Changeset 5804

Timestamp:

08/15/10 16:06:47 (3 years ago)

Author:

kosma

Message:

ICMP: load_raw support; tests for get_raw and load_raw

Location:

umpa/branches/protocols

Files:

• tests/a_unit/test_protocols/test_icmp.py (modified) (2 diffs)
• umit/umpa/protocols/ICMP.py (modified) (2 diffs)

umpa/branches/protocols/tests/a_unit/test_protocols/test_icmp.py

@@ -21 +21 @@
 
 import py.test
-from umit.umpa.protocols import ICMP, Payload
+from umit.umpa.protocols import ICMP
+from umit.umpa.utils.exceptions import UMPAException, UMPAAttributeException
+
+def raw(obj):
+    return obj._raw(0, 0, [obj], 0)
 
 class TestICMP(object):
 
-    # test if the active fields are set properly for supported ICMP types
     def test_active_fields(self):
+        """
+        test if the active fields are set properly for supported ICMP types
+        """
+
         i = ICMP()
         # default - no type
@@ -70 +77 @@
                 assert i.get_field(name).active == True
 
-#    # this tests pre/post raw methods as well
-#    def test_get_raw(self):
-#        ip = IP(src='1.2.3.4', dst='5.6.7.8')
-#        udp = UDP(srcport=9, dstport=10)
-#
-#        assert udp._raw(0, 0, [ip, udp], 0) == (0x9000A0008EFB7, 64)
-#
-#    # test length and checksum calculation
-#    def test_get_raw_with_payload(self):
-#        ip = IP(src='1.2.3.4', dst='5.6.7.8')
-#        udp = UDP(srcport=9, dstport=10)
-#        payload = Payload('12345678')
-#
-#        # the next two lines are normally done by Packet.get_raw()
-#        payload.get_raw([ip, udp, payload], 8*len(payload.data))
-#        udp.__dict__['payload'] = payload
-#
-#        assert udp._raw(0, 0, [ip, udp, payload], 8*len(payload.data)) == \
-#                                                       (0x9000a00101ed3L, 64)
+    def test_get_raw(self):
+        # empty objects have no type set
+        py.test.raises(UMPAException, "raw(ICMP())")
+
+        # basic tests, no complicated headers
+        assert raw(ICMP(type=0))   == (0x0000ffff00000000, 64)
+        assert raw(ICMP(type=255)) == (0xff0000ff00000000, 64)
+        assert raw(ICMP(type=0, code=0x42)) == (0x0042bdff00000000, 64)
+
+        # Echo
+        ret = raw(ICMP(type='ECHO', ident=0x1234, seq=0x5678, data='ABCD'))
+        assert ret == (0x08000acd1234567841424344, 96)
+        ret = raw(ICMP(type='ECHO_REPLY', ident=1, seq=2, data='foo'))
+        assert ret == (0x00008d2a00010002666f6f, 88)
+        # Address Mask
+        ret = raw(ICMP(type='ADDRESS_MASK_REQUEST', seq=0xffff))
+        assert ret == (0x1100eeff0000ffff00000000, 96)
+        ret = raw(ICMP(type='ADDRESS_MASK_REPLY', ident=0xbeef, addressmask='255.255.255.0'))
+        assert ret == (0x1200300fbeef0000ffffff00, 96)
+        # Timestamp
+        ret = raw(ICMP(type='TIMESTAMP', originate=0x0666))
+        assert ret == (0x0d00ec9900000000000006660000000000000000, 160)
+        ret = raw(ICMP(type='TIMESTAMP_REPLY', receive=2, transmit=3))
+        assert ret == (0x0e00f1fa00000000000000000000000200000003, 160)
+        # Information
+        ret = raw(ICMP(type='INFORMATION_REQUEST', code=9, ident=0xdead, seq=0xbeef))
+        assert ret == (0x0f095359deadbeef, 64)
+        ret = raw(ICMP(type='INFORMATION_REPLY', code=1))
+        assert ret == (0x1001effe00000000, 64)
+        # Redirect
+        ret = raw(ICMP(type='REDIRECT', redir_gw='8.8.8.8', data='\0\0\0\x42'))
+        assert ret == (0x0500eaad0808080800000042, 96)
+        # Destination Unreachable
+        ret = raw(ICMP(type='DESTINATION_UNREACHABLE', code=10, data=' '))
+        assert ret == (0x030adcf50000000020, 72)
+        # Time Exceeded
+        ret = raw(ICMP(type='TIME_EXCEEDED'))
+        assert ret == (0x0b00f4ff00000000, 64)
+        # Parameter Problem
+        ret = raw(ICMP(type='PARAMETER_PROBLEM', pointer=0x42, pointer_unused=1))
+        assert ret == (0x0c00b1fe42000001, 64)
+        # Source Quench
+        ret = raw(ICMP(type='SOURCE_QUENCH', unused=0xcafebabe, data=''))
+        assert ret == (0x04007642cafebabe, 64)
+
+    def test_load_raw(self):
+        i = ICMP()
+
+        # code
+        i.load_raw("\x00\x42\xbd\xff\x00\x00\x00\x00")
+        assert i.type == 0
+        assert i.code == 0x42
+
+        # Echo
+        i.load_raw("\x08\x00\x0a\xcd\x12\x34\x56\x78\x41\x42\x43\x44")
+        assert i.type == 8
+        assert i.code == 0
+        assert i._checksum == 0x0acd
+        assert i.ident == 0x1234
+        assert i.seq == 0x5678
+        assert i.data == 'ABCD'
+
+        # Address Mask
+        i.load_raw("\x12\x00\x30\x0f\xbe\xef\x00\x00\xff\xff\xff\x00")
+        assert i.type == 18
+        assert i.code == 0
+        assert i._checksum == 0x300f
+        assert i.ident == 0xbeef
+        assert i.seq == 0x0000
+        assert i.addressmask == (255,255,255,0)
+
+        # Timestamp
+        i.load_raw("\x0e\x00\xf1\xfa\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00\x03")
+        assert i.type == 14
+        assert i.code == 0
+        assert i._checksum == 0xf1fa
+        assert i.ident == 0x0000
+        assert i.seq == 0x0000
+        assert i.originate == 0
+        assert i.receive == 2
+        assert i.transmit == 3
+
+        # Redirect
+        i.load_raw("\x05\x00\xea\xad\x08\x08\x08\x08\x00\x00\x00\x42")
+        assert i.type == 5
+        assert i.code == 0
+        assert i._checksum == 0xeaad
+        assert i.redir_gw == (8,8,8,8)
+        assert i.data == '\0\0\0\x42'

umpa/branches/protocols/umit/umpa/protocols/ICMP.py

@@ -552 +552 @@
         header_format = '!BBHI'
         fields = struct.unpack(header_format, buffer[:header_size])
+        buffer = buffer[header_size:]
 
         self.type = fields[0]
@@ -557 +558 @@
         self._checksum = fields[2]
 
-        # TODO: fields[3] contains type-specific information
-        # TODO: unpack rest of the packet
-
-        return buffer[header_size:]
+        # fill in dynamic header fields depending on the type
+        if self.type in (_consts.ICMP_TYPE_ADDRESS_MASK_REQUEST,
+                         _consts.ICMP_TYPE_ADDRESS_MASK_REPLY,
+                         _consts.ICMP_TYPE_ECHO,
+                         _consts.ICMP_TYPE_ECHO_REPLY,
+                         _consts.ICMP_TYPE_INFORMATION_REQUEST,
+                         _consts.ICMP_TYPE_INFORMATION_REPLY,
+                         _consts.ICMP_TYPE_TIMESTAMP,
+                         _consts.ICMP_TYPE_TIMESTAMP_REPLY,):
+            self.ident = fields[3] >> 16
+            self.seq   = fields[3] & 0x0000FFFF
+        elif self.type in (_consts.ICMP_TYPE_REDIRECT, ):
+            self.redir_gw = ( (fields[3]>>24) & 0xff,
+                              (fields[3]>>16) & 0xff,
+                              (fields[3]>> 8) & 0xff,
+                              (fields[3]    ) & 0xff, )
+        elif self.type in (_consts.ICMP_TYPE_PARAMETER_PROBLEM, ):
+            self.pointer = fields[3] >> 24
+            self.pointer_unused = fields[3] & 0x00ffffff
+        else:
+            self.unused = fields[3]
+
+        # fill in data fields depending on the type
+        if self.type in (_consts.ICMP_TYPE_ADDRESS_MASK_REQUEST,
+                         _consts.ICMP_TYPE_ADDRESS_MASK_REPLY,):
+            data_size = 4
+            data_format = "!4B"
+            fields = struct.unpack(data_format, buffer[:data_size])
+            buffer = buffer[data_size:]
+
+            self.addressmask = fields[0:4]
+        elif self.type in (_consts.ICMP_TYPE_TIMESTAMP,
+                           _consts.ICMP_TYPE_TIMESTAMP_REPLY,):
+            data_size = 12
+            data_format = "!3I"
+            fields = struct.unpack(data_format, buffer[:data_size])
+            buffer = buffer[data_size:]
+
+            self.originate = fields[0]
+            self.receive = fields[1]
+            self.transmit = fields[2]
+        else:
+            # unknown/generic data
+            self.data = buffer
+
+        return buffer
 
 protocols = [ ICMP, ]