feat 🔌 shelly-mcp set_auth-Tool + Digest-Auth-Fallback (TP17 Geräte-PW)

- set_auth: setzt einheitliches Admin-PW (SHELLY_DEVICE_PW) via Shelly.SetAuth, realm=Geräte-ID, ha1=sha256("admin:realm:pw") pro Gerät; guarded (confirm/dry-run). - _rpc: HTTP-Digest-Fallback (user admin) bei 401 → Geräte bleiben nach set_auth managebar. - .env.example/README: SHELLY_DEVICE_PW dokumentiert (getrennt vom WLAN-PSK); Rollout mit PW-Schritt. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-11 07:17:16 +02:00
parent dcdf299568
commit 794082fd74
3 changed files with 69 additions and 10 deletions
--- 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
--- 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`.
--- 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:<realm>:<password>"). 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).