InventoryAllocation

Designing Immutable Audit-Log Schemas for Rate and Inventory Changes

An audit log is only worth keeping if you can prove nobody edited it, and that proof cannot rest on “our storage is locked down” — an auditor wants tamper-evidence baked into the records themselves. This page specifies the schema for an append-only audit record whose integrity is self-verifying: a SHA-256 hash chain that makes any retroactive edit mathematically detectable, sitting one level below the compliance and audit logging workflow that owns the end-to-end trail. The reader here is the Python engineer designing the record shape that will hold years of rate and inventory history.

The core idea is borrowed from block-linked ledgers without any of the distributed-consensus machinery: each record carries the hash of the one before it, so the records form a chain where altering entry n changes its hash and therefore breaks the prev_hash of entry n+1, and every entry after it. A verifier that walks the chain from a known genesis detects the first break and reports exactly where it is.

Prerequisites & environment

The schema and verifier below run wherever audit records are written and wherever they are later read for an integrity check. Pin these so the hash input is byte-identical on both sides:

The one hard requirement is determinism: if the bytes hashed at write time differ in any way from the bytes hashed at verify time, every record reads as tampered. Money and timestamps are the usual culprits, so both are normalized to strings in the record before hashing.

A hash-chained audit record sequence and where a tamper breaks it Three audit records are drawn left to right at sequence zero, one and two. Each record shows its sequence number, an actor and action, a before-after value pair, its prev_hash field, and its own record_hash. An arrow from each record's record_hash feeds the next record's prev_hash, forming a chain anchored at a GENESIS value. Below, a callout shows that editing the value in record one changes its record_hash, so record two's prev_hash no longer matches and the verifier flags the break at sequence one. Hash-chained audit records — each record_hash anchors the next prev_hash Editing any record changes its hash and breaks every link after it, so the verifier pinpoints the first tampered entry. GENESIS seq 0 · rate.published actor worker:pricing before — · after 180.00 prev_hash GENESIS record_hash 9f2a… seq 1 · rate.published actor user:rm_jbrown before 180.00 · after 185.00 prev_hash 9f2a… record_hash c71e… seq 2 · stop_sell actor worker:inventory before off · after on prev_hash c71e… record_hash 4d80… Tamper at seq 1 Editing after 185.00 → 95.00 changes seq 1's record_hash from c71e… to a new value. Seq 2's prev_hash still reads c71e… → mismatch → verifier returns 1, the first broken record.
Figure 1: three chained audit records — each entry's record_hash feeds the next entry's prev_hash, so editing sequence 1's value breaks the link to sequence 2 and the verifier reports the break at sequence 1.

Step-by-step implementation

Step 1 — Model the record with actor, action, and before/after

The record’s semantic core is who did what to which value. Capture the actor, the action type, and the before/after pair, plus the fields that let a verifier reproduce the hash: a monotonic sequence and the prev_hash it links to.

python
from datetime import datetime
from enum import Enum
from pydantic import BaseModel, Field, field_validator

class Action(str, Enum):
    RATE_PUBLISHED = "rate.published"
    INVENTORY_ADJUSTED = "inventory.adjusted"
    STOP_SELL_TOGGLED = "stop_sell.toggled"

class AuditRecord(BaseModel):
    sequence: int = Field(ge=0)             # monotonic, per property_id
    occurred_at: datetime                   # timezone-aware UTC
    property_id: str = Field(pattern=r"^[A-Z]{3}-[A-Z0-9]{3,}-\d{2}$")
    actor: str = Field(pattern=r"^(user|worker|system):[a-z0-9_\-]+$")
    action: Action
    subject: str                            # e.g. "DLX_KING/BAR" — what changed
    before: str | None                      # prior value as a string, None on create
    after: str                              # new value as a string
    prev_hash: str = Field(min_length=8)    # record_hash of sequence - 1, or "GENESIS"

    @field_validator("occurred_at")
    @classmethod
    def utc_only(cls, v: datetime) -> datetime:
        if v.tzinfo is None or v.utcoffset() is None:
            raise ValueError("occurred_at must be timezone-aware UTC")
        return v

Storing before and after as strings rather than native numbers is the deliberate choice that keeps the hash deterministic: a Decimal("180.00") and a float 180.0 render differently, so pinning the on-record form to a string means the digest cannot drift between a write in one service and a verify in another.

Step 2 — Compute the record hash over a canonical serialization

The hash covers the record’s content plus its sequence and prev_hash, serialized canonically so key order and whitespace never affect the digest. Exclude the record_hash field itself — it is the output, not an input.

python
import hashlib
import json

def compute_record_hash(record: AuditRecord) -> str:
    body = record.model_dump(mode="json", exclude={"prev_hash"})
    payload = json.dumps(body, sort_keys=True, separators=(",", ":"))
    # prev_hash is folded in explicitly so the chain link is part of the digest
    return hashlib.sha256(f"{record.prev_hash}|{payload}".encode()).hexdigest()

Using model_dump(mode="json") gives a JSON-safe dict (datetimes and enums already stringified) that matches exactly what gets stored, so the verifier can rebuild the identical byte string from the persisted record without a second, divergent serialization path.

Step 3 — Append a record by linking it to the chain head

Appending means reading the current head’s hash, allocating the next sequence number, computing this record’s hash, and writing it. The sequence and the prev_hash must come from the same head read to avoid a torn append under concurrency.

python
def append_record(new: AuditRecord, chain_head: AuditRecord | None) -> dict:
    prev_hash = "GENESIS" if chain_head is None else compute_record_hash(chain_head)
    expected_seq = 0 if chain_head is None else chain_head.sequence + 1
    if new.sequence != expected_seq:
        raise ValueError(f"sequence gap: expected {expected_seq}, got {new.sequence}")
    linked = new.model_copy(update={"prev_hash": prev_hash})
    return {**linked.model_dump(mode="json"),
            "record_hash": compute_record_hash(linked)}

Asserting new.sequence == chain_head.sequence + 1 before writing turns a concurrent double-append into a loud failure instead of a silent fork in the chain, so two workers racing to extend the same property’s log cannot both succeed with the same sequence number.

Step 4 — Verify the chain and pinpoint the first break

The verifier walks records in sequence order, recomputing each hash from the previous record’s hash. It returns the sequence of the first record whose stored hash or prev-link does not match — the tamper location.

python
def verify_chain(records: list[dict]) -> int | None:
    """Return the sequence of the first broken record, or None if the chain is intact."""
    prev_hash = "GENESIS"
    for raw in sorted(records, key=lambda r: r["sequence"]):
        stored_hash = raw["record_hash"]
        rebuilt = AuditRecord(**{k: v for k, v in raw.items()
                                 if k != "record_hash"})
        if rebuilt.prev_hash != prev_hash:
            return rebuilt.sequence                 # link to predecessor is broken
        if compute_record_hash(rebuilt) != stored_hash:
            return rebuilt.sequence                 # record body was edited
        prev_hash = stored_hash
    return None

Checking prev_hash continuity and recomputing the body hash catches two distinct attacks: editing a record’s content (body hash changes) and splicing a record out of the sequence (the link to the predecessor no longer matches), neither of which a single check alone would detect.

Gotchas & production notes

Verification snippet

Confirm the two behaviors the whole design rests on: an intact chain verifies clean, and any edit is caught at the exact record where it happened.

python
from datetime import datetime, timezone

def _rec(seq: int, after: str, prev: str) -> AuditRecord:
    return AuditRecord(sequence=seq, occurred_at=datetime.now(timezone.utc),
                       property_id="LON-STJ-01", actor="worker:pricing",
                       action="rate.published", subject="DLX_KING/BAR",
                       before=None, after=after, prev_hash=prev)

def test_intact_chain_verifies():
    r0 = {**_rec(0, "180.00", "GENESIS").model_dump(mode="json"),
          "record_hash": compute_record_hash(_rec(0, "180.00", "GENESIS"))}
    h0 = r0["record_hash"]
    r1 = {**_rec(1, "185.00", h0).model_dump(mode="json"),
          "record_hash": compute_record_hash(_rec(1, "185.00", h0))}
    assert verify_chain([r0, r1]) is None

def test_edit_is_caught_at_its_sequence():
    r0 = {**_rec(0, "180.00", "GENESIS").model_dump(mode="json"),
          "record_hash": compute_record_hash(_rec(0, "180.00", "GENESIS"))}
    h0 = r0["record_hash"]
    r1 = {**_rec(1, "185.00", h0).model_dump(mode="json"),
          "record_hash": compute_record_hash(_rec(1, "185.00", h0))}
    r1["after"] = "95.00"                      # tamper: edit the stored value in place
    assert verify_chain([r0, r1]) == 1         # break detected exactly at sequence 1

test_intact_chain_verifies()
test_edit_is_caught_at_its_sequence()

The second test is the one that matters: it mutates a stored record’s after field without recomputing its hash — exactly what an in-place database edit would do — and asserts the verifier returns the tampered sequence rather than a vague failure. Run this verifier on a schedule and alert on any non-None result, so the chain’s promise is continuously exercised rather than trusted in the abstract.

← Back to Compliance & Audit Logging for Sync Pipelines