chuanqi/cq2_sdk/scripts/probe_sms_endpoint_v2.py

"""Wider SMS endpoint probe — scan all 3 996box hosts × common paths.

Found 1 endpoint so far: /auth/v3/sms/sendLoginCode (exists but requires auth)
Looking for: the unauthenticated SMS-send endpoint that fires on first login.
"""
import asyncio
import sys
from pathlib import Path

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

from cq2_sdk.env_loader import load_env
load_env(_HERE.parent / ".env")

from cq2_sdk.login import Box996Login, ManualGeetestSolver

HOSTS = [
    "https://api.jiyuhz.cn",            # 密码登录用的 (api.jiyuhz.cn/auth/v3/login/safe/byPwd)
    "https://pc-api.996hezi.com",        # 业务 API (getTodayOpen 等)
    "https://user-api.996hezi.com",      # 用户 API
]

# 各种 SMS 相关 endpoint 命名模式
CANDIDATE_PATHS = [
    # api.jiyuhz.cn 风格 (auth/ 前缀)
    "/auth/v3/sms/sendLoginCode",       # 已确认 200 but 需要 auth
    "/auth/v3/login/sms/send",
    "/auth/v3/login/sendSmsCode",
    "/auth/v3/login/safe/sendSms",
    "/auth/v3/login/sendSms",
    "/auth/v3/login/sendLoginCode",
    "/auth/v3/login/sendCode",
    "/auth/v3/login/sendSmsCode",
    # 业务 API 风格 (无 /auth/ 前缀)
    "/v1/sms/sendCode",
    "/v1/sms/sendLoginCode",
    "/v1/sms/send",
    # user-api 风格
    "/v1/user/sendSms",
    "/v1/user/sendLoginCode",
    # 其它
    "/sms/send",
    "/sms/sendCode",
]


async def probe(box, host: str, path: str, phone: str = "00000000000") -> dict:
    """Returns dict with url, status, code, msg, is_real."""
    url = f"{host}{path}"
    try:
        r = await box._request("POST", host, path,
                               json_body={"phone": phone, "channel_id": "",
                                          "game_id": "", "inviter_uid": "",
                                          "needConfirmLogin": True})
        # 200 response: 996box returned valid JSON
        data_code = r.get("code", 0) if isinstance(r, dict) else 0
        msg = r.get("msg", "") if isinstance(r, dict) else ""
        # 区分: "需要登录" vs 真正 SMS 发送响应
        is_real_sms = (
            data_code == 200 and r.get("data", {}).get("expire") is not None
        )
        return {"url": url, "http": 200, "code": data_code, "msg": msg[:60],
                "is_real_sms": is_real_sms}
    except Exception as e:
        from cq2_sdk.login.box_996 import Box996Error, CaptchaRequiredError
        msg = str(e)[:80]
        if "404" in msg:
            return {"url": url, "http": 404, "code": 0, "msg": "404", "is_real_sms": False}
        if isinstance(e, CaptchaRequiredError):
            return {"url": url, "http": 200, "code": 101000, "msg": "geetest",
                    "is_real_sms": False}
        if isinstance(e, Box996Error):
            return {"url": url, "http": 200, "code": e.code, "msg": e.msg[:60],
                    "is_real_sms": False}
        return {"url": url, "http": 0, "code": 0, "msg": msg, "is_real_sms": False}


async def main() -> int:
    box = Box996Login(geetest=ManualGeetestSolver(), device_id="00:15:5d:0a:04:00")
    print(f"Probing {len(HOSTS)} hosts × {len(CANDIDATE_PATHS)} paths "
          f"= {len(HOSTS)*len(CANDIDATE_PATHS)} candidates\n")

    results = []
    for host in HOSTS:
        print(f"--- {host} ---")
        for path in CANDIDATE_PATHS:
            r = await probe(box, host, path)
            results.append(r)
            mark = ""
            if r["is_real_sms"]:
                mark = "[!!! REAL SMS !!!]"
            elif r["http"] == 200 and r["code"] in (200, 101000):
                mark = "[reachable]"
            elif r["http"] == 404:
                mark = "[404]"
            print(f"  {mark:<16} {r['url']:<55} code={r['code']:>4}  {r['msg']}")

    print()
    real = [r for r in results if r["is_real_sms"]]
    auth_req = [r for r in results if r["http"] == 200 and r["code"] not in (200, 101000)
               and "登录" in r["msg"]]
    if real:
        print(f"[!!! FOUND !!!] Real SMS endpoints: {len(real)}")
        for r in real:
            print(f"    {r['url']}")
    elif auth_req:
        print(f"[*] {len(auth_req)} endpoint(s) require auth (rejected our fake phone):")
        for r in auth_req:
            print(f"    {r['url']}  code={r['code']}  msg={r['msg']}")
    else:
        print("[!] No real SMS endpoint found. 996box may have a different flow now.")
        print("    → 建议你抓浏览器 Network 包: 打开 https://pc.hzwlan.com/ → 输手机号 → 点发验证码 → 看 F12 里的 URL")
    return 0


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))