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

Merge branch 'develop' into 568-csv-import-cf

Browse Source
This commit is contained in:
hSaria 2020-01-23 20:27:07 +00:00 committed by GitHub
commit 06398a9ac6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
20 changed files with 706 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
docs/additional-features/custom-scripts.md
View File

@ -124,7 +124,7 @@ Arbitrary text of any length. Renders as multi-line text input field.
Stored a numeric integer. Options include:
* `min_value:` - Minimum value
* `min_value` - Minimum value
* `max_value` - Maximum value
### BooleanVar
@ -158,9 +158,20 @@ A NetBox object. The list of available objects is defined by the queryset parame
An uploaded file. Note that uploaded files are present in memory only for the duration of the script's execution: They will not be save for future use.
### IPAddressVar
An IPv4 or IPv6 address, without a mask. Returns a `netaddr.IPAddress` object.
### IPAddressWithMaskVar
An IPv4 or IPv6 address with a mask. Returns a `netaddr.IPNetwork` object which includes the mask.
### IPNetworkVar
An IPv4 or IPv6 network with a mask.
An IPv4 or IPv6 network with a mask. Returns a `netaddr.IPNetwork` object. Two attributes are available to validate the provided mask:
* `min_prefix_length` - Minimum length of the mask (default: none)
* `max_prefix_length` - Maximum length of the mask (default: none)
### Default Options

7
docs/release-notes/version-2.7.md
View File

@ -3,6 +3,13 @@
## Enhancements
* [#568](https://github.com/netbox-community/netbox/issues/568) - Allow custom fields to be imported and exported using CSV
* [#3310](https://github.com/netbox-community/netbox/issues/3310) - Pre-select site/rack for B side when creating a new cable
* [#3509](https://github.com/netbox-community/netbox/issues/3509) - Add IP address variables for custom scripts
## Bug Fixes
* [#3983](https://github.com/netbox-community/netbox/issues/3983) - Permit the creation of multiple unnamed devices
* [#3989](https://github.com/netbox-community/netbox/issues/3989) - Correct HTTP content type assignment for webhooks
---

10
netbox/dcim/forms.py
View File

@ -3168,6 +3168,11 @@ class ConnectCableToDeviceForm(BootstrapMixin, ChainedFieldsMixin, forms.ModelFo
'termination_b_site', 'termination_b_rack', 'termination_b_device', 'termination_b_id', 'type', 'status',
'label', 'color', 'length', 'length_unit',
]
widgets = {
'status': StaticSelect2,
'type': StaticSelect2,
'length_unit': StaticSelect2,
}
class ConnectCableToConsolePortForm(ConnectCableToDeviceForm):
@ -3363,6 +3368,11 @@ class CableForm(BootstrapMixin, forms.ModelForm):
fields = [
'type', 'status', 'label', 'color', 'length', 'length_unit',
]
widgets = {
'status': StaticSelect2,
'type': StaticSelect2,
'length_unit': StaticSelect2,
}
class CableCSVForm(forms.ModelForm):

9
netbox/dcim/models/__init__.py
View File

@ -1445,10 +1445,11 @@ class Device(ChangeLoggedModel, ConfigContextModel, CustomFieldModel):
# Check for a duplicate name on a device assigned to the same Site and no Tenant. This is necessary
# because Django does not consider two NULL fields to be equal, and thus will not trigger a violation
# of the uniqueness constraint without manual intervention.
if self.tenant is None and Device.objects.exclude(pk=self.pk).filter(name=self.name, tenant__isnull=True):
raise ValidationError({
'name': 'A device with this name already exists.'
})
if self.name and self.tenant is None:
if Device.objects.exclude(pk=self.pk).filter(name=self.name, tenant__isnull=True):
raise ValidationError({
'name': 'A device with this name already exists.'
})
super().validate_unique(exclude)

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

@ -4,6 +4,7 @@ from netaddr import IPNetwork
from rest_framework import status
from circuits.models import Circuit, CircuitTermination, CircuitType, Provider
from dcim.api import serializers
from dcim.choices import *
from dcim.constants import *
from dcim.models import (
@ -595,6 +596,21 @@ class RackTest(APITestCase):
self.assertEqual(response.data['count'], 42)
def test_get_rack_elevation(self):
url = reverse('dcim-api:rack-elevation', kwargs={'pk': self.rack1.pk})
response = self.client.get(url, **self.header)
self.assertEqual(response.data['count'], 42)
def test_get_rack_elevation_svg(self):
url = '{}?render=svg'.format(reverse('dcim-api:rack-elevation', kwargs={'pk': self.rack1.pk}))
response = self.client.get(url, **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(response.get('Content-Type'), 'image/svg+xml')
def test_list_racks(self):
url = reverse('dcim-api:rack-list')
@ -1900,6 +1916,31 @@ class DeviceTest(APITestCase):
self.assertEqual(response.data['device_role']['id'], self.devicerole1.pk)
self.assertEqual(response.data['cluster']['id'], self.cluster1.pk)
def test_get_device_graphs(self):
device_ct = ContentType.objects.get_for_model(Device)
self.graph1 = Graph.objects.create(
type=device_ct,
name='Test Graph 1',
source='http://example.com/graphs.py?device={{ obj.name }}&foo=1'
)
self.graph2 = Graph.objects.create(
type=device_ct,
name='Test Graph 2',
source='http://example.com/graphs.py?device={{ obj.name }}&foo=2'
)
self.graph3 = Graph.objects.create(
type=device_ct,
name='Test Graph 3',
source='http://example.com/graphs.py?device={{ obj.name }}&foo=3'
)
url = reverse('dcim-api:device-graphs', kwargs={'pk': self.device1.pk})
response = self.client.get(url, **self.header)
self.assertEqual(len(response.data), 3)
self.assertEqual(response.data[0]['embed_url'], 'http://example.com/graphs.py?device=Test Device 1&foo=1')
def test_list_devices(self):
url = reverse('dcim-api:device-list')
@ -2134,6 +2175,31 @@ class ConsolePortTest(APITestCase):
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(ConsolePort.objects.count(), 2)
def test_trace_consoleport(self):
peer_device = Device.objects.create(
site=Site.objects.first(),
device_type=DeviceType.objects.first(),
device_role=DeviceRole.objects.first(),
name='Peer Device'
)
console_server_port = ConsoleServerPort.objects.create(
device=peer_device,
name='Console Server Port 1'
)
cable = Cable(termination_a=self.consoleport1, termination_b=console_server_port, label='Cable 1')
cable.save()
url = reverse('dcim-api:consoleport-trace', kwargs={'pk': self.consoleport1.pk})
response = self.client.get(url, **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(len(response.data), 1)
segment1 = response.data[0]
self.assertEqual(segment1[0]['name'], self.consoleport1.name)
self.assertEqual(segment1[1]['label'], cable.label)
self.assertEqual(segment1[2]['name'], console_server_port.name)
class ConsoleServerPortTest(APITestCase):
@ -2245,6 +2311,31 @@ class ConsoleServerPortTest(APITestCase):
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(ConsoleServerPort.objects.count(), 2)
def test_trace_consoleserverport(self):
peer_device = Device.objects.create(
site=Site.objects.first(),
device_type=DeviceType.objects.first(),
device_role=DeviceRole.objects.first(),
name='Peer Device'
)
console_port = ConsolePort.objects.create(
device=peer_device,
name='Console Port 1'
)
cable = Cable(termination_a=self.consoleserverport1, termination_b=console_port, label='Cable 1')
cable.save()
url = reverse('dcim-api:consoleserverport-trace', kwargs={'pk': self.consoleserverport1.pk})
response = self.client.get(url, **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(len(response.data), 1)
segment1 = response.data[0]
self.assertEqual(segment1[0]['name'], self.consoleserverport1.name)
self.assertEqual(segment1[1]['label'], cable.label)
self.assertEqual(segment1[2]['name'], console_port.name)
class PowerPortTest(APITestCase):
@ -2358,6 +2449,31 @@ class PowerPortTest(APITestCase):
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(PowerPort.objects.count(), 2)
def test_trace_powerport(self):
peer_device = Device.objects.create(
site=Site.objects.first(),
device_type=DeviceType.objects.first(),
device_role=DeviceRole.objects.first(),
name='Peer Device'
)
power_outlet = PowerOutlet.objects.create(
device=peer_device,
name='Power Outlet 1'
)
cable = Cable(termination_a=self.powerport1, termination_b=power_outlet, label='Cable 1')
cable.save()
url = reverse('dcim-api:powerport-trace', kwargs={'pk': self.powerport1.pk})
response = self.client.get(url, **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(len(response.data), 1)
segment1 = response.data[0]
self.assertEqual(segment1[0]['name'], self.powerport1.name)
self.assertEqual(segment1[1]['label'], cable.label)
self.assertEqual(segment1[2]['name'], power_outlet.name)
class PowerOutletTest(APITestCase):
@ -2469,6 +2585,31 @@ class PowerOutletTest(APITestCase):
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(PowerOutlet.objects.count(), 2)
def test_trace_poweroutlet(self):
peer_device = Device.objects.create(
site=Site.objects.first(),
device_type=DeviceType.objects.first(),
device_role=DeviceRole.objects.first(),
name='Peer Device'
)
power_port = PowerPort.objects.create(
device=peer_device,
name='Power Port 1'
)
cable = Cable(termination_a=self.poweroutlet1, termination_b=power_port, label='Cable 1')
cable.save()
url = reverse('dcim-api:poweroutlet-trace', kwargs={'pk': self.poweroutlet1.pk})
response = self.client.get(url, **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(len(response.data), 1)
segment1 = response.data[0]
self.assertEqual(segment1[0]['name'], self.poweroutlet1.name)
self.assertEqual(segment1[1]['label'], cable.label)
self.assertEqual(segment1[2]['name'], power_port.name)
class InterfaceTest(APITestCase):
@ -2673,6 +2814,262 @@ class InterfaceTest(APITestCase):
self.assertEqual(Interface.objects.count(), 2)
class FrontPortTest(APITestCase):
def setUp(self):
super().setUp()
site = Site.objects.create(name='Test Site 1', slug='test-site-1')
manufacturer = Manufacturer.objects.create(name='Test Manufacturer 1', slug='test-manufacturer-1')
devicetype = DeviceType.objects.create(
manufacturer=manufacturer, model='Test Device Type 1', slug='test-device-type-1'
)
devicerole = DeviceRole.objects.create(
name='Test Device Role 1', slug='test-device-role-1', color='ff0000'
)
self.device = Device.objects.create(
device_type=devicetype, device_role=devicerole, name='Test Device 1', site=site
)
rear_ports = RearPort.objects.bulk_create((
RearPort(device=self.device, name='Rear Port 1', type=PortTypeChoices.TYPE_8P8C),
RearPort(device=self.device, name='Rear Port 2', type=PortTypeChoices.TYPE_8P8C),
RearPort(device=self.device, name='Rear Port 3', type=PortTypeChoices.TYPE_8P8C),
RearPort(device=self.device, name='Rear Port 4', type=PortTypeChoices.TYPE_8P8C),
RearPort(device=self.device, name='Rear Port 5', type=PortTypeChoices.TYPE_8P8C),
RearPort(device=self.device, name='Rear Port 6', type=PortTypeChoices.TYPE_8P8C),
))
self.frontport1 = FrontPort.objects.create(device=self.device, name='Front Port 1', type=PortTypeChoices.TYPE_8P8C, rear_port=rear_ports[0])
self.frontport3 = FrontPort.objects.create(device=self.device, name='Front Port 2', type=PortTypeChoices.TYPE_8P8C, rear_port=rear_ports[1])
self.frontport1 = FrontPort.objects.create(device=self.device, name='Front Port 3', type=PortTypeChoices.TYPE_8P8C, rear_port=rear_ports[2])
def test_get_frontport(self):
url = reverse('dcim-api:frontport-detail', kwargs={'pk': self.frontport1.pk})
response = self.client.get(url, **self.header)
self.assertEqual(response.data['name'], self.frontport1.name)
def test_list_frontports(self):
url = reverse('dcim-api:frontport-list')
response = self.client.get(url, **self.header)
self.assertEqual(response.data['count'], 3)
def test_list_frontports_brief(self):
url = reverse('dcim-api:frontport-list')
response = self.client.get('{}?brief=1'.format(url), **self.header)
self.assertEqual(
sorted(response.data['results'][0]),
['cable', 'device', 'id', 'name', 'url']
)
def test_create_frontport(self):
rear_port = RearPort.objects.get(name='Rear Port 4')
data = {
'device': self.device.pk,
'name': 'Front Port 4',
'type': PortTypeChoices.TYPE_8P8C,
'rear_port': rear_port.pk,
'rear_port_position': 1,
}
url = reverse('dcim-api:frontport-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(FrontPort.objects.count(), 4)
frontport4 = FrontPort.objects.get(pk=response.data['id'])
self.assertEqual(frontport4.device_id, data['device'])
self.assertEqual(frontport4.name, data['name'])
def test_create_frontport_bulk(self):
rear_ports = RearPort.objects.filter(frontports__isnull=True)
data = [
{
'device': self.device.pk,
'name': 'Front Port 4',
'type': PortTypeChoices.TYPE_8P8C,
'rear_port': rear_ports[0].pk,
'rear_port_position': 1,
},
{
'device': self.device.pk,
'name': 'Front Port 5',
'type': PortTypeChoices.TYPE_8P8C,
'rear_port': rear_ports[1].pk,
'rear_port_position': 1,
},
{
'device': self.device.pk,
'name': 'Front Port 6',
'type': PortTypeChoices.TYPE_8P8C,
'rear_port': rear_ports[2].pk,
'rear_port_position': 1,
},
]
url = reverse('dcim-api:frontport-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(FrontPort.objects.count(), 6)
self.assertEqual(response.data[0]['name'], data[0]['name'])
self.assertEqual(response.data[1]['name'], data[1]['name'])
self.assertEqual(response.data[2]['name'], data[2]['name'])
def test_update_frontport(self):
rear_port = RearPort.objects.get(name='Rear Port 4')
data = {
'device': self.device.pk,
'name': 'Front Port X',
'type': PortTypeChoices.TYPE_110_PUNCH,
'rear_port': rear_port.pk,
'rear_port_position': 1,
}
url = reverse('dcim-api:frontport-detail', kwargs={'pk': self.frontport1.pk})
response = self.client.put(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(FrontPort.objects.count(), 3)
frontport1 = FrontPort.objects.get(pk=response.data['id'])
self.assertEqual(frontport1.name, data['name'])
self.assertEqual(frontport1.type, data['type'])
self.assertEqual(frontport1.rear_port, rear_port)
def test_delete_frontport(self):
url = reverse('dcim-api:frontport-detail', kwargs={'pk': self.frontport1.pk})
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(FrontPort.objects.count(), 2)
class RearPortTest(APITestCase):
def setUp(self):
super().setUp()
site = Site.objects.create(name='Test Site 1', slug='test-site-1')
manufacturer = Manufacturer.objects.create(name='Test Manufacturer 1', slug='test-manufacturer-1')
devicetype = DeviceType.objects.create(
manufacturer=manufacturer, model='Test Device Type 1', slug='test-device-type-1'
)
devicerole = DeviceRole.objects.create(
name='Test Device Role 1', slug='test-device-role-1', color='ff0000'
)
self.device = Device.objects.create(
device_type=devicetype, device_role=devicerole, name='Test Device 1', site=site
)
self.rearport1 = RearPort.objects.create(device=self.device, type=PortTypeChoices.TYPE_8P8C, name='Rear Port 1')
self.rearport3 = RearPort.objects.create(device=self.device, type=PortTypeChoices.TYPE_8P8C, name='Rear Port 2')
self.rearport1 = RearPort.objects.create(device=self.device, type=PortTypeChoices.TYPE_8P8C, name='Rear Port 3')
def test_get_rearport(self):
url = reverse('dcim-api:rearport-detail', kwargs={'pk': self.rearport1.pk})
response = self.client.get(url, **self.header)
self.assertEqual(response.data['name'], self.rearport1.name)
def test_list_rearports(self):
url = reverse('dcim-api:rearport-list')
response = self.client.get(url, **self.header)
self.assertEqual(response.data['count'], 3)
def test_list_rearports_brief(self):
url = reverse('dcim-api:rearport-list')
response = self.client.get('{}?brief=1'.format(url), **self.header)
self.assertEqual(
sorted(response.data['results'][0]),
['cable', 'device', 'id', 'name', 'url']
)
def test_create_rearport(self):
data = {
'device': self.device.pk,
'name': 'Front Port 4',
'type': PortTypeChoices.TYPE_8P8C,
}
url = reverse('dcim-api:rearport-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(RearPort.objects.count(), 4)
rearport4 = RearPort.objects.get(pk=response.data['id'])
self.assertEqual(rearport4.device_id, data['device'])
self.assertEqual(rearport4.name, data['name'])
def test_create_rearport_bulk(self):
data = [
{
'device': self.device.pk,
'name': 'Rear Port 4',
'type': PortTypeChoices.TYPE_8P8C,
},
{
'device': self.device.pk,
'name': 'Rear Port 5',
'type': PortTypeChoices.TYPE_8P8C,
},
{
'device': self.device.pk,
'name': 'Rear Port 6',
'type': PortTypeChoices.TYPE_8P8C,
},
]
url = reverse('dcim-api:rearport-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(RearPort.objects.count(), 6)
self.assertEqual(response.data[0]['name'], data[0]['name'])
self.assertEqual(response.data[1]['name'], data[1]['name'])
self.assertEqual(response.data[2]['name'], data[2]['name'])
def test_update_rearport(self):
data = {
'device': self.device.pk,
'name': 'Front Port X',
'type': PortTypeChoices.TYPE_110_PUNCH
}
url = reverse('dcim-api:rearport-detail', kwargs={'pk': self.rearport1.pk})
response = self.client.put(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(RearPort.objects.count(), 3)
rearport1 = RearPort.objects.get(pk=response.data['id'])
self.assertEqual(rearport1.name, data['name'])
self.assertEqual(rearport1.type, data['type'])
def test_delete_rearport(self):
url = reverse('dcim-api:rearport-detail', kwargs={'pk': self.rearport1.pk})
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(RearPort.objects.count(), 2)
class DeviceBayTest(APITestCase):
def setUp(self):

23
netbox/dcim/tests/test_models.py
View File

@ -285,7 +285,28 @@ class DeviceTestCase(TestCase):
name='Device Bay 1'
)
def test_device_duplicate_name_per_site(self):
def test_multiple_unnamed_devices(self):
device1 = Device(
site=self.site,
device_type=self.device_type,
device_role=self.device_role,
name=''
)
device1.save()
device2 = Device(
site=device1.site,
device_type=device1.device_type,
device_role=device1.device_role,
name=''
)
device2.full_clean()
device2.save()
self.assertEqual(Device.objects.filter(name='').count(), 2)
def test_device_duplicate_names(self):
device1 = Device(
site=self.site,

6
netbox/dcim/views.py
View File

@ -1945,6 +1945,12 @@ class CableCreateView(PermissionRequiredMixin, GetReturnURLMixin, View):
# Parse initial data manually to avoid setting field values as lists
initial_data = {k: request.GET[k] for k in request.GET}
# Set initial site and rack based on side A termination (if not already set)
if 'termination_b_site' not in initial_data:
initial_data['termination_b_site'] = getattr(self.obj.termination_a.parent, 'site', None)
if 'termination_b_rack' not in initial_data:
initial_data['termination_b_rack'] = getattr(self.obj.termination_a.parent, 'rack', None)
form = self.form_class(instance=self.obj, initial=initial_data)
return render(request, self.template_name, {

43
netbox/extras/scripts.py
View File

@ -14,10 +14,10 @@ from django.db import transaction
from mptt.forms import TreeNodeChoiceField, TreeNodeMultipleChoiceField
from mptt.models import MPTTModel
from ipam.formfields import IPFormField
from utilities.exceptions import AbortTransaction
from utilities.validators import MaxPrefixLengthValidator, MinPrefixLengthValidator
from ipam.formfields import IPAddressFormField, IPNetworkFormField
from ipam.validators import MaxPrefixLengthValidator, MinPrefixLengthValidator, prefix_validator
from .constants import LOG_DEFAULT, LOG_FAILURE, LOG_INFO, LOG_SUCCESS, LOG_WARNING
from utilities.exceptions import AbortTransaction
from .forms import ScriptForm
from .signals import purge_changelog
@ -27,6 +27,8 @@ __all__ = [
'ChoiceVar',
'FileVar',
'IntegerVar',
'IPAddressVar',
'IPAddressWithMaskVar',
'IPNetworkVar',
'MultiObjectVar',
'ObjectVar',
@ -48,15 +50,19 @@ class ScriptVariable:
def __init__(self, label='', description='', default=None, required=True):
# Default field attributes
self.field_attrs = {
'help_text': description,
'required': required
}
# Initialize field attributes
if not hasattr(self, 'field_attrs'):
self.field_attrs = {}
if description:
self.field_attrs['help_text'] = description
if label:
self.field_attrs['label'] = label
if default:
self.field_attrs['initial'] = default
if required:
self.field_attrs['required'] = True
if 'validators' not in self.field_attrs:
self.field_attrs['validators'] = []
def as_field(self):
"""
@ -196,17 +202,32 @@ class FileVar(ScriptVariable):
form_field = forms.FileField
class IPAddressVar(ScriptVariable):
"""
An IPv4 or IPv6 address without a mask.
"""
form_field = IPAddressFormField
class IPAddressWithMaskVar(ScriptVariable):
"""
An IPv4 or IPv6 address with a mask.
"""
form_field = IPNetworkFormField
class IPNetworkVar(ScriptVariable):
"""
An IPv4 or IPv6 prefix.
"""
form_field = IPFormField
form_field = IPNetworkFormField
field_attrs = {
'validators': [prefix_validator]
}
def __init__(self, min_prefix_length=None, max_prefix_length=None, *args, **kwargs):
super().__init__(*args, **kwargs)
self.field_attrs['validators'] = list()
# Optional minimum/maximum prefix lengths
if min_prefix_length is not None:
self.field_attrs['validators'].append(

56
netbox/extras/tests/test_scripts.py
View File

@ -1,6 +1,6 @@
from django.core.files.uploadedfile import SimpleUploadedFile
from django.test import TestCase
from netaddr import IPNetwork
from netaddr import IPAddress, IPNetwork
from dcim.models import DeviceRole
from extras.scripts import *
@ -186,6 +186,54 @@ class ScriptVariablesTest(TestCase):
self.assertTrue(form.is_valid())
self.assertEqual(form.cleaned_data['var1'], testfile)
def test_ipaddressvar(self):
class TestScript(Script):
var1 = IPAddressVar()
# Validate IP network enforcement
data = {'var1': '1.2.3'}
form = TestScript().as_form(data, None)
self.assertFalse(form.is_valid())
self.assertIn('var1', form.errors)
# Validate IP mask exclusion
data = {'var1': '192.0.2.0/24'}
form = TestScript().as_form(data, None)
self.assertFalse(form.is_valid())
self.assertIn('var1', form.errors)
# Validate valid data
data = {'var1': '192.0.2.1'}
form = TestScript().as_form(data, None)
self.assertTrue(form.is_valid())
self.assertEqual(form.cleaned_data['var1'], IPAddress(data['var1']))
def test_ipaddresswithmaskvar(self):
class TestScript(Script):
var1 = IPAddressWithMaskVar()
# Validate IP network enforcement
data = {'var1': '1.2.3'}
form = TestScript().as_form(data, None)
self.assertFalse(form.is_valid())
self.assertIn('var1', form.errors)
# Validate IP mask requirement
data = {'var1': '192.0.2.0'}
form = TestScript().as_form(data, None)
self.assertFalse(form.is_valid())
self.assertIn('var1', form.errors)
# Validate valid data
data = {'var1': '192.0.2.0/24'}
form = TestScript().as_form(data, None)
self.assertTrue(form.is_valid())
self.assertEqual(form.cleaned_data['var1'], IPNetwork(data['var1']))
def test_ipnetworkvar(self):
class TestScript(Script):
@ -198,6 +246,12 @@ class ScriptVariablesTest(TestCase):
self.assertFalse(form.is_valid())
self.assertIn('var1', form.errors)
# Validate host IP check
data = {'var1': '192.0.2.1/24'}
form = TestScript().as_form(data, None)
self.assertFalse(form.is_valid())
self.assertIn('var1', form.errors)
# Validate valid data
data = {'var1': '192.0.2.0/24'}
form = TestScript().as_form(data, None)

62
netbox/extras/tests/test_webhooks.py
View File

@ -1,11 +1,19 @@
import json
import uuid
from unittest.mock import patch
import django_rq
from django.contrib.contenttypes.models import ContentType
from django.http import HttpResponse
from django.urls import reverse
from requests import Session
from rest_framework import status
from dcim.models import Site
from extras.choices import ObjectChangeActionChoices
from extras.models import Webhook
from extras.webhooks import enqueue_webhooks, generate_signature
from extras.webhooks_worker import process_webhook
from utilities.testing import APITestCase
@ -22,11 +30,13 @@ class WebhookTest(APITestCase):
def setUpTestData(cls):
site_ct = ContentType.objects.get_for_model(Site)
PAYLOAD_URL = "http://localhost/"
DUMMY_URL = "http://localhost/"
DUMMY_SECRET = "LOOKATMEIMASECRETSTRING"
webhooks = Webhook.objects.bulk_create((
Webhook(name='Site Create Webhook', type_create=True, payload_url=PAYLOAD_URL),
Webhook(name='Site Update Webhook', type_update=True, payload_url=PAYLOAD_URL),
Webhook(name='Site Delete Webhook', type_delete=True, payload_url=PAYLOAD_URL),
Webhook(name='Site Create Webhook', type_create=True, payload_url=DUMMY_URL, secret=DUMMY_SECRET, additional_headers={'X-Foo': 'Bar'}),
Webhook(name='Site Update Webhook', type_update=True, payload_url=DUMMY_URL, secret=DUMMY_SECRET),
Webhook(name='Site Delete Webhook', type_delete=True, payload_url=DUMMY_URL, secret=DUMMY_SECRET),
))
for webhook in webhooks:
webhook.obj_type.set([site_ct])
@ -87,3 +97,47 @@ class WebhookTest(APITestCase):
self.assertEqual(job.args[1]['id'], site.pk)
self.assertEqual(job.args[2], 'site')
self.assertEqual(job.args[3], ObjectChangeActionChoices.ACTION_DELETE)
def test_webhooks_worker(self):
request_id = uuid.uuid4()
def dummy_send(_, request):
"""
A dummy implementation of Session.send() to be used for testing.
Always returns a 200 HTTP response.
"""
webhook = Webhook.objects.get(type_create=True)
signature = generate_signature(request.body, webhook.secret)
# Validate the outgoing request headers
self.assertEqual(request.headers['Content-Type'], webhook.http_content_type)
self.assertEqual(request.headers['X-Hook-Signature'], signature)
self.assertEqual(request.headers['X-Foo'], 'Bar')
# Validate the outgoing request body
body = json.loads(request.body)
self.assertEqual(body['event'], 'created')
self.assertEqual(body['timestamp'], job.args[4])
self.assertEqual(body['model'], 'site')
self.assertEqual(body['username'], 'testuser')
self.assertEqual(body['request_id'], str(request_id))
self.assertEqual(body['data']['name'], 'Site 1')
return HttpResponse()
# Enqueue a webhook for processing
site = Site.objects.create(name='Site 1', slug='site-1')
enqueue_webhooks(
instance=site,
user=self.user,
request_id=request_id,
action=ObjectChangeActionChoices.ACTION_CREATE
)
# Retrieve the job from queue
job = self.queue.jobs[0]
# Patch the Session object with our dummy_send() method, then process the webhook for sending
with patch.object(Session, 'send', dummy_send) as mock_send:
process_webhook(*job.args)

14
netbox/extras/webhooks.py
View File

@ -1,4 +1,6 @@
import datetime
import hashlib
import hmac
from django.contrib.contenttypes.models import ContentType
@ -8,6 +10,18 @@ from .choices import *
from .constants import *
def generate_signature(request_body, secret):
"""
Return a cryptographic signature that can be used to verify the authenticity of webhook data.
"""
hmac_prep = hmac.new(
key=secret.encode('utf8'),
msg=request_body.encode('utf8'),
digestmod=hashlib.sha512
)
return hmac_prep.hexdigest()
def enqueue_webhooks(instance, user, request_id, action):
"""
Find Webhook(s) assigned to this instance + action and enqueue them

14
netbox/extras/webhooks_worker.py
View File

@ -1,5 +1,3 @@
import hashlib
import hmac
import json
import requests
@ -7,6 +5,7 @@ from django_rq import job
from rest_framework.utils.encoders import JSONEncoder
from .choices import ObjectChangeActionChoices, WebhookContentTypeChoices
from .webhooks import generate_signature
@job('default')
@ -23,7 +22,7 @@ def process_webhook(webhook, data, model_name, event, timestamp, username, reque
'data': data
}
headers = {
'Content-Type': webhook.get_http_content_type_display(),
'Content-Type': webhook.http_content_type,
}
if webhook.additional_headers:
headers.update(webhook.additional_headers)
@ -43,12 +42,7 @@ def process_webhook(webhook, data, model_name, event, timestamp, username, reque
if webhook.secret != '':
# Sign the request with a hash of the secret key and its content.
hmac_prep = hmac.new(
key=webhook.secret.encode('utf8'),
msg=prepared_request.body.encode('utf8'),
digestmod=hashlib.sha512
)
prepared_request.headers['X-Hook-Signature'] = hmac_prep.hexdigest()
prepared_request.headers['X-Hook-Signature'] = generate_signature(prepared_request.body, webhook.secret)
with requests.Session() as session:
session.verify = webhook.ssl_verification
@ -56,7 +50,7 @@ def process_webhook(webhook, data, model_name, event, timestamp, username, reque
session.verify = webhook.ca_file_path
response = session.send(prepared_request)
if response.status_code >= 200 and response.status_code <= 299:
if 200 <= response.status_code <= 299:
return 'Status {} returned, webhook successfully processed.'.format(response.status_code)
else:
raise requests.exceptions.RequestException(

13
netbox/ipam/fields.py
View File

@ -2,13 +2,8 @@ from django.core.exceptions import ValidationError
from django.db import models
from netaddr import AddrFormatError, IPNetwork
from . import lookups
from .formfields import IPFormField
def prefix_validator(prefix):
if prefix.ip != prefix.cidr.ip:
raise ValidationError("{} is not a valid prefix. Did you mean {}?".format(prefix, prefix.cidr))
from . import lookups, validators
from .formfields import IPNetworkFormField
class BaseIPField(models.Field):
@ -38,7 +33,7 @@ class BaseIPField(models.Field):
return str(self.to_python(value))
def form_class(self):
return IPFormField
return IPNetworkFormField
def formfield(self, **kwargs):
defaults = {'form_class': self.form_class()}
@ -51,7 +46,7 @@ class IPNetworkField(BaseIPField):
IP prefix (network and mask)
"""
description = "PostgreSQL CIDR field"
default_validators = [prefix_validator]
default_validators = [validators.prefix_validator]
def db_type(self, connection):
return 'cidr'

35
netbox/ipam/formfields.py
View File

@ -1,13 +1,44 @@
from django import forms
from django.core.exceptions import ValidationError
from netaddr import IPNetwork, AddrFormatError
from django.core.validators import validate_ipv4_address, validate_ipv6_address
from netaddr import IPAddress, IPNetwork, AddrFormatError
#
# Form fields
#
class IPFormField(forms.Field):
class IPAddressFormField(forms.Field):
default_error_messages = {
'invalid': "Enter a valid IPv4 or IPv6 address (without a mask).",
}
def to_python(self, value):
if not value:
return None
if isinstance(value, IPAddress):
return value
# netaddr is a bit too liberal with what it accepts as a valid IP address. For example, '1.2.3' will become
# IPAddress('1.2.0.3'). Here, we employ Django's built-in IPv4 and IPv6 address validators as a sanity check.
try:
validate_ipv4_address(value)
except ValidationError:
try:
validate_ipv6_address(value)
except ValidationError:
raise ValidationError("Invalid IPv4/IPv6 address format: {}".format(value))
try:
return IPAddress(value)
except ValueError:
raise ValidationError('This field requires an IP address without a mask.')
except AddrFormatError:
raise ValidationError("Please specify a valid IPv4 or IPv6 address.")
class IPNetworkFormField(forms.Field):
default_error_messages = {
'invalid': "Enter a valid IPv4 or IPv6 address (with CIDR mask).",
}

24
netbox/ipam/validators.py
View File

@ -1,4 +1,26 @@
from django.core.validators import RegexValidator
from django.core.exceptions import ValidationError
from django.core.validators import BaseValidator, RegexValidator
def prefix_validator(prefix):
if prefix.ip != prefix.cidr.ip:
raise ValidationError("{} is not a valid prefix. Did you mean {}?".format(prefix, prefix.cidr))
class MaxPrefixLengthValidator(BaseValidator):
message = 'The prefix length must be less than or equal to %(limit_value)s.'
code = 'max_prefix_length'
def compare(self, a, b):
return a.prefixlen > b
class MinPrefixLengthValidator(BaseValidator):
message = 'The prefix length must be greater than or equal to %(limit_value)s.'
code = 'min_prefix_length'
def compare(self, a, b):
return a.prefixlen < b
DNSValidator = RegexValidator(

1
netbox/secrets/tests/test_form.py
View File

@ -29,5 +29,4 @@ class UserKeyFormTestCase(TestCase):
data={'public_key': SSH_PUBLIC_KEY},
instance=self.userkey,
)
print(form.is_valid())
self.assertFalse(form.is_valid())

21
netbox/templates/dcim/cable_connect.html
View File

@ -144,25 +144,8 @@
</div>
</div>
<div class="row">
<div class="col-md-4 col-md-offset-4">
<div class="panel panel-default">
<div class="panel-heading"><strong>Cable</strong></div>
<div class="panel-body">
{% render_field form.status %}
{% render_field form.type %}
{% render_field form.label %}
{% render_field form.color %}
<div class="form-group">
<label class="col-md-3 control-label" for="id_length">{{ form.length.label }}</label>
<div class="col-md-6">
{{ form.length }}
</div>
<div class="col-md-3">
{{ form.length_unit }}
</div>
</div>
</div>
</div>
<div class="col-md-6 col-md-offset-3">
{% include 'dcim/inc/cable_form.html' %}
</div>
</div>
<div class="form-group">

20
netbox/templates/dcim/cable_edit.html
View File

@ -1,23 +1,5 @@
{% extends 'utilities/obj_edit.html' %}
{% load form_helpers %}
{% block form %}
<div class="panel panel-default">
<div class="panel-heading"><strong>Cable</strong></div>
<div class="panel-body">
{% render_field form.type %}
{% render_field form.status %}
{% render_field form.label %}
{% render_field form.color %}
<div class="form-group">
<label class="col-md-3 control-label" for="id_length">{{ form.length.label }}</label>
<div class="col-md-6">
{{ form.length }}
</div>
<div class="col-md-3">
{{ form.length_unit }}
</div>
</div>
</div>
</div>
{% include 'dcim/inc/cable_form.html' %}
{% endblock %}

19
netbox/templates/dcim/inc/cable_form.html Normal file
View File

@ -0,0 +1,19 @@
{% load form_helpers %}
<div class="panel panel-default">
<div class="panel-heading"><strong>Cable</strong></div>
<div class="panel-body">
{% render_field form.status %}
{% render_field form.type %}
{% render_field form.label %}
{% render_field form.color %}
<div class="form-group">
<label class="col-md-3 control-label" for="id_length">{{ form.length.label }}</label>
<div class="col-md-5">
{{ form.length }}
</div>
<div class="col-md-4">
{{ form.length_unit }}
</div>
</div>
</div>
</div>

18
netbox/utilities/validators.py
View File

@ -1,6 +1,6 @@
import re
from django.core.validators import _lazy_re_compile, BaseValidator, URLValidator
from django.core.validators import _lazy_re_compile, URLValidator
class EnhancedURLValidator(URLValidator):
@ -26,19 +26,3 @@ class EnhancedURLValidator(URLValidator):
r'(?:[/?#][^\s]*)?' # Path
r'\Z', re.IGNORECASE)
schemes = AnyURLScheme()
class MaxPrefixLengthValidator(BaseValidator):
message = 'The prefix length must be less than or equal to %(limit_value)s.'
code = 'max_prefix_length'
def compare(self, a, b):
return a.prefixlen > b
class MinPrefixLengthValidator(BaseValidator):
message = 'The prefix length must be greater than or equal to %(limit_value)s.'
code = 'min_prefix_length'
def compare(self, a, b):
return a.prefixlen < b