- Annotate the `info` parameter in SharedObjectMixin.get_queryset() with the Info type for consistency with BaseObjectType.get_queryset() - Extend the SavedFilter and TableConfig visibility tests to assert that the owning user can still retrieve their own private object via both the REST detail endpoint and GraphQL
This commit is contained in:
parent
d9ccee6ef2
commit
6edb5ec8b7
|
|
@ -15,9 +15,19 @@ __all__ = (
|
|||
'ConfigContextQuerySetMixin',
|
||||
'ConfigTemplateRenderMixin',
|
||||
'RenderConfigMixin',
|
||||
'SharedObjectQuerySetMixin',
|
||||
)
|
||||
|
||||
|
||||
class SharedObjectQuerySetMixin:
|
||||
"""
|
||||
Restrict the queryset to shared objects, or those owned by the current user, unless the user is a superuser.
|
||||
This mirrors the visibility enforced in the UI by extras.utils.SharedObjectViewMixin.
|
||||
"""
|
||||
def get_queryset(self):
|
||||
return super().get_queryset().restrict_to_shared(self.request.user)
|
||||
|
||||
|
||||
class ConfigContextQuerySetMixin:
|
||||
"""
|
||||
Used by views that work with config context models (device and virtual machine).
|
||||
|
|
|
|||
|
|
@ -26,7 +26,7 @@ from utilities.request import copy_safe_request
|
|||
from utilities.rqworker import any_workers_for_queue
|
||||
|
||||
from . import serializers
|
||||
from .mixins import ConfigTemplateRenderMixin
|
||||
from .mixins import ConfigTemplateRenderMixin, SharedObjectQuerySetMixin
|
||||
|
||||
|
||||
class ExtrasRootView(APIRootView):
|
||||
|
|
@ -125,7 +125,7 @@ class ExportTemplateViewSet(SyncedDataMixin, NetBoxModelViewSet):
|
|||
# Saved filters
|
||||
#
|
||||
|
||||
class SavedFilterViewSet(NetBoxModelViewSet):
|
||||
class SavedFilterViewSet(SharedObjectQuerySetMixin, NetBoxModelViewSet):
|
||||
metadata_class = ContentTypeMetadata
|
||||
queryset = SavedFilter.objects.all()
|
||||
serializer_class = serializers.SavedFilterSerializer
|
||||
|
|
@ -136,7 +136,7 @@ class SavedFilterViewSet(NetBoxModelViewSet):
|
|||
# Table Configs
|
||||
#
|
||||
|
||||
class TableConfigViewSet(NetBoxModelViewSet):
|
||||
class TableConfigViewSet(SharedObjectQuerySetMixin, NetBoxModelViewSet):
|
||||
metadata_class = ContentTypeMetadata
|
||||
queryset = TableConfig.objects.all()
|
||||
serializer_class = serializers.TableConfigSerializer
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ from typing import TYPE_CHECKING, Annotated
|
|||
import strawberry
|
||||
import strawberry_django
|
||||
from strawberry.scalars import JSON
|
||||
from strawberry.types import Info
|
||||
|
||||
from core.graphql.mixins import SyncedDataMixin
|
||||
from extras import models
|
||||
|
|
@ -48,6 +49,17 @@ __all__ = (
|
|||
)
|
||||
|
||||
|
||||
class SharedObjectMixin:
|
||||
"""
|
||||
Restrict the queryset to shared objects, or those owned by the current user, unless the user is a superuser.
|
||||
This mirrors the visibility enforced in the UI (extras.utils.SharedObjectViewMixin) and the REST API.
|
||||
"""
|
||||
@classmethod
|
||||
def get_queryset(cls, queryset, info: Info, **kwargs):
|
||||
queryset = super().get_queryset(queryset, info, **kwargs)
|
||||
return queryset.restrict_to_shared(info.context.request.user)
|
||||
|
||||
|
||||
@strawberry_django.type(
|
||||
models.ConfigContextProfile,
|
||||
fields='__all__',
|
||||
|
|
@ -184,7 +196,7 @@ class NotificationGroupType(ObjectType):
|
|||
filters=SavedFilterFilter,
|
||||
pagination=True
|
||||
)
|
||||
class SavedFilterType(OwnerMixin, ObjectType):
|
||||
class SavedFilterType(SharedObjectMixin, OwnerMixin, ObjectType):
|
||||
user: Annotated["UserType", strawberry.lazy('users.graphql.types')] | None
|
||||
|
||||
|
||||
|
|
@ -203,7 +215,7 @@ class SubscriptionType(ObjectType):
|
|||
filters=TableConfigFilter,
|
||||
pagination=True
|
||||
)
|
||||
class TableConfigType(ObjectType):
|
||||
class TableConfigType(SharedObjectMixin, ObjectType):
|
||||
object_type: Annotated["ContentTypeType", strawberry.lazy('netbox.graphql.types')] | None
|
||||
user: Annotated["UserType", strawberry.lazy('users.graphql.types')] | None
|
||||
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ from extras.choices import *
|
|||
from extras.conditions import ConditionSet, InvalidCondition
|
||||
from extras.constants import *
|
||||
from extras.models.mixins import RenderTemplateMixin
|
||||
from extras.querysets import SharedObjectQuerySet
|
||||
from extras.utils import image_upload
|
||||
from netbox.config import get_config
|
||||
from netbox.events import get_event_type_choices
|
||||
|
|
@ -519,6 +520,8 @@ class SavedFilter(CloningMixin, ExportTemplatesMixin, OwnerMixin, ChangeLoggedMo
|
|||
verbose_name=_('parameters')
|
||||
)
|
||||
|
||||
objects = SharedObjectQuerySet.as_manager()
|
||||
|
||||
clone_fields = (
|
||||
'object_types', 'weight', 'enabled', 'parameters',
|
||||
)
|
||||
|
|
@ -606,6 +609,8 @@ class TableConfig(CloningMixin, ChangeLoggedModel):
|
|||
null=True,
|
||||
)
|
||||
|
||||
objects = SharedObjectQuerySet.as_manager()
|
||||
|
||||
clone_fields = ('object_type', 'table', 'enabled', 'shared', 'columns', 'ordering')
|
||||
|
||||
class Meta:
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ __all__ = (
|
|||
'ConfigContextModelQuerySet',
|
||||
'ConfigContextQuerySet',
|
||||
'NotificationQuerySet',
|
||||
'SharedObjectQuerySet',
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -190,3 +191,20 @@ class NotificationQuerySet(RestrictedQuerySet):
|
|||
Return only unread notifications.
|
||||
"""
|
||||
return self.filter(read__isnull=True)
|
||||
|
||||
|
||||
class SharedObjectQuerySet(RestrictedQuerySet):
|
||||
|
||||
def restrict_to_shared(self, user):
|
||||
"""
|
||||
Restrict the queryset to objects which are shared or owned by the given user. Superusers are exempt;
|
||||
anonymous users see only shared objects. This enforces consistent visibility across the UI, REST API,
|
||||
and GraphQL API.
|
||||
"""
|
||||
if user.is_superuser:
|
||||
return self
|
||||
if user.is_anonymous:
|
||||
return self.filter(shared=True)
|
||||
return self.filter(
|
||||
Q(shared=True) | Q(user=user)
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
import datetime
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
|
|
@ -18,7 +19,7 @@ from extras.models import *
|
|||
from extras.scripts import BooleanVar, IntegerVar, StringVar
|
||||
from extras.scripts import Script as PythonClass
|
||||
from users.constants import TOKEN_PREFIX
|
||||
from users.models import Group, Token, User
|
||||
from users.models import Group, ObjectPermission, Token, User
|
||||
from utilities.tables import get_table_for_model
|
||||
from utilities.testing import APITestCase, APIViewTestCases
|
||||
|
||||
|
|
@ -444,7 +445,24 @@ class CustomLinkTestCase(APIViewTestCases.APIViewTestCase):
|
|||
custom_link.object_types.set([site_type])
|
||||
|
||||
|
||||
class SavedFilterTestCase(APIViewTestCases.APIViewTestCase):
|
||||
class SharedObjectAPITestMixin:
|
||||
"""
|
||||
Helpers for testing the shared/owner visibility enforced on SavedFilter and TableConfig.
|
||||
"""
|
||||
def _grant_view_permission_and_authenticate(self, user, model):
|
||||
"""
|
||||
Grant `user` an unconstrained view permission on `model`, create an API token, and return the
|
||||
corresponding authentication header.
|
||||
"""
|
||||
obj_perm = ObjectPermission(name=f'{model._meta.model_name} view', actions=['view'])
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(model))
|
||||
token = Token.objects.create(user=user)
|
||||
return {'HTTP_AUTHORIZATION': f'Bearer {TOKEN_PREFIX}{token.key}.{token.token}'}
|
||||
|
||||
|
||||
class SavedFilterTestCase(SharedObjectAPITestMixin, APIViewTestCases.APIViewTestCase):
|
||||
model = SavedFilter
|
||||
brief_fields = ['description', 'display', 'id', 'name', 'slug', 'url']
|
||||
create_data = [
|
||||
|
|
@ -516,8 +534,55 @@ class SavedFilterTestCase(APIViewTestCases.APIViewTestCase):
|
|||
for i, savedfilter in enumerate(saved_filters):
|
||||
savedfilter.object_types.set([site_type])
|
||||
|
||||
def test_private_filter_not_visible_to_other_users(self):
|
||||
"""
|
||||
A private (shared=False) SavedFilter owned by another user must not be exposed via the REST API, even to
|
||||
a user holding an unconstrained view permission.
|
||||
"""
|
||||
site_type = ObjectType.objects.get_for_model(Site)
|
||||
owner = User.objects.create_user(username='filter-owner')
|
||||
private_filter = SavedFilter.objects.create(
|
||||
name='Private Filter',
|
||||
slug='private-filter',
|
||||
user=owner,
|
||||
shared=False,
|
||||
parameters={'status': ['active']},
|
||||
)
|
||||
private_filter.object_types.set([site_type])
|
||||
|
||||
class TableConfigTestCase(APIViewTestCases.APIViewTestCase):
|
||||
# Grant an unconstrained view permission (the common case)
|
||||
self.add_permissions('extras.view_savedfilter')
|
||||
|
||||
# The private filter must not appear in the list
|
||||
response = self.client.get(self._get_list_url(), **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
returned_ids = [obj['id'] for obj in response.data['results']]
|
||||
self.assertNotIn(private_filter.pk, returned_ids)
|
||||
|
||||
# The private filter must not be retrievable directly
|
||||
response = self.client.get(self._get_detail_url(private_filter), **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)
|
||||
|
||||
# The private filter must not be exposed via GraphQL either
|
||||
query = '{ saved_filter_list { id } }'
|
||||
response = self.client.post(reverse('graphql'), data={'query': query}, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
data = json.loads(response.content)
|
||||
returned_ids = [int(obj['id']) for obj in data['data']['saved_filter_list']]
|
||||
self.assertNotIn(private_filter.pk, returned_ids)
|
||||
|
||||
# The owner, however, must still be able to access their own private filter
|
||||
owner_header = self._grant_view_permission_and_authenticate(owner, SavedFilter)
|
||||
response = self.client.get(self._get_detail_url(private_filter), **owner_header)
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
response = self.client.post(reverse('graphql'), data={'query': query}, format='json', **owner_header)
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
data = json.loads(response.content)
|
||||
returned_ids = [int(obj['id']) for obj in data['data']['saved_filter_list']]
|
||||
self.assertIn(private_filter.pk, returned_ids)
|
||||
|
||||
|
||||
class TableConfigTestCase(SharedObjectAPITestMixin, APIViewTestCases.APIViewTestCase):
|
||||
model = TableConfig
|
||||
brief_fields = ['description', 'display', 'id', 'name', 'object_type', 'table', 'url']
|
||||
bulk_update_data = {
|
||||
|
|
@ -545,6 +610,7 @@ class TableConfigTestCase(APIViewTestCases.APIViewTestCase):
|
|||
object_type=site_type,
|
||||
table=site_table_name,
|
||||
user=users[0],
|
||||
shared=True,
|
||||
columns=['name', 'status'],
|
||||
),
|
||||
TableConfig(
|
||||
|
|
@ -552,6 +618,7 @@ class TableConfigTestCase(APIViewTestCases.APIViewTestCase):
|
|||
object_type=site_type,
|
||||
table=site_table_name,
|
||||
user=users[1],
|
||||
shared=True,
|
||||
columns=['name', 'region'],
|
||||
),
|
||||
TableConfig(
|
||||
|
|
@ -559,6 +626,7 @@ class TableConfigTestCase(APIViewTestCases.APIViewTestCase):
|
|||
object_type=site_type,
|
||||
table=site_table_name,
|
||||
user=users[2],
|
||||
shared=True,
|
||||
columns=['name', 'tenant'],
|
||||
),
|
||||
)
|
||||
|
|
@ -587,6 +655,54 @@ class TableConfigTestCase(APIViewTestCases.APIViewTestCase):
|
|||
},
|
||||
]
|
||||
|
||||
def test_private_table_config_not_visible_to_other_users(self):
|
||||
"""
|
||||
A private (shared=False) TableConfig owned by another user must not be exposed via the REST API, even to
|
||||
a user holding an unconstrained view permission.
|
||||
"""
|
||||
site_type = ObjectType.objects.get_for_model(Site)
|
||||
site_table_name = get_table_for_model(Site).__name__
|
||||
owner = User.objects.create_user(username='tableconfig-owner')
|
||||
private_config = TableConfig.objects.create(
|
||||
name='Private Table Config',
|
||||
object_type=site_type,
|
||||
table=site_table_name,
|
||||
user=owner,
|
||||
shared=False,
|
||||
columns=['name', 'status'],
|
||||
)
|
||||
|
||||
# Grant an unconstrained view permission (the common case)
|
||||
self.add_permissions('extras.view_tableconfig')
|
||||
|
||||
# The private table config must not appear in the list
|
||||
response = self.client.get(self._get_list_url(), **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
returned_ids = [obj['id'] for obj in response.data['results']]
|
||||
self.assertNotIn(private_config.pk, returned_ids)
|
||||
|
||||
# The private table config must not be retrievable directly
|
||||
response = self.client.get(self._get_detail_url(private_config), **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)
|
||||
|
||||
# The private table config must not be exposed via GraphQL either
|
||||
query = '{ table_config_list { id } }'
|
||||
response = self.client.post(reverse('graphql'), data={'query': query}, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
data = json.loads(response.content)
|
||||
returned_ids = [int(obj['id']) for obj in data['data']['table_config_list']]
|
||||
self.assertNotIn(private_config.pk, returned_ids)
|
||||
|
||||
# The owner, however, must still be able to access their own private table config
|
||||
owner_header = self._grant_view_permission_and_authenticate(owner, TableConfig)
|
||||
response = self.client.get(self._get_detail_url(private_config), **owner_header)
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
response = self.client.post(reverse('graphql'), data={'query': query}, format='json', **owner_header)
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
data = json.loads(response.content)
|
||||
returned_ids = [int(obj['id']) for obj in data['data']['table_config_list']]
|
||||
self.assertIn(private_config.pk, returned_ids)
|
||||
|
||||
|
||||
class BookmarkTestCase(
|
||||
APIViewTestCases.GetObjectViewTestCase,
|
||||
|
|
|
|||
|
|
@ -6,7 +6,6 @@ from django.core.exceptions import ImproperlyConfigured, SuspiciousFileOperation
|
|||
from django.core.files.storage import Storage, default_storage
|
||||
from django.core.files.utils import validate_file_name
|
||||
from django.db import models
|
||||
from django.db.models import Q
|
||||
from taggit.managers import _TaggableManager
|
||||
|
||||
from netbox.context import current_request
|
||||
|
|
@ -32,14 +31,7 @@ class SharedObjectViewMixin:
|
|||
"""
|
||||
Return only shared objects, or those owned by the current user, unless this is a superuser.
|
||||
"""
|
||||
queryset = super().get_queryset(request)
|
||||
if request.user.is_superuser:
|
||||
return queryset
|
||||
if request.user.is_anonymous:
|
||||
return queryset.filter(shared=True)
|
||||
return queryset.filter(
|
||||
Q(shared=True) | Q(user=request.user)
|
||||
)
|
||||
return super().get_queryset(request).restrict_to_shared(request.user)
|
||||
|
||||
|
||||
def filename_from_model(model: models.Model) -> str:
|
||||
|
|
|
|||
Loading…
Reference in New Issue