This commit is contained in:
parent
65454d30db
commit
d1919627ce
|
|
@ -5,6 +5,7 @@ from rest_framework.exceptions import AuthenticationFailed, PermissionDenied
|
|||
from netbox.api.fields import IPNetworkSerializer
|
||||
from netbox.api.serializers import ValidatedModelSerializer
|
||||
from users.models import Token
|
||||
from users.utils import user_may_grant_token
|
||||
|
||||
from .users import *
|
||||
|
||||
|
|
@ -50,9 +51,11 @@ class TokenSerializer(ValidatedModelSerializer):
|
|||
def validate(self, data):
|
||||
|
||||
# If the Token is being created on behalf of another user, enforce the grant_token permission.
|
||||
# Use user_may_grant_token() rather than has_perm(obj=None): the latter short-circuits to True
|
||||
# when the permission is present in cache without evaluating ObjectPermission constraints.
|
||||
request = self.context.get('request')
|
||||
token_user = data.get('user')
|
||||
if token_user and token_user != request.user and not request.user.has_perm('users.grant_token'):
|
||||
if token_user and token_user != request.user and not user_may_grant_token(request.user, token_user):
|
||||
raise PermissionDenied("This user does not have permission to create tokens for other users.")
|
||||
|
||||
return super().validate(data)
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ from netbox.registry import registry
|
|||
from users.choices import TokenVersionChoices
|
||||
from users.constants import *
|
||||
from users.models import *
|
||||
from users.utils import user_may_grant_token
|
||||
from utilities.data import flatten_dict
|
||||
from utilities.forms.fields import (
|
||||
ContentTypeMultipleChoiceField,
|
||||
|
|
@ -195,7 +196,7 @@ class TokenForm(UserTokenForm):
|
|||
if request is None:
|
||||
# Fail closed: we cannot verify that the acting user is authorized.
|
||||
raise forms.ValidationError(_("Unable to verify permission to create tokens."))
|
||||
if token_user != request.user and not request.user.has_perm('users.grant_token'):
|
||||
if token_user != request.user and not user_may_grant_token(request.user, token_user):
|
||||
raise forms.ValidationError(
|
||||
_("This user does not have permission to create tokens for other users.")
|
||||
)
|
||||
|
|
|
|||
|
|
@ -310,6 +310,109 @@ class TokenTestCase(
|
|||
response = self.client.post(url, data, format='json', **self.header)
|
||||
self.assertEqual(response.status_code, 201)
|
||||
|
||||
def test_grant_token_constrained_permission_is_enforced(self):
|
||||
"""
|
||||
Regression: SR-001 / VM-322 — constrained grant_token ObjectPermissions must not be
|
||||
bypassed. has_perm('users.grant_token', obj=None) short-circuits to True without
|
||||
evaluating constraints; the fix uses user_may_grant_token() which applies them.
|
||||
"""
|
||||
# Clear the unconstrained grant_token added by setUp.
|
||||
ObjectPermission.objects.filter(users=self.user, actions__contains=['grant']).delete()
|
||||
self.add_permissions('users.add_token')
|
||||
|
||||
superuser = User.objects.create_user(username='sec_superuser', is_superuser=True)
|
||||
regular = User.objects.create_user(username='sec_regular')
|
||||
|
||||
# Add a *constrained* grant_token permission: only tokens for non-superusers.
|
||||
token_ct = ObjectType.objects.get_by_natural_key('users', 'token')
|
||||
perm = ObjectPermission(
|
||||
name='constrained_grant_token',
|
||||
constraints={'user__is_superuser': False},
|
||||
actions=['grant'],
|
||||
)
|
||||
perm.save()
|
||||
perm.users.add(self.user)
|
||||
perm.object_types.add(token_ct)
|
||||
|
||||
url = reverse('users-api:token-list')
|
||||
|
||||
# Attempt to create a token for the superuser — must be denied (constraint violation).
|
||||
response = self.client.post(url, {'user': superuser.pk}, format='json', **self.header)
|
||||
self.assertEqual(
|
||||
response.status_code, 403,
|
||||
"Constrained grant_token must deny token creation for users that violate the constraint",
|
||||
)
|
||||
|
||||
# Attempt to create a token for the regular user — must succeed.
|
||||
response = self.client.post(url, {'user': regular.pk}, format='json', **self.header)
|
||||
self.assertEqual(
|
||||
response.status_code, 201,
|
||||
"Constrained grant_token must allow token creation for users that satisfy the constraint",
|
||||
)
|
||||
|
||||
def test_grant_token_superuser_always_allowed(self):
|
||||
"""
|
||||
Superusers must be able to create tokens for any user without an explicit
|
||||
grant_token ObjectPermission. Regression guard: _user_may_grant_token() must
|
||||
mirror ObjectPermissionMixin.has_perm's superuser bypass.
|
||||
"""
|
||||
ObjectPermission.objects.filter(users=self.user, actions__contains=['grant']).delete()
|
||||
self.add_permissions('users.add_token')
|
||||
self.user.is_superuser = True
|
||||
self.user.save()
|
||||
try:
|
||||
other = User.objects.create_user(username='superuser_grant_target')
|
||||
url = reverse('users-api:token-list')
|
||||
response = self.client.post(url, {'user': other.pk}, format='json', **self.header)
|
||||
self.assertEqual(response.status_code, 201, "Superuser must be able to grant tokens for any user")
|
||||
finally:
|
||||
self.user.is_superuser = False
|
||||
self.user.save()
|
||||
|
||||
def test_grant_token_self_only_constraint(self):
|
||||
"""
|
||||
A {"user": "$user"} constraint means "only grant tokens for yourself".
|
||||
Since creating a token for oneself bypasses the grant_token check entirely,
|
||||
this constraint effectively blocks all cross-user grants for the holder.
|
||||
Exercises the $user placeholder substitution path.
|
||||
"""
|
||||
ObjectPermission.objects.filter(users=self.user, actions__contains=['grant']).delete()
|
||||
self.add_permissions('users.add_token')
|
||||
token_ct = ObjectType.objects.get_by_natural_key('users', 'token')
|
||||
perm = ObjectPermission(name='self_only_grant', constraints={'user': '$user'}, actions=['grant'])
|
||||
perm.save()
|
||||
perm.users.add(self.user)
|
||||
perm.object_types.add(token_ct)
|
||||
|
||||
other = User.objects.create_user(username='self_only_target')
|
||||
url = reverse('users-api:token-list')
|
||||
response = self.client.post(url, {'user': other.pk}, format='json', **self.header)
|
||||
self.assertEqual(response.status_code, 403, "Self-only constraint must deny cross-user token grants")
|
||||
|
||||
def test_grant_token_non_user_field_constraint_fails_closed(self):
|
||||
"""
|
||||
A constraint referencing a non-user Token field (e.g. {"write_enabled": True})
|
||||
cannot be evaluated for an unsaved token; _user_may_grant_token() must fail
|
||||
closed and deny rather than bypass the constraint.
|
||||
"""
|
||||
ObjectPermission.objects.filter(users=self.user, actions__contains=['grant']).delete()
|
||||
self.add_permissions('users.add_token')
|
||||
token_ct = ObjectType.objects.get_by_natural_key('users', 'token')
|
||||
perm = ObjectPermission(
|
||||
name='non_user_field_grant', constraints={'write_enabled': True}, actions=['grant']
|
||||
)
|
||||
perm.save()
|
||||
perm.users.add(self.user)
|
||||
perm.object_types.add(token_ct)
|
||||
|
||||
other = User.objects.create_user(username='non_user_field_target')
|
||||
url = reverse('users-api:token-list')
|
||||
response = self.client.post(url, {'user': other.pk}, format='json', **self.header)
|
||||
self.assertEqual(
|
||||
response.status_code, 403,
|
||||
"Non-user Token field constraints must fail closed for new (unsaved) tokens",
|
||||
)
|
||||
|
||||
def test_create_token_returns_plaintext(self):
|
||||
"""
|
||||
Creating a Token via the REST API must return the usable plaintext value in the response.
|
||||
|
|
|
|||
|
|
@ -463,6 +463,78 @@ class TokenGrantPermissionTestCase(TestCase):
|
|||
self.assertEqual(token.user, owner)
|
||||
self.assertEqual(token.description, 'updated')
|
||||
|
||||
def _add_constrained_grant_permission(self, constraints):
|
||||
"""Add a constrained grant_token ObjectPermission to self.user."""
|
||||
token_ct = ObjectType.objects.get_by_natural_key('users', 'token')
|
||||
perm = ObjectPermission(
|
||||
name='constrained_grant_token',
|
||||
constraints=constraints,
|
||||
actions=['grant'],
|
||||
)
|
||||
perm.save()
|
||||
perm.users.add(self.user)
|
||||
perm.object_types.add(token_ct)
|
||||
|
||||
def test_create_token_via_form_constrained_grant_is_enforced(self):
|
||||
"""
|
||||
Regression: SR-001 / VM-322 — constrained grant_token ObjectPermissions must not
|
||||
be bypassed via the UI create form. has_perm(obj=None) short-circuits to True
|
||||
without evaluating constraints; user_may_grant_token() applies them correctly.
|
||||
"""
|
||||
superuser = User.objects.create_user(username='form_constrained_super', is_superuser=True)
|
||||
regular = User.objects.create_user(username='form_constrained_regular')
|
||||
self._add_constrained_grant_permission({'user__is_superuser': False})
|
||||
|
||||
# Constrained grant must deny token creation for a superuser.
|
||||
response = self.client.post(reverse('users:token_add'), data={
|
||||
'version': 2,
|
||||
'user': superuser.pk,
|
||||
'description': 'constrained denied',
|
||||
'enabled': 'on',
|
||||
})
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertFalse(Token.objects.filter(user=superuser).exists())
|
||||
|
||||
# Constrained grant must allow token creation for a non-superuser.
|
||||
response = self.client.post(reverse('users:token_add'), data={
|
||||
'version': 2,
|
||||
'user': regular.pk,
|
||||
'description': 'constrained allowed',
|
||||
'enabled': 'on',
|
||||
})
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertTrue(Token.objects.filter(description='constrained allowed', user=regular).exists())
|
||||
|
||||
def test_bulk_import_token_constrained_grant_is_enforced(self):
|
||||
"""
|
||||
Regression: SR-001 / VM-322 — constrained grant_token ObjectPermissions must not
|
||||
be bypassed via bulk CSV import. has_perm(obj=None) short-circuits to True
|
||||
without evaluating constraints; user_may_grant_token() applies them correctly.
|
||||
"""
|
||||
superuser = User.objects.create_user(username='bulk_constrained_super', is_superuser=True)
|
||||
regular = User.objects.create_user(username='bulk_constrained_regular')
|
||||
self._add_constrained_grant_permission({'user__is_superuser': False})
|
||||
|
||||
# Constrained grant must deny bulk import for a superuser.
|
||||
csv_data = '\n'.join(('user,description', f"{superuser.pk},bulk denied"))
|
||||
response = self.client.post(reverse('users:token_bulk_import'), data={
|
||||
'data': csv_data,
|
||||
'format': ImportFormatChoices.CSV,
|
||||
'csv_delimiter': CSVDelimiterChoices.AUTO,
|
||||
})
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertFalse(Token.objects.filter(user=superuser).exists())
|
||||
|
||||
# Constrained grant must allow bulk import for a non-superuser.
|
||||
csv_data = '\n'.join(('user,description', f"{regular.pk},bulk allowed"))
|
||||
response = self.client.post(reverse('users:token_bulk_import'), data={
|
||||
'data': csv_data,
|
||||
'format': ImportFormatChoices.CSV,
|
||||
'csv_delimiter': CSVDelimiterChoices.AUTO,
|
||||
})
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertTrue(Token.objects.filter(description='bulk allowed', user=regular).exists())
|
||||
|
||||
|
||||
class OwnerGroupTestCase(ViewTestCases.AdminModelViewTestCase):
|
||||
model = OwnerGroup
|
||||
|
|
|
|||
|
|
@ -1,12 +1,73 @@
|
|||
from django.conf import settings
|
||||
from django.db.models import Q
|
||||
from social_core.storage import NO_ASCII_REGEX, NO_SPECIAL_REGEX
|
||||
|
||||
__all__ = (
|
||||
'clean_username',
|
||||
'get_current_pepper',
|
||||
'user_may_grant_token',
|
||||
)
|
||||
|
||||
|
||||
def user_may_grant_token(requesting_user, token_user):
|
||||
"""
|
||||
Return True if *requesting_user* has permission to create a token for *token_user*,
|
||||
respecting ObjectPermission constraints on the users.grant_token permission.
|
||||
|
||||
``has_perm('users.grant_token', obj=None)`` always short-circuits to True when the
|
||||
permission is present in the cache (obj=None bypasses constraint evaluation in
|
||||
ObjectPermissionMixin.has_perm). Since the new token does not yet exist in the
|
||||
database we cannot pass an existing Token as obj. Instead we extract the raw
|
||||
constraint list, remap Token-field paths to User-field paths, and evaluate them
|
||||
directly against the target User record — the only variable in a new token creation.
|
||||
|
||||
Field remapping rules:
|
||||
``user__<field>`` → ``<field>`` (FK traversal into User)
|
||||
``user`` → ``pk`` (Token.user FK becomes User.pk)
|
||||
|
||||
Constraints referencing other Token fields cannot be evaluated for an unsaved
|
||||
token; the function returns False (deny) for those.
|
||||
"""
|
||||
from users.models import User
|
||||
|
||||
# Mirrors ObjectPermissionMixin.has_perm: superusers implicitly have all permissions.
|
||||
if requesting_user.is_active and requesting_user.is_superuser:
|
||||
return True
|
||||
|
||||
perm = 'users.grant_token'
|
||||
|
||||
# get_all_permissions() populates _object_perm_cache as a side effect.
|
||||
# Guard against missing cache key in case a non-standard backend is in use.
|
||||
if perm not in requesting_user.get_all_permissions():
|
||||
return False
|
||||
|
||||
constraints = getattr(requesting_user, '_object_perm_cache', {}).get(perm, [])
|
||||
|
||||
# An empty/null constraint means "no restriction" — allow any target.
|
||||
if any(not c for c in constraints):
|
||||
return True
|
||||
|
||||
# Substitute the $user token so {"user": "$user"} resolves to the requesting user.
|
||||
resolved_user_id = requesting_user.pk
|
||||
|
||||
q = Q()
|
||||
for constraint in constraints:
|
||||
user_constraint = {}
|
||||
for key, raw_val in constraint.items():
|
||||
val = resolved_user_id if raw_val == '$user' else raw_val
|
||||
if key == 'user':
|
||||
user_constraint['pk'] = val
|
||||
elif key.startswith('user__'):
|
||||
user_constraint[key.removeprefix('user__')] = val
|
||||
else:
|
||||
# Non-user Token field — cannot evaluate for a new (unsaved) token.
|
||||
# Fail closed.
|
||||
return False
|
||||
q |= Q(**user_constraint)
|
||||
|
||||
return User.objects.filter(q, pk=token_user.pk).exists()
|
||||
|
||||
|
||||
def clean_username(value):
|
||||
"""Clean username removing any unsupported character"""
|
||||
value = NO_ASCII_REGEX.sub('', value)
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ from netbox.ui.panels import (
|
|||
)
|
||||
from netbox.views import generic
|
||||
from users.ui import panels
|
||||
from users.utils import user_may_grant_token
|
||||
from utilities.query import count_related
|
||||
from utilities.views import GetRelatedModelsMixin, register_model_view
|
||||
|
||||
|
|
@ -86,7 +87,7 @@ class TokenBulkImportView(generic.BulkImportView):
|
|||
# creation; TokenImportForm disables the user field on update, so an existing Token's owner cannot change.
|
||||
token_user = object_form.cleaned_data.get('user')
|
||||
if object_form.instance._state.adding and token_user and token_user != request.user \
|
||||
and not request.user.has_perm('users.grant_token'):
|
||||
and not user_may_grant_token(request.user, token_user):
|
||||
raise ValidationError(_("This user does not have permission to create tokens for other users."))
|
||||
return super().save_object(object_form, request)
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue