✨ Refactor authentication decorators: move login_required to login_auth and clean up unused code

Author: pcrespov

services/web/server/src/simcore_service_webserver/login/_controller/rest/auth.py

@@ -287,20 +287,3 @@ async def logout(request: web.Request) -> web.Response:
         await security_web.forget_identity(request, response)
 
         return response
-
-
-@routes.get(f"/{API_VTAG}/auth:check", name="check_auth")
-@login_required
-async def check_auth(request: web.Request) -> web.Response:
-    """Lightweight endpoint for checking if users are authenticated & authorized to this product
-
-    Used primarily by Traefik auth middleware to verify session cookies
-    SEE https://doc.traefik.io/traefik/middlewares/http/forwardauth
-    """
-    # NOTE: for future development
-    # if database access is added here, services like jupyter-math
-    # which load a lot of resources will have a big performance hit
-    # consider caching some properties required by this endpoint or rely on Redis
-    assert request  # nosec
-
-    return web.json_response(status=status.HTTP_204_NO_CONTENT)

services/web/server/src/simcore_service_webserver/login/decorators.py

@@ -1,73 +1,7 @@
-import functools
-import inspect
-from typing import cast
+from ..login_auth.decorators import get_user_id, login_required
 
-from aiohttp import web
-from models_library.users import UserID
-from servicelib.aiohttp.typing_extension import HandlerAnyReturn
-from servicelib.request_keys import RQT_USERID_KEY
-
-from ..products import products_web
-from ..security import security_web
-
-
-def login_required(handler: HandlerAnyReturn) -> HandlerAnyReturn:
-    """Decorator that restrict access only for authorized users with permissions to access a given product
-
-    - User is considered authorized if check_authorized(request) raises no exception
-    - If authorized, it injects user_id in request[RQT_USERID_KEY]
-    - Use this decorator instead of aiohttp_security.api.login_required!
-
-    WARNING: Add always @router. decorator FIRST, e.g.
-
-        @router.get("/foo")
-        @login_required
-        async def get_foo(request: web.Request):
-            ...
-
-    and NOT as
-
-        @login_required
-        @router.get("/foo")
-        async def get_foo(request: web.Request):
-            ...
-
-    since the latter will register in `router` get_foo **without** `login_required`
-    """
-    assert set(inspect.signature(handler).parameters.values()) == {  # nosec
-        inspect.Parameter(
-            name="request",
-            kind=inspect.Parameter.POSITIONAL_OR_KEYWORD,
-            annotation=web.Request,
-        )
-    }, f"Expected {handler.__name__} with request as signature, got {handler.__annotations__}"
-
-    @functools.wraps(handler)
-    async def _wrapper(request: web.Request):
-        """
-        Raises:
-            HTTPUnauthorized: if unauthorized user
-            HTTPForbidden: if user not allowed in product
-        """
-        # WARNING: note that check_authorized is patched in some tests.
-        # Careful when changing the function signature
-        user_id = await security_web.check_user_authorized(request)
-        product_name = products_web.get_product_name(request)
-
-        await security_web.check_user_permission(
-            request,
-            security_web.PERMISSION_PRODUCT_LOGIN_KEY,
-            context=security_web.AuthContextDict(
-                product_name=product_name,
-                authorized_uid=user_id,
-            ),
-        )
-
-        request[RQT_USERID_KEY] = user_id
-        return await handler(request)
-
-    return _wrapper
-
-
-def get_user_id(request: web.Request) -> UserID:
-    return cast(UserID, request[RQT_USERID_KEY])
+__all__: tuple[str, ...] = (
+    "get_user_id",
+    "login_required",
+)
+# nopycln: file

services/web/server/src/simcore_service_webserver/login_auth/_controller_rest.py

@@ -5,7 +5,7 @@
 from servicelib.aiohttp import status
 
 from .._meta import API_VTAG
-from ..login.decorators import login_required  # FIXME: move this here
+from .decorators import login_required
 
 _logger = logging.getLogger(__name__)
 

services/web/server/src/simcore_service_webserver/login_auth/decorators.py

@@ -0,0 +1,73 @@
+import functools
+import inspect
+from typing import cast
+
+from aiohttp import web
+from models_library.users import UserID
+from servicelib.aiohttp.typing_extension import HandlerAnyReturn
+from servicelib.request_keys import RQT_USERID_KEY
+
+from ..products import products_web
+from ..security import security_web
+
+
+def login_required(handler: HandlerAnyReturn) -> HandlerAnyReturn:
+    """Decorator that restrict access only for authorized users with permissions to access a given product
+
+    - User is considered authorized if check_authorized(request) raises no exception
+    - If authorized, it injects user_id in request[RQT_USERID_KEY]
+    - Use this decorator instead of aiohttp_security.api.login_required!
+
+    WARNING: Add always @router. decorator FIRST, e.g.
+
+        @router.get("/foo")
+        @login_required
+        async def get_foo(request: web.Request):
+            ...
+
+    and NOT as
+
+        @login_required
+        @router.get("/foo")
+        async def get_foo(request: web.Request):
+            ...
+
+    since the latter will register in `router` get_foo **without** `login_required`
+    """
+    assert set(inspect.signature(handler).parameters.values()) == {  # nosec
+        inspect.Parameter(
+            name="request",
+            kind=inspect.Parameter.POSITIONAL_OR_KEYWORD,
+            annotation=web.Request,
+        )
+    }, f"Expected {handler.__name__} with request as signature, got {handler.__annotations__}"
+
+    @functools.wraps(handler)
+    async def _wrapper(request: web.Request):
+        """
+        Raises:
+            HTTPUnauthorized: if unauthorized user
+            HTTPForbidden: if user not allowed in product
+        """
+        # WARNING: note that check_authorized is patched in some tests.
+        # Careful when changing the function signature
+        user_id = await security_web.check_user_authorized(request)
+        product_name = products_web.get_product_name(request)
+
+        await security_web.check_user_permission(
+            request,
+            security_web.PERMISSION_PRODUCT_LOGIN_KEY,
+            context=security_web.AuthContextDict(
+                product_name=product_name,
+                authorized_uid=user_id,
+            ),
+        )
+
+        request[RQT_USERID_KEY] = user_id
+        return await handler(request)
+
+    return _wrapper
+
+
+def get_user_id(request: web.Request) -> UserID:
+    return cast(UserID, request[RQT_USERID_KEY])