Take account of admin/editor roles conferred by group membership

Fixes /projects?only_editable=true returning UPDATE:false for users who
hold the role via a group rather than direct team membership.

Author: apdavison

validation_service_api/validation_service/auth.py

@@ -88,6 +88,7 @@ def __init__(self, token, allow_anonymous=False):
         self.token = token
         self._identity = None
         self._teams = None
+        self._groups = None
         self._collab_info = {}
         self._connection_error = False
         self.username = None
@@ -111,6 +112,13 @@ def get_identity(self):
             self.username = username
         return self._identity
 
+    def get_groups(self):
+        if self._groups is None:
+            payload = _decode_jwt_payload(self.token.credentials)
+            raw = payload.get("group") or payload.get("groups") or []
+            self._groups = set(raw) if isinstance(raw, list) else set()
+        return self._groups
+
     async def _has_role(self, role, collab_name, client):
         roles_url = f"{settings.EBRAINS_IDM_API_URL}/teams/{collab_name}/{role}/users"
         headers = {"Authorization": f"Bearer {self.token.credentials}"}
@@ -123,6 +131,32 @@ async def _has_role(self, role, collab_name, client):
                 return True
         return False
 
+    async def _has_role_via_group(self, role, collab_name, client):
+        user_groups = self.get_groups()
+        if not user_groups:
+            return False
+        url = f"{settings.EBRAINS_IDM_API_URL}/teams/{collab_name}/{role}/groups"
+        headers = {"Authorization": f"Bearer {self.token.credentials}"}
+        try:
+            res = await client.get(url, headers=headers,
+                                   timeout=settings.AUTHENTICATION_TIMEOUT)
+            if res.status_code == 404:
+                return False
+            res.raise_for_status()
+        except Exception as err:
+            logger.warning("Group role check failed for %s/%s: %s",
+                           collab_name, role, err)
+            return False
+        role_group_names = {g.get("name") for g in res.json() if isinstance(g, dict)}
+        return bool(user_groups & role_group_names)
+
+    async def _user_has_role(self, role, collab_name, client):
+        direct, via_group = await asyncio.gather(
+            self._has_role(role, collab_name, client),
+            self._has_role_via_group(role, collab_name, client),
+        )
+        return direct or via_group
+
     async def get_teams(self):
         if self._teams is None:
             identity = self.get_identity()
@@ -144,9 +178,8 @@ async def get_teams(self):
                 )
             )
             for role in ("administrator", "editor"):
-                    # todo: get groups as well and check for group membership
                     async with AsyncClient() as client:
-                        found = await asyncio.gather(*[self._has_role(role, collab_name, client) for collab_name in collab_names])
+                        found = await asyncio.gather(*[self._user_has_role(role, collab_name, client) for collab_name in collab_names])
                         for include, collab_name in zip(found, collab_names.copy()):
                             if include:
                                 self._teams.append(f"collab-{collab_name}-{role}")

validation_service_api/validation_service/tests/fixtures.py

@@ -18,6 +18,11 @@
     not os.environ.get("VF_TEST_TOKEN"), reason="VF_TEST_TOKEN not set"
 )
 
+requires_group_token = pytest.mark.skipif(
+    not os.environ.get("VF_GROUP_TEST_TOKEN"),
+    reason="VF_GROUP_TEST_TOKEN not set",
+)
+
 client = TestClient(app)
 token = HTTPAuthorizationCredentials(
     scheme="Bearer",

validation_service_api/validation_service/tests/test_user.py

@@ -4,13 +4,16 @@
 Note: These will fail if the test user's permissions are too elevated.
 """
 
+import base64
+import json
 import os
+from unittest.mock import AsyncMock, MagicMock
 
 from fastapi.security import HTTPAuthorizationCredentials
 import pytest
 
 from ..auth import User
-from .fixtures import requires_token
+from .fixtures import requires_group_token, requires_token
 
 
 
@@ -19,6 +22,18 @@
 )
 
 
+def _make_token(payload: dict) -> HTTPAuthorizationCredentials:
+    """Build a syntactically-valid JWT whose payload decodes to the given dict.
+    Not signed — only suitable for tests that exercise _decode_jwt_payload."""
+    def b64(b: bytes) -> str:
+        return base64.urlsafe_b64encode(b).rstrip(b"=").decode()
+    header = b64(b'{"alg":"none","typ":"JWT"}')
+    body = b64(json.dumps(payload).encode())
+    return HTTPAuthorizationCredentials(
+        scheme="Bearer", credentials=f"{header}.{body}.sig"
+    )
+
+
 @pytest.mark.asyncio
 @requires_token
 async def test_user__is_admin():
@@ -60,3 +75,132 @@ async def test_can_edit_collab():
     user = User(token, allow_anonymous=False)
     can_edit = await user.can_edit_collab("model-validation")
     assert not can_edit
+
+
+# --- Unit tests for group-based role resolution (no live IDM calls) ---
+
+
+def test_get_groups_from_jwt():
+    user = User(
+        _make_token({"sub": "1", "preferred_username": "alice",
+                     "group": ["unit-x", "unit-y"]}),
+        allow_anonymous=False,
+    )
+    assert user.get_groups() == {"unit-x", "unit-y"}
+
+
+def test_get_groups_absent_claim_returns_empty_set():
+    user = User(
+        _make_token({"sub": "1", "preferred_username": "alice"}),
+        allow_anonymous=False,
+    )
+    assert user.get_groups() == set()
+
+
+def _mock_client(*, json_data=None, status_code=200):
+    resp = MagicMock()
+    resp.status_code = status_code
+    resp.json.return_value = json_data or []
+    resp.raise_for_status = MagicMock()
+    client = MagicMock()
+    client.get = AsyncMock(return_value=resp)
+    return client
+
+
+@pytest.mark.asyncio
+async def test_has_role_via_group_matches():
+    user = User(
+        _make_token({"sub": "1", "preferred_username": "alice",
+                     "group": ["unit-x"]}),
+        allow_anonymous=False,
+    )
+    client = _mock_client(json_data=[{"name": "unit-x"}, {"name": "other"}])
+    assert await user._has_role_via_group("administrator", "foo", client) is True
+
+
+@pytest.mark.asyncio
+async def test_has_role_via_group_no_intersection():
+    user = User(
+        _make_token({"sub": "1", "preferred_username": "alice",
+                     "group": ["unit-z"]}),
+        allow_anonymous=False,
+    )
+    client = _mock_client(json_data=[{"name": "unit-x"}])
+    assert await user._has_role_via_group("administrator", "foo", client) is False
+
+
+@pytest.mark.asyncio
+async def test_has_role_via_group_empty_groups_skips_call():
+    user = User(
+        _make_token({"sub": "1", "preferred_username": "alice"}),
+        allow_anonymous=False,
+    )
+    client = _mock_client()
+    assert await user._has_role_via_group("administrator", "foo", client) is False
+    client.get.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_has_role_via_group_404_returns_false():
+    user = User(
+        _make_token({"sub": "1", "preferred_username": "alice",
+                     "group": ["unit-x"]}),
+        allow_anonymous=False,
+    )
+    client = _mock_client(status_code=404)
+    assert await user._has_role_via_group("administrator", "foo", client) is False
+
+
+@pytest.mark.asyncio
+async def test_user_has_role_direct_only():
+    user = User(
+        _make_token({"sub": "1", "preferred_username": "alice"}),
+        allow_anonymous=False,
+    )
+    user._has_role = AsyncMock(return_value=True)
+    user._has_role_via_group = AsyncMock(return_value=False)
+    assert await user._user_has_role("administrator", "foo", MagicMock()) is True
+
+
+@pytest.mark.asyncio
+async def test_user_has_role_via_group_only():
+    user = User(
+        _make_token({"sub": "1", "preferred_username": "alice",
+                     "group": ["unit-x"]}),
+        allow_anonymous=False,
+    )
+    user._has_role = AsyncMock(return_value=False)
+    user._has_role_via_group = AsyncMock(return_value=True)
+    assert await user._user_has_role("administrator", "foo", MagicMock()) is True
+
+
+@pytest.mark.asyncio
+async def test_user_has_role_neither():
+    user = User(
+        _make_token({"sub": "1", "preferred_username": "alice"}),
+        allow_anonymous=False,
+    )
+    user._has_role = AsyncMock(return_value=False)
+    user._has_role_via_group = AsyncMock(return_value=False)
+    assert await user._user_has_role("administrator", "foo", MagicMock()) is False
+
+
+# --- Integration test: requires a token whose holder gets admin via a group ---
+
+
+@pytest.mark.asyncio
+@requires_group_token
+async def test_user_teams_includes_group_conferred_admin():
+    """With a token from a user who holds administrator on a collab via group
+    membership (e.g. closed-loop-motor-t3-4), the collab should be reported as
+    administrator, not viewer."""
+    group_token = HTTPAuthorizationCredentials(
+        scheme="Bearer",
+        credentials=os.environ["VF_GROUP_TEST_TOKEN"],
+    )
+    expected_collab = os.environ.get(
+        "VF_GROUP_TEST_COLLAB", "closed-loop-motor-t3-4"
+    )
+    user = User(group_token, allow_anonymous=False)
+    teams = await user.get_teams()
+    assert f"collab-{expected_collab}-administrator" in teams