GovOPlaN/govoplan
1
0
Fork 0
Code Issues 10 Pull Requests Actions 3 Packages Projects Releases Wiki Activity

fix(assessment): harden capability evidence trust

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-22 19:21:06 +02:00
parent 1dc9148ec3
commit 19c4b63ade
8 changed files with 1928 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files

190
tests/test_capability_fit_atomic_io.py Normal file
View File

@@ -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()

641
tests/test_capability_fit_evidence.py
View File

@@ -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