Coverage for app/venv/lib/python3.14/site-packages/weblate/middleware.py: 37%
258 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 07:15 +0000
1# Copyright © Michal Čihař <michal@weblate.org>
2#
3# SPDX-License-Identifier: GPL-3.0-or-later
5from __future__ import annotations
7from copy import deepcopy
8from typing import TYPE_CHECKING, Literal, cast
9from urllib.parse import urlparse
11from django.conf import settings
12from django.contrib import messages
13from django.core.exceptions import ValidationError
14from django.core.validators import validate_ipv46_address
15from django.http import Http404, HttpResponsePermanentRedirect
16from django.shortcuts import redirect
17from django.urls import is_valid_path, reverse
18from django.urls.exceptions import NoReverseMatch
19from django.utils.http import escape_leading_slashes
20from django.utils.translation import gettext_lazy
21from social_core.backends.oauth import OAuthAuth
22from social_core.backends.open_id import OpenIdAuth
23from social_django.utils import load_strategy
25from weblate.auth.utils import get_auth_backends
26from weblate.lang.models import Language
27from weblate.logger import LOGGER
28from weblate.trans.actions import ActionEvents
29from weblate.trans.models import Change, Component, Project
30from weblate.utils.errors import report_error
31from weblate.utils.site import get_site_url
32from weblate.utils.views import parse_path
34if TYPE_CHECKING: 34 ↛ 35line 34 didn't jump to line 35 because the condition on line 34 was never true
35 from django.http import HttpResponse
36 from django.http.request import HttpRequest
38 from weblate.accounts.strategy import WeblateStrategy
39 from weblate.auth.models import AuthenticatedHttpRequest
41 CSP_KIND = Literal[
42 "default-src",
43 "style-src",
44 "img-src",
45 "script-src",
46 "connect-src",
47 "object-src",
48 "font-src",
49 "frame-src",
50 "frame-ancestors",
51 "base-uri",
52 "form-action",
53 "manifest-src",
54 "worker-src",
55 ]
56 CSP_TYPE = dict[CSP_KIND, set[str]]
58CSP_DIRECTIVES: CSP_TYPE = {
59 "default-src": {"'none'"},
60 "style-src": {"'self'", "'unsafe-inline'"},
61 # "data:" is required for bootstrap 5 icons
62 "img-src": {"'self'", "data:"},
63 "script-src": {"'self'"},
64 "connect-src": {"'self'"},
65 "object-src": {"'none'"},
66 "font-src": {"'self'"},
67 "frame-src": {"'none'"},
68 "frame-ancestors": {"'none'"},
69 "base-uri": {"'none'"},
70 "form-action": {"'self'"},
71 "manifest-src": {"'self'"},
72 # Used by altcha
73 "worker-src": {"'self'", "blob:"},
74}
76# URLs requiring inline javascript
77INLINE_PATHS = {
78 "social:begin",
79 "djangosaml2idp:saml_login_process",
80}
83class ProxyMiddleware:
84 """
85 Middleware that updates REMOTE_ADDR from proxy.
87 Note that this can have security implications and settings have to match your actual
88 proxy setup.
89 """
91 def __init__(self, get_response=None) -> None:
92 self.get_response = get_response
94 def __call__(self, request: HttpRequest):
95 # Fake HttpRequest attribute to inject configured
96 # site name into build_absolute_uri
97 request.__dict__["_current_scheme_host"] = get_site_url()
99 # Actual proxy handling
100 proxy = None
101 if settings.IP_BEHIND_REVERSE_PROXY: 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true
102 proxy = request.META.get(settings.IP_PROXY_HEADER)
103 if proxy: 103 ↛ 105line 103 didn't jump to line 105 because the condition on line 103 was never true
104 # X_FORWARDED_FOR returns client1, proxy1, proxy2,...
105 address = proxy.split(",")[settings.IP_PROXY_OFFSET].strip()
106 try:
107 validate_ipv46_address(address)
108 request.META["REMOTE_ADDR"] = address
109 except ValidationError:
110 report_error("Invalid IP address")
112 return self.get_response(request)
115class RedirectMiddleware:
116 """
117 Middleware that handles URL redirecting.
119 This used for fuzzy lookups of projects, for example case insensitive
120 or after renaming.
121 """
123 def __init__(self, get_response=None) -> None:
124 self.get_response = get_response
126 def __call__(self, request: AuthenticatedHttpRequest) -> HttpResponse:
127 response = self.get_response(request)
128 # This is based on APPEND_SLASH handling in Django
129 if response.status_code == 404 and self.should_redirect_with_slash(request): 129 ↛ 130line 129 didn't jump to line 130 because the condition on line 129 was never true
130 new_path = request.get_full_path(force_append_slash=True)
131 # Prevent construction of scheme relative urls.
132 new_path = escape_leading_slashes(new_path)
133 return HttpResponsePermanentRedirect(new_path)
134 return response
136 def should_redirect_with_slash(self, request: AuthenticatedHttpRequest) -> bool:
137 path = request.path_info
138 # Avoid redirecting non GET requests, these would fail anyway due to
139 # missing parameters.
140 # Redirecting on API removes authentication headers in many cases,
141 # so avoid that as well.
142 # Redirecting requests for Sourcemap files will not do anything good
143 if (
144 path.endswith(("/", ".map"))
145 or request.method != "GET"
146 or (
147 path.startswith(f"{settings.URL_PREFIX}/api")
148 and not path.startswith(f"{settings.URL_PREFIX}/api/doc")
149 and not path.startswith(f"{settings.URL_PREFIX}/api/schema")
150 )
151 ):
152 return False
153 urlconf = getattr(request, "urlconf", None)
154 slash_path = f"{path}/"
155 return not is_valid_path(path, urlconf) and bool(
156 is_valid_path(slash_path, urlconf)
157 )
159 def fixup_language(self, lang: str) -> Language | None:
160 return Language.objects.fuzzy_get_strict(code=lang)
162 def fixup_project(self, slug, request: AuthenticatedHttpRequest) -> Project | None:
163 project: Project | None
164 try:
165 project = Project.objects.get(slug__iexact=slug)
166 except Project.MultipleObjectsReturned:
167 return None
168 except Project.DoesNotExist:
169 project = Change.objects.lookup_project_rename(slug)
170 if project is None:
171 return None
173 request.user.check_access(project)
174 return project
176 def fixup_component(
177 self, slug: str, request: AuthenticatedHttpRequest, project: Project
178 ) -> Component | None:
179 try:
180 # Try uncategorized component first
181 component = project.component_set.get(category=None, slug__iexact=slug)
182 except Component.DoesNotExist:
183 try:
184 # Fallback to any such named component in project
185 component = project.component_set.filter(slug__iexact=slug)[0]
186 except IndexError:
187 try:
188 # Look for renamed components in a project
189 component = cast(
190 "Component",
191 project.change_set.filter(
192 action=ActionEvents.RENAME_COMPONENT, old=slug
193 )
194 .order()[0]
195 .component,
196 )
197 except IndexError:
198 return None
200 request.user.check_access_component(component)
201 return component
203 def check_existing_translations(self, name: str, project: Project) -> bool:
204 """
205 Check in existing translations for specific language.
207 Return False if language translation not present, else True.
208 """
209 return any(lang.name == name for lang in project.languages)
211 def process_exception( # noqa: C901
212 self, request: AuthenticatedHttpRequest, exception
213 ) -> HttpResponse | None:
214 from weblate.utils.views import UnsupportedPathObjectError
216 if not isinstance(exception, Http404):
217 return None
219 try:
220 resolver_match = request.resolver_match
221 except AttributeError:
222 return None
224 if resolver_match is None:
225 return None
227 kwargs = dict(resolver_match.kwargs)
228 path = list(kwargs.get("path", ()))
229 language_name = None
230 if not path:
231 return None
233 if isinstance(exception, UnsupportedPathObjectError):
234 # Redirect to parent for unsupported locations
235 path = path[:-1]
236 else:
237 # Try using last part as a language
238 language_len = 0
239 if len(path) >= 3:
240 language = self.fixup_language(path[-1])
241 if language is not None:
242 path[-1] = language.code
243 language_name = language.name
244 language_len = 1
246 try:
247 # Check if project exists
248 project = parse_path(request, path[:1], (Project,))
249 except UnsupportedPathObjectError:
250 return None
251 except Http404:
252 project = self.fixup_project(path[0], request)
253 if project is None:
254 return None
255 path[0] = project.slug
257 if len(path) >= 2:
258 if path[1] != "-":
259 path_offset = len(path) - (language_len)
260 try:
261 # Check if component exists
262 component = parse_path(
263 request, path[:path_offset], (Component,)
264 )
265 except UnsupportedPathObjectError:
266 return None
267 except Http404:
268 component = self.fixup_component(
269 path[-1 - language_len], request, project
270 )
271 if component is None:
272 return None
273 path[:path_offset] = component.get_url_path()
275 if language_name:
276 existing_trans = self.check_existing_translations(
277 language_name, project
278 )
279 if not existing_trans:
280 messages.add_message(
281 request,
282 messages.INFO,
283 gettext_lazy(
284 "%s translation is currently not available, "
285 "but can be added."
286 )
287 % language_name,
288 )
289 return redirect(reverse("show", kwargs={"path": path[:-1]}))
291 if path != kwargs["path"]:
292 kwargs["path"] = path
293 query = request.META["QUERY_STRING"]
294 if query:
295 query = f"?{query}"
296 try:
297 new_url = reverse(resolver_match.url_name, kwargs=kwargs)
298 except NoReverseMatch:
299 return None
300 return HttpResponsePermanentRedirect(f"{new_url}{query}")
302 return None
305class CSPBuilder:
306 directives: CSP_TYPE
307 request: AuthenticatedHttpRequest
308 response: HttpResponse
310 def __init__(
311 self, request: AuthenticatedHttpRequest, response: HttpResponse
312 ) -> None:
313 self.directives = deepcopy(CSP_DIRECTIVES)
314 self.request = request
315 self.response = response
316 self.apply_csp_settings()
317 self.build_csp_inline()
318 self.build_csp_sentry()
319 self.build_csp_piwik()
320 self.build_csp_google_analytics()
321 self.build_csp_media_url()
322 self.build_csp_static_url()
323 self.build_csp_cdn()
324 self.build_csp_auth()
325 self.build_csp_redoc()
327 def apply_csp_settings(self) -> None:
328 setting_names: dict[str, CSP_KIND] = {
329 "CSP_STYLE_SRC": "style-src",
330 "CSP_SCRIPT_SRC": "script-src",
331 "CSP_IMG_SRC": "img-src",
332 "CSP_CONNECT_SRC": "connect-src",
333 "CSP_FONT_SRC": "font-src",
334 "CSP_FORM_SRC": "form-action",
335 }
336 for name, rule in setting_names.items():
337 value = getattr(settings, name)
338 if value: 338 ↛ 339line 338 didn't jump to line 339 because the condition on line 338 was never true
339 self.directives[rule].update(value)
341 def add_csp_host(self, url: str, *directives: CSP_KIND) -> str | None:
342 domain = urlparse(url).hostname
343 # Handle domain only URLs (OpenInfraOpenId uses that)
344 if not domain and ":" not in url and "/" not in url:
345 domain = url
346 if domain:
347 for directive in directives:
348 self.directives[directive].add(domain)
349 else:
350 LOGGER.error(
351 "could not parse domain from '%s', not adding to Content-Security-Policy",
352 url,
353 )
355 return domain
357 def build_csp_redoc(self) -> None:
358 if self.request.resolver_match and self.request.resolver_match.view_name in { 358 ↛ 362line 358 didn't jump to line 362 because the condition on line 358 was never true
359 "redoc",
360 "swagger",
361 }:
362 self.directives["script-src"].add("'unsafe-inline'")
363 self.directives["img-src"].add("data:")
365 def build_csp_inline(self) -> None:
366 if ( 366 ↛ 370line 366 didn't jump to line 370 because the condition on line 366 was never true
367 self.request.resolver_match
368 and self.request.resolver_match.view_name in INLINE_PATHS
369 ):
370 self.directives["script-src"].add("'unsafe-inline'")
372 def build_csp_sentry(self) -> None:
373 # Sentry user feedback
374 if settings.SENTRY_DSN and self.response.status_code == 500: 374 ↛ 375line 374 didn't jump to line 375 because the condition on line 374 was never true
375 domain = self.add_csp_host(settings.SENTRY_DSN, "script-src", "connect-src")
376 # Add appropriate frontend servers for sentry.io
377 if domain.endswith(".de.sentry.io"):
378 self.directives["connect-src"].add("de.sentry.io")
379 self.directives["script-src"].add("de.sentry.io")
380 elif domain.endswith(".sentry.io"):
381 self.directives["script-src"].add("sentry.io")
382 self.directives["connect-src"].add("sentry.io")
383 self.directives["script-src"].add("'unsafe-inline'")
384 self.directives["img-src"].add("data:")
386 def build_csp_piwik(self) -> None:
387 # Matomo (Piwik) analytics
388 if settings.MATOMO_URL: 388 ↛ 389line 388 didn't jump to line 389 because the condition on line 388 was never true
389 self.add_csp_host(
390 settings.MATOMO_URL, "script-src", "img-src", "connect-src"
391 )
393 def build_csp_google_analytics(self) -> None:
394 # Google Analytics
395 if settings.GOOGLE_ANALYTICS_ID: 395 ↛ 396line 395 didn't jump to line 396 because the condition on line 395 was never true
396 self.directives["script-src"].add("'unsafe-inline'")
397 self.directives["script-src"].add("www.google-analytics.com")
398 self.directives["img-src"].add("www.google-analytics.com")
400 def build_csp_media_url(self) -> None:
401 # External media URL
402 if "://" in settings.MEDIA_URL: 402 ↛ 403line 402 didn't jump to line 403 because the condition on line 402 was never true
403 self.add_csp_host(settings.MEDIA_URL, "img-src")
405 def build_csp_static_url(self) -> None:
406 # External static URL
407 if "://" in settings.STATIC_URL: 407 ↛ 408line 407 didn't jump to line 408 because the condition on line 407 was never true
408 self.add_csp_host(
409 settings.STATIC_URL, "script-src", "img-src", "style-src", "font-src"
410 )
412 def build_csp_cdn(self) -> None:
413 # CDN for fonts
414 if settings.FONTS_CDN_URL: 414 ↛ 415line 414 didn't jump to line 415 because the condition on line 414 was never true
415 self.add_csp_host(settings.FONTS_CDN_URL, "style-src", "font-src")
417 def build_csp_auth(self) -> None:
418 # When using external image for Auth0 provider, add it here
419 if "://" in settings.SOCIAL_AUTH_AUTH0_IMAGE: 419 ↛ 420line 419 didn't jump to line 420 because the condition on line 419 was never true
420 self.add_csp_host(settings.SOCIAL_AUTH_AUTH0_IMAGE, "img-src")
422 # Third-party login flow extensions
423 if self.request.resolver_match and ( 423 ↛ 428line 423 didn't jump to line 428 because the condition on line 423 was never true
424 self.request.resolver_match.view_name.startswith("social:")
425 or self.request.resolver_match.view_name in {"login", "profile", "register"}
426 ):
427 social_strategy: WeblateStrategy
428 if hasattr(self.request, "social_strategy"):
429 social_strategy = self.request.social_strategy
430 else:
431 social_strategy = load_strategy(self.request)
432 for backend in get_auth_backends().values():
433 urls: list[str] = []
435 # Handle OpenId redirect flow
436 if issubclass(backend, OpenIdAuth):
437 urls = [backend(social_strategy).openid_url()]
439 # Handle OAuth redirect flow
440 elif issubclass(backend, OAuthAuth):
441 urls = [backend(social_strategy).authorization_url()]
443 # Handle SAML redirect flow
444 elif hasattr(backend, "get_idp"):
445 # Lazily import here to avoid pulling in xmlsec
446 from social_core.backends.saml import SAMLAuth
448 assert issubclass(backend, SAMLAuth) # noqa: S101
450 saml_auth = backend(social_strategy)
451 urls = [
452 saml_auth.get_idp(idp_name).sso_url
453 for idp_name in getattr(
454 settings, "SOCIAL_AUTH_SAML_ENABLED_IDPS", {}
455 )
456 ]
458 for url in urls:
459 domain = self.add_csp_host(url, "form-action")
460 if domain and domain.endswith(".amazonaws.com"):
461 self.directives["form-action"].add("*.awsapps.com")
464class SecurityMiddleware:
465 """Middleware that sets Content-Security-Policy."""
467 def __init__(self, get_response=None) -> None:
468 self.get_response = get_response
470 def __call__(self, request: AuthenticatedHttpRequest):
471 response = self.get_response(request)
472 csp_builder = CSPBuilder(request, response)
474 response["Content-Security-Policy"] = "; ".join(
475 f"{name} {' '.join(rules)}"
476 for name, rules in csp_builder.directives.items()
477 )
478 if settings.SENTRY_SECURITY: 478 ↛ 479line 478 didn't jump to line 479 because the condition on line 478 was never true
479 response["Content-Security-Policy"] += (
480 f" report-uri {settings.SENTRY_SECURITY}"
481 )
482 response["Expect-CT"] = (
483 f'max-age=86400, enforce, report-uri="{settings.SENTRY_SECURITY}"'
484 )
486 # Opt-out from Google FLoC
487 response["Permissions-Policy"] = "interest-cohort=()"
489 return response