christianandClaude Opus 5.5
d06798adf0
Every 10 minutes the watchdog maps each Dante device: name and IP from Companion, MAC from NA Ingest's ARP table, switch port and PoE state from the Netgear switches in dante_switches, saved to dante-map.json. A DANTE hold on a missing PoE-powered device power-cycles the port it is on now (found live by MAC, or its last known port if it has aged out and nothing else is there); never a port with more than 2 devices. Re-patching, adding or renaming devices needs no config change. First live map: 19 devices, the 4 Studio B boxes on PoE. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
222 lines
9.4 KiB
Python
222 lines
9.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Netgear AV (M4250-class) switches on the Dante network, over their CLI (telnet).
|
|
|
|
The switch login is read from NA Ingest's login keychain (service
|
|
"netgear-status-panel", the account is the username) and is never printed. Store
|
|
it once in Terminal on NA Ingest (it asks for the password twice):
|
|
security add-generic-password -s netgear-status-panel -a admin -w
|
|
A login saved in the Passwords app cannot be read by scripts. The keychain is only
|
|
unlocked inside the logged-in session, so run this from a LaunchAgent or a
|
|
Terminal there, not over SSH.
|
|
|
|
python3 netgear.py lookup 00:1d:c1:51:18:84 ... which switch / port each MAC is on (read-only)
|
|
python3 netgear.py poe-status 10.66.1.102 0/14 PoE state of ports (read-only)
|
|
python3 netgear.py poe-cycle 10.66.1.102 0/14 [MAC] power-cycle one PoE port (refuses if MAC is elsewhere)
|
|
"""
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import telnetlib # stdlib in the macOS python3 (3.9)
|
|
|
|
SWITCHES = ["10.66.1.100", "10.66.1.102"]
|
|
KEYCHAIN_SERVICE = "netgear-status-panel"
|
|
|
|
|
|
def credentials(service=KEYCHAIN_SERVICE):
|
|
attrs = subprocess.run(["/usr/bin/security", "find-generic-password", "-s", service],
|
|
capture_output=True, text=True).stdout
|
|
m = re.search(r'"acct"<blob>="([^"]*)"', attrs)
|
|
password = subprocess.run(["/usr/bin/security", "find-generic-password", "-s", service, "-w"],
|
|
capture_output=True, text=True).stdout.rstrip("\n")
|
|
if not m or not password:
|
|
raise RuntimeError('keychain item "%s" not readable here (not stored yet, locked keychain, '
|
|
'or no access)' % service)
|
|
return m.group(1), password
|
|
|
|
|
|
class Switch:
|
|
PROMPT = re.compile(rb"[>#]\s*$")
|
|
|
|
def __init__(self, host, user, password, timeout=8):
|
|
self.t = telnetlib.Telnet(host, 23, timeout)
|
|
self.timeout = timeout
|
|
self.t.read_until(b"User:", timeout)
|
|
self.t.write(user.encode() + b"\n")
|
|
self.t.read_until(b"Password:", timeout)
|
|
self.t.write(password.encode() + b"\n")
|
|
_, m, text = self.t.expect([self.PROMPT, rb"(?i)fail|denied|incorrect"], timeout)
|
|
if m is None or not self.PROMPT.search(m.group(0)):
|
|
raise RuntimeError("%s: login failed" % host)
|
|
if text.rstrip().endswith(b">"):
|
|
self.t.write(b"enable\n")
|
|
idx, _, _ = self.t.expect([rb"Password:", self.PROMPT], timeout)
|
|
if idx == 0:
|
|
self.t.write(b"\n") # blank enable password
|
|
self.t.expect([self.PROMPT], timeout)
|
|
self.run("terminal length 0")
|
|
|
|
def run(self, command):
|
|
self.t.write(command.encode() + b"\n")
|
|
_, _, text = self.t.expect([self.PROMPT], self.timeout)
|
|
lines = text.decode("ascii", "replace").splitlines()
|
|
return "\n".join(lines[1:-1]) # drop the echoed command and the prompt
|
|
|
|
def close(self):
|
|
try:
|
|
self.t.write(b"logout\n")
|
|
finally:
|
|
self.t.close()
|
|
|
|
|
|
def lookup(macs):
|
|
user, password = credentials()
|
|
wanted = {m.lower(): m for m in macs}
|
|
for host in SWITCHES:
|
|
sw = Switch(host, user, password)
|
|
try:
|
|
model = [l.strip() for l in sw.run("show version").splitlines()
|
|
if re.search(r"(?i)machine model|system description", l)]
|
|
table = sw.run("show mac-addr-table")
|
|
print("== %s %s" % (host, " | ".join(model)))
|
|
hits = 0
|
|
for line in table.splitlines():
|
|
mm = re.search(r"([0-9A-Fa-f]{2}(?::[0-9A-Fa-f]{2}){5})\s+(\S+)", line)
|
|
if mm and mm.group(1).lower() in wanted:
|
|
print(" %s on port %s (%s)" % (mm.group(1).lower(), mm.group(2), line.strip()))
|
|
hits += 1
|
|
print(" %d of the MACs here; %d entries in the table" %
|
|
(hits, sum(1 for l in table.splitlines() if re.search(r"(?:[0-9A-Fa-f]{2}:){5}", l))))
|
|
finally:
|
|
sw.close()
|
|
|
|
|
|
def poe_status(host, ports):
|
|
"""Read-only: what the switch reports for PoE on these ports."""
|
|
user, password = credentials()
|
|
sw = Switch(host, user, password)
|
|
try:
|
|
for port in ports:
|
|
print("== %s port %s" % (host, port))
|
|
for line in sw.run("show poe port info %s" % port).splitlines():
|
|
if line.strip():
|
|
print(" " + line.rstrip())
|
|
finally:
|
|
sw.close()
|
|
|
|
|
|
MAC_RE = re.compile(r"(?:[0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}")
|
|
|
|
|
|
def poe_cycle(host, port, expect_mac=None):
|
|
# Used directly only for a known port; poe_cycle_device finds the port by MAC.
|
|
"""Power-cycle PoE on ONE port ("poe reset" in Interface Config mode). Nothing is saved
|
|
to the switch's startup config: the port's settings are unchanged afterwards.
|
|
|
|
With expect_mac, refuse if the port has been re-patched: if the switch has learned
|
|
other MACs on that port and not this one, a different device is plugged in there.
|
|
(No MACs at all is fine: a dead device ages out of the table.)"""
|
|
if not re.fullmatch(r"\d+/\d+", port):
|
|
raise ValueError("one port like 0/14, got %r" % port)
|
|
user, password = credentials()
|
|
sw = Switch(host, user, password)
|
|
try:
|
|
if expect_mac:
|
|
seen = {m.lower() for m in MAC_RE.findall(sw.run("show mac-addr-table interface %s" % port))}
|
|
if seen and expect_mac.lower() not in seen:
|
|
raise RuntimeError("%s %s has %s, not %s - re-patched? not power-cycling"
|
|
% (host, port, ", ".join(sorted(seen)), expect_mac.lower()))
|
|
sw.run("configure")
|
|
sw.run("interface %s" % port)
|
|
answer = sw.run("poe reset")
|
|
sw.run("exit")
|
|
sw.run("exit")
|
|
finally:
|
|
sw.close()
|
|
if re.search(r"(?i)invalid|error|fail|unrecognized", answer):
|
|
raise RuntimeError("%s %s: switch refused poe reset: %s" % (host, port, answer.strip()))
|
|
print("POE RESET %s %s ok%s" % (host, port, (": " + answer.strip()) if answer.strip() else ""))
|
|
|
|
|
|
# --- Find devices by MAC, wherever they are plugged in ---------------------
|
|
|
|
TABLE_RE = re.compile(r"((?:[0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2})\s+(\d+/\d+)")
|
|
MAX_SHARED = 2 # more devices than this on one port means an uplink: never power-cycle it
|
|
|
|
|
|
def parse_mac_table(text):
|
|
"""'show mac-addr-table' -> ({mac: port}, {port: number of MACs learned on it})."""
|
|
where, counts = {}, {}
|
|
for mac, port in TABLE_RE.findall(text or ""):
|
|
where[mac.lower()] = port
|
|
counts[port] = counts.get(port, 0) + 1
|
|
return where, counts
|
|
|
|
|
|
def best_ports(tables, macs):
|
|
"""tables: {switch: table text}. -> {mac: (switch, port, devices on that port)}.
|
|
A MAC is seen on every switch it crosses; its own port is where it has the fewest
|
|
neighbours (on the far switch it shows up on the uplink with everything else)."""
|
|
best = {}
|
|
for host, text in tables.items():
|
|
where, counts = parse_mac_table(text)
|
|
for mac in macs:
|
|
if mac in where:
|
|
cand = (host, where[mac], counts[where[mac]])
|
|
if mac not in best or cand[2] < best[mac][2]:
|
|
best[mac] = cand
|
|
return best
|
|
|
|
|
|
def poe_delivering(text):
|
|
return bool(re.search(r"(?i)delivering power", text or ""))
|
|
|
|
|
|
def survey(switches, macs):
|
|
"""Read-only: {mac: {"switch", "port", "shared", "poe"}} for every MAC found."""
|
|
macs = [m.lower() for m in macs]
|
|
user, password = credentials()
|
|
sessions, tables = {}, {}
|
|
try:
|
|
for host in switches:
|
|
sessions[host] = Switch(host, user, password)
|
|
tables[host] = sessions[host].run("show mac-addr-table")
|
|
found = {}
|
|
for mac, (host, port, shared) in best_ports(tables, macs).items():
|
|
poe = shared <= MAX_SHARED and poe_delivering(sessions[host].run("show poe port info %s" % port))
|
|
found[mac] = {"switch": host, "port": port, "shared": shared, "poe": poe}
|
|
return found
|
|
finally:
|
|
for sw in sessions.values():
|
|
sw.close()
|
|
|
|
|
|
def poe_cycle_device(switches, mac, last_known=None):
|
|
"""Power-cycle the PoE port a device is on NOW, found by its MAC. If it has aged out of
|
|
the switch tables (dead long enough), use its last known port - poe_cycle then refuses
|
|
unless that port is empty of other devices. Never touches a port shared by more than
|
|
MAX_SHARED devices. Returns (switch, port, how it was found)."""
|
|
mac = mac.lower()
|
|
live = survey(switches, [mac]).get(mac)
|
|
if live:
|
|
if live["shared"] > MAX_SHARED:
|
|
raise RuntimeError("%s is behind %s port %s with %d devices (an uplink) - not power-cycling"
|
|
% (mac, live["switch"], live["port"], live["shared"]))
|
|
host, port, how = live["switch"], live["port"], "found now"
|
|
elif last_known and last_known.get("switch") and last_known.get("port"):
|
|
host, port, how = last_known["switch"], last_known["port"], "last known port"
|
|
else:
|
|
raise RuntimeError("%s is not on any Dante switch and has no last known port" % mac)
|
|
poe_cycle(host, port, expect_mac=mac)
|
|
return host, port, how
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) >= 3 and sys.argv[1] == "lookup":
|
|
lookup(sys.argv[2:])
|
|
elif len(sys.argv) >= 4 and sys.argv[1] == "poe-status":
|
|
poe_status(sys.argv[2], sys.argv[3:])
|
|
elif len(sys.argv) in (4, 5) and sys.argv[1] == "poe-cycle":
|
|
poe_cycle(sys.argv[2], sys.argv[3], sys.argv[4] if len(sys.argv) == 5 else None)
|
|
else:
|
|
sys.exit(__doc__)
|