j3d1/webvirtcloud
1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-08 19:26:48 +00:00
Code Issues Projects Releases Wiki Activity
webvirtcloud/instances/test_security_mutations.py
catborise 5c22fc8436 fix(instances): a finished migration is recorded, and a stale request does not migrate 
What migrate_instance did around the migration itself:

- It checked the destination with a 1 s TCP probe of the libvirt port,
  which passes while libvirt is down or refuses the login. It now uses
  the compute's connection status, as the rest of the app does.
- Two requests for the same VM waited for the compute locks one after
  the other; the second then migrated with the VM's old location. The
  VM's record is read again under the locks, and the request is
  refused when the VM is no longer on the source.
- Restoring autostart on the destination came after the migration but
  before the database move; when it failed, the VM ran on the
  destination while the app recorded it on the source. Autostart is
  now read from libvirt and a failure to restore it is only logged.
- A database failure after a successful migration was logged as
  nothing; the log now says the VM migrated and recording it failed.
- When the destination's list had already made a record for the VM,
  the merge kept the owners but dropped the template mark.
2026-10-06 15:52:32 +03:00

419 lines
18 KiB
Python
Raw Blame History

from unittest.mock import MagicMock, patch
from django.core.exceptions import PermissionDenied
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Permission
from django.http import Http404
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
from instances.views import get_instance
class InstanceSecurityMutationsTestCase(TestCase):
def setUp(self):
User = get_user_model()
self.superuser = User.objects.create_superuser(
username="sec_super", email="super@example.com", password="password"
)
self.owner = User.objects.create_user(
username="sec_owner", password="password"
)
self.change_user = User.objects.create_user(
username="sec_change_user", password="password"
)
self.viewer = User.objects.create_user(
username="sec_viewer", password="password"
)
# Viewer only has global view_instances permission
view_perm = Permission.objects.get(codename="view_instances")
self.viewer.user_permissions.add(view_perm)
self.compute = Compute.objects.create(
name="sec-compute",
hostname="127.0.0.1",
login="root",
password="",
type=1,
)
self.instance = Instance.objects.create(
compute=self.compute,
name="sec-test-vm",
uuid="11111111-2222-3333-4444-555555555555",
)
# Owner has basic userinstance (power enabled, change/delete disabled)
self.owner_ui = UserInstance.objects.create(
instance=self.instance, user=self.owner, is_change=False, is_delete=False
)
# Change user has is_change=True
self.change_ui = UserInstance.objects.create(
instance=self.instance, user=self.change_user, is_change=True, is_delete=False
)
# Mock instance.proxy on instance dict
self.mock_proxy = MagicMock()
self.instance.__dict__["proxy"] = self.mock_proxy
def test_power_endpoints_reject_get_requests(self):
self.client.force_login(self.owner)
endpoints = [
reverse("instances:poweron", args=[self.instance.id]),
reverse("instances:poweroff", args=[self.instance.id]),
reverse("instances:powercycle", args=[self.instance.id]),
reverse("instances:force_off", args=[self.instance.id]),
reverse("instances:suspend", args=[self.instance.id]),
reverse("instances:resume", args=[self.instance.id]),
]
for url in endpoints:
res = self.client.get(url)
self.assertEqual(res.status_code, 405, f"GET {url} should return 405")
def test_viewer_without_ownership_cannot_power_mutate(self):
self.client.force_login(self.viewer)
endpoints = [
reverse("instances:poweron", args=[self.instance.id]),
reverse("instances:poweroff", args=[self.instance.id]),
reverse("instances:powercycle", args=[self.instance.id]),
reverse("instances:force_off", args=[self.instance.id]),
]
for url in endpoints:
res = self.client.post(url)
self.assertEqual(res.status_code, 403, f"POST {url} should be 403 for read-only viewer")
def test_owner_can_power_mutate(self):
self.client.force_login(self.owner)
with patch.object(Instance, "proxy", new_callable=lambda: self.mock_proxy):
res = self.client.post(reverse("instances:poweron", args=[self.instance.id]))
self.assertEqual(res.status_code, 302)
def test_set_root_pass_requires_is_change(self):
# Owner without is_change -> 403
self.client.force_login(self.owner)
res = self.client.post(reverse("instances:rootpasswd", args=[self.instance.id]), {"passwd": "newpass"})
self.assertEqual(res.status_code, 403)
# Viewer without ownership -> 403
self.client.force_login(self.viewer)
res_v = self.client.post(reverse("instances:rootpasswd", args=[self.instance.id]), {"passwd": "newpass"})
self.assertEqual(res_v.status_code, 403)
def test_drf_compute_pk_mismatch_returns_404(self):
other_compute = Compute.objects.create(
name="other-compute", hostname="127.0.0.2", login="root", type=1
)
self.client.force_login(self.owner)
# Query instance via wrong compute_pk in nested router
url = f"/api/v1/computes/{other_compute.id}/instances/{self.instance.id}/"
res = self.client.get(url)
self.assertEqual(res.status_code, 404)
def test_disk_media_and_snapshots_reject_get_requests(self):
self.client.force_login(self.change_user)
endpoints = [
reverse("instances:add_new_vol", args=[self.instance.id]),
reverse("instances:add_existing_vol", args=[self.instance.id]),
reverse("instances:edit_volume", args=[self.instance.id]),
reverse("instances:delete_vol", args=[self.instance.id]),
reverse("instances:detach_vol", args=[self.instance.id]),
reverse("instances:add_cdrom", args=[self.instance.id]),
reverse("instances:detach_cdrom", args=[self.instance.id, "hda"]),
reverse("instances:mount_iso", args=[self.instance.id]),
reverse("instances:unmount_iso", args=[self.instance.id]),
reverse("instances:snapshot", args=[self.instance.id]),
reverse("instances:delete_snapshot", args=[self.instance.id]),
reverse("instances:revert_snapshot", args=[self.instance.id]),
]
for url in endpoints:
res = self.client.get(url)
self.assertEqual(res.status_code, 405, f"GET {url} should return 405")
def test_disk_and_media_mutations_require_is_change(self):
# Viewer (global read-only) -> 403
self.client.force_login(self.viewer)
urls_to_test = [
reverse("instances:add_new_vol", args=[self.instance.id]),
reverse("instances:detach_cdrom", args=[self.instance.id, "hda"]),
reverse("instances:mount_iso", args=[self.instance.id]),
reverse("instances:delete_vol", args=[self.instance.id]),
]
for url in urls_to_test:
res = self.client.post(url, {})
self.assertEqual(res.status_code, 403, f"Viewer POST {url} should be 403")
# Owner without is_change (is_change=False, is_delete=False) -> 403
self.client.force_login(self.owner)
for url in urls_to_test:
res = self.client.post(url, {})
self.assertEqual(res.status_code, 403, f"Owner without is_change POST {url} should be 403")
def test_resize_views_reject_get_requests(self):
self.client.force_login(self.change_user)
endpoints = [
reverse("instances:resize_disk", args=[self.instance.id]),
reverse("instances:resizevm_cpu", args=[self.instance.id]),
reverse("instances:resize_memory", args=[self.instance.id]),
]
for url in endpoints:
res = self.client.get(url)
self.assertEqual(res.status_code, 405, f"GET {url} should return 405")
def test_resize_views_require_is_change(self):
self.client.force_login(self.viewer)
endpoints = [
reverse("instances:resize_disk", args=[self.instance.id]),
reverse("instances:resizevm_cpu", args=[self.instance.id]),
reverse("instances:resize_memory", args=[self.instance.id]),
]
for url in endpoints:
res = self.client.post(url, {})
self.assertEqual(res.status_code, 403, f"Viewer POST {url} should return 403")
self.client.force_login(self.owner)
for url in endpoints:
res = self.client.post(url, {})
self.assertEqual(res.status_code, 403, f"Owner without is_change POST {url} should return 403")
def test_get_instance_fail_closed_for_all_users_on_invalid_perm(self):
with self.assertRaises(PermissionDenied):
get_instance(self.superuser, self.instance.pk, perm_type="unknown_perm")
with self.assertRaises(PermissionDenied):
get_instance(self.owner, self.instance.pk, perm_type="unknown_perm")
def test_get_instance_staff_user_access(self):
User = get_user_model()
staff_user = User.objects.create_user(username="sec_staff", password="password", is_staff=True)
# Staff user without ownership or view_instances permission cannot view instance
with self.assertRaises(Http404):
get_instance(staff_user, self.instance.pk, perm_type="view")
# Staff user with explicit view_instances permission can view instance
view_perm = Permission.objects.get(codename="view_instances")
staff_user.user_permissions.add(view_perm)
staff_user = User.objects.get(pk=staff_user.pk)
inst = get_instance(staff_user, self.instance.pk, perm_type="view")
self.assertEqual(inst.pk, self.instance.pk)
# Staff user with view permission but without ownership cannot perform power, change, or delete operations
for perm in ["power", "change", "delete"]:
with self.assertRaises(PermissionDenied):
get_instance(staff_user, self.instance.pk, perm_type=perm)
def test_migrate_endpoint_requires_post(self):
self.client.force_login(self.superuser)
res = self.client.get(reverse("instances:migrate", args=[self.instance.id]))
self.assertEqual(res.status_code, 405)
def test_post_only_mutation_endpoints(self):
self.client.force_login(self.change_user)
# All mutation endpoints reject GET with 405 Method Not Allowed
endpoints_405 = ["change_options", "update_console", "rootpasswd", "add_public_key"]
for ep in endpoints_405:
res = self.client.get(reverse(f"instances:{ep}", args=[self.instance.id]))
self.assertEqual(res.status_code, 405, f"{ep} allowed GET request instead of 405")
def test_opposite_direction_migrations_no_deadlock(self):
import threading
import time
from instances.utils import migrate_instance
c1 = MagicMock()
c1.id = 101
c1.pk = 101
c1.hostname = "127.0.0.1"
c1.type = 1
c1.login = "root"
c1.password = ""
c2 = MagicMock()
c2.id = 102
c2.pk = 102
c2.hostname = "127.0.0.2"
c2.type = 1
c2.login = "root"
c2.password = ""
inst1 = MagicMock()
inst1.id = "inst1"
inst1.compute = c1
inst1.name = "inst1"
inst1.autostart = False
inst1.proxy = MagicMock()
inst1.uuid = "u1"
inst2 = MagicMock()
inst2.id = "inst2"
inst2.compute = c2
inst2.name = "inst2"
inst2.autostart = False
inst2.proxy = MagicMock()
inst2.uuid = "u2"
barrier = threading.Barrier(2)
errors = []
def worker_1_to_2():
try:
barrier.wait(timeout=5)
migrate_instance(c2, inst1, self.superuser)
except Exception as e:
errors.append(("worker_1_to_2", e))
def worker_2_to_1():
try:
barrier.wait(timeout=5)
migrate_instance(c1, inst2, self.superuser)
except Exception as e:
errors.append(("worker_2_to_1", e))
def slow_moveto(*args, **kwargs):
time.sleep(0.05)
with patch("instances.utils.wvmInstances") as mock_instances_cls, \
patch("instances.utils.wvmInstance"), \
patch("instances.utils.transaction.atomic"), \
patch("instances.utils.addlogmsg"), \
patch("instances.utils.Instance") as mock_instance_model:
mock_instances_cls.return_value.moveto.side_effect = slow_moveto
# under the locks each VM is still on its compute; the
# destination has no record of it
def rows(**query):
found = MagicMock(compute_id={"inst1": 101, "inst2": 102}[query["id"]]) if "id" in query else None
return MagicMock(**{"first.return_value": found})
mock_instance_model.objects.filter.side_effect = rows
t1 = threading.Thread(target=worker_1_to_2)
t2 = threading.Thread(target=worker_2_to_1)
t1.start()
t2.start()
t1.join(timeout=10)
t2.join(timeout=10)
self.assertFalse(t1.is_alive(), "Worker 1 to 2 timed out / deadlocked")
self.assertFalse(t2.is_alive(), "Worker 2 to 1 timed out / deadlocked")
self.assertEqual(errors, [], f"Unexpected errors during concurrent migrations: {errors}")
def test_resize_disk_empty_selection_warns_cleanly(self):
self.client.force_login(self.change_user)
with patch("instances.models.wvmInstance") as mock_wvm_cls:
mock_wvm_cls.return_value.get_disk_devices.return_value = []
res = self.client.post(reverse("instances:resize_disk", args=[self.instance.id]), {})
self.assertEqual(res.status_code, 302)
self.assertIn(f"/instances/{self.instance.id}/#resize", res.url)
def test_disk_mutation_denied_with_is_change(self):
# change_user has is_change=True, is_delete=False. Disk views are
# superuser-only, so is_change alone is not enough.
self.client.force_login(self.change_user)
with patch("instances.models.wvmInstance"), patch("instances.views.wvmStorage") as mock_storage:
res = self.client.post(
reverse("instances:delete_vol", args=[self.instance.id]),
{"storage": "default", "dev": "vda", "path": "/path"},
HTTP_REFERER=f"/instances/{self.instance.id}/",
)
self.assertEqual(res.status_code, 403)
mock_storage.return_value.del_volume.assert_not_called()
def test_destroy_vm_authorization(self):
# 1. User with is_change=True but is_delete=False gets 403 when trying to destroy VM
self.client.force_login(self.change_user)
res = self.client.post(reverse("instances:destroy", args=[self.instance.id]), {})
self.assertEqual(res.status_code, 403)
# 2. Staff user without ownership or is_delete cannot destroy VM
User = get_user_model()
staff_user = User.objects.create_user(username="staff_destroyer", password="password", is_staff=True)
self.client.force_login(staff_user)
res_staff = self.client.post(reverse("instances:destroy", args=[self.instance.id]), {})
self.assertEqual(res_staff.status_code, 404)
# 2b. Staff user with view_instances permission but is_delete=False gets 403
view_perm = Permission.objects.get(codename="view_instances")
staff_user.user_permissions.add(view_perm)
res_staff_view = self.client.post(reverse("instances:destroy", args=[self.instance.id]), {})
self.assertEqual(res_staff_view.status_code, 403)
# 3. Superuser can destroy VM
self.client.force_login(self.superuser)
with patch("instances.models.wvmInstance") as mock_wvm_cls:
mock_wvm_cls.return_value.get_status.return_value = 5 # shutoff
res = self.client.post(reverse("instances:destroy", args=[self.instance.id]), {})
self.assertEqual(res.status_code, 302)
self.assertFalse(Instance.objects.filter(id=self.instance.id).exists())
def test_migration_deduplicate_instances_logic(self):
import importlib
migration_mod = importlib.import_module("instances.migrations.0012_instance_unique_compute_uuid")
dedup_fn = getattr(migration_mod, "deduplicate_instances")
mock_apps = MagicMock()
mock_instance_model = MagicMock()
mock_userinstance_model = MagicMock()
def get_model(app_label, model_name):
if model_name == "Instance":
return mock_instance_model
elif model_name == "UserInstance":
return mock_userinstance_model
return MagicMock()
mock_apps.get_model = get_model
empty_inst = MagicMock(uuid="")
primary_inst = MagicMock(id=1)
dup_inst = MagicMock(id=2)
def mock_inst_filter(*args, **kwargs):
if args:
# Called for empty_uuid_instances query
return [empty_inst]
# Called with compute_id and uuid
ret = MagicMock()
ret.order_by.return_value = [primary_inst, dup_inst]
return ret
mock_instance_model.objects.filter.side_effect = mock_inst_filter
mock_instance_model.objects.exclude.return_value.exclude.return_value.values.return_value.annotate.return_value.filter.return_value = [
{"compute_id": 1, "uuid": "test-uuid-1234"}
]
primary_ui = MagicMock(is_change=False, is_delete=False, is_vnc=False)
dup_ui = MagicMock(user="user1", is_change=True, is_delete=False, is_vnc=True)
mock_userinstance_model.objects.filter.side_effect = lambda **kwargs: (
[dup_ui] if kwargs.get("instance") == dup_inst else
MagicMock(first=lambda: primary_ui)
)
dedup_fn(mock_apps, None)
self.assertTrue(len(empty_inst.uuid) > 0)
empty_inst.save.assert_called_once_with(update_fields=['uuid'])
dup_inst.delete.assert_called_once()
dup_ui.delete.assert_called_once()
self.assertTrue(primary_ui.is_change)
self.assertTrue(primary_ui.is_vnc)
self.assertFalse(primary_ui.is_delete)
primary_ui.save.assert_called_once()
def test_unique_compute_uuid_db_constraint_enforced(self):
from django.db import IntegrityError
with self.assertRaises(IntegrityError):
Instance.objects.create(
compute=self.compute,
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()
Reference in a new issue View git blame Copy permalink