fix(egress): wire reviewer-mode L2 import-audit (ast-based)

scripts/guard/egress_guard.py

@@ -1,13 +1,15 @@
 """Mode-guard: defense-in-depth egress control for the Sovereign reviewer path.
 
 Three levels, strongest first:
-  1. Mode declaration   — reviewer-mode refuses remote KB backends (this module).
-  2. Capability removal — audit reviewer-path scripts for network imports (Task 2).
-  3. Process sandbox    — run under `unshare -n` when available, else best-effort (Task 3).
+  1. Mode declaration   — reviewer-mode refuses remote KB backends (assert_local_only, wired in kb/query.py).
+  2. Capability removal — reviewer-mode audits the dispatched provider script for network
+                          imports (assert_local_imports, wired in kb/query.py).
+  3. Process sandbox    — run under `unshare -n` when available (run_sandboxed); scaffolded,
+                          no reviewer subprocess chokepoint wires it yet.
 
 The guard NEVER claims an isolation it cannot deliver — status() discloses the truth.
 """
-import re
+import ast
 import shutil
 import subprocess
 
@@ -35,24 +37,56 @@ def assert_local_only(backend, mode):
     "requests", "httpx", "urllib", "urllib2", "socket",
     "aiohttp", "http", "ftplib", "telnetlib",
 }
-_IMPORT_RE = re.compile(r"^\s*(?:import\s+(\w+)|from\s+(\w+)\s+import)")
+
+
+def _imported_tops(node):
+    """Top-level module names an ast Import/ImportFrom node brings in (absolute only)."""
+    if isinstance(node, ast.Import):
+        return [a.name.split(".")[0] for a in node.names]
+    if isinstance(node, ast.ImportFrom) and node.level == 0 and node.module:
+        return [node.module.split(".")[0]]
+    return []
 
 
 def audit_imports(paths):
-    """Scan .py files for network-library imports. Returns one violation per hit."""
+    """Scan .py files for network-library imports. Returns one violation per hit.
+
+    AST-based, so it catches dotted (`from urllib.request import x`), comma
+    (`import os, socket`), and aliased imports a line regex would miss.
+    """
     violations = []
     for path in paths:
+        if not path:
+            continue  # e.g. inspect.getsourcefile returned None — nothing to audit
         with open(path, "r", encoding="utf-8") as fh:
-            for lineno, line in enumerate(fh, start=1):
-                m = _IMPORT_RE.match(line)
-                if not m:
-                    continue
-                module = m.group(1) or m.group(2)
-                if module in NETWORK_MODULES:
-                    violations.append({"file": path, "module": module, "line": lineno})
+            src = fh.read()
+        try:
+            tree = ast.parse(src, filename=str(path))
+        except SyntaxError:
+            continue  # unparseable -> can't run either; nothing to audit
+        for node in ast.walk(tree):
+            for mod in _imported_tops(node):
+                if mod in NETWORK_MODULES:
+                    violations.append({"file": str(path), "module": mod, "line": node.lineno})
     return violations
 
 
+def assert_local_imports(paths, mode):
+    """In reviewer-mode, refuse reviewer-path scripts that import a network module.
+
+    Level-2 (capability removal) companion to assert_local_only. No-op outside reviewer-mode.
+    """
+    if mode != "reviewer":
+        return None
+    violations = audit_imports(paths)
+    if violations:
+        mods = sorted({v["module"] for v in violations})
+        raise EgressViolation(
+            f"reviewer-mode: network imports in reviewer-path script(s): {mods} ({violations})"
+        )
+    return None
+
+
 _UNSHARE = shutil.which("unshare")
 
 

scripts/kb/query.py

@@ -4,6 +4,7 @@
 Flow: egress_guard.assert_local_only → provider dispatch → ok envelope {evidence, finding}.
 Reviewer-mode + remote backend → guard raises → error envelope (no egress).
 """
+import inspect
 import sys
 from pathlib import Path
 
@@ -18,8 +19,10 @@ def main():
         req = json_io.read_input()
         backend = req["backend"]
         mode = req.get("mode", "author")
-        egress_guard.assert_local_only(backend, mode)  # may raise EgressViolation
+        egress_guard.assert_local_only(backend, mode)  # L1: refuse remote backend by name
         fn = provider.get_provider(backend)
+        # L2: in reviewer-mode, audit the dispatched provider's source for network imports.
+        egress_guard.assert_local_imports([inspect.getsourcefile(fn)], mode)
         result = fn(req["claim"], req["path"])
         json_io.emit(json_io.ok(result))
     except egress_guard.EgressViolation as exc:

tests/guard/test_egress_guard.py

@@ -38,6 +38,36 @@ def test_audit_imports_clean_file(tmp_path):
     assert g.audit_imports([str(f)]) == []
 
 
+def test_audit_imports_catches_dotted_and_comma(tmp_path):
+    # Regression: the old line-regex missed dotted from-imports and comma lists.
+    f = tmp_path / "sneaky.py"
+    f.write_text(
+        "from urllib.request import urlopen\n"
+        "import os, socket\n"
+        "from http.client import HTTPConnection\n"
+    )
+    mods = {v["module"] for v in g.audit_imports([str(f)])}
+    assert {"urllib", "socket", "http"} <= mods
+    assert "os" not in mods
+
+
+def test_assert_local_imports_blocks_network_in_reviewer(tmp_path):
+    f = tmp_path / "leaky.py"
+    f.write_text("import requests\n")
+    with pytest.raises(g.EgressViolation):
+        g.assert_local_imports([str(f)], mode="reviewer")
+
+
+def test_assert_local_imports_allows_clean_and_author(tmp_path):
+    clean = tmp_path / "clean.py"
+    clean.write_text("import json\n")
+    assert g.assert_local_imports([str(clean)], mode="reviewer") is None
+    leaky = tmp_path / "leaky.py"
+    leaky.write_text("import socket\n")
+    # author-mode is not egress-restricted, so a network import is allowed there.
+    assert g.assert_local_imports([str(leaky)], mode="author") is None
+
+
 def test_sandbox_claim_matches_reality():
     available = g.sandbox_available()
     result = g.run_sandboxed([__import__("sys").executable, "-c", "print('hi')"])

tests/kb/test_query_cli.py

@@ -23,6 +23,15 @@ def test_author_mode_folder_query(tmp_path):
     assert len(out["data"]["evidence"]) >= 1
 
 
+def test_reviewer_mode_folder_query_ok(tmp_path):
+    # L2 audits the (clean) folder provider -> reviewer-mode local query still succeeds.
+    (tmp_path / "a.md").write_text("Spirometry shows obstruction in asthma.\n")
+    out = run_cli({"claim": "spirometry obstruction asthma",
+                   "backend": "folder", "path": str(tmp_path), "mode": "reviewer"})
+    assert out["status"] == "ok"
+    assert len(out["data"]["evidence"]) >= 1
+
+
 def test_reviewer_mode_refuses_rag(tmp_path):
     out = run_cli({"claim": "x", "backend": "rag", "path": str(tmp_path), "mode": "reviewer"})
     assert out["status"] == "error"