diff --git a/.env.example b/.env.example index 631b306..1934d8d 100644 --- a/.env.example +++ b/.env.example @@ -14,8 +14,10 @@ ADMIN_PATH= # Логин и пароль администратора ADMIN_LOGIN= ADMIN_PASSWORD= -# Где хранить настройки (ссылка/интервал); в docker это том schedule_data -SETTINGS_FILE=/data/settings.json +# Где хранить настройки (ссылка/интервал). +# В docker по умолчанию /data/settings.json (том schedule_data); +# локально можно оставить пустым — тогда настройки живут в памяти. +SETTINGS_FILE= # --- порты и адреса ---------------------------------------------------------- API_PORT=8000 diff --git a/src/backend/admin.py b/src/backend/admin.py index 132acf2..7bd27e5 100644 --- a/src/backend/admin.py +++ b/src/backend/admin.py @@ -19,6 +19,7 @@ import hashlib import hmac import json import logging +import threading import time from dataclasses import dataclass @@ -32,6 +33,43 @@ log = logging.getLogger(__name__) COOKIE_NAME = "schedule_admin" SESSION_TTL_SECONDS = 12 * 3600 +# защита от перебора пароля: не больше N неудач с одного адреса за окно +LOGIN_MAX_FAILURES = 10 +LOGIN_WINDOW_SECONDS = 300.0 + + +class LoginThrottle: + """Простейший счётчик неудачных попыток входа по IP.""" + + def __init__( + self, + max_failures: int = LOGIN_MAX_FAILURES, + window: float = LOGIN_WINDOW_SECONDS, + ) -> None: + self._max = max_failures + self._window = window + self._lock = threading.Lock() + self._failures: dict[str, list[float]] = {} + + def _recent(self, key: str, now: float) -> list[float]: + return [ts for ts in self._failures.get(key, []) if now - ts < self._window] + + def blocked(self, key: str) -> bool: + now = time.time() + with self._lock: + recent = self._recent(key, now) + self._failures[key] = recent + return len(recent) >= self._max + + def register_failure(self, key: str) -> None: + now = time.time() + with self._lock: + self._failures[key] = [*self._recent(key, now), now] + + def reset(self, key: str) -> None: + with self._lock: + self._failures.pop(key, None) + @dataclass class AdminConfig: @@ -94,6 +132,7 @@ class IntervalBody(BaseModel): def create_admin_router(cache: ScheduleCache, config: AdminConfig) -> APIRouter: router = APIRouter(prefix="/api/admin", tags=["admin"]) + throttle = LoginThrottle() def require_auth(request: Request) -> None: token = request.cookies.get(COOKIE_NAME, "") @@ -104,13 +143,24 @@ def create_admin_router(cache: ScheduleCache, config: AdminConfig) -> APIRouter: proto = request.headers.get("x-forwarded-proto", "") or request.url.scheme return proto.split(",")[0].strip() == "https" + def _client_key(request: Request) -> str: + return request.client.host if request.client else "unknown" + @router.post("/login") def login(body: LoginBody, request: Request, response: Response) -> dict: + key = _client_key(request) + if throttle.blocked(key): + log.warning("Вход в админку: слишком много неудачных попыток с %s", key) + raise HTTPException( + status_code=429, detail="Слишком много попыток входа, попробуйте позже" + ) ok_login = _same(body.login.strip(), config.login) ok_password = _same(body.password, config.password) if not (ok_login and ok_password): + throttle.register_failure(key) log.warning("Неудачная попытка входа в админку (логин %r)", body.login[:40]) raise HTTPException(status_code=401, detail="Неверный логин или пароль") + throttle.reset(key) response.set_cookie( COOKIE_NAME, make_token(config.login, config.password), diff --git a/tests/test_admin.py b/tests/test_admin.py index 3554d99..e0d65f3 100644 --- a/tests/test_admin.py +++ b/tests/test_admin.py @@ -7,7 +7,7 @@ import pytest from fastapi.testclient import TestClient import main -from admin import AdminConfig, make_token, verify_token +from admin import LOGIN_MAX_FAILURES, AdminConfig, make_token, verify_token from fixtures import fixture_xlsx, make_cache from schedule import ScheduleError from schedule.settings import SettingsStore @@ -84,6 +84,14 @@ def test_logout(client): assert client.get("/api/admin/state").status_code == 401 +def test_login_throttled_after_failures(client): + for _ in range(LOGIN_MAX_FAILURES): + assert do_login(client, password="неверный").status_code == 401 + # дальше — отказ, даже с верным паролем, пока не истекло окно + assert do_login(client, password="неверный").status_code == 429 + assert do_login(client, password="s3cret").status_code == 429 + + def test_token_helpers(): token = make_token("admin", "s3cret", ttl=60) assert verify_token(token, "s3cret")