From 99d0be19df2d05aeafad7073a1a0368cd8df49e0 Mon Sep 17 00:00:00 2001 From: Erin Spencer Date: Sun, 12 Jul 2026 09:51:34 -0700 Subject: [PATCH 1/9] feat(evidence): add serialized bridge and certification envelopes --- ucns/evidence_envelope.py | 561 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 561 insertions(+) create mode 100644 ucns/evidence_envelope.py diff --git a/ucns/evidence_envelope.py b/ucns/evidence_envelope.py new file mode 100644 index 00000000..ffc71eac --- /dev/null +++ b/ucns/evidence_envelope.py @@ -0,0 +1,561 @@ +"""Versioned serialized evidence envelopes for UCNS consumers. + +Usage guidance +-------------- +Use :func:`bridge_record` to serialize canonical UCNS object identity and typed +domain prerequisite metadata without reconstructing or reimplementing the +algebra. Use :func:`factorization_evidence` to run the authoritative +:func:`ucns.factorization_result` path and serialize its search, coverage, +pruning, and certification evidence. + +Canonical JSON and SHA-256 evidence digests are tamper-evident, not signatures +or proof of who transmitted a record. Consumers must keep UCNS status evidence +separate from their own empirical validity claims. +""" + +# === MODULE_BUILD === +# id: ucns_evidence_envelope +# module_name: evidence_envelope +# module_kind: schema +# summary: versioned deterministic bridge records and factorization evidence envelopes binding UCNS stable identity, canonical serialization, typed domain status, exhaustive-search provenance, catalogue coverage, pruning policy, and negative-certification scope. +# owner: Erin Spencer +# public_surface: BRIDGE_RECORD_SCHEMA_ID, BRIDGE_RECORD_SCHEMA_VERSION, FACTORIZATION_EVIDENCE_SCHEMA_ID, FACTORIZATION_EVIDENCE_SCHEMA_VERSION, UCNSBridgeRecord, UCNSFactorizationEvidence, bridge_record, factorization_evidence +# internal_surface: _canonical_bytes, _digest, _tuple_of_strings, _status_values +# auth_boundary: none +# storage_boundary: deterministic serialization only; no persistence +# network_boundary: none +# user_data_boundary: none +# admin_only: false +# tests: tests.test_evidence_envelope +# rollout: default_enabled +# rollback: remove envelope exports while preserving object_record and factorization_result +# requires: ucns_object_record, ucns_factorization_result, ucns_serialization, ucns_domain_status +# since: 2026-07-12 +# unresolved: cryptographic producer authentication is not provided; evidence digests are tamper-evident content identities only +# === END MODULE_BUILD === + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from typing import Any, Iterable, Mapping, Optional + +from .canonical import UCNSObject +from .factorization_result import ( + NEGATIVE_CERTIFICATION_POLICY_VERSION, + FactorizationResultKind, + factorization_result, +) +from .object_record import object_record +from .serialization import CANONICAL_SERIALIZATION_VERSION, stable_hash + +BRIDGE_RECORD_SCHEMA_ID = "ucns.bridge-record" +BRIDGE_RECORD_SCHEMA_VERSION = "1.0.0" +FACTORIZATION_EVIDENCE_SCHEMA_ID = "ucns.factorization-evidence" +FACTORIZATION_EVIDENCE_SCHEMA_VERSION = "1.0.0" + + +def _canonical_bytes(value: Mapping[str, Any]) -> bytes: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + + +def _digest(value: Mapping[str, Any]) -> str: + return hashlib.sha256(_canonical_bytes(value)).hexdigest() + + +def _tuple_of_strings(name: str, values: Iterable[str]) -> tuple[str, ...]: + result = tuple(values) + for index, value in enumerate(result): + if not isinstance(value, str) or not value: + raise ValueError(f"{name}[{index}] must be a non-empty string") + return result + + +def _status_values(metadata: Any) -> tuple[str, ...]: + return tuple( + status.value if hasattr(status, "value") else str(status) + for status in metadata.statuses + ) + + +@dataclass(frozen=True, slots=True) +class UCNSBridgeRecord: + """Immutable serialized identity and domain-prerequisite record.""" + + object_hash: str + canonical_json: str + domain_label: str + domain_statuses: tuple[str, ...] + completeness_guaranteed: bool + seq_prime_claim_scope: str + depth: int + n_min: int + length: int + is_unit: bool + is_verified_domain: bool + is_frontier: bool + note: str + ucns_serialization_version: str = CANONICAL_SERIALIZATION_VERSION + schema_id: str = BRIDGE_RECORD_SCHEMA_ID + schema_version: str = BRIDGE_RECORD_SCHEMA_VERSION + evidence_digest: str = "" + + def __post_init__(self) -> None: + if self.schema_id != BRIDGE_RECORD_SCHEMA_ID: + raise ValueError(f"unsupported schema_id {self.schema_id!r}") + if self.schema_version != BRIDGE_RECORD_SCHEMA_VERSION: + raise ValueError(f"unsupported schema_version {self.schema_version!r}") + if self.ucns_serialization_version != CANONICAL_SERIALIZATION_VERSION: + raise ValueError( + "unsupported UCNS canonical serialization version " + f"{self.ucns_serialization_version!r}" + ) + if len(self.object_hash) != 64 or any( + character not in "0123456789abcdef" for character in self.object_hash + ): + raise ValueError("object_hash must be a lowercase hexadecimal SHA-256 digest") + if not isinstance(self.canonical_json, str) or not self.canonical_json: + raise ValueError("canonical_json must be non-empty") + + try: + canonical_data = json.loads(self.canonical_json) + except json.JSONDecodeError as exc: + raise ValueError("canonical_json must contain valid JSON") from exc + if not isinstance(canonical_data, dict): + raise ValueError("canonical_json must decode to an object") + if canonical_data.get("version") != self.ucns_serialization_version: + raise ValueError("canonical_json version does not match record serialization version") + observed_hash = hashlib.sha256(self.canonical_json.encode("utf-8")).hexdigest() + if observed_hash != self.object_hash: + raise ValueError("object_hash does not match canonical_json") + + statuses = _tuple_of_strings("domain_statuses", self.domain_statuses) + if not statuses: + raise ValueError("domain_statuses must not be empty") + object.__setattr__(self, "domain_statuses", statuses) + if self.depth < 0 or self.n_min < 1 or self.length < 0: + raise ValueError("depth, n_min, and length must be non-negative with n_min >= 1") + if self.is_verified_domain != self.completeness_guaranteed: + raise ValueError("is_verified_domain must match completeness_guaranteed") + + expected = _digest(self._evidence_fields()) + if self.evidence_digest and self.evidence_digest != expected: + raise ValueError("evidence_digest does not match bridge record contents") + object.__setattr__(self, "evidence_digest", expected) + + def _evidence_fields(self) -> dict[str, Any]: + return { + "schema_id": self.schema_id, + "schema_version": self.schema_version, + "ucns_serialization_version": self.ucns_serialization_version, + "object_hash": self.object_hash, + "canonical_json": self.canonical_json, + "domain_label": self.domain_label, + "domain_statuses": list(self.domain_statuses), + "completeness_guaranteed": self.completeness_guaranteed, + "seq_prime_claim_scope": self.seq_prime_claim_scope, + "depth": self.depth, + "n_min": self.n_min, + "length": self.length, + "is_unit": self.is_unit, + "is_verified_domain": self.is_verified_domain, + "is_frontier": self.is_frontier, + "note": self.note, + } + + def to_dict(self) -> dict[str, Any]: + data = self._evidence_fields() + data["evidence_digest"] = self.evidence_digest + return data + + def to_json(self) -> str: + return _canonical_bytes(self.to_dict()).decode("utf-8") + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> "UCNSBridgeRecord": + expected_fields = { + "schema_id", + "schema_version", + "ucns_serialization_version", + "object_hash", + "canonical_json", + "domain_label", + "domain_statuses", + "completeness_guaranteed", + "seq_prime_claim_scope", + "depth", + "n_min", + "length", + "is_unit", + "is_verified_domain", + "is_frontier", + "note", + "evidence_digest", + } + unknown = set(data) - expected_fields + missing = expected_fields - set(data) + if unknown: + raise ValueError(f"unknown bridge record fields: {sorted(unknown)!r}") + if missing: + raise ValueError(f"missing bridge record fields: {sorted(missing)!r}") + return cls( + schema_id=str(data["schema_id"]), + schema_version=str(data["schema_version"]), + ucns_serialization_version=str(data["ucns_serialization_version"]), + object_hash=str(data["object_hash"]), + canonical_json=str(data["canonical_json"]), + domain_label=str(data["domain_label"]), + domain_statuses=tuple(data["domain_statuses"]), + completeness_guaranteed=bool(data["completeness_guaranteed"]), + seq_prime_claim_scope=str(data["seq_prime_claim_scope"]), + depth=int(data["depth"]), + n_min=int(data["n_min"]), + length=int(data["length"]), + is_unit=bool(data["is_unit"]), + is_verified_domain=bool(data["is_verified_domain"]), + is_frontier=bool(data["is_frontier"]), + note=str(data["note"]), + evidence_digest=str(data["evidence_digest"]), + ) + + @classmethod + def from_json(cls, value: str) -> "UCNSBridgeRecord": + decoded = json.loads(value) + if not isinstance(decoded, dict): + raise ValueError("bridge record JSON must decode to an object") + return cls.from_dict(decoded) + + +@dataclass(frozen=True, slots=True) +class UCNSFactorizationEvidence: + """Immutable serialized output from UCNS's authoritative result policy.""" + + product_hash: str + product_domain_label: str + product_domain_statuses: tuple[str, ...] + completeness_guaranteed: bool + result_kind: str + factor_hashes: tuple[str, ...] + negative_result_certified: bool + seq_prime_is_absolute: bool + claim_scope: str + note: str + certification_policy_version: str + search_exhausted: bool + truncation_occurred: bool + catalogue_source: str + supplied_catalogue_size: int + supplied_catalogue_fingerprint: str + effective_catalogue_size: int + effective_catalogue_fingerprint: str + catalogue_coverage_status: str + catalogue_coverage_reason: str + catalogue_coverage_rule_version: str + required_catalogue_rule_version: str + required_catalogue_fingerprint: str + coverage_record_validated: bool + coverage_bound_to_search_report: bool + pruning_applied: bool + pruning_rule: str + pruning_rule_version: str + pruning_preserves_coverage: bool + uncertified_reasons: tuple[str, ...] + schema_id: str = FACTORIZATION_EVIDENCE_SCHEMA_ID + schema_version: str = FACTORIZATION_EVIDENCE_SCHEMA_VERSION + evidence_digest: str = "" + + def __post_init__(self) -> None: + if self.schema_id != FACTORIZATION_EVIDENCE_SCHEMA_ID: + raise ValueError(f"unsupported schema_id {self.schema_id!r}") + if self.schema_version != FACTORIZATION_EVIDENCE_SCHEMA_VERSION: + raise ValueError(f"unsupported schema_version {self.schema_version!r}") + if len(self.product_hash) != 64 or any( + character not in "0123456789abcdef" for character in self.product_hash + ): + raise ValueError("product_hash must be a lowercase hexadecimal SHA-256 digest") + statuses = _tuple_of_strings("product_domain_statuses", self.product_domain_statuses) + object.__setattr__(self, "product_domain_statuses", statuses) + factor_hashes = tuple(self.factor_hashes) + for value in factor_hashes: + if len(value) != 64 or any( + character not in "0123456789abcdef" for character in value + ): + raise ValueError("factor_hashes must contain SHA-256 digests") + object.__setattr__(self, "factor_hashes", factor_hashes) + reasons = tuple(str(value) for value in self.uncertified_reasons) + object.__setattr__(self, "uncertified_reasons", reasons) + + allowed = {kind.value for kind in FactorizationResultKind} + if self.result_kind not in allowed: + raise ValueError(f"unsupported result_kind {self.result_kind!r}") + if self.result_kind == FactorizationResultKind.FACTORS.value: + if len(self.factor_hashes) != 2: + raise ValueError("FACTORS evidence must contain exactly two factor hashes") + if self.negative_result_certified or self.seq_prime_is_absolute: + raise ValueError("factor evidence cannot claim negative certification") + else: + if self.factor_hashes: + raise ValueError("SEQ-PRIME evidence must not contain factor hashes") + + if self.negative_result_certified != self.seq_prime_is_absolute: + raise ValueError( + "negative_result_certified and seq_prime_is_absolute must agree" + ) + if self.negative_result_certified: + certification_requirements = { + "complete_domain": self.completeness_guaranteed, + "policy_version": ( + self.certification_policy_version + == NEGATIVE_CERTIFICATION_POLICY_VERSION + ), + "search_exhausted": self.search_exhausted, + "not_truncated": not self.truncation_occurred, + "coverage_validated": self.coverage_record_validated, + "coverage_bound": self.coverage_bound_to_search_report, + "pruning_preserves_coverage": self.pruning_preserves_coverage, + "no_uncertified_reasons": not self.uncertified_reasons, + } + failed = [name for name, passed in certification_requirements.items() if not passed] + if failed: + raise ValueError( + "certified negative evidence violates policy requirements: " + + ", ".join(failed) + ) + + for name in ("supplied_catalogue_size", "effective_catalogue_size"): + if getattr(self, name) < 0: + raise ValueError(f"{name} must be non-negative") + + expected = _digest(self._evidence_fields()) + if self.evidence_digest and self.evidence_digest != expected: + raise ValueError("evidence_digest does not match factorization evidence contents") + object.__setattr__(self, "evidence_digest", expected) + + def _evidence_fields(self) -> dict[str, Any]: + return { + "schema_id": self.schema_id, + "schema_version": self.schema_version, + "product_hash": self.product_hash, + "product_domain_label": self.product_domain_label, + "product_domain_statuses": list(self.product_domain_statuses), + "completeness_guaranteed": self.completeness_guaranteed, + "result_kind": self.result_kind, + "factor_hashes": list(self.factor_hashes), + "negative_result_certified": self.negative_result_certified, + "seq_prime_is_absolute": self.seq_prime_is_absolute, + "claim_scope": self.claim_scope, + "note": self.note, + "certification_policy_version": self.certification_policy_version, + "search_exhausted": self.search_exhausted, + "truncation_occurred": self.truncation_occurred, + "catalogue_source": self.catalogue_source, + "supplied_catalogue_size": self.supplied_catalogue_size, + "supplied_catalogue_fingerprint": self.supplied_catalogue_fingerprint, + "effective_catalogue_size": self.effective_catalogue_size, + "effective_catalogue_fingerprint": self.effective_catalogue_fingerprint, + "catalogue_coverage_status": self.catalogue_coverage_status, + "catalogue_coverage_reason": self.catalogue_coverage_reason, + "catalogue_coverage_rule_version": self.catalogue_coverage_rule_version, + "required_catalogue_rule_version": self.required_catalogue_rule_version, + "required_catalogue_fingerprint": self.required_catalogue_fingerprint, + "coverage_record_validated": self.coverage_record_validated, + "coverage_bound_to_search_report": self.coverage_bound_to_search_report, + "pruning_applied": self.pruning_applied, + "pruning_rule": self.pruning_rule, + "pruning_rule_version": self.pruning_rule_version, + "pruning_preserves_coverage": self.pruning_preserves_coverage, + "uncertified_reasons": list(self.uncertified_reasons), + } + + def to_dict(self) -> dict[str, Any]: + data = self._evidence_fields() + data["evidence_digest"] = self.evidence_digest + return data + + def to_json(self) -> str: + return _canonical_bytes(self.to_dict()).decode("utf-8") + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> "UCNSFactorizationEvidence": + expected_fields = { + "schema_id", + "schema_version", + "product_hash", + "product_domain_label", + "product_domain_statuses", + "completeness_guaranteed", + "result_kind", + "factor_hashes", + "negative_result_certified", + "seq_prime_is_absolute", + "claim_scope", + "note", + "certification_policy_version", + "search_exhausted", + "truncation_occurred", + "catalogue_source", + "supplied_catalogue_size", + "supplied_catalogue_fingerprint", + "effective_catalogue_size", + "effective_catalogue_fingerprint", + "catalogue_coverage_status", + "catalogue_coverage_reason", + "catalogue_coverage_rule_version", + "required_catalogue_rule_version", + "required_catalogue_fingerprint", + "coverage_record_validated", + "coverage_bound_to_search_report", + "pruning_applied", + "pruning_rule", + "pruning_rule_version", + "pruning_preserves_coverage", + "uncertified_reasons", + "evidence_digest", + } + unknown = set(data) - expected_fields + missing = expected_fields - set(data) + if unknown: + raise ValueError( + f"unknown factorization evidence fields: {sorted(unknown)!r}" + ) + if missing: + raise ValueError( + f"missing factorization evidence fields: {sorted(missing)!r}" + ) + return cls( + schema_id=str(data["schema_id"]), + schema_version=str(data["schema_version"]), + product_hash=str(data["product_hash"]), + product_domain_label=str(data["product_domain_label"]), + product_domain_statuses=tuple(data["product_domain_statuses"]), + completeness_guaranteed=bool(data["completeness_guaranteed"]), + result_kind=str(data["result_kind"]), + factor_hashes=tuple(data["factor_hashes"]), + negative_result_certified=bool(data["negative_result_certified"]), + seq_prime_is_absolute=bool(data["seq_prime_is_absolute"]), + claim_scope=str(data["claim_scope"]), + note=str(data["note"]), + certification_policy_version=str(data["certification_policy_version"]), + search_exhausted=bool(data["search_exhausted"]), + truncation_occurred=bool(data["truncation_occurred"]), + catalogue_source=str(data["catalogue_source"]), + supplied_catalogue_size=int(data["supplied_catalogue_size"]), + supplied_catalogue_fingerprint=str(data["supplied_catalogue_fingerprint"]), + effective_catalogue_size=int(data["effective_catalogue_size"]), + effective_catalogue_fingerprint=str(data["effective_catalogue_fingerprint"]), + catalogue_coverage_status=str(data["catalogue_coverage_status"]), + catalogue_coverage_reason=str(data["catalogue_coverage_reason"]), + catalogue_coverage_rule_version=str(data["catalogue_coverage_rule_version"]), + required_catalogue_rule_version=str(data["required_catalogue_rule_version"]), + required_catalogue_fingerprint=str(data["required_catalogue_fingerprint"]), + coverage_record_validated=bool(data["coverage_record_validated"]), + coverage_bound_to_search_report=bool(data["coverage_bound_to_search_report"]), + pruning_applied=bool(data["pruning_applied"]), + pruning_rule=str(data["pruning_rule"]), + pruning_rule_version=str(data["pruning_rule_version"]), + pruning_preserves_coverage=bool(data["pruning_preserves_coverage"]), + uncertified_reasons=tuple(data["uncertified_reasons"]), + evidence_digest=str(data["evidence_digest"]), + ) + + @classmethod + def from_json(cls, value: str) -> "UCNSFactorizationEvidence": + decoded = json.loads(value) + if not isinstance(decoded, dict): + raise ValueError("factorization evidence JSON must decode to an object") + return cls.from_dict(decoded) + + +def bridge_record(obj: Optional[UCNSObject]) -> UCNSBridgeRecord: + """Build a serialized bridge record from the canonical object inspector.""" + + record = object_record(obj) + metadata = record.domain_metadata + return UCNSBridgeRecord( + object_hash=str(record.object_hash), + canonical_json=str(record.canonical_json), + domain_label=str(record.domain_label), + domain_statuses=_status_values(metadata), + completeness_guaranteed=bool(metadata.completeness_guaranteed), + seq_prime_claim_scope=str(metadata.seq_prime_claim_scope), + depth=int(record.depth), + n_min=int(record.n_min), + length=int(record.length), + is_unit=bool(record.is_unit), + is_verified_domain=bool(record.is_verified_domain), + is_frontier=bool(record.is_frontier), + note=str(record.note), + ) + + +def factorization_evidence( + product: Optional[UCNSObject], + catalogue: Optional[Iterable[Optional[UCNSObject]]] = None, +) -> UCNSFactorizationEvidence: + """Run the authoritative UCNS result policy and serialize its evidence. + + No caller-supplied result object or certification boolean is accepted. + """ + + result = factorization_result(product, catalogue=catalogue) + metadata = result.product_domain_metadata + factor_hashes = ( + tuple(stable_hash(factor) for factor in result.factors) + if result.factors is not None + else () + ) + result_kind = ( + result.result_kind.value + if hasattr(result.result_kind, "value") + else str(result.result_kind) + ) + return UCNSFactorizationEvidence( + product_hash=str(result.product_hash), + product_domain_label=str(result.product_domain_label), + product_domain_statuses=_status_values(metadata), + completeness_guaranteed=bool(metadata.completeness_guaranteed), + result_kind=result_kind, + factor_hashes=factor_hashes, + negative_result_certified=bool(result.negative_result_certified), + seq_prime_is_absolute=bool(result.seq_prime_is_absolute), + claim_scope=str(result.claim_scope), + note=str(result.note), + certification_policy_version=str(result.certification_policy_version), + search_exhausted=bool(result.search_exhausted), + truncation_occurred=bool(result.truncation_occurred), + catalogue_source=str(result.catalogue_source), + supplied_catalogue_size=int(result.supplied_catalogue_size), + supplied_catalogue_fingerprint=str(result.supplied_catalogue_fingerprint), + effective_catalogue_size=int(result.effective_catalogue_size), + effective_catalogue_fingerprint=str(result.effective_catalogue_fingerprint), + catalogue_coverage_status=str(result.catalogue_coverage_status), + catalogue_coverage_reason=str(result.catalogue_coverage_reason), + catalogue_coverage_rule_version=str(result.catalogue_coverage_rule_version), + required_catalogue_rule_version=str(result.required_catalogue_rule_version), + required_catalogue_fingerprint=str(result.required_catalogue_fingerprint), + coverage_record_validated=bool(result.coverage_record_validated), + coverage_bound_to_search_report=bool(result.coverage_bound_to_search_report), + pruning_applied=bool(result.pruning_applied), + pruning_rule=str(result.pruning_rule), + pruning_rule_version=str(result.pruning_rule_version), + pruning_preserves_coverage=bool(result.pruning_preserves_coverage), + uncertified_reasons=tuple(result.uncertified_reasons), + ) + + +__all__ = [ + "BRIDGE_RECORD_SCHEMA_ID", + "BRIDGE_RECORD_SCHEMA_VERSION", + "FACTORIZATION_EVIDENCE_SCHEMA_ID", + "FACTORIZATION_EVIDENCE_SCHEMA_VERSION", + "UCNSBridgeRecord", + "UCNSFactorizationEvidence", + "bridge_record", + "factorization_evidence", +] From 73068dbbb3546976eb6e4e25f6dd0ec2aa81be3c Mon Sep 17 00:00:00 2001 From: Erin Spencer Date: Sun, 12 Jul 2026 09:53:10 -0700 Subject: [PATCH 2/9] feat(api): export UCNS evidence envelopes --- ucns/__init__.py | 27 ++++++++++++++++++++++++--- 1 file changed, 24 insertions(+), 3 deletions(-) diff --git a/ucns/__init__.py b/ucns/__init__.py index a6e47850..c5ce1634 100644 --- a/ucns/__init__.py +++ b/ucns/__init__.py @@ -7,9 +7,9 @@ Core recursive modules include ``canonical``, ``domains``, ``host_recovery``, ``payload_system``, ``witness_matrix``, ``factor_search_v08``, ``domain_status``, ``serialization``, ``factorization_result``, -``object_record``, ``recursive_codec``, ``left_quotient``, ``store``, -``catalogue``, ``canonical_factorization``, ``catalogue_pruning``, and -``geometry_bridge``. +``object_record``, ``evidence_envelope``, ``recursive_codec``, +``left_quotient``, ``store``, ``catalogue``, ``canonical_factorization``, +``catalogue_pruning``, and ``geometry_bridge``. """ from .canonical import ( @@ -95,6 +95,18 @@ # Object inspection record from .object_record import UCNSObjectRecord, object_record +# Versioned serialized evidence envelopes +from .evidence_envelope import ( + BRIDGE_RECORD_SCHEMA_ID, + BRIDGE_RECORD_SCHEMA_VERSION, + FACTORIZATION_EVIDENCE_SCHEMA_ID, + FACTORIZATION_EVIDENCE_SCHEMA_VERSION, + UCNSBridgeRecord, + UCNSFactorizationEvidence, + bridge_record, + factorization_evidence, +) + # Retrieval from .store import Match, OutOfDomainError, UCNSStore @@ -188,6 +200,15 @@ # object inspection record "UCNSObjectRecord", "object_record", + # versioned serialized evidence envelopes + "BRIDGE_RECORD_SCHEMA_ID", + "BRIDGE_RECORD_SCHEMA_VERSION", + "FACTORIZATION_EVIDENCE_SCHEMA_ID", + "FACTORIZATION_EVIDENCE_SCHEMA_VERSION", + "UCNSBridgeRecord", + "UCNSFactorizationEvidence", + "bridge_record", + "factorization_evidence", # retrieval "UCNSStore", "Match", From 3a6352d78eb3d4757381cbb73b0f4d7207f879fb Mon Sep 17 00:00:00 2001 From: Erin Spencer Date: Sun, 12 Jul 2026 09:53:47 -0700 Subject: [PATCH 3/9] test(evidence): cover bridge and certification envelopes --- tests/test_evidence_envelope.py | 149 ++++++++++++++++++++++++++++++++ 1 file changed, 149 insertions(+) create mode 100644 tests/test_evidence_envelope.py diff --git a/tests/test_evidence_envelope.py b/tests/test_evidence_envelope.py new file mode 100644 index 00000000..21030814 --- /dev/null +++ b/tests/test_evidence_envelope.py @@ -0,0 +1,149 @@ +from __future__ import annotations + +from dataclasses import replace +from fractions import Fraction + +import pytest + +from ucns import ( + BRIDGE_RECORD_SCHEMA_ID, + BRIDGE_RECORD_SCHEMA_VERSION, + FACTORIZATION_EVIDENCE_SCHEMA_ID, + FACTORIZATION_EVIDENCE_SCHEMA_VERSION, + NEGATIVE_CERTIFICATION_POLICY_VERSION, + S2, + UNIT, + FactorizationResultKind, + UCNSBridgeRecord, + UCNSFactorizationEvidence, + UCNSObject, + bridge_record, + factorization_evidence, + stable_hash, +) + +E = UCNSObject(1, 1, [(Fraction(0), UNIT)], [0]) +T2 = UCNSObject(1, 1, [(Fraction(0), E)], [0]) + + +def test_bridge_record_binds_canonical_json_stable_hash_and_domain_status() -> None: + record = bridge_record(S2) + + assert record.schema_id == BRIDGE_RECORD_SCHEMA_ID + assert record.schema_version == BRIDGE_RECORD_SCHEMA_VERSION + assert record.object_hash == stable_hash(S2) + assert record.domain_label == "depth-1" + assert record.domain_statuses == ("DEFENDED", "TEST_BACKED") + assert record.completeness_guaranteed is True + assert record.is_verified_domain is True + assert record.evidence_digest + + +def test_bridge_record_roundtrips_through_canonical_json() -> None: + record = bridge_record(S2) + assert UCNSBridgeRecord.from_dict(record.to_dict()) == record + assert UCNSBridgeRecord.from_json(record.to_json()) == record + assert record.to_json() == UCNSBridgeRecord.from_json(record.to_json()).to_json() + + +def test_unit_bridge_record_is_typed_and_not_a_prime_claim() -> None: + record = bridge_record(UNIT) + assert record.is_unit is True + assert record.length == 0 + assert record.domain_label == "depth-0" + assert record.seq_prime_claim_scope == "not-prime-unit-domain" + + +def test_bridge_record_rejects_hash_tamper_and_unknown_fields() -> None: + record = bridge_record(S2) + tampered = record.to_dict() + tampered["object_hash"] = "0" * 64 + tampered["evidence_digest"] = "" + with pytest.raises(ValueError, match="object_hash does not match"): + UCNSBridgeRecord.from_dict(tampered) + + unknown = record.to_dict() + unknown["invented"] = True + with pytest.raises(ValueError, match="unknown bridge record fields"): + UCNSBridgeRecord.from_dict(unknown) + + +def test_flat_prime_serializes_certified_negative_evidence() -> None: + evidence = factorization_evidence(S2) + + assert evidence.schema_id == FACTORIZATION_EVIDENCE_SCHEMA_ID + assert evidence.schema_version == FACTORIZATION_EVIDENCE_SCHEMA_VERSION + assert evidence.product_hash == stable_hash(S2) + assert evidence.result_kind == FactorizationResultKind.SEQ_PRIME.value + assert evidence.factor_hashes == () + assert evidence.negative_result_certified is True + assert evidence.seq_prime_is_absolute is True + assert evidence.certification_policy_version == NEGATIVE_CERTIFICATION_POLICY_VERSION + assert evidence.search_exhausted is True + assert evidence.truncation_occurred is False + assert evidence.coverage_record_validated is True + assert evidence.coverage_bound_to_search_report is True + assert evidence.pruning_preserves_coverage is True + assert evidence.uncertified_reasons == () + + +def test_certified_evidence_roundtrips_without_losing_policy_fields() -> None: + evidence = factorization_evidence(S2) + assert UCNSFactorizationEvidence.from_dict(evidence.to_dict()) == evidence + assert UCNSFactorizationEvidence.from_json(evidence.to_json()) == evidence + + +def test_unit_result_remains_explicitly_uncertified() -> None: + evidence = factorization_evidence(UNIT) + assert evidence.result_kind == FactorizationResultKind.SEQ_PRIME.value + assert evidence.negative_result_certified is False + assert evidence.seq_prime_is_absolute is False + assert evidence.claim_scope == "not-prime-unit-domain" + assert evidence.search_exhausted is False + assert evidence.uncertified_reasons == ("unit-domain-primality-inapplicable",) + + +def test_incomplete_catalogue_remains_uncertified_despite_exhaustion() -> None: + evidence = factorization_evidence(S2, catalogue=[]) + assert evidence.search_exhausted is True + assert evidence.negative_result_certified is False + assert evidence.seq_prime_is_absolute is False + assert any( + reason.startswith("catalogue-coverage-uncertified") + for reason in evidence.uncertified_reasons + ) + + +def test_recovered_factors_are_serialized_by_stable_hash_only() -> None: + evidence = factorization_evidence(T2) + assert evidence.result_kind == FactorizationResultKind.FACTORS.value + assert len(evidence.factor_hashes) == 2 + assert evidence.negative_result_certified is False + assert evidence.seq_prime_is_absolute is False + assert evidence.uncertified_reasons == ("factors-found",) + + +def test_certified_record_with_broken_policy_invariant_fails_closed() -> None: + evidence = factorization_evidence(S2) + with pytest.raises(ValueError, match="search_exhausted"): + replace(evidence, search_exhausted=False, evidence_digest="") + + with pytest.raises(ValueError, match="policy_version"): + replace( + evidence, + certification_policy_version="invented-policy", + evidence_digest="", + ) + + +def test_factorization_evidence_rejects_digest_tamper_and_unknown_fields() -> None: + evidence = factorization_evidence(S2) + tampered = evidence.to_dict() + tampered["claim_scope"] = "invented" + with pytest.raises(ValueError, match="evidence_digest does not match"): + UCNSFactorizationEvidence.from_dict(tampered) + + unknown = evidence.to_dict() + unknown["invented"] = True + with pytest.raises(ValueError, match="unknown factorization evidence fields"): + UCNSFactorizationEvidence.from_dict(unknown) From 6e8cb1a63ca2f99e0f44b36f33986338ad1f52c7 Mon Sep 17 00:00:00 2001 From: Erin Spencer Date: Sun, 12 Jul 2026 09:57:05 -0700 Subject: [PATCH 4/9] fix(evidence): support Python 3.8 and enforce strict serialized types --- ucns/evidence_envelope.py | 369 +++++++++++++++++++++++++------------- 1 file changed, 248 insertions(+), 121 deletions(-) diff --git a/ucns/evidence_envelope.py b/ucns/evidence_envelope.py index ffc71eac..b5d4fedc 100644 --- a/ucns/evidence_envelope.py +++ b/ucns/evidence_envelope.py @@ -20,7 +20,7 @@ # summary: versioned deterministic bridge records and factorization evidence envelopes binding UCNS stable identity, canonical serialization, typed domain status, exhaustive-search provenance, catalogue coverage, pruning policy, and negative-certification scope. # owner: Erin Spencer # public_surface: BRIDGE_RECORD_SCHEMA_ID, BRIDGE_RECORD_SCHEMA_VERSION, FACTORIZATION_EVIDENCE_SCHEMA_ID, FACTORIZATION_EVIDENCE_SCHEMA_VERSION, UCNSBridgeRecord, UCNSFactorizationEvidence, bridge_record, factorization_evidence -# internal_surface: _canonical_bytes, _digest, _tuple_of_strings, _status_values +# internal_surface: _canonical_bytes, _digest, _exact_fields, _strict_bool, _strict_int, _strict_str, _strict_string_tuple, _strict_hex_digest, _status_values # auth_boundary: none # storage_boundary: deterministic serialization only; no persistence # network_boundary: none @@ -39,7 +39,7 @@ import hashlib import json from dataclasses import dataclass -from typing import Any, Iterable, Mapping, Optional +from typing import Any, Iterable, Mapping, Optional, Sequence, Tuple from .canonical import UCNSObject from .factorization_result import ( @@ -54,6 +54,9 @@ BRIDGE_RECORD_SCHEMA_VERSION = "1.0.0" FACTORIZATION_EVIDENCE_SCHEMA_ID = "ucns.factorization-evidence" FACTORIZATION_EVIDENCE_SCHEMA_VERSION = "1.0.0" +BRIDGE_RECORD_PRODUCER_ID = "ucns.object_record" +FACTORIZATION_EVIDENCE_PRODUCER_ID = "ucns.factorization_result" +_HEX = frozenset("0123456789abcdef") def _canonical_bytes(value: Mapping[str, Any]) -> bytes: @@ -69,29 +72,77 @@ def _digest(value: Mapping[str, Any]) -> str: return hashlib.sha256(_canonical_bytes(value)).hexdigest() -def _tuple_of_strings(name: str, values: Iterable[str]) -> tuple[str, ...]: - result = tuple(values) - for index, value in enumerate(result): - if not isinstance(value, str) or not value: - raise ValueError(f"{name}[{index}] must be a non-empty string") +def _exact_fields(data: Mapping[str, Any], expected: Sequence[str], label: str) -> None: + expected_set = set(expected) + unknown = set(data) - expected_set + missing = expected_set - set(data) + if unknown: + raise ValueError("unknown %s fields: %r" % (label, sorted(unknown))) + if missing: + raise ValueError("missing %s fields: %r" % (label, sorted(missing))) + + +def _strict_bool(data: Mapping[str, Any], name: str) -> bool: + value = data[name] + if type(value) is not bool: + raise ValueError("%s must be a boolean" % name) + return value + + +def _strict_int(data: Mapping[str, Any], name: str) -> int: + value = data[name] + if type(value) is not int: + raise ValueError("%s must be an integer" % name) + return value + + +def _strict_str(data: Mapping[str, Any], name: str, allow_empty: bool = False) -> str: + value = data[name] + if not isinstance(value, str): + raise ValueError("%s must be a string" % name) + if not allow_empty and not value: + raise ValueError("%s must be non-empty" % name) + return value + + +def _strict_string_tuple( + data: Mapping[str, Any], + name: str, + allow_empty: bool = True, +) -> Tuple[str, ...]: + value = data[name] + if not isinstance(value, (list, tuple)): + raise ValueError("%s must be a list or tuple of strings" % name) + result = tuple(value) + if not allow_empty and not result: + raise ValueError("%s must not be empty" % name) + for index, item in enumerate(result): + if not isinstance(item, str) or not item: + raise ValueError("%s[%d] must be a non-empty string" % (name, index)) return result -def _status_values(metadata: Any) -> tuple[str, ...]: +def _strict_hex_digest(value: str, name: str) -> str: + if len(value) != 64 or any(character not in _HEX for character in value): + raise ValueError("%s must be a lowercase hexadecimal SHA-256 digest" % name) + return value + + +def _status_values(metadata: Any) -> Tuple[str, ...]: return tuple( status.value if hasattr(status, "value") else str(status) for status in metadata.statuses ) -@dataclass(frozen=True, slots=True) +@dataclass(frozen=True) class UCNSBridgeRecord: """Immutable serialized identity and domain-prerequisite record.""" object_hash: str canonical_json: str domain_label: str - domain_statuses: tuple[str, ...] + domain_statuses: Tuple[str, ...] completeness_guaranteed: bool seq_prime_claim_scope: str depth: int @@ -102,24 +153,24 @@ class UCNSBridgeRecord: is_frontier: bool note: str ucns_serialization_version: str = CANONICAL_SERIALIZATION_VERSION + producer_id: str = BRIDGE_RECORD_PRODUCER_ID schema_id: str = BRIDGE_RECORD_SCHEMA_ID schema_version: str = BRIDGE_RECORD_SCHEMA_VERSION evidence_digest: str = "" def __post_init__(self) -> None: if self.schema_id != BRIDGE_RECORD_SCHEMA_ID: - raise ValueError(f"unsupported schema_id {self.schema_id!r}") + raise ValueError("unsupported schema_id %r" % self.schema_id) if self.schema_version != BRIDGE_RECORD_SCHEMA_VERSION: - raise ValueError(f"unsupported schema_version {self.schema_version!r}") + raise ValueError("unsupported schema_version %r" % self.schema_version) + if self.producer_id != BRIDGE_RECORD_PRODUCER_ID: + raise ValueError("unsupported producer_id %r" % self.producer_id) if self.ucns_serialization_version != CANONICAL_SERIALIZATION_VERSION: raise ValueError( - "unsupported UCNS canonical serialization version " - f"{self.ucns_serialization_version!r}" + "unsupported UCNS canonical serialization version %r" + % self.ucns_serialization_version ) - if len(self.object_hash) != 64 or any( - character not in "0123456789abcdef" for character in self.object_hash - ): - raise ValueError("object_hash must be a lowercase hexadecimal SHA-256 digest") + _strict_hex_digest(self.object_hash, "object_hash") if not isinstance(self.canonical_json, str) or not self.canonical_json: raise ValueError("canonical_json must be non-empty") @@ -131,16 +182,31 @@ def __post_init__(self) -> None: raise ValueError("canonical_json must decode to an object") if canonical_data.get("version") != self.ucns_serialization_version: raise ValueError("canonical_json version does not match record serialization version") + normalized_canonical_json = json.dumps( + canonical_data, + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + ) + if normalized_canonical_json != self.canonical_json: + raise ValueError("canonical_json is not in canonical UCNS form") observed_hash = hashlib.sha256(self.canonical_json.encode("utf-8")).hexdigest() if observed_hash != self.object_hash: raise ValueError("object_hash does not match canonical_json") - statuses = _tuple_of_strings("domain_statuses", self.domain_statuses) + statuses = tuple(self.domain_statuses) if not statuses: raise ValueError("domain_statuses must not be empty") + for index, value in enumerate(statuses): + if not isinstance(value, str) or not value: + raise ValueError( + "domain_statuses[%d] must be a non-empty string" % index + ) object.__setattr__(self, "domain_statuses", statuses) if self.depth < 0 or self.n_min < 1 or self.length < 0: - raise ValueError("depth, n_min, and length must be non-negative with n_min >= 1") + raise ValueError( + "depth, n_min, and length must be non-negative with n_min >= 1" + ) if self.is_verified_domain != self.completeness_guaranteed: raise ValueError("is_verified_domain must match completeness_guaranteed") @@ -149,10 +215,11 @@ def __post_init__(self) -> None: raise ValueError("evidence_digest does not match bridge record contents") object.__setattr__(self, "evidence_digest", expected) - def _evidence_fields(self) -> dict[str, Any]: + def _evidence_fields(self) -> dict: return { "schema_id": self.schema_id, "schema_version": self.schema_version, + "producer_id": self.producer_id, "ucns_serialization_version": self.ucns_serialization_version, "object_hash": self.object_hash, "canonical_json": self.canonical_json, @@ -169,7 +236,7 @@ def _evidence_fields(self) -> dict[str, Any]: "note": self.note, } - def to_dict(self) -> dict[str, Any]: + def to_dict(self) -> dict: data = self._evidence_fields() data["evidence_digest"] = self.evidence_digest return data @@ -179,9 +246,10 @@ def to_json(self) -> str: @classmethod def from_dict(cls, data: Mapping[str, Any]) -> "UCNSBridgeRecord": - expected_fields = { + expected = ( "schema_id", "schema_version", + "producer_id", "ucns_serialization_version", "object_hash", "canonical_json", @@ -197,51 +265,55 @@ def from_dict(cls, data: Mapping[str, Any]) -> "UCNSBridgeRecord": "is_frontier", "note", "evidence_digest", - } - unknown = set(data) - expected_fields - missing = expected_fields - set(data) - if unknown: - raise ValueError(f"unknown bridge record fields: {sorted(unknown)!r}") - if missing: - raise ValueError(f"missing bridge record fields: {sorted(missing)!r}") + ) + _exact_fields(data, expected, "bridge record") return cls( - schema_id=str(data["schema_id"]), - schema_version=str(data["schema_version"]), - ucns_serialization_version=str(data["ucns_serialization_version"]), - object_hash=str(data["object_hash"]), - canonical_json=str(data["canonical_json"]), - domain_label=str(data["domain_label"]), - domain_statuses=tuple(data["domain_statuses"]), - completeness_guaranteed=bool(data["completeness_guaranteed"]), - seq_prime_claim_scope=str(data["seq_prime_claim_scope"]), - depth=int(data["depth"]), - n_min=int(data["n_min"]), - length=int(data["length"]), - is_unit=bool(data["is_unit"]), - is_verified_domain=bool(data["is_verified_domain"]), - is_frontier=bool(data["is_frontier"]), - note=str(data["note"]), - evidence_digest=str(data["evidence_digest"]), + schema_id=_strict_str(data, "schema_id"), + schema_version=_strict_str(data, "schema_version"), + producer_id=_strict_str(data, "producer_id"), + ucns_serialization_version=_strict_str( + data, "ucns_serialization_version" + ), + object_hash=_strict_str(data, "object_hash"), + canonical_json=_strict_str(data, "canonical_json"), + domain_label=_strict_str(data, "domain_label"), + domain_statuses=_strict_string_tuple( + data, "domain_statuses", allow_empty=False + ), + completeness_guaranteed=_strict_bool( + data, "completeness_guaranteed" + ), + seq_prime_claim_scope=_strict_str(data, "seq_prime_claim_scope"), + depth=_strict_int(data, "depth"), + n_min=_strict_int(data, "n_min"), + length=_strict_int(data, "length"), + is_unit=_strict_bool(data, "is_unit"), + is_verified_domain=_strict_bool(data, "is_verified_domain"), + is_frontier=_strict_bool(data, "is_frontier"), + note=_strict_str(data, "note"), + evidence_digest=_strict_str(data, "evidence_digest"), ) @classmethod def from_json(cls, value: str) -> "UCNSBridgeRecord": + if not isinstance(value, str): + raise ValueError("bridge record JSON must be a string") decoded = json.loads(value) if not isinstance(decoded, dict): raise ValueError("bridge record JSON must decode to an object") return cls.from_dict(decoded) -@dataclass(frozen=True, slots=True) +@dataclass(frozen=True) class UCNSFactorizationEvidence: """Immutable serialized output from UCNS's authoritative result policy.""" product_hash: str product_domain_label: str - product_domain_statuses: tuple[str, ...] + product_domain_statuses: Tuple[str, ...] completeness_guaranteed: bool result_kind: str - factor_hashes: tuple[str, ...] + factor_hashes: Tuple[str, ...] negative_result_certified: bool seq_prime_is_absolute: bool claim_scope: str @@ -265,43 +337,65 @@ class UCNSFactorizationEvidence: pruning_rule: str pruning_rule_version: str pruning_preserves_coverage: bool - uncertified_reasons: tuple[str, ...] + uncertified_reasons: Tuple[str, ...] + producer_id: str = FACTORIZATION_EVIDENCE_PRODUCER_ID schema_id: str = FACTORIZATION_EVIDENCE_SCHEMA_ID schema_version: str = FACTORIZATION_EVIDENCE_SCHEMA_VERSION evidence_digest: str = "" def __post_init__(self) -> None: if self.schema_id != FACTORIZATION_EVIDENCE_SCHEMA_ID: - raise ValueError(f"unsupported schema_id {self.schema_id!r}") + raise ValueError("unsupported schema_id %r" % self.schema_id) if self.schema_version != FACTORIZATION_EVIDENCE_SCHEMA_VERSION: - raise ValueError(f"unsupported schema_version {self.schema_version!r}") - if len(self.product_hash) != 64 or any( - character not in "0123456789abcdef" for character in self.product_hash - ): - raise ValueError("product_hash must be a lowercase hexadecimal SHA-256 digest") - statuses = _tuple_of_strings("product_domain_statuses", self.product_domain_statuses) + raise ValueError("unsupported schema_version %r" % self.schema_version) + if self.producer_id != FACTORIZATION_EVIDENCE_PRODUCER_ID: + raise ValueError("unsupported producer_id %r" % self.producer_id) + _strict_hex_digest(self.product_hash, "product_hash") + if self.certification_policy_version != NEGATIVE_CERTIFICATION_POLICY_VERSION: + raise ValueError("unsupported certification_policy_version %r" % ( + self.certification_policy_version, + )) + + statuses = tuple(self.product_domain_statuses) + if not statuses: + raise ValueError("product_domain_statuses must not be empty") + for index, value in enumerate(statuses): + if not isinstance(value, str) or not value: + raise ValueError( + "product_domain_statuses[%d] must be a non-empty string" % index + ) object.__setattr__(self, "product_domain_statuses", statuses) + factor_hashes = tuple(self.factor_hashes) for value in factor_hashes: - if len(value) != 64 or any( - character not in "0123456789abcdef" for character in value - ): - raise ValueError("factor_hashes must contain SHA-256 digests") + _strict_hex_digest(value, "factor_hashes item") object.__setattr__(self, "factor_hashes", factor_hashes) - reasons = tuple(str(value) for value in self.uncertified_reasons) + + reasons = tuple(self.uncertified_reasons) + for index, value in enumerate(reasons): + if not isinstance(value, str) or not value: + raise ValueError( + "uncertified_reasons[%d] must be a non-empty string" % index + ) object.__setattr__(self, "uncertified_reasons", reasons) - allowed = {kind.value for kind in FactorizationResultKind} + allowed = set(kind.value for kind in FactorizationResultKind) if self.result_kind not in allowed: - raise ValueError(f"unsupported result_kind {self.result_kind!r}") + raise ValueError("unsupported result_kind %r" % self.result_kind) if self.result_kind == FactorizationResultKind.FACTORS.value: if len(self.factor_hashes) != 2: raise ValueError("FACTORS evidence must contain exactly two factor hashes") if self.negative_result_certified or self.seq_prime_is_absolute: raise ValueError("factor evidence cannot claim negative certification") + if not self.uncertified_reasons: + raise ValueError("factor evidence must carry an uncertified reason") else: if self.factor_hashes: raise ValueError("SEQ-PRIME evidence must not contain factor hashes") + if not self.negative_result_certified and not self.uncertified_reasons: + raise ValueError( + "uncertified SEQ-PRIME evidence must carry explicit reasons" + ) if self.negative_result_certified != self.seq_prime_is_absolute: raise ValueError( @@ -310,10 +404,6 @@ def __post_init__(self) -> None: if self.negative_result_certified: certification_requirements = { "complete_domain": self.completeness_guaranteed, - "policy_version": ( - self.certification_policy_version - == NEGATIVE_CERTIFICATION_POLICY_VERSION - ), "search_exhausted": self.search_exhausted, "not_truncated": not self.truncation_occurred, "coverage_validated": self.coverage_record_validated, @@ -321,26 +411,32 @@ def __post_init__(self) -> None: "pruning_preserves_coverage": self.pruning_preserves_coverage, "no_uncertified_reasons": not self.uncertified_reasons, } - failed = [name for name, passed in certification_requirements.items() if not passed] + failed = [ + name + for name, passed in certification_requirements.items() + if not passed + ] if failed: raise ValueError( "certified negative evidence violates policy requirements: " + ", ".join(failed) ) - for name in ("supplied_catalogue_size", "effective_catalogue_size"): - if getattr(self, name) < 0: - raise ValueError(f"{name} must be non-negative") + if self.supplied_catalogue_size < 0 or self.effective_catalogue_size < 0: + raise ValueError("catalogue sizes must be non-negative") expected = _digest(self._evidence_fields()) if self.evidence_digest and self.evidence_digest != expected: - raise ValueError("evidence_digest does not match factorization evidence contents") + raise ValueError( + "evidence_digest does not match factorization evidence contents" + ) object.__setattr__(self, "evidence_digest", expected) - def _evidence_fields(self) -> dict[str, Any]: + def _evidence_fields(self) -> dict: return { "schema_id": self.schema_id, "schema_version": self.schema_version, + "producer_id": self.producer_id, "product_hash": self.product_hash, "product_domain_label": self.product_domain_label, "product_domain_statuses": list(self.product_domain_statuses), @@ -373,7 +469,7 @@ def _evidence_fields(self) -> dict[str, Any]: "uncertified_reasons": list(self.uncertified_reasons), } - def to_dict(self) -> dict[str, Any]: + def to_dict(self) -> dict: data = self._evidence_fields() data["evidence_digest"] = self.evidence_digest return data @@ -383,9 +479,10 @@ def to_json(self) -> str: @classmethod def from_dict(cls, data: Mapping[str, Any]) -> "UCNSFactorizationEvidence": - expected_fields = { + expected = ( "schema_id", "schema_version", + "producer_id", "product_hash", "product_domain_label", "product_domain_statuses", @@ -417,55 +514,83 @@ def from_dict(cls, data: Mapping[str, Any]) -> "UCNSFactorizationEvidence": "pruning_preserves_coverage", "uncertified_reasons", "evidence_digest", - } - unknown = set(data) - expected_fields - missing = expected_fields - set(data) - if unknown: - raise ValueError( - f"unknown factorization evidence fields: {sorted(unknown)!r}" - ) - if missing: - raise ValueError( - f"missing factorization evidence fields: {sorted(missing)!r}" - ) + ) + _exact_fields(data, expected, "factorization evidence") return cls( - schema_id=str(data["schema_id"]), - schema_version=str(data["schema_version"]), - product_hash=str(data["product_hash"]), - product_domain_label=str(data["product_domain_label"]), - product_domain_statuses=tuple(data["product_domain_statuses"]), - completeness_guaranteed=bool(data["completeness_guaranteed"]), - result_kind=str(data["result_kind"]), - factor_hashes=tuple(data["factor_hashes"]), - negative_result_certified=bool(data["negative_result_certified"]), - seq_prime_is_absolute=bool(data["seq_prime_is_absolute"]), - claim_scope=str(data["claim_scope"]), - note=str(data["note"]), - certification_policy_version=str(data["certification_policy_version"]), - search_exhausted=bool(data["search_exhausted"]), - truncation_occurred=bool(data["truncation_occurred"]), - catalogue_source=str(data["catalogue_source"]), - supplied_catalogue_size=int(data["supplied_catalogue_size"]), - supplied_catalogue_fingerprint=str(data["supplied_catalogue_fingerprint"]), - effective_catalogue_size=int(data["effective_catalogue_size"]), - effective_catalogue_fingerprint=str(data["effective_catalogue_fingerprint"]), - catalogue_coverage_status=str(data["catalogue_coverage_status"]), - catalogue_coverage_reason=str(data["catalogue_coverage_reason"]), - catalogue_coverage_rule_version=str(data["catalogue_coverage_rule_version"]), - required_catalogue_rule_version=str(data["required_catalogue_rule_version"]), - required_catalogue_fingerprint=str(data["required_catalogue_fingerprint"]), - coverage_record_validated=bool(data["coverage_record_validated"]), - coverage_bound_to_search_report=bool(data["coverage_bound_to_search_report"]), - pruning_applied=bool(data["pruning_applied"]), - pruning_rule=str(data["pruning_rule"]), - pruning_rule_version=str(data["pruning_rule_version"]), - pruning_preserves_coverage=bool(data["pruning_preserves_coverage"]), - uncertified_reasons=tuple(data["uncertified_reasons"]), - evidence_digest=str(data["evidence_digest"]), + schema_id=_strict_str(data, "schema_id"), + schema_version=_strict_str(data, "schema_version"), + producer_id=_strict_str(data, "producer_id"), + product_hash=_strict_str(data, "product_hash"), + product_domain_label=_strict_str(data, "product_domain_label"), + product_domain_statuses=_strict_string_tuple( + data, "product_domain_statuses", allow_empty=False + ), + completeness_guaranteed=_strict_bool( + data, "completeness_guaranteed" + ), + result_kind=_strict_str(data, "result_kind"), + factor_hashes=_strict_string_tuple(data, "factor_hashes"), + negative_result_certified=_strict_bool( + data, "negative_result_certified" + ), + seq_prime_is_absolute=_strict_bool(data, "seq_prime_is_absolute"), + claim_scope=_strict_str(data, "claim_scope"), + note=_strict_str(data, "note"), + certification_policy_version=_strict_str( + data, "certification_policy_version" + ), + search_exhausted=_strict_bool(data, "search_exhausted"), + truncation_occurred=_strict_bool(data, "truncation_occurred"), + catalogue_source=_strict_str( + data, "catalogue_source", allow_empty=True + ), + supplied_catalogue_size=_strict_int(data, "supplied_catalogue_size"), + supplied_catalogue_fingerprint=_strict_str( + data, "supplied_catalogue_fingerprint", allow_empty=True + ), + effective_catalogue_size=_strict_int(data, "effective_catalogue_size"), + effective_catalogue_fingerprint=_strict_str( + data, "effective_catalogue_fingerprint", allow_empty=True + ), + catalogue_coverage_status=_strict_str( + data, "catalogue_coverage_status", allow_empty=True + ), + catalogue_coverage_reason=_strict_str( + data, "catalogue_coverage_reason", allow_empty=True + ), + catalogue_coverage_rule_version=_strict_str( + data, "catalogue_coverage_rule_version", allow_empty=True + ), + required_catalogue_rule_version=_strict_str( + data, "required_catalogue_rule_version", allow_empty=True + ), + required_catalogue_fingerprint=_strict_str( + data, "required_catalogue_fingerprint", allow_empty=True + ), + coverage_record_validated=_strict_bool( + data, "coverage_record_validated" + ), + coverage_bound_to_search_report=_strict_bool( + data, "coverage_bound_to_search_report" + ), + pruning_applied=_strict_bool(data, "pruning_applied"), + pruning_rule=_strict_str(data, "pruning_rule", allow_empty=True), + pruning_rule_version=_strict_str( + data, "pruning_rule_version", allow_empty=True + ), + pruning_preserves_coverage=_strict_bool( + data, "pruning_preserves_coverage" + ), + uncertified_reasons=_strict_string_tuple( + data, "uncertified_reasons" + ), + evidence_digest=_strict_str(data, "evidence_digest"), ) @classmethod def from_json(cls, value: str) -> "UCNSFactorizationEvidence": + if not isinstance(value, str): + raise ValueError("factorization evidence JSON must be a string") decoded = json.loads(value) if not isinstance(decoded, dict): raise ValueError("factorization evidence JSON must decode to an object") @@ -552,8 +677,10 @@ def factorization_evidence( __all__ = [ "BRIDGE_RECORD_SCHEMA_ID", "BRIDGE_RECORD_SCHEMA_VERSION", + "BRIDGE_RECORD_PRODUCER_ID", "FACTORIZATION_EVIDENCE_SCHEMA_ID", "FACTORIZATION_EVIDENCE_SCHEMA_VERSION", + "FACTORIZATION_EVIDENCE_PRODUCER_ID", "UCNSBridgeRecord", "UCNSFactorizationEvidence", "bridge_record", From d493eda1620f4faea55250606872cc11a57af66e Mon Sep 17 00:00:00 2001 From: Erin Spencer Date: Sun, 12 Jul 2026 09:57:49 -0700 Subject: [PATCH 5/9] test(evidence): reject coerced serialized scalar types --- tests/test_evidence_envelope.py | 29 ++++++++++++++++++++++++++--- 1 file changed, 26 insertions(+), 3 deletions(-) diff --git a/tests/test_evidence_envelope.py b/tests/test_evidence_envelope.py index 21030814..e5267043 100644 --- a/tests/test_evidence_envelope.py +++ b/tests/test_evidence_envelope.py @@ -54,7 +54,7 @@ def test_unit_bridge_record_is_typed_and_not_a_prime_claim() -> None: assert record.seq_prime_claim_scope == "not-prime-unit-domain" -def test_bridge_record_rejects_hash_tamper_and_unknown_fields() -> None: +def test_bridge_record_rejects_hash_tamper_unknown_fields_and_type_coercion() -> None: record = bridge_record(S2) tampered = record.to_dict() tampered["object_hash"] = "0" * 64 @@ -67,6 +67,16 @@ def test_bridge_record_rejects_hash_tamper_and_unknown_fields() -> None: with pytest.raises(ValueError, match="unknown bridge record fields"): UCNSBridgeRecord.from_dict(unknown) + coerced = record.to_dict() + coerced["is_unit"] = "false" + with pytest.raises(ValueError, match="is_unit must be a boolean"): + UCNSBridgeRecord.from_dict(coerced) + + coerced = record.to_dict() + coerced["depth"] = "1" + with pytest.raises(ValueError, match="depth must be an integer"): + UCNSBridgeRecord.from_dict(coerced) + def test_flat_prime_serializes_certified_negative_evidence() -> None: evidence = factorization_evidence(S2) @@ -128,7 +138,7 @@ def test_certified_record_with_broken_policy_invariant_fails_closed() -> None: with pytest.raises(ValueError, match="search_exhausted"): replace(evidence, search_exhausted=False, evidence_digest="") - with pytest.raises(ValueError, match="policy_version"): + with pytest.raises(ValueError, match="certification_policy_version"): replace( evidence, certification_policy_version="invented-policy", @@ -136,7 +146,7 @@ def test_certified_record_with_broken_policy_invariant_fails_closed() -> None: ) -def test_factorization_evidence_rejects_digest_tamper_and_unknown_fields() -> None: +def test_factorization_evidence_rejects_tamper_unknown_fields_and_type_coercion() -> None: evidence = factorization_evidence(S2) tampered = evidence.to_dict() tampered["claim_scope"] = "invented" @@ -147,3 +157,16 @@ def test_factorization_evidence_rejects_digest_tamper_and_unknown_fields() -> No unknown["invented"] = True with pytest.raises(ValueError, match="unknown factorization evidence fields"): UCNSFactorizationEvidence.from_dict(unknown) + + coerced = evidence.to_dict() + coerced["negative_result_certified"] = "true" + with pytest.raises( + ValueError, + match="negative_result_certified must be a boolean", + ): + UCNSFactorizationEvidence.from_dict(coerced) + + coerced = evidence.to_dict() + coerced["supplied_catalogue_size"] = "1" + with pytest.raises(ValueError, match="supplied_catalogue_size must be an integer"): + UCNSFactorizationEvidence.from_dict(coerced) From c67751f0aecd353bc251d9c538898ce8f8cc5c9c Mon Sep 17 00:00:00 2001 From: Erin Spencer Date: Sun, 12 Jul 2026 09:58:24 -0700 Subject: [PATCH 6/9] docs(evidence): document bridge and certification records --- docs/evidence-envelopes.md | 77 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 77 insertions(+) create mode 100644 docs/evidence-envelopes.md diff --git a/docs/evidence-envelopes.md b/docs/evidence-envelopes.md new file mode 100644 index 00000000..66b591fe --- /dev/null +++ b/docs/evidence-envelopes.md @@ -0,0 +1,77 @@ +# UCNS serialized evidence envelopes + +## Bridge record + +Create a versioned canonical identity record for any UCNS object or the unit sentinel: + +```python +import ucns + +record = ucns.bridge_record(ucns.S2) +serialized = record.to_json() +restored = ucns.UCNSBridgeRecord.from_json(serialized) +assert restored == record +``` + +`UCNSBridgeRecord` binds: + +```text +canonical UCNS JSON +stable SHA-256 object hash +canonical serialization version +typed domain statuses +completeness prerequisite +SEQ-PRIME claim scope +depth, n_min, length, and unit/frontier flags +producer id and evidence digest +``` + +The constructor verifies that `object_hash` is the SHA-256 digest of the exact canonical JSON bytes. + +## Factorization evidence + +Create a serialized result only through the authoritative UCNS policy path: + +```python +import ucns + +evidence = ucns.factorization_evidence(ucns.S2) +assert evidence.negative_result_certified +``` + +`factorization_evidence()` accepts a product and optional catalogue. It does **not** accept a caller-built `FactorizationResult` or a certification boolean. It runs `ucns.factorization_result()` and serializes: + +```text +product stable hash and domain status +result kind and factor stable hashes +claim scope and note +negative-certification policy version +search exhaustion and truncation +catalogue source, sizes, and fingerprints +coverage status, validation, and report binding +pruning rule/version and coverage-preservation status +explicit uncertified reasons +producer id and evidence digest +``` + +A certified negative record fails construction unless every existing UCNS certification condition remains true. + +## Deserialization behavior + +Both records: + +- reject unknown and missing fields; +- require exact boolean, integer, string, and sequence types; +- reject unsupported schema versions and producer ids; +- recompute and verify their evidence digest; +- emit deterministic canonical JSON. + +## Trust boundary + +Evidence digests are tamper-evident content identities. They are not signatures and do not authenticate who transmitted a record. + +Domain statuses and negative certification are UCNS evidence. They do not automatically validate a consumer's empirical measurements, ontology, or external factual claims. + +## hmmm + +A future transport may add signed producer attestations. No signature or external trust system is claimed by version 1 of these envelopes. From 6d85ed96495d5c41615800930ba736af7da16493 Mon Sep 17 00:00:00 2001 From: Erin Spencer Date: Sun, 12 Jul 2026 10:03:12 -0700 Subject: [PATCH 7/9] ci(diagnostic): capture targeted evidence test output --- .github/workflows/evidence-diagnostic.yml | 33 +++++++++++++++++++++++ 1 file changed, 33 insertions(+) create mode 100644 .github/workflows/evidence-diagnostic.yml diff --git a/.github/workflows/evidence-diagnostic.yml b/.github/workflows/evidence-diagnostic.yml new file mode 100644 index 00000000..ccf92ec7 --- /dev/null +++ b/.github/workflows/evidence-diagnostic.yml @@ -0,0 +1,33 @@ +name: evidence diagnostic + +on: + pull_request: + +permissions: + contents: read + +jobs: + evidence-diagnostic: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + - name: Install package and test tools + run: | + python -m pip install --upgrade pip + python -m pip install -e .[dev] + - name: Capture targeted test output + run: | + set +e + python -m pytest tests/test_evidence_envelope.py -vv --tb=short > evidence-test.log 2>&1 + echo "$?" > evidence-status.txt + exit 0 + - name: Upload diagnostic output + uses: actions/upload-artifact@v4 + with: + name: evidence-diagnostic + path: | + evidence-test.log + evidence-status.txt From e484a2dddd1dd4019a7a76207ff1c20355c396e6 Mon Sep 17 00:00:00 2001 From: Erin Spencer Date: Sun, 12 Jul 2026 10:04:22 -0700 Subject: [PATCH 8/9] test(evidence): isolate bridge hash tamper assertion --- tests/test_evidence_envelope.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/test_evidence_envelope.py b/tests/test_evidence_envelope.py index e5267043..8f5dfac5 100644 --- a/tests/test_evidence_envelope.py +++ b/tests/test_evidence_envelope.py @@ -58,7 +58,6 @@ def test_bridge_record_rejects_hash_tamper_unknown_fields_and_type_coercion() -> record = bridge_record(S2) tampered = record.to_dict() tampered["object_hash"] = "0" * 64 - tampered["evidence_digest"] = "" with pytest.raises(ValueError, match="object_hash does not match"): UCNSBridgeRecord.from_dict(tampered) From f42d6183280c440b773f80350698729fd33fe6c6 Mon Sep 17 00:00:00 2001 From: Erin Spencer Date: Sun, 12 Jul 2026 10:05:32 -0700 Subject: [PATCH 9/9] ci(diagnostic): remove temporary evidence workflow --- .github/workflows/evidence-diagnostic.yml | 33 ----------------------- 1 file changed, 33 deletions(-) delete mode 100644 .github/workflows/evidence-diagnostic.yml diff --git a/.github/workflows/evidence-diagnostic.yml b/.github/workflows/evidence-diagnostic.yml deleted file mode 100644 index ccf92ec7..00000000 --- a/.github/workflows/evidence-diagnostic.yml +++ /dev/null @@ -1,33 +0,0 @@ -name: evidence diagnostic - -on: - pull_request: - -permissions: - contents: read - -jobs: - evidence-diagnostic: - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - - uses: actions/setup-python@v5 - with: - python-version: "3.11" - - name: Install package and test tools - run: | - python -m pip install --upgrade pip - python -m pip install -e .[dev] - - name: Capture targeted test output - run: | - set +e - python -m pytest tests/test_evidence_envelope.py -vv --tb=short > evidence-test.log 2>&1 - echo "$?" > evidence-status.txt - exit 0 - - name: Upload diagnostic output - uses: actions/upload-artifact@v4 - with: - name: evidence-diagnostic - path: | - evidence-test.log - evidence-status.txt