# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-provenance-chain-breaks-must-be-detected-in-multi-hop-data-transfers.md # Remnant Warrant: https://corpusfetch.com/c/what-provenance-chain-breaks-must-be-detected-in-multi-hop-data-transfers.warrant.json # Markdown SHA-256: 83e9a5b2bca35b8eb43a9cbc1c77841bd62d532c365d7b3bfd334ccdeefa6697 # Code AST SHA-256: e235e9cf6e440f036c748e98cbe1c23e5034f8713fe0ed5a0fa6f5468206d3c4 # Verified Classes: ChainVerdict, HopSignature, MultiHopState, ChainPolicy, ChainManifest, MultiHopProvenanceGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class ChainVerdict(Enum): CHAIN_VERIFIED = auto() CHAIN_BROKEN = auto() MISSING_HOP = auto() UNAUTHORIZED_HOP = auto() @dataclass(frozen=True) class HopSignature: hop_id: str data_hash: str previous_signature_hash: str hop_signature: str @dataclass(frozen=True) class MultiHopState: chain: tuple[HopSignature, ...] expected_hop_count: int authorized_hops: frozenset[str] @dataclass(frozen=True) class ChainPolicy: require_full_chain: bool @dataclass(frozen=True) class ChainManifest: verdict: ChainVerdict directive: str manifest_digest: str class MultiHopProvenanceGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_chain( cls, state: MultiHopState, policy: ChainPolicy ) -> ChainManifest: # Gate 1: Chain presence if not state.chain: return ChainManifest( verdict=ChainVerdict.CHAIN_BROKEN, directive="HALT: No hop signature chain provided.", manifest_digest=cls._hash("NO_CHAIN") ) # Gate 2: Chain length if len(state.chain) != state.expected_hop_count: return ChainManifest( verdict=ChainVerdict.MISSING_HOP, directive=f"HALT: Chain length {len(state.chain)} != expected {state.expected_hop_count}.", manifest_digest=cls._hash(f"LENGTH:{len(state.chain)}") ) # Gate 3: Continuity check for i, hop in enumerate(state.chain): if i == 0: continue previous = state.chain[i - 1] expected_prev_hash = cls._hash(previous.hop_signature) if hop.previous_signature_hash != expected_prev_hash: return ChainManifest( verdict=ChainVerdict.CHAIN_BROKEN, directive=f"HALT: Hop {hop.hop_id} breaks chain. Previous hash mismatch.", manifest_digest=cls._hash(f"BROKEN:{hop.hop_id}") ) # Gate 4: Authorization for hop in state.chain: if hop.hop_id not in state.authorized_hops: return ChainManifest( verdict=ChainVerdict.UNAUTHORIZED_HOP, directive=f"HALT: Hop {hop.hop_id} not in authorized list.", manifest_digest=cls._hash(f"UNAUTHORIZED:{hop.hop_id}") ) # Gate 5: Verified digest = cls._hash(json.dumps({ "chain_length": str(len(state.chain)), "first_hop": state.chain[0].hop_id, "last_hop": state.chain[-1].hop_id }, sort_keys=True)) return ChainManifest( verdict=ChainVerdict.CHAIN_VERIFIED, directive="PROCEED: Multi-hop provenance chain is continuous and authorized.", manifest_digest=digest )