Rate limiting¶

@classmethod
def disabled(cls) -> Self:
    """Build a preset that disables rate limiting for every auth endpoint.

    Returns:
        New config with every supported auth slot left unset.
    """
    return cls()

@classmethod
def from_shared_backend(
    cls,
    backend: RateLimiterBackend,
    *,
    options: SharedRateLimitConfigOptions | None = None,
) -> Self:
    """Build endpoint-specific limiters from the package-owned shared-backend recipe.

    The builder uses the private endpoint catalog for default scopes and namespace
    tokens, then applies override precedence in this order:

    1. ``backend`` for every enabled slot
    2. ``group_backends`` for slot groups such as ``totp`` or ``verification``
    3. ``endpoint_overrides`` for full slot replacement or explicit ``None`` disablement

    Args:
        backend: Default limiter backend for enabled auth slots.
        options: Optional shared builder options. Use
            ``SharedRateLimitConfigOptions(disabled={...})`` to leave built-in verification routes disabled.

    Returns:
        New config populated from the shared-backend builder inputs.
    """
    resolved_options = options or SharedRateLimitConfigOptions()
    group_backend_map: dict[AuthRateLimitEndpointGroup, RateLimiterBackend] = dict(
        resolved_options.group_backends or {},
    )
    endpoint_override_map = dict(resolved_options.endpoint_overrides or {})
    config_kwargs = dict(
        _iter_auth_rate_limit_config_items(
            _AuthRateLimitConfigItems(
                catalog=_SLOT_CATALOG,
                backend=backend,
                enabled=resolved_options.enabled,
                disabled=resolved_options.disabled,
                group_backends=group_backend_map,
                endpoint_overrides=endpoint_override_map,
                endpoint_type=cls._endpoint_rate_limit_type,
                trusted_proxy=resolved_options.trusted_proxy,
                identity_fields=resolved_options.identity_fields,
                trusted_headers=resolved_options.trusted_headers,
            ),
        ),
    )
    return cls(**cast("Any", config_kwargs))

@classmethod
def lenient(cls, *, backend: RateLimiterBackend) -> Self:
    """Build a lenient preset for internal or low-risk deployments.

    The supplied backend sets the broader budget used for the lower-risk
    login, change-password, refresh, and registration surfaces. Token- and secret-bearing
    flows still receive a stricter built-in limiter clone capped at five
    attempts per window so reset, verification, and TOTP endpoints do not
    inherit an overly permissive budget.

    Args:
        backend: Built-in shared backend instance for the lenient preset.

    Returns:
        New config with route-style namespaces for every supported auth slot.
    """
    strict_backend = _clone_backend_with_capped_attempts(
        backend,
        max_attempts_cap=_LENIENT_STRICT_MAX_ATTEMPTS_CAP,
    )
    config_kwargs = {
        recipe.slot.value: cls._endpoint_rate_limit_type(
            backend=backend if recipe.slot in _LENIENT_AUTH_RATE_LIMIT_SHARED_SLOTS else strict_backend,
            scope=recipe.default_scope,
            namespace=recipe.default_namespace,
        )
        for recipe in _SLOT_RECIPES
    }
    return cls(**cast("Any", config_kwargs))

@classmethod
def strict(cls, *, backend: RateLimiterBackend) -> Self:
    """Build a strict preset for public-facing sign-in surfaces.

    The provided backend should already be configured with the lower attempt
    budget you want to enforce. This preset wires that shared backend to the
    highest-risk credential entry points: login, change-password, register,
    and TOTP verify.

    Args:
        backend: Shared backend instance for the strict preset slots.

    Returns:
        New config with only the strict preset slots enabled.
    """
    config_kwargs = {
        recipe.slot.value: cls._endpoint_rate_limit_type(
            backend=backend,
            scope=recipe.default_scope,
            namespace=recipe.default_namespace,
        )
        for recipe in _SLOT_RECIPES
        if recipe.slot in _STRICT_AUTH_RATE_LIMIT_PRESET_SLOTS
    }
    return cls(**cast("Any", config_kwargs))

async def before_request(self, request: Request[Any, Any, Any]) -> None:
    """Reject the request with 429 when its key is over the configured limit.

    Security:
        Only set ``trusted_proxy=True`` when this service is behind a trusted
        proxy or load balancer that overwrites client IP headers. Otherwise,
        attackers can spoof headers like ``X-Forwarded-For`` and evade or
        poison rate-limiting keys.

    Raises:
        TooManyRequestsException: If the request exceeded the configured limit.
    """
    key = await self.build_key(request)
    if await self.backend.check(key):
        return

    retry_after = await self.backend.retry_after(key)
    logger.warning(
        "Rate limit exceeded",
        extra={
            "event": "rate_limit_triggered",
            "namespace": self.namespace,
            "scope": self.scope,
            "trusted_proxy": self.trusted_proxy,
        },
    )
    msg = "Too many requests."
    raise TooManyRequestsException(
        detail=msg,
        headers={"Retry-After": str(max(retry_after, 1))},
    )

async def build_key(self, request: Request[Any, Any, Any]) -> str:
    """Build the backend key for the given request.

    Returns:
        Namespaced rate-limit key for the request.
    """
    host = _client_host(request, trusted_proxy=self.trusted_proxy, trusted_headers=self.trusted_headers)
    parts = [self.namespace, _safe_key_part(host)]
    if self.scope == "ip_email":
        email = await _extract_email(request, identity_fields=self.identity_fields)
        if email:
            parts.append(_safe_key_part(email))
    if self.scope == "api_key_id":
        key_id = _extract_api_key_id(request)
        if key_id:
            parts.append(_safe_key_part(key_id))

    return ":".join(parts)

async def increment(self, request: Request[Any, Any, Any]) -> None:
    """Record a failed or rate-limited attempt for the current request."""
    await self.backend.increment(await self.build_key(request))

async def reset(self, request: Request[Any, Any, Any]) -> None:
    """Clear stored attempts for the current request key."""
    await self.backend.reset(await self.build_key(request))

async def check(self, key: str) -> bool:
    """Return whether ``key`` can perform another attempt."""
    async with self._lock:
        now = read_clock(self._clock)
        self._maybe_sweep(now)
        timestamps = self._prune(key, now)
        if timestamps is None:
            if self._is_at_capacity_after_prune(now):
                self._log_capacity_rejection()
                return False
            return True

        return len(timestamps) < self.max_attempts

async def increment(self, key: str) -> None:
    """Record a new attempt for ``key`` in the current window."""
    async with self._lock:
        now = read_clock(self._clock)
        self._maybe_sweep(now)
        timestamps = self._prune(key, now)
        if timestamps is None:
            if self._is_at_capacity_after_prune(now):
                self._log_capacity_rejection()
                return
            timestamps = deque()
            self._windows[key] = timestamps

        timestamps.append(now)

async def reset(self, key: str) -> None:
    """Clear the in-memory counter for ``key``."""
    async with self._lock:
        self._windows.pop(key, None)

async def retry_after(self, key: str) -> int:
    """Return the remaining block duration for ``key`` in whole seconds."""
    async with self._lock:
        now = read_clock(self._clock)
        timestamps = self._prune(key, now)
        if timestamps is None or len(timestamps) < self.max_attempts:
            return 0

        oldest_timestamp = timestamps[0]
        remaining = self.window_seconds - (now - oldest_timestamp)
        return max(math.ceil(remaining), 1)

async def check(self, key: str) -> bool:
    """Return whether ``key`` can perform another attempt."""
    count = self._decode_integer(
        await self._eval(
            self._CHECK_SCRIPT,
            key,
            read_clock(self._clock),
            self.window_seconds,
            self.max_attempts,
        ),
    )
    return count < self.max_attempts

async def increment(self, key: str) -> None:
    """Record a new attempt for ``key`` atomically in Redis."""
    now = read_clock(self._clock)
    await self._eval(
        self._INCREMENT_SCRIPT,
        key,
        now,
        self.window_seconds,
        f"{now:.9f}:{uuid4().hex}",
        self._ttl_seconds,
    )

async def reset(self, key: str) -> None:
    """Delete the Redis sorted set for ``key``."""
    await self.redis.delete(self._key(key))

async def retry_after(self, key: str) -> int:
    """Return the remaining block duration for ``key`` in whole seconds."""
    return max(
        self._decode_integer(
            await self._eval(
                self._RETRY_AFTER_SCRIPT,
                key,
                read_clock(self._clock),
                self.window_seconds,
                self.max_attempts,
            ),
        ),
        0,
    )

async def check(self, key: str) -> bool:
    """Return whether another attempt is allowed for ``key``."""

async def increment(self, key: str) -> None:
    """Record an attempt for ``key``."""

async def reset(self, key: str) -> None:
    """Clear tracked attempts for ``key``."""

async def retry_after(self, key: str) -> int:
    """Return the number of seconds until ``key`` can try again."""

async def before_request(self, endpoint: TotpSensitiveEndpoint, request: Request[Any, Any, Any]) -> None:
    """Run endpoint-specific before-request checks."""
    if limiter := self._limiters.get(endpoint):
        await limiter.before_request(request)

async def on_account_state_failure(self, endpoint: TotpSensitiveEndpoint, request: Request[Any, Any, Any]) -> None:
    """Apply endpoint-specific account-state failure behavior."""
    if endpoint in self._ACCOUNT_STATE_RESET_ENDPOINTS and (limiter := self._limiters.get(endpoint)):
        await limiter.reset(request)

async def on_invalid_attempt(self, endpoint: TotpSensitiveEndpoint, request: Request[Any, Any, Any]) -> None:
    """Record endpoint-specific invalid attempt failures."""
    if limiter := self._limiters.get(endpoint):
        await limiter.increment(request)

async def on_success(self, endpoint: TotpSensitiveEndpoint, request: Request[Any, Any, Any]) -> None:
    """Apply endpoint-specific success behavior."""
    if limiter := self._limiters.get(endpoint):
        await limiter.reset(request)