From 18dc25b71865fde7118ff8a625a8251453df2f03 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sun, 1 Nov 2026 15:09:28 +0100 Subject: [PATCH] feat(access-control-prover): authenticate fact authority --- .kiro/specs/access-control-prover/tasks.md | 2 +- src/openultrasast/mapping/fact_authority.py | 369 +++++++++++++++++++ tests/test_fact_authority.py | 373 ++++++++++++++++++++ 3 files changed, 743 insertions(+), 1 deletion(-) create mode 100644 src/openultrasast/mapping/fact_authority.py create mode 100644 tests/test_fact_authority.py diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index 2ebaeb1..84bb160 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -338,7 +338,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with - _Requirements: 13.1_ - _Boundary: Access-Control Family Contract_ - _Depends: 11.5_ -- [ ] 14.2 Implement externally bootstrapped fact authority +- [x] 14.2 Implement externally bootstrapped fact authority - Define the immutable authority bundle and separate host bootstrap policy over reviewed pack/projection references, policies, lineages, matcher identity, reviewer/store admission, forbidden identity pairs and conflict disclosure. - Re-derive canonical policy, attestation, disclosure and bundle identities before any target data is consumed; never let the bundle choose its own trust roots. - Done when a complete independently reviewed authority fixture authenticates and every missing, duplicate, substituted, self-authored or non-canonical variant fails before pack loading. diff --git a/src/openultrasast/mapping/fact_authority.py b/src/openultrasast/mapping/fact_authority.py new file mode 100644 index 0000000..9528a00 --- /dev/null +++ b/src/openultrasast/mapping/fact_authority.py @@ -0,0 +1,369 @@ +"""Externally bootstrapped authority for access-control entry facts.""" + +from __future__ import annotations + +import json +import re +from collections.abc import Mapping +from dataclasses import dataclass + +from openultrasast.core import canonical, family, provenance +from openultrasast.core.factpack import ( + FactPack, + IndependenceAttestation, + ReviewIndependencePolicy, + derive_fact_pack_content_digest, + derive_independence_attestation_digest, + derive_review_evidence_digest, + derive_review_independence_policy_digest, +) +from openultrasast.semantic.obligations.core_contract import validate_access_control_contract + +ENTRY_FACT_AUTHORITY_SCHEMA = "openultrasast.access-control.entry-fact-authority.v2" +ENTRY_AUTHORITY_BOOTSTRAP_SCHEMA = "openultrasast.access-control.entry-authority-bootstrap.v2" +AUTHORITY_CONFLICT_DISCLOSURE_SCHEMA = "openultrasast.access-control.authority-conflict-disclosure.v1" +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +class EntryFactAuthorityError(ValueError): + """Installed entry-fact authority cannot be authenticated externally.""" + + +def _require_text(value: object, field: str) -> None: + if not isinstance(value, str) or not value: + raise EntryFactAuthorityError(f"{field} must be nonempty text") + + +def _require_digest(value: object, field: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise EntryFactAuthorityError(f"{field} must be a full lowercase SHA-256 digest") + + +def _require_digest_tuple(values: object, field: str) -> None: + if not isinstance(values, tuple) or not values: + raise EntryFactAuthorityError(f"{field} must be a nonempty tuple") + for value in values: + _require_digest(value, field) + if len(set(values)) != len(values): + raise EntryFactAuthorityError(f"duplicate {field}") + + +def _require_forbidden_pairs(values: object) -> None: + if not isinstance(values, tuple): + raise EntryFactAuthorityError("forbidden identity pairs must be a tuple") + for pair in values: + if not isinstance(pair, tuple) or len(pair) != 2: + raise EntryFactAuthorityError("forbidden identity pairs must contain pairs") + _require_digest(pair[0], "forbidden producer") + _require_digest(pair[1], "forbidden reviewer") + if len(set(values)) != len(values): + raise EntryFactAuthorityError("duplicate forbidden identity pair") + + +def _require_store_identities(values: object) -> None: + if not isinstance(values, tuple) or not values: + raise EntryFactAuthorityError("trusted store identities must be a nonempty tuple") + for value in values: + _require_text(value, "trusted store identity") + if len(set(values)) != len(values): + raise EntryFactAuthorityError("duplicate trusted store identity") + + +@dataclass(frozen=True) +class EntryAuthorityBootstrapPolicy: + reviewed_bundle_reviewer_digests: tuple[str, ...] + forbidden_identity_pairs: tuple[tuple[str, str], ...] + trusted_store_identities: tuple[str, ...] + maximum_bundle_bytes: int + policy_digest: str + + def __post_init__(self) -> None: + _require_digest_tuple(self.reviewed_bundle_reviewer_digests, "reviewed bundle reviewer") + _require_forbidden_pairs(self.forbidden_identity_pairs) + _require_store_identities(self.trusted_store_identities) + if isinstance(self.maximum_bundle_bytes, bool) or not isinstance(self.maximum_bundle_bytes, int): + raise EntryFactAuthorityError("maximum bundle bytes must be a positive integer") + if self.maximum_bundle_bytes <= 0: + raise EntryFactAuthorityError("maximum bundle bytes must be a positive integer") + _require_digest(self.policy_digest, "bootstrap policy digest") + + +@dataclass(frozen=True) +class EntryFactAuthorityBundle: + contract: family.FamilyContract + packs: tuple[FactPack, ...] + recognition_projection: canonical.ContentRef + lineages: tuple[provenance.LineageManifest, ...] + custody_policy: provenance.PopulationCustodyPolicy + review_policy: ReviewIndependencePolicy + matcher_revision: str + matcher_behavior_digest: str + matcher_lineage_manifest_digest: str + authority_review: IndependenceAttestation + bundle_digest: str + + def __post_init__(self) -> None: + if not isinstance(self.contract, family.FamilyContract): + raise EntryFactAuthorityError("contract must be a FamilyContract") + if not isinstance(self.packs, tuple) or not all(isinstance(pack, FactPack) for pack in self.packs): + raise EntryFactAuthorityError("packs must be a tuple of FactPack values") + if not isinstance(self.recognition_projection, canonical.ContentRef): + raise EntryFactAuthorityError("recognition projection must be a ContentRef") + if not isinstance(self.lineages, tuple) or not all(isinstance(item, provenance.LineageManifest) for item in self.lineages): + raise EntryFactAuthorityError("lineages must be a tuple of LineageManifest values") + if not isinstance(self.custody_policy, provenance.PopulationCustodyPolicy): + raise EntryFactAuthorityError("custody policy must be a PopulationCustodyPolicy") + if not isinstance(self.review_policy, ReviewIndependencePolicy): + raise EntryFactAuthorityError("review policy must be a ReviewIndependencePolicy") + _require_text(self.matcher_revision, "matcher revision") + _require_digest(self.matcher_behavior_digest, "matcher behavior digest") + _require_digest(self.matcher_lineage_manifest_digest, "matcher lineage manifest digest") + if not isinstance(self.authority_review, IndependenceAttestation): + raise EntryFactAuthorityError("authority review must be an IndependenceAttestation") + _require_digest(self.bundle_digest, "bundle digest") + + +@dataclass(frozen=True) +class EntryFactAuthorityContext: + bundle: EntryFactAuthorityBundle + custody: provenance.PopulationCustodyPort + resolver: canonical.ContentResolver + + +def derive_entry_authority_bootstrap_policy_digest( + reviewed_bundle_reviewer_digests: tuple[str, ...], + forbidden_identity_pairs: tuple[tuple[str, str], ...], + trusted_store_identities: tuple[str, ...], + maximum_bundle_bytes: int, +) -> str: + """Derive the external bootstrap identity without consulting a bundle.""" + return canonical.content_digest( + { + "schema_version": ENTRY_AUTHORITY_BOOTSTRAP_SCHEMA, + "reviewed_bundle_reviewer_digests": tuple(sorted(reviewed_bundle_reviewer_digests)), + "forbidden_identity_pairs": tuple(sorted(forbidden_identity_pairs)), + "trusted_store_identities": tuple(sorted(trusted_store_identities)), + "maximum_bundle_bytes": maximum_bundle_bytes, + } + ) + + +def _reference_value(reference: canonical.ContentRef) -> canonical.JsonValue: + return { + "artifact_id": reference.artifact_id, + "content_digest": reference.content_digest, + "media_type": reference.media_type, + "byte_length": reference.byte_length, + "retention_policy_digest": reference.retention_policy_digest, + "storage_receipt_digest": reference.storage_receipt_digest, + } + + +def _pack_identity(pack: FactPack) -> Mapping[str, canonical.JsonValue]: + return { + "family_contract_digest": pack.family_contract_digest, + "family_id": pack.family_id, + "ecosystem": pack.ecosystem, + "language": pack.language, + "normalized_content_digest": pack.normalized_content_digest, + "review_evidence_digest": pack.review_evidence_digest, + "review_policy_digest": pack.review_policy_digest, + } + + +def derive_entry_fact_authority_bundle_digest(bundle: EntryFactAuthorityBundle) -> str: + """Derive the authority review subject, excluding the review and digest fields.""" + ordered_packs = sorted(bundle.packs, key=lambda pack: (pack.ecosystem, pack.language, pack.normalized_content_digest)) + pack_identities = tuple(_pack_identity(pack) for pack in ordered_packs) + return canonical.content_digest( + { + "schema_version": ENTRY_FACT_AUTHORITY_SCHEMA, + "family_contract_digest": family.family_contract_digest(bundle.contract), + "packs": pack_identities, + "recognition_projection": _reference_value(bundle.recognition_projection), + "lineage_manifest_digests": tuple(sorted(item.manifest_digest for item in bundle.lineages)), + "custody_policy_digest": bundle.custody_policy.policy_digest, + "review_policy_digest": bundle.review_policy.policy_digest, + "matcher_revision": bundle.matcher_revision, + "matcher_behavior_digest": bundle.matcher_behavior_digest, + "matcher_lineage_manifest_digest": bundle.matcher_lineage_manifest_digest, + } + ) + + +def _validate_bootstrap(policy: EntryAuthorityBootstrapPolicy) -> None: + policy.__post_init__() + expected = derive_entry_authority_bootstrap_policy_digest( + policy.reviewed_bundle_reviewer_digests, + policy.forbidden_identity_pairs, + policy.trusted_store_identities, + policy.maximum_bundle_bytes, + ) + if policy.policy_digest != expected: + raise EntryFactAuthorityError("bootstrap policy digest cannot be re-derived") + + +def _validate_ports(custody: object, resolver: object) -> None: + if not callable(getattr(custody, "current_log", None)): + raise EntryFactAuthorityError("population custody port is missing") + if not callable(getattr(resolver, "resolve", None)): + raise EntryFactAuthorityError("content resolver is missing") + + +def _validate_policies(bundle: EntryFactAuthorityBundle) -> None: + custody_expected = provenance.derive_population_custody_policy_digest(bundle.custody_policy.reviewed_custodian_digests) + if bundle.custody_policy.policy_digest != custody_expected: + raise EntryFactAuthorityError("custody policy digest cannot be re-derived") + review_expected = derive_review_independence_policy_digest( + bundle.review_policy.reviewed_reviewer_digests, + bundle.review_policy.forbidden_identity_pairs, + bundle.review_policy.trusted_store_identities, + bundle.review_policy.maximum_evidence_bytes, + ) + if bundle.review_policy.policy_digest != review_expected: + raise EntryFactAuthorityError("review policy digest cannot be re-derived") + + +def _validate_lineages(bundle: EntryFactAuthorityBundle) -> dict[str, provenance.LineageManifest]: + indexed: dict[str, provenance.LineageManifest] = {} + for lineage in bundle.lineages: + lineage.__post_init__() + expected = provenance.derive_lineage_manifest_digest(lineage.arm_id, lineage.subject_digest, lineage.uses) + if lineage.manifest_digest != expected: + raise EntryFactAuthorityError("lineage manifest digest cannot be re-derived") + if lineage.manifest_digest in indexed: + raise EntryFactAuthorityError("duplicate lineage manifest identity") + indexed[lineage.manifest_digest] = lineage + matcher = indexed.get(bundle.matcher_lineage_manifest_digest) + if matcher is None or matcher.subject_digest != bundle.matcher_behavior_digest: + raise EntryFactAuthorityError("matcher lineage is missing or names another behavior") + if not any(use.purpose is provenance.LineagePurpose.MATCHING_BEHAVIOR for use in matcher.uses): + raise EntryFactAuthorityError("matcher lineage lacks matching-behavior provenance") + return indexed + + +def _validate_packs(bundle: EntryFactAuthorityBundle, lineages: dict[str, provenance.LineageManifest]) -> None: + if not bundle.packs: + raise EntryFactAuthorityError("authority bundle requires reviewed packs") + identities = tuple((pack.ecosystem, pack.language) for pack in bundle.packs) + if len(set(identities)) != len(identities): + raise EntryFactAuthorityError("duplicate language-scoped pack identity") + contract_digest = family.family_contract_digest(bundle.contract) + for pack in bundle.packs: + pack.__post_init__() + if pack.family_contract_digest != contract_digest or pack.family_id != bundle.contract.family_id: + raise EntryFactAuthorityError("pack does not name the production family contract") + if pack.ecosystem not in bundle.contract.ecosystems: + raise EntryFactAuthorityError("pack ecosystem is outside the production contract") + if pack.review_policy_digest != bundle.review_policy.policy_digest: + raise EntryFactAuthorityError("pack review policy differs from authority") + if pack.normalized_content_digest != derive_fact_pack_content_digest(pack): + raise EntryFactAuthorityError("pack content digest cannot be re-derived") + if pack.review_evidence_digest != derive_review_evidence_digest(pack.review_evidence): + raise EntryFactAuthorityError("pack review evidence digest cannot be re-derived") + if not pack.review_evidence or pack.review_identity not in bundle.review_policy.reviewed_reviewer_digests: + raise EntryFactAuthorityError("pack lacks an independently reviewed identity") + _validate_pack_lineages(pack, lineages) + + +def _validate_pack_lineages(pack: FactPack, lineages: dict[str, provenance.LineageManifest]) -> None: + for evidence in pack.review_evidence: + lineage = lineages.get(evidence.lineage_manifest_digest) + if lineage is None or lineage.subject_digest != pack.normalized_content_digest: + raise EntryFactAuthorityError("semantic-fact lineage is missing or names another pack") + if not any(use.purpose is provenance.LineagePurpose.SEMANTIC_FACT for use in lineage.uses): + raise EntryFactAuthorityError("pack lineage lacks semantic-fact provenance") + + +def _validate_review(bundle: EntryFactAuthorityBundle, bootstrap: EntryAuthorityBootstrapPolicy) -> None: + review = bundle.authority_review + review.__post_init__() + if bundle.bundle_digest != derive_entry_fact_authority_bundle_digest(bundle): + raise EntryFactAuthorityError("authority bundle digest cannot be re-derived") + if review.subject_digest != bundle.bundle_digest: + raise EntryFactAuthorityError("authority review subject does not name the bundle") + if review.attestation_digest != derive_independence_attestation_digest(review): + raise EntryFactAuthorityError("authority attestation digest cannot be re-derived") + if review.producer_identity == review.reviewer_identity: + raise EntryFactAuthorityError("authority producer and reviewer must be independent") + if review.reviewer_identity not in bootstrap.reviewed_bundle_reviewer_digests: + raise EntryFactAuthorityError("authority reviewer is not trusted by bootstrap") + if (review.producer_identity, review.reviewer_identity) in bootstrap.forbidden_identity_pairs: + raise EntryFactAuthorityError("authority producer/reviewer pair is forbidden") + + +def _resolve(reference: canonical.ContentRef, bootstrap: EntryAuthorityBootstrapPolicy, resolver: canonical.ContentResolver) -> bytes: + try: + return canonical.resolve_content( + reference, + resolver, + trusted_store_identities=bootstrap.trusted_store_identities, + maximum_bytes=bootstrap.maximum_bundle_bytes, + ) + except (canonical.ArtifactResolutionError, canonical.ArtifactValidationError, ValueError, TypeError) as exc: + raise EntryFactAuthorityError("authority artifact cannot be resolved under bootstrap") from exc + + +def _validate_disclosure(content: bytes) -> None: + try: + raw = json.loads(content) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise EntryFactAuthorityError("authority conflict disclosure is not canonical JSON") from exc + if not isinstance(raw, dict) or set(raw) != {"schema_version", "conflicts", "resolved"}: + raise EntryFactAuthorityError("authority conflict disclosure has unknown fields") + conflicts = raw["conflicts"] + if raw["schema_version"] != AUTHORITY_CONFLICT_DISCLOSURE_SCHEMA: + raise EntryFactAuthorityError("authority conflict disclosure schema is unknown") + if not isinstance(conflicts, list) or not all(isinstance(item, str) and item for item in conflicts): + raise EntryFactAuthorityError("authority conflicts must be a list of nonempty text") + if raw["resolved"] is not True: + raise EntryFactAuthorityError("authority conflict disclosure is unresolved") + canonical_value: Mapping[str, canonical.JsonValue] = { + "schema_version": AUTHORITY_CONFLICT_DISCLOSURE_SCHEMA, + "conflicts": tuple(conflicts), + "resolved": True, + } + expected = canonical.canonical_record_bytes( + canonical_value, known_schema_versions=(AUTHORITY_CONFLICT_DISCLOSURE_SCHEMA,) + ) + if content != expected: + raise EntryFactAuthorityError("authority conflict disclosure bytes are not canonical") + + +def authenticate_entry_fact_authority( + bundle: EntryFactAuthorityBundle, + bootstrap: EntryAuthorityBootstrapPolicy, + custody: provenance.PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> EntryFactAuthorityContext: + """Authenticate installed authority without interpreting packs or reading custody state.""" + if not isinstance(bundle, EntryFactAuthorityBundle) or not isinstance(bootstrap, EntryAuthorityBootstrapPolicy): + raise EntryFactAuthorityError("authority and bootstrap must use the typed contracts") + _validate_ports(custody, resolver) + bundle.__post_init__() + _validate_bootstrap(bootstrap) + _validate_review(bundle, bootstrap) + disclosure = _resolve(bundle.authority_review.conflict_disclosure, bootstrap, resolver) + _validate_disclosure(disclosure) + validate_access_control_contract(bundle.contract) + _validate_policies(bundle) + lineages = _validate_lineages(bundle) + _validate_packs(bundle, lineages) + projection = _resolve(bundle.recognition_projection, bootstrap, resolver) + if not projection: + raise EntryFactAuthorityError("recognition projection must be nonempty") + return EntryFactAuthorityContext(bundle, custody, resolver) + + +__all__ = [ + "AUTHORITY_CONFLICT_DISCLOSURE_SCHEMA", + "ENTRY_AUTHORITY_BOOTSTRAP_SCHEMA", + "ENTRY_FACT_AUTHORITY_SCHEMA", + "EntryAuthorityBootstrapPolicy", + "EntryFactAuthorityBundle", + "EntryFactAuthorityContext", + "EntryFactAuthorityError", + "authenticate_entry_fact_authority", + "derive_entry_authority_bootstrap_policy_digest", + "derive_entry_fact_authority_bundle_digest", +] diff --git a/tests/test_fact_authority.py b/tests/test_fact_authority.py new file mode 100644 index 0000000..4cc35f4 --- /dev/null +++ b/tests/test_fact_authority.py @@ -0,0 +1,373 @@ +"""Externally bootstrapped access-control fact-authority contracts.""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass, replace + +import pytest +from detection_core_fixtures import InMemoryContentResolver, stored_artifact + +from openultrasast.core import canonical, provenance, semantics +from openultrasast.core.factpack import ( + FACT_PACK_SCHEMA, + FactPack, + IndependenceAttestation, + KnowledgeEvidence, + KnowledgeEvidenceKind, + RelationEffect, + ReviewIndependencePolicy, + SemanticFact, + StructuralRequirement, + StructuralRequirementKind, + derive_fact_pack_content_digest, + derive_independence_attestation_digest, + derive_review_evidence_digest, + derive_review_independence_policy_digest, +) +from openultrasast.mapping import fact_authority as authority_module +from openultrasast.mapping.fact_authority import ( + AUTHORITY_CONFLICT_DISCLOSURE_SCHEMA, + EntryAuthorityBootstrapPolicy, + EntryFactAuthorityBundle, + EntryFactAuthorityError, + authenticate_entry_fact_authority, + derive_entry_authority_bootstrap_policy_digest, + derive_entry_fact_authority_bundle_digest, +) +from openultrasast.semantic.obligations.core_contract import production_access_control_contract + + +@dataclass(frozen=True) +class AuthorityCase: + bundle: EntryFactAuthorityBundle + bootstrap: EntryAuthorityBootstrapPolicy + resolver: InMemoryContentResolver + + +class UnusedCustody: + """Authority authentication must not consume evaluation state or target packs.""" + + def current_log(self, population_digest: str) -> provenance.PopulationCustodyLog: + raise AssertionError(f"custody read before fact validation: {population_digest}") + + +def _merge_resolvers(*resolvers: InMemoryContentResolver) -> InMemoryContentResolver: + artifacts: dict[str, canonical.ResolvedContent] = {} + for resolver in resolvers: + artifacts.update(resolver.artifacts) + return InMemoryContentResolver(artifacts) + + +def _lineage(arm: str, subject: str, purpose: provenance.LineagePurpose) -> provenance.LineageManifest: + population = canonical.content_digest(f"{arm}:population") + receipt = canonical.content_digest(f"{arm}:receipt") + use = provenance.LineageUse(purpose, population, False, subject, receipt) + digest = provenance.derive_lineage_manifest_digest(arm, subject, (use,)) + return provenance.LineageManifest(arm, subject, (use,), digest) + + +def _review_policy(reviewer: str) -> ReviewIndependencePolicy: + stores = ("authority-store",) + maximum = 16_384 + digest = derive_review_independence_policy_digest((reviewer,), (), stores, maximum) + return ReviewIndependencePolicy((reviewer,), (), stores, maximum, digest) + + +def _pack( + language: str, + reviewer: str, + policy: ReviewIndependencePolicy, + evidence_ref: canonical.ContentRef, + disclosure_ref: canonical.ContentRef, +) -> tuple[FactPack, provenance.LineageManifest]: + producer = canonical.content_digest(f"{language}:pack-producer") + provisional_attestation = IndependenceAttestation(evidence_ref.content_digest, producer, reviewer, disclosure_ref, "0" * 64) + attestation = replace( + provisional_attestation, + attestation_digest=derive_independence_attestation_digest(provisional_attestation), + ) + provisional_evidence = KnowledgeEvidence( + KnowledgeEvidenceKind.SOURCE_CODE, + evidence_ref, + producer, + reviewer, + attestation, + "0" * 64, + ) + contract = production_access_control_contract() + fact = SemanticFact( + f"entry:test:{language}", + "entry_registration", + (StructuralRequirement("qualified", StructuralRequirementKind.QUALIFIED_NAME, "qualified_target", "app.get"),), + ( + RelationEffect( + semantics.RelationKind.REGISTRATION, + "entry_registration", + "framework_instance", + "entry_handler", + (), + ), + ), + ) + provisional_pack = FactPack( + FACT_PACK_SCHEMA, + canonical.content_digest( + { + "family_id": contract.family_id, + "family_version": contract.family_version, + "schema_version": contract.schema_version, + } + ), + contract.family_id, + "express", + language, + (fact,), + "0" * 64, + reviewer, + (provisional_evidence,), + "0" * 64, + policy.policy_digest, + ) + # Use the public contract digest required by FactPack rather than the local summary above. + from openultrasast.core.family import family_contract_digest + + provisional_pack = replace(provisional_pack, family_contract_digest=family_contract_digest(contract)) + normalized = derive_fact_pack_content_digest(provisional_pack) + lineage = _lineage(f"{language}:facts", normalized, provenance.LineagePurpose.SEMANTIC_FACT) + evidence = replace(provisional_evidence, lineage_manifest_digest=lineage.manifest_digest) + pack = replace( + provisional_pack, + normalized_content_digest=normalized, + review_evidence=(evidence,), + review_evidence_digest=derive_review_evidence_digest((evidence,)), + ) + return pack, lineage + + +def _case() -> AuthorityCase: + producer = canonical.content_digest("authority-producer") + reviewer = canonical.content_digest("authority-reviewer") + projection, projection_resolver = stored_artifact( + b'{"schema_version":"projection.v1"}', + artifact_id="recognition-projection", + store_identity="authority-store", + ) + evidence, evidence_resolver = stored_artifact( + b'{"source":"reviewed"}', + artifact_id="pack-review-evidence", + store_identity="authority-store", + ) + pack_disclosure, pack_disclosure_resolver = stored_artifact( + b'{"conflicts":[]}', + artifact_id="pack-conflicts", + store_identity="authority-store", + ) + disclosure_bytes = canonical.canonical_record_bytes( + { + "schema_version": AUTHORITY_CONFLICT_DISCLOSURE_SCHEMA, + "conflicts": (), + "resolved": True, + }, + known_schema_versions=(AUTHORITY_CONFLICT_DISCLOSURE_SCHEMA,), + ) + disclosure, disclosure_resolver = stored_artifact( + disclosure_bytes, + artifact_id="authority-conflicts", + store_identity="authority-store", + ) + resolver = _merge_resolvers( + projection_resolver, + evidence_resolver, + pack_disclosure_resolver, + disclosure_resolver, + ) + review_policy = _review_policy(reviewer) + javascript, javascript_lineage = _pack("javascript", reviewer, review_policy, evidence, pack_disclosure) + typescript, typescript_lineage = _pack("typescript", reviewer, review_policy, evidence, pack_disclosure) + matcher_behavior = canonical.content_digest("matcher-behavior") + matcher_lineage = _lineage("entry-matcher", matcher_behavior, provenance.LineagePurpose.MATCHING_BEHAVIOR) + custodian = canonical.content_digest("authority-custodian") + custody_policy = provenance.PopulationCustodyPolicy( + (custodian,), + provenance.derive_population_custody_policy_digest((custodian,)), + ) + provisional_review = IndependenceAttestation("0" * 64, producer, reviewer, disclosure, "0" * 64) + provisional_bundle = EntryFactAuthorityBundle( + production_access_control_contract(), + (javascript, typescript), + projection, + (javascript_lineage, typescript_lineage, matcher_lineage), + custody_policy, + review_policy, + "entry-matcher-v1", + matcher_behavior, + matcher_lineage.manifest_digest, + provisional_review, + "0" * 64, + ) + bundle_digest = derive_entry_fact_authority_bundle_digest(provisional_bundle) + review = replace(provisional_review, subject_digest=bundle_digest) + review = replace(review, attestation_digest=derive_independence_attestation_digest(review)) + bundle = replace(provisional_bundle, authority_review=review, bundle_digest=bundle_digest) + forbidden = ((canonical.content_digest("forbidden-producer"), reviewer),) + maximum = 16_384 + bootstrap_digest = derive_entry_authority_bootstrap_policy_digest((reviewer,), forbidden, ("authority-store",), maximum) + bootstrap = EntryAuthorityBootstrapPolicy((reviewer,), forbidden, ("authority-store",), maximum, bootstrap_digest) + return AuthorityCase(bundle, bootstrap, resolver) + + +def _authenticate(case: AuthorityCase) -> None: + context = authenticate_entry_fact_authority(case.bundle, case.bootstrap, UnusedCustody(), case.resolver) + assert context.bundle is case.bundle + assert context.resolver is case.resolver + + +def _with_unresolved_disclosure(case: AuthorityCase) -> AuthorityCase: + unresolved_bytes = canonical.canonical_record_bytes( + { + "schema_version": AUTHORITY_CONFLICT_DISCLOSURE_SCHEMA, + "conflicts": (), + "resolved": False, + }, + known_schema_versions=(AUTHORITY_CONFLICT_DISCLOSURE_SCHEMA,), + ) + unresolved, unresolved_resolver = stored_artifact( + unresolved_bytes, + artifact_id="authority-conflicts", + store_identity="authority-store", + ) + review = replace(case.bundle.authority_review, conflict_disclosure=unresolved) + review = replace(review, attestation_digest=derive_independence_attestation_digest(review)) + artifacts: Mapping[str, canonical.ResolvedContent] = {**case.resolver.artifacts, **unresolved_resolver.artifacts} + return replace( + case, + bundle=replace(case.bundle, authority_review=review), + resolver=InMemoryContentResolver(artifacts), + ) + + +def _with_independently_reviewed_bundle(case: AuthorityCase, bundle: EntryFactAuthorityBundle) -> AuthorityCase: + bundle_digest = derive_entry_fact_authority_bundle_digest(bundle) + review = replace(bundle.authority_review, subject_digest=bundle_digest) + review = replace(review, attestation_digest=derive_independence_attestation_digest(review)) + return replace(case, bundle=replace(bundle, authority_review=review, bundle_digest=bundle_digest)) + + +def test_independently_reviewed_authority_authenticates_without_consuming_target_state() -> None: + _authenticate(_case()) + + +@pytest.mark.parametrize("missing", ("custody", "resolver")) +def test_authority_refuses_missing_injected_ports(missing: str) -> None: + case = _case() + custody: object = object() if missing == "custody" else UnusedCustody() + resolver: object = object() if missing == "resolver" else case.resolver + + with pytest.raises(EntryFactAuthorityError, match="missing"): + authenticate_entry_fact_authority(case.bundle, case.bootstrap, custody, resolver) # type: ignore[arg-type] + + +@pytest.mark.parametrize("missing", ("packs", "projection", "lineage")) +def test_authority_refuses_missing_prerequisites(missing: str) -> None: + case = _case() + if missing == "packs": + bundle = replace(case.bundle, packs=()) + case = _with_independently_reviewed_bundle(case, bundle) + elif missing == "projection": + case = replace(case, resolver=InMemoryContentResolver({})) + else: + bundle = replace(case.bundle, lineages=case.bundle.lineages[:-1]) + case = _with_independently_reviewed_bundle(case, bundle) + + with pytest.raises(EntryFactAuthorityError): + _authenticate(case) + + +@pytest.mark.parametrize("duplicate", ("pack", "lineage")) +def test_authority_refuses_duplicate_identities(duplicate: str) -> None: + case = _case() + if duplicate == "pack": + bundle = replace(case.bundle, packs=(case.bundle.packs[0], case.bundle.packs[0])) + else: + bundle = replace(case.bundle, lineages=(*case.bundle.lineages, case.bundle.lineages[0])) + case = _with_independently_reviewed_bundle(case, bundle) + + with pytest.raises(EntryFactAuthorityError, match="duplicate"): + _authenticate(case) + + +@pytest.mark.parametrize("substitution", ("bundle", "projection", "matcher", "attestation")) +def test_authority_refuses_substituted_identity(substitution: str) -> None: + case = _case() + if substitution == "bundle": + bundle = replace(case.bundle, bundle_digest="f" * 64) + elif substitution == "projection": + bundle = replace(case.bundle, recognition_projection=replace(case.bundle.recognition_projection, content_digest="f" * 64)) + elif substitution == "matcher": + bundle = replace(case.bundle, matcher_behavior_digest="f" * 64) + else: + bundle = replace(case.bundle, authority_review=replace(case.bundle.authority_review, attestation_digest="f" * 64)) + + with pytest.raises(EntryFactAuthorityError): + _authenticate(replace(case, bundle=bundle)) + + +@pytest.mark.parametrize("self_authored", ("same-identity", "forbidden-pair", "untrusted-reviewer")) +def test_authority_refuses_self_authored_or_untrusted_review(self_authored: str) -> None: + case = _case() + review = case.bundle.authority_review + if self_authored == "same-identity": + review = replace(review, producer_identity=review.reviewer_identity) + elif self_authored == "forbidden-pair": + review = replace(review, producer_identity=case.bootstrap.forbidden_identity_pairs[0][0]) + else: + review = replace(review, reviewer_identity=canonical.content_digest("untrusted-reviewer")) + review = replace(review, attestation_digest=derive_independence_attestation_digest(review)) + + with pytest.raises(EntryFactAuthorityError): + _authenticate(replace(case, bundle=replace(case.bundle, authority_review=review))) + + +@pytest.mark.parametrize("invalid_authority", ("untrusted-reviewer", "invalid-attestation", "unresolved-disclosure")) +def test_external_authority_failure_stops_before_pack_validation( + invalid_authority: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + case = _case() + if invalid_authority == "untrusted-reviewer": + review = replace(case.bundle.authority_review, reviewer_identity=canonical.content_digest("untrusted-reviewer")) + review = replace(review, attestation_digest=derive_independence_attestation_digest(review)) + case = replace(case, bundle=replace(case.bundle, authority_review=review)) + elif invalid_authority == "invalid-attestation": + review = replace(case.bundle.authority_review, attestation_digest="f" * 64) + case = replace(case, bundle=replace(case.bundle, authority_review=review)) + else: + case = _with_unresolved_disclosure(case) + reached_pack_validation = False + + def fail_if_reached(*args: object, **kwargs: object) -> None: + nonlocal reached_pack_validation + reached_pack_validation = True + raise AssertionError("pack validation ran before external authority was established") + + monkeypatch.setattr(authority_module, "_validate_packs", fail_if_reached) + + with pytest.raises(EntryFactAuthorityError): + _authenticate(case) + assert not reached_pack_validation + + +@pytest.mark.parametrize("noncanonical", ("bootstrap", "bundle", "disclosure")) +def test_authority_refuses_noncanonical_policy_bundle_or_disclosure(noncanonical: str) -> None: + case = _case() + if noncanonical == "bootstrap": + case = replace(case, bootstrap=replace(case.bootstrap, policy_digest="f" * 64)) + elif noncanonical == "bundle": + case = replace(case, bundle=replace(case.bundle, matcher_revision="substituted")) + else: + case = _with_unresolved_disclosure(case) + + message = "unresolved" if noncanonical == "disclosure" else None + with pytest.raises(EntryFactAuthorityError, match=message): + _authenticate(case)