HCNP学习笔记之ICMP协议与ping原理以及用Python实现ping

一、ICMP协议分析

ICMP：Internet控制报文协议。由于IP协议并不是一个可靠的协议，它不保证数据被成功送达，那么，如何才能保证数据的可靠送达呢？ 这里就需要使用到一个重要的协议模块ICMP(网络控制报文)协议。它传递差错报文以及其他需要注意的信息，经常供IP层或更高层协议（TCP或UDP）使用。所以它经常被认为是IP层的一个组成部分。它在IP数据报文中的封装如下：

ICMP的数据报文格式如下所示。所有报文的前4个字节都是一样的，其他的因报文类型不同而不一样。类型字段可以有15个不同的值，用以描述不同的ICMP报文。校验和字段覆盖整个ICMP报文，使用了和IP首部检验和一样的算法，详细请搜索TCP/IP检验和算法。

 

 不同类型的报文是由类型字段和代码字段来共同决定。下表是各种类型的ICMP报文。

根据上表可知，ICMP协议大致分为两类，一种是查询报文，一种是差错报文。查询报文是用一对请求和应答定义的，它通常有以下几种用途:

ping查询

子网掩码查询（用于无盘工作站在初始化自身的时候初始化子网掩码）

时间戳查询（可以用来同步时间）

而差错报文通常包含了引起错误的IP数据报的第一个分片的IP首部（和选项），加上该分片数据部分的前8个字节。RFC 792规范中定义的这8个字节中包含了该分组运输层首部的所有分用信息，这样运输层协议就可以向正确的进程提交ICMP差错报文。

当传送IP数据包发生错误时，比如主机不可达，端口不可达等，ICMP协议就会把错误信息封包，然后传送回给主机。给主机一个处理错误的机会，这也就是为什么说建立在IP层以上的协议是可能做到安全的原因。由上面可知，ICMP数据包由8bit的错误类型和8bit的代码和16bit的校验和组成，而前 16bit就组成了ICMP所要传递的信息。由数据链路层所能发送的最大数据帧，即MTU（Maximum Transmission Unit）为1500，计算易知ICMP协议在实际传输中数据包为：20字节IP首部 + 8字节ICMP首部+ 1472字节（数据大小）。

尽管在大多数情况下，错误的包传送应该给出ICMP报文，但是在特殊情况下，是不产生ICMP错误报文的。如下

ICMP差错报文不会产生ICMP差错报文（出IMCP查询报文）（防止IMCP的无限产生和传送）

目的地址是广播地址或多播地址的IP数据报。

作为链路层广播的数据报。

不是IP分片的第一片。

源地址不是单个主机的数据报。这就是说，源地址不能为零地址、环回地址、广播地 址或多播地址。

二、ping程序原理分析

ping程序是由Mike Muuss编写，目的是为了测试另一 台主机是否可达，现在已经成为一个常用的网络状态检查工具。该程序发送一份 ICMP回显请求报文给远程主机，并等待返回 ICMP回显应答。利用ping这种原理，已经出现了许多基于ping的网络扫描器，比如nmap、arping、fping、hping3等。所以随着Internet安全意识的增强，现在有些提供访问控制策略的路由器和防火墙已经可以设置过滤特定ICMP报文请求。因此并不能通过简单的ping命令判断远程主机是否在线。

ping 使用的是ICMP协议，它发送icmp回送请求消息给目的主机。ICMP协议规定：目的主机必须返回ICMP回送应答消息给源主机。如果源主机在一定时间内收到应答，则认为主机可达。大多数的 TCP/IP 实现都在内核中直接支持Ping服务器，ICMP回显请求和回显应答报文如下图所示。

ping的原理是用类型码为0的ICMP发请 求，受到请求的主机则用类型码为8的ICMP回应。通过计算ICMP应答报文数量和与接受与发送报文之间的时间差，判断当前的网络状态。这个往返时间的计算方法是：ping命令在发送ICMP报文时将当前的时间值存储在ICMP报文中发出，当应答报文返回时，使用当前时间值减去存放在ICMP报文数据中存放发送请求的时间值来计算往返时间。ping返回接受到的数据报文字节大小、TTL值以及往返时间。

Unix系统在实现ping程序时是把ICMP报文中的标识符字段置成发送进程的 ID号。这样 即使在同一台主机上同时运行了多个 ping程序实例，ping程序也可以识别出返回的信息。

三、ICMP 的应用--Traceroute

Traceroute 是用来侦测主机到目的主机之间所经路由情况的重要工具,也是最便利的工具。前面说到,尽管 ping 工具也可以进行 侦测,但是,因为 ip 头的限制,ping 不能完全的记录下所经过的路由器。所以 Traceroute 正好就填补了这个缺憾。

Traceroute 的原理是非常非常的有意思,它受到目的主机的 IP 后,首先给目的主机发送一个 TTL=1(还记得 TTL 是什么吗?)的 UDP(后面就 知道 UDP 是什么了)数据包,而经过的第一个路由器收到这个数据包以后,就自动把 TTL 减1,而 TTL 变为0以后,路由 器就把这个包给抛弃了,并同时产生 一个主机不可达的 ICMP 数据报给主机。主机收到这个数据报以后再发一个 TTL=2的 UDP 数据报给目的主机,然后刺激第二个路由器给主机发 ICMP 数据 报。如此往复直到到达目的主机。这样,traceroute 就拿到了所有的路由器 ip。从而避开了 ip 头只能记录有限路由 IP 的问题。

有人要问,我怎么知道 UDP 到没到达目的主机呢?这就涉及一个技巧的问题,TCP 和 UDP 协议有一个端口号定义,而普通的网 络程序只监控少数的几个号码较 小的端口,比如说80,比如说23,等等。而 traceroute 发送的是端口号>30000(真变态)的 UDP 报,所以到 达目的主机的时候,目的 主机只能发送一个端口不可达的 ICMP 数据报给主机。主机接到这个报告以后就知道,主机到了,所以,说 Traceroute 是一个骗子一点也不为过。

Traceroute 程序里面提供了一些很有用的选项,甚至包含了 IP 选路的选项,请察看 man 文档来了解这些,这里就不赘述了。

四、python实现ping程序

 方法一、使用python脚本调用系统中的ping命令简单实现

import subprocess
import shlex

cmd = "ping -c 1 www.baidu.com"
args = shlex.split(cmd)

try:
subprocess.check_call(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
print "baidu server is up!"
except subprocess.CalledProcessError:
print "Failed to get ping."

但是，很多情况下，系统中的ping可执行文件是不可用，或者无法访问。这时，就需要使用一个纯python的检查脚本了。下面是ICMP ping的python实现脚本。这个脚本中定义了一个Pinger类，使用的一个校验检验和的do_checksum()方法，一个发送ping数据报文的send_ping()方法，接受ping数据报文的receive_ping()方法和一个执行这个类的ping()主方法。下面是具体的代码

# 定义Pinger类，初始化创建实例

import os
import argparse
import socket
import struct
import select
import time


ICMP_ECHO_REQUEST = 8 # Platform specific
DEFAULT_TIMEOUT = 2
DEFAULT_COUNT = 4


class Pinger(object):
""" Pings to a host -- the Pythonic way"""

def __init__(self, target_host, count=DEFAULT_COUNT, timeout=DEFAULT_TIMEOUT):
self.target_host = target_host
self.count = count
self.timeout = timeout

下面定义了do_checksum()方法，进行检验和的校验，校验方法如下：

把校验和字段置为0

将icmp包（包括header和data）以16bit（2个字节）为一组，并将所有组相加（二进制求和）

若高16bit不为0，则将高16bit与低16bit反复相加，直到高16bit的值为0，从而获得一个只有16bit长度的值

将此16bit值进行按位求反操作，将所得值替换到校验和字段

具体算法可以搜索：【TCP/IP】检验和算法。

def do_checksum(self, source_string):
""" Verify the packet integritity """
sum = 0
max_count = (len(source_string)/2)*2
count = 0
while count < max_count: # 分割数据每两比特(16bit)为一组
val = ord(source_string[count + 1])*256 + ord(source_string[count])
sum = sum + val
sum = sum & 0xffffffff
count = count + 2

if max_count<len(source_string): <span class="hljs-comment"># 如果数据长度为基数,则将最后一位单独相加</span>
sum = sum + ord(source_string[len(source_string) - 1])
sum = sum & 0xffffffff

sum = (sum >> 16) + (sum & 0xffff) # 将高16位与低16位相加直到高16位为0
sum = sum + (sum >> 16)
answer = ~sum
answer = answer & 0xffff
answer = answer >> 8 | (answer << 8 & 0xff00)
return answer # 返回的是十进制整数

下面是接受ICMP类型码为8的ICMP回应报文的方法。在未到达超时时间之前socket处于阻塞状态一直等待响应，当有数据传回时就接受响应，然后提取包含标识符ID的ICMP报文首部和包含发送时间值的ICMP内容部分，计算请求-响应的延时间隔。

def receive_ping(self, sock, ID, timeout):
"""
Receive ping from the socket.
"""
time_remaining = timeout
while True:
start_time = time.time()
readable = select.select([sock], [], [], time_remaining)
time_spent = (time.time() - start_time)
if readable[0] == []: # Timeout
return

time_received = time.time()
recv_packet, addr = sock.recvfrom(1024)
icmp_header = recv_packet[20:28]
type, code, checksum, packet_ID, sequence = struct.unpack(
"bbHHh", icmp_header
)
if packet_ID == ID:
bytes_In_double = struct.calcsize("d")
time_sent = struct.unpack("d", recv_packet[28:28 + bytes_In_double])[0]
return time_received - time_sent

time_remaining = time_remaining - time_spent
if time_remaining <= 0:
return

下面定义的send_ping()方法，获取远程主机的DNS主机名，然后使用struct模块创建一个ICMP_ECHO_REQUEST数据包，将查验请求的数据发送到目标主机。在此发送前也需要进行do_checksum()方法的校验。

def send_ping(self, sock, ID):
"""
Send ping to the target host
"""
target_addr = socket.gethostbyname(self.target_host)

my_checksum = 0

# Create a dummy heder with a 0 checksum.
header = struct.pack("bbHHh", ICMP_ECHO_REQUEST, 0, my_checksum, ID, 1)
bytes_In_double = struct.calcsize("d")
data = (192 - bytes_In_double) * "Q"
data = struct.pack("d", time.time()) + data

# Get the checksum on the data and the dummy header.
my_checksum = self.do_checksum(header + data)
header = struct.pack(
"bbHHh", ICMP_ECHO_REQUEST, 0, socket.htons(my_checksum), ID, 1
)
packet = header + data
sock.sendto(packet, (target_addr, 1))

下面定义了一个ping_once()方法，向远程主机发送一次查验：将ICMP协议传给socket()方法，创建一个原始的ICMP套接字。由于ping程序需要使用SOCK_RAW来构建数据包，所以需要root权限才能运行这个程序。因此，本程序需要使用root权限运行，下面的异常处理部分就是来负责未使用root运行时抛出的异常。

def ping_once(self):
"""
Returns the delay (in seconds) or none on timeout.
"""
icmp = socket.getprotobyname("icmp")
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, icmp)
except socket.error, (errno, msg):
if errno == 1:
# Not superuser, so operation not permitted
msg += "ICMP messages can only be sent from root user processes"
raise socket.error(msg)
except Exception, e:
print "Exception: %s" %(e)

my_ID = os.getpid() & 0xFFFF

self.send_ping(sock, my_ID)
delay = self.receive_ping(sock, my_ID, self.timeout)
sock.close()
return delay

下面这个ping()是执行这个类的主要方法。在for循环中调用ping_once()方法，发送ping数据报文，并返回结果。

def ping(self):
"""
Run the ping process
"""
for i in xrange(self.count):
print "Ping to %s..." % self.target_host,
try:
delay = self.ping_once()
except socket.gaierror, e:
print "Ping failed. (socket error: '%s')" % e[1]
break

if delay == None:
print "Ping failed. (timeout within %ssec.)" % self.timeout
else:
delay = delay * 1000
print "Get ping in %0.4fms" % delay

if __name__ == '__main__':
parser = argparse.ArgumentParser(description='Python ping')
parser.add_argument('--target-host', action="store", dest="target_host", required=True)
given_args = parser.parse_args()
target_host = given_args.target_host
pinger = Pinger(target_host=target_host)
pinger.ping()

​​完整代码​​

参考

​​参考1​​

​​参考2​​