diff --git a/src/backend/InvenTree/InvenTree/middleware.py b/src/backend/InvenTree/InvenTree/middleware.py index 0c6ff3c94b18..57c8f20dea92 100644 --- a/src/backend/InvenTree/InvenTree/middleware.py +++ b/src/backend/InvenTree/InvenTree/middleware.py @@ -1,447 +1,447 @@ -"""Middleware for InvenTree.""" - -import sys -from typing import Optional -from urllib.parse import urlsplit - -from django.conf import settings -from django.contrib.auth.middleware import PersistentRemoteUserMiddleware -from django.http import HttpRequest, HttpResponse, JsonResponse -from django.shortcuts import redirect, render -from django.urls import resolve, reverse, reverse_lazy -from django.utils.deprecation import MiddlewareMixin -from django.utils.http import is_same_domain, url_has_allowed_host_and_scheme -from django.utils.translation import gettext_lazy as _ - -import structlog -from error_report.middleware import ExceptionProcessor - -import InvenTree.helpers -from common.settings import get_global_setting -from InvenTree.cache import create_session_cache, delete_session_cache -from InvenTree.config import CONFIG_LOOKUPS, inventreeInstaller -from InvenTree.version import ( - inventreeApiVersion, - inventreePythonVersion, - inventreeVersion, -) -from users.models import ApiToken - -logger = structlog.get_logger('inventree') - - -def get_token_from_request(request): - """Extract token information from a request object.""" - auth_keys = ['Authorization', 'authorization'] - token_keys = ['token', 'bearer'] - - for k in auth_keys: - if auth_header := request.headers.get(k, None): - auth_header = auth_header.strip().lower().split() - - if len(auth_header) > 1: - if auth_header[0].strip().lower().replace(':', '') in token_keys: - token = auth_header[1] - return token - - return None - - -def ensure_slashes(path: str): - """Ensure that slashes are surrounding the passed path.""" - if not path.startswith('/'): - path = f'/{path}' - if not path.endswith('/'): - path = f'{path}/' - return path - - -# List of target URL endpoints where *do not* want to redirect to -urls = [ - reverse_lazy('account_login'), - reverse_lazy('admin:login'), - reverse_lazy('admin:logout'), -] - -paths_ignore_handling = [ - '/api/', - '/plugin/', - reverse('auth-check'), - settings.MEDIA_URL, - settings.STATIC_URL, -] -"""Paths that should not use InvenTrees own auth rejection behaviour, no host checking or redirecting. Security - are still enforced.""" -paths_own_security = [ - '/api/', # DRF handles API - '/o/', # oAuth2 library - has its own auth model - '/anymail/', # Mails - webhooks etc - '/accounts/', # allauth account management - has its own auth model - '/assets/', # Web assets - only used for testing, no security model needed - ensure_slashes( - settings.STATIC_URL - ), # Static files - static files are considered safe to serve - ensure_slashes( - settings.FRONTEND_URL_BASE - ), # Frontend files - frontend paths have their own security model -] -"""Paths that handle their own security model.""" -pages_mfa_bypass = [ - 'api-user-meta', - 'api-user-me', - 'api-user-roles', - 'api-inventree-info', - 'api-token', - # web platform urls - 'password_reset_confirm', - 'index', - 'web', - 'web-wildcard', - 'web-assets', -] -"""Exact page names that bypass MFA enforcement - normal security model is still enforced.""" -apps_mfa_bypass = [ - 'headless' # Headless allauth app - has its own security model -] -"""App namespaces that bypass MFA enforcement - normal security model is still enforced.""" - - -def csrf_failure(request, reason=''): - """Custom CSRF failure handler. - - Returns a JSON response for API/headless requests so the frontend can - provide a meaningful error message to the user - """ - from django.views.csrf import csrf_failure as django_default - - if ( - request.path.startswith('/_allauth/') - or request.path.startswith('/api/') - or 'application/json' in request.headers.get('Accept', '') - or 'application/json' in request.headers.get('Content-Type', '') - ): - return JsonResponse( - { - 'detail': _( - 'CSRF verification failed. Ensure INVENTREE_SITE_URL and INVENTREE_TRUSTED_ORIGINS are configured correctly.' - ) - }, - status=403, - ) - - return django_default(request, reason=reason) - - -class AuthRequiredMiddleware: - """Check for user to be authenticated.""" - - def __init__(self, get_response): - """Save response object.""" - self.get_response = get_response - - def check_token(self, request) -> bool: - """Check if the user is authenticated via token.""" - if token := get_token_from_request(request): - request.token = token - # Does the provided token match a valid user? - try: - token = ApiToken.objects.get(key=token) - - if token.active and token.user: - # Provide the user information to the request - request.user = token.user - return True - except ApiToken.DoesNotExist: # pragma: no cover - logger.warning( - 'Access denied for unknown token %s', - InvenTree.helpers.sanitize_token(str(token)), - ) # pragma: no cover - - return False - - def __call__(self, request): - """Check if user needs to be authenticated and is. - - Redirects to login if not authenticated. - """ - path: str = request.path_info - # Code to be executed for each request before - # the view (and later middleware) are called. - - assert hasattr(request, 'user') - - # API requests that are handled elsewhere - if any(path.startswith(a) for a in paths_own_security): - return self.get_response(request) - - # Is the function exempt from auth requirements? - path_func = resolve(request.path).func - if getattr(path_func, 'auth_exempt', False) is True: - return self.get_response(request) - - if not request.user.is_authenticated and not ( - path == f'/{settings.FRONTEND_URL_BASE}' or self.check_token(request) - ): - """ - Normally, a web-based session would use csrftoken based authentication. - - However when running an external application (e.g. the InvenTree app or Python library), - we must validate the user token manually. - """ - if path not in urls and not any( - path.startswith(p) for p in paths_ignore_handling - ): - # Validate next url is safe to redirect to - next_url = request.path - if not url_has_allowed_host_and_scheme( - url=next_url, - allowed_hosts=settings.ALLOWED_HOSTS, - require_https=request.is_secure(), - ): - return redirect(str(reverse_lazy('account_login'))) - # Save the 'next' parameter to pass through to the login view - return redirect(f'{reverse_lazy("account_login")}?next={next_url}') - # Return a 401 (Unauthorized) response code for this request - return HttpResponse('Unauthorized', status=401) - - response = self.get_response(request) - return response - - -class Check2FAMiddleware(MiddlewareMixin): - """Ensure that users have two-factor authentication enabled before they have access restricted endpoints. - - Adapted from https://github.com/pennersr/django-allauth/issues/3649 - """ - - require_2fa_message = _( - 'You must enable two-factor authentication before doing anything else.' - ) - - def on_require_2fa(self, request: HttpRequest) -> HttpResponse: - """Force user to mfa activation.""" - return JsonResponse( - {'id': 'mfa_register', 'error': self.require_2fa_message}, status=401 - ) - - def is_allowed_page(self, request: HttpRequest) -> bool: - """Check if the current page can be accessed without mfa.""" - match = request.resolver_match - return ( - False - if match is None - else any(ref in apps_mfa_bypass for ref in match.app_names) - or match.url_name in pages_mfa_bypass - or match.route == 'favicon.ico' - ) - - def is_multifactor_logged_in(self, request: HttpRequest) -> bool: - """Check if the user is logged in with multifactor authentication.""" - from allauth.account.authentication import get_authentication_records - from allauth.mfa.utils import is_mfa_enabled - from allauth.mfa.webauthn.internal.flows import did_use_passwordless_login - - authns = get_authentication_records(request) - - return is_mfa_enabled(request.user) and ( - did_use_passwordless_login(request) - or any(record.get('method') == 'mfa' for record in authns) - ) - - def process_view( - self, request: HttpRequest, view_func, view_args, view_kwargs - ) -> Optional[HttpResponse]: - """Determine if the server is set up enforce 2fa registration.""" - from django.conf import settings - - # Exit early if MFA is not enabled - if not settings.MFA_ENABLED: - return None - - if request.user.is_anonymous: - return None - if self.is_allowed_page(request): - return None - if self.is_multifactor_logged_in(request): - return None - if getattr( - request, 'token', get_token_from_request(request) - ): # Token based login can not do MFA - return None - - if self.enforce_2fa(request): - return self.on_require_2fa(request) - return None - - def enforce_2fa(self, request): - """Use setting to check if MFA should be enforced.""" - return get_global_setting( - 'LOGIN_ENFORCE_MFA', None, 'INVENTREE_LOGIN_ENFORCE_MFA' - ) - - -class InvenTreeRemoteUserMiddleware(PersistentRemoteUserMiddleware): - """Middleware to check if HTTP-header based auth is enabled and to set it up.""" - - header = settings.REMOTE_LOGIN_HEADER - - def process_request(self, request): - """Check if proxy login is enabled.""" - if not settings.REMOTE_LOGIN: - return - - return super().process_request(request) - - -class InvenTreeExceptionProcessor(ExceptionProcessor): - """Custom exception processor that respects blocked errors.""" - - def process_exception(self, request, exception): - """Check if kind is ignored before processing.""" - kind, _info, _data = sys.exc_info() - - # Check if the error is on the ignore list - if kind in settings.IGNORED_ERRORS: - return - - from error_report.settings import ERROR_DETAIL_SETTINGS - - # Error reporting is disabled - if not ERROR_DETAIL_SETTINGS.get('ERROR_DETAIL_ENABLE', True): - return - - path = request.build_absolute_uri() - - # Truncate the path to a reasonable length - # Otherwise we get a database error, - # because the path field is limited to 200 characters - if len(path) > 200: - path = path[:195] + '...' - - # Pass off to the exception reporter - from InvenTree.exceptions import log_error - - log_error(path) - - -class InvenTreeRequestCacheMiddleware(MiddlewareMixin): - """Middleware to perform caching against the request object. - - This middleware is used to cache data against the request object, - which can be used to store data for the duration of the request. - - In this fashion, we can avoid hitting the external cache multiple times, - much less the database! - """ - - def process_request(self, request): - """Create a request-specific cache object.""" - create_session_cache(request) - - def process_response(self, request, response): - """Clear the cache object.""" - delete_session_cache() - return response - - -class InvenTreeHostSettingsMiddleware(MiddlewareMixin): - """Middleware to check the host settings. - - Especially SITE_URL, trusted_origins. - """ - - def process_request(self, request): - """Check the host settings.""" - # Debug setups do not enforce these checks so we ignore that case - if settings.DEBUG: - return None - - # Handle commonly ignored paths that might also work without a correct setup (api, auth) - path = request.path_info - if path in urls or any(path.startswith(p) for p in paths_ignore_handling): - return None - - # treat the accessed scheme and host - accessed_scheme = request._current_scheme_host - referrer = urlsplit(accessed_scheme) - - site_url = urlsplit(settings.SITE_URL) - - # Check if the accessed URL matches the SITE_URL setting - site_url_match = ( - ( - # Exact match on domain - is_same_domain(referrer.netloc, site_url.netloc) - and referrer.scheme == site_url.scheme - ) - or ( - # Lax protocol match, accessed URL starts with SITE_URL - settings.SITE_LAX_PROTOCOL_CHECK - and accessed_scheme.startswith(settings.SITE_URL) - ) - or ( - # Lax protocol match, same domain - settings.SITE_LAX_PROTOCOL_CHECK - and referrer.hostname == site_url.hostname - ) - ) - - if not site_url_match: - # The accessed URL does not match the SITE_URL setting - if ( - isinstance(settings.CSRF_TRUSTED_ORIGINS, list) - and len(settings.CSRF_TRUSTED_ORIGINS) > 1 - ): - # The used url might not be the primary url - next check determines if in a trusted origins - pass - else: - source = CONFIG_LOOKUPS.get('INVENTREE_SITE_URL', {}).get( - 'source', 'unknown' - ) - dpl_method = inventreeInstaller() - 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}`' - logger.error(msg) - return render( - request, 'config_error.html', {'error_message': msg}, status=500 - ) - - trusted_origins_match = ( - # Matching domain found in allowed origins - any( - is_same_domain(referrer.netloc, host) - for host in [ - urlsplit(origin).netloc.lstrip('*') - for origin in settings.CSRF_TRUSTED_ORIGINS - ] - ) - ) or ( - # Lax protocol match allowed - settings.SITE_LAX_PROTOCOL_CHECK - and any( - referrer.hostname == urlsplit(origin).hostname - for origin in settings.CSRF_TRUSTED_ORIGINS - ) - ) - - # Check trusted origins - if not trusted_origins_match: - msg = f'INVE-E7: The used path `{accessed_scheme}` is not in the TRUSTED_ORIGINS' - logger.error(msg) - return render( - request, 'config_error.html', {'error_message': msg}, status=500 - ) - - # All checks passed - return None - - -class InvenTreeVersionHeaderMiddleware(MiddlewareMixin): - """Middleware to add the InvenTree version header to all responses.""" - - def process_response(self, request, response): - """Add the InvenTree version header to the response.""" - response['X-InvenTree-Version'] = inventreeVersion() - response['X-InvenTree-API'] = inventreeApiVersion() - response['X-InvenTree-Python'] = inventreePythonVersion() - response['X-InvenTree-Installer'] = inventreeInstaller() - return response +"""Middleware for InvenTree.""" + +import sys +from typing import Optional +from urllib.parse import urlsplit + +from django.conf import settings +from django.contrib.auth.middleware import PersistentRemoteUserMiddleware +from django.http import HttpRequest, HttpResponse, JsonResponse +from django.shortcuts import redirect, render +from django.urls import resolve, reverse, reverse_lazy +from django.utils.deprecation import MiddlewareMixin +from django.utils.http import is_same_domain, url_has_allowed_host_and_scheme +from django.utils.translation import gettext_lazy as _ + +import structlog +from error_report.middleware import ExceptionProcessor + +import InvenTree.helpers +from common.settings import get_global_setting +from InvenTree.cache import create_session_cache, delete_session_cache +from InvenTree.config import CONFIG_LOOKUPS, inventreeInstaller +from InvenTree.version import ( + inventreeApiVersion, + inventreePythonVersion, + inventreeVersion, +) +from users.models import ApiToken + +logger = structlog.get_logger('inventree') + + +def get_token_from_request(request): + """Extract token information from a request object.""" + auth_keys = ['Authorization', 'authorization'] + token_keys = ['token', 'bearer'] + + for k in auth_keys: + if auth_header := request.headers.get(k, None): + auth_header = auth_header.strip().lower().split() + + if len(auth_header) > 1: + if auth_header[0].strip().lower().replace(':', '') in token_keys: + token = auth_header[1] + return token + + return None + + +def ensure_slashes(path: str): + """Ensure that slashes are surrounding the passed path.""" + if not path.startswith('/'): + path = f'/{path}' + if not path.endswith('/'): + path = f'{path}/' + return path + + +# List of target URL endpoints where *do not* want to redirect to +urls = [ + reverse_lazy('account_login'), + reverse_lazy('admin:login'), + reverse_lazy('admin:logout'), +] + +paths_ignore_handling = [ + '/api/', + '/plugin/', + reverse('auth-check'), + settings.MEDIA_URL, + settings.STATIC_URL, +] +"""Paths that should not use InvenTrees own auth rejection behaviour, no host checking or redirecting. Security + are still enforced.""" +paths_own_security = [ + '/api/', # DRF handles API + '/o/', # oAuth2 library - has its own auth model + '/anymail/', # Mails - webhooks etc + '/accounts/', # allauth account management - has its own auth model + '/assets/', # Web assets - only used for testing, no security model needed + ensure_slashes( + settings.STATIC_URL + ), # Static files - static files are considered safe to serve + ensure_slashes( + settings.FRONTEND_URL_BASE + ), # Frontend files - frontend paths have their own security model +] +"""Paths that handle their own security model.""" +pages_mfa_bypass = [ + 'api-user-meta', + 'api-user-me', + 'api-user-roles', + 'api-inventree-info', + 'api-token', + # web platform urls + 'password_reset_confirm', + 'index', + 'web', + 'web-wildcard', + 'web-assets', +] +"""Exact page names that bypass MFA enforcement - normal security model is still enforced.""" +apps_mfa_bypass = [ + 'headless' # Headless allauth app - has its own security model +] +"""App namespaces that bypass MFA enforcement - normal security model is still enforced.""" + + +def csrf_failure(request, reason=''): + """Custom CSRF failure handler. + + Returns a JSON response for API/headless requests so the frontend can + provide a meaningful error message to the user + """ + from django.views.csrf import csrf_failure as django_default + + if ( + request.path.startswith('/_allauth/') + or request.path.startswith('/api/') + or 'application/json' in request.headers.get('Accept', '') + or 'application/json' in request.headers.get('Content-Type', '') + ): + return JsonResponse( + { + 'detail': _( + 'CSRF verification failed. Ensure INVENTREE_SITE_URL and INVENTREE_TRUSTED_ORIGINS are configured correctly.' + ) + }, + status=403, + ) + + return django_default(request, reason=reason) + + +class AuthRequiredMiddleware: + """Check for user to be authenticated.""" + + def __init__(self, get_response): + """Save response object.""" + self.get_response = get_response + + def check_token(self, request) -> bool: + """Check if the user is authenticated via token.""" + if token := get_token_from_request(request): + request.token = token + # Does the provided token match a valid user? + try: + token = ApiToken.objects.get(key=token) + + if token.active and token.user: + # Provide the user information to the request + request.user = token.user + return True + except ApiToken.DoesNotExist: # pragma: no cover + logger.warning( + 'Access denied for unknown token %s', + InvenTree.helpers.sanitize_token(str(token)), + ) # pragma: no cover + + return False + + def __call__(self, request): + """Check if user needs to be authenticated and is. + + Redirects to login if not authenticated. + """ + path: str = request.path_info + # Code to be executed for each request before + # the view (and later middleware) are called. + + assert hasattr(request, 'user') + + # API requests that are handled elsewhere + if any(path.startswith(a) for a in paths_own_security): + return self.get_response(request) + + # Is the function exempt from auth requirements? + path_func = resolve(request.path).func + if getattr(path_func, 'auth_exempt', False) is True: + return self.get_response(request) + + if not request.user.is_authenticated and not ( + path == f'/{settings.FRONTEND_URL_BASE}' or self.check_token(request) + ): + """ + Normally, a web-based session would use csrftoken based authentication. + + However when running an external application (e.g. the InvenTree app or Python library), + we must validate the user token manually. + """ + if path not in urls and not any( + path.startswith(p) for p in paths_ignore_handling + ): + # Validate next url is safe to redirect to + next_url = request.path + if not url_has_allowed_host_and_scheme( + url=next_url, + allowed_hosts=settings.ALLOWED_HOSTS, + require_https=request.is_secure(), + ): + return redirect(str(reverse_lazy('account_login'))) + # Save the 'next' parameter to pass through to the login view + return redirect(f'{reverse_lazy("account_login")}?next={next_url}') + # Return a 401 (Unauthorized) response code for this request + return HttpResponse('Unauthorized', status=401) + + response = self.get_response(request) + return response + + +class Check2FAMiddleware(MiddlewareMixin): + """Ensure that users have two-factor authentication enabled before they have access restricted endpoints. + + Adapted from https://github.com/pennersr/django-allauth/issues/3649 + """ + + require_2fa_message = _( + 'You must enable two-factor authentication before doing anything else.' + ) + + def on_require_2fa(self, request: HttpRequest) -> HttpResponse: + """Force user to mfa activation.""" + return JsonResponse( + {'id': 'mfa_register', 'error': self.require_2fa_message}, status=401 + ) + + def is_allowed_page(self, request: HttpRequest) -> bool: + """Check if the current page can be accessed without mfa.""" + match = request.resolver_match + return ( + False + if match is None + else any(ref in apps_mfa_bypass for ref in match.app_names) + or match.url_name in pages_mfa_bypass + or match.route == 'favicon.ico' + ) + + def is_multifactor_logged_in(self, request: HttpRequest) -> bool: + """Check if the user is logged in with multifactor authentication.""" + from allauth.account.authentication import get_authentication_records + from allauth.mfa.utils import is_mfa_enabled + from allauth.mfa.webauthn.internal.flows import did_use_passwordless_login + + authns = get_authentication_records(request) + + return is_mfa_enabled(request.user) and ( + did_use_passwordless_login(request) + or any(record.get('method') == 'mfa' for record in authns) + ) + + def process_view( + self, request: HttpRequest, view_func, view_args, view_kwargs + ) -> Optional[HttpResponse]: + """Determine if the server is set up enforce 2fa registration.""" + from django.conf import settings + + # Exit early if MFA is not enabled + if not settings.MFA_ENABLED: + return None + + if request.user.is_anonymous: + return None + if self.is_allowed_page(request): + return None + if self.is_multifactor_logged_in(request): + return None + if getattr( + request, 'token', get_token_from_request(request) + ): # Token based login can not do MFA + return None + + if self.enforce_2fa(request): + return self.on_require_2fa(request) + return None + + def enforce_2fa(self, request): + """Use setting to check if MFA should be enforced.""" + return get_global_setting( + 'LOGIN_ENFORCE_MFA', None, 'INVENTREE_LOGIN_ENFORCE_MFA' + ) + + +class InvenTreeRemoteUserMiddleware(PersistentRemoteUserMiddleware): + """Middleware to check if HTTP-header based auth is enabled and to set it up.""" + + header = settings.REMOTE_LOGIN_HEADER + + def process_request(self, request): + """Check if proxy login is enabled.""" + if not settings.REMOTE_LOGIN: + return + + return super().process_request(request) + + +class InvenTreeExceptionProcessor(ExceptionProcessor): + """Custom exception processor that respects blocked errors.""" + + def process_exception(self, request, exception): + """Check if kind is ignored before processing.""" + kind, _info, _data = sys.exc_info() + + # Check if the error is on the ignore list + if kind in settings.IGNORED_ERRORS: + return + + from error_report.settings import ERROR_DETAIL_SETTINGS + + # Error reporting is disabled + if not ERROR_DETAIL_SETTINGS.get('ERROR_DETAIL_ENABLE', True): + return + + path = request.build_absolute_uri() + + # Truncate the path to a reasonable length + # Otherwise we get a database error, + # because the path field is limited to 200 characters + if len(path) > 200: + path = path[:195] + '...' + + # Pass off to the exception reporter + from InvenTree.exceptions import log_error + + log_error(path) + + +class InvenTreeRequestCacheMiddleware(MiddlewareMixin): + """Middleware to perform caching against the request object. + + This middleware is used to cache data against the request object, + which can be used to store data for the duration of the request. + + In this fashion, we can avoid hitting the external cache multiple times, + much less the database! + """ + + def process_request(self, request): + """Create a request-specific cache object.""" + create_session_cache(request) + + def process_response(self, request, response): + """Clear the cache object.""" + delete_session_cache() + return response + + +class InvenTreeHostSettingsMiddleware(MiddlewareMixin): + """Middleware to check the host settings. + + Especially SITE_URL, trusted_origins. + """ + + def process_request(self, request): + """Check the host settings.""" + # Debug setups do not enforce these checks so we ignore that case + if settings.DEBUG: + return None + + # Handle commonly ignored paths that might also work without a correct setup (api, auth) + path = request.path_info + if path in urls or any(path.startswith(p) for p in paths_ignore_handling): + return None + + # treat the accessed scheme and host + accessed_scheme = request._current_scheme_host + referrer = urlsplit(accessed_scheme) + + site_url = urlsplit(settings.SITE_URL) + + # Check if the accessed URL matches the SITE_URL setting + site_url_match = ( + ( + # Exact match on domain + is_same_domain(referrer.netloc, site_url.netloc) + and referrer.scheme == site_url.scheme + ) + or ( + # Lax protocol match, accessed URL starts with SITE_URL + settings.SITE_LAX_PROTOCOL_CHECK + and accessed_scheme.startswith(settings.SITE_URL) + ) + or ( + # Lax protocol match, same domain + settings.SITE_LAX_PROTOCOL_CHECK + and referrer.hostname == site_url.hostname + ) + ) + + if not site_url_match: + # The accessed URL does not match the SITE_URL setting + if ( + isinstance(settings.CSRF_TRUSTED_ORIGINS, list) + and len(settings.CSRF_TRUSTED_ORIGINS) > 1 + ): + # The used url might not be the primary url - next check determines if in a trusted origins + pass + else: + source = CONFIG_LOOKUPS.get('INVENTREE_SITE_URL', {}).get( + 'source', 'unknown' + ) + dpl_method = inventreeInstaller() + 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}`' + logger.error(msg) + return render( + request, 'config_error.html', {'error_message': msg}, status=500 + ) + + trusted_origins_match = ( + # Matching domain found in allowed origins + any( + is_same_domain(referrer.netloc, host) + for host in [ + urlsplit(origin).netloc.lstrip('*') + for origin in settings.CSRF_TRUSTED_ORIGINS + ] + ) + ) or ( + # Lax protocol match allowed + settings.SITE_LAX_PROTOCOL_CHECK + and any( + referrer.hostname == urlsplit(origin).hostname + for origin in settings.CSRF_TRUSTED_ORIGINS + ) + ) + + # Check trusted origins + if not trusted_origins_match: + msg = f'INVE-E7: The used path `{accessed_scheme}` is not in the TRUSTED_ORIGINS' + logger.error(msg) + return render( + request, 'config_error.html', {'error_message': msg}, status=500 + ) + + # All checks passed + return None + + +class InvenTreeVersionHeaderMiddleware(MiddlewareMixin): + """Middleware to add the InvenTree version header to all responses.""" + + def process_response(self, request, response): + """Add the InvenTree version header to the response.""" + response['X-InvenTree-Version'] = inventreeVersion() + response['X-InvenTree-API'] = inventreeApiVersion() + response['X-InvenTree-Python'] = inventreePythonVersion() + response['X-InvenTree-Installer'] = inventreeInstaller() + return response diff --git a/src/backend/InvenTree/InvenTree/urls.py b/src/backend/InvenTree/InvenTree/urls.py index 01da5ace8dc3..c39902575ef0 100644 --- a/src/backend/InvenTree/InvenTree/urls.py +++ b/src/backend/InvenTree/InvenTree/urls.py @@ -1,200 +1,200 @@ -"""Top-level URL lookup for InvenTree application. - -Passes URL lookup downstream to each app as required. -""" - -from django.conf import settings -from django.conf.urls.static import static -from django.contrib import admin -from django.urls import include, path, re_path -from django.views.decorators.csrf import csrf_exempt -from django.views.generic.base import RedirectView - -from allauth.headless.urls import Client, build_urlpatterns -from drf_spectacular.views import SpectacularAPIView, SpectacularRedocView -from flags.urls import flagged_path -from oauth2_provider import urls as oauth2_urls -from sesame.views import LoginView - -import build.api -import common.api -import company.api -import importer.api -import InvenTree.logging # noqa: F401 - ensure logging handlers are registered -import machine.api -import order.api -import part.api -import plugin.api -import report.api -import stock.api -import users.api -from plugin.urls import get_plugin_urls -from web.urls import cui_compatibility_urls -from web.urls import urlpatterns as platform_urls - -from .api import ( - APISearchView, - InfoView, - LicenseView, - NotFoundView, - VersionTextView, - VersionView, -) -from .config import get_setting -from .magic_login import GetSimpleLoginView -from .views import auth_request - -# Set admin header from config or use default -admin.site.site_header = get_setting( - 'INVENTREE_SITE_HEADER', 'customize.site_header', 'InvenTree Admin' -) - - -apipatterns = [ - # Global search - path('admin/', include(common.api.admin_api_urls)), - path('bom/', include(part.api.bom_api_urls)), - path('build/', include(build.api.build_api_urls)), - path('company/', include(company.api.company_api_urls)), - path('importer/', include(importer.api.importer_api_urls)), - path('label/', include(report.api.label_api_urls)), - path('machine/', include(machine.api.machine_api_urls)), - path('order/', include(order.api.order_api_urls)), - path('part/', include(part.api.part_api_urls)), - path('report/', include(report.api.report_api_urls)), - path('search/', APISearchView.as_view(), name='api-search'), - path('settings/', include(common.api.settings_api_urls)), - path('stock/', include(stock.api.stock_api_urls)), - path( - 'generate/', - include([ - path( - 'batch-code/', - stock.api.GenerateBatchCode.as_view(), - name='api-generate-batch-code', - ), - path( - 'serial-number/', - stock.api.GenerateSerialNumber.as_view(), - name='api-generate-serial-number', - ), - ]), - ), - path('user/', include(users.api.user_urls)), - # Plugin endpoints - path('', include(plugin.api.plugin_api_urls)), - # Common endpoints endpoint - path('', include(common.api.common_api_urls)), - # OpenAPI Schema - path( - 'schema/', - SpectacularAPIView.as_view(custom_settings={'SCHEMA_PATH_PREFIX': '/api/'}), - name='schema', - ), - # InvenTree information endpoints - path('license/', LicenseView.as_view(), name='api-license'), # license info - path( - 'version-text', VersionTextView.as_view(), name='api-version-text' - ), # version text - path('version/', VersionView.as_view(), name='api-version'), # version info - path('', InfoView.as_view(), name='api-inventree-info'), # server info - # Auth API endpoints - path( - 'auth/', - include([ - path( - 'login-redirect/', - users.api.LoginRedirect.as_view(), - name='api-login-redirect', - ), - path( - '', - include( - (build_urlpatterns(Client.BROWSER), 'headless'), namespace='browser' - ), - ), # Allauth headless logic (only the browser client is included as we only use sessions based auth there) - ]), - ), - # Magic login URLs - path( - 'email/generate/', - csrf_exempt(GetSimpleLoginView().as_view()), - name='sesame-generate', - ), - path('email/login/', LoginView.as_view(), name='sesame-login'), - # Unknown endpoint - re_path(r'^.*$', NotFoundView.as_view(), name='api-404'), -] - - -backendpatterns = [ - path( - 'auth/', include('rest_framework.urls', namespace='rest_framework') - ), # Used for (DRF) browsable API auth - path( - 'auth/', auth_request, name='auth-check' - ), # Used for proxies to check if user is authenticated - path('accounts/', include('allauth.urls')), - # OAuth2 - flagged_path('OIDC', 'o/', include(oauth2_urls)), - path( - 'accounts/login/', - RedirectView.as_view(url=f'/{settings.FRONTEND_URL_BASE}', permanent=False), - name='account_login', - ), # Add a redirect for login views - path('anymail/', include('anymail.urls')), # Emails -] - -urlpatterns = [] - -if settings.INVENTREE_ADMIN_ENABLED: - admin_url = settings.INVENTREE_ADMIN_URL - - urlpatterns += [ - path(f'{admin_url}/error_log/', include('error_report.urls')), - path(f'{admin_url}/doc/', include('django.contrib.admindocs.urls')), - path(f'{admin_url}/', admin.site.urls, name='inventree-admin'), - ] - -urlpatterns += backendpatterns -urlpatterns += [ # API URLs - path('api/', include(apipatterns)), - path('api-doc/', SpectacularRedocView.as_view(url_name='schema'), name='api-doc'), -] -urlpatterns += platform_urls - -# Append custom plugin URLs (if custom plugin support is enabled) -if settings.PLUGINS_ENABLED: - urlpatterns.append(get_plugin_urls()) - -# Server running in "DEBUG" mode? -if settings.DEBUG: - # Static file access - urlpatterns += static(settings.STATIC_URL, document_root=settings.STATIC_ROOT) - - # Media file access - urlpatterns += static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT) - -# Redirect for favicon.ico -urlpatterns.append( - path( - 'favicon.ico', - RedirectView.as_view(url=f'{settings.STATIC_URL}img/favicon/favicon.ico'), - ) -) - -# Compatibility layer for old (CUI) URLs -if settings.FRONTEND_SETTINGS.get('url_compatibility'): - urlpatterns += cui_compatibility_urls(settings.FRONTEND_URL_BASE) - -if settings.DJANGO_SILK_ENABLED: - urlpatterns += [path('silk/', include('silk.urls', namespace='silk'))] - -# Send any unknown URLs to the index page -urlpatterns += [ - re_path( - r'^.*$', - RedirectView.as_view(url=f'/{settings.FRONTEND_URL_BASE}', permanent=False), - name='index', - ) -] +"""Top-level URL lookup for InvenTree application. + +Passes URL lookup downstream to each app as required. +""" + +from django.conf import settings +from django.conf.urls.static import static +from django.contrib import admin +from django.urls import include, path, re_path +from django.views.decorators.csrf import csrf_exempt +from django.views.generic.base import RedirectView + +from allauth.headless.urls import Client, build_urlpatterns +from drf_spectacular.views import SpectacularAPIView, SpectacularRedocView +from flags.urls import flagged_path +from oauth2_provider import urls as oauth2_urls +from sesame.views import LoginView + +import build.api +import common.api +import company.api +import importer.api +import InvenTree.logging # noqa: F401 - ensure logging handlers are registered +import machine.api +import order.api +import part.api +import plugin.api +import report.api +import stock.api +import users.api +from plugin.urls import get_plugin_urls +from web.urls import cui_compatibility_urls +from web.urls import urlpatterns as platform_urls + +from .api import ( + APISearchView, + InfoView, + LicenseView, + NotFoundView, + VersionTextView, + VersionView, +) +from .config import get_setting +from .magic_login import GetSimpleLoginView +from .views import auth_request + +# Set admin header from config or use default +admin.site.site_header = get_setting( + 'INVENTREE_SITE_HEADER', 'customize.site_header', 'InvenTree Admin' +) + + +apipatterns = [ + # Global search + path('admin/', include(common.api.admin_api_urls)), + path('bom/', include(part.api.bom_api_urls)), + path('build/', include(build.api.build_api_urls)), + path('company/', include(company.api.company_api_urls)), + path('importer/', include(importer.api.importer_api_urls)), + path('label/', include(report.api.label_api_urls)), + path('machine/', include(machine.api.machine_api_urls)), + path('order/', include(order.api.order_api_urls)), + path('part/', include(part.api.part_api_urls)), + path('report/', include(report.api.report_api_urls)), + path('search/', APISearchView.as_view(), name='api-search'), + path('settings/', include(common.api.settings_api_urls)), + path('stock/', include(stock.api.stock_api_urls)), + path( + 'generate/', + include([ + path( + 'batch-code/', + stock.api.GenerateBatchCode.as_view(), + name='api-generate-batch-code', + ), + path( + 'serial-number/', + stock.api.GenerateSerialNumber.as_view(), + name='api-generate-serial-number', + ), + ]), + ), + path('user/', include(users.api.user_urls)), + # Plugin endpoints + path('', include(plugin.api.plugin_api_urls)), + # Common endpoints endpoint + path('', include(common.api.common_api_urls)), + # OpenAPI Schema + path( + 'schema/', + SpectacularAPIView.as_view(custom_settings={'SCHEMA_PATH_PREFIX': '/api/'}), + name='schema', + ), + # InvenTree information endpoints + path('license/', LicenseView.as_view(), name='api-license'), # license info + path( + 'version-text', VersionTextView.as_view(), name='api-version-text' + ), # version text + path('version/', VersionView.as_view(), name='api-version'), # version info + path('', InfoView.as_view(), name='api-inventree-info'), # server info + # Auth API endpoints + path( + 'auth/', + include([ + path( + 'login-redirect/', + users.api.LoginRedirect.as_view(), + name='api-login-redirect', + ), + path( + '', + include( + (build_urlpatterns(Client.BROWSER), 'headless'), namespace='browser' + ), + ), # Allauth headless logic (only the browser client is included as we only use sessions based auth there) + ]), + ), + # Magic login URLs + path( + 'email/generate/', + csrf_exempt(GetSimpleLoginView().as_view()), + name='sesame-generate', + ), + path('email/login/', LoginView.as_view(), name='sesame-login'), + # Unknown endpoint + re_path(r'^.*$', NotFoundView.as_view(), name='api-404'), +] + + +backendpatterns = [ + path( + 'auth/', include('rest_framework.urls', namespace='rest_framework') + ), # Used for (DRF) browsable API auth + path( + 'auth/', auth_request, name='auth-check' + ), # Used for proxies to check if user is authenticated + path('accounts/', include('allauth.urls')), + # OAuth2 + flagged_path('OIDC', 'o/', include(oauth2_urls)), + path( + 'accounts/login/', + RedirectView.as_view(url=f'/{settings.FRONTEND_URL_BASE}', permanent=False), + name='account_login', + ), # Add a redirect for login views + path('anymail/', include('anymail.urls')), # Emails +] + +urlpatterns = [] + +if settings.INVENTREE_ADMIN_ENABLED: + admin_url = settings.INVENTREE_ADMIN_URL + + urlpatterns += [ + path(f'{admin_url}/error_log/', include('error_report.urls')), + path(f'{admin_url}/doc/', include('django.contrib.admindocs.urls')), + path(f'{admin_url}/', admin.site.urls, name='inventree-admin'), + ] + +urlpatterns += backendpatterns +urlpatterns += [ # API URLs + path('api/', include(apipatterns)), + path('api-doc/', SpectacularRedocView.as_view(url_name='schema'), name='api-doc'), +] +urlpatterns += platform_urls + +# Append custom plugin URLs (if custom plugin support is enabled) +if settings.PLUGINS_ENABLED: + urlpatterns.append(get_plugin_urls()) + +# Server running in "DEBUG" mode? +if settings.DEBUG: + # Static file access + urlpatterns += static(settings.STATIC_URL, document_root=settings.STATIC_ROOT) + + # Media file access + urlpatterns += static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT) + +# Redirect for favicon.ico +urlpatterns.append( + path( + 'favicon.ico', + RedirectView.as_view(url=f'{settings.STATIC_URL}img/favicon/favicon.ico'), + ) +) + +# Compatibility layer for old (CUI) URLs +if settings.FRONTEND_SETTINGS.get('url_compatibility'): + urlpatterns += cui_compatibility_urls(settings.FRONTEND_URL_BASE) + +if settings.DJANGO_SILK_ENABLED: + urlpatterns += [path('silk/', include('silk.urls', namespace='silk'))] + +# Send any unknown URLs to the index page +urlpatterns += [ + re_path( + r'^.*$', + RedirectView.as_view(url=f'/{settings.FRONTEND_URL_BASE}', permanent=False), + name='index', + ) +]