From b30cf970bd801d30038135d637bb372474d8e22c Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Tue, 22 Sep 2026 02:03:04 -0500 Subject: [PATCH 01/10] fix: publish P4 completion after verified cleanup and enrichment Successful hostile acceptance reports omitted the completed_phase_gate field required by the master collector. Publish P4 only after every core check and cleanup succeeds, and keep exact-wrapper reports pending until enrichment returns. Failed verification, malformed capture, interruption and nonzero exit cannot leave a newly completed report for the master consumer. Keep master validation, exact-source and signing rules, release verdicts and historical acceptance reports unchanged. Add producer/wrapper/consumer integration coverage including pending publication and failure paths. Validation: 24 focused and adjacent tests pass; full suite 637 pass, 2 skip, 1 existing symlink test blocked by Windows privilege 1314. P1 acceptance passes all 3 scenarios using a byte-identical source copy outside historical artifacts. Hosted CI, dual certification and deployment remain pending. --- scripts/p4_hostile_acceptance.py | 21 +++- scripts/p4_hostile_acceptance_exact.py | 18 +++ tests/test_p4_master_report_contract.py | 141 ++++++++++++++++++++++++ 3 files changed, 179 insertions(+), 1 deletion(-) create mode 100644 tests/test_p4_master_report_contract.py diff --git a/scripts/p4_hostile_acceptance.py b/scripts/p4_hostile_acceptance.py index b17dd0e..3ab0f36 100644 --- a/scripts/p4_hostile_acceptance.py +++ b/scripts/p4_hostile_acceptance.py @@ -136,6 +136,25 @@ def write_json(path: Path, value: Any) -> None: path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") +def write_acceptance_report(path: Path, report: dict[str, Any]) -> None: + """Publish phase completion only after checks and final cleanup succeed.""" + report.pop("completed_phase_gate", None) + checks = report.get("checks") + cleanup = report.get("cleanup") or {} + if ( + report.get("phase") == "P4" + and report.get("passed") is True + and report.get("run_outcome") == "COMPLETE" + and isinstance(checks, dict) + and checks + and all(value is True for value in checks.values()) + and cleanup.get("unrelated_container_ids_preserved") is True + and cleanup.get("ephemeral_private_files_removed") is True + ): + report["completed_phase_gate"] = "P4" + write_json(path, report) + + def docker_json(*arguments: str) -> Any: return json.loads(run(["docker", *arguments]).stdout) @@ -1859,7 +1878,7 @@ def main() -> int: report["error"] = {"type": "CleanupError", "message": "unrelated container identity changed"} return_code = 1 report["completed_at_utc"] = to_utc_iso(datetime.now(UTC)) - write_json(output, report) + write_acceptance_report(output, report) print( json.dumps( { diff --git a/scripts/p4_hostile_acceptance_exact.py b/scripts/p4_hostile_acceptance_exact.py index 0d6d61c..732ad99 100644 --- a/scripts/p4_hostile_acceptance_exact.py +++ b/scripts/p4_hostile_acceptance_exact.py @@ -147,6 +147,10 @@ def enrich_report( if not output_path.is_file(): raise RuntimeError(f"P4 harness did not produce an output report: {output_path}") report = json.loads(output_path.read_text(encoding="utf-8")) + # Enrichment is a further acceptance boundary; invalidate any earlier marker + # before parsing captured evidence, which can raise or be interrupted. + report.pop("completed_phase_gate", None) + output_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8") report["harness"] = { "path": str(HERE), "sha256": sha256_file(HERE), @@ -205,6 +209,10 @@ def enrich_report( ) return_code = 1 report["wrapper_return_code"] = return_code + if return_code != 0: + report["passed"] = False + report["run_outcome"] = "INFRA_FAILED" + report["release_verdict"] = "NOT_READY" output_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8") return report @@ -213,6 +221,15 @@ def main(arguments: list[str] | None = None) -> int: argv = list(sys.argv[1:] if arguments is None else arguments) sealed_manifest_path, output_path = parse_wrapper_paths(argv) harness = load_harness() + publish_report = harness.write_acceptance_report + + def write_pending_report(path: Path, report: dict[str, Any]) -> None: + report.pop("completed_phase_gate", None) + harness.write_json(path, report) + + # The core can finish before the exact verifier evidence is enriched. + # Keep its on-disk result pending until every wrapper layer has returned. + harness.write_acceptance_report = write_pending_report artifacts = resolve_runner_artifacts(harness, sealed_manifest_path) capture: dict[str, Any] = {} harness.execute_container = make_execute_container_override( @@ -234,6 +251,7 @@ def main(arguments: list[str] | None = None) -> int: return_code=return_code, ) final_code = 0 if report.get("passed") is True and report.get("run_outcome") == "COMPLETE" else 1 + publish_report(output_path, report) print( json.dumps( { diff --git a/tests/test_p4_master_report_contract.py b/tests/test_p4_master_report_contract.py new file mode 100644 index 0000000..4026343 --- /dev/null +++ b/tests/test_p4_master_report_contract.py @@ -0,0 +1,141 @@ +from __future__ import annotations + +import hashlib +import importlib.util +import json +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] + + +def load_script(name: str): + spec = importlib.util.spec_from_file_location(name, ROOT / "scripts" / f"{name}.py") + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def completed_report() -> dict: + # Contract fixture only: no real P4 execution or certification is claimed. + return { + "phase": "P4", + "passed": True, + "run_outcome": "COMPLETE", + "release_verdict": "NOT_READY", + "source_commit": "a" * 40, + "checks": {"sealed_twelve_image_manifest_verified": True, "public_verifier_passed": True}, + "cleanup": { + "unrelated_container_ids_preserved": True, + "ephemeral_private_files_removed": True, + }, + } + + +def test_successful_producer_report_is_consumable_without_changing_master_gate(tmp_path: Path) -> None: + producer = load_script("p4_hostile_acceptance") + consumer = load_script("master_acceptance") + report = completed_report() + output = tmp_path / "p4.json" + + producer.write_acceptance_report(output, report) + + gate = consumer._gate("P4", output, consumer._p4) + assert gate == {"state": "COMPLETE", "evidence_sha256": hashlib.sha256(output.read_bytes()).hexdigest()} + emitted = json.loads(output.read_text()) + assert emitted["source_commit"] == "a" * 40 + assert emitted["release_verdict"] == "NOT_READY" + + +@pytest.mark.parametrize("failure", ["checks", "empty_checks", "outcome", "passed", "containers", "private_files"]) +def test_incomplete_producer_report_cannot_satisfy_master_gate(tmp_path: Path, failure: str) -> None: + producer = load_script("p4_hostile_acceptance") + consumer = load_script("master_acceptance") + report = completed_report() + # A stale marker must be removed if a later verification or cleanup fails. + report["completed_phase_gate"] = "P4" + if failure == "checks": + report["checks"]["public_verifier_passed"] = False + elif failure == "empty_checks": + report["checks"] = {} + elif failure == "outcome": + report["run_outcome"] = "INFRA_FAILED" + elif failure == "passed": + report["passed"] = False + elif failure == "containers": + report["cleanup"]["unrelated_container_ids_preserved"] = False + else: + report["cleanup"]["ephemeral_private_files_removed"] = False + output = tmp_path / "p4.json" + + producer.write_acceptance_report(output, report) + + with pytest.raises(ValueError, match="P4 gate is incomplete"): + consumer._gate("P4", output, consumer._p4) + assert "completed_phase_gate" not in json.loads(output.read_text()) + + +def test_wrapper_failure_removes_previously_completed_marker(tmp_path: Path) -> None: + producer = load_script("p4_hostile_acceptance") + wrapper = load_script("p4_hostile_acceptance_exact") + consumer = load_script("master_acceptance") + output = tmp_path / "p4.json" + producer.write_acceptance_report(output, completed_report()) + + result = wrapper.enrich_report(output_path=output, harness=producer, capture={}, return_code=1) + + assert result["passed"] is False + assert "completed_phase_gate" not in result + with pytest.raises(ValueError, match="P4 gate is incomplete"): + consumer._gate("P4", output, consumer._p4) + + +@pytest.mark.parametrize("scenario", ["success", "nonzero", "malformed", "interrupted"]) +def test_exact_entrypoint_publishes_only_after_complete_enrichment(tmp_path: Path, monkeypatch, scenario: str) -> None: + producer = load_script("p4_hostile_acceptance") + wrapper = load_script("p4_hostile_acceptance_exact") + consumer = load_script("master_acceptance") + output = tmp_path / "p4.json" + capture = { + "result": { + "exit_code": 0, "timed_out": False, "oom_killed": False, + "containment": {"network_none": True}, + "logs": json.dumps({"passed": True, "private_key_loaded": False, + "sbom_valid": True, "admission": {"allowed": True}}), + }, + "artifacts": {}, "execution_limits": {}, + } + if scenario == "malformed": + capture["result"] = None + + def core_main(): + producer.write_acceptance_report(output, completed_report()) + # Even a successful core result is pending while the wrapper is running. + with pytest.raises(ValueError, match="P4 gate is incomplete"): + consumer._gate("P4", output, consumer._p4) + if scenario == "interrupted": + raise RuntimeError("simulated interruption after core serialization") + return 1 if scenario == "nonzero" else 0 + + def capture_override(original, *, artifacts, capture: dict): + capture.update(captured) + return original + + captured = capture + monkeypatch.setattr(producer, "main", core_main) + monkeypatch.setattr(wrapper, "load_harness", lambda: producer) + monkeypatch.setattr(wrapper, "parse_wrapper_paths", lambda argv: (tmp_path / "manifest.json", output)) + monkeypatch.setattr(wrapper, "resolve_runner_artifacts", lambda *args: {}) + monkeypatch.setattr(wrapper, "make_execute_container_override", capture_override) + if scenario in {"malformed", "interrupted"}: + with pytest.raises((AttributeError, RuntimeError)): + wrapper.main([]) + else: + assert wrapper.main([]) == (0 if scenario == "success" else 1) + if scenario == "success": + assert consumer._gate("P4", output, consumer._p4)["state"] == "COMPLETE" + else: + with pytest.raises(ValueError, match="P4 gate is incomplete"): + consumer._gate("P4", output, consumer._p4) From bc1b8ad9535ad6c736b3d03037298f6307e727fc Mon Sep 17 00:00:00 2001 From: echo-ops-certforge Date: Thu, 24 Sep 2026 15:16:59 -0500 Subject: [PATCH 02/10] fix(certforge): patched base image and cryptography 50 for P4 sealing (#22) Release hardening so the P4 hostile acceptance gate can seal again: - Bump the pinned python:3.12-alpine base for all six role images, the sandbox default image and the P4 gate to sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111 (Trivy: 0 HIGH/CRITICAL findings on the new base). - Bump the hash-locked cryptography pin to 50.0.0 (all PyPI wheel and sdist hashes locked). - deploy_forge.sh: the dispatcher unit uses Wants= instead of Requires= on the API service so an API restart no longer tears down in-flight runs; the source-fetch git credential helper is configurable (CERTFORGE_GIT_CREDENTIAL_HELPER); the trusted manifest default is the canonical-JSON digest that run_worker pins (08ba068c...), not the raw file-bytes digest. - Opt the repository into its own CertForge gate with a stdlib-only journey (scripts/certforge_journey.py) and pin tests (tests/test_release_pins.py). The worker environment identity digest is label-derived and does not change with this release (8e6466d2...). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0152mN1wMwj9vE2YmV2xZF4F --- .echo/certification.json | 4 + deploy/deploy_forge.sh | 14 +- images/anchor/Dockerfile | 4 +- images/custody/Dockerfile | 4 +- images/requirements.in | 2 +- images/requirements.lock | 94 ++-- images/runner/Dockerfile | 4 +- images/signer/Dockerfile | 4 +- images/verifier/Dockerfile | 4 +- images/worker/Dockerfile | 4 +- scripts/certforge_journey.py | 76 ++++ scripts/certforge_testkit.py | 562 ++++++++++++++++++++++++ scripts/run_p4_gate.sh | 2 +- src/echo_certification_forge/sandbox.py | 2 +- tests/test_release_pins.py | 69 +++ 15 files changed, 784 insertions(+), 65 deletions(-) create mode 100644 .echo/certification.json create mode 100644 scripts/certforge_journey.py create mode 100644 scripts/certforge_testkit.py create mode 100644 tests/test_release_pins.py diff --git a/.echo/certification.json b/.echo/certification.json new file mode 100644 index 0000000..b06c33e --- /dev/null +++ b/.echo/certification.json @@ -0,0 +1,4 @@ +{ + "version": 1, + "journey": ["python3", "-B", "scripts/certforge_journey.py"] +} diff --git a/deploy/deploy_forge.sh b/deploy/deploy_forge.sh index fd134ac..7301e49 100644 --- a/deploy/deploy_forge.sh +++ b/deploy/deploy_forge.sh @@ -22,13 +22,19 @@ PRODUCTION_E2E_ATTESTATION_DIR="${ECHO_CERTFORGE_PRODUCTION_E2E_ATTESTATION_DIR: PRODUCTION_E2E_TRUSTED_KEYS="${ECHO_CERTFORGE_TRUSTED_PRODUCTION_E2E_KEYS:-$STATE_ROOT/production-e2e/trusted-public-keys}" ADAPTER_DIR="${ECHO_CERTFORGE_PROD_ADAPTER_DIR:-$STATE_ROOT/p5}" ADAPTER_MODE="${CERTFORGE_ADAPTER_MODE:-required}" -TRUSTED_MANIFEST_SHA256="${ECHO_CERTFORGE_TRUSTED_MANIFEST_SHA256:-965106b00917268d556b325719f26f5096e6c3746551658ffecb9fd4a95ec342}" +# Canonical-JSON digest of policies/mandatory-rules.v2.json (== run_worker._PRODUCTION_MANIFEST_SHA256). +# The old default (965106b0...) was the raw file-bytes digest, which run_worker rejects. +TRUSTED_MANIFEST_SHA256="${ECHO_CERTFORGE_TRUSTED_MANIFEST_SHA256:-08ba068ceb3e14bfed2690337edbb94c546e3e0a1a89b1321f7657653d8eea43}" UNIT_PATH="/etc/systemd/system/$SERVICE.service" DISPATCH_UNIT_PATH="/etc/systemd/system/$DISPATCH_SERVICE.service" RELEASE_DROPIN="/etc/systemd/system/$SERVICE.service.d/10-release.conf" DISPATCH_RELEASE_DROPIN="/etc/systemd/system/$DISPATCH_SERVICE.service.d/10-release.conf" ENV_FILE="${CERTFORGE_ENV_FILE:-/home/forge/.config/echo/certforge.env}" -GITC=(-c credential.helper= -c credential.helper="store --file=/home/forge/.config/echo/omega_git_creds") +# Source fetch credentials. Default: the FORGE credential store. Repositories on the +# echoomegaprime account are fetched with a repo-scoped GitHub App helper instead +# (CERTFORGE_GIT_CREDENTIAL_HELPER), because that store cannot see them. +GIT_CREDENTIAL_HELPER="${CERTFORGE_GIT_CREDENTIAL_HELPER:-store --file=/home/forge/.config/echo/omega_git_creds}" +GITC=(-c credential.helper= -c credential.helper="$GIT_CREDENTIAL_HELPER") LOCK_FILE="${CERTFORGE_DEPLOY_LOCK:-/run/lock/echo-certforge-deploy.lock}" exec 9>"$LOCK_FILE" @@ -577,7 +583,9 @@ sudo tee "$DISPATCH_UNIT_PATH" >/dev/null < int: + print(f"CERTFORGE_SELF_JOURNEY_FAILED: {message}", file=sys.stderr) + return 1 + + +def canonical_digest(path: Path) -> str: + value = json.loads(path.read_text(encoding="utf-8")) + return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":"), + ensure_ascii=False).encode("utf-8")).hexdigest() + + +def main() -> int: + cert = json.loads((ROOT / ".echo/certification.json").read_text(encoding="utf-8")) + apps = json.loads((ROOT / ".echo/apps.json").read_text(encoding="utf-8")) + if cert.get("version") != 1 or apps.get("apps", {}).get("certification-forge", {}).get("enabled") is not True: + return fail("opt-in contract invalid") + + manifest_digest = canonical_digest(ROOT / "policies/mandatory-rules.v2.json") + run_worker = (ROOT / "src/echo_certification_forge/run_worker.py").read_text(encoding="utf-8") + pinned = re.search(r'_PRODUCTION_MANIFEST_SHA256 = \(\s*"([0-9a-f]{64})"', run_worker) + deploy = (ROOT / "deploy/deploy_forge.sh").read_text(encoding="utf-8") + deploy_default = re.search(r'TRUSTED_MANIFEST_SHA256="\$\{ECHO_CERTFORGE_TRUSTED_MANIFEST_SHA256:-([0-9a-f]{64})\}"', + deploy) + if not pinned or pinned.group(1) != manifest_digest: + return fail(f"run_worker manifest pin != canonical policy digest {manifest_digest}") + if not deploy_default or deploy_default.group(1) != manifest_digest: + return fail("deploy default trusted manifest digest != canonical policy digest") + + compiled = 0 + for path in sorted((ROOT / "src").rglob("*.py")): + try: + compile(path.read_text(encoding="utf-8"), str(path), "exec") + except SyntaxError as exc: + return fail(f"syntax error: {path.relative_to(ROOT)}: {exc}") + compiled += 1 + print(json.dumps({"manifest_sha256": manifest_digest, "modules_compiled": compiled})) + # The two release suites are self-contained text/contract checks; tests/conftest.py needs + # pydantic/cryptography, which the stdlib-only sandbox does not have. + return certforge_testkit.run(ROOT, SUITES, sys_paths=("tests", "."), budget_s=60.0, + min_passed=15, conftest=False) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/certforge_testkit.py b/scripts/certforge_testkit.py new file mode 100644 index 0000000..ce78108 --- /dev/null +++ b/scripts/certforge_testkit.py @@ -0,0 +1,562 @@ +"""Stdlib-only test runner for Certification Forge journeys (ops-20260924). + +The CertForge sandbox runs a journey in ``python:3.12-alpine`` with no network, no +third-party packages, a read-only checkout and a 90 s wall clock. Most ECHO suites are +written for pytest, so this module provides the small pytest surface they use: +``raises``, ``mark.parametrize`` / ``skip`` / ``skipif`` (other marks are no-ops), ``param``, +``skip()``, ``fail()``, ``importorskip()``, ``approx``, and ``@fixture`` functions (plain or +``yield``) from the suite module and its ``conftest.py`` files, plus the built-in fixtures +``tmp_path``, ``monkeypatch`` and ``capsys``. It also runs ``unittest.TestCase`` classes and +pytest-style ``Test*`` classes. + +Third-party modules that a suite imports but never exercises can be replaced with inert +stubs (``stub_modules``). Stubs are installed only when the real module is missing, and the +report names them. Every failure, error, import error, empty run or blown time budget fails +the journey (fail closed). +""" +from __future__ import annotations + +import contextlib +import importlib +import importlib.util +import inspect +import io +import itertools +import os +import re +import sys +import tempfile +import time +import traceback +import types +import unittest +from pathlib import Path +from typing import Any, Callable, Iterable + + +class Skipped(Exception): + """Raised by ``pytest.skip`` / ``importorskip``.""" + + +class Failed(AssertionError): + """Raised by ``pytest.fail``.""" + + +class BudgetExceeded(Exception): + """The journey ran out of its wall-clock budget (fails the journey).""" + + +# ── pytest compatibility surface ───────────────────────────────────────────────────────── +class _Raises: + def __init__(self, expected: Any, match: str | None = None) -> None: + self.expected = expected + self.match = match + self.value: BaseException | None = None + + def __enter__(self) -> "_Raises": + return self + + def __exit__(self, exc_type, exc, tb) -> bool: + if exc_type is None: + raise Failed(f"DID NOT RAISE {self.expected}") + if not issubclass(exc_type, self.expected): + return False + if self.match is not None and not re.search(self.match, str(exc)): + raise Failed(f"regex {self.match!r} did not match {str(exc)!r}") + self.value = exc + return True + + +class _Param: + def __init__(self, *values: Any, id: str | None = None, marks: Any = ()) -> None: # noqa: A002 + self.values = values + self.id = id + self.marks = marks + + +class _NoopMark: + def __call__(self, *args: Any, **kwargs: Any) -> Any: + if len(args) == 1 and not kwargs and callable(args[0]): + return args[0] + return lambda fn: fn + + +class _Mark: + @staticmethod + def parametrize(names: Any, values: Iterable[Any], **_: Any) -> Callable: + if isinstance(names, str): + names = [n.strip() for n in names.split(",") if n.strip()] + names = list(names) + rows = [] + for value in values: + if isinstance(value, _Param): + value = value.values if len(names) > 1 else value.values[0] + rows.append(tuple(value) if len(names) > 1 else (value,)) + + def deco(fn: Callable) -> Callable: + fn.__dict__.setdefault("_cf_params", []).insert(0, (names, rows)) + return fn + return deco + + @staticmethod + def skip(reason: str = "") -> Callable: + def deco(fn: Callable) -> Callable: + fn._cf_skip = reason or "skip" + return fn + return deco + + @staticmethod + def skipif(condition: Any, reason: str = "") -> Callable: + def deco(fn: Callable) -> Callable: + if condition: + fn._cf_skip = reason or "skipif" + return fn + return deco + + def __getattr__(self, name: str) -> _NoopMark: + return _NoopMark() + + +class _Approx: + def __init__(self, expected: Any, rel: float | None = None, abs: float | None = None) -> None: # noqa: A002 + self.expected, self.rel, self.abs = expected, rel, abs + + def _close(self, a: float, b: float) -> bool: + rel = 1e-6 if self.rel is None else self.rel + tol = max(rel * abs(b), 1e-12 if self.abs is None else self.abs) + return abs(a - b) <= tol + + def __eq__(self, other: Any) -> bool: + if isinstance(self.expected, (list, tuple)): + return len(other) == len(self.expected) and all( + self._close(a, b) for a, b in zip(other, self.expected)) + if isinstance(self.expected, dict): + return other.keys() == self.expected.keys() and all( + self._close(other[k], v) for k, v in self.expected.items()) + return self._close(other, self.expected) + + def __repr__(self) -> str: + return f"approx({self.expected!r})" + + +def _fixture(*args: Any, **kwargs: Any) -> Any: + def mark(fn: Callable) -> Callable: + fn._cf_fixture = True + return fn + if len(args) == 1 and callable(args[0]) and not kwargs: + return mark(args[0]) + return mark + + +def _skip(reason: str = "", **_: Any) -> None: + raise Skipped(reason) + + +def _fail(reason: str = "", **_: Any) -> None: + raise Failed(reason) + + +def _importorskip(name: str, *_: Any, **__: Any) -> Any: + try: + return importlib.import_module(name) + except ImportError as exc: + raise Skipped(f"{name} unavailable: {exc}") from None + + +def pytest_module() -> types.ModuleType: + mod = types.ModuleType("pytest") + mod.__dict__.update( + raises=_Raises, mark=_Mark(), param=_Param, fixture=_fixture, skip=_skip, + fail=_fail, importorskip=_importorskip, approx=_Approx, + __certforge_shim__=True, + ) + mod.skip.Exception = Skipped # type: ignore[attr-defined] + return mod + + +# ── built-in fixtures ──────────────────────────────────────────────────────────────────── +def _resolve_dotted(path: str) -> Any: + """Import the longest importable prefix of ``path``, then walk the remaining attributes.""" + parts = path.split(".") + for cut in range(len(parts), 0, -1): + try: + obj = importlib.import_module(".".join(parts[:cut])) + except ImportError: + continue + for attr in parts[cut:]: + obj = getattr(obj, attr) + return obj + raise ImportError(f"cannot resolve {path!r}") + + +class MonkeyPatch: + _MISSING = object() + + def __init__(self) -> None: + self._undo: list[Callable[[], None]] = [] + + def setattr(self, target: Any, name: Any, value: Any = _MISSING, raising: bool = True) -> None: + if value is self._MISSING: # "pkg.mod.attr" form + value = name + owner_path, _, name = str(target).rpartition(".") + target = _resolve_dotted(owner_path) + old = getattr(target, name, self._MISSING) + if old is self._MISSING and raising: + raise AttributeError(f"{target!r} has no attribute {name!r}") + setattr(target, name, value) + self._undo.append(lambda: delattr(target, name) if old is self._MISSING + else setattr(target, name, old)) + + def delattr(self, target: Any, name: str, raising: bool = True) -> None: + if not hasattr(target, name): + if raising: + raise AttributeError(name) + return + old = getattr(target, name) + delattr(target, name) + self._undo.append(lambda: setattr(target, name, old)) + + def setitem(self, mapping: Any, key: Any, value: Any) -> None: + old = mapping.get(key, self._MISSING) + mapping[key] = value + self._undo.append(lambda: mapping.pop(key, None) if old is self._MISSING + else mapping.__setitem__(key, old)) + + def delitem(self, mapping: Any, key: Any, raising: bool = True) -> None: + if key not in mapping: + if raising: + raise KeyError(key) + return + old = mapping.pop(key) + self._undo.append(lambda: mapping.__setitem__(key, old)) + + def setenv(self, name: str, value: Any, prepend: str | None = None) -> None: + value = str(value) + if prepend and name in os.environ: + value = value + prepend + os.environ[name] + self.setitem(os.environ, name, value) + + def delenv(self, name: str, raising: bool = True) -> None: + self.delitem(os.environ, name, raising) + + def syspath_prepend(self, path: Any) -> None: + sys.path.insert(0, str(path)) + self._undo.append(lambda: sys.path.remove(str(path))) + + def chdir(self, path: Any) -> None: + old = os.getcwd() + os.chdir(path) + self._undo.append(lambda: os.chdir(old)) + + def undo(self) -> None: + while self._undo: + self._undo.pop()() + + +class CapSys: + def __init__(self) -> None: + self._out, self._err = io.StringIO(), io.StringIO() + self._saved = (sys.stdout, sys.stderr) + sys.stdout, sys.stderr = self._out, self._err + + def readouterr(self) -> Any: + out, err = self._out.getvalue(), self._err.getvalue() + self._out.seek(0), self._out.truncate(), self._err.seek(0), self._err.truncate() + return types.SimpleNamespace(out=out, err=err) + + def close(self) -> None: + sys.stdout, sys.stderr = self._saved + + +# ── inert stubs for unexercised third-party imports ────────────────────────────────────── +class _StubObject: + def __init__(self, *args: Any, **kwargs: Any) -> None: + pass + + def __call__(self, *args: Any, **kwargs: Any) -> Any: + if len(args) == 1 and callable(args[0]) and not kwargs: + return args[0] # used as a decorator + return _StubObject() + + def __getattr__(self, name: str) -> Any: + return _StubObject() + + +def _stub_module(name: str) -> types.ModuleType: + mod = types.ModuleType(name) + mod.__path__ = [] # allow "import pkg.sub" + mod.__certforge_stub__ = True + + def __getattr__(attr: str) -> Any: + if attr.startswith("__"): + raise AttributeError(attr) + base = Exception if attr.endswith(("Error", "Exception")) else _StubObject + value = type(attr, (base,), {"__module__": name}) + setattr(mod, attr, value) + return value + mod.__getattr__ = __getattr__ # type: ignore[attr-defined] + return mod + + +def install_stubs(names: Iterable[str]) -> list[str]: + installed = [] + for name in names: + if name in sys.modules: + continue + try: + importlib.import_module(name) + continue + except ImportError: + pass + sys.modules[name] = _stub_module(name) + parent, _, child = name.rpartition(".") + if parent and parent in sys.modules: + setattr(sys.modules[parent], child, sys.modules[name]) + installed.append(name) + return installed + + +# ── collection and execution ───────────────────────────────────────────────────────────── +class _Session: + def __init__(self, root: Path, deadline: float) -> None: + self.root = root + self.deadline = deadline + self.passed = self.failed = self.skipped = 0 + self.failures: list[str] = [] + + # fixtures ---------------------------------------------------------------------------- + def _resolve(self, name: str, fixtures: dict, cache: dict, teardown: list) -> Any: + if name in cache: + return cache[name] + if name == "tmp_path": + value = Path(tempfile.mkdtemp(prefix="cf-")) + elif name == "monkeypatch": + value = MonkeyPatch() + teardown.append(value.undo) + elif name == "capsys": + value = CapSys() + teardown.append(value.close) + elif name == "request": + value = types.SimpleNamespace(param=None, node=None, config=None) + elif name in fixtures: + fn = fixtures[name] + kwargs = {p: self._resolve(p, fixtures, cache, teardown) + for p in inspect.signature(fn).parameters} + value = fn(**kwargs) + if inspect.isgenerator(value): + gen = value + value = next(gen) + teardown.append(lambda g=gen: next(g, None)) + else: + raise LookupError(f"fixture {name!r} not found") + cache[name] = value + return value + + def _call(self, label: str, fn: Callable, fixtures: dict, bound: dict) -> None: + if time.monotonic() > self.deadline: + raise BudgetExceeded(f"journey time budget exhausted before {label}") + skip = getattr(fn, "_cf_skip", None) + if skip: + self.skipped += 1 + return + cache: dict = dict(bound) + teardown: list = [] + try: + params = [p for p in inspect.signature(fn).parameters if p != "self"] + kwargs = {p: self._resolve(p, fixtures, cache, teardown) for p in params} + fn(**kwargs) + self.passed += 1 + except Skipped: + self.skipped += 1 + except (BudgetExceeded, KeyboardInterrupt): + raise + except BaseException as exc: # noqa: BLE001 - every error (incl. SystemExit) is a failure + self.failed += 1 + self.failures.append(f"{label}: {type(exc).__name__}: {exc}\n" + + "".join(traceback.format_exc(limit=6))[-1500:]) + finally: + for fin in reversed(teardown): + with contextlib.suppress(Exception): + fin() + + def _expand(self, label: str, fn: Callable, fixtures: dict) -> None: + grids = getattr(fn, "_cf_params", []) + if not grids: + self._call(label, fn, fixtures, {}) + return + for i, combo in enumerate(itertools.product(*[rows for _, rows in grids])): + bound: dict = {} + for (names, _), row in zip(grids, combo): + bound.update(zip(names, row)) + self._call(f"{label}[{i}]", fn, fixtures, bound) + + # suites ------------------------------------------------------------------------------ + def _conftest_fixtures(self, suite: Path) -> dict: + fixtures: dict = {} + if not getattr(self, "use_conftest", True): + return fixtures # self-contained suites; conftest needs packages the sandbox lacks + chain = [d for d in [suite.parent, *suite.parent.parents] if self.root in (d, *d.parents)] + for directory in reversed(chain): + conftest = directory / "conftest.py" + if conftest.is_file(): + mod = _load(conftest, "cf_conftest_" + _slug(conftest.relative_to(self.root))) + fixtures.update(_fixtures_of(mod)) + return fixtures + + def run_suite(self, suite: Path) -> None: + fixtures = self._conftest_fixtures(suite) + module = _load(suite, "cf_suite_" + _slug(suite.relative_to(self.root))) + fixtures.update(_fixtures_of(module)) + rel = str(suite.relative_to(self.root)) + for name, obj in list(vars(module).items()): + if isinstance(obj, type) and issubclass(obj, unittest.TestCase): + result = unittest.TestResult() + unittest.defaultTestLoader.loadTestsFromTestCase(obj).run(result) + self.passed += result.testsRun - len(result.failures) - len(result.errors) - len(result.skipped) + self.skipped += len(result.skipped) + for case, tb in result.failures + result.errors: + self.failed += 1 + self.failures.append(f"{rel}::{case.id()}\n{tb[-1500:]}") + elif isinstance(obj, type) and name.startswith("Test") and "__init__" not in vars(obj): + for meth in [m for m in dir(obj) if m.startswith("test")]: + inst = obj() + if hasattr(inst, "setup_method"): + inst.setup_method(getattr(inst, meth)) + try: + self._expand(f"{rel}::{name}::{meth}", getattr(inst, meth), fixtures) + finally: + if hasattr(inst, "teardown_method"): + inst.teardown_method(getattr(inst, meth)) + elif (name.startswith("test") and inspect.isfunction(obj) + and obj.__module__ == module.__name__): + self._expand(f"{rel}::{name}", obj, fixtures) + + +def _slug(path: Any) -> str: + return re.sub(r"[^0-9A-Za-z]+", "_", str(path)) + + +def _load(path: Path, name: str) -> types.ModuleType: + spec = importlib.util.spec_from_file_location(name, path) + if spec is None or spec.loader is None: + raise ImportError(f"cannot load {path}") + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +def _fixtures_of(module: types.ModuleType) -> dict: + return {n: f for n, f in vars(module).items() if callable(f) and getattr(f, "_cf_fixture", False)} + + +def run(root: Path, suites: Iterable[str], *, sys_paths: Iterable[str] = (), + stub_modules: Iterable[str] = (), budget_s: float = 75.0, min_passed: int = 1, + conftest: bool = True) -> int: + """Run ``suites`` (paths relative to ``root``); return a process exit code (0 = PASS).""" + import json + + start = time.monotonic() + sys.dont_write_bytecode = True # read-only checkout + for entry in reversed(list(sys_paths)): + sys.path.insert(0, str(root / entry)) + shim = importlib.util.find_spec("pytest") is None + if shim: + sys.modules["pytest"] = pytest_module() + stubs = install_stubs(stub_modules) + suites = list(suites) + missing = [s for s in suites if not (root / s).is_file()] + if missing: + print("CERTFORGE_TESTKIT_FAILED: missing suites: " + ", ".join(missing), file=sys.stderr) + return 1 + session = _Session(root, start + budget_s) + session.use_conftest = conftest + aborted = None + for suite in suites: + try: + session.run_suite(root / suite) + except BudgetExceeded as exc: + aborted = str(exc) + break + except BaseException as exc: # noqa: BLE001 - an import/collection error fails the suite + session.failed += 1 + session.failures.append(f"{suite}: collection error {type(exc).__name__}: {exc}\n" + + traceback.format_exc(limit=6)[-1500:]) + for failure in session.failures[:20]: + print("FAIL " + failure, file=sys.stderr) + report = { + "suites": len(suites), "passed": session.passed, "failed": session.failed, + "skipped": session.skipped, "pytest_shim": shim, "stubbed_modules": stubs, + "elapsed_s": round(time.monotonic() - start, 2), "aborted": aborted, + } + ok = aborted is None and session.failed == 0 and session.passed >= min_passed + print(("CERTFORGE_TESTKIT_OK " if ok else "CERTFORGE_TESTKIT_FAILED ") + + json.dumps(report, sort_keys=True)) + return 0 if ok else 1 + + +def run_isolated(root: Path, suites: Iterable[dict], *, budget_s: float = 60.0, + per_suite_timeout_s: float = 30.0, min_passed: int = 1) -> int: + """Run each suite in its own interpreter (no module-name collisions between suites). + + ``suites``: dicts with ``path`` plus optional ``sys_paths`` / ``stub_modules``. + """ + import json + import subprocess + + start = time.monotonic() + suites = list(suites) + totals = {"passed": 0, "failed": 0, "skipped": 0} + failed_suites: list[str] = [] + for spec in suites: + remaining = budget_s - (time.monotonic() - start) + if remaining <= 1: + failed_suites.append(f"{spec['path']}: journey time budget exhausted") + break + argv = [sys.executable, "-B", str(Path(__file__).resolve()), "--root", str(root), + "--suite", spec["path"]] + for entry in spec.get("sys_paths", []): + argv += ["--sys-path", entry] + for name in spec.get("stub_modules", []): + argv += ["--stub", name] + try: + proc = subprocess.run(argv, cwd=str(root), capture_output=True, text=True, + timeout=min(per_suite_timeout_s, remaining), check=False) + except subprocess.TimeoutExpired: + failed_suites.append(f"{spec['path']}: timed out") + continue + line = next((ln for ln in reversed(proc.stdout.splitlines()) + if ln.startswith("CERTFORGE_TESTKIT_")), "") + try: + report = json.loads(line.split(" ", 1)[1]) + except (IndexError, ValueError): + report = {} + for key in totals: + totals[key] += int(report.get(key, 0) or 0) + status = "ok" if proc.returncode == 0 else "FAILED" + print(f"suite {status} {spec['path']} passed={report.get('passed')} " + f"failed={report.get('failed')} skipped={report.get('skipped')} " + f"elapsed_s={report.get('elapsed_s')}") + if proc.returncode != 0: + failed_suites.append(spec["path"]) + sys.stderr.write(proc.stderr[-3000:]) + ok = not failed_suites and totals["failed"] == 0 and totals["passed"] >= min_passed + summary = dict(totals, suites=len(suites), failed_suites=failed_suites, + elapsed_s=round(time.monotonic() - start, 2)) + print(("CERTFORGE_SUITES_OK " if ok else "CERTFORGE_SUITES_FAILED ") + + json.dumps(summary, sort_keys=True)) + return 0 if ok else 1 + + +if __name__ == "__main__": # single-suite worker used by run_isolated + import argparse + + parser = argparse.ArgumentParser() + parser.add_argument("--root", required=True) + parser.add_argument("--suite", required=True) + parser.add_argument("--sys-path", action="append", default=[]) + parser.add_argument("--stub", action="append", default=[]) + parser.add_argument("--budget", type=float, default=60.0) + ns = parser.parse_args() + raise SystemExit(run(Path(ns.root), [ns.suite], sys_paths=ns.sys_path, + stub_modules=ns.stub, budget_s=ns.budget)) diff --git a/scripts/run_p4_gate.sh b/scripts/run_p4_gate.sh index 046b33b..21bce96 100644 --- a/scripts/run_p4_gate.sh +++ b/scripts/run_p4_gate.sh @@ -25,7 +25,7 @@ set -Eeuo pipefail REPO_URL="${REPO_URL:-https://github.com/ECHO-OMEGA-PRIME/echo-certification-forge}" BRANCH="${BRANCH:-feat/certforge-r5-negative-controls}" GIT_CRED_FILE="${GIT_CRED_FILE:-/home/forge/.config/echo/omega_git_creds}" -BASE_DIGEST="${BASE_DIGEST:-sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df}" +BASE_DIGEST="${BASE_DIGEST:-sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111}" CLAMAV_IMAGE="${CLAMAV_IMAGE:-clamav/clamav@sha256:7f5389ccaa2368c383fa80e167ccfe44348d71e685f926fce4755eed1757673a}" COSIGN="${COSIGN:-/home/forge/.cache/echo-certforge/p4-9c07eb7/tools/cosign/cosign}" TRIVY="${TRIVY:-/home/forge/.cache/echo-certforge/p4-9c07eb7/tools/trivy/trivy}" diff --git a/src/echo_certification_forge/sandbox.py b/src/echo_certification_forge/sandbox.py index a6ed098..96ed7f9 100644 --- a/src/echo_certification_forge/sandbox.py +++ b/src/echo_certification_forge/sandbox.py @@ -25,7 +25,7 @@ from typing import Callable # Pinned minimal Python base (same digest the P4 supply-chain pipeline pins). Override per policy. -DEFAULT_IMAGE = "python:3.12-alpine@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df" +DEFAULT_IMAGE = "python:3.12-alpine@sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111" class SandboxError(RuntimeError): diff --git a/tests/test_release_pins.py b/tests/test_release_pins.py new file mode 100644 index 0000000..152dc36 --- /dev/null +++ b/tests/test_release_pins.py @@ -0,0 +1,69 @@ +"""ops-20260924: the supply-chain pins stay consistent and patched (issue #22 security release). + +P4 image sealing fails closed on HIGH vulnerabilities with an available fix. This guards the +fix: one patched base image digest everywhere, the patched cryptography release hash-locked, +and the dispatcher unit decoupled from API restarts. +""" +from __future__ import annotations + +import re +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +PATCHED_BASE = "sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111" +VULNERABLE_BASE = "sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df" +ROLES = ("runner", "custody", "anchor", "signer", "worker", "verifier") + + +def _text(rel: str) -> str: + return (ROOT / rel).read_text(encoding="utf-8") + + +def test_every_role_image_uses_the_patched_base() -> None: + for role in ROLES: + dockerfile = _text(f"images/{role}/Dockerfile") + assert dockerfile.startswith(f"FROM python:3.12-alpine@{PATCHED_BASE}\n"), role + assert f'org.opencontainers.image.base.digest="{PATCHED_BASE}"' in dockerfile, role + assert VULNERABLE_BASE not in dockerfile, role + + +def test_sandbox_and_p4_gate_share_the_patched_base() -> None: + assert f'DEFAULT_IMAGE = "python:3.12-alpine@{PATCHED_BASE}"' in _text( + "src/echo_certification_forge/sandbox.py") + assert f'BASE_DIGEST="${{BASE_DIGEST:-{PATCHED_BASE}}}"' in _text("scripts/run_p4_gate.sh") + + +def test_cryptography_is_the_patched_hash_locked_release() -> None: + assert "cryptography==50.0.0" in _text("images/requirements.in") + lock = _text("images/requirements.lock").replace("\r\n", "\n") + block = re.search(r"^cryptography==(\S+) \\\n((?: --hash=sha256:[0-9a-f]{64}(?: \\)?\n)+)", lock, re.M) + assert block is not None + major = int(block.group(1).split(".")[0]) + assert major >= 50 + assert len(re.findall(r"sha256:[0-9a-f]{64}", block.group(2))) >= 20 + assert "cryptography==49" not in lock + + +def test_dispatcher_wants_but_does_not_require_the_api_service() -> None: + script = _text("deploy/deploy_forge.sh") + dispatcher = script[script.index('sudo tee "$DISPATCH_UNIT_PATH"'):] + unit = dispatcher[: dispatcher.index("[Service]")] + assert "Wants=$SERVICE.service" in unit + assert "Requires=$SERVICE.service" not in unit + assert "After=network.target $SERVICE.service" in unit + + +def test_source_fetch_credential_helper_is_configurable_with_safe_default() -> None: + script = _text("deploy/deploy_forge.sh") + assert ('GIT_CREDENTIAL_HELPER="${CERTFORGE_GIT_CREDENTIAL_HELPER:-store --file=' + '/home/forge/.config/echo/omega_git_creds}"') in script + assert 'GITC=(-c credential.helper= -c credential.helper="$GIT_CREDENTIAL_HELPER")' in script + + +def test_deploy_default_manifest_digest_matches_the_enforced_pin() -> None: + script = _text("deploy/deploy_forge.sh") + run_worker = _text("src/echo_certification_forge/run_worker.py") + pinned = re.search(r'_PRODUCTION_MANIFEST_SHA256 = \(\s*"([0-9a-f]{64})"', run_worker) + default = re.search(r'TRUSTED_MANIFEST_SHA256="\$\{ECHO_CERTFORGE_TRUSTED_MANIFEST_SHA256:-([0-9a-f]{64})\}"', + script) + assert pinned and default and default.group(1) == pinned.group(1) From d8ad9fd881365801354b78a758632d2a31ab6828 Mon Sep 17 00:00:00 2001 From: echo-ops-certforge Date: Thu, 24 Sep 2026 16:04:54 -0500 Subject: [PATCH 03/10] fix(p4): keep the worker health probe's deployment ledger on the workspace (#22) The fresh P4 hostile-acceptance run for bc1b8ad built and sealed all 12 images and passed the scans, the static and runtime attack matrices and the custody and anchor service probes. It then stopped as INFRA_FAILED: the worker container exited before /healthz answered. Cause: since P6 (bfc2159) the worker app opens its deployment ledger at import time. Its default path, /var/deployments.sqlite3, resolves to /opt/var on the read-only image root, so uvicorn died with "Read-only file system: '/opt/var'". The last green P4 (9c07eb7) predates P6, so no P6-era source could pass P4 until now. Fix: the hardened probe sets ECHO_CERTFORGE_DEPLOYMENT_LEDGER to /workspace/state/deployments.sqlite3, next to the other state paths. Only the harness changes; production and the image stay as they are. Reproduced with the sealed worker image: exit 1 without the variable, healthy with it. A new test pins every worker state path to the writable workspace; it fails without this change. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0152mN1wMwj9vE2YmV2xZF4F --- scripts/p4_hostile_acceptance.py | 4 +++ tests/test_p4_worker_service_env.py | 51 +++++++++++++++++++++++++++++ 2 files changed, 55 insertions(+) create mode 100644 tests/test_p4_worker_service_env.py diff --git a/scripts/p4_hostile_acceptance.py b/scripts/p4_hostile_acceptance.py index 3ab0f36..f05e19c 100644 --- a/scripts/p4_hostile_acceptance.py +++ b/scripts/p4_hostile_acceptance.py @@ -1394,6 +1394,10 @@ def service_health_probe(image: str, role: ImageRole, ownership_token: str, work "ECHO_CERTFORGE_DB": "/workspace/state/certforge.sqlite3", "ECHO_CERTFORGE_EVIDENCE_ROOT": "/workspace/state/evidence", "ECHO_CERTFORGE_TRUSTED_KEYS": "/workspace/state/trusted-public-keys", + # The worker app opens its P6 deployment ledger at import time. The default + # (/var/deployments.sqlite3) is on the read-only image root, so + # the ledger joins the other state paths on the writable workspace. + "ECHO_CERTFORGE_DEPLOYMENT_LEDGER": "/workspace/state/deployments.sqlite3", } ) command = ["-lc", "mkdir -p /workspace/state/trusted-public-keys /workspace/state/evidence && exec uvicorn echo_certification_forge.app:app --host 127.0.0.1 --port 8080 --no-access-log"] diff --git a/tests/test_p4_worker_service_env.py b/tests/test_p4_worker_service_env.py new file mode 100644 index 0000000..db4d071 --- /dev/null +++ b/tests/test_p4_worker_service_env.py @@ -0,0 +1,51 @@ +"""The P4 worker health probe keeps every state path on the writable workspace. + +The hardened probe runs the worker image with a read-only root. Since P6 the worker app opens its +deployment ledger at import time, and its default path lives on the image root, so the probe died with +"Read-only file system: '/opt/var'" before /healthz answered (P4 INFRA_FAILED, ops-20260924). +""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +SCRIPT = ROOT / "scripts" / "p4_hostile_acceptance.py" +STATE_PATHS = ( + "ECHO_CERTFORGE_DB", + "ECHO_CERTFORGE_EVIDENCE_ROOT", + "ECHO_CERTFORGE_TRUSTED_KEYS", + "ECHO_CERTFORGE_DEPLOYMENT_LEDGER", +) + + +class _Captured(Exception): + pass + + +def _load_harness(): + spec = importlib.util.spec_from_file_location("p4_hostile_acceptance_worker_env_test", SCRIPT) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_worker_health_probe_keeps_state_on_the_writable_workspace(monkeypatch, tmp_path: Path) -> None: + harness = _load_harness() + captured: dict = {} + + def fake_create_container(**kwargs): + captured.update(kwargs) + raise _Captured() + + monkeypatch.setattr(harness, "create_container", fake_create_container) + with pytest.raises(_Captured): + harness.service_health_probe("image@sha256:" + "0" * 64, harness.ImageRole.WORKER, "token-12345678", tmp_path) + environment = captured["environment"] + assert environment["ECHO_CERTFORGE_DEPLOYMENT_LEDGER"] == "/workspace/state/deployments.sqlite3" + for name in STATE_PATHS: + assert environment[name].startswith("/workspace/state/"), name From 413e47e72a344d52a289585f44dd037beb429920 Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Sat, 3 Oct 2026 08:41:20 -0500 Subject: [PATCH 04/10] Require patched cryptography for package deployments --- pyproject.toml | 2 +- tests/test_release_pins.py | 14 ++++++++++++++ 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 3f1a10e..0d44c78 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,7 +11,7 @@ requires-python = ">=3.12" license = {text = "Proprietary"} authors = [{name = "ECHO-OMEGA-PRIME", email = "bobbymcwilliams@echo-op.com"}] dependencies = [ - "cryptography>=44.0.0", + "cryptography>=50.0.0", "fastapi>=0.115.0", "pydantic>=2.10.0", "uvicorn>=0.34.0", diff --git a/tests/test_release_pins.py b/tests/test_release_pins.py index 152dc36..831ff73 100644 --- a/tests/test_release_pins.py +++ b/tests/test_release_pins.py @@ -7,8 +7,11 @@ from __future__ import annotations import re +import tomllib from pathlib import Path +from packaging.requirements import Requirement + ROOT = Path(__file__).resolve().parents[1] PATCHED_BASE = "sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111" VULNERABLE_BASE = "sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df" @@ -53,6 +56,17 @@ def test_dispatcher_wants_but_does_not_require_the_api_service() -> None: assert "After=network.target $SERVICE.service" in unit +def test_package_install_rejects_vulnerable_cryptography_versions() -> None: + # The deployment venv installs pyproject.toml, independently of the image lock. + dependencies = tomllib.loads(_text("pyproject.toml"))["project"]["dependencies"] + requirement = next( + item for item in map(Requirement, dependencies) if item.name == "cryptography" + ) + for vulnerable in ("44.0.0", "49.0.0", "49.0.1"): + assert not requirement.specifier.contains(vulnerable) + assert requirement.specifier.contains("50.0.0") + + def test_source_fetch_credential_helper_is_configurable_with_safe_default() -> None: script = _text("deploy/deploy_forge.sh") assert ('GIT_CREDENTIAL_HELPER="${CERTFORGE_GIT_CREDENTIAL_HELPER:-store --file=' From 81356ae2e31fe271a932ebfec44bde86f415a9ab Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Sat, 3 Oct 2026 08:43:05 -0500 Subject: [PATCH 05/10] Keep certification journey independent of packaging test dependencies --- tests/test_package_dependency_policy.py | 17 +++++++++++++++++ tests/test_release_pins.py | 14 -------------- 2 files changed, 17 insertions(+), 14 deletions(-) create mode 100644 tests/test_package_dependency_policy.py diff --git a/tests/test_package_dependency_policy.py b/tests/test_package_dependency_policy.py new file mode 100644 index 0000000..bfe83e1 --- /dev/null +++ b/tests/test_package_dependency_policy.py @@ -0,0 +1,17 @@ +"""Package-install policy, separate from the stdlib-only certification journey.""" +from pathlib import Path +import tomllib + +from packaging.requirements import Requirement + + +def test_package_install_rejects_vulnerable_cryptography_versions() -> None: + # The deployment venv installs pyproject.toml, independently of the image lock. + manifest = Path(__file__).resolve().parents[1] / "pyproject.toml" + dependencies = tomllib.loads(manifest.read_text(encoding="utf-8"))["project"]["dependencies"] + requirement = next( + item for item in map(Requirement, dependencies) if item.name == "cryptography" + ) + for vulnerable in ("44.0.0", "49.0.0", "49.0.1"): + assert not requirement.specifier.contains(vulnerable) + assert requirement.specifier.contains("50.0.0") diff --git a/tests/test_release_pins.py b/tests/test_release_pins.py index 831ff73..152dc36 100644 --- a/tests/test_release_pins.py +++ b/tests/test_release_pins.py @@ -7,11 +7,8 @@ from __future__ import annotations import re -import tomllib from pathlib import Path -from packaging.requirements import Requirement - ROOT = Path(__file__).resolve().parents[1] PATCHED_BASE = "sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111" VULNERABLE_BASE = "sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df" @@ -56,17 +53,6 @@ def test_dispatcher_wants_but_does_not_require_the_api_service() -> None: assert "After=network.target $SERVICE.service" in unit -def test_package_install_rejects_vulnerable_cryptography_versions() -> None: - # The deployment venv installs pyproject.toml, independently of the image lock. - dependencies = tomllib.loads(_text("pyproject.toml"))["project"]["dependencies"] - requirement = next( - item for item in map(Requirement, dependencies) if item.name == "cryptography" - ) - for vulnerable in ("44.0.0", "49.0.0", "49.0.1"): - assert not requirement.specifier.contains(vulnerable) - assert requirement.specifier.contains("50.0.0") - - def test_source_fetch_credential_helper_is_configurable_with_safe_default() -> None: script = _text("deploy/deploy_forge.sh") assert ('GIT_CREDENTIAL_HELPER="${CERTFORGE_GIT_CREDENTIAL_HELPER:-store --file=' From 424fc76fa09d9a83fcbe0d2ec12dc62ef97859c9 Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Sat, 3 Oct 2026 08:51:39 -0500 Subject: [PATCH 06/10] Publish authentic candidate CI and CodeQL using supported Actions --- .github/workflows/ci.yml | 19 +++++++++++----- .github/workflows/codeql.yml | 42 ++++++++++++++++++++++++++++++++++++ deploy/README.md | 8 +++++++ 3 files changed, 64 insertions(+), 5 deletions(-) create mode 100644 .github/workflows/codeql.yml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 01578a4..bc4b3d2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -13,13 +13,22 @@ permissions: jobs: test: - runs-on: ubuntu-latest + # Release Sentinel requires the authentic Actions check named CI. + name: CI + runs-on: ubuntu-24.04 timeout-minutes: 30 steps: - name: Check out exact source - uses: actions/checkout@v4 + uses: actions/checkout@v7 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + persist-credentials: false + - name: Verify candidate identity + env: + EXPECTED_SHA: ${{ github.event.pull_request.head.sha || github.sha }} + run: test "$(git rev-parse HEAD)" = "$EXPECTED_SHA" - name: Configure Python - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: "3.13" cache: pip @@ -37,9 +46,9 @@ jobs: run: python scripts/p7_acceptance.py - name: Upload verification evidence if: always() - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v7 with: - name: certification-forge-evidence-${{ github.sha }} + name: certification-forge-evidence-${{ github.event.pull_request.head.sha || github.sha }} path: artifacts/ if-no-files-found: error retention-days: 30 diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml new file mode 100644 index 0000000..446e95b --- /dev/null +++ b/.github/workflows/codeql.yml @@ -0,0 +1,42 @@ +name: certification-forge-codeql + +"on": + push: + branches: + - main + - feat/p5-adapter-execution + pull_request: + workflow_dispatch: + +permissions: + contents: read + +jobs: + analyze: + name: CodeQL + runs-on: ubuntu-24.04 + timeout-minutes: 30 + permissions: + contents: read + security-events: write + steps: + - name: Check out exact source + uses: actions/checkout@v7 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + persist-credentials: false + - name: Verify candidate identity + env: + EXPECTED_SHA: ${{ github.event.pull_request.head.sha || github.sha }} + run: test "$(git rev-parse HEAD)" = "$EXPECTED_SHA" + - name: Initialize Python analysis + uses: github/codeql-action/init@v4 + with: + languages: python + build-mode: none + - name: Analyze and upload results + uses: github/codeql-action/analyze@v4 + with: + category: /language:python + ref: ${{ github.event_name == 'pull_request' && format('refs/pull/{0}/head', github.event.pull_request.number) || github.ref }} + sha: ${{ github.event.pull_request.head.sha || github.sha }} diff --git a/deploy/README.md b/deploy/README.md index ca18a25..3c1d713 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -16,6 +16,14 @@ Staging failures retain the service log under `$STATE_ROOT/deploy-logs/staging-.log`. Listener ownership, signed readiness verification, live-smoke gates, and rollback requirements still apply. +# Source validation and release evidence + +The `CI` Actions job runs the full deterministic suite with coverage and P1, P6, +and P7 verification. The `CodeQL` job performs Python security analysis and uploads +its results. Both check out the immutable PR head (or event SHA on push), verify +that checkout, and bind their output to it. Their names match Release Sentinel's +required authentic checks; neither is a substitute status or a release verdict. + A green source-validation run does not prove current production identity or provide signed production E2E evidence. Every changed candidate needs fresh exact-SHA certification. Historical smoke reports and old certified tags must not From e1375b6bc4ad12a9703219f6955b2d1e6ef95163 Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Sun, 8 Nov 2026 13:18:45 -0600 Subject: [PATCH 07/10] 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 08/10] 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 09/10] 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 10/10] 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)