I am trying to write a mock snmp server using pysnmp library in python. /

I start the server using,

python mock_snmp_server.py

python --version Python 3.11.4

I test using snmpget command,

snmpget -v2c -c public 127.0.0.1:1611 .1.3.6.1.4.1.318.1.1.26.9.2.4.1.5 Timeout: No Response from 127.0.0.1:1611.

I get these logs when testing snmpget:

 Received SNMP request from ('127.0.0.1', 65278)
 request message is :  Message:
 version=version-2c
 community=public
 data=PDUs:
   get-request=GetRequestPDU:
    request-id=1008009258
    error-status=noError
    error-index=0
    variable-bindings=VarBindList:
     VarBind:
      name=1.3.6.1.4.1.318.1.1.26.9.2.4.1.5
      =_BindValue:
       unSpecified=



 pdu is :  GetRequestPDU:
  get-request=GetRequestPDU:
    request-id=1008009258
    error-status=noError
    error-index=0
    variable-bindings=VarBindList:
     VarBind:
      name=1.3.6.1.4.1.318.1.1.26.9.2.4.1.5
      =_BindValue:
       unSpecified=



request id is :  1008009258


OID: .1.3.6.1.4.1.318.1.1.26.9.2.4.1.5, Response: 1

Error processing SNMP request: Component value is tag-incompatible: <Integer32 value object, tagSet <TagSet object, tags 0:0:2>, subtypeSpec <ConstraintsIntersection object, consts <ValueRangeConstraint object, consts -2147483648, 2147483647>>, payload [1]> vs <NamedTypes object, types <NamedType object, type name=<ObjectName schema object, tagSet <TagSet object, tags 0:0:6>>>, <NamedType object, type =<_BindValue schema object, tagSet=<TagSet object, untagged>, subtypeSpec=<ConstraintsIntersection object, consts <ValueSizeConstraint object, consts 1, 1>>, componentType=<NamedTypes object, types <NamedType object, type value=<ObjectSyntax schema object, tagSet=<TagSet object, untagged>, subtypeSpec=<ConstraintsIntersection object, consts <ValueSizeConstraint object, consts 1, 1>>, componentType=<NamedTypes object, types <NamedType object, type simple=<SimpleSyntax schema object, tagSet=<TagSet object, untagged>, subtypeSpec=<ConstraintsIntersection object, consts <ValueSizeConstraint object, consts 1, 1>>, componentType=<NamedTypes object, types <NamedType object, type integer-value=<Integer schema object, tagSet <TagSet object, tags 0:0:2>, subtypeSpec <ConstraintsIntersection object, consts <ValueRangeConstraint object, consts -2147483648, 2147483647>>>>, <NamedType object, type string-value=<OctetString schema object, tagSet <TagSet object, tags 0:0:4>, subtypeSpec <ConstraintsIntersection object, consts <ValueSizeConstraint object, consts 0, 65535>>, encoding iso-8859-1>>, <NamedType object, type objectID-value=<ObjectIdentifier schema object, tagSet <TagSet object, tags 0:0:6>>>>, sizeSpec=>>, <NamedType object, type application-wide=<ApplicationSyntax schema object, tagSet=<TagSet object, untagged>, subtypeSpec=<ConstraintsIntersection object, consts <ValueSizeConstraint object, consts 1, 1>>, componentType=<NamedTypes object, types <NamedType object, type ipAddress-value=<IpAddress schema object, tagSet <TagSet object, tags 64:0:0>, subtypeSpec <ConstraintsIntersection object, consts <ValueSizeConstraint object, consts 0, 65535>, <ValueSizeConstraint object, consts 4, 4>>, encoding iso-8859-1>>, <NamedType object, type counter-value=<Counter32 schema object, tagSet <TagSet object, tags 64:0:1>, subtypeSpec <ConstraintsIntersection object, consts <ValueRangeConstraint object, consts 0, 4294967295>>>>, <NamedType object, type timeticks-value=<TimeTicks schema object, tagSet <TagSet object, tags 64:0:3>, subtypeSpec <ConstraintsIntersection object, consts <ValueRangeConstraint object, consts 0, 4294967295>>>>, <NamedType object, type arbitrary-value=<Opaque schema object, tagSet <TagSet object, tags 64:0:4>, encoding iso-8859-1>>, <NamedType object, type big-counter-value=<Counter64 schema object, tagSet <TagSet object, tags 64:0:6>, subtypeSpec <ConstraintsIntersection object, consts <ValueRangeConstraint object, consts 0, 18446744073709551615>>>>, <NamedType object, type gauge32-value=<Gauge32 schema object, tagSet <TagSet object, tags 64:0:2>, subtypeSpec <ConstraintsIntersection object, consts <ValueRangeConstraint object, consts 0, 4294967295>>>>>, sizeSpec=>>>, sizeSpec=>>, <NamedType object, type unSpecified=<Null value object, tagSet <TagSet object, tags 0:0:5>, subtypeSpec <ConstraintsIntersection object, consts <SingleValueConstraint object, consts b''>>, encoding iso-8859-1, payload []>>, <NamedType object, type noSuchObject=<NoSuchObject value object, tagSet <TagSet object, tags 128:0:0>, subtypeSpec <ConstraintsIntersection object, consts <SingleValueConstraint object, consts b''>>, encoding iso-8859-1, payload [No Such Object c...ists at this OID]>>, <NamedType object, type noSuchInstance=<NoSuchInstance value object, tagSet <TagSet object, tags 128:0:1>, subtypeSpec <ConstraintsIntersection object, consts <SingleValueConstraint object, consts b''>>, encoding iso-8859-1, payload [No Such Instance...ists at this OID]>>, <NamedType object, type endOfMibView=<EndOfMibView value object, tagSet <TagSet object, tags 128:0:2>, subtypeSpec <ConstraintsIntersection object, consts <SingleValueConstraint object, consts b''>>, encoding iso-8859-1, payload [No more variable...in this MIB View]>>>, sizeSpec=>>>

This is the mock_snmp_server.py file

import asyncio
import json
from pysnmp.proto import rfc1901
from pysnmp.proto.api import v2c
from pysnmp.proto.rfc1902 import Integer32, ObjectName, Null
from pysnmp.proto.rfc1905 import ResponsePDU, VarBind, VarBindList
from pyasn1.codec.ber import decoder, encoder

MOCK_RESPONSES_FILE = "../test/snmp/mock_responses.json"
SNMP_PORT = 1611
COMMUNITY_STRING = "public"


class SNMPServer:
def __init__(self, config_file):
    self.config_file = config_file
    self.responses = self.load_responses()

def load_responses(self):
    try:
        with open(self.config_file, "r") as file:
            responses = json.load(file)
            print(f"Loaded {len(responses)} mock responses.")
            return responses
    except Exception as e:
        print(f"Error loading mock responses: {e}")
        return {}

def get_response(self, oid, method):
    if oid in self.responses:
        for response in self.responses[oid]["responses"]:
            if response["method"] == method:
                return response["value"]

    print(f"No response found for OID: {oid}, method: {method}")
    return 0

def handle_incoming_get_request(self, pdu):
    response_var_binds = VarBindList()

    for var_bind in pdu["variable-bindings"]:
        oid = var_bind["name"]
        oid_str = "." + ".".join(map(str, oid))

        response_value = self.get_response(oid_str, "GetRequest")
        print(f"OID: {oid_str}, Response: {response_value}")

        response_var_bind = VarBind()
        response_var_bind.setComponentByPosition(0, ObjectName(oid))
        response_var_bind.setComponentByPosition(1, Integer32(response_value))

        response_var_binds.append(response_var_bind)

    return response_var_binds

def handle_incoming_packet(self, data):
    try:
        request_msg, _ = decoder.decode(data, asn1Spec=rfc1901.Message())
        print("request message is : ", request_msg)
        print("\n")

        pdu = request_msg["data"].getComponent()
        print("pdu is : ", pdu)
        print("\n")
        request_id = pdu["request-id"]
        print("request id is : ", request_id)
        print("\n")

        if pdu.tagSet == v2c.GetRequestPDU.tagSet:
            response_var_binds = self.handle_incoming_get_request(pdu)

            response_pdu = ResponsePDU()
            response_pdu.setComponentByPosition(0, request_id)
            response_pdu.setComponentByPosition(1, "noError")
            response_pdu.setComponentByPosition(2, 0)
            response_pdu.setComponentByPosition(3, response_var_binds)

            response_msg = rfc1901.Message()
            response_msg.setComponentByPosition(0, 1)
            response_msg.setComponentByPosition(1, COMMUNITY_STRING)
            response_msg.setComponentByPosition(2, response_pdu)

            encoded_response = encoder.encode(response_msg)
            print(f"Sending SNMP Response: {encoded_response.hex()}")
            return encoded_response
        else:
            print("Unsupported PDU type")
            return None
    except Exception as e:
        print(f"Error processing SNMP request: {e}")
        return None


class MockSNMPProtocol(asyncio.DatagramProtocol):
def __init__(self, server):
    self.server = server

def connection_made(self, transport):
    self.transport = transport
    print(f"Mock SNMP Server running on UDP {SNMP_PORT}...")

def datagram_received(self, data, addr):
    print(f"Received SNMP request from {addr}")
    response = self.server.handle_incoming_packet(data)
    if response:
        self.transport.sendto(response, addr)


async def main():
  loop = asyncio.get_running_loop()
  snmp_server = SNMPServer(MOCK_RESPONSES_FILE)
  listen = loop.create_datagram_endpoint(
    lambda: MockSNMPProtocol(snmp_server),
    local_addr=("127.0.0.1", SNMP_PORT),
  )
  transport, protocol = await listen

  try:
    await asyncio.Future()
  except asyncio.CancelledError:
    transport.close()


if __name__ == "__main__":
  asyncio.run(main())

I have a json file to define mock responses shown below.

 {
 ".1.3.6.1.4.1.318.1.1.26.9.2.4.1.5": {    
     "pduType": "abc",
     
     "powerSocket": 1,
     "OID_format": "<abcOID>.<PowerSocket>",
     "responses": [
        {
            "method": "GetRequest",
            "value": 1
        }
     ]
   } 
}