From 182746c73e2b100240068e2d5306dc35a97e2b01 Mon Sep 17 00:00:00 2001 From: catborise Date: Mon, 26 Oct 2026 12:07:45 +0300 Subject: [PATCH] fix(instances): an owner cannot power a VM an administrator paused Only administrators suspend and resume, and the page tells an owner of a paused VM "Administrator blocked your instance", but the power views let the owner force it off or cycle it, on the page and the API, and start it again: the block ended. The power permission now refuses a paused VM to everyone but administrators. --- instances/tests/test_paused_block.py | 44 ++++++++++++++++++++++++++++ instances/views.py | 20 +++++++++---- 2 files changed, 59 insertions(+), 5 deletions(-) create mode 100644 instances/tests/test_paused_block.py diff --git a/instances/tests/test_paused_block.py b/instances/tests/test_paused_block.py new file mode 100644 index 00000000..b76a7432 --- /dev/null +++ b/instances/tests/test_paused_block.py @@ -0,0 +1,44 @@ +"""A VM an administrator paused stays paused until an administrator resumes +it: its owners cannot power it off or cycle it, on the page or the API.""" + +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.test import TestCase +from django.urls import reverse +from instances.models import Instance +from libvirt import VIR_DOMAIN_PAUSED, VIR_DOMAIN_RUNNING + + +class PausedBlockTests(TestCase): + def setUp(self): + User = get_user_model() + self.owner = User.objects.create_user("pb-owner", password="pw") + self.admin = User.objects.create_superuser("pb-admin", "pb@example.com", "pw") + compute = Compute.objects.create(name="pb", hostname="192.0.2.1", login="u", password="p", type=1) + self.vm = Instance.objects.create(compute=compute, name="pb-vm", uuid="14141414-0000-0000-0000-000000000000") + UserInstance.objects.create(user=self.owner, instance=self.vm, is_change=True, is_delete=True) + patcher = patch("instances.models.wvmInstance") + self.proxy = patcher.start().return_value + self.addCleanup(patcher.stop) + + def test_an_owner_cannot_power_a_paused_vm(self): + self.proxy.get_status.return_value = VIR_DOMAIN_PAUSED + self.client.force_login(self.owner) + for view in ("force_off", "powercycle", "poweroff", "poweron"): + self.assertEqual(self.client.post(reverse(f"instances:{view}", args=[self.vm.pk])).status_code, 403, view) + api = reverse("compute-instance-forceoff", kwargs={"compute_pk": self.vm.compute_id, "pk": self.vm.pk}) + self.assertEqual(self.client.post(api).status_code, 403) + self.proxy.force_shutdown.assert_not_called() + self.proxy.shutdown.assert_not_called() + + def test_an_administrator_can_and_an_owner_of_a_running_vm_can(self): + self.proxy.get_status.return_value = VIR_DOMAIN_PAUSED + self.client.force_login(self.admin) + self.client.post(reverse("instances:force_off", args=[self.vm.pk])) + self.proxy.get_status.return_value = VIR_DOMAIN_RUNNING + self.client.force_login(self.owner) + self.client.post(reverse("instances:force_off", args=[self.vm.pk])) + self.assertEqual(self.proxy.force_shutdown.call_count, 2) diff --git a/instances/views.py b/instances/views.py index 6be749e2..d9ebffbe 100755 --- a/instances/views.py +++ b/instances/views.py @@ -33,7 +33,8 @@ from django.urls import reverse from django.utils import timezone from django.utils.translation import gettext_noop as _ from django.views.decorators.http import require_POST -from libvirt import (VIR_DOMAIN_UNDEFINE_KEEP_NVRAM, +from libvirt import (VIR_DOMAIN_PAUSED, + VIR_DOMAIN_UNDEFINE_KEEP_NVRAM, VIR_DOMAIN_UNDEFINE_MANAGED_SAVE, VIR_DOMAIN_UNDEFINE_SNAPSHOTS_METADATA, VIR_DOMAIN_UNDEFINE_NVRAM, @@ -546,6 +547,15 @@ def get_instance(user, pk, perm_type="view"): return instance +def _power_instance(user, pk): + """The VM to power on, off or cycle: a VM an administrator paused stays + paused until an administrator resumes it.""" + instance = get_instance(user, pk, perm_type="power") + if not user.is_superuser and instance.proxy.get_status() == VIR_DOMAIN_PAUSED: + raise PermissionDenied + return instance + + def from_script(request): """A form a page's script sends and answers itself: the create page's flavor forms (it swaps in the reply), the XML editor's dialog (JSON).""" @@ -684,7 +694,7 @@ def reject_disk_options(request, pk, invalid): @require_POST @serialize_instance_mutation def poweron(request, pk): - instance = get_instance(request.user, pk, perm_type="power") + instance = _power_instance(request.user, pk) if instance.is_template: messages.warning(request, _("Templates cannot be started.")) else: @@ -699,7 +709,7 @@ def poweron(request, pk): @require_POST @serialize_instance_mutation def powercycle(request, pk): - instance = get_instance(request.user, pk, perm_type="power") + instance = _power_instance(request.user, pk) if instance.is_template: messages.warning(request, _("Templates cannot be started.")) else: @@ -714,7 +724,7 @@ def powercycle(request, pk): @require_POST @serialize_instance_mutation def poweroff(request, pk): - instance = get_instance(request.user, pk, perm_type="power") + instance = _power_instance(request.user, pk) with audit_vm(request, instance, "power off"): instance.proxy.shutdown() @@ -750,7 +760,7 @@ def resume(request, pk): @require_POST @serialize_instance_mutation def force_off(request, pk): - instance = get_instance(request.user, pk, perm_type="power") + instance = _power_instance(request.user, pk) with audit_vm(request, instance, "force off"): instance.proxy.force_shutdown() return get_safe_redirect(