+ {# 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 %}
+
+
+ {% endfor %}
+
+
+
+ {% 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(