Implementing Step-Up Authentication for High-Impact DSRs

Within the broader Identity Verification & Attestation framework, this page builds the concrete high-impact path: the code that stops an erasure or portability request from executing until the requester has cleared a second authentication factor and, where warranted, proven identity with a short-lived government-ID check. The engineers who need this are privacy and platform teams whose intake already accepts requests but treats every one with the same trust level — a dangerous default, because an attacker who guesses a session or spoofs an email can trigger an irreversible deletion of someone else’s data. Step-up authentication raises the proof bar exactly for the operations where a wrong disclosure or deletion cannot be undone, while leaving low-impact requests fast. It sits one stage below the parent DSR Architecture & Intake Routing control plane and assumes a signed, schema-valid payload from the sibling Secure Intake Form Design.

The implementation flags high-impact requests, issues a TOTP step-up, accepts an ephemeral ID artifact, hashes it and purges the raw bytes, then records an attestation and emits the event that starts the statutory clock:

Prerequisites #

This implementation targets Python 3.11+ (for tuple[...] generics without from __future__ and the current datetime timezone API). Install the following:

• pydantic (v2) — strict validation of inputs and the immutable AttestationRecord. The code uses v2 idioms (model_config = ConfigDict(...), field_validator); v1 patterns (class Config, @validator) will not work.
• pyotp — RFC 6238 time-based one-time passcodes for the step-up factor.
• hashlib (stdlib) — SHA-256 digest of the verification artifact, a NIST SP 800-107 approved function.
• cryptography (optional) — for provisioning and encrypting the per-user TOTP secret at rest; the secret must never be stored in plaintext.

The TOTP shared secret is enrolled once per subject through a trusted channel and stored encrypted. This page owns only the step-up decision and attestation; it hands the attested request forward to the routing stage exactly as the parent cluster specifies.

Step-by-Step Implementation #

Step 1 — Classify the request and require step-up #

The classifier is a pure function that flags erasure and portability — the irreversible, high-disclosure operations — as high impact, along with anything touching special-category data. GDPR Art. 12(6) permits requesting additional information to confirm identity where the controller has reasonable doubts, and treating these request types as high impact is how that discretion becomes a consistent rule rather than a case-by-case judgement.

from enum import IntEnum

from pydantic import BaseModel, ConfigDict, Field


class Tier(IntEnum):
    STANDARD = 1
    HIGH_IMPACT = 2


HIGH_IMPACT_TYPES = frozenset({"erasure", "portability"})


class StepUpRequest(BaseModel):
    """Validated input to the step-up decision."""

    model_config = ConfigDict(strict=True, frozen=True)

    request_id: str = Field(pattern=r"^dsr_[0-9a-f]{32}$")
    request_type: str = Field(min_length=3, max_length=32)
    special_category: bool = False  # Art. 9 data in scope


def classify(req: StepUpRequest) -> Tier:
    """Flag erasure/portability and Art. 9 requests as high impact."""
    if req.request_type in HIGH_IMPACT_TYPES or req.special_category:
        return Tier.HIGH_IMPACT
    return Tier.STANDARD

A Tier.HIGH_IMPACT result means the request cannot proceed on session trust alone; it must clear the TOTP factor and the artifact check in the following steps.

Step 2 — Issue and verify the TOTP step-up #

The step-up factor is a time-based one-time passcode bound to the subject’s enrolled secret. Using pyotp, verification allows a small window for clock drift but no more, so a stale code cannot be replayed minutes later. The secret is loaded from encrypted storage per request and never logged.

import pyotp


def verify_totp(enrolled_secret: str, submitted_code: str) -> bool:
    """Verify a submitted TOTP against the subject's enrolled secret.

    `valid_window=1` tolerates one 30-second step of clock drift on either
    side; widening it weakens the factor, so keep it tight.
    """
    totp = pyotp.TOTP(enrolled_secret)
    return totp.verify(submitted_code, valid_window=1)

A failed verification never advances the request. The pipeline returns the requester to the challenge (subject to a retry cap) or rejects the attempt, and it records the failure — but it does not start the statutory clock, because identity is not yet attested.

Step 3 — Accept an ephemeral artifact, hash it, purge the raw bytes #

For the highest-assurance requests, TOTP is paired with a government-ID check. The critical discipline is that the document is ephemeral: it is compared to the account of record, a decision is recorded, and the raw bytes are hashed and discarded. Retaining the image would itself violate GDPR Art. 5(1)© data minimization, and an ID image is Art. 9 special-category-adjacent data, so keeping it converts a verification step into a standing liability.

import hashlib
from datetime import datetime, timezone

from pydantic import BaseModel, ConfigDict, Field


class AttestationRecord(BaseModel):
    """The only durable output of verification — no raw artifact retained."""

    model_config = ConfigDict(strict=True, frozen=True)

    request_id: str = Field(pattern=r"^dsr_[0-9a-f]{32}$")
    tier: int = Field(ge=1, le=2)
    method: str = Field(min_length=3, max_length=48)
    artifact_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
    matched: bool
    attested_at: datetime


def attest_artifact(request_id: str, artifact: bytes, matched: bool) -> AttestationRecord:
    """Hash the ephemeral ID artifact, then let its raw bytes fall out of scope.

    `artifact` is passed in memory from the ephemeral upload buffer. It is
    hashed here and must not be written to disk, cache, or logs anywhere.
    """
    digest = hashlib.sha256(artifact).hexdigest()
    return AttestationRecord(
        request_id=request_id,
        tier=int(Tier.HIGH_IMPACT),
        method="totp+gov_id",
        artifact_sha256=digest,
        matched=matched,
        attested_at=datetime.now(timezone.utc),
    )

The AttestationRecord carries the hash and the match decision — enough to prove an artifact was checked, and nothing that reconstructs the document. The upload buffer that held artifact is a short-TTL ephemeral store; once this function returns, the raw bytes exist nowhere in the system.

Step 4 — Emit the attestation event and start the clock #

A successful attestation is the event that anchors every downstream deadline. GDPR Art. 12(3) sets a one-month response window, but Art. 12(6) lets the controller pause for identity confirmation, so the clock starts at attestation, not submission. The event is emitted to the audit layer with the hash and decision — never the raw artifact.

def run_step_up(req: StepUpRequest, enrolled_secret: str, submitted_code: str,
                artifact: bytes, id_matches: bool, emit) -> AttestationRecord | None:
    """Drive the full high-impact step-up and emit the attestation event.

    `emit` is an audit sink callable. Returns the AttestationRecord on success,
    or None if the request is not high impact or the TOTP fails.
    """
    if classify(req) is not Tier.HIGH_IMPACT:
        return None  # standard tier handled elsewhere
    if not verify_totp(enrolled_secret, submitted_code):
        emit({"request_id": req.request_id, "event": "attestation_failed",
              "reason": "totp_invalid"})
        return None
    record = attest_artifact(req.request_id, artifact, id_matches)
    if not record.matched:
        emit({"request_id": req.request_id, "event": "attestation_failed",
              "reason": "id_mismatch"})
        return None
    emit({"request_id": record.request_id, "event": "attested",
          "tier": record.tier, "method": record.method,
          "artifact_sha256": record.artifact_sha256,
          "attested_at": record.attested_at.isoformat()})
    return record

The emitted attested event flows to the Audit Logging & Compliance Evidence layer, and its timestamp is what the SLA stage anchors the deadline to. Note that no branch of run_step_up ever puts the raw artifact into the audit sink.

Configuration Reference #

Verification #

Two assertions matter most: that a valid TOTP verifies and an invalid one does not, and that no raw ID artifact survives attestation. The first proves the factor works; the second proves the data-minimization guarantee.

import hashlib

import pyotp


def test_totp_verifies_current_code():
    secret = pyotp.random_base32()
    code = pyotp.TOTP(secret).now()
    assert verify_totp(secret, code) is True
    assert verify_totp(secret, "000000") is False


def test_raw_artifact_is_not_retained():
    artifact = b"fake-id-image-bytes"
    req = StepUpRequest(request_id="dsr_" + "a" * 32, request_type="erasure")
    record = attest_artifact(req.request_id, artifact, matched=True)
    # Only the hash survives; the serialized record never carries raw bytes.
    assert record.artifact_sha256 == hashlib.sha256(artifact).hexdigest()
    assert "fake-id-image-bytes" not in record.model_dump_json()
    assert record.matched is True

Expect a passing suite to confirm three things: classify sends erasure and portability to Tier 2, verify_totp accepts only a live code, and attest_artifact yields a record whose serialization contains the hash but never the source bytes. Run the non-retention assertion in CI so a future refactor that “temporarily” stashes the image fails the build rather than shipping a violation.

Troubleshooting #

TOTP always fails despite the user entering the right code — Root cause: server/device clock skew beyond the validation window, or a mismatched enrolled secret. Fix: sync the server clock via NTP, confirm the stored secret matches enrollment, and keep valid_window=1 rather than widening it to mask the drift.

Raw ID image appears in logs or object storage — Root cause: the upload was written or logged before hashing, or a debug handler captured the request body. Fix: hash in the same scope the bytes arrive, remove any body-logging middleware on the upload route, and assert non-retention in CI as shown above — this is an Art. 5(1)© violation, not a cosmetic bug.

Standard requests are being forced through step-up — Root cause: classify over-matching, or special_category set spuriously. Fix: verify HIGH_IMPACT_TYPES contains only erasure and portability and that the special-category flag is derived from actual data scope, so low-impact requests stay fast per the Art. 12(2) duty to facilitate rights.

The SLA clock starts before identity is proven — Root cause: the deadline is anchored to submission instead of the attested event. Fix: start the clock only from AttestationRecord.attested_at on a matched record, consistent with the parent Identity Verification & Attestation stage and Art. 12(6).

TOTP endpoint abused for brute force — Root cause: no retry cap or rate limit on code submission. Fix: enforce totp_retry_cap lockout per request and rate-limit issuance per account so the step-up factor cannot be guessed or used as an oracle.

Frequently Asked Questions #

Related #

• Identity Verification & Attestation — the parent stage defining risk-based tiers, hashing, and clock start.
• DSR Architecture & Intake Routing — the intake control plane this step-up path sits within.
• Secure Intake Form Design — the signed, schema-valid payload this flow consumes.
• 30-Day vs 45-Day SLA Mapping — how the attestation timestamp becomes a statutory deadline.
• Audit Logging & Compliance Evidence — where the emitted attestation event is recorded as tamper-evident evidence.