diff --git a/EvoScientist/channels/qq/__init__.py b/EvoScientist/channels/qq/__init__.py index 7f560c2..629edcc 100644 --- a/EvoScientist/channels/qq/__init__.py +++ b/EvoScientist/channels/qq/__init__.py @@ -10,8 +10,9 @@ Usage in config: from ..channel_manager import _parse_csv, register_channel from .channel import QQChannel, QQConfig +from .onboard import qr_register -__all__ = ["QQChannel", "QQConfig"] +__all__ = ["QQChannel", "QQConfig", "qr_register"] def create_from_config(config) -> QQChannel: diff --git a/EvoScientist/channels/qq/crypto.py b/EvoScientist/channels/qq/crypto.py new file mode 100644 index 0000000..325c980 --- /dev/null +++ b/EvoScientist/channels/qq/crypto.py @@ -0,0 +1,49 @@ +"""AES-256-GCM utilities for QQ Bot scan-to-configure credential decryption. + +Ported from hermes-agent/gateway/platforms/qqbot/crypto.py — the q.qq.com +``create_bind_task`` / ``poll_bind_result`` flow uses AES-256-GCM to keep +the bot's *client_secret* off the wire in plaintext. +""" + +from __future__ import annotations + +import base64 +import os + + +def generate_bind_key() -> str: + """Generate a 256-bit random AES key, base64-encoded. + + The key is sent to ``create_bind_task`` so the server can encrypt + the bot's *client_secret* before returning it. Only this client + holds the key, so the secret never travels in plaintext. + """ + return base64.b64encode(os.urandom(32)).decode() + + +def decrypt_secret(encrypted_base64: str, key_base64: str) -> str: + """Decrypt a base64-encoded AES-256-GCM ciphertext. + + Ciphertext layout (after base64-decoding):: + + IV (12 bytes) ‖ ciphertext (N bytes) ‖ AuthTag (16 bytes) + + Args: + encrypted_base64: The ``bot_encrypt_secret`` value returned by + ``poll_bind_result``. + key_base64: The base64 AES key produced by :func:`generate_bind_key`. + + Returns: + The decrypted *client_secret* as a UTF-8 string. + """ + from cryptography.hazmat.primitives.ciphers.aead import AESGCM + + key = base64.b64decode(key_base64) + raw = base64.b64decode(encrypted_base64) + + iv = raw[:12] + ciphertext_with_tag = raw[12:] # AESGCM expects ciphertext + tag concatenated + + aesgcm = AESGCM(key) + plaintext = aesgcm.decrypt(iv, ciphertext_with_tag, None) + return plaintext.decode("utf-8") diff --git a/EvoScientist/channels/qq/onboard.py b/EvoScientist/channels/qq/onboard.py new file mode 100644 index 0000000..a3b7976 --- /dev/null +++ b/EvoScientist/channels/qq/onboard.py @@ -0,0 +1,300 @@ +"""QQ Bot scan-to-configure (QR code onboard) flow. + +Ported from hermes-agent/gateway/platforms/qqbot/onboard.py. + +Calls the ``q.qq.com`` ``create_bind_task`` / ``poll_bind_result`` APIs to +generate a QR code URL and poll for scan completion. On success the caller +receives the bot's *app_id*, *client_secret* (decrypted locally), and the +scanner's *user_openid* — enough to fully configure the QQ channel. + +The bot must already be registered at https://q.qq.com — scanning binds +the QQ user (developer / admin) to the existing application; it does not +create a new one. + +Reference: https://bot.q.qq.com/wiki/develop/api-v2/ +""" + +from __future__ import annotations + +import logging +import os +import platform +import sys +import time +from enum import IntEnum +from urllib.parse import quote + +from .crypto import decrypt_secret, generate_bind_key + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Endpoints / timing +# --------------------------------------------------------------------------- + +# The portal domain is configurable for corporate proxies / sandbox routing. +PORTAL_HOST = os.getenv("QQ_PORTAL_HOST", "q.qq.com") + +ONBOARD_CREATE_PATH = "/lite/create_bind_task" +ONBOARD_POLL_PATH = "/lite/poll_bind_result" +QR_URL_TEMPLATE = ( + "https://q.qq.com/qqbot/openclaw/connect.html" + "?task_id={task_id}&_wv=2&source=evoscientist" +) + +ONBOARD_API_TIMEOUT = 10.0 +ONBOARD_POLL_INTERVAL = 2.0 + +_MAX_REFRESHES = 3 + + +# --------------------------------------------------------------------------- +# Bind status +# --------------------------------------------------------------------------- + + +class BindStatus(IntEnum): + """Status codes returned by ``poll_bind_result``.""" + + NONE = 0 + PENDING = 1 + COMPLETED = 2 + EXPIRED = 3 + + +# --------------------------------------------------------------------------- +# HTTP headers +# --------------------------------------------------------------------------- + + +def _get_evoscientist_version() -> str: + try: + from importlib.metadata import version + + return version("evoscientist") + except Exception: + return "dev" + + +def _build_user_agent() -> str: + py_version = ( + f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}" + ) + os_name = platform.system().lower() + return ( + f"EvoScientistQQ/1.0.0 (Python/{py_version}; {os_name}; " + f"EvoScientist/{_get_evoscientist_version()})" + ) + + +def _api_headers() -> dict[str, str]: + """Standard HTTP headers for q.qq.com onboard API requests. + + ``q.qq.com`` requires ``Accept: application/json`` — without it, + the server returns a JavaScript anti-bot challenge page. + """ + return { + "Content-Type": "application/json", + "Accept": "application/json", + "User-Agent": _build_user_agent(), + } + + +# --------------------------------------------------------------------------- +# QR rendering +# --------------------------------------------------------------------------- + +try: + import qrcode as _qrcode_mod +except (ImportError, TypeError): + _qrcode_mod = None # type: ignore[assignment] + + +def _render_qr(url: str) -> bool: + """Render a QR code to the terminal. Returns True on success.""" + if _qrcode_mod is None: + return False + try: + qr = _qrcode_mod.QRCode( + error_correction=_qrcode_mod.constants.ERROR_CORRECT_M, + border=2, + ) + qr.add_data(url) + qr.make(fit=True) + qr.print_ascii(invert=True) + return True + except Exception: + return False + + +# --------------------------------------------------------------------------- +# HTTP helpers +# --------------------------------------------------------------------------- + + +def _create_bind_task(timeout: float = ONBOARD_API_TIMEOUT) -> tuple[str, str]: + """Create a bind task and return *(task_id, aes_key_base64)*. + + Raises: + RuntimeError: if the API returns a non-zero ``retcode``. + """ + import httpx + + url = f"https://{PORTAL_HOST}{ONBOARD_CREATE_PATH}" + key = generate_bind_key() + + with httpx.Client(timeout=timeout, follow_redirects=True) as client: + resp = client.post(url, json={"key": key}, headers=_api_headers()) + resp.raise_for_status() + data = resp.json() + + if data.get("retcode") != 0: + raise RuntimeError(data.get("msg", "create_bind_task failed")) + + task_id = data.get("data", {}).get("task_id") + if not task_id: + raise RuntimeError("create_bind_task: missing task_id in response") + + logger.debug("create_bind_task ok: task_id=%s", task_id) + return task_id, key + + +def _poll_bind_result( + task_id: str, + timeout: float = ONBOARD_API_TIMEOUT, +) -> tuple[BindStatus, str, str, str]: + """Poll the bind result for *task_id*. + + Returns: + ``(status, bot_appid, bot_encrypt_secret, user_openid)``. + + Raises: + RuntimeError: if the API returns a non-zero ``retcode``. + """ + import httpx + + url = f"https://{PORTAL_HOST}{ONBOARD_POLL_PATH}" + + with httpx.Client(timeout=timeout, follow_redirects=True) as client: + resp = client.post(url, json={"task_id": task_id}, headers=_api_headers()) + resp.raise_for_status() + data = resp.json() + + if data.get("retcode") != 0: + raise RuntimeError(data.get("msg", "poll_bind_result failed")) + + d = data.get("data", {}) + return ( + BindStatus(d.get("status", 0)), + str(d.get("bot_appid", "")), + d.get("bot_encrypt_secret", ""), + d.get("user_openid", ""), + ) + + +def build_connect_url(task_id: str) -> str: + """Build the QR-code target URL for a given *task_id*.""" + return QR_URL_TEMPLATE.format(task_id=quote(task_id)) + + +# --------------------------------------------------------------------------- +# Public entry-point +# --------------------------------------------------------------------------- + + +def qr_register(timeout_seconds: int = 600) -> dict | None: + """Run the QQ Bot scan-to-configure QR registration flow. + + Handles create → display → poll → decrypt in one call. The QR + auto-refreshes up to ``_MAX_REFRESHES`` times if the user takes + too long to scan. + + Args: + timeout_seconds: Total wall-clock budget across all refreshes. + + Returns: + ``{"app_id": ..., "client_secret": ..., "user_openid": ...}`` on + success, or ``None`` on failure / expiry / cancellation. + """ + deadline = time.monotonic() + timeout_seconds + + for refresh_count in range(_MAX_REFRESHES + 1): + # ── Create bind task ── + try: + task_id, aes_key = _create_bind_task() + except Exception as exc: + logger.warning("[QQ onboard] Failed to create bind task: %s", exc) + return None + + url = build_connect_url(task_id) + + # ── Display QR code + URL ── + print() + if _render_qr(url): + print(f" Scan the QR code above, or open this URL on your phone:\n {url}") + else: + print(f" Open this URL in QQ on your phone:\n {url}") + print(" Tip: pip install qrcode to display a scannable QR code here") + print() + + # ── Poll loop ── + consecutive_errors = 0 + while time.monotonic() < deadline: + try: + status, app_id, encrypted_secret, user_openid = _poll_bind_result( + task_id + ) + except Exception as exc: + consecutive_errors += 1 + logger.warning( + "[QQ onboard] poll_bind_result failed (%d consecutive): %s", + consecutive_errors, + exc, + ) + if consecutive_errors >= 5: + print( + "\n Repeated polling failures — aborting." + " See logs for details." + ) + return None + time.sleep(ONBOARD_POLL_INTERVAL) + continue + consecutive_errors = 0 + + if status == BindStatus.COMPLETED: + try: + client_secret = decrypt_secret(encrypted_secret, aes_key) + except Exception as exc: + logger.warning("[QQ onboard] decrypt_secret failed: %s", exc) + return None + print() + print(f" QR scan complete! (App ID: {app_id})") + if user_openid: + print(f" Scanner's OpenID: {user_openid}") + return { + "app_id": app_id, + "client_secret": client_secret, + "user_openid": user_openid, + } + + if status == BindStatus.EXPIRED: + if refresh_count >= _MAX_REFRESHES: + logger.warning( + "[QQ onboard] QR code expired %d times — giving up", + _MAX_REFRESHES, + ) + return None + print( + f"\n QR code expired, refreshing... " + f"({refresh_count + 1}/{_MAX_REFRESHES})" + ) + break # next outer iteration creates a new task + + time.sleep(ONBOARD_POLL_INTERVAL) + else: + # deadline reached without completing + logger.warning("[QQ onboard] Poll timed out after %ds", timeout_seconds) + return None + + return None diff --git a/EvoScientist/config/onboard.py b/EvoScientist/config/onboard.py index c2d7150..84528f1 100644 --- a/EvoScientist/config/onboard.py +++ b/EvoScientist/config/onboard.py @@ -2442,7 +2442,7 @@ def _step_channels(config: EvoScientistConfig) -> dict[str, object]: "qrcode>=7.4", "certifi>=2024.0", ], - "qq": ["qq-botpy>=1.0"], + "qq": ["qq-botpy>=1.0", "cryptography>=41.0", "qrcode>=7.4"], } # Channel definitions: (value, display_name, required_fields, import_check, pip_extra) @@ -2653,6 +2653,81 @@ def _step_channels(config: EvoScientistConfig) -> dict[str, object]: enabled_channels.append("imessage") continue + # QQ: offer scan-to-configure before falling back to manual entry. + # The bot must already exist at q.qq.com — scanning binds the + # developer's QQ account to it and returns app_id + client_secret. + _qq_scanned = False + if ch_name == "qq": + scan_choices = [ + Choice( + title="Scan QR code (recommended — auto-fill App ID & Secret)", + value="scan", + ), + Choice(title="Enter App ID and Secret manually", value="manual"), + ] + scan_choice = questionary.select( + "Configure QQ Bot:", + choices=scan_choices, + default="scan", + style=WIZARD_STYLE, + qmark=f" {QMARK}", + use_indicator=True, + ).ask() + if scan_choice is None: + raise KeyboardInterrupt() + + if scan_choice == "scan": + # Preflight: AES-GCM decryption needs `cryptography`. + # `qrcode` is a soft dep — onboard.py degrades to URL-only display. + try: + import cryptography # noqa: F401 + except ImportError: + console.print( + ' [yellow]✗ QR scan requires "cryptography".[/yellow]' + ) + install_now = questionary.confirm( + 'Install "cryptography" now?', + default=True, + style=WIZARD_STYLE, + qmark=f" {QMARK}", + ).ask() + if install_now is None: + raise KeyboardInterrupt() from None + if install_now and install_library("cryptography>=41.0"): + console.print(" [green]✓ Installed cryptography.[/green]") + else: + console.print( + " [yellow]⚠ Falling back to manual entry.[/yellow]" + ) + scan_choice = "manual" + + if scan_choice == "scan": + from ..channels.qq.onboard import qr_register + + console.print( + " [dim]Make sure the bot is registered at" + " https://q.qq.com first — scanning binds an" + " existing app, it does not create one.[/dim]" + ) + try: + creds = qr_register() + except Exception as exc: + console.print(f" [red]✗ Scan failed: {exc}[/red]") + creds = None + + if creds: + updates["qq_app_id"] = creds["app_id"] + updates["qq_app_secret"] = creds["client_secret"] + console.print( + f" [green]✓ Bound QQ Bot (App ID: {creds['app_id']})[/green]" + ) + _qq_scanned = True + else: + console.print( + " [yellow]⚠ Scan did not complete — falling" + " back to manual entry.[/yellow]" + ) + # WeChat: pick backend (wecom / wechatmp / personal), then prompt # backend-specific fields. Personal-WeChat has no static credentials — # we offer an interactive QR-scan that obtains and persists them. @@ -2789,17 +2864,18 @@ def _step_channels(config: EvoScientistConfig) -> dict[str, object]: updates["wechat_personal_account_id"] = account_id.strip() # Prompt for required fields - for field_name, prompt_label in required_fields: - current = getattr(config, field_name, "") - value = questionary.text( - f"{prompt_label}:", - default=current, - style=WIZARD_STYLE, - qmark=f" {QMARK}", - ).ask() - if value is None: - raise KeyboardInterrupt() - updates[field_name] = value.strip() + if not _qq_scanned: + for field_name, prompt_label in required_fields: + current = getattr(config, field_name, "") + value = questionary.text( + f"{prompt_label}:", + default=current, + style=WIZARD_STYLE, + qmark=f" {QMARK}", + ).ask() + if value is None: + raise KeyboardInterrupt() + updates[field_name] = value.strip() # Feishu: subscription mode + optional fields if ch_name == "feishu": diff --git a/pyproject.toml b/pyproject.toml index acb04bc..9ea4d13 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -67,7 +67,11 @@ discord = ["discord.py>=2.3"] slack = ["slack-sdk>=3.27", "aiohttp>=3.9"] wechat = ["pycryptodome>=3.20", "aiohttp>=3.9", "qrcode>=7.4", "certifi>=2024.0"] feishu = ["lark-oapi>=1.4.0"] -qq = ["qq-botpy>=1.0"] +qq = [ + "qq-botpy>=1.0", + "cryptography>=41.0", + "qrcode>=7.4", +] stt = ["faster-whisper>=1.0"] oauth = ["ccproxy-api>=0.2.7"] all-channels = [ @@ -80,6 +84,8 @@ all-channels = [ "certifi>=2024.0", "lark-oapi>=1.4.0", "qq-botpy>=1.0", + "cryptography>=41.0", + "qrcode>=7.4", ] [project.urls] diff --git a/uv.lock b/uv.lock index aaf9a3b..f9f342d 100644 --- a/uv.lock +++ b/uv.lock @@ -907,6 +907,7 @@ dependencies = [ all-channels = [ { name = "aiohttp" }, { name = "certifi" }, + { name = "cryptography" }, { name = "discord-py" }, { name = "lark-oapi" }, { name = "pycryptodome" }, @@ -933,7 +934,9 @@ oauth = [ { name = "ccproxy-api" }, ] qq = [ + { name = "cryptography" }, { name = "qq-botpy" }, + { name = "qrcode" }, ] slack = [ { name = "aiohttp" }, @@ -971,6 +974,8 @@ requires-dist = [ { name = "ccproxy-api", marker = "extra == 'oauth'", specifier = ">=0.2.7" }, { name = "certifi", marker = "extra == 'all-channels'", specifier = ">=2024.0" }, { name = "certifi", marker = "extra == 'wechat'", specifier = ">=2024.0" }, + { name = "cryptography", marker = "extra == 'all-channels'", specifier = ">=41.0" }, + { name = "cryptography", marker = "extra == 'qq'", specifier = ">=41.0" }, { name = "deepagents", specifier = ">=0.5.7" }, { name = "discord-py", marker = "extra == 'all-channels'", specifier = ">=2.3" }, { name = "discord-py", marker = "extra == 'discord'", specifier = ">=2.3" }, @@ -1007,6 +1012,7 @@ requires-dist = [ { name = "qq-botpy", marker = "extra == 'all-channels'", specifier = ">=1.0" }, { name = "qq-botpy", marker = "extra == 'qq'", specifier = ">=1.0" }, { name = "qrcode", marker = "extra == 'all-channels'", specifier = ">=7.4" }, + { name = "qrcode", marker = "extra == 'qq'", specifier = ">=7.4" }, { name = "qrcode", marker = "extra == 'wechat'", specifier = ">=7.4" }, { name = "questionary", specifier = ">=2.1" }, { name = "rich", specifier = ">=15.0" },