Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions src/portfolio_truth_publish.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
)
Expand Down
198 changes: 198 additions & 0 deletions src/portfolio_truth_validate.py
Original file line number Diff line number Diff line change
Expand Up @@ -1810,6 +1810,204 @@ def canonicalize_truth_snapshot_payload(
return snapshot.to_dict()


_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(
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."
)
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,
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_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))
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."
) 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]:
Expand Down
Loading