"""Redis-based distributed rate limiting for AgentForms.
Uses sliding window counters in Redis for accurate rate limiting
across multiple worker instances.
Fallback: in-memory rate limiting when Redis is unavailable
(prevents fail-open security issue).
Usage:
from app.services.ratelimit import RateLimiter
limiter = RateLimiter()
# Check API rate limit (per user, 100 req/min)
if not limiter.allow("api:user:123"):
abort(429, "Rate limit exceeded")
# Check IP-based limit for public endpoints
if not limiter.allow("ip:" + request.remote_addr, max_requests=20, window=60):
abort(429, "Rate limit exceeded")
"""
import logging
import os
import threading
import time
logger = logging.getLogger("agentforms.ratelimit")
from app.services.logging import log as struct_log
class InMemoryRateLimiter:
"""Simple in-memory rate limiter using sliding window counters.
Used as fallback when Redis is unavailable. Per-process only
(not shared across workers), but prevents fail-open.
"""
def __init__(self):
self._counts = {} # key -> [(timestamp, count)]
self._lock = threading.Lock()
def check(self, key: str, max_requests: int = 100, window: int = 60) -> tuple:
"""Check rate limit using in-memory sliding window.
Returns:
(allowed, remaining, reset_at) — bool, int, float.
"""
now = time.time()
cutoff = now - window
with self._lock:
# Get or create window entry
if key not in self._counts:
self._counts[key] = []
# Remove expired entries
entries = self._counts[key]
self._counts[key] = [t for t in entries if t > cutoff]
entries = self._counts[key]
count = len(entries)
reset_at = (entries[0] + window) if entries else now + window
if count >= max_requests:
remaining = 0
allowed = False
else:
# Add this request
self._counts[key].append(now)
remaining = max(0, max_requests - count - 1)
allowed = True
# Periodic cleanup of old keys (every 1000 checks)
if len(self._counts) > 1000:
self._cleanup(cutoff)
return (allowed, remaining, reset_at)
def _cleanup(self, cutoff: float):
"""Remove expired entries (called while holding lock)."""
expired = [k for k, v in self._counts.items() if not v or max(v) < cutoff]
for k in expired:
del self._counts[k]
class RateLimiter:
"""Sliding window rate limiter backed by Redis.
Uses atomic Redis operations (INCR + EXPIRE) for correctness
across multiple worker instances.
"""
def __init__(self, redis_client=None):
self._redis = redis_client
self._connected = bool(redis_client)
self._local = InMemoryRateLimiter() # Fallback when Redis is down
def connect(self, redis_client):
"""Set the Redis client (called from app init)."""
self._redis = redis_client
self._connected = True
def allow(self, key: str, max_requests: int = 100, window: int = 60) -> bool:
"""Check if a request is allowed under the rate limit.
Args:
key: Unique rate limit key (e.g., "api:user:123")
max_requests: Maximum requests allowed in the window
window: Time window in seconds
Returns:
True if the request is allowed, False if rate limited.
Always returns True if Redis is unavailable (fail-open).
"""
allowed, _, _ = self.check(key, max_requests=max_requests, window=window)
return allowed
def check(self, key: str, max_requests: int = 100, window: int = 60) -> tuple:
"""Check rate limit and return detailed info.
Args:
key: Unique rate limit key (e.g., "api:user:123")
max_requests: Maximum requests allowed in the window
window: Time window in seconds
Returns:
(allowed, remaining, reset_at) — bool, int, float.
allowed: True if request is permitted.
remaining: Requests left in the current window.
reset_at: Unix timestamp when the window resets.
Falls back to in-memory rate limiter if Redis is unavailable
(prevents fail-open security issue).
"""
if not self._connected:
# Fallback to in-memory rate limiting — do NOT fail open
return self._local.check(key, max_requests=max_requests, window=window)
try:
pipe = self._redis.pipeline()
pipe.incr(key)
pipe.expire(key, window)
results = pipe.execute()
count = results[0]
remaining = max(0, max_requests - count)
reset_at = time.time() + window
allowed = count <= max_requests
if not allowed:
struct_log.warning(
"ratelimit",
"Rate limit exceeded",
key=key,
count=count,
max=max_requests,
window=window,
)
return (allowed, remaining, reset_at)
except Exception as e:
logger.error("Rate limiter Redis error: %s", e)
# Fallback to in-memory on Redis error — do NOT fail open
struct_log.warning("ratelimit", "Redis error, using in-memory fallback", error=str(e))
self._connected = False # Switch to fallback mode
return self._local.check(key, max_requests=max_requests, window=window)
def flush(self, pattern: str = "*") -> None:
"""Flush rate limit keys matching pattern. Use with care — intended for tests."""
if not self._connected:
return
try:
keys = self._redis.keys(pattern)
if keys:
self._redis.delete(*keys)
except Exception as e:
logger.error("Rate limiter flush error: %s", e)
def get_remaining(self, key: str, max_requests: int = 100, window: int = 60) -> tuple:
"""Get remaining requests and reset time for a key.
Returns:
(remaining, reset_at) — remaining requests and window reset timestamp.
Falls back to in-memory limiter if Redis unavailable.
"""
if not self._connected:
# In-memory fallback — estimate from last check result
allowed, remaining, reset_at = self._local.check(key, max_requests=max_requests, window=window)
return (remaining, reset_at)
try:
count = self._redis.get(key)
ttl = self._redis.ttl(key)
if count is None:
return (max_requests, time.time() + window)
remaining = max(0, max_requests - int(count))
reset_at = time.time() + (ttl if ttl > 0 else window)
return (remaining, reset_at)
except Exception as e:
logger.error("Rate limiter Redis error: %s", e)
return (max_requests, 0)
# Module-level singleton
limiter = RateLimiter()