""" 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'/>/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", )