Админка: защита от перебора пароля и правка дефолта SETTINGS_FILE
- не больше 10 неудачных входов с одного адреса за 5 минут, дальше 429 (панель доступна из интернета, простой перебор закрываем); успешный вход сбрасывает счётчик; - .env.example: SETTINGS_FILE пустой — локально настройки живут в памяти, в docker compose подставляет /data/settings.json по умолчанию.
This commit is contained in:
@@ -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),
|
||||
|
||||
Reference in New Issue
Block a user