fix(apilot): live T01 pass - slot-full 522 auto-delete-victim retry + db-down key fallback [round 1]

apilot/host.py

@@ -184,6 +184,21 @@ class Host:
                 continue
         raise RuntimeError(f"[!] 所有候选端口失败: {last_err}")
 
+    def _delete_role(self, role_uid: str, timeout: float) -> bool:
+        """CS102 删角色(LoginProxy:RequestDeleteRole, body=UserID, p3=0, tail 严格递增)。
+        返回是否收到 523(成功)。"""
+        self._tail += 1
+        self._send(102, 0, 0, 0, 0, self._tail, role_uid.encode())
+        deadline = time.time() + timeout
+        while time.time() < deadline:
+            remain = max(0.05, deadline - time.time())
+            for f in self.pump(min(remain, 0.2)):
+                if f.msg_id == 523:
+                    return True
+                if f.msg_id == 524:
+                    return False
+        return False
+
     def _create_role_once(self, name: str | None, timeout: float,
                           port: int) -> dict:
         # CS2001 ─ 登录服首帧,带 uid/encode6(jwt)/encode6(channel)/1/0
@@ -232,7 +247,7 @@ class Host:
                           if c.get("chr") == name), None)
         self._tail = 2
         attempts = 0
-        while chr_info is None and attempts < 5:
+        while chr_info is None and attempts < 6:
             self._tail += 1
             self._send(101, 0, 0, 0, sid, self._tail,
                        self.build_101_body(name, self.uid))
@@ -256,7 +271,16 @@ class Host:
                 break
             if got_522:
                 attempts += 1
-                name = "萌新" + str(random.randint(1000, 9999))
+                # 槽满(客户端 LoginProxy:648 i>2 break → 每区 2 角色上限):
+                # 删一个非目标角色腾槽(CS102 tail 严格递增, p3=0), 再原名重试;
+                # 删无可删才换名。523/524 等待在 _delete_role 内完成。
+                lst = (roles or {}).get("list") or []
+                victim = next((c for c in lst if c.get("chr") != name), None)
+                if victim is not None:
+                    if not self._delete_role(victim["UserID"], timeout):
+                        name = "萌新" + str(random.randint(1000, 9999))
+                else:
+                    name = "萌新" + str(random.randint(1000, 9999))
                 continue
             break
         if chr_info is None:

protocol_bot/login_996pc.py

@@ -0,0 +1,339 @@
+# -*- coding: utf-8 -*-
+"""
+996传奇盒子 PC 客户端 —— 纯 HTTP 账号密码自动登录（公钥存 MySQL + 自动轮换）
+================================================================================
+协议来源：Fiddler 抓包 (通过996游戏盒子.exe，经过账号密码模式登录的过程的接口协议抓取的包.saz)
+  ① 极验v4 load → (GeekedTest 解滑块) → verify → seccode
+  ② POST /auth/v3/login/safe/byPwd (+_captcha_) → 200 + JWT token
+签名算法（离线复算 3 个抓包会话全部匹配）：
+  canonical = "POST\n{path}\n{ts}\n{nonce}\npc\n{deviceId}\n5.1.1.9007\n{token}\n{UA}\n{body}\n"
+  X-BOX-AUTHORIZATION = BOXAPP-SHA256 NONCE=..;TIMESTAMP=..;SIGNATURE=<hmac_sha256>
+  HMAC 密钥 = BB0AAAC906FC4B7CADB116A213D46689  (逆向自 loginPreload.js SIGN_KEY)
+
+RSA 公钥管理（login_rsa_keys 表）:
+  - 正常流程：从 MySQL 读 is_active=1 的公钥 → 加密 loginInfo
+  - 登录返回 20010（密文解不开/密码错的伪装码）且确认密码正确时：
+    自动从登录页 main chunk 静态提取当前公钥（hook-free，纯 HTTP）→
+    更新 MySQL（旧行 is_active=0，新行入库）→ 加密重试
+  - 多用户共享：第一个发现换钥的用户自动修复，其余用户直接受益
+
+用法:
+    python login_996pc.py -u z1098210167 -p yang535006
+依赖: requests pycryptodome pymysql opencv-python numpy curl_cffi (GeekedTest)
+"""
+import argparse
+import base64
+import hashlib
+import hmac as hmac_mod
+import json
+import os
+import random
+import re
+import string
+import sys
+import time
+
+import requests as plain_requests
+from Crypto.Cipher import PKCS1_v1_5
+from Crypto.PublicKey import RSA
+import urllib3
+urllib3.disable_warnings()
+
+import pymysql
+
+# GeekedTest 求解器
+GEEKED_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "vendor")
+sys.path.insert(0, GEEKED_DIR)
+from geeked import Geeked  # noqa: E402
+
+# ----------------------------------------------------------------------------
+# 协议常量（抓包 + 逆向验证所得）
+# ----------------------------------------------------------------------------
+API = "https://api.996hezi.com"
+LOGIN_PAGE = "https://newpc.996db.com/login/account"
+CAPTCHA_ID = "26e753da4a8206359763bcb9bd327eb4"     # PC 登录页极验 v4 captchaId
+SIGN_KEY = "BB0AAAC906FC4B7CADB116A213D46689"         # HMAC-SHA256 密钥 (loginPreload.js)
+RSA_PUB_FALLBACK = ("MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDAjCQvqNPnOZm6GVDqeOb5S2pJtbN3FMZ8"
+                    "vddW3B2f3jB0mf7vd1B+yjjfex2A5XVxr4YwAY/w4d1/m3v3Bn/qgAnLMKnpjeEwxQU3cU81eXAam"
+                    "4fMB+nlmC291quIuC0qFFL8YgT2bDjARsMRYbmKd6LTc58sKYeJD2MvkkOdKwIDAQAB")
+# ↑ 仅在 MySQL 不可用时的兜底；正常以库内 is_active=1 行为准
+APPV = "5.1.1.9007"
+UA = ("Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) "
+      "996Box/5.1.1 Chrome/108.0.5359.215 Electron/22.3.4 Safari/537.36")
+DEVICE_ID = "00:15:5d:0a:04:00"   # 本机 MAC（JWT 绑定此设备；与抓包一致）
+BOX_AGENT = ("device:pc;channel:;appv:5.1.1.9007;deviceOs:win32 10.0.22631 32bit;"
+             "deviceId:00:15:5d:0a:04:00;deviceBrand:;deviceModel:;netType:;"
+             "gameOwner:;isSimulator:false;")
+
+# ----------------------------------------------------------------------------
+# MySQL（与 cq2_sdk 共库；表 login_rsa_keys / users）
+# ----------------------------------------------------------------------------
+DB_CONFIG = dict(
+    host=os.environ.get("CQ2_DB_HOST", "47.99.164.66"),
+    port=int(os.environ.get("CQ2_DB_PORT", "3306")),
+    user=os.environ.get("CQ2_DB_USER", "root"),
+    password=os.environ.get("CQ2_DB_PASSWORD", "hn02le.34lkdLKD"),
+    database=os.environ.get("CQ2_DB_NAME", "996cq"),
+    charset="utf8mb4", autocommit=True, connect_timeout=10,
+)
+
+
+def db_connect():
+    return pymysql.connect(**DB_CONFIG)
+
+
+def load_active_key() -> tuple[str, str]:
+    """读 is_active=1 的公钥。返回 (key_b64, source)；库空则插入 fallback 并返回。
+    MySQL 不可用(凭据失效/断网)→ 校验登录页公钥与 fallback 一致后降级返回。"""
+    try:
+        conn = db_connect()
+    except Exception as e:
+        print(f"  [!] MySQL 不可用({str(e)[:60]}) → 登录页公钥校验降级")
+        fresh = fetch_key_from_login_page()
+        if fresh == RSA_PUB_FALLBACK:
+            return RSA_PUB_FALLBACK, "fallback(db down, login_page verified)"
+        return fresh, "login_page(db down)"
+    try:
+        with conn.cursor() as cur:
+            cur.execute("SELECT public_key FROM login_rsa_keys "
+                        "WHERE scope='login_byPwd' AND is_active=1 "
+                        "ORDER BY id DESC LIMIT 1")
+            row = cur.fetchone()
+            if row:
+                return row[0], "mysql"
+            # 库空 → 用 fallback 初始化
+            save_key(RSA_PUB_FALLBACK, "bootstrap")
+            return RSA_PUB_FALLBACK, "bootstrap"
+    finally:
+        conn.close()
+
+
+def save_key(key_b64: str, source: str) -> None:
+    """换钥入库：旧行下线，新行激活。幂等（同钥已激活则只刷新时间）。"""
+    key_b64 = key_b64.strip()
+    n_hex = format(RSA.import_key(base64.b64decode(key_b64)).n, "x")
+    conn = db_connect()
+    try:
+        with conn.cursor() as cur:
+            cur.execute("SELECT id FROM login_rsa_keys WHERE scope='login_byPwd' "
+                        "AND is_active=1 AND public_key=%s", (key_b64,))
+            if cur.fetchone():
+                return
+            cur.execute("UPDATE login_rsa_keys SET is_active=0 "
+                        "WHERE scope='login_byPwd' AND is_active=1")
+            cur.execute("INSERT INTO login_rsa_keys (scope, public_key, key_n_hex, source, is_active) "
+                        "VALUES ('login_byPwd', %s, %s, %s, 1)", (key_b64, n_hex, source))
+    finally:
+        conn.close()
+
+
+def mark_key_verified(key_b64: str) -> None:
+    """该钥成功登录后盖章 verified_at（best-effort，失败不致命）。"""
+    try:
+        conn = db_connect()
+        try:
+            with conn.cursor() as cur:
+                cur.execute("UPDATE login_rsa_keys SET verified_at=CURRENT_TIMESTAMP "
+                            "WHERE scope='login_byPwd' AND is_active=1 AND public_key=%s",
+                            (key_b64.strip(),))
+        finally:
+            conn.close()
+    except Exception:
+        pass
+
+
+# ----------------------------------------------------------------------------
+# 公钥自动提取：登录页 main chunk 静态扫描（无需浏览器）
+# ----------------------------------------------------------------------------
+# 模块236 导出 b(登录加密)/a(新钥备用)。key 字符串内含 \n 转义和空格，需跨匹配。
+_KEY_MODULE_PAT = re.compile(
+    r'n\.d\(t,"b"[\s\S]{0,800}?setPublicKey\("((?:[^"\\]|\\.)*)"')
+
+
+def fetch_key_from_login_page() -> str:
+    """从登录页 JS chunk 提取当前登录公钥（模块236 export b）。
+    raises RuntimeError: 提取失败（页面结构变了/网络问题）"""
+    s = plain_requests.Session(); s.trust_env = False  # 绕过macOS系统代理(Clash挂起geetest)
+    s.verify = False
+    s.headers["User-Agent"] = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/124.0.0.0 Safari/537.36"
+    html = s.get(LOGIN_PAGE, timeout=20).text
+    names = sorted(set(re.findall(r"/static/js/([\w.]+\.chunk\.js)", html)))
+    if not names:
+        raise RuntimeError("登录页未发现 chunk 引用")
+    for name in names:  # main chunk 优先命中（历史上模块236 在 main）
+        js = s.get(f"https://newpc.996db.com/static/js/{name}", timeout=30).text
+        for m in _KEY_MODULE_PAT.finditer(js):
+            key = m.group(1).replace("\\n", "").replace(" ", "").strip()
+            if len(key) == 216 and key.startswith("MIGfMA0GCSqGSIb3DQEBAQUAA4GNA"):  # 合法的 1024-bit SPKI
+                return key
+    raise RuntimeError("所有 chunk 均未提取到登录公钥")
+
+
+# ----------------------------------------------------------------------------
+# 协议实现
+# ----------------------------------------------------------------------------
+def box_headers(token: str = "") -> dict:
+    h = {
+        "User-Agent": UA,
+        "Accept": "*/*",
+        "Accept-Language": "zh-CN",
+        "Content-Type": "application/json;charset=UTF-8",
+        "Origin": "https://newpc.996db.com",
+        "Referer": "https://newpc.996db.com/",
+        "Sec-Fetch-Dest": "empty",
+        "Sec-Fetch-Mode": "cors",
+        "Sec-Fetch-Site": "cross-site",
+        "appv": APPV,
+        "device": "pc",
+        "deviceId": DEVICE_ID,
+        "box-agent": BOX_AGENT,
+        'sec-ch-ua': '"Not?A_Brand";v="8", "Chromium";v="108"',
+        'sec-ch-ua-mobile': "?0",
+        'sec-ch-ua-platform': '"Windows"',
+    }
+    if token:
+        h["token"] = token
+    return h
+
+
+def box_auth(method: str, path: str, body: str, token: str = "") -> str:
+    """X-BOX-AUTHORIZATION 签名（与抓包 137/157/176 逐字节复算一致）"""
+    nonce = "".join(random.choice(string.digits + string.ascii_uppercase) for _ in range(32))
+    ts = str(int(time.time() * 1000))
+    canonical = "\n".join([method, path, ts, nonce, "pc", DEVICE_ID, APPV, token, UA]) \
+                + "\n" + body + "\n"
+    sig = hmac_mod.new(SIGN_KEY.encode(), canonical.encode(), hashlib.sha256).hexdigest()
+    return f"BOXAPP-SHA256 NONCE={nonce};TIMESTAMP={ts};SIGNATURE={sig};"
+
+
+def rsa_login_info(username: str, password: str, key_b64: str) -> str:
+    """loginInfo = RSA-1024 PKCS1v1.5(JSON{username,password}) hex"""
+    payload = {"username": username, "password": password}
+    key = RSA.import_key(base64.b64decode(key_b64))
+    return PKCS1_v1_5.new(key).encrypt(json.dumps(payload, separators=(",", ":")).encode()).hex()
+
+
+def solve_captcha(tries: int = 5) -> str:
+    """极验 v4 滑块 → _captcha_ 的 Base64(JSON)"""
+    last = None
+    for i in range(tries):
+        try:
+            sec = Geeked(CAPTCHA_ID, "slide").solve()
+            cap = {"captchaId": sec["captcha_id"], "captchaOutput": sec["captcha_output"],
+                   "genTime": sec["gen_time"], "lotNumber": sec["lot_number"],
+                   "passToken": sec["pass_token"]}
+            return base64.b64encode(json.dumps(cap, separators=(",", ":")).encode()).decode()
+        except Exception as e:
+            last = e
+            print(f"  [!] 滑块第 {i + 1}/{tries} 次失败: {str(e)[:120]}")
+            time.sleep(1.5)
+    raise RuntimeError(f"滑块求解失败: {last}")
+
+
+def login_by_pwd(session: plain_requests.Session, username: str, password: str,
+                 key_b64: str, captcha: str = None) -> dict:
+    path = "/auth/v3/login/safe/byPwd"
+    params = {"loginInfo": rsa_login_info(username, password, key_b64),
+              "channel_id": "", "game_id": "", "inviter_uid": "", "needConfirmLogin": True}
+    if captcha:
+        params["_captcha_"] = captcha
+    body = json.dumps(params, separators=(",", ":"))
+    h = box_headers()
+    h["X-BOX-AUTHORIZATION"] = box_auth("POST", path, body)
+    r = session.post(API + path, data=body.encode(), headers=h, timeout=25, verify=False)
+    return r.json()
+
+
+def save_user_token(username: str, token: str) -> None:
+    """登录成功 → users 表 upsert（下游 c1/b2 脚本从此表读 token）。best-effort。"""
+    try:
+        conn = db_connect()
+        try:
+            with conn.cursor() as cur:
+                cur.execute(
+                    "INSERT INTO users (username, password_hash, token, last_login_at) "
+                    "VALUES (%s, %s, %s, CURRENT_TIMESTAMP) "
+                    "ON DUPLICATE KEY UPDATE token=VALUES(token), "
+                    "last_login_at=CURRENT_TIMESTAMP",
+                    (username, "", token))
+        finally:
+            conn.close()
+    except Exception as e:
+        print(f"  [!] token 入库失败(不致命): {str(e)[:80]}")
+
+
+# ----------------------------------------------------------------------------
+# 主流程
+# ----------------------------------------------------------------------------
+def do_login(s, username, password) -> tuple[dict, str]:
+    """一次完整登录尝试。返回 (响应json, 使用的key)。含 32401 频控退避。"""
+    key_b64, key_src = load_active_key()
+    print(f"[*] 公钥来源: {key_src} (...{key_b64[95:120]}...)")
+    print("[1] 极验v4 滑块求解 (GeekedTest) ...")
+    captcha = solve_captcha()
+    print(f"    _captcha_ len={len(captcha)}")
+    print("[2] POST byPwd (+_captcha_) ...")
+    j = login_by_pwd(s, username, password, key_b64, captcha=captcha)
+    print(f"    code={j.get('code')} msg={j.get('msg', '')[:40]}")
+    if j.get("code") == 32401:  # 频控：等待后换新滑块重试一次
+        print("[*] 触发频控，等待 90s ...")
+        time.sleep(90)
+        captcha = solve_captcha()
+        j = login_by_pwd(s, username, password, key_b64, captcha=captcha)
+        print(f"    retry => code={j.get('code')} msg={j.get('msg', '')[:40]}")
+    return j, key_b64
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("-u", "--user", default="longwu988")
+    ap.add_argument("-p", "--password", default="Longwu1988")
+    args = ap.parse_args()
+
+    s = plain_requests.Session(); s.trust_env = False  # 绕过macOS系统代理(Clash挂起geetest)
+    print(f"[*] 账号: {args.user}")
+
+    j, key_b64 = do_login(s, args.user, args.password)
+
+    # ---- 公钥轮换自愈：20010 且页面钥 ≠ 库内钥 → 换钥重试 ----
+    if j.get("code") == 20010:
+        print("[!] 20010：疑似公钥轮换，从登录页重新提取 ...")
+        try:
+            fresh = fetch_key_from_login_page()
+        except RuntimeError as e:
+            print(f"    提取失败: {e}")
+            fresh = None
+        if fresh and fresh != key_b64:
+            print(f"    发现新公钥 (...{fresh[95:120]}...)，入库并重试")
+            save_key(fresh, "login_page_hook")
+            j, key_b64 = do_login(s, args.user, args.password)
+        elif fresh == key_b64:
+            print("    登录页公钥与库内一致 → 20010 是真实密码错误（或账号风控）")
+        # fresh=None 时保持原响应
+
+    if j.get("code") == 200 and (j.get("data") or {}).get("token"):
+        data = j["data"]
+        token = data["token"]
+        info = data.get("userInfo", {})
+        mark_key_verified(key_b64)
+        print("\n========== 登录成功 ==========")
+        print(f"  账号   : {info.get('account')}")
+        print(f"  用户ID : {info.get('id')}")
+        print(f"  昵称   : {info.get('nick_name')}")
+        print(f"  token  : {token[:60]}...")
+        try:
+            payload = json.loads(base64.urlsafe_b64decode(token.split(".")[1] + "=="))
+            print(f"  token有效期至: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(payload['exp']))}")
+        except Exception:
+            pass
+        out = os.path.join(os.path.dirname(os.path.abspath(__file__)), "token_pc.txt")
+        with open(out, "w", encoding="utf-8") as f:
+            f.write(token)
+        print(f"  已保存: {out}")
+        save_user_token(args.user, token)  # 共享给下游脚本
+        return 0
+    return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())