chuanqi/demo_login.py

"""
996盒子 Web 通行证 - 最小单文件 Demo
=====================================
目的：验证 POC 报告中的技术细节，确认能跑通账号密码登录。

用法：
    python demo_login.py --login    # 触发登录，输入账号密码
    python demo_login.py --check    # 用已保存的 token 探测登录态
    python demo_login.py --clear    # 清掉本地保存的 token
    python demo_login.py --info     # 打印 session 文件信息

注意：
- 这是 DEMO，token 明文存到本地（不带加密），不要在生产环境用
- 第一次跑 --login 前，先用环境变量或编辑下面的 CONFIG 配置游戏参数
- 如果服务端返回 "请先通过滑块验证" 等错误，说明你这次触发了风控，
  换 IP / 等几小时 / 用浏览器登录一次后再试

依赖：
    pip install requests
"""

import argparse
import hashlib
import hmac
import json
import os
import sys
import time
import urllib.parse
from pathlib import Path

try:
    import requests
except ImportError:
    print("缺少依赖，请先运行: pip install requests")
    sys.exit(1)


# ============================================================================
# 配置区 - 根据你 996盒子登录页 URL 的参数填
# ============================================================================
# 登录页 URL 形如：
# https://land.996box.com/chuanshi-login/chuanshi-login.html?ID=...&appid=155342&channel=PC%E5%8C%85&gameid=214256&promote_id=261888&sdkappid=2zn9bGS0DlJW&type=pc
#
# 下面的值是从示例 URL 抽出来的，你必须按你自己的 URL 替换

CONFIG = {
    # 996盒子登录接口域（前端源码里写死的）
    "register_url": "https://user-sdkv2.dhsf.996db.cn",

    # 从你登录页 URL 拿
    "appid": 155342,                # &appid=
    "game_id": 214256,              # &gameid=
    "promote_id": 261888,           # &promote_id=
    "sdkappid": "2zn9bGS0DlJW",     # &sdkappid=
    "sdkchannel": "PC包",           # &channel= (URL 解码后)
    "sdkid": "",                    # &ID= (留空即可，会自己生成)

    # 固定值
    "vType": "csvKey",

    # 签名 key（前端 JS 硬编码，所有人共用同一个）
    "SIGN_KEY": "k1ebbkxlojjdbnffnp3w2ykk3p625xnn",
}

# 本地 session 文件位置
SESSION_DIR = Path(os.environ.get("APPDATA", str(Path.home()))) / "chuanshi_demo"
SESSION_FILE = SESSION_DIR / "session.json"


# ============================================================================
# 核心工具
# ============================================================================
def key_sort(params: dict) -> str:
    """复现前端 keySort：按 key 字典序升序，拼成 key1=val1&key2=val2&..."""
    return "&".join(f"{k}={params[k]}" for k in sorted(params.keys()))


def hmac_str(sign_str: str, key: str) -> str:
    """复现前端 Hmacstr：HmacSHA1 后 hex 再 URL encode"""
    sig = hmac.new(key.encode("utf-8"), sign_str.encode("utf-8"), hashlib.sha1).hexdigest()
    return urllib.parse.quote(sig, safe="")


def make_signature(params: dict, key: str) -> str:
    """在 params 上加 sign 字段"""
    out = dict(params)
    out["sign"] = hmac_str(key_sort(out), key)
    return out


def make_device_info(sdkappid: str, sdkchannel: str, sdkid: str) -> str:
    """复现前端 Escalation：构造 PC 模式的 _device_info"""
    data = {
        "net_type": "UNKNOWN",
        "app_ver": "1.0.0",
        "appid": sdkappid,
        "channel": sdkchannel,
        "sdk_ver": "1.0.0",
        "device": {
            "type": "PC",
            "model": "pc",
            "id": sdkid or "demo-device-id-not-real",
            "id1": "",
            "id2": "",
            "id3": "",
            "os": "Win11",
            "width": 1920,
            "height": 1080,
            "lang": "zh-CN",
            "net_type": "UNKNOWN",
        },
    }
    return urllib.parse.quote(json.dumps(data, ensure_ascii=False), safe="")


# ============================================================================
# 登录
# ============================================================================
def login(username: str, password: str) -> dict:
    """调用 /api/user/v5w/login，返回完整响应 dict"""
    cfg = CONFIG
    params = {
        "username": username,
        "password": password,
        "game_id": cfg["game_id"],
        "vType": cfg["vType"],
        "appid": cfg["appid"],
        "promote_id": cfg["promote_id"],
        "_device_info": make_device_info(
            cfg["sdkappid"], cfg["sdkchannel"], cfg["sdkid"]
        ),
    }
    signed = make_signature(params, cfg["SIGN_KEY"])

    url = f"{cfg['register_url']}/api/user/v5w/login"
    print(f"[POST] {url}")
    print(f"[BODY] {json.dumps(signed, ensure_ascii=False, indent=2)[:500]}...")

    r = requests.post(url, json=signed, timeout=15)
    print(f"[STATUS] {r.status_code}")
    try:
        body = r.json()
    except Exception:
        body = {"raw": r.text}
    print(f"[RESPONSE] {json.dumps(body, ensure_ascii=False, indent=2)[:1500]}")
    return body


# ============================================================================
# 探测登录态
# ============================================================================
def check_session(token: str) -> dict:
    """
    用 token 调 /api/user/identification，看返回。
    - 服务端要求这个接口带 GET-X-Token，token 错的话会拒
    - 业务逻辑里这个接口是提交实名认证的，但只发 GET-X-Token 不发实名数据
      也能让服务端做 token 校验
    """
    cfg = CONFIG
    params = {
        "vType": cfg["vType"],
        "appid": cfg["appid"],
        "promote_id": cfg["promote_id"],
        "_device_info": make_device_info(
            cfg["sdkappid"], cfg["sdkchannel"], cfg["sdkid"]
        ),
    }
    signed = make_signature(params, cfg["SIGN_KEY"])

    url = f"{cfg['register_url']}/api/user/identification?GET-X-Token={token}"
    print(f"[POST] {url}")
    r = requests.post(url, json=signed, timeout=15)
    print(f"[STATUS] {r.status_code}")
    try:
        body = r.json()
    except Exception:
        body = {"raw": r.text}
    print(f"[RESPONSE] {json.dumps(body, ensure_ascii=False, indent=2)[:1500]}")
    return body


# ============================================================================
# Session 存储（DEMO 阶段明文存，不加密）
# ============================================================================
def save_session(data: dict) -> None:
    SESSION_DIR.mkdir(parents=True, exist_ok=True)
    data["_saved_at"] = int(time.time())
    SESSION_FILE.write_text(
        json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    print(f"[OK] Session 已保存到 {SESSION_FILE}")


def load_session() -> dict | None:
    if not SESSION_FILE.exists():
        return None
    try:
        return json.loads(SESSION_FILE.read_text(encoding="utf-8"))
    except Exception as e:
        print(f"[WARN] 读取 session 失败: {e}")
        return None


def clear_session() -> None:
    if SESSION_FILE.exists():
        SESSION_FILE.unlink()
        print(f"[OK] Session 已清掉: {SESSION_FILE}")
    else:
        print(f"[INFO] 没有 session 文件: {SESSION_FILE}")


# ============================================================================
# CLI
# ============================================================================
def prompt_account():
    print("=" * 60)
    print("首次登录 - 请输入账号密码（输入将用于本次登录，不保存）")
    print("=" * 60)
    username = input("账号: ").strip()
    password = input("密码: ").strip()
    return username, password


def cmd_login():
    session = load_session()
    if session and session.get("token"):
        print(f"[INFO] 已有 session: uid={session.get('uid')}, "
              f"username={session.get('username')}")
        ans = input("是否重新登录？(y/N): ").strip().lower()
        if ans != "y":
            return

    username, password = prompt_account()
    if not username or not password:
        print("[ERROR] 账号密码不能为空")
        return

    resp = login(username, password)

    if resp.get("code") == 200 and resp.get("data", {}).get("token"):
        data = resp["data"]
        save_session({
            "token": data["token"],
            "uid": data.get("uid"),
            "username": data.get("username"),
            "mobile": data.get("mobile"),
            "raw": data,
        })
        print()
        print("=" * 60)
        print("✅ 登录成功！")
        print(f"  token = {data['token'][:8]}...{data['token'][-4:]}")
        print(f"  uid   = {data.get('uid')}")
        print(f"  账号  = {data.get('username')}")
        print(f"  手机  = {data.get('mobile')}")
        print("=" * 60)
        print()
        print("下一步可以跑：")
        print("  python demo_login.py --check   # 用 token 探测登录态")
    else:
        print()
        print("=" * 60)
        print("❌ 登录失败")
        print(f"  code = {resp.get('code')}")
        print(f"  msg  = {resp.get('msg')}")
        print("=" * 60)
        if "极验" in str(resp) or "captcha" in str(resp).lower() or "jiyan" in str(resp).lower():
            print()
            print("⚠️  触发了滑块验证。本次 POC 暂不处理滑块，")
            print("   你可以：")
            print("   1) 换 IP / 用浏览器手动登录一次后再试")
            print("   2) 等几小时风控降级")
            print("   3) 启动 Playwright 走半自动重登（见报告）")


def cmd_check():
    session = load_session()
    if not session or not session.get("token"):
        print("[ERROR] 没有 session，请先跑 --login")
        return

    token = session["token"]
    print(f"[INFO] 用 uid={session.get('uid')} 的 token 探测登录态")
    print(f"[INFO] token = {token[:8]}...{token[-4:]}")
    print()
    resp = check_session(token)
    print()
    print("=" * 60)
    code = resp.get("code")
    msg = resp.get("msg", "")
    if code == 200:
        print("✅ Token 有效（接口返回 200）")
    elif code in (401, 403) or "token" in msg.lower() or "登录" in msg or "失效" in msg:
        print("❌ Token 失效或已过期，请重新跑 --login")
    else:
        print(f"⚠️  返回 code={code}, msg={msg!r}")
        print("    如果是 '参数错误'/'缺少参数' 类，说明 token 有效，")
        print("    只是这个接口需要更多 body 参数。")
    print("=" * 60)


def cmd_clear():
    clear_session()


def cmd_info():
    session = load_session()
    if not session:
        print(f"[INFO] 没有 session 文件: {SESSION_FILE}")
        return
    print(f"[INFO] Session 文件: {SESSION_FILE}")
    print(f"  uid        = {session.get('uid')}")
    print(f"  username   = {session.get('username')}")
    print(f"  mobile     = {session.get('mobile')}")
    print(f"  token      = {session.get('token', '')[:8]}..."
          f"{session.get('token', '')[-4:] if session.get('token') else ''}")
    print(f"  saved_at   = {session.get('_saved_at')}")
    if session.get("_saved_at"):
        age_hours = (time.time() - session["_saved_at"]) / 3600
        print(f"  距今       = {age_hours:.1f} 小时")


def main():
    parser = argparse.ArgumentParser(
        description="996盒子 Web 通行证 - 最小单文件 Demo",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    parser.add_argument("--login", action="store_true", help="执行登录流程")
    parser.add_argument("--check", action="store_true", help="探测已保存 token 是否有效")
    parser.add_argument("--clear", action="store_true", help="清掉本地 session")
    parser.add_argument("--info", action="store_true", help="显示当前 session 信息")
    args = parser.parse_args()

    if args.login:
        cmd_login()
    elif args.check:
        cmd_check()
    elif args.clear:
        cmd_clear()
    elif args.info:
        cmd_info()
    else:
        parser.print_help()


if __name__ == "__main__":
    main()