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.
118 lines
4.0 KiB
Python
118 lines
4.0 KiB
Python
"""Тесты ограничителя частоты запросов."""
|
|
import unittest
|
|
|
|
from splitapi.ratelimit import RateLimiter
|
|
|
|
|
|
class FakeClock:
|
|
def __init__(self):
|
|
self.t = 1000.0
|
|
|
|
def __call__(self):
|
|
return self.t
|
|
|
|
|
|
class RateLimiterTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.clock = FakeClock()
|
|
self.limiter = RateLimiter(clock=self.clock)
|
|
|
|
def test_allows_up_to_limit_then_blocks(self):
|
|
for _ in range(3):
|
|
self.assertTrue(self.limiter.check("k", 3, 60)[0])
|
|
allowed, retry = self.limiter.check("k", 3, 60)
|
|
self.assertFalse(allowed)
|
|
self.assertGreaterEqual(retry, 1)
|
|
|
|
def test_window_slides(self):
|
|
for _ in range(3):
|
|
self.limiter.check("k", 3, 60)
|
|
self.clock.t += 61
|
|
self.assertTrue(self.limiter.check("k", 3, 60)[0])
|
|
|
|
def test_retry_after_matches_oldest_hit(self):
|
|
self.limiter.check("k", 2, 60)
|
|
self.clock.t += 20
|
|
self.limiter.check("k", 2, 60)
|
|
allowed, retry = self.limiter.check("k", 2, 60)
|
|
self.assertFalse(allowed)
|
|
self.assertEqual(retry, 40)
|
|
|
|
def test_keys_are_independent(self):
|
|
self.assertTrue(self.limiter.check("a", 1, 60)[0])
|
|
self.assertFalse(self.limiter.check("a", 1, 60)[0])
|
|
self.assertTrue(self.limiter.check("b", 1, 60)[0])
|
|
|
|
def test_blocked_attempts_do_not_extend_the_block(self):
|
|
self.limiter.check("k", 1, 60)
|
|
for _ in range(50):
|
|
self.limiter.check("k", 1, 60)
|
|
self.clock.t += 61
|
|
self.assertTrue(self.limiter.check("k", 1, 60)[0])
|
|
|
|
def test_is_blocked_does_not_count_attempts(self):
|
|
for _ in range(10):
|
|
self.assertFalse(self.limiter.is_blocked("k", 2, 60)[0])
|
|
|
|
def test_recorded_failures_lead_to_block_and_expire(self):
|
|
self.limiter.record("k", 60)
|
|
self.limiter.record("k", 60)
|
|
blocked, retry = self.limiter.is_blocked("k", 2, 60)
|
|
self.assertTrue(blocked)
|
|
self.assertGreaterEqual(retry, 1)
|
|
self.clock.t += 61
|
|
self.assertFalse(self.limiter.is_blocked("k", 2, 60)[0])
|
|
|
|
def test_forgive_gives_back_the_latest_attempt(self):
|
|
self.limiter.check("k", 2, 60)
|
|
self.limiter.check("k", 2, 60)
|
|
self.assertFalse(self.limiter.check("k", 2, 60)[0])
|
|
self.limiter.forgive("k")
|
|
self.assertTrue(self.limiter.check("k", 2, 60)[0])
|
|
|
|
def test_forgive_unknown_key_is_harmless(self):
|
|
self.limiter.forgive("never-seen")
|
|
self.assertTrue(self.limiter.check("never-seen", 1, 60)[0])
|
|
|
|
def test_successful_attempts_can_be_forgiven_indefinitely(self):
|
|
for _ in range(50):
|
|
self.assertTrue(self.limiter.check("ok", 3, 60)[0])
|
|
self.limiter.forgive("ok")
|
|
|
|
def test_expired_entries_are_dropped_before_live_ones(self):
|
|
limiter = RateLimiter(clock=self.clock, max_keys=3)
|
|
limiter.check("old1", 1, 60)
|
|
limiter.check("old2", 1, 60)
|
|
self.clock.t += 100
|
|
limiter.check("new1", 1, 60)
|
|
limiter.check("new2", 1, 60) # тут превышение: просроченные уходят первыми
|
|
self.assertEqual(limiter.size(), 2)
|
|
self.assertFalse(limiter.check("new1", 1, 60)[0]) # живой счётчик не потерян
|
|
|
|
def test_concurrent_checks_never_exceed_the_limit(self):
|
|
import threading
|
|
allowed = []
|
|
gate = threading.Barrier(16)
|
|
|
|
def worker():
|
|
gate.wait()
|
|
allowed.append(self.limiter.check("race", 5, 60)[0])
|
|
|
|
threads = [threading.Thread(target=worker) for _ in range(16)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
self.assertEqual(sum(allowed), 5)
|
|
|
|
def test_memory_is_bounded(self):
|
|
limiter = RateLimiter(clock=self.clock, max_keys=100)
|
|
for i in range(500):
|
|
limiter.check(f"k{i}", 5, 60)
|
|
self.clock.t += 0.01
|
|
self.assertLessEqual(limiter.size(), 100)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|