NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-07-14 01:41:22 -06:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

Fixes #2514: Prevent new connections to already connected interfaces

Browse Source
This commit is contained in:
Jeremy Stretch 2018-10-16 16:41:12 -04:00
parent 409a9256a1
commit 0bb5d229e8
5 changed files with 87 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGELOG.md
View File

@ -2,6 +2,7 @@ v2.4.7 (FUTURE)
## Bug Fixes
* [#2514](https://github.com/digitalocean/netbox/issues/2514) - Prevent new connections to already connected interfaces
* [#2515](https://github.com/digitalocean/netbox/issues/2515) - Only use django-rq admin tmeplate if webhooks are enabled
---

59
netbox/dcim/api/serializers.py
View File

@ -472,10 +472,14 @@ class ConsoleServerPortSerializer(TaggitSerializer, ValidatedModelSerializer):
class NestedConsoleServerPortSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:consoleserverport-detail')
device = NestedDeviceSerializer(read_only=True)
is_connected = serializers.SerializerMethodField(read_only=True)
class Meta:
model = ConsoleServerPort
fields = ['id', 'url', 'device', 'name']
fields = ['id', 'url', 'device', 'name', 'is_connected']
def get_is_connected(self, obj):
return hasattr(obj, 'connected_console') and obj.connected_console is not None
#
@ -495,10 +499,14 @@ class ConsolePortSerializer(TaggitSerializer, ValidatedModelSerializer):
class NestedConsolePortSerializer(TaggitSerializer, ValidatedModelSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:consoleport-detail')
device = NestedDeviceSerializer(read_only=True)
is_connected = serializers.SerializerMethodField(read_only=True)
class Meta:
model = ConsolePort
fields = ['id', 'url', 'device', 'name']
fields = ['id', 'url', 'device', 'name', 'is_connected']
def get_is_connected(self, obj):
return obj.cs_port is not None
#
@ -518,10 +526,14 @@ class PowerOutletSerializer(TaggitSerializer, ValidatedModelSerializer):
class NestedPowerOutletSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:poweroutlet-detail')
device = NestedDeviceSerializer(read_only=True)
is_connected = serializers.SerializerMethodField(read_only=True)
class Meta:
model = PowerOutlet
fields = ['id', 'url', 'device', 'name']
fields = ['id', 'url', 'device', 'name', 'is_connected']
def get_is_connected(self, obj):
return hasattr(obj, 'connected_port') and obj.connected_port is not None
#
@ -541,23 +553,43 @@ class PowerPortSerializer(TaggitSerializer, ValidatedModelSerializer):
class NestedPowerPortSerializer(TaggitSerializer, ValidatedModelSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:powerport-detail')
device = NestedDeviceSerializer(read_only=True)
is_connected = serializers.SerializerMethodField(read_only=True)
class Meta:
model = PowerPort
fields = ['id', 'url', 'device', 'name']
fields = ['id', 'url', 'device', 'name', 'is_connected']
def get_is_connected(self, obj):
return obj.power_outlet is not None
#
# Interfaces
#
class NestedInterfaceSerializer(WritableNestedSerializer):
class IsConnectedMixin(object):
"""
Provide a method for setting is_connected on Interface serializers.
"""
def get_is_connected(self, obj):
"""
Return True if the interface has a connected interface or circuit.
"""
if obj.connection:
return True
if hasattr(obj, 'circuit_termination') and obj.circuit_termination is not None:
return True
return False
class NestedInterfaceSerializer(IsConnectedMixin, WritableNestedSerializer):
device = NestedDeviceSerializer(read_only=True)
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:interface-detail')
is_connected = serializers.SerializerMethodField(read_only=True)
class Meta:
model = Interface
fields = ['id', 'url', 'device', 'name']
fields = ['id', 'url', 'device', 'name', 'is_connected']
class InterfaceNestedCircuitSerializer(serializers.ModelSerializer):
@ -587,7 +619,7 @@ class InterfaceVLANSerializer(WritableNestedSerializer):
fields = ['id', 'url', 'vid', 'name', 'display_name']
class InterfaceSerializer(TaggitSerializer, ValidatedModelSerializer):
class InterfaceSerializer(TaggitSerializer, IsConnectedMixin, ValidatedModelSerializer):
device = NestedDeviceSerializer()
form_factor = ChoiceField(choices=IFACE_FF_CHOICES, required=False)
lag = NestedInterfaceSerializer(required=False, allow_null=True)
@ -631,19 +663,6 @@ class InterfaceSerializer(TaggitSerializer, ValidatedModelSerializer):
return super(InterfaceSerializer, self).validate(data)
def get_is_connected(self, obj):
"""
Return True if the interface has a connected interface or circuit termination.
"""
if obj.connection:
return True
try:
circuit_termination = obj.circuit_termination
return True
except CircuitTermination.DoesNotExist:
pass
return False
def get_interface_connection(self, obj):
if obj.connection:
context = {

8
netbox/dcim/forms.py
View File

@ -1328,7 +1328,7 @@ class ConsolePortConnectionForm(BootstrapMixin, ChainedFieldsMixin, forms.ModelF
label='Port',
widget=APISelect(
api_url='/api/dcim/console-server-ports/?device_id={{console_server}}',
disabled_indicator='connected_console',
disabled_indicator='is_connected',
)
)
@ -1419,7 +1419,7 @@ class ConsoleServerPortConnectionForm(BootstrapMixin, ChainedFieldsMixin, forms.
label='Port',
widget=APISelect(
api_url='/api/dcim/console-ports/?device_id={{device}}',
disabled_indicator='cs_port'
disabled_indicator='is_connected'
)
)
connection_status = forms.BooleanField(
@ -1597,7 +1597,7 @@ class PowerPortConnectionForm(BootstrapMixin, ChainedFieldsMixin, forms.ModelFor
label='Outlet',
widget=APISelect(
api_url='/api/dcim/power-outlets/?device_id={{pdu}}',
disabled_indicator='connected_port'
disabled_indicator='is_connected'
)
)
@ -1688,7 +1688,7 @@ class PowerOutletConnectionForm(BootstrapMixin, ChainedFieldsMixin, forms.Form):
label='Port',
widget=APISelect(
api_url='/api/dcim/power-ports/?device_id={{device}}',
disabled_indicator='power_outlet'
disabled_indicator='is_connected'
)
)
connection_status = forms.BooleanField(

57
netbox/dcim/models.py
View File

@ -2035,25 +2035,44 @@ class InterfaceConnection(models.Model):
csv_headers = ['device_a', 'interface_a', 'device_b', 'interface_b', 'connection_status']
def clean(self):
try:
if self.interface_a == self.interface_b:
raise ValidationError({
'interface_b': "Cannot connect an interface to itself."
})
if self.interface_a.form_factor in NONCONNECTABLE_IFACE_TYPES:
raise ValidationError({
'interface_a': '{} is not a connectable interface type.'.format(
self.interface_a.get_form_factor_display()
)
})
if self.interface_b.form_factor in NONCONNECTABLE_IFACE_TYPES:
raise ValidationError({
'interface_b': '{} is not a connectable interface type.'.format(
self.interface_b.get_form_factor_display()
)
})
except ObjectDoesNotExist:
pass
# An interface cannot be connected to itself
if self.interface_a == self.interface_b:
raise ValidationError({
'interface_b': "Cannot connect an interface to itself."
})
# Only connectable interface types are permitted
if self.interface_a.form_factor in NONCONNECTABLE_IFACE_TYPES:
raise ValidationError({
'interface_a': '{} is not a connectable interface type.'.format(
self.interface_a.get_form_factor_display()
)
})
if self.interface_b.form_factor in NONCONNECTABLE_IFACE_TYPES:
raise ValidationError({
'interface_b': '{} is not a connectable interface type.'.format(
self.interface_b.get_form_factor_display()
)
})
# Prevent the A side of one connection from being the B side of another
interface_a_connections = InterfaceConnection.objects.filter(
Q(interface_a=self.interface_a) |
Q(interface_b=self.interface_a)
).exclude(pk=self.pk)
if interface_a_connections.exists():
raise ValidationError({
'interface_a': "This interface is already connected."
})
interface_b_connections = InterfaceConnection.objects.filter(
Q(interface_a=self.interface_b) |
Q(interface_b=self.interface_b)
).exclude(pk=self.pk)
if interface_b_connections.exists():
raise ValidationError({
'interface_b': "This interface is already connected."
})
def to_csv(self):
return (

10
netbox/dcim/tests/test_api.py
View File

@ -1955,7 +1955,7 @@ class ConsolePortTest(APITestCase):
self.assertEqual(
sorted(response.data['results'][0]),
['device', 'id', 'name', 'url']
['device', 'id', 'is_connected', 'name', 'url']
)
def test_create_consoleport(self):
@ -2070,7 +2070,7 @@ class ConsoleServerPortTest(APITestCase):
self.assertEqual(
sorted(response.data['results'][0]),
['device', 'id', 'name', 'url']
['device', 'id', 'is_connected', 'name', 'url']
)
def test_create_consoleserverport(self):
@ -2181,7 +2181,7 @@ class PowerPortTest(APITestCase):
self.assertEqual(
sorted(response.data['results'][0]),
['device', 'id', 'name', 'url']
['device', 'id', 'is_connected', 'name', 'url']
)
def test_create_powerport(self):
@ -2296,7 +2296,7 @@ class PowerOutletTest(APITestCase):
self.assertEqual(
sorted(response.data['results'][0]),
['device', 'id', 'name', 'url']
['device', 'id', 'is_connected', 'name', 'url']
)
def test_create_poweroutlet(self):
@ -2432,7 +2432,7 @@ class InterfaceTest(APITestCase):
self.assertEqual(
sorted(response.data['results'][0]),
['device', 'id', 'name', 'url']
['device', 'id', 'is_connected', 'name', 'url']
)
def test_create_interface(self):