status-indicators/netgear.py

#!/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"]        # the CLI's default; the watchdog passes "dante_switches" from config
KEYCHAIN_SERVICE = "netgear-status-panel"


def config_switches():
    """"dante_switches" from this folder's config.json, for the command line (the watchdog passes its own)."""
    import json, os
    try:
        with open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json")) as f:
            cfg = json.load(f)
        global MAX_SHARED
        MAX_SHARED = cfg.get("dante_max_shared", MAX_SHARED)
        return cfg.get("dante_switches") or SWITCHES
    except (OSError, ValueError):
        return SWITCHES


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 config_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__)