如何通过Ping触发Python脚本及解决ICMP嗅探源IP显示异常问题
背景
我编写了一个Python脚本,可遍历网络中多台Cisco路由器并获取流量统计数据。已配置路由器在流量达到阈值后,向主机PC的指定地址发送Ping请求,需要触发该脚本运行。现有两个方案:
- 方案一:通过路由器发送的Ping触发脚本
- 方案二:通过FTP上传文件到PC文件夹触发,但后者需持续扫描文件夹,无法满足流量波动时的实时监控需求
我倾向采用方案一,即编写环回监听脚本,捕获特定源IP(路由器)的数据包后触发脚本。
尝试过程与遇到的问题
尝试过Socket监听,但路由器不支持原生发送UDP数据报,且Ping不涉及端口;尝试无端口IP监听,但此前仅做过Socket编程,缺乏ICMP Ping监听的实现经验(还需验证数据包源IP为路由器)。
之后参考《Black Hat Python》第3章代码实现了嗅探方案,功能基本正常,但存在核心问题:监听器输出的源IP和目的IP均解析为环回地址。我尝试从路由器及修改源地址的CMD发送Ping,源地址始终显示为环回地址,但Wireshark捕获到了正确的源IP。排查过是否捕获的是回显应答包,调整了socket.IPPROTO_ICMP参数及头部长度,问题仍未解决。
附上相关代码:
import socket import os import struct import ipaddress class IP: def __init__(self, buff=None): header = struct.unpack('<BBHHHBBH4s4s4s', buff) self.ver = header[0] >> 4 self.ihl = header[0] & 0xF self.tos = header[1] self.len = header[2] self.id = header[3] self.offset = header[4] self.ttl = header[5] self.protocol_num = header[6] self.sum = header[7] self.src = header[8] self.dst = header[9] self.trash = header[10] self.src_address = ipaddress.ip_address(self.src) self.dst_address = ipaddress.ip_address(self.dst) self.protocol_map = {1: "ICMP", 6: "TCP", 17: "UDP"} try: self.protocol = self.protocol_map[self.protocol_num] except Exception as e: print('%s No protocol for %s' % (e, self.protocol_num)) self.protocol = str(self.protocol_num) class ICMP: def __init__(self, buff): header = struct.unpack('<BBHHH', buff) self.type = header[0] self.code = header[1] self.sum = header[2] self.id = header[3] self.seq = header[4] def sniff(host): sniffer = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_IP) sniffer.bind((host, 0)) sniffer.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1) if os.name == 'nt': sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON) try: while True: raw_buffer = sniffer.recvfrom(65535)[0] ip_header = IP(raw_buffer[0:24]) print(f'Version: {ip_header.ver}') print(f'Header Length: {ip_header.ihl} TTL: {ip_header.ttl}') offset = ip_header.ihl * 4 buf = raw_buffer[offset:offset + 8] icmp_header = ICMP(buf) print('ICMP -> Type: %s Code: %s\n' % (icmp_header.type, icmp_header.code)) print('Protocol: %s %s -> %s' % (ip_header.protocol, ip_header.src_address, ip_header.dst_address)) except KeyboardInterrupt: if os.name == 'nt': sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF) exit() host = '192.168.56.1' sniff(host)
问题原因与解决方法
1. IP头部解析的字节序与格式错误
IP头部采用网络字节序(大端序),但代码中用了<(小端序)解析;同时格式字符串'<BBHHHBBH4s4s4s'会解析24字节,而标准IPv4头部最小是20字节,多余的4字节导致源IP、目的IP的位置偏移,解析出错误的地址。
修正后的IP类解析逻辑:
class IP: def __init__(self, buff=None): # 用网络字节序解析20字节标准IPv4头部 header = struct.unpack('!BBHHHBBH4s4s', buff[:20]) self.ver = header[0] >> 4 self.ihl = header[0] & 0xF self.tos = header[1] self.len = header[2] self.id = header[3] self.offset = header[4] self.ttl = header[5] self.protocol_num = header[6] self.sum = header[7] self.src = header[8] self.dst = header[9] self.src_address = ipaddress.ip_address(self.src) self.dst_address = ipaddress.ip_address(self.dst) self.protocol_map = {1: "ICMP", 6: "TCP", 17: "UDP"} try: self.protocol = self.protocol_map[self.protocol_num] except KeyError: self.protocol = str(self.protocol_num)
2. 过滤ICMP包类型
确保只处理路由器发送的ICMP请求包(Type=8),避免捕获本地生成的应答包(Type=0),可在嗅探逻辑中加入判断:
if icmp_header.type == 8: print(f"收到来自路由器 {ip_header.src_address} 的ICMP请求,触发流量统计脚本") # 此处调用你的流量统计脚本执行逻辑
3. 确认嗅探绑定地址
如果路由器是发送Ping到PC的物理网卡地址(如192.168.56.1),需确保嗅探器绑定的是该物理网卡地址,而非环回地址127.0.0.1;Windows下开启SIO_RCVALL可实现全网卡嗅探,Linux下无需额外设置即可捕获所有经过网卡的数据包。
内容的提问来源于stack exchange,提问作者hfakoor222

