AAP-32688: Support deletion of WCA API_KEY from Admin Portal (Backend) (

#1342)

ansible_ai_connect/ai/api/utils/seated_users_allow_list.py

@@ -215,6 +215,18 @@
         "timestamp": None,
         "plans": None,
     },
+    "modelApiKeyDelete": {
+        "duration": None,
+        "exception": None,
+        "problem": None,
+        "imageTags": None,
+        "hostname": None,
+        "groups": None,
+        "rh_user_has_seat": None,
+        "rh_user_org_id": None,
+        "timestamp": None,
+        "plans": None,
+    },
     "modelApiKeyValidate": {
         "duration": None,
         "exception": None,

ansible_ai_connect/ai/api/wca/api_key_views.py

@@ -19,7 +19,7 @@
 from drf_spectacular.utils import OpenApiResponse, extend_schema
 from oauth2_provider.contrib.rest_framework import IsAuthenticatedOrTokenHasScope
 from rest_framework.exceptions import ValidationError
-from rest_framework.generics import CreateAPIView, RetrieveAPIView
+from rest_framework.generics import CreateAPIView, DestroyAPIView, RetrieveAPIView
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
 from rest_framework.status import (
@@ -39,7 +39,11 @@
 from ansible_ai_connect.ai.api.serializers import WcaKeyRequestSerializer
 from ansible_ai_connect.ai.api.utils.segment import send_segment_event
 from ansible_ai_connect.ai.api.views import ServiceUnavailable
-from ansible_ai_connect.users.signals import user_set_wca_api_key
+from ansible_ai_connect.users.signals import (
+    user_delete_wca_api_key,
+    user_delete_wca_model_id,
+    user_set_wca_api_key,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -51,8 +55,8 @@
 ]
 
 
-class WCAApiKeyView(RetrieveAPIView, CreateAPIView):
-    required_scopes = ["read", "write"]
+class WCAApiKeyView(RetrieveAPIView, CreateAPIView, DestroyAPIView):
+    required_scopes = ["read", "write", "delete"]
     throttle_cache_key_suffix = "_wca_api_key"
     permission_classes = PERMISSION_CLASSES
 
@@ -180,6 +184,76 @@ def post(self, request, *args, **kwargs):
 
         return Response(status=HTTP_204_NO_CONTENT)
 
+    @extend_schema(
+        responses={
+            204: OpenApiResponse(description="Empty response"),
+            400: OpenApiResponse(description="Bad request"),
+            401: OpenApiResponse(description="Unauthorized"),
+            403: OpenApiResponse(description="Forbidden"),
+            429: OpenApiResponse(description="Request was throttled"),
+            500: OpenApiResponse(description="Internal service error"),
+        },
+        summary="DELETE WCA key for an Organization",
+        operation_id="wca_api_key_delete",
+    )
+    def delete(self, request, *args, **kwargs):
+        logger.debug("WCA API Key:: DELETE handler")
+
+        organization = None
+        exception = None
+        start_time = time.time()
+        try:
+            organization = request._request.user.organization
+            if not organization:
+                return Response(status=HTTP_400_BAD_REQUEST)
+
+            secret_manager = apps.get_app_config("ai").get_wca_secret_manager()
+
+            wca_key = secret_manager.get_secret(organization.id, Suffixes.API_KEY)
+            if wca_key is None:
+                return Response(status=HTTP_400_BAD_REQUEST)
+
+            secret_manager.delete_secret(organization.id, Suffixes.API_KEY)
+
+            # Audit trail/logging
+            user_delete_wca_api_key.send(
+                WCAApiKeyView.__class__,
+                user=request._request.user,
+                org_id=organization.id,
+                api_key=wca_key,
+            )
+            logger.info(f"Deleted API key secret for org_id '{organization.id}'")
+
+            # If model ID exists, delete it as well.
+            model_id = secret_manager.get_secret(organization.id, Suffixes.MODEL_ID)
+            if model_id is not None:
+                secret_manager.delete_secret(organization.id, Suffixes.MODEL_ID)
+
+                # Audit trail/logging
+                user_delete_wca_model_id.send(
+                    WCAApiKeyView.__class__,
+                    user=request._request.user,
+                    org_id=organization.id,
+                    model_id=model_id,
+                )
+                logger.info(f"Deleted model ID secret for org_id '{organization.id}'")
+
+        except Exception as e:
+            exception = e
+            logger.exception(e)
+            raise ServiceUnavailable(cause=e)
+
+        finally:
+            duration = round((time.time() - start_time) * 1000, 2)
+            event = {
+                "duration": duration,
+                "exception": exception is not None,
+                "problem": None if exception is None else exception.__class__.__name__,
+            }
+            send_segment_event(event, "modelApiKeyDelete", request.user)
+
+        return Response(status=HTTP_204_NO_CONTENT)
+
 
 class WCAApiKeyValidatorView(RetrieveAPIView):
     required_scopes = ["read"]

ansible_ai_connect/ai/api/wca/tests/test_api_key_views.py

@@ -13,7 +13,8 @@
 #  limitations under the License.
 
 from http import HTTPStatus
-from unittest.mock import patch
+from unittest import mock
+from unittest.mock import MagicMock, patch
 
 from django.apps import apps
 from django.test import override_settings
@@ -272,6 +273,254 @@ def test_set_key_throws_validation_exception(self, *args):
             self.assertEqual(r.status_code, HTTPStatus.BAD_REQUEST)
             self.assert_segment_log(log, "modelApiKeySet", "ValidationError")
 
+    @override_settings(SEGMENT_WRITE_KEY="DUMMY_KEY_VALUE")
+    def test_delete_key_without_org_id(self, *args):
+        self.client.force_authenticate(user=self.user)
+
+        with self.assertLogs(logger="root", level="DEBUG") as log:
+            r = self.client.delete(reverse("wca_api_key"))
+            self.assertEqual(r.status_code, HTTPStatus.BAD_REQUEST)
+            self.assert_segment_log(log, "modelApiKeyDelete", None)
+
+    @override_settings(SEGMENT_WRITE_KEY="DUMMY_KEY_VALUE")
+    def test_delete_key(self, *args):
+        self.user.organization = Organization.objects.get_or_create(id=123)[0]
+        mock_wca_client = apps.get_app_config("ai").model_mesh_client
+        mock_secret_manager = apps.get_app_config("ai").get_wca_secret_manager()
+        self.client.force_authenticate(user=self.user)
+
+        secrets = {
+            Suffixes.API_KEY: {"CreatedDate": timezone.now().isoformat()},
+            Suffixes.MODEL_ID: {"CreatedDate": timezone.now().isoformat()},
+        }
+
+        def get_secret(*args, **kwargs):
+            if args == (self.user.organization.id, Suffixes.API_KEY):
+                return secrets.get(Suffixes.API_KEY, None)
+            elif args == (self.user.organization.id, Suffixes.MODEL_ID):
+                return secrets.get(Suffixes.MODEL_ID, None)
+            else:
+                return Exception("exception occurred")
+
+        def delete_secret(*args, **kwargs):
+            if args == (self.user.organization.id, Suffixes.API_KEY):
+                return secrets.pop(Suffixes.API_KEY)
+            elif args == (self.user.organization.id, Suffixes.MODEL_ID):
+                return secrets.pop(Suffixes.MODEL_ID)
+            else:
+                return Exception("exception occurred")
+
+        mock_secret_manager.get_secret = MagicMock(side_effect=get_secret)
+        mock_secret_manager.delete_secret = MagicMock(side_effect=delete_secret)
+
+        r = self.client.get(reverse("wca_api_key"))
+        self.assertEqual(r.status_code, HTTPStatus.OK)
+        mock_secret_manager.get_secret.assert_called_with(
+            self.user.organization.id, Suffixes.API_KEY
+        )
+
+        # Delete key and model id
+        mock_wca_client.get_token.return_value = "token"
+        with self.assertLogs(logger="ansible_ai_connect.users.signals", level="DEBUG") as signals:
+            with self.assertLogs(logger="root", level="DEBUG") as log:
+                r = self.client.delete(reverse("wca_api_key"))
+                self.assertEqual(r.status_code, HTTPStatus.NO_CONTENT)
+                mock_secret_manager.delete_secret.assert_has_calls(
+                    [
+                        mock.call(self.user.organization.id, Suffixes.API_KEY),
+                        mock.call(self.user.organization.id, Suffixes.MODEL_ID),
+                    ]
+                )
+                self.assert_segment_log(log, "modelApiKeyDelete", None)
+
+            # Check audit entries
+            self.assertInLog(
+                f"User: '{self.user}' delete WCA Key for Organization "
+                f"'{self.user.organization.id}'",
+                signals,
+            )
+            self.assertInLog(
+                f"User: '{self.user}' delete WCA Model Id for Organization "
+                f"'{self.user.organization.id}'",
+                signals,
+            )
+
+        # Check Key was deleted
+        mock_secret_manager.get_secret.return_value = None
+        r = self.client.get(reverse("wca_api_key"))
+        self.assertEqual(r.status_code, HTTPStatus.OK)
+        mock_secret_manager.get_secret.assert_called_with(
+            self.user.organization.id, Suffixes.API_KEY
+        )
+
+    @override_settings(SEGMENT_WRITE_KEY="DUMMY_KEY_VALUE")
+    def test_delete_key_with_model_id_deletion_error(self, *args):
+        self.user.organization = Organization.objects.get_or_create(id=123)[0]
+        mock_wca_client = apps.get_app_config("ai").model_mesh_client
+        mock_secret_manager = apps.get_app_config("ai").get_wca_secret_manager()
+        self.client.force_authenticate(user=self.user)
+
+        secrets = {
+            Suffixes.API_KEY: {"CreatedDate": timezone.now().isoformat()},
+            Suffixes.MODEL_ID: {"CreatedDate": timezone.now().isoformat()},
+        }
+
+        def get_secret(*args, **kwargs):
+            if args == (self.user.organization.id, Suffixes.API_KEY):
+                return secrets.get(Suffixes.API_KEY, None)
+            elif args == (self.user.organization.id, Suffixes.MODEL_ID):
+                return secrets.get(Suffixes.MODEL_ID, None)
+            else:
+                return Exception("exception occurred")
+
+        def delete_secret(*args, **kwargs):
+            if args == (self.user.organization.id, Suffixes.API_KEY):
+                return secrets.pop(Suffixes.API_KEY)
+            elif args == (self.user.organization.id, Suffixes.MODEL_ID):
+                raise Exception("exception occurred")
+            else:
+                return Exception("exception occurred")
+
+        mock_secret_manager.get_secret = MagicMock(side_effect=get_secret)
+        mock_secret_manager.delete_secret = MagicMock(side_effect=delete_secret)
+
+        r = self.client.get(reverse("wca_api_key"))
+        self.assertEqual(r.status_code, HTTPStatus.OK)
+        mock_secret_manager.get_secret.assert_called_with(
+            self.user.organization.id, Suffixes.API_KEY
+        )
+
+        # Delete key and model id
+        mock_wca_client.get_token.return_value = "token"
+        with self.assertLogs(logger="root", level="DEBUG") as log:
+            r = self.client.delete(reverse("wca_api_key"))
+            self.assertEqual(r.status_code, HTTPStatus.SERVICE_UNAVAILABLE)
+            mock_secret_manager.delete_secret.assert_has_calls(
+                [
+                    mock.call(self.user.organization.id, Suffixes.MODEL_ID),
+                ]
+            )
+            self.assert_segment_log(log, "modelApiKeyDelete", None)
+
+    @override_settings(SEGMENT_WRITE_KEY="DUMMY_KEY_VALUE")
+    def test_delete_key_with_api_key_deletion_error(self, *args):
+        self.user.organization = Organization.objects.get_or_create(id=123)[0]
+        mock_wca_client = apps.get_app_config("ai").model_mesh_client
+        mock_secret_manager = apps.get_app_config("ai").get_wca_secret_manager()
+        self.client.force_authenticate(user=self.user)
+
+        secrets = {
+            Suffixes.API_KEY: {"CreatedDate": timezone.now().isoformat()},
+            Suffixes.MODEL_ID: {"CreatedDate": timezone.now().isoformat()},
+        }
+
+        def get_secret(*args, **kwargs):
+            if args == (self.user.organization.id, Suffixes.API_KEY):
+                return secrets.get(Suffixes.API_KEY, None)
+            elif args == (self.user.organization.id, Suffixes.MODEL_ID):
+                return secrets.get(Suffixes.MODEL_ID, None)
+            else:
+                return Exception("exception occurred")
+
+        def delete_secret(*args, **kwargs):
+            if args == (self.user.organization.id, Suffixes.API_KEY):
+                raise Exception("exception occurred")
+            elif args == (self.user.organization.id, Suffixes.MODEL_ID):
+                return secrets.pop(Suffixes.MODEL_ID)
+            else:
+                return Exception("exception occurred")
+
+        mock_secret_manager.get_secret = MagicMock(side_effect=get_secret)
+        mock_secret_manager.delete_secret = MagicMock(side_effect=delete_secret)
+
+        r = self.client.get(reverse("wca_api_key"))
+        self.assertEqual(r.status_code, HTTPStatus.OK)
+        mock_secret_manager.get_secret.assert_called_with(
+            self.user.organization.id, Suffixes.API_KEY
+        )
+
+        # Delete key and model id
+        mock_wca_client.get_token.return_value = "token"
+        with self.assertLogs(logger="root", level="DEBUG") as log:
+            r = self.client.delete(reverse("wca_api_key"))
+            self.assertEqual(r.status_code, HTTPStatus.SERVICE_UNAVAILABLE)
+            mock_secret_manager.delete_secret.assert_has_calls(
+                [
+                    mock.call(self.user.organization.id, Suffixes.API_KEY),
+                ]
+            )
+            self.assert_segment_log(log, "modelApiKeyDelete", None)
+
+    @override_settings(SEGMENT_WRITE_KEY="DUMMY_KEY_VALUE")
+    def test_delete_key_with_no_model_id(self, *args):
+        self.user.organization = Organization.objects.get_or_create(id=123)[0]
+        mock_wca_client = apps.get_app_config("ai").model_mesh_client
+        mock_secret_manager = apps.get_app_config("ai").get_wca_secret_manager()
+        self.client.force_authenticate(user=self.user)
+
+        def get_secret(*args, **kwargs):
+            if args == (self.user.organization.id, Suffixes.API_KEY):
+                return {"CreatedDate": timezone.now().isoformat()}
+            elif args == (self.user.organization.id, Suffixes.MODEL_ID):
+                return None
+            else:
+                return Exception("exception occurred")
+
+        mock_secret_manager.get_secret = MagicMock(side_effect=get_secret)
+        r = self.client.get(reverse("wca_api_key"))
+        self.assertEqual(r.status_code, HTTPStatus.OK)
+        mock_secret_manager.get_secret.assert_called_with(
+            self.user.organization.id, Suffixes.API_KEY
+        )
+
+        # Delete key and model id
+        mock_wca_client.get_token.return_value = "token"
+        with self.assertLogs(logger="ansible_ai_connect.users.signals", level="DEBUG") as signals:
+            with self.assertLogs(logger="root", level="DEBUG") as log:
+                r = self.client.delete(reverse("wca_api_key"))
+                self.assertEqual(r.status_code, HTTPStatus.NO_CONTENT)
+                mock_secret_manager.delete_secret.assert_has_calls(
+                    [mock.call(self.user.organization.id, Suffixes.API_KEY)]
+                )
+                self.assert_segment_log(log, "modelApiKeyDelete", None)
+
+            # Check audit entries
+            self.assertInLog(
+                f"User: '{self.user}' delete WCA Key for Organization "
+                f"'{self.user.organization.id}'",
+                signals,
+            )
+
+        # Check Key was deleted
+        mock_secret_manager.get_secret.return_value = None
+        r = self.client.get(reverse("wca_api_key"))
+        self.assertEqual(r.status_code, HTTPStatus.OK)
+        mock_secret_manager.get_secret.assert_called_with(
+            self.user.organization.id, Suffixes.API_KEY
+        )
+
+    def test_delete_key_with_no_key_no_model_id(self, *args):
+        self.user.organization = Organization.objects.get_or_create(id=123)[0]
+        mock_secret_manager = apps.get_app_config("ai").get_wca_secret_manager()
+        self.client.force_authenticate(user=self.user)
+
+        def get_secret(*args, **kwargs):
+            if args == (self.user.organization.id, Suffixes.API_KEY):
+                return None
+            elif args == (self.user.organization.id, Suffixes.MODEL_ID):
+                return None
+            else:
+                return Exception("exception occurred")
+
+        mock_secret_manager.get_secret = MagicMock(side_effect=get_secret)
+        r = self.client.delete(reverse("wca_api_key"))
+        self.assertEqual(r.status_code, HTTPStatus.BAD_REQUEST)
+
+        mock_secret_manager.get_secret.assert_called_with(
+            self.user.organization.id, Suffixes.API_KEY
+        )
+        mock_secret_manager.delete_secret.assert_not_called()
+
 
 @patch.object(IsOrganisationAdministrator, "has_permission", return_value=True)
 @patch.object(IsOrganisationLightspeedSubscriber, "has_permission", return_value=False)