120 lines
4.6 KiB
Python
120 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Export ap1-Daten, die im crs326-core Cutover-Fenster gebraucht werden.
|
|
|
|
Ausgaben (ohne Secrets im stdout):
|
|
- routeros/ap1-leases-export.json : alle DHCP-Leases (static + dynamic)
|
|
- routeros/ap1-wg-hub.privkey : wg-hub private-key (nur diese Zeile)
|
|
- routeros/ap1-cutover-summary.txt : menschliche Zusammenfassung
|
|
|
|
Quelle: ap1 via REST (10.40.10.1), Creds aus ~/.claude.json (mikromcp-env).
|
|
"""
|
|
import base64
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import sys
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
AP1_REST = "http://10.40.10.1/rest" # VLAN10 Mgmt-IP (ap1-Mgmt nur via VLAN10)
|
|
CLAUDE_JSON = pathlib.Path.home() / ".claude.json"
|
|
OUT_DIR = pathlib.Path(__file__).resolve().parent
|
|
|
|
|
|
def load_ap1_creds():
|
|
cfg = json.loads(CLAUDE_JSON.read_text())
|
|
env = cfg.get("mcpServers", {}).get("mikromcp", {}).get("env", {})
|
|
user = env.get("ROUTER_AP1_USER") or env.get("ROUTER_USER")
|
|
pw = env.get("ROUTER_AP1_PASS") or env.get("ROUTER_PASS")
|
|
if not user or not pw:
|
|
raise SystemExit("ROUTER_AP1_USER/PASS nicht in ~/.claude.json mikromcp.env gefunden")
|
|
return user, pw
|
|
|
|
|
|
def rest_get(path, user, pw):
|
|
url = f"{AP1_REST}{path}"
|
|
auth = base64.b64encode(f"{user}:{pw}".encode()).decode()
|
|
req = urllib.request.Request(url, headers={
|
|
"Authorization": f"Basic {auth}",
|
|
"Accept": "application/json",
|
|
})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=20) as resp:
|
|
raw = resp.read()
|
|
return json.loads(raw) if raw else []
|
|
except urllib.error.HTTPError as e:
|
|
sys.stderr.write(f"!! GET {url} -> {e.code} {e.read().decode()[:500]}\n")
|
|
raise
|
|
|
|
|
|
def main():
|
|
user, pw = load_ap1_creds()
|
|
print(f"ap1 REST {AP1_REST} (User: {user})")
|
|
|
|
# 1) DHCP-Leases
|
|
leases = rest_get("/ip/dhcp-server/lease", user, pw)
|
|
leases_path = OUT_DIR / "ap1-leases-export.json"
|
|
leases_path.write_text(json.dumps(leases, indent=2))
|
|
print(f" DHCP-Leases: {len(leases)} -> {leases_path}")
|
|
|
|
# 2) WG-Hub privkey (field may be redacted; fallback to script-run)
|
|
wg = rest_get("/interface/wireguard", user, pw)
|
|
hub = next((i for i in wg if i.get("name") == "wg-hub"), None)
|
|
privkey = ""
|
|
if hub:
|
|
privkey = hub.get("private-key") or ""
|
|
if not privkey:
|
|
# REST listet privkey manchmal nicht; über system/script ausgeben lassen
|
|
script = ':put [/interface/wireguard/get [find name="wg-hub"] private-key]'
|
|
try:
|
|
run_resp = rest_post("/system/script/run", {"source": script}, user, pw)
|
|
# output format variiert
|
|
privkey = str(run_resp).strip().strip('"')
|
|
except Exception as e:
|
|
sys.stderr.write(f" WG-Privkey-Fallback fehlgeschlagen: {e}\n")
|
|
wg_path = OUT_DIR / "ap1-wg-hub.privkey"
|
|
if privkey and privkey != "<<<none>>>":
|
|
wg_path.write_text(privkey + "\n")
|
|
print(f" WG-Hub Privkey -> {wg_path} ({len(privkey)} chars)")
|
|
else:
|
|
print(f" WARN: WG-Hub Privkey nicht ermittelt -> {wg_path} nicht geschrieben")
|
|
|
|
# 3) Zusammenfassung
|
|
summary = [
|
|
f"ap1 Cutover-Export vom {os.popen('date -Iseconds').read().strip()}",
|
|
f"DHCP-Leases gesamt: {len(leases)}",
|
|
f" statisch (dynamic=no): {sum(1 for l in leases if str(l.get('dynamic')).lower() == 'false')}",
|
|
f" dynamisch: {sum(1 for l in leases if str(l.get('dynamic')).lower() != 'false')}",
|
|
f"WG-Hub private-key erfasst: {'ja' if privkey else 'nein'}",
|
|
"",
|
|
"Nächste Schritte im Cutover-Fenster:",
|
|
"1. ap1-leases-export.json filtern: nur dynamic=no, server-Namen auf core dhcp-v10..v99 mappen.",
|
|
"2. ap1-wg-hub.privkey in core-Config einsetzen (ersetzt <<< AP1_WG_HUB_PRIVKEY >>>).",
|
|
"3. Generate Intended Config für crs326-core erneut laufen lassen -> Nautobot-Deployment.",
|
|
]
|
|
summary_path = OUT_DIR / "ap1-cutover-summary.txt"
|
|
summary_path.write_text("\n".join(summary) + "\n")
|
|
print(f" Summary -> {summary_path}")
|
|
|
|
|
|
def rest_post(path, data, user, pw):
|
|
url = f"{AP1_REST}{path}"
|
|
auth = base64.b64encode(f"{user}:{pw}".encode()).decode()
|
|
body = json.dumps(data).encode()
|
|
req = urllib.request.Request(url, data=body, headers={
|
|
"Authorization": f"Basic {auth}",
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
}, method="POST")
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=20) as resp:
|
|
raw = resp.read()
|
|
return json.loads(raw) if raw else {}
|
|
except urllib.error.HTTPError as e:
|
|
sys.stderr.write(f"!! POST {url} -> {e.code} {e.read().decode()[:500]}\n")
|
|
raise
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|