153 lines
6.4 KiB
Python
Executable File
153 lines
6.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""sonos-inv — zero-dep Sonos-Inventar & Verifikation über die :1400-HTTP-API.
|
|
|
|
KEIN MCP/Daemon, KEINE Fremd-Libs (nur Python-stdlib). Ersetzt für unseren Zweck
|
|
(Inventar + Umzugs-Verifikation) einen Sonos-MCP-Container. Sonos hat KEINE API zum
|
|
WLAN-Umstellen — das ist hier bewusst nicht abgebildet.
|
|
|
|
WICHTIG — Discovery ist bei Sonos pro L2/VLAN **partitioniert** (SSDP-Multicast quert
|
|
keine VLANs). Darum: SSDP auf dem lokalen Segment + optional **ein Seed je weiterer
|
|
Partition**. Bei BF: lokal (VLAN 1) + Seed 10.40.40.199 (VLAN 40).
|
|
|
|
Usage:
|
|
./sonos-inv.py # nur lokales Segment (SSDP)
|
|
./sonos-inv.py --seeds 10.40.40.199 # + VLAN-40-Partition
|
|
./sonos-inv.py --seeds 10.40.40.199,10.40.41.193 # explizit beide Partitionen
|
|
./sonos-inv.py --json # maschinenlesbar
|
|
./sonos-inv.py --verify 40 # Exit!=0 wenn ein Gerät NICHT in VLAN 40
|
|
|
|
Felder je Gerät: ip, vlan(rateil aus 3. Oktett), model, zone, mac, swgen(1=S1/2=S2),
|
|
sw, wifi(WifiEnabled), eth(EthLink).
|
|
"""
|
|
import argparse, html, json, re, socket, sys, urllib.request
|
|
|
|
TIMEOUT = 5
|
|
SOAP_BODY = (
|
|
'<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" '
|
|
's:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"><s:Body>'
|
|
'<u:GetZoneGroupState xmlns:u="urn:schemas-upnp-org:service:ZoneGroupTopology:1">'
|
|
'</u:GetZoneGroupState></s:Body></s:Envelope>'
|
|
)
|
|
|
|
def _get(ip, path, retries=2):
|
|
for _ in range(retries):
|
|
try:
|
|
return urllib.request.urlopen(f"http://{ip}:1400{path}", timeout=TIMEOUT).read().decode(errors="replace")
|
|
except Exception:
|
|
continue
|
|
return ""
|
|
|
|
def ssdp_discover(timeout=3):
|
|
"""SSDP M-SEARCH nach ZonePlayern auf dem lokalen Segment → set(IPs)."""
|
|
msg = "\r\n".join([
|
|
"M-SEARCH * HTTP/1.1", "HOST: 239.255.255.250:1900", 'MAN: "ssdp:discover"',
|
|
"MX: 2", "ST: urn:schemas-upnp-org:device:ZonePlayer:1", "", "",
|
|
])
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
|
|
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
s.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2)
|
|
s.settimeout(timeout)
|
|
found = set()
|
|
try:
|
|
s.sendto(msg.encode(), ("239.255.255.250", 1900))
|
|
import time
|
|
t = time.time()
|
|
while time.time() - t < timeout + 0.5:
|
|
try:
|
|
_, addr = s.recvfrom(2048)
|
|
found.add(addr[0])
|
|
except socket.timeout:
|
|
break
|
|
finally:
|
|
s.close()
|
|
return found
|
|
|
|
def zone_group_members(ip):
|
|
"""GetZoneGroupState an einem Gerät → {ip: {attr: val}} aller Member dieser Partition."""
|
|
req = urllib.request.Request(
|
|
f"http://{ip}:1400/ZoneGroupTopology/Control", data=SOAP_BODY.encode(),
|
|
headers={"Content-Type": "text/xml; charset=utf-8",
|
|
"SOAPACTION": '"urn:schemas-upnp-org:service:ZoneGroupTopology:1#GetZoneGroupState"'})
|
|
try:
|
|
d = urllib.request.urlopen(req, timeout=TIMEOUT).read().decode(errors="replace")
|
|
except Exception:
|
|
return {}
|
|
m = re.search(r"<ZoneGroupState>(.*)</ZoneGroupState>", d, re.S)
|
|
body = html.unescape(m.group(1)) if m else d
|
|
members = {}
|
|
for tag in re.finditer(r"<(?:ZoneGroupMember|Satellite)\s+([^>]*?)/?>", body):
|
|
attrs = dict(re.findall(r'(\w+)="([^"]*)"', tag.group(1)))
|
|
loc = attrs.get("Location", "")
|
|
mip = re.search(r"http://([\d.]+):", loc)
|
|
if mip:
|
|
members[mip.group(1)] = attrs
|
|
return members
|
|
|
|
def device_detail(ip):
|
|
"""Modell + MAC aus device_description/status/zp (was ZGS nicht liefert)."""
|
|
dd = _get(ip, "/xml/device_description.xml")
|
|
zp = _get(ip, "/status/zp")
|
|
def g(pat, src):
|
|
mm = re.search(pat, src)
|
|
return mm.group(1) if mm else ""
|
|
return {
|
|
"model": g(r"<modelName>(.*?)</modelName>", dd),
|
|
"modelnum": g(r"<modelNumber>(.*?)</modelNumber>", dd),
|
|
"mac": g(r"<SerialNumber>(.*?)</SerialNumber>", zp).split(":")[0], # "AA-BB-..-XX:3" → MAC
|
|
}
|
|
|
|
def vlan_of(ip):
|
|
o = ip.split(".")
|
|
if len(o) == 4 and o[0] == "10" and o[1] == "40":
|
|
return {"10": "10", "20": "20", "30": "30", "40": "40", "50": "50",
|
|
"90": "90", "99": "99", "41": "1/legacy"}.get(o[2], o[2])
|
|
return "?"
|
|
|
|
def inventory(seeds):
|
|
candidates = set(ssdp_discover()) | set(seeds)
|
|
members = {}
|
|
for ip in sorted(candidates):
|
|
if ip in members:
|
|
continue
|
|
members.update(zone_group_members(ip))
|
|
rows = []
|
|
for ip in sorted(members, key=lambda x: tuple(int(p) for p in x.split("."))):
|
|
a = members[ip]
|
|
det = device_detail(ip)
|
|
rows.append({
|
|
"ip": ip, "vlan": vlan_of(ip),
|
|
"model": f"{det['model']} ({det['modelnum']})".strip(),
|
|
"zone": a.get("ZoneName", ""), "mac": det["mac"],
|
|
"swgen": a.get("SWGen", ""), "sw": a.get("SoftwareVersion", ""),
|
|
"wifi": a.get("WifiEnabled", ""), "eth": a.get("EthLink", ""),
|
|
})
|
|
return rows
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description="Sonos-Inventar/Verifikation (zero-dep, :1400)")
|
|
ap.add_argument("--seeds", default="", help="Komma-IPs, je 1 pro weiterer VLAN-Partition (z.B. 10.40.40.199)")
|
|
ap.add_argument("--json", action="store_true")
|
|
ap.add_argument("--verify", metavar="VLAN", help="Exit!=0 wenn ein Gerät NICHT im genannten VLAN ist")
|
|
args = ap.parse_args()
|
|
seeds = [s.strip() for s in args.seeds.split(",") if s.strip()]
|
|
rows = inventory(seeds)
|
|
if args.json:
|
|
print(json.dumps(rows, indent=2, ensure_ascii=False))
|
|
else:
|
|
h = f"{'IP':15} {'VLAN':9} {'Modell':28} {'Zone':18} {'MAC':18} {'Gen':3} {'SW':14} {'Wifi':4} {'Eth':3}"
|
|
print(h); print("-" * len(h))
|
|
for r in rows:
|
|
print(f"{r['ip']:15} {r['vlan']:9} {r['model']:28.28} {r['zone']:18.18} {r['mac']:18} "
|
|
f"{r['swgen']:3} {r['sw']:14} {r['wifi']:4} {r['eth']:3}")
|
|
print(f"\n{len(rows)} Sonos. (Discovery pro VLAN partitioniert — Seeds: {seeds or 'nur lokal'})")
|
|
if args.verify:
|
|
bad = [r["ip"] for r in rows if r["vlan"] != args.verify]
|
|
if bad:
|
|
print(f"VERIFY FAIL: nicht in VLAN {args.verify}: {bad}", file=sys.stderr)
|
|
sys.exit(1)
|
|
print(f"VERIFY OK: alle {len(rows)} in VLAN {args.verify}")
|
|
return 0
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|