Przeglądaj źródła

Address code review findings on PR #22520

- Extract MACAddressShortcutMixin with create()/update() from both
  InterfaceSerializer and VMInterfaceSerializer into a new
  dcim/api/serializers_/mixins.py (~90 lines of duplication removed)
- Add dcim.add_macaddress permission check in InterfaceCommonForm.clean()
  using current_request ContextVar, preventing form-based MAC creation
  without the required permission
- Wrap MAC writes in transaction.atomic() in InterfaceCommonForm.save()
- Add instance.snapshot() before each secondary save in
  InterfaceCommonForm.save() so changelog entries show accurate diffs
- Replace f-string HTML construction in MACAddressActionsColumn.render()
  with format_html() for proper argument escaping
- Add test_mac_address_find_or_create: verifies that PATCHing with an
  existing MAC promotes it to primary without creating a duplicate record

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Brian Tiemann 1 tydzień temu
rodzic
commit
ca96d67be4

+ 2 - 55
netbox/dcim/api/serializers_/device_components.py

@@ -1,9 +1,7 @@
 from django.contrib.contenttypes.models import ContentType
-from django.db import transaction
 from django.utils.translation import gettext as _
 from netaddr import EUI, AddrFormatError
 from rest_framework import serializers
-from rest_framework.exceptions import PermissionDenied
 
 from dcim.choices import *
 from dcim.constants import *
@@ -14,7 +12,6 @@ from dcim.models import (
     FrontPort,
     Interface,
     InventoryItem,
-    MACAddress,
     ModuleBay,
     PortMapping,
     PowerOutlet,
@@ -39,11 +36,10 @@ from .base import ConnectedEndpointsSerializer, PortSerializer
 from .cables import CabledObjectSerializer
 from .devices import DeviceSerializer, MACAddressSerializer, ModuleSerializer, VirtualDeviceContextSerializer
 from .manufacturers import ManufacturerSerializer
+from .mixins import _UNSET, MACAddressShortcutMixin
 from .nested import NestedInterfaceSerializer
 from .roles import InventoryItemRoleSerializer
 
-_UNSET = object()
-
 __all__ = (
     'ConsolePortSerializer',
     'ConsoleServerPortSerializer',
@@ -203,6 +199,7 @@ class PowerOutletSerializer(
 
 
 class InterfaceSerializer(
+    MACAddressShortcutMixin,
     OwnerMixin,
     NetBoxModelSerializer,
     CabledObjectSerializer,
@@ -356,56 +353,6 @@ class InterfaceSerializer(
 
         return data
 
-    def create(self, validated_data):
-        mac_address = validated_data.pop('mac_address', None)
-        if mac_address is not None:
-            request = self.context.get('request')
-            if request and not request.user.has_perm('dcim.add_macaddress'):
-                raise PermissionDenied(_('You do not have permission to create MAC addresses.'))
-        with transaction.atomic():
-            instance = super().create(validated_data)
-            if mac_address is not None:
-                mac = MACAddress.objects.create(mac_address=mac_address, assigned_object=instance)
-                instance.primary_mac_address = mac
-                instance.save()
-                instance.__dict__.pop('mac_address', None)
-        return instance
-
-    def update(self, instance, validated_data):
-        mac_address = validated_data.pop('mac_address', _UNSET)
-
-        # Check permission and locate any existing MAC before any writes.
-        if mac_address not in (_UNSET, None):
-            existing_mac = instance.mac_addresses.filter(mac_address=mac_address).first()
-            if existing_mac is None:
-                request = self.context.get('request')
-                if request and not request.user.has_perm('dcim.add_macaddress'):
-                    raise PermissionDenied(_('You do not have permission to create MAC addresses.'))
-        else:
-            existing_mac = None
-
-        with transaction.atomic():
-            instance = super().update(instance, validated_data)
-            if mac_address is _UNSET:
-                pass
-            elif mac_address is None:
-                if instance.primary_mac_address_id is not None:
-                    instance.snapshot()
-                    instance.primary_mac_address = None
-                    instance.save()
-            else:
-                # Find-or-create: prefer existing MAC on this interface; create only if absent.
-                mac = existing_mac
-                if mac is None:
-                    mac = MACAddress.objects.create(mac_address=mac_address, assigned_object=instance)
-                if instance.primary_mac_address_id != mac.pk:
-                    instance.snapshot()
-                    instance.primary_mac_address = mac
-                    instance.save()
-
-        instance.__dict__.pop('mac_address', None)
-        return instance
-
 
 class RearPortMappingSerializer(serializers.ModelSerializer):
     position = serializers.IntegerField(

+ 69 - 0
netbox/dcim/api/serializers_/mixins.py

@@ -0,0 +1,69 @@
+from django.db import transaction
+from django.utils.translation import gettext as _
+from rest_framework.exceptions import PermissionDenied
+
+from dcim.models import MACAddress
+
+_UNSET = object()
+
+__all__ = (
+    '_UNSET',
+    'MACAddressShortcutMixin',
+)
+
+
+class MACAddressShortcutMixin:
+    """
+    Mixin for Interface and VMInterface serializers that adds a write-only `mac_address` shortcut
+    field for creating/updating the primary MACAddress in a single request.
+    """
+
+    def create(self, validated_data):
+        mac_address = validated_data.pop('mac_address', None)
+        if mac_address is not None:
+            request = self.context.get('request')
+            if request and not request.user.has_perm('dcim.add_macaddress'):
+                raise PermissionDenied(_('You do not have permission to create MAC addresses.'))
+        with transaction.atomic():
+            instance = super().create(validated_data)
+            if mac_address is not None:
+                mac = MACAddress.objects.create(mac_address=mac_address, assigned_object=instance)
+                instance.primary_mac_address = mac
+                instance.save()
+                instance.__dict__.pop('mac_address', None)
+        return instance
+
+    def update(self, instance, validated_data):
+        mac_address = validated_data.pop('mac_address', _UNSET)
+
+        # Check permission and locate any existing MAC before any writes.
+        if mac_address not in (_UNSET, None):
+            existing_mac = instance.mac_addresses.filter(mac_address=mac_address).first()
+            if existing_mac is None:
+                request = self.context.get('request')
+                if request and not request.user.has_perm('dcim.add_macaddress'):
+                    raise PermissionDenied(_('You do not have permission to create MAC addresses.'))
+        else:
+            existing_mac = None
+
+        with transaction.atomic():
+            instance = super().update(instance, validated_data)
+            if mac_address is _UNSET:
+                pass
+            elif mac_address is None:
+                if instance.primary_mac_address_id is not None:
+                    instance.snapshot()
+                    instance.primary_mac_address = None
+                    instance.save()
+            else:
+                # Find-or-create: prefer existing MAC on this interface; create only if absent.
+                mac = existing_mac
+                if mac is None:
+                    mac = MACAddress.objects.create(mac_address=mac_address, assigned_object=instance)
+                if instance.primary_mac_address_id != mac.pk:
+                    instance.snapshot()
+                    instance.primary_mac_address = mac
+                    instance.save()
+
+        instance.__dict__.pop('mac_address', None)
+        return instance

+ 24 - 12
netbox/dcim/forms/common.py

@@ -1,4 +1,5 @@
 from django import forms
+from django.db import transaction
 from django.utils.translation import gettext_lazy as _
 from netaddr import EUI, AddrFormatError
 
@@ -6,6 +7,7 @@ from dcim.choices import *
 from dcim.constants import *
 from dcim.models import MACAddress
 from dcim.utils import get_module_bay_positions, resolve_module_placeholder
+from netbox.context import current_request
 from utilities.forms import get_field_value
 
 __all__ = (
@@ -59,6 +61,12 @@ class InterfaceCommonForm(forms.Form):
                 raise forms.ValidationError({
                     'mac_address': _('Enter a valid MAC address (e.g. 00:11:22:33:44:55).')
                 })
+            # Require add_macaddress permission when a MAC value is provided (it may need to be created).
+            request = current_request.get()
+            if request is not None and not request.user.has_perm('dcim.add_macaddress'):
+                raise forms.ValidationError({
+                    'mac_address': _('You do not have permission to create MAC addresses.')
+                })
         parent_field = 'device' if 'device' in self.cleaned_data else 'virtual_machine'
         if 'tagged_vlans' in self.fields.keys():
             tagged_vlans = self.cleaned_data.get('tagged_vlans') if self.is_bound else \
@@ -93,19 +101,23 @@ class InterfaceCommonForm(forms.Form):
 
         mac_address = self.cleaned_data.get('mac_address')
 
-        if mac_address:
-            # Find an existing MACAddress on this interface with the target value, or create one.
-            # Using find-or-create avoids duplicating a MAC that already exists on this interface.
-            mac = instance.mac_addresses.filter(mac_address=mac_address).first()
-            if mac is None:
-                mac = MACAddress(mac_address=mac_address, assigned_object=instance)
-                mac.save()
-            if instance.primary_mac_address_id != mac.pk:
-                instance.primary_mac_address = mac
+        with transaction.atomic():
+            if mac_address:
+                # Find an existing MACAddress on this interface with the target value, or create one.
+                # Using find-or-create avoids duplicating a MAC that already exists on this interface.
+                mac = instance.mac_addresses.filter(mac_address=mac_address).first()
+                if mac is None:
+                    mac = MACAddress(mac_address=mac_address, assigned_object=instance)
+                    mac.save()
+                if instance.primary_mac_address_id != mac.pk:
+                    instance.snapshot()
+                    instance.primary_mac_address = mac
+                    instance.save()
+            else:
+                if instance.primary_mac_address_id is not None:
+                    instance.snapshot()
+                instance.primary_mac_address = None
                 instance.save()
-        else:
-            instance.primary_mac_address = None
-            instance.save()
 
         instance.__dict__.pop('mac_address', None)
         return instance

+ 9 - 10
netbox/dcim/tables/devices.py

@@ -1,6 +1,7 @@
 import django_tables2 as tables
 from django.middleware.csrf import get_token
 from django.urls import reverse
+from django.utils.html import format_html
 from django.utils.safestring import mark_safe
 from django.utils.translation import gettext_lazy as _
 from django_tables2.utils import Accessor
@@ -1242,19 +1243,17 @@ class MACAddressActionsColumn(columns.ActionsColumn):
             request = getattr(table, 'context', {}).get('request')
             if request:
                 url = reverse('dcim:macaddress_set_primary', kwargs={'pk': record.pk})
-                form_li = (
-                    f'<li>'
-                    f'<form method="post" action="{url}">'
-                    f'<input type="hidden" name="csrfmiddlewaretoken" value="{get_token(request)}">'
-                    f'<button type="submit" class="dropdown-item">'
-                    f'<i class="mdi mdi-star-outline"></i> {_("Set as primary")}'
-                    f'</button>'
-                    f'</form>'
-                    f'</li>'
+                form_li = format_html(
+                    '<li><form method="post" action="{}">'
+                    '<input type="hidden" name="csrfmiddlewaretoken" value="{}">'
+                    '<button type="submit" class="dropdown-item">'
+                    '<i class="mdi mdi-star-outline"></i> {}'
+                    '</button></form></li>',
+                    url, get_token(request), _('Set as primary'),
                 )
                 html_str = str(html)
                 if '</ul>' in html_str:
-                    html = mark_safe(html_str.replace('</ul>', form_li + '</ul>', 1))
+                    html = mark_safe(html_str.replace('</ul>', str(form_li) + '</ul>', 1))
 
         return html
 

+ 25 - 0
netbox/dcim/tests/test_api.py

@@ -2891,6 +2891,31 @@ class InterfaceTestCase(Mixins.ComponentTraceMixin, APIViewTestCases.APIViewTest
         self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
         self.assertIn('mac_address', response.data)
 
+    def test_mac_address_find_or_create(self):
+        """
+        Patching mac_address with a MAC that already exists on the interface promotes it to primary
+        without creating a duplicate MACAddress record.
+        """
+        self.add_permissions('dcim.change_interface', 'dcim.add_macaddress', 'dcim.change_macaddress')
+        iface = Interface.objects.first()
+
+        # Pre-create two MACs assigned to this interface
+        mac1 = MACAddress.objects.create(mac_address='CC:DD:EE:FF:00:01', assigned_object=iface)
+        mac2 = MACAddress.objects.create(mac_address='CC:DD:EE:FF:00:02', assigned_object=iface)
+        iface.primary_mac_address = mac1
+        iface.save()
+
+        mac_count_before = iface.mac_addresses.count()
+        url = self._get_detail_url(iface)
+
+        # PATCH with mac2's address — should promote mac2, not create a new record
+        response = self.client.patch(url, {'mac_address': 'CC:DD:EE:FF:00:02'}, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+
+        iface.refresh_from_db()
+        self.assertEqual(iface.primary_mac_address.pk, mac2.pk)
+        self.assertEqual(iface.mac_addresses.count(), mac_count_before)
+
 
 class FrontPortTestCase(APIViewTestCases.APIViewTestCase):
     model = FrontPort

+ 2 - 56
netbox/virtualization/api/serializers_/virtualmachines.py

@@ -1,16 +1,14 @@
-from django.db import transaction
 from django.utils.translation import gettext as _
 from drf_spectacular.utils import extend_schema_field
 from netaddr import EUI, AddrFormatError
 from rest_framework import serializers
-from rest_framework.exceptions import PermissionDenied
 
 from dcim.api.serializers_.devices import DeviceSerializer, MACAddressSerializer
+from dcim.api.serializers_.mixins import _UNSET, MACAddressShortcutMixin
 from dcim.api.serializers_.platforms import PlatformSerializer
 from dcim.api.serializers_.roles import DeviceRoleSerializer
 from dcim.api.serializers_.sites import SiteSerializer
 from dcim.choices import InterfaceModeChoices
-from dcim.models import MACAddress
 from extras.api.serializers_.configtemplates import ConfigTemplateSerializer
 from ipam.api.serializers_.ip import IPAddressSerializer
 from ipam.api.serializers_.vlans import VLANSerializer, VLANTranslationPolicySerializer
@@ -27,8 +25,6 @@ from ...models import VirtualDisk, VirtualMachine, VirtualMachineType, VMInterfa
 from .clusters import ClusterSerializer
 from .nested import NestedVMInterfaceSerializer
 
-_UNSET = object()
-
 __all__ = (
     'VMInterfaceSerializer',
     'VirtualDiskSerializer',
@@ -108,7 +104,7 @@ class VirtualMachineSerializer(PrimaryModelSerializer):
 # VM interfaces
 #
 
-class VMInterfaceSerializer(OwnerMixin, NetBoxModelSerializer):
+class VMInterfaceSerializer(MACAddressShortcutMixin, OwnerMixin, NetBoxModelSerializer):
     virtual_machine = VirtualMachineSerializer(nested=True)
     parent = NestedVMInterfaceSerializer(required=False, allow_null=True)
     bridge = NestedVMInterfaceSerializer(required=False, allow_null=True)
@@ -192,56 +188,6 @@ class VMInterfaceSerializer(OwnerMixin, NetBoxModelSerializer):
 
         return data
 
-    def create(self, validated_data):
-        mac_address = validated_data.pop('mac_address', None)
-        if mac_address is not None:
-            request = self.context.get('request')
-            if request and not request.user.has_perm('dcim.add_macaddress'):
-                raise PermissionDenied(_('You do not have permission to create MAC addresses.'))
-        with transaction.atomic():
-            instance = super().create(validated_data)
-            if mac_address is not None:
-                mac = MACAddress.objects.create(mac_address=mac_address, assigned_object=instance)
-                instance.primary_mac_address = mac
-                instance.save()
-                instance.__dict__.pop('mac_address', None)
-        return instance
-
-    def update(self, instance, validated_data):
-        mac_address = validated_data.pop('mac_address', _UNSET)
-
-        # Check permission and locate any existing MAC before any writes.
-        if mac_address not in (_UNSET, None):
-            existing_mac = instance.mac_addresses.filter(mac_address=mac_address).first()
-            if existing_mac is None:
-                request = self.context.get('request')
-                if request and not request.user.has_perm('dcim.add_macaddress'):
-                    raise PermissionDenied(_('You do not have permission to create MAC addresses.'))
-        else:
-            existing_mac = None
-
-        with transaction.atomic():
-            instance = super().update(instance, validated_data)
-            if mac_address is _UNSET:
-                pass
-            elif mac_address is None:
-                if instance.primary_mac_address_id is not None:
-                    instance.snapshot()
-                    instance.primary_mac_address = None
-                    instance.save()
-            else:
-                # Find-or-create: prefer existing MAC on this interface; create only if absent.
-                mac = existing_mac
-                if mac is None:
-                    mac = MACAddress.objects.create(mac_address=mac_address, assigned_object=instance)
-                if instance.primary_mac_address_id != mac.pk:
-                    instance.snapshot()
-                    instance.primary_mac_address = mac
-                    instance.save()
-
-        instance.__dict__.pop('mac_address', None)
-        return instance
-
 
 #
 # Virtual Disk