{"id":11032,"date":"2024-07-05T12:59:00","date_gmt":"2024-07-05T17:59:00","guid":{"rendered":"https:\/\/www.rushworth.us\/lisa\/?p=11032"},"modified":"2024-07-05T12:59:00","modified_gmt":"2024-07-05T17:59:00","slug":"snmp-simulator","status":"publish","type":"post","link":"https:\/\/www.rushworth.us\/lisa\/?p=11032","title":{"rendered":"SNMP Simulator"},"content":{"rendered":"\n

Background<\/h2>\n\n\n\n

As communication between development and production platforms is limited for security and data integrity reasons, this creates a challenge when testing changes in development: we cannot access \u201creal world\u201d data with which to perform tests. Having a limited set of data in development means testing may not illuminate issues that occur at high volume or on a large scale.<\/p>\n\n\n\n

Solution<\/h2>\n\n\n\n

While limiting communication between the prod and dev systems is reasonable, it would be beneficial to be able to replay production-like data within our development systems for testing purposes. While it is not cost effective to buy large network devices with thousands of interfaces for testing, the Python module snmpsim<\/a> provides \u201ccanned responses\u201d that simulate real devise on the production network. For simplicity, I have a bash script that launches the SNMP responder.<\/p>\n\n\n\n

server03:snmpsim # cat ..\/_playback.sh<\/p>\n\n\n\n

#!\/bin\/bash<\/p>\n\n\n\n

snmpsimd.py –data-dir=\/opt\/snmp\/snmpsim\/data –cache-dir=\/opt\/snmp\/snmpsim\/cache –agent-udpv4-endpoint=0.0.0.0:161 –process-user=ljrsnmp –process-group=ljrsnmp<\/p>\n\n\n\n

This responder will replay data stored in the directory \/opt\/snmp\/snmpsim\/data \u2013 any file ending in snmprec will be included in the response, and the filename prior to .snmprec is the community string to access the response data. E.G. public.snmprec is the data for the public community string<\/p>\n\n\n\n

The response files are in the format OID|TAG|VALUE where OID is the OID number of the SNMP object, TAG is an integer defined at https:\/\/pypi.org\/project\/snmpsim\/0.2.3\/<\/a><\/p>\n\n\n\n

Valid tag values and their corresponding ASN.1\/SNMP types are:<\/p>\n\n\n\n

ASN.1\/SNMP Type<\/strong><\/td>Tag Value<\/strong><\/td><\/tr>
Integer32<\/td>2<\/td><\/tr>
Octet String<\/td>4<\/td><\/tr>
Null<\/td>5<\/td><\/tr>
Object Identifier<\/td>6<\/td><\/tr>
IP Address<\/td>64<\/td><\/tr>
Counter32<\/td>65<\/td><\/tr>
Gauge32<\/td>66<\/td><\/tr>
Time Ticks<\/td>67<\/td><\/tr>
Opaque<\/td>68<\/td><\/tr>
Counter65<\/td>70<\/td><\/tr><\/tbody><\/table><\/figure>\n\n\n\n

And the value is the data to be returned for the OID object. As an example:<\/p>\n\n\n\n

1.3.6.1.2.1.1.3.0|67|2293092270<\/p>\n\n\n\n

1.3.6.1.2.1.1.3.0 is the sysUpTime, the data type is TimeTicks, and the system up time is 2293092270 hundredths of a second. Or 6375 hours, 20 minutes, and 24 seconds.<\/p>\n\n\n\n

Items within the response file need to be listed in ascending order.<\/p>\n\n\n\n

Generating Response Data<\/h2>\n\n\n\n

There are two methods for creating the data provided to an SNMP GET request. A response file can be created manually, populated with OID objects that should be included in the response as well as sample data. Alternatively, a network trace can be gathered from the production network and parsed to create the response file.<\/p>\n\n\n\n

Manually Generated Response File<\/h3>\n\n\n\n

While you can literally type data into a response file, but it is far easier to use a script to generate sample data. \/opt\/snmp\/snmpsim\/_genData.py is an example of creating a response file for about 1,000 interfaces<\/p>\n\n\n

\nfrom datetime import datetime\nimport random\n\niRangeMax = 1000\n\ndictTags = {'Integer': '2', 'OctetString': '4', 'NULL': '5', 'ObjectIdentifier': '6', 'IPAddress': '64', 'Counter32': '65', 'Gauge32': '66', 'TimeTicks': '67', 'Opaque': '68','Counter64': '70'}  # Valid tags per https:\/\/pypi.org\/project\/snmpsim\/0.2.3\/\n\ntoday = datetime.now()\n\niftable_snmp_objects = [\n    ('1.3.6.1.2.1.2.2.1.1', 'Integer', lambda i: i),  # ifIndex\n    ('1.3.6.1.2.1.2.2.1.2', 'OctetString', lambda i: f"SampleInterface{i}"),  # ifDescr\n    ('1.3.6.1.2.1.2.2.1.3', 'Integer', lambda i: 6),  # ifType\n    ('1.3.6.1.2.1.2.2.1.4', 'Integer', lambda i: 1500),  # ifMtu\n    ('1.3.6.1.2.1.2.2.1.5', 'Gauge32', lambda i: 100000000),  # ifSpeed\n    ('1.3.6.1.2.1.2.2.1.6', 'OctetString', lambda i: f"00:00:00:00:{format(i, '02x')[:2]}:{format(i, '02x')[-2:]}"),  # ifPhysAddress\n    ('1.3.6.1.2.1.2.2.1.7', 'Integer', lambda i: 1),  # ifAdminStatus\n    ('1.3.6.1.2.1.2.2.1.8', 'Integer', lambda i: 1),  # ifOperStatus\n    ('1.3.6.1.2.1.2.2.1.9', 'TimeTicks', lambda i: int((datetime.now() - datetime(2024, random.randint(1, today.month), random.randint(1, today.day))).total_seconds()) * 100),  # ifLastChange\n    ('1.3.6.1.2.1.2.2.1.10', 'Counter32', lambda i: random.randint(3, i*50000)),  # ifInOctets\n    ('1.3.6.1.2.1.2.2.1.11', 'Counter32', lambda i: random.randint(3, i*50000)),  # ifInUcastPkts\n    ('1.3.6.1.2.1.2.2.1.12', 'Counter32', lambda i: random.randint(0, 80)),  # ifInNUcastPkts\n    ('1.3.6.1.2.1.2.2.1.13', 'Counter32', lambda i: random.randint(0, 80)),  # ifInDiscards\n    ('1.3.6.1.2.1.2.2.1.14', 'Counter32', lambda i: random.randint(0, 80)),  # ifInErrors\n    ('1.3.6.1.2.1.2.2.1.15', 'Counter32', lambda i: random.randint(3, i*50000)),  # ifInUnknownProtos\n    ('1.3.6.1.2.1.2.2.1.16', 'Counter32', lambda i: random.randint(3, i*50000)),  # ifOutOctets\n    ('1.3.6.1.2.1.2.2.1.17', 'Counter32', lambda i: random.randint(3, i*50000)),  # ifOutUcastPkts\n    ('1.3.6.1.2.1.2.2.1.18', 'Counter32', lambda i: random.randint(3, i*50000)),  # ifOutNUcastPkts\n    ('1.3.6.1.2.1.2.2.1.19', 'Counter32', lambda i: random.randint(0, 80)),  # ifOutDiscards\n    ('1.3.6.1.2.1.2.2.1.20', 'Counter32', lambda i: random.randint(0, 80)),  # ifOutErrors\n]\n\nifxtable_snmp_objects = [\n    ('1.3.6.1.2.1.31.1.1.1.1', 'OctetString', lambda i: f"SampleInterface{i}"),  # ifName\n    ('1.3.6.1.2.1.31.1.1.1.15', 'Gauge32', lambda i: "100"),  # ifHighSpeed\n    ('1.3.6.1.2.1.31.1.1.1.6', 'Counter32', lambda i: random.randint(3, i*50000)),  # ifHCInOctets\n    ('1.3.6.1.2.1.31.1.1.1.10', 'Counter32', lambda i: random.randint(3, i*60000)),  # ifHCOutOctets\n]\n\n# Print IFTable data\nfor oid_base, tag_type, value_func in iftable_snmp_objects:\n    for i in range(1, iRangeMax+1):\n        value = value_func(i)\n        print(f"{oid_base}.{i}|{dictTags.get(tag_type)}|{value}")\n\n# IP-MIB objects for managing IP addressing\n# ipAdEntAddr: The IP address to which this entry's addressing information pertains\nprint(f"1.3.6.1.2.1.4.20.1.1|{dictTags.get('IPAddress')}|10.5.5.5")\n\n# ipAdEntIfIndex: The index value which uniquely identifies the interface to which this entry is applicable\nprint(f"1.3.6.1.2.1.4.20.1.2|{dictTags.get('OctetString')}|1")\n\n# ipAdEntNetMask: The subnet mask associated with the IP address of this entry\nprint(f"1.3.6.1.2.1.4.20.1.3|{dictTags.get('OctetString')}|255.255.255.0")\n\n# hrSWRunIndex: An index uniquely identifying a row in the hrSWRun table\nprint(f"1.3.6.1.2.1.25.4.2.1.1.1|{dictTags.get('Integer')}|1")\n\n# hrSWRunName: The name of the software running on this device\nprint(f"1.3.6.1.2.1.25.4.2.1.2.1|{dictTags.get('OctetString')}|LJRSNMPAgent")\n# hrSWRunID: The product ID of the software running on this device\nprint(f"1.3.6.1.2.1.25.4.2.1.3.1|{dictTags.get('ObjectIdentifier')}|1.3.6.1.4.1.25709.55")\n\n# hrSWRunPath: The path of the software running on this device\nprint(f"1.3.6.1.2.1.25.4.2.1.4.1|{dictTags.get('OctetString')}|\/opt\/snmp\/snmpsim\/_agent.sh")\n\n# hrSWRunParameters: Operational parameters for the software running on this device\nprint(f"1.3.6.1.2.1.25.4.2.1.5.1|{dictTags.get('OctetString')}|-L")\n\n# hrSWRunType: The type of software running (e.g., operating system, application)\nprint(f"1.3.6.1.2.1.25.4.2.1.6.1|{dictTags.get('Integer')}|4")\n\n# hrSWRunStatus: The status of this software (running, runnable, notRunnable, invalid)\nprint(f"1.3.6.1.2.1.25.4.2.1.7.1|{dictTags.get('Integer')}|1")\n\n\nfor oid_base, tag_type, value_func in ifxtable_snmp_objects:\n    for i in range(1, iRangeMax+1):\n        value = value_func(i)\n        print(f"{oid_base}.{i}|{dictTags.get(tag_type)}|{value}")\n\n<\/pre><\/div>\n\n\n
Network Capture<\/h3>\n\n\n\n
Even better, parse a network capture file.<\/p>\n\n\n\n
Capture Data<\/h4>\n\n\n\n
On the server that gathers SNMP data from the host we want to simulate, use a network capture utility to gather the SNMP communication between the server and the desired device.<\/p>\n\n\n\n
tcpdump -i <interface> -w <filename>.pcap<\/p>\n\n\n\n
E.G. to record the communication with 10.5.171.114<\/p>\n\n\n\n
tcpdump ‘host 10.5.171.114 and (tcp port 161 or tcp port 162 or udp port 161 or udp port 162)’ -w \/tmp\/ar.pcap<\/p>\n\n\n\n
Note \u2013 there Is no benefit to capturing more than one cycle of SNMP responses. If data is captured immediately, that means the devices were in the middle of a cycle. End the capture and start a new one shortly. There should be no packets captured for a bit, then packets during the SNMP polling cycle, and then another pause until the next cycle.<\/p>\n\n\n\n
Parsing The Capture Data Into A Response File<\/h4>\n\n\n\n
The following script parses the capture file into an snmprec response file \u2013 note, I needed to use 2.6.0rc1 of scapy to parse SNMP data. The 2.5.0 release version failed to parse most<\/em> of the packets which I believe is related to https:\/\/github.com\/secdev\/scapy\/issues\/3900<\/a><\/p>\n\n\n
\nfrom scapy.all import rdpcap, SNMP\nfrom scapy.layers.inet import UDP\nfrom scapy.packet import Raw\nfrom scapy.layers.snmp import SNMP, SNMPvarbind, SNMPresponse, SNMPbulk\nfrom scapy.all import conf, load_layer\nfrom scapy.utils import hexdump\n\nfrom scapy.all import UDP, load_contrib\nfrom scapy.packet import bind_layers\n\nimport os\nfrom datetime import datetime\nimport argparse\n\n# Ensure Scapy's SNMP contributions are loaded\nload_contrib("snmp")\n\ndef sort_by_oid(listSNMPResponses):\n    """\n    Sorts a list of "OID|TAG|Value" strings by the OID numerically and hierarchically.\n\n    :param listSNMPResponses: A list of "OID|TAG|Value" strings.\n    :return: A list of "OID|TAG|Value" strings sorted by OID.\n    """\n    # Split each element into a tuple of (OID list, original string), converting OID to integers for proper comparison\n    oid_tuples = [(list(map(int, element.split('|')[0].split('.'))), element) for element in listSNMPResponses]\n\n    # Sort the list of tuples by the OID part (the list of integers)\n    sorted_oid_tuples = sorted(oid_tuples, key=lambda x: x[0])\n\n    # Extract the original strings from the sorted list of tuples\n    sorted_listSNMPResponses = [element[1] for element in sorted_oid_tuples]\n\n    return sorted_listSNMPResponses\n\nparser = argparse.ArgumentParser(description='This script converts an SNMP packet capture into a snmpsim response file')\nparser.add_argument('--filename', '-f', help='The capture file to process', required=True)\n\nargs = parser.parse_args()\nstrFullCaptureFilePath = args.filename\nstrCaptureFilePath, strCaptureFileName = os.path.split(strFullCaptureFilePath)\n\n\n# Valid tags per https:\/\/pypi.org\/project\/snmpsim\/0.2.3\/\ndictTags = {'ASN1_INTEGER': '2', 'ASN1_STRING': '4', 'ASN1_NULL': '5', 'ASN1_OID': '6', 'ASN1_IPADDRESS': '64', 'ASN1_COUNTER32': '65', 'ASN1_GAUGE32': '66', 'ASN1_TIME_TICKS': '67', 'Opaque': '68','ASN1_COUNTER64': '70'}\n\nlistSNMPResponses = []\nlistSNMPResponses.append("1.3.6.1.2.1.25.4.2.1.1.1|2|1")\nlistSNMPResponses.append("1.3.6.1.2.1.25.4.2.1.2.1|4|LJRSNMPAgent")\nlistSNMPResponses.append("1.3.6.1.2.1.25.4.2.1.3.1|6|1.3.6.1.4.1.25709.55")\nlistSNMPResponses.append("1.3.6.1.2.1.25.4.2.1.4.1|4|\/opt\/snmp\/snmpsim\/_agent.sh")\nlistSNMPResponses.append("1.3.6.1.2.1.25.4.2.1.5.1|4|-L")\nlistSNMPResponses.append("1.3.6.1.2.1.25.4.2.1.6.1|2|4")\nlistSNMPResponses.append("1.3.6.1.2.1.25.4.2.1.7.1|2|1")\ni = 0\n\nif True:\n    packets = rdpcap(strFullCaptureFilePath)\n    # Packets are zero indexed, so packet 1 in script is packet 2 in Wireshark GUI\n    #for i in range(0,4):\n    for packet in packets:\n        print(f"Working on packet {i}")\n        i = i + 1\n        if SNMP in packet:\n            snmp_layer = packet[SNMP]\n            if isinstance(packet[SNMP].PDU,SNMPresponse):\n                snmp_response = snmp_layer.getfield_and_val('PDU')[1]\n                if hasattr(snmp_response, 'varbindlist') and snmp_response.varbindlist is not None:\n                    for varbind in snmp_response.varbindlist:\n                        strOID = varbind.oid.val if hasattr(varbind.oid, 'val') else str(varbind.oid)\n                        strValue = varbind.value.val if hasattr(varbind.value, 'val') else str(varbind.value)\n                        strType = type(varbind.value).__name__\n                        if dictTags.get(strType):\n                            iType = dictTags.get(strType)\n                        else:\n                            iType = strType\n\n                        if isinstance(strValue, bytes):\n                            print(f"Decoding {strValue}")\n                            strValue = strValue.decode('utf-8',errors='ignore')\n\n                        print(f"OID: {strOID}, Type: {strType}, Tag: {iType}, Value: {strValue}")\n                        listSNMPResponses.append(f"{strOID}|{iType}|{strValue}")\n            else:\n                print(f"Not a response -- type is {type(packet[SNMP].PDU)}")\n        elif Raw in packet:\n            print(f"I have a raw packet at {i}")\n        else:\n            print(dir(packet))\n            print(f"No SNMP or Raw in {i}: {packet}")\n\n# Sort by OID numbers\nlistSortedSNMPResponses = sort_by_oid(listSNMPResponses)\nf = open(f'\/opt\/snmp\/snmpsim\/data\/{datetime.now().strftime("%Y%m%d")}-{strCaptureFileName.rsplit(".", 1)[0]}.deactivated', "w")\nfor strSNMPResponse in listSortedSNMPResponses:\n    print(strSNMPResponse)\n    f.write(strSNMPResponse)\n    f.write("\\n")\nf.close()\n\n<\/pre><\/div>\n\n\n
This will create an snmpsim response file at \/opt\/snmp\/snmpsim\/data named as the capture file prefixed with the current year, month, and date. I.E. My ar.cap file results are \/opt\/snmp\/snmpsim\/data\/20240705-ar.deactivated \u2013 you can then copy the file to whatever community string you want \u2013 cp 20240705-ar.deactivated CommunityString.snmprec<\/p>\n\n\n\n
<\/p>\n","protected":false},"excerpt":{"rendered":"
Background As communication between development and production platforms is limited for security and data integrity reasons, this creates a challenge when testing changes in development: we cannot access \u201creal world\u201d data with which to perform tests. Having a limited set of data in development means testing may not illuminate issues that occur at high volume …<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1945,30],"tags":[664,2028],"class_list":["post-11032","post","type-post","status-publish","format-standard","hentry","category-python","category-system-administration","tag-python","tag-snmp"],"_links":{"self":[{"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/posts\/11032","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=11032"}],"version-history":[{"count":1,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/posts\/11032\/revisions"}],"predecessor-version":[{"id":11033,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/posts\/11032\/revisions\/11033"}],"wp:attachment":[{"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=11032"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=11032"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=11032"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}