root
3ae1062d65
- Remove POST /api/auth/register (404); no sign-up UI; users are admin-managed - Add admin-only POST/PATCH/DELETE /api/auth/users (forced canonical roles, self-lockout + reference guards) - Unify role model in app/core/roles.py; reject unknown roles at creation and at login/JWT validation; startup normalizes unambiguous legacy aliases - Login rate limiting ~5 fails/15 min per IP+email -> 429 (in-process, tunable) - WhatsApp webhook requires X-Webhook-Secret; fail-closed when env unset; GET handshake uses constant-time verify token (403 on mismatch) - GET /api/whatsapp/mock-log now requires auth - Security headers middleware: X-Frame-Options DENY, nosniff, CSP on HTML, HSTS behind TLS - Pagination: limit alias for page_size, hard cap enforced, both -> 422
83 lines
2.7 KiB
Python
83 lines
2.7 KiB
Python
"""Dependency-light login rate limiter.
|
|
|
|
Brute-force protection for ``POST /api/auth/login``: a sliding window of
|
|
failed attempts keyed by ``ip|email``. Defaults to ~5 failures / 15 minutes
|
|
(env-tunable via ``LOGIN_RATE_LIMIT_MAX_ATTEMPTS`` /
|
|
``LOGIN_RATE_LIMIT_WINDOW_SECONDS``).
|
|
|
|
In-process storage is intentional: the app currently runs a single uvicorn
|
|
worker, and keeping the limiter dependency-light avoids pulling slowapi in
|
|
for one endpoint. ``_now`` is a module-level hook so tests can fast-forward
|
|
the clock.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from collections import defaultdict, deque
|
|
|
|
from fastapi import HTTPException, status
|
|
|
|
from app.core.config import settings
|
|
|
|
|
|
def _now() -> float:
|
|
"""Wall-clock epoch seconds; overridable in tests via monkeypatch."""
|
|
return time.time()
|
|
|
|
|
|
class LoginRateLimiter:
|
|
"""Sliding-window failure limiter keyed by ``ip|email``."""
|
|
|
|
def __init__(self, max_attempts: int = 5, window_seconds: int = 15 * 60) -> None:
|
|
self.max_attempts = max(max_attempts, 1)
|
|
self.window_seconds = max(window_seconds, 1)
|
|
self._failures: defaultdict[str, deque[float]] = defaultdict(deque)
|
|
|
|
def key(self, ip: str, email: str) -> str:
|
|
return f"{ip}|{email.strip().lower()}"
|
|
|
|
def _prune(self, key: str, now: float | None = None) -> None:
|
|
now = now if now is not None else _now()
|
|
window_start = now - self.window_seconds
|
|
bucket = self._failures.get(key)
|
|
if bucket is None:
|
|
return
|
|
while bucket and bucket[0] <= window_start:
|
|
bucket.popleft()
|
|
if not bucket:
|
|
self._failures.pop(key, None)
|
|
|
|
def failure_count(self, key: str) -> int:
|
|
self._prune(key)
|
|
return len(self._failures.get(key, ()))
|
|
|
|
def is_blocked(self, key: str) -> bool:
|
|
return self.failure_count(key) >= self.max_attempts
|
|
|
|
def record_failure(self, key: str) -> None:
|
|
self._failures[key].append(_now())
|
|
self._prune(key)
|
|
|
|
def clear(self, key: str) -> None:
|
|
self._failures.pop(key, None)
|
|
|
|
def reset(self) -> None:
|
|
self._failures.clear()
|
|
|
|
def check_or_raise(self, key: str) -> None:
|
|
"""Raise HTTP 429 when the key has exhausted its attempts."""
|
|
if self.is_blocked(key):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
|
|
detail="Too many failed login attempts. Try again later.",
|
|
)
|
|
|
|
|
|
# Shared instance — module import is safe because the app fails closed at
|
|
# boot (app/core/config.py) before any request can reach the login route.
|
|
login_rate_limiter = LoginRateLimiter(
|
|
max_attempts=settings.LOGIN_RATE_LIMIT_MAX_ATTEMPTS,
|
|
window_seconds=settings.LOGIN_RATE_LIMIT_WINDOW_SECONDS,
|
|
)
|