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

[main author Brett Cannon] [co-authored with Philip E. Nu�ez] [ported to python3k by cerebus_mindless]

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, 216 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
215
216
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 string
        or None.

    packet
        binary representation of packet.
    
    EXAMPLE: (using Python as root)
		>>> import icmpLib
		>>> icmplib.ping('127.0.0.1')
    """

    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 string or None type."""
        if not isinstance(data, (str, bytes)) and not isinstance(data, type(None)):
            raise TypeError("value must be a subclass of string 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 != None:
            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("!%dH" % (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 3.0, only ping() has been tested