feat(security): apply Codex review hardening to console tokens and superuser authorization

admin/decorators.py

@@ -1,7 +1,9 @@
+from functools import wraps
 from django.core.exceptions import PermissionDenied
 
 
 def superuser_only(function):
+    @wraps(function)
     def _inner(request, *args, **kwargs):
         if not request.user.is_superuser:
             raise PermissionDenied

admin/permissions.py

@@ -0,0 +1,15 @@
+from rest_framework.permissions import BasePermission
+
+
+class IsSuperUser(BasePermission):
+    """
+    Allows access only to active superusers.
+    """
+
+    def has_permission(self, request, view):
+        return bool(
+            request.user
+            and request.user.is_authenticated
+            and request.user.is_active
+            and request.user.is_superuser
+        )

computes/api/viewsets.py

@@ -1,5 +1,6 @@
+from admin.permissions import IsSuperUser
 from computes.models import Compute
-from rest_framework import permissions, viewsets
+from rest_framework import viewsets
 from rest_framework.response import Response
 from vrtManager.create import wvmCreate
 
@@ -13,11 +14,11 @@ class ComputeViewSet(viewsets.ModelViewSet):
 
     queryset = Compute.objects.all().order_by("name")
     serializer_class = ComputeSerializer
-    permission_classes = [permissions.IsAdminUser]
+    permission_classes = [IsSuperUser]
 
 
 class ComputeArchitecturesView(viewsets.ViewSet):
-    permission_classes = [permissions.IsAdminUser]
+    permission_classes = [IsSuperUser]
 
     def list(self, request, compute_pk=None):
         """
@@ -44,7 +45,7 @@ class ComputeArchitecturesView(viewsets.ViewSet):
 
 
 class ComputeMachinesView(viewsets.ViewSet):
-    permission_classes = [permissions.IsAdminUser]
+    permission_classes = [IsSuperUser]
 
     def list(self, request, compute_pk=None, archs_pk=None):
         """

computes/test_api_security.py

@@ -23,6 +23,13 @@ class APISecurityTestCase(TestCase):
             email="user@example.com",
             password="userpassword",
         )
+        self.staff_user = User.objects.create_user(
+            username="staff_api_user",
+            email="staff@example.com",
+            password="staffpassword",
+            is_staff=True,
+            is_superuser=False,
+        )
         self.compute = Compute.objects.create(
             name="api-test-compute",
             hostname="127.0.0.1",
@@ -30,6 +37,13 @@ class APISecurityTestCase(TestCase):
             password="secret_hypervisor_password",
             type=1,
         )
+        self.empty_compute = Compute.objects.create(
+            name="empty-test-compute",
+            hostname="127.0.0.1",
+            login="libvirt_user",
+            password="secret_hypervisor_password",
+            type=1,
+        )
         self.flavor = Flavor.objects.create(
             label="test-flavor",
             vcpu=2,
@@ -60,11 +74,16 @@ class APISecurityTestCase(TestCase):
         res = self.client.get("/api/v1/computes/")
         self.assertIn(res.status_code, [302, 401, 403])
 
-        # Regular user - forbidden by IsAdminUser
+        # Regular user - forbidden by IsSuperUser
         self.client.force_login(self.regular_user)
         res = self.client.get("/api/v1/computes/")
         self.assertEqual(res.status_code, 403)
 
+        # Staff user (not superuser) - forbidden by IsSuperUser
+        self.client.force_login(self.staff_user)
+        res = self.client.get("/api/v1/computes/")
+        self.assertEqual(res.status_code, 403)
+
         # Admin user - allowed
         self.client.force_login(self.admin)
         res = self.client.get("/api/v1/computes/")
@@ -72,13 +91,22 @@ class APISecurityTestCase(TestCase):
 
     @patch("computes.utils.refresh_instance_database")
     def test_compute_instance_list_isolation(self, mock_refresh):
-        # Regular user should only see instance1
+        # Regular user should only see instance1 on compute
         self.client.force_login(self.regular_user)
         res = self.client.get(f"/api/v1/computes/{self.compute.id}/instances/")
         self.assertEqual(res.status_code, 200)
         names = [item["name"] for item in res.data]
         self.assertIn("inst-user-owned", names)
         self.assertNotIn("inst-other-owned", names)
+        mock_refresh.assert_called_once_with(self.compute)
+
+        # Regular user accessing compute where they have NO instances:
+        # returns empty list without calling refresh_instance_database!
+        mock_refresh.reset_mock()
+        res_empty = self.client.get(f"/api/v1/computes/{self.empty_compute.id}/instances/")
+        self.assertEqual(res_empty.status_code, 200)
+        self.assertEqual(res_empty.data, [])
+        mock_refresh.assert_not_called()
 
         # Admin user should see both instances
         self.client.force_login(self.admin)
@@ -101,6 +129,14 @@ class APISecurityTestCase(TestCase):
         )
         self.assertEqual(res.status_code, 403)
 
+        # Staff user (not superuser) cannot mutate flavors
+        self.client.force_login(self.staff_user)
+        res = self.client.post(
+            "/api/v1/flavor/",
+            {"label": "staffmutate", "vcpu": 4, "memory": 4096, "disk": 40},
+        )
+        self.assertEqual(res.status_code, 403)
+
         res = self.client.delete(f"/api/v1/flavor/{self.flavor.id}/")
         self.assertEqual(res.status_code, 403)
 
@@ -109,17 +145,33 @@ class APISecurityTestCase(TestCase):
         res = self.client.post("/api/v1/migrate/", {})
         self.assertEqual(res.status_code, 403)
 
+        self.client.force_login(self.staff_user)
+        res = self.client.post("/api/v1/migrate/", {})
+        self.assertEqual(res.status_code, 403)
+
     def test_network_viewset_permissions(self):
         self.client.force_login(self.regular_user)
         res = self.client.get(f"/api/v1/computes/{self.compute.id}/networks/")
         self.assertEqual(res.status_code, 403)
 
+        self.client.force_login(self.staff_user)
+        res = self.client.get(f"/api/v1/computes/{self.compute.id}/networks/")
+        self.assertEqual(res.status_code, 403)
+
     def test_interface_viewset_permissions(self):
         self.client.force_login(self.regular_user)
         res = self.client.get(f"/api/v1/computes/{self.compute.id}/interfaces/")
         self.assertEqual(res.status_code, 403)
 
+        self.client.force_login(self.staff_user)
+        res = self.client.get(f"/api/v1/computes/{self.compute.id}/interfaces/")
+        self.assertEqual(res.status_code, 403)
+
     def test_storage_viewset_permissions(self):
         self.client.force_login(self.regular_user)
         res = self.client.get(f"/api/v1/computes/{self.compute.id}/storages/")
         self.assertEqual(res.status_code, 403)
+
+        self.client.force_login(self.staff_user)
+        res = self.client.get(f"/api/v1/computes/{self.compute.id}/storages/")
+        self.assertEqual(res.status_code, 403)

console/novncd

@@ -99,7 +99,7 @@ def get_connection_infos(token):
 
     signer = TimestampSigner(salt="console.novnc")
     try:
-        unsigned = signer.unsign(token, max_age=600)
+        unsigned = signer.unsign(token, max_age=120)
         parts = unsigned.split(":")
         if len(parts) >= 2:
             host = int(parts[0])
@@ -107,7 +107,7 @@ def get_connection_infos(token):
             if len(parts) >= 3:
                 user_id = int(parts[2])
     except (BadSignature, SignatureExpired, ValueError, TypeError):
-        if getattr(settings, "DEBUG", False) or "test" in sys.argv:
+        if "test" in sys.argv:
             try:
                 temptoken = token.split("-", 1)
                 host = int(temptoken[0])
@@ -126,6 +126,8 @@ 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")
@@ -151,9 +153,7 @@ def get_connection_infos(token):
         console_port = conn.get_console_port()
         console_socket = conn.get_console_socket()
     except Exception as e:
-        logging.error(
-            "Fail to retrieve console connection infos for token %s : %s" % (token, e)
-        )
+        logging.error("Fail to retrieve console connection infos: %s", e)
         raise
     return (
         connhost,

console/tests.py

@@ -83,7 +83,7 @@ class ConsoleViewsTestCase(TestCase):
         self.assertEqual(response.status_code, 200)
         self.assertTemplateUsed(response, "console-vnc-lite.html")
         signer = TimestampSigner(salt="console.novnc")
-        unsigned = signer.unsign(response.cookies["token"].value, max_age=600)
+        unsigned = signer.unsign(response.cookies["token"].value, max_age=120)
         self.assertEqual(
             unsigned, f"{self.compute.id}:{self.instance.uuid}:{self.admin_user.id}"
         )
@@ -521,3 +521,63 @@ class NovncdDaemonLogicTestCase(TestCase):
 
         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(
+            username="inactive_console_user",
+            password="password",
+            is_active=False,
+        )
+        compute = Compute.objects.create(
+            name="novnc-inactive-compute",
+            hostname="127.0.0.1",
+            login="root",
+            password="",
+            type=1,
+        )
+        instance = Instance.objects.create(
+            compute=compute,
+            name="novnc-inactive-vm",
+            uuid="44444444-4444-4444-4444-444444444444",
+        )
+        UserInstance.objects.create(instance=instance, user=inactive_user)
+
+        signer = TimestampSigner(salt="console.novnc")
+        token = signer.sign(f"{compute.id}:{instance.uuid}:{inactive_user.id}")
+
+        with self.assertRaises(PermissionError):
+            novncd_mod.get_connection_infos(token)
+
+    @patch("sys.argv", ["novncd"])
+    @override_settings(DEBUG=True)
+    def test_get_connection_infos_debug_true_does_not_bypass(self):
+        with self.assertRaises(PermissionError):
+            novncd_mod.get_connection_infos("invalid-token-in-debug")
+
+    @patch("sys.argv", ["novncd"])
+    def test_get_connection_infos_expired_token(self):
+        User = get_user_model()
+        user = User.objects.filter(is_superuser=True).first()
+        compute = Compute.objects.create(
+            name="novnc-expired-compute",
+            hostname="127.0.0.1",
+            login="root",
+            password="",
+            type=1,
+        )
+        instance = Instance.objects.create(
+            compute=compute,
+            name="novnc-expired-vm",
+            uuid="55555555-5555-5555-5555-555555555555",
+        )
+
+        import time
+        from unittest.mock import patch
+        # Sign token 300 seconds in the past (> 120s TTL)
+        with patch("time.time", return_value=time.time() - 300):
+            signer = TimestampSigner(salt="console.novnc")
+            expired_token = signer.sign(f"{compute.id}:{instance.uuid}:{user.id}")
+
+        with self.assertRaises(PermissionError):
+            novncd_mod.get_connection_infos(expired_token)

console/views.py

@@ -58,7 +58,7 @@ def console(request):
 
     if token:
         try:
-            unsigned = signer.unsign(token, max_age=600)
+            unsigned = signer.unsign(token, max_age=120)
             parts = unsigned.split(":")
             if len(parts) >= 2:
                 host = int(parts[0])

instances/api/viewsets.py

@@ -1,3 +1,4 @@
+from admin.permissions import IsSuperUser
 from appsettings.settings import app_settings
 from computes import utils
 from computes.models import Compute
@@ -70,11 +71,19 @@ class InstanceViewSet(viewsets.ViewSet):
     def list(self, request, compute_pk=None):
         compute = get_object_or_404(Compute, pk=compute_pk)
 
+        is_admin = request.user.is_superuser or request.user.has_perm(
+            "instances.view_instances"
+        )
+
+        if not is_admin:
+            if not Instance.objects.filter(
+                compute=compute, userinstance__user=request.user
+            ).exists():
+                return Response([])
+
         utils.refresh_instance_database(compute)
 
-        if request.user.is_superuser or request.user.has_perm(
-            "instances.view_instances"
-        ):
+        if is_admin:
             queryset = Instance.objects.filter(compute=compute).prefetch_related(
                 "userinstance_set"
             )
@@ -134,7 +143,7 @@ class MigrateViewSet(viewsets.ViewSet):
     A viewset for migrating instances.
     """
 
-    permission_classes = [permissions.IsAdminUser]
+    permission_classes = [IsSuperUser]
     serializer_class = MigrateSerializer
     queryset = ""
 
@@ -180,7 +189,7 @@ class FlavorViewSet(viewsets.ModelViewSet):
     def get_permissions(self):
         if self.action in ["list", "retrieve"]:
             return [permissions.IsAuthenticated()]
-        return [permissions.IsAdminUser()]
+        return [IsSuperUser()]
 
 
 class CreateInstanceViewSet(viewsets.ViewSet):
@@ -188,7 +197,7 @@ class CreateInstanceViewSet(viewsets.ViewSet):
     A viewset for creating instances.
     """
 
-    permission_classes = [permissions.IsAdminUser]
+    permission_classes = [IsSuperUser]
     serializer_class = CreateInstanceSerializer
     queryset = ""
 

interfaces/api/viewsets.py

@@ -1,6 +1,7 @@
+from admin.permissions import IsSuperUser
 from django.shortcuts import get_object_or_404
 from computes.models import Compute
-from rest_framework import permissions, status, viewsets
+from rest_framework import status, viewsets
 from rest_framework.response import Response
 from vrtManager.interface import wvmInterfaces, wvmInterface
 
@@ -12,7 +13,7 @@ class InterfaceViewSet(viewsets.ViewSet):
     A viewset for listing retrieving interfaces.
     """
 
-    permission_classes = [permissions.IsAdminUser]
+    permission_classes = [IsSuperUser]
     
     def list(self, request, compute_pk=None):
         queryset = []

networks/api/viewsets.py

@@ -1,6 +1,7 @@
+from admin.permissions import IsSuperUser
 from computes.models import Compute
 from django.shortcuts import get_object_or_404
-from rest_framework import permissions, viewsets
+from rest_framework import viewsets
 from rest_framework.response import Response
 from vrtManager.network import wvmNetworks
 
@@ -12,7 +13,7 @@ class NetworkViewSet(viewsets.ViewSet):
     A viewset for listing retrieving networks.
     """
 
-    permission_classes = [permissions.IsAdminUser]
+    permission_classes = [IsSuperUser]
 
     def list(self, request, compute_pk=None):
 

storages/api/viewsets.py

@@ -1,7 +1,8 @@
+from admin.permissions import IsSuperUser
 from appsettings.settings import app_settings
 from computes.models import Compute
 from django.shortcuts import get_object_or_404
-from rest_framework import permissions, status, viewsets
+from rest_framework import status, viewsets
 from rest_framework.decorators import action
 from rest_framework.response import Response
 from vrtManager.storage import wvmStorage, wvmStorages
@@ -14,7 +15,7 @@ class StorageViewSet(viewsets.ViewSet):
     A viewset for listing retrieving storages.
     """
 
-    permission_classes = [permissions.IsAdminUser]
+    permission_classes = [IsSuperUser]
 
     def list(self, request, compute_pk=None):
 
@@ -105,7 +106,7 @@ class VolumeViewSet(viewsets.ViewSet):
     A simple ViewSet for listing or retrieving Storage Volumes.
     """
 
-    permission_classes = [permissions.IsAdminUser]
+    permission_classes = [IsSuperUser]
     serializer_class = VolumeSerializer
     lookup_value_regex = "[^/]+"