Building an Append-Only Audit Ledger in Postgres

This guide is part of the Audit Trail & Evidence Automation section within the broader Core Architecture & Compliance Mapping framework, and it solves one narrow problem: how do you record grant compliance events so that no one — not an intruder, not a careless administrator, not the application itself — can quietly rewrite history after the fact and have it go unnoticed?

A plain events table is not an audit trail. If the same database role that writes rows can also UPDATE or DELETE them, then an approval timestamp can be backdated, an actor can be swapped, and a rejected disbursement can be made to look approved — with nothing left to prove it happened. This guide builds a ledger where the database refuses mutation at the engine level and every row is sealed into a SHA-256 hash chain, so that a single-field edit or a removed row is detectable by anyone who re-walks the chain.

When to Use This Approach #

Reach for an append-only hash-chained ledger when all three of the following hold:

• The events are evidence, not telemetry. Grant compliance events — an artifact received, a budget approved, a subaward obligated — feed the internal-control regime that 2 CFR §200.303 requires a recipient to establish and maintain. If a row could be altered without a trace, the control is theatre. A tamper-evident ledger is what makes “we can prove what happened” a defensible claim rather than an assertion.
• You must survive a hostile or compromised operator. NIST SP 800-53 AU-9 calls for protecting audit information from unauthorized modification and deletion. Revoking write privileges from the application role, and chaining hashes so tampering is self-evident, are the two concrete mechanisms this page implements against that control.
• Records carry a legal retention obligation. Financial records tied to a federal award must be retained for the period set by 2 CFR §200.334 — generally three years from final report submission. A ledger you cannot edit is also a ledger you cannot accidentally shorten, so retention becomes a matter of when you stop appending, never of deleting rows.

Signing the compliance artifacts themselves is a separate concern and lives in the sibling guide on cryptographically signing grant compliance artifacts; this ledger stores only a subject_digest pointer to the signed object, never its bytes. Field-level protection of any personal data referenced by an event belongs to securing PII in nonprofit grant databases, and the vocabulary for event_type and retention_class values is governed by Compliance Metadata Standards. Those are explicitly out of scope here.

Step-by-Step Implementation #

The reference implementation targets PostgreSQL 14+ and Python 3.10+ with SQLAlchemy 2.x. The database enforcement (Steps 1–2) is pure DDL applied once by a migration role; the application code (Steps 3–5) connects as a lower-privileged role that holds only INSERT and SELECT. That privilege split is the point — the code that appends events is structurally incapable of rewriting them.

Step 1: Design the append-only ledger table #

The table carries exactly the fields an auditor needs to replay an event plus the two hash columns that chain rows together. subject_digest is a SHA-256 pointer to the compliance artifact the event concerns (not the artifact itself); retention_class names the disposal schedule so the retention automation can act on it later without touching row contents.

CREATE TABLE audit_ledger (
    entry_id        BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    occurred_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
    actor           TEXT        NOT NULL,
    event_type      TEXT        NOT NULL,
    subject_digest  CHAR(64)    NOT NULL,
    prev_hash       CHAR(64)    NOT NULL,
    entry_hash      CHAR(64)    NOT NULL UNIQUE,
    retention_class TEXT        NOT NULL DEFAULT 'financial-3yr'
);

CREATE INDEX audit_ledger_subject_idx ON audit_ledger (subject_digest, entry_id);

entry_id uses GENERATED ALWAYS AS IDENTITY so the application cannot supply or reorder ids, keeping the append order authoritative. The UNIQUE constraint on entry_hash is a cheap second line of defense: two distinct events can never collapse to the same sealed hash. The composite index on (subject_digest, entry_id) is what makes the audit replay in Step 5 an index range scan rather than a full-table read.

Step 2: Revoke UPDATE and DELETE and add a guard trigger #

Do both, not one. REVOKE stops the application role from issuing mutations, and the trigger catches any path that somehow reaches an UPDATE/DELETE — a mis-granted role, a future migration, an ORM cascade — by raising before the row changes. Enforcing this at the database is precisely the internal-control posture 2 CFR §200.303 expects, and it satisfies the modification-and-deletion protection in NIST SP 800-53 AU-9.

-- The application role may append and read; it may never mutate history.
REVOKE UPDATE, DELETE, TRUNCATE ON audit_ledger FROM grant_app;
GRANT INSERT, SELECT ON audit_ledger TO grant_app;

CREATE OR REPLACE FUNCTION deny_ledger_mutation() RETURNS trigger AS $$
BEGIN
    RAISE EXCEPTION 'append_only_violation: % on audit_ledger is forbidden', TG_OP
        USING ERRCODE = 'restrict_violation';
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER audit_ledger_no_mutation
    BEFORE UPDATE OR DELETE ON audit_ledger
    FOR EACH ROW EXECUTE FUNCTION deny_ledger_mutation();

Be honest about the boundary: a database superuser or the table owner can disable the trigger (session_replication_role = replica) and bypass REVOKE. That residual risk is exactly why the hash chain exists — the database guard stops the ordinary application, and the chain in Steps 3–4 makes any privileged bypass detectable after the fact. Defense in depth, not a single wall.

Step 3: Compute the hash chain and insert with bound parameters #

Serialize the entry to a canonical byte string (sorted keys, no incidental whitespace) so the hash is reproducible on any machine, then compute entry_hash = sha256(canonical(entry) || prev_hash). The insert reads the current head row’s entry_hash, uses it as prev_hash, and writes the new row — all inside one transaction guarded by a transaction-scoped advisory lock so two concurrent appenders cannot fork the chain. Every value goes in as a bound parameter; there is no string interpolation anywhere near SQL.

import hashlib
import json
import logging
from dataclasses import dataclass
from datetime import datetime, timezone

from sqlalchemy import Engine, text
from sqlalchemy.exc import IntegrityError

logger = logging.getLogger("grant.audit.ledger")

GENESIS_HASH: str = "0" * 64
LEDGER_LOCK_KEY: int = 0x6C65646772  # stable key for pg_advisory_xact_lock


class LedgerAppendError(RuntimeError):
    """Raised when an audit entry cannot be appended atomically."""


@dataclass(frozen=True)
class AuditEvent:
    actor: str
    event_type: str
    subject_digest: str
    retention_class: str = "financial-3yr"


def _canonical(event: AuditEvent, occurred_at: datetime) -> bytes:
    payload = {
        "occurred_at": occurred_at.astimezone(timezone.utc).isoformat(),
        "actor": event.actor,
        "event_type": event.event_type,
        "subject_digest": event.subject_digest,
        "retention_class": event.retention_class,
    }
    return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")


def _entry_hash(canonical: bytes, prev_hash: str) -> str:
    digest = hashlib.sha256()
    digest.update(canonical)
    digest.update(prev_hash.encode("ascii"))
    return digest.hexdigest()


def append_event(engine: Engine, event: AuditEvent) -> str:
    """Append one sealed audit entry and return its entry_hash."""
    occurred_at = datetime.now(timezone.utc)
    canonical = _canonical(event, occurred_at)
    try:
        with engine.begin() as conn:
            # Serialize appenders so the read-head-then-insert cannot fork the chain.
            conn.execute(text("SELECT pg_advisory_xact_lock(:k)"), {"k": LEDGER_LOCK_KEY})
            prev_hash = conn.execute(
                text("SELECT entry_hash FROM audit_ledger ORDER BY entry_id DESC LIMIT 1")
            ).scalar_one_or_none() or GENESIS_HASH

            entry_hash = _entry_hash(canonical, prev_hash)
            conn.execute(
                text(
                    "INSERT INTO audit_ledger "
                    "(occurred_at, actor, event_type, subject_digest, "
                    " prev_hash, entry_hash, retention_class) "
                    "VALUES (:occurred_at, :actor, :event_type, :subject_digest, "
                    "        :prev_hash, :entry_hash, :retention_class)"
                ),
                {
                    "occurred_at": occurred_at,
                    "actor": event.actor,
                    "event_type": event.event_type,
                    "subject_digest": event.subject_digest,
                    "prev_hash": prev_hash,
                    "entry_hash": entry_hash,
                    "retention_class": event.retention_class,
                },
            )
    except IntegrityError as exc:
        logger.error("ledger append rejected | actor=%s | %s", event.actor, exc.orig)
        raise LedgerAppendError("entry violated a ledger constraint") from exc

    logger.info(
        "appended audit entry | actor=%s event=%s subject=%s entry_hash=%s",
        event.actor, event.event_type, event.subject_digest[:12], entry_hash[:12],
    )
    return entry_hash

The first row ever written seals against GENESIS_HASH (sixty-four zeros), which anchors the chain to a known origin. Serializing to UTC ISO-8601 with sort_keys=True and compact separators is what guarantees verify_chain recomputes byte-identical input in Step 4 — a single stray space would look like tampering.

Step 4: Walk the chain with verify_chain to detect tampering #

verify_chain reads every row in entry_id order and applies two independent checks per row: the linkage check (prev_hash equals the previous row’s entry_hash) catches a deleted or reordered row, and the content check (recomputed entry_hash equals the stored value) catches any edited field. On the first failure it raises with the offending entry_id, so a compromised row is named, not just flagged.

class LedgerIntegrityError(RuntimeError):
    """Raised at the first row where the chain fails to verify."""

    def __init__(self, entry_id: int, reason: str) -> None:
        super().__init__(f"chain broken at entry_id={entry_id}: {reason}")
        self.entry_id = entry_id
        self.reason = reason


def verify_chain(engine: Engine) -> int:
    """Recompute the whole chain; return the count of verified entries or raise."""
    expected_prev = GENESIS_HASH
    checked = 0
    with engine.connect() as conn:
        rows = conn.execute(
            text(
                "SELECT entry_id, occurred_at, actor, event_type, "
                "       subject_digest, prev_hash, entry_hash, retention_class "
                "FROM audit_ledger ORDER BY entry_id ASC"
            )
        ).mappings()

        for row in rows:
            if row["prev_hash"] != expected_prev:
                raise LedgerIntegrityError(
                    row["entry_id"], "prev_hash does not match the preceding entry_hash"
                )
            event = AuditEvent(
                actor=row["actor"],
                event_type=row["event_type"],
                subject_digest=row["subject_digest"],
                retention_class=row["retention_class"],
            )
            recomputed = _entry_hash(_canonical(event, row["occurred_at"]), row["prev_hash"])
            if recomputed != row["entry_hash"]:
                raise LedgerIntegrityError(
                    row["entry_id"], "recomputed entry_hash differs from the stored value"
                )
            expected_prev = row["entry_hash"]
            checked += 1

    logger.info("verify_chain ok | entries=%d head=%s", checked, expected_prev[:12])
    return checked

Run verify_chain on a schedule and after any restore, and persist the returned head hash somewhere the database operator cannot reach — a second account, an object-lock bucket, a printed control report. Comparing today’s head against yesterday’s recorded head is what turns “the chain is internally consistent” into “no one truncated the tail,” closing the superuser-bypass gap noted in Step 2 as NIST SP 800-53 AU-9 intends.

Step 5: Replay the audit history for one subject #

An audit replay answers “everything that ever happened to this artifact, in order.” Filter by subject_digest with a bound parameter and return rows in entry_id order; the composite index from Step 1 makes it a range scan. Verify the whole chain first, then filter — replay reads a slice and trusts it only because verify_chain already proved the slice was not tampered with.

from typing import Any, Mapping


def replay_subject(engine: Engine, subject_digest: str) -> list[Mapping[str, Any]]:
    """Return the ordered audit history for one subject_digest."""
    with engine.connect() as conn:
        rows = conn.execute(
            text(
                "SELECT entry_id, occurred_at, actor, event_type, "
                "       prev_hash, entry_hash, retention_class "
                "FROM audit_ledger "
                "WHERE subject_digest = :subject "
                "ORDER BY entry_id ASC"
            ),
            {"subject": subject_digest},
        ).mappings().all()

    logger.info("replay | subject=%s entries=%d", subject_digest[:12], len(rows))
    return [dict(row) for row in rows]

Because the ledger is append-only, this history is complete by construction: there is no soft-delete flag to interpret and no “current version” to reconcile against prior ones. Every state the subject passed through is a row, and the row it points back to is provably the one before it.

Verification #

Confirm the ledger behaves correctly with four checks:

1. The database refuses mutation. Connect as grant_app and attempt UPDATE audit_ledger SET actor = 'x' and DELETE FROM audit_ledger; assert both raise — the REVOKE denies the privilege and, if privileges were mis-granted, the trigger raises append_only_violation. This proves history is immutable to the application.
2. A clean chain verifies. Append several events, then assert verify_chain(engine) returns the exact number appended without raising. This proves linkage and content hashes are computed consistently across insert and verify.
3. Tampering is detected. Simulate a privileged bypass (disable the trigger as owner, edit one actor value, re-enable) and assert verify_chain raises LedgerIntegrityError naming that entry_id with the content-mismatch reason. Do the same by removing a middle row and assert the next row fails the linkage check.
4. Replay is ordered and complete. Assert replay_subject returns events for one subject in ascending entry_id order and that the last row’s entry_hash matches the value append_event returned.

A compliant sequence emits one structured INFO line per append, one on a successful verify_chain, and an ERROR on a rejected append. Ship those logs to a write-once tier so the operational trail itself meets the three-year retention window of 2 CFR §200.334.

subject = "a" * 64
h1 = append_event(engine, AuditEvent("svc:intake", "artifact.received", subject))
h2 = append_event(engine, AuditEvent("user:cfo", "artifact.approved", subject))

assert verify_chain(engine) == 2
history = replay_subject(engine, subject)
assert [row["event_type"] for row in history] == ["artifact.received", "artifact.approved"]
assert history[-1]["entry_hash"] == h2
assert h1 != h2  # distinct events never share a sealed hash

Common Errors & Fixes #

Related #

• Parent section: Audit Trail & Evidence Automation
• The artifacts this ledger points at, sealed with detached signatures: Cryptographically Signing Grant Compliance Artifacts
• Where event_type and retention_class vocabularies are defined: Compliance Metadata Standards