live-hub-py/tools/full_auto_safe.py

#!/usr/bin/env python3
"""一键全自动(含新账号safe_auth滑块): 账号密码 -> 页面cookie(biztoken)。纯协议版。

2026-08-26: nonce 本地重算(tools/nonce_forge.py), 任意账号无需设备首登;
新/异地账号触发 safe_auth 时自动解 udbrtt 滑块并重发登录, 全程零人工。

用法:
  .venv/bin/python tools/full_auto_safe.py <账号如hy_xxx或300023887> <密码>

前提: evidence/cert_keycap.json 内有信封模板(任意账号共用, 仅需采集一次)
      + evidence/nonce_k1.json 或默认 k1 设备常量。
"""
import base64
import json
import struct
import sys
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "tools"))

from app_login_flow import QrAuthRequiredError, login_cred  # noqa: E402
from cert_forge import build_p1, decrypt_cert, forge_cert, parse_p1  # noqa: E402
from envelope_forge import Envelope  # noqa: E402
from nonce_forge import K1_DEFAULT, gen_nonce  # noqa: E402


def real_uid(acct: str, pwd: str) -> int:
    """登录响应解析真实uid(响应bean偏移235 u64be)。"""
    from app_login_flow import wup_password_login_raw
    resp = wup_password_login_raw(acct, pwd)
    s = resp.find(b"\x0a\x0a", 0x40)
    e = resp.find(b"_wup_header")
    uid = struct.unpack_from(">Q", resp[s:e], 235)[0]
    assert 1_100_000_000_000 < uid < 1_300_000_000_000, f"uid异常 {uid}"
    return uid


def forge_for_uid(env: Envelope, cred: bytes, uid: int, k1: str) -> bytes:
    """本地重算nonce + 换cred 铸证(任意账号可用)。"""
    orig = base64.b64decode(env.cert_b64)
    f = parse_p1(decrypt_cert(orig))
    st = int(time.time() * 1000)
    rnd = gen_nonce(uid, k1, service_time_ms=st, counter=0)
    print(f"      本地nonce: uid={uid} st={st} rnd={rnd.hex()}")
    P1 = build_p1(f["app_id"], f["fingerprint"], cred, rnd=rnd)
    return forge_cert(P1, key_idx=orig[1])


def main() -> int:
    acct, pwd = sys.argv[1], sys.argv[2]
    print("[1/4] 纯Python密码登录 (遇safe_auth自动过滑块) ...")
    try:
        cred = login_cred(acct, pwd)
    except QrAuthRequiredError as exc:
        print(f"❌ {exc}")
        return 2
    print(f"      cred OK: {len(cred)}B {cred[:8].hex()}")

    uid = real_uid(acct, pwd)
    print(f"[2/4] 解析真实uid={uid}, 本地重算nonce 铸证 ...")
    env = Envelope.load()
    cert = forge_for_uid(env, cred, uid, K1_DEFAULT)
    b64 = base64.b64encode(cert).decode()
    if len(b64) != env.cert_len:
        print(f"      cert b64 长度不符 {len(b64)} != {env.cert_len}, 中止")
        return 1
    raw = bytearray(env.raw)
    raw[env.cert_off:env.cert_off + env.cert_len] = b64.encode()
    if env.uid != uid:
        print(f"      patch uid {env.uid} -> {uid}")
        struct.pack_into(">Q", raw, env.uid_off, uid)
    wup = base64.b64encode(bytes(raw)).decode()

    print("[3/4] 扫码绑定四步流 ...")
    from urllib.parse import quote
    sys.path.insert(0, str(ROOT / "scripts"))
    from probe_huya_qr_bind import QrRole, web_behavior
    from core.huya.device_fingerprint import get_huya_sdid
    sdid = get_huya_sdid(allow_fallback=False).sdid
    pc = QrRole(pc=True, sdid=sdid)
    ph = QrRole(pc=False, sdid=sdid)
    beh, page = web_behavior()
    resp = pc.call("/qrLgn/getQrId", "70001",
                   {"behavior": beh, "type": "", "domainList": "", "page": page})
    qrid = resp["data"]["qrId"]
    cp = f"https://aq.huya.com/r/confirm.html?k={qrid}&id=5002"
    ph.call("/qrLgn/scanQrPicNotify", "70005",
            {"qrId": qrid, "wupData": wup,
             "behavior": quote("[]", safe=""), "page": quote(cp, safe="")})
    r2 = ph.call("/qrLgn/bindQrLoginUser", "70007",
                 {"qrId": qrid, "wupData": wup,
                  "behavior": quote("[]", safe=""), "page": quote(cp, safe="")})
    print(f"      bind rc={r2.get('returnCode')} {r2.get('message') or ''}")
    biztoken = None
    for _ in range(10):
        rt = pc.call("/qrLgn/tryQrLogin", "70003",
                     {"qrId": qrid, "remember": "1", "domainList": "",
                      "behavior": beh, "page": page})
        dt = rt.get("data") or {}
        if dt.get("stage") == 2:
            biztoken = dt.get("biztoken")
            print(f"      biztoken OK: {len(biztoken)}B")
            break
        time.sleep(2)
    if not biztoken:
        print("❌ 未取到biztoken")
        return 1

    print("[4/4] cookie verify 兑换 ...")
    sess = getattr(pc, 'session', None) or getattr(pc, 's', None) or getattr(pc, '_sess', None)
    r = sess.post("https://udblgn.huya.com/web/cookie/verify",
                  json={"appId": 5002}, timeout=15)
    print(f"      verify HTTP {r.status_code}")
    jar = []
    for c in sess.cookies:
        print(f"      {c.name} = {c.value[:50]}")
        jar.append(f"{c.name}={c.value}")
    out = ROOT / "evidence/web_cookies_full.txt"
    out.write_text(f"# account: {acct}\n" + "\n".join(jar) + "\n")
    print(f"\n🎉 纯协议全链路成功: 任意账号密码已换出全套cookie! ({len(jar)}个)")
    print("已写", out)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())