feat(hybrid-gateway): integrate muse-cli with Cloudflare netns isolation, symmetric sidechat routing, and 646-pip sync unblock

bin/box-query.py

@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+"""
+box-query.py — Lightweight client to query box.muse-dev.online APIs over HTTPS.
+
+Works inside containers and nodes without direct SSH access:
+Uses SSH signature authentication (?identity=bl&ts=...&sig=...) against
+the VM Box API.
+
+Usage:
+  box-query.py timers
+  box-query.py jobs
+  box-query.py agents
+  box-query.py dms [limit]
+"""
+
+import argparse
+import json
+import os
+import subprocess
+import sys
+import time
+import urllib.parse
+import urllib.request
+
+BOX_API_BASE = os.environ.get("BOX_API_BASE", "https://box.muse-dev.online")
+BOX_SIGN_KEY = os.environ.get("BOX_SIGN_KEY", os.path.expanduser("~/.ssh/id_ed25519"))
+
+def sign_request(identity, endpoint):
+    ts = str(int(time.time()))
+    payload = f"{ts}\n{endpoint}".encode()
+    try:
+        p = subprocess.run(
+            ["ssh-keygen", "-Y", "sign", "-f", BOX_SIGN_KEY, "-n", "box"],
+            input=payload, capture_output=True, timeout=15)
+        if p.returncode != 0:
+            return None, None
+        return ts, p.stdout.decode()
+    except Exception:
+        return None, None
+
+def query(endpoint):
+    ts, sig = sign_request("bl", endpoint)
+    if not ts or not sig:
+        sys.stderr.write("Failed to sign request (missing key or ssh-keygen error)\n")
+        sys.exit(1)
+
+    query_str = urllib.parse.urlencode({
+        "identity": "bl",
+        "ts": ts,
+        "sig": sig
+    })
+    url = f"{BOX_API_BASE}/api/box/{endpoint}?{query_str}"
+    req = urllib.request.Request(
+        url,
+        headers={"User-Agent": "NetVM-box-query/1.0 (container)"}
+    )
+    with urllib.request.urlopen(req, timeout=15) as resp:
+        data = resp.read().decode("utf-8")
+        try:
+            return json.loads(data)
+        except Exception:
+            return data
+
+def main():
+    p = argparse.ArgumentParser(description="Query box.muse-dev.online APIs via HTTPS signature auth")
+    p.add_argument("endpoint", choices=["timers", "jobs", "agents", "dms", "health"])
+    p.add_argument("--json", action="store_true")
+    args = p.parse_args()
+
+    res = query(args.endpoint)
+    print(json.dumps(res, indent=2))
+
+if __name__ == "__main__":
+    main()