From b30cf970bd801d30038135d637bb372474d8e22c Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Tue, 22 Sep 2026 02:03:04 -0500 Subject: [PATCH 1/3] 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 2/3] 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 3/3] 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