prep for next ai session
This commit is contained in:
@@ -0,0 +1,253 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user