1796 lines
68 KiB
Python
1796 lines
68 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
from copy import deepcopy
|
|
from datetime import UTC, datetime
|
|
import hashlib
|
|
from importlib import metadata
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
|
|
|
|
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.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,
|
|
review_installed_composition,
|
|
validate_payload,
|
|
)
|
|
from govoplan_assessment.installer_receipt import ( # noqa: E402
|
|
issue_installer_receipt,
|
|
)
|
|
from govoplan_release.artifact_identity import inspect_python_wheel # noqa: E402
|
|
from tests.test_capability_fit_review import signed_catalog # noqa: E402
|
|
|
|
|
|
class CapabilityFitEvidenceTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.assessment = load_json("docs/capability-fit-current.json")
|
|
cls.assessment_schema = load_json("docs/capability-fit.schema.json")
|
|
cls.installed_schema = load_json(
|
|
"docs/installed-composition-evidence.schema.json"
|
|
)
|
|
cls.boundary_schema = load_json(
|
|
"docs/capability-fit-boundary-evidence.schema.json"
|
|
)
|
|
cls.authority_schema = load_json(
|
|
"docs/capability-fit-proof-authority-keyring.schema.json"
|
|
)
|
|
|
|
def test_matching_installed_composition_proves_local_consistency_not_origin(
|
|
self,
|
|
) -> None:
|
|
catalog, keyring = signed_catalog(self.assessment)
|
|
evidence = matching_installed_evidence(self.assessment)
|
|
|
|
report = self.review(
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
installed_evidence=evidence,
|
|
)
|
|
|
|
self.assertEqual("current", report["status"])
|
|
self.assertTrue(report["proof_scope"]["installed_artifacts"]["valid"])
|
|
self.assertTrue(report["proof_scope"]["installed_record_integrity"]["valid"])
|
|
self.assertTrue(report["proof_scope"]["installed_source_provenance"]["valid"])
|
|
self.assertEqual(
|
|
"local-pep610-metadata",
|
|
report["proof_scope"]["installed_source_provenance"]["basis"],
|
|
)
|
|
self.assertFalse(
|
|
report["proof_scope"]["installed_source_provenance"][
|
|
"release_origin_bound"
|
|
]
|
|
)
|
|
self.assertFalse(report["proof_scope"]["installed_release_origin"]["checked"])
|
|
self.assertIsNone(report["proof_scope"]["installed_release_origin"]["valid"])
|
|
self.assertFalse(report["proof_scope"]["runtime_activation"]["checked"])
|
|
self.assertFalse(report["proof_scope"]["target_environment"]["checked"])
|
|
self.assertEqual(
|
|
canonical_sha256(evidence),
|
|
report["proof_scope"]["installed_artifacts"]["evidence_sha256"],
|
|
)
|
|
|
|
def test_signed_catalog_artifact_identities_bind_installed_release_origin(
|
|
self,
|
|
) -> None:
|
|
evidence = matching_installed_evidence(self.assessment)
|
|
catalog, keyring = signed_catalog(
|
|
self.assessment,
|
|
release_artifacts=catalog_artifacts_for(evidence),
|
|
)
|
|
|
|
report = self.review(
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
installed_evidence=evidence,
|
|
)
|
|
|
|
origin = report["proof_scope"]["installed_release_origin"]
|
|
self.assertEqual("current", report["status"])
|
|
self.assertTrue(origin["checked"])
|
|
self.assertTrue(origin["valid"])
|
|
self.assertTrue(origin["release_origin_bound"])
|
|
self.assertEqual(len(evidence["artifacts"]), origin["anchored_artifact_digest_count"])
|
|
self.assertEqual(0, origin["signed_installer_receipt_count"])
|
|
|
|
def test_role_scoped_installer_receipt_admits_transformed_artifacts_and_boundary(
|
|
self,
|
|
) -> None:
|
|
evidence = matching_installed_evidence(self.assessment)
|
|
artifact_rows = catalog_artifacts_for(evidence, requires_receipt=True)
|
|
catalog, keyring = signed_catalog(
|
|
self.assessment,
|
|
release_artifacts=artifact_rows,
|
|
)
|
|
receipt, installer_keys = signed_installer_receipt(
|
|
assessment=self.assessment,
|
|
installed=evidence,
|
|
catalog=catalog,
|
|
catalog_artifacts=artifact_rows,
|
|
)
|
|
proof, proof_keys = signed_boundary_evidence(
|
|
assessment=self.assessment,
|
|
installed=evidence,
|
|
claims=[boundary_claim("target_environment", "passed")],
|
|
allowed_scopes=["target_environment"],
|
|
)
|
|
|
|
report = self.review(
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
installed_evidence=evidence,
|
|
installer_receipt=receipt,
|
|
installer_authority_keyring=installer_keys,
|
|
boundary_evidence=proof,
|
|
authority_keyring=proof_keys,
|
|
installed_evidence_mode="imported_unsigned",
|
|
)
|
|
|
|
origin = report["proof_scope"]["installed_release_origin"]
|
|
self.assertTrue(origin["valid"])
|
|
self.assertEqual(0, origin["anchored_artifact_digest_count"])
|
|
self.assertEqual(len(evidence["artifacts"]), origin["signed_installer_receipt_count"])
|
|
self.assertTrue(report["proof_scope"]["target_environment"]["checked"])
|
|
self.assertTrue(report["proof_scope"]["target_environment"]["valid"])
|
|
self.assertTrue(
|
|
report["proof_scope"]["installed_evidence_observation"][
|
|
"receipt_authenticated"
|
|
]
|
|
)
|
|
self.assertNotIn(
|
|
"installed_evidence_unsigned_import",
|
|
{item["code"] for item in report["findings"]},
|
|
)
|
|
|
|
def test_installer_receipt_cannot_reuse_release_key_or_cover_forged_payload(
|
|
self,
|
|
) -> None:
|
|
evidence = matching_installed_evidence(self.assessment)
|
|
artifact_rows = catalog_artifacts_for(evidence, requires_receipt=True)
|
|
catalog, keyring = signed_catalog(
|
|
self.assessment,
|
|
release_artifacts=artifact_rows,
|
|
)
|
|
receipt, installer_keys = signed_installer_receipt(
|
|
assessment=self.assessment,
|
|
installed=evidence,
|
|
catalog=catalog,
|
|
catalog_artifacts=artifact_rows,
|
|
)
|
|
forged = deepcopy(evidence)
|
|
forged["artifacts"][0]["record_integrity"]["installed_payload_identity"][
|
|
"sha256"
|
|
] = "f" * 64
|
|
reused = deepcopy(installer_keys)
|
|
reused["keys"][0]["public_key"] = keyring["keys"][0]["public_key"]
|
|
|
|
forged_report = self.review(
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
installed_evidence=forged,
|
|
installer_receipt=receipt,
|
|
installer_authority_keyring=installer_keys,
|
|
)
|
|
reused_report = self.review(
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
installed_evidence=evidence,
|
|
installer_receipt=receipt,
|
|
installer_authority_keyring=reused,
|
|
)
|
|
|
|
self.assertIn(
|
|
"installer_receipt_evidence_mismatch",
|
|
{item["code"] for item in forged_report["findings"]},
|
|
)
|
|
self.assertIn(
|
|
"installer_authority_untrusted",
|
|
{item["code"] for item in reused_report["findings"]},
|
|
)
|
|
self.assertFalse(
|
|
forged_report["proof_scope"]["installed_release_origin"]["valid"]
|
|
)
|
|
self.assertFalse(
|
|
reused_report["proof_scope"]["installed_release_origin"]["valid"]
|
|
)
|
|
|
|
def test_receipt_issuer_refuses_file_like_or_unverified_observations(self) -> None:
|
|
evidence = matching_installed_evidence(self.assessment)
|
|
artifacts = catalog_artifacts_for(evidence, requires_receipt=True)
|
|
catalog, _ = signed_catalog(
|
|
self.assessment,
|
|
release_artifacts=artifacts,
|
|
)
|
|
private_key = Ed25519PrivateKey.generate()
|
|
|
|
with self.assertRaisesRegex(ValueError, "same-process"):
|
|
issue_installer_receipt(
|
|
assessment=self.assessment,
|
|
catalog=catalog,
|
|
installed_evidence=evidence,
|
|
receipt_id="install:test",
|
|
key_id="installer:test",
|
|
private_key=private_key,
|
|
)
|
|
evidence["artifacts"][0]["record_integrity"]["status"] = "partial"
|
|
evidence["artifacts"][0]["record_integrity"]["unverifiable_file_count"] = 1
|
|
with self.assertRaisesRegex(ValueError, "not a verified match"):
|
|
issue_installer_receipt(
|
|
assessment=self.assessment,
|
|
catalog=catalog,
|
|
installed_evidence=evidence,
|
|
receipt_id="install:test",
|
|
key_id="installer:test",
|
|
private_key=private_key,
|
|
same_process_observation=True,
|
|
consumed_artifacts={
|
|
str(item["package_name"]): Path("unused.whl")
|
|
for item in evidence["artifacts"]
|
|
},
|
|
)
|
|
|
|
def test_installer_receipt_and_observation_must_be_current_together(self) -> None:
|
|
evidence = matching_installed_evidence(self.assessment)
|
|
artifact_rows = catalog_artifacts_for(evidence, requires_receipt=True)
|
|
catalog, keyring = signed_catalog(
|
|
self.assessment,
|
|
release_artifacts=artifact_rows,
|
|
)
|
|
receipt, installer_keys = signed_installer_receipt(
|
|
assessment=self.assessment,
|
|
installed=evidence,
|
|
catalog=catalog,
|
|
catalog_artifacts=artifact_rows,
|
|
)
|
|
|
|
report = self.review(
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
installed_evidence=evidence,
|
|
installer_receipt=receipt,
|
|
installer_authority_keyring=installer_keys,
|
|
installed_evidence_mode="imported_unsigned",
|
|
verification_time=datetime(2026, 7, 23, 12, 5, 1, tzinfo=UTC),
|
|
)
|
|
|
|
self.assertIn(
|
|
"installer_receipt_time_invalid",
|
|
{item["code"] for item in report["findings"]},
|
|
)
|
|
self.assertFalse(
|
|
report["proof_scope"]["installed_release_origin"]["valid"]
|
|
)
|
|
self.assertFalse(
|
|
report["proof_scope"]["installed_evidence_observation"][
|
|
"receipt_authenticated"
|
|
]
|
|
)
|
|
|
|
def test_catalog_trust_blocker_invalidates_dependent_evidence_scopes(
|
|
self,
|
|
) -> None:
|
|
catalog, keyring = signed_catalog(self.assessment)
|
|
catalog["signatures"][0]["value"] = "AAAA"
|
|
installed = matching_installed_evidence(self.assessment)
|
|
proof, authority = signed_boundary_evidence(
|
|
assessment=self.assessment,
|
|
installed=installed,
|
|
claims=[boundary_claim("target_environment", "passed")],
|
|
allowed_scopes=["target_environment"],
|
|
)
|
|
|
|
report = self.review(
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
installed_evidence=installed,
|
|
boundary_evidence=proof,
|
|
authority_keyring=authority,
|
|
)
|
|
|
|
self.assertEqual("blocked", report["status"])
|
|
self.assertFalse(
|
|
report["proof_scope"]["catalog_signature_and_keyring"]["valid"]
|
|
)
|
|
self.assertFalse(report["proof_scope"]["installed_artifacts"]["checked"])
|
|
self.assertIsNone(report["proof_scope"]["installed_artifacts"]["valid"])
|
|
self.assertFalse(report["proof_scope"]["target_environment"]["checked"])
|
|
self.assertIsNone(report["proof_scope"]["target_environment"]["valid"])
|
|
self.assertIn(
|
|
"installed_catalog_trust_unavailable",
|
|
{item["code"] for item in report["findings"]},
|
|
)
|
|
|
|
def test_missing_extra_and_version_drift_have_deterministic_review_targets(
|
|
self,
|
|
) -> None:
|
|
catalog, keyring = signed_catalog(self.assessment)
|
|
evidence = matching_installed_evidence(self.assessment)
|
|
evidence["artifacts"] = [
|
|
item
|
|
for item in evidence["artifacts"]
|
|
if item["package_name"] != "govoplan-campaign"
|
|
]
|
|
access = next(
|
|
item
|
|
for item in evidence["artifacts"]
|
|
if item["package_name"] == "govoplan-access"
|
|
)
|
|
access["package_version"] = "9.9.9"
|
|
evidence["artifacts"].append(
|
|
artifact(
|
|
package_name="govoplan-unassessed",
|
|
version="1.0.0",
|
|
module_id="unassessed",
|
|
commit="f" * 40,
|
|
)
|
|
)
|
|
|
|
first = self.review(
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
installed_evidence=evidence,
|
|
)
|
|
second = self.review(
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
installed_evidence=deepcopy(evidence),
|
|
)
|
|
|
|
self.assertEqual("review_required", first["status"])
|
|
self.assertEqual(
|
|
json.dumps(first, sort_keys=True), json.dumps(second, sort_keys=True)
|
|
)
|
|
codes = {item["code"] for item in first["findings"]}
|
|
self.assertIn("installed_distribution_missing", codes)
|
|
self.assertIn("installed_distribution_extra", codes)
|
|
self.assertIn("installed_distribution_version_mismatch", codes)
|
|
targets = {item["id"] for item in first["review_targets"]}
|
|
self.assertIn("composition.campaigns", targets)
|
|
self.assertIn("composition.access", targets)
|
|
self.assertIn("assessment.installed_composition", targets)
|
|
self.assertTrue(first["proof_scope"]["release_metadata"]["valid"])
|
|
self.assertFalse(first["proof_scope"]["installed_artifacts"]["valid"])
|
|
|
|
def test_editable_source_and_partial_record_are_not_immutable_proof(self) -> None:
|
|
catalog, keyring = signed_catalog(self.assessment)
|
|
evidence = matching_installed_evidence(self.assessment)
|
|
campaign = next(
|
|
item
|
|
for item in evidence["artifacts"]
|
|
if item["package_name"] == "govoplan-campaign"
|
|
)
|
|
campaign["source_provenance"] = {
|
|
"basis": "local-pep610-metadata",
|
|
"kind": "editable-local",
|
|
}
|
|
campaign["record_integrity"] = {
|
|
**campaign["record_integrity"],
|
|
"status": "partial",
|
|
"unverifiable_file_count": 1,
|
|
}
|
|
|
|
report = self.review(
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
installed_evidence=evidence,
|
|
)
|
|
|
|
codes = {item["code"] for item in report["findings"]}
|
|
self.assertIn("installed_source_mutable", codes)
|
|
self.assertIn("installed_record_integrity_unverified", codes)
|
|
self.assertFalse(report["proof_scope"]["installed_source_provenance"]["valid"])
|
|
self.assertFalse(report["proof_scope"]["installed_record_integrity"]["valid"])
|
|
|
|
def test_malformed_or_wrongly_bound_installed_evidence_fails_closed(self) -> None:
|
|
catalog, keyring = signed_catalog(self.assessment)
|
|
evidence = matching_installed_evidence(self.assessment)
|
|
evidence["assessment_id"] = "another-assessment"
|
|
evidence["artifacts"].append(deepcopy(evidence["artifacts"][0]))
|
|
|
|
report = self.review(
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
installed_evidence=evidence,
|
|
)
|
|
|
|
self.assertEqual("blocked", report["status"])
|
|
codes = {item["code"] for item in report["findings"]}
|
|
self.assertIn("installed_evidence_assessment_mismatch", codes)
|
|
self.assertIn("installed_distribution_duplicate", codes)
|
|
self.assertFalse(report["proof_scope"]["installed_artifacts"]["valid"])
|
|
self.assertIn(
|
|
"composition.core", {item["id"] for item in report["review_targets"]}
|
|
)
|
|
self.assertEqual(
|
|
0,
|
|
report["proof_scope"]["installed_source_provenance"][
|
|
"mutable_distribution_count"
|
|
],
|
|
)
|
|
|
|
def test_schema_failure_does_not_echo_untrusted_evidence_values(self) -> None:
|
|
catalog, keyring = signed_catalog(self.assessment)
|
|
evidence = matching_installed_evidence(self.assessment)
|
|
evidence["artifacts"][0]["package_name"] = "secret-user@private-host/path"
|
|
|
|
report = self.review(
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
installed_evidence=evidence,
|
|
)
|
|
|
|
encoded = json.dumps(report, sort_keys=True)
|
|
self.assertEqual("blocked", report["status"])
|
|
self.assertIn("installed_evidence_schema", encoded)
|
|
self.assertNotIn("secret-user", encoded)
|
|
self.assertNotIn("private-host", encoded)
|
|
|
|
def test_missing_catalog_package_mapping_cannot_pass_vacuously(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),
|
|
catalog_trusted=True,
|
|
)
|
|
|
|
codes = {item.code for item in result.findings}
|
|
self.assertIn("installed_expected_catalog_entry_missing", codes)
|
|
self.assertIn("installed_expected_composition_empty", codes)
|
|
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)
|
|
distribution, payload_file = create_distribution(root)
|
|
|
|
with mock.patch.object(sys, "path", [str(root), *sys.path]):
|
|
evidence = collect_installed_composition(
|
|
assessment=self.assessment,
|
|
collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
|
|
distributions=[distribution],
|
|
)
|
|
|
|
self.assertEqual(
|
|
(), validate_payload(payload=evidence, schema=self.installed_schema)
|
|
)
|
|
observed = evidence["artifacts"][0]
|
|
self.assertEqual("editable-local", observed["source_provenance"]["kind"])
|
|
self.assertEqual(
|
|
"local-pep610-metadata",
|
|
observed["source_provenance"]["basis"],
|
|
)
|
|
self.assertEqual("verified", observed["record_integrity"]["status"])
|
|
encoded = json.dumps(evidence, sort_keys=True)
|
|
self.assertNotIn(str(root), encoded)
|
|
self.assertNotIn("file://", encoded)
|
|
|
|
payload_file.write_text("tampered\n", encoding="utf-8")
|
|
with mock.patch.object(sys, "path", [str(root), *sys.path]):
|
|
tampered = collect_installed_composition(
|
|
assessment=self.assessment,
|
|
collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
|
|
distributions=[distribution],
|
|
)
|
|
|
|
self.assertEqual(
|
|
"mismatch", tampered["artifacts"][0]["record_integrity"]["status"]
|
|
)
|
|
self.assertEqual(
|
|
1,
|
|
tampered["artifacts"][0]["record_integrity"]["mismatched_file_count"],
|
|
)
|
|
|
|
def test_collector_detects_payload_declared_in_record_but_missing(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir)
|
|
distribution, payload_file = create_distribution(root)
|
|
payload_file.unlink()
|
|
|
|
with mock.patch.object(sys, "path", [str(root), *sys.path]):
|
|
evidence = collect_installed_composition(
|
|
assessment=self.assessment,
|
|
collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
|
|
distributions=[distribution],
|
|
)
|
|
|
|
integrity = evidence["artifacts"][0]["record_integrity"]
|
|
self.assertEqual("mismatch", integrity["status"])
|
|
self.assertEqual(1, integrity["missing_file_count"])
|
|
|
|
def test_collector_rejects_duplicate_record_declarations(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir)
|
|
distribution, _ = create_distribution(root)
|
|
record_path = root / "govoplan_demo-1.0.0.dist-info" / "RECORD"
|
|
rows = record_path.read_text(encoding="utf-8").splitlines()
|
|
record_path.write_text(
|
|
"\n".join((*rows, rows[0])) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with mock.patch.object(sys, "path", [str(root), *sys.path]):
|
|
evidence = collect_installed_composition(
|
|
assessment=self.assessment,
|
|
collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
|
|
distributions=[distribution],
|
|
)
|
|
|
|
integrity = evidence["artifacts"][0]["record_integrity"]
|
|
self.assertEqual("partial", integrity["status"])
|
|
self.assertGreater(integrity["unverifiable_file_count"], 0)
|
|
|
|
def test_collector_rejects_malformed_record_declarations(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir)
|
|
distribution, _ = create_distribution(root)
|
|
record_path = root / "govoplan_demo-1.0.0.dist-info" / "RECORD"
|
|
record_path.write_text(
|
|
record_path.read_text(encoding="utf-8") + "malformed,row\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with mock.patch.object(sys, "path", [str(root), *sys.path]):
|
|
evidence = collect_installed_composition(
|
|
assessment=self.assessment,
|
|
collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
|
|
distributions=[distribution],
|
|
)
|
|
|
|
integrity = evidence["artifacts"][0]["record_integrity"]
|
|
self.assertEqual("partial", integrity["status"])
|
|
self.assertGreater(integrity["unverifiable_file_count"], 0)
|
|
|
|
def test_collector_verifies_declared_record_sizes(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir)
|
|
distribution, _ = create_distribution(root)
|
|
record_path = root / "govoplan_demo-1.0.0.dist-info" / "RECORD"
|
|
rows = record_path.read_text(encoding="utf-8").splitlines()
|
|
path, digest, size = rows[0].split(",")
|
|
rows[0] = f"{path},{digest},{int(size) + 1}"
|
|
record_path.write_text("\n".join(rows) + "\n", encoding="utf-8")
|
|
|
|
with mock.patch.object(sys, "path", [str(root), *sys.path]):
|
|
evidence = collect_installed_composition(
|
|
assessment=self.assessment,
|
|
collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
|
|
distributions=[distribution],
|
|
)
|
|
|
|
integrity = evidence["artifacts"][0]["record_integrity"]
|
|
self.assertEqual("mismatch", integrity["status"])
|
|
self.assertEqual(1, integrity["mismatched_file_count"])
|
|
|
|
def test_collector_rejects_symlinked_distribution_metadata_root(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir)
|
|
create_distribution(root)
|
|
metadata_root = root / "govoplan_demo-1.0.0.dist-info"
|
|
real_metadata_root = root / "real-govoplan-demo.dist-info"
|
|
metadata_root.rename(real_metadata_root)
|
|
metadata_root.symlink_to(real_metadata_root, target_is_directory=True)
|
|
distribution = metadata.PathDistribution(metadata_root)
|
|
|
|
with mock.patch.object(sys, "path", [str(root), *sys.path]):
|
|
evidence = collect_installed_composition(
|
|
assessment=self.assessment,
|
|
collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
|
|
distributions=[distribution],
|
|
)
|
|
|
|
integrity = evidence["artifacts"][0]["record_integrity"]
|
|
self.assertEqual("unavailable", integrity["status"])
|
|
self.assertEqual(0, integrity["hashed_file_count"])
|
|
|
|
def test_collector_accepts_real_core_wheel_payload_scripts_data_and_pyc(
|
|
self,
|
|
) -> None:
|
|
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"))
|
|
built_identity = inspect_python_wheel(wheel)
|
|
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(
|
|
built_identity.payload_sha256,
|
|
record["artifact_payload_identity"]["sha256"],
|
|
)
|
|
self.assertEqual(
|
|
built_identity.payload_file_count,
|
|
record["artifact_payload_identity"]["file_count"],
|
|
)
|
|
self.assertTrue(built_identity.requires_installer_receipt)
|
|
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)]
|
|
)
|
|
|
|
evidence = collect_installed_composition(
|
|
assessment=self.assessment,
|
|
collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
|
|
distributions=[distribution],
|
|
)
|
|
|
|
self.assertEqual(16, len(evidence["artifacts"][0]["modules"]))
|
|
self.assertIn(
|
|
{
|
|
"code": "module-entry-point-limit-exceeded",
|
|
"package_name": "govoplan-many",
|
|
},
|
|
evidence["collection_issues"],
|
|
)
|
|
|
|
def test_tied_distribution_rows_are_canonically_ordered(self) -> None:
|
|
first_distribution = FakeDistribution(
|
|
entry_points=[FakeEntryPoint(name="module-z")],
|
|
direct_url={"dir_info": {"editable": True}, "url": "file:///redacted-a"},
|
|
)
|
|
second_distribution = FakeDistribution(
|
|
entry_points=[FakeEntryPoint(name="module-a")],
|
|
direct_url={"url": "https://redacted.invalid/archive"},
|
|
)
|
|
collected_at = datetime(2026, 7, 22, 12, tzinfo=UTC)
|
|
|
|
forward = collect_installed_composition(
|
|
assessment=self.assessment,
|
|
collected_at=collected_at,
|
|
distributions=[first_distribution, second_distribution],
|
|
)
|
|
reversed_order = collect_installed_composition(
|
|
assessment=self.assessment,
|
|
collected_at=collected_at,
|
|
distributions=[second_distribution, first_distribution],
|
|
)
|
|
|
|
self.assertEqual(
|
|
json.dumps(forward, sort_keys=True),
|
|
json.dumps(reversed_order, sort_keys=True),
|
|
)
|
|
|
|
def test_signed_boundary_claim_waits_for_anchored_installed_release_origin(
|
|
self,
|
|
) -> None:
|
|
catalog, keyring = signed_catalog(self.assessment)
|
|
installed = matching_installed_evidence(self.assessment)
|
|
proof, authority = signed_boundary_evidence(
|
|
assessment=self.assessment,
|
|
installed=installed,
|
|
claims=[boundary_claim("target_environment", "passed")],
|
|
allowed_scopes=["target_environment"],
|
|
)
|
|
|
|
report = self.review(
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
installed_evidence=installed,
|
|
boundary_evidence=proof,
|
|
authority_keyring=authority,
|
|
)
|
|
|
|
self.assertEqual("blocked", report["status"])
|
|
self.assertFalse(report["proof_scope"]["target_environment"]["checked"])
|
|
self.assertIsNone(report["proof_scope"]["target_environment"]["valid"])
|
|
self.assertFalse(report["proof_scope"]["external_providers"]["checked"])
|
|
self.assertFalse(report["proof_scope"]["production_approval"]["checked"])
|
|
self.assertIn(
|
|
"boundary_installed_release_origin_untrusted",
|
|
{item["code"] for item in report["findings"]},
|
|
)
|
|
self.assertEqual(
|
|
self.assessment["deployment_profile"]["id"],
|
|
report["proof_scope"]["target_environment"]["expected_subject_id"],
|
|
)
|
|
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("blocked", accepted["status"])
|
|
provider_scope = accepted["proof_scope"]["external_providers"]
|
|
self.assertFalse(provider_scope["checked"])
|
|
self.assertEqual("provider:test", provider_scope["expected_subject_id"])
|
|
self.assertEqual("provider:test", provider_scope["observed_subject_id"])
|
|
self.assertIn(
|
|
"boundary_installed_release_origin_untrusted",
|
|
{item["code"] for item in accepted["findings"]},
|
|
)
|
|
|
|
def test_boundary_subject_mismatch_blocks_without_echoing_invalid_config(
|
|
self,
|
|
) -> 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_waits_for_installed_origin_binding(
|
|
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,
|
|
)
|
|
|
|
self.assertEqual("blocked", report["status"])
|
|
self.assertFalse(report["proof_scope"]["target_environment"]["checked"])
|
|
self.assertFalse(report["proof_scope"]["production_approval"]["checked"])
|
|
self.assertNotIn(
|
|
"boundary.target_environment.production-like-dev",
|
|
{item["id"] for item in report["review_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_malformed_authority_members_invalidate_the_whole_keyring(self) -> None:
|
|
catalog, keyring = signed_catalog(self.assessment)
|
|
installed = matching_installed_evidence(self.assessment)
|
|
proof, authority = signed_boundary_evidence(
|
|
assessment=self.assessment,
|
|
installed=installed,
|
|
claims=[boundary_claim("target_environment", "passed")],
|
|
allowed_scopes=["target_environment"],
|
|
)
|
|
malformed = deepcopy(authority)
|
|
malformed["keys"].append(
|
|
{
|
|
"key_id": "authority:malformed",
|
|
"status": "active",
|
|
"public_key": "not-base64",
|
|
"allowed_scopes": ["target_environment"],
|
|
}
|
|
)
|
|
inverted = deepcopy(authority)
|
|
inverted["keys"][0]["not_before"] = "2026-08-01T00:00:00Z"
|
|
inverted["keys"][0]["not_after"] = "2026-07-01T00:00:00Z"
|
|
|
|
cases = (
|
|
(malformed, "boundary_authority_key_invalid"),
|
|
(inverted, "boundary_authority_key_interval_invalid"),
|
|
)
|
|
for supplied_keyring, expected_code in cases:
|
|
with self.subTest(expected_code=expected_code):
|
|
report = self.review(
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
installed_evidence=installed,
|
|
boundary_evidence=proof,
|
|
authority_keyring=supplied_keyring,
|
|
)
|
|
|
|
self.assertEqual("blocked", report["status"])
|
|
self.assertIn(
|
|
expected_code,
|
|
{item["code"] for item in report["findings"]},
|
|
)
|
|
self.assertIn(
|
|
"boundary_authority_untrusted",
|
|
{item["code"] for item in report["findings"]},
|
|
)
|
|
self.assertFalse(
|
|
report["proof_scope"]["target_environment"]["checked"]
|
|
)
|
|
|
|
def test_historical_verification_and_generated_runtime_gap_are_visible(
|
|
self,
|
|
) -> None:
|
|
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",
|
|
catalog_trusted=True,
|
|
)
|
|
|
|
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)
|
|
installed = matching_installed_evidence(self.assessment)
|
|
proof, authority = signed_boundary_evidence(
|
|
assessment=self.assessment,
|
|
installed=installed,
|
|
claims=[
|
|
boundary_claim("target_environment", "passed"),
|
|
boundary_claim("production_approval", "approved"),
|
|
],
|
|
allowed_scopes=["target_environment"],
|
|
)
|
|
|
|
report = self.review(
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
installed_evidence=installed,
|
|
boundary_evidence=proof,
|
|
authority_keyring=authority,
|
|
)
|
|
|
|
self.assertEqual("blocked", report["status"])
|
|
self.assertIn(
|
|
"boundary_authority_untrusted",
|
|
{item["code"] for item in report["findings"]},
|
|
)
|
|
self.assertFalse(report["proof_scope"]["target_environment"]["checked"])
|
|
self.assertFalse(report["proof_scope"]["production_approval"]["checked"])
|
|
|
|
def test_boundary_semantics_interval_and_duplicates_fail_closed(self) -> None:
|
|
catalog, keyring = signed_catalog(self.assessment)
|
|
installed = matching_installed_evidence(self.assessment)
|
|
cases = []
|
|
semantic = [boundary_claim("production_approval", "passed")]
|
|
cases.append(("semantics", semantic, None))
|
|
duplicate = [
|
|
boundary_claim("target_environment", "passed"),
|
|
boundary_claim("target_environment", "failed"),
|
|
]
|
|
cases.append(("duplicate", duplicate, None))
|
|
cases.append(
|
|
(
|
|
"interval",
|
|
[boundary_claim("target_environment", "passed")],
|
|
("2026-07-24T00:00:00Z", "2026-07-23T00:00:00Z"),
|
|
)
|
|
)
|
|
for label, claims, interval in cases:
|
|
with self.subTest(label=label):
|
|
proof, authority = signed_boundary_evidence(
|
|
assessment=self.assessment,
|
|
installed=installed,
|
|
claims=claims,
|
|
allowed_scopes=[
|
|
"target_environment",
|
|
"production_approval",
|
|
],
|
|
interval=interval,
|
|
)
|
|
report = self.review(
|
|
catalog=catalog,
|
|
keyring=keyring,
|
|
installed_evidence=installed,
|
|
boundary_evidence=proof,
|
|
authority_keyring=authority,
|
|
)
|
|
self.assertEqual("blocked", report["status"])
|
|
self.assertFalse(
|
|
report["proof_scope"]["production_approval"]["checked"]
|
|
)
|
|
self.assertFalse(report["proof_scope"]["target_environment"]["checked"])
|
|
|
|
def review(
|
|
self,
|
|
*,
|
|
catalog: dict[str, object],
|
|
keyring: dict[str, object],
|
|
installed_evidence: dict[str, object] | None = None,
|
|
installer_receipt: dict[str, object] | None = None,
|
|
installer_authority_keyring: 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),
|
|
schema=self.assessment_schema,
|
|
catalog=catalog,
|
|
published_keyring=keyring,
|
|
trusted_keyring=keyring,
|
|
installed_evidence=installed_evidence,
|
|
installed_evidence_schema=(installed_schema or self.installed_schema)
|
|
if installed_evidence is not None
|
|
else None,
|
|
installer_receipt=installer_receipt,
|
|
installer_receipt_schema=load_json("docs/installer-receipt.schema.json")
|
|
if installer_receipt is not None
|
|
else None,
|
|
installer_authority_keyring=installer_authority_keyring,
|
|
installer_authority_keyring_schema=load_json(
|
|
"docs/installer-receipt-authority-keyring.schema.json"
|
|
)
|
|
if installer_receipt is not None
|
|
else None,
|
|
boundary_evidence=boundary_evidence,
|
|
boundary_evidence_schema=self.boundary_schema
|
|
if boundary_evidence is not None
|
|
else None,
|
|
boundary_authority_keyring=authority_keyring,
|
|
boundary_authority_keyring_schema=self.authority_schema
|
|
if boundary_evidence is not None
|
|
else None,
|
|
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.4.0",
|
|
"evidence_kind": "govoplan.installed-composition",
|
|
"assessment_id": assessment["assessment_id"],
|
|
"assessment_release": assessment["release"]["ref"],
|
|
"collected_at": "2026-07-23T11:58:00Z",
|
|
"scope": "current-python-environment.govoplan-distributions",
|
|
"artifacts": [
|
|
artifact(
|
|
package_name=str(component["repository"]),
|
|
version=str(component["manifest_version"]),
|
|
module_id=str(component["module_id"]),
|
|
commit=(str(component["commit"]) + "0" * 64)[:40],
|
|
)
|
|
for component in assessment["composition"]
|
|
if component["enabled"] is True
|
|
],
|
|
"collection_issues": [],
|
|
}
|
|
|
|
|
|
def artifact(
|
|
*, package_name: str, version: str, module_id: str, commit: str
|
|
) -> dict[str, object]:
|
|
return {
|
|
"package_name": package_name,
|
|
"package_version": version,
|
|
"modules": []
|
|
if module_id == "core"
|
|
else [{"module_id": module_id, "manifest_version": version}],
|
|
"source_provenance": {
|
|
"basis": "local-pep610-metadata",
|
|
"kind": "vcs-commit",
|
|
"commit": commit,
|
|
},
|
|
"record_integrity": {
|
|
"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,
|
|
"artifact_payload_identity": {
|
|
"algorithm": "govoplan-wheel-declared-payload-v1",
|
|
"sha256": hashlib.sha256(
|
|
f"artifact:{package_name}:{version}".encode()
|
|
).hexdigest(),
|
|
"file_count": 1,
|
|
},
|
|
"installed_payload_identity": {
|
|
"algorithm": "govoplan-installed-record-payload-v1",
|
|
"sha256": hashlib.sha256(
|
|
f"installed:{package_name}:{version}".encode()
|
|
).hexdigest(),
|
|
"file_count": 1,
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
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": subject_id or default_subject,
|
|
"control_ids": [f"control:{scope}"],
|
|
"artifacts": [
|
|
{
|
|
"artifact_id": f"result:{scope}",
|
|
"sha256": hashlib.sha256(scope.encode()).hexdigest(),
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def signed_boundary_evidence(
|
|
*,
|
|
assessment: dict[str, object],
|
|
installed: dict[str, object],
|
|
claims: list[dict[str, object]],
|
|
allowed_scopes: list[str],
|
|
interval: tuple[str, str] | None = None,
|
|
) -> tuple[dict[str, object], dict[str, object]]:
|
|
private_key = Ed25519PrivateKey.generate()
|
|
public_key = base64.b64encode(
|
|
private_key.public_key().public_bytes(
|
|
encoding=serialization.Encoding.Raw,
|
|
format=serialization.PublicFormat.Raw,
|
|
)
|
|
).decode("ascii")
|
|
issued_at, expires_at = interval or (
|
|
"2026-07-22T00:00:00Z",
|
|
"2026-07-24T00:00:00Z",
|
|
)
|
|
proof: dict[str, object] = {
|
|
"$schema": "./capability-fit-boundary-evidence.schema.json",
|
|
"schema_version": "0.1.0",
|
|
"evidence_kind": "govoplan.capability-fit-boundary-proof",
|
|
"proof_id": "proof:test",
|
|
"assessment_id": assessment["assessment_id"],
|
|
"assessment_release": assessment["release"]["ref"],
|
|
"installed_evidence_sha256": canonical_sha256(installed),
|
|
"issued_at": issued_at,
|
|
"expires_at": expires_at,
|
|
"claims": claims,
|
|
}
|
|
proof["signatures"] = [
|
|
{
|
|
"algorithm": "ed25519",
|
|
"key_id": "authority:test",
|
|
"value": base64.b64encode(private_key.sign(canonical_bytes(proof))).decode(
|
|
"ascii"
|
|
),
|
|
}
|
|
]
|
|
keyring = {
|
|
"$schema": "./capability-fit-proof-authority-keyring.schema.json",
|
|
"schema_version": "0.1.0",
|
|
"purpose": "govoplan.capability-fit-proof-authorities",
|
|
"keys": [
|
|
{
|
|
"key_id": "authority:test",
|
|
"status": "active",
|
|
"public_key": public_key,
|
|
"allowed_scopes": allowed_scopes,
|
|
"not_before": "2026-07-01T00:00:00Z",
|
|
"not_after": "2026-08-01T00:00:00Z",
|
|
}
|
|
],
|
|
}
|
|
return proof, keyring
|
|
|
|
|
|
def catalog_artifacts_for(
|
|
installed: dict[str, object], *, requires_receipt: bool = False
|
|
) -> list[dict[str, object]]:
|
|
rows: list[dict[str, object]] = []
|
|
for item in installed["artifacts"]:
|
|
record = item["record_integrity"]
|
|
rows.append(
|
|
{
|
|
"artifact_kind": "python-wheel",
|
|
"package_name": item["package_name"],
|
|
"package_version": item["package_version"],
|
|
"archive_sha256": hashlib.sha256(
|
|
f"wheel:{item['package_name']}:{item['package_version']}".encode()
|
|
).hexdigest(),
|
|
"archive_size": 1024,
|
|
"installed_payload": deepcopy(record["artifact_payload_identity"]),
|
|
"requires_installer_receipt": requires_receipt,
|
|
}
|
|
)
|
|
return rows
|
|
|
|
|
|
def signed_installer_receipt(
|
|
*,
|
|
assessment: dict[str, object],
|
|
installed: dict[str, object],
|
|
catalog: dict[str, object],
|
|
catalog_artifacts: list[dict[str, object]],
|
|
) -> tuple[dict[str, object], dict[str, object]]:
|
|
private_key = Ed25519PrivateKey.generate()
|
|
public_key = base64.b64encode(
|
|
private_key.public_key().public_bytes(
|
|
encoding=serialization.Encoding.Raw,
|
|
format=serialization.PublicFormat.Raw,
|
|
)
|
|
).decode("ascii")
|
|
catalog_by_package = {
|
|
str(item["package_name"]): item for item in catalog_artifacts
|
|
}
|
|
receipt: dict[str, object] = {
|
|
"$schema": "./installer-receipt.schema.json",
|
|
"schema_version": "0.1.0",
|
|
"evidence_kind": "govoplan.installer-receipt",
|
|
"receipt_id": "install:test",
|
|
"assessment_id": assessment["assessment_id"],
|
|
"assessment_release": assessment["release"]["ref"],
|
|
"installed_evidence_sha256": canonical_sha256(installed),
|
|
"catalog": {
|
|
"channel": catalog["channel"],
|
|
"sequence": catalog["sequence"],
|
|
"sha256": hashlib.sha256(
|
|
json.dumps(
|
|
catalog,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
ensure_ascii=True,
|
|
).encode("utf-8")
|
|
).hexdigest(),
|
|
},
|
|
"issued_at": "2026-07-23T11:59:00Z",
|
|
"artifacts": [
|
|
{
|
|
"package_name": item["package_name"],
|
|
"package_version": item["package_version"],
|
|
"catalog_archive_sha256": catalog_by_package[
|
|
str(item["package_name"])
|
|
]["archive_sha256"],
|
|
"installed_payload": deepcopy(
|
|
item["record_integrity"]["installed_payload_identity"]
|
|
),
|
|
}
|
|
for item in installed["artifacts"]
|
|
],
|
|
}
|
|
receipt["signatures"] = [
|
|
{
|
|
"algorithm": "ed25519",
|
|
"key_id": "installer:test",
|
|
"value": base64.b64encode(
|
|
private_key.sign(canonical_bytes(receipt))
|
|
).decode("ascii"),
|
|
}
|
|
]
|
|
keyring = {
|
|
"$schema": "./installer-receipt-authority-keyring.schema.json",
|
|
"schema_version": "0.1.0",
|
|
"purpose": "govoplan.installer-receipt-authorities",
|
|
"keys": [
|
|
{
|
|
"key_id": "installer:test",
|
|
"status": "active",
|
|
"public_key": public_key,
|
|
"allowed_scopes": ["installed_release_origin"],
|
|
"not_before": "2026-07-01T00:00:00Z",
|
|
"not_after": "2026-08-01T00:00:00Z",
|
|
}
|
|
],
|
|
}
|
|
return receipt, keyring
|
|
|
|
|
|
def create_distribution(
|
|
root: Path,
|
|
) -> tuple[metadata.PathDistribution, Path]:
|
|
payload_file = root / "govoplan_demo.py"
|
|
payload_file.write_text("value = 1\n", encoding="utf-8")
|
|
dist_info = root / "govoplan_demo-1.0.0.dist-info"
|
|
dist_info.mkdir()
|
|
metadata_file = dist_info / "METADATA"
|
|
metadata_file.write_text(
|
|
"Metadata-Version: 2.1\nName: govoplan-demo\nVersion: 1.0.0\n",
|
|
encoding="utf-8",
|
|
)
|
|
direct_url_file = dist_info / "direct_url.json"
|
|
direct_url_file.write_text(
|
|
json.dumps(
|
|
{
|
|
"url": f"file://{root}/private-user/govoplan-demo",
|
|
"dir_info": {"editable": True},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
record_file = dist_info / "RECORD"
|
|
rows = []
|
|
for path in (payload_file, metadata_file, direct_url_file):
|
|
relative = path.relative_to(root).as_posix()
|
|
content = path.read_bytes()
|
|
digest = (
|
|
base64.urlsafe_b64encode(hashlib.sha256(content).digest())
|
|
.decode("ascii")
|
|
.rstrip("=")
|
|
)
|
|
rows.append(f"{relative},sha256={digest},{len(content)}")
|
|
rows.append(f"{record_file.relative_to(root).as_posix()},,")
|
|
record_file.write_text("\n".join(rows) + "\n", encoding="utf-8")
|
|
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,
|
|
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.manifest_id,
|
|
version=self.manifest_version,
|
|
)
|
|
|
|
|
|
class FakeDistribution:
|
|
files = ()
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
entry_points: list[FakeEntryPoint],
|
|
direct_url: dict[str, object] | None = None,
|
|
name: str = "govoplan-many",
|
|
version: str = "1.0.0",
|
|
) -> None:
|
|
self.metadata = {"Name": name}
|
|
self.version = version
|
|
self.entry_points = entry_points
|
|
self.direct_url = direct_url
|
|
|
|
def read_text(self, filename: str):
|
|
return (
|
|
json.dumps(self.direct_url)
|
|
if filename == "direct_url.json" and self.direct_url
|
|
else None
|
|
)
|
|
|
|
|
|
def load_json(relative_path: str) -> dict[str, object]:
|
|
return json.loads((META_ROOT / relative_path).read_text(encoding="utf-8"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|