Skip to content

Rate Limiting

Rate limiting protects authentication endpoints from brute-force attacks and abuse. AuthFort uses a sliding window counter — no external dependencies required.

Pass RateLimitConfig to the constructor. All endpoints get sensible defaults.

from authfort import AuthFort, RateLimitConfig

auth = AuthFort(
    database_url="postgresql+asyncpg://...",
    rate_limit=RateLimitConfig(),
)
from authfort import AuthFort, RateLimitConfig

auth = AuthFort(
    database_url="postgresql+asyncpg://...",
    rate_limit=RateLimitConfig(),
)

That’s it. All 8 auth endpoints are now rate limited.

Endpoint Default Limit Type
login 5/min IP + email
signup 3/min IP + email
magic_link 5/min IP + email
otp 5/min IP + email
verify_email 5/min IP only
refresh 30/min IP only
oauth_authorize 10/min IP only

Format: "{count}/{period}" — period can be sec, min, hour, or day.

Override specific endpoints or disable them with None.

auth = AuthFort(
    database_url="postgresql+asyncpg://...",
    rate_limit=RateLimitConfig(
        login="10/min",
        signup="5/min",
        refresh=None,  # disable for refresh
    ),
)
auth = AuthFort(
    database_url="postgresql+asyncpg://...",
    rate_limit=RateLimitConfig(
        login="10/min",
        signup="5/min",
        refresh=None,  # disable for refresh
    ),
)

Rate limiting uses two strategies:

  • IP-based — applied to all endpoints. Prevents a single IP from flooding.
  • Email-based — applied to login, signup, magic link, and OTP. Catches distributed attacks targeting the same account from multiple IPs.

Both must pass for the request to proceed.

When a limit is exceeded, AuthFort returns:

  • 429 Too Many Requests status code
  • Retry-After header with seconds until the limit resets
@auth.on("rate_limit_exceeded")
async def on_rate_limit(event):
    print(f"{event.key_type} limit hit on {event.endpoint}: {event.ip_address}")
@auth.on("rate_limit_exceeded")
async def on_rate_limit(event):
    print(f"{event.key_type} limit hit on {event.endpoint}: {event.ip_address}")

The rate_limit_exceeded event fires on every rejected request with:

Field Type Description
endpoint str Which endpoint was hit
ip_address str | None Client IP
email str | None Email (for email-based limits)
limit str The limit that was exceeded
key_type str "ip" or "email"

When AuthFort runs behind a reverse proxy (nginx, traefik, Docker), request.client.host returns the proxy’s internal IP. Without proxy configuration, all users share a single rate limit bucket.

Option 1: Trust all proxies (simple, for single-proxy setups)

auth = AuthFort(
database_url="...",
trust_proxy=True,
rate_limit=RateLimitConfig(),
)

Option 2: Trust specific proxies (recommended for production)

auth = AuthFort(
database_url="...",
trusted_proxies=["172.18.0.0/16"],
rate_limit=RateLimitConfig(),
)

AuthFort reads X-Forwarded-For (leftmost value) then falls back to X-Real-IP. If neither header is present, request.client.host is used.

See Configuration for details on trust_proxy and trusted_proxies.

The default InMemoryStore is per-process. If you run multiple workers (uvicorn --workers 4, gunicorn, multiple replicas behind a load balancer), each process keeps its own buckets — so every configured limit is effectively multiplied by the worker count. Since rate limiting is the control that protects login and signup against brute force and enumeration probing, use the shared Redis store in any multi-process deployment:

Terminal window
uv add "authfort[redis]"
from authfort import AuthFort, RateLimitConfig, RedisRateLimitStore
auth = AuthFort(
database_url="...",
rate_limit=RateLimitConfig(),
rate_limit_store=RedisRateLimitStore.from_url("redis://localhost:6379"),
)

You can also pass a pre-configured client (e.g. one you already use elsewhere, or with TLS/cluster options):

import redis.asyncio as redis
from authfort import RedisRateLimitStore
client = redis.from_url("rediss://:password@redis.internal:6380")
store = RedisRateLimitStore(client, key_prefix="myapp:rl:")

The store uses one Redis sorted set per key with the same sliding-window semantics as InMemoryStore. It fails closed: if Redis is unreachable, rate-limited endpoints error rather than silently allowing unlimited traffic. Keep app-server clocks NTP-synced — timestamps come from the app processes.

For other backends, implement the RateLimitStore protocol. hit() and reset() may be sync or async — async implementations are awaited automatically.

class MyStore:
def hit(self, key, limit):
# Returns: (allowed: bool, remaining: int, retry_after: float)
...
def reset(self, key=None):
...
auth = AuthFort(..., rate_limit=RateLimitConfig(), rate_limit_store=MyStore())

See Server Config for the full RateLimitStore protocol.