Verifying Signed JSON Intake Payloads

Within the broader Secure Intake Form Design framework — part of the DSR Architecture & Intake Routing control plane — this page implements the cryptographic gate that proves an inbound Data Subject Request (DSR) JSON payload actually came from a trusted origin and was not altered or replayed in transit. Privacy and platform engineers hit this problem the moment a partner portal, a mobile app, or a public form submits requests over the open internet: you cannot start an irreversible erasure or a statutory clock on a body you cannot authenticate. Where the parent stage uses a shared-secret HMAC for a single form origin, asymmetric Ed25519 signatures scale to many independent submitters — each holding a private key, the intake service holding only public keys — so a compromised verifier never yields a signing capability. GDPR Art. 12(3) obliges a controller to act only on a verifiable request, and Art. 5(1)(f) requires processing that ensures integrity and confidentiality; a forged or tampered intake payload violates both. This page verifies the signature over a deterministically canonicalized body, rejects replays with a nonce and a timestamp window, validates the verified payload with Pydantic v2, and fails closed on any check — producing the trustworthy record the downstream jurisdiction-aware intake router consumes.

Verification runs as a strict sequence of gates, and a failure at any one of them fails closed to a single reject path that changes no downstream state:

Prerequisites #

The verifier targets Python 3.11+ and two runtime dependencies:

• cryptography — provides Ed25519PublicKey for fast, misuse-resistant asymmetric signature verification. Ed25519 has no parameter choices to get wrong (unlike RSA padding modes), which is why it is the recommended default for new signing systems.
• pydantic (v2) — strict validation of the verified payload. The code uses v2 idioms only (model_config = ConfigDict(...), model_validate, field_validator); v1 patterns (class Config, @validator, .parse_obj()) will not work.

Operationally, the verifier needs a registry mapping each sender’s key id to its Ed25519 public key (loaded from a secret manager or a pinned trust store, never from the payload itself), and a short-lived nonce store — Redis with a TTL equal to the skew window is the common choice. The signing side, owned by each submitter, canonicalizes the body identically and signs the resulting bytes; this page owns only the verify path.

Step-by-Step Implementation #

Step 1 — Load the sender’s public key from a trusted registry #

The public key must come from a store you control, keyed by a key_id carried in the payload envelope — never from the payload body, which would let an attacker present their own key. Resolving the key first also means an unknown sender fails closed before any cryptography runs.

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey


class VerificationError(ValueError):
    """Raised on any failed verification check. Always fails closed."""


# In production, resolve these from a secret manager / trust store, not code.
PUBLIC_KEY_REGISTRY: dict[str, Ed25519PublicKey] = {}


def load_public_key(key_id: str) -> Ed25519PublicKey:
    """Return the registered public key for a sender, or fail closed."""
    key = PUBLIC_KEY_REGISTRY.get(key_id)
    if key is None:
        raise VerificationError(f"Unknown signing key id: {key_id!r}")
    return key

Step 2 — Canonicalize the JSON deterministically before verifying #

A signature is over bytes, but JSON is a structural format — key order, whitespace, and Unicode escaping can all vary while representing the same object. Both signer and verifier must reduce the body to identical bytes, or valid signatures will fail. Canonicalize with sorted keys, no insignificant whitespace, and ensure_ascii=False so the byte sequence is stable and reproducible.

import json
from typing import Any


def canonicalize(payload: dict[str, Any]) -> bytes:
    """Produce the exact bytes that were signed, deterministically.

    Sorted keys and tight separators make the encoding reproducible on
    both sides; the signer must use the identical routine.
    """
    return json.dumps(
        payload,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
    ).encode("utf-8")

Canonicalization is applied to the inner claims object, excluding the detached signature field itself — signing a structure that contains its own signature is impossible. The envelope carries the signature and key_id alongside the claims object that is actually signed.

Step 3 — Verify the Ed25519 signature #

Ed25519PublicKey.verify raises InvalidSignature on any mismatch — a tampered body, a wrong key, or a truncated signature. Wrap it so every failure funnels into the same VerificationError, and never leak why a signature failed to the caller (a distinguishing error is an oracle).

import base64
from cryptography.exceptions import InvalidSignature


def verify_signature(claims: dict[str, Any], signature_b64: str, key: Ed25519PublicKey) -> None:
    """Verify the detached Ed25519 signature over the canonical claims."""
    message = canonicalize(claims)
    try:
        signature = base64.b64decode(signature_b64, validate=True)
        key.verify(signature, message)
    except (InvalidSignature, ValueError) as exc:
        raise VerificationError("Signature verification failed") from exc

Step 4 — Reject replays with a nonce and a timestamp window #

A valid signature alone does not stop replay: an attacker who captures a genuine signed erasure could resubmit it. Two independent checks close that gap. The timestamp must fall within a bounded skew window (so old captures expire), and the nonce must be unseen (so nothing inside the window is accepted twice). Store each accepted nonce with a TTL equal to the window; the two controls are complementary, not redundant.

import time


class NonceStore:
    """Records seen nonces with a TTL. Back with Redis SETNX in production."""

    def __init__(self, ttl_seconds: int = 300) -> None:
        self.ttl = ttl_seconds
        self._seen: dict[str, float] = {}

    def check_and_record(self, nonce: str) -> None:
        """Fail closed if the nonce was already used inside the window."""
        now = time.time()
        self._seen = {n: t for n, t in self._seen.items() if now - t < self.ttl}
        if nonce in self._seen:
            raise VerificationError("Replay detected: nonce already used")
        self._seen[nonce] = now


def check_timestamp(issued_epoch: int, max_skew_seconds: int = 300) -> None:
    """Fail closed if the submission is outside the allowed skew window."""
    if abs(time.time() - issued_epoch) > max_skew_seconds:
        raise VerificationError("Submission outside replay window")

Step 5 — Validate the verified payload with Pydantic v2 #

Only now — after the body is proven authentic, fresh, and unique — is it parsed into a strict schema. Ordering matters: validating before verifying would run the deserializer on untrusted input. Strict mode plus extra="forbid" rejects type coercion and smuggled keys, and frozen=True makes the accepted record immutable downstream.

from datetime import datetime, timezone
from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator


class VerifiedIntake(BaseModel):
    """A signed, replay-checked, schema-valid DSR intake claim."""
    model_config = ConfigDict(strict=True, extra="forbid", frozen=True)

    request_id: str = Field(min_length=8, max_length=64)
    subject_email: EmailStr
    request_type: str
    nonce: str = Field(min_length=16)
    issued_at: datetime

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


def verify_intake(envelope: dict[str, Any], nonces: NonceStore) -> VerifiedIntake:
    """Run all five gates in order; return the record or fail closed."""
    key = load_public_key(envelope["key_id"])          # 1
    claims = envelope["claims"]
    verify_signature(claims, envelope["signature"], key)  # 2 (canonicalize inside)
    nonces.check_and_record(claims["nonce"])           # 3
    check_timestamp(int(claims["issued_at_epoch"]))    # 4
    return VerifiedIntake.model_validate(claims)       # 5

Because every gate raises the same VerificationError (or a pydantic.ValidationError), the caller catches one failure family and returns a 401 without a distinguishing reason, satisfying the fail-closed guarantee and the integrity obligation of GDPR Art. 5(1)(f).

Configuration Reference #

Verification #

Prove the gate on three cases: a valid payload passes, a tampered body is rejected, and a replayed nonce is rejected even with a valid signature.

import base64
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey


def _sign(claims: dict, priv: Ed25519PrivateKey) -> str:
    return base64.b64encode(priv.sign(canonicalize(claims))).decode()


def test_valid_payload_passes():
    priv = Ed25519PrivateKey.generate()
    PUBLIC_KEY_REGISTRY["k1"] = priv.public_key()
    claims = {"request_id": "REQ-00001", "subject_email": "a@example.com",
              "request_type": "access", "nonce": "n" * 16,
              "issued_at": "2026-07-16T00:00:00+00:00",
              "issued_at_epoch": int(time.time())}
    env = {"key_id": "k1", "claims": claims, "signature": _sign(claims, priv)}
    record = verify_intake(env, NonceStore())
    assert record.request_id == "REQ-00001"


def test_tampered_body_is_rejected():
    priv = Ed25519PrivateKey.generate()
    PUBLIC_KEY_REGISTRY["k2"] = priv.public_key()
    claims = {"request_id": "REQ-00002", "subject_email": "a@example.com",
              "request_type": "access", "nonce": "m" * 16,
              "issued_at": "2026-07-16T00:00:00+00:00",
              "issued_at_epoch": int(time.time())}
    sig = _sign(claims, priv)
    claims["request_type"] = "deletion"   # tamper after signing
    env = {"key_id": "k2", "claims": claims, "signature": sig}
    try:
        verify_intake(env, NonceStore())
        assert False, "tampered body must be rejected"
    except VerificationError:
        pass


def test_replayed_nonce_is_rejected():
    priv = Ed25519PrivateKey.generate()
    PUBLIC_KEY_REGISTRY["k3"] = priv.public_key()
    claims = {"request_id": "REQ-00003", "subject_email": "a@example.com",
              "request_type": "access", "nonce": "r" * 16,
              "issued_at": "2026-07-16T00:00:00+00:00",
              "issued_at_epoch": int(time.time())}
    env = {"key_id": "k3", "claims": claims, "signature": _sign(claims, priv)}
    store = NonceStore()
    verify_intake(env, store)              # first use: accepted
    try:
        verify_intake(env, store)          # replay: rejected
        assert False, "replay must be rejected"
    except VerificationError:
        pass

Expect the tampered case to fail at gate 2 (the canonical bytes no longer match the signature) and the replay case to fail at gate 3 (the nonce is already recorded), even though its signature is genuine. Every rejection should write a redacted audit entry — key_id, the failing gate, a content hash of the body, never the raw PII — so the trail supports examination under GDPR Art. 30.

Troubleshooting #

Valid signatures intermittently fail — Root cause: the signer and verifier canonicalize differently (key order, whitespace, or ASCII escaping). Fix: share one canonicalization routine with sort_keys=True, separators=(",",":"), and ensure_ascii=False on both sides, and sign the exact bytes it emits.

Replay succeeds despite a nonce check — Root cause: the nonce TTL is shorter than the timestamp skew window, so a nonce expires while its timestamp is still valid. Fix: set NonceStore.ttl_seconds equal to max_skew_seconds so no in-window payload can be replayed.

Signature verifies but the wrong key was used — Root cause: the key_id was read from an untrusted field or the registry was seeded from the payload. Fix: resolve keys only from a pinned trust store keyed by an envelope key_id, and reject unknown ids before verifying.

Deserializer runs on unauthenticated input — Root cause: Pydantic validation was placed before signature verification. Fix: keep the gate order — verify, then replay-check, then validate — so untrusted bytes never reach the parser.

A distinguishing error leaks why verification failed — Root cause: returning different messages for bad signature versus expired timestamp gives an attacker an oracle. Fix: collapse all failures into one 401 with a generic reason and record the specific cause only in the internal audit log.

Frequently Asked Questions #

Related #

• Secure Intake Form Design — the parent framework this signature gate belongs to, including HMAC signing and SLA anchoring.
• Building a Jurisdiction-Aware Intake Router in Python — the downstream consumer that routes the verified payload this page produces.
• Jurisdiction Routing Logic — deterministic dispatch of the authenticated record to a jurisdiction-specific queue.
• GDPR vs CCPA Request Taxonomies — normalizing the verified request onto a canonical action.
• DSR Architecture & Intake Routing — the parent control plane this verification stage sits within.