diff --git a/compliance/frameworks/cis_azure_benchmark.json b/compliance/frameworks/cis_azure_benchmark.json index 10bd80ad..10bacd20 100644 --- a/compliance/frameworks/cis_azure_benchmark.json +++ b/compliance/frameworks/cis_azure_benchmark.json @@ -133,6 +133,11 @@ "control_name": "Ensure that 'OS patching' is enabled for virtual machines", "description": "The virtual machine does not have automatic OS patching enabled. CIS 8.3 requires that OS patches are applied in a timely manner. Unpatched VMs are vulnerable to known exploits targeting unpatched OS vulnerabilities." }, + "AZ-CMP-007": { + "control_id": "N/A-CMP-007", + "control_name": "Just-In-Time (JIT) VM access - Defender for Cloud recommendation, no numbered CIS Azure Foundations 2.0.0 control", + "description": "CIS Microsoft Azure Foundations Benchmark 2.0.0 has no numbered recommendation for Just-In-Time VM access (it is a Microsoft Defender for Cloud recommendation), so under the repository's one-CIS-ID-per-rule convention this rule is not assigned a fabricated control id. It is mapped under NIST CSF PR.AC-3, ISO 27001 A.13.1.1, and SOC 2 CC6.6 instead." + }, "AZ-KV-001": { "control_id": "N/A-KV-001", "control_name": "Key Vault soft-delete baseline (covered by the repository's CIS 8.5 purge-protection rule)", diff --git a/compliance/frameworks/iso27001.json b/compliance/frameworks/iso27001.json index 9044730f..bd5a4d2f 100644 --- a/compliance/frameworks/iso27001.json +++ b/compliance/frameworks/iso27001.json @@ -128,6 +128,11 @@ "control_name": "Management of technical vulnerabilities", "description": "The virtual machine does not have automatic OS patching enabled. A.12.6.1 requires that information about technical vulnerabilities is obtained and the organisation's exposure evaluated. Without automatic patching, known OS vulnerabilities remain unmitigated." }, + "AZ-CMP-007": { + "control_id": "A.13.1.1", + "control_name": "Network controls", + "description": "A VM has management ports (SSH/RDP) open to the internet with no Just-In-Time VM access policy covering them. A.13.1.1 requires network controls that manage and protect access to systems. JIT limits management-port exposure to approved, time-boxed windows." + }, "AZ-CMP-003": { "control_id": "A.12.2.1", "control_name": "Controls against malware", diff --git a/compliance/frameworks/nist_csf.json b/compliance/frameworks/nist_csf.json index 7fe766b0..89db1398 100644 --- a/compliance/frameworks/nist_csf.json +++ b/compliance/frameworks/nist_csf.json @@ -133,6 +133,11 @@ "control_name": "A vulnerability management plan is developed and implemented", "description": "The virtual machine does not have automatic OS patching enabled. PR.IP-12 requires that a vulnerability management plan is developed and implemented. Without automatic patching, known OS vulnerabilities remain unmitigated and exploitable." }, + "AZ-CMP-007": { + "control_id": "PR.AC-3", + "control_name": "Remote access is managed", + "description": "A VM has management ports (SSH/RDP) open to the internet with no Just-In-Time VM access policy covering them. PR.AC-3 requires that remote access is managed. JIT restricts management-port access to approved, time-boxed requests instead of leaving the ports standing open." + }, "AZ-KV-001": { "control_id": "PR.IP-4", "control_name": "Backups of information are conducted, maintained, and tested", diff --git a/compliance/frameworks/soc2.json b/compliance/frameworks/soc2.json index 38430730..36db57fb 100644 --- a/compliance/frameworks/soc2.json +++ b/compliance/frameworks/soc2.json @@ -148,6 +148,11 @@ "control_name": "System Vulnerabilities are Identified and Managed", "description": "The virtual machine does not have automatic OS patching enabled. CC7.1 requires that vulnerabilities in system components are identified and managed through a defined process. Without automatic patching, known OS vulnerabilities are left unmitigated and exploitable." }, + "AZ-CMP-007": { + "control_id": "CC6.6", + "control_name": "Restricts Access from Outside the Network Boundary", + "description": "A VM has management ports (SSH/RDP) open to the internet with no Just-In-Time VM access policy covering them. CC6.6 requires that access from outside the network boundary is restricted. JIT opens management ports only for approved, time-boxed requests instead of continuously." + }, "AZ-KV-001": { "control_id": "A1.2", "control_name": "Environmental Threats and Recovery", diff --git a/playbooks/cli/fix_az_cmp_007.sh b/playbooks/cli/fix_az_cmp_007.sh new file mode 100644 index 00000000..cf834231 --- /dev/null +++ b/playbooks/cli/fix_az_cmp_007.sh @@ -0,0 +1,39 @@ +#!/usr/bin/env bash +# fix_az_cmp_007.sh +# Enable Microsoft Defender for Cloud Just-In-Time (JIT) VM access so a VM's +# management ports (SSH 22 / RDP 3389) are only opened on approved, time-boxed +# requests instead of standing open to the internet. +# +# The shared 'default' JIT policy is a create-or-update whose PUT replaces the +# whole virtualMachines array, so this script first GETs the current policy and +# merges this VM in (preserving every other VM's JIT config) via +# jit_policy_merge.py before PUTting it back. +# +# Usage: ./fix_az_cmp_007.sh +# Requires: Microsoft Defender for Servers enabled on the subscription. + +set -euo pipefail + +RESOURCE_GROUP="${1:-}" +VM_NAME="${2:-}" +LOCATION="${3:-}" + +if [[ -z "$RESOURCE_GROUP" || -z "$VM_NAME" || -z "$LOCATION" ]]; then + echo "Usage: $0 " + exit 1 +fi + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +SUBSCRIPTION_ID="$(az account show --query id -o tsv)" +VM_ID="/subscriptions/${SUBSCRIPTION_ID}/resourceGroups/${RESOURCE_GROUP}/providers/Microsoft.Compute/virtualMachines/${VM_NAME}" +POLICY_URL="https://management.azure.com/subscriptions/${SUBSCRIPTION_ID}/resourceGroups/${RESOURCE_GROUP}/providers/Microsoft.Security/locations/${LOCATION}/jitNetworkAccessPolicies/default?api-version=2020-01-01" + +echo "Reading existing JIT policy (if any) in $LOCATION..." +EXISTING="$(az rest --method GET --url "$POLICY_URL" 2>/dev/null || true)" + +echo "Merging $VM_NAME (ports 22, 3389) into the policy without dropping other VMs..." +BODY="$(printf '%s' "$EXISTING" | python3 "${SCRIPT_DIR}/jit_policy_merge.py" "$VM_ID" 22 3389)" + +az rest --method PUT --url "$POLICY_URL" --body "$BODY" + +echo "Done. JIT policy 'default' now covers $VM_NAME (existing VMs preserved); management ports require an approved, time-boxed request." diff --git a/playbooks/cli/jit_policy_merge.py b/playbooks/cli/jit_policy_merge.py new file mode 100644 index 00000000..28719484 --- /dev/null +++ b/playbooks/cli/jit_policy_merge.py @@ -0,0 +1,70 @@ +#!/usr/bin/env python3 +"""Merge one VM's JIT entry into an existing Defender for Cloud JIT policy. + +The Defender for Cloud ``jitNetworkAccessPolicies`` PUT is a *create-or-update* +that replaces the whole ``properties.virtualMachines`` array. Sending a policy +that contains only the VM being remediated would therefore drop JIT coverage for +every other VM already in the shared ``default`` policy. + +This helper reads the current policy JSON from stdin (empty/whitespace means the +policy does not exist yet) and prints the full policy to PUT, preserving every +other VM's entry and replacing (not duplicating) the target VM's own entry. + +Usage: + az rest --method GET ... | jit_policy_merge.py [ ...] +""" + +from __future__ import annotations + +import json +import sys +from typing import Any, Dict, List + +_DEFAULT_MAX_DURATION = "PT3H" + + +def build_vm_entry(vm_id: str, ports: List[int]) -> Dict[str, Any]: + return { + "id": vm_id, + "ports": [ + { + "number": port, + "protocol": "*", + "allowedSourceAddressPrefix": "*", + "maxRequestAccessDuration": _DEFAULT_MAX_DURATION, + } + for port in ports + ], + } + + +def merge_policy(existing: Dict[str, Any] | None, vm_entry: Dict[str, Any]) -> Dict[str, Any]: + """Return a policy that keeps every other VM and (re)sets the target VM's entry.""" + if existing and isinstance(existing.get("properties"), dict): + policy = existing + else: + policy = {"kind": "Basic", "properties": {}} + properties = policy.setdefault("properties", {}) + others = [ + vm + for vm in (properties.get("virtualMachines") or []) + if isinstance(vm, dict) and vm.get("id") != vm_entry["id"] + ] + properties["virtualMachines"] = others + [vm_entry] + return policy + + +def main(argv: List[str]) -> int: + if len(argv) < 3: + print(f"usage: {argv[0]} [ ...]", file=sys.stderr) + return 2 + vm_id = argv[1] + ports = [int(port) for port in argv[2:]] + raw = sys.stdin.read().strip() + existing = json.loads(raw) if raw else None + print(json.dumps(merge_policy(existing, build_vm_entry(vm_id, ports)))) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv)) diff --git a/scanner/azure_client.py b/scanner/azure_client.py index dda5befa..bda05d46 100644 --- a/scanner/azure_client.py +++ b/scanner/azure_client.py @@ -431,6 +431,23 @@ def get_web_apps(self) -> List[Any]: logger.error("get_web_apps failed: %s", exc) return [] + def get_jit_network_access_policies(self) -> Optional[List[Any]]: + """List Microsoft Defender for Cloud Just-In-Time VM access policies. + + Returns a list (including an empty list) when Defender for Cloud responds, + or ``None`` when permissions, networking, the SDK, or a subscription + without Defender for Cloud prevent the collection from being evaluated. + Callers must treat ``None`` as "JIT coverage unknown", never as "no JIT". + """ + try: + from azure.mgmt.security import SecurityCenter + + client = SecurityCenter(self.credential, self.subscription_id) + return list(client.jit_network_access_policies.list()) + except Exception as exc: + logger.error("get_jit_network_access_policies failed: %s", exc) + return None + def get_function_app_security_posture(self) -> Optional[List[Dict[str, Any]]]: """Return a cached, secret-free posture for Function Apps.""" if self._function_apps_cache is not _UNSET: diff --git a/scanner/rules/az_cmp_007.py b/scanner/rules/az_cmp_007.py new file mode 100644 index 00000000..776c19ad --- /dev/null +++ b/scanner/rules/az_cmp_007.py @@ -0,0 +1,220 @@ +"""AZ-CMP-007: VM management ports open without Just-In-Time (JIT) VM access.""" + +import logging +from typing import Any, Dict, List + +RULE_ID = "AZ-CMP-007" +RULE_NAME = "VM Management Ports Open Without Just-In-Time (JIT) Access" +SEVERITY = "MEDIUM" +CATEGORY = "Compute" +FRAMEWORKS = { + "CIS": "N/A-CMP-007", + "NIST": "PR.AC-3", + "ISO27001": "A.13.1.1", + "SOC2": "CC6.6", +} +DESCRIPTION = ( + "A virtual machine has management ports (SSH 22 / RDP 3389) allowed inbound " + "from the internet by a network security group (on the NIC or its subnet), and no Microsoft Defender " + "for Cloud Just-In-Time (JIT) VM access policy covers those ports. The ports " + "are therefore open on a standing basis instead of only during an approved, " + "time-boxed request, leaving them continuously exposed to scanning and " + "brute-force attacks. VMs with no management ports open are not applicable." +) +REMEDIATION = ( + "Enable Just-In-Time VM access in Microsoft Defender for Cloud for the affected " + "VM and ports (or `az security jit-policy create`), so management ports are only " + "opened for an approved, time-limited window. Alternatively restrict the NSG " + "rule to trusted source ranges or use Azure Bastion for administrative access." +) +PLAYBOOK = "playbooks/cli/fix_az_cmp_007.sh" + +logger = logging.getLogger(__name__) + +# Inbound management ports this rule cares about (SSH, RDP). +_MANAGEMENT_PORTS = ("22", "3389") +# Source specifications that mean "reachable from anywhere on the internet". +_OPEN_SOURCES = {"*", "0.0.0.0/0", "Internet", "Any"} + + +def _spec_covers_port(spec: str, port: int) -> bool: + """True if an NSG destination-port spec covers ``port``. + + A spec is ``*`` (all ports), a single number (``22``), or an inclusive range + (``20-30``). Ranges are the case the earlier exact-match logic missed: a rule + with ``destination_port_range = "20-30"`` exposes SSH but read as closed. + """ + spec = spec.strip() + if not spec: + return False + if spec == "*": + return True + if "-" in spec: + low, _, high = spec.partition("-") + try: + return int(low) <= port <= int(high) + except ValueError: + return False + try: + return int(spec) == port + except ValueError: + return False + + +def _rule_allows_port_from_any(rule: Any, port: str) -> bool: + """True if an NSG security rule allows inbound traffic on ``port`` from any source.""" + if str(getattr(rule, "direction", "")).lower() != "inbound": + return False + if str(getattr(rule, "access", "")).lower() != "allow": + return False + + source = getattr(rule, "source_address_prefix", "") or "" + source_prefixes = getattr(rule, "source_address_prefixes", []) or [] + if source not in _OPEN_SOURCES and not any(s in _OPEN_SOURCES for s in source_prefixes): + return False + + port_int = int(port) + dest_specs = [str(getattr(rule, "destination_port_range", "") or "")] + dest_specs.extend(str(item) for item in (getattr(rule, "destination_port_ranges", []) or [])) + return any(_spec_covers_port(spec, port_int) for spec in dest_specs) + + +def _jit_coverage(policies: List[Any]) -> Dict[str, set]: + """Map lower-cased VM resource id -> set of JIT-covered management ports. + + A port covered as ``*`` (or the literal management port number) counts as + covered. An empty ``policies`` list yields an empty map (genuinely no JIT + coverage); the caller handles the indeterminate (``None``) case separately. + """ + coverage: Dict[str, set] = {} + for policy in policies or []: + for jit_vm in getattr(policy, "virtual_machines", []) or []: + vm_id = (getattr(jit_vm, "id", "") or "").lower() + if not vm_id: + continue + ports = {str(getattr(p, "number", "")) for p in (getattr(jit_vm, "ports", []) or [])} + coverage.setdefault(vm_id, set()).update(ports) + return coverage + + +def _applicable_nsgs( + azure_client: Any, + vm: Any, + nsg_by_id: Dict[str, Any], + nsg_by_subnet_id: Dict[str, Any], +) -> List[Any]: + """Every NSG that governs this VM's inbound traffic — NIC-level *and* subnet-level. + + An NSG can be attached directly to the NIC or to the NIC's subnet. A VM with + no NIC-level NSG can still be exposed through its subnet NSG, so both paths + must be considered or the rule reports a false NOT_APPLICABLE. + """ + nsgs: List[Any] = [] + seen: set = set() + network_profile = getattr(vm, "network_profile", None) + if not network_profile: + return nsgs + + def _add(nsg: Any) -> None: + if nsg is None: + return + key = (getattr(nsg, "id", "") or "").lower() or id(nsg) + if key not in seen: + seen.add(key) + nsgs.append(nsg) + + for nic_ref in getattr(network_profile, "network_interfaces", []) or []: + parsed = azure_client.parse_resource_id(getattr(nic_ref, "id", "")) + resource_group = parsed.get("resource_group", "") + nic_name = parsed.get("name", "") + if not resource_group or not nic_name: + continue + nic = azure_client.get_network_interface(resource_group, nic_name) + if not nic: + continue + nic_nsg_ref = getattr(nic, "network_security_group", None) + _add(nsg_by_id.get((getattr(nic_nsg_ref, "id", "") or "").lower())) + for ip_config in getattr(nic, "ip_configurations", []) or []: + subnet = getattr(ip_config, "subnet", None) + _add(nsg_by_subnet_id.get((getattr(subnet, "id", "") or "").lower())) + return nsgs + + +def _open_management_ports( + azure_client: Any, + vm: Any, + nsg_by_id: Dict[str, Any], + nsg_by_subnet_id: Dict[str, Any], +) -> set: + """Return the management ports open to the internet on any NSG that governs the VM.""" + open_ports: set = set() + for nsg in _applicable_nsgs(azure_client, vm, nsg_by_id, nsg_by_subnet_id): + for security_rule in getattr(nsg, "security_rules", []) or []: + for port in _MANAGEMENT_PORTS: + if _rule_allows_port_from_any(security_rule, port): + open_ports.add(port) + return open_ports + + +def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: + """Flag VMs with internet-open management ports not covered by a JIT policy. + + A VM is flagged only when at least one management port (22/3389) is open to + the internet through its NIC's NSG *and* that port is not covered by a + Defender for Cloud JIT policy for that VM. VMs with no management ports open + are treated as NOT_APPLICABLE and are not flagged. + """ + findings: List[Dict[str, Any]] = [] + + policies = azure_client.get_jit_network_access_policies() + if policies is None: + # Defender for Cloud could not be queried: coverage is indeterminate, so + # we cannot assert a VM has "no JIT policy". Skip rather than false-positive. + return findings + + coverage = _jit_coverage(policies) + all_nsgs = azure_client.get_network_security_groups() or [] + nsg_by_id = {(getattr(nsg, "id", "") or "").lower(): nsg for nsg in all_nsgs} + nsg_by_subnet_id: Dict[str, Any] = {} + for nsg in all_nsgs: + for subnet in getattr(nsg, "subnets", []) or []: + subnet_id = (getattr(subnet, "id", "") or "").lower() + if subnet_id: + nsg_by_subnet_id[subnet_id] = nsg + + for vm in azure_client.get_virtual_machines(): + open_ports = _open_management_ports(azure_client, vm, nsg_by_id, nsg_by_subnet_id) + if not open_ports: + continue # NOT_APPLICABLE: no management ports exposed + + covered_ports = coverage.get((getattr(vm, "id", "") or "").lower(), set()) + if "*" in covered_ports: + uncovered = set() + else: + uncovered = {port for port in open_ports if port not in covered_ports} + if not uncovered: + continue # every exposed port is covered by a JIT policy + + parsed = azure_client.parse_resource_id(getattr(vm, "id", "")) + findings.append( + { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": SEVERITY, + "category": CATEGORY, + "resource_id": getattr(vm, "id", ""), + "resource_name": getattr(vm, "name", None) or parsed.get("name", ""), + "resource_type": "Microsoft.Compute/virtualMachines", + "description": DESCRIPTION, + "remediation": REMEDIATION, + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + "metadata": { + "open_management_ports": sorted(open_ports), + "uncovered_ports": sorted(uncovered), + "jit_policy_present": bool(covered_ports), + }, + } + ) + + return findings diff --git a/tests/helpers/mock_azure.py b/tests/helpers/mock_azure.py index 5b8b859f..8fbc13f0 100644 --- a/tests/helpers/mock_azure.py +++ b/tests/helpers/mock_azure.py @@ -76,6 +76,7 @@ def __init__(self) -> None: self._kv_keys: Dict[str, List[Any]] = {} self._diagnostic_settings: Dict[str, Optional[bool]] = {} self._diagnostic_default: Optional[bool] = False + self._jit_policies: Optional[List[Any]] = [] self._conditional_access_policies: List[Any] = [] # Credential stub for Graph/SDK-based rules (idn_003..009). self.credential = _StubCredential() @@ -257,6 +258,14 @@ def set_disk(self, disk_id: str, disk: Optional[Any]) -> "MockAzureClient": def get_disk(self, disk_id: str) -> Optional[Any]: return self._disks.get(disk_id) + def set_jit_policies(self, policies: Optional[List[Any]]) -> "MockAzureClient": + """Configure the Defender for Cloud JIT policies; ``None`` represents an unreadable/indeterminate result.""" + self._jit_policies = policies + return self + + def get_jit_network_access_policies(self) -> Optional[List[Any]]: + return self._jit_policies + # ------------------------------------------------------------------ # # Storage — lifecycle & service logging (three-state: True/False/None) # # ------------------------------------------------------------------ # diff --git a/tests/test_jit_policy_merge.py b/tests/test_jit_policy_merge.py new file mode 100644 index 00000000..c665640a --- /dev/null +++ b/tests/test_jit_policy_merge.py @@ -0,0 +1,50 @@ +"""Tests for playbooks/cli/jit_policy_merge.py — the AZ-CMP-007 remediation helper. + +These prove the fix_az_cmp_007.sh remediation preserves an existing multi-VM JIT +policy instead of overwriting it with only the VM being remediated. +""" + +import importlib.util +import pathlib + +_MOD_PATH = pathlib.Path(__file__).resolve().parents[1] / "playbooks" / "cli" / "jit_policy_merge.py" +_spec = importlib.util.spec_from_file_location("jit_policy_merge", _MOD_PATH) +jit_policy_merge = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(jit_policy_merge) + + +def test_merge_preserves_existing_vms(): + """Remediating one VM must not drop other VMs already in the default policy.""" + existing = { + "kind": "Basic", + "properties": {"virtualMachines": [{"id": "/vm/other", "ports": [{"number": 22}]}]}, + } + entry = jit_policy_merge.build_vm_entry("/vm/new", [22, 3389]) + merged = jit_policy_merge.merge_policy(existing, entry) + ids = {vm["id"] for vm in merged["properties"]["virtualMachines"]} + assert ids == {"/vm/other", "/vm/new"} + + +def test_merge_replaces_same_vm_without_duplicating(): + """Re-remediating the same VM updates its entry in place rather than duplicating it.""" + existing = {"properties": {"virtualMachines": [{"id": "/vm/x", "ports": [{"number": 22}]}]}} + entry = jit_policy_merge.build_vm_entry("/vm/x", [3389]) + merged = jit_policy_merge.merge_policy(existing, entry) + vms = merged["properties"]["virtualMachines"] + assert len(vms) == 1 + assert vms[0]["id"] == "/vm/x" + assert [port["number"] for port in vms[0]["ports"]] == [3389] + + +def test_merge_creates_policy_when_none_exists(): + """With no existing policy, a fresh Basic policy containing just this VM is produced.""" + merged = jit_policy_merge.merge_policy(None, jit_policy_merge.build_vm_entry("/vm/a", [22])) + assert merged["kind"] == "Basic" + assert [vm["id"] for vm in merged["properties"]["virtualMachines"]] == ["/vm/a"] + + +def test_build_vm_entry_shape(): + entry = jit_policy_merge.build_vm_entry("/vm/a", [22, 3389]) + assert entry["id"] == "/vm/a" + assert [port["number"] for port in entry["ports"]] == [22, 3389] + assert all(port["maxRequestAccessDuration"] == "PT3H" for port in entry["ports"]) diff --git a/tests/test_rules_compute.py b/tests/test_rules_compute.py index af07c4c7..751bcb77 100644 --- a/tests/test_rules_compute.py +++ b/tests/test_rules_compute.py @@ -12,6 +12,7 @@ import scanner.rules.az_cmp_002 as az_cmp_002 import scanner.rules.az_cmp_003 as az_cmp_003 import scanner.rules.az_cmp_004 as az_cmp_004 +import scanner.rules.az_cmp_007 as az_cmp_007 from tests.helpers.mock_azure import make_resource try: @@ -380,3 +381,209 @@ def test_cmp_004_noncompliant_no_patching_returns_one_finding(mock_azure, subscr assert f["rule_id"] == "AZ-CMP-004" assert f["severity"] == "HIGH" assert f["resource_name"] == "vm-stale" + + +# ── AZ-CMP-007: management ports open without Just-In-Time (JIT) access ────── +# +# The rule resolves each VM's NIC -> NSG (by id, via get_network_security_groups) +# to find management ports (22/3389) open to the internet, then cross-references +# Defender for Cloud JIT policies (get_jit_network_access_policies) to see whether +# those ports are covered. These fixtures mirror that wiring. + + +def _nsg_id(name): + return f"/subscriptions/{_SUB}/resourceGroups/{_RG}/providers/Microsoft.Network/networkSecurityGroups/{name}" + + +def _sec_rule(port, direction="Inbound", access="Allow", source="Internet"): + return make_resource( + direction=direction, + access=access, + source_address_prefix=source, + source_address_prefixes=[], + destination_port_range=str(port), + destination_port_ranges=[], + ) + + +def _nsg(name, rules): + return make_resource(id=_nsg_id(name), name=name, security_rules=rules) + + +def _vm_on_nic(vm_name, nic_name): + return make_resource( + id=_vm_id(vm_name), + name=vm_name, + network_profile=make_resource(network_interfaces=[make_resource(id=_nic_id(nic_name))]), + ) + + +def _nic_on_nsg(nsg_name): + return make_resource(network_security_group=make_resource(id=_nsg_id(nsg_name))) + + +def _jit_policy(vm_name, ports): + return make_resource( + virtual_machines=[make_resource(id=_vm_id(vm_name), ports=[make_resource(number=p) for p in ports])] + ) + + +def _wire(mock_azure, vm, nic_name, nic, nsg, jit): + mock_azure.set_virtual_machines([vm]) + mock_azure.set_network_interface(_RG, nic_name, nic) + mock_azure.set_network_security_groups([nsg]) + mock_azure.set_jit_policies(jit) + + +def test_cmp_007_open_ssh_no_jit_returns_one_finding(mock_azure, subscription_id): + """SSH open to the internet with no JIT policy must be flagged MEDIUM.""" + _wire( + mock_azure, + _vm_on_nic("vm-open", "nic1"), + "nic1", + _nic_on_nsg("nsg1"), + _nsg("nsg1", [_sec_rule("22")]), + [], + ) + findings = az_cmp_007.scan(mock_azure, subscription_id) + assert len(findings) == 1 + f = findings[0] + assert _REQUIRED_FIELDS.issubset(f.keys()) + assert f["rule_id"] == "AZ-CMP-007" + assert f["severity"] == "MEDIUM" + assert f["resource_name"] == "vm-open" + assert f["metadata"]["open_management_ports"] == ["22"] + assert f["metadata"]["uncovered_ports"] == ["22"] + assert f["metadata"]["jit_policy_present"] is False + + +def test_cmp_007_open_ssh_with_jit_coverage_returns_no_findings(mock_azure, subscription_id): + """An open SSH port covered by a JIT policy for that VM is compliant.""" + _wire( + mock_azure, + _vm_on_nic("vm-jit", "nic1"), + "nic1", + _nic_on_nsg("nsg1"), + _nsg("nsg1", [_sec_rule("22")]), + [_jit_policy("vm-jit", [22])], + ) + assert az_cmp_007.scan(mock_azure, subscription_id) == [] + + +def test_cmp_007_no_management_ports_open_is_not_applicable(mock_azure, subscription_id): + """A VM whose NSG opens only non-management ports is NOT_APPLICABLE.""" + _wire( + mock_azure, + _vm_on_nic("vm-web", "nic1"), + "nic1", + _nic_on_nsg("nsg1"), + _nsg("nsg1", [_sec_rule("443")]), + [], + ) + assert az_cmp_007.scan(mock_azure, subscription_id) == [] + + +def test_cmp_007_management_port_from_trusted_source_not_flagged(mock_azure, subscription_id): + """RDP open only to a trusted CIDR (not the internet) must not be flagged.""" + _wire( + mock_azure, + _vm_on_nic("vm-trusted", "nic1"), + "nic1", + _nic_on_nsg("nsg1"), + _nsg("nsg1", [_sec_rule("3389", source="10.0.0.0/24")]), + [], + ) + assert az_cmp_007.scan(mock_azure, subscription_id) == [] + + +def test_cmp_007_partial_jit_coverage_flags_uncovered_port(mock_azure, subscription_id): + """When JIT covers SSH but RDP is also open, only the uncovered RDP port is reported.""" + _wire( + mock_azure, + _vm_on_nic("vm-partial", "nic1"), + "nic1", + _nic_on_nsg("nsg1"), + _nsg("nsg1", [_sec_rule("22"), _sec_rule("3389")]), + [_jit_policy("vm-partial", [22])], + ) + findings = az_cmp_007.scan(mock_azure, subscription_id) + assert len(findings) == 1 + assert findings[0]["metadata"]["open_management_ports"] == ["22", "3389"] + assert findings[0]["metadata"]["uncovered_ports"] == ["3389"] + assert findings[0]["metadata"]["jit_policy_present"] is True + + +def test_cmp_007_indeterminate_jit_is_not_flagged(mock_azure, subscription_id): + """When Defender for Cloud cannot be queried (None), coverage is unknown, so no finding.""" + _wire( + mock_azure, + _vm_on_nic("vm-x", "nic1"), + "nic1", + _nic_on_nsg("nsg1"), + _nsg("nsg1", [_sec_rule("3389")]), + None, + ) + assert az_cmp_007.scan(mock_azure, subscription_id) == [] + + +def _subnet_id(name): + return f"/subscriptions/{_SUB}/resourceGroups/{_RG}/providers/Microsoft.Network/virtualNetworks/vnet/subnets/{name}" + + +def test_cmp_007_port_range_covering_ssh_is_flagged(mock_azure, subscription_id): + """An NSG rule whose destination_port_range is a range (20-30) containing SSH (22) + exposes the port; the earlier exact-match logic treated 22 as closed.""" + _wire( + mock_azure, + _vm_on_nic("vm-range", "nic1"), + "nic1", + _nic_on_nsg("nsg1"), + _nsg("nsg1", [_sec_rule("20-30")]), + [], + ) + findings = az_cmp_007.scan(mock_azure, subscription_id) + assert len(findings) == 1 + assert findings[0]["metadata"]["open_management_ports"] == ["22"] + + +def test_cmp_007_port_range_in_ranges_list_covering_rdp(mock_azure, subscription_id): + """A range in destination_port_ranges (3380-3400) containing RDP (3389) is detected.""" + rule = make_resource( + direction="Inbound", + access="Allow", + source_address_prefix="Internet", + source_address_prefixes=[], + destination_port_range="", + destination_port_ranges=["3380-3400"], + ) + _wire( + mock_azure, + _vm_on_nic("vm-range2", "nic1"), + "nic1", + _nic_on_nsg("nsg1"), + _nsg("nsg1", [rule]), + [], + ) + findings = az_cmp_007.scan(mock_azure, subscription_id) + assert len(findings) == 1 + assert findings[0]["metadata"]["open_management_ports"] == ["3389"] + + +def test_cmp_007_subnet_level_nsg_exposure_is_flagged(mock_azure, subscription_id): + """A VM with no NIC-level NSG is still exposed if the NSG on its subnet opens SSH.""" + subnet_id = _subnet_id("subnet1") + nic = make_resource( + network_security_group=None, + ip_configurations=[make_resource(subnet=make_resource(id=subnet_id))], + ) + subnet_nsg = make_resource( + id=_nsg_id("nsg-sub"), + name="nsg-sub", + security_rules=[_sec_rule("22")], + subnets=[make_resource(id=subnet_id)], + ) + _wire(mock_azure, _vm_on_nic("vm-subnet", "nic1"), "nic1", nic, subnet_nsg, []) + findings = az_cmp_007.scan(mock_azure, subscription_id) + assert len(findings) == 1 + assert findings[0]["resource_name"] == "vm-subnet" + assert findings[0]["metadata"]["open_management_ports"] == ["22"]