diff --git a/instances/tests/test_claim_after_permission.py b/instances/tests/test_claim_after_permission.py new file mode 100644 index 00000000..ea679a2e --- /dev/null +++ b/instances/tests/test_claim_after_permission.py @@ -0,0 +1,47 @@ +"""A request the permission check refuses takes no claim: it cannot hold up +other requests on the VM or its compute.""" + +from contextlib import contextmanager +from unittest.mock import patch + +from django.contrib.auth import get_user_model +from django.contrib.auth.models import Permission +from django.test import TestCase +from django.urls import reverse + +from accounts.models import UserInstance +from computes.models import Compute +from instances.models import Instance + + +class ClaimAfterPermissionTestCase(TestCase): + def setUp(self): + compute = Compute.objects.create(name="claim-c", hostname="192.0.2.80", login="root", password="", type=1) + self.vm = Instance.objects.create(compute=compute, name="claim-vm", uuid="aaaaaaaa-0000-0000-0000-000000000081") + self.claims = [] + + @contextmanager + def record(*args, **kwargs): + self.claims.append(args) + yield + + for name in ("libvirt_instance_lock", "libvirt_compute_lock"): + patcher = patch(f"instances.views.{name}", record) + patcher.start() + self.addCleanup(patcher.stop) + + def test_a_read_only_viewer_takes_no_claim(self): + viewer = get_user_model().objects.create_user("claim-viewer", password="x") + viewer.user_permissions.add(Permission.objects.get(codename="view_instances")) + self.client.force_login(viewer) + for view in ("poweron", "destroy"): + with self.subTest(view): + self.assertEqual(self.client.post(reverse(f"instances:{view}", args=[self.vm.pk])).status_code, 403) + self.assertEqual(self.claims, []) + + def test_an_owner_who_may_not_delete_takes_no_compute_claim(self): + owner = get_user_model().objects.create_user("claim-owner", password="x") + UserInstance.objects.create(instance=self.vm, user=owner, is_delete=False) + self.client.force_login(owner) + self.assertEqual(self.client.post(reverse("instances:destroy", args=[self.vm.pk])).status_code, 403) + self.assertEqual(self.claims, []) diff --git a/instances/views.py b/instances/views.py index e34415d0..8df852e2 100755 --- a/instances/views.py +++ b/instances/views.py @@ -645,10 +645,12 @@ def _moved_meanwhile(pk, inst): def serialize_instance_mutation(func): - """Serialize one VM, after checking visibility, without blocking sibling VMs.""" + """Serialize one VM, without blocking sibling VMs. Only a user who may + change the VM somehow (power, the least right) takes its claim: a + refused request must not hold up others.""" @functools.wraps(func) def wrapper(request, pk, *args, **kwargs): - inst = get_instance(request.user, pk) + inst = get_instance(request.user, pk, perm_type="power") try: with libvirt_instance_lock(inst): if _moved_meanwhile(pk, inst): @@ -685,10 +687,12 @@ def serialize_user_quota(func): def serialize_compute_mutation(func): - """Hold the whole compute: disk deletions must not race another VM attaching that disk.""" + """Hold the whole compute: disk deletions must not race another VM attaching that disk. + Only a user who may delete the VM (destroy's right; the other views are + superuser-only) takes the claim: a refused request must not hold up others.""" @functools.wraps(func) def wrapper(request, pk, *args, **kwargs): - inst = get_instance(request.user, pk) + inst = get_instance(request.user, pk, perm_type="delete") if request.method in ("GET", "HEAD"): # e.g. the destroy confirmation page return func(request, pk, *args, **kwargs) try: