wallet/script/webview-inspect.py

#!/usr/bin/env python3
"""Drive the running wallet's webview through WebKitGTK's remote inspector.

Start the app with `WEBKIT_INSPECTOR_HTTP_SERVER=127.0.0.1:9223 cargo tauri dev`
(from crates/wallet-app), then:

  script/webview-inspect.py eval "document.body.innerText"
  script/webview-inspect.py eval "document.querySelector('form').requestSubmit()"
  script/webview-inspect.py console 3

Needs `pip install --user websockets`. This is how the shell was verified on a
Wayland desktop where screenshots are refused and ImageMagick cannot see
WebKit's GPU surface: the DOM is the truth, and JavaScript drives the UI.
"""
import asyncio, json, sys
import websockets

WS = "ws://127.0.0.1:9223/socket/1/1/WebPage"

class Inspector:
    def __init__(self, ws):
        self.ws, self.n, self.target = ws, 0, None
        self.events = []

    async def raw(self, method, params=None):
        self.n += 1
        i = self.n
        await self.ws.send(json.dumps({"id": i, "method": method, "params": params or {}}))
        return i

    async def pump(self, timeout=5.0):
        """Read one frame, unwrapping target messages; returns (kind, payload)."""
        msg = json.loads(await asyncio.wait_for(self.ws.recv(), timeout=timeout))
        if msg.get("method") == "Target.targetCreated":
            self.target = msg["params"]["targetInfo"]["targetId"]
            return ("target", self.target)
        if msg.get("method") == "Target.dispatchMessageFromTarget":
            inner = json.loads(msg["params"]["message"])
            if "id" in inner:
                return ("reply", inner)
            return ("event", inner)
        if "id" in msg:
            return ("outer-reply", msg)
        return ("outer-event", msg)

    async def ensure_target(self):
        if self.target:
            return
        await self.raw("Target.exists") if False else None
        # The inspector announces the page target on connect.
        for _ in range(20):
            kind, p = await self.pump()
            if kind == "target":
                return
        raise RuntimeError("no target announced")

    async def call(self, method, params=None):
        await self.ensure_target()
        self.n += 1
        i = self.n
        inner = json.dumps({"id": i, "method": method, "params": params or {}})
        await self.ws.send(json.dumps({"id": 10_000 + i, "method": "Target.sendMessageToTarget",
                                       "params": {"targetId": self.target, "message": inner}}))
        while True:
            kind, p = await self.pump()
            if kind == "reply" and p.get("id") == i:
                return p
            if kind == "event":
                self.events.append(p)

async def main():
    cmd = sys.argv[1]
    async with websockets.connect(WS, max_size=1 << 24) as ws:
        insp = Inspector(ws)
        if cmd == "eval":
            # WebKitGTK ignores `awaitPromise`: a promise evaluates to `{}`.
            # To read an async result, park it on `window` and poll:
            #   eval 'window.__r = 0; p().then(v => { window.__r = v }); 1'
            #   eval 'window.__r'
            r = await insp.call("Runtime.evaluate", {"expression": sys.argv[2], "returnByValue": True, "awaitPromise": True})
            res = r.get("result", {})
            if "error" in r:
                print("ERROR:", r["error"]); return
            if "result" not in res:
                print("RAW:", json.dumps(r)[:800]); return
            v = res.get("result", {})
            val = v.get("value", v)
            print(json.dumps(val, indent=1) if isinstance(val, (dict, list)) else val)
            if res.get("wasThrown"):
                print("THROWN")
        else:
            secs = float(sys.argv[2]) if len(sys.argv) > 2 else 3.0
            await insp.call("Console.enable")
            await insp.call("Runtime.enable")
            loop = asyncio.get_event_loop(); end = loop.time() + secs
            while loop.time() < end:
                try:
                    kind, p = await insp.pump(timeout=max(0.1, end - loop.time()))
                except asyncio.TimeoutError:
                    break
                if kind == "event" and p.get("method") == "Console.messageAdded":
                    m = p["params"]["message"]
                    print(f"[{m.get('level')}] {m.get('text')}  ({m.get('url','')}:{m.get('line','')})")
            for p in insp.events:
                if p.get("method") == "Console.messageAdded":
                    m = p["params"]["message"]
                    print(f"[{m.get('level')}] {m.get('text')}  ({m.get('url','')}:{m.get('line','')})")

asyncio.run(main())