From fb17417732908fe7d251da32f6da512b1984caa0 Mon Sep 17 00:00:00 2001 From: saagpatel <269905221+saagpatel@users.noreply.github.com> Date: Sun, 11 Oct 2026 06:47:44 -0700 Subject: [PATCH 1/2] fix: bind staged portfolio truth to manifest receipt identity --- CHANGELOG.md | 4 + .../portfolio_generation.py | 69 ++- tests/test_portfolio_generation.py | 291 +++++++++- tests/test_portfolio_generation_identity.py | 505 ++++++++++++++++++ 4 files changed, 867 insertions(+), 2 deletions(-) create mode 100644 tests/test_portfolio_generation_identity.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 294ea75..e1b64cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,10 @@ Format: [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) ## [Unreleased] +- Refuse new portfolio generation publication when staged truth lacks valid + GitHub security receipt and content identities or disagrees with the admitted + manifest; preserve historical generation readback compatibility. + - Evaluate security-cohort R3 staleness at the receipt observation time; report clock-only arrivals as pending security coverage and covered clock departures explicitly. Preserve metadata-only R3 failure diagnostics beside publication outputs. - Clarify locked contributor setup, optional test dependencies, and credential-free fixture verification without workstation discovery. diff --git a/src/github_repo_auditor/portfolio_generation.py b/src/github_repo_auditor/portfolio_generation.py index f6f2ceb..cd888af 100644 --- a/src/github_repo_auditor/portfolio_generation.py +++ b/src/github_repo_auditor/portfolio_generation.py @@ -552,6 +552,72 @@ def _pointer_entry(value: object, *, field: str) -> dict[str, str] | None: } +def _validate_staged_truth_security( + stage: Path, *, manifest: Mapping[str, Any] +) -> None: + """Bind newly published truth to its admitted security receipt identity. + + This publication-only check deliberately leaves historical generation + readback compatible with truth written before immutable input identities. + """ + from github_repo_auditor.portfolio_truth_types import SCHEMA_VERSION + + truth_record = next( + ( + record + for record in manifest["artifacts"] + if record["name"] == "portfolio-truth.json" + ), + None, + ) + if truth_record is None: + raise PortfolioGenerationError("portfolio-truth.json artifact is required") + if ( + truth_record["media_type"] != "application/json" + or truth_record["contract_version"] != f"ghra.portfolio_truth@{SCHEMA_VERSION}" + ): + raise PortfolioGenerationError("portfolio truth artifact contract mismatch") + content = _read_regular( + stage / "portfolio-truth.json", label="artifact portfolio-truth.json" + ) + try: + truth = json.loads(content) + except (json.JSONDecodeError, UnicodeDecodeError) as exc: + raise PortfolioGenerationError("portfolio truth is not valid JSON") from exc + if not isinstance(truth, dict): + raise PortfolioGenerationError("portfolio truth must be a JSON object") + if truth.get("schema_version") != SCHEMA_VERSION: + raise PortfolioGenerationError("portfolio truth schema mismatch") + inputs = truth.get("inputs") + if not isinstance(inputs, dict): + raise PortfolioGenerationError("portfolio truth inputs must be a JSON object") + github_security = inputs.get("github_security") + if not isinstance(github_security, dict): + raise PortfolioGenerationError( + "portfolio truth inputs.github_security must be a JSON object" + ) + receipt_id = github_security.get("receipt_id") + if ( + not isinstance(receipt_id, str) + or not receipt_id.startswith("sha256:") + or not SHA256_RE.fullmatch(receipt_id[7:]) + ): + raise PortfolioGenerationError( + "portfolio truth inputs.github_security.receipt_id is invalid" + ) + content_sha256 = _validate_sha256( + github_security.get("content_sha256"), + field="portfolio truth inputs.github_security.content_sha256", + ) + expected = manifest["github_security"] + if receipt_id != expected["receipt_id"]: + raise PortfolioGenerationError("portfolio truth security receipt_id mismatch") + if content_sha256 != expected["content_sha256"]: + raise PortfolioGenerationError( + "portfolio truth security content_sha256 mismatch" + ) + + def read_pointer(root: Path, *, validate_generations: bool = True) -> dict[str, Any]: payload = _read_object(root / POINTER_NAME, label="portfolio generation pointer") if set(payload) != {"schema_version", "current", "previous", "updated_at"}: @@ -652,11 +718,12 @@ def publish_generation( _write_fsynced(stage / MANIFEST_NAME, manifest_bytes, exclusive=True) _fsync_directory(stage) _crash_if_requested("after_manifest_written") - verify_generation_dir( + staged_manifest, _ = verify_generation_dir( stage, expected_generation_id=generation_id, expected_manifest_sha256=manifest_sha256, ) + _validate_staged_truth_security(stage, manifest=staged_manifest) _crash_if_requested("after_stage_validated") generation_dir = releases / generation_id if generation_dir.exists(): diff --git a/tests/test_portfolio_generation.py b/tests/test_portfolio_generation.py index 22b12b9..590c842 100644 --- a/tests/test_portfolio_generation.py +++ b/tests/test_portfolio_generation.py @@ -1,11 +1,15 @@ from __future__ import annotations +import hashlib import json import os import tempfile import unittest from datetime import UTC, datetime, timedelta from pathlib import Path +from unittest.mock import patch + +import github_repo_auditor.portfolio_generation as generation from github_repo_auditor.portfolio_generation import ( _terminal_observed_in_order, @@ -75,7 +79,18 @@ def inputs(self, marker: str = "one") -> list[ArtifactInput]: digest = self.base / f"digest-{marker}.json" markdown = self.base / f"digest-{marker}.md" truth.write_text( - json.dumps({"schema_version": "0.12.0", "marker": marker}) + json.dumps( + { + "schema_version": "0.12.0", + "marker": marker, + "inputs": { + "github_security": { + "receipt_id": security().receipt_id, + "content_sha256": security().content_sha256, + } + }, + } + ) ) digest.write_text( json.dumps( @@ -125,6 +140,280 @@ def test_publishes_content_derived_immutable_generation_and_atomic_pair(self): self.assertEqual(pointer["current"]["generation_id"], second.generation_id) self.assertEqual(pointer["previous"]["generation_id"], first.generation_id) self.assertEqual(resolve_current(root).manifest_sha256, second.manifest_sha256) + manifest = json.loads((second.generation_dir / "manifest.json").read_bytes()) + truth = json.loads( + (second.generation_dir / "portfolio-truth.json").read_bytes() + ) + for field in ("receipt_id", "content_sha256"): + self.assertEqual( + truth["inputs"]["github_security"][field], + manifest["github_security"][field], + ) + + def test_invalid_truth_identity_preserves_pointer_and_visible_generations( + self, + ) -> None: + root = self.base / "managed" + self.publish(root, "one", when=CREATED) + pointer_before = (root / "pointer.json").read_bytes() + releases_before = { + path.relative_to(root): path.read_bytes() + for path in (root / "releases").rglob("*") + if path.is_file() + } + cases = [] + for field in ("receipt_id", "content_sha256"): + for value in ( + None, + True, + 7, + [], + {}, + "", + "invalid", + " " + getattr(security(), field), + ): + cases.append((field, value, "invalid")) + cases.append((field, "missing", "invalid")) + cases.append((field, getattr(security(), field).upper(), "invalid")) + wrong = "sha256:" + "f" * 64 if field == "receipt_id" else "f" * 64 + cases.append((field, wrong, f"{field} mismatch")) + for index, (field, value, message) in enumerate(cases): + with self.subTest(field=field, value=value): + artifacts = self.inputs(f"invalid-{index}") + truth_path = artifacts[0].source + payload = json.loads(truth_path.read_bytes()) + identity = payload["inputs"]["github_security"] + if value == "missing": + del identity[field] + else: + identity[field] = value + truth_path.write_text(json.dumps(payload)) + with patch.object(generation, "_crash_if_requested") as crash_check: + with self.assertRaisesRegex(PortfolioGenerationError, message): + publish_generation( + root=root, + artifacts=artifacts, + producer=producer(), + security=security(), + created_at=CREATED + timedelta(minutes=1), + ) + steps = [call.args[0] for call in crash_check.call_args_list] + self.assertNotIn("after_generation_visible", steps) + self.assertEqual((root / "pointer.json").read_bytes(), pointer_before) + self.assertEqual( + { + path.relative_to(root): path.read_bytes() + for path in (root / "releases").rglob("*") + if path.is_file() + }, + releases_before, + ) + self.assertFalse(list(root.glob(".portfolio-generation-stage-*"))) + + def test_malformed_truth_and_input_objects_are_refused(self) -> None: + cases = [ + (b"{", "not valid JSON"), + (b"\xff", "not valid JSON"), + (b"[]", "must be a JSON object"), + (b"null", "must be a JSON object"), + (b"true", "must be a JSON object"), + (b'"truth"', "must be a JSON object"), + (b'{"schema_version":"0.11.0"}', "schema mismatch"), + (b"{}", "schema mismatch"), + ] + for value in (None, True, "invalid", [], {}): + cases.append( + ( + json.dumps({"schema_version": "0.12.0", "inputs": value}).encode(), + "inputs(\\.github_security)? must be a JSON object", + ) + ) + cases.append( + ( + json.dumps( + { + "schema_version": "0.12.0", + "inputs": {"github_security": value}, + } + ).encode(), + "(must be a JSON object|receipt_id is invalid)", + ) + ) + cases.append((b'{"schema_version":"0.12.0"}', "inputs must be a JSON object")) + for index, (content, message) in enumerate(cases): + with self.subTest(content=content): + root = self.base / f"malformed-{index}" + artifacts = self.inputs(f"malformed-{index}") + artifacts[0].source.write_bytes(content) + with self.assertRaisesRegex(PortfolioGenerationError, message): + publish_generation( + root=root, + artifacts=artifacts, + producer=producer(), + security=security(), + created_at=CREATED, + ) + self.assertFalse((root / "pointer.json").exists()) + self.assertFalse(list((root / "releases").iterdir())) + + def test_canonical_truth_artifact_and_contract_are_required(self) -> None: + cases = ( + ( + None, + "application/json", + "ghra.portfolio_truth@0.12.0", + "artifact is required", + ), + ( + "truth.json", + "application/json", + "ghra.portfolio_truth@0.12.0", + "artifact is required", + ), + ( + "portfolio-truth.json", + "text/plain", + "ghra.portfolio_truth@0.12.0", + "contract mismatch", + ), + ( + "portfolio-truth.json", + "application/json", + "ghra.portfolio_truth@0.11.0", + "contract mismatch", + ), + ) + for index, (name, media_type, contract_version, message) in enumerate(cases): + with self.subTest( + name=name, media_type=media_type, contract=contract_version + ): + root = self.base / f"contract-{index}" + artifacts = self.inputs(f"contract-{index}") + if name is None: + artifacts = artifacts[1:] + else: + artifacts[0] = ArtifactInput( + name, artifacts[0].source, media_type, contract_version + ) + with self.assertRaisesRegex(PortfolioGenerationError, message): + publish_generation( + root=root, + artifacts=artifacts, + producer=producer(), + security=security(), + created_at=CREATED, + ) + self.assertFalse((root / "pointer.json").exists()) + self.assertFalse(list((root / "releases").iterdir())) + + def test_truth_identity_check_uses_staged_bytes(self) -> None: + for stage_valid in (False, True): + with self.subTest(stage_valid=stage_valid): + root = self.base / f"staged-{stage_valid}" + artifacts = self.inputs(f"staged-{stage_valid}") + truth_path = artifacts[0].source + valid = truth_path.read_bytes() + invalid_payload = json.loads(valid) + invalid_payload["inputs"]["github_security"]["receipt_id"] = ( + "sha256:" + "f" * 64 + ) + invalid = json.dumps(invalid_payload).encode() + truth_path.write_bytes(valid if stage_valid else invalid) + + def change_source(step: str) -> None: + if step == "after_artifacts_written": + truth_path.write_bytes(invalid if stage_valid else valid) + + with patch.object( + generation, "_crash_if_requested", side_effect=change_source + ): + if stage_valid: + result = publish_generation( + root=root, + artifacts=artifacts, + producer=producer(), + security=security(), + created_at=CREATED, + ) + self.assertEqual( + ( + result.generation_dir / "portfolio-truth.json" + ).read_bytes(), + valid, + ) + else: + with self.assertRaisesRegex( + PortfolioGenerationError, "receipt_id mismatch" + ): + publish_generation( + root=root, + artifacts=artifacts, + producer=producer(), + security=security(), + created_at=CREATED, + ) + self.assertFalse((root / "pointer.json").exists()) + self.assertFalse(list((root / "releases").iterdir())) + + def test_legacy_truth_history_stays_readable_through_publish_and_swap(self) -> None: + root = self.base / "managed" + original = self.publish(root, "legacy", when=CREATED) + legacy_truth = json.dumps( + {"schema_version": "0.11.0", "marker": "legacy"} + ).encode() + (original.generation_dir / "portfolio-truth.json").write_bytes(legacy_truth) + manifest = json.loads((original.generation_dir / "manifest.json").read_bytes()) + for record in manifest["artifacts"]: + if record["name"] == "portfolio-truth.json": + record["sha256"] = hashlib.sha256(legacy_truth).hexdigest() + record["size_bytes"] = len(legacy_truth) + record["contract_version"] = "ghra.portfolio_truth@0.11.0" + core = { + key: value + for key, value in manifest.items() + if key not in {"generation_id", "content_identity_sha256"} + } + + def canonical_bytes(value: object) -> bytes: + return ( + json.dumps( + value, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ) + + "\n" + ).encode() + + identity = hashlib.sha256(canonical_bytes(core)).hexdigest() + legacy_id = f"portfolio-generation-{identity[:16]}" + manifest["generation_id"] = legacy_id + manifest["content_identity_sha256"] = identity + manifest_bytes = canonical_bytes(manifest) + (original.generation_dir / "manifest.json").write_bytes(manifest_bytes) + legacy_dir = original.generation_dir.with_name(legacy_id) + original.generation_dir.rename(legacy_dir) + pointer = json.loads((root / "pointer.json").read_bytes()) + pointer["current"] = { + "generation_id": legacy_id, + "manifest_sha256": hashlib.sha256(manifest_bytes).hexdigest(), + } + (root / "pointer.json").write_bytes(canonical_bytes(pointer)) + self.assertEqual(resolve_current(root).generation_id, legacy_id) + fresh = self.publish(root, "fresh", when=CREATED + timedelta(minutes=1)) + self.assertEqual(read_pointer(root)["previous"]["generation_id"], legacy_id) + self.assertEqual( + (legacy_dir / "portfolio-truth.json").read_bytes(), legacy_truth + ) + rolled_back = swap_current_previous( + root, updated_at=CREATED + timedelta(minutes=2) + ) + self.assertEqual(rolled_back["current"]["generation_id"], legacy_id) + self.assertEqual(resolve_current(root).generation_id, legacy_id) + rolled_forward = swap_current_previous( + root, updated_at=CREATED + timedelta(minutes=3) + ) + self.assertEqual( + rolled_forward["current"]["generation_id"], fresh.generation_id + ) def test_process_death_never_leaves_an_invalid_pointer(self): steps = ( diff --git a/tests/test_portfolio_generation_identity.py b/tests/test_portfolio_generation_identity.py new file mode 100644 index 0000000..589b00e --- /dev/null +++ b/tests/test_portfolio_generation_identity.py @@ -0,0 +1,505 @@ +"""Publisher CLI regressions using strictly validated, offline truth fixtures.""" + +from __future__ import annotations + +import dataclasses +import hashlib +import json +import os +import subprocess +import sys +import tempfile +import unittest +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any + +from github_repo_auditor import portfolio_generation as generation +from github_repo_auditor.github_security_coverage import ( + DEFAULT_COHORT_POLICY, + collect_security_coverage, + derive_default_attention_cohort, + load_security_coverage_receipt, + write_security_coverage_receipt, +) +from github_repo_auditor.portfolio_cohort_plan import PRODUCER_REPOSITORY +from github_repo_auditor.portfolio_cohort_plan_contract import ( + build_cohort_plan_payload, + validate_cohort_plan, +) +from github_repo_auditor.portfolio_decision_queue import ( + build_decision_digest, + render_decision_digest_markdown, +) +from github_repo_auditor.portfolio_truth_reconcile import build_portfolio_truth_snapshot +from github_repo_auditor.portfolio_truth_validate import ( + _snapshot_from_payload, + validate_truth_snapshot, +) +from github_repo_auditor.producer_preflight import inspect_canonical_producer + +OPERATOR_COMMIT = "b" * 40 +ARTIFACTS = ( + ("portfolio-truth.json", "application/json", "ghra.portfolio_truth@0.12.0"), + ("decision-queue-v2.json", "application/json", "portfolio_decision_digest_v2"), + ("portfolio-digest.md", "text/markdown", "portfolio_digest_markdown_v1"), +) + + +def _write(path: Path, value: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + data = ( + value + if isinstance(value, bytes) + else (json.dumps(value, indent=2) + "\n").encode() + ) + path.write_bytes(data) + path.chmod(0o600) + + +def _stamp(path: Path) -> dict[str, Any]: + return { + "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), + "size": path.stat().st_size, + } + + +def _environment(root: Path) -> dict[str, str]: + return { + "PATH": os.defpath, + "LANG": "C.UTF-8", + "PYTHONPATH": str(Path(generation.__file__).resolve().parents[1]), + "PYTHONDONTWRITEBYTECODE": "1", + "GIT_CONFIG_NOSYSTEM": "1", + "GIT_CONFIG_GLOBAL": os.devnull, + "TMPDIR": str(root / "tmp"), + } + + +def _git(repo: Path, *args: str) -> str: + return subprocess.run( + ["git", "-C", str(repo), *args], + env=_environment(repo.parent), + capture_output=True, + text=True, + check=True, + ).stdout.strip() + + +def _initialize_repo(repo: Path, repository: str) -> str: + repo.mkdir(parents=True, mode=0o700) + _git(repo, "init", "-q", "--initial-branch=main") + _git(repo, "config", "user.name", "Fixture") + _git(repo, "config", "user.email", "fixture@example.invalid") + _git(repo, "remote", "add", "origin", f"https://github.com/{repository}.git") + _write(repo / "README.md", b"# Fixture\n\nOwned offline fixture repository.\n") + _write( + repo / "AGENTS.md", + b"# Fixture Context\n\n## What This Project Is\n\n" + b"An owned offline product fixture for validating portfolio publication.\n\n" + b"## Current State\n\nThe fixture is ready for local contract checks and is actively maintained.\n\n" + b"## Stack\n\nPython and JSON provide the local fixture data and test runner.\n\n" + b"## How To Run\n\nRun the local unit tests with Python. No external credentials are required.\n\n" + b"## Known Risks\n\nProvider coverage remains unknown without credentials; this fixture is not a live audit.\n\n" + b"## Next Recommended Move\n\nVerify the receipt identity and preserve prior published generations.\n", + ) + _git(repo, "add", "README.md", "AGENTS.md") + _git(repo, "commit", "-qm", "Synthetic fixture") + return _git(repo, "rev-parse", "HEAD") + + +class _NoTransport: + def __init__(self) -> None: + self.headers: dict[str, str] = {} + + def get(self, *args: Any, **kwargs: Any) -> None: + raise AssertionError("offline collection attempted a request") + + def post(self, *args: Any, **kwargs: Any) -> None: + raise AssertionError("offline collection attempted a request") + + +def _binding(root: Path, commit: str, now: datetime) -> generation.SecurityBinding: + return generation.load_security_binding( + root / "live/security.json", + terminal_receipt_path=root / "live/terminal.json", + expected_producer_commit=commit, + expected_operator_commit=OPERATOR_COMMIT, + max_age_hours=1, + now=now, + ) + + +def replace_pair(root: Path, label: str) -> None: + for name in ("security.json", "terminal.json"): + staged = root / "live" / f"{name}.new" + _write(staged, (root / "pairs" / label / name).read_bytes()) + os.replace(staged, root / "live" / name) + + +def build_fixture(root: Path, *, clock: datetime | None = None) -> dict[str, Any]: + """Generate canonical truth locally, validate strictly, and retain all bytes.""" + root.mkdir(parents=True, mode=0o700) + (root / "tmp").mkdir(mode=0o700) + now = clock or (datetime.now(UTC).replace(microsecond=0) - timedelta(seconds=20)) + workspace = root / "workspace" + _initialize_repo(workspace / "Alpha", "fixture/alpha") + producer_repo = root / "producer" + commit = _initialize_repo(producer_repo, PRODUCER_REPOSITORY) + preflight = inspect_canonical_producer( + repo_root=producer_repo, + expected_repository=PRODUCER_REPOSITORY, + expected_ref=commit, + checkout_role="canonical", + now=now, + ) + assert preflight.state == "pass" and preflight.evidence is not None + _write(root / "producer.json", preflight.to_dict()) + catalog = root / "catalog.yaml" + _write( + catalog, + b"defaults:\n lifecycle_state: maintenance\n criticality: medium\n" + b" category: vanity\nrepos:\n Alpha:\n lifecycle_state: active\n" + b" owner: fixture\n category: commercial\n tool_provenance: codex\n" + b" operating_path: maintain\n", + ) + registry = root / "empty-registry.md" + _write(registry, b"# Project Registry\n") + candidate = build_portfolio_truth_snapshot( + workspace_root=workspace, + catalog_path=catalog, + legacy_registry_path=registry, + include_notion=False, + repo_status_by_name={}, + now=now, + ).snapshot.to_dict() + cohort = derive_default_attention_cohort(candidate, expected_count=1) + prior_bytes = (json.dumps(candidate, sort_keys=True) + "\n").encode() + _write(root / "prior-truth.json", prior_bytes) + plan_payload = build_cohort_plan_payload( + generated_at=now, + producer_commit=commit, + producer_repository=PRODUCER_REPOSITORY, + catalog_sha256=_stamp(catalog)["sha256"], + workspace_root=str(workspace), + prior_truth_path=str(root / "prior-truth.json"), + prior_truth_sha256=hashlib.sha256(prior_bytes).hexdigest(), + prior_truth_generated_at=now.isoformat(), + policy=DEFAULT_COHORT_POLICY, + prospective_repositories=cohort, + prior_published_repositories=cohort, + ) + plan = validate_cohort_plan(plan_payload) + _write(root / "plan.json", plan_payload) + admissions = {} + for index, label in enumerate(("A", "B")): + produced = now + timedelta(seconds=index) + receipt = collect_security_coverage( + candidate, + token=None, + expected_cohort_count=None, + cohort_plan=plan, + producer_commit=commit, + now=produced, + session=_NoTransport(), + truth_sha256=plan.prior_truth_sha256, + ) + loaded = write_security_coverage_receipt( + receipt, + root / "live/security.json", + expected_cohort_count=None, + ) + terminal = { + "schema": "AutomationTerminalStateV1", + "state": "succeeded", + "completed": True, + "partial": False, + "exit_code": 0, + "automation_id": "com.d.github-security-coverage.rehearsal", + "observed_at": (produced + timedelta(seconds=1)).isoformat(), + "invocation": {"kind": "offline-fixture"}, + "destination_readback": { + "verified": True, + "destination_id": str((root / "live/security.json").resolve()), + "evidence": { + "destination_sha256": loaded.content_sha256, + "producer_commit": commit, + "producer_generation": { + "repository": PRODUCER_REPOSITORY.split("/")[0] + + "/operator-scripts", + "commit": OPERATOR_COMMIT, + }, + "dispatch": {"dispatch_kind": "controlled"}, + }, + }, + } + _write(root / "live/terminal.json", terminal) + admissions[label] = dataclasses.asdict( + _binding(root, commit, now + timedelta(seconds=5)) + ) + for name in ("security.json", "terminal.json"): + _write(root / "pairs" / label / name, (root / "live" / name).read_bytes()) + assert admissions["A"]["receipt_id"] != admissions["B"]["receipt_id"] + assert admissions["A"]["content_sha256"] != admissions["B"]["content_sha256"] + replace_pair(root, "A") + _write(root / "pinned-A.json", (root / "live/security.json").read_bytes()) + security = load_security_coverage_receipt( + root / "pinned-A.json", + expected_cohort_count=None, + expected_producer_commit=commit, + require_cohort_transition=True, + max_age_hours=1, + now=now + timedelta(seconds=5), + ) + metadata = { + "source_id": "github-security-coverage-receipt", + "schema_version": security.schema_version, + "produced_at": security.produced_at, + "state": security.receipt_state, + "age_hours": security.age_hours, + "producer_commit": security.producer_commit, + "cohort_policy": security.cohort_policy, + "cohort_repository_count": len(cohort), + "path": security.source_path, + "receipt_id": security.receipt_id, + "content_sha256": security.content_sha256, + } + validated = [] + for index, marker in enumerate(("seed-old", "seed-current", "trial")): + instant = now + timedelta(seconds=2 + index) + built = build_portfolio_truth_snapshot( + workspace_root=workspace, + catalog_path=catalog, + legacy_registry_path=registry, + include_notion=False, + repo_status_by_name={}, + now=instant, + security_alerts_by_name=security.entries_by_full_name, + security_coverage_metadata=metadata, + security_cohort_required=security.required_repositories, + security_cohort_outgoing=security.outgoing_repositories, + security_cohort_transition=security.transition, + require_cohort_transition=True, + producer=preflight.evidence.to_dict(), + ) + validate_truth_snapshot(built.snapshot, security_max_age_hours=1) + truth = built.snapshot.to_dict() + folder = root / "artifacts" / marker + _write(folder / "portfolio-truth.json", truth) + # Validate what was serialized as well, with ordinary canonical rules. + roundtrip = _snapshot_from_payload( + json.loads((folder / "portfolio-truth.json").read_bytes()) + ) + validate_truth_snapshot(roundtrip, security_max_age_hours=1) + assert roundtrip.to_dict() == truth + assert len(truth["projects"]) == 1 + assert ( + truth["inputs"]["github_security"]["receipt_id"] + == admissions["A"]["receipt_id"] + ) + digest = build_decision_digest(truth) + _write(folder / "decision-queue-v2.json", digest) + _write( + folder / "portfolio-digest.md", + render_decision_digest_markdown(digest).encode(), + ) + validated.append(marker) + config = { + "commit": commit, + "admissions": admissions, + "strictly_validated_truth": validated, + "cohort": list(cohort), + "source_file": str(Path(generation.__file__).resolve()), + } + _write(root / "config.json", config) + return config + + +def worker(root: Path, marker: str, window: str) -> int: + """Run the real CLI body; trace insertion does not replace the final guard.""" + config = json.loads((root / "config.json").read_text()) + args = [ + "publish", + "--root", + str(root / "published"), + "--producer-receipt", + str(root / "producer.json"), + "--producer-repo-root", + str(root / "producer"), + "--expected-repository", + PRODUCER_REPOSITORY, + "--expected-commit", + config["commit"], + "--expected-operator-commit", + OPERATOR_COMMIT, + "--security-receipt", + str(root / "live/security.json"), + "--security-terminal-receipt", + str(root / "live/terminal.json"), + ] + for name, media, contract in ARTIFACTS: + args += [ + "--artifact", + f"{name}={root / 'artifacts' / marker / name}={media}={contract}", + ] + events = [] + if window == "before-entry": + replace_pair(root, "B") + events.append("replaced-before-main") + elif window == "mismatched-terminal": + _write( + root / "live/security.json", (root / "pairs/B/security.json").read_bytes() + ) + events.append("receipt-B-terminal-A-before-main") + switched = False + + def trace(frame: Any, event: str, arg: Any) -> Any: + nonlocal switched + if frame.f_code.co_filename != generation.__file__: + return trace + name = frame.f_code.co_name + if ( + name == "load_security_binding" + and event == "return" + and isinstance(arg, generation.SecurityBinding) + ): + events.append({"admitted": dataclasses.asdict(arg)}) + if name == "pre_pointer_check" and event == "call": + events.append("actual-pre-pointer-check-entry") + if window == "after-entry" and not switched: + replace_pair(root, "B") + switched = True + events.append("replaced-before-actual-guard-body") + if name == "pre_pointer_check" and event == "exception": + events.append({"guard_exception": str(arg[1])}) + return trace + + sys.settrace(trace) + try: + rc = generation.main(args) + finally: + sys.settrace(None) + _write( + root / f"{marker}-trace.json", + {"returncode": rc, "events": events, "argv": args}, + ) + return rc + + +def invoke(root: Path, marker: str, window: str) -> subprocess.CompletedProcess[str]: + outcome = subprocess.run( + [ + sys.executable, + str(Path(__file__).resolve()), + "worker", + str(root), + marker, + window, + ], + env=_environment(root), + capture_output=True, + text=True, + timeout=30, + ) + _write(root / f"{marker}-stdout.txt", outcome.stdout.encode()) + _write(root / f"{marker}-stderr.txt", outcome.stderr.encode()) + return outcome + + +def check_real_cli_valid_truth_identity_and_replacement_windows( + tmp_path: Path, window: str +) -> None: + build_fixture(tmp_path / "case") + root = tmp_path / "case" + for marker in ("seed-old", "seed-current"): + seed = invoke(root, marker, "unchanged") + assert seed.returncode == 0, seed.stderr + published = root / "published" + pointer_bytes = (published / "pointer.json").read_bytes() + before = json.loads(pointer_bytes) + preserved = { + str(p.relative_to(published)): _stamp(p) + for p in (published / "releases").rglob("*") + if p.is_file() + } + inputs = {n: _stamp(root / "artifacts/trial" / n) for n, _, _ in ARTIFACTS} + outcome = invoke(root, "trial", window) + after = generation.read_pointer(published) + for relative, recorded in preserved.items(): + assert _stamp(published / relative) == recorded + for name, recorded in inputs.items(): + assert _stamp(root / "artifacts/trial" / name) == recorded + old_ids = {Path(relative).parts[1] for relative in preserved} + added = {p.name for p in (published / "releases").iterdir()} - old_ids + if window == "unchanged": + assert outcome.returncode == 0, outcome.stderr + assert after["previous"] == before["current"] + assert after["current"] != before["current"] + selected = generation.resolve_current(published).generation_dir + manifest = json.loads((selected / "manifest.json").read_text()) + truth = json.loads((selected / "portfolio-truth.json").read_text()) + validate_truth_snapshot(_snapshot_from_payload(truth), security_max_age_hours=1) + for field in ("receipt_id", "content_sha256"): + assert ( + truth["inputs"]["github_security"][field] + == manifest["github_security"][field] + ) + assert len(added) == 1 + else: + assert outcome.returncode == 1, outcome.stdout + outcome.stderr + assert (published / "pointer.json").read_bytes() == pointer_bytes + if window == "after-entry": + assert ( + "security receipt changed before pointer publication" in outcome.stderr + ) + assert len(added) == 1 + trace = json.loads((root / "trial-trace.json").read_text()) + assert "actual-pre-pointer-check-entry" in trace["events"] + assert "replaced-before-actual-guard-body" in trace["events"] + else: + assert not added + if window == "mismatched-terminal": + assert "security terminal content hash mismatch" in outcome.stderr + else: + assert "identity" in outcome.stderr or "receipt_id" in outcome.stderr + generation.resolve_current(published) + + +class CanonicalTruthPublicationTests(unittest.TestCase): + def run_window(self, window: str) -> None: + with tempfile.TemporaryDirectory( + prefix="valid-truth-publication-", dir=os.environ.get("TMPDIR") + ) as folder: + check_real_cli_valid_truth_identity_and_replacement_windows( + Path(folder), window + ) + + def test_correct_publication(self) -> None: + self.run_window("unchanged") + + def test_replacement_before_admission(self) -> None: + self.run_window("before-entry") + + def test_replacement_after_admission(self) -> None: + self.run_window("after-entry") + + def test_mismatched_terminal(self) -> None: + self.run_window("mismatched-terminal") + + +if __name__ == "__main__": + + def deny_network(event: str, args: Any) -> None: + if event in { + "socket.connect", + "socket.connect_ex", + "socket.getaddrinfo", + "socket.bind", + }: + raise RuntimeError("offline publisher fixture forbids networking") + + sys.addaudithook(deny_network) + if sys.argv[1] == "worker": + raise SystemExit(worker(Path(sys.argv[2]), sys.argv[3], sys.argv[4])) + if sys.argv[1] == "build": + build_fixture(Path(sys.argv[2])) From 6c7c6351e8ce265dafda6b5c9000b2636eae0336 Mon Sep 17 00:00:00 2001 From: saagpatel <269905221+saagpatel@users.noreply.github.com> Date: Sun, 11 Oct 2026 06:55:34 -0700 Subject: [PATCH 2/2] test: isolate synthetic publication companion artifacts --- tests/test_portfolio_generation.py | 11 ++++----- tests/test_portfolio_generation_identity.py | 25 +++++++++++++++------ 2 files changed, 24 insertions(+), 12 deletions(-) diff --git a/tests/test_portfolio_generation.py b/tests/test_portfolio_generation.py index 590c842..3dd973f 100644 --- a/tests/test_portfolio_generation.py +++ b/tests/test_portfolio_generation.py @@ -9,8 +9,6 @@ from pathlib import Path from unittest.mock import patch -import github_repo_auditor.portfolio_generation as generation - from github_repo_auditor.portfolio_generation import ( _terminal_observed_in_order, ArtifactInput, @@ -189,7 +187,9 @@ def test_invalid_truth_identity_preserves_pointer_and_visible_generations( else: identity[field] = value truth_path.write_text(json.dumps(payload)) - with patch.object(generation, "_crash_if_requested") as crash_check: + with patch( + "github_repo_auditor.portfolio_generation._crash_if_requested" + ) as crash_check: with self.assertRaisesRegex(PortfolioGenerationError, message): publish_generation( root=root, @@ -325,8 +325,9 @@ def change_source(step: str) -> None: if step == "after_artifacts_written": truth_path.write_bytes(invalid if stage_valid else valid) - with patch.object( - generation, "_crash_if_requested", side_effect=change_source + with patch( + "github_repo_auditor.portfolio_generation._crash_if_requested", + side_effect=change_source, ): if stage_valid: result = publish_generation( diff --git a/tests/test_portfolio_generation_identity.py b/tests/test_portfolio_generation_identity.py index 589b00e..047e334 100644 --- a/tests/test_portfolio_generation_identity.py +++ b/tests/test_portfolio_generation_identity.py @@ -27,10 +27,6 @@ build_cohort_plan_payload, validate_cohort_plan, ) -from github_repo_auditor.portfolio_decision_queue import ( - build_decision_digest, - render_decision_digest_markdown, -) from github_repo_auditor.portfolio_truth_reconcile import build_portfolio_truth_snapshot from github_repo_auditor.portfolio_truth_validate import ( _snapshot_from_payload, @@ -297,11 +293,26 @@ def build_fixture(root: Path, *, clock: datetime | None = None) -> dict[str, Any truth["inputs"]["github_security"]["receipt_id"] == admissions["A"]["receipt_id"] ) - digest = build_decision_digest(truth) - _write(folder / "decision-queue-v2.json", digest) + # Companion files are explicitly synthetic opaque publication artifacts. + # This test validates canonical truth, identity binding and byte retention; + # decision-digest semantics belong to the digest builder's own tests. + _write( + folder / "decision-queue-v2.json", + { + "contract_version": "portfolio_decision_digest_v2", + "fixture": "offline publication companion", + "source": { + "github_security_receipt_id": admissions["A"]["receipt_id"], + "portfolio_truth_sha256": _stamp(folder / "portfolio-truth.json")[ + "sha256" + ], + }, + "decision_queue": [], + }, + ) _write( folder / "portfolio-digest.md", - render_decision_digest_markdown(digest).encode(), + b"# Offline publication companion\n\nSynthetic artifact fixture.\n", ) validated.append(marker) config = {