# SPDX-License-Identifier: GPL-2.0-only

# This file is part of Scapy

# See https://scapy.net/ for more information

# Copyright (C) Philippe Biondi <[email protected]>

"""

SNMP (Simple Network Management Protocol).

"""

from scapy.packet import bind_layers, bind_bottom_up

from scapy.asn1packet import ASN1_Packet

from scapy.asn1fields import ASN1F_INTEGER, ASN1F_IPADDRESS, ASN1F_OID, \

ASN1F_SEQUENCE, ASN1F_SEQUENCE_OF, ASN1F_STRING, ASN1F_TIME_TICKS, \

ASN1F_enum_INTEGER, ASN1F_field, ASN1F_CHOICE, ASN1F_optional, ASN1F_NULL

from scapy.asn1.asn1 import ASN1_Class_UNIVERSAL, ASN1_Codecs, ASN1_NULL, \

ASN1_SEQUENCE

from scapy.asn1.ber import BERcodec_SEQUENCE

from scapy.sendrecv import sr, sr1

from scapy.volatile import RandShort, IntAutoTime

from scapy.layers.inet import UDP, IP, ICMP

# Import needed to initialize conf.mib

from scapy.asn1.mib import conf # noqa: F401

##########

# SNMP #

##########

# [ ASN1 class ] #

class ASN1_Class_SNMP(ASN1_Class_UNIVERSAL):

name = "SNMP"

PDU_GET = 0xa0

PDU_NEXT = 0xa1

PDU_RESPONSE = 0xa2

PDU_SET = 0xa3

PDU_TRAPv1 = 0xa4

PDU_BULK = 0xa5

PDU_INFORM = 0xa6

PDU_TRAPv2 = 0xa7

class ASN1_SNMP_PDU_GET(ASN1_SEQUENCE):

tag = ASN1_Class_SNMP.PDU_GET

class ASN1_SNMP_PDU_NEXT(ASN1_SEQUENCE):

tag = ASN1_Class_SNMP.PDU_NEXT

class ASN1_SNMP_PDU_RESPONSE(ASN1_SEQUENCE):

tag = ASN1_Class_SNMP.PDU_RESPONSE

class ASN1_SNMP_PDU_SET(ASN1_SEQUENCE):

tag = ASN1_Class_SNMP.PDU_SET

class ASN1_SNMP_PDU_TRAPv1(ASN1_SEQUENCE):

tag = ASN1_Class_SNMP.PDU_TRAPv1

class ASN1_SNMP_PDU_BULK(ASN1_SEQUENCE):

tag = ASN1_Class_SNMP.PDU_BULK

class ASN1_SNMP_PDU_INFORM(ASN1_SEQUENCE):

tag = ASN1_Class_SNMP.PDU_INFORM

class ASN1_SNMP_PDU_TRAPv2(ASN1_SEQUENCE):

tag = ASN1_Class_SNMP.PDU_TRAPv2

# [ BER codecs ] #

class BERcodec_SNMP_PDU_GET(BERcodec_SEQUENCE):

tag = ASN1_Class_SNMP.PDU_GET

class BERcodec_SNMP_PDU_NEXT(BERcodec_SEQUENCE):

tag = ASN1_Class_SNMP.PDU_NEXT

class BERcodec_SNMP_PDU_RESPONSE(BERcodec_SEQUENCE):

tag = ASN1_Class_SNMP.PDU_RESPONSE

class BERcodec_SNMP_PDU_SET(BERcodec_SEQUENCE):

tag = ASN1_Class_SNMP.PDU_SET

class BERcodec_SNMP_PDU_TRAPv1(BERcodec_SEQUENCE):

tag = ASN1_Class_SNMP.PDU_TRAPv1

class BERcodec_SNMP_PDU_BULK(BERcodec_SEQUENCE):

tag = ASN1_Class_SNMP.PDU_BULK

class BERcodec_SNMP_PDU_INFORM(BERcodec_SEQUENCE):

tag = ASN1_Class_SNMP.PDU_INFORM

class BERcodec_SNMP_PDU_TRAPv2(BERcodec_SEQUENCE):

tag = ASN1_Class_SNMP.PDU_TRAPv2

# [ ASN1 fields ] #

class ASN1F_SNMP_PDU_GET(ASN1F_SEQUENCE):

ASN1_tag = ASN1_Class_SNMP.PDU_GET

class ASN1F_SNMP_PDU_NEXT(ASN1F_SEQUENCE):

ASN1_tag = ASN1_Class_SNMP.PDU_NEXT

class ASN1F_SNMP_PDU_RESPONSE(ASN1F_SEQUENCE):

ASN1_tag = ASN1_Class_SNMP.PDU_RESPONSE

class ASN1F_SNMP_PDU_SET(ASN1F_SEQUENCE):

ASN1_tag = ASN1_Class_SNMP.PDU_SET

class ASN1F_SNMP_PDU_TRAPv1(ASN1F_SEQUENCE):

ASN1_tag = ASN1_Class_SNMP.PDU_TRAPv1

class ASN1F_SNMP_PDU_BULK(ASN1F_SEQUENCE):

ASN1_tag = ASN1_Class_SNMP.PDU_BULK

class ASN1F_SNMP_PDU_INFORM(ASN1F_SEQUENCE):

ASN1_tag = ASN1_Class_SNMP.PDU_INFORM

class ASN1F_SNMP_PDU_TRAPv2(ASN1F_SEQUENCE):

ASN1_tag = ASN1_Class_SNMP.PDU_TRAPv2

# [ SNMP Packet ] #

SNMP_error = {0: "no_error",

1: "too_big",

2: "no_such_name",

3: "bad_value",

4: "read_only",

5: "generic_error",

6: "no_access",

7: "wrong_type",

8: "wrong_length",

9: "wrong_encoding",

10: "wrong_value",

11: "no_creation",

12: "inconsistent_value",

13: "resource_unavailable",

14: "commit_failed",

15: "undo_failed",

16: "authorization_error",

17: "not_writable",

18: "inconsistent_name",

}

SNMP_trap_types = {0: "cold_start",

1: "warm_start",

2: "link_down",

3: "link_up",

4: "auth_failure",

5: "egp_neigh_loss",

6: "enterprise_specific",

}

class SNMPvarbind(ASN1_Packet):

ASN1_codec = ASN1_Codecs.BER

ASN1_root = ASN1F_SEQUENCE(

ASN1F_OID("oid", "1.3"),

ASN1F_optional(

ASN1F_field("value", ASN1_NULL(0))

),

# exceptions in responses

ASN1F_optional(ASN1F_NULL("noSuchObject", None, implicit_tag=0x80)),

ASN1F_optional(ASN1F_NULL("noSuchInstance", None, implicit_tag=0x81)),

ASN1F_optional(ASN1F_NULL("endOfMibView", None, implicit_tag=0x82)),

)

class SNMPget(ASN1_Packet):

ASN1_codec = ASN1_Codecs.BER

ASN1_root = ASN1F_SNMP_PDU_GET(ASN1F_INTEGER("id", 0),

ASN1F_enum_INTEGER("error", 0, SNMP_error),

ASN1F_INTEGER("error_index", 0),

ASN1F_SEQUENCE_OF("varbindlist", [], SNMPvarbind) # noqa: E501

)

class SNMPnext(ASN1_Packet):

ASN1_codec = ASN1_Codecs.BER

ASN1_root = ASN1F_SNMP_PDU_NEXT(ASN1F_INTEGER("id", 0),

ASN1F_enum_INTEGER("error", 0, SNMP_error),

ASN1F_INTEGER("error_index", 0),

ASN1F_SEQUENCE_OF("varbindlist", [], SNMPvarbind) # noqa: E501

)

class SNMPresponse(ASN1_Packet):

ASN1_codec = ASN1_Codecs.BER

ASN1_root = ASN1F_SNMP_PDU_RESPONSE(ASN1F_INTEGER("id", 0),

ASN1F_enum_INTEGER("error", 0, SNMP_error), # noqa: E501

ASN1F_INTEGER("error_index", 0),

ASN1F_SEQUENCE_OF("varbindlist", [], SNMPvarbind) # noqa: E501

)

class SNMPset(ASN1_Packet):

ASN1_codec = ASN1_Codecs.BER

ASN1_root = ASN1F_SNMP_PDU_SET(ASN1F_INTEGER("id", 0),

ASN1F_enum_INTEGER("error", 0, SNMP_error),

ASN1F_INTEGER("error_index", 0),

ASN1F_SEQUENCE_OF("varbindlist", [], SNMPvarbind) # noqa: E501

)

class SNMPtrapv1(ASN1_Packet):

ASN1_codec = ASN1_Codecs.BER

ASN1_root = ASN1F_SNMP_PDU_TRAPv1(ASN1F_OID("enterprise", "1.3"),

ASN1F_IPADDRESS("agent_addr", "0.0.0.0"),

ASN1F_enum_INTEGER("generic_trap", 0, SNMP_trap_types), # noqa: E501

ASN1F_INTEGER("specific_trap", 0),

ASN1F_TIME_TICKS("time_stamp", IntAutoTime()), # noqa: E501

ASN1F_SEQUENCE_OF("varbindlist", [], SNMPvarbind) # noqa: E501

)

class SNMPbulk(ASN1_Packet):

ASN1_codec = ASN1_Codecs.BER

ASN1_root = ASN1F_SNMP_PDU_BULK(ASN1F_INTEGER("id", 0),

ASN1F_INTEGER("non_repeaters", 0),

ASN1F_INTEGER("max_repetitions", 0),

ASN1F_SEQUENCE_OF("varbindlist", [], SNMPvarbind) # noqa: E501

)

class SNMPinform(ASN1_Packet):

ASN1_codec = ASN1_Codecs.BER

ASN1_root = ASN1F_SNMP_PDU_INFORM(ASN1F_INTEGER("id", 0),

ASN1F_enum_INTEGER("error", 0, SNMP_error), # noqa: E501

ASN1F_INTEGER("error_index", 0),

ASN1F_SEQUENCE_OF("varbindlist", [], SNMPvarbind) # noqa: E501

)

class SNMPtrapv2(ASN1_Packet):

ASN1_codec = ASN1_Codecs.BER

ASN1_root = ASN1F_SNMP_PDU_TRAPv2(ASN1F_INTEGER("id", 0),

ASN1F_enum_INTEGER("error", 0, SNMP_error), # noqa: E501

ASN1F_INTEGER("error_index", 0),

ASN1F_SEQUENCE_OF("varbindlist", [], SNMPvarbind) # noqa: E501

)

class SNMP(ASN1_Packet):

ASN1_codec = ASN1_Codecs.BER

ASN1_root = ASN1F_SEQUENCE(

ASN1F_enum_INTEGER("version", 1, {0: "v1", 1: "v2c", 2: "v2", 3: "v3"}), # noqa: E501

ASN1F_STRING("community", "public"),

ASN1F_CHOICE("PDU", SNMPget(),

SNMPget, SNMPnext, SNMPresponse, SNMPset,

SNMPtrapv1, SNMPbulk, SNMPinform, SNMPtrapv2)

)

def answers(self, other):

return (isinstance(self.PDU, SNMPresponse) and

isinstance(other.PDU, (SNMPget, SNMPnext, SNMPset)) and

self.PDU.id == other.PDU.id)

bind_bottom_up(UDP, SNMP, sport=161)

bind_bottom_up(UDP, SNMP, dport=161)

bind_bottom_up(UDP, SNMP, sport=162)

bind_bottom_up(UDP, SNMP, dport=162)

bind_layers(UDP, SNMP, sport=161, dport=161)

def snmpget(dst, oid="1.0.8802.1.1.1.1.1.2.1.2.29", community="public"):

"""

SNMP get.

This can be used to perform a SNMP scan::

>>> snmpget("192.168.0.0/16", community="public")

"""

ans, _ = sr(

IP(dst=dst) / UDP(sport=RandShort()) / SNMP(

community=community,

PDU=SNMPnext(varbindlist=[SNMPvarbind(oid=oid)]),

),

timeout=2,

chainCC=1,

verbose=0,

retry=2,

)

for r in ans:

if ICMP in r.answer:

print(repr(r.answer))

return

print("[%-10s] %-40s: %r" % (

r.query.dst,

r.answer[SNMPvarbind].oid.val,

r.answer[SNMPvarbind].value,

))

def snmpwalk(dst, oid="1", community="public"):

"""

SNMP walk

"""

try:

while True:

r = sr1(

IP(dst=dst) / UDP(sport=RandShort()) / SNMP(

community=community,

PDU=SNMPnext(varbindlist=[SNMPvarbind(oid=oid)]),

),

timeout=2,

chainCC=1,

verbose=0,

retry=2,

)

if r is None:

print("No answers")

break

if ICMP in r:

print(repr(r))

break

print("%-40s: %r" % (r[SNMPvarbind].oid.val, r[SNMPvarbind].value))

oid = r[SNMPvarbind].oid

except KeyboardInterrupt:

pass