Coverage for src/backend/InvenTree/InvenTree/middleware.py: 61%
170 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
1"""Middleware for InvenTree."""
3import sys
4from typing import Optional
5from urllib.parse import urlsplit
7from django.conf import settings
8from django.contrib.auth.middleware import PersistentRemoteUserMiddleware
9from django.http import HttpRequest, HttpResponse, JsonResponse
10from django.shortcuts import redirect, render
11from django.urls import resolve, reverse, reverse_lazy
12from django.utils.deprecation import MiddlewareMixin
13from django.utils.http import is_same_domain, url_has_allowed_host_and_scheme
14from django.utils.translation import gettext_lazy as _
16import structlog
17from error_report.middleware import ExceptionProcessor
19import InvenTree.helpers
20from common.settings import get_global_setting
21from InvenTree.cache import create_session_cache, delete_session_cache
22from InvenTree.config import CONFIG_LOOKUPS, inventreeInstaller
23from InvenTree.version import (
24 inventreeApiVersion,
25 inventreePythonVersion,
26 inventreeVersion,
27)
28from users.models import ApiToken
30logger = structlog.get_logger('inventree')
33def get_token_from_request(request):
34 """Extract token information from a request object."""
35 auth_keys = ['Authorization', 'authorization']
36 token_keys = ['token', 'bearer']
38 for k in auth_keys:
39 if auth_header := request.headers.get(k, None):
40 auth_header = auth_header.strip().lower().split()
42 if len(auth_header) > 1: 42 ↛ 38line 42 didn't jump to line 38 because the condition on line 42 was always true
43 if auth_header[0].strip().lower().replace(':', '') in token_keys: 43 ↛ 44line 43 didn't jump to line 44 because the condition on line 43 was never true
44 token = auth_header[1]
45 return token
47 return None
50def ensure_slashes(path: str):
51 """Ensure that slashes are surrounding the passed path."""
52 if not path.startswith('/'):
53 path = f'/{path}'
54 if not path.endswith('/'):
55 path = f'{path}/'
56 return path
59# List of target URL endpoints where *do not* want to redirect to
60urls = [
61 reverse_lazy('account_login'),
62 reverse_lazy('admin:login'),
63 reverse_lazy('admin:logout'),
64]
66paths_ignore_handling = [
67 '/api/',
68 '/plugin/',
69 reverse('auth-check'),
70 settings.MEDIA_URL,
71 settings.STATIC_URL,
72]
73"""Paths that should not use InvenTrees own auth rejection behaviour, no host checking or redirecting. Security
74 are still enforced."""
75paths_own_security = [
76 '/api/', # DRF handles API
77 '/o/', # oAuth2 library - has its own auth model
78 '/anymail/', # Mails - webhooks etc
79 '/accounts/', # allauth account management - has its own auth model
80 '/assets/', # Web assets - only used for testing, no security model needed
81 ensure_slashes(
82 settings.STATIC_URL
83 ), # Static files - static files are considered safe to serve
84 ensure_slashes(
85 settings.FRONTEND_URL_BASE
86 ), # Frontend files - frontend paths have their own security model
87]
88"""Paths that handle their own security model."""
89pages_mfa_bypass = [
90 'api-user-meta',
91 'api-user-me',
92 'api-user-roles',
93 'api-inventree-info',
94 'api-token',
95 # web platform urls
96 'password_reset_confirm',
97 'index',
98 'web',
99 'web-wildcard',
100 'web-assets',
101]
102"""Exact page names that bypass MFA enforcement - normal security model is still enforced."""
103apps_mfa_bypass = [
104 'headless' # Headless allauth app - has its own security model
105]
106"""App namespaces that bypass MFA enforcement - normal security model is still enforced."""
109def csrf_failure(request, reason=''):
110 """Custom CSRF failure handler.
112 Returns a JSON response for API/headless requests so the frontend can
113 provide a meaningful error message to the user
114 """
115 from django.views.csrf import csrf_failure as django_default
117 if (
118 request.path.startswith('/_allauth/')
119 or request.path.startswith('/api/')
120 or 'application/json' in request.headers.get('Accept', '')
121 or 'application/json' in request.headers.get('Content-Type', '')
122 ):
123 return JsonResponse(
124 {
125 'detail': _(
126 'CSRF verification failed. Ensure INVENTREE_SITE_URL and INVENTREE_TRUSTED_ORIGINS are configured correctly.'
127 )
128 },
129 status=403,
130 )
132 return django_default(request, reason=reason)
135class AuthRequiredMiddleware:
136 """Check for user to be authenticated."""
138 def __init__(self, get_response):
139 """Save response object."""
140 self.get_response = get_response
142 def check_token(self, request) -> bool:
143 """Check if the user is authenticated via token."""
144 if token := get_token_from_request(request): 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true
145 request.token = token
146 # Does the provided token match a valid user?
147 try:
148 token = ApiToken.objects.get(key=token)
150 if token.active and token.user:
151 # Provide the user information to the request
152 request.user = token.user
153 return True
154 except ApiToken.DoesNotExist: # pragma: no cover
155 logger.warning(
156 'Access denied for unknown token %s',
157 InvenTree.helpers.sanitize_token(str(token)),
158 ) # pragma: no cover
160 return False
162 def __call__(self, request):
163 """Check if user needs to be authenticated and is.
165 Redirects to login if not authenticated.
166 """
167 path: str = request.path_info
168 # Code to be executed for each request before
169 # the view (and later middleware) are called.
171 assert hasattr(request, 'user')
173 # API requests that are handled elsewhere
174 if any(path.startswith(a) for a in paths_own_security):
175 return self.get_response(request)
177 # Is the function exempt from auth requirements?
178 path_func = resolve(request.path).func
179 if getattr(path_func, 'auth_exempt', False) is True: 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true
180 return self.get_response(request)
182 if not request.user.is_authenticated and not (
183 path == f'/{settings.FRONTEND_URL_BASE}' or self.check_token(request)
184 ):
185 """
186 Normally, a web-based session would use csrftoken based authentication.
188 However when running an external application (e.g. the InvenTree app or Python library),
189 we must validate the user token manually.
190 """
191 if path not in urls and not any( 191 ↛ 205line 191 didn't jump to line 205 because the condition on line 191 was always true
192 path.startswith(p) for p in paths_ignore_handling
193 ):
194 # Validate next url is safe to redirect to
195 next_url = request.path
196 if not url_has_allowed_host_and_scheme( 196 ↛ 201line 196 didn't jump to line 201 because the condition on line 196 was never true
197 url=next_url,
198 allowed_hosts=settings.ALLOWED_HOSTS,
199 require_https=request.is_secure(),
200 ):
201 return redirect(str(reverse_lazy('account_login')))
202 # Save the 'next' parameter to pass through to the login view
203 return redirect(f'{reverse_lazy("account_login")}?next={next_url}')
204 # Return a 401 (Unauthorized) response code for this request
205 return HttpResponse('Unauthorized', status=401)
207 response = self.get_response(request)
208 return response
211class Check2FAMiddleware(MiddlewareMixin):
212 """Ensure that users have two-factor authentication enabled before they have access restricted endpoints.
214 Adapted from https://github.com/pennersr/django-allauth/issues/3649
215 """
217 require_2fa_message = _(
218 'You must enable two-factor authentication before doing anything else.'
219 )
221 def on_require_2fa(self, request: HttpRequest) -> HttpResponse:
222 """Force user to mfa activation."""
223 return JsonResponse(
224 {'id': 'mfa_register', 'error': self.require_2fa_message}, status=401
225 )
227 def is_allowed_page(self, request: HttpRequest) -> bool:
228 """Check if the current page can be accessed without mfa."""
229 match = request.resolver_match
230 return (
231 False
232 if match is None
233 else any(ref in apps_mfa_bypass for ref in match.app_names)
234 or match.url_name in pages_mfa_bypass
235 or match.route == 'favicon.ico'
236 )
238 def is_multifactor_logged_in(self, request: HttpRequest) -> bool:
239 """Check if the user is logged in with multifactor authentication."""
240 from allauth.account.authentication import get_authentication_records
241 from allauth.mfa.utils import is_mfa_enabled
242 from allauth.mfa.webauthn.internal.flows import did_use_passwordless_login
244 authns = get_authentication_records(request)
246 return is_mfa_enabled(request.user) and (
247 did_use_passwordless_login(request)
248 or any(record.get('method') == 'mfa' for record in authns)
249 )
251 def process_view(
252 self, request: HttpRequest, view_func, view_args, view_kwargs
253 ) -> Optional[HttpResponse]:
254 """Determine if the server is set up enforce 2fa registration."""
255 from django.conf import settings
257 # Exit early if MFA is not enabled
258 if not settings.MFA_ENABLED: 258 ↛ 259line 258 didn't jump to line 259 because the condition on line 258 was never true
259 return None
261 if request.user.is_anonymous: 261 ↛ 263line 261 didn't jump to line 263 because the condition on line 261 was always true
262 return None
263 if self.is_allowed_page(request):
264 return None
265 if self.is_multifactor_logged_in(request):
266 return None
267 if getattr(
268 request, 'token', get_token_from_request(request)
269 ): # Token based login can not do MFA
270 return None
272 if self.enforce_2fa(request):
273 return self.on_require_2fa(request)
274 return None
276 def enforce_2fa(self, request):
277 """Use setting to check if MFA should be enforced."""
278 return get_global_setting(
279 'LOGIN_ENFORCE_MFA', None, 'INVENTREE_LOGIN_ENFORCE_MFA'
280 )
283class InvenTreeRemoteUserMiddleware(PersistentRemoteUserMiddleware):
284 """Middleware to check if HTTP-header based auth is enabled and to set it up."""
286 header = settings.REMOTE_LOGIN_HEADER
288 def process_request(self, request):
289 """Check if proxy login is enabled."""
290 if not settings.REMOTE_LOGIN: 290 ↛ 293line 290 didn't jump to line 293 because the condition on line 290 was always true
291 return
293 return super().process_request(request)
296class InvenTreeExceptionProcessor(ExceptionProcessor):
297 """Custom exception processor that respects blocked errors."""
299 def process_exception(self, request, exception):
300 """Check if kind is ignored before processing."""
301 kind, _info, _data = sys.exc_info()
303 # Check if the error is on the ignore list
304 if kind in settings.IGNORED_ERRORS:
305 return
307 from error_report.settings import ERROR_DETAIL_SETTINGS
309 # Error reporting is disabled
310 if not ERROR_DETAIL_SETTINGS.get('ERROR_DETAIL_ENABLE', True):
311 return
313 path = request.build_absolute_uri()
315 # Truncate the path to a reasonable length
316 # Otherwise we get a database error,
317 # because the path field is limited to 200 characters
318 if len(path) > 200:
319 path = path[:195] + '...'
321 # Pass off to the exception reporter
322 from InvenTree.exceptions import log_error
324 log_error(path)
327class InvenTreeRequestCacheMiddleware(MiddlewareMixin):
328 """Middleware to perform caching against the request object.
330 This middleware is used to cache data against the request object,
331 which can be used to store data for the duration of the request.
333 In this fashion, we can avoid hitting the external cache multiple times,
334 much less the database!
335 """
337 def process_request(self, request):
338 """Create a request-specific cache object."""
339 create_session_cache(request)
341 def process_response(self, request, response):
342 """Clear the cache object."""
343 delete_session_cache()
344 return response
347class InvenTreeHostSettingsMiddleware(MiddlewareMixin):
348 """Middleware to check the host settings.
350 Especially SITE_URL, trusted_origins.
351 """
353 def process_request(self, request):
354 """Check the host settings."""
355 # Debug setups do not enforce these checks so we ignore that case
356 if settings.DEBUG: 356 ↛ 357line 356 didn't jump to line 357 because the condition on line 356 was never true
357 return None
359 # Handle commonly ignored paths that might also work without a correct setup (api, auth)
360 path = request.path_info
361 if path in urls or any(path.startswith(p) for p in paths_ignore_handling):
362 return None
364 # treat the accessed scheme and host
365 accessed_scheme = request._current_scheme_host
366 referrer = urlsplit(accessed_scheme)
368 site_url = urlsplit(settings.SITE_URL)
370 # Check if the accessed URL matches the SITE_URL setting
371 site_url_match = (
372 (
373 # Exact match on domain
374 is_same_domain(referrer.netloc, site_url.netloc)
375 and referrer.scheme == site_url.scheme
376 )
377 or (
378 # Lax protocol match, accessed URL starts with SITE_URL
379 settings.SITE_LAX_PROTOCOL_CHECK
380 and accessed_scheme.startswith(settings.SITE_URL)
381 )
382 or (
383 # Lax protocol match, same domain
384 settings.SITE_LAX_PROTOCOL_CHECK
385 and referrer.hostname == site_url.hostname
386 )
387 )
389 if not site_url_match: 389 ↛ 408line 389 didn't jump to line 408 because the condition on line 389 was always true
390 # The accessed URL does not match the SITE_URL setting
391 if ( 391 ↛ 396line 391 didn't jump to line 396 because the condition on line 391 was never true
392 isinstance(settings.CSRF_TRUSTED_ORIGINS, list)
393 and len(settings.CSRF_TRUSTED_ORIGINS) > 1
394 ):
395 # The used url might not be the primary url - next check determines if in a trusted origins
396 pass
397 else:
398 source = CONFIG_LOOKUPS.get('INVENTREE_SITE_URL', {}).get(
399 'source', 'unknown'
400 )
401 dpl_method = inventreeInstaller()
402 msg = f'INVE-E7: The visited path `{accessed_scheme}` does not match the SITE_URL `{settings.SITE_URL}`. The INVENTREE_SITE_URL is set via `{source}` config method - deployment method `{dpl_method}`'
403 logger.error(msg)
404 return render(
405 request, 'config_error.html', {'error_message': msg}, status=500
406 )
408 trusted_origins_match = (
409 # Matching domain found in allowed origins
410 any(
411 is_same_domain(referrer.netloc, host)
412 for host in [
413 urlsplit(origin).netloc.lstrip('*')
414 for origin in settings.CSRF_TRUSTED_ORIGINS
415 ]
416 )
417 ) or (
418 # Lax protocol match allowed
419 settings.SITE_LAX_PROTOCOL_CHECK
420 and any(
421 referrer.hostname == urlsplit(origin).hostname
422 for origin in settings.CSRF_TRUSTED_ORIGINS
423 )
424 )
426 # Check trusted origins
427 if not trusted_origins_match:
428 msg = f'INVE-E7: The used path `{accessed_scheme}` is not in the TRUSTED_ORIGINS'
429 logger.error(msg)
430 return render(
431 request, 'config_error.html', {'error_message': msg}, status=500
432 )
434 # All checks passed
435 return None
438class InvenTreeVersionHeaderMiddleware(MiddlewareMixin):
439 """Middleware to add the InvenTree version header to all responses."""
441 def process_response(self, request, response):
442 """Add the InvenTree version header to the response."""
443 response['X-InvenTree-Version'] = inventreeVersion()
444 response['X-InvenTree-API'] = inventreeApiVersion()
445 response['X-InvenTree-Python'] = inventreePythonVersion()
446 response['X-InvenTree-Installer'] = inventreeInstaller()
447 return response