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

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# Settings modification rate limit - prevent abuse of settings endpoints 

26SETTINGS_RATE_LIMIT = _config.get("rate_limit_settings", "30 per minute") 

27 

28 

29def get_client_ip(): 

30 """ 

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

32 

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

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

35 """ 

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

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

38 if forwarded_for: 

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

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

41 

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

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

44 if real_ip: 

45 return real_ip.strip() 

46 

47 # Fallback to direct remote address 

48 return get_remote_address() 

49 

50 

51# Global limiter instance - will be initialized in app_factory 

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

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

54# 

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

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

57# configure Redis storage via RATELIMIT_STORAGE_URL environment variable: 

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

59limiter = Limiter( 

60 key_func=get_client_ip, 

61 default_limits=[DEFAULT_RATE_LIMIT], 

62 storage_uri="memory://", 

63 headers_enabled=True, 

64 enabled=is_rate_limiting_enabled, 

65) 

66 

67 

68# Shared rate limit decorators for authentication endpoints 

69# These can be imported and used directly on routes 

70login_limit = limiter.shared_limit( 

71 LOGIN_RATE_LIMIT, 

72 scope="login", 

73) 

74 

75registration_limit = limiter.shared_limit( 

76 REGISTRATION_RATE_LIMIT, 

77 scope="registration", 

78) 

79 

80settings_limit = limiter.shared_limit( 

81 SETTINGS_RATE_LIMIT, 

82 scope="settings", 

83)