"""Bounded installed-composition and externally authorized fit evidence.""" from __future__ import annotations import base64 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 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 MAX_GOVOPLAN_DISTRIBUTIONS = 256 MAX_DISTRIBUTION_FILES = 10_000 MAX_COLLECTION_RECORD_FILES = 50_000 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 digest.", "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 digest.", "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 digest.", "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 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.2.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", ) -> 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",), ) ) 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" findings.append( EvidenceFinding( "blocker", "installed_evidence_unsigned_import", "Imported unsigned installed evidence may be compared but cannot establish installed proof.", ("assessment.installed_composition",), ) ) if effective_verification_mode == "invalid": 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_anchored = 0 provenance_mutable = 0 provenance_unanchored = 0 artifact_valid = 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: artifact_valid = 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: artifact_valid = 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: artifact_valid = 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: artifact_valid = 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: artifact_valid = 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: artifact_valid = 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 "" ) 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_anchored += 1 else: artifact_valid = False _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 immutable VCS provenance for {package_name!r} does not match the assessed or signed selected-unit commit.", ) elif provenance_kind in {"editable-vcs", "editable-local", "local-directory"}: provenance_mutable += 1 artifact_valid = False _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} is editable or directory-backed and is not immutable release evidence.", ) else: provenance_unanchored += 1 artifact_valid = False _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 source provenance anchored to the assessed release.", ) expected_packages = set(expected_by_package) for package_name in sorted(set(artifacts_by_package) - expected_packages): artifact_valid = 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 "") artifact_valid = 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 ) installed_checked = observation_trusted record_checked = ( observation_trusted and binding_valid and exact_set and record_expected > 0 ) provenance_checked = ( observation_trusted and binding_valid and exact_set and provenance_expected > 0 ) proof_scope = { "installed_artifacts": { "checked": installed_checked, "valid": artifact_valid if installed_checked else None, "comparison_valid": artifact_valid, "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_anchored == provenance_expected if provenance_checked else None, "anchored_distribution_count": provenance_anchored, "mutable_distribution_count": provenance_mutable, "unanchored_distribution_count": provenance_unanchored, "expected_distribution_count": provenance_expected, }, "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, "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_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", {}) 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"), ) ) 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 {"kind": "malformed-direct-url"} if raw is None: return {"kind": "index-or-unknown"} if not isinstance(raw, str) or len(raw) > MAX_DIRECT_URL_BYTES: return {"kind": "malformed-direct-url"} if len(raw.encode("utf-8")) > MAX_DIRECT_URL_BYTES: return {"kind": "malformed-direct-url"} try: direct_url = json.loads(raw) except (TypeError, json.JSONDecodeError): return {"kind": "malformed-direct-url"} if not isinstance(direct_url, dict): return {"kind": "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 {"kind": "malformed-direct-url"} try: parsed_url = urlsplit(raw_url) except ValueError: return {"kind": "malformed-direct-url"} if parsed_url.password is not None or parsed_url.query or parsed_url.fragment: return {"kind": "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 {"kind": "malformed-direct-url"} if isinstance(vcs_info, dict): if vcs_info.get("vcs") != "git": return {"kind": "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 {"kind": "malformed-direct-url"} return {"kind": "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 {"kind": "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 {"kind": "archive", "sha256": digest_text} return {"kind": "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 {"kind": "malformed-direct-url"} return {"kind": "editable-local" if editable else "local-directory"} return {"kind": "malformed-direct-url"} 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, } try: files = tuple(distribution.files or ()) except Exception: files = () if not files: return {"status": "unavailable", **counts} if len(files) > MAX_DISTRIBUTION_FILES or len(files) > budget.remaining_files: budget.remaining_files = 0 return {"status": "limit-exceeded", **counts} budget.remaining_files -= len(files) try: distribution_root = Path(distribution.locate_file("")).resolve() installation_root = Path(sys.prefix).resolve() except OSError: return {"status": "unavailable", **counts} if not _trusted_distribution_root(distribution_root): return {"status": "unavailable", **counts} declared_scripts = _declared_distribution_scripts(distribution) owner_prefix = package_name.replace("-", "_") resolved_entries: list[tuple[object, Path | None]] = [ ( package_path, _resolve_record_path( distribution=distribution, package_path=package_path, distribution_root=distribution_root, installation_root=installation_root, owner_prefix=owner_prefix, declared_scripts=declared_scripts, ), ) for package_path in files ] hashed_payload_paths = { resolved_path for package_path, resolved_path in resolved_entries if resolved_path is not None and getattr(getattr(package_path, "hash", None), "mode", None) == "sha256" } total_hashed_bytes = 0 limit_exceeded = False for package_path, resolved_path in resolved_entries: hash_info = getattr(package_path, "hash", None) if resolved_path is None: counts["unverifiable_file_count"] += 1 continue if hash_info 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 getattr(hash_info, "mode", None) != "sha256": counts["unverifiable_file_count"] += 1 continue try: file_stat = resolved_path.stat() except OSError: counts["missing_file_count"] += 1 continue if not stat.S_ISREG(file_stat.st_mode): counts["missing_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(str(getattr(hash_info, "value", ""))) 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 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" return {"status": status, **counts} 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: 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": return ( hashed > 0 and counts["permitted_unhashed_file_count"] > 0 and unverifiable == 0 and missing == 0 and mismatched == 0 ) 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 _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() for item in authority_keyring.get("keys", []): if not isinstance(item, dict): continue key_id = str(item.get("key_id") or "") 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: 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 item.get("status") not in {"active", "next"}: continue not_before = _datetime(item.get("not_before")) not_after = _datetime(item.get("not_after")) 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(str(value) for value in item.get("allowed_scopes", [])), ) return 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}, "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