fix(plugin): protect endpoint listings (#40897)

This commit is contained in:
WH-2099
2026-08-21 02:17:58 +00:00
committed by GitHub
parent 3d08b2f89a
commit a5e284e6b1
2 changed files with 22 additions and 0 deletions
@@ -290,6 +290,8 @@ class EndpointListApi(Resource):
)
@setup_required
@login_required
@is_admin_or_owner_required
@rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MODEL_CONFIG, resource_required=False)
@account_initialization_required
@with_current_user_id
@with_current_tenant_id
@@ -318,6 +320,8 @@ class EndpointListForSinglePluginApi(Resource):
)
@setup_required
@login_required
@is_admin_or_owner_required
@rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_MODEL_CONFIG, resource_required=False)
@account_initialization_required
@with_current_user_id
@with_current_tenant_id
@@ -1,5 +1,6 @@
import inspect
from datetime import UTC, datetime
from types import FunctionType
from unittest.mock import patch
import pytest
@@ -23,6 +24,7 @@ from controllers.console.workspace.endpoint import (
EndpointUpdatePayload,
LegacyEndpointUpdatePayload,
)
from controllers.console.wraps import RBACPermission, RBACResourceScope
from core.entities.provider_entities import ProviderConfig, ProviderConfigType
from core.plugin.entities.endpoint import EndpointEntityWithInstance, EndpointProviderDeclaration
from core.plugin.impl.exc import PluginPermissionDeniedError
@@ -56,6 +58,22 @@ def _endpoint_entity() -> EndpointEntityWithInstance:
)
@pytest.mark.parametrize("method", [EndpointListApi.get, EndpointListForSinglePluginApi.get])
def test_endpoint_lists_require_management_permission(method: FunctionType) -> None:
legacy_wrapper = inspect.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 = inspect.unwrap(
method, stop=lambda wrapper: "rbac_permission_required" in wrapper.__code__.co_qualname
)
rbac_config = inspect.getclosurevars(rbac_wrapper).nonlocals
assert rbac_config["resource_type"] == RBACResourceScope.WORKSPACE
assert rbac_config["scene"] == RBACPermission.PLUGIN_MODEL_CONFIG
assert rbac_config["resource_required"] is False
class TestEndpointCollectionApi:
def test_create_success(self, app: Flask):
api = EndpointCollectionApi()