UN-2407 [FEAT] Add global API deployment keys for grouped API deployments

backend/api_v2/api_deployment_views.py

@@ -74,7 +74,12 @@ def post(
         organization = api.organization
 
         serializer = ExecutionRequestSerializer(
-            data=request.data, context={"api": api, "api_key": api_key}
+            data=request.data,
+            context={
+                "api": api,
+                "api_key": api_key,
+                "is_global_key": deployment_execution_dto.is_global_key,
+            },
         )
         serializer.is_valid(raise_exception=True)
         file_objs = serializer.validated_data.get(ApiExecution.FILES_FORM_DATA, [])

backend/api_v2/deployment_helper.py

@@ -9,6 +9,7 @@
 from configuration.models import Configuration
 from django.conf import settings
 from django.core.files.uploadedfile import InMemoryUploadedFile, UploadedFile
+from global_api_deployment_key.models import GlobalApiDeploymentKey
 from plugins.workflow_manager.workflow_v2.api_hub_usage_utils import APIHubUsageUtil
 from rest_framework.request import Request
 from rest_framework.serializers import Serializer
@@ -33,6 +34,7 @@
     InactiveAPI,
     InvalidAPIRequest,
     PresignedURLFetchError,
+    UnauthorizedKey,
 )
 from api_v2.key_helper import KeyHelper
 from api_v2.models import APIDeployment, APIKey
@@ -61,31 +63,63 @@ def validate_and_process(
         """Fetch API deployment and validate API key."""
         api_name = kwargs.get("api_name") or request.data.get("api_name")
         api_deployment = DeploymentHelper.get_deployment_by_api_name(api_name=api_name)
-        DeploymentHelper.validate_api(api_deployment=api_deployment, api_key=api_key)
+        global_key = DeploymentHelper.validate_api(
+            api_deployment=api_deployment, api_key=api_key
+        )
 
         deployment_execution_dto = DeploymentExecutionDTO(
-            api=api_deployment, api_key=api_key
+            api=api_deployment, api_key=api_key, global_key=global_key
         )
         kwargs["deployment_execution_dto"] = deployment_execution_dto
         return func(self, request, *args, **kwargs)
 
     @staticmethod
-    def validate_api(api_deployment: APIDeployment | None, api_key: str) -> None:
-        """Validating API and API key.
+    def validate_api(
+        api_deployment: APIDeployment | None, api_key: str
+    ) -> GlobalApiDeploymentKey | None:
+        """Validate API deployment and API key.
+
+        Tries deployment-specific key first. If that fails, falls back to
+        Global API Deployment Key validation.
 
         Args:
-            api_deployment (Optional[APIDeployment]): _description_
-            api_key (str): _description_
+            api_deployment: The API deployment instance
+            api_key: The bearer token value
+
+        Returns:
+            The authorizing ``GlobalApiDeploymentKey`` when the request was
+            authenticated via a global key, else ``None`` (deployment-specific
+            key). Returning the key (not a bool) preserves audit identity.
 
         Raises:
-            APINotFound: _description_
-            InactiveAPI: _description_
+            APINotFound: If deployment not found
+            InactiveAPI: If deployment is inactive
+            UnauthorizedKey: If key validation fails
         """
         if not api_deployment:
             raise APINotFound()
         if not api_deployment.is_active:
             raise InactiveAPI()
-        KeyHelper.validate_api_key(api_key=api_key, instance=api_deployment)
+
+        try:
+            KeyHelper.validate_api_key(api_key=api_key, instance=api_deployment)
+            return None
+        except UnauthorizedKey:
+            logger.debug(
+                "Deployment-specific key auth failed for API '%s'; falling back "
+                "to global API deployment key validation.",
+                api_deployment.api_name,
+            )
+            global_key = KeyHelper.validate_global_api_deployment_key(
+                api_key=api_key, api_deployment=api_deployment
+            )
+            logger.info(
+                "API '%s' authorized via global API deployment key '%s' (%s).",
+                api_deployment.api_name,
+                global_key.name,
+                global_key.id,
+            )
+            return global_key
 
     @staticmethod
     def validate_and_get_workflow(workflow_id: str) -> Workflow:

backend/api_v2/dto.py

@@ -1,11 +1,23 @@
 from dataclasses import dataclass
+from typing import TYPE_CHECKING
 
 from api_v2.models import APIDeployment
 
+if TYPE_CHECKING:
+    from global_api_deployment_key.models import GlobalApiDeploymentKey
+
 
 @dataclass
 class DeploymentExecutionDTO:
     """DTO for deployment execution viewset."""
 
     api: APIDeployment
     api_key: str
+    # The global API deployment key that authorized this execution, if any.
+    # Carrying the resolved key (not just a bool) preserves audit identity —
+    # "which named key authorized this?" — for downstream logging/incidents.
+    global_key: "GlobalApiDeploymentKey | None" = None
+
+    @property
+    def is_global_key(self) -> bool:
+        return self.global_key is not None

backend/api_v2/key_helper.py

@@ -1,6 +1,9 @@
+from __future__ import annotations
+
 import logging
 
 from django.core.exceptions import ValidationError
+from global_api_deployment_key.models import GlobalApiDeploymentKey
 from pipeline_v2.models import Pipeline
 from rest_framework.request import Request
 from workflow_manager.workflow_v2.workflow_helper import WorkflowHelper
@@ -61,6 +64,56 @@ def has_access(api_key: APIKey, instance: APIDeployment | Pipeline) -> bool:
             return api_key.pipeline == instance
         return False
 
+    @staticmethod
+    def validate_global_api_deployment_key(
+        api_key: str, api_deployment: APIDeployment
+    ) -> GlobalApiDeploymentKey:
+        """Validate a Global API Deployment Key for deployment execution.
+
+        Checks:
+        1. Key exists and is active
+        2. Key belongs to the same organization as the deployment
+        3. Key has access to the specific deployment (allow_all or listed)
+
+        Args:
+            api_key: The bearer token value
+            api_deployment: The API deployment being accessed
+
+        Returns:
+            GlobalApiDeploymentKey: The validated key instance
+
+        Raises:
+            UnauthorizedKey: If validation fails
+        """
+        try:
+            # UUIDField coerces/validates the key string via to_python, raising
+            # ValidationError for a malformed value — the same pattern
+            # ``validate_api_key`` relies on, so no manual uuid parsing is needed.
+            global_key = GlobalApiDeploymentKey.objects.get(key=api_key, is_active=True)
+        except (GlobalApiDeploymentKey.DoesNotExist, ValidationError):
+            # Unknown, inactive, or malformed key. Log the reason for
+            # observability; the client still gets a generic 401 (we don't
+            # leak which condition failed).
+            logger.warning(
+                "Global API key rejected (unknown/inactive/malformed) for "
+                "deployment %s (key ...%s).",
+                api_deployment.id,
+                str(api_key)[-4:],
+            )
+            raise UnauthorizedKey() from None
+
+        if not global_key.has_access_to_deployment(api_deployment):
+            logger.warning(
+                "Global API key '%s' (%s) rejected: no access to deployment %s "
+                "(out of scope or different organization).",
+                global_key.name,
+                global_key.id,
+                api_deployment.id,
+            )
+            raise UnauthorizedKey()
+
+        return global_key
+
     @staticmethod
     def validate_workflow_exists(workflow_id: str) -> None:
         """Validate that the specified workflow_id exists in the Workflow

backend/api_v2/serializers.py

@@ -399,6 +399,7 @@ def validate_llm_profile_id(self, value):
         # Get context from serializer
         api = self.context.get("api")
         api_key = self.context.get("api_key")
+        is_global_key = self.context.get("is_global_key", False)
 
         if not api or not api_key:
             raise ValidationError("Unable to validate LLM profile ownership")
@@ -409,6 +410,23 @@ def validate_llm_profile_id(self, value):
         except ProfileManager.DoesNotExist:
             raise ValidationError("Profile not found")
 
+        # Global API Keys are org-level (not tied to a single user), so the
+        # per-user ownership check below does not apply. We must still confirm
+        # the profile belongs to the same organization as the deployment,
+        # otherwise a caller could reference another org's profile by UUID.
+        # ``ProfileManager.objects`` is not org-scoped by default, so this
+        # check is load-bearing, not merely defense-in-depth.
+        if is_global_key:
+            profile_org_id = (
+                profile.prompt_studio_tool.organization_id
+                if profile.prompt_studio_tool_id
+                else None
+            )
+            if profile_org_id != api.organization_id:
+                # Generic error avoids confirming another org's profile exists.
+                raise ValidationError("Profile not found")
+            return value
+
         # Get the specific API key being used
         try:
             active_api_key = api.api_keys.get(api_key=api_key, is_active=True)

backend/backend/settings/base.py

@@ -390,6 +390,7 @@ def filter(self, record):
     "configuration",
     "dashboard_metrics",
     "platform_api",
+    "global_api_deployment_key",
 )
 TENANT_APPS = []
 

backend/backend/urls_v2.py

@@ -65,4 +65,5 @@
     path("execution/", include("workflow_manager.file_execution.urls")),
     path("metrics/", include("dashboard_metrics.urls")),
     path("platform-api/", include("platform_api.urls")),
+    path("global-api-deployment/", include("global_api_deployment_key.urls")),
 ]

backend/global_api_deployment_key/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class GlobalApiDeploymentKeyConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "global_api_deployment_key"

backend/global_api_deployment_key/migrations/0001_initial.py

@@ -0,0 +1,94 @@
+# Generated by Django 4.2.1 on 2026-03-25 18:22
+
+import uuid
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    initial = True
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ("account_v2", "0004_user_is_service_account"),
+        ("api_v2", "0003_add_organization_rate_limit"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="GlobalApiDeploymentKey",
+            fields=[
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("modified_at", models.DateTimeField(auto_now=True)),
+                (
+                    "id",
+                    models.UUIDField(
+                        default=uuid.uuid4,
+                        editable=False,
+                        primary_key=True,
+                        serialize=False,
+                    ),
+                ),
+                ("name", models.CharField(max_length=128)),
+                ("description", models.CharField(max_length=512)),
+                ("key", models.UUIDField(default=uuid.uuid4, unique=True)),
+                ("is_active", models.BooleanField(default=True)),
+                (
+                    "allow_all_deployments",
+                    models.BooleanField(
+                        db_comment="If True, this key can authenticate any API deployment in the org",
+                        default=False,
+                    ),
+                ),
+                (
+                    "api_deployments",
+                    models.ManyToManyField(
+                        blank=True,
+                        related_name="global_api_deployment_keys",
+                        to="api_v2.apideployment",
+                    ),
+                ),
+                (
+                    "created_by",
+                    models.ForeignKey(
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        related_name="global_api_deployment_keys_created",
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+                (
+                    "modified_by",
+                    models.ForeignKey(
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        related_name="+",
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+                (
+                    "organization",
+                    models.ForeignKey(
+                        blank=True,
+                        db_comment="Foreign key reference to the Organization model.",
+                        editable=False,
+                        null=True,
+                        on_delete=django.db.models.deletion.CASCADE,
+                        to="account_v2.organization",
+                    ),
+                ),
+            ],
+            options={
+                "db_table": "global_api_deployment_key",
+            },
+        ),
+        migrations.AddConstraint(
+            model_name="globalapideploymentkey",
+            constraint=models.UniqueConstraint(
+                fields=("name", "organization"),
+                name="unique_global_api_deployment_key_name_per_org",
+            ),
+        ),
+    ]