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