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"]
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):
    """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 ""))


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