ouyangbo/EvoScientist-Multi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(qq): add QR-code scan-to-configure onboarding for QQ Bot (#213)

Browse Source 
* feat(qq): add QR-code scan-to-configure onboarding for QQ Bot

Adds a `qr_register()` flow that drives q.qq.com's create_bind_task /
poll_bind_result APIs so the wizard can auto-fill `qq_app_id` and
`qq_app_secret` after the developer scans a QR code with a bound QQ
account, falling back to manual entry on failure or cancel.

- channels/qq/crypto.py: AES-256-GCM helpers for decrypting the bot's
  client_secret returned by poll_bind_result.
- channels/qq/onboard.py: portal API client + polling loop.
- channels/qq/__init__.py: re-export `qr_register`.
- config/onboard.py: QQ branch in `_step_channels` that offers
  "Scan QR code" vs "Enter manually", and skips the manual prompt
  loop when a scan succeeded.

* style(qq): fix ruff lint errors in onboard.py

Move `import os` to the top-level import block (E402), drop the legacy
`typing.Optional`/`typing.Tuple` imports (UP035), and use the PEP 585/604
builtin generics (`tuple[...]`, `X | None`) for the few annotations that
still referenced them (UP006/UP045). No behavior change.

* fix(qq): harden QR onboard error paths and declare scan deps

Address review feedback on PR #213:
- Declare cryptography>=41.0 and qrcode>=7.4 in [qq]/[all-channels]
  extras and in _CHANNEL_PIP_DEPS so the scan flow no longer fails
  with an opaque ImportError on a fresh `evoscientist[qq]` install.
- Polling loop logs each _poll_bind_result failure and aborts after
  5 consecutive errors instead of silently spinning until the 600s
  timeout, restoring the documented Raises: RuntimeError contract.
- Wrap decrypt_secret in try/except so failures honor the
  None-on-failure contract instead of letting exceptions escape.
- Preflight `import cryptography` in the scan branch and offer
  install or fall back to manual entry.

* style: ruff format collapse two over-wrapped log/console lines
This commit is contained in:
Ziheng Zhang
2026-05-08 19:08:52 +08:00
committed by GitHub
parent 4e04ac5b72
commit 89b0ecdbf3
6 changed files with 452 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
EvoScientist/channels/qq/__init__.py
+2 -1
View File
@@ -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:
EvoScientist/channels/qq/crypto.py
+49
View File
@@ -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")
EvoScientist/channels/qq/onboard.py
+300
View File
@@ -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
EvoScientist/config/onboard.py
+88 -12
View File
@@ -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":
pyproject.toml
+7 -1
View File
@@ -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]
uv.lock
Generated
+6
View File
@@ -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" },