Fixes #21988: Enforce view permissions when referencing related object by attributes in REST API
This commit is contained in:
parent
6c501413ee
commit
e713b4fd07
|
|
@ -168,6 +168,9 @@ Or by a set of attributes which uniquely identify the rack:
|
|||
|
||||
Note that if the provided parameters do not return exactly one object, a validation error is raised.
|
||||
|
||||
!!! note "Permissions"
|
||||
When a related object is referenced by a set of attributes, the lookup is restricted to only those objects which the requesting user has permission to view. This prevents the enumeration of objects by their attributes. Referencing a related object directly by its numeric ID is always permitted, regardless of the user's view permissions for that object.
|
||||
|
||||
### Generic Relations
|
||||
|
||||
Some objects within NetBox have attributes which can reference an object of multiple types, known as _generic relations_. For example, an IP address can be assigned to either a device interface _or_ a virtual machine interface. When making this assignment via the REST API, we must specify two attributes:
|
||||
|
|
|
|||
|
|
@ -449,7 +449,7 @@ class ChangeLogAPITestCase(APITestCase):
|
|||
}
|
||||
self.assertEqual(ObjectChange.objects.count(), 0)
|
||||
url = reverse('dcim-api:site-list')
|
||||
self.add_permissions('dcim.add_site')
|
||||
self.add_permissions('dcim.add_site', 'extras.view_tag')
|
||||
|
||||
response = self.client.post(url, data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
||||
|
|
@ -481,7 +481,7 @@ class ChangeLogAPITestCase(APITestCase):
|
|||
]
|
||||
}
|
||||
self.assertEqual(ObjectChange.objects.count(), 0)
|
||||
self.add_permissions('dcim.change_site')
|
||||
self.add_permissions('dcim.change_site', 'extras.view_tag')
|
||||
url = reverse('dcim-api:site-detail', kwargs={'pk': site.pk})
|
||||
|
||||
response = self.client.put(url, data, format='json', **self.header)
|
||||
|
|
|
|||
|
|
@ -1431,7 +1431,7 @@ class CustomFieldAPITestCase(APITestCase):
|
|||
site1 = Site.objects.get(name='Site 1')
|
||||
vlans = VLAN.objects.all()[:3]
|
||||
url = reverse('dcim-api:site-detail', kwargs={'pk': site1.pk})
|
||||
self.add_permissions('dcim.change_site')
|
||||
self.add_permissions('dcim.change_site', 'ipam.view_vlan')
|
||||
|
||||
# Set related objects by PK
|
||||
data = {
|
||||
|
|
|
|||
|
|
@ -149,7 +149,7 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
|
|||
]
|
||||
}
|
||||
url = reverse('dcim-api:site-list')
|
||||
self.add_permissions('dcim.add_site')
|
||||
self.add_permissions('dcim.add_site', 'extras.view_tag')
|
||||
response = self.client.post(url, data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
||||
self.assertEqual(Site.objects.count(), 1)
|
||||
|
|
@ -200,7 +200,7 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
|
|||
},
|
||||
]
|
||||
url = reverse('dcim-api:site-list')
|
||||
self.add_permissions('dcim.add_site')
|
||||
self.add_permissions('dcim.add_site', 'extras.view_tag')
|
||||
response = self.client.post(url, data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
||||
self.assertEqual(Site.objects.count(), 3)
|
||||
|
|
@ -234,7 +234,7 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
|
|||
]
|
||||
}
|
||||
url = reverse('dcim-api:site-detail', kwargs={'pk': site.pk})
|
||||
self.add_permissions('dcim.change_site')
|
||||
self.add_permissions('dcim.change_site', 'extras.view_tag')
|
||||
response = self.client.patch(url, data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
|
||||
|
|
@ -291,7 +291,7 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
|
|||
},
|
||||
]
|
||||
url = reverse('dcim-api:site-list')
|
||||
self.add_permissions('dcim.change_site')
|
||||
self.add_permissions('dcim.change_site', 'extras.view_tag')
|
||||
response = self.client.patch(url, data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
|
||||
|
|
|
|||
|
|
@ -50,7 +50,7 @@ class TaggedItemTestCase(APITestCase):
|
|||
{"name": "New Tag"},
|
||||
]
|
||||
}
|
||||
self.add_permissions('dcim.change_site')
|
||||
self.add_permissions('dcim.change_site', 'extras.view_tag')
|
||||
url = reverse('dcim-api:site-detail', kwargs={'pk': site.pk})
|
||||
|
||||
response = self.client.patch(url, data, format='json', **self.header)
|
||||
|
|
|
|||
|
|
@ -50,7 +50,9 @@ class BaseModelSerializer(serializers.ModelSerializer):
|
|||
# identifying a related object.
|
||||
if self.nested:
|
||||
queryset = self.Meta.model.objects.all()
|
||||
return get_related_object_by_attrs(queryset, data)
|
||||
request = self.context.get('request')
|
||||
user = request.user if request else None
|
||||
return get_related_object_by_attrs(queryset, data, user=user)
|
||||
|
||||
return super().to_internal_value(data)
|
||||
|
||||
|
|
|
|||
|
|
@ -17,7 +17,9 @@ class WritableNestedSerializer(BaseModelSerializer):
|
|||
"""
|
||||
def to_internal_value(self, data):
|
||||
queryset = self.Meta.model.objects.all()
|
||||
return get_related_object_by_attrs(queryset, data)
|
||||
request = self.context.get('request')
|
||||
user = request.user if request else None
|
||||
return get_related_object_by_attrs(queryset, data, user=user)
|
||||
|
||||
|
||||
# Declared here for use by PrimaryModelSerializer
|
||||
|
|
|
|||
|
|
@ -218,16 +218,31 @@ def get_annotations_for_serializer(serializer_class, fields=None, omit=None):
|
|||
return annotations
|
||||
|
||||
|
||||
def get_related_object_by_attrs(queryset, attrs):
|
||||
def get_related_object_by_attrs(queryset, attrs, user=None):
|
||||
"""
|
||||
Return an object identified by either a dictionary of attributes or its numeric primary key (ID). This is used
|
||||
for referencing related objects when creating/updating objects via the REST API.
|
||||
|
||||
When a dictionary of attributes is provided, the queryset is first restricted to only those objects on which the
|
||||
given user has been granted view permission. This prevents an unprivileged user from enumerating objects by their
|
||||
attributes. Referencing an object directly by its numeric ID is always permitted, regardless of the user's view
|
||||
permissions.
|
||||
|
||||
:param queryset: The base queryset from which to retrieve the related object
|
||||
:param attrs: A dictionary of attributes or a numeric primary key identifying the related object
|
||||
:param user: The user making the request (used to enforce view permissions on attribute-based lookups)
|
||||
"""
|
||||
if attrs is None:
|
||||
return None
|
||||
|
||||
# Dictionary of related object attributes
|
||||
if isinstance(attrs, dict):
|
||||
# Restrict the queryset to only those objects the user is permitted to view. This ensures that filtering by
|
||||
# attributes cannot be used to enumerate objects which the user is not otherwise permitted to see. Referencing
|
||||
# an object solely by its numeric ID (e.g. {"id": 123}) is equivalent to passing the ID directly, and is
|
||||
# always permitted regardless of the user's view permissions.
|
||||
if list(attrs) != ['id'] and user is not None and hasattr(queryset, 'restrict'):
|
||||
queryset = queryset.restrict(user, 'view')
|
||||
params = dict_to_filter_params(attrs)
|
||||
try:
|
||||
return queryset.get(**params)
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ from netbox.api.serializers import BaseModelSerializer
|
|||
from netbox.config import get_config
|
||||
from netbox.plugins import register_serializer_resolver
|
||||
from netbox.registry import registry
|
||||
from users.models import ObjectPermission
|
||||
from utilities.api import get_prefetches_for_serializer, get_serializer_for_model, get_view_name
|
||||
from utilities.testing import APITestCase, disable_warnings
|
||||
|
||||
|
|
@ -70,7 +71,7 @@ class WritableNestedSerializerTestCase(APITestCase):
|
|||
},
|
||||
}
|
||||
url = reverse('ipam-api:vlan-list')
|
||||
self.add_permissions('ipam.add_vlan')
|
||||
self.add_permissions('ipam.add_vlan', 'dcim.view_site')
|
||||
|
||||
response = self.client.post(url, data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
||||
|
|
@ -87,7 +88,7 @@ class WritableNestedSerializerTestCase(APITestCase):
|
|||
},
|
||||
}
|
||||
url = reverse('ipam-api:vlan-list')
|
||||
self.add_permissions('ipam.add_vlan')
|
||||
self.add_permissions('ipam.add_vlan', 'dcim.view_site')
|
||||
|
||||
with disable_warnings('django.request'):
|
||||
response = self.client.post(url, data, format='json', **self.header)
|
||||
|
|
@ -106,7 +107,7 @@ class WritableNestedSerializerTestCase(APITestCase):
|
|||
},
|
||||
}
|
||||
url = reverse('ipam-api:vlan-list')
|
||||
self.add_permissions('ipam.add_vlan')
|
||||
self.add_permissions('ipam.add_vlan', 'dcim.view_site')
|
||||
|
||||
with disable_warnings('django.request'):
|
||||
response = self.client.post(url, data, format='json', **self.header)
|
||||
|
|
@ -114,6 +115,100 @@ class WritableNestedSerializerTestCase(APITestCase):
|
|||
self.assertEqual(VLAN.objects.count(), 0)
|
||||
self.assertTrue(response.data['site'][0].startswith("Multiple objects match"))
|
||||
|
||||
def test_related_by_pk_without_view_permission(self):
|
||||
"""
|
||||
Referencing a related object by its numeric ID must be permitted even if the user has not been granted
|
||||
permission to view the object.
|
||||
"""
|
||||
data = {
|
||||
'vid': 100,
|
||||
'name': 'Test VLAN 100',
|
||||
'site': self.site1.pk,
|
||||
}
|
||||
url = reverse('ipam-api:vlan-list')
|
||||
self.add_permissions('ipam.add_vlan')
|
||||
|
||||
response = self.client.post(url, data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
||||
self.assertEqual(response.data['site']['id'], self.site1.pk)
|
||||
vlan = VLAN.objects.get(pk=response.data['id'])
|
||||
self.assertEqual(vlan.site, self.site1)
|
||||
|
||||
def test_related_by_id_attribute_without_view_permission(self):
|
||||
"""
|
||||
Referencing a related object by a dictionary containing only its numeric ID is equivalent to referencing it
|
||||
by ID directly, and must be permitted even without view permission.
|
||||
"""
|
||||
data = {
|
||||
'vid': 100,
|
||||
'name': 'Test VLAN 100',
|
||||
'site': {
|
||||
'id': self.site1.pk
|
||||
},
|
||||
}
|
||||
url = reverse('ipam-api:vlan-list')
|
||||
self.add_permissions('ipam.add_vlan')
|
||||
|
||||
response = self.client.post(url, data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
||||
self.assertEqual(response.data['site']['id'], self.site1.pk)
|
||||
vlan = VLAN.objects.get(pk=response.data['id'])
|
||||
self.assertEqual(vlan.site, self.site1)
|
||||
|
||||
def test_related_by_attributes_without_view_permission(self):
|
||||
"""
|
||||
Referencing a related object by a dictionary of attributes must enforce the user's view permissions,
|
||||
preventing enumeration of objects the user is not permitted to see.
|
||||
"""
|
||||
data = {
|
||||
'vid': 100,
|
||||
'name': 'Test VLAN 100',
|
||||
'site': {
|
||||
'name': 'Site 1'
|
||||
},
|
||||
}
|
||||
url = reverse('ipam-api:vlan-list')
|
||||
self.add_permissions('ipam.add_vlan')
|
||||
|
||||
with disable_warnings('django.request'):
|
||||
response = self.client.post(url, data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
||||
self.assertEqual(VLAN.objects.count(), 0)
|
||||
self.assertTrue(response.data['site'][0].startswith("Related object not found"))
|
||||
|
||||
def test_related_by_attributes_constrained_view_permission(self):
|
||||
"""
|
||||
When a user's view permission is constrained, only objects matching the constraint may be referenced by
|
||||
attributes.
|
||||
"""
|
||||
data = {
|
||||
'vid': 100,
|
||||
'name': 'Test VLAN 100',
|
||||
'site': {
|
||||
'name': 'Site 2'
|
||||
},
|
||||
}
|
||||
url = reverse('ipam-api:vlan-list')
|
||||
# Grant view permission only for Site 1
|
||||
self.add_permissions('ipam.add_vlan')
|
||||
obj_perm = ObjectPermission(name='Constrained view', constraints={'name': 'Site 1'}, actions=['view'])
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(Site))
|
||||
|
||||
# Referencing Site 2 by attributes must fail, as the user cannot view it
|
||||
with disable_warnings('django.request'):
|
||||
response = self.client.post(url, data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
||||
self.assertEqual(VLAN.objects.count(), 0)
|
||||
self.assertTrue(response.data['site'][0].startswith("Related object not found"))
|
||||
|
||||
# Referencing Site 1 by attributes must succeed
|
||||
data['site'] = {'name': 'Site 1'}
|
||||
response = self.client.post(url, data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
||||
self.assertEqual(response.data['site']['id'], self.site1.pk)
|
||||
|
||||
def test_related_by_invalid(self):
|
||||
data = {
|
||||
'vid': 100,
|
||||
|
|
|
|||
Loading…
Reference in New Issue