mirror of
https://github.com/inventree/InvenTree.git
synced 2026-09-27 14:16:02 +00:00
feat(backend): switch API token storage to hmac digest (#12850)
* [FR] switch API token storage to hmac digest Fixes #12027 * add pr number * fix implementation details allign token prefix and token end (add date issuance) * fix missing re-issuance * ensure we raise if not exsistant * fix test * add ignore * extend test * add information regarding v2 tokens * add fields for: - revocation details (reason, revoker) - issueance - version - pepper * bump apiversion * make revocation mesagge clearer * add missing update fields * ensure this only triggers for v2 * move comment * re-add v1 coverage * add missing annotation to api schema
This commit is contained in:
Matthias Mair
@@ -351,10 +351,9 @@ class InfoView(APIView):
|
||||
if token := get_token_from_request(request):
|
||||
# Does the provided token match a valid user?
|
||||
try:
|
||||
token = ApiToken.objects.get(key=token)
|
||||
|
||||
token = ApiToken.get_from_string(token)
|
||||
# Check if the token is active and the user is a staff member
|
||||
if token.active and token.user and token.user.is_staff:
|
||||
if token and token.active and token.user and token.user.is_staff:
|
||||
return True
|
||||
except ApiToken.DoesNotExist:
|
||||
pass
|
||||
|
||||
@@ -1,12 +1,15 @@
|
||||
"""InvenTree API version information."""
|
||||
|
||||
# InvenTree API version
|
||||
INVENTREE_API_VERSION = 546
|
||||
INVENTREE_API_VERSION = 547
|
||||
"""Increment this API version number whenever there is a significant change to the API that any clients need to know about."""
|
||||
|
||||
INVENTREE_API_TEXT = """
|
||||
|
||||
v546 -> 2026-09-12 : https://github.com/inventree/InvenTree/pull/12842/changes
|
||||
v547 -> 2026-09-12 : https://github.com/inventree/InvenTree/pull/12850
|
||||
- Added more details to API token management
|
||||
|
||||
v546 -> 2026-09-12 : https://github.com/inventree/InvenTree/pull/12842
|
||||
- Added setting flags to the settings APIs (read-only)
|
||||
|
||||
v545 -> 2026-09-08 : https://github.com/inventree/InvenTree/pull/12808
|
||||
|
||||
@@ -1150,3 +1150,8 @@ def sanitize_token(token_value: str, front=8, back=12) -> str:
|
||||
"""
|
||||
middle = len(token_value) - (front + back)
|
||||
return token_value[:front] + '*' * middle + token_value[-back:]
|
||||
|
||||
|
||||
def get_api_token_pepper() -> str:
|
||||
"""Return the secret 'pepper' used to compute v2 tokens."""
|
||||
return settings.SECRET_KEY
|
||||
|
||||
@@ -150,9 +150,9 @@ class AuthRequiredMiddleware:
|
||||
request.token = token
|
||||
# Does the provided token match a valid user?
|
||||
try:
|
||||
token = ApiToken.objects.get(key=token)
|
||||
token = ApiToken.get_from_string(token)
|
||||
|
||||
if token.active and token.user:
|
||||
if token and token.active and token.user:
|
||||
# Provide the user information to the request
|
||||
request.user = token.user
|
||||
return True
|
||||
|
||||
@@ -24,6 +24,7 @@ import InvenTree.permissions
|
||||
from InvenTree.fields import InvenTreeOutputOption, OutputConfiguration
|
||||
from InvenTree.filters import SEARCH_ORDER_FILTER
|
||||
from InvenTree.mixins import (
|
||||
CleanBase,
|
||||
ListAPI,
|
||||
ListCreateAPI,
|
||||
OutputOptionsMixin,
|
||||
@@ -389,30 +390,43 @@ class GetAuthToken(GenericAPIView):
|
||||
- Existing tokens are *never* exposed again via the API
|
||||
- Once the token is provided, it can be used for auth until it expires
|
||||
"""
|
||||
if not request.user.is_authenticated:
|
||||
raise exceptions.NotAuthenticated() # pragma: no cover
|
||||
|
||||
user = request.user
|
||||
name = request.query_params.get('name', '')
|
||||
|
||||
if not user.is_authenticated:
|
||||
raise exceptions.NotAuthenticated() # pragma: no cover
|
||||
|
||||
name = ApiToken.sanitize_name(name)
|
||||
|
||||
today = datetime.date.today()
|
||||
reissue_token = request.resolver_match.url_name == 'api-token'
|
||||
|
||||
# Find existing token, which has not expired
|
||||
token = ApiToken.objects.filter(
|
||||
user=user, name=name, revoked=False, expiry__gte=today
|
||||
).first()
|
||||
|
||||
if not token:
|
||||
if token and reissue_token:
|
||||
token.revoked = True
|
||||
token.revoked_by = user
|
||||
token.revocation_reason = (
|
||||
're-issued due to new token request to API with same name'
|
||||
)
|
||||
token.save(update_fields=['revoked', 'revoked_by', 'revocation_reason'])
|
||||
|
||||
if not token or reissue_token:
|
||||
# User is authenticated, and requesting a token against the provided name.
|
||||
token = ApiToken.objects.create(user=request.user, name=name)
|
||||
token = ApiToken.objects.create(user=user, name=name, issued_by=user)
|
||||
|
||||
logger.info(
|
||||
"Created new API token for user '%s' (name='%s')", user.username, name
|
||||
)
|
||||
|
||||
# Add some metadata about the request
|
||||
if token.token_version == 2 and token.hmac_digest and not token._raw_secret:
|
||||
raise exceptions.ValidationError(
|
||||
'Token is not newly created.'
|
||||
) # pragma: no cover
|
||||
|
||||
# Add some metadata about the request
|
||||
token.set_metadata('user_agent', request.headers.get('user-agent', ''))
|
||||
token.set_metadata('remote_addr', request.META.get('REMOTE_ADDR', ''))
|
||||
token.set_metadata('remote_host', request.META.get('REMOTE_HOST', ''))
|
||||
@@ -420,7 +434,11 @@ class GetAuthToken(GenericAPIView):
|
||||
token.set_metadata('server_name', request.META.get('SERVER_NAME', ''))
|
||||
token.set_metadata('server_port', request.META.get('SERVER_PORT', ''))
|
||||
|
||||
data = {'token': token.key, 'name': token.name, 'expiry': token.expiry}
|
||||
data = {
|
||||
'token': token.token if token.token_version == 2 else token.key,
|
||||
'name': token.name,
|
||||
'expiry': token.expiry,
|
||||
}
|
||||
|
||||
# Ensure that the users session is logged in
|
||||
if not get_user(request).is_authenticated:
|
||||
@@ -468,17 +486,32 @@ class TokenListView(TokenMixin, ListCreateAPI):
|
||||
'user__first_name',
|
||||
'user__last_name',
|
||||
'user__email',
|
||||
'revocation_reason',
|
||||
]
|
||||
ordering_fields = ['created', 'expiry', 'last_seen', 'user', 'name', 'revoked']
|
||||
filterset_fields = ['revoked', 'user']
|
||||
ordering_fields = [
|
||||
'created',
|
||||
'expiry',
|
||||
'last_seen',
|
||||
'user',
|
||||
'name',
|
||||
'revoked',
|
||||
'revoked_by',
|
||||
'issued_by',
|
||||
'token_version',
|
||||
'revocation_reason',
|
||||
]
|
||||
filterset_fields = ['revoked', 'user', 'issued_by', 'revoked_by']
|
||||
queryset = ApiToken.objects.none()
|
||||
|
||||
def perform_create(self, serializer):
|
||||
"""Save the new token and keep the secret (only available immediately after creation)."""
|
||||
serializer.save(issued_by=self.request.user)
|
||||
self._created_token = serializer.instance
|
||||
|
||||
def create(self, request, *args, **kwargs):
|
||||
"""Create token and show key to user."""
|
||||
resp = super().create(request, *args, **kwargs)
|
||||
resp.data['token'] = self.serializer_class.Meta.model.objects.get(
|
||||
id=resp.data['id']
|
||||
).key
|
||||
resp.data['token'] = self._created_token.token
|
||||
return resp
|
||||
|
||||
def get(self, request, *args, **kwargs):
|
||||
@@ -486,13 +519,32 @@ class TokenListView(TokenMixin, ListCreateAPI):
|
||||
return super().get(request, *args, **kwargs)
|
||||
|
||||
|
||||
class TokenDetailView(TokenMixin, DestroyAPIView, RetrieveAPI):
|
||||
class TokenDetailView(CleanBase, TokenMixin, DestroyAPIView, RetrieveAPI):
|
||||
"""Details for a user token."""
|
||||
|
||||
@extend_schema(
|
||||
parameters=[
|
||||
OpenApiParameter(
|
||||
name='revocation_reason',
|
||||
type=str,
|
||||
description='Reason for revoking the token.',
|
||||
default='',
|
||||
)
|
||||
]
|
||||
)
|
||||
def delete(self, request, *args, **kwargs):
|
||||
"""Revoke this specific user token."""
|
||||
return super().delete(request, *args, **kwargs)
|
||||
|
||||
def perform_destroy(self, instance):
|
||||
"""Revoke token."""
|
||||
instance.revoked = True
|
||||
instance.save()
|
||||
instance.revoked_by = self.request.user
|
||||
request_data = getattr(self.request, 'data', {})
|
||||
instance.revocation_reason = self.clean_string(
|
||||
'revocation_reason', str(request_data.get('revocation_reason', ''))
|
||||
)
|
||||
instance.save(update_fields=['revoked', 'revoked_by', 'revocation_reason'])
|
||||
|
||||
|
||||
class LoginRedirect(RedirectView):
|
||||
|
||||
@@ -24,7 +24,13 @@ 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)
|
||||
(user, token) = super().authenticate_credentials(key)
|
||||
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'))
|
||||
|
||||
+59
@@ -0,0 +1,59 @@
|
||||
# Generated by Django 5.2.17 on 2026-09-15 19:59
|
||||
|
||||
import django.core.validators
|
||||
import django.db.models.deletion
|
||||
from django.conf import settings
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
def mark_existing_tokens_as_v1(apps, schema_editor):
|
||||
"""Mark tokens created before v2 token support as legacy v1 tokens."""
|
||||
ApiToken = apps.get_model('users', 'ApiToken')
|
||||
ApiToken.objects.all().update(token_version=1)
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('users', '0016_remove_legacy_user_sessions_table'),
|
||||
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='apitoken',
|
||||
name='hmac_digest',
|
||||
field=models.CharField(blank=True, max_length=200, null=True),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='apitoken',
|
||||
name='issued_by',
|
||||
field=models.ForeignKey(blank=True, help_text='User who issued the token', null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='issued_api_tokens', to=settings.AUTH_USER_MODEL, verbose_name='Issued By'),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='apitoken',
|
||||
name='pepper_id',
|
||||
field=models.CharField(blank=True, help_text='Identifier for the pepper used in token hashing', max_length=100, null=True, verbose_name='Pepper ID'),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='apitoken',
|
||||
name='revocation_reason',
|
||||
field=models.TextField(blank=True, help_text='As entered by the user or action during revocation', null=True, verbose_name='Revocation Reason'),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='apitoken',
|
||||
name='revoked_by',
|
||||
field=models.ForeignKey(blank=True, help_text='User who revoked the token', null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='revoked_api_tokens', to=settings.AUTH_USER_MODEL, verbose_name='Revoked By'),
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='apitoken',
|
||||
name='token_version',
|
||||
field=models.PositiveSmallIntegerField(default=2, help_text='Version of the API token', verbose_name='Token Version'),
|
||||
),
|
||||
migrations.AlterField(
|
||||
model_name='apitoken',
|
||||
name='key',
|
||||
field=models.CharField(blank=True, db_index=True, max_length=100, unique=True, validators=[django.core.validators.MinLengthValidator(8)], verbose_name='Key'),
|
||||
),
|
||||
migrations.RunPython(mark_existing_tokens_as_v1, migrations.RunPython.noop),
|
||||
]
|
||||
@@ -1,6 +1,10 @@
|
||||
"""Database model definitions for the 'users' app."""
|
||||
|
||||
import datetime
|
||||
import hashlib
|
||||
import hmac
|
||||
import secrets
|
||||
from typing import Optional
|
||||
|
||||
from django.conf import settings
|
||||
from django.contrib import admin
|
||||
@@ -68,6 +72,12 @@ if settings.LDAP_AUTH: # pragma: no cover
|
||||
EmailAddress.objects.create(user=user, email=user.email, primary=True)
|
||||
|
||||
|
||||
API_TOKEN_PREFIX = 'inv-2-'
|
||||
API_TOKEN_IDENTIFIER_LENGTH = 20
|
||||
API_TOKEN_SECRET_LENGTH = 40
|
||||
|
||||
|
||||
# legacy TODO @matmair remove in the next breaking
|
||||
def default_token():
|
||||
"""Generate a default value for the token."""
|
||||
return ApiToken.generate_key()
|
||||
@@ -84,6 +94,10 @@ class ApiToken(AuthToken, InvenTree.models.MetadataMixin):
|
||||
Extensions:
|
||||
- Adds an 'expiry' date - tokens can be set to expire after a certain date
|
||||
- Adds a 'name' field - tokens can be given a custom name (in addition to the user information)
|
||||
|
||||
Token storage:
|
||||
- Old v1 tokens store the raw plaintext token value directly in the 'key' field.
|
||||
- Current v2 tokens only store a identifier in the 'key' field, and the secret is only persisted as hmac digest
|
||||
"""
|
||||
|
||||
class Meta:
|
||||
@@ -93,28 +107,143 @@ class ApiToken(AuthToken, InvenTree.models.MetadataMixin):
|
||||
verbose_name_plural = _('API Tokens')
|
||||
abstract = False
|
||||
|
||||
_raw_secret = None # Temp storage for secret
|
||||
|
||||
def __str__(self):
|
||||
"""String representation uses the redacted token."""
|
||||
return self.token
|
||||
|
||||
@classmethod
|
||||
def generate_key(cls, prefix='inv-'):
|
||||
"""Generate a new token key - with custom prefix."""
|
||||
"""Generate a new old token key - with custom prefix."""
|
||||
# Suffix is the date of creation
|
||||
suffix = '-' + str(datetime.datetime.now().date().isoformat().replace('-', ''))
|
||||
|
||||
return prefix + str(AuthToken.generate_key()) + suffix
|
||||
|
||||
# Override the 'key' field - force it to be unique
|
||||
def generate_v2_token(self) -> None:
|
||||
"""Generate new v2 token."""
|
||||
identifier = secrets.token_hex(API_TOKEN_IDENTIFIER_LENGTH // 2)
|
||||
secret = secrets.token_hex(API_TOKEN_SECRET_LENGTH // 2)
|
||||
suffix = '-' + str(datetime.datetime.now().date().isoformat().replace('-', ''))
|
||||
|
||||
self.key = identifier
|
||||
self.hmac_digest = self.calculate_digest(secret)
|
||||
self._raw_secret = f'{API_TOKEN_PREFIX}{identifier}.{secret}{suffix}'
|
||||
|
||||
# metadata
|
||||
self.token_version = 2
|
||||
self.pepper_id = self.calculate_pepper_id()
|
||||
|
||||
@staticmethod
|
||||
def calculate_digest(secret: str) -> str:
|
||||
"""Calculate the HMAC digest of the provided secret."""
|
||||
pepper = InvenTree.helpers.get_api_token_pepper()
|
||||
return hmac.new(
|
||||
pepper.encode('utf-8'), secret.encode('utf-8'), hashlib.sha256
|
||||
).hexdigest()
|
||||
|
||||
@classmethod
|
||||
def calculate_pepper_id(cls, length=8) -> str:
|
||||
"""Calculate the first 8 characters of the current pepper hashed."""
|
||||
pepper = InvenTree.helpers.get_api_token_pepper()
|
||||
return hashlib.sha256(pepper.encode('utf-8')).hexdigest()[:length]
|
||||
|
||||
@staticmethod
|
||||
def split_token(raw_token: str):
|
||||
"""Split a raw v2 token value into the required values."""
|
||||
if not raw_token:
|
||||
return None # pragma: no cover
|
||||
|
||||
value = (
|
||||
raw_token[len(API_TOKEN_PREFIX) :]
|
||||
if raw_token.startswith(API_TOKEN_PREFIX)
|
||||
else raw_token
|
||||
)
|
||||
|
||||
if '.' not in value:
|
||||
return None
|
||||
|
||||
identifier, _sep, secret = value.partition('.')
|
||||
if not identifier or not secret:
|
||||
return None # pragma: no cover
|
||||
|
||||
secret_parts = secret.rsplit('-', 1)
|
||||
if (
|
||||
len(secret_parts) == 2
|
||||
and len(secret_parts[1]) == 8
|
||||
and secret_parts[1].isdigit()
|
||||
):
|
||||
secret = secret_parts[0]
|
||||
|
||||
return identifier, secret
|
||||
|
||||
def match(self, raw_token: str) -> bool:
|
||||
"""Lightweight check for whether raw_token refers to *this* token instance."""
|
||||
if not raw_token:
|
||||
return False # pragma: no cover
|
||||
|
||||
# is a v2 token
|
||||
if self.hmac_digest:
|
||||
parts = self.split_token(raw_token)
|
||||
return bool(parts) and parts[0] == self.key
|
||||
|
||||
return raw_token == self.key
|
||||
|
||||
def validate(self, raw_token: str) -> bool:
|
||||
"""Validate a raw token value against this token."""
|
||||
if not raw_token:
|
||||
return False # pragma: no cover
|
||||
|
||||
# is a v2 token
|
||||
if self.hmac_digest:
|
||||
parts = self.split_token(raw_token)
|
||||
|
||||
# check id
|
||||
if not parts or parts[0] != self.key:
|
||||
return False # pragma: no cover
|
||||
# check secret
|
||||
return hmac.compare_digest(
|
||||
self.calculate_digest(parts[1]), self.hmac_digest
|
||||
)
|
||||
|
||||
# should be a v1 - compare directly
|
||||
return raw_token == self.key
|
||||
|
||||
@classmethod
|
||||
def lookup(cls, raw_token: str) -> 'ApiToken':
|
||||
"""Look up an ApiToken instance from a token string (v1 or v2). Not a full validation."""
|
||||
parts = cls.split_token(raw_token)
|
||||
return cls.objects.select_related('user').get(
|
||||
key=parts[0] if parts else raw_token
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def get_from_string(cls, raw_token: str) -> Optional['ApiToken']:
|
||||
"""Look up and fully validate an ApiToken from a raw token string."""
|
||||
try:
|
||||
token = cls.lookup(raw_token)
|
||||
except cls.DoesNotExist:
|
||||
return None
|
||||
|
||||
if not token.validate(raw_token):
|
||||
return None # pragma: no cover
|
||||
|
||||
return token
|
||||
|
||||
# in v1: private token; in v2: public identifier
|
||||
key = models.CharField(
|
||||
default=default_token,
|
||||
verbose_name=_('Key'),
|
||||
db_index=True,
|
||||
unique=True,
|
||||
max_length=100,
|
||||
validators=[MinLengthValidator(50)],
|
||||
blank=True,
|
||||
validators=[MinLengthValidator(8)],
|
||||
)
|
||||
|
||||
# in v2: HMAC digest of token; empty in v1
|
||||
hmac_digest = models.CharField(max_length=200, blank=True, null=True)
|
||||
|
||||
# Override the 'user' field, to allow multiple tokens per user
|
||||
user = models.ForeignKey(
|
||||
settings.AUTH_USER_MODEL,
|
||||
@@ -149,9 +278,51 @@ class ApiToken(AuthToken, InvenTree.models.MetadataMixin):
|
||||
default=False, verbose_name=_('Revoked'), help_text=_('Token has been revoked')
|
||||
)
|
||||
|
||||
revocation_reason = models.TextField(
|
||||
blank=True,
|
||||
null=True,
|
||||
verbose_name=_('Revocation Reason'),
|
||||
help_text=_('As entered by the user or action during revocation'),
|
||||
)
|
||||
|
||||
revoked_by = models.ForeignKey(
|
||||
settings.AUTH_USER_MODEL,
|
||||
on_delete=models.SET_NULL,
|
||||
blank=True,
|
||||
null=True,
|
||||
verbose_name=_('Revoked By'),
|
||||
help_text=_('User who revoked the token'),
|
||||
related_name='revoked_api_tokens',
|
||||
)
|
||||
|
||||
issued_by = models.ForeignKey(
|
||||
settings.AUTH_USER_MODEL,
|
||||
on_delete=models.SET_NULL,
|
||||
blank=True,
|
||||
null=True,
|
||||
verbose_name=_('Issued By'),
|
||||
help_text=_('User who issued the token'),
|
||||
related_name='issued_api_tokens',
|
||||
)
|
||||
|
||||
token_version = models.PositiveSmallIntegerField(
|
||||
default=2,
|
||||
verbose_name=_('Token Version'),
|
||||
help_text=_('Version of the API token'),
|
||||
)
|
||||
|
||||
pepper_id = models.CharField(
|
||||
max_length=100,
|
||||
blank=True,
|
||||
null=True,
|
||||
verbose_name=_('Pepper ID'),
|
||||
help_text=_('Identifier for the pepper used in token hashing'),
|
||||
)
|
||||
"""By default, the first 8 characters of the pepper hashed."""
|
||||
|
||||
@staticmethod
|
||||
def sanitize_name(name: str) -> str:
|
||||
"""Sanitize the provide name value."""
|
||||
"""Sanitize the provided name value."""
|
||||
name = str(name).strip()
|
||||
|
||||
# Remove any non-printable chars
|
||||
@@ -166,6 +337,17 @@ class ApiToken(AuthToken, InvenTree.models.MetadataMixin):
|
||||
|
||||
return name
|
||||
|
||||
def save(self, *args, **kwargs):
|
||||
"""Patch in token generations."""
|
||||
if self._state.adding and not self.key and not self.hmac_digest:
|
||||
if self.token_version != 2:
|
||||
raise ValueError(
|
||||
'Unsupported token version and scenario'
|
||||
) # pragma: no cover
|
||||
self.generate_v2_token()
|
||||
|
||||
super().save(*args, **kwargs)
|
||||
|
||||
@property
|
||||
@admin.display(description=_('Token'))
|
||||
def token(self) -> str:
|
||||
@@ -173,10 +355,18 @@ class ApiToken(AuthToken, InvenTree.models.MetadataMixin):
|
||||
|
||||
The *raw* key value should never be displayed anywhere!
|
||||
"""
|
||||
# If the token has not yet been saved, return the raw key
|
||||
# on generation: v2 token
|
||||
if self._raw_secret:
|
||||
return self._raw_secret
|
||||
|
||||
# on generation: v1
|
||||
if self.pk is None:
|
||||
return self.key # pragma: no cover
|
||||
|
||||
# v2 default path
|
||||
if self.hmac_digest:
|
||||
return f'{API_TOKEN_PREFIX}{self.key}.{"*" * API_TOKEN_SECRET_LENGTH}'
|
||||
|
||||
return InvenTree.helpers.sanitize_token(self.key)
|
||||
|
||||
@property
|
||||
|
||||
@@ -222,7 +222,7 @@ class ApiTokenSerializer(InvenTreeModelSerializer):
|
||||
|
||||
request = self.context.get('request')
|
||||
rq_token = get_token_from_request(request)
|
||||
return token.key == rq_token
|
||||
return token.match(rq_token)
|
||||
|
||||
class Meta:
|
||||
"""Meta options for ApiTokenSerializer."""
|
||||
@@ -240,6 +240,12 @@ class ApiTokenSerializer(InvenTreeModelSerializer):
|
||||
'user',
|
||||
'user_detail',
|
||||
'in_use',
|
||||
'revoked_by',
|
||||
'revoked_by_detail',
|
||||
'issued_by',
|
||||
'issued_by_detail',
|
||||
'token_version',
|
||||
'revocation_reason',
|
||||
]
|
||||
|
||||
def validate(self, data):
|
||||
@@ -262,6 +268,8 @@ class ApiTokenSerializer(InvenTreeModelSerializer):
|
||||
return super().validate(data)
|
||||
|
||||
user_detail = UserSerializer(source='user', read_only=True)
|
||||
revoked_by_detail = UserSerializer(source='revoked_by', read_only=True)
|
||||
issued_by_detail = UserSerializer(source='issued_by', read_only=True)
|
||||
|
||||
|
||||
class GroupSerializer(FilterableSerializerMixin, InvenTreeModelSerializer):
|
||||
|
||||
@@ -9,7 +9,7 @@ from django.urls import reverse
|
||||
from allauth.account.models import EmailAddress
|
||||
|
||||
from InvenTree.unit_test import InvenTreeAPITestCase
|
||||
from users.models import ApiToken
|
||||
from users.models import ApiToken, default_token
|
||||
from users.ruleset import RULESET_NAMES, get_ruleset_models
|
||||
|
||||
|
||||
@@ -406,20 +406,20 @@ class UserTokenTests(InvenTreeAPITestCase):
|
||||
# Request the token with the same name
|
||||
data = self.get(url, data={'name': 'cat'}, expected_code=200).data
|
||||
|
||||
self.assertEqual(data['token'], token.key)
|
||||
token.refresh_from_db()
|
||||
self.assertNotEqual(data['token'], token.key)
|
||||
self.assertTrue(data['token'].startswith('inv-2-'))
|
||||
self.assertTrue(token.revoked)
|
||||
|
||||
self.assertEqual(ApiToken.objects.count(), 3)
|
||||
|
||||
# Revoke the token, and then request again
|
||||
token.revoked = True
|
||||
token.save()
|
||||
self.assertEqual(ApiToken.objects.count(), 4)
|
||||
|
||||
# Request again, which issues another replacement token
|
||||
data = self.get(url, data={'name': 'cat'}, expected_code=200).data
|
||||
|
||||
self.assertNotEqual(data['token'], token.key)
|
||||
|
||||
# A new token has been generated
|
||||
self.assertEqual(ApiToken.objects.count(), 4)
|
||||
self.assertEqual(ApiToken.objects.count(), 5)
|
||||
|
||||
# Test with a really long name
|
||||
data = self.get(url, data={'name': 'cat' * 100}, expected_code=200).data
|
||||
@@ -460,7 +460,7 @@ class UserTokenTests(InvenTreeAPITestCase):
|
||||
# Grab the token, and update
|
||||
token = ApiToken.objects.first()
|
||||
assert token
|
||||
self.assertEqual(token.key, token_key)
|
||||
self.assertEqual(token.key, ApiToken.split_token(token_key)[0])
|
||||
self.assertIsNotNone(token.last_seen)
|
||||
|
||||
# Revoke the token
|
||||
@@ -501,7 +501,7 @@ class UserTokenTests(InvenTreeAPITestCase):
|
||||
url=reverse('api-token'), data={'name': 'race'}, expected_code=200
|
||||
).data['token']
|
||||
|
||||
token = ApiToken.objects.get(key=token_key)
|
||||
token = ApiToken.objects.get(key=ApiToken.split_token(token_key)[0])
|
||||
|
||||
# Force last_seen to be 'stale' so the auth backend attempts to update it
|
||||
ApiToken.objects.filter(pk=token.pk).update(
|
||||
@@ -537,24 +537,38 @@ class UserTokenTests(InvenTreeAPITestCase):
|
||||
# Get token
|
||||
response = self.get(reverse('api-token'), expected_code=200)
|
||||
self.assertIn('token', response.data)
|
||||
raw_token = response.data['token']
|
||||
|
||||
self.client.logout()
|
||||
self.client.credentials(HTTP_AUTHORIZATION=f'Token {raw_token}')
|
||||
|
||||
# Now there should be one token
|
||||
response = self.get(url, expected_code=200)
|
||||
self.assertEqual(len(response.data), 1)
|
||||
self.assertEqual(response.data[0]['active'], True)
|
||||
self.assertEqual(response.data[0]['revoked'], False)
|
||||
self.assertEqual(response.data[0]['in_use'], False)
|
||||
self.assertEqual(response.data[0]['in_use'], True)
|
||||
self.assertEqual(response.data[0]['issued_by'], self.user.pk)
|
||||
self.assertIsNone(response.data[0]['revoked_by'])
|
||||
self.assertIsNone(response.data[0]['revocation_reason'])
|
||||
expected_day = str(
|
||||
datetime.datetime.now().date() + datetime.timedelta(days=365)
|
||||
)
|
||||
self.assertEqual(response.data[0]['expiry'], expected_day)
|
||||
|
||||
# Destroy token
|
||||
token_id = response.data[0]['id']
|
||||
self.delete(
|
||||
reverse('api-token-detail', kwargs={'pk': response.data[0]['id']}),
|
||||
reverse('api-token-detail', kwargs={'pk': token_id}),
|
||||
data={'revocation_reason': 'No longer needed'},
|
||||
expected_code=204,
|
||||
)
|
||||
|
||||
token = ApiToken.objects.get(pk=token_id)
|
||||
self.assertTrue(token.revoked)
|
||||
self.assertEqual(token.revoked_by, self.user)
|
||||
self.assertEqual(token.revocation_reason, 'No longer needed')
|
||||
|
||||
# Get token without auth (should fail)
|
||||
self.client.logout()
|
||||
self.get(reverse('api-token'), expected_code=401)
|
||||
@@ -581,6 +595,40 @@ class UserTokenTests(InvenTreeAPITestCase):
|
||||
|
||||
self.assertEqual(ApiToken.objects.count(), 1)
|
||||
|
||||
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
|
||||
token = ApiToken.objects.create(
|
||||
user=self.user,
|
||||
key=default_token(),
|
||||
token_version=1,
|
||||
expiry=datetime.datetime.now() + datetime.timedelta(days=365),
|
||||
)
|
||||
token_key = token.key
|
||||
self.assertTrue(token_key.startswith('inv-'))
|
||||
|
||||
# Check match and validate functions
|
||||
self.assertTrue(token.match(token_key))
|
||||
self.assertTrue(token.validate(token_key))
|
||||
|
||||
# test api access with token
|
||||
self.logout()
|
||||
# false test - ensure that without the token, access is denied
|
||||
self.get(reverse('api-user-me'), expected_code=401)
|
||||
|
||||
# valid test
|
||||
self.client.credentials(HTTP_AUTHORIZATION=f'Token {token_key}')
|
||||
response = self.get(reverse('api-user-me'), expected_code=200)
|
||||
self.assertEqual(response.data['username'], self.user.username)
|
||||
|
||||
# check if info view also works
|
||||
response_data = self.get(
|
||||
reverse('api-inventree-info'), expected_code=200
|
||||
).json()
|
||||
# staff users are allowed to see the database field
|
||||
self.assertIn('database', response_data)
|
||||
self.assertIsNotNone(response_data.get('database'))
|
||||
|
||||
|
||||
class GroupDetailTests(InvenTreeAPITestCase):
|
||||
"""Tests for the GroupDetail API endpoint."""
|
||||
|
||||
@@ -14,13 +14,30 @@ class TestForwardMigrations(MigratorTestCase):
|
||||
def prepare(self):
|
||||
"""Setup the initial state of the database before migrations."""
|
||||
User = self.old_state.apps.get_model('auth', 'user')
|
||||
ApiToken = self.old_state.apps.get_model('users', 'ApiToken')
|
||||
|
||||
User.objects.create(username='fred', email='fred@fred.com', password='password')
|
||||
fred = User.objects.create(
|
||||
username='fred', email='fred@fred.com', password='password'
|
||||
)
|
||||
|
||||
User.objects.create(username='brad', email='brad@fred.com', password='password')
|
||||
|
||||
ApiToken.objects.create(key='legacy-token', user=fred)
|
||||
|
||||
def test_users_exist(self):
|
||||
"""Test that users exist in the database."""
|
||||
User = self.new_state.apps.get_model('auth', 'user')
|
||||
|
||||
self.assertEqual(User.objects.count(), 2)
|
||||
|
||||
def test_existing_tokens_are_marked_as_v1(self):
|
||||
"""Test that tokens created before v2 are marked as v1."""
|
||||
ApiToken = self.new_state.apps.get_model('users', 'ApiToken')
|
||||
|
||||
token = ApiToken.objects.get(key='legacy-token')
|
||||
|
||||
self.assertEqual(token.token_version, 1)
|
||||
|
||||
# a new token should be a v2 token now
|
||||
new_token = ApiToken.objects.create(key='new-token', user=token.user)
|
||||
self.assertEqual(new_token.token_version, 2)
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
"""Unit tests for the 'users' app."""
|
||||
|
||||
import datetime
|
||||
|
||||
from django.apps import apps
|
||||
from django.contrib.auth.models import Group
|
||||
from django.test import TestCase
|
||||
@@ -298,7 +300,10 @@ class OwnerModelTest(InvenTreeTestCase):
|
||||
self.client.login(username=self.username, password=self.password)
|
||||
# token get
|
||||
response = self.do_request(reverse('api-token'), {})
|
||||
self.assertEqual(response['token'], token.first().key)
|
||||
raw_token = response['token']
|
||||
self.assertTrue(raw_token.startswith('inv-2-'))
|
||||
token = ApiToken.get_from_string(raw_token)
|
||||
self.assertTrue(token.validate(raw_token))
|
||||
|
||||
# test user is associated with token
|
||||
response = self.do_request(
|
||||
@@ -404,6 +409,12 @@ class AdminTest(AdminTestCase):
|
||||
my_token = self.helper(
|
||||
model=ApiToken, model_kwargs={'user': self.user, 'name': 'test-token'}
|
||||
)
|
||||
self.assertTrue(
|
||||
my_token.token.endswith(
|
||||
f'-{datetime.datetime.now().date().isoformat().replace("-", "")}'
|
||||
)
|
||||
)
|
||||
self.assertTrue(my_token.validate(my_token.token))
|
||||
# Additionally test str fnc
|
||||
self.assertEqual(str(my_token), my_token.token)
|
||||
|
||||
|
||||
@@ -8,7 +8,17 @@ import useTable from '@lib/hooks/UseTable';
|
||||
import type { TableFilter } from '@lib/types/Filters';
|
||||
import { t } from '@lingui/core/macro';
|
||||
import { Trans } from '@lingui/react/macro';
|
||||
import { Badge, Code, Flex, Modal, Paper, Text } from '@mantine/core';
|
||||
import {
|
||||
Badge,
|
||||
Button,
|
||||
Code,
|
||||
Flex,
|
||||
Group,
|
||||
Modal,
|
||||
Paper,
|
||||
Text,
|
||||
Textarea
|
||||
} from '@mantine/core';
|
||||
import { useDisclosure } from '@mantine/hooks';
|
||||
import { IconCircleX } from '@tabler/icons-react';
|
||||
import { useCallback, useMemo, useState } from 'react';
|
||||
@@ -27,6 +37,8 @@ export function ApiTokenTable({
|
||||
}: Readonly<{ only_myself: boolean }>) {
|
||||
const [token, setToken] = useState<string>('');
|
||||
const [opened, { open, close }] = useDisclosure(false);
|
||||
const [revokeTokenId, setRevokeTokenId] = useState<string | null>(null);
|
||||
const [revocationReason, setRevocationReason] = useState<string>('');
|
||||
|
||||
const generateToken = useCreateApiFormModal({
|
||||
url: ApiEndpoints.user_me_token,
|
||||
@@ -100,6 +112,26 @@ export function ApiTokenTable({
|
||||
accessor: 'created',
|
||||
title: t`Created`,
|
||||
sortable: true
|
||||
},
|
||||
UserColumn({
|
||||
accessor: 'issued_by_detail',
|
||||
title: t`Issued By`,
|
||||
filtering: true,
|
||||
sortable: true
|
||||
}),
|
||||
UserColumn({
|
||||
accessor: 'revoked_by_detail',
|
||||
title: t`Revoked By`,
|
||||
filtering: true,
|
||||
sortable: true
|
||||
}),
|
||||
{
|
||||
accessor: 'token_version',
|
||||
sortable: true
|
||||
},
|
||||
{
|
||||
accessor: 'revocation_reason',
|
||||
sortable: false
|
||||
}
|
||||
];
|
||||
if (!only_myself) {
|
||||
@@ -129,6 +161,14 @@ export function ApiTokenTable({
|
||||
name: 'user',
|
||||
label: t`User`,
|
||||
description: t`Filter by user`
|
||||
}),
|
||||
UserFilter({
|
||||
name: 'issued_by',
|
||||
label: t`Issued By`
|
||||
}),
|
||||
UserFilter({
|
||||
name: 'revoked_by',
|
||||
label: t`Revoked By`
|
||||
})
|
||||
);
|
||||
}
|
||||
@@ -143,21 +183,25 @@ export function ApiTokenTable({
|
||||
hidden: !record.active || record.in_use,
|
||||
icon: <IconCircleX />,
|
||||
onClick: () => {
|
||||
revokeToken(record.id);
|
||||
setRevokeTokenId(record.id);
|
||||
}
|
||||
}
|
||||
];
|
||||
}, []);
|
||||
|
||||
const revokeToken = async (id: string) => {
|
||||
let targetUrl = apiUrl(ApiEndpoints.user_tokens, id);
|
||||
const revokeToken = async () => {
|
||||
if (!revokeTokenId) return;
|
||||
|
||||
let targetUrl = apiUrl(ApiEndpoints.user_tokens, revokeTokenId);
|
||||
if (!only_myself) {
|
||||
targetUrl += '?all_users=true';
|
||||
}
|
||||
api
|
||||
.delete(targetUrl)
|
||||
.delete(targetUrl, { data: { revocation_reason: revocationReason } })
|
||||
.then(() => {
|
||||
table.refreshTable();
|
||||
setRevokeTokenId(null);
|
||||
setRevocationReason('');
|
||||
})
|
||||
.catch((error) => {
|
||||
showApiErrorMessage({
|
||||
@@ -198,6 +242,38 @@ export function ApiTokenTable({
|
||||
</Modal>
|
||||
</>
|
||||
)}
|
||||
<Modal
|
||||
opened={revokeTokenId !== null}
|
||||
onClose={() => {
|
||||
setRevokeTokenId(null);
|
||||
setRevocationReason('');
|
||||
}}
|
||||
title={t`Revoke Token`}
|
||||
centered
|
||||
>
|
||||
<Textarea
|
||||
label={t`Revocation Reason`}
|
||||
placeholder={t`Enter a reason for revoking this token`}
|
||||
value={revocationReason}
|
||||
onChange={(event) => setRevocationReason(event.currentTarget.value)}
|
||||
autosize
|
||||
minRows={3}
|
||||
/>
|
||||
<Group justify='flex-end' mt='md'>
|
||||
<Button
|
||||
variant='default'
|
||||
onClick={() => {
|
||||
setRevokeTokenId(null);
|
||||
setRevocationReason('');
|
||||
}}
|
||||
>
|
||||
{t`Cancel`}
|
||||
</Button>
|
||||
<Button color='red' onClick={revokeToken}>
|
||||
{t`Revoke`}
|
||||
</Button>
|
||||
</Group>
|
||||
</Modal>
|
||||
<InvenTreeTable
|
||||
tableState={table}
|
||||
url={apiUrl(ApiEndpoints.user_tokens)}
|
||||
|
||||
Reference in New Issue
Block a user