Update role assignment check to handle service principals (#2466)

* Change get_user_role_assignments to get_identity_role_assignments

Update to handle user and servicePrincipal

Also update workspace list endpoint to gracefully fail if a workspace is missing auth info

* Fix: workspace auth written to wrong path

When running from the deployment repo, the WORKSPACE_API_CLIENT_ID value
was written under the AzureTRE folder

api_app/_version.py

@@ -1 +1 @@
-__version__ = "0.4.15"
+__version__ = "0.4.16"

api_app/api/routes/resource_helpers.py

@@ -62,9 +62,9 @@ def construct_location_header(operation: Operation) -> str:
     return f'/api{operation.resourcePath}/operations/{operation.id}'
 
 
-def get_user_role_assignments(user):
+def get_identity_role_assignments(user):
     access_service = get_access_service()
-    return access_service.get_user_role_assignments(user.id)
+    return access_service.get_identity_role_assignments(user.id)
 
 
 def get_app_user_roles_assignments_emails(app_obj_id):

api_app/api/routes/workspaces.py

@@ -20,12 +20,13 @@
 from models.schemas.workspace_service import WorkspaceServiceInCreate, WorkspaceServicesInList, WorkspaceServiceInResponse
 from models.schemas.resource import ResourcePatch
 from resources import strings
+from services.access_service import AuthConfigValidationError
 from services.authentication import get_current_admin_user, \
     get_access_service, get_current_workspace_owner_user, get_current_workspace_owner_or_researcher_user, get_current_tre_user_or_tre_admin, get_current_workspace_owner_or_researcher_user_or_tre_admin, get_current_workspace_owner_or_tre_admin
 from services.authentication import extract_auth_information
 from services.azure_resource_status import get_azure_resource_status
 from azure.cosmos.exceptions import CosmosAccessConditionFailedError
-from .resource_helpers import get_user_role_assignments, save_and_deploy_resource, construct_location_header, send_uninstall_message, \
+from .resource_helpers import get_identity_role_assignments, save_and_deploy_resource, construct_location_header, send_uninstall_message, \
     send_custom_action_message, send_resource_request_message
 from models.domain.request_action import RequestAction
 
@@ -58,15 +59,23 @@ async def retrieve_users_active_workspaces(request: Request, user=Depends(get_cu
         workspaces = workspace_repo.get_active_workspaces()
 
         access_service = get_access_service()
-        user_role_assignments = get_user_role_assignments(user)
-        user_workspaces = [workspace for workspace in workspaces if access_service.get_workspace_role(user, workspace, user_role_assignments) != WorkspaceRole.NoRole]
+        user_role_assignments = get_identity_role_assignments(user)
+
+        def _safe_get_workspace_role(user, workspace, user_role_assignments):
+            # provide graceful failure if there is a workspace without auth info
+            # to prevent it blocking listing other workspaces
+            try:
+                return access_service.get_workspace_role(user, workspace, user_role_assignments)
+            except AuthConfigValidationError:
+                return WorkspaceRole.NoRole
+        user_workspaces = [workspace for workspace in workspaces if _safe_get_workspace_role(user, workspace, user_role_assignments) != WorkspaceRole.NoRole]
         return WorkspacesInList(workspaces=user_workspaces)
 
 
 @workspaces_core_router.get("/workspaces/{workspace_id}", response_model=WorkspaceInResponse, name=strings.API_GET_WORKSPACE_BY_ID)
 async def retrieve_workspace_by_workspace_id(user=Depends(get_current_tre_user_or_tre_admin), workspace=Depends(get_workspace_by_id_from_path)) -> WorkspaceInResponse:
     access_service = get_access_service()
-    user_role_assignments = get_user_role_assignments(user)
+    user_role_assignments = get_identity_role_assignments(user)
     if access_service.get_workspace_role(user, workspace, user_role_assignments) != WorkspaceRole.NoRole:
         return WorkspaceInResponse(workspace=workspace)
     else:

api_app/resources/strings.py

@@ -86,6 +86,8 @@
 ACCESS_PLEASE_SUPPLY_CLIENT_ID = "Please supply the client_id for the AAD application"
 ACCESS_UNABLE_TO_GET_INFO_FOR_APP = "Unable to get app info for app:"
 ACCESS_UNABLE_TO_GET_ROLE_ASSIGNMENTS_FOR_USER = "Unable to get role assignments for user"
+ACCESS_UNABLE_TO_GET_ACCOUNT_TYPE = "Unable to look up account type"
+ACCESS_UNHANDLED_ACCOUNT_TYPE = "Unhandled account type"
 
 ACCESS_USER_IS_NOT_OWNER_OR_RESEARCHER = "Workspace Researcher or Owner rights are required"
 ACCESS_USER_IS_NOT_OWNER = "Workspace Owner rights are required"

api_app/services/aad_authentication.py

@@ -282,12 +282,39 @@ def _get_app_auth_info(self, client_id: str) -> dict:
 
         return authInfo
 
-    def _get_role_assignment_graph_data(self, user_id: str) -> dict:
+    def _get_role_assignment_graph_data_for_user(self, user_id: str) -> dict:
         msgraph_token = self._get_msgraph_token()
         user_endpoint = f"https://graph.microsoft.com/v1.0/users/{user_id}/appRoleAssignments"
         graph_data = requests.get(user_endpoint, headers=self._get_auth_header(msgraph_token)).json()
         return graph_data
 
+    def _get_role_assignment_graph_data_for_service_principal(self, principal_id: str) -> dict:
+        msgraph_token = self._get_msgraph_token()
+        user_endpoint = f"https://graph.microsoft.com/v1.0/servicePrincipals/{principal_id}/appRoleAssignments"
+        graph_data = requests.get(user_endpoint, headers=self._get_auth_header(msgraph_token)).json()
+        return graph_data
+
+    def _get_identity_type(self, id: str) -> str:
+        msgraph_token = self._get_msgraph_token()
+        objects_endpoint = "https://graph.microsoft.com/v1.0/directoryObjects/getByIds"
+        request_body = {"ids": [id], "types": ["user", "servicePrincipal"]}
+        graph_data = requests.post(
+            objects_endpoint,
+            headers=self._get_auth_header(msgraph_token),
+            json=request_body
+        ).json()
+
+        if "value" not in graph_data or len(graph_data["value"]) != 1:
+            logging.debug(graph_data)
+            raise AuthConfigValidationError(f"{strings.ACCESS_UNABLE_TO_GET_ACCOUNT_TYPE} {id}")
+
+        object_info = graph_data["value"][0]
+        if "@odata.type" not in object_info:
+            logging.debug(object_info)
+            raise AuthConfigValidationError(f"{strings.ACCESS_UNABLE_TO_GET_ACCOUNT_TYPE} {id}")
+
+        return object_info["@odata.type"]
+
     def extract_workspace_auth_information(self, data: dict) -> dict:
         if "client_id" not in data:
             raise AuthConfigValidationError(strings.ACCESS_PLEASE_SUPPLY_CLIENT_ID)
@@ -305,8 +332,15 @@ def extract_workspace_auth_information(self, data: dict) -> dict:
 
         return auth_info
 
-    def get_user_role_assignments(self, user_id: str) -> List[RoleAssignment]:
-        graph_data = self._get_role_assignment_graph_data(user_id)
+    def get_identity_role_assignments(self, user_id: str) -> List[RoleAssignment]:
+        identity_type = self._get_identity_type(user_id)
+        if identity_type == "#microsoft.graph.user":
+            graph_data = self._get_role_assignment_graph_data_for_user(user_id)
+        elif identity_type == "#microsoft.graph.servicePrincipal":
+            graph_data = self._get_role_assignment_graph_data_for_service_principal(user_id)
+        else:
+            logging.debug(graph_data)
+            raise AuthConfigValidationError(f"{strings.ACCESS_UNHANDLED_ACCOUNT_TYPE} {identity_type}")
 
         if 'value' not in graph_data:
             logging.debug(graph_data)

api_app/services/access_service.py

@@ -16,7 +16,7 @@ def extract_workspace_auth_information(self, data: dict) -> dict:
         pass
 
     @abstractmethod
-    def get_user_role_assignments(self, user_id: str) -> dict:
+    def get_identity_role_assignments(self, user_id: str) -> dict:
         pass
 
     @abstractmethod

api_app/tests_ma/test_api/test_routes/test_workspaces.py

@@ -219,7 +219,7 @@ def log_in_with_non_admin_user(self, app, non_admin_user):
 
     # [GET] /workspaces
     @patch("api.routes.workspaces.WorkspaceRepository.get_active_workspaces")
-    @patch("api.routes.workspaces.get_user_role_assignments", return_value=[])
+    @patch("api.routes.workspaces.get_identity_role_assignments", return_value=[])
     async def test_get_workspaces_returns_empty_list_when_no_resources_exist(self, access_service_mock, get_workspaces_mock, app, client) -> None:
         get_workspaces_mock.return_value = []
         access_service_mock.get_workspace_role.return_value = [WorkspaceRole.Owner]
@@ -229,7 +229,7 @@ async def test_get_workspaces_returns_empty_list_when_no_resources_exist(self, a
 
     # [GET] /workspaces
     @patch("api.routes.workspaces.WorkspaceRepository.get_active_workspaces")
-    @patch("api.routes.workspaces.get_user_role_assignments")
+    @patch("api.routes.workspaces.get_identity_role_assignments")
     async def test_get_workspaces_returns_correct_data_when_resources_exist(self, access_service_mock, get_workspaces_mock, app, client) -> None:
         auth_info_user_in_workspace_owner_role = {'sp_id': 'ab123', 'app_role_id_workspace_owner': 'ab124', 'app_role_id_workspace_researcher': 'ab125', 'app_role_id_workspace_airlock_manager': 'ab130'}
         auth_info_user_in_workspace_researcher_role = {'sp_id': 'ab123', 'app_role_id_workspace_owner': 'ab127', 'app_role_id_workspace_researcher': 'ab126', 'app_role_id_workspace_airlock_manager': 'ab130'}
@@ -251,7 +251,7 @@ async def test_get_workspaces_returns_correct_data_when_resources_exist(self, ac
 
     # [GET] /workspaces/{workspace_id}
     @patch("api.dependencies.workspaces.WorkspaceRepository.get_workspace_by_id")
-    @patch("api.routes.workspaces.get_user_role_assignments")
+    @patch("api.routes.workspaces.get_identity_role_assignments")
     async def test_get_workspace_by_id_get_returns_workspace_if_found(self, access_service_mock, get_workspace_mock, app, client):
         auth_info_user_in_workspace_owner_role = {'sp_id': 'ab123', 'app_role_id_workspace_owner': 'ab124', 'app_role_id_workspace_researcher': 'ab125', 'app_role_id_workspace_airlock_manager': 'ab130'}
         workspace = sample_workspace(auth_info=auth_info_user_in_workspace_owner_role)
@@ -264,15 +264,15 @@ async def test_get_workspace_by_id_get_returns_workspace_if_found(self, access_s
 
     # [GET] /workspaces/{workspace_id}
     @patch("api.dependencies.workspaces.WorkspaceRepository.get_workspace_by_id", side_effect=EntityDoesNotExist)
-    @patch("api.routes.workspaces.get_user_role_assignments")
+    @patch("api.routes.workspaces.get_identity_role_assignments")
     async def test_get_workspace_by_id_get_returns_404_if_resource_is_not_found(self, access_service_mock, _, app, client):
         access_service_mock.return_value = [RoleAssignment('ab123', 'ab124')]
         response = await client.get(app.url_path_for(strings.API_GET_WORKSPACE_BY_ID, workspace_id=WORKSPACE_ID))
         assert response.status_code == status.HTTP_404_NOT_FOUND
 
     # [GET] /workspaces/{workspace_id}
     @patch("api.dependencies.workspaces.WorkspaceRepository.get_workspace_by_id")
-    @patch("api.routes.workspaces.get_user_role_assignments")
+    @patch("api.routes.workspaces.get_identity_role_assignments")
     async def test_get_workspace_by_id_get_returns_422_if_workspace_id_is_not_a_uuid(self, access_service_mock, _, app, client):
         access_service_mock.return_value = [RoleAssignment('ab123', 'ab124')]
         response = await client.get(app.url_path_for(strings.API_GET_WORKSPACE_BY_ID, workspace_id="not_valid"))

api_app/tests_ma/test_services/test_aad_access_service.py

@@ -93,29 +93,29 @@ def test_extract_workspace__returns_sp_id_and_roles(get_app_sp_graph_data_mock):
                                         properties={'client_id': '1234', 'sp_id': 'abc127', 'app_role_id_workspace_owner': 'abc128', 'app_role_id_workspace_researcher': 'abc129', 'app_role_id_workspace_airlock_manager': 'abc130'}),
                               WorkspaceRole.AirlockManager)
                          ])
-@patch("services.aad_authentication.AzureADAuthorization.get_user_role_assignments")
-def test_get_workspace_role_returns_correct_owner(get_user_role_assignments_mock, user: User, workspace: Workspace, expected_role: WorkspaceRole):
+@patch("services.aad_authentication.AzureADAuthorization.get_identity_role_assignments")
+def test_get_workspace_role_returns_correct_owner(get_identity_role_assignments_mock, user: User, workspace: Workspace, expected_role: WorkspaceRole):
 
-    get_user_role_assignments_mock.return_value = user.roleAssignments
+    get_identity_role_assignments_mock.return_value = user.roleAssignments
 
     access_service = AzureADAuthorization()
-    actual_role = access_service.get_workspace_role(user, workspace, access_service.get_user_role_assignments(user.id))
+    actual_role = access_service.get_workspace_role(user, workspace, access_service.get_identity_role_assignments(user.id))
 
     assert actual_role == expected_role
 
 
-@patch("services.aad_authentication.AzureADAuthorization.get_user_role_assignments", return_value=[("ab123", "ab124")])
+@patch("services.aad_authentication.AzureADAuthorization.get_identity_role_assignments", return_value=[("ab123", "ab124")])
 def test_raises_auth_config_error_if_workspace_auth_config_is_not_set(_):
     access_service = AzureADAuthorization()
 
     user = User(id='123', name="test", email="t@t.com")
     workspace_with_no_auth_config = Workspace(id='abc', etag='', templateName='template-name', templateVersion='0.1.0', resourcePath="test")
 
     with pytest.raises(AuthConfigValidationError):
-        _ = access_service.get_workspace_role(user, workspace_with_no_auth_config, access_service.get_user_role_assignments(user.id))
+        _ = access_service.get_workspace_role(user, workspace_with_no_auth_config, access_service.get_identity_role_assignments(user.id))
 
 
-@patch("services.aad_authentication.AzureADAuthorization.get_user_role_assignments", return_value=[("ab123", "ab124")])
+@patch("services.aad_authentication.AzureADAuthorization.get_identity_role_assignments", return_value=[("ab123", "ab124")])
 def test_raises_auth_config_error_if_auth_info_has_incorrect_roles(_):
     access_service = AzureADAuthorization()
 
@@ -129,4 +129,4 @@ def test_raises_auth_config_error_if_auth_info_has_incorrect_roles(_):
         resourcePath="test")
 
     with pytest.raises(AuthConfigValidationError):
-        _ = access_service.get_workspace_role(user, workspace_with_auth_info_but_no_roles, access_service.get_user_role_assignments())
+        _ = access_service.get_workspace_role(user, workspace_with_auth_info_but_no_roles, access_service.get_identity_role_assignments())

devops/scripts/aad/create_workspace_application.sh

@@ -358,7 +358,7 @@ fi
 {
   echo "WORKSPACE_API_CLIENT_ID=\"${workspaceAppId}\""
   echo "WORKSPACE_API_CLIENT_SECRET=\"${spPassword}\""
-} >> "$DIR"/../../auth.env
+} >> "devops/auth.env"
 
 if [[ $grantAdminConsent -eq 0 ]]; then
     echo "NOTE: Make sure the API permissions of the app registrations have admin consent granted."