# Audit Trail & Evidence Retention

When an auditor asks “prove this flood-risk layer was the exact dataset you published on this date, and that no one quietly edited the record afterward,” a plain log file is worthless — it can be rewritten in seconds and leaves no trace. A tamper-evident audit trail solves this by binding each event to the content it describes with a SHA-256 digest and chaining records so that altering any one of them breaks every hash that follows. This guide, part of Spatial Data Audit Reporting & Compliance Governance, builds an append-only, hash-chained audit store for spatial datasets, adds a retention policy that prevents early deletion, and packages a verifiable evidence bundle an external reviewer can check independently.

# Prerequisites

1. Python 3.11+ in an isolated virtual environment (python -m venv .venv)
2. Standard-library hashlib, sqlite3, and json — the core of the audit store, no install required
3. pandas>=2.1 — optional, for querying and reporting over the audit table
4. Write access to append-only or WORM-capable storage for the chain database
5. A defined retention schedule mapping each record class to a minimum retention period
6. Environment variable AUDIT_DB pointing at the SQLite chain file, e.g. ./audit.db
7. Environment variable EVIDENCE_DIR for exported auditor bundles, e.g. ./evidence
8. A stable dataset-identifier scheme so records reference datasets unambiguously

Install the optional reporting dependency:

pip install "pandas>=2.1"

# Concept & Spec Reference

An audit trail is an append-only sequence of records, each attesting to an event that touched a dataset. Two primitives make it defensible. First, content addressing: the record stores a SHA-256 digest of the dataset bytes, so identity is bound to content rather than to an editable name. Second, the hash chain: each record’s own hash is computed over its fields and the previous record’s hash, so the log forms a Merkle-style chain where any retroactive edit invalidates every following record. This is the same content-hash primitive used to give datasets stable identities in spatial data lineage and provenance tracking, applied here to make the log itself trustworthy.

# Audit record fields

# Canonical serialization rule

The record hash is only reproducible if the bytes hashed are canonical. Serialize the fields with sorted keys and no incidental whitespace before hashing:

import hashlib
import json

def record_hash(fields: dict) -> str:
    """Deterministic SHA-256 over a record's canonical JSON form."""
    canonical = json.dumps(fields, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

# Implementation Walkthrough

# Step 1 — Content-address the dataset

Bind the record to the exact bytes of the dataset version. The rationale: a digest is the only field an attacker cannot forge without also producing a colliding file, so it anchors the entire record.

import hashlib
from pathlib import Path

def sha256_file(path: str | Path, chunk: int = 1 << 20) -> str:
    """Stream a file through SHA-256 without loading it into memory."""
    digest = hashlib.sha256()
    with open(path, "rb") as fh:
        for block in iter(lambda: fh.read(chunk), b""):
            digest.update(block)
    return digest.hexdigest()

# Step 2 — Append a hash-chained record

Read the current tail, compute the new record’s hash over its fields plus the tail’s hash, and insert. The rationale: chaining to the previous hash is what converts a mutable table into a tamper-evident ledger.

import sqlite3
import datetime as dt

GENESIS = "0" * 64

SCHEMA = """
CREATE TABLE IF NOT EXISTS audit_log (
    seq             INTEGER PRIMARY KEY,
    dataset_id      TEXT NOT NULL,
    dataset_sha256  TEXT NOT NULL,
    event           TEXT NOT NULL,
    actor           TEXT NOT NULL,
    recorded_at     TEXT NOT NULL,
    retention_class TEXT NOT NULL,
    prev_hash       TEXT NOT NULL,
    record_hash     TEXT NOT NULL
);
"""

def _tail(conn: sqlite3.Connection) -> tuple[int, str]:
    """Return (next_seq, prev_hash) for the current chain tail."""
    row = conn.execute(
        "SELECT seq, record_hash FROM audit_log ORDER BY seq DESC LIMIT 1"
    ).fetchone()
    if row is None:
        return 0, GENESIS
    return row[0] + 1, row[1]

def append_event(
    conn: sqlite3.Connection,
    dataset_id: str,
    dataset_path: str,
    event: str,
    actor: str,
    retention_class: str = "standard-7y",
) -> dict:
    """Append one tamper-evident record and return it."""
    conn.executescript(SCHEMA)
    seq, prev_hash = _tail(conn)
    fields = {
        "seq": seq,
        "dataset_id": dataset_id,
        "dataset_sha256": sha256_file(dataset_path),
        "event": event,
        "actor": actor,
        "recorded_at": dt.datetime.now(dt.timezone.utc).isoformat(),
        "retention_class": retention_class,
        "prev_hash": prev_hash,
    }
    fields["record_hash"] = record_hash(fields)
    conn.execute(
        "INSERT INTO audit_log (seq, dataset_id, dataset_sha256, event, actor, "
        "recorded_at, retention_class, prev_hash, record_hash) "
        "VALUES (:seq, :dataset_id, :dataset_sha256, :event, :actor, "
        ":recorded_at, :retention_class, :prev_hash, :record_hash)",
        fields,
    )
    conn.commit()
    return fields

# Step 3 — Apply a retention policy

Encode the minimum retention period per class and refuse to prune anything younger. The rationale: early deletion of evidence is itself a compliance failure, so retention must be enforced in code, not left to operator discretion.

import datetime as dt

RETENTION_YEARS = {"standard-7y": 7, "short-3y": 3, "permanent": None}

def is_prunable(record: dict, now: dt.datetime | None = None) -> bool:
    """True only if a record has passed its minimum retention period."""
    years = RETENTION_YEARS.get(record["retention_class"])
    if years is None:  # permanent
        return False
    now = now or dt.datetime.now(dt.timezone.utc)
    recorded = dt.datetime.fromisoformat(record["recorded_at"])
    age_days = (now - recorded).days
    return age_days >= years * 365

# Step 4 — Package evidence for auditors

Export a slice of the chain plus a manifest an external party can verify without your tooling. The rationale: evidence is only useful if a third party can check it independently, so the bundle must be self-describing.

import json
from pathlib import Path

def export_evidence(
    conn: sqlite3.Connection,
    dataset_id: str,
    out_dir: str,
) -> Path:
    """Write a verifiable JSON bundle of all records for one dataset."""
    cols = [
        "seq", "dataset_id", "dataset_sha256", "event", "actor",
        "recorded_at", "retention_class", "prev_hash", "record_hash",
    ]
    rows = conn.execute(
        f"SELECT {', '.join(cols)} FROM audit_log "
        "WHERE dataset_id = ? ORDER BY seq",
        (dataset_id,),
    ).fetchall()
    records = [dict(zip(cols, r)) for r in rows]
    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)
    bundle = out / f"{dataset_id}-evidence.json"
    bundle.write_text(json.dumps({
        "dataset_id": dataset_id,
        "algorithm": "sha256",
        "records": records,
    }, indent=2), encoding="utf-8")
    return bundle

# Validation & CI Integration

# Verify the chain

The single most important operation is chain verification: recompute every record’s hash and confirm each prev_hash matches the prior record. Any break localizes the tampering to a specific seq.

def verify_chain(records: list[dict]) -> list[str]:
    """Return a list of integrity violations; empty means the chain is intact."""
    problems: list[str] = []
    expected_prev = GENESIS
    for rec in sorted(records, key=lambda r: r["seq"]):
        fields = {k: rec[k] for k in rec if k != "record_hash"}
        if record_hash(fields) != rec["record_hash"]:
            problems.append(f"seq {rec['seq']}: record_hash mismatch (record altered)")
        if rec["prev_hash"] != expected_prev:
            problems.append(f"seq {rec['seq']}: prev_hash broken (chain cut)")
        expected_prev = rec["record_hash"]
    return problems

# GitHub Actions verification gate

Run verification on every change so a corrupted or hand-edited audit database fails the build. This fits the wider CI/CD validation and policy enforcement approach for spatial data.

# .github/workflows/audit-verify.yml
name: Audit chain verification
on:
  push:
    paths:
      - 'audit/**'
  pull_request:
    paths:
      - 'audit/**'
jobs:
  verify:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: python audit/verify_chain.py audit/audit.db

# Pytest coverage

import sqlite3
from audit.chain import append_event, verify_chain, export_evidence

def test_intact_chain_verifies(tmp_path):
    data = tmp_path / "layer.gpkg"
    data.write_bytes(b"payload-v1")
    conn = sqlite3.connect(":memory:")
    append_event(conn, "layer", str(data), "publish", "ci")
    data.write_bytes(b"payload-v2")
    append_event(conn, "layer", str(data), "reproject", "ci")
    records = conn.execute("SELECT * FROM audit_log").fetchall()
    cols = [d[0] for d in conn.execute("SELECT * FROM audit_log").description]
    assert verify_chain([dict(zip(cols, r)) for r in records]) == []

def test_tampered_record_is_detected(tmp_path):
    data = tmp_path / "layer.gpkg"
    data.write_bytes(b"payload")
    conn = sqlite3.connect(":memory:")
    append_event(conn, "layer", str(data), "publish", "ci")
    conn.execute("UPDATE audit_log SET actor = 'mallory' WHERE seq = 0")
    cols = [d[0] for d in conn.execute("SELECT * FROM audit_log").description]
    rows = conn.execute("SELECT * FROM audit_log").fetchall()
    assert verify_chain([dict(zip(cols, r)) for r in rows]), "tamper not detected"

# Derivative & Lineage Management

The audit trail and the provenance graph are complementary: provenance records how a dataset was made, the audit log records that the claim was made and has not changed since. Whenever a transformation appends a provenance graph, also append an audit record whose dataset_sha256 covers the new derived output and whose event names the operation, so the two stores cross-reference. Hash the serialized provenance file itself and store that digest in the record, making the captured PROV-O lineage tamper-evident too. Surface chain integrity as a first-class signal in your compliance dashboards for spatial catalogs, and when an auditor requests proof, generate the formal narrative through automated compliance report generation backed by the exported evidence bundle.

# Pitfalls & Resolution

# Related

• Spatial Data Audit Reporting & Compliance Governance — the parent guide framing evidence within the audit program
• Hashing Datasets for Tamper-Evident Audit Logs — a self-contained script for the hash-chained append-only log
• Spatial Data Lineage & Provenance Tracking — the provenance graphs this trail cross-references and protects
• Compliance Dashboards for Spatial Catalogs — surface chain integrity as a catalog-wide metric
• Automated Compliance Report Generation — build auditor deliverables from the exported evidence