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

fix(assessment): keep release origin unproven
Some checks failed
Dependency Audit / dependency-audit (push) Failing after 15s
Details
Security Audit / security-audit (push) Failing after 14s
Details

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-22 19:55:35 +02:00
parent 4c16069f88
commit ddee5c00dc
6 changed files with 689 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
docs/CAPABILITY_AND_INFRASTRUCTURE_FIT.md
View File

@@ -389,7 +389,7 @@ that observation to the assessment and signed catalog:
--output /var/tmp/govoplan-fit-review.json
```
The collector follows the strict version `0.2.0`
The collector follows the strict version `0.3.0`
[`installed-composition-evidence.schema.json`](installed-composition-evidence.schema.json)
contract. It enumerates all installed distributions whose normalized name starts
with `govoplan-`, compares the enabled assessed package and module-manifest
@@ -399,37 +399,47 @@ entries and evidence-backed conclusions.
For each distribution, the collector also:
- verifies every supported SHA-256 immutable wheel-payload entry in installed
`RECORD` metadata,
within fixed limits of 256 GovOPlaN distributions, 10,000 files and 512 MiB
hashed per distribution, 50,000 files and 2 GiB hashed for one collection,
and 64 MiB for any single file;
- reads the installed wheel `RECORD` file directly, with a 4 MiB metadata bound,
rather than relying on an interpreted distribution file list; missing files,
duplicate or malformed declarations, mismatched declared sizes and hashes all
prevent a `verified` result;
- verifies every supported SHA-256 wheel-payload declaration in that local
metadata, within fixed limits of 256 GovOPlaN distributions, 10,000 files and
512 MiB hashed per distribution, 50,000 files and 2 GiB hashed for one
collection, and 64 MiB for any single file;
- accepts package-owned data-file targets below the installation prefix and
declared `console_scripts`, while rejecting other traversal, undeclared
scripts, paths outside the prefix and leaf symlinks before opening a file;
- reports pip-generated, unhashed `__pycache__/*.pyc` entries separately only
when their derived source is a hashed payload entry, including generated
files below a confined package-owned runtime-data prefix;
- distinguishes coherent PEP 610 Git commits and archive hashes from editable installs,
local directories, package-index/unknown origins and malformed metadata;
- labels every PEP 610 observation with the explicit
`local-pep610-metadata` basis and distinguishes coherent declared Git commits
and archive hashes from editable installs, local directories,
package-index/unknown origins and malformed metadata;
- compares a non-editable VCS commit with the assessment commit and, when one is
present, the signed catalog `selected_units` commit;
- records only package/module identifiers, versions, bounded counters, hashes
and stable error codes. It never writes direct URLs, paths, hostnames,
usernames, exception text or file contents.
`RECORD` `verified` means complete matching coverage of the immutable wheel
payload, not every runtime byte. The proof scope exposes the aggregate count of
generated unhashed bytecode; runtime activation remains unchecked. `RECORD`
agreement proves that payload files agree with installed metadata, but does not
make self-consistent metadata a trusted release origin. Similarly, a version
string is not artifact integrity. PEP 610 VCS evidence is accepted only for
`RECORD` `verified` means complete matching coverage of the wheel payload
declared by the local metadata, not every runtime byte and not a release-origin
binding. The proof scope exposes the aggregate count of generated unhashed
bytecode; runtime activation remains unchecked. `RECORD` agreement proves that
payload files agree with installed metadata, but does not make self-consistent
metadata a trusted release origin. Similarly, a version string is not artifact
integrity. PEP 610 VCS metadata is accepted only for
`vcs: git`, a full 40–64 hexadecimal commit, one mutually exclusive provenance
form and a coherent bounded URL shape. Editable and directory-backed installs
remain mutable even when their small editable-install `RECORD` is valid. Archive
or index artifacts without a hash anchored by the assessed release remain
unanchored. These conditions cause review rather than being promoted to
immutable installed-release proof.
unanchored. A matching declaration is reported under
`installed_source_provenance` as local consistency only, with
`release_origin_bound: false`. The separate `installed_release_origin` scope
remains unchecked until an independently anchored artifact digest or signed
installer receipt binds each installed payload to the signed catalog. The tool
does not accept a digest merely asserted inside the installed evidence document.
Only evidence produced in-process by `--collect-installed-evidence` is a local
observation, and it must be no more than five minutes old (with at most 30
@@ -466,7 +476,10 @@ claims use `passed` or `failed`; production claims use `approved` or `rejected`.
One claim per scope, unique control/artifact IDs, `issued_at < expires_at`, current
validity and exact digest binding are mandatory. Any schema, binding, time,
signature or authority blocker leaves every supplied boundary claim unchecked;
an authorized negative result is checked but requires review.
one malformed authority member or invalid validity interval invalidates the
supplied authority set as a whole. An authorized negative result is checked but
requires review only after every prerequisite, including installed
release-origin binding, is established.
Signatures cover UTF-8 JSON with the `signatures` member omitted, object keys
sorted, compact `,`/`:` separators and non-ASCII characters escaped, matching
the tool's deterministic canonicalization.
@@ -483,11 +496,13 @@ the tool's deterministic canonicalization.
With `--installed-evidence`, this command performs comparison and proof-binding
diagnostics only: the imported installed document is unsigned, so neither it nor
the boundary claim becomes accepted proof. An accepted boundary review requires
a trusted orchestrator to retain the direct in-process observation, obtain the
external signature over its digest, and call the review library within the
five-minute observation window. A future signed collector-attestation format
should make that exchange durable across processes.
the boundary claim becomes accepted proof. Direct in-process collection removes
that unsigned-import gap, but still establishes only local metadata consistency.
No boundary claim is admitted by the current format because it has no separately
anchored artifact digest or signed installer receipt. A future evidence contract
must bind such an independently trusted release-artifact receipt to the installed
evidence digest; a signed collector-attestation format should also make that
exchange durable across processes.
Target-environment and production-approval claims must use the assessment's
`deployment_profile.id` as `subject_id`. External-provider claims require the
@@ -499,8 +514,10 @@ The authority keyring is a governance trust root. Do not download or generate it
from the proof bundle being checked. The checker rejects proof-authority public
keys reused by either the published or independently trusted catalog keyring,
and rejects the same proof public-key material assigned to multiple authority
IDs. Authority `not_after` is exclusive. A target operator or approver must validate
the referenced drill/result artifacts before signing. The rerun verifies the
IDs. A malformed key, an invalid or empty validity interval, or ambiguous key
identity invalidates the supplied authority set. Authority `not_after` is
exclusive. A target operator or approver must validate the referenced
drill/result artifacts before signing. The rerun verifies the
attestation and its bindings; it does not fetch or reinterpret those artifacts.
`--verification-time` exists only for reproducible historical review. Supplying
it always marks the machine and human report as a **HISTORICAL override**; callers

17
docs/installed-composition-evidence.schema.json
View File

@@ -20,7 +20,7 @@
"format": "uri-reference"
},
"schema_version": {
"const": "0.2.0"
"const": "0.3.0"
},
"evidence_kind": {
"const": "govoplan.installed-composition"
@@ -122,8 +122,11 @@
{
"type": "object",
"additionalProperties": false,
"required": ["kind", "commit"],
"required": ["basis", "kind", "commit"],
"properties": {
"basis": {
"const": "local-pep610-metadata"
},
"kind": {
"const": "vcs-commit"
},
@@ -136,8 +139,11 @@
{
"type": "object",
"additionalProperties": false,
"required": ["kind", "sha256"],
"required": ["basis", "kind", "sha256"],
"properties": {
"basis": {
"const": "local-pep610-metadata"
},
"kind": {
"const": "archive"
},
@@ -149,8 +155,11 @@
{
"type": "object",
"additionalProperties": false,
"required": ["kind"],
"required": ["basis", "kind"],
"properties": {
"basis": {
"const": "local-pep610-metadata"
},
"kind": {
"enum": [
"editable-local",

251
tests/test_capability_fit_evidence.py
View File

@@ -55,7 +55,7 @@ class CapabilityFitEvidenceTests(unittest.TestCase):
"docs/capability-fit-proof-authority-keyring.schema.json"
)
def test_matching_installed_composition_proves_versions_records_and_commits(
def test_matching_installed_composition_proves_local_consistency_not_origin(
self,
) -> None:
catalog, keyring = signed_catalog(self.assessment)
@@ -71,6 +71,17 @@ class CapabilityFitEvidenceTests(unittest.TestCase):
self.assertTrue(report["proof_scope"]["installed_artifacts"]["valid"])
self.assertTrue(report["proof_scope"]["installed_record_integrity"]["valid"])
self.assertTrue(report["proof_scope"]["installed_source_provenance"]["valid"])
self.assertEqual(
"local-pep610-metadata",
report["proof_scope"]["installed_source_provenance"]["basis"],
)
self.assertFalse(
report["proof_scope"]["installed_source_provenance"][
"release_origin_bound"
]
)
self.assertFalse(report["proof_scope"]["installed_release_origin"]["checked"])
self.assertIsNone(report["proof_scope"]["installed_release_origin"]["valid"])
self.assertFalse(report["proof_scope"]["runtime_activation"]["checked"])
self.assertFalse(report["proof_scope"]["target_environment"]["checked"])
self.assertEqual(
@@ -78,6 +89,40 @@ class CapabilityFitEvidenceTests(unittest.TestCase):
report["proof_scope"]["installed_artifacts"]["evidence_sha256"],
)
def test_catalog_trust_blocker_invalidates_dependent_evidence_scopes(
self,
) -> None:
catalog, keyring = signed_catalog(self.assessment)
catalog["signatures"][0]["value"] = "AAAA"
installed = matching_installed_evidence(self.assessment)
proof, authority = signed_boundary_evidence(
assessment=self.assessment,
installed=installed,
claims=[boundary_claim("target_environment", "passed")],
allowed_scopes=["target_environment"],
)
report = self.review(
catalog=catalog,
keyring=keyring,
installed_evidence=installed,
boundary_evidence=proof,
authority_keyring=authority,
)
self.assertEqual("blocked", report["status"])
self.assertFalse(
report["proof_scope"]["catalog_signature_and_keyring"]["valid"]
)
self.assertFalse(report["proof_scope"]["installed_artifacts"]["checked"])
self.assertIsNone(report["proof_scope"]["installed_artifacts"]["valid"])
self.assertFalse(report["proof_scope"]["target_environment"]["checked"])
self.assertIsNone(report["proof_scope"]["target_environment"]["valid"])
self.assertIn(
"installed_catalog_trust_unavailable",
{item["code"] for item in report["findings"]},
)
def test_missing_extra_and_version_drift_have_deterministic_review_targets(
self,
) -> None:
@@ -137,7 +182,10 @@ class CapabilityFitEvidenceTests(unittest.TestCase):
for item in evidence["artifacts"]
if item["package_name"] == "govoplan-campaign"
)
campaign["source_provenance"] = {"kind": "editable-local"}
campaign["source_provenance"] = {
"basis": "local-pep610-metadata",
"kind": "editable-local",
}
campaign["record_integrity"] = {
**campaign["record_integrity"],
"status": "partial",
@@ -212,6 +260,7 @@ class CapabilityFitEvidenceTests(unittest.TestCase):
schema=self.installed_schema,
observation_mode="direct_local",
verification_time=datetime(2026, 7, 23, 12, tzinfo=UTC),
catalog_trusted=True,
)
codes = {item.code for item in result.findings}
@@ -317,6 +366,10 @@ class CapabilityFitEvidenceTests(unittest.TestCase):
)
observed = evidence["artifacts"][0]
self.assertEqual("editable-local", observed["source_provenance"]["kind"])
self.assertEqual(
"local-pep610-metadata",
observed["source_provenance"]["basis"],
)
self.assertEqual("verified", observed["record_integrity"]["status"])
encoded = json.dumps(evidence, sort_keys=True)
self.assertNotIn(str(root), encoded)
@@ -338,6 +391,108 @@ class CapabilityFitEvidenceTests(unittest.TestCase):
tampered["artifacts"][0]["record_integrity"]["mismatched_file_count"],
)
def test_collector_detects_payload_declared_in_record_but_missing(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
distribution, payload_file = create_distribution(root)
payload_file.unlink()
with mock.patch.object(sys, "path", [str(root), *sys.path]):
evidence = collect_installed_composition(
assessment=self.assessment,
collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
distributions=[distribution],
)
integrity = evidence["artifacts"][0]["record_integrity"]
self.assertEqual("mismatch", integrity["status"])
self.assertEqual(1, integrity["missing_file_count"])
def test_collector_rejects_duplicate_record_declarations(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
distribution, _ = create_distribution(root)
record_path = root / "govoplan_demo-1.0.0.dist-info" / "RECORD"
rows = record_path.read_text(encoding="utf-8").splitlines()
record_path.write_text(
"\n".join((*rows, rows[0])) + "\n",
encoding="utf-8",
)
with mock.patch.object(sys, "path", [str(root), *sys.path]):
evidence = collect_installed_composition(
assessment=self.assessment,
collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
distributions=[distribution],
)
integrity = evidence["artifacts"][0]["record_integrity"]
self.assertEqual("partial", integrity["status"])
self.assertGreater(integrity["unverifiable_file_count"], 0)
def test_collector_rejects_malformed_record_declarations(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
distribution, _ = create_distribution(root)
record_path = root / "govoplan_demo-1.0.0.dist-info" / "RECORD"
record_path.write_text(
record_path.read_text(encoding="utf-8") + "malformed,row\n",
encoding="utf-8",
)
with mock.patch.object(sys, "path", [str(root), *sys.path]):
evidence = collect_installed_composition(
assessment=self.assessment,
collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
distributions=[distribution],
)
integrity = evidence["artifacts"][0]["record_integrity"]
self.assertEqual("partial", integrity["status"])
self.assertGreater(integrity["unverifiable_file_count"], 0)
def test_collector_verifies_declared_record_sizes(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
distribution, _ = create_distribution(root)
record_path = root / "govoplan_demo-1.0.0.dist-info" / "RECORD"
rows = record_path.read_text(encoding="utf-8").splitlines()
path, digest, size = rows[0].split(",")
rows[0] = f"{path},{digest},{int(size) + 1}"
record_path.write_text("\n".join(rows) + "\n", encoding="utf-8")
with mock.patch.object(sys, "path", [str(root), *sys.path]):
evidence = collect_installed_composition(
assessment=self.assessment,
collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
distributions=[distribution],
)
integrity = evidence["artifacts"][0]["record_integrity"]
self.assertEqual("mismatch", integrity["status"])
self.assertEqual(1, integrity["mismatched_file_count"])
def test_collector_rejects_symlinked_distribution_metadata_root(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
create_distribution(root)
metadata_root = root / "govoplan_demo-1.0.0.dist-info"
real_metadata_root = root / "real-govoplan-demo.dist-info"
metadata_root.rename(real_metadata_root)
metadata_root.symlink_to(real_metadata_root, target_is_directory=True)
distribution = metadata.PathDistribution(metadata_root)
with mock.patch.object(sys, "path", [str(root), *sys.path]):
evidence = collect_installed_composition(
assessment=self.assessment,
collected_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
distributions=[distribution],
)
integrity = evidence["artifacts"][0]["record_integrity"]
self.assertEqual("unavailable", integrity["status"])
self.assertEqual(0, integrity["hashed_file_count"])
def test_collector_accepts_real_core_wheel_payload_scripts_data_and_pyc(
self,
) -> None:
@@ -604,7 +759,9 @@ class CapabilityFitEvidenceTests(unittest.TestCase):
json.dumps(reversed_order, sort_keys=True),
)
def test_signed_scope_authority_can_validate_only_its_external_claim(self) -> None:
def test_signed_boundary_claim_waits_for_anchored_installed_release_origin(
self,
) -> None:
catalog, keyring = signed_catalog(self.assessment)
installed = matching_installed_evidence(self.assessment)
proof, authority = signed_boundary_evidence(
@@ -622,10 +779,15 @@ class CapabilityFitEvidenceTests(unittest.TestCase):
authority_keyring=authority,
)
self.assertEqual("current", report["status"])
self.assertTrue(report["proof_scope"]["target_environment"]["valid"])
self.assertEqual("blocked", report["status"])
self.assertFalse(report["proof_scope"]["target_environment"]["checked"])
self.assertIsNone(report["proof_scope"]["target_environment"]["valid"])
self.assertFalse(report["proof_scope"]["external_providers"]["checked"])
self.assertFalse(report["proof_scope"]["production_approval"]["checked"])
self.assertIn(
"boundary_installed_release_origin_untrusted",
{item["code"] for item in report["findings"]},
)
self.assertEqual(
self.assessment["deployment_profile"]["id"],
report["proof_scope"]["target_environment"]["expected_subject_id"],
@@ -666,10 +828,15 @@ class CapabilityFitEvidenceTests(unittest.TestCase):
"boundary_provider_subject_unconfigured",
{item["code"] for item in unconfigured["findings"]},
)
self.assertEqual("current", accepted["status"])
self.assertEqual("blocked", accepted["status"])
provider_scope = accepted["proof_scope"]["external_providers"]
self.assertFalse(provider_scope["checked"])
self.assertEqual("provider:test", provider_scope["expected_subject_id"])
self.assertEqual("provider:test", provider_scope["observed_subject_id"])
self.assertIn(
"boundary_installed_release_origin_untrusted",
{item["code"] for item in accepted["findings"]},
)
def test_boundary_subject_mismatch_blocks_without_echoing_invalid_config(
self,
@@ -711,7 +878,7 @@ class CapabilityFitEvidenceTests(unittest.TestCase):
self.assertNotIn("private.invalid", encoded)
self.assertNotIn("secret", encoded)
def test_authorized_negative_claim_creates_scope_specific_review_target(
def test_authorized_negative_claim_waits_for_installed_origin_binding(
self,
) -> None:
catalog, keyring = signed_catalog(self.assessment)
@@ -734,14 +901,12 @@ class CapabilityFitEvidenceTests(unittest.TestCase):
authority_keyring=authority,
)
targets = {item["id"] for item in report["review_targets"]}
self.assertIn(
self.assertEqual("blocked", report["status"])
self.assertFalse(report["proof_scope"]["target_environment"]["checked"])
self.assertFalse(report["proof_scope"]["production_approval"]["checked"])
self.assertNotIn(
"boundary.target_environment.production-like-dev",
targets,
)
self.assertIn(
"boundary.production_approval.production-like-dev",
targets,
{item["id"] for item in report["review_targets"]},
)
def test_proof_authorities_must_be_independent_and_key_ids_unambiguous(
@@ -818,6 +983,55 @@ class CapabilityFitEvidenceTests(unittest.TestCase):
{item["code"] for item in report["findings"]},
)
def test_malformed_authority_members_invalidate_the_whole_keyring(self) -> None:
catalog, keyring = signed_catalog(self.assessment)
installed = matching_installed_evidence(self.assessment)
proof, authority = signed_boundary_evidence(
assessment=self.assessment,
installed=installed,
claims=[boundary_claim("target_environment", "passed")],
allowed_scopes=["target_environment"],
)
malformed = deepcopy(authority)
malformed["keys"].append(
{
"key_id": "authority:malformed",
"status": "active",
"public_key": "not-base64",
"allowed_scopes": ["target_environment"],
}
)
inverted = deepcopy(authority)
inverted["keys"][0]["not_before"] = "2026-08-01T00:00:00Z"
inverted["keys"][0]["not_after"] = "2026-07-01T00:00:00Z"
cases = (
(malformed, "boundary_authority_key_invalid"),
(inverted, "boundary_authority_key_interval_invalid"),
)
for supplied_keyring, expected_code in cases:
with self.subTest(expected_code=expected_code):
report = self.review(
catalog=catalog,
keyring=keyring,
installed_evidence=installed,
boundary_evidence=proof,
authority_keyring=supplied_keyring,
)
self.assertEqual("blocked", report["status"])
self.assertIn(
expected_code,
{item["code"] for item in report["findings"]},
)
self.assertIn(
"boundary_authority_untrusted",
{item["code"] for item in report["findings"]},
)
self.assertFalse(
report["proof_scope"]["target_environment"]["checked"]
)
def test_historical_verification_and_generated_runtime_gap_are_visible(
self,
) -> None:
@@ -855,6 +1069,7 @@ class CapabilityFitEvidenceTests(unittest.TestCase):
observation_mode="direct_local",
verification_time=datetime(2026, 7, 23, 12, tzinfo=UTC),
verification_mode="https://private.invalid/secret",
catalog_trusted=True,
)
encoded = json.dumps(result.proof_scope, sort_keys=True)
@@ -981,7 +1196,7 @@ class CapabilityFitEvidenceTests(unittest.TestCase):
def matching_installed_evidence(assessment: dict[str, object]) -> dict[str, object]:
return {
"$schema": "./installed-composition-evidence.schema.json",
"schema_version": "0.2.0",
"schema_version": "0.3.0",
"evidence_kind": "govoplan.installed-composition",
"assessment_id": assessment["assessment_id"],
"assessment_release": assessment["release"]["ref"],
@@ -1010,7 +1225,11 @@ def artifact(
"modules": []
if module_id == "core"
else [{"module_id": module_id, "manifest_version": version}],
"source_provenance": {"kind": "vcs-commit", "commit": commit},
"source_provenance": {
"basis": "local-pep610-metadata",
"kind": "vcs-commit",
"commit": commit,
},
"record_integrity": {
"status": "verified",
"hashed_file_count": 1,

3
tests/test_capability_fit_review.py
View File

@@ -360,7 +360,8 @@ class CapabilityFitReviewTests(unittest.TestCase):
self.assertIn("Capability fit rerun: current", rendered)
self.assertIn(
"No installed-artifact, target-provider, or production proof", rendered
"No installed-composition, installed-release-origin, target-environment, external-provider, production-approval proof",
rendered,
)
def test_public_fetch_failure_is_generic_and_blocking(self) -> None:

33
tools/assessments/govoplan_assessment/capability_fit.py
View File

@@ -363,6 +363,9 @@ def review_capability_fit(
evidence_verification_mode = (
"historical_override" if verification_time is not None else "current"
)
catalog_dependency_trusted = not any(
item.severity == "blocker" for item in findings
)
installed_review = review_installed_composition(
assessment=assessment,
catalog_entries=catalog_entries,
@@ -373,6 +376,7 @@ def review_capability_fit(
observation_mode=installed_evidence_mode,
verification_time=evidence_verification_time,
verification_mode=evidence_verification_mode,
catalog_trusted=catalog_dependency_trusted,
)
findings.extend(
Finding(item.severity, item.code, item.message, item.assessment_ids)
@@ -884,7 +888,17 @@ def build_report(
else {
"installed_artifacts": {"checked": False, "valid": None},
"installed_record_integrity": {"checked": False, "valid": None},
"installed_source_provenance": {"checked": False, "valid": None},
"installed_source_provenance": {
"checked": False,
"valid": None,
"basis": "local-pep610-metadata",
"release_origin_bound": False,
},
"installed_release_origin": {
"checked": False,
"valid": None,
"release_origin_bound": False,
},
"runtime_activation": {"checked": False, "valid": None},
}
)
@@ -925,7 +939,7 @@ def build_report(
**boundary_scope,
}
return {
"report_version": "0.3.0",
"report_version": "0.4.0",
"status": status,
"assessment_id": assessment.get("assessment_id"),
"assessment_release": assessment.get("release", {}).get("ref")
@@ -950,6 +964,9 @@ def render_review(report: dict[str, Any]) -> str:
proof_scope = report.get("proof_scope", {})
installed_scope = proof_scope.get("installed_artifacts", {})
installed_checked = installed_scope.get("checked") is True
release_origin_checked = (
proof_scope.get("installed_release_origin", {}).get("checked") is True
)
installed_supplied = bool(installed_scope.get("evidence_sha256")) or (
"installed_evidence_observation" in proof_scope
)
@@ -960,7 +977,7 @@ def render_review(report: dict[str, Any]) -> str:
(
"Scope: schema, signed release metadata, optional local tag provenance, "
+ (
"and locally observed installed-composition evidence."
"and locally observed installed-composition consistency evidence; release origin remains a separate proof boundary."
if installed_checked
else (
"and supplied installed-composition evidence that did not establish local proof."
@@ -1006,12 +1023,12 @@ def render_review(report: dict[str, Any]) -> str:
if report.get("proof_scope", {}).get(key, {}).get("checked") is not True
]
if not installed_checked:
unchecked_boundaries.insert(0, "installed-artifact")
if not installed_checked and len(unchecked_boundaries) == 4:
lines.append(
"No installed-artifact, target-provider, or production proof was performed."
unchecked_boundaries.insert(0, "installed-composition")
if not release_origin_checked:
unchecked_boundaries.insert(
1 if not installed_checked else 0, "installed-release-origin"
)
elif unchecked_boundaries:
if unchecked_boundaries:
lines.append("No " + ", ".join(unchecked_boundaries) + " proof was performed.")
return "\n".join(lines) + "\n"

441
tools/assessments/govoplan_assessment/evidence.py
View File

@@ -3,12 +3,14 @@
from __future__ import annotations
import base64
import csv
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
import hashlib
import hmac
from importlib import metadata
from importlib.util import source_from_cache
import io
import json
import os
from pathlib import Path, PurePosixPath
@@ -27,6 +29,8 @@ from jsonschema.exceptions import SchemaError
MAX_GOVOPLAN_DISTRIBUTIONS = 256
MAX_DISTRIBUTION_FILES = 10_000
MAX_COLLECTION_RECORD_FILES = 50_000
MAX_RECORD_BYTES = 4 * 1024 * 1024
MAX_RECORD_PATH_CHARACTERS = 4096
MAX_HASHED_FILE_BYTES = 64 * 1024 * 1024
MAX_HASHED_DISTRIBUTION_BYTES = 512 * 1024 * 1024
MAX_HASHED_COLLECTION_BYTES = 2 * 1024 * 1024 * 1024
@@ -56,17 +60,17 @@ NEGATIVE_RESULTS = {
}
PROOF_REQUIREMENTS = {
"target_environment": (
"A target-run result bound to this assessment release and installed-evidence digest.",
"A target-run result bound to this assessment release and installed evidence whose release origin is independently anchored.",
"An independently trusted authority key permitted for target_environment proof.",
"Opaque control IDs and content hashes for the target acceptance artifacts.",
),
"external_providers": (
"Provider acceptance results bound to this assessment release and installed-evidence digest.",
"Provider acceptance results bound to this assessment release and installed evidence whose release origin is independently anchored.",
"An independently trusted authority key permitted for external_providers proof.",
"Opaque control IDs and content hashes; credentials and endpoint identifiers stay outside the report.",
),
"production_approval": (
"An explicit approval bound to this assessment release and installed-evidence digest.",
"An explicit approval bound to this assessment release and installed evidence whose release origin is independently anchored.",
"An independently provisioned authority key permitted for production_approval.",
"An unexpired signed proof bundle; an assessment or operator cannot approve itself.",
),
@@ -104,6 +108,14 @@ class CollectionHashBudget:
remaining_bytes: int = MAX_HASHED_COLLECTION_BYTES
@dataclass(frozen=True, slots=True)
class RecordEntry:
path: str
hash_mode: str | None
hash_value: str | None
size: int | None
def validate_payload(
*, payload: dict[str, Any], schema: dict[str, Any]
) -> tuple[str, ...]:
@@ -231,7 +243,7 @@ def collect_installed_composition(
sorted_issues.sort(key=lambda item: (item["package_name"], item["code"]))
return {
"$schema": "./installed-composition-evidence.schema.json",
"schema_version": "0.2.0",
"schema_version": "0.3.0",
"evidence_kind": "govoplan.installed-composition",
"assessment_id": _safe_opaque_id(
assessment.get("assessment_id"), fallback="invalid-assessment"
@@ -264,6 +276,7 @@ def review_installed_composition(
observation_mode: str = "imported_unsigned",
verification_time: datetime | None = None,
verification_mode: str = "current",
catalog_trusted: bool = False,
) -> InstalledEvidenceReview:
if evidence is None:
return InstalledEvidenceReview(
@@ -337,6 +350,15 @@ def review_installed_composition(
("assessment.installed_composition",),
)
)
if catalog_trusted is not True:
findings.append(
EvidenceFinding(
"blocker",
"installed_catalog_trust_unavailable",
"Installed-composition proof depends on a valid signed catalog.",
("assessment.release", "assessment.installed_composition"),
)
)
changes: list[dict[str, Any]] = []
changed_repositories: set[str] = set()
affected_module_ids: set[str] = set()
@@ -391,6 +413,8 @@ def review_installed_composition(
)
if effective_verification_mode == "invalid":
observation_trusted = False
if catalog_trusted is not True:
observation_trusted = False
expected_release = (
assessment.get("release", {}).get("ref")
if isinstance(assessment.get("release"), dict)
@@ -501,7 +525,7 @@ def review_installed_composition(
record_verified = 0
generated_unhashed_file_count = 0
provenance_expected = 0
provenance_anchored = 0
provenance_declared_match = 0
provenance_mutable = 0
provenance_unanchored = 0
artifact_valid = binding_valid and not duplicate_packages
@@ -648,6 +672,23 @@ def review_installed_composition(
provenance_kind = (
str(provenance.get("kind") or "") if isinstance(provenance, dict) else ""
)
provenance_basis = (
str(provenance.get("basis") or "") if isinstance(provenance, dict) else ""
)
if provenance_basis != "local-pep610-metadata":
artifact_valid = False
provenance_unanchored += 1
_record_change(
findings=findings,
changes=changes,
changed_repositories=changed_repositories,
affected_module_ids=affected_module_ids,
component=component,
kind="installed_source_metadata_invalid",
code="installed_source_metadata_basis_invalid",
message=f"Installed distribution {package_name!r} does not identify its source statement as local PEP 610 metadata.",
)
continue
if provenance_kind == "vcs-commit":
commit = str(provenance.get("commit") or "").lower()
selected_version = selected_versions.get(repository)
@@ -660,7 +701,7 @@ def review_installed_composition(
or commit == selected_commit.lower()
)
if commit_matches and selected_matches:
provenance_anchored += 1
provenance_declared_match += 1
else:
artifact_valid = False
_record_change(
@@ -671,7 +712,7 @@ def review_installed_composition(
component=component,
kind="installed_commit_mismatch",
code="installed_source_commit_mismatch",
message=f"Installed immutable VCS provenance for {package_name!r} does not match the assessed or signed selected-unit commit.",
message=f"Installed distribution {package_name!r} declares a local PEP 610 VCS commit that does not match the assessed or signed selected-unit commit.",
)
elif provenance_kind in {"editable-vcs", "editable-local", "local-directory"}:
provenance_mutable += 1
@@ -684,7 +725,7 @@ def review_installed_composition(
component=component,
kind="installed_source_mutable",
code="installed_source_mutable",
message=f"Installed distribution {package_name!r} is editable or directory-backed and is not immutable release evidence.",
message=f"Installed distribution {package_name!r} declares an editable or directory-backed local source.",
)
else:
provenance_unanchored += 1
@@ -697,7 +738,7 @@ def review_installed_composition(
component=component,
kind="installed_source_unanchored",
code="installed_source_unanchored",
message=f"Installed distribution {package_name!r} has no source provenance anchored to the assessed release.",
message=f"Installed distribution {package_name!r} has no local PEP 610 source statement matching the assessed release.",
)
expected_packages = set(expected_by_package)
@@ -785,13 +826,25 @@ def review_installed_composition(
},
"installed_source_provenance": {
"checked": provenance_checked,
"valid": provenance_anchored == provenance_expected
"valid": provenance_declared_match == provenance_expected
if provenance_checked
else None,
"anchored_distribution_count": provenance_anchored,
"basis": "local-pep610-metadata",
"declared_match_distribution_count": provenance_declared_match,
"mutable_distribution_count": provenance_mutable,
"unanchored_distribution_count": provenance_unanchored,
"expected_distribution_count": provenance_expected,
"release_origin_bound": False,
},
"installed_release_origin": {
"checked": False,
"valid": None,
"release_origin_bound": False,
"anchored_artifact_digest_count": 0,
"expected_distribution_count": provenance_expected,
"required_evidence": (
"An independently anchored artifact digest or signed installer receipt that binds each installed payload to the signed release catalog.",
),
},
"runtime_activation": {
"checked": False,
@@ -807,6 +860,7 @@ def review_installed_composition(
"collected_at": _iso_datetime(collected_at),
"evaluated_at": _iso_datetime(now),
"verification_mode": effective_verification_mode,
"catalog_trusted": catalog_trusted is True,
"freshness_window_seconds": int(MAX_LOCAL_OBSERVATION_AGE.total_seconds()),
},
}
@@ -949,6 +1003,9 @@ def review_boundary_evidence(
)
installed_digest = installed_review.evidence_sha256
installed_scope = installed_review.proof_scope.get("installed_artifacts", {})
installed_origin_scope = installed_review.proof_scope.get(
"installed_release_origin", {}
)
if (
installed_digest is None
or evidence.get("installed_evidence_sha256") != installed_digest
@@ -962,6 +1019,15 @@ def review_boundary_evidence(
("assessment.installed_composition", "assessment.external_proof"),
)
)
if installed_origin_scope.get("valid") is not True:
findings.append(
EvidenceFinding(
"blocker",
"boundary_installed_release_origin_untrusted",
"Boundary evidence requires an independently anchored digest or signed installer receipt for every installed release artifact; local PEP 610 and RECORD metadata are insufficient.",
("assessment.installed_composition", "assessment.external_proof"),
)
)
issued_at = _datetime(evidence.get("issued_at"))
expires_at = _datetime(evidence.get("expires_at"))
@@ -1295,19 +1361,19 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]:
try:
raw = distribution.read_text("direct_url.json")
except Exception:
return {"kind": "malformed-direct-url"}
return _local_source_metadata("malformed-direct-url")
if raw is None:
return {"kind": "index-or-unknown"}
return _local_source_metadata("index-or-unknown")
if not isinstance(raw, str) or len(raw) > MAX_DIRECT_URL_BYTES:
return {"kind": "malformed-direct-url"}
return _local_source_metadata("malformed-direct-url")
if len(raw.encode("utf-8")) > MAX_DIRECT_URL_BYTES:
return {"kind": "malformed-direct-url"}
return _local_source_metadata("malformed-direct-url")
try:
direct_url = json.loads(raw)
except (TypeError, json.JSONDecodeError):
return {"kind": "malformed-direct-url"}
return _local_source_metadata("malformed-direct-url")
if not isinstance(direct_url, dict):
return {"kind": "malformed-direct-url"}
return _local_source_metadata("malformed-direct-url")
raw_url = direct_url.get("url")
if (
not isinstance(raw_url, str)
@@ -1315,13 +1381,13 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]:
or len(raw_url) > 2048
or any(character.isspace() or ord(character) < 32 for character in raw_url)
):
return {"kind": "malformed-direct-url"}
return _local_source_metadata("malformed-direct-url")
try:
parsed_url = urlsplit(raw_url)
except ValueError:
return {"kind": "malformed-direct-url"}
return _local_source_metadata("malformed-direct-url")
if parsed_url.password is not None or parsed_url.query or parsed_url.fragment:
return {"kind": "malformed-direct-url"}
return _local_source_metadata("malformed-direct-url")
vcs_info = direct_url.get("vcs_info")
directory_info = direct_url.get("dir_info")
archive_info = direct_url.get("archive_info")
@@ -1335,10 +1401,10 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]:
if value is not None
)
if len(provenance_forms) != 1:
return {"kind": "malformed-direct-url"}
return _local_source_metadata("malformed-direct-url")
if isinstance(vcs_info, dict):
if vcs_info.get("vcs") != "git":
return {"kind": "malformed-direct-url"}
return _local_source_metadata("malformed-direct-url")
commit = str(vcs_info.get("commit_id") or "").lower()
requested_revision = vcs_info.get("requested_revision")
if (
@@ -1359,8 +1425,8 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]:
)
)
):
return {"kind": "malformed-direct-url"}
return {"kind": "vcs-commit", "commit": commit}
return _local_source_metadata("malformed-direct-url")
return _local_source_metadata("vcs-commit", commit=commit)
if isinstance(archive_info, dict):
if (
parsed_url.scheme not in {"file", "https"}
@@ -1370,7 +1436,7 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]:
and (not parsed_url.hostname or parsed_url.username)
)
):
return {"kind": "malformed-direct-url"}
return _local_source_metadata("malformed-direct-url")
hashes = archive_info.get("hashes")
digest = hashes.get("sha256") if isinstance(hashes, dict) else None
if digest is None:
@@ -1379,8 +1445,8 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]:
digest = legacy.removeprefix("sha256=")
digest_text = str(digest or "").lower()
if SHA256_PATTERN.fullmatch(digest_text):
return {"kind": "archive", "sha256": digest_text}
return {"kind": "malformed-direct-url"}
return _local_source_metadata("archive", sha256=digest_text)
return _local_source_metadata("malformed-direct-url")
if isinstance(directory_info, dict):
editable = directory_info.get("editable", False)
if (
@@ -1388,9 +1454,142 @@ def _source_provenance(distribution: metadata.Distribution) -> dict[str, str]:
or not parsed_url.path
or not isinstance(editable, bool)
):
return {"kind": "malformed-direct-url"}
return {"kind": "editable-local" if editable else "local-directory"}
return {"kind": "malformed-direct-url"}
return _local_source_metadata("malformed-direct-url")
return _local_source_metadata(
"editable-local" if editable else "local-directory"
)
return _local_source_metadata("malformed-direct-url")
def _local_source_metadata(kind: str, **values: str) -> dict[str, str]:
return {"basis": "local-pep610-metadata", "kind": kind, **values}
def _read_record_entries(
*,
distribution: metadata.Distribution,
distribution_root: Path,
max_entries: int,
) -> tuple[tuple[RecordEntry, ...], int, str]:
"""Read RECORD itself without importlib's missing-path filtering.
``Distribution.files`` is an interpreted convenience API rather than the
raw declaration being attested. Reading the metadata file directly lets
this collector retain missing and malformed declarations and enforce byte
and row limits before resolving any untrusted payload path.
"""
metadata_root_value = getattr(distribution, "_path", None)
if metadata_root_value is None:
return (), 0, "unavailable"
try:
metadata_root = Path(os.fspath(metadata_root_value))
if metadata_root.is_symlink():
return (), 0, "unavailable"
resolved_metadata_root = metadata_root.resolve()
except (OSError, TypeError, ValueError):
return (), 0, "unavailable"
if not _is_relative_to(resolved_metadata_root, distribution_root):
return (), 0, "unavailable"
record_path = resolved_metadata_root / "RECORD"
try:
if record_path.is_symlink():
return (), 0, "unavailable"
encoded, exceeded, stable = _bounded_regular_file_bytes(
record_path,
max_bytes=MAX_RECORD_BYTES,
)
except OSError:
return (), 0, "unavailable"
if exceeded:
return (), 0, "limit-exceeded"
if not stable:
return (), 1, "unavailable"
try:
text = encoded.decode("utf-8")
except UnicodeDecodeError:
return (), 1, "unavailable"
entries: list[RecordEntry] = []
malformed_rows = 0
declared_paths: set[str] = set()
try:
rows = csv.reader(io.StringIO(text, newline=""))
for row in rows:
if len(entries) + malformed_rows >= max_entries:
return (), 0, "limit-exceeded"
if len(row) != 3:
malformed_rows += 1
continue
raw_path, raw_hash, raw_size = row
if (
not raw_path
or len(raw_path) > MAX_RECORD_PATH_CHARACTERS
or any(ord(character) < 32 for character in raw_path)
or (raw_size and (len(raw_size) > 20 or not raw_size.isdecimal()))
):
malformed_rows += 1
continue
declared_path = str(PurePosixPath(raw_path.replace("\\", "/")))
if declared_path in declared_paths:
malformed_rows += 1
continue
declared_paths.add(declared_path)
if raw_hash:
hash_mode, separator, hash_value = raw_hash.partition("=")
if not separator or not hash_mode or not hash_value:
hash_mode = raw_hash
hash_value = ""
else:
hash_mode = None
hash_value = None
entries.append(
RecordEntry(
path=raw_path,
hash_mode=hash_mode,
hash_value=hash_value,
size=int(raw_size) if raw_size else None,
)
)
except (csv.Error, ValueError):
return (), 1, "unavailable"
return tuple(entries), malformed_rows, "ok"
def _bounded_regular_file_bytes(
path: Path,
*,
max_bytes: int,
) -> tuple[bytes, bool, bool]:
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
descriptor = os.open(path, flags)
chunks: list[bytes] = []
total = 0
try:
initial = os.fstat(descriptor)
if not stat.S_ISREG(initial.st_mode):
raise OSError("RECORD metadata is not a regular file")
if initial.st_size > max_bytes:
return b"", True, True
while total <= max_bytes:
chunk = os.read(descriptor, min(1024 * 1024, max_bytes + 1 - total))
if not chunk:
break
chunks.append(chunk)
total += len(chunk)
final = os.fstat(descriptor)
finally:
os.close(descriptor)
exceeded = total > max_bytes or final.st_size > max_bytes
stable = (
not exceeded
and total == initial.st_size
and final.st_size == initial.st_size
and (final.st_dev, final.st_ino) == (initial.st_dev, initial.st_ino)
and final.st_mtime_ns == initial.st_mtime_ns
and final.st_ctime_ns == initial.st_ctime_ns
)
return b"".join(chunks), exceeded, stable
def _record_integrity(
@@ -1407,53 +1606,84 @@ def _record_integrity(
"missing_file_count": 0,
"mismatched_file_count": 0,
}
try:
files = tuple(distribution.files or ())
except Exception:
files = ()
if not files:
return {"status": "unavailable", **counts}
if len(files) > MAX_DISTRIBUTION_FILES or len(files) > budget.remaining_files:
budget.remaining_files = 0
return {"status": "limit-exceeded", **counts}
budget.remaining_files -= len(files)
try:
distribution_root = Path(distribution.locate_file("")).resolve()
installation_root = Path(sys.prefix).resolve()
except OSError:
except Exception:
return {"status": "unavailable", **counts}
if not _trusted_distribution_root(distribution_root):
return {"status": "unavailable", **counts}
entries, malformed_rows, record_status = _read_record_entries(
distribution=distribution,
distribution_root=distribution_root,
max_entries=min(MAX_DISTRIBUTION_FILES, budget.remaining_files),
)
if record_status == "limit-exceeded":
budget.remaining_files = 0
return {"status": "limit-exceeded", **counts}
if record_status != "ok":
counts["unverifiable_file_count"] = malformed_rows
return {"status": "unavailable", **counts}
record_file_count = len(entries) + malformed_rows
if record_file_count == 0:
return {"status": "unavailable", **counts}
budget.remaining_files -= record_file_count
counts["unverifiable_file_count"] = malformed_rows
declared_scripts = _declared_distribution_scripts(distribution)
owner_prefix = package_name.replace("-", "_")
resolved_entries: list[tuple[object, Path | None]] = [
resolved_entries: list[tuple[RecordEntry, Path | None]] = [
(
package_path,
entry,
_resolve_record_path(
distribution=distribution,
package_path=package_path,
package_path=entry.path,
distribution_root=distribution_root,
installation_root=installation_root,
owner_prefix=owner_prefix,
declared_scripts=declared_scripts,
),
)
for package_path in files
for entry in entries
]
resolved_path_counts: dict[Path, int] = {}
for _, resolved_path in resolved_entries:
if resolved_path is not None:
resolved_path_counts[resolved_path] = (
resolved_path_counts.get(resolved_path, 0) + 1
)
duplicate_resolved_paths = {
path for path, count in resolved_path_counts.items() if count > 1
}
counts["unverifiable_file_count"] += sum(
resolved_path_counts[path] for path in duplicate_resolved_paths
)
hashed_payload_paths = {
resolved_path
for package_path, resolved_path in resolved_entries
for entry, resolved_path in resolved_entries
if resolved_path is not None
and getattr(getattr(package_path, "hash", None), "mode", None) == "sha256"
and resolved_path not in duplicate_resolved_paths
and entry.hash_mode == "sha256"
}
total_hashed_bytes = 0
limit_exceeded = False
for package_path, resolved_path in resolved_entries:
hash_info = getattr(package_path, "hash", None)
for entry, resolved_path in resolved_entries:
if resolved_path is None:
counts["unverifiable_file_count"] += 1
continue
if hash_info is None:
if resolved_path in duplicate_resolved_paths:
continue
try:
file_stat = resolved_path.lstat()
except OSError:
counts["missing_file_count"] += 1
continue
if not stat.S_ISREG(file_stat.st_mode):
counts["missing_file_count"] += 1
continue
if entry.size is not None and file_stat.st_size != entry.size:
counts["mismatched_file_count"] += 1
continue
if entry.hash_mode is None:
if _permitted_unhashed_record_path(
resolved_path,
distribution_root=distribution_root,
@@ -1468,17 +1698,9 @@ def _record_integrity(
else:
counts["unverifiable_file_count"] += 1
continue
if getattr(hash_info, "mode", None) != "sha256":
if entry.hash_mode != "sha256":
counts["unverifiable_file_count"] += 1
continue
try:
file_stat = resolved_path.stat()
except OSError:
counts["missing_file_count"] += 1
continue
if not stat.S_ISREG(file_stat.st_mode):
counts["missing_file_count"] += 1
continue
remaining_distribution_bytes = (
MAX_HASHED_DISTRIBUTION_BYTES - total_hashed_bytes
)
@@ -1490,7 +1712,7 @@ def _record_integrity(
if allowed_bytes <= 0 or file_stat.st_size > allowed_bytes:
limit_exceeded = True
continue
expected = _record_hash_bytes(str(getattr(hash_info, "value", "")))
expected = _record_hash_bytes(entry.hash_value or "")
if expected is None:
counts["unverifiable_file_count"] += 1
continue
@@ -1541,7 +1763,7 @@ def _resolve_record_path(
if unresolved_path.is_symlink():
return None
resolved_path = unresolved_path.resolve()
except OSError:
except (OSError, TypeError, ValueError):
return None
has_parent_traversal = ".." in record_path.parts
if not has_parent_traversal:
@@ -1704,10 +1926,49 @@ def _active_authority_keys(
seen: set[str] = set()
public_key_owners: dict[bytes, str] = {}
duplicate_public_keys: set[bytes] = set()
for item in authority_keyring.get("keys", []):
raw_keys = authority_keyring.get("keys")
if not isinstance(raw_keys, list):
return {}, (
EvidenceFinding(
"blocker",
"boundary_authority_key_invalid",
"Proof-authority keyring does not contain a valid key list.",
("assessment.external_proof",),
),
)
for item in raw_keys:
if not isinstance(item, dict):
findings.append(
EvidenceFinding(
"blocker",
"boundary_authority_key_invalid",
"Proof-authority keyring contains a malformed key entry.",
("assessment.external_proof",),
)
)
continue
key_id = str(item.get("key_id") or "")
status = item.get("status")
raw_allowed_scopes = item.get("allowed_scopes")
if (
len(key_id) > 160
or OPAQUE_ID_PATTERN.fullmatch(key_id) is None
or status not in {"active", "next", "revoked", "disabled", "retired"}
or not isinstance(raw_allowed_scopes, list)
or not raw_allowed_scopes
or any(not isinstance(scope, str) for scope in raw_allowed_scopes)
or len(raw_allowed_scopes) != len(set(raw_allowed_scopes))
or any(scope not in PROOF_REQUIREMENTS for scope in raw_allowed_scopes)
):
findings.append(
EvidenceFinding(
"blocker",
"boundary_authority_key_invalid",
"Proof-authority keyring contains a malformed key entry.",
("assessment.external_proof",),
)
)
continue
if key_id in seen:
findings.append(
EvidenceFinding(
@@ -1725,6 +1986,36 @@ def _active_authority_keys(
)
Ed25519PublicKey.from_public_bytes(public_key)
except ValueError:
findings.append(
EvidenceFinding(
"blocker",
"boundary_authority_key_invalid",
"Proof-authority keyring contains invalid Ed25519 public-key material.",
("assessment.external_proof",),
)
)
continue
raw_not_before = item.get("not_before")
raw_not_after = item.get("not_after")
not_before = _datetime(raw_not_before)
not_after = _datetime(raw_not_after)
if (
(raw_not_before is not None and not_before is None)
or (raw_not_after is not None and not_after is None)
or (
not_before is not None
and not_after is not None
and not_before >= not_after
)
):
findings.append(
EvidenceFinding(
"blocker",
"boundary_authority_key_interval_invalid",
"Proof-authority keyring contains an invalid or empty key-validity interval.",
("assessment.external_proof",),
)
)
continue
previous_owner = public_key_owners.get(public_key)
if previous_owner is not None and previous_owner != key_id:
@@ -1741,10 +2032,8 @@ def _active_authority_keys(
result.pop(previous_owner, None)
continue
public_key_owners[public_key] = key_id
if item.get("status") not in {"active", "next"}:
if status not in {"active", "next"}:
continue
not_before = _datetime(item.get("not_before"))
not_after = _datetime(item.get("not_after"))
if (not_before is not None and verification_time < not_before) or (
not_after is not None and verification_time >= not_after
):
@@ -1761,9 +2050,12 @@ def _active_authority_keys(
continue
result[key_id] = (
public_key,
frozenset(str(value) for value in item.get("allowed_scopes", [])),
frozenset(raw_allowed_scopes),
)
return result, tuple(findings)
# A supplied authority keyring is one trust root. Selecting usable members
# out of a malformed set would let a valid-looking sibling conceal an
# invalid interval, duplicate identity, or ambiguous key assignment.
return ({} if findings else result), tuple(findings)
def _record_change(
@@ -1810,7 +2102,20 @@ def _unchecked_installed_scope() -> dict[str, Any]:
),
},
"installed_record_integrity": {"checked": False, "valid": None},
"installed_source_provenance": {"checked": False, "valid": None},
"installed_source_provenance": {
"checked": False,
"valid": None,
"basis": "local-pep610-metadata",
"release_origin_bound": False,
},
"installed_release_origin": {
"checked": False,
"valid": None,
"release_origin_bound": False,
"required_evidence": (
"An independently anchored artifact digest or signed installer receipt bound to the signed release catalog.",
),
},
"runtime_activation": {
"checked": False,
"valid": None,