mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 19:26:48 +00:00
f94b3cee7a
Comments, docstrings and live test names cited item IDs of a roadmap that is not part of the repository. They now say what is meant, or nothing where the sentence stands without them. Test assertions and behavior are unchanged.
311 lines
13 KiB
Python
311 lines
13 KiB
Python
"""
|
|
Disk and media mutation views must be superuser-only (matching the UI, where
|
|
the Disk tab is only rendered for superusers), and even for superusers the
|
|
views must not act on volumes or option values taken blindly from the client.
|
|
"""
|
|
from unittest.mock import patch
|
|
|
|
from django.contrib.auth import get_user_model
|
|
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
|
|
|
|
DISK_ENDPOINTS = [
|
|
("add_new_vol", []),
|
|
("add_existing_vol", []),
|
|
("edit_volume", []),
|
|
("delete_vol", []),
|
|
("detach_vol", []),
|
|
("add_cdrom", []),
|
|
("detach_cdrom", ["hda"]),
|
|
("mount_iso", []),
|
|
("unmount_iso", []),
|
|
]
|
|
|
|
VM_DISK = {
|
|
"dev": "vda",
|
|
"bus": "virtio",
|
|
"image": "vm-disk.qcow2",
|
|
"storage": "default",
|
|
"path": "/var/lib/libvirt/images/vm-disk.qcow2",
|
|
"format": "qcow2",
|
|
}
|
|
|
|
|
|
class DiskViewsTenantIsolationTestCase(TestCase):
|
|
def setUp(self):
|
|
User = get_user_model()
|
|
self.superuser = User.objects.create_superuser(
|
|
username="disk_super", email="disk_super@example.com", password="password"
|
|
)
|
|
self.change_user = User.objects.create_user(
|
|
username="disk_change_user", password="password"
|
|
)
|
|
self.staff_change_user = User.objects.create_user(
|
|
username="disk_staff_change_user", password="password", is_staff=True
|
|
)
|
|
self.compute = Compute.objects.create(
|
|
name="disk-compute", hostname="127.0.0.1", login="root", password="", type=1
|
|
)
|
|
self.instance = Instance.objects.create(
|
|
compute=self.compute,
|
|
name="disk-test-vm",
|
|
uuid="aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
|
|
)
|
|
for user in (self.change_user, self.staff_change_user):
|
|
UserInstance.objects.create(
|
|
instance=self.instance, user=user, is_change=True, is_delete=False
|
|
)
|
|
self.referer = f"/instances/{self.instance.id}/"
|
|
|
|
def _url(self, name, extra=()):
|
|
return reverse(f"instances:{name}", args=[self.instance.id, *extra])
|
|
|
|
def _post(self, name, data, extra=()):
|
|
return self.client.post(self._url(name, extra), data, HTTP_REFERER=self.referer)
|
|
|
|
def test_owner_with_is_change_cannot_mutate_disks(self):
|
|
for user in (self.change_user, self.staff_change_user):
|
|
self.client.force_login(user)
|
|
with patch("instances.models.wvmInstance") as mock_wvm, patch(
|
|
"instances.views.wvmStorage"
|
|
) as mock_storage, patch("instances.views.wvmCreate") as mock_create:
|
|
for name, extra in DISK_ENDPOINTS:
|
|
res = self._post(
|
|
name,
|
|
{"storage": "default", "name": "victim.qcow2", "dev": "vda"},
|
|
extra,
|
|
)
|
|
self.assertEqual(
|
|
res.status_code, 403, f"{user.username} POST {name} should be 403"
|
|
)
|
|
mock_wvm.assert_not_called()
|
|
mock_storage.assert_not_called()
|
|
mock_create.assert_not_called()
|
|
|
|
def test_delete_vol_deletes_the_volume_attached_to_dev_not_the_posted_one(self):
|
|
self.client.force_login(self.superuser)
|
|
with patch("instances.models.wvmInstance") as mock_wvm, patch(
|
|
"instances.views.wvmStorage"
|
|
) as mock_storage:
|
|
mock_wvm.return_value.get_disk_devices.return_value = [VM_DISK]
|
|
res = self._post(
|
|
"delete_vol",
|
|
{"storage": "other-pool", "name": "victim.qcow2", "dev": "vda"},
|
|
)
|
|
self.assertEqual(res.status_code, 302)
|
|
mock_wvm.return_value.detach_disk.assert_called_once_with("vda")
|
|
self.assertEqual(mock_storage.call_args[0][-1], "default")
|
|
mock_storage.return_value.del_volume.assert_called_once_with("vm-disk.qcow2")
|
|
|
|
def test_delete_vol_refuses_dev_not_attached_to_vm(self):
|
|
self.client.force_login(self.superuser)
|
|
with patch("instances.models.wvmInstance") as mock_wvm, patch(
|
|
"instances.views.wvmStorage"
|
|
) as mock_storage:
|
|
mock_wvm.return_value.get_disk_devices.return_value = [VM_DISK]
|
|
res = self._post(
|
|
"delete_vol",
|
|
{"storage": "default", "name": "victim.qcow2", "dev": "vdz"},
|
|
)
|
|
self.assertEqual(res.status_code, 302)
|
|
mock_wvm.return_value.detach_disk.assert_not_called()
|
|
mock_storage.return_value.del_volume.assert_not_called()
|
|
|
|
def test_edit_volume_rejects_unknown_cache_mode(self):
|
|
self.client.force_login(self.superuser)
|
|
with patch("instances.models.wvmInstance") as mock_wvm:
|
|
proxy = mock_wvm.return_value
|
|
proxy.get_disk_devices.return_value = [VM_DISK]
|
|
proxy.get_media_devices.return_value = []
|
|
proxy.get_disk_bus_types.return_value = ["virtio", "sata", "scsi"]
|
|
proxy.get_cache_modes.return_value = {"default": "", "none": ""}
|
|
proxy.get_io_modes.return_value = {"default": "", "native": ""}
|
|
proxy.get_discard_modes.return_value = {"default": "", "unmap": ""}
|
|
proxy.get_detect_zeroes_modes.return_value = {"default": "", "on": ""}
|
|
res = self._post(
|
|
"edit_volume",
|
|
{
|
|
"edit_volume": "1",
|
|
"dev": "vda",
|
|
"vol_path": VM_DISK["path"],
|
|
"vol_bus_old": "virtio",
|
|
"vol_bus": "virtio",
|
|
"vol_format": "qcow2",
|
|
"vol_cache": "none' io='native",
|
|
},
|
|
)
|
|
self.assertEqual(res.status_code, 302)
|
|
proxy.edit_disk.assert_not_called()
|
|
proxy.attach_disk.assert_not_called()
|
|
proxy.detach_disk.assert_not_called()
|
|
|
|
def test_add_new_vol_rejects_unknown_bus(self):
|
|
self.client.force_login(self.superuser)
|
|
with patch("instances.models.wvmInstance") as mock_wvm, patch(
|
|
"instances.views.wvmCreate"
|
|
) as mock_create, patch("instances.views.wvmStorage"):
|
|
proxy = mock_wvm.return_value
|
|
proxy.get_disk_devices.return_value = [VM_DISK]
|
|
proxy.get_media_devices.return_value = []
|
|
proxy.get_disk_bus_types.return_value = ["virtio", "sata", "scsi"]
|
|
proxy.get_cache_modes.return_value = {"default": "", "none": ""}
|
|
res = self._post(
|
|
"add_new_vol",
|
|
{
|
|
"storage": "default",
|
|
"name": "new-disk",
|
|
"format": "qcow2",
|
|
"size": "10",
|
|
"bus": "virtio'/><disk",
|
|
"cache": "default",
|
|
},
|
|
)
|
|
self.assertEqual(res.status_code, 302)
|
|
mock_create.return_value.create_volume.assert_not_called()
|
|
proxy.attach_disk.assert_not_called()
|
|
|
|
def test_add_existing_vol_rejects_volume_not_in_pool(self):
|
|
self.client.force_login(self.superuser)
|
|
with patch("instances.models.wvmInstance") as mock_wvm, patch(
|
|
"instances.views.wvmStorage"
|
|
) as mock_storage:
|
|
proxy = mock_wvm.return_value
|
|
proxy.get_disk_devices.return_value = [VM_DISK]
|
|
proxy.get_media_devices.return_value = []
|
|
proxy.get_disk_bus_types.return_value = ["virtio", "sata", "scsi"]
|
|
proxy.get_cache_modes.return_value = {"default": "", "none": ""}
|
|
mock_storage.return_value.get_volumes.return_value = ["vm-disk.qcow2"]
|
|
res = self._post(
|
|
"add_existing_vol",
|
|
{
|
|
"selected_storage": "default",
|
|
"vols": "../../../etc/shadow",
|
|
"bus": "virtio",
|
|
"cache": "default",
|
|
},
|
|
)
|
|
self.assertEqual(res.status_code, 302)
|
|
proxy.attach_disk.assert_not_called()
|
|
|
|
def _mock_disk_options(self, proxy):
|
|
proxy.get_disk_devices.return_value = [VM_DISK]
|
|
proxy.get_media_devices.return_value = []
|
|
proxy.get_disk_bus_types.return_value = ["virtio", "sata", "scsi"]
|
|
proxy.get_cache_modes.return_value = {"default": "", "none": ""}
|
|
proxy.get_io_modes.return_value = {"default": "", "native": ""}
|
|
proxy.get_discard_modes.return_value = {"default": "", "unmap": ""}
|
|
proxy.get_detect_zeroes_modes.return_value = {"default": "", "on": ""}
|
|
|
|
def test_add_new_vol_rejects_unsafe_volume_name(self):
|
|
self.client.force_login(self.superuser)
|
|
with patch("instances.models.wvmInstance") as mock_wvm, patch(
|
|
"instances.views.wvmCreate"
|
|
) as mock_create, patch("instances.views.wvmStorage"):
|
|
self._mock_disk_options(mock_wvm.return_value)
|
|
res = self._post(
|
|
"add_new_vol",
|
|
{
|
|
"storage": "default",
|
|
"name": "x</name><key>/etc/shadow",
|
|
"format": "qcow2",
|
|
"size": "10",
|
|
"bus": "virtio",
|
|
"cache": "default",
|
|
},
|
|
)
|
|
self.assertEqual(res.status_code, 302)
|
|
mock_create.return_value.create_volume.assert_not_called()
|
|
|
|
def test_edit_volume_accepts_valid_options(self):
|
|
self.client.force_login(self.superuser)
|
|
with patch("instances.models.wvmInstance") as mock_wvm:
|
|
proxy = mock_wvm.return_value
|
|
self._mock_disk_options(proxy)
|
|
res = self._post(
|
|
"edit_volume",
|
|
{
|
|
"edit_volume": "1",
|
|
"dev": "vda",
|
|
"vol_path": VM_DISK["path"],
|
|
"vol_bus_old": "virtio",
|
|
"vol_bus": "virtio",
|
|
"vol_format": "qcow2",
|
|
"vol_serial": "SER-01",
|
|
"vol_cache": "none",
|
|
"vol_io_mode": "native",
|
|
"vol_discard_mode": "unmap",
|
|
"vol_detect_zeroes": "on",
|
|
},
|
|
)
|
|
self.assertEqual(res.status_code, 302)
|
|
proxy.edit_disk.assert_called_once_with(
|
|
"vda", VM_DISK["path"], False, False, "virtio", "SER-01", "qcow2",
|
|
"none", "native", "unmap", "on",
|
|
)
|
|
|
|
def _edit_volume_post(self, **extra):
|
|
data = {
|
|
"edit_volume": "1",
|
|
"dev": "vda",
|
|
"vol_path": VM_DISK["path"],
|
|
"vol_bus_old": "virtio",
|
|
"vol_bus": "virtio",
|
|
"vol_format": "qcow2",
|
|
"vol_cache": "none",
|
|
}
|
|
data.update(extra)
|
|
return self._post("edit_volume", data)
|
|
|
|
def test_edit_volume_bus_change_keeps_the_disk_and_applies_options(self):
|
|
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 = 1
|
|
res = self._edit_volume_post(vol_bus="sata")
|
|
self.assertEqual(res.status_code, 302)
|
|
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.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)
|
|
for value in ("None", ""):
|
|
with self.subTest(vol_format=value), patch("instances.models.wvmInstance") as mock_wvm:
|
|
proxy = mock_wvm.return_value
|
|
self._mock_disk_options(proxy)
|
|
self._edit_volume_post(vol_format=value)
|
|
proxy.edit_disk.assert_called_once()
|
|
self.assertEqual(proxy.edit_disk.call_args[0][6], VM_DISK["format"])
|
|
|
|
def test_disk_names_with_trailing_newline_are_rejected(self):
|
|
self.client.force_login(self.superuser)
|
|
with patch("instances.models.wvmInstance") as mock_wvm, patch(
|
|
"instances.views.wvmCreate"
|
|
) as mock_create, patch("instances.views.wvmStorage"):
|
|
self._mock_disk_options(mock_wvm.return_value)
|
|
self._post(
|
|
"add_new_vol",
|
|
{"storage": "default", "name": "disk\n", "format": "qcow2", "size": "1",
|
|
"bus": "virtio", "cache": "default"},
|
|
)
|
|
mock_create.return_value.create_volume.assert_not_called()
|
|
|
|
def test_edit_volume_accepts_a_disk_without_any_driver_format(self):
|
|
self.client.force_login(self.superuser)
|
|
no_format_disk = dict(VM_DISK, format=None)
|
|
with patch("instances.models.wvmInstance") as mock_wvm:
|
|
proxy = mock_wvm.return_value
|
|
self._mock_disk_options(proxy)
|
|
proxy.get_disk_devices.return_value = [no_format_disk]
|
|
self._edit_volume_post(vol_format="None")
|
|
proxy.edit_disk.assert_called_once()
|
|
self.assertEqual(proxy.edit_disk.call_args[0][6], "")
|