#!/usr/bin/env python3 """ HTTP admin debug dashboard for ``zkac-node serve``. Runs the ZKAC TCP node in a background thread and serves a read-only web UI on another port. Intended for loopback + I2P server-tunnel forwarding. **Security:** This page is fully transparent (registry metadata, live sessions, public keys). Do not expose it to untrusted networks without tunnel ACLs. Usage:: uv sync --extra demo uv run python demo/zkac_admin_serve.py alice \\ --node-host 127.0.0.1 --node-port 9800 \\ --web-host 127.0.0.1 --web-port 8766 """ from __future__ import annotations import argparse import json import os import sys import threading import time from pathlib import Path from flask import Flask, Response, jsonify, render_template_string, request from zkac_cli.paths import user_dir from zkac_cli.server import serve from zkac_cli.server_debug import ( ServerDebugState, collect_registry_debug, data_dir_tree, server_key_meta, ) _PAGE = """ {% if refresh_s %} {% endif %} ZKAC node — admin debug

ZKAC node — admin debug

User {{ snap.userid }} · TCP {{ snap.listen.host }}:{{ snap.listen.port }} · uptime {{ snap.uptime_s }}s {% if refresh_s %} · auto-refresh {{ refresh_s }}s {% endif %}

Status

Data directory{{ snap.data_dir }}
Started (wall){{ snap.started_wall }}
Server public key{{ snap.server_public_key_hex or "—" }}
Registries (boot){{ snap.registries_loaded_boot }}
Active TCP sessions {{ snap.active_connection_count }} live
Processpid={{ proc.pid }} threads={{ proc.threads }}

Server key file

{{ sk_meta | tojson(indent=2) }}

Registries (on disk + parsed)

{% if reg %} {% for r in reg %} {% endfor %}
ID (file) version state hash bytes (state / cert) parse
{{ r.file_registry_id_hex[:16] }}… {{ r.get("version", "—") }} {% if r.get("state_hash_hex") %}{{ r.state_hash_hex[:24] }}…{% else %}—{% endif %} {{ r.state_bytes }} / {{ r.cert_bytes }} {% if r.parsed_ok %}ok{% else %}fail{% endif %}
{% else %}

No registry state files yet.

{% endif %}

Session connections (live)

{% if snap.active_connections %} {% for c in snap.active_connections %} {% endfor %}
idpeerphaseoptranscript auth registryrolemgmt #echo bytes
{{ c.id }} {{ c.peer }} {{ c.phase }} {{ c.hello_op or "—" }} {% if c.transcript_hash_hex %}{{ c.transcript_hash_hex[:20] }}…{% else %}—{% endif %} {% if c.auth_registry_hex %}{{ c.auth_registry_hex[:16] }}…{% else %}—{% endif %} {% if c.auth_role_hex %}{{ c.auth_role_hex[:16] }}…{% else %}—{% endif %} {{ c.mgmt_commands or 0 }} {{ c.bytes_echoed or 0 }}
{% else %}

No active connections.

{% endif %}

Recent connections

{% if snap.recent_connections %} {% for c in snap.recent_connections %} {% endfor %}
idpeerphasemgmt #echo byteserror
{{ c.id }} {{ c.peer }} {{ c.phase }} {{ c.mgmt_commands or 0 }} {{ c.bytes_echoed or 0 }} {{ c.error or "—" }}
{% else %}

No recent disconnects recorded.

{% endif %}

Data directory tree (debug)

{{ files | tojson(indent=2) }}

Full debug JSON

Machine-readable snapshot (same as /api/debug.json).

{{ full_json }}

Auto-refresh 3s · 5s · static · debug.json

""" def _full_payload(debug: ServerDebugState, data_dir: Path) -> dict: snap = debug.snapshot() reg = collect_registry_debug(data_dir) sk = server_key_meta(data_dir) files = data_dir_tree(data_dir) return { "snapshot": snap, "server_key_file": sk, "registries": reg, "data_dir_files": files, "process": {"pid": os.getpid(), "threads": threading.active_count()}, "generated_wall": time.time(), } def create_app(debug: ServerDebugState, data_dir: Path) -> Flask: app = Flask(__name__) @app.get("/") def index(): refresh = request.args.get("refresh", "").strip() refresh_s = int(refresh) if refresh.isdigit() and 1 <= int(refresh) <= 60 else None data = _full_payload(debug, data_dir) snap = data["snapshot"] proc = data["process"] full_json = json.dumps(data, indent=2, sort_keys=True) return render_template_string( _PAGE, snap=snap, reg=data["registries"], sk_meta=data["server_key_file"], files=data["data_dir_files"], full_json=full_json, proc=proc, refresh_s=refresh_s, ) @app.get("/api/debug.json") def api_debug(): return jsonify(_full_payload(debug, data_dir)) @app.get("/healthz") def healthz(): return Response("ok", mimetype="text/plain") return app def main() -> None: p = argparse.ArgumentParser(description="ZKAC node TCP + HTTP admin debug dashboard") p.add_argument("userid", help="user whose ~/.zkac//server/ holds node state") p.add_argument("--data-dir", default=None, help="override server data directory") p.add_argument("--node-host", default="127.0.0.1") p.add_argument("--node-port", type=int, default=9800) p.add_argument("--web-host", default="127.0.0.1") p.add_argument("--web-port", type=int, default=8766) args = p.parse_args() data_dir = Path(args.data_dir) if args.data_dir else user_dir(args.userid) / "server" data_dir = data_dir.resolve() debug = ServerDebugState(userid=args.userid, data_dir=str(data_dir)) t = threading.Thread( target=serve, kwargs={ "data_dir": str(data_dir), "host": args.node_host, "port": args.node_port, "debug": debug, }, name="zkac-serve", daemon=True, ) t.start() time.sleep(0.15) if not t.is_alive(): print("ZKAC node thread died on startup; check stderr above.", file=sys.stderr) sys.exit(1) app = create_app(debug, data_dir) print(f"ZKAC TCP node: {args.node_host}:{args.node_port} (data {data_dir})") print(f"Admin debug UI: http://{args.web_host}:{args.web_port}/") print("Warning: admin UI exposes live sessions and registry metadata.", file=sys.stderr) app.run(host=args.web_host, port=args.web_port, debug=False, threaded=True) if __name__ == "__main__": main()