diff --git a/scripts/conformance/command_keys/README.md b/scripts/conformance/command_keys/README.md index 105bd7f..dded0dd 100644 --- a/scripts/conformance/command_keys/README.md +++ b/scripts/conformance/command_keys/README.md @@ -24,11 +24,40 @@ Store both as regular, single-link, owned files in owner-controlled directories. Before the experiment, check host/Docker capacity and pull both exact images with an empty Docker config. The controller itself does not pull images. -Run from an exact committed checkout: +Run from an exact committed checkout. Qualifying execution uses a literal trusted +bootstrap to verify the reviewed controller Git bytes before compiling them. +Set `REVIEWED_COMMIT` and `REVIEWED_SHA` from the independent review for the +chosen controller. The expected identity is supplied by that review. ```sh -python3 -I scripts/conformance/command_keys/run_response_loss.py \ - --runner-commit FULL_REVIEWED_RUNNER_COMMIT \ +BOOTSTRAP='import hashlib, os, re, subprocess, sys +from pathlib import Path +commit, relative, expected = sys.argv[1:4] +del sys.argv[1:4] +allowed = {"scripts/conformance/command_keys/" + name for name in + ("run_response_loss.py", "run_physical_readback.py", "run_synthetic_legacy.py")} +if (re.fullmatch(r"[0-9a-f]{40}", commit) is None + or re.fullmatch(r"[0-9a-f]{64}", expected) is None or relative not in allowed): + raise ValueError("require reviewed controller identities") +data = subprocess.check_output(["/usr/bin/git", "show", commit + ":" + relative], + env={"PATH": os.defpath}, timeout=10) +if hashlib.sha256(data).hexdigest() != expected: + raise ValueError("reviewed controller Git bytes differ") +context = {"__name__": "__main__", "__file__": str(Path.cwd() / relative)} +exec(compile(data, context["__file__"], "exec"), context) +' +``` + +The [frozen physical/loss invocation and pre-open substitution controls](https://github.com/durable-workflow/sdk-python/pull/321#issuecomment-6862370654) +show the complete recorded trust root. The adapter still captures and verifies +all host source/data files against the declared commit before creating its +workspace. Opening the host controller pathname first is insufficient for +qualification. + +```sh +/usr/bin/python3 -I -B -c "$BOOTSTRAP" "$REVIEWED_COMMIT" \ + scripts/conformance/command_keys/run_response_loss.py "$REVIEWED_SHA" \ + --runner-commit "$REVIEWED_COMMIT" \ --wheel /path/to/downloaded/durable_workflow-2.17.2-py3-none-any.whl \ --proxy /path/to/verified/committed_response_loss.py \ --output /path/to/private-disk-scratch/new-run @@ -101,8 +130,9 @@ reviewed response-loss runner. No host interpreter or executable input pathname is accepted. ```sh -python3 -I scripts/conformance/command_keys/run_physical_readback.py \ - --runner-commit FULL_REVIEWED_RUNNER_COMMIT \ +/usr/bin/python3 -I -B -c "$BOOTSTRAP" "$REVIEWED_COMMIT" \ + scripts/conformance/command_keys/run_physical_readback.py "$REVIEWED_SHA" \ + --runner-commit "$REVIEWED_COMMIT" \ --wheel /path/to/downloaded/durable_workflow-2.17.2-py3-none-any.whl \ --baseline /path/to/producer/stopped-backup.sqlite \ --seed /path/to/producer/seed-receipts.json \ @@ -147,3 +177,82 @@ A substituted parent, copied database, configuration, credential or report refus qualification. Known-container cleanup stays bound to captured IDs and preserves replacement paths. Workspace substitution controls must pass before installed response-loss or physical results qualify. + +## Synthetic SQLite legacy cases + +`run_synthetic_legacy.py` prepares real current-producer API admissions, stops +that owned runtime, then calls the unchanged accepted common helper on a private +stopped copy. Its helper source is Rust `4add871b175517f2031f71c868ce278bbfbba486`, +SHA-256 `6fc6054ce087c136a86d7b6b222daeefaa87c58fa0b613b155a5df39e6a26fad`. +The producer manifest is Server `d66e33bfb5c76f94c5b202f8c9f7a9e8cd75cf1c`, +SHA-256 `da1990000c611d7377690c7526487d6415fa4f9cb42d299f59585a7c2eebf769`. +The helper's manifest data lookup uses the captured accepted bytes. Its SQL, +preconditions, complete-copy comparison and exact allowed row changes are unchanged. +This setup is separate from the producer-owned physical historical seed. + +```sh +/usr/bin/python3 -I -B -c "$BOOTSTRAP" "$REVIEWED_COMMIT" \ + scripts/conformance/command_keys/run_synthetic_legacy.py "$REVIEWED_SHA" \ + --runner-commit "$REVIEWED_COMMIT" \ + --state open --scenario readback \ + --wheel /path/to/downloaded/durable_workflow-2.17.2-py3-none-any.whl \ + --snapshot-helper /path/to/accepted/sqlite_legacy.py \ + --manifest /path/to/accepted/command-key-legacy-v1-vectors.json \ + --output /path/to/private-disk-scratch/new-synthetic-run +``` + +Run the readback scenario separately for `open`, `completed` and `continued`. +The actual installed Worker creates closed/continued states and null update +results before the copy. No Worker runs during the readback window. Four +concurrent changed-input retries must retain original identities and unknown +intent; changed names and explicit signal-run targets must conflict. The Python +update helper uses its existing unscoped route. GET-only outcomes preserve the +pending/completed-null distinction. + +The other scenarios require `--state open`: `two-rows`, `legacy-token` and +`modern-raw` must refuse with explicit ambiguous/unverified/non-retryable fields. +`case-only` first checks exact-byte retained readback, then admits the lower-case +key in a separate write interval under SQLite's binary comparison. Complete +logical snapshots must match for readback/ambiguity windows. The distinct-key +interval has no zero-write claim. Its complete logical delta must preserve every +old row except the exact source-defined sequence/summary advances, add the two +receipt-bound command/intent rows and corresponding signal/update/history/wait +records, and leave all other tables and columns unchanged. The installed sender +fingerprint, official codec decode and retained/conflict controls bind the new +rows to the two intended fresh admissions. The stopped transformed/copy snapshot +must also equal the first restarted snapshot before any readback begins. + +Every phase executes captured adapter/bootstrap bytes in the fixed published +Python image with all nine dependency wheel hashes and the reviewed SDK wheel. +Lifecycle and cleanup use captured full creation IDs plus ownership labels. +Source and copied runtimes are sequential; at most one runtime and one consumer +coexist, each capped at 1 CPU, 1 GiB and 256 PIDs. All declared inputs and the +interpreter are rechecked after return, and both private database copies are +removed after exact-ID container/volume cleanup. Record each state/scenario and +any NONPASS separately. Independent review, exact-head checks and actual +installed evidence are required before qualification; another SDK's result does +not qualify this adapter or complete shared216. + +Physical and synthetic SQLite qualification is currently **HOLD**. SQLite's +native VFS can canonicalize even a real proc-FD URI and reopen its mutable +pathname. Historical business observations using that boundary remain useful, +but cannot qualify the +physical or synthetic cells. The separately qualified response-loss cell does +not read SQLite files and keeps its existing scope. + +`check_quiesced_snapshot.py` is a preparation-only synthetic control. It pauses +the captured owned writer ID, checks that state while reading held main/WAL/SHM +descriptors twice, and resumes the writer in `finally`, including on refusal. +Complete equality does not make an unquiesced byte capture a live snapshot. +The captured main/WAL bytes enter a fixed official interpreter container with +no host mounts or network. Official SQLite recovers them and backs up into +memory while root is the sole actor. The accepted snapshot helper consumes that +same memory connection after UID/GID/capability drop. The lower native-open +counterexample uses the exact C source from Rust PR265, commit701c17f. No +installed cell is authorized by these primitive controls. The physical runner +prepares this same reader for baseline, upgraded, pre-request and post-request +snapshots. Its schema marker checks consume the resulting typed snapshot rather +than opening a second host SQLite connection. That integration still needs +exact-source review and installed evidence. The synthetic runner's stopped +transform/copy path remains on hold. The stopped transform +helper's descriptor API is owned and reviewed separately in the Rust repository. diff --git a/scripts/conformance/command_keys/case_delta.py b/scripts/conformance/command_keys/case_delta.py new file mode 100644 index 0000000..d3c0f1b --- /dev/null +++ b/scripts/conformance/command_keys/case_delta.py @@ -0,0 +1,272 @@ +"""Exact logical delta for this fixture's two native SQLite fresh admissions. + +Derived from published5614274d runtime/{store,updates,wait_projection, +terminal_summary,command_context}.rs. No Worker executes in this interval. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from copy import deepcopy +from datetime import datetime, timezone + +ULID = re.compile(r"[0-9A-HJKMNP-TV-Z]{26}") + + +def value(cell): + if cell == ["null"]: + return None + if len(cell) != 2 or cell[0] not in {"text", "integer"}: + raise ValueError("unexpected case-delta field type") + return cell[1] + + +def typed(item): + if item is None: + return ["null"] + return ["integer" if type(item) is int else "text", item] + + +def timestamp(item): + parsed = datetime.fromisoformat(item.replace("Z", "+00:00")) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def rows(snapshot, table): + return [ + dict(zip(snapshot["tables"][table]["columns"], row, strict=True)) for row in snapshot["tables"][table]["rows"] + ] + + +def one(snapshot, table, field, expected): + found = [row for row in rows(snapshot, table) if value(row[field]) == expected] + if len(found) != 1: + raise ValueError("case delta needs one exact " + table + " identity") + return found[0] + + +def stable_catalog(before, after): + if ( + before["schema"] != after["schema"] + or before["user_version"] != after["user_version"] + or before["application_id"] != after["application_id"] + or set(before["tables"]) != set(after["tables"]) + ): + raise ValueError("case interval changed schema or table inventory") + if any(before["tables"][name]["columns"] != after["tables"][name]["columns"] for name in before["tables"]): + raise ValueError("case interval changed columns") + + +def finish_expected(expected, after): + for table in expected["tables"].values(): + table["rows"].sort(key=lambda row: json.dumps(row, sort_keys=True, separators=(",", ":"), ensure_ascii=True)) + if expected != after: + raise ValueError("case interval changed rows outside the exact admission delta") + + +def validate_restart(stopped, restarted): + if stopped != restarted: + raise ValueError("first restarted snapshot differs from the stopped copy") + + +def validate(before, after, seed, proof): + stable_catalog(before, after) + expected = deepcopy(before) + run_id, workflow_id = seed["run_id"], seed["workflow_id"] + run = one(before, "workflow_runs", "id", run_id) + instance = one(before, "workflow_instances", "id", workflow_id) + first_sequence = value(run["last_history_sequence"]) + events = proof["events"] + cells = proof["cells"] + if len(events) != 2 or [cell["kind"] for cell in cells] != ["signal", "update"]: + raise ValueError("case interval lacks its two public admission proofs") + began, ended = timestamp(proof["started_at"]), timestamp(proof["finished_at"]) + + def valid_time(observed, authoritative): + return began <= timestamp(value(observed)) == timestamp(authoritative) <= ended + + def append_row(table, old, new, changes, predicates=None): + candidate = deepcopy(old) + for field, item in changes.items(): + candidate[field] = typed(item) + for field, predicate in (predicates or {}).items(): + if not predicate(new[field]): + raise ValueError("case admission field differs: " + table + "." + field) + candidate[field] = new[field] + if candidate != new: + raise ValueError("case admission changed unexpected new-row fields: " + table) + expected["tables"][table]["rows"].append([new[column] for column in expected["tables"][table]["columns"]]) + + def update_counter(table, identity, increments): + original = one(before, table, "id", identity) + target = expected["tables"][table] + for row in target["rows"]: + if row[target["columns"].index("id")] == original["id"]: + for field, amount in increments.items(): + row[target["columns"].index(field)] = typed(value(original[field]) + amount) + return + raise ValueError("missing original sequence row") + + event_bytes = 0 + for offset, (cell, event) in enumerate(zip(cells, events, strict=True), start=1): + kind = cell["kind"] + original = seed["commands"][kind]["original"] + receipt = cell["receipt"] + command_id = receipt["command_id"] + if command_id == original["command_id"] or receipt["run_id"] != run_id: + raise ValueError("case-distinct receipt selected an original or foreign run") + payload = event["payload"] + command = payload["command"] + command_sequence = value(run["last_command_sequence"]) + offset + message_sequence = value(instance["last_message_sequence"]) + offset + if ( + event["sequence"] != first_sequence + offset + or payload["workflow_command_id"] != command_id + or command["id"] != command_id + or command["sequence"] != command_sequence + or command["message_sequence"] != message_sequence + ): + raise ValueError("new history sequence/receipt binding differs") + old_command = one(before, "workflow_commands", "id", original["command_id"]) + new_command = one(after, "workflow_commands", "id", command_id) + old_context = json.loads(value(old_command["context"])) + new_context = json.loads(value(new_command["context"])) + operation = new_context["server"]["operation_id"] + if not ULID.fullmatch(operation) or operation == old_context["server"]["operation_id"]: + raise ValueError("new admission operation identity is invalid") + context = deepcopy(old_context) + context["server"]["operation_id"] = operation + context["server"]["metadata"]["request_id"] = "casesensitive" + context["request"]["request_id"] = "casesensitive" + context["request"]["fingerprint"] = cell["request_fingerprint"] + if new_context != context: + raise ValueError("case interval changed unexpected command attribution") + if command["request_fingerprint"] != cell["request_fingerprint"]: + raise ValueError("stored request fingerprint differs from installed sender") + append_row( + "workflow_commands", + old_command, + new_command, + { + "id": command_id, + "request_id": kind + "-v1:" + hashlib.sha256(b"casesensitive").hexdigest(), + "command_sequence": command_sequence, + "message_sequence": message_sequence, + "payload": command["payload"]["blob"], + "context": value(new_command["context"]), + }, + {"accepted_at": lambda actual, at=command["accepted_at"]: valid_time(actual, at)}, + ) + intent = one(after, "dw_command_intents", "command_id", command_id) + if ( + set(intent) != {"command_id", "namespace", "intent_digest"} + or value(intent["namespace"]) != value(instance["namespace"]) + or re.fullmatch(r"[0-9a-f]{64}", value(intent["intent_digest"])) is None + or cell["same_intent_retained"] is not True + or cell["changed_intent_conflict"] is not True + ): + raise ValueError("new intent row is not bound to the verified fresh admission") + expected["tables"]["dw_command_intents"]["rows"].append( + [intent[column] for column in expected["tables"]["dw_command_intents"]["columns"]] + ) + old_event = one(before, "workflow_history_events", "workflow_command_id", original["command_id"]) + new_event = one(after, "workflow_history_events", "workflow_command_id", command_id) + if not ULID.fullmatch(value(new_event["id"])) or new_event["id"] == old_event["id"]: + raise ValueError("invalid new history identity") + old_payload = json.loads(value(old_event["payload"])) + desired = deepcopy(old_payload) + desired.update(workflow_command_id=command_id, arguments=payload["arguments"], command=command) + if kind == "signal": + desired.update(signal_id=payload["signal_id"], signal_wait_id="signal-command:" + command_id) + else: + desired.update(update_id=receipt["update_id"], request_id="casesensitive") + if desired != payload or json.loads(value(new_event["payload"])) != payload: + raise ValueError("new history payload differs from the declared source-derived event") + append_row( + "workflow_history_events", + old_event, + new_event, + { + "id": value(new_event["id"]), + "sequence": first_sequence + offset, + "workflow_command_id": command_id, + "payload": value(new_event["payload"]), + }, + {"recorded_at": lambda actual, at=event["timestamp"]: valid_time(actual, at)}, + ) + # This fixture uses only ASCII strings, small integers, booleans/null + # and opaque codec strings. Native's PHP-shaped event sizing is ordinary + # compact JSON for these values (no numeric-key maps/floats/RawValue). + event_bytes += len(event["event_type"].encode()) + len( + json.dumps(payload, separators=(",", ":"), ensure_ascii=True).encode() + ) + if kind == "signal": + old = one(before, "workflow_signal_records", "workflow_command_id", original["command_id"]) + new = one(after, "workflow_signal_records", "workflow_command_id", command_id) + append_row( + "workflow_signal_records", + old, + new, + { + "id": payload["signal_id"], + "workflow_command_id": command_id, + "command_sequence": command_sequence, + "signal_wait_id": "signal-command:" + command_id, + "arguments": payload["arguments"]["blob"], + }, + {"received_at": lambda actual, at=command["accepted_at"]: valid_time(actual, at)}, + ) + else: + old = one(before, "workflow_updates", "id", original["update_id"]) + new = one(after, "workflow_updates", "id", receipt["update_id"]) + append_row( + "workflow_updates", + old, + new, + { + "id": receipt["update_id"], + "workflow_command_id": command_id, + "command_sequence": command_sequence, + "arguments": payload["arguments"]["blob"], + }, + {"accepted_at": lambda actual, at=receipt["accepted_at"]: valid_time(actual, at)}, + ) + old_wait = one(before, "workflow_run_waits", "resume_source_id", original["update_id"]) + wait_id = "update:" + receipt["update_id"] + new_wait = one(after, "workflow_run_waits", "resume_source_id", receipt["update_id"]) + append_row( + "workflow_run_waits", + old_wait, + new_wait, + { + "id": hashlib.sha256((run_id + "|" + wait_id).encode()).hexdigest(), + "wait_id": wait_id, + "position": len(rows(before, "workflow_run_waits")), + "resume_source_id": receipt["update_id"], + "command_id": command_id, + "command_sequence": command_sequence, + }, + { + field: (lambda actual, at=event["timestamp"]: valid_time(actual, at)) + for field in ("opened_at", "created_at", "updated_at") + }, + ) + update_counter("workflow_runs", run_id, {"last_command_sequence": 2, "last_history_sequence": 2}) + update_counter("workflow_instances", workflow_id, {"last_message_sequence": 2}) + if any(value(row["id"]) == run_id for row in rows(before, "workflow_run_summaries")): + update_counter("workflow_run_summaries", run_id, {"history_event_count": 2, "history_size_bytes": event_bytes}) + finish_expected(expected, after) + return { + "new_commands": 2, + "new_intents": 2, + "new_signals": 1, + "new_updates": 1, + "new_history_events": 2, + "new_waits": 1, + "history_bytes_added": event_bytes, + "all_other_rows_preserved": True, + } diff --git a/scripts/conformance/command_keys/check_quiesced_snapshot.py b/scripts/conformance/command_keys/check_quiesced_snapshot.py new file mode 100644 index 0000000..fd6981d --- /dev/null +++ b/scripts/conformance/command_keys/check_quiesced_snapshot.py @@ -0,0 +1,300 @@ +"""Isolated synthetic controls for paused held-FD capture and private recovery. + +This command runs no SDK, Server, historical seed or installed conformance cell. +""" + +from __future__ import annotations + +import argparse +import base64 +import json +import os +import subprocess +import time +from contextlib import ExitStack +from datetime import datetime, timezone +from pathlib import Path +from types import ModuleType +from uuid import uuid4 + +HELPER_SHA256 = "6fc6054ce087c136a86d7b6b222daeefaa87c58fa0b613b155a5df39e6a26fad" +NATIVE_PROBE_SHA256 = "367b3c9ac78f16ff17fc8796763a2844ea332fddb6aea4cbc1ef3958ac47dfff" +WRITER = """import sqlite3,time +path='/fixture/sample.sqlite' +first=sqlite3.connect(path) +first.execute('CREATE TABLE sample(value INTEGER)') +first.execute('INSERT INTO sample VALUES (0)') +first.commit() +first.execute('PRAGMA journal_mode=WAL') +first.execute('PRAGMA wal_autocheckpoint=0') +second=sqlite3.connect(path) +second.execute('PRAGMA wal_autocheckpoint=0') +for n in range(1,601): + connection=first if n%2 else second + connection.execute('INSERT INTO sample VALUES (?)',(n,)) + connection.commit() + print(n,flush=True) + time.sleep(0.1) +""" + + +def module(data, name, filename): + result = ModuleType(name) + result.__file__ = filename + exec(compile(data, filename, "exec"), result.__dict__) + return result + + +def main(args): + root = Path(__file__).absolute().parents[3] + relative = "scripts/conformance/command_keys/run_response_loss.py" + source = subprocess.check_output( + ["/usr/bin/git", "-C", str(root), "show", args.runner_commit + ":" + relative], + env={"PATH": os.defpath}, + timeout=10, + ) + controls = module(source, "captured_controls", str(root / relative)) + with ExitStack() as stack: + inputs = controls.captured_inputs( + args, + stack, + source_files={ + "controller": "check_quiesced_snapshot.py", + "capture": "quiesced_capture.py", + "snapshot_bootstrap": "snapshot_bootstrap.py", + "native_control": "native_open_capture_control.py", + "snapshot_reader": "snapshot_runtime.py", + }, + data_files={ + "helper": (args.helper, 128 * 1024, HELPER_SHA256), + "native_probe": (args.native_probe_source, 64 * 1024, NATIVE_PROBE_SHA256), + }, + ) + workspace = stack.enter_context(controls.workspace_for(inputs, args.output)) + database = workspace.directory("database") + held = workspace.write("sample.sqlite", b"", parent=database, immutable=False) + nonce = uuid4().hex + environment = {"PATH": os.defpath, "TMPDIR": workspace.proc(workspace.root)} + + def cleanup_call(*arguments, check=True, timeout=45): + return subprocess.run( + ["docker", "--config", "/proc/self/fd", *arguments], + env=environment, + capture_output=True, + text=True, + check=check, + timeout=timeout, + ) + + def call(*arguments, check=True, timeout=45): + workspace.check() + result = cleanup_call(*arguments, check=check, timeout=timeout) + workspace.check() + return result + + image = "docker.io/library/python@" + controls.PYTHON_INDEX + interpreter = controls.interpreter_identity(json.loads(call("image", "inspect", image).stdout)) + report = { + "runner_commit": args.runner_commit, + "started_at": datetime.now(timezone.utc).isoformat(), + "scope": "isolated synthetic snapshot primitive only", + "interpreter": interpreter, + "inputs": {role: value.sha256 for role, value in inputs.items()}, + "owned": {}, + "cleanup": {}, + "complete": False, + } + report["owned_containers"] = {} + report["cleanup_by_id"] = report["cleanup"] + reader_module = module( + inputs["snapshot_reader"].data, "captured_snapshot_reader", "captured-snapshot-runtime.py" + ) + reader_inputs = {**inputs, "snapshot_helper": inputs["helper"]} + reader = reader_module.SnapshotReader( + workspace, + reader_inputs, + controls, + call, + cleanup_call, + report, + nonce, + lambda: [ + identity + for role, identity in report["owned"].items() + if role == "writer" and role not in report["cleanup"] + ], + ) + common = [ + "--init", + "--read-only", + "--network", + "none", + "--cpus", + "1", + "--label", + controls.OWNER_LABEL + "=sdk-python", + "--label", + controls.RUN_LABEL + "=" + nonce, + "--security-opt", + "no-new-privileges", + "--cap-drop", + "ALL", + ] + + def create(role, *arguments): + identity = controls.created_id( + call("create", "--name", "sdk-python-snapshot-" + role + "-" + nonce[:12], *common, *arguments).stdout + ) + report["owned"][role] = identity + report["owned_containers"][role] = {"id": identity} + if controls.inspect_owned(call, identity, nonce) is None: + raise ValueError("created snapshot control disappeared") + return identity + + def cleanup(role): + report["cleanup"][role] = controls.cleanup_owned(cleanup_call, report["owned"][role], nonce) + + def recover(role, files): + return reader.recover(files, {"primitive_role": role}) + + try: + writer = create( + "writer", + "--memory", + "512m", + "--pids-limit", + "64", + "--user", + str(os.getuid()) + ":" + str(os.getgid()), + "--mount", + "type=bind,src=" + workspace.mount_path(database) + ",dst=/fixture", + "--entrypoint", + "/usr/local/bin/python3", + image, + "-I", + "-B", + "-c", + WRITER, + ) + call("start", writer) + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + lines = call("logs", writer).stdout.splitlines() + if len(lines) >= 3: + break + time.sleep(0.1) + else: + raise TimeoutError("synthetic multi-connection WAL writer readiness") + files, report["capture"] = reader.capture_bytes(held) + if not files["wal"] or not files["shm"]: + raise ValueError("positive WAL control has no WAL/SHM") + report["paused_writer_verified"] = True + + def refuse_capture(*_args): + raise ValueError("injected refusal") + + actual_capture = reader.capture.capture + reader.capture.capture = refuse_capture + try: + try: + reader.capture_bytes(held) + except ValueError as error: + if str(error) != "injected refusal": + raise + else: + raise ValueError("injected capture refusal did not fire") + finally: + reader.capture.capture = actual_capture + state = controls.inspect_owned(call, writer, nonce) + if state is None or state["State"]["Paused"] or not state["State"]["Running"]: + raise ValueError("writer did not resume after refused capture") + previous = call("logs", writer).stdout.splitlines() + deadline = time.monotonic() + 3 + while time.monotonic() < deadline: + if len(call("logs", writer).stdout.splitlines()) > len(previous): + report["resume_after_refusal_verified"] = True + break + time.sleep(0.1) + else: + raise ValueError("writer made no progress after refused capture") + cleanup("writer") + with_wal = recover("with-wal", files) + main_only = recover("main-only", {"main": files["main"], "wal": b""}) + values = with_wal["snapshot"]["tables"]["sample"]["rows"] + integers = sorted(row[0][1] for row in values) + if integers != list(range(len(integers))) or len(integers) < 4: + raise ValueError("private recovery lost committed WAL rows") + if main_only["snapshot"]["tables"]["sample"]["rows"] != [[["integer", 0]]]: + raise ValueError("WAL positive did not distinguish the main file") + report["wal_committed_rows"] = len(integers) - 1 + report["main_only_rows"] = 1 + report["private_recovery"] = {key: value for key, value in with_wal.items() if key != "snapshot"} + # Compile the exact public Rust265 C probe from captured bytes, + # then keep the resulting library FD through child loader startup. + library = workspace.path / "native-open-race.so" + subprocess.run( + ["/usr/bin/cc", "-shared", "-fPIC", "-x", "c", "-", "-ldl", "-o", str(library)], + input=inputs["native_probe"].data, + env=environment, + check=True, + capture_output=True, + timeout=20, + ) + library.chmod(0o600) + held_library = workspace.capture(workspace.root, library.name, immutable=True, maximum=1024 * 1024) + native_inputs = {} + for role in ("workspace", "capture"): + native_inputs[role] = { + "sha256": inputs[role].sha256, + "base64": base64.b64encode(inputs[role].data).decode(), + } + native_result = subprocess.run( + [ + "/usr/bin/python3", + "-I", + "-B", + "-c", + inputs["native_control"].data.decode(), + str(workspace.path / "native-control"), + ], + input=json.dumps(native_inputs), + env={**environment, "LD_PRELOAD": "/proc/self/fd/" + str(held_library["fd"])}, + pass_fds=(held_library["fd"],), + capture_output=True, + text=True, + timeout=15, + ) + workspace.check() + if native_result.returncode or "intercepted native pathname open" not in native_result.stderr: + raise RuntimeError("lower native-open control did not reproduce") + native = json.loads(native_result.stdout) + native_files = {name: base64.b64decode(data, validate=True) for name, data in native.pop("files").items()} + native_recovery = recover("native-capture", native_files) + if native_recovery["snapshot"]["tables"]["sample"]["rows"] != [[["text", "original"]]]: + raise ValueError("captured native-race input was substituted") + report["native_open_control"] = native + report["native_probe_source_sha256"] = NATIVE_PROBE_SHA256 + report["native_probe_library_sha256"] = held_library["sha256"] + for value in inputs.values(): + value.recheck() + report["complete"] = True + finally: + for role, entry in report["owned_containers"].items(): + if role not in report["cleanup"]: + report["cleanup"][role] = controls.cleanup_owned(cleanup_call, entry["id"], nonce) + report["finished_at"] = datetime.now(timezone.utc).isoformat() + report["complete"] = report["complete"] and all( + value["absent"] and value["volumes_absent"] and value["oom_killed"] is False + for value in report["cleanup"].values() + ) + workspace.remove_database(database) + workspace.write("primitive.json", json.dumps(report, indent=2) + "\n") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--runner-commit", required=True) + parser.add_argument("--helper", required=True, type=Path) + parser.add_argument("--native-probe-source", required=True, type=Path) + parser.add_argument("--output", required=True, type=Path) + main(parser.parse_args()) diff --git a/scripts/conformance/command_keys/consumer_bootstrap.py b/scripts/conformance/command_keys/consumer_bootstrap.py index 60a5d13..93f3edc 100644 --- a/scripts/conformance/command_keys/consumer_bootstrap.py +++ b/scripts/conformance/command_keys/consumer_bootstrap.py @@ -37,7 +37,11 @@ def approved_environment_link(relative, target): def decode_inputs(bundle): scenario = bundle.get("scenario", "response_loss") - roles = ROLES if scenario == "response_loss" else PHYSICAL_ROLES if scenario == "physical_readback" else None + roles = ( + ROLES if scenario == "response_loss" + else PHYSICAL_ROLES if scenario in {"physical_readback", "synthetic_legacy"} + else None + ) if roles is None or set(bundle["sources"]) != roles: raise ValueError("captured source roles differ") result = {} diff --git a/scripts/conformance/command_keys/native_open_capture_control.py b/scripts/conformance/command_keys/native_open_capture_control.py new file mode 100644 index 0000000..04d884c --- /dev/null +++ b/scripts/conformance/command_keys/native_open_capture_control.py @@ -0,0 +1,78 @@ +"""Run only under Rust265's frozen native-open interposer, on synthetic files.""" + +from __future__ import annotations + +import base64 +import ctypes +import hashlib +import json +import os +import sqlite3 +import sys +from contextlib import closing +from pathlib import Path +from types import ModuleType + + +def load(item, name): + data = base64.b64decode(item["base64"], validate=True) + if hashlib.sha256(data).hexdigest() != item["sha256"]: + raise ValueError("native control captured module differs") + result = ModuleType(name) + exec(compile(data, name + ".py", "exec"), result.__dict__) + return result + + +def main(): + packet = json.load(sys.stdin) + guard = load(packet["workspace"], "captured_workspace") + capture = load(packet["capture"], "captured_capture") + with guard.Workspace(Path(sys.argv[1])) as workspace: + directory = workspace.directory("database") + held = workspace.write("sample.sqlite", b"", parent=directory, immutable=False) + path = workspace.path / "database/sample.sqlite" + for candidate, value in ((path, "original"), (Path(str(path) + ".replacement"), "replacement")): + with closing(sqlite3.connect(candidate)) as connection: + connection.execute("CREATE TABLE sample(value TEXT)") + connection.execute("INSERT INTO sample VALUES (?)", (value,)) + connection.commit() + candidate.chmod(0o600) + before = os.fstat(held["fd"]) + library = ctypes.CDLL(None) + library.getenv.restype = ctypes.c_char_p + os.environ["SQLITE_PROBE_TARGET"] = str(path) + os.environ["SQLITE_PROBE_ARMED"] = "1" + # Reproduce the lower VFS interval missed by a Python connect wrapper. + with closing(sqlite3.connect("file:/proc/self/fd/" + str(held["fd"]) + "?mode=ro", uri=True)) as connection: + observed = connection.execute("SELECT value FROM sample").fetchone()[0] + workspace.check() + after = os.fstat(held["fd"]) + if ( + observed != "replacement" + or library.getenv(b"SQLITE_PROBE_ARMED") is not None + or (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino) + ): + raise ValueError("native-open predecessor counterexample did not reproduce") + # Both fixture connections are closed; this child is the only fixture + # writer. Keep the lower native-open attack armed through actual capture. + os.environ["SQLITE_PROBE_ARMED"] = "1" + files, proof = capture.capture(workspace, held, lambda: None) + if library.getenv(b"SQLITE_PROBE_ARMED") != b"1": + raise ValueError("held capture reopened the mutable main pathname") + os.environ.pop("SQLITE_PROBE_ARMED") + print( + json.dumps( + { + "predecessor_observed": observed, + "held_inode_preserved": True, + "restored_basename_verified": True, + "native_open_not_called_by_capture": True, + "proof": proof, + "files": {name: base64.b64encode(data).decode() for name, data in files.items()}, + } + ) + ) + + +if __name__ == "__main__": + main() diff --git a/scripts/conformance/command_keys/quiesced_capture.py b/scripts/conformance/command_keys/quiesced_capture.py new file mode 100644 index 0000000..26b8b8c --- /dev/null +++ b/scripts/conformance/command_keys/quiesced_capture.py @@ -0,0 +1,96 @@ +"""Bounded held-FD capture after the controller proves every writer quiescent.""" + +from __future__ import annotations + +import hashlib +import os +import stat + +MAX_FILE = 64 * 1024 * 1024 + + +def identity(info): + return ( + info.st_dev, + info.st_ino, + info.st_mode, + info.st_uid, + info.st_gid, + info.st_nlink, + info.st_size, + info.st_mtime_ns, + info.st_ctime_ns, + ) + + +def capture(workspace, database, attest_quiescent): + """The callback must independently recheck the exact stopped/paused writers. + + Equality alone is not a live snapshot protocol. This function never asks + SQLite to open any host pathname, including a proc-FD URI. + """ + workspace.check() + if workspace.files.get((database["parent"], database["name"])) is not database: + raise ValueError("capture database is not held by this workspace") + descriptors = {"main": database["fd"]} + opened = [] + sidecars = {} + try: + attest_quiescent() + for suffix in ("-wal", "-shm", "-journal"): + name = database["name"] + suffix + try: + descriptor = os.open(name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=database["parent"]) + except FileNotFoundError: + sidecars[suffix] = None + continue + opened.append(descriptor) + info = os.fstat(descriptor) + if ( + not stat.S_ISREG(info.st_mode) + or info.st_uid != os.getuid() + or info.st_nlink != 1 + or info.st_mode & 0o077 + or info.st_size > MAX_FILE + ): + raise ValueError("capture sidecar is not a bounded private regular file") + sidecars[suffix] = (descriptor, identity(info)) + if suffix == "-journal": + if info.st_size: + raise ValueError("rollback journal requires a separate recovery contract") + else: + descriptors[suffix[1:]] = descriptor + + def recheck(): + attest_quiescent() + workspace.check() + for suffix, held in sidecars.items(): + name = database["name"] + suffix + try: + current = os.stat(name, dir_fd=database["parent"], follow_symlinks=False) + except FileNotFoundError: + if held is not None: + raise ValueError("capture sidecar vanished") from None + else: + if held is None or identity(current) != held[1] or identity(os.fstat(held[0])) != held[1]: + raise ValueError("capture sidecar identity changed") + + before = {role: identity(os.fstat(descriptor)) for role, descriptor in descriptors.items()} + if not 0 < before["main"][6] <= MAX_FILE: + raise ValueError("capture main database exceeds its bound") + first = {role: os.pread(descriptor, MAX_FILE + 1, 0) for role, descriptor in descriptors.items()} + recheck() + second = {role: os.pread(descriptor, MAX_FILE + 1, 0) for role, descriptor in descriptors.items()} + recheck() + if first != second or any(identity(os.fstat(fd)) != before[role] for role, fd in descriptors.items()): + raise ValueError("quiescent complete byte captures differ") + if any(len(first[role]) != before[role][6] for role in descriptors): + raise ValueError("quiescent byte capture is incomplete") + first.setdefault("wal", b"") + return first, { + role: {"identity": before.get(role), "bytes": len(data), "sha256": hashlib.sha256(data).hexdigest()} + for role, data in first.items() + } + finally: + for descriptor in opened: + os.close(descriptor) diff --git a/scripts/conformance/command_keys/run_physical_readback.py b/scripts/conformance/command_keys/run_physical_readback.py index ce91df2..773a350 100644 --- a/scripts/conformance/command_keys/run_physical_readback.py +++ b/scripts/conformance/command_keys/run_physical_readback.py @@ -43,8 +43,12 @@ def load_bytes(data, name, expected, *, filename=None): return module -def native_marker(connection, expected): - if connection.execute("SELECT engine,version FROM dw_server_schema").fetchall() != [("rust-development", expected)]: +def native_marker(snapshot, expected): + marker = snapshot["tables"]["dw_server_schema"] + if set(marker["columns"]) != {"engine", "version"} or len(marker["rows"]) != 1: + raise ValueError("native schema ownership/version marker shape differs") + row = dict(zip(marker["columns"], marker["rows"][0], strict=True)) + if row != {"engine": ["text", "rust-development"], "version": ["integer", expected]}: raise ValueError("native schema ownership/version marker differs") @@ -110,9 +114,10 @@ def run_bound(args, inputs, controls, workspace): with closing(sqlite3.connect(":memory:")) as catalog: catalog.execute("CREATE TABLE dw_server_schema(engine TEXT, version INTEGER)") catalog.executescript(inputs["step_sql"].data.decode()) - added_schema = [row for row in legacy.snapshot(catalog)["schema"] if row[2] != "dw_server_schema"] + added_schema = json.loads( + json.dumps([row for row in legacy.snapshot(catalog)["schema"] if row[2] != "dw_server_schema"]) + ) database = workspace.directory("database") - copied = Path(workspace.proc(database, "upgraded.sqlite")) config = workspace.directory("docker-config") workspace.write("config.json", '{"auths":{}}\n', parent=config) environment = {"PATH": os.defpath, "TMPDIR": workspace.proc(workspace.root)} @@ -138,12 +143,7 @@ def cleanup_call(*arguments, check=True, timeout=45): ) def snapshot(): - workspace.check() - with closing(sqlite3.connect(copied.as_uri() + "?mode=ro", uri=True, timeout=2)) as connection: - connection.execute("BEGIN") - result = legacy.snapshot(connection) - workspace.check() - return result + return reader.snapshot(held_database) image = "ghcr.io/durable-workflow/server@" + controls.EXPECTED_SERVER["index"] python_image = "docker.io/library/python@" + controls.PYTHON_INDEX @@ -206,13 +206,30 @@ def create(role, *arguments): raise ValueError("created object vanished") return identity + reader_module = load_bytes( + inputs["snapshot_reader"].data, "captured_snapshot_reader", inputs["snapshot_reader"].sha256 + ) + reader = reader_module.SnapshotReader( + workspace, + inputs, + controls, + call, + cleanup_call, + report, + nonce, + lambda: [ + entry["id"] + for role, entry in report["owned_containers"].items() + if role in {"upgrade", "runtime"} and role not in report["cleanup_by_id"] + ], + ) + try: for value in inputs.values(): value.recheck() held_database = workspace.write("upgraded.sqlite", inputs["baseline"].data, parent=database, immutable=False) - with closing(sqlite3.connect(copied)) as connection: - native_marker(connection, 12) baseline = snapshot() + native_marker(baseline, 12) began = time.time() upgrade_id = create( "upgrade", @@ -232,9 +249,9 @@ def create(role, *arguments): finished = time.time() report["upgrade"] = upgrade report["cleanup_by_id"]["upgrade"] = controls.cleanup_owned(cleanup_call, upgrade_id, nonce) - with closing(sqlite3.connect(copied)) as connection: - native_marker(connection, 13) - validate_upgrade(baseline, snapshot(), added_schema, began, finished) + upgraded = snapshot() + native_marker(upgraded, 13) + validate_upgrade(baseline, upgraded, added_schema, began, finished) report["upgrade_row_preservation"] = True report["journal_sha384"] = JOURNAL_SHA384 runtime_id = create( @@ -342,7 +359,9 @@ def create(role, *arguments): } report["finished_at"] = datetime.now(timezone.utc).isoformat() report["complete"] = ( - len(report["cleanup_by_id"]) == 3 + {"upgrade", "runtime", "consumer"}.issubset(report["cleanup_by_id"]) + and len(report.get("snapshots", [])) == 4 + and len(report["cleanup_by_id"]) == 7 and clean and report.get("consumer_complete") is True and all(value.get("oom_killed") is False for value in report["cleanup_by_id"].values()) @@ -370,6 +389,9 @@ def run(args): "support": "run_response_loss.py", "adapter": "physical_readback.py", "verifier": "response_loss.py", + "capture": "quiesced_capture.py", + "snapshot_bootstrap": "snapshot_bootstrap.py", + "snapshot_reader": "snapshot_runtime.py", }, data_files={ "wheel": (args.wheel, 4 * 1024 * 1024, controls.WHEEL_SHA256), diff --git a/scripts/conformance/command_keys/run_response_loss.py b/scripts/conformance/command_keys/run_response_loss.py index df620a5..9d9d5d7 100644 --- a/scripts/conformance/command_keys/run_response_loss.py +++ b/scripts/conformance/command_keys/run_response_loss.py @@ -165,7 +165,7 @@ def captured_inputs(args, stack, *, source_files=None, data_files=None): def consumer_result(stdout, inputs, *, scenario="response_loss"): - if scenario not in {"response_loss", "physical_readback"}: + if scenario not in {"response_loss", "physical_readback", "synthetic_legacy"}: raise ValueError("unknown installed consumer scenario") lines = [line[len(RESULT_PREFIX) :] for line in stdout.splitlines() if line.startswith(RESULT_PREFIX)] if len(lines) != 1: @@ -182,7 +182,7 @@ def consumer_result(stdout, inputs, *, scenario="response_loss"): raise ValueError("consumer result does not bind the captured inputs/toolchain") if scenario == "response_loss" and result.get("proxy_sha256") != PROXY_SHA256: raise ValueError("consumer proxy identity differs") - if scenario == "physical_readback" and result.get("seed_sha256") != inputs["seed"].sha256: + if scenario in {"physical_readback", "synthetic_legacy"} and result.get("seed_sha256") != inputs["seed"].sha256: raise ValueError("consumer seed identity differs") return result diff --git a/scripts/conformance/command_keys/run_synthetic_legacy.py b/scripts/conformance/command_keys/run_synthetic_legacy.py new file mode 100644 index 0000000..1d7298c --- /dev/null +++ b/scripts/conformance/command_keys/run_synthetic_legacy.py @@ -0,0 +1,399 @@ +"""Run one synthetic SQLite legacy cell using the accepted unchanged helper.""" + +from __future__ import annotations + +import argparse +import base64 +import hashlib +import json +import os +import re +import secrets +import sqlite3 +import subprocess +import time +import urllib.request +from contextlib import ExitStack, closing +from datetime import datetime, timezone +from pathlib import Path +from types import ModuleType, SimpleNamespace +from uuid import uuid4 + +HELPER_SHA256 = "6fc6054ce087c136a86d7b6b222daeefaa87c58fa0b613b155a5df39e6a26fad" +MANIFEST_SHA256 = "da1990000c611d7377690c7526487d6415fa4f9cb42d299f59585a7c2eebf769" +SCENARIOS = {"readback", "two-rows", "legacy-token", "modern-raw", "case-only"} + + +def load(data, name, filename): + module = ModuleType(name) + module.__file__ = filename + exec(compile(data, filename, "exec"), module.__dict__) + return module + + +def controls_from_git(args): + if re.fullmatch(r"[0-9a-f]{40}", args.runner_commit) is None: + raise ValueError("require an exact runner commit") + root = Path(__file__).absolute().parents[3] + relative = "scripts/conformance/command_keys/run_response_loss.py" + data = subprocess.check_output( + ["git", "-C", str(root), "show", args.runner_commit + ":" + relative], env={"PATH": os.defpath}, timeout=10 + ) + return load(data, "reviewed_synthetic_controls", str(root / relative)) + + +def accepted_helper(inputs): + helper = inputs["snapshot_helper"].data + manifest = inputs["manifest"].data + if hashlib.sha256(helper).hexdigest() != HELPER_SHA256 or hashlib.sha256(manifest).hexdigest() != MANIFEST_SHA256: + raise ValueError("accepted helper/manifest differs") + module = load(helper, "accepted_synthetic_helper", "captured-sqlite-legacy.py") + # The unchanged helper's manifest() opens its sibling file. Bind that single + # data lookup to the captured accepted bytes, avoiding any pathname reopen. + module.manifest = lambda: json.loads(manifest) + return module + + +def validate_cell(state, scenario): + if state not in {"open", "completed", "continued"} or scenario not in SCENARIOS: + raise ValueError("unknown synthetic cell") + if state != "open" and scenario != "readback": + raise ValueError("ambiguity/case scenarios require their own open cell") + + +def run_bound(args, inputs, controls, workspace): + validate_cell(args.state, args.scenario) + legacy = accepted_helper(inputs) + delta = load(inputs["delta"].data, "captured_case_delta", "captured-case-delta.py") + database_fd = workspace.directory("database") + database = Path(workspace.proc(database_fd)) + canonical_database = Path(workspace.mount_path(database_fd)) + held_databases = {} + config = workspace.directory("docker-config") + workspace.write("config.json", '{"auths":{}}\n', parent=config) + environment = {"PATH": os.defpath, "TMPDIR": workspace.proc(workspace.root)} + docker = ["docker", "--config", workspace.proc(config)] + + def call(*arguments, check=True, timeout=45): + workspace.check() + result = subprocess.run( + docker + list(arguments), env=environment, check=check, capture_output=True, text=True, timeout=timeout + ) + if arguments[0] != "create": + workspace.check() + return result + + def cleanup_call(*arguments, check=True, timeout=45): + return subprocess.run( + ["docker", "--config", "/proc/self/fd", *arguments], + env=environment, + check=check, + capture_output=True, + text=True, + timeout=timeout, + ) + + image = "ghcr.io/durable-workflow/server@" + controls.EXPECTED_SERVER["index"] + python_image = "docker.io/library/python@" + controls.PYTHON_INDEX + observed = controls.image_identity(json.loads(call("image", "inspect", image).stdout)) + interpreter = controls.interpreter_identity(json.loads(call("image", "inspect", python_image).stdout)) + nonce = uuid4().hex + token = secrets.token_hex(24) + credential = workspace.write( + "fixture.env", + "DW_AUTH_TOKEN=" + + token + + "\nDW_RUNTIME_CREDENTIALS_ENABLED=false\nDW_RUST_MAINTENANCE_LOOP=0\nDB_CONNECTION=sqlite\n", + ) + report = { + "runner_commit": args.runner_commit, + "started_at": datetime.now(timezone.utc).isoformat(), + "scope": "synthetic legacy " + args.scenario + " " + args.state, + "expected_server": controls.EXPECTED_SERVER, + "observed_image": observed, + "interpreter_image": interpreter, + "helper_source": "4add871b175517f2031f71c868ce278bbfbba486", + "manifest_source": "d66e33bfb5c76f94c5b202f8c9f7a9e8cd75cf1c", + "ownership_nonce": nonce, + "inputs": {role: {"sha256": item.sha256, "bytes": len(item.data)} for role, item in inputs.items()}, + "owned_containers": {}, + "cleanup_by_id": {}, + "complete": False, + } + + def bind_id(role, identity): + if identity in [entry["id"] for entry in report["owned_containers"].values()]: + raise ValueError("duplicate container identity") + report["owned_containers"][role] = {"id": identity} + if controls.inspect_owned(call, identity, nonce) is None: + raise ValueError("created object vanished") + return identity + + def start(role, filename): + identity = bind_id( + role, + controls.created_id( + call( + "create", + "--name", + "sdk-python-216-synthetic-" + role + "-" + nonce[:12], + "--label", + controls.OWNER_LABEL + "=sdk-python", + "--label", + controls.RUN_LABEL + "=" + nonce, + "--init", + "--read-only", + "--cpus", + "1", + "--memory", + "1g", + "--pids-limit", + "256", + "--tmpfs", + "/tmp:mode=1777,size=128m", + "--tmpfs", + "/app/storage:uid=1000,gid=1000,mode=0700,size=64m", + "--tmpfs", + "/app/database:uid=1000,gid=1000,mode=0700,size=16m", + "-v", + str(canonical_database) + ":/cohort:rw", + "--env-file", + workspace.proc(credential["fd"]), + "-e", + "DB_DATABASE=/cohort/" + filename, + "-p", + "127.0.0.1::8080", + "--entrypoint", + "/bin/sh", + image, + *workspace.native_arguments(database_fd, held_databases[filename]), + ).stdout + ), + ) + call("start", identity) + address = call("port", identity, "8080/tcp").stdout.strip() + host, port = address.split(":") + if host != "127.0.0.1" or not port.isdigit() or not 1 <= int(port) <= 65535: + raise ValueError("runtime is not numeric loopback") + upstream = "http://" + address + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + deadline = time.monotonic() + 45 + while time.monotonic() < deadline: + try: + with opener.open(upstream + "/api/health", timeout=2) as response: + if response.status == 200: + return identity, upstream + except OSError: + time.sleep(0.2) + raise TimeoutError("synthetic runtime readiness") + + def stop(role): + identity = report["owned_containers"][role]["id"] + if controls.inspect_owned(call, identity, nonce) is None: + raise ValueError("owned runtime vanished before stopped-copy attestation") + call("stop", "--time", "10", identity) + state = controls.inspect_owned(call, identity, nonce) + if state is None or state["State"]["Running"] or state["State"]["ExitCode"] != 0: + raise ValueError("owned runtime did not stop cleanly") + report["cleanup_by_id"][role] = controls.cleanup_owned(cleanup_call, identity, nonce) + + def snapshot(path): + return workspace.read_database(held_databases[path.name], legacy.snapshot) + + def consume(phase, upstream, seed=None): + for value in inputs.values(): + value.recheck() + payload = { + "schema": "sdk-python.synthetic-legacy-input.v1", + "manifest_sha256": MANIFEST_SHA256, + "phase": phase, + "state": args.state, + "scenario": args.scenario, + } + if seed is not None: + payload["seed"] = seed + data = json.dumps(payload, separators=(",", ":")).encode() + captured = { + role: value + for role, value in inputs.items() + if role in {"adapter", "bootstrap", "verifier", "wheel", "requirements"} + } + captured["seed"] = SimpleNamespace(data=data, sha256=hashlib.sha256(data).hexdigest()) + bundle = { + "scenario": "synthetic_legacy", + "upstream": upstream, + "token": token, + "server_version": controls.EXPECTED_SERVER["version"], + "sources": { + role: {"sha256": value.sha256, "base64": base64.b64encode(value.data).decode()} + for role, value in captured.items() + }, + } + identity = bind_id( + phase, controls.create_consumer(call, "sdk-python-216-synthetic-" + phase + "-" + nonce[:12], nonce, inputs) + ) + result = subprocess.run( + docker + ["start", "-a", "-i", identity], + input=json.dumps(bundle), + env=environment, + capture_output=True, + text=True, + timeout=600, + ) + workspace.write(phase + ".log", result.stdout + result.stderr) + state = controls.inspect_owned(call, identity, nonce) + if state is None or state["State"]["Running"] or state["State"]["ExitCode"] != 0 or result.returncode != 0: + raise RuntimeError("installed synthetic consumer failed") + proof = controls.consumer_result(result.stdout, captured, scenario="synthetic_legacy") + workspace.write(phase + ".json", json.dumps(proof, indent=2) + "\n") + for value in inputs.values(): + value.recheck() + report["cleanup_by_id"][phase] = controls.cleanup_owned(cleanup_call, identity, nonce) + return proof + + try: + held_databases["source.sqlite"] = workspace.write("source.sqlite", b"", parent=database_fd, immutable=False) + _, upstream = start("source", "source.sqlite") + seed = consume("prepare", upstream)["seed"] + stop("source") + source, copied = database / "source.sqlite", database / "copied.sqlite" + stopped_source = snapshot(source) + if args.scenario == "modern-raw": + # No transform is appropriate for this ambiguity cell. SQLite backup + # still creates a separately owned complete copy of the stopped DB. + held_databases["copied.sqlite"] = workspace.write("copied.sqlite", b"", parent=database_fd, immutable=False) + workspace.check() + with ( + closing(sqlite3.connect(workspace.database_uri(held_databases["source.sqlite"]), uri=True)) as original, + closing( + sqlite3.connect(workspace.database_uri(held_databases["copied.sqlite"], "rw"), uri=True) + ) as destination, + ): + original.backup(destination) + if snapshot(source) != stopped_source or snapshot(copied) != stopped_source: + raise ValueError("unmodified modern copy differs") + report["transform"] = {"transformed": 0, "manifest_sha256": MANIFEST_SHA256} + else: + workspace.check() + report["transform"] = legacy.transform_copy( + canonical_database / source.name, canonical_database / copied.name, seed["selections"], stopped=True + ) + workspace.check() + held_databases["copied.sqlite"] = workspace.capture(database_fd, "copied.sqlite") + stopped_copy = snapshot(copied) + stopped_fingerprint = legacy.fingerprint(stopped_copy) + if args.scenario != "modern-raw" and report["transform"]["after_sha256"] != stopped_fingerprint: + raise ValueError("stopped copy no longer matches the accepted transform") + report["stopped_copy_sha256"] = stopped_fingerprint + workspace.write("stopped-copy.json", json.dumps(stopped_copy, sort_keys=True) + "\n") + _, upstream = start("readback", copied.name) + before = snapshot(copied) + delta.validate_restart(stopped_copy, before) + report["restarted_copy_sha256"] = legacy.fingerprint(before) + workspace.write("restarted-copy.json", json.dumps(before, sort_keys=True) + "\n") + proof = consume("read", upstream, seed) + after = snapshot(copied) + if before != after: + raise ValueError("legacy readback/ambiguity window changed complete logical rows/schema") + report["zero_write_sha256"] = legacy.fingerprint(before) + report["cells"] = proof["cells"] + if args.scenario == "case-only": + before = snapshot(copied) + distinct_proof = consume("case_distinct", upstream, seed) + report["distinct_cells"] = distinct_proof["cells"] + after = snapshot(copied) + workspace.write("case-before.json", json.dumps(before, sort_keys=True) + "\n") + workspace.write("case-after.json", json.dumps(after, sort_keys=True) + "\n") + report["case_distinct_delta"] = delta.validate(before, after, seed, distinct_proof) + report["case_distinct_writes"] = { + "before_sha256": legacy.fingerprint(before), + "after_sha256": legacy.fingerprint(after), + } + report["consumer_complete"] = True + workspace.check() + report["workspace_identities_rechecked"] = True + except Exception as error: + report["failure_type"] = type(error).__name__ + raise + finally: + for role, entry in report["owned_containers"].items(): + if role not in report["cleanup_by_id"]: + try: + report["cleanup_by_id"][role] = controls.cleanup_owned(cleanup_call, entry["id"], nonce) + except Exception as error: + report["cleanup_by_id"][role] = {"id": entry["id"], "failure_type": type(error).__name__} + clean = bool(report["cleanup_by_id"]) and all( + value.get("absent") and value.get("volumes_absent") for value in report["cleanup_by_id"].values() + ) + workspace.remove(credential) + if clean: + workspace.remove_database(database_fd) + try: + for value in inputs.values(): + value.recheck() + report["input_identities_rechecked"] = True + if controls.interpreter_identity(json.loads(call("image", "inspect", python_image).stdout)) != interpreter: + raise ValueError("interpreter image identity changed") + report["interpreter_identity_rechecked"] = True + except Exception as error: + report["post_input_failure_type"] = type(error).__name__ + report["cleanup"] = { + "containers_and_volumes_absent": clean, + "database_removed": clean, + "credential_removed": True, + } + report["finished_at"] = datetime.now(timezone.utc).isoformat() + report["complete"] = ( + clean + and report.get("consumer_complete") is True + and report.get("workspace_identities_rechecked") is True + and "failure_type" not in report + and "post_input_failure_type" not in report + and all(value.get("oom_killed") is False for value in report["cleanup_by_id"].values()) + ) + workspace.write("run.json", json.dumps(report, indent=2) + "\n") + workspace.check() + if not clean: + raise RuntimeError("synthetic owned cleanup incomplete") + if not report["complete"] and "failure_type" not in report: + raise RuntimeError("synthetic evidence incomplete") + + +def run(args): + if not __debug__: + raise ValueError("run acceptance without optimization") + validate_cell(args.state, args.scenario) + controls = controls_from_git(args) + with ExitStack() as stack: + inputs = controls.captured_inputs( + args, + stack, + source_files={ + "controller": "run_synthetic_legacy.py", + "support": "run_response_loss.py", + "adapter": "synthetic_legacy.py", + "verifier": "response_loss.py", + "delta": "case_delta.py", + }, + data_files={ + "wheel": (args.wheel, 4 * 1024 * 1024, controls.WHEEL_SHA256), + "snapshot_helper": (args.snapshot_helper, 128 * 1024, HELPER_SHA256), + "manifest": (args.manifest, 128 * 1024, MANIFEST_SHA256), + }, + ) + workspace = stack.enter_context(controls.workspace_for(inputs, args.output)) + run_bound(args, inputs, controls, workspace) + + +def parse_args(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--runner-commit", required=True) + parser.add_argument("--state", choices=["open", "completed", "continued"], default="open") + parser.add_argument("--scenario", choices=sorted(SCENARIOS), default="readback") + for name in ("wheel", "snapshot-helper", "manifest", "output"): + parser.add_argument("--" + name, required=True, type=Path) + return parser.parse_args() + + +if __name__ == "__main__": + run(parse_args()) diff --git a/scripts/conformance/command_keys/snapshot_bootstrap.py b/scripts/conformance/command_keys/snapshot_bootstrap.py new file mode 100644 index 0000000..a515df5 --- /dev/null +++ b/scripts/conformance/command_keys/snapshot_bootstrap.py @@ -0,0 +1,147 @@ +"""Recover captured quiescent main/WAL bytes in a mount-free private container. + +The outer controller must prove all database writers stopped or paused before +capturing both files. This program does not make a live byte copy coherent. +""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import os +import sqlite3 +import stat +import sys +import time +from contextlib import closing +from types import ModuleType + +MAX_FILE = 64 * 1024 * 1024 +MAX_PACKET = 180 * 1024 * 1024 +HELPER_SHA256 = "6fc6054ce087c136a86d7b6b222daeefaa87c58fa0b613b155a5df39e6a26fad" + + +def decode(item, maximum): + if not isinstance(item, dict) or set(item) != {"base64", "sha256"}: + raise ValueError("captured input shape differs") + if not isinstance(item["base64"], str) or len(item["base64"]) > 4 * ((maximum + 2) // 3): + raise ValueError("captured input exceeds its bound") + data = base64.b64decode(item["base64"], validate=True) + if len(data) > maximum or hashlib.sha256(data).hexdigest() != item["sha256"]: + raise ValueError("captured input bytes differ") + return data + + +def recover(main, wal): + if os.getresuid() != (0, 0, 0): + raise ValueError("private recovery bootstrap requires root") + directory = os.open("/snapshot", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + try: + info = os.fstat(directory) + if info.st_uid != 0 or stat.S_IMODE(info.st_mode) != 0o700 or os.listdir(directory): + raise ValueError("recovery directory is not empty and root-private") + for name, data in (("captured.sqlite", main), ("captured.sqlite-wal", wal)): + if name.endswith("-wal") and not data: + continue + descriptor = os.open(name, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=directory) + try: + offset = 0 + while offset < len(data): + offset += os.pwrite(descriptor, data[offset:], offset) + if os.pread(descriptor, MAX_FILE + 1, 0) != data: + raise ValueError("private recovery copy differs") + finally: + os.close(descriptor) + memory = sqlite3.connect(":memory:") + try: + # Only this trusted bootstrap exists in the container. No host + # pathname or application code participates in SQLite's native open. + with closing(sqlite3.connect("file:/snapshot/captured.sqlite?mode=ro", uri=True, timeout=2)) as source: + source.execute("PRAGMA trusted_schema=OFF") + source.execute("PRAGMA query_only=ON") + source.execute("BEGIN") + source.execute("SELECT count(*) FROM sqlite_schema").fetchone() + deadline = time.monotonic() + 5 + + def progress(_status, _remaining, _total): + if time.monotonic() > deadline: + raise ValueError("private recovery backup exceeded its bound") + + source.backup(memory, pages=128, progress=progress, sleep=0.01) + if memory.execute("PRAGMA quick_check").fetchall() != [("ok",)]: + raise ValueError("private recovered database integrity check failed") + memory.execute("PRAGMA trusted_schema=OFF") + memory.execute("PRAGMA query_only=ON") + memory.execute("BEGIN") + for name in os.listdir(directory): + os.chmod(name, 0o444, dir_fd=directory, follow_symlinks=False) + os.fchmod(directory, 0o555) + return memory + except BaseException: + memory.close() + raise + finally: + os.close(directory) + + +def drop_privileges(): + os.setgroups([]) + os.setresgid(65534, 65534, 65534) + os.setresuid(65534, 65534, 65534) + with open("/proc/self/status", encoding="ascii") as stream: + status = dict(line.split(":", 1) for line in stream if ":" in line) + if ( + os.getresuid() != (65534, 65534, 65534) + or os.getresgid() != (65534, 65534, 65534) + or os.getgroups() + or any(int(status[key].strip(), 16) for key in ("CapEff", "CapPrm", "CapAmb")) + or status["NoNewPrivs"].strip() != "1" + ): + raise ValueError("snapshot helper privilege boundary differs") + try: + os.rename("/snapshot/captured.sqlite", "/snapshot/substituted.sqlite") + except PermissionError: + pass + else: + raise ValueError("snapshot helper can replace the private database") + + +def main(): + packet = sys.stdin.buffer.read(MAX_PACKET + 1) + if len(packet) > MAX_PACKET: + raise ValueError("snapshot packet exceeds its bound") + document = json.loads(packet) + if set(document) != {"schema", "main", "wal", "helper"} or document["schema"] != "sdk-python.quiesced-snapshot.v1": + raise ValueError("snapshot packet shape differs") + main_bytes = decode(document["main"], MAX_FILE) + wal = decode(document["wal"], MAX_FILE) + helper = decode(document["helper"], 128 * 1024) + if not main_bytes or hashlib.sha256(helper).hexdigest() != HELPER_SHA256: + raise ValueError("snapshot database/helper authority differs") + with closing(recover(main_bytes, wal)) as connection: + drop_privileges() + module = ModuleType("accepted_snapshot_helper") + module.__file__ = "captured-sqlite-legacy.py" + exec(compile(helper, module.__file__, "exec"), module.__dict__) + result = module.snapshot(connection) + output = json.dumps( + { + "schema": "sdk-python.quiesced-snapshot-result.v1", + "main_sha256": hashlib.sha256(main_bytes).hexdigest(), + "wal_sha256": hashlib.sha256(wal).hexdigest(), + "helper_sha256": HELPER_SHA256, + "sqlite_version": sqlite3.sqlite_version, + "uid": os.getuid(), + "snapshot": result, + }, + sort_keys=True, + separators=(",", ":"), + ) + if len(output.encode()) > 33 * 1024 * 1024: + raise ValueError("snapshot output exceeded its bound") + print(output) + + +if __name__ == "__main__": + main() diff --git a/scripts/conformance/command_keys/snapshot_runtime.py b/scripts/conformance/command_keys/snapshot_runtime.py new file mode 100644 index 0000000..69640ca --- /dev/null +++ b/scripts/conformance/command_keys/snapshot_runtime.py @@ -0,0 +1,162 @@ +"""Quiesce owned writers and recover held bytes outside the host namespace.""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import os +import subprocess +from types import ModuleType + + +class SnapshotReader: + def __init__(self, workspace, inputs, controls, call, cleanup_call, report, nonce, writers): + self.workspace = workspace + self.inputs = inputs + self.controls = controls + self.call = call + self.cleanup_call = cleanup_call + self.report = report + self.nonce = nonce + self.writers = writers + self.capture = ModuleType("captured_quiesced_capture") + exec(compile(inputs["capture"].data, "captured-quiesced-capture.py", "exec"), self.capture.__dict__) + + def capture_bytes(self, database): + paused = [] + original_states = {} + proof = {"writers": {}} + try: + for identity in self.writers(): + record = self.controls.inspect_owned(self.call, identity, self.nonce) + if record is None or record["State"]["Paused"]: + raise ValueError("snapshot writer is missing or already paused") + original_states[identity] = bool(record["State"]["Running"]) + proof["writers"][identity] = "paused" if record["State"]["Running"] else "stopped" + if record["State"]["Running"]: + paused.append(identity) + self.call("pause", identity) + + def attest(): + if set(self.writers()) != set(original_states): + raise ValueError("snapshot writer authority changed") + for identity, running in original_states.items(): + record = self.controls.inspect_owned(self.call, identity, self.nonce) + if ( + record is None + or bool(record["State"]["Running"]) != running + or bool(record["State"]["Paused"]) != running + ): + raise ValueError("snapshot writer is not quiescent") + + files, proof["files"] = self.capture.capture(self.workspace, database, attest) + return files, proof + finally: + # The immutable ID remains cleanup authority when a workspace path + # refuses. Resume every writer this invocation paused, even if one + # unpause fails; never inspect/recover ownership by reusable name. + failures = [] + for identity in reversed(paused): + try: + record = self.controls.inspect_owned(self.cleanup_call, identity, self.nonce) + if record is not None and record["State"]["Paused"]: + self.cleanup_call("unpause", identity) + record = self.controls.inspect_owned(self.cleanup_call, identity, self.nonce) + if record is None or record["State"]["Paused"] or not record["State"]["Running"]: + raise ValueError("snapshot writer resume is unverified") + except Exception as error: + failures.append(type(error).__name__) + if failures: + raise RuntimeError("snapshot writer resume failed: " + ",".join(failures)) + + def snapshot(self, database): + for value in self.inputs.values(): + value.recheck() + files, proof = self.capture_bytes(database) + return self.recover(files, proof)["snapshot"] + + def recover(self, files, proof): + role = "snapshot-" + str(len(self.report.setdefault("snapshots", []))) + proof["role"] = role + self.report["snapshots"].append(proof) + image = "docker.io/library/python@" + self.controls.PYTHON_INDEX + bootstrap = self.inputs["snapshot_bootstrap"].data + identity = self.controls.created_id( + self.call( + "create", + "--name", + "sdk-python-216-" + role + "-" + self.nonce[:12], + "--label", + self.controls.OWNER_LABEL + "=sdk-python", + "--label", + self.controls.RUN_LABEL + "=" + self.nonce, + "--init", + "--read-only", + "--network", + "none", + "--cpus", + "1", + "--memory", + "1g", + "--pids-limit", + "128", + "--user", + "0:0", + "--security-opt", + "no-new-privileges", + "--cap-drop", + "ALL", + "--cap-add", + "SETUID", + "--cap-add", + "SETGID", + "--tmpfs", + "/snapshot:mode=0700,size=192m,noexec,nosuid,nodev", + "-i", + "--entrypoint", + "/usr/local/bin/python3", + image, + "-I", + "-B", + "-c", + bootstrap.decode(), + ).stdout + ) + if identity in [entry["id"] for entry in self.report["owned_containers"].values()]: + raise ValueError("duplicate snapshot container identity") + self.report["owned_containers"][role] = {"id": identity} + record = self.controls.inspect_owned(self.call, identity, self.nonce) + if record is None or record["Mounts"] or record["HostConfig"]["NetworkMode"] != "none": + raise ValueError("private snapshot has a host mount or network") + packet = {"schema": "sdk-python.quiesced-snapshot.v1"} + for name, data in { + "main": files["main"], + "wal": files["wal"], + "helper": self.inputs["snapshot_helper"].data, + }.items(): + packet[name] = {"base64": base64.b64encode(data).decode(), "sha256": hashlib.sha256(data).hexdigest()} + result = subprocess.run( + ["docker", "--config", "/proc/self/fd", "start", "-a", "-i", identity], + input=json.dumps(packet), + env={"PATH": os.defpath, "TMPDIR": self.workspace.proc(self.workspace.root)}, + capture_output=True, + text=True, + timeout=30, + ) + self.workspace.write(role + ".log", result.stdout + result.stderr) + record = self.controls.inspect_owned(self.call, identity, self.nonce) + if record is None or result.returncode or record["State"]["Running"] or record["State"]["ExitCode"]: + raise RuntimeError("private snapshot recovery failed") + output = json.loads(result.stdout) + if ( + output["schema"] != "sdk-python.quiesced-snapshot-result.v1" + or output["uid"] != 65534 + or any(output[name + "_sha256"] != packet[name]["sha256"] for name in ("main", "wal", "helper")) + ): + raise ValueError("private snapshot provenance differs") + proof["recovery"] = {key: value for key, value in output.items() if key != "snapshot"} + self.report["cleanup_by_id"][role] = self.controls.cleanup_owned(self.cleanup_call, identity, self.nonce) + for value in self.inputs.values(): + value.recheck() + return output diff --git a/scripts/conformance/command_keys/synthetic_legacy.py b/scripts/conformance/command_keys/synthetic_legacy.py new file mode 100644 index 0000000..346f45b --- /dev/null +++ b/scripts/conformance/command_keys/synthetic_legacy.py @@ -0,0 +1,357 @@ +"""Installed Python API adapter for the accepted synthetic SQLite legacy cases.""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import stat +import time +from datetime import datetime, timezone +from urllib.parse import urlsplit +from uuid import uuid4 + +from durable_workflow import Client, Worker, workflow +from durable_workflow.errors import ServerError, SignalFailed, UpdateFailed +from durable_workflow.serializer import decode_envelope + +MANIFEST_SHA256 = "da1990000c611d7377690c7526487d6415fa4f9cb42d299f59585a7c2eebf769" +WORKFLOW_TYPE = "sdk-python-synthetic-legacy" +SCENARIOS = {"readback", "two-rows", "legacy-token", "modern-raw", "case-only"} +IDENTITIES = ("command_id", "run_id", "update_id", "accepted_at") + + +@workflow.defn(name=WORKFLOW_TYPE) +class LegacyCommands: + def __init__(self): + self.finish = False + self.rotate = False + + @workflow.signal("touch") + def touch(self, value): + assert value == 1 + + @workflow.update("mutate") + def mutate(self, value): + assert value == 1 + return None + + @workflow.signal("close") + def close(self, rotate): + self.rotate = rotate + self.finish = True + + def run(self, ctx, generation=0): + yield ctx.wait_condition(lambda: self.finish, key="control", timeout=600) + if self.rotate: + yield ctx.continue_as_new(generation + 1) + return None + + +def packet(data): + value = json.loads(data) + if ( + value.get("schema") != "sdk-python.synthetic-legacy-input.v1" + or value.get("manifest_sha256") != MANIFEST_SHA256 + or value.get("phase") not in {"prepare", "read", "case_distinct"} + or value.get("scenario") not in SCENARIOS + or value.get("state") not in {"open", "completed", "continued"} + or (value["scenario"] != "readback" and value["state"] != "open") + ): + raise ValueError("require a declared synthetic legacy cell") + if value["phase"] == "case_distinct" and value["scenario"] != "case-only": + raise ValueError("distinct-key admissions require the separate case-only interval") + if value["phase"] != "prepare": + seed = value.get("seed", {}) + if set(seed.get("commands", {})) != {"signal", "update"}: + raise ValueError("require both original API receipts") + for kind, command in seed["commands"].items(): + original = command["original"] + fields = IDENTITIES if kind == "update" else IDENTITIES[:2] + if any(not isinstance(original.get(key), str) or not original[key] for key in fields): + raise ValueError("original durable receipt identity is missing") + if original["run_id"] != seed["run_id"] or original["workflow_id"] != seed["workflow_id"]: + raise ValueError("original receipt target differs") + return value + + +def selection(kind, workflow_id, original, caller_key, legacy_key): + return { + "command_id": original["command_id"], + "command_type": kind, + "workflow_instance_id": workflow_id, + "workflow_run_id": original["run_id"], + "caller_key": caller_key, + "legacy_request_id": legacy_key, + } + + +def retained(original, received, kind): + if received.get("retained") is not True or received.get("intent_verified") is not False: + raise ValueError("legacy receipt must be retained and explicitly unverified") + for field in IDENTITIES if kind == "update" else IDENTITIES[:2]: + if received.get(field) != original[field]: + raise ValueError("legacy durable identity differs: " + field) + + +async def command(client, kind, seed, key, value=999, *, name=None, run_id=None): + if kind == "signal": + return await client.signal_workflow( + seed["workflow_id"], name or "touch", args=[value], request_id=key, run_id=run_id or seed["run_id"] + ) + return await client.update_workflow( + seed["workflow_id"], name or "mutate", args=[value], request_id=key, wait_for="accepted" + ) + + +async def prepare(client, value): + queue = "synthetic-legacy-" + uuid4().hex + worker = Worker(client, task_queue=queue, workflows=[LegacyCommands], poll_timeout=0.2) + task = asyncio.create_task(worker.run()) + try: + handle = await client.start_workflow( + workflow_type=WORKFLOW_TYPE, input=[0], workflow_id=queue, task_queue=queue + ) + # The producer admits updates only after the first real workflow task + # binds its contract. Stop before command admission so open updates stay pending. + deadline = time.monotonic() + 45 + while time.monotonic() < deadline: + events = [event async for event in client.iter_history(queue, handle.run_id)] + if any(event["event_type"] == "ConditionWaitOpened" for event in events): + break + await asyncio.sleep(0.1) + else: + raise TimeoutError("first workflow task did not bind the run contract") + finally: + await worker.stop() + await task + seed = {"workflow_id": handle.workflow_id, "run_id": handle.run_id, "commands": {}, "selections": []} + caller = "CaseSensitive" if value["scenario"] == "case-only" else "token-key" + for kind in ("signal", "update"): + original = await command(client, kind, seed, caller, 1) + seed["commands"][kind] = {"original": original, "caller_key": caller} + chosen = original + chosen_key = caller + legacy_key = caller + if value["scenario"] == "two-rows": + chosen_key = "second-key" + chosen = await command(client, kind, seed, chosen_key, 1) + elif value["scenario"] == "legacy-token": + legacy_key = kind + "-v1:" + hashlib.sha256(caller.encode()).hexdigest() + if value["scenario"] != "modern-raw": + seed["selections"].append(selection(kind, seed["workflow_id"], chosen, chosen_key, legacy_key)) + if value["state"] != "open": + worker = Worker(client, task_queue=queue, workflows=[LegacyCommands], poll_timeout=0.2) + task = asyncio.create_task(worker.run()) + try: + update = seed["commands"]["update"]["original"] + assert await client.update_outcome(queue, update["update_id"], wait_timeout_seconds=30) is None + await client.signal_workflow(queue, "close", args=[value["state"] == "continued"], run_id=handle.run_id) + deadline = time.monotonic() + 45 + while time.monotonic() < deadline: + description = await client.describe_workflow(queue) + if value["state"] == "completed" and description.status == "completed": + break + if value["state"] == "continued" and description.run_id != handle.run_id: + seed["successor_run_id"] = description.run_id + break + await asyncio.sleep(0.1) + else: + raise TimeoutError("original workflow did not reach the declared state") + finally: + await worker.stop() + await task + seed["history"] = [event async for event in client.iter_history(queue, handle.run_id)] + return seed + + +async def read(client, value): + seed = value["seed"] + observed = [event async for event in client.iter_history(seed["workflow_id"], seed["run_id"])] + assert observed == seed["history"] + cells = [] + for kind, original in seed["commands"].items(): + key = original["caller_key"] + receipt = original["original"] + cell = {"kind": kind, "original_command_id": receipt["command_id"], "original_run_id": receipt["run_id"]} + if value["scenario"] in {"two-rows", "legacy-token", "modern-raw"}: + if value["scenario"] == "modern-raw": + key = kind + "-v1:" + hashlib.sha256(key.encode()).hexdigest() + try: + await command(client, kind, seed, key) + except (ServerError, SignalFailed, UpdateFailed) as error: + assert error.status == 409 + assert error.body["reason"] == "request_id_ambiguous" + assert error.body["retained"] is True and error.body["intent_verified"] is False + assert error.body["retryable"] is False + cell["refusal"] = "request_id_ambiguous" + else: + raise AssertionError("ambiguous key selected a receipt") + else: + replies = await asyncio.gather(*(command(client, kind, seed, key) for _ in range(4))) + for reply in replies: + retained(receipt, reply, kind) + if kind == "update" and value["state"] != "open": + assert reply["update_status"] == "completed" and reply["result"] is None + try: + await command(client, kind, seed, key, name="other") + except (SignalFailed, UpdateFailed) as error: + assert error.reason == "request_id_conflict" + assert error.details["retained_command_id"] == receipt["command_id"] + else: + raise AssertionError("changed legacy name was admitted") + if kind == "signal": + try: + await command(client, kind, seed, key, run_id="01M00000000000000000000000") + except SignalFailed as error: + assert error.reason == "request_id_conflict" + else: + raise AssertionError("different explicit run selected the legacy receipt") + if kind == "update": + outcome = await client.update_outcome_receipt(seed["workflow_id"], receipt["update_id"]) + assert outcome["update_id"] == receipt["update_id"] + assert outcome["update_status"] == ("accepted" if value["state"] == "open" else "completed") + if value["state"] != "open": + assert outcome["result"] is None + assert await client.update_outcome(seed["workflow_id"], receipt["update_id"]) is None + cell.update(concurrent_retries=4, retained=True, intent_verified=False, changed_name="request_id_conflict") + cells.append(cell) + return cells + + +async def case_distinct(client, value): + seed = value["seed"] + cells = [] + sent = [] + + async def observe(response): + request = response.request + if request.method != "POST" or not request.url.path.endswith(("/signal/touch", "/update/mutate")): + return + body = json.loads(request.content) + headers = { + field: request.headers[name] + for name, field in (("x-request-id", "x_request_id"), ("x-correlation-id", "x_correlation_id")) + if name in request.headers + } + fingerprint = { + "method": "POST", + "path": request.url.path, + "payload": json.loads(json.dumps(body, sort_keys=True)), + "headers": headers or [], + } + sent.append( + { + "status": response.status_code, + "fingerprint": "sha256:" + + hashlib.sha256( + json.dumps(fingerprint, separators=(",", ":"), ensure_ascii=False).encode() + ).hexdigest(), + } + ) + + client._http.event_hooks["response"].append(observe) + try: + for kind, original in seed["commands"].items(): + start = len(sent) + admitted = await command(client, kind, seed, "casesensitive") + assert len(sent) == start + 1 and sent[start]["status"] == 202 + assert admitted["command_id"] != original["original"]["command_id"] + assert admitted["run_id"] == seed["run_id"] + assert admitted["request_id"] == "casesensitive" and admitted.get("retained") is not True + assert admitted["intent_verified"] is True + same = await command(client, kind, seed, "casesensitive") + assert same["command_id"] == admitted["command_id"] and same["retained"] is True + assert same["intent_verified"] is True + try: + await command(client, kind, seed, "casesensitive", 1000) + except (SignalFailed, UpdateFailed) as error: + assert error.reason == "request_id_conflict" + else: + raise AssertionError("fresh case-distinct intent was not fenced") + cells.append( + { + "kind": kind, + "receipt": admitted, + "request_fingerprint": sent[start]["fingerprint"], + "status": 202, + "same_intent_retained": True, + "changed_intent_conflict": True, + } + ) + finally: + client._http.event_hooks["response"].remove(observe) + history = [event async for event in client.iter_history(seed["workflow_id"], seed["run_id"])] + assert history[: len(seed["history"])] == seed["history"] + events = history[len(seed["history"]) :] + assert len(events) == 2 + for cell, event in zip(cells, events, strict=True): + payload = event["payload"] + snapshot = payload["command"] + old_event = next( + item + for item in seed["history"] + if item["payload"].get("workflow_command_id") == seed["commands"][cell["kind"]]["original"]["command_id"] + ) + expected = dict(old_event["payload"]["command"]) + expected.update( + id=cell["receipt"]["command_id"], + sequence=expected["sequence"] + 2, + message_sequence=expected["message_sequence"] + 2, + payload=snapshot["payload"], + request_id=cell["kind"] + "-v1:" + hashlib.sha256(b"casesensitive").hexdigest(), + request_fingerprint=cell["request_fingerprint"], + accepted_at=snapshot["accepted_at"], + ) + assert snapshot == expected + assert decode_envelope(payload["arguments"]) == [999] + assert decode_envelope(snapshot["payload"]) == { + "name": "touch" if cell["kind"] == "signal" else "mutate", + "arguments": [999], + "validation_errors": [], + } + return {"cells": cells, "events": events} + + +async def main(args): + if not __debug__: + raise ValueError("run acceptance without optimization") + value = packet(args.seed_data) + if hashlib.sha256(args.seed_data).hexdigest() != args.seed_sha256 or args.output.exists(): + raise ValueError("captured packet identity or output differs") + upstream = urlsplit(args.upstream) + if ( + upstream.scheme != "http" + or upstream.hostname != "127.0.0.1" + or args.upstream != f"http://127.0.0.1:{upstream.port}" + ): + raise ValueError("require canonical numeric loopback") + mode = args.credential_file.lstat().st_mode + if not stat.S_ISREG(mode) or mode & 0o077: + raise ValueError("fixture credential must be private") + token = dict(line.split("=", 1) for line in args.credential_file.read_text().splitlines())["DW_AUTH_TOKEN"] + report = { + "started_at": datetime.now(timezone.utc).isoformat(), + "adapter_sha256": args.adapter_sha256, + "seed_sha256": args.seed_sha256, + "sdk": args.package_identity, + "scope": "synthetic legacy " + value["scenario"] + " " + value["state"], + "complete": False, + } + try: + async with Client(args.upstream, token=token) as client: + assert (await client.get_cluster_info())["version"] == args.server_version + if value["phase"] == "prepare": + report["seed"] = await prepare(client, value) + elif value["phase"] == "case_distinct": + report.update(await case_distinct(client, value)) + else: + report["cells"] = await read(client, value) + report["complete"] = True + report["finished_at"] = datetime.now(timezone.utc).isoformat() + except Exception as error: + report["failure_type"] = type(error).__name__ + raise + finally: + args.output.write_text(json.dumps(report, indent=2) + "\n") + return report diff --git a/scripts/conformance/command_keys/workspace_guards.py b/scripts/conformance/command_keys/workspace_guards.py index 0a27e6d..b3d0e69 100644 --- a/scripts/conformance/command_keys/workspace_guards.py +++ b/scripts/conformance/command_keys/workspace_guards.py @@ -4,7 +4,9 @@ import hashlib import os +import sqlite3 import stat +from contextlib import closing from pathlib import Path NATIVE_PREFLIGHT = """set -eu @@ -106,6 +108,24 @@ def native_arguments(self, directory, database, *arguments): *arguments, ] + def database_uri(self, database, mode="ro"): + self.check() + if self.files.get((database["parent"], database["name"])) is not database: + raise ValueError("snapshot database is not a held workspace file") + if mode not in {"ro", "rw"}: + raise ValueError("unsupported held database mode") + return "file:/proc/self/fd/" + str(database["fd"]) + "?mode=" + mode + + def read_database(self, database, reader): + # Target the retained real file, not a mutable basename under a held + # directory. mode=ro includes active WAL; immutable=1 would omit it. + uri = self.database_uri(database) + with closing(sqlite3.connect(uri, uri=True, timeout=2)) as connection: + connection.execute("BEGIN") + result = reader(connection) + self.check() + return result + def check_file(self, item): current = os.stat(item["name"], dir_fd=item["parent"], follow_symlinks=False) held = os.fstat(item["fd"]) diff --git a/tests/test_command_keys_workspace.py b/tests/test_command_keys_workspace.py index 000196c..4ee565c 100644 --- a/tests/test_command_keys_workspace.py +++ b/tests/test_command_keys_workspace.py @@ -178,3 +178,68 @@ def test_cleanup_configuration_is_an_immutable_empty_kernel_namespace(): assert not path.exists() with pytest.raises(OSError): path.write_text("must not create") + + +def test_real_database_fd_refuses_the_predecessor_swap_open_restore_race(tmp_path, monkeypatch): + root = tmp_path / "work" + real_connect = sqlite3.connect + with module.Workspace(root) as workspace: + directory = workspace.directory("database") + held = workspace.write("copied.sqlite", b"", parent=directory, immutable=False) + target = root / "database/copied.sqlite" + other = root / "database/substitute.sqlite" + saved = root / "database/retained-original.sqlite" + for path, text in ((target, "original"), (other, "substituted")): + with real_connect(path) as connection: + connection.execute("CREATE TABLE values_table(value TEXT)") + connection.execute("INSERT INTO values_table VALUES (?)", (text,)) + path.chmod(0o600) + + def swapping_connect(*args, **kwargs): + target.rename(saved) + other.rename(target) + try: + return real_connect(*args, **kwargs) + finally: + target.rename(other) + saved.rename(target) + + monkeypatch.setattr(module.sqlite3, "connect", swapping_connect) + workspace.check() + # The321/322 predecessor retains a directory FD plus mutable basename. + with module.closing( + module.sqlite3.connect(Path(workspace.proc(directory, "copied.sqlite")).as_uri() + "?mode=ro", uri=True) + ) as connection: + assert connection.execute("SELECT value FROM values_table").fetchall() == [("substituted",)] + workspace.check() + assert workspace.read_database(held, lambda db: db.execute("SELECT value FROM values_table").fetchall()) == [ + ("original",) + ] + workspace.check() + + +def test_real_database_fd_reads_uncheckpointed_wal_from_another_connection(tmp_path): + root = tmp_path / "work" + with module.Workspace(root) as workspace: + directory = workspace.directory("database") + held = workspace.write("copied.sqlite", b"", parent=directory, immutable=False) + target = root / "database/copied.sqlite" + with module.closing(sqlite3.connect(target)) as initialize: + initialize.execute("CREATE TABLE values_table(value INTEGER)") + initialize.execute("INSERT INTO values_table VALUES (1)") + initialize.commit() + with module.closing(sqlite3.connect(target)) as writer: + assert writer.execute("PRAGMA journal_mode=WAL").fetchone()[0] == "wal" + writer.execute("PRAGMA wal_autocheckpoint=0") + writer.execute("INSERT INTO values_table VALUES (2)") + writer.commit() + assert Path(str(target) + "-wal").stat().st_size > 0 + # The positive proves the committed row is still WAL-only. + with module.closing( + sqlite3.connect("file:/proc/self/fd/" + str(held["fd"]) + "?immutable=1", uri=True) + ) as main_only: + assert main_only.execute("SELECT value FROM values_table ORDER BY value").fetchall() == [(1,)] + assert workspace.read_database( + held, lambda db: db.execute("SELECT value FROM values_table ORDER BY value").fetchall() + ) == [(1,), (2,)] + workspace.check() diff --git a/tests/test_physical_command_key_readback.py b/tests/test_physical_command_key_readback.py index d8d08fa..5e3613e 100644 --- a/tests/test_physical_command_key_readback.py +++ b/tests/test_physical_command_key_readback.py @@ -3,7 +3,6 @@ import copy import hashlib import importlib.util -import sqlite3 from pathlib import Path import pytest @@ -132,16 +131,14 @@ def test_physical_bundle_closes_seed_and_verifier_roles(): def test_schema_authority_uses_native_marker_not_sqlite_user_version(): - with sqlite3.connect(":memory:") as connection: - connection.execute("PRAGMA user_version=12") - connection.execute("CREATE TABLE dw_server_schema(engine,version)") - connection.execute("INSERT INTO dw_server_schema VALUES ('rust-development',13)") - controller.native_marker(connection, 13) - with pytest.raises(ValueError, match="marker differs"): - controller.native_marker(connection, 12) - connection.execute("INSERT INTO dw_server_schema VALUES ('rust-development',13)") - with pytest.raises(ValueError, match="marker differs"): - controller.native_marker(connection, 13) + marker = {"columns": ["engine", "version"], "rows": [[["text", "rust-development"], ["integer", 13]]]} + snapshot = {"user_version": 12, "tables": {"dw_server_schema": marker}} + controller.native_marker(snapshot, 13) + with pytest.raises(ValueError, match="marker differs"): + controller.native_marker(snapshot, 12) + marker["rows"].append(copy.deepcopy(marker["rows"][0])) + with pytest.raises(ValueError, match="marker shape differs"): + controller.native_marker(snapshot, 13) def upgrade_snapshots(): diff --git a/tests/test_quiesced_snapshot.py b/tests/test_quiesced_snapshot.py new file mode 100644 index 0000000..86c5911 --- /dev/null +++ b/tests/test_quiesced_snapshot.py @@ -0,0 +1,154 @@ +from __future__ import annotations + +import importlib.util +import os +import sqlite3 +from contextlib import closing +from pathlib import Path +from types import SimpleNamespace + +import pytest + +ROOT = Path(__file__).parents[1] / "scripts/conformance/command_keys" + + +def load(name): + spec = importlib.util.spec_from_file_location(name, ROOT / (name + ".py")) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +capture_module = load("quiesced_capture") +workspace_module = load("workspace_guards") + + +def initialize(path): + with closing(sqlite3.connect(path)) as connection: + connection.execute("CREATE TABLE sample(value INTEGER)") + connection.execute("INSERT INTO sample VALUES (1)") + connection.commit() + + +def test_capture_never_opens_sqlite_path_and_keeps_committed_wal(tmp_path, monkeypatch): + with workspace_module.Workspace(tmp_path / "work") as workspace: + directory = workspace.directory("database") + held = workspace.write("sample.sqlite", b"", parent=directory, immutable=False) + path = tmp_path / "work/database/sample.sqlite" + initialize(path) + with closing(sqlite3.connect(path)) as writer: + writer.execute("PRAGMA journal_mode=WAL") + writer.execute("PRAGMA wal_autocheckpoint=0") + writer.execute("INSERT INTO sample VALUES (2)") + writer.commit() + with closing(sqlite3.connect(path)) as other: + other.execute("INSERT INTO sample VALUES (3)") + other.commit() + monkeypatch.setattr(sqlite3, "connect", lambda *_a, **_kw: pytest.fail("capture reopened SQLite")) + calls = [] + data, proof = capture_module.capture(workspace, held, lambda: calls.append(True)) + assert len(calls) == 3 + assert data["main"] == os.pread(held["fd"], capture_module.MAX_FILE, 0) + assert data["wal"] == Path(str(path) + "-wal").read_bytes() + assert proof["wal"]["bytes"] > 0 + + +def test_quiescent_attestation_failure_refuses_before_capture(tmp_path, monkeypatch): + with workspace_module.Workspace(tmp_path / "work") as workspace: + directory = workspace.directory("database") + held = workspace.write("sample.sqlite", b"", parent=directory, immutable=False) + initialize(tmp_path / "work/database/sample.sqlite") + + def refuse(): + raise ValueError("writer is running") + + monkeypatch.setattr(capture_module.os, "pread", lambda *_a: pytest.fail("read before writer attestation")) + with pytest.raises(ValueError, match="writer is running"): + capture_module.capture(workspace, held, refuse) + + +def test_same_size_main_mutation_between_captures_refuses(tmp_path): + with workspace_module.Workspace(tmp_path / "work") as workspace: + directory = workspace.directory("database") + held = workspace.write("sample.sqlite", b"", parent=directory, immutable=False) + initialize(tmp_path / "work/database/sample.sqlite") + calls = 0 + + def invalid_attestation(): + nonlocal calls + calls += 1 + if calls == 2: + original = os.pread(held["fd"], 1, 100) + os.pwrite(held["fd"], bytes([original[0] ^ 1]), 100) + + with pytest.raises(ValueError, match="captures differ"): + capture_module.capture(workspace, held, invalid_attestation) + + +def test_wal_created_after_initial_absence_refuses(tmp_path): + with workspace_module.Workspace(tmp_path / "work") as workspace: + directory = workspace.directory("database") + held = workspace.write("sample.sqlite", b"", parent=directory, immutable=False) + path = tmp_path / "work/database/sample.sqlite" + initialize(path) + calls = 0 + + def invalid_attestation(): + nonlocal calls + calls += 1 + if calls == 2: + Path(str(path) + "-wal").write_bytes(b"substituted") + + with pytest.raises(ValueError, match="sidecar identity changed"): + capture_module.capture(workspace, held, invalid_attestation) + + +def test_sidecar_symlink_refuses(tmp_path): + with workspace_module.Workspace(tmp_path / "work") as workspace: + directory = workspace.directory("database") + held = workspace.write("sample.sqlite", b"", parent=directory, immutable=False) + path = tmp_path / "work/database/sample.sqlite" + initialize(path) + Path(str(path) + "-wal").symlink_to(path) + with pytest.raises(OSError): + capture_module.capture(workspace, held, lambda: None) + + +@pytest.mark.parametrize("failure", ["after_pause", "during_capture", "second_writer"]) +def test_id_bound_writer_resumes_after_workspace_or_capture_refusal(failure): + runtime = load("snapshot_runtime") + first, second = "a" * 64, "b" * 64 + states = {first: {"State": {"Running": True, "Paused": False}}} + calls = [] + + def call(action, identity): + calls.append((action, identity)) + if action == "pause": + states[identity]["State"]["Paused"] = True + if failure == "after_pause": + raise ValueError("workspace replaced after successful Docker pause") + elif action == "unpause": + states[identity]["State"]["Paused"] = False + + controls = SimpleNamespace(inspect_owned=lambda _call, identity, _nonce: states.get(identity)) + inputs = {"capture": SimpleNamespace(data=(ROOT / "quiesced_capture.py").read_bytes())} + reader = runtime.SnapshotReader( + None, + inputs, + controls, + call, + call, + {}, + "nonce", + lambda: [first, second] if failure == "second_writer" else [first], + ) + + def refuse(*_args): + raise ValueError("workspace replaced during capture") + + reader.capture.capture = refuse + with pytest.raises(ValueError): + reader.capture_bytes(None) + assert states[first]["State"] == {"Running": True, "Paused": False} + assert calls[-1] == ("unpause", first) diff --git a/tests/test_synthetic_command_key_readback.py b/tests/test_synthetic_command_key_readback.py new file mode 100644 index 0000000..26617fd --- /dev/null +++ b/tests/test_synthetic_command_key_readback.py @@ -0,0 +1,171 @@ +from __future__ import annotations + +import copy +import importlib.util +import json +from pathlib import Path +from types import SimpleNamespace + +import pytest + +ROOT = Path(__file__).parents[1] / "scripts/conformance/command_keys" + + +def module(name, filename): + spec = importlib.util.spec_from_file_location(name, ROOT / filename) + assert spec and spec.loader + value = importlib.util.module_from_spec(spec) + spec.loader.exec_module(value) + return value + + +consumer = module("synthetic_consumer", "synthetic_legacy.py") +controller = module("synthetic_controller", "run_synthetic_legacy.py") +bootstrap = module("synthetic_bootstrap", "consumer_bootstrap.py") + + +def packet(): + return { + "schema": "sdk-python.synthetic-legacy-input.v1", + "manifest_sha256": consumer.MANIFEST_SHA256, + "phase": "read", + "state": "open", + "scenario": "readback", + "seed": { + "workflow_id": "workflow", + "run_id": "run", + "commands": { + kind: { + "original": { + "workflow_id": "workflow", + "run_id": "run", + "command_id": kind, + "update_id": "update", + "accepted_at": "2026-01-01T00:00:00Z", + } + } + for kind in ("signal", "update") + }, + }, + } + + +@pytest.mark.parametrize("state", ["open", "completed", "continued"]) +def test_declared_readback_states_keep_original_api_receipts(state): + document = packet() + document["state"] = state + assert consumer.packet(json.dumps(document))["state"] == state + controller.validate_cell(state, "readback") + + +@pytest.mark.parametrize( + "change", + [ + lambda v: v.update(manifest_sha256="wrong"), + lambda v: v.update(phase="execute-arbitrary"), + lambda v: v.update(state="invented"), + lambda v: v.update(scenario="invented"), + lambda v: v.update(phase="case_distinct"), + lambda v: v.update(state="continued", scenario="two-rows"), + lambda v: v["seed"]["commands"].pop("signal"), + lambda v: v["seed"]["commands"]["update"]["original"].pop("update_id"), + lambda v: v["seed"]["commands"]["signal"]["original"].update(run_id="different"), + lambda v: v["seed"]["commands"]["signal"]["original"].update(workflow_id="different"), + ], +) +def test_wrong_or_incomplete_packet_refuses_before_api(change): + document = packet() + change(document) + with pytest.raises(ValueError): + consumer.packet(json.dumps(document)) + + +@pytest.mark.parametrize( + "change", + [ + {"retained": False}, + {"intent_verified": True}, + {"intent_verified": None}, + {"command_id": "new"}, + {"run_id": "successor"}, + {"update_id": "new"}, + {"accepted_at": "new"}, + ], +) +def test_retained_receipt_cannot_fabricate_intent_or_target(change): + original = packet()["seed"]["commands"]["update"]["original"] + reply = {**original, "retained": True, "intent_verified": False, **change} + with pytest.raises(ValueError): + consumer.retained(original, reply, "update") + + +def test_helper_execution_is_refused_before_wrong_bytes_can_run(): + inputs = { + "snapshot_helper": SimpleNamespace(data=b"raise RuntimeError('must not execute')"), + "manifest": SimpleNamespace(data=b"{}"), + } + with pytest.raises(ValueError, match="accepted helper/manifest"): + controller.accepted_helper(inputs) + + +@pytest.mark.parametrize( + "state,scenario", [("completed", "case-only"), ("continued", "modern-raw"), ("open", "unknown")] +) +def test_controller_refuses_unsupported_cell_combinations(state, scenario): + with pytest.raises(ValueError): + controller.validate_cell(state, scenario) + + +def test_case_delta_preserves_all_prior_rows_and_rejects_unrelated_writes(): + delta = module("case_delta_control", "case_delta.py") + before = { + "schema": [], + "user_version": 0, + "application_id": 0, + "tables": { + "workflow_commands": {"columns": ["id"], "rows": [[["text", "old"]]]}, + "dw_command_intents": {"columns": ["command_id"], "rows": [[["text", "old"]]]}, + "unrelated": {"columns": ["value"], "rows": [[["text", "preserve"]]]}, + }, + } + expected = copy.deepcopy(before) + for name in ("workflow_commands", "dw_command_intents"): + expected["tables"][name]["rows"] += [[["text", "new1"]], [["text", "new2"]]] + expected["tables"][name]["rows"].sort() + delta.stable_catalog(before, expected) + delta.finish_expected(copy.deepcopy(expected), expected) + unrelated = copy.deepcopy(expected) + unrelated["tables"]["unrelated"]["rows"][0][0][1] = "changed" + replaced = copy.deepcopy(expected) + replaced["tables"]["workflow_commands"]["rows"].remove([["text", "old"]]) + replaced["tables"]["workflow_commands"]["rows"].append([["text", "third-new"]]) + for changed in (unrelated, replaced): + # The predecessor count control accepts both corruptions. + assert all( + len(changed["tables"][name]["rows"]) == len(before["tables"][name]["rows"]) + 2 + for name in ("workflow_commands", "dw_command_intents") + ) + with pytest.raises(ValueError): + delta.finish_expected(copy.deepcopy(expected), changed) + columns = copy.deepcopy(expected) + columns["tables"]["unrelated"]["columns"] = ["different"] + with pytest.raises(ValueError): + delta.stable_catalog(before, columns) + + +def test_stopped_copy_must_equal_the_first_restart_before_later_zero_writes(): + delta = module("case_delta_restart", "case_delta.py") + stopped = {"business": [["original"]]} + restarted = {"business": [["substituted"]]} + # The predecessor only compares the two later snapshots and accepts this. + assert restarted == copy.deepcopy(restarted) + with pytest.raises(ValueError): + delta.validate_restart(stopped, restarted) + delta.validate_restart(stopped, copy.deepcopy(stopped)) + + +def test_synthetic_bundle_has_the_same_closed_roles_as_physical_not_the_proxy(): + with pytest.raises(ValueError, match="source roles"): + bootstrap.decode_inputs({"scenario": "synthetic_legacy", "sources": {"adapter": {}}}) + with pytest.raises(ValueError, match="source roles"): + bootstrap.decode_inputs({"scenario": "invented", "sources": {}})