# -*- 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 密钥 = 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", "nsyOfyOyoFCr%Brj-k_5PqbW"), 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())