Message Authentication Codes and the Data-Integrity Discipline for Trust-Coupled Pipelines
Python ships the tools for data integrity in the standard library. No pip install, no third-party dependency, no version-pinning ceremony. hashlib covers content hashing. hmac covers message authentication. secrets generates the signing keys. The three modules together are sufficient to implement the IntegrityEnvelope that today's Ops lesson specifies.
The reason to build this from hashlib and hmac rather than reaching for a cryptography library is deliberate. hashlib calls the system's OpenSSL backend; the implementation is C, and the performance overhead per call is negligible for message-sized payloads. hmac in the standard library has been audited for the specific timing vulnerability that naive == comparisons introduce. secrets uses the OS-level CSPRNG. The standard library combination is right for the pipeline-integrity use case: signing messages, verifying messages, no key exchange, no certificate infrastructure.
hashlib exposes hash algorithms through a consistent interface. Call hashlib.sha256() to get a hash object, call .update(data: bytes) one or more times to feed bytes, call .hexdigest() for a lowercase hex string.
import hashlib
def sha256_hex(data: bytes) -> str:
h = hashlib.sha256()
h.update(data)
return h.hexdigest()
Feed bytes, not str. If the payload starts as a string, encode to UTF-8 explicitly before hashing. Implicit encoding assumptions produce hashes that differ across environments when the default encoding differs.
hashlib.sha3_256() works identically but uses the Keccak construction. The SHA-3 family shares no design ancestry with SHA-2, so a theoretical weakness in one does not carry to the other. hashlib.algorithms_guaranteed lists the algorithms the current Python version guarantees regardless of the OpenSSL build; SHA-256 and SHA-3-256 are on that list in every CPython 3.6+ installation.
hmac.new(key: bytes, msg: bytes, digestmod) returns an HMAC object. Call .hexdigest() for the tag.
import hmac
import hashlib
def compute_hmac(key: bytes, data: bytes) -> str:
return hmac.new(key, data, hashlib.sha256).hexdigest()
The key must be at least 32 bytes for HMAC-SHA256. NIST recommends key length equal to the output length of the underlying hash (32 bytes for SHA-256). Generate keys with secrets.token_bytes(32), never with random.
hmac.compare_digest(a, b) compares two HMAC strings in constant time. The comparison does not short-circuit on the first mismatch. An attacker cannot learn anything about the expected tag by timing how long the comparison takes. Always use compare_digest for HMAC verification; never use ==.
import secrets signing_key: bytes = secrets.token_bytes(32)
Store the key in a secret manager at startup. Do not embed it in configuration files or environment variables that get logged. Fetch once at process start, hold in memory, rotate on a schedule determined by your key-management policy.
The IntegrityEnvelope is a signed payload container. An orchestrator calls sign() to create one; a consumer calls verify() to validate it before touching the contents.
from __future__ import annotations
import hashlib
import hmac
import json
import secrets
import time
from dataclasses import dataclass
from typing import Any
@dataclass(frozen=True)
class IntegrityEnvelope:
seq: int
ts: str
writer: str
payload_sha256: str
hmac_sha256: str
payload: dict[str, Any]
@classmethod
def sign(
cls,
payload: dict[str, Any],
seq: int,
writer: str,
key: bytes,
) -> IntegrityEnvelope:
ts = _utc_now()
canonical = _canonicalize(payload)
payload_hash = hashlib.sha256(canonical).hexdigest()
signing_input = _signing_input(seq, ts, writer, payload_hash)
tag = hmac.new(key, signing_input, hashlib.sha256).hexdigest()
return cls(
seq=seq,
ts=ts,
writer=writer,
payload_sha256=payload_hash,
hmac_sha256=tag,
payload=payload,
)
def verify(self, key: bytes) -> bool:
canonical = _canonicalize(self.payload)
expected_hash = hashlib.sha256(canonical).hexdigest()
if not hmac.compare_digest(expected_hash, self.payload_sha256):
return False
signing_input = _signing_input(
self.seq, self.ts, self.writer, self.payload_sha256
)
expected_tag = hmac.new(key, signing_input, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected_tag, self.hmac_sha256)
def _canonicalize(payload: dict[str, Any]) -> bytes:
return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
def _utc_now() -> str:
return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
def _signing_input(seq: int, ts: str, writer: str, payload_hash: str) -> bytes:
parts = f"{seq}\n{ts}\n{writer}\n{payload_hash}"
return parts.encode("utf-8")
frozen=True on the dataclass means the envelope is immutable after construction. _canonicalize sorts JSON keys and strips whitespace, so the hash is stable across serialization contexts. _signing_input binds the HMAC tag to the sequence number, timestamp, writer identity, and payload hash — a replay of an old envelope with the same payload but a different sequence number produces a different expected tag and fails verification.
key = secrets.token_bytes(32)
seq_counter = 0
def produce(payload: dict[str, Any]) -> IntegrityEnvelope:
global seq_counter
seq_counter += 1
return IntegrityEnvelope.sign(
payload, seq=seq_counter, writer="orchestrator-v2", key=key
)
def consume(envelope: IntegrityEnvelope, seen_seqs: set[int]) -> dict[str, Any]:
if envelope.seq in seen_seqs:
raise ValueError(f"Replay detected: seq {envelope.seq} already processed")
if not envelope.verify(key):
raise ValueError(f"Integrity check failed for seq {envelope.seq}")
seen_seqs.add(envelope.seq)
return envelope.payload
seen_seqs is a monotonic set the consumer maintains across its lifetime. In production, persist this set to durable storage so it survives restarts; otherwise a restart resets replay protection to zero.
The Ops lesson defined the threat model: corruption (hash failure) vs tampering (HMAC failure) vs replay (sequence number re-use). It named three primitives: the Content Hash, the Message Authentication Code, the Immutable Ledger. This lesson implements the first two in Python directly from the standard library.
The Immutable Ledger is the WORM audit store — S3 Object Lock or CloudTrail Lake in AWS terms, covered in today's Cert lesson. From Python's side: every IntegrityEnvelope produced by sign() carries all the fields an audit record needs (seq, ts, writer, payload hash, HMAC tag). Writing those fields to the WORM store as a side-effect of every sign() call is one additional line per production site, and it makes the audit trail automatic rather than optional.
The secrets module appeared in the 27 May lesson on trust-coupled workloads — that lesson used it for random-token generation and stressed that random is not cryptographically safe for security-sensitive material. Here the same module provides signing keys; the 27 May discipline carries directly.
The asyncio lesson on 16 June built bounded-queue producers and consumers with backpressure. The consume() function above is synchronous; in an async pipeline it becomes async def consume() with the same verification logic inside an async with semaphore: block. The verification itself is CPU-bound, so it runs synchronously before any await.
Type annotations on the dataclass carry the typing discipline from the Protocols and Generics lesson on 8 June. A consumer that depends on IntegrityEnvelope through a Protocol rather than the concrete class can swap implementations without changing the consumer code.
hashlib.sha256() detects corruption. hmac.new() detects tampering. compare_digest() prevents the timing attack. frozen=True prevents in-memory mutation after signing. The sequence number prevents replay.
The standard library gives everything needed. The discipline is using it at every seam rather than only at the perimeter. Verify every envelope. Reject anything that fails. Log the rejection with the sequence number and the failing check. That record is where the investigation begins.