fix(api): protect workspace credential mutations (#40790)

api/controllers/console/datasets/data_source.py

@@ -36,6 +36,7 @@ from ..wraps import (
     RBACPermission,
     RBACResourceScope,
     account_initialization_required,
+    is_admin_or_owner_required,
     model_validate,
     rbac_permission_required,
     setup_required,
@@ -187,6 +188,8 @@ class DataSourceApi(Resource):
 
     @setup_required
     @login_required
+    @is_admin_or_owner_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.CREDENTIAL_MANAGE, resource_required=False)
     @account_initialization_required
     @console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
     @with_current_tenant_id

api/controllers/console/workspace/tool_providers.py

@@ -580,6 +580,8 @@ class ToolBuiltinProviderAddApi(Resource):
     )
     @setup_required
     @login_required
+    @is_admin_or_owner_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.CREDENTIAL_CREATE, resource_required=False)
     @account_initialization_required
     @with_current_user
     @with_current_tenant_id

api/tests/unit_tests/controllers/console/test_workspace_credential_mutation_permissions.py

@@ -0,0 +1,28 @@
+from inspect import getclosurevars, unwrap
+from types import FunctionType
+
+import pytest
+
+from controllers.common.wraps import RBACPermission, RBACResourceScope
+from controllers.console.datasets.data_source import DataSourceApi
+from controllers.console.workspace.tool_providers import ToolBuiltinProviderAddApi
+
+
+@pytest.mark.parametrize(
+    ("method", "permission"),
+    [
+        (ToolBuiltinProviderAddApi.post, RBACPermission.CREDENTIAL_CREATE),
+        (DataSourceApi.patch, RBACPermission.CREDENTIAL_MANAGE),
+    ],
+)
+def test_workspace_credential_mutations_require_management_permission(
+    method: FunctionType, permission: RBACPermission
+) -> None:
+    legacy_wrapper = unwrap(method, stop=lambda wrapper: "is_admin_or_owner_required" in wrapper.__code__.co_qualname)
+    assert "is_admin_or_owner_required" in legacy_wrapper.__code__.co_qualname
+
+    rbac_wrapper = unwrap(method, stop=lambda wrapper: "rbac_permission_required" in wrapper.__code__.co_qualname)
+    rbac_config = getclosurevars(rbac_wrapper).nonlocals
+    assert rbac_config["resource_type"] == RBACResourceScope.WORKSPACE
+    assert rbac_config["scene"] == permission
+    assert rbac_config["resource_required"] is False

api/tests/unit_tests/controllers/console/workspace/test_tool_providers.py

@@ -336,6 +336,7 @@ def test_builtin_provider_add_passes_payload(
     app: Flask, controller_module: ModuleType, monkeypatch: pytest.MonkeyPatch
 ):
     user = _mock_account()
+    user.role = TenantAccountRole.ADMIN
     _set_current_account(monkeypatch, controller_module, user, "tenant-456")
 
     service_mock = MagicMock(return_value={"result": "success"})