Fix access to restricted endpoints with anonymous token or no token

CHANGELOG.md

@@ -54,6 +54,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
     `view-stats` action on a cluster in clusters list page.
   - Update dependencies to fix CVE-2024-55565 (nanoid) and CVE-2025-24010
     (vite).
+- pkgs: Bump dependency to RFL.web 1.3.0 to fix access to restricted endpoints
+  with anonymous token or no token (#460,#462→#461).
 
 ## [4.0.0] - 2024-11-28
 

pyproject.toml

@@ -19,7 +19,7 @@ dependencies = [
     "RFL.core >= 1.1.0",
     "RFL.log",
     "RFL.settings >= 1.1.1",
-    "RFL.web",
+    "RFL.web >= 1.3.0",
     "setuptools"
 ]
 readme = "README.md"

slurmweb/tests/lib/agent.py

@@ -13,7 +13,8 @@
 import werkzeug
 from flask import Blueprint, jsonify
 
-from rfl.authentication.user import AuthenticatedUser
+from rfl.authentication.user import AuthenticatedUser, AnonymousUser
+from rfl.permissions.rbac import ANONYMOUS_ROLE
 from slurmweb.apps import SlurmwebConfSeed
 from slurmweb.apps.agent import SlurmwebAppAgent
 from racksdb.errors import RacksDBFormatError, RacksDBSchemaError
@@ -56,6 +57,9 @@ def setup_client(
         additional_conf=None,
         racksdb_format_error=False,
         racksdb_schema_error=False,
+        anonymous_user=False,
+        anonymous_enabled=True,
+        use_token=True,
     ):
         # Generate JWT signing key
         key = tempfile.NamedTemporaryFile(mode="w+")
@@ -102,6 +106,9 @@ def setup_client(
                     conf=conf.name,
                 )
             )
+        if not anonymous_enabled:
+            self.app.policy.disable_anonymous()
+
         conf.close()
         key.close()
         self.app.config.update(
@@ -112,12 +119,13 @@ def setup_client(
 
         # Get token valid to get user role with all permissions as defined in
         # default policy.
-        token = self.app.jwt.generate(
-            user=AuthenticatedUser(
+        if anonymous_user:
+            self.user = AnonymousUser()
+        else:
+            self.user = AuthenticatedUser(
                 login="test", fullname="Testing User", groups=["group"]
-            ),
-            duration=3600,
-        )
+            )
+
         # werkzeug.test.TestResponse class does not have text and json
         # properties in werkzeug <= 0.15. When such version is installed, use
         # custom test response class to backport these text and json properties.
@@ -127,7 +135,12 @@ def setup_client(
         except AttributeError:
             self.app.response_class = SlurmwebCustomTestResponse
         self.client = self.app.test_client()
-        self.client.environ_base["HTTP_AUTHORIZATION"] = "Bearer " + token
+        if use_token:
+            token = self.app.jwt.generate(
+                user=self.user,
+                duration=3600,
+            )
+            self.client.environ_base["HTTP_AUTHORIZATION"] = "Bearer " + token
 
     def mock_slurmrestd_responses(self, slurm_version, assets):
         return mock_slurmrestd_responses(self.app.slurmrestd, slurm_version, assets)
@@ -146,13 +159,13 @@ def __enter__(self):
         for _role in self.policy.loader.roles:
             if _role.name == self.role:
                 _role.actions.remove(self.action)
-            if _role.name == "anonymous" and self.action in _role.actions:
+            if _role.name == ANONYMOUS_ROLE and self.action in _role.actions:
                 _role.actions.remove(self.action)
                 self.removed_in_anonymous = True
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         for _role in self.policy.loader.roles:
             if _role.name == self.role:
                 _role.actions.add(self.action)
-            if _role.name == "anonymous" and self.removed_in_anonymous:
+            if _role.name == ANONYMOUS_ROLE and self.removed_in_anonymous:
                 _role.actions.add(self.action)

slurmweb/tests/lib/gateway.py

@@ -8,9 +8,13 @@
 import tempfile
 import os
 
+import werkzeug
+
+from rfl.authentication.user import AuthenticatedUser, AnonymousUser
 from slurmweb.apps import SlurmwebConfSeed
 from slurmweb.apps.gateway import SlurmwebAppGateway
 
+from .utils import SlurmwebCustomTestResponse
 
 CONF = """
 [agents]
@@ -23,7 +27,7 @@
 
 class TestGatewayBase(unittest.TestCase):
 
-    def setup_app(self):
+    def setup_app(self, anonymous_user=False, use_token=True):
         # Generate JWT signing key
         key = tempfile.NamedTemporaryFile(mode="w+")
         key.write("hey")
@@ -57,3 +61,29 @@ def setup_app(self):
                 "TESTING": True,
             }
         )
+
+        # Get token valid to get user role with all permissions as defined in
+        # default policy.
+        if anonymous_user:
+            self.user = AnonymousUser()
+        else:
+            self.user = AuthenticatedUser(
+                login="test", fullname="Testing User", groups=["group"]
+            )
+
+        # werkzeug.test.TestResponse class does not have text and json
+        # properties in werkzeug <= 0.15. When such version is installed, use
+        # custom test response class to backport these text and json properties.
+        try:
+            getattr(werkzeug.test.TestResponse, "text")
+            getattr(werkzeug.test.TestResponse, "json")
+        except AttributeError:
+            self.app.response_class = SlurmwebCustomTestResponse
+
+        self.client = self.app.test_client()
+        if use_token:
+            token = self.app.jwt.generate(
+                user=self.user,
+                duration=3600,
+            )
+            self.client.environ_base["HTTP_AUTHORIZATION"] = "Bearer " + token

slurmweb/tests/lib/utils.py

@@ -45,6 +45,17 @@ def inner(self, *args, **kwargs):
     return inner
 
 
+def any_slurm_version(test):
+    """Return test with first slurm version"""
+
+    def inner(self, *args, **kwargs):
+        for slurm_version in slurm_versions():
+            test(self, slurm_version, *args, **kwargs)
+            break
+
+    return inner
+
+
 def flask_version():
     """Return version of Flask package as a tuple of integers."""
     import importlib

slurmweb/tests/views/test_agent.py

@@ -54,15 +54,6 @@ def test_info(self):
         self.assertIn("metrics", response.json)
         self.assertIsInstance(response.json["metrics"], bool)
 
-    def test_permissions(self):
-        response = self.client.get(f"/v{get_version()}/permissions")
-        self.assertEqual(response.status_code, 200)
-        self.assertIsInstance(response.json, dict)
-        self.assertEqual(len(response.json.keys()), 2)
-        self.assertIn("actions", response.json)
-        self.assertIn("roles", response.json)
-        self.assertCountEqual(response.json["roles"], ["user"])
-
     #
     # General error cases
     #
@@ -129,37 +120,6 @@ def test_request_agent_not_found(self):
             },
         )
 
-    def test_access_denied(self):
-        # Test agent permission denied with @rbac_action decorator by calling /accounts
-        # without authentication token, ie. as anonymous who is denied to access this
-        # route in Slurm-web default authorization policy.
-        del self.client.environ_base["HTTP_AUTHORIZATION"]
-        response = self.client.get(f"/v{get_version()}/accounts")
-        self.assertEqual(response.status_code, 403)
-        self.assertEqual(
-            response.json,
-            {
-                "code": 403,
-                "description": (
-                    "Anonymous role is not allowed to perform action view-accounts"
-                ),
-                "name": "Forbidden",
-            },
-        )
-
-    def test_invalid_token(self):
-        self.client.environ_base["HTTP_AUTHORIZATION"] = "Bearer failed"
-        response = self.client.get(f"/v{get_version()}/jobs")
-        self.assertEqual(response.status_code, 401)
-        self.assertEqual(
-            response.json,
-            {
-                "code": 401,
-                "description": "Unable to decode token: Not enough segments",
-                "name": "Unauthorized",
-            },
-        )
-
     #
     # slurmrestd ressources
     #

slurmweb/tests/views/test_agent_permissions.py

@@ -0,0 +1,151 @@
+# Copyright (c) 2025 Rackslab
+#
+# This file is part of Slurm-web.
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+from slurmweb.version import get_version
+
+from ..lib.agent import TestAgentBase
+from ..lib.utils import any_slurm_version
+
+
+class TestAgentPermissions(TestAgentBase):
+
+    def test_permissions_user(self):
+        self.setup_client()
+        response = self.client.get(f"/v{get_version()}/permissions")
+        self.assertEqual(response.status_code, 200)
+        self.assertIsInstance(response.json, dict)
+        self.assertEqual(len(response.json.keys()), 2)
+        self.assertIn("actions", response.json)
+        self.assertIn("roles", response.json)
+        self.assertCountEqual(response.json["roles"], ["user"])
+        self.assertCountEqual(
+            response.json["actions"], self.app.policy.roles_actions(self.user)[1]
+        )
+
+    def test_permissions_anonymous(self):
+        self.setup_client(anonymous_user=True)
+        self.assertTrue(self.app.policy.allow_anonymous)
+        response = self.client.get(f"/v{get_version()}/permissions")
+        self.assertEqual(response.status_code, 200)
+        self.assertIsInstance(response.json, dict)
+        self.assertEqual(len(response.json.keys()), 2)
+        self.assertIn("actions", response.json)
+        self.assertIn("roles", response.json)
+        # anonymous user should get the anonymous role and corresponding set of actions
+        # when anonymous mode is enabled in policy.
+        self.assertCountEqual(response.json["roles"], ["anonymous"])
+        self.assertCountEqual(
+            response.json["actions"], self.app.policy.roles_actions(self.user)[1]
+        )
+
+    def test_permissions_anonymous_disabled(self):
+        self.setup_client(anonymous_user=True, anonymous_enabled=False)
+        self.assertFalse(self.app.policy.allow_anonymous)
+        response = self.client.get(f"/v{get_version()}/permissions")
+        self.assertEqual(response.status_code, 200)
+        self.assertIsInstance(response.json, dict)
+        self.assertEqual(len(response.json.keys()), 2)
+        self.assertIn("actions", response.json)
+        self.assertIn("roles", response.json)
+        # anonymous user should get no role or action when anonymous mode is disabled in
+        # policy.
+        self.assertCountEqual(response.json["roles"], [])
+        self.assertCountEqual(response.json["actions"], [])
+
+    def test_permissions_no_token(self):
+        # permissions endpoint is guarded by @check_jwt decorator that must reply 403
+        # to requests without bearer token.
+        self.setup_client(use_token=False)
+        self.assertTrue(self.app.policy.allow_anonymous)
+        response = self.client.get(f"/v{get_version()}/permissions")
+        self.assertEqual(response.status_code, 403)
+        self.assertEqual(
+            response.json,
+            {
+                "code": 403,
+                "description": "Not allowed to access endpoint without bearer token",
+                "name": "Forbidden",
+            },
+        )
+
+    @any_slurm_version
+    def test_action_anonymous_ok(self, slurm_version):
+        # stats endpoint is authorized to anonymous role in default policy.
+        self.setup_client(anonymous_user=True)
+        [ping_asset, jobs_asset, nodes_asset] = self.mock_slurmrestd_responses(
+            slurm_version,
+            [
+                ("slurm-ping", "meta"),
+                ("slurm-jobs", "jobs"),
+                ("slurm-nodes", "nodes"),
+            ],
+        )
+        response = self.client.get(f"/v{get_version()}/stats")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("jobs", response.json)
+
+    def test_action_anonymous_disabled(self):
+        # stats endpoint must be denied to anonymous tokens when anonymous role is
+        # disabled in policy.
+        self.setup_client(anonymous_enabled=False, anonymous_user=True)
+        response = self.client.get(f"/v{get_version()}/stats")
+        self.assertEqual(response.status_code, 403)
+        self.assertEqual(
+            response.json,
+            {
+                "code": 403,
+                "description": (
+                    "Anonymous role is not allowed to perform action view-stats"
+                ),
+                "name": "Forbidden",
+            },
+        )
+
+    def test_action_no_token_denied(self):
+        # stats endpoint must be denied to requests without bearer token.
+        self.setup_client(use_token=False)
+        response = self.client.get(f"/v{get_version()}/stats")
+        self.assertEqual(response.status_code, 403)
+        self.assertEqual(
+            response.json,
+            {
+                "code": 403,
+                "description": ("Not allowed to access endpoint without bearer token"),
+                "name": "Forbidden",
+            },
+        )
+
+    def test_action_anonymous_denied(self):
+        # Test agent permission denied with @rbac_action decorator by calling /accounts
+        # without authentication token, ie. as anonymous who is denied to access this
+        # route in Slurm-web default authorization policy.
+        self.setup_client(anonymous_user=True)
+        response = self.client.get(f"/v{get_version()}/accounts")
+        self.assertEqual(response.status_code, 403)
+        self.assertEqual(
+            response.json,
+            {
+                "code": 403,
+                "description": (
+                    "Anonymous role is not allowed to perform action view-accounts"
+                ),
+                "name": "Forbidden",
+            },
+        )
+
+    def test_invalid_token(self):
+        self.setup_client()
+        self.client.environ_base["HTTP_AUTHORIZATION"] = "Bearer failed"
+        response = self.client.get(f"/v{get_version()}/jobs")
+        self.assertEqual(response.status_code, 401)
+        self.assertEqual(
+            response.json,
+            {
+                "code": 401,
+                "description": "Unable to decode token: Not enough segments",
+                "name": "Unauthorized",
+            },
+        )