From 6122143c4f0ebf8f9e135956e835c0ecdc0bea4a Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Sun, 8 Nov 2026 05:19:09 -0600 Subject: [PATCH] Add independently verified atomic production E2E intake --- README.md | 5 + docs/PRODUCTION_E2E_INTAKE.md | 118 ++++++ .../production_e2e.py | 401 ++++++++++++++++++ .../production_e2e_intake.py | 316 ++++++++++++++ tests/test_production_e2e.py | 301 +++++++++++++ tests/test_production_e2e_intake.py | 258 +++++++++++ 6 files changed, 1399 insertions(+) create mode 100644 docs/PRODUCTION_E2E_INTAKE.md create mode 100644 src/echo_certification_forge/production_e2e.py create mode 100644 src/echo_certification_forge/production_e2e_intake.py create mode 100644 tests/test_production_e2e.py create mode 100644 tests/test_production_e2e_intake.py diff --git a/README.md b/README.md index 43698f9..1f69787 100644 --- a/README.md +++ b/README.md @@ -141,3 +141,8 @@ production release. The bounded [Echo workspace observer](docs/ECHO_WORKSPACE_OBSERVER.md) exercises actual owner API journeys against a pinned isolated canary. Its unsigned observations retain `NOT_READY` and the remaining production admission requirements. + +The [production E2E owner intake](docs/PRODUCTION_E2E_INTAKE.md) verifies an +independently signed, exact-identity envelope and publishes it atomically to the +existing provider directory. It preserves signing custody, trust roots and the +deterministic release authority. diff --git a/docs/PRODUCTION_E2E_INTAKE.md b/docs/PRODUCTION_E2E_INTAKE.md new file mode 100644 index 0000000..a633c32 --- /dev/null +++ b/docs/PRODUCTION_E2E_INTAKE.md @@ -0,0 +1,118 @@ +# Production E2E owner intake + +`python -m echo_certification_forge.production_e2e_intake` implements the atomic +owner installation already specified by the deployed CertForge operations +contract. It supplies the existing `DirectoryProductionE2EProvider`; it does +not create a second release authority or a new mandatory release rule. + +The operator selects the canonical target and environment from the exact +certification run, the existing independently managed public-key directory, +and the existing attestation directory. The collector supplies only an +envelope. Its embedded public key never establishes trust. + +## Qualification before promotion + +The deployed sequence permits a production-shaped journey against an isolated +exact candidate before promotion. A candidate therefore does not need to be +already running in production to be observed. The observation must accurately +state its scope. For a private candidate lane, preserve fields such as +`execution_scope: private_exact_artifact_acceptance` and +`public_deployment_verified: false`; installation does not change those fields. + +An ordinary App staging result may establish the exact candidate and identify +the remaining production-E2E rule. That result is not itself a production +attestation. The actual collector still has to execute the required journeys, +negative controls, exact identity readbacks, at least three stability probes +and external acceptance, using its independently established authority. + +The existing six-check production contract, lifetime limit, signatures and +exact target/environment/source/deployment bindings remain unchanged. Nothing +in this tool converts an unsigned engineering report or successful process +exit into trusted production evidence. + +## Operator invocation + +The owner identity document contains exactly `target` and `environment`, using +the canonical `TargetIdentity` and `EnvironmentIdentity` fields recorded by the +control plane. Obtain its digest and the envelope's digest from the reviewed +owner handoff. A collector-controlled identity document is not an independent +binding. + +The production roots already documented for FORGE are: + +- Attestations: `/var/lib/echo-certforge/production-e2e/attestations` +- Public trust: `/var/lib/echo-certforge/production-e2e/trusted-keys` + +First verify the exact proposed bytes without writing an intake file or lock: + +```sh +python -m echo_certification_forge.production_e2e_intake \ + --envelope /absolute/reviewed/envelope.json \ + --envelope-sha256 EXACT_REVIEWED_ENVELOPE_FILE_SHA256 \ + --identities /absolute/owner/canonical-run-identities.json \ + --identities-sha256 EXACT_OWNER_IDENTITY_FILE_SHA256 \ + --attestation-directory /var/lib/echo-certforge/production-e2e/attestations \ + --trusted-key-directory /var/lib/echo-certforge/production-e2e/trusted-keys +``` + +Execute the same owner command with `--install` to publish the verified bytes. +The default owner UID is zero. The operational entrypoint requires protected +POSIX directories and normal operating-system write authority. It does not +elevate privileges or accept an owner UID override from the command line. + +For a renewal, additionally supply `--replace-content-sha256` with the exact +current installed file digest. A different existing document blocks the write +unless that digest matches. Repeating the identical verified document returns +`already_present`, without a second publication. + +After installation, use the ordinary certification/App workflow to reevaluate +the exact source and independently verify its new signed verdict. The intake +receipt says `release_verdict: NOT_ISSUED`; successful intake cannot issue a +release, deploy a target or change SDK scopes. + +## File and failure behavior + +The intake verifies reviewed input bytes, the existing signature, the full +production contract and current freshness before taking a lock. It reloads +public trust and checks freshness under the lock and again before publishing. +It rejects symlinked paths, nonregular files, oversized or ambiguous JSON, +unexpected envelope fields, writable trust material, unprotected directories +and wrong owners. It never reads a private key. + +Publication uses a private exclusive temporary file, file `fsync`, atomic +rename, directory `fsync` and exact readback. A nonblocking owner lock prevents +concurrent updates from bypassing the replacement comparison. Pre-publication +failure leaves the existing attestation unchanged. An I/O failure after a +publication attempt returns `state: unknown`; reconcile the installed bytes +before issuing a different write. The same input can then report +`already_present` after current signature, identity and freshness verification, +file and directory `fsync`, and exact readback. Continuing durability failure +keeps the result unknown. Post-publication temporary-file cleanup errors also +preserve that unknown outcome. + +## Contract provenance + +This branch restores `src/echo_certification_forge/production_e2e.py` unchanged +from the actual deployed CertForge revision +`d8ad9fd881365801354b78a758632d2a31ab6828`. Its SHA-256 is +`467a24dba4047e6f8450b79bd03dc752956d4cd4569662f39542d2e4d5ff0de6`. +The existing corresponding tests are restored from that revision as well. +The PR43 base did not include the deployed provider module. Reusing its exact +implementation avoids introducing an alternative signature or eligibility +validator in this owner tool. No mandatory-rule manifest, dispatcher or signer +is altered by this source addition. + +The new tests use ephemeral test-only Ed25519 keys and private temporary +directories. They verify actual signatures, actual provider consumption, +wrong identity, missing checks, expired evidence, untrusted embedded keys, +replacement conflicts, real locking, symlinks, nonregular inputs and an +uncertain post-publication I/O outcome. Test keys are never operational trust. + +## Current integration boundary + +This change implements the missing verified atomic installation operation. It +does not provide a generic collector private key, enroll a collector, or claim +that a production signer command exists. Existing runner enrollment and +transport-key rotation are scoped to transport and cannot supply that +authority. A newly written producer still needs the owner's real signing +custody workflow before its output can pass this intake. diff --git a/src/echo_certification_forge/production_e2e.py b/src/echo_certification_forge/production_e2e.py new file mode 100644 index 0000000..490f180 --- /dev/null +++ b/src/echo_certification_forge/production_e2e.py @@ -0,0 +1,401 @@ +"""Fail-closed validation for signed production end-to-end attestations. + +Source checks, a successful process exit, and a reachable URL are not production +evidence. This contract is evaluated by the deterministic verdict authority +after a trusted collector has signed the attestation. Target-controlled code +cannot manufacture the attestation or select its trust key. +""" +from __future__ import annotations + +import json +import re +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from .canonical import sha256_json +from .models import EnvironmentIdentity, SignedVerdictEnvelope, TargetIdentity +from .signing import TrustedPublicKeyRegistry + +RULE_ID = "production_e2e" +SCHEMA_VERSION = "certforge.production-e2e.v1" +GENERIC_PROFILE = "generic-production-v1" +ECHO_GITHUB_AUTONOMY_PROFILE = "echo-github-autonomy-remote-mcp-v2" +ECHO_GITHUB_AUTONOMY_REPOSITORY = "echoomegaprime/echo-github-autonomy" +ECHO_GITHUB_AUTONOMY_CANONICAL_MCP = "https://echo-ghub.grok.me/api/plugin/mcp" + +BASE_CHECKS = frozenset( + { + "runtime_or_artifact_executed", + "exact_identity_readback", + "critical_journeys_complete", + "negative_controls_pass", + "stability_verified", + "external_acceptance_verified", + } +) +ECHO_GITHUB_AUTONOMY_CHECKS = BASE_CHECKS | frozenset( + { + "canonical_mcp_health", + "oauth_discovery", + "mcp_initialize", + "tool_schema", + "repeated_tool_invocation", + "registry_persistence", + "three_account_reconciliation", + "private_public_visibility", + "read_write_certify", + "cross_client_consistency", + } +) +ECHO_GITHUB_ACCOUNTS = { + "echoomegaprime": 314902331, + "ECHO-OMEGA-PRIME": 264607697, + "bobmcwilliams4": 235318155, +} +ECHO_CLIENTS = frozenset({"chatgpt", "claude", "codex", "grok"}) + +_SHA = re.compile(r"^[0-9a-f]{40}(?:[0-9a-f]{24})?$") +_SHA256 = re.compile(r"^[0-9a-f]{64}$") +_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") +_SIGNING_KEY_ID = re.compile(r"^ed25519:[0-9a-f]{32}$") +_SECRET_VALUE = re.compile( + r"(?:gh[pousr]_[A-Za-z0-9]{20,}|github_pat_[A-Za-z0-9_]{20,}|" + r"-----BEGIN [A-Z ]*PRIVATE KEY-----|Bearer\s+[A-Za-z0-9._~-]{16,})" +) +_RESTRICTED_KEYS = { + "password", + "passwd", + "secret", + "token", + "private_key", + "authorization", + "cookie", + "api_key", +} +_VERIFIED_MARKER = object() + + +@dataclass(frozen=True, slots=True) +class VerifiedProductionE2E: + """An attestation whose envelope was verified by an independently pinned key registry.""" + + payload: dict[str, Any] + collector_key_id: str + envelope_sha256: str + _marker: object + + def __post_init__(self) -> None: + if self._marker is not _VERIFIED_MARKER: + raise ValueError("VerifiedProductionE2E must come from signature verification") + + def result_details(self) -> dict[str, Any]: + return { + **self.payload, + "signature_verified": True, + "collector_key_id": self.collector_key_id, + "attestation_envelope_sha256": self.envelope_sha256, + } + + +def _parse_time(value: object) -> datetime | None: + if not isinstance(value, str): + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + return parsed.astimezone(UTC) if parsed.tzinfo is not None else None + + +def _contains_restricted_value(value: Any, *, key_name: str = "") -> bool: + normalized = key_name.casefold().replace("-", "_") + if normalized in _RESTRICTED_KEYS: + return True + if isinstance(value, Mapping): + return any( + _contains_restricted_value(item, key_name=str(key)) for key, item in value.items() + ) + if isinstance(value, (list, tuple, set)): + return any(_contains_restricted_value(item) for item in value) + return isinstance(value, str) and bool(_SECRET_VALUE.search(value)) + + +def _source_repository(target: TargetIdentity) -> str | None: + reference = target.canonical_ref.casefold() + marker = "github.com/" + if marker not in reference: + return None + suffix = reference.split(marker, 1)[1].split("@", 1)[0] + if suffix.endswith(".git"): + suffix = suffix[:-4] + return suffix + + +def _validate_common( + payload: Mapping[str, Any], + target: TargetIdentity, + environment: EnvironmentIdentity, + *, + now: datetime, +) -> str | None: + if _contains_restricted_value(payload): + return "production_e2e_contains_restricted_value" + if payload.get("schema_version") != SCHEMA_VERSION: + return "production_e2e_schema_invalid" + if not _IDENTIFIER.fullmatch(str(payload.get("attestation_id") or "")): + return "production_e2e_attestation_id_invalid" + if payload.get("target_identity_digest") != target.identity_digest: + return "production_e2e_target_identity_mismatch" + if payload.get("environment_identity_digest") != environment.identity_digest: + return "production_e2e_environment_identity_mismatch" + if not _SIGNING_KEY_ID.fullmatch(str(payload.get("signing_key_id") or "")): + return "production_e2e_signing_key_id_invalid" + + observed_at = _parse_time(payload.get("observed_at")) + expires_at = _parse_time(payload.get("expires_at")) + current = now.astimezone(UTC) if now.tzinfo is not None else None + if current is None or observed_at is None or expires_at is None: + return "production_e2e_time_invalid" + if expires_at <= observed_at or (expires_at - observed_at).total_seconds() > 3600: + return "production_e2e_lifetime_invalid" + if observed_at > current or current >= expires_at: + return "production_e2e_attestation_not_current" + + source_commit = payload.get("source_commit") + deployment_sha = payload.get("deployment_sha") + if target.source_commit is not None: + expected = target.source_commit.casefold() + if not _SHA.fullmatch(expected): + return "production_e2e_target_source_commit_not_exact" + if source_commit != expected or deployment_sha != expected: + return "production_e2e_deployment_sha_mismatch" + elif source_commit is not None or deployment_sha is not None: + if source_commit != deployment_sha or not _SHA.fullmatch(str(source_commit)): + return "production_e2e_deployment_sha_invalid" + + required_checks = payload.get("required_checks") + checks = payload.get("checks") + if not isinstance(required_checks, list) or not isinstance(checks, Mapping): + return "production_e2e_checks_invalid" + if len(required_checks) != len(set(required_checks)): + return "production_e2e_checks_duplicated" + check_set = frozenset(required_checks) + if set(checks) != set(check_set) or any(checks.get(name) is not True for name in check_set): + return "production_e2e_checks_incomplete" + if not BASE_CHECKS <= check_set: + return "production_e2e_baseline_checks_missing" + if not isinstance(payload.get("stability_probe_count"), int) or payload["stability_probe_count"] < 3: + return "production_e2e_stability_probe_count_insufficient" + return None + + +def _validate_echo_github_autonomy(payload: Mapping[str, Any]) -> str | None: + if payload.get("canonical_target") != ECHO_GITHUB_AUTONOMY_CANONICAL_MCP: + return "production_e2e_canonical_mcp_mismatch" + if frozenset(payload.get("required_checks") or ()) != ECHO_GITHUB_AUTONOMY_CHECKS: + return "production_e2e_github_autonomy_checks_incomplete" + if payload.get("tool_count") != 30: + return "production_e2e_tool_count_mismatch" + + repositories = payload.get("sample_private_repositories") + if not isinstance(repositories, Mapping) or set(repositories) != set(ECHO_GITHUB_ACCOUNTS): + return "production_e2e_sample_repositories_incomplete" + sample_digests: dict[str, str] = {} + for login, sample in repositories.items(): + if not isinstance(sample, Mapping): + return "production_e2e_sample_repository_invalid" + if ( + not isinstance(sample.get("repository_id"), int) + or isinstance(sample["repository_id"], bool) + or sample["repository_id"] <= 0 + ): + return "production_e2e_sample_repository_id_invalid" + if not str(sample.get("node_id") or "").strip(): + return "production_e2e_sample_node_id_invalid" + if not str(sample.get("default_branch") or "").strip(): + return "production_e2e_sample_default_branch_invalid" + if not _SHA.fullmatch(str(sample.get("head_sha") or "")): + return "production_e2e_sample_head_sha_invalid" + digest = str(sample.get("fingerprint_sha256") or "") + if not _SHA256.fullmatch(digest): + return "production_e2e_sample_fingerprint_invalid" + sample_digests[login] = digest + + accounts = payload.get("accounts") + if not isinstance(accounts, Mapping) or set(accounts) != set(ECHO_GITHUB_ACCOUNTS): + return "production_e2e_accounts_incomplete" + for login, account_id in ECHO_GITHUB_ACCOUNTS.items(): + account = accounts.get(login) + if not isinstance(account, Mapping) or account.get("account_id") != account_id: + return "production_e2e_account_identity_mismatch" + counts = ( + account.get("enumerated_count"), + account.get("upstream_total_count"), + account.get("public_count"), + account.get("private_count"), + ) + if any(not isinstance(value, int) or isinstance(value, bool) or value < 0 for value in counts): + return "production_e2e_account_counts_invalid" + if counts[0] != counts[1] or counts[2] <= 0 or counts[3] <= 0: + return "production_e2e_account_reconciliation_failed" + if any(account.get(name) is not True for name in ("read", "write", "certify")): + return "production_e2e_account_authority_incomplete" + if account.get("credential_source") not in { + "github_app_installation", + "vault_user_token_fallback", + }: + return "production_e2e_credential_source_invalid" + if account.get("secret_exposed") is not False: + return "production_e2e_secret_boundary_failed" + + clients = payload.get("clients") + if not isinstance(clients, Mapping) or set(clients) != set(ECHO_CLIENTS): + return "production_e2e_clients_incomplete" + for client in clients.values(): + if not isinstance(client, Mapping) or client.get("accepted") is not True: + return "production_e2e_client_not_accepted" + fingerprints = client.get("repository_fingerprints") + if not isinstance(fingerprints, Mapping) or dict(fingerprints) != sample_digests: + return "production_e2e_cross_client_repository_mismatch" + return None + + +def validate_production_e2e( + payload: Mapping[str, Any] | None, + target: TargetIdentity, + environment: EnvironmentIdentity, + *, + now: datetime, +) -> tuple[bool, str]: + """Validate one trusted-collector payload against the exact run identities.""" + + if not isinstance(payload, Mapping): + return False, "production_e2e_attestation_missing" + error = _validate_common(payload, target, environment, now=now) + if error is not None: + return False, error + repository = _source_repository(target) + profile = payload.get("profile") + if repository == ECHO_GITHUB_AUTONOMY_REPOSITORY: + if profile != ECHO_GITHUB_AUTONOMY_PROFILE: + return False, "production_e2e_profile_mismatch" + error = _validate_echo_github_autonomy(payload) + return (error is None, error or "production_e2e_verified") + if profile != GENERIC_PROFILE: + return False, "production_e2e_profile_mismatch" + if frozenset(payload.get("required_checks") or ()) != BASE_CHECKS: + return False, "production_e2e_generic_checks_incomplete" + return True, "production_e2e_verified" + + +def validate_attestation_trust_metadata(payload: Mapping[str, Any]) -> tuple[bool, str]: + """Require the signature-verification metadata added by ``VerifiedProductionE2E``.""" + + if payload.get("signature_verified") is not True: + return False, "production_e2e_signature_not_verified" + if payload.get("collector_key_id") != payload.get("signing_key_id"): + return False, "production_e2e_collector_key_mismatch" + if not _SIGNING_KEY_ID.fullmatch(str(payload.get("collector_key_id") or "")): + return False, "production_e2e_collector_key_invalid" + if not _SHA256.fullmatch(str(payload.get("attestation_envelope_sha256") or "")): + return False, "production_e2e_envelope_sha256_invalid" + return True, "production_e2e_signature_verified" + + +def verify_signed_attestation( + envelope: SignedVerdictEnvelope, + trusted_keys: TrustedPublicKeyRegistry, +) -> VerifiedProductionE2E: + """Verify one collector envelope against an independently selected key registry.""" + + verified, reason = trusted_keys.verify(envelope) + if not verified: + raise ValueError(f"production E2E attestation is not trusted: {reason}") + if envelope.payload.get("schema_version") != SCHEMA_VERSION: + raise ValueError("production E2E attestation schema is unsupported") + if envelope.payload.get("signing_key_id") != envelope.key_id: + raise ValueError("production E2E attestation key binding is invalid") + envelope_sha256 = sha256_json( + { + "payload": envelope.payload, + "signature_b64": envelope.signature_b64, + "key_id": envelope.key_id, + } + ) + return VerifiedProductionE2E( + payload=dict(envelope.payload), + collector_key_id=envelope.key_id, + envelope_sha256=envelope_sha256, + _marker=_VERIFIED_MARKER, + ) + + +def load_signed_attestation(path: Path, trusted_key_directory: Path) -> VerifiedProductionE2E: + """Load and independently verify a collector envelope using pinned public keys.""" + + try: + document = json.loads(path.read_text(encoding="utf-8")) + envelope = SignedVerdictEnvelope( + payload=dict(document["payload"]), + signature_b64=str(document["signature_b64"]), + key_id=str(document["key_id"]), + public_key_pem=str(document.get("public_key_pem") or ""), + ) + except (OSError, ValueError, TypeError, KeyError, json.JSONDecodeError) as exc: + raise ValueError("production E2E attestation envelope is invalid") from exc + registry = TrustedPublicKeyRegistry.from_directory(trusted_key_directory) + return verify_signed_attestation(envelope, registry) + + +@dataclass(frozen=True, slots=True) +class DirectoryProductionE2EProvider: + """Resolve a signed attestation for one exact target/environment pair. + + The directory and trust roots are selected by the control plane. Target code + can neither choose an arbitrary file nor supply a public key. Missing or + untrusted envelopes are treated as absent so the deterministic verdict stays + ``NOT_READY``; a successfully verified envelope is still identity- and + freshness-checked by the executor. + """ + + attestation_directory: Path + trusted_key_directory: Path + + def __post_init__(self) -> None: + try: + attestation_root = self.attestation_directory.resolve(strict=True) + trust_root = self.trusted_key_directory.resolve(strict=True) + except OSError as exc: + raise ValueError("production E2E trust root is unavailable") from exc + if not attestation_root.is_dir() or not trust_root.is_dir(): + raise ValueError("production E2E trust root is not a directory") + object.__setattr__(self, "attestation_directory", attestation_root) + object.__setattr__(self, "trusted_key_directory", trust_root) + + def __call__( + self, + target: TargetIdentity, + environment: EnvironmentIdentity, + ) -> VerifiedProductionE2E | None: + target_digest = target.identity_digest + environment_digest = environment.identity_digest + if not _SHA256.fullmatch(target_digest) or not _SHA256.fullmatch(environment_digest): + return None + candidate = self.attestation_directory / ( + f"{target_digest}.{environment_digest}.json" + ) + try: + if candidate.is_symlink() or not candidate.is_file(): + return None + resolved = candidate.resolve(strict=True) + except OSError: + return None + if resolved.parent != self.attestation_directory: + return None + try: + return load_signed_attestation(resolved, self.trusted_key_directory) + except (OSError, TypeError, ValueError): + return None diff --git a/src/echo_certification_forge/production_e2e_intake.py b/src/echo_certification_forge/production_e2e_intake.py new file mode 100644 index 0000000..8d2273c --- /dev/null +++ b/src/echo_certification_forge/production_e2e_intake.py @@ -0,0 +1,316 @@ +"""Owner-operated, verified atomic intake for the existing production E2E provider. + +This module has no signing, key generation, trust enrollment or release-verdict +operation. A separate admitted collector must produce the envelope. The owner +selects the roots and canonical run identities independently of that envelope. +""" +from __future__ import annotations + +import argparse +import errno +import json +import os +import stat +from collections.abc import Callable +from contextlib import contextmanager +from datetime import datetime +from pathlib import Path +from typing import Any, Iterator +from uuid import uuid4 + +from .canonical import require_sha256, sha256_bytes, utc_now +from .echo_workspace_probe import MAX_BODY, ObservationError, bounded_json +from .models import EnvironmentIdentity, SignedVerdictEnvelope, TargetIdentity +from .production_e2e import ( + VerifiedProductionE2E, validate_production_e2e, verify_signed_attestation, +) +from .signing import TrustedPublicKeyRegistry + + +class IntakeError(RuntimeError): + """A fixed diagnostic code, without envelope, credential or key content.""" + + +class IntakeOutcomeUnknown(IntakeError): + """Publication was attempted; reconcile by readback before any new write.""" + + +def _require(condition: Any, code: str) -> None: + if not condition: + raise IntakeError(code) + + +def verify_intake_document( + raw: bytes, trusted_keys: TrustedPublicKeyRegistry, target: TargetIdentity, + environment: EnvironmentIdentity, *, now: datetime, expected_content_sha256: str, +) -> VerifiedProductionE2E: + """Verify reviewed bytes using the SAME contract consumed by the dispatcher.""" + try: + require_sha256(expected_content_sha256, "expected_content_sha256") + _require(sha256_bytes(raw) == expected_content_sha256, "content_digest_mismatch") + document = bounded_json(raw) + _require(set(document) in ( + {"payload", "signature_b64", "key_id"}, + {"payload", "signature_b64", "key_id", "public_key_pem"}, + ), "envelope_fields_invalid") + _require(isinstance(document["payload"], dict) and + all(isinstance(document.get(key), str) for key in ("signature_b64", "key_id")) and + isinstance(document.get("public_key_pem", ""), str), "envelope_fields_invalid") + envelope = SignedVerdictEnvelope( + payload=document["payload"], signature_b64=document["signature_b64"], + key_id=document["key_id"], public_key_pem=document.get("public_key_pem", ""), + ) + verified = verify_signed_attestation(envelope, trusted_keys) + eligible, reason = validate_production_e2e(verified.payload, target, environment, now=now) + _require(eligible, reason) + return verified + except ObservationError as error: + raise IntakeError("envelope_json_invalid") from error + except (ValueError, TypeError, KeyError) as error: + # The existing signature verifier returns fixed reasons, never material. + text = str(error) + reason = next((code for code in ("untrusted_signing_key", "invalid_signature", "payload_key_id_mismatch") + if text.endswith(code)), "envelope_invalid") + raise IntakeError(reason) from error + + +def _no_symlinks(path: Path) -> None: + _require(path.is_absolute() and ".." not in path.parts, "absolute_path_required") + for component in (path, *path.parents): + _require(not component.is_symlink(), "symlink_refused") + + +def _protected_directory(path: Path, owner_uid: int) -> None: + _require(os.name == "posix", "posix_owner_required") + _no_symlinks(path) + for component in (path, *path.parents): + info = component.stat() + _require(stat.S_ISDIR(info.st_mode), "directory_required") + if component == path: + _require(info.st_uid == owner_uid, "directory_owner_mismatch") + _require(not info.st_mode & 0o022, "unprotected_directory") + else: + _require(info.st_uid in (0, owner_uid), "parent_owner_mismatch") + # A sticky system temporary parent cannot rename another owner's child. + _require(not info.st_mode & 0o022 or + (info.st_uid == 0 and info.st_mode & stat.S_ISVTX), "unprotected_parent") + + +def _read_fd(fd: int, *, maximum: int = MAX_BODY, owner_uid: int | None = None) -> bytes: + before = os.fstat(fd) + _require(stat.S_ISREG(before.st_mode) and before.st_size <= maximum, "file_bound") + if owner_uid is not None: + _require(before.st_uid == owner_uid and not before.st_mode & 0o022, "unprotected_file") + chunks, size = [], 0 + while chunk := os.read(fd, min(65536, maximum + 1 - size)): + chunks.append(chunk) + size += len(chunk) + _require(size <= maximum, "file_bound") + after = os.fstat(fd) + _require((before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns) == + (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns), "file_changed_during_read") + return b"".join(chunks) + + +def _read_file(path: Path) -> bytes: + _no_symlinks(path) + try: + fd = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)) + try: + return _read_fd(fd) + finally: + os.close(fd) + except OSError as error: + raise IntakeError("source_unavailable") from error + + +class ProductionE2EIntake: + """Configured by the intake owner, never constructed from a target request. + + FORGE's production directories are root owned, so the operational default + is UID 0. A nonzero owner is useful only for a separately isolated test root. + """ + + def __init__(self, attestation_directory: Path, trusted_key_directory: Path, *, + owner_uid: int = 0, clock: Callable[[], datetime] = utc_now) -> None: + _require(type(owner_uid) is int and owner_uid >= 0, "owner_uid_invalid") + self.root, self.trust, self.owner_uid, self.clock = ( + attestation_directory, trusted_key_directory, owner_uid, clock, + ) + _protected_directory(self.root, self.owner_uid) + _protected_directory(self.trust, self.owner_uid) + + def _registry(self) -> TrustedPublicKeyRegistry: + _protected_directory(self.trust, self.owner_uid) + registry = TrustedPublicKeyRegistry.empty() + directory = os.open(self.trust, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + try: + names = sorted(name for name in os.listdir(directory) if name.endswith(".pem")) + _require(0 < len(names) <= 128, "trust_registry_bound") + for name in names: + try: + fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory) + try: + raw = _read_fd(fd, maximum=65536, owner_uid=self.owner_uid) + finally: + os.close(fd) + registry.add_pem(raw.decode("ascii")) + except (OSError, UnicodeError, ValueError, TypeError) as error: + raise IntakeError("trust_registry_invalid") from error + finally: + os.close(directory) + return registry + + def _verify(self, raw: bytes, target: TargetIdentity, environment: EnvironmentIdentity, + expected_content_sha256: str) -> VerifiedProductionE2E: + return verify_intake_document(raw, self._registry(), target, environment, + now=self.clock(), expected_content_sha256=expected_content_sha256) + + @staticmethod + def _name(target: TargetIdentity, environment: EnvironmentIdentity) -> str: + return f"{target.identity_digest}.{environment.identity_digest}.json" + + def _receipt(self, state: str, raw: bytes, verified: VerifiedProductionE2E, + target: TargetIdentity, environment: EnvironmentIdentity) -> dict[str, Any]: + return { + "schema_version": "certforge.production-e2e-intake.v1", "state": state, + "target_identity_digest": target.identity_digest, + "environment_identity_digest": environment.identity_digest, + "file_name": self._name(target, environment), "content_sha256": sha256_bytes(raw), + "envelope_sha256": verified.envelope_sha256, "collector_key_id": verified.collector_key_id, + "signature_verified": True, "release_verdict": "NOT_ISSUED", + } + + def preview(self, source: Path, target: TargetIdentity, environment: EnvironmentIdentity, *, + expected_content_sha256: str) -> dict[str, Any]: + raw = _read_file(source) + verified = self._verify(raw, target, environment, expected_content_sha256) + return self._receipt("verified_not_installed", raw, verified, target, environment) + + @contextmanager + def _locked_directory(self) -> Iterator[int]: + import fcntl # POSIX owner entrypoint, not an import-time Windows dependency. + + _protected_directory(self.root, self.owner_uid) + directory = os.open(self.root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + lock = None + try: + lock = os.open(".production-e2e-intake.lock", os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, + 0o600, dir_fd=directory) + info = os.fstat(lock) + _require(stat.S_ISREG(info.st_mode) and info.st_uid == self.owner_uid and + not info.st_mode & 0o022, "intake_lock_invalid") + try: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as error: + raise IntakeError("intake_busy") from error + yield directory + finally: + if lock is not None: + os.close(lock) + os.close(directory) + + def _existing(self, directory: int, name: str) -> bytes | None: + try: + fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory) + except FileNotFoundError: + return None + except OSError as error: + raise IntakeError("symlink_refused" if error.errno == errno.ELOOP else "intake_read_failed") from error + try: + return _read_fd(fd, owner_uid=self.owner_uid) + finally: + os.close(fd) + + def install(self, source: Path, target: TargetIdentity, environment: EnvironmentIdentity, *, + expected_content_sha256: str, replace_content_sha256: str | None = None) -> dict[str, Any]: + raw = _read_file(source) + # Reject unsigned/ineligible input before acquiring a write lock. + self._verify(raw, target, environment, expected_content_sha256) + name = self._name(target, environment) + with self._locked_directory() as directory: + # Reload independent trust and recheck expiry AFTER the lock acquisition. + verified = self._verify(raw, target, environment, expected_content_sha256) + old = self._existing(directory, name) + if old == raw: + try: + fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory) + try: + os.fsync(fd) + finally: + os.close(fd) + os.fsync(directory) + _require(self._existing(directory, name) == raw, "installed_readback_mismatch") + except (OSError, IntakeError) as error: + raise IntakeOutcomeUnknown("installation_outcome_unknown") from error + return self._receipt("already_present", raw, verified, target, environment) + _require((old is None and replace_content_sha256 is None) or + (old is not None and sha256_bytes(old) == replace_content_sha256), "replacement_conflict") + temporary = f".production-e2e-{uuid4().hex}.tmp" + publication_attempted = False + try: + fd = os.open(temporary, os.O_CREAT | os.O_EXCL | os.O_WRONLY | os.O_NOFOLLOW, + 0o600, dir_fd=directory) + with os.fdopen(fd, "wb") as stream: + stream.write(raw) + stream.flush() + os.fsync(stream.fileno()) + # A verified document must still be current immediately before publication. + verified = self._verify(raw, target, environment, expected_content_sha256) + publication_attempted = True + os.replace(temporary, name, src_dir_fd=directory, dst_dir_fd=directory) + os.fsync(directory) + _require(self._existing(directory, name) == raw, "installed_readback_mismatch") + except (OSError, IntakeError) as error: + if publication_attempted: + raise IntakeOutcomeUnknown("installation_outcome_unknown") from error + raise + finally: + try: + os.unlink(temporary, dir_fd=directory) + except FileNotFoundError: + pass + except OSError as error: + if publication_attempted: + raise IntakeOutcomeUnknown("installation_outcome_unknown") from error + raise IntakeError("temporary_cleanup_failed") from error + return self._receipt("installed" if old is None else "replaced", raw, verified, target, environment) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--envelope", type=Path, required=True) + parser.add_argument("--envelope-sha256", required=True) + parser.add_argument("--identities", type=Path, required=True, + help="Owner-selected canonical target and environment from the exact certification run") + parser.add_argument("--identities-sha256", required=True) + parser.add_argument("--attestation-directory", type=Path, required=True) + parser.add_argument("--trusted-key-directory", type=Path, required=True) + parser.add_argument("--replace-content-sha256") + parser.add_argument("--install", action="store_true", help="Atomically install the verified envelope") + args = parser.parse_args(argv) + try: + raw = _read_file(args.identities) + _require(sha256_bytes(raw) == args.identities_sha256, "identities_digest_mismatch") + identities = bounded_json(raw) + _require(set(identities) == {"target", "environment"}, "identities_fields_invalid") + target, environment = TargetIdentity(**identities["target"]), EnvironmentIdentity(**identities["environment"]) + intake = ProductionE2EIntake(args.attestation_directory, args.trusted_key_directory) + if args.install: + result = intake.install(args.envelope, target, environment, + expected_content_sha256=args.envelope_sha256, + replace_content_sha256=args.replace_content_sha256) + else: + _require(args.replace_content_sha256 is None, "replacement_requires_install") + result = intake.preview(args.envelope, target, environment, expected_content_sha256=args.envelope_sha256) + print(json.dumps(result, sort_keys=True)) + return 0 + except (IntakeError, ObservationError, OSError, TypeError, ValueError) as error: + print(json.dumps({"state": "unknown" if isinstance(error, IntakeOutcomeUnknown) else "refused", + "release_verdict": "NOT_ISSUED", + "reason": str(error) if isinstance(error, IntakeError) else "intake_input_invalid"})) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_production_e2e.py b/tests/test_production_e2e.py new file mode 100644 index 0000000..3358c55 --- /dev/null +++ b/tests/test_production_e2e.py @@ -0,0 +1,301 @@ +from __future__ import annotations + +import json +from datetime import UTC, datetime, timedelta + +import pytest + +from echo_certification_forge.models import EnvironmentIdentity, TargetIdentity +from echo_certification_forge.production_e2e import ( + DirectoryProductionE2EProvider, + ECHO_CLIENTS, + ECHO_GITHUB_ACCOUNTS, + ECHO_GITHUB_AUTONOMY_CANONICAL_MCP, + ECHO_GITHUB_AUTONOMY_CHECKS, + ECHO_GITHUB_AUTONOMY_PROFILE, + SCHEMA_VERSION, + load_signed_attestation, + validate_production_e2e, +) +from echo_certification_forge.signing import Ed25519VerdictSigner + +NOW = datetime(2026, 8, 30, 12, 0, tzinfo=UTC) +SOURCE_SHA = "5" * 40 + + +def _target() -> TargetIdentity: + return TargetIdentity( + tenant_id="echo-github-apps", + target_type="git", + canonical_ref=( + "https://github.com/echoomegaprime/echo-github-autonomy.git@" + SOURCE_SHA + ), + artifact_sha256="a" * 64, + source_commit=SOURCE_SHA, + dependency_sha256="b" * 64, + configuration_sha256="c" * 64, + ) + + +def _environment() -> EnvironmentIdentity: + return EnvironmentIdentity( + **{ + name: str(index) * 64 + for index, name in enumerate(EnvironmentIdentity.__dataclass_fields__, start=1) + } + ) + + +def _payload(*, signing_key_id: str = "ed25519:" + "d" * 32) -> dict: + target = _target() + environment = _environment() + samples = { + login: { + "repository_id": index, + "node_id": f"R_repo_{index}", + "default_branch": "main", + "head_sha": str(index) * 40, + "fingerprint_sha256": str(index) * 64, + } + for index, login in enumerate(ECHO_GITHUB_ACCOUNTS, start=1) + } + accounts = { + login: { + "account_id": account_id, + "enumerated_count": 10 + index, + "upstream_total_count": 10 + index, + "public_count": 5, + "private_count": 5 + index, + "read": True, + "write": True, + "certify": True, + "credential_source": ( + "github_app_installation" + if index == 1 + else "vault_user_token_fallback" + ), + "secret_exposed": False, + } + for index, (login, account_id) in enumerate(ECHO_GITHUB_ACCOUNTS.items(), start=1) + } + fingerprints = { + login: sample["fingerprint_sha256"] for login, sample in samples.items() + } + return { + "schema_version": SCHEMA_VERSION, + "attestation_id": "echo-github-autonomy-live-001", + "profile": ECHO_GITHUB_AUTONOMY_PROFILE, + "target_identity_digest": target.identity_digest, + "environment_identity_digest": environment.identity_digest, + "source_commit": SOURCE_SHA, + "deployment_sha": SOURCE_SHA, + "canonical_target": ECHO_GITHUB_AUTONOMY_CANONICAL_MCP, + "required_checks": sorted(ECHO_GITHUB_AUTONOMY_CHECKS), + "checks": {name: True for name in sorted(ECHO_GITHUB_AUTONOMY_CHECKS)}, + "stability_probe_count": 3, + "tool_count": 30, + "accounts": accounts, + "sample_private_repositories": samples, + "clients": { + name: {"accepted": True, "repository_fingerprints": fingerprints} + for name in sorted(ECHO_CLIENTS) + }, + "observed_at": NOW.isoformat(), + "expires_at": (NOW + timedelta(minutes=30)).isoformat(), + "signing_key_id": signing_key_id, + } + + +def test_echo_github_autonomy_requires_complete_exact_cross_client_e2e() -> None: + valid, reason = validate_production_e2e( + _payload(), _target(), _environment(), now=NOW + timedelta(minutes=1) + ) + assert valid, reason + + +@pytest.mark.parametrize( + ("mutation", "reason"), + [ + (lambda value: value.__setitem__("deployment_sha", "6" * 40), "production_e2e_deployment_sha_mismatch"), + (lambda value: value.__setitem__("tool_count", 27), "production_e2e_tool_count_mismatch"), + ( + lambda value: value["accounts"]["bobmcwilliams4"].__setitem__( + "private_count", 0 + ), + "production_e2e_account_reconciliation_failed", + ), + ( + lambda value: value["clients"]["grok"].__setitem__("accepted", False), + "production_e2e_client_not_accepted", + ), + ( + lambda value: value["accounts"]["bobmcwilliams4"].__setitem__( + "credential_source", "model_config_pat" + ), + "production_e2e_credential_source_invalid", + ), + ( + lambda value: value["accounts"]["bobmcwilliams4"].__setitem__( + "secret_exposed", True + ), + "production_e2e_secret_boundary_failed", + ), + ( + lambda value: value["sample_private_repositories"]["bobmcwilliams4"].__setitem__( + "repository_id", True + ), + "production_e2e_sample_repository_id_invalid", + ), + ], +) +def test_incomplete_or_mismatched_e2e_fails_closed(mutation, reason: str) -> None: + payload = _payload() + mutation(payload) + valid, actual = validate_production_e2e( + payload, _target(), _environment(), now=NOW + timedelta(minutes=1) + ) + assert not valid + assert actual == reason + + +def test_stale_e2e_attestation_fails_closed() -> None: + valid, reason = validate_production_e2e( + _payload(), _target(), _environment(), now=NOW + timedelta(hours=2) + ) + assert not valid + assert reason == "production_e2e_attestation_not_current" + + +def test_retired_account_and_v1_profile_are_rejected() -> None: + retired_account = _payload() + retired_account["accounts"]["Bmcbob76"] = { + **retired_account["accounts"]["bobmcwilliams4"], + "account_id": 203470412, + } + valid, reason = validate_production_e2e( + retired_account, _target(), _environment(), now=NOW + timedelta(minutes=1) + ) + assert not valid + assert reason == "production_e2e_accounts_incomplete" + + retired_profile = _payload() + retired_profile["profile"] = "echo-github-autonomy-remote-mcp-v1" + valid, reason = validate_production_e2e( + retired_profile, _target(), _environment(), now=NOW + timedelta(minutes=1) + ) + assert not valid + assert reason == "production_e2e_profile_mismatch" + + +def test_attestation_loader_requires_a_pinned_collector_key(tmp_path) -> None: + signer = Ed25519VerdictSigner.generate() + payload = _payload(signing_key_id=signer.key_id) + envelope = signer.sign_payload(payload) + envelope_path = tmp_path / "attestation.json" + envelope_path.write_text( + json.dumps( + { + "payload": envelope.payload, + "signature_b64": envelope.signature_b64, + "key_id": envelope.key_id, + "public_key_pem": envelope.public_key_pem, + } + ), + encoding="utf-8", + ) + trusted = tmp_path / "trusted" + trusted.mkdir() + with pytest.raises(ValueError, match="untrusted_signing_key"): + load_signed_attestation(envelope_path, trusted) + (trusted / "collector.pem").write_text(signer.public_key_pem, encoding="ascii") + verified = load_signed_attestation(envelope_path, trusted) + assert verified.payload == payload + assert verified.collector_key_id == signer.key_id + assert len(verified.envelope_sha256) == 64 + + +def test_self_selected_public_key_cannot_replace_the_pinned_collector(tmp_path) -> None: + trusted_signer = Ed25519VerdictSigner.generate() + attacker = Ed25519VerdictSigner.generate() + payload = _payload(signing_key_id=attacker.key_id) + envelope = attacker.sign_payload(payload) + envelope_path = tmp_path / "attestation.json" + envelope_path.write_text( + json.dumps( + { + "payload": envelope.payload, + "signature_b64": envelope.signature_b64, + "key_id": envelope.key_id, + "public_key_pem": envelope.public_key_pem, + } + ), + encoding="utf-8", + ) + trusted = tmp_path / "trusted" + trusted.mkdir() + (trusted / "collector.pem").write_text(trusted_signer.public_key_pem, encoding="ascii") + with pytest.raises(ValueError, match="untrusted_signing_key"): + load_signed_attestation(envelope_path, trusted) + + +def test_directory_provider_resolves_only_the_exact_target_and_environment(tmp_path) -> None: + signer = Ed25519VerdictSigner.generate() + payload = _payload(signing_key_id=signer.key_id) + envelope = signer.sign_payload(payload) + attestations = tmp_path / "attestations" + trusted = tmp_path / "trusted" + attestations.mkdir() + trusted.mkdir() + (trusted / "collector.pem").write_text(signer.public_key_pem, encoding="ascii") + expected_path = attestations / ( + f"{_target().identity_digest}.{_environment().identity_digest}.json" + ) + expected_path.write_text( + json.dumps( + { + "payload": envelope.payload, + "signature_b64": envelope.signature_b64, + "key_id": envelope.key_id, + "public_key_pem": envelope.public_key_pem, + } + ), + encoding="utf-8", + ) + + provider = DirectoryProductionE2EProvider(attestations, trusted) + verified = provider(_target(), _environment()) + assert verified is not None + assert verified.payload == payload + + +def test_directory_provider_fails_closed_for_missing_or_untrusted_envelopes(tmp_path) -> None: + trusted_signer = Ed25519VerdictSigner.generate() + attacker = Ed25519VerdictSigner.generate() + attestations = tmp_path / "attestations" + trusted = tmp_path / "trusted" + attestations.mkdir() + trusted.mkdir() + (trusted / "collector.pem").write_text( + trusted_signer.public_key_pem, + encoding="ascii", + ) + provider = DirectoryProductionE2EProvider(attestations, trusted) + assert provider(_target(), _environment()) is None + + payload = _payload(signing_key_id=attacker.key_id) + envelope = attacker.sign_payload(payload) + expected_path = attestations / ( + f"{_target().identity_digest}.{_environment().identity_digest}.json" + ) + expected_path.write_text( + json.dumps( + { + "payload": envelope.payload, + "signature_b64": envelope.signature_b64, + "key_id": envelope.key_id, + "public_key_pem": envelope.public_key_pem, + } + ), + encoding="utf-8", + ) + assert provider(_target(), _environment()) is None diff --git a/tests/test_production_e2e_intake.py b/tests/test_production_e2e_intake.py new file mode 100644 index 0000000..daa0131 --- /dev/null +++ b/tests/test_production_e2e_intake.py @@ -0,0 +1,258 @@ +"""Exercise the real signature boundary and the owner's atomic filesystem intake.""" +from __future__ import annotations + +import json +import os +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest + +from echo_certification_forge.canonical import canonical_json, sha256_bytes, to_utc_iso +from echo_certification_forge.models import EnvironmentIdentity, TargetIdentity +from echo_certification_forge.production_e2e import BASE_CHECKS, DirectoryProductionE2EProvider +from echo_certification_forge.production_e2e_intake import ( + IntakeError, IntakeOutcomeUnknown, ProductionE2EIntake, verify_intake_document, +) +from echo_certification_forge.signing import Ed25519VerdictSigner, TrustedPublicKeyRegistry + + +@pytest.fixture +def candidate(tmp_path): + now = datetime.now(UTC) + target = TargetIdentity("tenant-a", "repository", "https://github.com/example/product@" + "a" * 40, + "b" * 64, source_commit="a" * 40) + environment = EnvironmentIdentity(**{name: "c" * 64 for name in EnvironmentIdentity.__dataclass_fields__}) + signer = Ed25519VerdictSigner.generate() # Private, ephemeral TEST key; never operational trust. + registry = TrustedPublicKeyRegistry.empty() + registry.add_pem(signer.public_key_pem) + payload = { + "schema_version": "certforge.production-e2e.v1", "profile": "generic-production-v1", + "attestation_id": "test-collector-fresh-observation", "signing_key_id": signer.key_id, + "target_identity_digest": target.identity_digest, "environment_identity_digest": environment.identity_digest, + "source_commit": target.source_commit, "deployment_sha": target.source_commit, + "observed_at": to_utc_iso(now - timedelta(seconds=1)), "expires_at": to_utc_iso(now + timedelta(minutes=10)), + "required_checks": sorted(BASE_CHECKS), "checks": {name: True for name in BASE_CHECKS}, + "stability_probe_count": 3, "execution_scope": "private_exact_artifact_acceptance", + "public_deployment_verified": False, + } + + def encode(changes=None, use_signer=signer): + body = {**payload, **(changes or {}), "signing_key_id": use_signer.key_id} + signed = use_signer.sign_payload(body) + return canonical_json({"payload": signed.payload, "signature_b64": signed.signature_b64, + "key_id": signed.key_id, "public_key_pem": signed.public_key_pem}).encode() + + return now, target, environment, registry, signer, payload, encode + + +def test_signature_is_verified_against_independent_trust_and_exact_pair(candidate): + now, target, environment, registry, _, _, encode = candidate + raw = encode() + verified = verify_intake_document(raw, registry, target, environment, now=now, + expected_content_sha256=sha256_bytes(raw)) + assert verified.payload["execution_scope"] == "private_exact_artifact_acceptance" + assert verified.payload["public_deployment_verified"] is False + assert verified.collector_key_id in registry.keys + + +@pytest.mark.parametrize("changes,reason", [ + ({"target_identity_digest": "d" * 64}, "target_identity_mismatch"), + ({"environment_identity_digest": "d" * 64}, "environment_identity_mismatch"), + ({"deployment_sha": "d" * 40}, "deployment_sha_mismatch"), + ({"source_commit": "d" * 40}, "deployment_sha_mismatch"), + ({"checks": {name: name != "external_acceptance_verified" for name in BASE_CHECKS}}, "checks_incomplete"), + ({"stability_probe_count": 2}, "probe_count"), + ({"profile": "echo-github-autonomy-remote-mcp-v2"}, "profile_mismatch"), + ({"token": "should-not-be-persisted"}, "restricted_value"), +]) +def test_signed_but_ineligible_documents_never_reach_intake(candidate, changes, reason): + now, target, environment, registry, _, _, encode = candidate + raw = encode(changes) + with pytest.raises(IntakeError, match=reason): + verify_intake_document(raw, registry, target, environment, now=now, + expected_content_sha256=sha256_bytes(raw)) + + +def test_self_supplied_public_key_does_not_establish_trust(candidate): + now, target, environment, registry, _, _, encode = candidate + raw = encode(use_signer=Ed25519VerdictSigner.generate()) + with pytest.raises(IntakeError, match="untrusted_signing_key"): + verify_intake_document(raw, registry, target, environment, now=now, + expected_content_sha256=sha256_bytes(raw)) + + +def test_stale_tampered_or_unreviewed_bytes_are_rejected(candidate): + now, target, environment, registry, _, _, encode = candidate + raw = encode() + with pytest.raises(IntakeError, match="not_current"): + verify_intake_document(raw, registry, target, environment, now=now + timedelta(hours=2), + expected_content_sha256=sha256_bytes(raw)) + with pytest.raises(IntakeError, match="content_digest_mismatch"): + verify_intake_document(raw, registry, target, environment, now=now, expected_content_sha256="0" * 64) + document = json.loads(raw) + document["payload"]["stability_probe_count"] = 4 + changed = canonical_json(document).encode() + with pytest.raises(IntakeError, match="invalid_signature"): + verify_intake_document(changed, registry, target, environment, now=now, + expected_content_sha256=sha256_bytes(changed)) + + +@pytest.mark.parametrize("raw", [b'{}', b'{"payload":{},"payload":{}}', b'{"payload":NaN}', b'[]', b'x' * (1024 * 1024 + 1)], + ids=["empty", "duplicate", "nonfinite", "array", "oversized"]) +def test_unparseable_envelope_is_refused_before_any_write(candidate, raw): + now, target, environment, registry, *_ = candidate + with pytest.raises(IntakeError): + verify_intake_document(raw, registry, target, environment, now=now, + expected_content_sha256=sha256_bytes(raw)) + + +@pytest.fixture +def owner(tmp_path, candidate): + if os.name != "posix": + pytest.skip("The deployed directory intake uses POSIX owner/atomic-rename semantics") + now, target, environment, _, signer, _, encode = candidate + root, trust = tmp_path / "intake", tmp_path / "public-trust" + root.mkdir(mode=0o700) + trust.mkdir(mode=0o700) + (trust / "collector.pem").write_text(signer.public_key_pem) + lane = ProductionE2EIntake(root, trust, owner_uid=os.getuid(), clock=lambda: now) + source = tmp_path / "reviewed-envelope.json" + source.write_bytes(encode()) + return lane, source, root, trust, target, environment, encode + + +def install(owner, **kwargs): + lane, source, _, _, target, environment, _ = owner + return lane.install(source, target, environment, + expected_content_sha256=sha256_bytes(source.read_bytes()), **kwargs) + + +def test_atomic_install_is_consumed_by_actual_directory_provider(owner): + lane, source, root, trust, target, environment, _ = owner + receipt = install(owner) + path = root / (target.identity_digest + "." + environment.identity_digest + ".json") + assert path.read_bytes() == source.read_bytes() + assert receipt["state"] == "installed" and receipt["signature_verified"] is True + assert receipt["release_verdict"] == "NOT_ISSUED" + assert receipt["content_sha256"] == sha256_bytes(path.read_bytes()) + verified = DirectoryProductionE2EProvider(root, trust)(target, environment) + assert verified is not None and verified.envelope_sha256 == receipt["envelope_sha256"] + assert not any(path.name.endswith(".tmp") for path in root.iterdir()) + assert install(owner)["state"] == "already_present" + + +def test_replacement_requires_exact_old_bytes_and_preserves_old_on_conflict(owner): + _, source, root, _, target, environment, encode = owner + first = install(owner) + path = root / (target.identity_digest + "." + environment.identity_digest + ".json") + before = path.read_bytes() + source.write_bytes(encode({"attestation_id": "test-fresh-replacement"})) + with pytest.raises(IntakeError, match="replacement_conflict"): + install(owner) + with pytest.raises(IntakeError, match="replacement_conflict"): + install(owner, replace_content_sha256="0" * 64) + assert path.read_bytes() == before + result = install(owner, replace_content_sha256=first["content_sha256"]) + assert result["state"] == "replaced" and path.read_bytes() == source.read_bytes() + + +def test_dry_run_has_no_intake_or_lock_mutation(owner): + lane, source, root, _, target, environment, _ = owner + result = lane.preview(source, target, environment, expected_content_sha256=sha256_bytes(source.read_bytes())) + assert result["state"] == "verified_not_installed" and list(root.iterdir()) == [] + + +def test_untrusted_signature_cannot_replace_an_existing_attestation(owner): + _, source, root, trust, target, environment, encode = owner + first = install(owner) + path = root / (target.identity_digest + "." + environment.identity_digest + ".json") + before = path.read_bytes() + source.write_bytes(encode(use_signer=Ed25519VerdictSigner.generate())) + with pytest.raises(IntakeError, match="untrusted_signing_key"): + install(owner, replace_content_sha256=first["content_sha256"]) + assert path.read_bytes() == before + assert len(list(trust.iterdir())) == 1 + + +def test_symlinked_evidence_or_intake_is_not_followed(owner, tmp_path): + lane, source, root, trust, target, environment, _ = owner + link = tmp_path / "source-link" + link.symlink_to(source) + with pytest.raises(IntakeError, match="symlink"): + lane.install(link, target, environment, expected_content_sha256=sha256_bytes(source.read_bytes())) + alias = tmp_path / "root-link" + alias.symlink_to(root, target_is_directory=True) + with pytest.raises(IntakeError, match="symlink"): + ProductionE2EIntake(alias, trust, owner_uid=os.getuid()) + output = root / (target.identity_digest + "." + environment.identity_digest + ".json") + output.symlink_to(source) + with pytest.raises(IntakeError, match="symlink"): + install(owner) + + +def test_world_writable_trust_or_wrong_owner_is_refused(owner): + _, _, root, trust, *_ = owner + trust.chmod(0o777) + with pytest.raises(IntakeError, match="unprotected_directory"): + ProductionE2EIntake(root, trust, owner_uid=os.getuid()) + trust.chmod(0o700) + with pytest.raises(IntakeError, match="owner_mismatch"): + ProductionE2EIntake(root, trust, owner_uid=os.getuid() + 1) + + +def test_expiry_is_rechecked_on_install_not_only_preview(owner, candidate): + lane, source, root, trust, target, environment, _ = owner + now = candidate[0] + lane.preview(source, target, environment, expected_content_sha256=sha256_bytes(source.read_bytes())) + later = ProductionE2EIntake(root, trust, owner_uid=os.getuid(), clock=lambda: now + timedelta(hours=2)) + with pytest.raises(IntakeError, match="not_current"): + later.install(source, target, environment, expected_content_sha256=sha256_bytes(source.read_bytes())) + assert not list(root.glob("*.json")) + + +def test_concurrent_owner_operation_refuses_without_waiting_or_overwriting(owner): + lane, _, root, *_ = owner + with lane._locked_directory(): + with pytest.raises(IntakeError, match="intake_busy"): + install(owner) + assert not list(root.glob("*.json")) + assert install(owner)["state"] == "installed" + + +def test_nonregular_source_is_rejected_without_blocking(owner, tmp_path): + lane, _, _, _, target, environment, _ = owner + pipe = tmp_path / "not-an-envelope" + os.mkfifo(pipe) + with pytest.raises(IntakeError, match="file_bound"): + lane.preview(pipe, target, environment, expected_content_sha256="0" * 64) + + +def test_post_publish_io_failure_reports_unknown_and_is_reconciled_by_idempotent_readback(owner, monkeypatch): + real_fsync = os.fsync + + def fail_directory_fsync(fd): + import stat + if stat.S_ISDIR(os.fstat(fd).st_mode): + raise OSError("fixture: directory durability readback unavailable") + real_fsync(fd) + + with monkeypatch.context() as faults: + faults.setattr(os, "fsync", fail_directory_fsync) + with pytest.raises(IntakeOutcomeUnknown, match="installation_outcome_unknown"): + install(owner) + # Reading identical bytes alone must not turn failed durability into success. + with pytest.raises(IntakeOutcomeUnknown, match="installation_outcome_unknown"): + install(owner) + assert install(owner)["state"] == "already_present" + + +def test_cleanup_error_after_publication_preserves_unknown_outcome(owner, monkeypatch): + def fail_cleanup(*_args, **_kwargs): + raise OSError("fixture: cleanup state is unavailable") + + with monkeypatch.context() as faults: + faults.setattr(os, "unlink", fail_cleanup) + with pytest.raises(IntakeOutcomeUnknown, match="installation_outcome_unknown"): + install(owner) + assert install(owner)["state"] == "already_present"