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)
"""
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()


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:])
    else:
        sys.exit(__doc__)