75 lines
2.1 KiB
Python
75 lines
2.1 KiB
Python
|
#!/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()
|