diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index c63ecc5..2183485 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -73,7 +73,7 @@ - _Requirements: 1.1, 1.2, 1.3, 2.1, 2.2, 2.3, 2.4, 2.5, 10.1_ - [ ] 3. Add vulnerability-family contracts and unlike reference families -- [ ] 3.1 Validate a closed vulnerability-family declaration +- [x] 3.1 Validate a closed vulnerability-family declaration - Require family/schema versions, ecosystems, security-object fields, roles, decisions, outcome rules, completeness conditions, residual questions, and indeterminate reasons. - Enforce typed family fields and exact ecosystem/schema scope. - Declare whether each role accepts product evidence or requires reviewed excluded-dependency facts. diff --git a/src/openultrasast/core/family.py b/src/openultrasast/core/family.py new file mode 100644 index 0000000..eff0a02 --- /dev/null +++ b/src/openultrasast/core/family.py @@ -0,0 +1,381 @@ +"""Closed vulnerability-family declarations and evidence conformance.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum + +from openultrasast.core import canonical +from openultrasast.core.evidence import EvidenceRecord, NeutralOutcome +from openultrasast.core.semantics import ( + FieldKind, + FieldValue, + Observation, + ProvenanceKind, + RelationKind, + RelationStatus, + SemanticRelation, +) + +MISSING_REVIEWED_DEPENDENCY = "missing_reviewed_dependency_relation" + + +class FamilyContractError(ValueError): + """A family declaration or evidence record violates its closed contract.""" + + +class UnsupportedFamilyScope(FamilyContractError): + """Evidence names a family, schema, or ecosystem outside the declaration.""" + + +class RelationOriginPolicy(Enum): + PRODUCT_OR_REVIEWED_FACT = "product_or_reviewed_fact" + EXCLUDED_DEPENDENCY_FACT_ONLY = "excluded_dependency_fact_only" + + +def _text(value: object, field: str) -> None: + if not isinstance(value, str) or not value: + raise FamilyContractError(f"{field} must be nonempty text") + + +def _tuple(value: object, item_type: type[object], field: str, *, nonempty: bool = True) -> None: + if not isinstance(value, tuple) or (nonempty and not value) or not all(isinstance(item, item_type) for item in value): + raise FamilyContractError(f"{field} must be a{' nonempty' if nonempty else ''} tuple of {item_type.__name__}") + + +def _unique(values: object, field: str) -> None: + if not isinstance(values, tuple) or not all(isinstance(item, str) and item for item in values): + raise FamilyContractError(f"{field} must contain nonempty text") + if len(set(values)) != len(values): + raise FamilyContractError(f"duplicate {field}") + + +@dataclass(frozen=True) +class FieldSpec: + name: str + kind: FieldKind + required: bool + + def __post_init__(self) -> None: + _text(self.name, "field name") + if not isinstance(self.kind, FieldKind): + raise FamilyContractError("field kind must be a FieldKind") + if not isinstance(self.required, bool): + raise FamilyContractError("required must be boolean") + + +@dataclass(frozen=True) +class ObjectSpec: + kind: str + fields: tuple[FieldSpec, ...] + identity_fields: tuple[str, ...] + + def __post_init__(self) -> None: + _text(self.kind, "object kind") + _field_specs(self.fields, "security-object fields", nonempty=True) + _unique(self.identity_fields, "identity fields") + if not self.identity_fields: + raise FamilyContractError("identity fields must be nonempty") + fields = {field.name: field for field in self.fields} + if any(name not in fields or not fields[name].required for name in self.identity_fields): + raise FamilyContractError("identity fields must name declared required fields") + + +@dataclass(frozen=True) +class RoleSpec: + role: str + fields: tuple[FieldSpec, ...] + allowed_relation_kinds: tuple[RelationKind, ...] + origin_policy: RelationOriginPolicy + + def __post_init__(self) -> None: + _text(self.role, "role") + _field_specs(self.fields, f"role {self.role} fields", nonempty=False) + _tuple(self.allowed_relation_kinds, RelationKind, "allowed relation kinds", nonempty=False) + _unique(tuple(item.value for item in self.allowed_relation_kinds), "allowed relation kinds") + if not self.fields and not self.allowed_relation_kinds: + raise FamilyContractError("a role must declare fields or allowed relation kinds") + if not isinstance(self.origin_policy, RelationOriginPolicy): + raise FamilyContractError("origin_policy must be a RelationOriginPolicy") + + +@dataclass(frozen=True) +class DecisionRule: + family_code: str + outcome: NeutralOutcome + required_roles: tuple[str, ...] + allowed_statuses: tuple[RelationStatus, ...] + + def __post_init__(self) -> None: + _text(self.family_code, "family decision code") + if not isinstance(self.outcome, NeutralOutcome): + raise FamilyContractError("decision outcome must be a NeutralOutcome") + _unique(self.required_roles, "required roles") + _tuple(self.allowed_statuses, RelationStatus, "allowed statuses") + _unique(tuple(item.value for item in self.allowed_statuses), "allowed statuses") + if self.outcome is not NeutralOutcome.INDETERMINATE and self.allowed_statuses != (RelationStatus.COMPLETE,): + raise FamilyContractError("determinate decisions may allow only complete evidence") + if self.outcome is not NeutralOutcome.INDETERMINATE and not self.required_roles: + raise FamilyContractError("determinate decisions require minimum evidence roles") + + +@dataclass(frozen=True) +class ResidualQuestion: + question_id: str + required_roles: tuple[str, ...] + allowed_decisions: tuple[str, ...] + + def __post_init__(self) -> None: + _text(self.question_id, "question_id") + _unique(self.required_roles, "required roles") + _unique(self.allowed_decisions, "allowed decisions") + if not self.required_roles or not self.allowed_decisions: + raise FamilyContractError("residual questions require roles and allowed decisions") + + +@dataclass(frozen=True) +class FamilyContract: + family_id: str + family_version: str + schema_version: str + ecosystems: tuple[str, ...] + security_object: ObjectSpec + roles: tuple[RoleSpec, ...] + decisions: tuple[DecisionRule, ...] + residual_questions: tuple[ResidualQuestion, ...] + indeterminate_reasons: tuple[str, ...] + + def __post_init__(self) -> None: + validate_contract(self) + + +def _field_specs(values: object, field: str, *, nonempty: bool) -> None: + _tuple(values, FieldSpec, field, nonempty=nonempty) + assert isinstance(values, tuple) + _unique(tuple(item.name for item in values), field) + + +def validate_contract(contract: FamilyContract) -> None: + if not isinstance(contract, FamilyContract): + raise FamilyContractError("contract must be a FamilyContract") + for value, field in ( + (contract.family_id, "family_id"), + (contract.family_version, "family_version"), + (contract.schema_version, "schema_version"), + ): + _text(value, field) + _unique(contract.ecosystems, "ecosystems") + if not contract.ecosystems: + raise FamilyContractError("ecosystems must be nonempty") + if not isinstance(contract.security_object, ObjectSpec): + raise FamilyContractError("security_object must be an ObjectSpec") + _tuple(contract.roles, RoleSpec, "roles") + _tuple(contract.decisions, DecisionRule, "decisions") + _tuple(contract.residual_questions, ResidualQuestion, "residual questions") + _unique(tuple(item.role for item in contract.roles), "roles") + _unique(tuple(item.family_code for item in contract.decisions), "decisions") + _unique(tuple(item.question_id for item in contract.residual_questions), "residual questions") + _unique(contract.indeterminate_reasons, "indeterminate reasons") + if not contract.indeterminate_reasons: + raise FamilyContractError("indeterminate reasons must be nonempty") + _validate_contract_references(contract) + + +def _validate_contract_references(contract: FamilyContract) -> None: + roles = {item.role for item in contract.roles} + decisions = {item.family_code for item in contract.decisions} + for rule in contract.decisions: + if not set(rule.required_roles).issubset(roles): + raise FamilyContractError("decision references an undeclared role") + for question in contract.residual_questions: + if not set(question.required_roles).issubset(roles): + raise FamilyContractError("residual question references an undeclared role") + if not set(question.allowed_decisions).issubset(decisions): + raise FamilyContractError("residual question references an undeclared decision") + + +def _field_spec_value(value: FieldSpec) -> canonical.JsonValue: + return {"name": value.name, "kind": value.kind.value, "required": value.required} + + +def _contract_value(contract: FamilyContract) -> canonical.JsonValue: + return { + "family_id": contract.family_id, + "family_version": contract.family_version, + "schema_version": contract.schema_version, + "ecosystems": contract.ecosystems, + "security_object": { + "kind": contract.security_object.kind, + "fields": tuple(_field_spec_value(item) for item in contract.security_object.fields), + "identity_fields": contract.security_object.identity_fields, + }, + "roles": tuple( + { + "role": role.role, + "fields": tuple(_field_spec_value(item) for item in role.fields), + "allowed_relation_kinds": tuple(item.value for item in role.allowed_relation_kinds), + "origin_policy": role.origin_policy.value, + } + for role in contract.roles + ), + "decisions": tuple( + { + "family_code": rule.family_code, + "outcome": rule.outcome.value, + "required_roles": rule.required_roles, + "allowed_statuses": tuple(item.value for item in rule.allowed_statuses), + } + for rule in contract.decisions + ), + "residual_questions": tuple( + { + "question_id": question.question_id, + "required_roles": question.required_roles, + "allowed_decisions": question.allowed_decisions, + } + for question in contract.residual_questions + ), + "indeterminate_reasons": contract.indeterminate_reasons, + } + + +def family_contract_digest(contract: FamilyContract) -> str: + validate_contract(contract) + return canonical.content_digest(_contract_value(contract)) + + +def _identity_field_value(value: FieldValue) -> canonical.JsonValue: + return {"name": value.name, "kind": value.kind.value, "value": value.value} + + +def derive_security_object_id(spec: ObjectSpec, values: tuple[FieldValue, ...]) -> str: + """Bind an object kind to its typed identity fields in declared order.""" + if not isinstance(spec, ObjectSpec): + raise FamilyContractError("security object must be an ObjectSpec") + _validate_fields(values, spec.fields, "security object") + actual = {value.name: value for value in values} + payload: canonical.JsonValue = { + "kind": spec.kind, + "identity_fields": tuple(_identity_field_value(actual[name]) for name in spec.identity_fields), + } + return canonical.content_digest(payload) + + +def _validate_fields(values: tuple[FieldValue, ...], specs: tuple[FieldSpec, ...], context: str) -> None: + declared = {spec.name: spec for spec in specs} + actual = {value.name: value for value in values} + unknown = set(actual) - set(declared) + if unknown: + raise FamilyContractError(f"{context} contains an undeclared field: {sorted(unknown)[0]}") + missing = {spec.name for spec in specs if spec.required} - set(actual) + if missing: + raise FamilyContractError(f"{context} is missing required field: {sorted(missing)[0]}") + for name, value in actual.items(): + if value.kind is not declared[name].kind: + raise FamilyContractError(f"{context} field {name} has the wrong kind") + + +def _role_map(contract: FamilyContract) -> dict[str, RoleSpec]: + return {item.role: item for item in contract.roles} + + +def _validate_relation(relation: SemanticRelation, roles: dict[str, RoleSpec], packet_reasons: tuple[str, ...]) -> None: + if relation.role not in roles: + raise FamilyContractError(f"undeclared semantic role: {relation.role}") + spec = roles[relation.role] + if relation.kind not in spec.allowed_relation_kinds: + raise FamilyContractError(f"relation kind {relation.kind.value} is not declared for role {relation.role}") + _validate_origin(relation.status, relation.provenance.kind, spec, relation.reasons or packet_reasons) + + +def _validate_observation(observation: Observation, roles: dict[str, RoleSpec], packet_reasons: tuple[str, ...]) -> None: + if observation.role not in roles: + raise FamilyContractError(f"undeclared observation role: {observation.role}") + spec = roles[observation.role] + _validate_fields(observation.fields, spec.fields, f"observation role {observation.role}") + _validate_origin(observation.status, observation.provenance.kind, spec, packet_reasons) + + +def _validate_origin( + status: RelationStatus, + provenance: ProvenanceKind, + spec: RoleSpec, + reasons: tuple[str, ...], +) -> None: + if spec.origin_policy is not RelationOriginPolicy.EXCLUDED_DEPENDENCY_FACT_ONLY: + return + if status is RelationStatus.COMPLETE and provenance is not ProvenanceKind.SEMANTIC_FACT: + raise FamilyContractError(f"role {spec.role} requires a reviewed semantic fact for complete evidence") + if provenance is ProvenanceKind.PRODUCT_CODE and status is not RelationStatus.COMPLETE and MISSING_REVIEWED_DEPENDENCY not in reasons: + raise FamilyContractError(f"role {spec.role} requires {MISSING_REVIEWED_DEPENDENCY}") + + +def _validate_scope(contract: FamilyContract, record: EvidenceRecord) -> None: + packet = record.packet + if packet.family_id != contract.family_id: + raise UnsupportedFamilyScope("unsupported family scope; no cross-family fallback") + if packet.schema_version != contract.schema_version: + raise UnsupportedFamilyScope("unsupported schema scope; no fallback") + if packet.ecosystem not in contract.ecosystems: + raise UnsupportedFamilyScope("unsupported ecosystem scope; no fallback") + expected = family_contract_digest(contract) + if packet.family_contract_digest != expected: + raise FamilyContractError("packet family contract digest does not match declaration") + + +def _validate_decision(contract: FamilyContract, record: EvidenceRecord) -> None: + rules = {item.family_code: item for item in contract.decisions} + if record.decision.family_code not in rules: + raise FamilyContractError("decision code is not declared by the family") + rule = rules[record.decision.family_code] + if record.decision.outcome is not rule.outcome: + raise FamilyContractError("family decision outcome does not match its declared outcome") + if record.decision.outcome is NeutralOutcome.INDETERMINATE: + unknown = set(record.decision.reasons) - set(contract.indeterminate_reasons) + if unknown: + raise FamilyContractError(f"indeterminate reason is not declared: {sorted(unknown)[0]}") + statuses: dict[str, list[RelationStatus]] = {} + for relation in record.packet.transitions: + statuses.setdefault(relation.role, []).append(relation.status) + for observation in record.packet.observations: + statuses.setdefault(observation.role, []).append(observation.status) + for role in rule.required_roles: + if role not in statuses: + raise FamilyContractError(f"required role is missing: {role}") + if any(status not in rule.allowed_statuses for status in statuses[role]): + raise FamilyContractError(f"required role has a status outside the decision rule: {role}") + + +def validate_record(contract: FamilyContract, record: EvidenceRecord) -> None: + validate_contract(contract) + if not isinstance(record, EvidenceRecord): + raise FamilyContractError("record must be an EvidenceRecord") + _validate_scope(contract, record) + _validate_fields(record.packet.family_fields, contract.security_object.fields, "security object") + expected_object_id = derive_security_object_id(contract.security_object, record.packet.family_fields) + if record.packet.security_object_id != expected_object_id: + raise FamilyContractError("security object identity does not match its declared identity fields") + roles = _role_map(contract) + for relation in record.packet.transitions: + _validate_relation(relation, roles, record.packet.completeness_reasons) + for observation in record.packet.observations: + _validate_observation(observation, roles, record.packet.completeness_reasons) + _validate_decision(contract, record) + + +__all__ = [ + "DecisionRule", + "FamilyContract", + "FamilyContractError", + "FieldSpec", + "MISSING_REVIEWED_DEPENDENCY", + "ObjectSpec", + "RelationOriginPolicy", + "ResidualQuestion", + "RoleSpec", + "UnsupportedFamilyScope", + "derive_security_object_id", + "family_contract_digest", + "validate_contract", + "validate_record", +] diff --git a/tests/test_core_family.py b/tests/test_core_family.py new file mode 100644 index 0000000..678b5aa --- /dev/null +++ b/tests/test_core_family.py @@ -0,0 +1,468 @@ +"""Closed vulnerability-family declarations and record validation.""" + +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.provenance as provenance +import openultrasast.core.semantics as semantics + + +def _digest(value: str) -> str: + return canonical.content_digest(value) + + +def _location(line: int = 10) -> semantics.SourceLocation: + return semantics.SourceLocation("src/app.py", line, 2, "handle") + + +def _product() -> semantics.EvidenceProvenance: + return semantics.EvidenceProvenance(semantics.ProvenanceKind.PRODUCT_CODE, _digest("source"), None, None, None) + + +def _fact() -> semantics.EvidenceProvenance: + content = b"reviewed match" + retention = _digest("retention") + content_digest = canonical.bytes_digest(content) + receipt_digest = canonical.derive_storage_receipt_digest( + store_identity="reviewed-store", + content_digest=content_digest, + byte_length=len(content), + retention_policy_digest=retention, + ) + reference = canonical.ContentRef( + "match", + content_digest, + "application/json", + len(content), + retention, + receipt_digest, + ) + return semantics.EvidenceProvenance( + semantics.ProvenanceKind.SEMANTIC_FACT, + _digest("source"), + _digest("pack"), + "fact.constraint", + reference, + ) + + +def _relation( + *, + status: semantics.RelationStatus = semantics.RelationStatus.COMPLETE, + origin: semantics.EvidenceProvenance | None = None, + role: str = "constraint", +) -> semantics.SemanticRelation: + return semantics.make_relation( + kind=semantics.RelationKind.CONSTRAINT_EFFECT, + role=role, + source=semantics.SemanticNode("operation", "operation", _location()), + target=semantics.SemanticNode("policy", "policy", _location(11)), + location=_location(12), + status=status, + conditions=("policy_applies",), + reasons=() if status is semantics.RelationStatus.COMPLETE else ("missing_reviewed_dependency_relation",), + provenance=_product() if origin is None else origin, + ) + + +def _observation(*, field_name: str = "resource", kind: semantics.FieldKind = semantics.FieldKind.TEXT) -> semantics.Observation: + value: canonical.JsonValue = "document" if kind is semantics.FieldKind.TEXT else 7 + return semantics.make_observation( + role="security_object", + location=_location(13), + fields=(semantics.FieldValue(field_name, kind, value),), + status=semantics.RelationStatus.COMPLETE, + provenance=_product(), + ) + + +def _contract(*, fact_only: bool = False) -> family.FamilyContract: + origin = ( + family.RelationOriginPolicy.EXCLUDED_DEPENDENCY_FACT_ONLY if fact_only else family.RelationOriginPolicy.PRODUCT_OR_REVIEWED_FACT + ) + return family.FamilyContract( + family_id="obligation", + family_version="1.0.0", + schema_version=evidence.EVIDENCE_PACKET_SCHEMA, + ecosystems=("python",), + security_object=family.ObjectSpec( + kind="operation", + fields=(family.FieldSpec("operation", semantics.FieldKind.TEXT, True),), + identity_fields=("operation",), + ), + roles=( + family.RoleSpec( + "constraint", + (), + (semantics.RelationKind.CONSTRAINT_EFFECT,), + origin, + ), + family.RoleSpec( + "security_object", + (family.FieldSpec("resource", semantics.FieldKind.TEXT, True),), + (), + family.RelationOriginPolicy.PRODUCT_OR_REVIEWED_FACT, + ), + ), + decisions=( + family.DecisionRule( + "blocked", + evidence.NeutralOutcome.VIOLATION, + ("constraint", "security_object"), + (semantics.RelationStatus.COMPLETE,), + ), + family.DecisionRule( + "safe", + evidence.NeutralOutcome.SATISFIED, + ("constraint", "security_object"), + (semantics.RelationStatus.COMPLETE,), + ), + family.DecisionRule( + "unknown", + evidence.NeutralOutcome.INDETERMINATE, + ("constraint",), + (semantics.RelationStatus.INCOMPLETE, semantics.RelationStatus.UNRESOLVED), + ), + ), + residual_questions=(family.ResidualQuestion("resolve_constraint", ("constraint",), ("blocked", "safe", "unknown")),), + indeterminate_reasons=( + "missing_reviewed_dependency_relation", + "missing_required_role", + "incomplete_required_role", + "unsupported_scope", + ), + ) + + +def _stamp(contract: family.FamilyContract) -> provenance.ProvenanceStamp: + return provenance.ProvenanceStamp( + _digest("analyzer"), + family.family_contract_digest(contract), + _digest("facts"), + _digest("acquisition"), + _digest("configuration"), + _digest("population"), + _digest("change"), + ) + + +def _packet( + contract: family.FamilyContract, + *, + relation: semantics.SemanticRelation | None = None, + observation: semantics.Observation | None = None, + family_fields: tuple[semantics.FieldValue, ...] | None = None, + ecosystem: str = "python", +) -> evidence.EvidencePacket: + selected = _relation() if relation is None else relation + reasons = selected.reasons if selected.status is not semantics.RelationStatus.COMPLETE else () + selected_fields = (semantics.FieldValue("operation", semantics.FieldKind.TEXT, "read"),) if family_fields is None else family_fields + scope = evidence.AnalyzedScope( + "base", + "head", + _digest("change"), + (evidence.SourceRegion("src/app.py", 1, 30),), + ) + return evidence.EvidencePacket( + evidence.EVIDENCE_PACKET_SCHEMA, + contract.family_id, + family.family_contract_digest(contract), + family.derive_security_object_id(contract.security_object, selected_fields), + ecosystem, + scope, + (selected,), + (_observation() if observation is None else observation,), + selected_fields, + reasons, + _stamp(contract), + ) + + +def _record( + contract: family.FamilyContract, + *, + relation: semantics.SemanticRelation | None = None, + observation: semantics.Observation | None = None, + family_fields: tuple[semantics.FieldValue, ...] | None = None, + ecosystem: str = "python", + code: str = "blocked", +) -> evidence.EvidenceRecord: + packet = _packet( + contract, + relation=relation, + observation=observation, + family_fields=family_fields, + ecosystem=ecosystem, + ) + indeterminate = relation is not None and relation.status is not semantics.RelationStatus.COMPLETE + outcome = evidence.NeutralOutcome.INDETERMINATE if indeterminate else evidence.NeutralOutcome.VIOLATION + decision = evidence.Decision( + code, + outcome, + evidence.EvidenceBasis.DETERMINISTIC, + packet.scope, + packet.completeness_reasons, + evidence.derive_packet_digest(packet), + provenance.DecisionProvenance(None, None, None, None, None, None), + ) + return evidence.EvidenceRecord(packet, decision, "") + + +def test_closed_family_contract_and_record_are_accepted() -> None: + contract = _contract() + family.validate_contract(contract) + family.validate_record(contract, _record(contract)) + assert len(family.family_contract_digest(contract)) == 64 + + +@pytest.mark.parametrize( + "field", + [ + "family_id", + "family_version", + "schema_version", + "ecosystems", + "roles", + "decisions", + "residual_questions", + "indeterminate_reasons", + ], +) +def test_contract_requires_every_declared_section(field: str) -> None: + contract = _contract() + with pytest.raises(family.FamilyContractError): + if field == "family_id": + replace(contract, family_id="") + elif field == "family_version": + replace(contract, family_version="") + elif field == "schema_version": + replace(contract, schema_version="") + elif field == "ecosystems": + replace(contract, ecosystems=()) + elif field == "roles": + replace(contract, roles=()) + elif field == "decisions": + replace(contract, decisions=()) + elif field == "residual_questions": + replace(contract, residual_questions=()) + else: + replace(contract, indeterminate_reasons=()) + + +def test_security_object_identity_fields_must_be_required_and_declared() -> None: + contract = _contract() + with pytest.raises(family.FamilyContractError, match="identity"): + replace(contract, security_object=replace(contract.security_object, identity_fields=("missing",))) + optional = family.FieldSpec("operation", semantics.FieldKind.TEXT, False) + with pytest.raises(family.FamilyContractError, match="identity"): + replace(contract, security_object=replace(contract.security_object, fields=(optional,))) + with pytest.raises(family.FamilyContractError, match="identity"): + replace(contract, security_object=replace(contract.security_object, identity_fields=())) + + +def test_role_and_decision_declarations_are_closed_and_unique() -> None: + contract = _contract() + with pytest.raises(family.FamilyContractError, match="duplicate"): + replace(contract, roles=contract.roles + (contract.roles[0],)) + with pytest.raises(family.FamilyContractError, match="undeclared role"): + replace(contract, decisions=(replace(contract.decisions[0], required_roles=("other",)),) + contract.decisions[1:]) + with pytest.raises(family.FamilyContractError, match="complete"): + replace( + contract, + decisions=(replace(contract.decisions[0], allowed_statuses=(semantics.RelationStatus.INCOMPLETE,)),) + contract.decisions[1:], + ) + with pytest.raises(family.FamilyContractError, match="minimum evidence"): + replace(contract, decisions=(replace(contract.decisions[0], required_roles=()),) + contract.decisions[1:]) + + +def test_residual_questions_reference_only_declared_roles_and_decisions() -> None: + contract = _contract() + with pytest.raises(family.FamilyContractError, match="undeclared role"): + replace(contract, residual_questions=(replace(contract.residual_questions[0], required_roles=("other",)),)) + with pytest.raises(family.FamilyContractError, match="undeclared decision"): + replace(contract, residual_questions=(replace(contract.residual_questions[0], allowed_decisions=("other",)),)) + with pytest.raises(family.FamilyContractError, match="require roles"): + replace(contract, residual_questions=(replace(contract.residual_questions[0], required_roles=()),)) + + +def test_declaration_collections_require_immutable_tuples() -> None: + contract = _contract() + with pytest.raises(family.FamilyContractError, match="ecosystems"): + replace(contract, ecosystems=cast(tuple[str, ...], ["python"])) + with pytest.raises(family.FamilyContractError, match="indeterminate reasons"): + replace(contract, indeterminate_reasons=cast(tuple[str, ...], ["unsupported_scope"])) + + +def test_record_refuses_unsupported_family_ecosystem_and_schema_without_fallback() -> None: + contract = _contract() + record = _record(contract) + with pytest.raises(family.UnsupportedFamilyScope, match="family"): + family.validate_record(replace(contract, family_id="different"), record) + with pytest.raises(family.UnsupportedFamilyScope, match="ecosystem"): + family.validate_record(contract, _record(contract, ecosystem="javascript")) + with pytest.raises(family.UnsupportedFamilyScope, match="schema"): + family.validate_record(replace(contract, schema_version="other-schema"), record) + + +def test_record_refuses_unknown_and_mistyped_family_fields() -> None: + contract = _contract() + unknown = (semantics.FieldValue("other", semantics.FieldKind.TEXT, "read"),) + wrong_type = (semantics.FieldValue("operation", semantics.FieldKind.INTEGER, 7),) + with pytest.raises(family.FamilyContractError, match="field"): + family.validate_record(contract, _record(contract, family_fields=unknown)) + with pytest.raises(family.FamilyContractError, match="kind"): + family.validate_record(contract, _record(contract, family_fields=wrong_type)) + with pytest.raises(family.FamilyContractError, match="required"): + family.validate_record(contract, _record(contract, family_fields=())) + + +def _record_with_packet(record: evidence.EvidenceRecord, packet: evidence.EvidencePacket) -> evidence.EvidenceRecord: + decision = replace(record.decision, packet_digest=evidence.derive_packet_digest(packet)) + return evidence.EvidenceRecord(packet, decision, "") + + +def test_record_refuses_substituted_security_object_identity() -> None: + contract = _contract() + record = _record(contract) + substituted = replace(record.packet, security_object_id=_digest("other-object")) + with pytest.raises(family.FamilyContractError, match="security object identity"): + family.validate_record(contract, _record_with_packet(record, substituted)) + + +def test_identity_field_change_cannot_reuse_the_old_security_object_identity() -> None: + contract = _contract() + record = _record(contract) + changed = replace( + record.packet, + family_fields=(semantics.FieldValue("operation", semantics.FieldKind.TEXT, "write"),), + ) + with pytest.raises(family.FamilyContractError, match="security object identity"): + family.validate_record(contract, _record_with_packet(record, changed)) + + +def test_nonidentity_field_change_does_not_redirect_security_object_identity() -> None: + contract = _contract() + display = family.FieldSpec("display_name", semantics.FieldKind.TEXT, False) + extended = replace( + contract, + security_object=replace(contract.security_object, fields=contract.security_object.fields + (display,)), + ) + first = _record( + extended, + family_fields=( + semantics.FieldValue("operation", semantics.FieldKind.TEXT, "read"), + semantics.FieldValue("display_name", semantics.FieldKind.TEXT, "first"), + ), + ) + second = _record( + extended, + family_fields=( + semantics.FieldValue("operation", semantics.FieldKind.TEXT, "read"), + semantics.FieldValue("display_name", semantics.FieldKind.TEXT, "second"), + ), + ) + family.validate_record(extended, first) + family.validate_record(extended, second) + assert first.packet.security_object_id == second.packet.security_object_id + + +def test_security_object_identity_uses_declared_field_order_not_input_order() -> None: + spec = family.ObjectSpec( + "operation", + ( + family.FieldSpec("operation", semantics.FieldKind.TEXT, True), + family.FieldSpec("arity", semantics.FieldKind.INTEGER, True), + ), + ("operation", "arity"), + ) + values = ( + semantics.FieldValue("operation", semantics.FieldKind.TEXT, "read"), + semantics.FieldValue("arity", semantics.FieldKind.INTEGER, 1), + ) + assert family.derive_security_object_id(spec, values) == family.derive_security_object_id(spec, tuple(reversed(values))) + reversed_spec = replace(spec, identity_fields=("arity", "operation")) + assert family.derive_security_object_id(spec, values) != family.derive_security_object_id(reversed_spec, values) + + +def test_record_refuses_unknown_roles_relation_kinds_and_observation_fields() -> None: + contract = _contract() + with pytest.raises(family.FamilyContractError, match="role"): + family.validate_record(contract, _record(contract, relation=_relation(role="other"))) + wrong_kind = replace(_relation(), kind=semantics.RelationKind.CALL, relation_id="") + with pytest.raises(family.FamilyContractError, match="kind"): + family.validate_record(contract, _record(contract, relation=wrong_kind)) + with pytest.raises(family.FamilyContractError, match="field"): + family.validate_record(contract, _record(contract, observation=_observation(field_name="other"))) + + +def test_record_refuses_unknown_decision_and_outcome_mapping() -> None: + contract = _contract() + with pytest.raises(family.FamilyContractError, match="decision"): + family.validate_record(contract, _record(contract, code="other")) + record = _record(contract) + wrong = replace(record, decision=replace(record.decision, outcome=evidence.NeutralOutcome.SATISFIED), record_id="") + with pytest.raises(family.FamilyContractError, match="outcome"): + family.validate_record(contract, wrong) + + +def test_indeterminate_reasons_must_be_declared() -> None: + contract = _contract() + relation = _relation(status=semantics.RelationStatus.INCOMPLETE) + record = _record(contract, relation=relation, code="unknown") + family.validate_record(contract, record) + changed = replace(contract, indeterminate_reasons=("other",)) + with pytest.raises(family.FamilyContractError, match="reason"): + family.validate_record(changed, _record(changed, relation=relation, code="unknown")) + + +def test_determinate_rule_requires_every_role_and_allowed_status() -> None: + contract = _contract() + missing = replace(_record(contract).packet, observations=()) + decision = replace(_record(contract).decision, packet_digest=evidence.derive_packet_digest(missing)) + record = evidence.EvidenceRecord(missing, decision, "") + with pytest.raises(family.FamilyContractError, match="required role"): + family.validate_record(contract, record) + + +def test_excluded_dependency_complete_relation_requires_reviewed_fact() -> None: + contract = _contract(fact_only=True) + with pytest.raises(family.FamilyContractError, match="reviewed semantic fact"): + family.validate_record(contract, _record(contract)) + family.validate_record(contract, _record(contract, relation=_relation(origin=_fact()))) + + +def test_missing_reviewed_dependency_is_named_incomplete_not_product_complete() -> None: + contract = _contract(fact_only=True) + relation = _relation(status=semantics.RelationStatus.INCOMPLETE) + family.validate_record(contract, _record(contract, relation=relation, code="unknown")) + wrong_reason = replace(relation, reasons=("different",), relation_id="") + packet = replace( + _packet(contract, relation=wrong_reason), + completeness_reasons=("different",), + ) + decision = evidence.Decision( + "unknown", + evidence.NeutralOutcome.INDETERMINATE, + evidence.EvidenceBasis.DETERMINISTIC, + packet.scope, + packet.completeness_reasons, + evidence.derive_packet_digest(packet), + provenance.DecisionProvenance(None, None, None, None, None, None), + ) + with pytest.raises(family.FamilyContractError, match="missing_reviewed_dependency_relation"): + family.validate_record(contract, evidence.EvidenceRecord(packet, decision, "")) + + +def test_contract_digest_changes_with_every_declaration_surface() -> None: + contract = _contract() + variants = ( + replace(contract, family_version="1.0.1"), + replace(contract, ecosystems=("python", "java")), + replace(contract, indeterminate_reasons=contract.indeterminate_reasons + ("new_reason",)), + ) + assert len({family.family_contract_digest(contract), *(family.family_contract_digest(item) for item in variants)}) == 4