CDP improvment (#352)

improve CDP: more fields handled, well formed packets, and tests added

Author: mori-b

dpkt/cdp.py

@@ -29,7 +29,7 @@
 class CDP(dpkt.Packet):
     """Cisco Discovery Protocol.
 
-    See more about the BGP on \
+    See more on
     https://en.wikipedia.org/wiki/Cisco_Discovery_Protocol
 
     Attributes:
@@ -43,69 +43,106 @@ class CDP(dpkt.Packet):
         ('sum', 'H', 0)
     )
 
-    class Address(dpkt.Packet):
-        # XXX - only handle NLPID/IP for now
-        __hdr__ = (
-            ('ptype', 'B', 1),  # protocol type (NLPID)
-            ('plen', 'B', 1),  # protocol length
-            ('p', 'B', 0xcc),  # IP
-            ('alen', 'H', 4)  # address length
-        )
-
-        def unpack(self, buf):
-            dpkt.Packet.unpack(self, buf)
-            self.data = self.data[:self.alen]
-
     class TLV(dpkt.Packet):
+        """When constructing the packet, len is not mandatory: 
+        if not provided, then self.data must be this exact TLV payload"""
+
         __hdr__ = (
             ('type', 'H', 0),
-            ('len', 'H', 4)
+            ('len', 'H', 0)
         )
 
+        def data_len(self):
+            if self.len:
+                return self.len - self.__hdr_len__
+            return len(self.data)
+
         def unpack(self, buf):
             dpkt.Packet.unpack(self, buf)
-            self.data = self.data[:self.len - 4]
-            if self.type == CDP_ADDRESS:
-                n = struct.unpack('>I', self.data[:4])[0]
-                buf = self.data[4:]
-                l = []
-                for i in range(n):
-                    a = CDP.Address(buf)
-                    l.append(a)
-                    buf = buf[len(a):]
-                self.data = l
+            self.data = self.data[:self.data_len()]
 
         def __len__(self):
-            if self.type == CDP_ADDRESS:
-                n = 4 + sum(map(len, self.data))
-            else:
-                n = len(self.data)
-            return self.__hdr_len__ + n
+            return self.__hdr_len__ + len(self.data)
 
         def __bytes__(self):
-            self.len = len(self)
-            if self.type == CDP_ADDRESS:
-                s = struct.pack('>I', len(self.data)) + \
-                    b''.join(map(bytes, self.data))
-            else:
-                s = self.data
-            return self.pack_hdr() + s
+            if hasattr(self, 'len') and not self.len:
+                self.len = len(self)
+            return self.pack_hdr() + bytes(self.data)
+
+    class Address(TLV):
+        # XXX - only handle NLPID/IP for now
+        __hdr__ = (
+            ('ptype', 'B', 1),  # protocol type (NLPID)
+            ('plen', 'B', 1),  # protocol length
+            ('p', 'B', 0xcc),  # IP
+            ('alen', 'H', 4)  # address length
+        )
+        def data_len(self):
+            return self.alen
+
+    class TLV_Addresses(TLV):
+        __hdr__ = (
+            ('type', 'H', CDP_ADDRESS),
+            ('len', 'H', 0),    #17),
+            ('Addresses', 'L', 1),
+        )
 
     def unpack(self, buf):
         dpkt.Packet.unpack(self, buf)
         buf = self.data
         l = []
         while buf:
-            tlv = self.TLV(buf)
-            l.append(tlv)
+            # find the right TLV according to Type value
+            tlv_find_type = self.TLV(buf).type
+            # if this TLV is not in tlv_types, use the default TLV class
+            tlv = self.tlv_types.get(tlv_find_type, self.TLV)(buf)
+            l.append(bytes(tlv))
             buf = buf[len(tlv):]
-        self.data = l
+        self.tlvs = l
+        self.data = b''.join(l)
 
     def __len__(self):
-        return self.__hdr_len__ + sum(map(len, self.data))
+        return self.__hdr_len__ + len(self.data)
 
     def __bytes__(self):
-        data = b''.join(map(bytes, self.data))
+        data = bytes(self.data)
         if not self.sum:
             self.sum = dpkt.in_cksum(self.pack_hdr() + data)
         return self.pack_hdr() + data
+
+    # keep here the TLV classes whose header is different from the generic TLV header (example : TLV_Addresses)
+    tlv_types = {CDP_ADDRESS: TLV_Addresses}
+
+
+def test_cdp():
+    import socket
+    from . import ethernet
+
+    ss = (b'\x02\xb4\xdf\x93\x00\x01\x00\x09\x63\x69\x73\x63\x6f\x00\x02\x00\x11\x00\x00\x00\x01\x01\x01\xcc\x00\x04\xc0\xa8\x01\x67')
+    rr1 = CDP(ss)
+    assert bytes(rr1) == ss    
+
+    # construction
+    ss = (b'\x02\xb4\xdf\x93\x00\x01\x00\x09\x63\x69\x73\x63\x6f\x00\x02\x00\x11\x00\x00\x00\x01\x01\x01\xcc\x00\x04\xc0\xa8\x01\x67')
+    p1 = CDP.TLV_Addresses(data=CDP.Address(data=socket.inet_aton('192.168.1.103')))
+    p2 = CDP.TLV(type=CDP_DEVID, data=b'cisco')
+    data = p2.pack() + p1.pack()
+    rr2 = CDP(data=data)
+    assert bytes(rr2) == ss    
+
+    s = (b'\x01\x00\x0c\xcc\xcc\xcc\xc4\x022k\x00\x00\x01T\xaa\xaa\x03\x00\x00\x0c \x00\x02\xb4,B'
+         b'\x00\x01\x00\x06R2\x00\x05\x00\xffCisco IOS Software, 3700 Software (C3745-ADVENTERPRI'
+         b'SEK9_SNA-M), Version 12.4(25d), RELEASE SOFTWARE (fc1)\nTechnical Support: http://www.'
+         b'cisco.com/techsupport\nCopyright (c) 1986-2010 by Cisco Systems, Inc.\nCompiled Wed 18'
+         b'-Aug-10 08:18 by prod_rel_team\x00\x06\x00\x0eCisco 3745\x00\x02\x00\x11\x00\x00\x00\x01'
+         b'\x01\x01\xcc\x00\x04\n\x00\x00\x02\x00\x03\x00\x13FastEthernet0/0\x00\x04\x00\x08\x00'
+         b'\x00\x00)\x00\t\x00\x04\x00\x0b\x00\x05\x00')
+    eth = ethernet.Ethernet(s)
+    assert isinstance(eth.data.data, CDP)
+    assert len(eth.data.data.tlvs) == 8  # number of CDP TLVs; ensures they are decoded
+    assert str(eth) == str(s)
+    assert len(eth) == len(s)
+
+if __name__ == '__main__':
+    test_cdp()
+    print('Tests Successful...')

dpkt/ethernet.py

@@ -619,7 +619,7 @@ def test_eth_llc_snap_cdp():  # 802.3 Ethernet - LLC/SNAP - CDP
     # stack
     assert isinstance(eth.data, llc.LLC)
     assert isinstance(eth.data.data, cdp.CDP)
-    assert len(eth.data.data.data) == 8  # number of CDP TLVs; ensures they are decoded
+    assert len(eth.data.data.tlvs) == 8  # number of CDP TLVs; ensures they are decoded
     assert str(eth) == str(s), 'pack 1'
     assert str(eth) == str(s), 'pack 2'
     assert len(eth) == len(s)