diff --git a/netbox/netbox/graphql/pagination.py b/netbox/netbox/graphql/pagination.py index 53c159b12..abc62ac6b 100644 --- a/netbox/netbox/graphql/pagination.py +++ b/netbox/netbox/graphql/pagination.py @@ -2,6 +2,8 @@ import strawberry from strawberry.types.unset import UNSET from strawberry_django.pagination import _QS, apply +from netbox.config import get_config + __all__ = ( 'OffsetPaginationInfo', 'OffsetPaginationInput', @@ -47,4 +49,14 @@ def apply_pagination( # Ignore `offset` when `start` is set pagination.offset = 0 + # Enforce MAX_PAGE_SIZE on the pagination limit + max_page_size = get_config().MAX_PAGE_SIZE + if max_page_size: + if pagination is None: + pagination = OffsetPaginationInput(limit=max_page_size) + elif pagination.limit in (None, UNSET) or pagination.limit > max_page_size: + pagination.limit = max_page_size + elif pagination.limit <= 0: + pagination.limit = max_page_size + return apply(pagination, queryset, related_field_id=related_field_id) diff --git a/netbox/netbox/tests/test_graphql.py b/netbox/netbox/tests/test_graphql.py index cdf5cf3b5..3914a9493 100644 --- a/netbox/netbox/tests/test_graphql.py +++ b/netbox/netbox/tests/test_graphql.py @@ -283,6 +283,53 @@ class GraphQLAPITestCase(APITestCase): self.assertEqual(len(data['data']['site_list']), 1) self.assertEqual(data['data']['site_list'][0]['name'], 'Site 7') + @override_settings(MAX_PAGE_SIZE=3) + def test_max_page_size(self): + self.add_permissions('dcim.view_site') + url = reverse('graphql') + + # Request without explicit limit should be capped by MAX_PAGE_SIZE + query = """ + { + site_list { + id name + } + } + """ + response = self.client.post(url, data={'query': query}, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_200_OK) + data = json.loads(response.content) + self.assertNotIn('errors', data) + self.assertEqual(len(data['data']['site_list']), 3) + + # Request with limit exceeding MAX_PAGE_SIZE should be capped + query = """ + { + site_list(pagination: {limit: 100}) { + id name + } + } + """ + response = self.client.post(url, data={'query': query}, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_200_OK) + data = json.loads(response.content) + self.assertNotIn('errors', data) + self.assertEqual(len(data['data']['site_list']), 3) + + # Request with limit under MAX_PAGE_SIZE should be respected + query = """ + { + site_list(pagination: {limit: 2}) { + id name + } + } + """ + response = self.client.post(url, data={'query': query}, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_200_OK) + data = json.loads(response.content) + self.assertNotIn('errors', data) + self.assertEqual(len(data['data']['site_list']), 2) + def test_pagination_conflict(self): url = reverse('graphql') query = """