Check if user is active when checking token (#13148)

* Check if user is active when checking token

* Ensure token rejection logs a warning

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

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
 

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')

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):