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