diff --git a/scripts/conformance/command_keys/README.md b/scripts/conformance/command_keys/README.md new file mode 100644 index 0000000..84655be --- /dev/null +++ b/scripts/conformance/command_keys/README.md @@ -0,0 +1,83 @@ +# Installed command-response recovery + +This adapter exercises the published Python Client and actual Worker for +[shared command-key acceptance](https://github.com/durable-workflow/.github/issues/216). +It creates synthetic workflows and a disposable SQLite runtime. Its transport +fault comes from the independently reviewed shared loopback fixture. + +The controller binds the qualified native image/source/platform from +[Server's publication receipt](https://github.com/durable-workflow/server/pull/1209#issuecomment-6858140177). +Its interpreter is the published official Python image index +`sha256:935c3c5b12357a7e10caa9c1e6ca5692344d97cf229832ef584be2cec9fc03df`, +amd64 child `sha256:0f086103b026048cc50cb5da25ec435c023c5241a83fc44daea593377abc61c4`. +It starts a fresh private virtualenv in that image and installs all nine registry +packages from `consumer-requirements.txt`, with exact wheel-only hashes. The SDK +wheel is the reviewed published artifact. The interpreter/config and entire +installed environment are fingerprinted before and after the consumer. + +Download the published SDK wheel identified by `WHEEL_SHA256` in the controller. +Download the shared proxy from +[`117118c8`](https://github.com/durable-workflow/.github/blob/117118c83093822469b3824ba9587ff4068ed29f/conformance/command-keys/committed_response_loss.py). +Its SHA-256 is +`bd1e8ccbbf76f63fc2eafa278a023eb181d405b68d9c30016392e395a1eff725`. +Store both as regular, single-link, owned files in owner-controlled directories. +Before the experiment, check host/Docker capacity and pull both exact images +with an empty Docker config. The controller itself does not pull images. + +Run from an exact committed checkout: + +```sh +python3 -I scripts/conformance/command_keys/run_response_loss.py \ + --runner-commit FULL_REVIEWED_RUNNER_COMMIT \ + --wheel /path/to/downloaded/durable_workflow-2.17.2-py3-none-any.whl \ + --proxy /path/to/verified/committed_response_loss.py \ + --output /path/to/private-disk-scratch/new-run +``` + +The controller and executable source inputs must equal that commit's Git blobs. +It opens inputs without following symlinks, checks owner/mode/single-link/file +bounds and retains descriptors plus captured bytes. The adapter, bootstrap and +proxy execute from those bytes. Wheel inspection uses the same captured bytes. +Pathname, inode, content and parent-directory identities are checked again before +launch and after return. A substituted program cannot certify its own output by +reporting the original hash. The host does not supply an interpreter or virtualenv. + +The new private output directory must not exist; its parent must already exist. +One runtime and one consumer are each capped at 1 CPU, 1 GiB and 256 PIDs, with +read-only roots and `--init`. Bounded tmpfs paths remain within each memory cap. +The consumer's ordinary scratch is noexec. Its separate private wheel environment +permits native-extension mapping for the hash-locked codec wheel; it is not a +host mount and accepts no unverified executable input. +The runtime publishes one ephemeral loopback port. The consumer uses host +networking for the numeric loopback-only proxy and has no host mounts. Registry +installation precedes creation of its synthetic credential file. Its entire +bootstrap/input bundle travels through stdin; executable files are never reopened +from host paths. Controller readiness explicitly disables environment proxies. +The consumer has a ten-minute bound, and cleanup owns both containers, anonymous +volumes and the generated fixture credential. +Creation returns full container IDs bound to per-run ownership labels. Start, +inspection, port lookup, removal and absence verification use those IDs. A reused +human name grants no cleanup authority over its replacement or volumes. + +Six workflows exercise eight intentionally lost complete upstream successes: +automatic/manual recovery with generated/explicit keys on both routes, including +continuation. Sender and proxy request bytes agree, four concurrent identical +retries retain one admission, and changed intent gets a typed conflict. Business +execution/admission counts and closed/continued original identities are checked. +Explicit keys contain 255 Unicode code points and exceed 255 UTF-8 bytes. Nested +typed values include integers, booleans, bytes and Unicode. Query receipts are +read through their `result` field. + +Both `consumer.json` and `run.json` must report complete success, with captured +input and interpreter rechecks, equal environment inventories, no OOM and exact +cleanup, before qualification. Record the reviewed runner commit, input hashes, +tuple, timestamps and sanitized outcomes in GitHub. Keep failed attempts as +NONPASS and diagnostic captures private and temporary. The earlier host-venv +observation is retained under its Security provenance HOLD, not reused as the +corrected runner's qualification. + +Synthetic legacy transformations and physical upgrades remain separate cells. +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. diff --git a/scripts/conformance/command_keys/consumer-requirements.txt b/scripts/conformance/command_keys/consumer-requirements.txt new file mode 100644 index 0000000..dfd01a9 --- /dev/null +++ b/scripts/conformance/command_keys/consumer-requirements.txt @@ -0,0 +1,9 @@ +anyio==4.16.0 --hash=sha256:c66ae634e0b8228396b801741beb3294326d4b6102f7bba4041c0dfb58c29026 +certifi==2026.10.23 --hash=sha256:e9b54de76e530bb6cf0ccd036787103acb58fd42ba930ea8bf948ddbaa2f61d1 +durable-workflow==2.17.2 --hash=sha256:7dc77d3882fa5cbddbeaee6a4cc63b05285e280523c8ea40dc31e03cc12a40bc +fastavro==1.13.1 --hash=sha256:2f56a127d71e45083306d2650efff827cad0f4b0744dd42cb69c631d77943b1d +h11==0.16.0 --hash=sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86 +httpcore==1.0.9 --hash=sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55 +httpx==0.28.1 --hash=sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad +idna==3.20 --hash=sha256:ab7ae7122974553370f0bdb919e1a960b2cd1bc1ef0276416d896db81c14582c +typing_extensions==4.16.0 --hash=sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8 diff --git a/scripts/conformance/command_keys/consumer_bootstrap.py b/scripts/conformance/command_keys/consumer_bootstrap.py new file mode 100644 index 0000000..21067db --- /dev/null +++ b/scripts/conformance/command_keys/consumer_bootstrap.py @@ -0,0 +1,219 @@ +"""Trusted argv bootstrap inside the fixed published Python image. + +No host interpreter, virtualenv, source directory or credential directory is +mounted. The controller sends only already captured inputs and synthetic data. +""" + +from __future__ import annotations + +import asyncio +import base64 +import hashlib +import json +import linecache +import os +import stat +import subprocess +import sys +import tempfile +import venv +from pathlib import Path +from types import ModuleType, SimpleNamespace + +RESULT_PREFIX = "SDK_PYTHON_CONSUMER_RESULT=" +ROLES = {"adapter", "bootstrap", "proxy", "wheel", "requirements"} +MAX_BUNDLE_BYTES = 8 * 1024 * 1024 +IMAGE_INTERPRETER = Path("/usr/local/bin/python3.12") + + +def approved_environment_link(relative, target): + return ( + (relative == "lib64" and target == "lib") + or (relative in {"bin/python", "bin/python3"} and target == "python3.12") + or (relative == "bin/python3.12" and target == str(IMAGE_INTERPRETER)) + ) + + +def decode_inputs(bundle): + if set(bundle["sources"]) != ROLES: + raise ValueError("captured source roles differ") + result = {} + for role, entry in bundle["sources"].items(): + data = base64.b64decode(entry["base64"], validate=True) + if hashlib.sha256(data).hexdigest() != entry["sha256"]: + raise ValueError("captured bytes differ: " + role) + result[role] = data + return result + + +def execute_source(data, name): + module = ModuleType(name) + module.__file__ = "captured-" + name + ".py" + sys.modules[name] = module + linecache.cache[module.__file__] = (len(data), None, data.decode().splitlines(keepends=True), module.__file__) + # The bytes were captured before runtime start. No executable input path + # is reopened or imported through a loader. + exec(compile(data, module.__file__, "exec"), module.__dict__) + return module + + +def inventory(root): + entries = [] + total = 0 + for path in sorted(root.rglob("*")): + relative = str(path.relative_to(root)) + info = path.lstat() + if stat.S_ISDIR(info.st_mode): + continue + if path.is_symlink(): + target = os.readlink(path) + if not approved_environment_link(relative, target): + raise ValueError("unexpected installed-environment link") + entries.append([relative, "symlink", target]) + continue + if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_uid != os.getuid(): + raise ValueError("unexpected installed-environment file") + total += info.st_size + if total > 128 * 1024 * 1024 or len(entries) > 10_000: + raise ValueError("installed-environment inventory exceeds bound") + entries.append([relative, stat.S_IMODE(info.st_mode), hashlib.sha256(path.read_bytes()).hexdigest()]) + digest = hashlib.sha256(json.dumps(entries, separators=(",", ":")).encode()).hexdigest() + return {"sha256": digest, "files": len(entries), "bytes": total} + + +def prepare(bundle, sources): + if "prepared" in bundle: + raise ValueError("caller may not supply a prepared interpreter") + root = Path(tempfile.mkdtemp(prefix="sdk-python-installed-")) + environment = Path(tempfile.mkdtemp(prefix="environment-", dir="/opt/sdk-python")) / "venv" + # Scratch remains noexec. These fresh internal links may target only the + # immutable interpreter in the fixed image, never a host/caller program. + venv.EnvBuilder(with_pip=True, symlinks=True).create(environment) + lock = root / "requirements.txt" + lock.write_bytes(sources["requirements"]) + lock.chmod(0o400) + python = environment / "bin/python" + executable = python.lstat() + if ( + not stat.S_ISLNK(executable.st_mode) + or executable.st_nlink != 1 + or executable.st_uid != os.getuid() + or python.resolve() != IMAGE_INTERPRETER + ): + raise ValueError("fresh interpreter identity differs") + image_binary = IMAGE_INTERPRETER.lstat() + if ( + not stat.S_ISREG(image_binary.st_mode) + or image_binary.st_uid != 0 + or image_binary.st_nlink != 1 + or image_binary.st_mode & 0o022 + ): + raise ValueError("immutable image interpreter shape differs") + binary_sha256 = hashlib.sha256(python.read_bytes()).hexdigest() + if binary_sha256 != hashlib.sha256(IMAGE_INTERPRETER.read_bytes()).hexdigest(): + raise ValueError("virtualenv interpreter differs from fixed image interpreter") + clean_environment = {"PATH": os.defpath, "TMPDIR": str(root), "PIP_CONFIG_FILE": "/dev/null"} + subprocess.run( + [ + str(python), + "-I", + "-B", + "-m", + "pip", + "--isolated", + "install", + "--no-compile", + "--no-cache-dir", + "--only-binary=:all:", + "--require-hashes", + "--disable-pip-version-check", + "--progress-bar", + "off", + "--index-url", + "https://pypi.org/simple", + "-r", + str(lock), + ], + check=True, + env=clean_environment, + timeout=180, + ) + before = inventory(environment) + bundle["prepared"] = { + "root": str(environment), + "scratch": str(root), + "binary_sha256": binary_sha256, + "config_sha256": hashlib.sha256((environment / "pyvenv.cfg").read_bytes()).hexdigest(), + "inventory": before, + } + # Both the child program and its inputs come from the same captured bytes. + # The interpreter links here to the immutable image, never supplied + # by a host pathname. Its complete environment is checked around the call. + result = subprocess.run( + [str(python), "-I", "-B", "-c", sources["bootstrap"].decode(), "--prepared"], + input=json.dumps(bundle).encode(), + env=clean_environment, + timeout=300, + ) + if inventory(environment) != before: + raise ValueError("installed environment changed") + raise SystemExit(result.returncode) + + +def main(): + raw = sys.stdin.buffer.read(MAX_BUNDLE_BYTES + 1) + if len(raw) > MAX_BUNDLE_BYTES: + raise ValueError("captured bundle exceeds bound") + bundle = json.loads(raw) + sources = decode_inputs(bundle) + if sys.argv[1:] != ["--prepared"]: + prepare(bundle, sources) + prepared = bundle["prepared"] + environment = Path(prepared["root"]) + if Path(sys.prefix) != environment or Path(sys.executable).parent != environment / "bin": + raise ValueError("consumer did not start in its private registry environment") + if ( + hashlib.sha256(Path(sys.executable).read_bytes()).hexdigest() != prepared["binary_sha256"] + or hashlib.sha256((environment / "pyvenv.cfg").read_bytes()).hexdigest() != prepared["config_sha256"] + ): + raise ValueError("prepared interpreter/config identity changed") + before = inventory(environment) + if before != prepared["inventory"]: + raise ValueError("prepared environment changed before consumer execution") + root = Path(prepared["scratch"]) + credential = root / "fixture.env" + 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") + try: + result = asyncio.run( + adapter.main( + SimpleNamespace( + upstream=bundle["upstream"], + credential_file=credential, + proxy_data=sources["proxy"], + wheel_data=sources["wheel"], + adapter_sha256=bundle["sources"]["adapter"]["sha256"], + server_version=bundle["server_version"], + output=root / "consumer.json", + ) + ) + ) + finally: + credential.unlink(missing_ok=True) + after = inventory(environment) + if before != after: + raise ValueError("installed environment changed during consumer execution") + result["toolchain"] = { + "interpreter_sha256": prepared["binary_sha256"], + "venv_config_sha256": prepared["config_sha256"], + "environment_inventory": before, + "post_inventory_equal": True, + "requirements_sha256": bundle["sources"]["requirements"]["sha256"], + } + print(RESULT_PREFIX + json.dumps(result), flush=True) + + +if __name__ == "__main__": + main() diff --git a/scripts/conformance/command_keys/input_guards.py b/scripts/conformance/command_keys/input_guards.py new file mode 100644 index 0000000..c60887d --- /dev/null +++ b/scripts/conformance/command_keys/input_guards.py @@ -0,0 +1,95 @@ +"""Capture reviewed local inputs once and detect pathname/inode substitutions.""" + +from __future__ import annotations + +import hashlib +import os +import stat +from pathlib import Path + + +def identity(info): + return ( + info.st_dev, + info.st_ino, + info.st_mode, + info.st_uid, + info.st_gid, + info.st_nlink, + info.st_size, + info.st_mtime_ns, + info.st_ctime_ns, + ) + + +def directory_identity(info): + return (info.st_dev, info.st_ino, info.st_mode, info.st_uid, info.st_gid) + + +class CapturedInput: + """Retain no-follow descriptors and immutable bytes for one bounded input.""" + + def __init__(self, path, role, *, maximum, expected=None): + self.path = Path(path).absolute() + self.role = role + self.handles = [] + self.parents = [] + if ".." in self.path.parts: + raise ValueError("input path must be canonical") + try: + parent = os.open("/", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + self.handles.append(parent) + for part in self.path.parts[1:-1]: + 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("input parent is not owner-controlled") + self.parents.append((parent, part, descriptor, directory_identity(info))) + parent = descriptor + self.parent = parent + self.descriptor = os.open(self.path.name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=parent) + self.handles.append(self.descriptor) + info = os.fstat(self.descriptor) + if ( + not stat.S_ISREG(info.st_mode) + or info.st_uid != os.getuid() + or info.st_nlink != 1 + or info.st_mode & 0o022 + or not 0 < info.st_size <= maximum + ): + raise ValueError("input must be an owned bounded single-link regular file") + self.original = identity(info) + self.data = os.pread(self.descriptor, maximum + 1, 0) + self.sha256 = hashlib.sha256(self.data).hexdigest() + if len(self.data) != info.st_size or (expected is not None and self.sha256 != expected): + raise ValueError("captured input identity differs") + self.recheck() + except (OSError, ValueError): + self.close() + raise ValueError("reviewed input refused: " + role) from None + + def recheck(self): + for parent, name, descriptor, original in self.parents: + if ( + directory_identity(os.fstat(descriptor)) != original + or directory_identity(os.stat(name, dir_fd=parent, follow_symlinks=False)) != original + ): + raise ValueError("input parent was replaced: " + self.role) + current = os.stat(self.path.name, dir_fd=self.parent, follow_symlinks=False) + if identity(current) != self.original or identity(os.fstat(self.descriptor)) != self.original: + raise ValueError("held input was replaced or changed: " + self.role) + if hashlib.sha256(os.pread(self.descriptor, self.original[6] + 1, 0)).hexdigest() != self.sha256: + raise ValueError("held input bytes changed: " + self.role) + + 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/scripts/conformance/command_keys/response_loss.py b/scripts/conformance/command_keys/response_loss.py new file mode 100644 index 0000000..d2447bd --- /dev/null +++ b/scripts/conformance/command_keys/response_loss.py @@ -0,0 +1,409 @@ +"""Installed SDK adapter for shared #216's separately reviewed loopback fixture. + +Run only with a reviewed proxy file/hash and synthetic runtime credentials. +No transport fault implementation or product source overlay lives here. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import importlib.metadata +import importlib.util +import io +import json +import platform +import stat +import threading +import time +from collections import Counter +from datetime import datetime, timezone +from pathlib import Path +from types import ModuleType +from uuid import uuid4 +from zipfile import ZipFile + +import httpx + +from durable_workflow import Client, Worker, workflow +from durable_workflow.errors import SignalFailed, UpdateFailed +from durable_workflow.retry_policy import TransportRetryPolicy + +WORKFLOW_TYPE = "sdk-python-command-response-loss" +SDK_VERSION = "2.17.2" +WHEEL_SHA256 = "7dc77d3882fa5cbddbeaee6a4cc63b05285e280523c8ea40dc31e03cc12a40bc" +PROXY_COMMIT = "117118c83093822469b3824ba9587ff4068ed29f" +PROXY_SHA256 = "bd1e8ccbbf76f63fc2eafa278a023eb181d405b68d9c30016392e395a1eff725" +PAYLOAD = { + "number": 1, + "flag": True, + "large": 2**62 + 7, + "unicode": "Ünïcødé ☃ 日本", + "nested": [None, False, "text", 1.25, b"\x00\xff"], +} +IDENTITIES = ("command_id", "command_sequence", "run_id", "resolved_run_id", "update_id", "accepted_at") + + +def explicit_key(kind): + prefix = kind + "-" + uuid4().hex + # This is the corrected producer's 255-code-point boundary, deliberately + # longer than 255 UTF-8 bytes. Caller keys are opaque Unicode strings. + return prefix + "日" * (255 - len(prefix)) + + +def installed_identity(wheel): + """Compare installed code to the exact published wheel, without importing project source.""" + wheel_bytes = wheel if isinstance(wheel, bytes) else wheel.read_bytes() + if hashlib.sha256(wheel_bytes).hexdigest() != WHEEL_SHA256: + raise ValueError("published wheel identity mismatch") + distribution = importlib.metadata.distribution("durable-workflow") + if distribution.version != SDK_VERSION or distribution.read_text("direct_url.json") is not None: + raise ValueError("require a registry-only installation of the bound SDK") + with ZipFile(io.BytesIO(wheel_bytes)) as archive: + names = sorted( + name for name in archive.namelist() if name.startswith("durable_workflow/") and not name.endswith("/") + ) + for name in names: + if distribution.locate_file(name).read_bytes() != archive.read(name): + raise ValueError("installed SDK file differs from published wheel") + if len(names) != 46: + raise ValueError("incomplete published SDK inventory") + # The imported package must also come from that installation, even when a + # caller accidentally runs without Python's documented isolated flag. + import durable_workflow + + if Path(durable_workflow.__file__).resolve() != distribution.locate_file("durable_workflow/__init__.py").resolve(): + raise ValueError("SDK import is outside the verified installation") + return { + "version": SDK_VERSION, + "wheel_sha256": WHEEL_SHA256, + "runtime_files_verified": len(names), + "python": platform.python_version(), + "dependencies": {name: importlib.metadata.version(name) for name in ("httpx", "fastavro")}, + } + + +@workflow.defn(name=WORKFLOW_TYPE) +class Commands: + def __init__(self): + self.signals = 0 + self.updates = 0 + self.done = False + self.rotate = False + self.generation = 0 + + @workflow.signal("add") + def add(self, value): + assert value == PAYLOAD and type(value["number"]) is int and type(value["flag"]) is bool + self.signals += 1 + + @workflow.signal("finish") + def finish(self): + self.done = True + + @workflow.signal("continue") + def continue_run(self): + self.rotate = True + + @workflow.update("value") + def value(self, value): + self.updates += 1 + return {"count": self.updates, "payload": value} + + @workflow.query("state") + def state(self): + return {"signals": self.signals, "updates": self.updates, "generation": self.generation} + + def run(self, ctx, generation=0): + self.generation = generation + yield ctx.wait_condition(lambda: self.done or self.rotate, key="control", timeout=600) + if self.rotate: + yield ctx.continue_as_new(generation + 1) + return self.state() + + +async def history(client, handle): + return [event async for event in client.iter_history(handle.workflow_id, handle.run_id)] + + +async def until(client, handle, event_type): + deadline = time.monotonic() + 60 + while time.monotonic() < deadline: + events = await history(client, handle) + if any(event["event_type"] == event_type for event in events): + return events + await asyncio.sleep(0.1) + raise TimeoutError(event_type) + + +async def state(client, workflow_id, signals, updates, generation=0): + deadline = time.monotonic() + 60 + expected = {"signals": signals, "updates": updates, "generation": generation} + while time.monotonic() < deadline: + receipt = await client.query_workflow(workflow_id, "state") + assert receipt["workflow_id"] == workflow_id and receipt["query_name"] == "state" + assert receipt["reason"] is None + if receipt["result"] == expected: + return + await asyncio.sleep(0.1) + raise AssertionError("business state mismatch") + + +async def command(client, kind, handle, key=None, payload=PAYLOAD): + if kind == "signal": + return await client.signal_workflow( + handle.workflow_id, "add", args=[payload], request_id=key, run_id=handle.run_id + ) + return await client.update_workflow( + handle.workflow_id, "value", args=[payload], request_id=key, wait_for="accepted" + ) + + +def route(kind, handle): + if kind == "signal": + return f"/api/workflows/{handle.workflow_id}/runs/{handle.run_id}/signal/add" + return f"/api/workflows/{handle.workflow_id}/update/value" + + +def retained(original, retry): + assert retry["request_id"] == original["request_id"] + assert retry["retained"] is True and retry["intent_verified"] is True + for key in IDENTITIES: + if key in original: + assert retry.get(key) == original[key], key + + +def sanitized_record(record): + return { + key: record[key] + for key in ( + "method", + "path", + "request_body_bytes", + "request_body_sha256", + "upstream_status", + "upstream_complete", + "upstream_body_bytes", + "upstream_body_sha256", + "dropped", + ) + } + + +def loss_records(proxy, path, count): + records = [record for record in proxy.records if record["method"] == "POST" and record["path"] == path] + assert len(records) == count, (path, len(records), count) + assert records[0]["dropped"] is True and 200 <= records[0]["upstream_status"] < 300 + assert records[0]["upstream_body_bytes"] > 0 + assert all(record["upstream_complete"] is True for record in records) + assert len(records[0]["upstream_body_sha256"]) == 64 + assert all(record["dropped"] is False and 200 <= record["upstream_status"] < 300 for record in records[1:]) + return [sanitized_record(record) for record in records] + + +def compare_requests(records, requests): + """Independent sender and proxy observations must agree on every request byte.""" + assert len(records) == len(requests) + assert len({item["sha256"] for item in requests}) == 1 + for record, request in zip(records, requests, strict=True): + assert record["request_body_sha256"] == request["sha256"] + assert record["request_body_bytes"] == request["bytes"] + + +async def main(args): + if not __debug__: + raise ValueError("acceptance assertions require Python without optimization") + if args.output.exists(): + raise ValueError("refusing to overwrite an existing evidence file") + proxy_bytes = args.proxy_data + if hashlib.sha256(proxy_bytes).hexdigest() != PROXY_SHA256: + raise ValueError("accepted proxy source hash mismatch") + package = installed_identity(args.wheel_data) + module = ModuleType("shared_committed_response_loss") + exec(compile(proxy_bytes, "captured-committed-response-loss.py", "exec"), module.__dict__) + 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"] + proxy = module.CommittedResponseLoss(args.upstream) + thread = threading.Thread(target=proxy.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True) + thread.start() + report = { + "sdk": package, + "proxy_commit": PROXY_COMMIT, + "proxy_sha256": PROXY_SHA256, + "adapter_sha256": args.adapter_sha256, + "started_at": datetime.now(timezone.utc).isoformat(), + "cells": [], + "complete": False, + } + wire = [] + + async def observe(request): + path = request.url.path + if request.method == "POST" and (path.endswith("/signal/add") or path.endswith("/update/value")): + body = request.content + wire.append( + { + "path": path, + "sha256": hashlib.sha256(body).hexdigest(), + "bytes": len(body), + "request_id": json.loads(body)["request_id"], + } + ) + + def save(): + args.output.write_text(json.dumps(report, indent=2) + "\n") + + queue = "sdk-python-loss-" + uuid4().hex + try: + async with Client(args.upstream, token=token) as direct: + info = await direct.get_cluster_info() + if info["version"] != args.server_version: + raise ValueError("runtime version differs from the externally verified tuple") + report["server_version"] = info["version"] + worker = Worker(direct, task_queue=queue, workflows=[Commands], poll_timeout=0.2) + running = asyncio.create_task(worker.run()) + try: + for automatic, explicit, continued, kinds in ( + (True, False, False, ("signal", "update")), + (True, True, True, ("signal", "update")), + (False, False, False, ("signal",)), + (False, True, False, ("signal",)), + (False, False, False, ("update",)), + (False, True, False, ("update",)), + ): + handle = await direct.start_workflow( + workflow_type=WORKFLOW_TYPE, workflow_id=queue + "-" + uuid4().hex, task_queue=queue, input=[0] + ) + await until(direct, handle, "ConditionWaitOpened") + originals = {} + cell = { + "automatic_retry": automatic, + "explicit_key": explicit, + "continued": continued, + "commands": {}, + } + policy = None if automatic else TransportRetryPolicy(max_attempts=1) + async with Client(proxy.url, token=token, retry_policy=policy) as lossy: + lossy._http.event_hooks["request"].append(observe) + for kind in kinds: + path = route(kind, handle) + key = explicit_key(kind) if explicit else None + proxy.arm("POST", path) + diagnostic = None + if automatic: + receipt = await command(lossy, kind, handle, key) + assert receipt["retained"] is True and receipt["intent_verified"] is True + else: + try: + await command(lossy, kind, handle, key) + except httpx.TransportError as error: + diagnostic = type(error).__name__ + sent_key = getattr(error, "request_id", None) + assert isinstance(sent_key, str) and sent_key + assert sent_key == key if explicit else sent_key.startswith(f"sdk-python-{kind}-") + else: + raise AssertionError("selected committed response was not lost") + loss_records(proxy, path, 1) + receipt = await command(direct, kind, handle, sent_key) + assert receipt["retained"] is True and receipt["intent_verified"] is True + expected = 2 if automatic else 1 + records = loss_records(proxy, path, expected) + requests = [item for item in wire if item["path"] == path] + assert len(requests) == expected + compare_requests(records, requests) + assert all(item["request_id"] == receipt["request_id"] for item in requests) + assert ( + receipt["request_id"] == key + if explicit + else receipt["request_id"].startswith(f"sdk-python-{kind}-") + ) + assert receipt["command_id"] and type(receipt["command_sequence"]) is int + if kind == "update": + result = await direct.update_outcome( + handle.workflow_id, receipt["update_id"], wait_timeout_seconds=30 + ) + assert result["count"] == 1 and result["payload"] == PAYLOAD + assert ( + type(result["payload"]["number"]) is int and type(result["payload"]["flag"]) is bool + ) + originals[kind] = receipt + assert receipt.get("resolved_run_id", receipt.get("run_id")) == handle.run_id + # Concurrent retries must all resolve to one retained + # command before any close/continuation control. + retries = await asyncio.gather( + *(command(direct, kind, handle, receipt["request_id"]) for _ in range(4)) + ) + for retry in retries: + retained(receipt, retry) + try: + await command(direct, kind, handle, receipt["request_id"], {"different": True}) + except (SignalFailed, UpdateFailed) as error: + assert error.reason == "request_id_conflict" + assert error.request_id == receipt["request_id"] + else: + raise AssertionError("changed command intent was admitted") + cell["commands"][kind] = { + "key_characters": len(receipt["request_id"]), + "key_utf8_bytes": len(receipt["request_id"].encode()), + "wire": requests, + "loss": records, + "transport_exception": diagnostic, + "concurrent_identical_retries": len(retries), + "changed_intent": "request_id_conflict", + "identity": {key: receipt[key] for key in IDENTITIES if key in receipt}, + } + await state(direct, handle.workflow_id, int("signal" in kinds), int("update" in kinds)) + if continued: + await direct.signal_workflow(handle.workflow_id, "continue", run_id=handle.run_id) + await until(direct, handle, "WorkflowContinuedAsNew") + current = await direct.describe_workflow(handle.workflow_id) + assert current.run_id != handle.run_id + await state(direct, handle.workflow_id, 0, 0, 1) + for kind, original in originals.items(): + retained(original, await command(direct, kind, handle, original["request_id"])) + await state(direct, handle.workflow_id, 0, 0, 1) + # The original selected-run identity remains authoritative for recovery. + before = len(await history(direct, handle)) + await direct.signal_workflow(current.workflow_id, "finish", run_id=current.run_id) + await until(direct, current, "WorkflowCompleted") + else: + await direct.signal_workflow(handle.workflow_id, "finish", run_id=handle.run_id) + await until(direct, handle, "WorkflowCompleted") + before = len(await history(direct, handle)) + for kind, original in originals.items(): + retained(original, await command(direct, kind, handle, original["request_id"])) + events = await history(direct, handle) + assert len(events) == before, "retained closed-run reads wrote history" + counts = Counter(event["event_type"] for event in events) + assert counts["SignalReceived"] == int("signal" in kinds) + 1 + for event_type in ("UpdateAccepted", "UpdateApplied", "UpdateCompleted"): + assert counts[event_type] == int("update" in kinds), (event_type, counts) + cell["event_counts"] = dict(counts) + cell["business_executions"] = {"signal": int("signal" in kinds), "update": int("update" in kinds)} + report["cells"].append(cell) + save() + print(f"completed response-loss cell {len(report['cells'])}", flush=True) + finally: + await worker.stop() + await running + report["business_complete"] = True + except Exception as error: + report["failure_type"] = type(error).__name__ + raise + finally: + proxy.shutdown() + proxy.server_close() + thread.join(timeout=5) + report["proxy_thread_stopped"] = not thread.is_alive() + save() + assert not thread.is_alive() + report["complete"] = True + report["finished_at"] = datetime.now(timezone.utc).isoformat() + save() + return report + + +if __name__ == "__main__": + raise SystemExit("Use the bound run_response_loss.py controller; standalone executable paths are refused") diff --git a/scripts/conformance/command_keys/run_response_loss.py b/scripts/conformance/command_keys/run_response_loss.py new file mode 100644 index 0000000..da57ce6 --- /dev/null +++ b/scripts/conformance/command_keys/run_response_loss.py @@ -0,0 +1,422 @@ +"""Run the published SDK in a bound interpreter with captured executable inputs.""" + +from __future__ import annotations + +import argparse +import base64 +import hashlib +import json +import os +import re +import secrets +import subprocess +import time +import urllib.request +from contextlib import ExitStack +from datetime import datetime, timezone +from pathlib import Path +from types import ModuleType +from uuid import uuid4 + +EXPECTED_SERVER = { + "version": "3.9.2", + "source": "5614274d2629d03f92932a07dfee0fb10349ddd9", + "index": "sha256:8ffcdb2a36af3129a1987849aafe8715fe0d3bf9a9648467f0b5cb8eeef43ba9", + "linux_amd64": "sha256:813c9337c44be7d497efd29e1b463546519e488aa796b5ecf9dfc6f4f10d2128", + "qualification": "https://github.com/durable-workflow/server/pull/1209#issuecomment-6858140177", +} +PYTHON_INDEX = "sha256:935c3c5b12357a7e10caa9c1e6ca5692344d97cf229832ef584be2cec9fc03df" +PYTHON_AMD64 = "sha256:0f086103b026048cc50cb5da25ec435c023c5241a83fc44daea593377abc61c4" +PYTHON_VERSION = "3.12.16" +WHEEL_SHA256 = "7dc77d3882fa5cbddbeaee6a4cc63b05285e280523c8ea40dc31e03cc12a40bc" +PROXY_SHA256 = "bd1e8ccbbf76f63fc2eafa278a023eb181d405b68d9c30016392e395a1eff725" +RESULT_PREFIX = "SDK_PYTHON_CONSUMER_RESULT=" +OWNER_LABEL = "dev.durable-workflow.fixture-owner" +RUN_LABEL = "dev.durable-workflow.fixture-run" + + +def image_identity(document): + image = document[0] + if ( + image["Architecture"] != "amd64" + or image["Os"] != "linux" + or image["Config"]["Labels"]["org.opencontainers.image.revision"] != EXPECTED_SERVER["source"] + or not any(value.endswith("@" + EXPECTED_SERVER["index"]) for value in image["RepoDigests"]) + ): + raise ValueError("local runtime image differs from the qualified source/index/platform") + return { + "id": image["Id"], + "architecture": image["Architecture"], + "os": image["Os"], + "revision": image["Config"]["Labels"]["org.opencontainers.image.revision"], + "repo_digests": image["RepoDigests"], + } + + +def interpreter_identity(document): + image = document[0] + versions = [value for value in image["Config"]["Env"] if value.startswith("PYTHON_VERSION=")] + if ( + image["Architecture"] != "amd64" + or image["Os"] != "linux" + or not any(value.endswith("@" + PYTHON_INDEX) for value in image["RepoDigests"]) + or versions != ["PYTHON_VERSION=" + PYTHON_VERSION] + ): + raise ValueError("require the fixed published Python interpreter image") + return { + "id": image["Id"], + "index": PYTHON_INDEX, + "expected_amd64": PYTHON_AMD64, + "version": PYTHON_VERSION, + "architecture": "amd64", + } + + +def missing_object(result, name, *, volume=False): + expected = f"Error response from daemon: get {name}: no such volume" if volume else f"error: no such object: {name}" + return result.returncode == 1 and result.stderr.strip() == expected + + +def created_id(stdout): + value = stdout.strip() + if re.fullmatch(r"[0-9a-f]{64}", value) is None: + raise ValueError("Docker create did not return one full immutable container ID") + return value + + +def inspect_owned(call, container_id, nonce): + created_id(container_id) + response = call("inspect", container_id, check=False) + if missing_object(response, container_id): + return None + if response.returncode != 0: + raise RuntimeError("owned container state is unverified") + records = json.loads(response.stdout) + if len(records) != 1: + raise ValueError("owned container inspection is ambiguous") + record = records[0] + labels = record.get("Config", {}).get("Labels") or {} + if record["Id"] != container_id or labels.get(OWNER_LABEL) != "sdk-python" or labels.get(RUN_LABEL) != nonce: + raise ValueError("container ID/ownership identity differs; preserve it") + return record + + +def cleanup_owned(call, container_id, nonce): + record = inspect_owned(call, container_id, nonce) + volumes = [] + oom = None + if record is not None: + volumes = [mount["Name"] for mount in record["Mounts"] if mount["Type"] == "volume"] + oom = record["State"]["OOMKilled"] + call("rm", "-f", "-v", container_id) + # Never inspect, remove or recover by a reusable name. If the owned ID + # vanished, any same-name replacement and its volumes remain untouched. + absent = inspect_owned(call, container_id, nonce) is None + volumes_absent = all( + missing_object(call("volume", "inspect", volume, check=False), volume, volume=True) for volume in volumes + ) + return {"id": container_id, "absent": absent, "volumes_absent": volumes_absent, "oom_killed": oom} + + +def git_bytes(root, commit, relative): + return subprocess.check_output( + ["git", "-C", str(root), "show", commit + ":" + relative], env={"PATH": os.defpath}, timeout=10 + ) + + +def captured_inputs(args, stack): + 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] + head = subprocess.check_output( + ["git", "-C", str(root), "rev-parse", "HEAD"], text=True, env={"PATH": os.defpath}, timeout=10 + ).strip() + if head != args.runner_commit: + raise ValueError("checkout differs from the declared reviewed runner commit") + prefix = "scripts/conformance/command_keys/" + guard_bytes = git_bytes(root, head, prefix + "input_guards.py") + guards = ModuleType("reviewed_input_guards") + # 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 { + "controller": "run_response_loss.py", + "guard": "input_guards.py", + "adapter": "response_loss.py", + "bootstrap": "consumer_bootstrap.py", + "requirements": "consumer-requirements.txt", + }.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) + ) + return inputs + + +def consumer_result(stdout, inputs): + 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") + result = json.loads(lines[0]) + 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") + 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 call(*arguments, check=True, timeout=45): + return subprocess.run( + docker + list(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 + observed = image_identity(json.loads(call("image", "inspect", image).stdout)) + interpreter = interpreter_identity(json.loads(call("image", "inspect", python_image).stdout)) + nonce = uuid4().hex + 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") + report = { + "started_at": datetime.now(timezone.utc).isoformat(), + "runner_commit": args.runner_commit, + "expected_server": EXPECTED_SERVER, + "observed_image": observed, + "interpreter_image": interpreter, + "inputs": {role: {"sha256": value.sha256, "bytes": len(value.data)} for role, value in inputs.items()}, + "owned_containers": {}, + "ownership_nonce": nonce, + "complete": False, + } + try: + for value in inputs.values(): + value.recheck() + runtime_id = created_id( + call( + "create", + "--init", + "--read-only", + "--name", + names[0], + "--label", + OWNER_LABEL + "=sdk-python", + "--label", + RUN_LABEL + "=" + nonce, + "--cpus", + "1", + "--memory", + "1g", + "--pids-limit", + "256", + "--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=64m", + "-p", + "127.0.0.1::8080", + "--env-file", + str(credential), + "-e", + "DB_DATABASE=/app/database/response-loss.sqlite", + image, + ).stdout + ) + report["owned_containers"]["runtime"] = {"id": runtime_id, "requested_name": names[0]} + inspect_owned(call, runtime_id, nonce) + 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("synthetic runtime readiness") + bundle = { + "sources": { + role: {"sha256": value.sha256, "base64": base64.b64encode(value.data).decode()} + for role, value in inputs.items() + if role in {"adapter", "bootstrap", "requirements", "proxy", "wheel"} + }, + "upstream": upstream, + "token": token, + "server_version": EXPECTED_SERVER["version"], + } + for value in inputs.values(): + value.recheck() + # 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 + ) + 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]} + inspect_owned(call, consumer_id, nonce) + result = subprocess.run( + docker + ["start", "-a", "-i", consumer_id], + input=json.dumps(bundle), + capture_output=True, + text=True, + env=environment, + timeout=600, + ) + (root / "consumer.log").write_text(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") + report["consumer_exit_code"] = consumer_state["State"]["ExitCode"] + report["docker_attach_exit_code"] = result.returncode + for value in inputs.values(): + value.recheck() + report["input_identities_rechecked"] = True + if interpreter_identity(json.loads(call("image", "inspect", python_image).stdout)) != interpreter: + raise ValueError("interpreter image identity changed") + report["interpreter_identity_rechecked"] = True + 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") + report["consumer_complete"] = True + except Exception as error: + report["failure_type"] = type(error).__name__ + raise + finally: + clean = [] + oom = [] + try: + report["cleanup_by_id"] = {} + for role, owned in report["owned_containers"].items(): + try: + outcome = cleanup_owned(call, owned["id"], nonce) + except Exception as error: + report["cleanup_by_id"][role] = {"id": owned["id"], "failure_type": type(error).__name__} + clean.append(False) + oom.append(None) + continue + report["cleanup_by_id"][role] = outcome + oom.append(outcome["oom_killed"]) + clean.append(outcome["absent"] and outcome["volumes_absent"]) + except Exception as error: + report["cleanup_failure_type"] = type(error).__name__ + finally: + credential.unlink(missing_ok=True) + report["cleanup"] = { + "containers_and_volumes_absent": len(clean) == 2 and all(clean), + "credential_removed": not credential.exists(), + "oom_killed": ( + True + if any(value is True for value in oom) + else False + if len(oom) == 2 and all(value is False for value in oom) + else None + ), + } + report["finished_at"] = datetime.now(timezone.utc).isoformat() + report["complete"] = ( + report.get("consumer_complete") is True + and len(clean) == 2 + and all(clean) + and len(oom) == 2 + and all(value is False for value in oom) + and "failure_type" not in report + and "cleanup_failure_type" not in report + ) + (root / "run.json").write_text(json.dumps(report, indent=2) + "\n") + if not report["cleanup"]["containers_and_volumes_absent"]: + raise RuntimeError("owned runtime cleanup incomplete") + if not report["complete"] and "failure_type" not in report: + raise RuntimeError("installed cohort evidence is incomplete") + + +def run(args): + with ExitStack() as stack: + inputs = captured_inputs(args, stack) + run_bound(args, inputs) + + +def parse_args(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--runner-commit", required=True) + parser.add_argument("--wheel", type=Path, required=True) + parser.add_argument("--proxy", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + return parser.parse_args() + + +if __name__ == "__main__": + run(parse_args()) diff --git a/tests/test_command_keys_acceptance.py b/tests/test_command_keys_acceptance.py new file mode 100644 index 0000000..3ccf1f8 --- /dev/null +++ b/tests/test_command_keys_acceptance.py @@ -0,0 +1,167 @@ +"""Keep incomplete transport evidence from becoming an installed acceptance PASS.""" + +from __future__ import annotations + +import hashlib +import importlib.util +from pathlib import Path +from types import SimpleNamespace +from zipfile import ZipFile + +import pytest + +SOURCE = Path(__file__).parents[1] / "scripts/conformance/command_keys/response_loss.py" +SPEC = importlib.util.spec_from_file_location("command_keys_acceptance", SOURCE) +assert SPEC is not None and SPEC.loader is not None +adapter = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(adapter) +CONTROLLER_SPEC = importlib.util.spec_from_file_location( + "command_keys_controller", SOURCE.with_name("run_response_loss.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) + + +@pytest.mark.parametrize("kind", ["signal", "update"]) +def test_explicit_key_exercises_unicode_character_not_byte_limit(kind): + key = adapter.explicit_key(kind) + assert len(key) == 255 + assert len(key.encode()) > 255 + assert key != adapter.explicit_key(kind) + + +def record(**changes): + return { + "method": "POST", + "path": "/api/workflows/synthetic/update/value", + "request_body_bytes": 100, + "request_body_sha256": "a" * 64, + "upstream_status": 200, + "upstream_complete": True, + "upstream_body_bytes": 50, + "upstream_body_sha256": "b" * 64, + "dropped": True, + **changes, + } + + +def test_complete_loss_and_retry_agree_with_independent_sender(): + records = [record(), record(dropped=False)] + observed = adapter.loss_records(SimpleNamespace(records=records), records[0]["path"], 2) + adapter.compare_requests(observed, [{"sha256": "a" * 64, "bytes": 100}] * 2) + + +@pytest.mark.parametrize( + "changes", + [ + {"dropped": False}, + {"upstream_complete": False}, + {"upstream_status": 502}, + {"upstream_body_bytes": 0}, + {"upstream_body_sha256": ""}, + ], +) +def test_incomplete_or_unplanned_loss_refuses(changes): + item = record(**changes) + with pytest.raises(AssertionError): + adapter.loss_records(SimpleNamespace(records=[item]), item["path"], 1) + + +@pytest.mark.parametrize( + "requests", + [ + [], + [{"sha256": "a" * 64, "bytes": 99}], + [{"sha256": "c" * 64, "bytes": 100}], + [{"sha256": "a" * 64, "bytes": 100}, {"sha256": "c" * 64, "bytes": 100}], + ], +) +def test_truncated_or_changed_request_comparison_refuses(requests): + with pytest.raises(AssertionError): + adapter.compare_requests([record()], requests) + + +@pytest.mark.parametrize( + "change", + [ + {"request_id": "different"}, + {"command_id": "different"}, + {"retained": False}, + {"intent_verified": False}, + {"run_id": "different"}, + ], +) +def test_retained_identity_and_intent_are_required(change): + original = {"request_id": "synthetic-key", "command_id": "command-1", "run_id": "run-1"} + retry = {**original, "retained": True, "intent_verified": True, **change} + with pytest.raises(AssertionError): + adapter.retained(original, retry) + + +def test_installed_file_substitution_and_source_overlay_refuse(tmp_path, monkeypatch): + wheel = tmp_path / "synthetic.whl" + root = tmp_path / "installed" + names = ["durable_workflow/__init__.py"] + [f"durable_workflow/file{i}.py" for i in range(45)] + with ZipFile(wheel, "w") as archive: + for name in names: + archive.writestr(name, b"# synthetic fixture\n") + target = root / name + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(b"# synthetic fixture\n") + monkeypatch.setattr(adapter, "WHEEL_SHA256", hashlib.sha256(wheel.read_bytes()).hexdigest()) + distribution = SimpleNamespace( + version=adapter.SDK_VERSION, read_text=lambda _: None, locate_file=lambda name: root / name + ) + monkeypatch.setattr(adapter.importlib.metadata, "distribution", lambda _: distribution) + import durable_workflow + + monkeypatch.setattr(durable_workflow, "__file__", str(root / names[0])) + assert adapter.installed_identity(wheel)["runtime_files_verified"] == 46 + (root / names[-1]).write_bytes(b"# substituted\n") + with pytest.raises(ValueError, match="differs"): + adapter.installed_identity(wheel) + (root / names[-1]).write_bytes(b"# synthetic fixture\n") + monkeypatch.setattr(durable_workflow, "__file__", str(tmp_path / "overlay/__init__.py")) + with pytest.raises(ValueError, match="outside"): + adapter.installed_identity(wheel) + + +@pytest.mark.parametrize( + "volume,message", + [ + (False, "error: no such object: synthetic-owned"), + (True, "Error response from daemon: get synthetic-owned: no such volume"), + ], +) +def test_cleanup_recognizes_only_missing_owned_objects(volume, message): + result = SimpleNamespace(returncode=1, stderr=message + "\n") + assert controller.missing_object(result, "synthetic-owned", volume=volume) + assert not controller.missing_object(result, "someone-elses-object", volume=volume) + result.stderr = "Cannot connect to the Docker daemon" + assert not controller.missing_object(result, "synthetic-owned", volume=volume) + result.stderr = message + result.returncode = 0 + assert not controller.missing_object(result, "synthetic-owned", volume=volume) + + +@pytest.mark.parametrize( + "field,value", + [ + ("Architecture", "arm64"), + ("Os", "windows"), + ("RepoDigests", []), + ("Config", {"Labels": {"org.opencontainers.image.revision": "unqualified-source"}}), + ], +) +def test_unqualified_local_image_refuses(field, value): + image = { + "Architecture": "amd64", + "Os": "linux", + "Id": "synthetic-config-id", + "RepoDigests": ["ghcr.io/durable-workflow/server@" + controller.EXPECTED_SERVER["index"]], + "Config": {"Labels": {"org.opencontainers.image.revision": controller.EXPECTED_SERVER["source"]}}, + } + assert controller.image_identity([image])["revision"] == controller.EXPECTED_SERVER["source"] + with pytest.raises(ValueError, match="differs"): + controller.image_identity([{**image, field: value}]) diff --git a/tests/test_command_keys_captured_inputs.py b/tests/test_command_keys_captured_inputs.py new file mode 100644 index 0000000..56cb567 --- /dev/null +++ b/tests/test_command_keys_captured_inputs.py @@ -0,0 +1,181 @@ +"""Substituted files cannot become executable authority or certify a result.""" + +from __future__ import annotations + +import base64 +import hashlib +import importlib.util +import json +import runpy +from pathlib import Path + +import pytest + +ROOT = Path(__file__).parents[1] / "scripts/conformance/command_keys" + + +def module(name, path): + spec = importlib.util.spec_from_file_location(name, path) + assert spec is not None and spec.loader is not None + result = importlib.util.module_from_spec(spec) + spec.loader.exec_module(result) + return result + + +guards = module("capture_controls", ROOT / "input_guards.py") +bootstrap = module("bootstrap_controls", ROOT / "consumer_bootstrap.py") +controller = module("controller_controls", ROOT / "run_response_loss.py") +ROLES = ["adapter", "proxy", "wheel", "bootstrap", "requirements"] + + +@pytest.mark.parametrize("role", ROLES) +@pytest.mark.parametrize("substitution", ["symlink", "rename-identical", "rewrite", "hardlink"]) +def test_all_input_roles_reject_symlinks_hardlinks_and_path_or_inode_changes(tmp_path, role, substitution): + path = tmp_path / "input" + path.write_bytes(b"original input") + digest = hashlib.sha256(path.read_bytes()).hexdigest() + if substitution == "symlink": + alias = tmp_path / "alias" + alias.symlink_to(path) + with pytest.raises(ValueError): + guards.CapturedInput(alias, role, maximum=1024, expected=digest) + elif substitution == "hardlink": + import os + + os.link(path, tmp_path / "second-link") + with pytest.raises(ValueError): + guards.CapturedInput(path, role, maximum=1024, expected=digest) + else: + with guards.CapturedInput(path, role, maximum=1024, expected=digest) as captured: + if substitution == "rename-identical": + path.rename(tmp_path / "previous") + path.write_bytes(b"original input") + else: + path.write_bytes(b"substituted input") + assert captured.data == b"original input" + with pytest.raises(ValueError): + captured.recheck() + + +def test_parent_directory_rename_refuses_even_identical_replacement(tmp_path): + parent = tmp_path / "owned" + parent.mkdir(mode=0o700) + path = parent / "input" + path.write_bytes(b"original") + with guards.CapturedInput(path, "adapter", maximum=100) as captured: + parent.rename(tmp_path / "renamed") + parent.mkdir(mode=0o700) + path.write_bytes(b"original") + with pytest.raises(ValueError, match="parent was replaced"): + captured.recheck() + + +def test_substituted_adapter_cannot_self_certify_its_report(tmp_path): + path = tmp_path / "adapter.py" + path.write_text("def attest(): return {'complete': False}\n") + with guards.CapturedInput(path, "adapter", maximum=4096) as captured: + forged = {"complete": True, "adapter_sha256": captured.sha256} + path.rename(tmp_path / "original.py") + path.write_text("def attest(): return " + repr(forged) + "\n") + # The predecessor's hash-then-path-load and self-reported comparison + # accepts this synthetic forgery. The corrected execution sees only + # the captured original and its post-call identity check refuses. + predecessor = runpy.run_path(str(path))["attest"]() + assert predecessor["complete"] and predecessor["adapter_sha256"] == captured.sha256 + executed = bootstrap.execute_source(captured.data, "captured_adapter_control") + assert executed.attest()["complete"] is False + with pytest.raises(ValueError): + captured.recheck() + + +@pytest.mark.parametrize("role", ["adapter", "proxy", "wheel", "bootstrap", "requirements"]) +def test_bundle_byte_substitution_refuses_before_execution(role): + sources = { + key: {"base64": base64.b64encode(b"original").decode(), "sha256": hashlib.sha256(b"original").hexdigest()} + for key in bootstrap.ROLES + } + sources[role]["base64"] = base64.b64encode(b"substituted").decode() + with pytest.raises(ValueError, match="bytes differ"): + bootstrap.decode_inputs({"sources": sources}) + + +def test_no_host_interpreter_path_can_be_supplied(monkeypatch, capsys): + monkeypatch.setattr( + "sys.argv", + [ + "run_response_loss.py", + "--runner-commit", + "a" * 40, + "--wheel", + "wheel", + "--proxy", + "proxy", + "--output", + "new-output", + "--python", + "/arbitrary/program", + ], + ) + with pytest.raises(SystemExit): + controller.parse_args() + assert "unrecognized arguments: --python /arbitrary/program" in capsys.readouterr().err + + +@pytest.mark.parametrize( + "changes", + [ + {"Architecture": "arm64"}, + {"RepoDigests": ["python@sha256:" + "0" * 64]}, + {"Config": {"Env": ["PYTHON_VERSION=unqualified"]}}, + ], +) +def test_unqualified_interpreter_image_refuses(changes): + image = { + "Architecture": "amd64", + "Os": "linux", + "Id": "qualified-image", + "RepoDigests": ["python@" + controller.PYTHON_INDEX], + "Config": {"Env": ["PYTHON_VERSION=" + controller.PYTHON_VERSION]}, + } + assert controller.interpreter_identity([image])["version"] == controller.PYTHON_VERSION + with pytest.raises(ValueError, match="fixed published Python"): + controller.interpreter_identity([{**image, **changes}]) + + +def test_inventory_catches_installed_dependency_or_interpreter_mutation(tmp_path): + (tmp_path / "bin").mkdir() + (tmp_path / "bin/python").write_bytes(b"trusted interpreter fixture") + before = bootstrap.inventory(tmp_path) + (tmp_path / "bin/python").write_bytes(b"substituted interpreter fixture") + assert bootstrap.inventory(tmp_path) != before + + +@pytest.mark.parametrize( + "relative,target", + [ + ("bin/python", "python3.12"), + ("bin/python3", "python3.12"), + ("bin/python3.12", "/usr/local/bin/python3.12"), + ("lib64", "lib"), + ], +) +def test_fresh_venv_links_are_closed_to_the_immutable_image(relative, target): + assert bootstrap.approved_environment_link(relative, target) + assert not bootstrap.approved_environment_link(relative, "/host/arbitrary-program") + assert not bootstrap.approved_environment_link("lib/python3.12/site-packages/replaced", target) + + +def test_requirement_lock_covers_only_exact_wheel_hashes(): + lines = (ROOT / "consumer-requirements.txt").read_text().splitlines() + assert len(lines) == 9 + assert all("==" in line and " --hash=sha256:" in line for line in lines) + assert controller.WHEEL_SHA256 in "\n".join(lines) + # This JSON framing round trip retains bytes rather than an executable path. + original = b"compiled from captured bytes\n" + bundle = { + "sources": { + role: {"base64": base64.b64encode(original).decode(), "sha256": hashlib.sha256(original).hexdigest()} + for role in bootstrap.ROLES + } + } + assert bootstrap.decode_inputs(json.loads(json.dumps(bundle)))["adapter"] == original diff --git a/tests/test_command_keys_container_ownership.py b/tests/test_command_keys_container_ownership.py new file mode 100644 index 0000000..4ab1cb7 --- /dev/null +++ b/tests/test_command_keys_container_ownership.py @@ -0,0 +1,108 @@ +"""A reused human name grants no authority over another fixture's resources.""" + +from __future__ import annotations + +import copy +import importlib.util +import json +from pathlib import Path +from types import SimpleNamespace + +import pytest + +PATH = Path(__file__).parents[1] / "scripts/conformance/command_keys/run_response_loss.py" +SPEC = importlib.util.spec_from_file_location("id_bound_runner", PATH) +assert SPEC is not None and SPEC.loader is not None +controller = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(controller) +OWN_ID, FOREIGN_ID = "a" * 64, "b" * 64 +NONCE = "owned-run" + + +def document(identity, name, owner="sdk-python", nonce=NONCE): + return { + "Id": identity, + "Name": "/" + name, + "Config": {"Labels": {controller.OWNER_LABEL: owner, controller.RUN_LABEL: nonce}}, + "State": {"OOMKilled": False}, + "Mounts": [{"Type": "volume", "Name": identity + "-data"}], + } + + +class DockerModel: + def __init__(self, own_present=False): + self.objects = {FOREIGN_ID: document(FOREIGN_ID, "reused-name", owner="foreign", nonce="foreign-run")} + if own_present: + self.objects[OWN_ID] = document(OWN_ID, "original-renamed") + self.names = {value["Name"].removeprefix("/"): key for key, value in self.objects.items()} + self.volumes = {value["Mounts"][0]["Name"] for value in self.objects.values()} + self.calls = [] + + def __call__(self, *args, **_options): + self.calls.append(args) + if args[0] == "inspect": + identity = self.names.get(args[1], args[1]) + if identity in self.objects: + return SimpleNamespace(returncode=0, stdout=json.dumps([self.objects[identity]]), stderr="") + return SimpleNamespace(returncode=1, stdout="[]", stderr="error: no such object: " + args[1]) + if args[0] == "rm": + identity = self.names.get(args[-1], args[-1]) + value = self.objects.pop(identity) + self.volumes.remove(value["Mounts"][0]["Name"]) + self.names.pop(value["Name"].removeprefix("/")) + return SimpleNamespace(returncode=0, stdout=identity, stderr="") + if args[:2] == ("volume", "inspect"): + if args[2] in self.volumes: + return SimpleNamespace(returncode=0, stdout="[]", stderr="") + return SimpleNamespace( + returncode=1, stdout="[]", stderr="Error response from daemon: get " + args[2] + ": no such volume" + ) + raise AssertionError("unexpected Docker operation") + + +@pytest.mark.parametrize("own_present", [False, True]) +def test_foreign_same_name_and_its_volume_survive_id_bound_cleanup(own_present): + docker = DockerModel(own_present) + foreign = copy.deepcopy(docker.objects[FOREIGN_ID]) + result = controller.cleanup_owned(docker, OWN_ID, NONCE) + assert result["absent"] and result["volumes_absent"] + assert docker.objects[FOREIGN_ID] == foreign + assert FOREIGN_ID + "-data" in docker.volumes + assert all(call[-1] != "reused-name" for call in docker.calls) + assert [call for call in docker.calls if call[0] == "rm"] == ([("rm", "-f", "-v", OWN_ID)] if own_present else []) + + +def test_predecessor_name_cleanup_deletes_the_foreign_replacement(): + predecessor = DockerModel() + # The former controller resolved and removed by reusable name. The fake + # engine demonstrates that exact boundary without touching a real resource. + state = json.loads(predecessor("inspect", "reused-name").stdout)[0] + assert state["Id"] == FOREIGN_ID + predecessor("rm", "-f", "-v", "reused-name") + assert FOREIGN_ID not in predecessor.objects and FOREIGN_ID + "-data" not in predecessor.volumes + corrected = DockerModel() + controller.cleanup_owned(corrected, OWN_ID, NONCE) + assert FOREIGN_ID in corrected.objects and FOREIGN_ID + "-data" in corrected.volumes + + +@pytest.mark.parametrize( + "mutation", + [ + lambda value: value.update(Id=FOREIGN_ID), + lambda value: value["Config"]["Labels"].update({controller.OWNER_LABEL: "foreign"}), + lambda value: value["Config"]["Labels"].update({controller.RUN_LABEL: "different-run"}), + ], +) +def test_mismatched_id_or_ownership_refuses_without_mutating_anything(mutation): + docker = DockerModel(own_present=True) + mutation(docker.objects[OWN_ID]) + with pytest.raises(ValueError, match="ownership identity differs"): + controller.cleanup_owned(docker, OWN_ID, NONCE) + assert not any(call[0] == "rm" for call in docker.calls) + + +@pytest.mark.parametrize("value", ["reused-name", "a" * 12, "A" * 64, "a" * 64 + "\nforeign", ""]) +def test_only_a_full_created_id_can_authorize_operations(value): + with pytest.raises(ValueError, match="full immutable container ID"): + controller.created_id(value) + assert controller.created_id(OWN_ID + "\n") == OWN_ID