from flask import Flask, request, jsonify, render_template_string
from pysnmp.hlapi import *
app = Flask(__name__)
def snmp_get(ip, community, oid):
iterator = getCmd(
SnmpEngine(),
CommunityData(community, mpModel=1), # mpModel=1 for SNMP v2c
UdpTransportTarget((ip, 161)),
ContextData(),
ObjectType(ObjectIdentity(oid))
)
errorIndication, errorStatus, errorIndex, varBinds = next(iterator)
if errorIndication:
return {"error": str(errorIndication)}, 500
elif errorStatus:
return {"error": str(errorStatus.prettyPrint())}, 500
else:
for varBind in varBinds:
return {"value": varBind.prettyPrint().split("=")[-1].strip()}
def snmp_set(ip, community, oid, value, value_type):
type_map = {
'Integer': Integer,
'OctetString': OctetString,
'IpAddress': IpAddress,
'Counter32': Counter32,
'Gauge32': Gauge32,
'String': String,
'TimeTicks': TimeTicks
}
if value_type not in type_map:
return {"error": f"Unsupported value type: {value_type}"}, 400
data_type = type_map[value_type]
iterator = setCmd(
SnmpEngine(),
CommunityData(community, mpModel=1),
UdpTransportTarget((ip, 161)),
ContextData(),
ObjectType(ObjectIdentity(oid), data_type(value))
)
errorIndication, errorStatus, errorIndex, varBinds = next(iterator)
if errorIndication:
return {"error": str(errorIndication)}, 500
elif errorStatus:
return {"error": str(errorStatus.prettyPrint())}, 500
else:
return {"status": "Set request successful"}
def snmp_get_table(ip, community, base_oid):
table = []
tableBetter = {}
iterator = nextCmd(
SnmpEngine(),
CommunityData(community, mpModel=1),
UdpTransportTarget((ip, 161)),
ContextData(),
ObjectType(ObjectIdentity(base_oid)),
lexicographicMode=False
)
for errorIndication, errorStatus, errorIndex, varBinds in iterator:
if errorIndication:
return {"error": str(errorIndication)}, 500
elif errorStatus:
return {"error": str(errorStatus.prettyPrint())}, 500
else:
for varBind in varBinds:
tableBetter[str(varBind[0])] = str(varBind[1].prettyPrint())
return tableBetter
@app.route('/')
def index():
# Main HTML page with a simple web UI
return render_template_string("""
SNMP Web Interface
SNMP Web Interface
SNMP TABLE
""")
@app.route('/snmp/table', methods=['GET'])
def api_snmp_get_table():
ip = request.args.get('ip')
community = request.args.get('community', 'public')
base_oid = request.args.get('base_oid')
if not ip or not base_oid:
return {"error": "Missing required parameters: ip, base_oid"}, 400
result = snmp_get_table(ip, community, base_oid)
return jsonify(result)
@app.route('/snmp/set', methods=['POST'])
def api_snmp_set():
data = request.json
ip = data.get('ip')
community = data.get('community', 'public')
oid = data.get('oid')
value = data.get('value')
value_type = data.get('value_type', 'OctetString')
if not ip or not oid or value is None:
return {"error": "Missing required parameters: ip, oid, value"}, 400
result = snmp_set(ip, community, oid, value, value_type)
return jsonify(result)
class PDU:
def __init__(self, hostName, portNumber, community):
self.hostName = hostName
self.community = "t33cHm3"
self.port = 161
def getHost(self):
return self.hostName
def setHost(self, hostname):
self.hostName = hostName
return self.hostName
class PDUPort:
def __init__(self, pdu, portNumber, portName, powerState):
def setPortName(self, portName):
self.portName = portName
try:
snmp_set(self, self.community, , self.name, 'string')
else:
return False
return True
def setPDU(self, pdu):
self.pdu = pdu
return pdu.hostname
def setPowerState(self, powerState):
self.powerState = powerState
return self.powerState
def set
def getPortName(self):
return self.portName
def getPortNumber(self):
return self.portNumber
def getPowerState(self):
return self.powerState
if __name__ == '__main__':
# app.run(host='0.0.0.0', port=5032
table = snmp_get_table("apc-pdu-01.dezendorf.net", "t33cHm3", "1.3.6.1.4.1.318.1.1.4.4.2.1")
for k in table:
k1 = k.replace("1.3.6.1.4.1.318.1.1.4.4.2.1.", "")
print(f"{k1}: {table[k]}")