Coverage for src / local_deep_research / web / utils / rate_limiter.py: 92%

1""" 

2Rate limiting utility for authentication endpoints. 

3Provides a global limiter instance that can be imported by blueprints. 

4 

5Rate limits are configurable via UI settings (security.rate_limit_*) and 

6stored in server_config.json. Changes require server restart to take effect. 

7 

8Note: This is designed for single-instance local deployments. For multi-worker 

9production deployments, configure Redis storage via RATELIMIT_STORAGE_URL. 

10""" 

11 

12from flask import request 

13from flask_limiter import Limiter 

14from flask_limiter.util import get_remote_address 

15 

16from ...settings.env_registry import is_rate_limiting_enabled 

17from ..server_config import load_server_config 

18 

19# Load rate limits from server config (UI-configurable) 

20# Multiple limits can be separated by semicolons (e.g., "5000 per hour;50000 per day") 

21_config = load_server_config() 

22DEFAULT_RATE_LIMIT = _config["rate_limit_default"] 

23LOGIN_RATE_LIMIT = _config["rate_limit_login"] 

24REGISTRATION_RATE_LIMIT = _config["rate_limit_registration"] 

25 

26 

27def get_client_ip(): 

28 """ 

29 Get the real client IP address, respecting X-Forwarded-For headers. 

30 

31 This is important for deployments behind proxies/load balancers. 

32 Falls back to direct remote address if no forwarded headers present. 

33 """ 

34 # Check X-Forwarded-For header (set by proxies/load balancers) 

35 forwarded_for = request.environ.get("HTTP_X_FORWARDED_FOR") 

36 if forwarded_for: 

37 # Take the first IP in the chain (client IP) 

38 return forwarded_for.split(",")[0].strip() 

39 

40 # Check X-Real-IP header (alternative proxy header) 

41 real_ip = request.environ.get("HTTP_X_REAL_IP") 

42 if real_ip: 42 ↛ 43line 42 didn't jump to line 43 because the condition on line 42 was never true

43 return real_ip.strip() 

44 

45 # Fallback to direct remote address 

46 return get_remote_address() 

47 

48 

49# Global limiter instance - will be initialized in app_factory 

50# Rate limiting is disabled in CI unless ENABLE_RATE_LIMITING=true 

51# This allows the rate limiting test to run with rate limiting enabled 

52# 

53# Note: In-memory storage is used by default, which is suitable for single-instance 

54# deployments. For multi-instance production deployments behind a load balancer, 

55# configure Redis storage via RATELIMIT_STORAGE_URL environment variable: 

56# export RATELIMIT_STORAGE_URL="redis://localhost:6379" 

57limiter = Limiter( 

58 key_func=get_client_ip, 

59 default_limits=[DEFAULT_RATE_LIMIT], 

60 storage_uri="memory://", 

61 headers_enabled=True, 

62 enabled=is_rate_limiting_enabled, 

63) 

64 

65 

66# Shared rate limit decorators for authentication endpoints 

67# These can be imported and used directly on routes 

68login_limit = limiter.shared_limit( 

69 LOGIN_RATE_LIMIT, 

70 scope="login", 

71) 

72 

73registration_limit = limiter.shared_limit( 

74 REGISTRATION_RATE_LIMIT, 

75 scope="registration", 

76)