moves logic to policy

Author: pcrespov

services/web/server/src/simcore_service_webserver/security/_authz_policy.py

@@ -10,19 +10,22 @@
 from aiohttp_security.abc import (  # type: ignore[import-untyped]
     AbstractAuthorizationPolicy,
 )
+from common_library.users_enums import UserRole
 from models_library.products import ProductName
 from models_library.users import UserID
 from servicelib.aiohttp.db_asyncpg_engine import get_async_engine
 from servicelib.logging_errors import create_troubleshootting_log_kwargs
 from simcore_postgres_database.aiopg_errors import DatabaseError as AiopgDatabaseError
 from sqlalchemy.exc import DatabaseError as SQLAlchemyDatabaseError
 
+from ..groups import api as groups_service
 from ._authz_access_model import (
     AuthContextDict,
     OptionalContext,
     RoleBasedAccessModel,
     has_access_by_role,
 )
+from ._authz_access_roles import NAMED_GROUP_PERMISSIONS
 from ._authz_repository import (
     ActiveUserIdAndRole,
     get_active_user_or_none,
@@ -161,17 +164,39 @@ async def permits(
             "authorized_uid"
         ), f"{user_id}!={context.get('authorized_uid')}"
 
-        # product access
+        # PRODUCT access
         if permission == PERMISSION_PRODUCT_LOGIN_KEY:
             ok: bool = product_name is not None and await self._has_access_to_product(
                 user_id=user_id, product_name=product_name
             )
             return ok
 
-        # role-based access
-        return await has_access_by_role(
+        # ROLE-BASED access policy
+        role_allowed = await has_access_by_role(
             self._access_model,
             role=user_role,
             operation=permission,
             context=context,
         )
+
+        if role_allowed:
+            return True
+
+        # GROUP-BASED access policy (only if enabled in context and user is above GUEST role)
+        product_support_group_id = context.get("product_support_group_id", None)
+        if (
+            product_support_group_id is not None
+            and user_role > UserRole.GUEST
+            and permission in NAMED_GROUP_PERMISSIONS.get("PRODUCT_SUPPORT_GROUP", [])
+        ):
+            with contextlib.suppress(
+                Exception
+                # If product or group check fails, continue to deny access
+                # NOTE: Logging omitted to avoid exposing internal errors
+            ):
+                if await groups_service.is_user_in_group(
+                    self._app, user_id=user_id, group_id=product_support_group_id
+                ):
+                    return True
+
+        return False

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

@@ -1,13 +1,10 @@
-from contextlib import suppress
 from functools import wraps
 
 import aiohttp_security.api  # type: ignore[import-untyped]
 from aiohttp import web
 from servicelib.aiohttp.typing_extension import Handler
+from simcore_service_webserver.products import products_web
 
-from ..products import products_web
-from ._authz_access_model import AuthContextDict
-from ._authz_access_roles import NAMED_GROUP_PERMISSIONS
 from ._authz_web import check_user_authorized
 from .security_web import check_user_permission
 
@@ -46,31 +43,17 @@ def group_or_role_permission_required(permission: str):
     def _decorator(handler: Handler):
         @wraps(handler)
         async def _wrapped(request: web.Request):
-            context = AuthContextDict()
-            context["authorized_uid"] = user_id = await check_user_authorized(request)
-
-            # Check ROLE-BASED permissions first
-            role_allowed = await aiohttp_security.api.permits(
-                request, permission, context
-            )
-            if role_allowed:
+            context = {
+                "authorized_uid": await check_user_authorized(request),
+                "product_support_group_id": products_web.get_current_product(
+                    request
+                ).support_standard_group_id,
+            }
+
+            # Check both role-based and group-based permissions
+            if await aiohttp_security.api.permits(request, permission, context):
                 return await handler(request)
 
-            # Check GROUP-BASED permissions
-            with suppress(
-                Exception
-                # If product or group check fails, continue to deny access
-                # NOTE: Logging omitted to avoid exposing internal errors
-            ):
-                # FIXME: must be >GUEST !!
-                if permission in NAMED_GROUP_PERMISSIONS.get(
-                    "PRODUCT_SUPPORT_GROUP", []
-                ) and await products_web.is_user_in_product_support_group(
-                    request,
-                    user_id=user_id,
-                ):
-                    return await handler(request)
-
             # Neither role nor group permissions granted
             msg = f"You do not have sufficient access rights for {permission}"
             raise web.HTTPForbidden(text=msg)