chuanqi/cq2_sdk/scripts/cdp_captcha_extract.py

"""Extract _captcha_ from the LOGIN PAGE's localStorage (pc.hzwlan.com origin —
it was seen in the earlier dump) and re-trigger login with network capture
to see if the client auto-attaches it and what the server says.

Also screenshots the captcha widget to identify its type (slide vs icon).
"""
import asyncio
import base64
import json
import sys

import httpx
import websockets

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()


class Page:
    def __init__(self, ws):
        self.ws = ws
        self.mid = 0
        self.events = []

    async def call(self, method, params=None):
        self.mid += 1
        await self.ws.send(json.dumps({"id": self.mid, "method": method, "params": params or {}}))
        while True:
            msg = json.loads(await self.ws.recv())
            if msg.get("id") == self.mid:
                return msg.get("result", {})
            elif "method" in msg:
                self.events.append(msg)

    async def eval(self, expr):
        r = await self.call("Runtime.evaluate", {"expression": expr, "returnByValue": True})
        return r.get("result", {}).get("value")


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 target. Current targets:")
        for t in targets:
            print(f"  [{t['type']}] {t['url'][:70]}")
        return 1

    print(f"target: {login['url'][:60]}")
    async with websockets.connect(login["webSocketDebuggerUrl"], max_size=10**7) as ws:
        page = Page(ws)

        # 1. Read _captcha_ from THIS origin's localStorage
        cap = await page.eval("localStorage.getItem('_captcha_')")
        print(f"\n=== _captcha_ on pc.hzwlan.com ===")
        if cap:
            print(f"raw (first 80): {cap[:80]}")
            try:
                padded = cap + "=" * (-len(cap) % 4)
                decoded = base64.urlsafe_b64decode(padded).decode("utf-8", errors="replace")
                d = json.loads(decoded)
                print(f"decoded fields: {list(d.keys())}")
                print(f"  captchaId: {d.get('captchaId','?')}")
                print(f"  genTime:   {d.get('genTime','?')}  ({'stale' if (int(d.get('genTime',0) or 0) < 1789500000) else 'fresh'})")
                print(f"  lotNumber: {str(d.get('lotNumber','?'))[:30]}")
                print(f"  passToken: {str(d.get('passToken','?'))[:20]}...")
            except Exception as e:
                print(f"decode err: {e}")
        else:
            print("(empty on this origin)")

        # 2. Screenshot the page as-is (captcha state?)
        shot = await page.call("Page.captureScreenshot", {"format": "png"})
        if shot.get("data"):
            with open("cdp_captcha_state.png", "wb") as f:
                f.write(base64.b64decode(shot["data"]))
            print(f"\nscreenshot: cdp_captcha_state.png")

        # 3. Inspect the geetest widget structure (what TYPE of captcha?)
        gt_info = await page.eval("""
        (function() {
            const g = document.querySelector('.geetest_box, [class*=geetest_panel]');
            if (!g) return JSON.stringify({present: false});
            const r = g.getBoundingClientRect();
            // v4 widget internals
            const hasSlider = !!g.querySelector('[class*=slider_button], .geetest_btn_drag, [class*=btn_drag]');
            const hasIcons = !!g.querySelector('[class*=icon], .geetest_icon');
            const hasImage = !!g.querySelector('img, canvas, [class*=bg_img], [class*=puzzle]');
            const texts = Array.from(g.querySelectorAll('*')).filter(e => e.children.length===0 && (e.innerText||'').trim()).slice(0,8).map(e => e.innerText.trim().slice(0,30));
            return JSON.stringify({present: true, x: r.x, y: r.y, w: r.width, h: r.height, hasSlider, hasIcons, hasImage, texts});
        })()
        """)
        print(f"\n=== geetest widget ===\n{gt_info}")

        # 4. Enable network + click login again, watch for byPwd WITH _captcha_
        await page.call("Network.enable")
        page.events.clear()

        # ensure credentials still filled
        filled = await page.eval("""
        (function() {
            const u = document.querySelector('#username');
            const p = document.querySelector('#password');
            return JSON.stringify({u: u ? (u.value||'').length : -1, p: p ? (p.value||'').length : -1});
        })()
        """)
        fs = json.loads(filled or "{}")
        if fs.get("u", 0) < 3:
            await page.eval(f"""
            (function() {{
                const set = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value').set;
                const u = document.querySelector('#username');
                const p = document.querySelector('#password');
                set.call(u, 'longwu988'); u.dispatchEvent(new Event('input', {{bubbles:true}}));
                set.call(p, 'Longwu1988'); p.dispatchEvent(new Event('input', {{bubbles:true}}));
                return 'filled';
            }})()
            """)

        print("\nclicking login (stored _captcha_ should auto-attach)...")
        await page.eval("""
        (function() {
            const btn = document.querySelector('button.ant-btn.login-btn, button.login-btn, button.ant-btn');
            if (btn) { btn.click(); return 'ok'; }
            return 'no btn';
        })()
        """)

        # monitor 20s for the byPwd request + response
        print("monitoring 20s...")
        await asyncio.sleep(20)

        reqs = {}
        bodies = {}
        for e in page.events:
            m = e.get("method", "")
            p = e.get("params", {})
            if m == "Network.requestWillBeSent":
                url = p.get("request", {}).get("url", "")
                if "byPwd" in url:
                    reqs[p["requestId"]] = {
                        "url": url[:80],
                        "postData": (p["request"].get("postData") or "")[:400],
                    }
            elif m == "Network.responseReceived":
                if p.get("requestId") in reqs:
                    reqs[p["requestId"]]["status"] = p["response"].get("status")

        print(f"\n=== byPwd requests this round: {len(reqs)} ===")
        for rid, r in reqs.items():
            print(f"  -> {r['url']}")
            print(f"     status={r.get('status','?')}")
            pd = r.get("postData", "")
            if "_captcha_" in pd:
                print(f"     *** _captcha_ ATTACHED ***")
                # show structure (redact values)
                try:
                    j = json.loads(pd)
                    keys = list(j.keys())
                    print(f"     body keys: {keys}")
                except Exception:
                    pass
            else:
                print(f"     no _captcha_ in body")
            try:
                b = await page.call("Network.getResponseBody", {"requestId": rid})
                txt = b.get("body", "")
                if b.get("base64Encoded"):
                    txt = base64.b64decode(txt).decode("utf-8", errors="replace")
                print(f"     resp: {txt[:250]}")
            except Exception as ex:
                print(f"     body err: {type(ex).__name__}")

        # 5. final page state
        final = await page.eval("location.href.slice(0,70)")
        print(f"\nfinal url: {final}")

    return 0


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