diff --git a/docs/docs/api/index.md b/docs/docs/api/index.md index 55a8027c29..a5e96d4506 100644 --- a/docs/docs/api/index.md +++ b/docs/docs/api/index.md @@ -154,10 +154,17 @@ Examples: ```bash a:superuser g:read +g:write r:change:part r:delete:stock ``` +The general scopes `g:read` and `g:write` cover endpoints which are not associated with a specific role - `g:read` only grants read access (`GET`, `HEAD`, `OPTIONS`), write operations require `g:write`. + +!!! warning "Scopes restrict - they do not grant" + A request with an oAuth2 token must satisfy both the scopes of the token *and* the permissions of the user who authorized the application. A token can therefore never do more than the user is allowed to do. + oAuth2 tokens are only accepted by the API - they can not be used to log in to the web interface, or to issue (or manage) API tokens. + !!! info "Read the API docs" The API [documentation](#documentation) and [schema](./schema.md) list the required scopes for every API endpoint / interaction in the security sections. diff --git a/docs/docs/settings/error_codes.md b/docs/docs/settings/error_codes.md index d26bd1cbd6..7b02ffc46e 100644 --- a/docs/docs/settings/error_codes.md +++ b/docs/docs/settings/error_codes.md @@ -262,6 +262,13 @@ The `middleware` list provided in the configuration file does not include the au The middleware has been added automatically. To remove this warning, add it to the `middleware` list in your configuration file (after `django_structlog.middlewares.RequestMiddleware`). +#### INVE-W21 +**Unsupported class removed from configuration - Backend** + +The `middleware` or `authentication_backends` list provided in the configuration file includes a class which is not supported, as it would bypass security checks. + +The class has been removed automatically. To remove this warning, remove it from the list in your configuration file. + ### INVE-I (InvenTree Information) Information — These are not errors but information messages. They might point out potential issues or just provide information. diff --git a/src/backend/InvenTree/InvenTree/api_version.py b/src/backend/InvenTree/InvenTree/api_version.py index f56c6d1f1c..bf59311c2e 100644 --- a/src/backend/InvenTree/InvenTree/api_version.py +++ b/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 diff --git a/src/backend/InvenTree/InvenTree/config.py b/src/backend/InvenTree/InvenTree/config.py index 7ac1fe50db..78d64f9506 100644 --- a/src/backend/InvenTree/InvenTree/config.py +++ b/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'] diff --git a/src/backend/InvenTree/InvenTree/permissions.py b/src/backend/InvenTree/InvenTree/permissions.py index 68196292a1..30f61be551 100644 --- a/src/backend/InvenTree/InvenTree/permissions.py +++ b/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 diff --git a/src/backend/InvenTree/InvenTree/schema.py b/src/backend/InvenTree/InvenTree/schema.py index 6224f4c83b..dbca38c0e4 100644 --- a/src/backend/InvenTree/InvenTree/schema.py +++ b/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 diff --git a/src/backend/InvenTree/InvenTree/setting/spectacular.py b/src/backend/InvenTree/InvenTree/setting/spectacular.py index c83e1f8265..4133d9ef59 100644 --- a/src/backend/InvenTree/InvenTree/setting/spectacular.py +++ b/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', diff --git a/src/backend/InvenTree/InvenTree/settings.py b/src/backend/InvenTree/InvenTree/settings.py index 0dedd57c3e..71bfbcbc2b 100644 --- a/src/backend/InvenTree/InvenTree/settings.py +++ b/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 diff --git a/src/backend/InvenTree/InvenTree/test_api.py b/src/backend/InvenTree/InvenTree/test_api.py index 8491cfa6de..8e66bc3c95 100644 --- a/src/backend/InvenTree/InvenTree/test_api.py +++ b/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.""" diff --git a/src/backend/InvenTree/InvenTree/test_oauth2.py b/src/backend/InvenTree/InvenTree/test_oauth2.py new file mode 100644 index 0000000000..ab08ff1bd9 --- /dev/null +++ b/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)) diff --git a/src/backend/InvenTree/users/api.py b/src/backend/InvenTree/users/api.py index eb0bb84f3f..f42d6ed0d5 100644 --- a/src/backend/InvenTree/users/api.py +++ b/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): diff --git a/src/backend/InvenTree/users/authentication.py b/src/backend/InvenTree/users/authentication.py index 2f9172e292..9e6acc7e23 100644 --- a/src/backend/InvenTree/users/authentication.py +++ b/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() + ) diff --git a/src/backend/InvenTree/users/oauth2_scopes.py b/src/backend/InvenTree/users/oauth2_scopes.py index 80897a5f4d..dc2a57a127 100644 --- a/src/backend/InvenTree/users/oauth2_scopes.py +++ b/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',