From 7a91cc14829dbca80e06c538ba445388daed86fe Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Tue, 6 Oct 2026 12:43:11 +0200 Subject: [PATCH] search-with-proof: oracle placeholders for demos; Python apps run under the venv interpreter Demos may reference ${served_root}, ${database_path}, ${marker_dir} or ${callback_url}; the verifier substitutes them per side, so apps that take a directory or database argument can be wired to the oracle. Python apps start under the verifier interpreter (the one pip built the products for): the image carries a second system Python whose ABI tag broke compiled extensions. Co-Authored-By: Claude Fable 5.1 --- src/openultrasast/search/demo.py | 41 ++++--- src/openultrasast/search/oracles.py | 8 +- src/openultrasast/search/verify.py | 39 +++++-- src/openultrasast/search/worker.py | 15 ++- tests/test_search_demo.py | 36 ++++++ tests/test_search_verify.py | 164 ++++++++++++++++++++++++++++ 6 files changed, 270 insertions(+), 33 deletions(-) diff --git a/src/openultrasast/search/demo.py b/src/openultrasast/search/demo.py index 012a05e..6d9e01f 100644 --- a/src/openultrasast/search/demo.py +++ b/src/openultrasast/search/demo.py @@ -1,6 +1,6 @@ """Bounded declarative demonstrations; no executable agent content or private variables. -Only a previously captured application response can define a substitution. Runtime +Start configuration accepts oracle-owned fixture placeholders; steps accept prior response captures. Runtime paths and build manifests are checkout relative; request payloads remain data (and may contain traversal or injection strings being tested at the public interface). """ @@ -10,6 +10,7 @@ from __future__ import annotations import json import os import re +import sys import tomllib from pathlib import Path, PurePosixPath from typing import Any, cast @@ -77,7 +78,16 @@ BUILD_RECIPES = { "maven": ("/usr/bin/mvn", "--offline", "-f"), "gradle": ("/usr/bin/gradle", "--offline", "--no-daemon", "--project-dir"), } -RUNTIMES = {"python": ("/usr/bin/python3", "-I"), "node": ("/usr/bin/node",), "php": ("/usr/bin/php",), "java": ("/usr/bin/java", "-jar")} +# Python apps run under the verifier's own interpreter: it is the one pip installed the products for +# (the image carries a second, system Python with a different ABI tag; compiled extensions failed under it). +RUNTIMES = {"python": (sys.executable, "-I"), "node": ("/usr/bin/node",), "php": ("/usr/bin/php",), "java": ("/usr/bin/java", "-jar")} +ORACLE_PLACEHOLDERS = { + "path": {"served_root"}, + "sql": {"database_path", "database_url"}, + "command": {"marker_dir"}, + "ssrf": {"callback_url"}, + "xss": set(), +} _REFERENCE = re.compile(r"\$\{([a-z][a-z0-9_]{0,31})\}") _PRIVATE = re.compile(r"/(?:fixture|proc|sys|dev|deps|build)(?:/|\b)|(?:PROOF_MARKER|CALLBACK_URL|DATABASE|SERVED_ROOT)", re.I) @@ -274,11 +284,11 @@ def preparation_command(recipe: str, path: Path, checkout: Path, products: Path) def _text(value: Any, names: set[str], *, maximum: int = MAX_VALUE_BYTES) -> str: if not isinstance(value, str) or "\0" in value or len(value.encode()) > maximum: raise ValueError("invalid or oversized value") - if _PRIVATE.search(value): + if _PRIVATE.search(_REFERENCE.sub(lambda m: "" if m[1] in names else m[0], value)): raise ValueError("private verifier references are forbidden") references = _REFERENCE.findall(value) if any(name not in names for name in references) or "$" in _REFERENCE.sub("", value): - raise ValueError("only earlier response captures may be referenced") + raise ValueError("only allowed oracle placeholders or earlier response captures may be referenced") return value @@ -289,7 +299,8 @@ def _arguments(value: Any, names: set[str]) -> None: _text(arg, names) -def validate_start_config(start: dict[str, Any]) -> None: +def validate_start_config(start: dict[str, Any], names: set[str] | None = None) -> None: + names = names or set() environment = start.get("environment", {}) if not isinstance(environment, dict) or len(environment) > 32: raise ValueError("invalid environment map") @@ -304,7 +315,7 @@ def validate_start_config(start: dict[str, Any]) -> None: raise ValueError("forbidden environment variable") if not isinstance(value, str) or len(value) > 4096: raise ValueError("invalid environment value") - _text(value, set()) + _text(value, names) files = start.get("files", {}) if not isinstance(files, dict) or len(files) > 8: raise ValueError("invalid data file map") @@ -312,13 +323,15 @@ def validate_start_config(start: dict[str, Any]) -> None: checkout_path(name) if ".." in name or not any(name.endswith(ext) for ext in DATA_EXTENSIONS): raise ValueError("invalid data file extension or name") - _text(content, set(), maximum=MAX_DATA_BYTES) + _text(content, names, maximum=MAX_DATA_BYTES) -def write_start_files(checkout: Path, start: dict[str, Any]) -> None: +def write_start_files(checkout: Path, start: dict[str, Any], bindings: dict[str, str] | None = None) -> None: """Create only new private data files, before any app process is launched.""" - validate_start_config(start) + bindings = bindings or {} + validate_start_config(start, set(bindings)) for name, content in start.get("files", {}).items(): + content = substitute(content, bindings, maximum=MAX_DATA_BYTES) path = checkout / ".demo" / name directory = checkout for part in path.parent.relative_to(checkout).parts: @@ -361,7 +374,8 @@ def validate_demo(value: Any, *, checkout: Path | None = None) -> dict[str, Any] if checkout is not None: build_project(recipe, checkout / args[0], checkout) _keys(start, {"runtime", "path", "arguments", "mode"}, {"port", "environment", "files"}) - validate_start_config(start) + placeholders = ORACLE_PLACEHOLDERS[value["oracle"]] + validate_start_config(start, placeholders) ordinary = {**value, "start": {k: v for k, v in start.items() if k not in {"environment", "files"}}} if len(json.dumps(ordinary, allow_nan=False).encode()) > 65536: raise ValueError("demo excluding configuration exceeds size limit") @@ -370,7 +384,7 @@ def validate_demo(value: Any, *, checkout: Path | None = None) -> dict[str, Any] checkout_path(start["path"]) if start["path"].split("/")[0] == ".demo": raise ValueError("data files cannot be runtime entrypoints") - _arguments(start["arguments"], set()) + _arguments(start["arguments"], placeholders) if any( arg.startswith("/") or ".." in arg.split("/") or re.search(r"(?:^|[= :])/(?:etc|home|root|tmp|workspace|scratch)/", arg) for arg in start["arguments"] @@ -423,6 +437,7 @@ def validate_demo(value: Any, *, checkout: Path | None = None) -> dict[str, Any] or not re.fullmatch(r"[a-z][a-z0-9_]{0,31}", name) or name in names or name == "canary" + or any(name in allowed for allowed in ORACLE_PLACEHOLDERS.values()) or _PRIVATE.search(name) ): raise ValueError("invalid capture name") @@ -458,9 +473,9 @@ def load_demo(path: Path) -> dict[str, Any]: raise ValueError("demo nesting exceeds limit") from exc -def substitute(value: str, captures: dict[str, str]) -> str: +def substitute(value: str, captures: dict[str, str], *, maximum: int = MAX_VALUE_BYTES) -> str: result = _REFERENCE.sub(lambda m: captures[m.group(1)], value) - if len(result.encode()) > MAX_VALUE_BYTES or "\0" in result: + if len(result.encode()) > maximum or "\0" in result: raise ValueError("expanded value exceeds limit") return result diff --git a/src/openultrasast/search/oracles.py b/src/openultrasast/search/oracles.py index 50af637..8ab22da 100644 --- a/src/openultrasast/search/oracles.py +++ b/src/openultrasast/search/oracles.py @@ -39,6 +39,7 @@ class Canary: class Oracle(Protocol): namespace_pid: int | None + placeholders: dict[str, str] def prepare(self, canary: Canary) -> dict[str, str]: ... def capture(self, output: str) -> None: ... @@ -52,6 +53,7 @@ class OutputOracle: def prepare(self, canary: Canary) -> dict[str, str]: self.canary = canary self.seen = False + self.placeholders: dict[str, str] = {} return {} def capture(self, output: str) -> None: @@ -70,6 +72,7 @@ class PathOracle(OutputOracle): (canary.root / "public").mkdir() (canary.root / "canary").write_text(canary.nonce) (canary.root / "canary").chmod(0o600) + self.placeholders = {"served_root": "/fixture/public"} return {"SERVED_ROOT": "/fixture/public"} @@ -80,6 +83,7 @@ class SQLOracle(OutputOracle): 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)]) (canary.root / "fixture.db").chmod(0o600) + self.placeholders = {"database_path": "/fixture/fixture.db", "database_url": "sqlite:////fixture/fixture.db"} return {"DATABASE": "/fixture/fixture.db"} @@ -88,6 +92,7 @@ class CommandOracle(OutputOracle): super().prepare(canary) # Marker is in app-only writable scratch, inaccessible in artefact sandbox. self.marker = canary.root / "marker" + self.placeholders = {"marker_dir": "/scratch"} return {} def observe(self) -> tuple[bool, str]: @@ -132,7 +137,8 @@ class SSRFOracle(OutputOracle): self.namespace_pid, port = json.loads(line) if self.task_boundary: self.namespace_pid = None - return {"CALLBACK_URL": f"http://127.0.0.1:{port}/{canary.nonce}"} # loopback: oracle listener inside the private sandbox namespace + self.placeholders = {"callback_url": f"http://127.0.0.1:{port}/{canary.nonce}"} # loopback: private oracle listener + return {"CALLBACK_URL": self.placeholders["callback_url"]} def observe(self) -> tuple[bool, str]: assert self.process is not None and self.process.stdout is not None diff --git a/src/openultrasast/search/verify.py b/src/openultrasast/search/verify.py index 685992d..7ee627e 100644 --- a/src/openultrasast/search/verify.py +++ b/src/openultrasast/search/verify.py @@ -283,6 +283,7 @@ def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExe timings = dict(build=0.0, ready=0.0, run=0.0) app_diagnostics: dict[str, Any] = {} + private_values: list[str] = [] disk_peak = 0 root: Path | None = None @@ -301,7 +302,7 @@ def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExe outcome, tuple(observations), time.monotonic() - started, - diagnostic(reason, maximum=1500), + diagnostic(reason, maximum=1500, private=tuple(private_values)), timings["build"], timings["ready"] + app_diagnostics.get("ready_seconds", 0), max(0.0, timings["run"] - app_diagnostics.get("ready_seconds", 0)), @@ -310,8 +311,10 @@ def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExe chromium_no_sandbox=bool(getattr(browser, "no_sandbox", False)), phase=phase, exit_code=result.exit_code if result is not None else None, - stderr=diagnostic(result.stderr if result is not None else app_diagnostics.get("stderr", ""), maximum=1500), - stdout=diagnostic(getattr(result, "stdout", app_diagnostics.get("stdout", "")), maximum=1500), + stderr=diagnostic( + result.stderr if result is not None else app_diagnostics.get("stderr", ""), maximum=1500, private=tuple(private_values) + ), + stdout=diagnostic(getattr(result, "stdout", app_diagnostics.get("stdout", "")), maximum=1500, private=tuple(private_values)), ) try: @@ -359,12 +362,6 @@ def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExe return record("could_not_build", "build recipe failed", "build", compiled) if tuple(_tree(p) for p in immutable) != before: return record("inconclusive", "build changed immutable inputs") - write_start_files(checkout, schema["start"]) - if not task_boundary: - for name in schema["start"].get("files", {}): - _sandbox.writable_directory(checkout / ".demo" / name) - product_hash = _tree(products, generated_links=True) - checkout_hash = _tree(checkout, generated_links=True) mounts["/build"] = products start_spec = schema["start"] entry = checkout / start_spec["path"] @@ -400,14 +397,34 @@ def _side(side: Side, demo: Path, family: str, timeout: int, browser: BrowserExe assert oracle is not None try: canary = Canary.fresh(root / "fixture") + private_values.extend((canary.nonce, str(root), "/fixture", "/scratch")) if task_boundary and hasattr(oracle, "task_boundary"): oracle.task_boundary = True + oracle_env = oracle.prepare(canary) + bindings = oracle.placeholders + file_bindings = bindings + if task_boundary: + # Files are app data, so resolve their paths before writing; + # the task runner only translates argv and environment. + file_bindings = { + key: re.sub(r"/(?:fixture|scratch)(?=/|$)", lambda _, path=str(canary.root): path, value) + for key, value in bindings.items() + } + private_values.extend((*bindings.values(), *file_bindings.values())) + write_start_files(checkout, start_spec, file_bindings) + if not task_boundary: + for name in start_spec.get("files", {}): + _sandbox.writable_directory(checkout / ".demo" / name) + product_hash = _tree(products, generated_links=True) + checkout_hash = _tree(checkout, generated_links=True) + command = tuple(substitute(arg, bindings) for arg in command) app_env = { **{ - key: "/workspace/" + value if value.startswith(".demo/") else value + key: "/workspace/" + expanded if expanded.startswith(".demo/") else expanded for key, value in start_spec.get("environment", {}).items() + for expanded in [substitute(value, bindings)] }, - **oracle.prepare(canary), + **oracle_env, } if start_spec["runtime"] == "node": app_env["NODE_PATH"] = "/build/node_modules" diff --git a/src/openultrasast/search/worker.py b/src/openultrasast/search/worker.py index 7ba467b..8693659 100644 --- a/src/openultrasast/search/worker.py +++ b/src/openultrasast/search/worker.py @@ -91,22 +91,21 @@ DEMO_GUIDE = ( "HTTP mode requires port (1024-65535) and steps {type:http,method,path,headers?,body?}; " "CLI mode forbids port and uses {type:cli,arguments,stdin?}. Supply 1-32 ordered steps. " "Optional capture {name,source:output|json,path?} defines ${name} only for later steps from app responses. " - "Config via start.environment and small data files in start.files is allowed; no scripts or private canary references. " + "start.environment and start.files allow data, never scripts or raw private references. " + "Only the selected oracle's placeholders expand per side only in start.arguments, environment values and files contents. " "environment maps up to 32 uppercase names to values (4096 chars); loader/interpreter variables are forbidden. " "files maps up to 8 relative data filenames to contents (64 KiB each), created under .demo/ mode 0600; " "point environment values at .demo/name. Allowed: .pem .key .json .yaml .yml .txt .env .ini .toml .db .sqlite .crt." ) ORACLE_GUIDE = ( - "Family oracles: path plants a private nonce file outside the served directory and observes its contents in app output. " - "SQL plants public and secret rows in a fixture database and observes the secret nonce in app output. " - "Command reserves an app-only marker location and observes whether an injected command creates that file. " - "SSRF provides a private loopback listener with a fresh nonce URL and observes a request to it. " + 'Path observes a private nonce file outside ${served_root} in app output; e.g. "arguments": ["--dir", "${served_root}"]. ' + 'SQL observes secret rows in ${database_path} (${database_url} is its SQLite URL); e.g. "arguments": ["--db", "${database_path}"]. ' + 'Command observes marker creation at marker location ${marker_dir}/marker; e.g. "environment": {"MARKER_DIR": "${marker_dir}"}. ' + 'SSRF observes the nonce URL at its loopback listener ${callback_url}; e.g. "environment": {"INTERNAL_URL": "${callback_url}"}. ' "XSS installs a verifier-owned alert hook in a browser and observes actual execution of alert('ousast-xss'); " "reflection alone is insufficient. " "Set the required oracle field using the allowed choices below; for injection choose sql or command. " - "Other families have no owned oracle; report that gap instead of inventing one. " - + "Allowed family/oracle choices: " - + json.dumps(FAMILY_ORACLES) + "Other families have no oracle; report the gap. " + "Allowed family/oracle choices: " + json.dumps(FAMILY_ORACLES) ) REASON_PROMPT = ( "Your goal is a declarative demo conforming to demo.py's schema that makes the verifier observe the effect " diff --git a/tests/test_search_demo.py b/tests/test_search_demo.py index dce9a71..8ead1d4 100644 --- a/tests/test_search_demo.py +++ b/tests/test_search_demo.py @@ -225,3 +225,39 @@ def test_data_files_never_follow_links_or_overwrite(tmp_path, where): assert not list(outside.iterdir()) if where == "existing": assert (root / "keys/key.pem").read_text() == "original" + + +@pytest.mark.parametrize("field", ["arguments", "environment", "files"]) +@pytest.mark.parametrize( + "oracle,name", + [("path", "served_root"), ("sql", "database_path"), ("sql", "database_url"), ("command", "marker_dir"), ("ssrf", "callback_url")], +) +def test_oracle_placeholders_scoped_to_start(field, oracle, name): + value = document() + value["oracle"] = oracle + + def put(text): + value["start"][field] = [text] if field == "arguments" else {"CONFIG" if field == "environment" else "config.txt": text} + + put("${" + name + "}") + assert validate_demo(value) == value + value["oracle"] = "xss" + with pytest.raises(ValueError): + validate_demo(value) + value["oracle"] = oracle + for invalid in ("${nonsense}", "/fixture/public", "SERVED_ROOT", "DATABASE", "CALLBACK_URL", "${SERVED_ROOT}", "${bad-name}"): + put(invalid) + with pytest.raises(ValueError): + validate_demo(value) + + +@pytest.mark.parametrize("name", ["served_root", "database_path", "database_url", "marker_dir", "callback_url"]) +def test_fixture_placeholders_cannot_be_step_inputs_or_captures(name): + value = document() + value["steps"][0]["arguments"] = ["${" + name + "}"] + with pytest.raises(ValueError): + validate_demo(value) + value["steps"][0]["arguments"] = [] + value["steps"][0]["capture"] = {"name": name, "source": "output"} + with pytest.raises(ValueError): + validate_demo(value) diff --git a/tests/test_search_verify.py b/tests/test_search_verify.py index 9653cfe..98c2556 100644 --- a/tests/test_search_verify.py +++ b/tests/test_search_verify.py @@ -697,3 +697,167 @@ def test_http_driver_polls_eight_second_listener_before_get_and_steps(tmp_path, assert polls == [i * 0.5 for i in range(len(polls))] assert (tmp_path / "output").read_text() == "boot stdout" assert (tmp_path / "errors").read_text() == "boot stderr" + + +@pytest.mark.parametrize("task_boundary", [False, True]) +@pytest.mark.parametrize("config", ["arguments", "environment", "files"]) +def test_node_directory_option_uses_owned_fixture(pair, config, task_boundary): + sides, demo = pair + app = """const fs = require('fs'), path = require('path'); +const args = process.argv.slice(2); +const root = args[0] === '--dir' ? args[1] : + (process.env.PUBLIC_DIRECTORY || fs.readFileSync(path.join(__dirname, '.demo/root.txt'), 'utf8')); +if (!fs.statSync(root).isDirectory()) throw new Error('missing served directory'); +if (args.includes('--ready')) process.exit(0); +const target = path.resolve(root, args.at(-1)); +if (!fs.existsSync(path.join(__dirname, 'fixed')) || target.startsWith(root + '/')) + process.stdout.write(fs.readFileSync(target)); +""" + for side in sides: + (side.checkout / "app.js").write_text(app) + + def change(value): + value["start"].update(runtime="node", path="app.js") + if config == "arguments": + value["start"]["arguments"] = ["--dir", "${served_root}"] + elif config == "environment": + value["start"]["environment"] = {"PUBLIC_DIRECTORY": "${served_root}"} + else: + value["start"]["files"] = {"root.txt": "${served_root}"} + value["steps"] = [{"type": "cli", "arguments": ["../canary"]}] + + edit(demo, change) + original = (demo / "demo.json").read_bytes() + if task_boundary: + records = [verify_side_task(side, demo, "path") for side in sides] + else: + result = verify(*sides, demo, "path") + assert result.outcome == "demonstrated", result + records = result.sides + assert all(record.outcome == "observed" for record in records), records + assert all(run.observed for run in records[0].runs) + assert all(not run.observed for run in records[1].runs) + assert (demo / "demo.json").read_bytes() == original + assert all(not (side.checkout / ".demo").exists() for side in sides) + + +@pytest.mark.parametrize("placeholder", ["database_path", "database_url"]) +@pytest.mark.parametrize("task_boundary", [False, True]) +def test_sql_cli_database_fixture(tmp_path, placeholder, task_boundary): + sides, demos = materialise(tmp_path / "sql", "sql") + for side in sides: + path = side.checkout / "app.py" + source = path.read_text().replace('os.environ["DATABASE"]', "database") + source = source.replace( + "fixed = ", 'database = sys.argv.pop(1).removeprefix("sqlite:///")\nassert pathlib.Path(database).is_file()\nfixed = ' + ) + path.write_text(source) + edit(demos["real"], lambda value: value["start"].update(arguments=["${" + placeholder + "}"])) + if task_boundary: + records = [verify_side_task(side, demos["real"], "injection") for side in sides] + else: + result = verify(*sides, demos["real"], "injection") + assert result.outcome == "demonstrated", result + records = result.sides + assert all(record.outcome == "observed" for record in records), records + assert all(run.observed for run in records[0].runs) + assert all(not run.observed for run in records[1].runs) + + +@pytest.mark.parametrize("task_boundary", [False, True]) +def test_command_marker_directory_configuration(tmp_path, task_boundary): + sides, demos = materialise(tmp_path / "command", "command") + for side in sides: + path = side.checkout / "app.py" + source = path.read_text().replace( + 'subprocess.run("echo " + payload, shell=True, check=False)', + 'subprocess.run("echo " + payload, shell=True, check=False, cwd=os.environ["MARKER_DIRECTORY"])', + ) + path.write_text(source) + edit(demos["real"], lambda value: value["start"].update(environment={"MARKER_DIRECTORY": "${marker_dir}"})) + if task_boundary: + records = [verify_side_task(side, demos["real"], "injection") for side in sides] + else: + result = verify(*sides, demos["real"], "injection") + assert result.outcome == "demonstrated", result + records = result.sides + assert all(record.outcome == "observed" for record in records), records + assert all(run.observed for run in records[0].runs) + assert all(not run.observed for run in records[1].runs) + + +def test_fixture_values_and_nonce_redacted_from_diagnostics(pair, monkeypatch): + from dataclasses import asdict + + from openultrasast.search import oracles + + sides, demo = pair + nonces = [] + prepare = oracles.PathOracle.prepare + + def remember(self, canary): + nonces.append(canary.nonce) + return prepare(self, canary) + + monkeypatch.setattr(oracles.PathOracle, "prepare", remember) + edit(demo, lambda value: value["start"].update(environment={"PUBLIC_DIRECTORY": "${served_root}"})) + path = sides[0].checkout / "app.py" + path.write_text('import os\nprint(os.environ["PUBLIC_DIRECTORY"])\n' + path.read_text()) + result = verify(*sides, demo, "path") + assert result.outcome == "demonstrated", result + serialized = json.dumps(asdict(result)) + assert nonces and all(nonce not in serialized for nonce in nonces) + assert "/fixture" not in serialized and "ousast-proof-" not in serialized + + +def test_callback_placeholder_is_fresh_and_app_only_without_network(tmp_path, monkeypatch): + import io + from types import SimpleNamespace + + from openultrasast.sandbox import SandboxResult + from openultrasast.search import oracles + + sides, demos = materialise(tmp_path / "ssrf", "ssrf") + launched, commands = [], [] + + def listener(argv, **kwargs): + launched.append(argv[-1]) + return SimpleNamespace(stdout=io.StringIO("[123, 18765]\n"), terminate=lambda: None, wait=lambda **kw: None) + + class Selector: + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + def register(self, *args): + pass + + def select(self, timeout): + return [True] + + monkeypatch.setattr(oracles.subprocess, "Popen", listener) + monkeypatch.setattr(oracles.selectors, "DefaultSelector", Selector) + monkeypatch.setattr(_sandbox, "isolation_check", lambda **kw: None) + + def runner(job, **kwargs): + url = f"http://127.0.0.1:18765/{launched[-1]}" + assert job.command[-2 if job.command[-1] == "--ready" else -3] == url + assert kwargs["env"]["INTERNAL_URL"] == kwargs["env"]["CALLBACK_URL"] == url + assert (job.repo_root / ".demo/url.txt").read_text() == url + commands.append(job.command) + return SandboxResult(0, "safe", "", False) + + monkeypatch.setattr(_sandbox, "run", runner) + edit( + demos["real"], + lambda value: value["start"].update( + arguments=["${callback_url}"], environment={"INTERNAL_URL": "${callback_url}"}, files={"url.txt": "${callback_url}"} + ), + ) + result = verify(*sides, demos["real"], "untrusted_destination") + assert all(side.outcome == "observed" for side in result.sides), result + assert len(launched) == len(set(launched)) == 6 + assert len(commands) == 12 + assert "${callback_url}" in (demos["real"] / "demo.json").read_text()