ICMP协议编程实践：实现ping命令(Python语言)

注解

程序源码 可在本文末尾复制，或者在 Github 上下载：。

ICMP 报文承载在 IP 报文之上，头部结构非常简单：

注意到， ICMP 头部只有三个固定字段，其余部分因消息类型而异。固定字段如下：

type ， 消息类型 ；
code ，代码；
checksum ， 校验和 ；

ICMP 报文有很多不同的类型，由 type 和 code 字段区分。而命令使用其中两种：

ping命令原理

如上图，机器 A 通过 回显请求 ( Echo Request ) 询问机器 B ；机器 B 收到报文后通过 回显答复 ( Echo Reply ) 响应机器 A 。这两种报文的典型结构如下：

对应的 type 以及 code 字段值列举如下：

按照惯例，回显报文除了固定字段，其余部分组织成 3 个字段：

标识符 ( identifier )，一般填写进程 PID 以区分其他 ping 进程；
报文序号 ( sequence number )，用于编号报文序列；
数据 ( data )，可以是任意数据；

按 ICMP 规定， 回显答复 报文原封不动回传这些字段。因此，可以将 发送时间 封装在 数据负载 ( payload )中，收到答复后将其取出，用于计算 往返时间 ( round trip time )。

Python 标准库 struct 模块提供了用于 封装网络报文 的工具，可以这样封装数据负载：

封装报文头部也是类似的：

header = struct.pack('!BBHHH', _type, code, checksum, ident, seq)
icmp = header + payload

其中， B 表示长度为一个字节的无符号整数， H 表示长度为两个字节的无符号整数。

校验和

ICMP 报文校验和字段需要自行计算，计算步骤如下：

以 0 为校验和封装一个用于计算的 伪报文 ；
将报文分成两个字节一组，如果总字节数为奇数，则在末尾追加一个零字节；
对所有 双字节 进行按位求和；
将高于 16 位的进位取出相加，直到没有进位；
将校验和按位取反；

示例代码如下：

def calculate_checksum(icmp):
    if len(icmp) % 2:
        icmp += b'\00'
 
    checksum = 0
    for i in range(len(icmp)//2):
        word, = struct.unpack('!H', icmp[2i:2i+2])
        checksum += word
 
    while True:
        carry = checksum >> 16
        if carry:
            checksum = (checksum & 0xffff) + carry
        else:
            break
 
    checksum = ~checksum & 0xffff
 
    return struct.pack('!H', checksum)

编程实现网络通讯，离不开 套接字 ( socket )，收发 ICMP 报文当然也不例外：

from socket import socket, AF_INET, SOCK_RAW, IPPROTO_ICMP
 
s = socket(AF_INET, SOCK_RAW, IPPROTO_ICMP)

调用系统调用发送 ICMP 报文：

其中，第一个参数为封装好的 ICMP 报文；第二个参数为发送标志位，无特殊要求一般填 0 ；第三个参数为目的 IP 地址-端口对，端口这里填 0 。

调用 recvfrom 方法接收 ICMP 报文：

ip, (src_ip, _) = s.recvfrom(1500)

参数为接收缓冲区大小，这里用 1500 刚好是一个典型的 MTU 大小。注意到，系统调用返回 IP 报文，去掉前 20 字节的 IP 头部便得到 ICMP 报文。

注解

注意，创建 原始套接字 ( SOCK_RAW )需要超级用户权限。

程序实现

掌握基本原理后，便可着手编写代码了。

首先，实现 pack_icmp_echo_request 函数，用于封装 ICMP回显请求 报文：

def pack_icmp_echo_request(ident, seq, payload):
    pseudo = struct.pack(
        '!BBHHH',
        8,
        0,
        0,
        ident,
        seq,
    ) + payload
    checksum = calculate_checksum(pseudo)
    return pseudo[:2] + checksum + pseudo[4:]

对应地，实现 unpack_icmp_echo_reply 用于解析 ICMP回显答复 报文：

def unpackicmp_echo_reply(icmp):
    _type, code, , ident, seq, = struct.unpack(
        '!BBHHH',
        icmp[:8]
    )
    if _type != 0:
        return
    if code != 0:
        return
 
    payload = icmp[8:]
 
    return ident, seq, payload

接着，实现 send_routine 用于循环发送 ICMP回显请求 报文：

该函数需要 5 个参数，分别如下：

sock ，用于发送报文的 套接字 ；
addr ，目标 IP地址；
ident ， 标识符 ；
magic ，打包在数据负载中的魔性字符串；
stop ，停止发送标识；

第 3 行定义 报文序号 ，从 1 开始递增；接着是发送循环，不停发包，每次相隔一秒；第 7 行获取 发送时间戳 ；第 11 行将时间戳以及魔性字符串打包成 数据负载 ；第 14 行调用 pack_icmp_echo_request 封装 回显请求 报文；第 17 行调用系统调用 发送报文 ；第 19-20 行自增发送序号并等待一秒。

同样，实现 recv_routine 函数用于循环接收 ICMP回显答复 报文：

def recvroutine(sock, ident, magic):
    while True:
        # wait for another icmp packet
        ip, (src_addr, ) = sock.recvfrom(1500)
 
        # unpack it
        result = unpack_icmp_echo_reply(ip[20:])
        if not result:
            continue
 
        # print info
        _ident, seq, payload = result
        if _ident != ident:
            continue
        sending_ts, = struct.unpack('!d', payload[:8])
        print('%s seq=%d %5.2fms' % (
            src_addr,
            seq,
            (time.time()-sending_ts) * 1000,
        ))

第 4 行调用 recvfrom 系统调用接收 ICMP 报文；第 7 行调用 unpack_icmp_echo_reply解析报文 ；第 8-9 行忽略非回显答复报文；第 13-14 行检查标识符并忽略非法报文(可能是响应其他进程的)；第 16 行从 数据负载 中取出 发送时间戳 ；第 17-21 行，计算 往返时间 并输出提示。

报文发送和接收均实现完毕，如何让程序同时干两件事情呢？可以选用线程方案：

def ping(addr):
    # create socket for sending and receiving icmp packet
    sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)
 
    # id field
    ident = os.getpid()
    # magic string to pad
    magic = b'1234567890'
 
    # sender thread stop flag
    # append anything to stop
    sender_stop = []
 
    # start sender thread
    # call send_routine function to send icmp forever
    args = (sock, addr, ident, magic, sender_stop,)
    sender = threading.Thread(target=send_routine, args=args)
    sender.start()
 
    try:
        # receive icmp reply forever
        recv_routine(sock, ident, magic)
    except KeyboardInterrupt:
        pass
 
    # tell sender thread to stop
    sender_stop.append(True)
 
    # clean sender thread
    sender.join()
 
    print()

第 3 行创建用于发送、接收报文的 套接字 ；第 6 行获取进程 PID 作为 标识符 ；第 16-18 行启动一个 子线程 执行 报文发送 函数；第 20-24 行 主线程 执行 报文接收 函数直至用户按下 ctrl-C ；第 27 行程序退出前，通知发送线程退出并回收线程资源( join )。

将以上所有代码片段组装在一起，便得到 ping.py 命令。迫不及待想运行一下：

$ sudo python ping.py 8.8.8.8
8.8.8.8 seq=1 23.18ms
8.8.8.8 seq=2 22.25ms
8.8.8.8 seq=3 34.18ms

It works!

下一步

本节以语言为例，演示了 ICMP 编程方法。如果你对其他语言感兴趣，请按需取用：

ICMP协议编程实践：实现ping命令(C语言)

订阅更新，获取更多学习资料，请关注我们的：

小菜学编程

ICMP编程实践(Python语言)

ICMP协议编程实践：实现ping命令(Python语言)

校验和

程序实现

下一步