import demistomock as demisto # noqa: F401 from CommonServerPython import * # noqa: F401 import traceback import ipaddress import json def get_nsg_rules(subscription_id: str, rg_name: str, nsg_name: str, integration_instance: str) -> tuple[list, str]: """ Runs command 'azure-vn-security-rules-list' to fetch all rules in a given Azure NSG. Args: subscription_id (str): Azure Subscription ID rg_name (str): The name of the Azure Resource Group where the NSG is located nsg_name (str): The name of the NSG to fetch rules from integration_instance (str): The name of the Azure integration instance to use for the command Returns: tuple[list, str]: - List of retrieved NSG Inbound rules, sorted in ascending priority order - String with name of the Integration instance that successfully retrieved NSG rules Raises: DemistoException: If there's an error fetching NSG rules """ cmd_args = { "subscription_id": subscription_id, "resource_group_name": rg_name, "network_security_group_name": nsg_name, "using": integration_instance, } result = demisto.executeCommand("azure-vn-security-rules-list", cmd_args) nsg_rules = [] if result and len(result) > 1: # If multiple entries were returned, such as when multiple Azure integration instances are configured, # Identify the first entry with valid results. for entry in result: if not isError(entry): nsg_rules = [entry] break else: # If all entries are errors, use the first entry nsg_rules = [result[0]] else: nsg_rules = result if not nsg_rules: raise DemistoException( "Error retrieving security group details with command 'azure-vn-security-rules-list'.\n" "Error: No results returned." ) if isError(nsg_rules): raise DemistoException( f"Error retrieving security group details with command 'azure-vn-security-rules-list'.\n" f"Error: {json.dumps(nsg_rules[0].get('Contents', ''))}" ) instance_to_use = dict_safe_get(nsg_rules, (0, "Metadata", "instance")) # Get all Inbound rules inbound_rules = [ rule for rule in nsg_rules[0]["Contents"] if rule.get("properties", {}).get("direction", "").lower() == "inbound" ] # Sort inbound rules by priority (ascending order) sorted_rules = sorted(inbound_rules, key=lambda rule: rule.get("properties", {}).get("priority", 0)) if not sorted_rules: raise DemistoException("No inbound NSG rules found in the specified Network Security Group.") return sorted_rules, instance_to_use def find_matching_rule( port: int, protocol: str, destination_ips: list[ipaddress.IPv4Address | ipaddress.IPv6Address], nsg_rules: list[dict], ) -> tuple[str, int]: """ Find the first NSG Allow rule that matches the specified port, protocol, and any of the destination IPs. Args: port (int): The destination port to match protocol (str): The protocol to match (case-insensitive) destination_ips (list[IPv4Address | IPv6Address]): The destination IP addresses to match (rule matches if ANY IP matches) nsg_rules (list[dict]): List of NSG command results Returns: tuple[str, int]: (rule_name, rule_priority) Raises: DemistoException: If no matching NSG inbound rule is found """ # Normalize protocol for case-insensitive comparison protocol_normalized = protocol.lower() for rule in nsg_rules: properties = rule.get("properties", {}) # Skip rules that don't allow traffic if properties.get("access", "").lower() != "allow": continue # Check protocol match (case-insensitive) rule_protocol = properties.get("protocol", "").lower() if rule_protocol != "*" and rule_protocol != protocol_normalized: continue # Check port match if not _matches_port(port, properties): continue # Check if ANY of the destination IPs match if not _matches_destination_ip(destination_ips, properties): continue # Found a matching rule return rule.get("name", ""), properties.get("priority", 0) # No matching rule found raise DemistoException("No matching NSG inbound rule found for the specified IP/Port/Protocol.") def _matches_port(target_port: int, rule_properties: dict) -> bool: """ Check if the target port matches the rule's destination port configuration. Args: target_port (int): The port to check rule_properties (dict): The rule properties containing port information Returns: bool: True if the port matches, False otherwise """ # Check single port range field single_port_range = rule_properties.get("destinationPortRange", "") if single_port_range and _port_matches_range(target_port, single_port_range): return True # Check multiple port ranges field multiple_port_ranges = rule_properties.get("destinationPortRanges", []) return any(_port_matches_range(target_port, port_range) for port_range in multiple_port_ranges) def _port_matches_range(target_port: int, port_range: str) -> bool: """ Check if a port matches a port range specification. Args: target_port (int): The port to check port_range (str): Port range specification (e.g., "80", "80-90", "80,443,8080-8090", "*") Returns: bool: True if the port matches, False otherwise """ # Handle wildcard if port_range.strip() == "*": return True # Split by commas to handle multiple ports/ranges port_specs = [spec.strip() for spec in port_range.split(",")] for spec in port_specs: if "-" in spec: # Handle range (e.g., "8080-8090") try: start_port_str, end_port_str = spec.split("-", 1) start_port = int(start_port_str.strip()) end_port = int(end_port_str.strip()) if start_port <= target_port <= end_port: return True except (ValueError, IndexError): continue else: # Handle individual port try: if int(spec) == target_port: return True except ValueError: continue return False def _matches_destination_ip(target_ips: list[ipaddress.IPv4Address | ipaddress.IPv6Address], rule_properties: dict) -> bool: """ Check if any of the target IPs match the rule's destination address configuration. Args: target_ips (list[IPv4Address | IPv6Address]): The IP addresses to check (rule matches if ANY IP matches) rule_properties (dict): The rule properties containing address information Returns: bool: True if any IP matches, False otherwise """ # Check single destination address prefix single_prefix = rule_properties.get("destinationAddressPrefix", "") if single_prefix: for target_ip_obj in target_ips: if _ip_matches_prefix(target_ip_obj, single_prefix): return True # Check multiple destination address prefixes multiple_prefixes = rule_properties.get("destinationAddressPrefixes", []) for prefix in multiple_prefixes: for target_ip_obj in target_ips: if _ip_matches_prefix(target_ip_obj, prefix): return True return False def _ip_matches_prefix(target_ip_obj: ipaddress.IPv4Address | ipaddress.IPv6Address, address_prefix: str) -> bool: """ Check if an IP address matches an address prefix. Args: target_ip_obj: ipaddress.IPv4Address or ipaddress.IPv6Address object address_prefix (str): Address prefix (e.g., "10.0.0.0/24", "192.168.1.100", "*") Returns: bool: True if the IP matches, False otherwise """ if not address_prefix or address_prefix.strip() == "": return False address_prefix = address_prefix.strip() # Handle wildcard if address_prefix == "*": return True try: # Try to parse as network (CIDR notation) if "/" in address_prefix: network = ipaddress.ip_network(address_prefix, strict=False) return target_ip_obj in network else: # Try to parse as individual IP address prefix_ip = ipaddress.ip_address(address_prefix) return target_ip_obj == prefix_ip except ValueError: return False def find_available_priorities(target_rule_priority: int, nsg_rules: list, priority_count: int) -> list[int]: """ Identifies unused NSG rule priority values below the target priority that can be used to insert new rules above it. Args: target_rule_priority (int): The priority of the rule you want to find available priorities before. nsg_rules (list[dict]): List of NSG command results priority_count (int): Number of priorities needed Raises: DemistoException: if the requested number of available priorities are not found. Returns: list[int]: List of available priorities before the target rule priority """ rule_priorities = [] # Store all used priority values for rule in nsg_rules: rule_priorities.append(int(rule.get("properties", {}).get("priority", 0))) # Format values as a set for easier evaluation rule_priorities_set = set(rule_priorities) available_priorities = [] # Find available priorities counting down from target_rule_priority - 1 # Only goes down to 100, as this is the lowest value supported by Azure NSG rules for priority in range(target_rule_priority - 1, 99, -1): # Count down from target-1 to 100 if priority not in rule_priorities_set: available_priorities.append(priority) # Stop once we have found the requested number of priorities if len(available_priorities) >= priority_count: break # Raise error if requested priority count exceeds the number available if len(available_priorities) < priority_count: raise DemistoException( f"Requested {priority_count} available priority values, but only found {len(available_priorities)} " f"below the matching rule's priority of {target_rule_priority}." ) return available_priorities def process_nsg_info(args: dict[str, str]) -> CommandResults: """ Main command function to identify NSG rule causing an exposure. Args: args (Dict[str, Any]): Demisto.args() object Returns: CommandResults: Demisto CommandResults object containing: - MatchingRuleName: Name of the matching NSG rule - MatchingRulePriority: Priority of the matching rule - NextAvailablePriorityValues: List of available priorities - IntegrationInstance: Azure integration instance used Raises: ValueError: If required parameters are missing or invalid DemistoException: If Azure API operations fail """ subscription_id = args.get("subscription_id", "") rg_name = args.get("resource_group_name", "") nsg_name = args.get("network_security_group_name", "") destination_ip_input = args.get("private_ip_addresses", "") port = arg_to_number(args.get("port"), required=True) or 0 protocol = args.get("protocol", "") priority_count = arg_to_number(args.get("priority_count"), required=True) or 0 integration_instance = args.get("integration_instance", "") # Format provided IP addresses as a list. Handle both single IP and list of IPs destination_ips = [] if isinstance(destination_ip_input, list): destination_ips = destination_ip_input elif isinstance(destination_ip_input, str): # Handle comma-separated IPs in a single string or single IP destination_ips = [ip.strip() for ip in destination_ip_input.split(",") if ip.strip()] if not destination_ips: raise ValueError("At least one valid IP address must be provided in private_ip_address parameter") # Validate that all provided IPs are valid IP addresses and create IP Address objects valid_ips = [] for ip in destination_ips: try: valid_ips.append(ipaddress.ip_address(ip)) except ValueError: raise ValueError(f"Invalid IP address provided: {ip}") if not valid_ips: raise ValueError("No valid IP addresses found in private_ip_address parameter") # Retrieve NSG rules and identify the Azure integration instance to use nsg_rules, instance_to_use = get_nsg_rules(subscription_id, rg_name, nsg_name, integration_instance) # Find the name and priority value of the first Allow rule that matches the provided criteria matching_rule_name, priority = find_matching_rule(port, protocol, valid_ips, nsg_rules) # Identify available priority values to insert new rules ahead of the matched rule available_priorities = find_available_priorities(priority, nsg_rules, priority_count) outputs = { "MatchingRuleName": matching_rule_name, "MatchingRulePriority": priority, "NextAvailablePriorityValues": available_priorities, "IntegrationInstance": instance_to_use, } return CommandResults( outputs_prefix="AzurePublicExposure", outputs_key_field="MatchingRuleName", outputs=outputs, raw_response=outputs, ) def main(): try: return_results(process_nsg_info(demisto.args())) except Exception as ex: demisto.error(traceback.format_exc()) # print the traceback return_error(f"Failed to execute AzureIdentifyNSGExposureRule. Error: {str(ex)}") if __name__ in ("__main__", "__builtin__", "builtins"): # pragma: no cover main()