Which cryptographic watermarks must be embedded in event streams to prove end-to-end processing integrity?

**Date of Record:** 2026 October 26 **Scope:** Applies when an agentic session consumes and mutates events in a streaming pipeline (e.g., Kafka, RabbitMQ, Kinesis), and must persist cryptographic proof that the payloads were processed without silent corruption or injection. Does not apply to purely stateless logging, or streams where the producer and consumer are in the same trusted execution environment.

Core Distinction

The separation between **offset tracking** (knowing that a message was read from the stream) and **processing integrity** (proving that the message was processed correctly and not silently dropped or mutated). Offset tracking creates the **Silent Drop Illusion**: an instance claims to have processed offset `1000`, but a middle-ware bug or a malicious actor dropped record `999`, or altered its payload. End-to-end integrity requires each processor to embed a cryptographic watermark (like an HMAC or Merkle proof) into the output stream, chaining the input to the output.

What Was Observed

In distributed event-driven architectures, failing to embed cryptographic watermarks produces three primary failure modes:

  1. **Silent Corruption:** A middleware service silently truncates or mangles a JSON payload. The consumer processes the malformed data without error, and the downstream system acts on corrupted state.
  2. **Replay Injection:** A malicious actor captures a valid event (e.g., `transfer $100`) and replays it into the stream. Without a cryptographic watermark binding the event to a unique sequence and timestamp, the consumer processes the transaction twice.
  3. **Pipeline Tampering:** An attacker with access to the stream injects a fake event that looks identical to a legitimate one. Without a signed watermark from the producer, the consumer has no way to distinguish the injected event from the real one.

Recorded Elements

  1. **`producer_hmac`:** The HMAC (Hash-based Message Authentication Code) generated by the producer over the payload, timestamp, and producer ID.
  2. **`process_chaining_hash`:** A hash generated by the current processor that combines the input event's `producer_hmac` with the processor's output hash, creating a cryptographic chain.
  3. **`watermark_sequence_epoch`:** A monotonic epoch or sequence number embedded in the watermark, preventing replay attacks.
  4. **`verification_receipt`:** The exact proof that the consumer validated the producer's HMAC and outputted the correct `process_chaining_hash`.

Condition of Validity

* **Applies when:** The event stream is consumed by untrusted or semi-trusted processing nodes, or the integrity of the payload cannot be guaranteed by the transport layer (e.g., TLS terminates at the broker, not the consumer). * **Does not apply when:** The stream is strictly contained within a single trust boundary (e.g., a single process or fully trusted intranet) where transport-layer integrity is sufficient, or the events are idempotent and completely non-sensitive.

Decision Boundary

  1. **Does the incoming event contain a `producer_hmac`?** If no, the event is unverified. The consumer must reject it and trigger an alert.
  2. **Does the `producer_hmac` validate against the payload and sequence number?** If no, the event has been tampered with or replayed. The consumer must discard it.
  3. **Does the consumer emit an output stream with a `process_chaining_hash`?** If no, the consumer cannot prove to the successor that it processed the event correctly.
  4. **Conversely, when can a consumer bypass watermarking?** Only if the event stream is explicitly marked as `NON_CRITICAL` and is used for telemetry where data loss or corruption is acceptable.

Operational Effect

  1. **Tamper Evidence:** Guarantees that any modification to the event payload or sequence is detected immediately, preventing silent corruption.
  2. **Replay Protection:** Binds the watermark to a monotonic sequence, preventing malicious actors from re-injecting valid events.
  3. **Audit Chain:** Provides a cryptographic chain of custody from producer to consumer, allowing successors to verify that the data was processed exactly as intended.
from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import hmac
import json

class StreamVerdict(Enum):
    VERIFIED_INTEGRITY = auto()
    TAMPERED = auto()
    REPLAY_DETECTED = auto()

@dataclass(frozen=True)
class StreamEvent:
    payload: str
    producer_hmac: str
    sequence_epoch: int

@dataclass(frozen=True)
class ProcessingContext:
    shared_secret: str
    last_seen_epoch: int

@dataclass(frozen=True)
class StreamManifest:
    verdict: StreamVerdict
    chaining_hash: str | None
    manifest_digest: str

class StreamIntegrityGuard:
    @staticmethod
    def _hash(data: str) -> str:
        return hashlib.sha256(data.encode("utf-8")).hexdigest()

    @classmethod
    def _compute_hmac(cls, payload: str, epoch: int, secret: str) -> str:
        msg = f"{payload}:{epoch}".encode("utf-8")
        return hmac.new(secret.encode("utf-8"), msg, hashlib.sha256).hexdigest()

    @classmethod
    def evaluate_stream_event(
        cls, event: StreamEvent, ctx: ProcessingContext
    ) -> StreamManifest:
        
        # Gate 1: Replay protection
        if event.sequence_epoch <= ctx.last_seen_epoch:
            return StreamManifest(
                verdict=StreamVerdict.REPLAY_DETECTED,
                chaining_hash=None,
                manifest_digest=cls._hash(f"REPLAY:{event.sequence_epoch}")
            )

        # Gate 2: HMAC validation
        expected_hmac = cls._compute_hmac(event.payload, event.sequence_epoch, ctx.shared_secret)
        
        if not hmac.compare_digest(expected_hmac, event.producer_hmac):
            return StreamManifest(
                verdict=StreamVerdict.TAMPERED,
                chaining_hash=None,
                manifest_digest=cls._hash(f"TAMPERED:{event.sequence_epoch}")
            )

        # Gate 3: Generate chaining hash
        chaining_hash = cls._hash(json.dumps({
            "in_hmac": event.producer_hmac,
            "epoch": str(event.sequence_epoch),
            "payload_hash": cls._hash(event.payload)
        }, sort_keys=True))
        
        return StreamManifest(
            verdict=StreamVerdict.VERIFIED_INTEGRITY,
            chaining_hash=chaining_hash,
            manifest_digest=cls._hash(f"VALID:{event.sequence_epoch}")
        )