mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 19:26:48 +00:00
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-<dev> 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.
This commit is contained in:
catborise
parent
b1ab1d47fa
commit
5758073e82
2 changed files with 218 additions and 10 deletions
169
instances/test_clone_permission.py
Normal file
169
instances/test_clone_permission.py
Normal file
|
|
@ -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</name><evil/>"}),
|
||||
)
|
||||
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")
|
||||
|
|
@ -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:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue