Welcome, guest | Sign In | My Account | Store | Cart

[co-authored with Philip E. Nu�ez]

Python's stdlib does not have any included library for supporting ICMP packets; both reading them or creating them. But ICMP packets are common and useful; they are used for both the traceroute and ping utilities. And thus they can be useful to control to do network diagnostics.

The Packet class is what is used to create and read ICMP packets. To create a packet you instantiate the class, set the header and data fields, and then call the create() method which will the string representation that can be passed to a socket.

To read a packet, use the Packet.parse() classmethod, which will return an instance of Packet with the fields filled out.

To show its use you can also see the ping() method that is included. Just use the code as a script and pass in an address to ping. Response time is printed to stdout.

One word of warning, though, when using this module. Raw sockets tend to require root permissions on the process. Thus you might need to use sudo to execute the Python interpreter to make this all work. ping() does drop sudo permissions as soon as it can, though, for security reasons.

Python, 214 lines
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
import struct

class Packet(object):

    """Creates ICMPv4 and v6 packets.
    
    header
        two-item sequence containing the type and code of the packet,
        respectively.
    version
        Automatically set to version of protocol being used or None if ambiguous.
    data
        Contains data of the packet.  Can only assign a subclass of basestring
        or None.

    packet
        binary representation of packet.
    
    """

    header_table = {
                0 : (0, 4),
                #3 : (15, 4),  Overlap with ICMPv6
                3 : (15, None),
                #4 : (0, 4),  Deprecated by RFC 1812
                5 : (3, 4),
                8 : (0, 4),
                9 : (0, 4),
                10: (0, 4),
                11: (1, 4),
                12: (1, 4),
                13: (0, 4),
                14: (0, 4),
                15: (0, 4),
                16: (0, 4),
                17: (0, 4),
                18: (0, 4),

                1 : (4, 6),
                2 : (0, 6),
                #3 : (2, 6),  Overlap with ICMPv4
                #4 : (2, 6),  Type of 4 in ICMPv4 is deprecated
                4 : (2, None),
                128: (0, 6),
                129: (0, 6),
                130: (0, 6),
                131: (0, 6),
                132: (0, 6),
                133: (0, 6),
                134: (0, 6),
                135: (0, 6),
                136: (0, 6),
                137: (0, 6),
             }

    def _setheader(self, header):
        """Set type, code, and version for the packet."""
        if len(header) != 2:
            raise ValueError("header data must be in a two-item sequence")
        type_, code = header
        try:
            max_range, version = self.header_table[type_]
        except KeyError:
            raise ValueError("%s is not a valid type argument" % type_)
        else:
            if code > max_range:
                raise ValueError("%s is not a valid code value for type %s" %\
                                     (type_, code))
            self._type, self._code, self._version = type_, code, version

    header = property(lambda self: (self._type, self._code), _setheader,
                       doc="type and code of packet")

    version = property(lambda self: self._version,
                        doc="Protocol version packet is using or None if "
                            "ambiguous")

    def _setdata(self, data):
        """Setter for self.data; will only accept a basestring or None type."""
        if not isinstance(data, basestring) and not isinstance(data, type(None)):
            raise TypeError("value must be a subclass of basestring or None, "
                            "not %s" % type(data))
        self._data = data

    data = property(lambda self: self._data, _setdata,
                    doc="data contained within the packet")

    def __init__(self, header=(None, None), data=None):
        """Set instance attributes if given."""
        #XXX: Consider using __slots__
        # self._version initialized by setting self.header
        self.header = header
        self.data = data

    def __repr__(self):
        return "<ICMPv%s packet: type = %s, code = %s, data length = %s>" % \
                (self.version, self.type, self.code, len(self.data))

    def create(self):
        """Return a packet."""
        # Kept as a separate method instead of rolling into 'packet' property so
        # as to allow passing method around without having to define a lambda
        # method.
        args = [self.header[0], self.header[1], 0]
        pack_format = "!BBH"
        if self.data:
            pack_format += "%ss" % len(self.data)
            args.append(self.data)
        # ICMPv6 has the IP stack calculate the checksum
        # For ambiguous cases, just go ahead and calculate it just in case
        if self.version == 4 or not self.version:
            args[2] = self._checksum(struct.pack(pack_format, *args))
        return struct.pack(pack_format, *args)

    packet = property(create,
                       doc="Complete ICMP packet")

    def _checksum(self, checksum_packet):
        """Calculate checksum"""
        byte_count = len(checksum_packet)
        #XXX: Think there is an error here about odd number of bytes
        if byte_count % 2:
            odd_byte = ord(checksum_packet[-1])
            checksum_packet = checksum_packet[:-1]
        else:
            odd_byte = 0
        two_byte_chunks = struct.unpack("!%sH" % (len(checksum_packet)/2),
                                        checksum_packet)
        total = 0
        for two_bytes in two_byte_chunks:
            total += two_bytes
        else:
            total += odd_byte
        total = (total >> 16) + (total & 0xFFFF)
        total += total >> 16
        return ~total
        
    def parse(cls, packet):
        """Parse ICMP packet and return an instance of Packet"""
        string_len = len(packet) - 4 # Ignore IP header
        pack_format = "!BBH"
        if string_len:
            pack_format += "%ss" % string_len
        unpacked_packet = struct.unpack(pack_format, packet)
        type, code, checksum = unpacked_packet[:3]
        try:
            data = unpacked_packet[3]
        except IndexError:
            data = None
        return cls((type, code), data)

    parse = classmethod(parse)


#------------
# ping.py
#------------

import struct,socket,sys,time, os
#from icmplib import Packet

def main(addr):
    s = socket.socket(socket.AF_INET,socket.SOCK_RAW)
    s.connect((addr,))

datalen = 56
BUFSIZE = 1500


def ping(addr):
    print "PING (%s): %d data bytes" % (addr,datalen)

    ## create socket
    s = socket.socket(socket.AF_INET,socket.SOCK_RAW,
                        socket.getprotobyname('icmp'))
    s.connect((addr,22))

    ## setuid back to normal user
    os.setuid(os.getuid())

    seq_num = 0
    packet_count = 0
    process_id = os.getpid()
    base_packet = Packet((8,0))

    while 1:
    ## create ping packet 
        seq_num += 1
        pdata = struct.pack("!HHd",process_id,seq_num,time.time())
    
    ## send initial packet 
        base_packet.data = pdata
        s.send(base_packet.packet)
    
        ## recv packet
        buf = s.recv(BUFSIZE)
        current_time = time.time()

        ## parse packet; remove IP header first
        r = Packet.parse(buf[20:])

        ## parse ping data
        (ident,seq,timestamp) = struct.unpack("!HHd",r.data)

        ## calculate rounttrip time
        rtt =  current_time - timestamp
        rtt *= 1000
        print "%d bytes from %s: id=%s, seq=%u, rtt=%.3f ms" % (len(buf),
                                                        addr, ident, seq, rtt)
        time.sleep(1)
        
if __name__=='__main__':
    import sys
    ping(sys.argv[1])

ICMP packets are used for network debugging (traceroute and ping being the most common use). So this module could be used to easily detect if an address is up and running (assuming the address returns ping requests) or anything else ping and traceroute tend to be used for.

This module should work with Python 2.2, but has only be tested with Python 2.3 and higher. It was also written about three years ago and just recently rediscovered. Unit tests were not written for the module so we can't promise extensive testing. At least ping works. =)

7 comments

Mauro Baba 16 years, 1 month ago  # | flag

Problem, example and information. Hi, i'm using Python 2.5.

1: Problem) when i execute the code i got an error:

>

Traceback (most recent call last):

File "", line 1, in

IndexError: list index out of range

2: Exemple) can you do an example of how to use the code?

3: Info) there is a similar library to ARP packets?

My final intention is to ping an ip address (a range of ips in my lan) top see if its up, retrieve the host name and mac address.

For now i've use an escamotage: ive execute the ping with popen() and parsing the response (but there are various problems like different response for differents languages and o.s.).

Then, ive execute an "arp -a" command and does the same thing (and there are the same problems).

These problems force me to change the approach.

Any suggest?

Thanks

Kelvin Nicholson 15 years, 6 months ago  # | flag

To your #2:

(using Python as root)

>>> import icmplib
>>> icmplib.ping('127.0.0.1')
Jamie Andrews 14 years, 3 months ago  # | flag

At line 134, there is a conversion to twos complement. But the value is being fed back to an unsigned short So under some circumstances it works, but mostly it fails I commented out lines 134,135 and 136 and added

return total & 0xffff

This seems much more reliable

Gunter Ohrner 14 years, 2 months ago  # | flag

This code generates illegal ethernet frames with an insufficient length. The datalen-variable defined in the code is never used, the generated frames are not padded and only 50 bytes in size. This caused random paket delivery failures and even random misdeliveries with our network equipment.

The attached patch fixes this misbehaviour for us. It does not apply cleanly against the original code version however, as it's based on the module's variant currently used by us.

Gunter

Index: package/ud-wrtlib/site-python/ping.py
===================================================================
--- package/ud-wrtlib/site-python/ping.py       (Revision 1110)
+++ package/ud-wrtlib/site-python/ping.py       (Revision 1111)
@@ -165,25 +165,29 @@



-datalen = 56
+icmp_len = 0
BUFSIZE = 1500


def ping(addr):
        '''liefert RTT in ms oder 0 bei Fehler'''

+       global icmp_len
+
        ## create socket
        s = socket.socket(socket.AF_INET, socket.SOCK_RAW,
                                                                                socket.getprotobyname('icmp'))
        s.connect((addr,22))
        s.settimeout(2)

-       seq_num = 1
+       seq_num = 0
        process_id = os.getpid()
        base_packet = Packet((8,0))

-       ## create ping packet
-       pdata = struct.pack("!HHd",process_id,seq_num,time.time())
+       ## create ping packet
+       ## 16 = 4 bytes ICMP header, 2 bytes ID, 2 bytes seq no., 8 bytes timestamp
+       pad_len = max(14, icmp_len - 16)
+       pdata = struct.pack("!HHd%ds" % pad_len, process_id, seq_num, time.time(), pad_len * '*')

        try:
                ## send initial packet
@@ -203,7 +207,9 @@

        ## parse ping data
        try:
-               (ident,seq,timestamp) = struct.unpack("!HHd",r.data)
+               ## 12 = 2 bytes id, 2 bytes seq no., 8 bytes time
+               pad_len = max(0, len(r.data) - 12)
+               (ident, seq, timestamp, pad_bytes) = struct.unpack("!HHd%ds" % pad_len, r.data)
        except struct.error:
                ## Woran auch immer das liegen mag...
                return 0
Justok Jiang 12 years, 7 months ago  # | flag

I have a task that to create a ICMPv6 packet with specified souce ip and dst ip. But after reading your documentation, I tried but failed. Could you give a code example? my email is justok06@gmail.com. Thank you.

Steven Ayre 11 years, 6 months ago  # | flag

As Jamie Andrews points out, the checksum calculation fails to generate the correct checksums. In our tests this meant the code never received replies as the invalid ping request packets were dropped.

Unfortunately Jamie's suggestion of 'return total & 0xffff' also fails to return valid checksums.

This is the _checksum() replacement that we use, which does give correct checksums.

def _checksum(self, checksum_packet):
    """Calculate checksum"""
    total = 0

    # Add up 16-bit words
    num_words = len(checksum_packet) / 2
    for chunk in struct.unpack("!%sH" % num_words, checksum_packet[0:num_words*2]):
        total += chunk

    # Add any left over byte
    if len(checksum_packet) % 2:
        total += ord(checksum_packet[-1]) << 8

    # Fold 32-bits into 16-bits
    # Note the offset: in C this would return as a uint16_t type, but Python
    # returns it as signed which puts it in the wrong range for struct.pack's H cast
    # Adding the 0xffff offset moves it into the correct range, while the mask removes any overflow.
    total = (total >> 16) + (total & 0xffff)
    total += total >> 16
    return (~total + 0xffff & 0xffff)
Steven Ayre 11 years, 6 months ago  # | flag

Correction to previous code (caused by incorrectly rewriting decimal constant as hex):

def _checksum(self, checksum_packet): """Calculate checksum""" total = 0

# Add up 16-bit words
num_words = len(checksum_packet) / 2
for chunk in struct.unpack("!%sH" % num_words, checksum_packet[0:num_words*2]):
    total += chunk

# Add any left over byte
if len(checksum_packet) % 2:
    total += ord(checksum_packet[-1]) << 8

# Fold 32-bits into 16-bits
# Note the offset: in C this would return as a uint16_t type, but Python
# returns it as signed which puts it in the wrong range for struct.pack's H cast
# Adding the 0xffff offset moves it into the correct range, while the mask removes any overflow.
total = (total >> 16) + (total & 0xffff)
total += total >> 16
return (~total + 0x10000 & 0xffff)