mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-09 19:56:48 +00:00
fix(instances): restrict disk views to superusers and validate disk input
Disk and media mutation views (add_new_vol, add_existing_vol, edit_volume, delete_vol, detach_vol, add_cdrom, detach_cdrom, mount_iso, unmount_iso) were reachable by any owner with is_change, although the UI only shows the Disk tab to superusers. An ordinary user could delete any volume on the compute or attach arbitrary host files to their VM. - Require superuser for all disk/media views, matching the UI. - delete_vol resolves the volume from the VM's own disk XML by dev and ignores the posted storage/name. - Validate bus/cache/io/discard/detect_zeroes against the hypervisor's lists, and format/serial/volume name against strict patterns. - add_existing_vol only accepts volumes that exist in the selected pool. - edit_volume only accepts a dev attached to the VM. - vrtManager: escape all values interpolated into disk XML and pass dev as an XPath variable in detach_disk/edit_disk.
This commit is contained in:
catborise
parent
b1ab1d47fa
commit
8a3ee13b41
5 changed files with 573 additions and 179 deletions
249
instances/test_disk_isolation.py
Normal file
249
instances/test_disk_isolation.py
Normal file
|
|
@ -0,0 +1,249 @@
|
|||
"""
|
||||
Regression tests for ROADMAP Wave 0 item 1 (S-02, S-03, S-04):
|
||||
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 MagicMock, 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",
|
||||
)
|
||||
Loading…
Add table
Add a link
Reference in a new issue