chuanqi/cq2_sdk/scripts/cdp_login.py

"""Drive 996box login via CDP + capture the real game login traffic.

Sequence:
  1. (capture started separately — run capture_cq2.py first!)
  2. Fill #username / #password via CDP
  3. Click the login button (ant-btn)
  4. Poll: page navigates? Geetest iframe appears? cq2.exe spawns?
  5. Report state each step
"""
import asyncio
import json
import subprocess
import sys

import httpx
import websockets

USERNAME = "longwu988"
PASSWORD = "Longwu1988"


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_call(ws, msg_id: int, method: str, params: dict = None) -> dict:
    await ws.send(json.dumps({"id": msg_id, "method": method, "params": params or {}}))
    while True:
        msg = json.loads(await ws.recv())
        if msg.get("id") == msg_id:
            return msg.get("result", {})


async def cdp_eval(ws, expr: str, msg_id: int = 100) -> dict:
    return await cdp_call(ws, msg_id, "Runtime.evaluate", {"expression": expr, "returnByValue": True})


def cq2_running() -> bool:
    try:
        out = subprocess.run(
            ["powershell", "-NoProfile", "-Command",
             "Get-Process cq2 -ErrorAction SilentlyContinue | Select-Object -ExpandProperty Id"],
            capture_output=True, text=True, timeout=5,
        )
        return bool(out.stdout.strip())
    except Exception:
        return False


def gateway_conns() -> int:
    try:
        out = subprocess.run(
            ["powershell", "-NoProfile", "-Command",
             "(netstat -ano | Select-String '47.118.|:12000' | Measure-Object).Count"],
            capture_output=True, text=True, timeout=5,
        )
        return int(out.stdout.strip() or 0)
    except Exception:
        return 0


async def main() -> int:
    targets = await get_targets()
    login = next((t for t in targets if "login" in t.get("url", "")), None)
    if not login:
        print("ERROR: login page not found. Is 996box on the login screen?")
        return 1
    ws_url = login["webSocketDebuggerUrl"]
    print(f"login page: {login['url'][:60]}")

    async with websockets.connect(ws_url, max_size=10 * 1024 * 1024) as ws:
        # Step 1: fill username
        print("filling username...")
        await cdp_eval(ws, f"""
        (function() {{
            const el = document.querySelector('#username');
            if (!el) return 'no #username';
            const setter = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value').set;
            setter.call(el, '{USERNAME}');
            el.dispatchEvent(new Event('input', {{bubbles: true}}));
            el.dispatchEvent(new Event('change', {{bubbles: true}}));
            return 'filled';
        }})()
        """, 1)
        await asyncio.sleep(0.4)

        # Step 2: fill password
        print("filling password...")
        await cdp_eval(ws, f"""
        (function() {{
            const el = document.querySelector('#password');
            if (!el) return 'no #password';
            const setter = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value').set;
            setter.call(el, '{PASSWORD}');
            el.dispatchEvent(new Event('input', {{bubbles: true}}));
            el.dispatchEvent(new Event('change', {{bubbles: true}}));
            return 'filled';
        }})()
        """, 2)
        await asyncio.sleep(0.4)

        # verify both filled
        check = await cdp_eval(ws, """
        JSON.stringify({
            u: document.querySelector('#username')?.value?.length || 0,
            p: document.querySelector('#password')?.value?.length || 0
        })
        """, 3)
        print(f"verify: {check.get('result', {}).get('value')}")

        # Step 3: click the login button
        print("clicking login button...")
        await cdp_eval(ws, """
        (function() {
            const btn = document.querySelector('button.ant-btn.login-btn, button.login-btn, button.ant-btn');
            if (!btn) return 'no button';
            btn.click();
            return 'clicked: ' + (btn.innerText || '').trim();
        })()
        """, 4)

        # Step 4: monitor for 60s — captcha? navigation? cq2?
        print()
        print("=== monitoring 90s: captcha / navigation / cq2 / traffic ===")
        for i in range(30):
            await asyncio.sleep(3)
            state = await cdp_eval(ws, """
            (function() {
                const gt = document.querySelector('.geetest_box, .geetest_panel, [class*=geetest], iframe[src*=geetest]');
                const url = location.href;
                const err = document.querySelector('.ant-message-error, .login-error, [class*=error]');
                return JSON.stringify({
                    url: url.slice(0, 70),
                    geetest: !!gt,
                    errText: err ? (err.innerText||'').slice(0,40) : ''
                });
            })()
            """, 100 + i)

            val = state.get("result", {}).get("value", "{}")
            try:
                st = json.loads(val)
            except Exception:
                st = {"url": "?", "geetest": False, "errText": "?"}

            cq2 = cq2_running()
            conns = gateway_conns()
            t = (i + 1) * 3
            flags = []
            if st.get("geetest"):
                flags.append("GEETEST!")
            if cq2:
                flags.append("CQ2!")
            if conns:
                flags.append(f"conns={conns}")
            if st.get("errText"):
                flags.append(f"err={st['errText'][:20]}")
            print(f"  t+{t:>3}s  url={st.get('url','?')[:55]:<55}  {' '.join(flags)}")

            if cq2 and conns:
                print()
                print("*** GAME LAUNCHED WITH TRAFFIC — capture in progress ***")
                # let it run a bit more for the login frames
                await asyncio.sleep(20)
                return 0

        print()
        print("monitoring done (90s)")
        return 0


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