chuanqi/protocol_bot_backup/login_996pc.py

# -*- 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", "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())