feat(idp): plugin config meta retrieve api

src/bk-user/bkuser/apis/web/data_source/serializers.py

@@ -305,7 +305,7 @@ def validate(self, attrs):
             except PDValidationError as e:
                 raise ValidationError(_("密码规则配置不合法: {}").format(stringify_pydantic_error(e)))
         else:
-            attrs["password_rule"] = get_default_plugin_cfg(DataSourcePluginEnum.LOCAL).password_rule.to_rule()
+            attrs["password_rule"] = get_default_plugin_cfg(DataSourcePluginEnum.LOCAL).password_rule.to_rule()  # type: ignore
 
         return attrs
 

src/bk-user/bkuser/apis/web/data_source/views.py

@@ -75,6 +75,11 @@ def get(self, request, *args, **kwargs):
 
 
 class DataSourcePluginDefaultConfigApi(generics.RetrieveAPIView):
+    permission_classes = [IsAuthenticated, perm_class(PermAction.MANAGE_TENANT)]
+
+    queryset = DataSourcePlugin.objects.all()
+    lookup_url_kwarg = "id"
+
     @swagger_auto_schema(
         tags=["data_source_plugin"],
         operation_description="数据源插件默认配置",
@@ -84,8 +89,9 @@ class DataSourcePluginDefaultConfigApi(generics.RetrieveAPIView):
         },
     )
     def get(self, request, *args, **kwargs):
+        instance = self.get_object()
         try:
-            config = get_default_plugin_cfg(kwargs["id"])
+            config = get_default_plugin_cfg(instance.id)
         except NotImplementedError:
             raise error_codes.DATA_SOURCE_PLUGIN_NOT_DEFAULT_CONFIG
 

src/bk-user/bkuser/apis/web/idp/swagger.py → src/bk-user/bkuser/apis/web/idp/schema.py

@@ -8,17 +8,26 @@
 an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 specific language governing permissions and limitations under the License.
 """
-from typing import Dict
+import json
+from typing import Any, Dict
 
+import jsonref
 from drf_yasg import openapi
 
-from bkuser.idp_plugins.base import list_plugin_cls
+from bkuser.idp_plugins.base import get_plugin_cfg_cls, list_plugin_cls
 from bkuser.utils.pydantic import gen_openapi_schema
 
 
-def get_idp_plugin_cfg_schema_map() -> Dict[str, openapi.Schema]:
-    """获取认证插件配置类 JsonSchema 映射表"""
+def get_idp_plugin_cfg_openapi_schema_map() -> Dict[str, openapi.Schema]:
+    """获取认证插件配置类 JsonSchema 映射表，输出是以openapi schema格式"""
     return {
         f"plugin_config:{plugin_cls.id}": gen_openapi_schema(plugin_cls.config_class)
         for plugin_cls in list_plugin_cls()
     }
+
+
+def get_idp_plugin_cfg_json_schema(plugin_id: str) -> Dict[str, Any]:
+    """获取认证源插件配置类的 JsonSchema, without any jsonRef"""
+    json_schema = get_plugin_cfg_cls(plugin_id).model_json_schema()
+    # replace json refs
+    return jsonref.loads(json.dumps(json_schema))

src/bk-user/bkuser/apis/web/idp/serializers.py

@@ -32,6 +32,11 @@ class IdpPluginOutputSLZ(serializers.Serializer):
     logo = serializers.CharField(help_text="认证源插件 Logo")
 
 
+class IdpPluginConfigMetaRetrieveOutputSLZ(serializers.Serializer):
+    id = serializers.CharField(help_text="认证源插件唯一标识")
+    json_schema = serializers.JSONField(help_text="配置的JSON Schema")
+
+
 class IdpSearchInputSLZ(serializers.Serializer):
     keyword = serializers.CharField(help_text="搜索关键字", required=False)
 

src/bk-user/bkuser/apis/web/idp/urls.py

@@ -15,6 +15,12 @@
 urlpatterns = [
     # 认证源插件列表
     path("plugins/", views.IdpPluginListApi.as_view(), name="idp_plugin.list"),
+    # 认证源插件配置元数据
+    path(
+        "plugins/<str:id>/config-meta/",
+        views.IdpPluginConfigMetaRetrieveApi.as_view(),
+        name="idp_plugin_config_meta.retrieve",
+    ),
     # 认证源创建/获取列表
     path("", views.IdpListCreateApi.as_view(), name="idp.list_create"),
     # 认证源获取/更新

src/bk-user/bkuser/apis/web/idp/views.py

@@ -22,17 +22,18 @@
 from bkuser.apps.permission.permissions import perm_class
 from bkuser.common.error_codes import error_codes
 
+from .schema import get_idp_plugin_cfg_json_schema, get_idp_plugin_cfg_openapi_schema_map
 from .serializers import (
     IdpCreateInputSLZ,
     IdpCreateOutputSLZ,
     IdpPartialUpdateInputSLZ,
+    IdpPluginConfigMetaRetrieveOutputSLZ,
     IdpPluginOutputSLZ,
     IdpRetrieveOutputSLZ,
     IdpSearchInputSLZ,
     IdpSearchOutputSLZ,
     IdpUpdateInputSLZ,
 )
-from .swagger import get_idp_plugin_cfg_schema_map
 
 
 class IdpPluginListApi(generics.ListAPIView):
@@ -49,6 +50,30 @@ def get(self, request, *args, **kwargs):
         return self.list(request, *args, **kwargs)
 
 
+class IdpPluginConfigMetaRetrieveApi(generics.RetrieveAPIView):
+    permission_classes = [IsAuthenticated, perm_class(PermAction.MANAGE_TENANT)]
+
+    queryset = IdpPlugin.objects.all()
+    lookup_url_kwarg = "id"
+
+    @swagger_auto_schema(
+        tags=["idp_plugin"],
+        operation_description="认证源插件默认配置",
+        responses={status.HTTP_200_OK: IdpPluginConfigMetaRetrieveOutputSLZ()},
+    )
+    def get(self, request, *args, **kwargs):
+        instance = self.get_object()
+
+        try:
+            json_schema = get_idp_plugin_cfg_json_schema(instance.id)
+        except NotImplementedError:
+            raise error_codes.IDP_PLUGIN_NOT_LOAD
+
+        return Response(
+            IdpPluginConfigMetaRetrieveOutputSLZ(instance={"id": instance.id, "json_schema": json_schema}).data
+        )
+
+
 class IdpListCreateApi(CurrentUserTenantMixin, generics.ListCreateAPIView):
     permission_classes = [IsAuthenticated, perm_class(PermAction.MANAGE_TENANT)]
 
@@ -89,7 +114,7 @@ def get(self, request, *args, **kwargs):
         tags=["idp"],
         operation_description="新建认证源",
         request_body=IdpCreateInputSLZ(),
-        responses={status.HTTP_201_CREATED: IdpCreateOutputSLZ(), **get_idp_plugin_cfg_schema_map()},
+        responses={status.HTTP_201_CREATED: IdpCreateOutputSLZ(), **get_idp_plugin_cfg_openapi_schema_map()},
     )
     def post(self, request, *args, **kwargs):
         current_tenant_id = self.get_current_tenant_id()
@@ -128,7 +153,7 @@ def get_queryset(self):
         operation_description="认证源详情",
         responses={
             status.HTTP_200_OK: IdpRetrieveOutputSLZ(),
-            **get_idp_plugin_cfg_schema_map(),
+            **get_idp_plugin_cfg_openapi_schema_map(),
         },
     )
     def get(self, request, *args, **kwargs):
@@ -157,7 +182,7 @@ def patch(self, request, *args, **kwargs):
         tags=["idp"],
         operation_description="更新认证源",
         request_body=IdpUpdateInputSLZ(),
-        responses={status.HTTP_204_NO_CONTENT: "", **get_idp_plugin_cfg_schema_map()},
+        responses={status.HTTP_204_NO_CONTENT: "", **get_idp_plugin_cfg_openapi_schema_map()},
     )
     def put(self, request, *args, **kwargs):
         idp = self.get_object()

src/bk-user/bkuser/apps/permission/permissions.py

@@ -20,8 +20,8 @@
 
 from rest_framework.permissions import BasePermission
 
-from bkuser.apps.data_source.models import DataSource
-from bkuser.apps.idp.models import Idp
+from bkuser.apps.data_source.models import DataSource, DataSourcePlugin
+from bkuser.apps.idp.models import Idp, IdpPlugin
 from bkuser.apps.natural_user.models import DataSourceUserNaturalUserRelation
 from bkuser.apps.permission.constants import PermAction, UserRole
 from bkuser.apps.tenant.models import Tenant, TenantManager, TenantUser
@@ -47,7 +47,10 @@ def has_permission(self, request, view):
 
             return False
 
-        def has_object_permission(self, request, view, obj):
+        def has_object_permission(self, request, view, obj):  # noqa: C901
+            username = request.user.username
+            cur_tenant_id = request.user.get_property("tenant_id")
+
             if isinstance(obj, Tenant):
                 tenant_id = obj.id
             elif hasattr(obj, "tenant_id"):
@@ -60,12 +63,14 @@ def has_object_permission(self, request, view, obj):
                 tenant_id = obj.data_source.owner_tenant_id
             elif isinstance(obj, Idp):
                 tenant_id = obj.owner_tenant_id
+            elif isinstance(obj, (DataSourcePlugin, IdpPlugin)):
+                # 认证源插件和数据源插件的配置信息、默认配置等可能包含一些低敏感级别的信息，
+                # 所以需要确保用户可管理租户才可看到
+                tenant_id = cur_tenant_id
             else:
                 logger.exception("failed to get tenant id, obj: %s", obj)
                 return False
 
-            username = request.user.username
-            cur_tenant_id = request.user.get_property("tenant_id")
             if action == PermAction.MANAGE_PLATFORM:
                 return is_super_manager(tenant_id, username)
             if action == PermAction.MANAGE_TENANT:

src/bk-user/bkuser/common/error_codes.py

@@ -85,6 +85,7 @@ class ErrorCodes:
     CREATE_DATA_SOURCE_SYNC_TASK_FAILED = ErrorCode(_("创建数据源同步任务失败"))
 
     # 认证源
+    IDP_PLUGIN_NOT_LOAD = ErrorCode(_("认证源插件未加载"))
     CANNOT_UPDATE_IDP = ErrorCode(_("该认证源不允许更新配置"))
 
     # 租户

src/bk-user/tests/apis/web/data_source/test_data_source.py

@@ -106,7 +106,7 @@ def test_retrieve(self, api_client):
 
     def test_retrieve_not_exists(self, api_client):
         resp = api_client.get(reverse("data_source_plugin.default_config", args=["not_exists"]))
-        assert resp.status_code == status.HTTP_400_BAD_REQUEST
+        assert resp.status_code == status.HTTP_404_NOT_FOUND
 
 
 class TestDataSourceCreateApi: