diff --git a/scripts/conformance/command_keys/README.md b/scripts/conformance/command_keys/README.md index 84655be..105bd7f 100644 --- a/scripts/conformance/command_keys/README.md +++ b/scripts/conformance/command_keys/README.md @@ -81,3 +81,69 @@ The accepted common stopped-copy helper is owned by [SDK Rust260](https://github.com/durable-workflow/sdk-rust/pull/260); Server1210 owns the real historical seed. Namespace/run fences remain independently scoped. This adapter alone does not complete shared216. + +## Physical upgrade consumer + +The producer-owned [frozen handoff](https://github.com/durable-workflow/server/pull/1210#issuecomment-6860138558) +is accepted at contract source `a09f156399416216d2f097c257a2164f83c7eda7`. +The actual original seed execution remains `09a4ccda3da11853606af031a8970bbde4b2d6ec`, +driver SHA-256 `01c0fa1251ac3c809723e58d71817881fe2c5c9498579bc4c2e6097324a3618b`. +These are different authorities and must remain distinct in every receipt. + +`run_physical_readback.py` consumes that stopped backup/JSON plus the accepted +common snapshot helper and hash-bound tagged SQL step. It captures every input +with the same Git/owner/no-follow/held-identity boundary as the response-loss +runner. Only a new private copy is upgraded. Start/inspect/wait/log/remove/absence +checks use captured full creation IDs and ownership labels. Upgrade and runtime +are sequential; the runtime and registry consumer are the only concurrent pair. +The fixed image/bootstrap/hash-locked registry environment is shared with the +reviewed response-loss runner. No host interpreter or executable input pathname +is accepted. + +```sh +python3 -I scripts/conformance/command_keys/run_physical_readback.py \ + --runner-commit FULL_REVIEWED_RUNNER_COMMIT \ + --wheel /path/to/downloaded/durable_workflow-2.17.2-py3-none-any.whl \ + --baseline /path/to/producer/stopped-backup.sqlite \ + --seed /path/to/producer/seed-receipts.json \ + --snapshot-helper /path/to/accepted/sqlite_legacy.py \ + --step-sql /path/to/tagged/0013_command_intents.sql \ + --output /path/to/private-disk-scratch/new-physical-run +``` + +The controller binds the frozen backup/packet/helper/step hashes. It checks the +native marker transition and validates every business row/schema object, allowing +only the exact new empty intent catalog and source-defined marker/journal changes. +Every SQLx journal field is checked: version, description, tagged cumulative SQL +checksum, success, zero execution time and the actual upgrade timestamp window. +The installed Client's readbacks are bracketed by complete typed logical snapshots. +That interval ends before shutdown; subsequent repair work is not relabeled as +part of the zero-write claim. Original inputs are rechecked and preserved. + +For each original keyed update in open/completed/continued state, four concurrent +changed-input retries must return the original command/run/update/acceptance +identities with explicit `intent_verified: false`. Changed names must conflict; +GET-only outcome reads distinguish pending from completed-null. The Python update +helper uses its unscoped public route, so no new explicit-run update API is claimed. +Historical NULL-key signals receive no fabricated retry key. Their compatibility +control preserves their original history and business identities only. + +A producer upgrade receipt alone does not qualify this SDK. The exact consumer +source, installed proof, complete snapshots, immutable input/toolchain rechecks, +all three captured-ID cleanup records and independent review remain required. + +The host workspace also retains no-follow descriptors for every ancestor, +private directory and created file. Config/credential bytes are immutable; +the copied SQLite file may change contents but must keep its owned single-link +inode. Docker configuration and credential reads use held +`/proc//fd` paths. The canonical database bind has a preflight +inside the immutable published image: it verifies the mounted directory/file +device, inode, owner, mode and single-link identities before executing the +native entrypoint. Host snapshots and lifecycle calls check workspace +identities before and after use. Cleanup uses an explicit empty kernel FD +configuration directory, so workspace substitution cannot block known-ID cleanup. Evidence is created exclusively relative to the +held directory, with the creation descriptor retained through verification. +A substituted parent, copied database, configuration, credential or report refuses +qualification. Known-container cleanup stays bound to captured IDs and preserves +replacement paths. Workspace substitution controls must pass before installed +response-loss or physical results qualify. diff --git a/scripts/conformance/command_keys/consumer_bootstrap.py b/scripts/conformance/command_keys/consumer_bootstrap.py index 21067db..60a5d13 100644 --- a/scripts/conformance/command_keys/consumer_bootstrap.py +++ b/scripts/conformance/command_keys/consumer_bootstrap.py @@ -22,6 +22,7 @@ RESULT_PREFIX = "SDK_PYTHON_CONSUMER_RESULT=" ROLES = {"adapter", "bootstrap", "proxy", "wheel", "requirements"} +PHYSICAL_ROLES = {"adapter", "bootstrap", "verifier", "seed", "wheel", "requirements"} MAX_BUNDLE_BYTES = 8 * 1024 * 1024 IMAGE_INTERPRETER = Path("/usr/local/bin/python3.12") @@ -35,7 +36,9 @@ def approved_environment_link(relative, target): def decode_inputs(bundle): - if set(bundle["sources"]) != ROLES: + scenario = bundle.get("scenario", "response_loss") + roles = ROLES if scenario == "response_loss" else PHYSICAL_ROLES if scenario == "physical_readback" else None + if roles is None or set(bundle["sources"]) != roles: raise ValueError("captured source roles differ") result = {} for role, entry in bundle["sources"].items(): @@ -185,14 +188,23 @@ def main(): descriptor = os.open(credential, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(descriptor, "w") as stream: stream.write("DW_AUTH_TOKEN=" + bundle["token"] + "\n") - adapter = execute_source(sources["adapter"], "installed_command_keys") + scenario = bundle.get("scenario", "response_loss") + verifier = execute_source(sources.get("verifier", sources["adapter"]), "installed_package_verifier") + package = verifier.installed_identity(sources["wheel"]) + adapter = ( + verifier if scenario == "response_loss" else execute_source(sources["adapter"], "installed_physical_readback") + ) try: result = asyncio.run( adapter.main( SimpleNamespace( upstream=bundle["upstream"], credential_file=credential, - proxy_data=sources["proxy"], + proxy_data=sources.get("proxy"), + seed_data=sources.get("seed"), + seed_sha256=bundle["sources"].get("seed", {}).get("sha256"), + package_identity=package, + verifier_sha256=bundle["sources"].get("verifier", bundle["sources"]["adapter"])["sha256"], wheel_data=sources["wheel"], adapter_sha256=bundle["sources"]["adapter"]["sha256"], server_version=bundle["server_version"], diff --git a/scripts/conformance/command_keys/physical_readback.py b/scripts/conformance/command_keys/physical_readback.py new file mode 100644 index 0000000..095a131 --- /dev/null +++ b/scripts/conformance/command_keys/physical_readback.py @@ -0,0 +1,161 @@ +"""Installed SDK readbacks for the producer-owned physical seed, after upgrade. + +The outer runner owns the verified seed/database, exact image, upgrade, complete +zero-write snapshots and cleanup. This consumer neither seeds nor upgrades data. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import stat +from datetime import datetime, timezone +from urllib.parse import urlsplit + +from durable_workflow import Client +from durable_workflow.errors import UpdateFailed + +# Physical seed authority includes the producer's unkeyed-signal clarification. +# It is distinct from the older manifest bound to the synthetic-copy helper. +MANIFEST_SHA256 = "ceb4695643e48c7bfe3f81eea380c6ab9aa0b973da830cb552d070344047319f" + + +def seed_cases(document): + if ( + document.get("schema") != "durable-workflow.command-key-physical-seed.v1" + or document.get("manifest_sha256") != MANIFEST_SHA256 + ): + raise ValueError("require the producer-owned physical seed contract") + cases = document["cases"] + expected = {(kind, state) for kind in ("signal", "update") for state in ("open", "completed", "continued")} + if len(cases) != len(expected) or {(case["kind"], case["state"]) for case in cases} != expected: + raise ValueError("physical seed must contain each original state/type exactly once") + for case in cases: + receipt = case["receipt"] + fields = ("command_id", "run_id") + (("update_id", "accepted_at") if case["kind"] == "update" else ()) + if any(not isinstance(receipt.get(field), str) or not receipt[field] for field in fields): + raise ValueError("original receipt has incomplete durable identities") + if receipt["workflow_id"] != case["workflow_id"]: + raise ValueError("original receipt names a different workflow") + if case["kind"] == "update" and not case["original_body"].get("request_id"): + raise ValueError("original keyed update identity is missing") + if case["history"]["next_page_token"] is not None: + raise ValueError("physical seed history was not fully captured") + return cases + + +def legacy_receipt(original, received): + if received.get("retained") is not True or received.get("intent_verified") is not False: + raise ValueError("legacy readback must be retained and explicitly unverified") + for field in ("command_id", "run_id", "update_id", "accepted_at"): + if received.get(field) != original[field]: + raise ValueError("legacy readback changed original " + field) + + +async def main(args): + if not __debug__: + raise ValueError("acceptance assertions require Python without optimization") + if args.output.exists(): + raise ValueError("refusing to overwrite existing evidence") + upstream = urlsplit(args.upstream) + if ( + upstream.scheme != "http" + or upstream.hostname != "127.0.0.1" + or not upstream.port + or args.upstream != f"http://127.0.0.1:{upstream.port}" + ): + raise ValueError("synthetic runtime must be canonical numeric loopback") + raw_seed = args.seed_data + if hashlib.sha256(raw_seed).hexdigest() != args.seed_sha256: + raise ValueError("producer seed receipt hash mismatch") + cases = seed_cases(json.loads(raw_seed)) + package = args.package_identity + mode = args.credential_file.lstat().st_mode + if not stat.S_ISREG(mode) or mode & 0o077: + raise ValueError("fixture credential file must be owner-only") + token = dict(line.split("=", 1) for line in args.credential_file.read_text().splitlines())["DW_AUTH_TOKEN"] + report = { + "started_at": datetime.now(timezone.utc).isoformat(), + "sdk": package, + "seed_sha256": args.seed_sha256, + "verifier_sha256": args.verifier_sha256, + "adapter_sha256": args.adapter_sha256, + "scope": "installed SDK physical-upgrade readbacks only", + "cells": [], + "complete": False, + } + try: + async with Client(args.upstream, token=token) as client: + info = await client.get_cluster_info() + if info["version"] != args.server_version: + raise ValueError("runtime differs from the outer verified upgrade tuple") + report["server_version"] = info["version"] + for case in cases: + original = case["receipt"] + workflow_id = case["workflow_id"] + observed = [event async for event in client.iter_history(workflow_id, original["run_id"])] + if observed != case["history"]["events"]: + raise ValueError("upgrade history differs from the original producer capture") + cell = { + "kind": case["kind"], + "state": case["state"], + "command_id": original["command_id"], + "run_id": original["run_id"], + "original_history_preserved": True, + } + if case["kind"] == "signal": + # Published historical signals have NULL stored keys. Never + # manufacture keyed recovery from an unrelated current key. + cell["scope"] = "unkeyed historical signal compatibility only" + else: + key = case["original_body"]["request_id"] + name = original["update_name"] + receipts = await asyncio.gather( + *( + client.update_workflow(workflow_id, name, args=[999], request_id=key, wait_for="accepted") + for _ in range(4) + ) + ) + for receipt in receipts: + legacy_receipt(original, receipt) + assert receipt["request_id"] == key + if case["state"] != "open": + assert receipt["update_status"] == "completed" and receipt["result"] is None + try: + await client.update_workflow(workflow_id, "other", args=[999], request_id=key) + except UpdateFailed as error: + assert error.reason == "request_id_conflict" + assert error.details["retained_command_id"] == original["command_id"] + else: + raise AssertionError("changed legacy command name was admitted") + outcome = await client.update_outcome_receipt(workflow_id, original["update_id"]) + assert outcome["update_id"] == original["update_id"] + if case["state"] == "open": + assert outcome["update_status"] == "accepted" + else: + assert outcome["update_status"] == "completed" and outcome["result"] is None + assert await client.update_outcome(workflow_id, original["update_id"]) is None + cell.update( + { + "scope": "original keyed update; unscoped SDK route returns original run", + "update_id": original["update_id"], + "concurrent_retries": len(receipts), + "retained": True, + "intent_verified": False, + "changed_name": "request_id_conflict", + } + ) + report["cells"].append(cell) + report["complete"] = True + report["finished_at"] = datetime.now(timezone.utc).isoformat() + except Exception as error: + report["failure_type"] = type(error).__name__ + raise + finally: + args.output.write_text(json.dumps(report, indent=2) + "\n") + return report + + +if __name__ == "__main__": + raise SystemExit("Use the captured-input run_physical_readback.py controller") diff --git a/scripts/conformance/command_keys/run_physical_readback.py b/scripts/conformance/command_keys/run_physical_readback.py new file mode 100644 index 0000000..ce91df2 --- /dev/null +++ b/scripts/conformance/command_keys/run_physical_readback.py @@ -0,0 +1,395 @@ +"""Qualify one installed SDK consumer on a private copy of the producer seed.""" + +from __future__ import annotations + +import argparse +import base64 +import hashlib +import json +import os +import re +import secrets +import sqlite3 +import subprocess +import time +import urllib.request +from contextlib import ExitStack, closing +from copy import deepcopy +from datetime import datetime, timezone +from pathlib import Path +from types import ModuleType +from uuid import uuid4 + +HELPER_SHA256 = "6fc6054ce087c136a86d7b6b222daeefaa87c58fa0b613b155a5df39e6a26fad" +BASELINE_SHA256 = "7d1826f7681342bb3a5974509648e2e8d75e7a21d262f65f3ec1f58727ccbc0c" +SEED_SHA256 = "8c728ce827df3a11621dfb0e82cdd404bbab48784182a5e92c8d632d49047fc5" +STEP_SHA256 = "fc3ad1730682c09704027dcbabc17cd6ded9c642a3ff20d7988d5f4b4a5726fd" +JOURNAL_SHA384 = "53b752529a1f7addaefe62ab61160dd42b3990f805cfdc4ecd1b48927a6c2f7ce42526fc04522ac887a37208f343fc73" +JOURNAL_DESCRIPTION = ( + "full frozen PHP schema, native worker incarnations, session holders, stream reclaim deadlines, " + "service call deadlines, service admission counters and committed external payload deletions and " + "task ready origins and poll expiry indexes and bound run command contracts and task queue activity " + "dispatch pauses and keyed command intents" +) + + +def load_bytes(data, name, expected, *, filename=None): + if hashlib.sha256(data).hexdigest() != expected: + raise ValueError("required executable source identity differs") + module = ModuleType(name) + module.__file__ = filename or "captured-" + name + ".py" + exec(compile(data, module.__file__, "exec"), module.__dict__) + module.source_sha256 = expected + return module + + +def native_marker(connection, expected): + if connection.execute("SELECT engine,version FROM dw_server_schema").fetchall() != [("rust-development", expected)]: + raise ValueError("native schema ownership/version marker differs") + + +def validate_journal(table, began, finished): + columns = table["columns"] + if ( + set(columns) != {"version", "description", "installed_on", "success", "checksum", "execution_time"} + or len(table["rows"]) != 1 + ): + raise ValueError("upgraded migration journal shape differs") + row = dict(zip(columns, table["rows"][0], strict=True)) + expected = { + "version": ["integer", 13], + "description": ["text", JOURNAL_DESCRIPTION], + "success": ["integer", 1], + "checksum": ["blob", JOURNAL_SHA384], + "execution_time": ["integer", 0], + } + if any(row[key] != value for key, value in expected.items()): + raise ValueError("upgraded migration journal field differs from tagged source") + installed = row["installed_on"] + if installed[0] != "text": + raise ValueError("migration timestamp type differs") + timestamp = datetime.fromisoformat(installed[1]).replace(tzinfo=timezone.utc).timestamp() + if not int(began) <= timestamp <= finished: + raise ValueError("migration timestamp is outside the actual upgrade window") + + +def validate_upgrade(before, after, added_schema, began, finished): + # Preserve every business row and catalog entry. Only the exact source-defined + # marker/journal changes and empty new intent table may differ. + if sorted(before["schema"] + added_schema) != after["schema"]: + raise ValueError("upgrade changed unexpected schema objects") + validate_journal(after["tables"]["_sqlx_migrations"], began, finished) + expected = deepcopy(before) + expected["schema"] = after["schema"] + marker = expected["tables"]["dw_server_schema"] + version = marker["columns"].index("version") + if len(marker["rows"]) != 1 or marker["rows"][0][version] != ["integer", 12]: + raise ValueError("baseline marker differs") + marker["rows"][0][version] = ["integer", 13] + expected["tables"]["_sqlx_migrations"] = after["tables"]["_sqlx_migrations"] + expected["tables"]["dw_command_intents"] = {"columns": ["command_id", "namespace", "intent_digest"], "rows": []} + if expected != after: + raise ValueError("upgrade changed business rows or unexpected metadata") + + +def controls_from_git(args): + if re.fullmatch(r"[0-9a-f]{40}", args.runner_commit) is None: + raise ValueError("require one exact runner commit") + root = Path(__file__).absolute().parents[3] + relative = "scripts/conformance/command_keys/run_response_loss.py" + data = subprocess.check_output( + ["git", "-C", str(root), "show", args.runner_commit + ":" + relative], env={"PATH": os.defpath}, timeout=10 + ) + return load_bytes( + data, "reviewed_runtime_controls", hashlib.sha256(data).hexdigest(), filename=str(root / relative) + ) + + +def run_bound(args, inputs, controls, workspace): + legacy = load_bytes(inputs["snapshot_helper"].data, "accepted_snapshot", HELPER_SHA256) + with closing(sqlite3.connect(":memory:")) as catalog: + catalog.execute("CREATE TABLE dw_server_schema(engine TEXT, version INTEGER)") + catalog.executescript(inputs["step_sql"].data.decode()) + added_schema = [row for row in legacy.snapshot(catalog)["schema"] if row[2] != "dw_server_schema"] + database = workspace.directory("database") + copied = Path(workspace.proc(database, "upgraded.sqlite")) + config = workspace.directory("docker-config") + workspace.write("config.json", '{"auths":{}}\n', parent=config) + environment = {"PATH": os.defpath, "TMPDIR": workspace.proc(workspace.root)} + docker = ["docker", "--config", workspace.proc(config)] + + def call(*arguments, check=True, timeout=45): + workspace.check() + result = subprocess.run( + docker + list(arguments), env=environment, check=check, capture_output=True, text=True, timeout=timeout + ) + if arguments[0] != "create": + workspace.check() + return result + + def cleanup_call(*arguments, check=True, timeout=45): + return subprocess.run( + ["docker", "--config", "/proc/self/fd", *arguments], + env=environment, + check=check, + capture_output=True, + text=True, + timeout=timeout, + ) + + def snapshot(): + workspace.check() + with closing(sqlite3.connect(copied.as_uri() + "?mode=ro", uri=True, timeout=2)) as connection: + connection.execute("BEGIN") + result = legacy.snapshot(connection) + workspace.check() + return result + + image = "ghcr.io/durable-workflow/server@" + controls.EXPECTED_SERVER["index"] + python_image = "docker.io/library/python@" + controls.PYTHON_INDEX + observed = controls.image_identity(json.loads(call("image", "inspect", image).stdout)) + interpreter = controls.interpreter_identity(json.loads(call("image", "inspect", python_image).stdout)) + nonce = uuid4().hex + token = secrets.token_hex(24) + credential = workspace.write( + "fixture.env", + "DW_AUTH_TOKEN=" + token + "\nDW_RUNTIME_CREDENTIALS_ENABLED=false\n" + "DW_RUST_MAINTENANCE_LOOP=0\nDB_CONNECTION=sqlite\nDB_DATABASE=/cohort/upgraded.sqlite\n", + ) + report = { + "started_at": datetime.now(timezone.utc).isoformat(), + "runner_commit": args.runner_commit, + "accepted_contract_source": "a09f156399416216d2f097c257a2164f83c7eda7", + "actual_seed_source": "09a4ccda3da11853606af031a8970bbde4b2d6ec", + "actual_seed_driver_sha256": "01c0fa1251ac3c809723e58d71817881fe2c5c9498579bc4c2e6097324a3618b", + "expected_server": controls.EXPECTED_SERVER, + "observed_image": observed, + "interpreter_image": interpreter, + "ownership_nonce": nonce, + "owned_containers": {}, + "inputs": {role: {"sha256": value.sha256, "bytes": len(value.data)} for role, value in inputs.items()}, + "cleanup_by_id": {}, + "complete": False, + } + common = [ + "--init", + "--read-only", + "--cpus", + "1", + "--memory", + "1g", + "--pids-limit", + "256", + "--label", + controls.OWNER_LABEL + "=sdk-python", + "--label", + controls.RUN_LABEL + "=" + nonce, + "--tmpfs", + "/tmp:mode=1777,size=128m", + "--tmpfs", + "/app/storage:uid=1000,gid=1000,mode=0700,size=64m", + "--tmpfs", + "/app/database:uid=1000,gid=1000,mode=0700,size=16m", + "-v", + workspace.mount_path(database) + ":/cohort:rw", + "--env-file", + workspace.proc(credential["fd"]), + ] + + def create(role, *arguments): + name = "sdk-python-216-physical-" + role + "-" + nonce[:12] + identity = controls.created_id(call("create", "--name", name, *common, *arguments).stdout) + if identity in [entry["id"] for entry in report["owned_containers"].values()]: + raise ValueError("duplicate creation ID") + report["owned_containers"][role] = {"id": identity, "requested_name": name} + if controls.inspect_owned(call, identity, nonce) is None: + raise ValueError("created object vanished") + return identity + + try: + for value in inputs.values(): + value.recheck() + held_database = workspace.write("upgraded.sqlite", inputs["baseline"].data, parent=database, immutable=False) + with closing(sqlite3.connect(copied)) as connection: + native_marker(connection, 12) + baseline = snapshot() + began = time.time() + upgrade_id = create( + "upgrade", + "--network", + "none", + "--entrypoint", + "/bin/sh", + image, + *workspace.native_arguments(database, held_database, "schema-upgrade"), + ) + call("start", upgrade_id) + if call("wait", upgrade_id).stdout.strip() != "0": + raise RuntimeError("schema upgrade failed") + upgrade = json.loads(call("logs", upgrade_id).stdout) + if upgrade != {"backend": "sqlite", "from_schema_version": 12, "schema_version": 13, "status": "upgraded"}: + raise ValueError("unexpected upgrade result") + finished = time.time() + report["upgrade"] = upgrade + report["cleanup_by_id"]["upgrade"] = controls.cleanup_owned(cleanup_call, upgrade_id, nonce) + with closing(sqlite3.connect(copied)) as connection: + native_marker(connection, 13) + validate_upgrade(baseline, snapshot(), added_schema, began, finished) + report["upgrade_row_preservation"] = True + report["journal_sha384"] = JOURNAL_SHA384 + runtime_id = create( + "runtime", + "-p", + "127.0.0.1::8080", + "--entrypoint", + "/bin/sh", + image, + *workspace.native_arguments(database, held_database), + ) + call("start", runtime_id) + address = call("port", runtime_id, "8080/tcp").stdout.strip() + host, port = address.split(":") + if host != "127.0.0.1" or not port.isdigit() or not 1 <= int(port) <= 65535: + raise ValueError("runtime did not bind numeric loopback") + upstream = "http://" + address + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + deadline = time.monotonic() + 45 + while time.monotonic() < deadline: + try: + with opener.open(upstream + "/api/health", timeout=2) as response: + if response.status == 200: + break + except OSError: + time.sleep(0.2) + else: + raise TimeoutError("upgraded runtime readiness") + bundle = { + "scenario": "physical_readback", + "sources": { + role: {"sha256": value.sha256, "base64": base64.b64encode(value.data).decode()} + for role, value in inputs.items() + if role in {"adapter", "bootstrap", "requirements", "verifier", "seed", "wheel"} + }, + "upstream": upstream, + "token": token, + "server_version": controls.EXPECTED_SERVER["version"], + } + for value in inputs.values(): + value.recheck() + consumer_id = controls.create_consumer(call, "sdk-python-216-physical-consumer-" + nonce[:12], nonce, inputs) + if consumer_id in [entry["id"] for entry in report["owned_containers"].values()]: + raise ValueError("duplicate consumer ID") + report["owned_containers"]["consumer"] = {"id": consumer_id} + if controls.inspect_owned(call, consumer_id, nonce) is None: + raise ValueError("consumer vanished") + before = snapshot() + result = subprocess.run( + docker + ["start", "-a", "-i", consumer_id], + input=json.dumps(bundle), + env=environment, + capture_output=True, + text=True, + timeout=600, + ) + after = snapshot() + workspace.write("consumer.log", result.stdout + result.stderr) + report["request_window_before_sha256"] = legacy.fingerprint(before) + report["request_window_after_sha256"] = legacy.fingerprint(after) + if before != after: + raise ValueError("consumer window changed complete logical rows/schema") + state = controls.inspect_owned(call, consumer_id, nonce) + if state is None or state["State"]["Running"]: + raise ValueError("consumer final state unverified") + report["consumer_exit_code"] = state["State"]["ExitCode"] + if result.returncode != 0 or report["consumer_exit_code"] != 0: + raise RuntimeError("installed consumer failed") + for value in inputs.values(): + value.recheck() + report["input_identities_rechecked"] = True + if controls.interpreter_identity(json.loads(call("image", "inspect", python_image).stdout)) != interpreter: + raise ValueError("interpreter image changed") + report["interpreter_identity_rechecked"] = True + proof = controls.consumer_result(result.stdout, inputs, scenario="physical_readback") + workspace.write("consumer.json", json.dumps(proof, indent=2) + "\n") + workspace.check() + report["workspace_identities_rechecked"] = True + report["consumer_complete"] = True + except Exception as error: + report["failure_type"] = type(error).__name__ + raise + finally: + for role, owned in report["owned_containers"].items(): + if role in report["cleanup_by_id"]: + continue + try: + report["cleanup_by_id"][role] = controls.cleanup_owned(cleanup_call, owned["id"], nonce) + except Exception as error: + report["cleanup_by_id"][role] = {"id": owned["id"], "failure_type": type(error).__name__} + workspace.remove(credential) + clean = all(value.get("absent") and value.get("volumes_absent") for value in report["cleanup_by_id"].values()) + if clean: + workspace.remove_database(database) + try: + for value in inputs.values(): + value.recheck() + report["original_inputs_unchanged"] = True + except Exception as error: + report["post_input_failure_type"] = type(error).__name__ + report["cleanup"] = { + "containers_and_volumes_absent": clean, + "private_database_removed": clean, + "credential_removed": True, + } + report["finished_at"] = datetime.now(timezone.utc).isoformat() + report["complete"] = ( + len(report["cleanup_by_id"]) == 3 + and clean + and report.get("consumer_complete") is True + and all(value.get("oom_killed") is False for value in report["cleanup_by_id"].values()) + and "failure_type" not in report + and "post_input_failure_type" not in report + ) + workspace.write("run.json", json.dumps(report, indent=2) + "\n") + workspace.check() + if not clean: + raise RuntimeError("owned physical cleanup incomplete") + if not report["complete"] and "failure_type" not in report: + raise RuntimeError("physical evidence incomplete") + + +def run(args): + if not __debug__: + raise ValueError("acceptance requires assertions enabled") + controls = controls_from_git(args) + with ExitStack() as stack: + inputs = controls.captured_inputs( + args, + stack, + source_files={ + "controller": "run_physical_readback.py", + "support": "run_response_loss.py", + "adapter": "physical_readback.py", + "verifier": "response_loss.py", + }, + data_files={ + "wheel": (args.wheel, 4 * 1024 * 1024, controls.WHEEL_SHA256), + "baseline": (args.baseline, 64 * 1024 * 1024, BASELINE_SHA256), + "seed": (args.seed, 256 * 1024, SEED_SHA256), + "snapshot_helper": (args.snapshot_helper, 128 * 1024, HELPER_SHA256), + "step_sql": (args.step_sql, 16 * 1024, STEP_SHA256), + }, + ) + workspace = stack.enter_context(controls.workspace_for(inputs, args.output)) + run_bound(args, inputs, controls, workspace) + + +def parse_args(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--runner-commit", required=True) + for name in ("wheel", "baseline", "seed", "snapshot-helper", "step-sql", "output"): + parser.add_argument("--" + name, type=Path, required=True) + return parser.parse_args() + + +if __name__ == "__main__": + run(parse_args()) diff --git a/scripts/conformance/command_keys/run_response_loss.py b/scripts/conformance/command_keys/run_response_loss.py index da57ce6..df620a5 100644 --- a/scripts/conformance/command_keys/run_response_loss.py +++ b/scripts/conformance/command_keys/run_response_loss.py @@ -124,7 +124,7 @@ def git_bytes(root, commit, relative): ) -def captured_inputs(args, stack): +def captured_inputs(args, stack, *, source_files=None, data_files=None): if re.fullmatch(r"[0-9a-f]{40}", args.runner_commit) is None: raise ValueError("runner commit must be one exact full Git SHA") root = Path(__file__).absolute().parents[3] @@ -139,27 +139,34 @@ def captured_inputs(args, stack): # Load the immutable Git object, never a second lookup of a mutable file. exec(compile(guard_bytes, "reviewed-input-guards.py", "exec"), guards.__dict__) inputs = {} - for role, filename in { + sources = { "controller": "run_response_loss.py", "guard": "input_guards.py", "adapter": "response_loss.py", "bootstrap": "consumer_bootstrap.py", "requirements": "consumer-requirements.txt", - }.items(): + "workspace": "workspace_guards.py", + } + if source_files: + sources.update(source_files) + for role, filename in sources.items(): expected = hashlib.sha256(git_bytes(root, head, prefix + filename)).hexdigest() inputs[role] = stack.enter_context( guards.CapturedInput(root / prefix / filename, role, maximum=256 * 1024, expected=expected) ) - inputs["proxy"] = stack.enter_context( - guards.CapturedInput(args.proxy, "proxy", maximum=64 * 1024, expected=PROXY_SHA256) - ) - inputs["wheel"] = stack.enter_context( - guards.CapturedInput(args.wheel, "wheel", maximum=4 * 1024 * 1024, expected=WHEEL_SHA256) + data = ( + data_files + if data_files is not None + else {"proxy": (args.proxy, 64 * 1024, PROXY_SHA256), "wheel": (args.wheel, 4 * 1024 * 1024, WHEEL_SHA256)} ) + for role, (path, maximum, expected) in data.items(): + inputs[role] = stack.enter_context(guards.CapturedInput(path, role, maximum=maximum, expected=expected)) return inputs -def consumer_result(stdout, inputs): +def consumer_result(stdout, inputs, *, scenario="response_loss"): + if scenario not in {"response_loss", "physical_readback"}: + raise ValueError("unknown installed consumer scenario") lines = [line[len(RESULT_PREFIX) :] for line in stdout.splitlines() if line.startswith(RESULT_PREFIX)] if len(lines) != 1: raise ValueError("consumer did not emit exactly one bound result") @@ -167,29 +174,92 @@ def consumer_result(stdout, inputs): if ( result.get("complete") is not True or result.get("adapter_sha256") != inputs["adapter"].sha256 - or result.get("proxy_sha256") != PROXY_SHA256 or result.get("sdk", {}).get("wheel_sha256") != WHEEL_SHA256 or result.get("sdk", {}).get("python") != PYTHON_VERSION or result.get("toolchain", {}).get("post_inventory_equal") is not True or result["toolchain"].get("requirements_sha256") != inputs["requirements"].sha256 ): raise ValueError("consumer result does not bind the captured inputs/toolchain") + if scenario == "response_loss" and result.get("proxy_sha256") != PROXY_SHA256: + raise ValueError("consumer proxy identity differs") + if scenario == "physical_readback" and result.get("seed_sha256") != inputs["seed"].sha256: + raise ValueError("consumer seed identity differs") return result -def run_bound(args, inputs): - root = args.output.absolute() - root.mkdir(mode=0o700, parents=False, exist_ok=False) - config = root / "docker-config" - config.mkdir(mode=0o700) - (config / "config.json").write_text('{"auths":{}}\n') - environment = {"PATH": os.defpath, "TMPDIR": str(root)} - docker = ["docker", "--config", str(config)] +def create_consumer(call, name, nonce, inputs): + return created_id( + call( + "create", + "--init", + "--read-only", + "--name", + name, + "--label", + OWNER_LABEL + "=sdk-python", + "--label", + RUN_LABEL + "=" + nonce, + "--network", + "host", + "--cpus", + "1", + "--memory", + "1g", + "--pids-limit", + "256", + "--cap-drop", + "ALL", + "--security-opt", + "no-new-privileges", + "--user", + f"{os.getuid()}:{os.getgid()}", + "--tmpfs", + "/tmp:mode=1777,size=256m", + "--tmpfs", + f"/opt/sdk-python:mode=0700,uid={os.getuid()},gid={os.getgid()},size=256m,exec", + "-i", + "--entrypoint", + "/usr/local/bin/python3.12", + "docker.io/library/python@" + PYTHON_INDEX, + "-I", + "-B", + "-c", + inputs["bootstrap"].data.decode(), + ).stdout + ) + + +def workspace_for(inputs, path): + module = ModuleType("reviewed-workspace-guards") + exec(compile(inputs["workspace"].data, "captured-workspace-guards.py", "exec"), module.__dict__) + return module.Workspace(path) + + +def run_bound(args, inputs, workspace): + config = workspace.directory("docker-config") + workspace.write("config.json", '{"auths":{}}\n', parent=config) + environment = {"PATH": os.defpath, "TMPDIR": workspace.proc(workspace.root)} + docker = ["docker", "--config", workspace.proc(config)] def call(*arguments, check=True, timeout=45): - return subprocess.run( + workspace.check() + result = subprocess.run( docker + list(arguments), check=check, capture_output=True, text=True, timeout=timeout, env=environment ) + # Capture create's returned ownership ID before the caller's next guard. + if arguments[0] != "create": + workspace.check() + return result + + def cleanup_call(*arguments, check=True, timeout=45): + return subprocess.run( + ["docker", "--config", "/proc/self/fd", *arguments], + check=check, + capture_output=True, + text=True, + timeout=timeout, + env=environment, + ) image = "ghcr.io/durable-workflow/server@" + EXPECTED_SERVER["index"] python_image = "docker.io/library/python@" + PYTHON_INDEX @@ -199,10 +269,9 @@ def call(*arguments, check=True, timeout=45): suffix = nonce[:12] names = ["sdk-python-216-response-loss-" + suffix, "sdk-python-216-consumer-" + suffix] token = secrets.token_hex(24) - credential = root / "fixture.env" - descriptor = os.open(credential, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) - with os.fdopen(descriptor, "w") as stream: - stream.write("DW_AUTH_TOKEN=" + token + "\nDW_RUNTIME_CREDENTIALS_ENABLED=false\nDB_CONNECTION=sqlite\n") + credential = workspace.write( + "fixture.env", "DW_AUTH_TOKEN=" + token + "\nDW_RUNTIME_CREDENTIALS_ENABLED=false\nDB_CONNECTION=sqlite\n" + ) report = { "started_at": datetime.now(timezone.utc).isoformat(), "runner_commit": args.runner_commit, @@ -243,7 +312,7 @@ def call(*arguments, check=True, timeout=45): "-p", "127.0.0.1::8080", "--env-file", - str(credential), + workspace.proc(credential["fd"]), "-e", "DB_DATABASE=/app/database/response-loss.sqlite", image, @@ -284,45 +353,7 @@ def call(*arguments, check=True, timeout=45): # Host networking is needed only for the fixture's numeric loopback. # It has no host mounts or inherited host environment. Only the synthetic # runtime token is delivered through the captured stdin bundle. - consumer_id = created_id( - call( - "create", - "--init", - "--read-only", - "--name", - names[1], - "--label", - OWNER_LABEL + "=sdk-python", - "--label", - RUN_LABEL + "=" + nonce, - "--network", - "host", - "--cpus", - "1", - "--memory", - "1g", - "--pids-limit", - "256", - "--cap-drop", - "ALL", - "--security-opt", - "no-new-privileges", - "--user", - f"{os.getuid()}:{os.getgid()}", - "--tmpfs", - "/tmp:mode=1777,size=256m", - "--tmpfs", - f"/opt/sdk-python:mode=0700,uid={os.getuid()},gid={os.getgid()},size=256m,exec", - "-i", - "--entrypoint", - "/usr/local/bin/python3.12", - python_image, - "-I", - "-B", - "-c", - inputs["bootstrap"].data.decode(), - ).stdout - ) + consumer_id = create_consumer(call, names[1], nonce, inputs) if consumer_id == runtime_id: raise ValueError("Docker create reused an existing container ID") report["owned_containers"]["consumer"] = {"id": consumer_id, "requested_name": names[1]} @@ -335,7 +366,7 @@ def call(*arguments, check=True, timeout=45): env=environment, timeout=600, ) - (root / "consumer.log").write_text(result.stdout + result.stderr) + workspace.write("consumer.log", result.stdout + result.stderr) consumer_state = inspect_owned(call, consumer_id, nonce) if consumer_state is None or consumer_state["State"]["Running"]: raise ValueError("owned consumer final state is unverified") @@ -350,7 +381,9 @@ def call(*arguments, check=True, timeout=45): if result.returncode != 0 or report["consumer_exit_code"] != 0: raise RuntimeError("installed consumer failed; retain its private incomplete evidence") consumer = consumer_result(result.stdout, inputs) - (root / "consumer.json").write_text(json.dumps(consumer, indent=2) + "\n") + workspace.write("consumer.json", json.dumps(consumer, indent=2) + "\n") + workspace.check() + report["workspace_identities_rechecked"] = True report["consumer_complete"] = True except Exception as error: report["failure_type"] = type(error).__name__ @@ -362,7 +395,7 @@ def call(*arguments, check=True, timeout=45): report["cleanup_by_id"] = {} for role, owned in report["owned_containers"].items(): try: - outcome = cleanup_owned(call, owned["id"], nonce) + outcome = cleanup_owned(cleanup_call, owned["id"], nonce) except Exception as error: report["cleanup_by_id"][role] = {"id": owned["id"], "failure_type": type(error).__name__} clean.append(False) @@ -374,10 +407,10 @@ def call(*arguments, check=True, timeout=45): except Exception as error: report["cleanup_failure_type"] = type(error).__name__ finally: - credential.unlink(missing_ok=True) + workspace.remove(credential) report["cleanup"] = { "containers_and_volumes_absent": len(clean) == 2 and all(clean), - "credential_removed": not credential.exists(), + "credential_removed": True, "oom_killed": ( True if any(value is True for value in oom) @@ -396,7 +429,8 @@ def call(*arguments, check=True, timeout=45): and "failure_type" not in report and "cleanup_failure_type" not in report ) - (root / "run.json").write_text(json.dumps(report, indent=2) + "\n") + workspace.write("run.json", json.dumps(report, indent=2) + "\n") + workspace.check() if not report["cleanup"]["containers_and_volumes_absent"]: raise RuntimeError("owned runtime cleanup incomplete") if not report["complete"] and "failure_type" not in report: @@ -406,7 +440,8 @@ def call(*arguments, check=True, timeout=45): def run(args): with ExitStack() as stack: inputs = captured_inputs(args, stack) - run_bound(args, inputs) + workspace = stack.enter_context(workspace_for(inputs, args.output)) + run_bound(args, inputs, workspace) def parse_args(): diff --git a/scripts/conformance/command_keys/workspace_guards.py b/scripts/conformance/command_keys/workspace_guards.py new file mode 100644 index 0000000..0a27e6d --- /dev/null +++ b/scripts/conformance/command_keys/workspace_guards.py @@ -0,0 +1,208 @@ +"""Held, no-follow fixture workspace identities and descriptor-relative outputs.""" + +from __future__ import annotations + +import hashlib +import os +import stat +from pathlib import Path + +NATIVE_PREFLIGHT = """set -eu +test -d "$1" && test ! -L "$1" || exit 78 +test -f "$1/$2" && test ! -L "$1/$2" || exit 78 +test "$(stat -c '%d:%i:%u:%g:%a' -- "$1")" = "$3" || exit 78 +test "$(stat -c '%d:%i:%u:%g:%a:%h' -- "$1/$2")" = "$4" || exit 78 +shift 4 +exec /usr/local/bin/durable-workflow-server "$@" +""" + + +def shape(info): + return info.st_dev, info.st_ino, info.st_mode, info.st_uid, info.st_gid, info.st_nlink + + +class Workspace: + def __init__(self, path): + self.path = Path(path).absolute() + self.handles = [] + self.directories = [] + self.files = {} + if ".." in self.path.parts: + raise ValueError("workspace path is not canonical") + try: + parent = os.open("/", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + self.handles.append(parent) + for index, part in enumerate(self.path.parts[1:]): + if index == len(self.path.parts) - 2: + os.mkdir(part, mode=0o700, dir_fd=parent) + descriptor = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=parent) + self.handles.append(descriptor) + info = os.fstat(descriptor) + sticky_root = info.st_uid == 0 and bool(info.st_mode & stat.S_ISVTX) + if info.st_uid not in {0, os.getuid()} or (info.st_mode & 0o022 and not sticky_root): + raise ValueError("workspace parent is not owner-controlled") + self.directories.append((parent, part, descriptor, shape(info)[:5])) + parent = descriptor + self.root = parent + self.check() + except BaseException: + self.close() + raise + + def check(self): + for parent, name, descriptor, expected in self.directories: + if ( + shape(os.fstat(descriptor))[:5] != expected + or shape(os.stat(name, dir_fd=parent, follow_symlinks=False))[:5] != expected + ): + raise ValueError("held workspace directory was replaced") + for item in self.files.values(): + self.check_file(item) + + def directory(self, name): + self.check() + self.name(name) + os.mkdir(name, mode=0o700, dir_fd=self.root) + descriptor = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=self.root) + self.handles.append(descriptor) + self.directories.append((self.root, name, descriptor, shape(os.fstat(descriptor))[:5])) + self.check() + return descriptor + + @staticmethod + def name(name): + if not name or name in {".", ".."} or "/" in name or "\x00" in name: + raise ValueError("workspace entry must be one literal basename") + + @staticmethod + def proc(descriptor, name=None): + return "/proc/" + str(os.getpid()) + "/fd/" + str(descriptor) + ("/" + name if name else "") + + def mount_path(self, directory): + self.check() + entry = next(value for value in self.directories if value[2] == directory) + if entry[0] != self.root: + raise ValueError("bind source is not one held workspace child") + return str(self.path / entry[1]) + + def native_arguments(self, directory, database, *arguments): + self.check() + folder = os.fstat(directory) + file = os.fstat(database["fd"]) + directory_key = ( + f"{folder.st_dev}:{folder.st_ino}:{folder.st_uid}:{folder.st_gid}:{stat.S_IMODE(folder.st_mode):o}" + ) + file_key = ( + f"{file.st_dev}:{file.st_ino}:{file.st_uid}:{file.st_gid}:{stat.S_IMODE(file.st_mode):o}:{file.st_nlink}" + ) + return [ + "-c", + NATIVE_PREFLIGHT, + "checked-native", + "/cohort", + database["name"], + directory_key, + file_key, + *arguments, + ] + + def check_file(self, item): + current = os.stat(item["name"], dir_fd=item["parent"], follow_symlinks=False) + held = os.fstat(item["fd"]) + if shape(current) != item["shape"] or shape(held) != item["shape"]: + raise ValueError("held workspace file was replaced") + if held.st_size > item["maximum"]: + raise ValueError("workspace file exceeded its bound") + if item["sha256"] is not None: + data = os.pread(item["fd"], item["maximum"] + 1, 0) + if hashlib.sha256(data).hexdigest() != item["sha256"]: + raise ValueError("held workspace bytes changed") + + def capture(self, parent, name, *, immutable=False, maximum=64 * 1024 * 1024): + self.check() + self.name(name) + descriptor = os.open(name, os.O_RDWR | os.O_NOFOLLOW, dir_fd=parent) + return self.remember(parent, name, descriptor, immutable=immutable, maximum=maximum) + + def remember(self, parent, name, descriptor, *, immutable, maximum, initial_sha256=None): + self.handles.append(descriptor) + info = os.fstat(descriptor) + if ( + not stat.S_ISREG(info.st_mode) + or info.st_uid != os.getuid() + or info.st_nlink != 1 + or info.st_mode & 0o077 + or info.st_size > maximum + ): + raise ValueError("workspace file is not private, regular and single-link") + item = { + "parent": parent, + "name": name, + "fd": descriptor, + "shape": shape(info), + "maximum": maximum, + "sha256": hashlib.sha256(os.pread(descriptor, maximum + 1, 0)).hexdigest() if immutable else None, + } + if ( + initial_sha256 is not None + and hashlib.sha256(os.pread(descriptor, maximum + 1, 0)).hexdigest() != initial_sha256 + ): + raise ValueError("created workspace bytes differ") + self.files[(parent, name)] = item + self.check() + return item + + def write(self, name, data, *, parent=None, immutable=True): + self.check() + self.name(name) + parent = self.root if parent is None else parent + data = data.encode() if isinstance(data, str) else data + descriptor = os.open(name, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=parent) + try: + with os.fdopen(descriptor, "wb", closefd=False) as stream: + stream.write(data) + stream.flush() + os.fsync(descriptor) + except BaseException: + os.close(descriptor) + raise + item = self.remember( + parent, + name, + descriptor, + immutable=immutable, + maximum=max(64 * 1024 * 1024, len(data)), + initial_sha256=hashlib.sha256(data).hexdigest(), + ) + self.check() + return item + + def remove(self, item): + # Cleanup authority comes from the held parent/inode even if an outer + # pathname was renamed. A replacement entry is preserved. + self.check_file(item) + os.unlink(item["name"], dir_fd=item["parent"]) + del self.files[(item["parent"], item["name"])] + + def remove_database(self, directory): + self.check() + for name in os.listdir(directory): + item = self.files.get((directory, name)) + if item is None: + item = self.capture(directory, name) + self.remove(item) + entry = next(value for value in self.directories if value[2] == directory) + self.check() + os.rmdir(entry[1], dir_fd=entry[0]) + self.directories.remove(entry) + + def close(self): + for descriptor in reversed(self.handles): + os.close(descriptor) + self.handles.clear() + + def __enter__(self): + return self + + def __exit__(self, *_error): + self.close() diff --git a/tests/test_command_keys_workspace.py b/tests/test_command_keys_workspace.py new file mode 100644 index 0000000..000196c --- /dev/null +++ b/tests/test_command_keys_workspace.py @@ -0,0 +1,180 @@ +from __future__ import annotations + +import importlib.util +import os +import sqlite3 +from pathlib import Path + +import pytest + +PATH = Path(__file__).parents[1] / "scripts/conformance/command_keys/workspace_guards.py" +SPEC = importlib.util.spec_from_file_location("command_workspace", PATH) +assert SPEC and SPEC.loader +module = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(module) + + +def test_root_replacement_is_seen_and_descriptor_bind_keeps_original_database(tmp_path): + root = tmp_path / "work" + with module.Workspace(root) as workspace: + directory = workspace.directory("database") + workspace.write("copied.sqlite", b"original", parent=directory, immutable=False) + binding = Path(workspace.proc(directory)) / "copied.sqlite" + root.rename(tmp_path / "retained-original") + (root / "database").mkdir(parents=True) + (root / "database/copied.sqlite").write_bytes(b"substituted") + # The predecessor mount/snapshot path follows the replacement. + assert (root / "database/copied.sqlite").read_bytes() == b"substituted" + assert binding.read_bytes() == b"original" + with pytest.raises(ValueError, match="directory was replaced"): + workspace.check() + with pytest.raises(ValueError): + workspace.write("run.json", '{"complete":true}') + assert not (root / "run.json").exists() + + +@pytest.mark.parametrize("entry", ["database", "docker-config"]) +def test_subdirectory_rename_replacement_refuses_use(tmp_path, entry): + root = tmp_path / "work" + with module.Workspace(root) as workspace: + workspace.directory(entry) + (root / entry).rename(root / (entry + "-original")) + (root / entry).mkdir() + with pytest.raises(ValueError): + workspace.check() + + +@pytest.mark.parametrize("name", ["fixture.env", "consumer.log", "consumer.json", "run.json"]) +@pytest.mark.parametrize("replacement", ["rename", "symlink", "hardlink", "rewrite"]) +def test_workspace_output_and_credential_substitution_refuses(tmp_path, name, replacement): + root = tmp_path / "work" + with module.Workspace(root) as workspace: + item = workspace.write(name, b"original") + path = root / name + if replacement == "rewrite": + path.write_bytes(b"changed!") + else: + path.rename(root / (name + "-original")) + outside = tmp_path / "outside" + outside.write_bytes(b"foreign!") + if replacement == "rename": + path.write_bytes(b"foreign!") + path.chmod(0o600) + elif replacement == "symlink": + path.symlink_to(outside) + else: + os.link(outside, path) + # A pathname-only reader accepts the changed bytes; the held identity refuses. + assert path.read_bytes() != b"original" + with pytest.raises(ValueError): + workspace.check() + with pytest.raises(ValueError): + workspace.remove(item) + assert path.exists() + + +def test_evidence_writer_refuses_existing_symlink_without_touching_marker(tmp_path): + root = tmp_path / "work" + marker = tmp_path / "marker" + marker.write_bytes(b"preserve") + with module.Workspace(root) as workspace: + (root / "run.json").symlink_to(marker) + with pytest.raises(FileExistsError): + workspace.write("run.json", '{"complete":true}') + assert marker.read_bytes() == b"preserve" + + +def test_created_file_descriptor_is_held_without_reopen(tmp_path, monkeypatch): + root = tmp_path / "work" + with module.Workspace(root) as workspace: + original = workspace.remember + + def replace_before_remember(parent, name, descriptor, **kwargs): + path = root / name + path.rename(root / "original") + path.write_bytes(b"false-output") + path.chmod(0o600) + return original(parent, name, descriptor, **kwargs) + + monkeypatch.setattr(workspace, "remember", replace_before_remember) + with pytest.raises(ValueError): + workspace.write("consumer.json", b"true-output") + + +def test_mutable_database_keeps_inode_across_legitimate_sqlite_updates(tmp_path): + root = tmp_path / "work" + with module.Workspace(root) as workspace: + directory = workspace.directory("database") + workspace.write("copied.sqlite", b"", parent=directory, immutable=False) + path = Path(workspace.proc(directory, "copied.sqlite")) + with sqlite3.connect(path) as database: + database.execute("CREATE TABLE values_table(value INTEGER)") + database.execute("INSERT INTO values_table VALUES (1)") + workspace.check() + with sqlite3.connect(path.as_uri() + "?mode=ro", uri=True) as database: + assert database.execute("SELECT value FROM values_table").fetchall() == [(1,)] + workspace.check() + original = root / "database/copied.sqlite" + original.rename(root / "database/old.sqlite") + original.write_bytes(b"different") + original.chmod(0o600) + with pytest.raises(ValueError): + workspace.check() + + +def test_workspace_parent_replacement_refuses_before_create(tmp_path): + parent = tmp_path / "parent" + parent.mkdir(mode=0o700) + with module.Workspace(parent / "work") as workspace: + parent.rename(tmp_path / "retained-parent") + (parent / "work").mkdir(parents=True) + with pytest.raises(ValueError): + workspace.write("consumer.log", "must not write") + assert not (parent / "work/consumer.log").exists() + + +@pytest.mark.parametrize("replacement", ["none", "file", "directory", "symlink", "hardlink"]) +def test_native_preflight_refuses_substitution_before_native_execution(tmp_path, replacement): + import subprocess + + root = tmp_path / "work" + marker = tmp_path / "native-executed" + with module.Workspace(root) as workspace: + directory = workspace.directory("database") + database = workspace.write("copied.sqlite", b"original", parent=directory, immutable=False) + arguments = workspace.native_arguments(directory, database, str(marker)) + # Keep the actual inode/type preflight; substitute a harmless stand-in + # only for the immutable image's native executable in this unit control. + arguments[1] = arguments[1].replace("exec /usr/local/bin/durable-workflow-server", "exec /usr/bin/touch") + arguments[3] = str(root / "database") + path = root / "database/copied.sqlite" + if replacement == "directory": + (root / "database").rename(root / "original-database") + (root / "database").mkdir(mode=0o700) + path.write_bytes(b"foreign") + elif replacement != "none": + path.rename(root / "database/original.sqlite") + if replacement == "file": + path.write_bytes(b"foreign") + path.chmod(0o600) + elif replacement == "symlink": + path.symlink_to(root / "database/original.sqlite") + else: + os.link(root / "database/original.sqlite", path) + result = subprocess.run(["/bin/sh", *arguments], capture_output=True, text=True) + if replacement == "none": + assert result.returncode == 0 and marker.exists() + else: + assert result.returncode == 78 and not marker.exists() + # The predecessor entrypoint has no independent mounted-inode gate. + subprocess.run(["/usr/bin/touch", str(marker)], check=True) + assert marker.exists() + + +def test_cleanup_configuration_is_an_immutable_empty_kernel_namespace(): + # proc's FD directory contains only numeric names and cannot be seeded with + # a config.json after the controller's workspace/config path is replaced. + path = Path("/proc/self/fd/config.json") + assert not path.exists() + with pytest.raises(OSError): + path.write_text("must not create") diff --git a/tests/test_physical_command_key_readback.py b/tests/test_physical_command_key_readback.py new file mode 100644 index 0000000..d8d08fa --- /dev/null +++ b/tests/test_physical_command_key_readback.py @@ -0,0 +1,209 @@ +from __future__ import annotations + +import copy +import hashlib +import importlib.util +import sqlite3 +from pathlib import Path + +import pytest + +PATH = Path(__file__).parents[1] / "scripts/conformance/command_keys/physical_readback.py" +SPEC = importlib.util.spec_from_file_location("physical_readback", PATH) +assert SPEC is not None and SPEC.loader is not None +consumer = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(consumer) +CONTROLLER_SPEC = importlib.util.spec_from_file_location( + "physical_controller", PATH.with_name("run_physical_readback.py") +) +assert CONTROLLER_SPEC is not None and CONTROLLER_SPEC.loader is not None +controller = importlib.util.module_from_spec(CONTROLLER_SPEC) +CONTROLLER_SPEC.loader.exec_module(controller) +GUARD_SPEC = importlib.util.spec_from_file_location("physical_input_guards", PATH.with_name("input_guards.py")) +assert GUARD_SPEC is not None and GUARD_SPEC.loader is not None +guards = importlib.util.module_from_spec(GUARD_SPEC) +GUARD_SPEC.loader.exec_module(guards) +BOOTSTRAP_SPEC = importlib.util.spec_from_file_location("physical_bootstrap", PATH.with_name("consumer_bootstrap.py")) +assert BOOTSTRAP_SPEC is not None and BOOTSTRAP_SPEC.loader is not None +bootstrap = importlib.util.module_from_spec(BOOTSTRAP_SPEC) +BOOTSTRAP_SPEC.loader.exec_module(bootstrap) + + +def seed(): + cases = [] + for kind in ("signal", "update"): + for state in ("open", "completed", "continued"): + identity = kind + "-" + state + receipt = {"workflow_id": identity, "command_id": "command-" + identity, "run_id": "run-" + identity} + body = {} + if kind == "update": + receipt.update({"update_id": "update-" + identity, "accepted_at": "2026-01-01T00:00:00Z"}) + body["request_id"] = "original-update-key" + cases.append( + { + "kind": kind, + "state": state, + "workflow_id": identity, + "receipt": receipt, + "original_body": body, + "history": {"events": [], "next_page_token": None}, + } + ) + return { + "schema": "durable-workflow.command-key-physical-seed.v1", + "manifest_sha256": consumer.MANIFEST_SHA256, + "cases": cases, + } + + +def test_accepts_real_unkeyed_signal_compatibility_without_inventing_keys(): + cases = consumer.seed_cases(seed()) + assert all("request_id" not in case["original_body"] for case in cases if case["kind"] == "signal") + + +@pytest.mark.parametrize( + "mutation", + [ + lambda value: value.update(manifest_sha256="unreviewed"), + lambda value: value["cases"].pop(), + lambda value: value["cases"].append(copy.deepcopy(value["cases"][0])), + lambda value: value["cases"][0].update(state="unsupported"), + lambda value: value["cases"][0]["receipt"].pop("command_id"), + lambda value: value["cases"][0]["receipt"].update(workflow_id="other"), + lambda value: value["cases"][3]["original_body"].pop("request_id"), + lambda value: value["cases"][0]["history"].update(next_page_token="unfinished"), + ], +) +def test_incomplete_or_wrong_seed_refuses_before_client_creation(mutation): + document = seed() + mutation(document) + with pytest.raises(ValueError): + consumer.seed_cases(document) + + +@pytest.mark.parametrize( + "mutation", + [ + {"retained": False}, + {"intent_verified": True}, + {"intent_verified": None}, + {"command_id": "other"}, + {"run_id": "successor"}, + {"update_id": "other"}, + {"accepted_at": "changed"}, + ], +) +def test_legacy_receipt_cannot_claim_verification_or_change_original_identities(mutation): + original = seed()["cases"][3]["receipt"] + received = {**original, "retained": True, "intent_verified": False} + consumer.legacy_receipt(original, received) + with pytest.raises(ValueError): + consumer.legacy_receipt(original, {**received, **mutation}) + + +def test_input_hash_and_regular_file_refusal(tmp_path): + path = tmp_path / "synthetic" + path.write_bytes(b"original") + digest = hashlib.sha256(path.read_bytes()).hexdigest() + with guards.CapturedInput(path, "baseline", maximum=100, expected=digest) as captured: + assert captured.data == b"original" + path.write_bytes(b"changed") + with pytest.raises(ValueError, match="input refused"): + guards.CapturedInput(path, "baseline", maximum=100, expected=digest) + link = tmp_path / "linked" + link.symlink_to(path) + with pytest.raises(ValueError, match="input refused"): + guards.CapturedInput(link, "baseline", maximum=100, expected=hashlib.sha256(path.read_bytes()).hexdigest()) + + +def test_physical_bundle_closes_seed_and_verifier_roles(): + import base64 + + sources = { + role: {"base64": base64.b64encode(b"original").decode(), "sha256": hashlib.sha256(b"original").hexdigest()} + for role in bootstrap.PHYSICAL_ROLES + } + assert bootstrap.decode_inputs({"scenario": "physical_readback", "sources": sources})["seed"] == b"original" + sources["proxy"] = sources["seed"] + with pytest.raises(ValueError, match="roles differ"): + bootstrap.decode_inputs({"scenario": "physical_readback", "sources": sources}) + with pytest.raises(ValueError, match="roles differ"): + bootstrap.decode_inputs({"scenario": "unknown", "sources": sources}) + + +def test_schema_authority_uses_native_marker_not_sqlite_user_version(): + with sqlite3.connect(":memory:") as connection: + connection.execute("PRAGMA user_version=12") + connection.execute("CREATE TABLE dw_server_schema(engine,version)") + connection.execute("INSERT INTO dw_server_schema VALUES ('rust-development',13)") + controller.native_marker(connection, 13) + with pytest.raises(ValueError, match="marker differs"): + controller.native_marker(connection, 12) + connection.execute("INSERT INTO dw_server_schema VALUES ('rust-development',13)") + with pytest.raises(ValueError, match="marker differs"): + controller.native_marker(connection, 13) + + +def upgrade_snapshots(): + columns = ["version", "description", "installed_on", "success", "checksum", "execution_time"] + before = { + "schema": [("table", "business", "business", "business schema")], + "user_version": 0, + "application_id": 0, + "tables": { + "business": {"columns": ["id"], "rows": [[["text", "preserve"]]]}, + "dw_server_schema": { + "columns": ["engine", "version"], + "rows": [[["text", "rust-development"], ["integer", 12]]], + }, + "_sqlx_migrations": {"columns": columns, "rows": []}, + }, + } + added = [("table", "dw_command_intents", "dw_command_intents", "tagged schema")] + after = copy.deepcopy(before) + after["schema"] = sorted(before["schema"] + added) + after["tables"]["dw_server_schema"]["rows"][0][1] = ["integer", 13] + after["tables"]["dw_command_intents"] = {"columns": ["command_id", "namespace", "intent_digest"], "rows": []} + after["tables"]["_sqlx_migrations"]["rows"] = [ + [ + ["integer", 13], + ["text", controller.JOURNAL_DESCRIPTION], + ["text", "2026-01-01 00:00:01"], + ["integer", 1], + ["blob", controller.JOURNAL_SHA384], + ["integer", 0], + ] + ] + return before, after, added + + +@pytest.mark.parametrize( + "mutation", + [ + lambda value: value["tables"]["business"]["rows"].clear(), + lambda value: value.update(user_version=13), + lambda value: value["tables"]["dw_command_intents"]["rows"].append([["text", "unexpected"]]), + lambda value: value["tables"]["_sqlx_migrations"]["rows"][0].__setitem__(0, ["integer", 12]), + lambda value: value["tables"]["_sqlx_migrations"]["rows"][0].__setitem__(1, ["text", "other description"]), + lambda value: value["tables"]["_sqlx_migrations"]["rows"][0].__setitem__(2, ["text", "2025-01-01 00:00:00"]), + lambda value: value["tables"]["_sqlx_migrations"]["rows"][0].__setitem__(3, ["integer", 0]), + lambda value: value["tables"]["_sqlx_migrations"]["rows"][0].__setitem__(4, ["blob", "wrong checksum"]), + lambda value: value["tables"]["_sqlx_migrations"]["rows"][0].__setitem__(5, ["integer", 1]), + ], +) +def test_upgrade_preserves_all_business_rows_and_closes_every_journal_field(mutation): + before, after, added = upgrade_snapshots() + began, finished = 1767225600.5, 1767225602.5 + controller.validate_upgrade(before, after, added, began, finished) + mutation(after) + with pytest.raises(ValueError): + controller.validate_upgrade(before, after, added, began, finished) + + +def test_unapproved_helper_bytes_cannot_execute(tmp_path): + script = tmp_path / "helper.py" + marker = tmp_path / "executed" + script.write_text(f"from pathlib import Path\nPath({str(marker)!r}).touch()\n") + with pytest.raises(ValueError, match="source identity differs"): + controller.load_bytes(script.read_bytes(), "unapproved_fixture", "0" * 64) + assert not marker.exists()