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}")