commit 755013b9dba1f0f97e5b6d20fe8bed2c8e704644 Author: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Sun Oct 4 18:44:08 2026 +0800 初始提交：QQ 群入群验证机器人 基于 qqbotsdk-py 的入群验证：进群发按钮验证码、完成前禁言、 超时踢出（11253 时兜底长期禁言）、乱点他人按钮处罚、状态持久化 与重启恢复。38 个单元/集成测试。 diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..3007946 --- /dev/null +++ b/.env.example @@ -0,0 +1,34 @@ +# ===== QQ 开放平台凭据（必填，q.qq.com 开发者后台获取）===== +APPID=你的AppID +APPSECRET=你的AppSecret + +# 接入方式：websocket（默认）或 webhook；webhook 还需 WEBHOOK_HOST/PORT/PATH +# CONNECTER=websocket + +# ===== 验证业务参数（全部可选，注释值为默认值）===== + +# 进群后允许完成验证的时间（分钟），超时触发踢出/长期禁言 +# VERIFY_TIMEOUT_MINUTES=10 + +# 待验证期间的禁言时长（小时）。到期仍未通过将由超时流程接管 +# PENDING_MUTE_HOURS=24 + +# 超时后是否尝试移出群（官方接口处于内邀阶段，未开通返回 11253） +# TIMEOUT_KICK=true + +# 踢出失败（如 11253）时的兜底禁言时长（小时） +# TIMEOUT_FALLBACK_MUTE_HOURS=72 + +# 验证码选项个数（2~5） +# CAPTCHA_OPTIONS=4 + +# 通过验证后 N 天内退群重进免验证（0 = 每次进群都要验证） +# VERIFIED_TRUST_DAYS=0 + +# 非被试者点击他人验证按钮：计数达到该值后禁言（时间窗 30 分钟内） +# FOREIGN_CLICK_LIMIT=3 +# 达到次数后的禁言时长（分钟） +# FOREIGN_CLICK_MUTE_MINUTES=5 + +# 状态文件目录（pending/verified 持久化，重启自动恢复） +# DATA_DIR=data diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ed168c0 --- /dev/null +++ b/.gitignore @@ -0,0 +1,16 @@ +# 密钥与本地配置（绝不提交） +.env + +# Python 运行产物 +__pycache__/ +*.pyc +.pytest_cache/ + +# 虚拟环境 +.venv/ + +# 运行时状态（禁言/验证进度，含成员 openid） +data/ + +# ZCode 工作目录 +.zcode/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..5b1108f --- /dev/null +++ b/README.md @@ -0,0 +1,107 @@ +# AuthBot — QQ 群入群验证机器人 + +基于 [qqbotsdk-py](https://pypi.org/project/qqbotsdk-py/)（QQ 官方开放平台 SDK）的群验证机器人： + +新成员进群 → 机器人发**数学选择题 + 按钮验证码** → 完成验证前该成员处于**禁言**状态 → +限时内点对按钮立即解禁；**超时未通过则移出群**（若踢人接口未获官方内邀权限，自动降级为 +长期禁言 + 通知管理员）。 + +## 功能一览 + +- 进群自动验证：`GROUP_MEMBER_ADD` 事件触发，数学题 4 选 1 按钮（题目随机、答错换题、倒计时不重置） +- 验证前禁言：按 `member_openid` 单独禁言（不碰全员禁言），通过后立即解禁 +- 超时踢出：默认 10 分钟；踢人失败（业务码 11253）自动兜底为禁言 72 小时 + 群内提示管理员 +- 旧题防重放：按钮 data 携带题目 token，换题后旧按钮点击无效 +- 反捣乱：非被试者点击他人的验证按钮，30 分钟窗口内累计 3 次即禁言 5 分钟 + （次数与时长可配：`FOREIGN_CLICK_LIMIT` / `FOREIGN_CLICK_MUTE_MINUTES`） +- 状态持久化：`data/` 下 JSON 存储，进程重启自动恢复禁言与倒计时（已过期立即处理） +- 轻量管理指令：群里 `@机器人 验证状态` 查看待验证名单与剩余时间； + `@机器人 验证我`（或 `测试验证`）对发起人本人走一遍完整验证流程，方便上线前自测 +- 免验信任期（可选）：通过后 N 天内退群重进不再验证，默认 0（每次都验） + +## 环境要求 + +- Python **3.12+**（SDK 硬性要求；开发环境为 3.14） +- `qqbotsdk-py >= 0.2.1`（0.2.1 修复了按成员禁言的请求体——旧版平铺写法会被服务端静默忽略） +- QQ 开放平台机器人（[q.qq.com](https://q.qq.com) 注册，拿到 AppID / AppSecret） + +## 快速开始 + +```bash +# 1. 安装依赖 +python -m venv .venv +.venv\Scripts\pip install -r requirements.txt + +# 2. 配置凭据 +copy .env.example .env +# 编辑 .env，填入 APPID / APPSECRET + +# 3. 启动（WebSocket 方式，开箱即用） +.venv\Scripts\python main.py +``` + +事件订阅开关已配好在 `intents.toml`：`GROUP_AND_C2C_EVENT`（@消息）、 +`GROUP_MEMBER_EVENT`（进退群，**默认关闭的关键开关**）、`INTERACTION`（按钮回调）。 + +## 平台侧准备（重要） + +1. **markdown 消息与按钮（键盘）权限**：在 q.qq.com 开发者后台为机器人申请。 + 没有该权限验证码发不出去——机器人此时**不会禁言新成员**（防止把人永久锁死）， + 并在日志输出 CRITICAL/ERROR 提示排查方向。 +2. **机器人入群并设为群管理员**：禁言（`restrict_chat_setting`）、移出成员都需要管理员身份。 +3. **踢人接口处于官方"内邀接入中"**：未开通的机器人调用返回业务码 11253。 + 机器人会自动降级：延长禁言（默认 72 小时，可配）并在群里提示管理员手动处理。 +4. 群消息接收方式保持默认（@机器人）即可，无需"接收所有消息"权限。 + +## 配置项（.env） + +| 变量 | 默认 | 说明 | +| --- | --- | --- | +| `APPID` / `APPSECRET` | 必填 | 开放平台凭据 | +| `CONNECTER` | `websocket` | `websocket` 或 `webhook` | +| `VERIFY_TIMEOUT_MINUTES` | `10` | 验证限时，超时触发处理 | +| `PENDING_MUTE_HOURS` | `24` | 待验证期间禁言时长 | +| `TIMEOUT_KICK` | `true` | 超时是否尝试移出群 | +| `TIMEOUT_FALLBACK_MUTE_HOURS` | `72` | 踢出失败后的兜底禁言时长 | +| `CAPTCHA_OPTIONS` | `4` | 选项个数（2~5） | +| `VERIFIED_TRUST_DAYS` | `0` | 通过后免验天数（0=每次都验） | +| `FOREIGN_CLICK_LIMIT` | `3` | 他人乱点验证按钮的处罚阈值（30 分钟窗口） | +| `FOREIGN_CLICK_MUTE_MINUTES` | `5` | 乱点处罚的禁言时长（分钟） | +| `DATA_DIR` | `data` | 状态文件目录 | + +## 目录结构 + +``` +main.py 入口 +authbot/ + config.py .env 业务参数读取与校验 + captcha.py 验证码生成、按钮编解码、消息模板 + state.py pending/verified 持久化 + 超时定时器 + handlers.py 事件处理器（进群/按钮/退群/@指令/超时流程） +intents.toml 事件订阅开关 +data/ 运行时状态（自动创建） +tests/ pytest 单元测试 +``` + +## 已知限制（官方 API 能力边界） + +- 群场景只有 **openid**（`group_openid` / `member_openid`），拿不到真实 QQ 号， + 所以提示信息里只能显示 openid 尾号。 +- **群主/管理员/机器人无法被禁言**（官方限制，接口会静默返回成功但不生效）； + 机器人会在回查禁言列表后发现这一点，`验证我` 测试时会明确提示。因此请用 + 普通成员账号测试禁言效果。 +- 禁言接口请求体为嵌套结构 `{"members": [{"op": "add|update|del", ...}]}` + （平铺写法会被服务端静默忽略，实测踩坑；SDK 0.2.1 起已内置正确封装）； + 写入后机器人会回查确认生效。 +- 被动回复有窗口限制：群内同一凭据 5 分钟内最多 5 条；主动消息默认未开通， + 超时通知发送失败时仅记录日志（禁言/踢人不受影响）。 +- 兜底禁言同样需要机器人是群管理员。 + +## 测试 + +```bash +.venv\Scripts\python -m pytest tests/ -v +``` + +覆盖验证码生成/编解码、状态持久化、定时器触发与取消、重启恢复、配置校验、 +处理器装配（不发起任何网络请求）。 diff --git a/authbot/__init__.py b/authbot/__init__.py new file mode 100644 index 0000000..e2e2f4c --- /dev/null +++ b/authbot/__init__.py @@ -0,0 +1,7 @@ +"""AuthBot：QQ 群入群验证机器人业务包。 + +进群 → 按钮验证码 → 通过前禁言 → 超时踢出（失败则长期禁言兜底）。 +事件接入与鉴权由 qqbotsdk 提供，本包只含验证业务逻辑。 +""" + +__version__ = "0.1.0" diff --git a/authbot/captcha.py b/authbot/captcha.py new file mode 100644 index 0000000..028bd82 --- /dev/null +++ b/authbot/captcha.py @@ -0,0 +1,102 @@ +"""数学选择题验证码：题目生成、选项按钮组装与按钮数据编解码。 + +按钮 data 格式为 ``v||<选项值>``：token 标识"当前这道题"， +答错换题后旧消息上的按钮因 token 不匹配而失效，防止回放旧题。 +""" + +from __future__ import annotations + +import random +from dataclasses import dataclass +from uuid import uuid4 + +from qqbotsdk.api import button, keyboard + +_BUTTON_PREFIX = "v" + + +@dataclass(slots=True) +class Captcha: + """一道验证码；a/op/b 保留运算数便于测试与排查。""" + + question: str + answer: int + options: list[int] + token: str + a: int + op: str + b: int + + def body(self) -> str: + """题面正文（不含欢迎/提示头），换题重发时复用。""" + return f"### {self.a} {self.op} {self.b} = ?" + + def keyboard(self) -> dict: + """选项按钮键盘，每行 2 个。""" + rows = [ + [button(str(v), encode_button_data(self.token, v)) for v in self.options[i : i + 2]] + for i in range(0, len(self.options), 2) + ] + return keyboard(*rows) + + +def encode_button_data(token: str, value: int) -> str: + return f"{_BUTTON_PREFIX}|{token}|{value}" + + +def parse_button_data(data: str) -> tuple[str, int] | None: + """解析按钮 data 为 (token, 选项值)；非本机器人按钮返回 None。""" + parts = data.split("|") + if len(parts) != 3 or parts[0] != _BUTTON_PREFIX: + return None + try: + return parts[1], int(parts[2]) + except ValueError: + return None + + +def welcome_markdown(cap: Captcha, timeout_minutes: int, subject: str = "新同学") -> str: + """进群首题/测试指令消息（markdown + 按钮一起发）；subject 为称呼前缀。""" + return ( + "**入群验证**\n" + f"{subject}请在 **{timeout_minutes} 分钟内** 点击下方正确答案完成验证，" + "超时未通过将被移出本群。\n\n" + cap.body() + ) + + +def retry_markdown(cap: Captcha) -> str: + """答错后换题重发的消息。""" + return "答错了，换一题再试试：\n\n" + cap.body() + + +def generate(option_count: int = 4) -> Captcha: + """随机生成一道加减/乘法选择题；干扰项取答案邻域且互不重复。""" + op = random.choice(("+", "-", "×")) + match op: + case "+": + a, b = random.randint(1, 20), random.randint(1, 20) + case "-": + a = random.randint(5, 30) + b = random.randint(1, a) # 保证差非负 + case _: + a, b = random.randint(2, 9), random.randint(2, 9) + answer = a + b if op == "+" else a - b if op == "-" else a * b + + lo = max(0, answer - 12) + distractors: set[int] = set() + while len(distractors) < option_count - 1: + cand = random.randint(lo, answer + 12) + if cand != answer: + distractors.add(cand) + options = [answer, *distractors] + random.shuffle(options) + + return Captcha( + question=f"{a} {op} {b}", + answer=answer, + options=options, + token=uuid4().hex[:12], + a=a, + op=op, + b=b, + ) diff --git a/authbot/config.py b/authbot/config.py new file mode 100644 index 0000000..dce6220 --- /dev/null +++ b/authbot/config.py @@ -0,0 +1,70 @@ +"""验证业务配置：SDK 的 .env（AppID/AppSecret 等）之上再读验证参数。 + +全部项都有内置默认值，.env 里按需覆盖；取值越界或非法启动即报错。 +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path + +from dotenv import load_dotenv + +# 幂等：qqbotsdk 导入时也会 load_dotenv()，且不会覆盖已存在的环境变量 +load_dotenv() + + +def _int_env(name: str, default: int, lo: int, hi: int) -> int: + raw = os.getenv(name, "").strip() + if not raw: + return default + try: + value = int(raw) + except ValueError: + raise ValueError(f"{name} 应为整数，实际: {raw!r}") from None + if not lo <= value <= hi: + raise ValueError(f"{name} 应在 {lo}~{hi} 之间，实际: {value}") + return value + + +def _bool_env(name: str, default: bool) -> bool: + raw = os.getenv(name, "").strip().lower() + if not raw: + return default + if raw in ("1", "true", "yes", "on"): + return True + if raw in ("0", "false", "no", "off"): + return False + raise ValueError(f"{name} 应为 true/false，实际: {raw!r}") + + +@dataclass(frozen=True, slots=True) +class AuthConfig: + """验证业务参数，含义与默认值见 .env.example。""" + + verify_timeout_minutes: int + pending_mute_hours: int + timeout_kick: bool + timeout_fallback_mute_hours: int + captcha_options: int + verified_trust_days: int + foreign_click_limit: int + foreign_click_mute_minutes: int + data_dir: Path + + @classmethod + def load(cls) -> "AuthConfig": + return cls( + verify_timeout_minutes=_int_env("VERIFY_TIMEOUT_MINUTES", 10, 1, 1440), + pending_mute_hours=_int_env("PENDING_MUTE_HOURS", 24, 1, 24 * 30), + timeout_kick=_bool_env("TIMEOUT_KICK", True), + timeout_fallback_mute_hours=_int_env( + "TIMEOUT_FALLBACK_MUTE_HOURS", 72, 1, 24 * 30 + ), + captcha_options=_int_env("CAPTCHA_OPTIONS", 4, 2, 5), + verified_trust_days=_int_env("VERIFIED_TRUST_DAYS", 0, 0, 365), + foreign_click_limit=_int_env("FOREIGN_CLICK_LIMIT", 3, 1, 100), + foreign_click_mute_minutes=_int_env("FOREIGN_CLICK_MUTE_MINUTES", 5, 1, 1440), + data_dir=Path(os.getenv("DATA_DIR", "data")), + ) diff --git a/authbot/handlers.py b/authbot/handlers.py new file mode 100644 index 0000000..b6da693 --- /dev/null +++ b/authbot/handlers.py @@ -0,0 +1,451 @@ +"""事件处理器：进群验证主流程。 + +流程（与 README 一致）： +- GROUP_MEMBER_ADD → 发按钮验证码 → 发送成功后禁言 → 挂超时定时器； + 发送失败（通常是 markdown/按钮权限未开通）则不禁言，避免把成员 + 永久锁死，同时记日志提示排查方向； +- INTERACTION_CREATE → 回应互动 → 校验点击者/题目 → 答对解禁+记录， + 答错换题（倒计时不重置），非本人或旧题点击礼貌忽略； +- 超时 → 尝试移出群（官方"内邀"能力，未开通时返回 11253）→ 失败则 + 延长禁言兜底并通知管理员； +- GROUP_MEMBER_REMOVE → 清理状态；@机器人 可查"验证状态"。 + +所有 handler 均含网络调用，注册为 background 避免 429 重试阻塞事件流。 +""" + +from __future__ import annotations + +import time +from datetime import datetime, timedelta + +from loguru import logger + +from qqbotsdk import Bot +from qqbotsdk.api import ApiError, BotApi +from qqbotsdk.api.v2 import MUTE_OP_ADD, MUTE_OP_DEL, MUTE_OP_UPDATE +from qqbotsdk.events import Event +from qqbotsdk.payloads import GroupAtMessage, GroupMemberChange, Interaction, Ready + +from . import captcha +from .config import AuthConfig +from .state import ( + ForeignClickTracker, + PendingRecord, + RecentSet, + VerifyState, + pending_key, +) + +# 官方业务码：移出群成员能力处于内邀阶段，未开通的机器人返回 11253 +KICK_NOT_ALLOWED = 11253 + + +def _rfc3339(dt: datetime) -> str: + return dt.astimezone().isoformat(timespec="seconds") + + +async def set_member_mute( + api: BotApi, group_openid: str, member_openid: str, until: datetime | None +) -> bool: + """按成员设置禁言；until=None 解除禁言。返回禁言是否确认生效。 + + 网络层走 SDK 的 mute_group_members（qqbotsdk-py 0.2.1 修复版：官方 + 请求体为嵌套 {"members": [{"op": ...}]}，平铺写法会被服务端静默 + 忽略——旧版 SDK 即踩此坑）。服务端两类"假成功"仍需自防： + - HTTP 200 但响应体带业务错误码； + - 禁言群主/管理员/机器人时静默忽略（官方限制）。 + 因此未来时间的禁言必须回查禁言列表，确认成员真的在列。 + """ + expire = "" if until is None else _rfc3339(until) + + async def _post(op: str) -> None: + result = await api.mute_group_members( + group_openid, + [{"op": op, "member_openid": member_openid, "mute_expire_at": expire}], + ) + code = result.get("code") if isinstance(result, dict) else None + if isinstance(result, dict) and code not in (None, 0): + raise ApiError(200, code, result.get("message")) + + if until is None: + await _post(MUTE_OP_DEL) + return True # 解禁请求被接受即可（回查"不在列表"无法区分从未禁言） + + try: + await _post(MUTE_OP_ADD) + except ApiError: + await _post(MUTE_OP_UPDATE) # 已在禁言中则改为更新到期时间，仍失败向上抛 + + state = await api.get_group_mute_state(group_openid) + members = state.get("members") if isinstance(state, dict) else None + if not isinstance(members, list): + return False + for entry in members: + if not isinstance(entry, dict) or entry.get("member_openid") != member_openid: + continue + try: + confirmed = datetime.fromisoformat(entry.get("mute_expire_at")) >= until - timedelta(minutes=2) + except (TypeError, ValueError): + confirmed = True # 时间解析不了就只认"人在列表里" + if confirmed: + return True + logger.warning( + f"禁言未生效 (…{member_openid[-6:]})：服务端返回成功但成员不在禁言列表，" + "通常是目标为群主/管理员/机器人（QQ 不允许对其禁言）" + ) + return False + + +async def _reply_text(api: BotApi, group_openid: str, text: str, *, event_id: str | None = None, msg_id: str | None = None) -> None: + """被动回复文本；窗口过期等失败降级为日志，不影响主流程。""" + try: + await api.post_group_message(group_openid, content=text, event_id=event_id, msg_id=msg_id) + except ApiError as e: + logger.warning(f"回复消息失败 (group={group_openid}): {e}") + + +async def _reply_captcha(api: BotApi, group_openid: str, md: str, kb: dict, event_id: str) -> None: + await api.post_group_message(group_openid, markdown=md, keyboard=kb, event_id=event_id) + + +async def _notify(api: BotApi, group_openid: str, text: str) -> None: + """主动通知（无被动凭据）。主动消息默认未开通，失败只记日志。""" + try: + await api.post_group_message(group_openid, content=text) + except ApiError as e: + logger.warning(f"主动通知失败（可能未开通主动消息）: {e}") + + +async def _start_verification( + api: BotApi, + cfg: AuthConfig, + state: VerifyState, + group_openid: str, + member_openid: str, + *, + reply_event_id: str | None = None, + reply_msg_id: str | None = None, + keep_deadline: float | None = None, + subject: str = "新同学", +) -> str: + """对一名成员发起验证：发题 → 发送成功后禁言 → 登记 pending。 + + 进群事件与"验证我"测试指令共用。keep_deadline 用于重新出题时保留 + 原截止时间（限时自进群/发起起算）。返回状态： + - "send_failed"：验证码发送失败（此时绝不禁言，防止成员被永久锁死）； + - "mute_ineffective"：题已发出，但禁言未生效（群主/管理员不可禁言）； + - "ok"。 + """ + cap = captcha.generate(cfg.captcha_options) + try: + await api.post_group_message( + group_openid, + markdown=captcha.welcome_markdown(cap, cfg.verify_timeout_minutes, subject), + keyboard=cap.keyboard(), + msg_id=reply_msg_id, + event_id=reply_event_id, + ) + except ApiError as e: + logger.error( + f"验证码发送失败，本次不禁言 (group={group_openid}, …{member_openid[-6:]}): {e}\n" + ">>> 请到 q.qq.com 开发者后台为机器人申请 markdown 消息与按钮（键盘）权限，" + "并确认机器人已添加到群、intents.toml 配置正确。" + ) + return "send_failed" + + mute_ok = True + try: + mute_ok = await set_member_mute( + api, group_openid, member_openid, + datetime.now() + timedelta(hours=cfg.pending_mute_hours), + ) + except ApiError as e: + logger.warning(f"禁言失败（请将机器人设为群管理员）(…{member_openid[-6:]}): {e}") + + state.put(PendingRecord( + group_openid=group_openid, + member_openid=member_openid, + deadline=keep_deadline if keep_deadline is not None + else time.time() + cfg.verify_timeout_minutes * 60, + token=cap.token, + answer=cap.answer, + created_at=time.time(), + )) + logger.info(f"成员 …{member_openid[-6:]} 开始验证（{cfg.verify_timeout_minutes} 分钟限时）") + return "ok" if mute_ok else "mute_ineffective" + + +async def _handle_foreign_click( + api: BotApi, + cfg: AuthConfig, + tracker: ForeignClickTracker, + group_openid: str, + member_openid: str, + *, + event_id: str | None, +) -> None: + """非被试者点击他人的验证按钮：计数警告，达到上限禁言一段时间。""" + key = pending_key(group_openid, member_openid) + count = tracker.register(key) + if count < cfg.foreign_click_limit: + remaining = cfg.foreign_click_limit - count + await _reply_text( + api, group_openid, + f"这道题不是给你的～（再乱点 {remaining} 次将被禁言 " + f"{cfg.foreign_click_mute_minutes} 分钟）", + event_id=event_id, + ) + return + + tracker.reset(key) # 处罚后清零，下一轮重新计 + muted = False + try: + muted = await set_member_mute( + api, group_openid, member_openid, + datetime.now() + timedelta(minutes=cfg.foreign_click_mute_minutes), + ) + except ApiError as e: + logger.warning(f"处罚禁言失败（需群管理员身份）: {e}") + if muted: + logger.info( + f"成员 …{member_openid[-6:]} 多次点击他人验证按钮，" + f"禁言 {cfg.foreign_click_mute_minutes} 分钟" + ) + await _reply_text( + api, group_openid, + f"多次乱点他人的验证按钮，已禁言 {cfg.foreign_click_mute_minutes} 分钟。", + event_id=event_id, + ) + else: + await _reply_text( + api, group_openid, + "多次乱点他人的验证按钮！禁言未能生效（群主/管理员无法被禁言），" + "请管理员处理。", + event_id=event_id, + ) + + +async def _handle_timeout(bot: Bot, cfg: AuthConfig, state: VerifyState, record: PendingRecord) -> None: + """超时处理：优先踢出，未开通（11253 等失败）则延长禁言兜底。""" + api = bot.api + state.pop(record.key) # 先摘除，避免重复触发 + tag = record.member_openid[-6:] + + if cfg.timeout_kick: + try: + result = await api.remove_group_members(record.group_openid, [record.member_openid]) + # QQ 失败响应可能仍是 HTTP 200，业务码只能从响应体读 + code = result.get("code") if isinstance(result, dict) else None + if isinstance(result, dict) and code not in (None, 0): + raise ApiError(200, code, result.get("message")) + except ApiError as e: + logger.warning( + f"移出群成员失败 (…{tag}, code={e.code}): {e}；转入禁言兜底" + ) + else: + logger.info(f"成员 …{tag} 超时未验证，已移出群") + await _notify(api, record.group_openid, + f"有成员（…{tag}）超时未完成验证，已移出本群。") + return + + try: + if not await set_member_mute( + api, record.group_openid, record.member_openid, + datetime.now() + timedelta(hours=cfg.timeout_fallback_mute_hours), + ): + logger.warning("兜底禁言未生效（目标可能是群主/管理员）") + else: + logger.info(f"成员 …{tag} 超时未验证，已延长禁言 {cfg.timeout_fallback_mute_hours} 小时") + except ApiError as e: + logger.error(f"兜底禁言失败（请检查机器人是否为群管理员）: {e}") + await _notify( + api, record.group_openid, + f"成员（…{tag}）超时未完成验证。移出群能力未生效（需群管理员身份，" + f"踢人接口可能仍处官方内邀阶段），已禁言 {cfg.timeout_fallback_mute_hours} 小时，" + "请管理员酌情手动处理。", + ) + + +def setup(bot: Bot, cfg: AuthConfig) -> VerifyState: + """注册全部事件处理器，返回验证状态实例。""" + + async def on_timeout(record: PendingRecord) -> None: + await _handle_timeout(bot, cfg, state, record) + + state = VerifyState(cfg.data_dir, on_timeout=on_timeout) + seen_joins = RecentSet() + seen_at_msgs = RecentSet() + foreign_clicks = ForeignClickTracker(limit=cfg.foreign_click_limit) + + # ---------- 启动恢复 ---------- + + @bot.on("READY") + async def on_ready(_: Ready, api: BotApi) -> None: + pending = state.all_pending() + logger.info(f"网关就绪，恢复 {len(pending)} 条待验证记录") + for record in pending: + # 停机期间禁言可能已到期/被解除，防御性重新禁言；失败不阻断 + try: + await set_member_mute( + api, record.group_openid, record.member_openid, + datetime.now() + timedelta(hours=cfg.pending_mute_hours), + ) + except ApiError as e: + logger.warning(f"恢复禁言失败 (…{record.member_openid[-6:]}): {e}") + await state.restore() + + # ---------- 进群：发题 + 禁言 ---------- + + @bot.on("GROUP_MEMBER_ADD", background=True) + async def on_member_add(d: GroupMemberChange, event: Event, api: BotApi) -> None: + if not d.group_openid or not d.member_openid: + return + if not seen_joins.add((d.group_openid, d.member_openid, d.timestamp)): + return + key = pending_key(d.group_openid, d.member_openid) + if state.is_verified_recently(key, cfg.verified_trust_days): + logger.info(f"成员 …{d.member_openid[-6:]} 在信任期内，跳过验证") + return + + status = await _start_verification( + api, cfg, state, d.group_openid, d.member_openid, + reply_event_id=event.event_id or None, + ) + if status == "mute_ineffective": + logger.warning("新成员禁言未生效（目标为群主/管理员？），验证流程继续") + # ---------- 点按钮：判卷 ---------- + + @bot.on("INTERACTION_CREATE", background=True) + async def on_interaction(inter: Interaction, event: Event, api: BotApi) -> None: + # 互动必须先回应，否则点击端一直转圈；重推导致的失败忽略 + try: + await api.respond_interaction(inter.id or "") + except ApiError as e: + logger.debug(f"回应互动失败（可能为重复推送）: {e}") + + if inter.chat_type != 1 or not inter.group_openid or not inter.group_member_openid: + return + parsed = captcha.parse_button_data(inter.button_data) + if parsed is None: + return + token, value = parsed + key = pending_key(inter.group_openid, inter.group_member_openid) + + record = state.get(key) + owner = state.get_by_token(token) + is_subject = record is not None and token == record.token + + # 点的是别人的当前题：无论点击者自己什么身份，计入乱点处罚 + if owner is not None and not is_subject: + await _handle_foreign_click( + api, cfg, foreign_clicks, inter.group_openid, inter.group_member_openid, + event_id=event.event_id, + ) + return + + if record is None: + if state.is_verified_recently(key, cfg.verified_trust_days): + await _reply_text(api, inter.group_openid, "你已经通过验证啦，不用再答题～", + event_id=event.event_id) + else: + # 旧题/已失效按钮，礼貌忽略不计处罚 + await _reply_text(api, inter.group_openid, "这道题不是给你的～", + event_id=event.event_id) + return + if not is_subject: + await _reply_text(api, inter.group_openid, "这题已经被换掉啦，请回答最新一题", + event_id=event.event_id) + return + + if value == record.answer: + try: + await set_member_mute(api, inter.group_openid, inter.group_member_openid, None) + except ApiError as e: + logger.warning(f"解除禁言失败（已通过验证，请手动解禁）: {e}") + state.pop(key) + state.mark_verified(key) + await _reply_text(api, inter.group_openid, "验证通过，已解除禁言，欢迎加入本群！", + event_id=event.event_id) + logger.info(f"成员 …{inter.group_member_openid[-6:]} 验证通过") + else: + cap = captcha.generate(cfg.captcha_options) + # 换题但保留原截止时间：限时自进群起算 + state.put(PendingRecord( + group_openid=record.group_openid, + member_openid=record.member_openid, + deadline=record.deadline, + token=cap.token, + answer=cap.answer, + created_at=record.created_at, + )) + try: + await _reply_captcha(api, inter.group_openid, captcha.retry_markdown(cap), + cap.keyboard(), event_id=event.event_id or "") + except ApiError as e: + logger.warning(f"换题发送失败 (…{inter.group_member_openid[-6:]}): {e}") + + # ---------- 退群：清理 ---------- + + @bot.on("GROUP_MEMBER_REMOVE", background=True) + async def on_member_remove(d: GroupMemberChange) -> None: + if d.group_openid and d.member_openid: + state.pop(pending_key(d.group_openid, d.member_openid)) + + # ---------- 群消息指令：验证状态 / 验证我 / 帮助 ---------- + + # 平台"群消息接收方式"两种模式对应两个事件名，都注册： + # 仅@消息（GROUP_AT_MESSAGE_CREATE）与全部消息（GROUP_MESSAGE_CREATE）。 + # 后者不带 @ 也能触发指令，内容匹配已对 @ 前缀免疫。 + async def on_group_command(msg: GroupAtMessage, api: BotApi) -> None: + if not msg.group_openid or (msg.id and not seen_at_msgs.add(msg.id)): + return + text = (msg.content or "").strip() + if "验证状态" in text or "验证进度" in text: + items = state.summary() + if not items: + await _reply_text(api, msg.group_openid, "当前没有待验证的成员。", + msg_id=msg.id) + return + lines = [f"待验证 {len(items)} 人："] + for record, remain in items: + minutes = max(0, int(remain // 60)) + seconds = max(0, int(remain % 60)) + lines.append(f"- 成员 …{record.member_openid[-6:]}：剩余 {minutes} 分 {seconds} 秒") + await _reply_text(api, msg.group_openid, "\n".join(lines), msg_id=msg.id) + elif "验证我" in text or "测试验证" in text: + # 测试指令：对发起人本人走一遍完整验证流程（发题→禁言→点按钮解禁） + member = msg.user_openid + if not member: + await _reply_text(api, msg.group_openid, "无法识别你的身份，测试发起失败。", + msg_id=msg.id) + return + existing = state.get(pending_key(msg.group_openid, member)) + status = await _start_verification( + api, cfg, state, msg.group_openid, member, + reply_msg_id=msg.id, + keep_deadline=existing.deadline if existing else None, + subject="你发起了验证测试，请", + ) + if status == "send_failed": + await _reply_text(api, msg.group_openid, + "验证码发送失败，请联系管理员检查机器人权限。", + msg_id=msg.id) + elif status == "mute_ineffective": + await _reply_text( + api, msg.group_openid, + "提示：你的禁言未能生效（群主/管理员/机器人无法被禁言），" + "本次测试看不到禁言效果；普通成员进群会正常禁言。", + msg_id=msg.id, + ) + elif "帮助" in text or "help" in text.lower(): + await _reply_text( + api, msg.group_openid, + "我是入群验证机器人：新成员点击验证码按钮即可通过。\n" + "指令：验证状态｜验证我（对自己发起一次验证测试）", + msg_id=msg.id, + ) + + for event_name in ("GROUP_AT_MESSAGE_CREATE", "GROUP_MESSAGE_CREATE"): + bot.on(event_name, on_group_command, background=True) + + return state diff --git a/authbot/state.py b/authbot/state.py new file mode 100644 index 0000000..f4dfef5 --- /dev/null +++ b/authbot/state.py @@ -0,0 +1,224 @@ +"""验证状态：待验证/已验证双表 JSON 持久化与超时定时器。 + +- pending 表按 ``group_openid|member_openid`` 为键，超时定时器随记录 + 增删自动挂载/取消；进程重启后经 restore() 按剩余时间重新挂载， + 已过期的立即触发超时回调； +- verified 表记录最近通过时间，配合 VERIFIED_TRUST_DAYS 决定退群重进 + 是否免验。 +""" + +from __future__ import annotations + +import asyncio +import json +import time +from collections import OrderedDict +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any, Awaitable, Callable + +from loguru import logger + + +@dataclass(slots=True) +class PendingRecord: + """一名待验证成员的当前题目与截止时间。""" + + group_openid: str + member_openid: str + deadline: float # unix 秒；超时倒计时自进群起算，换题不重置 + token: str + answer: int + created_at: float + + @property + def key(self) -> str: + return pending_key(self.group_openid, self.member_openid) + + +def pending_key(group_openid: str, member_openid: str) -> str: + return f"{group_openid}|{member_openid}" + + +class RecentSet: + """有界去重集合：官方会重推同一 msg_id/事件，超出容量的旧记录淘汰。""" + + def __init__(self, maxsize: int = 1024) -> None: + self._maxsize = maxsize + self._items: OrderedDict[Any, None] = OrderedDict() + + def add(self, item: Any) -> bool: + """首次出现返回 True，重复出现返回 False。""" + if item in self._items: + return False + self._items[item] = None + if len(self._items) > self._maxsize: + self._items.popitem(last=False) + return True + + +class ForeignClickTracker: + """非本人点击计数：窗口内累计达到上限即应处罚，处罚后清零重新计。 + + 仅内存计数（重启清零），用于实时反捣乱；窗口外的旧点击不计入。 + """ + + def __init__(self, limit: int = 3, window: float = 1800.0) -> None: + self._limit = limit + self._window = window + self._clicks: dict[str, list[float]] = {} + + def register(self, key: str, now: float | None = None) -> int: + """记录一次点击，返回时间窗口内的累计次数（含本次）。""" + now = time.time() if now is None else now + clicks = [t for t in self._clicks.get(key, []) if now - t <= self._window] + clicks.append(now) + self._clicks[key] = clicks + return len(clicks) + + def reset(self, key: str) -> None: + self._clicks.pop(key, None) + + +class VerifyState: + def __init__( + self, + data_dir: Path, + on_timeout: Callable[[PendingRecord], Awaitable[None]], + ) -> None: + self._dir = data_dir + self._on_timeout = on_timeout + self._pending: dict[str, PendingRecord] = {} + self._verified: dict[str, float] = {} + self._timers: dict[str, asyncio.Task[None]] = {} + self._load() + + # ---------- 持久化 ---------- + + @property + def _pending_file(self) -> Path: + return self._dir / "pending.json" + + @property + def _verified_file(self) -> Path: + return self._dir / "verified.json" + + def _load(self) -> None: + self._pending = { + r["key"]: PendingRecord( + group_openid=r["group_openid"], + member_openid=r["member_openid"], + deadline=r["deadline"], + token=r["token"], + answer=r["answer"], + created_at=r["created_at"], + ) + for r in self._read_json(self._pending_file, list) + if isinstance(r, dict) and {"group_openid", "member_openid", "deadline", "token", "answer"} <= r.keys() + } + verified = self._read_json(self._verified_file, dict) + self._verified = {k: v for k, v in verified.items() if isinstance(k, str) and isinstance(v, (int, float))} + + def _read_json(self, path: Path, expect: type) -> Any: + if not path.exists(): + return expect() + try: + data = json.loads(path.read_text("utf-8")) + if not isinstance(data, expect): + raise ValueError(f"应为 {expect.__name__}") + return data + except (ValueError, OSError) as e: + bad = path.with_name(path.name + ".bad") + path.replace(bad) + logger.error(f"状态文件损坏已备份到 {bad}，从空状态开始: {e}") + return expect() + + def _save_pending(self) -> None: + self._dir.mkdir(parents=True, exist_ok=True) + data = [ + {"key": k, **asdict(r)} for k, r in self._pending.items() + ] + self._pending_file.write_text( + json.dumps(data, ensure_ascii=False, indent=2), "utf-8" + ) + + def _save_verified(self) -> None: + self._dir.mkdir(parents=True, exist_ok=True) + self._verified_file.write_text( + json.dumps(self._verified, ensure_ascii=False, indent=2), "utf-8" + ) + + # ---------- verified 表 ---------- + + def is_verified_recently(self, key: str, trust_days: int) -> bool: + if trust_days <= 0: + return False + ts = self._verified.get(key) + return ts is not None and time.time() - ts < trust_days * 86400 + + def mark_verified(self, key: str) -> None: + self._verified[key] = time.time() + self._save_verified() + + # ---------- pending 表与定时器 ---------- + + def get(self, key: str) -> PendingRecord | None: + return self._pending.get(key) + + def get_by_token(self, token: str) -> PendingRecord | None: + """按题目 token 反查所属记录，用于识别"点的是谁的题"。""" + for record in self._pending.values(): + if record.token == token: + return record + return None + + def all_pending(self) -> list[PendingRecord]: + return list(self._pending.values()) + + def summary(self) -> list[tuple[PendingRecord, float]]: + """(记录, 剩余秒) 列表，剩余为负表示已到点待处理。""" + now = time.time() + return [(r, r.deadline - now) for r in self._pending.values()] + + def put(self, record: PendingRecord) -> None: + """新增/覆盖同键记录并（重）挂定时器。""" + self._pending[record.key] = record + self._save_pending() + self.arm(record.key) + + def pop(self, key: str) -> PendingRecord | None: + """删除记录并取消定时器（验证通过/成员退出/超时已处理）。""" + record = self._pending.pop(key, None) + if (timer := self._timers.pop(key, None)) is not None: + timer.cancel() + if record is not None: + self._save_pending() + return record + + def arm(self, key: str) -> None: + """为 pending 记录挂载/重挂超时任务；已过期的延迟取 0 立即触发。""" + record = self._pending[key] + if (old := self._timers.get(key)) is not None: + old.cancel() + delay = max(0.0, record.deadline - time.time()) + self._timers[key] = asyncio.create_task(self._tick(key, delay)) + + async def _tick(self, key: str, delay: float) -> None: + try: + await asyncio.sleep(delay) + except asyncio.CancelledError: + raise + record = self._pending.get(key) + if record is None: # 已被 pop，正常路径 + return + try: + # 超时处理由回调全权负责（含 pop 与兜底禁言/踢人） + await self._on_timeout(record) + except Exception: + logger.exception(f"超时处理异常 (key={key})") + + async def restore(self) -> None: + """重启恢复：READY 后重挂全部 pending 定时器（过期的随即触发）。""" + for key in list(self._pending): + self.arm(key) + logger.info(f"已恢复 {len(self._pending)} 条待验证记录的定时器") diff --git a/intents.toml b/intents.toml new file mode 100644 index 0000000..4025124 --- /dev/null +++ b/intents.toml @@ -0,0 +1,12 @@ +# 事件订阅开关（QQ 开放平台按"分组位掩码"订阅，组内事件无法单独开关） +# 完整分组名见 SDK 的 Intent 枚举；未知分组名/非布尔值会导致启动报错 + +# 群 @机器人消息（GROUP_AT_MESSAGE_CREATE，"验证状态"指令） +GROUP_AND_C2C_EVENT = true + +# 群成员进群/退群事件（GROUP_MEMBER_ADD / GROUP_MEMBER_REMOVE） +# 注意：该分组默认关闭，是进群验证的关键开关 +GROUP_MEMBER_EVENT = true + +# 按钮点击回调（INTERACTION_CREATE） +INTERACTION = true diff --git a/main.py b/main.py new file mode 100644 index 0000000..a2ea404 --- /dev/null +++ b/main.py @@ -0,0 +1,43 @@ +"""AuthBot 入口：组装 qqbotsdk 的 Bot 与验证业务并启动。 + +首次使用请复制 .env.example 为 .env 并填入 APPID/APPSECRET， +事件订阅开关在 intents.toml。WebSocket 方式直接 `python main.py`。 +""" + +from loguru import logger + +from qqbotsdk import Bot +from qqbotsdk.events import Event +from qqbotsdk.model import Opcode + +from authbot.config import AuthConfig +from authbot.handlers import setup + + +def _log_every_event(bot: Bot) -> None: + """给分发器加一层观测：任何业务事件到达即记日志。 + + SDK 对未注册 handler 的事件是静默跳过的，线上排查"平台到底推没推" + 必须能看到每一个到达的 op=0 事件。 + """ + orig_emit = bot._emitter.emit + + async def logging_emit(payload: dict) -> None: + t = payload.get("t") + if payload.get("op") == Opcode.DISPATCH and isinstance(t, str): + logger.info(f"收到网关事件: {t}") + await orig_emit(payload) + + bot._emitter.emit = logging_emit + + +def main() -> None: + bot = Bot() # 构造时读 .env，缺 APPID/APPSECRET 会直接报错退出 + cfg = AuthConfig.load() + setup(bot, cfg) + _log_every_event(bot) + bot.run() + + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..1df274e --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +qqbotsdk-py>=0.2.1 +pytest>=8.0 diff --git a/start.bat b/start.bat new file mode 100644 index 0000000..0cf61c3 --- /dev/null +++ b/start.bat @@ -0,0 +1,29 @@ +@echo off +rem AuthBot launcher - ASCII only: cmd parses batch files with the ANSI +rem codepage, so non-ASCII text here gets mangled on Chinese Windows. +title AuthBot - QQ Group Verify Bot +cd /d "%~dp0" + +if not exist ".venv\Scripts\python.exe" ( + echo [AuthBot] venv not found. Creating it and installing deps... + python -m venv .venv + if errorlevel 1 ( + echo [AuthBot] Failed to create venv. Install Python 3.12+ and add it to PATH. + pause + exit /b 1 + ) + ".venv\Scripts\python.exe" -m pip install -r requirements.txt +) + +if not exist ".env" ( + echo [AuthBot] .env missing. Copy .env.example to .env and fill APPID / APPSECRET. + pause + exit /b 1 +) + +echo [AuthBot] starting... +".venv\Scripts\python.exe" main.py + +echo. +echo [AuthBot] exited. If this was unexpected, check the log above. +pause diff --git a/tests/test_captcha.py b/tests/test_captcha.py new file mode 100644 index 0000000..0570a26 --- /dev/null +++ b/tests/test_captcha.py @@ -0,0 +1,70 @@ +"""验证码生成与编解码测试。""" + +import pytest + +from authbot import captcha + + +def test_generate_basic_properties(): + for _ in range(50): + cap = captcha.generate(4) + assert len(cap.options) == 4 + assert len(set(cap.options)) == 4 # 选项互不重复 + assert cap.answer in cap.options + assert all(v >= 0 for v in cap.options) + # 运算式与答案自洽 + expected = cap.a + cap.b if cap.op == "+" else cap.a - cap.b if cap.op == "-" else cap.a * cap.b + assert expected == cap.answer + assert cap.op in ("+", "-", "×") + + +def test_generate_option_count_variants(): + for n in (2, 3, 5): + cap = captcha.generate(n) + assert len(cap.options) == n + assert len(set(cap.options)) == n + + +def test_token_unique_per_captcha(): + tokens = {captcha.generate().token for _ in range(100)} + assert len(tokens) == 100 + + +def test_parse_button_data_roundtrip(): + token, value = "abc123", 42 + encoded = captcha.encode_button_data(token, value) + assert captcha.parse_button_data(encoded) == (token, value) + + +@pytest.mark.parametrize( + "data", + ["", "v|", "v|abc", "v|abc|x", "hello|abc|1", "cmd|abc|1", "v|abc|1|2"], +) +def test_parse_button_data_rejects_invalid(data): + assert captcha.parse_button_data(data) is None + + +def test_keyboard_structure(): + cap = captcha.generate(4) + kb = cap.keyboard() + rows = kb["content"]["rows"] + assert len(rows) == 2 # 每行 2 个按钮 + flat = [btn for row in rows for btn in row["buttons"]] + assert len(flat) == 4 + labels = {btn["render_data"]["label"] for btn in flat} + assert labels == {str(v) for v in cap.options} + for btn in flat: + token, value = captcha.parse_button_data(btn["action"]["data"]) + assert token == cap.token + assert value in cap.options + assert btn["action"]["type"] == 1 # 回调按钮 + + +def test_markdown_templates(): + cap = captcha.generate(4) + welcome = captcha.welcome_markdown(cap, 10) + assert "10 分钟" in welcome + assert cap.question in welcome + retry = captcha.retry_markdown(cap) + assert cap.question in retry + assert "换一题" in retry diff --git a/tests/test_command.py b/tests/test_command.py new file mode 100644 index 0000000..88eec1e --- /dev/null +++ b/tests/test_command.py @@ -0,0 +1,223 @@ +"""@机器人 指令的端到端装配测试（FakeApi 注入，不发起网络请求）。""" + +import asyncio +import os + +from qqbotsdk import Bot +from qqbotsdk.api import BotApi + +from authbot.config import AuthConfig +from authbot.handlers import setup +from authbot.state import PendingRecord, pending_key + + +class FakeApi(BotApi): + """替身：只记录调用，不出网。""" + + def __init__(self): # noqa: D107 - 跳过 BaseApi.__init__（不需要其组件） + self.group_messages: list[dict] = [] + self.raw_posts: list[tuple[str, dict | None]] = [] + self.mute_effective = True # 置 False 模拟禁言被服务端静默忽略 + self._muted: dict[str, str] = {} + + async def post_group_message( + self, group_openid, content="", msg_id=None, msg_seq=1, + media=None, event_id=None, markdown=None, keyboard=None, **extra, + ): + self.group_messages.append({ + "group_openid": group_openid, + "content": content, + "msg_id": msg_id, + "event_id": event_id, + "markdown": markdown, + "keyboard": keyboard, + }) + return {"id": "fake-msg"} + + async def post(self, path, **kwargs): + self.raw_posts.append((path, kwargs.get("json"))) + return {} + + async def mute_group_members(self, group_openid, members): + self.raw_posts.append((f"/v2/groups/{group_openid}/restrict_chat_setting", + {"members": members})) + for entry in members: + if entry.get("op") == "del": + self._muted.pop(entry.get("member_openid"), None) + else: + self._muted[entry.get("member_openid")] = entry.get("mute_expire_at") + return {} + + async def get_group_mute_state(self, group_openid): + members = [ + {"member_openid": m, "mute_expire_at": exp} + for m, exp in self._muted.items() + ] if self.mute_effective else [] + return {"global_rule": {"mode": "none"}, "members": members} + + async def get(self, path, **kwargs): + return {} + + +def make_bot() -> Bot: + os.environ.setdefault("APPID", "test-appid") + os.environ.setdefault("APPSECRET", "test-secret") + return Bot(intents=0) + + +def at_message_payload(content, member="MEMBER123456", msg_id="m1", group="GROUP1"): + return { + "op": 0, + "id": msg_id, + "t": "GROUP_AT_MESSAGE_CREATE", + "d": { + "id": msg_id, + "content": content, + "group_openid": group, + "author": {"member_openid": member, "union_openid": ""}, + }, + } + + +def group_message_payload(content, member="MEMBER123456", msg_id="g1", group="GROUP1"): + """全量消息模式（GROUP_MESSAGE_CREATE），字段与 @ 消息一致。""" + payload = at_message_payload(content, member, msg_id, group) + payload["t"] = "GROUP_MESSAGE_CREATE" + return payload + + +def test_command_works_in_all_message_mode(tmp_path, monkeypatch): + """平台"接收全部消息"模式下，指令经 GROUP_MESSAGE_CREATE 到达也能触发。""" + + async def scenario(): + monkeypatch.setenv("DATA_DIR", str(tmp_path)) + bot = make_bot() + cfg = AuthConfig.load() + state = setup(bot, cfg) + fake = FakeApi() + bot.services[BotApi] = fake + + await bot._emitter.emit(group_message_payload(" @AuthBot 验证状态")) + await asyncio.sleep(0.05) + + assert len(fake.group_messages) == 1 + assert "没有待验证" in (fake.group_messages[0]["content"] or "") + assert state.all_pending() == [] + + asyncio.run(scenario()) + + +def test_verify_me_command_runs_full_flow(tmp_path, monkeypatch): + async def scenario(): + monkeypatch.setenv("DATA_DIR", str(tmp_path)) + bot = make_bot() + cfg = AuthConfig.load() + state = setup(bot, cfg) + fake = FakeApi() + bot.services[BotApi] = fake # handler 按类型标注从 services 拿到替身 + + await bot._emitter.emit(at_message_payload(" 验证我")) + await asyncio.sleep(0.05) # background handler 异步执行 + + # 发出了带按钮的验证码，且是对指令消息的被动回复（msg_id 凭据） + assert len(fake.group_messages) == 1 + sent = fake.group_messages[0] + assert sent["msg_id"] == "m1" + assert sent["markdown"] and "验证测试" in sent["markdown"] + assert sent["keyboard"] + + # 对发起人本人按成员禁言（官方嵌套结构：members[].op=add） + mute_calls = [j for p, j in fake.raw_posts if p.endswith("/restrict_chat_setting")] + assert len(mute_calls) == 1 + entry = mute_calls[0]["members"][0] + assert entry["op"] == "add" + assert entry["member_openid"] == "MEMBER123456" + assert "mute_expire_at" in entry + + # 进入 pending，按钮 data 里的 token 与记录一致 + key = pending_key("GROUP1", "MEMBER123456") + record = state.get(key) + assert record is not None + btn = sent["keyboard"]["content"]["rows"][0]["buttons"][0] + assert btn["action"]["data"].split("|")[1] == record.token + + state.pop(key) # 清理，避免定时器任务悬挂到用例外 + + asyncio.run(scenario()) + + +def test_verify_me_keeps_original_deadline(tmp_path, monkeypatch): + """已在验证中再次发起测试：换题但倒计时不重置。""" + + async def scenario(): + import time + + monkeypatch.setenv("DATA_DIR", str(tmp_path)) + bot = make_bot() + cfg = AuthConfig.load() + state = setup(bot, cfg) + bot.services[BotApi] = FakeApi() + + key = pending_key("GROUP1", "MEMBER123456") + old_deadline = time.time() + 123.0 + state.put(PendingRecord( + group_openid="GROUP1", member_openid="MEMBER123456", + deadline=old_deadline, token="old", answer=1, created_at=time.time(), + )) + + await bot._emitter.emit(at_message_payload(" 测试验证", msg_id="m2")) + await asyncio.sleep(0.05) + + record = state.get(key) + assert record is not None + assert record.deadline == old_deadline + assert record.token != "old" # 题目已更换 + state.pop(key) + + asyncio.run(scenario()) + + +def test_verify_me_reports_ineffective_mute(tmp_path, monkeypatch): + """禁言被服务端静默忽略时（群主/管理员），验证我 应明确提示原因。""" + + async def scenario(): + monkeypatch.setenv("DATA_DIR", str(tmp_path)) + bot = make_bot() + cfg = AuthConfig.load() + state = setup(bot, cfg) + fake = FakeApi() + fake.mute_effective = False + bot.services[BotApi] = fake + + await bot._emitter.emit(at_message_payload(" 验证我")) + await asyncio.sleep(0.05) + + # 验证码 + 一条"禁言未生效"提示 + assert len(fake.group_messages) == 2 + assert fake.group_messages[0]["markdown"] + assert "禁言未能生效" in (fake.group_messages[1]["content"] or "") + key = pending_key("GROUP1", "MEMBER123456") + assert state.get(key) is not None # 验证流程仍然继续 + state.pop(key) + + asyncio.run(scenario()) + + +def test_status_command_no_side_effects(tmp_path, monkeypatch): + async def scenario(): + monkeypatch.setenv("DATA_DIR", str(tmp_path)) + bot = make_bot() + cfg = AuthConfig.load() + state = setup(bot, cfg) + fake = FakeApi() + bot.services[BotApi] = fake + + await bot._emitter.emit(at_message_payload("验证状态", msg_id="m3")) + await asyncio.sleep(0.05) + + assert len(fake.group_messages) == 1 + assert "没有待验证" in (fake.group_messages[0]["content"] or "") + assert fake.raw_posts == [] # 不触发禁言等写操作 + assert state.all_pending() == [] + + asyncio.run(scenario()) diff --git a/tests/test_interaction.py b/tests/test_interaction.py new file mode 100644 index 0000000..a71c2a3 --- /dev/null +++ b/tests/test_interaction.py @@ -0,0 +1,178 @@ +"""INTERACTION_CREATE 按钮回调流程测试（FakeApi 注入，无网络）。 + +覆盖：本人答对通过、非本人点击计数与处罚、旧题点击不处罚。 +""" + +import asyncio +import os +import time + +from qqbotsdk import Bot +from qqbotsdk.api import BotApi + +from authbot import captcha +from authbot.config import AuthConfig +from authbot.handlers import setup +from authbot.state import PendingRecord, pending_key + + +class FakeApi(BotApi): + def __init__(self): # noqa: D107 - 跳过 BaseApi.__init__（不需要其组件） + self.group_messages: list[dict] = [] + self.raw_posts: list[tuple[str, dict | None]] = [] + self.interactions_responded: list[str] = [] + self.mute_effective = True + self._muted: dict[str, str] = {} + + async def post_group_message( + self, group_openid, content="", msg_id=None, msg_seq=1, + media=None, event_id=None, markdown=None, keyboard=None, **extra, + ): + self.group_messages.append({ + "group_openid": group_openid, + "content": content, + "msg_id": msg_id, + "event_id": event_id, + "markdown": markdown, + "keyboard": keyboard, + }) + return {"id": "fake-msg"} + + async def post(self, path, **kwargs): + self.raw_posts.append((path, kwargs.get("json"))) + return {} + + async def mute_group_members(self, group_openid, members): + self.raw_posts.append((f"/v2/groups/{group_openid}/restrict_chat_setting", + {"members": members})) + for entry in members: + if entry.get("op") == "del": + self._muted.pop(entry.get("member_openid"), None) + else: + self._muted[entry.get("member_openid")] = entry.get("mute_expire_at") + return {} + + async def get_group_mute_state(self, group_openid): + members = [ + {"member_openid": m, "mute_expire_at": exp} + for m, exp in self._muted.items() + ] if self.mute_effective else [] + return {"global_rule": {"mode": "none"}, "members": members} + + async def get(self, path, **kwargs): + return {} + + async def respond_interaction(self, interaction_id, code=0): + self.interactions_responded.append(interaction_id) + return {} + + +def make_bot() -> Bot: + os.environ.setdefault("APPID", "test-appid") + os.environ.setdefault("APPSECRET", "test-secret") + return Bot(intents=0) + + +def interaction_payload(token, value, member, group="GROUP1", seq=1): + return { + "op": 0, + "id": f"INTERACTION_CREATE:evt-{seq}", + "t": "INTERACTION_CREATE", + "d": { + "id": f"inner-uuid-{seq}", + "type": 11, + "chat_type": 1, + "group_openid": group, + "group_member_openid": member, + "data": {"resolved": {"button_data": captcha.encode_button_data(token, value)}}, + }, + } + + +def test_subject_correct_answer_passes(tmp_path, monkeypatch): + async def scenario(): + monkeypatch.setenv("DATA_DIR", str(tmp_path)) + bot = make_bot() + cfg = AuthConfig.load() + state = setup(bot, cfg) + fake = FakeApi() + bot.services[BotApi] = fake + + key = pending_key("GROUP1", "MEMBER_A") + state.put(PendingRecord( + group_openid="GROUP1", member_openid="MEMBER_A", + deadline=time.time() + 600, token="subj", answer=7, created_at=time.time(), + )) + + await bot._emitter.emit(interaction_payload("subj", 7, "MEMBER_A", seq=1)) + await asyncio.sleep(0.05) + + # 解禁（op=del）+ 通过播报 + mutes = [j for p, j in fake.raw_posts if p.endswith("/restrict_chat_setting")] + assert len(mutes) == 1 + entry = mutes[0]["members"][0] + assert entry["op"] == "del" + assert entry["member_openid"] == "MEMBER_A" + assert "验证通过" in (fake.group_messages[-1]["content"] or "") + assert state.get(key) is None # pending 已清除 + assert state.is_verified_recently(key, 99) + + asyncio.run(scenario()) + + +def test_foreign_clicker_muted_on_third_click(tmp_path, monkeypatch): + async def scenario(): + monkeypatch.setenv("DATA_DIR", str(tmp_path)) + bot = make_bot() + cfg = AuthConfig.load() + state = setup(bot, cfg) + fake = FakeApi() + bot.services[BotApi] = fake + + # MEMBER_A 的待验证题 + state.put(PendingRecord( + group_openid="GROUP1", member_openid="MEMBER_A", + deadline=time.time() + 600, token="subj", answer=7, created_at=time.time(), + )) + + # 旁观者 MEMBER_B 连点 3 次别人的题 + for seq in range(1, 4): + await bot._emitter.emit(interaction_payload("subj", 9, "MEMBER_B", seq=seq)) + await asyncio.sleep(0.03) + + # 每次点击都回应了互动 + assert len(fake.interactions_responded) == 3 + # 前两次只有警告，第三次才禁言 5 分钟 + warns = [m["content"] for m in fake.group_messages[:2]] + assert all("不是给你的" in (c or "") for c in warns) + mutes = [j for p, j in fake.raw_posts if p.endswith("/restrict_chat_setting")] + assert len(mutes) == 1 + entry = mutes[0]["members"][0] + assert entry["op"] == "add" + assert entry["member_openid"] == "MEMBER_B" + assert "已禁言 5 分钟" in (fake.group_messages[-1]["content"] or "") + # 被试者的记录不受影响 + assert state.get(pending_key("GROUP1", "MEMBER_A")) is not None + state.pop(pending_key("GROUP1", "MEMBER_A")) + + asyncio.run(scenario()) + + +def test_stale_button_click_not_punished(tmp_path, monkeypatch): + async def scenario(): + monkeypatch.setenv("DATA_DIR", str(tmp_path)) + bot = make_bot() + cfg = AuthConfig.load() + state = setup(bot, cfg) + fake = FakeApi() + bot.services[BotApi] = fake + + # token 不属于任何在途题目（旧题残留按钮） + for seq in range(1, 4): + await bot._emitter.emit(interaction_payload("ghost", 1, "MEMBER_C", seq=seq)) + await asyncio.sleep(0.03) + + assert fake.raw_posts == [] # 三次都不触发禁言 + assert all("不是给你的" in (m["content"] or "") for m in fake.group_messages) + + asyncio.run(scenario()) diff --git a/tests/test_state.py b/tests/test_state.py new file mode 100644 index 0000000..26cfa12 --- /dev/null +++ b/tests/test_state.py @@ -0,0 +1,229 @@ +"""验证状态持久化与超时定时器测试。""" + +import asyncio +import json +import tempfile +import time +from pathlib import Path + +from authbot.state import ( + ForeignClickTracker, + PendingRecord, + RecentSet, + VerifyState, + pending_key, +) + + +def make_record(group="G1", member="M1", deadline=None, token="tok1", answer=7): + return PendingRecord( + group_openid=group, + member_openid=member, + deadline=deadline if deadline is not None else time.time() + 600, + token=token, + answer=answer, + created_at=time.time(), + ) + + +def test_put_get_pop_roundtrip(tmp_path): + async def scenario(): + async def noop(record): + pass + + state = VerifyState(tmp_path, on_timeout=noop) + rec = make_record() + state.put(rec) + assert state.get(rec.key) is not None + + # 新实例从磁盘恢复（模拟重启） + state2 = VerifyState(tmp_path, on_timeout=noop) + restored = state2.get(rec.key) + assert restored is not None + assert restored.token == rec.token + assert restored.answer == rec.answer + assert restored.deadline == rec.deadline + + assert state2.pop(rec.key).token == rec.token + assert state2.get(rec.key) is None + assert state2.pop(rec.key) is None # 重复 pop 安全 + + asyncio.run(scenario()) + + +def test_pop_persists_deletion(tmp_path): + async def scenario(): + async def noop(record): + pass + + state = VerifyState(tmp_path, on_timeout=noop) + rec = make_record() + state.put(rec) + state.pop(rec.key) + assert VerifyState(tmp_path, on_timeout=noop).get(rec.key) is None + + asyncio.run(scenario()) + + +def test_verified_trust_window(tmp_path): + async def scenario(): + async def noop(record): + pass + + state = VerifyState(tmp_path, on_timeout=noop) + key = pending_key("G1", "M1") + state.mark_verified(key) + + # trust_days=0：即使刚通过也不免验 + assert state.is_verified_recently(key, 0) is False + assert state.is_verified_recently(key, 1) is True + + # 手动把通过时间改到 3 天前 + verified = json.loads((tmp_path / "verified.json").read_text("utf-8")) + verified[key] = time.time() - 3 * 86400 + (tmp_path / "verified.json").write_text(json.dumps(verified), "utf-8") + state2 = VerifyState(tmp_path, on_timeout=noop) + assert state2.is_verified_recently(key, 1) is False + assert state2.is_verified_recently(key, 7) is True + + asyncio.run(scenario()) + + +def test_summary_remaining_seconds(tmp_path): + async def scenario(): + async def noop(record): + pass + + state = VerifyState(tmp_path, on_timeout=noop) + rec = make_record(deadline=time.time() + 120) + state.put(rec) + [(got, remain)] = state.summary() + assert got.key == rec.key + assert 110 < remain <= 120 + + asyncio.run(scenario()) + + +def test_timeout_timer_fires_and_pop_cancels(tmp_path): + async def scenario(): + fired = [] + + async def on_timeout(record): + fired.append(record.key) + + state = VerifyState(tmp_path, on_timeout=on_timeout) + # 已到点的：立即触发 + state.put(make_record(member="EXPIRED", deadline=time.time() - 1)) + # 未到点的随后被 pop：不应触发 + state.put(make_record(member="CANCELLED", deadline=time.time() + 5)) + state.pop(pending_key("G1", "CANCELLED")) + await asyncio.sleep(0.3) + assert fired == [pending_key("G1", "EXPIRED")] + + asyncio.run(scenario()) + + +def test_restore_rearms_from_disk(tmp_path): + async def scenario(): + async def noop(record): + pass + + # 先写入一条已过期与一条未过期的记录 + VerifyState(tmp_path, on_timeout=noop).put( + make_record(member="OLD", deadline=time.time() - 1) + ) + VerifyState(tmp_path, on_timeout=noop).put( + make_record(member="NEW", deadline=time.time() + 600) + ) + + fired = [] + + async def on_timeout(record): + fired.append(record.key) + + state = VerifyState(tmp_path, on_timeout=on_timeout) + await state.restore() + await asyncio.sleep(0.3) + # 只有已过期的触发，未到期的不触发 + assert fired == [pending_key("G1", "OLD")] + assert state.get(pending_key("G1", "NEW")) is not None + + asyncio.run(scenario()) + + +def test_timeout_callback_exception_isolated(tmp_path): + """回调抛异常不炸定时器任务本身，也不影响其他记录。""" + + async def scenario(): + async def boom(record): + raise RuntimeError("boom") + + state = VerifyState(tmp_path, on_timeout=boom) + state.put(make_record(member="A", deadline=time.time() - 1)) + state.put(make_record(member="B", deadline=time.time() - 1)) + await asyncio.sleep(0.3) # 两条回调都跑完且异常被记录 + + asyncio.run(scenario()) + + +def test_corrupt_state_file_backed_up(tmp_path): + async def scenario(): + async def noop(record): + pass + + (tmp_path / "pending.json").write_text("{not json", "utf-8") + state = VerifyState(tmp_path, on_timeout=noop) + assert state.all_pending() == [] + assert (tmp_path / "pending.json.bad").exists() + + asyncio.run(scenario()) + + +def test_recent_set(): + s = RecentSet(maxsize=3) + assert s.add("a") is True + assert s.add("a") is False + assert s.add("b") is True + assert s.add("c") is True + assert s.add("d") is True # 挤掉最旧的 a + assert s.add("a") is True # a 已被淘汰，重新视为首次 + + +def test_foreign_click_tracker_counts_and_resets(): + t = ForeignClickTracker(limit=3, window=1800.0) + now = 1000.0 + assert t.register("k", now) == 1 + assert t.register("k", now + 10) == 2 + assert t.register("k", now + 20) == 3 # 到达上限 + t.reset("k") + assert t.register("k", now + 30) == 1 # 处罚后重新计数 + + +def test_foreign_click_tracker_window_prunes_old(): + t = ForeignClickTracker(limit=3, window=60.0) + now = 1000.0 + assert t.register("k", now) == 1 + assert t.register("k", now + 61) == 1 # 窗口外的第一次点击不计入 + assert t.register("k", now + 61 + 30) == 2 + + +def test_foreign_click_tracker_keys_independent(): + t = ForeignClickTracker(limit=2) + assert t.register("a") == 1 + assert t.register("b") == 1 + assert t.register("a") == 2 + + +def test_get_by_token(): + async def scenario(): + async def noop(record): + pass + + state = VerifyState(Path(tempfile.mkdtemp()), on_timeout=noop) + rec = make_record(token="tok-xyz") + state.put(rec) + assert state.get_by_token("tok-xyz").key == rec.key + assert state.get_by_token("nope") is None + state.pop(rec.key) + + asyncio.run(scenario()) diff --git a/tests/test_wiring.py b/tests/test_wiring.py new file mode 100644 index 0000000..04687e7 --- /dev/null +++ b/tests/test_wiring.py @@ -0,0 +1,67 @@ +"""处理器装配与配置装配测试（不发起任何网络请求）。""" + +from qqbotsdk import Bot + +from authbot.config import AuthConfig +from authbot.handlers import KICK_NOT_ALLOWED, setup + + +def _bot() -> Bot: + import os + + os.environ.setdefault("APPID", "test-appid") + os.environ.setdefault("APPSECRET", "test-secret") + return Bot(intents=0) # 掩码直传，避免读 intents.toml + + +def test_config_defaults(monkeypatch, tmp_path): + monkeypatch.delenv("VERIFY_TIMEOUT_MINUTES", raising=False) + monkeypatch.delenv("TIMEOUT_KICK", raising=False) + monkeypatch.delenv("DATA_DIR", raising=False) + cfg = AuthConfig.load() + assert cfg.verify_timeout_minutes == 10 + assert cfg.timeout_kick is True + assert cfg.timeout_fallback_mute_hours == 72 + assert cfg.captcha_options == 4 + assert cfg.verified_trust_days == 0 + + +def test_config_env_override(monkeypatch, tmp_path): + monkeypatch.setenv("VERIFY_TIMEOUT_MINUTES", "5") + monkeypatch.setenv("TIMEOUT_KICK", "false") + monkeypatch.setenv("DATA_DIR", str(tmp_path)) + cfg = AuthConfig.load() + assert cfg.verify_timeout_minutes == 5 + assert cfg.timeout_kick is False + assert cfg.data_dir == tmp_path + + +def test_config_rejects_out_of_range(monkeypatch): + monkeypatch.setenv("VERIFY_TIMEOUT_MINUTES", "0") + try: + AuthConfig.load() + except ValueError as e: + assert "VERIFY_TIMEOUT_MINUTES" in str(e) + else: + raise AssertionError("应当拒绝越界值") + + +def test_setup_registers_all_handlers(tmp_path, monkeypatch): + monkeypatch.setenv("DATA_DIR", str(tmp_path)) + bot = _bot() + cfg = AuthConfig.load() + state = setup(bot, cfg) + registered = { + event + for event, handlers in bot._emitter._handlers.items() # noqa: SLF001 + if handlers + } + assert { + "READY", + "GROUP_MEMBER_ADD", + "INTERACTION_CREATE", + "GROUP_MEMBER_REMOVE", + "GROUP_AT_MESSAGE_CREATE", + } <= registered + assert state.all_pending() == [] + assert KICK_NOT_ALLOWED == 11253