Agent Zero
604a2b7648
H1: Sanitize error endpoint context (strip tokens/passwords, limit depth/size) H2: Rate limiter IP spoofing fix (trusted proxy CIDR check for X-Forwarded-For) H3: CSRF middleware uses Redis singleton instead of per-request connection H4: WebSocket origin verification added to both kommunikation and ai_ui_control H5: Storage path traversal protection, get_url() returns relative URL not filesystem path H6: Security headers middleware (HSTS, X-Content-Type-Options, X-Frame-Options, CSP, Referrer-Policy) H7: Forward-repair migration 0045 for databases that ran original 0021/0027 Also: add trusted_proxy_cidrs to config, add verify_ws_origin to auth
77 lines
2.5 KiB
Python
77 lines
2.5 KiB
Python
"""Redis-based rate limiting for auth endpoints."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi import HTTPException, Request, status
|
|
|
|
from app.core.auth import get_redis
|
|
|
|
|
|
async def check_rate_limit(
|
|
redis_key: str,
|
|
max_attempts: int,
|
|
window_seconds: int,
|
|
) -> None:
|
|
"""Check rate limit using Redis INCR + EXPIRE.
|
|
Raises 429 if limit exceeded.
|
|
"""
|
|
redis = get_redis()
|
|
current = await redis.incr(redis_key)
|
|
if current == 1:
|
|
await redis.expire(redis_key, window_seconds)
|
|
if current > max_attempts:
|
|
ttl = await redis.ttl(redis_key)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
|
|
detail={
|
|
"detail": "Rate limit exceeded",
|
|
"code": "rate_limited",
|
|
"retry_after": ttl,
|
|
},
|
|
headers={"Retry-After": str(ttl)} if ttl > 0 else {},
|
|
)
|
|
|
|
|
|
async def reset_rate_limit(redis_key: str) -> None:
|
|
"""Reset a rate limit counter (e.g. on successful login)."""
|
|
redis = get_redis()
|
|
await redis.delete(redis_key)
|
|
|
|
|
|
def get_client_ip(request: Request) -> str:
|
|
"""Extract client IP from request.
|
|
|
|
Only trusts X-Forwarded-For if the direct client is a trusted proxy
|
|
(configured via TRUSTED_PROXY_CIDRS env var, comma-separated CIDRs).
|
|
This prevents IP spoofing to bypass rate limits.
|
|
"""
|
|
direct_ip = request.client.host if request.client else "unknown"
|
|
|
|
# Check if the direct client is a trusted proxy
|
|
from app.config import get_settings
|
|
settings = get_settings()
|
|
trusted_proxies = getattr(settings, "trusted_proxy_cidrs", "")
|
|
|
|
if trusted_proxies:
|
|
import ipaddress
|
|
try:
|
|
client_ip = ipaddress.ip_address(direct_ip)
|
|
for cidr in trusted_proxies.split(","):
|
|
cidr = cidr.strip()
|
|
if cidr and client_ip in ipaddress.ip_network(cidr, strict=False):
|
|
# Trusted proxy — use X-Forwarded-For
|
|
forwarded = request.headers.get("x-forwarded-for")
|
|
if forwarded:
|
|
# Use the leftmost (original client) IP
|
|
return forwarded.split(",")[0].strip()
|
|
# Fallback to X-Real-IP
|
|
real_ip = request.headers.get("x-real-ip")
|
|
if real_ip:
|
|
return real_ip.strip()
|
|
break
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
# Not a trusted proxy or no trusted proxies configured — use direct IP
|
|
return direct_ip
|