Geospatial Compliance Metadata Automation Hub

Hashing Datasets for Tamper-Evident Audit Logs

To build a tamper-evident audit log, compute a SHA-256 digest of each dataset and store it in an append-only entry whose own hash is derived from those fields plus the previous entry’s hash — chaining the entries so that altering any past record breaks every hash after it and the change becomes provable.

This is the concrete script behind Audit Trail & Evidence Retention, which covers retention policy and evidence packaging around this core, and it sits within the Spatial Data Audit Reporting & Compliance Governance program. The focus here is a single, dependency-free implementation of the hash chain and its verifier.

Automated Python Implementation

The script uses only the Python standard library — hashlib, json, sqlite3, and datetime — so it runs anywhere with no installs. It appends an entry for a dataset, then demonstrates verification on the resulting chain.

#!/usr/bin/env python3
"""Hash-chained, append-only audit log for spatial datasets.

Usage:
    python audit_log.py append --db audit.db --dataset roads.gpkg --event publish
    python audit_log.py verify --db audit.db
"""
from __future__ import annotations

import argparse
import datetime as dt
import hashlib
import json
import sqlite3
from pathlib import Path

GENESIS = "0" * 64

SCHEMA = """
CREATE TABLE IF NOT EXISTS audit_entries (
    seq            INTEGER PRIMARY KEY,
    dataset        TEXT NOT NULL,
    dataset_sha256 TEXT NOT NULL,
    event          TEXT NOT NULL,
    recorded_at    TEXT NOT NULL,
    prev_hash      TEXT NOT NULL,
    entry_hash     TEXT NOT NULL
);
"""

HASHED_FIELDS = ("seq", "dataset", "dataset_sha256", "event", "recorded_at", "prev_hash")


def sha256_file(path: Path, chunk: int = 1 << 20) -> str:
    """SHA-256 of a file, streamed so large rasters do not exhaust memory."""
    digest = hashlib.sha256()
    with path.open("rb") as fh:  # binary mode is essential for stable digests
        for block in iter(lambda: fh.read(chunk), b""):
            digest.update(block)
    return digest.hexdigest()


def entry_hash(entry: dict) -> str:
    """Deterministic SHA-256 over the canonical form of the hashed fields."""
    payload = {k: entry[k] for k in HASHED_FIELDS}
    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def append(db: str, dataset: str, event: str) -> dict:
    """Append one hash-chained entry for a dataset and return it."""
    conn = sqlite3.connect(db)
    try:
        conn.executescript(SCHEMA)
        tail = conn.execute(
            "SELECT seq, entry_hash FROM audit_entries ORDER BY seq DESC LIMIT 1"
        ).fetchone()
        seq, prev_hash = (0, GENESIS) if tail is None else (tail[0] + 1, tail[1])
        entry = {
            "seq": seq,
            "dataset": dataset,
            "dataset_sha256": sha256_file(Path(dataset)),
            "event": event,
            "recorded_at": dt.datetime.now(dt.timezone.utc).isoformat(),
            "prev_hash": prev_hash,
        }
        entry["entry_hash"] = entry_hash(entry)
        conn.execute(
            "INSERT INTO audit_entries (seq, dataset, dataset_sha256, event, "
            "recorded_at, prev_hash, entry_hash) VALUES "
            "(:seq, :dataset, :dataset_sha256, :event, :recorded_at, "
            ":prev_hash, :entry_hash)",
            entry,
        )
        conn.commit()
        return entry
    finally:
        conn.close()


def verify(db: str) -> list[str]:
    """Recompute hashes and links; return violations (empty means intact)."""
    conn = sqlite3.connect(db)
    try:
        conn.row_factory = sqlite3.Row
        rows = conn.execute(
            "SELECT * FROM audit_entries ORDER BY seq"
        ).fetchall()
    finally:
        conn.close()

    problems: list[str] = []
    expected_prev = GENESIS
    for row in rows:
        entry = dict(row)
        if entry_hash(entry) != entry["entry_hash"]:
            problems.append(f"seq {entry['seq']}: entry_hash mismatch — record altered")
        if entry["prev_hash"] != expected_prev:
            problems.append(f"seq {entry['seq']}: prev_hash broken — chain cut here")
        expected_prev = entry["entry_hash"]
    return problems


def main() -> None:
    parser = argparse.ArgumentParser(description="Hash-chained audit log.")
    sub = parser.add_subparsers(dest="cmd", required=True)

    ap = sub.add_parser("append")
    ap.add_argument("--db", required=True)
    ap.add_argument("--dataset", required=True)
    ap.add_argument("--event", default="publish")

    vp = sub.add_parser("verify")
    vp.add_argument("--db", required=True)

    args = parser.parse_args()
    if args.cmd == "append":
        entry = append(args.db, args.dataset, args.event)
        print(f"seq {entry['seq']} appended: {entry['entry_hash'][:16]}…")
    elif args.cmd == "verify":
        violations = verify(args.db)
        if violations:
            for msg in violations:
                print(f"BROKEN: {msg}")
            raise SystemExit(1)
        print("Chain intact — all entries verified.")


if __name__ == "__main__":
    main()

Validation and pipeline integration

Append a couple of entries and verify the chain end to end:

printf 'roads-v1' > roads.gpkg
python audit_log.py append --db audit.db --dataset roads.gpkg --event publish

printf 'roads-v2' > roads.gpkg
python audit_log.py append --db audit.db --dataset roads.gpkg --event reproject

python audit_log.py verify --db audit.db   # -> Chain intact

Now prove that tampering is caught. Editing any past row makes verification exit non-zero:

sqlite3 audit.db "UPDATE audit_entries SET event='deleted' WHERE seq=0;"
python audit_log.py verify --db audit.db   # -> BROKEN: seq 0 ... ; exit 1

Lock the behaviour into a pytest so a refactor cannot weaken the guarantee:

from audit_log import append, verify, sha256_file
from pathlib import Path
import sqlite3

def test_chain_verifies_then_breaks_on_edit(tmp_path):
    db = str(tmp_path / "audit.db")
    data = tmp_path / "layer.gpkg"
    data.write_bytes(b"v1")
    append(db, str(data), "publish")
    data.write_bytes(b"v2")
    append(db, str(data), "reproject")
    assert verify(db) == []

    conn = sqlite3.connect(db)
    conn.execute("UPDATE audit_entries SET event='x' WHERE seq=0")
    conn.commit()
    conn.close()
    assert verify(db), "tamper should be detected"

Run verification in CI so a corrupted audit database blocks the merge, consistent with the wider CI/CD validation and policy enforcement for spatial data practice:

# .github/workflows/audit-log.yml
name: Audit log verification
on:
  push:
    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/audit_log.py verify --db audit/audit.db

Long-term compliance best practices

  • Hash in binary mode, always. Opening a dataset as text can rewrite line endings and change the digest; open(path, "rb") keeps the SHA-256 stable across platforms.
  • Serialize canonically before hashing. Sorted keys and separators=(",", ":") guarantee the same entry produces the same hash on every machine, so verification travels with the bundle.
  • Use a fixed genesis sentinel. A 64-zero prev_hash for the first entry makes the start of the chain explicit and verifiable rather than an empty-string special case.
  • Serialize appends through a single writer. Concurrent appends that read the same tail create duplicate seq values and fork the chain; guard with a transaction or a queue.
  • Anchor the tail externally. Periodically publish the latest entry_hash somewhere append-only — a signed release note or a separate ledger — so even a wholesale rewrite of the local database is detectable.
  • Store the dataset digest, not the file, in the log. The chain proves which bytes existed; keep the heavy artifacts in your evidence store and let the hash link the two.