Don't let AI subnet probes scan public neighboring hosts

src/network_prober.py

@@ -146,21 +146,24 @@ 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
     if isinstance(targets, str):
         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)

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 == []