UN-3853 [FIX] Attribute platform-key-created resources to the API key's creator

backend/adapter_processor_v2/views.py

@@ -15,7 +15,6 @@
     IsOwnerOrSharedUserOrSharedToOrg,
 )
 from permissions.resource_share_views import ResourceShareManagementMixin
-from permissions.roles import ResourceRole
 from plugins import get_plugin
 from rest_framework import status
 from rest_framework.decorators import action
@@ -271,9 +270,7 @@ def create(self, request: Any) -> Response:
             instance = serializer.save(organization=UserContext.get_organization())
             # ``created_by`` is audit-only; the creator's access flows through
             # an OWNER membership row (UN-2202 co-owners).
-            instance.memberships.get_or_create(
-                user_id=request.user.id, defaults={"role": ResourceRole.OWNER}
-            )
+            instance.grant_owner(request.user)
             organization_member = OrganizationMemberService.get_user_by_id(
                 request.user.id
             )

backend/api_v2/api_deployment_views.py

@@ -9,7 +9,6 @@
 from permissions.membership_views import OwnerManagementMixin
 from permissions.permission import IsOwner, IsOwnerOrSharedUserOrSharedToOrg
 from permissions.resource_share_views import ResourceShareManagementMixin
-from permissions.roles import ResourceRole
 from platform_api.openapi_schema import PlatformKeyAutoSchema
 from plugins import get_plugin
 from prompt_studio.prompt_studio_registry_v2.models import PromptStudioRegistry
@@ -360,9 +359,7 @@ def create(
         self.perform_create(serializer)
         # ``created_by`` is audit-only; the creator's access flows through an
         # OWNER membership row (UN-2202 co-owners).
-        serializer.instance.memberships.get_or_create(
-            user_id=request.user.id, defaults={"role": ResourceRole.OWNER}
-        )
+        serializer.instance.grant_owner(request.user)
         api_key = DeploymentHelper.create_api_key(serializer=serializer, request=request)
         response_serializer = DeploymentResponseSerializer(
             {"api_key": api_key.api_key, **serializer.data}
@@ -406,8 +403,11 @@ def by_prompt_studio_tool(self, request: Request) -> Response:
             # Get API deployments for these workflows the user can access —
             # ``created_by`` is audit-only; access flows through memberships,
             # sharing, and the admin/SA bypasses (UN-2202).
-            deployments = APIDeployment.objects.for_user(request.user).filter(
-                workflow_id__in=workflow_ids
+            deployments = (
+                APIDeployment.objects.for_user(request.user)
+                .select_related("created_by")
+                .prefetch_related("memberships__user")
+                .filter(workflow_id__in=workflow_ids)
             )
 
             serializer = APIDeploymentListSerializer(deployments, many=True)

backend/api_v2/serializers.py

@@ -510,6 +510,7 @@ class APIDeploymentListSerializer(ModelSerializer):
     last_run_time = SerializerMethodField()
     is_owner = SerializerMethodField()
     co_owners_count = SerializerMethodField()
+    owner_emails = SerializerMethodField()
 
     class Meta:
         model = APIDeployment
@@ -529,6 +530,7 @@ class Meta:
             "last_run_time",
             "is_owner",
             "co_owners_count",
+            "owner_emails",
         ]
 
     def get_created_by_email(self, obj) -> str | None:
@@ -542,6 +544,12 @@ def get_is_owner(self, obj) -> bool:
     def get_co_owners_count(self, obj) -> int:
         return obj.co_owners_count()
 
+    def get_owner_emails(self, obj) -> list[str]:
+        """Email of each owner, earliest first. Empty if none is a person."""
+        # Published field: APIDeploymentSummary inherits it, so it also
+        # reaches platform-key callers.
+        return obj.owner_emails()
+
     # Both read the list view's annotations when they are there, and fall back
     # to a query for the callers that serialize a plain queryset. A deployment
     # that has never run annotates to `None`, so absence is what decides, not

backend/connector_v2/views.py

@@ -11,7 +11,6 @@
 from permissions.membership_views import OwnerManagementMixin
 from permissions.permission import IsOwner, IsOwnerOrSharedUserOrSharedToOrg
 from permissions.resource_share_views import ResourceShareManagementMixin
-from permissions.roles import ResourceRole
 from plugins import get_plugin
 from rest_framework import status, viewsets
 from rest_framework.decorators import action
@@ -258,9 +257,7 @@ def create(self, request: Any) -> Response:
             )
         # ``created_by`` is audit-only; the creator's access flows through an
         # OWNER membership row (UN-2202 co-owners).
-        serializer.instance.memberships.get_or_create(
-            user_id=request.user.id, defaults={"role": ResourceRole.OWNER}
-        )
+        serializer.instance.grant_owner(request.user)
         headers = self.get_success_headers(serializer.data)
         return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)
 

backend/permissions/models.py

@@ -76,3 +76,12 @@ def is_owner(self, user: Any) -> bool:
             m.user_id == user.id and m.role == ResourceRole.OWNER
             for m in self.memberships.all()  # type: ignore[attr-defined]
         )
+
+    def grant_owner(self, user: Any) -> None:
+        """Grant OWNER on create, resolving a platform key to its creator."""
+        # Lazy: platform_api.services imports models at import time.
+        from platform_api.services import owner_user_for
+
+        self.memberships.get_or_create(  # type: ignore[attr-defined]
+            user=owner_user_for(user), defaults={"role": ResourceRole.OWNER}
+        )

backend/permissions/permission.py

@@ -282,6 +282,8 @@ def has_object_permission(
     ) -> bool:
         if obj.is_friction_less:
             return True
+        if _is_service_account(request):
+            return True
         if _is_resource_owner(request.user, obj):
             return True
         return _is_organization_admin(request)

backend/pipeline_v2/serializers/crud.py

@@ -32,6 +32,7 @@ class PipelineSerializer(IntegrityErrorMixin, AuditSerializer):
     next_run_time = SerializerMethodField()
     is_owner = SerializerMethodField()
     co_owners_count = SerializerMethodField()
+    owner_emails = SerializerMethodField()
     # ``shared_groups`` is no longer an M2M on Pipeline — declare it
     # explicitly so ``fields = "__all__"`` continues to expose it. Share
     # mutations go through ``POST /pipeline/{id}/share/`` (UN-2977 plan §B).
@@ -224,6 +225,11 @@ def get_is_owner(self, obj) -> bool:
     def get_co_owners_count(self, obj) -> int:
         return obj.co_owners_count()
 
+    def get_owner_emails(self, obj) -> list[str]:
+        # Names the actual owner in "Owned By"; ``created_by`` is audit-only
+        # (UN-2202) and stays the service account on platform-key creates.
+        return obj.owner_emails()
+
     def get_last_5_run_statuses(self, instance: Pipeline) -> list[dict]:
         """Fetch the last 5 execution statuses with timestamps for this pipeline."""
         return WorkflowExecution.get_last_run_statuses(instance.id, limit=5)

backend/pipeline_v2/views.py

@@ -12,7 +12,6 @@
 from permissions.membership_views import OwnerManagementMixin
 from permissions.permission import IsOwner, IsOwnerOrSharedUserOrSharedToOrg
 from permissions.resource_share_views import ResourceShareManagementMixin
-from permissions.roles import ResourceRole
 from plugins import get_plugin
 from rest_framework import serializers, status, viewsets
 from rest_framework.decorators import action
@@ -158,9 +157,7 @@ def create(self, request: Request) -> Response:
                 pipeline_instance = serializer.save()
                 # Grant before the API key so the creator's access is committed
                 # with the row itself, matching api_deployment_views.create().
-                pipeline_instance.memberships.get_or_create(
-                    user_id=request.user.id, defaults={"role": ResourceRole.OWNER}
-                )
+                pipeline_instance.grant_owner(request.user)
                 # Create API key using the created instance
                 KeyHelper.create_api_key(pipeline_instance, request)
         except IntegrityError:

backend/platform_api/services.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 import re
 import uuid as _uuid
 from typing import TYPE_CHECKING
@@ -15,6 +16,15 @@
 
     from platform_api.models import PlatformApiKey
 
+logger = logging.getLogger(__name__)
+
+# Distinguishes "caller said nothing" from an explicit None in transfer_ownership.
+_SAME_AS_TO_USER: object = object()
+
+# Reserved domain for service-account addresses. The frontend matches on it to
+# label an ownerless resource "Platform key" instead of naming a machine.
+SERVICE_ACCOUNT_EMAIL_DOMAIN = "platform.internal"
+
 # Business app labels whose models may carry created_by / membership rows.
 # Restricts transfer_ownership to avoid scanning Django built-in and third-party models.
 _BUSINESS_APP_LABELS = {
@@ -45,7 +55,7 @@ def create_api_user_for_key(
         name_slug = _slugify_for_email(platform_api_key.name)
         user = User(
             username=f"svc-{name_slug}-{uid[:8]}",
-            email=f"{name_slug}-{uid[:8]}@platform.internal",
+            email=f"{name_slug}-{uid[:8]}@{SERVICE_ACCOUNT_EMAIL_DOMAIN}",
             user_id=uid,
             is_service_account=True,
         )
@@ -63,6 +73,63 @@ def create_api_user_for_key(
     return user
 
 
+def live_key_creator(platform_api_key: PlatformApiKey) -> User | None:
+    """The key's creator if they still belong to the key's organization.
+
+    ``_is_resource_owner`` grants on any surviving OWNER row without checking
+    live membership, which is why ``cleanup_user_org_access`` purges those rows
+    when a user leaves. Handing an ex-member a fresh row -- at create, on key
+    deletion, or in a backfill -- reopens that rejoin backdoor, so every path
+    that names a successor asks this one question.
+    """
+    creator = platform_api_key.created_by
+    if creator is None:
+        return None
+    # ``_base_manager`` because the default manager is org-scoped by
+    # ``UserContext``, which is None outside a request — an empty result would
+    # silently strip every resource of its owner. The org is filtered here.
+    if not OrganizationMember._base_manager.filter(
+        user=creator, organization=platform_api_key.organization
+    ).exists():
+        return None
+    return creator
+
+
+def owner_user_for(user: User) -> User:
+    """Resolve the human who should own a resource created by ``user``.
+
+    Service accounts are filtered out of every owner surface, so granting to
+    one leaves no human owner; attribute it to the key's live creator instead.
+    Returns ``user`` unchanged for a normal session, or when no live creator
+    can be named -- the UI then labels the resource "Platform key".
+    """
+    if not getattr(user, "is_service_account", False):
+        return user
+
+    # Imported here so the module keeps its models import behind TYPE_CHECKING.
+    from platform_api.models import PlatformApiKey
+
+    key = (
+        PlatformApiKey.objects.filter(api_user=user)
+        .select_related("created_by", "organization")
+        .first()
+    )
+    if key is None:
+        logger.warning(
+            "Service account %s backs no platform key; resource gets no human owner",
+            user.id,
+        )
+        return user
+    creator = live_key_creator(key)
+    if creator is None:
+        logger.warning(
+            "Platform key %s has no live creator; resource gets no human owner",
+            key.id,
+        )
+        return user
+    return creator
+
+
 def _get_user_fk_fields(model: type) -> list[str]:
     """Return names of all ForeignKey fields pointing to User."""
     return [
@@ -150,7 +217,9 @@ def _transfer_membership_rows(from_user: User, to_user: User) -> None:
         row.delete()
 
 
-def transfer_ownership(from_user: User, to_user: User | None) -> None:
+def transfer_ownership(
+    from_user: User, to_user: User | None, membership_to: User | None = _SAME_AS_TO_USER
+) -> None:
     """Transfer all resource ownership from one user to another.
 
     Replaces from_user with to_user across business models:
@@ -159,25 +228,41 @@ def transfer_ownership(from_user: User, to_user: User | None) -> None:
     - OWNER/VIEWER membership rows (custom-through, UN-2202) — re-pointed,
       reconciling by role precedence when to_user already holds a row so the
       resource keeps an owner.
+
+    ``membership_to`` splits the two halves. Audit fields may follow a user who
+    has left the org -- nulling them has no security value and breaks deletes
+    that dereference ``created_by`` -- while an OWNER row may not.
     """
-    if not to_user:
-        return
+    if membership_to is _SAME_AS_TO_USER:
+        membership_to = to_user
 
     with transaction.atomic():
-        for model in apps.get_models():
-            if model._meta.app_label not in _BUSINESS_APP_LABELS:
-                continue
-            _transfer_model_ownership(model, from_user, to_user)
+        if to_user:
+            for model in apps.get_models():
+                if model._meta.app_label not in _BUSINESS_APP_LABELS:
+                    continue
+                _transfer_model_ownership(model, from_user, to_user)
         # Memberships live in one polymorphic table — transfer once, not per model.
-        _transfer_membership_rows(from_user, to_user)
+        if membership_to:
+            _transfer_membership_rows(from_user, membership_to)
 
 
 def delete_api_user_for_key(platform_api_key: PlatformApiKey) -> None:
-    """Transfer ownership to key creator, then delete the service account."""
+    """Transfer ownership to the key's creator, then delete the service account.
+
+    Audit fields follow ``created_by`` even if they have left the org: deleting
+    the account is ``SET_NULL`` on those FKs, and a null ``created_by`` breaks
+    callers that dereference it. An OWNER row is a live-membership question, so
+    it goes only to a creator ``live_key_creator`` still admits.
+    """
     api_user = platform_api_key.api_user
     if not api_user:
         return
 
     with transaction.atomic():
-        transfer_ownership(from_user=api_user, to_user=platform_api_key.created_by)
+        transfer_ownership(
+            from_user=api_user,
+            to_user=platform_api_key.created_by,
+            membership_to=live_key_creator(platform_api_key),
+        )
         api_user.delete()