box/bin/meta-acct.py

#!/usr/bin/env python3
"""Meta Accounts Center read API (via authenticated browser session).

Usage:
  meta-acct.py list-linked <agent>       # linked profiles in this Meta Account
  meta-acct.py security-status <agent>   # login & recovery / security checkup summary
  meta-acct.py login-activity <agent>    # "Where you're logged in" sessions

The agent's browser must have an active Meta session (phone/email OTP login).
Reads NODES.md for the CDP port. Outputs JSON.

Scope: Accounts Center linkage/security surface only. No writes.
"""
import json, sys, time, urllib.request, os

NETVM = os.environ.get("NETVM_DIR", os.path.expanduser("~/Projects/NetVM"))
AC_BASE = "https://accountscenter.meta.com"

def cdp_port(agent):
    for line in open(os.path.join(NETVM, "NODES.md")):
        line = line.strip()
        if not line.startswith("|"):
            continue
        cells = [c.strip() for c in line.strip("|").split("|")]
        if len(cells) >= 4 and cells[0] == agent:
            return int(cells[3])
    raise SystemExit(f"meta-acct: unknown agent '{agent}' (not in NODES.md)")

def cdp_get(port, path, method="GET", timeout=10):
    req = urllib.request.Request(f"http://127.0.0.1:{port}{path}", method=method)
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read())

def get_ac_tab(port):
    tabs = cdp_get(port, "/json/list")
    for t in tabs:
        if "accountscenter.meta.com" in t.get("url", "") and t.get("type") == "page":
            return t
    # create one
    return cdp_get(port, f"/json/new?{AC_BASE}/", method="PUT")

def evaluate(port, ws_url, js, timeout=15):
    import websocket
    ws = websocket.create_connection(ws_url, timeout=timeout)
    try:
        ws.send(json.dumps({"id": 1, "method": "Page.navigate",
                            "params": {"url": js[0]}}))
        ws.recv()
        time.sleep(4)
        ws.send(json.dumps({"id": 2, "method": "Runtime.evaluate",
            "params": {"expression": js[1], "returnByValue": True}}))
        resp = json.loads(ws.recv())
        return json.loads(resp["result"]["result"]["value"])
    finally:
        ws.close()

def cmd_list_linked(port):
    tab = get_ac_tab(port)
    data = evaluate(port, tab["webSocketDebuggerUrl"], (
        f"{AC_BASE}/account_overview/",
        """JSON.stringify((() => {
          const h2 = [...document.querySelectorAll('h2')]
            .find(e=>e.innerText.trim()==='Profiles');
          // Full section is 3 levels up (H2 > DIV > DIV > MAIN)
          const container = h2?.parentElement?.parentElement?.parentElement;
          return {
            email: (document.body.innerText.match(
              /[\\w.+-]+@[\\w-]+\\.[\\w.]+/)||[])[0]||null,
            section: container?.innerText?.slice(0,1000) || null
          };
        })())"""))
    # Parse: lines after "Profiles" are name/type pairs until "Add more",
    # then "AI and devices" section follows
    profiles = []
    section = data.get("section", "")
    if section:
        lines = [l.strip() for l in section.split("\n") if l.strip()]
        try:
            start = lines.index("Profiles") + 1
        except ValueError:
            start = 0
        i = start
        while i < len(lines):
            if lines[i] == "Add more":
                i += 1
                continue
            if lines[i] == "AI and devices":
                # Next line is the device/service name
                if i + 1 < len(lines):
                    profiles.append({"name": lines[i+1], "type": "ai_device"})
                break
            # Expect name + type pair
            if i + 1 < len(lines) and lines[i+1] not in (
                    "Add more", "AI and devices", "Profiles"):
                profiles.append({"name": lines[i],
                                 "type": lines[i+1].lower()})
                i += 2
            else:
                i += 1
    return {"email": data.get("email"), "profiles": profiles}

def cmd_security_status(port):
    tab = get_ac_tab(port)
    data = evaluate(port, tab["webSocketDebuggerUrl"], (
        f"{AC_BASE}/password_and_security/",
        """JSON.stringify({
          checkup: document.body.innerText.match(
            /Meta Security Checkup\\s*(\\d+) recommended actions/)?.[1] || null,
          sections: [...document.querySelectorAll('h2')]
            .map(e=>e.innerText.trim()).slice(0,10),
          body: document.body.innerText.slice(0,2000)
        })"""))
    return {"security_checkup_actions": data.get("checkup"),
            "sections": data.get("sections")}

def cmd_login_activity(port):
    tab = get_ac_tab(port)
    # "Where you're logged in" is on the password_and_security page
    data = evaluate(port, tab["webSocketDebuggerUrl"], (
        f"{AC_BASE}/password_and_security/",
        """JSON.stringify({
          body: document.body.innerText.slice(0,4000)
        })"""))
    body = data.get("body", "")
    # Extract the "Where you're logged in" section
    idx = body.find("Where you're logged in")
    section = body[idx:idx+1500] if idx >= 0 else ""
    return {"where_logged_in": section}

def main():
    if len(sys.argv) != 3:
        print(__doc__.strip().split("\n")[0])
        print("Usage: meta-acct.py <list-linked|security-status|login-activity> <agent>")
        sys.exit(2)
    cmd, agent = sys.argv[1], sys.argv[2]
    port = cdp_port(agent)
    try:
        if cmd == "list-linked":
            out = cmd_list_linked(port)
        elif cmd == "security-status":
            out = cmd_security_status(port)
        elif cmd == "login-activity":
            out = cmd_login_activity(port)
        else:
            raise SystemExit(f"meta-acct: unknown command '{cmd}'")
    except Exception as e:
        print(json.dumps({"error": str(e), "agent": agent}))
        sys.exit(1)
    out["agent"] = agent
    out["checked_at"] = int(time.time())
    print(json.dumps(out, indent=2))

if __name__ == "__main__":
    main()