What provenance chain breaks must be detected in multi-hop data transfers?

**Date of Record:** 2026 October 26 **Scope:** Applies when data moves through multiple agents, systems, or storage layers (multi-hop), and the receiving agent must verify that the provenance chain has no breaks (unattributed transfers, missing signatures, or orphaned hops). Does not apply to single-hop transfers or data that originates and terminates within a single trust boundary.

Core Distinction

The separation between **endpoint provenance** (the data is signed by the original producer and the final consumer) and **continuous chain provenance** (every intermediate hop has signed the data, proving it was legitimately transferred without unauthorized modifications or unauthorized intermediaries). Relying on endpoints creates the **Middle-Hop Blindness**: an attacker inserts themselves into the transfer chain, sees the data, and forwards it without modifying it. The endpoint signatures still validate, but the data has been exposed to an unauthorized party.

What Was Observed

In multi-hop data pipelines and agent relays, failing to detect provenance chain breaks produces three primary failure modes:

  1. **Relay Injection:** An attacker intercepts a data transfer, forwards it to their own system, and then to the intended receiver. Endpoint signatures match, but the data passed through an unauthorized hop.
  2. **Signature Stripping:** An intermediate hop strips the previous hop's signature to avoid detection, resigning the data as if it originated from itself. Without chain continuity, the receiver cannot detect the provenance break.
  3. **Orphaned Hops:** An intermediate hop crashes and restarts without persisting its signature. The chain has a gap, but the endpoints are intact.

Recorded Elements

  1. **`hop_signature_chain`:** An ordered list of cryptographic signatures from every hop in the transfer chain, each signature binding the data hash to the hop's identity and the previous hop's signature.
  2. **`expected_hop_count`:** The exact number of hops the data is expected to traverse, allowing the receiver to detect missing or extra hops.
  3. **`hop_authorization_manifest`:** The explicit list of authorized hops and the cryptographic proof that each was authorized to handle the data.
  4. **`chain_continuity_proof`:** The cryptographic proof that each hop's signature chains to the previous, with no gaps or unlinked signatures.

Condition of Validity

* **Applies when:** Data flows through multiple agents or systems, each hop could expose the data to unauthorized parties, and the receiver must verify continuous custody. * **Does not apply when:** The transfer is single-hop from producer to consumer, or all hops are within a single trust boundary (e.g., internal service mesh with mTLS).

Decision Boundary

  1. **Does the data arrive with a `hop_signature_chain`?** If no, the provenance is undocumented. Reject or treat as unverified.
  2. **Does the chain length match `expected_hop_count`?** If no, there is a missing or extra hop. Reject.
  3. **Does each hop's signature chain to the previous hop's signature?** If no, there is a provenance break. Reject.
  4. **Conversely, when can a receiver accept data without a full chain?** Only when all hops are within a single cryptographically-verified trust boundary (e.g., a sealed enclave), or when the data is explicitly marked as `LOW_CONFIDENTIALITY` and exposure to additional hops is acceptable.

Operational Effect

  1. **Middle-Hop Detection:** Prevents unauthorized intermediaries from handling data by requiring each hop to sign the chain.
  2. **Chain Continuity Enforcement:** Detects signature stripping or orphaned hops, ensuring the chain is complete.
  3. **Authorization Verification:** Verifies that each hop was authorized to handle the data, closing a common multi-hop vulnerability.
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
        )