mirror of
https://github.com/inventree/InvenTree.git
synced 2026-11-10 10:57:52 +00:00
Token fixes (#13175)
* Prevent junk token from bypassing MFA * updated docs * Fix read-only token fields * Fix API version ref
This commit is contained in:
1 parent
315b5bd134
commit
d8f09fa078
11 files changed
+209
-31
No files matched your search
@@ -1,11 +1,15 @@
|
||||
"""InvenTree API version information."""
|
||||
|
||||
# InvenTree API version
|
||||
INVENTREE_API_VERSION = 569
|
||||
INVENTREE_API_VERSION = 570
|
||||
"""Increment this API version number whenever there is a significant change to the API that any clients need to know about."""
|
||||
|
||||
INVENTREE_API_TEXT = """
|
||||
|
||||
v570 -> 2026-11-03 : https://github.com/inventree/InvenTree/pull/13175
|
||||
- Token API (/api/user/tokens/): "revoked", "revoked_by", "issued_by", "token_version", "revocation_reason" and "last_seen" fields are read-only
|
||||
- Token API (/api/user/tokens/): "expiry" must be between today and 365 days in the future
|
||||
|
||||
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/)
|
||||
|
||||
@@ -18,7 +18,6 @@ from error_report.middleware import ExceptionProcessor
|
||||
from rest_framework.exceptions import AuthenticationFailed
|
||||
|
||||
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 (
|
||||
@@ -26,7 +25,7 @@ from InvenTree.version import (
|
||||
inventreePythonVersion,
|
||||
inventreeVersion,
|
||||
)
|
||||
from users.authentication import validate_api_token
|
||||
from users.authentication import mfa_enforced, validate_api_token
|
||||
|
||||
logger = structlog.get_logger('inventree')
|
||||
|
||||
@@ -49,6 +48,32 @@ def get_token_from_request(request):
|
||||
return None
|
||||
|
||||
|
||||
def get_api_token(request):
|
||||
"""Return the validated ApiToken provided in the request headers, if any.
|
||||
|
||||
A valid token is cached on the request as 'request.token'.
|
||||
This attribute is *only* set once the token has been fully validated.
|
||||
"""
|
||||
if token := getattr(request, 'token', None):
|
||||
return token
|
||||
|
||||
if not (raw_token := get_token_from_request(request)):
|
||||
return None
|
||||
|
||||
try:
|
||||
token = validate_api_token(raw_token)
|
||||
except AuthenticationFailed as exc:
|
||||
logger.warning(
|
||||
'Access denied for token %s: %s',
|
||||
InvenTree.helpers.sanitize_token(str(raw_token)),
|
||||
exc.detail,
|
||||
)
|
||||
return None
|
||||
|
||||
request.token = token
|
||||
return token
|
||||
|
||||
|
||||
def ensure_slashes(path: str):
|
||||
"""Ensure that slashes are surrounding the passed path."""
|
||||
if not path.startswith('/'):
|
||||
@@ -147,19 +172,7 @@ class AuthRequiredMiddleware:
|
||||
|
||||
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 (active) user?
|
||||
try:
|
||||
token = validate_api_token(token)
|
||||
except AuthenticationFailed as exc:
|
||||
logger.warning(
|
||||
'Access denied for token %s: %s',
|
||||
InvenTree.helpers.sanitize_token(str(token)),
|
||||
exc.detail,
|
||||
)
|
||||
return False
|
||||
|
||||
if token := get_api_token(request):
|
||||
# Provide the user information to the request
|
||||
request.user = token.user
|
||||
return True
|
||||
@@ -271,9 +284,9 @@ class Check2FAMiddleware(MiddlewareMixin):
|
||||
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
|
||||
# Token based login can not do MFA - but the token must be valid,
|
||||
# and belong to the same user (otherwise a junk header bypasses MFA)
|
||||
if (token := get_api_token(request)) and token.user_id == request.user.pk:
|
||||
return None
|
||||
|
||||
if self.enforce_2fa(request):
|
||||
@@ -282,9 +295,7 @@ class Check2FAMiddleware(MiddlewareMixin):
|
||||
|
||||
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'
|
||||
)
|
||||
return mfa_enforced()
|
||||
|
||||
|
||||
class InvenTreeRemoteUserMiddleware(PersistentRemoteUserMiddleware):
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from django.contrib.auth.models import User
|
||||
from django.http import Http404, HttpRequest
|
||||
from django.urls import reverse
|
||||
|
||||
@@ -104,6 +105,59 @@ class MiddlewareTests(InvenTreeTestCase):
|
||||
# Not enforced - token can be generated
|
||||
self.check_path(url)
|
||||
|
||||
def test_Check2FAMiddleware_token_header(self):
|
||||
"""Test that a token header only bypasses MFA if it validly authenticates the same user."""
|
||||
target = reverse('api-license')
|
||||
token_url = reverse('api-token')
|
||||
|
||||
other_user = User.objects.create_user(username='other', password='password')
|
||||
other_token = ApiToken.objects.create(user=other_user, name='other').token
|
||||
|
||||
with patch.dict('os.environ', {'INVENTREE_LOGIN_ENFORCE_MFA': 'True'}):
|
||||
for header in [
|
||||
'Bearer junk',
|
||||
'token: junk',
|
||||
'Token junk',
|
||||
f'Token {other_token}',
|
||||
f'Bearer {other_token}',
|
||||
]:
|
||||
response = self.check_path(target, 401, auth_header=header)
|
||||
self.assertContains(response, 'mfa_register', status_code=401)
|
||||
|
||||
# Token issuance must not be reachable either
|
||||
response = self.check_path(token_url, 401, auth_header=header)
|
||||
self.assertContains(response, 'mfa_register', status_code=401)
|
||||
|
||||
self.assertFalse(ApiToken.objects.filter(user=self.user).exists())
|
||||
|
||||
def test_token_auth_mfa_enforced(self):
|
||||
"""Test that tokens of users without MFA are refused while MFA is enforced."""
|
||||
api_target = reverse('api-license')
|
||||
target = reverse('auth-check')
|
||||
|
||||
raw_token = ApiToken.objects.create(user=self.user, name='mfa-test').token
|
||||
self.client.logout()
|
||||
|
||||
self.check_path(api_target, auth_header=f'Token {raw_token}')
|
||||
self.check_path(target, auth_header=f'Token {raw_token}')
|
||||
|
||||
with patch.dict('os.environ', {'INVENTREE_LOGIN_ENFORCE_MFA': 'True'}):
|
||||
# User has no MFA set up - token is refused
|
||||
response = self.check_path(
|
||||
api_target, 401, auth_header=f'Token {raw_token}'
|
||||
)
|
||||
self.assertContains(
|
||||
response,
|
||||
'Multi-factor authentication must be enabled for this account',
|
||||
status_code=401,
|
||||
)
|
||||
self.check_path(target, 401, auth_header=f'Token {raw_token}')
|
||||
|
||||
# User sets up MFA - token works again
|
||||
get_codes(self.user)
|
||||
self.check_path(api_target, auth_header=f'Token {raw_token}')
|
||||
self.check_path(target, auth_header=f'Token {raw_token}')
|
||||
|
||||
def test_token_auth(self):
|
||||
"""Test auth with token auth."""
|
||||
target = reverse('api-license')
|
||||
|
||||
@@ -1192,7 +1192,7 @@ SYSTEM_SETTINGS: dict[str, InvenTreeSettingsKeyType] = {
|
||||
'validator': bool,
|
||||
'confirm': True,
|
||||
'confirm_text': _(
|
||||
'Enabling this setting will require all users to set up multifactor authentication. All sessions will be disconnected immediately.'
|
||||
'Enabling this setting will require all users to set up multifactor authentication. All sessions will be disconnected immediately, and API tokens of users without multifactor authentication will stop working.'
|
||||
),
|
||||
'flags': [SettingFlag.SECURITY],
|
||||
'after_save': enforce_mfa,
|
||||
|
||||
@@ -13,21 +13,26 @@ import users.models
|
||||
from common.settings import get_global_setting
|
||||
|
||||
|
||||
def mfa_enforced() -> bool:
|
||||
"""Check if MFA is enforced globally for all users."""
|
||||
return bool(
|
||||
settings.MFA_ENABLED
|
||||
and get_global_setting('LOGIN_ENFORCE_MFA', None, 'INVENTREE_LOGIN_ENFORCE_MFA')
|
||||
)
|
||||
|
||||
|
||||
def password_auth_allowed(user) -> bool:
|
||||
"""Check if a password alone is allowed to authenticate the user."""
|
||||
if is_mfa_enabled(user):
|
||||
return False
|
||||
return not (
|
||||
settings.MFA_ENABLED
|
||||
and get_global_setting('LOGIN_ENFORCE_MFA', None, 'INVENTREE_LOGIN_ENFORCE_MFA')
|
||||
)
|
||||
return not mfa_enforced()
|
||||
|
||||
|
||||
class InvenTreeBasicAuthentication(BasicAuthentication):
|
||||
"""Custom basic auth class.
|
||||
|
||||
Checks:
|
||||
- If MFA is required and dissalows basic auth
|
||||
- If MFA is required and disallows basic auth
|
||||
"""
|
||||
|
||||
def authenticate_credentials(self, userid, password, request=None):
|
||||
@@ -49,7 +54,8 @@ def validate_api_token(raw_token: str) -> users.models.ApiToken:
|
||||
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
|
||||
AuthenticationFailed: If the token is unknown, revoked, expired, belongs to an inactive user,
|
||||
or belongs to a user without MFA while MFA is enforced
|
||||
"""
|
||||
token = users.models.ApiToken.get_from_string(raw_token) if raw_token else None
|
||||
|
||||
@@ -65,6 +71,12 @@ def validate_api_token(raw_token: str) -> users.models.ApiToken:
|
||||
if token.expired:
|
||||
raise exceptions.AuthenticationFailed(_('Token has expired'))
|
||||
|
||||
# Tokens issued before MFA was enforced must not outlive the enforcement
|
||||
if mfa_enforced() and not is_mfa_enabled(token.user):
|
||||
raise exceptions.AuthenticationFailed(
|
||||
_('Multi-factor authentication must be enabled for this account')
|
||||
)
|
||||
|
||||
today = InvenTree.helpers.current_date()
|
||||
|
||||
if token.last_seen != today:
|
||||
|
||||
@@ -10,13 +10,21 @@ from django.utils.translation import gettext_lazy as _
|
||||
from rest_framework import serializers
|
||||
from rest_framework.exceptions import PermissionDenied
|
||||
|
||||
import InvenTree.helpers
|
||||
from InvenTree.serializers import (
|
||||
FilterableSerializerMixin,
|
||||
InvenTreeModelSerializer,
|
||||
OptionalField,
|
||||
)
|
||||
|
||||
from .models import ApiToken, Owner, RuleSet, TipResult, UserProfile
|
||||
from .models import (
|
||||
ApiToken,
|
||||
Owner,
|
||||
RuleSet,
|
||||
TipResult,
|
||||
UserProfile,
|
||||
default_token_expiry,
|
||||
)
|
||||
from .permissions import check_user_role, prefetch_rule_sets
|
||||
from .ruleset import RULESET_CHOICES, RULESET_PERMISSIONS, RuleSetEnum
|
||||
|
||||
@@ -259,6 +267,26 @@ class ApiTokenSerializer(InvenTreeModelSerializer):
|
||||
'token_version',
|
||||
'revocation_reason',
|
||||
]
|
||||
read_only_fields = [
|
||||
'last_seen',
|
||||
'revoked',
|
||||
'revoked_by',
|
||||
'issued_by',
|
||||
'token_version',
|
||||
'revocation_reason',
|
||||
]
|
||||
|
||||
def validate_expiry(self, expiry):
|
||||
"""Ensure that the expiry date is within the allowed token lifetime."""
|
||||
if expiry < InvenTree.helpers.current_date():
|
||||
raise serializers.ValidationError(_('Expiry date cannot be in the past'))
|
||||
|
||||
if expiry > default_token_expiry():
|
||||
raise serializers.ValidationError(
|
||||
_('Expiry date cannot be more than 365 days in the future')
|
||||
)
|
||||
|
||||
return expiry
|
||||
|
||||
def validate(self, data):
|
||||
"""Validate the data for the serializer."""
|
||||
|
||||
@@ -8,6 +8,7 @@ from django.urls import reverse
|
||||
|
||||
from allauth.account.models import EmailAddress
|
||||
|
||||
import InvenTree.helpers
|
||||
from common.models import NotificationMessage
|
||||
from InvenTree.unit_test import InvenTreeAPITestCase
|
||||
from users.models import ApiToken, default_token
|
||||
@@ -720,6 +721,50 @@ class UserTokenTests(InvenTreeAPITestCase):
|
||||
|
||||
self.assertEqual(ApiToken.objects.count(), 1)
|
||||
|
||||
def test_token_create_fields(self):
|
||||
"""Test that protected token fields cannot be set when creating a token."""
|
||||
url = reverse('api-token-list')
|
||||
today = InvenTree.helpers.current_date()
|
||||
other = User.objects.create_user(username='other', password='password')
|
||||
|
||||
# Expiry date must be within the allowed token lifetime
|
||||
for expiry in [today - datetime.timedelta(days=1), datetime.date(9999, 12, 31)]:
|
||||
response = self.post(
|
||||
url, data={'name': 'test', 'expiry': str(expiry)}, expected_code=400
|
||||
)
|
||||
self.assertIn('expiry', response.data)
|
||||
|
||||
for expiry in [today, today + datetime.timedelta(days=365)]:
|
||||
response = self.post(
|
||||
url,
|
||||
data={'name': f'test-{expiry}', 'expiry': str(expiry)},
|
||||
expected_code=201,
|
||||
)
|
||||
self.assertEqual(response.data['expiry'], str(expiry))
|
||||
|
||||
# Read-only fields are ignored
|
||||
response = self.post(
|
||||
url,
|
||||
data={
|
||||
'name': 'protected',
|
||||
'revoked': True,
|
||||
'revoked_by': other.pk,
|
||||
'issued_by': other.pk,
|
||||
'revocation_reason': 'forged',
|
||||
'token_version': 1,
|
||||
'last_seen': str(today),
|
||||
},
|
||||
expected_code=201,
|
||||
)
|
||||
|
||||
token = ApiToken.objects.get(pk=response.data['id'])
|
||||
self.assertFalse(token.revoked)
|
||||
self.assertIsNone(token.revoked_by)
|
||||
self.assertEqual(token.issued_by, self.user)
|
||||
self.assertIsNone(token.revocation_reason)
|
||||
self.assertEqual(token.token_version, 2)
|
||||
self.assertIsNone(token.last_seen)
|
||||
|
||||
def test_token_v1(self):
|
||||
"""Test that v1 API tokens still work."""
|
||||
# Create a v1 token via model - this is NOT recommended; use v2 tokens
|
||||
|
||||
Reference in new issue
Block a user