diff --git a/app/auth.py b/app/auth.py index ff4eac3..e6b0dbf 100644 --- a/app/auth.py +++ b/app/auth.py @@ -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) diff --git a/app/main.py b/app/main.py index eaab847..38b5baa 100644 --- a/app/main.py +++ b/app/main.py @@ -42,6 +42,9 @@ from utils import ( from auth import ( get_current_user, has_access, issue_auth_cookie, issue_csrf_cookie, require_admin, require_user, user_is_valid, validate_csrf, verify_password, hash_password, + check_login_rate_limit, + record_login_failure, + record_login_success, ) from runtime import ( acquire_universal_slot, acquire_web_pool_slot, allocator_lock, @@ -1160,8 +1163,13 @@ def login( if not cookie_csrf or csrf_token != cookie_csrf: raise HTTPException(status_code=403, detail="CSRF failed") + ip = _get_real_ip(request) + if check_login_rate_limit(ip): + raise HTTPException(status_code=429, detail="Слишком много попыток входа. Попробуйте через 15 минут.") + user = db.scalar(select(User).where(User.username == username)) if not user or not verify_password(password, user.password_hash): + record_login_failure(ip) csrf = request.cookies.get(CSRF_COOKIE) or secrets.token_urlsafe(24) response = templates.TemplateResponse( "login.html", @@ -1189,6 +1197,7 @@ def login( response.set_cookie(CSRF_COOKIE, csrf, httponly=False, secure=True, samesite="lax", path="/") return response + record_login_success(ip) response = RedirectResponse(url="/", status_code=303) issue_auth_cookie(response, user) issue_csrf_cookie(response)