From c1608722fd8134f86bc61084a116c8676febf0dc Mon Sep 17 00:00:00 2001 From: saagpatel Date: Fri, 31 Jul 2026 10:37:45 -0700 Subject: [PATCH] fix: bind security receipt publication to collection intent --- src/app/portfolio_truth.py | 33 ++- src/github_security_coverage.py | 313 ++++++++++++++++++++-- src/portfolio_truth_publish.py | 75 +++++- src/portfolio_truth_validate.py | 22 ++ tests/test_github_security_coverage.py | 226 ++++++++++++++++ tests/test_portfolio_truth.py | 344 ++++++++++++++++++++++++- 6 files changed, 972 insertions(+), 41 deletions(-) diff --git a/src/app/portfolio_truth.py b/src/app/portfolio_truth.py index 90ef2be..2c46298 100644 --- a/src/app/portfolio_truth.py +++ b/src/app/portfolio_truth.py @@ -5,7 +5,11 @@ from typing import Any from src.cli_output import print_info -from src.github_security_coverage import DEFAULT_EXPECTED_GITHUB_COHORT_COUNT +from src.github_security_coverage import ( + DEFAULT_EXPECTED_GITHUB_COHORT_COUNT, + SecurityCoverageError, + SecurityCoverageReceiptBinding, +) from src.portfolio_context_recovery import ( apply_context_recovery_plan, build_context_recovery_plan, @@ -46,6 +50,9 @@ def run_portfolio_truth_mode(args: Any) -> None: producer_repo_root = ( Path(producer_repo_root_value) if producer_repo_root_value else None ) + require_producer_evidence = bool( + os.environ.get("GHRA_REQUIRE_PRODUCER_EVIDENCE", "1") == "1" + ) release_count_by_name: dict[str, int] | None = None if getattr(args, "portfolio_truth_include_release_count", False): release_count_by_name = load_release_count_by_name( @@ -54,6 +61,7 @@ def run_portfolio_truth_mode(args: Any) -> None: ) security_alerts_by_name: dict[str, dict] | None = None security_coverage_metadata: dict[str, object] | None = None + security_receipt_binding: SecurityCoverageReceiptBinding | None = None if getattr(args, "portfolio_truth_include_security", False): receipt_path_value = getattr(args, "portfolio_truth_security_receipt", None) loaded_security = load_security_coverage_by_full_name( @@ -72,6 +80,13 @@ def run_portfolio_truth_mode(args: Any) -> None: ), ) if loaded_security is not None: + try: + security_receipt_binding = loaded_security.binding() + except SecurityCoverageError as exc: + if require_producer_evidence: + raise SystemExit( + f"Canonical PortfolioTruth security publication refused: {exc}" + ) from exc security_alerts_by_name = loaded_security.entries_by_full_name security_coverage_metadata = { "source_id": "github-security-coverage-receipt", @@ -86,6 +101,17 @@ def run_portfolio_truth_mode(args: Any) -> None: ), "path": loaded_security.source_path, } + if loaded_security.receipt_id is not None: + security_coverage_metadata["receipt_id"] = loaded_security.receipt_id + if loaded_security.content_sha256 is not None: + security_coverage_metadata["content_sha256"] = ( + loaded_security.content_sha256 + ) + elif require_producer_evidence: + raise SystemExit( + "Canonical PortfolioTruth security publication requires a valid " + "identity-bound GitHub security receipt." + ) repo_status_by_name = load_live_repo_status_by_name( username=args.username, token=getattr(args, "token", None), @@ -112,12 +138,11 @@ def run_portfolio_truth_mode(args: Any) -> None: release_count_by_name=release_count_by_name, security_alerts_by_name=security_alerts_by_name, security_coverage_metadata=security_coverage_metadata, + security_receipt_binding=security_receipt_binding, repo_status_by_name=repo_status_by_name, producer_evidence=producer_evidence, producer_repo_root=producer_repo_root, - require_producer_evidence=bool( - os.environ.get("GHRA_REQUIRE_PRODUCER_EVIDENCE", "1") == "1" - ), + require_producer_evidence=require_producer_evidence, ) except (PortfolioTruthPublishError, ValueError) as exc: raise SystemExit(str(exc)) from exc diff --git a/src/github_security_coverage.py b/src/github_security_coverage.py index 1defb59..184de98 100644 --- a/src/github_security_coverage.py +++ b/src/github_security_coverage.py @@ -8,14 +8,18 @@ from __future__ import annotations import argparse +import fcntl +import hashlib import json import os import re import subprocess +import tempfile +from contextlib import contextmanager from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path -from typing import Any +from typing import Any, Iterator import requests @@ -76,6 +80,8 @@ DEFAULT_QUOTA_RESERVE = 100 _COMMIT_RE = re.compile(r"^[0-9a-f]{40}$") _REPOSITORY_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") +_RECEIPT_ID_RE = re.compile(r"^sha256:[0-9a-f]{64}$") +_SHA256_RE = re.compile(r"^[0-9a-f]{64}$") _ENDPOINTS = { "dependabot": "dependabot/alerts", @@ -102,6 +108,19 @@ class SecurityCoverageError(ValueError): """Raised when a coverage receipt or bounded collection contract is invalid.""" +@dataclass(frozen=True) +class SecurityCoverageReceiptBinding: + """Immutable identity needed to authorize a PortfolioTruth publication.""" + + source_path: str + receipt_id: str + content_sha256: str + receipt_state: str + max_age_hours: int + expected_cohort_count: int + expected_producer_commit: str | None + + @dataclass(frozen=True) class LoadedSecurityCoverage: entries_by_full_name: dict[str, dict[str, Any]] @@ -113,6 +132,64 @@ class LoadedSecurityCoverage: receipt_state: str age_hours: float source_path: str + receipt_id: str | None = None + content_sha256: str | None = None + max_age_hours: int = 24 + expected_cohort_count: int = DEFAULT_EXPECTED_GITHUB_COHORT_COUNT + expected_producer_commit: str | None = None + + def binding(self) -> SecurityCoverageReceiptBinding: + """Return the strict byte binding used by canonical publication. + + Legacy V1 receipts remain readable, but they cannot authorize a new + canonical PortfolioTruth publication because they do not identify the + exact semantic receipt or the exact file bytes that were consumed. + """ + if not self.receipt_id: + raise SecurityCoverageError( + "security coverage receipt is missing immutable receipt_id provenance" + ) + if not self.content_sha256: + raise SecurityCoverageError( + "security coverage receipt is missing an exact content digest" + ) + if not self.source_path: + raise SecurityCoverageError( + "security coverage receipt is missing its canonical source path" + ) + return SecurityCoverageReceiptBinding( + source_path=self.source_path, + receipt_id=self.receipt_id, + content_sha256=self.content_sha256, + receipt_state=self.receipt_state, + max_age_hours=self.max_age_hours, + expected_cohort_count=self.expected_cohort_count, + expected_producer_commit=self.expected_producer_commit, + ) + + +@dataclass(frozen=True) +class SecurityCoverageReceiptWriter: + """Exclusive writer intent for one canonical security receipt. + + The lock is intentionally held before the prior receipt is read and remains + held until the replacement bytes land. That gives PortfolioTruth a + concrete signal that a newer same-cycle receipt is in flight even before + the canonical pointer has changed. + """ + + path: Path + expected_cohort_count: int + + def load_prior(self) -> dict[str, Any] | None: + return _load_json_object(self.path) if self.path.is_file() else None + + def write(self, payload: dict[str, Any]) -> LoadedSecurityCoverage: + return _write_security_coverage_receipt_unlocked( + payload, + self.path, + expected_cohort_count=self.expected_cohort_count, + ) @dataclass @@ -159,6 +236,83 @@ def _mapping(value: Any) -> dict[str, Any]: return value if isinstance(value, dict) else {} +def _receipt_id_for_payload(payload: dict[str, Any]) -> str: + """Return a deterministic semantic identity for a receipt payload.""" + unsigned = {key: value for key, value in payload.items() if key != "receipt_id"} + canonical = json.dumps( + unsigned, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + return f"sha256:{hashlib.sha256(canonical).hexdigest()}" + + +def _with_receipt_id(payload: dict[str, Any]) -> dict[str, Any]: + bound = dict(payload) + declared = bound.get("receipt_id") + expected = _receipt_id_for_payload(bound) + if declared is None: + bound["receipt_id"] = expected + elif declared != expected: + raise SecurityCoverageError( + "security coverage receipt_id does not match the receipt payload" + ) + return bound + + +def _receipt_lock_path(path: Path) -> Path: + return path.with_name(f".{path.name}.lock") + + +@contextmanager +def _receipt_lock( + path: Path, + *, + exclusive: bool, + create: bool, + blocking: bool = True, +) -> Iterator[None]: + lock_path = _receipt_lock_path(path) + flags = os.O_RDWR | (os.O_CREAT if create else 0) + try: + descriptor = os.open(lock_path, flags, 0o600) + except FileNotFoundError as exc: + raise SecurityCoverageError( + f"security coverage receipt lock is missing: {lock_path}" + ) from exc + try: + operation = fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH + if not blocking: + operation |= fcntl.LOCK_NB + try: + fcntl.flock(descriptor, operation) + except BlockingIOError as exc: + raise SecurityCoverageError( + "security coverage receipt collection is active; retry " + "PortfolioTruth publication after the current receipt lands" + ) from exc + yield + finally: + fcntl.flock(descriptor, fcntl.LOCK_UN) + os.close(descriptor) + + +@contextmanager +def security_coverage_receipt_writer( + path: Path, + *, + expected_cohort_count: int = DEFAULT_EXPECTED_GITHUB_COHORT_COUNT, +) -> Iterator[SecurityCoverageReceiptWriter]: + """Hold exclusive writer intent across collection and receipt replacement.""" + path.parent.mkdir(parents=True, exist_ok=True) + with _receipt_lock(path, exclusive=True, create=True): + yield SecurityCoverageReceiptWriter( + path=path, + expected_cohort_count=expected_cohort_count, + ) + + def _canonical_repo(value: Any) -> str: full_name = _text(value) if not _REPOSITORY_RE.fullmatch(full_name): @@ -1616,6 +1770,7 @@ def validate_security_coverage_receipt( expected_producer_commit: str | None = None, now: datetime | None = None, source_path: str = "", + content_sha256: str | None = None, ) -> LoadedSecurityCoverage: """Validate provenance/freshness and return normalized full-name entries.""" if max_age_hours <= 0: @@ -1627,6 +1782,22 @@ def validate_security_coverage_receipt( "unexpected security coverage receipt schema: " f"{payload.get('schema_version')!r}" ) + receipt_id_value = payload.get("receipt_id") + receipt_id: str | None = None + if receipt_id_value is not None: + receipt_id = _text(receipt_id_value) + if not _RECEIPT_ID_RE.fullmatch(receipt_id): + raise SecurityCoverageError( + "security coverage receipt_id must be a sha256 identity" + ) + if receipt_id != _receipt_id_for_payload(payload): + raise SecurityCoverageError( + "security coverage receipt_id does not match the receipt payload" + ) + if content_sha256 is not None and not _SHA256_RE.fullmatch(content_sha256): + raise SecurityCoverageError( + "security coverage content_sha256 must be a lowercase SHA-256 digest" + ) produced_at = _parse_datetime(payload.get("produced_at"), field_name="produced_at") current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) age_hours = (current - produced_at).total_seconds() / 3600 @@ -1752,6 +1923,11 @@ def validate_security_coverage_receipt( receipt_state="stale" if receipt_is_stale else "fresh", age_hours=round(age_hours, 3), source_path=source_path, + receipt_id=receipt_id, + content_sha256=content_sha256, + max_age_hours=max_age_hours, + expected_cohort_count=expected_cohort_count, + expected_producer_commit=expected_producer_commit, ) @@ -1762,9 +1938,27 @@ def load_security_coverage_receipt( expected_cohort_count: int = DEFAULT_EXPECTED_GITHUB_COHORT_COUNT, expected_producer_commit: str | None = None, now: datetime | None = None, +) -> LoadedSecurityCoverage: + return _load_security_coverage_receipt_unlocked( + path, + max_age_hours=max_age_hours, + expected_cohort_count=expected_cohort_count, + expected_producer_commit=expected_producer_commit, + now=now, + ) + + +def _load_security_coverage_receipt_unlocked( + path: Path, + *, + max_age_hours: int, + expected_cohort_count: int, + expected_producer_commit: str | None, + now: datetime | None, ) -> LoadedSecurityCoverage: try: - payload = json.loads(path.read_text()) + raw_bytes = path.read_bytes() + payload = json.loads(raw_bytes) except FileNotFoundError as exc: raise SecurityCoverageError( f"security coverage receipt not found: {path}" @@ -1780,25 +1974,94 @@ def load_security_coverage_receipt( expected_producer_commit=expected_producer_commit, now=now, source_path=str(path), + content_sha256=hashlib.sha256(raw_bytes).hexdigest(), ) +@contextmanager +def verified_security_coverage_receipt_binding( + binding: SecurityCoverageReceiptBinding, +) -> Iterator[LoadedSecurityCoverage]: + """Hold the collector lock while revalidating the bound receipt bytes.""" + path = Path(binding.source_path) + with _receipt_lock(path, exclusive=False, create=False, blocking=False): + loaded = _load_security_coverage_receipt_unlocked( + path, + max_age_hours=binding.max_age_hours, + expected_cohort_count=binding.expected_cohort_count, + expected_producer_commit=binding.expected_producer_commit, + now=None, + ) + if loaded.receipt_id != binding.receipt_id: + raise SecurityCoverageError( + "security coverage receipt identity changed after it was loaded" + ) + if loaded.content_sha256 != binding.content_sha256: + raise SecurityCoverageError( + "security coverage receipt bytes changed after they were loaded" + ) + if loaded.receipt_state != binding.receipt_state: + raise SecurityCoverageError( + "security coverage receipt freshness changed after it was loaded" + ) + yield loaded + + def write_security_coverage_receipt( payload: dict[str, Any], path: Path, *, expected_cohort_count: int = DEFAULT_EXPECTED_GITHUB_COHORT_COUNT, -) -> None: - """Atomically write a validated receipt payload.""" - validate_security_coverage_receipt( - payload, +) -> LoadedSecurityCoverage: + """Atomically write a validated, identity-bound receipt payload.""" + path.parent.mkdir(parents=True, exist_ok=True) + with _receipt_lock(path, exclusive=True, create=True): + return _write_security_coverage_receipt_unlocked( + payload, + path, + expected_cohort_count=expected_cohort_count, + ) + + +def _write_security_coverage_receipt_unlocked( + payload: dict[str, Any], + path: Path, + *, + expected_cohort_count: int, +) -> LoadedSecurityCoverage: + """Write a receipt while the caller already holds exclusive writer intent.""" + bound_payload = _with_receipt_id(payload) + serialized = ( + json.dumps(bound_payload, indent=2, sort_keys=True, ensure_ascii=False) + "\n" + ).encode("utf-8") + content_sha256 = hashlib.sha256(serialized).hexdigest() + loaded = validate_security_coverage_receipt( + bound_payload, max_age_hours=24 * 365, expected_cohort_count=expected_cohort_count, + source_path=str(path), + content_sha256=content_sha256, ) path.parent.mkdir(parents=True, exist_ok=True) - temporary = path.with_name(f".{path.name}.tmp") - temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") - temporary.replace(path) + temporary_path: Path | None = None + try: + with tempfile.NamedTemporaryFile( + "wb", + delete=False, + dir=path.parent, + prefix=f".{path.name}.", + suffix=".tmp", + ) as handle: + temporary_path = Path(handle.name) + handle.write(serialized) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary_path, path) + temporary_path = None + finally: + if temporary_path is not None: + temporary_path.unlink(missing_ok=True) + return loaded def _load_json_object(path: Path) -> dict[str, Any]: @@ -1873,27 +2136,29 @@ def main() -> None: "receipt_state": loaded.receipt_state, "age_hours": loaded.age_hours, "cohort_count": len(loaded.cohort_repositories), + "receipt_id": loaded.receipt_id, + "content_sha256": loaded.content_sha256, }, indent=2, ) ) return - truth = _load_json_object(args.truth) - prior = _load_json_object(args.output) if args.output.is_file() else None - receipt = collect_security_coverage( - truth, - token=os.environ.get("GITHUB_TOKEN"), - expected_cohort_count=args.expected_cohort_count, - base_request_limit=args.base_request_limit, - total_request_limit=args.total_request_limit, - quota_reserve=args.quota_reserve, - prior_receipt=prior, - ) - write_security_coverage_receipt( - receipt, + with security_coverage_receipt_writer( args.output, expected_cohort_count=args.expected_cohort_count, - ) + ) as writer: + truth = _load_json_object(args.truth) + prior = writer.load_prior() + receipt = collect_security_coverage( + truth, + token=os.environ.get("GITHUB_TOKEN"), + expected_cohort_count=args.expected_cohort_count, + base_request_limit=args.base_request_limit, + total_request_limit=args.total_request_limit, + quota_reserve=args.quota_reserve, + prior_receipt=prior, + ) + written = writer.write(receipt) except SecurityCoverageError as exc: raise SystemExit(str(exc)) from exc print( @@ -1903,6 +2168,8 @@ def main() -> None: "path": str(args.output), "cohort_count": receipt["cohort"]["repository_count"], "request_budget": receipt["request_budget"], + "receipt_id": written.receipt_id, + "content_sha256": written.content_sha256, }, indent=2, ) diff --git a/src/portfolio_truth_publish.py b/src/portfolio_truth_publish.py index 4cd1b50..9b7f66c 100644 --- a/src/portfolio_truth_publish.py +++ b/src/portfolio_truth_publish.py @@ -2,9 +2,15 @@ import json import tempfile +from contextlib import nullcontext from dataclasses import dataclass from pathlib import Path +from src.github_security_coverage import ( + SecurityCoverageError, + SecurityCoverageReceiptBinding, + verified_security_coverage_receipt_binding, +) from src.portfolio_truth_reconcile import ( build_portfolio_truth_snapshot, load_prior_notion_context, @@ -91,6 +97,7 @@ def publish_portfolio_truth( release_count_by_name: dict[str, int] | None = None, security_alerts_by_name: dict[str, dict] | None = None, security_coverage_metadata: dict[str, object] | None = None, + security_receipt_binding: SecurityCoverageReceiptBinding | None = None, repo_status_by_name: dict[str, dict] | None = None, producer_evidence: ProducerEvidence | None = None, producer_repo_root: Path | None = None, @@ -109,6 +116,15 @@ def publish_portfolio_truth( verify_evidence_still_current(producer_repo_root, producer_evidence) except ValueError as exc: raise PortfolioTruthPublishError(str(exc)) from exc + _validate_security_receipt_binding( + metadata=security_coverage_metadata, + binding=security_receipt_binding, + required=require_producer_evidence + and ( + security_alerts_by_name is not None + or security_coverage_metadata is not None + ), + ) validate_publish_targets( workspace_root=workspace_root, output_dir=output_dir, @@ -134,11 +150,6 @@ def publish_portfolio_truth( prior_notion_generated_at=prior_notion_generated_at, ) validate_truth_snapshot(build_result.snapshot) - if producer_evidence is not None and producer_repo_root is not None: - try: - verify_evidence_still_current(producer_repo_root, producer_evidence) - except ValueError as exc: - raise PortfolioTruthPublishError(str(exc)) from exc snapshot_stamp = build_result.snapshot.generated_at.strftime("%Y-%m-%dT%H%M%SZ") snapshot_path = output_dir / f"portfolio-truth-{snapshot_stamp}.json" @@ -186,14 +197,22 @@ def publish_portfolio_truth( originals = {path: (path.read_text() if path.exists() else None) for path in targets} published: list[Path] = [] + publication_guard = ( + verified_security_coverage_receipt_binding(security_receipt_binding) + if security_receipt_binding is not None + else nullcontext() + ) try: - for path, staged in temp_files.items(): - if path in {registry_output, portfolio_report_output} and not changed[path]: - staged.unlink(missing_ok=True) - continue - staged.replace(path) - published.append(path) - except Exception: + with publication_guard: + if producer_evidence is not None and producer_repo_root is not None: + verify_evidence_still_current(producer_repo_root, producer_evidence) + for path, staged in temp_files.items(): + if path in {registry_output, portfolio_report_output} and not changed[path]: + staged.unlink(missing_ok=True) + continue + staged.replace(path) + published.append(path) + except Exception as exc: for path in reversed(published): original = originals[path] if original is None: @@ -202,6 +221,8 @@ def publish_portfolio_truth( path.write_text(original) for staged in temp_files.values(): staged.unlink(missing_ok=True) + if isinstance(exc, (SecurityCoverageError, ValueError)): + raise PortfolioTruthPublishError(str(exc)) from exc raise for staged in temp_files.values(): @@ -219,6 +240,36 @@ def publish_portfolio_truth( ) +def _validate_security_receipt_binding( + *, + metadata: dict[str, object] | None, + binding: SecurityCoverageReceiptBinding | None, + required: bool, +) -> None: + if binding is None: + if required: + raise PortfolioTruthPublishError( + "Canonical security publication requires immutable receipt identity." + ) + return + if metadata is None: + raise PortfolioTruthPublishError( + "Security receipt binding requires PortfolioTruth input metadata." + ) + if metadata.get("receipt_id") != binding.receipt_id: + raise PortfolioTruthPublishError( + "Security receipt_id metadata does not match the bound receipt." + ) + if metadata.get("content_sha256") != binding.content_sha256: + raise PortfolioTruthPublishError( + "Security content_sha256 metadata does not match the bound receipt bytes." + ) + if metadata.get("path") != binding.source_path: + raise PortfolioTruthPublishError( + "Security receipt path metadata does not match the bound receipt." + ) + + def _stage_text(target: Path, content: str) -> Path: target.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile( diff --git a/src/portfolio_truth_validate.py b/src/portfolio_truth_validate.py index d539a58..e192f41 100644 --- a/src/portfolio_truth_validate.py +++ b/src/portfolio_truth_validate.py @@ -1,5 +1,6 @@ from __future__ import annotations +import re from pathlib import Path from src.portfolio_pathing import ( @@ -145,6 +146,27 @@ def _validate_contract_envelope(snapshot: PortfolioTruthSnapshot) -> None: raise ValueError("Live Notion input cannot declare a carried-forward origin.") if mode == "verified-snapshot" and not notion.get("observed_at"): raise ValueError("Verified Notion snapshot input requires an observation timestamp.") + github_security = ( + snapshot.inputs.get("github_security") + if isinstance(snapshot.inputs, dict) + else None + ) + if isinstance(github_security, dict): + receipt_id = github_security.get("receipt_id") + content_sha256 = github_security.get("content_sha256") + if (receipt_id is None) != (content_sha256 is None): + raise ValueError( + "GitHub security input identity requires both receipt_id and " + "content_sha256." + ) + if receipt_id is not None and not re.fullmatch( + r"sha256:[0-9a-f]{64}", str(receipt_id) + ): + raise ValueError("GitHub security receipt_id is malformed.") + if content_sha256 is not None and not re.fullmatch( + r"[0-9a-f]{64}", str(content_sha256) + ): + raise ValueError("GitHub security content_sha256 is malformed.") def validate_publish_targets( diff --git a/tests/test_github_security_coverage.py b/tests/test_github_security_coverage.py index df6b17b..818a152 100644 --- a/tests/test_github_security_coverage.py +++ b/tests/test_github_security_coverage.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import subprocess import sys from datetime import datetime, timedelta, timezone from pathlib import Path @@ -17,7 +18,10 @@ derive_default_attention_cohort, load_security_coverage_receipt, main, + security_coverage_receipt_writer, validate_security_coverage_receipt, + verified_security_coverage_receipt_binding, + write_security_coverage_receipt, ) from src.portfolio_truth_reconcile import _select_security_entry from src.portfolio_truth_status import load_security_coverage_by_full_name @@ -106,6 +110,42 @@ def _collect( ) +def _assert_binding_revalidation_fails_in_child( + binding: Any, + *, + expected: str, +) -> None: + script = """ +import json +import sys + +from src.github_security_coverage import ( + SecurityCoverageError, + SecurityCoverageReceiptBinding, + verified_security_coverage_receipt_binding, +) + +binding = SecurityCoverageReceiptBinding(**json.loads(sys.stdin.read())) +try: + with verified_security_coverage_receipt_binding(binding): + pass +except SecurityCoverageError as exc: + print(str(exc)) + raise SystemExit(0) +raise SystemExit("binding unexpectedly revalidated") +""" + result = subprocess.run( + [sys.executable, "-c", script], + input=json.dumps(binding.__dict__), + text=True, + capture_output=True, + cwd=Path(__file__).resolve().parents[1], + check=False, + ) + assert result.returncode == 0, result.stderr or result.stdout + assert expected in result.stdout + + def _prior_with_private_unavailable(count: int = 6) -> dict[str, Any]: prior = _collect() for index in range(count): @@ -688,6 +728,192 @@ def test_receipt_loader_uses_embedded_provenance_not_newer_mtime( ) +def test_written_receipt_binds_semantic_identity_and_exact_bytes( + tmp_path: Path, +) -> None: + receipt = _collect(cohort_count=DEFAULT_EXPECTED_GITHUB_COHORT_COUNT) + canonical = tmp_path / GITHUB_SECURITY_RECEIPT_FILENAME + + written = write_security_coverage_receipt(receipt, canonical) + loaded = load_security_coverage_receipt(canonical, now=NOW) + + assert written.receipt_id == loaded.receipt_id + assert written.content_sha256 == loaded.content_sha256 + assert written.receipt_id is not None + assert written.receipt_id.startswith("sha256:") + assert written.content_sha256 is not None + assert len(written.content_sha256) == 64 + assert json.loads(canonical.read_text())["receipt_id"] == written.receipt_id + assert canonical.with_name(f".{canonical.name}.lock").is_file() + + +def test_collector_cli_holds_writer_intent_before_truth_read_and_replacement( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + truth_path = tmp_path / "portfolio-truth-latest.json" + receipt_path = tmp_path / GITHUB_SECURITY_RECEIPT_FILENAME + truth = _truth(DEFAULT_EXPECTED_GITHUB_COHORT_COUNT) + truth_path.write_text(json.dumps(truth)) + first = _collect(cohort_count=DEFAULT_EXPECTED_GITHUB_COHORT_COUNT) + write_security_coverage_receipt(first, receipt_path) + binding = load_security_coverage_receipt(receipt_path, now=NOW).binding() + second = collect_security_coverage( + truth, + token="opaque-test-token", + expected_cohort_count=DEFAULT_EXPECTED_GITHUB_COHORT_COUNT, + session=_Session(), + now=NOW + timedelta(minutes=1), + producer_commit="a" * 40, + api_base_url="https://api.example.test", + ) + observed: dict[str, object] = {} + from src import github_security_coverage as coverage_module + + original_load_json_object = coverage_module._load_json_object + + def load_json_with_truth_read_probe(path: Path) -> dict[str, Any]: + if path == truth_path: + _assert_binding_revalidation_fails_in_child( + binding, + expected="security coverage receipt collection is active", + ) + observed["publisher_blocked_during_truth_read"] = True + return original_load_json_object(path) + + def fake_collect_security_coverage( + _truth_payload: dict[str, Any], + **kwargs: Any, + ) -> dict[str, Any]: + prior_receipt = kwargs["prior_receipt"] + assert prior_receipt is not None + observed["prior_receipt_id"] = prior_receipt["receipt_id"] + _assert_binding_revalidation_fails_in_child( + binding, + expected="security coverage receipt collection is active", + ) + observed["publisher_blocked_during_collection"] = True + return second + + monkeypatch.setattr( + "src.github_security_coverage._load_json_object", + load_json_with_truth_read_probe, + ) + monkeypatch.setattr( + "src.github_security_coverage.collect_security_coverage", + fake_collect_security_coverage, + ) + monkeypatch.setattr( + sys, + "argv", + [ + "github-security-coverage", + "--truth", + str(truth_path), + "--output", + str(receipt_path), + "--expected-cohort-count", + str(DEFAULT_EXPECTED_GITHUB_COHORT_COUNT), + ], + ) + + main() + loaded = load_security_coverage_receipt(receipt_path, now=NOW + timedelta(minutes=1)) + + assert observed == { + "publisher_blocked_during_truth_read": True, + "prior_receipt_id": binding.receipt_id, + "publisher_blocked_during_collection": True, + } + assert loaded.receipt_id != binding.receipt_id + + +def test_writer_intent_interruption_does_not_leave_stale_lock( + tmp_path: Path, +) -> None: + receipt = _collect(cohort_count=DEFAULT_EXPECTED_GITHUB_COHORT_COUNT) + canonical = tmp_path / GITHUB_SECURITY_RECEIPT_FILENAME + write_security_coverage_receipt(receipt, canonical) + binding = load_security_coverage_receipt( + canonical, + max_age_hours=24 * 365, + now=NOW, + ).binding() + + with pytest.raises(RuntimeError, match="collector interrupted"): + with security_coverage_receipt_writer(canonical): + raise RuntimeError("collector interrupted") + + assert canonical.with_name(f".{canonical.name}.lock").is_file() + with verified_security_coverage_receipt_binding(binding) as loaded: + assert loaded.receipt_id == binding.receipt_id + + +def test_replacement_after_load_fails_bound_revalidation(tmp_path: Path) -> None: + canonical = tmp_path / GITHUB_SECURITY_RECEIPT_FILENAME + first = _collect(cohort_count=DEFAULT_EXPECTED_GITHUB_COHORT_COUNT) + write_security_coverage_receipt(first, canonical) + binding = load_security_coverage_receipt(canonical, now=NOW).binding() + + second = collect_security_coverage( + _truth(DEFAULT_EXPECTED_GITHUB_COHORT_COUNT), + token="opaque-test-token", + expected_cohort_count=DEFAULT_EXPECTED_GITHUB_COHORT_COUNT, + session=_Session(), + now=NOW + timedelta(minutes=1), + producer_commit="a" * 40, + api_base_url="https://api.example.test", + ) + write_security_coverage_receipt(second, canonical) + + with pytest.raises(SecurityCoverageError, match="changed after it was loaded"): + with verified_security_coverage_receipt_binding(binding): + pass + + +def test_byte_change_with_same_receipt_id_fails_bound_revalidation( + tmp_path: Path, +) -> None: + canonical = tmp_path / GITHUB_SECURITY_RECEIPT_FILENAME + write_security_coverage_receipt( + _collect(cohort_count=DEFAULT_EXPECTED_GITHUB_COHORT_COUNT), canonical + ) + binding = load_security_coverage_receipt(canonical, now=NOW).binding() + payload = json.loads(canonical.read_text()) + canonical.write_text(json.dumps(payload, separators=(",", ":"))) + + with pytest.raises(SecurityCoverageError, match="bytes changed"): + with verified_security_coverage_receipt_binding(binding): + pass + + +def test_legacy_receipt_remains_readable_but_cannot_authorize_publication( + tmp_path: Path, +) -> None: + legacy = _collect(cohort_count=DEFAULT_EXPECTED_GITHUB_COHORT_COUNT) + canonical = tmp_path / GITHUB_SECURITY_RECEIPT_FILENAME + canonical.write_text(json.dumps(legacy)) + + loaded = load_security_coverage_receipt(canonical, now=NOW) + + assert loaded.receipt_id is None + assert loaded.content_sha256 is not None + with pytest.raises(SecurityCoverageError, match="missing immutable receipt_id"): + loaded.binding() + + +def test_malformed_or_mismatched_receipt_identity_fails_closed() -> None: + receipt = _collect() + receipt["receipt_id"] = "sha256:" + "0" * 64 + + with pytest.raises(SecurityCoverageError, match="does not match"): + validate_security_coverage_receipt( + receipt, + expected_cohort_count=16, + now=NOW, + ) + + def test_receipt_loader_honors_explicit_nondefault_cohort_count( tmp_path: Path, ) -> None: diff --git a/tests/test_portfolio_truth.py b/tests/test_portfolio_truth.py index c567976..dce179e 100644 --- a/tests/test_portfolio_truth.py +++ b/tests/test_portfolio_truth.py @@ -5,12 +5,18 @@ import os import subprocess import time -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from pathlib import Path import pytest from src.cli import main +from src.github_security_coverage import ( + collect_security_coverage, + load_security_coverage_receipt, + write_security_coverage_receipt, +) +from src.portfolio_decision_queue import build_decision_queue from src.portfolio_context_recovery import ( apply_context_recovery_plan, build_context_recovery_plan, @@ -30,7 +36,10 @@ _git_remote_full_name, load_safe_notion_project_context, ) -from src.portfolio_truth_validate import validate_portfolio_report_markdown +from src.portfolio_truth_validate import ( + validate_portfolio_report_markdown, + validate_truth_snapshot, +) from src.project_registry import build_project_registry from src.registry_parser import parse_registry @@ -1449,6 +1458,106 @@ def test_security_overlay_populates_and_force_elevates( assert alpha_dict["security"]["dependabot_critical"] == 1 +def test_bound_security_identity_and_high_findings_reach_decision_queue( + portfolio_workspace: Path, + portfolio_catalog: Path, + legacy_registry: Path, +) -> None: + now = datetime.fromtimestamp(1_700_200_000, tz=timezone.utc) + alpha_path = portfolio_workspace / "Alpha" + subprocess.run(["git", "init"], cwd=alpha_path, capture_output=True, check=True) + subprocess.run( + ["git", "remote", "add", "origin", "https://github.com/d/Alpha.git"], + cwd=alpha_path, + capture_output=True, + check=True, + ) + observed_at = now.isoformat() + security = { + "d/Alpha": { + "repo_full_name": "d/Alpha", + "cohort_member": True, + "cohort_policy": "portfolio-default-attention-v1", + "receipt_schema_version": "GitHubSecurityCoverageReceiptV1", + "receipt_state": "fresh", + "source_produced_at": observed_at, + "repository": { + "source": "github-graphql-default-branch-head-v1", + "state": "observed", + "reason_code": "observed", + "reason": None, + "observed_at": observed_at, + "default_branch": "main", + "head_sha": "b" * 40, + "archived": False, + }, + "providers": { + "dependabot": { + "state": "observed", + "observed_at": observed_at, + "pagination_complete": True, + "counts": {"critical": 0, "high": 2, "medium": 1, "low": 0}, + }, + "code_scanning": { + "state": "observed", + "observed_at": observed_at, + "pagination_complete": True, + "counts": {"critical": 0, "high": 0, "warning": 0, "note": 0}, + }, + "secret_scanning": { + "state": "observed", + "observed_at": observed_at, + "pagination_complete": True, + "counts": {"open": 0}, + }, + }, + } + } + metadata = { + "source_id": "github-security-coverage-receipt", + "schema_version": "GitHubSecurityCoverageReceiptV1", + "produced_at": observed_at, + "state": "fresh", + "age_hours": 0.0, + "producer_commit": "a" * 40, + "cohort_policy": "portfolio-default-attention-v1", + "cohort_repository_count": 1, + "path": "/evidence/github-security-coverage-latest.json", + "receipt_id": "sha256:" + "c" * 64, + "content_sha256": "d" * 64, + } + + result = build_portfolio_truth_snapshot( + workspace_root=portfolio_workspace, + catalog_path=portfolio_catalog, + legacy_registry_path=legacy_registry, + include_notion=False, + now=now, + security_alerts_by_name=security, + security_coverage_metadata=metadata, + ) + truth = result.snapshot.to_dict() + alpha = next( + project + for project in truth["projects"] + if project["identity"]["display_name"] == "Alpha" + ) + decision = next( + item for item in build_decision_queue(truth) if item["project"] == "Alpha" + ) + + assert truth["inputs"]["github_security"] == metadata + assert alpha["risk"]["security_risk"] is True + assert alpha["security"]["dependabot_high"] == 2 + assert alpha["security"]["dependabot_medium"] == 1 + assert decision["decision_type"] == "security follow-up" + assert "critical=0, high=2" in decision["evidence"][0] + + result.snapshot.inputs["github_security"].pop("content_sha256") + with pytest.raises(ValueError, match="requires both receipt_id"): + validate_truth_snapshot(result.snapshot) + + def test_security_overlay_absent_leaves_repos_unscanned( portfolio_workspace: Path, portfolio_catalog: Path, @@ -2364,6 +2473,112 @@ def _boom(_snapshot, _latest_json_path): assert not list(output_dir.glob("*.tmp")) +def test_publish_refuses_receipt_pointer_replacement_after_load( + portfolio_workspace: Path, + portfolio_catalog: Path, + legacy_registry: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + now = datetime.now(timezone.utc).replace(microsecond=0) + output_dir = tmp_path / "output" + output_dir.mkdir() + receipt_path = output_dir / "github-security-coverage-latest.json" + receipt_truth = { + "projects": [ + { + "identity": {"repo_full_name": "d/Alpha"}, + "derived": {"attention_state": "active-product"}, + } + ] + } + first_receipt = collect_security_coverage( + receipt_truth, + token=None, + expected_cohort_count=1, + now=now, + producer_commit="a" * 40, + ) + write_security_coverage_receipt( + first_receipt, + receipt_path, + expected_cohort_count=1, + ) + loaded = load_security_coverage_receipt( + receipt_path, + expected_cohort_count=1, + expected_producer_commit="a" * 40, + now=now, + ) + binding = loaded.binding() + metadata = { + "source_id": "github-security-coverage-receipt", + "schema_version": loaded.schema_version, + "produced_at": loaded.produced_at, + "state": loaded.receipt_state, + "age_hours": loaded.age_hours, + "producer_commit": loaded.producer_commit, + "cohort_policy": loaded.cohort_policy, + "cohort_repository_count": len(loaded.cohort_repositories), + "path": loaded.source_path, + "receipt_id": loaded.receipt_id, + "content_sha256": loaded.content_sha256, + } + registry_output = portfolio_workspace / "project-registry.md" + report_output = portfolio_workspace / "PORTFOLIO-AUDIT-REPORT.md" + registry_output.write_text("sentinel-registry\n") + report_output.write_text("sentinel-report\n") + + from src import portfolio_truth_publish as publish_module + + original_stage = publish_module._stage_text + replaced = False + + def stage_then_replace_receipt(target: Path, content: str) -> Path: + nonlocal replaced + staged = original_stage(target, content) + if not replaced: + replaced = True + newer_receipt = collect_security_coverage( + receipt_truth, + token=None, + expected_cohort_count=1, + now=now + timedelta(minutes=1), + producer_commit="a" * 40, + ) + write_security_coverage_receipt( + newer_receipt, + receipt_path, + expected_cohort_count=1, + ) + return staged + + monkeypatch.setattr(publish_module, "_stage_text", stage_then_replace_receipt) + + with pytest.raises( + PortfolioTruthPublishError, + match="changed after it was loaded", + ): + publish_portfolio_truth( + workspace_root=portfolio_workspace, + output_dir=output_dir, + registry_output=registry_output, + portfolio_report_output=report_output, + catalog_path=portfolio_catalog, + legacy_registry_path=legacy_registry, + include_notion=False, + security_alerts_by_name=loaded.entries_by_full_name, + security_coverage_metadata=metadata, + security_receipt_binding=binding, + ) + + assert replaced is True + assert registry_output.read_text() == "sentinel-registry\n" + assert report_output.read_text() == "sentinel-report\n" + assert not (output_dir / "portfolio-truth-latest.json").exists() + assert not list(output_dir.glob("portfolio-truth-*.json")) + + def test_publish_requires_producer_evidence_before_touching_outputs( portfolio_workspace: Path, portfolio_catalog: Path, @@ -2757,6 +2972,131 @@ def fake_publish(**_kwargs): assert captured["repo_status_cache"] is None +def test_portfolio_truth_app_carries_security_receipt_binding_to_publisher( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from types import SimpleNamespace + + from src.app.portfolio_truth import run_portfolio_truth_mode + from src.github_security_coverage import SecurityCoverageReceiptBinding + + receipt_path = tmp_path / "github-security-coverage-latest.json" + binding = SecurityCoverageReceiptBinding( + source_path=str(receipt_path), + receipt_id="sha256:" + "b" * 64, + content_sha256="c" * 64, + receipt_state="fresh", + max_age_hours=12, + expected_cohort_count=3, + expected_producer_commit=None, + ) + loaded = SimpleNamespace( + entries_by_full_name={}, + schema_version="GitHubSecurityCoverageReceiptV1", + produced_at="2026-07-31T12:00:00+00:00", + receipt_state="fresh", + age_hours=0.0, + producer_commit="a" * 40, + cohort_policy="portfolio-default-attention-v1", + cohort_repositories=("d/one", "d/two", "d/three"), + source_path=str(receipt_path), + receipt_id=binding.receipt_id, + content_sha256=binding.content_sha256, + binding=lambda: binding, + ) + captured: dict[str, object] = {} + + monkeypatch.setattr( + "src.app.portfolio_truth.load_security_coverage_by_full_name", + lambda **_kwargs: loaded, + ) + monkeypatch.setattr( + "src.app.portfolio_truth.load_live_repo_status_by_name", lambda **_kwargs: {} + ) + + def fake_publish(**kwargs): + captured.update(kwargs) + return SimpleNamespace( + latest_path=tmp_path / "latest.json", + snapshot_path=tmp_path / "history.json", + registry_output=tmp_path / "registry.md", + portfolio_report_output=tmp_path / "report.md", + project_count=0, + registry_changed=False, + report_changed=False, + ) + + monkeypatch.setattr("src.app.portfolio_truth.publish_portfolio_truth", fake_publish) + monkeypatch.setenv("GHRA_REQUIRE_PRODUCER_EVIDENCE", "0") + args = SimpleNamespace( + output_dir=str(tmp_path / "output"), + workspace_root=str(tmp_path), + registry_output=str(tmp_path / "registry.md"), + portfolio_report_output=str(tmp_path / "report.md"), + registry=None, + catalog=None, + username="testuser", + token=None, + no_cache=True, + portfolio_truth_include_release_count=False, + portfolio_truth_include_security=True, + portfolio_truth_security_receipt=str(receipt_path), + portfolio_truth_security_max_age_hours=12, + portfolio_truth_security_cohort_count=3, + portfolio_truth_allow_empty_notion=False, + ) + + run_portfolio_truth_mode(args) + + assert captured["security_receipt_binding"] == binding + metadata = captured["security_coverage_metadata"] + assert metadata["receipt_id"] == binding.receipt_id + assert metadata["content_sha256"] == binding.content_sha256 + + +def test_canonical_portfolio_truth_refuses_legacy_security_receipt_identity( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from types import SimpleNamespace + + from src.app.portfolio_truth import run_portfolio_truth_mode + from src.github_security_coverage import SecurityCoverageError + + def missing_binding(): + raise SecurityCoverageError( + "security coverage receipt is missing immutable receipt_id provenance" + ) + + loaded = SimpleNamespace(binding=missing_binding) + monkeypatch.setattr( + "src.app.portfolio_truth.load_security_coverage_by_full_name", + lambda **_kwargs: loaded, + ) + monkeypatch.setenv("GHRA_REQUIRE_PRODUCER_EVIDENCE", "1") + args = SimpleNamespace( + output_dir=str(tmp_path / "output"), + workspace_root=str(tmp_path), + registry_output=str(tmp_path / "registry.md"), + portfolio_report_output=str(tmp_path / "report.md"), + registry=None, + catalog=None, + username="testuser", + token=None, + no_cache=True, + portfolio_truth_include_release_count=False, + portfolio_truth_include_security=True, + portfolio_truth_security_receipt=None, + portfolio_truth_security_max_age_hours=24, + portfolio_truth_security_cohort_count=9, + portfolio_truth_allow_empty_notion=False, + ) + + with pytest.raises(SystemExit, match="missing immutable receipt_id"): + run_portfolio_truth_mode(args) + + def test_portfolio_truth_app_passes_validated_producer_receipt_to_publisher( tmp_path: Path, monkeypatch: pytest.MonkeyPatch,