ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

ICMP报文构造与网络诊断:Python raw socket实战

ICMP报文构造与网络诊断:Python raw socket实战

简介:围绕ICMP报文构造与验证整理的一份小型代码资料,适合正在学习TCP/IP协议、网络编程及网络诊断的初、中级读者,也可供网络管理员排查连通性问题时参考。内容聚焦ICMP差错报告与查询报文的类型、代码及数据字段,演示如何将ICMP报文封装进IP数据报,并借助抓包工具核对发送与接收结果,帮助读者理解ping回显请求/应答在协议层的实际形态。压缩包共3个文件,约5KB,包含cpp源码、h头文件与txt说明文档,分别承载核心实现、结构声明和阅读指引,体量轻便,便于结合现有工程对照研读。已有1608人学习或下载,说明其在小巧示例层面具有一定参考价值;读者可从中获得报文构造思路、字段设置方法、IP封装要点以及抓包验证与排错线索,为后续网络协议实验和安全分析打基础。

1. 排障时那行“请求超时”,其实值 8 个字节

上周帮朋友看一台自研网关,ping返回“请求超时”,但业务端口连通正常,日志里也没有异常。命令行ping只给你一行结论,给不了每跳 RTT、给不了回包里的字段,更没法告诉你对端到底有没有按协议填 Type 和 Code。这就是一份 ICMP 数据包构造的源码最实在的用处:自己把 8 字节 ICMP 头和载荷拼出来,发出去,再把回来的包拆开逐字段看。读完这份代码,你能拿到一个不依赖系统ping的可运行探针,能改 Type/Code 构造各种探测报文,能顺着 TTL 阶梯跑一遍路径,也能在没有第三方 SDK 的环境里做链路质量采样。适合正在写网络诊断工具、做设备联调、或者想把 TCP/IP 协议从课本搬到手上的人。

2. 报文结构拆解:ICMP 头这 8 个字节怎么填

2.1 Type 与 Code 的组合表,先认清自己要发哪一种

ICMP 报文头固定 8 字节,按顺序是 Type(1 字节)、Code(1 字节)、Checksum(2 字节)、后面 4 字节随 Type 变化。Echo 类报文这 4 字节拆成 Identifier(2 字节)和 Sequence Number(2 字节);Destination Unreachable 类则用来携带触发该报文的原始 IP 头和 8 字节数据。字段名背下来不难,难的是知道什么时候该填哪个组合,下面这张表我一般贴在代码注释里。

TypeCode名称实际会遇到的场景
00Echo Reply对端回应你发出的 Echo Request
30Network Unreachable路由表里没有到目标的路径
31Host Unreachable目标网段可达,ARP 拿不到对端 MAC
33Port UnreachableUDP 打到没人监听的端口,对端回这个
34Fragmentation Needed报文超过 MTU 且置了 DF 位,这条最关键
50~3Redirect网关告诉你换一条更优路径
80Echo Request你主动发出去的探测包
110Time Exceeded(TTL)中间路由器把 TTL 减到 0,回这个
111Time Exceeded(重组)分片重组超时
13 / 140Timestamp Request / Reply不带状态的时间探测

要留意的点:除了 Type 3、5、11 之外,绝大多数 Type 的 Code 都固定为 0,填错之后对端不会报错,而是静默丢弃,这是新手最容易误以为“网络不通”的地方。另外 Type 3 的报文里,ICMP 头后面还会跟着触发它的那个原始 IP 头的前 20 字节加至少 8 字节数据,解析时要按ihl跳过 IP 头,再按外层 ICMP 头长 8 字节往后偏移,才能摸到内层内容。做 MTU 探测的人得靠这段内层信息判断哪个流被卡住了。

2.2 校验和:RFC 1071 的反码求和与三个易错点

Checksum 覆盖范围是整个 ICMP 报文(头 + 数据),算法是 16 位反码求和。看起来简单,实际写起来翻车率很高,我见过不止一个人在这四行里卡一个下午。

def checksum(data: bytes) -> int: """RFC 1071 的 16 位反码求和,返回值直接填进 Checksum 字段""" if len(data) % 2: data += b'\x00' # 奇数字节必须补零,不能直接跳过最后一位 total = 0 for i in range(0, len(data), 2): total += (data[i] << 8) | data[i + 1] # 按网络字节序取 16 位 total = (total & 0xFFFF) + (total >> 16) # 每次累加后回卷进位,避免溢出 return (~total) & 0xFFFF # 取反并截断到 16 位

逻辑说明:把数据按两个字节一组当成大端 16 位整数累加,累加过程中一旦超过 0xFFFF,就把高位折回低位继续加(这一步叫回卷),最后对结果取反。参数只有一个data,传进去的必须是校验和字段已经置 0的完整报文,否则算出来的值永远对不上。

三个易错点挨个说。第一,奇数字节长度必须补一个\x00,不补的话最后一字节会被漏掉,长载荷(比如 15 字节的魔术字)几乎必错。第二,回卷这一步很多人写成先全加完再处理溢出,遇到大载荷会丢进位,正确做法是边加边回卷。第三,字节序。struct.pack('!BBHHH', ...)里的!就是网络字节序,漏掉它在本机 x86 上跑出来的包 Wireshark 会直接标 “checksum incorrect”,但本机回环测试可能又“看起来正常”,属于典型的玄学问题。验证手段很简单:抓一次系统ping发出的包,把它的 ICMP 段抠出来喂给你的函数,算出来是 0 就说明算法对了(含校验和的报文再算一遍结果应为 0)。

2.3 载荷设计:时间戳、序列号与魔术字

载荷部分协议没规定内容,但工程上得放点有用的东西。我一般按三段排:开头 8 字节放发送时刻(4 字节秒 + 4 字节微秒),中间 4 字节放一个固定的魔术字,剩下的用递增字节填充。

放时间戳是为了不依赖本地记录就能算出 RTT——收到回包时用当前时间减去载荷里的发送时刻,多进程、多线程并发跑也不会串账。放魔术字是为了识别“这包是不是我发的”。系统ping也在发 ICMP,identifier 撞车的概率虽然低,但在容器共享网络命名空间的环境里并不罕见,一个固定魔术字能省掉大量排查。用递增字节填充则是为了做链路误码检测,某些交换机在极端负载下会把载荷改掉,回包里的填充段和发出的对不上,就说明中间设备出了问题。

载荷长度直接影响是否会触发分片。默认我写 56 字节,加上 8 字节 ICMP 头和 20 字节 IP 头刚好 84 字节,远小于 1500 的典型 MTU,路径上不会分片,测出来的 RTT 干净。要做 MTU 探测时再专门放大,这一点在第 4 章展开。

3. 构造链路:从 raw socket 逐字节拼装到 scapy 分层写法

3.1 Linux raw socket 版本:内核帮你补 IP 头

构造 ICMP 最直接的方式是AF_INET+SOCK_RAW+IPPROTO_ICMP。这里有个必须先说清的坑:用IPPROTO_ICMP创建的套接字,内核会自动帮你补上 IP 头,你只管拼 ICMP 部分,长度字段、源地址、TTL 全由内核填。只有当你用IPPROTO_RAW或者开了IP_HDRINCL时,才需要自己拼 IP 头。网上很多示例把这两条路混着写,照抄下来要么发包失败,要么发出去对端收不到。

import os, struct, socket, time ICMP_ECHO_REQUEST = 8 ICMP_ECHO_REPLY = 0 MAGIC = b'PRB0' # 魔术字,用来区分自己的包和系统 ping def build_echo(seq: int, payload: bytes, ident: int) -> bytes: """拼一个完整的 ICMP Echo Request,返回可以直接 sendto 的字节串""" body = struct.pack('!d', time.time()) + MAGIC + payload # 8 字节时间戳打头 header = struct.pack('!BBHHH', ICMP_ECHO_REQUEST, 0, 0, ident, seq) csum = checksum(header + body) # 此时校验和字段是 0 header = struct.pack('!BBHHH', ICMP_ECHO_REQUEST, 0, csum, ident, seq) return header + body

逻辑说明:先算载荷,再算把头(校验和位置填 0),然后两次pack——第一次只为拿长度对齐的字节串去算校验和,第二次把算出来的csum填回去。参数里ident建议传os.getpid() & 0xFFFF,seq从 1 开始单调递增,payload长度控制在 56 字节以内。

权限是绕不过去的一道坎。raw socket 需要CAP_NET_RAW,普通用户直接跑会拿到PermissionError: [Errno 1] Operation not permitted。开发阶段sudo最省事,但交付时我一般用能力位而不是给整个脚本加 sudo:

sudo setcap cap_net_raw+eip /usr/bin/python3.11 # 验证,能打印出 cap_net_raw 就说明生效了 getcap /usr/bin/python3.11

注意这条命令是给解释器加能力,等于这台机器上所有 Python 脚本都能开 raw socket,生产环境慎用,更稳妥的做法是打包成独立可执行文件后只给那个文件加能力位。

3.2 收发与匹配:identifier 和 sequence 是唯一的后悔药

发出去只是第一步,收回来才是麻烦的开始。raw socket 收到的是带着 IP 头的完整数据包,得先按ihl切片才能看到 ICMP 段。同时因为同一个套接字会收到本机所有 ICMP 流量,匹配逻辑必须写严。

def ping_once(host: str, seq: int = 1, timeout: float = 1.0, ttl: int = 64): ident = os.getpid() & 0xFFFF pkt = build_echo(seq, b'\x00' * 56, ident) with socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP) as sock: sock.settimeout(timeout) sock.setsockopt(socket.IPPROTO_IP, socket.IP_TTL, ttl) # 控制跳数,traceroute 靠它 start = time.perf_counter() sock.sendto(pkt, (host, 0)) # 端口号对 ICMP 无意义,填 0 while True: data, addr = sock.recvfrom(2048) ihl = (data[0] & 0x0F) * 4 # IP 头长度以 4 字节为单位 icmp = data[ihl:] rtype, rcode, _, rid, rseq = struct.unpack('!BBHHH', icmp[:8]) if rtype != ICMP_ECHO_REPLY: continue # 不是回显应答,交给上层别的逻辑 if rid != ident or rseq != seq: continue # 别人的回包,丢掉 rtt = (time.perf_counter() - start) * 1000 return rtt, addr[0], icmp[8:] # 载荷原样带回,供校验魔术字

逻辑说明:setsockopt里的IP_TTL在 traceroute 场景下会反复改,这里先固定 64。recvfrom拿到的data从第 0 字节起就是 IP 头,(data[0] & 0x0F) * 4取的是 IHL 字段,常见值 20,带有选项时会更长,写死 20 是另一个高频翻车点。匹配条件用了rtype、rid、rseq三重判断,少任何一个都可能拿到系统ping或其他进程的回包,表现为 RTT 忽大忽小、时对时错。

参数含义再强调一下:seq必须和构造时一致,否则永远匹配不上;timeout是settimeout的秒数,局域网 0.5 秒足够,跨地域建议 2 秒,设太小会把正常的慢回包误判成丢包。

3.3 scapy 版本:二十行写完成同样的功能

如果项目里不介意引入依赖,scapy 能把上面这一堆字节操作压缩到几行,协议分层的表达也更直观。

import os from scapy.all import IP, ICMP, Raw, sr1 def ping_scapy(host: str, seq: int = 1, ttl: int = 64, timeout: float = 1.0): ident = os.getpid() & 0xFFFF pkt = IP(dst=host, ttl=ttl) / ICMP(type=8, id=ident, seq=seq) / Raw(load=b'\x00' * 56) resp = sr1(pkt, timeout=timeout, verbose=0) # sr1 = send and receive 1,只等第一个回包 if resp is None: return None if resp.haslayer(ICMP) and resp[ICMP].type == 0: return resp[IP].src, resp[ICMP].payload.load return None

/是 scapy 的重载运算符,表示协议分层,等价于“把后者的字节接到前者后面”。sr1会自动处理 IP 头切片和校验和计算,省掉了手写checksum的环节,收包时按层名访问字段也比struct.unpack可读。代价是启动慢(导入 scapy 通常要 0.3~1 秒),打包体积大,某些精简镜像里还会缺依赖。两条路的对比如下:

维度raw socketscapy
启动开销几乎为零首次导入数百毫秒
依赖标准库需要额外安装
字段可读性依赖 struct 格式串按层名访问,直观
精细控制完全可控,能构造畸形包需要手动改字段或Raw
适合场景交付给终端用户的轻量工具内部调试、教学、快速验证

我自己的做法是两套都留:核心链路用 raw socket,保证没有第三方依赖;写测试用例和抓包对齐时用 scapy 起一个对照实现,两边 RTT 差在 1 毫秒以内,基本可以确认构造逻辑没问题。

4. 从 echo 到诊断:TTL 阶梯与 DF 位探测

4.1 Traceroute 的 TTL 递增循环与 Type 11 解析

Echo 只能告诉你“通不通”,路径信息要靠 TTL 一层层往上加。原理不复杂:TTL 减到 0 时,中间路由器会回一个 Type 11 的 Time Exceeded,把回包源地址记下来就是这一跳的地址;直到某一跳回的是 Type 0 Echo Reply,说明到终点了。

def trace(host: str, max_hop: int = 30, timeout: float = 1.5): ident = os.getpid() & 0xFFFF for ttl in range(1, max_hop + 1): pkt = build_echo(seq=ttl, b'\x00' * 32, ident) with socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP) as sock: sock.settimeout(timeout) sock.setsockopt(socket.IPPROTO_IP, socket.IP_TTL, ttl) start = time.perf_counter() sock.sendto(pkt, (host, 0)) try: data, addr = sock.recvfrom(2048) except socket.timeout: print(f'{ttl:>2} * request timeout') continue rtt = (time.perf_counter() - start) * 1000 ihl = (data[0] & 0x0F) * 4 rtype, rcode = struct.unpack('!BB', data[ihl:ihl + 2]) if rtype == 11: # Time Exceeded,中间跳 print(f'{ttl:>2} {addr[0]:<16} {rtt:6.2f} ms') elif rtype == 0: # Echo Reply,到终点 print(f'{ttl:>2} {addr[0]:<16} {rtt:6.2f} ms reached') return

逻辑说明:外层循环把 TTL 从 1 加到 30,每轮新建一个套接字(复用一个套接字也可以,但要记得setsockopt在每次发送前重设)。seq直接用 TTL 值,省掉额外的计数器。收到 Type 11 打印中间跳,收到 Type 0 打印终点并返回。超时用*占位,和系统traceroute的输出风格一致,方便对照。

这里有个容易被忽略的细节:Type 11 的报文里,ICMP 头之后还塞着触发它的原始 IP 头 + 8 字节数据,也就是你发出去的 Echo 报文的前 28 字节。想确认“这个超时确实是我这轮发的包触发的”,得从data[ihl+8:]再切一层,比对内层的 identifier 和 seq。不做这层校验的话,路径上中间设备主动发的 ICMP 超时会被误记成正常跳,输出里就会冒出莫名其妙的地址。这个血泪经验在跨运营商链路上特别值钱。

4.2 DF 位与 MTU 探测:Type 3 Code 4 才是有效信号

MTU 探测的逻辑是:发一个指定长度的 Echo Request,把 IP 头的 DF(Don't Fragment)位置 1,中间链路 MTU 不够时,路由器不能分片,只能回 Type 3 Code 4 告诉你“需要分片但设了 DF”,同时报文里会带上它这一跳的 MTU 值(新版 ICMP 规范里放在 ICMP 头后 2 字节)。把长度二分搜索一遍,就能测出路径 MTU。

import socket, struct IP_MTU_DISCOVER = 10 # Linux 下这两个常量标准库没暴露,得自己定义 IP_PMTUDISC_DO = 2 def probe_mtu(host: str, low: int = 576, high: int = 1500) -> int: """二分找路径 MTU,返回不触发分片的最大载荷长度""" while low < high: mid = (low + high + 1) // 2 with socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP) as sock: sock.settimeout(2.0) sock.setsockopt(socket.IPPROTO_IP, IP_MTU_DISCOVER, IP_PMTUDISC_DO) pkt = build_echo(seq=mid, b'\x00' * (mid - 8 - 20), os.getpid() & 0xFFFF) sock.sendto(pkt, (host, 0)) try: data, _ = sock.recvfrom(4096) ihl = (data[0] & 0x0F) * 4 rtype, rcode = struct.unpack('!BB', data[ihl:ihl + 2]) if rtype == 3 and rcode == 4: high = mid - 1 # 太大,被拒 else: low = mid # 能过 except socket.timeout: high = mid - 1 # 超时也按过大处理,保守收敛 return low

逻辑说明:IP_MTU_DISCOVER设成IP_PMTUDISC_DO等价于在 IP 头上置 DF 位,这是不自己拼 IP 头时最省事的做法。载荷长度换算成mid - 8 - 20,减掉 ICMP 头和 IP 头才是真正的 data 长度。二分区间从 576(IPv4 最小重组缓冲)到 1500(以太网典型 MTU),如果目标是 PPPoE 或隧道链路,实际值可能落在 1400 出头,二分能很快收敛。

注意超时的情况我也当成“过大”处理,这是一种保守策略——丢包和 MTU 超限在客户端看起来都是没收到回包,不做区分会让结果偏大。要区分的话,得在抓到 Type 3 Code 4 时把内层报文的 identifier 和 seq 拿出来比对,确认这条错误确实是冲着这轮探测来的。

5. 避坑与排查:构造器跑不通时按这五条对表

5.1 权限、防火墙、序列号:收发链路上的三条翻车记录

第一条,权限。

现象:脚本一跑就抛PermissionError: [Errno 1] Operation not permitted,换成sudo就正常。

原因:创建SOCK_RAW套接字需要CAP_NET_RAW能力,普通用户默认没有。部分发行版的ping命令之所以能免 sudo,是因为它带着setuid位或能力位。

解决:开发期用sudo快速验证,交付前把脚本打成独立可执行文件再setcap cap_net_raw+eip,别给整个解释器加能力。容器环境里除了容器本身的 capability,还要确认--cap-add=NET_RAW有没有传进去,这两层权限是叠加的。

第二条,防火墙。

现象:sendto返回的字节数正常,说明包发出去了,但recvfrom一直超时;同一台机器上用系统ping打同一个地址却通。

原因:本机iptables/nftables或云上安全组把入向 ICMP 丢了。还有一种情况是ping命令走的是SOCK_DGRAM的 ICMP 套接字(某些系统上带net.ipv4.ping_group_range支持),走了不同的路径,所以行为不一致。

解决:先在发送端抓包确认包真的出去了,再在接收端抓包确认回包到了网卡。两端都有包而应用收不到,八成是防火墙规则;只有一端有包,问题在路由。排查时顺手iptables -L -n -v看一眼 ICMP 相关的 DROP 计数,比盲猜快得多。

第三条,序列号与 identifier 撞车。

现象:RTT 数据忽大忽小,偶尔出现几百毫秒的异常值,或者回包的载荷和发出去的对不上。

原因:identifier 用了固定值 0,或者每次调用随机生成,同一台机器上并发跑多个探测进程时互相串包。系统ping也在发 ICMP,撞车概率不低。

解决:identifier 固定用os.getpid() & 0xFFFF,sequence 单调递增,收到包后三重校验 type、id、seq 全对上才算数。再加一层魔术字比对载荷,能挡住绝大多数串包。这个后悔药必须在第一版就吃上,等线上数据脏了再回头改,历史数据全得重采。

5.2 校验和、字节序、IP 头长度:报文内容上的两条翻车记录

第四条,校验和算错。

现象:Wireshark 里 ICMP 那一行标红,提示 checksum incorrect,但对端有时候居然会回包,有时候不回,表现非常随机。

原因:三种可能——计算校验和时没把 Checksum 字段置 0;报文长度是奇数没补零;struct.pack漏了!导致按小端写入。有些设备对校验和不做校验直接回,有些严格丢弃,所以你看到的现象就是玄学。

解决:写完checksum先单独测。拿系统ping发出的包做样本,把 ICMP 段(含原校验和)喂进去,结果应该是 0。再用自己构造的包算一遍然后重新计算,同样应该是 0。两个测试都过,算法就没问题。抓包确认字节序,0800 0000这种才是对的。

第五条,IP 头长度写死 20。

现象:解析回包时字段全错位,rtype打印出来是个莫名其妙的数字,有时是 69 有时是 0。

原因:recvfrom返回的数据从 IP 头开始,代码里写死了data[20:]。某些路径上 IP 头带选项(比如记录路由),长度会变成 24 甚至更长,写死 20 就把 ICMP 头的中间几个字节当成了 Type。

解决:统一用ihl = (data[0] & 0x0F) * 4动态计算偏移,任何需要切 ICMP 段的地方都用这个变量。调试期我会在切片前打一行data[:32].hex(),肉眼确认前几个字节的版本号是 4、IHL 是 5 或更大,这个习惯救过我好几次。

6. 二次开发:把它改成你手上那台设备的专用探针

这套代码本身不复杂,价值在于它是个干净的骨架。真要落到具体项目里,我一般按“探针注册 + 统一采集”的路子改,而不是在主流程里堆 if-else。

先加一个极简的注册表,每个探测能力是一个函数,靠装饰器挂进去:

PROBES = {} def probe(name: str): def deco(fn): PROBES[name] = fn return fn return deco @probe('echo') def run_echo(host, **kw): return [ping_once(host, seq=i) for i in range(1, 5)] # 连发 4 个取统计 @probe('mtu') def run_mtu(host, **kw): return {'path_mtu': probe_mtu(host, high=kw.get('high', 1500))}

PROBES用字典存函数引用,@probe('echo')在模块导入时自动注册,主流程只认字符串命令名,新增探测类型不用改调度代码。参数统一收**kw,避免每加一个探针就改一次函数签名——这个设计我是在改到第七个探针时才想明白的,前期图省事直接传位置参数,后期加个超时选项要动五处调用。

几个具体扩展点,按投入产出排:

扩展方向改哪里验证方式
加时间戳请求(Type 13)新增build_timestamp,复用checksum抓包看回复 Type 是不是 14
加 IPv6 支持换AF_INET6+IPPROTO_ICMPV6,校验和算法含伪首部与系统ping6结果对齐
输出改成 JSON在run_*返回值上包一层序列化用jq过一遍确认字段完整
加丢包率统计在run_echo里统计 None 比例人为丢包 20% 看统计是否吻合
接入监控把 RTT 推到时间序列库对比本地抓包时间戳

IPv6 那条要特别注意:ICMPv6 的校验和覆盖范围包含伪首部(源地址、目的地址、长度、下一头部),跟 IPv4 完全不是一回事,直接把checksum拿来用会全错。我一般会为 v6 单独写一份,别图省事去复用。

验证方法上我养成一个固定套路:任何新探针先打127.0.0.1,这时候 RTT 应该在 0.1 毫秒量级,Type 和 Code 必然正确,排除了链路因素;再打网关,验证二层可达和 ARP;最后才上真实目标。三步都过,才认为这个探针写完了。跳过前两步直接上真机,出问题时根本分不清是自己构造错了还是链路不通。

从那以后我每次动这套代码,改完都强制拿回环地址跑一遍全量探针,再抓一次包对齐ping的结果,两个都对上才允许合进去。这个习惯拦下过至少三次字节序和切片偏移的低级错误。希望帮到你。

本文还有配套的精品资源,点击获取

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进