From 5758073e825420ec8176f9f36bcc40a59e6cc744 Mon Sep 17 00:00:00 2001 From: catborise Date: Fri, 2 Oct 2026 19:07:53 +0300 Subject: [PATCH] fix(instances): require change permission to clone and stop trusting clone POST data clone only required "view" access, so a user with clone_instances and the global view_instances permission could clone anyone's VM and read its disks. It also copied every POST field into clone_data, which let the client override the server-set disk_owner_uid/gid, and disk- values went unvalidated into volume XML and os.path.join. - Require change permission on the source VM, except for templates, which any user who can view them may clone. - Build clone_data from an allowlist: name, title and description for everyone; disk names and MAC addresses only from superusers. - For non-superusers, derive disk names (get_clone_disk_name) and MAC addresses (DHCP or random) on the server; their form only sent the source VM's own values, which could never work. - Validate every MAC address (not only the first) and every disk name; a missing MAC is an error message instead of a KeyError. --- instances/test_clone_permission.py | 169 +++++++++++++++++++++++++++++ instances/views.py | 59 ++++++++-- 2 files changed, 218 insertions(+), 10 deletions(-) create mode 100644 instances/test_clone_permission.py diff --git a/instances/test_clone_permission.py b/instances/test_clone_permission.py new file mode 100644 index 00000000..6cd04289 --- /dev/null +++ b/instances/test_clone_permission.py @@ -0,0 +1,169 @@ +""" +Regression tests for ROADMAP Wave 0 item 4 (S-05): cloning another user's VM +must require change permission on it (templates excepted), and the clone +data must not be taken blindly from the POST body. +""" +from contextlib import contextmanager +from unittest.mock import patch + +from django.contrib.auth import get_user_model +from django.contrib.auth.models import Permission +from django.test import TestCase +from django.urls import reverse + +from accounts.models import UserInstance +from appsettings.settings import app_settings +from computes.models import Compute +from instances.models import Instance + +SRC_DISK = { + "dev": "vda", + "image": "src-vm.qcow2", + "storage": "default", + "path": "/var/lib/libvirt/images/src-vm.qcow2", + "size": 10 << 30, + "format": "qcow2", +} + + +class ClonePermissionTestCase(TestCase): + def setUp(self): + User = get_user_model() + clone_perm = Permission.objects.get(codename="clone_instances") + view_perm = Permission.objects.get(codename="view_instances") + + self.superuser = User.objects.create_superuser( + username="clone_super", email="clone_super@example.com", password="password" + ) + self.viewer = User.objects.create_user(username="clone_viewer", password="password") + self.viewer.user_permissions.add(clone_perm, view_perm) + self.owner_readonly = User.objects.create_user( + username="clone_owner_ro", password="password" + ) + self.owner_readonly.user_permissions.add(clone_perm) + self.owner_change = User.objects.create_user( + username="clone_owner_rw", password="password" + ) + self.owner_change.user_permissions.add(clone_perm) + + self.compute = Compute.objects.create( + name="clone-compute", hostname="127.0.0.1", login="root", password="", type=1 + ) + self.instance = Instance.objects.create( + compute=self.compute, + name="src-vm", + uuid="12121212-3434-5656-7878-909090909090", + ) + UserInstance.objects.create( + instance=self.instance, user=self.owner_readonly, is_change=False + ) + UserInstance.objects.create( + instance=self.instance, user=self.owner_change, is_change=True + ) + + @contextmanager + def mocked_libvirt(self): + with patch("instances.models.wvmInstance") as mock_wvm, patch( + "instances.views.utils.check_user_quota", return_value="" + ): + proxy = mock_wvm.return_value + proxy.get_disk_devices.return_value = [SRC_DISK] + proxy.get_net_devices.return_value = [{"mac": "52:54:00:aa:bb:cc"}] + proxy.get_vcpu.return_value = 1 + proxy.get_memory.return_value = 512 + proxy.clone_instance.return_value = "abababab-cdcd-efef-0101-232323232323" + yield proxy + + def _clone(self, user, data): + self.client.force_login(user) + return self.client.post( + reverse("instances:clone", args=[self.instance.id]), + data, + HTTP_REFERER=f"/instances/{self.instance.id}/", + ) + + def _valid_post(self, **extra): + data = { + "name": "clone-vm", + "clone-net-mac-0": "52:54:00:12:34:56", + "disk-vda": "clone-vm.qcow2", + "clone-title": "", + "clone-description": "", + } + data.update(extra) + return data + + def test_viewer_cannot_clone_someone_elses_vm(self): + with self.mocked_libvirt() as proxy: + res = self._clone(self.viewer, self._valid_post()) + self.assertEqual(res.status_code, 403) + proxy.clone_instance.assert_not_called() + + def test_owner_without_is_change_cannot_clone(self): + with self.mocked_libvirt() as proxy: + res = self._clone(self.owner_readonly, self._valid_post()) + self.assertEqual(res.status_code, 403) + proxy.clone_instance.assert_not_called() + + def test_viewer_can_clone_a_template(self): + self.instance.is_template = True + self.instance.save(update_fields=["is_template"]) + with self.mocked_libvirt() as proxy: + self._clone(self.viewer, self._valid_post()) + proxy.clone_instance.assert_called_once() + + def test_posted_disk_owner_and_unknown_keys_are_ignored(self): + with self.mocked_libvirt() as proxy: + self._clone( + self.owner_change, + self._valid_post(disk_owner_uid="4242", disk_owner_gid="4242", evil="x"), + ) + proxy.clone_instance.assert_called_once() + clone_data = proxy.clone_instance.call_args[0][0] + self.assertEqual( + clone_data["disk_owner_uid"], + int(app_settings.INSTANCE_VOLUME_DEFAULT_OWNER_UID), + ) + self.assertEqual( + clone_data["disk_owner_gid"], + int(app_settings.INSTANCE_VOLUME_DEFAULT_OWNER_GID), + ) + self.assertNotIn("evil", clone_data) + + def test_non_superuser_disk_names_are_derived_by_the_server(self): + with self.mocked_libvirt() as proxy: + self._clone( + self.owner_change, + self._valid_post(**{"disk-vda": "../../../etc/passwd"}), + ) + proxy.clone_instance.assert_called_once() + clone_data = proxy.clone_instance.call_args[0][0] + self.assertEqual(clone_data["disk-vda"], "clone-vm.qcow2") + + def test_superuser_unsafe_disk_name_is_rejected(self): + with self.mocked_libvirt() as proxy: + self._clone( + self.superuser, + self._valid_post(**{"disk-vda": "x"}), + ) + proxy.clone_instance.assert_not_called() + + def test_superuser_invalid_second_mac_is_rejected(self): + with self.mocked_libvirt() as proxy: + self._clone( + self.superuser, + self._valid_post(**{"clone-net-mac-1": "not-a-mac"}), + ) + proxy.clone_instance.assert_not_called() + + def test_superuser_valid_clone_passes_posted_values(self): + with self.mocked_libvirt() as proxy: + self._clone( + self.superuser, + self._valid_post(**{"disk-vda": "custom-name.qcow2", "meta-vda": "true"}), + ) + proxy.clone_instance.assert_called_once() + clone_data = proxy.clone_instance.call_args[0][0] + self.assertEqual(clone_data["disk-vda"], "custom-name.qcow2") + self.assertEqual(clone_data["meta-vda"], "true") + self.assertEqual(clone_data["clone-net-mac-0"], "52:54:00:12:34:56") diff --git a/instances/views.py b/instances/views.py index 1c45f9d3..19b3bf55 100755 --- a/instances/views.py +++ b/instances/views.py @@ -1541,25 +1541,39 @@ def del_owner(request, pk): return redirect(request.META.get("HTTP_REFERER") + "#users") +MAC_ADDRESS_RE = re.compile(r"^([0-9A-F]{2})(:?[0-9A-F]{2}){5}$", re.IGNORECASE) +CLONE_DISK_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.+-]*$") +CLONE_POST_KEY_RE = re.compile(r"^(clone-net-mac-\d+|disk-[a-z0-9]+|meta-[a-z0-9]+)$") + + @require_POST @permission_required("instances.clone_instances", raise_exception=True) def clone(request, pk): + # Cloning copies the source disks, so it needs change permission on the + # source VM. Templates are meant to be deployed from, so viewing is enough. instance = get_instance(request.user, pk) + if not instance.is_template: + instance = get_instance(request.user, pk, perm_type="change") clone_data = dict() - clone_data["name"] = request.POST.get("name", "") + clone_data["name"] = request.POST.get("name", "").strip() + clone_data["clone-title"] = request.POST.get("clone-title", "").strip() + clone_data["clone-description"] = request.POST.get("clone-description", "").strip() disk_sum = sum([disk["size"] >> 30 for disk in instance.disks]) quota_msg = utils.check_user_quota( request.user, 1, instance.vcpu, instance.memory, disk_sum ) - check_instance = Instance.objects.filter(name=clone_data["name"]) clone_data["disk_owner_uid"] = int(app_settings.INSTANCE_VOLUME_DEFAULT_OWNER_UID) clone_data["disk_owner_gid"] = int(app_settings.INSTANCE_VOLUME_DEFAULT_OWNER_GID) - for post in request.POST: - clone_data[post] = request.POST.get(post, "").strip() + # Only superusers may choose disk names and MAC addresses; the form only + # shows those fields to them. + if request.user.is_superuser: + for key, value in request.POST.items(): + if CLONE_POST_KEY_RE.match(key): + clone_data[key] = value.strip() if app_settings.CLONE_INSTANCE_AUTO_NAME == "True" and not clone_data["name"]: auto_vname = utils.get_clone_free_names()[0] @@ -1570,6 +1584,30 @@ def clone(request, pk): disk_name = utils.get_clone_disk_name(disk, instance.name, auto_vname) clone_data[disk_dev] = disk_name + if not request.user.is_superuser: + for disk in instance.disks: + clone_data[f"disk-{disk['dev']}"] = utils.get_clone_disk_name( + disk, instance.name, clone_data["name"] + ) + for num in range(max(1, len(instance.networks))): + key = f"clone-net-mac-{num}" + if not clone_data.get(key): + clone_data[key] = ( + utils.get_dhcp_mac_address(clone_data["name"]) if num == 0 else "" + ) or utils.get_random_mac_address() + + check_instance = Instance.objects.filter(name=clone_data["name"]) + invalid_macs = [ + value + for key, value in clone_data.items() + if key.startswith("clone-net-mac-") and not MAC_ADDRESS_RE.match(value) + ] + invalid_disks = [ + value + for key, value in clone_data.items() + if key.startswith("disk-") and not CLONE_DISK_NAME_RE.match(value or "") + ] + if not request.user.is_superuser and quota_msg: msg = _( "User '%(quota_msg)s' quota reached, cannot create '%(clone_name)s'!" @@ -1588,13 +1626,14 @@ def clone(request, pk): "clone_name": clone_data["name"] } messages.error(request, msg) - elif not re.match( - r"^([0-9A-F]{2})(:?[0-9A-F]{2}){5}$", - clone_data["clone-net-mac-0"], - re.IGNORECASE, - ): + elif "clone-net-mac-0" not in clone_data or invalid_macs: msg = _("Instance MAC '%(clone_mac)s' invalid format!") % { - "clone_mac": clone_data["clone-net-mac-0"] + "clone_mac": ", ".join(invalid_macs) + } + messages.error(request, msg) + elif invalid_disks: + msg = _("Disk name '%(disk_name)s' contains invalid characters!") % { + "disk_name": ", ".join(str(v) for v in invalid_disks) } messages.error(request, msg) else: