chuanqi/cq2_sdk/scripts/cdp_session_check.py

"""Check 996box session state via CDP: localStorage tokens, then try
navigating straight to the main page (skipping login if session alive).
"""
import asyncio
import json
import sys

import httpx
import websockets


async def get_targets():
    async with httpx.AsyncClient(timeout=5) as c:
        r = await c.get("http://127.0.0.1:9222/json/list")
        return r.json()


async def cdp_eval(ws, expr: str, msg_id: int) -> dict:
    payload = json.dumps({"id": msg_id, "method": "Runtime.evaluate",
                          "params": {"expression": expr, "returnByValue": True}})
    await ws.send(payload)
    while True:
        msg = json.loads(await ws.recv())
        if msg.get("id") == msg_id:
            return msg.get("result", {})


async def cdp_navigate(ws, url: str, msg_id: int) -> dict:
    payload = json.dumps({"id": msg_id, "method": "Page.navigate",
                          "params": {"url": url}})
    await ws.send(payload)
    while True:
        msg = json.loads(await ws.recv())
        if msg.get("id") == msg_id:
            return msg.get("result", {})


async def main() -> int:
    targets = await get_targets()
    print("=== current CDP targets ===")
    for t in targets:
        print(f"  [{t['type']}] {t.get('title','')[:30]}  {t['url'][:75]}")

    login = next((t for t in targets if "login" in t.get("url", "")), None)
    if not login:
        print("\nno login page target — checking any hzwlan page")
        login = next((t for t in targets if "hzwlan" in t.get("url", "") or "996box" in t.get("url", "")), None)
    if not login:
        print("no usable target")
        return 1

    ws_url = login["webSocketDebuggerUrl"]
    print(f"\nusing: {login['url'][:60]}")

    async with websockets.connect(ws_url, max_size=10 * 1024 * 1024) as ws:
        # 1. localStorage recon
        print("\n=== localStorage keys ===")
        res = await cdp_eval(ws, """
        (function() {
            const out = {};
            for (let i = 0; i < localStorage.length; i++) {
                const k = localStorage.key(i);
                const v = localStorage.getItem(k) || '';
                out[k] = v.length > 60 ? v.slice(0, 60) + '...' : v;
            }
            return JSON.stringify(out, null, 1);
        })()
        """, 1)
        val = res.get("result", {}).get("value", "{}")
        try:
            ls = json.loads(val)
            for k, v in ls.items():
                print(f"  {k} = {v}")
        except Exception:
            print(val[:500])

        # 2. Check for token presence
        res2 = await cdp_eval(ws, """
        (function() {
            const t = localStorage.getItem('token') || localStorage.getItem('TOKEN') || '';
            return JSON.stringify({hasToken: t.length > 0, len: t.length});
        })()
        """, 2)
        print(f"\ntoken check: {res2.get('result', {}).get('value')}")

        # 3. Navigate to the main/game page — session cookie may carry us through
        print("\n=== navigating to main page (game list) ===")
        await cdp_navigate(ws, "https://h5qfdt.996box.com/", 3)
        await asyncio.sleep(6)
        res3 = await cdp_eval(ws, "location.href + ' | title=' + document.title", 4)
        print(f"after nav: {res3.get('result', {}).get('value')}")

    return 0


sys.exit(asyncio.run(main()))