diff --git a/src/openultrasast/search/probe.py b/src/openultrasast/search/probe.py index 363f280..7167637 100644 --- a/src/openultrasast/search/probe.py +++ b/src/openultrasast/search/probe.py @@ -15,16 +15,16 @@ import subprocess import tempfile import time import urllib.request -from dataclasses import asdict +from dataclasses import asdict, replace from importlib.resources import files from pathlib import Path from typing import Any from . import _sandbox -from .demo import FAMILY_ORACLES +from .demo import FAMILY_ORACLES, load_demo from .oracles import BrowserExecutor from .task_storage import exception_reason -from .verify import Side, verify +from .verify import Side, VerificationRecord, verify, verify_side_task FAMILIES = ("sql", "path", "command", "ssrf", "xss") PAYLOADS = { @@ -105,6 +105,55 @@ def namespace_facts() -> dict[str, Any]: return facts +def _verify_task_pair(sides: tuple[Side, Side], demo: Path, family: str) -> VerificationRecord: + """Exercise the task executor on packaged toys only, not arbitrary checkouts. + + Repetitions get fresh scratch, but share this probe task's outer boundary. + This is an instrument self-test, not a fresh-task dispatcher for evidence. + """ + started = time.monotonic() + records = [] + try: + load_demo(demo) + except (OSError, ValueError) as exc: + return VerificationRecord( + "inconclusive", + (), + time.monotonic() - started, + "invalid declarative demo: " + exception_reason(exc), + isolation_mode="task-boundary", + ) + search_family = next(f for f, kinds in FAMILY_ORACLES.items() if family in kinds) + for side in sides: + runs = [verify_side_task(side, demo, search_family) for _ in range(3)] + first = next((run for run in runs if run.outcome != "observed"), runs[0]) + records.append( + replace( + first, + runs=tuple(observation for run in runs for observation in run.runs), + elapsed_seconds=sum(run.elapsed_seconds for run in runs), + build_seconds=sum(run.build_seconds for run in runs), + ready_seconds=sum(run.ready_seconds for run in runs), + run_seconds=sum(run.run_seconds for run in runs), + scratch_peak_bytes=max(run.scratch_peak_bytes for run in runs), + reason=first.reason if first.outcome != "observed" else "three probe task executions completed", + ) + ) + demonstrated = ( + all(side.outcome == "observed" and len(side.runs) == 3 for side in records) + and all(run.observed for run in records[0].runs) + and not any(run.observed for run in records[1].runs) + ) + return VerificationRecord( + "demonstrated" if demonstrated else "inconclusive", + tuple(records), + time.monotonic() - started, + "owned effect in 3/3 affected runs and 0/3 safe runs" if demonstrated else "no consistent differential effect", + isolation_mode="task-boundary", + chromium_no_sandbox=any(side.chromium_no_sandbox for side in records), + ) + + def probe(root: Path) -> dict[str, Any]: record: dict[str, Any] = { "status": "ok", @@ -114,26 +163,54 @@ def probe(root: Path) -> dict[str, Any]: "isolation_check": None, "isolation_mode": _sandbox.isolation_mode(), } + namespaces = record["namespaces"] + if namespaces and all(namespaces[name]["exit_code"] != 0 for name in ("user", "pid", "net", "ipc", "uts", "mount")): + record["isolation_mode"] = "task-boundary" + task_boundary = record["isolation_mode"] == "task-boundary" + if task_boundary: + # Real verification still needs fresh side tasks; only these packaged + # self-test applications share the probe task's boundary. + record["requires_fresh_side_tasks"] = True start = time.monotonic() try: - _sandbox.isolation_check() - record["isolation_check"] = {"status": "ok", "input_bytes": len(b"isolation-ready\n")} + if task_boundary: + record["isolation_check"] = {"status": "skipped", "reason": "task-boundary"} + else: + _sandbox.isolation_check() + record["isolation_check"] = {"status": "ok", "input_bytes": len(b"isolation-ready\n")} for family in FAMILIES: sides, demos = materialise(root / family, family) results: dict[str, Any] = {} record["oracles"][family] = results for name, demo in demos.items(): - result = verify( - *sides, - demo, - next(f for f, kinds in FAMILY_ORACLES.items() if family in kinds), - browser=BrowserExecutor() if family == "xss" else None, - ) + search_family = next(f for f, kinds in FAMILY_ORACLES.items() if family in kinds) + if task_boundary: + tick = time.monotonic() + try: + result = _verify_task_pair(sides, demo, family) + except Exception as exc: + result = VerificationRecord( + "inconclusive", (), time.monotonic() - tick, exception_reason(exc), isolation_mode="task-boundary" + ) + else: + result = verify( + *sides, + demo, + search_family, + oracle=family, + browser=BrowserExecutor() if family == "xss" else None, + ) results[name] = asdict(result) + if task_boundary and (result.outcome == "demonstrated") != (name == "real") and record["status"] == "ok": + expected = "demonstrated" if name == "real" else "not demonstrated" + record.update( + status="instrument_failure", + exception=f"{family}/{name}: expected {expected}, got {result.outcome}", + ) # SSRF/browser need additional namespace capabilities beyond the # generic preflight. These failures must not look like oracle misses. for side in result.sides: - if side.reason.startswith("verification unavailable or refused:"): + if not task_boundary and side.reason.startswith("verification unavailable or refused:"): raise _sandbox.IsolationUnavailable(f"{family}/{name}: {side.reason}") except Exception as exc: record.update(status="instrument_failure", exception=exception_reason(exc)) diff --git a/src/openultrasast/search/verify.py b/src/openultrasast/search/verify.py index 6946255..bf1c22f 100644 --- a/src/openultrasast/search/verify.py +++ b/src/openultrasast/search/verify.py @@ -437,10 +437,11 @@ def _run_task( the filesystem/network boundary. Its launch manifest must contain no keys. """ translations = {"/workspace": str(job.repo_root), "/scratch": str(scratch), **{k: str(v) for k, v in (mounts or {}).items()}} - pattern = re.compile("|".join(re.escape(key) for key in sorted(translations, key=len, reverse=True))) + pattern = re.compile("(?:" + "|".join(re.escape(key) for key in sorted(translations, key=len, reverse=True)) + r")(?=/|$|['\"])") def translate(value: str) -> str: - # One substitution pass: host temp paths may themselves contain /scratch. + # One pass, at path-component boundaries: /fixture/fixture.db must + # translate its mount prefix without also rewriting the filename. return pattern.sub(lambda match: translations[match.group(0)], value) command = tuple(translate(arg) for arg in job.command) diff --git a/tests/test_search_probe.py b/tests/test_search_probe.py index 669b9e1..8f7126e 100644 --- a/tests/test_search_probe.py +++ b/tests/test_search_probe.py @@ -2,12 +2,22 @@ import json import sys from pathlib import Path +import pytest + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from openultrasast.search import probe from openultrasast.search._sandbox import IsolationUnavailable +@pytest.fixture(autouse=True) +def namespace_diagnostics(monkeypatch): + facts = {name: {"exit_code": 0} for name in ("user", "pid", "net", "ipc", "uts", "mount")} + monkeypatch.setattr(probe, "namespace_facts", lambda: facts) + monkeypatch.setattr(probe._sandbox, "isolation_mode", lambda: "userns") + return facts + + def test_isolation_failure_is_recorded_and_nonzero(monkeypatch, capsys): def fail(**kwargs): raise IsolationUnavailable("gVisor denied namespace") @@ -118,3 +128,121 @@ def test_oracle_namespace_failure_is_instrument_failure(tmp_path, monkeypatch): assert record["status"] == "instrument_failure" assert "net namespace denied" in record["exception"] assert record["isolation_check"]["status"] == "ok" + + +@pytest.mark.parametrize("execution", ["real", "stubbed"]) +def test_task_boundary_probe_executes_packaged_pairs(tmp_path, monkeypatch, namespace_diagnostics, execution): + if execution == "real": + import socket + + try: + with socket.socket() as listener: + listener.bind(("127.0.0.1", 0)) + except PermissionError: + pytest.skip("host policy prohibits localhost sockets required by SSRF and Chromium") + else: + from openultrasast.search.verify import RunObservation, SideRecord + + def run(side, demo, family): + assert (side.checkout / "app.py").read_bytes() + observed = not (side.checkout / "fixed").exists() + return SideRecord( + "observed", + (RunObservation(observed, "stubbed owned effect", 0.5),), + 0.7, + "stubbed execution", + 0, + 0.2, + 0.5, + 4096, + "task-boundary", + json.loads((demo / "demo.json").read_text())["oracle"] == "xss", + ) + + monkeypatch.setattr(probe, "verify_side_task", run) + monkeypatch.setattr(probe._sandbox, "isolation_mode", lambda: "task-boundary") + monkeypatch.setattr(probe, "image_facts", lambda: {}) + monkeypatch.setattr(probe._sandbox, "isolation_check", lambda **kw: pytest.fail("task boundary must skip preflight")) + monkeypatch.setattr(probe, "verify", lambda *a, **kw: pytest.fail("must use task executor")) + record = probe.probe(tmp_path) + assert record["status"] == "ok", record + assert record["isolation_check"] == {"status": "skipped", "reason": "task-boundary"} + assert record["isolation_mode"] == "task-boundary" + assert record["namespaces"] == namespace_diagnostics + assert set(record["oracles"]) == set(probe.FAMILIES) + for family, demos in record["oracles"].items(): + assert set(demos) == {"real", "observer_access", "revision_patch"} + for name, result in demos.items(): + assert (result["outcome"] == "demonstrated") == (name == "real") + assert result["isolation_mode"] == "task-boundary" + assert result["elapsed_seconds"] > 0 + real = demos["real"] + assert real["chromium_no_sandbox"] == (family == "xss") + for index, side in enumerate(real["sides"]): + assert len(side["runs"]) == 3 + assert all(run["observed"] == (index == 0) for run in side["runs"]) + assert side["isolation_mode"] == "task-boundary" + assert side["chromium_no_sandbox"] == (family == "xss") + assert side["build_seconds"] == 0 + assert side["ready_seconds"] > 0 + assert side["run_seconds"] > 0 + assert side["scratch_peak_bytes"] > 0 + assert record["wall_seconds"] > 0 + assert record["peak_rss_kib"]["sum"] > 0 + assert record["scratch_disk_bytes"] > 0 + assert record["verification_scratch_peak_bytes"] > 0 + assert record["scratch_filesystem_used_bytes"] > 0 + + +@pytest.mark.parametrize("family", ["sql", "path", "command"]) +def test_task_boundary_real_local_oracles(tmp_path, family): + sides, demos = probe.materialise(tmp_path / family, family) + result = probe._verify_task_pair(sides, demos["real"], family) + assert result.outcome == "demonstrated", result + assert all(len(side.runs) == 3 for side in result.sides) + + +def test_task_boundary_xss_uses_browser_mode(tmp_path, monkeypatch): + from openultrasast.search.oracles import BrowserExecutor + + calls = [] + + def capture(self, document, nonce): + assert self.task_boundary and self.no_sandbox + calls.append(document) + return '' in document + + monkeypatch.setattr(BrowserExecutor, "__call__", capture) + sides, demos = probe.materialise(tmp_path / "xss", "xss") + result = probe._verify_task_pair(sides, demos["real"], "xss") + assert result.outcome == "demonstrated" + assert result.chromium_no_sandbox + assert all(side.chromium_no_sandbox for side in result.sides) + assert len(calls) == 6 + + +@pytest.mark.parametrize("mismatch", ["real", "observer_access", "revision_patch", "exception"]) +def test_task_boundary_mismatch_names_first_and_finishes(tmp_path, monkeypatch, mismatch): + from openultrasast.search.verify import VerificationRecord + + # On AX the diagnostics select the boundary before the failed bwrap preflight. + facts = {name: {"exit_code": 1} for name in ("user", "pid", "net", "ipc", "uts", "mount")} + monkeypatch.setattr(probe, "namespace_facts", lambda: facts) + monkeypatch.setattr(probe, "image_facts", lambda: {}) + monkeypatch.setattr(probe._sandbox, "isolation_check", lambda **kw: pytest.fail("must skip preflight")) + + def result(sides, demo, family): + if mismatch == "exception": + raise OSError("executor unavailable") + demonstrated = demo.name == "real" + if demo.name == mismatch: + demonstrated = not demonstrated + return VerificationRecord("demonstrated" if demonstrated else "inconclusive", (), 1, "", "task-boundary") + + monkeypatch.setattr(probe, "_verify_task_pair", result) + record = probe.probe(tmp_path) + assert record["status"] == "instrument_failure" + assert record["exception"].startswith(f"sql/{'real' if mismatch == 'exception' else mismatch}:") + assert record["namespaces"] == facts + assert record["isolation_check"] == {"status": "skipped", "reason": "task-boundary"} + assert sum(len(demos) for demos in record["oracles"].values()) == 15