Rate Limiting

Rate limiting is a critical security measure for authentication systems. Without it, attackers can brute-force passwords, flood OTP endpoints, and abuse password reset flows. AllSafe Fast includes a built-in rate limiter with two backends: Redis for production and in-memory for development.

Why Rate Limiting Matters for Auth

Authentication endpoints are prime targets for abuse. The specific threats rate limiting addresses include:

Sliding Window Algorithm

AllSafe Fast uses a sliding window algorithm for rate limiting. Unlike a fixed window counter (which allows bursts at window boundaries), the sliding window tracks the actual number of requests in the trailing time window:

# Sliding window example with limit=5, window=900s (15 minutes)
#
# Timeline (timestamps in seconds):
#   t=0:   Request 1 -> count=1 -> ALLOW
#   t=100: Request 2 -> count=2 -> ALLOW
#   t=200: Request 3 -> count=3 -> ALLOW
#   t=300: Request 4 -> count=4 -> ALLOW
#   t=400: Request 5 -> count=5 -> ALLOW
#   t=500: Request 6 -> count=6 -> DENY (6 > 5)
#   t=950: Request 7 -> count=6 -> DENY (requests 1-6 still in window)
#   t=901: Request 8 -> count=5 -> ALLOW (request 1 expired, count=5)
#
# The window "slides" — it always covers the last 900 seconds

Redis Backend (Production)

The Redis rate limiter uses sorted sets and an atomic pipeline for accurate, distributed rate limiting. This is the recommended backend for production:

How It Works

The Redis backend uses four Redis operations executed as an atomic pipeline:

import time
import redis.asyncio as redis

class RedisRateLimiter:
    def __init__(self, redis_client, default_limit=5, default_window=900):
        self.redis = redis_client
        self.default_limit = default_limit
        self.default_window = default_window

    async def check(self, key: str, limit=None, window=None) -> bool:
        limit = limit or self.default_limit
        window = window or self.default_window
        now = time.time()
        window_start = now - window

        # Atomic pipeline — all operations execute together
        async with self.redis.pipeline() as pipe:
            pipe.zremrangebyscore(key, 0, window_start)  # Remove expired
            pipe.zadd(key, {str(now): now})             # Add current request
            pipe.zcard(key)                             # Count in window
            pipe.expire(key, int(window))                # Set TTL
            results = await pipe.execute()

        count = results[2]  # zcard result
        return count <= limit

In-Memory Backend (Development)

The in-memory rate limiter uses a defaultdict(list) to track request timestamps. It implements the same sliding window algorithm but stores data in process memory:

import time
from collections import defaultdict

class InMemoryRateLimiter:
    def __init__(self, default_limit=5, default_window=900):
        self.default_limit = default_limit
        self.default_window = default_window
        self._requests: dict[str, list[float]] = defaultdict.__call__(list)

    async def check(self, key: str, limit=None, window=None) -> bool:
        limit = limit or self.default_limit
        window = window or self.default_window
        now = time.time()
        window_start = now - window

        # Remove expired timestamps
        self._requests[key] = [
            ts for ts in self._requests[key] if ts > window_start
        ]

        # Check limit
        if len(self._requests[key]) >= limit:
            return False

        # Add current request
        self._requests[key].append(now)
        return True

Configuration

Rate limiting is configured through environment variables or the AuthConfig constructor:

# Via environment variables
# export ALLSAFE_RATE_LIMIT=5
# export ALLSAFE_RATE_LIMIT_WINDOW=900
# export ALLSAFE_REDIS_URL=redis://localhost:6379/0

# Via AuthConfig
config = AuthConfig(
    rate_limit=5,           # 5 requests per window
    rate_limit_window=900,    # 15-minute window
    redis_url="redis://localhost:6379/0",  # Use Redis backend
)

RateLimiter API

The RateLimiter class provides three methods for checking and managing rate limits:

from allsafe_fast import RateLimiter

# RateLimiter is created internally by Auth, but you can use it directly
limiter = RateLimiter(
    redis=redis_client,       # Redis client or None for in-memory
    default_limit=5,
    default_window=900,
)

# Check if a request is allowed (returns bool)
allowed = await limiter.check(key="sign_in:alice@example.com")
if not allowed:
    raise RateLimitError(retry_after=900)

# Check with custom limit and window
allowed = await limiter.check(
    key="otp:bob@example.com",
    limit=3,      # Stricter limit for OTP
    window=3600,   # 1-hour window
)

# Get remaining requests in the window (returns int)
remaining = await limiter.remaining(key="sign_in:alice@example.com")
# remaining = 3 (out of 5 allowed)

# Reset the rate limit for a key (clears all requests)
await limiter.reset(key="sign_in:alice@example.com")

Protecting Auth Endpoints

AllSafe Fast automatically applies rate limiting to sensitive authentication endpoints. Here is how different endpoints are protected:

Sign-In Endpoint

# Rate limited by email + IP address
# Key: "sign_in:{email}:{ip}"
# Default: 5 attempts per 15 minutes
#
# After 5 failed attempts, the endpoint returns 429 Too Many Requests
# with a Retry-After header indicating when to try again

@app.post("/auth/sign-in")
async def sign_in(request, email: str, password: str):
    try:
        result = await auth.sign_in_with_password(email, password)
        # Reset rate limit on successful sign-in
        await auth.rate_limiter.reset(f"sign_in:{email}:{request.client.host}")
        return result
    except RateLimitError as e:
        raise HTTPException(
            status_code=429,
            detail="Too many attempts. Try again later.",
            headers={"Retry-After": str(e.retry_after)},
        )

Email OTP and Magic Link Endpoints

# Rate limited by email address
# Key: "otp:{email}" or "magic_link:{email}"
# Recommended: 3 requests per hour (stricter than sign-in)
#
# This prevents OTP flooding and email spam

@app.post("/auth/otp/request")
async def request_otp(email: str):
    key = f"otp:{email}"
    allowed = await auth.rate_limiter.check(key, limit=3, window=3600)
    if not allowed:
        raise RateLimitError(retry_after=3600)
    await auth.send_otp(email)
    return {"message": "OTP sent"}

Password Reset Endpoint

# Rate limited by email address
# Key: "password_reset:{email}"
# Recommended: 3 requests per hour
#
# Prevents reset email spam and token enumeration

@app.post("/auth/password-reset/request")
async def request_reset(email: str):
    key = f"password_reset:{email}"
    allowed = await auth.rate_limiter.check(key, limit=3, window=3600)
    if not allowed:
        raise RateLimitError(retry_after=3600)
    await auth.request_password_reset(email)
    return {"message": "If the email is registered, a reset link has been sent"}

RateLimitError

When a rate limit is exceeded, AllSafe Fast raises a RateLimitError with a retry_after attribute indicating how many seconds the client should wait before retrying:

from allsafe_fast import RateLimitError

try:
    await auth.sign_in_with_password(email, password)
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
    # e.retry_after -> 900 (seconds until the window resets)