security: add brute-force rate limiting to /login
5 failed attempts per IP within 5 minutes triggers 15-minute block. Counter resets on successful login. State is per-worker (in-memory).
This commit is contained in:
+42
@@ -100,3 +100,45 @@ def has_access(db: Session, user_id: int, service_id: int) -> bool:
|
||||
UserServiceAccess.service_id == service_id,
|
||||
)
|
||||
return db.scalar(q) is not None
|
||||
|
||||
import threading
|
||||
import time
|
||||
|
||||
_login_attempts: dict = {}
|
||||
_login_lock = threading.Lock()
|
||||
|
||||
_LOGIN_MAX_ATTEMPTS = 5
|
||||
_LOGIN_WINDOW = 300 # сброс счётчика через 5 минут
|
||||
_LOGIN_BLOCK = 900 # блокировка на 15 минут
|
||||
|
||||
|
||||
def check_login_rate_limit(ip: str) -> Optional[float]:
|
||||
"""Возвращает blocked_until (time.monotonic) если IP заблокирован, иначе None."""
|
||||
now = time.monotonic()
|
||||
with _login_lock:
|
||||
entry = _login_attempts.get(ip)
|
||||
if not entry:
|
||||
return None
|
||||
if entry["blocked_until"] > now:
|
||||
return entry["blocked_until"]
|
||||
if now - entry["first"] > _LOGIN_WINDOW:
|
||||
del _login_attempts[ip]
|
||||
return None
|
||||
|
||||
|
||||
def record_login_failure(ip: str) -> None:
|
||||
now = time.monotonic()
|
||||
with _login_lock:
|
||||
entry = _login_attempts.get(ip)
|
||||
if not entry or now - entry["first"] > _LOGIN_WINDOW:
|
||||
_login_attempts[ip] = {"count": 1, "first": now, "blocked_until": 0.0}
|
||||
entry = _login_attempts[ip]
|
||||
else:
|
||||
entry["count"] += 1
|
||||
if entry["count"] >= _LOGIN_MAX_ATTEMPTS:
|
||||
entry["blocked_until"] = now + _LOGIN_BLOCK
|
||||
|
||||
|
||||
def record_login_success(ip: str) -> None:
|
||||
with _login_lock:
|
||||
_login_attempts.pop(ip, None)
|
||||
|
||||
Reference in New Issue
Block a user