diff --git a/.github/workflows/qualification.yml b/.github/workflows/qualification.yml index cb0482e4c3d..61d0c63402a 100644 --- a/.github/workflows/qualification.yml +++ b/.github/workflows/qualification.yml @@ -63,6 +63,14 @@ jobs: with: fetch-depth: 0 persist-credentials: false + - name: Set up Python for inert preparation controls + uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6 + with: + python-version: "3.13" + - name: Run inert preparation controller controls + run: >- + python -I -S -B -m unittest discover + -s security/browser-headers -p test_rust_preparation.py -v - name: Set up Node uses: actions/setup-node@249970729cb0ef3589644e2896645e5dc5ba9c38 # v6 with: diff --git a/security/browser-headers/README.md b/security/browser-headers/README.md index 4b5189d345d..0e8011ab81e 100644 --- a/security/browser-headers/README.md +++ b/security/browser-headers/README.md @@ -119,6 +119,8 @@ only its inert source controls. The optional browser entrypoint requires its own exact run admission; it makes no provider or production request. `rust-preparation-proposal.md` supplies the next separately reviewed acquisition and immutable-image custody design. No acquisition, seal or prepared image exists. +The separately gated [controller source](rust-preparation-controller.md) has +inert fault controls; its operational admission remains unset. ## Rust-only staged acceptance and rollback diff --git a/security/browser-headers/rust-preparation-controller.md b/security/browser-headers/rust-preparation-controller.md new file mode 100644 index 00000000000..d480227963a --- /dev/null +++ b/security/browser-headers/rust-preparation-controller.md @@ -0,0 +1,225 @@ +# Bounded preparation controller source + +This implements the source-only commission after +[Docs #479](https://github.com/durable-workflow/durable-workflow.github.io/pull/479#issuecomment-6649284339). +The [reviewed design](rust-preparation-proposal.md) remains authoritative. +`rust_preparation.py` contains input authentication and allocation/cleanup +rules. `rust_preparation_platform.py` contains the guarded Linux adapter. +`rust_preparation_authority.py` verifies a separately reviewed installed-tool, +runtime, manager and daemon catalog; no real catalog is supplied or admitted. +`test_rust_preparation.py` exercises them with synthetic files/archives, +in-memory seal operations, a fake daemon and a fake service manager. + +**Operational and installed-authority admissions remain `None`.** CLI and native entrypoints refuse +before any acquisition, memory seal, service, Docker/image, Node/browser or +provider effect. There is no command-line or environment bypass. The source +does not accept a real handoff, prepare an image or qualify the host runtime. +No controller invocation or filesystem path is approved by these source files. +The current suite has 47 test methods, including the per-edge fault subcases. +Security's source/executable HOLD remains in force pending review of these +corrections: durable mutation binding, descriptor-first transport creation, +bounded evidence retention, installed authority and the joined terminal collector. +The [selected retention contract](https://github.com/durable-workflow/durable-workflow.github.io/pull/480#issuecomment-6656594364) +supersedes the earlier proposal's transport deletion requirement. No transport +pathname deletion is attempted by this controller. + +Run the source controls with Python 3.11 or later on Linux: + +```sh +python3 -I -S -B -m unittest discover -s security/browser-headers -p test_rust_preparation.py -v +``` + +The controls load only the Python controller and standard library. They read +the existing browser packet as data, never import a Node module, and never call +the real process, network, kernel-seal or Docker/service adapters. Their unsafe +baselines are explicitly labelled test countermodels, not historical browser +or preparation runs. Fake lifecycle success is not native lifecycle acceptance. + +## Before any later operation + +Security must review the exact source and fault controls. General separately +binds the host/principal, complete trusted interpreter/runtime and existing +user-manager capabilities; exact authenticated predownloaded ZIP metadata; +fresh protected-run/deployment/live receipt; nine packet-file bindings; staging +and Docker-filesystem locations; competing work and capacity; and one invocation. +Those private fields are deliberately absent here. Missing/expired/unbound +fields stop the attempt. The current artifact still expires November 6 at +03:10:05 UTC and may require a fresh source retarget before any decision. +The exact future invocation must also bind the evidence owner, a custody +checkpoint within 24 hours, and the owning GitHub reconciliation record. This +checkpoint schedules reconciliation; it does not grant automatic deletion. + +`rust-preparation-source.json` records the exact source closure. A trusted +external isolated Python preloader must first authenticate that manifest against +the SHA-256 copied from the reviewed GitHub record, then authenticate every +member before compiling any controller/platform code. Capture and execute the +same verified byte buffers. Do not import a worktree module for verification, +reopen a verified code path, or add modules to the execution closure. + +The following literal preloader is source-only and is **not an operation +invocation**. Copy it from the reviewed GitHub record, never an unverified task +file. Its directory argument must contain only the manifest and its four +members. It authenticates all members before compiling the three controller modules +and refuses an operational admission value. No such staging directory is +created here. The current trusted manifest SHA is pinned in the code below. + +```sh +/usr/bin/python3 -I -S -B - "$controller_source_directory" <<'PY' +import hashlib, json, os, pwd, stat, sys, types +from pathlib import Path + +root = Path(sys.argv[1]) +if not (root.is_absolute() and root.resolve() == root and root.is_dir()): + raise RuntimeError('source-root') +if root.is_relative_to(Path(pwd.getpwuid(os.getuid()).pw_dir)): + raise RuntimeError('source-home') +names = {'rust_preparation.py', 'rust_preparation_platform.py', 'rust_preparation_authority.py', 'test_rust_preparation.py'} +root_fd = os.open('/', os.O_RDONLY | os.O_DIRECTORY) +try: + for part in root.parts[1:]: + next_fd = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=root_fd) + os.close(root_fd) + root_fd = next_fd + if set(os.listdir(root_fd)) != names | {'rust-preparation-source.json'}: + raise RuntimeError('source-closure') +except BaseException: + os.close(root_fd) + raise +def capture(name, limit): + fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=root_fd) + try: + info = os.fstat(fd) + if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_size > limit: + raise RuntimeError('source-member') + with os.fdopen(os.dup(fd), 'rb') as stream: + raw = stream.read(limit + 1) + if len(raw) > limit: + raise RuntimeError('source-limit') + return raw + finally: + os.close(fd) +try: + raw = capture('rust-preparation-source.json', 16384) + expected_manifest = 'bc15b52893ad2b82128846643c7a5d0ff3baad242059ecca21bb816b55eb74cf' + if hashlib.sha256(raw).hexdigest() != expected_manifest: + raise RuntimeError('source-manifest-hash') + manifest = json.loads(raw) + if manifest['status'] != 'source-only' or manifest['operation_admission'] is not None or manifest['installed_authority_admission'] is not None or set(manifest['files']) != names: + raise RuntimeError('source-status') + captured = {name: capture(name, 262144) for name in names} + if any(hashlib.sha256(captured[n]).hexdigest() != manifest['files'][n] for n in names): + raise RuntimeError('source-member-hash') +finally: + os.close(root_fd) +modules = {} +for name in ('rust_preparation', 'rust_preparation_authority', 'rust_preparation_platform'): + module = types.ModuleType(name) + sys.modules[name] = module + exec(compile(captured[name + '.py'], '', 'exec'), module.__dict__) + modules[name] = module +controller = modules['rust_preparation'] +controller._VERIFIED_MODULES = {n: captured[n + '.py'] for n in modules} +modules['rust_preparation_authority'].VERIFIED_SOURCE_SHA256 = hashlib.sha256(captured['rust_preparation_authority.py']).hexdigest() +modules['rust_preparation_platform'].VERIFIED_SOURCE_SHA256 = hashlib.sha256(captured['rust_preparation_platform.py']).hexdigest() +if controller.OPERATION_ADMISSION is not None or modules['rust_preparation_authority'].AUTHORITY_ADMISSION is not None: + raise RuntimeError('operation-remains-held') +sys.argv = [''] +raise SystemExit(controller.cli()) +PY +``` + +Only a separately reviewed source/invocation change can bind +`OPERATION_ADMISSION` to the canonical admitted-spec SHA-256. No such hash is +set. Installed authority requires its own independently reviewed manifest hash +in `AUTHORITY_ADMISSION`; no such manifest or hash is supplied. The external +preloader supplies the three authenticated module byte buffers +as `_VERIFIED_MODULES` and installs the authority/platform modules from those buffers, +binding `VERIFIED_SOURCE_SHA256` before `cli()`. Normal source tests do not do +this with operational admission enabled. + +The future unit command captures all three verified module buffers in its argument +payload. Worker and independent stop command compile those captured bytes, +with isolated Python, site startup disabled and no bytecode writes. They do not +read controller code from mutable host paths after registration. Effective unit +commands, deadlines, limits and active state are read back before allocation. +An explicit `env -i` boundary prevents the manager's inherited credentials, +proxy or startup settings from reaching the worker and stop interpreter. +The independent stop command is registered before transport directories are +created. A partial private transport without its completed ownership record +is unresolved evidence, not permission to delete an unverified directory. +Worker readiness binds the effective main PID and running interpreter; stop +cleanup binds the effective control PID and exact captured stop command. +Creation-window drift prevents use but cannot prevent nonce-owned cleanup. +No service has been registered or read back by this implementation's tests. + +The controller records a pending daemon operation before issuing its request. +A killed/timed-out client leaves that record unresolved. Cleanup still removes +only independently discovered nonce-owned containers, but reports daemon +terminal state unknown and retains bounded evidence. It does not claim a +successful cleanup or allow another phase merely because a client stopped. +Unknown native work requires General's decision and independent reconciliation. +CLI exit zero does not complete a mutation journal. Typed returned identity and +independent readback must be durably appended and synced before a completion +record can append. Pending and completed entries remain in the bounded journal; +no journal filename is deleted. Copy readback is verified before commit as well as after the image +snapshot. Malformed output or loss between exit and persistence stays unresolved. + +Transport creation uses a held parent descriptor and immediately opens/binds +the root before any child or record write. Child directories are also held and +their device/inode identities enter the ownership record. All writes use these +descriptors. A detected root/child namespace swap refuses use and never adopts +the replacement for writes. Reopened worker/stop transports must match all three +directory identities. Docker config and HOME resolve through the verified child +descriptors, rather than mutable child names; unbound runners use an unavailable +kernel FD path. The anonymous registry config contains no credentials or helpers +and disables platform helper-store discovery through its explicit empty registry +entry ([Docker loader](https://github.com/docker/cli/blob/master/cli/config/config.go), +[auth-map check](https://github.com/docker/cli/blob/master/cli/config/configfile/file.go)). + +Finalization performs no transport unlink or rmdir. The exact private tree is +retained as bounded attempt evidence: three directories and five regular files +(ownership, operation results, mutation journal, worker context and anonymous +Docker config), at most 144 KiB of file contents. Every member, mode, owner, +root/child identity and namespace binding must match; pending, extra or unknown +members prevent acceptance. The receipt records root device/inode, member hashes, +owner, checkpoint and GitHub reconciliation reference. Namespace swaps leave +both the original bound inode and replacement untouched and remain unresolved. +Later deletion requires separate reviewed authority; retaining this evidence +does not select a cleanup command or permit unbounded accumulation. Controls +include the exact predecessor functions and root/child swap successors. + +The installed-authority manifest must come from an independent complete-loader +and package review. It binds a boot generation, exact entry aliases and resolved +ELF bytes for Python, env, Docker, busctl and systemd-run; full catalog membership +and bytes for their reviewed library/runtime roots; loader configuration and +package receipts; and the complete effective isolated Python search path. +Every tool must use the whole pinned catalog. Additional members, changed bytes, +an unbound search root or an incomplete outer receipt refuse use. Root-owned +non-writable files and ancestors are required. Repeated reads cache hashes only +under unchanged kernel metadata, while reopening files and checking membership. +The initial isolated interpreter and this observer still require the outer +trusted runtime review before they may consume that manifest. + +Both worker and stop repeat this installed identity check. Manager/daemon PIDs, +start ticks, user IDs, executable and loaded-file paths bind the processes; +the manager's actual D-Bus owner PID and the Docker socket's kernel peer PID +must match. Command execution uses a checked open ELF descriptor. Socket, +catalog or process replacement refuses further native requests. This is an +implementation for later exact host review, not acceptance of any installed +host/runtime, package catalog, service-manager or daemon instance. + +The worker's image receipt explicitly leaves cleanup pending. The independent +launcher waits for the unit and captures both worker and stop output through +finite pipes, then joins exactly two terminal records. It binds invocation, +attempt/source/authority/command hashes and distinct main/control PIDs; actual +worker exit and cleanup status; exact base/dependency identities; seal proof, +image/config/layer identities and readback; durably bound operations; no pending +work or owned survivors; foreign-container preservation; and bounded transport +evidence custody with the exact owner and checkpoint. +Lost/duplicate/spliced output, nonzero wait, malformed identities or unresolved +cleanup cannot yield a preparation receipt. Collector loss leaves no acceptance; +independent stop cleanup remains the service manager's responsibility. + +Only that combined receipt can report prepared inputs. Derived images remain +local and bound by full content ID. No browser execution or publication follows. +All public policy flags remain false and browser outcome not-run. diff --git a/security/browser-headers/rust-preparation-source.json b/security/browser-headers/rust-preparation-source.json new file mode 100644 index 00000000000..6f00314ce76 --- /dev/null +++ b/security/browser-headers/rust-preparation-source.json @@ -0,0 +1,13 @@ +{ + "schema": "durable-workflow.preparation-source/v1", + "status": "source-only", + "operation_admission": null, + "node_packet_sha256": "c6c735624906c4680d162ba989ebd55f00a8124e6d8ca72f781f8fee646ae281", + "files": { + "rust_preparation.py": "a6a95daacafbbd4120a3189ecace7843e4f2ff9b0536cf9c2f7aa69ce0aefcc0", + "rust_preparation_platform.py": "819d93c2df9d762babfb2ac061268c4205ea7d18f18244cc79eb9684f3f078d4", + "test_rust_preparation.py": "c78b5c42d42e9f61124668a24b226e2f3383a61bf5ce03b6ee2d533665da2e1d", + "rust_preparation_authority.py": "c6d14a86283339de5924d59d4ac7a6d2bff0fe6260f63e4ca7993030adfb9950" + }, + "installed_authority_admission": null +} diff --git a/security/browser-headers/rust_preparation.py b/security/browser-headers/rust_preparation.py new file mode 100644 index 00000000000..123d4e0ff24 --- /dev/null +++ b/security/browser-headers/rust_preparation.py @@ -0,0 +1,546 @@ +"""One-shot preparation source; operational admission is deliberately unset. + +The inert controls use the same validation/lifecycle functions with fake effects. +No package/Node module is imported, and import/CLI cannot allocate a resource. +""" + +from __future__ import annotations + +import base64 +import datetime as dt +import gzip +import hashlib +import io +import json +import os +import re +import stat +import sys +import tarfile +import zipfile +from dataclasses import dataclass +from pathlib import PurePosixPath + + +OPERATION_ADMISSION = None +OWNER = "durable-workflow-github-io" +PACKET_SHA = "c6c735624906c4680d162ba989ebd55f00a8124e6d8ca72f781f8fee646ae281" +ZIP_SHA = "ea80b0d3191da525152b8f02ff2e3253a29ed93a723dfeae4c344eefb392cada" +ZIP_BYTES = 4067287 +SUBJECT_EXPIRY = "2026-11-06T03:10:05+00:00" +BASE_IMAGE = "mcr.microsoft.com/playwright@sha256:bc72a8df40831ded821e3cbc79c318c21b44d6e51dff9fd89ace7a8249c5b462" +BASE_ID = "sha256:847f71bbb3f17fc1032d50d75bf6fb3fdcf188b16dff81f24667fe9ae9409e01" +SUBJECT = { + "repository": "durable-workflow/sdk-rust", + "run_id": 41959739724, + "artifact_id": 13384761426, + "tooling_revision": "46cf36514f7a39e819213771abf9efc82321c062", + "zip_bytes": ZIP_BYTES, + "zip_sha256": ZIP_SHA, +} +INVENTORY_NAME = "rust-inventory.json" +PACKET_MEMBERS = ( + "rust-browser.test.cjs", "rust-browser-inputs.cjs", + "rust-browser-source.test.cjs", "rust-policy.json", INVENTORY_NAME, + "rust-rule.candidate.json", "runtime/package.json", "runtime/package-lock.json", +) +LABEL_OWNER = "com.durable-workflow.agent" +LABEL_NONCE = "com.durable-workflow.attempt" +LABEL_ROLE = "com.durable-workflow.purpose" +ROLES = ("assemble", "verify") +MIB = 1024 * 1024 +FILE_LIMIT = 10000 +TREE_LIMIT = 64 * MIB +TAR_LIMIT = 128 * MIB +HEX_64 = r"[0-9a-f]{64}" +IMAGE_ID = r"sha256:[0-9a-f]{64}" +RETAINED_EVIDENCE = "retained-bounded-evidence" +EVIDENCE_LIMIT = 144 * 1024 +DEPENDENCY_INTEGRITY = { + "playwright": "sha512-kVzTnYFYEyQ9RGzBjNCd2+yEqWmVbxl3AHPynv3qQ6JTjv4a+kwRcnqQwIpuP3NnrnIrE/Q6wWAIAvMfIE2/Yg==", + "playwright-core": "sha512-T9r+MZkTECl2+oUcZ26YgZzTrYidTsIhYYkJix+6iDOymOb1FB2RLLSOxRGnZGv5DStt+aduIVbt+E8WygXYpg==", +} + + +class Refusal(RuntimeError): + """Fixed diagnostic only; input values/paths/credentials are not formatted.""" + + +def require(condition, code): + if not condition: + raise Refusal(code) + + +def digest(data): + return hashlib.sha256(data).hexdigest() + + +def strict_json(data): + def pairs(values): + result = {} + for key, value in values: + require(key not in result, "duplicate-json-key") + result[key] = value + return result + + def invalid_constant(_value): + raise Refusal("nonfinite-json") + + return json.loads(data, object_pairs_hook=pairs, parse_constant=invalid_constant) + + +def safe_name(name): + require(isinstance(name, str) and 0 < len(name) <= 256, "member-name") + require(not any(c in name for c in ("\\", "\x00", "\r", "\n")), "member-name") + path = PurePosixPath(name) + require(not path.is_absolute(), "absolute-member") + require(all(p not in ("", ".", "..") for p in name.split("/")), "member-traversal") + require(path.as_posix() == name, "noncanonical-member") + return name + + +@dataclass(frozen=True) +class Entry: + path: str + data: bytes + mode: int = 0o444 + + def __post_init__(self): + safe_name(self.path) + require(type(self.data) is bytes, "mutable-entry") + require(self.mode in (0o444, 0o555), "entry-mode") + + +def index(entries): + result = {} + total = 0 + for entry in entries: + require(entry.path not in result, "duplicate-member") + result[entry.path] = entry + total += len(entry.data) + require(len(result) <= FILE_LIMIT and total <= TREE_LIMIT, "tree-limit") + for name in result: + require(not any(p.as_posix() in result for p in PurePosixPath(name).parents + if p.as_posix() != "."), "file-parent") + return result + + +def normalized_mode(mode): + return 0o555 if mode & 0o111 else 0o444 + + +def tar_entries(data, prefix=""): + require(type(data) is bytes and len(data) <= TAR_LIMIT, "tar-limit") + result = [] + seen = set() + total = 0 + with tarfile.open(fileobj=io.BytesIO(data), mode="r:") as archive: + for member in archive: + name = member.name.removeprefix("./").rstrip("/") + if name in ("", "."): + require(member.isdir(), "tar-root") + continue + safe_name(name) + require(name not in seen, "duplicate-member") + seen.add(name) + require(len(seen) <= 20000, "member-limit") + require(member.isfile() or member.isdir(), "nonregular-member") + require(not member.sparse and not member.mode & 0o7000, "special-member") + if prefix: + require(name == prefix or name.startswith(prefix + "/"), "archive-prefix") + if member.isdir(): + continue + require(name != prefix, "archive-prefix-file") + total += member.size + require(0 <= member.size <= TREE_LIMIT and total <= TREE_LIMIT, "tree-limit") + raw = archive.extractfile(member).read(member.size + 1) + require(len(raw) == member.size, "short-member") + relative = name[len(prefix) + 1:] if prefix else name + result.append(Entry(relative, raw, normalized_mode(member.mode))) + index(result) + return tuple(result) + + +def packet_from_entries(entries): + files = index(entries) + names = set(PACKET_MEMBERS) | {"rust-local-browser-packet.json"} + require(set(files) == names, "packet-closure") + raw = files["rust-local-browser-packet.json"].data + require(digest(raw) == PACKET_SHA, "packet-manifest-hash") + packet = strict_json(raw) + require(packet["run_status"] == "not-run" and set(packet["files"]) == set(PACKET_MEMBERS), "packet-status") + for name, expected in packet["files"].items(): + require(digest(files[name].data) == expected, "packet-member-hash") + inventory = strict_json(files[INVENTORY_NAME].data) + candidate = strict_json(files["rust-rule.candidate.json"].data) + policy = strict_json(files["rust-policy.json"].data) + for value in (packet, inventory, candidate, policy): + require(value["input_artifact"] == SUBJECT, "subject-drift") + require(candidate["rule"]["enabled"] is False, "enabled-rule") + require(candidate["rollout_approved"] is False and candidate["compatibility_accepted"] is False, "approval-drift") + require(candidate["source_preparation"] is None and inventory["source_preparation"] is None, "prepared-subject") + require(candidate["inventory_sha256"] == digest(files[INVENTORY_NAME].data), "inventory-hash") + return inventory, strict_json(files["runtime/package-lock.json"].data) + + +def artifact_from_zip(raw, inventory, now): + require(now < dt.datetime.fromisoformat(SUBJECT_EXPIRY), "expired-artifact") + require(len(raw) == ZIP_BYTES and digest(raw) == ZIP_SHA, "artifact-zip-hash") + with zipfile.ZipFile(io.BytesIO(raw)) as archive: + members = archive.infolist() + require(len(members) == 1 and members[0].orig_filename == "artifact.tar", "zip-membership") + require(members[0].file_size <= TAR_LIMIT, "zip-inflate-limit") + payload = archive.read(members[0]) + entries = tar_entries(payload) + actual = index(entries) + expected = inventory["files"] + require(len(expected) == 484 and len(actual) == 484, "artifact-count") + require(len({row["path"] for row in expected}) == 484, "inventory-duplicates") + require(set(actual) == {row["path"] for row in expected}, "artifact-membership") + for row in expected: + entry = actual[row["path"]] + require(len(entry.data) == row["bytes"] and digest(entry.data) == row["sha256"], "artifact-file-hash") + require(strict_json(actual["__build.json"].data) == inventory["build_identity"], "build-identity") + return entries + + +def dependency_entries(raw, name, lock): + require(name in ("playwright", "playwright-core") and len(raw) <= 16 * MIB, "dependency-input") + record = lock["packages"]["node_modules/" + name] + expected_url = "https://registry.npmjs.org/" + name + "/-/" + name + "-1.64.0.tgz" + require(record["resolved"] == expected_url and record["version"] == "1.64.0", "dependency-lock") + expected = "sha512-" + base64.b64encode(hashlib.sha512(raw).digest()).decode("ascii") + require(expected == record["integrity"], "dependency-integrity") + with gzip.GzipFile(fileobj=io.BytesIO(raw)) as compressed: + unpacked = compressed.read(TAR_LIMIT + 1) + require(len(unpacked) <= TAR_LIMIT, "dependency-inflate-limit") + entries = tar_entries(unpacked, "package") + package = strict_json(index(entries)["package.json"].data) + require(package["name"] == name and package["version"] == "1.64.0", "package-identity") + dependencies = {"playwright-core": "1.64.0"} if name == "playwright" else {} + require(package.get("dependencies", {}) == dependencies, "package-graph") + for key in ("optionalDependencies", "peerDependencies", "bundledDependencies", "bundleDependencies"): + require(not package.get(key), "package-graph") + return tuple(Entry("runtime/node_modules/" + name + "/" + e.path, e.data, e.mode) for e in entries) + + +def public_tar(packet, artifact, dependencies): + entries = tuple(Entry("packet/" + e.path, e.data, e.mode) for e in packet) + entries += tuple(Entry("site/" + e.path, e.data, e.mode) for e in artifact) + entries += tuple(dependencies) + files = index(entries) + directories = {p.as_posix() for name in files for p in PurePosixPath(name).parents if str(p) != "."} + output = io.BytesIO() + with tarfile.open(fileobj=output, mode="w", format=tarfile.USTAR_FORMAT) as archive: + for name in sorted(directories): + info = tarfile.TarInfo(name) + info.type = tarfile.DIRTYPE + info.uid = info.gid = 1000 + info.mode = 0o555 + archive.addfile(info) + for name, entry in sorted(files.items()): + info = tarfile.TarInfo(name) + info.uid = info.gid = 1000 + info.mode = entry.mode + info.size = len(entry.data) + archive.addfile(info, io.BytesIO(entry.data)) + raw = output.getvalue() + require(len(raw) <= TAR_LIMIT, "public-tar-limit") + return raw, tuple(files[name] for name in sorted(files)) + + +def held_regular(path, binding): + """Read from one no-follow descriptor; never reopen a validated pathname.""" + require(isinstance(path, str) and path.startswith("/") and path != "/", "handoff-path") + parts = path.split("/")[1:] + require(all(p not in ("", ".", "..") for p in parts), "handoff-path") + descriptor = os.open("/", os.O_RDONLY | os.O_DIRECTORY) + try: + for part in parts[:-1]: + following = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=descriptor) + os.close(descriptor) + descriptor = following + parent = os.fstat(descriptor) + require(stat.S_IMODE(parent.st_mode) == binding["parent_mode"], "handoff-parent-mode") + require((parent.st_uid, parent.st_gid) == (binding["uid"], binding["gid"]), "handoff-parent-owner") + leaf = os.open(parts[-1], os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=descriptor) + try: + observed = os.fstat(leaf) + require(stat.S_ISREG(observed.st_mode) and observed.st_nlink == 1, "handoff-type") + for key, value in (("uid", observed.st_uid), ("gid", observed.st_gid), + ("mode", stat.S_IMODE(observed.st_mode)), ("device", observed.st_dev), + ("inode", observed.st_ino), ("bytes", observed.st_size)): + require(value == binding[key], "handoff-metadata") + require(0 <= binding["bytes"] <= TAR_LIMIT, "handoff-limit") + chunks = [] + remaining = binding["bytes"] + 1 + while remaining: + chunk = os.read(leaf, min(remaining, MIB)) + if not chunk: + break + chunks.append(chunk) + remaining -= len(chunk) + raw = b"".join(chunks) + require(len(raw) == binding["bytes"] and digest(raw) == binding["sha256"], "handoff-buffer") + return raw + finally: + os.close(leaf) + finally: + os.close(descriptor) + + +@dataclass(frozen=True) +class Ownership: + nonce: str + + def __post_init__(self): + require(re.fullmatch(HEX_64, self.nonce) is not None, "attempt-nonce") + + def name(self, role): + require(role in ROLES, "container-role") + return OWNER + "-rust-custody-" + role + + def labels(self, role): + self.name(role) + return {LABEL_OWNER: OWNER, LABEL_NONCE: self.nonce, LABEL_ROLE: "rust-custody-" + role} + + +def owned_identity(record, ownership, role): + require(re.fullmatch(HEX_64, record["Id"]) is not None, "container-id") + require(record["Name"] == "/" + ownership.name(role), "container-name") + labels = record["Config"]["Labels"] + require(all(labels.get(k) == v for k, v in ownership.labels(role).items()), "container-labels") + return record["Id"] + + +def inspect_allocation(record, ownership, role, image): + cid = owned_identity(record, ownership, role) + require(record["Image"] == image, "container-image") + require(record["State"]["Status"] == "created" and record["State"]["Running"] is False, "container-started") + require(record.get("WithinPreparationWindow") is True, "allocation-window") + require(not record["Mounts"] and not record["Config"].get("Volumes"), "container-mounts") + host = record["HostConfig"] + require(host["NetworkMode"] == "none" and host["ReadonlyRootfs"] is (role == "verify"), "container-isolation") + require(host["NanoCpus"] == 1000000000 and host["Memory"] == 512 * MIB and host["PidsLimit"] == 64, "container-limits") + require(host["Init"] is True and host["Privileged"] is False and host["CapDrop"] == ["ALL"], "container-privilege") + require(host["PidMode"] == "" and host["IpcMode"] == "private" and host["UTSMode"] == "", "container-namespace") + require(host["PublishAllPorts"] is False and not host.get("DeviceRequests") and not host.get("Tmpfs"), "container-extra-isolation") + require(not host.get("CapAdd") and not host.get("Devices") and not host.get("Binds") and not host.get("PortBindings"), "container-extra-access") + require(host["SecurityOpt"] == ["no-new-privileges:true"], "container-security") + require(record["Config"]["Cmd"] == ["/bin/false"] and not record["Config"].get("Entrypoint"), "container-command") + return cid + + +def discover_owned(effects, ownership, role): + candidates = effects.discover(ownership, role) + require(len(candidates) <= 1, "ambiguous-allocation") + if not candidates: + return None + cid = candidates[0] + require(re.fullmatch(HEX_64, cid) is not None, "discovered-id") + record = effects.inspect(cid) + owned_identity(record, ownership, role) + return record + + +def allocate(effects, ownership, role, image): + effects.supervisor_ready() + require(not effects.name_exists(ownership.name(role)), "foreign-name-collision") + require(not effects.discover(ownership, role), "existing-attempt-allocation") + returned = effects.create(ownership, role, image) + record = discover_owned(effects, ownership, role) + require(re.fullmatch(HEX_64, returned) is not None, "returned-id") + require(record is not None and record["Id"] == returned, "returned-discovered-mismatch") + cid = inspect_allocation(record, ownership, role, image) + effects.complete_operation("create-" + role, {"container_id": cid}) + effects.supervisor_ready() + return cid + + +def cleanup(effects, ownership): + """Independent stop path rediscovers ownership; a bad returned ID is irrelevant.""" + problems = [] + for role in ROLES: + try: + record = discover_owned(effects, ownership, role) + if record is not None: + final_record = effects.inspect(record["Id"]) + cid = owned_identity(final_record, ownership, role) + effects.remove(cid) + if final_record.get("WithinPreparationWindow") is not True: + problems.append("allocation-window") + require(not effects.discover(ownership, role), "owned-survivor") + except Exception: + problems.append("cleanup-identity-or-survivor") + if effects.uncertain_daemon_operation(): + problems.append("daemon-terminal-unknown") + return tuple(problems) + + +def verify_image(record, base_layers): + require(re.fullmatch(IMAGE_ID, record["Id"]) is not None, "image-id") + layers = record["RootFS"]["Layers"] + require(layers[:-1] == base_layers and len(layers) == len(base_layers) + 1, "image-layer-chain") + require(not record["Config"].get("Volumes") and not record["Config"].get("OnBuild"), "image-hooks") + require(record["Config"]["Cmd"] == ["/bin/false"] and not record["Config"].get("Entrypoint"), "image-command") + return record["Id"] + + +def assemble(effects, ownership, raw, expected, base_layers): + """Effects are injected; the CLI never supplies live effects without admission.""" + effects.supervisor_ready() + sealed = effects.seal(raw) + try: + require(sealed.verified_hash() == digest(raw), "seal-readback") + cid = allocate(effects, ownership, "assemble", BASE_ID) + effects.require_safe_prefixes(cid) + effects.supervisor_ready() + effects.copy_sealed(cid, sealed) + require(effects.readback(cid) == expected, "copy-readback") + effects.complete_operation("copy", {"container_id": cid, "input_tar_sha256": digest(raw)}) + effects.supervisor_ready() + returned = effects.commit(cid) + require(re.fullmatch(IMAGE_ID, returned) is not None, "returned-image-id") + image = effects.image_inspect(returned) + require(verify_image(image, base_layers) == returned, "image-id-mismatch") + effects.complete_operation("commit", {"image_id": returned, "config_sha256": returned[7:], + "rootfs_layers": image["RootFS"]["Layers"]}) + owned_identity(effects.inspect(cid), ownership, "assemble") + effects.remove(cid) + require(not effects.discover(ownership, "assemble"), "owned-survivor") + verifier = allocate(effects, ownership, "verify", returned) + require(effects.readback(verifier) == expected, "image-readback") + return returned + finally: + sealed.close() + + +def terminal_process_binding(records, expected): + for record in records: + require(record["schema"] == "durable-workflow.preparation-stage/v1", "terminal-stage-schema") + for key in ("unit", "attempt_sha256", "source_bundle_sha256", "authority_sha256"): + require(record[key] == expected[key], "terminal-binding-drift") + require(re.fullmatch(r"[0-9a-f]{32}", record["invocation_id"]) is not None, "terminal-invocation") + require(type(record["pid"]) is int and record["pid"] > 1, "terminal-pid") + worker = next(r for r in records if r["role"] == "worker") + stop = next(r for r in records if r["role"] == "stop") + require(worker["invocation_id"] == stop["invocation_id"], "terminal-unit-generation") + require(worker["command_sha256"] == expected["worker_command_sha256"] + and stop["command_sha256"] == expected["stop_command_sha256"], "terminal-command") + require(stop["worker_pid"] == worker["pid"] and stop["pid"] != worker["pid"], "terminal-worker-stop-pid") + statuses = [stop[k] for k in ("worker_exit_code", "worker_exit_status", "cleanup_exit_status")] + require(all(type(v) is int for v in statuses) and statuses == [1, 0, 0], "terminal-worker-stop-status") + require(worker["outcome"] == "prepared-cleanup-pending" and stop["outcome"] == "complete", "terminal-stage-outcome") + return worker, stop + + +def terminal_input_binding(worker, expected): + require(worker["subject"] == SUBJECT and worker["node_packet_sha256"] == PACKET_SHA, "terminal-subject") + require(worker["base_image_id"] == BASE_ID, "terminal-base") + require(re.fullmatch(HEX_64, worker["input_tar_sha256"]) is not None, "terminal-input-tar") + require(worker["seal"]["sha256"] == worker["input_tar_sha256"] + and worker["seal"]["flags"] == expected["seal_flags"], "terminal-seal") + require(re.fullmatch(IMAGE_ID, worker["image"]["id"]) is not None + and worker["image"]["config_sha256"] == worker["image"]["id"][7:], "terminal-image") + require(worker["image"]["layers"][:-1] == worker["base_layers"] + and len(worker["image"]["layers"]) == len(worker["base_layers"]) + 1, "terminal-image-layers") + require(worker["base_layers"] and all(re.fullmatch(IMAGE_ID, layer) is not None + for layer in worker["image"]["layers"]), "terminal-layer-identities") + require(worker["readback_verified"] is True and re.fullmatch(HEX_64, worker["readback_sha256"]) is not None, "terminal-readback") + require(set(worker["dependencies"]) == {"playwright", "playwright-core"}, "terminal-dependencies") + for name, dependency in worker["dependencies"].items(): + require(dependency["version"] == "1.64.0" and dependency["integrity"] == DEPENDENCY_INTEGRITY[name] + and re.fullmatch(HEX_64, dependency["sha256"]) is not None, "terminal-dependency-identity") + + +def terminal_cleanup_binding(worker, stop, expected): + require(stop["pending"] is False and stop["owned_survivors"] == [] and stop["problems"] == [], "terminal-unresolved-work") + require(stop["foreign_preserved"] is True, "terminal-foreign-preservation") + terminal_retained_evidence(stop["transport"], expected) + require(worker["foreign_before_sha256"] == stop["foreign_before_sha256"], "terminal-foreign-baseline") + require(re.fullmatch(HEX_64, worker["foreign_before_sha256"]) is not None, "terminal-foreign-baseline-hash") + operations = stop["operation_results"] + require([r["operation"] for r in operations] == ["pull", "create-assemble", "copy", "commit", "create-verify"], "terminal-operation-closure") + require(all(r["attempt_sha256"] == expected["attempt_sha256"] for r in operations), "terminal-operation-attempt") + commit = operations[3]["identity"] + require(commit == {"image_id": worker["image"]["id"], "config_sha256": worker["image"]["config_sha256"], + "rootfs_layers": worker["image"]["layers"]}, "terminal-commit-binding") + require(operations[0]["identity"] == {"base_image_id": BASE_ID, "rootfs_layers": worker["base_layers"]}, "terminal-pull-binding") + require(operations[2]["identity"]["input_tar_sha256"] == worker["input_tar_sha256"], "terminal-copy-binding") + assembler, verifier = (operations[i]["identity"]["container_id"] for i in (1, 4)) + require(re.fullmatch(HEX_64, assembler) is not None and re.fullmatch(HEX_64, verifier) is not None + and assembler != verifier and operations[2]["identity"]["container_id"] == assembler, "terminal-container-binding") + + +def terminal_retained_evidence(evidence, expected): + require(type(evidence) is dict and evidence["state"] == RETAINED_EVIDENCE and evidence["owner"] == OWNER, + "terminal-transport-custody") + require(evidence["attempt_sha256"] == expected["attempt_sha256"] + and evidence["custody_due"] == expected["custody_due"] + and evidence["reconciliation_ref"] == expected["reconciliation_ref"], "terminal-transport-attempt") + require(all(type(evidence[k]) is int and evidence[k] >= 0 for k in ("device", "inode", "uid", "gid")) + and evidence["inode"] > 0 and evidence["mode"] == 0o700, "terminal-transport-identity") + require((evidence["uid"], evidence["gid"]) == (expected["transport_uid"], expected["transport_gid"]), "terminal-transport-owner") + rows = evidence["files"] + require(len(rows) == 5 and {r["path"] for r in rows} + == {"ownership.json", "operation-results.jsonl", "worker-context.json", "pending-operation.json", "docker/config.json"}, + "terminal-transport-members") + require(all(type(r["bytes"]) is int and r["bytes"] >= 0 and re.fullmatch(HEX_64, r["sha256"]) is not None + for r in rows), "terminal-transport-member-identity") + require(evidence["bytes"] == sum(r["bytes"] for r in rows) <= EVIDENCE_LIMIT, "terminal-transport-size") + require(evidence["sha256"] == digest(json.dumps(rows, sort_keys=True, separators=(",", ":")).encode()), + "terminal-transport-manifest") + + +def collect_terminal(raw, expected, wait_exit_code): + """Independent join; neither a worker line nor launch success is acceptance.""" + require(type(wait_exit_code) is int and wait_exit_code == 0, "unit-wait-failed") + require(type(raw) is bytes and len(raw) <= MIB, "terminal-receipt-limit") + records = [strict_json(line) for line in raw.splitlines() if line.strip()] + require(len(records) == 2 and all(type(r) is dict for r in records) + and {r.get("role") for r in records} == {"worker", "stop"}, "terminal-record-pair") + worker, stop = terminal_process_binding(records, expected) + terminal_input_binding(worker, expected) + terminal_cleanup_binding(worker, stop, expected) + return {"schema": "durable-workflow.preparation-terminal/v1", "outcome": "prepared-inputs-only", + "unit": expected["unit"], "invocation_id": worker["invocation_id"], "attempt_sha256": expected["attempt_sha256"], + "source_bundle_sha256": expected["source_bundle_sha256"], "authority_sha256": expected["authority_sha256"], + "worker_pid": worker["pid"], "stop_pid": stop["pid"], "unit_wait_exit": 0, + "subject": SUBJECT, "base_image_id": BASE_ID, "dependencies": worker["dependencies"], + "seal": worker["seal"], "image": worker["image"], "readback_sha256": worker["readback_sha256"], + "pending": False, "owned_survivors": [], "foreign_preserved": True, "transport": stop["transport"], + "browser": "not-run", "provider": "not-run", "compatibility_accepted": False, "rollout_approved": False} + + +def main(): + # Enabling this requires a separately reviewed exact source/invocation change. + # There is deliberately no command-line/environment override for admission. + if OPERATION_ADMISSION is None: + print("Preparation admission is unset; source controls only.") + return 78 + modules = globals().get("_VERIFIED_MODULES") + require(type(modules) is dict and set(modules) == {"rust_preparation", "rust_preparation_authority", "rust_preparation_platform"}, "external-source-verification-required") + # The external preloader installs the authenticated platform module from its + # captured bytes, before calling main. Never resolve it from a worktree. + platform = sys.modules.get("rust_preparation_platform") + require(platform is not None and getattr(platform, "VERIFIED_SOURCE_SHA256", None) + == digest(modules["rust_preparation_platform"]), "platform-source-binding") + authority = sys.modules.get("rust_preparation_authority") + require(authority is not None and getattr(authority, "VERIFIED_SOURCE_SHA256", None) + == digest(modules["rust_preparation_authority"]), "authority-source-binding") + return platform.entrypoint(sys.argv[1:], modules, OPERATION_ADMISSION) + + +def cli(): + try: + return main() + except Exception as error: + reason = error.args[0] if isinstance(error, Refusal) and error.args else "controller-refused" + if not isinstance(reason, str) or re.fullmatch(r"[a-z-]{1,80}", reason) is None: + reason = "controller-refused" + print(json.dumps({"outcome": "refused", "reason": reason, "browser": "not-run"})) + return 1 + + +if __name__ == "__main__": + raise SystemExit(cli()) diff --git a/security/browser-headers/rust_preparation_authority.py b/security/browser-headers/rust_preparation_authority.py new file mode 100644 index 00000000000..d6386a55eed --- /dev/null +++ b/security/browser-headers/rust_preparation_authority.py @@ -0,0 +1,257 @@ +"""Finite installed authority, requiring its own independently reviewed digest. + +No host catalog is supplied or admitted here. Import performs no observation. +The outer reviewer must establish that the catalog covers the complete loader, +configuration, package/runtime and Python search closure, not a chosen subset. +""" + +import hashlib +import json +import os +from pathlib import Path +import socket +import stat +import struct +import sys + +import rust_preparation as core + + +AUTHORITY_ADMISSION = None +TOOLS = frozenset(("python", "env", "docker", "busctl", "systemd-run")) +PROCESSES = frozenset(("manager", "daemon")) +ENTRY_PATHS = {"python": "/usr/bin/python3", "env": "/usr/bin/env", "docker": "/usr/bin/docker", + "busctl": "/usr/bin/busctl", "systemd-run": "/usr/bin/systemd-run"} +LOADER_CONFIGS = frozenset(("/etc/ld.so.cache", "/etc/ld.so.conf", "/etc/nsswitch.conf")) +MAX_AUTHORITY_BYTES = 8 * 1024 * core.MIB + + +def file_fingerprint(info): + # Reading may update atime. Content/security metadata must stay stable. + return (info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns, info.st_ctime_ns, + info.st_mode, info.st_uid, info.st_gid, info.st_nlink) + + +def native_admitted(): + core.require(isinstance(core.OPERATION_ADMISSION, str) and isinstance(AUTHORITY_ADMISSION, str), "installed-authority-unadmitted") + + +def canonical_bytes(value): + return json.dumps(value, sort_keys=True, separators=(",", ":")).encode() + + +def catalog_schema(record): + required = {"schema", "tools", "processes", "catalogs", "python_search", "socket", "closure_receipt", + "loader_configs", "package_receipts", "boot_id"} + core.require(set(record) == required and record["schema"] == "durable-workflow.installed-authority/v1", "authority-schema") + core.require(set(record["tools"]) == TOOLS and set(record["processes"]) == PROCESSES, "authority-role-closure") + core.require(record["catalogs"] and record["python_search"], "authority-runtime-closure") + core.require(set(record["loader_configs"]) == LOADER_CONFIGS and record["package_receipts"], "authority-loader-packages") + for path in record["package_receipts"]: + core.require(any(path.startswith(root + "/") for root in ("/var/lib/dpkg", "/var/lib/rpm", "/usr/lib/sysimage/rpm")), "authority-package-path") + core.require(set(record["closure_receipt"]) == {"independent_review", "package_catalog_sha256", "complete_loader_closure"}, "authority-outer-receipt") + core.require(record["closure_receipt"]["complete_loader_closure"] is True + and record["closure_receipt"]["independent_review"], "authority-outer-review") + core.require(core.digest(canonical_bytes(record["package_receipts"])) + == record["closure_receipt"]["package_catalog_sha256"], "authority-package-receipt") + catalog_roots = set(record["catalogs"]) + for role, tool in record["tools"].items(): + core.require(tool["entry_path"] == ENTRY_PATHS[role], "authority-entry-role") + # Completeness comes from the separately pinned outer receipt. A tool + # cannot narrow that reviewed runtime footprint at invocation time. + core.require(set(tool["runtime_roots"]) == catalog_roots, "authority-tool-runtime-subset") + core.require(tool["binding"]["uid"] == 0 and not tool["binding"]["mode"] & 0o22, "authority-tool-writable") + for path in record["python_search"]: + core.require(any(path == root or path.startswith(root + "/") for root in catalog_roots), "authority-python-root-subset") + + +class HostAuthority: + def __init__(self, raw, observer, expected_digest=None): + expected = AUTHORITY_ADMISSION if expected_digest is None else expected_digest + core.require(isinstance(expected, str) and core.digest(raw) == expected, "installed-authority-unadmitted") + self.record = core.strict_json(raw) + catalog_schema(self.record) + self.digest = expected + self.observer = observer + self.verify() + + def verify(self): + expected = self.record + core.require(self.observer.boot_id() == expected["boot_id"], "authority-boot-drift") + core.require(tuple(self.observer.python_search()) == tuple(expected["python_search"]), "effective-python-search-drift") + for root, catalog in expected["catalogs"].items(): + core.require(self.observer.catalog(root) == catalog, "installed-runtime-catalog-drift") + for role, tool in expected["tools"].items(): + core.require(self.observer.file(tool["binding"]["path"]) == tool["binding"], "installed-tool-drift") + core.require(self.observer.resolve(tool["entry_path"]) == tool["binding"]["path"], "selected-tool-drift") + for group in ("loader_configs", "package_receipts"): + for path, binding in expected[group].items(): + core.require(self.observer.resolve(path) == binding["path"] + and self.observer.file(binding["path"]) == binding, "installed-configuration-drift") + for role, process in expected["processes"].items(): + core.require(self.observer.process(process["pid"]) == process, "effective-process-drift") + for path in process["mapped_files"]: + core.require(any(path == root or path.startswith(root + "/") for root in expected["catalogs"]) + or path in {t["binding"]["path"] for t in expected["tools"].values()} + or path in {p["exe"]["path"] for p in expected["processes"].values()}, "unbound-process-library") + core.require(self.observer.socket_identity(expected["socket"]["path"]) == expected["socket"], "daemon-socket-drift") + core.require(expected["socket"]["peer_pid"] == expected["processes"]["daemon"]["pid"], "socket-daemon-causality") + + def open_command(self, entry_path): + self.verify() + matches = [t for t in self.record["tools"].values() if t["entry_path"] == entry_path + or t["binding"]["path"] == entry_path] + core.require(len(matches) == 1, "unbound-command") + return self.observer.open_verified(matches[0]["binding"]) + + def command_path(self, role): + core.require(role in TOOLS, "unbound-command-role") + return self.record["tools"][role]["binding"]["path"] + + +class NativeObserver: + """Read-only outer evidence consumption; no commands or vault fallback.""" + + def __init__(self): + self.bytes_hashed = 0 + self.hash_cache = {} + + def boot_id(self): + native_admitted() + value = Path("/proc/sys/kernel/random/boot_id").read_text().strip() + core.require(len(value) == 36, "authority-boot-id") + return value + + def resolve(self, path): + resolved = Path(path).resolve(strict=True) + core.require(resolved.is_absolute(), "authority-canonical-path") + return str(resolved) + + def _open(self, path): + native_admitted() + target = Path(path) + core.require(target.is_absolute() and str(target) == self.resolve(path), "authority-file-path") + descriptor = os.open("/", os.O_RDONLY | os.O_DIRECTORY) + try: + for part in target.parts[1:-1]: + following = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=descriptor) + os.close(descriptor) + descriptor = following + parent = os.fstat(descriptor) + core.require(parent.st_uid == 0 and not parent.st_mode & 0o22, "authority-parent-writable") + return os.open(target.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=descriptor) + finally: + os.close(descriptor) + + def _metadata(self, path, descriptor): + info = os.fstat(descriptor) + core.require(stat.S_ISREG(info.st_mode) and info.st_uid == 0 and not info.st_mode & 0o22, "authority-file-writable") + core.require(info.st_size <= 256 * core.MIB, "authority-file-limit") + # Reviewed files and every ancestor are root-owned/non-writable by the + # executor. Cache only bytes for an unchanged kernel stat identity; + # still re-open and recheck membership/metadata on every observation. + key = file_fingerprint(info) + sha256 = self.hash_cache.get(key) + if sha256 is None: + sha256 = self._hash(descriptor) + core.require(file_fingerprint(os.fstat(descriptor)) == key, "authority-file-changed-during-read") + self.hash_cache[key] = sha256 + return {"path": path, "uid": info.st_uid, "gid": info.st_gid, "mode": stat.S_IMODE(info.st_mode), + "device": info.st_dev, "inode": info.st_ino, "bytes": info.st_size, "sha256": sha256} + + def _hash(self, descriptor): + observed = hashlib.sha256() + os.lseek(descriptor, 0, os.SEEK_SET) + while True: + chunk = os.read(descriptor, core.MIB) + if not chunk: + break + self.bytes_hashed += len(chunk) + core.require(self.bytes_hashed <= MAX_AUTHORITY_BYTES, "authority-read-budget") + observed.update(chunk) + os.lseek(descriptor, 0, os.SEEK_SET) + return observed.hexdigest() + + def file(self, path): + descriptor = self._open(path) + try: + return self._metadata(path, descriptor) + finally: + os.close(descriptor) + + def open_verified(self, binding): + descriptor = self._open(binding["path"]) + try: + core.require(self._metadata(binding["path"], descriptor) == binding, "consumed-tool-drift") + core.require(os.read(descriptor, 4) == b"\x7fELF", "authority-non-elf-tool") + os.lseek(descriptor, 0, os.SEEK_SET) + return descriptor + except BaseException: + os.close(descriptor) + raise + + def python_search(self): + return tuple(sys.path) + + def catalog(self, root): + native_admitted() + # Private/home/config/credential trees are never catalog inputs. + path = Path(root) + allowed = (Path("/usr/lib"), Path("/usr/local/lib"), Path("/etc/ld.so.conf.d")) + core.require(path.is_absolute() and any(path == p or path.is_relative_to(p) for p in allowed), "authority-runtime-root") + if not path.exists(): + core.require(not path.is_symlink(), "authority-missing-root-link") + return {"kind": "absent"} + if path.is_file(): + return {"kind": "file", "binding": self.file(root)} + core.require(str(path.resolve()) == root and not path.is_symlink(), "authority-root-link") + rows = {} + for directory, dirs, files in os.walk(root, followlinks=False): + info = os.lstat(directory) + core.require(info.st_uid == 0 and not info.st_mode & 0o22, "authority-directory-writable") + for name in sorted(dirs + files): + member = Path(directory) / name + relative = member.relative_to(path).as_posix() + current = member.lstat() + core.require(current.st_uid == 0, "authority-member-owner") + if member.is_symlink(): + target = member.resolve(strict=True) + core.require(target.is_relative_to(path), "authority-library-link-escape") + rows[relative] = {"kind": "link", "target": os.readlink(member)} + elif member.is_file(): + rows[relative] = {"kind": "file", "binding": self.file(str(member))} + else: + core.require(member.is_dir(), "authority-special-file") + core.require(not current.st_mode & 0o22, "authority-member-writable") + rows[relative] = {"kind": "directory", "mode": stat.S_IMODE(current.st_mode)} + core.require(len(rows) <= 100000, "authority-catalog-limit") + return {"kind": "directory", "members": rows} + + def process(self, pid): + native_admitted() + core.require(type(pid) is int and pid > 1, "authority-process-pid") + root = Path("/proc") / str(pid) + text = (root / "stat").read_text() + start = text[text.rfind(")") + 2:].split()[19] + status = (root / "status").read_text().splitlines() + uids = next([int(v) for v in line.split()[1:]] for line in status if line.startswith("Uid:")) + exe = str((root / "exe").resolve(strict=True)) + mapped = set() + for line in (root / "maps").read_text().splitlines(): + fields = line.split(maxsplit=5) + if len(fields) == 6 and fields[5].startswith("/"): + core.require(not fields[5].endswith(" (deleted)"), "authority-deleted-mapping") + mapped.add(fields[5]) + return {"pid": pid, "start_ticks": start, "uids": uids, "exe": self.file(exe), "mapped_files": sorted(mapped)} + + def socket_identity(self, path): + native_admitted() + core.require(path == "/run/docker.sock" and str(Path(path).resolve()) == path, "authority-socket-path") + info = os.lstat(path) + core.require(stat.S_ISSOCK(info.st_mode), "authority-socket-type") + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection: + connection.settimeout(5) + connection.connect(path) + pid, uid, gid = struct.unpack("3i", connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)) + return {"path": path, "device": info.st_dev, "inode": info.st_ino, "uid": info.st_uid, + "gid": info.st_gid, "mode": stat.S_IMODE(info.st_mode), "peer_pid": pid, "peer_uid": uid, "peer_gid": gid} diff --git a/security/browser-headers/rust_preparation_platform.py b/security/browser-headers/rust_preparation_platform.py new file mode 100644 index 00000000000..7e610b3a50e --- /dev/null +++ b/security/browser-headers/rust_preparation_platform.py @@ -0,0 +1,946 @@ +"""Guarded Linux adapter. Import has no effects; admission is unset in source. + +All native calls are behind rust_preparation.OPERATION_ADMISSION. Inert tests +inject a command runner and never instantiate the admitted native execution. +""" + +import base64 +import datetime as dt +import fcntl +import hashlib +import http.client +import json +import os +from pathlib import Path +import platform +import pwd +import re +import secrets +import selectors +import socket +import stat +import subprocess +import sys +import time +import zlib + +import rust_preparation as core +import rust_preparation_authority as installed + + +UNIT = core.OWNER + "-rust-preparation.service" +SERVICE = "org.freedesktop.systemd1" +INTERFACE = SERVICE + ".Service" +UNIT_INTERFACE = SERVICE + ".Unit" +PYTHON = "/usr/bin/python3" +ENV = "/usr/bin/env" +DOCKER = "/usr/bin/docker" +BUSCTL = "/usr/bin/busctl" +SYSTEMD_RUN = "/usr/bin/systemd-run" +BUS_JSON = "--json=short" +FD_PREFIX = "/proc/self/fd/" +CONFIG_FILE = "config.json" +RUNTIME_LIMIT = 600 +VERIFIED_SOURCE_SHA256 = None +OWNERSHIP_FILE = "ownership.json" +PENDING_FILE = "pending-operation.json" +RESULTS_FILE = "operation-results.jsonl" +CONTEXT_FILE = "worker-context.json" +# A nonempty auth map with an empty registry entry prevents Docker CLI's +# platform default-store detection, while supplying no credential or helper. +ANONYMOUS_CONFIG = b'{"auths":{"mcr.microsoft.com":{}}}' +_HOST_AUTHORITY = None + + +def admitted(): + core.require(isinstance(core.OPERATION_ADMISSION, str) + and re.fullmatch(core.HEX_64, core.OPERATION_ADMISSION) is not None, "preparation-unadmitted") + + +def require_installed_authority(): + admitted() + core.require(isinstance(_HOST_AUTHORITY, installed.HostAuthority) + and _HOST_AUTHORITY.digest == installed.AUTHORITY_ADMISSION, "installed-authority-unadmitted") + _HOST_AUTHORITY.verify() + return _HOST_AUTHORITY + + +def install_authority(spec): + admitted() + core.require(isinstance(installed.AUTHORITY_ADMISSION, str), "installed-authority-unadmitted") + binding = spec["authority_binding"] + raw = core.held_regular(binding["path"], binding) + authority = installed.HostAuthority(raw, installed.NativeObserver()) + global _HOST_AUTHORITY + _HOST_AUTHORITY = authority + + +def encoded(value): + return base64.b64encode(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).decode("ascii") + + +def decoded(value): + core.require(len(value) <= 65536, "argument-limit") + return core.strict_json(base64.b64decode(value, validate=True)) + + +def admission(spec, now): + required = {"artifact_handoff", "packet_bindings", "staging_path", "uid", "gid", + "docker_root", "metadata_at", "metadata_expires", "metadata_subject", + "live_build_sha256", "python_sha256", "env_sha256", "authority_binding", "retention"} + core.require(set(spec) == required, "admission-fields") + core.require(type(spec["uid"]) is int and type(spec["gid"]) is int, "admission-owner") + core.require((spec["uid"], spec["gid"]) == (os.getuid(), os.getgid()), "executor-owner") + core.require(spec["artifact_handoff"]["bytes"] == core.ZIP_BYTES + and spec["artifact_handoff"]["sha256"] == core.ZIP_SHA, "admission-subject") + expected_metadata = {**core.SUBJECT, "deployment_id": 7564108107, "event": "push", "branch": "main", + "attempt": 1, "run_conclusion": "success", "deployment_state": "success"} + core.require(spec["metadata_subject"] == expected_metadata, "metadata-subject") + core.require(set(spec["packet_bindings"]) == set(core.PACKET_MEMBERS) | {"rust-local-browser-packet.json"}, "admission-packet") + core.require(spec["live_build_sha256"] == "3e8d1827d1df8bf3c47ee8f0efcbd2f3692cb1bd8741f40c1271ef5c74f4a183", "live-subject") + observed = dt.datetime.fromisoformat(spec["metadata_at"]) + expiry = dt.datetime.fromisoformat(spec["metadata_expires"]) + core.require(0 <= (now - observed).total_seconds() <= 300, "stale-metadata") + core.require(expiry == dt.datetime.fromisoformat(core.SUBJECT_EXPIRY) + and now + dt.timedelta(seconds=RUNTIME_LIMIT + 15) < expiry, "artifact-window") + retention = spec["retention"] + core.require(set(retention) == {"owner", "custody_due", "reconciliation_ref"} + and retention["owner"] == core.OWNER, "retention-owner") + due = dt.datetime.fromisoformat(retention["custody_due"]) + core.require(now + dt.timedelta(seconds=RUNTIME_LIMIT + 15) < due <= now + dt.timedelta(hours=24), "retention-window") + core.require(re.fullmatch(r"https://github\.com/durable-workflow/(?:durable-workflow\.github\.io/pull/480|\.github/issues/268)#issuecomment-\d+", + retention["reconciliation_ref"], flags=re.ASCII) is not None, "retention-reconciliation") + for path in (spec["staging_path"], spec["docker_root"]): + core.require(re.fullmatch(r"/[A-Za-z0-9_./-]+", path) is not None, "native-path") + core.require(str(Path(path)) == path and ".." not in Path(path).parts, "native-path") + home = Path(pwd.getpwuid(os.getuid()).pw_dir) + bindings = [spec["artifact_handoff"], *spec["packet_bindings"].values()] + for binding in bindings: + path = Path(binding["path"]) + core.require(path.is_absolute() and not path.is_relative_to(home), "credential-tree-input") + core.require((binding["uid"], binding["gid"]) == (spec["uid"], spec["gid"]), "input-owner") + core.require(binding["mode"] == 0o600 and binding["parent_mode"] == 0o700, "input-mode") + + +class CommandRunner: + """Finite pipes; raw stderr is never included in a refusal or receipt.""" + + def __init__(self, environment): + self.environment = dict(environment) + self.environment.update({"HOME": "/proc/self/fd/-1", "DOCKER_CONFIG": "/proc/self/fd/-1"}) + self.pass_fds = () + + def bind_transport(self, transport): + self.environment = clean_environment(Path(FD_PREFIX + str(transport.fd))) + self.environment["HOME"] = FD_PREFIX + str(transport.children["home"]) + self.environment["DOCKER_CONFIG"] = FD_PREFIX + str(transport.children["docker"]) + self.pass_fds = (transport.fd, *transport.children.values()) + + def run(self, args, timeout=15, limit=core.MIB, stdin=None): + admitted() + authority = require_installed_authority() + binary = authority.open_command(args[0]) + try: + process = subprocess.Popen(args, executable=FD_PREFIX + str(binary), + stdin=stdin if stdin is not None else subprocess.DEVNULL, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=self.environment, + pass_fds=tuple(set(self.pass_fds + (binary,)))) + except BaseException: + os.close(binary) + raise + output = bytearray() + errors = 0 + deadline = time.monotonic() + timeout + try: + with selectors.DefaultSelector() as selector: + selector.register(process.stdout, selectors.EVENT_READ, "stdout") + selector.register(process.stderr, selectors.EVENT_READ, "stderr") + while selector.get_map(): + remaining = deadline - time.monotonic() + core.require(remaining > 0, "command-timeout") + for key, _event in selector.select(min(remaining, 0.25)): + chunk = os.read(key.fileobj.fileno(), 65536) + if not chunk: + selector.unregister(key.fileobj) + elif key.data == "stdout": + output.extend(chunk) + core.require(len(output) <= limit, "command-output-limit") + else: + errors += len(chunk) + core.require(errors <= 65536, "command-error-limit") + code = process.wait(timeout=max(0.01, deadline - time.monotonic())) + core.require(code == 0, "command-failed") + return bytes(output) + finally: + if process.poll() is None: + process.kill() + process.wait() + process.stdout.close() + process.stderr.close() + os.close(binary) + + +class SealKernel: + def create(self): + admitted() + require_installed_authority() + return os.memfd_create("public-browser-input", os.MFD_CLOEXEC | os.MFD_ALLOW_SEALING) + + def write(self, fd, raw): + with os.fdopen(os.dup(fd), "wb") as stream: + stream.write(raw) + + def add_seals(self, fd, flags): + fcntl.fcntl(fd, fcntl.F_ADD_SEALS, flags) + + def seals(self, fd): + return fcntl.fcntl(fd, fcntl.F_GET_SEALS) + + def rewind(self, fd): + os.lseek(fd, 0, os.SEEK_SET) + + def read(self, fd): + return os.read(fd, core.MIB) + + def close(self, fd): + os.close(fd) + + +class Seal: + def __init__(self, raw, kernel=None): + admitted() + core.require(type(raw) is bytes and len(raw) <= core.TAR_LIMIT, "seal-input-limit") + self.kernel = kernel or SealKernel() + self.fd = self.kernel.create() + try: + self.kernel.write(self.fd, raw) + self.flags = fcntl.F_SEAL_WRITE | fcntl.F_SEAL_GROW | fcntl.F_SEAL_SHRINK | fcntl.F_SEAL_SEAL + self.kernel.add_seals(self.fd, self.flags) + self.expected = core.digest(raw) + except BaseException: + self.kernel.close(self.fd) + raise + + def verified_hash(self): + core.require(self.kernel.seals(self.fd) == self.flags, "missing-seals") + self.kernel.rewind(self.fd) + observed = hashlib.sha256() + while True: + chunk = self.kernel.read(self.fd) + if not chunk: + break + observed.update(chunk) + self.kernel.rewind(self.fd) + core.require(observed.hexdigest() == self.expected, "sealed-buffer-drift") + return observed.hexdigest() + + def close(self): + self.kernel.close(self.fd) + + +def fetch_dependency(name, runner): + admitted() + core.require(name in ("playwright", "playwright-core"), "dependency-name") + url = "https://registry.npmjs.org/" + name + "/-/" + name + "-1.64.0.tgz" + # The child has an absolute host deadline and finite pipes. A per-read + # socket timeout alone would allow a slow response to renew its deadline. + code = ("import sys,urllib.request\n" + "class Stop(urllib.request.HTTPRedirectHandler):\n" + " def redirect_request(self,*a,**k): raise RuntimeError('redirect')\n" + "o=urllib.request.build_opener(urllib.request.ProxyHandler({}),Stop())\n" + "u=" + repr(url) + "\n" + "with o.open(urllib.request.Request(u,headers={'Accept-Encoding':'identity'}),timeout=30) as r:\n" + " if r.status!=200 or r.url!=u: raise RuntimeError('response')\n" + " b=r.read(16777217)\n" + " if len(b)>16777216: raise RuntimeError('size')\n" + " sys.stdout.buffer.write(b)\n") + return runner.run([PYTHON, "-I", "-S", "-B", "-c", code], timeout=30, limit=16 * core.MIB) + + +def clean_environment(root): + return {"PATH": "/usr/bin:/bin", "LANG": "C.UTF-8", "HOME": str(root / "home"), + "DOCKER_CONFIG": str(root / "docker"), "DOCKER_HOST": "unix:///var/run/docker.sock", + "XDG_RUNTIME_DIR": "/run/user/" + str(os.getuid()), + "DBUS_SESSION_BUS_ADDRESS": "unix:path=/run/user/" + str(os.getuid()) + "/bus"} + + +def unit_stub(modules): + packed = base64.b64encode(zlib.compress(json.dumps({k: base64.b64encode(v).decode("ascii") + for k, v in modules.items()}, sort_keys=True).encode())).decode("ascii") + return ("import base64,json,sys,types,zlib; " + "s={k:base64.b64decode(v) for k,v in json.loads(zlib.decompress(base64.b64decode('" + packed + "'))).items()}; " + "c=types.ModuleType('rust_preparation');sys.modules[c.__name__]=c;exec(compile(s[c.__name__],'','exec'),c.__dict__); " + "a=types.ModuleType('rust_preparation_authority');sys.modules[a.__name__]=a;exec(compile(s[a.__name__],'','exec'),a.__dict__); " + "p=types.ModuleType('rust_preparation_platform');sys.modules[p.__name__]=p;exec(compile(s[p.__name__],'','exec'),p.__dict__); " + "c._VERIFIED_MODULES=s;a.VERIFIED_SOURCE_SHA256=c.digest(s[a.__name__]);p.VERIFIED_SOURCE_SHA256=c.digest(s[p.__name__]);sys.exit(c.cli())") + + +def unit_argv(mode, spec, runtime, modules): + core.require(mode in ("worker", "cleanup"), "unit-mode") + environment = clean_environment(Path(spec.get("staging_path", "/unbound-source-fixture"))) + assignments = [name + "=" + value for name, value in sorted(environment.items())] + env_path = _HOST_AUTHORITY.command_path("env") if _HOST_AUTHORITY else ENV + python_path = _HOST_AUTHORITY.command_path("python") if _HOST_AUTHORITY else PYTHON + return [env_path, "-i", *assignments, python_path, "-I", "-S", "-B", "-c", unit_stub(modules), mode, encoded(spec), encoded(runtime)] + + +def stop_property(argv): + # ':' disables systemd environment substitution; %% protects specifiers. + quoted = ['"' + a.replace("\\", "\\\\").replace('"', '\\"').replace("%", "%%") + '"' for a in argv] + return "ExecStopPost=:" + " ".join(quoted) + + +def bus_value(runner, path, interface, property_name): + raw = runner.run([BUSCTL, "--user", BUS_JSON, "get-property", SERVICE, path, interface, property_name]) + value = core.strict_json(raw) + core.require(set(value) == {"type", "data"}, "bus-property-shape") + return value["data"] + + +def service_path(runner): + raw = runner.run([BUSCTL, "--user", BUS_JSON, "call", SERVICE, "/org/freedesktop/systemd1", + SERVICE + ".Manager", "GetUnit", "s", UNIT]) + value = core.strict_json(raw) + core.require(value["type"] == "o" and len(value["data"]) == 1, "unit-identity") + return value["data"][0] + + +def check_manager_identity(runner): + authority = require_installed_authority() + raw = runner.run([BUSCTL, "--user", BUS_JSON, "call", "org.freedesktop.DBus", "/org/freedesktop/DBus", + "org.freedesktop.DBus", "GetConnectionUnixProcessID", "s", SERVICE]) + observed = core.strict_json(raw) + core.require(observed["type"] == "u" and observed["data"] == [authority.record["processes"]["manager"]["pid"]], "manager-interface-causality") + + +def check_supervisor(runner, spec, runtime, modules): + path = service_path(runner) + core.require(bus_value(runner, path, UNIT_INTERFACE, "ActiveState") == "active", "unit-inactive") + expected = {"RuntimeMaxUSec": 600000000, "TimeoutStopUSec": 15000000, + "KillMode": "control-group", "Restart": "no", "MemoryMax": 1024 * core.MIB, + "TasksMax": 64, "UMask": 0o77, "CPUQuotaPerSecUSec": 1000000, "MainPID": os.getpid()} + for name, value in expected.items(): + core.require(bus_value(runner, path, INTERFACE, name) == value, "unit-limit-drift") + for property_name, mode in (("ExecStart", "worker"), ("ExecStopPost", "cleanup")): + observed = bus_value(runner, path, INTERFACE, property_name) + expected_argv = unit_argv(mode, spec, runtime, modules) + core.require(len(observed) == 1 and observed[0][0] == expected_argv[0] + and observed[0][1] == expected_argv, "unit-source-drift") + core.require(time.monotonic() < runtime["deadline"], "whole-attempt-timeout") + + +def check_stop_authority(runner, spec, runtime, modules): + path = service_path(runner) + core.require(bus_value(runner, path, INTERFACE, "ControlPID") == os.getpid(), "cleanup-control-pid") + core.require(bus_value(runner, path, UNIT_INTERFACE, "ActiveState") == "deactivating", "cleanup-unit-state") + observed = bus_value(runner, path, INTERFACE, "ExecStopPost") + expected_argv = unit_argv("cleanup", spec, runtime, modules) + core.require(len(observed) == 1 and observed[0][0] == expected_argv[0] + and observed[0][1] == expected_argv, "cleanup-source-authority") + + +def capacity(spec): + core.require(platform.machine() == "x86_64" and platform.system() == "Linux", "host-platform") + with open("/proc/meminfo", encoding="ascii") as source: + available = next(int(line.split()[1]) * 1024 for line in source if line.startswith("MemAvailable:")) + fs = os.statvfs(spec["docker_root"]) + core.require(available >= 4 * 1024 * core.MIB, "host-memory") + core.require(fs.f_bavail * fs.f_frsize >= 20 * 1024 * core.MIB, "host-disk") + core.require(os.getloadavg()[0] <= (os.cpu_count() or 1), "host-load") + + +class DockerConnection(http.client.HTTPConnection): + def connect(self): + admitted() + require_installed_authority() + self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.sock.settimeout(self.timeout) + self.sock.connect("/var/run/docker.sock") + + +def parent_descriptor(path): + target = Path(path) + core.require(target.is_absolute() and target.resolve() == target, "transport-canonical-path") + descriptor = os.open("/", os.O_RDONLY | os.O_DIRECTORY) + try: + for part in target.parts[1:-1]: + following = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=descriptor) + os.close(descriptor) + descriptor = following + return descriptor, target.name + except BaseException: + os.close(descriptor) + raise + + +class TransportDirectory: + """All child operations use the same held root, including after a rename.""" + + def __init__(self, spec, runtime): + self.spec = spec + self.parent, self.name = parent_descriptor(spec["staging_path"]) + self.fd = None + self.children = {} + try: + self.fd = os.open(self.name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=self.parent) + self.identity = os.fstat(self.fd) + core.require(stat.S_IMODE(self.identity.st_mode) == 0o700 + and self.identity.st_uid == spec["uid"] and self.identity.st_gid == spec["gid"], "transport-owner") + record = core.strict_json(self.read(OWNERSHIP_FILE, 4096)) + expected = {"nonce": runtime["nonce"], "device": self.identity.st_dev, "inode": self.identity.st_ino} + core.require(set(record) == set(expected) | {"children"} and all(record[k] == v for k, v in expected.items()), + "transport-record-binding") + self.child_bindings = record["children"] + core.require(set(self.child_bindings) == {"home", "docker"}, "transport-child-closure") + for name in ("home", "docker"): + child = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=self.fd) + self.children[name] = child + info = os.fstat(child) + core.require({"device": info.st_dev, "inode": info.st_ino} == self.child_bindings[name] + and (info.st_uid, info.st_gid) == (spec["uid"], spec["gid"]) + and stat.S_IMODE(info.st_mode) == 0o700, "transport-child-binding") + self.verify_namespace() + except BaseException: + self.close() + raise + + def close(self): + for child in self.children.values(): + os.close(child) + self.children.clear() + if self.fd is not None: + os.close(self.fd) + self.fd = None + if self.parent is not None: + os.close(self.parent) + self.parent = None + + def namespace_matches(self): + current = os.stat(self.name, dir_fd=self.parent, follow_symlinks=False) + return (current.st_dev, current.st_ino) == (self.identity.st_dev, self.identity.st_ino) + + def verify_namespace(self): + core.require(self.namespace_matches(), "transport-namespace-changed") + for name, binding in self.child_bindings.items(): + info = os.stat(name, dir_fd=self.fd, follow_symlinks=False) + core.require(stat.S_ISDIR(info.st_mode) and {"device": info.st_dev, "inode": info.st_ino} == binding, + "transport-child-swapped") + + def exists(self, name): + try: + os.stat(name, dir_fd=self.fd, follow_symlinks=False) + return True + except FileNotFoundError: + return False + + def read(self, name, limit, parent=None): + descriptor = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=self.fd if parent is None else parent) + try: + info = os.fstat(descriptor) + core.require(stat.S_ISREG(info.st_mode) and info.st_nlink == 1 + and info.st_uid == self.identity.st_uid and stat.S_IMODE(info.st_mode) == 0o600, + "transport-record") + with os.fdopen(os.dup(descriptor), "rb") as stream: + raw = stream.read(limit + 1) + core.require(len(raw) <= limit, "transport-record-limit") + return raw + finally: + os.close(descriptor) + + def write_pending(self, value): + core.require(not self.pending_operation(), "daemon-terminal-unknown") + self.append_record(PENDING_FILE, {**value, "state": "pending"}, 16384) + + def append_record(self, name, value, limit): + descriptor = os.open(name, os.O_WRONLY | os.O_APPEND | os.O_CREAT | os.O_NOFOLLOW, 0o600, dir_fd=self.fd) + with os.fdopen(descriptor, "wb") as file: + observed = os.fstat(file.fileno()) + core.require(stat.S_ISREG(observed.st_mode) and observed.st_nlink == 1 + and observed.st_uid == self.identity.st_uid and stat.S_IMODE(observed.st_mode) == 0o600, + "operation-result-type") + raw = (json.dumps(value, sort_keys=True) + "\n").encode() + core.require(observed.st_size + len(raw) <= limit, "operation-journal-limit") + file.write(raw) + file.flush() + os.fsync(file.fileno()) + os.fsync(self.fd) + + def pending_operation(self): + if not self.exists(PENDING_FILE): + return None + rows = [core.strict_json(line) for line in self.read(PENDING_FILE, 16384).splitlines()] + core.require(0 < len(rows) <= 16, "operation-journal-count") + pending = None + for row in rows: + core.require(set(row) == {"operation", "nonce", "state"}, "operation-journal-fields") + if row["state"] == "pending": + core.require(pending is None, "operation-journal-overlap") + pending = {k: row[k] for k in ("operation", "nonce")} + else: + core.require(row["state"] == "bound" and pending == {k: row[k] for k in ("operation", "nonce")}, + "operation-journal-terminal") + pending = None + return pending + + def bind_operation(self, operation, identity, nonce): + pending = self.pending_operation() + core.require(pending == {"operation": operation, "nonce": nonce}, "pending-operation-identity") + self.append_record(RESULTS_FILE, {"operation": operation, "nonce": nonce, "identity": identity}, 65536) + # Only typed, durably recorded caller validation makes this terminal. + # Append completion; do not unlink a mutable filename after validation. + self.append_record(PENDING_FILE, {"operation": operation, "nonce": nonce, "state": "bound"}, 16384) + + def write_context(self, value): + raw = json.dumps(value, sort_keys=True).encode() + core.require(len(raw) <= 65536, "transport-context-limit") + descriptor = os.open(CONTEXT_FILE, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=self.fd) + with os.fdopen(descriptor, "wb") as file: + file.write(raw) + file.flush() + os.fsync(file.fileno()) + os.fsync(self.fd) + + def results(self): + if not self.exists(RESULTS_FILE): + return [] + raw = self.read(RESULTS_FILE, 65536) + rows = [core.strict_json(line) for line in raw.splitlines()] + core.require(len(rows) <= 16, "operation-result-count") + return rows + + def retained_file(self, parent, name, limit, path): + raw = self.read(name, limit, parent) + return {"path": path, "bytes": len(raw), "sha256": core.digest(raw)} + + def retain_evidence(self, runtime): + # Read-only finalization. No unlink/rmdir of a mutable namespace entry: + # POSIX pathname deletion cannot condition on our held inode identity. + core.require(self.namespace_matches(), "transport-namespace-changed") + expected = {"home", "docker", OWNERSHIP_FILE, RESULTS_FILE, CONTEXT_FILE, PENDING_FILE} + core.require(set(os.listdir(self.fd)) == expected, "transport-extra-or-pending") + core.require(self.pending_operation() is None, "daemon-terminal-unknown") + children = self.children + self.verify_namespace() + try: + core.require(os.listdir(children["home"]) == [], "transport-home-extra") + core.require(os.listdir(children["docker"]) == [CONFIG_FILE], "transport-config-extra") + core.require(self.read(CONFIG_FILE, 4096, children["docker"]) == ANONYMOUS_CONFIG, "transport-config-changed") + rows = [self.retained_file(self.fd, name, limit, name) for name, limit in + ((OWNERSHIP_FILE, 4096), (RESULTS_FILE, 65536), (CONTEXT_FILE, 65536), (PENDING_FILE, 16384))] + rows.append(self.retained_file(children["docker"], CONFIG_FILE, 4096, "docker/config.json")) + for name, descriptor in children.items(): + observed = os.stat(name, dir_fd=self.fd, follow_symlinks=False) + opened = os.fstat(descriptor) + core.require((observed.st_dev, observed.st_ino) == (opened.st_dev, opened.st_ino), "transport-child-swapped") + core.require(self.namespace_matches() and set(os.listdir(self.fd)) == expected, "transport-namespace-changed") + core.require(sum(row["bytes"] for row in rows) <= core.EVIDENCE_LIMIT, "transport-evidence-limit") + retention = self.spec["retention"] + return {"state": core.RETAINED_EVIDENCE, "owner": core.OWNER, "attempt_sha256": core.digest(runtime["nonce"].encode()), + "device": self.identity.st_dev, "inode": self.identity.st_ino, + "uid": self.identity.st_uid, "gid": self.identity.st_gid, "mode": 0o700, + "files": rows, "bytes": sum(row["bytes"] for row in rows), + "sha256": core.digest(installed.canonical_bytes(rows)), "custody_due": retention["custody_due"], + "reconciliation_ref": retention["reconciliation_ref"]} + finally: + self.verify_namespace() + + +class LiveEffects: + def __init__(self, spec, runtime, modules, runner=None): + admitted() + self.spec, self.runtime, self.modules = spec, runtime, modules + self.root = Path(spec["staging_path"]) + self.runner = runner or CommandRunner(clean_environment(self.root)) + self.transport = None + self.ownership = core.Ownership(runtime["nonce"]) + + def supervisor_ready(self): + if self.transport is not None: + self.transport.verify_namespace() + check_manager_identity(self.runner) + check_supervisor(self.runner, self.spec, self.runtime, self.modules) + capacity(self.spec) + core.require(not self.uncertain_daemon_operation(), "daemon-terminal-unknown") + + def command(self, args, timeout=15, limit=core.MIB, stdin=None, mutation=None): + admitted() + if mutation is not None: + self.supervisor_ready() + self.require_transport().write_pending({"operation": mutation, "nonce": self.ownership.nonce}) + remaining = max(0.01, self.runtime["deadline"] - time.monotonic()) + result = self.runner.run([DOCKER, *args], min(timeout, remaining), limit, stdin) + return result + + def require_transport(self): + if self.transport is None: + self.transport = TransportDirectory(self.spec, self.runtime) + if isinstance(self.runner, CommandRunner): + self.runner.bind_transport(self.transport) + return self.transport + + def complete_operation(self, operation, identity): + if operation == "commit": + core.require(set(identity) == {"image_id", "config_sha256", "rootfs_layers"}, "commit-binding-fields") + core.require(re.fullmatch(core.IMAGE_ID, identity["image_id"]) is not None + and identity["config_sha256"] == identity["image_id"][7:], "commit-binding-id") + record = self.image_inspect(identity["image_id"]) + core.require(record["Id"] == identity["image_id"] + and record["RootFS"]["Layers"] == identity["rootfs_layers"], "commit-binding-readback") + self.require_transport().bind_operation(operation, identity, self.ownership.nonce) + + def discover(self, owner, role): + args = ["container", "ls", "--all", "--no-trunc", "--filter", "name=^/" + owner.name(role) + "$"] + for name, value in owner.labels(role).items(): + args += ["--filter", "label=" + name + "=" + value] + return self.command([*args, "--format", "{{.ID}}"]).decode("ascii").split() + + def name_exists(self, name): + raw = self.command(["container", "ls", "--all", "--filter", "name=^/" + name + "$", "--format", "{{.Names}}"]) + return bool(raw.strip()) + + def inspect(self, cid): + core.require(re.fullmatch(core.HEX_64, cid) is not None, "inspect-id") + record = core.strict_json(self.command(["container", "inspect", cid]))[0] + try: + created = dt.datetime.fromisoformat(record["Created"].replace("Z", "+00:00")) + start = dt.datetime.fromisoformat(self.runtime["started_at"]) + within = created.tzinfo is not None and 0 <= (created - start).total_seconds() <= RUNTIME_LIMIT + except (ValueError, TypeError, KeyError): + within = False + record["WithinPreparationWindow"] = within + return record + + def create(self, owner, role, image): + args = ["create", "--pull=never", "--platform=linux/amd64", "--name", owner.name(role), + "--network=none", "--cpus=1", "--memory=512m", "--pids-limit=64", "--init", + "--cap-drop=ALL", "--security-opt=no-new-privileges:true", "--entrypoint", ""] + if role == "verify": + args += ["--read-only"] + for name, value in owner.labels(role).items(): + args += ["--label", name + "=" + value] + raw = self.command([*args, image, "/bin/false"], mutation="create-" + role) + return raw.decode("ascii").strip() + + def remove(self, cid): + # Independent cleanup has its own deadline; an unresolved journal does + # not prevent removal of an independently verified owned container. + core.require(re.fullmatch(core.HEX_64, cid) is not None, "remove-id") + self.runner.run([DOCKER, "rm", "--force", cid], timeout=5) + + def uncertain_daemon_operation(self): + if self.transport is None and not self.root.exists(): + return False + return self.require_transport().pending_operation() is not None + + def seal(self, raw): + sealed = Seal(raw) + try: + self.seal_proof = {"sha256": sealed.verified_hash(), "flags": sealed.flags} + return sealed + except BaseException: + sealed.close() + raise + + def foreign_snapshot(self): + raw = self.command(["container", "ls", "--all", "--no-trunc", "--format", "{{.ID}}\t{{.Names}}"]) + result = {} + for line in raw.decode("utf-8").splitlines(): + cid, name = line.split("\t") + core.require(re.fullmatch(core.HEX_64, cid) is not None and cid not in result, "foreign-snapshot") + result[cid] = name + return result + + def require_safe_prefixes(self, cid): + for path in ("/packet", "/site", "/runtime"): + connection = DockerConnection("localhost", timeout=5) + try: + connection.request("HEAD", "/containers/" + cid + "/archive?path=" + path) + response = connection.getresponse() + core.require(response.status == 404, "base-prefix-present-or-unknown") + # A disappeared container must not turn an unknown prefix into + # a successful ENOENT assumption. + core.inspect_allocation(self.inspect(cid), self.ownership, "assemble", core.BASE_ID) + finally: + connection.close() + + def copy_sealed(self, cid, sealed): + sealed.verified_hash() + self.command(["cp", "-a", "-", cid + ":/"], timeout=60, stdin=sealed.fd, mutation="copy") + + def commit(self, cid): + raw = self.command(["commit", "--pause=false", cid], timeout=60, mutation="commit") + return raw.decode("ascii").strip() + + def image_inspect(self, image): + core.require(re.fullmatch(core.IMAGE_ID, image) is not None, "inspect-image-id") + return core.strict_json(self.command(["image", "inspect", image]))[0] + + def readback(self, cid): + result = [] + for root in ("packet", "site", "runtime/node_modules"): + raw = self.command(["cp", cid + ":/" + root, "-"], timeout=60, limit=core.TAR_LIMIT) + validate_readback_metadata(raw) + entries = core.tar_entries(raw) + for entry in entries: + name = "runtime/" + entry.path if root.startswith("runtime/") else entry.path + result.append(core.Entry(name, entry.data, entry.mode)) + return tuple(sorted(result, key=lambda entry: entry.path)) + + def acquire_base(self): + info = core.strict_json(self.command(["info", "--format", "{{json .}}"])) + core.require(not any(info.get(k) for k in ("HTTPProxy", "HTTPSProxy", "NoProxy")), "daemon-proxy") + core.require(info["DockerRootDir"] == self.spec["docker_root"], "daemon-filesystem") + self.command(["pull", "--platform=linux/amd64", core.BASE_IMAGE], timeout=240, + limit=4 * core.MIB, mutation="pull") + record = self.image_inspect(core.BASE_ID) + core.require(core.BASE_IMAGE in record["RepoDigests"] and record["Architecture"] == "amd64" + and record["Os"] == "linux" and record["Size"] <= 8 * 1024 * core.MIB, "base-image") + core.require(not record["Config"].get("Volumes") and not record["Config"].get("OnBuild"), "base-hooks") + self.complete_operation("pull", {"base_image_id": core.BASE_ID, "rootfs_layers": record["RootFS"]["Layers"]}) + return record["RootFS"]["Layers"] + + +def validate_readback_metadata(raw): + import tarfile + import io + core.require(len(raw) <= core.TAR_LIMIT, "readback-limit") + with tarfile.open(fileobj=io.BytesIO(raw), mode="r:") as archive: + for member in archive: + core.require(member.uid == 1000 and member.gid == 1000, "readback-owner") + modes = (0o555,) if member.isdir() else (0o444, 0o555) + core.require(member.mode in modes, "readback-mode") + + +def packet_capture(spec): + entries = [] + for name, binding in spec["packet_bindings"].items(): + core.safe_name(name) + raw = core.held_regular(binding["path"], binding) + entries.append(core.Entry(name, raw)) + return tuple(entries) + + +def runtime_identity(spec): + require_installed_authority() + core.require((os.getuid(), os.getgid()) == (spec["uid"], spec["gid"]), "effective-executor") + core.require(core.digest(Path("/proc/self/exe").read_bytes()) == spec["python_sha256"], "effective-interpreter") + + +def expected_receipt(spec, runtime, modules): + authority = require_installed_authority() + source_hashes = {name: core.digest(raw) for name, raw in modules.items()} + return {"unit": UNIT, "attempt_sha256": core.digest(runtime["nonce"].encode()), + "source_bundle_sha256": core.digest(installed.canonical_bytes(source_hashes)), + "authority_sha256": authority.digest, + "worker_command_sha256": core.digest(installed.canonical_bytes(unit_argv("worker", spec, runtime, modules))), + "stop_command_sha256": core.digest(installed.canonical_bytes(unit_argv("cleanup", spec, runtime, modules))), + "custody_due": spec["retention"]["custody_due"], "reconciliation_ref": spec["retention"]["reconciliation_ref"], + "transport_uid": spec["uid"], "transport_gid": spec["gid"], + "seal_flags": fcntl.F_SEAL_WRITE | fcntl.F_SEAL_GROW | fcntl.F_SEAL_SHRINK | fcntl.F_SEAL_SEAL} + + +def stage_receipt(role, runner, spec, runtime, modules): + expected = expected_receipt(spec, runtime, modules) + path = service_path(runner) + invocation = bus_value(runner, path, UNIT_INTERFACE, "InvocationID") + core.require(isinstance(invocation, list) and len(invocation) == 16 and any(invocation), "unit-invocation-id") + return {"schema": "durable-workflow.preparation-stage/v1", "role": role, + "unit": UNIT, "attempt_sha256": expected["attempt_sha256"], + "source_bundle_sha256": expected["source_bundle_sha256"], "authority_sha256": expected["authority_sha256"], + "pid": os.getpid(), "invocation_id": bytes(invocation).hex(), + "command_sha256": expected["worker_command_sha256" if role == "worker" else "stop_command_sha256"]} + + +def entry_inventory_sha256(entries): + rows = [{"path": e.path, "bytes": len(e.data), "sha256": core.digest(e.data), "mode": e.mode} + for e in sorted(entries, key=lambda entry: entry.path)] + return core.digest(installed.canonical_bytes(rows)) + + +def run_worker(spec, runtime, modules): + runtime_identity(spec) + effects = LiveEffects(spec, runtime, modules) + effects.supervisor_ready() + create_transport(spec, runtime["nonce"]) + effects.require_transport() + stage = stage_receipt("worker", effects.runner, spec, runtime, modules) + foreign_before = effects.foreign_snapshot() + effects.require_transport().write_context({"foreign_before": foreign_before, "worker_pid": os.getpid(), + "invocation_id": stage["invocation_id"]}) + packet = packet_capture(spec) + inventory, lock = core.packet_from_entries(packet) + binding = spec["artifact_handoff"] + artifact = core.artifact_from_zip(core.held_regular(binding["path"], binding), inventory, + dt.datetime.now(dt.timezone.utc)) + core.require(set(lock["packages"]) == {"", "node_modules/playwright", "node_modules/playwright-core"}, "lock-closure") + dependencies = () + dependency_receipts = {} + for name in ("playwright", "playwright-core"): + effects.supervisor_ready() + package_raw = fetch_dependency(name, effects.runner) + dependencies += core.dependency_entries(package_raw, name, lock) + dependency_receipts[name] = {"version": "1.64.0", "integrity": lock["packages"]["node_modules/" + name]["integrity"], + "sha256": core.digest(package_raw)} + raw, expected = core.public_tar(packet, artifact, dependencies) + layers = effects.acquire_base() + image = core.assemble(effects, effects.ownership, raw, expected, layers) + image_record = effects.image_inspect(image) + core.require(core.verify_image(image_record, layers) == image, "final-image-readback") + stage.update({"outcome": "prepared-cleanup-pending", "subject": core.SUBJECT, + "node_packet_sha256": core.PACKET_SHA, "base_image_id": core.BASE_ID, + "base_layers": layers, "dependencies": dependency_receipts, "seal": effects.seal_proof, + "image": {"id": image, "config_sha256": image[7:], "layers": image_record["RootFS"]["Layers"]}, + "readback_sha256": entry_inventory_sha256(expected), "readback_verified": True, + "input_tar_sha256": core.digest(raw), "files": len(expected), + "foreign_before_sha256": core.digest(installed.canonical_bytes(foreign_before))}) + print(json.dumps(stage, sort_keys=True)) + return 0 + + +def create_transport(spec, nonce): + admitted() + root = Path(spec["staging_path"]) + home = Path(pwd.getpwuid(os.getuid()).pw_dir) + core.require(not root.is_relative_to(home), "transport-location") + core.require((os.getuid(), os.getgid()) == (spec["uid"], spec["gid"]), "executor-owner") + parent, leaf = parent_descriptor(root) + descriptor, children = None, {} + try: + os.mkdir(leaf, mode=0o700, dir_fd=parent) + descriptor = os.open(leaf, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=parent) + info = os.fstat(descriptor) + core.require((info.st_uid, info.st_gid) == (spec["uid"], spec["gid"]) + and stat.S_IMODE(info.st_mode) == 0o700 and os.listdir(descriptor) == [], "created-transport-identity") + def verify(): + current = os.stat(leaf, dir_fd=parent, follow_symlinks=False) + core.require((current.st_dev, current.st_ino) == (info.st_dev, info.st_ino), "transport-namespace-changed") + verify() + bindings = {} + for name in ("home", "docker"): + verify() + os.mkdir(name, mode=0o700, dir_fd=descriptor) + child = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=descriptor) + children[name] = child + observed = os.fstat(child) + core.require(os.listdir(child) == [] and (observed.st_uid, observed.st_gid) == (spec["uid"], spec["gid"]) + and stat.S_IMODE(observed.st_mode) == 0o700, "created-child-identity") + bindings[name] = {"device": observed.st_dev, "inode": observed.st_ino} + owner_record = {"nonce": nonce, "device": info.st_dev, "inode": info.st_ino, "children": bindings} + for directory, name, raw in ((children["docker"], CONFIG_FILE, ANONYMOUS_CONFIG), + (descriptor, OWNERSHIP_FILE, json.dumps(owner_record).encode())): + verify() + for child_name, binding in bindings.items(): + current = os.stat(child_name, dir_fd=descriptor, follow_symlinks=False) + core.require({"device": current.st_dev, "inode": current.st_ino} == binding, "transport-child-swapped") + file_fd = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=directory) + with os.fdopen(file_fd, "wb") as file: + file.write(raw) + file.flush() + os.fsync(file.fileno()) + os.fsync(directory) + verify() + os.fsync(descriptor) + return owner_record + finally: + for child in children.values(): + os.close(child) + if descriptor is not None: + os.close(descriptor) + os.close(parent) + + +def retain_transport(spec, runtime): + admitted() + directory = TransportDirectory(spec, runtime) + try: + return directory.retain_evidence(runtime) + finally: + directory.close() + + +def launch(spec, modules): + admitted() + nonce = secrets.token_hex(32) + runtime = {"nonce": nonce, "deadline": time.monotonic() + RUNTIME_LIMIT, + "started_at": dt.datetime.now(dt.timezone.utc).isoformat()} + runner = CommandRunner(clean_environment(Path(spec["staging_path"]))) + check_manager_identity(runner) + source = Path(PYTHON).read_bytes() + core.require(core.digest(source) == spec["python_sha256"], "interpreter-identity") + core.require(core.digest(Path(ENV).read_bytes()) == spec["env_sha256"], "environment-launcher-identity") + core.require(not Path(spec["staging_path"]).exists() and not Path(spec["staging_path"]).is_symlink(), "transport-preexisting") + worker = unit_argv("worker", spec, runtime, modules) + stop = unit_argv("cleanup", spec, runtime, modules) + args = [SYSTEMD_RUN, "--user", "--wait", "--pipe", "--quiet", "--unit=" + UNIT, "--expand-environment=no", "--description=bounded-public-input-preparation", + "--property=Type=exec", "--property=RuntimeMaxSec=600", "--property=TimeoutStopSec=15", + "--property=KillMode=control-group", "--property=Restart=no", "--property=CPUQuota=100%", + "--property=MemoryMax=1G", "--property=TasksMax=64", "--property=UMask=0077", + "--property=" + stop_property(stop), *worker] + # No pull/create occurs in this launcher. The worker reads back the installed + # independent stop command and effective bounds before its first operation. + # This observer is outside worker/stop. Missing/failed wait or collector loss + # produces no acceptance; systemd still owns independent bounded cleanup. + raw = runner.run(args, timeout=RUNTIME_LIMIT + 15, limit=core.MIB) + receipt = core.collect_terminal(raw, expected_receipt(spec, runtime, modules), 0) + print(json.dumps(receipt, sort_keys=True)) + return 0 + + +def entrypoint(argv, modules, admitted_hash): + core.require(argv and argv[0] in ("launch", "worker", "cleanup"), "controller-mode") + core.require(len(argv) == (2 if argv[0] == "launch" else 3), "controller-arguments") + spec = decoded(argv[1]) + core.require(core.digest(json.dumps(spec, sort_keys=True, separators=(",", ":")).encode()) == admitted_hash, "unadmitted-invocation") + install_authority(spec) + if argv[0] != "cleanup": + admission(spec, dt.datetime.now(dt.timezone.utc)) + if argv[0] == "launch": + return launch(spec, modules) + runtime = decoded(argv[2]) + core.Ownership(runtime["nonce"]) + if argv[0] == "worker": + return run_worker(spec, runtime, modules) + runner = CommandRunner(clean_environment(Path(spec["staging_path"]))) + runtime_identity(spec) + check_manager_identity(runner) + check_stop_authority(runner, spec, runtime, modules) + original_runtime = runtime + runtime = {**runtime, "deadline": time.monotonic() + 12} + effects = LiveEffects(spec, runtime, modules) + stage = stage_receipt("stop", runner, spec, original_runtime, modules) + path = service_path(runner) + worker_pid = bus_value(runner, path, INTERFACE, "ExecMainPID") + worker_code = bus_value(runner, path, INTERFACE, "ExecMainCode") + worker_status = bus_value(runner, path, INTERFACE, "ExecMainStatus") + problems = core.cleanup(effects, effects.ownership) + problems = list(problems) + operations, context, survivors = [], {}, [] + pending, foreign_preserved, transport = True, False, None + try: + directory = effects.require_transport() + operations = directory.results() + context = core.strict_json(directory.read(CONTEXT_FILE, 65536)) + core.require(context["worker_pid"] == worker_pid and context["invocation_id"] == stage["invocation_id"], + "terminal-context-generation") + pending = effects.uncertain_daemon_operation() + for role in core.ROLES: + survivors += effects.discover(effects.ownership, role) + foreign_after = effects.foreign_snapshot() + foreign_preserved = all(foreign_after.get(cid) == name for cid, name in context["foreign_before"].items()) + if not problems and not pending and not survivors and foreign_preserved: + transport = retain_transport(spec, runtime) + except Exception: + problems.append("terminal-cleanup-evidence-incomplete") + clean = not problems and not pending and not survivors and foreign_preserved and transport is not None + attempt = stage["attempt_sha256"] + sanitized = [{"operation": row["operation"], "attempt_sha256": core.digest(row["nonce"].encode()), + "identity": row["identity"]} for row in operations] + core.require(all(row["attempt_sha256"] == attempt for row in sanitized), "terminal-operation-nonce") + stage.update({"outcome": "complete" if clean else "unresolved", "worker_pid": worker_pid, + "worker_exit_code": worker_code, "worker_exit_status": worker_status, "cleanup_exit_status": 0 if clean else 1, + "operation_results": sanitized, "pending": pending, "owned_survivors": survivors, + "foreign_preserved": foreign_preserved, + "foreign_before_sha256": core.digest(installed.canonical_bytes(context.get("foreign_before", {}))), + "transport": transport, "problems": problems}) + print(json.dumps(stage, sort_keys=True)) + return 0 if clean else 1 diff --git a/security/browser-headers/test_rust_preparation.py b/security/browser-headers/test_rust_preparation.py new file mode 100644 index 00000000000..fd7e8842bca --- /dev/null +++ b/security/browser-headers/test_rust_preparation.py @@ -0,0 +1,1205 @@ +"""Inert controls: synthetic archives/records and fake daemon/supervisor only. + +Unsafe baselines below are explicit test countermodels, not historical runs. +No seal, socket, network, service, image or container is created by this suite. +""" + +import copy +import base64 +import dataclasses +import datetime as dt +import gzip +import hashlib +import io +import json +import os +import shutil +from pathlib import Path +import stat +import tarfile +import tempfile +import unittest +import zipfile +import time +from types import SimpleNamespace +from unittest import mock + +import rust_preparation as prep +import rust_preparation_platform as native +import rust_preparation_authority as authority + + +# Exact effect functions captured from aad23d45, retained only as unsafe +# predecessor controls. These run solely against disposable synthetic trees. +PREDECESSOR_CREATE = 'def create_transport(spec, nonce):\n admitted()\n root = Path(spec["staging_path"])\n home = Path(pwd.getpwuid(os.getuid()).pw_dir)\n core.require(not root.is_relative_to(home) and root.resolve() == root, "transport-location")\n core.require((os.getuid(), os.getgid()) == (spec["uid"], spec["gid"]), "executor-owner")\n root.mkdir(mode=0o700)\n for name in ("home", "docker"):\n (root / name).mkdir(mode=0o700)\n observed = root.stat()\n owner_record = {"nonce": nonce, "device": observed.st_dev, "inode": observed.st_ino}\n for name, raw in (("docker/config.json", b"{}"), (OWNERSHIP_FILE, json.dumps(owner_record).encode())):\n with os.fdopen(os.open(root / name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600), "wb") as file:\n file.write(raw)\n return owner_record\n' +PREDECESSOR_CREATE_SHA256 = '571789f1853a14dd2eafbf98cfbadc682495f41b007b79d70d1d0a3a41acfde3' +PREDECESSOR_REMOVE = 'def remove_contents(self):\n core.require(self.namespace_matches(), "transport-namespace-changed")\n expected = {"home", "docker", OWNERSHIP_FILE}\n if self.exists(RESULTS_FILE):\n expected.add(RESULTS_FILE)\n if self.exists(CONTEXT_FILE):\n expected.add(CONTEXT_FILE)\n core.require(set(os.listdir(self.fd)) == expected, "transport-extra-or-pending")\n children = {}\n try:\n for name in ("home", "docker"):\n child = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=self.fd)\n children[name] = child\n core.require(os.fstat(child).st_uid == self.identity.st_uid, "transport-child-owner")\n core.require(os.listdir(children["home"]) == [], "transport-home-extra")\n core.require(os.listdir(children["docker"]) == ["config.json"], "transport-config-extra")\n os.unlink("config.json", dir_fd=children["docker"])\n for name in (OWNERSHIP_FILE, RESULTS_FILE, CONTEXT_FILE):\n if self.exists(name):\n os.unlink(name, dir_fd=self.fd)\n for name in ("home", "docker"):\n os.rmdir(name, dir_fd=self.fd)\n core.require(self.namespace_matches(), "transport-namespace-changed")\n os.rmdir(self.name, dir_fd=self.parent)\n core.require(os.fstat(self.fd).st_nlink == 0, "bound-transport-survivor")\n finally:\n for child in children.values():\n os.close(child)\n' +PREDECESSOR_REMOVE_SHA256 = '6a65cb51d6f2484645465c6d1409eaeefda15b88c72970a647e050360dc1f8c9' + + +class HardStop(BaseException): + pass + + +class FakeSeal: + def __init__(self, raw, corrupt=False): + self.raw = raw + self.corrupt = corrupt + self.closed = False + + def verified_hash(self): + return "bad" if self.corrupt else prep.digest(self.raw) + + def close(self): + self.closed = True + + +class FakeEffects: + def __init__(self, expected=()): + self.containers = {} + self.images = {} + self.expected = expected + self.guardian = False + self.fresh_sample = False + self.alive = True + self.calls = [] + self.inspected = [] + self.fault = None + self.bad_return = None + self.uncertain = False + self.bad_record = None + self.bad_image = False + self.bad_readback = False + self.bad_prefix = False + self.bad_seal = False + self.next_id = 1 + self.last_seal = None + self.pending_operation = None + self.bound_operations = [] + + def edge(self, name): + self.calls.append(name) + if self.fault == name: + self.fault = None + raise HardStop(name) + + def register(self): + self.edge("register-before") + self.guardian = True + self.fresh_sample = True + self.edge("register-after") + + def supervisor_ready(self): + prep.require(self.guardian and self.alive and self.fresh_sample, "supervisor-unready") + prep.require(not self.uncertain, "daemon-terminal-unknown") + + def name_exists(self, name): + return any(r["Name"] == "/" + name for r in self.containers.values()) + + def discover(self, owner, role): + return [cid for cid, record in self.containers.items() + if record["Name"] == "/" + owner.name(role) + and all(record["Config"]["Labels"].get(k) == v for k, v in owner.labels(role).items())] + + def inspect(self, cid): + self.inspected.append(cid) + self.edge("inspect-" + self.containers[cid]["Config"]["Labels"][prep.LABEL_ROLE]) + return copy.deepcopy(self.containers[cid]) + + def create(self, owner, role, image): + self.supervisor_ready() + self.pending_operation = "create-" + role + self.edge("create-" + role + "-before") + cid = f"{self.next_id:064x}" + self.next_id += 1 + record = allocation(cid, owner, role, image) + if self.bad_record: + self.bad_record(record) + self.containers[cid] = record + self.edge("create-" + role + "-after") + return cid if self.bad_return is None else self.bad_return + + def remove(self, cid): + self.edge("remove-before") + del self.containers[cid] + self.edge("remove-after") + + def seal(self, raw): + self.last_seal = FakeSeal(raw, self.bad_seal) + return self.last_seal + + def require_safe_prefixes(self, _cid): + prep.require(not self.bad_prefix, "base-prefix-symlink") + + def copy_sealed(self, _cid, _sealed): + self.pending_operation = "copy" + self.edge("copy-before") + self.edge("copy-after") + + def commit(self, _cid): + self.pending_operation = "commit" + self.edge("commit-before") + image_id = "sha256:" + "c" * 64 + record = {"Id": image_id, "RootFS": {"Layers": ["base", "public"]}, + "Config": {"Cmd": ["/bin/false"], "Entrypoint": None}} + if self.bad_image: + record["RootFS"]["Layers"][0] = "other" + self.images[image_id] = record + self.edge("commit-after") + return image_id + + def image_inspect(self, image_id): + self.edge("image-inspect") + return copy.deepcopy(self.images[image_id]) + + def readback(self, _cid): + self.edge("readback-before") + self.edge("readback-after") + return () if self.bad_readback else self.expected + + def uncertain_daemon_operation(self): + return self.uncertain or self.pending_operation is not None + + def complete_operation(self, operation, identity): + prep.require(self.pending_operation == operation, "operation-binding") + self.edge("bind-" + operation + "-before") + self.bound_operations.append({"operation": operation, **identity}) + self.edge("bind-" + operation + "-recorded") + self.pending_operation = None + + +def allocation(cid, owner, role, image=prep.BASE_ID): + return {"Id": cid, "Name": "/" + owner.name(role), "Image": image, + "WithinPreparationWindow": True, + "Config": {"Labels": owner.labels(role), "Cmd": ["/bin/false"], "Entrypoint": None}, + "State": {"Status": "created", "Running": False}, "Mounts": [], + "HostConfig": {"NetworkMode": "none", "ReadonlyRootfs": role == "verify", + "PidMode": "", "IpcMode": "private", "UTSMode": "", "PublishAllPorts": False, + "NanoCpus": 1000000000, "Memory": 512 * prep.MIB, + "PidsLimit": 64, "Init": True, "Privileged": False, + "CapDrop": ["ALL"], "SecurityOpt": ["no-new-privileges:true"]}} + + +def archive(members): + output = io.BytesIO() + with tarfile.open(fileobj=output, mode="w") as value: + for name, data, kind in members: + info = tarfile.TarInfo(name) + info.type = kind + info.mode = 0o644 + info.size = len(data) if kind == tarfile.REGTYPE else 0 + if kind in (tarfile.SYMTYPE, tarfile.LNKTYPE): + info.linkname = "/outside" + value.addfile(info, io.BytesIO(data) if info.size else None) + return output.getvalue() + + +def transport_fixture(parent): + root = Path(parent) / "transport" + root.mkdir(mode=0o700) + (root / "home").mkdir(mode=0o700) + (root / "docker").mkdir(mode=0o700) + (root / "docker/config.json").write_bytes(native.ANONYMOUS_CONFIG) + (root / "docker/config.json").chmod(0o600) + observed = root.stat() + nonce = "a" * 64 + (root / native.OWNERSHIP_FILE).write_text(json.dumps({"nonce": nonce, "device": observed.st_dev, "inode": observed.st_ino, "children": {name: {"device": (root / name).stat().st_dev, + "inode": (root / name).stat().st_ino} for name in ("home", "docker")}})) + (root / native.OWNERSHIP_FILE).chmod(0o600) + return ({"staging_path": str(root), "uid": os.getuid(), "gid": os.getgid(), + "retention": {"owner": prep.OWNER, "custody_due": "2026-11-06T00:00:00+00:00", "reconciliation_ref": "synthetic"}}, + {"nonce": nonce, "deadline": time.monotonic() + 30, + "started_at": dt.datetime.now(dt.timezone.utc).isoformat()}) + + +class InputControls(unittest.TestCase): + def test_current_packet_authenticates_without_loading_it(self): + root = Path(__file__).parent + entries = tuple(prep.Entry(n, (root / n).read_bytes()) + for n in (*prep.PACKET_MEMBERS, "rust-local-browser-packet.json")) + inventory, lock = prep.packet_from_entries(entries) + self.assertEqual(len(inventory["files"]), 484) + self.assertEqual(len(lock["packages"]), 3) + mutated = list(entries) + mutated[0] = dataclasses.replace(mutated[0], data=b"changed") + changed_entries = tuple(mutated) + with self.assertRaises(prep.Refusal): + prep.packet_from_entries(changed_entries) + shadow_entries = entries + (prep.Entry("node_modules/shadow", b"x"),) + with self.assertRaises(prep.Refusal): + prep.packet_from_entries(shadow_entries) + + def test_archive_refuses_traversal_links_devices_duplicates_and_limits(self): + good = archive([("package/file", b"ok", tarfile.REGTYPE)]) + self.assertEqual(prep.tar_entries(good, "package")[0].data, b"ok") + bad = [ + [("../outside", b"x", tarfile.REGTYPE)], + [("/absolute", b"x", tarfile.REGTYPE)], + [("package/link", b"", tarfile.SYMTYPE)], + [("package/link", b"", tarfile.LNKTYPE)], + [("package/device", b"", tarfile.CHRTYPE)], + [("package/f", b"a", tarfile.REGTYPE), ("package/f", b"b", tarfile.REGTYPE)], + [("package/f", b"a", tarfile.REGTYPE), ("package/f/child", b"b", tarfile.REGTYPE)], + ] + for members in bad: + raw = archive(members) + with self.subTest(members=members), self.assertRaises(prep.Refusal): + prep.tar_entries(raw, "package") + with mock.patch.object(prep, "TREE_LIMIT", 1), self.assertRaises(prep.Refusal): + prep.tar_entries(good, "package") + + def test_json_duplicate_and_nonfinite_refusals(self): + for raw in (b'{"a":1,"a":2}', b'{"a":NaN}'): + with self.assertRaises(prep.Refusal): + prep.strict_json(raw) + + def test_handoff_owner_inode_mode_bytes_and_symlink_controls(self): + with tempfile.TemporaryDirectory(prefix="dw-source-fixture-") as directory: + path = Path(directory) / "synthetic.zip" + path.write_bytes(b"synthetic source fixture") + s = path.stat() + binding = {"uid": s.st_uid, "gid": s.st_gid, "mode": stat.S_IMODE(s.st_mode), + "parent_mode": stat.S_IMODE(path.parent.stat().st_mode), + "device": s.st_dev, "inode": s.st_ino, "bytes": s.st_size, + "sha256": prep.digest(path.read_bytes())} + self.assertEqual(prep.held_regular(str(path), binding), path.read_bytes()) + for key in ("uid", "gid", "mode", "parent_mode", "device", "inode", "bytes"): + wrong = {**binding, key: binding[key] + 1} + with self.subTest(key=key), self.assertRaises(prep.Refusal): + prep.held_regular(str(path), wrong) + path.write_bytes(b"x" * s.st_size) + with self.assertRaises(prep.Refusal): + prep.held_regular(str(path), binding) + link = Path(directory) / "alias" + link.symlink_to(path) + with self.assertRaises(OSError): + prep.held_regular(str(link), binding) + + def test_canonical_tar_has_only_public_roots_and_normalized_modes(self): + raw, entries = prep.public_tar((prep.Entry("source.cjs", b"s"),), + (prep.Entry("index.html", b"h"),), + (prep.Entry("runtime/node_modules/p/file", b"p"),)) + decoded = prep.tar_entries(raw) + self.assertEqual({e.path: e.data for e in decoded}, {e.path: e.data for e in entries}) + with self.assertRaises(dataclasses.FrozenInstanceError): + entries[0].data = b"changed" + + def test_default_cli_guard_precedes_any_effect(self): + with mock.patch("builtins.print") as output: + self.assertEqual(prep.main(), 78) + output.assert_called_once() + self.assertIsNone(prep.OPERATION_ADMISSION) + + def test_dependency_archives_authenticate_the_complete_two_package_graph(self): + name = "playwright" + package = {"name": name, "version": "1.64.0", "dependencies": {"playwright-core": "1.64.0"}} + def compressed(value): + return gzip.compress(archive([("package/package.json", json.dumps(value).encode(), tarfile.REGTYPE)])) + raw = compressed(package) + record = {"resolved": "https://registry.npmjs.org/playwright/-/playwright-1.64.0.tgz", "version": "1.64.0", + "integrity": "sha512-" + base64.b64encode(hashlib.sha512(raw).digest()).decode()} + lock = {"packages": {"node_modules/playwright": record}} + self.assertEqual(len(prep.dependency_entries(raw, name, lock)), 1) + with self.assertRaises(prep.Refusal): + prep.dependency_entries(raw + b"tampered", name, lock) + bad = compressed({**package, "optionalDependencies": {"unexpected": "1.0.0"}}) + changed = {"packages": {"node_modules/playwright": {**record, + "integrity": "sha512-" + base64.b64encode(hashlib.sha512(bad).digest()).decode()}}} + with self.assertRaises(prep.Refusal): + prep.dependency_entries(bad, name, changed) + with mock.patch.object(prep, "TAR_LIMIT", 2), self.assertRaises(prep.Refusal): + prep.dependency_entries(raw, name, lock) + + def test_synthetic_zip_digest_inventory_build_and_expiry(self): + build = {"version": "synthetic"} + members = [("__build.json", json.dumps(build).encode(), tarfile.REGTYPE)] + members += [(f"file-{i}", b"synthetic", tarfile.REGTYPE) for i in range(483)] + tar = archive(members) + output = io.BytesIO() + with zipfile.ZipFile(output, "w", zipfile.ZIP_DEFLATED) as value: + value.writestr("artifact.tar", tar) + raw = output.getvalue() + inventory = {"build_identity": build, "files": [{"path": n, "bytes": len(b), "sha256": prep.digest(b)} + for n, b, _kind in members]} + now = dt.datetime(2026, 11, 5, tzinfo=dt.timezone.utc) + with mock.patch.object(prep, "ZIP_BYTES", len(raw)), mock.patch.object(prep, "ZIP_SHA", prep.digest(raw)): + self.assertEqual(len(prep.artifact_from_zip(raw, inventory, now)), 484) + bad = copy.deepcopy(inventory) + bad["files"][0]["sha256"] = "f" * 64 + with self.assertRaises(prep.Refusal): + prep.artifact_from_zip(raw, bad, now) + expired = now + dt.timedelta(days=2) + with self.assertRaises(prep.Refusal): + prep.artifact_from_zip(raw, inventory, expired) + with self.assertRaises(prep.Refusal): + prep.artifact_from_zip(raw + b"x", inventory, now) + + def test_handoff_path_swap_cannot_change_the_captured_descriptor(self): + with tempfile.TemporaryDirectory(prefix="dw-fd-fixture-") as directory: + path = Path(directory) / "synthetic.zip" + raw = b"authenticated fixture" + path.write_bytes(raw) + info = path.stat() + binding = {"uid": info.st_uid, "gid": info.st_gid, "mode": stat.S_IMODE(info.st_mode), + "parent_mode": stat.S_IMODE(path.parent.stat().st_mode), "device": info.st_dev, + "inode": info.st_ino, "bytes": len(raw), "sha256": prep.digest(raw)} + original_open = os.open + def swapping_open(name, *args, **kwargs): + fd = original_open(name, *args, **kwargs) + if name == "synthetic.zip": + path.rename(path.with_name("old.zip")) + path.write_bytes(b"replacement pathname content") + return fd + with mock.patch("os.open", side_effect=swapping_open): + self.assertEqual(prep.held_regular(str(path), binding), raw) + self.assertNotEqual(path.read_bytes(), raw) + + def test_external_preloader_refuses_code_manifest_extra_and_link_before_load(self): + source = Path(__file__).parent + plan = (source / "rust-preparation-controller.md").read_text() + code = plan.split("<<'PY'\n", 1)[1].split("\nPY\n", 1)[0] + import shutil + import sys + with tempfile.TemporaryDirectory(prefix="dw-preloader-fixture-") as directory: + root = Path(directory) + names = ["rust-preparation-source.json", "rust_preparation.py", "rust_preparation_platform.py", "rust_preparation_authority.py", "test_rust_preparation.py"] + for name in names: + shutil.copyfile(source / name, root / name) + def check(error=None): + with mock.patch.dict("sys.modules"), mock.patch.object(sys, "argv", ["preloader", directory]), \ + mock.patch("builtins.print"), mock.patch("subprocess.Popen", side_effect=AssertionError("process")), \ + mock.patch("os.memfd_create", side_effect=AssertionError("seal")): + if error: + with self.assertRaisesRegex(RuntimeError, error): + exec(code, {}) + else: + with self.assertRaises(SystemExit) as result: + exec(code, {}) + self.assertEqual(result.exception.code, 78) + check() + target = root / "rust_preparation.py" + original = target.read_bytes() + target.write_bytes(b"raise AssertionError('unverified code loaded')\n") + check("source-member-hash") + target.write_bytes(original) + target.unlink() + target.symlink_to(source / "rust_preparation.py") + with self.assertRaises(OSError): + check() + target.unlink() + target.write_bytes(original) + extra = root / "json.py" + extra.write_bytes(b"raise AssertionError('shadow imported')") + check("source-closure") + extra.unlink() + target = root / "rust-preparation-source.json" + target.write_bytes(target.read_bytes() + b"\n") + check("source-manifest-hash") + + +class LifecycleControls(unittest.TestCase): + def setUp(self): + self.owner = prep.Ownership("a" * 64) + self.expected = (prep.Entry("site/index.html", b"ok"),) + + def supervised(self, effects): + try: + effects.register() + return prep.assemble(effects, self.owner, b"synthetic tar", self.expected, ["base"]) + finally: + # Models systemd's independent stop path, including hard worker loss. + # It is not called by assemble's cooperative cleanup path. + if effects.guardian: + prep.cleanup(effects, self.owner) + + def test_positive_and_guardian_before_create_countermodel(self): + effects = FakeEffects(self.expected) + self.assertTrue(self.supervised(effects).startswith("sha256:")) + self.assertEqual(effects.containers, {}) + self.assertTrue(effects.last_seal.closed) + unsafe = FakeEffects() + # Unsafe baseline: allocation precedes guardian registration. + unsafe.containers["f" * 64] = allocation("f" * 64, self.owner, "assemble") + self.assertFalse(unsafe.guardian) + self.assertEqual(len(unsafe.containers), 1) + safe = FakeEffects() + with self.assertRaises(prep.Refusal): + prep.allocate(safe, self.owner, "assemble", prep.BASE_ID) + self.assertEqual(safe.containers, {}) + + def test_hard_stop_at_every_create_copy_commit_readback_and_cleanup_edge(self): + edges = ["register-before", "register-after", "create-assemble-before", "create-assemble-after", + "inspect-rust-custody-assemble", "copy-before", "copy-after", "commit-before", + "commit-after", "image-inspect", "remove-before", "remove-after", + "create-verify-before", "create-verify-after", "inspect-rust-custody-verify", + "readback-before", "readback-after"] + for edge in edges: + with self.subTest(edge=edge): + effects = FakeEffects(self.expected) + effects.fault = edge + with self.assertRaises(HardStop): + self.supervised(effects) + self.assertEqual(effects.containers, {}) + if effects.last_seal: + self.assertTrue(effects.last_seal.closed) + + def test_lost_malformed_and_mismatched_return_still_clean_owned_allocation(self): + for returned in ("", "bad", "f" * 64): + with self.subTest(returned=returned): + effects = FakeEffects(self.expected) + effects.bad_return = returned + with self.assertRaises(prep.Refusal): + self.supervised(effects) + self.assertEqual(effects.containers, {}) + self.assertNotIn("copy-before", effects.calls) + + def test_foreign_name_nonce_collision_never_inspected_or_removed(self): + effects = FakeEffects(self.expected) + foreign = prep.Ownership("b" * 64) + cid = "f" * 64 + effects.containers[cid] = allocation(cid, foreign, "assemble") + original = copy.deepcopy(effects.containers) + with self.assertRaises(prep.Refusal): + self.supervised(effects) + self.assertEqual(effects.containers, original) + self.assertNotIn(cid, effects.inspected) + + def test_seal_prefix_config_layer_and_readback_mismatch_stop_and_clean(self): + for field in ("bad_seal", "bad_prefix", "bad_image", "bad_readback"): + with self.subTest(field=field): + effects = FakeEffects(self.expected) + setattr(effects, field, True) + with self.assertRaises(prep.Refusal): + self.supervised(effects) + self.assertEqual(effects.containers, {}) + effects = FakeEffects(self.expected) + effects.bad_record = lambda r: r["HostConfig"].update(NetworkMode="bridge") + with self.assertRaises(prep.Refusal): + self.supervised(effects) + self.assertEqual(effects.containers, {}) + + def test_continuing_daemon_work_is_unknown_and_never_accepted_as_cleanup(self): + effects = FakeEffects(self.expected) + effects.register() + prep.allocate(effects, self.owner, "assemble", prep.BASE_ID) + effects.uncertain = True + self.assertEqual(prep.cleanup(effects, self.owner), ("daemon-terminal-unknown",)) + self.assertEqual(effects.containers, {}) + with self.assertRaises(prep.Refusal): + effects.supervisor_ready() + + def test_unready_or_dead_supervisor_blocks_allocation(self): + for attribute in ("fresh_sample", "alive"): + effects = FakeEffects() + effects.register() + setattr(effects, attribute, False) + with self.assertRaises(prep.Refusal): + prep.allocate(effects, self.owner, "assemble", prep.BASE_ID) + self.assertEqual(effects.containers, {}) + + def test_late_owned_allocation_is_refused_for_use_but_cleanup_still_removes_it(self): + effects = FakeEffects() + effects.register() + record = allocation("f" * 64, self.owner, "assemble") + record["WithinPreparationWindow"] = False + effects.containers[record["Id"]] = record + with self.assertRaises(prep.Refusal): + prep.inspect_allocation(record, self.owner, "assemble", prep.BASE_ID) + self.assertEqual(prep.cleanup(effects, self.owner), ("allocation-window",)) + self.assertEqual(effects.containers, {}) + + +class MemoryKernel: + def __init__(self): + self.raw = b"" + self.position = 0 + self.flags = 0 + self.closed = False + + def create(self): + return 42 + + def write(self, _fd, raw): + prep.require(not self.flags, "sealed-write") + self.raw = raw + + def add_seals(self, _fd, flags): + self.flags = flags + + def seals(self, _fd): + return self.flags + + def rewind(self, _fd): + self.position = 0 + + def read(self, _fd): + result = self.raw[self.position:] + self.position = len(self.raw) + return result + + def close(self, _fd): + self.closed = True + + +class TerminalControls(unittest.TestCase): + def pair(self): + expected = {"unit": "synthetic.service", "attempt_sha256": "a" * 64, + "source_bundle_sha256": "b" * 64, "authority_sha256": "c" * 64, + "worker_command_sha256": "d" * 64, "stop_command_sha256": "e" * 64, "seal_flags": 15, + "custody_due": "2026-11-06T12:00:00+00:00", "reconciliation_ref": "synthetic", + "transport_uid": 1000, "transport_gid": 1000} + common = {"schema": "durable-workflow.preparation-stage/v1", "invocation_id": "f" * 32, + **{k: expected[k] for k in ("unit", "attempt_sha256", "source_bundle_sha256", "authority_sha256")}} + image = {"id": "sha256:" + "1" * 64, "config_sha256": "1" * 64, + "layers": ["sha256:" + "2" * 64, "sha256:" + "3" * 64]} + worker = {**common, "role": "worker", "pid": 100, "command_sha256": expected["worker_command_sha256"], + "outcome": "prepared-cleanup-pending", "subject": prep.SUBJECT, "node_packet_sha256": prep.PACKET_SHA, + "base_image_id": prep.BASE_ID, "base_layers": image["layers"][:1], "image": image, + "seal": {"sha256": "4" * 64, "flags": 15}, "input_tar_sha256": "4" * 64, + "readback_verified": True, "readback_sha256": "5" * 64, "foreign_before_sha256": "6" * 64, + "dependencies": {n: {"version": "1.64.0", "integrity": value, "sha256": "7" * 64} + for n, value in prep.DEPENDENCY_INTEGRITY.items()}} + operations = [ + ("pull", {"base_image_id": prep.BASE_ID, "rootfs_layers": worker["base_layers"]}), + ("create-assemble", {"container_id": "8" * 64}), + ("copy", {"container_id": "8" * 64, "input_tar_sha256": worker["input_tar_sha256"]}), + ("commit", {"image_id": image["id"], "config_sha256": image["config_sha256"], "rootfs_layers": image["layers"]}), + ("create-verify", {"container_id": "9" * 64}), + ] + rows = [{"path": n, "bytes": 1, "sha256": "a" * 64} for n in + (native.OWNERSHIP_FILE, native.RESULTS_FILE, native.CONTEXT_FILE, native.PENDING_FILE, "docker/config.json")] + transport = {"state": prep.RETAINED_EVIDENCE, "owner": prep.OWNER, "attempt_sha256": expected["attempt_sha256"], + "device": 1, "inode": 2, "uid": 1000, "gid": 1000, "mode": 0o700, "files": rows, + "bytes": 5, "sha256": prep.digest(authority.canonical_bytes(rows)), "custody_due": expected["custody_due"], "reconciliation_ref": expected["reconciliation_ref"]} + stop = {**common, "role": "stop", "pid": 101, "worker_pid": worker["pid"], + "command_sha256": expected["stop_command_sha256"], "outcome": "complete", "worker_exit_code": 1, + "worker_exit_status": 0, "cleanup_exit_status": 0, "pending": False, "owned_survivors": [], + "problems": [], "transport": transport, "foreign_preserved": True, + "foreign_before_sha256": worker["foreign_before_sha256"], + "operation_results": [{"operation": name, "identity": identity, "attempt_sha256": expected["attempt_sha256"]} + for name, identity in operations]} + return expected, worker, stop + + def capture(self, records): + return b"\n".join(json.dumps(r).encode() for r in records) + + def test_acceptance_joins_both_terminal_stages_and_successful_unit_wait(self): + expected, worker, stop = self.pair() + raw = self.capture([worker, stop]) + result = prep.collect_terminal(raw, expected, 0) + self.assertEqual(result["outcome"], "prepared-inputs-only") + self.assertEqual((result["worker_pid"], result["stop_pid"]), (100, 101)) + self.assertFalse(result["compatibility_accepted"]) + self.assertFalse(result["rollout_approved"]) + self.assertEqual(result["browser"], "not-run") + for code in (1, -9, None, False): + with self.subTest(code=code), self.assertRaises(prep.Refusal): + prep.collect_terminal(raw, expected, code) + + def test_worker_only_stop_only_duplicate_or_lost_output_cannot_pass(self): + expected, worker, stop = self.pair() + for records in ([], [worker], [stop], [worker, worker], [worker, stop, stop], [None, stop]): + raw = self.capture(records) + with self.subTest(records=len(records)), self.assertRaises(prep.Refusal): + prep.collect_terminal(raw, expected, 0) + + def test_cross_attempt_pid_command_source_and_authority_splices_refuse(self): + expected, worker, stop = self.pair() + for field, wrong in (("unit", "other.service"), ("attempt_sha256", "0" * 64), + ("source_bundle_sha256", "0" * 64), ("authority_sha256", "0" * 64), + ("invocation_id", "0" * 32), ("worker_pid", 999), ("pid", worker["pid"]), + ("command_sha256", "0" * 64), ("worker_exit_status", 1), ("cleanup_exit_status", 1)): + changed = copy.deepcopy(stop) + changed[field] = wrong + raw = self.capture([worker, changed]) + with self.subTest(field=field), self.assertRaises(prep.Refusal): + prep.collect_terminal(raw, expected, 0) + + def test_pending_survivor_foreign_transport_or_operation_drift_cannot_pass(self): + expected, worker, stop = self.pair() + for field, wrong in (("pending", True), ("owned_survivors", ["8" * 64]), ("problems", ["unresolved"]), + ("foreign_preserved", False), ("transport", None), ("foreign_before_sha256", "0" * 64)): + changed = copy.deepcopy(stop) + changed[field] = wrong + raw = self.capture([worker, changed]) + with self.subTest(field=field), self.assertRaises(prep.Refusal): + prep.collect_terminal(raw, expected, 0) + mutations = ((0, "base_image_id", "sha256:" + "0" * 64), (1, "container_id", "malformed"), + (2, "container_id", "0" * 64), (3, "image_id", "sha256:" + "0" * 64), + (4, "container_id", "8" * 64)) + for index, field, wrong in mutations: + changed = copy.deepcopy(stop) + changed["operation_results"][index]["identity"][field] = wrong + raw = self.capture([worker, changed]) + with self.subTest(index=index), self.assertRaises(prep.Refusal): + prep.collect_terminal(raw, expected, 0) + + def test_wrong_dependency_seal_layer_or_readback_refuses_despite_cleanup_success(self): + expected, worker, stop = self.pair() + for field, wrong in (("readback_verified", False), ("node_packet_sha256", "0" * 64), + ("seal", {"sha256": "0" * 64, "flags": 15}), ("base_layers", ["sha256:" + "0" * 64])): + changed = copy.deepcopy(worker) + changed[field] = wrong + raw = self.capture([changed, stop]) + with self.subTest(field=field), self.assertRaises(prep.Refusal): + prep.collect_terminal(raw, expected, 0) + changed = copy.deepcopy(worker) + changed["dependencies"]["playwright"]["integrity"] = "sha512-unreviewed" + raw = self.capture([changed, stop]) + with self.assertRaises(prep.Refusal): + prep.collect_terminal(raw, expected, 0) + + def test_retained_evidence_requires_closed_bounded_custody_and_attempt(self): + expected, worker, stop = self.pair() + for field, wrong in (("owner", "foreign"), ("inode", 0), ("mode", 0o777), + ("attempt_sha256", "0" * 64), ("custody_due", "other"), + ("bytes", prep.EVIDENCE_LIMIT + 1), ("sha256", "0" * 64), ("files", [])): + changed = copy.deepcopy(stop) + changed["transport"][field] = wrong + raw = self.capture([worker, changed]) + with self.subTest(field=field), self.assertRaises(prep.Refusal): + prep.collect_terminal(raw, expected, 0) + + +class AuthorityControls(unittest.TestCase): + def test_runtime_hash_allows_read_atime_but_rejects_content_metadata_race(self): + before = SimpleNamespace(st_mode=stat.S_IFREG | 0o644, st_uid=0, st_gid=0, st_dev=1, + st_ino=1, st_size=4, st_mtime_ns=1, st_ctime_ns=1, st_nlink=1, st_atime_ns=1) + after = copy.copy(before) + after.st_atime_ns = 2 + observer = authority.NativeObserver() + with mock.patch("os.fstat", side_effect=[before, after]), mock.patch("os.read", side_effect=[b"ELF!", b""]), \ + mock.patch("os.lseek"): + observed = observer._metadata("/synthetic/root-owned-runtime", 999) + self.assertEqual(observed["sha256"], prep.digest(b"ELF!")) + after.st_mtime_ns = 2 + observer = authority.NativeObserver() + with mock.patch("os.fstat", side_effect=[before, after]), mock.patch("os.read", side_effect=[b"ELF!", b""]), \ + mock.patch("os.lseek"): + with self.assertRaises(prep.Refusal): + observer._metadata("/synthetic/root-owned-runtime", 999) + + def record(self): + def binding(path): + return {"path": path, "uid": 0, "gid": 0, "mode": 0o755, "device": 1, + "inode": 2, "bytes": 4, "sha256": "f" * 64} + roots = {"/usr/lib/fixture": {"kind": "directory", "members": {"lib.so": {"kind": "file"}}}} + tools = {name: {"entry_path": path, "binding": binding(path), "runtime_roots": list(roots)} + for name, path in authority.ENTRY_PATHS.items()} + processes = {name: {"pid": n + 10, "start_ticks": "1", "uids": [0, 0, 0, 0], + "exe": binding("/usr/bin/" + name), "mapped_files": ["/usr/lib/fixture/lib.so"]} + for n, name in enumerate(sorted(authority.PROCESSES))} + packages = {"/var/lib/dpkg/status": binding("/var/lib/dpkg/status")} + return {"schema": "durable-workflow.installed-authority/v1", "boot_id": "synthetic-boot", "tools": tools, "processes": processes, + "catalogs": roots, "python_search": ["/usr/lib/fixture"], + "loader_configs": {p: binding(p) for p in authority.LOADER_CONFIGS}, "package_receipts": packages, + "socket": {"path": "/run/docker.sock", "peer_pid": processes["daemon"]["pid"]}, + "closure_receipt": {"independent_review": "synthetic-source-control", + "package_catalog_sha256": prep.digest(authority.canonical_bytes(packages)), + "complete_loader_closure": True}} + + def observer(self, actual): + class Observer: + def boot_id(self): + return actual["boot_id"] + def python_search(self): + return actual["python_search"] + def catalog(self, root): + return actual["catalogs"][root] + def file(self, path): + values = [t["binding"] for t in actual["tools"].values()] + values += list(actual["loader_configs"].values()) + list(actual["package_receipts"].values()) + return next(v for v in values if v["path"] == path) + def resolve(self, path): + return path + def process(self, pid): + return next(p for p in actual["processes"].values() if p["pid"] == pid) + def socket_identity(self, _path): + return actual["socket"] + return Observer() + + def test_unbound_installed_authority_refuses_before_observation(self): + raw = authority.canonical_bytes(self.record()) + with self.assertRaises(prep.Refusal): + authority.HostAuthority(raw, None) + observer = authority.NativeObserver() + with self.assertRaises(prep.Refusal): + observer.socket_identity("/run/docker.sock") + + def test_complete_catalog_rejects_extra_library_tool_process_and_socket_drift(self): + record = self.record() + raw = authority.canonical_bytes(record) + expected = prep.digest(raw) + authority.HostAuthority(raw, self.observer(copy.deepcopy(record)), expected) + cases = [] + for field, key, wrong in (("catalogs", "/usr/lib/fixture", {"kind": "directory", "members": {"extra.so": {}}}), + ("python_search", None, ["/unbound"]), ("boot_id", None, "other-boot")): + changed = copy.deepcopy(record) + if key is None: + changed[field] = wrong + else: + changed[field][key] = wrong + cases.append(changed) + for tool in authority.TOOLS: + changed = copy.deepcopy(record) + changed["tools"][tool]["binding"]["sha256"] = "a" * 64 + cases.append(changed) + for process in authority.PROCESSES: + changed = copy.deepcopy(record) + changed["processes"][process]["start_ticks"] = "2" + cases.append(changed) + changed = copy.deepcopy(record) + changed["socket"]["peer_pid"] = 999 + cases.append(changed) + for actual in cases: + observer = self.observer(actual) + with self.assertRaises(prep.Refusal): + authority.HostAuthority(raw, observer, expected) + + def test_invocation_cannot_narrow_the_pinned_tool_and_library_closure(self): + for field in ("tools", "catalogs"): + record = self.record() + record[field].pop(next(iter(record[field]))) + raw = authority.canonical_bytes(record) + observer = self.observer(record) + expected = prep.digest(raw) + with self.assertRaises(prep.Refusal): + authority.HostAuthority(raw, observer, expected) + + +class PlatformControls(unittest.TestCase): + def test_unbound_native_entrypoints_refuse_before_any_effect(self): + with mock.patch("subprocess.Popen", side_effect=AssertionError("unexpected process")), \ + mock.patch("os.memfd_create", side_effect=AssertionError("unexpected memfd")): + calls = [lambda: native.CommandRunner({}).run(["unused"]), + lambda: native.Seal(b"synthetic"), + lambda: native.fetch_dependency("playwright", None), + lambda: native.create_transport({}, "a" * 64), + lambda: native.launch({}, {}), + lambda: native.DockerConnection("localhost").connect()] + for call in calls: + with self.subTest(call=call), self.assertRaises(prep.Refusal): + call() + + def test_captured_source_unit_refuses_without_resolving_worktree_modules(self): + directory = Path(__file__).parent + modules = {name: (directory / (name + ".py")).read_bytes() + for name in ("rust_preparation", "rust_preparation_authority", "rust_preparation_platform")} + stub = native.unit_stub(modules) + with mock.patch.dict("sys.modules"), mock.patch("builtins.print"), \ + mock.patch("subprocess.Popen", side_effect=AssertionError("unexpected process")), \ + mock.patch("os.memfd_create", side_effect=AssertionError("unexpected seal")): + with self.assertRaises(SystemExit) as result: + exec(stub, {}) + self.assertEqual(result.exception.code, 78) + + def test_real_seal_logic_uses_fake_kernel_and_refuses_mutation_missing_flags(self): + kernel = MemoryKernel() + with mock.patch.object(prep, "OPERATION_ADMISSION", "e" * 64): + seal = native.Seal(b"synthetic public bytes", kernel) + self.assertEqual(seal.verified_hash(), prep.digest(kernel.raw)) + with self.assertRaises(prep.Refusal): + kernel.write(42, b"changed") + flags = kernel.flags + kernel.flags = 0 + with self.assertRaises(prep.Refusal): + seal.verified_hash() + kernel.flags = flags + kernel.raw = b"changed despite simulated seals" + with self.assertRaises(prep.Refusal): + seal.verified_hash() + seal.close() + self.assertTrue(kernel.closed) + + def test_daemon_operation_journal_precedes_request_and_survives_timeout(self): + with tempfile.TemporaryDirectory(prefix="dw-journal-fixture-") as directory: + spec, runtime = transport_fixture(directory) + pending = Path(spec["staging_path"]) / native.PENDING_FILE + class Runner: + def run(self, *_args, **_kwargs): + if not pending.is_file(): + raise AssertionError("daemon request preceded journal") + raise prep.Refusal("command-timeout") + with mock.patch.object(prep, "OPERATION_ADMISSION", "e" * 64): + effects = native.LiveEffects(spec, runtime, {}, Runner()) + try: + with mock.patch.object(effects, "supervisor_ready"): + with self.assertRaises(prep.Refusal): + effects.command(["create"], mutation="create-assemble") + self.assertTrue(effects.uncertain_daemon_operation()) + self.assertEqual(json.loads(pending.read_bytes())["operation"], "create-assemble") + finally: + effects.transport.close() + + def test_successful_malformed_commit_and_post_exit_loss_remain_unresolved(self): + for response in (b"not-an-image-id\n", b"sha256:" + b"c" * 64 + b"\n"): + with self.subTest(response=response), tempfile.TemporaryDirectory(prefix="dw-commit-fixture-") as directory: + spec, runtime = transport_fixture(directory) + class Runner: + def run(self, *_args, **_kwargs): + return response + with mock.patch.object(prep, "OPERATION_ADMISSION", "e" * 64): + effects = native.LiveEffects(spec, runtime, {}, Runner()) + try: + with mock.patch.object(effects, "supervisor_ready"): + returned = effects.commit("f" * 64) + self.assertTrue(effects.uncertain_daemon_operation()) + self.assertEqual(effects.require_transport().results(), []) + if response.startswith(b"not-"): + self.assertIsNone(__import__("re").fullmatch(prep.IMAGE_ID, returned)) + else: + # Process exit has happened; loss before binding must + # leave the same unresolved journal and no acceptance. + with self.assertRaises(HardStop): + raise HardStop("after-exit-before-binding") + self.assertTrue(effects.uncertain_daemon_operation()) + finally: + effects.transport.close() + + def retained_fixture(self, directory): + spec, runtime = transport_fixture(directory) + root = Path(spec["staging_path"]) + nonce = runtime["nonce"] + rows = [{"operation": "pull", "nonce": nonce, "state": state} for state in ("pending", "bound")] + for name, raw in ((native.PENDING_FILE, b"\n".join(json.dumps(r).encode() for r in rows)), + (native.RESULTS_FILE, b'{"operation":"pull"}\n'), (native.CONTEXT_FILE, b"{}")): + (root / name).write_bytes(raw) + (root / name).chmod(0o600) + return spec, runtime + + def test_retention_never_deletes_empty_or_nonempty_replacement_at_any_edge(self): + edges = ("root-open", "list-root", "child-open", "child-read", "final-namespace") + for empty in (True, False): + for edge in edges: + with self.subTest(empty=empty, edge=edge), tempfile.TemporaryDirectory(prefix="dw-transport-swap-fixture-") as directory: + spec, runtime = self.retained_fixture(directory) + root = Path(spec["staging_path"]) + original_stat = root.stat() + real_open, real_list = os.open, os.listdir + real_namespace = native.TransportDirectory.namespace_matches + state = {"root_fd": None, "swapped": False, "namespace_checks": 0} + def swap(): + if not state["swapped"]: + root.rename(root.with_name("bound-original")) + root.mkdir(mode=0o700) + if not empty: + (root / "foreign").write_bytes(b"replacement bytes") + state["swapped"] = True + def opened(name, *args, **kwargs): + fd = real_open(name, *args, **kwargs) + if name == root.name: + state["root_fd"] = fd + if edge == "root-open": + swap() + if name == "home" and edge == "child-open": + swap() + if name == "config.json" and edge == "child-read": + swap() + return fd + def listed(value): + answer = real_list(value) + if value == state["root_fd"] and edge == "list-root": + swap() + return answer + def namespace(instance): + state["namespace_checks"] += 1 + if edge == "final-namespace" and state["namespace_checks"] == 2: + swap() + return real_namespace(instance) + with mock.patch.object(prep, "OPERATION_ADMISSION", "e" * 64), \ + mock.patch("os.open", side_effect=opened), mock.patch("os.listdir", side_effect=listed), \ + mock.patch.object(native.TransportDirectory, "namespace_matches", namespace), \ + mock.patch("os.unlink", side_effect=AssertionError("unexpected transport deletion")), \ + mock.patch("os.rmdir", side_effect=AssertionError("unexpected transport deletion")): + with self.assertRaises(prep.Refusal): + native.retain_transport(spec, runtime) + self.assertTrue(state["swapped"]) + self.assertTrue(root.is_dir()) + self.assertEqual(list(root.iterdir()), [] if empty else [root / "foreign"]) + if not empty: + self.assertEqual((root / "foreign").read_bytes(), b"replacement bytes") + moved = root.with_name("bound-original").stat() + self.assertEqual((moved.st_dev, moved.st_ino), (original_stat.st_dev, original_stat.st_ino)) + + def test_exact_predecessor_adopts_first_stat_swap_and_deletes_final_empty_swap(self): + self.assertEqual(prep.digest(PREDECESSOR_CREATE.encode()), PREDECESSOR_CREATE_SHA256) + self.assertEqual(prep.digest(PREDECESSOR_REMOVE.encode()), PREDECESSOR_REMOVE_SHA256) + namespace = dict(native.__dict__) + exec(PREDECESSOR_CREATE, namespace) + exec(PREDECESSOR_REMOVE, namespace) + with tempfile.TemporaryDirectory(prefix="dw-old-create-fixture-") as directory: + root = Path(directory) / "transport" + spec = {"staging_path": str(root), "uid": os.getuid(), "gid": os.getgid()} + real_stat = Path.stat + state = {"swapped": False} + def observed(path, *args, **kwargs): + if path == root and not state["swapped"]: + state["swapped"] = True + root.rename(root.with_name("bound-original")) + root.mkdir(mode=0o700) + (root / "home").mkdir(mode=0o700) + (root / "docker").mkdir(mode=0o700) + (root / "foreign").write_bytes(b"foreign creation marker") + return real_stat(path, *args, **kwargs) + # Avoid a recursive Path.exists() inside the intercepted stat. + def intercepted(path, *args, **kwargs): + if path == root and os.path.isdir(root) and not state["swapped"]: + return observed(path, *args, **kwargs) + return real_stat(path, *args, **kwargs) + with mock.patch.object(prep, "OPERATION_ADMISSION", "e" * 64), mock.patch.object(Path, "stat", intercepted): + namespace["create_transport"](spec, "a" * 64) + self.assertTrue((root / native.OWNERSHIP_FILE).is_file()) + self.assertTrue((root / "docker/config.json").is_file()) + self.assertEqual((root / "foreign").read_bytes(), b"foreign creation marker") + with tempfile.TemporaryDirectory(prefix="dw-old-remove-fixture-") as directory: + spec, runtime = transport_fixture(directory) + root = Path(spec["staging_path"]) + transport = native.TransportDirectory(spec, runtime) + real_rmdir = os.rmdir + def removed(name, *args, **kwargs): + if name == root.name: + root.rename(root.with_name("bound-original")) + root.mkdir(mode=0o700) + return real_rmdir(name, *args, **kwargs) + try: + with mock.patch("os.rmdir", side_effect=removed): + with self.assertRaises(prep.Refusal): + namespace["remove_contents"](transport) + self.assertFalse(root.exists()) + self.assertTrue(root.with_name("bound-original").is_dir()) + finally: + transport.close() + + def test_creation_binds_descriptor_before_first_identity_capture_and_preserves_swap(self): + for child_swap in (False, True): + with self.subTest(child_swap=child_swap), tempfile.TemporaryDirectory(prefix="dw-create-swap-fixture-") as directory: + root = Path(directory) / "transport" + spec = {"staging_path": str(root), "uid": os.getuid(), "gid": os.getgid()} + real_open, real_fstat = os.open, os.fstat + state = {"target_fd": None, "swapped": False, "bound": None} + def opened(name, *args, **kwargs): + fd = real_open(name, *args, **kwargs) + if name == ("docker" if child_swap else root.name): + state["target_fd"] = fd + return fd + def observed(fd): + answer = real_fstat(fd) + if fd == state["target_fd"] and not state["swapped"]: + target = root / "docker" if child_swap else root + state["bound"] = (answer.st_dev, answer.st_ino) + target.rename(target.with_name("bound-original")) + target.mkdir(mode=0o700) + (target / "foreign").write_bytes(b"foreign creation marker") + state["swapped"] = True + return answer + with mock.patch.object(prep, "OPERATION_ADMISSION", "e" * 64), \ + mock.patch("os.open", side_effect=opened), mock.patch("os.fstat", side_effect=observed): + with self.assertRaises(prep.Refusal): + native.create_transport(spec, "a" * 64) + target = root / "docker" if child_swap else root + self.assertEqual(list(target.iterdir()), [target / "foreign"]) + self.assertEqual((target / "foreign").read_bytes(), b"foreign creation marker") + moved = target.with_name("bound-original").stat() + actual_identity = (moved.st_dev, moved.st_ino) + expected_identity = state["bound"] + self.assertEqual(actual_identity, expected_identity) + + def test_descriptor_creation_binds_both_children_and_empty_credential_transport(self): + with tempfile.TemporaryDirectory(prefix="dw-created-transport-fixture-") as directory: + root = Path(directory) / "transport" + spec = {"staging_path": str(root), "uid": os.getuid(), "gid": os.getgid()} + runtime = {"nonce": "a" * 64} + with mock.patch.object(prep, "OPERATION_ADMISSION", "e" * 64): + record = native.create_transport(spec, runtime["nonce"]) + self.assertEqual(set(record["children"]), {"home", "docker"}) + transport = native.TransportDirectory(spec, runtime) + try: + runner = native.CommandRunner(native.clean_environment(root)) + self.assertEqual(runner.environment["DOCKER_CONFIG"], "/proc/self/fd/-1") + runner.bind_transport(transport) + self.assertEqual(runner.environment["DOCKER_CONFIG"], "/proc/self/fd/" + str(transport.children["docker"])) + expected_fds = {transport.fd, *transport.children.values()} + self.assertEqual(set(runner.pass_fds), expected_fds) + finally: + transport.close() + + def test_child_directory_swap_preserves_both_trees_and_refuses_custody(self): + for empty in (True, False): + with self.subTest(empty=empty), tempfile.TemporaryDirectory(prefix="dw-child-swap-fixture-") as directory: + spec, runtime = self.retained_fixture(directory) + root = Path(spec["staging_path"]) + real_open = os.open + def opened(name, *args, **kwargs): + fd = real_open(name, *args, **kwargs) + if name == "docker": + (root / "docker").rename(root / "original-docker") + (root / "docker").mkdir(mode=0o700) + if not empty: + (root / "docker/foreign").write_bytes(b"foreign child") + return fd + with mock.patch.object(prep, "OPERATION_ADMISSION", "e" * 64), mock.patch("os.open", side_effect=opened), \ + mock.patch("os.unlink", side_effect=AssertionError("unexpected deletion")), \ + mock.patch("os.rmdir", side_effect=AssertionError("unexpected deletion")): + with self.assertRaises(prep.Refusal): + native.retain_transport(spec, runtime) + self.assertEqual((root / "original-docker/config.json").read_bytes(), native.ANONYMOUS_CONFIG) + self.assertTrue((root / "docker").is_dir()) + if not empty: + self.assertEqual((root / "docker/foreign").read_bytes(), b"foreign child") + + def test_retained_transport_has_finite_closed_manifest_and_no_deletion(self): + with tempfile.TemporaryDirectory(prefix="dw-retention-fixture-") as directory: + spec, runtime = self.retained_fixture(directory) + with mock.patch.object(prep, "OPERATION_ADMISSION", "e" * 64), \ + mock.patch("os.unlink", side_effect=AssertionError("unexpected deletion")), \ + mock.patch("os.rmdir", side_effect=AssertionError("unexpected deletion")): + receipt = native.retain_transport(spec, runtime) + self.assertEqual(receipt["state"], prep.RETAINED_EVIDENCE) + self.assertEqual(receipt["owner"], prep.OWNER) + self.assertLessEqual(receipt["bytes"], 144 * 1024) + self.assertEqual(len(receipt["files"]), 5) + prep.terminal_retained_evidence(receipt, {"attempt_sha256": receipt["attempt_sha256"], "custody_due": receipt["custody_due"], + "reconciliation_ref": receipt["reconciliation_ref"], "transport_uid": os.getuid(), "transport_gid": os.getgid()}) + + def test_bound_mutations_append_terminal_journal_without_unlink(self): + with tempfile.TemporaryDirectory(prefix="dw-append-journal-fixture-") as directory: + spec, runtime = transport_fixture(directory) + with mock.patch("os.unlink", side_effect=AssertionError("unexpected journal deletion")): + transport = native.TransportDirectory(spec, runtime) + try: + for name in ("pull", "create-assemble"): + transport.write_pending({"operation": name, "nonce": runtime["nonce"]}) + self.assertIsNotNone(transport.pending_operation()) + transport.bind_operation(name, {"synthetic": True}, runtime["nonce"]) + self.assertIsNone(transport.pending_operation()) + self.assertEqual([r["operation"] for r in transport.results()], ["pull", "create-assemble"]) + finally: + transport.close() + + def test_clean_environment_has_no_ambient_credentials_or_proxy(self): + with mock.patch.dict(os.environ, {"GH_TOKEN": "synthetic", "HTTPS_PROXY": "synthetic", + "NODE_OPTIONS": "synthetic", "SSH_AUTH_SOCK": "synthetic"}): + environment = native.clean_environment(Path("/synthetic-public-input")) + self.assertTrue(set(environment).isdisjoint({"GH_TOKEN", "HTTPS_PROXY", "NODE_OPTIONS", "SSH_AUTH_SOCK"})) + self.assertEqual(environment["DOCKER_CONFIG"], "/synthetic-public-input/docker") + + def test_effective_unit_source_limits_and_readiness_are_required(self): + runtime = {"deadline": __import__("time").monotonic() + 30} + modules = {"rust_preparation": b"core", "rust_preparation_platform": b"platform", "rust_preparation_authority": b"authority"} + values = {"ActiveState": "active", "RuntimeMaxUSec": 600000000, + "TimeoutStopUSec": 15000000, "KillMode": "control-group", "Restart": "no", + "MemoryMax": 1024 * prep.MIB, "TasksMax": 64, "UMask": 0o77, + "CPUQuotaPerSecUSec": 1000000, "MainPID": os.getpid()} + for key, mode in (("ExecStart", "worker"), ("ExecStopPost", "cleanup")): + values[key] = [[native.ENV, native.unit_argv(mode, {}, runtime, modules)]] + def value(_runner, _path, _interface, name): + return values[name] + with mock.patch.object(native, "service_path", return_value="/synthetic-unit"), \ + mock.patch.object(native, "bus_value", side_effect=value): + native.check_supervisor(None, {}, runtime, modules) + for name, bad in (("ExecStopPost", []), ("RuntimeMaxUSec", 0), + ("ActiveState", "dead"), ("TasksMax", 0)): + original = values[name] + values[name] = bad + with self.subTest(name=name), self.assertRaises(prep.Refusal): + native.check_supervisor(None, {}, runtime, modules) + values[name] = original + + def test_unit_environment_is_cleared_before_both_interpreters(self): + runtime = {"nonce": "a" * 64, "deadline": 1} + modules = {"rust_preparation": b"source", "rust_preparation_platform": b"platform", "rust_preparation_authority": b"authority"} + for mode in ("worker", "cleanup"): + argv = native.unit_argv(mode, {"staging_path": "/synthetic"}, runtime, modules) + self.assertEqual(argv[:2], [native.ENV, "-i"]) + self.assertLess(argv.index("-i"), argv.index(native.PYTHON)) + self.assertIn("HOME=/synthetic/home", argv) + self.assertIn("-S", argv) + + def test_readback_refuses_owner_and_permission_drift_not_just_file_bytes(self): + output = io.BytesIO() + with tarfile.open(fileobj=output, mode="w") as value: + info = tarfile.TarInfo("packet/file") + info.uid = info.gid = 1000 + info.mode = 0o444 + info.size = 1 + value.addfile(info, io.BytesIO(b"x")) + raw = output.getvalue() + native.validate_readback_metadata(raw) + for owner, mode in ((0, 0o444), (1000, 0o644)): + changed = io.BytesIO() + with tarfile.open(fileobj=changed, mode="w") as value: + info.uid = owner + info.mode = mode + value.addfile(info, io.BytesIO(b"x")) + changed_raw = changed.getvalue() + with self.assertRaises(prep.Refusal): + native.validate_readback_metadata(changed_raw) + + def test_capture_and_transport_happen_only_after_supervisor_readback(self): + events = [] + class Effects: + def __init__(self, *_args): + pass + def supervisor_ready(self): + events.append("supervised") + def transport(*_args): + events.append("transport") + raise prep.Refusal("fixture-stop") + with mock.patch.object(native, "runtime_identity"), mock.patch.object(native, "LiveEffects", Effects), \ + mock.patch.object(native, "create_transport", side_effect=transport): + with self.assertRaises(prep.Refusal): + native.run_worker({}, {"nonce": "a" * 64}, {}) + self.assertEqual(events, ["supervised", "transport"]) + + def test_cleanup_entrypoint_requires_actual_stop_pid_and_exact_captured_command(self): + runtime = {"nonce": "a" * 64, "deadline": 1} + modules = {"rust_preparation": b"source", "rust_preparation_platform": b"platform", "rust_preparation_authority": b"authority"} + values = {"ControlPID": os.getpid(), "ActiveState": "deactivating", + "ExecStopPost": [[native.ENV, native.unit_argv("cleanup", {}, runtime, modules)]]} + def value(_runner, _path, _interface, name): + return values[name] + with mock.patch.object(native, "service_path", return_value="/synthetic-unit"), \ + mock.patch.object(native, "bus_value", side_effect=value): + native.check_stop_authority(None, {}, runtime, modules) + for field, wrong in (("ControlPID", -1), ("ActiveState", "active"), ("ExecStopPost", [])): + original = values[field] + values[field] = wrong + with self.subTest(field=field), self.assertRaises(prep.Refusal): + native.check_stop_authority(None, {}, runtime, modules) + values[field] = original + + def test_admission_binds_fresh_metadata_modes_subject_and_executor(self): + now = dt.datetime(2026, 11, 5, tzinfo=dt.timezone.utc) + binding = {"path": "/synthetic/zip", "uid": os.getuid(), "gid": os.getgid(), "mode": 0o600, + "parent_mode": 0o700, "bytes": prep.ZIP_BYTES, "sha256": prep.ZIP_SHA} + metadata = {**prep.SUBJECT, "deployment_id": 7564108107, "event": "push", "branch": "main", + "attempt": 1, "run_conclusion": "success", "deployment_state": "success"} + spec = {"artifact_handoff": binding, "packet_bindings": {name: dict(binding) for name in + (*prep.PACKET_MEMBERS, "rust-local-browser-packet.json")}, "staging_path": "/synthetic/staging", + "uid": os.getuid(), "gid": os.getgid(), "docker_root": "/synthetic/docker", + "metadata_at": now.isoformat(), "metadata_expires": prep.SUBJECT_EXPIRY, "metadata_subject": metadata, + "live_build_sha256": "3e8d1827d1df8bf3c47ee8f0efcbd2f3692cb1bd8741f40c1271ef5c74f4a183", + "python_sha256": "f" * 64, "env_sha256": "e" * 64, "authority_binding": dict(binding), + "retention": {"owner": prep.OWNER, "custody_due": (now + dt.timedelta(hours=23)).isoformat(), + "reconciliation_ref": "https://github.com/durable-workflow/durable-workflow.github.io/pull/480#issuecomment-1"}} + native.admission(spec, now) + invalid = [] + bad = copy.deepcopy(spec) + bad["metadata_at"] = (now - dt.timedelta(minutes=10)).isoformat() + invalid.append(bad) + bad = copy.deepcopy(spec) + bad["metadata_subject"]["artifact_id"] = 1 + invalid.append(bad) + bad = copy.deepcopy(spec) + bad["artifact_handoff"]["mode"] = 0o666 + invalid.append(bad) + bad = copy.deepcopy(spec) + bad["uid"] += 1 + invalid.append(bad) + for candidate in invalid: + with self.assertRaises(prep.Refusal): + native.admission(candidate, now) + + +if __name__ == "__main__": + unittest.main()