"""Bounded installed-composition and externally authorized fit evidence.""" from __future__ import annotations import base64 import csv from dataclasses import dataclass from datetime import UTC, datetime, timedelta import hashlib import hmac from importlib import metadata from importlib.util import source_from_cache import io import json import os from pathlib import Path, PurePosixPath import re import stat import sys from typing import Any, Iterable, Mapping, Sequence from urllib.parse import urlsplit from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from jsonschema import Draft202012Validator, FormatChecker from jsonschema.exceptions import SchemaError from govoplan_release.artifact_identity import ( INSTALLED_PAYLOAD_ALGORITHM, WHEEL_PAYLOAD_ALGORITHM, payload_identity_sha256, ) MAX_GOVOPLAN_DISTRIBUTIONS = 256 MAX_DISTRIBUTION_FILES = 10_000 MAX_COLLECTION_RECORD_FILES = 50_000 MAX_RECORD_BYTES = 4 * 1024 * 1024 MAX_RECORD_PATH_CHARACTERS = 4096 MAX_HASHED_FILE_BYTES = 64 * 1024 * 1024 MAX_HASHED_DISTRIBUTION_BYTES = 512 * 1024 * 1024 MAX_HASHED_COLLECTION_BYTES = 2 * 1024 * 1024 * 1024 MAX_DIRECT_URL_BYTES = 64 * 1024 MAX_LOCAL_OBSERVATION_AGE = timedelta(minutes=5) MAX_LOCAL_OBSERVATION_FUTURE_SKEW = timedelta(seconds=30) INSTALLED_OBSERVATION_MODES = frozenset({"direct_local", "imported_unsigned"}) VERIFICATION_MODES = frozenset({"current", "historical_override"}) PACKAGE_PATTERN = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$") OPAQUE_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]*$") VERSION_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+!-]*$") ENTRY_POINT_VALUE_PATTERN = re.compile( r"^[A-Za-z_][A-Za-z0-9_.]*(?::[A-Za-z_][A-Za-z0-9_.]*)?(?:\s*\[[A-Za-z0-9_., -]+\])?$" ) VCS_REVISION_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/+~-]{0,159}$") GIT_OBJECT_PATTERN = re.compile(r"^[0-9a-f]{40,64}$") SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") POSITIVE_RESULTS = { "target_environment": "passed", "external_providers": "passed", "production_approval": "approved", } NEGATIVE_RESULTS = { "target_environment": "failed", "external_providers": "failed", "production_approval": "rejected", } PROOF_REQUIREMENTS = { "target_environment": ( "A target-run result bound to this assessment release and installed evidence whose release origin is independently anchored.", "An independently trusted authority key permitted for target_environment proof.", "Opaque control IDs and content hashes for the target acceptance artifacts.", ), "external_providers": ( "Provider acceptance results bound to this assessment release and installed evidence whose release origin is independently anchored.", "An independently trusted authority key permitted for external_providers proof.", "Opaque control IDs and content hashes; credentials and endpoint identifiers stay outside the report.", ), "production_approval": ( "An explicit approval bound to this assessment release and installed evidence whose release origin is independently anchored.", "An independently provisioned authority key permitted for production_approval.", "An unexpired signed proof bundle; an assessment or operator cannot approve itself.", ), } @dataclass(frozen=True, slots=True) class EvidenceFinding: severity: str code: str message: str assessment_ids: tuple[str, ...] = () @dataclass(frozen=True, slots=True) class InstalledEvidenceReview: findings: tuple[EvidenceFinding, ...] changes: tuple[dict[str, Any], ...] changed_repositories: frozenset[str] affected_module_ids: frozenset[str] proof_scope: Mapping[str, Any] evidence_sha256: str | None @dataclass(frozen=True, slots=True) class BoundaryEvidenceReview: findings: tuple[EvidenceFinding, ...] proof_scope: Mapping[str, Any] review_targets: tuple[dict[str, Any], ...] = () @dataclass(slots=True) class CollectionHashBudget: remaining_files: int = MAX_COLLECTION_RECORD_FILES remaining_bytes: int = MAX_HASHED_COLLECTION_BYTES @dataclass(frozen=True, slots=True) class RecordEntry: path: str hash_mode: str | None hash_value: str | None size: int | None def validate_payload( *, payload: dict[str, Any], schema: dict[str, Any] ) -> tuple[str, ...]: try: Draft202012Validator.check_schema(schema) except SchemaError as exc: return (f"$schema: {exc.message}",) validator = Draft202012Validator(schema, format_checker=FormatChecker()) return tuple( f"{_json_path(error.absolute_path)}: failed {error.validator!r} validation" for error in sorted( validator.iter_errors(payload), key=lambda item: ( tuple(str(part) for part in item.absolute_path), str(item.validator), item.message, ), ) ) def collect_installed_composition( *, assessment: dict[str, Any], collected_at: datetime | None = None, distributions: Iterable[metadata.Distribution] | None = None, ) -> dict[str, Any]: """Collect only bounded, non-identifying GovOPlaN distribution evidence. Loading ``govoplan.modules`` entry points executes installed module code. The collector catches failures and records only a stable error code; it never serializes exception text, URLs, local paths, hostnames, or usernames. """ artifacts: list[dict[str, Any]] = [] issues: list[dict[str, str]] = [] hash_budget = CollectionHashBudget() candidates: list[tuple[str, metadata.Distribution]] = [] source = metadata.distributions() if distributions is None else distributions for distribution in source: try: raw_package_name = distribution.metadata["Name"] if not isinstance(raw_package_name, str) or len(raw_package_name) > 256: continue package_name = normalize_package_name(raw_package_name) except (Exception, SystemExit): continue if ( package_name.startswith("govoplan-") and len(package_name) <= 128 and PACKAGE_PATTERN.fullmatch(package_name) is not None ): candidates.append((package_name, distribution)) candidates.sort(key=lambda item: (item[0], _distribution_version(item[1]))) if len(candidates) > MAX_GOVOPLAN_DISTRIBUTIONS: candidates = candidates[:MAX_GOVOPLAN_DISTRIBUTIONS] issues.append( { "code": "collection-limit-exceeded", "package_name": "govoplan-collector", } ) name_counts: dict[str, int] = {} for package_name, _ in candidates: name_counts[package_name] = name_counts.get(package_name, 0) + 1 for package_name, count in sorted(name_counts.items()): if count > 1: issues.append( { "code": "duplicate-distribution", "package_name": package_name, } ) for package_name, distribution in candidates: version = _distribution_version(distribution) if not version: issues.append( { "code": "distribution-metadata-invalid", "package_name": package_name if PACKAGE_PATTERN.fullmatch(package_name) else "govoplan-invalid", } ) continue modules, module_issues = _distribution_modules( package_name=package_name, distribution=distribution, ) issues.extend(module_issues) artifacts.append( { "package_name": package_name, "package_version": version, "modules": modules, "source_provenance": _source_provenance(distribution), "record_integrity": _record_integrity( distribution, package_name=package_name, budget=hash_budget, ), } ) observed_at = collected_at or datetime.now(tz=UTC) if observed_at.tzinfo is None: observed_at = observed_at.replace(tzinfo=UTC) release = assessment.get("release") assessment_release = release.get("ref") if isinstance(release, dict) else None sorted_issues = sorted( _unique_dicts(issues), key=lambda item: (item["package_name"], item["code"]), ) if len(sorted_issues) > 256: limit_issue = { "code": "collection-limit-exceeded", "package_name": "govoplan-collector", } sorted_issues = sorted_issues[:255] if limit_issue not in sorted_issues: sorted_issues.append(limit_issue) sorted_issues.sort(key=lambda item: (item["package_name"], item["code"])) return { "$schema": "./installed-composition-evidence.schema.json", "schema_version": "0.4.0", "evidence_kind": "govoplan.installed-composition", "assessment_id": _safe_opaque_id( assessment.get("assessment_id"), fallback="invalid-assessment" ), "assessment_release": _safe_opaque_id( assessment_release, fallback="invalid-release" ), "collected_at": observed_at.astimezone(UTC).isoformat().replace("+00:00", "Z"), "scope": "current-python-environment.govoplan-distributions", "artifacts": sorted( artifacts, key=lambda item: ( item["package_name"], item["package_version"], canonical_bytes(item), ), ), "collection_issues": sorted_issues, } def review_installed_composition( *, assessment: dict[str, Any], catalog_entries: Mapping[str, Mapping[str, Any]], selected_versions: Mapping[str, str], selected_commits: Mapping[str, str], evidence: dict[str, Any] | None, schema: dict[str, Any] | None, observation_mode: str = "imported_unsigned", verification_time: datetime | None = None, verification_mode: str = "current", catalog_trusted: bool = False, catalog_artifacts: object = None, catalog_sha256: str | None = None, catalog_channel: str | None = None, catalog_sequence: int | None = None, installer_receipt: dict[str, Any] | None = None, installer_receipt_schema: dict[str, Any] | None = None, installer_authority_keyring: dict[str, Any] | None = None, installer_authority_keyring_schema: dict[str, Any] | None = None, forbidden_installer_public_keys: frozenset[bytes] = frozenset(), ) -> InstalledEvidenceReview: if evidence is None: return InstalledEvidenceReview( findings=(), changes=(), changed_repositories=frozenset(), affected_module_ids=frozenset(), proof_scope=_unchecked_installed_scope(), evidence_sha256=None, ) digest = canonical_sha256(evidence) if schema is None: finding = EvidenceFinding( "blocker", "installed_evidence_schema_missing", "Installed evidence was supplied without its validation schema.", ("assessment.installed_composition",), ) return InstalledEvidenceReview( findings=(finding,), changes=(), changed_repositories=frozenset(), affected_module_ids=frozenset(), proof_scope=_invalid_installed_scope(digest=digest), evidence_sha256=digest, ) schema_errors = validate_payload(payload=evidence, schema=schema) if schema_errors: findings = tuple( EvidenceFinding( "blocker", "installed_evidence_schema", message, ("assessment.installed_composition",), ) for message in schema_errors ) return InstalledEvidenceReview( findings=findings, changes=(), changed_repositories=frozenset(), affected_module_ids=frozenset(), proof_scope=_invalid_installed_scope(digest=digest), evidence_sha256=digest, ) findings: list[EvidenceFinding] = [] effective_observation_mode = ( observation_mode if observation_mode in INSTALLED_OBSERVATION_MODES else "invalid" ) effective_verification_mode = ( verification_mode if verification_mode in VERIFICATION_MODES else "invalid" ) if effective_observation_mode == "invalid": findings.append( EvidenceFinding( "blocker", "installed_evidence_observation_mode", "Installed evidence observation mode is invalid.", ("assessment.installed_composition",), ) ) if effective_verification_mode == "invalid": findings.append( EvidenceFinding( "blocker", "installed_evidence_verification_mode", "Installed evidence verification mode is invalid.", ("assessment.installed_composition",), ) ) if catalog_trusted is not True: findings.append( EvidenceFinding( "blocker", "installed_catalog_trust_unavailable", "Installed-composition proof depends on a valid signed catalog.", ("assessment.release", "assessment.installed_composition"), ) ) changes: list[dict[str, Any]] = [] changed_repositories: set[str] = set() affected_module_ids: set[str] = set() now = verification_time or datetime.now(tz=UTC) if now.tzinfo is None: now = now.replace(tzinfo=UTC) collected_at = _datetime(evidence.get("collected_at")) observation_trusted = False freshness = "invalid" if effective_observation_mode == "direct_local": if collected_at is None: findings.append( EvidenceFinding( "blocker", "installed_evidence_time_invalid", "Direct installed evidence has no valid collection time.", ("assessment.installed_composition",), ) ) elif collected_at > now + MAX_LOCAL_OBSERVATION_FUTURE_SKEW: freshness = "future" findings.append( EvidenceFinding( "blocker", "installed_evidence_from_future", "Direct installed evidence is newer than the permitted clock skew.", ("assessment.installed_composition",), ) ) elif now - collected_at > MAX_LOCAL_OBSERVATION_AGE: freshness = "stale" findings.append( EvidenceFinding( "blocker", "installed_evidence_stale", "Direct installed evidence is older than the five-minute acceptance window.", ("assessment.installed_composition",), ) ) else: freshness = "fresh" observation_trusted = True elif effective_observation_mode == "imported_unsigned": freshness = "unsigned_import" if effective_verification_mode == "invalid": observation_trusted = False if catalog_trusted is not True: observation_trusted = False expected_release = ( assessment.get("release", {}).get("ref") if isinstance(assessment.get("release"), dict) else None ) binding_valid = True if evidence.get("assessment_id") != assessment.get("assessment_id"): binding_valid = False findings.append( EvidenceFinding( "blocker", "installed_evidence_assessment_mismatch", "Installed evidence is bound to a different assessment ID.", ("assessment.installed_composition",), ) ) if evidence.get("assessment_release") != expected_release: binding_valid = False findings.append( EvidenceFinding( "blocker", "installed_evidence_release_mismatch", "Installed evidence is bound to a different assessment release.", ("assessment.release", "assessment.installed_composition"), ) ) artifact_rows = [ item for item in evidence.get("artifacts", []) if isinstance(item, dict) ] artifacts_by_package: dict[str, dict[str, Any]] = {} duplicate_packages: set[str] = set() for artifact in artifact_rows: package_name = normalize_package_name(str(artifact.get("package_name") or "")) if package_name in artifacts_by_package: duplicate_packages.add(package_name) else: artifacts_by_package[package_name] = artifact for package_name in sorted(duplicate_packages): binding_valid = False findings.append( EvidenceFinding( "blocker", "installed_distribution_duplicate", f"Installed evidence repeats distribution {package_name!r}.", ("assessment.installed_composition",), ) ) expected_by_package: dict[str, tuple[dict[str, Any], Mapping[str, Any]]] = {} assessment_by_module: dict[str, dict[str, Any]] = {} for component in assessment.get("composition", []): if not isinstance(component, dict): continue module_id = str(component.get("module_id") or "") assessment_by_module[module_id] = component if component.get("enabled") is not True: continue catalog_entry = catalog_entries.get(module_id) if catalog_entry is None: binding_valid = False findings.append( EvidenceFinding( "blocker", "installed_expected_catalog_entry_missing", f"Enabled module {module_id!r} cannot be mapped to a catalog package.", (f"composition.{module_id}",), ) ) continue package_name = normalize_package_name(str(catalog_entry.get("package") or "")) if not package_name: binding_valid = False findings.append( EvidenceFinding( "blocker", "installed_expected_package_missing", f"Enabled module {module_id!r} has no valid catalog package name.", (f"composition.{module_id}",), ) ) continue if package_name in expected_by_package: binding_valid = False findings.append( EvidenceFinding( "blocker", "installed_expected_package_ambiguous", f"Multiple enabled composition entries map to {package_name!r}.", ("assessment.installed_composition",), ) ) continue expected_by_package[package_name] = (component, catalog_entry) if not expected_by_package: binding_valid = False findings.append( EvidenceFinding( "blocker", "installed_expected_composition_empty", "Installed proof requires at least one enabled assessment component mapped to a catalog package.", ("assessment.installed_composition",), ) ) record_expected = 0 record_verified = 0 generated_unhashed_file_count = 0 provenance_expected = 0 provenance_declared_match = 0 provenance_mutable = 0 provenance_unanchored = 0 payload_consistent = binding_valid and not duplicate_packages for package_name, (component, catalog_entry) in sorted(expected_by_package.items()): module_id = str(component["module_id"]) repository = str(component["repository"]) artifact = artifacts_by_package.get(package_name) record_expected += 1 provenance_expected += 1 if package_name in duplicate_packages: payload_consistent = False _record_change( findings=findings, changes=changes, changed_repositories=changed_repositories, affected_module_ids=affected_module_ids, component=component, kind="installed_distribution_duplicate", code="installed_distribution_duplicate_ambiguous", message=f"Installed distribution {package_name!r} is duplicated, so no candidate was selected for integrity or provenance conclusions.", ) continue if artifact is None: payload_consistent = False _record_change( findings=findings, changes=changes, changed_repositories=changed_repositories, affected_module_ids=affected_module_ids, component=component, kind="installed_distribution_missing", code="installed_distribution_missing", message=f"Enabled module {module_id!r} has no installed distribution {package_name!r} in the evidence.", ) continue expected_version = str(component["manifest_version"]) package_version = str(artifact.get("package_version") or "") catalog_version = str(catalog_entry.get("version") or "") if package_version != expected_version or package_version != catalog_version: payload_consistent = False _record_change( findings=findings, changes=changes, changed_repositories=changed_repositories, affected_module_ids=affected_module_ids, component=component, kind="installed_version_mismatch", code="installed_distribution_version_mismatch", message=( f"Installed {package_name!r} version {package_version!r} does not match " f"assessment {expected_version!r} and catalog {catalog_version!r}." ), extra={ "installed_version": package_version, "assessed_version": expected_version, "catalog_version": catalog_version, }, ) modules = [ item for item in artifact.get("modules", []) if isinstance(item, dict) ] module_rows: dict[str, dict[str, Any]] = {} duplicate_module_ids: set[str] = set() for item in modules: observed_id = str(item.get("module_id") or "") if observed_id in module_rows: duplicate_module_ids.add(observed_id) else: module_rows[observed_id] = item expected_manifest = module_rows.get(module_id) manifest_mismatch = ( expected_manifest is None if module_id != "core" else False ) or ( expected_manifest is not None and str(expected_manifest.get("manifest_version") or "") != expected_version ) if manifest_mismatch: payload_consistent = False _record_change( findings=findings, changes=changes, changed_repositories=changed_repositories, affected_module_ids=affected_module_ids, component=component, kind="installed_manifest_mismatch", code="installed_manifest_version_mismatch", message=f"Installed distribution {package_name!r} does not expose module {module_id!r} at version {expected_version!r}.", ) unexpected_modules = sorted(set(module_rows) - {module_id}) if duplicate_module_ids or unexpected_modules: payload_consistent = False _record_change( findings=findings, changes=changes, changed_repositories=changed_repositories, affected_module_ids=affected_module_ids, component=component, kind="installed_manifest_set_mismatch", code="installed_manifest_set_mismatch", message=f"Installed distribution {package_name!r} exposes an unexpected or duplicate module entry-point set.", ) record = artifact.get("record_integrity") record_status = record.get("status") if isinstance(record, dict) else None record_semantics_valid = ( _record_integrity_semantics_valid(record) if isinstance(record, dict) else False ) if isinstance(record, dict): generated_count = record.get("generated_unhashed_file_count") if isinstance(generated_count, int) and not isinstance( generated_count, bool ): generated_unhashed_file_count += max(generated_count, 0) if record_status == "verified" and record_semantics_valid: record_verified += 1 else: payload_consistent = False _record_change( findings=findings, changes=changes, changed_repositories=changed_repositories, affected_module_ids=affected_module_ids, component=component, kind="installed_record_unverified", code=( "installed_record_integrity_unverified" if record_semantics_valid else "installed_record_integrity_invalid" ), message=( f"Installed distribution {package_name!r} does not have complete, matching RECORD hash coverage." if record_semantics_valid else f"Installed distribution {package_name!r} has contradictory RECORD status and counters." ), extra={"record_status": record_status}, ) provenance = artifact.get("source_provenance") provenance_kind = ( str(provenance.get("kind") or "") if isinstance(provenance, dict) else "" ) provenance_basis = ( str(provenance.get("basis") or "") if isinstance(provenance, dict) else "" ) if provenance_basis != "local-pep610-metadata": provenance_unanchored += 1 _record_change( findings=findings, changes=changes, changed_repositories=changed_repositories, affected_module_ids=affected_module_ids, component=component, kind="installed_source_metadata_invalid", code="installed_source_metadata_basis_invalid", message=f"Installed distribution {package_name!r} does not identify its source statement as local PEP 610 metadata.", ) continue if provenance_kind == "vcs-commit": commit = str(provenance.get("commit") or "").lower() selected_version = selected_versions.get(repository) selected_commit = selected_commits.get(repository) assessed_commit = str(component.get("commit") or "").lower() commit_matches = commit.startswith(assessed_commit) selected_matches = ( selected_version != expected_version or selected_commit is None or commit == selected_commit.lower() ) if commit_matches and selected_matches: provenance_declared_match += 1 else: _record_change( findings=findings, changes=changes, changed_repositories=changed_repositories, affected_module_ids=affected_module_ids, component=component, kind="installed_commit_mismatch", code="installed_source_commit_mismatch", message=f"Installed distribution {package_name!r} declares a local PEP 610 VCS commit that does not match the assessed or signed selected-unit commit.", ) elif provenance_kind in {"editable-vcs", "editable-local", "local-directory"}: provenance_mutable += 1 _record_change( findings=findings, changes=changes, changed_repositories=changed_repositories, affected_module_ids=affected_module_ids, component=component, kind="installed_source_mutable", code="installed_source_mutable", message=f"Installed distribution {package_name!r} declares an editable or directory-backed local source.", ) else: provenance_unanchored += 1 _record_change( findings=findings, changes=changes, changed_repositories=changed_repositories, affected_module_ids=affected_module_ids, component=component, kind="installed_source_unanchored", code="installed_source_unanchored", message=f"Installed distribution {package_name!r} has no local PEP 610 source statement matching the assessed release.", ) expected_packages = set(expected_by_package) for package_name in sorted(set(artifacts_by_package) - expected_packages): payload_consistent = False matching_module_id = next( ( module_id for module_id, entry in catalog_entries.items() if normalize_package_name(str(entry.get("package") or "")) == package_name ), None, ) component = assessment_by_module.get(str(matching_module_id or "")) repository = ( str(component.get("repository")) if isinstance(component, dict) else None ) if repository: changed_repositories.add(repository) if matching_module_id and component is not None: affected_module_ids.add(str(matching_module_id)) target_ids = ( (f"composition.{matching_module_id}",) if matching_module_id and component is not None else ("assessment.installed_composition",) ) findings.append( EvidenceFinding( "review", "installed_distribution_extra", f"Installed GovOPlaN distribution {package_name!r} is outside the enabled assessed composition.", target_ids, ) ) changes.append( { "module_id": str(matching_module_id or package_name), "repository": repository, "kind": "installed_distribution_extra", "package_name": package_name, } ) for issue in evidence.get("collection_issues", []): if not isinstance(issue, dict): continue package_name = str(issue.get("package_name") or "") code = str(issue.get("code") or "") payload_consistent = False findings.append( EvidenceFinding( "review", "installed_collection_incomplete", f"Installed evidence collection reported {code!r} for {package_name!r}.", ("assessment.installed_composition",), ) ) exact_set = ( set(artifacts_by_package) == expected_packages and not duplicate_packages ) origin_scope, origin_findings = _review_installed_release_origin( expected_by_package=expected_by_package, installed_artifacts=artifacts_by_package, installed_evidence=evidence, installed_evidence_sha256=digest, installed_composition_valid=( binding_valid and exact_set and payload_consistent and record_verified == record_expected and record_expected > 0 ), observation_trusted=observation_trusted, catalog_artifacts=catalog_artifacts, catalog_sha256=catalog_sha256, catalog_channel=catalog_channel, catalog_sequence=catalog_sequence, catalog_trusted=catalog_trusted, receipt=installer_receipt, receipt_schema=installer_receipt_schema, authority_keyring=installer_authority_keyring, authority_keyring_schema=installer_authority_keyring_schema, verification_time=now, forbidden_public_keys=forbidden_installer_public_keys, ) findings.extend(origin_findings) receipt_authenticated = ( origin_scope.get("valid") is True and origin_scope.get("signed_installer_receipt_count") == record_expected and record_expected > 0 ) effective_observation_trusted = observation_trusted or receipt_authenticated if ( effective_observation_mode == "imported_unsigned" and not receipt_authenticated ): findings.append( EvidenceFinding( "blocker", "installed_evidence_unsigned_import", "Imported installed evidence requires a valid independent installer receipt before it can establish installed proof.", ("assessment.installed_composition",), ) ) installed_checked = effective_observation_trusted record_checked = ( effective_observation_trusted and binding_valid and exact_set and record_expected > 0 ) provenance_checked = ( effective_observation_trusted and binding_valid and exact_set and provenance_expected > 0 ) proof_scope = { "installed_artifacts": { "checked": installed_checked, "valid": payload_consistent if installed_checked else None, "comparison_valid": payload_consistent, "evidence_sha256": digest, "expected_distribution_count": len(expected_packages), "observed_distribution_count": len(artifact_rows), }, "installed_record_integrity": { "checked": record_checked, "valid": record_verified == record_expected if record_checked else None, "verified_distribution_count": record_verified, "expected_distribution_count": record_expected, "generated_unhashed_file_count": generated_unhashed_file_count, }, "installed_source_provenance": { "checked": provenance_checked, "valid": provenance_declared_match == provenance_expected if provenance_checked else None, "basis": "local-pep610-metadata", "declared_match_distribution_count": provenance_declared_match, "mutable_distribution_count": provenance_mutable, "unanchored_distribution_count": provenance_unanchored, "expected_distribution_count": provenance_expected, "release_origin_bound": False, }, "installed_release_origin": origin_scope, "runtime_activation": { "checked": False, "valid": None, "required_evidence": ( "A separately collected runtime registry/configuration snapshot bound to the installed evidence digest.", ), }, "installed_evidence_observation": { "checked": True, "mode": effective_observation_mode, "freshness": freshness, "collected_at": _iso_datetime(collected_at), "evaluated_at": _iso_datetime(now), "verification_mode": effective_verification_mode, "catalog_trusted": catalog_trusted is True, "receipt_authenticated": receipt_authenticated, "freshness_window_seconds": int(MAX_LOCAL_OBSERVATION_AGE.total_seconds()), }, } return InstalledEvidenceReview( findings=tuple(findings), changes=tuple(changes), changed_repositories=frozenset(changed_repositories), affected_module_ids=frozenset(affected_module_ids), proof_scope=proof_scope, evidence_sha256=digest, ) def _review_installed_release_origin( *, expected_by_package: Mapping[str, tuple[dict[str, Any], Mapping[str, Any]]], installed_artifacts: Mapping[str, dict[str, Any]], installed_evidence: dict[str, Any], installed_evidence_sha256: str, installed_composition_valid: bool, observation_trusted: bool, catalog_artifacts: object, catalog_sha256: str | None, catalog_channel: str | None, catalog_sequence: int | None, catalog_trusted: bool, receipt: dict[str, Any] | None, receipt_schema: dict[str, Any] | None, authority_keyring: dict[str, Any] | None, authority_keyring_schema: dict[str, Any] | None, verification_time: datetime, forbidden_public_keys: frozenset[bytes], ) -> tuple[dict[str, Any], tuple[EvidenceFinding, ...]]: findings: list[EvidenceFinding] = [] expected_count = len(expected_by_package) scope: dict[str, Any] = { "checked": False, "valid": None, "release_origin_bound": False, "anchored_artifact_digest_count": 0, "signed_installer_receipt_count": 0, "expected_distribution_count": expected_count, "catalog_sha256": catalog_sha256, "installer_receipt_sha256": None, "required_evidence": ( "A signed catalog identity computed from each built immutable artifact, or a role-scoped installer receipt signed by an independently provisioned authority.", ), } manifest_supplied = catalog_artifacts is not None receipt_supplied = receipt is not None if not manifest_supplied and not receipt_supplied: return scope, () scope["checked"] = True scope["valid"] = False if not installed_composition_valid or catalog_trusted is not True: findings.append( EvidenceFinding( "blocker", "installed_origin_dependency_untrusted", "Installed release-origin proof requires a trusted catalog and valid fresh installed-composition evidence.", ("assessment.release", "assessment.installed_composition"), ) ) return scope, tuple(findings) catalog_by_package, catalog_findings = _catalog_artifact_identities( catalog_artifacts ) findings.extend(catalog_findings) if catalog_findings: return scope, tuple(findings) direct_bound: set[str] = set() receipt_required: set[str] = set() for package_name, (component, catalog_entry) in sorted(expected_by_package.items()): expected_version = str(catalog_entry.get("version") or "") identity = catalog_by_package.get(package_name) if identity is None or identity.get("package_version") != expected_version: findings.append( EvidenceFinding( "review", "installed_origin_catalog_artifact_missing", f"The signed catalog has no built-artifact identity for {package_name!r} at the selected version.", (f"composition.{component['module_id']}",), ) ) continue if identity.get("requires_installer_receipt") is True: receipt_required.add(package_name) continue if not observation_trusted: continue installed = installed_artifacts.get(package_name) record = installed.get("record_integrity") if isinstance(installed, dict) else None observed = ( record.get("artifact_payload_identity") if isinstance(record, dict) else None ) expected = identity.get("installed_payload") if observed == expected: direct_bound.add(package_name) else: findings.append( EvidenceFinding( "blocker", "installed_origin_artifact_payload_mismatch", f"Installed payload for {package_name!r} does not match the identity independently computed from the signed catalog artifact.", (f"composition.{component['module_id']}",), ) ) receipt_bound: set[str] = set() if receipt_supplied: receipt_bound, receipt_findings, receipt_digest = _verify_installer_receipt( expected_by_package=expected_by_package, installed_artifacts=installed_artifacts, installed_evidence=installed_evidence, installed_evidence_sha256=installed_evidence_sha256, catalog_by_package=catalog_by_package, catalog_sha256=catalog_sha256, catalog_channel=catalog_channel, catalog_sequence=catalog_sequence, receipt=receipt, receipt_schema=receipt_schema, authority_keyring=authority_keyring, authority_keyring_schema=authority_keyring_schema, verification_time=verification_time, forbidden_public_keys=forbidden_public_keys, ) findings.extend(receipt_findings) scope["installer_receipt_sha256"] = receipt_digest elif receipt_required: findings.append( EvidenceFinding( "review", "installed_origin_receipt_required", "At least one selected wheel produces installer-transformed files, so its signed installer receipt is required for complete origin proof.", ("assessment.installed_composition",), ) ) bound = direct_bound | receipt_bound scope["anchored_artifact_digest_count"] = len(direct_bound) scope["signed_installer_receipt_count"] = len(receipt_bound) valid = len(bound) == expected_count and not any( finding.severity == "blocker" for finding in findings ) scope["valid"] = valid scope["release_origin_bound"] = valid return scope, tuple(findings) def _catalog_artifact_identities( value: object, ) -> tuple[dict[str, dict[str, Any]], tuple[EvidenceFinding, ...]]: if value is None: return {}, () if not isinstance(value, list) or len(value) > MAX_GOVOPLAN_DISTRIBUTIONS: return {}, ( EvidenceFinding( "blocker", "catalog_artifact_manifest_invalid", "Signed catalog release artifacts must be a bounded list.", ("assessment.release",), ), ) result: dict[str, dict[str, Any]] = {} invalid = False for item in value: if not isinstance(item, dict) or set(item) != { "artifact_kind", "package_name", "package_version", "archive_sha256", "archive_size", "installed_payload", "requires_installer_receipt", }: invalid = True continue package_name = item.get("package_name") version = item.get("package_version") archive_size = item.get("archive_size") payload = item.get("installed_payload") if ( item.get("artifact_kind") != "python-wheel" or not isinstance(package_name, str) or PACKAGE_PATTERN.fullmatch(package_name) is None or not isinstance(version, str) or VERSION_PATTERN.fullmatch(version) is None or not isinstance(item.get("archive_sha256"), str) or SHA256_PATTERN.fullmatch(str(item["archive_sha256"])) is None or not isinstance(archive_size, int) or isinstance(archive_size, bool) or not 0 < archive_size <= MAX_HASHED_DISTRIBUTION_BYTES or not isinstance(item.get("requires_installer_receipt"), bool) or not _payload_identity_semantics_valid( payload, algorithm=WHEEL_PAYLOAD_ALGORITHM, maximum_files=MAX_DISTRIBUTION_FILES, ) or package_name in result ): invalid = True continue result[package_name] = item if invalid: return {}, ( EvidenceFinding( "blocker", "catalog_artifact_manifest_invalid", "Signed catalog release artifacts contain malformed or duplicate package identities.", ("assessment.release",), ), ) return result, () def _verify_installer_receipt( *, expected_by_package: Mapping[str, tuple[dict[str, Any], Mapping[str, Any]]], installed_artifacts: Mapping[str, dict[str, Any]], installed_evidence: dict[str, Any], installed_evidence_sha256: str, catalog_by_package: Mapping[str, dict[str, Any]], catalog_sha256: str | None, catalog_channel: str | None, catalog_sequence: int | None, receipt: dict[str, Any], receipt_schema: dict[str, Any] | None, authority_keyring: dict[str, Any] | None, authority_keyring_schema: dict[str, Any] | None, verification_time: datetime, forbidden_public_keys: frozenset[bytes], ) -> tuple[set[str], tuple[EvidenceFinding, ...], str]: receipt_digest = canonical_sha256(receipt) findings: list[EvidenceFinding] = [] if receipt_schema is None or authority_keyring_schema is None: findings.append( EvidenceFinding( "blocker", "installer_receipt_schema_missing", "Installer receipt proof requires both bounded validation schemas.", ("assessment.installed_composition",), ) ) return set(), tuple(findings), receipt_digest receipt_errors = validate_payload(payload=receipt, schema=receipt_schema) keyring_errors = ( validate_payload(payload=authority_keyring, schema=authority_keyring_schema) if isinstance(authority_keyring, dict) else ("$: an independently provisioned installer authority keyring is required",) ) findings.extend( EvidenceFinding( "blocker", "installer_receipt_schema", message, ("assessment.installed_composition",), ) for message in receipt_errors ) findings.extend( EvidenceFinding( "blocker", "installer_authority_keyring_schema", message, ("assessment.installed_composition",), ) for message in keyring_errors ) if receipt_errors or keyring_errors or not isinstance(authority_keyring, dict): return set(), tuple(findings), receipt_digest expected_release = installed_evidence.get("assessment_release") expected_assessment = installed_evidence.get("assessment_id") catalog_binding = receipt.get("catalog") binding_valid = True if ( receipt.get("assessment_id") != expected_assessment or receipt.get("assessment_release") != expected_release or receipt.get("installed_evidence_sha256") != installed_evidence_sha256 ): binding_valid = False findings.append( EvidenceFinding( "blocker", "installer_receipt_evidence_mismatch", "Installer receipt is not bound to the supplied assessment and installed-evidence digest.", ("assessment.installed_composition",), ) ) if not isinstance(catalog_binding, dict) or ( catalog_binding.get("sha256") != catalog_sha256 or catalog_binding.get("channel") != catalog_channel or catalog_binding.get("sequence") != catalog_sequence ): binding_valid = False findings.append( EvidenceFinding( "blocker", "installer_receipt_catalog_mismatch", "Installer receipt is not bound to the supplied signed catalog identity.", ("assessment.release", "assessment.installed_composition"), ) ) issued_at = _datetime(receipt.get("issued_at")) collected_at = _datetime(installed_evidence.get("collected_at")) if ( issued_at is None or collected_at is None or issued_at < collected_at or issued_at - collected_at > MAX_LOCAL_OBSERVATION_AGE or issued_at > verification_time + MAX_LOCAL_OBSERVATION_FUTURE_SKEW or verification_time - issued_at > MAX_LOCAL_OBSERVATION_AGE ): binding_valid = False findings.append( EvidenceFinding( "blocker", "installer_receipt_time_invalid", "Installer receipt issuance must follow the observation within five minutes and remain fresh at the selected verification time.", ("assessment.installed_composition",), ) ) receipt_rows = [ item for item in receipt.get("artifacts", []) if isinstance(item, dict) ] rows_by_package: dict[str, dict[str, Any]] = {} duplicate_packages: set[str] = set() for row in receipt_rows: package_name = str(row.get("package_name") or "") if package_name in rows_by_package: duplicate_packages.add(package_name) else: rows_by_package[package_name] = row expected_packages = set(expected_by_package) if duplicate_packages or set(rows_by_package) != expected_packages: binding_valid = False findings.append( EvidenceFinding( "blocker", "installer_receipt_artifact_set_mismatch", "Installer receipt must bind exactly one row for every expected installed distribution.", ("assessment.installed_composition",), ) ) bound: set[str] = set() for package_name, (_, catalog_entry) in sorted(expected_by_package.items()): row = rows_by_package.get(package_name) installed = installed_artifacts.get(package_name) record = installed.get("record_integrity") if isinstance(installed, dict) else None observed_payload = ( record.get("installed_payload_identity") if isinstance(record, dict) else None ) catalog_identity = catalog_by_package.get(package_name) if ( row is None or catalog_identity is None or row.get("package_version") != catalog_entry.get("version") or row.get("catalog_archive_sha256") != catalog_identity.get("archive_sha256") or row.get("installed_payload") != observed_payload ): binding_valid = False findings.append( EvidenceFinding( "blocker", "installer_receipt_artifact_mismatch", f"Installer receipt does not bind the observed {package_name!r} payload to its signed catalog artifact.", ("assessment.installed_composition",), ) ) else: bound.add(package_name) keys, key_findings = _installer_receipt_keys( authority_keyring=authority_keyring, issued_at=issued_at, forbidden_public_keys=forbidden_public_keys, ) findings.extend(key_findings) signature_payload = dict(receipt) signature_payload.pop("signatures", None) signature_bytes = canonical_bytes(signature_payload) signature_valid = False seen_ids: set[str] = set() duplicate_ids = False for signature in receipt.get("signatures", []): if not isinstance(signature, dict): continue key_id = str(signature.get("key_id") or "") if key_id in seen_ids: duplicate_ids = True continue seen_ids.add(key_id) public_key = keys.get(key_id) if public_key is None or signature.get("algorithm") != "ed25519": continue try: encoded = base64.b64decode(str(signature.get("value") or ""), validate=True) Ed25519PublicKey.from_public_bytes(public_key).verify( encoded, signature_bytes ) signature_valid = True except (ValueError, InvalidSignature): continue if duplicate_ids or not signature_valid: findings.append( EvidenceFinding( "blocker", "installer_receipt_signature_untrusted", "Installer receipt has no unique valid signature from an authorized independent installer key.", ("assessment.installed_composition",), ) ) if not binding_valid or not signature_valid or key_findings: return set(), tuple(findings), receipt_digest return bound, tuple(findings), receipt_digest def _installer_receipt_keys( *, authority_keyring: dict[str, Any], issued_at: datetime | None, forbidden_public_keys: frozenset[bytes], ) -> tuple[dict[str, bytes], tuple[EvidenceFinding, ...]]: findings: list[EvidenceFinding] = [] result: dict[str, bytes] = {} raw_keys = authority_keyring.get("keys") if not isinstance(raw_keys, list) or issued_at is None: return {}, ( EvidenceFinding( "blocker", "installer_authority_untrusted", "Installer receipt authority keyring or issuance time is invalid.", ("assessment.installed_composition",), ), ) seen_ids: set[str] = set() seen_material: set[bytes] = set() malformed = False for item in raw_keys: if not isinstance(item, dict): malformed = True continue key_id = str(item.get("key_id") or "") try: public_key = base64.b64decode(str(item.get("public_key") or ""), validate=True) Ed25519PublicKey.from_public_bytes(public_key) except ValueError: malformed = True continue not_before = _datetime(item.get("not_before")) not_after = _datetime(item.get("not_after")) interval_valid = not ( not_before is not None and not_after is not None and not_before >= not_after ) if ( OPAQUE_ID_PATTERN.fullmatch(key_id) is None or key_id in seen_ids or public_key in seen_material or public_key in forbidden_public_keys or item.get("allowed_scopes") != ["installed_release_origin"] or not interval_valid ): malformed = True continue seen_ids.add(key_id) seen_material.add(public_key) status = item.get("status") active_at_issue = ( (not_before is None or issued_at >= not_before) and (not_after is None or issued_at < not_after) ) historically_retired = status == "retired" and not_after is not None if active_at_issue and ( status in {"active", "next"} or historically_retired ): result[key_id] = public_key if malformed or not result: findings.append( EvidenceFinding( "blocker", "installer_authority_untrusted", "Installer authority keyring is malformed, reuses another trust-domain key, or has no role-scoped key valid at receipt issuance.", ("assessment.installed_composition",), ) ) return {}, tuple(findings) return result, () def review_boundary_evidence( *, assessment: dict[str, Any], installed_review: InstalledEvidenceReview, evidence: dict[str, Any] | None, evidence_schema: dict[str, Any] | None, authority_keyring: dict[str, Any] | None, authority_keyring_schema: dict[str, Any] | None, verification_time: datetime | None = None, verification_mode: str = "current", expected_external_provider_subject: str | None = None, forbidden_authority_public_keys: frozenset[bytes] = frozenset(), ) -> BoundaryEvidenceReview: deployment_profile = assessment.get("deployment_profile") raw_deployment_subject = ( deployment_profile.get("id") if isinstance(deployment_profile, dict) else None ) deployment_subject = _safe_opaque_id_or_none(raw_deployment_subject) provider_subject = _safe_opaque_id_or_none(expected_external_provider_subject) effective_verification_mode = ( verification_mode if verification_mode in VERIFICATION_MODES else "invalid" ) expected_subjects = { "target_environment": deployment_subject, "external_providers": provider_subject, "production_approval": deployment_subject, } scopes = _unchecked_boundary_scope(expected_subjects=expected_subjects) findings: list[EvidenceFinding] = [] if raw_deployment_subject is not None and deployment_subject is None: findings.append( EvidenceFinding( "blocker", "boundary_deployment_subject_invalid", "The assessment deployment profile has no valid opaque subject ID.", ("assessment.deployment_profile",), ) ) if expected_external_provider_subject is not None and provider_subject is None: findings.append( EvidenceFinding( "blocker", "boundary_provider_subject_invalid", "The expected external-provider subject must be a bounded opaque ID.", ("assessment.external_proof",), ) ) if effective_verification_mode == "invalid": findings.append( EvidenceFinding( "blocker", "boundary_verification_mode_invalid", "Boundary evidence verification mode is invalid.", ("assessment.external_proof",), ) ) if evidence is None: return BoundaryEvidenceReview(findings=tuple(findings), proof_scope=scopes) now = verification_time or datetime.now(tz=UTC) if now.tzinfo is None: now = now.replace(tzinfo=UTC) scopes["boundary_evidence_verification"] = { "checked": True, "evaluated_at": _iso_datetime(now), "verification_mode": effective_verification_mode, "historical": effective_verification_mode == "historical_override", } if evidence_schema is None or authority_keyring_schema is None: findings.append( EvidenceFinding( "blocker", "boundary_evidence_schema_missing", "Boundary evidence was supplied without both validation schemas.", ("assessment.external_proof",), ) ) return BoundaryEvidenceReview(tuple(findings), scopes) evidence_errors = validate_payload(payload=evidence, schema=evidence_schema) keyring_errors = ( validate_payload(payload=authority_keyring, schema=authority_keyring_schema) if isinstance(authority_keyring, dict) else ("$: an independently provisioned proof-authority keyring is required",) ) findings.extend( EvidenceFinding( "blocker", "boundary_evidence_schema", message, ("assessment.external_proof",), ) for message in evidence_errors ) findings.extend( EvidenceFinding( "blocker", "boundary_authority_keyring_schema", message, ("assessment.external_proof",), ) for message in keyring_errors ) if evidence_errors or keyring_errors or not isinstance(authority_keyring, dict): return BoundaryEvidenceReview(tuple(findings), scopes) expected_release = ( assessment.get("release", {}).get("ref") if isinstance(assessment.get("release"), dict) else None ) if evidence.get("assessment_id") != assessment.get("assessment_id"): findings.append( EvidenceFinding( "blocker", "boundary_assessment_mismatch", "Boundary evidence is bound to a different assessment ID.", ("assessment.external_proof",), ) ) if evidence.get("assessment_release") != expected_release: findings.append( EvidenceFinding( "blocker", "boundary_release_mismatch", "Boundary evidence is bound to a different assessment release.", ("assessment.release", "assessment.external_proof"), ) ) installed_digest = installed_review.evidence_sha256 installed_scope = installed_review.proof_scope.get("installed_artifacts", {}) installed_origin_scope = installed_review.proof_scope.get( "installed_release_origin", {} ) if ( installed_digest is None or evidence.get("installed_evidence_sha256") != installed_digest or installed_scope.get("valid") is not True ): findings.append( EvidenceFinding( "blocker", "boundary_installed_evidence_untrusted", "Boundary evidence must bind to the supplied, valid installed-composition evidence digest.", ("assessment.installed_composition", "assessment.external_proof"), ) ) if installed_origin_scope.get("valid") is not True: findings.append( EvidenceFinding( "blocker", "boundary_installed_release_origin_untrusted", "Boundary evidence requires an independently anchored digest or signed installer receipt for every installed release artifact; local PEP 610 and RECORD metadata are insufficient.", ("assessment.installed_composition", "assessment.external_proof"), ) ) issued_at = _datetime(evidence.get("issued_at")) expires_at = _datetime(evidence.get("expires_at")) time_valid = True if issued_at is None or expires_at is None or issued_at >= expires_at: time_valid = False findings.append( EvidenceFinding( "blocker", "boundary_evidence_interval_invalid", "Boundary evidence must have issued_at earlier than expires_at.", ("assessment.external_proof",), ) ) elif now < issued_at or now >= expires_at: time_valid = False findings.append( EvidenceFinding( "blocker", "boundary_evidence_not_current", "Boundary evidence is not valid at the verification time.", ("assessment.external_proof",), ) ) claims = [item for item in evidence.get("claims", []) if isinstance(item, dict)] claims_by_scope: dict[str, dict[str, Any]] = {} duplicate_scopes: set[str] = set() semantic_invalid: set[str] = set() subject_invalid: set[str] = set() for claim in claims: scope = str(claim.get("scope") or "") if scope in claims_by_scope: duplicate_scopes.add(scope) else: claims_by_scope[scope] = claim allowed_results = {POSITIVE_RESULTS.get(scope), NEGATIVE_RESULTS.get(scope)} if claim.get("result") not in allowed_results: semantic_invalid.add(scope) controls = claim.get("control_ids") if not isinstance(controls, list) or len(controls) != len(set(controls)): semantic_invalid.add(scope) artifacts = claim.get("artifacts") artifact_ids = [ str(item.get("artifact_id") or "") for item in artifacts or [] if isinstance(item, dict) ] if len(artifact_ids) != len(set(artifact_ids)): semantic_invalid.add(scope) observed_subject = str(claim.get("subject_id") or "") expected_subject = expected_subjects.get(scope) if scope in scopes: scopes[scope]["observed_subject_id"] = observed_subject if expected_subject is None or observed_subject != expected_subject: subject_invalid.add(scope) for scope in sorted(duplicate_scopes): findings.append( EvidenceFinding( "blocker", "boundary_claim_duplicate", f"Boundary evidence repeats claim scope {scope!r}.", ("assessment.external_proof",), ) ) for scope in sorted(semantic_invalid): findings.append( EvidenceFinding( "blocker", "boundary_claim_semantics", f"Boundary claim {scope!r} has an invalid result or duplicate control/artifact IDs.", ("assessment.external_proof",), ) ) for scope in sorted(subject_invalid): expected_subject = expected_subjects.get(scope) findings.append( EvidenceFinding( "blocker", ( "boundary_provider_subject_unconfigured" if scope == "external_providers" and expected_subject is None else "boundary_claim_subject_mismatch" ), ( "External-provider proof requires an explicit expected opaque subject ID." if scope == "external_providers" and expected_subject is None else f"Boundary claim {scope!r} is not bound to its expected assessment subject." ), ("assessment.external_proof",), ) ) authority_keys, key_findings = _active_authority_keys( authority_keyring=authority_keyring, verification_time=now, forbidden_public_keys=forbidden_authority_public_keys, ) findings.extend(key_findings) signature_payload = dict(evidence) signature_payload.pop("signatures", None) signature_bytes = canonical_bytes(signature_payload) authorized_scopes: set[str] = set() seen_signature_key_ids: set[str] = set() duplicate_signature_key_ids: set[str] = set() for signature in evidence.get("signatures", []): if not isinstance(signature, dict): continue key_id = str(signature.get("key_id") or "") if key_id in seen_signature_key_ids: duplicate_signature_key_ids.add(key_id) continue seen_signature_key_ids.add(key_id) key = authority_keys.get(key_id) if key is None or signature.get("algorithm") != "ed25519": continue try: signature_value = base64.b64decode( str(signature.get("value") or ""), validate=True ) Ed25519PublicKey.from_public_bytes(key[0]).verify( signature_value, signature_bytes ) except (ValueError, InvalidSignature): continue authorized_scopes.update(key[1]) if duplicate_signature_key_ids: findings.append( EvidenceFinding( "blocker", "boundary_signature_duplicate", "Boundary evidence repeats one or more signature key IDs.", ("assessment.external_proof",), ) ) unauthorized_scopes = { scope for scope in claims_by_scope if scope in PROOF_REQUIREMENTS and scope not in authorized_scopes } for scope in sorted(unauthorized_scopes): findings.append( EvidenceFinding( "blocker", "boundary_authority_untrusted", f"No valid independently trusted signature is authorized for {scope!r}.", ("assessment.external_proof",), ) ) global_valid = not any(item.severity == "blocker" for item in findings) evidence_digest = canonical_sha256(evidence) review_targets: list[dict[str, Any]] = [] for scope, claim in sorted(claims_by_scope.items()): if scope not in PROOF_REQUIREMENTS: continue if scope in unauthorized_scopes: continue if ( scope in duplicate_scopes or scope in semantic_invalid or scope in subject_invalid or not time_valid ): continue if not global_valid: continue result = str(claim.get("result")) valid = result == POSITIVE_RESULTS[scope] scopes[scope] = { "checked": True, "valid": valid, "evidence_sha256": evidence_digest, "result": result, "control_count": len(claim.get("control_ids") or []), "artifact_count": len(claim.get("artifacts") or []), "expected_subject_id": expected_subjects[scope], "observed_subject_id": claim.get("subject_id"), "evaluated_at": _iso_datetime(now), "verification_mode": effective_verification_mode, "required_evidence": PROOF_REQUIREMENTS[scope], } if not valid: findings.append( EvidenceFinding( "review", "boundary_claim_negative", f"Authorized boundary claim {scope!r} has result {result!r}.", ("assessment.external_proof",), ) ) subject = str(expected_subjects[scope]) review_targets.append( { "id": f"boundary.{scope}.{subject}", "section": scope, "reason": f"Authorized boundary evidence reported {result!r} for {scope!r}.", } ) return BoundaryEvidenceReview( tuple(findings), scopes, tuple( { item["id"]: item for item in sorted(review_targets, key=lambda value: value["id"]) }.values() ), ) def normalize_package_name(value: str) -> str: return re.sub(r"[-_.]+", "-", value.strip().lower()) def canonical_bytes(payload: object) -> bytes: return json.dumps( payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True ).encode("utf-8") def canonical_sha256(payload: object) -> str: return hashlib.sha256(canonical_bytes(payload)).hexdigest() def public_key_material_from_keyrings( *payloads: Mapping[str, Any], ) -> frozenset[bytes]: result: set[bytes] = set() for payload in payloads: keys = payload.get("keys") values = ( [item.get("public_key") for item in keys if isinstance(item, dict)] if isinstance(keys, list) else list(payload.values()) ) for value in values: if not isinstance(value, str): continue try: decoded = base64.b64decode(value, validate=True) Ed25519PublicKey.from_public_bytes(decoded) except ValueError: continue result.add(decoded) return frozenset(result) def _distribution_version(distribution: metadata.Distribution) -> str: try: value = str(distribution.version) except Exception: return "" if len(value) > 128 or VERSION_PATTERN.fullmatch(value) is None: return "" return value def _distribution_modules( *, package_name: str, distribution: metadata.Distribution ) -> tuple[list[dict[str, str]], list[dict[str, str]]]: modules: list[dict[str, str]] = [] issues: list[dict[str, str]] = [] try: all_entry_points = [] for entry_point in distribution.entry_points: if entry_point.group != "govoplan.modules": continue name = str(entry_point.name) value = str(entry_point.value) if ( len(name) > 160 or OPAQUE_ID_PATTERN.fullmatch(name) is None or len(value) > 512 or ENTRY_POINT_VALUE_PATTERN.fullmatch(value) is None ): issues.append( { "code": "module-entry-point-invalid", "package_name": package_name, } ) continue all_entry_points.append((name, value, entry_point)) all_entry_points.sort(key=lambda item: (item[0], item[1])) except Exception: return [], [ { "code": "distribution-metadata-invalid", "package_name": package_name, } ] if len(all_entry_points) > 16: issues.append( { "code": "module-entry-point-limit-exceeded", "package_name": package_name, } ) for _, _, entry_point in all_entry_points[:16]: try: manifest = entry_point.load()() module_id = str(manifest.id) manifest_version = str(manifest.version) except (Exception, SystemExit): issues.append( { "code": "module-entry-point-load-failed", "package_name": package_name, } ) continue if ( len(module_id) > 160 or OPAQUE_ID_PATTERN.fullmatch(module_id) is None or len(manifest_version) > 128 or VERSION_PATTERN.fullmatch(manifest_version) is None ): issues.append( { "code": "module-entry-point-invalid", "package_name": package_name, } ) continue modules.append({"module_id": module_id, "manifest_version": manifest_version}) return sorted(modules, key=lambda item: item["module_id"]), issues def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]: try: raw = distribution.read_text("direct_url.json") except Exception: return _local_source_metadata("malformed-direct-url") if raw is None: return _local_source_metadata("index-or-unknown") if not isinstance(raw, str) or len(raw) > MAX_DIRECT_URL_BYTES: return _local_source_metadata("malformed-direct-url") if len(raw.encode("utf-8")) > MAX_DIRECT_URL_BYTES: return _local_source_metadata("malformed-direct-url") try: direct_url = json.loads(raw) except (TypeError, json.JSONDecodeError): return _local_source_metadata("malformed-direct-url") if not isinstance(direct_url, dict): return _local_source_metadata("malformed-direct-url") raw_url = direct_url.get("url") if ( not isinstance(raw_url, str) or not raw_url or len(raw_url) > 2048 or any(character.isspace() or ord(character) < 32 for character in raw_url) ): return _local_source_metadata("malformed-direct-url") try: parsed_url = urlsplit(raw_url) except ValueError: return _local_source_metadata("malformed-direct-url") if parsed_url.password is not None or parsed_url.query or parsed_url.fragment: return _local_source_metadata("malformed-direct-url") vcs_info = direct_url.get("vcs_info") directory_info = direct_url.get("dir_info") archive_info = direct_url.get("archive_info") provenance_forms = tuple( name for name, value in ( ("vcs", vcs_info), ("directory", directory_info), ("archive", archive_info), ) if value is not None ) if len(provenance_forms) != 1: return _local_source_metadata("malformed-direct-url") if isinstance(vcs_info, dict): if vcs_info.get("vcs") != "git": return _local_source_metadata("malformed-direct-url") commit = str(vcs_info.get("commit_id") or "").lower() requested_revision = vcs_info.get("requested_revision") if ( GIT_OBJECT_PATTERN.fullmatch(commit) is None or parsed_url.scheme not in {"file", "git", "git+https", "git+ssh", "https", "ssh"} or not parsed_url.path or ( parsed_url.scheme != "file" and (not parsed_url.hostname or not parsed_url.path.endswith(".git")) ) or (parsed_url.scheme in {"https", "git+https"} and parsed_url.username) or ( requested_revision is not None and ( not isinstance(requested_revision, str) or VCS_REVISION_PATTERN.fullmatch(requested_revision) is None ) ) ): return _local_source_metadata("malformed-direct-url") return _local_source_metadata("vcs-commit", commit=commit) if isinstance(archive_info, dict): if ( parsed_url.scheme not in {"file", "https"} or not parsed_url.path or ( parsed_url.scheme == "https" and (not parsed_url.hostname or parsed_url.username) ) ): return _local_source_metadata("malformed-direct-url") hashes = archive_info.get("hashes") digest = hashes.get("sha256") if isinstance(hashes, dict) else None if digest is None: legacy = str(archive_info.get("hash") or "") if legacy.startswith("sha256="): digest = legacy.removeprefix("sha256=") digest_text = str(digest or "").lower() if SHA256_PATTERN.fullmatch(digest_text): return _local_source_metadata("archive", sha256=digest_text) return _local_source_metadata("malformed-direct-url") if isinstance(directory_info, dict): editable = directory_info.get("editable", False) if ( parsed_url.scheme != "file" or not parsed_url.path or not isinstance(editable, bool) ): return _local_source_metadata("malformed-direct-url") return _local_source_metadata( "editable-local" if editable else "local-directory" ) return _local_source_metadata("malformed-direct-url") def _local_source_metadata(kind: str, **values: str) -> dict[str, str]: return {"basis": "local-pep610-metadata", "kind": kind, **values} def _read_record_entries( *, distribution: metadata.Distribution, distribution_root: Path, max_entries: int, ) -> tuple[tuple[RecordEntry, ...], int, str]: """Read RECORD itself without importlib's missing-path filtering. ``Distribution.files`` is an interpreted convenience API rather than the raw declaration being attested. Reading the metadata file directly lets this collector retain missing and malformed declarations and enforce byte and row limits before resolving any untrusted payload path. """ metadata_root_value = getattr(distribution, "_path", None) if metadata_root_value is None: return (), 0, "unavailable" try: metadata_root = Path(os.fspath(metadata_root_value)) if metadata_root.is_symlink(): return (), 0, "unavailable" resolved_metadata_root = metadata_root.resolve() except (OSError, TypeError, ValueError): return (), 0, "unavailable" if not _is_relative_to(resolved_metadata_root, distribution_root): return (), 0, "unavailable" record_path = resolved_metadata_root / "RECORD" try: if record_path.is_symlink(): return (), 0, "unavailable" encoded, exceeded, stable = _bounded_regular_file_bytes( record_path, max_bytes=MAX_RECORD_BYTES, ) except OSError: return (), 0, "unavailable" if exceeded: return (), 0, "limit-exceeded" if not stable: return (), 1, "unavailable" try: text = encoded.decode("utf-8") except UnicodeDecodeError: return (), 1, "unavailable" entries: list[RecordEntry] = [] malformed_rows = 0 declared_paths: set[str] = set() try: rows = csv.reader(io.StringIO(text, newline="")) for row in rows: if len(entries) + malformed_rows >= max_entries: return (), 0, "limit-exceeded" if len(row) != 3: malformed_rows += 1 continue raw_path, raw_hash, raw_size = row if ( not raw_path or len(raw_path) > MAX_RECORD_PATH_CHARACTERS or any(ord(character) < 32 for character in raw_path) or (raw_size and (len(raw_size) > 20 or not raw_size.isdecimal())) ): malformed_rows += 1 continue declared_path = str(PurePosixPath(raw_path.replace("\\", "/"))) if declared_path in declared_paths: malformed_rows += 1 continue declared_paths.add(declared_path) if raw_hash: hash_mode, separator, hash_value = raw_hash.partition("=") if not separator or not hash_mode or not hash_value: hash_mode = raw_hash hash_value = "" else: hash_mode = None hash_value = None entries.append( RecordEntry( path=raw_path, hash_mode=hash_mode, hash_value=hash_value, size=int(raw_size) if raw_size else None, ) ) except (csv.Error, ValueError): return (), 1, "unavailable" return tuple(entries), malformed_rows, "ok" def _bounded_regular_file_bytes( path: Path, *, max_bytes: int, ) -> tuple[bytes, bool, bool]: flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0) descriptor = os.open(path, flags) chunks: list[bytes] = [] total = 0 try: initial = os.fstat(descriptor) if not stat.S_ISREG(initial.st_mode): raise OSError("RECORD metadata is not a regular file") if initial.st_size > max_bytes: return b"", True, True while total <= max_bytes: chunk = os.read(descriptor, min(1024 * 1024, max_bytes + 1 - total)) if not chunk: break chunks.append(chunk) total += len(chunk) final = os.fstat(descriptor) finally: os.close(descriptor) exceeded = total > max_bytes or final.st_size > max_bytes stable = ( not exceeded and total == initial.st_size and final.st_size == initial.st_size and (final.st_dev, final.st_ino) == (initial.st_dev, initial.st_ino) and final.st_mtime_ns == initial.st_mtime_ns and final.st_ctime_ns == initial.st_ctime_ns ) return b"".join(chunks), exceeded, stable def _record_integrity( distribution: metadata.Distribution, *, package_name: str, budget: CollectionHashBudget, ) -> dict[str, Any]: counts = { "hashed_file_count": 0, "permitted_unhashed_file_count": 0, "generated_unhashed_file_count": 0, "unverifiable_file_count": 0, "missing_file_count": 0, "mismatched_file_count": 0, } identities = { "artifact_payload_identity": None, "installed_payload_identity": None, } try: distribution_root = Path(distribution.locate_file("")).resolve() installation_root = Path(sys.prefix).resolve() except Exception: return {"status": "unavailable", **counts, **identities} if not _trusted_distribution_root(distribution_root): return {"status": "unavailable", **counts, **identities} entries, malformed_rows, record_status = _read_record_entries( distribution=distribution, distribution_root=distribution_root, max_entries=min(MAX_DISTRIBUTION_FILES, budget.remaining_files), ) if record_status == "limit-exceeded": budget.remaining_files = 0 return {"status": "limit-exceeded", **counts, **identities} if record_status != "ok": counts["unverifiable_file_count"] = malformed_rows return {"status": "unavailable", **counts, **identities} record_file_count = len(entries) + malformed_rows if record_file_count == 0: return {"status": "unavailable", **counts, **identities} budget.remaining_files -= record_file_count counts["unverifiable_file_count"] = malformed_rows declared_scripts = _declared_distribution_scripts(distribution) owner_prefix = package_name.replace("-", "_") resolved_entries: list[tuple[RecordEntry, Path | None]] = [ ( entry, _resolve_record_path( distribution=distribution, package_path=entry.path, distribution_root=distribution_root, installation_root=installation_root, owner_prefix=owner_prefix, declared_scripts=declared_scripts, ), ) for entry in entries ] resolved_path_counts: dict[Path, int] = {} for _, resolved_path in resolved_entries: if resolved_path is not None: resolved_path_counts[resolved_path] = ( resolved_path_counts.get(resolved_path, 0) + 1 ) duplicate_resolved_paths = { path for path, count in resolved_path_counts.items() if count > 1 } counts["unverifiable_file_count"] += sum( resolved_path_counts[path] for path in duplicate_resolved_paths ) hashed_payload_paths = { resolved_path for entry, resolved_path in resolved_entries if resolved_path is not None and resolved_path not in duplicate_resolved_paths and entry.hash_mode == "sha256" } total_hashed_bytes = 0 limit_exceeded = False artifact_payload_rows: list[dict[str, object]] = [] installed_payload_rows: list[dict[str, object]] = [] for entry, resolved_path in resolved_entries: if resolved_path is None: counts["unverifiable_file_count"] += 1 continue if resolved_path in duplicate_resolved_paths: continue try: file_stat = resolved_path.lstat() except OSError: counts["missing_file_count"] += 1 continue if not stat.S_ISREG(file_stat.st_mode): counts["missing_file_count"] += 1 continue if entry.size is not None and file_stat.st_size != entry.size: counts["mismatched_file_count"] += 1 continue if entry.hash_mode is None: if _permitted_unhashed_record_path( resolved_path, distribution_root=distribution_root, owner_prefix=owner_prefix, ): counts["permitted_unhashed_file_count"] += 1 elif _generated_unhashed_pyc( resolved_path, hashed_payload_paths=hashed_payload_paths, ): counts["generated_unhashed_file_count"] += 1 else: counts["unverifiable_file_count"] += 1 continue if entry.hash_mode != "sha256": counts["unverifiable_file_count"] += 1 continue remaining_distribution_bytes = ( MAX_HASHED_DISTRIBUTION_BYTES - total_hashed_bytes ) allowed_bytes = min( MAX_HASHED_FILE_BYTES, remaining_distribution_bytes, budget.remaining_bytes, ) if allowed_bytes <= 0 or file_stat.st_size > allowed_bytes: limit_exceeded = True continue expected = _record_hash_bytes(entry.hash_value or "") if expected is None: counts["unverifiable_file_count"] += 1 continue try: actual, hashed_bytes, exceeded = _bounded_file_sha256( resolved_path, max_bytes=allowed_bytes, ) except OSError: counts["missing_file_count"] += 1 continue total_hashed_bytes += hashed_bytes budget.remaining_bytes -= hashed_bytes if exceeded: limit_exceeded = True continue if hmac.compare_digest(actual, expected): counts["hashed_file_count"] += 1 installed_path = _installed_payload_path( resolved_path, distribution_root=distribution_root, installation_root=installation_root, declared_scripts=declared_scripts, ) if installed_path is None: counts["unverifiable_file_count"] += 1 continue row = { "path": installed_path, "sha256": actual.hex(), "size": file_stat.st_size, } installed_payload_rows.append(row) artifact_path = _artifact_payload_path( resolved_path, distribution_root=distribution_root, installation_root=installation_root, declared_scripts=declared_scripts, ) if artifact_path is not None and not _installer_generated_metadata( resolved_path, distribution_root=distribution_root ): artifact_payload_rows.append({**row, "path": artifact_path}) else: counts["mismatched_file_count"] += 1 if limit_exceeded: status = "limit-exceeded" elif counts["missing_file_count"] or counts["mismatched_file_count"]: status = "mismatch" elif counts["hashed_file_count"] == 0: status = "unavailable" elif counts["unverifiable_file_count"]: status = "partial" else: status = "verified" if status == "verified": installed_identity = _payload_identity( algorithm=INSTALLED_PAYLOAD_ALGORITHM, rows=installed_payload_rows, ) artifact_identity = _payload_identity( algorithm=WHEEL_PAYLOAD_ALGORITHM, rows=artifact_payload_rows, ) if installed_identity is None or artifact_identity is None: status = "partial" counts["unverifiable_file_count"] += 1 else: identities["installed_payload_identity"] = installed_identity identities["artifact_payload_identity"] = artifact_identity return {"status": status, **counts, **identities} def _payload_identity( *, algorithm: str, rows: list[dict[str, object]] ) -> dict[str, object] | None: ordered = sorted(rows, key=lambda item: str(item["path"])) paths = [str(item["path"]) for item in ordered] if not ordered or len(set(paths)) != len(paths): return None return { "algorithm": algorithm, "sha256": payload_identity_sha256(algorithm=algorithm, rows=ordered), "file_count": len(ordered), } def _installed_payload_path( path: Path, *, distribution_root: Path, installation_root: Path, declared_scripts: frozenset[str], ) -> str | None: if _is_relative_to(path, distribution_root): return path.relative_to(distribution_root).as_posix() for script_root in ( (installation_root / "bin").resolve(), (installation_root / "Scripts").resolve(), ): if path.parent == script_root and path.name in declared_scripts: return f"@scripts/{path.name}" if _is_relative_to(path, installation_root): return f"@data/{path.relative_to(installation_root).as_posix()}" return None def _artifact_payload_path( path: Path, *, distribution_root: Path, installation_root: Path, declared_scripts: frozenset[str], ) -> str | None: installed = _installed_payload_path( path, distribution_root=distribution_root, installation_root=installation_root, declared_scripts=declared_scripts, ) return None if installed is None or installed.startswith("@scripts/") else installed def _installer_generated_metadata( path: Path, *, distribution_root: Path ) -> bool: if not _is_relative_to(path, distribution_root): return False relative = path.relative_to(distribution_root) return ( len(relative.parts) >= 2 and relative.parent.name.endswith(".dist-info") and relative.name in {"direct_url.json", "INSTALLER", "REQUESTED"} ) def _resolve_record_path( *, distribution: metadata.Distribution, package_path: object, distribution_root: Path, installation_root: Path, owner_prefix: str, declared_scripts: frozenset[str], ) -> Path | None: record_path = PurePosixPath(str(package_path).replace("\\", "/")) if record_path.is_absolute() or not record_path.parts: return None try: unresolved_path = Path(distribution.locate_file(package_path)) if unresolved_path.is_symlink(): return None resolved_path = unresolved_path.resolve() except (OSError, TypeError, ValueError): return None has_parent_traversal = ".." in record_path.parts if not has_parent_traversal: return ( resolved_path if _is_relative_to(resolved_path, distribution_root) else None ) script_roots = ( (installation_root / "bin").resolve(), (installation_root / "Scripts").resolve(), ) for script_root in script_roots: if ( resolved_path.parent == script_root and resolved_path.name in declared_scripts ): return resolved_path if not _is_relative_to(resolved_path, installation_root): return None relative = resolved_path.relative_to(installation_root) if not relative.parts: return None owned_root = relative.parts[0] if owned_root == owner_prefix or owned_root.startswith(f"{owner_prefix}_"): return resolved_path return None def _declared_distribution_scripts( distribution: metadata.Distribution, ) -> frozenset[str]: scripts: set[str] = set() try: entry_points = distribution.entry_points except Exception: return frozenset() for entry_point in entry_points: if entry_point.group != "console_scripts": continue name = str(entry_point.name) if len(name) > 160 or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", name) is None: continue scripts.update((name, f"{name}.exe", f"{name}-script.py")) return frozenset(scripts) def _bounded_file_sha256(path: Path, *, max_bytes: int) -> tuple[bytes, int, bool]: flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) descriptor = os.open(path, flags) digest = hashlib.sha256() total = 0 exceeded = False try: file_stat = os.fstat(descriptor) if not stat.S_ISREG(file_stat.st_mode): raise OSError("RECORD path is not a regular file") if file_stat.st_size > max_bytes: return digest.digest(), 0, True expected_size = file_stat.st_size with os.fdopen(descriptor, "rb", closefd=False) as handle: while total < expected_size: chunk = handle.read(min(1024 * 1024, expected_size - total)) if not chunk: break digest.update(chunk) total += len(chunk) final_stat = os.fstat(descriptor) exceeded = total != expected_size or final_stat.st_size != expected_size finally: os.close(descriptor) return digest.digest(), total, exceeded def _record_hash_bytes(value: str) -> bytes | None: try: decoded = base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)) except (ValueError, TypeError): return None return decoded if len(decoded) == hashlib.sha256().digest_size else None def _record_integrity_semantics_valid(record: Mapping[str, Any]) -> bool: count_names = ( "hashed_file_count", "permitted_unhashed_file_count", "generated_unhashed_file_count", "unverifiable_file_count", "missing_file_count", "mismatched_file_count", ) counts: dict[str, int] = {} for name in count_names: value = record.get(name) if isinstance(value, bool) or not isinstance(value, int) or value < 0: return False counts[name] = value if sum(counts.values()) > MAX_DISTRIBUTION_FILES: return False status = record.get("status") hashed = counts["hashed_file_count"] unverifiable = counts["unverifiable_file_count"] missing = counts["missing_file_count"] mismatched = counts["mismatched_file_count"] if status == "verified": artifact_identity = record.get("artifact_payload_identity") installed_identity = record.get("installed_payload_identity") return ( hashed > 0 and counts["permitted_unhashed_file_count"] > 0 and unverifiable == 0 and missing == 0 and mismatched == 0 and _payload_identity_semantics_valid( artifact_identity, algorithm=WHEEL_PAYLOAD_ALGORITHM, maximum_files=hashed, ) and _payload_identity_semantics_valid( installed_identity, algorithm=INSTALLED_PAYLOAD_ALGORITHM, maximum_files=hashed, exact_files=hashed, ) ) if status == "partial": return hashed > 0 and unverifiable > 0 and missing == 0 and mismatched == 0 if status == "mismatch": return missing + mismatched > 0 if status == "unavailable": return hashed == 0 and missing == 0 and mismatched == 0 if status == "limit-exceeded": return True return False def _payload_identity_semantics_valid( value: object, *, algorithm: str, maximum_files: int, exact_files: int | None = None, ) -> bool: if not isinstance(value, Mapping) or set(value) != { "algorithm", "sha256", "file_count", }: return False file_count = value.get("file_count") return ( value.get("algorithm") == algorithm and isinstance(value.get("sha256"), str) and SHA256_PATTERN.fullmatch(str(value["sha256"])) is not None and isinstance(file_count, int) and not isinstance(file_count, bool) and 0 < file_count <= maximum_files and (exact_files is None or file_count == exact_files) ) def _permitted_unhashed_record_path( path: Path, *, distribution_root: Path, owner_prefix: str, ) -> bool: if not _is_relative_to(path, distribution_root): return False if path.name not in {"RECORD", "RECORD.jws", "RECORD.p7s"}: return False parent = path.parent.name return parent.endswith(".dist-info") and ( parent == f"{owner_prefix}.dist-info" or parent.startswith(f"{owner_prefix}-") ) def _generated_unhashed_pyc( path: Path, *, hashed_payload_paths: set[Path], ) -> bool: if path.suffix != ".pyc" or "__pycache__" not in path.parts: return False try: source = Path(source_from_cache(str(path))).resolve() except (ValueError, OSError): return False return source in hashed_payload_paths def _active_authority_keys( *, authority_keyring: dict[str, Any], verification_time: datetime, forbidden_public_keys: frozenset[bytes], ) -> tuple[dict[str, tuple[bytes, frozenset[str]]], tuple[EvidenceFinding, ...]]: result: dict[str, tuple[bytes, frozenset[str]]] = {} findings: list[EvidenceFinding] = [] seen: set[str] = set() public_key_owners: dict[bytes, str] = {} duplicate_public_keys: set[bytes] = set() raw_keys = authority_keyring.get("keys") if not isinstance(raw_keys, list): return {}, ( EvidenceFinding( "blocker", "boundary_authority_key_invalid", "Proof-authority keyring does not contain a valid key list.", ("assessment.external_proof",), ), ) for item in raw_keys: if not isinstance(item, dict): findings.append( EvidenceFinding( "blocker", "boundary_authority_key_invalid", "Proof-authority keyring contains a malformed key entry.", ("assessment.external_proof",), ) ) continue key_id = str(item.get("key_id") or "") status = item.get("status") raw_allowed_scopes = item.get("allowed_scopes") if ( len(key_id) > 160 or OPAQUE_ID_PATTERN.fullmatch(key_id) is None or status not in {"active", "next", "revoked", "disabled", "retired"} or not isinstance(raw_allowed_scopes, list) or not raw_allowed_scopes or any(not isinstance(scope, str) for scope in raw_allowed_scopes) or len(raw_allowed_scopes) != len(set(raw_allowed_scopes)) or any(scope not in PROOF_REQUIREMENTS for scope in raw_allowed_scopes) ): findings.append( EvidenceFinding( "blocker", "boundary_authority_key_invalid", "Proof-authority keyring contains a malformed key entry.", ("assessment.external_proof",), ) ) continue if key_id in seen: findings.append( EvidenceFinding( "blocker", "boundary_authority_key_duplicate", "Proof-authority keyring repeats a key ID.", ("assessment.external_proof",), ) ) continue seen.add(key_id) try: public_key = base64.b64decode( str(item.get("public_key") or ""), validate=True ) Ed25519PublicKey.from_public_bytes(public_key) except ValueError: findings.append( EvidenceFinding( "blocker", "boundary_authority_key_invalid", "Proof-authority keyring contains invalid Ed25519 public-key material.", ("assessment.external_proof",), ) ) continue raw_not_before = item.get("not_before") raw_not_after = item.get("not_after") not_before = _datetime(raw_not_before) not_after = _datetime(raw_not_after) if ( (raw_not_before is not None and not_before is None) or (raw_not_after is not None and not_after is None) or ( not_before is not None and not_after is not None and not_before >= not_after ) ): findings.append( EvidenceFinding( "blocker", "boundary_authority_key_interval_invalid", "Proof-authority keyring contains an invalid or empty key-validity interval.", ("assessment.external_proof",), ) ) continue previous_owner = public_key_owners.get(public_key) if previous_owner is not None and previous_owner != key_id: if public_key not in duplicate_public_keys: findings.append( EvidenceFinding( "blocker", "boundary_authority_public_key_duplicate", "Proof-authority keyring assigns one public key to multiple key IDs.", ("assessment.external_proof",), ) ) duplicate_public_keys.add(public_key) result.pop(previous_owner, None) continue public_key_owners[public_key] = key_id if status not in {"active", "next"}: continue if (not_before is not None and verification_time < not_before) or ( not_after is not None and verification_time >= not_after ): continue if public_key in forbidden_public_keys: findings.append( EvidenceFinding( "blocker", "boundary_authority_reuses_release_key", "A proof-authority public key is also trusted for catalog release signatures.", ("assessment.external_proof",), ) ) continue result[key_id] = ( public_key, frozenset(raw_allowed_scopes), ) # A supplied authority keyring is one trust root. Selecting usable members # out of a malformed set would let a valid-looking sibling conceal an # invalid interval, duplicate identity, or ambiguous key assignment. return ({} if findings else result), tuple(findings) def _record_change( *, findings: list[EvidenceFinding], changes: list[dict[str, Any]], changed_repositories: set[str], affected_module_ids: set[str], component: Mapping[str, Any], kind: str, code: str, message: str, extra: Mapping[str, Any] | None = None, ) -> None: module_id = str(component.get("module_id")) repository = str(component.get("repository")) changed_repositories.add(repository) affected_module_ids.add(module_id) findings.append( EvidenceFinding( "review", code, message, (f"composition.{module_id}",), ) ) change: dict[str, Any] = { "module_id": module_id, "repository": repository, "kind": kind, } if extra: change.update(extra) changes.append(change) def _unchecked_installed_scope() -> dict[str, Any]: return { "installed_artifacts": { "checked": False, "valid": None, "required_evidence": ( "A bounded installed-composition evidence document collected from the Python environment under review.", ), }, "installed_record_integrity": {"checked": False, "valid": None}, "installed_source_provenance": { "checked": False, "valid": None, "basis": "local-pep610-metadata", "release_origin_bound": False, }, "installed_release_origin": { "checked": False, "valid": None, "release_origin_bound": False, "required_evidence": ( "An independently anchored artifact digest or signed installer receipt bound to the signed release catalog.", ), }, "runtime_activation": { "checked": False, "valid": None, "required_evidence": ( "A separately collected runtime registry/configuration snapshot bound to installed evidence.", ), }, } def _invalid_installed_scope(*, digest: str) -> dict[str, Any]: scope = _unchecked_installed_scope() scope["installed_artifacts"] = { "checked": False, "valid": None, "comparison_valid": False, "evidence_sha256": digest, } return scope def _unchecked_boundary_scope( *, expected_subjects: Mapping[str, str | None] ) -> dict[str, Any]: return { scope: { "checked": False, "valid": None, "expected_subject_id": expected_subjects.get(scope), "required_evidence": requirements, } for scope, requirements in PROOF_REQUIREMENTS.items() } def _datetime(value: object) -> datetime | None: if not isinstance(value, str) or not value: return None try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC) def _iso_datetime(value: datetime | None) -> str | None: if value is None: return None return value.astimezone(UTC).isoformat().replace("+00:00", "Z") def _safe_opaque_id(value: object, *, fallback: str) -> str: text = value if isinstance(value, str) else "" return ( text if len(text) <= 160 and OPAQUE_ID_PATTERN.fullmatch(text) is not None else fallback ) def _safe_opaque_id_or_none(value: object) -> str | None: if not isinstance(value, str): return None if len(value) > 160 or OPAQUE_ID_PATTERN.fullmatch(value) is None: return None return value def _json_path(parts: Iterable[object]) -> str: path = "$" for part in parts: path += f"[{part}]" if isinstance(part, int) else f".{part}" return path def _unique_dicts(values: Sequence[dict[str, str]]) -> list[dict[str, str]]: result: list[dict[str, str]] = [] seen: set[tuple[tuple[str, str], ...]] = set() for value in values: key = tuple(sorted(value.items())) if key not in seen: seen.add(key) result.append(value) return result def _is_relative_to(path: Path, root: Path) -> bool: try: path.relative_to(root) except ValueError: return False return True def _trusted_distribution_root(root: Path) -> bool: for value in sys.path: try: candidate = Path(value or ".").resolve() except OSError: continue if candidate.parent == candidate: continue if root == candidate or _is_relative_to(root, candidate): return True return False