refactor: rename AuthInfoDict to ActiveUserIdAndRole and update related methods

Author: pcrespov

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

@@ -25,7 +25,7 @@
     has_access_by_role,
 )
 from ._authz_repository import (
-    AuthInfoDict,
+    ActiveUserIdAndRole,
     get_active_user_or_none,
     is_user_in_product_name,
 )
@@ -72,7 +72,9 @@ def __init__(self, app: web.Application, access_model: RoleBasedAccessModel):
         namespace=__name__,
         key_builder=lambda f, *ag, **kw: f"{f.__name__}/{kw['email']}",
     )
-    async def _get_auth_or_none(self, *, email: str) -> AuthInfoDict | None:
+    async def _get_authorized_user_or_none(
+        self, *, email: str
+    ) -> ActiveUserIdAndRole | None:
         """
         Raises:
             web.HTTPServiceUnavailable: if database raises an exception
@@ -105,7 +107,7 @@ def access_model(self) -> RoleBasedAccessModel:
 
     async def clear_cache(self):
         # pylint: disable=no-member
-        for fun in (self._get_auth_or_none, self._has_access_to_product):
+        for fun in (self._get_authorized_user_or_none, self._has_access_to_product):
             autz_cache: BaseCache = fun.cache
             await autz_cache.clear()
 
@@ -119,7 +121,9 @@ async def authorized_userid(self, identity: IdentityStr) -> int | None:
         Return the user_id of the user identified by the identity
         or "None" if no user exists related to the identity.
         """
-        user_info: AuthInfoDict | None = await self._get_auth_or_none(email=identity)
+        user_info: ActiveUserIdAndRole | None = await self._get_authorized_user_or_none(
+            email=identity
+        )
         if user_info is None:
             return None
 
@@ -142,25 +146,33 @@ async def permits(
         if identity is None or permission is None:
             return False
 
-        auth_info = await self._get_auth_or_none(email=identity)
-        if auth_info is None:
+        # authorized user info
+        authorized_user_info = await self._get_authorized_user_or_none(email=identity)
+        if authorized_user_info is None:
             return False
 
-        # make sure context is a copy to avoid side effects
-        context = deepcopy(context) if context else AuthContextDict()
+        user_id = authorized_user_info["id"]
+        user_role = authorized_user_info["role"]
+
+        # context info: product_name
+        context = context or AuthContextDict()
+        product_name = context.get("product_name")
+
+        assert user_id == context.get(  # nosec
+            "authorized_uid"
+        ), f"{user_id}!={context.get('authorized_uid')}"
 
         # product access
         if permission == PERMISSION_PRODUCT_LOGIN_KEY:
-            product_name = context.get("product_name")
             ok: bool = product_name is not None and await self._has_access_to_product(
-                user_id=auth_info["id"], product_name=product_name
+                user_id=user_id, product_name=product_name
             )
             return ok
 
         # role-based access
         return await has_access_by_role(
             self._access_model,
-            role=auth_info["role"],
+            role=user_role,
             operations=permission,
             context=context,
         )

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

@@ -16,14 +16,14 @@
 _logger = logging.getLogger(__name__)
 
 
-class AuthInfoDict(TypedDict, total=True):
+class ActiveUserIdAndRole(TypedDict, total=True):
     id: IdInt
     role: UserRole
 
 
 async def get_active_user_or_none(
     engine: AsyncEngine, *, email: str
-) -> AuthInfoDict | None:
+) -> ActiveUserIdAndRole | None:
     """Gets a user with email if ACTIVE othewise return None
 
     Raises:
@@ -46,7 +46,7 @@ async def get_active_user_or_none(
             row is None or TypeAdapter(UserRole).validate_python(row.role) is not None
         )
 
-        return AuthInfoDict(id=row.id, role=row.role) if row else None
+        return ActiveUserIdAndRole(id=row.id, role=row.role) if row else None
 
 
 async def is_user_in_product_name(

services/web/server/tests/unit/isolated/test_security__authz.py

@@ -29,7 +29,7 @@
     UserRole,
 )
 from simcore_service_webserver.security._authz_policy import AuthorizationPolicy
-from simcore_service_webserver.security._authz_repository import AuthInfoDict
+from simcore_service_webserver.security._authz_repository import ActiveUserIdAndRole
 
 
 @pytest.fixture
@@ -248,9 +248,9 @@ def mock_db(mocker: MockerFixture) -> MagicMock:
         return_value="FAKE-ENGINE",
     )
 
-    users_db: dict[str, AuthInfoDict] = {
-        "[email protected]": AuthInfoDict(id=1, role=UserRole.GUEST),
-        "[email protected]": AuthInfoDict(id=55, role=UserRole.GUEST),
+    users_db: dict[str, ActiveUserIdAndRole] = {
+        "[email protected]": ActiveUserIdAndRole(id=1, role=UserRole.GUEST),
+        "[email protected]": ActiveUserIdAndRole(id=55, role=UserRole.GUEST),
     }
 
     async def _fake_db(engine, email):
@@ -280,11 +280,11 @@ async def test_authorization_policy_cache(mocker: MockerFixture, mock_db: MagicM
     # cache under test
 
     # pylint: disable=no-member
-    autz_cache: BaseCache = authz_policy._get_auth_or_none.cache
+    autz_cache: BaseCache = authz_policy._get_authorized_user_or_none.cache
 
     assert not (await autz_cache.exists("_get_auth_or_none/[email protected]"))
     for _ in range(3):
-        got = await authz_policy._get_auth_or_none(email="[email protected]")
+        got = await authz_policy._get_authorized_user_or_none(email="[email protected]")
         assert mock_db.call_count == 1
         assert got["id"] == 1
 
@@ -295,15 +295,15 @@ async def test_authorization_policy_cache(mocker: MockerFixture, mock_db: MagicM
     assert (await autz_cache.get("_get_auth_or_none/[email protected]"))["id"] == 1
 
     # gets cache, db is NOT called
-    got = await authz_policy._get_auth_or_none(email="[email protected]")
+    got = await authz_policy._get_authorized_user_or_none(email="[email protected]")
     assert mock_db.call_count == 1
     assert got["id"] == 1
 
     # clear cache
     await authz_policy.clear_cache()
 
     # gets new value
-    got = await authz_policy._get_auth_or_none(email="[email protected]")
+    got = await authz_policy._get_authorized_user_or_none(email="[email protected]")
     assert mock_db.call_count == 2
     assert got["id"] == 2
 
@@ -312,10 +312,10 @@ async def test_authorization_policy_cache(mocker: MockerFixture, mock_db: MagicM
 
     for _ in range(4):
         # NOTE: None
-        assert await authz_policy._get_auth_or_none(email="[email protected]")
+        assert await authz_policy._get_authorized_user_or_none(email="[email protected]")
         assert await autz_cache.exists("_get_auth_or_none/[email protected]")
         assert mock_db.call_count == 3
 
     # should raise web.HTTPServiceUnavailable on db failure
     with pytest.raises(web.HTTPServiceUnavailable):
-        await authz_policy._get_auth_or_none(email="[email protected]")
+        await authz_policy._get_authorized_user_or_none(email="[email protected]")