Scapy实战:Python数据包构造、收发解析与网络排障
前段时间排查一个局域网里的通信异常抓包软件翻来覆去看了半天最后定位到一个自己拼的畸形报文上。当时我就在想要是能直接用代码把包拼出来、发出去、再把回包收回来解析整个流程脚本化比在图形界面里点来点去省事得多。Python 的 scapy 库干的就是这个活——它把网络协议的每一层都抽象成可以随手拼装的 Python 对象你写几行代码就能构造出一个完整的数据包发出去收回来拆开看。它不是那种调用一个函数帮你搞定一切的高层封装恰恰相反它把细节全摊在你面前你可以精确到某个标志位、某个序列号去改。这篇文章面向的是已经会一点 Python、想上手网络层实操的读者也适合平时做协议验证、内网排障、教学演示的朋友。下面我按自己摸索的顺序从装环境到写出能跑的脚本把坑和技巧一并说清楚。1. 环境准备与版本选择的那些事1.1 三个平台的安装差异scapy 的安装本身不复杂一条pip install scapy就能把纯 Python 部分装好但真正让人卡住的是底层抓包驱动。它依赖系统提供的数据链路层访问能力Linux 上靠的是 libpcapmacOS 自带 BSD 的抓包接口基本可用Windows 上则必须额外装一个 Npcap 驱动而且安装时要勾选支持 WinPcap API 兼容模式这个选项否则 scapy 起来之后conf.iface会是空的或者一发包就抛出找不到网卡的异常。我第一次在 Windows 上折腾就是因为漏勾了那个框白白查了半个多小时。Linux 侧的坑主要在权限。普通用户默认没有打开原始套接字raw socket的权限直接用会收到 Permission denied。最省事的做法是临时sudo跑脚本但长期这么干不优雅。比较稳妥的方式是给 Python 解释器加CAP_NET_RAW和CAP_NET_ADMIN两个能力用setcap命令打上去这样普通用户也能抓包发裸包。需要注意的是每次升级 Python 解释器这个能力都会丢得重新打一次这个细节文档里很少提但在自动化环境里特别容易踩。macOS 上还有一个隐性的问题较新版本系统对网卡访问做了限制某些虚拟网卡比如容器或虚拟化软件创建的网桥在列表里能看到但抓包时只能收到本机发出的包收不到外部进来的。这种情况不是你代码写错了换物理网卡验证一下就能确认。提示安装完先别急着写业务代码跑一句from scapy.all import conf; print(conf.iface)能打印出网卡名就说明链路层通了。这一步没通过后面所有操作都是空中楼阁。1.2 依赖库与能跑起来的最小验证scapy 本身是可选依赖设计核心功能不装额外库也能用但如果你想解析 pcap 文件、做 TLS 层解析、画个简单的报文图就需要补装几个。常见的搭配是pip install scapy[complete]它会把 cryptography、matplotlib 这些一并拉下来。我的建议是先用最小安装跑通确定自己真的需要哪一块再补不然装一堆用不上的东西依赖冲突排查起来更烦。最小验证我一般分三步走。第一步构造一个 ICMP 包并打印它的结构确认分层拼装正常第二步用sr1向本机回环地址发一个包确认发送接收链路正常第三步用sniff抓几个包确认嗅探功能正常。这三步覆盖了 scapy 最核心的三条腿构造、收发、嗅探任何一步失败都能快速定位问题在哪个环节。from scapy.all import IP, ICMP, sr1, sniff # 第一步构造只看不设置网络 pkt IP(dst127.0.0.1) / ICMP() print(pkt.summary()) # 第二步发送并等一个回包 resp sr1(pkt, timeout2, verboseFalse) print(resp.summary() if resp else 没收到回包) # 第三步抓三个包看看 sniff(count3, prnlambda p: p.summary())回环地址做验证有个好处完全不出本机不打扰任何人也避开了防火墙和路由的干扰。等这三步都过了再去碰真实网段。我见过不少人一上来就往外部地址发包结果收不到回应分不清是自己代码写错了、网络不通、还是对方策略拦了白白绕远路。1.3 虚拟环境与版本兼容的取舍scapy 的版本迭代挺活跃2.4 到 2.5 再到 2.6API 有过几次调整比如网卡列表函数在 2.5 之后推荐用show_interfaces()替代老的show_if_list()路由表相关的一些属性也有变动。如果你照着两三年前的教程写很可能遇到函数不存在或者属性为空这类问题。我的习惯是固定版本在虚拟环境里锁住scapy2.5.0这种写法等项目稳定了再考虑升级。虚拟环境这块还有个细节scapy 的配置对象conf是全局单例脚本里改过之后同一个进程内后续所有操作都会继承。这在写单文件小脚本时没什么但如果你把代码封装成模块被别的程序调用就可能出现别人调你的函数结果发现自己程序的默认网卡被改了这种诡异现象。解决办法是在自己模块内部显式指定参数不去动全局配置需要改的时候用上下文管理器包一层用完恢复。2. 分层构造把数据包当成积木来搭2.1 / 操作符背后的对象模型scapy 最核心的设计就是那个斜杠。IP()/TCP()看上去像除法实际上是重载过的运算符作用是把两个 Packet 对象按层次串起来。前一个对象成为后一个的底层承载后一个对象挂在前一个的 payload 属性上。这个模型非常贴近真实的协议栈以太网头承载 IP 头IP 头承载 TCP 头TCP 头承载应用数据。你用斜杠连起来就等于在描述谁装在谁里面。理解了这个模型很多操作就顺理成章了。比如想取某一层用下标访问pkt[IP]或者pkt[TCP]它会在包里找对应类型的层想判断有没有某一层用IP in pkt或者pkt.haslayer(IP)想从某一层开始截取剩下的部分用pkt[IP:]切片。这些操作在解析复杂报文时特别顺手比如你收到一个带 VLAN 标签的包想跳过以太网头和 802.1Q 头直接看 IP 层一行pkt[IP]就拿到了不用自己去算偏移量。各层的字段用属性方式读写。ip IP(dst192.168.1.1); ip.ttl 32这样就行。这里有个容易忽视的点如果你修改了某一层的某个字段而这个字段的值会影响另一层比如 IP 头的长度字段、TCP 的校验和scapy 在发送时会自动重算但你在内存里直接show()看到的还是旧值。想看重算之后的样子用show2()它会先做一遍组装再显示。这个差别我最初没注意一直纳闷为什么改完长度字段显示没变。from scapy.all import Ether, IP, TCP, Raw pkt Ether() / IP(dst192.168.1.10) / TCP(dport80) / Raw(loadhello) pkt.show() # 显示当前内存中的字段值 pkt[IP].ttl 12 # 改 TTL pkt.show2() # 重新组装后再显示长度、校验和会更新2.2 字段默认值、随机值与自动计算scapy 给每个字段都设了默认值这一点对新手既友好又危险。友好在于你写IP()/TCP()就能得到一个结构完整、可以直接发出去的包危险在于你可能根本没意识到某些字段是怎么来的。比如 TCP 的源端口默认是 20IP 头的 TTL 默认是 64以太网的源 MAC 在没指定时会取当前网卡的地址目的 MAC 在知道目的 IP 的情况下会去查 ARP 缓存自动填充。这些贴心行为在正常场景下省事在做精确控制的时候就得留个心眼。想要更贴近真实流量的随机性scapy 提供了一批 Rand 开头的字段生成器。RandIP()生成随机 IPRandShort()生成随机 16 位整数RandMAC()生成随机 MAC。注意它们不是直接赋值而是作为字段的默认值传入比如IP(srcRandIP())。这种写法在同一个包被多次组装时会每次生成新值这一点和直接写死一个随机数不一样批量发送的时候要留意。字段值还有一种延迟计算的用法把字段设成Nonescapy 会在组装时自动推算。最典型的是 IP 头的长度字段和校验和字段你写IP(lenNone, chksumNone)它会在发送前根据实际的载荷长度和内容算出来。这个特性在你自己拼接不定长载荷时特别有用不用手动去数字节数。同理TCP 的校验和涉及伪首部计算手算容易错设成 None 交给库处理最稳妥。注意手动指定了校验和但又改了载荷内容会导致包被对端丢弃。如果你的脚本先设置chksum再追加数据务必在最后把chksum重置为 None 让它重算。3. 发出去与收回来send/sr 系列 API 的选择3.1 几个发送函数的区别与使用场景scapy 的发送函数名字很像功能差别不小用错了轻则收不到回应重则报文根本发不出去。我的经验是可以按发哪一层和要不要答案两个维度来分。发三层IP 层及以上用send()它只发不收适合做单向的探测或者批量灌包。发二层以太网帧用sendp()需要你显式指定网卡因为二层帧没有路由信息库不知道该从哪个网卡出去。这两个是只管发的版本对应的发并等答案版本是sr()和srp()返回两个列表收到了回应的和没收到回应的。只要第一个回应的版本是sr1()它返回单个包对象或者 None写交互式脚本时最方便。选择哪种关键看你关心什么。如果你只是想让对端收到一个报文比如测某个端口开不开、触发一次特定的协议交互用send就够。如果你需要拿到回包做后续分析用sr1最简洁它内部会帮你匹配请求和响应。如果是批量探测比如对一段地址逐个发用sr能拿到完整的成功失败列表方便统计。函数工作层次是否收包典型场景send三层否单向灌包、触发动作sendp二层否构造特定以太网帧sr三层是返回全部批量探测、统计成功率sr1三层是返回第一个单次交互、拿回包解析srp二层是ARP 类二层交互有个细节值得说sr和sr1默认会把回包和请求做匹配靠的是 IP ID、端口这类字段的对应关系。如果你的请求包字段都是默认值批量发的时候可能出现匹配错乱拿到别人的回包。解决办法是给每个包一个不一样的标识比如递增的 IP ID 或者源端口这样匹配就准了。3.2 超时、重试与结果对象的解析超时设置是个必须认真对待的参数。默认超时比较短跨网段或者对端响应慢的时候很容易假失败。我通常按目标距离来设同局域网 1 到 2 秒足够跨路由 3 到 5 秒跨运营商或者对方响应本来就慢的给到 8 秒以上。超时设太长的副作用是脚本整体耗时飙升尤其是批量扫的时候几百个地址每个等 8 秒跑一次要小半天。retry参数控制重发次数默认是 1 次也就是不重发。网络有轻微丢包的环境下设成 2 到 3 次能明显提高成功率代价是失败的目标耗时翻倍。还有个inter参数控制发包间隔默认是 0 也就是连着发。这个参数在探测带速率限制的设备时特别重要间隔设成 0.05 到 0.1 秒能避开大部分设备的限速策略成功率会好看很多。结果对象的解析是新手容易迷糊的地方。sr返回的是一个二元组第一个是发了且有回应的列表每个元素是(发出的包, 收到的包)这样的配对第二个是发了没回应的列表。想看汇总直接调.summary()就能打印成可读格式。想遍历就for sent, recv in ans:这样拆包。我建议先把ans的长度打出来看一眼再看几条具体内容心里有个数再往下写逻辑。from scapy.all import IP, ICMP, sr ans, unans sr( IP(dst192.168.1.0/28) / ICMP(), timeout2, retry2, inter0.05, verboseFalse, ) print(f有回应 {len(ans)} 个无回应 {len(unans)} 个) for sent, recv in ans: print(sent[IP].dst, -, recv[IP].src, recv[ICMP].type)4. 嗅探与落地sniff 和 pcap 读写4.1 sniff 的参数组合与回调写法sniff是日常用得最多的函数。它的核心参数就那么几个iface指定网卡count指定抓多少个包就停prn是每抓到一个包就调一次的回调函数filter是 BPF 过滤表达式timeout是抓多久就停store决定要不要把包留在内存里。prn的写法直接决定脚本的可读性。最简单的是传个 lambdaprnlambda p: p.summary()一行搞定打印。稍微复杂点就定义个具名函数在里面做解析、判断、落盘。我推荐后者尤其是逻辑超过三行的时候lambda 里塞太多东西调试起来费劲出错了堆栈指向也不清晰。storeFalse这个参数值得单独提。默认情况下 sniff 会把抓到的每个包都存在一个列表里抓几千个包内存还好抓几十万个就明显吃内存了。如果你的处理逻辑是来一个处理一个比如统计、落盘、实时告警把 store 设成 False包处理完就释放内存占用能压到很低。我之前写一个长时间运行的监控脚本忘了设这个参数跑了一晚上内存吃了两个多 G第二天才反应过来。filter用的是 BPF 语法它由底层的抓包库解析效率比在 Python 侧用if判断高得多因为过滤发生在内核态不匹配的包根本不会送到用户态。语法上常见的写法有tcp port 80、udp and host 10.0.0.5、icmp、not arp。注意如果系统里没装 libpcap这个参数会报错提示 filter 不可用这时候要么补装依赖要么退回用 Python 侧判断。from scapy.all import sniff, wrpcap buffer [] def on_packet(pkt): if pkt.haslayer(IP): buffer.append(pkt) print(pkt[IP].src, -, pkt[IP].dst) sniff( ifaceeth0, filtertcp port 443, prnon_packet, count100, timeout30, storeFalse, ) if buffer: wrpcap(collected.pcap, buffer)4.2 离线 pcap 的读写与回放scapy 对 pcap 文件的支持做得挺完整。rdpcap()一次性把文件读成包列表wrpcap()把列表写回文件PcapWriter适合边抓边写、持续追加的场景。读出来的东西就是普通的 Packet 对象列表可以像操作内存里的包一样做切片、过滤、字段修改。这里有个实用技巧读大文件的时候rdpcap会把全部内容载入内存几百兆的文件就可能把机器拖慢。scapy 提供了PcapReader它是迭代器一次只读一个包配合生成器写法可以做到流式处理。处理逻辑如果只是统计、筛选、转换格式用这个更合适。回放是个有意思的用法。你把 pcap 读出来改掉源 IP、目的 IP再用sendp发出去相当于重现一段流量。这在协议测试、设备联调时很有用。但要注意两点一是回放会改变原始的时间间隔sendp是尽最大速度发的想要保真得自己按时间戳加延时二是回放的包如果带校验和改了字段之后要把它重置为 None 让库重算否则对端会直接丢掉。还有一种场景是边抓边分析。我用过一个思路sniff 的prn里不做重活只把包塞进一个队列另起一个线程从队列里取出来做解析和统计。这样抓包线程不会因为解析慢而丢包。这个模式在高速流量下几乎是必须的单线程边抓边做复杂解析丢包率会很高。5. 两个真实场景的实操记录5.1 局域网地址探测与响应分析我经常需要在自建的小型实验网络里确认某些地址是否有设备在线。用常规的 ping 命令一个个敲太慢写个 scapy 脚本批量发 ICMP 更省事。思路很简单构造一批IP(dst...)/ICMP()用sr发出去看哪些有回。from scapy.all import IP, ICMP, sr, conf targets [f192.168.1.{i} for i in range(1, 33)] ans, unans sr( IP(dsttargets) / ICMP(), timeout2, retry1, inter0.03, verboseFalse, ) online sorted({recv[IP].src for _, recv in ans}) print(在线地址, online)这段代码在实际跑的时候有几个点要注意。第一ICMP 回应的类型字段要看一下正常回应是 type 0但也可能收到 type 3目标不可达之类的那说明中间有设备在回你不代表目标主机在线。第二有些设备会屏蔽 ICMP 但端口是开的所以光靠 ICMP 判断在线会有漏网想更全面就配合 TCP SYN 探测。第三批量发的时候注意间隔太密集可能触发某些设备的防护策略反而什么都收不到。ARP 层也可以做类似的事情而且比 ICMP 更可靠因为 ARP 是二层协议不受三层防火墙影响。发一个广播的 ARP 请求谁在线谁就会回应。这个操作只应该在自己管理或明确授权的网络里做别在别人的网络环境里随便扫。5.2 自定义协议字段的抓包解析scapy 另一个让我觉得好用的地方是能定义自己的协议层。假设我在做一个自研的通信模块报文格式是固定的1 字节类型、2 字节长度、4 字节序号、然后是不定长的载荷。用 scapy 定义出来之后抓包时就能像看标准协议一样一层层展开。from scapy.all import Packet, ByteField, ShortField, IntField, StrLenField from scapy.all import UDP, bind_layers class MyProto(Packet): name MyProto fields_desc [ ByteField(msg_type, 1), ShortField(length, None), IntField(seq, 0), StrLenField(payload, b, length_fromlambda p: p.length), ] bind_layers(UDP, MyProto, dport9999)定义好之后抓到的包只要 UDP 目的端口是 9999就会自动按这个结构解析。length_from这个参数是关键它让载荷长度跟随前面定义的 length 字段走实现变长解析。如果没有它载荷会被当成定长多出来的字节就丢了。bind_layers的绑定是有条件的可以指定端口、类型字段等。多个绑定同时命中时后绑定的优先级更高这个顺序在调试自定义协议时容易忽略。我遇到过一次解析结果不对查了半天发现是另一个端口相同的协议绑定把解析抢走了把绑定条件写得更具体就解决了。5.3 时间戳与性能数据的一次实测链路层的响应时间是个值得关注的指标。scapy 抓到的每个包都带时间戳pkt.time是个十进制浮点数单位秒。想算往返时延就拿回包时间减去发包时间。这里要注意时钟精度有些平台的抓包时间戳精度只到微秒跨机器比对会有偏差同机比对才比较准。我做过一次小规模实测在同一台机器上对 32 个内网地址发 ICMP间隔 0.03 秒超时 2 秒。统计下来局域网内正常响应的时延集中在 1 到 3 毫秒少数设备在 8 到 15 毫秒还有两个地址完全无响应。整个脚本跑完不到三秒。对比用系统 ping 命令逐个执行同样的地址集合要二十多秒。这个差距主要来自进程启动开销和串行等待scapy 在脚本里批量构造、异步收发效率高得多。不过也别迷信这个数字。scapy 是纯 Python 实现构造大量复杂包的时候Python 层的开销会显现出来。我试过构造一万个带自定义载荷的包光拼接就花了两三秒。如果对吞吐有硬要求可以考虑把构造和发送分开用多进程并行或者干脆换更底层的库。多数排障和验证场景下scapy 的速度完全够用。6. 常见问题与排查技巧实录6.1 权限、网卡、过滤器三座大山新手遇到的问题里九成集中在三件事上没权限、网卡选错、过滤器写错。这三个问题的报错信息往往很含糊容易让人往错误的方向查。权限问题的典型表现是PermissionError或者Operation not permitted。在 Linux 上要么加 sudo要么给解释器打能力标记。Windows 上则是 Npcap 没装好或者装的版本不匹配表现可能是能构造包但发不出去或者发出去没反应。网卡选错的表现更隐蔽脚本不报错但抓不到任何包或者只能抓到本机的流量。这是因为默认网卡可能是个虚拟网卡或者回环网卡。排查办法是先打印conf.iface看看当前默认是哪块再show_interfaces()列出全部手动指定对的。名字里带 docker、vmnet、vbox 的一般是虚拟网卡别选错。过滤器写错会直接抛异常提示语法错误。BPF 的语法和常见编程语言的逻辑表达式不太一样and、or、not是小写字段名要符合协议规范比如tcp[tcpflags] tcp-syn ! 0这种位运算写法。建议先用最简单的tcp或icmp验证能用再逐步加条件。注意在容器环境里抓包默认只能看到容器自己网卡上的流量。想抓宿主机或者别的容器的流量需要把网卡共享出来这是环境配置问题不是代码问题。6.2 报错与现象速查表下面这张表是我这些年攒下来的遇到问题先对一遍能省不少时间。现象可能原因处理方式导入报错找不到 conf装的是精简版或装错包确认pip show scapy有输出重装发送无报错但收不到回应目标屏蔽、防火墙、超时太短加长 timeout换目标验证sniff 一个包都抓不到网卡选错、过滤器太严去掉 filter先不指定 iface抓到的包解析成 Raw协议层没识别检查端口用 bind_layers 绑定自定义层校验和错误导致丢包改了载荷没重算把 chksum 设为 None 重新组装内存持续上涨sniff 没设 storeFalse关掉存储改成边处理边释放批量发送匹配错乱请求标识重复给每个包唯一的 IP ID 或源端口时间戳全为 0驱动不支持高精度时间戳换物理网卡或忽略该字段除了表里这些还有一个不算问题但常让人困惑的现象show()显示的长度和实际发出去的不一致。前面提过show()看的是内存里的值show2()才会触发组装重算。遇到长度、校验和相关的疑惑一律用show2()看。6.3 几个我踩过的坑和应对思路第一个坑是网卡名称。Linux 下的网卡名可能是eth0、ens33、enp0s3各种形式硬编码在脚本里换个机器就废了。我的做法是不指定 iface让它走默认需要明确指定的时候从conf.iface读出来作为默认值或者做成命令行参数。这样脚本的可移植性好很多。第二个坑是发包过快导致本机网卡队列溢出。批量发的时候如果inter设成 0包的构造速度超过网卡发送速度本机会直接丢掉一部分你以为是网络丢包其实是自己发太猛。加个几十毫秒的间隔问题就没了。这个现象在虚拟机和配置较低的机器上尤其明显。第三个坑是混淆了三层和二层发送。用send发一个只有 IP 层的包库会自己查路由、找网卡、补以太网头用sendp发同样的包因为没有以太网层会直接报错。反过来用send发一个带了完整以太网头的包那个以太网头会被当成载荷塞进 IP 包里面发出去的东西完全不是你想的样子。判断标准很简单包里有没有Ether()这一层有就用sendp没有就用send。第四个坑是脚本跑在后台时的输出。scapy 的verbose默认是开的会往终端打一堆进度信息。放在交互式终端里无所谓做成定时任务或者后台服务的时候这些输出可能把日志撑爆或者因为终端不可写而抛异常。养成习惯正式脚本里把verboseFalse加上。最后一个关于结果的可靠性。sr1返回 None 的时候可能是真的没回应也可能是回应来晚了、被中间设备拦了、或者匹配逻辑没对上。别一看到 None 就下结论说目标不通多试几次换个探测方式交叉验证一下。网络这东西单次结果的说服力很有限多跑几轮看趋势才靠谱。