Audit Log Schema Contracts

An audit log is only as trustworthy as the weakest record it contains, and the fastest way to lose an auditor’s confidence is to hand over an event stream where some rows are missing an actor, carry a stringly-typed timestamp, or smuggle an undeclared field that nobody can explain. Within the Schema Validation for Compliance & Audit Readiness framework, this guide builds the structural half of a defensible audit trail: a $jsonSchema validator attached to an append-only event collection that guarantees every persisted record carries the actor, action, resource, timestamp, request_id, and tamper-evidence fields an assessor expects — and, through additionalProperties: false, that no record can omit a mandatory field or slip an unexpected one past the write path. The scope here is the collection contract and the writer that feeds it; the honest caveat that a validator constrains shape but cannot by itself forbid an in-place edit is developed in the companion Enforcing Immutable Audit Fields with Validators page, which pairs this schema with the least-privilege role grant that actually makes the collection append-only. The deliverable is a versioned, idempotently deployable validator plus a production pymongo writer that constructs a conforming, hash-chained audit event and rejects a malformed one before it ever reaches durable storage.

Architectural Context & Enforcement Boundaries

A schema validator and an append-only guarantee solve two different problems, and conflating them is the most common design error in audit-log projects. The validator is a structural gate: on every insert, MongoDB evaluates the incoming BSON against $jsonSchema and rejects any document that lacks a required field, carries a wrong bsonType, uses an unregistered action, or contains a property the contract never declared. What the validator cannot do is stop someone with update privileges from later editing a field on a record that already exists — a validator inspects the resulting document, not the delta, so an $set that rewrites actor to a different valid string sails through. Append-only is therefore an authorization property, enforced by denying update and delete privileges on the collection, not a schema property. The schema is the necessary structural half; the role grant is the other half.

Because the validator lives in the collection’s options.validator metadata, it applies to every writer regardless of driver, service, or an ad-hoc mongosh session — there is no code path that appends an event while skipping the contract, short of an aggregation $out/$merge into the collection, which bypasses validators entirely and must be forbidden by policy on an audit target. This non-bypassable property is exactly what makes a collection-level validator suitable as compliance evidence: an assessor can read the registered schema and know that every historical insert was tested against it. Tightening the contract with additionalProperties: false and an enum-bounded action also closes the field-injection surface described in Security Boundaries in Schema Design, so an attacker who reaches the write path cannot annotate audit records with fields your export tooling does not expect. The validation telemetry this contract emits — how many events were rejected, on which keyword — becomes an auditable control signal in its own right, which is the subject of SOC 2 Evidence from Validation Telemetry.

Prerequisites & Operational Requirements

Confirm each item before attaching a validator to a collection you intend to treat as an evidence source.

• MongoDB version: 5.0 or later. The structured errInfo.details.schemaRulesNotSatisfied object that pinpoints the failing keyword — indispensable when a rejected audit event needs to be traced back to a producer bug — is only emitted from 5.0 onward. On 4.x the rejection reason is an opaque string.
• Topology: a replica set with w: "majority" write concern. An audit record that is acknowledged by the primary but lost to a rollback after a failover is worse than no record; majority acknowledgement is the durability floor for evidence.
• Driver: PyMongo 4.x (pip install "pymongo>=4.6,<5"). Pin it in the writer’s deployment image so WriteError.code, errInfo, and BSON datetime handling stay stable across builds.
• Time source: all hosts running the writer must be NTP-synchronised. The timestamp field is a date generated by the application, and clock skew between producers directly corrupts event ordering — see the edge cases below.
• Privileges — two principals. The deploy principal needs the collMod and createCollection actions (the dbAdmin role on the database). The runtime writer principal needs only insert and find on the audit collection and must be denied update and remove; provisioning that least-privilege role is covered in Enforcing Immutable Audit Fields with Validators and is what converts the schema into an actual append-only guarantee.

Idempotent Deployment / Implementation Workflow

Creating an audit collection with its contract is a deterministic sequence. Each step is independently verifiable and safe to re-run, so the same script can run on every CI/CD deploy without churning metadata or acquiring a needless lock.

1. Define the contract as a versioned artifact. Keep the schema in source control next to the writer so the rule and the code that satisfies it evolve together. The audit contract:

   const auditSchema = {
     bsonType: "object",
     required: ["actor", "action", "resource", "timestamp", "request_id", "prev_hash", "hash"],
     properties: {
       actor:      { bsonType: "string", minLength: 1 },
       action:     { enum: ["create", "read", "update", "delete", "login", "logout", "export"] },
       resource:   { bsonType: "string", minLength: 1 },
       timestamp:  { bsonType: "date" },
       request_id: { bsonType: "string", pattern: "^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" },
       prev_hash:  { bsonType: "string", pattern: "^[0-9a-f]{64}$" },
       hash:       { bsonType: "string", pattern: "^[0-9a-f]{64}$" },
       metadata:   { bsonType: "object" }
     },
     additionalProperties: false
   };

2. Create the collection with the validator attached. Attach the contract at creation time in strict + error mode — an audit collection is greenfield by definition, so there is no legacy backlog to grandfather, and it should reject non-conforming events from its first write:

   db.createCollection("audit_events", {
     validator: { $jsonSchema: auditSchema },
     validationLevel: "strict",
     validationAction: "error"
   })
   // Expected output: { ok: 1 }

3. Create the ordering index. Evidence export reads events in chronological order, and the hash chain is verified in insertion order; back both with a compound index so neither is a collection scan:

   db.audit_events.createIndex({ timestamp: 1, _id: 1 })
   // Expected output: "timestamp_1__id_1"

4. Reconcile idempotently on redeploy. On subsequent runs the collection already exists, so compare the registered validator against the target and issue collMod only on divergence. The pymongo deployer in the next section encapsulates this comparison so a repeated pipeline run is a no-op rather than a redundant metadata write.

5. Confirm the live contract. Read the registered validator straight from the catalog to prove the deploy landed, rather than trusting the deploy script’s intent:

   db.getCollectionInfos({ name: "audit_events" })[0].options.validator
   // => { $jsonSchema: { bsonType: "object", required: [...], ... } }

Production-Ready Automation Implementation

The following module is the runtime audit writer. It reconciles the validator idempotently on startup, then constructs a fully conforming audit record — including a timezone-aware UTC timestamp, a UUID request_id, and a prev_hash/hash chain link computed over the canonical serialization of the record — and inserts it. Because the writer computes the hash from the same fields the schema constrains, a record that the validator would reject can never be silently mis-hashed into the chain; the two controls reinforce each other.

import hashlib
import json
import logging
import uuid
from datetime import datetime, timezone
from typing import Any, Optional

from pymongo import ASCENDING, MongoClient
from pymongo.collection import Collection
from pymongo.errors import WriteError

logger = logging.getLogger("audit-writer")

GENESIS_HASH = "0" * 64  # prev_hash of the very first event in the chain

AUDIT_SCHEMA: dict[str, Any] = {
    "bsonType": "object",
    "required": ["actor", "action", "resource", "timestamp", "request_id", "prev_hash", "hash"],
    "properties": {
        "actor": {"bsonType": "string", "minLength": 1},
        "action": {"enum": ["create", "read", "update", "delete", "login", "logout", "export"]},
        "resource": {"bsonType": "string", "minLength": 1},
        "timestamp": {"bsonType": "date"},
        "request_id": {"bsonType": "string",
                       "pattern": "^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"},
        "prev_hash": {"bsonType": "string", "pattern": "^[0-9a-f]{64}$"},
        "hash": {"bsonType": "string", "pattern": "^[0-9a-f]{64}$"},
        "metadata": {"bsonType": "object"},
    },
    "additionalProperties": False,
}


class AuditLogWriter:
    """Append tamper-evident, schema-validated audit events to an insert-only collection."""

    def __init__(self, collection: Collection):
        self.collection = collection

    def ensure_contract(self) -> None:
        """Idempotently attach the validator and ordering index (deploy principal)."""
        db = self.collection.database
        name = self.collection.name
        desired = {"$jsonSchema": AUDIT_SCHEMA}
        info = db.command("listCollections", filter={"name": name})["cursor"]["firstBatch"]
        if not info:
            db.create_collection(name, validator=desired,
                                 validationLevel="strict", validationAction="error")
            logger.info("created audit collection %s with validator", name)
        elif info[0].get("options", {}).get("validator") != desired:
            db.command("collMod", name, validator=desired,
                       validationLevel="strict", validationAction="error")
            logger.info("reconciled validator on %s", name)
        self.collection.create_index([("timestamp", ASCENDING), ("_id", ASCENDING)])

    def _tip_hash(self) -> str:
        """Return the hash of the most recent event, or the genesis hash if empty."""
        tip = self.collection.find_one(sort=[("timestamp", -1), ("_id", -1)], projection={"hash": 1})
        return tip["hash"] if tip else GENESIS_HASH

    @staticmethod
    def _digest(record: dict[str, Any]) -> str:
        """Deterministic SHA-256 over the record's evidentiary fields (excluding 'hash')."""
        payload = {
            "actor": record["actor"],
            "action": record["action"],
            "resource": record["resource"],
            "timestamp": record["timestamp"].isoformat(),
            "request_id": record["request_id"],
            "prev_hash": record["prev_hash"],
            "metadata": record.get("metadata", {}),
        }
        canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

    def record(self, actor: str, action: str, resource: str,
               request_id: Optional[str] = None, metadata: Optional[dict] = None) -> str:
        """Construct a conforming, hash-chained audit event and append it. Returns its hash."""
        event = {
            "actor": actor,
            "action": action,
            "resource": resource,
            # timezone-aware UTC; pymongo stores it as a BSON date the schema requires.
            "timestamp": datetime.now(timezone.utc),
            "request_id": request_id or str(uuid.uuid4()),
            "prev_hash": self._tip_hash(),
            "metadata": metadata or {},
        }
        event["hash"] = self._digest(event)
        try:
            self.collection.insert_one(event)
        except WriteError as exc:
            if exc.code == 121:
                rules = (exc.details or {}).get("errInfo", {}).get("details", {})
                # A 121 here is a producer bug, not bad audit data — fail loud.
                logger.error("audit event rejected by validator: %s", rules.get("schemaRulesNotSatisfied"))
            raise
        return event["hash"]


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    client = MongoClient("mongodb://localhost:27017/?replicaSet=rs0&w=majority")
    writer = AuditLogWriter(client.compliance.audit_events)
    writer.ensure_contract()
    h = writer.record(actor="user:8842", action="export", resource="report:q3-financials")
    logger.info("appended audit event %s", h)

The writer never mutates an existing event, so it needs only insert and find at runtime — the ensure_contract step is a separate, privileged deploy concern. That separation is deliberate: the process that appends events in production should be provisioned with a role that cannot update or delete, which is what makes the “append-only” claim defensible rather than aspirational.

Diagnostic Fingerprints & Fast Resolution

A validator on an audit collection changes the meaning of a rejection. On a normal data collection a code 121 is often expected traffic to be quarantined; on an audit collection it is a signal that a producer tried to write a malformed event, which is almost always a logging bug caught at the database — the write path did exactly its job. Treat every 121 on audit_events as a defect to fix in the emitter, not a document to salvage.

Signature	Where it appears	Root cause	Resolution
WriteError code 121, schemaRulesNotSatisfied names required	PyMongo writer at insert_one	Producer omitted a field, e.g. forgot to set actor on a background job	Fix the emitter to populate every required field; do not weaken the schema
121 with failing keyword enum on action	Writer	A new action verb was logged that the contract never registered	Add the verb to the action enum via a reviewed collMod, then redeploy
121 with failing keyword bsonType on timestamp	Writer	Timestamp was passed as an ISO string instead of a datetime	Pass a timezone-aware datetime; pymongo encodes it as a BSON date
121 with failing keyword additionalProperties	Writer	Emitter attached an undeclared field	Remove the field or add it to properties deliberately; treat as a possible tampering signal
OperationFailure code 13 (Unauthorized) on update/remove	Writer or operator	The append-only role correctly denied a mutation	Expected — investigate why a mutation was attempted at all

To pull the precise reason a specific audit event was rejected, catch the driver exception and read the structured detail rather than the summary string:

from pymongo.errors import WriteError

try:
    writer.record(actor="", action="login", resource="console")  # empty actor violates minLength
except WriteError as exc:
    assert exc.code == 121
    print("Failed rule:", exc.details["errInfo"]["details"]["schemaRulesNotSatisfied"])
    # => Failed rule: [{'operatorName': 'properties', 'propertiesNotSatisfied': [...'actor'...]}]

Edge Cases, Gotchas & Known Limitations

• A validator alone does not make a collection append-only. $jsonSchema constrains a document’s shape but cannot forbid an update to an existing record — it validates the resulting document, not the change. Immutability comes from denying update/remove privileges, detailed in Enforcing Immutable Audit Fields with Validators. Do not ship an audit contract without the paired role grant.
• date versus timestamp are different BSON types. Use bsonType: "date" for the human-meaningful event time; it maps to a Python datetime and is what you sort and export by. BSON timestamp is an internal replication type (seconds plus an ordinal counter) intended for oplog use — validating timestamp as a date will reject a BSON timestamp value, which is the correct behaviour. Never store bsonType: "timestamp" as your audit event time.
• Clock skew corrupts ordering, not validation. The schema accepts any valid date, including one from a host whose clock is minutes off. Two events can then persist out of causal order while both passing validation. Keep every writer NTP-synchronised and rely on the prev_hash/hash chain — not the wall-clock timestamp — as the authoritative ordering, because the chain encodes actual insertion sequence.
• Naive versus timezone-aware datetimes. A naive datetime.utcnow() is stored by pymongo as UTC but read back as naive, inviting a downstream comparison bug. Always construct datetime.now(timezone.utc); the schema treats both as a date, so this is a correctness discipline the validator will not catch for you.
• Aggregation $merge/$out bypass the validator. A pipeline that writes into audit_events via $merge skips the contract entirely and can inject non-conforming or unchained records. Forbid these operations on the audit target by policy and by role.
• The enum on action is a change-controlled surface. Widening it is a legitimate schema evolution, but doing it silently defeats the point of a bounded vocabulary. Route every action addition through the same review that governs any breaking schema change.

Verification & Rollback Procedures

Prove the contract is enforcing before you rely on the collection as evidence. The only positive proof is that a known-malformed event is actually rejected:

from pymongo.errors import WriteError

# A conforming event must succeed; a malformed one must raise 121.
good = writer.record(actor="svc:etl", action="create", resource="dataset:events")
assert len(good) == 64  # a hash came back

try:
    writer.collection.insert_one({"actor": "x", "action": "purge", "resource": "all"})
    raise AssertionError("audit validator is NOT enforcing")
except WriteError as exc:
    assert exc.code == 121  # 'purge' not in enum, and required hash fields absent
    print("contract confirmed: malformed audit event rejected with 121.")

You can also verify the tamper-evidence chain independently of the schema by walking the collection in order and recomputing each digest — a break means a record was altered or removed after the fact:

def verify_chain(collection) -> bool:
    prev = "0" * 64
    for ev in collection.find(sort=[("timestamp", 1), ("_id", 1)]):
        if ev["prev_hash"] != prev or AuditLogWriter._digest(ev) != ev["hash"]:
            print("chain break at", ev["_id"])
            return False
        prev = ev["hash"]
    return True

Rollback on an audit collection is deliberately asymmetric. You may loosen enforcement in an emergency — for example if a mislabelled action is blocking a legitimate producer — but you must never delete audit data to “fix” a rejection. The soft loosen keeps the schema attached for telemetry while it stops blocking; restore error the moment the emitter is patched:

// Soft loosen — keep the contract logging, stop blocking a legitimate producer.
db.runCommand({ collMod: "audit_events", validationAction: "warn" })

// Restore enforcement after the producer is fixed.
db.runCommand({ collMod: "audit_events", validationAction: "error" })

Both take effect on the primary immediately and replicate through the oplog, so time-to-recover is the lag of your slowest secondary — typically seconds. A warn window on an audit collection must be short and itself logged, because any event that persists during it is, by definition, unvalidated evidence.

Frequently Asked Questions

Related

• Schema Validation for Compliance & Audit Readiness — the parent reference framing how validation contracts produce audit and compliance evidence.
• Enforcing Immutable Audit Fields with Validators — the least-privilege role grant and hash-chain check that turn this schema into a genuine append-only guarantee.
• SOC 2 Evidence from Validation Telemetry — treating validation rejection counts and keywords as an auditable control signal.
• Security Boundaries in Schema Design — using additionalProperties: false and enums to close the field-injection surface on sensitive collections.