Coverage for netbox/settings_utils.py: 34%
114 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1"""Startup helpers for settings.py. Import-safe: no Django settings access at import time."""
3import importlib
4import importlib.util
5import os
6import sys
7import threading
8import warnings
9from typing import NamedTuple
11from django.core.exceptions import ImproperlyConfigured
12from rq.exceptions import TimeoutFormatError
13from rq.queue import Queue
14from rq.utils import parse_timeout
16__all__ = (
17 'InstallPaths',
18 'get_configuration_dir',
19 'load_configuration',
20 'load_ldap_config',
21 'parse_job_timeout',
22 'resolve_install_paths',
23 'secret_key_hint',
24 'validate_webhook_default_timeout',
25)
28class InstallPaths(NamedTuple):
29 """Filesystem layout resolved from the install mode (wheel vs. source checkout)."""
30 install_mode: str # 'wheel' or 'checkout'
31 base_dir: str # package data root (BASE_DIR)
32 netbox_root: str # instance root for mutable files (NETBOX_ROOT)
33 docs_root: str # documentation sources on a checkout, the pre-rendered site in a wheel (DOCS_ROOT default)
34 static_docs_root: str # built documentation, source of the STATICFILES 'docs' prefix
37def resolve_install_paths(settings_dir, environ):
38 """Resolve the install mode and filesystem roots for this NetBox installation.
40 A wheel bundles package data (including the pre-rendered documentation site)
41 under netbox/_data and keeps mutable instance files under an external instance root
42 (NETBOX_ROOT, default /opt/netbox); a source checkout keeps the historical layout,
43 where both roots are the project directory. All wheel-vs-checkout branching lives
44 here so settings.py stays declarative.
45 """
46 bundled_data = os.path.join(settings_dir, '_data')
47 if os.path.isdir(bundled_data): 47 ↛ 48line 47 didn't jump to line 48 because the condition on line 47 was never true
48 install_mode = 'wheel'
49 base_dir = bundled_data
50 netbox_root = os.path.abspath(environ.get('NETBOX_ROOT', '/opt/netbox'))
51 docs_root = os.path.join(base_dir, 'docs')
52 # The wheel bundles the pre-rendered documentation site at _data/docs; it serves as
53 # both the DOCS_ROOT default and the STATICFILES 'docs' prefix source.
54 static_docs_root = docs_root
55 else:
56 install_mode = 'checkout'
57 base_dir = os.path.dirname(settings_dir)
58 netbox_root = base_dir
59 docs_root = os.path.join(os.path.dirname(base_dir), 'docs')
60 static_docs_root = os.path.join(base_dir, 'project-static', 'docs')
61 return InstallPaths(
62 install_mode=install_mode,
63 base_dir=base_dir,
64 netbox_root=netbox_root,
65 docs_root=docs_root,
66 static_docs_root=static_docs_root,
67 )
70def secret_key_hint(install_mode, base_dir):
71 """Return the command to suggest in the SECRET_KEY-too-short error, based on install mode.
73 generate_secret_key.py is not packaged in a wheel, so a wheel install points at the
74 `netbox secret-key` console command instead of the (nonexistent) script path.
75 """
76 if install_mode == 'wheel':
77 return 'netbox secret-key'
78 return f'python {base_dir}/generate_secret_key.py'
81def parse_job_timeout(value):
82 """Normalize an RQ job timeout (i.e. RQ_DEFAULT_TIMEOUT) to a number of seconds.
84 RQ accepts a timeout as an integer, as a numeric string, or as a duration string such as
85 "1h", so its own parser is used to arrive at a value which can be compared against webhook
86 timeouts. A negative timeout (-1 by convention) disables RQ's death penalty; that is reported
87 as None, meaning that job execution is unbounded. An absent or zero timeout is *not* unbounded:
88 RQ falls back to the queue's own default, which is reported in its place.
89 """
90 try:
91 timeout = parse_timeout(value)
92 except (TimeoutFormatError, TypeError):
93 raise ImproperlyConfigured(
94 f"RQ_DEFAULT_TIMEOUT must be a number of seconds or a duration string such as '1h' "
95 f"(found {value!r})"
96 )
97 if timeout is None or timeout == 0: 97 ↛ 99line 97 didn't jump to line 99 because the condition on line 97 was never true
98 # Queue treats a null or zero default timeout as unset and substitutes its class default.
99 return Queue.DEFAULT_TIMEOUT
100 if timeout < 0: 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true
101 return None
102 return timeout
105def validate_webhook_default_timeout(timeout, job_timeout):
106 """Validate WEBHOOK_DEFAULT_TIMEOUT, including against the background job timeout.
108 job_timeout is the normalized RQ_DEFAULT_TIMEOUT (see parse_job_timeout()), or None if job
109 execution is unbounded. A webhook timeout which meets or exceeds the job timeout leaves no
110 room for the request's own timeout to apply, as the worker will terminate the job first.
111 """
112 if not isinstance(timeout, int) or not 1 <= timeout <= 3600: 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true
113 raise ImproperlyConfigured(
114 f"WEBHOOK_DEFAULT_TIMEOUT must be an integer between 1 and 3600 (found {timeout!r})"
115 )
116 if job_timeout is not None and timeout >= job_timeout: 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true
117 raise ImproperlyConfigured(
118 f"WEBHOOK_DEFAULT_TIMEOUT ({timeout}) must be less than RQ_DEFAULT_TIMEOUT ({job_timeout} seconds), "
119 f"which caps the total runtime of the background job."
120 )
123def _import_module(name):
124 """Import a configuration module by dotted path.
126 Preserve NetBox's historical behavior: a friendly ImproperlyConfigured when the module
127 itself is absent, but re-raise the original error when the module exists yet imports
128 something else that is missing.
129 """
130 try:
131 return importlib.import_module(name)
132 except ModuleNotFoundError as e:
133 if e.name == name:
134 raise ImproperlyConfigured(
135 f"Specified configuration module ({name}) not found. Please define "
136 f"netbox/netbox/configuration.py per the documentation, or specify an alternate "
137 f"module in the NETBOX_CONFIGURATION environment variable."
138 )
139 raise
142# Serializes cache checks, module execution, and the temporary sys.path change.
143# Reentrant because configuration code may load another path-based module.
144_import_lock = threading.RLock()
147def _import_from_path(module_name, path):
148 """Load a configuration module from an explicit file path.
150 The module is registered in sys.modules while it executes, and the file's directory is
151 placed on sys.path for that duration so the module can import siblings, matching normal
152 import semantics closely enough for configuration files. A module already loaded under the
153 same name from the same path is reused, while the same name from a different path replaces
154 it. A failed load leaves the previous entry in place. Loading is serialized so that a
155 concurrent caller cannot observe a module mid-execution.
156 """
157 path = os.path.abspath(path)
158 with _import_lock:
159 existing = sys.modules.get(module_name)
160 existing_path = getattr(existing, '__file__', None)
161 if existing_path and os.path.abspath(existing_path) == path:
162 return existing
163 module_dir = os.path.dirname(path)
164 spec = importlib.util.spec_from_file_location(module_name, path)
165 if spec is None or spec.loader is None:
166 raise ImproperlyConfigured(f"Unable to load configuration file {path}")
167 module = importlib.util.module_from_spec(spec)
168 sys.modules[module_name] = module
169 sys.path.insert(0, module_dir)
170 try:
171 spec.loader.exec_module(module)
172 except Exception:
173 if sys.modules.get(module_name) is module:
174 if existing is None:
175 del sys.modules[module_name]
176 else:
177 sys.modules[module_name] = existing
178 raise
179 finally:
180 # Remove only the entry this helper inserted at index 0.
181 if sys.path and sys.path[0] == module_dir:
182 sys.path.pop(0)
183 return module
186def get_configuration_dir(module):
187 """Return the directory containing a loaded configuration module (None if unknown)."""
188 source = getattr(module, '__file__', None)
189 return os.path.dirname(os.path.abspath(source)) if source else None
192def load_configuration(*, install_mode, install_root, environ):
193 """Import and return NetBox's configuration module.
195 An explicit NETBOX_CONFIGURATION module always wins. In wheel mode, prefer
196 <install_root>/conf/configuration.py, loaded by file path (so a stale source tree at
197 <install_root>/netbox cannot shadow it and no generic 'configuration' module is left in
198 sys.modules), then
199 fall back to the legacy <install_root>/netbox/netbox/configuration.py with a migration
200 warning. In checkout mode, keep the historical default module.
201 """
202 explicit = environ.get('NETBOX_CONFIGURATION')
203 if explicit: 203 ↛ 204line 203 didn't jump to line 204 because the condition on line 203 was never true
204 return _import_module(explicit)
206 if install_mode == 'wheel': 206 ↛ 207line 206 didn't jump to line 207 because the condition on line 206 was never true
207 conf_dir = os.path.join(install_root, 'conf')
208 preferred = os.path.join(conf_dir, 'configuration.py')
209 legacy = os.path.join(install_root, 'netbox', 'netbox', 'configuration.py')
210 if os.path.isfile(preferred):
211 if os.path.isfile(legacy):
212 warnings.warn(
213 f"Both {preferred} and the legacy {legacy} exist; using {preferred} and "
214 f"ignoring the legacy file.",
215 RuntimeWarning,
216 )
217 return _import_from_path('netbox_local_configuration', preferred)
218 if os.path.isfile(legacy):
219 warnings.warn(
220 f"Loaded NetBox configuration from the legacy source-tree path {legacy}. For a "
221 f"pip-installed NetBox, move it to {preferred}.",
222 RuntimeWarning,
223 )
224 return _import_from_path('netbox_legacy_configuration', legacy)
225 raise ImproperlyConfigured(
226 f"No NetBox configuration found. For a pip-installed NetBox, create {preferred}, "
227 f"or set NETBOX_CONFIGURATION to an importable module."
228 )
230 return _import_module('netbox.configuration')
233def load_ldap_config(config_dir, *, allow_legacy_fallback=False):
234 """Load ldap_config.py from the active configuration directory (settings.CONFIGURATION_DIR).
236 One rule for every install method: the active ldap_config.py is the one next to the
237 active configuration.py. Checkout installs may additionally allow a legacy fallback to
238 the historical netbox/netbox/ldap_config.py module, because a custom NETBOX_CONFIGURATION
239 can live outside the source tree while LDAP config stayed inside it; the fallback warns
240 so those installs can migrate to the sibling rule.
241 """
242 path = os.path.join(config_dir, 'ldap_config.py') if config_dir else None
243 if path and os.path.isfile(path):
244 return _import_from_path('netbox.ldap_config', path)
245 if allow_legacy_fallback:
246 try:
247 module = importlib.import_module('netbox.ldap_config')
248 except ModuleNotFoundError as e:
249 if e.name != 'netbox.ldap_config':
250 raise
251 else:
252 warnings.warn(
253 "Loaded LDAP configuration from the legacy netbox/netbox/ldap_config.py module. "
254 "Move ldap_config.py into the directory containing the active configuration.py; "
255 "this fallback may be removed in a future release.",
256 RuntimeWarning,
257 )
258 return module
259 if not config_dir:
260 raise ImproperlyConfigured(
261 "LDAP configuration file not found: unable to determine the directory containing "
262 "configuration.py."
263 )
264 raise ImproperlyConfigured(
265 "LDAP configuration file not found: Check that ldap_config.py has been created "
266 "alongside configuration.py. For a pip-installed NetBox, this is "
267 "NETBOX_ROOT/conf/ldap_config.py."
268 )