diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 2a02657..91101da 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -263,7 +263,7 @@ - Keep durable append, compare-and-swap execution, label release, persistence, crash handling, and concurrency outside core. - Done when only the exact spent successor authorizes scoring and every other receipt shape is refused without an external side effect. - _Requirements: 12.2, 13.4_ -- [ ] 8.7 Validate the authenticated post-spend case release +- [x] 8.7 Validate the authenticated post-spend case release - Define canonical released-case bindings over case identity, scope, role, scoped pair identity, exact result digest, evidence-input digests, change-input digests, and input-manifest digest. - Bind the release to the population, manifest, authorization, preparation, exact current-chain spend receipt, custodian identity, and an independent review attestation. - Validate the release only after the exact spend successor and require one canonical binding for every supplied case and no others. diff --git a/src/openultrasast/core/_admission_case_shape.py b/src/openultrasast/core/_admission_case_shape.py new file mode 100644 index 0000000..d9bcf87 --- /dev/null +++ b/src/openultrasast/core/_admission_case_shape.py @@ -0,0 +1,138 @@ +"""Closed object-graph whitelist for immutable admission cases.""" + +from __future__ import annotations + +from openultrasast.core._closed_graph import closed_graph_problem +from openultrasast.core._model_result import ModelResult, ModelStatus, PreservedModelResponse +from openultrasast.core.budget import BudgetDerivation, ChangeComplexity, ReferenceHardware, ReferenceWorkload +from openultrasast.core.canonical import ContentRef +from openultrasast.core.coverage import CoverageFunnel, InputManifest, InputReadReceipt, ReasonCount, RelationStatusCount +from openultrasast.core.custody import LineagePurpose +from openultrasast.core.evidence import ( + AnalyzedScope, + Decision, + EvidenceBasis, + EvidencePacket, + EvidenceRecord, + NeutralOutcome, + SourceRegion, +) +from openultrasast.core.family import ( + DecisionRule, + FamilyContract, + FieldSpec, + ObjectSpec, + RelationOriginPolicy, + ResidualQuestion, + RoleSpec, +) +from openultrasast.core.lineage import LineageManifest, LineageUse +from openultrasast.core.measurement import ScopeKey +from openultrasast.core.model import CompletenessItem, ModelProgramDeclaration, ModelRequest +from openultrasast.core.proof import ( + ControlBinding, + DifferentialProofRecord, + ProofAttachment, + ProofOutcome, + ProofProducerIdentity, + ProofSideResult, + ProofVerificationReceipt, +) +from openultrasast.core.provenance import ( + DecisionProvenance, + ModelCallIdentity, + NamedInput, + NamedInputs, + ProvenanceInputRole, + ProvenanceStamp, +) +from openultrasast.core.semantics import ( + EvidenceProvenance, + FieldKind, + FieldValue, + Observation, + ProvenanceKind, + RelationKind, + RelationStatus, + SemanticNode, + SemanticRelation, + SourceLocation, +) +from openultrasast.core.validation import EvidenceValidationInput +from openultrasast.core.verdict_v2 import BoundedSatisfaction, BoundedSilence, DefectClaim + +_DATACLASSES: tuple[type[object], ...] = ( + ScopeKey, + EvidenceValidationInput, + EvidenceRecord, + EvidencePacket, + Decision, + AnalyzedScope, + SourceRegion, + SourceLocation, + SemanticNode, + EvidenceProvenance, + FieldValue, + SemanticRelation, + Observation, + ProvenanceStamp, + DecisionProvenance, + ModelCallIdentity, + NamedInput, + NamedInputs, + ContentRef, + ModelRequest, + ModelProgramDeclaration, + CompletenessItem, + ModelResult, + PreservedModelResponse, + FamilyContract, + FieldSpec, + ObjectSpec, + RoleSpec, + DecisionRule, + ResidualQuestion, + LineageManifest, + LineageUse, + DefectClaim, + BoundedSatisfaction, + BoundedSilence, + ProofAttachment, + DifferentialProofRecord, + ProofProducerIdentity, + ProofSideResult, + ControlBinding, + ProofVerificationReceipt, + CoverageFunnel, + RelationStatusCount, + ReasonCount, + InputManifest, + InputReadReceipt, + ChangeComplexity, + BudgetDerivation, + ReferenceWorkload, + ReferenceHardware, +) + +_ENUMS: tuple[type[object], ...] = ( + NeutralOutcome, + EvidenceBasis, + FieldKind, + RelationKind, + RelationStatus, + ProvenanceKind, + RelationOriginPolicy, + ProvenanceInputRole, + LineagePurpose, + ModelStatus, + ProofOutcome, +) + + +def admission_case_problem(value: object, case_type: type[object], role_type: type[object]) -> str | None: + return closed_graph_problem( + value, + dataclass_types=(case_type, *_DATACLASSES), + enum_types=(role_type, *_ENUMS), + path="admission case result", + ) diff --git a/src/openultrasast/core/_admission_release.py b/src/openultrasast/core/_admission_release.py new file mode 100644 index 0000000..6c03803 --- /dev/null +++ b/src/openultrasast/core/_admission_release.py @@ -0,0 +1,299 @@ +"""Validation helpers for authenticated post-spend admission-case releases.""" + +from __future__ import annotations + +import re +from collections import Counter +from collections.abc import Sequence + +from openultrasast.core import canonical +from openultrasast.core.admission_case import ( + ADMISSION_CASE_RELEASE_SCHEMA, + AdmissionCaseRelease, + AdmissionCaseReleaseError, + AdmissionCaseResult, + AdmissionCaseRole, + ReleasedCaseBinding, + derive_admission_case_release_digest, + derive_admission_case_release_payload_digest, + derive_admission_case_result_digest, + derive_released_case_binding_digest, + validate_admission_case_shape, +) +from openultrasast.core.admission_context import AdmissionValidationContext +from openultrasast.core.admission_review import AdmissionManifestV2 +from openultrasast.core.admission_spend import AdmissionAuthorization, AdmissionPreparation +from openultrasast.core.coverage import derive_input_manifest_digest, validate_instrument +from openultrasast.core.custody import PopulationCustodyPort, PopulationUseReceipt +from openultrasast.core.factpack import ( + FactPackValidationError, + IndependenceAttestation, + ReviewIndependencePolicy, + derive_independence_attestation_digest, + derive_review_independence_policy_digest, +) +from openultrasast.core.measurement import ScopeKey +from openultrasast.core.provenance import ProvenanceInputRole +from openultrasast.core.validation import derive_evidence_validation_input_digest, validate_evidence_input + +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +def _digest(value: object, field: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise AdmissionCaseReleaseError(f"{field} must be a full lowercase SHA-256 digest") + + +def _digest_tuple(value: object, field: str, *, nonempty: bool = True) -> None: + if type(value) is not tuple or (nonempty and not value): + raise AdmissionCaseReleaseError(f"{field} must be a{' nonempty' if nonempty else ''} exact tuple") + for item in value: + _digest(item, field) + if value != tuple(sorted(set(value))): + raise AdmissionCaseReleaseError(f"{field} must be unique and canonically ordered") + + +def _validate_policy(policy: ReviewIndependencePolicy) -> None: + try: + policy.__post_init__() + except FactPackValidationError as exc: + raise AdmissionCaseReleaseError(f"release review policy is invalid: {exc}") from exc + 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 AdmissionCaseReleaseError("release review policy digest cannot be re-derived") + + +def _validate_review( + release: AdmissionCaseRelease, + receipt: PopulationUseReceipt, + policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, +) -> None: + review = release.review + if type(review) is not IndependenceAttestation: + raise AdmissionCaseReleaseError("release review must be an exact independence attestation") + try: + review.__post_init__() + review.conflict_disclosure.__post_init__() + except (FactPackValidationError, ValueError) as exc: + raise AdmissionCaseReleaseError(f"release review is invalid: {exc}") from exc + if review.subject_digest != release.release_payload_digest: + raise AdmissionCaseReleaseError("release review subject does not match payload digest") + if review.producer_identity != release.custodian_identity or review.producer_identity != receipt.custodian_identity: + raise AdmissionCaseReleaseError("release review producer must be the exact spend custodian") + if review.producer_identity == review.reviewer_identity: + raise AdmissionCaseReleaseError("release review producer and reviewer must be independent") + if review.reviewer_identity not in policy.reviewed_reviewer_digests: + raise AdmissionCaseReleaseError("release reviewer is not admitted by policy") + if (review.producer_identity, review.reviewer_identity) in policy.forbidden_identity_pairs: + raise AdmissionCaseReleaseError("release producer/reviewer pair is forbidden") + if review.attestation_digest != derive_independence_attestation_digest(review): + raise AdmissionCaseReleaseError("release review attestation digest cannot be re-derived") + content = canonical.resolve_content( + review.conflict_disclosure, + resolver, + trusted_store_identities=policy.trusted_store_identities, + maximum_bytes=policy.maximum_evidence_bytes, + ) + if not content: + raise AdmissionCaseReleaseError("release conflict disclosure must be nonempty") + + +def _validate_binding_shape(binding: ReleasedCaseBinding) -> None: + if type(binding) is not ReleasedCaseBinding: + raise AdmissionCaseReleaseError("release binding must use the exact contract type") + if type(binding.scope) is not ScopeKey or type(binding.role) is not AdmissionCaseRole: + raise AdmissionCaseReleaseError("release binding scope and role must use exact contract types") + binding.scope.__post_init__() + for value, field in ( + (binding.case_id_digest, "case identity digest"), + (binding.case_result_digest, "case result digest"), + (binding.input_manifest_digest, "input manifest digest"), + (binding.binding_digest, "binding digest"), + ): + _digest(value, field) + if binding.role is AdmissionCaseRole.ORDINARY: + if binding.pair_id_digest is not None: + raise AdmissionCaseReleaseError("ordinary release binding cannot carry a pair identity") + else: + _digest(binding.pair_id_digest, "pair identity digest") + _digest_tuple(binding.evidence_input_digests, "evidence input digests") + _digest_tuple(binding.change_input_digests, "change input digests") + if binding.binding_digest != derive_released_case_binding_digest(binding): + raise AdmissionCaseReleaseError("release binding digest cannot be re-derived") + + +def _validate_pair_integrity(bindings: tuple[ReleasedCaseBinding, ...]) -> None: + pairs: dict[tuple[str, str, str], Counter[AdmissionCaseRole]] = {} + for binding in bindings: + if binding.pair_id_digest is None: + continue + key = (binding.scope.family_id, binding.scope.ecosystem, binding.pair_id_digest) + pairs.setdefault(key, Counter())[binding.role] += 1 + expected = Counter({AdmissionCaseRole.VULNERABLE: 1, AdmissionCaseRole.FIXED: 1}) + if any(counts != expected for counts in pairs.values()): + raise AdmissionCaseReleaseError("every scoped release pair must contain one vulnerable and one fixed case") + + +def _validate_release_shape(release: AdmissionCaseRelease) -> None: + if type(release) is not AdmissionCaseRelease or release.schema_version != ADMISSION_CASE_RELEASE_SCHEMA: + raise AdmissionCaseReleaseError("unknown or substituted admission case-release schema") + for value, field in ( + (release.population_digest, "population digest"), + (release.manifest_digest, "manifest digest"), + (release.authorization_digest, "authorization digest"), + (release.preparation_digest, "preparation digest"), + (release.label_release_receipt_digest, "label release receipt digest"), + (release.custodian_identity, "custodian identity"), + (release.release_payload_digest, "release payload digest"), + (release.release_digest, "release digest"), + ): + _digest(value, field) + if type(release.bindings) is not tuple or not release.bindings: + raise AdmissionCaseReleaseError("release bindings must be a nonempty exact tuple") + for binding in release.bindings: + _validate_binding_shape(binding) + case_ids = tuple(item.case_id_digest for item in release.bindings) + if case_ids != tuple(sorted(case_ids)) or len(case_ids) != len(set(case_ids)): + raise AdmissionCaseReleaseError("release bindings must have unique canonical case identities") + _validate_pair_integrity(release.bindings) + if release.release_payload_digest != derive_admission_case_release_payload_digest(release): + raise AdmissionCaseReleaseError("release payload digest cannot be re-derived") + if release.release_digest != derive_admission_case_release_digest(release): + raise AdmissionCaseReleaseError("release digest cannot be re-derived") + + +def _validate_header( + manifest: AdmissionManifestV2, + preparation: AdmissionPreparation, + receipt: PopulationUseReceipt, + authorization: AdmissionAuthorization, + release: AdmissionCaseRelease, +) -> None: + expected = ( + manifest.identity.population_digest, + manifest.manifest_digest, + authorization.authorization_digest, + preparation.preparation_digest, + receipt.receipt_digest, + receipt.custodian_identity, + ) + actual = ( + release.population_digest, + release.manifest_digest, + release.authorization_digest, + release.preparation_digest, + release.label_release_receipt_digest, + release.custodian_identity, + ) + if actual != expected: + fields = ( + "population digest", + "manifest digest", + "authorization digest", + "preparation digest", + "label release receipt digest", + "custodian", + ) + changed = next(field for field, left, right in zip(fields, actual, expected, strict=True) if left != right) + raise AdmissionCaseReleaseError(f"release {changed} does not match the exact spent admission") + + +def _validated_case_inputs( + case: AdmissionCaseResult, + context: AdmissionValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> tuple[tuple[str, ...], tuple[str, ...]]: + evidence: list[str] = [] + changes: set[str] = set() + for value in case.evidence_inputs: + if value.input_digest != derive_evidence_validation_input_digest(value): + raise AdmissionCaseReleaseError("case evidence input digest cannot be re-derived") + validate_evidence_input(value, context.evidence, custody, resolver) + evidence.append(value.input_digest) + matches = tuple(item.identity for item in value.named_inputs.inputs if item.role is ProvenanceInputRole.CHANGE_INPUT) + if len(matches) != 1 or matches[0] != value.record.packet.provenance.change_input_digest: + raise AdmissionCaseReleaseError("case change input cannot be re-derived from validated provenance") + changes.add(matches[0]) + ordered_evidence = tuple(sorted(evidence)) + ordered_changes = tuple(sorted(changes)) + _digest_tuple(ordered_evidence, "case evidence input digests") + _digest_tuple(ordered_changes, "case change input digests") + return ordered_evidence, ordered_changes + + +def _binding_for_case( + case: AdmissionCaseResult, + context: AdmissionValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> ReleasedCaseBinding: + validate_admission_case_shape(case) + expected_manifest = derive_input_manifest_digest(case.input_manifest.tracked_input_ids) + if case.input_manifest.manifest_digest != expected_manifest: + raise AdmissionCaseReleaseError("case input manifest digest cannot be re-derived") + validate_instrument(case.input_manifest, case.input_receipts) + if case.result_digest != derive_admission_case_result_digest(case): + raise AdmissionCaseReleaseError("case result digest cannot be re-derived") + evidence, changes = _validated_case_inputs(case, context, custody, resolver) + draft = ReleasedCaseBinding( + case.case_id_digest, + case.scope, + case.role, + case.pair_id_digest, + case.result_digest, + evidence, + changes, + case.input_manifest.manifest_digest, + "0" * 64, + ) + return draft + + +def _canonical_binding( + case: AdmissionCaseResult, + context: AdmissionValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> ReleasedCaseBinding: + draft = _binding_for_case(case, context, custody, resolver) + return ReleasedCaseBinding( + draft.case_id_digest, + draft.scope, + draft.role, + draft.pair_id_digest, + draft.case_result_digest, + draft.evidence_input_digests, + draft.change_input_digests, + draft.input_manifest_digest, + derive_released_case_binding_digest(draft), + ) + + +def validate_case_bindings( + release: AdmissionCaseRelease, + cases: Sequence[AdmissionCaseResult], + context: AdmissionValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> None: + selected = tuple(cases) + if not selected: + raise AdmissionCaseReleaseError("case population must be nonempty") + case_ids = tuple(case.case_id_digest for case in selected) + if len(case_ids) != len(set(case_ids)): + raise AdmissionCaseReleaseError("case population contains duplicate identities") + expected = tuple( + sorted( + (_canonical_binding(case, context, custody, resolver) for case in selected), + key=lambda item: item.case_id_digest, + ) + ) + if release.bindings != expected: + raise AdmissionCaseReleaseError("authenticated release binding does not match the exact case population") diff --git a/src/openultrasast/core/admission_case.py b/src/openultrasast/core/admission_case.py new file mode 100644 index 0000000..82ffc9d --- /dev/null +++ b/src/openultrasast/core/admission_case.py @@ -0,0 +1,280 @@ +"""Immutable admission-case contracts and canonical commitments.""" + +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass +from enum import Enum + +from openultrasast.core import canonical +from openultrasast.core._admission_case_shape import admission_case_problem +from openultrasast.core.admission_context import AdmissionValidationContext +from openultrasast.core.admission_review import AdmissionManifestV2 +from openultrasast.core.admission_spend import AdmissionPreparation, validate_pre_label_spend +from openultrasast.core.budget import BudgetDerivation, ChangeComplexity +from openultrasast.core.coverage import CoverageFunnel, InputManifest, InputReadReceipt +from openultrasast.core.custody import PopulationCustodyPort, PopulationUseReceipt +from openultrasast.core.factpack import IndependenceAttestation +from openultrasast.core.measurement import DenominatorPolicy, ScopeKey +from openultrasast.core.proof import ProofAttachment +from openultrasast.core.readiness import ReadinessResult +from openultrasast.core.validation import EvidenceValidationInput +from openultrasast.core.verdict_v2 import BoundedSatisfaction, BoundedSilence, DefectClaim + +ADMISSION_CASE_RELEASE_SCHEMA = "openultrasast.detection-core.admission-case-release.v2" +ADMISSION_CASE_RELEASE_PAYLOAD_SCHEMA = "openultrasast.detection-core.admission-case-release-payload.v2" +RELEASED_CASE_BINDING_SCHEMA = "openultrasast.detection-core.released-case-binding.v2" + + +class AdmissionCaseReleaseError(ValueError): + """The post-spend case release cannot authorize admission scoring.""" + + +class AdmissionCaseRole(Enum): + VULNERABLE = "vulnerable" + FIXED = "fixed" + ORDINARY = "ordinary" + + +@dataclass(frozen=True) +class AdmissionCaseResult: + case_id_digest: str + scope: ScopeKey + role: AdmissionCaseRole + pair_id_digest: str | None + evidence_inputs: tuple[EvidenceValidationInput, ...] + defect_claims: tuple[DefectClaim, ...] + satisfactions: tuple[BoundedSatisfaction, ...] + silence: tuple[BoundedSilence, ...] + proof_attachments: tuple[ProofAttachment, ...] + coverage: CoverageFunnel + input_manifest: InputManifest + input_receipts: tuple[InputReadReceipt, ...] + complexity: ChangeComplexity + budget: BudgetDerivation + total_reply_ms: int + result_digest: str + + +@dataclass(frozen=True) +class ReleasedCaseBinding: + case_id_digest: str + scope: ScopeKey + role: AdmissionCaseRole + pair_id_digest: str | None + case_result_digest: str + evidence_input_digests: tuple[str, ...] + change_input_digests: tuple[str, ...] + input_manifest_digest: str + binding_digest: str + + +@dataclass(frozen=True) +class AdmissionCaseRelease: + schema_version: str + population_digest: str + manifest_digest: str + authorization_digest: str + preparation_digest: str + label_release_receipt_digest: str + custodian_identity: str + bindings: tuple[ReleasedCaseBinding, ...] + release_payload_digest: str + review: IndependenceAttestation + release_digest: str + + +def _coverage_value(value: CoverageFunnel) -> canonical.JsonValue: + return { + "inputs_read": value.inputs_read, + "questions_formed": value.questions_formed, + "sites_considered": value.sites_considered, + "sites_recognized": value.sites_recognized, + "transitions_materialized": value.transitions_materialized, + "security_objects_constructed": value.security_objects_constructed, + "constraints_evaluated": value.constraints_evaluated, + "determinate_outcomes": value.determinate_outcomes, + "findings_emitted": value.findings_emitted, + "relation_status_counts": tuple((item.status.value, item.count) for item in value.relation_status_counts), + "unresolved_by_reason": tuple((item.reason, item.count) for item in value.unresolved_by_reason), + } + + +def derive_case_coverage_digest(case_id_digest: str, scope: ScopeKey, coverage: CoverageFunnel) -> str: + return canonical.content_digest( + { + "case_id_digest": case_id_digest, + "scope": {"family_id": scope.family_id, "ecosystem": scope.ecosystem}, + "coverage": _coverage_value(coverage), + } + ) + + +def _claim_value(value: DefectClaim) -> canonical.JsonValue: + location = value.location + return { + "record_id": value.record.record_id, + "basis": value.basis.value, + "location": {"path": location.path, "line": location.line, "column": location.column, "symbol": location.symbol}, + "anchor": (value.change_anchor.path, value.change_anchor.start_line, value.change_anchor.end_line), + } + + +def derive_admission_case_result_digest(value: AdmissionCaseResult) -> str: + return canonical.content_digest( + { + "case_id_digest": value.case_id_digest, + "scope": (value.scope.family_id, value.scope.ecosystem), + "role": value.role.value, + "pair_id_digest": value.pair_id_digest, + "evidence_input_digests": tuple(item.input_digest for item in value.evidence_inputs), + "defect_claims": tuple(_claim_value(item) for item in value.defect_claims), + "satisfactions": tuple((item.question_ids, tuple(record.record_id for record in item.records)) for item in value.satisfactions), + "silence": tuple( + (item.question_ids, item.reasons, item.evidence_record_digests, item.coverage_digest) for item in value.silence + ), + "proof_attachment_ids": tuple(item.proof_id for item in value.proof_attachments), + "coverage": _coverage_value(value.coverage), + "input_manifest_digest": value.input_manifest.manifest_digest, + "input_receipts": tuple( + (item.input_id, item.path, item.opened, item.byte_count, item.content_digest) for item in value.input_receipts + ), + "complexity": ( + value.complexity.change_id, + value.complexity.workload_class, + value.complexity.changed_regions, + value.complexity.transition_count, + value.complexity.changed_bytes, + value.complexity.measurement.content_digest, + ), + "budget_derivation_digest": value.budget.derivation_digest, + "total_reply_ms": value.total_reply_ms, + } + ) + + +def validate_admission_case_shape(value: AdmissionCaseResult) -> None: + problem = admission_case_problem(value, AdmissionCaseResult, AdmissionCaseRole) + if problem is not None: + raise ValueError(f"closed admission case violation: {problem}") + + +def create_admission_case_result(**values: object) -> AdmissionCaseResult: + draft = AdmissionCaseResult(**values, result_digest="0" * 64) # type: ignore[arg-type] + validate_admission_case_shape(draft) + return AdmissionCaseResult(**values, result_digest=derive_admission_case_result_digest(draft)) # type: ignore[arg-type] + + +def _scope_value(scope: ScopeKey) -> dict[str, canonical.JsonValue]: + return {"family_id": scope.family_id, "ecosystem": scope.ecosystem} + + +def _binding_value(binding: ReleasedCaseBinding) -> dict[str, canonical.JsonValue]: + return { + "schema_version": RELEASED_CASE_BINDING_SCHEMA, + "case_id_digest": binding.case_id_digest, + "scope": _scope_value(binding.scope), + "role": binding.role.value, + "pair_id_digest": binding.pair_id_digest, + "case_result_digest": binding.case_result_digest, + "evidence_input_digests": binding.evidence_input_digests, + "change_input_digests": binding.change_input_digests, + "input_manifest_digest": binding.input_manifest_digest, + } + + +def derive_released_case_binding_digest(binding: ReleasedCaseBinding) -> str: + return canonical.content_digest(_binding_value(binding)) + + +def _content_ref_value(reference: canonical.ContentRef) -> dict[str, 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 _review_value(review: IndependenceAttestation) -> dict[str, canonical.JsonValue]: + return { + "subject_digest": review.subject_digest, + "producer_identity": review.producer_identity, + "reviewer_identity": review.reviewer_identity, + "conflict_disclosure": _content_ref_value(review.conflict_disclosure), + "attestation_digest": review.attestation_digest, + } + + +def _release_payload_value(release: AdmissionCaseRelease) -> dict[str, canonical.JsonValue]: + return { + "schema_version": ADMISSION_CASE_RELEASE_PAYLOAD_SCHEMA, + "release_schema_version": release.schema_version, + "population_digest": release.population_digest, + "manifest_digest": release.manifest_digest, + "authorization_digest": release.authorization_digest, + "preparation_digest": release.preparation_digest, + "label_release_receipt_digest": release.label_release_receipt_digest, + "custodian_identity": release.custodian_identity, + "bindings": tuple({**_binding_value(item), "binding_digest": item.binding_digest} for item in release.bindings), + } + + +def derive_admission_case_release_payload_digest(release: AdmissionCaseRelease) -> str: + return canonical.content_digest(_release_payload_value(release)) + + +def derive_admission_case_release_digest(release: AdmissionCaseRelease) -> str: + return canonical.content_digest( + { + "schema_version": ADMISSION_CASE_RELEASE_SCHEMA, + "release_payload_digest": release.release_payload_digest, + "review": _review_value(release.review), + } + ) + + +def validate_admission_case_release( + manifest: AdmissionManifestV2, + denominator: DenominatorPolicy, + readiness: Sequence[ReadinessResult], + context: AdmissionValidationContext, + preparation: AdmissionPreparation, + receipt: PopulationUseReceipt, + release: AdmissionCaseRelease, + cases: Sequence[AdmissionCaseResult], + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> AdmissionCaseRelease: + """Validate the custodian's exact post-spend case-to-operational-input mapping.""" + from openultrasast.core._admission_release import ( + _validate_header, + _validate_policy, + _validate_release_shape, + _validate_review, + validate_case_bindings, + ) + + try: + authorization = validate_pre_label_spend( + manifest, + denominator, + readiness, + context, + preparation, + receipt, + custody, + resolver, + ) + _validate_policy(context.admission_review_policy) + _validate_release_shape(release) + _validate_header(manifest, preparation, receipt, authorization, release) + _validate_review(release, receipt, context.admission_review_policy, resolver) + validate_case_bindings(release, cases, context, custody, resolver) + return release + except AdmissionCaseReleaseError: + raise + except (AttributeError, LookupError, TypeError, ValueError) as exc: + raise AdmissionCaseReleaseError(f"admission case release refused: {exc}") from exc diff --git a/tests/test_core_admission_release.py b/tests/test_core_admission_release.py new file mode 100644 index 0000000..dde1227 --- /dev/null +++ b/tests/test_core_admission_release.py @@ -0,0 +1,411 @@ +"""Authenticated post-spend case-release contract (12.2, 12.5, 12.6, 13.4, 15.1).""" + +from __future__ import annotations + +from dataclasses import fields, replace +from typing import NamedTuple + +import pytest +from detection_core_fixtures import InMemoryContentResolver +from test_core_admission_context import AdmissionCase, _attestation +from test_core_admission_spend import _feasible_case, _prepare, _spend +from test_core_consumer_v2 import _request +from test_core_model import _record as _model_record +from test_core_validation_v2 import _named_inputs, _with_fact_set + +from openultrasast.core import evidence, validation +from openultrasast.core.admission_case import ( + ADMISSION_CASE_RELEASE_SCHEMA, + AdmissionCaseRelease, + AdmissionCaseReleaseError, + AdmissionCaseResult, + AdmissionCaseRole, + ReleasedCaseBinding, + derive_admission_case_release_digest, + derive_admission_case_release_payload_digest, + derive_admission_case_result_digest, + derive_released_case_binding_digest, + validate_admission_case_release, +) +from openultrasast.core.admission_spend import AdmissionPreparation +from openultrasast.core.canonical import content_digest +from openultrasast.core.coverage import ( + CoverageFunnel, + InputManifest, + InputReadReceipt, + RelationStatusCount, + derive_input_manifest_digest, +) +from openultrasast.core.custody import PopulationUseReceipt +from openultrasast.core.measurement import ScopeKey +from openultrasast.core.semantics import RelationStatus + + +class _ReleaseSetup(NamedTuple): + admission: AdmissionCase + preparation: AdmissionPreparation + receipt: PopulationUseReceipt + cases: tuple[AdmissionCaseResult, ...] + release: AdmissionCaseRelease + resolver: InMemoryContentResolver + + +def _merge_resolvers(*values: InMemoryContentResolver) -> InMemoryContentResolver: + artifacts = {} + for value in values: + artifacts.update(value.artifacts) + return InMemoryContentResolver(artifacts) + + +def _coverage() -> CoverageFunnel: + return CoverageFunnel( + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 0, + ( + RelationStatusCount(RelationStatus.COMPLETE, 1), + RelationStatusCount(RelationStatus.INCOMPLETE, 0), + RelationStatusCount(RelationStatus.UNRESOLVED, 0), + ), + (), + ) + + +def _instrument(name: str) -> tuple[InputManifest, tuple[InputReadReceipt, ...]]: + manifest = InputManifest((name,), derive_input_manifest_digest((name,))) + receipt = InputReadReceipt(name, f"/{name}.py", True, 10, content_digest(name)) + return manifest, (receipt,) + + +def _case_result( + value: validation.EvidenceValidationInput, + role: AdmissionCaseRole, + pair: str | None, + resolver: InMemoryContentResolver, +) -> tuple[AdmissionCaseResult, InMemoryContentResolver]: + request, combined = _request(value, resolver) + manifest, receipts = _instrument(role.value) + values: dict[str, object] = { + "case_id_digest": content_digest(f"release-{role.value}"), + "scope": ScopeKey("release-family", "python"), + "role": role, + "pair_id_digest": pair, + "evidence_inputs": (value,), + "defect_claims": (), + "satisfactions": (), + "silence": (), + "proof_attachments": (), + "coverage": _coverage(), + "input_manifest": manifest, + "input_receipts": receipts, + "complexity": request.complexity, + "budget": request.budget, + "total_reply_ms": 10, + } + draft = AdmissionCaseResult(**values, result_digest="0" * 64) # type: ignore[arg-type] + return replace(draft, result_digest=derive_admission_case_result_digest(draft)), combined + + +def _canonical_evidence_inputs( + admission: AdmissionCase, +) -> tuple[tuple[validation.EvidenceValidationInput, ...], InMemoryContentResolver]: + contract = admission.context.evidence.family_catalog.entries[0].contract + base = _with_fact_set(_model_record(contract), ()) + values: list[validation.EvidenceValidationInput] = [] + resolvers: list[InMemoryContentResolver] = [] + for suffix in ("vulnerable", "fixed", "ordinary"): + scope = replace(base.packet.scope, base_revision=f"base-{suffix}", head_revision=f"head-{suffix}") + packet = replace(base.packet, scope=scope) + decision = replace(base.decision, bounded_scope=scope, packet_digest=evidence.derive_packet_digest(packet)) + record = evidence.EvidenceRecord(packet, decision, "") + named_inputs, resolver = _named_inputs(record, contract, ()) + values.append(validation.make_evidence_validation_input(record=record, named_inputs=named_inputs)) + resolvers.append(resolver) + return tuple(values), _merge_resolvers(*resolvers) + + +def _binding(case: AdmissionCaseResult) -> ReleasedCaseBinding: + evidence = tuple(sorted(item.input_digest for item in case.evidence_inputs)) + changes = tuple(sorted({item.record.packet.provenance.change_input_digest for item in case.evidence_inputs})) + draft = ReleasedCaseBinding( + case.case_id_digest, + case.scope, + case.role, + case.pair_id_digest, + case.result_digest, + evidence, + changes, + case.input_manifest.manifest_digest, + "0" * 64, + ) + return replace(draft, binding_digest=derive_released_case_binding_digest(draft)) + + +def _resign_release( + release: AdmissionCaseRelease, + admission: AdmissionCase, + receipt: PopulationUseReceipt, +) -> AdmissionCaseRelease: + draft = replace(release, release_payload_digest="0" * 64, release_digest="0" * 64) + payload = derive_admission_case_release_payload_digest(draft) + review = _attestation( + payload, + receipt.custodian_identity, + admission.manifest.review.reviewer_identity, + admission.manifest.review.conflict_disclosure, + ) + reviewed = replace(draft, release_payload_digest=payload, review=review) + return replace(reviewed, release_digest=derive_admission_case_release_digest(reviewed)) + + +def _setup() -> _ReleaseSetup: + admission = _feasible_case() + preparation = _prepare(admission) + receipt = _spend(admission, preparation) + evidence_inputs, evidence_resolver = _canonical_evidence_inputs(admission) + pair = content_digest("release-pair") + vulnerable_input, fixed_input, ordinary_input = evidence_inputs + vulnerable, vulnerable_resolver = _case_result(vulnerable_input, AdmissionCaseRole.VULNERABLE, pair, evidence_resolver) + fixed, fixed_resolver = _case_result(fixed_input, AdmissionCaseRole.FIXED, pair, evidence_resolver) + ordinary, ordinary_resolver = _case_result(ordinary_input, AdmissionCaseRole.ORDINARY, None, evidence_resolver) + cases = (vulnerable, fixed, ordinary) + bindings = tuple(sorted((_binding(case) for case in cases), key=lambda item: item.case_id_digest)) + draft = AdmissionCaseRelease( + ADMISSION_CASE_RELEASE_SCHEMA, + admission.manifest.identity.population_digest, + admission.manifest.manifest_digest, + preparation.authorization.authorization_digest, + preparation.preparation_digest, + receipt.receipt_digest, + receipt.custodian_identity, + bindings, + "0" * 64, + admission.manifest.review, + "0" * 64, + ) + release = _resign_release(draft, admission, receipt) + resolver = _merge_resolvers(admission.resolver, vulnerable_resolver, fixed_resolver, ordinary_resolver) + return _ReleaseSetup(admission, preparation, receipt, cases, release, resolver) + + +def _validate( + setup: _ReleaseSetup, + *, + release: AdmissionCaseRelease | None = None, + cases: tuple[AdmissionCaseResult, ...] | None = None, +) -> None: + validate_admission_case_release( + setup.admission.manifest, + setup.admission.denominator, + setup.admission.readiness, + setup.admission.context, + setup.preparation, + setup.receipt, + release or setup.release, + cases or setup.cases, + setup.admission.port, + setup.resolver, + ) + + +def test_exact_post_spend_release_authorizes_case_associations() -> None: + setup = _setup() + + _validate(setup) + + +def test_release_is_refused_before_exact_spend_successor() -> None: + setup = _setup() + population = setup.admission.manifest.identity.population_digest + setup.admission.port.logs[population] = replace( + setup.admission.port.logs[population], + receipts=(), + receipt_count=0, + head_receipt_digest=None, + status=setup.receipt.prior_status, + log_digest=setup.preparation.authorization.custody_log_digest, + ) + + with pytest.raises((AdmissionCaseReleaseError, ValueError), match="successor|spent|custody"): + _validate(setup) + + +@pytest.mark.parametrize( + "field,value", + ( + ("population_digest", content_digest("other-population")), + ("manifest_digest", content_digest("other-manifest")), + ("authorization_digest", content_digest("other-authorization")), + ("preparation_digest", content_digest("other-preparation")), + ("label_release_receipt_digest", content_digest("other-receipt")), + ("custodian_identity", content_digest("other-custodian")), + ), +) +def test_self_consistent_release_cannot_change_spend_binding(field: str, value: str) -> None: + setup = _setup() + changed = _resign_release(replace(setup.release, **{field: value}), setup.admission, setup.receipt) + + with pytest.raises(AdmissionCaseReleaseError, match=field.replace("_", " ") + "|custodian"): + _validate(setup, release=changed) + + +@pytest.mark.parametrize("field", ("schema_version", "release_payload_digest", "release_digest")) +def test_release_schema_and_canonical_digests_are_rederived(field: str) -> None: + setup = _setup() + changed = replace(setup.release, **{field: content_digest(field)}) + + with pytest.raises(AdmissionCaseReleaseError, match="schema|payload digest|release digest"): + _validate(setup, release=changed) + + +@pytest.mark.parametrize("mutation", ("missing", "extra", "duplicate")) +def test_release_requires_exact_one_to_one_case_population(mutation: str) -> None: + setup = _setup() + bindings = setup.release.bindings + if mutation == "missing": + changed_bindings = bindings[:-1] + elif mutation == "extra": + extra = replace(bindings[-1], case_id_digest=content_digest("extra-case"), binding_digest="0" * 64) + extra = replace(extra, binding_digest=derive_released_case_binding_digest(extra)) + changed_bindings = tuple(sorted((*bindings, extra), key=lambda item: item.case_id_digest)) + else: + changed_bindings = (*bindings, bindings[-1]) + changed = _resign_release(replace(setup.release, bindings=changed_bindings), setup.admission, setup.receipt) + + with pytest.raises(AdmissionCaseReleaseError, match="binding|case population|canonical|pair"): + _validate(setup, release=changed) + + +@pytest.mark.parametrize( + "field", + ( + "scope", + "role", + "pair_id_digest", + "case_result_digest", + "evidence_input_digests", + "change_input_digests", + "input_manifest_digest", + ), +) +def test_release_binding_mutations_are_refused(field: str) -> None: + setup = _setup() + original = setup.release.bindings[0] + other_role = AdmissionCaseRole.FIXED if original.role is not AdmissionCaseRole.FIXED else AdmissionCaseRole.VULNERABLE + values: dict[str, object] = { + "scope": ScopeKey(original.scope.family_id, "other-ecosystem"), + "role": other_role, + "pair_id_digest": content_digest("other-pair"), + "case_result_digest": content_digest("other-result"), + "evidence_input_digests": (content_digest("other-evidence"),), + "change_input_digests": (content_digest("other-change"),), + "input_manifest_digest": content_digest("other-input-manifest"), + } + changed_binding = replace(original, **{field: values[field]}, binding_digest="0" * 64) + changed_binding = replace(changed_binding, binding_digest=derive_released_case_binding_digest(changed_binding)) + bindings = tuple(sorted((changed_binding, *setup.release.bindings[1:]), key=lambda item: item.case_id_digest)) + changed = _resign_release(replace(setup.release, bindings=bindings), setup.admission, setup.receipt) + + with pytest.raises(AdmissionCaseReleaseError, match="binding|pair|case"): + _validate(setup, release=changed) + + +def test_binding_digest_is_rederived() -> None: + setup = _setup() + changed_binding = replace(setup.release.bindings[0], binding_digest=content_digest("changed-binding")) + bindings = (changed_binding, *setup.release.bindings[1:]) + changed = replace(setup.release, bindings=bindings) + + with pytest.raises(AdmissionCaseReleaseError, match="binding digest"): + _validate(setup, release=changed) + + +def test_rehashed_vulnerable_fixed_operational_payload_swap_is_refused() -> None: + setup = _setup() + vulnerable, fixed, ordinary = setup.cases + operational_fields = tuple( + field.name + for field in fields(AdmissionCaseResult) + if field.name not in {"case_id_digest", "scope", "role", "pair_id_digest", "result_digest"} + ) + vulnerable_values = {name: getattr(fixed, name) for name in operational_fields} + fixed_values = {name: getattr(vulnerable, name) for name in operational_fields} + changed_vulnerable = replace(vulnerable, **vulnerable_values, result_digest="0" * 64) + changed_fixed = replace(fixed, **fixed_values, result_digest="0" * 64) + changed_vulnerable = replace(changed_vulnerable, result_digest=derive_admission_case_result_digest(changed_vulnerable)) + changed_fixed = replace(changed_fixed, result_digest=derive_admission_case_result_digest(changed_fixed)) + + with pytest.raises(AdmissionCaseReleaseError, match="binding"): + _validate(setup, cases=(changed_vulnerable, changed_fixed, ordinary)) + + +def test_release_refuses_tampered_named_input_payload_with_stale_input_digest() -> None: + setup = _setup() + vulnerable, fixed, ordinary = setup.cases + evidence_input = vulnerable.evidence_inputs[0] + first, *remaining = evidence_input.named_inputs.inputs + tampered_named = replace(first, identity=content_digest("tampered-named-input")) + stale_named_inputs = replace(evidence_input.named_inputs, inputs=(tampered_named, *remaining)) + stale_input = replace(evidence_input, named_inputs=stale_named_inputs) + changed = replace(vulnerable, evidence_inputs=(stale_input,)) + + with pytest.raises(AdmissionCaseReleaseError, match="named input|input digest|provenance"): + _validate(setup, cases=(changed, fixed, ordinary)) + + +def test_release_refuses_tampered_tracked_inputs_with_stale_manifest_digest() -> None: + setup = _setup() + vulnerable, fixed, ordinary = setup.cases + stale_manifest = replace(vulnerable.input_manifest, tracked_input_ids=("substituted-input",)) + changed = replace(vulnerable, input_manifest=stale_manifest) + + with pytest.raises(AdmissionCaseReleaseError, match="manifest digest|tracked input"): + _validate(setup, cases=(changed, fixed, ordinary)) + + +def test_release_review_must_be_independent_and_custodian_produced() -> None: + setup = _setup() + payload = setup.release.release_payload_digest + disclosure = setup.release.review.conflict_disclosure + reviewer = setup.release.review.reviewer_identity + wrong_producer = _attestation(payload, content_digest("wrong-producer"), reviewer, disclosure) + self_review = _attestation(payload, reviewer, reviewer, disclosure) + + for review in (wrong_producer, self_review): + changed = replace(setup.release, review=review, release_digest="0" * 64) + changed = replace(changed, release_digest=derive_admission_case_release_digest(changed)) + with pytest.raises(AdmissionCaseReleaseError, match="producer|independent"): + _validate(setup, release=changed) + + +def test_release_review_subject_reviewer_digest_and_disclosure_are_revalidated() -> None: + setup = _setup() + review = setup.release.review + mutations = ( + replace(review, subject_digest=content_digest("wrong-subject")), + _attestation( + review.subject_digest, + review.producer_identity, + content_digest("unknown-reviewer"), + review.conflict_disclosure, + ), + replace(review, attestation_digest=content_digest("wrong-attestation")), + _attestation( + review.subject_digest, + review.producer_identity, + review.reviewer_identity, + replace(review.conflict_disclosure, artifact_id="missing-release-disclosure"), + ), + ) + + for changed_review in mutations: + changed = replace(setup.release, review=changed_review, release_digest="0" * 64) + changed = replace(changed, release_digest=derive_admission_case_release_digest(changed)) + with pytest.raises(AdmissionCaseReleaseError, match="subject|reviewer|attestation|artifact"): + _validate(setup, release=changed)