Merge pull request #23374 from netbox-community/23371-avoid-deep-copying-cached-config-context-during-api

Closes #23371: Avoid deep-copying cached Config Context in REST and GraphQL responses

netbox/dcim/api/serializers_/devices.py

@@ -137,7 +137,7 @@ class DeviceSerializer(PrimaryModelSerializer):
 
     @extend_schema_field(serializers.JSONField(allow_null=True))
     def get_config_context(self, obj):
-        return obj.get_config_context()
+        return obj.get_config_context(copy_data=False)
 
 
 class VirtualDeviceContextSerializer(PrimaryModelSerializer):

netbox/dcim/tests/test_api.py

@@ -1,3 +1,4 @@
+import copy
 import json
 import os
 import tempfile
@@ -2782,6 +2783,29 @@ class DeviceTestCase(Mixins.FieldPrefetchMixin, APIViewTestCases.APIViewTestCase
             filters.append(f'site_id={site.pk}')
         return [filters[0], '&'.join(filters)]
 
+    def test_config_context_read_without_copy(self):
+        """
+        REST and GraphQL read cached config context without copying it.
+        """
+        self.add_permissions('dcim.view_device')
+        Device.objects.filter(name='Device 1').update(_config_context_data={'A': 'cached'})
+        Device.objects.filter(name='Device 2').update(_config_context_data={})
+        # Device 3 has no cache and renders its local context data
+        expected = {'Device 1': {'A': 'cached'}, 'Device 2': {}, 'Device 3': {'C': 3}}
+        query = '{ device_list { name config_context } }'
+
+        with patch('extras.models.configs.copy', wraps=copy) as copy_module:
+            rest = self.client.get(self._get_list_url(), **self.header)
+            graphql = self.client.post(reverse('graphql'), data={'query': query}, format='json', **self.header)
+        copy_module.deepcopy.assert_not_called()
+
+        self.assertHttpStatus(rest, status.HTTP_200_OK)
+        self.assertEqual({row['name']: row['config_context'] for row in rest.data['results']}, expected)
+        self.assertHttpStatus(graphql, status.HTTP_200_OK)
+        data = graphql.json()
+        self.assertNotIn('errors', data)
+        self.assertEqual({row['name']: row['config_context'] for row in data['data']['device_list']}, expected)
+
     def test_config_context_included_by_default_in_list_view(self):
         """
         Check that config context data is included by default in the devices list.
@@ -3016,6 +3040,31 @@ class DeviceTestCase(Mixins.FieldPrefetchMixin, APIViewTestCases.APIViewTestCase
         self.assertHttpStatus(response, status.HTTP_200_OK)
         self.assertEqual(response.data['content'], f'Config for device {device.name}')
 
+    def test_render_config_leaves_config_context_cache_unchanged(self):
+        """
+        A config template which mutates nested context data leaves the cached config context unchanged.
+        """
+        configtemplate = ConfigTemplate.objects.create(
+            name='Config Template 1',
+            template_code="{{ servers.append('192.0.2.2') or '' }}{{ servers|join(',') }}"
+        )
+        device = Device.objects.first()
+        device.config_template = configtemplate
+        device.save()
+        Device.objects.filter(pk=device.pk).update(_config_context_data={'servers': ['192.0.2.1']})
+
+        self.add_permissions('dcim.render_config_device', 'dcim.view_device')
+        url = reverse('dcim-api:device-render-config', kwargs={'pk': device.pk})
+        with patch.object(
+            Device, 'get_config_context', autospec=True, side_effect=Device.get_config_context
+        ) as get_config_context:
+            response = self.client.post(url, {}, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertEqual(response.data['content'], '192.0.2.1,192.0.2.2')
+        # A leak would only show on the view's in-memory instance
+        instance = get_config_context.call_args.args[0]
+        self.assertEqual(instance._config_context_data, {'servers': ['192.0.2.1']})
+
     def test_render_config_without_permission(self):
         configtemplate = ConfigTemplate.objects.create(
             name='Config Template 1',

netbox/dcim/tests/test_views.py

@@ -3,6 +3,7 @@ import datetime
 import json
 from decimal import Decimal
 from io import StringIO
+from unittest.mock import patch
 from urllib.parse import quote
 from zoneinfo import ZoneInfo
 
@@ -3260,6 +3261,31 @@ class DeviceTestCase(ViewTestCases.PrimaryObjectViewTestCase):
         self.assertHttpStatus(response, 200)
         self.assertIn(b'Error rendering template', response.content)
 
+    def test_device_renderconfig_leaves_config_context_cache_unchanged(self):
+        """
+        A config template which mutates nested context data leaves the cached config context unchanged.
+        """
+        configtemplate = ConfigTemplate.objects.create(
+            name='Test Config Template',
+            template_code="{{ servers.append('192.0.2.2') or '' }}{{ servers|join(',') }}"
+        )
+        device = Device.objects.first()
+        device.config_template = configtemplate
+        device.save()
+        Device.objects.filter(pk=device.pk).update(_config_context_data={'servers': ['192.0.2.1']})
+
+        self.add_permissions('dcim.view_device', 'dcim.render_config_device')
+        url = reverse('dcim:device_render-config', kwargs={'pk': device.pk})
+        with patch.object(
+            Device, 'get_config_context', autospec=True, side_effect=Device.get_config_context
+        ) as get_config_context:
+            response = self.client.get(url)
+        self.assertHttpStatus(response, 200)
+        self.assertIn(b'192.0.2.1,192.0.2.2', response.content)
+        # A leak would only show on the view's in-memory instance
+        instance = get_config_context.call_args.args[0]
+        self.assertEqual(instance._config_context_data, {'servers': ['192.0.2.1']})
+
     def test_device_configcontext_is_not_cacheable(self):
         """
         The config context tab renders the merged context data, which may contain sensitive

netbox/extras/graphql/mixins.py

@@ -44,7 +44,7 @@ class ConfigContextMixin:
     # is requested, so the warm-cache read path requires no additional queries.
     @strawberry_django.field(only=['_config_context_data', 'local_context_data'])
     def config_context(self) -> strawberry.scalars.JSON:
-        return self.get_config_context()
+        return self.get_config_context(copy_data=False)
 
 
 @strawberry.type

netbox/extras/models/configs.py

@@ -362,19 +362,21 @@ class ConfigContextModel(models.Model):
     class Meta:
         abstract = True
 
-    def get_config_context(self):
+    def get_config_context(self, *, copy_data=True):
         """
         Return the merged config context for this object. If a pre-rendered cache is present
-        (`_config_context_data`), return a copy of it. Otherwise, fall back to rendering on demand.
+        (`_config_context_data`), return a copy of it unless copy_data is False. Otherwise, fall
+        back to rendering on demand.
 
-        The returned dict is always safe for callers to mutate (e.g. ObjectRenderConfigView merges
+        By default, the returned dict is safe for callers to mutate (e.g. ObjectRenderConfigView merges
         in additional context with .update()): the cached blob is deep-copied so mutations cannot
         leak back into this instance's in-memory cache, matching the fresh-dict guarantee of the
-        on-demand render path.
+        on-demand render path. Read-only callers may pass copy_data=False to skip the copy. The
+        cached blob is then returned as is and must not be modified.
         """
         cached = getattr(self, '_config_context_data', None)
         if cached is not None:
-            return copy.deepcopy(cached)
+            return copy.deepcopy(cached) if copy_data else cached
         return self.render_config_context()
 
     def render_config_context(self):

netbox/extras/tests/test_configcontext_cache.py

@@ -54,11 +54,30 @@ class ConfigContextCacheReadPathTest(TestCase):
         device = Device.objects.get(pk=self.device.pk)
         self.assertEqual(device.get_config_context(), cached)
 
+    def test_cached_value_is_copied(self):
+        """
+        Mutating the returned context leaves the cache unchanged.
+        """
+        _set_cache(self.device, {'servers': ['192.0.2.1']})
+        device = Device.objects.get(pk=self.device.pk)
+        device.get_config_context()['servers'].append('192.0.2.2')
+        self.assertEqual(device._config_context_data, {'servers': ['192.0.2.1']})
+
+    def test_cached_value_is_returned_without_copy(self):
+        """
+        copy_data=False returns the cache itself, including an empty one.
+        """
+        for cached in ({'cached': True}, {}):
+            _set_cache(self.device, cached)
+            device = Device.objects.get(pk=self.device.pk)
+            self.assertIs(device.get_config_context(copy_data=False), device._config_context_data)
+
     def test_null_cache_falls_back_to_render(self):
         ConfigContext.objects.create(name='CC', weight=100, data={'rendered': True})
         device = Device.objects.get(pk=self.device.pk)
         self.assertIsNone(device._config_context_data)
         self.assertEqual(device.get_config_context(), {'rendered': True})
+        self.assertEqual(device.get_config_context(copy_data=False), {'rendered': True})
 
     def test_render_matches_legacy_path(self):
         ConfigContext.objects.create(name='A', weight=100, data={'a': 1})

netbox/virtualization/api/serializers_/virtualmachines.py

@@ -97,7 +97,7 @@ class VirtualMachineSerializer(PrimaryModelSerializer):
 
     @extend_schema_field(serializers.JSONField(allow_null=True))
     def get_config_context(self, obj):
-        return obj.get_config_context()
+        return obj.get_config_context(copy_data=False)
 
 
 #

netbox/virtualization/tests/test_api.py

@@ -1,5 +1,7 @@
+import copy
 import logging
 from decimal import Decimal
+from unittest.mock import patch
 
 from django.db import connection
 from django.db.models.signals import post_init
@@ -459,6 +461,31 @@ class VirtualMachineTestCase(APIViewTestCases.APIViewTestCase):
         self.assertIsNone(vm.vcpus)
         self.assertIsNone(vm.memory)
 
+    def test_config_context_read_without_copy(self):
+        """
+        REST and GraphQL read cached config context without copying it.
+        """
+        self.add_permissions('virtualization.view_virtualmachine')
+        VirtualMachine.objects.filter(name='Virtual Machine 1').update(_config_context_data={'A': 'cached'})
+        VirtualMachine.objects.filter(name='Virtual Machine 2').update(_config_context_data={})
+        # Virtual Machine 3 has no cache and renders its local context data
+        expected = {'Virtual Machine 1': {'A': 'cached'}, 'Virtual Machine 2': {}, 'Virtual Machine 3': {'C': 3}}
+        query = '{ virtual_machine_list { name config_context } }'
+
+        with patch('extras.models.configs.copy', wraps=copy) as copy_module:
+            rest = self.client.get(self._get_list_url(), **self.header)
+            graphql = self.client.post(reverse('graphql'), data={'query': query}, format='json', **self.header)
+        copy_module.deepcopy.assert_not_called()
+
+        self.assertHttpStatus(rest, status.HTTP_200_OK)
+        self.assertEqual({row['name']: row['config_context'] for row in rest.data['results']}, expected)
+        self.assertHttpStatus(graphql, status.HTTP_200_OK)
+        data = graphql.json()
+        self.assertNotIn('errors', data)
+        self.assertEqual(
+            {row['name']: row['config_context'] for row in data['data']['virtual_machine_list']}, expected
+        )
+
     def test_config_context_included_by_default_in_list_view(self):
         """
         Check that config context data is included by default in the virtual machines list.