Writing WORM Audit Logs with Python

This page implements one concrete task inside the broader immutable audit trails guide: a self-contained Python module that appends hash-chained disaster-recovery drill records to an Amazon S3 bucket with Object Lock (WORM) in compliance mode, then verifies the chain by re-hashing every record. Each drill verdict that flows out of the checksum validation pipeline becomes one canonical JSON object, SHA-256 linked to its predecessor and written under a retention date that no principal — root included — can shorten. The module is the evidence writer that compliance evidence and control mapping later queries by drill_id, and its verifier returns strict exit codes so it can gate a job the same way error categorization gates on a classified failure: 0 for a verified chain, 1 for a detected tamper or gap, 2 for a usage error.

Architecture and Execution Model

Figure. A record is canonicalized, linked by SHA-256 to the prior record, written to S3 under a retention lock, its head persisted, and later re-verified to yield a POSIX exit code.

The module has two entry points sharing one canonicalization routine: an append path that writes a new record, and a verify path that replays the whole chain. Both derive the digest from identical canonical bytes, which is the only way a record written today can be re-hashed and matched years later. S3 Object Lock enforces the physical half of immutability — the retention date makes the object undeletable — while the SHA-256 linkage enforces the logical half, so a deletion, insertion, or edit anywhere in the sequence surfaces on the next verify.

Prerequisites

• Python 3.8+ on the writer host.

• The AWS SDK for Python:

  pip install "boto3>=1.34"
  

• A versioned S3 bucket created with Object Lock enabled. Object Lock can only be turned on at bucket-creation time (or via AWS Support for existing buckets); enabling bucket versioning afterward is not sufficient.

• A default or per-object retention configured in compliance mode, so records cannot be deleted before the retention date by any principal.

• An IAM writer principal granted s3:PutObject and s3:PutObjectRetention but explicitly denied s3:DeleteObject, s3:DeleteObjectVersion, and s3:BypassGovernanceRetention, enforcing write-once at the identity layer as well.

• A separate reader principal for verification holding only s3:GetObject and s3:ListBucket.

Production Implementation

The module keeps a small canonical core so the writer and the verifier can never disagree on which bytes were hashed. Canonicalization sorts keys, strips insignificant whitespace, and fixes UTF-8 encoding; the digest is taken over the linked envelope {"prev_digest": ..., "payload": ...}, and the object key is the digest itself so the store is content-addressed and idempotent.

#!/usr/bin/env python3
"""WORM audit-log writer/verifier for DR-drill records on S3 Object Lock.

Usage:
    python3 worm_audit.py append <bucket> <retention_days> <record.json>
    python3 worm_audit.py verify <bucket>

Exit codes:
    0  chain verified intact (verify) / record appended (append)
    1  tamper or gap detected during verification
    2  usage or configuration error
"""
from __future__ import annotations

import datetime as dt
import hashlib
import json
import sys
from typing import Dict, List, Optional, Tuple

import boto3
from botocore.exceptions import BotoCoreError, ClientError

GENESIS = "0" * 64
HEAD_KEY = "_chain/head.json"      # not lock-protected; the chain itself is the truth


def canonical_bytes(obj: Dict[str, object]) -> bytes:
    """One deterministic byte representation shared by writer and verifier."""
    return json.dumps(
        obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False
    ).encode("utf-8")


def link(prev_digest: str, payload: Dict[str, object]) -> Tuple[str, dict]:
    """Return (digest, envelope) for a payload chained to prev_digest."""
    envelope = {"prev_digest": prev_digest, "payload": payload}
    digest = hashlib.sha256(canonical_bytes(envelope)).hexdigest()
    return digest, {"digest": digest, **envelope}

The append function reads the current head, links the new payload, and PUTs the object with an explicit ObjectLockMode="COMPLIANCE" and a RetainUntilDate. The retention argument is computed as an absolute UTC timestamp because Object Lock stores a fixed date, not a relative window — the clock starts at write time and cannot later be shortened.

def read_head(s3, bucket: str) -> str:
    """Fetch the current chain head digest, or GENESIS on an empty chain."""
    try:
        obj = s3.get_object(Bucket=bucket, Key=HEAD_KEY)
    except ClientError as exc:
        if exc.response["Error"]["Code"] in ("NoSuchKey", "404"):
            return GENESIS
        raise
    return json.loads(obj["Body"].read())["head"]


def append(s3, bucket: str, retention_days: int,
           payload: Dict[str, object]) -> str:
    """Write one chained, lock-protected record; return the new head digest."""
    head = read_head(s3, bucket)
    digest, entry = link(head, payload)
    retain_until = dt.datetime.now(dt.timezone.utc) + dt.timedelta(days=retention_days)
    s3.put_object(
        Bucket=bucket,
        Key=f"records/{digest}.json",
        Body=canonical_bytes(entry),
        ContentType="application/json",
        ObjectLockMode="COMPLIANCE",
        ObjectLockRetainUntilDate=retain_until,
    )
    # Advance the head pointer last, so a crash leaves an orphan, never a gap.
    s3.put_object(Bucket=bucket, Key=HEAD_KEY,
                  Body=json.dumps({"head": digest}).encode("utf-8"))
    return digest

The verifier lists every record, orders them by following prev_digest from genesis, and re-hashes each envelope. It reports the first record whose stored digest does not match its recomputed digest, or the first break in linkage — an inserted, deleted, or reordered record.

def load_records(s3, bucket: str) -> Dict[str, dict]:
    """Load all records keyed by their stored digest."""
    records: Dict[str, dict] = {}
    paginator = s3.get_paginator("list_objects_v2")
    for page in paginator.paginate(Bucket=bucket, Prefix="records/"):
        for item in page.get("Contents", []):
            body = s3.get_object(Bucket=bucket, Key=item["Key"])["Body"].read()
            entry = json.loads(body)
            records[entry["digest"]] = entry
    return records


def verify(s3, bucket: str) -> Tuple[int, Optional[str]]:
    """Walk the chain from genesis; return (exit_code, message)."""
    records = load_records(s3, bucket)
    if not records:
        return 0, "empty chain"
    by_prev: Dict[str, dict] = {r["prev_digest"]: r for r in records.values()}
    ordered: List[dict] = []
    cursor = GENESIS
    while cursor in by_prev:
        entry = by_prev[cursor]
        ordered.append(entry)
        cursor = entry["digest"]
    if len(ordered) != len(records):
        return 1, f"chain gap: {len(records) - len(ordered)} record(s) unreachable"
    for i, entry in enumerate(ordered, start=1):
        envelope = {"prev_digest": entry["prev_digest"], "payload": entry["payload"]}
        recomputed = hashlib.sha256(canonical_bytes(envelope)).hexdigest()
        if recomputed != entry["digest"]:
            return 1, f"TAMPER at record {i}: digest mismatch"
    return 0, f"chain intact: {len(ordered)} records verified"


def main(argv: List[str]) -> int:
    if len(argv) >= 2 and argv[1] == "append" and len(argv) == 5:
        try:
            days = int(argv[3])
            payload = json.loads(open(argv[4], encoding="utf-8").read())
        except (ValueError, OSError, json.JSONDecodeError) as exc:
            print(f"config error: {exc}", file=sys.stderr)
            return 2
        try:
            s3 = boto3.client("s3")
            digest = append(s3, argv[2], days, payload)
        except (BotoCoreError, ClientError) as exc:
            print(f"S3 error: {exc}", file=sys.stderr)
            return 2
        print(f"appended {digest[:12]} retained {days}d")
        return 0
    if len(argv) == 3 and argv[1] == "verify":
        try:
            s3 = boto3.client("s3")
            code, message = verify(s3, argv[2])
        except (BotoCoreError, ClientError) as exc:
            print(f"S3 error: {exc}", file=sys.stderr)
            return 2
        stream = sys.stderr if code else sys.stdout
        print(message, file=stream)
        return code
    print(__doc__, file=sys.stderr)
    return 2


if __name__ == "__main__":
    sys.exit(main(sys.argv))

Step-by-Step Execution Walkthrough

1. Provision the bucket once. Create a versioned S3 bucket with Object Lock enabled and a default retention in compliance mode. Confirm the writer principal is denied delete and bypass permissions.
2. Emit the record. After a drill completes, serialize its outcome (drill id, artifact, manifest hash, verdict, actor, timestamps, phase durations) to a JSON file — this is the payload.
3. Append it. Run python3 worm_audit.py append my-audit-bucket 2555 drill_outcome.json. The retention here is seven years in days; the record is written under a compliance-mode lock and the head pointer advances only after the record lands.
4. Read the exit code. A 0 means the record was chained and locked; a 2 means a usage or S3 error occurred and nothing was written.
5. Verify on demand. Run python3 worm_audit.py verify my-audit-bucket. Exit 0 confirms every record re-hashes and the chain is unbroken; exit 1 names the first tampered or unreachable record.
6. Gate on it. Wire the verify exit code into a scheduled compliance job so an integrity break pages on-call rather than waiting for an annual audit.

Verification and Expected Output

A healthy append followed by a verify prints the following and exits 0:

$ python3 worm_audit.py append my-audit-bucket 2555 drill_outcome.json
appended 9f2bc41ad3e0 retained 2555d
$ python3 worm_audit.py verify my-audit-bucket
chain intact: 3 records verified
$ echo $?
0

If an object were somehow altered or a record removed, the verifier localizes the break and exits 1:

$ python3 worm_audit.py verify my-audit-bucket
chain gap: 1 record(s) unreachable
$ echo $?
1

Success means three conditions held together: the head pointer advanced only after the record object was durably written, every record re-hashed to its stored digest, and each prev_digest resolved to exactly one successor from genesis to head. A non-zero verify exit is the contract the compliance job branches on — treat exit 1 as a security event, not a retryable infrastructure error.

Failure Modes and Troubleshooting

Integration Notes

The module returns clean POSIX exit codes, so any scheduler drives it without glue:

• Airflow — append from a PythonOperator immediately after the validation task, and run verify from a separate daily DAG whose failure routes to the compliance channel; the DAG run history mirrors the audit history.
• Celery — dispatch the append from the task that finalizes a drill so the record is written before the “PASS” is reported anywhere, and let a periodic task run verify on a schedule.
• cron / systemd — schedule verify directly; because it exits 0/1/2 cleanly, an OnFailure= unit can page on a 1 without parsing logs.

Feed the classified verify outcome into your error categorization so a chain break and a transient S3 error escalate as distinct events, and let compliance evidence and control mapping query the verified records by drill_id when assembling control evidence.

Frequently Asked Questions

Related

• Immutable audit trails — the parent guide covering record schema, hash chaining, and WORM design this module implements.
• Compliance evidence and control mapping — how the verified records become control evidence queried by drill id.
• Validation telemetry and metrics — the operational counters emitted alongside each audit write.
• Checksum validation pipelines — the source of the manifest digests each record pins.

This script is one component of the broader Immutable Audit Trails guide.

For authoritative behavior of the underlying APIs, consult the Amazon S3 Object Lock documentation, the Python hashlib reference, and the Boto3 S3 put_object reference.