diff --git a/netbox/dcim/api/serializers_/device_components.py b/netbox/dcim/api/serializers_/device_components.py index 8125dc238..04ae102af 100644 --- a/netbox/dcim/api/serializers_/device_components.py +++ b/netbox/dcim/api/serializers_/device_components.py @@ -364,6 +364,7 @@ class InterfaceSerializer( mac_address = _UNSET if isinstance(data, dict): mac_address = data.pop('mac_address', _UNSET) + self._validate_no_mac_conflict(data, mac_address) if not self.nested and isinstance(data, dict): if mac_address not in (_UNSET, None): diff --git a/netbox/dcim/api/serializers_/mixins.py b/netbox/dcim/api/serializers_/mixins.py index 4a856c813..d68cae198 100644 --- a/netbox/dcim/api/serializers_/mixins.py +++ b/netbox/dcim/api/serializers_/mixins.py @@ -1,9 +1,10 @@ +from django.core.exceptions import ValidationError as DjangoValidationError 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.models import MACAddress - _UNSET = object() __all__ = ( @@ -14,56 +15,72 @@ __all__ = ( 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. + Mixin for Interface and VMInterface serializers that adds a `mac_address` shortcut field for + creating/updating the primary MACAddress in a single request. The validated write is centralized + on the interface model (BaseInterface.set_primary_mac_address); this mixin is a thin adapter that + owns the permission check, the shortcut-vs-primary_mac_address conflict guard, and translating the + model's validation errors into API errors. """ + @staticmethod + def _validate_no_mac_conflict(data, mac_address): + # The mac_address shortcut and the primary_mac_address field both set the primary MAC. Reject + # only when both are supplied AND they disagree, so a read-modify-write round-trip (which echoes + # both readable fields with matching values) is accepted while a genuine conflict is rejected. + if mac_address is _UNSET or not isinstance(data, dict) or 'primary_mac_address' not in data: + return + + primary = data['primary_mac_address'] + primary_value = primary.mac_address if primary is not None else None + try: + shortcut_value = EUI(mac_address, version=48) if mac_address is not None else None + except (AddrFormatError, ValueError, TypeError): + # Leave an invalid shortcut value to the format check in each serializer's validate(). + return + + if shortcut_value != primary_value: + raise serializers.ValidationError( + _("The provided 'mac_address' and 'primary_mac_address' values conflict.") + ) + + def _check_add_mac_permission(self, instance, mac_address): + # A submitted value that doesn't already exist on this interface will be created, which requires + # add_macaddress. An existing value is only reassigned, so it doesn't. + if instance is not None and instance.mac_addresses.filter(mac_address=mac_address).exists(): + return + 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.')) + 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.')) + self._check_add_mac_permission(None, mac_address) 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) + self._set_primary_mac(instance, mac_address) 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 - + self._check_add_mac_permission(instance, mac_address) 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) + if mac_address is not _UNSET: + self._set_primary_mac(instance, mac_address) return instance + + def _set_primary_mac(self, instance, mac_address): + # Surface model/custom validation raised by the centralized operation as a DRF 400 rather than + # a 500 (it runs after DRF's own validation phase). The viewset's discard_events_on_rollback() + # clears any event queued for the rolled-back interface save. + try: + instance.set_primary_mac_address_from_value(mac_address) + except DjangoValidationError as e: + # Field-scoped errors (e.g. an interface-level check like qinq_svlan) keep their field key; + # non-field errors are attributed to the mac_address shortcut that triggered the operation. + if hasattr(e, 'error_dict'): + raise serializers.ValidationError(e.message_dict) + raise serializers.ValidationError({'mac_address': e.messages}) diff --git a/netbox/dcim/forms/common.py b/netbox/dcim/forms/common.py index 0c1673c34..ea25ac6bc 100644 --- a/netbox/dcim/forms/common.py +++ b/netbox/dcim/forms/common.py @@ -1,13 +1,13 @@ from django import forms -from django.db import transaction +from django.core.exceptions import ValidationError from django.utils.translation import gettext_lazy as _ from netaddr import EUI, AddrFormatError 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.exceptions import AbortRequest from utilities.forms import get_field_value __all__ = ( @@ -61,12 +61,14 @@ 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.') - }) + # Require add_macaddress only when the field is actually being changed (a MAC may need to be + # created). A pre-populated primary MAC left untouched must not gate unrelated edits. + if 'mac_address' in self.changed_data: + 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 \ @@ -96,30 +98,14 @@ class InterfaceCommonForm(forms.Form): def save(self, commit=True): instance = super().save(commit=commit) - if not commit or 'mac_address' not in self.changed_data: - return instance + if commit and 'mac_address' in self.changed_data: + try: + instance.set_primary_mac_address_from_value(self.cleaned_data.get('mac_address')) + except ValidationError as e: + # Surface a model/custom validation failure (e.g. a MACAddress CustomValidator) as a + # clean request abort rather than letting it escape as a 500. + raise AbortRequest('; '.join(e.messages)) - mac_address = self.cleaned_data.get('mac_address') - - 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() - - instance.__dict__.pop('mac_address', None) return instance diff --git a/netbox/dcim/models/device_components.py b/netbox/dcim/models/device_components.py index c8ac488cb..486c59721 100644 --- a/netbox/dcim/models/device_components.py +++ b/netbox/dcim/models/device_components.py @@ -5,7 +5,7 @@ from django.contrib.postgres.fields import ArrayField from django.contrib.postgres.indexes import GistIndex from django.core.exceptions import ObjectDoesNotExist, ValidationError from django.core.validators import MaxValueValidator, MinValueValidator -from django.db import models +from django.db import models, router, transaction from django.utils.translation import gettext_lazy as _ from dcim.choices import * @@ -882,20 +882,24 @@ class BaseInterface(models.Model): 'qinq_svlan': _("Only Q-in-Q interfaces may specify a service VLAN.") }) - # Check that the primary MAC address (if any) is assigned to this interface - if ( - self.primary_mac_address and - self.primary_mac_address.assigned_object is not None and - self.primary_mac_address.assigned_object != self - ): - raise ValidationError({ - 'primary_mac_address': _( - "MAC address {mac_address} is assigned to a different interface ({interface})." - ).format( - mac_address=self.primary_mac_address, - interface=self.primary_mac_address.assigned_object, + # A primary MAC address must belong to this interface. On create the MAC is assigned by a + # post_save signal after this runs, so an as-yet-unassigned MAC is only rejected on update + # (self._state.adding is False), where no such signal fires. These are raised as non-field + # errors: primary_mac_address is not an InterfaceForm field (it's edited via the mac_address + # shortcut), so a field-keyed error would raise in the form's add_error() rather than render. + if self.primary_mac_address: + if self.primary_mac_address.assigned_object is None: + if not self._state.adding: + raise ValidationError( + _("Only a MAC address assigned to this interface can be its primary MAC address.") + ) + elif self.primary_mac_address.assigned_object != self: + raise ValidationError( + _("MAC address {mac_address} is assigned to a different interface ({interface}).").format( + mac_address=self.primary_mac_address, + interface=self.primary_mac_address.assigned_object, + ) ) - }) def save(self, *args, **kwargs): @@ -909,6 +913,67 @@ class BaseInterface(models.Model): return super().save(*args, **kwargs) + def set_primary_mac_address(self, mac): + """ + Set (or clear) this interface's primary MAC address as a single atomic, validated operation. + Pass a MACAddress instance to designate it primary, or None to clear the primary MAC. The + callers own permission checks; this method owns the validated write. To set from a submitted + address string (find-or-create on this interface) use set_primary_mac_address_from_value(). + """ + self._set_primary_mac_address(mac=mac) + set_primary_mac_address.alters_data = True + + def set_primary_mac_address_from_value(self, mac_address): + """ + Set this interface's primary MAC address from a submitted address string, finding an existing + MAC on the interface or creating one, all within the operation's locked transaction. An empty + value clears the primary MAC. For the form and API adapters, which receive a string. + """ + self._set_primary_mac_address(mac_value=mac_address or None) + set_primary_mac_address_from_value.alters_data = True + + def _set_primary_mac_address(self, mac=None, mac_value=None): + """ + Shared implementation of the two public setters. Locks this interface's row, resolves a + submitted string to a MACAddress (find-or-create, inside the lock so concurrent requests can't + both create the same one), validates, and saves. Callers pass either a resolved MACAddress + (`mac`) or an address string (`mac_value`), never both. + """ + with transaction.atomic(using=router.db_for_write(type(self))): + # Lock and re-fetch this interface so concurrent set-primary/find-or-create requests + # serialize, and mutate the freshly-loaded row rather than the caller's in-memory instance. + # The re-fetch resets change-tracking state (e.g. _original_device) to the persisted values, + # so full_clean() validates the persisted object plus this one change, not unrelated edits the + # adapter already validated and saved. + locked = type(self).objects.select_for_update().get(pk=self.pk) + + # Resolve a submitted string to a MAC inside the lock, so two concurrent requests setting the + # same new value can't both miss the lookup and both create a duplicate. + if mac_value is not None: + mac = locked.mac_addresses.filter(mac_address=mac_value).first() + if mac is None: + mac = locked.mac_addresses.model(mac_address=mac_value, assigned_object=locked) + mac.full_clean() + mac.save() + + target_id = mac.pk if mac is not None else None + if locked.primary_mac_address_id == target_id: + self.primary_mac_address = mac + self.__dict__.pop('mac_address', None) + return + + # Snapshot the locked row (refetched after any adapter save this request) so the changelog + # records the correct pre-change state for this MAC change, not an earlier field edit. + locked.snapshot() + locked.primary_mac_address = mac + locked.full_clean(validate_unique=False) + locked.save() + + # Reflect the change on the caller's instance (for success messages and API responses) and + # invalidate the cached read-side mac_address property. + self.primary_mac_address = mac + self.__dict__.pop('mac_address', None) + @property def tunnel_termination(self): return self.tunnel_terminations.first() diff --git a/netbox/dcim/tables/devices.py b/netbox/dcim/tables/devices.py index ac10e8bd2..d879ac6c3 100644 --- a/netbox/dcim/tables/devices.py +++ b/netbox/dcim/tables/devices.py @@ -1,3 +1,5 @@ +from urllib.parse import quote + import django_tables2 as tables from django.middleware.csrf import get_token from django.urls import reverse @@ -1257,17 +1259,36 @@ class MACAddressActionsColumn(columns.ActionsColumn): request = getattr(table, 'context', {}).get('request') if request: url = reverse('dcim:macaddress_set_primary', kwargs={'pk': record.pk}) - form_li = format_html( - '
', - url, get_token(request), _('Set as primary'), - ) + # Return the user where they came from, the same way the parent's GET actions + # (edit/delete/changelog) do. In an embedded panel ObjectsTablePanel injects the parent + # object's URL as ?return_url=, so this lands on the interface; on the list view it falls + # back to the list path. + return_url = request.GET.get('return_url', request.get_full_path()) + url = f'{url}?return_url={quote(return_url)}' + # Embedded tables need their own form; list tables reuse the surrounding bulk form. + if getattr(table, 'embedded', False): + # No surrounding form: a self-contained POST form is valid and carries its own CSRF token. + action_li = format_html( + '', + url, get_token(request), _('Set as primary'), + ) + else: + # Inside the bulk-edit