ayurishchevandClaude Opus 5.5
e1f197ccff
Пункт 12 ревью 2026-09-28 17:35 (docs/changes/026): tests/test_app.py (936 строк) разнесён на test_security, test_devices, test_operations, test_backups, test_events, test_ids_migrations, test_architecture; общие хелперы — tests/helpers.py. Тела тестов не менялись (AST совпадает). security.reset_all_failures() — фикстура сбрасывает счётчики попыток по всем ключам, тесты не зависят от порядка. Тесты: 36 из 36 полным прогоном, каждый модуль отдельно и в обратном порядке; тест-линтер после переноса проверен мутацией. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
126 lines
5.5 KiB
Python
126 lines
5.5 KiB
Python
import hmac
|
||
import ipaddress
|
||
import threading
|
||
import time
|
||
|
||
from cryptography.fernet import Fernet
|
||
from fastapi import Depends, HTTPException, Request
|
||
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
||
|
||
from app.config import get_settings, trusted_networks
|
||
|
||
_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
|
||
|
||
|
||
def _in_networks(ip: str, networks) -> bool:
|
||
try:
|
||
addr = ipaddress.ip_address(ip)
|
||
except ValueError:
|
||
return False
|
||
return any(addr in net for net in networks)
|
||
|
||
|
||
def client_ip(request: Request) -> str:
|
||
"""IP клиента с учётом TRUSTED_PROXIES. Пустой список или peer не из доверенной сети — X-Forwarded-For
|
||
полностью игнорируется (подделка заголовка бессмысленна), возвращается адрес соединения.
|
||
|
||
Иначе цепочка `X-Forwarded-For` (все заголовки, значения через запятую) разбирается справа налево:
|
||
доверенные прокси пропускаются, первый адрес не из доверенной сети — IP клиента. Невалидная запись
|
||
обрывает разбор — возвращается последний валидный разобранный адрес (или peer, если такого не было).
|
||
Если все адреса цепочки доверенные (или заголовка нет) — возвращается peer."""
|
||
peer = request.client.host if request.client else "unknown"
|
||
networks = trusted_networks(get_settings().trusted_proxies)
|
||
if not networks or not _in_networks(peer, networks):
|
||
return peer
|
||
chain = [part.strip() for header in request.headers.getlist("x-forwarded-for") for part in header.split(",")]
|
||
last_valid = None
|
||
for hop in reversed(chain):
|
||
try:
|
||
ipaddress.ip_address(hop)
|
||
except ValueError:
|
||
return last_valid if last_valid is not None else peer
|
||
if not _in_networks(hop, networks):
|
||
return hop
|
||
last_valid = hop
|
||
return peer
|
||
|
||
|
||
# --- защита от перебора: вход в 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)
|
||
|
||
|
||
def reset_all_failures() -> None:
|
||
"""Сбрасывает счётчики неудачных попыток по всем ключам (для тестов и будущего администрирования)."""
|
||
with _guard:
|
||
_fails.clear()
|
||
_locked_until.clear()
|
||
|
||
|
||
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")
|