fix(pipeline): decouple Runner from extension permissions (LANG-54) (#2723)

* fix(pipeline): decouple runner from extension allowlist

Co-authored-by: multica-agent <[email protected]>

* fix(pipeline): complete runner diagnostics and real runtime coverage

Co-authored-by: multica-agent <[email protected]>

---------

Co-authored-by: multica-agent <[email protected]>
This commit is contained in:
RockChinQ
2026-10-19 23:07:11 +08:00
committed by GitHub
co-authored by multica-agent
parent 5a6ddbff6f
commit d9e4e7d9df
24 changed files with 1012 additions and 35 deletions
+268 -10
View File
@@ -7,6 +7,9 @@ loads a deterministic Runner plugin through the real SDK Plugin Runtime.
from __future__ import annotations
import shutil
import json
import os
import signal
import socket
import sqlite3
import subprocess
@@ -31,15 +34,18 @@ QA_RUNNER_ID = 'plugin:e2e/agent-runner-qa/default'
@pytest.fixture(scope='session')
def runner_e2e_port():
"""Port for the Runner plugin-runtime E2E process."""
return 15310
return _free_port()
@pytest.fixture(scope='session')
def runner_e2e_tmpdir():
"""Create temporary directory for Runner E2E testing."""
tmpdir = Path(tempfile.mkdtemp(prefix='langbot_runner_e2e_'))
root = find_project_root() / '.qa' / 'lang54'
root.mkdir(parents=True, exist_ok=True)
tmpdir = Path(tempfile.mkdtemp(prefix='runner-runtime-', dir=root))
yield tmpdir
shutil.rmtree(tmpdir, ignore_errors=True)
if not os.environ.get('LANGBOT_KEEP_RUNNER_E2E'):
shutil.rmtree(tmpdir, ignore_errors=True)
def _write_qa_runner_plugin(plugin_root: Path) -> None:
@@ -74,6 +80,15 @@ def _write_qa_runner_plugin(plugin_root: Path) -> None:
Runner:
fromDirs:
- path: components/runner/
Tool:
fromDirs:
- path: components/tools/
Command:
fromDirs:
- path: components/commands/
EventListener:
fromDirs:
- path: components/event_listener/
pages: []
execution:
python:
@@ -115,10 +130,19 @@ def _write_qa_runner_plugin(plugin_root: Path) -> None:
zh_Hans: 回显输入并验证运行级状态 API。
spec:
usages: [agent]
config: []
config:
- name: prefix
type: string
label:
en_US: Echo prefix
zh_Hans: 回显前缀
default: 'e2e echo'
required: false
capabilities:
streaming: false
permissions: {}
tool_calling: true
permissions:
tools: [detail, call]
execution:
python:
path: default.py
@@ -134,6 +158,8 @@ def _write_qa_runner_plugin(plugin_root: Path) -> None:
from __future__ import annotations
from typing import AsyncGenerator
import json
from pathlib import Path
from langbot_plugin.api.definition.components.runner.runner import Runner
from langbot_plugin.api.entities.builtin.runner.context import RunnerContext
@@ -144,9 +170,26 @@ def _write_qa_runner_plugin(plugin_root: Path) -> None:
class DefaultRunner(Runner):
async def run(self, ctx: RunnerContext) -> AsyncGenerator[RunnerResult, None]:
text = ctx.input.to_text()
content = f'{ctx.config.get("prefix", "e2e echo")}: {text}'
if text.startswith('boundary probe'):
tools = await ctx.get_available_tools()
names = [item['name'] for item in tools]
tool_name = next((name for name in names if 'qa_boundary_tool' in name), 'qa_boundary_tool')
try:
result = await ctx.call_tool(tool_name, {})
denied = 'QA_TOOL_EXECUTED' not in json.dumps(result)
except Exception as exc:
result = str(exc)
denied = True
content = json.dumps({
'tool_result': str(result),
'tool_names': names,
'tool_denied': denied,
'listener_hits': getattr(self.plugin, 'boundary_listener_hits', 0),
})
yield RunnerResult.message_completed(
ctx.run_id,
Message(role='assistant', content=f'e2e echo: {text}'),
Message(role='assistant', content=content),
)
yield RunnerResult.state_updated(
ctx.run_id,
@@ -161,6 +204,75 @@ def _write_qa_runner_plugin(plugin_root: Path) -> None:
encoding='utf-8',
)
import yaml
components = {
'tools': (
'Tool',
'qa_boundary_tool',
'BoundaryTool',
{'parameters': {}, 'llm_prompt': 'QA boundary tool'},
"""
from langbot_plugin.api.definition.components.tool.tool import Tool
class BoundaryTool(Tool):
async def call(self, params, session, query_id):
return {'result': 'QA_TOOL_EXECUTED'}
""",
),
'commands': (
'Command',
'qa_boundary_command',
'BoundaryCommand',
{},
"""
from langbot_plugin.api.definition.components.command.command import Command
from langbot_plugin.api.entities.builtin.command.context import CommandReturn
class BoundaryCommand(Command):
async def initialize(self):
await super().initialize()
@self.subcommand('')
async def handle(self, ctx):
yield CommandReturn(text='QA_COMMAND_EXECUTED')
""",
),
'event_listener': (
'EventListener',
'boundary',
'BoundaryListener',
{},
"""
from pathlib import Path
from langbot_plugin.api.definition.components.common.event_listener import EventListener
from langbot_plugin.api.entities import events
class BoundaryListener(EventListener):
async def initialize(self):
await super().initialize()
@self.handler(events.PersonNormalMessageReceived)
async def handle(ctx):
self.plugin.boundary_listener_hits = getattr(self.plugin, 'boundary_listener_hits', 0) + 1
""",
),
}
for directory, (kind, name, attr, spec, source) in components.items():
target = plugin_root / 'components' / directory
target.mkdir(parents=True, exist_ok=True)
(target / f'{name}.yaml').write_text(
yaml.safe_dump(
{
'apiVersion': 'v1',
'kind': kind,
'metadata': {'name': name, 'label': {'en_US': name}, 'description': {'en_US': name}},
'spec': spec,
'execution': {'python': {'path': f'{name}.py', 'attr': attr}},
}
),
encoding='utf-8',
)
(target / f'{name}.py').write_text(textwrap.dedent(source), encoding='utf-8')
processor_dir = runner_dir
(processor_dir / 'welcome.yaml').write_text(
textwrap.dedent("""
@@ -237,6 +349,12 @@ def runner_e2e_config_path(runner_e2e_tmpdir, runner_e2e_port, runner_runtime_po
config['plugin']['enable_marketplace'] = False
config['box']['enabled'] = False
config['system']['jwt']['secret'] = 'e2e-agent-runner-secret-key'
adapters = find_project_root() / 'src' / 'langbot' / 'pkg' / 'platform' / 'sources'
config['system']['disabled_adapters'] = [
yaml.safe_load(path.read_text(encoding='utf-8'))['metadata']['name']
for path in sorted(adapters.glob('*.yaml'))
if path.stem != 'websocket'
]
with open(config_path, 'w', encoding='utf-8') as f:
yaml.safe_dump(config, f, default_flow_style=False)
@@ -275,12 +393,14 @@ def runner_runtime_process(runner_e2e_tmpdir, runner_runtime_ports):
start_new_session=True,
)
yield proc
proc.terminate()
try:
os.killpg(proc.pid, signal.SIGTERM)
proc.wait(timeout=5)
except subprocess.TimeoutExpired:
proc.kill()
os.killpg(proc.pid, signal.SIGKILL)
proc.wait()
except ProcessLookupError:
pass
stdout_file.close()
stderr_file.close()
@@ -298,8 +418,9 @@ def runner_langbot_process(
project_root=project_root,
work_dir=runner_e2e_tmpdir,
port=runner_e2e_port,
timeout=180,
debug=True,
timeout=360,
debug=False,
collect_coverage=False,
cli_args=['--standalone-runtime'],
)
@@ -565,3 +686,140 @@ def test_event_processor_real_runtime_logs_actions_and_instance_isolation(
foreign = client.get(f'/api/v1/agents/{created[1]}/runs/{run["run_id"]}/events', headers=headers)
assert foreign.status_code == 400
assert client.get(f'/api/v1/agents/{created[1]}/runs', headers=headers).json()['data']['items'] == []
def _create_boundary_pipeline(client, token):
headers = {'Authorization': f'Bearer {token}'}
created = client.post(
'/api/v1/pipelines', headers=headers, json={'name': 'Runner boundary QA', 'description': ''}
).json()
assert created['code'] == 0, created
pipeline_id = created['data']['uuid']
config = client.get(f'/api/v1/pipelines/{pipeline_id}', headers=headers).json()['data']['pipeline']['config']
config['ai']['runner'] = {'id': QA_RUNNER_ID, 'expire-time': 0}
config['ai']['runner_config'] = {QA_RUNNER_ID: {'prefix': 'e2e echo', 'hidden': {'keep': [None, False, ' raw ']}}}
saved = client.put(f'/api/v1/pipelines/{pipeline_id}', headers=headers, json={'config': config}).json()
assert saved['code'] == 0, saved
return pipeline_id
def _pipeline_failure_diagnostics(directory, pipeline_id):
with sqlite3.connect(directory / 'data' / 'langbot.db') as connection:
messages = connection.execute(
'SELECT status, level, runner_name FROM monitoring_messages WHERE pipeline_id = ?',
(pipeline_id,),
).fetchall()
calls = connection.execute(
'SELECT status, error_message FROM monitoring_llm_calls WHERE pipeline_id = ?',
(pipeline_id,),
).fetchall()
runs = connection.execute(
'SELECT status, status_reason FROM agent_run WHERE binding_id = ?',
(pipeline_id,),
).fetchall()
errors = connection.execute(
'SELECT error_type, error_message, stack_trace FROM monitoring_errors WHERE pipeline_id = ?',
(pipeline_id,),
).fetchall()
diagnostic = {'messages': messages, 'calls': calls, 'runs': runs, 'errors': errors}
target = find_project_root() / '.qa' / 'lang54' / 'pipeline-failure.json'
target.write_text(json.dumps(diagnostic, indent=2), encoding='utf-8')
return diagnostic
def test_mixed_plugin_runner_does_not_authorize_extensions(runner_client, runner_e2e_tmpdir):
token = _init_and_auth(runner_client)
_ensure_qa_plugin(runner_client, token, runner_e2e_tmpdir / 'agent-runner-qa.zip')
pipeline_id = _create_boundary_pipeline(runner_client, token)
headers = {'Authorization': f'Bearer {token}'}
def run(text):
response = runner_client.post(
f'/api/v1/pipelines/{pipeline_id}/run', headers=headers, json={'message': text}
).json()
assert response['code'] == 0, response
if response['data']['status'] != 'completed':
pytest.fail(
f'{response}\nPersisted diagnostics: {_pipeline_failure_diagnostics(runner_e2e_tmpdir, pipeline_id)}'
)
return response['data']['reply']
enabled = json.loads(run('boundary probe enabled'))
assert enabled['listener_hits'] > 0, enabled
assert any('qa_boundary_tool' in name for name in enabled['tool_names']), enabled
assert enabled['tool_denied'] is False, enabled
assert 'QA_COMMAND_EXECUTED' in run('!qa_boundary_command')
last_enabled = json.loads(run('boundary probe before disabling'))
disabled = runner_client.put(
f'/api/v1/pipelines/{pipeline_id}/extensions',
headers=headers,
json={
'enable_all_plugins': False,
'bound_plugins': [],
'enable_all_mcp_servers': False,
'bound_mcp_servers': [],
'enable_all_skills': False,
'bound_skills': [],
},
).json()
assert disabled['code'] == 0, disabled
restricted = json.loads(run('boundary probe disabled'))
assert restricted['listener_hits'] == last_enabled['listener_hits'], restricted
assert not any('qa_boundary_tool' in name for name in restricted['tool_names']), restricted
assert restricted['tool_denied'] is True, restricted
command_reply = run('!qa_boundary_command')
assert 'QA_COMMAND_EXECUTED' not in command_reply
assert 'qa_boundary_command' in command_reply
def test_chromium_pipeline_save_copy_execute_real_runtime(runner_client, runner_e2e_tmpdir):
token = _init_and_auth(runner_client)
_ensure_qa_plugin(runner_client, token, runner_e2e_tmpdir / 'agent-runner-qa.zip')
pipeline_id = _create_boundary_pipeline(runner_client, token)
headers = {'Authorization': f'Bearer {token}'}
response = runner_client.put(
f'/api/v1/pipelines/{pipeline_id}/extensions',
headers=headers,
json={
'enable_all_plugins': False,
'bound_plugins': [],
'enable_all_mcp_servers': False,
'bound_mcp_servers': [],
'enable_all_skills': False,
'bound_skills': [],
},
).json()
assert response['code'] == 0, response
env = os.environ.copy()
env.update(
{
'LANGBOT_RUNNER_API_URL': str(runner_client.base_url).rstrip('/'),
'LANGBOT_RUNNER_WEB_PORT': str(_free_port()),
'LANGBOT_RUNNER_TOKEN': token,
'LANGBOT_RUNNER_PIPELINE_ID': pipeline_id,
'LANGBOT_RUNNER_BROWSER_OUTPUT': str(runner_e2e_tmpdir / 'browser-results'),
}
)
browser = subprocess.Popen(
['node', 'node_modules/@playwright/test/cli.js', 'test', '--config=playwright.runner-runtime.config.ts'],
cwd=find_project_root() / 'web',
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
start_new_session=True,
)
try:
stdout, stderr = browser.communicate(timeout=300)
assert browser.returncode == 0, f'Chromium real-stack E2E:\n{stdout}\n{stderr}'
print(stdout)
finally:
try:
os.killpg(browser.pid, signal.SIGTERM)
except ProcessLookupError:
pass
try:
browser.wait(timeout=5)
except subprocess.TimeoutExpired:
os.killpg(browser.pid, signal.SIGKILL)
browser.wait(timeout=5)
+11 -10
View File
@@ -8,7 +8,7 @@ import pytest
from langbot.pkg.agent.runner.registry import RunnerRegistry
from langbot.pkg.agent.runner.descriptor import RunnerDescriptor
from langbot.pkg.agent.runner.errors import RunnerNotFoundError, RunnerNotAuthorizedError
from langbot.pkg.agent.runner.errors import RunnerNotFoundError
from langbot.pkg.api.http.context import ExecutionContext
@@ -242,8 +242,9 @@ class TestRegistryGet:
assert ap.plugin_connector.list_runners.await_count == 2
@pytest.mark.asyncio
async def test_get_runner_with_bound_plugins_filter(self):
"""Get runner with bound plugins authorization."""
@pytest.mark.parametrize('plugins', [None, [], ['langbot-team/LocalAgent']])
async def test_get_runner_independent_of_extension_filter(self, plugins):
"""Selecting a Runner must not grant or require extension authorization."""
ap = FakeApplication()
registry = RunnerRegistry(ap)
@@ -255,13 +256,13 @@ class TestRegistryGet:
)
assert descriptor is not None
# Not authorized - plugin not in bound list
with pytest.raises(RunnerNotAuthorizedError):
await registry.get(
TEST_CONTEXT,
'plugin:alice/my-agent/custom',
bound_plugins=['langbot-team/LocalAgent'],
)
descriptor = await registry.get(
TEST_CONTEXT,
'plugin:alice/my-agent/custom',
bound_plugins=plugins,
)
assert descriptor.id == 'plugin:alice/my-agent/custom'
assert plugins in (None, [], ['langbot-team/LocalAgent'])
class TestRegistryMetadataForPipeline:
@@ -0,0 +1,39 @@
import ast
import yaml
from langbot_plugin.api.entities.builtin.runner.manifest import RunnerManifest
from langbot_plugin.api.definition.components.manifest import ComponentManifest
from tests.e2e.test_runner_plugin_runtime import _write_qa_runner_plugin
def test_real_runtime_fixture_contains_all_four_component_kinds(tmp_path):
_write_qa_runner_plugin(tmp_path)
manifest = yaml.safe_load((tmp_path / 'manifest.yaml').read_text())
assert set(manifest['spec']['components']) == {'Runner', 'Tool', 'Command', 'EventListener'}
kinds = set()
for path in (tmp_path / 'components').rglob('*.yaml'):
raw = yaml.safe_load(path.read_text())
component = ComponentManifest(
owner='e2e/agent-runner-qa', manifest=raw, rel_path=str(path.relative_to(tmp_path))
)
kinds.add(component.kind)
if component.kind == 'Runner':
spec = raw['spec']
runner = RunnerManifest(
id=f'plugin:e2e/agent-runner-qa/{raw["metadata"]["name"]}',
name=raw['metadata']['name'],
label=raw['metadata']['label'],
usages=spec['usages'],
supported_event_patterns=spec.get('events') or ['*'],
capabilities=spec.get('capabilities') or {},
permissions=spec.get('permissions') or {},
config_schema=spec.get('config') or [],
)
if path.stem == 'default':
assert runner.capabilities.tool_calling is True
assert set(runner.permissions.tools) == {'detail', 'call'}
assert kinds == {'Runner', 'Tool', 'Command', 'EventListener'}
for path in tmp_path.rglob('*.py'):
ast.parse(path.read_text(), filename=str(path))
@@ -0,0 +1,124 @@
"""Runner selection must survive restricted extension preferences on disk."""
from types import SimpleNamespace as NS
import json
from unittest.mock import AsyncMock, Mock
import pytest
import sqlalchemy as sa
from langbot.pkg.api.http.service.pipeline import PipelineService
from langbot.pkg.entity.persistence.pipeline import LegacyPipeline
from langbot.pkg.workspace.errors import WorkspaceNotFoundError
from tests.unit_tests.api.service.test_pipeline_migration import env, context, RID # noqa: F401
@pytest.mark.asyncio
async def test_http_and_mcp_share_real_persistence_without_expanding_extensions(env, monkeypatch): # noqa: F811
from quart import Quart
from langbot.pkg.api.http.controller.groups.pipelines.pipelines import PipelinesRouterGroup
from langbot.pkg.api.mcp.server import LangBotMCPServer
monkeypatch.setattr('langbot.pkg.api.http.service.workspace_limits.workspace_limits', AsyncMock(return_value={}))
monkeypatch.setattr('langbot.pkg.api.mcp.server.get_request_context', context)
monkeypatch.setattr('langbot.pkg.api.mcp.server.get_workspace_language', AsyncMock(return_value='en_US'))
monkeypatch.setattr('langbot.pkg.api.http.controller.group.get_workspace_language', AsyncMock(return_value='en_US'))
env.ap.ver_mgr = NS(get_current_version=Mock(return_value='4.10'))
env.ap.pipeline_mgr.load_pipeline = AsyncMock()
env.ap.pipeline_mgr.remove_pipeline = AsyncMock()
env.ap.pipeline_service = PipelineService(env.ap)
app = Quart(__name__)
routes = PipelinesRouterGroup(env.ap, app)
monkeypatch.setattr(routes, '_authenticate_api_key', AsyncMock(return_value=context()))
monkeypatch.setattr(routes, '_record_operation', AsyncMock())
await routes.initialize()
http = app.test_client()
mcp = LangBotMCPServer(env.ap)
headers = {'X-API-Key': 'isolated-test-key'}
config = {
'ai': {
'runner': {'id': RID, 'expire-time': 0},
'runner_config': {RID: {'unknown': {'keep': [None, False, ' raw ']}}},
}
}
prefs = {'enable_all_plugins': False, 'plugins': [], 'enable_all_mcp_servers': False, 'mcp_servers': []}
async with env.engine.begin() as conn:
await conn.execute(
sa.update(LegacyPipeline)
.where(LegacyPipeline.uuid == 'one')
.values(config=config, extensions_preferences=prefs)
)
response = await http.put('/api/v1/pipelines/one', headers=headers, json={'config': config})
assert response.status_code == 200, await response.get_data(as_text=True)
assert (await response.get_json())['code'] == 0
result = await mcp.mcp.call_tool('get_pipeline', {'pipeline_uuid': 'one'})
saved = json.loads(result[0][0].text)
assert saved['config'] == config
assert saved['extensions_preferences'] == prefs
config['ai']['runner_config'][RID]['edited'] = 'via MCP'
result = await mcp.mcp.call_tool('update_pipeline', {'pipeline_uuid': 'one', 'pipeline_data': {'config': config}})
assert json.loads(result[0][0].text) == {'ok': True}
copied_response = await http.post('/api/v1/pipelines/one/copy', headers=headers)
assert copied_response.status_code == 200, await copied_response.get_data(as_text=True)
copied = (await copied_response.get_json())['data']['uuid']
for pipeline_id in ('one', copied):
response = await http.get(f'/api/v1/pipelines/{pipeline_id}', headers=headers)
saved = (await response.get_json())['data']['pipeline']
assert saved['config'] == config
assert saved['extensions_preferences']['plugins'] == []
assert saved['extensions_preferences']['enable_all_plugins'] is False
async with env.engine.connect() as conn:
row = (
await conn.execute(
sa.select(LegacyPipeline.config, LegacyPipeline.extensions_preferences).where(
LegacyPipeline.uuid == pipeline_id
)
)
).one()
assert row.config == config
assert row.extensions_preferences['plugins'] == []
assert row.extensions_preferences['enable_all_plugins'] is False
body = {'name': 'HTTP creation', 'description': '', 'config': config, 'extensions_preferences': prefs}
response = await http.post('/api/v1/pipelines', headers=headers, json=body)
assert response.status_code == 200, await response.get_data(as_text=True)
http_id = (await response.get_json())['data']['uuid']
body['name'] = 'MCP creation'
result = await mcp.mcp.call_tool('create_pipeline', {'pipeline_data': body})
mcp_id = json.loads(result[0][0].text)['uuid']
for pipeline_id in (http_id, mcp_id):
saved = await env.ap.pipeline_service.get_pipeline(context(), pipeline_id, include_secret=True)
assert saved['config'] == config
assert saved['extensions_preferences'] == prefs
@pytest.mark.asyncio
async def test_edit_and_copy_do_not_expand_extension_permissions(env, monkeypatch): # noqa: F811
monkeypatch.setattr(
'langbot.pkg.api.http.service.workspace_limits.workspace_limits',
AsyncMock(return_value={}),
)
env.ap.ver_mgr = NS(get_current_version=Mock(return_value='4.10'))
env.ap.pipeline_mgr.load_pipeline = AsyncMock()
env.ap.pipeline_mgr.remove_pipeline = AsyncMock()
service = PipelineService(env.ap)
config = {'ai': {'runner': {'id': RID}, 'runner_config': {RID: {}}}}
prefs = {'enable_all_plugins': False, 'plugins': [], 'enable_all_mcp_servers': False, 'mcp_servers': []}
# Existing, pre-fix configuration: selected Runner with no extensions.
async with env.engine.begin() as conn:
await conn.execute(
sa.update(LegacyPipeline)
.where(LegacyPipeline.uuid == 'one')
.values(
config=config,
extensions_preferences=prefs,
)
)
await service.update_pipeline(context(), 'one', {'config': config, 'name': 'Edited'})
copied = await service.copy_pipeline(context(), 'one')
for pid in ('one', copied):
saved = await service.get_pipeline(context(), pid, include_secret=True)
assert saved['config'] == config
assert saved['extensions_preferences']['plugins'] == []
assert saved['extensions_preferences']['enable_all_plugins'] is False
with pytest.raises(WorkspaceNotFoundError):
await service.copy_pipeline(context(), 'foreign')
@@ -32,7 +32,8 @@ def configure_connector(connector) -> None:
@pytest.mark.asyncio
async def test_plugin_list_filter_by_component_kinds():
@pytest.mark.parametrize('independent_kind', ['KnowledgeEngine', 'Runner'])
async def test_plugin_list_filter_by_component_kinds(independent_kind):
"""Test that plugins can be filtered by component kinds."""
from langbot.pkg.plugin.connector import PluginRuntimeConnector
@@ -70,7 +71,7 @@ async def test_plugin_list_filter_by_component_kinds():
}
}
},
'components': [{'manifest': {'manifest': {'kind': 'KnowledgeEngine', 'metadata': {'name': 'retriever1'}}}}],
'components': [{'manifest': {'manifest': {'kind': independent_kind, 'metadata': {'name': 'retriever1'}}}}],
},
{
'debug': False,
@@ -107,7 +108,7 @@ async def test_plugin_list_filter_by_component_kinds():
}
},
'components': [
{'manifest': {'manifest': {'kind': 'KnowledgeEngine', 'metadata': {'name': 'retriever2'}}}},
{'manifest': {'manifest': {'kind': independent_kind, 'metadata': {'name': 'retriever2'}}}},
{'manifest': {'manifest': {'kind': 'Tool', 'metadata': {'name': 'tool2'}}}},
],
},