Rate Limiting
Rate limiting protects authentication endpoints from brute-force attacks and abuse. AuthFort uses a sliding window counter — no external dependencies required.
Enable Rate Limiting
Section titled “Enable Rate Limiting”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.
Default Limits
Section titled “Default Limits”| 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.
Custom Limits
Section titled “Custom Limits”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
),
)How It Works
Section titled “How It Works”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-Afterheader with seconds until the limit resets
Events
Section titled “Events”@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" |
Behind a Reverse Proxy
Section titled “Behind a Reverse Proxy”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.
Multi-Worker Deployments: Redis Store
Section titled “Multi-Worker Deployments: Redis Store”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:
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 redisfrom 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.
Custom Storage Backend
Section titled “Custom Storage Backend”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.