From f745033d29bfcbc0337d85c867a305f6fbd6d5d7 Mon Sep 17 00:00:00 2001 From: catborise Date: Thu, 29 Oct 2026 11:28:21 +0300 Subject: [PATCH] feat(networks): a network's DHCP leases on its page - A DHCP leases card on the page of a network that serves DHCP: each lease's address, MAC or an IPv6 client's DUID, the host name the client sent, the VM whose card has that MAC, linked, and when it ends. - A lease that is a fixed address is marked; any other in one of the network's DHCP subnets has a Make fixed button that fills in the fixed address dialog with its address, MAC or DUID and name. - Leases that cannot be read are said on the card; the rest of the page stays. A network that does not run has none. - The VMs on a network are read with the MACs of their cards on it, from the definitions already read. --- networks/templates/network.html | 66 +++++++++- networks/tests/test_network_leases.py | 118 ++++++++++++++++++ networks/views.py | 20 +++ .../icon_cache/pin-angle_None_None_None.svg | 8 ++ vrtManager/network.py | 42 +++++-- 5 files changed, 245 insertions(+), 9 deletions(-) create mode 100644 networks/tests/test_network_leases.py create mode 100644 static/icon_cache/pin-angle_None_None_None.svg diff --git a/networks/templates/network.html b/networks/templates/network.html index 5b185c50..ae9c17c4 100644 --- a/networks/templates/network.html +++ b/networks/templates/network.html @@ -189,6 +189,69 @@
{% trans "The network serves no DHCP, so it has no fixed addresses." %}
{% endif %} + + {% if dhcp_networks %} +
+
{% trans "DHCP leases" %} {% if leases %}{{ leases|length }}{% endif %}
+ {% if leases_error %} +
{% blocktrans with error=leases_error %}The leases could not be read: {{ error }}{% endblocktrans %}
+ {% elif leases %} +
+ + + + + + + + + + + + + {% for lease in leases %} + + + + + + + + + {% endfor %} + +
{% trans "Address" %}{% trans "MAC / DUID" %}{% trans "Host name" %}{% trans "VM" %}{% trans "Ends" %}
+ {{ lease.ip }} + {% if lease.fixed %}{% trans "Fixed" %}{% endif %} + {{ lease.key|default:"–" }}{{ lease.hostname|default:"–" }} + {% if lease.vm.id %} + {{ lease.vm.name }} + {% else %} + {{ lease.vm.name|default:"–" }} + {% endif %} + + {% if lease.expires %} +
{% blocktrans with time=lease.expires|timeuntil %}in {{ time }}{% endblocktrans %}
+
{{ lease.expires|date:"Y-m-d H:i" }}
+ {% else %} + {% trans "Never" %} + {% endif %} + 		+ {# a lease kept as a fixed address: the dialog of a new one, filled in #} + {% if not lease.fixed and lease.key and lease.ip_index is not None %} + + {% endif %} +
+
+ {% elif state %} +
{% trans "No current DHCP leases." %}
+ {% else %} +
{% trans "The network is not running, so it gives no leases." %}
+ {% endif %} +
+ {% endif %} {% endif %}
@@ -413,7 +476,8 @@ form.elements.address.value = data.address || ""; form.elements.mac.value = data.mac || ""; form.elements.name.value = data.name || ""; - if (data.ipIndex) form.elements.ip_index.value = data.ipIndex; + // Add starts from the first subnet, not the one a row chose before + form.elements.ip_index.value = data.ipIndex || form.elements.ip_index.options[0].value; document.getElementById("fixed_subnet_row").hidden = Boolean(data.original); }); fillFrom("DhcpRange", function (form, data) { diff --git a/networks/tests/test_network_leases.py b/networks/tests/test_network_leases.py new file mode 100644 index 00000000..593e8f89 --- /dev/null +++ b/networks/tests/test_network_leases.py @@ -0,0 +1,118 @@ +"""A network's DHCP leases on its page: each with the VM whose card has its +MAC, a mark when it is a fixed address, and a "Make fixed" button that fills +in the fixed address dialog, by MAC for IPv4 and by DUID for IPv6. Leases +that cannot be read leave the rest of the page. Runs on a private libvirt +test driver, which has no leases: they are given.""" + +import tempfile +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +import libvirt +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 instances.models import Instance +from lxml import html +from networks.tests.driver import without_inactive_flag +from networks.tests.test_network_pages import NODE +from vrtManager.network import wvmNetwork + +DUID = "00:01:00:01:2e:8b:3c:41:52:54:00:00:00:40" + + +def lease(ip, mac=None, expiry=1793265489, hostname=None, clientid=None, family=0): + return {"iface": "virbr9", "expirytime": expiry, "type": family, "mac": mac, "ipaddr": ip, "prefix": 24, + "hostname": hostname, "clientid": clientid, "iaid": None} + + +class LeasesTests(SimpleTestCase): + def network(self, leases, active=True): + net = wvmNetwork.__new__(wvmNetwork) + net.net = MagicMock(**{"isActive.return_value": active, "DHCPLeases.return_value": leases}) + return net + + def test_leases_by_address_ipv4_first(self): + net = self.network([lease("2001:db8:1::40", family=1, clientid=DUID, expiry=0), + lease("192.0.2.100", "52:54:00:AA:00:01", hostname="db"), + lease("192.0.2.9", "52:54:00:aa:00:02")]) + ipv4_9, ipv4_100, ipv6 = net.get_leases() + self.assertEqual([str(ipv4_9["ip"]), str(ipv4_100["ip"]), str(ipv6["ip"])], ["192.0.2.9", "192.0.2.100", "2001:db8:1::40"]) + self.assertEqual((ipv4_100["mac"], ipv4_100["hostname"], ipv4_100["expires"]), + ("52:54:00:aa:00:01", "db", datetime.fromtimestamp(1793265489, timezone.utc))) + # an IPv6 lease: no MAC, its client's DUID; one that never ends + self.assertEqual((ipv6["mac"], ipv6["client_id"], ipv6["expires"]), ("", DUID, None)) + + def test_a_network_that_does_not_run_has_none(self): + net = self.network([lease("192.0.2.9")], active=False) + self.assertEqual(net.get_leases(), []) + net.net.DHCPLeases.assert_not_called() + + +class LeasesCardTests(TestCase): + def setUp(self): + node = tempfile.NamedTemporaryFile("w", suffix=".xml") + self.addCleanup(node.close) + node.write(NODE) + node.flush() + self.conn = libvirt.open(f"test://{node.name}") + self.addCleanup(self.conn.close) + patcher = patch("vrtManager.connection.connection_manager.get_connection", return_value=self.conn) + patcher.start() + self.addCleanup(patcher.stop) + without_inactive_flag(self) + self.compute = Compute.objects.create(name="nl", hostname="localhost", login="", password="", type=4) + self.client.force_login(get_user_model().objects.create_superuser("nl", "nl@example.com", "pw")) + web = self.conn.lookupByName("web") + self.vm = Instance.objects.create(compute=self.compute, name="web", uuid=web.UUIDString()) + self.mac = html.fromstring(web.XMLDesc(0)).xpath("string(//interface/mac/@address)") + + def rows(self, leases): + with patch.object(libvirt.virNetwork, "DHCPLeases", return_value=leases, create=True): + response = self.client.get(reverse("network", args=[self.compute.id, "lab"])) + self.assertEqual(response.status_code, 200) + page = html.fromstring(response.content) + card = page.xpath("//div[@class='card mb-3'][div[@class='card-header'][starts-with(normalize-space(), 'DHCP leases')]]")[0] + return card, card.xpath(".//tbody/tr") + + def test_each_lease_with_its_vm_and_whether_it_is_fixed(self): + _, (by_vm, fixed, stale, ipv6) = self.rows([ + lease("192.0.2.15", self.mac.upper(), hostname="web-guest"), + lease("192.0.2.31", "52:54:00:00:00:31"), + lease("198.51.100.5", "52:54:00:00:00:99"), # a removed VM's, outside the network's subnets + lease("2001:db8:1::40", family=1, clientid=DUID), + ]) + self.assertEqual(by_vm.xpath("td[4]/a/@href"), [reverse("instances:instance", args=[self.vm.id])]) + button = by_vm.xpath(".//button[@data-bs-target='#FixedAddress']")[0] + self.assertEqual((button.get("data-ip-index"), button.get("data-address"), button.get("data-mac"), button.get("data-name")), + ("0", "192.0.2.15", self.mac.lower(), "web-guest")) + self.assertIn("Fixed", fixed.xpath("string(td[1])")) + self.assertFalse(fixed.xpath(".//button")) + self.assertEqual((stale.xpath("normalize-space(td[4])"), stale.xpath(".//button")), ("–", [])) + # an IPv6 fixed address is known by the client's DUID, not a MAC + button = ipv6.xpath(".//button")[0] + self.assertEqual((button.get("data-ip-index"), button.get("data-mac"), button.get("data-name")), ("1", DUID, "")) + + def test_an_ipv6_lease_is_known_by_its_duid_and_its_address_however_written(self): + _, (fixed, other) = self.rows([ + lease("2001:db8:1:0:0:0:0:31", family=1, clientid="0:1:0:1"), # the fixed 2001:db8:1::31 + lease("2001:db8:1::41", "52:54:00:00:00:41", family=1, clientid=DUID), + ]) + self.assertEqual((fixed.xpath("normalize-space(td[1])"), fixed.xpath(".//button")), ("2001:db8:1::31 Fixed", [])) + self.assertEqual(other.xpath(".//button/@data-mac"), [DUID]) + + def test_a_vm_the_panel_does_not_know_is_named_not_linked(self): + self.vm.delete() + _, (row,) = self.rows([lease("192.0.2.15", self.mac)]) + self.assertEqual((row.xpath("normalize-space(td[4])"), row.xpath("td[4]/a")), ("web", [])) + # without a host name, the fixed address takes the VM's name + self.assertEqual(row.xpath(".//button/@data-name"), ["web"]) + + def test_leases_that_cannot_be_read_leave_the_page(self): + with patch.object(libvirt.virNetwork, "DHCPLeases", side_effect=libvirt.libvirtError("status file unreadable"), + create=True): + response = self.client.get(reverse("network", args=[self.compute.id, "lab"])) + page = html.fromstring(response.content) + self.assertIn("The leases could not be read: status file unreadable", page.text_content()) + self.assertTrue(page.xpath("//tbody[@class='searchable']/tr"), "the fixed addresses are still shown") diff --git a/networks/views.py b/networks/views.py index 69d32f7e..54ac18e2 100644 --- a/networks/views.py +++ b/networks/views.py @@ -1,3 +1,5 @@ +import ipaddress + from admin.decorators import superuser_only from computes.models import Compute from instances.models import Instance @@ -279,6 +281,24 @@ def network(request, compute_id, pool): except libvirtError as lib_err: messages.error(request, lib_err) return HttpResponseRedirect(reverse("networks", args=[compute_id])) + + # leases that cannot be read leave the rest of the page + leases, leases_error = [], None + if dhcp_networks: + try: + leases = conn.get_leases() + except libvirtError as lib_err: + leases_error = lib_err + vm_by_mac = {mac: vm for vm in vms for mac in vm["macs"]} + fixed_ips = {ipaddress.ip_address(host["ip"]) for host in fixed_addresses if host["ip"]} + for lease in leases: + lease["vm"] = vm_by_mac.get(lease["mac"]) + lease["name"] = lease["hostname"] or (lease["vm"]["name"] if lease["vm"] else "") # for a fixed address + lease["fixed"] = lease["ip"] in fixed_ips + # what a fixed address for it is known by: its MAC, or an IPv6 client's DUID + lease["key"] = lease["mac"] if lease["ip"].version == 4 else lease["client_id"] + lease["ip_index"] = next((ip["index"] for ip in dhcp_networks + if ip["subnet"] and lease["ip"] in ip["subnet"]), None) conn.close() return render(request, "network.html", locals()) diff --git a/static/icon_cache/pin-angle_None_None_None.svg b/static/icon_cache/pin-angle_None_None_None.svg new file mode 100644 index 00000000..4818ac3e --- /dev/null +++ b/static/icon_cache/pin-angle_None_None_None.svg @@ -0,0 +1,8 @@ + + + + + + + + diff --git a/vrtManager/network.py b/vrtManager/network.py index 03a96aa1..56d2978e 100644 --- a/vrtManager/network.py +++ b/vrtManager/network.py @@ -1,5 +1,6 @@ import copy import ipaddress +from datetime import datetime, timezone from functools import cached_property from libvirt import ( @@ -240,25 +241,50 @@ class wvmNetwork(wvmConnect): return network_mode(self.tree) def get_vms(self): - """The VMs whose definition names this network, by name, and whether - each is running on it now: one listing, and one XML read per stopped - VM, two per running one (its running and its saved definition).""" + """The VMs whose definition names this network, by name, whether each + is running on it now, and the MACs of its cards on it (as it runs, or + as saved when it does not): one listing, and one XML read per stopped + VM, two per running one that is not on it now.""" name = self.net.name() - def on_it(xml): - return etree.fromstring(xml).xpath("devices/interface[@type='network']/source[@network=$name]", name=name) + def cards(xml): + return etree.fromstring(xml).xpath("devices/interface[@type='network'][source/@network=$name]", name=name) vms = [] for dom in self.wvm.listAllDomains(): try: - connected = dom.isActive() and bool(on_it(dom.XMLDesc(0))) - if connected or on_it(dom.XMLDesc(VIR_DOMAIN_XML_INACTIVE)): - vms.append({"name": dom.name(), "connected": connected}) + on_it = cards(dom.XMLDesc(0)) if dom.isActive() else [] + connected = bool(on_it) + if not connected: + on_it = cards(dom.XMLDesc(VIR_DOMAIN_XML_INACTIVE)) except libvirtError as err: # a VM undefined meanwhile is not on it if err.get_error_code() != VIR_ERR_NO_DOMAIN: raise + continue + if on_it: + macs = [card.xpath("string(mac/@address)").lower() for card in on_it] + vms.append({"name": dom.name(), "connected": connected, "macs": [mac for mac in macs if mac]}) return sorted(vms, key=lambda vm: vm["name"]) + def get_leases(self): + """The network's DHCP leases as its dnsmasq keeps them, by address: IP, + MAC (an IPv6 lease may have none), the host name and client ID (an + IPv6 lease's DUID) the client sent, and when it ends (None: never). + A network that does not run has none.""" + if not self.is_active(): + return [] + leases = [] + for lease in self.net.DHCPLeases(): + expiry = lease["expirytime"] + leases.append({ + "ip": ipaddress.ip_address(lease["ipaddr"]), + "mac": (lease.get("mac") or "").lower(), + "hostname": lease.get("hostname") or "", + "client_id": lease.get("clientid") or "", + "expires": datetime.fromtimestamp(expiry, timezone.utc) if expiry else None, + }) + return sorted(leases, key=lambda lease: (lease["ip"].version, int(lease["ip"]))) + def _vms_on_it(self, running_only): name = self.net.name() return self.domains_using(