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

fix(instances): only administrators choose where a console listens

Browse source 
A VM owner who may manage the console (is_change and is_vnc) could set
its listen address to all interfaces and remove its password: a VNC
console reachable from the network without a password. Checked on a
test host: the owner's form changed the VM's listen address to 0.0.0.0
and then removed the password.

- The listen address field is disabled for users who are not
  superusers; a disabled field ignores posted values. They can still
  set or change the password and the keymap.
- A console whose persistent definition does not bind it to loopback
  (or a Unix socket) only loses its password through an administrator.
  A binding the XML does not pin down counts as not local: listen type
  network, or no address (the host's default in qemu.conf).

On the test host, the owner's form now leaves the listen address alone
and is refused when it removes the password of a console listening on
0.0.0.0; with the console on localhost the password can be removed.
The new tests fail on the previous code.
This commit is contained in:
catborise 2026-10-05 15:07:31 +03:00
parent eab27f8aba
commit f3929b8e8f
4 changed files with 147 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

9
instances/forms.py
View file

@ -25,12 +25,17 @@ class ConsoleForm(forms.Form):
clear_keymap = forms.BooleanField(label=_("Clear keymap"), required=False)
keymap = forms.ChoiceField(label=_("Keymap"), required=False)
def __init__(self, *args, **kwargs):
def __init__(self, *args, can_set_listener=True, **kwargs):
super(ConsoleForm, self).__init__(*args, **kwargs)
keymap_choices = [("auto", _("Auto"))] + list((c, c) for c in QEMU_KEYMAPS)
# Only administrators choose the listen address: beyond localhost the
# console is reachable from the network. A disabled field ignores
# posted values.
self.fields["listen_on"] = forms.ChoiceField(
label=_("Listen on"),
choices=QEMU_CONSOLE_LISTENER_ADDRESSES
choices=QEMU_CONSOLE_LISTENER_ADDRESSES,
disabled=not can_set_listener,
required=can_set_listener,
)
self.fields["keymap"] = forms.ChoiceField(label=_("Keymap"), choices=keymap_choices)

98
instances/test_console_listener.py Normal file
View file

@ -0,0 +1,98 @@
"""Only administrators choose where a VM's console listens. Other users who may
manage the console cannot make it listen beyond localhost, and cannot remove
the password of a console that does."""
from unittest.mock import PropertyMock, patch
from accounts.models import UserInstance
from computes.models import Compute
from django.contrib.auth import get_user_model
from django.contrib.messages import get_messages
from django.test import TestCase
from django.urls import reverse
from instances.models import Instance
from vrtManager.instance import PERSISTENT_XML, wvmInstance
class ConsoleListenerTests(TestCase):
def setUp(self):
User = get_user_model()
self.owner = User.objects.create_user("console-owner", password="pw")
self.admin = User.objects.create_superuser("console-admin", password="pw", email="ca@example.com")
compute = Compute.objects.create(name="cl", hostname="10.0.0.1", login="u", password="p", type=1)
self.vm = Instance.objects.create(compute=compute, name="cl-vm", uuid="78787878-0000-0000-0000-000000000000")
UserInstance.objects.create(user=self.owner, instance=self.vm, is_change=True, is_vnc=True)
def post(self, user, data, local=True):
self.client.force_login(user)
with patch("instances.models.wvmInstance") as wvm:
proxy = wvm.return_value
proxy.console_listens_locally.return_value = local
proxy.set_console_passwd.return_value = True
response = self.client.post(reverse("instances:update_console", args=[self.vm.id]), data)
return proxy, [str(m) for m in get_messages(response.wsgi_request)]
def test_owner_cannot_change_the_listen_address(self):
proxy, _ = self.post(self.owner, {"listen_on": "0.0.0.0", "keymap": "auto"})
proxy.set_console_listener_addr.assert_not_called()
def test_owner_cannot_remove_the_password_of_a_console_listening_beyond_localhost(self):
proxy, messages = self.post(self.owner, {"clear_password": "on", "keymap": "auto"}, local=False)
proxy.set_console_passwd.assert_not_called()
self.assertTrue(any("password" in m.lower() for m in messages), messages)
def test_owner_can_remove_the_password_of_a_local_console(self):
proxy, _ = self.post(self.owner, {"clear_password": "on", "keymap": "auto"})
proxy.set_console_passwd.assert_called_once_with("")
def test_owner_can_set_a_password_of_any_console(self):
proxy, _ = self.post(self.owner, {"password": "s3cret-pass", "keymap": "auto"}, local=False)
proxy.set_console_passwd.assert_called_once_with("s3cret-pass")
def test_superuser_can_open_the_console_to_all_interfaces(self):
proxy, _ = self.post(self.admin, {"listen_on": "0.0.0.0", "clear_password": "on", "keymap": "auto"}, local=False)
proxy.set_console_listener_addr.assert_called_once_with("0.0.0.0")
proxy.set_console_passwd.assert_called_once_with("")
def test_owner_sees_the_listen_address_but_cannot_edit_it(self):
self.client.force_login(self.owner)
with patch("instances.models.wvmInstance") as wvm, patch.object(
Compute, "status", new_callable=PropertyMock, return_value=True
):
proxy = 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_console_listener_addr.return_value = "127.0.0.1"
for name in ("get_networks", "get_ifaces", "get_storages", "get_disk_devices", "get_media_devices", "get_net_devices"):
getattr(proxy, name).return_value = []
proxy.get_status.return_value = 5
response = self.client.get(reverse("instances:instance", args=[self.vm.id]))
form = response.context["console_form"]
self.assertTrue(form.fields["listen_on"].disabled)
class ConsoleListensLocallyTests(TestCase):
"""Only a binding the persistent XML pins to loopback (or a Unix socket) is local."""
def listens_locally(self, graphics):
proxy = wvmInstance.__new__(wvmInstance) # no connection: only the XML is read
xml = f"<domain><devices>{graphics}</devices></domain>"
proxy._XMLDesc = lambda flags: xml if flags == PERSISTENT_XML else "<wrong-flags/>"
return proxy.console_listens_locally()
def test_cases(self):
cases = {
"<graphics type='vnc' listen='127.0.0.1'><listen type='address' address='127.0.0.1'/></graphics>": True,
"<graphics type='vnc'><listen type='address' address='::1'/></graphics>": True,
"<graphics type='vnc'><listen type='socket' socket='/run/vnc.sock'/></graphics>": True,
"": True, # no console
"<graphics type='vnc' listen='0.0.0.0'><listen type='address' address='0.0.0.0'/></graphics>": False,
"<graphics type='vnc'><listen type='network' network='default'/></graphics>": False,
"<graphics type='vnc' autoport='yes'/>": False, # the host default decides
"<graphics type='vnc'><listen type='address' address='localhost'/></graphics>": False,
"<graphics type='vnc' listen='127.0.0.1'><listen type='address' address='10.0.0.5'/></graphics>": False,
}
for graphics, expected in cases.items():
with self.subTest(graphics=graphics):
self.assertIs(self.listens_locally(graphics), expected)

19
instances/views.py
View file

@ -110,7 +110,8 @@ def instance(request, pk):
"listen_on": instance.console_listener_address,
"password": instance.console_passwd,
"keymap": instance.console_keymap,
}
},
can_set_listener=request.user.is_superuser,
)
console_listener_addresses = settings.QEMU_CONSOLE_LISTENER_ADDRESSES
bottom_bar = app_settings.VIEW_INSTANCE_DETAIL_BOTTOM_BAR
@ -1887,7 +1888,7 @@ def update_console(request, pk):
instance = get_instance(request.user, pk, perm_type="change")
if utils.can_manage_console(request.user, instance):
form = ConsoleForm(request.POST or None)
form = ConsoleForm(request.POST or None, can_set_listener=request.user.is_superuser)
if form.is_valid():
if (
"generate_password" in form.changed_data
@ -1901,7 +1902,19 @@ def update_console(request, pk):
else:
password = form.cleaned_data["password"]
if not instance.proxy.set_console_passwd(password):
# A console that listens beyond localhost (set by an
# administrator) keeps its password unless an administrator
# removes it. The definition being changed is the persistent one.
if (
not password
and not request.user.is_superuser
and not instance.proxy.console_listens_locally()
):
messages.error(
request,
_("The console listens beyond localhost; only an administrator can remove its password."),
)
elif not instance.proxy.set_console_passwd(password):
msg = _(
"Error setting console password. "
+ "You should check that your instance have an graphic device."

26
vrtManager/instance.py
View file

@ -1,4 +1,5 @@
import contextlib
import ipaddress
import json
import logging
import os.path
@ -1092,6 +1093,31 @@ class wvmInstance(wvmConnect):
return "127.0.0.1"
return listener_addr
def console_listens_locally(self):
"""True if the persistent definition binds the console to loopback
addresses or a Unix socket only. Anything the XML does not pin down
counts as not local: listen type network, or no address at all (the
host's default in qemu.conf)."""
graphic = etree.fromstring(self._XMLDesc(PERSISTENT_XML)).find("devices/graphics")
if graphic is None:
return True # no console
addresses = [graphic.get("listen")]
local_types = 0
for listen in graphic.findall("listen"):
if listen.get("type") == "address":
addresses.append(listen.get("address"))
elif listen.get("type") in ("socket", "none"):
local_types += 1
else:
return False
addresses = [a for a in addresses if a]
if not addresses:
return local_types > 0
try:
return all(ipaddress.ip_address(a).is_loopback for a in addresses)
except ValueError: # a host name
return False
def set_console_listener_addr(self, listener_addr):
root = etree.fromstring(self._XMLDesc(PERSISTENT_XML))
# The console is the first graphics device of the definition being