j3d1/webvirtcloud
1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-08 11:16:47 +00:00
Code Issues Projects Releases Wiki Activity

fix(instances): CD-ROM bus, USB controller and MACs of a new VM

Browse source 
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.
This commit is contained in:
catborise 2026-10-04 18:33:28 +03:00
parent 525018dd99
commit 03ac93ea53
9 changed files with 130 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

8
instances/api/serializers.py
View file

@ -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 = [

6
instances/templates/create_instance_w2.html
View file

@ -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>

75
instances/test_create_instance.py
View file

@ -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)

24
instances/test_live_create.py
View file

@ -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/",

14
instances/utils.py
View file

@ -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),

15
instances/views.py
View file

@ -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"]:

11
vrtManager/create.py
View file

@ -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}'/>"""

1
vrtManager/test_util.py
View file

@ -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"))

2
vrtManager/util.py
View file

@ -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"