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/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

@@ -21,11 +21,13 @@
 from bkuser.apps.permission.constants import PermAction
 from bkuser.apps.permission.permissions import perm_class
 from bkuser.common.error_codes import error_codes
+from bkuser.idp_plugins.base import get_plugin_cfg_cls
 
 from .serializers import (
     IdpCreateInputSLZ,
     IdpCreateOutputSLZ,
     IdpPartialUpdateInputSLZ,
+    IdpPluginConfigMetaRetrieveOutputSLZ,
     IdpPluginOutputSLZ,
     IdpRetrieveOutputSLZ,
     IdpSearchInputSLZ,
@@ -49,6 +51,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_plugin_cfg_cls(instance.id).model_json_schema()
+        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)]
 

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:

src/bk-user/tests/apis/web/idp/test_idp.py

@@ -13,6 +13,7 @@
 import pytest
 from bkuser.apps.data_source.models import DataSource
 from bkuser.apps.idp.models import Idp, IdpPlugin
+from bkuser.idp_plugins.base import list_plugin_cls
 from bkuser.idp_plugins.constants import BuiltinIdpPluginEnum
 from django.urls import reverse
 from rest_framework import status
@@ -79,6 +80,14 @@ def test_list(self, api_client):
         assert BuiltinIdpPluginEnum.LOCAL in [i["id"] for i in resp.data]
 
 
+class TestIdpPluginConfigMetaRetrieveApi:
+    def test_retrieve(self, api_client):
+        for plugin_cls in list_plugin_cls():
+            resp = api_client.get(reverse("idp_plugin_config_meta.retrieve", kwargs={"id": plugin_cls.id}))
+            assert resp.data["id"] == plugin_cls.id
+            assert resp.data["json_schema"] == plugin_cls.config_class.model_json_schema()
+
+
 class TestIdpCreateApi:
     def test_create_with_wecom_idp(self, api_client, wecom_plugin_cfg, data_source_match_rules):
         resp = api_client.post(