snmp.py

import binascii
from enum import IntEnum

# class UniversalClassTags(IntEnum):
#     sequence_of = 16
#     set_of = 17
#     numeric_string = 18
#     printable_string = 19
#     t61_string = 20
#     videotex_string = 21
#     ia5string = 22
#     utc_time = 23
#     generalized_time = 24
#     graphic_string = 25
#     visible_string = 26
#     general_string = 27
#     universal_string = 28
#     character_string = 29
#     bmp_string = 30

class BER:
    """Basic Encoding Rules"""

    class TagClass(IntEnum):
        universal = 0
        application = 1
        context_specific = 2
        private = 3

    universal_class_tags = ['eoc', 'boolean', 'integer', 'bit_string', 'octet_string', 'null', 'object_identifier',
                            'object_descriptor', 'external', 'real',
                            'enumerate', 'embedded_pdv', 'utf8_string', 'releative_oid', 'reserved', 'reserved', 'sequence',
                            'set', 'numeric_string', 'printable_string',
                            't61_string', 'videotex_string', 'ia5_string', 'utc_time', 'generalized_time', 'graphic_string',
                            'visible_string', 'general_string',
                            'universal_string', 'character_string', 'bmp_string']

    def __init__(self, tag_class, is_complex, tag_id, octet_length, value):
        self.tag_class = tag_class
        self.is_complex = is_complex
        self.tag_id = tag_id
        self.octet_length = octet_length
        self.value = value

    def __repr__(self):
        return "<class=%s complex=%s tag=%s length=%d value=%s>\n" % (
            self.tag_class, self.is_complex, self.tag_id if self.tag_class != self.TagClass.universal else self.universal_class_tags[self.tag_id], self.octet_length, self.value)

    @classmethod
    def parse_from(cls, buffer, offset=0):
        tag_class = cls.TagClass(buffer[offset] >> 6)
        is_complex = bool(buffer[offset] & 0b100000)
        tag_id = buffer[offset] & 0b11111
        if tag_id == 31:
            # long-form identifier
            tag_id = 0
            while buffer[offset] & 0b10000000:
                tag_id |= buffer[offset] & 0b01111111
                tag_id <<= 7
                offset += 1
            else:
                tag_id |= buffer[offset]  # & 0b01111111

        offset += 1
        # length
        # null is zero length?
        if buffer[offset] == 128:
            try:
                offset += 1
                octet_length = buffer.index(b'\0' * 2, offset) - offset
            except ValueError:
                raise Exception("Malformed packet: Indefinite form has no end-of-content terminator.")
        elif buffer[offset] < 128:
            octet_length = buffer[offset]
            offset += 1
        else:  # buffer[offset] > 128
            length_octets = buffer[offset] & 0b01111111
            offset += 1
            octet_length = sum([value << (octet * 8) for (octet, value) in
                                enumerate(reversed(buffer[offset:offset + length_octets]))])
            offset += length_octets
        stop_octet = offset + octet_length
        if is_complex:
            # value is wrapped in another tlv
            value = []
            while offset < stop_octet:
                (subobj, offset) = cls.parse_from(buffer, offset)
                value.append(subobj)
        else:
            value = buffer[offset:stop_octet]
        return cls(tag_class, is_complex, tag_id, octet_length, value), stop_octet


class SNMP:
    def __init__(self, version, community, pdu):
        self.version = version
        self.community = community
        self.pdu = pdu

    @classmethod
    def decode(cls, buffer, offset=0):
        """Decodes SNMP packet"""
        (object, offset) = BER.parse_from(buffer, offset)
        assert len(buffer) == offset
        (version, community, pdu) = object.value
        return cls(version, community, pdu)

    def __repr__(self):
        return "<SNMP version=%s, community=%s, pdu=%s>" % (self.version, self.community, self.pdu)

if __name__ == '__main__':
    snmp_datagrams = map(binascii.a2b_hex, ['302602010004067075626c6963a119020452447053020100020100300b300906052b060102010500',
                                            '307902010004067075626c6963a26c020452447053020100020100305e305c06082b0601020101010004504c696e7578206e696e6a616c6170707920342e322e352d312d4152434820233120534d5020505245454d505420547565204f63742032372030383a31333a3238204345542032303135207838365f3634',
                                            '302902010004067075626c6963a11c020452447054020100020100300e300c06082b060102010101000500',
                                            '303302010004067075626c6963a2260204524470540201000201003018301606082b06010201010200060a2b06010401bf0803020a'])
    for datagram in snmp_datagrams:
        snmp = SNMP.decode(datagram)
        print(snmp)