From 742469d6711d56f1df93e6ff834e322cf46e48cb Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Mon, 5 Oct 2026 14:09:48 +0200 Subject: [PATCH] search-with-proof group 2: verifier with owned oracles and differential The verifier, not the artefact, observes the effect: SQL, path, command and SSRF oracles owned by the verifier (XSS needs a browser executor, else no_oracle); artefact exit code and output ignored; immutable checkout and dependencies; 3 runs per side; one-sided build or readiness failure is inconclusive. An isolation failure raises IsolationUnavailable instead of reading as could_not_build (the NPROC limit now applies inside the user namespace). 35 tests, including gaming attempts and 300 host processes. Co-Authored-By: Claude Opus 5.5 --- .kiro/specs/search-with-proof/tasks.md | 19 +- .../measurements/2026-09-08-module-audit.json | 38 +- src/openultrasast/search/__init__.py | 1 + src/openultrasast/search/_sandbox.py | 148 +++++++ src/openultrasast/search/oracles.py | 189 ++++++++ src/openultrasast/search/verify.py | 257 +++++++++++ tests/fixtures/search/app.py | 33 ++ tests/test_search_verify.py | 411 ++++++++++++++++++ 8 files changed, 1089 insertions(+), 7 deletions(-) create mode 100644 src/openultrasast/search/__init__.py create mode 100644 src/openultrasast/search/_sandbox.py create mode 100644 src/openultrasast/search/oracles.py create mode 100644 src/openultrasast/search/verify.py create mode 100644 tests/fixtures/search/app.py create mode 100644 tests/test_search_verify.py 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"]