admin/bf-infra
1
0
Fork 0
Code Issues 21 Pull Requests Actions Packages Projects Releases Wiki Activity

Initial import from bf-network-r6.opencode — opencode migration

Browse Source
This commit is contained in:
Christoph Emsenhuber
2026-08-26 12:44:21 +02:00
commit 5ee0716dda
297 changed files with 72324 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
mcp/shelly/.env.example Normal file
View File

@@ -0,0 +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

40
mcp/shelly/README.md Normal file
View File

@@ -0,0 +1,40 @@
# shelly-mcp — Shelly-Verwaltung & VLAN1→VLAN30-Migration (TP17)
Tooling, um die ~26 Shellys von der alten cudy-WLAN/VLAN1 (`10.40.41.2xx`) auf die
IOT-SSID **`dirtyvader-iot`** (VLAN30, `dhcp-v30`/pool-v30) umzuklemmen und danach
dauerhaft zu verwalten. Steuerung/Status: `docs/teilprojekte.md` → **TP17**.
## Ist-Stand (Recon 2026-07-11)
- **16 live**, alle **Gen3/Gen4** → uniforme RPC-API (`WiFi.SetConfig`), `auth=off`.
Kein Gen1-Sonderfall.
- **10 nicht erreichbar:** `weihnachten_*` (s1/s2/s3, saisonal ausgesteckt) +
s12/s14/s16/s18/s22/s23 (aus / DHCP-drift). s24 hat keine MAC im Inventar.
- Recon-Snapshot: `python3 discover.py 10.40.41.180-250 --json`.
## Migrationsmechanismus (per Gerät)
Gen2+ RPC (HTTP POST `http://<ip>/rpc`):
```
WiFi.SetConfig {"config":{"sta":{"ssid":"dirtyvader-iot","pass":"<PSK>","enable":true}}}
```
→ `Shelly.Reboot` (oder greift nach kurzer Zeit selbst) → Gerät verschwindet aus VLAN1,
kommt per `dhcp-v30` auf `10.40.30.x` (mit MAC-Reservierung = stabile IP).
⚠️ **Falsches PSK → Gerät geht in AP-Mode (SSID `Shelly*`) → physischer Reset nötig.**
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` — 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 (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`.

124
mcp/shelly/discover.py Normal file
View File

@@ -0,0 +1,124 @@
#!/usr/bin/env python3
"""Shelly recon: sweep an IPv4 range for Shelly devices and identify them.
Read-only. Hits each host's unauthenticated `/shelly` endpoint (works on Gen1
and Gen2+), collects gen/model/fw/mac, and cross-references the project
inventory (inventory/bf.csv) by MAC so DHCP-drifted devices are still matched
to their logical name.
Usage:
python3 discover.py # default sweep 10.40.41.200-240
python3 discover.py 10.40.41.180-250
python3 discover.py --json # machine-readable to stdout
"""
import csv, json, os, sys, urllib.request, socket
from concurrent.futures import ThreadPoolExecutor
REPO = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
INVENTORY = os.path.join(REPO, "inventory", "bf.csv")
def load_inventory():
"""MAC(lower, colon) -> {name, ip, section} for shelly rows."""
by_mac = {}
if not os.path.exists(INVENTORY):
return by_mac
with open(INVENTORY) as f:
for row in csv.DictReader(f):
host = (row.get("hostname") or "")
if "shelly" not in host.lower():
continue
mac = (row.get("mac") or "").strip().lower()
entry = {"name": (row.get("note") or host).strip(),
"shelly_id": host.replace("shelly:", "").strip(),
"inv_ip": (row.get("ip") or "").strip(),
"mac": mac}
if mac:
by_mac[mac] = entry
return by_mac
def probe(ip, timeout=3.0):
"""Return identity dict for a Shelly at ip, or None."""
try:
req = urllib.request.Request(f"http://{ip}/shelly")
with urllib.request.urlopen(req, timeout=timeout) as r:
d = json.load(r)
except Exception:
return None
gen = d.get("gen", 1)
mac = (d.get("mac") or "").lower()
# normalise MAC to colon form for inventory matching
if mac and ":" not in mac and len(mac) == 12:
mac = ":".join(mac[i:i+2] for i in range(0, 12, 2))
return {
"ip": ip,
"gen": gen,
"model": d.get("model") or d.get("type") or "?",
"fw": d.get("fw") or d.get("ver") or "?",
"mac": mac,
"auth": d.get("auth", d.get("auth_en", None)),
"name_dev": d.get("name"),
}
def expand(spec):
"""'10.40.41.200-240' or '10.40.41.208' -> list of IPs."""
if "-" in spec.split(".")[-1]:
base, rng = spec.rsplit(".", 1)
lo, hi = rng.split("-")
return [f"{base}.{i}" for i in range(int(lo), int(hi) + 1)]
return [spec]
def main():
args = [a for a in sys.argv[1:] if not a.startswith("--")]
as_json = "--json" in sys.argv
spec = args[0] if args else "10.40.41.200-240"
ips = expand(spec)
inv = load_inventory()
found = []
with ThreadPoolExecutor(max_workers=32) as ex:
for res in ex.map(probe, ips):
if res:
found.append(res)
# match to inventory
seen_macs = set()
for f in found:
m = inv.get(f["mac"])
f["inv_name"] = m["name"] if m else None
f["inv_id"] = m["shelly_id"] if m else None
f["inv_ip"] = m["inv_ip"] if m else None
f["ip_drifted"] = bool(m and m["inv_ip"] and m["inv_ip"] != f["ip"])
if m:
seen_macs.add(f["mac"])
missing = [{"mac": mac, **e} for mac, e in inv.items() if mac not in seen_macs]
if as_json:
print(json.dumps({"found": found, "missing_from_inventory": missing,
"swept": spec, "reachable": len(found),
"inventory_total": len(inv)}, indent=2))
return
found.sort(key=lambda x: x["inv_id"] or "zzz")
print(f"Sweep {spec}: {len(found)} Shellys reachable "
f"(inventory has {len(inv)} with MAC)\n")
print(f"{'shelly':7} {'ip':15} {'gen':4} {'model':18} {'fw':10} {'name':26} drift")
print("-" * 100)
for f in found:
drift = f"was {f['inv_ip']}" if f["ip_drifted"] else ""
print(f"{(f['inv_id'] or '?'):7} {f['ip']:15} g{f['gen']:<3} "
f"{str(f['model']):18} {str(f['fw']):10} "
f"{str(f['inv_name'] or f.get('name_dev') or '?'):26} {drift}")
if missing:
print(f"\n{len(missing)} in inventory but NOT reachable in sweep "
f"(offline / seasonal / outside range):")
for m in sorted(missing, key=lambda x: x["shelly_id"]):
print(f" {m['shelly_id']:5} {m['inv_ip']:15} {m['mac']:18} {m['name']}")
if __name__ == "__main__":
main()

321
mcp/shelly/migrate.py Normal file
View File

@@ -0,0 +1,321 @@
#!/usr/bin/env python3
"""TP17 migration runner — anchored VLAN1→VLAN30 cutover for Shellys.
Per-device sequence (Gen2+ RPC), ordered so recovery is ALWAYS possible:
1. probe old IP — must be reachable; record current sta for the log
2. ANCHOR: ap.enable=true — device now also broadcasts its own AP
(Shelly*-<mac>, open) as an emergency anchor.
AP+STA run concurrently, no reboot, sta untouched.
3. set sta -> dirtyvader-iot / VLAN30 static 10.40.30.<octet>
4. Shelly.Reboot
5. poll new IP (10.40.30.<octet>) until it answers, or time out
-> success: AP stays on until end-of-run `deanchor`
-> fail: device is still reachable via its own AP (step 2) — no reset
Only AFTER step 2 do we overwrite sta, so the working fallback that vanishes on
reboot (old sta) is replaced by a fallback that does NOT depend on any WLAN infra
(the device's own AP). That's the whole point of the anchor.
Reuses discover.py (probe/expand/load_inventory) and the same .env as server.py.
Commands:
python3 migrate.py show <ip> read WiFi.GetConfig (read-only)
python3 migrate.py migrate <ip...> [--go] anchor+sta+reboot+verify (dry-run w/o --go)
python3 migrate.py deanchor <ip...> [--go] turn the AP anchor back off (.30.x IPs)
python3 migrate.py auth <ip...> [--go] set uniform admin password (last step)
python3 migrate.py eco <ip...> [--go] turn device.eco_mode OFF (--on to re-enable)
python3 migrate.py table final tabular overview (old/new/model/fw)
<ip...> accepts single IPs, ranges (10.40.41.205-233) or 'all' (the default sweep).
Batches run sequentially and STOP on the first failed verify (safety).
"""
import hashlib
import json
import os
import sys
import time
import urllib.error
import urllib.request
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import discover # noqa: E402
# ---- shared .env loader (same as server.py) ----
_envfile = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")
if os.path.exists(_envfile):
for _line in open(_envfile):
_line = _line.strip()
if _line and not _line.startswith("#") and "=" in _line:
_k, _v = _line.split("=", 1)
os.environ.setdefault(_k.strip(), _v.strip().strip('"').strip("'"))
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", "")
V30_NET, V30_MASK, V30_GW = "10.40.30", "255.255.255.0", "10.40.30.1"
LOG = os.path.join(os.path.dirname(os.path.abspath(__file__)), "migration-log.jsonl")
def _rpc(ip, method, params=None, timeout=8.0):
body = json.dumps({"id": 1, "method": method, "params": params or {}}).encode()
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 _octet(ip):
return ip.rsplit(".", 1)[-1]
def _name_for(mac):
m = discover.load_inventory().get((mac or "").lower())
return m["name"] if m else None
def _log(rec):
with open(LOG, "a") as f:
f.write(json.dumps(rec) + "\n")
def _resolve(specs):
"""Expand CLI ip args into a concrete IP list. 'all' -> default VLAN1 sweep."""
if not specs or specs == ["all"]:
return discover.expand(os.environ.get("SHELLY_SWEEP", "10.40.41.180-250"))
ips = []
for s in specs:
ips.extend(discover.expand(s))
return ips
# ---------------- commands ----------------
def cmd_show(ip):
print(json.dumps(_rpc(ip, "WiFi.GetConfig"), indent=2))
def _anchor(ip):
"""Enable the device's own AP as an emergency anchor. Live, no reboot.
This firmware refuses an OPEN AP (`-103 'pass' should not be empty`), so the
anchor is WPA2 secured with the IOT PSK — a value the operator already has,
and it's removed again by `deanchor`. AP+STA run concurrently; sta untouched.
"""
res = _rpc(ip, "WiFi.SetConfig",
{"config": {"ap": {"enable": True, "is_open": False, "pass": IOT_PSK}}})
if res.get("error"):
return {"enabled": False, "error": res["error"]}
ap = _rpc(ip, "WiFi.GetConfig").get("result", {}).get("ap", {})
return {"enabled": bool(ap.get("enable")), "ap_ssid": ap.get("ssid"),
"secured": not ap.get("is_open")}
def _migrate_one(ip, go, poll=90):
ident = discover.probe(ip)
if not ident:
return {"ip": ip, "ok": False, "error": "not reachable on old IP"}
if not IOT_PSK:
return {"ip": ip, "ok": False, "error": "SHELLY_IOT_PSK not set"}
cur = _rpc(ip, "WiFi.GetConfig").get("result", {}).get("sta", {})
new_ip = f"{V30_NET}.{_octet(ip)}"
name = _name_for(ident["mac"]) or ident.get("name_dev") or "?"
sta = {"ssid": IOT_SSID, "pass": IOT_PSK, "enable": True, "ipv4mode": "static",
"ip": new_ip, "netmask": V30_MASK, "gw": V30_GW, "nameserver": V30_GW}
plan = {"ip": ip, "name": name, "mac": ident["mac"], "model": ident["model"],
"fw": ident["fw"], "old_ssid": cur.get("ssid"), "new_ip": new_ip,
"new_ssid": IOT_SSID}
if not go:
return {**plan, "ok": False, "dry_run": True,
"would": "anchor AP -> set sta(static .30) -> reboot -> verify"}
# 2. anchor
anc = _anchor(ip)
if not anc["enabled"]:
return {**plan, "ok": False, "error": "AP anchor did not enable — aborting before sta write"}
print(f" anchor up: AP '{anc['ap_ssid']}' (WPA2, PSK=IOT-PSK)")
# 3+4. set sta, reboot
_rpc(ip, "WiFi.SetConfig", {"config": {"sta": sta}})
_rpc(ip, "Shelly.Reboot")
print(f" sta set -> {new_ip}, rebooting; polling…")
# 5. poll new IP
t0 = time.time()
while time.time() - t0 < poll:
time.sleep(5)
got = discover.probe(new_ip)
if got and got["mac"].lower() == ident["mac"].lower():
st = _rpc(new_ip, "WiFi.GetStatus").get("result", {})
rec = {**plan, "ok": True, "verified_ssid": st.get("ssid"),
"rssi": st.get("rssi"), "secs": round(time.time() - t0)}
_log(rec)
return rec
return {**plan, "ok": False,
"error": f"did not appear on {new_ip} within {poll}s — "
f"reach it via its AP '{anc['ap_ssid']}' (no reset needed)"}
def cmd_migrate(specs, go):
ips = _resolve(specs)
print(f"{'GO' if go else 'DRY-RUN'} migrate {len(ips)} device(s) "
f"-> {IOT_SSID}/VLAN30 (PSK {'set' if IOT_PSK else 'MISSING'})\n")
ok = 0
for ip in ips:
if not discover.probe(ip):
continue # skip empty slots in a range
print(f"[{ip}] …")
r = _migrate_one(ip, go)
tag = "OK " if r.get("ok") else ("dry" if r.get("dry_run") else "FAIL")
print(f" {tag}: {r.get('name','?')} {ip} -> {r.get('new_ip','?')} "
f"{r.get('error') or r.get('would') or ('rssi '+str(r.get('rssi')))}")
if r.get("ok"):
ok += 1
elif go and not r.get("dry_run"):
print(" >> STOP: batch halted on failure (device recoverable via its AP).")
break
print(f"\n{ok} migrated OK.")
def cmd_deanchor(specs, go):
ips = _resolve(specs) if specs != ["all"] else discover.expand("10.40.30.180-250")
for ip in ips:
if not discover.probe(ip):
continue
if not go:
print(f"dry: would disable AP on {ip}")
continue
_rpc(ip, "WiFi.SetConfig", {"config": {"ap": {"enable": False}}})
cfg = _rpc(ip, "WiFi.GetConfig").get("result", {}).get("ap", {})
print(f"{ip}: AP enable -> {cfg.get('enable')}")
def cmd_auth(specs, go):
if not DEVICE_PW:
print("SHELLY_DEVICE_PW not set — aborting.")
return
for ip in _resolve(specs):
if not discover.probe(ip):
continue
info = _rpc(ip, "Shelly.GetDeviceInfo").get("result", {})
realm = info.get("id")
ha1 = hashlib.sha256(f"admin:{realm}:{DEVICE_PW}".encode()).hexdigest()
if not go:
print(f"dry: {ip} realm={realm} ha1={ha1[:12]}…")
continue
_rpc(ip, "Shelly.SetAuth", {"user": "admin", "realm": realm, "ha1": ha1})
print(f"{ip}: auth set (realm {realm})")
def cmd_eco(specs, go, want=False):
"""Set device.eco_mode across the fleet (TP17.3, 2026-08-23).
eco_mode throttles the CPU clock and TX power. It was applied in TP17 by an
RSSI rule and turned out to cause real misbehaviour: the s23 phase-cut dimmer
flickered, and s26's `momentary` input swallowed the first edge. Both were
fixed by switching it off, so the fleet decision is: OFF everywhere.
No reboot needed — the device answers `restart_required: false` to this write.
Devices that don't support it report eco_mode=None and are skipped (s24).
"""
# Weak links (s20 sits at -85 dBm) time out intermittently. A single timeout
# must not abort the sweep, so every call is retried and failures are
# collected and reported at the end instead of raising.
def _try(ip, method, params=None, tries=3):
for n in range(tries):
try:
return _rpc(ip, method, params, timeout=10.0)
except Exception as e:
last = e
raise last
failed = []
for ip in _resolve(specs):
ident = discover.probe(ip)
if not ident:
continue
name = _name_for(ident["mac"]) or ident.get("name_dev") or "?"
try:
cur = _try(ip, "Sys.GetConfig").get("result", {}).get("device", {}).get("eco_mode")
if cur is None:
print(f" {ip:15} {name:26} eco not supported — skipped")
continue
if cur == want:
print(f" {ip:15} {name:26} already {want}")
continue
if not go:
print(f" dry: {ip:11} {name:26} eco {cur} -> {want}")
continue
res = _try(ip, "Sys.SetConfig", {"config": {"device": {"eco_mode": want}}})
new = _try(ip, "Sys.GetConfig").get("result", {}).get("device", {}).get("eco_mode")
rr = (res.get("result") or {}).get("restart_required")
ok = "OK " if new == want else "MISMATCH"
print(f" {ip:15} {name:26} eco {cur} -> {new} {ok}"
+ (" (restart_required was already set before this write)" if rr else ""))
if new != want:
failed.append((ip, name, f"read back {new}"))
except Exception as e:
print(f" {ip:15} {name:26} FAILED: {type(e).__name__}")
failed.append((ip, name, type(e).__name__))
if failed:
print(f"\n{len(failed)} device(s) need a retry:")
for ip, name, why in failed:
print(f" {ip:15} {name:26} {why}")
def cmd_table():
inv = discover.load_inventory()
ips = discover.expand("10.40.30.180-250")
from concurrent.futures import ThreadPoolExecutor
rows = []
with ThreadPoolExecutor(max_workers=32) as ex:
for r in ex.map(discover.probe, ips):
if r:
m = inv.get(r["mac"])
rows.append({"id": (m["shelly_id"] if m else "?"),
"name": (m["name"] if m else r.get("name_dev") or "?"),
"old": f"10.40.41.{_octet(r['ip'])}", "new": r["ip"],
"gen": r["gen"], "model": r["model"], "fw": r["fw"]})
rows.sort(key=lambda x: x["id"])
print(f"\n{'id':5} {'name':26} {'old IP':15} {'new IP':15} {'gen':4} {'model':18} fw")
print("-" * 108)
for x in rows:
print(f"{x['id']:5} {x['name']:26} {x['old']:15} {x['new']:15} "
f"g{x['gen']:<3} {x['model']:18} {x['fw']}")
print(f"\n{len(rows)} Shellys on VLAN30.")
def main():
a = sys.argv[1:]
if not a:
print(__doc__)
return
cmd, rest = a[0], a[1:]
go = "--go" in rest
specs = [x for x in rest if not x.startswith("--")]
if cmd == "show":
cmd_show(specs[0])
elif cmd == "migrate":
cmd_migrate(specs, go)
elif cmd == "deanchor":
cmd_deanchor(specs or ["all"], go)
elif cmd == "auth":
cmd_auth(specs, go)
elif cmd == "eco":
cmd_eco(specs, go, want=("--on" in rest))
elif cmd == "table":
cmd_table()
else:
print(f"unknown command: {cmd}\n{__doc__}")
if __name__ == "__main__":
main()

16
mcp/shelly/migration-log.jsonl Normal file
View File

@@ -0,0 +1,16 @@
{"ip": "10.40.41.233", "name": "bad_luefter", "mac": "98:a3:16:7a:a3:54", "model": "S4SW-002P16EU", "fw": "1.7.5", "old_ssid": "elektrosmog", "new_ip": "10.40.30.233", "new_ssid": "dirtyvader-iot", "ok": true, "verified_ssid": "dirtyvader-iot", "rssi": -65, "secs": 8}
{"ip": "10.40.41.211", "name": "kueche_mikrowelle", "mac": "e4:b0:63:e4:0f:30", "model": "S3PL-00112EU", "fw": "1.7.5", "old_ssid": "elektrosmog.c", "new_ip": "10.40.30.211", "new_ssid": "dirtyvader-iot", "ok": true, "verified_ssid": "dirtyvader-iot", "rssi": -93, "secs": 7}
{"ip": "10.40.41.212", "name": "kueche_backofen", "mac": "d8:85:ac:0c:5c:c8", "model": "S3PL-30116EU", "fw": "1.7.99-plugpmg3prod1", "old_ssid": "elektrosmog.c", "new_ip": "10.40.30.212", "new_ssid": "dirtyvader-iot", "ok": true, "verified_ssid": "dirtyvader-iot", "rssi": -68, "secs": 7}
{"ip": "10.40.41.213", "name": "kueche_kuehlschrank", "mac": "d0:cf:13:d8:18:fc", "model": "S3PL-00112EU", "fw": "1.7.5", "old_ssid": "elektrosmog.c", "new_ip": "10.40.30.213", "new_ssid": "dirtyvader-iot", "ok": true, "verified_ssid": "dirtyvader-iot", "rssi": -67, "secs": 7}
{"ip": "10.40.41.214", "name": "kueche_tiefkuehlschrank", "mac": "d0:cf:13:d9:94:e8", "model": "S3PL-00112EU", "fw": "1.7.5", "old_ssid": "elektrosmog.c", "new_ip": "10.40.30.214", "new_ssid": "dirtyvader-iot", "ok": true, "verified_ssid": "dirtyvader-iot", "rssi": -85, "secs": 7}
{"ip": "10.40.41.215", "name": "kueche_geschirrspueler", "mac": "e4:b0:63:fa:fd:e0", "model": "S3PL-00112EU", "fw": "1.7.5", "old_ssid": "elektrosmog.c", "new_ip": "10.40.30.215", "new_ssid": "dirtyvader-iot", "ok": true, "verified_ssid": "dirtyvader-iot", "rssi": -71, "secs": 7}
{"ip": "10.40.41.216", "name": "wohnzimmer_sideboard", "mac": "e8:f6:0a:61:88:08", "model": "S4PL-00416EU", "fw": "1.7.99-powerstripg4prod1", "old_ssid": "elektrosmog.c", "new_ip": "10.40.30.216", "new_ssid": "dirtyvader-iot", "ok": true, "verified_ssid": "dirtyvader-iot", "rssi": -66, "secs": 13}
{"ip": "10.40.41.217", "name": "bad_sonos", "mac": "e4:b0:63:fc:57:e4", "model": "S3PL-00112EU", "fw": "1.7.5", "old_ssid": "elektrosmog.c", "new_ip": "10.40.30.217", "new_ssid": "dirtyvader-iot", "ok": true, "verified_ssid": "dirtyvader-iot", "rssi": -64, "secs": 7}
{"ip": "10.40.41.218", "name": "kinderzimmer_sonos", "mac": "e4:b0:63:fa:6c:00", "model": "S3PL-00112EU", "fw": "2.0.0", "old_ssid": "elektrosmog.c", "new_ip": "10.40.30.218", "new_ssid": "dirtyvader-iot", "ok": true, "verified_ssid": "dirtyvader-iot", "rssi": -57, "secs": 7}
{"ip": "10.40.41.222", "name": "technik_pve", "mac": "e4:b0:63:fa:d5:34", "model": "S3PL-00112EU", "fw": "1.7.5", "old_ssid": "elektrosmog.c", "new_ip": "10.40.30.222", "new_ssid": "dirtyvader-iot", "ok": true, "verified_ssid": "dirtyvader-iot", "rssi": -26, "secs": 7}
{"ip": "10.40.41.226", "name": "kueche_steckdosen3", "mac": "10:20:ba:fb:70:fc", "model": "S4SW-002P16EU", "fw": "1.7.5", "old_ssid": "elektrosmog.c", "new_ip": "10.40.30.226", "new_ssid": "dirtyvader-iot", "ok": true, "verified_ssid": "dirtyvader-iot", "rssi": -79, "secs": 8}
{"ip": "10.40.41.206", "name": "kueche_steckdosen2", "mac": "98:a3:16:7c:5f:ec", "model": "S4SW-002P16EU", "fw": "1.7.5", "old_ssid": "elektrosmog.c", "new_ip": "10.40.30.206", "new_ssid": "dirtyvader-iot", "ok": true, "verified_ssid": "dirtyvader-iot", "rssi": -73, "secs": 8}
{"ip": "10.40.41.207", "name": "kueche_steckdosen4", "mac": "20:6e:f1:01:f9:f8", "model": "S4SW-002P16EU", "fw": "1.7.5", "old_ssid": "elektrosmog.c", "new_ip": "10.40.30.207", "new_ssid": "dirtyvader-iot", "ok": true, "verified_ssid": "dirtyvader-iot", "rssi": -72, "secs": 8}
{"ip": "10.40.41.205", "name": "bad_lichtschalter", "mac": "98:a3:16:79:38:50", "model": "S4SW-002P16EU", "fw": "1.7.5", "old_ssid": "elektrosmog", "new_ip": "10.40.30.205", "new_ssid": "dirtyvader-iot", "ok": true, "verified_ssid": "dirtyvader-iot", "rssi": -62, "secs": 13}
{"ip": "10.40.41.220", "name": "arbeitszimmer_technik", "mac": "48:f6:ee:dd:6a:f4", "model": "S4PL-00416EU", "fw": "1.7.99-powerstripg4prod1", "old_ssid": "elektrosmog.c", "new_ip": "10.40.30.220", "new_ssid": "dirtyvader-iot", "ok": true, "verified_ssid": "dirtyvader-iot", "rssi": -63, "secs": 8}
{"ip": "10.40.41.224", "name": "bad_steckdosen", "mac": "10:20:ba:fb:6d:64", "model": "S4SW-002P16EU", "fw": "1.7.5", "old_ssid": "elektrosmog.c", "new_ip": "10.40.30.224", "new_ssid": "dirtyvader-iot", "ok": true, "verified_ssid": "dirtyvader-iot", "rssi": -81, "secs": 8}

295
mcp/shelly/server.py Normal file
View File

@@ -0,0 +1,295 @@
#!/usr/bin/env python3
"""shelly-mcp — MCP server for managing Shellys and the VLAN1→VLAN30 migration (TP17).
Transport: StreamableHTTP (endpoint /mcp), matching the repo's other MCP servers.
Also serves a browser-viewable status page at `/` and JSON health at `/health`,
so it works as a local dev server (launch.json / preview_start).
Config via env (see .env.example):
SHELLY_IOT_SSID target IOT SSID (default: dirtyvader-iot)
SHELLY_IOT_PSK target IOT PSK (required for set_sta)
SHELLY_SWEEP default discovery range (default: 10.40.41.180-250)
SHELLY_MCP_HOST bind host (default: 127.0.0.1)
SHELLY_MCP_PORT bind port (default: 3030)
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
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import discover # probe(), load_inventory(), expand()
from mcp.server.fastmcp import FastMCP
from starlette.requests import Request
from starlette.responses import HTMLResponse, JSONResponse
# ---- tiny .env loader (no external deps) ----
_envfile = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")
if os.path.exists(_envfile):
for _line in open(_envfile):
_line = _line.strip()
if _line and not _line.startswith("#") and "=" in _line:
_k, _v = _line.split("=", 1)
os.environ.setdefault(_k.strip(), _v.strip().strip('"').strip("'"))
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"))
mcp = FastMCP("shelly-mcp", host=HOST, port=PORT, stateless_http=True,
instructions="Manage Shellys and drive the VLAN1→VLAN30 (dirtyvader-iot) "
"migration. Read tools are safe; set_sta reconfigures live "
"Wi-Fi and requires confirm=true.")
# ---------------- helpers ----------------
def _rpc(ip: str, method: str, params: dict | None = None, timeout: float = 6.0) -> dict:
"""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()
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]:
inv = discover.load_inventory()
ips = discover.expand(spec or SWEEP)
found = []
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=32) as ex:
for res in ex.map(discover.probe, ips):
if res:
m = inv.get(res["mac"])
res["inv_name"] = m["name"] if m else None
res["inv_id"] = m["shelly_id"] if m else None
res["inv_ip"] = m["inv_ip"] if m else None
res["ip_drifted"] = bool(m and m["inv_ip"] and m["inv_ip"] != res["ip"])
found.append(res)
found.sort(key=lambda x: (x["inv_id"] or "zzz"))
return found
# ---------------- MCP tools ----------------
@mcp.tool()
def discover_shellys(sweep: str | None = None) -> dict:
"""Sweep the LAN for Shellys (read-only) and match them to inventory by MAC.
sweep: IPv4 range like '10.40.41.180-250' or a single IP. Defaults to SHELLY_SWEEP.
Returns reachable devices (ip, gen, model, fw, mac, inventory name, ip_drift).
"""
found = _sweep(sweep)
return {"sweep": sweep or SWEEP, "reachable": len(found), "devices": found}
@mcp.tool()
def get_device(ip: str) -> dict:
"""Read a single Shelly's identity + current Wi-Fi STA status (read-only)."""
ident = discover.probe(ip)
if not ident:
return {"ip": ip, "reachable": False}
out = {"ip": ip, "reachable": True, "identity": ident}
try:
out["wifi_status"] = _rpc(ip, "WiFi.GetStatus")
out["wifi_config"] = _rpc(ip, "WiFi.GetConfig")
except Exception as e:
out["rpc_error"] = f"{type(e).__name__}: {e}"
return out
V30_NET = "10.40.30"
V30_MASK = "255.255.255.0"
V30_GW = "10.40.30.1"
def _sta_cfg(ssid: str, static: bool, target_ip: str) -> dict:
"""Build the WiFi STA config. Project decision (2026-07-11): all Shellys get a
STATIC IP on-device (keeping the last octet) plus a matching in-pool DHCP
reservation as backup — not DHCP-only."""
sta = {"ssid": ssid, "pass": IOT_PSK, "enable": True}
if static:
sta.update({"ipv4mode": "static", "ip": target_ip, "netmask": V30_MASK,
"gw": V30_GW, "nameserver": V30_GW})
else:
sta["ipv4mode"] = "dhcp"
return sta
@mcp.tool()
def set_sta(ip: str, confirm: bool = False, ssid: str | None = None,
static: bool = True) -> dict:
"""MIGRATION ACTION — point a Shelly's Wi-Fi STA at the IOT SSID, then reboot.
Reconfigures a LIVE device: it drops off VLAN1 and rejoins on VLAN30. Per the
project decision it sets a STATIC IP (10.40.30.<same-last-octet>, gw/dns .30.1);
a matching in-pool DHCP reservation exists as backup. Pass static=False to use DHCP.
A wrong PSK leaves it in AP-mode (physical reset needed). Therefore:
- confirm MUST be true to actually write,
- SHELLY_IOT_PSK must be set in the environment.
With confirm=false this is a dry-run (returns the exact config that would be sent).
"""
target_ssid = ssid or IOT_SSID
target_ip = f"{V30_NET}.{ip.rsplit('.', 1)[-1]}"
ident = discover.probe(ip)
plan = {"ip": ip, "target_ssid": target_ssid, "target_ip": target_ip,
"mode": "static" if static else "dhcp",
"device": ident, "psk_configured": bool(IOT_PSK)}
if not ident:
return {**plan, "ok": False, "error": "device not reachable"}
if not IOT_PSK:
return {**plan, "ok": False, "error": "SHELLY_IOT_PSK not set in environment"}
sta = _sta_cfg(target_ssid, static, target_ip)
if not confirm:
redacted = {**sta, "pass": "***"}
return {**plan, "ok": False, "dry_run": True,
"would_send": {"method": "WiFi.SetConfig",
"params": {"config": {"sta": redacted}}}}
try:
res = _rpc(ip, "WiFi.SetConfig", {"config": {"sta": sta}})
reboot = _rpc(ip, "Shelly.Reboot")
return {**plan, "ok": True, "set_result": res, "reboot": reboot,
"note": f"device rebooting; expect it at {target_ip} shortly"}
except Exception as e:
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).
ip = the device's NEW (10.40.30.x) address. Confirms reachability + SSID.
"""
ident = discover.probe(ip)
if not ident:
return {"ip": ip, "on_iot": False, "reachable": False}
out = {"ip": ip, "reachable": True, "identity": ident}
try:
st = _rpc(ip, "WiFi.GetStatus")
ssid = (st.get("result") or {}).get("ssid")
out["ssid"] = ssid
out["on_iot"] = (ssid == IOT_SSID) and ip.startswith("10.40.30.")
except Exception as e:
out["rpc_error"] = f"{type(e).__name__}: {e}"
out["on_iot"] = False
return out
@mcp.tool()
def reservation_plan(sweep: str | None = None) -> dict:
"""Build the VLAN30 DHCP-reservation plan (MAC → suggested .30.x, keeping last octet).
Suggests preserving each Shelly's current last octet under 10.40.30.0/24.
Output feeds MikroMCP manage_dhcp_lease. Read-only.
"""
found = _sweep(sweep)
rows = []
for f in found:
last = f["ip"].split(".")[-1]
rows.append({"name": f.get("inv_name") or f.get("inv_id") or "?",
"mac": f["mac"], "current_ip": f["ip"],
"suggested_v30_ip": f"10.40.30.{last}", "gen": f["gen"]})
return {"count": len(rows), "server": "dhcp-v30", "reservations": rows}
# ---------------- browser-viewable routes ----------------
# NOTE: sync route fns — Starlette runs them in a threadpool, so the blocking
# network sweep never freezes the async event loop (which also serves /mcp).
@mcp.custom_route("/health", methods=["GET"])
def health(_req: Request) -> JSONResponse:
return JSONResponse({"status": "ok", "server": "shelly-mcp",
"iot_ssid": IOT_SSID, "psk_configured": bool(IOT_PSK),
"mcp_endpoint": "/mcp"})
@mcp.custom_route("/", methods=["GET"])
def index(_req: Request) -> HTMLResponse:
try:
found = _sweep()
except Exception as e:
found = []
err = f"{type(e).__name__}: {e}"
else:
err = None
rows = "".join(
f"<tr><td>{f.get('inv_id') or '?'}</td><td>{f['ip']}</td>"
f"<td>g{f['gen']}</td><td>{f['model']}</td><td>{f['fw']}</td>"
f"<td>{f['mac']}</td><td>{f.get('inv_name') or ''}</td>"
f"<td>{'⚠︎ '+(f.get('inv_ip') or '') if f.get('ip_drifted') else ''}</td></tr>"
for f in found)
html = f"""<!doctype html><meta charset=utf-8>
<title>shelly-mcp</title>
<style>body{{font:14px/1.5 system-ui;margin:2rem;color:#222}}
h1{{font-size:1.3rem}} table{{border-collapse:collapse;margin-top:1rem}}
td,th{{border:1px solid #ccc;padding:.25rem .6rem;text-align:left}}
th{{background:#f4f4f4}} .m{{color:#666}}</style>
<h1>shelly-mcp <span class=m>· TP17 VLAN1→VLAN30</span></h1>
<p class=m>MCP endpoint <code>/mcp</code> · target SSID <b>{IOT_SSID}</b> ·
PSK {'set' if IOT_PSK else '<b>NOT set</b>'} · sweep <code>{SWEEP}</code></p>
{'<p style=color:#b00>'+err+'</p>' if err else ''}
<p>{len(found)} Shellys reachable:</p>
<table><tr><th>id<th>ip<th>gen<th>model<th>fw<th>mac<th>name<th>drift</tr>{rows}</table>
"""
return HTMLResponse(html)
if __name__ == "__main__":
mcp.run(transport="streamable-http")