From 08b0eeaac28b4890c609dc5ebaddb02abbe01e99 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 31 Oct 2026 15:28:14 +0100 Subject: [PATCH] feat(detection-core): validate reviewed fact packs --- .kiro/specs/detection-core/tasks.md | 2 +- src/openultrasast/core/__init__.py | 3 +- src/openultrasast/core/_factpack_digest.py | 138 ++++++ src/openultrasast/core/_legacy_factpack.py | 59 +++ src/openultrasast/core/factpack.py | 491 ++++++++++++++++++--- tests/test_core_factpack.py | 406 +++++++++++++++++ 6 files changed, 1042 insertions(+), 57 deletions(-) create mode 100644 src/openultrasast/core/_factpack_digest.py create mode 100644 src/openultrasast/core/_legacy_factpack.py create mode 100644 tests/test_core_factpack.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 0103e77..cd50114 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -125,7 +125,7 @@ - _Requirements: 8.3, 8.4, 8.5, 15.5_ - [ ] 5. Replace semantic extension points with reviewed fact packs -- [ ] 5.1 Validate versioned packs and independent review evidence +- [x] 5.1 Validate versioned packs and independent review evidence - Bind schema, family, ecosystem, language, stable facts, declared roles, normalized content, and review evidence. - Accept source, official-documentation, or independently reviewed behavioral evidence under a trusted review policy. - Resolve conflict disclosures and require producer/reviewer independence where behavioral evidence is used. diff --git a/src/openultrasast/core/__init__.py b/src/openultrasast/core/__init__.py index 5dc400a..f915df5 100644 --- a/src/openultrasast/core/__init__.py +++ b/src/openultrasast/core/__init__.py @@ -4,6 +4,7 @@ Owns contracts only -- no analysis logic. Arms implement the trace contract; runners and consumers adopt coverage, budget, provenance, and admission contracts. """ +from openultrasast.core._legacy_factpack import LegacyFactPack as FactPack from openultrasast.core.admission import ( REPORT_SCHEMA_KEYS, AdmissionManifest, @@ -14,7 +15,7 @@ from openultrasast.core.admission import ( from openultrasast.core.budget import Budget, BudgetGateFailure, Complexity, check_budget, derive_budget from openultrasast.core.consumer import PushRequest, PushVerdict, consume_verdict from openultrasast.core.coverage import Coverage, InstrumentFailure, ResolutionOutcome, check_reads -from openultrasast.core.factpack import FactPack, FactPackError, admit_fact_pack, verify_fact_pack +from openultrasast.core.factpack import FactPackError, admit_fact_pack, verify_fact_pack from openultrasast.core.probes import ( FittedComponentBlocked, ProbeResult, diff --git a/src/openultrasast/core/_factpack_digest.py b/src/openultrasast/core/_factpack_digest.py new file mode 100644 index 0000000..3345ff6 --- /dev/null +++ b/src/openultrasast/core/_factpack_digest.py @@ -0,0 +1,138 @@ +"""Canonical value projections for v2 fact-pack identities.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from openultrasast.core import canonical + +if TYPE_CHECKING: + from openultrasast.core.factpack import ( + FactPack, + IndependenceAttestation, + KnowledgeEvidence, + RelationEffect, + SemanticFact, + StructuralRequirement, + ) + +ATTESTATION_SCHEMA = "openultrasast.detection-core.independence-attestation.v2" +REVIEW_POLICY_SCHEMA = "openultrasast.detection-core.fact-review-policy.v2" +REVIEW_EVIDENCE_SCHEMA = "openultrasast.detection-core.fact-review-evidence.v2" + + +def _ref_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 attestation_digest(attestation: IndependenceAttestation) -> str: + return canonical.content_digest( + { + "schema_version": ATTESTATION_SCHEMA, + "subject_digest": attestation.subject_digest, + "producer_identity": attestation.producer_identity, + "reviewer_identity": attestation.reviewer_identity, + "conflict_disclosure": _ref_value(attestation.conflict_disclosure), + } + ) + + +def review_policy_digest( + reviewed_reviewer_digests: tuple[str, ...], + forbidden_identity_pairs: tuple[tuple[str, str], ...], + trusted_store_identities: tuple[str, ...], + maximum_evidence_bytes: int, +) -> str: + return canonical.content_digest( + { + "schema_version": REVIEW_POLICY_SCHEMA, + "reviewed_reviewer_digests": tuple(sorted(reviewed_reviewer_digests)), + "forbidden_identity_pairs": tuple(sorted(forbidden_identity_pairs)), + "trusted_store_identities": tuple(sorted(trusted_store_identities)), + "maximum_evidence_bytes": maximum_evidence_bytes, + } + ) + + +def _attestation_value(attestation: IndependenceAttestation) -> canonical.JsonValue: + return { + "subject_digest": attestation.subject_digest, + "producer_identity": attestation.producer_identity, + "reviewer_identity": attestation.reviewer_identity, + "conflict_disclosure": _ref_value(attestation.conflict_disclosure), + "attestation_digest": attestation.attestation_digest, + } + + +def _evidence_value(evidence: KnowledgeEvidence, *, include_lineage: bool) -> canonical.JsonValue: + """Project evidence; lineage is omitted only from the digest it names as subject.""" + value: dict[str, canonical.JsonValue] = { + "kind": evidence.kind.value, + "content": _ref_value(evidence.content), + "producer_identity": evidence.producer_identity, + "reviewer_identity": evidence.reviewer_identity, + "independence": _attestation_value(evidence.independence), + } + if include_lineage: + value["lineage_manifest_digest"] = evidence.lineage_manifest_digest + return value + + +def _requirement_value(requirement: StructuralRequirement) -> canonical.JsonValue: + return { + "requirement_id": requirement.requirement_id, + "kind": requirement.kind.value, + "field": requirement.field, + "expected": requirement.expected, + } + + +def _effect_value(effect: RelationEffect) -> canonical.JsonValue: + return { + "kind": effect.kind.value, + "role": effect.role, + "source_role": effect.source_role, + "target_role": effect.target_role, + "conditions": effect.conditions, + } + + +def _fact_value(fact: SemanticFact) -> canonical.JsonValue: + return { + "fact_id": fact.fact_id, + "role": fact.role, + "structural_requirements": tuple(_requirement_value(item) for item in fact.structural_requirements), + "relation_effects": tuple(_effect_value(item) for item in fact.relation_effects), + } + + +def pack_content_digest(pack: FactPack) -> str: + return canonical.content_digest( + { + "schema_version": pack.schema_version, + "family_contract_digest": pack.family_contract_digest, + "family_id": pack.family_id, + "ecosystem": pack.ecosystem, + "language": pack.language, + "facts": tuple(_fact_value(fact) for fact in pack.facts), + "review_identity": pack.review_identity, + "review_evidence": tuple(_evidence_value(item, include_lineage=False) for item in pack.review_evidence), + "review_policy_digest": pack.review_policy_digest, + } + ) + + +def evidence_digest(review_evidence: tuple[KnowledgeEvidence, ...]) -> str: + return canonical.content_digest( + { + "schema_version": REVIEW_EVIDENCE_SCHEMA, + "review_evidence": tuple(_evidence_value(item, include_lineage=True) for item in review_evidence), + } + ) diff --git a/src/openultrasast/core/_legacy_factpack.py b/src/openultrasast/core/_legacy_factpack.py new file mode 100644 index 0000000..d06ad1c --- /dev/null +++ b/src/openultrasast/core/_legacy_factpack.py @@ -0,0 +1,59 @@ +"""Compatibility facade for the pre-v2 file-backed fact-pack API.""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Any + + +class FactPackError(ValueError): + """A legacy fact pack is unreviewed or inconsistent.""" + + +@dataclass(frozen=True) +class LegacyFactPack: + family: str + facts: tuple[dict[str, Any], ...] + review_digest: str + fact_set_digest: str + + +def _digest(payload: object) -> str: + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + return hashlib.sha256(encoded).hexdigest()[:12] + + +def admit_fact_pack(family: str, facts: Sequence[Mapping[str, Any]], *, review_digest: str) -> LegacyFactPack: + if not review_digest: + raise FactPackError("unreviewed fact pack refused: no review digest") + normalized = tuple(dict(fact) for fact in facts) + return LegacyFactPack(family, normalized, review_digest, _digest([family, normalized])) + + +def verify_fact_pack(pack: LegacyFactPack, facts: Sequence[Mapping[str, Any]]) -> None: + normalized = tuple(dict(fact) for fact in facts) + if _digest([pack.family, normalized]) != pack.fact_set_digest: + raise FactPackError(f"fact content does not match the stamped digest of pack {pack.family!r}") + + +def load_pack_file(path: str | Path) -> LegacyFactPack: + payload = json.loads(Path(path).read_text(encoding="utf-8")) + review = payload.get("review", {}) + note = review.get("note", "") + if not note or review.get("digest") != hashlib.sha256(note.encode()).hexdigest(): + raise FactPackError(f"pack {path} lacks a verifiable review attestation") + facts = payload.get("facts") + if not isinstance(facts, list) or not all(isinstance(fact, dict) for fact in facts): + raise FactPackError(f"pack {path}: facts must be a list of objects") + return admit_fact_pack(str(payload.get("family", "")), facts, review_digest=review["digest"]) + + +def admit_pack_file(path: str | Path, *, review_digest: str) -> LegacyFactPack: + pack = load_pack_file(path) + if review_digest != pack.review_digest: + raise FactPackError(f"pack {path}: supplied review digest does not match the embedded attestation") + return pack diff --git a/src/openultrasast/core/factpack.py b/src/openultrasast/core/factpack.py index aeed2e8..38d95ef 100644 --- a/src/openultrasast/core/factpack.py +++ b/src/openultrasast/core/factpack.py @@ -1,80 +1,461 @@ -"""Family fact-pack admission contract for the detection core (Req 10). - -New vulnerability families enter only as reviewed fact packs: the pack carries its -family tag and a review digest, and its fact-set digest is stamped so consumers can -refuse tampered content. The core never reads pool or admission labels. -""" +"""Pure versioned semantic fact-pack admission contracts.""" from __future__ import annotations -import hashlib -import json +import re from collections.abc import Mapping, Sequence from dataclasses import dataclass -from pathlib import Path -from typing import Any +from enum import Enum +from types import MappingProxyType + +from openultrasast.core import _factpack_digest as digest_projection +from openultrasast.core import canonical +from openultrasast.core._legacy_factpack import FactPackError, admit_fact_pack, admit_pack_file, load_pack_file, verify_fact_pack +from openultrasast.core.custody import LineagePurpose, PopulationCustodyPolicy, PopulationCustodyPort +from openultrasast.core.family import FamilyContract, family_contract_digest, validate_contract +from openultrasast.core.lineage import LineageExpectation, LineageManifest, validate_lineage +from openultrasast.core.semantics import RelationKind + +FACT_PACK_SCHEMA = "openultrasast.detection-core.fact-pack.v2" +_DIGEST = re.compile(r"[0-9a-f]{64}") -class FactPackError(ValueError): - """A fact pack is unreviewed or its content does not match its digest.""" +class FactPackValidationError(ValueError): + """A v2 fact pack cannot be reproduced under the supplied trust context.""" + + +class KnowledgeEvidenceKind(Enum): + SOURCE_CODE = "source_code" + OFFICIAL_DOCUMENTATION = "official_documentation" + INDEPENDENT_BEHAVIOR = "independent_behavior" + + +class StructuralRequirementKind(Enum): + QUALIFIED_NAME = "qualified_name" + AST_SHAPE = "ast_shape" + RECEIVER_TYPE = "receiver_type" + CALL_TARGET = "call_target" + ARGUMENT_SHAPE = "argument_shape" + CONFIGURATION = "configuration" + + +def _text(value: object, field: str) -> None: + if not isinstance(value, str) or not value: + raise FactPackValidationError(f"{field} must be nonempty text") + + +def _digest(value: object, field: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise FactPackValidationError(f"{field} must be a full lowercase SHA-256 digest") + + +def _unique(values: tuple[str, ...], field: str) -> None: + if len(set(values)) != len(values): + raise FactPackValidationError(f"duplicate {field}") + + +def _digest_tuple(values: object, field: str) -> None: + if not isinstance(values, tuple) or not values: + raise FactPackValidationError(f"{field} must be a nonempty tuple") + for value in values: + _digest(value, field) + if len(set(values)) != len(values): + raise FactPackValidationError(f"duplicate {field}") + + +def _forbidden_pairs(values: object) -> None: + if not isinstance(values, tuple): + raise FactPackValidationError("forbidden identity pairs must be a tuple") + for pair in values: + if not isinstance(pair, tuple) or len(pair) != 2: + raise FactPackValidationError("forbidden identity pairs must contain pairs") + _digest(pair[0], "forbidden producer") + _digest(pair[1], "forbidden reviewer") + if len(set(values)) != len(values): + raise FactPackValidationError("duplicate forbidden identity pair") + + +def _text_tuple(values: object, field: str) -> None: + if not isinstance(values, tuple) or not values: + raise FactPackValidationError(f"{field} must be a nonempty tuple") + for value in values: + _text(value, field) + if len(set(values)) != len(values): + raise FactPackValidationError(f"duplicate {field}") + + +def _freeze_json(value: canonical.JsonValue) -> canonical.JsonValue: + if isinstance(value, Mapping): + return MappingProxyType({key: _freeze_json(item) for key, item in value.items()}) + if isinstance(value, tuple): + return tuple(_freeze_json(item) for item in value) + return value + + +@dataclass(frozen=True) +class IndependenceAttestation: + subject_digest: str + producer_identity: str + reviewer_identity: str + conflict_disclosure: canonical.ContentRef + attestation_digest: str + + def __post_init__(self) -> None: + for value, field in ( + (self.subject_digest, "attestation subject"), + (self.producer_identity, "attestation producer"), + (self.reviewer_identity, "attestation reviewer"), + (self.attestation_digest, "attestation digest"), + ): + _digest(value, field) + if not isinstance(self.conflict_disclosure, canonical.ContentRef): + raise FactPackValidationError("conflict disclosure must be a ContentRef") + + +@dataclass(frozen=True) +class ReviewIndependencePolicy: + reviewed_reviewer_digests: tuple[str, ...] + forbidden_identity_pairs: tuple[tuple[str, str], ...] + trusted_store_identities: tuple[str, ...] + maximum_evidence_bytes: int + policy_digest: str + + def __post_init__(self) -> None: + _digest_tuple(self.reviewed_reviewer_digests, "reviewed reviewer") + _forbidden_pairs(self.forbidden_identity_pairs) + _text_tuple(self.trusted_store_identities, "trusted store identity") + if isinstance(self.maximum_evidence_bytes, bool) or not isinstance(self.maximum_evidence_bytes, int): + raise FactPackValidationError("maximum evidence bytes must be a positive integer") + if self.maximum_evidence_bytes <= 0: + raise FactPackValidationError("maximum evidence bytes must be a positive integer") + _digest(self.policy_digest, "review policy digest") + + +@dataclass(frozen=True) +class KnowledgeEvidence: + kind: KnowledgeEvidenceKind + content: canonical.ContentRef + producer_identity: str + reviewer_identity: str + independence: IndependenceAttestation + lineage_manifest_digest: str + + def __post_init__(self) -> None: + if not isinstance(self.kind, KnowledgeEvidenceKind): + raise FactPackValidationError("knowledge evidence kind is unknown") + if not isinstance(self.content, canonical.ContentRef): + raise FactPackValidationError("knowledge evidence content must be a ContentRef") + _digest(self.producer_identity, "evidence producer") + _digest(self.reviewer_identity, "evidence reviewer") + if not isinstance(self.independence, IndependenceAttestation): + raise FactPackValidationError("evidence independence must be an attestation") + _digest(self.lineage_manifest_digest, "evidence lineage manifest") + + +@dataclass(frozen=True) +class StructuralRequirement: + requirement_id: str + kind: StructuralRequirementKind + field: str + expected: canonical.JsonValue + + def __post_init__(self) -> None: + _text(self.requirement_id, "requirement_id") + if not isinstance(self.kind, StructuralRequirementKind): + raise FactPackValidationError("structural requirement kind is unknown") + _text(self.field, "structural requirement field") + frozen = _freeze_json(self.expected) + canonical.canonical_bytes(frozen) + object.__setattr__(self, "expected", frozen) + + +@dataclass(frozen=True) +class RelationEffect: + kind: RelationKind + role: str + source_role: str + target_role: str + conditions: tuple[str, ...] + + def __post_init__(self) -> None: + if not isinstance(self.kind, RelationKind): + raise FactPackValidationError("relation effect kind is unknown") + for value, field in ( + (self.role, "effect role"), + (self.source_role, "effect source role"), + (self.target_role, "effect target role"), + ): + _text(value, field) + if not isinstance(self.conditions, tuple): + raise FactPackValidationError("effect conditions must be a tuple") + for condition in self.conditions: + _text(condition, "effect condition") + _unique(self.conditions, "effect condition") + + +@dataclass(frozen=True) +class SemanticFact: + fact_id: str + role: str + structural_requirements: tuple[StructuralRequirement, ...] + relation_effects: tuple[RelationEffect, ...] + + def __post_init__(self) -> None: + _text(self.fact_id, "fact_id") + _text(self.role, "fact role") + if not isinstance(self.structural_requirements, tuple) or not self.structural_requirements: + raise FactPackValidationError("a fact requires a tuple of structural requirements") + if not all(isinstance(item, StructuralRequirement) for item in self.structural_requirements): + raise FactPackValidationError("structural requirements have the wrong type") + if not isinstance(self.relation_effects, tuple) or not self.relation_effects: + raise FactPackValidationError("a fact requires a tuple of relation effects") + if not all(isinstance(item, RelationEffect) for item in self.relation_effects): + raise FactPackValidationError("relation effects have the wrong type") + _unique(tuple(item.requirement_id for item in self.structural_requirements), "requirement_id") @dataclass(frozen=True) class FactPack: - """A reviewed family fact pack with provenance.""" + schema_version: str + family_contract_digest: str + family_id: str + ecosystem: str + language: str + facts: tuple[SemanticFact, ...] + normalized_content_digest: str + review_identity: str + review_evidence: tuple[KnowledgeEvidence, ...] + review_evidence_digest: str + review_policy_digest: str - family: str - facts: tuple[dict[str, Any], ...] - review_digest: str - fact_set_digest: str + def __post_init__(self) -> None: + _text(self.schema_version, "schema_version") + _digest(self.family_contract_digest, "family contract digest") + for value, field in ((self.family_id, "family_id"), (self.ecosystem, "ecosystem"), (self.language, "language")): + _text(value, field) + if not isinstance(self.facts, tuple) or not all(isinstance(item, SemanticFact) for item in self.facts): + raise FactPackValidationError("facts must be a tuple of SemanticFact values") + for value, field in ( + (self.normalized_content_digest, "normalized content digest"), + (self.review_identity, "review identity"), + (self.review_evidence_digest, "review evidence digest"), + (self.review_policy_digest, "review policy digest"), + ): + _digest(value, field) + if not isinstance(self.review_evidence, tuple) or not all(isinstance(item, KnowledgeEvidence) for item in self.review_evidence): + raise FactPackValidationError("review evidence must be a tuple of KnowledgeEvidence values") -def _digest(payload: object) -> str: - return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest()[:12] +def derive_independence_attestation_digest(attestation: IndependenceAttestation) -> str: + return digest_projection.attestation_digest(attestation) -def admit_fact_pack(family: str, facts: Sequence[Mapping[str, Any]], *, review_digest: str) -> FactPack: - """Admit a pack only with review evidence; stamp its fact-set digest (Req 10.1).""" - if not review_digest: - raise FactPackError("unreviewed fact pack refused: no review digest") - normalized = tuple(dict(fact) for fact in facts) - return FactPack( - family=family, - facts=normalized, - review_digest=review_digest, - fact_set_digest=_digest([family, normalized]), +def derive_review_independence_policy_digest( + reviewed_reviewer_digests: tuple[str, ...], + forbidden_identity_pairs: tuple[tuple[str, str], ...], + trusted_store_identities: tuple[str, ...], + maximum_evidence_bytes: int, +) -> str: + return digest_projection.review_policy_digest( + reviewed_reviewer_digests, + forbidden_identity_pairs, + trusted_store_identities, + maximum_evidence_bytes, ) -def verify_fact_pack(pack: FactPack, facts: Sequence[Mapping[str, Any]]) -> None: - """Refuse content that does not match the pack's stamped digest.""" - normalized = tuple(dict(fact) for fact in facts) - if _digest([pack.family, normalized]) != pack.fact_set_digest: - raise FactPackError(f"fact content does not match the stamped digest of pack {pack.family!r}") +def derive_fact_pack_content_digest(pack: FactPack) -> str: + return digest_projection.pack_content_digest(pack) -def load_pack_file(path: str | Path) -> FactPack: - """Load a pack file whose embedded review attestation is self-consistent. - - ``review.digest`` must equal sha256 of ``review.note``; anything else is refused. - """ - payload = json.loads(Path(path).read_text(encoding="utf-8")) - review = payload.get("review", {}) - note = review.get("note", "") - if not note or review.get("digest") != hashlib.sha256(note.encode()).hexdigest(): - raise FactPackError(f"pack {path} lacks a verifiable review attestation") - facts = payload.get("facts") - if not isinstance(facts, list) or not all(isinstance(fact, dict) for fact in facts): - raise FactPackError(f"pack {path}: facts must be a list of objects") - return admit_fact_pack(str(payload.get("family", "")), facts, review_digest=review["digest"]) +def derive_review_evidence_digest(review_evidence: tuple[KnowledgeEvidence, ...]) -> str: + return digest_projection.evidence_digest(review_evidence) -def admit_pack_file(path: str | Path, *, review_digest: str) -> FactPack: - """Admit a pack file only when its embedded attestation matches the supplied review digest.""" - pack = load_pack_file(path) - if review_digest != pack.review_digest: - raise FactPackError(f"pack {path}: supplied review digest does not match the embedded attestation") - return pack +def _revalidate_shapes(pack: FactPack, policy: ReviewIndependencePolicy) -> None: + pack.__post_init__() + policy.__post_init__() + for fact in pack.facts: + fact.__post_init__() + for requirement in fact.structural_requirements: + requirement.__post_init__() + for effect in fact.relation_effects: + effect.__post_init__() + for evidence in pack.review_evidence: + evidence.__post_init__() + evidence.independence.__post_init__() + evidence.content.__post_init__() + evidence.independence.conflict_disclosure.__post_init__() -__all__ = ["FactPack", "FactPackError", "admit_fact_pack", "admit_pack_file", "load_pack_file", "verify_fact_pack"] +def _validate_scope(pack: FactPack, contract: FamilyContract) -> None: + validate_contract(contract) + if pack.schema_version != FACT_PACK_SCHEMA: + raise FactPackValidationError("unknown fact-pack schema") + if pack.family_id != contract.family_id: + raise FactPackValidationError("fact-pack family does not match contract") + if pack.family_contract_digest != family_contract_digest(contract): + raise FactPackValidationError("family contract digest does not match contract") + if pack.ecosystem not in contract.ecosystems: + raise FactPackValidationError("fact-pack ecosystem is outside contract scope") + if not pack.facts: + raise FactPackValidationError("fact pack must contain facts") + _unique(tuple(fact.fact_id for fact in pack.facts), "fact identity") + + +def _validate_roles(pack: FactPack, contract: FamilyContract) -> None: + roles = {role.role: role for role in contract.roles} + for fact in pack.facts: + if fact.role not in roles: + raise FactPackValidationError(f"fact uses undeclared role: {fact.role}") + for effect in fact.relation_effects: + if any(role not in roles for role in (effect.role, effect.source_role, effect.target_role)): + raise FactPackValidationError("relation effect uses an undeclared role") + if effect.kind not in roles[effect.role].allowed_relation_kinds: + raise FactPackValidationError("relation kind is not allowed for its effect role") + + +def _validate_policy(pack: FactPack, policy: ReviewIndependencePolicy) -> None: + expected = derive_review_independence_policy_digest( + policy.reviewed_reviewer_digests, + policy.forbidden_identity_pairs, + policy.trusted_store_identities, + policy.maximum_evidence_bytes, + ) + if policy.policy_digest != expected: + raise FactPackValidationError("review policy digest cannot be re-derived") + if pack.review_policy_digest != policy.policy_digest: + raise FactPackValidationError("pack review policy digest does not match supplied policy") + + +def _resolve_evidence( + evidence: KnowledgeEvidence, + policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, +) -> None: + attestation = evidence.independence + if attestation.subject_digest != evidence.content.content_digest: + raise FactPackValidationError("attestation subject does not match evidence content digest") + if attestation.producer_identity != evidence.producer_identity: + raise FactPackValidationError("attestation producer does not match evidence producer") + if attestation.reviewer_identity != evidence.reviewer_identity: + raise FactPackValidationError("attestation reviewer does not match evidence reviewer") + if attestation.attestation_digest != derive_independence_attestation_digest(attestation): + raise FactPackValidationError("attestation digest cannot be re-derived") + if evidence.reviewer_identity not in policy.reviewed_reviewer_digests: + raise FactPackValidationError("evidence reviewer is not trusted") + if (evidence.producer_identity, evidence.reviewer_identity) in policy.forbidden_identity_pairs: + raise FactPackValidationError("evidence producer/reviewer pair is forbidden") + if evidence.kind is KnowledgeEvidenceKind.INDEPENDENT_BEHAVIOR and evidence.producer_identity == evidence.reviewer_identity: + raise FactPackValidationError("behavioral evidence producer and reviewer must be independent") + for reference in (evidence.content, attestation.conflict_disclosure): + content = canonical.resolve_content( + reference, + resolver, + trusted_store_identities=policy.trusted_store_identities, + maximum_bytes=policy.maximum_evidence_bytes, + ) + if not content: + raise FactPackValidationError("review evidence and conflict disclosures must be nonempty") + + +def _lineage_for(evidence: KnowledgeEvidence, lineages: Sequence[LineageManifest]) -> LineageManifest: + matches = tuple(item for item in lineages if item.manifest_digest == evidence.lineage_manifest_digest) + if len(matches) != 1: + raise FactPackValidationError("evidence lineage manifest is missing or ambiguous") + return matches[0] + + +def _validate_evidence_lineage( + pack: FactPack, + evidence: KnowledgeEvidence, + lineages: Sequence[LineageManifest], + protected_population_digest: str | None, + custody: PopulationCustodyPort, + custody_policy: PopulationCustodyPolicy, +) -> None: + lineage = _lineage_for(evidence, lineages) + if lineage.subject_digest != pack.normalized_content_digest and pack.normalized_content_digest != derive_fact_pack_content_digest(pack): + raise FactPackValidationError("normalized content digest cannot be re-derived") + expectations = ( + LineageExpectation(LineagePurpose.SEMANTIC_FACT, pack.normalized_content_digest, 1, 1), + LineageExpectation(LineagePurpose.PROMOTION, pack.normalized_content_digest, 0, 1), + ) + decision = validate_lineage( + lineage, + pack.normalized_content_digest, + expectations, + protected_population_digest, + True, + custody, + custody_policy, + ) + if not decision.allowed: + raise FactPackValidationError("fact lineage refused: " + ",".join(decision.reasons)) + + +def _validate_digests(pack: FactPack) -> None: + if pack.normalized_content_digest != derive_fact_pack_content_digest(pack): + raise FactPackValidationError("normalized content digest cannot be re-derived") + if pack.review_evidence_digest != derive_review_evidence_digest(pack.review_evidence): + raise FactPackValidationError("review evidence digest cannot be re-derived") + + +def validate_fact_pack( + pack: FactPack, + contract: FamilyContract, + protected_population_digest: str | None, + lineages: Sequence[LineageManifest], + custody: PopulationCustodyPort, + custody_policy: PopulationCustodyPolicy, + review_policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, +) -> None: + """Revalidate a closed v2 pack against exact scope, trust, artifacts, and lineage.""" + if not isinstance(pack, FactPack) or not isinstance(review_policy, ReviewIndependencePolicy): + raise FactPackValidationError("pack and review policy must use v2 contract types") + _revalidate_shapes(pack, review_policy) + _validate_scope(pack, contract) + _validate_roles(pack, contract) + _validate_policy(pack, review_policy) + if not pack.review_evidence: + raise FactPackValidationError("fact pack requires review evidence") + _unique(tuple(item.independence.attestation_digest for item in pack.review_evidence), "review attestation") + for evidence in pack.review_evidence: + _resolve_evidence(evidence, review_policy, resolver) + _validate_evidence_lineage( + pack, + evidence, + lineages, + protected_population_digest, + custody, + custody_policy, + ) + if pack.review_identity not in {item.reviewer_identity for item in pack.review_evidence}: + raise FactPackValidationError("pack review identity is absent from review evidence") + _validate_digests(pack) + + +__all__ = [ + "FACT_PACK_SCHEMA", + "FactPack", + "FactPackError", + "FactPackValidationError", + "IndependenceAttestation", + "KnowledgeEvidence", + "KnowledgeEvidenceKind", + "RelationEffect", + "ReviewIndependencePolicy", + "SemanticFact", + "StructuralRequirement", + "StructuralRequirementKind", + "admit_fact_pack", + "admit_pack_file", + "derive_fact_pack_content_digest", + "derive_independence_attestation_digest", + "derive_review_evidence_digest", + "derive_review_independence_policy_digest", + "load_pack_file", + "validate_fact_pack", + "verify_fact_pack", +] diff --git a/tests/test_core_factpack.py b/tests/test_core_factpack.py new file mode 100644 index 0000000..d02dece --- /dev/null +++ b/tests/test_core_factpack.py @@ -0,0 +1,406 @@ +"""Versioned semantic fact-pack and independent-review contracts.""" + +from __future__ import annotations + +from collections.abc import Mapping, MutableMapping +from dataclasses import dataclass, replace +from typing import cast + +import pytest +from detection_core_fixtures import ( + InMemoryContentResolver, + InMemoryPopulationCustody, + custody_contract, + stored_artifact, +) + +from openultrasast.core import canonical, provenance +from openultrasast.core.evidence import NeutralOutcome +from openultrasast.core.factpack import ( + FACT_PACK_SCHEMA, + FactPack, + FactPackValidationError, + 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, + validate_fact_pack, +) +from openultrasast.core.family import ( + DecisionRule, + FamilyContract, + FieldSpec, + ObjectSpec, + RelationOriginPolicy, + ResidualQuestion, + RoleSpec, + family_contract_digest, +) +from openultrasast.core.semantics import FieldKind, RelationKind, RelationStatus + + +@dataclass(frozen=True) +class PackCase: + pack: FactPack + contract: FamilyContract + lineages: tuple[provenance.LineageManifest, ...] + custody: InMemoryPopulationCustody + custody_policy: provenance.PopulationCustodyPolicy + review_policy: ReviewIndependencePolicy + resolver: InMemoryContentResolver + + +def _contract() -> FamilyContract: + return FamilyContract( + family_id="reference-family", + family_version="1", + schema_version="family.v2", + ecosystems=("python",), + security_object=ObjectSpec("resource", (FieldSpec("name", FieldKind.TEXT, True),), ("name",)), + roles=( + RoleSpec("flow", (), (RelationKind.CALL,), RelationOriginPolicy.PRODUCT_OR_REVIEWED_FACT), + RoleSpec("source", (FieldSpec("name", FieldKind.TEXT, False),), (), RelationOriginPolicy.PRODUCT_OR_REVIEWED_FACT), + RoleSpec("target", (FieldSpec("name", FieldKind.TEXT, False),), (), RelationOriginPolicy.PRODUCT_OR_REVIEWED_FACT), + ), + decisions=(DecisionRule("unknown", NeutralOutcome.INDETERMINATE, (), (RelationStatus.UNRESOLVED,)),), + residual_questions=(ResidualQuestion("classify", ("flow",), ("unknown",)),), + indeterminate_reasons=("missing",), + ) + + +def _manifest( + subject: str, + custody: InMemoryPopulationCustody, + custody_policy: provenance.PopulationCustodyPolicy, + custodian: str, + *, + labels_accessed: bool = False, +) -> provenance.LineageManifest: + initial = custody.current_log(canonical.content_digest("population")) + request = provenance.PopulationUseRequest( + population_digest=initial.population_digest, + arm_id="fact-arm", + purpose=provenance.LineagePurpose.SEMANTIC_FACT, + labels_accessed=labels_accessed, + influenced_artifact_digest=subject, + expected_log_digest=initial.log_digest, + ) + receipt = custody.append(request, custodian) + use = provenance.LineageUse( + purpose=request.purpose, + source_population_digest=request.population_digest, + labels_accessed=labels_accessed, + influenced_artifact_digest=subject, + population_use_receipt_digest=receipt.receipt_digest, + ) + digest = provenance.derive_lineage_manifest_digest("fact-arm", subject, (use,)) + return provenance.LineageManifest("fact-arm", subject, (use,), digest) + + +def _case(kind: KnowledgeEvidenceKind = KnowledgeEvidenceKind.SOURCE_CODE, *, labels_accessed: bool = False) -> PackCase: + contract = _contract() + producer = canonical.content_digest("producer") + reviewer = canonical.content_digest("reviewer") + content, content_resolver = stored_artifact(b'{"reviewed":"content"}', artifact_id="knowledge") + disclosure, disclosure_resolver = stored_artifact(b'{"conflicts":[]}', artifact_id="conflicts") + resolver = InMemoryContentResolver({**content_resolver.artifacts, **disclosure_resolver.artifacts}) + provisional_attestation = IndependenceAttestation(content.content_digest, producer, reviewer, disclosure, "0" * 64) + attestation = replace( + provisional_attestation, + attestation_digest=derive_independence_attestation_digest(provisional_attestation), + ) + provisional_evidence = KnowledgeEvidence(kind, content, producer, reviewer, attestation, "0" * 64) + reviewer_digests = (reviewer,) + forbidden_pairs: tuple[tuple[str, str], ...] = () + stores = ("trusted-store",) + policy = ReviewIndependencePolicy( + reviewer_digests, + forbidden_pairs, + stores, + 4096, + derive_review_independence_policy_digest(reviewer_digests, forbidden_pairs, stores, 4096), + ) + fact = SemanticFact( + "stable.fact.v1", + "flow", + (StructuralRequirement("qualified-name", StructuralRequirementKind.QUALIFIED_NAME, "callee", "pkg.call"),), + (RelationEffect(RelationKind.CALL, "flow", "source", "target", ("receiver_resolved",)),), + ) + provisional_pack = FactPack( + FACT_PACK_SCHEMA, + family_contract_digest(contract), + contract.family_id, + "python", + "python", + (fact,), + "0" * 64, + reviewer, + (provisional_evidence,), + "0" * 64, + policy.policy_digest, + ) + normalized = derive_fact_pack_content_digest(provisional_pack) + initial, custody_policy, custodian = custody_contract() + custody = InMemoryPopulationCustody(initial, custody_policy) + lineage = _manifest(normalized, custody, custody_policy, custodian, labels_accessed=labels_accessed) + 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 PackCase(pack, contract, (lineage,), custody, custody_policy, policy, resolver) + + +def _validate(case: PackCase) -> None: + validate_fact_pack( + case.pack, + case.contract, + None, + case.lineages, + case.custody, + case.custody_policy, + case.review_policy, + case.resolver, + ) + + +def _with_policy(case: PackCase, policy: ReviewIndependencePolicy) -> PackCase: + pack = replace(case.pack, review_policy_digest=policy.policy_digest) + pack = replace(pack, normalized_content_digest=derive_fact_pack_content_digest(pack)) + initial, custody_policy, custodian = custody_contract() + custody = InMemoryPopulationCustody(initial, custody_policy) + lineage = _manifest(pack.normalized_content_digest, custody, custody_policy, custodian) + evidence = replace(pack.review_evidence[0], lineage_manifest_digest=lineage.manifest_digest) + pack = replace(pack, review_evidence=(evidence,), review_evidence_digest=derive_review_evidence_digest((evidence,))) + return replace( + case, + pack=pack, + lineages=(lineage,), + custody=custody, + custody_policy=custody_policy, + review_policy=policy, + ) + + +@pytest.mark.parametrize("kind", tuple(KnowledgeEvidenceKind)) +def test_reviewed_label_isolated_pack_accepts_each_evidence_kind(kind: KnowledgeEvidenceKind) -> None: + case = _case(kind) + + _validate(case) + + +@pytest.mark.parametrize( + ("field", "value", "message"), + ( + ("schema_version", "fact-pack.v1", "schema"), + ("family_id", "other-family", "family"), + ("ecosystem", "javascript", "ecosystem"), + ("family_contract_digest", "f" * 64, "contract digest"), + ("normalized_content_digest", "f" * 64, "normalized content digest"), + ("review_evidence_digest", "f" * 64, "review evidence digest"), + ("review_policy_digest", "f" * 64, "review policy digest"), + ), +) +def test_pack_rejects_scope_or_digest_substitution(field: str, value: str, message: str) -> None: + case = _case() + if field == "schema_version": + pack = replace(case.pack, schema_version=value) + elif field == "family_id": + pack = replace(case.pack, family_id=value) + elif field == "ecosystem": + pack = replace(case.pack, ecosystem=value) + elif field == "family_contract_digest": + pack = replace(case.pack, family_contract_digest=value) + elif field == "normalized_content_digest": + pack = replace(case.pack, normalized_content_digest=value) + elif field == "review_evidence_digest": + pack = replace(case.pack, review_evidence_digest=value) + else: + assert field == "review_policy_digest" + pack = replace(case.pack, review_policy_digest=value) + changed = replace(case, pack=pack) + + with pytest.raises(FactPackValidationError, match=message): + _validate(changed) + + +def test_pack_rejects_missing_review_and_duplicate_fact_ids() -> None: + case = _case() + missing = replace(case, pack=replace(case.pack, review_evidence=())) + duplicate = replace(case, pack=replace(case.pack, facts=(case.pack.facts[0], case.pack.facts[0]))) + + with pytest.raises(FactPackValidationError, match="review evidence"): + _validate(missing) + with pytest.raises(FactPackValidationError, match="duplicate fact"): + _validate(duplicate) + + +@pytest.mark.parametrize("part", ("fact", "effect", "source", "target", "kind")) +def test_pack_rejects_undeclared_roles_and_relation_kinds(part: str) -> None: + case = _case() + fact = case.pack.facts[0] + effect = fact.relation_effects[0] + if part == "fact": + fact = replace(fact, role="undeclared") + elif part == "kind": + fact = replace(fact, relation_effects=(replace(effect, kind=RelationKind.RETURN),)) + elif part == "effect": + fact = replace(fact, relation_effects=(replace(effect, role="undeclared"),)) + elif part == "source": + fact = replace(fact, relation_effects=(replace(effect, source_role="undeclared"),)) + else: + assert part == "target" + fact = replace(fact, relation_effects=(replace(effect, target_role="undeclared"),)) + changed = replace(case, pack=replace(case.pack, facts=(fact,))) + + with pytest.raises(FactPackValidationError, match="role|relation kind"): + _validate(changed) + + +def test_pack_rejects_attestation_substitution_and_unreadable_conflict_disclosure() -> None: + case = _case() + evidence = case.pack.review_evidence[0] + wrong_subject = replace(evidence.independence, subject_digest="f" * 64) + changed = replace(case, pack=replace(case.pack, review_evidence=(replace(evidence, independence=wrong_subject),))) + + with pytest.raises(FactPackValidationError, match="attestation subject"): + _validate(changed) + + unreadable = replace( + case, + resolver=InMemoryContentResolver({key: value for key, value in case.resolver.artifacts.items() if key != "conflicts"}), + ) + with pytest.raises(canonical.ArtifactResolutionError): + _validate(unreadable) + + +@pytest.mark.parametrize("failure", ("self_review", "unknown_reviewer", "forbidden_pair")) +def test_behavioral_evidence_requires_independent_trusted_review(failure: str) -> None: + case = _case(KnowledgeEvidenceKind.INDEPENDENT_BEHAVIOR) + evidence = case.pack.review_evidence[0] + if failure == "self_review": + evidence = replace(evidence, producer_identity=evidence.reviewer_identity) + elif failure == "unknown_reviewer": + unknown = canonical.content_digest("unknown") + evidence = replace(evidence, reviewer_identity=unknown) + else: + pair = ((evidence.producer_identity, evidence.reviewer_identity),) + policy = replace( + case.review_policy, + forbidden_identity_pairs=pair, + policy_digest=derive_review_independence_policy_digest( + case.review_policy.reviewed_reviewer_digests, + pair, + case.review_policy.trusted_store_identities, + case.review_policy.maximum_evidence_bytes, + ), + ) + case = _with_policy(case, policy) + changed = replace(case, pack=replace(case.pack, review_evidence=(evidence,))) + + with pytest.raises(FactPackValidationError, match="producer|reviewer|forbidden"): + _validate(changed) + + +def test_source_evidence_may_be_produced_by_its_trusted_reviewer() -> None: + case = _case(KnowledgeEvidenceKind.SOURCE_CODE) + evidence = case.pack.review_evidence[0] + same = evidence.reviewer_identity + attestation = replace(evidence.independence, producer_identity=same) + attestation = replace(attestation, attestation_digest=derive_independence_attestation_digest(attestation)) + evidence = replace(evidence, producer_identity=same, independence=attestation) + pack = replace(case.pack, review_evidence=(evidence,)) + pack = replace( + pack, + normalized_content_digest=derive_fact_pack_content_digest(pack), + review_evidence_digest=derive_review_evidence_digest((evidence,)), + ) + # The lineage subject must follow the amended normalized review content. + initial, policy, custodian = custody_contract() + custody = InMemoryPopulationCustody(initial, policy) + lineage = _manifest(pack.normalized_content_digest, custody, policy, custodian) + evidence = replace(evidence, lineage_manifest_digest=lineage.manifest_digest) + pack = replace(pack, review_evidence=(evidence,), review_evidence_digest=derive_review_evidence_digest((evidence,))) + + _validate(replace(case, pack=pack, lineages=(lineage,), custody=custody, custody_policy=policy)) + + +def test_pack_rejects_missing_wrong_or_label_derived_lineage() -> None: + case = _case() + with pytest.raises(FactPackValidationError, match="lineage"): + _validate(replace(case, lineages=())) + + wrong = replace(case.lineages[0], subject_digest=canonical.content_digest("other")) + with pytest.raises((FactPackValidationError, provenance.LineageValidationError), match="lineage|subject"): + _validate(replace(case, lineages=(wrong,))) + + labelled = _case(labels_accessed=True) + with pytest.raises(FactPackValidationError, match="label_access_forbidden"): + _validate(labelled) + + +def test_pack_revalidates_policy_and_resolver_trust_boundaries() -> None: + case = _case() + tampered = replace(case.review_policy, policy_digest="f" * 64) + with pytest.raises(FactPackValidationError, match="policy digest"): + _validate(replace(case, review_policy=tampered)) + + too_small = replace( + case.review_policy, + maximum_evidence_bytes=1, + policy_digest=derive_review_independence_policy_digest( + case.review_policy.reviewed_reviewer_digests, + case.review_policy.forbidden_identity_pairs, + case.review_policy.trusted_store_identities, + 1, + ), + ) + with pytest.raises(canonical.ArtifactValidationError, match="maximum"): + _validate(_with_policy(case, too_small)) + + +def test_structural_requirement_defensively_freezes_nested_caller_json() -> None: + nested_item: dict[str, canonical.JsonValue] = {"name": "original"} + caller_sequence: tuple[canonical.JsonValue, ...] = ("first", nested_item, "last") + caller_value: dict[str, canonical.JsonValue] = {"items": caller_sequence} + requirement = StructuralRequirement("shape", StructuralRequirementKind.AST_SHAPE, "node", caller_value) + case = _case() + fact = replace(case.pack.facts[0], structural_requirements=(requirement,)) + pack = replace(case.pack, facts=(fact,)) + original_digest = derive_fact_pack_content_digest(pack) + + nested_item["name"] = "mutated" + caller_value["added"] = True + exposed = cast(MutableMapping[str, canonical.JsonValue], requirement.expected) + with pytest.raises(TypeError): + exposed["added"] = False + + frozen = cast(Mapping[str, canonical.JsonValue], requirement.expected) + frozen_items = cast(tuple[canonical.JsonValue, ...], frozen["items"]) + frozen_nested = cast(Mapping[str, canonical.JsonValue], frozen_items[1]) + assert tuple(frozen_items) == ("first", frozen_items[1], "last") + assert frozen_nested["name"] == "original" + assert "added" not in frozen + assert derive_fact_pack_content_digest(pack) == original_digest + + +def test_pack_revalidation_freezes_and_rejects_post_construction_expected_tamper() -> None: + case = _case() + requirement = case.pack.facts[0].structural_requirements[0] + object.__setattr__(requirement, "expected", {"callee": "tampered"}) + + with pytest.raises(FactPackValidationError, match="normalized content digest"): + _validate(case) + assert isinstance(requirement.expected, Mapping) + with pytest.raises(TypeError): + cast(MutableMapping[str, canonical.JsonValue], requirement.expected)["callee"] = "again"