From 53efbe0a00f42d7df39e4b8a3fa8147c903b0553 Mon Sep 17 00:00:00 2001 From: Martin Hauser Date: Mon, 10 Aug 2026 16:04:16 +0200 Subject: [PATCH] Fixes #22805: Prevent repeated execution of LDAP configuration (#22809) --- netbox/netbox/settings_utils.py | 59 ++++++++++++++-------- netbox/netbox/tests/test_settings_utils.py | 56 ++++++++++++++++++++ 2 files changed, 94 insertions(+), 21 deletions(-) diff --git a/netbox/netbox/settings_utils.py b/netbox/netbox/settings_utils.py index c19316a3d..8f1a28c26 100644 --- a/netbox/netbox/settings_utils.py +++ b/netbox/netbox/settings_utils.py @@ -4,6 +4,7 @@ import importlib import importlib.util import os import sys +import threading import warnings from typing import NamedTuple @@ -91,32 +92,48 @@ def _import_module(name): raise +# Serializes cache checks, module execution, and the temporary sys.path change. +# Reentrant because configuration code may load another path-based module. +_import_lock = threading.RLock() + + def _import_from_path(module_name, path): """Load a configuration module from an explicit file path. - The module is registered in sys.modules (and removed again if execution fails), and the - file's directory is placed on sys.path for the duration of execution so the module can - import siblings, matching normal import semantics closely enough for configuration files. + The module is registered in sys.modules while it executes, and the file's directory is + placed on sys.path for that duration so the module can import siblings, matching normal + import semantics closely enough for configuration files. A module already loaded under the + same name from the same path is reused, while the same name from a different path replaces + it. A failed load leaves the previous entry in place. Loading is serialized so that a + concurrent caller cannot observe a module mid-execution. """ path = os.path.abspath(path) - module_dir = os.path.dirname(path) - spec = importlib.util.spec_from_file_location(module_name, path) - if spec is None or spec.loader is None: - raise ImproperlyConfigured(f"Unable to load configuration file {path}") - module = importlib.util.module_from_spec(spec) - sys.modules[module_name] = module - sys.path.insert(0, module_dir) - try: - spec.loader.exec_module(module) - except Exception: - if sys.modules.get(module_name) is module: - del sys.modules[module_name] - raise - finally: - # Remove only the entry this helper inserted at index 0. - if sys.path and sys.path[0] == module_dir: - sys.path.pop(0) - return module + with _import_lock: + existing = sys.modules.get(module_name) + existing_path = getattr(existing, '__file__', None) + if existing_path and os.path.abspath(existing_path) == path: + return existing + module_dir = os.path.dirname(path) + spec = importlib.util.spec_from_file_location(module_name, path) + if spec is None or spec.loader is None: + raise ImproperlyConfigured(f"Unable to load configuration file {path}") + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + sys.path.insert(0, module_dir) + try: + spec.loader.exec_module(module) + except Exception: + if sys.modules.get(module_name) is module: + if existing is None: + del sys.modules[module_name] + else: + sys.modules[module_name] = existing + raise + finally: + # Remove only the entry this helper inserted at index 0. + if sys.path and sys.path[0] == module_dir: + sys.path.pop(0) + return module def get_configuration_dir(module): diff --git a/netbox/netbox/tests/test_settings_utils.py b/netbox/netbox/tests/test_settings_utils.py index 573fd8eae..5870e9604 100644 --- a/netbox/netbox/tests/test_settings_utils.py +++ b/netbox/netbox/tests/test_settings_utils.py @@ -132,6 +132,52 @@ class LoadConfigurationTest(SimpleTestCase): finally: sys.path[:] = saved + def test_import_from_path_reuses_module_loaded_from_same_path(self): + """A repeated load of the same path returns the first module and runs the file only once.""" + with tempfile.TemporaryDirectory() as root: + marker = os.path.join(root, 'executions') + path = os.path.join(root, 'cached_cfg.py') + with open(path, 'w') as handle: + handle.write(f'with open({marker!r}, "a") as handle:\n handle.write("x")\n') + self.addCleanup(sys.modules.pop, 'netbox_test_cached_cfg', None) + first = settings_utils._import_from_path('netbox_test_cached_cfg', path) + second = settings_utils._import_from_path('netbox_test_cached_cfg', path) + self.assertIs(second, first) + with open(marker) as handle: + self.assertEqual(handle.read(), 'x') + + def test_import_from_path_replaces_module_loaded_from_another_path(self): + """The same module name at a different path is loaded fresh, not served from the cache.""" + with tempfile.TemporaryDirectory() as root: + first_path = os.path.join(root, 'first_cfg.py') + second_path = os.path.join(root, 'second_cfg.py') + with open(first_path, 'w') as handle: + handle.write('ALLOWED_HOSTS = ["first"]\n') + with open(second_path, 'w') as handle: + handle.write('ALLOWED_HOSTS = ["second"]\n') + self.addCleanup(sys.modules.pop, 'netbox_test_switched_cfg', None) + settings_utils._import_from_path('netbox_test_switched_cfg', first_path) + module = settings_utils._import_from_path('netbox_test_switched_cfg', second_path) + self.assertEqual(module.ALLOWED_HOSTS, ['second']) + self.assertIs(sys.modules['netbox_test_switched_cfg'], module) + + def test_import_from_path_restores_previous_module_on_failure(self): + """A failed replacement does not evict the previously loaded module.""" + with tempfile.TemporaryDirectory() as root: + module_name = 'netbox_test_restore_cfg' + first_path = os.path.join(root, 'first_cfg.py') + broken_path = os.path.join(root, 'broken_cfg.py') + with open(first_path, 'w') as handle: + handle.write('ALLOWED_HOSTS = ["first"]\n') + with open(broken_path, 'w') as handle: + handle.write('raise RuntimeError("Simulated configuration error")\n') + self.addCleanup(sys.modules.pop, module_name, None) + first = settings_utils._import_from_path(module_name, first_path) + with self.assertRaisesMessage(RuntimeError, 'Simulated configuration error'): + settings_utils._import_from_path(module_name, broken_path) + self.assertIs(sys.modules[module_name], first) + self.assertIs(settings_utils._import_from_path(module_name, first_path), first) + def test_wheel_both_configs_present_warns_and_prefers_conf(self): with tempfile.TemporaryDirectory() as root: conf = os.path.join(root, 'conf') @@ -221,6 +267,16 @@ class LoadLdapConfigTest(SimpleTestCase): self.assertEqual(module.AUTH_LDAP_SERVER_URI, 'ldaps://example') self.assertIs(sys.modules['netbox.ldap_config'], module) + def test_repeated_calls_reuse_the_sibling_module(self): + """Two calls with an unchanged sibling ldap_config.py return the same module object.""" + with tempfile.TemporaryDirectory() as conf_dir: + with open(os.path.join(conf_dir, 'ldap_config.py'), 'w') as handle: + handle.write('AUTH_LDAP_SERVER_URI = "ldaps://example"\n') + self.addCleanup(sys.modules.pop, 'netbox.ldap_config', None) + first = settings_utils.load_ldap_config(conf_dir) + second = settings_utils.load_ldap_config(conf_dir) + self.assertIs(second, first) + def test_legacy_fallback_loads_historical_module_with_warning(self): legacy = ModuleType('netbox.ldap_config') legacy.AUTH_LDAP_SERVER_URI = 'ldaps://legacy'