用 Python 构建轻量级安全巡检工具：端口扫描、ARP 监测与流量分析

1. 从一个真实需求谈起

我没有系统学过网络安全，也不认为自己是“黑客”或者“红队研究员”。但这些年做后端开发和运维自动化，Python 一直是我最顺手的工具。有一次某公司的内部系统上线前做安全自查，安全团队给了一份漏洞清单，我随手看了一下，发现很多条目其实是重复劳动：比如定时扫端口、检查某些服务是否暴露到外网、分析某个时间段的异常流量。他们用商业扫描器能做，但价格不便宜，而且不好定制，遇到一些内部特有的端口规范只能手动补。

那时候我产生了一个想法：与其堆命令和脱机的表格，不如用 Python 把这些巡检动作做成一个小工具集。于是就有了一个大概 1200 行的轻量级检测脚本集合，包含端口自检、内网网段扫描、ARP 监测、流量统计几个模块。我把它起了个名字，叫“轻量级安全巡检助手”。这个项目让我对 Python 在网络安全领域的定位有了非常具体的认知：它不是万能钥匙，但绝对是效率放大器。

这篇文章我想把这套工具集的选型思路、核心实现和踩坑记录完整写出来。不是那种“知识点罗列”的教学文，而是更像你在公司内部看到一份项目总结。如果你是运维、后端开发，或者刚开始接触网络安全方向，想用 Python 做点实际的安全自动化工作，这里面的思路和代码可以直接拿去改。

2. 整体设计思路与关键技术选型

2.1 为什么选择 Python 而不是直接上商业工具

先明确一个前提：真正的大型企业安全建设，不可能只靠脚本去顶，商业扫描器、EDR、NTA 这类平台必须上。但很多中小型团队的情况是——有明确的安全需求，却不一定有预算和人力去维护一套复杂系统。这时候 Python 的价值体现在三个方面。

第一是胶水能力。安全排查通常要串联多种数据源，比如端口状态、网络连接、系统日志、ARP 表。Python 可以借用系统命令的解析能力，也能直接调用底层库做数据包解析，中间的粘合成本很低。第二是快速定制。商业工具的参数和报告模板固定，但每个公司的内网规范不同，比如某些端口段业务自己约定不允许外放，这些规则在通用工具里很难表达，在你自己的脚本里就是一个配置文件的事。第三是迭代速度。我做一个新检测逻辑，改完代码直接跑，不需要等厂商更新规则库。

2.2 核心框架选型：Scapy 与 psutil 的组合

我调研过几个 Python 的网络安全相关库，最终保留了 Scapy 和 psutil 作为主力，辅助用了一部分 dpkt 和 socket 标准库。

Scapy 是 Python 生态里最著名的网络包操作库，可以构造、发送、解析网络数据包。我最早用它是做 TCP 三次握手细节的实验，后来发现它在网络安全里真正的优势不是“发包”——虽然这也是它的强项——而是把抓包和解析做成了一套高度抽象的命令。你不需要处理 pcap 文件的二进制格式，直接写sniff(iface='eth0', prn=callback, count=100)，就能拿到解析好的数据帧，比手写套接字省太多事。

psutil 是系统信息采集库，进程、CPU、内存、网络连接都能查。它在做安全巡检时的作用是提供“主机侧”的视野。比如我想知道某个端口对应的进程是谁，一条psutil.net_connections()就能列出本地端口与进程 PID 的关系，再通过psutil.Process(pid).name()拿到进程名，一套链路就通了。

dpkt 则是更底层的 pcap 解析库，体积小，运行快。我在做离线流量分析时用 dpkt 比较多，因为可以直接读取 tcpdump 生成的 pcap 文件，按自己的逻辑统计最关心的字段，不依赖图形界面。

2.3 整体功能规划：一个可扩展的巡检集合

我没有把它做成一个大一统的“平台”，而是做一个模块集合。每个模块独立成文件，主入口只负责分发参数。这样做的好处是：单模块崩溃不影响整体，而且后面新增安全检查只需要加新文件。

整个项目最初规划了五个模块：

每个模块各有侧重，但又都围绕同一个核心：用自动化的方式替代人工在命令行里反复敲命令。比如端口扫描，很多人第一反应就是装个 Nmap。但如果你要扫的只是某几个固定端口，并且希望结果直接汇总到统一格式的 JSON 里，自己用 socket 写一个同步扫描可能要不了 100 行。这不是重复造轮子，而是按需定制。

3. 核心实现与关键代码拆解

3.1 端口扫描模块：从同步到并发

端口扫描是网络巡检最基础的部分。它的技术本质是尝试与目标主机的指定端口建立 TCP 连接，根据连接是否成功判断端口是否开放。

最朴素的写法是用 socket 的connect_ex方法逐个连接：

import socket def check_port(host, port, timeout=3): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(timeout) result = sock.connect_ex((host, port)) sock.close() return result == 0

这里有个关键点：connect_ex返回 0 表示连接成功，其它值表示失败，比如 10061（Windows 上）表示主动拒绝，11001 表示主机不可达。直接用这个返回值能省掉异常处理。

但这种写法在扫描多个端口时会非常慢，因为网络请求是串行。我用concurrent.futures.ThreadPoolExecutor做了并发改造，同时开 50 个线程。你可能会问：为什么不直接用协程？因为 TCP 连接涉及系统调用和内核协议栈，线程在这种 I/O 密集场景下反而更直观，Python 的 GIL 不会对系统调用部分造成明显影响。

from concurrent.futures import ThreadPoolExecutor, as_completed def scan_range(host, ports, timeout=3, max_workers=50): results = {} with ThreadPoolExecutor(max_workers=max_workers) as executor: future_map = { executor.submit(check_port, host, port, timeout): port for port in ports } for future in as_completed(future_map): port = future_map[future] try: is_open = future.result() except Exception as exc: is_open = False print(f"[WARN] 检测端口 {port} 时发生异常: {exc}") if is_open: results[port] = "open" return results

实际使用中我还发现了两个值得注意的地方。

第一，超时时间不能一律用 3 秒。如果扫描目标是内网机器，1 秒足够；如果扫的是跨地域的公网地址，3 秒可能不够。我建议采用动态超时：先 ping 一次判断 RTT，再根据 RTT 设置 1.5 倍或 2 倍超时，能明显提升扫描效率。

第二，线程数不是越大越好。当线程开到 200 以上时，本机可能会因为文件描述符耗尽而出现 Socket 创建失败。我实测 50 到 100 之间效率提升最明显，再往上边际收益很小，反而增加误报概率。

3.2 基于 Scapy 的 ARP 异常检测

ARP 监测这个模块我一开始是拒绝的，因为它看起来太“上古”了。但在实际排障中，我发现 ARP 层面的异常往往能暴露早期横向移动的迹象。当一个主机在短时间内发出大量针对不同 IP 的 ARP 请求，也就是所谓的 ARP 扫描，通常意味着有人在探测内网拓扑。

用 Scapy 实现 ARP 嗅探最方便的写法是这样：

from scapy.all import sniff, ARP def arp_callback(pkt): if pkt.haslayer(ARP): op = pkt[ARP].op # 1 表示请求，2 表示应答 src_ip = pkt[ARP].psrc dst_ip = pkt[ARP].pdst src_mac = pkt[ARP].hwsrc if op == 1: # 记录请求来源与目标地址 print(f"[ARP-REQ] {src_mac} ({src_ip}) 询问 {dst_ip}") def start_sniffer(iface, count=200): sniff(iface=iface, prn=arp_callback, count=count, store=False)

这里有个非常容易踩的坑：在arp_callback里做字符串拼接和打印没问题，但如果你要统计高频来源，千万不要直接用 Python 字典在回调函数里更新计数而不考虑并发——Scapy 的 sniff 虽然是顺序回调，但如果你结合多线程跑多个 sniff 实例，对同一个字典操作就存在竞态条件。我后来改成使用collections.Counter加锁，或者干脆不用多线程，按网卡顺序轮流嗅探，逻辑简单很多也不容易出错。

另一个问题是误报率。一些网络设备（比如路由器和交换机）本身就会定期发送 ARP 请求做邻居探测，这些行为从数据包层面看和“扫描”没有本质区别。所以我会加一个聚类逻辑：某个来源 MAC 在 10 秒内请求了超过 20 个不同的 IP，才判定为异常。

后来我把这个逻辑做成一个阈值配置文件：

{ "time_window_seconds": 10, "min_arp_requests": 20, "watch_ips": ["192.168.1.1", "192.168.1.2"] }

在抓包时只记录目标 IP 属于 watch_ips 的请求，既减少数据量，又能针对性防护重要资产。

3.3 离线流量统计：用 dpkt 分析 pcap

大多数情况下，我们需要分析的流量不是实时抓的，而是 tcpdump 或者 Wireshark 导出的 pcap 文件。dpkt 库能把 pcap 文件解析成以太网帧，然后逐层剥出 IP、TCP、UDP 头。

我最常用的分析维度有三个：Top N 目的 IP、协议分布、可疑的高频 DNS 请求。

import dpkt import collections def analyze_pcap(pcap_file): ip_counter = collections.Counter() proto_counter = collections.Counter() dns_counter = collections.Counter() with open(pcap_file, 'rb') as f: pcap = dpkt.pcap.Reader(f) for timestamp, buf in pcap: eth = dpkt.ethernet.Ethernet(buf) if isinstance(eth.data, dpkt.ip.IP): ip = eth.data ip_counter[ip.dst] += 1 proto_counter[ip.p] += 1 if ip.p == dpkt.ip.IP_PROTO_UDP: udp = ip.data if hasattr(udp, 'dport') and udp.dport == 53: try: dns = dpkt.dns.DNS(udp.data) if dns.qr == 0: # 查询请求 if dns.qd: dns_counter[dns.qd[0].name] += 1 except Exception: pass return ip_counter, proto_counter, dns_counter

这个模块在跑大数据量文件时性能很关键。我拿一个 2GB 的 pcap 做过测试，纯 Python 循环大概需要 12 分钟，后来把协议判断和 DNS 解析部分改用多进程分片处理，时间压缩到了 5 分钟以内。分片方法就是按 pcap 文件的大小切块，每个进程独立解析一块，最后合并 Counter。需要留意的是，dpkt 解析某些异常包时会抛异常，所以每个包的处理都要包在 try-except 里，否则一个坏包会中断整个分析任务。

3.4 主机侧进程与外连检测

这个模块实现起来其实很轻量，因为它只是把 psutil 的能力组合起来。逻辑很简单：列出所有进程的网络连接，提取出“非本机回环地址”的目标地址，然后和配置好的白名单比对。

import psutil def find_suspicious_connections(whitelist_ips=None, whitelist_pids=None): suspicious = [] for conn in psutil.net_connections(kind='inet'): if conn.status != psutil.CONN_ESTABLISHED: continue laddr = conn.laddr raddr = conn.raddr if not raddr or raddr.ip == '127.0.0.1': continue pid = conn.pid if pid is None: continue if whitelist_pids and pid in whitelist_pids: continue if whitelist_ips and raddr.ip in whitelist_ips: continue try: proc = psutil.Process(pid) proc_name = proc.name() except psutil.NoSuchProcess: proc_name = f"[已退出进程 pid={pid}]" suspicious.append({ "pid": pid, "process": proc_name, "local_ip": laddr.ip, "local_port": laddr.port, "remote_ip": raddr.ip, "remote_port": raddr.port }) return suspicious

我发现这个模块的真正价值不在于“找出恶意程序”，因为现代恶意样本不会大摇大摆地建立稳定连接等你来抓。它更实用的场景是发现被遗忘的服务——比如某台服务器上有一个三年前启动的测试进程，还在向外部云存储上传数据。这种连接平时不会有人注意，但用脚本拉出全部外连列表后，你会惊讶于内网竟然有这么多“隐形沟通”。

4. 实操过程中遇到的高频问题与排查方法

4.1 Scapy 在 Windows 和 Linux 上的行为差异

这个坑几乎每个用 Scapy 的人都会遇到。在 Windows 上运行 sniff 需要安装 Npcap 并选择“WinPcap 兼容模式”，否则会报找不到网卡的错误。而 Linux 上需要 root 权限，如果你用普通用户跑脚本，会在打开原始套接字时直接被拒绝。

我的建议是：开发调试的时候可以用高性能模式下的 CentOS 或者 Ubuntu 虚拟机，直接 sudo 跑；生产环境必须给脚本一个最小权限账号，不要无脑用 root 跑整个项目，避免因为一个子模块的漏洞导致主机沦陷。

4.2 高并发端口扫描触发本机防火墙限制

在扫描公网 IP 时，本机的安全软件或者云平台的安全组可能因为短时间大量连接而封禁你的出口 IP。这个情况发生时，扫描结果会出现大范围的“超时”，你很难区分是被扫描主机过滤了还是本机被封了。

排查方法是：在扫描脚本里增加一个小功能，记录连接失败的类型——超时和无响应是两回事。如果连续 N 个端口都是 timeout，大概率是本机被限流了。这时候你需要降低并发度，或者加一个随机延迟，把扫描速率控制在每秒 100 个连接以内。

4.3 pcap 文件时间戳精度问题

dpkt 的 pcap.Reader 返回的时间戳是 Unix 时间戳，但精度只到秒，而 tcpdump 默认保存的时间戳是有微秒精度的。如果你要分析某个毫秒级时间段内的流量细节，直接用 dpkt 默认值会丢失精度。

解决办法是打开 pcap 文件时用dpkt.pcap.Reader的timestamp_precision参数，或者在保存的时候用tcpdump -tttt微秒格式。这个问题在跨主机对比数据包时序时尤其关键，我最早没注意，结果分析出的“先后顺序”和 Wireshark 里看到的对不上。

4.4 ARP 监测的持续运行与内存泄漏

Scapy 的 sniff 函数如果长时间运行，而不设置 count 和 timeout，它会把所有抓到的包保存在内存里（即使 store=False 也可能在某些版本中有累积）。我遇到过内存增长率过高的问题，排查后发现是回调函数里引用了全局的 list 保存所有包摘要。后来我在回调里只保留统计结果，不保存原始包引用，并且设置每抓 5000 个包就主动重启一次 sniff 实例，把内存稳定在 50MB 左右。

下面是排障速查表，是我实际调项目时核对过的：

5. 从项目实战到一套可持续的安全巡检机制

这套工具集最终没有做成一个复杂平台，而是以命令行的方式跑在每周五凌晨两点，用 crontab 触发，结果推到内部的消息通道。我自己的真实体会是：网络安全项目最怕的不是技术难，而是“做完就完了”。如果只是写一次脚本出一次报告，那它的价值很有限；真正有价值的是把巡检变成一种固定的机制。

所以后来我往里加了几个“让机制转起来”的细节：

第一，报告必须包含增量对比。比如这次扫描发现的开放端口如果比上次多了三个，那这三个端口必须高亮显示。实现方式很简单，把上一次的结果存成 JSON，本次运行后做 set 差集。

第二，告警不能全量发给所有人，否则三天一过就没人看了。我设置了一个简单的分级规则：高危（比如 22 端口暴露到公网）只发给运维负责人和安全管理人；中危（比如内网出现 ARP 异常）发到值班群；低危只记录不告警。

第三，所有检测模块的阈值都集中到一个 YAML 文件里，改配置不用改代码。因为在实际运行中你会发现，阈值是需要根据网络环境不断调优的——刚上线时觉得 10 秒内 20 个 ARP 请求是异常，后来发现正常的视频会议系统就有这种行为，就得调成 50。

关于 Python 和网络安全的关系，很多文章喜欢从“攻击”的角度讲，写爬虫、写漏洞利用、写爆破脚本。但我更推荐把重心放在可观察性和自我巡检上。先知道资产有什么、端口是什么状态、哪些进程在通信，再谈怎么防护。一个稳定的网络巡检基线，比一个高超的利用技巧更有价值。如果你已经能用 Python 写业务代码，不妨拿这些模块练手，你会发现它跟写业务接口完全是两个维度——你思考的粒度会从“怎么处理用户请求”切换到“怎么解读每一层网络数据背后的含义”。这个视角的转换，往往就是理解网络安全的开始。