125 lines
4.4 KiB
Python
125 lines
4.4 KiB
Python
#!/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()
|