mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 11:16:47 +00:00
aab8dbf1df
A running VM's page read its addresses while it rendered: the guest agent, when its channel is connected but it does not answer, holds the page (and a gunicorn worker) for its whole 10 s timeout, on every load. The page now shows them once it has loaded: vm_addresses.js asks the new instances/<pk>/addresses/ for each NIC's addresses and fills the header's IPv4 and each network card's IPv4 and IPv6.
325 lines
18 KiB
Python
325 lines
18 KiB
Python
"""
|
|
The VM page renders a control only for users the backend lets use it:
|
|
power to superusers and owners; the console, virt-viewer file and VDI URL to
|
|
superusers and owners (can_open_console); root password, SSH key, resize and
|
|
options to superusers and owners with is_change; per-vCPU hotplug, video
|
|
model and guest agent to superusers; clone with clone_instances on a template
|
|
or with is_change; snapshots with snapshot_instances and is_change, on a
|
|
template only for staff. is_staff and the global view_instances permission
|
|
alone are read-only.
|
|
"""
|
|
import shutil
|
|
import subprocess
|
|
import unittest
|
|
from unittest.mock import PropertyMock, 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 lxml import html as lxml_html
|
|
|
|
from accounts.models import UserInstance
|
|
from appsettings.models import AppSettings
|
|
from computes.models import Compute
|
|
from instances.models import Instance
|
|
from operations.models import Operation
|
|
|
|
CONSOLE = {"console", "vv file", "vdi"}
|
|
POWER = {"power"}
|
|
CHANGE = {"root password", "ssh key", "resize", "options"}
|
|
SUPERUSER_ONLY = {"vcpu hotplug", "video model", "guest agent"}
|
|
|
|
|
|
class PageControls:
|
|
def setUp(self):
|
|
AppSettings.objects.filter(
|
|
key__in=["SHOW_ACCESS_ROOT_PASSWORD", "SHOW_ACCESS_SSH_KEYS"]
|
|
).update(value="True")
|
|
self.compute = Compute.objects.create(
|
|
name="ctl-compute", hostname="127.0.0.1:1", login="root", password="", type=1
|
|
)
|
|
self.instance = Instance.objects.create(
|
|
compute=self.compute, name="ctl-vm", uuid="dddddddd-eeee-ffff-0000-111111111111"
|
|
)
|
|
|
|
def user(self, name, staff=False, perms=(), owner=None):
|
|
user = get_user_model().objects.create_user(name, password="x", is_staff=staff)
|
|
for codename in perms:
|
|
user.user_permissions.add(Permission.objects.get(codename=codename))
|
|
if owner is not None:
|
|
UserInstance.objects.create(instance=self.instance, user=user, **owner)
|
|
return user
|
|
|
|
def controls(self, user, status=1):
|
|
self.client.force_login(user)
|
|
with patch("instances.models.wvmInstance") as mock_wvm, patch.object(
|
|
Compute, "status", new_callable=PropertyMock, return_value=True
|
|
):
|
|
proxy = mock_wvm.return_value
|
|
proxy.get_memory.return_value = 1024
|
|
proxy.get_cur_memory.return_value = 1024
|
|
proxy.get_max_memory.return_value = 4096
|
|
proxy.get_vcpu.return_value = 2
|
|
proxy.get_cur_vcpu.return_value = 1
|
|
proxy.get_max_cpus.return_value = [1, 2, 3, 4]
|
|
# a running VM with hotpluggable vCPUs shows the per-vCPU buttons
|
|
proxy.get_vcpus.return_value = {
|
|
0: {"enabled": "yes", "hotpluggable": "no"},
|
|
1: {"enabled": "no", "hotpluggable": "yes"},
|
|
}
|
|
for getter in ("get_networks", "get_ifaces", "get_storages", "get_disk_devices",
|
|
"get_media_devices", "get_net_devices", "get_snapshot",
|
|
"get_snapshot_jobs"):
|
|
getattr(proxy, getter).return_value = []
|
|
proxy.get_hostdevs.return_value = ([], [], [])
|
|
proxy.get_console_type.return_value = "vnc"
|
|
proxy.get_status.return_value = status
|
|
response = self.client.get(reverse("instances:instance", args=[self.instance.id]))
|
|
self.assertEqual(response.status_code, 200)
|
|
self.last_page = response.content.decode()
|
|
doc = lxml_html.fromstring(response.content)
|
|
pk = self.instance.id
|
|
|
|
def form(name):
|
|
return bool(doc.xpath("//form[@action=$a]", a=reverse(f"instances:{name}", args=[pk])))
|
|
|
|
found = {
|
|
"console": bool(doc.xpath("//button[@id='console_button']")),
|
|
"vv file": bool(doc.xpath("//a[@href=$a]", a=reverse("instances:getvvfile", args=[pk]))),
|
|
"vdi": bool(doc.xpath("//*[@id='vdi_url']")),
|
|
"power": form("poweron") or bool(doc.xpath("//button[@formaction=$a]", a=reverse("instances:poweroff", args=[pk]))),
|
|
"root password": form("rootpasswd"),
|
|
"ssh key": form("add_public_key"),
|
|
# a running VM with hotplug shows per-vCPU buttons instead of the CPU form
|
|
"resize": form("resize_memory"),
|
|
"options": form("change_options"),
|
|
"console settings": form("update_console"),
|
|
# host pages (superuser-only views)
|
|
"host links": bool(doc.xpath("//a[@href=$a]", a=reverse("overview", args=[self.compute.id]))),
|
|
# a link to the destroy confirmation page
|
|
"destroy": bool(doc.xpath("//a[@href=$a]", a=reverse("instances:destroy", args=[pk]))),
|
|
"vcpu hotplug": form("set_vcpu"),
|
|
"video model": form("set_video_model"),
|
|
"guest agent": form("set_guest_agent"),
|
|
"clone": bool(doc.xpath("//*[@id='clone']")),
|
|
"snapshots": bool(doc.xpath("//*[@id='snapshots']")),
|
|
}
|
|
return {name for name, present in found.items() if present}
|
|
|
|
|
|
|
|
class PageControlsTestCase(PageControls, TestCase):
|
|
@unittest.skipUnless(shutil.which("node"), "node is not installed")
|
|
def test_the_page_scripts_parse(self):
|
|
superuser = get_user_model().objects.create_superuser("ctl_js", "ctl_js@example.com", "x")
|
|
viewer = self.user("ctl_js_viewer", perms=("view_instances",))
|
|
for user in (superuser, viewer):
|
|
self.controls(user)
|
|
scripts = lxml_html.fromstring(self.last_page).xpath("//script[not(@src) and not(@type='application/json')]/text()")
|
|
for number, script in enumerate(scripts):
|
|
with self.subTest(user=user.username, script=number):
|
|
result = subprocess.run(
|
|
["node", "--check", "-"], input=script, capture_output=True, text=True, timeout=30
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_the_header_names_the_vm_its_place_and_resources(self):
|
|
"""Host links for administrators, the VM list for its owner; the
|
|
guest agent's state in words; each disk with its device name."""
|
|
disks = [{"dev": "vda", "size": 2 << 30}, {"dev": "vdb", "size": 1 << 30}]
|
|
with patch.object(Instance, "disks", new_callable=PropertyMock, return_value=disks), \
|
|
patch.object(Instance, "guest_agent", new_callable=PropertyMock, return_value=True), \
|
|
patch.object(Instance, "guest_agent_ready", new_callable=PropertyMock, return_value=False):
|
|
self.controls(get_user_model().objects.create_superuser("hdr-admin", "a@example.com", "x"))
|
|
admin_page = lxml_html.fromstring(self.last_page)
|
|
self.controls(self.user("hdr-owner", owner={}))
|
|
owner_page = lxml_html.fromstring(self.last_page)
|
|
links = "//nav[@aria-label='Location']/a"
|
|
self.assertEqual([a.text_content().strip() for a in admin_page.xpath(links)], ["ctl-compute", "Instances"])
|
|
self.assertEqual([a.text_content().strip() for a in owner_page.xpath(links)], ["Instances"])
|
|
self.assertIn("Guest agent not connected", self.last_page)
|
|
self.assertIn("vda 2.0\xa0GB · vdb 1.0\xa0GB", self.last_page)
|
|
# filled by its script once the page has loaded ("—" when it has none)
|
|
self.assertEqual(owner_page.xpath("//dt[.='IPv4']/following-sibling::dd/@data-empty"), ["—"])
|
|
|
|
def test_superuser_sees_every_control(self):
|
|
superuser = get_user_model().objects.create_superuser("ctl_super", "ctl_super@example.com", "x")
|
|
self.assertEqual(
|
|
self.controls(superuser),
|
|
CONSOLE | POWER | CHANGE | SUPERUSER_ONLY
|
|
| {"clone", "snapshots", "console settings", "destroy", "host links"},
|
|
)
|
|
|
|
def test_the_job_panel_gets_the_jobs_just_submitted_once(self):
|
|
superuser = get_user_model().objects.create_superuser("ctl_jobs", "ctl_jobs@example.com", "x")
|
|
self.client.force_login(superuser)
|
|
session = self.client.session
|
|
other = {"resource": "instance:aaaaaaaa-bbbb-cccc-0000-111111111111", "id": "4343"}
|
|
session["operations_shown"] = [{"resource": f"instance:{self.instance.uuid}", "id": "4242"}, other]
|
|
session.save()
|
|
self.controls(superuser)
|
|
self.assertIn('["4242"]', self.last_page)
|
|
self.assertNotIn('"4343"', self.last_page)
|
|
# a job remembered for another VM waits for that VM's page
|
|
self.assertEqual(self.client.session["operations_shown"], [other])
|
|
self.controls(superuser)
|
|
self.assertNotIn('"4242"', self.last_page)
|
|
|
|
def test_global_viewers_see_nothing_to_act_on(self):
|
|
perms = ("view_instances", "clone_instances", "snapshot_instances")
|
|
for user in (self.user("ctl_viewer", perms=perms), self.user("ctl_staff", staff=True, perms=perms)):
|
|
with self.subTest(user=user.username):
|
|
self.assertEqual(self.controls(user), set())
|
|
|
|
def test_owner_without_change_powers_and_opens_the_console(self):
|
|
owner = self.user("ctl_owner", perms=("clone_instances", "snapshot_instances"), owner={})
|
|
self.assertEqual(self.controls(owner), CONSOLE | POWER)
|
|
|
|
def test_owner_with_change_gets_change_clone_and_snapshots(self):
|
|
changer = self.user("ctl_changer", perms=("clone_instances", "snapshot_instances"),
|
|
owner={"is_change": True})
|
|
self.assertEqual(self.controls(changer), CONSOLE | POWER | CHANGE | {"clone", "snapshots"})
|
|
|
|
def test_a_paused_vm_can_be_destroyed(self):
|
|
# destroy stops a paused VM like a running one
|
|
owner = self.user("paused_deleter", owner={"is_delete": True})
|
|
self.assertIn("destroy", self.controls(owner, status=3))
|
|
|
|
def test_disks_grow_while_running_paused_or_shut_off(self):
|
|
# QEMU grows a running or paused VM's disk itself (blockResize)
|
|
changer = self.user("disk_resizer", owner={"is_change": True})
|
|
for status, enabled in ((1, True), (3, True), (5, True), (7, False)):
|
|
with self.subTest(status=status):
|
|
self.controls(changer, status=status)
|
|
buttons = lxml_html.fromstring(self.last_page).xpath(
|
|
"//form[@action=$a]//button[@type='submit']",
|
|
a=reverse("instances:resize_disk", args=[self.instance.id]),
|
|
)
|
|
self.assertEqual(bool(buttons), enabled)
|
|
|
|
def test_a_remote_viewer_only_for_a_console_beyond_the_host_s_loopback(self):
|
|
admin = get_user_model().objects.create_superuser("vdi_admin", "vdi@example.com", "x")
|
|
for listen, shown in (("0.0.0.0", True), ("192.0.2.5", True), ("127.0.0.2", False), ("::1", False)):
|
|
with self.subTest(listen=listen), patch("instances.models.Instance.console_listener_address",
|
|
new_callable=PropertyMock, return_value=listen):
|
|
self.controls(admin)
|
|
self.assertEqual('id="vdi_url"' in self.last_page, shown)
|
|
|
|
def test_the_power_tab_offers_the_state_s_actions_and_confirms_a_power_cut(self):
|
|
admin = get_user_model().objects.create_superuser("power_admin", "pa@example.com", "x")
|
|
owner = self.user("power_owner", owner={})
|
|
# each action under the state it leads to (active, suspended, off), the power cut ones after them
|
|
for user, status, names in ((admin, 1, ["suspend", "poweroff", "powerforce", "powercycle"]),
|
|
(admin, 3, ["resume", "powerforce"]), (owner, 3, []), (admin, 5, ["poweron"])):
|
|
with self.subTest(user=user.username, status=status):
|
|
self.controls(user, status=status)
|
|
(tab,) = lxml_html.fromstring(self.last_page).xpath("//div[@id='power']")
|
|
self.assertEqual(tab.xpath(".//button/@name"), names)
|
|
# the power-cut actions ask first
|
|
danger = tab.xpath(".//button[contains(@class, 'danger')]")
|
|
self.assertEqual(len(danger), len({"powerforce", "powercycle"} & set(names)))
|
|
self.assertTrue(all(b.get("data-confirm") for b in danger))
|
|
if user == owner:
|
|
self.assertIn("An administrator can resume it.", tab.text_content())
|
|
if status == 5:
|
|
self.assertEqual(tab.xpath(".//form/@data-please-wait"), [""])
|
|
|
|
def test_the_power_tab_marks_the_current_state_or_names_another(self):
|
|
admin = get_user_model().objects.create_superuser("power_strip_admin", "ps@example.com", "x")
|
|
for status, now in ((1, "Active"), (3, "Suspended"), (5, "Off"), (6, None)):
|
|
with self.subTest(status=status):
|
|
self.controls(admin, status=status)
|
|
(tab,) = lxml_html.fromstring(self.last_page).xpath("//div[@id='power']")
|
|
marked = [s.text_content() for s in tab.xpath(".//span[@class='fw-semibold']")]
|
|
self.assertEqual(marked, [now] if now else [])
|
|
self.assertEqual("The VM's state: Crashed" in tab.text_content(), now is None)
|
|
|
|
def test_a_running_paused_or_shut_off_vm_can_be_cloned(self):
|
|
changer = self.user("vm_cloner", perms=("clone_instances",), owner={"is_change": True})
|
|
for status, enabled in ((1, True), (3, True), (5, True), (4, False)):
|
|
with self.subTest(status=status):
|
|
self.controls(changer, status=status)
|
|
buttons = lxml_html.fromstring(self.last_page).xpath(
|
|
"//form[@action=$a]//button[@type='submit']",
|
|
a=reverse("instances:clone", args=[self.instance.id]),
|
|
)
|
|
self.assertEqual(bool(buttons), enabled)
|
|
|
|
def test_owner_with_change_and_no_extra_permissions(self):
|
|
changer = self.user("ctl_plain_changer", owner={"is_change": True})
|
|
self.assertEqual(self.controls(changer), CONSOLE | POWER | CHANGE)
|
|
|
|
|
|
class TemplatePageControlsTestCase(PageControls, TestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.instance.is_template = True
|
|
self.instance.save()
|
|
|
|
def test_a_template_is_cloned_by_viewing_it(self):
|
|
viewer = self.user("tpl_viewer", perms=("view_instances", "clone_instances"))
|
|
self.assertEqual(self.controls(viewer, status=5), {"clone"})
|
|
|
|
def test_only_staff_owners_change_a_template(self):
|
|
# its access (root password, SSH key) is for a VM that runs: not on a template's page
|
|
owner = self.user("tpl_plain_changer", owner={"is_change": True})
|
|
staff = self.user("tpl_staff_owner", staff=True, owner={"is_change": True})
|
|
self.assertEqual(self.controls(owner, status=5) & CHANGE, set())
|
|
self.assertEqual(self.controls(staff, status=5) & CHANGE, {"resize", "options"})
|
|
|
|
def test_a_template_s_page_has_no_console_settings_or_snapshots(self):
|
|
# a template is cloned, not run
|
|
flags = {"is_change": True, "is_vnc": True}
|
|
perms = ("clone_instances", "snapshot_instances")
|
|
for user in (self.user("tpl_vnc_owner", perms=perms, owner=flags),
|
|
self.user("tpl_vnc_staff", staff=True, perms=perms, owner=flags)):
|
|
with self.subTest(user=user.username):
|
|
controls = self.controls(user, status=5)
|
|
self.assertNotIn("console settings", controls)
|
|
self.assertNotIn("snapshots", controls)
|
|
|
|
def test_a_template_s_page_opens_on_its_clone_form(self):
|
|
viewer = self.user("tpl_cloner", perms=("view_instances", "clone_instances"))
|
|
self.controls(viewer, status=5)
|
|
doc = lxml_html.fromstring(self.last_page)
|
|
self.assertEqual(doc.xpath("string(//section[@aria-labelledby='vm-name']//span[@class='badge text-bg-light border'])"),
|
|
"Template")
|
|
self.assertTrue(doc.xpath("//button[@id='clone-button']"))
|
|
self.assertEqual(doc.xpath("string(//section[@id='vm-summary']/@data-opens-on)"), "#clone")
|
|
for pane in ("power", "access", "graphics", "vncsettings", "hostdevs"):
|
|
self.assertEqual(doc.xpath("//*[@id=$p]", p=pane), [], pane)
|
|
|
|
def test_only_staff_owners_destroy_a_template(self):
|
|
owner = self.user("tpl_deleter", owner={"is_delete": True})
|
|
staff = self.user("tpl_staff_deleter", staff=True, owner={"is_delete": True})
|
|
self.assertNotIn("destroy", self.controls(owner, status=5))
|
|
self.assertIn("destroy", self.controls(staff, status=5))
|
|
|
|
|
|
class TemplateSnapshotRuleTestCase(PageControls, TestCase):
|
|
"""The backend side of the template rule: only staff owners (and
|
|
superusers) take a snapshot of a template; other owners with is_change
|
|
and snapshot_instances are turned away without a libvirt call."""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.instance.is_template = True
|
|
self.instance.save()
|
|
|
|
def snapshot(self, user):
|
|
self.client.force_login(user)
|
|
with patch("instances.models.wvmInstance"), patch("instances.jobs.wvmInstance") as mock_wvm:
|
|
mock_wvm.return_value.instance.snapshotListNames.return_value = []
|
|
self.client.post(
|
|
reverse("instances:snapshot", args=[self.instance.id]),
|
|
{"name": "s1", "description": "", "submission_key": "key-snap0001"},
|
|
)
|
|
return Operation.objects.filter(kind="snapshot.create").exists()
|
|
|
|
def test_only_staff_owners_snapshot_a_template(self):
|
|
perms = ("snapshot_instances",)
|
|
owner = self.user("snap_changer", perms=perms, owner={"is_change": True})
|
|
staff = self.user("snap_staff_changer", staff=True, perms=perms, owner={"is_change": True})
|
|
self.assertFalse(self.snapshot(owner))
|
|
self.assertTrue(self.snapshot(staff))
|