Generating SOX 404 Evidence Bundles in Python

Once a reconciliation period closes, an internal or external auditor testing SOX 404 controls does not want raw database access — they want a self-contained bundle: the population of audit events for the period, a reproducible sample drawn from that population, and cryptographic proof that nothing in the bundle was altered after packaging. This page implements that packaging step end to end in Python, sitting downstream of SOX Evidence Packaging: pulling the period’s events from a hash-chained audit log, drawing a deterministic seeded sample, building a manifest of SHA-256 digests, and signing the resulting archive so an auditor can verify it independently of the system that produced it.

Prerequisites

Step 1 — Define the population and collect audit events

The “population” is every audit event the control claims to cover for the period — not a query result you happen to get back. Before sampling, assert the event count against an independent count of the chain (e.g., a running counter maintained alongside the hash-chained ledger) so a truncated fetch is caught immediately rather than silently under-sampling.

python
from pydantic import BaseModel, ConfigDict
from uuid import UUID
from decimal import Decimal
from datetime import datetime, timezone
import logging

log = logging.getLogger("sox.evidence")

class AuditEvent(BaseModel):
    model_config = ConfigDict(frozen=True)

    event_id: UUID
    occurred_at: datetime          # timezone-aware UTC
    control_id: str                # e.g. "SOX-404-RECON-03"
    amount_usd: Decimal
    match_decision: str            # the reconciliation outcome this event records
    source_hash: str               # digest of the originating ledger row
    chain_hash: str                # this event's position in the hash chain

def collect_population(
    rows: list[dict],
    expected_count: int,
    period_start: datetime,
    period_end: datetime,
) -> list[AuditEvent]:
    events = [AuditEvent(**row) for row in rows]
    in_period = [e for e in events if period_start <= e.occurred_at < period_end]
    if len(in_period) != expected_count:
        raise ValueError(
            f"INCOMPLETE_POPULATION: expected {expected_count}, got {len(in_period)}"
        )
    # Sort by event_id so downstream sampling never depends on fetch order.
    return sorted(in_period, key=lambda e: e.event_id)

expected_count should come from a source independent of the query itself — a chain-length counter, a row-count trigger, or a nightly reconciliation of the audit table against its own hash chain. If the two disagree, the population is incomplete and no sample should be drawn from it.

Step 2 — Draw a deterministic, reproducible sample

Auditors routinely re-run a sample selection to confirm it matches what was delivered. Reproducibility means the same population plus the same seed always yields the same sample, independent of machine, Python version, or PYTHONHASHSEED. random.Random seeded with an explicit integer — never relying on set iteration order or dict hashing — satisfies that.

python
import random
import hashlib

def derive_seed(control_id: str, period_end: datetime, base_seed: int) -> int:
    blob = f"{control_id}|{period_end.isoformat()}|{base_seed}"
    digest = hashlib.sha256(blob.encode("utf-8")).hexdigest()
    return int(digest[:16], 16)

def select_sample(
    population: list[AuditEvent],
    sampling_rate: float,
    base_seed: int,
) -> tuple[list[AuditEvent], int]:
    if not population:
        raise ValueError("INCOMPLETE_POPULATION: empty population, nothing to sample")
    seed = derive_seed(population[0].control_id, population[-1].occurred_at, base_seed)
    rng = random.Random(seed)
    sample_size = max(1, round(len(population) * sampling_rate))
    # population is already sorted by event_id (Step 1) — required for reproducibility.
    indices = rng.sample(range(len(population)), sample_size)
    sample = [population[i] for i in sorted(indices)]
    return sample, seed

The seed is derived from the control ID, period end, and a base seed rather than hardcoded, so each control/period combination gets a distinct but still reproducible draw. Persist the returned seed alongside the bundle — an auditor needs it to re-run select_sample against the same population and confirm the identical sample comes back.

Step 3 — Build a manifest of SHA-256 digests

The manifest is the trust anchor for the bundle: a canonical JSON digest for every included artifact (population summary, sample, and each individual event), plus one top-level digest over the manifest entries themselves. Canonicalize before hashing — sort keys and fix separators — so the same logical content always produces the same digest regardless of dict insertion order.

python
import json

def canonical_digest(obj: dict) -> str:
    blob = json.dumps(obj, sort_keys=True, separators=(",", ":"), default=str)
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()

def build_manifest(
    population: list[AuditEvent],
    sample: list[AuditEvent],
    seed: int,
    sampling_rate: float,
) -> dict:
    entries = {
        str(e.event_id): canonical_digest(e.model_dump(mode="json"))
        for e in sample
    }
    manifest = {
        "population_count": len(population),
        "sample_count": len(sample),
        "sampling_rate": str(sampling_rate),
        "seed": seed,
        "entries": entries,
        "population_digest": canonical_digest(
            {"ids": [str(e.event_id) for e in population]}
        ),
    }
    manifest["manifest_digest"] = canonical_digest(manifest)
    return manifest

manifest_digest is computed over everything except itself, then attached last — that ordering is what verify_manifest in the test fixture below replays to confirm nothing was tampered with after generation.

Step 4 — Package and sign the bundle

Write the population summary, the sampled events, and the manifest into a zip archive, then sign the manifest digest with the referenced key. Signing happens against a detached signature file so the archive itself stays a plain, auditor-readable zip — only the signature needs the private key.

python
import zipfile
import hmac
import io

def sign_manifest(manifest_digest: str, signing_key_ref: str, key_material: bytes) -> str:
    # key_material is fetched from KMS/HSM by signing_key_ref at call time —
    # it is never stored in the bundle or logged.
    mac = hmac.new(key_material, manifest_digest.encode("utf-8"), hashlib.sha256)
    return mac.hexdigest()

def write_bundle(
    path: str,
    population: list[AuditEvent],
    sample: list[AuditEvent],
    manifest: dict,
    signing_key_ref: str,
    key_material: bytes,
    bundle_format: str = "zip",
) -> str:
    if bundle_format != "zip":
        raise ValueError(f"unsupported bundle_format: {bundle_format}")

    signature = sign_manifest(manifest["manifest_digest"], signing_key_ref, key_material)

    with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as zf:
        zf.writestr("manifest.json", json.dumps(manifest, indent=2, sort_keys=True))
        zf.writestr(
            "sample_events.json",
            json.dumps([e.model_dump(mode="json") for e in sample], indent=2, default=str),
        )
        zf.writestr(
            "population_ids.json",
            json.dumps([str(e.event_id) for e in population], indent=2),
        )
        zf.writestr(
            "signature.json",
            json.dumps(
                {"signing_key_ref": signing_key_ref, "algorithm": "HMAC-SHA256", "signature": signature},
                indent=2,
            ),
        )
    return signature

Every bundle must contain signature.json before it leaves the packaging process — an archive without it fails control testing outright, since there is no way to prove the manifest wasn’t edited after the fact.

Step 5 — Emit the structured audit log for the packaging run

The packaging run itself is a reconciliation-relevant event and belongs in the same audit trail as the transactions it evidences. Log a single structured record carrying the required trace_id/source_hash/match_decision triple, with match_decision recording the outcome of the packaging attempt.

python
from uuid import uuid4

def log_bundle_sealed(
    manifest: dict,
    signature: str,
    control_id: str,
    period_start: datetime,
    period_end: datetime,
) -> None:
    trace_id = str(uuid4())
    log.info(
        "sox.evidence.bundle_sealed",
        extra={
            "trace_id": trace_id,
            "source_hash": manifest["population_digest"],
            "match_decision": "BUNDLE_SEALED",
            "control_id": control_id,
            "period_start": period_start.isoformat(),
            "period_end": period_end.isoformat(),
            "sample_count": manifest["sample_count"],
            "population_count": manifest["population_count"],
            "manifest_digest": manifest["manifest_digest"],
            "signature": signature,
            "sealed_at": datetime.now(timezone.utc).isoformat(),
        },
    )

Because this record lands in the same append-only trail as the events it describes, an auditor can later confirm the packaging run itself was never retroactively altered — the same guarantee the hash-chained audit log provides for the underlying reconciliation events. For the broader control catalogue this bundle supports, see Compliance, Audit Trails & Financial Controls.

Configuration boundary table

Parameter Default Valid range Notes
sampling_rate 0.10 0.01–1.0 Fraction of population selected for substantive testing
seed period-derived any 64-bit int Derived per control/period so re-runs reproduce the same sample
bundle_format zip zip | tar.gz Container format written to evidence storage
signing_key_ref kms://sox-evidence/2026 KMS/HSM URI Reference only — key material fetched at sign time, never persisted
retention_days 2555 1825–3650 ~7 years, per typical SOX record-retention policy

Verification and testing

A fixture with a small fixed population confirms three things: the manifest digest recomputes correctly, the sample is reproducible across independent calls with the same seed, and tampering with a sampled event is detectable.

python
def make_event(n: int) -> AuditEvent:
    return AuditEvent(
        event_id=UUID(int=n),
        occurred_at=datetime(2026, 6, 30, tzinfo=timezone.utc),
        control_id="SOX-404-RECON-03",
        amount_usd=Decimal("1000.00") + Decimal(n),
        match_decision="AUTO_MATCHED",
        source_hash=f"src{n}",
        chain_hash=f"chain{n}",
    )

def test_sample_is_reproducible():
    population = [make_event(n) for n in range(20)]
    sample_a, seed_a = select_sample(population, sampling_rate=0.25, base_seed=42)
    sample_b, seed_b = select_sample(population, sampling_rate=0.25, base_seed=42)
    assert seed_a == seed_b
    assert [e.event_id for e in sample_a] == [e.event_id for e in sample_b]

def test_manifest_digest_detects_tampering():
    population = [make_event(n) for n in range(20)]
    sample, seed = select_sample(population, sampling_rate=0.25, base_seed=42)
    manifest = build_manifest(population, sample, seed, sampling_rate=0.25)

    recomputed = {k: v for k, v in manifest.items() if k != "manifest_digest"}
    assert canonical_digest(recomputed) == manifest["manifest_digest"]

    tampered = dict(manifest)
    tampered["entries"] = dict(manifest["entries"])
    tampered["entries"][str(sample[0].event_id)] = "0" * 64
    recomputed_tampered = {k: v for k, v in tampered.items() if k != "manifest_digest"}
    assert canonical_digest(recomputed_tampered) != manifest["manifest_digest"]

Run the full pipeline (collect_population → select_sample → build_manifest → write_bundle) as a scheduled job immediately after period close, before any downstream correction can touch the source rows. Export bundle_seal_duration_ms, sample_size, and signature_verify_failures as gauges, and alert if a scheduled packaging run does not complete before the period’s evidence-delivery deadline.

Troubleshooting

  • UNSIGNED_EVIDENCE — bundle delivered without a valid signature.json. Root cause: write_bundle was interrupted before signing, or a bundle was hand-assembled outside the pipeline. Fix: reject any archive missing signature.json at intake, and make signing the last step inside the same transaction as archive creation, not a separate follow-up job.
  • SAMPLE_NONREPRODUCIBLE — re-running select_sample with the recorded seed yields a different sample. Root cause: the population was sorted by iteration order (a set or unordered dict) instead of by event_id, so PYTHONHASHSEED or fetch order changed the input to rng.sample. Fix: always sort the population deterministically (Step 1) before sampling, and never sample directly from a set.
  • MANIFEST_HASH_MISMATCH — the auditor’s recomputed manifest_digest doesn’t match the delivered one. Root cause: the bundle was modified after sealing, or JSON was re-serialized with different key ordering before hashing. Fix: always hash with sort_keys=True and fixed separators, and treat any mismatch as a chain-of-custody failure requiring re-issuance, not a formatting bug to patch around.
  • INCOMPLETE_POPULATION — collect_population raises because the fetched row count is below expected_count. Root cause: pagination truncation on the audit-log query, or a genuine gap in the hash chain. Fix: verify the chain is gapless (each chain_hash links to the prior event’s digest) before packaging, and re-fetch with pagination fixed rather than sampling from a partial set.
  • SIGNING_KEY_UNAVAILABLE — sign_manifest fails to obtain key_material. Root cause: signing_key_ref points to a rotated or revoked key. Fix: pin the key version in signing_key_ref (not just an alias) and keep prior key versions retrievable for the full retention_days window so historical bundles remain verifiable.

Part of SOX Evidence Packaging within Compliance, Audit Trails & Financial Controls.