diff --git a/.kiro/specs/search-with-proof/tasks.md b/.kiro/specs/search-with-proof/tasks.md
index c07afb3..71d2ce0 100644
--- a/.kiro/specs/search-with-proof/tasks.md
+++ b/.kiro/specs/search-with-proof/tasks.md
@@ -9,15 +9,15 @@ gates, runs every networked or paid step.
injection), and whether a no-egress task class exists for verify runs. _Requirements: 5.1, 3.2_
- [ ] 1.2 Record the answers in `kube-ax-cluster` memory and design.md; no key value anywhere. _Requirements: 5.2_
-- [ ] 2. Verifier and oracles (no model calls)
- - [ ] 2.1 `src/openultrasast/search/verify.py`: fresh-sandbox runner per side; build, readiness check, read-only
+- [x] 2. Verifier and oracles (no model calls)
+ - [x] 2.1 `src/openultrasast/search/verify.py`: fresh-sandbox runner per side; build, readiness check, read-only
checkout and dependencies, untracked-file and runtime-patch refusal, 3 runs per side, consistency check,
outcomes `demonstrated|inconclusive|could_not_build|could_not_run|no_oracle`. Artefact exit code ignored.
_Requirements: 3.1-3.4_
- - [ ] 2.2 Oracles owned by the verifier: SQL (database fixture + query log), path (file-access watch on a planted
+ - [x] 2.2 Oracles owned by the verifier: SQL (database fixture + query log), path (file-access watch on a planted
canary), XSS (headless browser execution check), SSRF (in-sandbox listener), command injection (process watch).
Observers unreachable from the artefact. _Requirements: 3.1, 3.2, 3.5_
- - [ ] 2.3 Unit tests with hand-written vulnerable/fixed toy apps per oracle, including gaming attempts (reading the
+ - [x] 2.3 Unit tests with hand-written vulnerable/fixed toy apps per oracle, including gaming attempts (reading the
canary, revision sniffing, patching a dependency, contacting the listener directly): each must not yield
`demonstrated`. _Requirements: 3.2, 3.3_
@@ -54,3 +54,14 @@ gates, runs every networked or paid step.
_Requirements: 6.1, 6.4_
- [ ] 6.4 Exit decision against 6.2 and report to the maintainer; on pass, spec tasks for qualification and the
unseen-pool arm. _Requirements: 6.2_
+
+## Implementation Notes
+
+- Group 2 (2026-10-05): independently reviewed; `tests/test_search_verify.py` has 32 passing tests and one
+ skipped SSRF integration test because this host forbids even localhost sockets. The namespace handoff and
+ listener request parsing are tested separately; the full SSRF effect still needs the task 3.2 probe before
+ claiming that oracle operational. The module audit and neighbouring sandbox/regression tests pass.
+- The local lane requires Linux Bubblewrap and util-linux, reuses `SandboxJob`/`SandboxResult`, and fails closed
+ without isolation. Artefacts receive a batch CLI `TARGET` bridge and private scratch; HTTP apps need a trusted
+ CLI adapter. Only trusted `Side.build_command` outputs can enter the app runtime (read-only `/build`);
+ artefact `build.sh` writes scratch only. XSS requires a trusted browser executor, otherwise `no_oracle`.
diff --git a/benchmarks/measurements/2026-09-08-module-audit.json b/benchmarks/measurements/2026-09-08-module-audit.json
index b70b7d6..767e64a 100644
--- a/benchmarks/measurements/2026-09-08-module-audit.json
+++ b/benchmarks/measurements/2026-09-08-module-audit.json
@@ -1,7 +1,7 @@
{
- "modules": 159,
+ "modules": 163,
"counts": {
- "load_bearing": 154,
+ "load_bearing": 158,
"standalone_capability": 5
},
"orphaned": [],
@@ -1553,7 +1553,9 @@
"regress.candidate",
"regress.recipes",
"regress.safety",
- "regress.verdict"
+ "regress.verdict",
+ "search._sandbox",
+ "search.verify"
]
},
{
@@ -1592,6 +1594,36 @@
"scoring"
]
},
+ {
+ "module": "search",
+ "classification": "load_bearing",
+ "reason": "reachable from gate, map_gate, pair_gate, cli",
+ "importers": [
+ "search.verify"
+ ]
+ },
+ {
+ "module": "search._sandbox",
+ "classification": "load_bearing",
+ "reason": "reachable from gate, map_gate, pair_gate, cli",
+ "importers": [
+ "search.verify"
+ ]
+ },
+ {
+ "module": "search.oracles",
+ "classification": "load_bearing",
+ "reason": "python -m openultrasast.search.oracles",
+ "importers": [
+ "search.verify"
+ ]
+ },
+ {
+ "module": "search.verify",
+ "classification": "load_bearing",
+ "reason": "python -m openultrasast.search.verify",
+ "importers": []
+ },
{
"module": "semantic",
"classification": "load_bearing",
diff --git a/src/openultrasast/search/__init__.py b/src/openultrasast/search/__init__.py
new file mode 100644
index 0000000..ff712e4
--- /dev/null
+++ b/src/openultrasast/search/__init__.py
@@ -0,0 +1 @@
+"""Verifier-owned differential demonstrations (search-with-proof Req 3)."""
diff --git a/src/openultrasast/search/_sandbox.py b/src/openultrasast/search/_sandbox.py
new file mode 100644
index 0000000..97f7894
--- /dev/null
+++ b/src/openultrasast/search/_sandbox.py
@@ -0,0 +1,148 @@
+"""Local CLI lane using the existing sandbox job/result contract.
+
+Bubblewrap is required, never emulated with chmod. Each process sees only the
+immutable runtime, declared inputs, and its own scratch. No host /proc or home.
+The seccomp filter denies sockets for artefacts and builds, even on hosts where
+creating a network namespace is unavailable. SSRF apps join the oracle's private
+loopback namespace. This is a Linux lane; other hosts fail closed.
+"""
+
+from __future__ import annotations
+
+import os
+import platform
+import resource
+import signal
+import struct
+import subprocess
+import tempfile
+from contextlib import suppress
+from pathlib import Path
+
+from ..sandbox import SandboxJob, SandboxResult
+
+
+class IsolationUnavailable(RuntimeError):
+ """The isolation instrument failed, so no product outcome can be recorded."""
+
+
+def isolation_check(*, timeout_seconds: int = 5) -> None:
+ """Prove a trivial job can read its input through the real sandbox path."""
+ try:
+ with tempfile.TemporaryDirectory(prefix="isolation-check-") as directory:
+ root = Path(directory)
+ (root / "probe").write_text("isolation-ready\n")
+ scratch = root / "scratch"
+ scratch.mkdir()
+ job = SandboxJob("", ("/bin/cat", "/workspace/probe"), root, {}, timeout_seconds, 256, 128)
+ result = run(job, scratch=scratch)
+ except (OSError, subprocess.SubprocessError) as exc:
+ raise IsolationUnavailable(f"sandbox isolation preflight could not start: {exc}") from exc
+ if result.exit_code or result.timed_out or result.stdout != "isolation-ready\n":
+ raise IsolationUnavailable(
+ f"sandbox isolation preflight failed (exit={result.exit_code}, timeout={result.timed_out}): {result.stderr}"
+ )
+
+
+def _filter(network: bool) -> bytes:
+ # Audit architecture check also excludes x32 syscall-number bypasses.
+ if platform.machine() == "x86_64":
+ arch, forbidden = 0xC000003E, [101, 165, 166, 272, 308, 321]
+ sockets = [41, 42, 43, 49, 50, 53, 288]
+ elif platform.machine() == "aarch64":
+ arch, forbidden = 0xC00000B7, [39, 40, 97, 117, 268, 280]
+ sockets = [198, 199, 200, 201, 202, 203, 242]
+ else:
+ raise OSError("unsupported seccomp architecture")
+ ins = [(0x20, 0, 0, 4), (0x15, 1, 0, arch), (6, 0, 0, 0x80000000), (0x20, 0, 0, 0), (0x35, 0, 1, 0x40000000), (6, 0, 0, 0x80000000)]
+ for number in forbidden + ([] if network else sockets):
+ ins.extend([(0x15, 0, 1, number), (6, 0, 0, 0x50001)])
+ ins.append((6, 0, 0, 0x7FFF0000))
+ return b"".join(struct.pack("HBBI", *i) for i in ins)
+
+
+def run(
+ job: SandboxJob,
+ *,
+ scratch: Path,
+ mounts: dict[str, Path] | None = None,
+ env: dict[str, str] | None = None,
+ namespace_pid: int | None = None,
+) -> SandboxResult:
+ """Run with bounded files, memory, processes, and time; no inherited env."""
+
+ def limits() -> None:
+ resource.setrlimit(resource.RLIMIT_FSIZE, (1024 * 1024, 1024 * 1024))
+ resource.setrlimit(resource.RLIMIT_AS, (job.memory_mb * 1024**2,) * 2)
+ resource.setrlimit(resource.RLIMIT_CPU, (job.timeout_seconds + 1,) * 2)
+
+ argv = ["bwrap", "--die-with-parent", "--new-session", "--unshare-pid", "--unshare-ipc", "--unshare-uts"]
+ if namespace_pid is None:
+ argv += ["--unshare-user"]
+ else:
+ argv = [
+ "/usr/bin/nsenter",
+ "--preserve-credentials",
+ f"--user=/proc/{namespace_pid}/ns/user",
+ f"--net=/proc/{namespace_pid}/ns/net",
+ "--",
+ *argv,
+ ]
+ argv += ["--cap-drop", "ALL", "--clearenv"]
+ for path in ("/usr", "/lib", "/lib64", "/bin"):
+ if Path(path).exists():
+ argv += ["--ro-bind", path, path]
+ argv += ["--ro-bind", str(job.repo_root), "/workspace", "--bind", str(scratch), "/scratch"]
+ for target, source in (mounts or {}).items():
+ argv += ["--ro-bind", str(source), target]
+ argv += [
+ "--remount-ro",
+ "/",
+ "--chdir",
+ "/scratch",
+ "--setenv",
+ "PATH",
+ "/usr/bin:/bin",
+ "--setenv",
+ "HOME",
+ "/scratch",
+ "--setenv",
+ "TMPDIR",
+ "/scratch",
+ ]
+ for key, value in (env or {}).items():
+ argv += ["--setenv", key, value]
+ with tempfile.TemporaryFile() as policy, tempfile.TemporaryFile() as out, tempfile.TemporaryFile() as err:
+ policy.write(_filter(namespace_pid is not None))
+ policy.seek(0)
+ # Set NPROC after entering the user namespace (per-namespace since Linux 5.14),
+ # so the host uid's existing processes cannot prevent bwrap from starting.
+ argv += [
+ "--seccomp", str(policy.fileno()), "--",
+ "/usr/bin/prlimit", f"--nproc={job.pids_limit}:{job.pids_limit}", "--", *job.command,
+ ]
+ proc = subprocess.Popen(
+ argv,
+ stdout=out,
+ stderr=err,
+ pass_fds=(policy.fileno(),),
+ start_new_session=True,
+ preexec_fn=limits,
+ env={"PATH": "/usr/bin:/bin"},
+ )
+ timed_out = False
+ try:
+ proc.wait(timeout=job.timeout_seconds)
+ except subprocess.TimeoutExpired:
+ timed_out = True
+ os.killpg(proc.pid, signal.SIGKILL)
+ proc.wait()
+ finally:
+ # Also terminate descendants left behind after a successful shell exit.
+ with suppress(ProcessLookupError):
+ os.killpg(proc.pid, signal.SIGKILL)
+ out.seek(0)
+ err.seek(0)
+ return SandboxResult(
+ proc.returncode, out.read(1024 * 1024).decode(errors="replace"), err.read(1024 * 1024).decode(errors="replace"), timed_out
+ )
diff --git a/src/openultrasast/search/oracles.py b/src/openultrasast/search/oracles.py
new file mode 100644
index 0000000..ff2e9b9
--- /dev/null
+++ b/src/openultrasast/search/oracles.py
@@ -0,0 +1,189 @@
+"""Owned canaries. Only app captures enter these oracles, never attack output.
+
+The attacker may supply CLI arguments through TARGET. Fixtures and app config
+are mounted only in the app sandbox. Nonces are fresh per run and absent from
+artefact env. A browser executor is trusted verifier code, never a demo script.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import secrets
+import selectors
+import sqlite3
+import subprocess
+from collections.abc import Callable
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Protocol
+
+
+@dataclass(frozen=True)
+class Canary:
+ root: Path
+ nonce: str
+
+ @classmethod
+ def fresh(cls, root: Path) -> Canary:
+ root.mkdir()
+ return cls(root, secrets.token_hex(24))
+
+
+class Oracle(Protocol):
+ namespace_pid: int | None
+
+ def prepare(self, canary: Canary) -> dict[str, str]: ...
+ def capture(self, output: str) -> None: ...
+ def observe(self) -> tuple[bool, str]: ...
+ def close(self) -> None: ...
+
+
+class OutputOracle:
+ namespace_pid: int | None = None
+
+ def prepare(self, canary: Canary) -> dict[str, str]:
+ self.canary = canary
+ self.seen = False
+ return {}
+
+ def capture(self, output: str) -> None:
+ self.seen |= self.canary.nonce in output
+
+ 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 close(self) -> None:
+ pass
+
+
+class PathOracle(OutputOracle):
+ def prepare(self, canary: Canary) -> dict[str, str]:
+ super().prepare(canary)
+ (canary.root / "public").mkdir()
+ (canary.root / "canary").write_text(canary.nonce)
+ return {"SERVED_ROOT": "/fixture/public"}
+
+
+class SQLOracle(OutputOracle):
+ def prepare(self, canary: Canary) -> dict[str, str]:
+ super().prepare(canary)
+ with sqlite3.connect(canary.root / "fixture.db") as db:
+ db.execute("CREATE TABLE records (name TEXT, value TEXT, public INTEGER)")
+ db.executemany("INSERT INTO records VALUES (?, ?, ?)", [("ordinary", "public value", 1), ("secret", canary.nonce, 0)])
+ return {"DATABASE": "/fixture/fixture.db"}
+
+
+class CommandOracle(OutputOracle):
+ def prepare(self, canary: Canary) -> dict[str, str]:
+ super().prepare(canary)
+ # Marker is in app-only writable scratch, inaccessible in artefact sandbox.
+ self.marker = canary.root / canary.nonce
+ return {"PROOF_MARKER": "/scratch/" + canary.nonce}
+
+ def observe(self) -> tuple[bool, str]:
+ observed = self.marker.is_file() and not self.marker.is_symlink()
+ return observed, "injected command marker exists" if observed else "no command marker"
+
+
+class SSRFOracle(OutputOracle):
+ """Separate listener process and private loopback namespace; no external egress."""
+
+ process: subprocess.Popen[str] | None = None
+
+ def prepare(self, canary: Canary) -> dict[str, str]:
+ super().prepare(canary)
+ self.process = subprocess.Popen(
+ [
+ "/usr/bin/unshare",
+ "--user",
+ "--map-root-user",
+ "--net",
+ "/usr/bin/python3",
+ "-I",
+ str(Path(__file__).resolve()),
+ "--listener",
+ canary.nonce,
+ ],
+ stdout=subprocess.PIPE,
+ stderr=subprocess.DEVNULL,
+ text=True,
+ env={"PATH": "/usr/bin:/bin"},
+ )
+ assert self.process.stdout is not None
+ with selectors.DefaultSelector() as selector:
+ selector.register(self.process.stdout, selectors.EVENT_READ)
+ if not selector.select(3):
+ raise OSError("listener namespace readiness timed out")
+ line = self.process.stdout.readline()
+ if not line:
+ raise OSError("private loopback listener unavailable")
+ self.namespace_pid, port = json.loads(line)
+ return {"CALLBACK_URL": f"http://127.0.0.1:{port}/{canary.nonce}"} # loopback: oracle listener inside the private sandbox namespace
+
+ def observe(self) -> tuple[bool, str]:
+ assert self.process is not None and self.process.stdout is not None
+ with selectors.DefaultSelector() as selector:
+ selector.register(self.process.stdout, selectors.EVENT_READ)
+ observed = bool(selector.select(0.2)) and self.process.stdout.readline().strip() == "observed"
+ return observed, "nonce request received" if observed else "no nonce request"
+
+ def close(self) -> None:
+ if self.process is not None:
+ self.process.terminate()
+ self.process.wait(timeout=3)
+ if self.process.stdout:
+ self.process.stdout.close()
+
+
+BrowserExecutor = Callable[[str, str], bool]
+
+
+class XSSOracle(OutputOracle):
+ def __init__(self, executor: BrowserExecutor) -> None:
+ self.executor = executor
+
+ def capture(self, output: str) -> None:
+ self.seen |= self.executor(output, self.canary.nonce)
+
+
+def oracle_for(family: str, browser: BrowserExecutor | None = None) -> Oracle | None:
+ factories = {"path": PathOracle, "sql": SQLOracle, "command": CommandOracle, "ssrf": SSRFOracle}
+ if family == "xss":
+ return XSSOracle(browser) if browser else None
+ factory = factories.get(family)
+ return factory() if factory else None
+
+
+def _listen(nonce: str) -> None:
+ import fcntl
+ import socket
+ import struct
+ from http.server import BaseHTTPRequestHandler, HTTPServer
+
+ # unshare created only loopback; bring it up without iproute2 or any egress.
+ with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
+ fcntl.ioctl(sock, 0x8914, struct.pack("16sH14s", b"lo", 0x49, b""))
+
+ class Handler(BaseHTTPRequestHandler):
+ def do_GET(self) -> None:
+ if self.path.split("?", 1)[0] == "/" + nonce:
+ print("observed", flush=True)
+ self.send_response(200)
+ self.end_headers()
+ self.wfile.write(b"ok")
+
+ do_POST = do_PUT = do_PATCH = do_DELETE = do_HEAD = do_OPTIONS = do_TRACE = do_CONNECT = do_GET
+
+ def log_message(self, format: str, *args: object) -> None:
+ pass
+
+ with HTTPServer(("127.0.0.1", 0), Handler) as server:
+ print(json.dumps([os.getpid(), server.server_port]), flush=True)
+ server.serve_forever()
+
+
+if __name__ == "__main__":
+ import sys
+
+ _listen(sys.argv[2])
diff --git a/src/openultrasast/search/verify.py b/src/openultrasast/search/verify.py
new file mode 100644
index 0000000..d4c2ffb
--- /dev/null
+++ b/src/openultrasast/search/verify.py
@@ -0,0 +1,257 @@
+"""Offline differential verification through an isolated, batch CLI interface.
+
+Side.command and readiness_args are trusted integration configuration, not demo
+contents. TARGET is an executable CLI bridge: each invocation queues arguments
+for the app. It deliberately has no synchronous response channel. Alternatively
+request.json contains one argument vector. HTTP apps need a trusted CLI adapter;
+this lane never treats an artefact's stdout, exit status, or claimed result as
+proof. Build scripts validate the immutable checkout and may write scratch only;
+only Side.build_command (trusted integration config) can produce /build runtime outputs.
+
+Run: python -m openultrasast.search.verify vulnerable fixed demo path --entry app.py
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import shutil
+import stat
+import tempfile
+import time
+from dataclasses import asdict, dataclass
+from pathlib import Path
+
+from ..sandbox import SandboxJob
+from . import _sandbox
+from .oracles import BrowserExecutor, Canary, oracle_for
+
+_BRIDGE = """#!/usr/bin/python3 -I
+import json, sys
+with open('/scratch/requests.jsonl', 'a') as stream:
+ stream.write(json.dumps(sys.argv[1:]) + '\\n')
+"""
+
+
+@dataclass(frozen=True)
+class Side:
+ checkout: Path
+ command: tuple[str, ...]
+ readiness_args: tuple[str, ...] = ("--ready",)
+ dependencies: tuple[Path, ...] = ()
+ build_command: tuple[str, ...] = ()
+
+
+@dataclass(frozen=True)
+class RunObservation:
+ observed: bool
+ evidence: str
+ elapsed_seconds: float
+
+
+@dataclass(frozen=True)
+class SideRecord:
+ outcome: str
+ runs: tuple[RunObservation, ...]
+ elapsed_seconds: float
+ reason: str
+
+
+@dataclass(frozen=True)
+class VerificationRecord:
+ outcome: str
+ sides: tuple[SideRecord, ...]
+ elapsed_seconds: float
+ reason: str
+
+
+def _tree(root: Path) -> str:
+ """Read every file; hash names, modes and bytes, refusing links/devices."""
+ digest = hashlib.sha256()
+ if not root.is_dir() or root.is_symlink():
+ raise ValueError("input must be a real directory")
+ for path in [root, *sorted(root.rglob("*"))]:
+ mode = path.lstat().st_mode
+ if not (stat.S_ISREG(mode) or stat.S_ISDIR(mode)):
+ raise ValueError(f"links and special files refused: {path.name}")
+ digest.update(str(path.relative_to(root)).encode())
+ digest.update(str(mode).encode())
+ if path.is_file():
+ digest.update(path.read_bytes())
+ return digest.hexdigest()
+
+
+def _requests(scratch: Path, demo: Path) -> list[list[str]]:
+ path = demo / "request.json"
+ if path.is_file():
+ values = [json.loads(path.read_text())]
+ else:
+ queue = scratch / "requests.jsonl"
+ if not queue.exists():
+ return []
+ if queue.is_symlink() or not queue.is_file() or queue.stat().st_size > 65536:
+ raise ValueError("invalid request queue")
+ values = [json.loads(line) for line in queue.read_text().splitlines()]
+ if len(values) > 32 or any(
+ not isinstance(v, list) or not all(isinstance(a, str) and "\0" not in a for a in v) or sum(map(len, v)) > 16384 for v in values
+ ):
+ raise ValueError("invalid or excessive CLI requests")
+ return values
+
+
+def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExecutor | None) -> SideRecord:
+ started = time.monotonic()
+ observations: list[RunObservation] = []
+
+ def record(outcome: str, reason: str) -> SideRecord:
+ return SideRecord(outcome, tuple(observations), time.monotonic() - started, reason)
+
+ try:
+ originals = (side.checkout, *side.dependencies, demo)
+ initial = tuple(_tree(p) for p in originals)
+ for _ in range(3):
+ tick = time.monotonic()
+ with tempfile.TemporaryDirectory(prefix="ousast-proof-") as temporary:
+ root = Path(temporary)
+ checkout = shutil.copytree(side.checkout, root / "checkout")
+ artefact = shutil.copytree(demo, root / "demo")
+ dependencies = [shutil.copytree(p, root / f"dep-{i}") for i, p in enumerate(side.dependencies)]
+ immutable = (checkout, *dependencies, artefact)
+ before = tuple(_tree(p) for p in immutable)
+ mounts = {f"/deps/{i}": p for i, p in enumerate(dependencies)}
+
+ def job(command: tuple[str, ...], checkout: Path = checkout) -> SandboxJob:
+ return SandboxJob("", command, checkout, {}, timeout, 256, 128)
+
+ products = root / "products"
+ products.mkdir()
+ if side.build_command:
+ compiled = _sandbox.run(job(side.build_command), scratch=products, mounts=mounts)
+ if compiled.exit_code or compiled.timed_out:
+ return record("could_not_build", "trusted build failed: " + compiled.stderr[:300])
+ product_hash = _tree(products)
+ mounts["/build"] = products
+ build_scratch = root / "build"
+ build_scratch.mkdir()
+ built = _sandbox.run(job(("/bin/sh", "/demo/build.sh")), scratch=build_scratch, mounts={**mounts, "/demo": artefact})
+ if built.exit_code or built.timed_out:
+ return record("could_not_build", "build failed: " + built.stderr[:300])
+ if tuple(_tree(p) for p in immutable) != before or _tree(products) != product_hash:
+ return record("inconclusive", "build changed immutable inputs")
+
+ oracle = oracle_for(family, browser)
+ assert oracle is not None
+ try:
+ canary = Canary.fresh(root / "fixture")
+ app_env = oracle.prepare(canary)
+ app_scratch = canary.root if family == "command" else root / "app"
+ app_scratch.mkdir(exist_ok=True)
+ app_mounts = {**mounts, "/fixture": canary.root}
+ ready = _sandbox.run(
+ job((*side.command, *side.readiness_args)),
+ scratch=app_scratch,
+ mounts=app_mounts,
+ env=app_env,
+ namespace_pid=oracle.namespace_pid,
+ )
+ if ready.exit_code or ready.timed_out:
+ return record("could_not_run", "app readiness failed: " + ready.stderr[:300])
+ if family == "ssrf" and oracle.observe()[0]:
+ return record("inconclusive", "readiness triggered the SSRF oracle")
+ # Readiness cannot supply the attack's evidence or leave a command marker.
+ if family == "command" and any(canary.root.iterdir()):
+ return record("inconclusive", "readiness modified command fixture")
+ attack_scratch = root / "attack"
+ attack_scratch.mkdir()
+ if (artefact / "attack.sh").is_file():
+ bridge = root / "target"
+ bridge.write_text(_BRIDGE)
+ bridge.chmod(0o555)
+ attacker_input = root / "attacker-input"
+ attacker_input.mkdir()
+ attack = _sandbox.run(
+ job(("/bin/sh", "/demo/attack.sh"), checkout=attacker_input),
+ scratch=attack_scratch,
+ mounts={"/demo": artefact, "/target": bridge},
+ env={"TARGET": "/target", "SCRATCH": "/scratch"},
+ )
+ # Output/status are never evidence. Timeout and explicit filesystem
+ # denials veto verification; read-only mounts enforce the boundary.
+ if attack.timed_out:
+ return record("inconclusive", "attack timed out")
+ if "Read-only file system" in attack.stderr or "Permission denied" in attack.stderr:
+ return record("inconclusive", "attacker boundary refused")
+ for arguments in _requests(attack_scratch, artefact):
+ result = _sandbox.run(
+ job((*side.command, *arguments)),
+ scratch=app_scratch,
+ mounts=app_mounts,
+ env=app_env,
+ namespace_pid=oracle.namespace_pid,
+ )
+ if result.timed_out:
+ return record("inconclusive", "app attack timed out")
+ oracle.capture(result.stdout)
+ observed, evidence = oracle.observe()
+ if (
+ tuple(_tree(p) for p in immutable) != before
+ or tuple(_tree(p) for p in originals) != initial
+ or _tree(products) != product_hash
+ ):
+ return record("inconclusive", "immutable input or dependency changed")
+ observations.append(RunObservation(observed, evidence, time.monotonic() - tick))
+ finally:
+ oracle.close()
+ return record("observed", "three fresh runs completed")
+ except (OSError, ValueError, AssertionError) as exc:
+ return record("could_not_run", f"verification unavailable or refused: {exc}")
+
+
+def verify(
+ affected: Side,
+ safe: Side,
+ artefact: Path,
+ family: str,
+ *,
+ timeout_seconds: int = 5,
+ browser: BrowserExecutor | None = None,
+) -> VerificationRecord:
+ """Affected is vulnerable/head; safe is fixed/base. No negative verdict exists."""
+ started = time.monotonic()
+ if oracle_for(family, browser) is None:
+ return VerificationRecord("no_oracle", (), time.monotonic() - started, "no configured owned oracle for " + family)
+ if timeout_seconds < 1:
+ raise ValueError("timeout must be positive")
+ _sandbox.isolation_check(timeout_seconds=timeout_seconds)
+ sides = tuple(_side(s, artefact, family, timeout_seconds, browser) for s in (affected, safe))
+ if sides[1].outcome != "observed":
+ outcome, reason = "inconclusive", "safe side did not complete build/readiness/verification"
+ elif sides[0].outcome != "observed":
+ outcome, reason = sides[0].outcome, sides[0].reason
+ elif all(r.observed for r in sides[0].runs) and not any(r.observed for r in sides[1].runs):
+ outcome, reason = "demonstrated", "owned effect in 3/3 affected runs and 0/3 safe runs"
+ else:
+ outcome, reason = "inconclusive", "no consistent differential effect"
+ return VerificationRecord(outcome, sides, time.monotonic() - started, reason)
+
+
+def main() -> None:
+ import argparse
+
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("affected", type=Path)
+ parser.add_argument("safe", type=Path)
+ parser.add_argument("artefact", type=Path)
+ parser.add_argument("family")
+ parser.add_argument("--entry", required=True, help="trusted Python CLI entry relative to checkout")
+ args = parser.parse_args()
+ entry = Path(args.entry)
+ if entry.is_absolute() or ".." in entry.parts:
+ parser.error("entry must stay inside checkout")
+ command = ("/usr/bin/python3", "-I", "/workspace/" + str(entry))
+ result = verify(Side(args.affected, command), Side(args.safe, command), args.artefact, args.family)
+ print(json.dumps(asdict(result), indent=2))
+
+
+if __name__ == "__main__":
+ main()
diff --git a/tests/fixtures/search/app.py b/tests/fixtures/search/app.py
new file mode 100644
index 0000000..f7db753
--- /dev/null
+++ b/tests/fixtures/search/app.py
@@ -0,0 +1,33 @@
+"""CLI toy: arguments are its sole public interface; config belongs to the app."""
+
+import os
+import pathlib
+import sqlite3
+import subprocess
+import sys
+import urllib.request
+
+fixed = pathlib.Path("/workspace/fixed").exists()
+if sys.argv[1:] == ["--ready"]:
+ sys.exit(1 if pathlib.Path("/workspace/broken").exists() else 0)
+family, payload = sys.argv[1:]
+if family == "path":
+ root = pathlib.Path(os.environ["SERVED_ROOT"])
+ path = (root / payload).resolve()
+ if not fixed or path.is_relative_to(root):
+ value = path.read_text()
+ if not pathlib.Path("/workspace/flaky").exists() or value.startswith("a"):
+ print(value)
+elif family == "sql":
+ with sqlite3.connect(os.environ["DATABASE"]) as db:
+ if fixed:
+ rows = db.execute("SELECT value FROM records WHERE public=1 AND name=?", (payload,))
+ else:
+ rows = db.execute("SELECT value FROM records WHERE public=1 AND name='" + payload + "'")
+ print(list(rows))
+elif family == "command":
+ if not fixed:
+ subprocess.run("echo " + payload, shell=True, check=False)
+elif family == "ssrf":
+ if not fixed:
+ print(urllib.request.urlopen(os.path.expandvars(payload), timeout=2).read())
diff --git a/tests/test_search_verify.py b/tests/test_search_verify.py
new file mode 100644
index 0000000..f8b029c
--- /dev/null
+++ b/tests/test_search_verify.py
@@ -0,0 +1,411 @@
+"""Owned observations and adversarial artefacts, with real isolated toy processes."""
+
+import os
+import shutil
+import signal
+import socket
+from pathlib import Path
+
+import pytest
+
+from openultrasast.search.verify import Side, verify
+
+FIXTURE = Path(__file__).parent / "fixtures/search/app.py"
+PAYLOADS = {
+ "path": "../canary",
+ "sql": "' OR public=0 --",
+ "command": '; touch "$PROOF_MARKER"',
+ "ssrf": "$CALLBACK_URL",
+}
+
+
+@pytest.fixture
+def pair(tmp_path):
+ sides = []
+ for name in ("vulnerable", "fixed"):
+ root = tmp_path / name
+ root.mkdir()
+ shutil.copyfile(FIXTURE, root / "app.py")
+ if name == "fixed":
+ (root / "fixed").touch()
+ sides.append(Side(root, ("/usr/bin/python3", "-I", "/workspace/app.py")))
+ demo = tmp_path / "demo"
+ demo.mkdir()
+ (demo / "build.sh").write_text("test -s /workspace/app.py\n")
+ return sides, demo
+
+
+def request(demo, family):
+ import json
+
+ (demo / "request.json").write_text(json.dumps([family, PAYLOADS[family]]))
+
+
+@pytest.mark.parametrize("family", ["path", "sql", "command", "ssrf"])
+def test_real_differential(pair, family):
+ if family == "ssrf":
+ try:
+ with socket.socket():
+ pass
+ except PermissionError:
+ pytest.skip("host policy prohibits even localhost sockets")
+ sides, demo = pair
+ request(demo, family)
+ result = verify(*sides, demo, family)
+ assert result.outcome == "demonstrated", result
+ assert [r.observed for r in result.sides[0].runs] == [True] * 3
+ assert [r.observed for r in result.sides[1].runs] == [False] * 3
+ assert result.elapsed_seconds > 0
+
+
+@pytest.mark.parametrize(
+ "script",
+ [
+ "cat /fixture/canary; echo success; exit 42",
+ "if test ! -f /workspace/fixed; then echo demonstrated; fi",
+ "echo hacked > /workspace/sitecustomize.py",
+ "echo hacked > /usr/lib/python3.12/sitecustomize.py",
+ "echo hacked > /deps/0/module.py",
+ "cat /proc/1/environ; echo demonstrated",
+ '/usr/bin/python3 -I -c "import socket; socket.create_connection(("127.0.0.1",12345))"',
+ ],
+)
+def test_gaming_cannot_demonstrate(pair, tmp_path, script):
+ sides, demo = pair
+ deps = tmp_path / "deps"
+ deps.mkdir()
+ (deps / "module.py").write_text("original")
+ sides = [Side(s.checkout, s.command, dependencies=(deps,)) for s in sides]
+ (demo / "attack.sh").write_text(script)
+ result = verify(*sides, demo, "path")
+ assert result.outcome != "demonstrated"
+ assert (deps / "module.py").read_text() == "original"
+ assert not (sides[0].checkout / "sitecustomize.py").exists()
+
+
+def test_attack_exit_and_stdout_ignored(pair):
+ sides, demo = pair
+ (demo / "attack.sh").write_text('"$TARGET" path ../canary\necho fake\nexit 99\n')
+ assert verify(*sides, demo, "path").outcome == "demonstrated"
+
+
+def test_safe_build_failure_is_inconclusive(pair):
+ sides, demo = pair
+ request(demo, "path")
+ (demo / "build.sh").write_text("test ! -f /workspace/fixed")
+ result = verify(*sides, demo, "path")
+ assert result.outcome == "inconclusive"
+ assert result.sides[1].outcome == "could_not_build"
+
+
+def test_readiness_failure(pair):
+ sides, demo = pair
+ request(demo, "path")
+ (sides[1].checkout / "broken").touch()
+ result = verify(*sides, demo, "path")
+ assert result.outcome == "inconclusive"
+ assert result.sides[1].outcome == "could_not_run"
+
+
+@pytest.mark.parametrize("family", ["xss", "deserialisation", "access_control", "secrets"])
+def test_no_oracle(pair, family):
+ sides, demo = pair
+ assert verify(*sides, demo, family).outcome == "no_oracle"
+
+
+def test_flaky_observation_is_inconclusive(pair, monkeypatch):
+ from openultrasast.search import oracles
+
+ nonces = iter(["a" * 48, "b" * 48, "c" * 48] * 2)
+ monkeypatch.setattr(oracles.secrets, "token_hex", lambda _: next(nonces))
+ sides, demo = pair
+ (sides[0].checkout / "flaky").touch()
+ request(demo, "path")
+ result = verify(*sides, demo, "path")
+ assert result.outcome == "inconclusive"
+ assert [r.observed for r in result.sides[0].runs] == [True, False, False]
+
+
+def test_environment_injection_stays_in_artefact(pair, monkeypatch):
+ sides, demo = pair
+ monkeypatch.setenv("PYTHONPATH", "/bogus")
+ monkeypatch.setenv("NODE_OPTIONS", "--require=/bogus")
+ monkeypatch.setenv("LD_PRELOAD", "/bogus")
+ (demo / "attack.sh").write_text(
+ 'test -z "$PYTHONPATH$NODE_OPTIONS$LD_PRELOAD" || exit 3\n'
+ 'echo "raise RuntimeError()" > /scratch/sitecustomize.py\n'
+ "export PYTHONPATH=/scratch NODE_OPTIONS=--require=/scratch/sitecustomize.py LD_PRELOAD=/bogus\n"
+ '"$TARGET" path ../canary\n'
+ )
+ assert verify(*sides, demo, "path").outcome == "demonstrated"
+
+
+def test_untracked_file_outside_scratch_refused(pair):
+ sides, demo = pair
+ (demo / "attack.sh").write_text('echo hacked > /untracked\n"$TARGET" path ../canary\n')
+ assert verify(*sides, demo, "path").outcome == "inconclusive"
+
+
+def test_unknown_app_failure_cannot_prove_effect(pair):
+ sides, demo = pair
+ (sides[0].checkout / "app.py").write_text('raise RuntimeError("no app")')
+ request(demo, "path")
+ assert verify(*sides, demo, "path").outcome == "could_not_run"
+
+
+def test_affected_build_failure(pair):
+ sides, demo = pair
+ request(demo, "path")
+ (demo / "build.sh").write_text("test -f /workspace/fixed")
+ assert verify(*sides, demo, "path").outcome == "could_not_build"
+
+
+def test_missing_isolation_fails_closed(pair, monkeypatch):
+ from openultrasast.search import _sandbox
+
+ def unavailable(*args, **kwargs):
+ raise FileNotFoundError("bwrap unavailable")
+
+ monkeypatch.setattr(_sandbox, "run", unavailable)
+ sides, demo = pair
+ request(demo, "path")
+ with pytest.raises(_sandbox.IsolationUnavailable, match="bwrap unavailable"):
+ verify(*sides, demo, "path")
+
+
+def test_bwrap_failure_raises_before_product_outcomes(pair, monkeypatch):
+ from openultrasast.sandbox import SandboxResult
+ from openultrasast.search import _sandbox
+
+ calls = []
+ stderr = "bwrap: Creating new namespace failed: Resource temporarily unavailable"
+
+ def failed(job, **kwargs):
+ calls.append(job)
+ return SandboxResult(1, "", stderr, False)
+
+ monkeypatch.setattr(_sandbox, "run", failed)
+ sides, demo = pair
+ with pytest.raises(_sandbox.IsolationUnavailable, match=stderr):
+ verify(*sides, demo, "path")
+ assert len(calls) == 1
+ assert calls[0].command == ("/bin/cat", "/workspace/probe")
+
+
+def test_verification_with_many_host_processes(pair, monkeypatch):
+ from openultrasast.search import _sandbox
+
+ checks = []
+ original = _sandbox.isolation_check
+
+ def checked(**kwargs):
+ checks.append(True)
+ return original(**kwargs)
+
+ monkeypatch.setattr(_sandbox, "isolation_check", checked)
+ sides, demo = pair
+ request(demo, "path")
+ children = []
+ try:
+ for _ in range(300):
+ pid = os.fork()
+ if pid == 0:
+ try:
+ while True:
+ signal.pause()
+ finally:
+ os._exit(0)
+ children.append(pid)
+ assert verify(*sides, demo, "path").outcome == "demonstrated"
+ assert checks == [True]
+ finally:
+ for pid in children:
+ os.kill(pid, signal.SIGKILL)
+ for pid in children:
+ os.waitpid(pid, 0)
+
+
+def test_symlink_input_refused(pair):
+ sides, demo = pair
+ (demo / "escape").symlink_to("/etc/passwd")
+ request(demo, "path")
+ assert verify(*sides, demo, "path").outcome == "inconclusive"
+
+
+def test_xss_executor_contract(tmp_path):
+ from openultrasast.search.oracles import Canary, oracle_for
+
+ calls = []
+
+ def browser(document, nonce):
+ calls.append((document, nonce))
+ return document == ""
+
+ oracle = oracle_for("xss", browser)
+ canary = Canary.fresh(tmp_path / "fixture")
+ assert oracle.prepare(canary) == {}
+ oracle.capture("")
+ assert oracle.observe()[0]
+ assert calls == [("", canary.nonce)]
+
+
+def test_six_fresh_canaries(pair, monkeypatch):
+ from openultrasast.search import oracles
+
+ nonces = []
+ prepare = oracles.PathOracle.prepare
+
+ def record(self, canary):
+ nonces.append(canary.nonce)
+ return prepare(self, canary)
+
+ monkeypatch.setattr(oracles.PathOracle, "prepare", record)
+ sides, demo = pair
+ request(demo, "path")
+ assert verify(*sides, demo, "path").outcome == "demonstrated"
+ assert len(set(nonces)) == 6
+
+
+def test_readiness_effect_cannot_prove_ssrf(pair, monkeypatch):
+ from openultrasast.search import verify as module
+ from openultrasast.search.oracles import OutputOracle
+
+ class ReadinessOracle(OutputOracle):
+ def observe(self):
+ return True, "request during readiness"
+
+ monkeypatch.setattr(module, "oracle_for", lambda *args: ReadinessOracle())
+ sides, demo = pair
+ result = module.verify(*sides, demo, "ssrf")
+ assert result.outcome == "inconclusive"
+ assert not result.sides[0].runs
+
+
+def test_dependency_hash_detects_replacement(pair, tmp_path, monkeypatch):
+ from openultrasast.search import _sandbox
+
+ original = _sandbox.run
+ dependency = tmp_path / "dependency"
+ dependency.mkdir()
+ (dependency / "module.py").write_text("original")
+ sides, demo = pair
+ sides = [Side(s.checkout, s.command, dependencies=(dependency,)) for s in sides]
+ request(demo, "path")
+
+ def changed(job, **kwargs):
+ result = original(job, **kwargs)
+ if job.command[-1] == "/demo/build.sh":
+ (kwargs["mounts"]["/deps/0"] / "module.py").write_text("replacement")
+ return result
+
+ monkeypatch.setattr(_sandbox, "run", changed)
+ result = verify(*sides, demo, "path")
+ assert result.outcome == "inconclusive"
+ assert result.sides[0].reason == "build changed immutable inputs"
+
+
+def test_trusted_build_products_are_used_readonly(pair):
+ sides, demo = pair
+ sides = [
+ Side(s.checkout, ("/usr/bin/python3", "-I", "/build/app.py"), build_command=("/bin/cp", "/workspace/app.py", "/scratch/app.py"))
+ for s in sides
+ ]
+ request(demo, "path")
+ assert verify(*sides, demo, "path").outcome == "demonstrated"
+ (demo / "build.sh").write_text("echo patched > /build/app.py")
+ assert verify(*sides, demo, "path").outcome == "inconclusive"
+
+
+def test_private_namespace_handoff_without_sockets(tmp_path):
+ import subprocess
+
+ from openultrasast.sandbox import SandboxJob
+ from openultrasast.search import _sandbox
+
+ # Proves the actual nsenter -> bwrap handoff without requiring socket().
+ process = subprocess.Popen(
+ [
+ "/usr/bin/unshare",
+ "--user",
+ "--map-root-user",
+ "--net",
+ "/usr/bin/python3",
+ "-I",
+ "-c",
+ "import os,time; print(os.getpid(), flush=True); time.sleep(10)",
+ ],
+ stdout=subprocess.PIPE,
+ text=True,
+ )
+ try:
+ pid = int(process.stdout.readline())
+ scratch = tmp_path / "scratch"
+ scratch.mkdir()
+ job = SandboxJob("", ("/usr/bin/python3", "-I", "-c", 'print("namespace-ready")'), tmp_path, {}, 3, 256, 128)
+ result = _sandbox.run(job, scratch=scratch, namespace_pid=pid)
+ assert result.exit_code == 0, result.stderr
+ assert result.stdout.strip() == "namespace-ready"
+ finally:
+ process.terminate()
+ process.wait(timeout=3)
+ process.stdout.close()
+
+
+def test_attacker_sees_only_external_bridge_and_scratch(pair):
+ sides, demo = pair
+ (demo / "attack.sh").write_text(
+ "test ! -e /workspace/app.py || exit 3\n"
+ "test ! -e /fixture/canary || exit 3\n"
+ "test ! -e /build/app.py || exit 3\n"
+ "test ! -e /deps/0 || exit 3\n"
+ 'test -z "$CALLBACK_URL$PROOF_MARKER$DATABASE$SERVED_ROOT" || exit 3\n'
+ '"$TARGET" path ../canary\n'
+ )
+ assert verify(*sides, demo, "path").outcome == "demonstrated"
+
+
+def test_listener_accepts_nonce_path_for_post_without_sockets(monkeypatch, capsys):
+ import fcntl
+ import http.server
+ import io
+
+ from openultrasast.search import oracles
+
+ class Socket:
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *args):
+ pass
+
+ class Connection:
+ def __init__(self, path):
+ self.input = io.BytesIO(b"POST " + path + b" HTTP/1.0\r\nContent-Length: 0\r\n\r\n")
+
+ def makefile(self, *args):
+ return self.input
+
+ def sendall(self, data):
+ pass
+
+ class Server:
+ server_port = 12345
+
+ def __init__(self, address, handler):
+ self.handler = handler
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *args):
+ pass
+
+ def serve_forever(self):
+ for path in (b"/wrong", b"/secret?query=1"):
+ self.handler(Connection(path), ("127.0.0.1", 1), self)
+
+ monkeypatch.setattr(socket, "socket", lambda *args: Socket())
+ monkeypatch.setattr(fcntl, "ioctl", lambda *args: None)
+ monkeypatch.setattr(http.server, "HTTPServer", Server)
+ oracles._listen("secret")
+ assert capsys.readouterr().out.splitlines()[1:] == ["observed"]