使用socket和dpkt模块向本地主机收发ICMP数据包遇阻塞问题求助

我尝试向本地主机发送ICMP回显数据包，通过Wireshark确认发送成功，但接收函数一直卡在循环中无法获取响应。现有代码如下：

发送函数 #

def send_echo(socket, data, id, seq, dest):
    echo = dpkt.icmp.ICMP.Echo()
    echo.id = id
    echo.seq = seq
    echo.payload = bytes(data, 'utf-8')

    icmp = dpkt.icmp.ICMP()
    icmp.type = dpkt.icmp.ICMP_ECHO
    icmp.data = echo

    icmp_packed = icmp.pack()
    socket.sendto(icmp_packed, destination) #sending packet here 

接收函数 #

def recv_response():
    mySocket = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP) 
    while True:
    try:
        received, addr = mySocket.recvfrom(1024)
        break
    except socket.error:
        test = 0
    return received

主函数 #

def main():
    host_port = tuple(["127.0.0.1", 65432])
    send_echo(make_socket(50,50), "data", 0, 0, host_port)    
    recv_response()

问题分析与修复方案 #

接收函数的核心问题 #

1. 无超时设置：recvfrom会一直阻塞直到收到数据包，若响应丢失或未匹配，程序会无限挂起
2. 未过滤响应类型：收到的ICMP包可能不是回显响应（ICMP_ECHO_REPLY），需要筛选目标包
3. 错误处理无效：捕获socket.error后仅赋值变量，无任何日志或退出逻辑，无法排查问题
4. 发送端的隐性问题：
   • 变量名错误：send_echo参数是dest，但代码里用了未定义的destination，会直接抛出NameError
   • ICMP是无端口协议，host_port里的端口参数无效，只需传递("127.0.0.1", 0)即可

修复后的完整代码 #

import socket
import dpkt

def send_echo(sock, data, id, seq, dest):
    echo = dpkt.icmp.ICMP.Echo()
    echo.id = id
    echo.seq = seq
    echo.payload = bytes(data, 'utf-8')

    icmp = dpkt.icmp.ICMP()
    icmp.type = dpkt.icmp.ICMP_ECHO
    icmp.data = echo

    icmp_packed = icmp.pack()
    # 修复变量名错误，使用参数dest而非destination
    sock.sendto(icmp_packed, dest)

def recv_response(expected_id, expected_seq, timeout=5):
    mySocket = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)
    # 设置超时，避免无限阻塞
    mySocket.settimeout(timeout)
    try:
        while True:
            received, addr = mySocket.recvfrom(1024)
            # 解析IP头部（跳过前20字节是IP头）
            ip = dpkt.ip.IP(received)
            # 确认是ICMP包且为回显响应
            if isinstance(ip.data, dpkt.icmp.ICMP) and ip.data.type == dpkt.icmp.ICMP_ECHO_REPLY:
                echo_reply = ip.data.data
                # 匹配发送时的ID和序列号，确保是对应响应
                if echo_reply.id == expected_id and echo_reply.seq == expected_seq:
                    return received, addr
    except socket.timeout:
        print("接收超时，未收到响应")
        return None, None
    except socket.error as e:
        print(f"接收出错: {e}")
        return None, None
    finally:
        mySocket.close()

def make_socket():
    # 创建RAW ICMP套接字（注意：需要管理员权限）
    return socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)

def main():
    dest = ("127.0.0.1", 0)  # ICMP不需要端口，设为0即可
    echo_id = 1234  # 自定义ID，用于匹配响应
    echo_seq = 0
    sock = make_socket()
    try:
        send_echo(sock, "test_data", echo_id, echo_seq, dest)
        print("已发送ICMP回显请求")
        response, addr = recv_response(echo_id, echo_seq)
        if response:
            print(f"收到来自{addr}的响应")
    finally:
        sock.close()

if __name__ == "__main__":
    main()

关键修改说明 #

• 给接收套接字添加settimeout，避免无限阻塞
• 解析IP和ICMP头部，只处理ICMP_ECHO_REPLY类型的数据包
• 匹配发送时的ID和序列号，确保收到的是对应请求的响应
• 修复发送函数的变量名错误，移除无效的端口参数
• 添加finally块确保套接字正常关闭，避免资源泄漏
• 完善错误处理，添加超时和错误日志

内容的提问来源于stack exchange，提问作者user19945252