From 11ed8a23067b8767b85698ee6e8c60644684eb1d Mon Sep 17 00:00:00 2001 From: Hyu Date: Mon, 12 Oct 2026 13:36:26 +0800 Subject: [PATCH] feat(onebot): support main-port WebSocket transports and filter protocol events --- skills/scripts/e2e/onebot-runtime-probe.py | 11 +- .../e2e/wizard-onebot-agent-runtime.mjs | 28 +- .../cases/wizard-onebot-agent-runtime.yaml | 2 +- .../api/http/controller/groups/webhooks.py | 30 ++ src/langbot/pkg/api/http/service/bot.py | 4 +- src/langbot/pkg/platform/adapter_names.py | 3 +- .../platform/adapters/aiocqhttp/adapter.py | 139 ++++++-- .../adapters/aiocqhttp/event_converter.py | 26 +- .../platform/adapters/aiocqhttp/manifest.yaml | 81 +++-- .../platform/adapters/aiocqhttp/transport.py | 125 +++++++ src/langbot/pkg/platform/botmgr.py | 3 + tests/e2e/live_aiocqhttp_eba_probe.py | 18 +- .../platform/test_aiocqhttp_eba_adapter.py | 33 +- .../platform/test_botmgr_tenancy.py | 3 +- .../platform/test_omni_adapter_names.py | 45 ++- .../platform/test_onebot_transport.py | 329 ++++++++++++++++++ .../home/bots/components/bot-form/BotForm.tsx | 4 + web/src/app/wizard/page.tsx | 4 + 18 files changed, 752 insertions(+), 136 deletions(-) create mode 100644 src/langbot/pkg/platform/adapters/aiocqhttp/transport.py create mode 100644 tests/unit_tests/platform/test_onebot_transport.py diff --git a/skills/scripts/e2e/onebot-runtime-probe.py b/skills/scripts/e2e/onebot-runtime-probe.py index a080571fa..c919803bf 100644 --- a/skills/scripts/e2e/onebot-runtime-probe.py +++ b/skills/scripts/e2e/onebot-runtime-probe.py @@ -6,14 +6,15 @@ from __future__ import annotations import argparse import asyncio import json +import os import time import websockets -async def run(port: int) -> None: - uri = f'ws://127.0.0.1:{port}/ws' +async def run(uri: str) -> None: headers = { + 'Authorization': 'Bearer ' + os.environ['ONEBOT_PROBE_TOKEN'], 'X-Self-ID': '900001', 'X-Client-Role': 'Universal', 'User-Agent': 'LangBot-E2E-OneBot/1.0', @@ -27,7 +28,7 @@ async def run(port: int) -> None: except OSError: await asyncio.sleep(0.25) if connection is None: - raise RuntimeError(f'OneBot reverse WebSocket did not open on port {port}') + raise RuntimeError('OneBot reverse WebSocket did not open') actions: list[str] = [] event_deadline = time.monotonic() + 20 @@ -88,6 +89,6 @@ async def run(port: int) -> None: if __name__ == '__main__': parser = argparse.ArgumentParser() - parser.add_argument('--port', type=int, required=True) + parser.add_argument('--url', required=True) args = parser.parse_args() - asyncio.run(run(args.port)) + asyncio.run(run(args.url)) diff --git a/skills/scripts/e2e/wizard-onebot-agent-runtime.mjs b/skills/scripts/e2e/wizard-onebot-agent-runtime.mjs index dc94efe31..29f5b5506 100644 --- a/skills/scripts/e2e/wizard-onebot-agent-runtime.mjs +++ b/skills/scripts/e2e/wizard-onebot-agent-runtime.mjs @@ -1,7 +1,7 @@ #!/usr/bin/env node import { spawn } from "node:child_process"; -import { createServer } from "node:net"; +import { randomBytes } from "node:crypto"; import { dirname, join } from "node:path"; import { fileURLToPath } from "node:url"; import { @@ -60,19 +60,6 @@ const result = { evidence_collected: ["ui", "screenshot", "console", "api_diagnostic"], }; -async function getFreePort() { - const server = createServer(); - await new Promise((resolve, reject) => { - server.once("error", reject); - server.listen(0, "127.0.0.1", resolve); - }); - const address = server.address(); - const port = typeof address === "object" && address ? address.port : 0; - await new Promise((resolve) => server.close(resolve)); - if (!port) throw new Error("Could not allocate a temporary OneBot port."); - return port; -} - async function api(page, path, options = {}) { return await page.evaluate( async ({ baseUrl, path, options, token }) => { @@ -96,7 +83,7 @@ async function api(page, path, options = {}) { ); } -async function runProbe(port) { +async function runProbe(url, accessToken) { const repo = await resolveLangBotRepo(); const python = join(repo, ".venv", "bin", "python"); const script = join( @@ -107,8 +94,9 @@ async function runProbe(port) { throw new Error(`LangBot virtualenv Python is missing: ${python}`); } return await new Promise((resolve, reject) => { - const child = spawn(python, [script, "--port", String(port)], { + const child = spawn(python, [script, "--url", url], { cwd: repo, + env: { ...process.env, ONEBOT_PROBE_TOKEN: accessToken }, }); let stdout = ""; let stderr = ""; @@ -132,7 +120,7 @@ async function runProbe(port) { try { if (!frontendUrl) throw new Error("LANGBOT_FRONTEND_URL is not configured."); if (!backendUrl) throw new Error("LANGBOT_BACKEND_URL is not configured."); - const port = await getFreePort(); + const accessToken = randomBytes(32).toString("hex"); browser = await createBrowser(paths); const { page } = browser; @@ -190,9 +178,7 @@ try { .getByText(/Configure Your Bot|配置机器人|ボットを設定/) .first() .waitFor({ timeout: 15_000 }); - const portInput = page.getByRole("spinbutton").first(); - await portInput.fill(String(port)); - await portInput.press("Tab"); + await page.getByLabel(/Access token|访问令牌|アクセストークン/i).fill(accessToken); await page .getByRole("button", { name: /Save & Enable Bot|保存并启用|保存して有効化/, @@ -270,7 +256,7 @@ try { } result.visible_signals.push("route-persisted"); - result.runtime.onebot = await runProbe(port); + result.runtime.onebot = await runProbe(`${backendUrl.replace(/^http/, "ws").replace(/\/$/, "")}/bots/${botId}/onebot/v11/ws`, accessToken); let routeStatus; for (let attempt = 0; attempt < 30; attempt += 1) { routeStatus = await api( diff --git a/skills/skills/langbot-testing/cases/wizard-onebot-agent-runtime.yaml b/skills/skills/langbot-testing/cases/wizard-onebot-agent-runtime.yaml index 220c5b3f5..de4d56cb0 100644 --- a/skills/skills/langbot-testing/cases/wizard-onebot-agent-runtime.yaml +++ b/skills/skills/langbot-testing/cases/wizard-onebot-agent-runtime.yaml @@ -32,7 +32,7 @@ preconditions: - "The LangBot virtualenv includes the websockets package used by the local OneBot probe." steps: - "Open Quick Start and choose Welcome new members with the OneBot v11 channel." - - "Create and enable a temporary OneBot Bot on an available local port." + - "Create and enable a temporary OneBot Bot using the main-port reverse WS URL and a generated access token." - "Select Local Agent and confirm the editable welcome prompt is prefilled." - "Choose QA Deterministic Runner and finish the wizard." - "Connect a local OneBot reverse WebSocket and send a group_increase notice." diff --git a/src/langbot/pkg/api/http/controller/groups/webhooks.py b/src/langbot/pkg/api/http/controller/groups/webhooks.py index 8499772cd..cbc886b15 100644 --- a/src/langbot/pkg/api/http/controller/groups/webhooks.py +++ b/src/langbot/pkg/api/http/controller/groups/webhooks.py @@ -1,6 +1,7 @@ from __future__ import annotations import quart +import quart_cors import traceback from .. import group @@ -10,6 +11,35 @@ from .....utils import bounded_executor @group.group_class('webhooks', '/bots') class WebhookRouterGroup(group.RouterGroup): async def initialize(self) -> None: + @self.quart_app.websocket(self.path + '//onebot/v11/ws') + @quart_cors.cors_exempt + async def onebot_websocket(bot_uuid: str): + # Protocol clients omit Origin; bearer authentication protects this endpoint. + from .....platform.adapters.aiocqhttp.adapter import AiocqhttpAdapter + + runtime = await self.ap.platform_mgr.resolve_public_bot(bot_uuid) + if runtime is None or not runtime.enable or not isinstance(runtime.adapter, AiocqhttpAdapter): + quart.abort(404) + adapter = runtime.adapter + socket = quart.websocket._get_current_object() + if adapter.config['connection_mode'] != 'reverse_ws': + quart.abort(409) + if not adapter.authenticate(socket.headers): + quart.abort(401) + if socket.headers.get('X-Client-Role', '').lower() != 'universal': + quart.abort(400) + self_id = socket.headers.get('X-Self-ID', '') + expected = str(adapter.config.get('self_id', '') or '') + if not self_id or (expected and self_id != expected): + quart.abort(400) + if adapter.bot.connection is not None: + quart.abort(409) + await socket.accept() + try: + await adapter.serve_connection(socket.send, socket.receive, socket.close, self_id) + except Exception: + await socket.close(1011) + @self.route('/', methods=['GET', 'POST'], auth_type=group.AuthType.NONE) async def handle_webhook(bot_uuid: str): """处理 bot webhook 回调(无子路径)""" diff --git a/src/langbot/pkg/api/http/service/bot.py b/src/langbot/pkg/api/http/service/bot.py index 3e34c1af7..59ab6fe58 100644 --- a/src/langbot/pkg/api/http/service/bot.py +++ b/src/langbot/pkg/api/http/service/bot.py @@ -52,13 +52,13 @@ class BotService: return None def _adapter_declares_webhook_url(self, adapter_name: str) -> bool: - """Whether the adapter manifest declares a generated webhook URL config item.""" + """Whether the manifest needs a generated public bot ingress URL.""" component = self._get_adapter_component(adapter_name) if component is None: return False for config_item in component.spec.get('config', []): - if config_item.get('type') == 'webhook-url': + if config_item.get('type') == 'webhook-url' or config_item.get('name') == '__system.onebot_ws_url': return True return False diff --git a/src/langbot/pkg/platform/adapter_names.py b/src/langbot/pkg/platform/adapter_names.py index 448182af6..be5c1f56b 100644 --- a/src/langbot/pkg/platform/adapter_names.py +++ b/src/langbot/pkg/platform/adapter_names.py @@ -2,7 +2,6 @@ OMNI_ADAPTER_NAMES = frozenset( { - 'aiocqhttp', 'telegram', 'discord', 'qqofficial', @@ -20,6 +19,8 @@ OMNI_ADAPTER_NAMES = frozenset( def canonical_adapter_name(name: str) -> str: """Accept previous built-in IDs without renaming legacy or custom adapters.""" + if name in ('aiocqhttp-omni', 'aiocqhttp-eba'): + return 'onebot-v11' if name.endswith('-eba') and name[:-4] in OMNI_ADAPTER_NAMES: return f'{name[:-4]}-omni' return name diff --git a/src/langbot/pkg/platform/adapters/aiocqhttp/adapter.py b/src/langbot/pkg/platform/adapters/aiocqhttp/adapter.py index 7d97cac81..6840d7131 100644 --- a/src/langbot/pkg/platform/adapters/aiocqhttp/adapter.py +++ b/src/langbot/pkg/platform/adapters/aiocqhttp/adapter.py @@ -2,6 +2,12 @@ from __future__ import annotations import asyncio import traceback +import hmac +from urllib.parse import urlsplit + +from websockets.asyncio.client import connect +from langbot.pkg.core.task_boundary import create_detached_task +from langbot.pkg.platform.adapters.aiocqhttp.transport import OneBotClient, OneBotConnection import typing import aiocqhttp @@ -20,7 +26,10 @@ from langbot_plugin.api.entities.builtin.platform.errors import NotSupportedErro class AiocqhttpAdapter(AiocqhttpAPIMixin, abstract_platform_adapter.AbstractPlatformAdapter): - bot: aiocqhttp.CQHttp = pydantic.Field(exclude=True) + bot: typing.Any = pydantic.Field(exclude=True) + _runtime: typing.Any = pydantic.PrivateAttr(default=None) + _stopped: asyncio.Event = pydantic.PrivateAttr(default_factory=asyncio.Event) + _connection_task: typing.Any = pydantic.PrivateAttr(default=None) message_converter: AiocqhttpMessageConverter = AiocqhttpMessageConverter() event_converter: AiocqhttpEventConverter = AiocqhttpEventConverter() @@ -36,24 +45,64 @@ class AiocqhttpAdapter(AiocqhttpAPIMixin, abstract_platform_adapter.AbstractPlat arbitrary_types_allowed = True def __init__(self, config: dict, logger: abstract_platform_logger.AbstractEventLogger): - run_config = dict(config) + config = dict(config) + config.setdefault('connection_mode', 'reverse_ws') + if not isinstance(config.get('access-token'), str) or not config['access-token'].strip(): + raise ValueError('OneBot requires a non-empty access token') + if config['connection_mode'] not in ('reverse_ws', 'forward_ws'): + raise ValueError('Unsupported OneBot connection mode') + if config['connection_mode'] == 'forward_ws': + url = urlsplit(config.get('ws_url', '')) + if url.scheme not in ('ws', 'wss') or not url.hostname or url.username or url.password or url.fragment: + raise ValueError('OneBot ws_url must be a ws:// or wss:// endpoint without embedded credentials') + super().__init__(config=config, logger=logger, bot=OneBotClient(), bot_account_id='', listeners={}) - async def shutdown_trigger_placeholder(): - while True: - await asyncio.sleep(1) + def bind_runtime(self, runtime): + self._runtime = runtime - run_config['shutdown_trigger'] = shutdown_trigger_placeholder - access_token = run_config.pop('access-token', '') or None - bot = aiocqhttp.CQHttp(access_token=access_token) + async def _validate_connection(self): + if self._stopped.is_set() or self._runtime is None: + raise RuntimeError('OneBot runtime is inactive') + current = await self._runtime.ap.platform_mgr.resolve_public_bot(self._runtime.bot_entity.uuid) + if current is not self._runtime or not current.enable: + raise RuntimeError('OneBot runtime placement changed') - super().__init__( - config=run_config, - logger=logger, - bot=bot, - bot_account_id='', - listeners={}, + def authenticate(self, headers): + token = str(self.config.get('access-token', '')) + supplied = headers.get('Authorization', '') + return bool(token) and hmac.compare_digest(supplied.encode(), ('Bearer ' + token).encode()) + + async def serve_connection(self, send, receive, close, self_id=''): + await self._validate_connection() + if self.bot.connection is not None: + raise RuntimeError('OneBot already has an active connection') + expected = str(self.config.get('self_id', '') or '') + if expected and self_id and expected != self_id: + raise ValueError('OneBot account mismatch') + connection = OneBotConnection( + send, + receive, + close, + self._receive_event, + self._validate_connection, + self_id=expected or self_id, ) - self._register_native_handlers() + self.bot.connection = connection + self.bot_account_id = expected or self_id + self._lookup.clear() + task = create_detached_task(connection.run(), workspace_uuid=self._runtime.workspace_uuid) + self._connection_task = task + try: + await task + finally: + await connection.close() + if self.bot.connection is connection: + self.bot.connection = None + self._connection_task = None + self._lookup.clear() + + async def _receive_event(self, payload): + await self._handle_native_event(aiocqhttp.Event.from_payload(payload)) def get_supported_events(self) -> list[str]: return [ @@ -121,33 +170,51 @@ class AiocqhttpAdapter(AiocqhttpAPIMixin, abstract_platform_adapter.AbstractPlat self.listeners.pop(event_type, None) async def run_async(self): - await self.bot._server_app.run_task(**self.config) + if not self.config.get('access-token'): + raise ValueError('OneBot requires a non-empty access token') + if self.config['connection_mode'] == 'reverse_ws': + await self._stopped.wait() + return + delay = 1 + while not self._stopped.is_set(): + await self._validate_connection() + try: + async with connect( + self.config['ws_url'], + additional_headers={'Authorization': 'Bearer ' + self.config['access-token']}, + open_timeout=10, + close_timeout=5, + ping_interval=20, + ping_timeout=20, + max_size=8 * 1024 * 1024, + proxy=None, + ) as socket: + delay = 1 + await self.serve_connection(socket.send, socket.recv, socket.close) + except asyncio.CancelledError: + raise + except Exception as exc: + await self.logger.warning(f'OneBot connection lost ({type(exc).__name__}); reconnecting') + try: + await asyncio.wait_for(self._stopped.wait(), timeout=delay) + except TimeoutError: + pass + delay = min(delay * 2, 30) async def kill(self) -> bool: + self._stopped.set() + task = self._connection_task + if task is not None: + task.cancel() + await asyncio.gather(task, return_exceptions=True) self._lookup.clear() - return False - - def _register_native_handlers(self): - @self.bot.on_message() - async def on_message(event: aiocqhttp.Event): - await self._handle_native_event(event) - - @self.bot.on_notice() - async def on_notice(event: aiocqhttp.Event): - await self._handle_native_event(event) - - @self.bot.on_request() - async def on_request(event: aiocqhttp.Event): - await self._handle_native_event(event) - - @self.bot.on_websocket_connection - async def on_websocket_connection(event: aiocqhttp.Event): - self.bot_account_id = str(getattr(event, 'self_id', '') or self.bot_account_id) - await self.logger.info(f'WebSocket connection established, bot id: {self.bot_account_id}') - await self._dispatch_native_event(event) + return True async def _handle_native_event(self, event: aiocqhttp.Event): self.bot_account_id = str(getattr(event, 'self_id', '') or self.bot_account_id) + # Connection metadata belongs to the adapter, never business event routing. + if getattr(event, 'type', None) == 'meta_event': + return if getattr(event, 'type', None) == 'message' and str(getattr(event, 'user_id', '')) == self.bot_account_id: return try: diff --git a/src/langbot/pkg/platform/adapters/aiocqhttp/event_converter.py b/src/langbot/pkg/platform/adapters/aiocqhttp/event_converter.py index d2c950c5f..d1c39d277 100644 --- a/src/langbot/pkg/platform/adapters/aiocqhttp/event_converter.py +++ b/src/langbot/pkg/platform/adapters/aiocqhttp/event_converter.py @@ -34,8 +34,6 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter): return AiocqhttpEventConverter.notice_to_eba(event, bot_user_id) if event_type == 'request': return AiocqhttpEventConverter.request_to_eba(event) - if event_type == 'meta_event': - return AiocqhttpEventConverter.platform_specific(event, f'meta.{getattr(event, "detail_type", "")}') return None @staticmethod @@ -90,7 +88,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter): ) return platform_events.MessageReceivedEvent( type='message.received', - adapter_name='aiocqhttp-omni', + adapter_name='onebot-v11', message_id=getattr(event, 'message_id', ''), message_chain=message_chain, sender=sender, @@ -111,7 +109,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter): if notice_type in ('group_recall', 'friend_recall'): return platform_events.MessageDeletedEvent( type='message.deleted', - adapter_name='aiocqhttp-omni', + adapter_name='onebot-v11', message_id=getattr(event, 'message_id', ''), operator=AiocqhttpEventConverter.user(getattr(event, 'operator_id', None)), chat_type=platform_entities.ChatType.GROUP @@ -129,7 +127,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter): if AiocqhttpEventConverter._is_bot_user(getattr(event, 'user_id', None), bot_user_id, event): return platform_events.BotInvitedToGroupEvent( type='bot.invited_to_group', - adapter_name='aiocqhttp-omni', + adapter_name='onebot-v11', group=group, inviter=AiocqhttpEventConverter.user(inviter_id) if inviter_id else None, timestamp=float(getattr(event, 'time', 0) or 0), @@ -137,7 +135,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter): ) return platform_events.MemberJoinedEvent( type='group.member_joined', - adapter_name='aiocqhttp-omni', + adapter_name='onebot-v11', group=group, member=user, inviter=AiocqhttpEventConverter.user(inviter_id) if inviter_id else None, @@ -151,7 +149,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter): if AiocqhttpEventConverter._is_bot_user(getattr(event, 'user_id', None), bot_user_id, event): return platform_events.BotRemovedFromGroupEvent( type='bot.removed_from_group', - adapter_name='aiocqhttp-omni', + adapter_name='onebot-v11', group=group, operator=operator, timestamp=float(getattr(event, 'time', 0) or 0), @@ -159,7 +157,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter): ) return platform_events.MemberLeftEvent( type='group.member_left', - adapter_name='aiocqhttp-omni', + adapter_name='onebot-v11', group=group, member=AiocqhttpEventConverter.user(getattr(event, 'user_id', '')), is_kicked=getattr(event, 'sub_type', '') in ('kick', 'kick_me'), @@ -175,7 +173,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter): event_cls = platform_events.BotMutedEvent if duration > 0 else platform_events.BotUnmutedEvent kwargs: dict[str, typing.Any] = { 'type': 'bot.muted' if duration > 0 else 'bot.unmuted', - 'adapter_name': 'aiocqhttp-omni', + 'adapter_name': 'onebot-v11', 'group': group, 'operator': operator, 'timestamp': float(getattr(event, 'time', 0) or 0), @@ -187,7 +185,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter): if duration > 0: return platform_events.MemberBannedEvent( type='group.member_banned', - adapter_name='aiocqhttp-omni', + adapter_name='onebot-v11', group=group, member=AiocqhttpEventConverter.user(getattr(event, 'user_id', '')), operator=operator, @@ -198,7 +196,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter): if notice_type == 'friend_add': return platform_events.FriendAddedEvent( type='friend.added', - adapter_name='aiocqhttp-omni', + adapter_name='onebot-v11', user=AiocqhttpEventConverter.user(getattr(event, 'user_id', '')), timestamp=float(getattr(event, 'time', 0) or 0), source_platform_object=event, @@ -211,7 +209,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter): if request_type == 'friend': return platform_events.FriendRequestReceivedEvent( type='friend.request_received', - adapter_name='aiocqhttp-omni', + adapter_name='onebot-v11', request_id=getattr(event, 'flag', ''), user=AiocqhttpEventConverter.user(getattr(event, 'user_id', '')), message=getattr(event, 'comment', None), @@ -221,7 +219,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter): if request_type == 'group' and getattr(event, 'sub_type', '') == 'invite': return platform_events.BotInvitedToGroupEvent( type='bot.invited_to_group', - adapter_name='aiocqhttp-omni', + adapter_name='onebot-v11', group=AiocqhttpEventConverter.group_from_event(event), inviter=AiocqhttpEventConverter.user(getattr(event, 'user_id', '')), request_id=getattr(event, 'flag', ''), @@ -258,7 +256,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter): def platform_specific(event: aiocqhttp.Event, action: str) -> platform_events.PlatformSpecificEvent: return platform_events.PlatformSpecificEvent( type='platform.specific', - adapter_name='aiocqhttp-omni', + adapter_name='onebot-v11', action=action, data={key: value for key, value in dict(event).items() if key not in {'message'}}, timestamp=float(getattr(event, 'time', 0) or 0), diff --git a/src/langbot/pkg/platform/adapters/aiocqhttp/manifest.yaml b/src/langbot/pkg/platform/adapters/aiocqhttp/manifest.yaml index e14581cfb..3b5dec81b 100644 --- a/src/langbot/pkg/platform/adapters/aiocqhttp/manifest.yaml +++ b/src/langbot/pkg/platform/adapters/aiocqhttp/manifest.yaml @@ -2,7 +2,7 @@ apiVersion: v1 kind: MessagePlatformAdapter metadata: - name: aiocqhttp-omni + name: onebot-v11 label: en_US: OneBot v11 zh_Hans: OneBot v11 @@ -22,42 +22,59 @@ spec: en: https://langbot.app/docs/en/usage/platforms/qq/aiocqhttp/napcat ja: https://langbot.app/docs/ja/usage/platforms/qq/aiocqhttp/napcat config: - - name: host - label: - en_US: Host - zh_Hans: 主机 - zh_Hant: 主機 - description: - en_US: The host that OneBot v11 listens on for reverse WebSocket connections. Unless you know what you're doing, use 0.0.0.0 - zh_Hans: OneBot v11 反向 WebSocket 监听主机,除非你知道自己在做什么,否则请写 0.0.0.0 - zh_Hant: OneBot v11 反向 WebSocket 監聽主機,除非你知道自己在做什麼,否則請填 0.0.0.0 - type: string + - name: connection_mode + label: {en_US: Connection mode, zh_Hans: 连接模式, zh_Hant: 連線模式, ja_JP: 接続方式} + type: select required: true - default: 0.0.0.0 - - name: port - label: - en_US: Port - zh_Hans: 端口 - zh_Hant: 連接埠 + default: reverse_ws + options: + - name: reverse_ws + label: {en_US: 'Reverse WS — protocol endpoint connects to LangBot', zh_Hans: '反向 WS — 协议端连接 LangBot', zh_Hant: '反向 WS — 協定端連接 LangBot', ja_JP: '逆方向 WS — プロトコル側から LangBot に接続'} + - name: forward_ws + label: {en_US: 'Forward WS — LangBot connects to protocol endpoint', zh_Hans: '正向 WS — LangBot 连接协议端', zh_Hant: '正向 WS — LangBot 連接協定端', ja_JP: '順方向 WS — LangBot からプロトコル側に接続'} description: - en_US: Reverse WebSocket listen port - zh_Hans: 反向 WebSocket 监听端口 - zh_Hant: 反向 WebSocket 監聽連接埠 - type: integer - required: true - default: 2280 - - name: access-token - label: - en_US: Access Token - zh_Hans: 访问令牌 - zh_Hant: 存取令牌 + en_US: Reverse WS uses LangBot's main service port. Forward WS requires a protocol endpoint reachable from LangBot. HTTP is not supported. + zh_Hans: 反向 WS 由协议端连接 LangBot 主服务端口,无需额外监听端口;正向 WS 由 LangBot 连接可访问的协议端。不支持 HTTP。 + zh_Hant: 反向 WS 由協定端連接 LangBot 主服務連接埠;正向 WS 由 LangBot 連接可存取的協定端。不支援 HTTP。 + ja_JP: 逆方向 WS は LangBot のメインポートを使用。順方向 WS は LangBot から到達可能なプロトコル側に接続します。HTTP は非対応です。 + - name: __system.onebot_ws_url + label: {en_US: Protocol endpoint connection URL, zh_Hans: 协议端连接地址, zh_Hant: 協定端連線位址, ja_JP: プロトコル側の接続先 URL} description: - en_US: Custom connection token for the protocol endpoint. Leave empty if the endpoint has no token configured - zh_Hans: 自定义的协议端连接令牌;若协议端未设置,则不填 - zh_Hant: 自訂的協定端連線令牌;若協定端未設定,則不填 + en_US: Save the bot, then copy this URL into the protocol endpoint's WebSocket client (Universal). Old host/port settings are ignored; update the protocol endpoint URL. + zh_Hans: 保存机器人后,将此地址填入协议端的 WebSocket 客户端(Universal)配置。旧 host/port 配置不再生效,请同步修改协议端地址。 + zh_Hant: 儲存後將此位址填入協定端 WebSocket 用戶端(Universal)。舊 host/port 不再生效,請同步更新協定端位址。 + ja_JP: 保存後、この URL をプロトコル側の WebSocket クライアント(Universal)に設定。旧 host/port は無効です。接続先を更新してください。 type: string required: false - default: "" + default: '' + show_if: {field: connection_mode, operator: eq, value: reverse_ws} + - name: ws_url + label: {en_US: Protocol endpoint WebSocket URL, zh_Hans: 协议端 WebSocket 地址, zh_Hant: 協定端 WebSocket 位址, ja_JP: プロトコル側 WebSocket URL} + description: + en_US: LangBot connects to this ws:// or wss:// Universal endpoint. It must be reachable from LangBot. + zh_Hans: LangBot 将主动连接此 ws:// 或 wss:// 地址,使用协议端同时提供事件和 API 的 Universal 端点。请确保 LangBot 能访问它。 + zh_Hant: LangBot 主動連接此 ws:// 或 wss:// Universal 位址,請確保可存取。 + ja_JP: LangBot がこの ws:// または wss:// Universal URL に接続します。到達可能な URL を指定してください。 + type: string + required: true + default: '' + show_if: {field: connection_mode, operator: eq, value: forward_ws} + - name: access-token + label: {en_US: Access token, zh_Hans: 访问令牌, zh_Hant: 存取權杖, ja_JP: アクセストークン} + description: + en_US: Required in both modes. Configure the same secret token on LangBot and the protocol endpoint. Sent using Authorization Bearer. + zh_Hans: 两种模式均必填。在 LangBot 与协议端设置相同的保密令牌,连接时使用 Authorization Bearer 验证。 + zh_Hant: 兩種模式皆必填。在 LangBot 與協定端設定相同的保密權杖。 + ja_JP: 両方式で必須。LangBot とプロトコル側に同じ秘密トークンを設定してください。 + type: string + required: true + default: '' + - name: self_id + label: {en_US: Expected bot account (optional), zh_Hans: 预期机器人账号(可选), zh_Hant: 預期機器人帳號(選填), ja_JP: ボットアカウント(任意)} + description: {en_US: Reject connections or events from a different account., zh_Hans: 拒绝其他账号的连接或事件。, zh_Hant: 拒絕其他帳號的連線或事件。, ja_JP: 異なるアカウントの接続やイベントを拒否します。} + type: string + required: false + default: '' supported_events: - message.received diff --git a/src/langbot/pkg/platform/adapters/aiocqhttp/transport.py b/src/langbot/pkg/platform/adapters/aiocqhttp/transport.py new file mode 100644 index 000000000..5499b7981 --- /dev/null +++ b/src/langbot/pkg/platform/adapters/aiocqhttp/transport.py @@ -0,0 +1,125 @@ +"""OneBot v11 duplex transport, independent of an HTTP server.""" + +from __future__ import annotations + +import asyncio +import functools +import json +import uuid + +from aiocqhttp.exceptions import ActionFailed, ApiNotAvailable + + +class OneBotConnection: + def __init__(self, send, receive, close, on_event, validate, *, self_id='', timeout=30): + self.send = send + self.receive = receive + self.close_socket = close + self.on_event = on_event + self.validate = validate + self.self_id = self_id + self.timeout = timeout + self.pending = {} + self.events = asyncio.Queue(maxsize=256) + self.closed = False + self.send_lock = asyncio.Lock() + + async def call_action(self, action, **params): + await self.validate() + if self.closed: + raise ApiNotAvailable('OneBot connection closed') + if len(self.pending) >= 256: + raise ApiNotAvailable('Too many pending OneBot actions') + echo = uuid.uuid4().hex + future = asyncio.get_running_loop().create_future() + self.pending[echo] = future + try: + async with asyncio.timeout(self.timeout): + async with self.send_lock: + await self.validate() + await self.send(json.dumps({'action': action, 'params': params, 'echo': echo})) + result = await future + if result.get('status') not in ('ok', 'async') or result.get('retcode') not in (0, 1): + raise ActionFailed(result) + return result.get('data') + finally: + self.pending.pop(echo, None) + if not future.done(): + future.cancel() + + async def _read(self): + while True: + raw = await self.receive() + if len(raw) > 8 * 1024 * 1024: + raise ValueError('OneBot frame too large') + data = json.loads(raw) + if not isinstance(data, dict): + raise ValueError('OneBot frame must be an object') + await self.validate() + if 'post_type' in data: + account = str(data.get('self_id', '')) + if not account or (self.self_id and account != self.self_id): + raise ValueError('OneBot account mismatch') + self.self_id = account + # Never block responses behind an event that itself calls an API. + self.events.put_nowait(data) + else: + echo = data.get('echo') + future = self.pending.get(echo) if isinstance(echo, str) else None + if future is not None and not future.done(): + future.set_result(data) + + async def _consume(self): + while True: + data = await self.events.get() + await self.validate() + await self.on_event(data) + + async def _watch_placement(self): + while True: + await asyncio.sleep(5) + await self.validate() + + async def run(self): + tasks = [asyncio.create_task(coro()) for coro in (self._read, self._consume, self._watch_placement)] + try: + done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) + for task in done: + task.result() + finally: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + await self.close() + + async def close(self): + if self.closed: + return + self.closed = True + for future in self.pending.values(): + if not future.done(): + future.set_exception(ApiNotAvailable('OneBot connection closed')) + self.pending.clear() + await self.close_socket() + + +class OneBotClient: + """Keep the existing converter/API call surface without CQHttp's server.""" + + def __init__(self): + self.connection = None + + async def call_action(self, action, **params): + connection = self.connection + if connection is None: + raise ApiNotAvailable('OneBot protocol endpoint is not connected') + return await connection.call_action(action, **params) + + def __getattr__(self, action): + if action.startswith('_'): + raise AttributeError(action) + return functools.partial(self.call_action, action) + + async def send(self, event, message, **kwargs): + params = {key: event[key] for key in ('message_type', 'user_id', 'group_id', 'discuss_id') if key in event} + return await self.call_action('send_msg', **(params | {'message': message} | kwargs)) diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index 9ffc4cf57..dfb99d292 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -2357,6 +2357,9 @@ class PlatformManager: execution_context=execution_context, ) + if hasattr(adapter_inst, 'bind_runtime'): + adapter_inst.bind_runtime(runtime_bot) + await runtime_bot.initialize() bot_key = self._runtime_bot_key(runtime_bot) diff --git a/tests/e2e/live_aiocqhttp_eba_probe.py b/tests/e2e/live_aiocqhttp_eba_probe.py index 493d208e3..dc193c9bd 100644 --- a/tests/e2e/live_aiocqhttp_eba_probe.py +++ b/tests/e2e/live_aiocqhttp_eba_probe.py @@ -7,6 +7,7 @@ import json import time from collections import Counter from pathlib import Path +from types import SimpleNamespace from langbot.pkg.platform.adapters.aiocqhttp.adapter import AiocqhttpAdapter from langbot_plugin.api.definition.abstract.platform.event_logger import AbstractEventLogger @@ -36,9 +37,8 @@ def dump_event(event: platform_events.Event) -> dict: async def main(): parser = argparse.ArgumentParser(description='Live OneBot v11 / aiocqhttp EBA probe for Matcha or a real endpoint.') - parser.add_argument('--host', default='127.0.0.1') - parser.add_argument('--port', type=int, default=2280) - parser.add_argument('--access-token', default='') + parser.add_argument('--ws-url', required=True, help='Protocol endpoint Universal WebSocket URL') + parser.add_argument('--access-token', required=True) parser.add_argument('--timeout', type=int, default=120) parser.add_argument('--target-type', choices=['private', 'group'], default=None) parser.add_argument('--target-id', default=None) @@ -54,10 +54,18 @@ async def main(): out_fp = out_path.open('a', encoding='utf-8') adapter = AiocqhttpAdapter( - {'host': args.host, 'port': args.port, 'access-token': args.access_token}, + {'connection_mode': 'forward_ws', 'ws_url': args.ws_url, 'access-token': args.access_token}, ProbeLogger(), ) + # This standalone converter probe uses a single explicit local runtime. + runtime = SimpleNamespace(enable=True, workspace_uuid='probe', bot_entity=SimpleNamespace(uuid='probe')) + + async def resolve_probe(_): + return runtime + + runtime.ap = SimpleNamespace(platform_mgr=SimpleNamespace(resolve_public_bot=resolve_probe)) + adapter.bind_runtime(runtime) observed: list[platform_events.Event] = [] first_message = asyncio.Event() @@ -79,7 +87,7 @@ async def main(): return None task = asyncio.create_task(adapter.run_async()) - print(f'Listening on ws://{args.host}:{args.port}/ws/ . Trigger events from Matcha now.') + print('Connecting to the protocol endpoint. Trigger events now.') api_results: dict[str, str] = {} try: diff --git a/tests/unit_tests/platform/test_aiocqhttp_eba_adapter.py b/tests/unit_tests/platform/test_aiocqhttp_eba_adapter.py index 584d39d91..58473bf00 100644 --- a/tests/unit_tests/platform/test_aiocqhttp_eba_adapter.py +++ b/tests/unit_tests/platform/test_aiocqhttp_eba_adapter.py @@ -34,7 +34,7 @@ class DummyLogger(AbstractEventLogger): def make_adapter() -> AiocqhttpAdapter: - return AiocqhttpAdapter({'host': '127.0.0.1', 'port': 2280, 'access-token': ''}, DummyLogger()) + return AiocqhttpAdapter({'host': '127.0.0.1', 'port': 2280, 'access-token': 'test-secret'}, DummyLogger()) def manifest() -> dict: @@ -48,13 +48,40 @@ def manifest() -> dict: / 'aiocqhttp' / 'manifest.yaml' ) - return yaml.safe_load(manifest_path.read_text()) + return yaml.safe_load(manifest_path.read_text(encoding='utf-8')) def onebot_event(payload: dict) -> aiocqhttp.Event: return aiocqhttp.Event.from_payload(payload) +@pytest.mark.parametrize('meta_type', ['heartbeat', 'lifecycle', 'future_transport_metadata']) +async def test_protocol_metadata_never_reaches_business_listeners(meta_type): + adapter = make_adapter() + listener = AsyncMock() + adapter.register_listener(platform_events.EBAEvent, listener) + payload = {'post_type': 'meta_event', 'meta_event_type': meta_type, 'self_id': 42, 'time': 1700000000} + assert await AiocqhttpEventConverter.target2yiri(onebot_event(payload)) is None + await adapter._receive_event(payload) + listener.assert_not_awaited() + assert adapter.bot_account_id == '42' + + # Genuine platform-specific user interactions still reach business routing. + await adapter._receive_event( + { + 'post_type': 'notice', + 'notice_type': 'notify', + 'sub_type': 'poke', + 'self_id': 42, + 'time': 1700000001, + 'user_id': 10001, + 'target_id': 42, + } + ) + listener.assert_awaited_once() + assert listener.await_args.args[0].type == 'platform.specific' + + def test_aiocqhttp_supported_events_match_manifest(): assert make_adapter().get_supported_events() == manifest()['spec']['supported_events'] @@ -255,7 +282,7 @@ async def test_aiocqhttp_event_converter_maps_private_and_group_messages(): assert isinstance(private_event, platform_events.MessageReceivedEvent) assert private_event.type == 'message.received' - assert private_event.adapter_name == 'aiocqhttp-omni' + assert private_event.adapter_name == 'onebot-v11' assert private_event.chat_type == platform_entities.ChatType.PRIVATE assert private_event.chat_id == 10001 assert private_event.sender.nickname == 'Alice' diff --git a/tests/unit_tests/platform/test_botmgr_tenancy.py b/tests/unit_tests/platform/test_botmgr_tenancy.py index 9e6b61667..6a1b0a6d2 100644 --- a/tests/unit_tests/platform/test_botmgr_tenancy.py +++ b/tests/unit_tests/platform/test_botmgr_tenancy.py @@ -243,7 +243,8 @@ async def test_reload_stops_and_drops_existing_platform_runtimes(): @pytest.mark.asyncio @pytest.mark.parametrize( ('saved_adapter', 'registered_adapter'), - [('probe', 'probe')] + [(f'{name}-eba', f'{name}-omni') for name in sorted(OMNI_ADAPTER_NAMES)], + [('probe', 'probe'), ('aiocqhttp-eba', 'onebot-v11'), ('aiocqhttp-omni', 'onebot-v11')] + + [(f'{name}-eba', f'{name}-omni') for name in sorted(OMNI_ADAPTER_NAMES)], ) async def test_cloud_startup_reuses_validated_platform_binding(saved_adapter, registered_adapter): class TenantUow: diff --git a/tests/unit_tests/platform/test_omni_adapter_names.py b/tests/unit_tests/platform/test_omni_adapter_names.py index 5b8aa038c..7651b26bf 100644 --- a/tests/unit_tests/platform/test_omni_adapter_names.py +++ b/tests/unit_tests/platform/test_omni_adapter_names.py @@ -13,13 +13,17 @@ from langbot.pkg.platform.botmgr import PlatformManager @pytest.mark.asyncio -@pytest.mark.parametrize('name', sorted(OMNI_ADAPTER_NAMES)) -async def test_saved_adapter_ids_are_normalized_across_bot_api(name): - saved = {'uuid': 'bot', 'adapter': f'{name}-eba', 'adapter_config': {}} +@pytest.mark.parametrize( + 'saved_name,public_name', + [(f'{name}-eba', f'{name}-omni') for name in sorted(OMNI_ADAPTER_NAMES)] + + [('aiocqhttp-eba', 'onebot-v11'), ('aiocqhttp-omni', 'onebot-v11'), ('onebot-v11', 'onebot-v11')], +) +async def test_saved_adapter_ids_are_normalized_across_bot_api(saved_name, public_name): + saved = {'uuid': 'bot', 'adapter': saved_name, 'adapter_config': {}} component = SimpleNamespace( - metadata=SimpleNamespace(name=f'{name}-omni'), + metadata=SimpleNamespace(name=public_name), spec={'config': [{'type': 'webhook-url'}]}, - to_plain_dict=lambda: {'metadata': {'name': f'{name}-omni'}}, + to_plain_dict=lambda: {'metadata': {'name': public_name}}, ) result = SimpleNamespace(all=lambda: [saved], first=lambda: saved) application = SimpleNamespace( @@ -30,18 +34,20 @@ async def test_saved_adapter_ids_are_normalized_across_bot_api(name): ), ) service = BotService(application) - assert (await service.get_bots('workspace'))[0]['adapter'] == f'{name}-omni' - assert (await service.get_bot('workspace', 'bot'))['adapter'] == f'{name}-omni' + assert (await service.get_bots('workspace'))[0]['adapter'] == public_name + assert (await service.get_bot('workspace', 'bot'))['adapter'] == public_name assert service._adapter_declares_webhook_url(saved['adapter']) - assert (await service._prepare_bot_data('workspace', saved, include_uuid=True))['adapter'] == f'{name}-omni' - assert saved['adapter'] == f'{name}-eba' + assert (await service._prepare_bot_data('workspace', saved, include_uuid=True))['adapter'] == public_name + assert saved['adapter'] == saved_name manager = PlatformManager(application) manager.adapter_components = [component] assert manager.get_available_adapter_manifest_by_name(saved['adapter']) is component - assert manager.get_available_adapter_info_by_name(saved['adapter'])['metadata']['name'] == f'{name}-omni' + assert manager.get_available_adapter_info_by_name(saved['adapter'])['metadata']['name'] == public_name -@pytest.mark.parametrize('name', ['telegram', 'custom-eba', 'my-telegram-eba', 'telegram-omni', 'websocket']) +@pytest.mark.parametrize( + 'name', ['aiocqhttp', 'onebot-v11', 'telegram', 'custom-eba', 'my-telegram-eba', 'telegram-omni', 'websocket'] +) def test_legacy_and_custom_adapter_ids_are_unchanged(name): assert canonical_adapter_name(name) == name @@ -49,14 +55,23 @@ def test_legacy_and_custom_adapter_ids_are_unchanged(name): def test_discovered_manifests_expose_only_omni_names(): root = Path(__file__).resolve().parents[3] / 'src/langbot/pkg/platform/adapters' for name in OMNI_ADAPTER_NAMES: - metadata = yaml.safe_load((root / name / 'manifest.yaml').read_text())['metadata'] + metadata = yaml.safe_load((root / name / 'manifest.yaml').read_text(encoding='utf-8'))['metadata'] assert metadata['name'] == f'{name}-omni' @pytest.mark.asyncio -@pytest.mark.parametrize('disabled_name', ['telegram-eba', 'telegram-omni']) -async def test_old_disabled_adapter_config_still_hides_omni(disabled_name): - component = SimpleNamespace(metadata=SimpleNamespace(name='telegram-omni')) +@pytest.mark.parametrize( + 'disabled_name,public_name', + [ + ('telegram-eba', 'telegram-omni'), + ('telegram-omni', 'telegram-omni'), + ('aiocqhttp-eba', 'onebot-v11'), + ('aiocqhttp-omni', 'onebot-v11'), + ('onebot-v11', 'onebot-v11'), + ], +) +async def test_old_disabled_adapter_config_still_hides_omni(disabled_name, public_name): + component = SimpleNamespace(metadata=SimpleNamespace(name=public_name)) application = SimpleNamespace( storage_mgr=SimpleNamespace(storage_provider=SimpleNamespace(delete_dir_recursive=AsyncMock())), instance_config=SimpleNamespace(data={'system': {'disabled_adapters': [disabled_name]}}), diff --git a/tests/unit_tests/platform/test_onebot_transport.py b/tests/unit_tests/platform/test_onebot_transport.py new file mode 100644 index 000000000..114cb450a --- /dev/null +++ b/tests/unit_tests/platform/test_onebot_transport.py @@ -0,0 +1,329 @@ +import asyncio +import json +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from quart import Quart +import quart_cors +from quart.testing.connections import WebsocketResponseError +from aiocqhttp.exceptions import ActionFailed, ApiNotAvailable +from websockets.asyncio.server import serve + +from langbot.pkg.platform.adapters.aiocqhttp.transport import OneBotConnection +from langbot.pkg.api.http.controller.groups.webhooks import WebhookRouterGroup +from tests.unit_tests.platform.test_aiocqhttp_eba_adapter import make_adapter + + +def bind(adapter): + runtime = SimpleNamespace( + adapter=adapter, enable=True, workspace_uuid='workspace-a', bot_entity=SimpleNamespace(uuid='bot-a') + ) + ap = SimpleNamespace(platform_mgr=SimpleNamespace(resolve_public_bot=AsyncMock(return_value=runtime))) + runtime.ap = ap + adapter.bind_runtime(runtime) + adapter.config['access-token'] = 'secret' + return runtime, ap + + +async def test_echo_out_of_order_and_event_api_does_not_deadlock(): + incoming, outgoing = asyncio.Queue(), asyncio.Queue() + handled = asyncio.Event() + + async def event(data): + assert await connection.call_action('get_group_info', group_id=1) == {'name': 'group'} + handled.set() + + connection = OneBotConnection(outgoing.put, incoming.get, AsyncMock(), event, AsyncMock()) + task = asyncio.create_task(connection.run()) + first = asyncio.create_task(connection.call_action('first')) + second = asyncio.create_task(connection.call_action('second')) + a, b = json.loads(await outgoing.get()), json.loads(await outgoing.get()) + for request in (b, a): + await incoming.put( + json.dumps({'echo': request['echo'], 'status': 'ok', 'retcode': 0, 'data': request['action']}) + ) + assert await first == 'first' + assert await second == 'second' + await incoming.put(json.dumps({'post_type': 'message', 'self_id': 42})) + request = json.loads(await outgoing.get()) + await incoming.put(json.dumps({'echo': request['echo'], 'status': 'ok', 'retcode': 0, 'data': {'name': 'group'}})) + await asyncio.wait_for(handled.wait(), 1) + task.cancel() + await asyncio.gather(task, return_exceptions=True) + assert not connection.pending + + +async def test_timeout_disconnect_and_failure(): + outgoing = asyncio.Queue() + connection = OneBotConnection(outgoing.put, AsyncMock(), AsyncMock(), AsyncMock(), AsyncMock(), timeout=0.01) + with pytest.raises(TimeoutError): + await connection.call_action('timeout') + assert not connection.pending + await outgoing.get() + connection.timeout = 2 + action = asyncio.create_task(connection.call_action('failure')) + request = json.loads(await outgoing.get()) + connection.pending[request['echo']].set_result({'status': 'failed', 'retcode': 100}) + with pytest.raises(ActionFailed): + await action + action = asyncio.create_task(connection.call_action('disconnect')) + await outgoing.get() + await connection.close() + with pytest.raises(ApiNotAvailable): + await action + + +@pytest.mark.parametrize('payload', ['[]', 'broken', '{"post_type":"message","self_id":99}']) +async def test_invalid_frames_and_account_mismatch(payload): + incoming = asyncio.Queue() + await incoming.put(payload) + connection = OneBotConnection(AsyncMock(), incoming.get, AsyncMock(), AsyncMock(), AsyncMock(), self_id='42') + with pytest.raises(ValueError): + await connection.run() + assert connection.closed + + +async def test_main_port_reverse_ws_auth_and_placement(): + adapter = make_adapter() + runtime, ap = bind(adapter) + app = Quart(__name__) + quart_cors.cors(app, allow_origin='*') + await WebhookRouterGroup(ap, app).initialize() + client = app.test_client() + headers = {'Authorization': 'Bearer secret', 'X-Self-ID': '42', 'X-Client-Role': 'Universal'} + for changes, status in [({'Authorization': 'Bearer wrong'}, 401), ({'X-Client-Role': 'Event'}, 400)]: + with pytest.raises(WebsocketResponseError) as error: + async with client.websocket('/bots/bot-a/onebot/v11/ws', headers=headers | changes): + await asyncio.sleep(0.01) + assert error.value.response.status_code == status + async with client.websocket('/bots/bot-a/onebot/v11/ws', headers=headers | {'X-Workspace-Id': 'attacker'}) as ws: + await ws.send(json.dumps({'post_type': 'meta_event', 'meta_event_type': 'heartbeat', 'self_id': 42})) + for _ in range(100): + if adapter.bot.connection: + break + await asyncio.sleep(0.001) + action = asyncio.create_task(adapter.bot.get_login_info()) + request = json.loads(await ws.receive()) + await ws.send(json.dumps({'echo': request['echo'], 'status': 'ok', 'retcode': 0, 'data': {'user_id': 42}})) + assert await asyncio.wait_for(action, 1) == {'user_id': 42} + ap.platform_mgr.resolve_public_bot.return_value = None + with pytest.raises(RuntimeError, match='placement'): + await adapter.bot.get_login_info() + assert runtime.workspace_uuid == 'workspace-a' + await adapter.kill() + + +async def test_forward_ws_round_trip_and_shutdown(): + connected = asyncio.Event() + + async def endpoint(socket): + assert socket.request.headers['Authorization'] == 'Bearer secret' + await socket.send(json.dumps({'post_type': 'meta_event', 'meta_event_type': 'heartbeat', 'self_id': 42})) + connected.set() + async for raw in socket: + request = json.loads(raw) + await socket.send( + json.dumps({'echo': request['echo'], 'status': 'ok', 'retcode': 0, 'data': {'message_id': 7}}) + ) + + async with serve(endpoint, '127.0.0.1', 0) as server: + adapter = make_adapter() + bind(adapter) + adapter.config.update( + connection_mode='forward_ws', ws_url=f'ws://127.0.0.1:{server.sockets[0].getsockname()[1]}' + ) + task = asyncio.create_task(adapter.run_async()) + await asyncio.wait_for(connected.wait(), 2) + for _ in range(100): + if adapter.bot.connection: + break + await asyncio.sleep(0.001) + assert await adapter.bot.send_private_msg(user_id=1, message='hello') == {'message_id': 7} + await adapter.kill() + await asyncio.gather(task, return_exceptions=True) + assert adapter.bot.connection is None + + +async def test_idle_placement_revocation_closes_connection(): + adapter = make_adapter() + _, ap = bind(adapter) + closed = asyncio.Event() + + async def close(): + closed.set() + + incoming = asyncio.Queue() + task = asyncio.create_task(adapter.serve_connection(AsyncMock(), incoming.get, close, '42')) + for _ in range(100): + if adapter.bot.connection: + break + await asyncio.sleep(0.001) + with pytest.raises(RuntimeError, match='active connection'): + await adapter.serve_connection(AsyncMock(), incoming.get, close, '42') + ap.platform_mgr.resolve_public_bot.return_value = None + await asyncio.wait_for(closed.wait(), 7) + with pytest.raises(RuntimeError, match='placement'): + await task + assert adapter.bot.connection is None + + +async def test_forward_reconnects_after_endpoint_disconnect(): + count = 0 + reconnected = asyncio.Event() + + async def endpoint(socket): + nonlocal count + count += 1 + if count == 1: + await socket.close() + else: + reconnected.set() + await socket.wait_closed() + + async with serve(endpoint, '127.0.0.1', 0) as server: + adapter = make_adapter() + bind(adapter) + adapter.config.update( + connection_mode='forward_ws', ws_url=f'ws://127.0.0.1:{server.sockets[0].getsockname()[1]}' + ) + task = asyncio.create_task(adapter.run_async()) + await asyncio.wait_for(reconnected.wait(), 4) + await adapter.kill() + await asyncio.gather(task, return_exceptions=True) + assert count == 2 + + +def test_legacy_manifest_keeps_independent_adapter(): + from langbot.pkg.discover.engine import ComponentDiscoveryEngine + from langbot.pkg.platform.adapters.aiocqhttp.adapter import AiocqhttpAdapter as NewAdapter + from langbot.pkg.platform.sources.aiocqhttp import AiocqhttpAdapter as LegacyAdapter + + engine = ComponentDiscoveryEngine(SimpleNamespace()) + legacy = engine.load_component_manifest('pkg/platform/sources/aiocqhttp.yaml') + modern = engine.load_component_manifest('pkg/platform/adapters/aiocqhttp/manifest.yaml') + assert legacy.get_python_component_class() is LegacyAdapter + assert modern.get_python_component_class() is NewAdapter + assert modern.metadata.name == 'onebot-v11' + assert LegacyAdapter is not NewAdapter + fields = {item['name']: item for item in legacy.spec['config']} + assert fields['port']['default'] == 2280 + assert fields['access-token']['required'] is False + assert 'connection_mode' not in fields + + +def test_configuration_rejects_invalid_credentials_and_url(): + from langbot.pkg.platform.adapters.aiocqhttp.adapter import AiocqhttpAdapter + from tests.unit_tests.platform.test_aiocqhttp_eba_adapter import DummyLogger + + with pytest.raises(ValueError, match='token'): + AiocqhttpAdapter({}, DummyLogger()) + for url in ('https://example.com', 'ws://user:password@example.com', ''): + with pytest.raises(ValueError, match='ws_url'): + AiocqhttpAdapter({'access-token': 'secret', 'connection_mode': 'forward_ws', 'ws_url': url}, DummyLogger()) + + +@pytest.mark.parametrize('mode', ['reverse_ws', 'forward_ws', 'legacy']) +async def test_message_event_to_reply_roundtrip(mode): + """Exercise real converters and callbacks against a simulated protocol peer.""" + from langbot.pkg.platform.sources.aiocqhttp import AiocqhttpAdapter as LegacyAdapter + from langbot_plugin.api.entities.builtin.platform import events, message + from tests.unit_tests.platform.test_aiocqhttp_eba_adapter import DummyLogger + + adapter = LegacyAdapter({'access-token': 'secret'}, DummyLogger()) if mode == 'legacy' else make_adapter() + finished = asyncio.Event() + received = [] + + async def callback(event, source): + received.append(str(event.message_chain)) + await source.send_message('person', '10001', message.MessageChain([message.Plain(text='roundtrip reply')])) + finished.set() + + adapter.register_listener(events.FriendMessage, callback) + + async def protocol_peer(send, receive): + await send( + json.dumps( + { + 'time': 1700000000, + 'self_id': 42, + 'post_type': 'message', + 'message_type': 'private', + 'sub_type': 'friend', + 'message_id': 9, + 'user_id': 10001, + 'message': [{'type': 'text', 'data': {'text': 'roundtrip incoming'}}], + 'raw_message': 'roundtrip incoming', + 'font': 0, + 'sender': {'user_id': 10001, 'nickname': 'Protocol test'}, + } + ) + ) + while not finished.is_set(): + request = json.loads(await asyncio.wait_for(receive(), 3)) + action = request['action'] + if action == 'send_private_msg': + assert str(request['params']['user_id']) == '10001' + assert 'roundtrip reply' in json.dumps(request['params']['message']) + data = {'message_id': 10} + elif action == 'get_stranger_info': + data = {'user_id': 10001, 'nickname': 'Protocol test'} + else: + raise AssertionError(f'Unexpected action: {action}') + await send(json.dumps({'echo': request['echo'], 'status': 'ok', 'retcode': 0, 'data': data})) + if action == 'send_private_msg': + await asyncio.wait_for(finished.wait(), 3) + + if mode == 'forward_ws': + bind(adapter) + + async def endpoint(ws): + assert ws.request.headers['Authorization'] == 'Bearer secret' + await protocol_peer(ws.send, ws.recv) + + async with serve(endpoint, '127.0.0.1', 0) as server: + adapter.config.update(connection_mode=mode, ws_url=f'ws://127.0.0.1:{server.sockets[0].getsockname()[1]}') + task = asyncio.create_task(adapter.run_async()) + try: + await asyncio.wait_for(finished.wait(), 5) + finally: + await adapter.kill() + await asyncio.gather(task, return_exceptions=True) + else: + if mode == 'legacy': + app, path = adapter.bot._server_app, '/ws/' + else: + _, ap = bind(adapter) + app, path = Quart(__name__), '/bots/bot-a/onebot/v11/ws' + await WebhookRouterGroup(ap, app).initialize() + headers = {'Authorization': 'Bearer secret', 'X-Self-ID': '42', 'X-Client-Role': 'Universal'} + async with app.test_client().websocket(path, headers=headers) as ws: + await protocol_peer(ws.send, ws.receive) + await adapter.kill() + assert received == ['roundtrip incoming'] + + +@pytest.mark.parametrize( + 'name,path', + [ + ('aiocqhttp', 'pkg/platform/sources/aiocqhttp.yaml'), + ('onebot-v11', 'pkg/platform/adapters/aiocqhttp/manifest.yaml'), + ('aiocqhttp-omni', 'pkg/platform/adapters/aiocqhttp/manifest.yaml'), + ('aiocqhttp-eba', 'pkg/platform/adapters/aiocqhttp/manifest.yaml'), + ], +) +async def test_public_url_only_available_for_new_onebot_adapter(name, path): + from langbot.pkg.discover.engine import ComponentDiscoveryEngine + from langbot.pkg.api.http.service.bot import BotService + + ap = SimpleNamespace( + instance_config=SimpleNamespace(data={'api': {'webhook_prefix': 'https://bot.example.com'}}), + platform_mgr=SimpleNamespace(get_bot_by_uuid=AsyncMock(return_value=None)), + ) + ap.discover = ComponentDiscoveryEngine(ap) + ap.discover.load_component_manifest(path) + service = BotService(ap) + service.get_bot = AsyncMock(return_value={'uuid': 'bot-a', 'adapter': name, 'adapter_config': {}}) + result = await service.get_runtime_bot_info('workspace-a', 'bot-a') + expected = 'https://bot.example.com/bots/bot-a' if name != 'aiocqhttp' else None + assert result['adapter_runtime_values']['webhook_full_url'] == expected diff --git a/web/src/app/home/bots/components/bot-form/BotForm.tsx b/web/src/app/home/bots/components/bot-form/BotForm.tsx index be107e631..0d036c221 100644 --- a/web/src/app/home/bots/components/bot-form/BotForm.tsx +++ b/web/src/app/home/bots/components/bot-form/BotForm.tsx @@ -288,6 +288,10 @@ const BotForm = forwardRef(function BotForm( () => ({ bot_uuid: initBotId, webhook_url: webhookUrl, + onebot_ws_url: webhookUrl + ? webhookUrl.replace(/^http/, 'ws').replace(/\/$/, '') + + '/onebot/v11/ws' + : '', extra_webhook_url: extraWebhookUrl, adapter_config: { title: embedTitle }, outbound_ips: systemInfo.outbound_ips, diff --git a/web/src/app/wizard/page.tsx b/web/src/app/wizard/page.tsx index 397427e0d..6083526ec 100644 --- a/web/src/app/wizard/page.tsx +++ b/web/src/app/wizard/page.tsx @@ -1823,6 +1823,10 @@ function StepBotConfig({ systemContext={{ is_wizard: true, webhook_url: webhookUrl, + onebot_ws_url: webhookUrl + ? webhookUrl.replace(/^http/, 'ws').replace(/\/$/, '') + + '/onebot/v11/ws' + : '', extra_webhook_url: extraWebhookUrl, outbound_ips: systemInfo.outbound_ips, }}