diff --git a/interfaces/forms.py b/interfaces/forms.py index c62f1816..83923c8c 100644 --- a/interfaces/forms.py +++ b/interfaces/forms.py @@ -6,7 +6,7 @@ from django.utils.translation import gettext_lazy as _ class AddInterface(forms.Form): - name = forms.CharField(max_length=10, required=True) + name = forms.CharField(max_length=15, required=True) itype = forms.ChoiceField( required=True, choices=(("bridge", "bridge"), ("ethernet", "ethernet")) ) @@ -19,7 +19,8 @@ class AddInterface(forms.Form): required=True, choices=(("dhcp", "dhcp"), ("static", "static"), ("none", "none")), ) - ipv4_addr = forms.CharField(max_length=18, required=False) + # room for a dotted netmask: 192.0.2.10/255.255.255.0 + ipv4_addr = forms.CharField(max_length=31, required=False) ipv4_gw = forms.CharField(max_length=15, required=False) ipv6_type = forms.ChoiceField( required=True, @@ -28,81 +29,92 @@ class AddInterface(forms.Form): ipv6_addr = forms.CharField(max_length=100, required=False) ipv6_gw = forms.CharField(max_length=100, required=False) stp = forms.ChoiceField(required=False, choices=(("on", "on"), ("off", "off"))) - delay = forms.IntegerField(required=False) + delay = forms.IntegerField(required=False, min_value=0) + + def __init__(self, *args, netdevs=(), **kwargs): + # The host's network devices: a bridge member, or the device an + # ethernet interface configures, must be one of them. + super().__init__(*args, **kwargs) + self.netdevs = set(netdevs) def clean_ipv4_addr(self): ipv4_addr = self.cleaned_data["ipv4_addr"] - have_symbol = re.match("^[0-9./]+$", ipv4_addr) - if not have_symbol: - raise forms.ValidationError( - _("The IPv4 address must not contain any special characters") - ) - elif len(ipv4_addr) > 20: - raise forms.ValidationError( - _("The IPv4 address must not exceed 20 characters") - ) - return ipv4_addr + if not ipv4_addr: + return ipv4_addr + try: + # Always address/prefix, as the interface XML needs both + return str(ipaddress.IPv4Interface(ipv4_addr)) + except ValueError: + raise forms.ValidationError(_("The IPv4 address is not valid (e.g. 192.0.2.10/24)")) def clean_ipv4_gw(self): ipv4_gw = self.cleaned_data["ipv4_gw"] - have_symbol = re.match("^[0-9.]+$", ipv4_gw) - if not have_symbol: - raise forms.ValidationError( - _("The IPv4 gateway must not contain any special characters") - ) - elif len(ipv4_gw) > 20: - raise forms.ValidationError( - _("The IPv4 gateway must not exceed 20 characters") - ) - return ipv4_gw + if not ipv4_gw: + return ipv4_gw + try: + return str(ipaddress.IPv4Address(ipv4_gw)) + except ValueError: + raise forms.ValidationError(_("The IPv4 gateway is not valid")) + # libvirt's interface schema has no IPv6 scope id (fe80::1%eth0) def clean_ipv6_addr(self): ipv6_addr = self.cleaned_data["ipv6_addr"] if not ipv6_addr: return ipv6_addr try: - # Always address/prefix, as the interface XML needs both - return str(ipaddress.IPv6Interface(ipv6_addr)) + address = ipaddress.IPv6Interface(ipv6_addr) except ValueError: + address = None + if address is None or address.scope_id is not None: raise forms.ValidationError(_("The IPv6 address is not valid (e.g. 2001:db8::10/64)")) + # Always address/prefix, as the interface XML needs both + return str(address) def clean_ipv6_gw(self): ipv6_gw = self.cleaned_data["ipv6_gw"] if not ipv6_gw: return ipv6_gw try: - return str(ipaddress.IPv6Address(ipv6_gw)) + gateway = ipaddress.IPv6Address(ipv6_gw) except ValueError: + gateway = None + if gateway is None or gateway.scope_id is not None: raise forms.ValidationError(_("The IPv6 gateway is not valid")) - - def clean(self): - cleaned = super().clean() - if cleaned.get("ipv6_type") == "static" and not cleaned.get("ipv6_addr") and "ipv6_addr" not in self.errors: - self.add_error("ipv6_addr", _("A static IPv6 configuration needs an address")) - return cleaned + return str(gateway) def clean_name(self): + # A Linux interface name: up to 15 characters, no spaces or slashes name = self.cleaned_data["name"] - have_symbol = re.match("^[a-z0-9.]+$", name) - if not have_symbol: + if not re.fullmatch(r"[A-Za-z0-9_.-]{1,15}", name) or name in (".", ".."): raise forms.ValidationError( - _("The interface must not contain any special characters") - ) - elif len(name) > 10: - raise forms.ValidationError( - _("The interface must not exceed 10 characters") + _("The interface name may have up to 15 letters, digits and . _ -") ) return name def clean_netdev(self): netdev = self.cleaned_data["netdev"] - have_symbol = re.match("^[a-z0-9.:]+$", netdev) - if not have_symbol: + if not re.fullmatch(r"[A-Za-z0-9_.-]{1,15}", netdev) or netdev in (".", ".."): raise forms.ValidationError( - _("The interface must not contain any special characters") - ) - elif len(netdev) > 10: - raise forms.ValidationError( - _("The interface must not exceed 10 characters") + _("The device name may have up to 15 letters, digits and . _ -") ) return netdev + + def clean(self): + cleaned = super().clean() + for family in ("ipv4", "ipv6"): + addr = f"{family}_addr" + if cleaned.get(f"{family}_type") == "static" and not cleaned.get(addr) and addr not in self.errors: + self.add_error(addr, _("A static configuration needs an address")) + name, netdev = cleaned.get("name"), cleaned.get("netdev") + if cleaned.get("itype") == "bridge": + if not cleaned.get("stp") and "stp" not in self.errors: + self.add_error("stp", _("A bridge needs STP on or off")) + if cleaned.get("delay") is None and "delay" not in self.errors: + self.add_error("delay", _("A bridge needs a forward delay")) + if netdev and netdev not in self.netdevs: + self.add_error("netdev", _("The device is not on this host")) + elif netdev and netdev == name: + self.add_error("netdev", _("A bridge cannot contain itself")) + elif cleaned.get("itype") == "ethernet" and name and name not in self.netdevs: + self.add_error("name", _("An ethernet interface configures a device of this host")) + return cleaned diff --git a/interfaces/templates/create_iface_block.html b/interfaces/templates/create_iface_block.html index aad6c07d..9837ba28 100644 --- a/interfaces/templates/create_iface_block.html +++ b/interfaces/templates/create_iface_block.html @@ -18,7 +18,7 @@
- +
@@ -96,7 +96,7 @@
- +
diff --git a/interfaces/test_forms.py b/interfaces/test_forms.py index e1bea3c3..b5a36456 100644 --- a/interfaces/test_forms.py +++ b/interfaces/test_forms.py @@ -1,8 +1,11 @@ -"""Host interface form: IPv6 values are validated before they go into the interface XML.""" +"""Host interface form: every value is validated before it goes into the interface XML.""" -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch -from django.test import SimpleTestCase +from computes.models import Compute +from django.contrib.auth import get_user_model +from django.test import SimpleTestCase, TestCase +from django.urls import reverse from lxml import etree from vrtManager.interface import wvmInterfaces @@ -15,8 +18,12 @@ BASE = { } -def form(**ipv6): - return AddInterface(data={**BASE, **ipv6}) +# the host's network devices, as interfaces/views.py passes them +NETDEVS = ["eth0", "eth1", "enp129s0f0np0", "eth0.100", "bond-lan"] + + +def form(**values): + return AddInterface(data={**BASE, **values}, netdevs=NETDEVS) class Ipv6FormTestCase(SimpleTestCase): @@ -67,3 +74,154 @@ class Ipv6XmlTestCase(SimpleTestCase): def test_no_empty_route_without_gateway(self): self.assertIsNone(self.xml("").find("protocol[@family='ipv6']/route")) + + +class Ipv4FormTestCase(SimpleTestCase): + def test_dhcp_and_none_need_no_address(self): + for ipv4_type in ("dhcp", "none"): + with self.subTest(ipv4_type=ipv4_type): + f = form(ipv4_type=ipv4_type, ipv4_addr="", ipv4_gw="", ipv6_type="none") + self.assertTrue(f.is_valid(), f.errors) + + def test_valid_address_and_gateway(self): + f = form(ipv4_addr="192.0.2.10/24", ipv4_gw="192.0.2.1", ipv6_type="none") + self.assertTrue(f.is_valid(), f.errors) + self.assertEqual(f.cleaned_data["ipv4_addr"], "192.0.2.10/24") + + def test_address_without_prefix_gets_one(self): + f = form(ipv4_addr="192.0.2.10", ipv4_gw="", ipv6_type="none") + self.assertTrue(f.is_valid(), f.errors) + self.assertEqual(f.cleaned_data["ipv4_addr"], "192.0.2.10/32") + + def test_invalid_values_are_rejected(self): + for field, value in ( + ("ipv4_addr", "192.0.2.300/24"), + ("ipv4_addr", "192.0.2.10/24/1"), + ("ipv4_addr", "192.0.2.10/33"), + ("ipv4_gw", "192.0.2.300"), + ("ipv4_gw", "192.0.2.1/24"), + ): + with self.subTest(field=field, value=value): + f = form(**{"ipv6_type": "none", field: value}) + self.assertFalse(f.is_valid()) + self.assertIn(field, f.errors) + + def test_static_needs_an_address(self): + f = form(ipv4_addr="", ipv4_gw="", ipv6_type="none") + self.assertFalse(f.is_valid()) + self.assertIn("ipv4_addr", f.errors) + + +class InterfaceNameTestCase(SimpleTestCase): + def test_linux_interface_names_are_accepted(self): + for name in ("br0", "bond-lan", "br_mgmt", "vlan.100", "BR0", "a" * 15): + with self.subTest(name=name): + f = form(name=name, ipv6_type="none") + self.assertTrue(f.is_valid(), f.errors) + for netdev in ("enp129s0f0np0", "eth0.100", "bond-lan"): + with self.subTest(netdev=netdev): + f = form(netdev=netdev, ipv6_type="none") + self.assertTrue(f.is_valid(), f.errors) + + def test_other_names_are_rejected(self): + for field in ("name", "netdev"): + for value in ("br 1", "br/1", "br'1", "br1\nx", "a" * 16, "", ".", ".."): + with self.subTest(field=field, value=value): + f = form(**{"ipv6_type": "none", field: value}) + self.assertFalse(f.is_valid()) + self.assertIn(field, f.errors) + + def test_surrounding_whitespace_is_stripped(self): + f = form(name="br1\n", ipv6_type="none") + self.assertTrue(f.is_valid(), f.errors) + self.assertEqual(f.cleaned_data["name"], "br1") + + def test_a_new_interface_name_has_no_alias_colon(self): + f = form(name="br0:1", ipv6_type="none") + self.assertFalse(f.is_valid()) + self.assertIn("name", f.errors) + + +class Ipv4XmlTestCase(SimpleTestCase): + def xml(self, addr, gw): + conn = wvmInterfaces.__new__(wvmInterfaces) + conn.define_iface = MagicMock() + conn.get_iface = MagicMock() + conn.create_iface("br1", "bridge", "onboot", "eth1", "static", addr, gw, "none", "", "", "on", 0) + return etree.fromstring(conn.define_iface.call_args[0][0]) + + def test_static_address_with_gateway(self): + proto = self.xml("192.0.2.10/24", "192.0.2.1").find("protocol[@family='ipv4']") + self.assertEqual(proto.find("ip").attrib, {"address": "192.0.2.10", "prefix": "24"}) + self.assertEqual(proto.find("route").get("gateway"), "192.0.2.1") + + def test_no_empty_route_without_gateway(self): + self.assertIsNone(self.xml("192.0.2.10/24", "").find("protocol[@family='ipv4']/route")) + + +class HostDeviceTestCase(SimpleTestCase): + def test_a_bridge_member_must_be_a_host_device(self): + for netdev in ("eth9", "eth0:1"): + with self.subTest(netdev=netdev): + f = form(netdev=netdev, ipv6_type="none") + self.assertFalse(f.is_valid()) + self.assertIn("netdev", f.errors) + + def test_a_bridge_cannot_contain_itself(self): + f = form(name="eth1", netdev="eth1", ipv6_type="none") + self.assertFalse(f.is_valid()) + self.assertIn("netdev", f.errors) + + def test_an_ethernet_interface_configures_a_host_device(self): + self.assertTrue(form(itype="ethernet", name="eth0", ipv6_type="none").is_valid()) + f = form(itype="ethernet", name="eth9", ipv6_type="none") + self.assertFalse(f.is_valid()) + self.assertIn("name", f.errors) + + +class BridgeSettingsTestCase(SimpleTestCase): + def test_a_bridge_needs_stp_and_a_delay(self): + for field in ("stp", "delay"): + with self.subTest(field=field): + f = form(**{"ipv6_type": "none", field: ""}) + self.assertFalse(f.is_valid()) + self.assertIn(field, f.errors) + + def test_the_delay_is_not_negative(self): + f = form(delay=-1, ipv6_type="none") + self.assertFalse(f.is_valid()) + self.assertIn("delay", f.errors) + + +class AddressNotationTestCase(SimpleTestCase): + def test_ipv4_netmask_notation_becomes_a_prefix(self): + f = form(ipv4_addr="192.0.2.10/255.255.255.0", ipv6_type="none") + self.assertTrue(f.is_valid(), f.errors) + self.assertEqual(f.cleaned_data["ipv4_addr"], "192.0.2.10/24") + + def test_ipv6_scope_ids_are_rejected(self): + for field, value in (("ipv6_addr", "fe80::1%eth0/64"), ("ipv6_gw", "fe80::1%eth0")): + with self.subTest(field=field): + f = form(**{"ipv6_addr": "2001:db8::10/64", "ipv6_gw": "", field: value}) + self.assertFalse(f.is_valid()) + self.assertIn(field, f.errors) + + +class CreateViewTestCase(TestCase): + """The page validates the device against the host's devices before it + defines an interface.""" + + def test_only_a_host_device_becomes_a_bridge_member(self): + admin = get_user_model().objects.create_superuser("iface_admin", "a@example.com", "x") + compute = Compute.objects.create(name="iface-c", hostname="127.0.0.1:1", login="root", password="", type=1) + self.client.force_login(admin) + for netdev, created in (("eth9", False), ("eth0", True)): + with self.subTest(netdev=netdev), patch("interfaces.views.wvmInterfaces") as conn_cls: + conn = conn_cls.return_value + conn.get_ifaces.return_value = [] + conn.get_net_devices.return_value = ["eth0", "eth1"] + self.client.post( + reverse("interfaces", args=[compute.id]), + {**BASE, "netdev": netdev, "ipv6_type": "none", "create": ""}, + ) + self.assertEqual(conn.create_iface.called, created) diff --git a/interfaces/views.py b/interfaces/views.py index ec82ea52..6453e08f 100644 --- a/interfaces/views.py +++ b/interfaces/views.py @@ -30,7 +30,7 @@ def interfaces(request, compute_id): ) ifaces = conn.get_ifaces() - netdevs = ["eth0", "eth1"] + netdevs = [] with contextlib.suppress(Exception): netdevs = conn.get_net_devices() @@ -46,7 +46,7 @@ def interfaces(request, compute_id): if request.method == "POST": if "create" in request.POST: - form = AddInterface(request.POST) + form = AddInterface(request.POST, netdevs=netdevs) if form.is_valid(): data = form.cleaned_data conn.create_iface( diff --git a/vrtManager/interface.py b/vrtManager/interface.py index 52d40704..9a8830c0 100644 --- a/vrtManager/interface.py +++ b/vrtManager/interface.py @@ -34,9 +34,10 @@ class wvmInterfaces(wvmConnect): if ipv4_type == "static": address, prefix = ipv4_addr.split("/") xml += f""" - - - """ + """ + if ipv4_gw: + xml += f"""""" + xml += """""" if ipv6_type == "dhcp": xml += """