diff --git a/docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md b/docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md index a483104..4ae2bd0 100644 --- a/docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md +++ b/docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md @@ -389,7 +389,7 @@ that observation to the assessment and signed catalog: --output /var/tmp/govoplan-fit-review.json ``` -The collector follows the strict version `0.2.0` +The collector follows the strict version `0.3.0` [`installed-composition-evidence.schema.json`](installed-composition-evidence.schema.json) contract. It enumerates all installed distributions whose normalized name starts with `govoplan-`, compares the enabled assessed package and module-manifest @@ -399,37 +399,47 @@ entries and evidence-backed conclusions. For each distribution, the collector also: -- verifies every supported SHA-256 immutable wheel-payload entry in installed - `RECORD` metadata, - within fixed limits of 256 GovOPlaN distributions, 10,000 files and 512 MiB - hashed per distribution, 50,000 files and 2 GiB hashed for one collection, - and 64 MiB for any single file; +- reads the installed wheel `RECORD` file directly, with a 4 MiB metadata bound, + rather than relying on an interpreted distribution file list; missing files, + duplicate or malformed declarations, mismatched declared sizes and hashes all + prevent a `verified` result; +- verifies every supported SHA-256 wheel-payload declaration in that local + metadata, within fixed limits of 256 GovOPlaN distributions, 10,000 files and + 512 MiB hashed per distribution, 50,000 files and 2 GiB hashed for one + collection, and 64 MiB for any single file; - accepts package-owned data-file targets below the installation prefix and declared `console_scripts`, while rejecting other traversal, undeclared scripts, paths outside the prefix and leaf symlinks before opening a file; - reports pip-generated, unhashed `__pycache__/*.pyc` entries separately only when their derived source is a hashed payload entry, including generated files below a confined package-owned runtime-data prefix; -- distinguishes coherent PEP 610 Git commits and archive hashes from editable installs, - local directories, package-index/unknown origins and malformed metadata; +- labels every PEP 610 observation with the explicit + `local-pep610-metadata` basis and distinguishes coherent declared Git commits + and archive hashes from editable installs, local directories, + package-index/unknown origins and malformed metadata; - compares a non-editable VCS commit with the assessment commit and, when one is present, the signed catalog `selected_units` commit; - records only package/module identifiers, versions, bounded counters, hashes and stable error codes. It never writes direct URLs, paths, hostnames, usernames, exception text or file contents. -`RECORD` `verified` means complete matching coverage of the immutable wheel -payload, not every runtime byte. The proof scope exposes the aggregate count of -generated unhashed bytecode; runtime activation remains unchecked. `RECORD` -agreement proves that payload files agree with installed metadata, but does not -make self-consistent metadata a trusted release origin. Similarly, a version -string is not artifact integrity. PEP 610 VCS evidence is accepted only for +`RECORD` `verified` means complete matching coverage of the wheel payload +declared by the local metadata, not every runtime byte and not a release-origin +binding. The proof scope exposes the aggregate count of generated unhashed +bytecode; runtime activation remains unchecked. `RECORD` agreement proves that +payload files agree with installed metadata, but does not make self-consistent +metadata a trusted release origin. Similarly, a version string is not artifact +integrity. PEP 610 VCS metadata is accepted only for `vcs: git`, a full 40–64 hexadecimal commit, one mutually exclusive provenance form and a coherent bounded URL shape. Editable and directory-backed installs remain mutable even when their small editable-install `RECORD` is valid. Archive or index artifacts without a hash anchored by the assessed release remain -unanchored. These conditions cause review rather than being promoted to -immutable installed-release proof. +unanchored. A matching declaration is reported under +`installed_source_provenance` as local consistency only, with +`release_origin_bound: false`. The separate `installed_release_origin` scope +remains unchecked until an independently anchored artifact digest or signed +installer receipt binds each installed payload to the signed catalog. The tool +does not accept a digest merely asserted inside the installed evidence document. Only evidence produced in-process by `--collect-installed-evidence` is a local observation, and it must be no more than five minutes old (with at most 30 @@ -466,7 +476,10 @@ claims use `passed` or `failed`; production claims use `approved` or `rejected`. One claim per scope, unique control/artifact IDs, `issued_at < expires_at`, current validity and exact digest binding are mandatory. Any schema, binding, time, signature or authority blocker leaves every supplied boundary claim unchecked; -an authorized negative result is checked but requires review. +one malformed authority member or invalid validity interval invalidates the +supplied authority set as a whole. An authorized negative result is checked but +requires review only after every prerequisite, including installed +release-origin binding, is established. Signatures cover UTF-8 JSON with the `signatures` member omitted, object keys sorted, compact `,`/`:` separators and non-ASCII characters escaped, matching the tool's deterministic canonicalization. @@ -483,11 +496,13 @@ the tool's deterministic canonicalization. With `--installed-evidence`, this command performs comparison and proof-binding diagnostics only: the imported installed document is unsigned, so neither it nor -the boundary claim becomes accepted proof. An accepted boundary review requires -a trusted orchestrator to retain the direct in-process observation, obtain the -external signature over its digest, and call the review library within the -five-minute observation window. A future signed collector-attestation format -should make that exchange durable across processes. +the boundary claim becomes accepted proof. Direct in-process collection removes +that unsigned-import gap, but still establishes only local metadata consistency. +No boundary claim is admitted by the current format because it has no separately +anchored artifact digest or signed installer receipt. A future evidence contract +must bind such an independently trusted release-artifact receipt to the installed +evidence digest; a signed collector-attestation format should also make that +exchange durable across processes. Target-environment and production-approval claims must use the assessment's `deployment_profile.id` as `subject_id`. External-provider claims require the @@ -499,8 +514,10 @@ The authority keyring is a governance trust root. Do not download or generate it from the proof bundle being checked. The checker rejects proof-authority public keys reused by either the published or independently trusted catalog keyring, and rejects the same proof public-key material assigned to multiple authority -IDs. Authority `not_after` is exclusive. A target operator or approver must validate -the referenced drill/result artifacts before signing. The rerun verifies the +IDs. A malformed key, an invalid or empty validity interval, or ambiguous key +identity invalidates the supplied authority set. Authority `not_after` is +exclusive. A target operator or approver must validate the referenced +drill/result artifacts before signing. The rerun verifies the attestation and its bindings; it does not fetch or reinterpret those artifacts. `--verification-time` exists only for reproducible historical review. Supplying it always marks the machine and human report as a **HISTORICAL override**; callers diff --git a/docs/installed-composition-evidence.schema.json b/docs/installed-composition-evidence.schema.json index eb94a1b..9e607ca 100644 --- a/docs/installed-composition-evidence.schema.json +++ b/docs/installed-composition-evidence.schema.json @@ -20,7 +20,7 @@ "format": "uri-reference" }, "schema_version": { - "const": "0.2.0" + "const": "0.3.0" }, "evidence_kind": { "const": "govoplan.installed-composition" @@ -122,8 +122,11 @@ { "type": "object", "additionalProperties": false, - "required": ["kind", "commit"], + "required": ["basis", "kind", "commit"], "properties": { + "basis": { + "const": "local-pep610-metadata" + }, "kind": { "const": "vcs-commit" }, @@ -136,8 +139,11 @@ { "type": "object", "additionalProperties": false, - "required": ["kind", "sha256"], + "required": ["basis", "kind", "sha256"], "properties": { + "basis": { + "const": "local-pep610-metadata" + }, "kind": { "const": "archive" }, @@ -149,8 +155,11 @@ { "type": "object", "additionalProperties": false, - "required": ["kind"], + "required": ["basis", "kind"], "properties": { + "basis": { + "const": "local-pep610-metadata" + }, "kind": { "enum": [ "editable-local", diff --git a/tests/test_capability_fit_evidence.py b/tests/test_capability_fit_evidence.py index 272626c..a7f3591 100644 --- a/tests/test_capability_fit_evidence.py +++ b/tests/test_capability_fit_evidence.py @@ -55,7 +55,7 @@ class CapabilityFitEvidenceTests(unittest.TestCase): "docs/capability-fit-proof-authority-keyring.schema.json" ) - def test_matching_installed_composition_proves_versions_records_and_commits( + def test_matching_installed_composition_proves_local_consistency_not_origin( self, ) -> None: catalog, keyring = signed_catalog(self.assessment) @@ -71,6 +71,17 @@ class CapabilityFitEvidenceTests(unittest.TestCase): self.assertTrue(report["proof_scope"]["installed_artifacts"]["valid"]) self.assertTrue(report["proof_scope"]["installed_record_integrity"]["valid"]) self.assertTrue(report["proof_scope"]["installed_source_provenance"]["valid"]) + self.assertEqual( + "local-pep610-metadata", + report["proof_scope"]["installed_source_provenance"]["basis"], + ) + self.assertFalse( + report["proof_scope"]["installed_source_provenance"][ + "release_origin_bound" + ] + ) + self.assertFalse(report["proof_scope"]["installed_release_origin"]["checked"]) + self.assertIsNone(report["proof_scope"]["installed_release_origin"]["valid"]) self.assertFalse(report["proof_scope"]["runtime_activation"]["checked"]) self.assertFalse(report["proof_scope"]["target_environment"]["checked"]) self.assertEqual( @@ -78,6 +89,40 @@ class CapabilityFitEvidenceTests(unittest.TestCase): report["proof_scope"]["installed_artifacts"]["evidence_sha256"], ) + def test_catalog_trust_blocker_invalidates_dependent_evidence_scopes( + self, + ) -> None: + catalog, keyring = signed_catalog(self.assessment) + catalog["signatures"][0]["value"] = "AAAA" + installed = matching_installed_evidence(self.assessment) + proof, authority = signed_boundary_evidence( + assessment=self.assessment, + installed=installed, + claims=[boundary_claim("target_environment", "passed")], + allowed_scopes=["target_environment"], + ) + + report = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=installed, + boundary_evidence=proof, + authority_keyring=authority, + ) + + self.assertEqual("blocked", report["status"]) + self.assertFalse( + report["proof_scope"]["catalog_signature_and_keyring"]["valid"] + ) + self.assertFalse(report["proof_scope"]["installed_artifacts"]["checked"]) + self.assertIsNone(report["proof_scope"]["installed_artifacts"]["valid"]) + self.assertFalse(report["proof_scope"]["target_environment"]["checked"]) + self.assertIsNone(report["proof_scope"]["target_environment"]["valid"]) + self.assertIn( + "installed_catalog_trust_unavailable", + {item["code"] for item in report["findings"]}, + ) + def test_missing_extra_and_version_drift_have_deterministic_review_targets( self, ) -> None: @@ -137,7 +182,10 @@ class CapabilityFitEvidenceTests(unittest.TestCase): for item in evidence["artifacts"] if item["package_name"] == "govoplan-campaign" ) - campaign["source_provenance"] = {"kind": "editable-local"} + campaign["source_provenance"] = { + "basis": "local-pep610-metadata", + "kind": "editable-local", + } campaign["record_integrity"] = { **campaign["record_integrity"], "status": "partial", @@ -212,6 +260,7 @@ class CapabilityFitEvidenceTests(unittest.TestCase): schema=self.installed_schema, observation_mode="direct_local", verification_time=datetime(2026, 7, 23, 12, tzinfo=UTC), + catalog_trusted=True, ) codes = {item.code for item in result.findings} @@ -317,6 +366,10 @@ class CapabilityFitEvidenceTests(unittest.TestCase): ) observed = evidence["artifacts"][0] self.assertEqual("editable-local", observed["source_provenance"]["kind"]) + self.assertEqual( + "local-pep610-metadata", + observed["source_provenance"]["basis"], + ) self.assertEqual("verified", observed["record_integrity"]["status"]) encoded = json.dumps(evidence, sort_keys=True) self.assertNotIn(str(root), encoded) @@ -338,6 +391,108 @@ class CapabilityFitEvidenceTests(unittest.TestCase): tampered["artifacts"][0]["record_integrity"]["mismatched_file_count"], ) + def test_collector_detects_payload_declared_in_record_but_missing(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + distribution, payload_file = create_distribution(root) + payload_file.unlink() + + with mock.patch.object(sys, "path", [str(root), *sys.path]): + evidence = collect_installed_composition( + assessment=self.assessment, + collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC), + distributions=[distribution], + ) + + integrity = evidence["artifacts"][0]["record_integrity"] + self.assertEqual("mismatch", integrity["status"]) + self.assertEqual(1, integrity["missing_file_count"]) + + def test_collector_rejects_duplicate_record_declarations(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + distribution, _ = create_distribution(root) + record_path = root / "govoplan_demo-1.0.0.dist-info" / "RECORD" + rows = record_path.read_text(encoding="utf-8").splitlines() + record_path.write_text( + "\n".join((*rows, rows[0])) + "\n", + encoding="utf-8", + ) + + with mock.patch.object(sys, "path", [str(root), *sys.path]): + evidence = collect_installed_composition( + assessment=self.assessment, + collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC), + distributions=[distribution], + ) + + integrity = evidence["artifacts"][0]["record_integrity"] + self.assertEqual("partial", integrity["status"]) + self.assertGreater(integrity["unverifiable_file_count"], 0) + + def test_collector_rejects_malformed_record_declarations(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + distribution, _ = create_distribution(root) + record_path = root / "govoplan_demo-1.0.0.dist-info" / "RECORD" + record_path.write_text( + record_path.read_text(encoding="utf-8") + "malformed,row\n", + encoding="utf-8", + ) + + with mock.patch.object(sys, "path", [str(root), *sys.path]): + evidence = collect_installed_composition( + assessment=self.assessment, + collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC), + distributions=[distribution], + ) + + integrity = evidence["artifacts"][0]["record_integrity"] + self.assertEqual("partial", integrity["status"]) + self.assertGreater(integrity["unverifiable_file_count"], 0) + + def test_collector_verifies_declared_record_sizes(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + distribution, _ = create_distribution(root) + record_path = root / "govoplan_demo-1.0.0.dist-info" / "RECORD" + rows = record_path.read_text(encoding="utf-8").splitlines() + path, digest, size = rows[0].split(",") + rows[0] = f"{path},{digest},{int(size) + 1}" + record_path.write_text("\n".join(rows) + "\n", encoding="utf-8") + + with mock.patch.object(sys, "path", [str(root), *sys.path]): + evidence = collect_installed_composition( + assessment=self.assessment, + collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC), + distributions=[distribution], + ) + + integrity = evidence["artifacts"][0]["record_integrity"] + self.assertEqual("mismatch", integrity["status"]) + self.assertEqual(1, integrity["mismatched_file_count"]) + + def test_collector_rejects_symlinked_distribution_metadata_root(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + create_distribution(root) + metadata_root = root / "govoplan_demo-1.0.0.dist-info" + real_metadata_root = root / "real-govoplan-demo.dist-info" + metadata_root.rename(real_metadata_root) + metadata_root.symlink_to(real_metadata_root, target_is_directory=True) + distribution = metadata.PathDistribution(metadata_root) + + with mock.patch.object(sys, "path", [str(root), *sys.path]): + evidence = collect_installed_composition( + assessment=self.assessment, + collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC), + distributions=[distribution], + ) + + integrity = evidence["artifacts"][0]["record_integrity"] + self.assertEqual("unavailable", integrity["status"]) + self.assertEqual(0, integrity["hashed_file_count"]) + def test_collector_accepts_real_core_wheel_payload_scripts_data_and_pyc( self, ) -> None: @@ -604,7 +759,9 @@ class CapabilityFitEvidenceTests(unittest.TestCase): json.dumps(reversed_order, sort_keys=True), ) - def test_signed_scope_authority_can_validate_only_its_external_claim(self) -> None: + def test_signed_boundary_claim_waits_for_anchored_installed_release_origin( + self, + ) -> None: catalog, keyring = signed_catalog(self.assessment) installed = matching_installed_evidence(self.assessment) proof, authority = signed_boundary_evidence( @@ -622,10 +779,15 @@ class CapabilityFitEvidenceTests(unittest.TestCase): authority_keyring=authority, ) - self.assertEqual("current", report["status"]) - self.assertTrue(report["proof_scope"]["target_environment"]["valid"]) + self.assertEqual("blocked", report["status"]) + self.assertFalse(report["proof_scope"]["target_environment"]["checked"]) + self.assertIsNone(report["proof_scope"]["target_environment"]["valid"]) self.assertFalse(report["proof_scope"]["external_providers"]["checked"]) self.assertFalse(report["proof_scope"]["production_approval"]["checked"]) + self.assertIn( + "boundary_installed_release_origin_untrusted", + {item["code"] for item in report["findings"]}, + ) self.assertEqual( self.assessment["deployment_profile"]["id"], report["proof_scope"]["target_environment"]["expected_subject_id"], @@ -666,10 +828,15 @@ class CapabilityFitEvidenceTests(unittest.TestCase): "boundary_provider_subject_unconfigured", {item["code"] for item in unconfigured["findings"]}, ) - self.assertEqual("current", accepted["status"]) + self.assertEqual("blocked", accepted["status"]) provider_scope = accepted["proof_scope"]["external_providers"] + self.assertFalse(provider_scope["checked"]) self.assertEqual("provider:test", provider_scope["expected_subject_id"]) self.assertEqual("provider:test", provider_scope["observed_subject_id"]) + self.assertIn( + "boundary_installed_release_origin_untrusted", + {item["code"] for item in accepted["findings"]}, + ) def test_boundary_subject_mismatch_blocks_without_echoing_invalid_config( self, @@ -711,7 +878,7 @@ class CapabilityFitEvidenceTests(unittest.TestCase): self.assertNotIn("private.invalid", encoded) self.assertNotIn("secret", encoded) - def test_authorized_negative_claim_creates_scope_specific_review_target( + def test_authorized_negative_claim_waits_for_installed_origin_binding( self, ) -> None: catalog, keyring = signed_catalog(self.assessment) @@ -734,14 +901,12 @@ class CapabilityFitEvidenceTests(unittest.TestCase): authority_keyring=authority, ) - targets = {item["id"] for item in report["review_targets"]} - self.assertIn( + self.assertEqual("blocked", report["status"]) + self.assertFalse(report["proof_scope"]["target_environment"]["checked"]) + self.assertFalse(report["proof_scope"]["production_approval"]["checked"]) + self.assertNotIn( "boundary.target_environment.production-like-dev", - targets, - ) - self.assertIn( - "boundary.production_approval.production-like-dev", - targets, + {item["id"] for item in report["review_targets"]}, ) def test_proof_authorities_must_be_independent_and_key_ids_unambiguous( @@ -818,6 +983,55 @@ class CapabilityFitEvidenceTests(unittest.TestCase): {item["code"] for item in report["findings"]}, ) + def test_malformed_authority_members_invalidate_the_whole_keyring(self) -> None: + catalog, keyring = signed_catalog(self.assessment) + installed = matching_installed_evidence(self.assessment) + proof, authority = signed_boundary_evidence( + assessment=self.assessment, + installed=installed, + claims=[boundary_claim("target_environment", "passed")], + allowed_scopes=["target_environment"], + ) + malformed = deepcopy(authority) + malformed["keys"].append( + { + "key_id": "authority:malformed", + "status": "active", + "public_key": "not-base64", + "allowed_scopes": ["target_environment"], + } + ) + inverted = deepcopy(authority) + inverted["keys"][0]["not_before"] = "2026-08-01T00:00:00Z" + inverted["keys"][0]["not_after"] = "2026-07-01T00:00:00Z" + + cases = ( + (malformed, "boundary_authority_key_invalid"), + (inverted, "boundary_authority_key_interval_invalid"), + ) + for supplied_keyring, expected_code in cases: + with self.subTest(expected_code=expected_code): + report = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=installed, + boundary_evidence=proof, + authority_keyring=supplied_keyring, + ) + + self.assertEqual("blocked", report["status"]) + self.assertIn( + expected_code, + {item["code"] for item in report["findings"]}, + ) + self.assertIn( + "boundary_authority_untrusted", + {item["code"] for item in report["findings"]}, + ) + self.assertFalse( + report["proof_scope"]["target_environment"]["checked"] + ) + def test_historical_verification_and_generated_runtime_gap_are_visible( self, ) -> None: @@ -855,6 +1069,7 @@ class CapabilityFitEvidenceTests(unittest.TestCase): observation_mode="direct_local", verification_time=datetime(2026, 7, 23, 12, tzinfo=UTC), verification_mode="https://private.invalid/secret", + catalog_trusted=True, ) encoded = json.dumps(result.proof_scope, sort_keys=True) @@ -981,7 +1196,7 @@ class CapabilityFitEvidenceTests(unittest.TestCase): def matching_installed_evidence(assessment: dict[str, object]) -> dict[str, object]: return { "$schema": "./installed-composition-evidence.schema.json", - "schema_version": "0.2.0", + "schema_version": "0.3.0", "evidence_kind": "govoplan.installed-composition", "assessment_id": assessment["assessment_id"], "assessment_release": assessment["release"]["ref"], @@ -1010,7 +1225,11 @@ def artifact( "modules": [] if module_id == "core" else [{"module_id": module_id, "manifest_version": version}], - "source_provenance": {"kind": "vcs-commit", "commit": commit}, + "source_provenance": { + "basis": "local-pep610-metadata", + "kind": "vcs-commit", + "commit": commit, + }, "record_integrity": { "status": "verified", "hashed_file_count": 1, diff --git a/tests/test_capability_fit_review.py b/tests/test_capability_fit_review.py index 768b65a..7555a8d 100644 --- a/tests/test_capability_fit_review.py +++ b/tests/test_capability_fit_review.py @@ -360,7 +360,8 @@ class CapabilityFitReviewTests(unittest.TestCase): self.assertIn("Capability fit rerun: current", rendered) self.assertIn( - "No installed-artifact, target-provider, or production proof", rendered + "No installed-composition, installed-release-origin, target-environment, external-provider, production-approval proof", + rendered, ) def test_public_fetch_failure_is_generic_and_blocking(self) -> None: diff --git a/tools/assessments/govoplan_assessment/capability_fit.py b/tools/assessments/govoplan_assessment/capability_fit.py index de89ac4..d514854 100644 --- a/tools/assessments/govoplan_assessment/capability_fit.py +++ b/tools/assessments/govoplan_assessment/capability_fit.py @@ -363,6 +363,9 @@ def review_capability_fit( evidence_verification_mode = ( "historical_override" if verification_time is not None else "current" ) + catalog_dependency_trusted = not any( + item.severity == "blocker" for item in findings + ) installed_review = review_installed_composition( assessment=assessment, catalog_entries=catalog_entries, @@ -373,6 +376,7 @@ def review_capability_fit( observation_mode=installed_evidence_mode, verification_time=evidence_verification_time, verification_mode=evidence_verification_mode, + catalog_trusted=catalog_dependency_trusted, ) findings.extend( Finding(item.severity, item.code, item.message, item.assessment_ids) @@ -884,7 +888,17 @@ def build_report( else { "installed_artifacts": {"checked": False, "valid": None}, "installed_record_integrity": {"checked": False, "valid": None}, - "installed_source_provenance": {"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, + }, "runtime_activation": {"checked": False, "valid": None}, } ) @@ -925,7 +939,7 @@ def build_report( **boundary_scope, } return { - "report_version": "0.3.0", + "report_version": "0.4.0", "status": status, "assessment_id": assessment.get("assessment_id"), "assessment_release": assessment.get("release", {}).get("ref") @@ -950,6 +964,9 @@ def render_review(report: dict[str, Any]) -> str: proof_scope = report.get("proof_scope", {}) installed_scope = proof_scope.get("installed_artifacts", {}) installed_checked = installed_scope.get("checked") is True + release_origin_checked = ( + proof_scope.get("installed_release_origin", {}).get("checked") is True + ) installed_supplied = bool(installed_scope.get("evidence_sha256")) or ( "installed_evidence_observation" in proof_scope ) @@ -960,7 +977,7 @@ def render_review(report: dict[str, Any]) -> str: ( "Scope: schema, signed release metadata, optional local tag provenance, " + ( - "and locally observed installed-composition evidence." + "and locally observed installed-composition consistency evidence; release origin remains a separate proof boundary." if installed_checked else ( "and supplied installed-composition evidence that did not establish local proof." @@ -1006,12 +1023,12 @@ def render_review(report: dict[str, Any]) -> str: if report.get("proof_scope", {}).get(key, {}).get("checked") is not True ] if not installed_checked: - unchecked_boundaries.insert(0, "installed-artifact") - if not installed_checked and len(unchecked_boundaries) == 4: - lines.append( - "No installed-artifact, target-provider, or production proof was performed." + unchecked_boundaries.insert(0, "installed-composition") + if not release_origin_checked: + unchecked_boundaries.insert( + 1 if not installed_checked else 0, "installed-release-origin" ) - elif unchecked_boundaries: + if unchecked_boundaries: lines.append("No " + ", ".join(unchecked_boundaries) + " proof was performed.") return "\n".join(lines) + "\n" diff --git a/tools/assessments/govoplan_assessment/evidence.py b/tools/assessments/govoplan_assessment/evidence.py index 3c86b82..741462f 100644 --- a/tools/assessments/govoplan_assessment/evidence.py +++ b/tools/assessments/govoplan_assessment/evidence.py @@ -3,12 +3,14 @@ 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 @@ -27,6 +29,8 @@ from jsonschema.exceptions import SchemaError 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 @@ -56,17 +60,17 @@ NEGATIVE_RESULTS = { } PROOF_REQUIREMENTS = { "target_environment": ( - "A target-run result bound to this assessment release and installed-evidence digest.", + "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 digest.", + "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 digest.", + "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.", ), @@ -104,6 +108,14 @@ class CollectionHashBudget: 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, ...]: @@ -231,7 +243,7 @@ def collect_installed_composition( sorted_issues.sort(key=lambda item: (item["package_name"], item["code"])) return { "$schema": "./installed-composition-evidence.schema.json", - "schema_version": "0.2.0", + "schema_version": "0.3.0", "evidence_kind": "govoplan.installed-composition", "assessment_id": _safe_opaque_id( assessment.get("assessment_id"), fallback="invalid-assessment" @@ -264,6 +276,7 @@ def review_installed_composition( observation_mode: str = "imported_unsigned", verification_time: datetime | None = None, verification_mode: str = "current", + catalog_trusted: bool = False, ) -> InstalledEvidenceReview: if evidence is None: return InstalledEvidenceReview( @@ -337,6 +350,15 @@ def review_installed_composition( ("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() @@ -391,6 +413,8 @@ def review_installed_composition( ) 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) @@ -501,7 +525,7 @@ def review_installed_composition( record_verified = 0 generated_unhashed_file_count = 0 provenance_expected = 0 - provenance_anchored = 0 + provenance_declared_match = 0 provenance_mutable = 0 provenance_unanchored = 0 artifact_valid = binding_valid and not duplicate_packages @@ -648,6 +672,23 @@ def review_installed_composition( 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": + artifact_valid = False + 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) @@ -660,7 +701,7 @@ def review_installed_composition( or commit == selected_commit.lower() ) if commit_matches and selected_matches: - provenance_anchored += 1 + provenance_declared_match += 1 else: artifact_valid = False _record_change( @@ -671,7 +712,7 @@ def review_installed_composition( 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.", + 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 @@ -684,7 +725,7 @@ def review_installed_composition( 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.", + message=f"Installed distribution {package_name!r} declares an editable or directory-backed local source.", ) else: provenance_unanchored += 1 @@ -697,7 +738,7 @@ def review_installed_composition( 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.", + message=f"Installed distribution {package_name!r} has no local PEP 610 source statement matching the assessed release.", ) expected_packages = set(expected_by_package) @@ -785,13 +826,25 @@ def review_installed_composition( }, "installed_source_provenance": { "checked": provenance_checked, - "valid": provenance_anchored == provenance_expected + "valid": provenance_declared_match == provenance_expected if provenance_checked else None, - "anchored_distribution_count": provenance_anchored, + "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": { + "checked": False, + "valid": None, + "release_origin_bound": False, + "anchored_artifact_digest_count": 0, + "expected_distribution_count": provenance_expected, + "required_evidence": ( + "An independently anchored artifact digest or signed installer receipt that binds each installed payload to the signed release catalog.", + ), }, "runtime_activation": { "checked": False, @@ -807,6 +860,7 @@ def review_installed_composition( "collected_at": _iso_datetime(collected_at), "evaluated_at": _iso_datetime(now), "verification_mode": effective_verification_mode, + "catalog_trusted": catalog_trusted is True, "freshness_window_seconds": int(MAX_LOCAL_OBSERVATION_AGE.total_seconds()), }, } @@ -949,6 +1003,9 @@ def review_boundary_evidence( ) 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 @@ -962,6 +1019,15 @@ def review_boundary_evidence( ("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")) @@ -1295,19 +1361,19 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]: try: raw = distribution.read_text("direct_url.json") except Exception: - return {"kind": "malformed-direct-url"} + return _local_source_metadata("malformed-direct-url") if raw is None: - return {"kind": "index-or-unknown"} + return _local_source_metadata("index-or-unknown") if not isinstance(raw, str) or len(raw) > MAX_DIRECT_URL_BYTES: - return {"kind": "malformed-direct-url"} + return _local_source_metadata("malformed-direct-url") if len(raw.encode("utf-8")) > MAX_DIRECT_URL_BYTES: - return {"kind": "malformed-direct-url"} + return _local_source_metadata("malformed-direct-url") try: direct_url = json.loads(raw) except (TypeError, json.JSONDecodeError): - return {"kind": "malformed-direct-url"} + return _local_source_metadata("malformed-direct-url") if not isinstance(direct_url, dict): - return {"kind": "malformed-direct-url"} + return _local_source_metadata("malformed-direct-url") raw_url = direct_url.get("url") if ( not isinstance(raw_url, str) @@ -1315,13 +1381,13 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]: or len(raw_url) > 2048 or any(character.isspace() or ord(character) < 32 for character in raw_url) ): - return {"kind": "malformed-direct-url"} + return _local_source_metadata("malformed-direct-url") try: parsed_url = urlsplit(raw_url) except ValueError: - return {"kind": "malformed-direct-url"} + return _local_source_metadata("malformed-direct-url") if parsed_url.password is not None or parsed_url.query or parsed_url.fragment: - return {"kind": "malformed-direct-url"} + 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") @@ -1335,10 +1401,10 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]: if value is not None ) if len(provenance_forms) != 1: - return {"kind": "malformed-direct-url"} + return _local_source_metadata("malformed-direct-url") if isinstance(vcs_info, dict): if vcs_info.get("vcs") != "git": - return {"kind": "malformed-direct-url"} + return _local_source_metadata("malformed-direct-url") commit = str(vcs_info.get("commit_id") or "").lower() requested_revision = vcs_info.get("requested_revision") if ( @@ -1359,8 +1425,8 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]: ) ) ): - return {"kind": "malformed-direct-url"} - return {"kind": "vcs-commit", "commit": commit} + 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"} @@ -1370,7 +1436,7 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]: and (not parsed_url.hostname or parsed_url.username) ) ): - return {"kind": "malformed-direct-url"} + 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: @@ -1379,8 +1445,8 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]: 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"} + 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 ( @@ -1388,9 +1454,142 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]: 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"} + 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( @@ -1407,53 +1606,84 @@ def _record_integrity( "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: + except Exception: return {"status": "unavailable", **counts} if not _trusted_distribution_root(distribution_root): return {"status": "unavailable", **counts} + 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} + if record_status != "ok": + counts["unverifiable_file_count"] = malformed_rows + return {"status": "unavailable", **counts} + record_file_count = len(entries) + malformed_rows + if record_file_count == 0: + return {"status": "unavailable", **counts} + 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[object, Path | None]] = [ + resolved_entries: list[tuple[RecordEntry, Path | None]] = [ ( - package_path, + entry, _resolve_record_path( distribution=distribution, - package_path=package_path, + package_path=entry.path, distribution_root=distribution_root, installation_root=installation_root, owner_prefix=owner_prefix, declared_scripts=declared_scripts, ), ) - for package_path in files + 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 package_path, resolved_path in resolved_entries + for entry, resolved_path in resolved_entries if resolved_path is not None - and getattr(getattr(package_path, "hash", None), "mode", None) == "sha256" + and resolved_path not in duplicate_resolved_paths + and entry.hash_mode == "sha256" } total_hashed_bytes = 0 limit_exceeded = False - for package_path, resolved_path in resolved_entries: - hash_info = getattr(package_path, "hash", None) + for entry, resolved_path in resolved_entries: if resolved_path is None: counts["unverifiable_file_count"] += 1 continue - if hash_info is None: + 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, @@ -1468,17 +1698,9 @@ def _record_integrity( else: counts["unverifiable_file_count"] += 1 continue - if getattr(hash_info, "mode", None) != "sha256": + if entry.hash_mode != "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 ) @@ -1490,7 +1712,7 @@ def _record_integrity( if allowed_bytes <= 0 or file_stat.st_size > allowed_bytes: limit_exceeded = True continue - expected = _record_hash_bytes(str(getattr(hash_info, "value", ""))) + expected = _record_hash_bytes(entry.hash_value or "") if expected is None: counts["unverifiable_file_count"] += 1 continue @@ -1541,7 +1763,7 @@ def _resolve_record_path( if unresolved_path.is_symlink(): return None resolved_path = unresolved_path.resolve() - except OSError: + except (OSError, TypeError, ValueError): return None has_parent_traversal = ".." in record_path.parts if not has_parent_traversal: @@ -1704,10 +1926,49 @@ def _active_authority_keys( seen: set[str] = set() public_key_owners: dict[bytes, str] = {} duplicate_public_keys: set[bytes] = set() - for item in authority_keyring.get("keys", []): + 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( @@ -1725,6 +1986,36 @@ def _active_authority_keys( ) 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: @@ -1741,10 +2032,8 @@ def _active_authority_keys( result.pop(previous_owner, None) continue public_key_owners[public_key] = key_id - if item.get("status") not in {"active", "next"}: + if 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 ): @@ -1761,9 +2050,12 @@ def _active_authority_keys( continue result[key_id] = ( public_key, - frozenset(str(value) for value in item.get("allowed_scopes", [])), + frozenset(raw_allowed_scopes), ) - return result, tuple(findings) + # 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( @@ -1810,7 +2102,20 @@ def _unchecked_installed_scope() -> dict[str, Any]: ), }, "installed_record_integrity": {"checked": False, "valid": None}, - "installed_source_provenance": {"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,