diff --git a/mcp/shelly/.env.example b/mcp/shelly/.env.example index 33fb7f1..13ce89b 100644 --- a/mcp/shelly/.env.example +++ b/mcp/shelly/.env.example @@ -1,6 +1,7 @@ # shelly-mcp config — copy to .env (gitignored) and fill in. SHELLY_IOT_SSID=dirtyvader-iot SHELLY_IOT_PSK= # IOT SSID passphrase (from ap1 w2-iot). Required for set_sta. +SHELLY_DEVICE_PW= # uniform Shelly admin password (Shelly.SetAuth). Required for set_auth. Separate from the WLAN PSK. SHELLY_SWEEP=10.40.41.180-250 # discovery range (VLAN1 today) SHELLY_MCP_HOST=127.0.0.1 SHELLY_MCP_PORT=3030 diff --git a/mcp/shelly/README.md b/mcp/shelly/README.md index ead67a5..2324264 100644 --- a/mcp/shelly/README.md +++ b/mcp/shelly/README.md @@ -25,11 +25,16 @@ Immer erst `WiFi.GetStatus`/`Sys.GetStatus` gegenchecken, Pilot vor Batch. ## Dateien - `discover.py` — read-only Sweep, identifiziert Shellys per `/shelly`, matcht MAC gegen `inventory/bf.csv`, meldet DHCP-Drift und Inventar-Lücken. Basis des MCP-`discover`-Tools. -- `server.py` — (in Arbeit) MCP-Server: `discover` / `get` / `set_sta` (Migration, guarded) / `verify`. +- `server.py` — MCP-Server: `discover` / `get` / `set_sta` (WLAN→VLAN30, guarded) / + `set_auth` (einheitliches Admin-PW `SHELLY_DEVICE_PW`, guarded, Digest-Auth-Fallback) / + `verify` / `reservation_plan`. Go-Runbook: `docs/tp17-shelly-go-runbook.md`. ## Rollout-Reihenfolge (User-Entscheid: Pilot→Rest, Reservierungen je MAC) -1. VLAN30-DHCP-Reservierungen je MAC anlegen (MikroMCP `manage_dhcp_lease`). -2. **Pilot** 2–3 unkritische Always-on (z.B. `bad_luefter` .233 / `technik_pve` .222) umklemmen. -3. VLAN30-Reachability + HA/ioBroker-Wiederanbindung verifizieren. -4. Rest in Batches; `weihnachten_*` beim nächsten Einstecken. +1. ✅ VLAN30-DHCP-Reservierungen je MAC (liegen; s6-MAC 2026-07-11 gefixt). +2. **Pilot** unkritischer Always-on (`set_sta`) → **VLAN30-Verify** → **`set_auth`** (Admin-PW) → HA-Credential. +3. VLAN30-Reachability + HA/ioBroker-Wiederanbindung verifizieren (Entities bleiben unverändert). +4. Rest in Batches; saisonale (`weihnachten_*`, Twinkly, Creality) beim nächsten Einstecken. 5. cudy-Alt-WLAN abbauen; DNS/Inventar/`docs/ip-plan.md` nachziehen. + +Status/Detail: **6/26 schon auf VLAN30**, 16 morgen. Volle Runbook-Tabelle + Netz-Abgleich in +`docs/tp17-shelly-go-runbook.md`. diff --git a/mcp/shelly/server.py b/mcp/shelly/server.py index 0230ef9..78c30d6 100644 --- a/mcp/shelly/server.py +++ b/mcp/shelly/server.py @@ -14,9 +14,11 @@ Config via env (see .env.example): Run: python3 server.py (loads ./.env if present) """ +import hashlib import json import os import sys +import urllib.error import urllib.request # reuse the read-only recon logic @@ -38,6 +40,7 @@ if os.path.exists(_envfile): IOT_SSID = os.environ.get("SHELLY_IOT_SSID", "dirtyvader-iot") IOT_PSK = os.environ.get("SHELLY_IOT_PSK", "") +DEVICE_PW = os.environ.get("SHELLY_DEVICE_PW", "") # uniform admin password (Shelly.SetAuth) SWEEP = os.environ.get("SHELLY_SWEEP", "10.40.41.180-250") HOST = os.environ.get("SHELLY_MCP_HOST", "127.0.0.1") PORT = int(os.environ.get("SHELLY_MCP_PORT", "3030")) @@ -50,12 +53,27 @@ mcp = FastMCP("shelly-mcp", host=HOST, port=PORT, stateless_http=True, # ---------------- helpers ---------------- def _rpc(ip: str, method: str, params: dict | None = None, timeout: float = 6.0) -> dict: - """Gen2+ Shelly RPC over HTTP POST /rpc.""" + """Gen2+ Shelly RPC over HTTP POST /rpc. + + Falls back to HTTP digest auth (user 'admin' / SHELLY_DEVICE_PW) if the + device requires authentication — so devices stay manageable after set_auth. + """ body = json.dumps({"id": 1, "method": method, "params": params or {}}).encode() - req = urllib.request.Request(f"http://{ip}/rpc", data=body, - headers={"Content-Type": "application/json"}) - with urllib.request.urlopen(req, timeout=timeout) as r: - return json.load(r) + + def _mk(): + return urllib.request.Request(f"http://{ip}/rpc", data=body, + headers={"Content-Type": "application/json"}) + try: + with urllib.request.urlopen(_mk(), timeout=timeout) as r: + return json.load(r) + except urllib.error.HTTPError as e: + if e.code == 401 and DEVICE_PW: + mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm() + mgr.add_password(None, f"http://{ip}/rpc", "admin", DEVICE_PW) + opener = urllib.request.build_opener(urllib.request.HTTPDigestAuthHandler(mgr)) + with opener.open(_mk(), timeout=timeout) as r: + return json.load(r) + raise def _sweep(spec: str | None = None) -> list[dict]: @@ -138,6 +156,41 @@ def set_sta(ip: str, confirm: bool = False, ssid: str | None = None) -> dict: return {**plan, "ok": False, "error": f"{type(e).__name__}: {e}"} +@mcp.tool() +def set_auth(ip: str, confirm: bool = False) -> dict: + """MIGRATION ACTION — set the uniform admin password (SHELLY_DEVICE_PW) on a Shelly (Gen2+). + + Enables HTTP digest auth: user 'admin', realm = the device id (per-device), + ha1 = sha256("admin::"). Same cleartext password across the + fleet, different ha1 per device. After this, _rpc authenticates automatically. + + Run this AFTER the device is verified on VLAN30 (last step per device). + - confirm MUST be true to write; SHELLY_DEVICE_PW must be set. + - confirm=false is a dry-run: returns the computed realm (no write). + """ + if not DEVICE_PW: + return {"ip": ip, "ok": False, "error": "SHELLY_DEVICE_PW not set in environment"} + try: + info = _rpc(ip, "Shelly.GetDeviceInfo").get("result") or {} + except Exception as e: + return {"ip": ip, "ok": False, "error": f"{type(e).__name__}: {e}"} + realm = info.get("id") + if not realm: + return {"ip": ip, "ok": False, "error": "could not read device id (realm)"} + ha1 = hashlib.sha256(f"admin:{realm}:{DEVICE_PW}".encode()).hexdigest() + if not confirm: + return {"ip": ip, "ok": False, "dry_run": True, "realm": realm, + "would_send": {"method": "Shelly.SetAuth", + "params": {"user": "admin", "realm": realm, + "ha1": ha1[:12] + "…"}}} + try: + res = _rpc(ip, "Shelly.SetAuth", {"user": "admin", "realm": realm, "ha1": ha1}) + return {"ip": ip, "ok": True, "realm": realm, "set_result": res, + "note": "admin auth enabled; future RPC uses digest via SHELLY_DEVICE_PW"} + except Exception as e: + return {"ip": ip, "ok": False, "error": f"{type(e).__name__}: {e}"} + + @mcp.tool() def verify_on_iot(ip: str) -> dict: """Verify a device now sits on the IOT SSID / VLAN30 (post-migration check).