diff --git a/proof/region/v1/semantic/__init__.py b/proof/region/v1/semantic/__init__.py index 22317a1c..06f951e5 100644 --- a/proof/region/v1/semantic/__init__.py +++ b/proof/region/v1/semantic/__init__.py @@ -10,11 +10,12 @@ SemanticVerificationReceiptV1, SemanticVerificationRejectedV1, ) -from .verifier import verify_transcript +from .verifier import bind_transcript_v1, verify_transcript __all__ = [ "SemanticVerificationReasonV1", "SemanticVerificationReceiptV1", "SemanticVerificationRejectedV1", + "bind_transcript_v1", "verify_transcript", ] diff --git a/proof/region/v1/semantic/verifier.py b/proof/region/v1/semantic/verifier.py index b0abc30e..e28a6a98 100644 --- a/proof/region/v1/semantic/verifier.py +++ b/proof/region/v1/semantic/verifier.py @@ -33,12 +33,18 @@ def _foreign_binding(detail: str) -> SemanticVerificationRejectedV1: return _reject(SemanticVerificationReasonV1.FOREIGN_BINDING, 0, detail) -def verify_transcript( +def bind_transcript_v1( job: protocol.ProofJobV1, comparator: protocol.ContentResolvedComparatorManifestV2, transcript: protocol.DecisionTranscriptV1, run: protocol.RunClaimV1, -) -> VerificationResultV1: +) -> SemanticVerificationRejectedV1 | None: + """The single binding surface between one replay and one transcript. + + Returns the typed rejection when any input is foreign or any coordinate + drifts, and `None` once the four canonical objects bind one verification. + """ + if ( type(job) is not protocol.ProofJobV1 or type(comparator) is not protocol.ContentResolvedComparatorManifestV2 @@ -65,6 +71,19 @@ def verify_transcript( return _foreign_binding("run claim binds a foreign transcript") if transcript.point_count != job.domain.point_count: return _foreign_binding("transcript point count drifts from the bound domain") + return None + + +def verify_transcript( + job: protocol.ProofJobV1, + comparator: protocol.ContentResolvedComparatorManifestV2, + transcript: protocol.DecisionTranscriptV1, + run: protocol.RunClaimV1, +) -> VerificationResultV1: + binding = bind_transcript_v1(job, comparator, transcript, run) + if binding is not None: + return binding + # Binary, invocation and platform are declared execution coordinates. # Their causality belongs to the source-bound controller's receipt; the # semantic verifier binds the run through job, comparator and transcript diff --git a/proof/region/v1/tests/test_verification_assembly.py b/proof/region/v1/tests/test_verification_assembly.py new file mode 100644 index 00000000..fd95aad4 --- /dev/null +++ b/proof/region/v1/tests/test_verification_assembly.py @@ -0,0 +1,361 @@ +#!/usr/bin/env python3 +"""Hostile contract for laned semantic verification. + +A full-domain transcript can only be verified by independent replay, and one +sequential replay of 2^24 points never fits a single verification process. +The laned path therefore splits the domain into packing-aligned windows, +replays each window independently in the exhausted ordinal-prefix grant +regime, and seals exactly one `SemanticVerificationReceiptV1` — the same +receipt the monolithic verifier seals — only when the lane-assembled replay +is byte-identical to the verified transcript. No lane cover, no receipt. +""" + +from __future__ import annotations + +import hashlib +import sys +import unittest +from pathlib import Path + +PROOF = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(PROOF)) + +import corpus # noqa: E402 +import corpus_assembly # noqa: E402 +import corpus_lane # noqa: E402 +import region_proof_protocol as protocol # noqa: E402 +import verification_assembly # noqa: E402 +from semantic.verifier import verify_transcript # noqa: E402 + +SHARD_POINTS = 16 +WINDOW_POINTS = 32 +DOMAIN_ORDINALS = tuple(range(2 * WINDOW_POINTS)) + + +def digest(label: int) -> bytes: + return hashlib.sha256(f"verification-assembly-{label}".encode("ascii")).digest() + + +def _exhausted_policy(policy: protocol.ProofPolicyV1) -> protocol.ProofPolicyV1: + return protocol.ProofPolicyV1( + policy.equality_release, + tuple( + protocol.ComparatorBudgetV1( + budget.kind, budget.precision_ladder, budget.per_point_work, 0 + ) + for budget in policy.comparators + ), + ) + + +def test_job() -> protocol.ProofJobV1: + base = protocol.ProofJobV1.parse(corpus_lane.FIXTURE_JOB_V1.read_bytes()) + return protocol.ProofJobV1( + base.definition, + base.formula_spec, + protocol.ReducedDomainManifestV1.from_ordinals(DOMAIN_ORDINALS), + _exhausted_policy(base.policy), + ) + + +def honest_transcript( + job: protocol.ProofJobV1, + comparator: protocol.ContentResolvedComparatorManifestV2, +) -> protocol.DecisionTranscriptV1: + runner = corpus.ShardCorpusRunnerV1(job, comparator) + shards = [runner.run_shard(start, end) for start, end in corpus.shard_plan_v1( + job.domain, SHARD_POINTS + )] + assembled = corpus.assemble_transcript_from_shards_v1( + job, comparator, shards, runner.accounting_digest + ) + if type(assembled) is not protocol.DecisionTranscriptV1: + raise AssertionError(f"honest transcript did not assemble: {assembled!r}") + return assembled + + +def run_claim(transcript: protocol.DecisionTranscriptV1) -> protocol.RunClaimV1: + return protocol.RunClaimV1.for_transcript( + test_job(), comparator(), transcript, digest(1), digest(2), digest(3) + ) + + +def comparator() -> protocol.ContentResolvedComparatorManifestV2: + return corpus_lane.lane_comparator_v1() + + +def window_lanes( + job: protocol.ProofJobV1, + comparator_manifest: protocol.ContentResolvedComparatorManifestV2, +) -> tuple: + return tuple( + corpus.run_window_lane_v1( + job, comparator_manifest, start, WINDOW_POINTS, SHARD_POINTS + ) + for start in range(0, job.domain.point_count, WINDOW_POINTS) + ) + + +def lanes( + job: protocol.ProofJobV1, + comparator_manifest: protocol.ContentResolvedComparatorManifestV2, +) -> tuple: + return tuple( + corpus_assembly.AdmittedLaneV1( + lane.window_start, + lane.window_points, + lane.shards, + lane.accounting_records, + ) + for lane in window_lanes(job, comparator_manifest) + ) + + +class LanedSemanticVerificationTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.job = test_job() + cls.comparator = comparator() + cls.transcript = honest_transcript(cls.job, cls.comparator) + cls.run_claim = run_claim(cls.transcript) + cls.monolithic = verify_transcript( + cls.job, cls.comparator, cls.transcript, cls.run_claim + ) + + def test_monolithic_baseline_seals_a_receipt(self) -> None: + from semantic.receipt import SemanticVerificationReceiptV1 + + self.assertIs(type(self.monolithic), SemanticVerificationReceiptV1) + + def test_laned_verification_seals_the_same_receipt(self) -> None: + result = verification_assembly.assemble_semantic_verification_v1( + self.job, self.comparator, self.transcript, self.run_claim, lanes( + self.job, self.comparator + ) + ) + self.assertIs(type(result), type(self.monolithic)) + self.assertEqual(result.identity, self.monolithic.identity) + self.assertTrue( + result.binds(self.job, self.comparator, self.run_claim, self.transcript) + ) + + def test_laned_verification_rejects_foreign_inputs(self) -> None: + from semantic.receipt import ( + SemanticVerificationReasonV1, + SemanticVerificationRejectedV1, + ) + + good_lanes = lanes(self.job, self.comparator) + + def hostile_cover(error_type: type[Exception]) -> object: + def _cover(): + yield good_lanes[0] + raise error_type("hostile lane iterator") + + return _cover() + + cases = ( + (object(), self.comparator, self.transcript, self.run_claim, good_lanes), + (self.job, object(), self.transcript, self.run_claim, good_lanes), + (self.job, self.comparator, object(), self.run_claim, good_lanes), + (self.job, self.comparator, self.transcript, object(), good_lanes), + (self.job, self.comparator, self.transcript, self.run_claim, ()), + (self.job, self.comparator, self.transcript, self.run_claim, (object(),)), + ( + self.job, + self.comparator, + self.transcript, + self.run_claim, + hostile_cover(ValueError), + ), + ( + self.job, + self.comparator, + self.transcript, + self.run_claim, + hostile_cover(RuntimeError), + ), + ) + for case in cases: + result = verification_assembly.assemble_semantic_verification_v1(*case) + self.assertIs( + type(result), + SemanticVerificationRejectedV1, + f"foreign input was not rejected: {case!r}", + ) + self.assertEqual(result.reason, SemanticVerificationReasonV1.INVALID_INPUT) + + def test_laned_verification_rejects_foreign_bindings(self) -> None: + from semantic.receipt import ( + SemanticVerificationReasonV1, + SemanticVerificationRejectedV1, + ) + + good_lanes = lanes(self.job, self.comparator) + foreign_transcript = honest_transcript( + protocol.ProofJobV1( + self.job.definition, + self.job.formula_spec, + protocol.ReducedDomainManifestV1.from_ordinals(DOMAIN_ORDINALS[:-4]), + self.job.policy, + ), + self.comparator, + ) + # transcript bound to a foreign job + result = verification_assembly.assemble_semantic_verification_v1( + self.job, self.comparator, foreign_transcript, self.run_claim, good_lanes + ) + self.assertIs(type(result), SemanticVerificationRejectedV1) + self.assertEqual(result.reason, SemanticVerificationReasonV1.FOREIGN_BINDING) + # run claim forged against a foreign transcript identity + foreign_run = protocol.RunClaimV1( + self.job.identity, + self.comparator.identity, + digest(1), + digest(2), + digest(3), + foreign_transcript.identity, + ) + result = verification_assembly.assemble_semantic_verification_v1( + self.job, self.comparator, self.transcript, foreign_run, good_lanes + ) + self.assertIs(type(result), SemanticVerificationRejectedV1) + self.assertEqual(result.reason, SemanticVerificationReasonV1.FOREIGN_BINDING) + + def test_incomplete_lane_cover_never_seals(self) -> None: + from semantic.receipt import SemanticVerificationRejectedV1 + + all_lanes = lanes(self.job, self.comparator) + for drop in range(len(all_lanes)): + partial = all_lanes[:drop] + all_lanes[drop + 1 :] + result = verification_assembly.assemble_semantic_verification_v1( + self.job, self.comparator, self.transcript, self.run_claim, partial + ) + self.assertIs( + type(result), SemanticVerificationRejectedV1, + f"dropping lane {drop} still sealed a receipt", + ) + + def test_reordered_or_overlapping_lanes_never_seal(self) -> None: + from semantic.receipt import SemanticVerificationRejectedV1 + + all_lanes = lanes(self.job, self.comparator) + reversed_cover = tuple(reversed(all_lanes)) + result = verification_assembly.assemble_semantic_verification_v1( + self.job, self.comparator, self.transcript, self.run_claim, reversed_cover + ) + self.assertIs(type(result), SemanticVerificationRejectedV1) + duplicated = all_lanes + all_lanes[-1:] + result = verification_assembly.assemble_semantic_verification_v1( + self.job, self.comparator, self.transcript, self.run_claim, duplicated + ) + self.assertIs(type(result), SemanticVerificationRejectedV1) + + def test_diverging_replay_never_seals(self) -> None: + from semantic.receipt import ( + SemanticVerificationReasonV1, + SemanticVerificationRejectedV1, + ) + + # A transcript whose committed accounting digest does not replay from + # the lane records is a foreign transcript for this evidence. + mutated = bytearray(self.transcript.accounting_digest) + mutated[0] ^= 0xFF + foreign = protocol.DecisionTranscriptV1( + self.transcript.job_identity, + self.transcript.domain_identity, + self.transcript.comparator_identity, + self.transcript.point_count, + self.transcript.decision_bits, + self.transcript.counters, + self.transcript.exact_equality_count, + bytes(mutated), + self.transcript.witness_store, + ) + foreign_run = protocol.RunClaimV1.for_transcript( + self.job, self.comparator, foreign, digest(1), digest(2), digest(3) + ) + result = verification_assembly.assemble_semantic_verification_v1( + self.job, + self.comparator, + foreign, + foreign_run, + lanes(self.job, self.comparator), + ) + self.assertIs(type(result), SemanticVerificationRejectedV1) + self.assertEqual(result.reason, SemanticVerificationReasonV1.DECISION_MISMATCH) + + def test_lanes_replayed_over_a_shifted_window_never_seal(self) -> None: + from semantic.receipt import SemanticVerificationRejectedV1 + + # Lanes whose windows are shifted by one alignment unit replay real + # points but never cover the verified domain, so no receipt seals. + shifted = tuple( + corpus.run_window_lane_v1( + self.job, self.comparator, start, WINDOW_POINTS, SHARD_POINTS + ) + for start in range(4, 4 + self.job.domain.point_count, WINDOW_POINTS) + ) + result = verification_assembly.assemble_semantic_verification_v1( + self.job, self.comparator, self.transcript, self.run_claim, shifted + ) + self.assertIs(type(result), SemanticVerificationRejectedV1) + + + def test_wire_lane_round_trip_seals_the_same_receipt(self) -> None: + import tempfile + + with tempfile.TemporaryDirectory() as root: + from pathlib import Path + + directories = [] + for index, lane in enumerate(window_lanes(self.job, self.comparator)): + out = Path(root) / f"lane-{index:05d}" + corpus_lane.write_lane_artifacts_v1( + lane, self.job, self.comparator, SHARD_POINTS, out + ) + directories.append(out) + loaded = [ + corpus_assembly.load_lane_v1(directory, self.job, self.comparator) + for directory in directories + ] + for lane in loaded: + self.assertIs(type(lane), corpus_assembly.AdmittedLaneV1) + result = verification_assembly.assemble_semantic_verification_v1( + self.job, self.comparator, self.transcript, self.run_claim, tuple(loaded) + ) + self.assertIs(type(result), type(self.monolithic)) + self.assertEqual(result.identity, self.monolithic.identity) + + def test_wire_lanes_bound_to_a_foreign_comparator_never_admit(self) -> None: + import hashlib as _hashlib + import tempfile + + foreign = protocol.ContentResolvedComparatorManifestV2.admit( + protocol.ComparatorManifestV2( + protocol.ComparatorKindV1.ARB, + *( + _hashlib.sha256(f"foreign-verification-{i}".encode()).digest() + for i in range(10) + ), + ), + { + _hashlib.sha256(f"foreign-verification-{i}".encode()).digest(): + f"foreign-verification-{i}".encode() + for i in range(10) + }.get, + ) + with tempfile.TemporaryDirectory() as root: + from pathlib import Path + + lane = window_lanes(self.job, self.comparator)[0] + out = Path(root) / "lane-00000" + corpus_lane.write_lane_artifacts_v1( + lane, self.job, self.comparator, SHARD_POINTS, out + ) + loaded = corpus_assembly.load_lane_v1(out, self.job, foreign) + self.assertIsNot(type(loaded), corpus_assembly.AdmittedLaneV1) + + +if __name__ == "__main__": + unittest.main() diff --git a/proof/region/v1/verification_assembly.py b/proof/region/v1/verification_assembly.py new file mode 100644 index 00000000..f237929b --- /dev/null +++ b/proof/region/v1/verification_assembly.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python3 +"""Laned assembly of the independent semantic verification receipt. + +One sequential semantic replay of the exact full 2^24 point domain never +fits a single verification process, so the replay runs as packing-aligned +window lanes: each lane independently replays its window of the job's domain +in the exhausted ordinal-prefix grant regime (the exact lane machinery the +full-domain RUN uses). Assembly admits the lanes through the same lane +admission as the corpus RUN, reassembles the monolithic transcript from the +lane fragments, and seals the one `SemanticVerificationReceiptV1` — the same +receipt the monolithic verifier seals — only when the reassembled +independent replay is byte-identical to the verified transcript. Any gap, +overlap, reorder, drift or identity mismatch returns a typed rejection and +nothing ever seals. +""" + +from __future__ import annotations + +import corpus +import corpus_assembly +import region_proof_protocol as protocol +from semantic import verifier as semantic_verifier +from semantic.receipt import ( + SemanticVerificationReasonV1, + SemanticVerificationReceiptV1, + SemanticVerificationRejectedV1, +) + +VerificationAssemblyResultV1 = ( + SemanticVerificationReceiptV1 | SemanticVerificationRejectedV1 +) + + +def _reject( + reason: SemanticVerificationReasonV1, detail: str +) -> SemanticVerificationRejectedV1: + return SemanticVerificationRejectedV1(reason, 0, detail) + + +def assemble_semantic_verification_v1( + job: protocol.ProofJobV1, + comparator: protocol.ContentResolvedComparatorManifestV2, + transcript: protocol.DecisionTranscriptV1, + run: protocol.RunClaimV1, + lanes: object, +) -> VerificationAssemblyResultV1: + """Seal the semantic verification receipt from a complete lane cover. + + The lanes are the same admitted wire lanes the corpus RUN assembly uses: + independently replayed windows of the job's domain. The receipt seals + exactly when their cover is exact and their reassembled replay binds the + verified transcript byte for byte. + """ + + binding = semantic_verifier.bind_transcript_v1(job, comparator, transcript, run) + if binding is not None: + return binding + try: + lane_tuple = tuple(lanes) # type: ignore[arg-type] + except Exception: + # The lane cover is a hostile boundary: any iterator failure — + # not just a non-iterable — must land as the typed rejection. + return _reject( + SemanticVerificationReasonV1.INVALID_INPUT, + "laned verification requires an iterable lane cover", + ) + if not lane_tuple or any( + type(lane) is not corpus_assembly.AdmittedLaneV1 for lane in lane_tuple + ): + return _reject( + SemanticVerificationReasonV1.INVALID_INPUT, + "laned verification requires canonical admitted lane evidence", + ) + assembled = corpus_assembly.assemble_lanes_v1(job, comparator, lane_tuple) + if type(assembled) is not protocol.DecisionTranscriptV1: + reason = ( + SemanticVerificationReasonV1.INVALID_INPUT + if assembled.reason is corpus.ShardCorpusReasonV1.FOREIGN_INPUT + else SemanticVerificationReasonV1.FOREIGN_BINDING + ) + return _reject(reason, f"the lane cover does not assemble: {assembled.detail}") + if assembled.identity != transcript.identity: + return _reject( + SemanticVerificationReasonV1.DECISION_MISMATCH, + "the independent lane replay does not bind the verified transcript", + ) + return SemanticVerificationReceiptV1._seal(job, comparator, run, transcript)