在不使用原始套接字发送UDP请求后,从端口读取ICMP响应会遇到一些挑战,因为ICMP协议通常用于网络诊断和错误报告,而不是用于常规的数据传输。ICMP响应通常是由网络设备(如路由器)在检测到某些网络问题时生成的,例如目标不可达、超时等。
ICMP(Internet Control Message Protocol):是一种网络层协议,用于发送错误消息和操作信息。它通常用于诊断网络连接问题。
原始套接字:允许应用程序发送和接收原始IP数据包,包括ICMP包。使用原始套接字可以直接处理ICMP协议。
虽然题目要求不使用原始套接字,但为了完整性,这里简要说明如何使用原始套接字来处理ICMP响应:
import socket
import struct
import time
ICMP_ECHO_REQUEST = 8
def checksum(source_string):
sum = 0
count_to = (len(source_string) // 2) * 2
count = 0
while count < count_to:
this_val = source_string[count + 1] * 256 + source_string[count]
sum = sum + this_val
sum = sum & 0xffffffff
count = count + 2
if count_to < len(source_string):
sum = sum + source_string[len(source_string) - 1]
sum = sum & 0xffffffff
sum = (sum >> 16) + (sum & 0xffff)
sum = sum + (sum >> 16)
answer = ~sum
answer = answer & 0xffff
answer = answer >> 8 | (answer << 8 & 0xff00)
return answer
def send_one_ping(sock, dest_addr, id):
packet = struct.pack('!BBHHH', ICMP_ECHO_REQUEST, 0, 0, id, 1)
checksum_value = checksum(packet)
packet = struct.pack('!BBHHH', ICMP_ECHO_REQUEST, 0, checksum_value, id, 1)
sock.sendto(packet, (dest_addr, 1))
def receive_one_ping(sock, id, timeout):
time_left = timeout
while True:
start_time = time.time()
readable = select.select([sock], [], [], time_left)
time_spent = time.time() - start_time
if not readable[0]: # Timeout
return None
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:
return time_received - start_time
time_left -= time_spent
if time_left <= 0:
return None
def do_one(dest_addr, timeout):
icmp = socket.getprotobyname("icmp")
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, icmp)
except PermissionError:
print("Permission denied. You need to run this script as root.")
return
my_id = os.getpid() & 0xFFFF
send_one_ping(sock, dest_addr, my_id)
delay = receive_one_ping(sock, my_id, timeout)
sock.close()
return delay
# Example usage
import os
import select
dest_addr = "8.8.8.8" # Google's public DNS server
timeout = 1
delay = do_one(dest_addr, timeout)
if delay is not None:
print(f"Reply from {dest_addr}: time={delay:.3f}ms")
else:
print("Request timed out.")
如果不允许使用原始套接字,可以考虑使用现有的库,如ping3
,它封装了ICMP请求的处理:
from ping3 import ping, verbose_ping
# Simple ping
response_time = ping("8.8.8.8")
print(f"Response time: {response_time} ms")
# Verbose ping with more details
verbose_ping("8.8.8.8", count=4)
通过上述方法,可以在不使用原始套接字的情况下,有效地处理ICMP响应。
领取专属 10元无门槛券
手把手带您无忧上云