fix: add an authorization matrix test and close the endpoints it finds

Add webvirtcloud/test_authorization_matrix.py, which walks every named URL
and checks that a user without any VM/compute relation only reaches an
explicit allowlist, and that the global view_instances permission does not
open consoles. It found:

- /computes/<id>/... views without any check (compute_graph,
  get_compute_*, get_dom_capabilities, storages.get_volumes): the whole
  computes URL namespace is now wrapped with superuser_only, so a view
  added there later cannot be exposed by a missing decorator.
- guess_mac_address (leaks dhcpd.conf host/MAC pairs), random_mac_address
  and guess_clone_name: superuser-only, like the buttons that use them.
- check_instance: cross-tenant VM name lookup with no caller; removed.
- getvvfile and get_vdi_url: reachable with only view_instances, and the
  .vv file contains the VNC password.

Console access now follows one rule, instances.utils.can_open_console:
an active superuser or an active owner of the VM. The global
view_instances permission is read-only. It is used by the console page,
novncd, getvvfile and get_vdi_url (getvvfile and the VDI tab are kept for
virt-viewer users). The console page answers 403 instead of 500 when
access is denied, with the same answer for a missing VM.

computes/urls.py

@@ -1,6 +1,7 @@
 from virtsecrets.views import secrets
 
-from django.urls import include, path
+from admin.decorators import superuser_only
+from django.urls import URLPattern, URLResolver, include, path
 from interfaces.views import interface, interfaces
 from networks.views import network, networks
 from nwfilters.views import nwfilter, nwfilters
@@ -8,6 +9,22 @@ from storages.views import create_volume, get_volumes, storage, storages
 
 from . import forms, views
 
+
+def superuser_only_patterns(patterns):
+    """
+    Wrap every view below these patterns with superuser_only. Everything under
+    /computes/ is host administration (ROADMAP S-10), so the check is applied
+    to the whole namespace instead of per view, and a view added here later
+    cannot be exposed by forgetting a decorator.
+    """
+    for pattern in patterns:
+        if isinstance(pattern, URLResolver):
+            superuser_only_patterns(pattern.url_patterns)
+        elif isinstance(pattern, URLPattern):
+            pattern.callback = superuser_only(pattern.callback)
+    return patterns
+
+
 urlpatterns = [
     path("", views.computes, name="computes"),
     path(
@@ -77,3 +94,5 @@ urlpatterns = [
         ),
     ),
 ]
+
+superuser_only_patterns(urlpatterns)

console/novncd

@@ -89,8 +89,8 @@ def get_connection_infos(token):
     from django.conf import settings
     from django.contrib.auth import get_user_model
     from django.core.signing import BadSignature, SignatureExpired, TimestampSigner
-    from accounts.models import UserInstance
     from instances.models import Instance
+    from instances.utils import can_open_console
     from vrtManager.instance import wvmInstance
 
     host = None
@@ -126,13 +126,12 @@ def get_connection_infos(token):
             User = get_user_model()
             try:
                 user = User.objects.get(id=user_id)
-                if not user.is_active:
-                    raise PermissionError("User account is inactive or disabled")
-                if not user.is_superuser and not user.has_perm("instances.view_instances"):
-                    if not UserInstance.objects.filter(instance=instance, user=user).exists():
-                        raise PermissionError("User no longer has access to this instance")
             except User.DoesNotExist:
                 raise PermissionError("User not found")
+            if not user.is_active:
+                raise PermissionError("User account is inactive or disabled")
+            if not can_open_console(user, instance):
+                raise PermissionError("User no longer has access to this instance")
 
         conn = wvmInstance(
             instance.compute.hostname,

console/tests.py

@@ -154,11 +154,13 @@ class ConsoleViewsTestCase(TestCase):
         response = self.client.get(url)
 
         # Non-permitted user attempting to view other user's instance
-        self.assertEqual(response.status_code, 500)
+        self.assertEqual(response.status_code, 403)
         self.assertIn("permission", response.content.decode("utf-8").lower())
 
     @patch("console.views.wvmInstance")
-    def test_console_user_with_view_instances_perm(self, mock_wvm):
+    def test_console_view_instances_perm_does_not_open_console(self, mock_wvm):
+        # ROADMAP S-09: global view_instances is a read-only role; the console
+        # is for superusers and VM owners only.
         mock_conn = MagicMock()
         mock_conn.get_console_type.return_value = "vnc"
         mock_conn.get_console_websocket_port.return_value = None
@@ -171,7 +173,8 @@ class ConsoleViewsTestCase(TestCase):
         url = reverse("console") + f"?token={self.token}"
         response = self.client.get(url)
 
-        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.status_code, 403)
+        self.assertTemplateNotUsed(response, "console-vnc-lite.html")
 
     @patch("console.views.wvmInstance")
     @override_settings(WS_PUBLIC_PATH="/novncd/")
@@ -522,6 +525,25 @@ class NovncdDaemonLogicTestCase(TestCase):
         with self.assertRaises(PermissionError):
             novncd_mod.get_connection_infos(token)
 
+    def test_get_connection_infos_view_instances_user_is_not_enough(self):
+        User = get_user_model()
+        viewer = User.objects.create_user(username="novnc_viewer", password="password")
+        viewer.user_permissions.add(Permission.objects.get(codename="view_instances"))
+        compute = Compute.objects.create(
+            name="novnc-viewer-compute", hostname="127.0.0.1", login="root", password="", type=1
+        )
+        instance = Instance.objects.create(
+            compute=compute,
+            name="novnc-viewer-vm",
+            uuid="55555555-5555-5555-5555-555555555555",
+        )
+
+        signer = TimestampSigner(salt="console.novnc")
+        token = signer.sign(f"{compute.id}:{instance.uuid}:{viewer.id}")
+
+        with self.assertRaises(PermissionError):
+            novncd_mod.get_connection_infos(token)
+
     def test_get_connection_infos_inactive_user(self):
         User = get_user_model()
         inactive_user = User.objects.create_user(

console/views.py

@@ -1,14 +1,14 @@
 # pylint: disable=no-name-in-module,no-member
 import re
 
-from accounts.models import UserInstance
 from appsettings.settings import app_settings
 from django.conf import settings
 from django.core.signing import BadSignature, SignatureExpired, TimestampSigner
-from django.http.response import HttpResponseServerError
+from django.http.response import HttpResponseForbidden
 from django.shortcuts import render
 from django.utils.translation import gettext_lazy as _
 from instances.models import Instance
+from instances.utils import can_open_console
 from libvirt import libvirtError
 from vrtManager.instance import wvmInstance
 
@@ -79,24 +79,16 @@ def console(request):
         if host is None or uuid is None:
             raise ValueError("Invalid token")
 
-        if not request.user.is_superuser and not request.user.has_perm(
-            "instances.view_instances"
-        ):
-            try:
-                userInstance = UserInstance.objects.get(
-                    instance__compute_id=host,
-                    instance__uuid=uuid,
-                    user__id=request.user.id,
-                )
-                instance = Instance.objects.get(compute_id=host, uuid=uuid)
-            except UserInstance.DoesNotExist:
-                instance = None
-                console_error = _(
-                    "User does not have permission to access console or host/instance not exist"
-                )
-                return HttpResponseServerError(console_error)
-        else:
-            instance = Instance.objects.get(compute_id=host, uuid=uuid)
+        instance = Instance.objects.filter(compute_id=host, uuid=uuid).first()
+        if instance is None and request.user.is_superuser:
+            raise Instance.DoesNotExist
+        # Same answer for "no such VM" and "not allowed", so the response
+        # does not reveal whether a VM exists.
+        if instance is None or not can_open_console(request.user, instance):
+            console_error = _(
+                "User does not have permission to access console or host/instance not exist"
+            )
+            return HttpResponseForbidden(console_error)
 
         # Generate secure signed token for noVNC daemon
         token = signer.sign(f"{host}:{uuid}:{request.user.id}")

datasource/views.py

@@ -3,8 +3,11 @@ import socket
 
 from accounts.models import UserInstance, UserSSHKey
 from computes.models import Compute
+from django.core.exceptions import PermissionDenied
 from django.http import Http404, HttpResponse
 from django.shortcuts import get_object_or_404, render
+from instances.models import Instance
+from instances.utils import can_open_console
 from libvirt import libvirtError
 from vrtManager.instance import wvmInstance
 
@@ -95,6 +98,10 @@ def get_vdi_url(request, compute_id, vname):
     :return:
     """
     compute = get_object_or_404(Compute, pk=compute_id)
+    # The VDI URL points straight at the VM's console: same rule as the console.
+    instance = Instance.objects.filter(compute=compute, name=vname).first()
+    if instance is None or not can_open_console(request.user, instance):
+        raise PermissionDenied
 
     try:
         conn = wvmInstance(

instances/test_console_rule.py

@@ -0,0 +1,74 @@
+"""
+ROADMAP Wave 0 item 5 (S-09): one rule decides who may open a VM's console
+(noVNC page, novncd, the virt-viewer .vv file, the VDI URL): a superuser or
+an owner of the VM. The global view_instances permission is read-only.
+"""
+from unittest.mock import MagicMock, 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 accounts.models import UserInstance
+from computes.models import Compute
+from instances.models import Instance
+from instances.utils import can_open_console
+
+
+class CanOpenConsoleTestCase(TestCase):
+    def setUp(self):
+        User = get_user_model()
+        self.superuser = User.objects.create_superuser(
+            username="cons_super", email="cons_super@example.com", password="x"
+        )
+        self.owner = User.objects.create_user(username="cons_owner", password="x")
+        self.inactive_owner = User.objects.create_user(
+            username="cons_inactive", password="x", is_active=False
+        )
+        self.viewer = User.objects.create_user(username="cons_viewer", password="x")
+        self.viewer.user_permissions.add(Permission.objects.get(codename="view_instances"))
+        self.stranger = User.objects.create_user(username="cons_stranger", password="x")
+
+        self.compute = Compute.objects.create(
+            name="cons-compute", hostname="127.0.0.1:1", login="root", password="", type=1
+        )
+        self.instance = Instance.objects.create(
+            compute=self.compute,
+            name="cons-vm",
+            uuid="77777777-8888-9999-aaaa-bbbbbbbbbbbb",
+        )
+        UserInstance.objects.create(instance=self.instance, user=self.owner)
+        UserInstance.objects.create(instance=self.instance, user=self.inactive_owner)
+
+    def test_superuser_and_owner_may_open_console(self):
+        self.assertTrue(can_open_console(self.superuser, self.instance))
+        self.assertTrue(can_open_console(self.owner, self.instance))
+
+    def test_view_instances_stranger_and_inactive_owner_may_not(self):
+        self.assertFalse(can_open_console(self.viewer, self.instance))
+        self.assertFalse(can_open_console(self.stranger, self.instance))
+        self.assertFalse(can_open_console(self.inactive_owner, self.instance))
+
+    def test_owner_can_download_vv_file(self):
+        self.client.force_login(self.owner)
+        with patch("instances.views.wvmInstances") as mock_conn_cls:
+            conn = mock_conn_cls.return_value
+            conn.graphics_type.return_value = "vnc"
+            conn.graphics_listen.return_value = "127.0.0.1"
+            conn.graphics_port.return_value = "5900"
+            conn.domain_name.return_value = "cons-vm"
+            conn.graphics_passwd.return_value = "secret"
+            res = self.client.get(reverse("instances:getvvfile", args=[self.instance.id]))
+        self.assertEqual(res.status_code, 200)
+        self.assertIn(b"[virt-viewer]", res.content)
+
+    def test_owner_can_get_vdi_url(self):
+        self.client.force_login(self.owner)
+        with patch("datasource.views.wvmInstance", MagicMock()), patch(
+            "datasource.views.get_hostname_by_ip", return_value="host"
+        ):
+            res = self.client.get(
+                reverse("vdi_url", args=[self.compute.id, self.instance.name])
+            )
+        self.assertEqual(res.status_code, 200)

instances/tests.py

@@ -805,13 +805,6 @@ class InstancesTestCase(TestCase):
         )
         self.assertEqual(response.status_code, 200)
 
-    def test_check_instance(self):
-        response = self.client.get(
-            reverse("instances:check_instance", args=["test-vm"])
-        )
-        self.assertEqual(response.status_code, 200)
-        self.assertJSONEqual(response.content, {"vname": "test-vm", "exists": True})
-
     def test_start_template(self):
         # starting templates must fail
         self.assertEqual(self.instance.status, 5)

instances/urls.py

@@ -82,6 +82,5 @@ urlpatterns = [
     ),
     path("guess_clone_name/", views.guess_clone_name, name="guess_clone_name"),
     path("random_mac_address/", views.random_mac_address, name="random_mac_address"),
-    path("check_instance/<vname>/", views.check_instance, name="check_instance"),
     path("<int:pk>/sshkeys/", views.sshkeys, name="sshkeys"),
 ]

instances/utils.py

@@ -14,6 +14,20 @@ from vrtManager.instance import wvmInstance, wvmInstances
 from .models import Instance
 
 
+def can_open_console(user, instance):
+    """
+    The single rule for console access (noVNC page, novncd, the virt-viewer
+    .vv file and the VDI URL): an active superuser or an active owner of the
+    VM. The global view_instances permission is read-only and does not grant
+    an interactive console (ROADMAP S-09).
+    """
+    if not user.is_active:
+        return False
+    if user.is_superuser:
+        return True
+    return UserInstance.objects.filter(instance=instance, user=user).exists()
+
+
 def get_clone_free_names(size=10):
     prefix = app_settings.CLONE_INSTANCE_DEFAULT_PREFIX
     free_names = []

instances/views.py

@@ -255,6 +255,7 @@ def osinfo(request, pk):
     return JsonResponse(results)
 
 
+@superuser_only
 def guess_mac_address(request, vname):
     data = {"vname": vname}
     mac = utils.get_dhcp_mac_address(vname)
@@ -264,12 +265,14 @@ def guess_mac_address(request, vname):
     return HttpResponse(json.dumps(data))
 
 
+@superuser_only
 def random_mac_address(request):
     data = dict()
     data["mac"] = utils.get_random_mac_address()
     return HttpResponse(json.dumps(data))
 
 
+@superuser_only
 def guess_clone_name(request):
     dhcp_file = "/srv/webvirtcloud/dhcpd.conf"
     prefix = app_settings.CLONE_INSTANCE_DEFAULT_PREFIX
@@ -288,14 +291,6 @@ def guess_clone_name(request):
     return HttpResponse(json.dumps({}))
 
 
-def check_instance(request, vname):
-    instance = Instance.objects.filter(name=vname)
-    data = {"vname": vname, "exists": False}
-    if instance:
-        data["exists"] = True
-    return JsonResponse(data)
-
-
 def sshkeys(request, pk):
     """
     :param request:
@@ -1729,6 +1724,9 @@ def change_options(request, pk):
 
 def getvvfile(request, pk):
     instance = get_instance(request.user, pk)
+    # The .vv file contains the VNC password: same rule as the console.
+    if not utils.can_open_console(request.user, instance):
+        raise PermissionDenied
     conn = wvmInstances(
         instance.compute.hostname,
         instance.compute.login,

webvirtcloud/test_authorization_matrix.py

@@ -0,0 +1,151 @@
+"""
+Authorization matrix (ROADMAP Wave 0 item 5, S-09/S-10).
+
+Walks every named URL pattern and checks that an authenticated user without
+any relation to a VM or compute cannot reach it, unless the URL is in the
+explicit allowlist below. A new endpoint added without an authorization check
+makes this test fail, so the allowlist must be extended deliberately.
+
+Limitation: the compute points at a closed port, so a view that turns a
+libvirt connection error into 404 looks "denied" here even without an
+authorization check. Endpoints where that matters (consoles) get dedicated
+tests with libvirt mocked.
+"""
+from unittest.mock import MagicMock, patch
+
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import Permission
+from django.test import Client, TestCase
+from django.urls import NoReverseMatch, URLPattern, URLResolver, get_resolver, reverse
+
+from computes.models import Compute
+from instances.models import Instance
+
+# URL name -> why any authenticated user may reach it.
+ALLOWED_FOR_ANY_USER = {
+    "index": "landing page",
+    "instances:index": "lists only the user's own instances",
+    "accounts:login": "login page",
+    "accounts:email_otp": "part of the login flow (S-12 tracks its own issues)",
+    "accounts:logout": "logout",
+    "accounts:profile": "the user's own profile",
+    "accounts:ssh_key_create": "the user's own SSH keys",
+    "set_language": "UI language switch",
+    "rest_framework:login": "DRF browsable API login",
+    "rest_framework:logout": "DRF browsable API logout",
+    "console": "renders an error without a valid token; access is checked per VM",
+    "ds_openstack_index": "static cloud-init datasource index (F-14)",
+    "instance-list": "API: filtered to the user's own instances",
+    "compute-instance-list": "API: filtered to the user's own instances",
+    "instance-flavor-list": "API: flavor catalogue",
+    "instance-flavor-detail": "API: flavor catalogue",
+    "schema": "OpenAPI schema (SERVE_PERMISSIONS is S-21, Wave 3)",
+    "schema-json": "OpenAPI schema (SERVE_PERMISSIONS is S-21, Wave 3)",
+    "schema-redoc": "API docs (SERVE_PERMISSIONS is S-21, Wave 3)",
+    "schema-swagger-ui": "API docs (SERVE_PERMISSIONS is S-21, Wave 3)",
+}
+
+# Endpoints that expose a VM's console (or its VNC password) and therefore
+# must follow the console rule: superuser or VM owner, not global view_instances.
+CONSOLE_ENDPOINTS = ["instances:getvvfile", "vdi_url"]
+
+DENIED = (403, 404)
+
+
+def iter_patterns(patterns, prefix="", params=None):
+    params = params or {}
+    for p in patterns:
+        own = {**{g: None for g in p.pattern.regex.groupindex}, **p.pattern.converters}
+        if isinstance(p, URLResolver):
+            ns = f"{prefix}{p.namespace}:" if p.namespace else prefix
+            yield from iter_patterns(p.url_patterns, ns, {**params, **own})
+        elif isinstance(p, URLPattern) and p.name:
+            yield prefix + p.name, {**params, **own}
+
+
+class AuthorizationMatrixTestCase(TestCase):
+    def setUp(self):
+        User = get_user_model()
+        self.user = User.objects.create_user(username="matrix_plain", password="x")
+        self.viewer = User.objects.create_user(username="matrix_viewer", password="x")
+        self.viewer.user_permissions.add(Permission.objects.get(codename="view_instances"))
+        # Port 1 is closed, so any view that reaches libvirt fails fast.
+        self.compute = Compute.objects.create(
+            name="matrix-compute", hostname="127.0.0.1:1", login="root", password="", type=1
+        )
+        self.instance = Instance.objects.create(
+            compute=self.compute,
+            name="matrix-vm",
+            uuid="22222222-3333-4444-5555-666666666666",
+        )
+        self.client = Client(raise_request_exception=False)
+
+    def _kwargs(self, params):
+        kwargs = {}
+        for name, converter in params.items():
+            if name in ("compute_id", "compute_pk"):
+                kwargs[name] = self.compute.pk
+            elif name in ("pk", "instance_id"):
+                kwargs[name] = self.instance.pk if converter else str(self.instance.pk)
+            elif name == "vname":
+                kwargs[name] = self.instance.name
+            elif name == "format":
+                kwargs[name] = "json"
+            elif converter is not None and type(converter).__name__ == "IntConverter":
+                kwargs[name] = 1
+            else:
+                kwargs[name] = "x"
+        return kwargs
+
+    def _request(self, user, url):
+        self.client.force_login(user)
+        response = self.client.get(url)
+        if response.status_code == 405:
+            self.client.force_login(user)
+            response = self.client.post(url, {})
+        return response
+
+    def _url(self, name, params):
+        return reverse(name, kwargs=self._kwargs(params))
+
+    def test_plain_user_cannot_reach_anything_outside_the_allowlist(self):
+        reachable = []
+        for name, params in iter_patterns(get_resolver().url_patterns):
+            if name in ALLOWED_FOR_ANY_USER:
+                continue
+            try:
+                url = self._url(name, params)
+            except NoReverseMatch:
+                reachable.append(f"{name}: cannot build URL, extend _kwargs()")
+                continue
+            response = self._request(self.user, url)
+            if response.status_code not in DENIED:
+                reachable.append(f"{name} ({url}) -> {response.status_code}")
+        self.assertEqual(
+            reachable,
+            [],
+            "Reachable by a user without any VM/compute relation:\n" + "\n".join(reachable),
+        )
+
+    def test_allowlist_only_names_existing_urls(self):
+        names = {name for name, _ in iter_patterns(get_resolver().url_patterns)}
+        self.assertEqual(sorted(set(ALLOWED_FOR_ANY_USER) - names), [])
+
+    def test_global_view_permission_does_not_open_consoles(self):
+        params = dict(iter_patterns(get_resolver().url_patterns))
+        reachable = []
+        # Mock libvirt so the views would succeed if authorization let them through.
+        with patch("datasource.views.wvmInstance", MagicMock()), patch(
+            "datasource.views.get_hostname_by_ip", return_value="host"
+        ), patch("instances.views.wvmInstances", MagicMock()):
+            for name in CONSOLE_ENDPOINTS:
+                url = self._url(name, params[name])
+                response = self._request(self.viewer, url)
+                if response.status_code not in DENIED:
+                    reachable.append(f"{name} ({url}) -> {response.status_code}")
+        self.assertEqual(
+            reachable,
+            [],
+            "Console endpoints reachable with only global view_instances:\n"
+            + "\n".join(reachable),
+        )