#!/usr/bin/env python3 """ snmp-switch-audit.py — READ-ONLY switch inventory via SNMP. Gathers interface status/speed/errors, LLDP neighbor topology, port-channel (LAG) membership, and VLAN membership from any SNMPv2c-capable switch. Designed for Dell/Radlan (Neyland) and standard IF/LLDP/Q-BRIDGE MIB switches, but works on any SNMP-manageable device. All parameters configurable via env vars or CLI flags so this works on any network: SNMP_COMMUNITY SNMPv2c community string (env, default: public) SWITCH_IPS space-separated switch IPs (env, or pass as args) OUTPUT_DIR where to write per-switch (env, default: returned-logs/snmp) Usage: SNMP_COMMUNITY=kn3lmgmt SWITCH_IPS="192.168.0.9 192.168.0.12" \\ python3 snmp-switch-audit.py # or pass IPs as positional args: SNMP_COMMUNITY=kn3lmgmt python3 snmp-switch-audit.py 192.168.0.9 192.168.0.12 Requires: pysnmp (pip install pysnmp) or net-snmp utils (snmpwalk) on PATH. Outputs: per-switch JSON + human-readable text in OUTPUT_DIR. Read-only: sends only SNMP GET/GETNEXT/GETBULK. Never SETs anything. """ import argparse import json import os import re import subprocess import sys COMMUNITY = os.environ.get("SNMP_COMMUNITY", "public") OUTPUT_DIR = os.environ.get("OUTPUT_DIR", os.path.join(os.path.dirname(__file__), "..", "..", "returned-logs", "snmp")) # OID constants OID_SYSDESCR = "1.3.6.1.2.1.1.1.0" OID_SYSNAME = "1.3.6.1.2.1.1.5.0" OID_IF_NAME = "1.3.6.1.2.1.31.1.1.1.1" OID_IF_SPEED = "1.3.6.1.2.1.2.2.1.5" OID_IF_OPER = "1.3.6.1.2.1.2.2.1.8" OID_IF_INERR = "1.3.6.1.2.1.2.2.1.14" OID_IF_OUTERR = "1.3.6.1.2.1.2.2.1.20" OID_IF_INOCT = "1.3.6.1.2.1.31.1.1.1.6" OID_IF_OUTOCT = "1.3.6.1.2.1.31.1.1.1.10" OID_LACP_LAG = "1.2.840.10006.300.43.1.1.1.1" OID_LLDP_REM_PORT = "1.0.8802.1.1.2.1.4.1.1.7" OID_LLDP_REM_SYSNAME = "1.0.8802.1.1.2.1.4.1.1.9" OID_LLDP_REM_CHASSIS = "1.0.8802.1.1.2.1.4.1.1.6" OID_LLDP_REM_LOCALPORT = "1.0.8802.1.1.2.1.4.1.1.3" OID_QBRIDGE_VLAN = "1.3.6.1.2.1.17.7.1.4.3.1.1" def snmpget(ip, oid): """Single SNMP GET, returns string value or None.""" try: r = subprocess.run( ["snmpget", "-Oqv", "-v2c", "-c", COMMUNITY, ip, oid], capture_output=True, text=True, timeout=10) if r.returncode == 0 and r.stdout.strip(): return r.stdout.strip().strip('"') except (subprocess.TimeoutExpired, FileNotFoundError): pass return None def snmpwalk(ip, oid): """SNMP BULKWALK, returns dict of ifIndex -> value.""" try: r = subprocess.run( ["snmpbulkwalk", "-Oqv", "-v2c", "-c", COMMUNITY, ip, oid], capture_output=True, text=True, timeout=30) if r.returncode != 0: r = subprocess.run( ["snmpwalk", "-Oqv", "-v2c", "-c", COMMUNITY, ip, oid], capture_output=True, text=True, timeout=30) except (subprocess.TimeoutExpired, FileNotFoundError): return {} result = {} for i, line in enumerate(r.stdout.strip().split("\n"), 1): line = line.strip().strip('"') if line: result[i] = line return result def walk_indexed(ip, oid): """SNMP walk preserving OID index. Returns dict: index_str -> value.""" try: r = subprocess.run( ["snmpbulkwalk", "-v2c", "-c", COMMUNITY, ip, oid], capture_output=True, text=True, timeout=30) if r.returncode != 0: r = subprocess.run( ["snmpwalk", "-v2c", "-c", COMMUNITY, ip, oid], capture_output=True, text=True, timeout=30) except (subprocess.TimeoutExpired, FileNotFoundError): return {} result = {} for line in r.stdout.strip().split("\n"): m = re.search(r'(\d+)\s*=\s*(.+)', line) if not m: m = re.search(r'\.(\d+)\s*=\s*(.+)', line) if m: idx = m.group(1).split(".")[-1] val = m.group(2).strip() val = re.sub(r'^(INTEGER: |STRING: |Hex-STRING: |Gauge32: |Counter32: |Counter64: )', '', val) result[idx] = val return result def audit_switch(ip): """Gather all data for one switch.""" data = {"ip": ip} data["sysDescr"] = snmpget(ip, OID_SYSDESCR) data["sysName"] = snmpget(ip, OID_SYSNAME) if not data["sysDescr"]: return data names = snmpwalk(ip, OID_IF_NAME) speeds = snmpwalk(ip, OID_IF_SPEED) oper = snmpwalk(ip, OID_IF_OPER) inerr = snmpwalk(ip, OID_IF_INERR) outerr = snmpwalk(ip, OID_IF_OUTERR) interfaces = [] for idx in sorted(names.keys()): if idx not in names: continue speed_raw = speeds.get(idx, "0") try: speed_mbps = int(re.sub(r'\D', '', str(speed_raw))) // 1000000 except (ValueError, TypeError): speed_mbps = 0 is_up = str(oper.get(idx, "0")).strip() == "1" interfaces.append({ "ifIndex": idx, "name": names[idx], "speedMbps": speed_mbps, "up": is_up, "inErrors": inerr.get(idx, "0"), "outErrors": outerr.get(idx, "0"), }) data["interfaces"] = interfaces # LLDP neighbors rem_ports = walk_indexed(ip, OID_LLDP_REM_PORT) rem_sysnames = walk_indexed(ip, OID_LLDP_REM_SYSNAME) rem_chassis = walk_indexed(ip, OID_LLDP_REM_CHASSIS) rem_local = walk_indexed(ip, OID_LLDP_REM_LOCALPORT) lldp = [] for idx in rem_ports: lldp.append({ "localPort": rem_local.get(idx, "?"), "remotePort": rem_ports[idx], "remoteSysName": rem_sysnames.get(idx, ""), "remoteChassis": rem_chassis.get(idx, ""), }) data["lldpNeighbors"] = lldp # LACP LAG table lag_data = walk_indexed(ip, OID_LACP_LAG) data["lagTable"] = lag_data # VLAN membership vlan_data = walk_indexed(ip, OID_QBRIDGE_VLAN) data["vlans"] = vlan_data return data def print_switch(data): """Human-readable summary.""" print(f"\n{'='*60}") print(f" {data.get('sysName', data['ip'])} ({data['ip']})") print(f" {data.get('sysDescr', '?')}") print(f"{'='*60}") print(f"\n Active ports (UP only):") print(f" {'Port':<12} {'Speed':>10} {'InErrors':>10} {'OutErrors':>10}") print(f" {'-'*12} {'-'*10} {'-'*10} {'-'*10}") for iface in data.get("interfaces", []): if iface["up"]: print(f" {iface['name']:<12} {iface['speedMbps']:>8}Mb " f"{iface['inErrors']:>10} {iface['outErrors']:>10}") err_ports = [i for i in data.get("interfaces", []) if i["up"] and (int(i["inErrors"] or 0) > 0 or int(i["outErrors"] or 0) > 0)] if err_ports: print(f"\n *** PORTS WITH ERRORS ***") for p in err_ports: print(f" {p['name']}: inErr={p['inErrors']} outErr={p['outErrors']}") if data.get("lldpNeighbors"): print(f"\n LLDP neighbors:") for n in data["lldpNeighbors"]: sysname = n.get("remoteSysName", "") or "(unknown)" print(f" local={n['localPort']:<6} remote={n['remotePort']:<20} {sysname}") if data.get("lagTable"): print(f"\n LACP/LAG table entries: {len(data['lagTable'])}") def main(): ap = argparse.ArgumentParser( description="READ-ONLY SNMP switch audit (portable, config-driven)") ap.add_argument("switches", nargs="*", help="switch IPs (env: SWITCH_IPS)") ap.add_argument("--community", default=COMMUNITY, help=f"SNMPv2c community (env: SNMP_COMMUNITY, default: {COMMUNITY})") ap.add_argument("--output", default=OUTPUT_DIR, help=f"output dir (env: OUTPUT_DIR)") args = ap.parse_args() community = args.community ips = args.switches if not ips: env_ips = os.environ.get("SWITCH_IPS", "") ips = env_ips.split() if not ips: ap.error("no switch IPs provided (pass as args or set SWITCH_IPS)") os.makedirs(args.output, exist_ok=True) all_data = [] for ip in ips: globals()["COMMUNITY"] = community data = audit_switch(ip.strip()) all_data.append(data) print_switch(data) outpath = os.path.join(args.output, f"switch-{ip}.json") with open(outpath, "w") as f: json.dump(data, f, indent=2) print(f"\n -> {outpath}") combined = os.path.join(args.output, "switches-all.json") with open(combined, "w") as f: json.dump(all_data, f, indent=2) print(f"\n Combined: {combined}") return 0 if __name__ == "__main__": sys.exit(main())