Secure Intake Form Design for DSR Pipelines

Within the broader DSR Architecture & Intake Routing framework, the intake form is the cryptographic perimeter of the entire pipeline — the single point where untrusted input from a public web form or a partner API crosses into a compliance-bound processing system. Before any Data Subject Request (DSR) enters the queue, the intake stage must reject malformed payloads, prove the submission actually came from your form, and stamp the request with the jurisdictional and timing metadata every downstream worker will rely on. Naive intake — a free-text ticket or an unvalidated JSON body — fails in three predictable ways: the statutory clock starts from the wrong event, an attacker forges or replays a submission, and divergent regulatory vocabulary (right to erasure versus right to delete) is never normalized, so the same request routes inconsistently. Each of those is a reportable compliance defect under GDPR Art. 12(3), which requires acting only on a verifiable request.

The gap this page closes is the distance between “the form accepts input” and “the form produces a signed, validated, jurisdiction-tagged, SLA-anchored canonical record that the Jurisdiction Routing Logic can dispatch without re-parsing untrusted data.” A secure intake form is not a data-collection surface; it is the compliance contract that governs everything downstream.

Phase 1: Signature Verification at the Perimeter #

Signature verification runs before schema parsing so that forged or replayed bodies are rejected without ever being deserialized. Public-facing forms and partner APIs both submit to a webhook endpoint that carries an X-DSR-Signature header — an HMAC-SHA256 digest of the raw request body computed with a shared secret held only by the form origin and the intake service. The comparison must be constant-time (hmac.compare_digest) to avoid timing side-channels, and a submission timestamp bounds the replay window so a captured payload cannot be resubmitted days later.

This is the perimeter control the DSR Architecture & Intake Routing pipeline treats as non-negotiable: the raw bytes and their signature are preserved as received, because GDPR Art. 12(3) obliges the controller to act only on a request it can prove is genuine.

import hmac
import hashlib
import time
from pydantic import SecretStr


class SignatureError(ValueError):
    """Raised when an intake submission fails HMAC or replay-window checks."""


def verify_intake_signature(
    raw_body: bytes,
    signature_header: str,
    submitted_epoch: int,
    secret: SecretStr,
    max_skew_seconds: int = 300,
) -> None:
    """Verify an intake POST before any deserialization occurs.

    Fails closed on skew, malformed header, or digest mismatch.
    """
    if abs(time.time() - submitted_epoch) > max_skew_seconds:
        raise SignatureError("Submission outside replay window")

    expected = hmac.new(
        secret.get_secret_value().encode(),
        raw_body,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(expected, signature_header):
        raise SignatureError("HMAC signature mismatch")

Because the shared secret is a high-sensitivity credential, it is resolved from a secret manager at call time rather than an environment variable, on a rotation schedule aligned with the key-lifecycle guidance in NIST SP 800-57 Part 1. Only bodies that clear this gate proceed to schema validation.

Phase 2: Strict Schema Validation & Payload Sanitization #

Ingestion fidelity dictates pipeline stability. Once a body is authenticated, it is validated against a Pydantic v2 model in strict mode so that no silent type coercion can corrupt a downstream processor. Strict mode rejects a string "true" where a boolean is expected, an integer where a string jurisdiction code belongs, and any field outside the declared schema — surfacing a precise, field-level error the client can remediate. This mirrors the validation discipline applied in the Schema Validation Rules stage, narrowed here to the intake contract.

from datetime import datetime, timezone
from enum import Enum
from pydantic import BaseModel, ConfigDict, EmailStr, Field, ValidationError


class RequestType(str, Enum):
    ACCESS = "access"
    DELETION = "deletion"
    OPT_OUT = "opt_out"
    CORRECTION = "correction"


class Jurisdiction(str, Enum):
    GDPR = "GDPR"
    CCPA = "CCPA"
    CPRA = "CPRA"
    VCDPA = "VCDPA"


class DSRIntakePayload(BaseModel):
    """Canonical, validated intake record. Strict mode forbids coercion."""

    model_config = ConfigDict(strict=True, extra="forbid", frozen=True)

    request_id: str = Field(min_length=32, max_length=64)
    email: EmailStr
    request_type: RequestType
    jurisdiction: Jurisdiction
    submitted_at: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc)
    )
    consent_token: str | None = None


def validate_intake(raw_json: dict) -> DSRIntakePayload:
    """Parse an authenticated body into the canonical record or raise."""
    try:
        return DSRIntakePayload(**raw_json)
    except ValidationError as exc:
        # Field-level errors are safe to return; the raw payload is not.
        raise ValueError(f"Intake schema violation: {exc.errors()}") from exc

extra="forbid" closes the injection vector of a caller smuggling unexpected keys into the record, and frozen=True makes the validated payload immutable so no later stage can silently mutate the request vector. Modeling jurisdiction and request_type as enums — rather than free-form regex-checked strings — means the taxonomy layer in Phase 3 can exhaustively match on known members and fail closed on anything else.

Phase 3: Jurisdiction Taxonomy Normalization #

Regulatory frameworks use divergent terminology for functionally identical operations, so normalization must occur immediately after validation and before any routing decision. GDPR’s right to erasure (Art. 17) and CCPA’s right to delete (§1798.105) both resolve to an internal deletion operation, but they carry different obligations, deadlines, and exemptions. The GDPR vs CCPA Request Taxonomies page details the full mapping; the intake form’s job is narrower — collapse the validated (jurisdiction, request_type) pair into a single deterministic queue identifier that downstream workers treat as ground truth.

TAXONOMY_ROUTING: dict[Jurisdiction, dict[RequestType, str]] = {
    Jurisdiction.GDPR: {
        RequestType.ACCESS: "eu_access_v2",
        RequestType.DELETION: "eu_erasure_v2",
        RequestType.CORRECTION: "eu_rectify_v2",
    },
    Jurisdiction.CCPA: {
        RequestType.ACCESS: "ca_access_v1",
        RequestType.DELETION: "ca_delete_v1",
        RequestType.OPT_OUT: "ca_optout_v1",
    },
    Jurisdiction.CPRA: {
        RequestType.ACCESS: "ca_access_v1",
        RequestType.DELETION: "ca_delete_v1",
        RequestType.OPT_OUT: "ca_optout_v1",
        RequestType.CORRECTION: "ca_correct_v1",
    },
    Jurisdiction.VCDPA: {
        RequestType.ACCESS: "va_access_v1",
        RequestType.DELETION: "va_delete_v1",
        RequestType.OPT_OUT: "va_optout_v1",
        RequestType.CORRECTION: "va_correct_v1",
    },
}


def resolve_queue(payload: DSRIntakePayload) -> str:
    """Map a validated request to a deterministic queue, or fail closed."""
    try:
        return TAXONOMY_ROUTING[payload.jurisdiction][payload.request_type]
    except KeyError as exc:
        raise RuntimeError(
            f"Unsupported jurisdiction/request combination: "
            f"{payload.jurisdiction}/{payload.request_type}"
        ) from exc

Note that not every request type is valid in every jurisdiction: CCPA has no correction right pre-CPRA, so a (CCPA, correction) pair fails closed rather than routing to an inappropriate handler. Failing closed here is a compliance feature — an unroutable combination is surfaced for human review instead of silently processed under the wrong legal basis.

Phase 4: Deterministic SLA Anchoring #

Deadline calculation must be auditable from the exact moment identity is proven — never from when the HTTP request arrived, and never inferred at runtime. Response windows vary by region: GDPR Art. 12(3) fixes the window at one month from receipt of a verifiable request, while CCPA §1798.130(a)(2) sets 45 days. Both allow a documented extension. The 30-Day vs 45-Day SLA Mapping page covers the full statutory decision matrix; the intake form’s responsibility is to compute one UTC-anchored sla_deadline and hand it downstream so every worker, scheduler, and escalation rule reads a single source of truth.

from datetime import timedelta

# Base statutory windows, in days from verified receipt.
SLA_DAYS: dict[Jurisdiction, int] = {
    Jurisdiction.GDPR: 30,
    Jurisdiction.CCPA: 45,
    Jurisdiction.CPRA: 45,
    Jurisdiction.VCDPA: 45,
}
# Extension REPLACES the base window (total days from verified receipt).
EXTENDED_SLA_DAYS: dict[Jurisdiction, int] = {
    Jurisdiction.GDPR: 90,   # Art. 12(3): up to two further months
    Jurisdiction.CCPA: 90,   # §1798.130(a)(2): one 45-day extension
    Jurisdiction.CPRA: 90,
    Jurisdiction.VCDPA: 90,
}


def compute_sla_deadline(
    payload: DSRIntakePayload,
    requires_extension: bool = False,
) -> datetime:
    """Return a UTC deadline. Extension must be set explicitly, not inferred."""
    table = EXTENDED_SLA_DAYS if requires_extension else SLA_DAYS
    # submitted_at is timezone-aware UTC, so arithmetic never drifts across DST.
    return payload.submitted_at + timedelta(days=table[payload.jurisdiction])

Anchoring to UTC is deliberate. As documented in the Python datetime module, timezone-aware arithmetic on UTC-anchored values is immune to the daylight-saving anomalies that corrupt naive local-time deadlines. The requires_extension flag is never guessed at runtime — it is set explicitly by the intake layer against documented complexity thresholds or a legal-counsel directive, and a scope_complexity metadata tag records the justification so a downstream worker can confirm the extension was legitimately granted before honoring the longer deadline.

Edge Cases & Conflict Resolution #

Real submissions violate assumptions, and the intake form must resolve each ambiguity deterministically rather than defaulting to whatever is convenient:

• UNKNOWN or absent jurisdiction. When geographic signals conflict or are missing, the payload cannot route. Rather than guessing, tag it jurisdiction=UNKNOWN, route to a manual-review queue, and start the shortest applicable clock (the GDPR one-month window) so the request is never accidentally treated as lower-urgency while its scope is resolved.
• Multi-jurisdiction overlap. A subject who qualifies under both CPRA and GDPR is handled under the most restrictive posture: the shorter deadline and the superset of rights. The intake record stores every matched framework so the response can document which obligations were applied.
• Valid signature, invalid schema. The submission is authenticated but malformed. It is logged (redacted) as a genuine attempt and returned a 400 with field-level errors — not silently dropped, because a discarded verifiable request is itself a compliance failure.
• Duplicate request_id. Idempotency is enforced at intake: a repeated request_id returns the original canonical record’s status instead of creating a second pipeline run, preventing double-execution of a deletion.
• Unroutable taxonomy pair. As in Phase 3, an unsupported (jurisdiction, request_type) combination fails closed to human review rather than routing under an incorrect legal basis.

Performance & Scale Considerations #

Intake sits in front of a public endpoint, so it must absorb bursts (a breach-notification wave, a viral privacy campaign) without dropping verifiable requests:

• Adaptive rate limiting protects the endpoint per source IP and per partner API key, but a request that clears signature verification is queued rather than rejected — a legitimate DSR must never be lost to a rate limiter.
• Cache jurisdiction lookups in Redis. Geographic-signal resolution and taxonomy tables are read-heavy and change rarely; caching them keeps validation off the hot path during high-volume intake.
• Keep validation synchronous, dispatch asynchronous. Signature verification and Pydantic parsing are cheap and run inline so the client gets an immediate 202 or 400. The heavier work — audit write, queue insertion — is handed to a worker pool, and Kafka partitioning by jurisdiction isolates a slow regional handler from the rest.
• Bound every external call. The secret-manager fetch and the audit-log write carry strict timeouts so a downstream stall cannot back-pressure the perimeter.

Testing & Compliance Verification #

The intake contract is verified with a payload matrix that pins every branch — valid, forged, replayed, malformed, and each jurisdiction/request combination — so a regression in taxonomy or SLA logic is caught before deploy.

import pytest


@pytest.mark.parametrize(
    "jurisdiction,request_type,expected_queue",
    [
        (Jurisdiction.GDPR, RequestType.DELETION, "eu_erasure_v2"),
        (Jurisdiction.CCPA, RequestType.OPT_OUT, "ca_optout_v1"),
        (Jurisdiction.VCDPA, RequestType.ACCESS, "va_access_v1"),
    ],
)
def test_resolve_queue_known_pairs(jurisdiction, request_type, expected_queue):
    payload = DSRIntakePayload(
        request_id="a" * 32,
        email="subject@example.com",
        request_type=request_type,
        jurisdiction=jurisdiction,
    )
    assert resolve_queue(payload) == expected_queue


def test_unroutable_pair_fails_closed():
    payload = DSRIntakePayload(
        request_id="b" * 32,
        email="subject@example.com",
        request_type=RequestType.CORRECTION,
        jurisdiction=Jurisdiction.CCPA,  # no pre-CPRA correction right
    )
    with pytest.raises(RuntimeError):
        resolve_queue(payload)


def test_gdpr_extension_replaces_window():
    payload = DSRIntakePayload(
        request_id="c" * 32,
        email="subject@example.com",
        request_type=RequestType.ACCESS,
        jurisdiction=Jurisdiction.GDPR,
    )
    base = compute_sla_deadline(payload)
    extended = compute_sla_deadline(payload, requires_extension=True)
    assert (extended - base).days == 60  # 90 total vs 30 base

Beyond unit coverage, every intake payload is written to an immutable, append-only audit log with PII redacted — the email reduced to a salted hash, the raw body replaced by its content hash — so the trail satisfies regulatory examination under GDPR Art. 30 without violating data-minimization principles. Held-out regulatory regions (a newly effective state statute, for example) are run against the matrix before their enum members are enabled in production, so an unsupported region fails closed rather than routing incorrectly on day one.

Frequently Asked Questions #

Related #

• DSR Architecture & Intake Routing — the parent pipeline architecture this intake stage feeds
• Jurisdiction Routing Logic — deterministic dispatch of the validated, signed record to a jurisdiction-specific queue
• GDPR vs CCPA Request Taxonomies — the full vocabulary mapping the intake form normalizes against
• 30-Day vs 45-Day SLA Mapping — statutory deadline translation for GDPR and CCPA/CPRA
• Building a Jurisdiction-Aware Intake Router in Python — end-to-end implementation that consumes this intake contract