Audit Report Generation for Waste Route Compliance

An audit does not fail because the work was wrong — it fails because the record cannot be reproduced. When a municipality answers a public-records request or an environmental regulator opens a file on a hazardous-waste hauler, the question is not “did you run the route” but “prove this report is exactly what your systems produced, unchanged, on that date.” If the report is a PDF assembled from whatever the database happened to return that afternoon, with floating-point tonnages and dictionary key order that shifts between Python versions, there is no way to demonstrate it was not edited after the fact. The report is only as trustworthy as your ability to recompute it and get the same bytes.

This is the reporting sub-system of the Compliance Reporting & Automation pipeline. It bundles the artifacts that already exist elsewhere in the system — route plans, hours-of-service logs, EPA e-Manifests, and telemetry — into a single signed, reproducible report. The technique is deliberately boring and dependency-light: canonicalize every source record to sorted-key JSON, hash each into a Merkle leaf, chain the leaves into a single root, sign the root with HMAC, and emit a verification manifest that lets an auditor recompute the root from the same inputs and confirm the signature. What breaks without it is the whole point of keeping records: an audit that cannot be reproduced or verified byte-for-byte is not evidence, it is an assertion. Every code block is complete and pasteable into a Python 3.10+ compliance worker.

Prerequisites

Pin only what is needed to validate inputs — the cryptography is entirely standard library. Deliberately keeping a heavy PDF renderer out of the pipeline is a design choice, not a shortcut: a structured JSON report plus a verification manifest is reproducible byte-for-byte, whereas a rendered PDF embeds timestamps and font subsets that change the bytes without changing the content.

python --version   # 3.10 or newer
pip install \
  pydantic==2.9.2 \
  pytest==8.2.0

Hashing (hashlib), signing (hmac), and serialization (json) all come from the standard library. The one rule that governs every record before it is hashed: it must be deterministically serializable. Any field that cannot be canonicalized identically on every machine — most often a float — poisons the whole tree, because a single differing byte changes the leaf hash and cascades to a different root. The AuditReport model captures the report envelope and enforces that discipline:

from __future__ import annotations
from datetime import datetime, timezone
from typing import Any
from pydantic import BaseModel, Field, field_validator


class SourceRecord(BaseModel):
    """One compliance artifact to be sealed into the report."""
    kind: str = Field(min_length=1)          # e.g. "route_plan", "hos_log"
    record_id: str = Field(min_length=1)
    body: dict[str, Any]                      # the artifact itself

    @field_validator("body")
    @classmethod
    def reject_floats_in_body(cls, body: dict[str, Any]) -> dict[str, Any]:
        _assert_no_float(body)
        return body


class AuditReport(BaseModel):
    report_id: str = Field(min_length=1)
    generated_at: datetime
    key_id: str = Field(min_length=1)         # which signing key sealed this
    records: list[SourceRecord] = Field(min_length=1)
    merkle_root: str | None = None            # hex, filled at seal time
    signature: str | None = None              # hex HMAC, filled at seal time

    @field_validator("generated_at")
    @classmethod
    def require_utc(cls, v: datetime) -> datetime:
        if v.tzinfo is None or v.utcoffset() is None:
            raise ValueError("generated_at must be timezone-aware UTC")
        return v


def _assert_no_float(value: Any) -> None:
    """Reject floats anywhere in a to-be-hashed structure (bool is allowed)."""
    if isinstance(value, float):
        raise ValueError("float in a hashed field: represent quantities as int or str")
    if isinstance(value, dict):
        for v in value.values():
            _assert_no_float(v)
    elif isinstance(value, (list, tuple)):
        for v in value:
            _assert_no_float(v)

Quantities that seem to demand a float — a tonnage of 12.4, a latitude of 40.4406 — are carried as integers in a fixed unit (kilograms, micro-degrees) or as strings, exactly as the Capacity & Weight Limits model keeps payload in integer kilograms. This is not fussiness; 0.1 + 0.2 does not equal 0.3 in IEEE-754, and a report that hashes differently depending on how a number was assembled is not reproducible.

Core Implementation

The construction has four deterministic steps: canonicalize each record, hash it into a leaf, chain the leaves into a Merkle root, and sign the root. Each is pure — same input, same output, on any machine.

Canonicalize and hash a leaf

Canonicalization is the load-bearing step. Sorted keys, no insignificant whitespace, and UTF-8 encoding make the serialization of a record identical everywhere. A one-byte domain-separation prefix distinguishes a leaf hash from an internal node hash so no record can be passed off as a subtree — the classic Merkle second-preimage defense.

import hashlib
import hmac
import json

LEAF_PREFIX = b"\x00"
NODE_PREFIX = b"\x01"


def canonical_bytes(record: dict) -> bytes:
    """Deterministic serialization: sorted keys, tight separators, UTF-8."""
    return json.dumps(
        record, sort_keys=True, separators=(",", ":"), ensure_ascii=False
    ).encode("utf-8")


def leaf_hash(record: dict) -> bytes:
    return hashlib.sha256(LEAF_PREFIX + canonical_bytes(record)).digest()


def node_hash(left: bytes, right: bytes) -> bytes:
    return hashlib.sha256(NODE_PREFIX + left + right).digest()

Chain leaves into a Merkle root

Each internal node is the hash of the concatenation of its two children,

$H_{\text{parent}} = \text{SHA-256}\!\left(\texttt{0x01} \,\Vert\, H_{\text{left}} \,\Vert\, H_{\text{right}}\right)$

and the tree has height $\lceil \log_2 n \rceil$ for $n$ leaves, so verifying one record’s inclusion costs only that many hashes rather than re-reading the whole set. When a level has an odd number of nodes, the last node is duplicated to form a pair — a fixed, documented rule so the root is a pure function of the ordered leaves.

def merkle_root(leaves: list[bytes]) -> bytes:
    if not leaves:
        raise ValueError("cannot build a Merkle root over zero records")
    level = list(leaves)
    while len(level) > 1:
        if len(level) % 2 == 1:
            level.append(level[-1])            # duplicate the last to pair it
        level = [node_hash(level[i], level[i + 1]) for i in range(0, len(level), 2)]
    return level[0]

Seal the report: root, signature, manifest

Sealing binds the ordered records to a root, signs the root with HMAC under a named key, and emits a verification manifest — the ordered leaf hashes plus the algorithm parameters and the key_id, but never the key itself. An auditor holding the manifest and the source records can recompute the root and, with the key resolved from the key_id, recompute the signature.

def sign_root(root: bytes, key: bytes) -> str:
    return hmac.new(key, root, hashlib.sha256).hexdigest()


def seal_report(report: AuditReport, key: bytes) -> tuple[AuditReport, dict]:
    """Compute the Merkle root, sign it, and return (sealed report, manifest)."""
    leaves = [leaf_hash(r.body) for r in report.records]
    root = merkle_root(leaves)
    report.merkle_root = root.hex()
    report.signature = sign_root(root, key)

    manifest = {
        "report_id": report.report_id,
        "generated_at": report.generated_at.isoformat(),
        "key_id": report.key_id,
        "hash_algorithm": "sha256",
        "leaf_prefix": LEAF_PREFIX.hex(),
        "node_prefix": NODE_PREFIX.hex(),
        "odd_node_rule": "duplicate_last",
        "leaves": [{"kind": r.kind, "record_id": r.record_id, "leaf": leaves[i].hex()}
                   for i, r in enumerate(report.records)],
        "merkle_root": report.merkle_root,
        "signature": report.signature,
    }
    return report, manifest

The records sealed here are the same artifacts other subsystems produce: the route plan from the VRP Route Optimization Algorithms engine, the driving intervals from FMCSA ELD Sync, the hazardous-waste record from EPA e-Manifest Integration, and the duty-status log from DOT Hours-of-Service Logging. The signing key itself never lives in application code; it is resolved through the Security & Access Boundaries layer, which is also where key rotation and the HMAC chaining pattern belong.

Regulatory Mapping

The retention periods and recordkeeping obligations a report satisfies are statutory, so pin each to the rule rather than asserting it in prose. The report’s generated_at, record set, and retention class map directly to these sources:

Report concern	Regulatory source	Obligation
Hazardous-waste generator records	40 CFR 262.40	Retain manifests and related records for at least 3 years
Driver Records of Duty Status	49 CFR 395.8(k)	Carrier retains supporting RODS documents for at least 6 months
e-Manifest recordkeeping	EPA e-Manifest program (40 CFR 262 Subpart B)	Retain the signed electronic manifest and tracking number
Municipal public-records retention	Local records-retention schedule	Produce the requested record intact on demand

The differing retention windows — three years for generator records, six months for RODS supporting documents, longer for many municipal schedules — mean a single report often carries records governed by multiple clocks. Sealing them together under one Merkle root does not merge their retention obligations; it lets you prove that whichever record is later produced under whichever clock is exactly the one that was sealed. The reconciliation of tonnage in the sealed manifest back to the weighbridge and the DOT/FMCSA rule mapping hours-of-service records is what makes the bundle internally consistent, and the overall structure follows the Core Architecture & Compliance Mapping contract for how compliance artifacts are versioned and stored.

Validation & Verification

A tamper-evident report is worthless if you never demonstrate that tampering is detected. The tests below recompute the root from the manifest, confirm a clean report verifies, and confirm that mutating any single record changes the root and invalidates the signature.

KEY = b"unit-test-signing-key-not-for-production"


def _report() -> AuditReport:
    return AuditReport(
        report_id="RPT-2026-07-16-0007",
        generated_at=datetime(2026, 7, 16, 12, 0, tzinfo=timezone.utc),
        key_id="hmac-2026-q3",
        records=[
            SourceRecord(kind="route_plan", record_id="R-88",
                         body={"vehicle": 3, "stops": [11, 12, 13], "payload_kg": 9300}),
            SourceRecord(kind="hos_log", record_id="H-88",
                         body={"driver": "D-4417", "driving_min": 402}),
            SourceRecord(kind="manifest", record_id="M-88",
                         body={"mtn": "0123456789ABC", "waste_codes": ["D001"], "qty_kg": 9300}),
        ],
    )


def recompute_root(manifest: dict) -> str:
    leaves = [bytes.fromhex(leaf["leaf"]) for leaf in manifest["leaves"]]
    return merkle_root(leaves).hex()


def test_sealed_report_verifies() -> None:
    report, manifest = seal_report(_report(), KEY)
    assert recompute_root(manifest) == report.merkle_root
    assert sign_root(bytes.fromhex(report.merkle_root), KEY) == report.signature


def test_tampered_record_breaks_the_root() -> None:
    report, manifest = seal_report(_report(), KEY)
    original_root = report.merkle_root

    # An auditor is handed the records but one quantity was altered after sealing.
    tampered = _report()
    tampered.records[2].body["qty_kg"] = 9301          # 1 kg heavier
    recomputed = merkle_root([leaf_hash(r.body) for r in tampered.records]).hex()

    assert recomputed != original_root                  # tamper changes the root
    assert sign_root(bytes.fromhex(recomputed), KEY) != report.signature


def test_canonicalization_is_key_order_independent() -> None:
    a = leaf_hash({"b": 2, "a": 1})
    b = leaf_hash({"a": 1, "b": 2})
    assert a == b                                        # sorted keys make order irrelevant

In production, emit each seal as a structured record so verification is itself auditable. The JSONFormatter fields here are distinct to report sealing:

import logging


class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "level": record.levelname,
            "event": record.getMessage(),
            "report_id": getattr(record, "report_id", None),
            "record_count": getattr(record, "record_count", None),
            "merkle_root": getattr(record, "merkle_root", None),
            "key_id": getattr(record, "key_id", None),
        }
        return json.dumps(payload, separators=(",", ":"))


logger = logging.getLogger("audit_report")
_handler = logging.StreamHandler()
_handler.setFormatter(JSONFormatter())
logger.addHandler(_handler)
logger.setLevel(logging.INFO)

The merkle_root and key_id fields are what an auditor cross-references: the root proves which record set was sealed, and the key_id names the key needed to re-verify the signature after any rotation.

Failure Modes & Edge Cases

Every failure here produces a report that looks fine and verifies wrong. Guard against each explicitly.

1. Non-deterministic serialization. If two machines serialize the same record to different bytes — unsorted keys, differing whitespace, ASCII escaping — their roots diverge and the report fails to verify even though nothing was tampered with. The canonical_bytes function pins sorted keys, tight separators, and UTF-8; never hash str(record) or a default json.dumps.

2. Floating-point in hashed fields. A float is the sharpest edge: 0.1 + 0.2 is not 0.3, and the same tonnage assembled two ways can hash differently. The _assert_no_float guard rejects floats at model construction so they never reach a leaf.

def safe_seal(report: AuditReport, key: bytes) -> tuple[AuditReport, dict]:
    """Fail loudly before hashing if any record is non-deterministic."""
    for r in report.records:
        try:
            json.dumps(r.body, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
        except TypeError as exc:
            raise ValueError(f"record {r.record_id!r} is not JSON-serializable: {exc}") from exc
        _assert_no_float(r.body)
    return seal_report(report, key)

3. Missing source record. If a record that existed at seal time is absent at verification, the recomputed root will not match — which is correct, but uninformative. The manifest lists each leaf’s kind and record_id, so a verifier can name exactly which record is missing rather than reporting a bare root mismatch.

4. Key rotation. HMAC verification needs the same key that signed the report. Store the key_id in the report and manifest and resolve the actual key through a keyring, so a report signed under last quarter’s key still verifies after the key is rotated. Never hardcode the key or assume a single current key.

def verify_signature(report: AuditReport, manifest: dict, keyring: dict[str, bytes]) -> bool:
    key = keyring.get(report.key_id)
    if key is None:
        raise KeyError(f"no key for key_id {report.key_id!r}; cannot verify")
    recomputed = recompute_root(manifest)
    return recomputed == report.merkle_root and sign_root(bytes.fromhex(recomputed), key) == report.signature

Integration Checklist

Complete every item before an audit report is treated as evidence:

Every source record is JSON-serializable with sorted keys and carries no float in any hashed field.
Quantities are stored as integers in a fixed unit or as strings, never as floats.
Leaf and node hashes use distinct domain-separation prefixes.
The odd-node rule is fixed and recorded in the manifest so the root is a pure function of the ordered leaves.
The root is signed with HMAC-SHA256 under a named key_id, and the key is resolved from a keyring, never hardcoded.
The verification manifest lists every leaf’s kind, record_id, and hash so a mismatch names the offending record.
A CI test recomputes the root from the manifest and asserts a one-field tamper both changes the root and invalidates the signature.
Sealed reports are written to a write-once, append-only store and never mutated in place.
Each seal emits a structured record with merkle_root and key_id.

FAQ

Why a Merkle tree instead of hashing the whole report at once?

A single hash over the concatenated report proves the whole set is intact but cannot prove one record’s inclusion without re-reading everything, and it gives no structure for naming which record changed. A Merkle tree lets you verify any one record against the root with about log-base-2 of n hashes and lets the manifest pinpoint exactly which leaf diverged, which matters when an auditor asks about one manifest out of thousands.

Is HMAC signing as good as a digital signature here?

For a single organization proving its own records were not altered after sealing, HMAC-SHA256 under a rotated, access-controlled key is sufficient and simple. It is symmetric, so a verifier needs the shared key, which is fine when the carrier and the auditor operate under the same keyring governance. If records must be verifiable by an outside party who should never hold the signing key, move to an asymmetric signature over the same Merkle root — the tree construction does not change.

Why forbid floats in hashed fields so strictly?

Because IEEE-754 arithmetic is not associative and its decimal representation is platform- and assembly-order dependent. Two systems can hold the same tonnage yet serialize it to different bytes, producing different leaf hashes and a report that fails to verify despite no tampering. Carrying quantities as integers in a fixed unit or as strings removes the ambiguity entirely.

What does the verification manifest contain that the report does not?

The report is the sealed artifact; the manifest is the recipe to re-derive its root. It lists the ordered leaf hashes with each record’s kind and id, the hash algorithm, the domain-separation prefixes, the odd-node rule, the root, and the key_id. An auditor with the manifest and the original records can recompute the root deterministically and, resolving the key from the key_id, confirm the signature — without trusting the report’s own claim about itself.

Up: Compliance Reporting & Automation

Generating Tamper-Evident Compliance Reports — the end-to-end walkthrough from source records to a signed, verifiable artifact
EPA e-Manifest Integration — the hazardous-waste records sealed into the report
FMCSA ELD Sync — the duty-status timeline bundled as a source record
Security & Access Boundaries — key resolution, rotation, and the HMAC chaining pattern
Core Architecture & Compliance Mapping — how compliance artifacts are versioned and stored