From 51dedb7b8f87a7a5f8b306a9ed60b11a1daf9293 Mon Sep 17 00:00:00 2001 From: catborise Date: Sat, 3 Oct 2026 19:11:44 +0300 Subject: [PATCH] fix(instances): disk edits keep their XML; changes lock per VM Disk edits: - edit_disk built a new from the form and replaced the old one, losing what the form does not show (e.g. encryption and virtio driver options); a bus change detached and re-attached the disk. The disk of the persistent definition is now edited in place and the domain defined once; a bus change picks a free device name and drops the old address and virtio-only options. Pending persistent changes of a running VM are kept. - Clone omits empty disk devices, refuses a disk without a supported source and destination, and checks the name before allocating storage. Locks: - Every VM change held its compute's exclusive lock, so one slow operation blocked all VMs on that compute. A VM change now holds its own VM lock plus a shared compute lock; reconciliation, migration and cloning hold the compute exclusively (two clones of different VMs must not pass the name check together). - Quota-checked changes of one user (CPU, memory and disk resize, clone) run one at a time under a per-user lock, also across computes, so concurrent requests cannot all pass the check against the old allocation. It is the outermost lock; superusers, who have no quota, skip it. - A busy lock answers 409 to the API and a message in the web UI. API power and destroy actions return the view's error instead of success. --- computes/test_api_security.py | 12 ++ computes/test_operation_locks.py | 60 ++++++++ computes/tests.py | 2 +- computes/utils.py | 143 +++++++++---------- instances/api/viewsets.py | 28 ++-- instances/test_clone_permission.py | 16 ++- instances/test_disk_isolation.py | 13 +- instances/test_quota_lock.py | 62 ++++++++ instances/test_security_mutations.py | 9 ++ instances/views.py | 66 ++++++--- vrtManager/instance.py | 164 ++++++++++----------- vrtManager/test_xml_safety.py | 204 ++++++++++++++++++--------- 12 files changed, 510 insertions(+), 269 deletions(-) create mode 100644 computes/test_operation_locks.py create mode 100644 instances/test_quota_lock.py diff --git a/computes/test_api_security.py b/computes/test_api_security.py index 8d73d177..0ac10bda 100644 --- a/computes/test_api_security.py +++ b/computes/test_api_security.py @@ -175,3 +175,15 @@ class APISecurityTestCase(TestCase): self.client.force_login(self.staff_user) res = self.client.get(f"/api/v1/computes/{self.compute.id}/storages/") self.assertEqual(res.status_code, 403) + + def test_busy_instance_api_does_not_report_success(self): + self.client.force_login(self.regular_user) + with patch("instances.views.libvirt_instance_lock", side_effect=TimeoutError("busy")), patch( + "instances.models.wvmInstance" + ) as proxy: + response = self.client.post( + f"/api/v1/computes/{self.compute.pk}/instances/{self.instance1.pk}/poweron/", {} + ) + self.assertEqual(response.status_code, 409) + self.assertIn("busy", response.json()["detail"]) + proxy.assert_not_called() diff --git a/computes/test_operation_locks.py b/computes/test_operation_locks.py new file mode 100644 index 00000000..e9e50622 --- /dev/null +++ b/computes/test_operation_locks.py @@ -0,0 +1,60 @@ +import threading +from types import SimpleNamespace + +from django.test import SimpleTestCase + +from computes.utils import libvirt_compute_lock, libvirt_instance_lock + + +class OperationLocksTestCase(SimpleTestCase): + def setUp(self): + self.compute = SimpleNamespace(pk=987654321) + self.first = SimpleNamespace(pk=987654321, compute=self.compute) + self.second = SimpleNamespace(pk=987654322, compute=self.compute) + + def test_sibling_vm_can_run_while_same_vm_and_reconcile_are_excluded(self): + acquired = threading.Event() + release = threading.Event() + errors = [] + + def hold_first_vm(): + try: + with libvirt_instance_lock(self.first): + acquired.set() + if not release.wait(5): + raise TimeoutError("Test did not release the VM lock") + except Exception as err: + errors.append(err) + + worker = threading.Thread(target=hold_first_vm) + worker.start() + try: + self.assertTrue(acquired.wait(2)) + with libvirt_instance_lock(self.second, timeout=0.1): + pass + with self.assertRaises(TimeoutError): + with libvirt_instance_lock(self.first, timeout=0.05): + pass + with self.assertRaises(TimeoutError): + with libvirt_compute_lock(self.compute, timeout=0.05): + pass + finally: + release.set() + worker.join(5) + self.assertFalse(worker.is_alive()) + self.assertEqual(errors, []) + with libvirt_compute_lock(self.compute, timeout=0.1): + pass + + def test_instance_lock_is_reentrant_without_releasing_the_outer_lock(self): + with libvirt_instance_lock(self.first): + with libvirt_instance_lock(self.first): + pass + with self.assertRaisesRegex(RuntimeError, "Cannot upgrade"): + with libvirt_compute_lock(self.compute, timeout=0.05): + pass + + def test_compute_exclusive_owner_can_take_an_instance_lock(self): + with libvirt_compute_lock(self.compute): + with libvirt_instance_lock(self.first): + pass diff --git a/computes/tests.py b/computes/tests.py index 9a7b82bd..bf034203 100644 --- a/computes/tests.py +++ b/computes/tests.py @@ -276,7 +276,7 @@ class ComputeConcurrencyTestCase(TransactionTestCase): return curr_time[0] with patch("fcntl.flock", side_effect=BlockingIOError("Resource temporarily unavailable")): - with patch("time.time", side_effect=advance_time): + with patch("time.monotonic", side_effect=advance_time): refresh_instance_database(compute) self.assertEqual(Instance.objects.filter(compute=compute).count(), 1) diff --git a/computes/utils.py b/computes/utils.py index 1fa40628..97bd13a0 100644 --- a/computes/utils.py +++ b/computes/utils.py @@ -51,7 +51,7 @@ def _get_lock_directory(): if (stat.S_IMODE(stat_info.st_mode) & 0o077) != 0: continue - test_file = os.path.join(candidate, f".test_write_{os.getpid()}") + test_file = os.path.join(candidate, f".test_write_{os.getpid()}_{threading.get_ident()}") with open(test_file, "w") as tf: tf.write("1") os.unlink(test_file) @@ -71,85 +71,80 @@ def _get_thread_held_locks(): @contextmanager -def libvirt_compute_lock(compute, timeout=15.0): - """ - Context manager providing multi-thread and multi-process synchronization - for libvirt compute operations and instance database synchronization. - Supports re-entrant locking within the same thread. - Fails closed on lock directory or permission errors. - """ +def _libvirt_lock(key, filename, timeout=15.0, shared=False): + """Use flock for both threads and processes; exclusive locks also use a mutex.""" + held_locks = _get_thread_held_locks() + if key in held_locks: + if held_locks[key] and not shared: + raise RuntimeError("Cannot upgrade a shared compute lock inside a VM operation") + yield + return + + thread_lock = None if shared else _get_compute_thread_lock(key) + start_time = time.monotonic() + if thread_lock is not None and not thread_lock.acquire(timeout=timeout): + raise TimeoutError(f"Timeout waiting for lock on {key}") + try: + lock_dir = _get_lock_directory() + if not lock_dir: + raise OSError(f"Could not acquire secure lock directory for {key}") + path = os.path.join(lock_dir, filename) + if os.path.islink(path): + raise PermissionError(f"Lock file {path} is a symlink") + fd = os.open(path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) + acquired = False + try: + mode = fcntl.LOCK_SH if shared else fcntl.LOCK_EX + while True: + try: + fcntl.flock(fd, mode | fcntl.LOCK_NB) + acquired = True + break + except BlockingIOError: + if time.monotonic() - start_time >= timeout: + raise TimeoutError(f"Timeout waiting for lock on {key}") + time.sleep(0.05) + held_locks[key] = shared + try: + yield + finally: + held_locks.pop(key, None) + finally: + try: + if acquired: + fcntl.flock(fd, fcntl.LOCK_UN) + finally: + os.close(fd) + finally: + if thread_lock is not None: + thread_lock.release() + + +@contextmanager +def libvirt_compute_lock(compute, timeout=15.0, *, shared=False): + """Exclude reconciliation/migration while allowing independent VM operations.""" compute_pk = getattr(compute, "pk", getattr(compute, "id", None)) if not compute_pk: yield return + with _libvirt_lock(compute_pk, f"sync_compute_{compute_pk}.lock", timeout, shared): + yield - held_locks = _get_thread_held_locks() - if compute_pk in held_locks: - held_locks[compute_pk] += 1 - try: + +@contextmanager +def user_quota_lock(user, timeout=15.0): + """One user's quota check and the change it allows run one at a time, on every compute. + Take it before any compute or VM lock.""" + with _libvirt_lock(("quota", user.pk), f"quota_user_{user.pk}.lock", timeout): + yield + + +@contextmanager +def libvirt_instance_lock(instance, timeout=15.0): + """Serialize one VM and share the compute barrier with other VM operations.""" + with libvirt_compute_lock(instance.compute, timeout, shared=True): + with _libvirt_lock(("instance", instance.pk), f"instance_{instance.pk}.lock", timeout): yield - finally: - held_locks[compute_pk] -= 1 - if held_locks[compute_pk] <= 0: - held_locks.pop(compute_pk, None) - return - - thread_lock = _get_compute_thread_lock(compute_pk) - start_time = time.time() - acquired_thread = thread_lock.acquire(timeout=timeout) - if not acquired_thread: - logger.error("Timeout waiting for in-process thread lock on compute %s; aborting to fail closed", compute_pk) - raise TimeoutError(f"Timeout waiting for lock on compute {compute_pk}") - - try: - lock_dir = _get_lock_directory() - if not lock_dir: - logger.error("Failed to acquire application lock directory for compute %s; aborting to fail closed", compute_pk) - raise OSError(f"Could not acquire secure lock directory for compute {compute_pk}") - - lock_file_path = os.path.join(lock_dir, f"sync_compute_{compute_pk}.lock") - if os.path.islink(lock_file_path): - logger.error("Lock file %s is a symlink! Aborting to fail closed", lock_file_path) - raise PermissionError(f"Compute lock file {lock_file_path} is a symlink") - - flags = os.O_CREAT | os.O_RDWR | getattr(os, "O_NOFOLLOW", 0) - try: - lock_fd = os.open(lock_file_path, flags, 0o600) - except OSError as e: - logger.error("Could not open synchronization lock file for compute %s: %s; aborting to fail closed", compute_pk, e) - raise - - lock_acquired = False - try: - while time.time() - start_time < timeout: - try: - fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) - lock_acquired = True - break - except (BlockingIOError, OSError): - time.sleep(0.05) - - if not lock_acquired: - logger.error("Timeout waiting for cross-process synchronization lock on compute %s; aborting to fail closed", compute_pk) - raise TimeoutError(f"Timeout waiting for lock on compute {compute_pk}") - - held_locks[compute_pk] = 1 - try: - yield - finally: - held_locks.pop(compute_pk, None) - finally: - if lock_acquired: - try: - fcntl.flock(lock_fd, fcntl.LOCK_UN) - except OSError: - pass - try: - os.close(lock_fd) - except OSError: - pass - finally: - thread_lock.release() def refresh_instance_database(compute): diff --git a/instances/api/viewsets.py b/instances/api/viewsets.py index 2c8fd36a..cbe92ff9 100644 --- a/instances/api/viewsets.py +++ b/instances/api/viewsets.py @@ -33,6 +33,13 @@ from .serializers import ( ) +def instance_action_response(action, request, pk, message): + response = action(request, pk) + if response.status_code >= 400: + return response + return Response({"status": message}) + + class InstancesViewSet(viewsets.ViewSet): """ A simple ViewSet for listing or retrieving ALL/Compute Instances. @@ -117,8 +124,7 @@ class InstanceViewSet(viewsets.ViewSet): queryset = get_instance(request.user, pk, perm_type="delete") if compute_pk is not None and str(queryset.compute_id) != str(compute_pk): raise Http404() - instance_destroy(request, pk) - return Response({"status": "Instance is destroyed"}) + return instance_action_response(instance_destroy, request, pk, "Instance is destroyed") @extend_schema(request=None, responses=StatusSerializer) @action(detail=True, methods=["post"]) @@ -126,8 +132,7 @@ class InstanceViewSet(viewsets.ViewSet): queryset = get_instance(request.user, pk, perm_type="power") if compute_pk is not None and str(queryset.compute_id) != str(compute_pk): raise Http404() - poweron(request, pk) - return Response({"status": "poweron command send"}) + return instance_action_response(poweron, request, pk, "poweron command send") @extend_schema(request=None, responses=StatusSerializer) @action(detail=True, methods=["post"]) @@ -135,8 +140,7 @@ class InstanceViewSet(viewsets.ViewSet): queryset = get_instance(request.user, pk, perm_type="power") if compute_pk is not None and str(queryset.compute_id) != str(compute_pk): raise Http404() - poweroff(request, pk) - return Response({"status": "poweroff command send"}) + return instance_action_response(poweroff, request, pk, "poweroff command send") @extend_schema(request=None, responses=StatusSerializer) @action(detail=True, methods=["post"]) @@ -144,8 +148,7 @@ class InstanceViewSet(viewsets.ViewSet): queryset = get_instance(request.user, pk, perm_type="power") if compute_pk is not None and str(queryset.compute_id) != str(compute_pk): raise Http404() - powercycle(request, pk) - return Response({"status": "powercycle command send"}) + return instance_action_response(powercycle, request, pk, "powercycle command send") @extend_schema(request=None, responses=StatusSerializer) @action(detail=True, methods=["post"]) @@ -153,8 +156,7 @@ class InstanceViewSet(viewsets.ViewSet): queryset = get_instance(request.user, pk, perm_type="power") if compute_pk is not None and str(queryset.compute_id) != str(compute_pk): raise Http404() - force_off(request, pk) - return Response({"status": "force off command send"}) + return instance_action_response(force_off, request, pk, "force off command send") @extend_schema(request=None, responses=StatusSerializer) @action(detail=True, methods=["post"]) @@ -162,8 +164,7 @@ class InstanceViewSet(viewsets.ViewSet): queryset = get_instance(request.user, pk) if compute_pk is not None and str(queryset.compute_id) != str(compute_pk): raise Http404() - suspend(request, pk) - return Response({"status": "suspend command send"}) + return instance_action_response(suspend, request, pk, "suspend command send") @extend_schema(request=None, responses=StatusSerializer) @action(detail=True, methods=["post"]) @@ -171,8 +172,7 @@ class InstanceViewSet(viewsets.ViewSet): queryset = get_instance(request.user, pk) if compute_pk is not None and str(queryset.compute_id) != str(compute_pk): raise Http404() - resume(request, pk) - return Response({"status": "resume command send"}) + return instance_action_response(resume, request, pk, "resume command send") class MigrateViewSet(viewsets.ViewSet): diff --git a/instances/test_clone_permission.py b/instances/test_clone_permission.py index 9794c8fe..f517872b 100644 --- a/instances/test_clone_permission.py +++ b/instances/test_clone_permission.py @@ -156,6 +156,20 @@ class ClonePermissionTestCase(TestCase): ) proxy.clone_instance.assert_not_called() + def test_clone_holds_the_compute_exclusively(self): + # Clones of different VMs must not race for the destination name. + calls = [] + + @contextmanager + def record(compute, timeout=15.0, *, shared=False): + calls.append(shared) + yield + + with self.mocked_libvirt() as proxy, patch("instances.views.libvirt_compute_lock", record): + self._clone(self.superuser, self._valid_post()) + proxy.clone_instance.assert_called_once() + self.assertEqual(calls, [False]) + def test_superuser_valid_clone_passes_posted_values(self): with self.mocked_libvirt() as proxy: self._clone( @@ -170,7 +184,7 @@ class ClonePermissionTestCase(TestCase): def test_non_superuser_clone_skips_disks_without_a_volume(self): empty_disk = {"dev": "vdb", "image": None, "storage": None, "path": None, - "size": 0, "format": None} + "size": None, "format": None} with self.mocked_libvirt() as proxy: proxy.get_disk_devices.return_value = [SRC_DISK, empty_disk] self._clone(self.owner_change, self._valid_post()) diff --git a/instances/test_disk_isolation.py b/instances/test_disk_isolation.py index 951551a8..2d0044ad 100644 --- a/instances/test_disk_isolation.py +++ b/instances/test_disk_isolation.py @@ -262,25 +262,20 @@ class DiskViewsTenantIsolationTestCase(TestCase): return self._post("edit_volume", data) def test_edit_volume_bus_change_keeps_the_disk_and_applies_options(self): - # autospec enforces the real attach_disk signature; a plain MagicMock - # would accept the invalid driver_type argument. from vrtManager.instance import wvmInstance self.client.force_login(self.superuser) with patch("instances.models.wvmInstance", autospec=wvmInstance) as mock_wvm: proxy = mock_wvm.return_value self._mock_disk_options(proxy) - proxy.get_status.return_value = 5 + proxy.get_status.return_value = 1 res = self._edit_volume_post(vol_bus="sata") self.assertEqual(res.status_code, 302) - # The disk keeps its type and source; only target dev/bus change, then - # the other options are applied to the disk on its new device. - proxy.change_disk_bus.assert_called_once_with("vda", proxy.change_disk_bus.call_args[0][1], "sata") - new_dev = proxy.change_disk_bus.call_args[0][1] proxy.attach_disk.assert_not_called() + proxy.detach_disk.assert_not_called() proxy.edit_disk.assert_called_once() - self.assertEqual(proxy.edit_disk.call_args[0][0], new_dev) - self.assertEqual(proxy.edit_disk.call_args[0][4], "sata") + self.assertEqual(proxy.edit_disk.call_args.args[0], "vda") + self.assertEqual(proxy.edit_disk.call_args.args[4], "sata") def test_edit_volume_keeps_current_format_when_field_is_none_or_empty(self): self.client.force_login(self.superuser) diff --git a/instances/test_quota_lock.py b/instances/test_quota_lock.py new file mode 100644 index 00000000..214a718d --- /dev/null +++ b/instances/test_quota_lock.py @@ -0,0 +1,62 @@ +import threading +import time +from unittest.mock import patch + +from accounts.models import UserInstance +from computes.models import Compute +from django.contrib.auth import get_user_model +from django.db import connection +from django.test import Client, TransactionTestCase, override_settings +from django.urls import reverse +from instances.models import Instance + +QUOTA_CPUS = 4 + + +# Sessions and messages in cookies: concurrent writes to the in-memory test +# database would fail for reasons unrelated to the quota race. +@override_settings( + SESSION_ENGINE="django.contrib.sessions.backends.signed_cookies", + MESSAGE_STORAGE="django.contrib.messages.storage.cookie.CookieStorage", +) +class QuotaRaceTestCase(TransactionTestCase): + """Two concurrent resizes of one user's VMs must not both pass the quota.""" + + def setUp(self): + self.user = get_user_model().objects.create_user("quota_user", password="pw") + compute = Compute.objects.create(name="quota-compute", hostname="127.0.0.1", login="", password="", type=4) + self.vms = [ + Instance.objects.create(compute=compute, name=f"quota-vm-{i}", uuid=f"1111111{i}-2222-3333-4444-555555555555") + for i in range(2) + ] + for vm in self.vms: + UserInstance.objects.create(user=self.user, instance=vm, is_change=True) + self.allocated = 2 # two VMs with one vCPU each + + def fake_quota(self, user, instance, cpu, memory, disk): + seen = self.allocated + time.sleep(0.3) # widen the window between the check and the change + return "cpu" if seen + cpu > QUOTA_CPUS else "" + + def fake_resize(self, cur_vcpu, vcpu): + self.allocated += int(vcpu) - 1 + + def resize(self, vm): + client = Client() + client.force_login(self.user) + client.post(reverse("instances:resizevm_cpu", args=[vm.id]), {"vcpu": 3, "cur_vcpu": 3}) + connection.close() + + def test_second_concurrent_resize_is_refused(self): + with patch("instances.views.utils.check_user_quota", self.fake_quota), patch( + "instances.models.wvmInstance" + ) as wvm: + wvm.return_value.get_vcpu.return_value = 1 + wvm.return_value.resize_cpu.side_effect = self.fake_resize + threads = [threading.Thread(target=self.resize, args=(vm,)) for vm in self.vms] + for t in threads: + t.start() + for t in threads: + t.join() + self.assertEqual(wvm.return_value.resize_cpu.call_count, 1) + self.assertLessEqual(self.allocated, QUOTA_CPUS) diff --git a/instances/test_security_mutations.py b/instances/test_security_mutations.py index e8af20ae..d9f20954 100644 --- a/instances/test_security_mutations.py +++ b/instances/test_security_mutations.py @@ -400,3 +400,12 @@ class InstanceSecurityMutationsTestCase(TestCase): name="duplicate_vm", uuid=self.instance.uuid, ) + + def test_busy_instance_redirects_without_executing_the_operation(self): + self.client.force_login(self.owner) + with patch("instances.views.libvirt_instance_lock", side_effect=TimeoutError("busy")), patch( + "instances.models.wvmInstance" + ) as proxy: + response = self.client.post(reverse("instances:poweron", args=[self.instance.pk]), {}) + self.assertRedirects(response, reverse("instances:instance", args=[self.instance.pk]), fetch_redirect_response=False) + proxy.assert_not_called() diff --git a/instances/views.py b/instances/views.py index 75c85eee..8e7962a3 100755 --- a/instances/views.py +++ b/instances/views.py @@ -16,7 +16,7 @@ from admin.decorators import superuser_only from appsettings.models import AppSettings from appsettings.settings import app_settings from computes.models import Compute -from computes.utils import libvirt_compute_lock +from computes.utils import libvirt_compute_lock, libvirt_instance_lock, user_quota_lock from django.conf import settings from django.contrib import messages from django.contrib.auth.decorators import permission_required @@ -373,20 +373,39 @@ def get_instance(user, pk, perm_type="view"): return instance +def _busy(request, pk): + if hasattr(request, "accepted_renderer"): + return JsonResponse({"detail": "Instance operation is busy. Please retry."}, status=409) + messages.error(request, _("Instance operation is busy. Please retry.")) + return get_safe_redirect(request, default=reverse("instances:instance", args=[pk])) + + def serialize_instance_mutation(func): - """ - Decorator that ensures any view mutating an instance's libvirt state - acquires libvirt_compute_lock for the instance's compute. - Works re-entrantly with inner calls such as refresh_instance_database. - """ + """Serialize one VM, after checking visibility, without blocking sibling VMs.""" @functools.wraps(func) def wrapper(request, pk, *args, **kwargs): + inst = get_instance(request.user, pk) try: - inst = Instance.objects.only("compute_id").get(pk=pk) - with libvirt_compute_lock(inst.compute): + with libvirt_instance_lock(inst): return func(request, pk, *args, **kwargs) - except Instance.DoesNotExist: + except TimeoutError: + return _busy(request, pk) + + return wrapper + + +def serialize_user_quota(func): + """Outermost lock for quota-checked changes: concurrent requests of one + user (also on different computes) must not pass the check together.""" + @functools.wraps(func) + def wrapper(request, pk, *args, **kwargs): + if request.user.is_superuser: # quotas do not apply return func(request, pk, *args, **kwargs) + try: + with user_quota_lock(request.user): + return func(request, pk, *args, **kwargs) + except TimeoutError: + return _busy(request, pk) return wrapper @@ -436,7 +455,7 @@ def reject_disk_options(request, invalid): request, _("Invalid disk options: %(options)s") % {"options": ", ".join(invalid)}, ) - return redirect(request.META.get("HTTP_REFERER") + "#disks") + return get_safe_redirect(request) @require_POST @@ -521,10 +540,11 @@ def force_off(request, pk): ) +@serialize_instance_mutation def destroy(request, pk): if request.method in ["POST", "DELETE"]: instance = get_instance(request.user, pk, perm_type="delete") - with libvirt_compute_lock(instance.compute): + with libvirt_instance_lock(instance): if instance.proxy.get_status() == 1: instance.proxy.force_shutdown() @@ -694,6 +714,7 @@ def add_public_key(request, pk): @require_POST +@serialize_user_quota @serialize_instance_mutation def resizevm_cpu(request, pk): instance = get_instance(request.user, pk, perm_type="change") @@ -729,6 +750,7 @@ def resizevm_cpu(request, pk): @require_POST +@serialize_user_quota @serialize_instance_mutation def resize_memory(request, pk): instance = get_instance(request.user, pk, perm_type="change") @@ -773,6 +795,7 @@ def resize_memory(request, pk): @require_POST +@serialize_user_quota @serialize_instance_mutation def resize_disk(request, pk): instance = get_instance(request.user, pk, perm_type="change") @@ -958,9 +981,9 @@ def edit_volume(request, pk): new_path = request.POST.get("vol_path", "") shareable = bool(request.POST.get("vol_shareable", False)) readonly = bool(request.POST.get("vol_readonly", False)) - disk_type = request.POST.get("vol_type", "") - bus = request.POST.get("vol_bus_old", "") - new_bus = request.POST.get("vol_bus", bus) + disks = instance.proxy.get_disk_devices(config=True) + current = next((disk for disk in disks if disk["dev"] == target_dev), {}) + new_bus = request.POST.get("vol_bus", current.get("bus")) serial = request.POST.get("vol_serial", "") format = request.POST.get("vol_format", "") cache = request.POST.get( @@ -975,7 +998,6 @@ def edit_volume(request, pk): ) # The form shows "None" (or nothing) for a disk without driver type. if format in ("", "None"): - current = next((d for d in instance.disks if d["dev"] == target_dev), {}) format = current.get("format") or "" invalid = invalid_disk_options( @@ -989,16 +1011,11 @@ def edit_volume(request, pk): format=format or None, serial=serial, ) - if target_dev not in [disk["dev"] for disk in instance.disks]: + if not current: invalid.append("dev") if invalid: return reject_disk_options(request, invalid) - new_target_dev = utils.get_new_disk_dev(instance.media, instance.disks, new_bus) - - if new_bus != bus: - instance.proxy.change_disk_bus(target_dev, new_target_dev, new_bus) - target_dev = new_target_dev instance.proxy.edit_disk( target_dev, new_path, @@ -1609,6 +1626,7 @@ CLONE_POST_KEY_RE = re.compile(r"^(clone-net-mac-\d+|disk-[a-z0-9]+|meta-[a-z0-9 @require_POST @permission_required("instances.clone_instances", raise_exception=True) +@serialize_user_quota def clone(request, pk): # Cloning copies the source disks, so it needs change permission on the # source VM. Templates are meant to be deployed from, so viewing is enough. @@ -1621,7 +1639,7 @@ def clone(request, pk): clone_data["clone-title"] = request.POST.get("clone-title", "").strip() clone_data["clone-description"] = request.POST.get("clone-description", "").strip() - disk_sum = sum([disk["size"] >> 30 for disk in instance.disks]) + disk_sum = sum(int(disk.get("size") or 0) >> 30 for disk in instance.disks) quota_msg = utils.check_user_quota( request.user, 1, instance.vcpu, instance.memory, disk_sum ) @@ -1702,6 +1720,8 @@ def clone(request, pk): messages.error(request, msg) else: try: + # The whole compute: the destination name and disk names must not be + # taken by a concurrent clone of another VM while volumes are copied. with libvirt_compute_lock(instance.compute): new_uuid = instance.proxy.clone_instance(clone_data) new_instance = Instance.objects.get_or_create( @@ -1735,7 +1755,7 @@ def clone(request, pk): except Exception as e: messages.error(request, e) - return redirect(request.META.get("HTTP_REFERER") + "#clone") + return get_safe_redirect(request, default=reverse("instances:instance", args=[pk])) @require_POST diff --git a/vrtManager/instance.py b/vrtManager/instance.py index 6c418d7d..ef9c91e4 100644 --- a/vrtManager/instance.py +++ b/vrtManager/instance.py @@ -2,6 +2,7 @@ import contextlib import json import logging import os.path +import string import time logger = logging.getLogger(__name__) @@ -13,6 +14,7 @@ try: VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_AGENT, VIR_DOMAIN_RUNNING, VIR_DOMAIN_XML_SECURE, + VIR_DOMAIN_XML_INACTIVE, VIR_MIGRATE_AUTO_CONVERGE, VIR_MIGRATE_COMPRESSED, VIR_MIGRATE_LIVE, @@ -450,7 +452,7 @@ class wvmInstance(wvmConnect): return util.get_xml_path(self._XMLDesc(0), func=networks) - def get_disk_devices(self): + def get_disk_devices(self, config=False): def disks(doc): result = [] @@ -537,7 +539,8 @@ class wvmInstance(wvmConnect): ) return result - return util.get_xml_path(self._XMLDesc(0), func=disks) + flags = VIR_DOMAIN_XML_INACTIVE | VIR_DOMAIN_XML_SECURE if config else 0 + return util.get_xml_path(self._XMLDesc(flags), func=disks) def get_media_devices(self): def disks(doc): @@ -844,32 +847,6 @@ class wvmInstance(wvmConnect): if self.get_status() == 5: self.instance.detachDeviceFlags(xml_disk, VIR_DOMAIN_AFFECT_CONFIG) - def change_disk_bus(self, target_dev, new_target_dev, new_bus): - """ - Move a disk to another bus in the persistent config. The disk element - is reused as is (type, source, auth, driver), only changes - and the old bus address is dropped. If attaching fails, the original - disk is attached again. - """ - tree = etree.fromstring(self._XMLDesc(0)) - disk_el = tree.xpath("./devices/disk/target[@dev=$dev]", dev=target_dev)[ - 0 - ].getparent() - old_xml = etree.tostring(disk_el).decode() - - disk_el.find("target").set("dev", new_target_dev) - disk_el.find("target").set("bus", new_bus) - for address in disk_el.findall("address"): - disk_el.remove(address) - new_xml = etree.tostring(disk_el).decode() - - self.instance.detachDeviceFlags(old_xml, VIR_DOMAIN_AFFECT_CONFIG) - try: - self.instance.attachDeviceFlags(new_xml, VIR_DOMAIN_AFFECT_CONFIG) - except libvirtError: - self.instance.attachDeviceFlags(old_xml, VIR_DOMAIN_AFFECT_CONFIG) - raise - def edit_disk( self, target_dev, @@ -884,61 +861,70 @@ class wvmInstance(wvmConnect): discard_mode, detect_zeroes_mode, ): - tree = etree.fromstring(self._XMLDesc(0)) - disk_el = tree.xpath("./devices/disk/target[@dev=$dev]", dev=target_dev)[ - 0 - ].getparent() - old_disk_type = disk_el.get("type") - old_disk_device = disk_el.get("device") - old_driver_name = disk_el.xpath("driver/@name")[0] - old_target_bus = disk_el.xpath("target/@bus")[0] + """Edit the persistent disk in place and commit its domain XML once.""" + tree = etree.fromstring( + self._XMLDesc(VIR_DOMAIN_XML_INACTIVE | VIR_DOMAIN_XML_SECURE) + ) + matches = tree.xpath("./devices/disk/target[@dev=$dev]", dev=target_dev) + if not matches: + raise ValueError(f"Disk {target_dev} is not in the persistent configuration") + target = matches[0] + disk = target.getparent() - # Values below are interpolated into disk XML; escape them all. - esc = util.xml_escape - target_dev, source, target_bus = esc(target_dev), esc(source), esc(target_bus) - serial, format = esc(serial), esc(format) - cache_mode, io_mode = esc(cache_mode), esc(io_mode) - discard_mode, detect_zeroes_mode = esc(discard_mode), esc(detect_zeroes_mode) - old_disk_type, old_disk_device = esc(old_disk_type), esc(old_disk_device) - old_driver_name = esc(old_driver_name) + if target.get("bus") != target_bus: + prefix = {"virtio": "vd", "ide": "hd", "fdc": "fd"}.get(target_bus, "sd") + used = set(tree.xpath("./devices/disk/target/@dev")) - {target_dev} + new_dev = target_dev if target_dev.startswith(prefix) else next( + (prefix + letter for letter in string.ascii_lowercase if prefix + letter not in used), + None, + ) + if new_dev is None: + raise ValueError("No available device name for the requested disk bus") + target.set("dev", new_dev) + target.set("bus", target_bus) + for address in disk.findall("address"): + disk.remove(address) - additionals = "" - if cache_mode is not None and cache_mode != "default": - additionals += f"cache='{cache_mode}' " - if io_mode is not None and io_mode != "default": - additionals += f"io='{io_mode}' " - if discard_mode is not None and discard_mode != "default": - additionals += f"discard='{discard_mode}' " - if detect_zeroes_mode is not None and detect_zeroes_mode != "default": - additionals += f"detect_zeroes='{detect_zeroes_mode}' " + driver = disk.find("driver") + if driver is None: + driver = etree.Element("driver", name="qemu") + disk.insert(0, driver) + if target_bus != "virtio": + for key in ("queues", "queue_size", "iothread", "iommu", "ats", "packed", "page_per_vq"): + driver.attrib.pop(key, None) + for iothreads in driver.findall("iothreads"): + driver.remove(iothreads) + disk.attrib.pop("model", None) + for key, value in ( + ("type", format), ("cache", cache_mode), ("io", io_mode), + ("discard", discard_mode), ("detect_zeroes", detect_zeroes_mode), + ): + if value and value != "default": + driver.set(key, value) + else: + driver.attrib.pop(key, None) - xml_disk = f"" - # A disk without a driver type keeps having none. - type_attr = f"type='{format}' " if format else "" - if old_disk_device == "cdrom": - xml_disk += f"" - elif old_disk_device == "disk": - xml_disk += f"" + if disk.get("type") == "file": + source_el = disk.find("source") + if source_el is None: + source_el = etree.SubElement(disk, "source") + source_el.set("file", source) - if disk_el.get("type") == "file": - xml_disk += f"" - else: - # Block/network disks: keep their source (and auth) untouched; - # the form only edits the path of file disks. - for name in ("auth", "source"): - el = disk_el.find(name) - if el is not None: - xml_disk += etree.tostring(el).decode() - xml_disk += f"" - if readonly: - xml_disk += """""" - if shareable: - xml_disk += """""" - if serial is not None and serial != "None" and serial != "": - xml_disk += f"""{serial}""" - xml_disk += """""" + for tag, enabled in (("readonly", readonly), ("shareable", shareable)): + element = disk.find(tag) + if enabled and element is None: + etree.SubElement(disk, tag) + elif not enabled and element is not None: + disk.remove(element) + serial_el = disk.find("serial") + if serial and serial != "None": + if serial_el is None: + serial_el = etree.SubElement(disk, "serial") + serial_el.text = serial + elif serial_el is not None: + disk.remove(serial_el) - self.instance.updateDeviceFlags(xml_disk, VIR_DOMAIN_AFFECT_CONFIG) + self._defineXML(etree.tostring(tree).decode()) def cpu_usage(self): cpu_usage = {} @@ -1510,12 +1496,17 @@ class wvmInstance(wvmConnect): def clone_instance(self, clone_data): clone_dev_path = [] + if clone_data["name"] in self.get_instances(): + raise ValueError("An instance with the clone name already exists") + xml = self._XMLDesc(VIR_DOMAIN_XML_SECURE) tree = etree.fromstring(xml) name = tree.find("name") name.text = clone_data["name"] uuid = tree.find("uuid") - tree.remove(uuid) + if uuid is None: + uuid = etree.SubElement(tree, "uuid") + uuid.text = util.randomUUID() options = { "title": clone_data.get("clone-title", ""), @@ -1523,6 +1514,21 @@ class wvmInstance(wvmConnect): } self._set_options(tree, options) + for disk in list(tree.findall("devices/disk")): + if disk.get("device") != "disk": + continue + source = disk.find("source") + if source is None or not any(source.get(key) for key in ("file", "dev", "name", "volume")): + disk.getparent().remove(disk) + continue + supported = source.get("file") or source.get("dev") or ( + source.get("protocol") == "rbd" and source.get("name") + ) + target = disk.find("target") + dev = target.get("dev") if target is not None else None + if not supported or not dev or not clone_data.get("disk-" + dev): + raise ValueError("Cannot clone a disk without a supported source and destination") + src_nvram_path = self.get_nvram() if src_nvram_path: # Change XML for nvram diff --git a/vrtManager/test_xml_safety.py b/vrtManager/test_xml_safety.py index 91d3ac13..4dca4cf9 100644 --- a/vrtManager/test_xml_safety.py +++ b/vrtManager/test_xml_safety.py @@ -185,6 +185,7 @@ class TestDiskXmlEscaping(unittest.TestCase): inst = wvmInstance.__new__(wvmInstance) inst.get_status = MagicMock(return_value=5) # shutoff inst.instance = MagicMock() + inst._defineXML = MagicMock() return inst def test_attach_disk_escapes_source_and_serial(self): @@ -225,8 +226,9 @@ class TestDiskXmlEscaping(unittest.TestCase): "default", "default", ) - xml = inst.instance.updateDeviceFlags.call_args[0][0] - self.assertIn("file='/images/b'/><x y=''", xml) + xml = inst._defineXML.call_args.args[0] + from lxml import etree + self.assertEqual(etree.fromstring(xml).find("devices/disk/source").get("file"), "/images/b'/>" - "" - "" - "" - "" - "
" - "" - ) - - def _instance(self): - inst = wvmInstance.__new__(wvmInstance) - inst.instance = MagicMock() - inst._XMLDesc = MagicMock(return_value=self.RBD_DOMAIN) - return inst - - def test_keeps_disk_source_and_auth_and_only_changes_target(self): +class TestPersistentDiskEditing(unittest.TestCase): + def setUp(self): + import libvirt from lxml import etree - inst = self._instance() - inst.change_disk_bus("vda", "sda", "sata") + self.conn = libvirt.open("test:///default") + self.addCleanup(self.conn.close) + self.domain = self.conn.lookupByName("test") + original = self.domain.XMLDesc(libvirt.VIR_DOMAIN_XML_INACTIVE | libvirt.VIR_DOMAIN_XML_SECURE) + self.addCleanup(self.conn.defineXML, original) + config = etree.fromstring(original) + disk = config.find("devices/disk") + disk.set("type", "file") + driver = disk.find("driver") + if driver is None: + driver = etree.SubElement(disk, "driver") + driver.set("name", "qemu") + driver.set("type", "qcow2") + driver.set("queues", "4") + etree.SubElement(disk, "address", type="pci", slot="0x05") + encryption = etree.SubElement(disk, "encryption", format="luks") + etree.SubElement(encryption, "secret", type="passphrase", uuid="aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee") + etree.SubElement(etree.SubElement(disk, "iotune"), "total_bytes_sec").text = "1024" + # A pending change and a VNC password outside the edited disk must survive. + config.find("memory").text = "123456" + etree.SubElement(config.find("devices"), "graphics", type="vnc", passwd="keep-secret") + self.conn.defineXML(etree.tostring(config).decode()) + self.inst = wvmInstance.__new__(wvmInstance) + self.inst.wvm = self.conn + self.inst.instance = self.domain + self.target = disk.find("target").get("dev") - new_xml = inst.instance.attachDeviceFlags.call_args[0][0] - disk = etree.fromstring(new_xml) - self.assertEqual(disk.get("type"), "network") - self.assertEqual(disk.find("source").get("name"), "pool/vm-disk") - self.assertEqual(disk.find("auth/secret").get("uuid"), "aaaa") - self.assertEqual(disk.find("target").get("dev"), "sda") + def config(self): + import libvirt + from lxml import etree + return etree.fromstring(self.domain.XMLDesc(libvirt.VIR_DOMAIN_XML_INACTIVE | libvirt.VIR_DOMAIN_XML_SECURE)) + + def edit(self, bus="virtio", format="qcow2"): + self.inst.edit_disk( + self.target, "/images/a'&.qcow2", False, False, bus, + "serial<&", format, "none", "native", "unmap", "on", + ) + + def test_running_bus_change_preserves_disk_and_pending_domain_configuration(self): + from lxml import etree + self.assertTrue(self.domain.isActive()) + live_before = self.domain.XMLDesc(0) + self.edit(bus="sata") + config = self.config() + disk = config.find("devices/disk") self.assertEqual(disk.find("target").get("bus"), "sata") - self.assertIsNone(disk.find("address")) + self.assertTrue(disk.find("target").get("dev").startswith("sd")) + self.assertNotEqual(disk.find("address").get("type"), "pci") + self.assertIsNone(disk.find("driver").get("queues")) + self.assertEqual(disk.find("driver").get("cache"), "none") + self.assertEqual(disk.findtext("iotune/total_bytes_sec"), "1024") + self.assertEqual(disk.find("encryption/secret").get("uuid"), "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee") + self.assertEqual(disk.find("source").get("file"), "/images/a'&.qcow2") + self.assertEqual(disk.findtext("serial"), "serial<&") + self.assertEqual(config.findtext("memory"), "123456") + self.assertEqual(config.find("devices/graphics").get("passwd"), "keep-secret") + self.assertEqual(self.domain.XMLDesc(0), live_before) + # The resulting disk can be edited again before the VM restarts. + self.target = disk.find("target").get("dev") + self.edit(bus="sata") + self.assertEqual(self.config().find("devices/disk/target").get("dev"), self.target) - def test_reattaches_the_old_disk_when_attach_fails(self): + def test_failed_definition_leaves_the_original_configuration_intact(self): from libvirt import libvirtError + from unittest.mock import patch + before = self.domain.XMLDesc(2) + with patch.object(self.inst, "_defineXML", side_effect=libvirtError("definition failed")): + with self.assertRaises(libvirtError): + self.edit(bus="sata") + self.assertEqual(self.domain.XMLDesc(2), before) - inst = self._instance() - inst.instance.attachDeviceFlags.side_effect = [libvirtError("boom"), None] + def test_network_disk_keeps_auth_source_and_encryption(self): + from lxml import etree + config = self.config() + disk = config.find("devices/disk") + disk.set("type", "network") + disk.remove(disk.find("source")) + source = etree.SubElement(disk, "source", protocol="rbd", name="pool/vm-disk") + etree.SubElement(source, "host", name="mon1", port="6789") + auth = etree.SubElement(disk, "auth", username="libvirt") + etree.SubElement(auth, "secret", type="ceph", uuid="aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee") + self.conn.defineXML(etree.tostring(config).decode()) + self.edit(bus="sata") + result = self.config().find("devices/disk") + self.assertEqual(result.find("source").get("name"), "pool/vm-disk") + self.assertEqual(result.find("source/host").get("name"), "mon1") + self.assertEqual(result.find("auth").get("username"), "libvirt") + self.assertIsNotNone(result.find("encryption")) - with self.assertRaises(libvirtError): - inst.change_disk_bus("vda", "sda", "sata") + def test_empty_format_omits_driver_type(self): + self.edit(format="") + self.assertIsNone(self.config().find("devices/disk/driver").get("type")) - detached_xml = inst.instance.detachDeviceFlags.call_args[0][0] - restored_xml = inst.instance.attachDeviceFlags.call_args_list[1][0][0] - self.assertEqual(restored_xml, detached_xml) + def test_virtio_options_survive_editing_on_the_same_bus(self): + self.edit() + self.assertEqual(self.config().find("devices/disk/driver").get("queues"), "4") - def test_edit_disk_keeps_source_and_auth_of_a_network_disk(self): - from lxml import etree - - inst = self._instance() - inst.edit_disk( - "vda", "ignored-path", False, False, "virtio", "", "raw", - "default", "default", "default", "default", - ) - - disk = etree.fromstring(inst.instance.updateDeviceFlags.call_args[0][0]) - self.assertEqual(disk.get("type"), "network") - self.assertEqual(disk.find("source").get("name"), "pool/vm-disk") - self.assertEqual(disk.find("source/host").get("name"), "mon1") - self.assertEqual(disk.find("auth/secret").get("uuid"), "aaaa") - - def test_edit_disk_without_format_omits_the_driver_type(self): + +class TestEmptyDiskClone(unittest.TestCase): + def test_clone_omits_empty_disks_before_real_libvirt_definition(self): + import libvirt from lxml import etree + conn = libvirt.open("test:///default") + self.addCleanup(conn.close) + domain = conn.lookupByName("test") + config = etree.fromstring(domain.XMLDesc(libvirt.VIR_DOMAIN_XML_INACTIVE)) + devices = config.find("devices") + for disk in devices.findall("disk"): + source = disk.find("source") + if source is not None: + disk.remove(source) + for interface in devices.findall("interface"): + devices.remove(interface) inst = wvmInstance.__new__(wvmInstance) - inst.instance = MagicMock() - inst._XMLDesc = MagicMock( - return_value=( - "" - "" - "" - ) - ) - inst.edit_disk( - "vda", "/images/vm.img", False, False, "virtio", "", "", - "default", "default", "default", "default", - ) - disk = etree.fromstring(inst.instance.updateDeviceFlags.call_args[0][0]) - self.assertIsNone(disk.find("driver").get("type")) + inst.wvm = conn + inst.instance = domain + # Exercise defensive handling of incomplete disk metadata, while the + # clone is still defined and looked up through the real test driver. + inst._XMLDesc = MagicMock(return_value=etree.tostring(config).decode()) + clone_uuid = inst.clone_instance({"name": "empty-disk-clone"}) + clone = conn.lookupByUUIDString(clone_uuid) + self.addCleanup(clone.undefine) + clone_config = etree.fromstring(clone.XMLDesc(libvirt.VIR_DOMAIN_XML_INACTIVE)) + self.assertEqual(clone_config.findtext("name"), "empty-disk-clone") + self.assertEqual(clone_config.findall("devices/disk"), []) + self.assertTrue(config.findall("devices/disk")) + + def test_clone_rejects_existing_name_before_allocating_storage(self): + import libvirt + + conn = libvirt.open("test:///default") + self.addCleanup(conn.close) + inst = wvmInstance.__new__(wvmInstance) + inst.wvm = conn + inst.instance = conn.lookupByName("test") + inst._XMLDesc = MagicMock() + with self.assertRaisesRegex(ValueError, "already exists"): + inst.clone_instance({"name": "test"}) + inst._XMLDesc.assert_not_called()