diff --git a/src/network_prober.py b/src/network_prober.py index 8341ece..b422404 100644 --- a/src/network_prober.py +++ b/src/network_prober.py @@ -146,8 +146,11 @@ def probe_targets(targets: list[str], ports: list[int], """Probe specific *targets* on specific *ports* — used by AI-directed probing. Skips the live-host sweep; tests each target:port pair directly. - Only RFC1918 (or *allowed_net*) addresses are probed, and the cartesian - product is capped at ``_MAX_PROBE_PAIRS``. + Only RFC1918 addresses are probed. ``allowed_net`` further *restricts* + to that subnet (the target's /24) — it must never enlarge the allowlist + to a public network, which is what happens if a public target's + ``_subnet_of()`` /24 replaces RFC1918. + The cartesian product is capped at ``_MAX_PROBE_PAIRS``. """ if threads is not None: max_workers = threads @@ -155,12 +158,12 @@ def probe_targets(targets: list[str], ports: list[int], targets = [targets] if not targets or not ports: return [] - nets: list[ipaddress.IPv4Network | ipaddress.IPv6Network] = list(_RFC1918) + allowed: ipaddress.IPv4Network | ipaddress.IPv6Network | None = None if allowed_net: try: - nets = [ipaddress.ip_network(allowed_net, strict=False)] + allowed = ipaddress.ip_network(allowed_net, strict=False) except ValueError: - nets = list(_RFC1918) + allowed = None filtered: list[str] = [] seen: set[str] = set() for h in targets: @@ -171,7 +174,10 @@ def probe_targets(targets: list[str], ports: list[int], addr = ipaddress.ip_address(host) except ValueError: continue - if not any(addr in net for net in nets): + # RFC1918 is mandatory — allowed_net may only narrow, never expand. + if not _is_rfc1918(host): + continue + if allowed is not None and addr not in allowed: continue seen.add(host) filtered.append(host) diff --git a/test_ip_scope.py b/test_ip_scope.py index 7a9c6b6..e3671c8 100644 --- a/test_ip_scope.py +++ b/test_ip_scope.py @@ -123,3 +123,37 @@ def test_merge_agent_dedupes_and_skips_tech_inventory(): def test_probe_targets_drops_public_hosts(): from src.network_prober import probe_targets assert probe_targets(["8.8.8.8", "127.0.0.1", "169.254.1.1"], [80, 443]) == [] + + +def test_probe_targets_public_allowed_net_does_not_enable_public_scans(): + """AI-directed probes pass allowed_net=_subnet_of(target_ip). For a public + target that is a public /24 — it must not replace the RFC1918 gate and + scan neighboring hosts the operator never authorized.""" + from src.network_prober import probe_targets, _subnet_of + + public_ip = "45.33.32.156" # scanme.nmap.org-class public address + neighbors = ["45.33.32.1", "45.33.32.2", "8.8.8.8"] + found = probe_targets( + neighbors, [80, 443], + allowed_net=_subnet_of(public_ip), + ) + assert found == [] + + +def test_probe_targets_allowed_net_narrows_rfc1918_only(monkeypatch): + """On a private target, allowed_net still limits probes to that /24.""" + from src import network_prober + + monkeypatch.setattr(network_prober, "_tcp_connect", lambda host, port, timeout: True) + + in_subnet = network_prober.probe_targets( + ["10.0.0.5", "10.0.0.9"], [22], + allowed_net="10.0.0.0/24", + ) + assert {h.ip for h in in_subnet} == {"10.0.0.5", "10.0.0.9"} + + other_rfc1918 = network_prober.probe_targets( + ["10.1.1.5", "172.16.0.4"], [22], + allowed_net="10.0.0.0/24", + ) + assert other_rfc1918 == []