Merge commit from fork

* fix oauth scopes

* fix typing

src/backend/InvenTree/InvenTree/api_version.py

@@ -1,11 +1,15 @@
 """InvenTree API version information."""
 
 # InvenTree API version
-INVENTREE_API_VERSION = 568
+INVENTREE_API_VERSION = 569
 """Increment this API version number whenever there is a significant change to the API that any clients need to know about."""
 
 INVENTREE_API_TEXT = """
 
+v569 -> 2026-10-30 : https://github.com/inventree/InvenTree/security/advisories
+    - Hardens OAuth2 token usage in the API
+    - OAuth2 tokens can not be used to issue or manage API tokens (/api/user/me/token/, /api/user/tokens/)
+
 v568 -> 2026-10-30 : https://github.com/inventree/InvenTree/pull/13141
     - token endpoints now have a warning if notifications cannot be sent
 

src/backend/InvenTree/InvenTree/config.py

@@ -93,6 +93,21 @@ def ensure_middleware(
     return middleware
 
 
+def ensure_removed(values: list[str], forbidden: str, setting: str) -> list[str]:
+    """Ensure that a forbidden class is not present in a list setting."""
+    values = list(values)
+
+    if forbidden not in values:
+        return values
+
+    logger.warning(
+        'INVE-W21: %s is not supported in %s configuration - removing automatically',
+        forbidden,
+        setting,
+    )
+    return [value for value in values if value != forbidden]
+
+
 def is_true(x):
     """Shortcut function to determine if a value "looks" like a boolean."""
     return str(x).strip().lower() in ['1', 'y', 'yes', 't', 'true', 'on']

src/backend/InvenTree/InvenTree/permissions.py

@@ -3,7 +3,6 @@
 from functools import wraps
 from typing import Optional
 
-from oauth2_provider.contrib.rest_framework import TokenMatchesOASRequirements
 from oauth2_provider.contrib.rest_framework.authentication import OAuth2Authentication
 from rest_framework import permissions
 
@@ -13,6 +12,7 @@ from users.oauth2_scopes import (
     DEFAULT_READ,
     DEFAULT_STAFF,
     DEFAULT_SUPERUSER,
+    DEFAULT_WRITE,
     _roles,
     get_granular_scope,
 )
@@ -41,6 +41,22 @@ def get_model_for_view(view):
     raise AttributeError(f'Serializer class not specified for {view.__class__}')
 
 
+def get_roles_for_view(view) -> list[str]:
+    """Return the user roles associated with the model of an API view."""
+    try:
+        model = get_model_for_view(view)
+    except Exception:
+        # Fall back to the model of the queryset (as checked by ModelPermission)
+        model = getattr(getattr(view, 'queryset', None), 'model', None)
+
+    if model is None:
+        return []
+
+    return precalculated_roles.get(
+        f'{model._meta.app_label}_{model._meta.model_name}', []
+    )
+
+
 def map_scope(
     roles: Optional[list[str]] = None,
     only_read=False,
@@ -92,24 +108,55 @@ for role, tables in roles.items():
         precalculated_roles[table].append(role)
 
 
+def enforce_oauth2_scopes(has_permission):
+    """Decorate a `has_permission` method, so that OAuth2 tokens must hold the required scopes.
+
+    Requests which are not authenticated via OAuth2 are checked by the decorated method only.
+    """
+
+    @wraps(has_permission)
+    def wrapped_has_permission(self, request, view):
+        if self.is_oauth2ed(request) and not self.has_scope_permission(request, view):
+            return False
+
+        return has_permission(self, request, view)
+
+    wrapped_has_permission.enforces_oauth2_scopes = True
+    return wrapped_has_permission
+
+
 class OASTokenMixin:
     """Mixin that combines the permissions of normal classes and token classes."""
 
-    ENFORCE_USER_PERMS: bool = False
+    def __init_subclass__(cls, **kwargs):
+        """Ensure that the scope check is applied to the `has_permission` method of the subclass."""
+        super().__init_subclass__(**kwargs)
 
+        if not getattr(cls.has_permission, 'enforces_oauth2_scopes', False):
+            cls.has_permission = enforce_oauth2_scopes(cls.has_permission)
+
+    @enforce_oauth2_scopes
     def has_permission(self, request, view):
-        """Check if the user has the required scopes or was authenticated another way."""
-        if self.ENFORCE_USER_PERMS:
-            return super().has_permission(request, view)
-        return self.check_oauth2_authentication(
-            request, view
-        ) or super().has_permission(request, view)
+        """Check the normal permissions (the scopes are checked by the decorator)."""
+        return super().has_permission(request, view)
 
-    def check_oauth2_authentication(self, request, view):
-        """Check if the user is authenticated using OAuth2 and has the required scopes."""
-        return self.is_oauth2ed(
-            request
-        ) and TokenMatchesOASRequirements().has_permission(request, view)
+    def has_scope_permission(self, request, view):
+        """Check if the OAuth2 token holds one of the scope alternatives required for this request."""
+        token = request.auth
+        alternatives = self.get_required_alternate_scopes(request, view).get(
+            request.method.upper(), []
+        )
+
+        return hasattr(token, 'is_valid') and any(
+            scopes and token.is_valid(scopes) for scopes in alternatives
+        )
+
+    def get_required_alternate_scopes(self, request, view):
+        """Return the required scope alternatives for each request method.
+
+        By default, OAuth2 tokens can not access the view.
+        """
+        return {}
 
     def is_oauth2ed(self, request):
         """Check if the user is authenticated using OAuth2."""
@@ -122,49 +169,43 @@ class OASTokenMixin:
 
 
 class InvenTreeRoleScopeMixin(OASTokenMixin):
-    """Permission that discovers the required scopes from the OpenAPI schema."""
+    """Permission that discovers the required scopes from the user roles of the view."""
 
     def get_required_alternate_scopes(self, request, view):
         """Return the required scopes for the current request."""
         if hasattr(view, 'required_alternate_scopes'):
             return view.required_alternate_scopes
-        try:
-            # Extract the model name associated with this request
-            model = get_model_for_view(view)
-            calc = precalculated_roles.get(
-                f'{model._meta.app_label}_{model._meta.model_name}', []
-            )
 
-            if model is None or not calc:
-                return map_scope(only_read=True)
-            return map_scope(roles=calc)
-        except AttributeError:
-            # We will assume that if the serializer class does *not* have a Meta,
-            # then we don't need a permission
-            return map_scope(only_read=True)
-        except Exception:
-            return map_scope(only_read=True)
+        # The required role may be defined for the view class (see RolePermission)
+        if role := getattr(view, 'role_required', None):
+            if '.' in role:
+                role, permission = role.split('.')
+                return map_scope(roles=[role], override_all_actions=permission)
+            return map_scope(roles=[role])
+
+        if roles := get_roles_for_view(view):
+            return map_scope(roles=roles)
+
+        # No role is associated with this view - writing requires the general write scope
+        return map_scope(
+            only_read=True,
+            read_name=DEFAULT_WRITE,
+            map_read=list(permissions.SAFE_METHODS),
+        )
 
 
-class InvenTreeTokenMatchesOASRequirements(InvenTreeRoleScopeMixin):
+class InvenTreeTokenMatchesOASRequirements(
+    InvenTreeRoleScopeMixin, permissions.BasePermission
+):
     """Combines InvenTree role-based scope handling with OpenAPI schema token requirements.
 
     Used as default permission class.
     """
 
     def has_permission(self, request, view):
-        """Check if the user has the required scopes or was authenticated another way."""
-        if self.is_oauth2ed(request):
-            # Check if the user is authenticated using OAuth2 and has the required scopes
-            return super().has_permission(request, view)
-
-        # If the user is authenticated using another method, check if they have the required permissions
+        """Check that the user is authenticated - the user permissions are checked by the other default classes."""
         return bool(request.user and request.user.is_authenticated)
 
-    def has_object_permission(self, request, view, obj):
-        """Return `True` if permission is granted, `False` otherwise."""
-        return True
-
 
 class ModelPermission(permissions.DjangoModelPermissions):
     """Custom ModelPermission implementation which provides cached lookup of queryset.
@@ -324,11 +365,27 @@ class IsSuperuserOrReadOnlyOrScope(OASTokenMixin, permissions.IsAdminUser):
 
 
 class IsAuthenticatedOrReadScope(OASTokenMixin, permissions.IsAuthenticated):
-    """Allows access only to authenticated users or read scope tokens."""
+    """Allows access only to authenticated users or read / write scope tokens."""
 
     def get_required_alternate_scopes(self, request, view):
         """Return the required scopes for the current request."""
-        return map_scope(only_read=True)
+        return map_scope(
+            only_read=True,
+            read_name=DEFAULT_WRITE,
+            map_read=list(permissions.SAFE_METHODS),
+        )
+
+
+class IsAuthenticatedNotOAuth2(OASTokenMixin, permissions.IsAuthenticated):
+    """Allows access only to authenticated users - OAuth2 tokens are never allowed.
+
+    Used for endpoints which issue or manage credentials (e.g. API tokens):
+    a scoped OAuth2 token must not be able to obtain an unscoped credential.
+    """
+
+    def get_required_alternate_scopes(self, request, view):
+        """No scope grants access to this endpoint."""
+        return {}
 
 
 class IsStaffOrReadOnlyScope(OASTokenMixin, permissions.IsAuthenticated):
@@ -359,7 +416,7 @@ class IsAdminOrAdminScope(OASTokenMixin, permissions.IsAdminUser):
 
 
 class AllowAnyOrReadScope(OASTokenMixin, permissions.AllowAny):
-    """Allows access to any user or read scope tokens."""
+    """Allows access to any user or read / write scope tokens."""
 
     def has_permission(self, request, view):
         """Anyone is allowed."""
@@ -367,7 +424,11 @@ class AllowAnyOrReadScope(OASTokenMixin, permissions.AllowAny):
 
     def get_required_alternate_scopes(self, request, view):
         """Return the required scopes for the current request."""
-        return map_scope(only_read=True)
+        return map_scope(
+            only_read=True,
+            read_name=DEFAULT_WRITE,
+            map_read=list(permissions.SAFE_METHODS),
+        )
 
 
 def auth_exempt(view_func):
@@ -405,7 +466,11 @@ class UserSettingsPermissionsOrScope(OASTokenMixin, permissions.BasePermission):
 
     def get_required_alternate_scopes(self, request, view):
         """Return the required scopes for the current request."""
-        return map_scope(only_read=True)
+        return map_scope(
+            only_read=True,
+            read_name=DEFAULT_WRITE,
+            map_read=list(permissions.SAFE_METHODS),
+        )
 
 
 class GlobalSettingsPermissions(OASTokenMixin, permissions.BasePermission):
@@ -435,8 +500,6 @@ class GlobalSettingsPermissions(OASTokenMixin, permissions.BasePermission):
 class DataImporterPermission(OASTokenMixin, permissions.BasePermission):
     """Mixin class for determining if the user has correct permissions."""
 
-    ENFORCE_USER_PERMS = True
-
     def has_permission(self, request, view):
         """Class level permission checks are handled via InvenTree.permissions.IsAuthenticatedOrReadScope."""
         return request.user and request.user.is_authenticated
@@ -471,8 +534,6 @@ class DataImporterPermission(OASTokenMixin, permissions.BasePermission):
 class ContentTypePermission(OASTokenMixin, permissions.BasePermission):
     """Mixin class for determining if the user has correct permissions."""
 
-    ENFORCE_USER_PERMS = True
-
     def has_permission(self, request, view):
         """Class level permission checks are handled via InvenTree.permissions.IsAuthenticatedOrReadScope."""
         return request.user and request.user.is_authenticated

src/backend/InvenTree/InvenTree/schema.py

@@ -1,7 +1,7 @@
 """Schema processing functions for cleaning up generated schema."""
 
 from itertools import chain
-from typing import Any, Optional
+from typing import Any, Optional, TypedDict
 
 from django.conf import settings
 
@@ -20,7 +20,14 @@ from rest_framework import serializers, viewsets
 from rest_framework.pagination import LimitOffsetPagination
 
 from InvenTree.permissions import OASTokenMixin
-from users.oauth2_scopes import oauth2_scopes
+from users.oauth2_scopes import DEFAULT_READ, oauth2_scopes
+
+
+class _SchemaPathDetails(TypedDict):
+    method: str
+    oauth: list[str]
+    alternatives: list[list[str]]
+    sec: bool
 
 
 class ExtendedOAuth2Scheme(DjangoOAuthToolkitScheme):
@@ -35,13 +42,37 @@ class ExtendedOAuth2Scheme(DjangoOAuthToolkitScheme):
             return ret
 
         # If no security requirement is found, try if the view uses our OASTokenMixin
+        # A token has to satisfy *all* of these permissions, so their alternatives are combined
+        alternatives = None
         for permission in auto_schema.view.get_permissions():
             if isinstance(permission, OASTokenMixin):
                 alt_scopes = permission.get_required_alternate_scopes(
                     auto_schema.view.request, auto_schema.view
                 )
                 alt_scopes = alt_scopes.get(auto_schema.method, [])
-                return [{self.name: group} for group in alt_scopes]
+                alternatives = (
+                    alt_scopes
+                    if alternatives is None
+                    else combine_scope_alternatives(alternatives, alt_scopes)
+                )
+
+        if alternatives is not None:
+            return [{self.name: group} for group in alternatives]
+
+
+def combine_scope_alternatives(
+    first: list[list[str]], second: list[list[str]]
+) -> list[list[str]]:
+    """Combine two lists of scope alternatives, which both have to be satisfied."""
+    combined = []
+
+    for a in first:
+        for b in second:
+            group = [*a, *(scope for scope in b if scope not in a)]
+            if group not in combined:
+                combined.append(group)
+
+    return combined
 
 
 class ExtendedAutoSchema(AutoSchema):
@@ -281,7 +312,7 @@ def postprocess_required_nullable(result, generator, request, public):
 
 def postprocess_print_stats(result, generator, request, public):
     """Prints statistics against schema."""
-    rlt_dict = {}
+    rlt_dict: dict[str, _SchemaPathDetails] = {}
     for path in result['paths']:
         for method in result['paths'][path]:
             sec = result['paths'][path][method].get('security', [])
@@ -289,6 +320,7 @@ def postprocess_print_stats(result, generator, request, public):
             rlt_dict[f'{path}:{method}'] = {
                 'method': method,
                 'oauth': list(chain(*scopes)),
+                'alternatives': scopes,
                 'sec': sec is None,
             }
 
@@ -335,6 +367,20 @@ def postprocess_print_stats(result, generator, request, public):
             '\n\nPlease check the schema and add oauth2 scopes where necessary.'
         )
 
+    # Raise error if a write operation can be performed with the general read scope
+    read_scopes = {DEFAULT_READ, 'openid'}
+    read_writable = [
+        path
+        for path, details in rlt_dict.items()
+        if details['method'] not in ('get', 'head', 'options')
+        and any(set(group) <= read_scopes for group in details['alternatives'])
+    ]
+    if len(read_writable) > 0:
+        warn(
+            f'Found {len(read_writable)} write operations which are accessible with the read scope:\n{", ".join(read_writable)}. '
+            '\n\nPlease check the scopes of these paths.'
+        )
+
     return result
 
 

src/backend/InvenTree/InvenTree/setting/spectacular.py

@@ -18,6 +18,7 @@ def get_spectacular_settings():
         },
         'VERSION': str(inventreeApiVersion()),
         'SERVE_INCLUDE_SCHEMA': False,
+        'SERVE_PERMISSIONS': ['InvenTree.permissions.AllowAnyOrReadScope'],
         'SCHEMA_PATH_PREFIX': '/api/',
         'POSTPROCESSING_HOOKS': [
             'InvenTree.schema.postprocess_schema_enums',

src/backend/InvenTree/InvenTree/settings.py

@@ -27,6 +27,7 @@ import InvenTree.backup
 from InvenTree.cache import get_cache_config, is_global_cache_enabled
 from InvenTree.config import (
     ensure_middleware,
+    ensure_removed,
     get_boolean_setting,
     get_oidc_private_key,
     get_setting,
@@ -399,7 +400,6 @@ MIDDLEWARE = CONFIG.get(
         'django.middleware.clickjacking.XFrameOptionsMiddleware',
         'InvenTree.middleware.AuthRequiredMiddleware',
         'InvenTree.middleware.Check2FAMiddleware',  # Check if the user should be forced to use MFA
-        'oauth2_provider.middleware.OAuth2TokenMiddleware',  # oauth2_provider
         'maintenance_mode.middleware.MaintenanceModeMiddleware',
         'InvenTree.middleware.InvenTreeExceptionProcessor',  # Error reporting
         'InvenTree.middleware.InvenTreeRequestCacheMiddleware',  # Request caching
@@ -418,6 +418,12 @@ MIDDLEWARE = ensure_middleware(
     after='django_structlog.middlewares.RequestMiddleware',
 )
 
+# OAuth2 tokens must only be handled by the API authentication class (which enforces token scopes)
+# The oauth2_provider middleware would turn a token into a full (session) user login - ensure that it is never present
+MIDDLEWARE = ensure_removed(
+    MIDDLEWARE, 'oauth2_provider.middleware.OAuth2TokenMiddleware', 'MIDDLEWARE'
+)
+
 # Audit log configuration (django-auditlog)
 # Ref: https://django-auditlog.readthedocs.io/en/latest/usage.html#settings
 # Note: AUDITLOG_LOGENTRY_MODEL is a swappable model, and must not be changed once migrations have been applied
@@ -476,7 +482,6 @@ QUERYCOUNT = {
 
 
 default_auth_backends = [
-    'oauth2_provider.backends.OAuth2Backend',  # OAuth2 provider
     'django.contrib.auth.backends.RemoteUserBackend',  # proxy login
     'django.contrib.auth.backends.ModelBackend',
     'allauth.account.auth_backends.AuthenticationBackend',  # SSO login via external providers
@@ -489,6 +494,13 @@ AUTHENTICATION_BACKENDS = (
     else default_auth_backends
 )
 
+# The oauth2_provider backend accepts an OAuth2 token for *any* authentication attempt (e.g. a login form) - ensure that it is never present
+AUTHENTICATION_BACKENDS = ensure_removed(
+    AUTHENTICATION_BACKENDS,
+    'oauth2_provider.backends.OAuth2Backend',
+    'AUTHENTICATION_BACKENDS',
+)
+
 # LDAP support
 LDAP_AUTH = get_boolean_setting('INVENTREE_LDAP_ENABLED', 'ldap.enabled', False)
 LDAP_DEBUG = (
@@ -1232,6 +1244,13 @@ OAUTH2_PROVIDER = {
 OAUTH2_CHECK_EXCLUDED = [  # This setting mutes schema checks for these rule/method combinations
     '/api/email/generate/:post',
     '/api/webhook/{endpoint}/:post',
+    # API token management is not available to OAuth2 tokens
+    '/api/user/token/:get',
+    '/api/user/me/token/:get',
+    '/api/user/tokens/:get',
+    '/api/user/tokens/:post',
+    '/api/user/tokens/{id}/:get',
+    '/api/user/tokens/{id}/:delete',
 ]
 
 if SITE_URL and not TESTING:  # pragma: no cover

src/backend/InvenTree/InvenTree/test_api.py

@@ -262,6 +262,16 @@ class OAuth2ApplicationAPITests(InvenTreeAPITestCase):
         self.assertIn('theme', profile)
         self.assertIn('widgets', profile)
 
+        # The token is limited to its (read) scope - even with a matching CSRF cookie / header
+        self.client.cookies['csrftoken'] = 'A' * 32
+        response = self.patch(
+            reverse('api-user-detail', kwargs={'pk': self.user.pk}),
+            {'first_name': 'Changed'},
+            HTTP_AUTHORIZATION=f'Bearer {access_token}',
+            HTTP_X_CSRFTOKEN='A' * 32,
+            expected_code=403,
+        )
+
 
 class ApiAccessTests(InvenTreeAPITestCase):
     """Tests for various access scenarios with the InvenTree API."""

src/backend/InvenTree/InvenTree/test_oauth2.py

@@ -0,0 +1,280 @@
+"""Tests for the enforcement of OAuth2 token scopes."""
+
+import json
+import re
+from datetime import timedelta
+
+from django.conf import settings
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import Group
+from django.test import override_settings
+from django.urls import Resolver404, resolve, reverse
+from django.utils import timezone
+
+from allauth.account.models import EmailAddress
+from drf_spectacular.generators import EndpointEnumerator
+from oauth2_provider.contrib.rest_framework.authentication import OAuth2Authentication
+from oauth2_provider.models import AccessToken, Application
+from rest_framework import permissions
+from rest_framework.test import APIClient
+
+from InvenTree.permissions import IsAuthenticatedOrReadScope, OASTokenMixin
+from InvenTree.schema import combine_scope_alternatives
+from InvenTree.unit_test import InvenTreeAPITestCase
+from users.authentication import ExtendedOAuth2Authentication
+from users.models import ApiToken
+from users.oauth2_scopes import DEFAULT_READ, DEFAULT_STAFF, oauth2_scopes
+
+# Any client which is not a browser can set a matching CSRF cookie and header
+CSRF_TOKEN = 'A' * 32
+
+
+@override_settings(
+    SITE_URL='http://testserver', CSRF_TRUSTED_ORIGINS=['http://testserver']
+)
+class OAuth2ScopeTests(InvenTreeAPITestCase):
+    """Tests that OAuth2 tokens can only be used within their scopes."""
+
+    roles = ['admin.view', 'admin.change', 'part_category.view', 'part_category.add']
+    auto_login = False
+
+    @classmethod
+    def setUpTestData(cls):
+        """Create an OAuth2 application to issue tokens for."""
+        super().setUpTestData()
+
+        cls.application = Application.objects.create(
+            name='Scope test application',
+            client_type=Application.CLIENT_CONFIDENTIAL,
+            authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
+            redirect_uris='https://example.com/callback',
+        )
+
+    def get_token(self, scope: str, user=None) -> AccessToken:
+        """Issue an OAuth2 access token with the provided scopes."""
+        return AccessToken.objects.create(
+            user=user or self.user,
+            application=self.application,
+            token=f'oauth2-test-token-{AccessToken.objects.count()}',
+            expires=timezone.now() + timedelta(hours=1),
+            scope=scope,
+        )
+
+    def oauth2_request(self, token: AccessToken, method: str, url: str, data=None):
+        """Perform an API request with an OAuth2 token, the way a non-browser client can.
+
+        CSRF checks are enforced, but the client provides a matching CSRF cookie and header.
+        """
+        client = APIClient(enforce_csrf_checks=True, raise_request_exception=False)
+        client.cookies['csrftoken'] = CSRF_TOKEN
+
+        return client.generic(
+            method,
+            url,
+            data=json.dumps(data or {}),
+            content_type='application/json',
+            HTTP_AUTHORIZATION=f'Bearer {token.token}',
+            HTTP_X_CSRFTOKEN=CSRF_TOKEN,
+        )
+
+    def test_reported_user_escalation(self):
+        """A read-only token of a staff user must not be able to modify users or their groups."""
+        victim = get_user_model().objects.create_user('victim', password='password')
+        admin_group = Group.objects.create(name='Administrators')
+        url = reverse('api-user-detail', kwargs={'pk': victim.pk})
+        data = {'is_staff': True, 'group_ids': [admin_group.pk]}
+
+        token = self.get_token(f'openid {DEFAULT_READ}')
+
+        response = self.oauth2_request(token, 'GET', url)
+        self.assertEqual(response.status_code, 200)
+
+        for method in ['PATCH', 'PUT', 'DELETE']:
+            response = self.oauth2_request(token, method, url, data)
+            self.assertEqual(response.status_code, 403, method)
+
+        victim.refresh_from_db()
+        self.assertFalse(victim.is_staff)
+        self.assertFalse(victim.groups.exists())
+
+        # With the staff scope, the token can perform the action (as the user has the permissions)
+        token = self.get_token(f'openid {DEFAULT_READ} {DEFAULT_STAFF}')
+        response = self.oauth2_request(token, 'PATCH', url, data)
+        self.assertEqual(response.status_code, 200)
+
+        victim.refresh_from_db()
+        self.assertTrue(victim.is_staff)
+
+    def test_write_requires_role_scope(self):
+        """Creating an object requires the 'add' scopes for the associated roles."""
+        url = reverse('api-part-category-list')
+
+        token = self.get_token(f'openid {DEFAULT_READ}')
+        response = self.oauth2_request(token, 'POST', url, {'name': 'Category'})
+        self.assertEqual(response.status_code, 403)
+
+        token = self.get_token('openid r:add:build r:add:part_category')
+        response = self.oauth2_request(token, 'POST', url, {'name': 'Category'})
+        self.assertEqual(response.status_code, 201)
+
+    def test_scopes_do_not_extend_user_permissions(self):
+        """A token can never do more than the user is allowed to do - regardless of the scopes."""
+        token = self.get_token(' '.join(oauth2_scopes))
+
+        # Superuser scope, but the user is not a superuser
+        url = reverse('api-user-set-password', kwargs={'pk': self.user.pk})
+        response = self.oauth2_request(token, 'PUT', url, {'password': 'new-pw'})
+        self.assertEqual(response.status_code, 403)
+
+        # Part scope, but the user does not have the part role
+        response = self.oauth2_request(
+            token, 'POST', reverse('api-part-list'), {'name': 'Part'}
+        )
+        self.assertEqual(response.status_code, 403)
+
+    def test_token_cannot_issue_credentials(self):
+        """An OAuth2 token must not be exchanged for an API token or a session."""
+        token = self.get_token(' '.join(oauth2_scopes))
+        n_tokens = ApiToken.objects.count()
+
+        response = self.oauth2_request(token, 'GET', reverse('api-token'))
+        self.assertEqual(response.status_code, 403)
+        self.assertNotIn('sessionid', response.cookies)
+
+        response = self.oauth2_request(
+            token, 'POST', reverse('api-token-list'), {'name': 'token'}
+        )
+        self.assertEqual(response.status_code, 403)
+
+        self.assertEqual(ApiToken.objects.count(), n_tokens)
+
+    def test_token_only_accepted_by_api(self):
+        """OAuth2 tokens are only handled by the API, not by the session based views."""
+        self.assertNotIn(
+            'oauth2_provider.middleware.OAuth2TokenMiddleware', settings.MIDDLEWARE
+        )
+        self.assertNotIn(
+            'oauth2_provider.backends.OAuth2Backend', settings.AUTHENTICATION_BACKENDS
+        )
+
+        token = self.get_token(' '.join(oauth2_scopes))
+
+        response = self.oauth2_request(token, 'GET', '/api/auth/v1/auth/session')
+        self.assertEqual(response.status_code, 401)
+
+        response = self.oauth2_request(
+            token, 'POST', '/api/auth/v1/account/email', {'email': 'new@example.com'}
+        )
+        self.assertNotEqual(response.status_code, 200)
+        self.assertFalse(EmailAddress.objects.filter(email='new@example.com').exists())
+
+    def test_scope_check_cannot_be_bypassed(self):
+        """Permission classes which override `has_permission` still enforce the token scopes."""
+
+        class AllowEverything(OASTokenMixin, permissions.BasePermission):
+            def has_permission(self, request, view):
+                return True
+
+            def get_required_alternate_scopes(self, request, view):
+                return {'GET': [[DEFAULT_READ]]}
+
+        class AllowAnyFirst(permissions.AllowAny, IsAuthenticatedOrReadScope):
+            pass
+
+        class OAuth2Request:
+            def __init__(self, token):
+                self.user = token.user
+                self.auth = token
+                self.method = 'GET'
+                self.successful_authenticator = OAuth2Authentication()
+
+        no_scope = OAuth2Request(self.get_token('openid'))
+        read_scope = OAuth2Request(self.get_token(f'openid {DEFAULT_READ}'))
+
+        for permission in [AllowEverything(), AllowAnyFirst()]:
+            self.assertFalse(permission.has_permission(no_scope, None))
+            self.assertTrue(permission.has_permission(read_scope, None))
+
+    def test_combine_scope_alternatives(self):
+        """The schema combines the scopes of all permission classes of a view."""
+        self.assertEqual(
+            combine_scope_alternatives([['a', 'b']], [['b', 'c']]), [['a', 'b', 'c']]
+        )
+        self.assertEqual(
+            combine_scope_alternatives([['a'], ['b']], [['c']]),
+            [['a', 'c'], ['b', 'c']],
+        )
+        self.assertEqual(combine_scope_alternatives([['a']], []), [])
+
+    def test_api_endpoints(self):
+        """Check the scope enforcement of all API endpoints, using tokens of a superuser.
+
+        - A token without any scope can not access any endpoint
+        - A token with the read scope can not perform any write operation
+        - A token with all scopes can read from all endpoints (unless excluded from OAuth2)
+        """
+        admin = get_user_model().objects.create_superuser(
+            'oauth2_admin', 'admin@example.com', 'password'
+        )
+        no_scope = self.get_token('openid', admin)
+        read_scope = self.get_token(f'openid {DEFAULT_READ}', admin)
+        all_scopes = self.get_token(' '.join(oauth2_scopes), admin)
+
+        # Path parameters which are validated before the permission checks
+        parameters = {
+            'model': 'part',
+            'lookup_field': 'pk',
+            'statusmodel': 'StockStatus',
+        }
+
+        errors = []
+        checked = 0
+
+        for path, _regex, method, callback in EndpointEnumerator().get_api_endpoints():
+            if not path.startswith('/api/'):
+                continue
+
+            # Views which do not use OAuth2 authentication ignore the token
+            authentication_classes = getattr(callback, 'initkwargs', {}).get(
+                'authentication_classes', callback.cls.authentication_classes
+            )
+            if ExtendedOAuth2Authentication not in authentication_classes:
+                continue
+
+            # Fill in the path parameters - the permission checks run before any lookup
+            url = re.sub(
+                r'{([^}]+)}', lambda match: parameters.get(match[1], '1'), path
+            )
+
+            try:
+                if resolve(url).func != callback:
+                    continue
+            except Resolver404:
+                continue
+
+            checked += 1
+            excluded = (
+                f'{path.replace("{pk}", "{id}")}:{method.lower()}'
+                in settings.OAUTH2_CHECK_EXCLUDED
+            )
+
+            response = self.oauth2_request(no_scope, method, url)
+            if response.status_code != 403:
+                errors.append(f'{method} {path}: no scope -> {response.status_code}')
+
+            if method not in permissions.SAFE_METHODS:
+                response = self.oauth2_request(read_scope, method, url)
+                if response.status_code != 403:
+                    errors.append(
+                        f'{method} {path}: read scope -> {response.status_code}'
+                    )
+
+            elif method == 'GET' and not excluded:
+                response = self.oauth2_request(all_scopes, method, url)
+                if response.status_code in [401, 403]:
+                    errors.append(
+                        f'{method} {path}: all scopes -> {response.status_code}'
+                    )
+
+        self.assertGreater(checked, 500)
+        self.assertEqual(errors, [], '\n'.join(errors))

src/backend/InvenTree/users/api.py

@@ -378,7 +378,7 @@ class RuleSetDetail(RuleSetMixin, RetrieveUpdateDestroyAPI):
 class GetAuthToken(GenericAPIView):
     """Return authentication token for an authenticated user."""
 
-    permission_classes = [InvenTree.permissions.IsAuthenticatedOrReadScope]
+    permission_classes = [InvenTree.permissions.IsAuthenticatedNotOAuth2]
     serializer_class = GetAuthTokenSerializer
 
     @extend_schema(
@@ -462,7 +462,7 @@ class GetAuthToken(GenericAPIView):
 class TokenMixin:
     """Mixin for API token endpoints."""
 
-    permission_classes = (InvenTree.permissions.IsAuthenticatedOrReadScope,)
+    permission_classes = (InvenTree.permissions.IsAuthenticatedNotOAuth2,)
     serializer_class = ApiTokenSerializer
 
     def get_queryset(self):

src/backend/InvenTree/users/authentication.py

@@ -63,3 +63,25 @@ class ApiTokenAuthentication(TokenAuthentication):
 
 class ExtendedOAuth2Authentication(OAuth2ProtectedResourceAuthentication):
     """Custom implementation of OAuth2Authentication class to support custom scope rendering."""
+
+    def authenticate(self, request):
+        """Authenticate the request, if the requested view enforces OAuth2 token scopes."""
+        result = super().authenticate(request)
+
+        if result is not None and not self.view_enforces_scopes(request):
+            raise exceptions.PermissionDenied(
+                _('OAuth2 tokens are not accepted for this endpoint')
+            )
+
+        return result
+
+    def view_enforces_scopes(self, request) -> bool:
+        """Check if any permission class of the requested view enforces OAuth2 token scopes."""
+        from InvenTree.permissions import OASTokenMixin
+
+        view = (getattr(request, 'parser_context', None) or {}).get('view')
+
+        return view is not None and any(
+            isinstance(permission, OASTokenMixin)
+            for permission in view.get_permissions()
+        )

src/backend/InvenTree/users/oauth2_scopes.py

@@ -34,11 +34,13 @@ calculated = {
 
 
 DEFAULT_READ = get_granular_scope('read', type='g')
+DEFAULT_WRITE = get_granular_scope('write', type='g')
 DEFAULT_STAFF = get_granular_scope('staff', type='a')
 DEFAULT_SUPERUSER = get_granular_scope('superuser', type='a')
 # This is actually used
 oauth2_scopes = {
     DEFAULT_READ: 'General Read scope',
+    DEFAULT_WRITE: 'General Write scope',
     'openid': 'OpenID Connect scope',
     # Admin scopes
     DEFAULT_STAFF: 'User Role Staff',