Coverage for extras/api/views.py: 76%
221 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1from django.core.exceptions import NON_FIELD_ERRORS
2from django.core.exceptions import ValidationError as DjangoValidationError
3from django.http import Http404
4from django.shortcuts import get_object_or_404
5from django.utils.translation import gettext_lazy as _
6from drf_spectacular.utils import OpenApiResponse, OpenApiTypes, extend_schema
7from rest_framework.decorators import action
8from rest_framework.exceptions import PermissionDenied, ValidationError
9from rest_framework.generics import RetrieveUpdateDestroyAPIView
10from rest_framework.mixins import CreateModelMixin, ListModelMixin, RetrieveModelMixin, UpdateModelMixin
11from rest_framework.renderers import JSONRenderer
12from rest_framework.response import Response
13from rest_framework.routers import APIRootView
15from core.choices import ManagedFileRootPathChoices
16from extras import filtersets
17from extras.jobs import ScriptJob
18from extras.models import *
19from extras.scripts import EXEC_PARAM_FIELDS, prepare_script_form
20from netbox.api.authentication import IsAuthenticatedOrLoginNotRequired, TokenWritePermission
21from netbox.api.features import SyncedDataMixin
22from netbox.api.metadata import ContentTypeMetadata
23from netbox.api.renderers import TextRenderer
24from netbox.api.viewsets import BaseViewSet, NetBoxModelViewSet
25from netbox.api.viewsets.mixins import ObjectValidationMixin
26from users.models import Token
27from utilities.exceptions import RQWorkerNotRunningException
28from utilities.request import copy_safe_request
29from utilities.rqworker import any_workers_for_queue
31from . import serializers
32from .mixins import ConfigTemplateRenderMixin, SharedObjectQuerySetMixin
35class ExtrasRootView(APIRootView):
36 """
37 Extras API root view
38 """
39 def get_view_name(self):
40 return 'Extras'
43#
44# EventRules
45#
47class EventRuleViewSet(NetBoxModelViewSet):
48 metadata_class = ContentTypeMetadata
49 queryset = EventRule.objects.all()
50 serializer_class = serializers.EventRuleSerializer
51 filterset_class = filtersets.EventRuleFilterSet
54#
55# Webhooks
56#
58class WebhookViewSet(NetBoxModelViewSet):
59 metadata_class = ContentTypeMetadata
60 queryset = Webhook.objects.all()
61 serializer_class = serializers.WebhookSerializer
62 filterset_class = filtersets.WebhookFilterSet
65#
66# Custom fields
67#
69class CustomFieldViewSet(NetBoxModelViewSet):
70 metadata_class = ContentTypeMetadata
71 queryset = CustomField.objects.select_related('choice_set')
72 serializer_class = serializers.CustomFieldSerializer
73 filterset_class = filtersets.CustomFieldFilterSet
76class CustomFieldChoiceSetViewSet(NetBoxModelViewSet):
77 queryset = CustomFieldChoiceSet.objects.all()
78 serializer_class = serializers.CustomFieldChoiceSetSerializer
79 filterset_class = filtersets.CustomFieldChoiceSetFilterSet
81 @action(detail=True)
82 def choices(self, request, pk):
83 """
84 Provides an endpoint to iterate through each choice in a set.
85 """
86 choiceset = get_object_or_404(self.queryset, pk=pk)
87 choices = choiceset.choices
89 # Enable filtering
90 if q := request.GET.get('q'):
91 q = q.lower()
92 choices = [c for c in choices if q in c[0].lower() or q in c[1].lower()]
94 # Paginate data
95 if page := self.paginate_queryset(choices):
96 data = [
97 {'id': c[0], 'display': c[1]} for c in page
98 ]
99 else:
100 data = []
102 return self.get_paginated_response(data)
105#
106# Custom links
107#
109class CustomLinkViewSet(NetBoxModelViewSet):
110 metadata_class = ContentTypeMetadata
111 queryset = CustomLink.objects.all()
112 serializer_class = serializers.CustomLinkSerializer
113 filterset_class = filtersets.CustomLinkFilterSet
116#
117# Export templates
118#
120class ExportTemplateViewSet(SyncedDataMixin, NetBoxModelViewSet):
121 metadata_class = ContentTypeMetadata
122 queryset = ExportTemplate.objects.all()
123 serializer_class = serializers.ExportTemplateSerializer
124 filterset_class = filtersets.ExportTemplateFilterSet
127#
128# Saved filters
129#
131class SavedFilterViewSet(SharedObjectQuerySetMixin, NetBoxModelViewSet):
132 metadata_class = ContentTypeMetadata
133 queryset = SavedFilter.objects.all()
134 serializer_class = serializers.SavedFilterSerializer
135 filterset_class = filtersets.SavedFilterFilterSet
138#
139# Table Configs
140#
142class TableConfigViewSet(SharedObjectQuerySetMixin, NetBoxModelViewSet):
143 metadata_class = ContentTypeMetadata
144 queryset = TableConfig.objects.all()
145 serializer_class = serializers.TableConfigSerializer
146 filterset_class = filtersets.TableConfigFilterSet
149#
150# Bookmarks
151#
153class BookmarkViewSet(NetBoxModelViewSet):
154 metadata_class = ContentTypeMetadata
155 queryset = Bookmark.objects.all()
156 serializer_class = serializers.BookmarkSerializer
157 filterset_class = filtersets.BookmarkFilterSet
160#
161# Notifications & subscriptions
162#
164class NotificationViewSet(NetBoxModelViewSet):
165 metadata_class = ContentTypeMetadata
166 queryset = Notification.objects.all()
167 serializer_class = serializers.NotificationSerializer
170class NotificationGroupViewSet(NetBoxModelViewSet):
171 queryset = NotificationGroup.objects.all()
172 serializer_class = serializers.NotificationGroupSerializer
175class SubscriptionViewSet(NetBoxModelViewSet):
176 metadata_class = ContentTypeMetadata
177 queryset = Subscription.objects.all()
178 serializer_class = serializers.SubscriptionSerializer
181#
182# Tags
183#
185class TagViewSet(NetBoxModelViewSet):
186 queryset = Tag.objects.all()
187 serializer_class = serializers.TagSerializer
188 filterset_class = filtersets.TagFilterSet
191class TaggedItemViewSet(RetrieveModelMixin, ListModelMixin, BaseViewSet):
192 queryset = TaggedItem.objects.prefetch_related(
193 'content_type', 'content_object', 'tag'
194 ).order_by('tag__weight', 'tag__name')
195 serializer_class = serializers.TaggedItemSerializer
196 filterset_class = filtersets.TaggedItemFilterSet
199#
200# Image attachments
201#
203class ImageAttachmentViewSet(NetBoxModelViewSet):
204 metadata_class = ContentTypeMetadata
205 queryset = ImageAttachment.objects.all()
206 serializer_class = serializers.ImageAttachmentSerializer
207 filterset_class = filtersets.ImageAttachmentFilterSet
210#
211# Journal entries
212#
214class JournalEntryViewSet(NetBoxModelViewSet):
215 metadata_class = ContentTypeMetadata
216 queryset = JournalEntry.objects.all()
217 serializer_class = serializers.JournalEntrySerializer
218 filterset_class = filtersets.JournalEntryFilterSet
221#
222# Config contexts
223#
225class ConfigContextProfileViewSet(SyncedDataMixin, NetBoxModelViewSet):
226 queryset = ConfigContextProfile.objects.all()
227 serializer_class = serializers.ConfigContextProfileSerializer
228 filterset_class = filtersets.ConfigContextProfileFilterSet
231class ConfigContextViewSet(SyncedDataMixin, NetBoxModelViewSet):
232 queryset = ConfigContext.objects.all()
233 serializer_class = serializers.ConfigContextSerializer
234 filterset_class = filtersets.ConfigContextFilterSet
237#
238# Config templates
239#
241class ConfigTemplateViewSet(SyncedDataMixin, ConfigTemplateRenderMixin, NetBoxModelViewSet):
242 queryset = ConfigTemplate.objects.all()
243 serializer_class = serializers.ConfigTemplateSerializer
244 filterset_class = filtersets.ConfigTemplateFilterSet
246 def get_permissions(self):
247 # For render action, check only token write ability (not model permissions)
248 if self.action == 'render':
249 return [TokenWritePermission()]
250 return super().get_permissions()
252 @extend_schema(
253 request=OpenApiTypes.OBJECT,
254 responses={
255 200: OpenApiResponse(
256 response=serializers.RenderedConfigSerializer,
257 description=_(
258 "The rendered config template. When the client requests `text/plain`, the raw "
259 "rendered content is returned in place of the JSON object."
260 ),
261 ),
262 500: OpenApiResponse(
263 response=OpenApiTypes.OBJECT,
264 description=_("An error occurred while rendering the config template."),
265 ),
266 },
267 )
268 @action(detail=True, methods=['post'], renderer_classes=[JSONRenderer, TextRenderer])
269 def render(self, request, pk):
270 """
271 Render a ConfigTemplate using the context data provided (if any). The request body should be a
272 mapping of context variables to make available to the template. If the client requests "text/plain"
273 data, return the raw rendered content, rather than serialized JSON.
274 """
275 # Override restrict() on the default queryset to enforce the render & view actions
276 self.queryset = self.queryset.model.objects.restrict(request.user, 'render').restrict(request.user, 'view')
277 configtemplate = self.get_object()
279 context = request.data
281 return self.render_configtemplate(request, configtemplate, context)
284#
285# Scripts
286#
288class ScriptModuleViewSet(ObjectValidationMixin, CreateModelMixin, UpdateModelMixin, BaseViewSet):
289 queryset = ScriptModule.objects.filter(file_root=ManagedFileRootPathChoices.SCRIPTS)
290 serializer_class = serializers.ScriptModuleSerializer
291 lookup_value_regex = '[^/]+' # Allow dots
293 def get_object(self):
294 """
295 Retrieve a ScriptModule by numeric ID or by file name (e.g. my_script.py).
296 """
297 queryset = self.filter_queryset(self.get_queryset())
298 lookup = self.kwargs.get(self.lookup_url_kwarg or self.lookup_field, '')
300 # Support lookup by numeric PK or by file_path. Treat all-decimal values as PKs
301 # to preserve normal detail-route behavior; otherwise resolve the value as a
302 # script module filename, e.g. "myscript.py".
303 if lookup.isdecimal():
304 obj = get_object_or_404(queryset, pk=int(lookup))
305 else:
306 obj = get_object_or_404(queryset, file_path=lookup)
308 self.check_object_permissions(self.request, obj)
309 return obj
312class ScriptViewSet(ListModelMixin, RetrieveModelMixin, BaseViewSet):
313 # Individual scripts are created, modified, and deleted through their module (see ScriptModuleViewSet),
314 # so the standard write actions are intentionally omitted here. Only listing/retrieving a script (GET)
315 # and running one (POST to the detail route) are supported.
316 permission_classes = [IsAuthenticatedOrLoginNotRequired]
317 queryset = Script.objects.all()
318 serializer_class = serializers.ScriptSerializer
319 filterset_class = filtersets.ScriptFilterSet
321 lookup_value_regex = '[^/]+' # Allow dots
323 def get_serializer(self, *args, **kwargs):
324 # A POST to the detail route runs the script, taking ScriptInputSerializer as its request body.
325 # (This is keyed on the request method rather than on self.action, which is unset when generating
326 # OPTIONS metadata.) ScriptInputSerializer is instantiated directly rather than via BaseViewSet,
327 # which would pass it the fields/omit kwargs supported only by BaseModelSerializer.
328 if getattr(self.request, 'method', None) == 'POST':
329 kwargs.setdefault('context', self.get_serializer_context())
330 return serializers.ScriptInputSerializer(*args, **kwargs)
331 return super().get_serializer(*args, **kwargs)
333 def get_serializer_context(self):
334 context = super().get_serializer_context()
336 # ScriptInputSerializer resolves its field defaults and validates scheduling against the script
337 # being run (set by run() below).
338 context['script'] = getattr(self, 'script', None)
340 return context
342 def _get_script(self, pk):
343 # Retrieve the script by ID if the PK is all decimal digits. (isdecimal() rather than isnumeric(),
344 # as the latter also matches characters which cannot be cast to an integer.)
345 if pk.isdecimal():
346 try:
347 pk = int(pk)
348 except ValueError:
349 raise Http404
350 return get_object_or_404(self.queryset, pk=pk)
352 # Default to retrieval by module & name
353 try:
354 module_name, script_name = pk.split('.', maxsplit=1)
355 except ValueError:
356 raise Http404
358 return get_object_or_404(self.queryset, module__file_path=f'{module_name}.py', name=script_name)
360 def retrieve(self, request, pk, **kwargs):
361 script = self._get_script(pk)
362 serializer = serializers.ScriptDetailSerializer(script, context={'request': request})
364 return Response(serializer.data)
366 @extend_schema(
367 operation_id='extras_scripts_run',
368 request=serializers.ScriptInputSerializer,
369 responses={
370 200: OpenApiResponse(
371 response=serializers.ScriptDetailSerializer,
372 description=_("The script has been enqueued for execution."),
373 ),
374 },
375 )
376 def run(self, request, pk, **kwargs):
377 """
378 Run a Script identified by its numeric PK or module & name and return the pending Job as the result
379 """
380 # Bound to POST on the detail route by ScriptRouter
382 # Reject read-only tokens before resolving the script, so that an insufficient token is always
383 # reported as such. (Not via TokenWritePermission, which permits token auth only.)
384 if isinstance(request.auth, Token) and not request.auth.write_enabled: 384 ↛ 385line 384 didn't jump to line 385 because the condition on line 384 was never true
385 raise PermissionDenied(_("This token does not permit write operations (running a script)."))
387 # An unauthenticated user can never run a script; report that explicitly, as restrict() below would
388 # match no scripts and yield a misleading 404.
389 if not request.user.is_authenticated: 389 ↛ 390line 389 didn't jump to line 390 because the condition on line 389 was never true
390 raise PermissionDenied(_("This user does not have permission to run this script."))
392 # Running a script is a 'run' operation (not the 'add' that BaseViewSet maps to POST), so restrict
393 # the QuerySet on 'run' before resolving the script. A script the user cannot run yields a 404.
394 self.queryset = self.queryset.model.objects.restrict(request.user, 'run')
395 self.script = script = self._get_script(pk)
397 # A script whose Python class cannot be resolved (e.g. its module has been modified or the script has
398 # been deleted, retaining the record for its jobs) cannot be run
399 if not script.is_executable or script.python_class is None:
400 raise ValidationError(_("This script is not currently executable."))
402 input_serializer = self.get_serializer(data=request.data)
404 # Check that at least one RQ worker is running
405 if not any_workers_for_queue('default'):
406 raise RQWorkerNotRunningException()
408 input_serializer.is_valid(raise_exception=True)
410 validated = input_serializer.validated_data
412 payload = validated['data']
414 # Guaranteed non-None by the is_executable check above
415 script_class = script.python_class
416 script_instance = script_class()
418 form = prepare_script_form(script_instance, payload, files=request.FILES)
419 if not form.is_valid():
420 # Exec params are validated separately via ScriptInputSerializer. Excluded by name
421 # rather than by '_' prefix, which would also strip Django's NON_FIELD_ERRORS
422 # key ('__all__').
423 errors = {k: v for k, v in form.errors.items() if k not in EXEC_PARAM_FIELDS}
424 if not errors:
425 # Every error was on an exec-param field, which a client can bind by naming one
426 # in 'data' (e.g. {"_interval": "abc"}). NON_FIELD_ERRORS is never among them --
427 # the filter above retains '__all__' -- so there is nothing to re-surface here;
428 # report a generic message rather than an empty body.
429 errors = {NON_FIELD_ERRORS: [_('Invalid script input.')]}
430 # Nest under 'data' so script-variable errors can't collide with the
431 # serializer's own top-level fields (commit, schedule_at, interval, ...).
432 raise ValidationError({'data': errors})
434 data = form.cleaned_data.copy()
435 for k in EXEC_PARAM_FIELDS:
436 data.pop(k, None)
438 try:
439 ScriptJob.enqueue(
440 instance=script,
441 user=request.user,
442 data=data,
443 request=copy_safe_request(request),
444 commit=validated.get('commit'),
445 job_timeout=script_class.job_timeout,
446 schedule_at=validated.get('schedule_at'),
447 interval=validated.get('interval'),
448 notifications=validated.get('notifications'),
449 )
450 except DjangoValidationError as e:
451 # The script's execution configuration is invalid (see #22872). Surface it as a 400 rather than
452 # allowing the exception to bubble up as an HTTP 500. These are script-level config errors, not
453 # request-field errors, so report them under the non-field "detail" key.
454 raise ValidationError({'detail': e.messages}) from e
456 serializer = serializers.ScriptDetailSerializer(script, context={'request': request})
457 return Response(serializer.data)
460#
461# User dashboard
462#
464class DashboardView(RetrieveUpdateDestroyAPIView):
465 queryset = Dashboard.objects.all()
466 serializer_class = serializers.DashboardSerializer
468 def get_object(self):
469 return Dashboard.objects.filter(user=self.request.user).first()