From 328406ad415997805bcbd50955495207a75ea43d Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Tue, 6 Oct 2026 17:07:24 +0200 Subject: [PATCH] =?UTF-8?q?search-with-proof:=20positive=20control=20for?= =?UTF-8?q?=20path=20oracle=20=E2=80=94=20honest=20serve=20signal,=20no=20?= =?UTF-8?q?false=20negatives?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The HTTP verifier could not distinguish a mis-wired or non-serving app from a genuine no-effect: both yielded `inconclusive`. This adds a positive control for the path oracle. - PathOracle plants a benign in-root marker file (random token from os.urandom, distinct from the secret canary) and exposes it via a new Oracle.control() hook (default None; only the path oracle implements it today). - On the first run of an HTTP side, _side runs a benign GET for that marker as its own fully-guarded app start. The marker returning proves the interface serves the fixture; its absence — or any control failure — is recorded in the new SideRecord.control_observed field but never changes the verdict, since a valid app may answer only its own routes. A crash is still caught by the existing could_not_run path. A 2xx-based "nothing served" block was rejected: the verifier deliberately treats any HTTP status as the app being alive (a 503-only app is a valid observed side), so status-based blocking would false-fail valid apps. Soundly converting a mis-wired-but-responding app into could_not_run requires a demo-declared control, which is a separate follow-up. Tests: new tests/test_search_control.py covers marker observed, absent marker not blocking, control failure not blocking a working attack, and crash still could_not_run. Full search-verify/probe/worker suites pass; ruff clean; mypy unchanged (only the pre-existing file_bindings lambda note). Co-Authored-By: Claude Opus 4.8 --- src/openultrasast/search/oracles.py | 31 ++++++ src/openultrasast/search/verify.py | 43 +++++++- tests/test_search_control.py | 153 ++++++++++++++++++++++++++++ 3 files changed, 226 insertions(+), 1 deletion(-) create mode 100644 tests/test_search_control.py diff --git a/src/openultrasast/search/oracles.py b/src/openultrasast/search/oracles.py index 8ab22da..61c7e07 100644 --- a/src/openultrasast/search/oracles.py +++ b/src/openultrasast/search/oracles.py @@ -37,6 +37,18 @@ class Canary: return cls(root, secrets.token_hex(24)) +@dataclass(frozen=True) +class ControlProbe: + """A benign interaction a correctly functioning app must answer by returning + ``token`` in its response body. It proves the demo's interface is wired and the + app can serve before any absence of the secret effect is trusted as a negative.""" + + method: str + path: str + token: str + body: str = "" + + class Oracle(Protocol): namespace_pid: int | None placeholders: dict[str, str] @@ -44,6 +56,7 @@ class Oracle(Protocol): def prepare(self, canary: Canary) -> dict[str, str]: ... def capture(self, output: str) -> None: ... def observe(self) -> tuple[bool, str]: ... + def control(self) -> ControlProbe | None: ... def close(self) -> None: ... @@ -62,19 +75,37 @@ class OutputOracle: def observe(self) -> tuple[bool, str]: return self.seen, "canary in verifier-captured app output" if self.seen else "no canary in app output" + def control(self) -> ControlProbe | None: + # No positive control by default. SQL/command/ssrf/xss controls are a + # documented follow-up; only the path oracle plants one for now. + return None + def close(self) -> None: pass class PathOracle(OutputOracle): + control_name = "control" + def prepare(self, canary: Canary) -> dict[str, str]: super().prepare(canary) (canary.root / "public").mkdir() (canary.root / "canary").write_text(canary.nonce) (canary.root / "canary").chmod(0o600) + # A benign in-root control file, distinct from the out-of-root secret, so a + # correctly serving app returns it on BOTH revisions without any traversal. + # Drawn from os.urandom, not secrets.token_hex, so it never perturbs callers + # that monkeypatch the nonce source and count its draws. + self.control_token = os.urandom(24).hex() + control_file = canary.root / "public" / self.control_name + control_file.write_text(self.control_token) + control_file.chmod(0o600) self.placeholders = {"served_root": "/fixture/public"} return {"SERVED_ROOT": "/fixture/public"} + def control(self) -> ControlProbe | None: + return ControlProbe(method="GET", path="/" + self.control_name, token=self.control_token) + class SQLOracle(OutputOracle): def prepare(self, canary: Canary) -> dict[str, str]: diff --git a/src/openultrasast/search/verify.py b/src/openultrasast/search/verify.py index 5fc7aaf..041a88e 100644 --- a/src/openultrasast/search/verify.py +++ b/src/openultrasast/search/verify.py @@ -67,6 +67,7 @@ class SideRecord: exit_code: int | None = None stderr: str = "" stdout: str = "" + control_observed: bool | None = None @dataclass(frozen=True) @@ -285,6 +286,7 @@ def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExe private_values: list[str] = [] disk_peak = 0 + control_observed: bool | None = None root: Path | None = None def timed(stage: str, operation: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: @@ -315,13 +317,14 @@ def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExe result.stderr if result is not None else app_diagnostics.get("stderr", ""), maximum=1500, private=tuple(private_values) ), stdout=diagnostic(getattr(result, "stdout", app_diagnostics.get("stdout", "")), maximum=1500, private=tuple(private_values)), + control_observed=control_observed, ) try: schema = load_demo(demo) originals = (side.checkout, *side.dependencies, demo, *((side.products,) if side.products else ())) initial = tuple(_tree(p) for p in originals) - for _ in range(1 if task_boundary else 3): + for run_index in range(1 if task_boundary else 3): tick = time.monotonic() with tempfile.TemporaryDirectory(prefix="ousast-proof-") as temporary: root = Path(temporary) @@ -475,6 +478,44 @@ def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExe captures: dict[str, str] = {} if schema["start"]["mode"] == "http": + # Positive control: on the first run, a benign in-root request + # for a verifier-planted marker, run as its OWN app start and + # fully guarded so it can only annotate, never change the + # verdict. Its marker returning proves the interface serves the + # fixture; its absence (or any control failure) is recorded, not + # trusted, since a valid app may answer only its own routes. + probe = oracle.control() + if run_index == 0 and probe is not None: + control_schema = { + **schema, + "steps": [ + { + "type": "http", + "method": probe.method, + "path": probe.path, + **({"body": probe.body} if probe.body else {}), + } + ], + } + try: + control_outputs = timed( + "run", + _http_steps, + control_schema, + command, + job, + app_scratch, + app_mounts, + app_env, + oracle.namespace_pid, + canary.nonce, + runner, + http_ready_check, + {}, + ) + control_observed = bool(control_outputs) and probe.token in control_outputs[0] + except (CommandFailure, OSError, ValueError, AssertionError): + control_observed = False outputs = timed( "run", _http_steps, diff --git a/tests/test_search_control.py b/tests/test_search_control.py new file mode 100644 index 0000000..c876f68 --- /dev/null +++ b/tests/test_search_control.py @@ -0,0 +1,153 @@ +"""Positive-control keystone for the path oracle (HTTP mode). + +The verifier prepends a benign in-root request for a planted control marker to the +first run. Its marker returning proves the interface serves the fixture +(``control_observed`` True); its absence is recorded, not trusted as a verdict, +because a valid app may answer only its own routes. A crash on the shared app start +is caught by the existing could_not_run path. These tests fake the HTTP driver so no +real sandbox is needed; they exercise the real PathOracle and the real _side wiring.""" + +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from openultrasast.search import verify as verify_module +from openultrasast.search.oracles import Canary, ControlProbe, PathOracle +from openultrasast.search.verify import CommandFailure, Side, VerificationRecord, _side, verify + + +def _http_demo(tmp_path: Path) -> tuple[Path, Side]: + """A minimal HTTP path demo plus a checkout whose entrypoint exists.""" + checkout = tmp_path / "checkout" + checkout.mkdir() + (checkout / "app.js").write_text("// entrypoint presence only; the driver is faked\n") + demo_dir = tmp_path / "demo" + demo_dir.mkdir() + schema = { + "oracle": "path", + "build": {"recipe": "none", "arguments": []}, + "start": {"runtime": "node", "path": "app.js", "arguments": ["--port", "8080"], "mode": "http", "port": 8080}, + "steps": [{"type": "http", "method": "GET", "path": "/read"}], + } + (demo_dir / "demo.json").write_text(json.dumps(schema)) + return demo_dir, Side(checkout=checkout) + + +def _spy_oracle(monkeypatch): + """Keep a handle on the real PathOracle _side builds, so a fake driver can echo + its control token and the nonce the way a working app would.""" + holder: dict[str, PathOracle] = {} + real = verify_module.oracle_for + + def spy(family, browser=None): + oracle = real(family, browser) + if isinstance(oracle, PathOracle): + holder["oracle"] = oracle + return oracle + + monkeypatch.setattr(verify_module, "oracle_for", spy) + return holder + + +def _is_control_run(schema): + return schema["steps"][0]["path"] == "/control" + + +def test_path_oracle_plants_control_distinct_from_secret(tmp_path): + oracle = PathOracle() + canary = Canary.fresh(tmp_path / "fixture") + oracle.prepare(canary) + probe = oracle.control() + assert isinstance(probe, ControlProbe) + assert probe.path == "/control" + control_file = canary.root / "public" / "control" + assert control_file.read_text() == probe.token + assert probe.token != canary.nonce # the control is benign, never the secret + assert (canary.root / "canary").read_text() == canary.nonce # secret stays outside served root + + +def test_control_marker_observed_sets_flag(tmp_path, monkeypatch): + demo_dir, side = _http_demo(tmp_path) + holder = _spy_oracle(monkeypatch) + seen = {"control_runs": 0} + + def fake_http_steps(schema, command, job, scratch, mounts, env, ns, nonce, runner, ready_check, diagnostics): + if _is_control_run(schema): + seen["control_runs"] += 1 + return [holder["oracle"].control().token] # the control's own app start returns the marker + return [nonce] # the attack start reflects the effect + + monkeypatch.setattr(verify_module, "_http_steps", fake_http_steps) + record = _side(side, demo_dir, "path", 30, None) + assert record.outcome == "observed" + assert len(record.runs) == 3 and all(r.observed for r in record.runs) + assert record.control_observed is True + assert seen["control_runs"] == 1 # the control runs once, on the first run only + + +def test_absent_control_marker_does_not_block(tmp_path, monkeypatch): + demo_dir, side = _http_demo(tmp_path) + _spy_oracle(monkeypatch) + + def fake_http_steps(schema, command, job, scratch, mounts, env, ns, nonce, runner, ready_check, diagnostics): + if _is_control_run(schema): + return ["a custom-route body with no control marker"] + return [nonce] + + monkeypatch.setattr(verify_module, "_http_steps", fake_http_steps) + record = _side(side, demo_dir, "path", 30, None) + # A valid custom-route app that never serves /control must NOT be failed: + # the effect is still observed and only the annotation records the miss. + assert record.outcome == "observed" + assert len(record.runs) == 3 and all(r.observed for r in record.runs) + assert record.control_observed is False + + +def test_control_failure_does_not_block_a_working_attack(tmp_path, monkeypatch): + # The control's own app start fails (timeout/closed/oversize), but the attack + # route works. The verdict must stand; only the annotation records the miss. + demo_dir, side = _http_demo(tmp_path) + _spy_oracle(monkeypatch) + + def fake_http_steps(schema, command, job, scratch, mounts, env, ns, nonce, runner, ready_check, diagnostics): + if _is_control_run(schema): + raise CommandFailure("run", 1, "control route hung", "HTTP application/steps failed") + return [nonce] + + monkeypatch.setattr(verify_module, "_http_steps", fake_http_steps) + record = _side(side, demo_dir, "path", 30, None) + assert record.outcome == "observed" + assert len(record.runs) == 3 and all(r.observed for r in record.runs) + assert record.control_observed is False + + +def test_control_crash_is_could_not_run(tmp_path, monkeypatch): + demo_dir, side = _http_demo(tmp_path) + _spy_oracle(monkeypatch) + + def fake_http_steps(schema, command, job, scratch, mounts, env, ns, nonce, runner, ready_check, diagnostics): + raise CommandFailure("start", -9, "TypeError: mime.lookup is not a function", "HTTP application/steps failed") + + monkeypatch.setattr(verify_module, "_http_steps", fake_http_steps) + record = _side(side, demo_dir, "path", 30, None) + assert record.outcome == "could_not_run" + assert record.runs == () + + +def test_verify_never_demonstrates_when_app_crashes(tmp_path, monkeypatch): + demo_dir, affected = _http_demo(tmp_path) + safe = Side(checkout=affected.checkout) + _spy_oracle(monkeypatch) + monkeypatch.setattr(verify_module._sandbox, "isolation_check", lambda **kwargs: None) + monkeypatch.setattr(verify_module._sandbox, "isolation_mode", lambda: "userns") + + def fake_http_steps(schema, command, job, scratch, mounts, env, ns, nonce, runner, ready_check, diagnostics): + raise CommandFailure("start", -9, "boom", "HTTP application/steps failed") + + monkeypatch.setattr(verify_module, "_http_steps", fake_http_steps) + record = verify(affected, safe, demo_dir, "path", timeout_seconds=30) + assert isinstance(record, VerificationRecord) + assert record.outcome != "demonstrated" + assert record.outcome in ("could_not_run", "inconclusive")