Building Hash-Chained Audit Logs for 21 CFR Part 11

An auditor asks you to prove that no one altered a serial’s history after the fact, and a plain append-only table cannot answer that question — a database administrator with UPDATE access can rewrite a row and the timestamp column will happily lie for them. This guide solves that exact gap within Data Security & Encryption Boundaries, the DSCSA Compliance Architecture & Standards Mapping area’s control set for confidentiality and integrity: an audit log where every entry embeds the SHA-256 digest of the entry before it, so altering or deleting any record breaks a chain that a verifier can check in one pass. That property — not encryption, not access control — is what 21 CFR Part 11 means by an electronic record trustworthy enough to stand in for a signature.

Prerequisites #

• Python 3.10+ — the snippets use X | Y union syntax, list[dict] generics, and datetime.timezone.utc.
• Pydantic v2 (pydantic>=2.0) — for the AuditLogEntry model, field_validator, and model_copy to produce sealed, immutable entries.
• Standard library hashlib and json — no third-party crypto package is required for SHA-256 chaining; canonical JSON serialization plus hashlib.sha256 is sufficient and auditable.
• DSCSA data prerequisites — a serialized unit identified by GTIN (01) and serial (21), ideally with lot (10) and expiry (17) available for cross-reference, plus a stable acting-identity string (a service account name or an authenticated user ID) for every lifecycle transition you log.
• A write-once or append-only sink — object storage with object-lock, a WORM-configured filesystem, or an insert-only database table. The hash chain proves tampering happened; it does not prevent a privileged operator from deleting the underlying file, so the storage layer still matters.

Step-by-Step Solution #

Step 1 — Define the audit entry as a canonical, hashable Pydantic model #

Every lifecycle transition — commission, aggregate, ship, decommission, verify — becomes one immutable entry. The model captures the acting identity and a UTC timestamp, and it exposes a deterministic byte representation so two processes hashing the same entry always get the same digest, regardless of dict key ordering or locale.

from __future__ import annotations

import hashlib
import json
from datetime import datetime, timezone
from enum import Enum
from typing import Optional

from pydantic import BaseModel, field_validator

GENESIS_HASH = "0" * 64  # conventional predecessor for sequence 0


class LifecycleEvent(str, Enum):
    COMMISSION = "commission"
    AGGREGATE = "aggregate"
    SHIP = "ship"
    DECOMMISSION = "decommission"
    VERIFY = "verify"


class AuditLogEntry(BaseModel):
    model_config = {"frozen": True}

    sequence: int
    event_type: LifecycleEvent
    gtin: str            # AI (01)
    serial: str           # AI (21)
    actor_id: str          # acting identity: service account or authenticated user
    recorded_at: datetime  # UTC timestamp of the transition
    previous_hash: str
    entry_hash: Optional[str] = None

    @field_validator("recorded_at")
    @classmethod
    def _must_be_utc_aware(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("recorded_at must carry a timezone offset")
        return v.astimezone(timezone.utc)

    @field_validator("previous_hash", "entry_hash")
    @classmethod
    def _hash_shape(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return v
        if len(v) != 64 or any(c not in "0123456789abcdef" for c in v):
            raise ValueError("hash must be a 64-char lowercase hex SHA-256 digest")
        return v

    def canonical_payload(self) -> bytes:
        """Deterministic bytes hashed into entry_hash — sorted keys, no whitespace."""
        payload = {
            "sequence": self.sequence,
            "event_type": self.event_type.value,
            "gtin": self.gtin,
            "serial": self.serial,
            "actor_id": self.actor_id,
            "recorded_at": self.recorded_at.isoformat(),
            "previous_hash": self.previous_hash,
        }
        return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()

    def compute_hash(self) -> str:
        return hashlib.sha256(self.canonical_payload()).hexdigest()

    def sealed(self) -> "AuditLogEntry":
        """Return an immutable copy with entry_hash populated from the payload."""
        return self.model_copy(update={"entry_hash": self.compute_hash()})

DSCSA/GS1 note: gtin and serial tie every entry back to a specific saleable unit, so a verifier or investigator can filter the chain down to one serial’s lifecycle without touching any other record — the exact reconstruction 21 CFR Part 11 audit trails must support.

Step 2 — Wrap entries in an append-only writer that owns the chain #

The application never sets previous_hash or entry_hash directly; the writer does, by reading the tail of the chain before constructing the next entry. This keeps the chaining logic in one place and makes it impossible to append an entry out of sequence.

class HashChainedAuditLog:
    """Append-only log. Swap the in-memory list for a WORM-backed store in production."""

    def __init__(self) -> None:
        self._entries: list[AuditLogEntry] = []

    def append(
        self,
        event_type: LifecycleEvent,
        gtin: str,
        serial: str,
        actor_id: str,
        recorded_at: datetime | None = None,
    ) -> AuditLogEntry:
        previous_hash = self._entries[-1].entry_hash if self._entries else GENESIS_HASH
        draft = AuditLogEntry(
            sequence=len(self._entries),
            event_type=event_type,
            gtin=gtin,
            serial=serial,
            actor_id=actor_id,
            recorded_at=recorded_at or datetime.now(timezone.utc),
            previous_hash=previous_hash,
        )
        entry = draft.sealed()
        self._entries.append(entry)
        return entry

    def entries(self) -> list[AuditLogEntry]:
        return list(self._entries)

DSCSA/GS1 note: a commission entry corresponds to an ObjectEvent with a commissioning business step, aggregate to an AggregationEvent, and ship to a TransactionEvent — the audit log is a parallel, tamper-evident ledger of the same lifecycle the EPCIS repository already stores, not a replacement for it.

Step 3 — Persist sealed entries as append-only JSON Lines #

A durable log has to survive process restarts. Serialize each sealed entry as one JSON line and open the file in append mode only — never in a mode that permits seeking backward and rewriting a prior line.

def write_entry(path: str, entry: AuditLogEntry) -> None:
    with open(path, mode="a", encoding="utf-8") as fh:
        fh.write(entry.model_dump_json() + "\n")
        fh.flush()


def load_entries(path: str) -> list[AuditLogEntry]:
    entries: list[AuditLogEntry] = []
    with open(path, encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if line:
                entries.append(AuditLogEntry.model_validate_json(line))
    return entries

DSCSA/GS1 note: storing one JSON object per line, in sequence order, means the reconstruction an inspector demands for a single serial is a linear scan and filter — no reconciliation step, no derived state that could itself drift from the source records.

Step 4 — Verify the chain and name exactly where it breaks #

The verifier recomputes every hash from the stored fields and confirms each entry’s previous_hash matches the prior entry’s entry_hash. It also checks sequence is contiguous, which catches a deleted entry even if the surrounding hashes were somehow patched consistently.

class ChainIntegrityError(Exception):
    """Raised when a hash-chained audit log fails verification."""


def verify_chain(entries: list[AuditLogEntry]) -> None:
    expected_previous = GENESIS_HASH
    for position, entry in enumerate(entries):
        if entry.sequence != position:
            raise ChainIntegrityError(
                f"gap at position {position}: found sequence {entry.sequence} "
                "— an entry was deleted or reordered"
            )
        if entry.previous_hash != expected_previous:
            raise ChainIntegrityError(
                f"entry {entry.sequence}: previous_hash does not match the "
                "prior entry's stored hash — chain broken before this point"
            )
        if entry.compute_hash() != entry.entry_hash:
            raise ChainIntegrityError(
                f"entry {entry.sequence}: recomputed hash does not match "
                "entry_hash — a field was altered after sealing"
            )
        expected_previous = entry.entry_hash

DSCSA/GS1 note: the error names the exact sequence where trust ends, so a compliance officer investigating a discrepancy gets a precise starting point instead of re-verifying six years of records by hand.

Step 5 — Record a serial’s full lifecycle and prove it is intact #

Tie the pieces together: append every transition as it happens, then run the verifier as a scheduled job or before any export to an inspector.

log = HashChainedAuditLog()
log.append(LifecycleEvent.COMMISSION, gtin="00312345000010", serial="SERIAL001",
           actor_id="svc-packaging-line-4")
log.append(LifecycleEvent.AGGREGATE, gtin="00312345000010", serial="SERIAL001",
           actor_id="svc-case-aggregator")
log.append(LifecycleEvent.SHIP, gtin="00312345000010", serial="SERIAL001",
           actor_id="user:jdoe@3pl.example")
log.append(LifecycleEvent.VERIFY, gtin="00312345000010", serial="SERIAL001",
           actor_id="vrs-router")

verify_chain(log.entries())  # raises ChainIntegrityError if anything was touched

DSCSA/GS1 note: logging verify alongside commission, aggregate, ship, and decommission means the audit trail captures not just physical custody events but every trading-partner verification lookup answered for the unit — evidence regulators increasingly expect during a suspect-product review, the same review path covered in Suspect Product Investigation Workflows.

Verification #

Confirm both the happy path and the failure path before trusting this in production. The happy-path test appends a short lifecycle and asserts verify_chain is silent; the tamper test mutates a sealed entry’s field in place (simulating a row edited directly in storage, bypassing the API) and asserts the verifier catches it.

import pytest
from datetime import datetime, timezone


def _sample_log() -> HashChainedAuditLog:
    log = HashChainedAuditLog()
    ts = datetime(2026, 7, 17, 12, 0, tzinfo=timezone.utc)
    log.append(LifecycleEvent.COMMISSION, "00312345000010", "SERIAL001", "svc-line-4", ts)
    log.append(LifecycleEvent.SHIP, "00312345000010", "SERIAL001", "user:jdoe", ts)
    return log


def test_intact_chain_verifies_clean():
    verify_chain(_sample_log().entries())  # no exception


def test_altered_field_is_detected():
    entries = _sample_log().entries()
    tampered = entries[0].model_copy(update={"actor_id": "attacker"})
    entries[0] = tampered  # bypasses sealed() — entry_hash now stale
    with pytest.raises(ChainIntegrityError, match="recomputed hash"):
        verify_chain(entries)


def test_deleted_entry_is_detected():
    entries = _sample_log().entries()
    del entries[0]  # remove the commission entry entirely
    with pytest.raises(ChainIntegrityError, match="gap at position"):
        verify_chain(entries)

Run this against your real store, too: load entries with load_entries, call verify_chain on the result, and schedule the check to run before every export bundle you hand to an inspector or trading partner. A verification pass that only runs against in-memory fixtures proves the algorithm works; it does not prove your storage layer preserved the file untouched, so the recurring job against the persisted log is the check that actually matters for audit readiness.

Gotchas & Edge Cases #

• Non-monotonic wall clocks. NTP adjustments or container restarts can make datetime.now(timezone.utc) jump backward between entries. The chain’s integrity does not depend on timestamp ordering, but if your compliance report also asserts entries are chronological, capture a monotonic counter alongside recorded_at rather than trusting wall-clock time alone.
• Concurrent appenders. Two workers computing previous_hash from the same tail at once will both mint an entry claiming to be the true successor. Serialize appends through a single writer process, a database row lock, or a distributed lock — never let two processes read-then-append against the same tail unguarded.
• Rotating the log file. If you roll to a new file after N entries, carry the last entry_hash forward as the new file’s first previous_hash (do not reset to GENESIS_HASH), or verify_chain will see a false break at the rotation boundary.
• JSON float and Unicode drift. json.dumps must use the same separators, key order, and float formatting on every machine that computes a hash. Pin sort_keys=True, separators=(",", ":") exactly as shown; a differently configured serializer on a second server produces a different digest for identical data.
• Re-sealing after a schema migration. Adding a field to AuditLogEntry changes canonical_payload() and therefore every future hash. Existing sealed entries must keep validating against the old payload shape, or store a schema version in the payload itself so old and new entries hash consistently within their own era.

Related #

• Up to the parent section: Data Security & Encryption Boundaries
• Implementing AES-256 for Serialized Data at Rest — the encryption-at-rest control this audit log complements
• Suspect Product Investigation Workflows — where hash-chained history is pulled during an investigation
• DSCSA Compliance Architecture & Standards Mapping — the area this control set supports