1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-08 11:16:47 +00:00

fix(instances): a request refused by the permission check takes no claim

The VM and compute claim decorators checked only that the user sees the
VM, took the claim, and left the action's own right to the view inside:
a read-only viewer, or an owner who may not delete the VM, briefly held
the VM's or the whole compute's claim and could make other requests
there busy. They now check the least right first: power for a VM change,
delete for the compute-wide claim (destroy; its other views are
superuser-only).
This commit is contained in:
catborise 2026-11-05 16:37:18 +03:00
parent 7d82cc7d6b
commit d97e44b77f
2 changed files with 55 additions and 4 deletions

View file

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

View file

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