TOTP¶

def build_recovery_code_index(
    codes: tuple[str, ...],
    *,
    lookup_secret: bytes,
    password_helper: PasswordHelper | None = None,
) -> dict[str, str]:
    """Build a keyed lookup index for TOTP recovery-code hashes.

    Returns:
        Mapping of HMAC-SHA-256 lookup hex digests to Argon2 hashes.
    """
    helper = password_helper or PasswordHelper.from_defaults()
    return {
        _recovery_code_lookup_hex(code, lookup_secret=lookup_secret): helper.hash(code.casefold()) for code in codes
    }

def generate_totp_recovery_codes(*, count: int = DEFAULT_TOTP_RECOVERY_CODE_COUNT) -> tuple[str, ...]:
    """Return distinct plaintext recovery codes for a TOTP enrollment.

    Returns:
        A tuple of unique 112-bit hex recovery codes.

    Raises:
        ValueError: If ``count`` is negative.
    """
    if count < 0:
        msg = "Recovery-code count cannot be negative."
        raise ValueError(msg)

    codes: set[str] = set()
    while len(codes) < count:
        codes.add(secrets.token_hex(TOTP_RECOVERY_CODE_HEX_BYTES))
    return tuple(codes)

def generate_totp_secret(algorithm: TotpAlgorithm = TOTP_ALGORITHM) -> str:
    """Generate a base32-encoded TOTP secret sized to the algorithm's HMAC output.

    Per RFC 4226 Section 4, the shared secret length should match the HMAC
    output length: 32 bytes for SHA-256 or 64 bytes for SHA-512.

    Args:
        algorithm: TOTP hash algorithm; determines secret byte length.

    Returns:
        A random base32 secret without RFC padding.
    """
    secret_bytes = _SECRET_BYTES_BY_ALGORITHM[_validate_totp_algorithm(algorithm)]
    random_bytes = secrets.token_bytes(secret_bytes)
    return base64.b32encode(random_bytes).decode("ascii").rstrip("=")

def generate_totp_uri(
    secret: str,
    email: str,
    issuer: str,
    *,
    algorithm: TotpAlgorithm = TOTP_ALGORITHM,
) -> str:
    """Build an otpauth URI suitable for QR-code generation.

    Returns:
        An ``otpauth://`` URI for authenticator apps.
    """
    algorithm = _validate_totp_algorithm(algorithm)
    label = quote(f"{issuer}:{email}")
    query_params: dict[str, str] = {
        "secret": secret,
        "issuer": issuer,
        "digits": str(TOTP_DIGITS),
        "period": str(TIME_STEP_SECONDS),
        "algorithm": algorithm,
    }
    query = urlencode(query_params)
    return f"otpauth://totp/{label}?{query}"

def verify_totp(secret: str, code: str, *, algorithm: TotpAlgorithm = TOTP_ALGORITHM) -> bool:
    """Validate a TOTP code for the current time window only.

    Returns:
        ``True`` when the code matches the current time step, otherwise ``False``.
    """
    return _verify_totp_counter(secret, code, algorithm=algorithm) is not None

async def verify_totp_with_store(
    secret: str,
    code: str,
    *,
    replay: TotpReplayProtection,
    algorithm: TotpAlgorithm = TOTP_ALGORITHM,
) -> bool:
    """Validate a TOTP code and optionally reject same-window replays.

    Returns:
        ``True`` when the code is valid and has not already been used for ``replay.user_id``.

    Raises:
        ConfigurationError: If replay protection is required and no replay store is configured
            outside testing mode.
    """
    counter = _verify_totp_counter(secret, code, algorithm=algorithm)
    if counter is None:
        logger.warning("TOTP verification failed.", extra={"event": "totp_failed", "user_id": str(replay.user_id)})
        return False

    if replay.used_tokens_store is None:
        if replay.require_replay_protection and not replay.unsafe_testing:
            msg = "TOTP replay protection is required in production. Configure a UsedTotpCodeStore."
            raise ConfigurationError(msg)
        warnings.warn(
            "TOTP replay protection is DISABLED because used_tokens_store=None.",
            SecurityWarning,
            stacklevel=2,
        )
        return True

    mark_result = await replay.used_tokens_store.mark_used(replay.user_id, counter, USED_TOTP_CODE_TTL_SECONDS)
    if mark_result.stored:
        return True
    if mark_result.rejected_as_replay:
        logger.warning("TOTP replay detected.", extra={"event": "totp_replay", "user_id": str(replay.user_id)})
    else:
        logger.warning(
            "TOTP used-code store rejected verification under capacity pressure (fail closed).",
            extra={"event": "totp_replay_store_capacity", "user_id": str(replay.user_id)},
        )
    return False

def build_pending_totp_client_binding(
    request: Request[Any, Any, Any],
    *,
    pending_secret: str,
    trusted_proxy: bool = False,
    trusted_headers: tuple[str, ...] = ("X-Forwarded-For",),
) -> PendingTotpClientBinding:
    """Return keyed client-IP and User-Agent fingerprints for a TOTP pending token."""
    client_ip = _client_host(request, trusted_proxy=trusted_proxy, trusted_headers=trusted_headers)
    user_agent = (request.headers.get("User-Agent") or request.headers.get("user-agent") or "")[
        :_USER_AGENT_FINGERPRINT_MAX_BYTES
    ]
    key = pending_secret.encode()
    return PendingTotpClientBinding(
        client_ip_fingerprint=_fingerprint_client_binding_value(client_ip, key=key),
        user_agent_fingerprint=_fingerprint_client_binding_value(user_agent, key=key),
    )