100 lines
3.1 KiB
Python
100 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Lädt eine .rsc-/Script-Datei per RouterOS REST API auf crs326-core hoch
|
|
und führt sie als /system/script aus.
|
|
|
|
Usage: python3 rest-core-script.py <pfad/zur/datei.rsc> [--host 10.40.41.211]
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
CLAUDe_JSON = Path.home() / ".claude.json"
|
|
|
|
if len(sys.argv) < 2:
|
|
print(__doc__, file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
src_path = Path(sys.argv[1]).resolve()
|
|
host = "10.40.41.211"
|
|
if len(sys.argv) > 2 and sys.argv[2] == "--host" and len(sys.argv) > 3:
|
|
host = sys.argv[3]
|
|
|
|
if not src_path.exists():
|
|
sys.exit(f"Datei nicht gefunden: {src_path}")
|
|
|
|
env = json.loads(CLAUDe_JSON.read_text())["mcpServers"]["mikromcp"]["env"]
|
|
user = env.get("CRS326_CORE_USER", "admin")
|
|
pw = env.get("CRS326_CORE_PASS")
|
|
if not pw:
|
|
sys.exit("CRS326_CORE_PASS fehlt in ~/.claude.json")
|
|
|
|
base_url = f"http://{host}/rest"
|
|
script_name = f"tp15_{src_path.stem}"
|
|
source = src_path.read_text()
|
|
|
|
|
|
def api(path, body=None, method="POST"):
|
|
url = base_url + path
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
req = urllib.request.Request(url, data=data, method=method)
|
|
req.add_header("Content-Type", "application/json")
|
|
import base64
|
|
req.add_header("Authorization", "Basic " + base64.b64encode(f"{user}:{pw}".encode()).decode())
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
return resp.status, resp.read().decode()
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, (e.read().decode() or e.reason)
|
|
|
|
|
|
def api_get(path):
|
|
url = base_url + path
|
|
req = urllib.request.Request(url)
|
|
import base64
|
|
req.add_header("Authorization", "Basic " + base64.b64encode(f"{user}:{pw}".encode()).decode())
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
return resp.status, resp.read().decode()
|
|
|
|
|
|
# [1] Altes Script mit gleichem Namen entfernen
|
|
print(f"Suche ggf. altes Script '{script_name}' ...")
|
|
status, body = api_get("/system/script")
|
|
if status == 200:
|
|
try:
|
|
for item in json.loads(body):
|
|
if item.get("name") == script_name:
|
|
print(f"Entferne altes Script id={item.get('.id')} ...")
|
|
api("/system/script/remove", {"id": item[".id"]})
|
|
break
|
|
except Exception:
|
|
pass
|
|
|
|
# [2] Script anlegen
|
|
print(f"Erstelle Script '{script_name}' ({len(source)} Zeichen) ...")
|
|
status, body = api("/system/script/add", {"name": script_name, "source": source})
|
|
print(f"add -> HTTP {status}: {body[:400]}")
|
|
if status >= 400:
|
|
sys.exit("Script konnte nicht angelegt werden.")
|
|
|
|
try:
|
|
created = json.loads(body)
|
|
except json.JSONDecodeError:
|
|
created = {}
|
|
|
|
script_id = created.get(".id")
|
|
|
|
# [3] Script ausführen
|
|
if script_id:
|
|
print(f"Starte Script id={script_id} ...")
|
|
status, body = api("/system/script/run", {"id": script_id})
|
|
print(f"run -> HTTP {status}: {body[:2000]}")
|
|
else:
|
|
# Try to find by name
|
|
print("Suche Script-ID per Name ...")
|
|
status, body = api_get(f"/system/script?name={script_name}")
|
|
print(f"getall -> HTTP {status}: {body[:800]}")
|
|
|
|
print("OK: REST-Aufrufe abgeschlossen")
|