""" Rate limiting configuration using slowapi with Redis backend. This module provides rate limiting functionality to protect against brute force attacks and DoS attempts on sensitive endpoints. """ import logging import os from slowapi import Limiter from slowapi.util import get_remote_address from app.core.config import settings logger = logging.getLogger(__name__) def _get_storage_uri() -> str: """ Determine the appropriate storage URI for rate limiting. Priority: 1. Use memory storage if TESTING environment variable is set 2. Try Redis if available 3. Fall back to memory storage if Redis is unavailable (with warning) Note: Memory storage is acceptable for development but Redis should be used in production for consistent rate limiting across workers. """ # Use memory storage for testing testing = os.environ.get("TESTING", "").lower() in ("true", "1", "yes") if testing: return "memory://" # Try to connect to Redis redis_url = settings.REDIS_URL try: import redis r = redis.Redis( host=settings.REDIS_HOST, port=settings.REDIS_PORT, db=settings.REDIS_DB, socket_connect_timeout=1, ) r.ping() logger.info("Rate limiter using Redis storage") return redis_url except Exception as e: logger.warning( f"Redis unavailable for rate limiting ({e}). " "Falling back to in-memory storage. " "Note: In production, Redis should be running for consistent " "rate limiting across multiple workers." ) return "memory://" _storage_uri = _get_storage_uri() # Create limiter instance with appropriate storage # Uses the client's remote address (IP) as the key for rate limiting limiter = Limiter( key_func=get_remote_address, storage_uri=_storage_uri, strategy="fixed-window", # Fixed window strategy for predictable rate limiting )