chuanqi/cq2_sdk/scripts/hitl_login.py

"""Human-in-the-loop login: automate fill+click, user drags slider.

Steps:
  1. (capture running in background)
  2. CDP: fill credentials + click login -> geetest pops up
  3. Bring the 996box window to foreground (user drags slider)
  4. Poll up to 240s: cq2.exe spawn? gateway connections? capture lines?
  5. On traffic: report and exit (analysis happens after)
"""
import asyncio
import json
import subprocess
import sys
import time

import httpx
import websockets

USERNAME = "longwu988"
PASSWORD = "Longwu1988"
CDP = "http://127.0.0.1:9222"


async def get_targets():
    async with httpx.AsyncClient(timeout=5) as c:
        r = await c.get(f"{CDP}/json/list")
        return r.json()


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


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


def gateway_conns():
    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


def capture_lines():
    try:
        out = subprocess.run(
            ["powershell", "-NoProfile", "-Command",
             "(Get-Content 'D:\\传奇\\cq2_sdk\\real_login.jsonl' -ErrorAction SilentlyContinue | Measure-Object).Count"],
            capture_output=True, text=True, timeout=5)
        return int(out.stdout.strip() or 0)
    except Exception:
        return 0


def focus_996box():
    subprocess.run(
        ["powershell", "-NoProfile", "-Command",
         "$p = Get-Process 996box -ErrorAction SilentlyContinue | Where-Object { $_.MainWindowTitle } | Select-Object -First 1;"
         "if ($p) { Add-Type '@\nusing System;using System.Runtime.InteropServices;\npublic class F{[DllImport(\"user32.dll\")] public static extern bool SetForegroundWindow(IntPtr h);}\n@'; [F]::SetForegroundWindow($p.MainWindowHandle) | Out-Null; 'focused: ' + $p.MainWindowTitle }"],
        capture_output=True, text=True, timeout=10)


async def main():
    targets = await get_targets()
    login = next((t for t in targets if "login" in t.get("url", "") and t["type"] == "page"), None)
    if not login:
        print("no login page. Targets:")
        for t in targets:
            print(f"  [{t['type']}] {t['url'][:70]}")
        return 1

    async with websockets.connect(login["webSocketDebuggerUrl"], max_size=10**7) as ws:
        # fill + click
        r = await cdp_eval(ws, f"""
        (function() {{
            const set = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value').set;
            const u = document.querySelector('#username');
            const p = document.querySelector('#password');
            if (!u || !p) return 'missing fields';
            set.call(u, '{USERNAME}'); u.dispatchEvent(new Event('input', {{bubbles:true}}));
            set.call(p, '{PASSWORD}'); p.dispatchEvent(new Event('input', {{bubbles:true}}));
            const btn = document.querySelector('button.ant-btn.login-btn, button.login-btn, button.ant-btn');
            if (!btn) return 'no button';
            btn.click();
            return 'clicked';
        }})()
        """, 1)
        print(f"fill+click: {r}")

        await asyncio.sleep(2.5)
        gt = await cdp_eval(ws, "!!document.querySelector('.geetest_box, [class*=geetest]')", 2)
        print(f"geetest visible: {gt}")

    # focus the window for the user
    focus_996box()
    print()
    print("*** 滑块已弹出 — 请在 996box 窗口里拖动滑块完成验证 ***")
    print("*** 后台抓包运行中，登录成功后自动进游戏，无需其他操作 ***")
    print()

    # monitor
    t0 = time.time()
    deadline = t0 + 300
    seen_traffic = False
    while time.time() < deadline:
        await asyncio.sleep(5)
        el = int(time.time() - t0)
        pid = cq2_pid()
        conns = gateway_conns()
        lines = capture_lines()
        flags = []
        if pid:
            flags.append(f"CQ2={pid}")
        if conns:
            flags.append(f"conns={conns}")
        if lines:
            flags.append(f"cap={lines}")
        print(f"  t+{el:>3}s {' '.join(flags) or '...waiting'}")
        if lines > 0:
            seen_traffic = True
            # let the login frames accumulate
            if lines >= 3 or (lines >= 1 and el > 60):
                break
        if pid and conns and el > 30:
            break

    print()
    if seen_traffic:
        print("*** CAPTURED REAL TRAFFIC — analyze real_login.jsonl now ***")
        return 0
    print("no traffic captured within window")
    return 1


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