From 876e92a52161801d71907541135e7d4409479f93 Mon Sep 17 00:00:00 2001 From: saagpatel Date: Wed, 5 Aug 2026 00:39:59 -0700 Subject: [PATCH 1/2] fix(portfolio): accept bounded prior truth metadata --- src/portfolio_truth_publish.py | 4 +- src/portfolio_truth_validate.py | 100 ++++++++++++++++++ tests/test_portfolio_truth.py | 181 ++++++++++++++++++++++++++++++++ 3 files changed, 283 insertions(+), 2 deletions(-) diff --git a/src/portfolio_truth_publish.py b/src/portfolio_truth_publish.py index 9e0fd8ab..184b9a70 100644 --- a/src/portfolio_truth_publish.py +++ b/src/portfolio_truth_publish.py @@ -28,7 +28,7 @@ from src.portfolio_truth_types import truth_latest_path from src.producer_preflight import ProducerEvidence, verify_evidence_still_current from src.portfolio_truth_validate import ( - canonicalize_truth_snapshot_payload, + canonicalize_prior_security_truth_payload, validate_portfolio_report_markdown, validate_publish_targets, validate_registry_markdown, @@ -104,7 +104,7 @@ def _load_prior_security_alerts( payload = json.loads(content) if not isinstance(payload, dict): raise ValueError("snapshot must be an object") - canonical = canonicalize_truth_snapshot_payload( + canonical = canonicalize_prior_security_truth_payload( payload, security_max_age_hours=security_max_age_hours, ) diff --git a/src/portfolio_truth_validate.py b/src/portfolio_truth_validate.py index 1dd1e4c5..75220cea 100644 --- a/src/portfolio_truth_validate.py +++ b/src/portfolio_truth_validate.py @@ -1802,6 +1802,106 @@ def canonicalize_truth_snapshot_payload( return snapshot.to_dict() +_LEGACY_PRIOR_SECURITY_DISCOVERY_POLICY_VERSION = "workspace_discovery.v2" + + +def canonicalize_prior_security_truth_payload( + payload: Mapping[str, Any], + *, + security_max_age_hours: int = 24, +) -> dict[str, Any]: + """Canonicalize prior security evidence across one bounded metadata upgrade. + + Current snapshots always take the ordinary strict path. The compatibility + path accepts only the immediately preceding discovery envelope, before + checkout-collision metadata existed, and reconstructs those two unrelated + metadata envelopes in memory before running the complete current validator. + """ + try: + return canonicalize_truth_snapshot_payload( + payload, + security_max_age_hours=security_max_age_hours, + ) + except ValueError as strict_error: + summary = payload.get("source_summary") + exclusions = payload.get("exclusions") + is_bounded_legacy_payload = ( + isinstance(summary, Mapping) + and "checkout_collisions" not in summary + and isinstance(exclusions, Mapping) + and exclusions.get("policy_version") + == _LEGACY_PRIOR_SECURITY_DISCOVERY_POLICY_VERSION + ) + if not is_bounded_legacy_payload: + raise + + try: + snapshot = _snapshot_from_payload(payload) + _validate_runtime_dataclass(snapshot, "snapshot") + if any( + "checkout_authority" in project.repository_state + for project in snapshot.projects + ): + raise ValueError( + "Legacy prior PortfolioTruth cannot contain checkout authority " + "without a collision summary." + ) + + current_summary = build_source_summary( + workspace_root=snapshot.workspace_root, + projects=snapshot.projects, + catalog_errors=summary["catalog_errors"], + catalog_warnings=summary["catalog_warnings"], + legacy_registry_rows=summary["legacy_registry_rows"], + notion_context_rows=summary["notion_context_rows"], + notion_context_carried_forward=summary[ + "notion_context_carried_forward" + ], + checkout_collisions=[], + ) + expected_legacy_summary = deepcopy(current_summary) + expected_legacy_summary.pop("checkout_collisions") + if dict(summary) != expected_legacy_summary: + raise ValueError( + "Legacy prior PortfolioTruth source summary differs from " + "producer facts." + ) + expected_warnings = build_warnings( + catalog_errors=summary["catalog_errors"], + catalog_warnings=summary["catalog_warnings"], + unresolved_duplicates=summary[ + "unresolved_duplicate_display_names" + ], + checkout_collisions=[], + ) + if snapshot.warnings != expected_warnings: + raise ValueError( + "Legacy prior PortfolioTruth warnings differ from producer facts." + ) + + exclusion_counts = exclusions.get("counts") + if set(exclusions) != {"policy_version", "counts"} or not isinstance( + exclusion_counts, Mapping + ): + raise ValueError( + "Legacy prior PortfolioTruth exclusions envelope is invalid." + ) + + migrated = deepcopy(dict(payload)) + migrated["source_summary"] = current_summary + migrated["exclusions"] = build_exclusions(dict(exclusion_counts)) + return canonicalize_truth_snapshot_payload( + migrated, + security_max_age_hours=security_max_age_hours, + ) + except (KeyError, TypeError) as legacy_error: + raise ValueError( + "Legacy prior PortfolioTruth metadata envelope is invalid." + ) from legacy_error + except ValueError as legacy_error: + raise legacy_error from strict_error + + def _without_documented_contract_canaries( payload: Mapping[str, Any], ) -> dict[str, Any]: diff --git a/tests/test_portfolio_truth.py b/tests/test_portfolio_truth.py index ee521a34..0be101c6 100644 --- a/tests/test_portfolio_truth.py +++ b/tests/test_portfolio_truth.py @@ -53,6 +53,7 @@ load_safe_notion_project_context, ) from src.portfolio_truth_validate import ( + canonicalize_prior_security_truth_payload, validate_portfolio_report_markdown, validate_truth_snapshot, ) @@ -519,6 +520,186 @@ def legacy_registry(tmp_path: Path) -> Path: return path +def _legacy_prior_security_payload( + *, + portfolio_workspace: Path, + portfolio_catalog: Path, + legacy_registry: Path, +) -> tuple[dict, dict]: + now = datetime(2026, 8, 4, 12, tzinfo=timezone.utc) + alpha = portfolio_workspace / "Alpha" + subprocess.run(["git", "init"], cwd=alpha, capture_output=True, check=True) + subprocess.run( + [ + "git", + "remote", + "add", + "origin", + "https://github.com/d/Alpha.git", + ], + cwd=alpha, + capture_output=True, + check=True, + ) + observed_at = now.isoformat() + provider_counts = { + "dependabot": {"critical": 0, "high": 1, "medium": 0, "low": 0}, + "code_scanning": {"critical": 0, "high": 0, "warning": 0, "note": 0}, + "secret_scanning": {"open": 0}, + } + security = { + "d/Alpha": { + "repo_full_name": "d/Alpha", + "cohort_member": True, + "cohort_policy": "portfolio-default-attention-v1", + "receipt_schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "receipt_state": "fresh", + "source_produced_at": observed_at, + "repository": _remote_repository_result( + state="observed", + observed_at=observed_at, + default_branch="main", + head_sha="b" * 40, + archived=False, + ), + "providers": { + provider: _provider_result( + provider, + state="observed", + observed_at=observed_at, + http_status=200, + pagination_complete=True, + counts=counts, + ) + for provider, counts in provider_counts.items() + }, + } + } + metadata = { + "source_id": "github-security-coverage-receipt", + "schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "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, + ) + payload = result.snapshot.to_dict() + payload["source_summary"].pop("checkout_collisions") + payload["exclusions"]["policy_version"] = "workspace_discovery.v2" + return payload, metadata + + +def test_prior_security_loader_accepts_bounded_legacy_truth( + tmp_path: Path, + portfolio_workspace: Path, + portfolio_catalog: Path, + legacy_registry: Path, +) -> None: + import src.portfolio_truth_publish as publish_mod + + payload, metadata = _legacy_prior_security_payload( + portfolio_workspace=portfolio_workspace, + portfolio_catalog=portfolio_catalog, + legacy_registry=legacy_registry, + ) + latest = tmp_path / "portfolio-truth-latest.json" + latest.write_text(json.dumps(payload), encoding="utf-8") + current_metadata = { + **metadata, + "produced_at": "2026-08-04T12:01:00+00:00", + } + + evidence = publish_mod._load_prior_security_alerts( + latest, + current_security_metadata=current_metadata, + security_max_age_hours=24, + ) + + assert evidence.final_cohort_repositories == ("d/Alpha",) + assert evidence.alerts_by_full_name["d/Alpha"]["dependabot_high"] == 1 + assert evidence.content_sha256 == hashlib.sha256(latest.read_bytes()).hexdigest() + + +@pytest.mark.parametrize( + ("mutation", "expected_message"), + ( + ("duplicate-repository", "one canonical project per origin"), + ("receipt-binding", "requires both receipt_id and content_sha256"), + ("provider-count", "does not match the normalized dependabot provider"), + ("future-receipt", "GitHub security input is future-dated"), + ), +) +def test_bounded_legacy_prior_security_remains_fail_closed( + mutation: str, + expected_message: str, + portfolio_workspace: Path, + portfolio_catalog: Path, + legacy_registry: Path, +) -> None: + payload, _ = _legacy_prior_security_payload( + portfolio_workspace=portfolio_workspace, + portfolio_catalog=portfolio_catalog, + legacy_registry=legacy_registry, + ) + if mutation == "duplicate-repository": + other = next( + project + for project in payload["projects"] + if project["identity"]["project_key"] != "Alpha" + ) + other["identity"]["repo_full_name"] = "d/Alpha" + elif mutation == "receipt-binding": + payload["inputs"]["github_security"].pop("content_sha256") + elif mutation == "provider-count": + alpha = next( + project + for project in payload["projects"] + if project["identity"]["project_key"] == "Alpha" + ) + alpha["security"]["providers"]["dependabot"]["counts"]["high"] = 2 + else: + payload["inputs"]["github_security"]["produced_at"] = ( + "2026-08-05T12:00:00+00:00" + ) + + with pytest.raises(ValueError, match=expected_message): + canonicalize_prior_security_truth_payload(payload) + + +def test_current_prior_truth_failure_cannot_use_legacy_fallback( + portfolio_workspace: Path, + portfolio_catalog: Path, + legacy_registry: Path, +) -> None: + legacy, _ = _legacy_prior_security_payload( + portfolio_workspace=portfolio_workspace, + portfolio_catalog=portfolio_catalog, + legacy_registry=legacy_registry, + ) + current = canonicalize_prior_security_truth_payload(legacy) + current["source_summary"]["checkout_collisions"]["state"] = "unknown" + + with pytest.raises( + ValueError, + match="Checkout collision summary state does not match group authority", + ): + canonicalize_prior_security_truth_payload(current) + + def test_truth_snapshot_respects_declared_and_derived_fields( portfolio_workspace: Path, portfolio_catalog: Path, From f4a20112954861417e2dea824e88b6a2a3f30b62 Mon Sep 17 00:00:00 2001 From: saagpatel Date: Wed, 5 Aug 2026 00:57:27 -0700 Subject: [PATCH 2/2] fix(portfolio): validate legacy security decisions --- src/portfolio_truth_validate.py | 102 +++++++++++++++++++++++++++++++- tests/test_portfolio_truth.py | 19 +++--- 2 files changed, 112 insertions(+), 9 deletions(-) diff --git a/src/portfolio_truth_validate.py b/src/portfolio_truth_validate.py index 211f0152..5c2cd12f 100644 --- a/src/portfolio_truth_validate.py +++ b/src/portfolio_truth_validate.py @@ -1811,6 +1811,50 @@ def canonicalize_truth_snapshot_payload( _LEGACY_PRIOR_SECURITY_DISCOVERY_POLICY_VERSION = "workspace_discovery.v2" +_LEGACY_SECURITY_RISK_LABEL = "open high/critical security alerts" +_CURRENT_SECURITY_RISK_LABEL = "blocking GitHub security findings" + + +def _project_decision_for_security_policy( + project: PortfolioTruthProject, + *, + legacy_dependabot_only: bool, +) -> tuple[dict[str, Any], str]: + if legacy_dependabot_only: + coverage_state = project.security.coverage_state + high_alerts = project.security.dependabot_high or 0 + critical_alerts = project.security.dependabot_critical or 0 + else: + security_admission = derive_security_admission(project.security.to_dict()) + coverage_state = security_admission.effective_coverage_state + high_alerts = security_admission.total_open_high + critical_alerts = ( + security_admission.total_open_critical + + security_admission.total_open_secrets + ) + risk, attention = build_project_decision( + display_name=project.identity.display_name, + operating_path=project.declared.operating_path, + path_override=project.derived.path_override, + context_quality=project.derived.context_quality, + activity_status=project.derived.activity_status, + archived=project.derived.archived, + lifecycle_state=project.declared.lifecycle_state, + category=project.declared.category, + criticality=project.declared.criticality, + doctor_standard=project.declared.doctor_standard, + known_risks_present=project.derived.known_risks_present, + run_instructions_present=project.derived.run_instructions_present, + security_coverage_state=coverage_state, + security_high_alerts=high_alerts, + security_critical_alerts=critical_alerts, + ) + if legacy_dependabot_only: + risk["risk_summary"] = risk["risk_summary"].replace( + _CURRENT_SECURITY_RISK_LABEL, + _LEGACY_SECURITY_RISK_LABEL, + ) + return risk, attention def canonicalize_prior_security_truth_payload( @@ -1854,6 +1898,21 @@ def canonicalize_prior_security_truth_payload( "Legacy prior PortfolioTruth cannot contain checkout authority " "without a collision summary." ) + for project in snapshot.projects: + legacy_risk, legacy_attention = _project_decision_for_security_policy( + project, + legacy_dependabot_only=True, + ) + if project.risk.to_dict() != legacy_risk: + raise ValueError( + "Legacy prior PortfolioTruth project risk differs from " + f"its bounded derivation: {project.identity.project_key}." + ) + if project.derived.attention_state != legacy_attention: + raise ValueError( + "Legacy prior PortfolioTruth project attention differs from " + f"its bounded derivation: {project.identity.project_key}." + ) current_summary = build_source_summary( workspace_root=snapshot.workspace_root, @@ -1896,12 +1955,51 @@ def canonicalize_prior_security_truth_payload( ) migrated = deepcopy(dict(payload)) - migrated["source_summary"] = current_summary + migrated_projects = migrated.get("projects") + if not isinstance(migrated_projects, list) or len(migrated_projects) != len( + snapshot.projects + ): + raise ValueError( + "Legacy prior PortfolioTruth project envelope is invalid." + ) + for raw_project, project in zip( + migrated_projects, + snapshot.projects, + strict=True, + ): + if not isinstance(raw_project, dict) or not isinstance( + raw_project.get("derived"), dict + ): + raise ValueError( + "Legacy prior PortfolioTruth project envelope is invalid." + ) + current_risk, current_attention = ( + _project_decision_for_security_policy( + project, + legacy_dependabot_only=False, + ) + ) + raw_project["risk"] = current_risk + raw_project["derived"]["attention_state"] = current_attention + migrated_snapshot = _snapshot_from_payload(migrated) + migrated["source_summary"] = build_source_summary( + workspace_root=migrated_snapshot.workspace_root, + projects=migrated_snapshot.projects, + catalog_errors=summary["catalog_errors"], + catalog_warnings=summary["catalog_warnings"], + legacy_registry_rows=summary["legacy_registry_rows"], + notion_context_rows=summary["notion_context_rows"], + notion_context_carried_forward=summary[ + "notion_context_carried_forward" + ], + checkout_collisions=[], + ) migrated["exclusions"] = build_exclusions(dict(exclusion_counts)) - return canonicalize_truth_snapshot_payload( + canonicalize_truth_snapshot_payload( migrated, security_max_age_hours=security_max_age_hours, ) + return snapshot.to_dict() except (KeyError, TypeError) as legacy_error: raise ValueError( "Legacy prior PortfolioTruth metadata envelope is invalid." diff --git a/tests/test_portfolio_truth.py b/tests/test_portfolio_truth.py index 95e3554a..33ebcd96 100644 --- a/tests/test_portfolio_truth.py +++ b/tests/test_portfolio_truth.py @@ -591,7 +591,7 @@ def _legacy_prior_security_payload( portfolio_workspace: Path, portfolio_catalog: Path, legacy_registry: Path, -) -> tuple[dict, dict]: +) -> tuple[dict, dict, dict]: now = datetime(2026, 8, 4, 12, tzinfo=timezone.utc) alpha = portfolio_workspace / "Alpha" subprocess.run(["git", "init"], cwd=alpha, capture_output=True, check=True) @@ -663,10 +663,16 @@ def _legacy_prior_security_payload( security_alerts_by_name=security, security_coverage_metadata=metadata, ) - payload = result.snapshot.to_dict() + current = result.snapshot.to_dict() + payload = deepcopy(current) payload["source_summary"].pop("checkout_collisions") payload["exclusions"]["policy_version"] = "workspace_discovery.v2" - return payload, metadata + for project in payload["projects"]: + project["risk"]["risk_summary"] = project["risk"]["risk_summary"].replace( + "blocking GitHub security findings", + "open high/critical security alerts", + ) + return payload, metadata, current def test_prior_security_loader_accepts_bounded_legacy_truth( @@ -677,7 +683,7 @@ def test_prior_security_loader_accepts_bounded_legacy_truth( ) -> None: import src.portfolio_truth_publish as publish_mod - payload, metadata = _legacy_prior_security_payload( + payload, metadata, _ = _legacy_prior_security_payload( portfolio_workspace=portfolio_workspace, portfolio_catalog=portfolio_catalog, legacy_registry=legacy_registry, @@ -716,7 +722,7 @@ def test_bounded_legacy_prior_security_remains_fail_closed( portfolio_catalog: Path, legacy_registry: Path, ) -> None: - payload, _ = _legacy_prior_security_payload( + payload, _, _ = _legacy_prior_security_payload( portfolio_workspace=portfolio_workspace, portfolio_catalog=portfolio_catalog, legacy_registry=legacy_registry, @@ -751,12 +757,11 @@ def test_current_prior_truth_failure_cannot_use_legacy_fallback( portfolio_catalog: Path, legacy_registry: Path, ) -> None: - legacy, _ = _legacy_prior_security_payload( + _, _, current = _legacy_prior_security_payload( portfolio_workspace=portfolio_workspace, portfolio_catalog=portfolio_catalog, legacy_registry=legacy_registry, ) - current = canonicalize_prior_security_truth_payload(legacy) current["source_summary"]["checkout_collisions"]["state"] = "unknown" with pytest.raises(