📅  最后修改于: 2023-12-03 15:36:04.673000             🧑  作者: Mango
互联网控制消息协议 (ICMP) 是互联网协议套件中的一个子协议,它在网络中传递错误信息和操作控制信息。ICMP 消息通常由网络设备和主机生成和发送。
ICMP 消息具有以下特点:
ICMP 消息是在 IP 数据包中封装的。ICMP 消息的格式如下:
0 1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Type | Code | Checksum |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Data |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
ICMP 消息格式字段的含义:
ICMP 消息有多种类型,常用的类型如下:
在各种编程语言中,可以使用套接字库来发送 ICMP 消息。下面是在 Python 中通过套接字发送 ICMP Echo Request 消息的示例代码:
import socket
import struct
import select
import time
def checksum(string):
csum = 0
count_to = (len(string) / 2) * 2
count = 0
while count < count_to:
this_val = ord(string[count + 1]) * 256 + ord(string[count])
csum = csum + this_val
csum = csum & 0xffffffff
count = count + 2
if count_to < len(string):
csum = csum + ord(string[len(string) - 1])
csum = csum & 0xffffffff
csum = (csum >> 16) + (csum & 0xffff)
csum = csum + (csum >> 16)
answer = ~csum
answer = answer & 0xffff
answer = answer >> 8 | (answer << 8 & 0xff00)
return answer
def receive_ping(my_socket, ID, timeout):
time_left = timeout
while True:
started_select = time.time()
what_ready = select.select([my_socket], [], [], time_left)
how_long_in_select = (time.time() - started_select)
if what_ready[0] == []:
return "Request timed out."
time_received = time.time()
rec_packet, addr = my_socket.recvfrom(1024)
icmp_header = rec_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", rec_packet[28 : 28 + bytes_in_double])[0]
return time_received - time_sent
time_left = time_left - how_long_in_select
if time_left <= 0:
return "Request timed out."
def send_ping(my_socket, dest_addr, ID):
dest_addr = socket.gethostbyname(dest_addr)
my_checksum = 0
header = struct.pack("bbHHh", 8, 0, my_checksum, ID, 1)
padding = struct.pack("d", time.time())
my_checksum = checksum(header + padding)
header = struct.pack(
"bbHHh", 8, 0, socket.htons(my_checksum), ID, 1
)
packet = header + padding
my_socket.sendto(packet, (dest_addr, 1))
def ping(dest_addr, timeout=2, count=4):
my_socket = socket.socket(
socket.AF_INET, socket.SOCK_RAW, socket.getprotobyname("icmp")
)
for i in range(count):
send_ping(my_socket, dest_addr, i)
delay = receive_ping(my_socket, i, timeout)
print("ping %s..." % dest_addr, end="")
if isinstance(delay, str):
print(delay)
else:
delay = delay * 1000
print("获取响应时间 %.4f 毫秒" % delay)
my_socket.close()
这段代码使用套接字发送 ICMP Echo Request 消息,并等待接收 ICMP Echo Reply 消息。如果超时,则返回“Request timed out.”。如果收到 ICMP Echo Reply 消息,则返回响应时间。