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

@@ -31,6 +31,7 @@
     InactiveAPI,
     InvalidAPIRequest,
     PresignedURLFetchError,
+    UnauthorizedKey,
 )
 from api_v2.key_helper import KeyHelper
 from api_v2.models import APIDeployment, APIKey
@@ -59,31 +60,48 @@ 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)
+        is_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, is_global_key=is_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) -> bool:
+        """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:
+            bool: True if authenticated via Global API Deployment Key, False otherwise
 
         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 False
+        except UnauthorizedKey:
+            KeyHelper.validate_global_api_deployment_key(
+                api_key=api_key, api_deployment=api_deployment
+            )
+            return True
 
     @staticmethod
     def validate_and_get_workflow(workflow_id: str) -> Workflow:

backend/api_v2/dto.py

@@ -9,3 +9,4 @@ class DeploymentExecutionDTO:
 
     api: APIDeployment
     api_key: str
+    is_global_key: bool = False

backend/api_v2/key_helper.py

@@ -1,4 +1,8 @@
+from __future__ import annotations
+
 import logging
+import uuid as uuid_module
+from typing import TYPE_CHECKING
 
 from django.core.exceptions import ValidationError
 from pipeline_v2.models import Pipeline
@@ -7,6 +11,9 @@
 
 from api_v2.exceptions import UnauthorizedKey
 from api_v2.models import APIDeployment, APIKey
+
+if TYPE_CHECKING:
+    from global_api_deployment_key.models import GlobalApiDeploymentKey
 from api_v2.serializers import APIKeySerializer
 
 logger = logging.getLogger(__name__)
@@ -61,6 +68,46 @@ 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
+        """
+        from global_api_deployment_key.models import GlobalApiDeploymentKey
+
+        try:
+            key_uuid = uuid_module.UUID(api_key)
+        except (ValueError, AttributeError):
+            raise UnauthorizedKey()
+
+        try:
+            global_key = GlobalApiDeploymentKey.objects.select_related(
+                "organization"
+            ).get(key=key_uuid, is_active=True)
+        except GlobalApiDeploymentKey.DoesNotExist:
+            raise UnauthorizedKey()
+
+        if not global_key.has_access_to_deployment(api_deployment):
+            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

@@ -381,6 +381,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")
@@ -391,6 +392,10 @@ def validate_llm_profile_id(self, value):
         except ProfileManager.DoesNotExist:
             raise ValidationError("Profile not found")
 
+        # Global API Keys are org-level; skip per-user ownership check
+        if is_global_key:
+            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

@@ -345,6 +345,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.TextField(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.",
+                        default=None,
+                        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",
+            ),
+        ),
+    ]

backend/global_api_deployment_key/models.py

@@ -0,0 +1,58 @@
+import uuid
+
+from account_v2.models import User
+from api_v2.models import APIDeployment
+from django.db import models
+from utils.models.base_model import BaseModel
+from utils.models.organization_mixin import DefaultOrganizationMixin
+
+
+class GlobalApiDeploymentKey(DefaultOrganizationMixin, BaseModel):
+    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+    name = models.CharField(max_length=128)
+    description = models.TextField(max_length=512)
+    key = models.UUIDField(default=uuid.uuid4, unique=True)
+    is_active = models.BooleanField(default=True)
+    allow_all_deployments = models.BooleanField(
+        default=False,
+        db_comment="If True, this key can authenticate any API deployment in the org",
+    )
+    api_deployments = models.ManyToManyField(
+        APIDeployment,
+        blank=True,
+        related_name="global_api_deployment_keys",
+    )
+    created_by = models.ForeignKey(
+        User,
+        on_delete=models.SET_NULL,
+        null=True,
+        related_name="global_api_deployment_keys_created",
+    )
+    modified_by = models.ForeignKey(
+        User,
+        on_delete=models.SET_NULL,
+        null=True,
+        related_name="+",
+    )
+
+    class Meta:
+        db_table = "global_api_deployment_key"
+        constraints = [
+            models.UniqueConstraint(
+                fields=["name", "organization"],
+                name="unique_global_api_deployment_key_name_per_org",
+            ),
+        ]
+
+    def __str__(self):
+        return f"{self.name} ({self.organization})"
+
+    def has_access_to_deployment(self, api_deployment):
+        """Check if this key can authenticate the given API deployment."""
+        if not self.is_active:
+            return False
+        if self.organization_id != api_deployment.organization_id:
+            return False
+        if self.allow_all_deployments:
+            return True
+        return self.api_deployments.filter(id=api_deployment.id).exists()

backend/global_api_deployment_key/permissions.py

@@ -0,0 +1,3 @@
+from platform_api.permissions import IsOrganizationAdmin
+
+__all__ = ["IsOrganizationAdmin"]