From 741338f207bd0eb91733e56d873ac34160495f49 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 31 Oct 2026 14:32:37 +0100 Subject: [PATCH] feat(detection-core): build exact family catalog resolution --- .kiro/specs/detection-core/tasks.md | 2 +- src/openultrasast/core/family_catalog.py | 180 ++++++++++++++++ tests/test_core_family_catalog.py | 257 +++++++++++++++++++++++ 3 files changed, 438 insertions(+), 1 deletion(-) create mode 100644 src/openultrasast/core/family_catalog.py create mode 100644 tests/test_core_family_catalog.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 2183485..0c6881b 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -80,7 +80,7 @@ - Refuse undeclared fields, roles, decisions, reasons, or determinate outcomes without their minimum evidence. - Done when the family registration matrix accepts only declared combinations and reports unsupported scope without cross-family fallback. - _Requirements: 1.3, 2.1, 2.4, 3.1, 3.2, 3.3, 3.4, 4.4, 9.1_ -- [ ] 3.2 Build exact family catalog resolution and capability declarations +- [x] 3.2 Build exact family catalog resolution and capability declarations - Re-derive every contract and catalog digest and require canonical ordering and unique family/version/schema identities. - Resolve a contract only when family, digest, and ecosystem identify exactly one entry. - Validate that each family capability names its contract, readiness fixtures, development population, and admission boundary. diff --git a/src/openultrasast/core/family_catalog.py b/src/openultrasast/core/family_catalog.py new file mode 100644 index 0000000..b81d5ad --- /dev/null +++ b/src/openultrasast/core/family_catalog.py @@ -0,0 +1,180 @@ +"""Exact immutable family catalogs and capability declarations.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +from openultrasast.core import canonical +from openultrasast.core.family import FamilyContract, family_contract_digest, validate_contract + +FAMILY_CATALOG_SCHEMA = "openultrasast.detection-core.family-catalog.v2" +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +class FamilyCatalogError(ValueError): + """A family catalog violates its canonical closed-set contract.""" + + +class FamilyCatalogResolutionError(FamilyCatalogError, LookupError): + """No exact family, digest, and ecosystem catalog entry exists.""" + + +class FamilyCapabilityError(FamilyCatalogError): + """A family capability declaration is incomplete or substituted.""" + + +def _text(value: object, field: str, error: type[FamilyCatalogError] = FamilyCatalogError) -> None: + if not isinstance(value, str) or not value: + raise error(f"{field} must be nonempty text") + + +def _digest(value: object, field: str, error: type[FamilyCatalogError] = FamilyCatalogError) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise error(f"{field} must be a full lowercase SHA-256 digest") + + +@dataclass(frozen=True) +class FamilyCatalogEntry: + contract: FamilyContract + contract_digest: str + + def __post_init__(self) -> None: + if not isinstance(self.contract, FamilyContract): + raise FamilyCatalogError("entry contract must be a FamilyContract") + _digest(self.contract_digest, "contract digest") + validate_contract(self.contract) + if self.contract_digest != family_contract_digest(self.contract): + raise FamilyCatalogError("contract digest does not match contract content") + + +def catalog_entry_key(entry: FamilyCatalogEntry) -> tuple[str, str, str, str]: + if not isinstance(entry, FamilyCatalogEntry): + raise FamilyCatalogError("catalog entry must be a FamilyCatalogEntry") + contract = entry.contract + return (contract.family_id, contract.family_version, contract.schema_version, entry.contract_digest) + + +def _entry_value(entry: FamilyCatalogEntry) -> canonical.JsonValue: + contract = entry.contract + return { + "family_id": contract.family_id, + "family_version": contract.family_version, + "schema_version": contract.schema_version, + "ecosystems": contract.ecosystems, + "contract_digest": entry.contract_digest, + } + + +def derive_catalog_digest(entries: tuple[FamilyCatalogEntry, ...]) -> str: + if not isinstance(entries, tuple) or not entries: + raise FamilyCatalogError("entries must be a nonempty tuple") + if not all(isinstance(entry, FamilyCatalogEntry) for entry in entries): + raise FamilyCatalogError("entries must contain FamilyCatalogEntry values") + return canonical.content_digest({"schema_version": FAMILY_CATALOG_SCHEMA, "entries": tuple(_entry_value(entry) for entry in entries)}) + + +@dataclass(frozen=True) +class FamilyCatalog: + entries: tuple[FamilyCatalogEntry, ...] + catalog_digest: str + + def __post_init__(self) -> None: + validate_catalog(self) + + +def _validate_unique_entries(entries: tuple[FamilyCatalogEntry, ...]) -> None: + identities = tuple((item.contract.family_id, item.contract.family_version, item.contract.schema_version) for item in entries) + if len(set(identities)) != len(identities): + raise FamilyCatalogError("duplicate family contract identity") + digests = tuple(item.contract_digest for item in entries) + if len(set(digests)) != len(digests): + raise FamilyCatalogError("duplicate contract digest") + + +def validate_catalog(value: FamilyCatalog) -> None: + if not isinstance(value, FamilyCatalog): + raise FamilyCatalogError("catalog must be a FamilyCatalog") + if not isinstance(value.entries, tuple) or not value.entries: + raise FamilyCatalogError("entries must be a nonempty tuple") + if not all(isinstance(entry, FamilyCatalogEntry) for entry in value.entries): + raise FamilyCatalogError("entries must contain FamilyCatalogEntry values") + _digest(value.catalog_digest, "catalog digest") + if value.entries != tuple(sorted(value.entries, key=catalog_entry_key)): + raise FamilyCatalogError("catalog entries must be in canonical order") + _validate_unique_entries(value.entries) + for entry in value.entries: + validate_contract(entry.contract) + if entry.contract_digest != family_contract_digest(entry.contract): + raise FamilyCatalogError("contract digest does not match contract content") + if value.catalog_digest != derive_catalog_digest(value.entries): + raise FamilyCatalogError("catalog digest does not match canonical entries") + + +def resolve_contract( + value: FamilyCatalog, + family_id: str, + contract_digest: str, + ecosystem: str, +) -> FamilyContract: + validate_catalog(value) + _text(family_id, "family_id", FamilyCatalogResolutionError) + _digest(contract_digest, "contract digest", FamilyCatalogResolutionError) + _text(ecosystem, "ecosystem", FamilyCatalogResolutionError) + matches = tuple( + entry.contract + for entry in value.entries + if entry.contract.family_id == family_id and entry.contract_digest == contract_digest and ecosystem in entry.contract.ecosystems + ) + if len(matches) != 1: + raise FamilyCatalogResolutionError("family, digest, and ecosystem must resolve one exact contract") + return matches[0] + + +@dataclass(frozen=True) +class FamilyCapabilityDeclaration: + arm_id: str + family_contract_digest: str + readiness_fixture_digest: str + development_population_digest: str + admission_boundary_digest: str + + def __post_init__(self) -> None: + _validate_capability_fields(self) + + +def _validate_capability_fields(declaration: FamilyCapabilityDeclaration) -> None: + _text(declaration.arm_id, "arm_id", FamilyCapabilityError) + for value, field in ( + (declaration.family_contract_digest, "family contract digest"), + (declaration.readiness_fixture_digest, "readiness fixture digest"), + (declaration.development_population_digest, "development population digest"), + (declaration.admission_boundary_digest, "admission boundary digest"), + ): + _digest(value, field, FamilyCapabilityError) + + +def validate_capability(declaration: FamilyCapabilityDeclaration, value: FamilyCatalog) -> None: + if not isinstance(declaration, FamilyCapabilityDeclaration): + raise FamilyCapabilityError("declaration must be a FamilyCapabilityDeclaration") + _validate_capability_fields(declaration) + validate_catalog(value) + matches = tuple(entry for entry in value.entries if entry.contract_digest == declaration.family_contract_digest) + if len(matches) != 1: + raise FamilyCapabilityError("capability contract digest must identify one exact catalog entry") + + +__all__ = [ + "FAMILY_CATALOG_SCHEMA", + "FamilyCapabilityDeclaration", + "FamilyCapabilityError", + "FamilyCatalog", + "FamilyCatalogEntry", + "FamilyCatalogError", + "FamilyCatalogResolutionError", + "catalog_entry_key", + "derive_catalog_digest", + "resolve_contract", + "validate_capability", + "validate_catalog", +] diff --git a/tests/test_core_family_catalog.py b/tests/test_core_family_catalog.py new file mode 100644 index 0000000..7e51798 --- /dev/null +++ b/tests/test_core_family_catalog.py @@ -0,0 +1,257 @@ +"""Exact family-catalog resolution and capability declaration contracts.""" + +from dataclasses import replace +from typing import cast + +import pytest + +import openultrasast.core.canonical as canonical +import openultrasast.core.evidence as evidence +import openultrasast.core.family as family +import openultrasast.core.family_catalog as catalog +import openultrasast.core.semantics as semantics + + +def _digest(value: str) -> str: + return canonical.content_digest(value) + + +def _contract( + family_id: str = "obligation", + family_version: str = "1.0.0", + schema_version: str = evidence.EVIDENCE_PACKET_SCHEMA, + ecosystems: tuple[str, ...] = ("python",), +) -> family.FamilyContract: + role = family.RoleSpec( + "operation", + (family.FieldSpec("operation", semantics.FieldKind.TEXT, True),), + (semantics.RelationKind.OPERATION_EFFECT,), + family.RelationOriginPolicy.PRODUCT_OR_REVIEWED_FACT, + ) + return family.FamilyContract( + family_id, + family_version, + schema_version, + ecosystems, + family.ObjectSpec( + "operation", + (family.FieldSpec("operation", semantics.FieldKind.TEXT, True),), + ("operation",), + ), + (role,), + ( + family.DecisionRule( + "unsafe", + evidence.NeutralOutcome.VIOLATION, + ("operation",), + (semantics.RelationStatus.COMPLETE,), + ), + family.DecisionRule( + "unknown", + evidence.NeutralOutcome.INDETERMINATE, + ("operation",), + ( + semantics.RelationStatus.COMPLETE, + semantics.RelationStatus.INCOMPLETE, + semantics.RelationStatus.UNRESOLVED, + ), + ), + ), + (family.ResidualQuestion("operation-known", ("operation",), ("unsafe", "unknown")),), + ("missing_operation", "unsupported_scope"), + ) + + +def _entry(contract: family.FamilyContract) -> catalog.FamilyCatalogEntry: + return catalog.FamilyCatalogEntry(contract, family.family_contract_digest(contract)) + + +def _catalog(*contracts: family.FamilyContract) -> catalog.FamilyCatalog: + entries = tuple(sorted((_entry(contract) for contract in contracts), key=catalog.catalog_entry_key)) + return catalog.FamilyCatalog(entries, catalog.derive_catalog_digest(entries)) + + +def _capability(contract: family.FamilyContract) -> catalog.FamilyCapabilityDeclaration: + return catalog.FamilyCapabilityDeclaration( + "obligation-arm", + family.family_contract_digest(contract), + _digest("readiness fixtures"), + _digest("development population"), + _digest("admission boundary"), + ) + + +def test_catalog_rederives_contract_and_catalog_digests() -> None: + contract = _contract() + value = _catalog(contract) + catalog.validate_catalog(value) + assert value.entries[0].contract_digest == family.family_contract_digest(contract) + assert value.catalog_digest == catalog.derive_catalog_digest(value.entries) + + +def test_catalog_requires_canonical_entry_order() -> None: + first = _contract("alpha") + second = _contract("zeta") + entries = (_entry(second), _entry(first)) + with pytest.raises(catalog.FamilyCatalogError, match="canonical order"): + catalog.FamilyCatalog(entries, catalog.derive_catalog_digest(tuple(reversed(entries)))) + + +def test_catalog_refuses_duplicate_contract_identity_and_digest() -> None: + contract = _contract() + entry = _entry(contract) + with pytest.raises(catalog.FamilyCatalogError, match="duplicate family contract identity"): + catalog.FamilyCatalog((entry, entry), _digest("catalog")) + duplicate_digest = _entry(_contract("other")) + object.__setattr__(duplicate_digest, "contract_digest", entry.contract_digest) + entries = tuple(sorted((entry, duplicate_digest), key=catalog.catalog_entry_key)) + with pytest.raises(catalog.FamilyCatalogError, match="duplicate contract digest"): + catalog.FamilyCatalog(entries, _digest("catalog")) + + +def test_catalog_refuses_substituted_contract_and_catalog_digest() -> None: + contract = _contract() + with pytest.raises(catalog.FamilyCatalogError, match="contract digest"): + catalog.FamilyCatalogEntry(contract, _digest("substituted")) + value = _catalog(contract) + with pytest.raises(catalog.FamilyCatalogError, match="catalog digest"): + catalog.FamilyCatalog(value.entries, _digest("substituted catalog")) + + +def test_catalog_identity_includes_family_version_and_schema() -> None: + contracts = ( + _contract(family_version="1.0.0"), + _contract(family_version="2.0.0"), + _contract(family_version="3.0.0", schema_version="evidence-v3"), + ) + value = _catalog(*contracts) + assert len(value.entries) == 3 + catalog.validate_catalog(value) + + +def test_catalog_collections_are_immutable_tuples() -> None: + value = _catalog(_contract()) + with pytest.raises(catalog.FamilyCatalogError, match="entries"): + replace(value, entries=cast(tuple[catalog.FamilyCatalogEntry, ...], list(value.entries))) + + +def test_resolve_contract_requires_exact_family_digest_and_ecosystem() -> None: + python = _contract("obligation", ecosystems=("python",)) + java = _contract("taint", ecosystems=("java",)) + value = _catalog(python, java) + assert catalog.resolve_contract(value, "obligation", family.family_contract_digest(python), "python") is python + for family_id, digest, ecosystem in ( + ("missing", family.family_contract_digest(python), "python"), + ("obligation", family.family_contract_digest(java), "python"), + ("obligation", family.family_contract_digest(python), "java"), + ): + with pytest.raises(catalog.FamilyCatalogResolutionError, match="exact"): + catalog.resolve_contract(value, family_id, digest, ecosystem) + + +def test_resolution_never_falls_back_to_another_version() -> None: + first = _contract(family_version="1.0.0") + second = _contract(family_version="2.0.0") + value = _catalog(first, second) + assert ( + catalog.resolve_contract( + value, + first.family_id, + family.family_contract_digest(second), + "python", + ) + is second + ) + with pytest.raises(catalog.FamilyCatalogResolutionError, match="exact"): + catalog.resolve_contract(value, first.family_id, _digest("unknown version"), "python") + + +def test_capability_names_exact_catalog_contract_and_boundaries() -> None: + contract = _contract() + value = _catalog(contract) + capability = _capability(contract) + catalog.validate_capability(capability, value) + assert capability.family_contract_digest == value.entries[0].contract_digest + + +@pytest.mark.parametrize( + "field", + ( + "family_contract_digest", + "readiness_fixture_digest", + "development_population_digest", + "admission_boundary_digest", + ), +) +@pytest.mark.parametrize("bad", ("", "not-a-digest", "A" * 64, None)) +def test_capability_refuses_every_missing_or_corrupt_digest(field: str, bad: object) -> None: + capability = _capability(_contract()) + with pytest.raises((catalog.FamilyCapabilityError, canonical.DigestFormatError), match="digest"): + _replace_capability_digest(capability, field, cast(str, bad)) + + +def _replace_capability_digest( + capability: catalog.FamilyCapabilityDeclaration, + field: str, + value: str, +) -> catalog.FamilyCapabilityDeclaration: + if field == "family_contract_digest": + return replace(capability, family_contract_digest=value) + if field == "readiness_fixture_digest": + return replace(capability, readiness_fixture_digest=value) + if field == "development_population_digest": + return replace(capability, development_population_digest=value) + return replace(capability, admission_boundary_digest=value) + + +@pytest.mark.parametrize("bad", ("", 7, None)) +def test_capability_requires_nonempty_text_arm_id(bad: object) -> None: + capability = _capability(_contract()) + with pytest.raises(catalog.FamilyCapabilityError, match="arm_id"): + replace(capability, arm_id=cast(str, bad)) + + +def test_capability_refuses_contract_substitution() -> None: + contract = _contract() + other = _contract("other") + value = _catalog(contract, other) + capability = _capability(contract) + catalog.validate_capability(capability, value) + substituted = replace(capability, family_contract_digest=_digest("not in catalog")) + with pytest.raises(catalog.FamilyCapabilityError, match="catalog"): + catalog.validate_capability(substituted, value) + + +@pytest.mark.parametrize("bad", ("", 7, None)) +def test_capability_validator_rechecks_tampered_arm_id(bad: object) -> None: + contract = _contract() + value = _catalog(contract) + capability = _capability(contract) + object.__setattr__(capability, "arm_id", bad) + with pytest.raises(catalog.FamilyCapabilityError, match="arm_id"): + catalog.validate_capability(capability, value) + + +@pytest.mark.parametrize( + "field", + ( + "family_contract_digest", + "readiness_fixture_digest", + "development_population_digest", + "admission_boundary_digest", + ), +) +@pytest.mark.parametrize("bad", ("", 7, None, "not-a-digest", "A" * 64)) +def test_capability_validator_rechecks_every_tampered_digest(field: str, bad: object) -> None: + contract = _contract() + value = _catalog(contract) + capability = _capability(contract) + object.__setattr__(capability, field, bad) + with pytest.raises(catalog.FamilyCapabilityError, match="digest"): + catalog.validate_capability(capability, value) + + +def test_catalog_digest_covers_contract_identity_and_content() -> None: + contract = _contract() + changed = replace(contract, indeterminate_reasons=contract.indeterminate_reasons + ("new_reason",)) + assert _catalog(contract).catalog_digest != _catalog(changed).catalog_digest