j3d1/webvirtcloud
1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-08 11:16:47 +00:00
Code Issues Projects Releases Wiki Activity

fix(instances): edit_volume bus change and format, full-string validation

Browse source 
- edit_volume passed driver_type to attach_disk, which has no such
  parameter, so every bus change failed after the disk was detached; pass
  format_type.
- A disk without a driver type shows "None" in the edit form; keep the
  disk's current format instead of rejecting it.
- Validate with fullmatch, so a trailing newline is no longer accepted,
  and use one volume-name pattern for disk views and clone.
This commit is contained in:
catborise 2026-10-02 22:31:25 +03:00
parent fe5bb578d8
commit d250915e17
2 changed files with 62 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

51
instances/test_disk_isolation.py
View file

@ -247,3 +247,54 @@ class DiskViewsTenantIsolationTestCase(TestCase):
"vda", VM_DISK["path"], False, False, "virtio", "SER-01", "qcow2",
"none", "native", "unmap", "on",
)
def _edit_volume_post(self, **extra):
data = {
"edit_volume": "1",
"dev": "vda",
"vol_path": VM_DISK["path"],
"vol_bus_old": "virtio",
"vol_bus": "virtio",
"vol_format": "qcow2",
"vol_cache": "none",
}
data.update(extra)
return self._post("edit_volume", data)
def test_edit_volume_bus_change_passes_valid_arguments_to_attach_disk(self):
# autospec enforces the real attach_disk signature; a plain MagicMock
# would accept the invalid driver_type argument.
from vrtManager.instance import wvmInstance
self.client.force_login(self.superuser)
with patch("instances.models.wvmInstance", autospec=wvmInstance) as mock_wvm:
proxy = mock_wvm.return_value
self._mock_disk_options(proxy)
proxy.get_status.return_value = 5
res = self._edit_volume_post(vol_bus="sata")
self.assertEqual(res.status_code, 302)
proxy.attach_disk.assert_called_once()
self.assertEqual(proxy.attach_disk.call_args.kwargs["format_type"], "qcow2")
def test_edit_volume_keeps_current_format_when_field_is_none_or_empty(self):
self.client.force_login(self.superuser)
for value in ("None", ""):
with self.subTest(vol_format=value), patch("instances.models.wvmInstance") as mock_wvm:
proxy = mock_wvm.return_value
self._mock_disk_options(proxy)
self._edit_volume_post(vol_format=value)
proxy.edit_disk.assert_called_once()
self.assertEqual(proxy.edit_disk.call_args[0][6], VM_DISK["format"])
def test_disk_names_with_trailing_newline_are_rejected(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)
self._post(
"add_new_vol",
{"storage": "default", "name": "disk\n", "format": "qcow2", "size": "1",
"bus": "virtio", "cache": "default"},
)
mock_create.return_value.create_volume.assert_not_called()

19
instances/views.py
View file

@ -421,8 +421,8 @@ def invalid_disk_options(
"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))),
"format": (format, lambda v: bool(DISK_FORMAT_RE.fullmatch(v))),
"serial": (serial, lambda v: bool(DISK_SERIAL_RE.fullmatch(v))),
}
return [
name
@ -823,7 +823,7 @@ def add_new_vol(request, pk):
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):
if not VOLUME_NAME_RE.fullmatch(name):
invalid.append("name")
if invalid:
return reject_disk_options(request, invalid)
@ -964,6 +964,10 @@ def edit_volume(request, pk):
zeroes = request.POST.get(
"vol_detect_zeroes", app_settings.INSTANCE_VOLUME_DEFAULT_DETECT_ZEROES
)
# The form shows "None" (or nothing) for a disk without driver type.
if format in ("", "None"):
current = next((d for d in instance.disks if d["dev"] == target_dev), {})
format = current.get("format") or ""
invalid = invalid_disk_options(
instance,
@ -988,7 +992,7 @@ def edit_volume(request, pk):
new_target_dev,
new_path,
target_bus=new_bus,
driver_type=format,
format_type=format,
cache_mode=cache,
readonly=readonly,
shareable=shareable,
@ -1603,7 +1607,6 @@ def del_owner(request, pk):
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]+)$")
@ -1633,7 +1636,7 @@ def clone(request, pk):
# shows those fields to them.
if request.user.is_superuser:
for key, value in request.POST.items():
if CLONE_POST_KEY_RE.match(key):
if CLONE_POST_KEY_RE.fullmatch(key):
clone_data[key] = value.strip()
if app_settings.CLONE_INSTANCE_AUTO_NAME == "True" and not clone_data["name"]:
@ -1661,12 +1664,12 @@ def clone(request, pk):
invalid_macs = [
value
for key, value in clone_data.items()
if key.startswith("clone-net-mac-") and not MAC_ADDRESS_RE.match(value)
if key.startswith("clone-net-mac-") and not MAC_ADDRESS_RE.fullmatch(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 key.startswith("disk-") and not VOLUME_NAME_RE.fullmatch(value or "")
]
if not request.user.is_superuser and quota_msg: