Create async auth rules for user privileges

SDESK-7460

Author: eos87

pyproject.toml

@@ -20,5 +20,5 @@ exclude = '''
 [tool.pytest.ini_options]
 testpaths = ["tests", "superdesk", "apps", "content_api"]
 python_files = "*_test.py *_tests.py test_*.py tests_*.py tests.py test.py"
-asyncio_mode = "strict"
+asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "function"

superdesk/core/auth/privilege_rules.py

@@ -0,0 +1,62 @@
+from typing import Any, cast
+
+from quart_babel import gettext
+from superdesk.errors import SuperdeskApiError
+from superdesk.users.async_service import get_privileges, is_admin
+from superdesk.core.types import HTTP_METHOD, AuthRule, Request
+
+
+def required_privilege_rule(privilege: str) -> AuthRule:
+    """
+    Creates an authentication rule that restricts access based on a specific privilege.
+
+    Args:
+        privilege (str): The privilege required to access the resource.
+
+    Returns:
+        AuthRule: An asynchronous rule function that checks if the user has the required privilege.
+
+    The rule verifies:
+    - If the user is an admin, access is granted without further checks.
+    - Otherwise, the user's privileges are retrieved and checked against the required privilege.
+
+    Raises:
+        SuperdeskApiError: If the user does not have the required privilege.
+    """
+
+    async def internal_rule(request: Request):
+        current_user = request.user
+
+        # TODO-ASYNC: Check if for admin we need to pull all privileges instead
+        if is_admin(current_user):
+            return None
+
+        user_role = request.storage.request.get("role")
+        user_privileges = get_privileges(request.user, user_role)
+
+        if user_privileges.get(privilege, False):
+            return None
+
+        raise SuperdeskApiError.forbiddenError(message=gettext("Insufficient privileges for the requested operation."))
+
+    return internal_rule
+
+
+def privilege_based_rules(rules: dict[HTTP_METHOD, str]) -> dict[str, AuthRule]:
+    """
+    Generates a dictionary of authentication rules for HTTP methods based on privileges.
+
+    Args:
+        rules (dict[HTTP_METHOD, str]): A mapping of HTTP methods (e.g., GET, POST) to the required privileges
+        for those methods.
+
+    Returns:
+        dict[str, list[AuthRule]]: A dictionary where each HTTP method is mapped to a list of authentication
+        rules that restrict access based on the specified privilege.
+    """
+
+    auth_rules: dict[str, AuthRule] = {}
+    for method, privilege_name in rules.items():
+        auth_rules[method] = required_privilege_rule(privilege_name)
+
+    return auth_rules

superdesk/core/auth/token_auth.py

@@ -1,20 +1,35 @@
+import base64
+import arrow
+
 from typing import Any, cast
 from datetime import timedelta
 
-from superdesk.core import get_app_config
+from superdesk.utc import utcnow
+from superdesk.core import json, get_app_config
 from superdesk.core.types import Request
 from superdesk import get_resource_service
 from superdesk.errors import SuperdeskApiError
 from superdesk.resource_fields import LAST_UPDATED, ID_FIELD
-from superdesk.utc import utcnow
 
 from .user_auth import UserAuthProtocol
 
 
 class TokenAuthorization(UserAuthProtocol):
+    @staticmethod
+    def _decode_token(token: str) -> str:
+        """
+        Tries to decode the auth header token. Decode logic based on internal logic from
+        `werkzeug.datastructures.Authorization`
+        """
+        try:
+            return base64.b64decode(token).decode().partition(":")[0]
+        except Exception:
+            return token
+
     async def authenticate(self, request: Request):
         token = request.get_header("Authorization")
         new_session = True
+
         if token:
             token = token.strip()
             if token.lower().startswith(("token", "bearer")):
@@ -29,6 +44,10 @@ async def authenticate(self, request: Request):
 
         # Check provided token is valid
         auth_service = get_resource_service("auth")
+
+        # tokens are no longer decoded internally by flask/quart
+        # so we need to do it ourselves
+        token = self._decode_token(token)
         auth_token = auth_service.find_one(token=token, req=None)
 
         if not auth_token:
@@ -54,12 +73,15 @@ async def start_session(self, request: Request, user: dict[str, Any], **kwargs)
             await self.stop_session(request)
             raise SuperdeskApiError.unauthorizedError()
 
-        request.storage.session.set("session_token", auth_token)
+        request.storage.session.set("session_token", json.dumps(auth_token))
         await super().start_session(request, user, **kwargs)
 
     async def continue_session(self, request: Request, user: dict[str, Any], **kwargs) -> None:
         auth_token = request.storage.session.get("session_token")
 
+        if isinstance(auth_token, str):
+            auth_token = json.loads(auth_token)
+
         if not auth_token:
             await self.stop_session(request)
             raise SuperdeskApiError.unauthorizedError()
@@ -74,7 +96,9 @@ async def continue_session(self, request: Request, user: dict[str, Any], **kwarg
             now = utcnow()
             auth_updated = False
             session_update_seconds = cast(int, get_app_config("SESSION_UPDATE_SECONDS", 30))
-            if auth_token[LAST_UPDATED] + timedelta(seconds=session_update_seconds) < now:
+            auth_last_updated = arrow.get(auth_token[LAST_UPDATED])
+
+            if auth_last_updated + timedelta(seconds=session_update_seconds) < now:
                 auth_service = get_resource_service("auth")
                 auth_service.update_session({LAST_UPDATED: now})
                 auth_updated = True

superdesk/core/auth/user_auth.py

@@ -1,5 +1,6 @@
 from typing import Any, cast
 
+from superdesk.utils import flatten
 from superdesk.errors import SuperdeskApiError
 from superdesk.core.types import Request, AuthRule
 
@@ -27,7 +28,10 @@ async def authorize(self, request: Request) -> Any | None:
         elif isinstance(endpoint_rules, dict):
             endpoint_rules = cast(list[AuthRule], endpoint_rules.get(request.method) or [])
 
-        for rule in self.get_default_auth_rules() + (endpoint_rules or []):
+        auth_rules = self.get_default_auth_rules()
+        auth_rules.append(endpoint_rules or [])
+
+        for rule in flatten(auth_rules):
             response = await rule(request)
             if response is not None:
                 return response

superdesk/errors.py

@@ -130,7 +130,7 @@ def to_dict(self):
         """Create dict for json response."""
         rv = {}
         rv[STATUS] = STATUS_ERR
-        rv["_message"] = self.message or ""
+        rv["_message"] = str(self.message or "")
         if hasattr(self, "payload"):
             rv[ISSUES] = self.payload
         return rv

superdesk/tests/__init__.py

@@ -479,8 +479,12 @@ async def setup_db_user(context, user):
 
         user["password"] = original_password
         auth_data = json.dumps({"username": user["username"], "password": user["password"]})
+
+        context.headers.append(["Content-Type", "application/json"])
         auth_response = await context.client.post(
-            get_prefixed_url(context.app, "/auth_db"), data=auth_data, headers=context.headers
+            get_prefixed_url(context.app, "/auth_db"),
+            data=auth_data,
+            headers=context.headers,
         )
 
         auth_data = json.loads(await auth_response.get_data())

superdesk/utils.py

@@ -388,3 +388,19 @@ def format_content_type_name(item: Dict[str, str], default_name: str) -> str:
     """
     name = item.get("output_name") or item.get("label", default_name)
     return re.compile("[^0-9a-zA-Z_]").sub("", name)
+
+
+def flatten(nested_list) -> list:
+    """
+    Recursively flattens an arbitrarily nested list into a single flat list.
+
+    Returns:
+        list: A single flat list containing all elements from the nested structure.
+    """
+    result = []
+    for i in nested_list:
+        if isinstance(i, list):
+            result.extend(flatten(i))
+        else:
+            result.append(i)
+    return result

tests/core/modules/module_with_privileges.py

@@ -1,12 +1,27 @@
 from quart_babel import lazy_gettext
 
+from superdesk.core.auth.privilege_rules import privilege_based_rules
 from superdesk.core.module import Module
 from superdesk.core.privileges import Privilege
+from superdesk.core.resources import ResourceConfig, ResourceModel, RestEndpointConfig
+
+test_privileges = ["can_read", "can_create"]
+
+resource_config = ResourceConfig(
+    name="privileged_resource",
+    data_class=ResourceModel,
+    rest_endpoints=RestEndpointConfig(
+        resource_methods=["GET", "POST"],
+        auth=privilege_based_rules({"GET": test_privileges[0], "POST": test_privileges[1]}),
+    ),
+)
+
 
 module = Module(
     name="tests.module_with_privileges",
+    resources=[resource_config],
     privileges=[
-        Privilege(name="can_test", description=lazy_gettext("Test privilege can test")),
-        Privilege(name="can_play", description=lazy_gettext("Test privilege can play")),
+        Privilege(name="can_read", description=lazy_gettext("Test privilege can read")),
+        Privilege(name="can_create", description=lazy_gettext("Test privilege can create")),
     ],
 )

tests/core/privilege_rules_test.py

@@ -0,0 +1,73 @@
+import pytest
+from unittest.mock import patch, MagicMock
+from superdesk.errors import SuperdeskApiError
+from superdesk.core.auth.privilege_rules import privilege_based_rules, required_privilege_rule
+
+
+@patch("superdesk.core.auth.privilege_rules.is_admin")
+@patch("superdesk.core.auth.privilege_rules.get_privileges")
+async def test_required_privilege_rule_admin(mock_get_privileges, mock_is_admin):
+    mock_is_admin.return_value = True
+    mock_request = MagicMock()
+
+    rule = required_privilege_rule("test_privilege")
+    result = await rule(mock_request)
+
+    # admin user should pass without checking privileges
+    assert result is None
+    mock_get_privileges.assert_not_called()
+
+
+@patch("superdesk.core.auth.privilege_rules.is_admin")
+@patch("superdesk.core.auth.privilege_rules.get_privileges")
+async def test_required_privilege_rule_insufficient_privileges(mock_get_privileges, mock_is_admin):
+    mock_is_admin.return_value = False
+    mock_get_privileges.return_value = {"test_privilege": False}
+
+    mock_request = MagicMock()
+    mock_request.user = "test_user"
+    mock_request.storage.request.get.return_value = "user_role"
+
+    rule = required_privilege_rule("test_privilege")
+
+    with pytest.raises(SuperdeskApiError) as error:
+        await rule(mock_request)
+
+    assert error.value.status_code == 403
+    assert "Insufficient privileges" in str(error.value)
+    mock_get_privileges.assert_called_once_with("test_user", "user_role")
+
+
+@patch("superdesk.core.auth.privilege_rules.is_admin")
+@patch("superdesk.core.auth.privilege_rules.get_privileges")
+async def test_required_privilege_rule_sufficient_privileges(mock_get_privileges, mock_is_admin):
+    mock_is_admin.return_value = False
+    mock_get_privileges.return_value = {"test_privilege": True}
+
+    mock_request = MagicMock()
+    mock_request.user = "test_user"
+    mock_request.storage.request.get.return_value = "user_role"
+
+    rule = required_privilege_rule("test_privilege")
+
+    result = await rule(mock_request)
+    assert result is None
+    mock_get_privileges.assert_called_once_with("test_user", "user_role")
+
+
+def test_privilege_based_rules():
+    rules = {
+        "GET": "read_articles",
+        "POST": "create_articles",
+        "DELETE": "delete_articles",
+    }
+
+    auth_rules = privilege_based_rules(rules)
+
+    assert "GET" in auth_rules
+    assert "POST" in auth_rules
+    assert "DELETE" in auth_rules
+
+    assert len(auth_rules["GET"]) == 1
+    assert len(auth_rules["POST"]) == 1
+    assert len(auth_rules["DELETE"]) == 1