From 03ac93ea53b449b56e1ecbdbe65866c6459665e7 Mon Sep 17 00:00:00 2001
From: catborise <catborise@gmail.com>
Date: Sun, 4 Oct 2026 18:33:28 +0300
Subject: [PATCH] fix(instances): CD-ROM bus, USB controller and MACs of a new
 VM

Creating each option on a RHEL qemu-kvm host showed:

- CD-ROM: the create page offered every disk bus, but the empty CD-ROM
  of a new VM works only on ide, sata and scsi. fdc and virtio failed
  ("Invalid bus type", "does not support ejectable media"), usb failed
  with a device named 'Nonea' because vol_dev_type() had no prefix for
  usb, and after that libvirt refuses an empty usb disk. Only ide, sata
  and scsi are offered now, and another value is refused before any
  storage is allocated. vol_dev_type() names usb disks sdX, which also
  fixes data volumes on the usb bus.
- USB controller: any input device added a bare <controller
  type='usb'/>, which libvirt turns into piix3-uhci (USB 1.1) on q35
  instead of its default qemu-xhci. The bare controller is gone; libvirt
  adds the machine's default for USB inputs.
- MACs: with more networks than MAC addresses, create_instance indexed
  past the list and the request failed with a 500. NICs without a given
  MAC now get one from libvirt. MAC addresses are validated (format, not
  more than networks) by the page and the API before anything is
  created; they went into the XML unchecked.

Unit tests cover the refused values, the MAC rule and the usb prefix.
Live tests create a VM with every offered CD-ROM bus and start it,
check the USB controller with a USB input device and create a VM with
two networks and one MAC.
---
 instances/api/serializers.py                |  8 +++
 instances/templates/create_instance_w2.html |  6 +-
 instances/test_create_instance.py           | 75 ++++++++++++++++-----
 instances/test_live_create.py               | 24 +++++++
 instances/utils.py                          | 14 ++++
 instances/views.py                          | 15 ++++-
 vrtManager/create.py                        | 11 +--
 vrtManager/test_util.py                     |  1 +
 vrtManager/util.py                          |  2 +-
 9 files changed, 130 insertions(+), 26 deletions(-)

diff --git a/instances/api/serializers.py b/instances/api/serializers.py
index 999cfa03..5c2847c8 100644
--- a/instances/api/serializers.py
+++ b/instances/api/serializers.py
@@ -1,4 +1,5 @@
 from instances.models import CreateInstance, Flavor, Instance, MigrateInstance
+from instances.utils import nic_macs
 from rest_framework import serializers
 
 
@@ -58,6 +59,13 @@ class CreateInstanceSerializer(serializers.ModelSerializer):
     virtio = serializers.BooleanField(initial=True)
     qemu_ga = serializers.BooleanField(initial=True)
 
+    def validate(self, attrs):
+        try:
+            nic_macs(attrs.get("mac"), attrs.get("networks", ""))
+        except ValueError as err:
+            raise serializers.ValidationError({"mac": str(err)})
+        return attrs
+
     class Meta:
         model = CreateInstance
         fields = [
diff --git a/instances/templates/create_instance_w2.html b/instances/templates/create_instance_w2.html
index e70260c3..648f8341 100644
--- a/instances/templates/create_instance_w2.html
+++ b/instances/templates/create_instance_w2.html
@@ -251,7 +251,7 @@
                                                                         <div class="col-sm-7">
                                                                             <select class="form-select" id="add_cdrom" name="add_cdrom">
                                                                                 <option value="None" {% if bus == default_cdrom %}selected{% endif %}>--{% trans "False" %}--</option>
-                                                                                {% for bus in disk_buses %}
+                                                                                {% for bus in cdrom_buses %}
                                                                                     <option value="{{ bus }}" {% if bus == default_cdrom %}selected {% endif %}>
                                                                                         {% trans bus %}
                                                                                     </option>
@@ -504,7 +504,7 @@
                             <div class="col-sm-7">
                                 <select class="form-select" id="add_cdrom" name="add_cdrom">
                                     <option value="None" {% if bus == default_cdrom %}selected{% endif %} >--{% trans "False" %}--</option>
-                                    {% for bus in disk_buses %}
+                                    {% for bus in cdrom_buses %}
                                         <option value="{{ bus }}" {% if bus == default_cdrom %}selected {% endif %}>
                                             {% trans bus %}
                                         </option>
@@ -744,7 +744,7 @@
                                 <div class="col-sm-7">
                                     <select class="form-select" id="add_cdrom" name="add_cdrom">
                                         <option value="None" {% if bus == default_cdrom %}selected{% endif %}>--{% trans "False" %}--</option>
-                                        {% for bus in disk_buses %}
+                                        {% for bus in cdrom_buses %}
                                             <option value="{{ bus }}" {% if bus == default_cdrom %}selected {% endif %}>
                                                 {% trans bus %}
                                             </option>
diff --git a/instances/test_create_instance.py b/instances/test_create_instance.py
index 241fa609..23a7d7fc 100644
--- a/instances/test_create_instance.py
+++ b/instances/test_create_instance.py
@@ -6,28 +6,58 @@ from computes.models import Compute
 from django.contrib.auth import get_user_model
 from django.test import TestCase
 from django.urls import reverse
+from instances.api.serializers import CreateInstanceSerializer
 from instances.models import Instance
+from instances.utils import nic_macs
 
 
 class CreateInstanceViewTests(TestCase):
     @patch("instances.views.wvmCreate")
     @patch("instances.views.NewVMForm")
     def test_successful_create_redirects_after_saving_instance(self, form_class, connection_class):
-        self.create(form_class, connection_class, "BIOS")
+        response, conn = self.create(form_class, connection_class)
+
+        instance = Instance.objects.get(name="new-vm")
+        self.assertRedirects(
+            response, reverse("instances:instance", args=[instance.id]),
+            fetch_redirect_response=False,
+        )
+        conn.create_instance.assert_called_once()
 
     @patch("instances.views.wvmCreate")
     @patch("instances.views.NewVMForm")
     def test_secure_boot_loader_on_q35_requests_secure_boot(self, form_class, connection_class):
-        self.create(form_class, connection_class, "UEFI x86_64: /usr/share/edk2/ovmf/OVMF_CODE.secboot.fd")
+        self.create(form_class, connection_class, firmware="UEFI x86_64: /usr/share/edk2/ovmf/OVMF_CODE.secboot.fd")
         firmware = connection_class.return_value.create_instance.call_args.kwargs["firmware"]
         self.assertEqual(firmware["secure"], "yes")
 
-    def create(self, form_class, connection_class, firmware):
-        user = get_user_model().objects.create_superuser(
-            username="create-test-admin", password="password", email="create@example.com"
+    @patch("instances.views.wvmCreate")
+    @patch("instances.views.NewVMForm")
+    def test_invalid_devices_are_refused_before_anything_is_created(self, form_class, connection_class):
+        for data in (
+            {"add_cdrom": "virtio"},  # no ejectable media
+            {"add_cdrom": "fdc"},
+            {"add_cdrom": "usb"},  # libvirt refuses an empty usb disk
+            {"mac": "52:54:00:00:00:01,52:54:00:00:00:02"},  # one network
+            {"mac": "52:54:00:00:00:01'/><evil/>"},
+        ):
+            with self.subTest(data):
+                Instance.objects.all().delete()
+                response, conn = self.create(form_class, connection_class, **data)
+                self.assertEqual(response.status_code, 200)
+                self.assertFalse(Instance.objects.exists())
+                conn.create_volume.assert_not_called()
+                conn.create_instance.assert_not_called()
+                connection_class.reset_mock()
+
+    def create(self, form_class, connection_class, **data):
+        user = get_user_model().objects.filter(username="create-test-admin").first() or (
+            get_user_model().objects.create_superuser(
+                username="create-test-admin", password="password", email="create@example.com"
+            )
         )
         self.client.force_login(user)
-        compute = Compute.objects.create(
+        compute, _ = Compute.objects.get_or_create(
             name="create-test-compute", hostname="localhost", type=4,
             login="", password="",
         )
@@ -36,7 +66,7 @@ class CreateInstanceViewTests(TestCase):
         conn.get_video_models.return_value = ["vga"]
         conn.get_cache_modes.return_value = {"default": "Default"}
         conn.get_disk_device_types.return_value = ["disk"]
-        conn.get_disk_bus_types.return_value = ["virtio"]
+        conn.get_disk_bus_types.return_value = ["fdc", "virtio", "sata"]
         conn.get_networks.return_value = ["default"]
         conn.get_nwfilters.return_value = []
         conn.get_network_models.return_value = ["virtio"]
@@ -53,25 +83,20 @@ class CreateInstanceViewTests(TestCase):
         form.cleaned_data = {
             "name": "new-vm", "meta_prealloc": False, "hdd_size": 0,
             "template": "", "images": "base.qcow2", "cache_mode": "default",
-            "firmware": firmware, "net_model": "virtio", "memory": 1024,
+            "firmware": "BIOS", "net_model": "virtio", "memory": 1024,
             "vcpu": 1, "vcpu_mode": "host-model", "networks": "default",
             "virtio": True, "listener_addr": "0.0.0.0", "nwfilter": "",
             "graphics": "vnc", "video": "vga", "console_pass": "",
             "mac": "52:54:00:00:00:01", "qemu_ga": False,
             "add_cdrom": "sata", "add_input": "default",
+            **data,
         }
 
         response = self.client.post(
             reverse("instances:create_instance", args=[compute.id, "x86_64", "q35"]),
             {"create": "1"},
         )
-
-        instance = Instance.objects.get(compute=compute, name="new-vm")
-        self.assertRedirects(
-            response, reverse("instances:instance", args=[instance.id]),
-            fetch_redirect_response=False,
-        )
-        conn.create_instance.assert_called_once()
+        return response, conn
 
     @patch("instances.views.wvmCreate")
     def test_firmware_list_uses_the_chosen_machine_type(self, connection_class):
@@ -99,3 +124,23 @@ class CreateInstanceViewTests(TestCase):
         self.assertEqual(response.status_code, 200)
         conn.find_uefi_path_for_arch.assert_called_once_with("x86_64", "q35")
 
+
+class NicMacsTestCase(TestCase):
+    def test_macs_follow_the_networks(self):
+        self.assertEqual(nic_macs("52:54:00:00:00:01", "a,b"), ["52:54:00:00:00:01"])
+        self.assertEqual(nic_macs("", "a,b"), [])
+
+    def test_invalid_or_extra_macs_are_refused(self):
+        for mac, networks in (("52:54:00:00:00:01,52:54:00:00:00:02", "a"), ("not-a-mac", "a")):
+            with self.subTest(mac=mac), self.assertRaises(ValueError):
+                nic_macs(mac, networks)
+
+    def test_api_serializer_refuses_extra_macs(self):
+        serializer = CreateInstanceSerializer(data={
+            "name": "api-vm", "vcpu": 1, "vcpu_mode": "host-model", "memory": 512,
+            "networks": "default", "mac": "52:54:00:00:00:01,52:54:00:00:00:02", "nwfilter": "",
+            "storage": "default", "hdd_size": 1, "cache_mode": "none", "meta_prealloc": False,
+            "virtio": True, "qemu_ga": True, "console_pass": "", "video": "vga", "listener_addr": "0.0.0.0",
+        })
+        self.assertFalse(serializer.is_valid())
+        self.assertIn("mac", serializer.errors)
diff --git a/instances/test_live_create.py b/instances/test_live_create.py
index 9bdafaf0..5f9bbe4b 100644
--- a/instances/test_live_create.py
+++ b/instances/test_live_create.py
@@ -143,6 +143,30 @@ class LiveCreateTestCase(TestCase):
         dom.create()
         self.assertTrue(dom.isActive())
 
+    def test_every_offered_cdrom_bus_creates_a_vm_that_starts(self):
+        page = self.client.get(reverse("instances:create_instance", args=[self.compute.id, "x86_64", "q35"]))
+        buses = page.context["cdrom_buses"]
+        self.assertTrue(buses)
+        for i, bus in enumerate(buses):
+            with self.subTest(bus=bus):
+                self.create(f"cr-cd-{bus}", add_cdrom=bus, mac=f"52:54:00:aa:03:1{i}")
+                dom = self.conn.lookupByName(P + f"cr-cd-{bus}")
+                self.assertEqual(etree.fromstring(dom.XMLDesc(0)).xpath("devices/disk[@device='cdrom']/target/@bus"), [bus])
+                dom.create()
+                self.assertTrue(dom.isActive())
+
+    def test_input_device_keeps_the_machine_usb_controller(self):
+        self.create("cr-input", add_input="usb")
+        tree = etree.fromstring(self.conn.lookupByName(P + "cr-input").XMLDesc(0))
+        self.assertEqual(tree.xpath("devices/controller[@type='usb']/@model"), ["qemu-xhci"])
+        self.assertTrue(tree.xpath("devices/input[@type='tablet' and @bus='usb']"))
+
+    def test_networks_without_a_mac_of_their_own(self):
+        self.create("cr-macs", networks="default,default")
+        macs = etree.fromstring(self.conn.lookupByName(P + "cr-macs").XMLDesc(0)).xpath("devices/interface/mac/@address")
+        self.assertEqual(len(macs), 2)
+        self.assertEqual(macs[0], "52:54:00:aa:03:01")
+
     def test_api_rejected_definition_removes_the_new_disk(self):
         response = self.client.post(
             f"/api/v1/computes/{self.compute.id}/instances/create/x86_64/q35/",
diff --git a/instances/utils.py b/instances/utils.py
index c09cb31b..9da9ba0b 100644
--- a/instances/utils.py
+++ b/instances/utils.py
@@ -246,6 +246,20 @@ def get_dhcp_mac_address(vname):
     return mac
 
 
+MAC_ADDRESS_RE = re.compile(r"^([0-9A-F]{2})(:?[0-9A-F]{2}){5}$", re.IGNORECASE)
+
+
+def nic_macs(mac, networks):
+    """The MAC addresses given for a new VM's NICs, in network order. NICs
+    without one get theirs from libvirt."""
+    macs = [m.strip() for m in mac.split(",")] if mac else []
+    if not all(MAC_ADDRESS_RE.fullmatch(m) for m in macs):
+        raise ValueError(_("Invalid MAC address"))
+    if len(macs) > len(networks.split(",")):
+        raise ValueError(_("More MAC addresses than networks"))
+    return macs
+
+
 def get_random_mac_address():
     mac = settings.MAC_OUI + ":%02x:%02x:%02x" % (
         random.randint(0x00, 0xFF),
diff --git a/instances/views.py b/instances/views.py
index 52119cb2..e04e4347 100755
--- a/instances/views.py
+++ b/instances/views.py
@@ -1704,7 +1704,9 @@ def del_owner(request, pk):
     return _back(request, pk, "users")
 
 
-MAC_ADDRESS_RE = re.compile(r"^([0-9A-F]{2})(:?[0-9A-F]{2}){5}$", re.IGNORECASE)
+# Buses for the empty CD-ROM of a new VM: fdc and virtio have no ejectable
+# media, and libvirt refuses an empty usb disk.
+CDROM_BUSES = ("ide", "sata", "scsi")
 CLONE_POST_KEY_RE = re.compile(r"^(clone-net-mac-\d+|disk-[a-z0-9]+|meta-[a-z0-9]+)$")
 
 
@@ -1763,7 +1765,7 @@ def clone(request, pk):
     invalid_macs = [
         value
         for key, value in clone_data.items()
-        if key.startswith("clone-net-mac-") and not MAC_ADDRESS_RE.fullmatch(value)
+        if key.startswith("clone-net-mac-") and not utils.MAC_ADDRESS_RE.fullmatch(value)
     ]
     invalid_disks = [
         value
@@ -2071,6 +2073,7 @@ def create_instance(request, compute_id, arch, machine):
         mac_auto = util.randomMAC()
         disk_devices = conn.get_disk_device_types(arch, machine)
         disk_buses = conn.get_disk_bus_types(arch, machine)
+        cdrom_buses = [bus for bus in disk_buses if bus in CDROM_BUSES]
         default_bus = app_settings.INSTANCE_VOLUME_DEFAULT_BUS
         networks = sorted(conn.get_networks())
         nwfilters = conn.get_nwfilters()
@@ -2132,6 +2135,14 @@ def create_instance(request, compute_id, arch, machine):
                             )
                         if data["cache_mode"] not in conn.get_cache_modes():
                             raise util.OperationError(_("Invalid cache mode"))
+                        if data["add_cdrom"] != "None" and data["add_cdrom"] not in cdrom_buses:
+                            raise util.OperationError(
+                                _("A CD-ROM cannot use the %(bus)s bus") % {"bus": data["add_cdrom"]}
+                            )
+                        try:
+                            utils.nic_macs(data["mac"], data["networks"])
+                        except ValueError as err:
+                            raise util.OperationError(str(err))
 
                         firmware = dict()
                         if "UEFI" in data["firmware"]:
diff --git a/vrtManager/create.py b/vrtManager/create.py
index 3304ed03..91f846de 100644
--- a/vrtManager/create.py
+++ b/vrtManager/create.py
@@ -306,12 +306,12 @@ class wvmCreate(wvmConnect):
                 xml += """<target dev='%s%s' bus='%s'/>""" % (dev_prefix, get_letter(add_cdrom), add_cdrom)
             xml += """</disk>"""
 
-        if mac:
-            macs = mac.split(',')
+        # NICs without a given MAC get one from libvirt
+        macs = mac.split(",") if mac else []
         for idx, net in enumerate(networks.split(",")):
             xml += """<interface type='network'>"""
-            if mac:
-                xml += f"""<mac address='{macs[idx]}'/>"""
+            if idx < len(macs):
+                xml += f"""<mac address='{macs[idx].strip()}'/>"""
             xml += f"""<source network='{net}'/>"""
             if nwfilter:
                 xml += f"""<filterref filter='{nwfilter}'/>"""
@@ -324,8 +324,9 @@ class wvmCreate(wvmConnect):
         elif console_pass != "":
             console_pass = "passwd='" + console_pass + "'"
 
+        # No bare USB controller: libvirt adds the machine's default
+        # (qemu-xhci on q35) for USB inputs; a bare one is piix3-uhci.
         if add_input != "None":
-            xml += """<controller type='usb'/>"""
             if add_input in dom_caps["disk_bus"]:
                 xml += f"""<input type='mouse' bus='{add_input}'/>"""
                 xml += f"""<input type='keyboard' bus='{add_input}'/>"""
diff --git a/vrtManager/test_util.py b/vrtManager/test_util.py
index fb174231..b1279563 100644
--- a/vrtManager/test_util.py
+++ b/vrtManager/test_util.py
@@ -123,6 +123,7 @@ class TestVrtManagerUtil(unittest.TestCase):
         self.assertEqual(vol_dev_type("ide"), "hd")
         self.assertEqual(vol_dev_type("fdc"), "fd")
         self.assertEqual(vol_dev_type("virtio"), "vd")
+        self.assertEqual(vol_dev_type("usb"), "sd")
         self.assertEqual(vol_dev_type("scsi"), "sd")
         self.assertEqual(vol_dev_type("sata"), "sd")
         self.assertIsNone(vol_dev_type("unknown_bus"))
diff --git a/vrtManager/util.py b/vrtManager/util.py
index 26ca01f9..de32a1d3 100644
--- a/vrtManager/util.py
+++ b/vrtManager/util.py
@@ -208,6 +208,6 @@ def vol_dev_type(vol_bus):
         return "fd"
     elif vol_bus == "virtio":
         return "vd"
-    elif vol_bus in ["scsi", "sata"]:
+    elif vol_bus in ["scsi", "sata", "usb"]:
         return "sd"