zabbix-deploy/collector/port_status.py

#!/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()