fz-mirrors/AstrBot
1
0
Fork 0
mirror of https://github.com/AstrBotDevs/AstrBot synced 2026-07-15 17:30:13 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: add qq official qr binding

Browse Source 
Add QQ Official Bot QR binding registration flow for the WebSocket adapter, wire dashboard credential autofill, and mark the WebSocket template as recommended.
This commit is contained in:
Weilong Liao
2026-06-16 19:08:41 +08:00
committed by GitHub
parent 898c800c96
commit f1854df620
9 changed files with 531 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
astrbot/core/config/default.py
View File

@@ -328,7 +328,7 @@ CONFIG_METADATA_2 = {
"description": "消息平台适配器",
"type": "list",
"config_template": {
"QQ 官方机器人(WebSocket)": {
"QQ 官方机器人(Websocket, 推荐)": {
"id": "default",
"type": "qq_official",
"enable": True,

272
astrbot/core/platform/sources/qqofficial/login_registration.py Normal file
View File

@@ -0,0 +1,272 @@
from __future__ import annotations
import base64
import secrets
from dataclasses import dataclass
from typing import Any
from urllib.parse import quote
import httpx
from Crypto.Cipher import AES
DEFAULT_QQOFFICIAL_BIND_HOST = "q.qq.com"
DEFAULT_QQOFFICIAL_QR_POLL_INTERVAL = 2
DEFAULT_QQOFFICIAL_API_TIMEOUT_MS = 10_000
QQOFFICIAL_BIND_STATUS_NONE = 0
QQOFFICIAL_BIND_STATUS_PENDING = 1
QQOFFICIAL_BIND_STATUS_COMPLETED = 2
QQOFFICIAL_BIND_STATUS_EXPIRED = 3
@dataclass
class QQOfficialLoginRegistration:
task_id: str
bind_key: str
qrcode: str
interval: int
def _string_field(data: dict[str, Any], key: str) -> str:
value = data.get(key)
if isinstance(value, str):
return value.strip()
return ""
def _int_config(value: Any, default: int, minimum: int) -> int:
try:
parsed = int(value)
except (TypeError, ValueError):
parsed = default
return max(parsed, minimum)
def _bind_host(platform_config: dict[str, Any]) -> str:
host = _string_field(platform_config, "qqofficial_bind_host")
if not host:
host = DEFAULT_QQOFFICIAL_BIND_HOST
host = host.removeprefix("https://").removeprefix("http://").rstrip("/")
return host or DEFAULT_QQOFFICIAL_BIND_HOST
def _connect_url(task_id: str, host: str) -> str:
return (
f"https://{host}/qqbot/openclaw/connect.html"
f"?task_id={quote(task_id, safe='')}&_wv=2"
)
async def _post_json(
*,
url: str,
payload: dict[str, Any],
timeout_ms: int,
) -> dict[str, Any]:
async with httpx.AsyncClient(timeout=timeout_ms / 1000) as client:
response = await client.post(
url,
json=payload,
headers={"Accept": "application/json"},
)
response.raise_for_status()
data = response.json()
if not isinstance(data, dict):
raise RuntimeError("QQ 机器人绑定接口响应格式异常")
retcode = data.get("retcode")
if retcode is not None:
try:
retcode_ok = int(retcode) == 0
except (TypeError, ValueError):
retcode_ok = False
if retcode_ok:
return data
message = (
_string_field(data, "msg")
or _string_field(data, "message")
or "QQ 机器人绑定接口返回失败"
)
raise RuntimeError(message)
return data
def generate_qqofficial_bind_key() -> str:
"""Generate a base64 AES-256 key for QQ bot binding.
Returns:
A base64-encoded 32-byte key.
"""
return base64.b64encode(secrets.token_bytes(32)).decode("ascii")
def decrypt_qqofficial_secret(encrypted_secret: str, bind_key: str) -> str:
"""Decrypt the AppSecret returned by QQ bot QR binding.
Args:
encrypted_secret: Base64 payload containing 12-byte nonce, ciphertext,
and 16-byte GCM tag.
bind_key: Base64 AES-256 key sent when creating the bind task.
Returns:
The decrypted QQ bot AppSecret.
Raises:
ValueError: If the encrypted payload is malformed or decryption fails.
"""
try:
key = base64.b64decode(bind_key)
raw = base64.b64decode(encrypted_secret)
except Exception as exc:
raise ValueError("QQ 机器人凭证解码失败") from exc
if len(key) != 32 or len(raw) <= 28:
raise ValueError("QQ 机器人凭证密文格式异常")
nonce = raw[:12]
tag = raw[-16:]
ciphertext = raw[12:-16]
cipher = AES.new(key, AES.MODE_GCM, nonce=nonce)
try:
return cipher.decrypt_and_verify(ciphertext, tag).decode("utf-8")
except Exception as exc:
raise ValueError("QQ 机器人凭证解密失败") from exc
def qqofficial_login_result(data: dict[str, Any], *, bind_key: str) -> dict[str, Any]:
"""Map QQ bot bind polling payloads to AstrBot registration statuses.
Args:
data: Response data from `/lite/poll_bind_result`.
bind_key: Base64 AES-256 key originally used for the bind task.
Returns:
A registration status payload for the dashboard polling flow.
"""
payload = data.get("data")
if not isinstance(payload, dict):
payload = {}
try:
raw_status = int(payload.get("status", QQOFFICIAL_BIND_STATUS_NONE))
except (TypeError, ValueError):
raw_status = QQOFFICIAL_BIND_STATUS_NONE
if raw_status == QQOFFICIAL_BIND_STATUS_COMPLETED:
appid = str(payload.get("bot_appid") or "").strip()
encrypted_secret = str(payload.get("bot_encrypt_secret") or "").strip()
if not appid or not encrypted_secret:
return {
"status": "error",
"qr_status": raw_status,
"message": "扫码成功但未返回完整 QQ 机器人凭证",
}
try:
secret = decrypt_qqofficial_secret(encrypted_secret, bind_key)
except ValueError as exc:
return {
"status": "error",
"qr_status": raw_status,
"message": str(exc),
}
return {
"status": "created",
"qr_status": raw_status,
"appid": appid,
"secret": secret,
"platform_id_suffix": f"_{appid}",
}
if raw_status == QQOFFICIAL_BIND_STATUS_EXPIRED:
return {
"status": "expired",
"qr_status": raw_status,
"message": "二维码已过期",
}
return {"status": "pending", "qr_status": raw_status}
async def request_qqofficial_login_qr(
platform_config: dict[str, Any],
) -> QQOfficialLoginRegistration:
"""Request a QR binding task for QQ Official Bot credentials.
Args:
platform_config: Platform configuration from the dashboard.
Returns:
QR binding registration data used by the dashboard.
"""
host = _bind_host(platform_config)
timeout_ms = _int_config(
platform_config.get("qqofficial_api_timeout_ms"),
DEFAULT_QQOFFICIAL_API_TIMEOUT_MS,
1_000,
)
interval = _int_config(
platform_config.get("qqofficial_qr_poll_interval"),
DEFAULT_QQOFFICIAL_QR_POLL_INTERVAL,
1,
)
bind_key = generate_qqofficial_bind_key()
data = await _post_json(
url=f"https://{host}/lite/create_bind_task",
payload={"key": bind_key},
timeout_ms=timeout_ms,
)
payload = data.get("data")
if not isinstance(payload, dict):
payload = {}
task_id = str(payload.get("task_id") or "").strip()
if not task_id:
raise RuntimeError("QQ 机器人绑定任务响应缺少 task_id")
return QQOfficialLoginRegistration(
task_id=task_id,
bind_key=bind_key,
qrcode=_connect_url(task_id, host),
interval=interval,
)
async def poll_qqofficial_login_once(
*,
platform_config: dict[str, Any],
task_id: str,
bind_key: str,
) -> dict[str, Any]:
"""Poll a QQ Official Bot QR binding task once.
Args:
platform_config: Platform configuration from the dashboard.
task_id: Task ID returned by `request_qqofficial_login_qr`.
bind_key: Base64 AES-256 key returned with the task.
Returns:
A registration status payload for the dashboard polling flow.
Raises:
ValueError: If `task_id` or `bind_key` is missing.
"""
if not task_id:
raise ValueError("Missing task_id")
if not bind_key:
raise ValueError("Missing bind_key")
host = _bind_host(platform_config)
timeout_ms = _int_config(
platform_config.get("qqofficial_api_timeout_ms"),
DEFAULT_QQOFFICIAL_API_TIMEOUT_MS,
1_000,
)
data = await _post_json(
url=f"https://{host}/lite/poll_bind_result",
payload={"task_id": task_id},
timeout_ms=timeout_ms,
)
return qqofficial_login_result(data, bind_key=bind_key)

45
astrbot/dashboard/services/platform_service.py
View File

@@ -15,6 +15,10 @@ from astrbot.core.platform.sources.lark.app_registration import (
request_app_registration,
)
from astrbot.core.platform.sources.lark.bot_info import request_lark_bot_info
from astrbot.core.platform.sources.qqofficial.login_registration import (
poll_qqofficial_login_once,
request_qqofficial_login_qr,
)
from astrbot.core.platform.sources.weixin_oc.login_registration import (
poll_weixin_oc_login_once,
request_weixin_oc_login_qr,
@@ -95,6 +99,12 @@ class PlatformService:
)
if platform_type == "dingtalk":
return await self._handle_dingtalk_registration(action, payload)
if platform_type == "qq_official":
return await self._handle_qqofficial_registration(
action,
payload,
platform_config,
)
raise PlatformServiceError(
f"Unsupported platform registration: {platform_type}",
@@ -185,6 +195,41 @@ class PlatformService:
raise PlatformServiceError(f"Unsupported action: {action}", 400)
async def _handle_qqofficial_registration(
self,
action: str,
payload: dict,
platform_config: dict,
) -> dict:
if action == "start":
registration = await request_qqofficial_login_qr(platform_config)
return {
"status": "pending",
"registration_code": registration.task_id,
"task_id": registration.task_id,
"bind_key": registration.bind_key,
"qrcode": registration.qrcode,
"qrcode_img_content": registration.qrcode,
"interval": registration.interval,
}
if action == "poll":
task_id = str(
payload.get("task_id") or payload.get("registration_code") or ""
).strip()
bind_key = str(payload.get("bind_key") or "").strip()
if not task_id:
raise PlatformServiceError("Missing task_id", 400)
if not bind_key:
raise PlatformServiceError("Missing bind_key", 400)
return await poll_qqofficial_login_once(
platform_config=platform_config,
task_id=task_id,
bind_key=bind_key,
)
raise PlatformServiceError(f"Unsupported action: {action}", 400)
async def _handle_weixin_oc_registration(
self,
action: str,

76
dashboard/src/components/platform/AddNewPlatform.vue
View File

@@ -142,6 +142,7 @@
<PlatformRegistrationAction
:platform-config="selectedPlatformConfig"
:active="dingtalkCreationMode === 'scan'"
@created="handlePlatformRegistrationCreated"
@success="showSuccess"
@error="showError"
/>
@@ -170,6 +171,61 @@
</div>
</div>
<div v-else-if="isQqOfficialPlatform">
<div class="creation-mode-title mt-4 mb-1">
{{ tm("registrationAction.mode.title") }}
</div>
<v-radio-group
v-model="qqOfficialCreationMode"
class="creation-mode-group"
hide-details
>
<v-radio
value="scan"
:label="tm('registrationAction.mode.scan')"
></v-radio>
<v-radio
value="manual"
:label="tm('registrationAction.mode.manual')"
></v-radio>
</v-radio-group>
<div
v-if="qqOfficialCreationMode === 'scan'"
class="registration-inline mt-3"
>
<PlatformRegistrationAction
:platform-config="selectedPlatformConfig"
:active="qqOfficialCreationMode === 'scan'"
@created="handlePlatformRegistrationCreated"
@success="showSuccess"
@error="showError"
/>
</div>
<div
v-else-if="qqOfficialCreationMode === 'manual'"
class="mt-2"
>
<div class="platform-action-row">
<v-btn
color="info"
variant="tonal"
@click="openTutorial"
class="mt-2"
>
<v-icon start>mdi-book-open-variant</v-icon>
{{ tm("dialog.viewTutorial") }}
</v-btn>
</div>
<AstrBotConfig
:iterable="selectedPlatformConfig"
:metadata="metadata['platform_group']?.metadata"
metadataKey="platform"
/>
</div>
</div>
<div
v-else-if="isWeixinOcPlatform"
class="weixin-oc-registration-inline mt-4"
@@ -777,6 +833,7 @@ export default {
selectedPlatformConfig: null,
larkCreationMode: "",
dingtalkCreationMode: "",
qqOfficialCreationMode: "",
aBConfigRadioVal: "0",
selectedAbConfId: "default",
@@ -878,6 +935,19 @@ export default {
}
}
if (this.isQqOfficialPlatform && !this.qqOfficialCreationMode) {
return false;
}
if (this.isQqOfficialPlatform && this.qqOfficialCreationMode === "scan") {
if (
!this.selectedPlatformConfig?.appid ||
!this.selectedPlatformConfig?.secret
) {
return false;
}
}
if (
this.isWeixinOcPlatform &&
!this.selectedPlatformConfig?.weixin_oc_token
@@ -974,6 +1044,9 @@ export default {
isDingtalkPlatform() {
return this.selectedPlatformConfig?.type === "dingtalk";
},
isQqOfficialPlatform() {
return this.selectedPlatformConfig?.type === "qq_official";
},
},
watch: {
selectedPlatformType(newType) {
@@ -983,10 +1056,12 @@ export default {
);
this.larkCreationMode = "";
this.dingtalkCreationMode = "";
this.qqOfficialCreationMode = "";
} else {
this.selectedPlatformConfig = null;
this.larkCreationMode = "";
this.dingtalkCreationMode = "";
this.qqOfficialCreationMode = "";
}
},
selectedAbConfId(newConfigId) {
@@ -1073,6 +1148,7 @@ export default {
this.selectedPlatformConfig = null;
this.larkCreationMode = "";
this.dingtalkCreationMode = "";
this.qqOfficialCreationMode = "";
this.aBConfigRadioVal = "0";
this.selectedAbConfId = "default";

26
dashboard/src/components/platform/PlatformRegistrationAction.vue
View File

@@ -70,6 +70,13 @@ const REGISTRATION_ACTIONS = {
scanTitleKey: 'registrationAction.dingtalk.scanTitle',
successKey: 'registrationAction.dingtalk.created',
},
qq_official: {
icon: 'mdi-qrcode',
titleKey: 'registrationAction.qqOfficial.title',
scanTitleKey: 'registrationAction.qqOfficial.scanTitle',
successKey: 'registrationAction.qqOfficial.created',
statusKeyPrefix: 'registrationAction.qqOfficial.status',
},
};
export default {
@@ -202,12 +209,19 @@ export default {
if (!this.action || !this.flow.registration_code) {
return;
}
const pollPayload = {
registration_code: this.flow.registration_code,
};
if (this.flow.task_id) {
pollPayload.task_id = this.flow.task_id;
}
if (this.flow.bind_key) {
pollPayload.bind_key = this.flow.bind_key;
}
try {
const res = await botApi.registration(
this.platformConfig.type,
this.buildPayload('poll', {
registration_code: this.flow.registration_code,
}),
this.buildPayload('poll', pollPayload),
);
if (res.data.status !== 'ok') {
throw new Error(res.data.message || this.tm('registrationAction.pollFailed'));
@@ -254,6 +268,12 @@ export default {
if (data.app_secret) {
this.platformConfig.app_secret = data.app_secret;
}
if (data.appid) {
this.platformConfig.appid = data.appid;
}
if (data.secret) {
this.platformConfig.secret = data.secret;
}
if (data.domain) {
this.platformConfig.domain = data.domain;
}

14
dashboard/src/i18n/locales/en-US/features/platform.json
View File

@@ -163,6 +163,20 @@
"scanTitle": "Scan with DingTalk on your phone",
"created": "Setup Complete"
},
"qqOfficial": {
"title": "QQ Official Bot QR Binding",
"scanTitle": "Scan with QQ on your phone",
"created": "Binding Complete",
"status": {
"idle": "Not Started",
"starting": "Getting QR Code",
"pending": "Waiting for Scan",
"created": "Binding Complete - remember to click the save button!",
"denied": "Canceled",
"expired": "Expired",
"error": "Binding Failed"
}
},
"start": "Start Setup",
"created": "Setup Complete",
"startFailed": "Failed to start QR setup",

16
dashboard/src/i18n/locales/ru-RU/features/platform.json
View File

@@ -1,4 +1,4 @@
{
{
"title": "Боты",
"subtitle": "Управление адаптерами платформ для подключения к мессенджерам",
"adapters": "Адаптеры платформ",
@@ -163,6 +163,20 @@
"scanTitle": "Отсканируйте в мобильном DingTalk",
"created": "Настройка завершена"
},
"qqOfficial": {
"title": "Привязка QQ Official Bot по QR",
"scanTitle": "Отсканируйте в мобильном QQ",
"created": "Привязка завершена",
"status": {
"idle": "Не начато",
"starting": "Получение QR",
"pending": "Ожидание сканирования",
"created": "Привязка завершена - не забудьте нажать кнопку сохранения!",
"denied": "Отменено",
"expired": "Истекло",
"error": "Ошибка привязки"
}
},
"start": "Начать настройку",
"created": "Настройка завершена",
"startFailed": "Не удалось начать QR настройку",

14
dashboard/src/i18n/locales/zh-CN/features/platform.json
View File

@@ -163,6 +163,20 @@
"scanTitle": "请使用手机钉钉扫码",
"created": "创建成功"
},
"qqOfficial": {
"title": "QQ 官方机器人扫码绑定",
"scanTitle": "请使用手机 QQ 扫码",
"created": "绑定成功",
"status": {
"idle": "未开始",
"starting": "正在获取二维码",
"pending": "等待扫码确认",
"created": "绑定成功，记得点击下方保存按钮！",
"denied": "用户取消",
"expired": "已过期",
"error": "绑定失败"
}
},
"start": "开始创建",
"created": "创建成功",
"startFailed": "发起扫码创建失败",

71
tests/test_qqofficial_login_registration.py Normal file
View File

@@ -0,0 +1,71 @@
import base64
import pytest
from Crypto.Cipher import AES
from astrbot.core.platform.sources.qqofficial.login_registration import (
QQOFFICIAL_BIND_STATUS_COMPLETED,
QQOFFICIAL_BIND_STATUS_EXPIRED,
QQOFFICIAL_BIND_STATUS_PENDING,
decrypt_qqofficial_secret,
generate_qqofficial_bind_key,
qqofficial_login_result,
)
def test_generate_qqofficial_bind_key_returns_base64_aes_key():
bind_key = generate_qqofficial_bind_key()
assert len(base64.b64decode(bind_key)) == 32
def test_qqofficial_login_result_maps_completed_payload():
bind_key = base64.b64encode(bytes(range(32))).decode("ascii")
nonce = b"123456789012"
cipher = AES.new(base64.b64decode(bind_key), AES.MODE_GCM, nonce=nonce)
ciphertext, tag = cipher.encrypt_and_digest(b"secret-value")
encrypted_secret = base64.b64encode(nonce + ciphertext + tag).decode("ascii")
result = qqofficial_login_result(
{
"data": {
"status": QQOFFICIAL_BIND_STATUS_COMPLETED,
"bot_appid": "123456789",
"bot_encrypt_secret": encrypted_secret,
},
},
bind_key=bind_key,
)
assert result == {
"status": "created",
"qr_status": QQOFFICIAL_BIND_STATUS_COMPLETED,
"appid": "123456789",
"secret": "secret-value",
"platform_id_suffix": "_123456789",
}
def test_qqofficial_login_result_maps_pending_and_expired_payloads():
bind_key = base64.b64encode(bytes(range(32))).decode("ascii")
assert qqofficial_login_result(
{"data": {"status": QQOFFICIAL_BIND_STATUS_PENDING}},
bind_key=bind_key,
) == {"status": "pending", "qr_status": QQOFFICIAL_BIND_STATUS_PENDING}
assert qqofficial_login_result(
{"data": {"status": QQOFFICIAL_BIND_STATUS_EXPIRED}},
bind_key=bind_key,
) == {
"status": "expired",
"qr_status": QQOFFICIAL_BIND_STATUS_EXPIRED,
"message": "二维码已过期",
}
def test_decrypt_qqofficial_secret_rejects_invalid_payload():
bind_key = base64.b64encode(bytes(range(32))).decode("ascii")
with pytest.raises(ValueError):
decrypt_qqofficial_secret("invalid", bind_key)