goyimnose/Medios-Macina
1
0
Fork 0
Code Issues Pull Requests Packages Releases 1 Wiki Activity

j

Browse Source
This commit is contained in:
Nose
2026-01-14 04:27:54 -08:00
parent cd60c86868
commit 187a230e98
8 changed files with 318 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

260
cmdnat/zerotier.py
View File

@@ -1,151 +1,153 @@
from __future__ import annotations
import os
from typing import Any, Dict, List, Optional, Sequence, Tuple
import sys
import requests
from pathlib import Path
from typing import Any, Dict, List, Optional, Sequence
# Add project root to sys.path
root = Path(__file__).resolve().parent.parent
if str(root) not in sys.path:
sys.path.insert(0, str(root))
from cmdlet._shared import Cmdlet, CmdletArg
from SYS.result_table import ResultTable
from SYS.config import load_config
from SYS.result_table import ResultTable
from API import zerotier as zt
def exec(pipe: Sequence[Any], args: Sequence[str], options: Dict[str, Any]) -> ResultTable:
table = ResultTable(title="ZeroTier Status", max_columns=10)
def exec_zerotier(result: Any, args: Sequence[str], config: Dict[str, Any]) -> int:
# Use provided config or fall back to CWD load
cfg = config if config else load_config(Path.cwd())
cfg = load_config()
table = ResultTable("ZeroTier Status")
# 1. Local Node Status
node_id = "unknown"
# 1. Local Hub Status
row = table.add_row()
row.add_column("TYPE", "HOST")
row.add_column("NAME", "localhost")
# Try to get node ID via CLI info
node_id = "???"
try:
# Best effort to get local node ID
st = zt._run_cli_json("status")
if isinstance(st, dict):
node_id = st.get("address") or node_id
except: pass
if hasattr(zt, "_run_cli_json"):
info = zt._run_cli_json("info", "-j")
node_id = info.get("address", "???")
except:
pass
row.add_column("ID", node_id)
# 2. Hosting Status
zt_net = cfg.get("networking", {}).get("zerotier", {})
serve_target = zt_net.get("serve")
if serve_target:
port = zt_net.get("port") or 999
net_id = zt_net.get("network_id") or "all"
# Check if local server is responsive
try:
# endpoint is /health for remote_storage_server
# We try 127.0.0.1 first with a more generous timeout
# Using a list of potential local hits to be robust against Windows networking quirks
import socket
status = "OFFLINE"
detail = ""
# Try to find the local ZT address for this network
zt_addrs = []
if net_id and net_id != "all":
zt_addrs = zt.get_assigned_addresses(net_id)
# We probe localhost for hosting status, but show ZT IP in the table
display_addr = zt_addrs[0] if zt_addrs else "localhost"
# Try probes
# Using 127.0.0.1 is often more reliable than 'localhost' on Windows
probe_targets = [f"http://127.0.0.1:{port}/health"]
if zt_addrs:
probe_targets.insert(0, f"http://{zt_addrs[0]}:{port}/health")
def get_local_ip():
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
ip = s.getsockname()[0]
s.close()
return ip
except:
return None
from API.HTTP import HTTPClient
with HTTPClient(timeout=1.0, retries=0) as client:
for url in probe_targets:
hosts = ["127.0.0.1", "localhost"]
local_ip = get_local_ip()
if local_ip:
hosts.append(local_ip)
success = False
last_err = ""
# Try multiple times if server just started
import time
for attempt in range(3):
for host in hosts:
try:
resp = client.get(url)
resp = requests.get(f"http://{host}:999/health", timeout=3, proxies={"http": None, "https": None})
if resp.status_code == 200:
status = "ONLINE"
payload = resp.json()
detail = f"Serving {payload.get('name') or serve_target}"
row.add_column("STATUS", "ONLINE")
row.add_column("ADDRESS", f"{host}:999")
row.add_column("DETAIL", f"Serving {cfg.get('active_store', 'default')}")
success = True
break
else:
status = f"HTTP {resp.status_code}"
except Exception as exc:
if not detail: # Keep the first failure reason if all fail
detail = f"Probe failed: {exc}"
if status == "OFFLINE" and not zt_addrs:
detail = "No ZeroTier IP assigned yet. Check 'zerotier-cli listnetworks'."
row = table.add_row()
row.add_column("TYPE", "HOST")
row.add_column("NAME", serve_target)
row.add_column("ID", net_id)
row.add_column("ADDRESS", f"{display_addr}:{port}")
row.add_column("STATUS", status)
row.add_column("DETAIL", detail)
# 3. Connections (Remote Stores)
zt_stores = cfg.get("store", {}).get("zerotier", {})
if zt_stores:
for name, sconf in zt_stores.items():
net_id = sconf.get("NETWORK_ID") or sconf.get("network_id") or ""
host = sconf.get("HOST") or sconf.get("host") or ""
port = sconf.get("PORT") or sconf.get("port") or 999
svc = sconf.get("SERVICE") or sconf.get("service") or "remote"
status = "probing..."
detail = ""
if not host:
status = "MISCONFIGURED"
detail = "No host IP"
else:
try:
from API.HTTP import HTTPClient
with HTTPClient(timeout=2.0) as client:
# Paths depend on service type
path = "/api_version" if svc == "hydrus" else "/health"
resp = client.get(f"http://{host}:{port}{path}")
if resp.status_code == 200:
status = "ONLINE"
if svc == "remote":
p = resp.json()
detail = f"Remote store: {p.get('name', 'unknown')}"
else:
detail = "Hydrus API"
else:
status = f"HTTP {resp.status_code}"
except Exception as exc:
status = "OFFLINE"
detail = str(exc)
row = table.add_row()
row.add_column("TYPE", "REMOTE")
row.add_column("NAME", name)
row.add_column("ID", net_id)
row.add_column("ADDRESS", f"{host}:{port}")
row.add_column("STATUS", status)
row.add_column("DETAIL", detail)
# 4. Networking Networks (Raw ZT status)
try:
nets = zt.list_networks()
if not nets:
row = table.add_row()
row.add_column("TYPE", "INFO")
row.add_column("NAME", "ZeroTier CLI")
row.add_column("STATUS", "No networks found or CLI error")
row.add_column("DETAIL", f"CLI Path: {zt._get_cli_path() or 'Not found'}")
elif resp.status_code == 401:
row.add_column("STATUS", "Serving (Locked)")
row.add_column("ADDRESS", f"{host}:999")
row.add_column("DETAIL", "401 Unauthorized - API Key required")
success = True
break
except Exception as e:
last_err = str(e)
continue
if success:
break
time.sleep(1) # Wait between attempts
for n in nets:
row = table.add_row()
row.add_column("TYPE", "NETWORK")
row.add_column("NAME", n.name)
row.add_column("ID", n.id)
row.add_column("ADDRESS", ", ".join(n.assigned_addresses))
row.add_column("STATUS", n.status)
row.add_column("DETAIL", "")
except Exception as exc:
row = table.add_row()
row.add_column("TYPE", "ERROR")
row.add_column("NAME", "ZeroTier CLI")
row.add_column("STATUS", "EXCEPTION")
row.add_column("DETAIL", str(exc))
if not success:
row.add_column("STATUS", "OFFLINE")
row.add_column("ADDRESS", "127.0.0.1:999")
row.add_column("DETAIL", f"Server not responding on port 999. Last attempt ({hosts[-1]}): {last_err}")
return table
except Exception as e:
row.add_column("STATUS", "OFFLINE")
row.add_column("ADDRESS", "127.0.0.1:999")
row.add_column("DETAIL", f"Status check failed: {e}")
# 2. Add Networks
if zt.is_available():
try:
networks = zt.list_networks()
for net in networks:
row = table.add_row()
row.add_column("TYPE", "NETWORK")
row.add_column("NAME", getattr(net, "name", "Unnamed"))
row.add_column("ID", getattr(net, "id", ""))
status = getattr(net, "status", "OK")
assigned = getattr(net, "assigned_addresses", [])
ip_str = assigned[0] if assigned else ""
row.add_column("STATUS", status)
row.add_column("ADDRESS", ip_str)
except Exception as e:
row = table.add_row()
row.add_column("TYPE", "ERROR")
row.add_column("DETAIL", f"Failed to list networks: {e}")
else:
row = table.add_row()
row.add_column("TYPE", "SYSTEM")
row.add_column("NAME", "ZeroTier")
row.add_column("STATUS", "NOT FOUND")
row.add_column("DETAIL", "zerotier-cli not in path")
# Output
try:
from cmdnat.out_table import TableOutput
TableOutput().render(table)
except Exception:
# Fallback for raw CLI
print(f"\n--- {table.title} ---")
for r in table.rows:
# Use the get_column method from ResultRow
t = r.get_column("TYPE") or ""
n = r.get_column("NAME") or ""
s = r.get_column("STATUS") or ""
a = r.get_column("ADDRESS") or ""
id = r.get_column("ID") or ""
d = r.get_column("DETAIL") or ""
print(f"[{t:7}] {n:15} | {s:15} | {a:20} | {id} | {d}")
print("-" * 100)
return 0
CMDLET = Cmdlet(
name=".zerotier",
summary="Check ZeroTier hosting and connection status",
summary="Check ZeroTier node and hosting status",
usage=".zerotier",
exec=exec
exec=exec_zerotier,
)
if __name__ == "__main__":
exec_zerotier(None, sys.argv[1:], {})