Building Append-Only Audit Logs in PostgreSQL

When a state primacy agency or an EPA auditor asks how a utility arrived at a particular compliance determination, the only defensible answer is a record that provably has not been touched since it was written. This page is the relational implementation inside the parent Audit Trail & Data Lineage Storage Patterns section: an insert-only PostgreSQL table that captures every determination — a running annual average crossing an MCL, a monitoring gap classification, a public-notification decision — and then physically refuses to let anyone alter or erase it. The technique combines three enforcement layers that reinforce one another: a BEFORE UPDATE OR DELETE trigger that aborts any mutation, a per-row hash chain that makes silent tampering detectable, and least-privilege grants that hand the writing service nothing but INSERT and SELECT. Where the Parquet-on-object-storage sibling leans on immutable object versioning for cold archival, this approach keeps determinations queryable in the same transactional database that produces them, so lineage and live compliance state never drift apart.

Prerequisites & Environment Setup

You need PostgreSQL 14 or newer (identity columns and clock_timestamp() behave consistently there) and the modern psycopg 3 driver. The examples run identically against a managed instance or a container. The role that owns the schema is separate from the role the application connects as — that separation is the whole point, and it mirrors the read-only egress discipline described in Security Boundary Design.

python3 -m venv .venv && source .venv/bin/activate
pip install "psycopg[binary]==3.2.*"
# Optional: run a throwaway Postgres for local testing
docker run --rm -d --name audit-pg -e POSTGRES_PASSWORD=devpass \
  -p 5432:5432 postgres:16

Set the connection string for the owner role once so the DDL step can create objects, and provision a distinct low-privilege login (auditor_writer) that the ingestion service will actually use:

export AUDIT_ADMIN_DSN="postgresql://postgres:devpass@localhost:5432/postgres"
export AUDIT_WRITER_DSN="postgresql://auditor_writer:writerpass@localhost:5432/postgres"

Step-by-Step Implementation

Step 1 — Define the append-only table and immutability trigger

The schema pins one determination per row. id is GENERATED ALWAYS AS IDENTITY, so even a client that tries to supply its own key is rejected — sequence values are assigned by the server and cannot be forged. recorded_at uses clock_timestamp() rather than now() so that rows inserted inside a single long transaction still receive distinct wall-clock instants. The two BYTEA columns, prev_hash and row_hash, carry the chain that Step 3 populates.

The immutability guarantee lives in a plpgsql function wired to a BEFORE UPDATE OR DELETE trigger. Any attempt to mutate an existing row raises an exception, which aborts the surrounding transaction. This is stricter than a revoked UPDATE grant alone, because it also stops the table owner and superuser paths that grants do not cover.

import os
import psycopg

DDL = """
CREATE TABLE IF NOT EXISTS compliance_audit_log (
    id             BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    recorded_at    TIMESTAMPTZ  NOT NULL DEFAULT clock_timestamp(),
    facility_id    TEXT         NOT NULL,
    parameter      TEXT         NOT NULL,
    determination  TEXT         NOT NULL,
    payload        JSONB        NOT NULL,
    prev_hash      BYTEA,
    row_hash       BYTEA        NOT NULL,
    CONSTRAINT row_hash_len CHECK (octet_length(row_hash) = 32)
);

CREATE OR REPLACE FUNCTION reject_audit_mutation()
RETURNS TRIGGER LANGUAGE plpgsql AS $fn$
BEGIN
    RAISE EXCEPTION
        'compliance_audit_log is append-only: % is not permitted', TG_OP
        USING ERRCODE = 'restrict_violation';
END;
$fn$;

DROP TRIGGER IF EXISTS trg_no_mutation ON compliance_audit_log;
CREATE TRIGGER trg_no_mutation
    BEFORE UPDATE OR DELETE ON compliance_audit_log
    FOR EACH ROW EXECUTE FUNCTION reject_audit_mutation();
"""


def apply_schema(admin_dsn: str) -> None:
    with psycopg.connect(admin_dsn, autocommit=True) as conn:
        conn.execute(DDL)


if __name__ == "__main__":
    apply_schema(os.environ["AUDIT_ADMIN_DSN"])

Step 2 — Lock down privileges to INSERT and SELECT

A trigger stops accidental mutation; grants stop the application from ever asking. Revoke the default public privileges, then hand auditor_writer exactly two verbs. Because id is GENERATED ALWAYS, the writer needs no sequence grant, and withholding UPDATE/DELETE/TRUNCATE means the ingestion credential is structurally incapable of rewriting history even if it is compromised.

GRANTS = """
REVOKE ALL ON compliance_audit_log FROM PUBLIC;
GRANT INSERT, SELECT ON compliance_audit_log TO auditor_writer;
"""


def apply_grants(admin_dsn: str, writer_role: str = "auditor_writer",
                 writer_password: str = "writerpass") -> None:
    with psycopg.connect(admin_dsn, autocommit=True) as conn:
        conn.execute(
            "DO $$ BEGIN "
            "  IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname = %s) THEN "
            "    EXECUTE format('CREATE ROLE %I LOGIN PASSWORD %L', %s, %s); "
            "  END IF; "
            "END $$;",
            (writer_role, writer_role, writer_password),
        )
        conn.execute(GRANTS)

Keep TRUNCATE in mind separately: it is not a row-level operation, so the FOR EACH ROW trigger never fires for it. Never granting TRUNCATE is what closes that door, which is why the grant list above is an allowlist rather than a set of revocations.

Step 3 — Chain each record to its predecessor

A trigger and grants protect the row in this database. The hash chain protects the sequence of rows against anyone who bypasses the database entirely — a restored-from-backup edit, a storage-layer tamper, an out-of-band psql session run as superuser. Each record stores the SHA-256 digest of its own canonical content concatenated with the previous record’s digest, so the chain is defined by

hi=SHA256(hi−1∥ci),h0=032h_i = \mathrm{SHA256}\bigl(h_{i-1} \parallel c_i\bigr), \qquad h_0 = \mathbf{0}_{32}

where cic_i is the canonical serialization of row ii and ∥\parallel is byte concatenation. Altering any earlier cic_i changes every hjh_j for j≥ij \ge i, so a single edit invalidates the entire tail — the same tamper-evidence property the cryptographic-checksum sibling applies to individual records, extended here across the whole ledger.

Append-only audit records linked by a SHA-256 previous-hash chain Three record boxes left to right. A genesis zero value feeds the first record's previous-hash input; each record outputs a row hash that becomes the next record's previous-hash input, so any change to an earlier record invalidates all later hashes. h0 = 0 Record 1 Record 2 Record 3 payload + prev_hash payload + prev_hash payload + prev_hash h1 h2

The writer computes the digest in Python and inserts it atomically. Reading the current tail and inserting the new head must share one transaction with SERIALIZABLE isolation, otherwise two concurrent writers could both chain onto the same predecessor and fork the ledger.

import hashlib
import json
from typing import Any


def canonical_bytes(facility_id: str, parameter: str,
                    determination: str, payload: dict[str, Any]) -> bytes:
    """Deterministic serialization used as the hash input."""
    body = {
        "facility_id": facility_id,
        "parameter": parameter,
        "determination": determination,
        "payload": payload,
    }
    return json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")


GENESIS = bytes(32)  # 32 zero bytes anchor the first record


def append_determination(writer_dsn: str, facility_id: str, parameter: str,
                         determination: str, payload: dict[str, Any]) -> bytes:
    content = canonical_bytes(facility_id, parameter, determination, payload)
    with psycopg.connect(writer_dsn, autocommit=False) as conn:
        conn.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")
        prev = conn.execute(
            "SELECT row_hash FROM compliance_audit_log "
            "ORDER BY id DESC LIMIT 1"
        ).fetchone()
        prev_hash = prev[0] if prev else GENESIS
        row_hash = hashlib.sha256(prev_hash + content).digest()
        conn.execute(
            "INSERT INTO compliance_audit_log "
            "(facility_id, parameter, determination, payload, prev_hash, row_hash) "
            "VALUES (%s, %s, %s, %s, %s, %s)",
            (facility_id, parameter, determination,
             psycopg.types.json.Jsonb(payload), prev_hash, row_hash),
        )
        conn.commit()
    return row_hash

Step 4 — Query a retention window

Auditors rarely want the whole table; they want every determination for a facility over a defined period — a quarter, a monitoring cycle, the statutory retention window. Because rows are never updated, a plain range scan on recorded_at is always a faithful point-in-time reconstruction. Add a composite index so the window query stays cheap as the ledger grows.

def create_window_index(admin_dsn: str) -> None:
    with psycopg.connect(admin_dsn, autocommit=True) as conn:
        conn.execute(
            "CREATE INDEX IF NOT EXISTS ix_audit_facility_time "
            "ON compliance_audit_log (facility_id, recorded_at)"
        )


def determinations_in_window(reader_dsn: str, facility_id: str,
                             days: int) -> list[tuple]:
    with psycopg.connect(reader_dsn) as conn:
        rows = conn.execute(
            "SELECT id, recorded_at, parameter, determination, payload "
            "FROM compliance_audit_log "
            "WHERE facility_id = %s "
            "  AND recorded_at >= now() - make_interval(days => %s) "
            "ORDER BY recorded_at",
            (facility_id, days),
        ).fetchall()
    return rows

Configuration Reference

The table below is the contract between the schema, the trigger, and the grant policy. Treat the column widths and the grant set as versioned configuration, not as values to be edited casually against a live compliance table.

Column / object Type Purpose Notes
id BIGINT identity Monotonic row key GENERATED ALWAYS; client cannot supply it
recorded_at TIMESTAMPTZ Server-assigned write instant Uses clock_timestamp(), distinct within a txn
facility_id TEXT Public water system / facility identifier Indexed with recorded_at
parameter TEXT Contaminant or metric evaluated e.g. turbidity, TTHM
determination TEXT The decision recorded e.g. MCL_EXCEEDANCE, NO_VIOLATION
payload JSONB Full inputs and computed values Serialized into the hash
prev_hash BYTEA(32) Predecessor’s row_hash GENESIS for the first row
row_hash BYTEA(32) SHA-256 of prev_hash || content CHECK enforces 32-byte length
trg_no_mutation trigger Blocks UPDATE/DELETE BEFORE ... FOR EACH ROW, raises exception
grant to auditor_writer privilege Writer capability INSERT, SELECT only — no UPDATE/DELETE/TRUNCATE

Verification & Testing

Prove immutability directly: insert a row as the writer, then confirm both that the trigger rejects an update and that the chain links correctly. The test below uses the writer DSN so it exercises the real privilege set rather than an owner shortcut.

import pytest
import psycopg


def test_update_and_delete_are_rejected(writer_dsn):
    append_determination(writer_dsn, "PWS-01", "turbidity",
                         "NO_VIOLATION", {"ntu": 0.12})
    with psycopg.connect(writer_dsn) as conn:
        with pytest.raises(psycopg.errors.RaiseException):
            conn.execute(
                "UPDATE compliance_audit_log SET determination = 'x' "
                "WHERE id = (SELECT max(id) FROM compliance_audit_log)"
            )
        conn.rollback()
        with pytest.raises(psycopg.errors.InsufficientPrivilege):
            conn.execute("DELETE FROM compliance_audit_log")
        conn.rollback()


def test_chain_links_to_predecessor(writer_dsn):
    h1 = append_determination(writer_dsn, "PWS-02", "TTHM",
                              "MCL_EXCEEDANCE", {"raa_mg_l": 0.082})
    h2 = append_determination(writer_dsn, "PWS-02", "TTHM",
                              "NO_VIOLATION", {"raa_mg_l": 0.061})
    with psycopg.connect(writer_dsn) as conn:
        prev = conn.execute(
            "SELECT prev_hash FROM compliance_audit_log "
            "ORDER BY id DESC LIMIT 1"
        ).fetchone()[0]
    assert bytes(prev) == h1 and h2 != h1

Acceptance criteria before this table backs any regulatory report:

Troubleshooting & Gotchas

  • The owner can still delete rows. Table-level triggers fire for the owner, but a superuser can ALTER TABLE ... DISABLE TRIGGER. Never let the application role own the table, keep superuser off the ingestion path per Security Boundary Design, and rely on the hash chain plus off-box digest anchoring to detect anything that slips past the trigger.
  • TRUNCATE wipes the table without firing the trigger. FOR EACH ROW triggers never see TRUNCATE. The defense is purely the grant list — do not grant TRUNCATE, and audit pg_class privileges periodically. A statement-level BEFORE TRUNCATE trigger adds belt-and-suspenders coverage.
  • Concurrent writers fork the chain. Two transactions reading the same tail under default READ COMMITTED isolation both chain onto it, producing two rows with identical prev_hash. Use SERIALIZABLE as shown and retry on serialization failure, or serialize writes through a single worker as the Parquet sibling does for its object writes.
  • Non-canonical JSON breaks verification. If the payload is serialized with different key ordering or whitespace at write time versus verify time, the recomputed digest will not match. Always hash the sort_keys=True, tight-separator form, and store exactly what you hashed.
  • Logical-replication or backup edits go unnoticed. A restore that quietly rewrites a row leaves the trigger and grants untouched but breaks the chain. Periodically re-walk the full chain and compare the final row_hash against a value anchored in a separate system, using the record-level checks in verifying record integrity with cryptographic checksums.

Related pages