Pup - the AI's own browser

@@ -0,0 +1,145 @@+#!/usr/bin/env python3+"""not_intercepted.py — the SDK's NOT-INTERCEPTED audit (adom/pup-bridge#61, ab#86).++ab is supposed to FORWARD a bridge's verbs and relay the response verbatim. When ab core carries its+own implementation of one of your verbs, nothing errors: the caller gets a plausible, well-formed+answer that is not yours, and the blame lands on your repo. The tell is the KEY SET.++For each verb this calls pup's OWN port directly (POST /command) and the same verb through+adom-bridge, then diffs the CONTENT keys. ab lifts pup's {success, output:"<json>"} envelope to top+level by design, so both sides are unwrapped before comparing — otherwise every verb looks wrong and+real interception hides in the noise.++    tools/not_intercepted.py [TARGET]        # default AdomLapper++Run after ANY ab upgrade. Read-only verbs only; it must never mutate the user's desktop.++Two hard-won rules live in this file:+  * a probe that could not run is NOT a pass — it is reported and makes the run inconclusive;+  * the request body is verified on disk before the direct call, because a stale body file made an+    earlier version compare pup_status against every verb and report six false mismatches.+"""+import json+import subprocess+import sys++TARGET = sys.argv[1] if len(sys.argv) > 1 else "AdomLapper"+CLI = ["adom-bridge", "--target", TARGET]+BODY_WIN = r"C:\Users\john\AppData\Local\Temp\pup-audit-body.json"+++def cli(verb, args, timeout=90):+    try:+        p = subprocess.run(CLI + [verb, json.dumps(args)], capture_output=True, text=True, timeout=timeout)+        raw = p.stdout or ""+        i = raw.find("{")+        return json.loads(raw[i:]) if i >= 0 else {}+    except Exception:+        return {}+++def content_keys(payload):+    """Keys that describe the RESPONSE CONTENT, with the output envelope unwrapped and ab's+    transport/telemetry additions removed."""+    ENVELOPE = {+        "durationMs", "status", "success", "error", "aiThread", "window", "sessionId", "ok",+        "timeoutSeconds", "_hint", "_next", "related", "_related", "_cacheHint", "narrative",+        "_reportIssues", "_timeoutHint", "output",+        # ab adds statusVerb from the bridge manifest on its way out — a documented transport+        # addition, present on verbs it demonstrably forwards, so it is not an interception tell.+        "statusVerb",+    }+    keys = set(payload.keys())+    out = payload.get("output")+    if isinstance(out, str):+        try:+            keys |= set(json.loads(out).keys())+        except Exception:+            pass+    return keys - ENVELOPE+++def direct_call(port, verb, args):+    """Call pup's own port. The body goes to a file (quoting a JSON body through bash -> CLI JSON ->+    cmd defeated three attempts), and the file is READ BACK to prove it holds THIS verb."""+    body = json.dumps({"command": verb, "args": args})+    cli("desktop_write_file", {"path": BODY_WIN, "content": body,+                               "reason": f"NOT-INTERCEPTED audit: stage the request body for {verb}"})+    check = cli("shell_execute", {"command": r"type %TEMP%\pup-audit-body.json",+                                  "reason": "NOT-INTERCEPTED audit: verify the staged body"})+    staged = (check.get("stdout") or "").strip()+    if verb not in staged:+        return None, f"body file does not contain {verb} (staged: {staged[:60]!r})"+    r = cli("shell_execute", {+        "command": (r'curl.exe -s -X POST -H "Content-Type: application/json" '+                    rf"--data-binary @%TEMP%\pup-audit-body.json http://127.0.0.1:{port}/command"),+        "reason": f"NOT-INTERCEPTED audit: call pup's own port directly for {verb}"})+    out = (r.get("stdout") or "").strip()+    if not out:+        return None, "empty response from pup's port"+    try:+        return json.loads(out), None+    except Exception as e:+        return None, f"unparseable: {e}"+++def main():+    bl = cli("bridge_list", {})+    port = None+    for b in bl.get("bridges") or []:+        if isinstance(b, dict) and b.get("name") == "puppeteer":+            port = (b.get("spawn") or {}).get("runtimePort") or b.get("port")+    if not port:+        print(f"could not resolve pup's port on {TARGET}")+        return 1+    print(f"== NOT-INTERCEPTED audit on {TARGET} (pup port {port}) ==")++    sid = "ni-audit-probe"+    cli("pup_open_window", {"sessionId": sid, "url": "about:blank", "foreground": False}, timeout=150)++    verbs = [+        ("pup_status", {}), ("pup_readiness", {}), ("pup_list_windows", {}), ("pup_describe", {}),+        ("pup_window_info", {"sessionId": sid}), ("pup_list_tabs", {"sessionId": sid}),+        ("pup_lower_os_window", {"sessionId": sid}),+        # no foregroundReason on purpose: both sides should REFUSE identically, and a refusal that+        # differs is as telling as a success that differs+        ("pup_raise_os_window", {"sessionId": sid}),+        ("pup_alert_window", {"sessionId": sid, "stop": True}),+        ("pup_progress", {"sessionId": sid, "mode": "off"}),+    ]+    bad, skipped, findings = 0, 0, []+    for verb, args in verbs:+        d, err = direct_call(port, verb, args)+        if d is None:+            print(f"  ?? {verb} — DIRECT CALL FAILED ({err}); probe did not run, this is NOT a pass")+            skipped += 1+            continue+        routed = cli(verb, args)+        dk, rk = content_keys(d), content_keys(routed)+        only_ab, only_pup = rk - dk, dk - rk+        if only_ab or only_pup:+            bad += 1+            findings.append((verb, sorted(only_ab), sorted(only_pup)))+            print(f"  MISMATCH {verb}")+            if only_ab:+                print(f"      only via ab  (ab may be answering): {sorted(only_ab)}")+            if only_pup:+                print(f"      only direct  (ab dropped these):   {sorted(only_pup)}")+        else:+            print(f"  ok {verb} — {len(dk)} content keys, identical both ways")++    cli("pup_close_window", {"sessionId": sid})+    checked = len(verbs) - skipped+    print()+    if skipped:+        print(f"== INCONCLUSIVE: {checked} of {len(verbs)} compared, {skipped} probe(s) failed to run ==")+        return bad + skipped+    if bad:+        print(f"== {bad} of {checked} verbs differ — report these on the ab issue ==")+    else:+        print(f"== CLEAN: ab forwards all {checked} audited verbs verbatim ==")+    return bad+++if __name__ == "__main__":+    sys.exit(main())