WORM Retention and SHA-256 Hashing for DVIR Records

A sealed DVIR record has to satisfy two independent proofs during a DOT audit: that it has not been altered since it was written, and that it could not have been deleted before its retention window closed. This guide implements both with a single storage call — a SHA-256 digest over the record’s canonical JSON that lets any reader detect a byte-level change, and an S3 Object Lock retain-until date computed from the retention minimum in 49 CFR § 396.3© so the object physically cannot be overwritten or deleted early. It is the concrete boto3 build behind the sealing step described in Immutable Audit Trail and WORM Retention, and it sits inside the wider Core DVIR Architecture & FMCSA Compliance Mapping pipeline.

The failure this prevents is quiet and expensive: a record that looks fine in the console but whose bytes shifted between write and read — because two “equal” records serialized differently — or a record aged out a month early because a retain-until date was computed from a device clock that was wrong. Get the canonicalization and the clock source right and the rest is a small amount of boto3.

• Python 3.10+ — the code uses X | Y unions and modern typing.
• boto3 1.34+ with credentials whose role can PutObject and GetObject but not PutObjectRetention with BypassGovernanceRetention.
• A bucket created with Object Lock enabled and versioning on — Object Lock cannot be added to an existing bucket.
• hashlib and json (stdlib) — SHA-256 over canonical JSON.
• The canonical SealedDVIREvent field contract from the parent guide (record_id, inspection_ts, sealed_at, defect_category, severity_score, compliance_guardrail, version) and the SHA-256 audit_hash convention used across the pipeline.

Every integrity guarantee rests on one property — two logically-equal records must produce byte-identical JSON. Sort keys, use compact separators, and preserve unicode explicitly. Do not rely on Python dict insertion order or Pydantic field order; make canonicalization independent of both.

import hashlib
import json

from pydantic import BaseModel


def canonical_bytes(model: BaseModel) -> bytes:
    """Deterministic serialization for hashing. Sorted keys make the
    output independent of field/insertion order; ensure_ascii=False
    fixes a single UTF-8 representation for non-ASCII text."""
    payload = json.loads(model.model_dump_json())  # normalize types via JSON round-trip
    return json.dumps(
        payload,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
    ).encode("utf-8")


def compute_audit_hash(model: BaseModel) -> str:
    """SHA-256 over canonical JSON — the pipeline-wide audit_hash."""
    return hashlib.sha256(canonical_bytes(model)).hexdigest()

Retain-until is a legal deadline, so derive it from an authoritative server clock — never from inspection_ts, which comes off a driver’s device and can be wrong by hours or years. Apply the greater of the statutory 3-month floor and any carrier-policy retention, and always return a timezone-aware UTC datetime because S3 requires one.

from datetime import datetime, timedelta, timezone

STATUTORY_MIN = timedelta(days=92)  # ~3 months, 49 CFR 396.3(c); round up, never down


def retain_until(sealed_at: datetime, carrier_policy: timedelta | None = None) -> datetime:
    """Later of the statutory floor and carrier policy. Policy may only
    extend the window, never shorten it below the federal minimum."""
    if sealed_at.tzinfo is None:
        raise ValueError("sealed_at must be timezone-aware")
    window = max(STATUTORY_MIN, carrier_policy or STATUTORY_MIN)
    return (sealed_at.astimezone(timezone.utc) + window)

Set the lock in the same put_object that writes the body. Writing first and locking second leaves a window in which the object exists but is deletable — close it by making the write and the lock a single atomic request.

import boto3

s3 = boto3.client("s3")


def seal_record(bucket: str, model: BaseModel, record_id: str,
                sealed_at: datetime, carrier_policy: timedelta | None = None) -> str:
    body = canonical_bytes(model)
    digest = hashlib.sha256(body).hexdigest()
    s3.put_object(
        Bucket=bucket,
        Key=f"dvir/{record_id}.json",
        Body=body,
        ObjectLockMode="COMPLIANCE",  # not GOVERNANCE — no bypass, even for root
        ObjectLockRetainUntilDate=retain_until(sealed_at, carrier_policy),
        Metadata={"audit-hash": digest},
    )
    return digest

Reading is where the hash earns its keep. Fetch the object, re-hash the returned bytes, and compare against the stored digest. Use hmac.compare_digest for a constant-time comparison so an attacker cannot infer the expected value from timing.

import hmac


def verify_on_read(bucket: str, record_id: str) -> bool:
    resp = s3.get_object(Bucket=bucket, Key=f"dvir/{record_id}.json")
    body = resp["Body"].read()
    stored = resp["Metadata"]["audit-hash"]
    recomputed = hashlib.sha256(body).hexdigest()
    if not hmac.compare_digest(stored, recomputed):
        raise IntegrityError(f"{record_id}: audit_hash mismatch — record altered")
    return True


class IntegrityError(Exception):
    ...

A legal hold is orthogonal to retain_until: it has no expiry and blocks deletion until an authorized custodian releases it. Apply it when a DOT investigation or litigation-hold notice lands, on top of the retention date.

def apply_legal_hold(bucket: str, record_id: str) -> None:
    s3.put_object_legal_hold(
        Bucket=bucket,
        Key=f"dvir/{record_id}.json",
        LegalHold={"Status": "ON"},
    )

Two properties must hold: the digest is stable for logically-equal records regardless of key order, and a locked object cannot be overwritten before retain_until. Test the first as a pure unit; test the second against a bucket (or a moto/localstack mock that honors Object Lock).

import pytest


def test_hash_is_stable_for_logically_equal_records():
    a = SealedDVIREvent(record_id="r1", inspection_ts=TS, sealed_at=TS,
                        defect_category="brakes", severity_score=72,
                        compliance_guardrail="OOS", version=1)
    # same values, constructed in a different order — must hash identically
    b = SealedDVIREvent(version=1, compliance_guardrail="OOS", severity_score=72,
                        defect_category="brakes", sealed_at=TS,
                        inspection_ts=TS, record_id="r1")
    assert compute_audit_hash(a) == compute_audit_hash(b)


def test_object_cannot_be_overwritten_before_retain_until(worm_bucket):
    digest = seal_record(worm_bucket, EVENT, "r2", sealed_at=NOW)
    with pytest.raises(s3.exceptions.ClientError) as exc:
        s3.delete_object(Bucket=worm_bucket, Key="dvir/r2.json")
    assert exc.value.response["Error"]["Code"] in {"AccessDenied", "InvalidRequest"}
    # and the record still verifies clean
    assert verify_on_read(worm_bucket, "r2") is True


def test_retain_until_never_below_statutory_floor():
    ru = retain_until(NOW, carrier_policy=timedelta(days=10))  # policy shorter than floor
    assert ru - NOW >= STATUTORY_MIN

• Canonicalization drift. The single most common integrity bug is a digest computed over a non-canonical form on write and a canonical form on read (or vice versa). Serialize through one function everywhere. Watch floats (72 vs 72.0), trailing-zero decimals, and datetime formatting — pin them by round-tripping through model_dump_json() before hashing so the JSON, not the Python object, is the source of truth.
• Unicode representation. A signature field containing an accented driver name hashes differently if one path emits é and another emits the raw byte sequence. Fix ensure_ascii=False and a single UTF-8 encoding everywhere, and normalize the string form (NFC) before it ever reaches the hasher.
• Clock source for retain-until. Computing retain_until from inspection_ts — a device timestamp — can set the deadline in the past on a misconfigured tablet, aging the record out immediately, or years out, over-retaining. Always derive it from a trusted server clock and validate it is at least the statutory floor ahead of now before the write.
• Legal hold vs retention confusion. Releasing a legal hold does not delete the object, and letting retain_until pass does not remove a legal hold. Both must clear before deletion is possible. Track them independently so an ended investigation does not accidentally leave a record deletable while it is still inside its statutory window.

• Immutable Audit Trail and WORM Retention — the full audit-store design this implements.
• Audit Logging and Tamper-Evident Event Chains — links each sealed record into a verifiable chain.
• Encrypting DVIR Data at Rest and in Transit — protects the sealed bytes from disclosure, which WORM does not.
• JSON Schema Validation for Electronic DVIRs — the validation gate a record clears before it is hashed and sealed.

Back to Immutable Audit Trail and WORM Retention, part of Core DVIR Architecture & FMCSA Compliance Mapping.