Closes #22462: Allow terminating an L2VPN to an IP address (#23592)

* Closes #22462: Allow terminating an L2VPN to an IP address

Add IP addresses as an L2VPN termination type, e.g. for the far-end PE of a
pseudowire, with the pseudowire ID recorded as the L2VPN's identifier. An IP
address may terminate any number of L2VPNs, but each one only once, so the
unique constraint now covers the L2VPN as well; interfaces and VLANs keep
their one-L2VPN limit through model validation.

* Reject a duplicate IP address termination in form validation

Model forms skip unique constraints on fields they don't include, so a
second termination of the same L2VPN to the same IP address reached the
database and raised an IntegrityError. Check for it in clean(), as is
already done for interfaces and VLANs.

* Drop the parent column from the IP address's L2VPN terminations table

* Make the L2VPN termination API list test independent of row order

* Address review feedback on L2VPN IP address terminations

- Extend the termination site, region, device and VM filters to IP addresses assigned to interfaces
- Show a VLAN termination's site, and drop a duplicate branch in assigned_object_parent
- Reject a termination import row with no assigned object instead of failing on insert
- Import L2VPNTerminationType for type checking in the IP address GraphQL type

* Address further review feedback on L2VPN IP address terminations

docs/features/l2vpn-overlay.md

@@ -2,4 +2,4 @@
 
 L2VPN and overlay networks, such as VXLAN and EVPN, can be defined in NetBox and tied to interfaces and VLANs. This allows for easy tracking of overlay assets and their relationships with underlay resources.
 
-Each L2VPN instance has a type and optional unique identifier. Like VRFs, L2VPNs can also have import and export route targets assigned to them. Terminations can then be created to assign VLANs and/or device and virtual machine interfaces to the overlay.
+Each L2VPN instance has a type and optional unique identifier. Like VRFs, L2VPNs can also have import and export route targets assigned to them. Terminations can then be created to assign VLANs, device and virtual machine interfaces, and/or IP addresses (such as the far-end of a pseudowire) to the overlay.

docs/models/vpn/l2vpn.md

@@ -1,6 +1,6 @@
 # L2VPN
 
-A L2VPN object is NetBox is a representation of a layer 2 bridge technology such as VXLAN, VPLS, or EPL. Each L2VPN can be identified by name as well as by an optional unique identifier (VNI would be an example). Once created, L2VPNs can be terminated to [interfaces](../dcim/interface.md) and [VLANs](../ipam/vlan.md).
+A L2VPN object is NetBox is a representation of a layer 2 bridge technology such as VXLAN, VPLS, or EPL. Each L2VPN can be identified by name as well as by an optional unique identifier (VNI would be an example). Once created, L2VPNs can be terminated to [interfaces](../dcim/interface.md), [VLANs](../ipam/vlan.md), and [IP addresses](../ipam/ipaddress.md).
 
 ## Fields
 

docs/models/vpn/l2vpntermination.md

@@ -1,6 +1,6 @@
 # L2VPN Termination
 
-A L2VPN termination is the attachment of an [L2VPN](./l2vpn.md) to an [interface](../dcim/interface.md) or [VLAN](../ipam/vlan.md). Note that the L2VPNs of the following types may have only two terminations assigned to them:
+A L2VPN termination is the attachment of an [L2VPN](./l2vpn.md) to an [interface](../dcim/interface.md), [VLAN](../ipam/vlan.md), or [IP address](../ipam/ipaddress.md). Note that the L2VPNs of the following types may have only two terminations assigned to them:
 
 * VPWS
 * EPL
@@ -13,6 +13,8 @@ A L2VPN termination is the attachment of an [L2VPN](./l2vpn.md) to an [interface
 
 The [L2VPN](./l2vpn.md) instance.
 
-### VLAN or Interface
+### VLAN, Interface, or IP Address
 
-The [VLAN](../ipam/vlan.md), [device interface](../dcim/interface.md), or [virtual machine interface](../virtualization/virtualmachine.md) attached to the L2VPN.
+The [VLAN](../ipam/vlan.md), [device interface](../dcim/interface.md), [virtual machine interface](../virtualization/virtualmachine.md), or [IP address](../ipam/ipaddress.md) attached to the L2VPN.
+
+An interface or VLAN can terminate only one L2VPN. An IP address can terminate any number of L2VPNs, but each L2VPN only once. For example, a pseudowire can be terminated to the IP address of its far-end provider edge (PE) device, with the pseudowire ID recorded as the L2VPN's identifier.

netbox/ipam/graphql/filters.py

@@ -27,7 +27,7 @@ if TYPE_CHECKING:
     from core.graphql.filters import ContentTypeFilter
     from dcim.graphql.filters import SiteFilter
     from netbox.graphql.filter_lookups import BigIntegerLookup, IntegerLookup, IntegerRangeArrayLookup
-    from vpn.graphql.filters import L2VPNFilter
+    from vpn.graphql.filters import L2VPNFilter, L2VPNTerminationFilter
 
     from .enums import *
 
@@ -198,6 +198,9 @@ class IPAddressFilter(ContactFilterMixin, TenancyFilterMixin, PrimaryModelFilter
     )
     nat_outside_id: ID | None = strawberry_django.filter_field()
     dns_name: StrFilterLookup | None = strawberry_django.filter_field()
+    l2vpn_terminations: Annotated['L2VPNTerminationFilter', strawberry.lazy('vpn.graphql.filters')] | None = (
+        strawberry_django.filter_field()
+    )
 
     @strawberry_django.filter_field()
     def assigned(self, value: bool, prefix) -> Q:

netbox/ipam/graphql/types.py

@@ -34,7 +34,7 @@ if TYPE_CHECKING:
     )
     from tenancy.graphql.types import TenantType
     from virtualization.graphql.types import ClusterGroupType, ClusterType, VirtualMachineType, VMInterfaceType
-    from vpn.graphql.types import L2VPNType, TunnelTerminationType
+    from vpn.graphql.types import L2VPNTerminationType, L2VPNType, TunnelTerminationType
     from wireless.graphql.types import WirelessLANType
 
 __all__ = (
@@ -167,6 +167,7 @@ class IPAddressType(ContactsMixin, PrimaryObjectType):
     nat_inside: Annotated['IPAddressType', strawberry.lazy('ipam.graphql.types')] | None
     nat_outside: list[Annotated['IPAddressType', strawberry.lazy('ipam.graphql.types')]]
     tunnel_terminations: list[Annotated['TunnelTerminationType', strawberry.lazy('vpn.graphql.types')]]
+    l2vpn_terminations: list[Annotated['L2VPNTerminationType', strawberry.lazy('vpn.graphql.types')]]
     services: list[Annotated['ServiceType', strawberry.lazy('ipam.graphql.types')]]
 
     # Note that self is the Django model instance, so self.family resolves to the model's family property

netbox/ipam/models/ip.py

@@ -1,5 +1,5 @@
 import netaddr
-from django.contrib.contenttypes.fields import GenericForeignKey
+from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
 from django.contrib.contenttypes.models import ContentType
 from django.contrib.postgres.indexes import GistIndex
 from django.core.exceptions import ValidationError
@@ -1013,6 +1013,12 @@ class IPAddress(ContactsMixin, PrimaryModel):
         verbose_name=_('DNS name'),
         help_text=_('Hostname or FQDN (not case-sensitive)')
     )
+    l2vpn_terminations = GenericRelation(
+        to='vpn.L2VPNTermination',
+        content_type_field='assigned_object_type',
+        object_id_field='assigned_object_id',
+        related_query_name='ipaddress',
+    )
 
     objects = IPAddressManager()
 

netbox/ipam/views.py

@@ -1148,6 +1148,12 @@ class IPAddressView(generic.ObjectView):
                 filters={'ip_address_id': lambda ctx: ctx['object'].pk},
                 title=_('Application services'),
             ),
+            ObjectsTablePanel(
+                'vpn.l2vpntermination',
+                filters={'ipaddress_id': lambda ctx: ctx['object'].pk},
+                exclude_columns=['assigned_object_type', 'assigned_object', 'assigned_object_parent'],
+                title=_('L2VPN terminations'),
+            ),
         ],
     )
 

netbox/vpn/constants.py

@@ -2,6 +2,12 @@ from django.db.models import Q
 
 L2VPN_ASSIGNMENT_MODELS = Q(
     Q(app_label='dcim', model='interface') |
+    Q(app_label='ipam', model='ipaddress') |
     Q(app_label='ipam', model='vlan') |
     Q(app_label='virtualization', model='vminterface')
 )
+
+# Models which may terminate more than one L2VPN (e.g. an IP address which is the far-end PE of several pseudowires)
+L2VPN_MULTI_TERMINATION_MODELS = (
+    'ipam.ipaddress',
+)

netbox/vpn/filtersets.py

@@ -5,6 +5,7 @@ from django.utils.translation import gettext as _
 from core.models import ObjectType
 from dcim.models import Device, Interface
 from ipam.models import VLAN, IPAddress, RouteTarget
+from ipam.utils import parse_inet_addresses
 from netbox.filtersets import NetBoxModelFilterSet, OrganizationalModelFilterSet, PrimaryModelFilterSet
 from tenancy.filtersets import ContactModelFilterSet, TenancyFilterSet
 from utilities.filters import (
@@ -422,24 +423,28 @@ class L2VPNTerminationFilterSet(NetBoxModelFilterSet):
         label=_('Site (ID)'),
     )
     device = NonUniqueModelMultipleChoiceFilter(
-        field_name='interface__device__name',
+        method='filter_device',
+        field_name='name',
         queryset=Device.objects.all(),
         to_field_name='name',
         label=_('Device (name)'),
     )
     device_id = django_filters.ModelMultipleChoiceFilter(
-        field_name='interface__device',
+        method='filter_device',
+        field_name='pk',
         queryset=Device.objects.all(),
         label=_('Device (ID)'),
     )
     virtual_machine = NonUniqueModelMultipleChoiceFilter(
-        field_name='vminterface__virtual_machine__name',
+        method='filter_virtual_machine',
+        field_name='name',
         queryset=VirtualMachine.objects.all(),
         to_field_name='name',
         label=_('Virtual machine (name)'),
     )
     virtual_machine_id = django_filters.ModelMultipleChoiceFilter(
-        field_name='vminterface__virtual_machine',
+        method='filter_virtual_machine',
+        field_name='pk',
         queryset=VirtualMachine.objects.all(),
         label=_('Virtual machine (ID)'),
     )
@@ -480,6 +485,15 @@ class L2VPNTerminationFilterSet(NetBoxModelFilterSet):
         queryset=VLAN.objects.all(),
         label=_('VLAN (ID)'),
     )
+    ipaddress = MultiValueCharFilter(
+        method='filter_ipaddress',
+        label=_('IP address'),
+    )
+    ipaddress_id = django_filters.ModelMultipleChoiceFilter(
+        field_name='ipaddress',
+        queryset=IPAddress.objects.all(),
+        label=_('IP address (ID)'),
+    )
     assigned_object_type_id = django_filters.ModelMultipleChoiceFilter(
         queryset=ObjectType.objects.all(),
         distinct=False,
@@ -508,17 +522,48 @@ class L2VPNTerminationFilterSet(NetBoxModelFilterSet):
             Q(
                 Q(**{'vlan__site__{}__in'.format(name): value}) |
                 Q(**{'interface__device__site__{}__in'.format(name): value}) |
-                Q(**{'vminterface__virtual_machine__site__{}__in'.format(name): value})
+                Q(**{'vminterface__virtual_machine__site__{}__in'.format(name): value}) |
+                Q(**{'ipaddress__interface__device__site__{}__in'.format(name): value}) |
+                Q(**{'ipaddress__vminterface__virtual_machine__site__{}__in'.format(name): value})
             )
         )
         return qs
 
+    def filter_ipaddress(self, queryset, name, value):
+        # Match an address given without a mask regardless of the IP address's mask. Discard any invalid addresses.
+        value = parse_inet_addresses(value)
+        if not value:
+            return queryset.none()
+        return queryset.filter(ipaddress__address__net_in=value)
+
+    def filter_device(self, queryset, name, value):
+        if not value:
+            return queryset
+        # Match on the name (rather than the object) when filtering by name, as devices may share a name
+        values = [getattr(device, name) for device in value]
+        return queryset.filter(
+            Q(**{f'interface__device__{name}__in': values}) |
+            Q(**{f'ipaddress__interface__device__{name}__in': values})
+        )
+
+    def filter_virtual_machine(self, queryset, name, value):
+        if not value:
+            return queryset
+        # Match on the name (rather than the object) when filtering by name, as virtual machines may share a name
+        values = [getattr(vm, name) for vm in value]
+        return queryset.filter(
+            Q(**{f'vminterface__virtual_machine__{name}__in': values}) |
+            Q(**{f'ipaddress__vminterface__virtual_machine__{name}__in': values})
+        )
+
     def filter_region(self, queryset, name, value):
         qs = queryset.filter(
             Q(
                 Q(**{'vlan__site__region__{}__in'.format(name): value}) |
                 Q(**{'interface__device__site__region__{}__in'.format(name): value}) |
-                Q(**{'vminterface__virtual_machine__site__region__{}__in'.format(name): value})
+                Q(**{'vminterface__virtual_machine__site__region__{}__in'.format(name): value}) |
+                Q(**{'ipaddress__interface__device__site__region__{}__in'.format(name): value}) |
+                Q(**{'ipaddress__vminterface__virtual_machine__site__region__{}__in'.format(name): value})
             )
         )
         return qs

netbox/vpn/forms/bulk_import.py

@@ -2,7 +2,7 @@ from django.core.exceptions import ValidationError
 from django.utils.translation import gettext_lazy as _
 
 from dcim.models import Device, Interface
-from ipam.models import VLAN, IPAddress
+from ipam.models import VLAN, VRF, IPAddress
 from netbox.forms import NetBoxModelImportForm, OrganizationalModelImportForm, PrimaryModelImportForm
 from tenancy.models import Tenant
 from utilities.forms.fields import CSVChoiceField, CSVModelChoiceField, CSVModelMultipleChoiceField
@@ -289,14 +289,14 @@ class L2VPNTerminationImportForm(NetBoxModelImportForm):
         queryset=Device.objects.all(),
         required=False,
         to_field_name='name',
-        help_text=_('Parent device (for interface)')
+        help_text=_('Parent device (for interface or IP address)')
     )
     virtual_machine = CSVModelChoiceField(
         label=_('Virtual machine'),
         queryset=VirtualMachine.objects.all(),
         required=False,
         to_field_name='name',
-        help_text=_('Parent virtual machine (for interface)')
+        help_text=_('Parent virtual machine (for interface or IP address)')
     )
     interface = CSVModelChoiceField(
         label=_('Interface'),
@@ -312,36 +312,72 @@ class L2VPNTerminationImportForm(NetBoxModelImportForm):
         to_field_name='name',
         help_text=_('Assigned VLAN')
     )
+    ipaddress = CSVModelChoiceField(
+        label=_('IP address'),
+        queryset=IPAddress.objects.all(),
+        required=False,
+        to_field_name='address',
+        help_text=_('Assigned IP address, with mask (e.g. the far-end PE of a pseudowire)')
+    )
+    ipaddress_vrf = CSVModelChoiceField(
+        label=_('IP address VRF'),
+        queryset=VRF.objects.all(),
+        required=False,
+        to_field_name='name',
+        help_text=_('VRF of the assigned IP address')
+    )
 
     class Meta:
         model = L2VPNTermination
-        fields = ('l2vpn', 'device', 'virtual_machine', 'interface', 'vlan', 'tags')
+        fields = ('l2vpn', 'device', 'virtual_machine', 'interface', 'vlan', 'ipaddress', 'ipaddress_vrf', 'tags')
 
     def __init__(self, data=None, *args, **kwargs):
         super().__init__(data, *args, **kwargs)
 
         if data:
 
-            # Limit interface queryset by device or VM
+            # Limit interface and IP address querysets by device or VM
             if data.get('device'):
                 self.fields['interface'].queryset = Interface.objects.filter(
                     **{f"device__{self.fields['device'].to_field_name}": data['device']}
                 )
+                self.fields['ipaddress'].queryset = IPAddress.objects.filter(
+                    **{f"interface__device__{self.fields['device'].to_field_name}": data['device']}
+                )
             elif data.get('virtual_machine'):
                 self.fields['interface'].queryset = VMInterface.objects.filter(
                     **{f"virtual_machine__{self.fields['virtual_machine'].to_field_name}": data['virtual_machine']}
                 )
+                self.fields['ipaddress'].queryset = IPAddress.objects.filter(
+                    **{
+                        f"vminterface__virtual_machine__{self.fields['virtual_machine'].to_field_name}":
+                            data['virtual_machine']
+                    }
+                )
+
+            # Limit IP address queryset by VRF
+            if data.get('ipaddress_vrf'):
+                self.fields['ipaddress'].queryset = self.fields['ipaddress'].queryset.filter(
+                    **{f"vrf__{self.fields['ipaddress_vrf'].to_field_name}": data['ipaddress_vrf']}
+                )
 
     def clean(self):
         super().clean()
 
         if self.cleaned_data.get('device') and self.cleaned_data.get('virtual_machine'):
             raise ValidationError(_('Cannot import device and VM interface terminations simultaneously.'))
-        if not self.instance and not (self.cleaned_data.get('interface') or self.cleaned_data.get('vlan')):
-            raise ValidationError(_('Each termination must specify either an interface or a VLAN.'))
-        if self.cleaned_data.get('interface') and self.cleaned_data.get('vlan'):
-            raise ValidationError(_('Cannot assign both an interface and a VLAN.'))
-
-        # if this is an update we might not have interface or vlan in the form data
-        if self.cleaned_data.get('interface') or self.cleaned_data.get('vlan'):
-            self.instance.assigned_object = self.cleaned_data.get('interface') or self.cleaned_data.get('vlan')
+        assigned_objects = [
+            obj for obj in (
+                self.cleaned_data.get('interface'),
+                self.cleaned_data.get('vlan'),
+                self.cleaned_data.get('ipaddress'),
+            ) if obj
+        ]
+        if not self.instance.pk and not assigned_objects:
+            raise ValidationError(_('Each termination must specify an interface, a VLAN, or an IP address.'))
+        if len(assigned_objects) > 1:
+            raise ValidationError(_('Cannot assign more than one of an interface, a VLAN, and an IP address.'))
+
+        # if this is an update we might not have an assigned object in the form data
+        if assigned_objects:
+            self.instance.assigned_object = assigned_objects[0]

netbox/vpn/forms/filtersets.py

@@ -3,7 +3,7 @@ from django.contrib.contenttypes.models import ContentType
 from django.utils.translation import gettext as _
 
 from dcim.models import Device, Region, Site
-from ipam.models import VLAN, RouteTarget
+from ipam.models import VLAN, IPAddress, RouteTarget
 from netbox.forms import NetBoxModelFilterSetForm, OrganizationalModelFilterSetForm, PrimaryModelFilterSetForm
 from tenancy.forms import ContactModelFilterForm, TenancyFilterForm
 from utilities.forms.fields import (
@@ -259,7 +259,7 @@ class L2VPNTerminationFilterForm(NetBoxModelFilterSetForm):
         FieldSet('filter_id', 'tag', 'l2vpn_id'),
         FieldSet(
             'assigned_object_type_id', 'region_id', 'site_id', 'device_id', 'virtual_machine_id', 'vlan_id',
-            name=_('Assigned Object')
+            'ipaddress_id', name=_('Assigned Object')
         ),
     )
     l2vpn_id = DynamicModelChoiceField(
@@ -314,4 +314,9 @@ class L2VPNTerminationFilterForm(NetBoxModelFilterSetForm):
         },
         label=_('Virtual Machine')
     )
+    ipaddress_id = DynamicModelMultipleChoiceField(
+        queryset=IPAddress.objects.all(),
+        required=False,
+        label=_('IP Address')
+    )
     tag = TagFilterField(model)

netbox/vpn/forms/model_forms.py

@@ -506,6 +506,13 @@ class L2VPNTerminationForm(NetBoxModelForm):
         selector=True,
         label=_('Interface')
     )
+    ipaddress = DynamicModelChoiceField(
+        queryset=IPAddress.objects.all(),
+        required=False,
+        selector=True,
+        label=_('IP Address'),
+        help_text=_('For example, the far-end PE of a pseudowire')
+    )
 
     fieldsets = (
         FieldSet(
@@ -514,6 +521,7 @@ class L2VPNTerminationForm(NetBoxModelForm):
                 FieldSet('vlan', name=_('VLAN')),
                 FieldSet('interface', name=_('Device')),
                 FieldSet('vminterface', name=_('Virtual Machine')),
+                FieldSet('ipaddress', name=_('IP Address')),
             ),
             'tags',
         ),
@@ -534,6 +542,8 @@ class L2VPNTerminationForm(NetBoxModelForm):
                 initial['vlan'] = instance.assigned_object
             elif type(instance.assigned_object) is VMInterface:
                 initial['vminterface'] = instance.assigned_object
+            elif type(instance.assigned_object) is IPAddress:
+                initial['ipaddress'] = instance.assigned_object
             kwargs['initial'] = initial
 
         super().__init__(*args, **kwargs)
@@ -544,10 +554,13 @@ class L2VPNTerminationForm(NetBoxModelForm):
         interface = self.cleaned_data.get('interface')
         vminterface = self.cleaned_data.get('vminterface')
         vlan = self.cleaned_data.get('vlan')
+        ipaddress = self.cleaned_data.get('ipaddress')
 
-        if not (interface or vminterface or vlan):
-            raise ValidationError(_('A termination must specify an interface or VLAN.'))
-        if len([x for x in (interface, vminterface, vlan) if x]) > 1:
-            raise ValidationError(_('A termination can only have one terminating object (an interface or VLAN).'))
+        if not (interface or vminterface or vlan or ipaddress):
+            raise ValidationError(_('A termination must specify an interface, VLAN, or IP address.'))
+        if len([x for x in (interface, vminterface, vlan, ipaddress) if x]) > 1:
+            raise ValidationError(
+                _('A termination can only have one terminating object (an interface, VLAN, or IP address).')
+            )
 
-        self.instance.assigned_object = interface or vminterface or vlan
+        self.instance.assigned_object = interface or vminterface or vlan or ipaddress

netbox/vpn/graphql/types.py

@@ -5,7 +5,7 @@ import strawberry_django
 
 from dcim.models import Interface
 from extras.graphql.mixins import ContactsMixin, CustomFieldsMixin, TagsMixin
-from ipam.models import VLAN
+from ipam.models import VLAN, IPAddress
 from netbox.graphql.optimization import build_gfk_prefetch
 from netbox.graphql.types import (
     NetBoxObjectType,
@@ -162,12 +162,14 @@ class L2VPNTerminationType(NetBoxObjectType):
                 Interface,
                 VMInterface,
                 VLAN,
+                IPAddress,
             ],
         ),
         only=['assigned_object_type', 'assigned_object_id'],
     )
     def assigned_object(self) -> Annotated[
         Annotated['InterfaceType', strawberry.lazy('dcim.graphql.types')]
+        | Annotated['IPAddressType', strawberry.lazy('ipam.graphql.types')]
         | Annotated['VLANType', strawberry.lazy('ipam.graphql.types')]
         | Annotated['VMInterfaceType', strawberry.lazy('virtualization.graphql.types')],
         strawberry.union('L2VPNAssignmentType'),

netbox/vpn/migrations/0014_l2vpntermination_unique_per_l2vpn.py

@@ -0,0 +1,26 @@
+# Generated by Django 6.1.1 on 2026-11-02 19:40
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('contenttypes', '0002_remove_content_type_name'),
+        ('extras', '0145_taggeditem_object_id_bigint'),
+        ('vpn', '0013_consolidate_unique_constraints'),
+    ]
+
+    operations = [
+        migrations.RemoveConstraint(
+            model_name='l2vpntermination',
+            name='vpn_l2vpntermination_assigned_object',
+        ),
+        migrations.AddConstraint(
+            model_name='l2vpntermination',
+            constraint=models.UniqueConstraint(
+                fields=('l2vpn', 'assigned_object_type', 'assigned_object_id'),
+                name='vpn_l2vpntermination_l2vpn_assigned_object',
+            ),
+        ),
+    ]

netbox/vpn/models/l2vpn.py

@@ -5,9 +5,11 @@ from django.utils.functional import cached_property
 from django.utils.translation import gettext_lazy as _
 
 from core.models import ObjectType
+from ipam.models import VLAN, IPAddress
 from netbox.models import NetBoxModel, PrimaryModel
 from netbox.models.features import ContactsMixin
 from vpn.choices import L2VPNStatusChoices, L2VPNTypeChoices
+from vpn.constants import L2VPN_MULTI_TERMINATION_MODELS
 
 __all__ = (
     'L2VPN',
@@ -109,8 +111,8 @@ class L2VPNTermination(NetBoxModel):
         ordering = ('l2vpn',)
         constraints = (
             models.UniqueConstraint(
-                fields=('assigned_object_type', 'assigned_object_id'),
-                name='vpn_l2vpntermination_assigned_object'
+                fields=('l2vpn', 'assigned_object_type', 'assigned_object_id'),
+                name='vpn_l2vpntermination_l2vpn_assigned_object'
             ),
         )
         verbose_name = _('L2VPN termination')
@@ -126,8 +128,18 @@ class L2VPNTermination(NetBoxModel):
         if self.assigned_object:
             obj_id = self.assigned_object.pk
             obj_type = ObjectType.objects.get_for_model(self.assigned_object)
-            terminations = L2VPNTermination.objects.filter(assigned_object_id=obj_id, assigned_object_type=obj_type)
-            if terminations.exclude(pk=self.pk).exists():
+            terminations = L2VPNTermination.objects.filter(
+                assigned_object_id=obj_id, assigned_object_type=obj_type
+            ).exclude(pk=self.pk)
+            # Some objects may terminate many L2VPNs, but each only once. Others may terminate only one L2VPN.
+            if self.assigned_object._meta.label_lower in L2VPN_MULTI_TERMINATION_MODELS:
+                if terminations.filter(l2vpn_id=self.l2vpn_id).exists():
+                    raise ValidationError(
+                        _('{assigned_object} already terminates {l2vpn}').format(
+                            assigned_object=self.assigned_object, l2vpn=self.l2vpn
+                        )
+                    )
+            elif terminations.exists():
                 raise ValidationError(
                     _('L2VPN Termination already assigned ({assigned_object})').format(
                         assigned_object=self.assigned_object
@@ -147,15 +159,16 @@ class L2VPNTermination(NetBoxModel):
 
     @property
     def assigned_object_parent(self):
-        obj_type = ObjectType.objects.get_for_model(self.assigned_object)
-        if obj_type.model == 'vminterface':
-            return self.assigned_object.virtual_machine
-        if obj_type.model == 'interface':
-            return self.assigned_object.device
-        if obj_type.model == 'vminterface':
-            return self.assigned_object.virtual_machine
-        return None
+        obj = self.assigned_object
+        if isinstance(obj, IPAddress):
+            # The interface the IP address is assigned to, if any
+            obj = obj.assigned_object
+        return getattr(obj, 'parent_object', None)
 
     @property
     def assigned_object_site(self):
-        return self.assigned_object_parent.site
+        if isinstance(self.assigned_object, VLAN):
+            return self.assigned_object.site
+        if parent := self.assigned_object_parent:
+            return parent.site
+        return None

netbox/vpn/tests/query_counts.json

@@ -12,7 +12,7 @@
   "l2vpn:api_list_objects": 14,
   "l2vpn:list_objects_with_permission": 16,
   "l2vpntermination:api_list_objects": 15,
-  "l2vpntermination:list_objects_with_permission": 21,
+  "l2vpntermination:list_objects_with_permission": 20,
   "tunnel:api_list_objects": 13,
   "tunnel:list_objects_with_permission": 17,
   "tunnelgroup:api_list_objects": 12,

netbox/vpn/tests/test_api.py

@@ -1,9 +1,11 @@
+import json
+
 from django.urls import reverse
 from rest_framework import status
 
 from dcim.choices import InterfaceTypeChoices
 from dcim.models import Interface
-from ipam.models import VLAN
+from ipam.models import VLAN, IPAddress
 from utilities.testing import APITestCase, APIViewTestCases, create_test_device
 from vpn.choices import *
 from vpn.models import *
@@ -676,10 +678,12 @@ class L2VPNTerminationTestCase(APIViewTestCases.APIViewTestCase):
         )
         L2VPN.objects.bulk_create(l2vpns)
 
+        ip_address = IPAddress.objects.create(address='192.0.2.1/32')
+
         l2vpnterminations = (
             L2VPNTermination(l2vpn=l2vpns[0], assigned_object=vlans[0]),
             L2VPNTermination(l2vpn=l2vpns[0], assigned_object=vlans[1]),
-            L2VPNTermination(l2vpn=l2vpns[0], assigned_object=vlans[2])
+            L2VPNTermination(l2vpn=l2vpns[1], assigned_object=ip_address)
         )
         L2VPNTermination.objects.bulk_create(l2vpnterminations)
 
@@ -695,9 +699,10 @@ class L2VPNTerminationTestCase(APIViewTestCases.APIViewTestCase):
                 'assigned_object_id': vlans[4].pk,
             },
             {
-                'l2vpn': l2vpns[0].pk,
-                'assigned_object_type': 'ipam.vlan',
-                'assigned_object_id': vlans[5].pk,
+                # The same IP address may terminate another L2VPN
+                'l2vpn': l2vpns[2].pk,
+                'assigned_object_type': 'ipam.ipaddress',
+                'assigned_object_id': ip_address.pk,
             },
         ]
 
@@ -708,3 +713,17 @@ class L2VPNTerminationTestCase(APIViewTestCases.APIViewTestCase):
         cls.bulk_update_invalid_data = {
             'l2vpn': 99999,
         }
+
+    def test_graphql_ip_address_l2vpn_terminations_filter(self):
+        """IP addresses can be filtered over GraphQL by the L2VPNs they terminate."""
+        self.add_permissions('ipam.view_ipaddress')
+        IPAddress.objects.create(address='192.0.2.2/32')
+        l2vpn = L2VPN.objects.get(name='L2VPN 2')
+        query = (
+            f'{{ ip_address_list(filters: {{l2vpn_terminations: {{l2vpn_id: "{l2vpn.pk}"}}}}) {{ address }} }}'
+        )
+        response = self.client.post(reverse('graphql'), data={'query': query}, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        data = json.loads(response.content)
+        self.assertNotIn('errors', data)
+        self.assertEqual([ip['address'] for ip in data['data']['ip_address_list']], ['192.0.2.1/32'])

netbox/vpn/tests/test_filtersets.py

@@ -900,6 +900,13 @@ class L2VPNTerminationTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
         )
         L2VPN.objects.bulk_create(l2vpns)
 
+        ip_addresses = (
+            IPAddress(address='192.0.2.1/32', assigned_object=interfaces[0]),
+            IPAddress(address='192.0.2.2/32', assigned_object=vminterfaces[0]),
+            IPAddress(address='192.0.2.3/32'),
+        )
+        IPAddress.objects.bulk_create(ip_addresses)
+
         l2vpnterminations = (
             L2VPNTermination(l2vpn=l2vpns[0], assigned_object=vlans[0]),
             L2VPNTermination(l2vpn=l2vpns[1], assigned_object=vlans[1]),
@@ -910,6 +917,8 @@ class L2VPNTerminationTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
             L2VPNTermination(l2vpn=l2vpns[0], assigned_object=vminterfaces[0]),
             L2VPNTermination(l2vpn=l2vpns[1], assigned_object=vminterfaces[1]),
             L2VPNTermination(l2vpn=l2vpns[2], assigned_object=vminterfaces[2]),
+            L2VPNTermination(l2vpn=l2vpns[2], assigned_object=ip_addresses[0]),
+            L2VPNTermination(l2vpn=l2vpns[2], assigned_object=ip_addresses[1]),
         )
         L2VPNTermination.objects.bulk_create(l2vpnterminations)
 
@@ -954,6 +963,17 @@ class L2VPNTerminationTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
         params = {'vlan': ['VLAN 1', 'VLAN 2']}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
 
+    def test_ipaddress(self):
+        ip_addresses = IPAddress.objects.all()[:2]
+        params = {'ipaddress_id': [ip_addresses[0].pk, ip_addresses[1].pk]}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+        params = {'ipaddress': ['192.0.2.1/32', '192.0.2.2/32']}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+        params = {'ipaddress': ['192.0.2.1', '192.0.2.2']}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+        params = {'ipaddress': ['invalid']}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 0)
+
     def test_vlan_shared_name(self):
         # A VLAN name shared by several groups should resolve to a single filter value
         vlan = VLAN.objects.first()
@@ -965,16 +985,16 @@ class L2VPNTerminationTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
     def test_site(self):
         site = Site.objects.all().first()
         params = {'site_id': [site.pk]}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
         params = {'site': ['site-1']}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
 
     def test_device(self):
         device = Device.objects.all().first()
         params = {'device_id': [device.pk]}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
         params = {'device': ['Device 1']}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
 
     def test_device_shared_name(self):
         # A device name shared by several sites should resolve to a single filter value
@@ -988,9 +1008,9 @@ class L2VPNTerminationTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
     def test_virtual_machine(self):
         virtual_machine = VirtualMachine.objects.all().first()
         params = {'virtual_machine_id': [virtual_machine.pk]}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
         params = {'virtual_machine': ['Virtual Machine 1']}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
 
     def test_virtual_machine_shared_name(self):
         # A virtual machine name shared by several clusters should resolve to a single filter value

netbox/vpn/tests/test_forms.py

@@ -0,0 +1,82 @@
+from django.test import TestCase
+
+from dcim.models import Interface
+from ipam.models import VRF, IPAddress
+from utilities.testing import create_test_device
+from vpn.choices import L2VPNTypeChoices
+from vpn.forms import L2VPNTerminationForm, L2VPNTerminationImportForm
+from vpn.models import L2VPN, L2VPNTermination
+
+
+class L2VPNTerminationFormTestCase(TestCase):
+
+    @classmethod
+    def setUpTestData(cls):
+        cls.ip_address = IPAddress.objects.create(address='192.0.2.1/32')
+        cls.l2vpns = (
+            L2VPN.objects.create(name='L2VPN 1', slug='l2vpn-1', type=L2VPNTypeChoices.TYPE_VPLS),
+            L2VPN.objects.create(name='L2VPN 2', slug='l2vpn-2', type=L2VPNTypeChoices.TYPE_VPLS),
+        )
+        L2VPNTermination.objects.create(l2vpn=cls.l2vpns[0], assigned_object=cls.ip_address)
+
+    def test_ipaddress_terminates_another_l2vpn(self):
+        form = L2VPNTerminationForm(data={'l2vpn': self.l2vpns[1].pk, 'ipaddress': self.ip_address.pk})
+        self.assertTrue(form.is_valid(), form.errors)
+
+    def test_duplicate_ipaddress_termination(self):
+        form = L2VPNTerminationForm(data={'l2vpn': self.l2vpns[0].pk, 'ipaddress': self.ip_address.pk})
+        self.assertFalse(form.is_valid())
+        self.assertIn('already terminates', str(form.errors))
+
+
+class L2VPNTerminationImportFormTestCase(TestCase):
+
+    @classmethod
+    def setUpTestData(cls):
+        L2VPN.objects.create(name='L2VPN 1', slug='l2vpn-1', type=L2VPNTypeChoices.TYPE_VPLS)
+
+        # The same address exists globally, in a VRF, and on a device's interface in another VRF
+        cls.vrfs = (
+            VRF.objects.create(name='VRF 1'),
+            VRF.objects.create(name='VRF 2'),
+        )
+        cls.device = create_test_device('Device 1')
+        interface = Interface.objects.create(device=cls.device, name='Interface 1', type='1000baset')
+        cls.ip_addresses = (
+            IPAddress.objects.create(address='192.0.2.1/32'),
+            IPAddress.objects.create(address='192.0.2.1/32', vrf=cls.vrfs[0]),
+            IPAddress.objects.create(address='192.0.2.1/32', vrf=cls.vrfs[1], assigned_object=interface),
+        )
+
+    def test_missing_assigned_object(self):
+        form = L2VPNTerminationImportForm(data={'l2vpn': 'L2VPN 1'}, instance=L2VPNTermination())
+        self.assertFalse(form.is_valid())
+        self.assertIn('must specify an interface, a VLAN, or an IP address', str(form.errors))
+
+    def get_assigned_object(self, data):
+        form = L2VPNTerminationImportForm(data={'l2vpn': 'L2VPN 1', **data}, instance=L2VPNTermination())
+        self.assertTrue(form.is_valid(), form.errors)
+        return form.instance.assigned_object
+
+    def test_ipaddress_ambiguous(self):
+        form = L2VPNTerminationImportForm(
+            data={'l2vpn': 'L2VPN 1', 'ipaddress': '192.0.2.1/32'}, instance=L2VPNTermination()
+        )
+        self.assertFalse(form.is_valid())
+        self.assertIn('ipaddress', form.errors)
+
+    def test_ipaddress_vrf(self):
+        assigned_object = self.get_assigned_object({'ipaddress': '192.0.2.1/32', 'ipaddress_vrf': 'VRF 1'})
+        self.assertEqual(assigned_object, self.ip_addresses[1])
+
+    def test_ipaddress_device(self):
+        assigned_object = self.get_assigned_object({'ipaddress': '192.0.2.1/32', 'device': 'Device 1'})
+        self.assertEqual(assigned_object, self.ip_addresses[2])
+
+    def test_ipaddress_not_on_device(self):
+        form = L2VPNTerminationImportForm(
+            data={'l2vpn': 'L2VPN 1', 'ipaddress': '192.0.2.1/32', 'ipaddress_vrf': 'VRF 1', 'device': 'Device 1'},
+            instance=L2VPNTermination()
+        )
+        self.assertFalse(form.is_valid())
+        self.assertIn('ipaddress', form.errors)

netbox/vpn/tests/test_models.py

@@ -2,7 +2,7 @@ from django.core.exceptions import ValidationError
 from django.test import TestCase
 
 from dcim.models import Device, DeviceRole, DeviceType, Interface, Manufacturer, Site
-from ipam.models import VLAN
+from ipam.models import VLAN, IPAddress
 from vpn.models import *
 
 
@@ -77,3 +77,50 @@ class L2VPNTerminationTestCase(TestCase):
         L2VPNTermination.objects.create(l2vpn=l2vpn, assigned_object=vlan)
         duplicate = L2VPNTermination(l2vpn=l2vpn, assigned_object=vlan)
         self.assertRaises(ValidationError, duplicate.clean)
+
+    def test_ipaddress_terminates_multiple_l2vpns(self):
+        ip_address = IPAddress.objects.create(address='192.0.2.1/32')
+        l2vpns = L2VPN.objects.all()[:2]
+
+        L2VPNTermination.objects.create(l2vpn=l2vpns[0], assigned_object=ip_address)
+        termination = L2VPNTermination(l2vpn=l2vpns[1], assigned_object=ip_address)
+        termination.full_clean()
+        termination.save()
+
+        self.assertEqual(ip_address.l2vpn_terminations.count(), 2)
+
+    def test_duplicate_ipaddress_terminations(self):
+        ip_address = IPAddress.objects.create(address='192.0.2.1/32')
+        l2vpn = L2VPN.objects.first()
+
+        L2VPNTermination.objects.create(l2vpn=l2vpn, assigned_object=ip_address)
+        duplicate = L2VPNTermination(l2vpn=l2vpn, assigned_object=ip_address)
+        self.assertRaises(ValidationError, duplicate.clean)
+
+    def test_ipaddress_termination_parent(self):
+        interface = Interface.objects.first()
+        assigned_ip = IPAddress.objects.create(address='192.0.2.1/32', assigned_object=interface)
+        unassigned_ip = IPAddress.objects.create(address='192.0.2.2/32')
+        l2vpn = L2VPN.objects.first()
+
+        termination = L2VPNTermination.objects.create(l2vpn=l2vpn, assigned_object=assigned_ip)
+        self.assertEqual(termination.assigned_object_parent, interface.device)
+        self.assertEqual(termination.assigned_object_site, interface.device.site)
+
+        termination = L2VPNTermination.objects.create(l2vpn=l2vpn, assigned_object=unassigned_ip)
+        self.assertIsNone(termination.assigned_object_parent)
+        self.assertIsNone(termination.assigned_object_site)
+
+    def test_vlan_termination_site(self):
+        site = Site.objects.first()
+        vlan = VLAN.objects.create(name='VLAN 8', vid=658, site=site)
+        termination = L2VPNTermination.objects.create(l2vpn=L2VPN.objects.first(), assigned_object=vlan)
+        self.assertIsNone(termination.assigned_object_parent)
+        self.assertEqual(termination.assigned_object_site, site)
+
+    def test_delete_ipaddress_deletes_terminations(self):
+        ip_address = IPAddress.objects.create(address='192.0.2.1/32')
+        termination = L2VPNTermination.objects.create(l2vpn=L2VPN.objects.first(), assigned_object=ip_address)
+
+        ip_address.delete()
+        self.assertFalse(L2VPNTermination.objects.filter(pk=termination.pk).exists())

netbox/vpn/tests/test_views.py

@@ -1,7 +1,12 @@
+from django.db import connection
+from django.test.utils import CaptureQueriesContext
+from django.urls import reverse
+
 from dcim.choices import InterfaceTypeChoices
 from dcim.models import Interface
-from ipam.models import VLAN, RouteTarget
-from utilities.testing import ViewTestCases, create_tags, create_test_device
+from ipam.models import VLAN, IPAddress, RouteTarget
+from utilities.testing import ViewTestCases, create_tags, create_test_device, create_test_virtualmachine
+from virtualization.models import VMInterface
 from vpn.choices import *
 from vpn.models import *
 
@@ -664,11 +669,13 @@ class L2VPNTerminationTestCase(
             'interface': interface.pk,
         }
 
+        IPAddress.objects.create(address='192.0.2.1/32')
+
         cls.csv_data = (
-            "l2vpn,vlan",
-            "L2VPN 1,Vlan 4",
-            "L2VPN 1,Vlan 5",
-            "L2VPN 1,Vlan 6",
+            "l2vpn,vlan,ipaddress",
+            "L2VPN 1,Vlan 4,",
+            "L2VPN 1,Vlan 5,",
+            "L2VPN 1,,192.0.2.1/32",
         )
 
         cls.csv_update_data = (
@@ -684,6 +691,43 @@ class L2VPNTerminationTestCase(
     def test_bulk_update_objects_with_permission(self):
         pass
 
+    def test_list_objects_query_count(self):
+        """
+        Listing terminations should not need extra queries for each row to display its parent and site.
+        """
+        self.add_permissions('vpn.view_l2vpntermination')
+        self.user.config.set(
+            'tables.L2VPNTerminationTable.columns',
+            ['l2vpn', 'assigned_object_type', 'assigned_object', 'assigned_object_parent', 'assigned_object_site'],
+            commit=True
+        )
+        device = create_test_device('Device 2')
+        virtual_machine = create_test_virtualmachine('Virtual Machine 1')
+
+        def get_queries(count):
+            l2vpn = L2VPN.objects.create(
+                name=f'L2VPN {count}x', slug=f'l2vpn-{count}x', type=L2VPNTypeChoices.TYPE_VPLS
+            )
+            for i in range(count):
+                interface = Interface.objects.create(device=device, name=f'{count}x {i}', type='1000baset')
+                vminterface = VMInterface.objects.create(virtual_machine=virtual_machine, name=f'{count}x {i}')
+                vlan = VLAN.objects.create(name=f'VLAN {count}x {i}', vid=100 * count + i, site=device.site)
+                for obj in (
+                    interface,
+                    vminterface,
+                    vlan,
+                    IPAddress.objects.create(address=f'198.51.{count}.{i}/32', assigned_object=interface),
+                    IPAddress.objects.create(address=f'203.0.{count}.{i}/32', assigned_object=vminterface),
+                ):
+                    L2VPNTermination.objects.create(l2vpn=l2vpn, assigned_object=obj)
+            with CaptureQueriesContext(connection) as queries:
+                response = self.client.get(f"{reverse('vpn:l2vpntermination_list')}?l2vpn_id={l2vpn.pk}")
+            self.assertHttpStatus(response, 200)
+            self.assertContains(response, device.site.name)
+            return len(queries)
+
+        self.assertEqual(get_queries(1), get_queries(3))
+
     #
     # Custom assertions
     #

netbox/vpn/views.py

@@ -1,6 +1,9 @@
+from django.contrib.contenttypes.prefetch import GenericPrefetch
 from django.utils.translation import gettext_lazy as _
 
+from dcim.models import Interface
 from extras.ui.panels import CustomFieldsPanel, TagsPanel
+from ipam.models import VLAN, IPAddress
 from ipam.tables import RouteTargetTable
 from netbox.object_actions import AddObject, BulkDelete, BulkEdit, BulkExport, BulkImport
 from netbox.ui import actions, layout
@@ -14,6 +17,7 @@ from netbox.ui.panels import (
 from netbox.views import generic
 from utilities.query import count_related
 from utilities.views import GetRelatedModelsMixin, register_model_view
+from virtualization.models import VMInterface
 
 from . import filtersets, forms, tables
 from .models import *
@@ -761,7 +765,20 @@ class L2VPNBulkDeleteView(generic.BulkDeleteView):
 
 @register_model_view(L2VPNTermination, 'list', path='', detail=False)
 class L2VPNTerminationListView(generic.ObjectListView):
-    queryset = L2VPNTermination.objects.all()
+    # Fetch each terminating object's parent and site, which the table may display
+    queryset = L2VPNTermination.objects.prefetch_related(
+        GenericPrefetch('assigned_object', [
+            Interface.objects.select_related('device__site'),
+            VMInterface.objects.select_related('virtual_machine__site'),
+            VLAN.objects.select_related('site'),
+            IPAddress.objects.prefetch_related(
+                GenericPrefetch('assigned_object', [
+                    Interface.objects.select_related('device__site'),
+                    VMInterface.objects.select_related('virtual_machine__site'),
+                ])
+            ),
+        ])
+    )
     table = tables.L2VPNTerminationTable
     filterset = filtersets.L2VPNTerminationFilterSet
     filterset_form = forms.L2VPNTerminationFilterForm