From a5ebc69a6e0dfadc79e7e2c0b307868a15e265d4 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sun, 1 Nov 2026 16:56:26 +0100 Subject: [PATCH] feat(access-control-prover): decode installed fact authority --- .kiro/specs/access-control-prover/tasks.md | 2 +- .../_installed_fact_authority_codec.py | 248 ++++++++++++++++++ .../mapping/installed_fact_authority.py | 242 +++++++++++++++++ tests/test_installed_fact_authority.py | 247 +++++++++++++++++ 4 files changed, 738 insertions(+), 1 deletion(-) create mode 100644 src/openultrasast/mapping/_installed_fact_authority_codec.py create mode 100644 src/openultrasast/mapping/installed_fact_authority.py create mode 100644 tests/test_installed_fact_authority.py diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index 8785701..e6076f9 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -345,7 +345,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with - _Requirements: 13.2, 13.3_ - _Boundary: Fact Authority_ - _Depends: 14.1_ -- [ ] 14.3 Implement installed authority codecs and resolver +- [x] 14.3 Implement installed authority codecs and resolver - Implement closed canonical JSON decoders for the bootstrap policy, installed authority, v2 fact packs, nested policies/lineages/evidence and immutable content references. - Implement the checked-in digest-addressed resolver using only validated `content_digest` for filesystem addressing and re-derivable storage receipts under an injected bootstrap-admitted store identity. - Done when canonical synthetic installed fixtures round-trip to Task 14.2 typed values and unknown/missing fields, noncanonical bytes, traversal-like artifact IDs, digest paths or byte substitutions fail before authentication. diff --git a/src/openultrasast/mapping/_installed_fact_authority_codec.py b/src/openultrasast/mapping/_installed_fact_authority_codec.py new file mode 100644 index 0000000..5951c70 --- /dev/null +++ b/src/openultrasast/mapping/_installed_fact_authority_codec.py @@ -0,0 +1,248 @@ +"""Nested value codecs for the installed fact-authority wire boundary.""" + +from __future__ import annotations + +import json +from collections.abc import Callable, Mapping +from dataclasses import fields, is_dataclass +from enum import Enum +from typing import TypeVar, cast + +from openultrasast.core import canonical, provenance, semantics +from openultrasast.core.factpack import ( + IndependenceAttestation, + KnowledgeEvidence, + KnowledgeEvidenceKind, + RelationEffect, + ReviewIndependencePolicy, + SemanticFact, + StructuralRequirement, + StructuralRequirementKind, +) + +_T = TypeVar("_T") +_REF_FIELDS = ("artifact_id", "content_digest", "media_type", "byte_length", "retention_policy_digest", "storage_receipt_digest") +_ATTESTATION_FIELDS = ("subject_digest", "producer_identity", "reviewer_identity", "conflict_disclosure", "attestation_digest") + + +class InstalledAuthorityWireError(ValueError): + """Installed authority bytes do not satisfy the closed wire contract.""" + + +def wire_value(value: object) -> canonical.JsonValue: + """Project dataclasses, enums, and parsed JSON into canonical values.""" + if isinstance(value, Enum): + return cast(canonical.JsonValue, value.value) + if is_dataclass(value) and not isinstance(value, type): + return {item.name: wire_value(getattr(value, item.name)) for item in fields(value)} + if isinstance(value, Mapping): + if not all(isinstance(key, str) for key in value): + raise InstalledAuthorityWireError("JSON object keys must be text") + return {cast(str, key): wire_value(item) for key, item in value.items()} + if isinstance(value, (tuple, list)): + return tuple(wire_value(item) for item in value) + if value is None or isinstance(value, (str, bool, int, float)): + return value + raise InstalledAuthorityWireError(f"unsupported wire value: {type(value).__name__}") + + +def closed_object(value: object, expected: tuple[str, ...], label: str) -> dict[str, object]: + """Require a JSON object with exactly the declared field set.""" + if not isinstance(value, dict) or not all(isinstance(key, str) for key in value): + raise InstalledAuthorityWireError(f"{label} must be an object") + record = cast(dict[str, object], value) + missing = set(expected) - set(record) + unknown = set(record) - set(expected) + if missing or unknown: + details = [f"missing {item}" for item in sorted(missing)] + details.extend(f"unknown {item}" for item in sorted(unknown)) + raise InstalledAuthorityWireError(f"{label} has {', '.join(details)}") + return record + + +def items(value: object, label: str) -> tuple[object, ...]: + if not isinstance(value, list): + raise InstalledAuthorityWireError(f"{label} must be an array") + return tuple(value) + + +def text(value: object, label: str) -> str: + if not isinstance(value, str) or not value: + raise InstalledAuthorityWireError(f"{label} must be nonempty text") + return value + + +def integer(value: object, label: str) -> int: + if isinstance(value, bool) or not isinstance(value, int): + raise InstalledAuthorityWireError(f"{label} must be an integer") + return value + + +def boolean(value: object, label: str) -> bool: + if not isinstance(value, bool): + raise InstalledAuthorityWireError(f"{label} must be boolean") + return value + + +def strings(value: object, label: str) -> tuple[str, ...]: + return tuple(text(item, label) for item in items(value, label)) + + +def pairs(value: object, label: str) -> tuple[tuple[str, str], ...]: + decoded: list[tuple[str, str]] = [] + for item in items(value, label): + pair = items(item, label) + if len(pair) != 2: + raise InstalledAuthorityWireError(f"{label} entries must be pairs") + decoded.append((text(pair[0], label), text(pair[1], label))) + return tuple(decoded) + + +def parse_record(content: bytes, schema: str, expected: tuple[str, ...], label: str) -> dict[str, object]: + """Parse canonical JSON and enforce its top-level schema and fields.""" + if not isinstance(content, bytes): + raise InstalledAuthorityWireError(f"{label} content must be bytes") + try: + raw: object = json.loads(content) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise InstalledAuthorityWireError(f"{label} is not valid JSON") from exc + record = closed_object(raw, expected, label) + if record["schema_version"] != schema: + raise InstalledAuthorityWireError(f"unknown {label} schema") + if canonical.canonical_bytes(wire_value(record)) != content: + raise InstalledAuthorityWireError(f"{label} bytes are not canonical") + return record + + +def decode_value(label: str, build: Callable[[], _T]) -> _T: + """Normalize typed construction failures to the wire error boundary.""" + try: + return build() + except InstalledAuthorityWireError: + raise + except (KeyError, TypeError, ValueError) as exc: + raise InstalledAuthorityWireError(f"invalid {label}: {exc}") from exc + + +def content_ref(value: object) -> canonical.ContentRef: + row = closed_object(value, _REF_FIELDS, "content reference") + return canonical.ContentRef( + text(row["artifact_id"], "artifact_id"), + text(row["content_digest"], "content_digest"), + text(row["media_type"], "media_type"), + integer(row["byte_length"], "byte_length"), + text(row["retention_policy_digest"], "retention_policy_digest"), + text(row["storage_receipt_digest"], "storage_receipt_digest"), + ) + + +def attestation(value: object) -> IndependenceAttestation: + row = closed_object(value, _ATTESTATION_FIELDS, "attestation") + return IndependenceAttestation( + text(row["subject_digest"], "subject_digest"), + text(row["producer_identity"], "producer_identity"), + text(row["reviewer_identity"], "reviewer_identity"), + content_ref(row["conflict_disclosure"]), + text(row["attestation_digest"], "attestation_digest"), + ) + + +def evidence(value: object) -> KnowledgeEvidence: + expected = ("kind", "content", "producer_identity", "reviewer_identity", "independence", "lineage_manifest_digest") + row = closed_object(value, expected, "knowledge evidence") + return KnowledgeEvidence( + KnowledgeEvidenceKind(text(row["kind"], "evidence kind")), + content_ref(row["content"]), + text(row["producer_identity"], "producer_identity"), + text(row["reviewer_identity"], "reviewer_identity"), + attestation(row["independence"]), + text(row["lineage_manifest_digest"], "lineage_manifest_digest"), + ) + + +def requirement(value: object) -> StructuralRequirement: + row = closed_object(value, ("requirement_id", "kind", "field", "expected"), "structural requirement") + return StructuralRequirement( + text(row["requirement_id"], "requirement_id"), + StructuralRequirementKind(text(row["kind"], "requirement kind")), + text(row["field"], "requirement field"), + wire_value(row["expected"]), + ) + + +def effect(value: object) -> RelationEffect: + row = closed_object(value, ("kind", "role", "source_role", "target_role", "conditions"), "relation effect") + return RelationEffect( + semantics.RelationKind(text(row["kind"], "relation kind")), + text(row["role"], "effect role"), + text(row["source_role"], "source role"), + text(row["target_role"], "target role"), + strings(row["conditions"], "conditions"), + ) + + +def fact(value: object) -> SemanticFact: + row = closed_object(value, ("fact_id", "role", "structural_requirements", "relation_effects"), "semantic fact") + requirements = items(row["structural_requirements"], "structural_requirements") + effects = items(row["relation_effects"], "relation_effects") + return SemanticFact( + text(row["fact_id"], "fact_id"), + text(row["role"], "fact role"), + tuple(requirement(item) for item in requirements), + tuple(effect(item) for item in effects), + ) + + +def lineage(value: object) -> provenance.LineageManifest: + row = closed_object(value, ("arm_id", "subject_digest", "uses", "manifest_digest"), "lineage") + expected = ( + "purpose", + "source_population_digest", + "labels_accessed", + "influenced_artifact_digest", + "population_use_receipt_digest", + ) + uses = tuple(_lineage_use(item, expected) for item in items(row["uses"], "lineage uses")) + return provenance.LineageManifest( + text(row["arm_id"], "arm_id"), + text(row["subject_digest"], "subject_digest"), + uses, + text(row["manifest_digest"], "manifest_digest"), + ) + + +def _lineage_use(value: object, expected: tuple[str, ...]) -> provenance.LineageUse: + row = closed_object(value, expected, "lineage use") + return provenance.LineageUse( + provenance.LineagePurpose(text(row["purpose"], "lineage purpose")), + text(row["source_population_digest"], "source population"), + boolean(row["labels_accessed"], "labels_accessed"), + text(row["influenced_artifact_digest"], "influenced artifact"), + text(row["population_use_receipt_digest"], "population receipt"), + ) + + +def custody_policy(value: object) -> provenance.PopulationCustodyPolicy: + row = closed_object(value, ("reviewed_custodian_digests", "policy_digest"), "custody policy") + return provenance.PopulationCustodyPolicy( + strings(row["reviewed_custodian_digests"], "custodian digests"), + text(row["policy_digest"], "custody policy digest"), + ) + + +def review_policy(value: object) -> ReviewIndependencePolicy: + expected = ( + "reviewed_reviewer_digests", + "forbidden_identity_pairs", + "trusted_store_identities", + "maximum_evidence_bytes", + "policy_digest", + ) + row = closed_object(value, expected, "review policy") + return ReviewIndependencePolicy( + strings(row["reviewed_reviewer_digests"], "reviewer digests"), + pairs(row["forbidden_identity_pairs"], "forbidden pairs"), + strings(row["trusted_store_identities"], "store identities"), + integer(row["maximum_evidence_bytes"], "maximum evidence bytes"), + text(row["policy_digest"], "review policy digest"), + ) diff --git a/src/openultrasast/mapping/installed_fact_authority.py b/src/openultrasast/mapping/installed_fact_authority.py new file mode 100644 index 0000000..6450a20 --- /dev/null +++ b/src/openultrasast/mapping/installed_fact_authority.py @@ -0,0 +1,242 @@ +"""Closed canonical wire format for installed entry-fact authority.""" + +from __future__ import annotations + +import re +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path +from typing import cast + +from openultrasast.core import canonical +from openultrasast.core.factpack import FACT_PACK_SCHEMA, FactPack +from openultrasast.core.family import FamilyContract +from openultrasast.mapping._installed_fact_authority_codec import ( + InstalledAuthorityWireError, + attestation, + content_ref, + custody_policy, + decode_value, + evidence, + fact, + integer, + items, + lineage, + pairs, + parse_record, + review_policy, + strings, + text, + wire_value, +) +from openultrasast.mapping.fact_authority import ( + ENTRY_AUTHORITY_BOOTSTRAP_SCHEMA, + EntryAuthorityBootstrapPolicy, + EntryFactAuthorityBundle, +) + +INSTALLED_AUTHORITY_SCHEMA = "openultrasast.access-control.installed-entry-authority.v1" +PACK_FILES = ("express.javascript.v2.json", "express.typescript.v2.json") +_PACK_LANGUAGES = ("javascript", "typescript") +_DIGEST = re.compile(r"[0-9a-f]{64}") +_PACK_FIELDS = ( + "schema_version", + "family_contract_digest", + "family_id", + "ecosystem", + "language", + "facts", + "normalized_content_digest", + "review_identity", + "review_evidence", + "review_evidence_digest", + "review_policy_digest", +) +_BOOTSTRAP_FIELDS = ( + "schema_version", + "reviewed_bundle_reviewer_digests", + "forbidden_identity_pairs", + "trusted_store_identities", + "maximum_bundle_bytes", + "policy_digest", +) +_AUTHORITY_FIELDS = ( + "schema_version", + "pack_files", + "recognition_projection", + "lineages", + "custody_policy", + "review_policy", + "matcher_revision", + "matcher_behavior_digest", + "matcher_lineage_manifest_digest", + "authority_review", + "bundle_digest", +) + + +def _record_bytes(value: object, schema: str) -> bytes: + record = cast(Mapping[str, canonical.JsonValue], wire_value(value)) + return canonical.canonical_record_bytes(record, known_schema_versions=(schema,)) + + +def fact_pack_bytes(value: FactPack) -> bytes: + """Encode every public fact-pack field as canonical JSON.""" + return _record_bytes(value, FACT_PACK_SCHEMA) + + +def decode_fact_pack_bytes(content: bytes) -> FactPack: + """Decode only the closed v2 fact-pack representation.""" + + def build() -> FactPack: + row = parse_record(content, FACT_PACK_SCHEMA, _PACK_FIELDS, "fact pack") + value = FactPack( + FACT_PACK_SCHEMA, + text(row["family_contract_digest"], "family_contract_digest"), + text(row["family_id"], "family_id"), + text(row["ecosystem"], "ecosystem"), + text(row["language"], "language"), + tuple(fact(item) for item in items(row["facts"], "facts")), + text(row["normalized_content_digest"], "normalized_content_digest"), + text(row["review_identity"], "review_identity"), + tuple(evidence(item) for item in items(row["review_evidence"], "review_evidence")), + text(row["review_evidence_digest"], "review_evidence_digest"), + text(row["review_policy_digest"], "review_policy_digest"), + ) + if fact_pack_bytes(value) != content: + raise InstalledAuthorityWireError("fact pack does not round-trip canonically") + return value + + return decode_value("fact pack", build) + + +def bootstrap_bytes(value: EntryAuthorityBootstrapPolicy) -> bytes: + """Encode the external bootstrap policy without authority data.""" + row = cast(dict[str, canonical.JsonValue], wire_value(value)) + return canonical.canonical_record_bytes( + {"schema_version": ENTRY_AUTHORITY_BOOTSTRAP_SCHEMA, **row}, + known_schema_versions=(ENTRY_AUTHORITY_BOOTSTRAP_SCHEMA,), + ) + + +def decode_bootstrap_bytes(content: bytes) -> EntryAuthorityBootstrapPolicy: + """Decode a closed external bootstrap policy.""" + + def build() -> EntryAuthorityBootstrapPolicy: + row = parse_record(content, ENTRY_AUTHORITY_BOOTSTRAP_SCHEMA, _BOOTSTRAP_FIELDS, "bootstrap") + value = EntryAuthorityBootstrapPolicy( + strings(row["reviewed_bundle_reviewer_digests"], "bundle reviewers"), + pairs(row["forbidden_identity_pairs"], "forbidden pairs"), + strings(row["trusted_store_identities"], "store identities"), + integer(row["maximum_bundle_bytes"], "maximum bundle bytes"), + text(row["policy_digest"], "policy digest"), + ) + if bootstrap_bytes(value) != content: + raise InstalledAuthorityWireError("bootstrap does not round-trip canonically") + return value + + return decode_value("bootstrap", build) + + +def _authority_record(value: EntryFactAuthorityBundle) -> dict[str, canonical.JsonValue]: + return { + "schema_version": INSTALLED_AUTHORITY_SCHEMA, + "pack_files": PACK_FILES, + "recognition_projection": wire_value(value.recognition_projection), + "lineages": wire_value(value.lineages), + "custody_policy": wire_value(value.custody_policy), + "review_policy": wire_value(value.review_policy), + "matcher_revision": value.matcher_revision, + "matcher_behavior_digest": value.matcher_behavior_digest, + "matcher_lineage_manifest_digest": value.matcher_lineage_manifest_digest, + "authority_review": wire_value(value.authority_review), + "bundle_digest": value.bundle_digest, + } + + +def installed_authority_bytes(value: EntryFactAuthorityBundle) -> bytes: + """Encode installed authority while leaving bootstrap trust external.""" + return canonical.canonical_record_bytes( + _authority_record(value), + known_schema_versions=(INSTALLED_AUTHORITY_SCHEMA,), + ) + + +def decode_installed_authority_bytes( + content: bytes, + packs: tuple[FactPack, ...], + contract: FamilyContract, +) -> EntryFactAuthorityBundle: + """Decode closed installed authority and attach separately decoded packs.""" + + def build() -> EntryFactAuthorityBundle: + row = parse_record(content, INSTALLED_AUTHORITY_SCHEMA, _AUTHORITY_FIELDS, "installed authority") + if strings(row["pack_files"], "pack_files") != PACK_FILES: + raise InstalledAuthorityWireError("pack_files must use the fixed order") + if tuple(pack.language for pack in packs) != _PACK_LANGUAGES: + raise InstalledAuthorityWireError("packs must use the fixed language order") + value = EntryFactAuthorityBundle( + contract, + packs, + content_ref(row["recognition_projection"]), + tuple(lineage(item) for item in items(row["lineages"], "lineages")), + custody_policy(row["custody_policy"]), + review_policy(row["review_policy"]), + text(row["matcher_revision"], "matcher_revision"), + text(row["matcher_behavior_digest"], "matcher_behavior_digest"), + text(row["matcher_lineage_manifest_digest"], "matcher_lineage_manifest_digest"), + attestation(row["authority_review"]), + text(row["bundle_digest"], "bundle_digest"), + ) + if installed_authority_bytes(value) != content: + raise InstalledAuthorityWireError("installed authority does not round-trip canonically") + return value + + return decode_value("installed authority", build) + + +@dataclass(frozen=True) +class DigestDirectoryResolver: + """Read immutable artifacts by validated content digest only.""" + + root: Path + store_identity: str + + def resolve(self, reference: canonical.ContentRef) -> canonical.ResolvedContent: + if _DIGEST.fullmatch(reference.content_digest) is None: + raise canonical.ArtifactResolutionError("content digest is not lowercase SHA-256") + path = self.root / "artifacts" / "sha256" / reference.content_digest + try: + content = path.read_bytes() + except OSError as exc: + raise canonical.ArtifactResolutionError("digest-addressed artifact is unavailable") from exc + if len(content) != reference.byte_length or canonical.bytes_digest(content) != reference.content_digest: + raise canonical.ArtifactResolutionError("digest-addressed artifact bytes do not match reference") + receipt_digest = canonical.derive_storage_receipt_digest( + store_identity=self.store_identity, + content_digest=reference.content_digest, + byte_length=reference.byte_length, + retention_policy_digest=reference.retention_policy_digest, + ) + receipt = canonical.StorageReceipt( + self.store_identity, + reference.content_digest, + reference.byte_length, + reference.retention_policy_digest, + receipt_digest, + ) + return canonical.ResolvedContent(content, receipt) + + +__all__ = [ + "DigestDirectoryResolver", + "INSTALLED_AUTHORITY_SCHEMA", + "InstalledAuthorityWireError", + "PACK_FILES", + "bootstrap_bytes", + "decode_bootstrap_bytes", + "decode_fact_pack_bytes", + "decode_installed_authority_bytes", + "fact_pack_bytes", + "installed_authority_bytes", +] diff --git a/tests/test_installed_fact_authority.py b/tests/test_installed_fact_authority.py new file mode 100644 index 0000000..2548510 --- /dev/null +++ b/tests/test_installed_fact_authority.py @@ -0,0 +1,247 @@ +"""Closed installed-authority wire and digest-addressed resolver tests.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from openultrasast.core import canonical, provenance, semantics +from openultrasast.core.factpack import ( + FACT_PACK_SCHEMA, + FactPack, + IndependenceAttestation, + KnowledgeEvidence, + KnowledgeEvidenceKind, + RelationEffect, + ReviewIndependencePolicy, + SemanticFact, + StructuralRequirement, + StructuralRequirementKind, +) +from openultrasast.core.family import family_contract_digest +from openultrasast.mapping.fact_authority import ( + EntryAuthorityBootstrapPolicy, + EntryFactAuthorityBundle, + derive_entry_authority_bootstrap_policy_digest, +) +from openultrasast.mapping.installed_fact_authority import ( + DigestDirectoryResolver, + InstalledAuthorityWireError, + bootstrap_bytes, + decode_bootstrap_bytes, + decode_fact_pack_bytes, + decode_installed_authority_bytes, + fact_pack_bytes, + installed_authority_bytes, +) +from openultrasast.semantic.obligations.core_contract import production_access_control_contract + + +def _reference(label: str = "artifact") -> canonical.ContentRef: + digest = canonical.content_digest(label) + retention = canonical.content_digest("retain") + receipt = canonical.derive_storage_receipt_digest( + store_identity="installed-store", + content_digest=digest, + byte_length=len(label), + retention_policy_digest=retention, + ) + return canonical.ContentRef(label, digest, "application/json", len(label), retention, receipt) + + +def _canonical_json(value: object) -> bytes: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode() + + +def _pack(language: str) -> FactPack: + contract = production_access_control_contract() + producer = canonical.content_digest(f"producer:{language}") + reviewer = canonical.content_digest("reviewer") + content = _reference(f"evidence:{language}") + attestation = IndependenceAttestation( + content.content_digest, + producer, + reviewer, + _reference(f"disclosure:{language}"), + canonical.content_digest(f"attestation:{language}"), + ) + evidence = KnowledgeEvidence( + KnowledgeEvidenceKind.SOURCE_CODE, + content, + producer, + reviewer, + attestation, + canonical.content_digest(f"lineage:{language}"), + ) + fact = SemanticFact( + f"entry:test:{language}", + "entry_registration", + (StructuralRequirement("target", StructuralRequirementKind.QUALIFIED_NAME, "target", ("app", "get")),), + (RelationEffect(semantics.RelationKind.REGISTRATION, "entry_registration", "framework_instance", "entry_handler", ()),), + ) + return FactPack( + FACT_PACK_SCHEMA, + family_contract_digest(contract), + contract.family_id, + "express", + language, + (fact,), + canonical.content_digest(f"pack:{language}"), + reviewer, + (evidence,), + canonical.content_digest(f"review-evidence:{language}"), + canonical.content_digest("review-policy"), + ) + + +def _bootstrap() -> EntryAuthorityBootstrapPolicy: + reviewer = (canonical.content_digest("reviewer"),) + forbidden = ((canonical.content_digest("bad-producer"), reviewer[0]),) + stores = ("installed-store",) + digest = derive_entry_authority_bootstrap_policy_digest(reviewer, forbidden, stores, 4096) + return EntryAuthorityBootstrapPolicy(reviewer, forbidden, stores, 4096, digest) + + +def _bundle(packs: tuple[FactPack, ...]) -> EntryFactAuthorityBundle: + subject = canonical.content_digest("matcher") + use = provenance.LineageUse( + provenance.LineagePurpose.MATCHING_BEHAVIOR, + canonical.content_digest("population"), + False, + subject, + canonical.content_digest("receipt"), + ) + manifest = provenance.LineageManifest( + "matcher", + subject, + (use,), + provenance.derive_lineage_manifest_digest("matcher", subject, (use,)), + ) + custodian = (canonical.content_digest("custodian"),) + custody = provenance.PopulationCustodyPolicy( + custodian, + provenance.derive_population_custody_policy_digest(custodian), + ) + reviewer = (canonical.content_digest("reviewer"),) + review_policy = ReviewIndependencePolicy(reviewer, (), ("installed-store",), 4096, canonical.content_digest("policy")) + review = IndependenceAttestation( + canonical.content_digest("bundle"), + canonical.content_digest("producer"), + reviewer[0], + _reference("disclosure"), + canonical.content_digest("attestation"), + ) + return EntryFactAuthorityBundle( + production_access_control_contract(), + packs, + _reference("projection"), + (manifest,), + custody, + review_policy, + "matcher-v1", + subject, + manifest.manifest_digest, + review, + canonical.content_digest("bundle"), + ) + + +def test_closed_codecs_round_trip_typed_values() -> None: + packs = (_pack("javascript"), _pack("typescript")) + bootstrap = _bootstrap() + bundle = _bundle(packs) + + assert decode_bootstrap_bytes(bootstrap_bytes(bootstrap)) == bootstrap + assert tuple(decode_fact_pack_bytes(fact_pack_bytes(pack)) for pack in packs) == packs + encoded = installed_authority_bytes(bundle) + assert decode_installed_authority_bytes(encoded, packs, bundle.contract) == bundle + + +@pytest.mark.parametrize("kind", ("unknown", "missing", "noncanonical")) +def test_bootstrap_decoder_refuses_open_or_noncanonical_json(kind: str) -> None: + content = bootstrap_bytes(_bootstrap()) + raw = json.loads(content) + if kind == "unknown": + raw["extra"] = True + content = _canonical_json(raw) + elif kind == "missing": + del raw["policy_digest"] + content = _canonical_json(raw) + else: + content = json.dumps(raw, indent=2).encode() + + with pytest.raises(InstalledAuthorityWireError): + decode_bootstrap_bytes(content) + + +def test_pack_decoder_refuses_unknown_nested_fields() -> None: + content = fact_pack_bytes(_pack("javascript")) + raw = json.loads(content) + raw["facts"][0]["structural_requirements"][0]["extra"] = "forbidden" + + with pytest.raises(InstalledAuthorityWireError): + decode_fact_pack_bytes(_canonical_json(raw)) + + +def test_authority_decoder_requires_fixed_pack_file_order() -> None: + packs = (_pack("javascript"), _pack("typescript")) + bundle = _bundle(packs) + raw = json.loads(installed_authority_bytes(bundle)) + raw["pack_files"].reverse() + + with pytest.raises(InstalledAuthorityWireError, match="pack_files"): + decode_installed_authority_bytes(_canonical_json(raw), packs, bundle.contract) + + +def test_resolver_ignores_traversal_like_artifact_id_and_uses_digest(tmp_path: Path) -> None: + content = b'{"reviewed":true}' + digest = canonical.bytes_digest(content) + store = tmp_path / "artifacts" / "sha256" + store.mkdir(parents=True) + (store / digest).write_bytes(content) + retention = canonical.content_digest("retain") + receipt = canonical.derive_storage_receipt_digest( + store_identity="installed-store", + content_digest=digest, + byte_length=len(content), + retention_policy_digest=retention, + ) + reference = canonical.ContentRef("../../outside", digest, "application/json", len(content), retention, receipt) + + resolved = DigestDirectoryResolver(tmp_path, "installed-store").resolve(reference) + + assert resolved.content == content + assert resolved.receipt.receipt_digest == receipt + + +def test_resolver_rejects_traversal_like_content_digest_before_read(tmp_path: Path) -> None: + reference = _reference("logical") + object.__setattr__(reference, "content_digest", "../../outside") + + with pytest.raises(canonical.ArtifactResolutionError, match="lowercase SHA-256"): + DigestDirectoryResolver(tmp_path, "installed-store").resolve(reference) + + +@pytest.mark.parametrize("mutation", ("missing", "bytes", "digest-path")) +def test_resolver_refuses_missing_or_substituted_digest_content(tmp_path: Path, mutation: str) -> None: + content = b"trusted" + digest = canonical.bytes_digest(content) + store = tmp_path / "artifacts" / "sha256" + store.mkdir(parents=True) + retention = canonical.content_digest("retain") + receipt = canonical.derive_storage_receipt_digest( + store_identity="installed-store", + content_digest=digest, + byte_length=len(content), + retention_policy_digest=retention, + ) + reference = canonical.ContentRef("logical", digest, "application/json", len(content), retention, receipt) + if mutation == "bytes": + (store / digest).write_bytes(b"changed") + elif mutation == "digest-path": + (store / digest.upper()).write_bytes(content) + + with pytest.raises(canonical.ArtifactResolutionError): + DigestDirectoryResolver(tmp_path, "installed-store").resolve(reference)