diff --git a/instances/forms.py b/instances/forms.py index 38867454..156bc3c6 100644 --- a/instances/forms.py +++ b/instances/forms.py @@ -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) diff --git a/instances/test_console_listener.py b/instances/test_console_listener.py new file mode 100644 index 00000000..8bbcb80a --- /dev/null +++ b/instances/test_console_listener.py @@ -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"{graphics}" + proxy._XMLDesc = lambda flags: xml if flags == PERSISTENT_XML else "" + return proxy.console_listens_locally() + + def test_cases(self): + cases = { + "": True, + "": True, + "": True, + "": True, # no console + "": False, + "": False, + "": False, # the host default decides + "": False, + "": False, + } + for graphics, expected in cases.items(): + with self.subTest(graphics=graphics): + self.assertIs(self.listens_locally(graphics), expected) diff --git a/instances/views.py b/instances/views.py index 59ae796b..40ce2e75 100755 --- a/instances/views.py +++ b/instances/views.py @@ -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." diff --git a/vrtManager/instance.py b/vrtManager/instance.py index df563b7d..6f2f2a29 100644 --- a/vrtManager/instance.py +++ b/vrtManager/instance.py @@ -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