fix: 处理组件获取connection token获取不到的问题 (#8629)

* fix: 处理组件获取connection token获取不到的问题

* fix: ViewSet 显示获取资源用户

* fix: ViewSet 显示获取资源用户

Co-authored-by: Jiangjie.Bai <bugatti_it@163.com>

apps/authentication/api/connection_token.py

@@ -1,3 +1,4 @@
+import abc
 import os
 import json
 import base64
@@ -34,9 +35,12 @@ class ConnectionTokenMixin:
         if not is_valid:
             raise PermissionDenied(error)
 
-    @staticmethod
-    def get_request_resources(serializer):
-        user = serializer.validated_data.get('user')
+    @abc.abstractmethod
+    def get_request_resource_user(self, serializer):
+        raise NotImplementedError
+
+    def get_request_resources(self, serializer):
+        user = self.get_request_resource_user(serializer)
         asset = serializer.validated_data.get('asset')
         application = serializer.validated_data.get('application')
         system_user = serializer.validated_data.get('system_user')
@@ -226,6 +230,17 @@ class ConnectionTokenViewSet(ConnectionTokenMixin, RootOrgViewMixin, JMSModelVie
     def get_queryset(self):
         return ConnectionToken.objects.filter(user=self.request.user)
 
+    def get_request_resource_user(self, serializer):
+        return self.request.user
+
+    def get_object(self):
+        if self.request.user.is_service_account:
+            # TODO: 组件获取 token 详情，将来放在 Super-connection-token API 中
+            obj = get_object_or_404(ConnectionToken, pk=self.kwargs.get('pk'))
+        else:
+            obj = super(ConnectionTokenViewSet, self).get_object()
+        return obj
+
     def create_connection_token(self):
         data = self.request.query_params if self.request.method == 'GET' else self.request.data
         serializer = self.get_serializer(data=data)
@@ -293,6 +308,9 @@ class SuperConnectionTokenViewSet(ConnectionTokenViewSet):
         'renewal': 'authentication.add_superconnectiontoken'
     }
 
+    def get_request_resource_user(self, serializer):
+        return serializer.validated_data.get('user')
+
     @action(methods=['PATCH'], detail=False)
     def renewal(self, request, *args, **kwargs):
         from common.utils.timezone import as_current_tz