Building a Jurisdiction-Aware Intake Router in Python

Within the broader DSR Architecture & Intake Routing framework, the intake router is the first executable control plane a data subject request (DSR) touches — it must resolve which statute governs the request, start the correct statutory clock, and hand off an encrypted, jurisdiction-tagged envelope before any personal data reaches a fulfillment worker. This page is the concrete Python implementation of the decision described in the parent Jurisdiction Routing Logic specification: the engineers who build it are privacy or platform teams that receive DSRs from many regions and cannot afford routing ambiguity, because a mis-routed request is a reportable compliance defect — an EU resident given 45 days instead of the one month GDPR Art. 12(3) allows, or a California opt-out executed with erasure semantics. The router resolves conflicting residency signals, normalizes heterogeneous request labels, and fails closed to manual triage on genuine ambiguity, so that every downstream stage reads a deterministic, auditable directive rather than raw consumer wording.

The router resolves each payload through gating, multi-signal jurisdiction resolution, SLA computation, and a three-tier fallback before handing off to fulfillment:

Jurisdiction-aware intake router: payload flow with fail-closed branches A DSR intake payload flows down the router's main path. First a schema gate: an invalid payload is rejected at the edge, a valid one is stamped with a correlation ID. It then reaches the multi-signal jurisdiction resolver, which weighs an explicit declaration at 0.95, billing country at 0.70, and IP geolocation at 0.50. A confidence gate at 0.60 diverts under-confident payloads to manual triage with the SLA clock paused. Confident payloads have their statutory SLA deadline computed, then pass through KMS envelope encryption: an encryption error is diverted to the encryption dead-letter queue for exponential-backoff retry, while success hands off a PII-free envelope to the fulfillment worker. Intake payload — schema gate Multi-signal jurisdiction resolver Compute SLA deadline Fulfillment worker (PII-free) Schema valid? Confidence≥ 0.60? KMS envelopeencrypt OK? Reject at edge Manual triage Encryption DLQ no yes — correlation ID no — SLA paused yes error ok retry: exponential backoff Signal weights: explicit 0.95 billing 0.70 ip_geo 0.50

Prerequisites

The router targets Python 3.11+ (required for the standard-library zoneinfo module and tuple[...] / list[...] generic syntax without from __future__). Install the following runtime dependencies:

  • pydantic (v2) — strict runtime validation of the intake payload at the edge. The code below uses v2 idioms (model_config = ConfigDict(...), @field_validator); v1 patterns (class Config, @validator) will not work.
  • cryptography — AESGCM authenticated encryption for the envelope data-encryption key (DEK).
  • A KMS client — e.g. boto3 for AWS KMS, or google-cloud-kms. The examples assume an AWS-style encrypt(KeyId=..., Plaintext=...) interface; adapt the call for your provider.
  • zoneinfo (stdlib) — timezone-aware SLA arithmetic. For jurisdiction-specific public-holiday calendars in production, add a maintained library such as holidays or workalendar.

Infra-side, the router expects three durable message queues (FULFILLMENT_WORKER, MANUAL_TRIAGE, ENCRYPTION_DLQ) and a hardware-backed KMS key aliased alias/dsr-intake-key. This page owns only the intake decision; it assumes a signed, deduplicated payload arriving from Secure Intake Form Design.

Step-by-Step Implementation

Step 1 — Deterministic payload gating and idempotency

Every intake event is validated against a strict schema before it is committed to any queue, and is stamped with a cryptographically unique correlation ID the moment it is received. That idempotency key propagates through every routing decision, SLA calculation, and audit record, so retries never inflate the SLA clock or trigger duplicate extraction jobs. Pydantic v2’s strict=True, extra="forbid" rejects malformed or over-broad submissions at the edge rather than letting them fail silently downstream.

import uuid
import logging
from typing import Optional, Literal
from datetime import datetime, timezone
from pydantic import BaseModel, Field, field_validator, ConfigDict

logger = logging.getLogger("dsr.intake_router")

class DSRIntakePayload(BaseModel):
    """Validated, immutable representation of a single inbound DSR."""
    model_config = ConfigDict(strict=True, extra="forbid")

    request_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    consumer_email: str
    explicit_jurisdiction: Optional[str] = None
    request_type: Literal["access", "deletion", "rectification", "opt_out_sale"]
    raw_ip: Optional[str] = None
    billing_country: Optional[str] = None
    submitted_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

    @field_validator("consumer_email")
    @classmethod
    def validate_email(cls, v: str) -> str:
        if "@" not in v or len(v) > 254:
            raise ValueError("Invalid consumer identifier format")
        return v.lower()

Step 2 — Multi-signal jurisdiction resolution

Jurisdiction resolution parses multiple residency indicators — an explicit consumer declaration, billing country, and IP geolocation — each carrying a confidence weight. When signals align, routing is trivial; when they diverge, the resolver picks the highest-confidence signal and breaks ties toward the strictest applicable regulation. The full precedence rationale (and the canonical-action mapping this feeds) lives in GDPR vs CCPA Request Taxonomies; the code below returns both the decision and an audit trail of every signal considered.

from enum import Enum
from dataclasses import dataclass

class Jurisdiction(Enum):
    GDPR = "gdpr"
    CCPA_CPRA = "ccpa_cpra"
    VCDPA = "vcdpa"
    UNKNOWN = "unknown"

@dataclass(frozen=True)
class JurisdictionSignal:
    source: str
    jurisdiction: Jurisdiction
    confidence: float  # 0.0 to 1.0

def resolve_primary_jurisdiction(payload: DSRIntakePayload) -> tuple[Jurisdiction, list[dict]]:
    """Return the governing jurisdiction and a full signal audit trail."""
    signals: list[JurisdictionSignal] = []

    # 1. Explicit declaration (highest weight)
    if payload.explicit_jurisdiction:
        signals.append(JurisdictionSignal("explicit", Jurisdiction(payload.explicit_jurisdiction), 0.95))

    # 2. Billing country (medium weight)
    if payload.billing_country:
        mapping = {"GB": Jurisdiction.GDPR, "DE": Jurisdiction.GDPR, "US-CA": Jurisdiction.CCPA_CPRA}
        signals.append(JurisdictionSignal("billing", mapping.get(payload.billing_country, Jurisdiction.UNKNOWN), 0.70))

    # 3. IP geolocation (lower weight, fallback)
    if payload.raw_ip:
        # In production, call internal GeoIP service here
        signals.append(JurisdictionSignal("ip_geo", Jurisdiction.UNKNOWN, 0.50))

    # Precedence matrix: strictest applicable regulation wins on ties.
    # GDPR > CCPA/CPRA > VCDPA > UNKNOWN
    strictness_order = [Jurisdiction.GDPR, Jurisdiction.CCPA_CPRA, Jurisdiction.VCDPA, Jurisdiction.UNKNOWN]

    valid_signals = [s for s in signals if s.jurisdiction != Jurisdiction.UNKNOWN]
    if not valid_signals:
        return Jurisdiction.UNKNOWN, []

    # Resolve conflicts by highest confidence, then strictest fallback.
    resolved = max(valid_signals, key=lambda s: (s.confidence, -strictness_order.index(s.jurisdiction)))

    audit_trail = [{"source": s.source, "jurisdiction": s.jurisdiction.value, "weight": s.confidence} for s in signals]
    return resolved.jurisdiction, audit_trail

Step 3 — SLA calculation and statutory tolling

Once jurisdiction is resolved, the router computes the statutory response deadline. GDPR Art. 12(3) fixes the response window at one month from receipt of a verifiable request; CCPA §1798.130(a)(2) fixes it at 45 days. When signals imply dual scope — a California IP with a UK billing profile — the router computes both and applies the shorter deadline while flagging the payload for compliance review. The full deadline-translation logic, including CCPA’s 45+45-day and GDPR’s two-month complexity extensions, is detailed in 30-Day vs 45-Day SLA Mapping; the router only needs the base deadline plus any tolling already applied.

from datetime import timedelta
from zoneinfo import ZoneInfo

def calculate_sla_deadline(jurisdiction: Jurisdiction, submitted_at: datetime, tolling_days: int = 0) -> datetime:
    """Compute a UTC deadline from the governing statute's response window."""
    base_days = {
        Jurisdiction.GDPR: 30,
        Jurisdiction.CCPA_CPRA: 45,
        Jurisdiction.VCDPA: 45,
    }.get(jurisdiction, 45)

    # Business-day arithmetic (simplified; production should use `holidays` / `workalendar`).
    current = submitted_at.astimezone(ZoneInfo("UTC"))
    business_days_elapsed = 0
    while business_days_elapsed < base_days + tolling_days:
        current += timedelta(days=1)
        if current.weekday() < 5:  # skip Sat/Sun
            business_days_elapsed += 1

    return current

Step 4 — Secure PII envelope and handoff

The router never stores raw consumer identifiers in plaintext. It applies envelope encryption — a fresh 256-bit DEK per request encrypts the PII with AES-GCM, and the KMS wraps the DEK — then routes only the ciphertext, the wrapped DEK, and non-sensitive metadata to the fulfillment queue. This aligns with NIST SP 800-57 Part 1 Rev. 5 key-management guidance.

import os
import json
import base64
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

class SecureIntakeEnvelope:
    def __init__(self, kms_client):
        self.kms = kms_client

    def wrap_payload(self, payload: DSRIntakePayload, jurisdiction: Jurisdiction, deadline: datetime) -> dict:
        """Encrypt PII under a KMS-wrapped DEK and return a PII-free routing envelope."""
        # 1. Serialize raw PII.
        raw_pii = json.dumps({"email": payload.consumer_email, "request_id": payload.request_id}).encode()

        # 2. Generate DEK and encrypt (envelope pattern).
        dek = AESGCM.generate_key(bit_length=256)
        aesgcm = AESGCM(dek)
        nonce = os.urandom(12)
        ciphertext = aesgcm.encrypt(nonce, raw_pii, None)

        # 3. Wrap the DEK with KMS.
        encrypted_dek = self.kms.encrypt(KeyId="alias/dsr-intake-key", Plaintext=dek)

        # 4. Construct routing envelope (NO raw PII).
        return {
            "correlation_id": payload.request_id,
            "jurisdiction": jurisdiction.value,
            "sla_deadline": deadline.isoformat(),
            "request_type": payload.request_type,
            "encrypted_dek_blob": encrypted_dek["CiphertextBlob"],
            "ciphertext_b64": base64.b64encode(nonce + ciphertext).decode(),
            "kms_key_id": encrypted_dek["KeyId"],
        }

Step 5 — Fallback routing and escalation

No intake system operates without failure. Network partitions, KMS throttling, or ambiguous jurisdictional signals require deterministic fallback. The router applies a three-tier escalation matrix:

  1. Circuit-breaker fallback — if KMS or the GeoIP resolver exceeds its timeout, the payload goes to the ENCRYPTION_DLQ in a PENDING_ENCRYPTION state; a background worker retries with exponential backoff.
  2. Jurisdictional ambiguity — when confidence falls below 0.60 or a conflict cannot be resolved, jurisdiction is set to UNKNOWN and the payload is routed to MANUAL_TRIAGE with the SLA clock paused until a compliance officer assigns a definitive statute.
  3. Taxonomy mismatch — if request_type normalization is ambiguous (e.g. “remove my data” between deletion and opt-out), the payload is parked in CLARIFICATION_HOLD and a clarification email is dispatched.
class FallbackRouter:
    def route_with_fallback(self, envelope: dict, confidence: float) -> str:
        """Deterministic three-tier fallback: fail closed to human review on ambiguity."""
        if confidence < 0.60:
            return self._send_to_queue("MANUAL_TRIAGE", envelope, status="AMBIGUOUS_JURISDICTION")
        if envelope.get("kms_error"):
            return self._send_to_queue("ENCRYPTION_DLQ", envelope, status="KMS_FAILURE")
        return self._send_to_queue("FULFILLMENT_WORKER", envelope, status="READY")

Configuration Reference

Parameter Type Default Compliance note
confidence_threshold float 0.60 Below this, the router fails closed to MANUAL_TRIAGE rather than guessing a statute.
strictness_order list[Jurisdiction] [GDPR, CCPA_CPRA, VCDPA, UNKNOWN] Tie-break order; strictest regime wins so under-protection is never the default.
base_days (GDPR) int 30 GDPR Art. 12(3) one-month response window.
base_days (CCPA/CPRA) int 45 CCPA §1798.130(a)(2) 45-day response window.
tolling_days int 0 Statutory extension already granted (e.g. CCPA +45 with notice); added to the base window.
kms_key_id str alias/dsr-intake-key Hardware-backed key per NIST SP 800-57; rotate per key-management policy.
dek_bit_length int 256 AES-256-GCM data-encryption key generated fresh per request.
retry_backoff str exponential Applied to ENCRYPTION_DLQ replays to avoid hammering a throttled KMS.

Verification

Confirm correctness before trusting the router in production. The critical assertions are that an under-confident payload fails closed to manual triage, and that a resolved jurisdiction produces the right statutory deadline.

from datetime import datetime, timezone

def test_low_confidence_fails_closed():
    """A payload with only a weak IP signal must not be auto-routed."""
    payload = DSRIntakePayload(consumer_email="a@example.com", request_type="access", raw_ip="203.0.113.7")
    jurisdiction, trail = resolve_primary_jurisdiction(payload)
    assert jurisdiction is Jurisdiction.UNKNOWN  # no signal above threshold

def test_gdpr_deadline_is_one_month_window():
    """A UK billing profile resolves to GDPR and gets the 30-day base window."""
    payload = DSRIntakePayload(consumer_email="b@example.com", request_type="deletion", billing_country="GB")
    jurisdiction, _ = resolve_primary_jurisdiction(payload)
    assert jurisdiction is Jurisdiction.GDPR
    deadline = calculate_sla_deadline(jurisdiction, datetime(2026, 1, 1, tzinfo=timezone.utc))
    assert deadline > payload.submitted_at

Expect the structured audit log to record, for every routed request, a correlation_id, the signal_matrix_snapshot (the returned audit trail), the jurisdiction_override_reason, sla_computed_at, and the routing_destination_queue. When debugging duplicate submissions or SLA drift, query the audit store by correlation ID: idempotency at the ingress gateway prevents double-processing, and every state transition is traceable across the distributed queues.

Troubleshooting

Under-confident payload routed to fulfillment instead of triage — Root cause: confidence_threshold set too low or a signal weight overestimated. Fix: keep the threshold at 0.60, and verify that IP geolocation stays at 0.50 so it can never alone clear the gate.

Signals conflict but the wrong statute wins — Root cause: tie-break not falling back to the strictest regime. Fix: confirm strictness_order lists GDPR first and that max(...) uses -strictness_order.index(...) so lower index (stricter) wins on equal confidence.

SLA deadline drifts or lands on a non-business day — Root cause: naive calendar-day arithmetic or a non-UTC submitted_at. Fix: normalize to UTC via astimezone(ZoneInfo("UTC")) before counting, and integrate a jurisdiction-specific holiday calendar (holidays/workalendar) rather than only skipping weekends. See Python zoneinfo documentation.

KMS throttling stalls intake — Root cause: synchronous KMS calls under burst load. Fix: catch the throttle, set kms_error on the envelope so route_with_fallback diverts to ENCRYPTION_DLQ, and replay with exponential backoff — never block the ingress path.

Duplicate submissions inflate the SLA clock or trigger redundant extraction — Root cause: missing idempotency check at the gateway. Fix: dedupe on request_id before routing so retries reuse the original correlation ID and never restart the deadline.