diff --git a/src/backend/InvenTree/InvenTree/api.py b/src/backend/InvenTree/InvenTree/api.py index d68a4ed2a9..3549d27e59 100644 --- a/src/backend/InvenTree/InvenTree/api.py +++ b/src/backend/InvenTree/InvenTree/api.py @@ -19,6 +19,7 @@ import structlog from django_q.models import OrmQ from drf_spectacular.utils import OpenApiParameter, OpenApiResponse, extend_schema from rest_framework import permissions, serializers, viewsets +from rest_framework.exceptions import AuthenticationFailed from rest_framework.generics import GenericAPIView from rest_framework.request import clone_request from rest_framework.response import Response @@ -38,7 +39,6 @@ from InvenTree.proof import calculate_proof from InvenTree.tasks import batch_offload_tasks from plugin.base.event.events import batch_events from plugin.serializers import MetadataSerializer -from users.models import ApiToken from users.permissions import check_user_permission, prefetch_rule_sets from .helpers import plugins_info, str2bool @@ -365,15 +365,13 @@ class InfoView(APIView): def check_auth_header(self, request): """Check if user is authenticated via a token in the header.""" from InvenTree.middleware import get_token_from_request + from users.authentication import validate_api_token if token := get_token_from_request(request): - # Does the provided token match a valid user? + # Check if the token is valid and the user is a staff member try: - token = ApiToken.get_from_string(token) - # Check if the token is active and the user is a staff member - if token and token.active and token.user and token.user.is_staff: - return True - except ApiToken.DoesNotExist: + return validate_api_token(token).user.is_staff + except AuthenticationFailed: pass return False diff --git a/src/backend/InvenTree/InvenTree/middleware.py b/src/backend/InvenTree/InvenTree/middleware.py index 40da4e1c98..a13969f3f8 100644 --- a/src/backend/InvenTree/InvenTree/middleware.py +++ b/src/backend/InvenTree/InvenTree/middleware.py @@ -15,6 +15,7 @@ from django.utils.translation import gettext_lazy as _ import structlog from error_report.middleware import ExceptionProcessor +from rest_framework.exceptions import AuthenticationFailed import InvenTree.helpers from common.settings import get_global_setting @@ -25,7 +26,7 @@ from InvenTree.version import ( inventreePythonVersion, inventreeVersion, ) -from users.models import ApiToken +from users.authentication import validate_api_token logger = structlog.get_logger('inventree') @@ -148,19 +149,20 @@ class AuthRequiredMiddleware: """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? + # Does the provided token match a valid (active) user? try: - token = ApiToken.get_from_string(token) - - if token and token.active and token.user: - # Provide the user information to the request - request.user = token.user - return True - except ApiToken.DoesNotExist: # pragma: no cover + token = validate_api_token(token) + except AuthenticationFailed as exc: logger.warning( - 'Access denied for unknown token %s', + 'Access denied for token %s: %s', InvenTree.helpers.sanitize_token(str(token)), - ) # pragma: no cover + exc.detail, + ) + return False + + # Provide the user information to the request + request.user = token.user + return True return False diff --git a/src/backend/InvenTree/InvenTree/test_middleware.py b/src/backend/InvenTree/InvenTree/test_middleware.py index 3b155f7308..026590209f 100644 --- a/src/backend/InvenTree/InvenTree/test_middleware.py +++ b/src/backend/InvenTree/InvenTree/test_middleware.py @@ -10,6 +10,7 @@ from error_report.models import Error from InvenTree.exceptions import log_error from InvenTree.helpers_mfa import get_codes from InvenTree.unit_test import InvenTreeTestCase +from users.models import ApiToken class MiddlewareTests(InvenTreeTestCase): @@ -114,6 +115,73 @@ class MiddlewareTests(InvenTreeTestCase): # Request something that is not on the API - should still work self.check_path(reverse('auth-check'), auth_header=f'Token {token}') + def test_token_auth_inactive_user(self): + """Tokens for inactive users, or revoked tokens, must be rejected outside the API.""" + target = reverse('auth-check') + + token = ApiToken.objects.create(user=self.user, name='middleware-test') + raw_token = token.token + self.client.logout() + + self.check_path(target, auth_header=f'Token {raw_token}') + + # Inactive users are rejected + self.user.is_active = False + self.user.save() + self.check_path(target, 401, auth_header=f'Token {raw_token}') + self.user.is_active = True + self.user.save() + + # Revoked tokens are rejected + token.revoked = True + token.save() + self.check_path(target, 401, auth_header=f'Token {raw_token}') + + def test_token_auth_rejection_logged(self): + """Rejected tokens are logged (sanitized) along with the reason.""" + target = reverse('auth-check') + + token = ApiToken.objects.create(user=self.user, name='log-test') + raw_token = token.token + token.revoked = True + token.save() + self.client.logout() + + with self.assertLogs('inventree', level='WARNING') as captured: + self.check_path(target, 401, auth_header=f'Token {raw_token}') + + output = '\n'.join(captured.output) + self.assertIn('Access denied for token', output) + self.assertIn('Token has been revoked', output) + # The secret must not appear in the logs + self.assertNotIn(raw_token, output) + + with self.assertLogs('inventree', level='WARNING') as captured: + self.check_path( + target, 401, auth_header='Token inv-unknown-token-value-0123456789' + ) + + self.assertIn('Invalid token', '\n'.join(captured.output)) + + def test_info_view_inactive_staff(self): + """InfoView must not treat a token for an inactive staff user as staff.""" + self.user.is_staff = True + self.user.save() + + raw_token = ApiToken.objects.create(user=self.user, name='info-test').token + self.client.logout() + + # Use the 'Bearer' scheme, so that DRF token auth does not handle the request, + # and InfoView falls back to its own token check + url = reverse('api-inventree-info') + response = self.check_path(url, auth_header=f'Bearer {raw_token}') + self.assertIsNotNone(response.json()['database']) + + self.user.is_active = False + self.user.save() + response = self.check_path(url, auth_header=f'Bearer {raw_token}') + self.assertIsNone(response.json()['database']) + def test_error_exceptions(self): """Test that ignored errors are not logged.""" self.assignRole('part.view') diff --git a/src/backend/InvenTree/users/authentication.py b/src/backend/InvenTree/users/authentication.py index 0903553989..0d1910ba5b 100644 --- a/src/backend/InvenTree/users/authentication.py +++ b/src/backend/InvenTree/users/authentication.py @@ -11,6 +11,39 @@ from rest_framework.authentication import TokenAuthentication import users.models +def validate_api_token(raw_token: str) -> users.models.ApiToken: + """Look up and fully validate a raw API token value. + + This is the single source of truth for API token validation, + used by both the DRF authentication class and the middleware stack. + + Raises: + AuthenticationFailed: If the token is unknown, revoked, expired, or belongs to an inactive user + """ + token = users.models.ApiToken.get_from_string(raw_token) if raw_token else None + + if token is None or token.user is None: + raise exceptions.AuthenticationFailed(_('Invalid token.')) + + if not token.user.is_active: + raise exceptions.AuthenticationFailed(_('User inactive or deleted.')) + + if token.revoked: + raise exceptions.AuthenticationFailed(_('Token has been revoked')) + + if token.expired: + raise exceptions.AuthenticationFailed(_('Token has expired')) + + if token.last_seen != datetime.date.today(): + # Update the last-seen date + # Note: Use update_fields to avoid clobbering concurrent changes to + # other fields on this token (e.g. a concurrent revocation) + token.last_seen = datetime.date.today() + token.save(update_fields=['last_seen']) + + return token + + class ApiTokenAuthentication(TokenAuthentication): """Custom implementation of TokenAuthentication class, with custom features. @@ -23,29 +56,8 @@ class ApiTokenAuthentication(TokenAuthentication): def authenticate_credentials(self, key): """Adds additional checks to the default token authentication method.""" - # If this runs without error, then the token is valid (so far) - token = self.model.get_from_string(key) - if token is None: - raise exceptions.AuthenticationFailed(_('Invalid token.')) - - user = token.user - if not user.is_active: - raise exceptions.AuthenticationFailed(_('User inactive or deleted.')) - - if token.revoked: - raise exceptions.AuthenticationFailed(_('Token has been revoked')) - - if token.expired: - raise exceptions.AuthenticationFailed(_('Token has expired')) - - if token.last_seen != datetime.date.today(): - # Update the last-seen date - # Note: Use update_fields to avoid clobbering concurrent changes to - # other fields on this token (e.g. a concurrent revocation) - token.last_seen = datetime.date.today() - token.save(update_fields=['last_seen']) - - return (user, token) + token = validate_api_token(key) + return (token.user, token) class ExtendedOAuth2Authentication(OAuth2ProtectedResourceAuthentication):