From 244336ec57cd5749f7366591b53008724de74124 Mon Sep 17 00:00:00 2001 From: RockChinQ Date: Sun, 1 Nov 2026 04:51:22 +0000 Subject: [PATCH] fix(api-tools): enforce strict workspace check on handler actions and remove mock fallbacks (LANG-227) Co-authored-by: multica-agent --- .../pkg/agent/runner/resource_builder.py | 51 +++---- src/langbot/pkg/api/application/__init__.py | 1 - src/langbot/pkg/plugin/handler.py | 113 +++++++------- .../api/test_runner_workspace_operations.py | 138 +++++++++++++++++- 4 files changed, 206 insertions(+), 97 deletions(-) diff --git a/src/langbot/pkg/agent/runner/resource_builder.py b/src/langbot/pkg/agent/runner/resource_builder.py index 30c59d031..c5bfec3ad 100644 --- a/src/langbot/pkg/agent/runner/resource_builder.py +++ b/src/langbot/pkg/agent/runner/resource_builder.py @@ -172,48 +172,31 @@ class AgentResourceBuilder: if not allowed_ops: return [] + # Processor identity is strictly required; do not fall back to event ID + if not binding.processor_id: + return [] + + workspace_service = getattr(self.ap, 'workspace_service', None) + if workspace_service is None: + return [] + from ...api.application.processor_context import build_processor_request_context from ...api.application.granted_operations import list_granted workspace_uuid = execution_context.workspace_uuid or event.workspace_id or '' processor_kind = binding.processor_type or 'agent' - processor_uuid = binding.processor_id or event.event_id trigger_actor = event.actor.actor_id if event.actor else None try: - if getattr(self.ap, 'workspace_service', None) is not None: - processor_context = await build_processor_request_context( - self.ap, - workspace_uuid=workspace_uuid, - processor_kind=processor_kind, - processor_uuid=processor_uuid, - run_id=event.event_id, - operation_ids=allowed_ops, - trigger_actor=trigger_actor, - ) - else: - from ...api.context import PrincipalContext, PrincipalType, RequestContext, WorkspaceContext - - principal = PrincipalContext( - principal_type=PrincipalType.PROCESSOR, - processor_kind=processor_kind, - processor_uuid=processor_uuid, - run_id=event.event_id, - trigger_actor=trigger_actor, - ) - workspace = WorkspaceContext( - workspace_uuid=workspace_uuid, - permissions=frozenset(), - ) - processor_context = RequestContext( - instance_uuid='test-instance', - placement_generation=1, - request_id=event.event_id, - auth_type='processor', - principal=principal, - workspace=workspace, - allowed_operation_ids=frozenset(allowed_ops), - ) + processor_context = await build_processor_request_context( + self.ap, + workspace_uuid=workspace_uuid, + processor_kind=processor_kind, + processor_uuid=binding.processor_id, + run_id=event.event_id, + operation_ids=allowed_ops, + trigger_actor=trigger_actor, + ) granted_ops = list_granted( self.ap, processor_context, diff --git a/src/langbot/pkg/api/application/__init__.py b/src/langbot/pkg/api/application/__init__.py index 33223e073..91532d38a 100644 --- a/src/langbot/pkg/api/application/__init__.py +++ b/src/langbot/pkg/api/application/__init__.py @@ -13,4 +13,3 @@ __all__ = [ 'get_workspace_operations_catalog', 'validate_grant_change', ] - diff --git a/src/langbot/pkg/plugin/handler.py b/src/langbot/pkg/plugin/handler.py index da8d5c63d..cf07fe9d0 100644 --- a/src/langbot/pkg/plugin/handler.py +++ b/src/langbot/pkg/plugin/handler.py @@ -2828,18 +2828,47 @@ class RuntimeConnectionHandler(handler.Handler): }, ) - @self.action(_plugin_runtime_action('LIST_APPLICATION_OPERATIONS', 'list_application_operations')) - async def list_application_operations(data: dict[str, Any]) -> handler.ActionResponse: + async def _validate_run_workspace_operations( + data: dict[str, Any], + ) -> tuple[dict[str, Any] | None, handler.ActionResponse | None]: run_id = data.get('run_id') if not run_id: - return handler.ActionResponse.error(message='run_id is required') + return None, handler.ActionResponse.error(message='run_id is required') + + caller_plugin_identity = data.get('caller_plugin_identity') + if not caller_plugin_identity: + return None, handler.ActionResponse.error( + message=f'caller_plugin_identity is required for run_id {run_id}' + ) + + action_context = self.current_action_context + if action_context is None or not getattr(action_context, 'workspace_uuid', None): + return None, handler.ActionResponse.error( + message='Runtime action is missing a trusted Workspace context' + ) + session_registry = get_session_registry() session = await session_registry.get(run_id) if not session: - return handler.ActionResponse.error(message=f'Run session {run_id} not found or expired') - caller_plugin_identity = data.get('caller_plugin_identity') - if caller_plugin_identity and caller_plugin_identity != session.get('plugin_identity'): - return handler.ActionResponse.error(message=f'Plugin identity mismatch for run_id {run_id}') + return None, handler.ActionResponse.error(message=f'Run session {run_id} not found or expired') + + session_plugin_identity = session.get('plugin_identity') + if caller_plugin_identity != session_plugin_identity: + return None, handler.ActionResponse.error( + message=f'Plugin identity mismatch: caller {caller_plugin_identity} is not authorized for run_id {run_id}' + ) + + session_workspace_id = session.get('authorization', {}).get('workspace_id') + if not session_workspace_id or session_workspace_id != action_context.workspace_uuid: + return None, handler.ActionResponse.error(message='Run session belongs to another Workspace') + + return session, None + + @self.action(_plugin_runtime_action('LIST_APPLICATION_OPERATIONS', 'list_application_operations')) + async def list_application_operations(data: dict[str, Any]) -> handler.ActionResponse: + session, error = await _validate_run_workspace_operations(data) + if error: + return error resources = session.get('authorization', {}).get('resources', {}) operations = resources.get('workspace_operations') or session.get('workspace_operations', []) @@ -2856,17 +2885,11 @@ class RuntimeConnectionHandler(handler.Handler): @self.action(_plugin_runtime_action('CALL_APPLICATION_OPERATION', 'call_application_operation')) async def call_application_operation(data: dict[str, Any]) -> handler.ActionResponse: - run_id = data.get('run_id') - if not run_id: - return handler.ActionResponse.error(message='run_id is required') - session_registry = get_session_registry() - session = await session_registry.get(run_id) - if not session: - return handler.ActionResponse.error(message=f'Run session {run_id} not found or expired') - caller_plugin_identity = data.get('caller_plugin_identity') - if caller_plugin_identity and caller_plugin_identity != session.get('plugin_identity'): - return handler.ActionResponse.error(message=f'Plugin identity mismatch for run_id {run_id}') + session, error = await _validate_run_workspace_operations(data) + if error: + return error + run_id = data.get('run_id') operation_id = data.get('operation_id') if not operation_id or not isinstance(operation_id, str): return handler.ActionResponse.error(message='operation_id is required') @@ -2877,12 +2900,15 @@ class RuntimeConnectionHandler(handler.Handler): return handler.ActionResponse.error(message='arguments must be an object') authorization = session.get('authorization', {}) - workspace_id = authorization.get('workspace_id') or session.get('workspace_id') + workspace_id = authorization.get('workspace_id') if not workspace_id: return handler.ActionResponse.error(message='workspace_id not found in run session') - processor_type = authorization.get('processor_type') or session.get('processor_type', 'agent') - processor_id = authorization.get('processor_id') or session.get('processor_id') or run_id + processor_type = authorization.get('processor_type') or session.get('processor_type') + processor_id = authorization.get('processor_id') or session.get('processor_id') + if not processor_type or not processor_id: + return handler.ActionResponse.error(message=f'processor_id is required for run_id {run_id}') + trigger_actor = authorization.get('trigger_actor') or session.get('trigger_actor') # Frozen granted operations from session @@ -2890,44 +2916,23 @@ class RuntimeConnectionHandler(handler.Handler): frozen_ops = resources.get('workspace_operations') or session.get('workspace_operations', []) granted_ids = [op.get('id') if isinstance(op, dict) else getattr(op, 'id', str(op)) for op in frozen_ops] + workspace_service = getattr(self.ap, 'workspace_service', None) + if workspace_service is None: + return handler.ActionResponse.error(message='Workspace execution service is unavailable') + from ..api.application.processor_context import build_processor_request_context import langbot.pkg.api.application.granted_operations as granted_operations try: - if getattr(self.ap, 'workspace_service', None) is not None: - processor_context = await build_processor_request_context( - self.ap, - workspace_uuid=workspace_id, - processor_kind=processor_type, - processor_uuid=processor_id, - run_id=run_id, - operation_ids=granted_ids, - trigger_actor=trigger_actor, - ) - else: - from ..api.context import PrincipalContext, PrincipalType, RequestContext, WorkspaceContext - - principal = PrincipalContext( - principal_type=PrincipalType.PROCESSOR, - processor_kind=processor_type, - processor_uuid=processor_id, - run_id=run_id, - trigger_actor=trigger_actor, - ) - workspace = WorkspaceContext( - workspace_uuid=workspace_id, - permissions=frozenset(), - ) - processor_context = RequestContext( - instance_uuid='test-instance', - placement_generation=1, - request_id=str(uuid.uuid4()), - auth_type='processor', - principal=principal, - workspace=workspace, - allowed_operation_ids=frozenset(granted_ids), - ) - + processor_context = await build_processor_request_context( + self.ap, + workspace_uuid=workspace_id, + processor_kind=processor_type, + processor_uuid=processor_id, + run_id=run_id, + operation_ids=granted_ids, + trigger_actor=trigger_actor, + ) result = await granted_operations.call( self.ap, processor_context, diff --git a/tests/unit_tests/api/test_runner_workspace_operations.py b/tests/unit_tests/api/test_runner_workspace_operations.py index 22aaeb6a7..425f62cd9 100644 --- a/tests/unit_tests/api/test_runner_workspace_operations.py +++ b/tests/unit_tests/api/test_runner_workspace_operations.py @@ -63,6 +63,7 @@ def test_env(monkeypatch): handler = RuntimeConnectionHandler(Mock(), AsyncMock(return_value=True), host) handler.register_installation_binding(binding, plugin_author='langbot-team', plugin_name='LocalAgent') + handler._current_action_context.set(binding) host.plugin_connector = SimpleNamespace(handler=handler) exec_api = handler.execution_api @@ -272,7 +273,8 @@ async def test_call_application_operation_success_and_audit(test_env): # 1. 资源下发 (Delivery) assert 'workspace_operations' in resources assert len(resources['workspace_operations']) == 1 - print(f"\n[CHAIN] 1. 资源下发 -> operations: {[op['id'] for op in resources['workspace_operations']]}") + ops_ids = [op['id'] for op in resources['workspace_operations']] + print(f'\n[CHAIN] 1. 资源下发 -> operations: {ops_ids}') # 2. 列出授权操作 (List) list_result = await test_env.exec_api.invoke( @@ -282,7 +284,10 @@ async def test_call_application_operation_success_and_audit(test_env): ) assert len(list_result['operations']) == 1 assert list_result['operations'][0]['id'] == 'bot.list' - print(f"[CHAIN] 2. list_application_operations -> operations: {[op['id'] for op in list_result['operations']]}, domains: {list_result['domains']}") + print( + f'[CHAIN] 2. list_application_operations -> ' + f'operations: {[op["id"] for op in list_result["operations"]]}, domains: {list_result["domains"]}' + ) # 3. 调用授权操作 (Call) call_result = await test_env.exec_api.invoke( @@ -292,7 +297,7 @@ async def test_call_application_operation_success_and_audit(test_env): ) assert call_result['ok'] is True assert isinstance(call_result['result'], list) - print(f"[CHAIN] 3. call_application_operation -> ok: {call_result['ok']}, result: {call_result['result']}") + print(f'[CHAIN] 3. call_application_operation -> ok: {call_result["ok"]}, result: {call_result["result"]}') # 4. 审计记录验证 (Audit Record) assert len(test_env.recorded_audits) > 0 @@ -302,7 +307,12 @@ async def test_call_application_operation_success_and_audit(test_env): assert audit['detail']['processor_kind'] == 'agent' assert audit['detail']['processor_uuid'] == 'agent-uuid-999' assert audit['detail']['run_id'] == run_id - print(f"[CHAIN] 4. 审计记录 (_record) -> route: {audit['route']}, principal_type: {audit['principal_type']}, processor_kind: {audit['detail']['processor_kind']}, processor_uuid: {audit['detail']['processor_uuid']}, run_id: {audit['detail']['run_id']}") + detail = audit['detail'] + print( + f'[CHAIN] 4. 审计记录 (_record) -> route: {audit["route"]}, ' + f'principal_type: {audit["principal_type"]}, processor_kind: {detail["processor_kind"]}, ' + f'processor_uuid: {detail["processor_uuid"]}, run_id: {detail["run_id"]}' + ) # 4. call 调用未授权的操作，返回 not_found @@ -375,7 +385,7 @@ async def test_call_outside_run_is_rejected(test_env): assert exc_info2.value.error.code == 'permission_denied' -# 6. 跨工作区的 run 被拒绝 +# 6. 跨工作区的 run 被拒绝（覆盖 ExecutionAPI 与 SDK handler actions 两种路径） @pytest.mark.asyncio async def test_cross_workspace_run_is_rejected(test_env): run_id = 'run-cross-ws' @@ -392,22 +402,134 @@ async def test_cross_workspace_run_is_rejected(test_env): 'skills': [], 'storage': {}, 'platform_capabilities': {}, + 'workspace_operations': [{'id': 'bot.list'}], }, - workspace_id='workspace-OTHER', # Different workspace + workspace_id='workspace-OTHER', # Different workspace from test_env.binding (workspace-a) + processor_type='agent', + processor_id='agent-cross-1', ) - # Inbound context belongs to workspace-a + # 1. 经 ExecutionAPI 调用（from_handler=False 路径） exec_context = ExecutionAPIContext( binding=test_env.binding, # workspace-a run_id=run_id, claimed_identity='langbot-team/LocalAgent', ) - with pytest.raises(ApplicationAPIException) as exc_info: await test_env.exec_api.invoke('execution.list_application_operations', exec_context, {}) assert exc_info.value.error.code == 'permission_denied' assert 'does not belong to this installation Workspace' in exc_info.value.error.message + with pytest.raises(ApplicationAPIException) as exc_info_call: + await test_env.exec_api.invoke( + 'execution.call_application_operation', + exec_context, + {'operation_id': 'bot.list', 'arguments': {}}, + ) + assert exc_info_call.value.error.code == 'permission_denied' + + # 2. 经 SDK handler actions 调用（from_handler=True 路径） + handler_payload = { + 'run_id': run_id, + 'caller_plugin_identity': 'langbot-team/LocalAgent', + } + resp_list = await test_env.handler.actions['list_application_operations'](handler_payload) + assert resp_list.code != 0 + assert 'Run session belongs to another Workspace' in resp_list.message + + call_payload = { + 'run_id': run_id, + 'caller_plugin_identity': 'langbot-team/LocalAgent', + 'operation_id': 'bot.list', + 'arguments': {}, + } + resp_call = await test_env.handler.actions['call_application_operation'](call_payload) + assert resp_call.code != 0 + assert 'Run session belongs to another Workspace' in resp_call.message + + # 3. session 缺少 workspace_id 时同样严格拒绝 + run_id_no_ws = 'run-no-ws' + await registry.register( + run_id=run_id_no_ws, + runner_id='plugin:langbot-team/LocalAgent/default', + query_id=None, + plugin_identity='langbot-team/LocalAgent', + resources={'workspace_operations': [{'id': 'bot.list'}]}, + workspace_id=None, + processor_type='agent', + processor_id='agent-no-ws', + ) + resp_no_ws = await test_env.handler.actions['list_application_operations']( + { + 'run_id': run_id_no_ws, + 'caller_plugin_identity': 'langbot-team/LocalAgent', + } + ) + assert resp_no_ws.code != 0 + assert 'Run session belongs to another Workspace' in resp_no_ws.message + + # 4. 缺少 caller_plugin_identity 时严格拒绝 + resp_no_caller = await test_env.handler._raw_actions['list_application_operations']( + { + 'run_id': run_id, + } + ) + assert resp_no_caller.code != 0 + assert 'caller_plugin_identity is required' in resp_no_caller.message + + # 5. caller_plugin_identity 伪造/不匹配时被拒绝 + resp_bad_caller = await test_env.handler.actions['list_application_operations']( + { + 'run_id': run_id, + 'caller_plugin_identity': 'evil-author/evil-plugin', + } + ) + assert resp_bad_caller.code != 0 + assert 'Caller plugin identity does not match the installation binding' in resp_bad_caller.message + + +# 处理器身份不能伪造：缺少 processor_id 时不下发、拒绝调用，不回退到 run/event ID +@pytest.mark.asyncio +async def test_processor_id_required_and_cannot_be_faked(test_env): + run_id = 'run-no-proc-id' + builder = AgentResourceBuilder(test_env.host) + event = make_event(event_id='event-fake-id') + descriptor = make_local_agent_descriptor() + ctx = ExecutionContext(instance_uuid='instance-a', workspace_uuid='workspace-a', placement_generation=1) + + # 缺少 processor_id 时不下发操作（返回空列表，不使用 event.event_id 回退） + binding_no_proc = AgentBinding( + binding_id='bind-no-proc', + runner_id=descriptor.id, + processor_type='agent', + processor_id=None, + resource_policy=ResourcePolicy(allowed_workspace_operations=['bot.list']), + ) + resources_no_proc = await builder.build_resources_from_binding(ctx, event, binding_no_proc, descriptor) + assert resources_no_proc.get('workspace_operations') == [] + + # session 缺少 processor_id 时拒绝调用，不使用 run_id 回退 + registry = get_session_registry() + await registry.register( + run_id=run_id, + runner_id=descriptor.id, + query_id=None, + plugin_identity=descriptor.get_plugin_id(), + resources={'workspace_operations': [{'id': 'bot.list'}]}, + workspace_id='workspace-a', + processor_type='agent', + processor_id=None, + ) + call_payload = { + 'run_id': run_id, + 'caller_plugin_identity': 'langbot-team/LocalAgent', + 'operation_id': 'bot.list', + 'arguments': {}, + } + resp_call = await test_env.handler.actions['call_application_operation'](call_payload) + assert resp_call.code != 0 + assert 'processor_id is required' in resp_call.message + # 7. 运行期间修改配置，不影响本次运行 @pytest.mark.asyncio