Merge pull request #1436 from dod-ccpo/azure-disable-user

azure disable user call

atst/domain/csp/cloud/azure_cloud_provider.py

@@ -311,6 +311,41 @@ class AzureCloudProvider(CloudProviderInterface):
             management_group_id=management_group_id,
         )
 
+    def disable_user(self, tenant_id, role_assignment_cloud_id):
+        sp_token = self._get_tenant_principal_token(tenant_id)
+        if sp_token is None:
+            raise AuthenticationException("Could not resolve token in disable user")
+        headers = {
+            "Authorization": f"Bearer {sp_token}",
+        }
+
+        try:
+            result = self.sdk.requests.delete(
+                f"{self.sdk.cloud.endpoints.resource_manager}/{role_assignment_cloud_id}?api-version=2015-07-01",
+                headers=headers,
+                timeout=30,
+            )
+            result.raise_for_status()
+            return result.json()
+
+        except self.sdk.requests.exceptions.ConnectionError:
+            app.logger.error(
+                f"Could not disable user. Connection Error", exc_info=1,
+            )
+            raise ConnectionException("connection error azure disable user")
+        except self.sdk.requests.exceptions.Timeout:
+            app.logger.error(
+                f"Could not disable user. Request timed out.", exc_info=1,
+            )
+            raise ConnectionException("timout error azure disable user")
+        except self.sdk.requests.exceptions.HTTPError as exc:
+            app.logger.error(
+                result.status_code, "azure application error disable user", exc_info=1,
+            )
+            raise UnknownServerException(
+                result.status_code, f"azure application error disable user. {str(exc)}",
+            )
+
     def create_tenant(self, payload: TenantCSPPayload):
         sp_token = self._get_root_provisioning_token()
         if sp_token is None:

atst/domain/csp/cloud/cloud_provider_interface.py

@@ -54,13 +54,13 @@ class CloudProviderInterface:  # pragma: no cover
         """
         raise NotImplementedError()
 
-    def disable_user(self, auth_credentials: Dict, csp_user_id: str) -> bool:
+    def disable_user(self, tenant_id: str, role_assignment_cloud_id: str) -> bool:
         """Revoke all privileges for a user. Used to prevent user access while a full
         delete is being processed.
 
         Arguments:
-            auth_credentials -- Object containing CSP account credentials
-            csp_user_id -- CSP internal user identifier
+            tenant_id -- CSP internal tenant identifier
+            role_assignment_cloud_id -- CSP name of the role assignment to delete.
 
         Returns:
             bool -- True on success

atst/domain/csp/cloud/mock_cloud_provider.py

@@ -417,14 +417,13 @@ class MockCloudProvider(CloudProviderInterface):
         self._maybe_raise(self.UNAUTHORIZED_RATE, self.AUTHORIZATION_EXCEPTION)
         return self._id()
 
-    def disable_user(self, auth_credentials, csp_user_id):
-        self._authorize(auth_credentials)
+    def disable_user(self, tenant_id, role_assignment_cloud_id):
         self._maybe_raise(self.NETWORK_FAILURE_PCT, self.NETWORK_EXCEPTION)
         self._maybe_raise(self.SERVER_FAILURE_PCT, self.SERVER_EXCEPTION)
 
         self._maybe_raise(
             self.ATAT_ADMIN_CREATE_FAILURE_PCT,
-            UserRemovalException(csp_user_id, "Could not disable user."),
+            UserRemovalException(tenant_id, "Could not disable user."),
         )
 
         return self._maybe(12)

tests/domain/cloud/test_azure_csp.py

@@ -206,6 +206,51 @@ def test_create_policy_definition_succeeds(mock_azure: AzureCloudProvider):
     )
 
 
+def test_disable_user(mock_azure: AzureCloudProvider):
+
+    assignment_guid = str(uuid4())
+    management_group_id = str(uuid4())
+    assignment_id = f"/providers/Microsoft.Management/managementGroups/{management_group_id}/providers/Microsoft.Authorization/roleAssignments/{assignment_guid}"
+
+    mock_result = Mock()
+    mock_result.json.return_value = {
+        "properties": {
+            "roleDefinitionId": "/subscriptions/subId/providers/Microsoft.Authorization/roleDefinitions/roledefinitionId",
+            "principalId": "Pid",
+            "scope": "/subscriptions/subId/resourcegroups/rgname",
+        },
+        "id": assignment_id,
+        "type": "Microsoft.Authorization/roleAssignments",
+        "name": assignment_guid,
+    }
+
+    mock_result.status_code = 200
+    mock_http_error_resp = mock_requests_response(
+        status=500,
+        raise_for_status=mock_azure.sdk.requests.exceptions.HTTPError(
+            "500 Server Error"
+        ),
+    )
+    mock_azure.sdk.requests.delete.side_effect = [
+        mock_azure.sdk.requests.exceptions.ConnectionError,
+        mock_azure.sdk.requests.exceptions.Timeout,
+        mock_http_error_resp,
+        mock_result,
+    ]
+    mock_azure = mock_get_secret(mock_azure)
+
+    tenant_id = "60ff9d34-82bf-4f21-b565-308ef0533435"
+    with pytest.raises(ConnectionException):
+        mock_azure.disable_user(tenant_id, assignment_guid)
+    with pytest.raises(ConnectionException):
+        mock_azure.disable_user(tenant_id, assignment_guid)
+    with pytest.raises(UnknownServerException, match=r".*500 Server Error.*"):
+        mock_azure.disable_user(tenant_id, assignment_guid)
+
+    result = mock_azure.disable_user(tenant_id, assignment_guid)
+    assert result.get("name") == assignment_guid
+
+
 def test_create_tenant(mock_azure: AzureCloudProvider):
     mock_result = Mock()
     mock_result.json.return_value = {

tests/domain/cloud/test_mock_csp.py

@@ -79,4 +79,4 @@ def test_create_or_update_user(mock_csp: MockCloudProvider):
 
 
 def test_disable_user(mock_csp: MockCloudProvider):
-    assert mock_csp.disable_user(CREDENTIALS, "csp_user_id")
+    assert mock_csp.disable_user("tenant_id", "role_assignment_cloud_id")