Skip to content

SNMP Python ​

py
import socket

def encode_length(n):
    if n < 0x80:
        return bytes([n])
    enc = []
    while n > 0:
        enc.insert(0, n & 0xFF)
        n >>= 8
    return bytes([0x80 | len(enc)]) + bytes(enc)

def encode_tlv(tag, value):
    return bytes([tag]) + encode_length(len(value)) + value

def encode_integer(v):
    body = v.to_bytes((v.bit_length() // 8) + 1, 'big', signed=True) if v else b'\x00'
    return encode_tlv(0x02, body)

def encode_octet_string(s):
    return encode_tlv(0x04, s.encode() if isinstance(s, str) else s)

def encode_null():
    return encode_tlv(0x05, b'')

def encode_oid(oid):
    parts = [int(x) for x in oid.strip('.').split('.')]
    body = bytes([parts[0]*40 + parts[1]])
    for p in parts[2:]:
        chunk = [p & 0x7F]
        p >>= 7
        while p:
            chunk.insert(0, (p & 0x7F) | 0x80)
            p >>= 7
        body += bytes(chunk)
    return encode_tlv(0x06, body)

def encode_seq(tag, *items):
    return encode_tlv(tag, b''.join(items))

def read_tlv(data, pos):
    tag = data[pos]
    length = data[pos+1]
    pos += 2
    if length & 0x80:
        nbytes = length & 0x7F
        length = int.from_bytes(data[pos:pos+nbytes], 'big')
        pos += nbytes
    value = data[pos:pos+length]
    return tag, value, pos+length

def decode_oid(data):
    first = data[0]
    parts = [first // 40, first % 40]
    i, val = 1, 0
    while i < len(data):
        b = data[i]
        val = (val << 7) | (b & 0x7F)
        if not (b & 0x80):
            parts.append(val)
            val = 0
        i += 1
    return '.'.join(str(p) for p in parts)

def snmp_get_multi(ip, oids, community='public', port=161, timeout=3, version=1):
    varbinds = b''.join(encode_seq(0x30, encode_oid(oid), encode_null()) for oid in oids)
    pdu = encode_seq(0xA0,
        encode_integer(1),
        encode_integer(0),
        encode_integer(0),
        encode_seq(0x30, varbinds)
    )
    packet = encode_seq(0x30,
        encode_integer(version),   # 0 = SNMPv1, 1 = SNMPv2c
        encode_octet_string(community),
        pdu
    )
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.settimeout(timeout)
    s.sendto(packet, (ip, port))
    data, _ = s.recvfrom(4096)
    s.close()

    # Parsing générique
    _, content, _ = read_tlv(data, 0)
    p = 0
    _, _, p = read_tlv(content, p)          # version
    _, _, p = read_tlv(content, p)          # community
    _, pdu_body, p = read_tlv(content, p)   # PDU
    pp = 0
    _, _, pp = read_tlv(pdu_body, pp)       # request-id
    _, _, pp = read_tlv(pdu_body, pp)       # error-status
    _, _, pp = read_tlv(pdu_body, pp)       # error-index
    _, varbindlist, pp = read_tlv(pdu_body, pp)

    results = {}
    vp = 0
    while vp < len(varbindlist):
        _, vb, vp = read_tlv(varbindlist, vp)
        ip2 = 0
        _, oid_bytes, ip2 = read_tlv(vb, ip2)
        val_tag, val_bytes, ip2 = read_tlv(vb, ip2)
        oid_str = decode_oid(oid_bytes)
        if val_tag == 0x02:
            value = int.from_bytes(val_bytes, 'big', signed=True)
        elif val_tag == 0x04:
            value = val_bytes.decode(errors='replace')
        else:
            value = val_bytes.hex()
        results[oid_str] = value
    return results

oids = {
    '1.3.6.1.2.1.33.1.2.4.0':     'Battery (%)',
    '1.3.6.1.2.1.33.1.2.5.0':     'Battery Voltage (V x10)',
    '1.3.6.1.2.1.33.1.3.3.1.3.1': 'Input Voltage (V)',
    '1.3.6.1.2.1.33.1.3.3.1.2.1': 'Input Frequency (Hz x10)',
    '1.3.6.1.2.1.33.1.4.4.1.2.1': 'Output Voltage (V)',
    '1.3.6.1.2.1.33.1.4.2.0':     'Output Frequency (Hz x10)',
    '1.3.6.1.2.1.33<|fim_middle|>
}

result = snmp_get_multi('<UPS_IP>', list(oids.keys()))
for oid, val in result.items():
    print(f"{oids.get(oid, oid)}: {val}")