From 361c216402fb3420172758c248c123bd516a2045 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 31 Oct 2026 19:08:14 +0100 Subject: [PATCH] feat(detection-core): validate reviewed admission payloads --- .kiro/specs/detection-core/tasks.md | 2 +- .../core/_admission_review_shape.py | 110 ++++ src/openultrasast/core/admission_review.py | 432 ++++++++++++++++ tests/test_core_admission_review.py | 485 ++++++++++++++++++ 4 files changed, 1028 insertions(+), 1 deletion(-) create mode 100644 src/openultrasast/core/_admission_review_shape.py create mode 100644 src/openultrasast/core/admission_review.py create mode 100644 tests/test_core_admission_review.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 520b772..cd88b44 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -235,7 +235,7 @@ - Refuse zero or clean-looking results unless tracked input reads and prerequisite semantic stages passed. - Done when missing metrics, counts, failure fields, readiness ancestry, or holdout-report identity fields invalidate the corresponding record. - _Requirements: 13.2, 15.1, 15.2, 15.4, 15.5_ -- [ ] 8.3 Validate independently reviewed census and manifest payloads +- [x] 8.3 Validate independently reviewed census and manifest payloads - Canonically bind every substantive census and manifest field into separate review-subject digests. - Require producer/reviewer separation, admitted reviewer identity, allowed identity pairing, and resolver-validated conflict disclosure. - Reject self-review, copied attestations, unknown reviewers, malformed review digests, and unreadable disclosures. diff --git a/src/openultrasast/core/_admission_review_shape.py b/src/openultrasast/core/_admission_review_shape.py new file mode 100644 index 0000000..1842c5e --- /dev/null +++ b/src/openultrasast/core/_admission_review_shape.py @@ -0,0 +1,110 @@ +"""Recursive closed-graph checks for reviewed admission payloads.""" + +from __future__ import annotations + +from dataclasses import fields +from typing import Any, cast + +_SCALAR_TYPES: tuple[type[object], ...] = (str, int, float, bool, type(None)) +_ENUM_STATE_FIELDS = frozenset(("_value_", "_name_", "__objclass__", "_sort_order_")) + + +def closed_graph_problem( + value: object, + *, + dataclass_types: tuple[type[object], ...], + enum_types: tuple[type[object], ...], + path: str = "payload", +) -> str | None: + """Return the first non-closed node in a whitelisted payload graph.""" + value_type = type(value) + if value_type in dataclass_types: + return _dataclass_problem(value, value_type, dataclass_types, enum_types, path) + if value_type is tuple: + for index, item in enumerate(cast(tuple[object, ...], value)): + problem = closed_graph_problem( + item, + dataclass_types=dataclass_types, + enum_types=enum_types, + path=f"{path}[{index}]", + ) + if problem is not None: + return problem + return None + if value_type in enum_types: + return _enum_problem(value, value_type, path) + if value_type in _SCALAR_TYPES: + return None + return f"{path} uses non-closed type {value_type.__name__}" + + +def _dataclass_problem( + value: object, + value_type: type[object], + dataclass_types: tuple[type[object], ...], + enum_types: tuple[type[object], ...], + path: str, +) -> str | None: + declared_fields = fields(cast(Any, value_type)) + declared_names = {field.name for field in declared_fields} + try: + actual_names = set(vars(value)) + except TypeError: + actual_names = declared_names + if actual_names != declared_names: + return f"{path} must have exactly its declared fields" + for field in declared_fields: + try: + field_value = getattr(value, field.name) + except AttributeError: + return f"{path}.{field.name} is missing" + problem = closed_graph_problem( + field_value, + dataclass_types=dataclass_types, + enum_types=enum_types, + path=f"{path}.{field.name}", + ) + if problem is not None: + return problem + return None + + +def _enum_problem(value: object, enum_type: type[object], path: str) -> str | None: + state = vars(value) + if set(state) != _ENUM_STATE_FIELDS: + return f"{path} enum member must have exactly its canonical state fields" + expected = _canonical_enum_state(value, enum_type) + if expected is None: + return f"{path} enum type has aliases or nonunique member mappings" + expected_name, expected_value, expected_order = expected + if ( + state["_name_"] is not expected_name + or state["_value_"] is not expected_value + or state["__objclass__"] is not enum_type + or type(state["_sort_order_"]) is not int + or state["_sort_order_"] != expected_order + ): + return f"{path} enum member state is not canonical" + return None + + +def _canonical_enum_state(value: object, enum_type: type[object]) -> tuple[object, object, int] | None: + enum_class = cast(Any, enum_type) + canonical_members = tuple(enum_class) + name_entries = tuple(enum_class.__members__.items()) + value_entries = tuple(enum_class._value2member_map_.items()) + if len(name_entries) != len(canonical_members) or len(value_entries) != len(canonical_members): + return None + if any(sum(candidate is member for _, candidate in name_entries) != 1 for member in canonical_members): + return None + if any(sum(candidate is member for _, candidate in value_entries) != 1 for member in canonical_members): + return None + names = tuple(name for name, member in name_entries if member is value) + values = tuple(item for item, member in value_entries if member is value) + orders = tuple(index for index, member in enumerate(canonical_members) if member is value) + if len(names) != 1 or len(values) != 1 or len(orders) != 1: + return None + return names[0], values[0], orders[0] + + +__all__ = ["closed_graph_problem"] diff --git a/src/openultrasast/core/admission_review.py b/src/openultrasast/core/admission_review.py new file mode 100644 index 0000000..41fc4cf --- /dev/null +++ b/src/openultrasast/core/admission_review.py @@ -0,0 +1,432 @@ +"""Reviewed label-blind census and admission-manifest contracts.""" + +from __future__ import annotations + +import math +import re +from dataclasses import dataclass + +from openultrasast.core import canonical +from openultrasast.core._admission_review_shape import closed_graph_problem +from openultrasast.core.factpack import ( + FactPackValidationError, + IndependenceAttestation, + ReviewIndependencePolicy, + derive_independence_attestation_digest, + derive_review_independence_policy_digest, +) +from openultrasast.core.measurement import ( + BaselineIdentity, + Comparator, + MetricId, + ScopeKey, + Statistic, + derive_baseline_identity_digest, +) + +CENSUS_PAYLOAD_SCHEMA = "openultrasast.detection-core.population-census-payload.v2" +CENSUS_SCHEMA = "openultrasast.detection-core.population-census.v2" +MANIFEST_PAYLOAD_SCHEMA = "openultrasast.detection-core.admission-manifest-payload.v2" +MANIFEST_SCHEMA = "openultrasast.detection-core.admission-manifest.v2" +THRESHOLD_SET_SCHEMA = "openultrasast.detection-core.threshold-set.v2" +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +class AdmissionReviewError(ValueError): + """A census or manifest review contract is invalid.""" + + +@dataclass(frozen=True) +class ScopeCensus: + scope: ScopeKey + total_cases: int + pair_count: int + vulnerable_side_count: int + fixed_side_count: int + ordinary_case_count: int + semantic_site_capacity: int + model_call_capacity: int + + +@dataclass(frozen=True) +class PopulationCensus: + population_digest: str + rows: tuple[ScopeCensus, ...] + case_labels_exposed: bool + census_payload_digest: str + review: IndependenceAttestation + census_digest: str + + +@dataclass(frozen=True) +class AdmissionThreshold: + metric_id: MetricId + statistic: Statistic + comparator: Comparator + target: float + + +@dataclass(frozen=True) +class AdmissionManifestV2: + manifest_payload_digest: str + review: IndependenceAttestation + label_isolated: bool + labels_opened: bool + identity: BaselineIdentity + census: PopulationCensus + readiness_digests: tuple[str, ...] + custody_log_digest: str + denominator_policy_digest: str + thresholds: tuple[AdmissionThreshold, ...] + threshold_set_digest: str + threshold_lineage_manifest_digest: str + validation_context_digest: str + manifest_digest: str + + +_GRAPH_DATACLASS_TYPES: tuple[type[object], ...] = ( + ScopeCensus, + PopulationCensus, + AdmissionThreshold, + AdmissionManifestV2, + ScopeKey, + BaselineIdentity, + IndependenceAttestation, + ReviewIndependencePolicy, + canonical.ContentRef, +) +_GRAPH_ENUM_TYPES: tuple[type[object], ...] = (MetricId, Statistic, Comparator) + + +def _closed_graph(value: object, field: str) -> None: + problem = closed_graph_problem( + value, + dataclass_types=_GRAPH_DATACLASS_TYPES, + enum_types=_GRAPH_ENUM_TYPES, + path=field, + ) + if problem is not None: + raise AdmissionReviewError(f"closed payload graph violation: {problem}") + + +def derive_census_payload_digest(census: PopulationCensus) -> str: + _closed_graph(census, "census") + _validate_census_shape(census) + return canonical.content_digest(_census_payload_value(census)) + + +def derive_census_digest(census: PopulationCensus) -> str: + _closed_graph(census, "census") + _validate_census_shape(census) + return canonical.content_digest( + { + "schema_version": CENSUS_SCHEMA, + "census_payload_digest": census.census_payload_digest, + "review": _attestation_value(census.review), + } + ) + + +def derive_manifest_payload_digest(manifest: AdmissionManifestV2) -> str: + _closed_graph(manifest, "manifest") + _validate_manifest_shape(manifest) + return canonical.content_digest(_manifest_payload_value(manifest)) + + +def derive_manifest_digest(manifest: AdmissionManifestV2) -> str: + _closed_graph(manifest, "manifest") + _validate_manifest_shape(manifest) + return canonical.content_digest( + { + "schema_version": MANIFEST_SCHEMA, + "manifest_payload_digest": manifest.manifest_payload_digest, + "review": _attestation_value(manifest.review), + } + ) + + +def derive_threshold_set_digest(thresholds: tuple[AdmissionThreshold, ...]) -> str: + _closed_graph(thresholds, "thresholds") + _validate_thresholds(thresholds) + return canonical.content_digest( + {"schema_version": THRESHOLD_SET_SCHEMA, "thresholds": tuple(_threshold_value(item) for item in thresholds)} + ) + + +def _digest(value: object, field: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise AdmissionReviewError(f"{field} must be a full lowercase SHA-256 digest") + + +def _count(value: object, field: str) -> None: + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise AdmissionReviewError(f"{field} must be a non-negative integer") + + +def _content_ref_value(reference: canonical.ContentRef) -> canonical.JsonValue: + return { + "artifact_id": reference.artifact_id, + "content_digest": reference.content_digest, + "media_type": reference.media_type, + "byte_length": reference.byte_length, + "retention_policy_digest": reference.retention_policy_digest, + "storage_receipt_digest": reference.storage_receipt_digest, + } + + +def _attestation_value(attestation: IndependenceAttestation) -> canonical.JsonValue: + return { + "subject_digest": attestation.subject_digest, + "producer_identity": attestation.producer_identity, + "reviewer_identity": attestation.reviewer_identity, + "conflict_disclosure": _content_ref_value(attestation.conflict_disclosure), + "attestation_digest": attestation.attestation_digest, + } + + +def _scope_value(scope: ScopeKey) -> canonical.JsonValue: + return {"family_id": scope.family_id, "ecosystem": scope.ecosystem} + + +def _row_value(row: ScopeCensus) -> canonical.JsonValue: + return { + "scope": _scope_value(row.scope), + "total_cases": row.total_cases, + "pair_count": row.pair_count, + "vulnerable_side_count": row.vulnerable_side_count, + "fixed_side_count": row.fixed_side_count, + "ordinary_case_count": row.ordinary_case_count, + "semantic_site_capacity": row.semantic_site_capacity, + "model_call_capacity": row.model_call_capacity, + } + + +def _census_payload_value(census: PopulationCensus) -> canonical.JsonValue: + return { + "schema_version": CENSUS_PAYLOAD_SCHEMA, + "population_digest": census.population_digest, + "rows": tuple(_row_value(row) for row in census.rows), + "case_labels_exposed": census.case_labels_exposed, + } + + +def _threshold_value(threshold: AdmissionThreshold) -> canonical.JsonValue: + return { + "metric_id": threshold.metric_id.value, + "statistic": threshold.statistic.value, + "comparator": threshold.comparator.value, + "target": threshold.target, + } + + +def _manifest_payload_value(manifest: AdmissionManifestV2) -> canonical.JsonValue: + return { + "schema_version": MANIFEST_PAYLOAD_SCHEMA, + "label_isolated": manifest.label_isolated, + "labels_opened": manifest.labels_opened, + "baseline_identity_digest": derive_baseline_identity_digest(manifest.identity), + "census_digest": manifest.census.census_digest, + "readiness_digests": manifest.readiness_digests, + "custody_log_digest": manifest.custody_log_digest, + "denominator_policy_digest": manifest.denominator_policy_digest, + "thresholds": tuple(_threshold_value(item) for item in manifest.thresholds), + "threshold_set_digest": manifest.threshold_set_digest, + "threshold_lineage_manifest_digest": manifest.threshold_lineage_manifest_digest, + "validation_context_digest": manifest.validation_context_digest, + } + + +def _validate_row(row: ScopeCensus) -> None: + row.scope.__post_init__() + for field in ( + "total_cases", + "pair_count", + "vulnerable_side_count", + "fixed_side_count", + "ordinary_case_count", + "semantic_site_capacity", + "model_call_capacity", + ): + _count(getattr(row, field), field) + if row.total_cases != row.vulnerable_side_count + row.fixed_side_count + row.ordinary_case_count: + raise AdmissionReviewError("census total does not reconcile with aggregate roles") + if row.pair_count > min(row.vulnerable_side_count, row.fixed_side_count): + raise AdmissionReviewError("census pair count exceeds side capacity") + + +def _validate_census_shape(census: PopulationCensus) -> None: + _digest(census.population_digest, "census population digest") + if not census.rows: + raise AdmissionReviewError("census rows must be a nonempty tuple") + for row in census.rows: + _validate_row(row) + scopes = tuple(row.scope for row in census.rows) + if scopes != tuple(sorted(scopes)) or len(set(scopes)) != len(scopes): + raise AdmissionReviewError("census scopes must be unique and canonically ordered") + if not isinstance(census.case_labels_exposed, bool): + raise AdmissionReviewError("case_labels_exposed must be boolean") + for value, field in ( + (census.census_payload_digest, "census payload digest"), + (census.census_digest, "census digest"), + ): + _digest(value, field) + + +def _validate_thresholds(thresholds: tuple[AdmissionThreshold, ...]) -> None: + if not thresholds: + raise AdmissionReviewError("thresholds must be a nonempty tuple") + for threshold in thresholds: + if not isinstance(threshold.metric_id, MetricId) or not isinstance(threshold.statistic, Statistic): + raise AdmissionReviewError("threshold metric and statistic must use closed enums") + if not isinstance(threshold.comparator, Comparator): + raise AdmissionReviewError("threshold comparator must use the closed enum") + if isinstance(threshold.target, bool) or not isinstance(threshold.target, float) or not math.isfinite(threshold.target): + raise AdmissionReviewError("threshold target must be a finite float") + values = tuple((item.metric_id.value, item.statistic.value) for item in thresholds) + if values != tuple(sorted(values)) or len(set(values)) != len(values): + raise AdmissionReviewError("thresholds must be unique and canonically ordered") + + +def _validate_manifest_shape(manifest: AdmissionManifestV2) -> None: + _validate_census_shape(manifest.census) + derive_baseline_identity_digest(manifest.identity) + if not isinstance(manifest.label_isolated, bool) or not isinstance(manifest.labels_opened, bool): + raise AdmissionReviewError("manifest label flags must be boolean") + if not manifest.readiness_digests: + raise AdmissionReviewError("readiness digests must be a nonempty tuple") + for item in manifest.readiness_digests: + _digest(item, "readiness digest") + if manifest.readiness_digests != tuple(sorted(manifest.readiness_digests)): + raise AdmissionReviewError("readiness digests must use canonical order") + if len(set(manifest.readiness_digests)) != len(manifest.readiness_digests): + raise AdmissionReviewError("readiness digests contain duplicates") + for value, field in ( + (manifest.manifest_payload_digest, "manifest payload digest"), + (manifest.custody_log_digest, "custody log digest"), + (manifest.denominator_policy_digest, "denominator policy digest"), + (manifest.threshold_set_digest, "threshold set digest"), + (manifest.threshold_lineage_manifest_digest, "threshold lineage manifest digest"), + (manifest.validation_context_digest, "validation context digest"), + (manifest.manifest_digest, "manifest digest"), + ): + _digest(value, field) + _validate_thresholds(manifest.thresholds) + + +def _validate_policy(policy: ReviewIndependencePolicy) -> None: + try: + policy.__post_init__() + except FactPackValidationError as exc: + raise AdmissionReviewError(f"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 AdmissionReviewError("review policy digest cannot be re-derived") + + +def _validate_review( + review: IndependenceAttestation, + subject: str, + policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, +) -> None: + try: + review.__post_init__() + review.conflict_disclosure.__post_init__() + except (FactPackValidationError, ValueError) as exc: + raise AdmissionReviewError(f"review attestation is invalid: {exc}") from exc + if review.subject_digest != subject: + raise AdmissionReviewError("review subject does not match payload digest") + if review.producer_identity == review.reviewer_identity: + raise AdmissionReviewError("review producer and reviewer must be independent") + if review.reviewer_identity not in policy.reviewed_reviewer_digests: + raise AdmissionReviewError("reviewer is not admitted by policy") + if (review.producer_identity, review.reviewer_identity) in policy.forbidden_identity_pairs: + raise AdmissionReviewError("review producer/reviewer pair is forbidden") + if review.attestation_digest != derive_independence_attestation_digest(review): + raise AdmissionReviewError("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 AdmissionReviewError("review conflict disclosure must be nonempty") + + +def _validate_census( + census: PopulationCensus, + policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, +) -> None: + _validate_census_shape(census) + if census.case_labels_exposed: + raise AdmissionReviewError("census must not expose case labels") + expected_census_payload = derive_census_payload_digest(census) + if census.census_payload_digest != expected_census_payload: + raise AdmissionReviewError("census payload digest cannot be re-derived") + _validate_review(census.review, expected_census_payload, policy, resolver) + if census.census_digest != derive_census_digest(census): + raise AdmissionReviewError("census digest cannot be re-derived") + + +def _validate_manifest( + manifest: AdmissionManifestV2, + census: PopulationCensus, + policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, +) -> None: + _validate_manifest_shape(manifest) + if not manifest.label_isolated or manifest.labels_opened: + raise AdmissionReviewError("manifest must remain label isolated and unopened") + if manifest.census != census: + raise AdmissionReviewError("manifest census does not match reviewed census") + if manifest.identity.population_digest != census.population_digest: + raise AdmissionReviewError("manifest population does not match reviewed census") + if manifest.denominator_policy_digest != manifest.identity.denominator_policy_digest: + raise AdmissionReviewError("manifest denominator does not match baseline identity") + expected_thresholds = derive_threshold_set_digest(manifest.thresholds) + if manifest.threshold_set_digest != expected_thresholds: + raise AdmissionReviewError("threshold set digest cannot be re-derived") + if manifest.identity.threshold_set_digest != expected_thresholds: + raise AdmissionReviewError("manifest thresholds do not match baseline identity") + expected_manifest_payload = derive_manifest_payload_digest(manifest) + if manifest.manifest_payload_digest != expected_manifest_payload: + raise AdmissionReviewError("manifest payload digest cannot be re-derived") + _validate_review(manifest.review, expected_manifest_payload, policy, resolver) + if manifest.manifest_digest != derive_manifest_digest(manifest): + raise AdmissionReviewError("manifest digest cannot be re-derived") + + +def validate_reviewed_admission_payloads( + census: PopulationCensus, + manifest: AdmissionManifestV2, + policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, +) -> None: + _closed_graph(census, "census") + _closed_graph(manifest, "manifest") + _closed_graph(policy, "review policy") + _validate_policy(policy) + _validate_census(census, policy, resolver) + _validate_manifest(manifest, census, policy, resolver) + + +__all__ = [ + "AdmissionManifestV2", + "AdmissionReviewError", + "AdmissionThreshold", + "PopulationCensus", + "ScopeCensus", + "derive_census_digest", + "derive_census_payload_digest", + "derive_manifest_digest", + "derive_manifest_payload_digest", + "derive_threshold_set_digest", + "validate_reviewed_admission_payloads", +] diff --git a/tests/test_core_admission_review.py b/tests/test_core_admission_review.py new file mode 100644 index 0000000..864cbc1 --- /dev/null +++ b/tests/test_core_admission_review.py @@ -0,0 +1,485 @@ +"""Independently reviewed label-blind admission payload contracts.""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +from typing import cast + +import pytest +from detection_core_fixtures import InMemoryContentResolver, stored_artifact + +from openultrasast.core import canonical +from openultrasast.core.admission_review import ( + AdmissionManifestV2, + AdmissionReviewError, + AdmissionThreshold, + PopulationCensus, + ScopeCensus, + derive_census_digest, + derive_census_payload_digest, + derive_manifest_digest, + derive_manifest_payload_digest, + derive_threshold_set_digest, + validate_reviewed_admission_payloads, +) +from openultrasast.core.factpack import ( + IndependenceAttestation, + ReviewIndependencePolicy, + derive_independence_attestation_digest, + derive_review_independence_policy_digest, +) +from openultrasast.core.measurement import ( + BaselineIdentity, + Comparator, + MetricId, + ScopeKey, + Statistic, + derive_baseline_identity_digest, +) + + +def _policy(reviewer: str, forbidden: tuple[tuple[str, str], ...] = ()) -> ReviewIndependencePolicy: + stores = ("trusted-store",) + return ReviewIndependencePolicy( + (reviewer,), + forbidden, + stores, + 4096, + derive_review_independence_policy_digest((reviewer,), forbidden, stores, 4096), + ) + + +def _attestation(subject: str, producer: str, reviewer: str, disclosure: canonical.ContentRef) -> IndependenceAttestation: + provisional = IndependenceAttestation(subject, producer, reviewer, disclosure, "0" * 64) + return replace(provisional, attestation_digest=derive_independence_attestation_digest(provisional)) + + +def _baseline(population: str, scope: ScopeKey) -> BaselineIdentity: + digest = canonical.content_digest + return BaselineIdentity( + "arm-a", + population, + digest("freeze"), + (scope,), + (digest("contract"),), + digest("catalog"), + digest("fact-schema"), + digest("fact-set"), + digest("acquisition"), + "analyzer-v2", + digest("configuration"), + (), + None, + digest("denominator"), + digest("threshold-set"), + digest("outcomes"), + ) + + +def _case() -> tuple[PopulationCensus, AdmissionManifestV2, ReviewIndependencePolicy, InMemoryContentResolver]: + producer = canonical.content_digest("benchmark-producer") + reviewer = canonical.content_digest("independent-reviewer") + disclosure, disclosure_resolver = stored_artifact(b'{"conflicts":[]}', artifact_id="admission-conflicts") + policy = _policy(reviewer) + population = canonical.content_digest("untouched-population") + scope = ScopeKey("reference-family", "python") + row = ScopeCensus(scope, 8, 2, 2, 2, 4, 12, 3) + provisional_review = _attestation("0" * 64, producer, reviewer, disclosure) + provisional_census = PopulationCensus(population, (row,), False, "0" * 64, provisional_review, "0" * 64) + census_payload = derive_census_payload_digest(provisional_census) + census_review = _attestation(census_payload, producer, reviewer, disclosure) + census = replace(provisional_census, census_payload_digest=census_payload, review=census_review) + census = replace(census, census_digest=derive_census_digest(census)) + + threshold = AdmissionThreshold(MetricId.VULNERABLE_RECALL, Statistic.WILSON_LOWER, Comparator.AT_LEAST, 0.5) + thresholds = (threshold,) + threshold_set = derive_threshold_set_digest(thresholds) + identity = replace(_baseline(population, scope), threshold_set_digest=threshold_set) + provisional_manifest = AdmissionManifestV2( + "0" * 64, + provisional_review, + True, + False, + identity, + census, + (canonical.content_digest("readiness-development"),), + canonical.content_digest("custody-head"), + identity.denominator_policy_digest, + thresholds, + threshold_set, + canonical.content_digest("threshold-lineage"), + canonical.content_digest("validation-context"), + "0" * 64, + ) + payload = derive_manifest_payload_digest(provisional_manifest) + review = _attestation(payload, producer, reviewer, disclosure) + manifest = replace(provisional_manifest, manifest_payload_digest=payload, review=review) + manifest = replace(manifest, manifest_digest=derive_manifest_digest(manifest)) + return census, manifest, policy, disclosure_resolver + + +def _validate(case: tuple[PopulationCensus, AdmissionManifestV2, ReviewIndependencePolicy, InMemoryContentResolver]) -> None: + validate_reviewed_admission_payloads(*case) + + +def test_label_blind_census_and_manifest_validate_and_preserve_only_aggregates() -> None: + case = _case() + _validate(case) + census, manifest, _, _ = case + assert census.rows[0].total_cases == 8 + assert not hasattr(census, "case_ids") + assert not hasattr(census, "labels") + assert census.review.subject_digest == census.census_payload_digest + assert manifest.review.subject_digest == manifest.manifest_payload_digest + assert derive_baseline_identity_digest(manifest.identity) + + +def test_every_census_and_manifest_field_is_bound_to_its_own_review_subject() -> None: + census, manifest, policy, resolver = _case() + changed_row = replace(census.rows[0], semantic_site_capacity=13) + changed_census = replace(census, rows=(changed_row,)) + with pytest.raises(AdmissionReviewError, match="census payload"): + _validate((changed_census, manifest, policy, resolver)) + + changed_manifest = replace(manifest, readiness_digests=(canonical.content_digest("other-readiness"),)) + with pytest.raises(AdmissionReviewError, match="manifest payload"): + _validate((census, changed_manifest, policy, resolver)) + + +def test_copied_attestation_cannot_review_another_payload() -> None: + census, manifest, policy, resolver = _case() + copied = replace(manifest, review=census.review) + with pytest.raises(AdmissionReviewError, match="subject"): + _validate((census, copied, policy, resolver)) + + +@pytest.mark.parametrize("failure", ("self", "unknown", "forbidden", "digest")) +def test_independence_policy_rejects_invalid_reviews(failure: str) -> None: + census, manifest, policy, resolver = _case() + review = manifest.review + if failure == "self": + review = replace(review, producer_identity=review.reviewer_identity) + elif failure == "unknown": + review = replace(review, reviewer_identity=canonical.content_digest("unknown-reviewer")) + elif failure == "forbidden": + policy = _policy(review.reviewer_identity, ((review.producer_identity, review.reviewer_identity),)) + else: + review = replace(review, attestation_digest=canonical.content_digest("forged-attestation")) + changed = replace(manifest, review=review) + with pytest.raises(AdmissionReviewError, match="review|attestation|independent|forbidden"): + _validate((census, changed, policy, resolver)) + + +def test_unreadable_or_empty_conflict_disclosure_is_refused() -> None: + census, manifest, policy, resolver = _case() + unreadable = InMemoryContentResolver({}) + with pytest.raises(canonical.ArtifactResolutionError): + _validate((census, manifest, policy, unreadable)) + + empty_ref, empty_resolver = stored_artifact(b"", artifact_id="empty-conflicts") + review = _attestation(manifest.manifest_payload_digest, manifest.review.producer_identity, manifest.review.reviewer_identity, empty_ref) + changed = replace(manifest, review=review) + combined = InMemoryContentResolver({**resolver.artifacts, **empty_resolver.artifacts}) + with pytest.raises(AdmissionReviewError, match="nonempty"): + _validate((census, changed, policy, combined)) + + +@pytest.mark.parametrize("target", ("census", "manifest")) +def test_label_exposure_is_refused_before_manifest_validation(target: str) -> None: + census, manifest, policy, resolver = _case() + if target == "census": + census = replace(census, case_labels_exposed=True) + else: + manifest = replace(manifest, labels_opened=True) + with pytest.raises(AdmissionReviewError, match="label"): + _validate((census, manifest, policy, resolver)) + + +@dataclass(frozen=True) +class _CensusWithCases(PopulationCensus): + case_ids: tuple[str, ...] + + +@dataclass(frozen=True) +class _ManifestWithLabels(AdmissionManifestV2): + per_case_labels: tuple[str, ...] + + +class _HiddenTuple(tuple[object, ...]): + def __new__(cls, values: tuple[object, ...], field: str, hidden: object) -> _HiddenTuple: + instance = super().__new__(cls, values) + setattr(instance, field, hidden) + return instance + + +class _HiddenStr(str): + hidden: str + + def __new__(cls, value: str) -> _HiddenStr: + instance = super().__new__(cls, value) + instance.hidden = "case-data" + return instance + + +class _HiddenInt(int): + hidden: str + + def __new__(cls, value: int) -> _HiddenInt: + instance = super().__new__(cls, value) + instance.hidden = "case-data" + return instance + + +class _HiddenFloat(float): + hidden: str + + def __new__(cls, value: float) -> _HiddenFloat: + instance = super().__new__(cls, value) + instance.hidden = "case-data" + return instance + + +@pytest.mark.parametrize("target", ("census_subclass", "manifest_subclass", "census_injected", "manifest_injected")) +def test_payload_graph_refuses_subtypes_and_injected_fields(target: str) -> None: + census, manifest, policy, resolver = _case() + if target == "census_subclass": + census = _CensusWithCases( + census.population_digest, + census.rows, + census.case_labels_exposed, + census.census_payload_digest, + census.review, + census.census_digest, + (canonical.content_digest("case-identity"),), + ) + manifest = replace(manifest, census=census) + elif target == "manifest_subclass": + manifest = _ManifestWithLabels( + manifest.manifest_payload_digest, + manifest.review, + manifest.label_isolated, + manifest.labels_opened, + manifest.identity, + manifest.census, + manifest.readiness_digests, + manifest.custody_log_digest, + manifest.denominator_policy_digest, + manifest.thresholds, + manifest.threshold_set_digest, + manifest.threshold_lineage_manifest_digest, + manifest.validation_context_digest, + manifest.manifest_digest, + ("vulnerable",), + ) + elif target == "census_injected": + object.__setattr__(census, "case_ids", (canonical.content_digest("case-identity"),)) + else: + object.__setattr__(manifest, "per_case_labels", ("vulnerable",)) + + with pytest.raises(AdmissionReviewError, match="closed|exact|field|type"): + _validate((census, manifest, policy, resolver)) + + +@pytest.mark.parametrize("target", ("row", "scope", "threshold", "baseline", "review", "disclosure", "policy")) +def test_nested_payload_graph_refuses_undeclared_fields(target: str) -> None: + census, manifest, policy, resolver = _case() + values = { + "row": census.rows[0], + "scope": census.rows[0].scope, + "threshold": manifest.thresholds[0], + "baseline": manifest.identity, + "review": manifest.review, + "disclosure": manifest.review.conflict_disclosure, + "policy": policy, + } + object.__setattr__(values[target], "undeclared", "not-canonical") + + with pytest.raises(AdmissionReviewError, match="closed|exact|field|type"): + _validate((census, manifest, policy, resolver)) + + +@pytest.mark.parametrize("target", ("rows", "readiness")) +def test_payload_graph_refuses_tuple_subclasses_with_hidden_case_data(target: str) -> None: + census, manifest, policy, resolver = _case() + if target == "rows": + rows = _HiddenTuple(census.rows, "case_ids", (canonical.content_digest("hidden-case"),)) + census = replace(census, rows=cast(tuple[ScopeCensus, ...], rows)) + manifest = replace(manifest, census=census) + else: + readiness = _HiddenTuple(manifest.readiness_digests, "per_case_labels", ("vulnerable",)) + manifest = replace(manifest, readiness_digests=cast(tuple[str, ...], readiness)) + + with pytest.raises(AdmissionReviewError, match="tuple|closed|exact"): + _validate((census, manifest, policy, resolver)) + + +@pytest.mark.parametrize( + "target", + ("thresholds", "scopes", "contracts", "models", "reviewers", "pairs", "pair", "stores"), +) +def test_all_nested_tuple_nodes_require_plain_tuples(target: str) -> None: + census, manifest, policy, resolver = _case() + + def hidden(values: tuple[object, ...]) -> _HiddenTuple: + return _HiddenTuple(values, "hidden", "case-data") + + if target == "thresholds": + changed = cast(tuple[AdmissionThreshold, ...], hidden(manifest.thresholds)) + manifest = replace(manifest, thresholds=changed) + elif target == "scopes": + changed_scopes = cast(tuple[ScopeKey, ...], hidden(manifest.identity.claimed_scopes)) + identity = replace(manifest.identity, claimed_scopes=changed_scopes) + manifest = replace(manifest, identity=identity) + elif target == "contracts": + changed_contracts = cast(tuple[str, ...], hidden(manifest.identity.family_contract_digests)) + identity = replace(manifest.identity, family_contract_digests=changed_contracts) + manifest = replace(manifest, identity=identity) + elif target == "models": + changed_models = cast(tuple[str, ...], hidden(manifest.identity.model_program_digests)) + identity = replace(manifest.identity, model_program_digests=changed_models) + manifest = replace(manifest, identity=identity) + elif target == "reviewers": + changed_reviewers = cast(tuple[str, ...], hidden(policy.reviewed_reviewer_digests)) + policy = replace(policy, reviewed_reviewer_digests=changed_reviewers) + elif target == "pairs": + changed_pairs = cast(tuple[tuple[str, str], ...], hidden(policy.forbidden_identity_pairs)) + policy = replace(policy, forbidden_identity_pairs=changed_pairs) + elif target == "pair": + other = _HiddenTuple( + (canonical.content_digest("other-producer"), canonical.content_digest("other-reviewer")), + "hidden", + "case-data", + ) + policy = _policy(manifest.review.reviewer_identity, (cast(tuple[str, str], other),)) + else: + changed_stores = cast(tuple[str, ...], hidden(policy.trusted_store_identities)) + policy = replace(policy, trusted_store_identities=changed_stores) + + with pytest.raises(AdmissionReviewError, match="tuple|closed|exact"): + _validate((census, manifest, policy, resolver)) + + +def _payload_with_hidden_scalar( + target: str, + census: PopulationCensus, + manifest: AdmissionManifestV2, +) -> tuple[PopulationCensus, AdmissionManifestV2]: + if target == "census_digest": + census = replace(census, population_digest=_HiddenStr(census.population_digest)) + identity = replace(manifest.identity, population_digest=census.population_digest) + manifest = replace(manifest, census=census, identity=identity) + elif target == "manifest_digest": + manifest = replace(manifest, custody_log_digest=_HiddenStr(manifest.custody_log_digest)) + elif target == "scope_text": + scope = replace(census.rows[0].scope, family_id=_HiddenStr(census.rows[0].scope.family_id)) + census = replace(census, rows=(replace(census.rows[0], scope=scope),)) + manifest = replace(manifest, census=census, identity=replace(manifest.identity, claimed_scopes=(scope,))) + elif target == "baseline_text": + manifest = replace(manifest, identity=replace(manifest.identity, arm_id=_HiddenStr(manifest.identity.arm_id))) + elif target == "baseline_digest": + identity = replace(manifest.identity, freeze_digest=_HiddenStr(manifest.identity.freeze_digest)) + manifest = replace(manifest, identity=identity) + elif target == "count": + row = replace(census.rows[0], total_cases=_HiddenInt(census.rows[0].total_cases)) + census = replace(census, rows=(row,)) + manifest = replace(manifest, census=census) + elif target == "threshold_target": + threshold = replace(manifest.thresholds[0], target=_HiddenFloat(manifest.thresholds[0].target)) + manifest = replace(manifest, thresholds=(threshold,)) + else: + review = replace(manifest.review, producer_identity=_HiddenStr(manifest.review.producer_identity)) + manifest = replace(manifest, review=review) + return census, manifest + + +def _reference_with_hidden_scalar(target: str, reference: canonical.ContentRef) -> canonical.ContentRef: + if target == "ref_artifact": + return replace(reference, artifact_id=_HiddenStr(reference.artifact_id)) + if target == "ref_digest": + return replace(reference, content_digest=_HiddenStr(reference.content_digest)) + if target == "ref_media": + return replace(reference, media_type=_HiddenStr(reference.media_type)) + if target == "ref_retention": + return replace(reference, retention_policy_digest=_HiddenStr(reference.retention_policy_digest)) + if target == "ref_receipt": + return replace(reference, storage_receipt_digest=_HiddenStr(reference.storage_receipt_digest)) + return replace(reference, byte_length=_HiddenInt(reference.byte_length)) + + +@pytest.mark.parametrize( + "target", + ( + "census_digest", + "manifest_digest", + "scope_text", + "baseline_text", + "baseline_digest", + "count", + "threshold_target", + "review_identity", + "ref_artifact", + "ref_digest", + "ref_media", + "ref_retention", + "ref_receipt", + "ref_length", + "policy_identity", + "policy_store", + "policy_bytes", + ), +) +def test_recursive_payload_graph_refuses_scalar_subclasses(target: str) -> None: + census, manifest, policy, resolver = _case() + if target.startswith("ref_"): + changed_ref = _reference_with_hidden_scalar(target, manifest.review.conflict_disclosure) + manifest = replace(manifest, review=replace(manifest.review, conflict_disclosure=changed_ref)) + elif target == "policy_identity": + policy = replace(policy, reviewed_reviewer_digests=(_HiddenStr(policy.reviewed_reviewer_digests[0]),)) + elif target == "policy_store": + policy = replace(policy, trusted_store_identities=(_HiddenStr(policy.trusted_store_identities[0]),)) + elif target == "policy_bytes": + policy = replace(policy, maximum_evidence_bytes=_HiddenInt(policy.maximum_evidence_bytes)) + else: + census, manifest = _payload_with_hidden_scalar(target, census, manifest) + + with pytest.raises(AdmissionReviewError, match="closed|exact|scalar|type"): + _validate((census, manifest, policy, resolver)) + + +def test_each_public_digest_entry_closes_its_complete_input_graph() -> None: + census, manifest, _, _ = _case() + hidden_census = replace(census, census_digest=_HiddenStr(census.census_digest)) + for derive_census in (derive_census_payload_digest, derive_census_digest): + with pytest.raises(AdmissionReviewError, match="closed|type"): + derive_census(hidden_census) + + hidden_manifest = replace(manifest, manifest_digest=_HiddenStr(manifest.manifest_digest)) + for derive_manifest in (derive_manifest_payload_digest, derive_manifest_digest): + with pytest.raises(AdmissionReviewError, match="closed|type"): + derive_manifest(hidden_manifest) + + hidden_threshold = replace(manifest.thresholds[0], target=_HiddenFloat(manifest.thresholds[0].target)) + with pytest.raises(AdmissionReviewError, match="closed|type"): + derive_threshold_set_digest((hidden_threshold,)) + + +@pytest.mark.parametrize( + ("member", "hidden_field"), + ( + pytest.param(MetricId.VULNERABLE_RECALL, "case_ids", id="metric-id"), + pytest.param(Statistic.WILSON_LOWER, "per_case_labels", id="statistic"), + pytest.param(Comparator.AT_LEAST, "case_ids", id="comparator"), + ), +) +def test_payload_graph_refuses_mutated_exact_enum_members(member: object, hidden_field: str) -> None: + _, manifest, _, _ = _case() + original_state = dict(vars(member)) + try: + object.__setattr__(member, hidden_field, ("hidden-case-data",)) + with pytest.raises(AdmissionReviewError, match="closed|enum|field|state"): + derive_threshold_set_digest(manifest.thresholds) + finally: + current_state = vars(member) + current_state.clear() + current_state.update(original_state) + assert vars(member) == original_state