AzureIdentifyNSGExposureRule

Identifies the first inbound Allow rule in a specified NSG that matches network exposure criteria. Returns rule details and a list of available priority numbers for inserting higher-priority rules.

Type

python

Pack

CloudSecurityPolicyManagement

Source

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()

README