box/bin/meta-ac-snapshot.py

#!/usr/bin/env python3
"""Meta Accounts Center change-detection harness.

Captures a structural snapshot of the accountscenter.meta.com auth flow
via CDP inside a NetVM netns, diffs against the stored baseline.

Outcomes:
  PASS    - matches baseline (or first run establishes it)
  CHANGED - structural diff detected; needs human review, baseline untouched
  FAIL    - automation itself broke (browser/CDP/network error)

Usage:
  meta-ac-snapshot.py [--node NAME] [--promote] [--snapshot-dir DIR]

  --node      NetVM node to run in (default: phone)
  --promote   after human review, promote the latest snapshot to baseline
  --snapshot-dir  where snapshots live (default: ~/Projects/NetVM/snapshots/meta-ac)
"""
import argparse, base64, datetime, json, os, subprocess, sys, time
import urllib.parse, urllib.request

CDP_PORT = 19744

def log(*a):
    print(*a, flush=True)

def ns_exec(node, cmd):
    return subprocess.run(
        ["sudo", "-n", "ip", "netns", "exec", f"warp-{node}"] + cmd,
        capture_output=True, text=True)

def norm_url(u):
    """Strip query/fragment — nonces change every visit."""
    p = urllib.parse.urlparse(u)
    return f"{p.scheme}://{p.host}{p.path}" if hasattr(p, 'host') else f"{p.scheme}://{p.hostname}{p.path}"

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--node", default="phone")
    ap.add_argument("--promote", action="store_true")
    ap.add_argument("--snapshot-dir", default=os.path.expanduser(
        "~/Projects/NetVM/snapshots/meta-ac"))
    args = ap.parse_args()
    os.makedirs(args.snapshot_dir, exist_ok=True)
    baseline_path = os.path.join(args.snapshot_dir, "baseline.json")

    if args.promote:
        snaps = sorted(f for f in os.listdir(args.snapshot_dir)
                       if f.startswith("snap-") and f.endswith(".json"))
        if not snaps:
            log("no snapshots to promote"); return 2
        latest = os.path.join(args.snapshot_dir, snaps[-1])
        data = json.load(open(latest))
        data["promoted_at"] = datetime.datetime.now(datetime.timezone.utc).isoformat()
        json.dump(data, open(baseline_path, "w"), indent=2)
        log(f"promoted {snaps[-1]} -> baseline.json")
        return 0

    profile_dir = "/tmp/meta-ac-snap-profile"
    subprocess.run(["rm", "-rf", profile_dir])
    os.makedirs(profile_dir, exist_ok=True)

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

    # launch chromium inside the netns via a wrapper script
    wrapper = "/tmp/meta-ac-snap-run.py"
    open(wrapper, "w").write(WRAPPER_SRC)
    log(f"launching chromium in warp-{args.node} (CDP {CDP_PORT})...")
    proc = subprocess.Popen(
        ["sudo", "-n", "ip", "netns", "exec", f"warp-{args.node}",
         "python3", wrapper, str(CDP_PORT), profile_dir],
        stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
    try:
        out, _ = proc.communicate(timeout=120)
    except subprocess.TimeoutExpired:
        proc.kill(); log("FAIL: harness timed out"); return 1
    print(out)
    # wrapper prints SNAPSHOT_JSON=<json> on success
    snap = None
    for line in out.splitlines():
        if line.startswith("SNAPSHOT_JSON="):
            snap = json.loads(line[len("SNAPSHOT_JSON="):])
    if not snap:
        log("FAIL: no snapshot captured"); return 1

    snap["node"] = args.node
    snap["captured_at"] = datetime.datetime.now(datetime.timezone.utc).isoformat()

    # egress ip for context
    try:
        r = ns_exec(args.node, ["curl", "-s", "--max-time", "8",
                                "https://api.ipify.org"])
        snap["egress_ip"] = r.stdout.strip()
    except Exception:
        snap["egress_ip"] = "unknown"

    ts = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%d-%H%M%S")
    snap_path = os.path.join(args.snapshot_dir, f"snap-{ts}.json")
    json.dump(snap, open(snap_path, "w"), indent=2)
    log(f"snapshot saved: {snap_path}")

    if not os.path.exists(baseline_path):
        json.dump(snap, open(baseline_path, "w"), indent=2)
        log("PASS: baseline established (first run)")
        return 0

    baseline = json.load(open(baseline_path))
    diffs = diff_snapshots(baseline, snap)
    if not diffs:
        log("PASS: matches baseline")
        return 0
    log("CHANGED: structural diff detected (baseline untouched):")
    for d in diffs:
        log(f"  - {d}")
    log("review with: diff baseline.json snap-<ts>.json")
    log("promote after review with: --promote")
    return 3

def diff_snapshots(base, snap):
    diffs = []
    b_chain = [norm_url(u) for u in base.get("redirect_chain", [])]
    s_chain = [norm_url(u) for u in snap.get("redirect_chain", [])]
    if b_chain != s_chain:
        diffs.append(f"redirect_chain changed: {b_chain} -> {s_chain}")
    for key in ("forms", "inputs", "buttons"):
        b = sorted(base.get("dom_markers", {}).get(key, []))
        s = sorted(snap.get("dom_markers", {}).get(key, []))
        if b != s:
            added = [x for x in s if x not in b]
            removed = [x for x in b if x not in s]
            diffs.append(f"dom_markers.{key}: added={added} removed={removed}")
    if base.get("final_title") != snap.get("final_title"):
        diffs.append(f"final_title: {base.get('final_title')!r} -> {snap.get('final_title')!r}")
    return diffs

WRAPPER_SRC = '''
import json, subprocess, sys, time, os, urllib.request, base64
CDP_PORT = int(sys.argv[1])
PROFILE_DIR = sys.argv[2]
def http(path):
    with urllib.request.urlopen(f"http://127.0.0.1:{CDP_PORT}{path}", timeout=5) as r:
        return json.loads(r.read())
logf = open("/tmp/meta-ac-snap-chrome.log", "w")
proc = subprocess.Popen(["chromium", "--headless=new", "--disable-gpu", "--no-sandbox",
    "--disable-dev-shm-usage", f"--user-data-dir={PROFILE_DIR}",
    f"--remote-debugging-port={CDP_PORT}", "--remote-allow-origins=*", "about:blank"],
    stdout=logf, stderr=subprocess.STDOUT)
try:
    for i in range(30):
        try:
            ver = http("/json/version")
            if "webSocketDebuggerUrl" in ver: break
        except Exception: pass
        time.sleep(1)
    else:
        print("FAIL: CDP never came up"); sys.exit(1)
    import websocket
    bws = websocket.create_connection(ver["webSocketDebuggerUrl"], timeout=20)
    bws.send(json.dumps({"id": 1, "method": "Target.createTarget",
        "params": {"url": "https://accountscenter.meta.com"}}))
    target_id = json.loads(bws.recv())["result"]["targetId"]
    bws.close()
    # redirect chain: seed with the navigation target (we always start
    # there), then poll for where Meta sends us. Seeding fixes the race
    # where a fast redirect is missed by the poll interval.
    START_URL = "https://accountscenter.meta.com/"
    chain, seen = [START_URL], {START_URL}
    for _ in range(24):
        time.sleep(2)
        for t in http("/json/list"):
            if t.get("id") == target_id or "meta.com" in t.get("url", ""):
                u = t["url"]
                if u not in seen:
                    seen.add(u); chain.append(u)
                title = t.get("title", "")
                break
    # dom markers from the final tab
    tab_ws = None
    for t in http("/json/list"):
        if t.get("id") == target_id or "meta.com" in t.get("url", ""):
            tab_ws = t["webSocketDebuggerUrl"]; final_url = t["url"]; break
    ws = websocket.create_connection(tab_ws, timeout=20)
    js = """JSON.stringify({
        forms: [...document.forms].map(f => f.id || f.name || '(anon)'),
        inputs: [...document.querySelectorAll('input')].map(i => i.name || i.type || '(anon)'),
        buttons: [...document.querySelectorAll('button, [role=button]')].map(b => (b.innerText||'').trim()).filter(Boolean)
    })"""
    ws.send(json.dumps({"id": 1, "method": "Runtime.evaluate",
        "params": {"expression": js, "returnByValue": True}}))
    markers = json.loads(json.loads(ws.recv())["result"]["result"]["value"])
    # dedupe buttons, keep order
    markers["buttons"] = list(dict.fromkeys(markers["buttons"]))
    ws.close()
    snap = {"redirect_chain": chain, "final_url": final_url,
            "final_title": title, "dom_markers": markers}
    print("SNAPSHOT_JSON=" + json.dumps(snap))
finally:
    proc.terminate()
'''

if __name__ == "__main__":
    sys.exit(main())