Add route to retrieve users assigned to a workspace with UI (#4058)

* View users who are assigned roles within a workspace
Fixes #4049

* Fix iteration through roles.

* remove unused metadata

---------

Co-authored-by: Tim Allen <[email protected]>

api_app/_version.py

@@ -1 +1 @@
-__version__ = "0.18.11"
+__version__ = "0.19.2"

api_app/api/routes/api.py

@@ -24,7 +24,7 @@
     {"name": "workspaces", "description": " **Workspace Owners and Researchers** can view their own workspaces"},
     {"name": "workspace services", "description": "**Workspace Owners** administer workspace services, **Workspace Owners and Researchers** can view services in the workspaces they belong to"},
     {"name": "user resources", "description": "**Researchers** administer and can view their own researchers, **Workspace Owners** can view/update/delete all user resources in their workspaces"},
-    {"name": "shared services", "description": "**TRE administratiors** administer shared services"},
+    {"name": "shared services", "description": "**TRE administratiors** administer shared services"}
 ]
 
 # Root

api_app/api/routes/workspaces.py

@@ -21,6 +21,7 @@
 from models.schemas.workspace_service import WorkspaceServiceInCreate, WorkspaceServicesInList, WorkspaceServiceInResponse
 from models.schemas.resource import ResourceHistoryInList, ResourcePatch
 from models.schemas.resource_template import ResourceTemplateInformationInList
+from models.schemas.users import UsersInResponse
 from resources import strings
 from services.access_service import AuthConfigValidationError
 from services.authentication import get_current_admin_user, \
@@ -187,6 +188,13 @@ async def invoke_action_on_workspace(response: Response, action: str, user=Depen
     return OperationInResponse(operation=operation)
 
 
+@workspaces_shared_router.get("/workspaces/{workspace_id}/users", response_model=UsersInResponse, name=strings.API_GET_WORKSPACE_USERS)
+async def get_workspace_users(workspace=Depends(get_workspace_by_id_from_path)) -> UsersInResponse:
+    access_service = get_access_service()
+    users = access_service.get_workspace_users(workspace)
+    return UsersInResponse(users=users)
+
+
 # workspace operations
 # This method only returns templates that the authenticated user is authorized to use
 @workspaces_shared_router.get("/workspaces/{workspace_id}/workspace-service-templates", response_model=ResourceTemplateInformationInList, name=strings.API_GET_WORKSPACE_SERVICE_TEMPLATES_IN_WORKSPACE)

api_app/models/domain/authentication.py

@@ -10,6 +10,6 @@
 class User(BaseModel):
     id: str
     name: str
-    email: str
+    email: str = Field(None)
     roles: List[str] = Field([])
     roleAssignments: List[RoleAssignment] = Field([])

api_app/models/schemas/users.py

@@ -0,0 +1,28 @@
+from pydantic import BaseModel, Field
+from typing import List
+
+from models.domain.authentication import User
+
+
+class UsersInResponse(BaseModel):
+    users: List[User] = Field(..., title="Users", description="List of users assigned to the workspace")
+
+    class Config:
+        schema_extra = {
+            "example": {
+                "users": [
+                    {
+                        "id": 1,
+                        "name": "John Doe",
+                        "email": "[email protected]",
+                        "roles": ["WorkspaceOwner", "WorkspaceResearcher"]
+                    },
+                    {
+                        "id": 2,
+                        "name": "Jane Smith",
+                        "email": "[email protected]",
+                        "roles": ["WorkspaceResearcher"]
+                    }
+                ]
+            }
+        }

api_app/resources/strings.py

@@ -15,6 +15,8 @@
 API_UPDATE_WORKSPACE = "Update an existing workspace"
 API_INVOKE_ACTION_ON_WORKSPACE = "Invoke action on a workspace"
 
+API_GET_WORKSPACE_USERS = "Get all users for a workspace"
+
 API_GET_ALL_WORKSPACE_SERVICES = "Get all workspace services for workspace"
 API_GET_WORKSPACE_SERVICE_BY_ID = "Get workspace service by Id"
 API_CREATE_WORKSPACE_SERVICE = "Create a workspace service"

api_app/services/aad_authentication.py

@@ -219,19 +219,19 @@ def _get_service_principal_endpoint(client_id) -> str:
 
     @staticmethod
     def _get_service_principal_assigned_roles_endpoint(client_id) -> str:
-        return f"{MICROSOFT_GRAPH_URL}/v1.0/serviceprincipals/{client_id}/appRoleAssignedTo?$select=appRoleId,principalId,principalType"
+        return f"{MICROSOFT_GRAPH_URL}/v1.0/serviceprincipals/{client_id}/appRoleAssignedTo?$select=appRoleId,principalId,principalType,principalDisplayName"
 
     @staticmethod
     def _get_batch_endpoint() -> str:
         return f"{MICROSOFT_GRAPH_URL}/v1.0/$batch"
 
     @staticmethod
     def _get_users_endpoint(user_object_id) -> str:
-        return "/users/" + user_object_id + "?$select=mail,id"
+        return "/users/" + user_object_id + "?$select=displayName,mail,id"
 
     @staticmethod
     def _get_group_members_endpoint(group_object_id) -> str:
-        return "/groups/" + group_object_id + "/transitiveMembers?$select=mail,id"
+        return "/groups/" + group_object_id + "/transitiveMembers?$select=displayName,mail,id"
 
     def _get_app_sp_graph_data(self, client_id: str) -> dict:
         msgraph_token = self._get_msgraph_token()
@@ -243,7 +243,7 @@ def _get_user_role_assignments(self, client_id, msgraph_token):
         sp_roles_endpoint = self._get_service_principal_assigned_roles_endpoint(client_id)
         return requests.get(sp_roles_endpoint, headers=self._get_auth_header(msgraph_token)).json()
 
-    def _get_user_emails(self, roles_graph_data, msgraph_token):
+    def _get_user_details(self, roles_graph_data, msgraph_token):
         batch_endpoint = self._get_batch_endpoint()
         batch_request_body = self._get_batch_users_by_role_assignments_body(roles_graph_data)
         headers = self._get_auth_header(msgraph_token)
@@ -262,43 +262,59 @@ def _get_user_emails(self, roles_graph_data, msgraph_token):
 
         return users_graph_data
 
-    def _get_user_emails_from_response(self, users_graph_data):
-        user_emails = {}
-        for user_data in users_graph_data["responses"]:
-            # Handle user endpoint response
-            if "users" in user_data["body"]["@odata.context"] and user_data["body"]["mail"] is not None:
-                user_emails[user_data["body"]["id"]] = [user_data["body"]["mail"]]
-            # Handle group endpoint response
-            if "directoryObjects" in user_data["body"]["@odata.context"]:
-                group_members_emails = []
-                for group_member in user_data["body"]["value"]:
-                    if group_member["mail"] is not None and group_member["mail"] not in group_members_emails:
-                        group_members_emails.append(group_member["mail"])
-                user_emails[user_data["id"]] = group_members_emails
-        return user_emails
+    def _get_roles_for_principal(self, user_id, roles_graph_data, app_id_to_role_name):
+        roles = []
+        for role_assignment in roles_graph_data["value"]:
+            if role_assignment["principalId"] == user_id:
+                roles.append(app_id_to_role_name[role_assignment["appRoleId"]])
+        return roles
 
-    def get_workspace_role_assignment_details(self, workspace: Workspace):
-        msgraph_token = self._get_msgraph_token()
-        app_role_ids = {role_name: workspace.properties[role_id] for role_name, role_id in self.WORKSPACE_ROLES_DICT.items()}
-        inverted_app_role_ids = {role_id: role_name for role_name, role_id in app_role_ids.items()}
+    def _get_users_inc_groups_from_response(self, users_graph_data, roles_graph_data, app_id_to_role_name) -> List[User]:
+        users = []
+        for user_data in users_graph_data["responses"]:
+            if "users" in user_data["body"]["@odata.context"]:
+                # Handle user endpoint response
+                user_id = user_data["body"]["id"]
+                user_name = user_data["body"]["displayName"]
 
-        sp_id = workspace.properties["sp_id"]
-        roles_graph_data = self._get_user_role_assignments(sp_id, msgraph_token)
-        users_graph_data = self._get_user_emails(roles_graph_data, msgraph_token)
-        user_emails = self._get_user_emails_from_response(users_graph_data)
+                if "users" in user_data["body"]["@odata.context"]:
+                    user_email = user_data["body"]["mail"]
+                    # if user with id does not already exist in users
+                    if not any(user.id == user_id for user in users):
+                        users.append(User(id=user_id, name=user_name, email=user_email, roles=self._get_roles_for_principal(user_id, roles_graph_data, app_id_to_role_name)))
 
-        workspace_role_assignments_details = defaultdict(list)
-        for role_assignment in roles_graph_data["value"]:
-            principal_id = role_assignment["principalId"]
-            principal_type = role_assignment["principalType"]
+            # Handle group endpoint response
+            elif "directoryObjects" in user_data["body"]["@odata.context"]:
+                group_id = user_data["id"]
+                for group_member in user_data["body"]["value"]:
+                    user_id = group_member["id"]
+                    user_name = group_member["displayName"]
+                    user_email = group_member["mail"]
 
-            if principal_type != "ServicePrincipal" and principal_id in user_emails:
-                app_role_id = role_assignment["appRoleId"]
-                app_role_name = inverted_app_role_ids.get(app_role_id)
+                    if not any(user.id == user_id for user in users):
+                        users.append(User(id=user_id, name=user_name, email=user_email, roles=self._get_roles_for_principal(group_id, roles_graph_data, app_id_to_role_name)))
 
-                if app_role_name:
-                    workspace_role_assignments_details[app_role_name].extend(user_emails[principal_id])
+        return users
 
+    def get_workspace_users(self, workspace: Workspace) -> List[User]:
+        msgraph_token = self._get_msgraph_token()
+        sp_graph_data = self._get_app_sp_graph_data(workspace.properties["client_id"])
+        app_id_to_role_name = {app_role["id"]: app_role["value"] for app_role in sp_graph_data["value"][0]["appRoles"]}
+        roles_graph_data = self._get_user_role_assignments(workspace.properties["sp_id"], msgraph_token)
+        users_graph_data = self._get_user_details(roles_graph_data, msgraph_token)
+        users_inc_groups = self._get_users_inc_groups_from_response(users_graph_data, roles_graph_data, app_id_to_role_name)
+
+        return users_inc_groups
+
+    def get_workspace_user_emails_by_role_assignment(self, workspace: Workspace):
+        users = self.get_workspace_users(workspace)
+        workspace_role_assignments_details = {}
+        for user in users:
+            if user.email:
+                for role in user.roles:
+                    if role not in workspace_role_assignments_details:
+                        workspace_role_assignments_details[role] = []
+                    workspace_role_assignments_details[role].append(user.email)
         return workspace_role_assignments_details
 
     def _get_batch_users_by_role_assignments_body(self, roles_graph_data):

api_app/services/access_service.py

@@ -20,7 +20,11 @@ def get_identity_role_assignments(self, user_id: str) -> dict:
         pass
 
     @abstractmethod
-    def get_workspace_role_assignment_details(self, workspace: Workspace) -> dict:
+    def get_workspace_users(self, workspace: Workspace) -> List[User]:
+        pass
+
+    @abstractmethod
+    def get_workspace_user_emails_by_role_assignment(self, workspace: Workspace) -> dict:
         pass
 
     @staticmethod

api_app/services/airlock.py

@@ -273,7 +273,7 @@ async def _handle_existing_review_resource(existing_resource: AirlockReviewUserR
 async def save_and_publish_event_airlock_request(airlock_request: AirlockRequest, airlock_request_repo: AirlockRequestRepository, user: User, workspace: Workspace):
 
     access_service = get_access_service()
-    role_assignment_details = access_service.get_workspace_role_assignment_details(workspace)
+    role_assignment_details = access_service.get_workspace_user_emails_by_role_assignment(workspace)
     if config.ENABLE_AIRLOCK_EMAIL_CHECK:
         check_email_exists(role_assignment_details)
 
@@ -332,7 +332,7 @@ async def update_and_publish_event_airlock_request(
         logger.debug(f"Sending status changed event for airlock request item: {airlock_request.id}")
         await send_status_changed_event(airlock_request=updated_airlock_request, previous_status=airlock_request.status)
         access_service = get_access_service()
-        role_assignment_details = access_service.get_workspace_role_assignment_details(workspace)
+        role_assignment_details = access_service.get_workspace_user_emails_by_role_assignment(workspace)
         await send_airlock_notification_event(updated_airlock_request, workspace, role_assignment_details)
         return updated_airlock_request
     except Exception:

api_app/tests_ma/test_api/test_routes/test_airlock.py

@@ -135,7 +135,7 @@ def log_in_with_researcher_user(self, app, researcher_user):
                 patch("api.routes.workspaces.OperationRepository.resource_has_deployed_operation"), \
                 patch("api.routes.airlock.AirlockRequestRepository.save_item"), \
                 patch("api.dependencies.workspaces.WorkspaceRepository.get_workspace_by_id"), \
-                patch("services.aad_authentication.AzureADAuthorization.get_workspace_role_assignment_details", return_value={"WorkspaceResearcher": ["[email protected]"], "WorkspaceOwner": ["[email protected]"], "AirlockManager": ["[email protected]"]}):
+                patch("services.aad_authentication.AzureADAuthorization.get_workspace_user_emails_by_role_assignment", return_value={"WorkspaceResearcher": ["[email protected]"], "WorkspaceOwner": ["[email protected]"], "AirlockManager": ["[email protected]"]}):
             yield
         app.dependency_overrides = {}