darko
fcafe1ae15
API аккаунтов на чистой стандартной библиотеке Python (server/splitapi): - регистрация по почте: пароль задаётся по ссылке из письма, аккаунт создаётся только после подтверждения - вход, выход, сброс пароля по почте (токены одноразовые, в базе только SHA-256), удаление аккаунта - пароли scrypt, лимиты попыток по паре «адрес + IP», IP и аккаунту, защита от CSRF - хранение прогресса пользователя с оптимистичной блокировкой версии - SMTP через переменные окружения; без SMTP на публичном адресе API не стартует Клиент: public/auth.js (вход, регистрация, сброс, синхронизация), public/sync.js (чистое слияние прогресса). Без API сайт работает как раньше. Развёртывание: Dockerfile.api, сервис splitapi в docker-compose, проксирование /api/ в nginx с ограничением частоты, deploy/install-api-debian.sh (systemd), .env.example. Тесты: 216 для бэкенда (unittest, покрытие около 97%), 16 для клиентского слияния. Документация: раздел «Аккаунты и почта» в docs/DEPLOY.md, раздел 9 в docs/ИНСТРУКЦИЯ.md.
80 lines
3.4 KiB
Python
80 lines
3.4 KiB
Python
"""Ограничитель частоты запросов со скользящим окном (в памяти одного процесса)."""
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import threading
|
|
import time
|
|
from collections import OrderedDict, deque
|
|
from typing import Callable
|
|
|
|
|
|
class RateLimiter:
|
|
def __init__(self, clock: Callable[[], float] = time.monotonic, max_keys: int = 50_000, purge_interval: int = 60) -> None:
|
|
self._clock = clock
|
|
self._max_keys = max_keys
|
|
self._purge_interval = purge_interval
|
|
self._last_purge = float("-inf")
|
|
self._lock = threading.Lock()
|
|
self._hits: OrderedDict[str, tuple[deque[float], int]] = OrderedDict()
|
|
|
|
def check(self, key: str, limit: int, window: int) -> tuple[bool, int]:
|
|
"""Учитывает попытку. Возвращает (разрешено, через сколько секунд повторить)."""
|
|
now = self._clock()
|
|
with self._lock:
|
|
entry = self._hits.get(key)
|
|
hits = entry[0] if entry else deque()
|
|
while hits and hits[0] <= now - window:
|
|
hits.popleft()
|
|
if len(hits) >= limit:
|
|
self._hits[key] = (hits, window)
|
|
return False, max(1, math.ceil(hits[0] + window - now))
|
|
hits.append(now)
|
|
self._hits[key] = (hits, window)
|
|
self._hits.move_to_end(key)
|
|
self._evict(now)
|
|
return True, 0
|
|
|
|
def is_blocked(self, key: str, limit: int, window: int) -> tuple[bool, int]:
|
|
"""Проверяет лимит, не засчитывая попытку (вместе с record() считает только неудачи)."""
|
|
now = self._clock()
|
|
with self._lock:
|
|
entry = self._hits.get(key)
|
|
hits = entry[0] if entry else deque()
|
|
while hits and hits[0] <= now - window:
|
|
hits.popleft()
|
|
if len(hits) >= limit:
|
|
return True, max(1, math.ceil(hits[0] + window - now))
|
|
return False, 0
|
|
|
|
def record(self, key: str, window: int) -> None:
|
|
now = self._clock()
|
|
with self._lock:
|
|
entry = self._hits.get(key)
|
|
hits = entry[0] if entry else deque()
|
|
hits.append(now)
|
|
self._hits[key] = (hits, window)
|
|
self._hits.move_to_end(key)
|
|
self._evict(now)
|
|
|
|
def forgive(self, key: str) -> None:
|
|
"""Возвращает последнюю учтённую попытку (успешный вход не должен приближать блокировку)."""
|
|
with self._lock:
|
|
entry = self._hits.get(key)
|
|
if entry and entry[0]:
|
|
entry[0].pop()
|
|
|
|
def _evict(self, now: float) -> None:
|
|
if len(self._hits) <= self._max_keys:
|
|
return
|
|
# Полный проход по ключам дорог, поэтому не чаще раза в purge_interval; иначе просто вытесняем самые старые.
|
|
if now - self._last_purge >= self._purge_interval:
|
|
self._last_purge = now
|
|
for key in [k for k, (hits, window) in self._hits.items() if not hits or hits[-1] <= now - window]:
|
|
del self._hits[key]
|
|
while len(self._hits) > self._max_keys:
|
|
self._hits.popitem(last=False)
|
|
|
|
def size(self) -> int:
|
|
with self._lock:
|
|
return len(self._hits)
|