box/bin/accounts-health.py

#!/usr/bin/env python3
"""Per-account session vitality check (runs INSIDE the node's netns).

Usage: sudo ip netns exec warp-<node> python3 accounts-health.py <cdp_port>

Probes the account's browser via CDP:
  - browser reachable
  - muse.ai tab present
  - login markers (heuristic: "Log in" button vs user content)

Prints JSON to stdout. Exit 0 on success, 1 if the browser is unreachable.
"""
import json, sys, urllib.request

CDP_PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 9410

def http(path, timeout=5):
    with urllib.request.urlopen(f"http://127.0.0.1:{CDP_PORT}{path}",
                                timeout=timeout) as r:
        return json.loads(r.read())

result = {"browser_up": False, "muse_tab": False,
          "session_alive": None, "title": "", "url": "", "detail": ""}

try:
    http("/json/version")
    result["browser_up"] = True
except Exception as e:
    result["detail"] = f"CDP unreachable: {e}"
    print(json.dumps(result)); sys.exit(1)

try:
    tabs = http("/json/list")
except Exception as e:
    result["detail"] = f"tab list failed: {e}"
    print(json.dumps(result)); sys.exit(0)

muse_tab = None
for t in tabs:
    url = t.get("url", "")
    if "muse.ai" in url and t.get("type") == "page":
        muse_tab = t
        break

if not muse_tab:
    result["detail"] = "no muse.ai tab open"
    print(json.dumps(result)); sys.exit(0)

result["muse_tab"] = True
result["url"] = muse_tab.get("url", "")
result["title"] = muse_tab.get("title", "")

# DOM heuristic via the tab's debugger socket
try:
    import websocket
    ws = websocket.create_connection(muse_tab["webSocketDebuggerUrl"], timeout=10)
    js = """JSON.stringify({
        loginButtons: [...document.querySelectorAll('button')].filter(
            b => /^\\s*log\\s*in\\s*$/i.test(b.innerText)).map(b => b.innerText.trim()),
        hasAvatar: !!document.querySelector(
            'img[alt*="avatar" i], [data-testid*="avatar" i], [aria-label*="profile" i]'),
        title: document.title,
        url: location.href
    })"""
    ws.send(json.dumps({"id": 1, "method": "Runtime.evaluate",
                        "params": {"expression": js, "returnByValue": True}}))
    resp = json.loads(ws.recv())
    ws.close()
    dom = json.loads(resp["result"]["result"]["value"])
    # Heuristic: login buttons present + no avatar => logged out.
    # access/verification gate => needs verification.
    # No login buttons (or avatar present) => likely logged in.
    if "access/verification" in dom["url"]:
        result["session_alive"] = False
        result["detail"] = "age verification gate (access/verification)"
    elif dom["loginButtons"] and not dom["hasAvatar"]:
        result["session_alive"] = False
        result["detail"] = f"login wall visible: {dom['loginButtons'][:2]}"
    else:
        result["session_alive"] = True
        result["detail"] = "no login wall detected"
except Exception as e:
    result["detail"] = f"DOM check failed: {e}"

print(json.dumps(result))