ayurishchevandClaude Opus 5.5
c4079724c5
Безопасность, пункты 1–4 ревью (docs/changes/021): - приложение не стартует со слабыми секретами: API_TOKEN и SESSION_SECRET пустые, change-me или короче 32 символов, ADMIN_PASSWORD короче 12, невалидный SECRET_KEY; значения секретов в ошибку не попадают; - вход в UI: 5 неверных попыток за 10 минут блокируют IP на 10 минут (429, пароль не проверяется); неверный пароль — 401; событие auth.locked пишется один раз, журнал не засыпается перебором; - SESSION_COOKIE_SECURE — флаг Secure у cookie сессии (по умолчанию выкл.); - _safe_next: редирект только на локальный путь (/ui/move, удаление бэкапов); - .env.bak-* в .gitignore. Тесты: 28 из 28. Стенд: change-me → 401, новый токен → 200, отказ старта со слабым токеном, блокировка входа и редиректы проверены, боевые данные не изменены. Ручная проверка UI пользователем на момент коммита не подтверждена. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
85 lines
3.4 KiB
Python
85 lines
3.4 KiB
Python
import hmac
|
|
import threading
|
|
import time
|
|
|
|
from cryptography.fernet import Fernet
|
|
from fastapi import Depends, HTTPException
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
|
|
from app.config import get_settings
|
|
|
|
_bearer = HTTPBearer(auto_error=False)
|
|
|
|
|
|
def _fernet() -> Fernet:
|
|
key = get_settings().secret_key
|
|
if not key:
|
|
raise RuntimeError("SECRET_KEY не задан (ключ Fernet)")
|
|
return Fernet(key.encode())
|
|
|
|
|
|
def encrypt(value: str) -> str:
|
|
return _fernet().encrypt(value.encode()).decode()
|
|
|
|
|
|
def decrypt(token: str) -> str:
|
|
return _fernet().decrypt(token.encode()).decode()
|
|
|
|
|
|
def check_admin(user: str, password: str) -> bool:
|
|
s = get_settings()
|
|
ok_user = hmac.compare_digest(user.encode(), s.admin_user.encode())
|
|
ok_pass = hmac.compare_digest(password.encode(), s.admin_password.encode())
|
|
return ok_user and ok_pass
|
|
|
|
|
|
# --- защита от перебора: вход в UI (по IP) и подтверждение действий паролем (очистка журнала, по пользователю) ---
|
|
FAIL_LIMIT, WINDOW_S, LOCK_S = 5, 600, 600 # 5 неверных за 10 минут → блокировка на 10 минут
|
|
_guard = threading.Lock()
|
|
_fails: dict[str, list[float]] = {}
|
|
_locked_until: dict[str, float] = {}
|
|
|
|
|
|
def verify_password(user: str, password: str) -> bool:
|
|
"""Пароль пользователя сессии (единственный пользователь — ADMIN_USER); сравнение за постоянное время."""
|
|
s = get_settings()
|
|
return (hmac.compare_digest(user.encode(), s.admin_user.encode())
|
|
and hmac.compare_digest(password.encode(), s.admin_password.encode()))
|
|
|
|
|
|
def lockout_remaining(key: str) -> int:
|
|
"""Сколько секунд осталось до конца блокировки по ключу (0 — не заблокирован).
|
|
|
|
Ключ — своё пространство на каждый вид перебора (например, `f"login:{ip}"` для входа в UI
|
|
и имя пользователя для очистки журнала), чтобы счётчики не пересекались."""
|
|
with _guard:
|
|
return max(0, int(_locked_until.get(key, 0) - time.monotonic() + 0.999))
|
|
|
|
|
|
def register_failure(key: str) -> int:
|
|
"""Учитывает неверную попытку по ключу; возвращает число оставшихся попыток (0 — ключ заблокирован)."""
|
|
now = time.monotonic()
|
|
with _guard:
|
|
recent = [t for t in _fails.get(key, []) if now - t < WINDOW_S] + [now]
|
|
_fails[key] = recent
|
|
if len(recent) >= FAIL_LIMIT:
|
|
_locked_until[key] = now + LOCK_S
|
|
_fails[key] = []
|
|
return 0
|
|
return FAIL_LIMIT - len(recent)
|
|
|
|
|
|
def reset_failures(key: str) -> None:
|
|
with _guard:
|
|
_fails.pop(key, None)
|
|
_locked_until.pop(key, None)
|
|
|
|
|
|
async def require_api_token(cred: HTTPAuthorizationCredentials | None = Depends(_bearer)) -> None:
|
|
expected = get_settings().api_token
|
|
if cred is None or not hmac.compare_digest(cred.credentials.encode(), expected.encode()):
|
|
raise HTTPException(status_code=401, detail="Invalid or missing API token")
|
|
from app.services import events # локальный импорт: security загружается раньше сервисов
|
|
|
|
events.set_actor("api")
|