diff --git a/netbox_custom_objects/__init__.py b/netbox_custom_objects/__init__.py index 6ff1f3b7..96ad85b9 100644 --- a/netbox_custom_objects/__init__.py +++ b/netbox_custom_objects/__init__.py @@ -18,6 +18,11 @@ # Context variable to track if we're currently running migrations _is_migrating = contextvars.ContextVar('is_migrating', default=False) +# Guards get_models() against re-entrancy (issues #685/#686): generating a COT +# model can trigger Django to rebuild its relation graph, which calls +# apps.get_models() again while we're still mid-generation. +_generating_models = contextvars.ContextVar('generating_models', default=False) + # Cache for migration check to avoid repeated expensive filesystem/database operations _migrations_checked = None _checking_migrations = False @@ -560,52 +565,63 @@ def get_models(self, include_auto_created=False, include_swapped=False): for model in super().get_models(include_auto_created, include_swapped): yield model - # Suppress warnings about database calls during model loading. - # See the corresponding block in ready() for the rationale for using - # module-based filtering instead of message-content matching. - with warnings.catch_warnings(): - warnings.filterwarnings( - "ignore", category=RuntimeWarning, - module=r"django\.db\.backends\..*", - ) - warnings.filterwarnings( - "ignore", category=UserWarning, - module=r"netbox_branching\..*", - ) - - # Skip dynamic model generation until ready() has completed. - # Other apps' ready() calls (e.g. dcim) trigger _relation_tree → - # apps.get_models() before our ready() runs. At that point _model_cache - # is empty, so get_model() would regenerate every COT from scratch — - # including ContentType DB lookups that may fail. After our ready() - # finishes, _app_ready is True and get_model() returns cached models - # without any ContentType lookups. - if not _app_ready: - return - - # Skip custom object type model loading if dynamic models can't be created yet - if self.should_skip_dynamic_model_creation(): - return - - # Add custom object type models - from .models import CustomObjectType - - try: - with transaction.atomic(): - custom_object_types = CustomObjectType.objects.all() - for custom_type in custom_object_types: - model = custom_type.get_model() - if model: - yield model + # Re-entrant call (issues #685/#686): generate_model() registers a COT's + # model synchronously, before get_model() can trigger this recursion, so + # super().get_models() above already covers everything safely visible + # here -- fall back to that instead of regenerating. + if _generating_models.get(): + return - # If include_auto_created is True, also yield through models - if include_auto_created and hasattr(model, '_through_models'): - for through_model in model._through_models: - yield through_model - except (ProgrammingError, OperationalError): - # DB schema is incomplete (unapplied migrations). Yield nothing — - # dynamic models will be available once migrations have run. - return + token = _generating_models.set(True) + try: + # Suppress warnings about database calls during model loading. + # See the corresponding block in ready() for the rationale for using + # module-based filtering instead of message-content matching. + with warnings.catch_warnings(): + warnings.filterwarnings( + "ignore", category=RuntimeWarning, + module=r"django\.db\.backends\..*", + ) + warnings.filterwarnings( + "ignore", category=UserWarning, + module=r"netbox_branching\..*", + ) + + # Skip dynamic model generation until ready() has completed. + # Other apps' ready() calls (e.g. dcim) trigger _relation_tree → + # apps.get_models() before our ready() runs. At that point _model_cache + # is empty, so get_model() would regenerate every COT from scratch — + # including ContentType DB lookups that may fail. After our ready() + # finishes, _app_ready is True and get_model() returns cached models + # without any ContentType lookups. + if not _app_ready: + return + + # Skip custom object type model loading if dynamic models can't be created yet + if self.should_skip_dynamic_model_creation(): + return + + # Add custom object type models + from .models import CustomObjectType + + try: + with transaction.atomic(): + custom_object_types = CustomObjectType.objects.all() + for custom_type in custom_object_types: + model = custom_type.get_model() + if model: + yield model + + # If include_auto_created is True, also yield through models + if include_auto_created and hasattr(model, '_through_models'): + for through_model in model._through_models: + yield through_model + except (ProgrammingError, OperationalError): + # DB schema is incomplete (unapplied migrations). Yield nothing — + # dynamic models will be available once migrations have run. + return + finally: + _generating_models.reset(token) config = CustomObjectsPluginConfig diff --git a/netbox_custom_objects/tests/schema/test_schema_api.py b/netbox_custom_objects/tests/schema/test_schema_api.py index 78b52afa..22addd40 100644 --- a/netbox_custom_objects/tests/schema/test_schema_api.py +++ b/netbox_custom_objects/tests/schema/test_schema_api.py @@ -13,6 +13,9 @@ """ +from unittest import mock + +from django.apps import apps as django_apps from django.urls import reverse from django.test import TransactionTestCase from rest_framework import status @@ -22,6 +25,7 @@ from users.models import ObjectPermission from utilities.testing import create_test_user +import netbox_custom_objects as nco_pkg from netbox_custom_objects.schema.exporter import export_cot from netbox_custom_objects.models import CustomObjectType @@ -388,3 +392,191 @@ def test_apply_allow_destructive_string_returns_400(self): ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("allow_destructive", resp.data) + + +# --------------------------------------------------------------------------- +# get_models() re-entrancy (issue #685) +# --------------------------------------------------------------------------- + +class SchemaApplyMultiCOTRecursionTestCase(_SchemaAPIBase): + """Regression test for issue #685: applying a document that creates + multiple new, cross-referencing Custom Object Types in one request + raised RecursionError.""" + + def setUp(self): + super().setUp() + perm = ObjectPermission(name='schema_apply_recursion_cot_perm', actions=['add', 'change']) + perm.save() + perm.users.add(self.user) + perm.object_types.add(ObjectType.objects.get_for_model(CustomObjectType)) + + def test_apply_two_new_cross_referencing_cots_in_one_request(self): + django_apps.clear_cache() + + # should_skip_dynamic_model_creation() disables get_models()'s + # CustomObjectType loop under `manage.py test`; bypass it so this test + # exercises the real vulnerable code path. + app_config = django_apps.get_app_config('netbox_custom_objects') + self.enterContext(mock.patch.object(nco_pkg, '_app_ready', True)) + self.enterContext( + mock.patch.object(app_config, 'should_skip_dynamic_model_creation', return_value=False) + ) + + schema_doc = { + "schema_version": "1", + "types": [ + { + "name": "ospf_instance", + "slug": "ospf-instances", + "fields": [ + {"id": 1, "name": "name", "type": "text", "primary": True, "required": True, "unique": True}, + ], + }, + { + "name": "ospf_area", + "slug": "ospf-areas", + "fields": [ + {"id": 1, "name": "name", "type": "text", "primary": True, "required": True, "unique": True}, + { + "id": 2, + "name": "instance", + "type": "object", + "required": True, + "related_object_type": "custom-objects/ospf-instances", + }, + ], + }, + ], + } + resp = self.client.post(self.apply_url, data=self._apply_body(schema_doc), format="json") + self.assertEqual(resp.status_code, status.HTTP_200_OK, resp.content) + self.assertTrue(resp.data["applied"]) + instance_cot = CustomObjectType.objects.get(slug="ospf-instances") + area_cot = CustomObjectType.objects.get(slug="ospf-areas") + self.assertIn( + instance_cot.get_model(), + django_apps.get_models(), + "Generated model should be returned by apps.get_models().", + ) + self.assertIn( + area_cot.get_model(), + django_apps.get_models(), + "Generated model should be returned by apps.get_models().", + ) + self.assertFalse(nco_pkg._generating_models.get()) + + def test_apply_three_new_cots_chained_references_in_one_request(self): + """The issue notes a 3-type chain (interface -> area -> instance) fails + identically to the 2-type case; cover it too.""" + django_apps.clear_cache() + + app_config = django_apps.get_app_config('netbox_custom_objects') + self.enterContext(mock.patch.object(nco_pkg, '_app_ready', True)) + self.enterContext( + mock.patch.object(app_config, 'should_skip_dynamic_model_creation', return_value=False) + ) + + schema_doc = { + "schema_version": "1", + "types": [ + { + "name": "ospf_instance", + "slug": "ospf-instances", + "fields": [ + {"id": 1, "name": "name", "type": "text", "primary": True, "required": True, "unique": True}, + ], + }, + { + "name": "ospf_area", + "slug": "ospf-areas", + "fields": [ + {"id": 1, "name": "name", "type": "text", "primary": True, "required": True, "unique": True}, + { + "id": 2, + "name": "instance", + "type": "object", + "required": True, + "related_object_type": "custom-objects/ospf-instances", + }, + ], + }, + { + "name": "ospf_interface", + "slug": "ospf-interfaces", + "fields": [ + {"id": 1, "name": "name", "type": "text", "primary": True, "required": True, "unique": True}, + { + "id": 2, + "name": "area", + "type": "object", + "required": True, + "related_object_type": "custom-objects/ospf-areas", + }, + ], + }, + ], + } + resp = self.client.post(self.apply_url, data=self._apply_body(schema_doc), format="json") + self.assertEqual(resp.status_code, status.HTTP_200_OK, resp.content) + self.assertTrue(resp.data["applied"]) + instance_cot = CustomObjectType.objects.get(slug="ospf-instances") + area_cot = CustomObjectType.objects.get(slug="ospf-areas") + interface_cot = CustomObjectType.objects.get(slug="ospf-interfaces") + self.assertIn( + instance_cot.get_model(), + django_apps.get_models(), + "Generated model should be returned by apps.get_models().", + ) + self.assertIn( + area_cot.get_model(), + django_apps.get_models(), + "Generated model should be returned by apps.get_models().", + ) + self.assertIn( + interface_cot.get_model(), + django_apps.get_models(), + "Generated model should be returned by apps.get_models().", + ) + self.assertFalse(nco_pkg._generating_models.get()) + + def test_get_models_guards_against_reentrant_cot_generation(self): + """Deterministic counterpart to the two end-to-end tests above: whether + those actually blow the recursion limit depends on incidental process + state, so this simulates the re-entrant get_models() call directly and + asserts get_model() isn't invoked twice for an already-generated COT.""" + from collections import defaultdict + + cot1 = self.create_custom_object_type(name='Reentrancy Source', slug='reentrancy-source') + cot2 = self.create_custom_object_type(name='Reentrancy Target', slug='reentrancy-target') + app_config = django_apps.get_app_config('netbox_custom_objects') + + call_counts = defaultdict(int) + real_get_model = CustomObjectType.get_model + test_case = self + + def spying_get_model(self, *args, **kwargs): + call_counts[self.pk] += 1 + if call_counts[self.pk] > 10: + test_case.fail( + f"get_model() called {call_counts[self.pk]} times for COT " + f"{self.pk} -- unbounded re-entrant generation (issue #685/#686)" + ) + result = real_get_model(self, *args, **kwargs) + if call_counts[self.pk] == 1: + # Simulate Django re-entering get_models() mid-generation. + list(app_config.get_models()) + return result + + with ( + mock.patch.object(CustomObjectType, 'get_model', spying_get_model), + mock.patch.object(nco_pkg, '_app_ready', True), + mock.patch.object(app_config, 'should_skip_dynamic_model_creation', return_value=False), + ): + # Not calling django_apps.clear_cache() here -- it walks + # apps.get_models() itself, which would drive this loop once + # already before the call below even starts. + list(app_config.get_models()) + + self.assertFalse(nco_pkg._generating_models.get()) + self.assertEqual(call_counts[cot1.pk], 1) + self.assertEqual(call_counts[cot2.pk], 1) diff --git a/netbox_custom_objects/tests/test_polymorphic_fields.py b/netbox_custom_objects/tests/test_polymorphic_fields.py index e4138ce9..aa23d46b 100644 --- a/netbox_custom_objects/tests/test_polymorphic_fields.py +++ b/netbox_custom_objects/tests/test_polymorphic_fields.py @@ -2108,3 +2108,72 @@ def test_unwire_does_not_remove_descriptor_owned_by_different_field(self): Site.__dict__.get("co_shared_ref"), descriptor_b, "_unwire must not remove a descriptor owned by a different CO field", ) + + +# --------------------------------------------------------------------------- +# get_models() re-entrancy (issue #686) +# --------------------------------------------------------------------------- + +class PolymorphicReverseDescriptorRecursionTestCase( + TransactionCleanupMixin, CustomObjectsTestCase, TransactionTestCase +): + """Regression test for issue #686: generating a polymorphic field's + reverse descriptor raised RecursionError.""" + + def test_get_model_with_polymorphic_related_name_does_not_recurse(self): + from unittest import mock + + from django.apps import apps as django_apps + + import netbox_custom_objects as nco_pkg + + site_ot = ObjectType.objects.get(app_label="dcim", model="site") + prefix_ot = ObjectType.objects.get(app_label="ipam", model="prefix") + + cot = CustomObjectType.objects.create( + name="RecursionRevTest", slug="recursion-rev-test", + verbose_name_plural="Recursion Rev Tests", + ) + CustomObjectTypeField.objects.create( + custom_object_type=cot, name="name", type="text", primary=True, required=True, + ) + field = CustomObjectTypeField.objects.create( + custom_object_type=cot, + name="target_obj", label="Target", type="object", + is_polymorphic=True, + related_name="rev_recursion_test", + ) + field.related_object_types.set([site_ot, prefix_ot]) + + # Force the model out of cache and the relation-tree cold, so + # get_model() -> _after_model_generation() -> related_object_types.all() + # hits the "first access" path that triggers apps.get_models(). + cot.clear_model_cache(cot.id) + django_apps.clear_cache() + + # should_skip_dynamic_model_creation() disables get_models()'s + # CustomObjectType loop under `manage.py test`; bypass it so this test + # exercises the real vulnerable code path. + app_config = django_apps.get_app_config('netbox_custom_objects') + with ( + mock.patch.object(nco_pkg, '_app_ready', True), + mock.patch.object(app_config, 'should_skip_dynamic_model_creation', return_value=False), + ): + # Must not raise RecursionError. + model = cot.get_model() + self.assertFalse(nco_pkg._generating_models.get()) + # Look up the registered class directly rather than trusting `model`: + # a nested get_models() call triggered mid-generation (as above) can + # leave get_model()'s return value out of sync with what's actually + # registered in the app registry -- see #688. + registered_model = django_apps.get_model(APP_LABEL, model.__name__) + self.assertIn( + registered_model, + django_apps.get_models(), + "Generated model should be returned by apps.get_models().", + ) + self.assertTrue( + hasattr(Site, "rev_recursion_test"), + "Reverse descriptor must still be set on Site after get_model()", + ) + self.assertIsNotNone(model)