[FEAT] Platform API gaps for org-to-org clone: service-account group management + name filters

backend/api_v2/api_deployment_views.py

@@ -269,7 +269,7 @@ def get_queryset(self) -> QuerySet | None:
         if search:
             queryset = queryset.filter(display_name__icontains=search)
 
-        # Exact-match api_name filter for migration SDK's get-or-create flow.
+        # Exact-match lookup (distinct from the icontains search above).
         api_name = self.request.query_params.get("api_name")
         if api_name:
             queryset = queryset.filter(api_name=api_name)

backend/connector_v2/serializers.py

@@ -36,9 +36,7 @@ class Meta:
         fields = "__all__"
         extra_kwargs = {
             "connector_name": {"required": False},
-            # connector_mode is overridden in to_representation from the catalog,
-            # so any client-supplied value is silently discarded — mark it read_only
-            # to make that explicit (and to keep DRF OPTIONS schema honest).
+            # connector_mode is derived from the catalog in to_representation.
             "connector_mode": {"read_only": True},
             "shared_users": {"read_only": True},
             "shared_to_org": {"read_only": True},

backend/pipeline_v2/views.py

@@ -78,7 +78,7 @@ def get_queryset(self) -> QuerySet:
         if search:
             queryset = queryset.filter(pipeline_name__icontains=search)
 
-        # Exact-match name filter for migration SDK's get-or-create flow.
+        # Exact-match lookup (distinct from the icontains search above).
         pipeline_name = self.request.query_params.get(PK.PIPELINE_NAME)
         if pipeline_name:
             queryset = queryset.filter(pipeline_name=pipeline_name)

backend/tenant_account_v2/group_serializers.py

@@ -86,11 +86,14 @@ class GroupMemberSerializer(serializers.ModelSerializer):
     user_id = serializers.IntegerField(source="user.id", read_only=True)
     email = serializers.CharField(source="user.email", read_only=True)
     display_name = serializers.SerializerMethodField()
+    is_service_account = serializers.BooleanField(
+        source="user.is_service_account", read_only=True
+    )
     joined_at = serializers.DateTimeField(source="created_at", read_only=True)
 
     class Meta:
         model = GroupMembership
-        fields = ("user_id", "email", "display_name", "joined_at")
+        fields = ("user_id", "email", "display_name", "is_service_account", "joined_at")
 
     def get_display_name(self, obj: GroupMembership) -> str:
         user = obj.user

backend/tenant_account_v2/group_views.py

@@ -48,8 +48,26 @@ def _is_org_admin(request: Request) -> bool:
     return is_org_admin(request.user)
 
 
+def _is_service_account(request: Request) -> bool:
+    """Whether the request user is a platform-key service account."""
+    return getattr(request.user, "is_service_account", False)
+
+
+def _is_admin_or_service_account(request: Request) -> bool:
+    """Write gate for group management.
+
+    Service accounts bypass authorization — they already bypass other access
+    controls (see ShareAuthorizationService) and platform-key automation needs
+    to manage groups and memberships.
+    """
+    return _is_service_account(request) or _is_org_admin(request)
+
+
 class IsOrgAdminForWrite(BasePermission):
-    """Read for any authenticated org member; write for org admins only."""
+    """Read for any authenticated org member; write for org admins only.
+
+    Service accounts (platform-key auth) are also allowed to write.
+    """
 
     message = "Only organization admins can manage groups."
 
@@ -58,7 +76,7 @@ def has_permission(self, request: Request, view: Any) -> bool:
             return False
         if request.method in ("GET", "HEAD", "OPTIONS"):
             return True
-        return _is_org_admin(request)
+        return _is_admin_or_service_account(request)
 
 
 class OrganizationGroupViewSet(viewsets.ModelViewSet):
@@ -86,7 +104,7 @@ def get_queryset(self) -> QuerySet[OrganizationGroup]:
     def list(self, request: Request, *args: Any, **kwargs: Any) -> Response:
         organization = _current_organization()
         member_filter = request.query_params.get("member")
-        is_admin = _is_org_admin(request)
+        is_admin = _is_admin_or_service_account(request)
 
         if member_filter == "me":
             qs = list_groups_with_member_counts(
@@ -128,7 +146,7 @@ def members(self, request: Request, pk: str | None = None) -> Response:
             return Response(data)
 
         # POST → bulk add
-        if not _is_org_admin(request):
+        if not _is_admin_or_service_account(request):
             raise PermissionDenied(IsOrgAdminForWrite.message)
         serializer = GroupMemberAddSerializer(data=request.data, context={"group": group})
         serializer.is_valid(raise_exception=True)
@@ -150,7 +168,7 @@ def members(self, request: Request, pk: str | None = None) -> Response:
     def remove_member(
         self, request: Request, pk: str | None = None, user_id: str | None = None
     ) -> Response:
-        if not _is_org_admin(request):
+        if not _is_admin_or_service_account(request):
             raise PermissionDenied(IsOrgAdminForWrite.message)
         try:
             user_id_int = int(user_id)  # type: ignore[arg-type]
@@ -169,7 +187,7 @@ def resources(self, request: Request, pk: str | None = None) -> Response:
         # Admin-only: this is the delete blast-radius view. Leaving it open to
         # any org member would leak names/UUIDs of resources shared with groups
         # they are not in (org admin has no implicit resource access).
-        if not _is_org_admin(request):
+        if not _is_admin_or_service_account(request):
             raise PermissionDenied(IsOrgAdminForWrite.message)
         group = self._get_group_or_404(pk)
         payload = _collect_resources_shared_with_group(group)

backend/tenant_account_v2/organization_member_service.py

@@ -60,7 +60,9 @@ def get_user_by_id(id: str) -> OrganizationMember | None:
 
     @staticmethod
     def get_members() -> list[OrganizationMember]:
-        return OrganizationMember.objects.filter(user__is_service_account=False)
+        return OrganizationMember.objects.select_related("user").filter(
+            user__is_service_account=False
+        )
 
     @staticmethod
     def get_members_by_role(role: str) -> list[OrganizationMember]:

backend/tenant_account_v2/serializer.py

@@ -30,10 +30,13 @@ class OrganizationMemberSerializer(serializers.ModelSerializer):
     email = serializers.CharField(source="user.email", read_only=True)
     id = serializers.CharField(source="user.id", read_only=True)
     is_admin = serializers.SerializerMethodField()
+    is_service_account = serializers.BooleanField(
+        source="user.is_service_account", read_only=True
+    )
 
     class Meta:
         model = OrganizationMember
-        fields = ("id", "email", "role", "is_admin")
+        fields = ("id", "email", "role", "is_admin", "is_service_account")
 
     def get_is_admin(self, obj: OrganizationMember) -> bool:
         # Admin determination is auth-plugin specific (OSS "admin" vs Auth0

backend/tenant_account_v2/tests.py

@@ -18,9 +18,11 @@
 from django.core.exceptions import FieldDoesNotExist
 from django.test import TestCase
 from rest_framework.exceptions import PermissionDenied
+from rest_framework.test import APIRequestFactory, force_authenticate
 from utils.user_context import UserContext
 from workflow_manager.workflow_v2.models.workflow import Workflow
 
+from tenant_account_v2.group_views import OrganizationGroupViewSet
 from tenant_account_v2.models import (
     GroupMembership,
     OrganizationGroup,
@@ -33,6 +35,7 @@
     get_resource_share_groups,
     set_resource_share_groups,
 )
+from tenant_account_v2.users_view import OrganizationUserViewSet
 
 
 def _make_user(email: str, **kwargs) -> User:
@@ -250,6 +253,144 @@ def test_resource_delete_purges_group_shares(self) -> None:
         )
 
 
+class GroupViewSetServiceAccountTests(GroupSharingTestBase):
+    """Service accounts (platform-key auth) may manage groups; plain members may not.
+
+    Admin resolution is patched to always-False so any write that succeeds
+    does so via the service-account allowance, not an accidental admin role.
+    """
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.svc = _make_user("svc@example.com", is_service_account=True)
+        self.factory = APIRequestFactory()
+        patcher = patch(
+            "tenant_account_v2.sharing_helpers.is_org_admin", return_value=False
+        )
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def _call(self, actions, method, user, data=None, **url_kwargs):
+        view = OrganizationGroupViewSet.as_view(actions)
+        request = getattr(self.factory, method)("/groups/", data, format="json")
+        force_authenticate(request, user=user)
+        return view(request, **url_kwargs)
+
+    def test_service_account_can_create_group(self) -> None:
+        response = self._call(
+            {"post": "create"}, "post", self.svc, data={"name": "Cloned"}
+        )
+        self.assertEqual(response.status_code, 201)
+        self.assertTrue(
+            OrganizationGroup.objects.filter(
+                organization=self.org, name="Cloned"
+            ).exists()
+        )
+
+    def test_service_account_can_add_members(self) -> None:
+        response = self._call(
+            {"post": "members"},
+            "post",
+            self.svc,
+            data={"user_ids": [self.outsider.id]},
+            pk=str(self.group.pk),
+        )
+        self.assertEqual(response.status_code, 201)
+        self.assertTrue(
+            GroupMembership.objects.filter(group=self.group, user=self.outsider).exists()
+        )
+
+    def test_service_account_can_remove_member(self) -> None:
+        response = self._call(
+            {"delete": "remove_member"},
+            "delete",
+            self.svc,
+            pk=str(self.group.pk),
+            user_id=str(self.member.id),
+        )
+        self.assertEqual(response.status_code, 204)
+        self.assertFalse(
+            GroupMembership.objects.filter(group=self.group, user=self.member).exists()
+        )
+
+    def test_service_account_can_list_group_resources(self) -> None:
+        set_resource_share_groups(self.workflow, [self.group.id])
+        response = self._call(
+            {"get": "resources"}, "get", self.svc, pk=str(self.group.pk)
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(
+            str(self.workflow.id), [item["resource_id"] for item in response.data]
+        )
+
+    def test_non_admin_member_still_403_on_writes(self) -> None:
+        cases = [
+            ({"post": "create"}, "post", {"name": "Nope"}, {}),
+            (
+                {"post": "members"},
+                "post",
+                {"user_ids": [self.outsider.id]},
+                {"pk": str(self.group.pk)},
+            ),
+            (
+                {"delete": "remove_member"},
+                "delete",
+                None,
+                {"pk": str(self.group.pk), "user_id": str(self.member.id)},
+            ),
+            ({"get": "resources"}, "get", None, {"pk": str(self.group.pk)}),
+        ]
+        for actions, method, data, url_kwargs in cases:
+            with self.subTest(action=next(iter(actions.values()))):
+                response = self._call(
+                    actions, method, self.member, data=data, **url_kwargs
+                )
+                self.assertEqual(response.status_code, 403)
+
+    def test_members_listing_flags_service_accounts(self) -> None:
+        GroupMembership.objects.create(group=self.group, user=self.svc)
+        response = self._call({"get": "members"}, "get", self.svc, pk=str(self.group.pk))
+        self.assertEqual(response.status_code, 200)
+        flags = {row["email"]: row["is_service_account"] for row in response.data}
+        self.assertTrue(flags[self.svc.email])
+        self.assertFalse(flags[self.member.email])
+
+    def test_non_admin_member_cannot_delete_group(self) -> None:
+        response = self._call(
+            {"delete": "destroy"}, "delete", self.member, pk=str(self.group.pk)
+        )
+        self.assertEqual(response.status_code, 403)
+        response = self._call(
+            {"delete": "destroy"}, "delete", self.svc, pk=str(self.group.pk)
+        )
+        self.assertEqual(response.status_code, 204)
+        self.assertFalse(OrganizationGroup.objects.filter(pk=self.group.pk).exists())
+
+
+class OrganizationMembersListingTests(GroupSharingTestBase):
+    """GET /users/ excludes service accounts; human rows carry the flag."""
+
+    def test_users_listing_excludes_service_accounts(self) -> None:
+        svc = _make_user("svc@example.com", is_service_account=True)
+        OrganizationMember.objects.create(organization=self.org, user=svc, role="user")
+
+        request = APIRequestFactory().get("/users/")
+        force_authenticate(request, user=self.admin)
+        view = OrganizationUserViewSet.as_view({"get": "get_organization_members"})
+        with patch(
+            "tenant_account_v2.users_view.UserSessionUtils.get_organization_id",
+            return_value=self.org.organization_id,
+        ):
+            response = view(request)
+
+        self.assertEqual(response.status_code, 200)
+        flags = {
+            row["email"]: row["is_service_account"] for row in response.data["members"]
+        }
+        self.assertNotIn(svc.email, flags)
+        self.assertFalse(flags[self.member.email])
+
+
 class ShareableResourceRegistryTests(TestCase):
     """Each installed descriptor must resolve and expose its declared fields."""