extends coverage and rm unsued functionality

Author: pcrespov

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

@@ -7,7 +7,6 @@
 
 import inspect
 import logging
-import re
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
 from typing import TypeAlias, TypedDict
@@ -148,32 +147,10 @@ def from_rawdata(cls, raw: dict):
         return RoleBasedAccessModel(roles)
 
 
-_OPERATORS_REGEX_PATTERN = re.compile(r"(&|\||\bAND\b|\bOR\b)")
-
-
 async def has_access_by_role(
     model: RoleBasedAccessModel,
     role: UserRole,
-    operations: str,
+    operation: str,
     context: OptionalContext = None,
 ) -> bool:
-    """Extends `RoleBasedAccessModel.can` to check access to boolean expressions of operations
-
-    Returns True if a user with a role has permission on a given context
-    """
-    tokens = _OPERATORS_REGEX_PATTERN.split(operations)
-    if len(tokens) == 1:
-        return await model.can(role, tokens[0], context)
-
-    if len(tokens) == 3:
-        tokens = [t.strip() for t in tokens if t.strip() != ""]
-        lhs, op, rhs = tokens
-        can_lhs = await model.can(role, lhs, context)
-        if op in ["AND", "&"]:
-            if can_lhs:
-                return await model.can(role, rhs, context)
-            return False
-        return can_lhs or (await model.can(role, rhs, context))
-
-    msg = f"Invalid expression '{operations}': only supports at most two operands"
-    raise NotImplementedError(msg)
+    return await model.can(role=role, operation=operation, context=context)

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

@@ -172,6 +172,6 @@ async def permits(
         return await has_access_by_role(
             self._access_model,
             role=user_role,
-            operations=permission,
+            operation=permission,
             context=context,
         )

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

@@ -258,18 +258,6 @@ async def test_check_access_expressions(access_model: RoleBasedAccessModel):
 
     assert await has_access_by_role(access_model, R.ANONYMOUS, "study.stop")
 
-    assert await has_access_by_role(
-        access_model, R.ANONYMOUS, "study.stop |study.node.create"
-    )
-
-    assert not await has_access_by_role(
-        access_model, R.ANONYMOUS, "study.stop & study.node.create"
-    )
-
-    assert await has_access_by_role(
-        access_model, R.USER, "study.stop & study.node.create"
-    )
-
 
 @pytest.fixture
 def mock_db(mocker: MockerFixture) -> MagicMock:
@@ -314,17 +302,17 @@ async def test_authorization_policy_cache(mocker: MockerFixture, mock_db: MagicM
     # pylint: disable=no-member
     autz_cache: BaseCache = authz_policy._get_authorized_user_or_none.cache
 
-    assert not (await autz_cache.exists("_get_auth_or_none/[email protected]"))
+    assert not (await autz_cache.exists("_get_authorized_user_or_none/[email protected]"))
     for _ in range(3):
         got = await authz_policy._get_authorized_user_or_none(email="[email protected]")
         assert mock_db.call_count == 1
         assert got["id"] == 1
 
-    assert await autz_cache.exists("_get_auth_or_none/[email protected]")
+    assert await autz_cache.exists("_get_authorized_user_or_none/[email protected]")
 
     # new value in db
     mock_db.users_db["[email protected]"]["id"] = 2
-    got = await autz_cache.get("_get_auth_or_none/[email protected]")
+    got = await autz_cache.get("_get_authorized_user_or_none/[email protected]")
     assert got["id"] == 1
 
     # gets cache, db is NOT called
@@ -341,12 +329,12 @@ async def test_authorization_policy_cache(mocker: MockerFixture, mock_db: MagicM
     assert got["id"] == 2
 
     # other email has other key
-    assert not (await autz_cache.exists("_get_auth_or_none/[email protected]"))
+    assert not (await autz_cache.exists("_get_authorized_user_or_none/[email protected]"))
 
     for _ in range(4):
         # NOTE: None
         assert await authz_policy._get_authorized_user_or_none(email="[email protected]")
-        assert await autz_cache.exists("_get_auth_or_none/[email protected]")
+        assert await autz_cache.exists("_get_authorized_user_or_none/[email protected]")
         assert mock_db.call_count == 3
 
     # should raise web.HTTPServiceUnavailable on db failure