1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-08 19:26:48 +00:00
webvirtcloud/instances/tests/test_xss.py
catborise 51d24fce30 feat(instances): migrate a VM with its disks copied into pools of the destination
A migration needed every disk on storage both hosts share: it was refused
when the destination lacked a disk file, and a file of the same name
there was taken for the VM's own disk.

The Migrate tab now lists the VM's disk files for the chosen destination.
For each disk it offers the destination's pools that keep disk files
(directory, filesystem and NFS pools). A disk on storage both hosts share
can also stay where it is.

A directory counts as shared when both hosts have it as a storage pool
and a marker file created on the source exists on the destination. A
directory the probe cannot tell is never taken as shared.

libvirt copies the chosen disks during the migration: migrate3 with
NON_SHARED_DISK for those targets, and destination and persistent
definitions naming the new paths, where libvirt creates the files.

- A shut-off VM is started paused for the copy and shut off on the
  destination; its CPUs never run. Its UEFI variables are copied first,
  as the destination would start from the firmware's template. If the
  migration fails, the VM is shut off on the source again.
- Refused before anything changes, each disk with its reason:
  - a disk neither shared nor given a pool, also one only the next
    start's definition has;
  - CD-ROM images, read-only or encrypted disks, formats other than raw
    and qcow2, and disks with a backing file;
  - a pool that is not of disk files, or that has a file of that name,
    which libvirt would reuse and overwrite;
  - two disks copied to one file.
- The job claims each copy's name in its pool, and the source volume.
- A failed or cancelled migration deletes the copies libvirt made, and
  the UEFI variables copy, unless a VM on the destination has them.
- The source files stay and are named in the log. With "Delete the
  source files of copied disks" they go, except one another VM or volume
  uses; a failed delete is logged and the migration stands.

The NVRAM sharing check uses the same probe: a shut-off move stops when
it cannot be told, and the source file stays.

The API's migrate call takes disk_pools ({target: pool}) and
delete_source.
2026-10-29 19:46:31 +03:00

160 lines
6.7 KiB
Python

"""
JSON endpoints must not be served as text/html (reflected XSS), and
templates must not build HTML by concatenating server data that contains
user-controlled names (stored XSS).
"""
import re
from pathlib import Path
from unittest.mock import patch
from django.contrib.auth import get_user_model
from django.test import TestCase
from django.urls import reverse
from accounts.models import UserInstance, UserSSHKey
from computes.models import Compute
from instances.models import Instance
ROOT = Path(__file__).resolve().parent.parent.parent
PAYLOAD = "<script>alert(1)</script>"
class JsonEndpointContentTypeTestCase(TestCase):
def setUp(self):
User = get_user_model()
self.superuser = User.objects.create_superuser(
username="xss_super", email="xss_super@example.com", password="password"
)
self.compute = Compute.objects.create(
name="xss-compute", hostname="127.0.0.1", login="root", password="", type=1
)
self.instance = Instance.objects.create(
compute=self.compute,
name="xss-vm",
uuid="99999999-8888-7777-6666-555555555555",
)
UserInstance.objects.create(instance=self.instance, user=self.superuser)
UserSSHKey.objects.create(
user=self.superuser,
keyname="k",
keypublic=f"ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI {PAYLOAD}",
)
self.client.force_login(self.superuser)
def assertJson(self, response):
self.assertEqual(response.status_code, 200)
self.assertEqual(response["Content-Type"], "application/json")
def test_guess_mac_address_reflects_vname_as_json(self):
res = self.client.get(
reverse("instances:guess_mac_address", args=["<img src=x onerror=alert(1)>"])
)
self.assertJson(res)
def test_random_mac_address_is_json(self):
self.assertJson(self.client.get(reverse("instances:random_mac_address")))
def test_guess_clone_name_is_json(self):
self.assertJson(self.client.get(reverse("instances:guess_clone_name")))
def test_sshkeys_is_json(self):
self.assertJson(
self.client.get(reverse("instances:sshkeys", args=[self.instance.id]))
)
def test_sshkeys_plain_is_text_plain(self):
res = self.client.get(
reverse("instances:sshkeys", args=[self.instance.id]) + "?plain=true"
)
self.assertEqual(res.status_code, 200)
self.assertTrue(res["Content-Type"].startswith("text/plain"))
def test_storage_volumes_is_json(self):
with patch("storages.views.wvmStorage") as mock_storage:
mock_storage.return_value.get_volumes.return_value = [PAYLOAD]
res = self.client.get(
reverse("volumes", args=[self.compute.id, "default"])
)
self.assertJson(res)
def test_datasource_metadata_is_json(self):
self.assertJson(
self.client.get(reverse("ds_openstack_metadata", args=["latest"]))
)
def test_compute_json_endpoints_are_json(self):
urls = [
reverse("compute_graph", args=[self.compute.id]),
reverse("machines", args=[self.compute.id, "x86_64"]),
reverse("buses", args=[self.compute.id, "x86_64", "q35", "disk"]),
]
with patch("computes.views.ComputeManager") as mock_mgr:
for method in ("compute_graph", "get_machine_types", "get_disk_buses"):
getattr(mock_mgr.return_value, method).return_value = "{}"
for url in urls:
with self.subTest(url=url):
self.assertJson(self.client.get(url))
def test_vm_logs_is_json(self):
self.assertJson(self.client.get(reverse("vm_logs", args=[self.instance.pk])))
class TemplateHtmlConcatenationTestCase(TestCase):
"""
The log table and the volume dropdown are filled by JavaScript from JSON
that contains user-entered names. They must use DOM APIs (.text()/.val())
instead of concatenating the values into an HTML string.
"""
def _read(self, relative):
return (ROOT / relative).read_text()
def test_logs_table_does_not_concatenate_log_fields_into_html(self):
src = self._read("instances/templates/instances/stats_tab.html")
for field in ("date", "user", "message"):
self.assertNotIn(f"+row['{field}']+", src)
self.assertNotIn('$("#logs_table > tbody").html(', src)
def test_volume_dropdown_does_not_concatenate_volume_names_into_html(self):
src = self._read("instances/templates/instance.html")
self.assertNotIn("'<option value=' + item", src)
self.assertNotIn('pool + "<span', src)
def test_create_wizard_chipset_list_does_not_concatenate_into_html(self):
src = self._read("instances/templates/create_instance_w1.html")
self.assertNotIn("""append('<option value="' + item""", src)
def test_create_wizard_volume_lists_do_not_concatenate_volume_names_into_html(self):
src = self._read("instances/templates/create_instance_w2.html")
self.assertNotIn("'<option value=' + item", src)
# Every template line that builds HTML from JavaScript must be listed here with
# a reason; a new one fails the test and has to be reviewed (use .text() /
# .val() / DOM APIs for server data instead).
HTML_SINK = re.compile(r"""\.html\(|innerHTML\s*=|append\(\s*['"]<""")
ALLOWED_HTML_SINKS = {
("accounts/templates/login.html", "$btn.html("): "static spinner markup",
("accounts/templates/accounts/otp_login.html", "$btn.html("): "static spinner markup",
("instances/templates/instance.html", "//sto_input.innerHTML = pool;"): "comment",
("instances/templates/instance.html", "hostDevices.innerHTML = html;"): "the panel's own fragment, host_devices_form.html, escaped by Django",
("instances/templates/instance.html", "migratePlan.innerHTML = html;"): "the panel's own fragment, migrate_plan.html, escaped by Django",
}
class TemplateHtmlSinkInventoryTestCase(TestCase):
def test_every_html_sink_in_templates_is_reviewed(self):
found = set()
for path in ROOT.glob("*/templates/**/*.html"):
rel = str(path.relative_to(ROOT))
for line in path.read_text().splitlines():
if HTML_SINK.search(line):
found.add((rel, line.strip()))
templates_root = ROOT / "templates"
for path in templates_root.glob("**/*.html"):
rel = str(path.relative_to(ROOT))
for line in path.read_text().splitlines():
if HTML_SINK.search(line):
found.add((rel, line.strip()))
self.assertEqual(sorted(found - set(ALLOWED_HTML_SINKS)), [], "unreviewed HTML sinks")
self.assertEqual(sorted(set(ALLOWED_HTML_SINKS) - found), [], "stale allowlist entries")