mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-11-13 17:58:18 +00:00
feat(detection-core): validate reviewed fact packs
This commit is contained in:
Marius Ciepluch
@@ -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,
|
||||
|
||||
@@ -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),
|
||||
}
|
||||
)
|
||||
@@ -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
|
||||
@@ -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",
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user