From e1375b6bc4ad12a9703219f6955b2d1e6ef95163 Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Sun, 8 Nov 2026 13:18:45 -0600 Subject: [PATCH 1/4] feat: validate unsigned Finisher observation journeys through admitted probes --- docs/FINISHER_COLLECTOR.md | 219 +++++++++ scripts/collect_finisher_e2e.py | 67 +++ .../finisher_collector.py | 431 ++++++++++++++++++ tests/test_finisher_collector.py | 403 ++++++++++++++++ 4 files changed, 1120 insertions(+) create mode 100644 docs/FINISHER_COLLECTOR.md create mode 100644 scripts/collect_finisher_e2e.py create mode 100644 src/echo_certification_forge/finisher_collector.py create mode 100644 tests/test_finisher_collector.py diff --git a/docs/FINISHER_COLLECTOR.md b/docs/FINISHER_COLLECTOR.md new file mode 100644 index 0000000..3b944d0 --- /dev/null +++ b/docs/FINISHER_COLLECTOR.md @@ -0,0 +1,219 @@ +# Finisher observation collector candidate + +This collector validates fresh, authenticated, independently recalled Finisher +journeys through a separately admitted owner probe. Its only successful output is +`UNSIGNED_CANDIDATE`, with `release_verdict: NOT_READY` and +`signature_verified: false`. It does not sign, enroll a collector, install an +attestation, create a worker grant, change the staging gate or deploy a target. + +The implementation uses the installed `generic-production-v1` contract from +CertForge source `d8ad9fd881365801354b78a758632d2a31ab6828`. It reuses the canonical +JSON hash and existing restricted-content checks. Existing trust verification +rejects its unsigned payload. A passing test fixture is not live evidence. + +## Current missing prerequisites + +There is no owner-probe executable included in this candidate and no newly +admitted collector or signer. The CLI fails with `owner_probe_unconfigured` when +that dependency is absent. An implementation that returns configured booleans or +replays a saved observation is not a valid owner probe. + +The independently admitted owner must supply and review the real adapter to the +existing authentication, control, observation and receipt services. Its entire +executable dependency closure must be immutable and hash pinned, and its runtime +identity must have actual admission for this exact policy, target and operation. +Hash pins provide integrity, not authorization. The implementer of this candidate +cannot make themselves an independent collector, signer or reviewer by changing +their role label. Existing Maximalist signatures and the Forge integrity +bootstrap do not attest to Finisher. + +There are also concrete selected-runtime constraints. In transport source +`1b79e2ee932927b3e408ff2ed43eb8d248875aaf`: + +* `SYSTEMS/finisher_cloud_integration/cloud_front.py` treats `start` as + reconciliation of an already running singleton. A stopped singleton requires + an independently admitted owner bootstrap operation; frontend `start` does not + provide one. +* `SYSTEMS/finisher_cloud_integration/ui_runtime.py` returns a read-only + `echo.finisher.ui.snapshot.v1` for status. Accepted/applied are journal control + fields, not a status mutation receipt. +* Native controller halt is `halted`, not `stopped`. The Cloudflare observer in + `SYSTEMS/cf_terminal/cloud/container_control.py` currently requires a live + process for `readback_verified`. It cannot prove terminal halt after that + process exits. The UI projection also omits the controller's `halted` field. + The owner probe must not invent terminal proof; that source/runtime dependency + must be repaired, independently reviewed and actually observed first. +* Worker-image rollback alone does not restore a compatible checkpoint. The + runtime rejects a checkpoint whose source identity differs. A separately + authorized compatible checkpoint/config restoration and exact readback are + prerequisites. This collector does not provide a reset or data-restore route. + +These constraints remain holds. They are not requests to weaken the gate or to +reuse another product's trust identity. + +## Invocation and private configuration + +Run on an admitted Linux observer host using its normal opaque scoped broker +environment. Do not copy another client's auth state or put credentials in the +policy, command arguments, receipt or output. + +```sh +python scripts/collect_finisher_e2e.py \ + --policy /ABSOLUTE/PRIVATE/owner-policy.json \ + --policy-sha256 ACTUAL_POLICY_FILE_SHA256 \ + --output /ABSOLUTE/PRIVATE/new-unsigned-observations.json +``` + +The policy file must be regular, at most 64 KiB, owned by root or the observer +identity, and not writable by group/other. Its exact bytes must match the pin. +The output parent must be a canonical private directory owned by the observer; +output is created exclusively with mode 0600. No existing output is overwritten. +Exit 0 means collection succeeded **unsigned**; exit 2 is a verified hold; exit 3 +is invalid input or I/O failure. Fixed reason codes do not include raw provider +responses or credentials. Structured logs contain only operation names, receipt +identifiers and hashes. + +Policy fields: + +| Field | Required value | +| --- | --- | +| `schema` | `echo.finisher.collector-policy.v1` | +| `tenant_id`, `collector_id`, `authority_id`, `execution_id`, `admission_id` | Actual distinct scoped identities; bounded identifiers | +| `bindings.target_identity_digest`, `environment_identity_digest` | Exact canonical CertForge identity hashes | +| `bindings.source_commit` | Exact native or transport source being certified | +| `bindings.native_source_commit`, `transport_source_commit` | Full 40-character immutable source commits | +| `bindings.artifact_sha256`, `native_artifact_sha256` | Actual artifact byte hashes, not a tag or source SHA | +| `bindings.image_digest` | Actual `sha256:` OCI index/image digest, distinct from archive byte hash | +| `bindings.account_id`, `application_id`, `durable_object_id` | Complete actual Cloudflare tuple; 32 hex account and 64 hex DO IDs | +| `bindings.installation_id`, `owner_id`, `cloudflare_origin` | Exact installation, admitted owner and HTTPS origin without userinfo/query | +| `rollback_artifact_sha256`, `rollback_source_commit` | Actual distinct compatible prior artifact/source | +| `providers` | Exactly claude/codex/gemini/grok, each with exact `model` and nonpersonal `account_sha256` | +| `probe.command` | Exactly `[ABSOLUTE_INTERPRETER, ABSOLUTE_REVIEWED_PROBE_SCRIPT]` | +| `probe.sha256` | Exact SHA256 pins for those two files in the same order | +| `probe.dependencies` | Absolute dependency paths mapped to immutable SHA256 pins | +| `probe.timeout` | Whole process deadline from 1 to 120 seconds, default 120 | + +The transport runs the interpreter in isolated mode, accepts no shell text or +provider flags, bounds stdin to 8192 bytes and stdout to 128 KiB, discards stderr, +and checks executable pins before and after each operation. Nonblocking stdin +and stdout share the same deadline, including a child that never reads input. +The process owns its group and reaps it on every exit. Environment inheritance is +limited to normal PATH/HOME/locale/TLS paths and opaque SOL broker/run/role/state +variables, including `SOL_STATE_DB`. No API billing or custom provider route +fallback is added. Owner admission must make pinned files/closure immutable to +untrusted concurrent writers; rehashing is not a sandbox or an administrative +grant. + +## Public owner-probe protocol + +Each invocation receives one JSON object on stdin and returns one JSON object on +stdout. The owner adapter performs real authenticated requests; it must not load +responses from fixtures, accept a caller-supplied `verified: true`, mint a grant, +or turn a blocked command into an indirect interpreter/SSH/container execution. + +Request schema `echo.finisher.owner-query.v1` contains: + +* `operation`: `admission`, `observe`, or `recall` only. +* A new unpredictable `nonce`, plus canonical `policy_sha256`. +* Exact `admission_id`, `execution_id`, `tenant_id`, `collector_id`, and complete + `bindings` as above. +* For `observe`: collection `challenge`, finite `kind` and `name` below. +* For `recall`: exact `receipt_id` returned by the previous actual observation. + +Response schema `echo.finisher.owner-response.v1` contains the identical nonce, +`request_sha256` over canonical full request, actual `authority_id`, independently +established `authenticated: true` and `authorized: true`, a fresh timezone-aware +`observed_at`, and `result`. These fields are trusted only because the owner has +independently admitted and pinned the real adapter and its authenticated service +routes. This candidate is a verifier/transport, not that authority service. + +For `admission`, result must contain `allowed: true`, `independent: true`, exact +policy hash, all request identities/bindings, an expiry in the next hour, +`operations: ["observe", "recall"]`, and `restoration_allowed: true`. The actual +admission must cover the complete finite journey including its controls, +negative checks, prior snapshot/config rollback, and owner restoration. The +collector rechecks admission before every observe and recall, and at completion. + +For `observe`, result contains a new actual `receipt_id`. The collector then +requests `recall`; recall result contains the identical ID, exact `content`, and +canonical `content_sha256`. Content schema `echo.finisher.observation.v1` contains +the receipt ID, `mode: live`, exact collection challenge/kind/name, identities and +bindings, a fresh `observed_at`, and the journey result. No receipt or provider +invocation may be reused. The complete collection is bounded to 30 minutes. + +The receipt service must preserve actual authenticated provenance and permit +exact readback. Merely echoing the just-submitted request or returning an +unverifiable success flag does not satisfy this contract. The separately +admitted signer must independently verify the receipt manifest before signing. + +## Finite journey and result contract + +The ordered fronts are `auto`, `cauto`, `gauto`, `grauto`, `CFauto`, `CFcauto`, +`CFgauto`, `CFgrauto`, mapped to claude/codex/gemini/grok respectively. + +1. `front/` performs the authorized real front journey. Its result binds + actual provider/model/account, run/worker/session and native invocation IDs; + observed model/subscription identity; exact resumed session; real scoped MCP, + hook capture and exact capture readback hashes; live heartbeat; uncertain + replay suppression; and process cleanup. It requires zero inference caused + by readonly status. `start` has the actual `singleton_observed` shape with + `already_running`, verified readback and exact target. `status_snapshot` is + the real UI snapshot, with no top-level accepted/applied/request ID. It must + be fresh, online, installation/resource/heartbeat/service verified, and bound + to the actual full Cloudflare target and transport source. `controls` contains + exactly pause/resume/halt, each with a distinct actual request ID, accepted + and applied proof, and independently observed paused/running/halted state. + An accepted write, pending journal or unavailable terminal readback must fail. +2. Immediately after **each** halt, `restoration/` requires + `method: admitted_owner_operation`, the same admission ID, previous run ID, + a new actual owner operation ID, and a fresh healthy exact-target snapshot. + This is a normal independently authorized owner bootstrap, never a synthetic + frontend start receipt. Without an actual admitted restoration route the + collection stops; it does not continue to the next front in a stopped state. +3. `negative/` checks unauthenticated, wrong_owner, wrong_installation, + wrong_source, invalid_token, expired_token, stale_control, replayed_control, + out_of_scope, redirect and forbidden_native_route. Each result must prove + denial, no side effects, and identical bounded before/after state hashes. + Credentials and personal identity never enter the receipt. A redirect check + must refuse rather than follow an unadmitted destination. +4. `rollback/restore-and-recover` requires actual prior and recovered artifact + and source readbacks, both checkpoint source matches, actual checkpoint and + configuration SHA256 values, and verified restored/recovered state. It runs + **before** final stability. The prior source and candidate source are pinned; + container replacement alone does not meet this check. +5. `stability/0`, `/1`, `/2` require healthy, resource-admitted, fresh heartbeat, + zero unexpected restarts and healthy exact-target snapshots for **all eight + fronts**. The first-to-third observation interval must be at least 30 seconds. +6. `external/acceptance` requires a distinct independent reviewer, actual Access + session, immutable UI assets, exact HTTPS origin and accepted live behavior. + +Tests use explicitly synthetic authority fixtures to exercise forged, replayed, +stale, mismatched, revoked and incomplete receipts. They also exercise real local +child processes on FORGE for small-pipe/nonreading deadlines, output bounds and +opaque broker state inheritance. None of these tests call a real provider or +produce admission, signing, deployment or production-acceptance evidence. + +## Unchanged signing and deployment boundaries + +Only after real observations and independent review may an already admitted +independent signer add its actual key identity and produce the existing signed +production-E2E envelope. The normal trusted-key verifier must accept that exact +envelope and its source/target/environment/manifest bindings before re-running +certification. Private signing keys never belong in this collector or its worker. + +The existing deployment admission hook remains unchanged: + +```sh +python /home/forge/echo-certification-forge-current/scripts/deployment_admission_hook.py \ + admit --forge-url ADMITTED_FORGE_ORIGIN --tenant ACTUAL_TENANT \ + --artifact-digest ACTUAL_ARTIFACT_BYTE_SHA256 --environment staging \ + --environment-identity-digest ACTUAL_ENVIRONMENT_IDENTITY_DIGEST \ + --rule-manifest-digest ACTUAL_RULE_MANIFEST_DIGEST \ + --deployment-id ACTUAL_DEPLOYMENT_ID --requested-by ACTUAL_ADMITTED_OWNER +``` + +Any credential required by that existing hook stays opaque in its normal owner +environment; do not paste values into this command or handoff. Staging and +production gates remain enforced. Actual outcome registration and exact readback +remain owner responsibilities after an admitted deployment. This collector's +source, tests, draft PR or unsigned output cannot clear those gates. diff --git a/scripts/collect_finisher_e2e.py b/scripts/collect_finisher_e2e.py new file mode 100644 index 0000000..c82e955 --- /dev/null +++ b/scripts/collect_finisher_e2e.py @@ -0,0 +1,67 @@ +#!/usr/bin/env python3 +"""Run the owner-admitted Finisher collector; output is never a signed verdict.""" +from __future__ import annotations + +import argparse +import hashlib +import json +import logging +import os +from pathlib import Path +import stat +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) +from echo_certification_forge.finisher_collector import ( # noqa: E402 + CollectorError, FinisherCollector, PinnedOwnerProbe, bounded_document, require, +) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--policy", type=Path, required=True) + parser.add_argument("--policy-sha256", required=True) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + logging.basicConfig(level=logging.INFO, format="%(message)s") + try: + require(os.name == "posix", "collector_platform_unsupported") + require(args.policy.is_absolute() and not args.policy.is_symlink(), "policy_path_invalid") + descriptor = os.open(args.policy, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + with os.fdopen(descriptor, "rb") as handle: + metadata = os.fstat(handle.fileno()) + require(stat.S_ISREG(metadata.st_mode) and metadata.st_size <= 65536, "policy_file_invalid") + require(metadata.st_uid in (0, os.geteuid()) and not metadata.st_mode & 0o022, "policy_file_permissions") + raw = handle.read(65537) + require(hashlib.sha256(raw).hexdigest() == args.policy_sha256, "policy_pin_mismatch") + policy = bounded_document(raw, 65536) + probe = policy.get("probe") + require(isinstance(probe, dict), "owner_probe_unconfigured") + require(args.output.is_absolute() and not args.output.exists() and not args.output.is_symlink(), "output_path_invalid") + parent = args.output.parent + require(parent.resolve(strict=True) == parent and parent.is_dir(), "output_parent_invalid") + metadata = parent.stat() + require(metadata.st_uid == os.geteuid() and not metadata.st_mode & 0o077, "output_parent_not_private") + owner = PinnedOwnerProbe(probe.get("command"), probe.get("sha256"), + dependencies=probe.get("dependencies"), timeout=probe.get("timeout", 120)) + result = FinisherCollector(policy, owner).collect() + data = (json.dumps(result, sort_keys=True, separators=(",", ":")) + "\n").encode() + descriptor = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) + with os.fdopen(descriptor, "wb") as handle: + handle.write(data) + handle.flush() + os.fsync(handle.fileno()) + print(json.dumps({"status": "UNSIGNED_CANDIDATE", "release_verdict": "NOT_READY", + "signature_verified": False, "sha256": hashlib.sha256(data).hexdigest(), + "observations": len(result["evidence"])})) + return 0 + except CollectorError as exc: + print(json.dumps({"status": "HELD", "release_verdict": "NOT_READY", "reason": str(exc)})) + return 2 + except (OSError, ValueError, TypeError, KeyError): + print(json.dumps({"status": "HELD", "release_verdict": "NOT_READY", "reason": "collector_input_or_io_failed"})) + return 3 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/echo_certification_forge/finisher_collector.py b/src/echo_certification_forge/finisher_collector.py new file mode 100644 index 0000000..e8a5b58 --- /dev/null +++ b/src/echo_certification_forge/finisher_collector.py @@ -0,0 +1,431 @@ +"""Finisher observation collection through a separately admitted owner probe. + +This module never signs, deploys, enrolls a worker or changes a release policy. +The owner must independently review/admit the probe and pin its entire closure. +Successful collection produces an UNSIGNED_CANDIDATE, always NOT_READY. +""" +from __future__ import annotations + +from collections.abc import Callable, Mapping +from datetime import UTC, datetime, timedelta +import hashlib +import json +import logging +import os +from pathlib import Path +import re +import selectors +import signal +import subprocess +import time +from typing import Any, Protocol +from urllib.parse import urlsplit +import uuid + +from .canonical import sha256_json +from .production_e2e import BASE_CHECKS, GENERIC_PROFILE, SCHEMA_VERSION, _contains_restricted_value + +FRONTS = {"auto": "claude", "cauto": "codex", "gauto": "gemini", "grauto": "grok", + "CFauto": "claude", "CFcauto": "codex", "CFgauto": "gemini", "CFgrauto": "grok"} +NEGATIVE_CONTROLS = ("unauthenticated", "wrong_owner", "wrong_installation", "wrong_source", + "invalid_token", "expired_token", "stale_control", "replayed_control", + "out_of_scope", "redirect", "forbidden_native_route") +CONTROL_STATES = {"pause": "paused", "resume": "running", "halt": "halted"} +IDENTIFIER = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}\Z") +DIGEST = re.compile(r"[0-9a-f]{64}\Z") +COMMIT = re.compile(r"[0-9a-f]{40}\Z") +LOG = logging.getLogger(__name__) + + +class CollectorError(RuntimeError): + """A fixed nonsecret reason code; raw probe responses are never interpolated.""" + + +class OwnerProbe(Protocol): + def call(self, request: dict[str, Any]) -> dict[str, Any]: ... + + +def require(condition: bool, reason: str) -> None: + if not condition: + raise CollectorError(reason) + + +def identifier(value: Any) -> bool: + return isinstance(value, str) and IDENTIFIER.fullmatch(value) is not None + + +def digest(value: Any) -> bool: + return isinstance(value, str) and DIGEST.fullmatch(value) is not None + + +def timestamp(value: Any) -> datetime: + require(isinstance(value, str) and len(value) <= 64, "time_invalid") + try: + result = datetime.fromisoformat(value.replace("Z", "+00:00")) + require(result.tzinfo is not None, "time_invalid") + return result.astimezone(UTC) + except (ValueError, OverflowError) as exc: + raise CollectorError("time_invalid") from exc + + +def bounded_document(data: bytes, limit: int = 131072) -> dict[str, Any]: + require(len(data) <= limit, "document_size") + def pairs(items): + result = {} + for key, value in items: + require(key not in result, "document_duplicate_key") + result[key] = value + return result + try: + result = json.loads(data, object_pairs_hook=pairs, + parse_constant=lambda _: (_ for _ in ()).throw(CollectorError("document_number"))) + except (ValueError, RecursionError, UnicodeError) as exc: + raise CollectorError("document_invalid") from exc + require(isinstance(result, dict), "document_invalid") + try: + require(not _contains_restricted_value(result), "restricted_content") + except RecursionError as exc: + raise CollectorError("document_depth") from exc + return result + + +class PinnedOwnerProbe: + """Bounded Linux process transport for a previously reviewed owner probe. + + Hashes are integrity pins, NOT administrative admission. The probe must query + the real authenticated control plane for admission and every observation. + No command, shell text, interpreter switch or provider argument is accepted + from an observation. Dependencies must be pinned by the owner as well. + """ + def __init__(self, command: list[str], pins: list[str], *, + dependencies: Mapping[str, str] | None = None, timeout: float = 120): + require(isinstance(command, list) and len(command) == 2 and isinstance(pins, list) and len(pins) == 2, + "probe_command_invalid") + require(type(timeout) in (int, float) and 1 <= timeout <= 120, "probe_timeout_invalid") + try: + self.command = [str(Path(item).resolve(strict=True)) for item in command] + except (OSError, ValueError, TypeError) as exc: + raise CollectorError("probe_command_invalid") from exc + self.pins = dict(zip(self.command, pins, strict=True)) + for path, pin in (dependencies or {}).items(): + resolved = str(Path(path).resolve(strict=True)) + require(resolved not in self.pins, "probe_pin_duplicate") + self.pins[resolved] = pin + self.timeout = timeout + self._verify() + + def _verify(self) -> None: + for name, expected in self.pins.items(): + require(digest(expected), "probe_pin_invalid") + path = Path(name) + require(path.is_file() and not path.is_symlink(), "probe_pin_file") + value = hashlib.sha256() + with path.open("rb") as handle: + for block in iter(lambda: handle.read(65536), b""): + value.update(block) + require(value.hexdigest() == expected, "probe_pin_mismatch") + + def call(self, request: dict[str, Any]) -> dict[str, Any]: + self._verify() + require(os.name == "posix", "probe_platform_unsupported") + data = json.dumps(request, sort_keys=True, separators=(",", ":")).encode() + require(len(data) <= 8192, "probe_request_size") + # Credentials remain opaque in the inherited normal scoped broker path. + allowed = {"PATH", "HOME", "LANG", "LC_ALL", "SSL_CERT_FILE", "SSL_CERT_DIR", + "SOL_BROKER_TOKEN", "SOL_RUN_ID", "SOL_STATE", "SOL_STATE_DB", "SOL_ROLE"} + environment = {name: value for name, value in os.environ.items() if name in allowed} + environment["PYTHONDONTWRITEBYTECODE"] = "1" + deadline = time.monotonic() + self.timeout + process = subprocess.Popen([self.command[0], "-I", self.command[1]], stdin=subprocess.PIPE, + stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, + env=environment, start_new_session=True, close_fds=True, bufsize=0) + selector = selectors.DefaultSelector() + output = bytearray() + try: + # Both directions share one deadline, including a child that never + # reads stdin. No assumption about the actual pipe capacity is made. + os.set_blocking(process.stdin.fileno(), False) + os.set_blocking(process.stdout.fileno(), False) + selector.register(process.stdin, selectors.EVENT_WRITE) + selector.register(process.stdout, selectors.EVENT_READ) + sent = 0 + while selector.get_map(): + require(time.monotonic() < deadline, "probe_timeout") + for key, _ in selector.select(min(.1, max(0, deadline - time.monotonic()))): + try: + if key.fileobj is process.stdin: + sent += os.write(process.stdin.fileno(), data[sent:sent + 4096]) + if sent == len(data): + selector.unregister(process.stdin) + process.stdin.close() + continue + block = os.read(key.fileobj.fileno(), 16384) + except BlockingIOError: + continue + if not block: + selector.unregister(key.fileobj) + continue + output.extend(block) + require(len(output) <= 131072, "probe_output_size") + try: + process.wait(timeout=max(.01, deadline - time.monotonic())) + except subprocess.TimeoutExpired as exc: + raise CollectorError("probe_timeout") from exc + require(process.returncode == 0, "probe_failed") + self._verify() + return bounded_document(bytes(output)) + except (OSError, subprocess.SubprocessError) as exc: + raise CollectorError("probe_transport_failed") from exc + finally: + selector.close() + # The probe owns this process group, including any unsettled children. + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + process.wait(timeout=5) + process.stdout.close() + process.stdin.close() + + +class FinisherCollector: + def __init__(self, policy: dict[str, Any], probe: OwnerProbe, *, + clock: Callable[[], datetime] = lambda: datetime.now(UTC)): + self.policy = bounded_document(json.dumps(policy, allow_nan=False).encode(), 65536) + self.probe, self.clock = probe, clock + p = self.policy + require(p.get("schema") == "echo.finisher.collector-policy.v1", "policy_schema") + require(all(identifier(p.get(key)) for key in + ("tenant_id", "collector_id", "authority_id", "execution_id", "admission_id")), "policy_identity") + bindings = p.get("bindings") + require(isinstance(bindings, dict), "policy_bindings") + expected_keys = {"target_identity_digest", "environment_identity_digest", "source_commit", + "native_source_commit", "transport_source_commit", "artifact_sha256", + "native_artifact_sha256", "image_digest", "installation_id", "owner_id", "cloudflare_origin", + "account_id", "application_id", "durable_object_id"} + require(set(bindings) == expected_keys, "policy_bindings") + require(all(digest(bindings[key]) for key in + ("target_identity_digest", "environment_identity_digest", "artifact_sha256", "native_artifact_sha256")), "policy_digest") + require(all(isinstance(bindings[key], str) and COMMIT.fullmatch(bindings[key]) for key in + ("source_commit", "native_source_commit", "transport_source_commit")), "policy_source") + require(bindings["source_commit"] in (bindings["native_source_commit"], bindings["transport_source_commit"]), "policy_source") + require(isinstance(bindings["image_digest"], str) and bindings["image_digest"].startswith("sha256:") + and digest(bindings["image_digest"][7:]), "policy_image") + require(identifier(bindings["installation_id"]) and identifier(bindings["owner_id"]), "policy_installation") + require(isinstance(bindings["account_id"], str) and re.fullmatch("[a-f0-9]{32}", bindings["account_id"]) + and digest(bindings["durable_object_id"]) and isinstance(bindings["application_id"], str) + and re.fullmatch("[A-Za-z0-9_-]{1,128}", bindings["application_id"]), "policy_cloudflare_identity") + origin = urlsplit(bindings["cloudflare_origin"]) + require(origin.scheme == "https" and bool(origin.hostname) and not origin.username and not origin.password + and origin.path in ("", "/") and not origin.query and not origin.fragment, "policy_origin") + require(digest(p.get("rollback_artifact_sha256")) and p["rollback_artifact_sha256"] != bindings["artifact_sha256"], "policy_rollback") + require(isinstance(p.get("rollback_source_commit"), str) + and COMMIT.fullmatch(p["rollback_source_commit"]) + and p["rollback_source_commit"] != bindings["source_commit"], "policy_rollback") + require(isinstance(p.get("providers"), dict) and set(p["providers"]) == set(FRONTS.values()), "policy_providers") + for provider in p["providers"].values(): + require(isinstance(provider, dict) and set(provider) == {"model", "account_sha256"} + and identifier(provider["model"]) and digest(provider["account_sha256"]), "policy_provider") + self.policy_hash = sha256_json(p) + self.challenge = uuid.uuid4().hex + self.started = self.clock() + require(self.started.tzinfo is not None, "clock_invalid") + self.ids: set[str] = set() + self.invocations: set[str] = set() + self.control_ids: set[str] = set() + self.restoration_ids: set[str] = set() + self.evidence: list[dict[str, str]] = [] + + def _query(self, operation: str, **extra: Any) -> dict[str, Any]: + now = self.clock() + require(0 <= (now - self.started).total_seconds() <= 1800, "collection_deadline") + request = {"schema": "echo.finisher.owner-query.v1", "operation": operation, + "nonce": uuid.uuid4().hex, "policy_sha256": self.policy_hash, + **{key: self.policy[key] for key in + ("admission_id", "execution_id", "tenant_id", "collector_id", "bindings")}, **extra} + LOG.info(json.dumps({"event": "owner_query", "operation": operation, "policy_sha256": self.policy_hash})) + response = self.probe.call(request) + require(isinstance(response, dict), "owner_response_invalid") + response = bounded_document(json.dumps(response, allow_nan=False).encode()) + require(response.get("schema") == "echo.finisher.owner-response.v1" + and response.get("nonce") == request["nonce"] + and response.get("request_sha256") == sha256_json(request) + and response.get("authority_id") == self.policy["authority_id"] + and response.get("authenticated") is True and response.get("authorized") is True, + "owner_response_binding") + observed = timestamp(response.get("observed_at")) + require(now - timedelta(seconds=1) <= observed <= self.clock() + timedelta(seconds=1), "owner_response_time") + require(isinstance(response.get("result"), dict), "owner_response_result") + return response["result"] + + def _admission(self) -> None: + result = self._query("admission") + require(result.get("allowed") is True and result.get("independent") is True, "admission_denied") + require(all(result.get(key) == self.policy[key] for key in + ("admission_id", "collector_id", "execution_id", "tenant_id", "bindings")), "admission_binding") + require(result.get("policy_sha256") == self.policy_hash and result.get("restoration_allowed") is True, + "admission_journey_not_allowed") + expiry = timestamp(result.get("expires_at")) + require(self.clock() < expiry <= self.clock() + timedelta(hours=1), "admission_expired") + require(result.get("operations") == ["observe", "recall"], "admission_operations") + + def _observe(self, kind: str, name: str) -> tuple[dict[str, Any], datetime]: + self._admission() + produced = self._query("observe", challenge=self.challenge, kind=kind, name=name) + receipt_id = produced.get("receipt_id") + require(identifier(receipt_id) and receipt_id not in self.ids, "receipt_replay") + self.ids.add(receipt_id) + self._admission() + recalled = self._query("recall", receipt_id=receipt_id) + content = recalled.get("content") + require(recalled.get("receipt_id") == receipt_id and isinstance(content, dict), "receipt_readback") + require(digest(recalled.get("content_sha256")) and recalled["content_sha256"] == sha256_json(content), "receipt_hash") + require(content.get("schema") == "echo.finisher.observation.v1" + and content.get("receipt_id") == receipt_id and content.get("mode") == "live" + and content.get("challenge") == self.challenge and content.get("kind") == kind + and content.get("name") == name, "observation_binding") + require(all(content.get(key) == self.policy[key] for key in + ("tenant_id", "execution_id", "collector_id", "bindings")), "observation_identity") + observed = timestamp(content.get("observed_at")) + require(self.started <= observed <= self.clock() and (self.clock() - observed).total_seconds() <= 900, "observation_time") + result = content.get("result") + require(isinstance(result, dict), "observation_result") + self.evidence.append({"kind": kind, "name": name, "receipt_id": receipt_id, + "sha256": recalled["content_sha256"], "observed_at": observed.isoformat()}) + LOG.info(json.dumps({"event": "observation_recalled", "kind": kind, "name": name, + "receipt_id": receipt_id, "sha256": recalled["content_sha256"]})) + return result, observed + + def _target(self, target: Any, run_id: str | None = None) -> None: + binding = self.policy["bindings"] + expected = {"provider": "cloudflare", **{key: binding[key] for key in + ("account_id", "application_id", "durable_object_id", "installation_id")}, + "source_sha": binding["transport_source_commit"]} + require(isinstance(target, dict) and set(target) == {*expected, "run_id"} + and all(target.get(key) == value for key, value in expected.items()) + and identifier(target.get("run_id")) and (run_id is None or target["run_id"] == run_id), + "snapshot_target") + + def _snapshot(self, value: Any, run_id: str | None = None) -> None: + require(isinstance(value, dict) and value.get("schema") == "echo.finisher.ui.snapshot.v1" + and not any(key in value for key in ("accepted", "applied", "request_id")), "status_snapshot_invalid") + self._target(value.get("target"), run_id) + require(value.get("source_sha") == self.policy["bindings"]["transport_source_commit"] + and value.get("online") is True and value.get("readback_verified") is True + and value.get("lifecycle") in ("online", "idle", "running", "executing"), "status_snapshot_unhealthy") + checks = value.get("checks") + require(isinstance(checks, dict) and all(checks.get(key) is True for key in + ("service_active", "installation_verified", "heartbeat_fresh", "resource_admitted")), "status_snapshot_unhealthy") + stamp = value.get("observed_at") + if type(stamp) in (int, float): + require(0 <= stamp <= 253402300799, "status_snapshot_time") + stamp = datetime.fromtimestamp(stamp, UTC) + else: + stamp = timestamp(stamp) + require(self.started <= stamp <= self.clock() and (self.clock() - stamp).total_seconds() <= 30, + "status_snapshot_time") + + def _front(self, name: str, result: dict[str, Any]) -> None: + provider = FRONTS[name] + require(result.get("provider") == provider and all(result.get(key) == value for key, value in + self.policy["providers"][provider].items()), "front_provider_identity") + require(all(result.get(key) is True for key in ("model_observed", "account_verified", "subscription_auth", + "heartbeat_verified", "scoped_mcp_verified", "uncertain_replay_suppressed", "cleanup_verified")), "front_incomplete") + require(type(result.get("status_inference_count")) is int and result["status_inference_count"] == 0, "front_status_inference") + require(all(identifier(result.get(key)) for key in ("run_id", "worker_id", "session_id", + "provider_invocation_id", "hook_receipt_id", "capture_receipt_id")), "front_receipt_missing") + require(result.get("resumed_session_id") == result["session_id"], "front_resume_mismatch") + require(digest(result.get("hook_readback_sha256")) and digest(result.get("capture_readback_sha256")), "front_capture_readback") + require(result["provider_invocation_id"] not in self.invocations, "provider_replay") + self.invocations.add(result["provider_invocation_id"]) + self._snapshot(result.get("status_snapshot"), result["run_id"]) + start = result.get("start") + require(isinstance(start, dict) and start.get("state") == "singleton_observed" + and start.get("already_running") is True and start.get("readback_verified") is True, + "front_start_unverified") + self._target(start.get("target"), result["run_id"]) + controls = result.get("controls") + require(isinstance(controls, dict) and set(controls) == set(CONTROL_STATES), "front_controls_missing") + for action, expected in CONTROL_STATES.items(): + control = controls[action] + require(isinstance(control, dict) and identifier(control.get("request_id")), "front_control_invalid") + require(control["request_id"] not in self.control_ids, "control_replay") + self.control_ids.add(control["request_id"]) + require(control.get("accepted") is True and control.get("applied") is True and control.get("state") == expected, + "front_control_not_applied") + + def _restoration(self, name: str, previous_run_id: str) -> None: + result, _ = self._observe("restoration", name) + require(result.get("method") == "admitted_owner_operation" + and result.get("admission_id") == self.policy["admission_id"] + and result.get("previous_run_id") == previous_run_id + and identifier(result.get("operation_id")) + and result["operation_id"] not in self.restoration_ids, "restoration_unverified") + self.restoration_ids.add(result["operation_id"]) + self._snapshot(result.get("snapshot")) + + def _rollback(self) -> None: + rollback, _ = self._observe("rollback", "restore-and-recover") + require(rollback.get("verified") is True and rollback.get("restored_readback_verified") is True + and rollback.get("recovered_readback_verified") is True + and rollback.get("restored_artifact_sha256") == self.policy["rollback_artifact_sha256"] + and rollback.get("recovered_artifact_sha256") == self.policy["bindings"]["artifact_sha256"] + and rollback.get("restored_source_commit") == self.policy["rollback_source_commit"] + and rollback.get("restored_checkpoint_source_commit") == self.policy["rollback_source_commit"] + and rollback.get("recovered_source_commit") == self.policy["bindings"]["source_commit"] + and rollback.get("recovered_checkpoint_source_commit") == self.policy["bindings"]["source_commit"] + and all(digest(rollback.get(key)) for key in ("restored_checkpoint_sha256", "recovered_checkpoint_sha256", + "restored_configuration_sha256", "recovered_configuration_sha256")), "rollback_failed") + + def collect(self) -> dict[str, Any]: + require(not self.evidence, "collector_already_used") + self._admission() + for front in FRONTS: + result, _ = self._observe("front", front) + self._front(front, result) + # A halted singleton cannot serve the next front. Its independent + # owner must use an already admitted restoration operation; frontend + # start reconciliation itself is not a bootstrap or reset grant. + self._restoration(front, result["run_id"]) + for name in NEGATIVE_CONTROLS: + result, _ = self._observe("negative", name) + require(result.get("denied") is True and result.get("side_effects") is False + and digest(result.get("before_sha256")) and result.get("before_sha256") == result.get("after_sha256"), "negative_control_failed") + # Restore the prior compatible checkpoint/config and then the candidate + # before measuring final health. A Worker image rollback alone is not + # checkpoint restoration. + self._rollback() + probes = [] + for index in range(3): + result, observed = self._observe("stability", str(index)) + require(all(result.get(key) is True for key in ("healthy", "resource_admitted", "heartbeat_verified")) + and type(result.get("unexpected_restarts")) is int and result["unexpected_restarts"] == 0, "stability_failed") + snapshots = result.get("snapshots") + require(isinstance(snapshots, dict) and set(snapshots) == set(FRONTS), "stability_fronts_missing") + for snapshot in snapshots.values(): + self._snapshot(snapshot) + probes.append(observed) + require(probes[0] < probes[1] < probes[2] and (probes[2] - probes[0]).total_seconds() >= 30, "stability_interval") + external, _ = self._observe("external", "acceptance") + require(external.get("accepted") is True and external.get("ui_assets_verified") is True + and external.get("access_session_verified") is True + and identifier(external.get("reviewer_id")) and external["reviewer_id"] != self.policy["collector_id"] + and external.get("origin") == self.policy["bindings"]["cloudflare_origin"], "external_acceptance_failed") + self._admission() + finished = self.clock() + require((finished - self.started).total_seconds() <= 1800, "collection_deadline") + payload = {"schema_version": SCHEMA_VERSION, "profile": GENERIC_PROFILE, + "attestation_id": "finisher-" + self.challenge, + "target_identity_digest": self.policy["bindings"]["target_identity_digest"], + "environment_identity_digest": self.policy["bindings"]["environment_identity_digest"], + "source_commit": self.policy["bindings"]["source_commit"], + "deployment_sha": self.policy["bindings"]["source_commit"], + "observed_at": self.started.isoformat(), + "expires_at": (self.started + timedelta(hours=1)).isoformat(), + "required_checks": sorted(BASE_CHECKS), "checks": {key: True for key in sorted(BASE_CHECKS)}, + "stability_probe_count": len(probes), "collector_execution_id": self.policy["execution_id"], + "observation_manifest_sha256": sha256_json(self.evidence), "policy_sha256": self.policy_hash} + return {"schema": "echo.finisher.unsigned-collection.v1", "status": "UNSIGNED_CANDIDATE", + "release_verdict": "NOT_READY", "signature_verified": False, "owner_signing_required": True, + "attestation_payload": payload, "evidence": self.evidence, + "finished_at": finished.isoformat(), "bindings": self.policy["bindings"]} diff --git a/tests/test_finisher_collector.py b/tests/test_finisher_collector.py new file mode 100644 index 0000000..9feb5f7 --- /dev/null +++ b/tests/test_finisher_collector.py @@ -0,0 +1,403 @@ +"""Synthetic authority fixtures exercise rejection paths; none is live evidence.""" +from copy import deepcopy +from datetime import UTC, datetime, timedelta +import json +import hashlib +import os +import sys +import subprocess +import time +from pathlib import Path + +import pytest + +from echo_certification_forge.canonical import sha256_json +from echo_certification_forge.finisher_collector import ( + CollectorError, FRONTS, NEGATIVE_CONTROLS, FinisherCollector, PinnedOwnerProbe, +) +from echo_certification_forge.production_e2e import BASE_CHECKS, validate_attestation_trust_metadata + +NOW = datetime(2026, 11, 7, 6, 0, tzinfo=UTC) + + +def policy(): + return { + "schema": "echo.finisher.collector-policy.v1", "tenant_id": "tenant-fixture", + "collector_id": "independent-fixture", "authority_id": "owner-fixture", + "execution_id": "execution-fixture", "admission_id": "admission-fixture", + "bindings": {"target_identity_digest": "1" * 64, + "environment_identity_digest": "2" * 64, + "source_commit": "3" * 40, "native_source_commit": "4" * 40, + "transport_source_commit": "3" * 40, "artifact_sha256": "5" * 64, + "image_digest": "sha256:" + "6" * 64, + "native_artifact_sha256": "7" * 64, + "installation_id": "installation-fixture", "owner_id": "owner-fixture", + "account_id": "a" * 32, "application_id": "application-fixture", + "durable_object_id": "b" * 64, + "cloudflare_origin": "https://finisher.example.test"}, + "rollback_artifact_sha256": "8" * 64, + "rollback_source_commit": "9" * 40, + "providers": {v: {"model": v + "-fixture-model", "account_sha256": "a" * 64} + for v in set(FRONTS.values())}, + } + + +class OwnerFixture: + """In-memory test authority, not a provider, deployment or signing service.""" + def __init__(self, config): + self.config = deepcopy(config) + self.time = NOW + self.receipts = {} + self.calls = [] + self.modify = lambda value: value + self.admitted = True + self.nonce_override = None + + def clock(self): + return self.time + + def snapshot(self, name): + binding = self.config["bindings"] + return {"schema": "echo.finisher.ui.snapshot.v1", "observed_at": self.time.isoformat(), + "source_sha": binding["transport_source_commit"], "online": True, + "lifecycle": "idle", "readback_verified": True, + "checks": {key: True for key in ("installation_verified", "service_active", + "heartbeat_fresh", "resource_admitted")}, + "target": {"provider": "cloudflare", **{key: binding[key] for key in + ("account_id", "application_id", "durable_object_id", "installation_id")}, + "source_sha": binding["transport_source_commit"], "run_id": "run-" + name}} + + def call(self, request): + self.calls.append(deepcopy(request)) + self.time += timedelta(seconds=1) + op = request["operation"] + common = {"schema": "echo.finisher.owner-response.v1", "nonce": request["nonce"], + "request_sha256": sha256_json(request), "authority_id": self.config["authority_id"], + "authenticated": True, "authorized": True, + "observed_at": self.time.isoformat()} + if self.nonce_override: + common["nonce"] = self.nonce_override + if op == "admission": + value = {"allowed": self.admitted, "admission_id": self.config["admission_id"], + "collector_id": self.config["collector_id"], + "execution_id": self.config["execution_id"], "tenant_id": self.config["tenant_id"], + "bindings": self.config["bindings"], "expires_at": (self.time + timedelta(minutes=15)).isoformat(), + "operations": ["observe", "recall"], "independent": True, + "policy_sha256": request["policy_sha256"], "restoration_allowed": True} + elif op == "observe": + kind, name = request["kind"], request["name"] + if kind == "stability": + self.time += timedelta(seconds=15) + receipt_id = "receipt-" + str(len(self.receipts)) + value = {"receipt_id": receipt_id} + data = {"schema": "echo.finisher.observation.v1", "receipt_id": receipt_id, + "tenant_id": self.config["tenant_id"], "execution_id": self.config["execution_id"], + "collector_id": self.config["collector_id"], "bindings": deepcopy(self.config["bindings"]), + "challenge": request["challenge"], "kind": kind, "name": name, + "observed_at": self.time.isoformat(), "mode": "live"} + if kind == "front": + provider = FRONTS[name] + data["result"] = { + "provider": provider, **self.config["providers"][provider], + "provider_invocation_id": "invoke-" + name, + "model_observed": True, "account_verified": True, "subscription_auth": True, + "run_id": "run-" + name, "worker_id": "worker-" + name, + "session_id": "session-" + name, "resumed_session_id": "session-" + name, + "heartbeat_verified": True, "scoped_mcp_verified": True, + "hook_receipt_id": "hook-" + name, "capture_receipt_id": "capture-" + name, + "hook_readback_sha256": "b" * 64, "capture_readback_sha256": "c" * 64, + "uncertain_replay_suppressed": True, "status_inference_count": 0, + "status_snapshot": self.snapshot(name), + "start": {"state": "singleton_observed", "already_running": True, + "readback_verified": True, "target": self.snapshot(name)["target"]}, + "controls": {action: {"request_id": name + "-" + action, "accepted": True, + "applied": True, "state": state} + for action, state in {"pause": "paused", "resume": "running", "halt": "halted"}.items()}, + "cleanup_verified": True, + } + elif kind == "restoration": + data["result"] = {"method": "admitted_owner_operation", "operation_id": "restore-" + name, + "admission_id": self.config["admission_id"], + "previous_run_id": "run-" + name, "snapshot": self.snapshot(name)} + elif kind == "negative": + data["result"] = {"denied": True, "side_effects": False, + "before_sha256": "d" * 64, "after_sha256": "d" * 64} + elif kind == "stability": + data["result"] = {"healthy": True, "resource_admitted": True, + "heartbeat_verified": True, "unexpected_restarts": 0, + "snapshots": {front: self.snapshot(front) for front in FRONTS}} + elif kind == "external": + data["result"] = {"accepted": True, "reviewer_id": "other-reviewer", + "ui_assets_verified": True, "access_session_verified": True, + "origin": self.config["bindings"]["cloudflare_origin"]} + elif kind == "rollback": + data["result"] = {"restored_artifact_sha256": self.config["rollback_artifact_sha256"], + "recovered_artifact_sha256": self.config["bindings"]["artifact_sha256"], + "restored_source_commit": self.config["rollback_source_commit"], + "recovered_source_commit": self.config["bindings"]["source_commit"], + "restored_checkpoint_source_commit": self.config["rollback_source_commit"], + "recovered_checkpoint_source_commit": self.config["bindings"]["source_commit"], + "restored_checkpoint_sha256": "1" * 64, + "recovered_checkpoint_sha256": "2" * 64, + "restored_configuration_sha256": "3" * 64, + "recovered_configuration_sha256": "4" * 64, + "restored_readback_verified": True, "recovered_readback_verified": True, + "verified": True} + self.receipts[receipt_id] = self.modify(data) + else: + data = deepcopy(self.receipts[request["receipt_id"]]) + value = {"receipt_id": request["receipt_id"], "content": data, "content_sha256": sha256_json(data)} + return {**common, "result": value} + + +def collect(config=None, fixture=None): + config = config or policy() + fixture = fixture or OwnerFixture(config) + return FinisherCollector(config, fixture, clock=fixture.clock).collect() + + +def test_all_eight_fronts_are_observed_then_independently_recalled(): + config = policy() + owner = OwnerFixture(config) + result = collect(config, owner) + assert result["status"] == "UNSIGNED_CANDIDATE" + assert result["release_verdict"] == "NOT_READY" + assert result["signature_verified"] is False + assert result["owner_signing_required"] is True + assert "signing_key_id" not in result["attestation_payload"] + assert validate_attestation_trust_metadata(result["attestation_payload"]) == (False, "production_e2e_signature_not_verified") + assert set(result["attestation_payload"]["checks"]) == BASE_CHECKS + assert {x["name"] for x in owner.calls if x["operation"] == "observe" and x["kind"] == "front"} == set(FRONTS) + observed = [x for x in owner.calls if x["operation"] == "observe"] + recalled = [x for x in owner.calls if x["operation"] == "recall"] + assert len(observed) == len(recalled) == 16 + len(NEGATIVE_CONTROLS) + 3 + 2 + journey = [(x["kind"], x["name"]) for x in observed] + for index, name in enumerate(FRONTS): + assert journey[index * 2:index * 2 + 2] == [("front", name), ("restoration", name)] + assert journey.index(("rollback", "restore-and-recover")) < journey.index(("stability", "0")) + + +def test_missing_admission_stops_before_any_observation(): + owner = OwnerFixture(policy()) + owner.admitted = False + with pytest.raises(CollectorError, match="admission_denied"): + collect(fixture=owner) + assert [x["operation"] for x in owner.calls] == ["admission"] + + +@pytest.mark.parametrize("field,value", [("nonce", "forged"), ("request_sha256", "0" * 64), + ("authenticated", False), ("authority_id", "candidate-author")]) +def test_forged_owner_response_rejected(field, value): + owner = OwnerFixture(policy()) + original = owner.call + owner.call = lambda request: {**original(request), field: value} + with pytest.raises(CollectorError, match="owner_response"): + collect(fixture=owner) + + +@pytest.mark.parametrize("mutation", [ + lambda d: d.update(mode="fixture"), + lambda d: d.update(challenge="prior-run"), + lambda d: d["bindings"].update(source_commit="f" * 40), + lambda d: d["bindings"].update(durable_object_id="f" * 64), + lambda d: d["result"]["status_snapshot"]["target"].update(account_id="f" * 32), + lambda d: d["result"]["status_snapshot"].update(online=False), + lambda d: d["result"]["status_snapshot"].update(request_id="not-a-mutation"), + lambda d: d["result"]["controls"]["halt"].update(state="stopped"), + lambda d: d.update(observed_at=(NOW - timedelta(hours=2)).isoformat()), + lambda d: d["result"].update(model="invented"), + lambda d: d["result"].update(account_sha256="f" * 64), + lambda d: d["result"].update(subscription_auth=False), + lambda d: d["result"].update(resumed_session_id="latest-other-session"), + lambda d: d["result"]["controls"]["halt"].update(applied=False), + lambda d: d["result"].update(status_inference_count=1), + lambda d: d["result"].update(capture_readback_sha256="bad"), +]) +def test_forged_or_incomplete_front_proof_is_not_promoted(mutation): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "front": + mutation(data) + return data + owner.modify = change + with pytest.raises(CollectorError): + collect(fixture=owner) + + +def test_recall_hash_mismatch_is_rejected(): + owner = OwnerFixture(policy()) + original = owner.call + def wrong(request): + value = original(request) + if request["operation"] == "recall": + value["result"]["content_sha256"] = "f" * 64 + return value + owner.call = wrong + with pytest.raises(CollectorError, match="receipt_hash"): + collect(fixture=owner) + + +@pytest.mark.parametrize("kind,field,value", [ + ("negative", "side_effects", True), ("negative", "after_sha256", "e" * 64), + ("stability", "unexpected_restarts", 1), ("stability", "healthy", False), + ("external", "reviewer_id", "independent-fixture"), + ("external", "access_session_verified", False), + ("rollback", "restored_artifact_sha256", "f" * 64), + ("rollback", "restored_checkpoint_source_commit", "f" * 40), + ("rollback", "recovered_readback_verified", False), + ("restoration", "method", "unadmitted_reset"), + ("restoration", "previous_run_id", "wrong-run"), +]) +def test_negative_stability_external_and_rollback_fail_closed(kind, field, value): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == kind: + data["result"][field] = value + return data + owner.modify = change + with pytest.raises(CollectorError): + collect(fixture=owner) + + +def test_sensitive_receipt_content_is_rejected_without_echoing_value(): + owner = OwnerFixture(policy()) + owner.modify = lambda data: {**data, "authorization": "test-sensitive-value"} + with pytest.raises(CollectorError) as error: + collect(fixture=owner) + assert "test-sensitive-value" not in str(error.value) + + +def test_pinned_probe_rejects_source_drift_before_execution(tmp_path): + path = tmp_path / "probe.py" + path.write_text("raise SystemExit('must not execute')") + with pytest.raises(CollectorError, match="probe_pin"): + PinnedOwnerProbe([sys.executable, str(path)], ["0" * 64, "0" * 64]) + + +def test_policy_rejects_userinfo_origin_and_boolean_counts(): + config = policy() + config["bindings"]["cloudflare_origin"] = "https://name:password@example.test" + with pytest.raises(CollectorError, match="policy"): + collect(config) + + +def test_duplicate_provider_invocation_receipt_is_rejected(): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "front": + data["result"]["provider_invocation_id"] = "reused-invocation" + return data + owner.modify = change + with pytest.raises(CollectorError, match="replay"): + collect(fixture=owner) + + +def make_probe(tmp_path, source, timeout=1): + path = tmp_path / "probe.py" + path.write_text(source) + command = [str(Path(sys.executable).resolve()), str(path)] + pins = [hashlib.sha256(Path(item).read_bytes()).hexdigest() for item in command] + return PinnedOwnerProbe(command, pins, timeout=timeout) + + +@pytest.mark.skipif(os.name != "posix", reason="Linux owner-probe transport") +def test_nonreading_probe_has_whole_request_deadline_with_small_pipe(tmp_path, monkeypatch): + import fcntl + probe = make_probe(tmp_path, "import time; time.sleep(4)") + original = subprocess.Popen + def small_pipe(*args, **kwargs): + process = original(*args, **kwargs) + fcntl.fcntl(process.stdin.fileno(), fcntl.F_SETPIPE_SZ, 4096) + return process + monkeypatch.setattr(subprocess, "Popen", small_pipe) + started = time.monotonic() + with pytest.raises(CollectorError, match="probe_timeout"): + probe.call({"padding": "x" * 6000}) + assert time.monotonic() - started < 2.5 + + +@pytest.mark.skipif(os.name != "posix", reason="Linux owner-probe transport") +def test_normal_broker_state_db_is_preserved_and_route_overrides_removed(tmp_path, monkeypatch): + monkeypatch.setenv("SOL_STATE_DB", "/private/fixture.db") + monkeypatch.setenv("GOOGLE_APPLICATION_CREDENTIALS", "/must/not/inherit") + probe = make_probe(tmp_path, "import json,os,sys; json.load(sys.stdin); print(json.dumps({'state_ok':os.environ.get('SOL_STATE_DB')=='/private/fixture.db','override_present':'GOOGLE_APPLICATION_CREDENTIALS' in os.environ}))") + assert probe.call({"request": "fixture"}) == {"state_ok": True, "override_present": False} + + +@pytest.mark.skipif(os.name != "posix", reason="Linux owner-probe transport") +def test_probe_output_overflow_is_bounded(tmp_path): + probe = make_probe(tmp_path, "import sys; sys.stdin.read(); sys.stdout.write('x'*140000); sys.stdout.flush()") + with pytest.raises(CollectorError, match="probe_output_size"): + probe.call({"request": "fixture"}) + + +def test_healthy_boolean_cannot_replace_all_fronts_stability_readback(): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "stability": + data["result"]["snapshots"]["CFgrauto"]["online"] = False + return data + owner.modify = change + with pytest.raises(CollectorError): + collect(fixture=owner) + + +@pytest.mark.parametrize("field", ["account_id", "application_id", "durable_object_id"]) +def test_policy_requires_the_complete_cloudflare_identity_tuple(field): + config = policy() + del config["bindings"][field] + with pytest.raises(CollectorError, match="policy_bindings"): + collect(config) + + +def test_admission_revocation_is_checked_before_exact_recall(): + owner = OwnerFixture(policy()) + original = owner.call + def revoke(request): + value = original(request) + if request["operation"] == "observe": + owner.admitted = False + return value + owner.call = revoke + with pytest.raises(CollectorError, match="admission_denied"): + collect(fixture=owner) + assert not any(request["operation"] == "recall" for request in owner.calls) + + +def test_admission_cannot_authorize_a_different_journey_policy(): + owner = OwnerFixture(policy()) + original = owner.call + def wrong(request): + value = original(request) + if request["operation"] == "admission": + value["result"]["policy_sha256"] = "f" * 64 + return value + owner.call = wrong + with pytest.raises(CollectorError, match="admission_journey"): + collect(fixture=owner) + + +@pytest.mark.skipif(os.name != "posix", reason="Linux collector CLI") +@pytest.mark.parametrize("case,reason", [("missing_probe", "owner_probe_unconfigured"), + ("wrong_pin", "policy_pin_mismatch"), + ("symlink", "policy_path_invalid")]) +def test_cli_refuses_missing_authority_or_bad_policy_without_output(tmp_path, case, reason): + tmp_path.chmod(0o700) + path = tmp_path / "policy.json" + raw = json.dumps(policy()).encode() + path.write_bytes(raw) + path.chmod(0o600) + pin = hashlib.sha256(raw).hexdigest() + if case == "wrong_pin": + pin = "0" * 64 + if case == "symlink": + link = tmp_path / "link.json" + link.symlink_to(path) + path = link + output = tmp_path / "unsigned.json" + process = subprocess.run([sys.executable, str(Path(__file__).resolve().parents[1] / "scripts/collect_finisher_e2e.py"), + "--policy", str(path), "--policy-sha256", pin, "--output", str(output)], + capture_output=True, text=True, timeout=5) + assert process.returncode == 2 + assert json.loads(process.stdout) == {"status": "HELD", "release_verdict": "NOT_READY", "reason": reason} + assert not output.exists() From 434452b9c455cfdea4d86372784ce3acf744894a Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Sun, 8 Nov 2026 13:18:49 -0600 Subject: [PATCH 2/4] fix: bind each Finisher control to exact native readback --- docs/FINISHER_COLLECTOR.md | 15 ++++- .../finisher_collector.py | 62 ++++++++++++++----- tests/test_finisher_collector.py | 50 ++++++++++++++- 3 files changed, 108 insertions(+), 19 deletions(-) diff --git a/docs/FINISHER_COLLECTOR.md b/docs/FINISHER_COLLECTOR.md index 3b944d0..99744c3 100644 --- a/docs/FINISHER_COLLECTOR.md +++ b/docs/FINISHER_COLLECTOR.md @@ -161,9 +161,18 @@ The ordered fronts are `auto`, `cauto`, `gauto`, `grauto`, `CFauto`, `CFcauto`, the real UI snapshot, with no top-level accepted/applied/request ID. It must be fresh, online, installation/resource/heartbeat/service verified, and bound to the actual full Cloudflare target and transport source. `controls` contains - exactly pause/resume/halt, each with a distinct actual request ID, accepted - and applied proof, and independently observed paused/running/halted state. - An accepted write, pending journal or unavailable terminal readback must fail. + exactly pause/resume/halt, each with distinct actual `request_id` and + `native_request_id`, matching `action`, exact full `target` including run, + `submitted_at`, accepted/applied proof and paused/running/halted `state`. + Each `readback` must independently match that target and transport source, + have fresh `observed_at >= submitted_at`, verified installation/readback, + and `controller.last_control_id` equal to the exact native request ID. + The actual paused/halted flags must agree with the action. Pause/resume + require active service and fresh heartbeat. Halt permits a fresh independently + established `terminal_halt_verified` observation after process exit without + pretending that the exited controller still has a live heartbeat. A stale + observation, replayed native ID, other run/installation, accepted write, + pending journal or unavailable terminal readback must fail. 2. Immediately after **each** halt, `restoration/` requires `method: admitted_owner_operation`, the same admission ID, previous run ID, a new actual owner operation ID, and a fresh healthy exact-target snapshot. diff --git a/src/echo_certification_forge/finisher_collector.py b/src/echo_certification_forge/finisher_collector.py index e8a5b58..e362cb2 100644 --- a/src/echo_certification_forge/finisher_collector.py +++ b/src/echo_certification_forge/finisher_collector.py @@ -233,6 +233,7 @@ def __init__(self, policy: dict[str, Any], probe: OwnerProbe, *, self.ids: set[str] = set() self.invocations: set[str] = set() self.control_ids: set[str] = set() + self.native_control_ids: set[str] = set() self.restoration_ids: set[str] = set() self.evidence: list[dict[str, str]] = [] @@ -306,6 +307,16 @@ def _target(self, target: Any, run_id: str | None = None) -> None: and identifier(target.get("run_id")) and (run_id is None or target["run_id"] == run_id), "snapshot_target") + def _observation_time(self, stamp: Any) -> datetime: + if type(stamp) in (int, float): + require(0 <= stamp <= 253402300799, "status_snapshot_time") + stamp = datetime.fromtimestamp(stamp, UTC) + else: + stamp = timestamp(stamp) + require(self.started <= stamp <= self.clock() and (self.clock() - stamp).total_seconds() <= 30, + "status_snapshot_time") + return stamp + def _snapshot(self, value: Any, run_id: str | None = None) -> None: require(isinstance(value, dict) and value.get("schema") == "echo.finisher.ui.snapshot.v1" and not any(key in value for key in ("accepted", "applied", "request_id")), "status_snapshot_invalid") @@ -316,14 +327,40 @@ def _snapshot(self, value: Any, run_id: str | None = None) -> None: checks = value.get("checks") require(isinstance(checks, dict) and all(checks.get(key) is True for key in ("service_active", "installation_verified", "heartbeat_fresh", "resource_admitted")), "status_snapshot_unhealthy") - stamp = value.get("observed_at") - if type(stamp) in (int, float): - require(0 <= stamp <= 253402300799, "status_snapshot_time") - stamp = datetime.fromtimestamp(stamp, UTC) - else: - stamp = timestamp(stamp) - require(self.started <= stamp <= self.clock() and (self.clock() - stamp).total_seconds() <= 30, - "status_snapshot_time") + self._observation_time(value.get("observed_at")) + + def _control(self, action: str, control: Any, run_id: str) -> None: + require(isinstance(control, dict) and identifier(control.get("request_id")) + and identifier(control.get("native_request_id")) and control.get("action") == action, + "front_control_invalid") + require(control["request_id"] not in self.control_ids + and control["native_request_id"] not in self.native_control_ids, "control_replay") + self.control_ids.add(control["request_id"]) + self.native_control_ids.add(control["native_request_id"]) + self._target(control.get("target"), run_id) + require(control.get("accepted") is True and control.get("applied") is True + and control.get("state") == CONTROL_STATES[action], "front_control_not_applied") + submitted = self._observation_time(control.get("submitted_at")) + observed = control.get("readback") + require(isinstance(observed, dict) and observed.get("target") == control["target"] + and observed.get("source_sha") == self.policy["bindings"]["transport_source_commit"] + and observed.get("readback_verified") is True, "control_readback_binding") + require(self._observation_time(observed.get("observed_at")) >= submitted, "control_readback_stale") + checks, controller = observed.get("checks"), observed.get("controller") + require(isinstance(checks, dict) and checks.get("installation_verified") is True + and isinstance(controller, dict) and controller.get("last_control_id") == control["native_request_id"], + "control_readback_identity") + require(type(controller.get("paused")) is bool and type(controller.get("halted")) is bool, + "control_readback_state") + expected = {"pause": controller["paused"] and not controller["halted"], + "resume": not controller["paused"] and not controller["halted"], + "halt": controller["halted"]} + require(expected[action], "control_readback_state") + live = checks.get("service_active") is True and checks.get("heartbeat_fresh") is True + terminal = action == "halt" and observed.get("terminal_halt_verified") is True + # A legitimate fresh terminal observation proves process exit through + # the owner observer. It must not fabricate a live controller heartbeat. + require(live or terminal, "control_readback_unavailable") def _front(self, name: str, result: dict[str, Any]) -> None: provider = FRONTS[name] @@ -346,13 +383,8 @@ def _front(self, name: str, result: dict[str, Any]) -> None: self._target(start.get("target"), result["run_id"]) controls = result.get("controls") require(isinstance(controls, dict) and set(controls) == set(CONTROL_STATES), "front_controls_missing") - for action, expected in CONTROL_STATES.items(): - control = controls[action] - require(isinstance(control, dict) and identifier(control.get("request_id")), "front_control_invalid") - require(control["request_id"] not in self.control_ids, "control_replay") - self.control_ids.add(control["request_id"]) - require(control.get("accepted") is True and control.get("applied") is True and control.get("state") == expected, - "front_control_not_applied") + for action in CONTROL_STATES: + self._control(action, controls[action], result["run_id"]) def _restoration(self, name: str, previous_run_id: str) -> None: result, _ = self._observe("restoration", name) diff --git a/tests/test_finisher_collector.py b/tests/test_finisher_collector.py index 9feb5f7..055f3bc 100644 --- a/tests/test_finisher_collector.py +++ b/tests/test_finisher_collector.py @@ -111,7 +111,18 @@ def call(self, request): "start": {"state": "singleton_observed", "already_running": True, "readback_verified": True, "target": self.snapshot(name)["target"]}, "controls": {action: {"request_id": name + "-" + action, "accepted": True, - "applied": True, "state": state} + "applied": True, "state": state, "action": action, + "native_request_id": "native-" + name + "-" + action, + "submitted_at": self.time.isoformat(), + "target": self.snapshot(name)["target"], + "readback": {"observed_at": self.time.isoformat(), + "target": self.snapshot(name)["target"], + "source_sha": self.config["bindings"]["transport_source_commit"], + "readback_verified": True, "terminal_halt_verified": action == "halt", + "checks": {"installation_verified": True, + "service_active": action != "halt", "heartbeat_fresh": action != "halt"}, + "controller": {"paused": action == "pause", "halted": action == "halt", + "last_control_id": "native-" + name + "-" + action}}} for action, state in {"pause": "paused", "resume": "running", "halt": "halted"}.items()}, "cleanup_verified": True, } @@ -401,3 +412,40 @@ def test_cli_refuses_missing_authority_or_bad_policy_without_output(tmp_path, ca assert process.returncode == 2 assert json.loads(process.stdout) == {"status": "HELD", "release_verdict": "NOT_READY", "reason": reason} assert not output.exists() + + +@pytest.mark.parametrize("mutation", [ + lambda c: c["pause"]["target"].update(run_id="another-run"), + lambda c: c["pause"]["readback"]["target"].update(durable_object_id="f" * 64), + lambda c: c["pause"]["readback"].update(source_sha="f" * 40), + lambda c: c["resume"]["readback"]["controller"].update(last_control_id="different-request"), + lambda c: c["resume"]["readback"]["controller"].update(paused=True), + lambda c: c["halt"]["readback"]["controller"].update(halted=False), + lambda c: c["halt"]["readback"].update(terminal_halt_verified=False), + lambda c: c["pause"]["readback"]["checks"].update(heartbeat_fresh=False), + lambda c: c["pause"]["readback"].update(observed_at=(NOW - timedelta(seconds=31)).isoformat()), + lambda c: c["pause"].update(submitted_at=(NOW + timedelta(minutes=5)).isoformat()), + lambda c: c["resume"].update(action="pause"), +]) +def test_action_readback_cannot_cross_target_run_time_or_actual_state(mutation): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "front": + mutation(data["result"]["controls"]) + return data + owner.modify = change + with pytest.raises(CollectorError): + collect(fixture=owner) + + +def test_native_control_id_cannot_be_replayed_with_new_front_request_id(): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "front": + control = data["result"]["controls"]["pause"] + control["native_request_id"] = "reused-native-id" + control["readback"]["controller"]["last_control_id"] = "reused-native-id" + return data + owner.modify = change + with pytest.raises(CollectorError, match="control_replay"): + collect(fixture=owner) From cabda419d383b029ac4413480f3e7fae240dfbdd Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Sun, 8 Nov 2026 13:18:53 -0600 Subject: [PATCH 3/4] fix: require terminal cleanup before accepting Finisher halt --- docs/FINISHER_COLLECTOR.md | 8 +++++--- .../finisher_collector.py | 5 +++-- tests/test_finisher_collector.py | 18 ++++++++++++++++++ 3 files changed, 26 insertions(+), 5 deletions(-) diff --git a/docs/FINISHER_COLLECTOR.md b/docs/FINISHER_COLLECTOR.md index 99744c3..45d25a7 100644 --- a/docs/FINISHER_COLLECTOR.md +++ b/docs/FINISHER_COLLECTOR.md @@ -168,9 +168,11 @@ The ordered fronts are `auto`, `cauto`, `gauto`, `grauto`, `CFauto`, `CFcauto`, have fresh `observed_at >= submitted_at`, verified installation/readback, and `controller.last_control_id` equal to the exact native request ID. The actual paused/halted flags must agree with the action. Pause/resume - require active service and fresh heartbeat. Halt permits a fresh independently - established `terminal_halt_verified` observation after process exit without - pretending that the exited controller still has a live heartbeat. A stale + require active service and fresh heartbeat. Halt requires a fresh independently + established `terminal_halt_verified: true`, `lifecycle: halted` and + `online: false` observation after process exit without pretending that the + exited controller still has a live heartbeat. A live controller setting its + halted flag before cleanup is insufficient. A stale observation, replayed native ID, other run/installation, accepted write, pending journal or unavailable terminal readback must fail. 2. Immediately after **each** halt, `restoration/` requires diff --git a/src/echo_certification_forge/finisher_collector.py b/src/echo_certification_forge/finisher_collector.py index e362cb2..5f8bfb4 100644 --- a/src/echo_certification_forge/finisher_collector.py +++ b/src/echo_certification_forge/finisher_collector.py @@ -357,10 +357,11 @@ def _control(self, action: str, control: Any, run_id: str) -> None: "halt": controller["halted"]} require(expected[action], "control_readback_state") live = checks.get("service_active") is True and checks.get("heartbeat_fresh") is True - terminal = action == "halt" and observed.get("terminal_halt_verified") is True + terminal = (observed.get("terminal_halt_verified") is True + and observed.get("lifecycle") == "halted" and observed.get("online") is False) # A legitimate fresh terminal observation proves process exit through # the owner observer. It must not fabricate a live controller heartbeat. - require(live or terminal, "control_readback_unavailable") + require(terminal if action == "halt" else live, "control_readback_unavailable") def _front(self, name: str, result: dict[str, Any]) -> None: provider = FRONTS[name] diff --git a/tests/test_finisher_collector.py b/tests/test_finisher_collector.py index 055f3bc..fa0320e 100644 --- a/tests/test_finisher_collector.py +++ b/tests/test_finisher_collector.py @@ -119,6 +119,8 @@ def call(self, request): "target": self.snapshot(name)["target"], "source_sha": self.config["bindings"]["transport_source_commit"], "readback_verified": True, "terminal_halt_verified": action == "halt", + "lifecycle": "halted" if action == "halt" else "running", + "online": action != "halt", "checks": {"installation_verified": True, "service_active": action != "halt", "heartbeat_fresh": action != "halt"}, "controller": {"paused": action == "pause", "halted": action == "halt", @@ -449,3 +451,19 @@ def change(data): owner.modify = change with pytest.raises(CollectorError, match="control_replay"): collect(fixture=owner) + + +@pytest.mark.parametrize("mutation", [ + lambda d: d.update(terminal_halt_verified=False, checks={"installation_verified": True, "service_active": True, "heartbeat_fresh": True}), + lambda d: d.update(lifecycle="running"), + lambda d: d.update(online=True), +]) +def test_terminal_halt_requires_actual_stopped_cleanup_proof(mutation): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "front": + mutation(data["result"]["controls"]["halt"]["readback"]) + return data + owner.modify = change + with pytest.raises(CollectorError): + collect(fixture=owner) From 2e0d592b031c5482bf69f150aec74012dc953bf8 Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Sun, 8 Nov 2026 13:34:07 -0600 Subject: [PATCH 4/4] fix: bind owner route receipts and exact rollback identities --- docs/FINISHER_COLLECTOR.md | 39 ++++-- .../finisher_collector.py | 28 ++++- tests/test_finisher_collector.py | 115 ++++++++++++++++-- 3 files changed, 159 insertions(+), 23 deletions(-) diff --git a/docs/FINISHER_COLLECTOR.md b/docs/FINISHER_COLLECTOR.md index 45d25a7..5faf747 100644 --- a/docs/FINISHER_COLLECTOR.md +++ b/docs/FINISHER_COLLECTOR.md @@ -87,12 +87,21 @@ Policy fields: | `bindings.account_id`, `application_id`, `durable_object_id` | Complete actual Cloudflare tuple; 32 hex account and 64 hex DO IDs | | `bindings.installation_id`, `owner_id`, `cloudflare_origin` | Exact installation, admitted owner and HTTPS origin without userinfo/query | | `rollback_artifact_sha256`, `rollback_source_commit` | Actual distinct compatible prior artifact/source | +| `rollback_checkpoint_sha256`, `rollback_configuration_sha256` | Owner-approved immutable prior checkpoint/configuration digest pins | +| `recovered_checkpoint_sha256`, `recovered_configuration_sha256` | Owner-approved recovered candidate checkpoint/configuration digest pins | | `providers` | Exactly claude/codex/gemini/grok, each with exact `model` and nonpersonal `account_sha256` | | `probe.command` | Exactly `[ABSOLUTE_INTERPRETER, ABSOLUTE_REVIEWED_PROBE_SCRIPT]` | | `probe.sha256` | Exact SHA256 pins for those two files in the same order | | `probe.dependencies` | Absolute dependency paths mapped to immutable SHA256 pins | | `probe.timeout` | Whole process deadline from 1 to 120 seconds, default 120 | +The four checkpoint/configuration pins must come from the owner's reviewed, +immutable prior-baseline and recovered-candidate manifests **before** collection, +not from a candidate's self-reported readback. Mutable runtime checkpoints must +be deliberately frozen or version-addressed for meaningful exact comparison. +The owner admission verifies these immutable policy values as part of the +canonical policy hash; arbitrary well-formed digests are not rollback evidence. + The transport runs the interpreter in isolated mode, accepts no shell text or provider flags, bounds stdin to 8192 bytes and stdout to 128 KiB, discards stderr, and checks executable pins before and after each operation. Nonblocking stdin @@ -131,8 +140,10 @@ For `admission`, result must contain `allowed: true`, `independent: true`, exact policy hash, all request identities/bindings, an expiry in the next hour, `operations: ["observe", "recall"]`, and `restoration_allowed: true`. The actual admission must cover the complete finite journey including its controls, -negative checks, prior snapshot/config rollback, and owner restoration. The -collector rechecks admission before every observe and recall, and at completion. +negative checks, prior snapshot/config rollback, owner restoration, and all +24 separately recalled `stability-front/:` route observations. +The collector rechecks admission before every observe and recall, and at +completion. An aggregate health result is not authority for a named route. For `observe`, result contains a new actual `receipt_id`. The collector then requests `recall`; recall result contains the identical ID, exact `content`, and @@ -188,13 +199,23 @@ The ordered fronts are `auto`, `cauto`, `gauto`, `grauto`, `CFauto`, `CFcauto`, Credentials and personal identity never enter the receipt. A redirect check must refuse rather than follow an unadmitted destination. 4. `rollback/restore-and-recover` requires actual prior and recovered artifact - and source readbacks, both checkpoint source matches, actual checkpoint and - configuration SHA256 values, and verified restored/recovered state. It runs - **before** final stability. The prior source and candidate source are pinned; - container replacement alone does not meet this check. -5. `stability/0`, `/1`, `/2` require healthy, resource-admitted, fresh heartbeat, - zero unexpected restarts and healthy exact-target snapshots for **all eight - fronts**. The first-to-third observation interval must be at least 30 seconds. + and source readbacks, both checkpoint source matches, and verified restored + and recovered state. All four actual checkpoint/configuration SHA256 readbacks + must **equal their separately owner-approved immutable policy pins**. A digest + of the right length, a truthful-looking boolean, or only a Worker image + replacement is insufficient. It runs **before** final stability. +5. `stability/0`, `/1`, `/2` establish aggregate health, resource admission, + fresh heartbeat and zero unexpected restarts only. Legacy aggregate + `snapshots` are rejected; they cannot count as eight-front evidence. + Each aggregate probe is followed by **eight separate** owner-issued, + independently recalled `stability-front/:` receipts, each + bound to its exact challenge/kind/name, front and expected provider. Each + route result requires `route_readback_verified: true`, a fresh healthy + `snapshot`, and `snapshot_sha256` equal to the canonical digest of that + snapshot. Reusing another route's receipt/result fails named-route binding. + All routes may legitimately share the single Cloudflare singleton `run_id`; + route receipt provenance, not distinct run IDs, proves eight-front coverage. + The first-to-third aggregate observation interval must be at least 30 seconds. 6. `external/acceptance` requires a distinct independent reviewer, actual Access session, immutable UI assets, exact HTTPS origin and accepted live behavior. diff --git a/src/echo_certification_forge/finisher_collector.py b/src/echo_certification_forge/finisher_collector.py index 5f8bfb4..90cbf90 100644 --- a/src/echo_certification_forge/finisher_collector.py +++ b/src/echo_certification_forge/finisher_collector.py @@ -222,6 +222,11 @@ def __init__(self, policy: dict[str, Any], probe: OwnerProbe, *, require(isinstance(p.get("rollback_source_commit"), str) and COMMIT.fullmatch(p["rollback_source_commit"]) and p["rollback_source_commit"] != bindings["source_commit"], "policy_rollback") + # The approved immutable baseline and recovered candidate must pin their + # actual checkpoint/config identities before any rollback evidence is read. + require(all(digest(p.get(key)) for key in + ("rollback_checkpoint_sha256", "rollback_configuration_sha256", + "recovered_checkpoint_sha256", "recovered_configuration_sha256")), "policy_rollback_identity") require(isinstance(p.get("providers"), dict) and set(p["providers"]) == set(FRONTS.values()), "policy_providers") for provider in p["providers"].values(): require(isinstance(provider, dict) and set(provider) == {"model", "account_sha256"} @@ -407,8 +412,12 @@ def _rollback(self) -> None: and rollback.get("restored_checkpoint_source_commit") == self.policy["rollback_source_commit"] and rollback.get("recovered_source_commit") == self.policy["bindings"]["source_commit"] and rollback.get("recovered_checkpoint_source_commit") == self.policy["bindings"]["source_commit"] - and all(digest(rollback.get(key)) for key in ("restored_checkpoint_sha256", "recovered_checkpoint_sha256", - "restored_configuration_sha256", "recovered_configuration_sha256")), "rollback_failed") + and all(rollback.get(observed) == self.policy[expected] for observed, expected in { + "restored_checkpoint_sha256": "rollback_checkpoint_sha256", + "restored_configuration_sha256": "rollback_configuration_sha256", + "recovered_checkpoint_sha256": "recovered_checkpoint_sha256", + "recovered_configuration_sha256": "recovered_configuration_sha256", + }.items()), "rollback_failed") def collect(self) -> dict[str, Any]: require(not self.evidence, "collector_already_used") @@ -433,9 +442,18 @@ def collect(self) -> dict[str, Any]: result, observed = self._observe("stability", str(index)) require(all(result.get(key) is True for key in ("healthy", "resource_admitted", "heartbeat_verified")) and type(result.get("unexpected_restarts")) is int and result["unexpected_restarts"] == 0, "stability_failed") - snapshots = result.get("snapshots") - require(isinstance(snapshots, dict) and set(snapshots) == set(FRONTS), "stability_fronts_missing") - for snapshot in snapshots.values(): + # Aggregate health is not eight-front evidence. Each route requires + # its own fresh, owner-issued and independently recalled observation. + require("snapshots" not in result, "stability_snapshots_unadmitted") + for front, provider in FRONTS.items(): + route, _ = self._observe("stability-front", f"{index}:{front}") + snapshot = route.get("snapshot") + require(route.get("front") == front and route.get("provider") == provider + and route.get("route_readback_verified") is True + and digest(route.get("snapshot_sha256")) + and route["snapshot_sha256"] == sha256_json(snapshot), "stability_route_binding") + # The Cloudflare singleton run_id may legitimately be shared + # across fronts; identity is the independently recalled route. self._snapshot(snapshot) probes.append(observed) require(probes[0] < probes[1] < probes[2] and (probes[2] - probes[0]).total_seconds() >= 30, "stability_interval") diff --git a/tests/test_finisher_collector.py b/tests/test_finisher_collector.py index fa0320e..cd2bb37 100644 --- a/tests/test_finisher_collector.py +++ b/tests/test_finisher_collector.py @@ -37,6 +37,10 @@ def policy(): "cloudflare_origin": "https://finisher.example.test"}, "rollback_artifact_sha256": "8" * 64, "rollback_source_commit": "9" * 40, + "rollback_checkpoint_sha256": "1" * 64, + "rollback_configuration_sha256": "3" * 64, + "recovered_checkpoint_sha256": "2" * 64, + "recovered_configuration_sha256": "4" * 64, "providers": {v: {"model": v + "-fixture-model", "account_sha256": "a" * 64} for v in set(FRONTS.values())}, } @@ -52,6 +56,10 @@ def __init__(self, config): self.modify = lambda value: value self.admitted = True self.nonce_override = None + self.shared_run_id = False + + def run_id(self, name): + return "run-shared-singleton" if self.shared_run_id else "run-" + name def clock(self): return self.time @@ -65,7 +73,7 @@ def snapshot(self, name): "heartbeat_fresh", "resource_admitted")}, "target": {"provider": "cloudflare", **{key: binding[key] for key in ("account_id", "application_id", "durable_object_id", "installation_id")}, - "source_sha": binding["transport_source_commit"], "run_id": "run-" + name}} + "source_sha": binding["transport_source_commit"], "run_id": self.run_id(name)}} def call(self, request): self.calls.append(deepcopy(request)) @@ -101,7 +109,7 @@ def call(self, request): "provider": provider, **self.config["providers"][provider], "provider_invocation_id": "invoke-" + name, "model_observed": True, "account_verified": True, "subscription_auth": True, - "run_id": "run-" + name, "worker_id": "worker-" + name, + "run_id": self.run_id(name), "worker_id": "worker-" + name, "session_id": "session-" + name, "resumed_session_id": "session-" + name, "heartbeat_verified": True, "scoped_mcp_verified": True, "hook_receipt_id": "hook-" + name, "capture_receipt_id": "capture-" + name, @@ -131,14 +139,20 @@ def call(self, request): elif kind == "restoration": data["result"] = {"method": "admitted_owner_operation", "operation_id": "restore-" + name, "admission_id": self.config["admission_id"], - "previous_run_id": "run-" + name, "snapshot": self.snapshot(name)} + "previous_run_id": self.run_id(name), "snapshot": self.snapshot(name)} elif kind == "negative": data["result"] = {"denied": True, "side_effects": False, "before_sha256": "d" * 64, "after_sha256": "d" * 64} elif kind == "stability": data["result"] = {"healthy": True, "resource_admitted": True, - "heartbeat_verified": True, "unexpected_restarts": 0, - "snapshots": {front: self.snapshot(front) for front in FRONTS}} + "heartbeat_verified": True, "unexpected_restarts": 0} + elif kind == "stability-front": + probe_index, front = name.split(":", 1) + assert probe_index in ("0", "1", "2") + snapshot = self.snapshot(front) + data["result"] = {"front": front, "provider": FRONTS[front], + "route_readback_verified": True, "snapshot": snapshot, + "snapshot_sha256": sha256_json(snapshot)} elif kind == "external": data["result"] = {"accepted": True, "reviewer_id": "other-reviewer", "ui_assets_verified": True, "access_session_verified": True, @@ -183,11 +197,24 @@ def test_all_eight_fronts_are_observed_then_independently_recalled(): assert {x["name"] for x in owner.calls if x["operation"] == "observe" and x["kind"] == "front"} == set(FRONTS) observed = [x for x in owner.calls if x["operation"] == "observe"] recalled = [x for x in owner.calls if x["operation"] == "recall"] - assert len(observed) == len(recalled) == 16 + len(NEGATIVE_CONTROLS) + 3 + 2 + assert len(observed) == len(recalled) == 16 + len(NEGATIVE_CONTROLS) + 3 + 3 * len(FRONTS) + 2 journey = [(x["kind"], x["name"]) for x in observed] for index, name in enumerate(FRONTS): assert journey[index * 2:index * 2 + 2] == [("front", name), ("restoration", name)] assert journey.index(("rollback", "restore-and-recover")) < journey.index(("stability", "0")) + assert [(kind, name) for kind, name in journey if kind == "stability-front"] == [ + ("stability-front", f"{index}:{front}") for index in range(3) for front in FRONTS + ] + + +def test_all_eight_routes_may_share_one_authorized_cloudflare_singleton_run(): + config = policy() + owner = OwnerFixture(config) + owner.shared_run_id = True + result = collect(config, owner) + assert result["status"] == "UNSIGNED_CANDIDATE" + assert result["release_verdict"] == "NOT_READY" + assert len([x for x in owner.calls if x["operation"] == "recall"]) == 16 + len(NEGATIVE_CONTROLS) + 3 + 3 * len(FRONTS) + 2 def test_missing_admission_stops_before_any_observation(): @@ -272,6 +299,32 @@ def change(data): collect(fixture=owner) +@pytest.mark.parametrize("field", [ + "rollback_checkpoint_sha256", "rollback_configuration_sha256", + "recovered_checkpoint_sha256", "recovered_configuration_sha256", +]) +def test_missing_owner_approved_rollback_digest_pin_fails_before_observation(field): + config = policy() + del config[field] + with pytest.raises(CollectorError, match="policy_rollback_identity"): + collect(config) + + +@pytest.mark.parametrize("field", [ + "restored_checkpoint_sha256", "restored_configuration_sha256", + "recovered_checkpoint_sha256", "recovered_configuration_sha256", +]) +def test_well_formed_but_wrong_rollback_digest_fails_closed(field): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "rollback": + data["result"][field] = "f" * 64 + return data + owner.modify = change + with pytest.raises(CollectorError, match="rollback_failed"): + collect(fixture=owner) + + def test_sensitive_receipt_content_is_rejected_without_echoing_value(): owner = OwnerFixture(policy()) owner.modify = lambda data: {**data, "authorization": "test-sensitive-value"} @@ -344,14 +397,58 @@ def test_probe_output_overflow_is_bounded(tmp_path): probe.call({"request": "fixture"}) -def test_healthy_boolean_cannot_replace_all_fronts_stability_readback(): +def test_healthy_boolean_cannot_replace_per_route_stability_readback(): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "stability-front" and data["name"] == "2:CFgrauto": + snapshot = data["result"]["snapshot"] + snapshot["online"] = False + # A correct route digest cannot turn an unhealthy snapshot into health. + data["result"]["snapshot_sha256"] = sha256_json(snapshot) + return data + owner.modify = change + with pytest.raises(CollectorError, match="status_snapshot_unhealthy"): + collect(fixture=owner) + + +def test_cloned_owner_route_result_is_rejected_for_another_front(): + owner = OwnerFixture(policy()) + first = {} + def change(data): + if data["kind"] == "stability-front" and data["name"] == "0:auto": + first["result"] = deepcopy(data["result"]) + if data["kind"] == "stability-front" and data["name"] == "0:cauto": + # The singleton target may match; route provenance still must not. + data["result"] = deepcopy(first["result"]) + return data + owner.modify = change + with pytest.raises(CollectorError, match="stability_route_binding"): + collect(fixture=owner) + + +@pytest.mark.parametrize("field,value", [ + ("front", "auto"), ("provider", "claude"), ("route_readback_verified", False), + ("snapshot_sha256", "f" * 64), +]) +def test_stability_route_provenance_tampering_fails_closed(field, value): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "stability-front" and data["name"] == "0:cauto": + data["result"][field] = value + return data + owner.modify = change + with pytest.raises(CollectorError, match="stability_route_binding"): + collect(fixture=owner) + + +def test_legacy_aggregate_snapshots_cannot_count_as_route_receipts(): owner = OwnerFixture(policy()) def change(data): if data["kind"] == "stability": - data["result"]["snapshots"]["CFgrauto"]["online"] = False + data["result"]["snapshots"] = {front: owner.snapshot(front) for front in FRONTS} return data owner.modify = change - with pytest.raises(CollectorError): + with pytest.raises(CollectorError, match="stability_snapshots_unadmitted"): collect(fixture=owner)