Coverage for extras/models/configs.py: 47%
205 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import copy
2import os
3import re
4import sys
5import traceback
7import jsonschema
8from django.conf import settings
9from django.core.validators import ValidationError
10from django.db import models
11from django.db.models import Q
12from django.urls import reverse
13from django.utils.translation import gettext_lazy as _
14from jinja2.exceptions import TemplateError
15from jsonschema.exceptions import ValidationError as JSONValidationError
17from extras.models.mixins import RenderTemplateMixin
18from extras.querysets import ConfigContextQuerySet
19from netbox.models import ChangeLoggedModel, PrimaryModel
20from netbox.models.features import CloningMixin, CustomLinksMixin, ExportTemplatesMixin, SyncedDataMixin, TagsMixin
21from netbox.models.mixins import OwnerMixin
22from utilities.data import deepmerge
23from utilities.jsonschema import validate_schema
25__all__ = (
26 'ConfigContext',
27 'ConfigContextModel',
28 'ConfigContextProfile',
29 'ConfigTemplate',
30)
33#
34# Config contexts
35#
37class ConfigContextProfile(SyncedDataMixin, PrimaryModel):
38 """
39 A profile which can be used to enforce parameters on a ConfigContext.
40 """
41 name = models.CharField(
42 verbose_name=_('name'),
43 max_length=100,
44 unique=True
45 )
46 description = models.CharField(
47 verbose_name=_('description'),
48 max_length=200,
49 blank=True
50 )
51 schema = models.JSONField(
52 blank=True,
53 null=True,
54 validators=[validate_schema],
55 verbose_name=_('schema'),
56 help_text=_('A JSON schema specifying the structure of the context data for this profile')
57 )
59 clone_fields = ('schema',)
61 class Meta:
62 ordering = ('name',)
63 verbose_name = _('config context profile')
64 verbose_name_plural = _('config context profiles')
66 def __str__(self):
67 return self.name
69 def sync_data(self):
70 """
71 Synchronize schema from the designated DataFile (if any).
72 """
73 self.schema = self.validate_synced_value('schema', self.data_file.get_data())
74 sync_data.alters_data = True
77class ConfigContext(SyncedDataMixin, CloningMixin, CustomLinksMixin, OwnerMixin, ChangeLoggedModel):
78 """
79 A ConfigContext represents a set of arbitrary data available to any Device or VirtualMachine matching its assigned
80 qualifiers (region, site, etc.). For example, the data stored in a ConfigContext assigned to site A and tenant B
81 will be available to a Device in site A assigned to tenant B. Data is stored in JSON format.
82 """
83 name = models.CharField(
84 verbose_name=_('name'),
85 max_length=100,
86 unique=True
87 )
88 profile = models.ForeignKey(
89 to='extras.ConfigContextProfile',
90 on_delete=models.PROTECT,
91 blank=True,
92 null=True,
93 related_name='config_contexts',
94 )
95 weight = models.PositiveSmallIntegerField(
96 verbose_name=_('weight'),
97 default=1000
98 )
99 description = models.CharField(
100 verbose_name=_('description'),
101 max_length=200,
102 blank=True
103 )
104 is_active = models.BooleanField(
105 verbose_name=_('is active'),
106 default=True,
107 )
108 regions = models.ManyToManyField(
109 to='dcim.Region',
110 related_name='+',
111 blank=True
112 )
113 site_groups = models.ManyToManyField(
114 to='dcim.SiteGroup',
115 related_name='+',
116 blank=True
117 )
118 sites = models.ManyToManyField(
119 to='dcim.Site',
120 related_name='+',
121 blank=True
122 )
123 locations = models.ManyToManyField(
124 to='dcim.Location',
125 related_name='+',
126 blank=True
127 )
128 device_types = models.ManyToManyField(
129 to='dcim.DeviceType',
130 related_name='+',
131 blank=True
132 )
133 roles = models.ManyToManyField(
134 to='dcim.DeviceRole',
135 related_name='+',
136 blank=True
137 )
138 platforms = models.ManyToManyField(
139 to='dcim.Platform',
140 related_name='+',
141 blank=True
142 )
143 cluster_types = models.ManyToManyField(
144 to='virtualization.ClusterType',
145 related_name='+',
146 blank=True
147 )
148 cluster_groups = models.ManyToManyField(
149 to='virtualization.ClusterGroup',
150 related_name='+',
151 blank=True
152 )
153 clusters = models.ManyToManyField(
154 to='virtualization.Cluster',
155 related_name='+',
156 blank=True
157 )
158 tenant_groups = models.ManyToManyField(
159 to='tenancy.TenantGroup',
160 related_name='+',
161 blank=True
162 )
163 tenants = models.ManyToManyField(
164 to='tenancy.Tenant',
165 related_name='+',
166 blank=True
167 )
168 tags = models.ManyToManyField(
169 to='extras.Tag',
170 related_name='+',
171 blank=True
172 )
173 data = models.JSONField()
175 objects = ConfigContextQuerySet.as_manager()
177 clone_fields = (
178 'weight', 'profile', 'is_active', 'regions', 'site_groups', 'sites', 'locations', 'device_types', 'roles',
179 'platforms', 'cluster_types', 'cluster_groups', 'clusters', 'tenant_groups', 'tenants', 'tags', 'data',
180 )
182 class Meta:
183 ordering = ['weight', 'name']
184 indexes = (
185 models.Index(fields=('weight', 'name')), # Default ordering
186 )
187 verbose_name = _('config context')
188 verbose_name_plural = _('config contexts')
190 def __str__(self):
191 return self.name
193 def get_absolute_url(self):
194 return reverse('extras:configcontext', kwargs={'pk': self.pk})
196 @property
197 def docs_url(self):
198 return f'{settings.STATIC_URL}docs/models/extras/configcontext/'
200 def clean(self):
201 super().clean()
203 # Verify that JSON data is provided as an object
204 if type(self.data) is not dict:
205 raise ValidationError(
206 {'data': _('JSON data must be in object form. Example:') + ' {"foo": 123}'}
207 )
209 # Validate config data against the assigned profile's schema (if any)
210 if self.profile and self.profile.schema:
211 try:
212 jsonschema.validate(self.data, schema=self.profile.schema)
213 except JSONValidationError as e:
214 raise ValidationError(_("Data does not conform to profile schema: {error}").format(error=e))
216 def sync_data(self):
217 """
218 Synchronize context data from the designated DataFile (if any).
219 """
220 self.data = self.validate_synced_value('data', self.data_file.get_data())
221 sync_data.alters_data = True
223 def get_affected_objects(self, using=None):
224 """
225 Return a (device_qs, vm_qs) tuple of all Devices and VirtualMachines that fall within this
226 ConfigContext's scope. This is the inverse of ConfigContextQuerySet.get_for_object().
227 Used to determine which pre-rendered context caches must be invalidated when this
228 ConfigContext changes.
230 `using` pins every query (both the scope lookups and the returned querysets) to the given
231 database alias; None defers to the router, as an unpinned query would.
232 """
233 from dcim.models import Device
234 from virtualization.models import VirtualMachine
236 device_q, vm_q = self._get_affected_object_filters(using=using)
237 return (
238 Device.objects.using(using).filter(device_q),
239 VirtualMachine.objects.using(using).filter(vm_q),
240 )
242 def _get_affected_object_filters(self, using=None):
243 """
244 Build the Q expressions matching Devices and VirtualMachines in this context's scope.
245 Returns (device_q, vm_q). Does NOT consider `is_active` — callers that need that should
246 check it separately. For invalidation purposes, we want the scope set regardless of
247 whether the context is currently active (toggling is_active also requires invalidation).
248 `using` pins the scope lookups to the given database alias.
249 """
250 from extras.models.tags import TaggedItem
252 def _nested_scope_q(m2m, object_path):
253 # Match objects whose `object_path` ltree column is a descendant-or-equal of any node
254 # selected in this nested-group m2m (regions, locations, etc.). This is the inverse of
255 # the forward `<object>__path__ancestor_or_equal` match in ConfigContextQuerySet: there
256 # a CC's node must be an ancestor of the object's node; here the object's node must fall
257 # within a CC node's subtree. Returns None if the m2m is empty (no scope restriction).
258 paths = list(m2m.using(using).values_list('path', flat=True))
259 if not paths:
260 return None
261 q = Q()
262 for path in paths:
263 q |= Q(**{f'{object_path}__descendant_or_equal': path})
264 return q
266 def _direct_pks(m2m):
267 pks = list(m2m.using(using).values_list('pk', flat=True))
268 return pks or None
270 # Shared filters (applicable to both Device and VirtualMachine)
271 shared = Q()
273 region_q = _nested_scope_q(self.regions, 'site__region__path')
274 if region_q is not None:
275 shared &= region_q
277 site_group_q = _nested_scope_q(self.site_groups, 'site__group__path')
278 if site_group_q is not None:
279 shared &= site_group_q
281 role_q = _nested_scope_q(self.roles, 'role__path')
282 if role_q is not None:
283 shared &= role_q
285 platform_q = _nested_scope_q(self.platforms, 'platform__path')
286 if platform_q is not None:
287 shared &= platform_q
289 for m2m, path in (
290 (self.sites, 'site'),
291 (self.cluster_types, 'cluster__type'),
292 (self.cluster_groups, 'cluster__group'),
293 (self.clusters, 'cluster'),
294 (self.tenant_groups, 'tenant__group'),
295 (self.tenants, 'tenant'),
296 ):
297 pks = _direct_pks(m2m)
298 if pks is not None:
299 shared &= Q(**{f'{path}__in': pks})
301 # Tag-scoped contexts: object must be tagged with at least one of the context's tags
302 tag_pks = _direct_pks(self.tags)
304 device_q = Q(shared)
305 vm_q = Q(shared)
307 # Device-only filters: location (nested/ltree) and device_type (direct)
308 location_q = _nested_scope_q(self.locations, 'location__path')
309 if location_q is not None:
310 device_q &= location_q
311 device_type_pks = _direct_pks(self.device_types)
312 if device_type_pks is not None:
313 device_q &= Q(device_type__in=device_type_pks)
314 # For VMs, locations and device_types must be empty for the context to apply
315 if location_q is not None or device_type_pks is not None:
316 vm_q &= Q(pk__in=())
318 if tag_pks is not None:
319 device_tagged = TaggedItem.objects.using(using).filter(
320 tag_id__in=tag_pks,
321 content_type__app_label='dcim',
322 content_type__model='device',
323 ).values_list('object_id', flat=True)
324 vm_tagged = TaggedItem.objects.using(using).filter(
325 tag_id__in=tag_pks,
326 content_type__app_label='virtualization',
327 content_type__model='virtualmachine',
328 ).values_list('object_id', flat=True)
329 device_q &= Q(pk__in=device_tagged)
330 vm_q &= Q(pk__in=vm_tagged)
332 return device_q, vm_q
335class ConfigContextModel(models.Model):
336 """
337 A model which includes local configuration context data. This local data will override any inherited data from
338 ConfigContexts.
339 """
340 # Pre-rendered config context cache. NULL means "invalidated; render on demand". Populated by
341 # extras.jobs.RenderConfigContextJob in the background.
342 _config_context_data = models.JSONField(
343 blank=True,
344 null=True,
345 editable=False,
346 )
347 # Monotonic counter bumped each time the cache is invalidated. The background renderer captures
348 # this value before rendering and only writes the result back if it is unchanged, so a fresh
349 # invalidation that lands mid-render is never overwritten by a stale value (compare-and-set).
350 _config_context_generation = models.PositiveBigIntegerField(
351 default=0,
352 editable=False,
353 )
354 local_context_data = models.JSONField(
355 blank=True,
356 null=True,
357 help_text=_(
358 "Local config context data takes precedence over source contexts in the final rendered config context"
359 )
360 )
362 class Meta:
363 abstract = True
365 def get_config_context(self):
366 """
367 Return the merged config context for this object. If a pre-rendered cache is present
368 (`_config_context_data`), return a copy of it. Otherwise, fall back to rendering on demand.
370 The returned dict is always safe for callers to mutate (e.g. ObjectRenderConfigView merges
371 in additional context with .update()): the cached blob is deep-copied so mutations cannot
372 leak back into this instance's in-memory cache, matching the fresh-dict guarantee of the
373 on-demand render path.
374 """
375 cached = getattr(self, '_config_context_data', None)
376 if cached is not None: 376 ↛ 377line 376 didn't jump to line 377 because the condition on line 376 was never true
377 return copy.deepcopy(cached)
378 return self.render_config_context()
380 def render_config_context(self):
381 """
382 Compile all config data, overwriting lower-weight values with higher-weight values where a collision occurs.
383 Return the rendered configuration context for a device or VM. This bypasses the pre-rendered cache
384 (`_config_context_data`); use get_config_context() for the cached read path.
385 """
386 data = {}
388 if not hasattr(self, 'config_context_data'):
389 # The annotation is not available, so we fall back to manually querying for the config context objects
390 config_context_data = ConfigContext.objects.get_for_object(self, aggregate_data=True) or []
391 else:
392 # The attribute may exist, but the annotated value could be None if there is no config context data
393 config_context_data = self.config_context_data or []
395 for context in config_context_data: 395 ↛ 396line 395 didn't jump to line 396 because the loop on line 395 never started
396 data = deepmerge(data, context)
398 # If the object has local config context data defined, merge it last
399 if self.local_context_data: 399 ↛ 400line 399 didn't jump to line 400 because the condition on line 399 was never true
400 data = deepmerge(data, self.local_context_data)
402 return data
404 def clean(self):
405 super().clean()
407 # Verify that JSON data is provided as an object
408 if self.local_context_data is not None and type(self.local_context_data) is not dict: 408 ↛ 409line 408 didn't jump to line 409 because the condition on line 408 was never true
409 raise ValidationError(
410 {'local_context_data': _('JSON data must be in object form. Example:') + ' {"foo": 123}'}
411 )
413 def serialize_object(self, exclude=None):
414 # Exclude the pre-rendered cache and its generation counter from change-log snapshots;
415 # they are derived fields and would otherwise produce noisy diffs.
416 exclude = list(exclude or [])
417 for field in ('_config_context_data', '_config_context_generation'):
418 if field not in exclude: 418 ↛ 417line 418 didn't jump to line 417 because the condition on line 418 was always true
419 exclude.append(field)
420 return super().serialize_object(exclude=exclude)
423#
424# Config templates
425#
427class ConfigTemplate(
428 RenderTemplateMixin,
429 SyncedDataMixin,
430 CustomLinksMixin,
431 ExportTemplatesMixin,
432 OwnerMixin,
433 TagsMixin,
434 ChangeLoggedModel,
435):
436 name = models.CharField(
437 verbose_name=_('name'),
438 max_length=100
439 )
440 description = models.CharField(
441 verbose_name=_('description'),
442 max_length=200,
443 blank=True
444 )
445 debug = models.BooleanField(
446 verbose_name=_('debug'),
447 default=False,
448 help_text=_(
449 'Enable verbose error output when rendering this template. Not recommended for production use.'
450 )
451 )
453 class Meta:
454 ordering = ('name',)
455 indexes = (
456 models.Index(fields=('name',)), # Default ordering
457 )
458 verbose_name = _('config template')
459 verbose_name_plural = _('config templates')
461 def __str__(self):
462 return self.name
464 def get_absolute_url(self):
465 return reverse('extras:configtemplate', args=[self.pk])
467 def sync_data(self):
468 """
469 Synchronize template content from the designated DataFile (if any).
470 """
471 self.template_code = self.validate_synced_value('template_code', self.data_file.data_as_string)
472 sync_data.alters_data = True
474 def get_environment_params(self):
475 """
476 Config templates render plain text (network configs, scripts), not HTML. Force
477 autoescape off so environment_params cannot enable it and create a latent XSS sink
478 if output is ever rendered in an HTML context.
479 """
480 params = super().get_environment_params()
481 params['autoescape'] = False
482 return params
484 def format_render_error(self, exc):
485 """
486 Return a formatted error string for a rendering exception. When debug is enabled, the full
487 traceback for the provided exception is returned. Otherwise, a concise, user-facing message
488 is returned.
489 """
490 if self.debug:
491 # Strip deployment-specific path prefixes from File "..." lines to avoid disclosing
492 # the server's filesystem layout. install_root covers all NetBox source files plus
493 # any venv co-located inside the repo. When the venv lives outside the repo
494 # (the typical production pattern, e.g. ~/.venv/netbox/), sys.prefix differs from
495 # sys.base_prefix and the venv root is stripped separately so that the deployment
496 # user's home directory is not exposed. Stdlib paths not under either prefix are
497 # left as-is — they reveal only standard OS locations, not deployment structure.
498 install_root = os.path.dirname(settings.BASE_DIR) + os.sep
499 prefixes_to_strip = [install_root]
500 if sys.prefix != sys.base_prefix:
501 venv_root = sys.prefix + os.sep
502 if venv_root != install_root:
503 prefixes_to_strip.append(venv_root)
504 tb = ''.join(traceback.format_exception(exc))
505 for prefix in prefixes_to_strip:
506 tb = re.sub(r'(File ")' + re.escape(prefix), r'\1', tb)
507 return tb
508 if isinstance(exc, TemplateError):
509 parts = [f"{type(exc).__name__}: {exc}"]
510 if getattr(exc, 'name', None):
511 parts.append(_("Template: {name}").format(name=exc.name))
512 if getattr(exc, 'lineno', None):
513 parts.append(_("Line: {lineno}").format(lineno=exc.lineno))
514 return "\n".join(parts)
515 return f"{type(exc).__name__}: {exc}"