diff --git a/docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md b/docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md index 220d811..a483104 100644 --- a/docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md +++ b/docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md @@ -360,7 +360,11 @@ windows. Only `active` and `next` entries are trusted; unknown statuses, duplicate key IDs and malformed structured keyrings fail closed, while revoked, disabled and retired entries are ignored. Do not establish this trust file by downloading it in the same rerun. Public JSON responses are bounded to 16 MiB. -Use `--json` or `--output PATH` for automation. +Use `--json` or `--output PATH` for automation. Evidence and report files are +written through a bounded same-directory atomic replacement, with mode `0600` +and file and directory `fsync`. All parent directories must already exist, no +parent component may be a symlink, and an existing symlink or non-regular output +target is rejected. Exit status `0` means the assessed release metadata is current, `2` means explicit review is required, and `1` means the schema or trust checks are @@ -385,7 +389,7 @@ that observation to the assessment and signed catalog: --output /var/tmp/govoplan-fit-review.json ``` -The collector follows the strict +The collector follows the strict version `0.2.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 @@ -395,11 +399,18 @@ entries and evidence-backed conclusions. For each distribution, the collector also: -- verifies every supported SHA-256 entry in installed wheel `RECORD` metadata, +- 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; -- distinguishes immutable VCS commits and archive hashes from editable installs, +- 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; - compares a non-editable VCS commit with the assessment commit and, when one is present, the signed catalog `selected_units` commit; @@ -407,16 +418,29 @@ For each distribution, the collector also: and stable error codes. It never writes direct URLs, paths, hostnames, usernames, exception text or file contents. -`RECORD` agreement proves that the files agree with their installed metadata; it -does not make self-consistent metadata a trusted release origin. Similarly, a -version string is not artifact integrity. Editable and directory-backed installs +`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 +`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. -Use `--installed-evidence PATH` to review evidence collected in a separate -environment. Evidence input and output are bounded to 4 MiB. Collection loads +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 +seconds of future clock skew) at the selected verification time. Use +`--installed-evidence PATH` to compare evidence collected in a separate +environment. Imported JSON is unsigned and therefore always carries a blocker: +it can identify differences but cannot establish checked or valid installed +proof. A future signed collector-attestation format is required to cross that +boundary. The proof scope records observation mode, collection/evaluation time, +freshness, and whether evaluation used current time or an explicit historical +override. Evidence input and output are bounded to 4 MiB. Collection loads the `govoplan.modules` entry-point factories in order to read module IDs and manifest versions; that executes installed GovOPlaN manifest code. Run it only inside the installation being assessed and with the same isolation expected for @@ -453,16 +477,35 @@ the tool's deterministic canonicalization. --trusted-keyring /srv/govoplan/trust/catalog-keyring.json \ --installed-evidence /srv/govoplan/evidence/installed.json \ --boundary-evidence /srv/govoplan/evidence/target-proof.json \ - --boundary-authority-keyring /srv/govoplan/trust/proof-authorities.json + --boundary-authority-keyring /srv/govoplan/trust/proof-authorities.json \ + --expected-external-provider-subject provider-production ``` +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. + +Target-environment and production-approval claims must use the assessment's +`deployment_profile.id` as `subject_id`. External-provider claims require the +operator to supply a bounded opaque expected subject with +`--expected-external-provider-subject`; without it, such a claim remains +unchecked and blocks. Expected and observed IDs are retained in proof scope. + The authority keyring is a governance trust root. Do not download or generate it -from the proof bundle being checked, and do not reuse catalog signing authority -as implicit production authority. A target operator or approver must validate +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 attestation and its bindings; it does not fetch or reinterpret those artifacts. -`--verification-time` exists for reproducible historical review. Live admission -must use the actual current time. +`--verification-time` exists only for reproducible historical review. Supplying +it always marks the machine and human report as a **HISTORICAL override**; callers +cannot relabel it as current. Live admission must omit it and use the actual +current time. No boundary bundle or production authority has been supplied for this current assessment. Target environment, provider and production proof therefore remain diff --git a/docs/installed-composition-evidence.schema.json b/docs/installed-composition-evidence.schema.json index 702e7bd..eb94a1b 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.1.0" + "const": "0.2.0" }, "evidence_kind": { "const": "govoplan.installed-composition" @@ -125,7 +125,7 @@ "required": ["kind", "commit"], "properties": { "kind": { - "enum": ["vcs-commit", "editable-vcs"] + "const": "vcs-commit" }, "commit": { "type": "string", @@ -170,6 +170,7 @@ "status", "hashed_file_count", "permitted_unhashed_file_count", + "generated_unhashed_file_count", "unverifiable_file_count", "missing_file_count", "mismatched_file_count" @@ -190,6 +191,9 @@ "permitted_unhashed_file_count": { "$ref": "#/$defs/count" }, + "generated_unhashed_file_count": { + "$ref": "#/$defs/count" + }, "unverifiable_file_count": { "$ref": "#/$defs/count" }, @@ -199,7 +203,51 @@ "mismatched_file_count": { "$ref": "#/$defs/count" } - } + }, + "allOf": [ + { + "if": { "properties": { "status": { "const": "verified" } } }, + "then": { + "properties": { + "hashed_file_count": { "minimum": 1 }, + "permitted_unhashed_file_count": { "minimum": 1 }, + "unverifiable_file_count": { "const": 0 }, + "missing_file_count": { "const": 0 }, + "mismatched_file_count": { "const": 0 } + } + } + }, + { + "if": { "properties": { "status": { "const": "partial" } } }, + "then": { + "properties": { + "hashed_file_count": { "minimum": 1 }, + "unverifiable_file_count": { "minimum": 1 }, + "missing_file_count": { "const": 0 }, + "mismatched_file_count": { "const": 0 } + } + } + }, + { + "if": { "properties": { "status": { "const": "mismatch" } } }, + "then": { + "anyOf": [ + { "properties": { "missing_file_count": { "minimum": 1 } } }, + { "properties": { "mismatched_file_count": { "minimum": 1 } } } + ] + } + }, + { + "if": { "properties": { "status": { "const": "unavailable" } } }, + "then": { + "properties": { + "hashed_file_count": { "const": 0 }, + "missing_file_count": { "const": 0 }, + "mismatched_file_count": { "const": 0 } + } + } + } + ] }, "collection_issue": { "type": "object", diff --git a/tests/test_capability_fit_atomic_io.py b/tests/test_capability_fit_atomic_io.py new file mode 100644 index 0000000..5ea6544 --- /dev/null +++ b/tests/test_capability_fit_atomic_io.py @@ -0,0 +1,190 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +import stat +import sys +import tempfile +import unittest +from unittest import mock + + +META_ROOT = Path(__file__).resolve().parents[1] +ASSESSMENT_TOOLS_ROOT = META_ROOT / "tools" / "assessments" +RELEASE_TOOLS_ROOT = META_ROOT / "tools" / "release" +for tools_root in (ASSESSMENT_TOOLS_ROOT, RELEASE_TOOLS_ROOT): + if str(tools_root) not in sys.path: + sys.path.insert(0, str(tools_root)) + +from govoplan_assessment.atomic_io import ( # noqa: E402 + AtomicJsonWriteError, + atomic_write_json, +) + + +class CapabilityFitAtomicIOTests(unittest.TestCase): + def test_writes_private_json_with_expected_content(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + target = Path(temporary_directory) / "evidence.json" + payload = {"z": [1, 2], "message": "Grüße"} + created_in: list[Path] = [] + original_mkstemp = tempfile.mkstemp + + def observed_mkstemp(*args, **kwargs): + created_in.append(Path(kwargs["dir"])) + return original_mkstemp(*args, **kwargs) + + with ( + mock.patch( + "govoplan_assessment.atomic_io.tempfile.mkstemp", + side_effect=observed_mkstemp, + ), + mock.patch( + "govoplan_assessment.atomic_io.os.fsync", + wraps=os.fsync, + ) as fsync, + ): + atomic_write_json(target, payload, max_bytes=1024) + + self.assertEqual(payload, json.loads(target.read_text(encoding="utf-8"))) + self.assertTrue(target.read_bytes().endswith(b"\n")) + self.assertEqual(0o600, stat.S_IMODE(target.stat().st_mode)) + self.assertEqual([target.parent], created_in) + self.assertEqual(2, fsync.call_count) + + def test_replaces_existing_regular_file_and_secures_mode(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + target = Path(temporary_directory) / "report.json" + target.write_text('{"old": true}\n', encoding="utf-8") + target.chmod(0o644) + old_handle = target.open("rb") + self.addCleanup(old_handle.close) + + atomic_write_json(target, {"new": True}, max_bytes=1024) + + self.assertEqual(b'{"old": true}\n', old_handle.read()) + self.assertEqual({"new": True}, json.loads(target.read_text("utf-8"))) + self.assertEqual(0o600, stat.S_IMODE(target.stat().st_mode)) + + def test_size_bound_leaves_existing_target_unchanged(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + directory = Path(temporary_directory) + target = directory / "evidence.json" + original = b'{"original": true}\n' + target.write_bytes(original) + + with self.assertRaisesRegex(AtomicJsonWriteError, "size limit"): + atomic_write_json(target, {"large": "x" * 100}, max_bytes=32) + + self.assertEqual(original, target.read_bytes()) + self.assertEqual([], list(directory.glob(".govoplan-json-*.tmp"))) + + @unittest.skipUnless(hasattr(os, "symlink"), "symbolic links are unavailable") + def test_rejects_symlink_without_modifying_link_or_referent(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + directory = Path(temporary_directory) + referent = directory / "outside.json" + referent.write_bytes(b'{"keep": true}\n') + target = directory / "evidence.json" + target.symlink_to(referent.name) + + with self.assertRaisesRegex(AtomicJsonWriteError, "symbolic link"): + atomic_write_json(target, {"replace": True}, max_bytes=1024) + + self.assertTrue(target.is_symlink()) + self.assertEqual(b'{"keep": true}\n', referent.read_bytes()) + self.assertEqual([], list(directory.glob(".govoplan-json-*.tmp"))) + + @unittest.skipUnless(hasattr(os, "symlink"), "symbolic links are unavailable") + def test_rejects_symlinked_parent_component(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + directory = Path(temporary_directory) + real_parent = directory / "real-parent" + real_parent.mkdir() + linked_parent = directory / "linked-parent" + linked_parent.symlink_to(real_parent, target_is_directory=True) + target = linked_parent / "new" / "evidence.json" + + with self.assertRaisesRegex(AtomicJsonWriteError, "parent path"): + atomic_write_json(target, {"unsafe": True}, max_bytes=1024) + + self.assertFalse((real_parent / "new").exists()) + + def test_requires_existing_parent_without_creating_directories(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + directory = Path(temporary_directory) + missing_parent = directory / "missing" / "nested" + target = missing_parent / "evidence.json" + + with self.assertRaisesRegex(AtomicJsonWriteError, "already exist"): + atomic_write_json(target, {"safe": True}, max_bytes=1024) + + self.assertFalse(missing_parent.exists()) + + def test_rejects_post_replace_mode_or_identity_change(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + directory = Path(temporary_directory) + target = directory / "evidence.json" + original_replace = os.replace + + def insecure_replace(source, destination): + original_replace(source, destination) + Path(destination).chmod(0o644) + + with mock.patch( + "govoplan_assessment.atomic_io.os.replace", + side_effect=insecure_replace, + ): + with self.assertRaisesRegex( + AtomicJsonWriteError, + "permissions changed and were restored", + ): + atomic_write_json(target, {"safe": True}, max_bytes=1024) + + self.assertEqual(0o600, stat.S_IMODE(target.stat().st_mode)) + self.assertEqual({"safe": True}, json.loads(target.read_text("utf-8"))) + + def test_does_not_unlink_unknown_post_replace_inode(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + directory = Path(temporary_directory) + target = directory / "evidence.json" + original_replace = os.replace + + def replaced_again(source, destination): + original_replace(source, destination) + Path(destination).unlink() + Path(destination).write_bytes(b"unknown replacement\n") + + with mock.patch( + "govoplan_assessment.atomic_io.os.replace", + side_effect=replaced_again, + ): + with self.assertRaisesRegex( + AtomicJsonWriteError, + "did not preserve", + ): + atomic_write_json(target, {"secret": True}, max_bytes=1024) + + self.assertEqual(b"unknown replacement\n", target.read_bytes()) + + def test_replace_failure_removes_private_temporary_file(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + directory = Path(temporary_directory) + target = directory / "evidence.json" + + with mock.patch( + "govoplan_assessment.atomic_io.os.replace", + side_effect=OSError("simulated replacement failure"), + ): + with self.assertRaisesRegex( + AtomicJsonWriteError, "could not be written atomically" + ): + atomic_write_json(target, {"safe": True}, max_bytes=1024) + + self.assertFalse(target.exists()) + self.assertEqual([], list(directory.glob(".govoplan-json-*.tmp"))) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_capability_fit_evidence.py b/tests/test_capability_fit_evidence.py index 9ffc880..272626c 100644 --- a/tests/test_capability_fit_evidence.py +++ b/tests/test_capability_fit_evidence.py @@ -7,6 +7,7 @@ import hashlib from importlib import metadata import json from pathlib import Path +import subprocess import sys import tempfile from types import SimpleNamespace @@ -24,8 +25,12 @@ for tools_root in (ASSESSMENT_TOOLS_ROOT, RELEASE_TOOLS_ROOT): if str(tools_root) not in sys.path: sys.path.insert(0, str(tools_root)) -from govoplan_assessment.capability_fit import review_capability_fit # noqa: E402 +from govoplan_assessment.capability_fit import ( # noqa: E402 + render_review, + review_capability_fit, +) from govoplan_assessment.evidence import ( # noqa: E402 + _bounded_file_sha256, canonical_bytes, canonical_sha256, collect_installed_composition, @@ -205,6 +210,8 @@ class CapabilityFitEvidenceTests(unittest.TestCase): selected_commits={}, evidence=evidence, schema=self.installed_schema, + observation_mode="direct_local", + verification_time=datetime(2026, 7, 23, 12, tzinfo=UTC), ) codes = {item.code for item in result.findings} @@ -213,6 +220,86 @@ class CapabilityFitEvidenceTests(unittest.TestCase): self.assertTrue(result.proof_scope["installed_artifacts"]["checked"]) self.assertFalse(result.proof_scope["installed_artifacts"]["valid"]) + def test_imported_unsigned_installed_evidence_is_comparison_only(self) -> None: + catalog, keyring = signed_catalog(self.assessment) + evidence = matching_installed_evidence(self.assessment) + + report = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=evidence, + installed_evidence_mode="imported_unsigned", + ) + + installed = report["proof_scope"]["installed_artifacts"] + self.assertEqual("blocked", report["status"]) + self.assertFalse(installed["checked"]) + self.assertIsNone(installed["valid"]) + self.assertTrue(installed["comparison_valid"]) + self.assertEqual( + "unsigned_import", + report["proof_scope"]["installed_evidence_observation"]["freshness"], + ) + self.assertIn( + "installed_evidence_unsigned_import", + {item["code"] for item in report["findings"]}, + ) + + def test_direct_installed_evidence_must_be_fresh_and_not_from_future(self) -> None: + catalog, keyring = signed_catalog(self.assessment) + cases = ( + ("2026-07-23T11:54:59Z", "installed_evidence_stale", "stale"), + ("2026-07-23T12:00:31Z", "installed_evidence_from_future", "future"), + ) + for collected_at, expected_code, expected_freshness in cases: + with self.subTest(collected_at=collected_at): + evidence = matching_installed_evidence(self.assessment) + evidence["collected_at"] = collected_at + report = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=evidence, + ) + + self.assertEqual("blocked", report["status"]) + self.assertFalse( + report["proof_scope"]["installed_artifacts"]["checked"] + ) + self.assertIn( + expected_code, + {item["code"] for item in report["findings"]}, + ) + self.assertEqual( + expected_freshness, + report["proof_scope"]["installed_evidence_observation"][ + "freshness" + ], + ) + + def test_forged_verified_record_counters_fail_semantic_validation(self) -> None: + catalog, keyring = signed_catalog(self.assessment) + evidence = matching_installed_evidence(self.assessment) + evidence["artifacts"][0]["record_integrity"].update( + hashed_file_count=0, + mismatched_file_count=1, + ) + relaxed_schema = deepcopy(self.installed_schema) + relaxed_schema["$defs"]["record_integrity"].pop("allOf") + + report = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=evidence, + installed_schema=relaxed_schema, + ) + + self.assertEqual("review_required", report["status"]) + self.assertIn( + "installed_record_integrity_invalid", + {item["code"] for item in report["findings"]}, + ) + self.assertFalse(report["proof_scope"]["installed_record_integrity"]["valid"]) + def test_collector_verifies_record_and_redacts_direct_url(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) @@ -251,6 +338,225 @@ class CapabilityFitEvidenceTests(unittest.TestCase): tampered["artifacts"][0]["record_integrity"]["mismatched_file_count"], ) + def test_collector_accepts_real_core_wheel_payload_scripts_data_and_pyc( + self, + ) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + wheels = root / "wheels" + prefix = root / "prefix" + wheels.mkdir() + build_python = ( + Path(sys.base_prefix) + / "bin" + / (f"python{sys.version_info.major}.{sys.version_info.minor}") + ) + subprocess.run( + ( + str(build_python), + "-m", + "pip", + "wheel", + "--no-deps", + "--no-build-isolation", + "--wheel-dir", + str(wheels), + str(META_ROOT.parent / "govoplan-core"), + ), + check=True, + capture_output=True, + text=True, + ) + wheel = next(wheels.glob("govoplan_core-*.whl")) + subprocess.run( + ( + str(build_python), + "-m", + "pip", + "install", + "--no-deps", + "--ignore-installed", + "--prefix", + str(prefix), + str(wheel), + ), + check=True, + capture_output=True, + text=True, + ) + site_packages = next(prefix.glob("lib/python*/site-packages")) + dist_info = next(site_packages.glob("govoplan_core-*.dist-info")) + distribution = metadata.PathDistribution(dist_info) + record_paths = {str(item) for item in distribution.files or ()} + self.assertTrue( + any( + path.startswith("../../../govoplan_core_runtime/") + for path in record_paths + ) + ) + self.assertTrue( + any(path.startswith("../../../bin/govoplan-") for path in record_paths) + ) + with ( + mock.patch.object(sys, "path", [str(site_packages), *sys.path]), + mock.patch.object(sys, "prefix", str(prefix)), + ): + evidence = collect_installed_composition( + assessment=self.assessment, + collected_at=datetime(2026, 7, 23, 11, 58, tzinfo=UTC), + distributions=[distribution], + ) + + record = evidence["artifacts"][0]["record_integrity"] + self.assertEqual("verified", record["status"]) + self.assertGreater(record["hashed_file_count"], 0) + self.assertGreater(record["permitted_unhashed_file_count"], 0) + self.assertGreater(record["generated_unhashed_file_count"], 0) + self.assertEqual(0, record["unverifiable_file_count"]) + self.assertEqual( + (), validate_payload(payload=evidence, schema=self.installed_schema) + ) + + def test_record_traversal_and_leaf_symlinks_are_never_hashed(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + prefix = Path(temp_dir) / "prefix" + site_packages = prefix / "lib" / "python-test" / "site-packages" + site_packages.mkdir(parents=True) + traversal = create_traversal_distribution(prefix, site_packages) + normal, payload = create_distribution(site_packages) + outside = Path(temp_dir) / "outside.py" + outside.write_text(payload.read_text(encoding="utf-8"), encoding="utf-8") + payload.unlink() + payload.symlink_to(outside) + with ( + mock.patch.object(sys, "path", [str(site_packages), *sys.path]), + mock.patch.object(sys, "prefix", str(prefix)), + ): + evidence = collect_installed_composition( + assessment=self.assessment, + collected_at=datetime(2026, 7, 23, 11, 58, tzinfo=UTC), + distributions=[traversal, normal], + ) + + records = { + item["package_name"]: item["record_integrity"] + for item in evidence["artifacts"] + } + self.assertEqual("partial", records["govoplan-traversal"]["status"]) + self.assertGreaterEqual( + records["govoplan-traversal"]["unverifiable_file_count"], 2 + ) + self.assertEqual("partial", records["govoplan-demo"]["status"]) + self.assertGreater(records["govoplan-demo"]["unverifiable_file_count"], 0) + + def test_bounded_hash_stops_at_opened_file_snapshot(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + path = Path(temp_dir) / "growing.bin" + path.write_bytes(b"0123456789") + actual_stat = path.stat() + initial_stat = SimpleNamespace(st_mode=actual_stat.st_mode, st_size=4) + final_stat = SimpleNamespace(st_mode=actual_stat.st_mode, st_size=10) + with mock.patch( + "govoplan_assessment.evidence.os.fstat", + side_effect=(initial_stat, final_stat), + ): + _, hashed_bytes, exceeded = _bounded_file_sha256(path, max_bytes=8) + + self.assertEqual(4, hashed_bytes) + self.assertTrue(exceeded) + + def test_collector_rejects_unbounded_or_secret_shaped_metadata(self) -> None: + distributions = [ + FakeDistribution( + name="govoplan-secret-user@private-host/path", + entry_points=[], + ), + FakeDistribution( + version="1.0.0/private-user@host", + entry_points=[], + ), + FakeDistribution( + entry_points=[ + FakeEntryPoint( + name="safe-module", + manifest_id="secret-user@private-host/path", + ) + ], + ), + ] + + evidence = collect_installed_composition( + assessment=self.assessment, + collected_at=datetime(2026, 7, 23, 11, 58, tzinfo=UTC), + distributions=distributions, + ) + + encoded = json.dumps(evidence, sort_keys=True) + self.assertNotIn("secret-user", encoded) + self.assertNotIn("private-host", encoded) + self.assertNotIn("private-user", encoded) + self.assertTrue( + any( + item["code"] + in { + "distribution-metadata-invalid", + "module-entry-point-invalid", + } + for item in evidence["collection_issues"] + ) + ) + + def test_pep610_git_provenance_requires_one_coherent_git_form(self) -> None: + valid_commit = "a" * 40 + cases = ( + ( + "valid", + { + "url": "https://example.invalid/govoplan-many.git", + "vcs_info": {"vcs": "git", "commit_id": valid_commit}, + }, + "vcs-commit", + ), + ( + "svn", + { + "url": "https://example.invalid/govoplan-many.git", + "vcs_info": {"vcs": "svn", "commit_id": valid_commit}, + }, + "malformed-direct-url", + ), + ( + "missing-vcs", + { + "url": "https://example.invalid/govoplan-many.git", + "vcs_info": {"commit_id": valid_commit}, + }, + "malformed-direct-url", + ), + ( + "ambiguous", + { + "url": "file:///govoplan-many", + "vcs_info": {"vcs": "git", "commit_id": valid_commit}, + "dir_info": {"editable": True}, + }, + "malformed-direct-url", + ), + ) + for label, direct_url, expected_kind in cases: + with self.subTest(label=label): + evidence = collect_installed_composition( + assessment=self.assessment, + collected_at=datetime(2026, 7, 23, 11, 58, tzinfo=UTC), + distributions=[ + FakeDistribution(entry_points=[], direct_url=direct_url) + ], + ) + self.assertEqual( + expected_kind, + evidence["artifacts"][0]["source_provenance"]["kind"], + ) + def test_entry_point_limit_is_explicit_instead_of_silent_truncation(self) -> None: distribution = FakeDistribution( entry_points=[FakeEntryPoint(name=f"module-{index}") for index in range(17)] @@ -320,6 +626,248 @@ class CapabilityFitEvidenceTests(unittest.TestCase): self.assertTrue(report["proof_scope"]["target_environment"]["valid"]) self.assertFalse(report["proof_scope"]["external_providers"]["checked"]) self.assertFalse(report["proof_scope"]["production_approval"]["checked"]) + self.assertEqual( + self.assessment["deployment_profile"]["id"], + report["proof_scope"]["target_environment"]["expected_subject_id"], + ) + self.assertEqual( + self.assessment["deployment_profile"]["id"], + report["proof_scope"]["target_environment"]["observed_subject_id"], + ) + + def test_external_provider_claim_requires_explicit_matching_subject(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("external_providers", "passed")], + allowed_scopes=["external_providers"], + ) + + unconfigured = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=installed, + boundary_evidence=proof, + authority_keyring=authority, + ) + accepted = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=installed, + boundary_evidence=proof, + authority_keyring=authority, + expected_external_provider_subject="provider:test", + ) + + self.assertEqual("blocked", unconfigured["status"]) + self.assertIn( + "boundary_provider_subject_unconfigured", + {item["code"] for item in unconfigured["findings"]}, + ) + self.assertEqual("current", accepted["status"]) + provider_scope = accepted["proof_scope"]["external_providers"] + self.assertEqual("provider:test", provider_scope["expected_subject_id"]) + self.assertEqual("provider:test", provider_scope["observed_subject_id"]) + + def test_boundary_subject_mismatch_blocks_without_echoing_invalid_config( + 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", + subject_id="another-deployment", + ) + ], + allowed_scopes=["target_environment"], + ) + mismatch = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=installed, + boundary_evidence=proof, + authority_keyring=authority, + ) + invalid_config = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=installed, + expected_external_provider_subject="https://private.invalid/secret", + ) + + self.assertIn( + "boundary_claim_subject_mismatch", + {item["code"] for item in mismatch["findings"]}, + ) + encoded = json.dumps(invalid_config, sort_keys=True) + self.assertIn("boundary_provider_subject_invalid", encoded) + self.assertNotIn("private.invalid", encoded) + self.assertNotIn("secret", encoded) + + def test_authorized_negative_claim_creates_scope_specific_review_target( + 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", "failed"), + boundary_claim("production_approval", "rejected"), + ], + allowed_scopes=["target_environment", "production_approval"], + ) + + report = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=installed, + boundary_evidence=proof, + authority_keyring=authority, + ) + + targets = {item["id"] for item in report["review_targets"]} + self.assertIn( + "boundary.target_environment.production-like-dev", + targets, + ) + self.assertIn( + "boundary.production_approval.production-like-dev", + targets, + ) + + def test_proof_authorities_must_be_independent_and_key_ids_unambiguous( + 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"], + ) + release_public_key = keyring["keys"][0]["public_key"] + reused = deepcopy(authority) + reused["keys"][0]["public_key"] = release_public_key + duplicate = deepcopy(authority) + duplicate["keys"].append( + { + **deepcopy(duplicate["keys"][0]), + "key_id": "authority:duplicate-material", + "allowed_scopes": ["production_approval"], + } + ) + + reused_report = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=installed, + boundary_evidence=proof, + authority_keyring=reused, + ) + duplicate_report = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=installed, + boundary_evidence=proof, + authority_keyring=duplicate, + ) + + self.assertIn( + "boundary_authority_reuses_release_key", + {item["code"] for item in reused_report["findings"]}, + ) + self.assertIn( + "boundary_authority_public_key_duplicate", + {item["code"] for item in duplicate_report["findings"]}, + ) + self.assertFalse( + duplicate_report["proof_scope"]["target_environment"]["checked"] + ) + + def test_authority_not_after_is_exclusive(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"], + ) + authority["keys"][0]["not_after"] = "2026-07-23T12:00:00Z" + + report = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=installed, + boundary_evidence=proof, + authority_keyring=authority, + ) + + self.assertIn( + "boundary_authority_untrusted", + {item["code"] for item in report["findings"]}, + ) + + def test_historical_verification_and_generated_runtime_gap_are_visible( + self, + ) -> None: + catalog, keyring = signed_catalog(self.assessment) + evidence = matching_installed_evidence(self.assessment) + evidence["artifacts"][0]["record_integrity"][ + "generated_unhashed_file_count" + ] = 3 + + report = self.review( + catalog=catalog, + keyring=keyring, + installed_evidence=evidence, + ) + + observation = report["proof_scope"]["installed_evidence_observation"] + self.assertEqual("historical_override", observation["verification_mode"]) + self.assertEqual( + 3, + report["proof_scope"]["installed_record_integrity"][ + "generated_unhashed_file_count" + ], + ) + self.assertIn("HISTORICAL override", render_review(report)) + + def test_invalid_internal_verification_mode_is_bounded_and_blocking(self) -> None: + evidence = matching_installed_evidence(self.assessment) + result = review_installed_composition( + assessment=self.assessment, + catalog_entries={}, + selected_versions={}, + selected_commits={}, + evidence=evidence, + schema=self.installed_schema, + observation_mode="direct_local", + verification_time=datetime(2026, 7, 23, 12, tzinfo=UTC), + verification_mode="https://private.invalid/secret", + ) + + encoded = json.dumps(result.proof_scope, sort_keys=True) + self.assertIn( + "installed_evidence_verification_mode", + {item.code for item in result.findings}, + ) + self.assertEqual( + "invalid", + result.proof_scope["installed_evidence_observation"]["verification_mode"], + ) + self.assertFalse(result.proof_scope["installed_artifacts"]["checked"]) + self.assertNotIn("private.invalid", encoded) def test_unauthorized_production_claim_keeps_every_claim_unchecked(self) -> None: catalog, keyring = signed_catalog(self.assessment) @@ -401,6 +949,10 @@ class CapabilityFitEvidenceTests(unittest.TestCase): installed_evidence: dict[str, object] | None = None, boundary_evidence: dict[str, object] | None = None, authority_keyring: dict[str, object] | None = None, + installed_evidence_mode: str = "direct_local", + installed_schema: dict[str, object] | None = None, + expected_external_provider_subject: str | None = None, + verification_time: datetime | None = datetime(2026, 7, 23, 12, tzinfo=UTC), ) -> dict[str, object]: return review_capability_fit( assessment=deepcopy(self.assessment), @@ -409,7 +961,7 @@ class CapabilityFitEvidenceTests(unittest.TestCase): published_keyring=keyring, trusted_keyring=keyring, installed_evidence=installed_evidence, - installed_evidence_schema=self.installed_schema + installed_evidence_schema=(installed_schema or self.installed_schema) if installed_evidence is not None else None, boundary_evidence=boundary_evidence, @@ -420,18 +972,20 @@ class CapabilityFitEvidenceTests(unittest.TestCase): boundary_authority_keyring_schema=self.authority_schema if boundary_evidence is not None else None, - verification_time=datetime(2026, 7, 23, 12, tzinfo=UTC), + verification_time=verification_time, + installed_evidence_mode=installed_evidence_mode, + expected_external_provider_subject=expected_external_provider_subject, ) def matching_installed_evidence(assessment: dict[str, object]) -> dict[str, object]: return { "$schema": "./installed-composition-evidence.schema.json", - "schema_version": "0.1.0", + "schema_version": "0.2.0", "evidence_kind": "govoplan.installed-composition", "assessment_id": assessment["assessment_id"], "assessment_release": assessment["release"]["ref"], - "collected_at": "2026-07-22T12:00:00Z", + "collected_at": "2026-07-23T11:58:00Z", "scope": "current-python-environment.govoplan-distributions", "artifacts": [ artifact( @@ -461,6 +1015,7 @@ def artifact( "status": "verified", "hashed_file_count": 1, "permitted_unhashed_file_count": 1, + "generated_unhashed_file_count": 0, "unverifiable_file_count": 0, "missing_file_count": 0, "mismatched_file_count": 0, @@ -468,11 +1023,19 @@ def artifact( } -def boundary_claim(scope: str, result: str) -> dict[str, object]: +def boundary_claim( + scope: str, + result: str, + *, + subject_id: str | None = None, +) -> dict[str, object]: + default_subject = ( + "provider:test" if scope == "external_providers" else "production-like-dev" + ) return { "scope": scope, "result": result, - "subject_id": f"subject:{scope}", + "subject_id": subject_id or default_subject, "control_ids": [f"control:{scope}"], "artifacts": [ { @@ -579,19 +1142,72 @@ def create_distribution( return metadata.PathDistribution(dist_info), payload_file +def create_traversal_distribution( + prefix: Path, + site_packages: Path, +) -> metadata.PathDistribution: + dist_info = site_packages / "govoplan_traversal-1.0.0.dist-info" + dist_info.mkdir() + metadata_file = dist_info / "METADATA" + metadata_file.write_text( + "Metadata-Version: 2.1\nName: govoplan-traversal\nVersion: 1.0.0\n", + encoding="utf-8", + ) + unowned_file = prefix / "unowned_runtime" / "private.py" + unowned_file.parent.mkdir() + unowned_file.write_text("private = True\n", encoding="utf-8") + undeclared_script = prefix / "bin" / "govoplan-not-declared" + undeclared_script.parent.mkdir() + undeclared_script.write_text("#!/bin/sh\n", encoding="utf-8") + metadata_path = metadata_file.relative_to(site_packages).as_posix() + unowned_path = "../../../unowned_runtime/private.py" + record_path = (dist_info / "RECORD").relative_to(site_packages).as_posix() + (dist_info / "RECORD").write_text( + "\n".join( + ( + record_row(metadata_path, metadata_file.read_bytes()), + record_row(unowned_path, unowned_file.read_bytes()), + record_row( + "../../../bin/govoplan-not-declared", + undeclared_script.read_bytes(), + ), + f"{record_path},,", + ) + ) + + "\n", + encoding="utf-8", + ) + return metadata.PathDistribution(dist_info) + + +def record_row(path: str, content: bytes) -> str: + digest = base64.urlsafe_b64encode(hashlib.sha256(content).digest()).decode("ascii") + return f"{path},sha256={digest.rstrip('=')},{len(content)}" + + class FakeEntryPoint: group = "govoplan.modules" value = "fixture:get_manifest" - def __init__(self, *, name: str) -> None: + def __init__( + self, + *, + name: str, + manifest_id: str | None = None, + manifest_version: str = "1.0.0", + ) -> None: self.name = name + self.manifest_id = manifest_id or name + self.manifest_version = manifest_version def load(self): - return lambda: SimpleNamespace(id=self.name, version="1.0.0") + return lambda: SimpleNamespace( + id=self.manifest_id, + version=self.manifest_version, + ) class FakeDistribution: - version = "1.0.0" files = () def __init__( @@ -599,8 +1215,11 @@ class FakeDistribution: *, entry_points: list[FakeEntryPoint], direct_url: dict[str, object] | None = None, + name: str = "govoplan-many", + version: str = "1.0.0", ) -> None: - self.metadata = {"Name": "govoplan-many"} + self.metadata = {"Name": name} + self.version = version self.entry_points = entry_points self.direct_url = direct_url diff --git a/tools/assessments/capability-fit.py b/tools/assessments/capability-fit.py index 421b5bb..dc1bbe4 100755 --- a/tools/assessments/capability-fit.py +++ b/tools/assessments/capability-fit.py @@ -7,6 +7,7 @@ import argparse import json import os from pathlib import Path +import re import sys @@ -22,6 +23,10 @@ from govoplan_assessment import ( # noqa: E402 render_review, review_capability_fit, ) +from govoplan_assessment.atomic_io import ( # noqa: E402 + AtomicJsonWriteError, + atomic_write_json, +) from govoplan_assessment.evidence import validate_payload # noqa: E402 from govoplan_release.catalog import DEFAULT_PUBLIC_BASE_URL, fetch_json # noqa: E402 @@ -29,6 +34,8 @@ from govoplan_release.catalog import DEFAULT_PUBLIC_BASE_URL, fetch_json # noqa TRUSTED_KEYRING_FILE_ENV = "GOVOPLAN_MODULE_PACKAGE_CATALOG_TRUSTED_KEYS_FILE" MAX_ASSESSMENT_INPUT_BYTES = 16 * 1024 * 1024 MAX_EVIDENCE_INPUT_BYTES = 4 * 1024 * 1024 +MAX_REPORT_OUTPUT_BYTES = 16 * 1024 * 1024 +OPAQUE_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,159}$") def parse_args(argv: list[str] | None = None) -> argparse.Namespace: @@ -130,6 +137,13 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: "--verification-time", help="Optional ISO-8601 time for reproducible boundary-proof verification.", ) + parser.add_argument( + "--expected-external-provider-subject", + help=( + "Opaque provider-environment ID that external-provider proof must match; " + "never supply a URL, credential, or endpoint identifier." + ), + ) parser.add_argument( "--json", action="store_true", help="Print the machine-readable review report." ) @@ -157,6 +171,13 @@ def main(argv: list[str] | None = None) -> int: raise SystemExit( "--boundary-evidence requires --installed-evidence or --collect-installed-evidence" ) + if ( + args.expected_external_provider_subject is not None + and OPAQUE_ID_PATTERN.fullmatch(args.expected_external_provider_subject) is None + ): + raise SystemExit( + "--expected-external-provider-subject must be a bounded opaque ID" + ) if ( args.collect_installed_evidence is not None and args.output is not None @@ -232,7 +253,12 @@ def main(argv: list[str] | None = None) -> int: raise SystemExit( "Collected installed evidence did not satisfy its bounded schema" ) - write_object(args.collect_installed_evidence, installed_evidence) + write_object( + args.collect_installed_evidence, + installed_evidence, + max_bytes=MAX_EVIDENCE_INPUT_BYTES, + label="installed evidence", + ) boundary_evidence = None boundary_evidence_schema = None @@ -260,6 +286,7 @@ def main(argv: list[str] | None = None) -> int: max_bytes=MAX_EVIDENCE_INPUT_BYTES, ) + verification_time = parse_datetime(args.verification_time) report = review_capability_fit( assessment=assessment, schema=schema, @@ -275,12 +302,22 @@ def main(argv: list[str] | None = None) -> int: boundary_evidence_schema=boundary_evidence_schema, boundary_authority_keyring=boundary_authority_keyring, boundary_authority_keyring_schema=boundary_authority_keyring_schema, - verification_time=parse_datetime(args.verification_time), + verification_time=verification_time, + installed_evidence_mode=( + "direct_local" + if args.collect_installed_evidence is not None + else "imported_unsigned" + ), + expected_external_provider_subject=args.expected_external_provider_subject, ) encoded = json.dumps(report, indent=2, sort_keys=True) + "\n" if args.output is not None: - args.output.parent.mkdir(parents=True, exist_ok=True) - args.output.write_text(encoded, encoding="utf-8") + write_object( + args.output, + report, + max_bytes=MAX_REPORT_OUTPUT_BYTES, + label="review report", + ) sys.stdout.write(encoded if args.json else render_review(report)) return {"current": 0, "blocked": 1, "review_required": 2}[report["status"]] @@ -307,16 +344,19 @@ def read_object(path: Path | None, *, label: str, max_bytes: int) -> dict[str, o return payload -def write_object(path: Path | None, payload: dict[str, object]) -> None: +def write_object( + path: Path | None, + payload: dict[str, object], + *, + max_bytes: int, + label: str, +) -> None: if path is None: - raise SystemExit("Missing installed evidence output path") - encoded = json.dumps(payload, indent=2, sort_keys=True) + "\n" - if len(encoded.encode("utf-8")) > MAX_EVIDENCE_INPUT_BYTES: - raise SystemExit( - "Collected installed evidence exceeds the bounded output limit" - ) - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(encoded, encoding="utf-8") + raise SystemExit(f"Missing {label} output path") + try: + atomic_write_json(path, payload, max_bytes=max_bytes) + except AtomicJsonWriteError as exc: + raise SystemExit(f"Could not securely write {label}") from exc def parse_datetime(value: str | None): diff --git a/tools/assessments/govoplan_assessment/atomic_io.py b/tools/assessments/govoplan_assessment/atomic_io.py new file mode 100644 index 0000000..56a7202 --- /dev/null +++ b/tools/assessments/govoplan_assessment/atomic_io.py @@ -0,0 +1,230 @@ +"""Bounded, private, atomic JSON output for assessment evidence.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +import stat +import tempfile +from typing import Any + + +class AtomicJsonWriteError(RuntimeError): + """Raised when JSON output cannot be written with the required guarantees.""" + + +def atomic_write_json( + path: str | os.PathLike[str], + payload: Any, + *, + max_bytes: int, +) -> None: + """Write bounded JSON through a private same-directory temporary file. + + A successful return means the file contents and the containing directory + entry have both been flushed. If flushing the directory fails after the + atomic replacement, the replacement may be visible but its crash + durability is unknown, so the function reports failure. Every parent must + already exist as a real directory; symbolic-link components are rejected. + """ + + if isinstance(max_bytes, bool) or not isinstance(max_bytes, int) or max_bytes <= 0: + raise ValueError("max_bytes must be a positive integer") + + encoded = _encode_bounded_json(payload, max_bytes=max_bytes) + target = Path(os.path.abspath(os.fspath(path))) + parent = target.parent + descriptor = -1 + verification_descriptor = -1 + temporary_path: Path | None = None + temporary_identity: tuple[int, int] | None = None + + try: + _assert_real_parent(parent) + _assert_replaceable_target(target) + try: + descriptor, temporary_name = tempfile.mkstemp( + dir=parent, + prefix=".govoplan-json-", + suffix=".tmp", + ) + temporary_path = Path(temporary_name) + os.fchmod(descriptor, 0o600) + metadata = os.fstat(descriptor) + if ( + not stat.S_ISREG(metadata.st_mode) + or stat.S_IMODE(metadata.st_mode) != 0o600 + ): + raise AtomicJsonWriteError( + "Atomic JSON temporary file could not be secured." + ) + temporary_identity = (metadata.st_dev, metadata.st_ino) + verification_descriptor = os.dup(descriptor) + + handle = os.fdopen(descriptor, "wb") + descriptor = -1 + with handle: + handle.write(encoded) + handle.flush() + os.fsync(handle.fileno()) + + # Recheck after creating and flushing the temporary file so an + # unsafe target discovered before replacement is never followed. + _assert_real_parent(parent) + _assert_replaceable_target(target) + os.replace(temporary_path, target) + temporary_path = None + _assert_installed_target( + target, + expected_identity=temporary_identity, + recovery_descriptor=verification_descriptor, + ) + _fsync_directory(parent) + finally: + try: + if descriptor >= 0: + os.close(descriptor) + finally: + try: + if verification_descriptor >= 0: + os.close(verification_descriptor) + finally: + if temporary_path is not None: + try: + temporary_path.unlink() + except FileNotFoundError: + pass + except AtomicJsonWriteError: + raise + except OSError as exc: + raise AtomicJsonWriteError( + "JSON output could not be written atomically." + ) from exc + + +def _encode_bounded_json(payload: Any, *, max_bytes: int) -> bytes: + encoder = json.JSONEncoder( + allow_nan=False, + ensure_ascii=False, + indent=2, + sort_keys=True, + ) + chunks: list[bytes] = [] + encoded_size = 1 # The document always ends with one newline. + for chunk in encoder.iterencode(payload): + encoded_chunk = chunk.encode("utf-8") + encoded_size += len(encoded_chunk) + if encoded_size > max_bytes: + raise AtomicJsonWriteError("JSON output exceeds its size limit.") + chunks.append(encoded_chunk) + return b"".join(chunks) + b"\n" + + +def _assert_replaceable_target(path: Path) -> None: + try: + metadata = path.lstat() + except FileNotFoundError: + return + except OSError as exc: + raise AtomicJsonWriteError( + "JSON output target could not be inspected safely." + ) from exc + if stat.S_ISLNK(metadata.st_mode): + raise AtomicJsonWriteError("JSON output target must not be a symbolic link.") + if not stat.S_ISREG(metadata.st_mode): + raise AtomicJsonWriteError("JSON output target must be a regular file.") + + +def _assert_real_parent(path: Path) -> None: + current = Path(path.anchor) + for part in path.parts[1:]: + current /= part + try: + metadata = current.lstat() + except FileNotFoundError as exc: + raise AtomicJsonWriteError( + "JSON output parent directory must already exist." + ) from exc + except OSError as exc: + raise AtomicJsonWriteError( + "JSON output parent directory could not be inspected safely." + ) from exc + if stat.S_ISLNK(metadata.st_mode): + raise AtomicJsonWriteError( + "JSON output parent path must not contain symbolic links." + ) + if not stat.S_ISDIR(metadata.st_mode): + raise AtomicJsonWriteError( + "JSON output parent path must contain only directories." + ) + + +def _assert_installed_target( + path: Path, + *, + expected_identity: tuple[int, int] | None, + recovery_descriptor: int, +) -> None: + try: + metadata = path.lstat() + except OSError as exc: + raise AtomicJsonWriteError( + "Atomic JSON replacement could not be verified." + ) from exc + observed_identity = (metadata.st_dev, metadata.st_ino) + if ( + expected_identity is None + or observed_identity != expected_identity + or not stat.S_ISREG(metadata.st_mode) + ): + raise AtomicJsonWriteError( + "Atomic JSON replacement did not preserve the secured regular file." + ) + if stat.S_IMODE(metadata.st_mode) == 0o600: + return + try: + recovery_metadata = os.fstat(recovery_descriptor) + if ( + not stat.S_ISREG(recovery_metadata.st_mode) + or (recovery_metadata.st_dev, recovery_metadata.st_ino) != expected_identity + ): + raise AtomicJsonWriteError( + "Atomic JSON replacement recovery descriptor is invalid." + ) + os.fchmod(recovery_descriptor, 0o600) + os.fsync(recovery_descriptor) + recovered_metadata = os.fstat(recovery_descriptor) + installed_metadata = path.lstat() + except OSError as exc: + raise AtomicJsonWriteError( + "Atomic JSON replacement permissions could not be restored." + ) from exc + if ( + stat.S_IMODE(recovered_metadata.st_mode) != 0o600 + or (installed_metadata.st_dev, installed_metadata.st_ino) != expected_identity + or not stat.S_ISREG(installed_metadata.st_mode) + or stat.S_IMODE(installed_metadata.st_mode) != 0o600 + ): + raise AtomicJsonWriteError( + "Atomic JSON replacement permissions could not be restored." + ) + _fsync_directory(path.parent) + raise AtomicJsonWriteError( + "Atomic JSON replacement permissions changed and were restored." + ) + + +def _fsync_directory(path: Path) -> None: + flags = os.O_RDONLY + if hasattr(os, "O_DIRECTORY"): + flags |= os.O_DIRECTORY + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + descriptor = os.open(path, flags) + try: + if not stat.S_ISDIR(os.fstat(descriptor).st_mode): + raise AtomicJsonWriteError("JSON output parent path must be a directory.") + os.fsync(descriptor) + finally: + os.close(descriptor) diff --git a/tools/assessments/govoplan_assessment/capability_fit.py b/tools/assessments/govoplan_assessment/capability_fit.py index 97c621e..de89ac4 100644 --- a/tools/assessments/govoplan_assessment/capability_fit.py +++ b/tools/assessments/govoplan_assessment/capability_fit.py @@ -28,6 +28,7 @@ from govoplan_release.version_alignment import candidate_catalog_version_issues from .evidence import ( BoundaryEvidenceReview, InstalledEvidenceReview, + public_key_material_from_keyrings, review_boundary_evidence, review_installed_composition, ) @@ -63,6 +64,8 @@ def review_capability_fit( boundary_authority_keyring: dict[str, Any] | None = None, boundary_authority_keyring_schema: dict[str, Any] | None = None, verification_time: datetime | None = None, + installed_evidence_mode: str = "imported_unsigned", + expected_external_provider_subject: str | None = None, ) -> dict[str, Any]: """Return a secret-free release-drift report for one fit assessment.""" @@ -356,6 +359,10 @@ def review_capability_fit( ) ) + evidence_verification_time = verification_time or datetime.now(tz=UTC) + evidence_verification_mode = ( + "historical_override" if verification_time is not None else "current" + ) installed_review = review_installed_composition( assessment=assessment, catalog_entries=catalog_entries, @@ -363,6 +370,9 @@ def review_capability_fit( selected_commits=source_selection.selected_commits, evidence=installed_evidence, schema=installed_evidence_schema, + observation_mode=installed_evidence_mode, + verification_time=evidence_verification_time, + verification_mode=evidence_verification_mode, ) findings.extend( Finding(item.severity, item.code, item.message, item.assessment_ids) @@ -378,7 +388,13 @@ def review_capability_fit( evidence_schema=boundary_evidence_schema, authority_keyring=boundary_authority_keyring, authority_keyring_schema=boundary_authority_keyring_schema, - verification_time=verification_time, + verification_time=evidence_verification_time, + verification_mode=evidence_verification_mode, + expected_external_provider_subject=expected_external_provider_subject, + forbidden_authority_public_keys=public_key_material_from_keyrings( + published_keyring, + trusted_keyring, + ), ) findings.extend( Finding(item.severity, item.code, item.message, item.assessment_ids) @@ -396,6 +412,7 @@ def review_capability_fit( assessment=assessment, installed_review=installed_review, ), + boundary_review.review_targets, ) return build_report( assessment=assessment, @@ -908,7 +925,7 @@ def build_report( **boundary_scope, } return { - "report_version": "0.2.0", + "report_version": "0.3.0", "status": status, "assessment_id": assessment.get("assessment_id"), "assessment_release": assessment.get("release", {}).get("ref") @@ -930,9 +947,11 @@ def build_report( def render_review(report: dict[str, Any]) -> str: - installed_checked = ( - report.get("proof_scope", {}).get("installed_artifacts", {}).get("checked") - is True + proof_scope = report.get("proof_scope", {}) + installed_scope = proof_scope.get("installed_artifacts", {}) + installed_checked = installed_scope.get("checked") is True + installed_supplied = bool(installed_scope.get("evidence_sha256")) or ( + "installed_evidence_observation" in proof_scope ) lines = [ f"Capability fit rerun: {report['status']}", @@ -941,12 +960,29 @@ def render_review(report: dict[str, Any]) -> str: ( "Scope: schema, signed release metadata, optional local tag provenance, " + ( - "and supplied installed-composition evidence." + "and locally observed installed-composition evidence." if installed_checked - else "and no installed-composition evidence." + else ( + "and supplied installed-composition evidence that did not establish local proof." + if installed_supplied + else "and no installed-composition evidence." + ) ) ), ] + verification = proof_scope.get("boundary_evidence_verification") or proof_scope.get( + "installed_evidence_observation" + ) + if isinstance(verification, dict) and verification.get("evaluated_at"): + mode = verification.get("verification_mode") + mode_label = ( + "HISTORICAL override" + if mode == "historical_override" + else "current-time evaluation" + ) + lines.append( + f"Evidence verification: {mode_label} at {verification['evaluated_at']}." + ) findings = report.get("findings") or [] if findings: lines.append("Findings:") diff --git a/tools/assessments/govoplan_assessment/evidence.py b/tools/assessments/govoplan_assessment/evidence.py index a5b8713..3c86b82 100644 --- a/tools/assessments/govoplan_assessment/evidence.py +++ b/tools/assessments/govoplan_assessment/evidence.py @@ -4,15 +4,19 @@ from __future__ import annotations import base64 from dataclasses import dataclass -from datetime import UTC, datetime +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 @@ -27,9 +31,17 @@ 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 = { @@ -83,6 +95,7 @@ class InstalledEvidenceReview: class BoundaryEvidenceReview: findings: tuple[EvidenceFinding, ...] proof_scope: Mapping[str, Any] + review_targets: tuple[dict[str, Any], ...] = () @dataclass(slots=True) @@ -132,10 +145,17 @@ def collect_installed_composition( source = metadata.distributions() if distributions is None else distributions for distribution in source: try: - package_name = normalize_package_name(str(distribution.metadata["Name"])) + 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-"): + 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]))) @@ -162,7 +182,7 @@ def collect_installed_composition( for package_name, distribution in candidates: version = _distribution_version(distribution) - if not version or PACKAGE_PATTERN.fullmatch(package_name) is None: + if not version: issues.append( { "code": "distribution-metadata-invalid", @@ -185,6 +205,7 @@ def collect_installed_composition( "source_provenance": _source_provenance(distribution), "record_integrity": _record_integrity( distribution, + package_name=package_name, budget=hash_budget, ), } @@ -210,10 +231,14 @@ 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.1.0", + "schema_version": "0.2.0", "evidence_kind": "govoplan.installed-composition", - "assessment_id": str(assessment.get("assessment_id") or "invalid-assessment"), - "assessment_release": str(assessment_release or "invalid-release"), + "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( @@ -236,6 +261,9 @@ def review_installed_composition( 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( @@ -283,9 +311,86 @@ def review_installed_composition( ) 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) @@ -394,6 +499,7 @@ def review_installed_composition( record_expected = 0 record_verified = 0 + generated_unhashed_file_count = 0 provenance_expected = 0 provenance_anchored = 0 provenance_mutable = 0 @@ -503,7 +609,18 @@ def review_installed_composition( record = artifact.get("record_integrity") record_status = record.get("status") if isinstance(record, dict) else None - if record_status == "verified": + 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 @@ -514,8 +631,16 @@ def review_installed_composition( affected_module_ids=affected_module_ids, component=component, kind="installed_record_unverified", - code="installed_record_integrity_unverified", - message=f"Installed distribution {package_name!r} does not have complete, matching RECORD hash coverage.", + 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}, ) @@ -635,13 +760,18 @@ def review_installed_composition( exact_set = ( set(artifacts_by_package) == expected_packages and not duplicate_packages ) - installed_checked = True - record_checked = binding_valid and exact_set and record_expected > 0 - provenance_checked = binding_valid and exact_set and provenance_expected > 0 + 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, + "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), @@ -651,6 +781,7 @@ def review_installed_composition( "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, @@ -669,6 +800,15 @@ def review_installed_composition( "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), @@ -689,11 +829,64 @@ def review_boundary_evidence( 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: - scopes = _unchecked_boundary_scope() - if evidence is None: - return BoundaryEvidenceReview(findings=(), proof_scope=scopes) + 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( @@ -770,9 +963,6 @@ def review_boundary_evidence( ) ) - now = verification_time or datetime.now(tz=UTC) - if now.tzinfo is None: - now = now.replace(tzinfo=UTC) issued_at = _datetime(evidence.get("issued_at")) expires_at = _datetime(evidence.get("expires_at")) time_valid = True @@ -801,6 +991,7 @@ def review_boundary_evidence( 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: @@ -821,6 +1012,12 @@ def review_boundary_evidence( ] 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( @@ -839,10 +1036,29 @@ def review_boundary_evidence( ("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) @@ -898,12 +1114,18 @@ def review_boundary_evidence( ) 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 not time_valid: + 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 @@ -916,6 +1138,10 @@ def review_boundary_evidence( "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: @@ -927,7 +1153,24 @@ def review_boundary_evidence( ("assessment.external_proof",), ) ) - return BoundaryEvidenceReview(tuple(findings), scopes) + 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: @@ -944,11 +1187,37 @@ 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: - return str(distribution.version) + value = str(distribution.version) except Exception: return "" + if len(value) > 128 or VERSION_PATTERN.fullmatch(value) is None: + return "" + return value def _distribution_modules( @@ -957,14 +1226,27 @@ def _distribution_modules( modules: list[dict[str, str]] = [] issues: list[dict[str, str]] = [] try: - all_entry_points = sorted( - ( - entry_point - for entry_point in distribution.entry_points - if entry_point.group == "govoplan.modules" - ), - key=lambda item: (item.name, item.value), - ) + 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 [], [ { @@ -979,7 +1261,7 @@ def _distribution_modules( "package_name": package_name, } ) - for entry_point in all_entry_points[:16]: + for _, _, entry_point in all_entry_points[:16]: try: manifest = entry_point.load()() module_id = str(manifest.id) @@ -993,9 +1275,10 @@ def _distribution_modules( ) continue if ( - OPAQUE_ID_PATTERN.fullmatch(module_id) is None - or VERSION_PATTERN.fullmatch(manifest_version) is None + 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( { @@ -1015,6 +1298,8 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]: 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: @@ -1023,21 +1308,69 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]: 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") - editable = ( - isinstance(directory_info, dict) and directory_info.get("editable") is True - ) - if isinstance(vcs_info, dict): - commit = str(vcs_info.get("commit_id") or "").lower() - if GIT_OBJECT_PATTERN.fullmatch(commit): - return { - "kind": "editable-vcs" if editable else "vcs-commit", - "commit": commit, - } - return {"kind": "malformed-direct-url"} 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: @@ -1049,6 +1382,13 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]: 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"} @@ -1056,11 +1396,13 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]: 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, @@ -1082,17 +1424,47 @@ def _record_integrity( return {"status": "unavailable", **counts} if not _trusted_distribution_root(distribution_root): return {"status": "unavailable", **counts} - script_roots = ( - (installation_root / "bin").resolve(), - (installation_root / "Scripts").resolve(), - ) + 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 in files: + 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(str(package_path)): + 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 @@ -1100,58 +1472,44 @@ def _record_integrity( counts["unverifiable_file_count"] += 1 continue try: - resolved_path = Path(distribution.locate_file(package_path)).resolve() + file_stat = resolved_path.stat() except OSError: counts["missing_file_count"] += 1 continue - record_path = PurePosixPath(str(package_path).replace("\\", "/")) - has_parent_traversal = ".." in record_path.parts - path_allowed = ( - not record_path.is_absolute() - and not has_parent_traversal - and _is_relative_to(resolved_path, distribution_root) - ) or ( - not record_path.is_absolute() - and has_parent_traversal - and any(_is_relative_to(resolved_path, root) for root in script_roots) + 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 ) - if not path_allowed: - counts["unverifiable_file_count"] += 1 - continue - try: - stat = resolved_path.stat() - except OSError: - counts["missing_file_count"] += 1 - continue - if not resolved_path.is_file(): - counts["missing_file_count"] += 1 - continue - if stat.st_size > MAX_HASHED_FILE_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 - total_hashed_bytes += stat.st_size - if ( - total_hashed_bytes > MAX_HASHED_DISTRIBUTION_BYTES - or stat.st_size > budget.remaining_bytes - ): - limit_exceeded = True - budget.remaining_bytes = 0 - continue - budget.remaining_bytes -= stat.st_size expected = _record_hash_bytes(str(getattr(hash_info, "value", ""))) if expected is None: counts["unverifiable_file_count"] += 1 continue - digest = hashlib.sha256() try: - with resolved_path.open("rb") as handle: - for chunk in iter(lambda: handle.read(1024 * 1024), b""): - digest.update(chunk) + actual, hashed_bytes, exceeded = _bounded_file_sha256( + resolved_path, + max_bytes=allowed_bytes, + ) except OSError: counts["missing_file_count"] += 1 continue - counts["hashed_file_count"] += 1 - if not hmac.compare_digest(digest.digest(), expected): + 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" @@ -1166,6 +1524,96 @@ def _record_integrity( 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)) @@ -1174,19 +1622,88 @@ def _record_hash_bytes(value: str) -> bytes | None: return decoded if len(decoded) == hashlib.sha256().digest_size else None -def _permitted_unhashed_record_path(value: str) -> bool: - normalized = value.replace("\\", "/") - return normalized.endswith( - (".dist-info/RECORD", ".dist-info/RECORD.jws", ".dist-info/RECORD.p7s") +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 + *, + 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 @@ -1202,14 +1719,6 @@ def _active_authority_keys( ) continue seen.add(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 try: public_key = base64.b64decode( str(item.get("public_key") or ""), validate=True @@ -1217,6 +1726,39 @@ def _active_authority_keys( 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", [])), @@ -1282,18 +1824,22 @@ def _unchecked_installed_scope() -> dict[str, Any]: def _invalid_installed_scope(*, digest: str) -> dict[str, Any]: scope = _unchecked_installed_scope() scope["installed_artifacts"] = { - "checked": True, - "valid": False, + "checked": False, + "valid": None, + "comparison_valid": False, "evidence_sha256": digest, } return scope -def _unchecked_boundary_scope() -> dict[str, Any]: +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() @@ -1310,6 +1856,29 @@ def _datetime(value: object) -> datetime | 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: