From c70e807ea0ce4a371c0382a26a7119c829bd3e78 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=81ukasz=20Minarowski?= Date: Thu, 2 Jul 2026 22:08:09 +0200 Subject: [PATCH] fix(egress): wire reviewer-mode L2 import-audit; ast-based audit_imports egress_guard: audit_imports rewritten on ast (catches dotted `from a.b import`, comma `import a, b`, aliased imports the line-regex missed); add assert_local_imports (reviewer-mode; no-op otherwise); drop _IMPORT_RE/re; docstring L2 wired, L3 scaffolded kb/query.py: reviewer-mode audits dispatched provider source via assert_local_imports (L2) tests: dotted/comma audit, assert_local_imports reviewer/author, reviewer+folder e2e ok Co-Authored-By: Claude Opus 4.8 (1M context) --- scripts/guard/egress_guard.py | 60 +++++++++++++++++++++++++------- scripts/kb/query.py | 5 ++- tests/guard/test_egress_guard.py | 30 ++++++++++++++++ tests/kb/test_query_cli.py | 9 +++++ 4 files changed, 90 insertions(+), 14 deletions(-) diff --git a/scripts/guard/egress_guard.py b/scripts/guard/egress_guard.py index 9a765ac..2664c84 100644 --- a/scripts/guard/egress_guard.py +++ b/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") diff --git a/scripts/kb/query.py b/scripts/kb/query.py index 006fe5c..6421506 100644 --- a/scripts/kb/query.py +++ b/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: diff --git a/tests/guard/test_egress_guard.py b/tests/guard/test_egress_guard.py index 25dee40..e487b14 100644 --- a/tests/guard/test_egress_guard.py +++ b/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')"]) diff --git a/tests/kb/test_query_cli.py b/tests/kb/test_query_cli.py index ab3e1fc..46c5c29 100644 --- a/tests/kb/test_query_cli.py +++ b/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"