feat: Add dashboard admin feature

Author: KShivendu

supertokens_python/framework/request.py

@@ -25,6 +25,10 @@ def __init__(self):
         self.wrapper_used = True
         self.request = None
 
+    @abstractmethod
+    def get_original_url(self) -> str:
+        pass
+
     @abstractmethod
     def get_query_param(
         self, key: str, default: Union[str, None] = None

supertokens_python/recipe/dashboard/__init__.py

@@ -14,7 +14,7 @@
 
 from __future__ import annotations
 
-from typing import Callable, Optional, Union
+from typing import Callable, Optional, List
 
 from supertokens_python import AppInfo, RecipeModule
 
@@ -26,10 +26,12 @@
 
 
 def init(
-    api_key: Union[str, None] = None,
+    api_key: Optional[str] = None,
+    admins: Optional[List[str]] = None,
     override: Optional[InputOverrideConfig] = None,
 ) -> Callable[[AppInfo], RecipeModule]:
     return DashboardRecipe.init(
         api_key,
+        admins,
         override,
     )

supertokens_python/recipe/dashboard/api/api_key_protector.py

@@ -13,6 +13,7 @@
 # under the License.
 from __future__ import annotations
 
+import json
 from typing import TYPE_CHECKING, Callable, Optional, Awaitable, Dict, Any
 
 from supertokens_python.framework import BaseResponse
@@ -29,6 +30,8 @@
     send_non_200_response_with_message,
 )
 
+from ..exceptions import DashboardOperationNotAllowedError
+
 
 async def api_key_protector(
     api_implementation: APIInterface,
@@ -39,9 +42,25 @@ async def api_key_protector(
     ],
     user_context: Dict[str, Any],
 ) -> Optional[BaseResponse]:
-    should_allow_access = await api_options.recipe_implementation.should_allow_access(
-        api_options.request, api_options.config, user_context
-    )
+    should_allow_access = False
+
+    try:
+        should_allow_access = (
+            await api_options.recipe_implementation.should_allow_access(
+                api_options.request, api_options.config, user_context
+            )
+        )
+    except DashboardOperationNotAllowedError as _:
+        # api_options.response.set_status_code(403)
+        # api_options.response.set_json_content({
+        #     "message": "You are not permitted to perform this operation"
+        # })
+        # return None
+        return send_non_200_response_with_message(
+            json.dumps({"message": "You are not permitted to perform this operation"}),
+            403,
+            api_options.response,
+        )
 
     if should_allow_access is False:
         return send_non_200_response_with_message(

supertokens_python/recipe/dashboard/exceptions.py

@@ -3,3 +3,7 @@
 
 class SuperTokensDashboardError(SuperTokensError):
     pass
+
+
+class DashboardOperationNotAllowedError(SuperTokensDashboardError):
+    pass

supertokens_python/recipe/dashboard/recipe.py

@@ -87,12 +87,14 @@ def __init__(
         self,
         recipe_id: str,
         app_info: AppInfo,
-        api_key: Union[str, None],
-        override: Union[InputOverrideConfig, None] = None,
+        api_key: Optional[str],
+        admins: Optional[List[str]],
+        override: Optional[InputOverrideConfig] = None,
     ):
         super().__init__(recipe_id, app_info)
         self.config = validate_and_normalise_user_input(
             api_key,
+            admins,
             override,
         )
         recipe_implementation = RecipeImplementation()
@@ -349,15 +351,17 @@ def get_all_cors_headers(self) -> List[str]:
 
     @staticmethod
     def init(
-        api_key: Union[str, None],
-        override: Union[InputOverrideConfig, None] = None,
+        api_key: Optional[str],
+        admins: Optional[List[str]],
+        override: Optional[InputOverrideConfig] = None,
     ):
         def func(app_info: AppInfo):
             if DashboardRecipe.__instance is None:
                 DashboardRecipe.__instance = DashboardRecipe(
                     DashboardRecipe.recipe_id,
                     app_info,
                     api_key,
+                    admins,
                     override,
                 )
                 return DashboardRecipe.__instance

supertokens_python/recipe/dashboard/recipe_implementation.py

@@ -18,10 +18,16 @@
 from supertokens_python.constants import DASHBOARD_VERSION
 from supertokens_python.framework import BaseRequest
 from supertokens_python.normalised_url_path import NormalisedURLPath
+from supertokens_python.utils import log_debug_message
 from supertokens_python.querier import Querier
+from supertokens_python.recipe.dashboard.constants import (
+    DASHBOARD_ANALYTICS_API,
+    EMAIL_PASSSWORD_SIGNOUT,
+)
 
 from .interfaces import RecipeInterface
 from .utils import DashboardConfig, validate_api_key
+from .exceptions import DashboardOperationNotAllowedError
 
 
 class RecipeImplementation(RecipeInterface):
@@ -34,9 +40,9 @@ async def should_allow_access(
         config: DashboardConfig,
         user_context: Dict[str, Any],
     ) -> bool:
-        if config.auth_mode == "email-password":
+        # For cases where we're not using the API key, the JWT is being used; we allow their access by default
+        if config.api_key is not None:
             auth_header_value = request.get_header("authorization")
-
             if not auth_header_value:
                 return False
 
@@ -47,8 +53,40 @@ async def should_allow_access(
                     {"sessionId": auth_header_value},
                 )
             )
-            return (
-                "status" in session_verification_response
-                and session_verification_response["status"] == "OK"
-            )
+            if session_verification_response.get("status") != "OK":
+                return False
+
+            # For all non GET requests we also want to check if the
+            # user is allowed to perform this operation
+            if request.method() != "GET":  # TODO: Use normalize http method?
+                # We dont want to block the analytics API
+                if request.get_original_url().startswith(DASHBOARD_ANALYTICS_API):
+                    return True
+
+                # We do not want to block the sign out request
+                if request.get_original_url().endswith(EMAIL_PASSSWORD_SIGNOUT):
+                    return True
+
+                admins = config.admins
+
+                # If the user has provided no admins, allow
+                if len(admins) == 0:
+                    return True
+
+                email_in_headers = request.get_header("email")
+
+                if email_in_headers is None:
+                    log_debug_message(
+                        "User Dashboard: Returniing OPERATION_NOT_ALLOWED because no email was provided in headers"
+                    )
+                    return False
+
+                if email_in_headers not in admins:
+                    log_debug_message(
+                        "User Dashboard: Throwing OPERATION_NOT_ALLOWED because user is not an admin"
+                    )
+                    raise DashboardOperationNotAllowedError()
+
+            return True
+
         return validate_api_key(request, config, user_context)

supertokens_python/recipe/dashboard/utils.py

@@ -43,7 +43,7 @@
     get_user_by_id as tppless_get_user_by_id,
 )
 from supertokens_python.types import User
-from supertokens_python.utils import Awaitable
+from supertokens_python.utils import Awaitable, log_debug_message, normalise_email
 
 from ...normalised_url_path import NormalisedURLPath
 from .constants import (
@@ -181,24 +181,38 @@ def __init__(
 
 class DashboardConfig:
     def __init__(
-        self, api_key: Union[str, None], override: OverrideConfig, auth_mode: str
+        self,
+        api_key: Optional[str],
+        admins: List[str],
+        override: OverrideConfig,
+        auth_mode: str,
     ):
         self.api_key = api_key
+        self.admins = admins
         self.override = override
         self.auth_mode = auth_mode
 
 
 def validate_and_normalise_user_input(
     # app_info: AppInfo,
     api_key: Union[str, None],
+    admins: Optional[List[str]],
     override: Optional[InputOverrideConfig] = None,
 ) -> DashboardConfig:
 
     if override is None:
         override = InputOverrideConfig()
 
+    if api_key is not None and admins is not None:
+        log_debug_message(
+            "User Dashboard: Providing 'admins' has no effect when using an api key."
+        )
+
+    admins = [normalise_email(a) for a in admins] if admins is not None else []
+
     return DashboardConfig(
         api_key,
+        admins,
         OverrideConfig(
             functions=override.functions,
             apis=override.apis,

supertokens_python/utils.py

@@ -360,3 +360,7 @@ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any):
 
         if exc_type is not None:
             raise exc_type(exc_value).with_traceback(traceback)
+
+
+def normalise_email(email: str) -> str:
+    return email.strip().lower()