Generating a Regulator-Ready Closure Manifest

A Data Subject Request (DSR) is only closed when you can hand a regulator a single file that proves it — a sealed artifact they can verify offline, without logging into your system. This page is the concrete Python build of that artifact described in the parent Closure Manifests & Evidence Packaging specification, and it sits within the wider Audit Logging & Compliance Evidence framework. The engineers who need it are privacy and platform teams whose auditors have started asking not “is it done?” but “prove it was done correctly, and let me check the proof myself.” The failure mode this prevents is the manifest that looks authoritative but verifies nothing: a JSON blob with a "signed": true flag and no way to recompute anything. Below we build a typed ClosureManifest, assemble it from an audit ledger, seal it with a digest over canonical JSON, and — the part that matters — write an offline verify_manifest() that recomputes the hash chain from genesis to head and confirms every dispatched sub-task carries a completion proof. The whole point is that GDPR Art. 5(2) accountability is demonstrated by recomputation, not asserted by a boolean.

Prerequisites #

The build targets Python 3.11+ (for datetime, zoneinfo, and PEP 604 X | None unions without from __future__). Dependencies are deliberately minimal so the verifier has no heavy runtime:

• pydantic (v2) — strict validation of the manifest and proof models. The code uses v2 idioms (model_config = ConfigDict(...), field_validator, model_validate_json, model_dump(mode="json")); v1 patterns (class Config, @validator, .dict()) will not work.
• hashlib and hmac (stdlib) — SHA-256 digests and the seal signature. This example seals with HMAC-SHA256 for a self-contained, copy-pasteable build; in production, swap the HMAC for an asymmetric signature from a KMS or HSM per NIST SP 800-57 key-management guidance so the seal is publicly verifiable without sharing the signing key.

This page assumes an existing hash-chained ledger of the kind built in Building a Hash-Chained Audit Ledger in Python; here we read from it rather than construct it.

Step-by-step implementation #

Step 1 — Model the events, proofs, and manifest #

Everything is a typed model so a malformed manifest fails loudly at construction rather than silently at audit time. A ManifestEvent carries the fields that were hashed into the ledger link plus the recorded digest, so the verifier can recompute the link and compare. A TaskProof binds a completed sub-task to the audit event that recorded its terminal transition.

import hashlib
import hmac
import json
from datetime import datetime, timezone
from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, Field, field_validator

GENESIS = "0" * 64


def _utc(value: datetime) -> datetime:
    """Normalize any aware datetime to UTC; reject naive input."""
    if value.tzinfo is None:
        raise ValueError("timestamps must be timezone-aware (UTC)")
    return value.astimezone(timezone.utc)


class ManifestEvent(BaseModel):
    """One chained audit event, copied verbatim from the ledger."""

    model_config = ConfigDict(strict=True)

    seq: int = Field(ge=0)
    request_id: str = Field(pattern=r"^dsr_[0-9a-f]{32}$")
    stage: str = Field(min_length=1, max_length=32)
    legal_basis: str = Field(min_length=3, max_length=64)
    actor: str = Field(min_length=1, max_length=128)
    at: datetime
    digest: str = Field(pattern=r"^[0-9a-f]{64}$")

    @field_validator("at")
    @classmethod
    def _aware(cls, v: datetime) -> datetime:
        return _utc(v)

    def canonical_bytes(self) -> bytes:
        """Deterministic bytes hashed into the chain link (excludes digest)."""
        payload = {
            "seq": self.seq,
            "request_id": self.request_id,
            "stage": self.stage,
            "legal_basis": self.legal_basis,
            "actor": self.actor,
            "at": self.at.isoformat(),
        }
        return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()


class TaskProof(BaseModel):
    """Evidence that one dispatched sub-task completed."""

    model_config = ConfigDict(strict=True)

    task_id: str = Field(min_length=1, max_length=64)
    system: str = Field(min_length=1, max_length=64)
    action: Literal["extracted", "redacted", "deleted", "opted_out"]
    record_count: int = Field(ge=0)
    result_digest: str = Field(pattern=r"^[0-9a-f]{64}$")
    event_seq: int = Field(ge=0)  # references the ManifestEvent that recorded it


def link(prev_hash: str, event: ManifestEvent) -> str:
    """Recompute a chain link digest from its predecessor and the event."""
    hasher = hashlib.sha256()
    hasher.update(prev_hash.encode())
    hasher.update(event.canonical_bytes())
    return hasher.hexdigest()

Step 2 — Define the manifest and build it from a ledger #

The ClosureManifest bundles the ordered event chain, the resolved framework and its inputs, the per-task proofs, the set of tasks that were dispatched (for the coverage check), the outcome, and a retention-until timestamp. The Seal and manifest_digest start empty and are filled by seal().

class Seal(BaseModel):
    model_config = ConfigDict(strict=True)

    alg: str
    key_id: str
    signature: str = Field(pattern=r"^[0-9a-f]{64}$")


class ClosureManifest(BaseModel):
    model_config = ConfigDict(strict=True)

    manifest_version: str = "1.0.0"
    request_id: str = Field(pattern=r"^dsr_[0-9a-f]{32}$")
    framework: Literal["GDPR", "CCPA"]
    framework_inputs: dict[str, Any]
    event_chain: list[ManifestEvent]
    ledger_head: str = Field(pattern=r"^[0-9a-f]{64}$")
    task_proofs: list[TaskProof]
    dispatched_tasks: list[str]
    outcome: Literal["fulfilled", "denied"]
    legal_basis: str = Field(min_length=3, max_length=256)
    retention_until: datetime
    manifest_digest: str | None = None
    seal: Seal | None = None

    @field_validator("retention_until")
    @classmethod
    def _aware(cls, v: datetime) -> datetime:
        return _utc(v)

    def compute_digest(self) -> str:
        """SHA-256 over canonical JSON of the unsealed body."""
        body = self.model_dump(
            mode="json", exclude={"manifest_digest", "seal"}
        )
        encoded = json.dumps(body, sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(encoded.encode("utf-8")).hexdigest()


def build_manifest(
    request_id: str,
    ledger: list[ManifestEvent],
    dispatched_tasks: list[str],
    proofs: list[TaskProof],
    framework: Literal["GDPR", "CCPA"],
    framework_inputs: dict[str, Any],
    outcome: Literal["fulfilled", "denied"],
    legal_basis: str,
    retention_until: datetime,
) -> ClosureManifest:
    """Assemble an unsealed manifest from a request's ledger and proofs."""
    if not ledger:
        raise ValueError("cannot build a manifest from an empty ledger")
    return ClosureManifest(
        request_id=request_id,
        framework=framework,
        framework_inputs=framework_inputs,
        event_chain=ledger,
        ledger_head=ledger[-1].digest,
        task_proofs=proofs,
        dispatched_tasks=dispatched_tasks,
        outcome=outcome,
        legal_basis=legal_basis,
        retention_until=retention_until,
    )

Step 3 — Compute the canonical digest and seal #

Sealing computes the digest over the canonical body and signs it. Using model_dump(mode="json") converts nested models and timestamps to their JSON forms first, so the sorted-key json.dumps is byte-reproducible on any machine.

def seal_manifest(manifest: ClosureManifest, key: bytes, key_id: str) -> ClosureManifest:
    """Return a sealed copy: digest computed, signature attached."""
    digest = manifest.compute_digest()
    signature = hmac.new(key, digest.encode("utf-8"), hashlib.sha256).hexdigest()
    return manifest.model_copy(
        update={
            "manifest_digest": digest,
            "seal": Seal(alg="HMAC-SHA256", key_id=key_id, signature=signature),
        }
    )

Step 4 — Verify offline #

verify_manifest() is the heart of independent verifiability. It takes only the serialized bytes and a verification key, then reaches a verdict with no database, no network, and no trust in the producer. It recomputes the chain from genesis to head, enforces that every dispatched task has a proof pointing at a real event, re-derives the manifest digest, and checks the seal.

class VerificationError(Exception):
    """Raised when a manifest fails any integrity check."""


def verify_manifest(raw: bytes, verify_key: bytes) -> ClosureManifest:
    """Validate a serialized manifest offline; raise on any failure."""
    manifest = ClosureManifest.model_validate_json(raw)
    if manifest.manifest_digest is None or manifest.seal is None:
        raise VerificationError("manifest is not sealed")

    # 1. Recompute the hash chain from genesis to the recorded head.
    prev = GENESIS
    for event in manifest.event_chain:
        recomputed = link(prev, event)
        if recomputed != event.digest:
            raise VerificationError(f"chain broken at seq {event.seq}")
        prev = recomputed
    if prev != manifest.ledger_head:
        raise VerificationError("ledger head does not match recomputed chain")

    # 2. Coverage: every dispatched task has a proof anchored to a real event.
    event_seqs = {e.seq for e in manifest.event_chain}
    proven = {p.task_id for p in manifest.task_proofs}
    missing = set(manifest.dispatched_tasks) - proven
    if missing:
        raise VerificationError(f"dispatched tasks without proof: {sorted(missing)}")
    for proof in manifest.task_proofs:
        if proof.event_seq not in event_seqs:
            raise VerificationError(f"proof {proof.task_id} cites unknown event")

    # 3. Re-derive the digest and check the seal signature.
    recomputed_digest = manifest.compute_digest()
    if recomputed_digest != manifest.manifest_digest:
        raise VerificationError("manifest digest mismatch (body was altered)")
    expected = hmac.new(
        verify_key, recomputed_digest.encode("utf-8"), hashlib.sha256
    ).hexdigest()
    if not hmac.compare_digest(expected, manifest.seal.signature):
        raise VerificationError("seal signature does not verify")

    return manifest

Step 5 — Serialize to JSON #

The sealed manifest serializes with Pydantic’s own JSON encoder, which is the exact form the verifier reads back. This is the byte stream you write to write-once storage and later hand to a regulator; providing an access-request copy in a structured electronic form also satisfies the GDPR Art. 15(3) obligation to supply the data in a commonly used electronic format.

def to_bytes(manifest: ClosureManifest) -> bytes:
    """Serialize a sealed manifest to the canonical JSON bytes to store."""
    return manifest.model_dump_json().encode("utf-8")

Configuration reference #

Verification #

The one test that proves the seal is not decorative: tamper with a sealed manifest and confirm verification fails. A round trip must pass, and any single-byte mutation must raise.

import pytest


def _fixture() -> tuple[ClosureManifest, bytes]:
    key = b"unit-test-key-not-for-production"
    rid = "dsr_" + "a" * 32
    e0 = ManifestEvent(
        seq=0, request_id=rid, stage="intake", legal_basis="art15_access",
        actor="svc:intake", at=datetime(2026, 1, 1, tzinfo=timezone.utc),
        digest=link(GENESIS, ManifestEvent(
            seq=0, request_id=rid, stage="intake", legal_basis="art15_access",
            actor="svc:intake", at=datetime(2026, 1, 1, tzinfo=timezone.utc),
            digest="0" * 64)),
    )
    e1 = ManifestEvent(
        seq=1, request_id=rid, stage="closed", legal_basis="art15_access",
        actor="svc:closer", at=datetime(2026, 1, 10, tzinfo=timezone.utc),
        digest=link(e0.digest, ManifestEvent(
            seq=1, request_id=rid, stage="closed", legal_basis="art15_access",
            actor="svc:closer", at=datetime(2026, 1, 10, tzinfo=timezone.utc),
            digest="0" * 64)),
    )
    proof = TaskProof(
        task_id="crm", system="salesforce", action="extracted",
        record_count=3, result_digest="b" * 64, event_seq=1,
    )
    manifest = build_manifest(
        request_id=rid, ledger=[e0, e1], dispatched_tasks=["crm"],
        proofs=[proof], framework="GDPR", framework_inputs={"jurisdiction": "DE"},
        outcome="fulfilled", legal_basis="art15_access",
        retention_until=datetime(2028, 1, 10, tzinfo=timezone.utc),
    )
    sealed = seal_manifest(manifest, key, key_id="kms-dsr-2026")
    return sealed, to_bytes(sealed)


def test_valid_manifest_verifies():
    _, raw = _fixture()
    assert verify_manifest(raw, b"unit-test-key-not-for-production")


def test_tampered_manifest_fails():
    sealed, _ = _fixture()
    # Flip a proof's record_count without re-sealing.
    tampered = sealed.model_copy(deep=True)
    tampered.task_proofs[0] = tampered.task_proofs[0].model_copy(
        update={"record_count": 999}
    )
    raw = to_bytes(tampered)
    with pytest.raises(VerificationError, match="digest mismatch"):
        verify_manifest(raw, b"unit-test-key-not-for-production")


def test_missing_proof_fails():
    sealed, _ = _fixture()
    stripped = sealed.model_copy(update={"task_proofs": []})
    resealed = seal_manifest(
        stripped.model_copy(update={"manifest_digest": None, "seal": None}),
        b"unit-test-key-not-for-production", key_id="kms-dsr-2026",
    )
    with pytest.raises(VerificationError, match="without proof"):
        verify_manifest(to_bytes(resealed), b"unit-test-key-not-for-production")

The tamper test is the assertion that matters: mutating the body changes the recomputed digest, so compute_digest() no longer matches the stored manifest_digest and verification raises. The missing-proof test proves the coverage invariant — a manifest that re-seals cleanly still fails because a dispatched task has no proof.

Troubleshooting #

Digest mismatch on a manifest nobody touched — Root cause: non-canonical serialization, usually a naive (timezone-less) datetime or a library that doesn’t sort keys. Fix: normalize every timestamp to UTC at construction (the field_validator above) and always hash via model_dump(mode="json") with sort_keys=True; never hash str(manifest).

Chain verifies but the head does not match — Root cause: event_chain was re-sorted on a wall-clock timestamp instead of preserving the ledger’s intrinsic seq order. Fix: copy events in their stored chain order; the recomputed prev after the last event must equal ledger_head.

Verification needs a database call — Root cause: proofs reference records by primary key rather than embedding a result_digest. Fix: put a content digest in each proof so the verifier commits to the result offline; the bulk data stays in WORM storage, out of the manifest.

Seal verifies for anyone who has the key — Root cause: symmetric HMAC lets any holder of the key forge a seal. Fix: for regulator-facing non-repudiation, sign with an asymmetric KMS/HSM key (NIST SP 800-57) so the regulator verifies with your public key and cannot themselves forge a seal.

A denied request produces no manifest — Root cause: the packager only runs on fulfilled. Fix: build and seal manifests for denied outcomes too, carrying the legal_basis; the refusal is exactly what a regulator reviews.

Frequently Asked Questions #

Related #

• Closure Manifests & Evidence Packaging — the parent specification defining what a manifest bundles and why.
• Building a Hash-Chained Audit Ledger in Python — the ledger this build reads its events and head digest from.
• Verifying Audit Chain Integrity After Tampering — the recompute-and-localize technique the verifier applies to the event chain.
• WORM Storage for DSR Evidence — the write-once store the sealed JSON bytes are written to under a retention lock.
• Audit Logging & Compliance Evidence — the framework this closure step completes.