Implementing Hash-Chained Audit Logs in Postgres

A reconciliation system’s audit trail is only as trustworthy as its resistance to silent edits. Row-level updated_at timestamps and application-layer is_deleted flags do not stop a database administrator, a compromised service account, or a bad migration from rewriting history — they only record that a write happened, not whether an earlier one was altered afterward. This page builds an append-only audit table in Postgres where every row carries a prev_hash pointer and a sha256 digest of its own payload chained to that pointer, so any retroactive edit breaks the chain and is detectable by recomputation rather than by trust. It is a concrete implementation of Immutable Append-Only Ledger Design, the pattern that underlies every control surface in Compliance, Audit Trails & Financial Controls, and the chained hashes produced here are exactly what gets bundled into evidence packages in Generating SOX 404 Evidence Bundles in Python.

The design has four moving parts: a schema that physically refuses UPDATE and DELETE, a Python append path that computes the chained hash before the row ever reaches the database, a verify_chain() walk that recomputes every hash from the first row forward, and a periodic integrity job that runs the walk on a schedule and alerts on the first break. None of this replaces access control — a role that can DROP TABLE can still destroy evidence — but it converts silent tampering into detectable tampering, which is the property auditors actually ask for.

This matters specifically because reconciliation systems generate the exact kind of records someone might want quietly rewritten: a match decision that moved a transaction out of an exception queue, a manual override that approved a batch outside normal tolerance, a routing rule change that suppressed alerts for a specific counterparty. A hash chain does not stop any of those actions from happening — it stops them from happening invisibly. Once a row is written, its digest is a function of every row before it, so editing row 4,000 changes what row 4,001’s prev_hash should have been, and that mismatch surfaces the moment anyone runs the walk, whether that is an automated nightly job or an external auditor handed a read-only replica.

Prerequisites

• A Postgres 13+ instance with a role dedicated to audit writes, distinct from the application’s general-purpose role.
• The audit table sits downstream of Immutable Append-Only Ledger Design — you understand which events (match decisions, exception escalations, manual overrides) must be captured.
• Python 3.10+ with psycopg (v3) and pydantic v2 installed.
• Monetary fields in any audited payload use Decimal, never float; timestamps are timezone-aware UTC.
• A structured logger configured to emit trace_id, source_hash, and match_decision on every append.

Step 1 — Build an append-only schema

The table itself is ordinary, but three controls make it structurally append-only rather than append-only by convention: revoked UPDATE/DELETE grants for the application role, and a trigger that raises on any mutation attempt even from a role that still holds the privilege (defense in depth against a misconfigured grant). payload_hash is the digest of the row’s own JSON payload; row_hash folds in prev_hash so the chain, not just the row, is verifiable.

CREATE TABLE audit_log (
    id            BIGSERIAL PRIMARY KEY,
    recorded_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
    trace_id      UUID NOT NULL,
    event_type    TEXT NOT NULL,
    payload       JSONB NOT NULL,
    payload_hash  CHAR(64) NOT NULL,
    prev_hash     CHAR(64),
    row_hash      CHAR(64) NOT NULL UNIQUE
);

CREATE INDEX audit_log_trace_id_idx ON audit_log (trace_id);

-- Least privilege: the writer role can only INSERT and SELECT.
REVOKE UPDATE, DELETE, TRUNCATE ON audit_log FROM audit_writer;
GRANT INSERT, SELECT ON audit_log TO audit_writer;

-- Defense in depth even if a grant is ever misconfigured.
CREATE OR REPLACE FUNCTION reject_audit_mutation() RETURNS TRIGGER AS $$
BEGIN
    RAISE EXCEPTION 'audit_log is append-only: % is not permitted on row %',
        TG_OP, COALESCE(OLD.id, NEW.id);
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER audit_log_no_update
    BEFORE UPDATE ON audit_log
    FOR EACH ROW EXECUTE FUNCTION reject_audit_mutation();

CREATE TRIGGER audit_log_no_delete
    BEFORE DELETE ON audit_log
    FOR EACH ROW EXECUTE FUNCTION reject_audit_mutation();

row_hash UNIQUE is deliberate: two rows can never legitimately share a chained hash, so a UNIQUE violation on insert is itself an early signal of a replay or a broken chain computation upstream, before you even run a full verification pass.

Two grant-layer decisions are worth calling out. First, TRUNCATE is revoked alongside UPDATE and DELETE — it is easy to forget that TRUNCATE bypasses row-level triggers entirely in Postgres, so a role that still holds it can erase the whole chain without ever firing reject_audit_mutation(). Second, the application’s audit writer role should be distinct from any role used for schema migrations; a migration role legitimately needs ALTER TABLE, and if that same role can also write rows, a compromised migration script becomes a path to forging history. Keep the two roles separate even though it adds a connection pool.

Step 2 — Compute the chained hash in Python before the write

The chain must be computed application-side, in the same order the rows will be committed, because Postgres itself has no notion of “the previous row’s hash” without a round trip. The writer reads the current chain tip, computes payload_hash over a canonical JSON encoding of the payload, folds it with prev_hash into row_hash, and inserts in a single transaction so no other writer can interleave between the tip read and the insert.

import hashlib
import json
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal
from uuid import UUID, uuid4
import logging

import psycopg

log = logging.getLogger("audit.chain")


def canonical_json(payload: dict) -> str:
    """Deterministic encoding so identical payloads always hash identically."""
    def default(obj):
        if isinstance(obj, Decimal):
            return str(obj)
        if isinstance(obj, datetime):
            return obj.astimezone(timezone.utc).isoformat()
        raise TypeError(f"unhashable type: {type(obj)!r}")

    return json.dumps(payload, sort_keys=True, separators=(",", ":"), default=default)


def sha256_hex(data: str) -> str:
    return hashlib.sha256(data.encode("utf-8")).hexdigest()


@dataclass(frozen=True)
class AuditAppendResult:
    row_id: int
    row_hash: str


def append_audit_row(
    conn: psycopg.Connection,
    trace_id: UUID,
    event_type: str,
    payload: dict,
    match_decision: str,
) -> AuditAppendResult:
    payload_json = canonical_json(payload)
    payload_hash = sha256_hex(payload_json)

    with conn.transaction():
        with conn.cursor() as cur:
            # Serialize on the chain tip so concurrent writers cannot fork it.
            cur.execute("SELECT row_hash FROM audit_log ORDER BY id DESC LIMIT 1 FOR UPDATE")
            row = cur.fetchone()
            prev_hash = row[0] if row else None

            row_hash = sha256_hex(f"{prev_hash}|{payload_hash}|{trace_id}")

            cur.execute(
                """
                INSERT INTO audit_log
                    (trace_id, event_type, payload, payload_hash, prev_hash, row_hash)
                VALUES (%s, %s, %s, %s, %s, %s)
                RETURNING id
                """,
                (str(trace_id), event_type, payload_json, payload_hash, prev_hash, row_hash),
            )
            row_id = cur.fetchone()[0]

    log.info(
        "audit.appended",
        extra={
            "trace_id": str(trace_id),
            "source_hash": payload_hash,
            "match_decision": match_decision,
            "row_id": row_id,
            "row_hash": row_hash,
        },
    )
    return AuditAppendResult(row_id=row_id, row_hash=row_hash)

SELECT ... FOR UPDATE on the tip row (rather than a table-level lock) is what makes concurrent appends safe: two workers racing to write the next link both block on the same tip row, and the second one recomputes prev_hash against the first worker’s freshly committed row_hash once it acquires the lock. This is a narrower cousin of the FOR UPDATE SKIP LOCKED pattern used in Manual Review Queue Design — here we want strict serialization, not lock-skipping, because the chain has exactly one valid next link.

A concrete call site looks like this, using Decimal for the audited amount and a UUID for the trace:

with psycopg.connect(DSN, autocommit=False) as conn:
    append_audit_row(
        conn,
        trace_id=uuid4(),
        event_type="MATCH_DECISION",
        payload={
            "ledger_batch_ref": "BATCH-2026-07-14",
            "amount_usd": Decimal("18420.50"),
            "matched_at": datetime.now(timezone.utc),
        },
        match_decision="AUTO_MATCHED",
    )

Step 3 — Walk and verify the chain

verify_chain() reconstructs the chain from the first row and recomputes each row_hash independently of what is stored, comparing at every step. Because it recomputes rather than trusts, it catches both a mutated payload (the payload_hash no longer matches the recomputed digest) and a mutated prev_hash or row_hash (the link itself no longer resolves).

from dataclasses import dataclass


@dataclass(frozen=True)
class ChainBreak:
    row_id: int
    reason: str


def verify_chain(conn: psycopg.Connection, batch_size: int = 5000) -> list[ChainBreak]:
    breaks: list[ChainBreak] = []
    prev_hash: str | None = None
    last_id = 0

    with conn.cursor() as cur:
        while True:
            cur.execute(
                """
                SELECT id, trace_id, payload, payload_hash, prev_hash, row_hash
                FROM audit_log
                WHERE id > %s
                ORDER BY id ASC
                LIMIT %s
                """,
                (last_id, batch_size),
            )
            rows = cur.fetchall()
            if not rows:
                break

            for row_id, trace_id, payload, stored_payload_hash, stored_prev_hash, stored_row_hash in rows:
                recomputed_payload_hash = sha256_hex(canonical_json(payload))
                if recomputed_payload_hash != stored_payload_hash:
                    breaks.append(ChainBreak(row_id, "PAYLOAD_HASH_MISMATCH"))
                if stored_prev_hash != prev_hash:
                    breaks.append(ChainBreak(row_id, "PREV_HASH_MISMATCH"))

                recomputed_row_hash = sha256_hex(f"{prev_hash}|{recomputed_payload_hash}|{trace_id}")
                if recomputed_row_hash != stored_row_hash:
                    breaks.append(ChainBreak(row_id, "ROW_HASH_MISMATCH"))

                prev_hash = stored_row_hash
                last_id = row_id

    log.info(
        "audit.chain_verified",
        extra={
            "trace_id": "chain-walk",
            "source_hash": prev_hash or "genesis",
            "match_decision": "CLEAN" if not breaks else "CHAIN_BREAK_DETECTED",
            "break_count": len(breaks),
        },
    )
    return breaks

Note that prev_hash continues to advance to stored_row_hash even when a mismatch is found — you want a single scan to enumerate every break, not to stop at the first one, since one tampered row and a later independent tamper should both surface in the same report.

The walk is deliberately linear and single-threaded. It is tempting to parallelize verification across ID ranges to speed up a full re-walk over a multi-million-row table, but the chain has a sequential dependency by construction — worker N cannot know the correct prev_hash for its first row without worker N-1 finishing. The practical compromise is to keep the walk single-threaded but checkpointed, so a full re-walk resumes from the last confirmed id/row_hash pair instead of restarting from genesis, and to reserve parallel scanning for read-only spot checks that compare a sampled row’s payload_hash against a separately stored digest rather than for the authoritative chain verification itself.

Step 4 — Run a periodic integrity job

Verification is cheap enough to run continuously on new rows but expensive to run in full over a multi-year table, so split it into an incremental tail check (every few minutes) and a full re-walk (nightly, or before quarterly audit close). Persist the last verified id/row_hash pair somewhere outside audit_log itself — a small audit_checkpoint table — so a restart resumes rather than re-verifying from genesis.

import time


def run_integrity_job(conn: psycopg.Connection, checkpoint_table: str = "audit_checkpoint") -> None:
    with conn.cursor() as cur:
        cur.execute(f"SELECT last_verified_id FROM {checkpoint_table} LIMIT 1")
        row = cur.fetchone()
        start_id = row[0] if row else 0

    started = time.monotonic()
    breaks = verify_chain(conn, batch_size=5000)
    duration_s = time.monotonic() - started

    if breaks:
        for b in breaks:
            log.error(
                "audit.integrity_breach",
                extra={
                    "trace_id": str(b.row_id),
                    "source_hash": "n/a",
                    "match_decision": "CHAIN_BREAK_DETECTED",
                    "reason": b.reason,
                },
            )
        raise RuntimeError(f"{len(breaks)} audit chain break(s) detected; halting checkpoint advance")

    with conn.cursor() as cur:
        cur.execute(
            f"UPDATE {checkpoint_table} SET last_verified_id = "
            "(SELECT max(id) FROM audit_log), verified_at = now(), duration_s = %s",
            (duration_s,),
        )
    conn.commit()

Wire this into the same scheduler used for the SLA and escalation jobs elsewhere on the site (Celery beat, APScheduler, or a Kubernetes CronJob) and export audit_verify_duration_seconds and audit_chain_breaks_total as Prometheus metrics. A single non-zero audit_chain_breaks_total should page whoever owns the compliance surface — unlike a match-quality alert, this one has no acceptable non-zero baseline.

Treat the checkpoint table itself as part of the trust boundary. If last_verified_id can be rewound by an ordinary application role, an attacker who tampers with an old row can simply reset the checkpoint to skip past it on the next run, and the incremental job will never re-examine the affected range until the next full genesis-to-tip walk. Grant write access to audit_checkpoint only to the same restricted role that runs run_integrity_job(), and schedule the full re-walk frequently enough — nightly is reasonable for most reconciliation volumes — that this window stays small even if the incremental job is compromised.

Configuration boundary table

Verification and testing

A fixture with three chained rows, one deliberately corrupted after insertion, exercises both the happy path and the break-detection path without needing a live Postgres instance for the hashing logic itself — only verify_chain()'s SQL needs a real connection, which you can point at a disposable test database or a psycopg connection wrapping a transaction that’s rolled back at the end of the test.

def test_verify_chain_detects_tamper(conn: psycopg.Connection):
    trace_a, trace_b, trace_c = uuid4(), uuid4(), uuid4()

    r1 = append_audit_row(conn, trace_a, "MATCH_DECISION",
                           {"amount_usd": Decimal("100.00")}, "AUTO_MATCHED")
    r2 = append_audit_row(conn, trace_b, "MATCH_DECISION",
                           {"amount_usd": Decimal("250.75")}, "AUTO_MATCHED")
    r3 = append_audit_row(conn, trace_c, "EXCEPTION_ROUTED",
                           {"amount_usd": Decimal("9999.99")}, "ROUTED_TO_REVIEW")

    assert verify_chain(conn) == []  # clean chain, no breaks

    # Simulate tampering by bypassing the trigger with a superuser session
    # (in production this path should not exist for the application role).
    with conn.cursor() as cur:
        cur.execute("ALTER TABLE audit_log DISABLE TRIGGER audit_log_no_update")
        cur.execute(
            "UPDATE audit_log SET payload = %s WHERE id = %s",
            (canonical_json({"amount_usd": Decimal("90000.00")}), r2.row_id),
        )
        cur.execute("ALTER TABLE audit_log ENABLE TRIGGER audit_log_no_update")
    conn.commit()

    breaks = verify_chain(conn)
    reasons = {b.reason for b in breaks if b.row_id == r2.row_id}
    assert "PAYLOAD_HASH_MISMATCH" in reasons
    # Downstream row_hash for r2 no longer matches what r3 was chained against.
    assert any(b.row_id == r2.row_id and b.reason == "ROW_HASH_MISMATCH" for b in breaks)

Disabling the trigger to simulate tampering is intentional: it proves the detection lives in the hash chain itself, not in the trigger, which is the property you actually need — a superuser or a pg_dump/restore that skips triggers should not be able to forge history undetected.

Troubleshooting

• CHAIN_BREAK_DETECTED — verify_chain() returns one or more ChainBreak entries on a table nobody admits to touching. Root cause: usually a restore from an out-of-order backup, or a migration that ran COPY and reset row ordering. Fix: restore from the last checkpoint with matching row_hash, and audit who ran the restoring migration.
• ROW_HASH_MISMATCH on the row immediately after a known-good one. Root cause: prev_hash was computed against a stale tip because two writers bypassed the FOR UPDATE lock (e.g., one used a different connection pool with autocommit=True, skipping the transaction wrapper). Fix: ensure every writer path goes through append_audit_row(), and add a NOT VALID check that rejects direct INSERTs missing row_hash.
• PSYCOPG_LOCK_TIMEOUT — appends start failing under load with a lock-wait error on the tip row. Root cause: chain_lock_timeout_ms is too low for write concurrency, or a long-running transaction is holding the tip lock open. Fix: raise chain_lock_timeout_ms modestly and audit for any transaction that calls append_audit_row() without committing promptly.
• GENESIS_HASH_MISMATCH — verification fails on row 1 specifically. Root cause: the genesis row’s prev_hash was not NULL (e.g., seeded from a template with a placeholder value) or verify_chain() was pointed at the wrong table partition. Fix: confirm prev_hash IS NULL for the first row and that last_verified_id in the checkpoint table was reset to 0 for a from-scratch walk.
• DUPLICATE_ROW_HASH — the row_hash UNIQUE constraint rejects an insert. Root cause: two payloads with identical content and the same trace_id/prev_hash were appended (a retried write without idempotency), producing an identical digest. Fix: derive trace_id from a stable idempotency key upstream and dedupe before calling append_audit_row(), mirroring the idempotency pattern in Immutable Append-Only Ledger Design.

Related

• Immutable Append-Only Ledger Design
• Generating SOX 404 Evidence Bundles in Python
• Gating Reconciliation Deploys with GitHub Actions

Part of Immutable Append-Only Ledger Design within Compliance, Audit Trails & Financial Controls.