mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-09 03:36:48 +00:00
fix(interfaces): a host interface can be created with DHCP or no IPv4, and every value is validated
The IPv4 address and gateway were required even for DHCP or no IPv4, so such an interface could not be created; malformed addresses passed and a missing prefix crashed the XML builder. They are validated like the IPv6 fields now (netmask notation accepted, IPv6 scope ids rejected), and a static address without a gateway gets no empty route. Interface names follow the Linux rule (up to 15 letters, digits, . _ -). A bridge member, or the device an ethernet interface configures, must be a device of the host (the page no longer falls back to made-up names), a bridge cannot contain itself, and needs STP and a non-negative delay.
This commit is contained in:
parent
3c0d5efb18
commit
443b31d86b
5 changed files with 229 additions and 58 deletions
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue