diff --git a/instances/test_disk_isolation.py b/instances/test_disk_isolation.py new file mode 100644 index 00000000..85e8b553 --- /dev/null +++ b/instances/test_disk_isolation.py @@ -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'/>/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", + ) diff --git a/instances/test_security_mutations.py b/instances/test_security_mutations.py index 0b047d2a..e8af20ae 100644 --- a/instances/test_security_mutations.py +++ b/instances/test_security_mutations.py @@ -297,17 +297,18 @@ class InstanceSecurityMutationsTestCase(TestCase): self.assertEqual(res.status_code, 302) self.assertIn(f"/instances/{self.instance.id}/#resize", res.url) - def test_subresource_mutation_allowed_with_is_change(self): - # change_user has is_change=True, is_delete=False + def test_disk_mutation_denied_with_is_change(self): + # change_user has is_change=True, is_delete=False. Disk views are + # superuser-only (S-02/S-03/S-04), so is_change alone is not enough. self.client.force_login(self.change_user) - with patch("instances.models.wvmInstance"), patch("instances.views.wvmStorage"): - # delete_vol uses perm_type="change", so it passes get_instance without 403 + 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, 302) + 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 diff --git a/instances/views.py b/instances/views.py index 1c45f9d3..d0ee249d 100755 --- a/instances/views.py +++ b/instances/views.py @@ -392,6 +392,54 @@ def serialize_instance_mutation(func): return wrapper +DISK_FORMAT_RE = re.compile(r"^[a-z0-9]+$") +DISK_SERIAL_RE = re.compile(r"^[A-Za-z0-9_.+-]*$") +VOLUME_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.+-]*$") + + +def invalid_disk_options( + instance, + bus=None, + cache=None, + io=None, + discard=None, + zeroes=None, + format=None, + serial=None, +): + """ + Return the names of the given disk options whose values are not allowed. + These values end up in libvirt disk XML, so they must come from the + hypervisor's own lists (or match a strict pattern). None means "not given". + """ + proxy = instance.proxy + checks = { + "bus": ( + bus, + lambda v: v in proxy.get_disk_bus_types(instance.arch, instance.machine), + ), + "cache": (cache, lambda v: v in proxy.get_cache_modes()), + "io": (io, lambda v: v in proxy.get_io_modes()), + "discard": (discard, lambda v: v in proxy.get_discard_modes()), + "detect_zeroes": (zeroes, lambda v: v in proxy.get_detect_zeroes_modes()), + "format": (format, lambda v: bool(DISK_FORMAT_RE.match(v))), + "serial": (serial, lambda v: bool(DISK_SERIAL_RE.match(v))), + } + return [ + name + for name, (value, is_valid) in checks.items() + if value is not None and not is_valid(value) + ] + + +def reject_disk_options(request, invalid): + messages.error( + request, + _("Invalid disk options: %(options)s") % {"options": ", ".join(invalid)}, + ) + return redirect(request.META.get("HTTP_REFERER") + "#disks") + + @require_POST @serialize_instance_mutation def poweron(request, pk): @@ -761,135 +809,142 @@ def resize_disk(request, pk): @require_POST +@superuser_only @serialize_instance_mutation def add_new_vol(request, pk): instance = get_instance(request.user, pk, perm_type="change") - allow_admin_or_not_template = ( - request.user.is_superuser or request.user.is_staff or not instance.is_template + media = instance.proxy.get_media_devices() + disks = instance.proxy.get_disk_devices() + storage = request.POST.get("storage", "") + name = request.POST.get("name", "") + format = request.POST.get("format", app_settings.INSTANCE_VOLUME_DEFAULT_FORMAT) + size = request.POST.get("size", 0) + meta_prealloc = True if request.POST.get("meta_prealloc", False) else False + bus = request.POST.get("bus", app_settings.INSTANCE_VOLUME_DEFAULT_BUS) + cache = request.POST.get("cache", app_settings.INSTANCE_VOLUME_DEFAULT_CACHE) + + invalid = invalid_disk_options(instance, bus=bus, cache=cache, format=format) + if not VOLUME_NAME_RE.match(name): + invalid.append("name") + if invalid: + return reject_disk_options(request, invalid) + + conn_create = wvmCreate( + instance.compute.hostname, + instance.compute.login, + instance.compute.password, + instance.compute.type, + ) + target_dev = utils.get_new_disk_dev(media, disks, bus) + + source = conn_create.create_volume( + storage, + name, + size, + format, + meta_prealloc, + int(app_settings.INSTANCE_VOLUME_DEFAULT_OWNER_UID), + int(app_settings.INSTANCE_VOLUME_DEFAULT_OWNER_GID), ) - if allow_admin_or_not_template: - media = instance.proxy.get_media_devices() - disks = instance.proxy.get_disk_devices() - conn_create = wvmCreate( - instance.compute.hostname, - instance.compute.login, - instance.compute.password, - instance.compute.type, - ) - storage = request.POST.get("storage", "") - name = request.POST.get("name", "") - format = request.POST.get("format", app_settings.INSTANCE_VOLUME_DEFAULT_FORMAT) - size = request.POST.get("size", 0) - meta_prealloc = True if request.POST.get("meta_prealloc", False) else False - bus = request.POST.get("bus", app_settings.INSTANCE_VOLUME_DEFAULT_BUS) - cache = request.POST.get("cache", app_settings.INSTANCE_VOLUME_DEFAULT_CACHE) - target_dev = utils.get_new_disk_dev(media, disks, bus) + conn_pool = wvmStorage( + instance.compute.hostname, + instance.compute.login, + instance.compute.password, + instance.compute.type, + storage, + ) - source = conn_create.create_volume( - storage, - name, - size, - format, - meta_prealloc, - int(app_settings.INSTANCE_VOLUME_DEFAULT_OWNER_UID), - int(app_settings.INSTANCE_VOLUME_DEFAULT_OWNER_GID), - ) + pool_type = conn_pool.get_type() + disk_type = conn_pool.get_volume_type(os.path.basename(source)) - conn_pool = wvmStorage( - instance.compute.hostname, - instance.compute.login, - instance.compute.password, - instance.compute.type, - storage, - ) + if pool_type == "rbd": + source_info = conn_pool.get_rbd_source() + else: # add more disk types to handle different pool and disk types + source_info = None - pool_type = conn_pool.get_type() - disk_type = conn_pool.get_volume_type(os.path.basename(source)) - - if pool_type == "rbd": - source_info = conn_pool.get_rbd_source() - else: # add more disk types to handle different pool and disk types - source_info = None - - instance.proxy.attach_disk( - target_dev, - source, - source_info=source_info, - pool_type=pool_type, - disk_type=disk_type, - target_bus=bus, - format_type=format, - cache_mode=cache, - ) - msg = _("Attach new disk: %(name)s (%(format)s)") % { - "name": name, - "format": format, - } - addlogmsg(request.user.username, instance.compute.name, instance.name, msg) + instance.proxy.attach_disk( + target_dev, + source, + source_info=source_info, + pool_type=pool_type, + disk_type=disk_type, + target_bus=bus, + format_type=format, + cache_mode=cache, + ) + msg = _("Attach new disk: %(name)s (%(format)s)") % { + "name": name, + "format": format, + } + addlogmsg(request.user.username, instance.compute.name, instance.name, msg) return redirect(request.META.get("HTTP_REFERER") + "#disks") @require_POST +@superuser_only @serialize_instance_mutation def add_existing_vol(request, pk): instance = get_instance(request.user, pk, perm_type="change") - allow_admin_or_not_template = ( - request.user.is_superuser or request.user.is_staff or not instance.is_template + storage = request.POST.get("selected_storage", "") + name = request.POST.get("vols", "") + bus = request.POST.get("bus", app_settings.INSTANCE_VOLUME_DEFAULT_BUS) + cache = request.POST.get("cache", app_settings.INSTANCE_VOLUME_DEFAULT_CACHE) + + invalid = invalid_disk_options(instance, bus=bus, cache=cache) + if invalid: + return reject_disk_options(request, invalid) + + media = instance.proxy.get_media_devices() + disks = instance.proxy.get_disk_devices() + + conn_create = wvmStorage( + instance.compute.hostname, + instance.compute.login, + instance.compute.password, + instance.compute.type, + storage, ) - if allow_admin_or_not_template: - storage = request.POST.get("selected_storage", "") - name = request.POST.get("vols", "") - bus = request.POST.get("bus", app_settings.INSTANCE_VOLUME_DEFAULT_BUS) - cache = request.POST.get("cache", app_settings.INSTANCE_VOLUME_DEFAULT_CACHE) - media = instance.proxy.get_media_devices() - disks = instance.proxy.get_disk_devices() + # Only volumes that really exist in the selected pool; this also rules + # out path traversal through the volume name. + if name not in conn_create.get_volumes(): + return reject_disk_options(request, ["vols"]) - conn_create = wvmStorage( - instance.compute.hostname, - instance.compute.login, - instance.compute.password, - instance.compute.type, - storage, - ) + format_type = conn_create.get_volume_format_type(name) + disk_type = conn_create.get_volume_type(name) + pool_type = conn_create.get_type() + if pool_type == "rbd": + source_info = conn_create.get_rbd_source() + path = conn_create.get_source_name() + else: + source_info = None + path = conn_create.get_target_path() - format_type = conn_create.get_volume_format_type(name) - disk_type = conn_create.get_volume_type(name) - pool_type = conn_create.get_type() - if pool_type == "rbd": - source_info = conn_create.get_rbd_source() - path = conn_create.get_source_name() - else: - source_info = None - path = conn_create.get_target_path() + target_dev = utils.get_new_disk_dev(media, disks, bus) + source = f"{path}/{name}" - target_dev = utils.get_new_disk_dev(media, disks, bus) - source = f"{path}/{name}" - - instance.proxy.attach_disk( - target_dev, - source, - source_info=source_info, - pool_type=pool_type, - disk_type=disk_type, - target_bus=bus, - format_type=format_type, - cache_mode=cache, - ) - msg = _("Attach Existing disk: %(target_dev)s") % {"target_dev": target_dev} - addlogmsg(request.user.username, instance.compute.name, instance.name, msg) + instance.proxy.attach_disk( + target_dev, + source, + source_info=source_info, + pool_type=pool_type, + disk_type=disk_type, + target_bus=bus, + format_type=format_type, + cache_mode=cache, + ) + msg = _("Attach Existing disk: %(target_dev)s") % {"target_dev": target_dev} + addlogmsg(request.user.username, instance.compute.name, instance.name, msg) return redirect(request.META.get("HTTP_REFERER") + "#disks") @require_POST +@superuser_only @serialize_instance_mutation def edit_volume(request, pk): instance = get_instance(request.user, pk, perm_type="change") - allow_admin_or_not_template = ( - request.user.is_superuser or request.user.is_staff or not instance.is_template - ) - if "edit_volume" in request.POST and allow_admin_or_not_template: + if "edit_volume" in request.POST: target_dev = request.POST.get("dev", "") new_path = request.POST.get("vol_path", "") @@ -910,6 +965,22 @@ def edit_volume(request, pk): zeroes = request.POST.get( "vol_detect_zeroes", app_settings.INSTANCE_VOLUME_DEFAULT_DETECT_ZEROES ) + + invalid = invalid_disk_options( + instance, + bus=new_bus, + cache=cache, + io=io, + discard=discard, + zeroes=zeroes, + format=format, + serial=serial, + ) + if target_dev not in [disk["dev"] for disk in instance.disks]: + invalid.append("dev") + if invalid: + return reject_disk_options(request, invalid) + new_target_dev = utils.get_new_disk_dev(instance.media, instance.disks, new_bus) if new_bus != bus: @@ -959,122 +1030,113 @@ def edit_volume(request, pk): @require_POST +@superuser_only @serialize_instance_mutation def delete_vol(request, pk): instance = get_instance(request.user, pk, perm_type="change") - allow_admin_or_not_template = ( - request.user.is_superuser or request.user.is_staff or not instance.is_template - ) - if allow_admin_or_not_template: - storage = request.POST.get("storage", "") - conn_delete = wvmStorage( - instance.compute.hostname, - instance.compute.login, - instance.compute.password, - instance.compute.type, - storage, + dev = request.POST.get("dev", "") + + # Resolve the volume from the VM's own XML; never trust the posted + # storage/name, which could point at any volume on the compute. + disk = next((d for d in instance.disks if d["dev"] == dev), None) + if disk is None or not disk["storage"] or not disk["image"]: + messages.error( + request, + _("Disk %(dev)s is not a storage volume of this instance") % {"dev": dev}, ) - dev = request.POST.get("dev", "") - path = request.POST.get("path", "") - name = request.POST.get("name", "") + return redirect(request.META.get("HTTP_REFERER") + "#disks") - msg = _("Delete disk: %(dev)s") % {"dev": dev} - instance.proxy.detach_disk(dev) - conn_delete.del_volume(name) + conn_delete = wvmStorage( + instance.compute.hostname, + instance.compute.login, + instance.compute.password, + instance.compute.type, + disk["storage"], + ) - addlogmsg(request.user.username, instance.compute.name, instance.name, msg) + msg = _("Delete disk: %(dev)s") % {"dev": dev} + instance.proxy.detach_disk(dev) + conn_delete.del_volume(disk["image"]) + + addlogmsg(request.user.username, instance.compute.name, instance.name, msg) return redirect(request.META.get("HTTP_REFERER") + "#disks") @require_POST +@superuser_only @serialize_instance_mutation def detach_vol(request, pk): instance = get_instance(request.user, pk, perm_type="change") - allow_admin_or_not_template = ( - request.user.is_superuser or request.user.is_staff or not instance.is_template - ) - - if allow_admin_or_not_template: - dev = request.POST.get("dev", "") - path = request.POST.get("path", "") - instance.proxy.detach_disk(dev) - msg = _("Detach disk: %(dev)s") % {"dev": dev} - addlogmsg(request.user.username, instance.compute.name, instance.name, msg) + dev = request.POST.get("dev", "") + instance.proxy.detach_disk(dev) + msg = _("Detach disk: %(dev)s") % {"dev": dev} + addlogmsg(request.user.username, instance.compute.name, instance.name, msg) return redirect(request.META.get("HTTP_REFERER") + "#disks") @require_POST +@superuser_only @serialize_instance_mutation def add_cdrom(request, pk): instance = get_instance(request.user, pk, perm_type="change") - allow_admin_or_not_template = ( - request.user.is_superuser or request.user.is_staff or not instance.is_template + bus = request.POST.get("bus", "ide" if instance.machine == "pc" else "sata") + invalid = invalid_disk_options(instance, bus=bus) + if invalid: + return reject_disk_options(request, invalid) + + target = utils.get_new_disk_dev(instance.media, instance.disks, bus) + instance.proxy.attach_disk( + target, + "", + disk_device="cdrom", + cache_mode="none", + target_bus=bus, + readonly=True, ) - if allow_admin_or_not_template: - bus = request.POST.get("bus", "ide" if instance.machine == "pc" else "sata") - target = utils.get_new_disk_dev(instance.media, instance.disks, bus) - instance.proxy.attach_disk( - target, - "", - disk_device="cdrom", - cache_mode="none", - target_bus=bus, - readonly=True, - ) - msg = _("Add CD-ROM: %(target)s") % {"target": target} - addlogmsg(request.user.username, instance.compute.name, instance.name, msg) + msg = _("Add CD-ROM: %(target)s") % {"target": target} + addlogmsg(request.user.username, instance.compute.name, instance.name, msg) return redirect(request.META.get("HTTP_REFERER") + "#disks") @require_POST +@superuser_only @serialize_instance_mutation def detach_cdrom(request, pk, dev): instance = get_instance(request.user, pk, perm_type="change") - allow_admin_or_not_template = ( - request.user.is_superuser or request.user.is_staff or not instance.is_template - ) - - if allow_admin_or_not_template: - # dev = request.POST.get('detach_cdrom', '') - instance.proxy.detach_disk(dev) - msg = _("Detach CD-ROM: %(dev)s") % {"dev": dev} - addlogmsg(request.user.username, instance.compute.name, instance.name, msg) + # dev = request.POST.get('detach_cdrom', '') + instance.proxy.detach_disk(dev) + msg = _("Detach CD-ROM: %(dev)s") % {"dev": dev} + addlogmsg(request.user.username, instance.compute.name, instance.name, msg) return redirect(request.META.get("HTTP_REFERER") + "#disks") @require_POST +@superuser_only @serialize_instance_mutation def unmount_iso(request, pk): instance = get_instance(request.user, pk, perm_type="change") - allow_admin_or_not_template = ( - request.user.is_superuser or request.user.is_staff or not instance.is_template - ) - if allow_admin_or_not_template: - image = request.POST.get("path", "") - dev = request.POST.get("umount_iso", "") - instance.proxy.umount_iso(dev, image) - msg = _("Mount media: %(dev)s") % {"dev": dev} - addlogmsg(request.user.username, instance.compute.name, instance.name, msg) + image = request.POST.get("path", "") + dev = request.POST.get("umount_iso", "") + instance.proxy.umount_iso(dev, image) + msg = _("Mount media: %(dev)s") % {"dev": dev} + addlogmsg(request.user.username, instance.compute.name, instance.name, msg) return redirect(request.META.get("HTTP_REFERER") + "#disks") @require_POST +@superuser_only @serialize_instance_mutation def mount_iso(request, pk): instance = get_instance(request.user, pk, perm_type="change") - allow_admin_or_not_template = ( - request.user.is_superuser or request.user.is_staff or not instance.is_template - ) - if allow_admin_or_not_template: - image = request.POST.get("media", "") - dev = request.POST.get("mount_iso", "") - instance.proxy.mount_iso(dev, image) - msg = _("Unmount media: %(dev)s") % {"dev": dev} - addlogmsg(request.user.username, instance.compute.name, instance.name, msg) + image = request.POST.get("media", "") + dev = request.POST.get("mount_iso", "") + instance.proxy.mount_iso(dev, image) + msg = _("Unmount media: %(dev)s") % {"dev": dev} + addlogmsg(request.user.username, instance.compute.name, instance.name, msg) return redirect(request.META.get("HTTP_REFERER") + "#disks") diff --git a/vrtManager/instance.py b/vrtManager/instance.py index a4be0d3d..ea0a2320 100644 --- a/vrtManager/instance.py +++ b/vrtManager/instance.py @@ -760,6 +760,15 @@ class wvmInstance(wvmConnect): discard_mode=None, detect_zeroes_mode=None, ): + # Every value below is interpolated into disk XML; escape them all so + # a quote in a path, serial or option cannot inject XML. + esc = util.xml_escape + target_dev, source, target_bus = esc(target_dev), esc(source), esc(target_bus) + disk_type, disk_device = esc(disk_type), esc(disk_device) + driver_name, format_type = esc(driver_name), esc(format_type) + serial = esc(serial) + cache_mode, io_mode = esc(cache_mode), esc(io_mode) + discard_mode, detect_zeroes_mode = esc(discard_mode), esc(detect_zeroes_mode) additionals = "" if ( @@ -822,7 +831,7 @@ class wvmInstance(wvmConnect): def detach_disk(self, target_dev): tree = etree.fromstring(self._XMLDesc(0)) - disk_el = tree.xpath("./devices/disk/target[@dev='{}']".format(target_dev))[ + disk_el = tree.xpath("./devices/disk/target[@dev=$dev]", dev=target_dev)[ 0 ].getparent() xml_disk = etree.tostring(disk_el).decode() @@ -850,7 +859,7 @@ class wvmInstance(wvmConnect): detect_zeroes_mode, ): tree = etree.fromstring(self._XMLDesc(0)) - disk_el = tree.xpath("./devices/disk/target[@dev='{}']".format(target_dev))[ + disk_el = tree.xpath("./devices/disk/target[@dev=$dev]", dev=target_dev)[ 0 ].getparent() old_disk_type = disk_el.get("type") @@ -858,6 +867,15 @@ class wvmInstance(wvmConnect): old_driver_name = disk_el.xpath("driver/@name")[0] old_target_bus = disk_el.xpath("target/@bus")[0] + # Values below are interpolated into disk XML; escape them all. + esc = util.xml_escape + target_dev, source, target_bus = esc(target_dev), esc(source), esc(target_bus) + serial, format = esc(serial), esc(format) + cache_mode, io_mode = esc(cache_mode), esc(io_mode) + discard_mode, detect_zeroes_mode = esc(discard_mode), esc(detect_zeroes_mode) + old_disk_type, old_disk_device = esc(old_disk_type), esc(old_disk_device) + old_driver_name = esc(old_driver_name) + additionals = "" if cache_mode is not None and cache_mode != "default": additionals += f"cache='{cache_mode}' " diff --git a/vrtManager/test_xml_safety.py b/vrtManager/test_xml_safety.py index 70bef485..78ac7f83 100644 --- a/vrtManager/test_xml_safety.py +++ b/vrtManager/test_xml_safety.py @@ -178,3 +178,67 @@ class TestXmlSafety(unittest.TestCase): if __name__ == "__main__": unittest.main() + + +class TestDiskXmlEscaping(unittest.TestCase): + def _instance(self): + inst = wvmInstance.__new__(wvmInstance) + inst.get_status = MagicMock(return_value=5) # shutoff + inst.instance = MagicMock() + return inst + + def test_attach_disk_escapes_source_and_serial(self): + inst = self._instance() + inst.attach_disk( + "vdb", + "/images/a'/>S1</serial><evil/>", xml) + self.assertNotIn("", xml) + + def test_edit_disk_escapes_source_and_serial(self): + inst = self._instance() + inst._XMLDesc = MagicMock( + return_value=( + "" + "" + "" + "" + "" + ) + ) + inst.edit_disk( + "vda", + "/images/b'/>", xml) + + def test_detach_disk_does_not_evaluate_dev_as_xpath(self): + inst = self._instance() + inst._XMLDesc = MagicMock( + return_value=( + "" + "" + "" + "" + ) + ) + with self.assertRaises(IndexError): + inst.detach_disk("x' or '1'='1") + inst.instance.detachDeviceFlags.assert_not_called()