Python手写UDP主机发现工具:解码IP头部实现内网存活探测
1. 为什么我要自己写一个UDP主机发现工具
做网络资产梳理的时候,第一步永远是搞清楚"网段里到底有哪些主机活着"。市面上现成的扫描器不少,但真到内网环境里,很多工具要么依赖ICMP被防火墙拦得死死的,要么需要root权限跑原始套接字,要么行为特征太明显容易被流量审计设备盯上。我在几个不同规模的局域网里折腾过一轮之后,最后落地的方案是自己用Python写一个基于UDP的主机发现工具,核心思路是发UDP探测包、收ICMP端口不可达回包,然后手动解码IP头部来判断目标是否存活。
这个工具解决的问题很具体:在没有ICMP回显权限、或者目标主机禁ping的环境下,依然能判断主机是否在线。它适合有一定Python基础、了解基本网络协议、需要做内网资产清点的从业者参考。整篇内容我会从设计思路讲到代码实现,再到实际踩过的坑,尽量把每个"为什么这么写"都交代清楚,你照着抄作业也能跑起来。
先说清楚一个前提:这个工具只做"发现",不做端口扫描、不做漏洞探测,探测行为本身要控制频率和范围,只在你拥有明确授权的网络里使用。这一点不是客套话,是实操底线。
2. 整体设计思路与协议选型拆解
2.1 为什么选UDP而不是TCP SYN或ICMP
主机发现的本质是"发一个包,看有没有回应"。常见三条路:ICMP Echo(ping)、TCP SYN扫描、UDP探测。三条路各有各的坑。
ICMP Echo最直接,但太多主机和防火墙默认丢弃ICMP,尤其是一些加固过的服务器,ping不通但服务活得好好的。TCP SYN扫描准确率高,但半开扫描需要构造原始包,权限要求高,而且行为特征在流量层面非常显眼。UDP探测的思路是利用协议规范:当一个UDP包发往目标主机的某个关闭端口时,目标主机按RFC 792的规定应该回一个ICMP Type 3 Code 3(端口不可达)的报文。只要收到这个回包,就说明目标主机在线——因为回包是主机协议栈发出来的,主机不在线根本不会有任何回应。
这个方案的巧妙之处在于:它不依赖目标主机开放任何服务,也不依赖ICMP Echo是否被放行。很多防火墙会拦Echo请求,但对"端口不可达"这类错误回包相对宽松,因为它们属于正常的协议交互。实测下来,在禁ping的内网环境里,UDP探测的存活识别率明显高于纯ICMP方案。
当然它也有代价:需要能收到ICMP错误报文,这意味着要么有原始套接字权限,要么在特定网络位置。这一点后面会详细讲。
2.2 解码IP header的必要性
很多人写这类工具时直接用现成库解析回包,我一开始也这么干,后来发现两个问题。第一,ICMP错误报文里嵌套了原始IP头和UDP头,标准库的解析层级不一定对得上,容易拿错字段。第二,我需要从回包里提取"原始目标IP"来确认是哪台主机回的,这个信息藏在ICMP载荷的内层IP头里,不手动解码根本拿不到准确值。
所以这个工具的核心技术点其实是手动解码IP头部:外层IP头20字节、ICMP头8字节、内层原始IP头20字节、内层UDP头8字节,一层层剥开,从内层IP头的源地址字段反推出是哪台探测目标触发了这个回包。这个解码过程是整个工具准确性的关键,也是标题里"解码IP header"的由来。
2.3 整体架构分层
我把工具拆成四层,每层职责单一,方便调试:
| 层级 | 职责 | 关键实现 |
|---|---|---|
| 探测层 | 构造并发送UDP探测包 | 原始套接字或普通UDP套接字 |
| 接收层 | 抓取ICMP回包 | 原始套接字监听 |
| 解码层 | 逐层解析IP/ICMP/UDP头 | struct模块按偏移解包 |
| 调度层 | 并发控制、超时管理、结果汇总 | 多线程+队列 |
这样分层的好处是,解码层可以单独拿真实抓包数据做单元测试,不用每次都跑完整扫描。我在调试阶段就是把一段pcap里的回包字节流喂给解码函数,反复验证字段偏移对不对,比盲扫高效得多。
3. 核心细节解析与实操要点
3.1 IP头部结构逐字段拆解
IPv4头部固定20字节(不含选项),字段布局必须背下来,因为解码全靠偏移量。我用struct的格式串来解,先看标准布局:
| 偏移(字节) | 字段 | 长度 | 说明 |
|---|---|---|---|
| 0 | 版本+首部长度 | 1 | 高4位版本,低4位首部长度(单位4字节) |
| 1 | 服务类型 | 1 | 一般忽略 |
| 2-3 | 总长度 | 2 | 整个IP包长度,网络字节序 |
| 4-5 | 标识 | 2 | 分片重组用 |
| 6-7 | 标志+片偏移 | 2 | 高3位标志 |
| 8 | 生存时间TTL | 1 | 每经一跳减1 |
| 9 | 协议 | 1 | 1=ICMP, 6=TCP, 17=UDP |
| 10-11 | 首部校验和 | 2 | 校验用 |
| 12-15 | 源IP | 4 | 关键字段 |
| 16-19 | 目的IP | 4 | 关键字段 |
解码时最容易踩的坑是首部长度字段。低4位乘以4才是真实首部长度,通常是5(即20字节),但如果带选项就不是20了。我第一版代码硬编码偏移20去读ICMP头,遇到带选项的包直接解析错位,回包全乱。正确做法是先读第0字节算出首部长度,再动态定位后续字段。
import struct def decode_ip_header(data): # data 至少20字节 ver_ihl = data[0] version = ver_ihl >> 4 ihl = (ver_ihl & 0x0F) * 4 # 真实首部长度 total_len = struct.unpack('!H', data[2:4])[0] ttl = data[8] proto = data[9] src_ip = '.'.join(map(str, data[12:16])) dst_ip = '.'.join(map(str, data[16:20])) return { 'version': version, 'ihl': ihl, 'total_len': total_len, 'ttl': ttl, 'proto': proto, 'src_ip': src_ip, 'dst_ip': dst_ip, 'payload': data[ihl:] # 动态偏移,别硬编码20 }这段代码里payload: data[ihl:]是精髓,用算出来的首部长度切载荷,兼容带选项的情况。
3.2 ICMP错误报文的内嵌结构
收到ICMP Type 3(目的不可达)时,报文结构是:外层IP头 + ICMP头(8字节) + 原始IP头 + 原始UDP头 + 原始数据前8字节。ICMP头8字节布局:
| 偏移 | 字段 | 说明 |
|---|---|---|
| 0 | 类型 | 3表示不可达 |
| 1 | 代码 | 3表示端口不可达 |
| 2-3 | 校验和 | |
| 4-7 | 未使用/其他 | Type3下通常为0 |
关键在ICMP载荷里嵌套的原始IP头。这个内层IP头的目的地址就是我们探测的目标,源地址是目标主机自己。所以从内层IP头读源地址,就能确认是哪台主机回的包。这里有个细节:内层IP头同样要按首部长度动态偏移,不能假设20字节。
def decode_icmp_payload(icmp_payload): # icmp_payload 是去掉外层IP头后的ICMP部分 icmp_type = icmp_payload[0] icmp_code = icmp_payload[1] if icmp_type != 3: return None # 跳过8字节ICMP头,进入内嵌原始IP头 inner = icmp_payload[8:] inner_ip = decode_ip_header(inner) return { 'icmp_type': icmp_type, 'icmp_code': icmp_code, 'original_src': inner_ip['src_ip'], # 目标主机 'original_dst': inner_ip['dst_ip'], # 我们探测的地址 }3.3 套接字权限与平台差异
这是实操里最现实的一道坎。接收ICMP错误报文需要原始套接字(socket.SOCK_RAW),在Linux上需要root或CAP_NET_RAW能力,Windows上需要管理员权限。没有权限时socket()直接抛PermissionError。
我的处理策略是:发送端用普通UDP套接字(不需要特权),接收端用原始套接字。这样至少发送逻辑在无权限时也能跑,方便调试。另外不同平台对原始套接字的协议参数要求不同,Linux上socket(AF_INET, SOCK_RAW, IPPROTO_ICMP)即可,Windows上同样写法但需要管理员。跨平台代码里我会先尝试创建原始套接字,失败则降级提示。
注意:原始套接字收到的是完整IP包(含IP头),而普通套接字收到的是纯载荷。解码函数要按"是否含IP头"区分处理,否则会多剥一层导致错位。我建议统一用原始套接字接收,解码时从IP头开始,逻辑一致。
3.4 探测端口的选择逻辑
UDP探测要发往一个"大概率关闭"的端口,才能触发端口不可达。如果发往开放端口,目标可能正常处理不回错误,就收不到回包了。常见做法是选高位随机端口,比如40000以上,这些端口在绝大多数主机上都是关闭的。
但这里有个反直觉的点:某些系统对关闭端口的UDP包会限速回ICMP错误(RFC 1812建议每秒不超过一定数量),所以探测频率不能太高,否则回包被限速丢弃,造成漏判。我实测把单目标探测间隔控制在50ms以上,配合重试,识别率比较稳。
4. 完整实操流程与核心代码实现
4.1 环境准备与依赖确认
整个工具只依赖Python标准库,socket、struct、threading、queue、time、ipaddress,不需要装第三方包。Python版本建议3.8以上,因为用到了ipaddress模块的一些便利方法。
先确认运行环境有原始套接字权限:
# Linux下确认当前用户是否有CAP_NET_RAW getcap /usr/bin/python3 2>/dev/null # 或者直接跑一段测试 python3 -c "import socket; s=socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP); print('raw socket ok'); s.close()"如果报PermissionError,Linux下用sudo跑,或者给Python二进制加能力;Windows下用管理员终端。这一步不通过,后面全白搭。
4.2 发送端:构造UDP探测包
发送端逻辑很轻,普通UDP套接字发一个短载荷到目标的高位端口即可。载荷内容无所谓,我习惯发一个固定标记字符串,方便在抓包时识别。
import socket import random PROBE_PAYLOAD = b'PROBE_DISCOVERY_2024' def send_udp_probe(target_ip, port=None, timeout=1.0): if port is None: port = random.randint(40000, 60000) s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.settimeout(timeout) try: s.sendto(PROBE_PAYLOAD, (target_ip, port)) return port except OSError: return None finally: s.close()这里sendto对关闭端口通常不会立即报错(UDP无连接),错误会以ICMP形式异步回来,所以发送端只管发,判断交给接收端。
4.3 接收端:原始套接字抓ICMP回包
接收端是核心。开一个原始套接字监听ICMP,循环recvfrom,每收到一个包就解码,判断是不是Type 3 Code 3,再提取内层原始目标地址。
import socket import struct def listen_icmp(stop_event, result_queue): try: sniffer = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP) except PermissionError: print('需要原始套接字权限,请用管理员/root运行') return sniffer.settimeout(0.5) while not stop_event.is_set(): try: packet, addr = sniffer.recvfrom(65535) except socket.timeout: continue # packet 含完整IP头 outer = decode_ip_header(packet) if outer['proto'] != 1: # 只要ICMP continue icmp_data = outer['payload'] if len(icmp_data) < 8: continue info = decode_icmp_payload(icmp_data) if info and info['icmp_code'] == 3: result_queue.put(info['original_dst']) sniffer.close()注意recvfrom的缓冲区给65535,IP包最大就这么大,给小了会截断导致解码失败。另外原始套接字在Linux上会收到本机发出的所有ICMP,包括我们自己的探测包触发的回包,这正是我们要的。
4.4 调度层:并发探测与超时管理
单线程串行探测太慢,一个/24网段254个地址,每个等1秒就是4分钟。我用"发送线程池+单接收线程"的模式:多个线程并发发探测包,一个线程专门收,结果进队列,主线程汇总。
import threading import queue import ipaddress import time def discover_hosts(network_cidr, max_workers=50, wait=3.0): net = ipaddress.ip_network(network_cidr, strict=False) targets = [str(ip) for ip in net.hosts()] result_queue = queue.Queue() stop_event = threading.Event() receiver = threading.Thread(target=listen_icmp, args=(stop_event, result_queue), daemon=True) receiver.start() time.sleep(0.2) # 给接收线程一点启动时间 def worker(ip): send_udp_probe(ip) time.sleep(0.05) # 控制频率,避免触发限速 threads = [] for ip in targets: t = threading.Thread(target=worker, args=(ip,)) t.start() threads.append(t) if len(threads) >= max_workers: for t in threads: t.join() threads = [] for t in threads: t.join() time.sleep(wait) # 等回包陆续到达 stop_event.set() receiver.join(timeout=1.0) alive = set() while not result_queue.empty(): alive.add(result_queue.get()) return sorted(alive)max_workers控制并发度,50是个折中值,太高容易把本机文件描述符打满,也容易触发目标侧限速。wait给回包留到达时间,内网一般3秒够,跨网段可以加到5秒。
4.5 解码函数的完整实现与验证
把前面几段拼起来,解码链路是:外层IP头 → ICMP头 → 内层IP头 → 提取源地址。完整验证方法是用一段真实抓包字节流喂进去,对比Wireshark里看到的字段。我在调试时专门构造了一个已知回包,逐字节打印偏移,确认每个字段位置无误后才上真实网段。
def full_decode(packet): outer = decode_ip_header(packet) if outer['proto'] != 1: return None icmp = outer['payload'] if len(icmp) < 8 or icmp[0] != 3: return None inner = icmp[8:] if len(inner) < 20: return None inner_ip = decode_ip_header(inner) return { 'responder': inner_ip['src_ip'], 'probed': inner_ip['dst_ip'], 'code': icmp[1] }responder是回包主机(即存活主机),probed是我们探测的地址,两者一致说明匹配成功。实际网络里可能有其他ICMP流量混入,靠probed是否在本次探测目标集合里来过滤,避免误报。
5. 常见问题与排查技巧实录
5.1 收不到任何回包怎么办
这是最高频的问题。排查顺序我总结成一张表:
| 现象 | 可能原因 | 排查方法 |
|---|---|---|
| 完全无回包 | 无原始套接字权限 | 检查是否root/管理员运行 |
| 完全无回包 | 目标网段禁ICMP错误 | 换TCP探测对比验证 |
| 部分目标无回包 | 探测频率过高被限速 | 加大间隔、增加重试 |
| 回包地址对不上 | 解码偏移错误 | 用已知抓包单测解码函数 |
| 程序卡死 | 接收线程阻塞 | 检查settimeout是否设置 |
我遇到过一次"完全无回包",折腾半天发现是跑在容器里,容器网络命名空间隔离导致原始套接字收不到宿主网段的ICMP。换到宿主机跑就正常了。这类环境问题比代码问题更隐蔽,建议先在能ping通的环境里验证工具本身。
5.2 误报与漏报的平衡
误报主要来自把无关ICMP流量当成探测结果。解决办法是维护一个"本次探测目标集合",只有回包里的probed字段命中集合才计入。漏报主要来自限速和丢包,解决办法是重试:每个目标发2到3次探测,间隔拉长,任意一次收到回包即判定存活。
提示:重试次数不是越多越好。我试过每个目标发5次,识别率提升有限,但扫描时长翻了几倍,还更容易触发目标侧防护。2到3次是性价比最高的区间。
5.3 跨平台解码差异
Windows和Linux在原始套接字返回的数据格式上有细微差别。Linux返回完整IP包,Windows某些情况下返回的包结构需要额外确认。我的做法是在解码入口先判断第一个字节的高4位是不是4(IPv4版本号),不是就跳过或按平台调整。这个判断很便宜,但能挡掉不少格式错位导致的崩溃。
另外字节序要统一用网络字节序(struct的!前缀),别用本机字节序,否则换台机器结果就变了。这个坑我在早期版本踩过,同一段代码在开发机正常、部署机全错,查了半天是字节序问题。
5.4 性能与安全的边界
一个/24网段,50并发,单目标间隔50ms,整体扫描时间大概十几秒,这个速度对内网资产清点够用。再往上提并发,收益递减且风险上升。我个人的经验是:扫描行为要可解释、可控制,频率、范围、时长都要有明确边界,别为了快把网段打爆。
还有一点,探测载荷我固定用一个无害标记字符串,不带任何攻击性内容,这样即使被流量审计记录,行为也是清晰可解释的。工具是工具,怎么用取决于人,边界意识得自己守住。
6. 我在实际使用中的几点体会
这套工具我前后迭代了三个版本,第一版硬编码偏移、第二版加了动态首部长度、第三版才把并发和重试理顺。最大的体会是:解码的健壮性比扫描速度重要得多。一个解码错位的工具,扫得再快结果也是错的,反而误导判断。所以我现在写这类工具,第一件事永远是把解码函数用真实抓包数据单测透,再谈其他。
另一个体会是权限和环境适配要前置。很多人卡在"跑不起来"这一步就放弃了,其实大部分是权限或网络命名空间的问题,跟代码无关。先把最小可运行版本跑通——哪怕只探测一个已知存活的主机——再逐步扩大范围,这个顺序能省掉大量无效调试。
后续这个工具还能往两个方向扩展:一是把结果导出成结构化格式,方便和资产管理系统对接;二是增加对IPv6的支持,不过IPv6的ICMPv6错误报文结构不同,解码逻辑要重写,这是另一个话题了。眼下这个版本,应付常规内网清点已经够用。