Seguridad: rate limiting, códigos de un solo uso, rotación de FERNET_KEY
- utils/rate_limiter.py: ventana deslizante 20 req/60s por chat_id, aplicado en todos los handlers de telegram_bot.py - database.py: tabla invite_codes (un solo uso por código), funciones crear_invite_code() y verificar_y_consumir_invite_code() - onboarding.py: verifica primero en DB (un solo uso), fallback al INVITE_CODE del .env (ilimitado, para uso interno) - scripts/generar_codigo.py: genera 1 o N códigos de invitación - scripts/rotar_fernet_key.py: re-cifra todos los tenants y el .env con nueva FERNET_KEY, backup automático de .env antes de escribir Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
44
utils/rate_limiter.py
Normal file
44
utils/rate_limiter.py
Normal file
@@ -0,0 +1,44 @@
|
||||
"""
|
||||
utils/rate_limiter.py — Rate limiting por chat_id.
|
||||
|
||||
Ventana deslizante en memoria: máximo N requests por M segundos.
|
||||
Si el proceso se reinicia el contador se resetea (aceptable en v0).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from collections import defaultdict, deque
|
||||
|
||||
|
||||
class RateLimiter:
|
||||
"""
|
||||
Ventana deslizante: registra timestamps de requests por chat_id.
|
||||
Thread-safe para uso dentro de un solo event loop asyncio.
|
||||
"""
|
||||
|
||||
def __init__(self, max_requests: int = 20, window_seconds: int = 60) -> None:
|
||||
self._max = max_requests
|
||||
self._window = window_seconds
|
||||
self._history: dict[int, deque[float]] = defaultdict(deque)
|
||||
|
||||
def is_allowed(self, chat_id: int) -> bool:
|
||||
"""
|
||||
Retorna True y registra el request si está dentro del límite.
|
||||
Retorna False sin registrar si el límite fue excedido.
|
||||
"""
|
||||
now = time.monotonic()
|
||||
hist = self._history[chat_id]
|
||||
|
||||
# Descartar entradas fuera de la ventana
|
||||
while hist and hist[0] < now - self._window:
|
||||
hist.popleft()
|
||||
|
||||
if len(hist) >= self._max:
|
||||
return False
|
||||
|
||||
hist.append(now)
|
||||
return True
|
||||
|
||||
def reset(self, chat_id: int) -> None:
|
||||
"""Limpia el historial de un chat_id (ej: tras un resetear)."""
|
||||
self._history.pop(chat_id, None)
|
||||
Reference in New Issue
Block a user