Coverage for app/venv/lib/python3.14/site-packages/weblate/utils/ratelimit.py: 48%

1# Copyright © Michal Čihař <michal@weblate.org> 

2# 

3# SPDX-License-Identifier: GPL-3.0-or-later 

4 

5from __future__ import annotations 

6 

7from contextlib import suppress 

8from functools import wraps 

9from typing import TYPE_CHECKING 

10 

11from django.conf import settings 

12from django.contrib.auth import logout 

13from django.core.cache import cache 

14from django.middleware.csrf import rotate_token 

15from django.shortcuts import redirect 

16from django.template.loader import render_to_string 

17from siphashc import siphash 

18 

19from weblate.logger import LOGGER 

20from weblate.utils import messages 

21from weblate.utils.docs import get_doc_url 

22from weblate.utils.hash import calculate_checksum 

23from weblate.utils.request import get_ip_address 

24 

25if TYPE_CHECKING: 25 ↛ 26line 25 didn't jump to line 26 because the condition on line 25 was never true

26 from django.http import HttpResponse 

27 

28 from weblate.auth.models import AuthenticatedHttpRequest, User 

29 

30 

31def get_rate_setting(scope: str, suffix: str): 

32 key = f"RATELIMIT_{scope.upper()}_{suffix}" 

33 if hasattr(settings, key): 

34 return getattr(settings, key) 

35 return getattr(settings, f"RATELIMIT_{suffix}") 

36 

37 

38def reset_rate_limit( 

39 scope, request: AuthenticatedHttpRequest | None = None, address=None, user=None 

40) -> None: 

41 """Reset rate limit.""" 

42 limiter = RateLimitHttpRequest(scope, request, address, user) 

43 limiter.reset() 

44 

45 

46def revert_rate_limit(scope, request: AuthenticatedHttpRequest) -> None: 

47 """ 

48 Revert rate limit to previous state. 

49 

50 This can be used when rate limiting POST, but ignoring some events. 

51 """ 

52 limiter = RateLimitHttpRequest(scope, request) 

53 limiter.revert() 

54 

55 

56def check_rate_limit(scope: str, request: AuthenticatedHttpRequest) -> bool: 

57 """Check authentication rate limit.""" 

58 if request.user.is_superuser: 

59 return True 

60 

61 limiter = RateLimitHttpRequest(scope, request) 

62 is_exceeded, _ = limiter.is_limit_exceeded() 

63 

64 if is_exceeded: 

65 # Set key to longer expiry for lockout period 

66 limiter.touch(get_rate_setting(scope, "LOCKOUT")) 

67 LOGGER.info( 

68 "rate-limit lockout for %s in %s scope from %s", 

69 limiter.key, 

70 scope, 

71 get_ip_address(request), 

72 ) 

73 return False 

74 

75 return True 

76 

77 

78def session_ratelimit_post(scope: str, logout_user: bool = True): 

79 """Session based rate limiting for POST requests.""" 

80 

81 def _session_ratelimit_post_controller(function): 

82 def _rate_wrap( 

83 request: AuthenticatedHttpRequest, *args, **kwargs 

84 ) -> HttpResponse: 

85 if request.method == "POST" and not check_rate_limit(scope, request): 

86 # Rotate session token 

87 rotate_token(request) 

88 if not logout_user: 

89 messages.error( 

90 request, 

91 render_to_string( 

92 "ratelimit.html", {"do_logout": False, "user": request.user} 

93 ), 

94 ) 

95 return redirect(request.get_full_path()) 

96 # Logout user 

97 do_logout = request.user.is_authenticated 

98 if do_logout: 

99 logout(request) 

100 messages.error( 

101 request, 

102 render_to_string( 

103 "ratelimit.html", {"do_logout": do_logout, "user": request.user} 

104 ), 

105 ) 

106 return redirect("login") 

107 return function(request, *args, **kwargs) 

108 

109 return wraps(function)(_rate_wrap) 

110 

111 return _session_ratelimit_post_controller 

112 

113 

114def rate_limit_notify(address: str) -> tuple[bool, str]: 

115 """ 

116 Multi-level rate limiting for email notifications. 

117 

118 Returns: tuple: (is_blocked, reason) 

119 """ 

120 rate_limits = settings.RATELIMIT_NOTIFICATION_LIMITS 

121 encoded_email = siphash("Weblate notifier", address) 

122 limiter = RateLimitNotify(f"{encoded_email}", rate_limits) 

123 return limiter.is_limit_exceeded() 

124 

125 

126class RateLimitBase: 

127 cache_items: list[CacheCounterItem] 

128 key: str 

129 

130 def __init__(self, base_key: str, rate_limits: list[tuple[int, int]]): 

131 self.key = base_key 

132 

133 self.cache_items = [ 

134 CacheCounterItem(self.key, attempts, window) 

135 for attempts, window in rate_limits 

136 ] 

137 

138 def is_limit_exceeded(self) -> tuple[bool, str]: 

139 # Check all without decrementing 

140 for cache_item in self.cache_items: 

141 if cache_item.count_remaining <= 0: 

142 return ( 

143 True, 

144 f"rate limit exceeded ({cache_item.attempts}/{cache_item.window}s)", 

145 ) 

146 # If we get here, we can allow the operation - so decrement all counters 

147 for cache_item in self.cache_items: 

148 cache_item.decrement() 

149 return False, "" 

150 

151 def touch(self, timeout: int): 

152 for cache_item in self.cache_items: 

153 cache_item.touch(timeout) 

154 

155 def revert(self): 

156 for cache_item in self.cache_items: 

157 cache_item.increment() 

158 

159 def reset(self): 

160 for cache_item in self.cache_items: 

161 cache_item.delete() 

162 

163 

164class RateLimitNotify(RateLimitBase): 

165 def __init__(self, base_key: str, rate_limits: list[tuple[int, int]]): 

166 RateLimitBase.__init__(self, f"notify:rate:{base_key}", rate_limits) 

167 

168 

169class RateLimitHttpRequest(RateLimitBase): 

170 def __init__( 

171 self, 

172 scope: str, 

173 request: AuthenticatedHttpRequest | None = None, 

174 address: str | None = None, 

175 user: User | None = None, 

176 ): 

177 if request is not None and request.user.is_authenticated and user is None: 

178 user = request.user 

179 key: int | str 

180 if user is not None: 

181 key = user.id 

182 origin = "user" 

183 else: 

184 if address is None: 

185 address = get_ip_address(request) 

186 if not address: 

187 LOGGER.error( 

188 "could not obtain remote IP address, see %s", 

189 get_doc_url("admin/install", "reverse-proxy"), 

190 ) 

191 origin = "ip" 

192 key = calculate_checksum(address) 

193 

194 base_key = f"ratelimit-{origin}-{scope}-{key}" 

195 window = get_rate_setting(scope, "WINDOW") 

196 attempts = get_rate_setting(scope, "ATTEMPTS") 

197 

198 RateLimitBase.__init__(self, base_key, [(attempts, window)]) 

199 

200 

201class CacheCounterItem: 

202 cache_key: str 

203 attempts: int 

204 window: int 

205 

206 def __init__(self, base_key: str, attempts: int, window: int): 

207 self.cache_key = f"{base_key}:{attempts}:{window}" 

208 self.attempts = attempts 

209 self.window = window 

210 cache.add(self.cache_key, attempts, window) 

211 

212 @property 

213 def count_remaining(self) -> int: 

214 return cache.get(self.cache_key, 0) 

215 

216 def increment(self) -> None: 

217 with suppress(ValueError): 

218 cache.incr(self.cache_key) 

219 

220 def decrement(self) -> None: 

221 with suppress(ValueError): 

222 cache.decr(self.cache_key) 

223 

224 def touch(self, timeout: int) -> None: 

225 cache.touch(self.cache_key, timeout) 

226 

227 def delete(self) -> None: 

228 cache.delete(self.cache_key)