#!/usr/bin/env python3 # -*- coding: utf-8 -*- # Read-only port status collector: IF-MIB oper/admin/speed + FDB learned MAC per port. import json, urllib.request, subprocess, re, sys, collections import concurrent.futures as cf import l2cfg TOK = 'd3abb125eb301a021ef80a2847a803a7cd2d0b1d3c8337a14d09db9d571369b5' API = 'http://192.168.87.20:8080/api_jsonrpc.php' def call(m, p): d = json.dumps({"jsonrpc": "2.0", "method": m, "params": p, "id": 1}).encode() q = urllib.request.Request(API, data=d, headers={'Content-Type': 'application/json', 'Authorization': 'Bearer ' + TOK}) return json.loads(urllib.request.urlopen(q, timeout=60).read().decode()) AUTH = {0: 'MD5', 1: 'SHA', 2: 'SHA-224', 3: 'SHA-256', 4: 'SHA-384', 5: 'SHA-512'} PRIV = {0: 'DES', 1: 'AES', 2: 'AES-192', 3: 'AES-256', 4: 'AES', 5: 'AES-256'} def creds(i): det = i.get('details') or {}; ver = str(det.get('version', '1')) if ver == '3': lvl = int(det.get('securitylevel', 0)) a = ['-v3', '-l', {0: 'noAuthNoPriv', 1: 'authNoPriv', 2: 'authPriv'}.get(lvl, 'authPriv'), '-u', det.get('securityname', '')] if lvl >= 1: a += ['-a', AUTH.get(int(det.get('authprotocol', 0)), 'SHA'), '-A', det.get('authpassphrase', '')] if lvl >= 2: a += ['-x', PRIV.get(int(det.get('privprotocol', 0)), 'AES'), '-X', det.get('privpassphrase', '')] return a return ['-v2c', '-c', det.get('community', 'public')] def walk(args, ip, oid): try: r = subprocess.run(['snmpbulkwalk', '-On'] + args + ['-t', '3', '-r', '1', ip, oid], capture_output=True, text=True, timeout=40) return [l for l in r.stdout.splitlines() if ' = ' in l and 'No Such' not in l] except Exception: return [] def num(v): m = re.search(r'-?\d+', v); return int(m.group()) if m else None def strval(v): return v.split(':', 1)[1].strip().strip('"') if ':' in v else v.strip() O = { 'ifIndex':'1.3.6.1.2.1.2.2.1.1','ifDescr':'1.3.6.1.2.1.2.2.1.2','ifType':'1.3.6.1.2.1.2.2.1.3', 'ifSpeed':'1.3.6.1.2.1.2.2.1.5','ifAdmin':'1.3.6.1.2.1.2.2.1.7','ifOper':'1.3.6.1.2.1.2.2.1.8', 'ifLastChange':'1.3.6.1.2.1.2.2.1.9','ifName':'1.3.6.1.2.1.31.1.1.1.1','ifHighSpeed':'1.3.6.1.2.1.31.1.1.1.15', 'ifAlias':'1.3.6.1.2.1.31.1.1.1.18','ifConnector':'1.3.6.1.2.1.31.1.1.1.17','duplex':'1.3.6.1.2.1.10.7.2.1.19', 'fdbPort':'1.3.6.1.2.1.17.4.3.1.2','bpIf':'1.3.6.1.2.1.17.1.4.1.2','fdbStatus':'1.3.6.1.2.1.17.4.3.1.3', } POE_STD = '1.3.6.1.2.1.105.1.1.1.1.1' # RFC 3621 pethPsePortAdminEnable POE_HW_JUDGE = '1.3.6.1.4.1.2011.5.25.195.4.1.2' # HUAWEI-POE-MIB hwPoePortJudgeIsSupport (TruthValue) def parse(lines, base): out = [] for ln in lines: l, r = ln.split(' = ', 1); o = l.strip().lstrip('.') if o.startswith(base): out.append((o[len(base):].lstrip('.'), r.strip())) return out def collect(t): ip, nm, iface = t; args = creds(iface) d = {'ip': ip, 'name': nm, 'ok': False} idx = [n for s, v in parse(walk(args, ip, O['ifIndex']), O['ifIndex']) for n in [num(s.split('.')[-1])] if n is not None] if not idx: # fall back to ifName walk presence fn = parse(walk(args, ip, O['ifName']), O['ifName']) if not fn: return d def m1(key): return {num(s.split('.')[-1]): strval(v) for s, v in parse(walk(args, ip, O[key]), O[key])} def n1(key): return {num(s.split('.')[-1]): num(strval(v)) for s, v in parse(walk(args, ip, O[key]), O[key])} ifname = m1('ifName'); ifdescr = m1('ifDescr'); ifalias = m1('ifAlias') iftype = n1('ifType'); ifadmin = n1('ifAdmin'); ifoper = n1('ifOper') ifspeed = n1('ifSpeed'); ifhigh = n1('ifHighSpeed'); lastchg = n1('ifLastChange') conn = n1('ifConnector'); dup = n1('duplex') # PoE capability: standard RFC 3621 PSE table, plus Huawei's TruthValue # capability table. A row in the standard table means a PSE port exists; # Huawei reports true(1)/false(2) for each interface. This is capability, # not current PD power, so it stays stable when no device is attached. std_poe = parse(walk(args, ip, POE_STD), POE_STD) hw_poe = parse(walk(args, ip, POE_HW_JUDGE), POE_HW_JUDGE) poe = bool(std_poe) or any(num(v) == 1 for _, v in hw_poe) # FDB per port -> count and sample macs bpif = {num(s.split('.')[-1]): num(v) for s, v in parse(walk(args, ip, O['bpIf']), O['bpIf'])} st = {num(s.split('.')[-1]): num(v) for s, v in parse(walk(args, ip, O['fdbStatus']), O['fdbStatus'])} mac_cnt = collections.Counter(); mac_by_if = collections.defaultdict(list) for s, v in parse(walk(args, ip, O['fdbPort']), O['fdbPort']): port = num(v) if port is None: continue ifx = bpif.get(port) if ifx is None: continue mac_cnt[ifx] = mac_cnt.get(ifx, 0) + 1 if len(mac_by_if[ifx]) < 6: parts = s.split('.') macs = ':'.join('%02x' % int(p) for p in parts[-6:]) if len(parts) >= 6 else '' mac_by_if[ifx].append(macs) d.update({'ok': True, 'ifname': ifname, 'ifdescr': ifdescr, 'ifalias': ifalias, 'iftype': iftype, 'ifadmin': ifadmin, 'ifoper': ifoper, 'ifspeed': ifspeed, 'ifhigh': ifhigh, 'lastchg': lastchg, 'conn': conn, 'dup': dup, 'mac_cnt': dict(mac_cnt), 'mac_by_if': {k: v for k, v in mac_by_if.items()}, 'poe': poe}) return d def main(): hs = l2cfg.hosts(call, output=['hostid', 'name']) tasks = [] for h in hs: for i in h['interfaces']: if i.get('type') == '2': tasks.append((i['ip'], h['name'], i)); break res = [] with cf.ThreadPoolExecutor(max_workers=30) as ex: for r in ex.map(collect, tasks): res.append(r) ok = [r for r in res if r.get('ok')] rows = [] for r in ok: allif = set(r['ifname']) | set(r['ifdescr']) for ifx in allif: it = r['iftype'].get(ifx, 0) admin = r['ifadmin'].get(ifx); oper = r['ifoper'].get(ifx) sp = r['ifhigh'].get(ifx) if sp is None or sp == 0: s_bps = r['ifspeed'].get(ifx) or 0 sp = int(s_bps / 1000000) if s_bps else 0 rows.append([r['name'], r['ip'], str(ifx), r['ifname'].get(ifx, '') or r['ifdescr'].get(ifx, ''), r['ifalias'].get(ifx, ''), str(it), {1:'up',2:'down',3:'testing',4:'unknown',5:'dormant',6:'notPresent',7:'lowerLayerDown'}.get(admin, str(admin)), {1:'up',2:'down',3:'testing',4:'unknown',5:'dormant',6:'notPresent',7:'lowerLayerDown'}.get(oper, str(oper)), str(sp if sp is not None else ''), {1:'half',2:'full',3:'unknown'}.get(r['dup'].get(ifx), ''), str(r['conn'].get(ifx, '')), str(r['lastchg'].get(ifx, '')), str(r['mac_cnt'].get(ifx, 0)), ' '.join(r['mac_by_if'].get(ifx, [])[:3])]) with open('/tmp/port_status.tsv', 'w', encoding='utf-8') as f: f.write('switch_name\tswitch_ip\tifIndex\tifName\tifAlias\tifType\tadmin_status\toper_status\tspeed_mbps\tduplex\tconnector_present\tlast_change_ticks\tmac_count\tsample_macs\n') for row in rows: f.write('\t'.join(row) + '\n') with open('/tmp/poe_switches.tsv', 'w', encoding='utf-8') as f: f.write('switch_ip\tpoe\n') for r in ok: f.write('%s\t%s\n' % (r['ip'], '1' if r.get('poe') else '0')) poe_count = sum(1 for r in ok if r.get('poe')) print('poe_switches=%d' % poe_count) # summary phys = [r for r in rows if r[5] in ('6', '62', '117', '161', '117')] up = [r for r in rows if r[7] == 'up'] spd = collections.Counter(r[8] for r in up) with open('/tmp/port_status_summary.txt', 'w', encoding='utf-8') as f: f.write('switches_ok=%d/%d\n' % (len(ok), len(tasks))) f.write('total_if=%d\n' % len(rows)) f.write('oper_up=%d oper_down=%d\n' % (len(up), sum(1 for r in rows if r[7] == 'down'))) f.write('up_speed_dist=%s\n' % dict(spd)) f.write('up_with_mac=%d\n' % sum(1 for r in rows if r[7]=='up' and int(r[12] or 0) > 0)) print(open('/tmp/port_status_summary.txt').read()) if __name__ == '__main__': main()