简介这是一份面向计算机相关专业毕业生与Python学习者的完整课设/毕设项目围绕网络入侵检测与防御场景提供可直接运行的代码和配套文档适合用于毕业设计、课程设计或期末大作业。资源采用zip压缩包形式共38个文件整体约97KB。主体为14个Python源文件与9个编译生成的pyc文件涵盖核心业务逻辑、路由配置与功能模块另有4个HTML页面、3个JavaScript脚本和1个CSS样式文件用于前端交互与界面展示同时包含dockerfile、docker-compose.yml、requirements.txt、install.sh等部署运维文件便于快速搭建运行环境README.md则对项目结构和启动方式做了说明。目前已有63人学习浏览属于轻量但结构完整的小型安全类项目。下载后既能直接运行查看效果也可参照源码与文档理解入侵检测流程、模块划分和部署思路适合需要快速完成课设或入门网络安全方向的读者。1. 用 Python 写网络入侵检测与防御系统先想清楚“检测”再谈“防御”用 Python 写一套网络入侵检测与防御系统作为毕业设计难点往往不是“能跑”而是“怎么证明它真能检测”。很多人拿到源码包第一件事就是python main.py看到界面弹出来就以为完成了一半结果答辩时老师问“你的系统检测到了什么特征基线怎么定的误报多少”直接卡住。这类系统本质上是一条流水线抓包、解析、特征提取、检测判定、告警与封禁。每个环节都有独立的坑而源码和文档的作用是给你一个能改、能测、能讲清楚的骨架而不是一个黑匣子。这篇笔记就按这条流水线拆开讲适合做毕设的学生也想自己搭轻量级 IDS 的运维新手。2. 系统拆解与 Python 选型理由一个毕设级 IDS 的模块边界在哪里2.1 抓包层为什么选 Scapy三种方案的取舍Python 做流量采集绕不开三个常用库Scapy、dpkt、以及基于 tshark 的 pyshark。Scapy 的特点是协议解析内置、交互式调试方便写pkt[IP]就能取到 IP 层字段对新手极其友好。dpkt 更快但协议解析基本靠手写适合对性能敏感、又愿意读 RFC 的老手。pyshark 解析能力最强但它依赖 Wireshark 的 tshark 二进制部署时要额外装一坨东西。毕设场景里流量规模通常是小网卡抓包或离线 pcap 回放Scapy 完全够用而且它在 pcap 文件读写、构造测试报文方面的生态也最完整。从源码结构来看抓包模块用 Scapy 也是最容易讲清楚的一层一个回调函数处理一个包逻辑独立方便后面做单元测试。方案性能依赖解析能力适合场景Scapy中等仅 libpcap内置常见协议毕设、中小流量、快速原型dpkt较高无需手写解析追求吞吐、熟悉协议格式pyshark低tshark 二进制最强需要 Wireshark 级解析结果2.2 检测引擎的两条路线规则匹配与行为基线抓包只是拿到原料真正决定系统价值的是检测引擎。常见做法是两条路线都做一是规则匹配像 Snort 签名那样在报文负载里找特征字符串或正则优点是解释性强命中即告警答辩时可以直接讲“这条规则为什么这么写”二是行为基线统计每个源 IP 在时间窗口内的连接数、包长均值、协议分布建立正常画像偏离到一定程度就告警优点是能检出“未知攻击”缺点是误报率高阈值全靠调。两条路线在系统里不冲突反而互补。规则负责已知攻击的精确识别基线负责可疑行为的兜底。毕设的创新点通常就落在基线统计上——比如传统规则引擎对慢速扫描基本无感但窗口内 SYN 包数量显著偏离均值就能暴露它。2.3 整体模块划分源码应该长成什么样一个可维护的毕设项目代码结构应该能一眼看出职责边界。我一般会分成五个模块capture/管抓包parse/管协议解析与字段抽取features/管特征向量生成detect/管规则与统计检测action/管告警入库和防火墙联动。另加一个web/目录做结果展示用 Flask 起一个简单的仪表盘。这个划分的价值在于答辩时你能指着一个文件说“这是特征提取”“这是检测阈值”而不是在一坨代码里翻半天。文档说明部分也按这个结构写每章对应一个模块评审老师顺着目录就能看懂你的设计思路。3. 流量采集与特征提取从网卡到结构化数据的完整链路3.1 用 Scapy 实时抓包主循环怎么写才不丢包先把最小的实时抓包跑通。下面这段代码解决两个问题一是回调函数必须短二是抓包线程要独立不能阻塞主程序。很多人直接把抓包写在主线程里界面一卡包就开始丢。from scapy.all import sniff import queue import threading # 用有界队列隔离抓包与处理避免回调里做重活 packet_queue queue.Queue(maxsize1000) def packet_callback(pkt): # 回调里只入队不做协议解析保证sniff主循环足够快 packet_queue.put(pkt) def start_capture(ifaceeth0): # daemon线程让抓包在后台运行主程序可以做别的事 sniffer threading.Thread( targetsniff, kwargs{ iface: iface, prn: packet_callback, store: False, # 不把原始包保存在内存中 }, daemonTrue, ) sniffer.start() return sniffer # 使用示例启动后每秒看一次队列积压量 start_capture(eth0)这段代码有几个参数值得较真。storeFalse是必开的否则 Scapy 会把所有经过的包缓存在内存里跑一晚上就能吃掉几个 GB这是最常见的翻车点。prn指定回调函数队列的maxsize1000起到背压作用如果消费者处理不过来队列满了抓包线程会阻塞间接保护了整个系统的内存稳定性。网卡名建议先执行ip link确认eth0在云服务器上经常不叫这个名字。3.2 特征提取从一个包到一个字典拿到的 raw packet 要转成结构化特征这一步是为了让检测引擎不依赖 Scapy 的对象模型。每个包提取五元组、协议号、包长、TCP 标志位和时间戳后续无论是规则匹配还是统计基线都用同一份特征香。from scapy.all import IP, TCP, UDP def extract_features(pkt): # 非IP包直接跳过抓包环境里常有ARP、LLDP等无意义报文 if not pkt.haslayer(IP): return None ip pkt[IP] feat { ts: pkt.time, src: ip.src, dst: ip.dst, proto: ip.proto, pkt_len: len(pkt), } if pkt.haslayer(TCP): tcp pkt[TCP] feat[sport] tcp.sport feat[dport] tcp.dport feat[flags] int(tcp.flags) # 转成整型便于后续位运算匹配 elif pkt.haslayer(UDP): udp pkt[UDP] feat[sport] udp.sport feat[dport] udp.dport feat[flags] 0 # 有负载时截前64字节作负载样本既能识别特征又控制体积 if pkt.haslayer(Raw): feat[payload_head] bytes(pkt[Raw].load)[:64] else: feat[payload_head] b return feat参数说明pkt.time是抓包时刻的 Unix 时间戳离线回放时要注意它可能和当前时间不一致pkt_len用的是len(pkt)而不是ip.len前者是链路上实际收到的字节数后者是 IP 头里的声明长度两者在分片和填充场景下会不同。负载只截前 64 字节主要为了控制入库体积和内存占用规则匹配不需要完整负载。提取失败的包直接返回None调用方过滤即可。3.3 结构化存储为什么用 SQLite 而不是 MySQL特征存哪里直接影响系统的可演示性。选 SQLite 的理由很实际单文件数据库答辩时拷走就带走不用单独装服务并发写简单。不要用check_same_threadFalse去跨线程共享同一个连接这是 SQLite 多线程写入报错的重灾区。正确做法是消费线程自己独享连接。import sqlite3 import time def writer_loop(queue_ref, db_pathflows.db): # 每个线程单独创建连接不跨线程复用 conn sqlite3.connect(db_path) conn.execute(PRAGMA journal_modeWAL) conn.execute( CREATE TABLE IF NOT EXISTS flows ( ts REAL, src TEXT, dst TEXT, proto INTEGER, sport INTEGER, dport INTEGER, flags INTEGER, pkt_len INTEGER, payload_head BLOB ) ) while True: try: pkt queue_ref.get(timeout1) except queue.Empty: continue feat extract_features(pkt) if feat is None: continue # 预编译语句 批量写入比逐条execute快一个量级 conn.execute( INSERT INTO flows VALUES (?,?,?,?,?,?,?,?,?), (feat[ts], feat[src], feat[dst], feat[proto], feat.get(sport), feat.get(dport), feat.get(flags), feat[pkt_len], feat[payload_head]), ) conn.commit()这里PRAGMA journal_modeWAL是关键参数。WAL 模式下读写不互斥能明显降低“database is locked”的出现频率但要注意它只是缓解不是根治真正根治是保证只有这一个线程在写。timeout1配合queue.Empty让消费线程空转时不空耗 CPU。生产环境里 commit 的频率可以考虑攒批再提交比如每 100 条一次毕设场景直接每条提交也够用。4. 检测引擎的三个实现层次规则、统计与轻量学习4.1 从 Snort 规则到 Python 匹配器最少代码做精确识别规则匹配器是检测引擎的“脸面”因为它最好讲、最好调试。规则文件我通常自己定义一种极简格式兼容常见 Snort 规则的关键语义协议、方向、目的端口、告警消息、特征字符串。读取规则后在内存里编译成正则每次特征进来逐条匹配。import re import os RULE_FILE rules.txt # 规则格式: alert tcp any any - any 80 msg:敏感路径探测 content:/etc/passwd def load_rules(pathRULE_FILE): rules [] if not os.path.exists(path): return rules with open(path, r, encodingutf-8) as f: for line in f: line line.strip() if not line or line.startswith(#): continue parts line.split() # parts: [alert, tcp, any, any, -, any, 80, msg:..., content:...] rule { proto: parts[1], dport: int(parts[6]) if parts[6].isdigit() else None, msg: re.search(rmsg:(.*?), line).group(1), content: re.search(rcontent:(.*?), line).group(1).encode(), } rule[content_re] re.compile(re.escape(rule[content])) rules.append(rule) return rules def match_rules(feat, rules): for rule in rules: if rule[proto] tcp and feat.get(proto) ! 6: continue if rule[proto] udp and feat.get(proto) ! 17: continue if rule[dport] and feat.get(dport) ! rule[dport]: continue # 在负载头里找特征字符串找不到就继续下一条规则 if rule[content_re].search(feat.get(payload_head, b)): return rule[msg], rule[content] return None, None几个关键点re.escape保证规则里的特殊字符不会被当成正则语法解释这是新手最容易踩的坑——写一条content:/etc/passwd结果因为.匹配任意字符而疯狂误报。规则匹配是纯字符串查找故意不查全负载用第 3 章截出来的前 64 字节一方面匹配速度极快另一方面对分片攻击也有一定敏感性。如果你的毕设选题偏向 Web 攻击检测可以在content之外再加一个uri字段匹配 HTTP 请求行那通常解析不出来需要单独拆 HTTP 层。4.2 统计基线用滑动窗口识别“不正常”规则匹配对未知攻击无能为力统计基线就是用来补这块的。思想很简单每个源 IP 在最近 N 秒内发了多少个包、多少种标志位这些数字在正常流量里是相对稳定的。一旦某个窗口内的包数超过历史均值的 3 个标准差就判定异常。用deque实现滑动窗口最省内存。from collections import defaultdict, deque import statistics WINDOW_SEC 60 STD_MULTIPLIER 3.0 # 结构: ip - deque([(ts, 1), ...])只用ts和时间戳算窗内总数 ip_packet_times defaultdict(deque) def update_baseline(feat): src feat[src] q ip_packet_times[src] q.append((feat[ts], 1)) # 弹出窗口外的旧数据 cutoff feat[ts] - WINDOW_SEC while q and q[0][0] cutoff: q.popleft() if len(q) 10: return None, 样本太少继续攒数据 # 每秒包数量近似 窗口内包数 / 窗口秒数 sample [t for t, _ in q] mean statistics.mean(sample) stdev statistics.stdev(sample) if len(sample) 1 else 0.0 if stdev 0: return None, 流量过于稳定无法计算波动 # 最近10秒的包数比历史均值偏移超过阈值才告警 recent_10s sum(1 for t, _ in q if t feat[ts] - 10) if recent_10s mean STD_MULTIPLIER * stdev: return src, f最近10秒包数 {recent_10s}均值 {mean:.1f}标准差 {stdev:.1f} return None, 正常参数要说明白WINDOW_SEC60决定基线的记忆长度60 秒适合检测扫描型攻击太长了会稀释突发现象太短了基线本身就不稳定。STD_MULTIPLIER3.0是 3σ 原则理论上正态分布下只有 0.3% 的正常点会被误判但网络流量不是正态分布实际使用里 2.5 到 4 之间都要测。len(q) 10这个条件是防止冷启动阶段误报——新 IP 前几个包就触发告警没有任何意义这是绝大多数统计检测入门代码里没有的细节。4.3 轻量机器学习孤立森林怎么融进现有流程如果你想让毕设多一些“智能”成分可以加一个孤立森林做无监督异常检测但不要搞深度学习。无监督的意义是不需要标注好的攻击数据只要拿一段正常流量的特征训练然后对实时特征打分即可。对毕设来说可解释性和训练成本远比 AUC 高 0.01 重要。import joblib from sklearn.ensemble import IsolationForest def train_model(feature_vectors_pathnormal_features.npy): # 读入正常流量的特征矩阵每行是 [src_entropy, pkt_len_mean, ports_per_window] import numpy as np X np.load(feature_vectors_path) model IsolationForest( contamination0.1, # 假定的异常比例 n_estimators100, # 树的数量100够用调大只会增加耗时 random_state42 ) model.fit(X) joblib.dump(model, isf_model.pkl) return model def predict_anomaly(feat, model): import numpy as np # 特征格式要和训练时完全一致 vec np.array([[feat[src_entropy], feat[pkt_len_mean], feat[ports_per_window]]]) # score_samples返回负值值越小越异常decision_function小于0判异常 score model.score_samples(vec)[0] return score -0.1, scorecontamination0.1是最容易让人困惑的参数它表示模型假设训练集里有 10% 的异常样本实际上你的“正常流量”里可能一点异常都没有这时设 1% 更合适。另一个血泪经验特征向量必须和训练时完全对得上第 3 章存 SQLite 时就要同步落一份 numpy 格式的特征文件别在训练完以后才想起来特征代码改过。训练完一定要 dump 成 pkl演示时直接加载现场训练 30 秒会非常尴尬。5. 从检测到防御联动 iptables 的落地细节与五条避坑记录5.1 封禁动作怎么做才算“防御系统”检测的结果如果不能触发动作那它只能叫“监控系统”不叫“防御系统”。最常见的动作是调用 iptables 封禁源 IP封禁时长要有限制否则误报一次就把正常用户关外面了。下面这段代码同时实现了插入封禁规则和定时解封。import subprocess import threading def block_ip(ip, block_seconds300): # 先查是否已封禁避免重复插入同样规则 check_cmd fiptables -C INPUT -s {ip} -j DROP if subprocess.run(check_cmd, shellTrue).returncode 0: return False apply_cmd fiptables -I INPUT -s {ip} -j DROP subprocess.run(apply_cmd, shellTrue, checkTrue) print(f[ACTION] 已封禁 {ip}{block_seconds} 秒后自动解封) # 定时器到点后解封使用独立线程避免阻塞检测主循环 threading.Timer(block_seconds, unblock_ip, args(ip,)).start() return True def unblock_ip(ip): cmd fiptables -D INPUT -s {ip} -j DROP subprocess.run(cmd, shellTrue, checkTrue) print(f[ACTION] 已解封 {ip})这段代码里我特意先-C查规则是否存在这是个容易被忽略的幂等问题连续触发告警时如果没有查重同一个 IP 会被插入几十条 DROP 规则。-I是插到链首-D按同样条件删除正好配对。更工业化的做法是用 ipset 维护封禁集合避免规则数膨胀但毕设里 iptables 直连已经足够。运行这段代码必须有 root 权限给程序加上 sudo 执行时的日志记录很有必要。5.2 五条避坑记录现象、原因、解决现象一sniff启动后完全没有输出程序也不报错。原因多半是没权限——Linux 下用 libpcap 抓包需要 root 或CAP_NET_RAW能力普通用户调用 Scapy 时包回调一次都不触发。也可能是网卡名写错云服务器上常见ens33、eth0混用。 解决先sudo -s切到 root 再跑网卡名用ip link确认后再填。抓包丢包问题也常被当成玄学实际上多数是权限和网卡名这两个低级错误。现象二程序跑几分钟后弹出sqlite3.OperationalError: database is locked。原因肯定有多个线程在同时写同一个数据库文件。即使你只在 writer 线程里写只要别的地方调了sqlite3.connect()也可能引起锁竞争最常见的就是 Flask 展示页面里顺手查了一下数据库。 解决确保所有读写入口收敛到一个连接或者展示层用只读连接打开modero。不要用check_same_threadFalse硬顶着那只是把崩溃推迟。现象三检测引擎告警了iptables 也执行了但测试流量还是能正常访问。原因封禁规则插到了 INPUT 链可流量是从 FORWARD 链过的比如你这套系统部署在路由/网桥模式图省事只写了 INPUT。也可能是目标机器走 IPv6而 iptables 只管 IPv4。 解决先确认拓扑——是旁路镜像还是串联网关串联网关就要同时操作 FORWARD 链IPv6 环境还要同步跑ip6tables -I INPUT -s ip -j DROP。现象四正常用户被误封一天内封禁列表里全是误报。原因阈值设得太激进。统计基线里STD_MULTIPLIER设成 2.0 以下加上窗口期样本不足就开始判断偶发的浏览器预取流量就会触发告警。规则匹配那边也更危险一条宽泛的content规则能匹配所有含该字符串的正常包。 解决规则匹配的告警不要直接封禁先进入一个“观察名单”连续 3 次命中才动作统计基线把阈值回调到 3.0 以上并保证窗口内至少有 30 条样本再参与判断。毕设论文里把这两层延迟机制写上反而是个加分项。现象五离线回放 pcap 测试时告警时间戳全部乱套展板上的时间线成了乱码。原因tcpreplay默认以最快速度发包pcap 里的时间戳被压缩到几秒内打完特征里的pkt.time全部挤在一起滑动窗口统计直接失真。 解决回放时用tcpreplay --pps200限制每秒发包数或者回放前用editcap重置时间戳。回放的体积也要控制一次灌进去几 GB 的 pcap内存和队里都会承受不了分片回放更稳。6. 验证效果与进阶方向用可控重放实验给毕设一份可信结论毕设答辩时最怕的是“功能演示完但拿不出数据支撑”。正确做法是设计一个可控实验先采集 30 分钟正常办公流量建立基线然后用 Scapy 构造若干条带特征字符串的测试报文注入网络模拟一次敏感路径探测再观察系统是否能在大约 10 秒内告警并触发封禁。每次实验记录四个指标真正例数、假正例数、假负例数、封禁延迟。表格形式写进论文足够交代方法论。这套系统的进阶方向有三条值得投入把单机规则引擎改成动态热加载运行时用inotify监听规则文件变化免重启生效把 SQLite 换成 TimescaleDB 之类的时序库让特征曲线能以秒级粒度回溯在统计基线之上做关联分析比如同一源 IP 在 5 分钟内同时触发“端口扫描”和“敏感路径探测”则告警优先级自动上调。这三条任选一条深度都足够支撑毕业论文的一章“进一步工作”。我的习惯是每次调参都存一份日志把当时的阈值、窗口大小、实验数据量、结果指标写进一个 JSON 文件。答辩时直接翻日志讲“我把标准差阈值从 2.0 调到 3.0误报率降了多少”比口头说“我调过很多次”有说服力得多。这套系统真正的价值不是替你挡住所有攻击而是给你一个能反复验证、能讲清楚每次决策依据的检测闭环。希望帮到你。本文还有配套的精品资源点击获取 基于Python的网络入侵检测与防御系统：从抓包到封禁的完整实现
简介这是一份面向计算机相关专业毕业生与Python学习者的完整课设/毕设项目围绕网络入侵检测与防御场景提供可直接运行的代码和配套文档适合用于毕业设计、课程设计或期末大作业。资源采用zip压缩包形式共38个文件整体约97KB。主体为14个Python源文件与9个编译生成的pyc文件涵盖核心业务逻辑、路由配置与功能模块另有4个HTML页面、3个JavaScript脚本和1个CSS样式文件用于前端交互与界面展示同时包含dockerfile、docker-compose.yml、requirements.txt、install.sh等部署运维文件便于快速搭建运行环境README.md则对项目结构和启动方式做了说明。目前已有63人学习浏览属于轻量但结构完整的小型安全类项目。下载后既能直接运行查看效果也可参照源码与文档理解入侵检测流程、模块划分和部署思路适合需要快速完成课设或入门网络安全方向的读者。1. 用 Python 写网络入侵检测与防御系统先想清楚“检测”再谈“防御”用 Python 写一套网络入侵检测与防御系统作为毕业设计难点往往不是“能跑”而是“怎么证明它真能检测”。很多人拿到源码包第一件事就是python main.py看到界面弹出来就以为完成了一半结果答辩时老师问“你的系统检测到了什么特征基线怎么定的误报多少”直接卡住。这类系统本质上是一条流水线抓包、解析、特征提取、检测判定、告警与封禁。每个环节都有独立的坑而源码和文档的作用是给你一个能改、能测、能讲清楚的骨架而不是一个黑匣子。这篇笔记就按这条流水线拆开讲适合做毕设的学生也想自己搭轻量级 IDS 的运维新手。2. 系统拆解与 Python 选型理由一个毕设级 IDS 的模块边界在哪里2.1 抓包层为什么选 Scapy三种方案的取舍Python 做流量采集绕不开三个常用库Scapy、dpkt、以及基于 tshark 的 pyshark。Scapy 的特点是协议解析内置、交互式调试方便写pkt[IP]就能取到 IP 层字段对新手极其友好。dpkt 更快但协议解析基本靠手写适合对性能敏感、又愿意读 RFC 的老手。pyshark 解析能力最强但它依赖 Wireshark 的 tshark 二进制部署时要额外装一坨东西。毕设场景里流量规模通常是小网卡抓包或离线 pcap 回放Scapy 完全够用而且它在 pcap 文件读写、构造测试报文方面的生态也最完整。从源码结构来看抓包模块用 Scapy 也是最容易讲清楚的一层一个回调函数处理一个包逻辑独立方便后面做单元测试。方案性能依赖解析能力适合场景Scapy中等仅 libpcap内置常见协议毕设、中小流量、快速原型dpkt较高无需手写解析追求吞吐、熟悉协议格式pyshark低tshark 二进制最强需要 Wireshark 级解析结果2.2 检测引擎的两条路线规则匹配与行为基线抓包只是拿到原料真正决定系统价值的是检测引擎。常见做法是两条路线都做一是规则匹配像 Snort 签名那样在报文负载里找特征字符串或正则优点是解释性强命中即告警答辩时可以直接讲“这条规则为什么这么写”二是行为基线统计每个源 IP 在时间窗口内的连接数、包长均值、协议分布建立正常画像偏离到一定程度就告警优点是能检出“未知攻击”缺点是误报率高阈值全靠调。两条路线在系统里不冲突反而互补。规则负责已知攻击的精确识别基线负责可疑行为的兜底。毕设的创新点通常就落在基线统计上——比如传统规则引擎对慢速扫描基本无感但窗口内 SYN 包数量显著偏离均值就能暴露它。2.3 整体模块划分源码应该长成什么样一个可维护的毕设项目代码结构应该能一眼看出职责边界。我一般会分成五个模块capture/管抓包parse/管协议解析与字段抽取features/管特征向量生成detect/管规则与统计检测action/管告警入库和防火墙联动。另加一个web/目录做结果展示用 Flask 起一个简单的仪表盘。这个划分的价值在于答辩时你能指着一个文件说“这是特征提取”“这是检测阈值”而不是在一坨代码里翻半天。文档说明部分也按这个结构写每章对应一个模块评审老师顺着目录就能看懂你的设计思路。3. 流量采集与特征提取从网卡到结构化数据的完整链路3.1 用 Scapy 实时抓包主循环怎么写才不丢包先把最小的实时抓包跑通。下面这段代码解决两个问题一是回调函数必须短二是抓包线程要独立不能阻塞主程序。很多人直接把抓包写在主线程里界面一卡包就开始丢。from scapy.all import sniff import queue import threading # 用有界队列隔离抓包与处理避免回调里做重活 packet_queue queue.Queue(maxsize1000) def packet_callback(pkt): # 回调里只入队不做协议解析保证sniff主循环足够快 packet_queue.put(pkt) def start_capture(ifaceeth0): # daemon线程让抓包在后台运行主程序可以做别的事 sniffer threading.Thread( targetsniff, kwargs{ iface: iface, prn: packet_callback, store: False, # 不把原始包保存在内存中 }, daemonTrue, ) sniffer.start() return sniffer # 使用示例启动后每秒看一次队列积压量 start_capture(eth0)这段代码有几个参数值得较真。storeFalse是必开的否则 Scapy 会把所有经过的包缓存在内存里跑一晚上就能吃掉几个 GB这是最常见的翻车点。prn指定回调函数队列的maxsize1000起到背压作用如果消费者处理不过来队列满了抓包线程会阻塞间接保护了整个系统的内存稳定性。网卡名建议先执行ip link确认eth0在云服务器上经常不叫这个名字。3.2 特征提取从一个包到一个字典拿到的 raw packet 要转成结构化特征这一步是为了让检测引擎不依赖 Scapy 的对象模型。每个包提取五元组、协议号、包长、TCP 标志位和时间戳后续无论是规则匹配还是统计基线都用同一份特征香。from scapy.all import IP, TCP, UDP def extract_features(pkt): # 非IP包直接跳过抓包环境里常有ARP、LLDP等无意义报文 if not pkt.haslayer(IP): return None ip pkt[IP] feat { ts: pkt.time, src: ip.src, dst: ip.dst, proto: ip.proto, pkt_len: len(pkt), } if pkt.haslayer(TCP): tcp pkt[TCP] feat[sport] tcp.sport feat[dport] tcp.dport feat[flags] int(tcp.flags) # 转成整型便于后续位运算匹配 elif pkt.haslayer(UDP): udp pkt[UDP] feat[sport] udp.sport feat[dport] udp.dport feat[flags] 0 # 有负载时截前64字节作负载样本既能识别特征又控制体积 if pkt.haslayer(Raw): feat[payload_head] bytes(pkt[Raw].load)[:64] else: feat[payload_head] b return feat参数说明pkt.time是抓包时刻的 Unix 时间戳离线回放时要注意它可能和当前时间不一致pkt_len用的是len(pkt)而不是ip.len前者是链路上实际收到的字节数后者是 IP 头里的声明长度两者在分片和填充场景下会不同。负载只截前 64 字节主要为了控制入库体积和内存占用规则匹配不需要完整负载。提取失败的包直接返回None调用方过滤即可。3.3 结构化存储为什么用 SQLite 而不是 MySQL特征存哪里直接影响系统的可演示性。选 SQLite 的理由很实际单文件数据库答辩时拷走就带走不用单独装服务并发写简单。不要用check_same_threadFalse去跨线程共享同一个连接这是 SQLite 多线程写入报错的重灾区。正确做法是消费线程自己独享连接。import sqlite3 import time def writer_loop(queue_ref, db_pathflows.db): # 每个线程单独创建连接不跨线程复用 conn sqlite3.connect(db_path) conn.execute(PRAGMA journal_modeWAL) conn.execute( CREATE TABLE IF NOT EXISTS flows ( ts REAL, src TEXT, dst TEXT, proto INTEGER, sport INTEGER, dport INTEGER, flags INTEGER, pkt_len INTEGER, payload_head BLOB ) ) while True: try: pkt queue_ref.get(timeout1) except queue.Empty: continue feat extract_features(pkt) if feat is None: continue # 预编译语句 批量写入比逐条execute快一个量级 conn.execute( INSERT INTO flows VALUES (?,?,?,?,?,?,?,?,?), (feat[ts], feat[src], feat[dst], feat[proto], feat.get(sport), feat.get(dport), feat.get(flags), feat[pkt_len], feat[payload_head]), ) conn.commit()这里PRAGMA journal_modeWAL是关键参数。WAL 模式下读写不互斥能明显降低“database is locked”的出现频率但要注意它只是缓解不是根治真正根治是保证只有这一个线程在写。timeout1配合queue.Empty让消费线程空转时不空耗 CPU。生产环境里 commit 的频率可以考虑攒批再提交比如每 100 条一次毕设场景直接每条提交也够用。4. 检测引擎的三个实现层次规则、统计与轻量学习4.1 从 Snort 规则到 Python 匹配器最少代码做精确识别规则匹配器是检测引擎的“脸面”因为它最好讲、最好调试。规则文件我通常自己定义一种极简格式兼容常见 Snort 规则的关键语义协议、方向、目的端口、告警消息、特征字符串。读取规则后在内存里编译成正则每次特征进来逐条匹配。import re import os RULE_FILE rules.txt # 规则格式: alert tcp any any - any 80 msg:敏感路径探测 content:/etc/passwd def load_rules(pathRULE_FILE): rules [] if not os.path.exists(path): return rules with open(path, r, encodingutf-8) as f: for line in f: line line.strip() if not line or line.startswith(#): continue parts line.split() # parts: [alert, tcp, any, any, -, any, 80, msg:..., content:...] rule { proto: parts[1], dport: int(parts[6]) if parts[6].isdigit() else None, msg: re.search(rmsg:(.*?), line).group(1), content: re.search(rcontent:(.*?), line).group(1).encode(), } rule[content_re] re.compile(re.escape(rule[content])) rules.append(rule) return rules def match_rules(feat, rules): for rule in rules: if rule[proto] tcp and feat.get(proto) ! 6: continue if rule[proto] udp and feat.get(proto) ! 17: continue if rule[dport] and feat.get(dport) ! rule[dport]: continue # 在负载头里找特征字符串找不到就继续下一条规则 if rule[content_re].search(feat.get(payload_head, b)): return rule[msg], rule[content] return None, None几个关键点re.escape保证规则里的特殊字符不会被当成正则语法解释这是新手最容易踩的坑——写一条content:/etc/passwd结果因为.匹配任意字符而疯狂误报。规则匹配是纯字符串查找故意不查全负载用第 3 章截出来的前 64 字节一方面匹配速度极快另一方面对分片攻击也有一定敏感性。如果你的毕设选题偏向 Web 攻击检测可以在content之外再加一个uri字段匹配 HTTP 请求行那通常解析不出来需要单独拆 HTTP 层。4.2 统计基线用滑动窗口识别“不正常”规则匹配对未知攻击无能为力统计基线就是用来补这块的。思想很简单每个源 IP 在最近 N 秒内发了多少个包、多少种标志位这些数字在正常流量里是相对稳定的。一旦某个窗口内的包数超过历史均值的 3 个标准差就判定异常。用deque实现滑动窗口最省内存。from collections import defaultdict, deque import statistics WINDOW_SEC 60 STD_MULTIPLIER 3.0 # 结构: ip - deque([(ts, 1), ...])只用ts和时间戳算窗内总数 ip_packet_times defaultdict(deque) def update_baseline(feat): src feat[src] q ip_packet_times[src] q.append((feat[ts], 1)) # 弹出窗口外的旧数据 cutoff feat[ts] - WINDOW_SEC while q and q[0][0] cutoff: q.popleft() if len(q) 10: return None, 样本太少继续攒数据 # 每秒包数量近似 窗口内包数 / 窗口秒数 sample [t for t, _ in q] mean statistics.mean(sample) stdev statistics.stdev(sample) if len(sample) 1 else 0.0 if stdev 0: return None, 流量过于稳定无法计算波动 # 最近10秒的包数比历史均值偏移超过阈值才告警 recent_10s sum(1 for t, _ in q if t feat[ts] - 10) if recent_10s mean STD_MULTIPLIER * stdev: return src, f最近10秒包数 {recent_10s}均值 {mean:.1f}标准差 {stdev:.1f} return None, 正常参数要说明白WINDOW_SEC60决定基线的记忆长度60 秒适合检测扫描型攻击太长了会稀释突发现象太短了基线本身就不稳定。STD_MULTIPLIER3.0是 3σ 原则理论上正态分布下只有 0.3% 的正常点会被误判但网络流量不是正态分布实际使用里 2.5 到 4 之间都要测。len(q) 10这个条件是防止冷启动阶段误报——新 IP 前几个包就触发告警没有任何意义这是绝大多数统计检测入门代码里没有的细节。4.3 轻量机器学习孤立森林怎么融进现有流程如果你想让毕设多一些“智能”成分可以加一个孤立森林做无监督异常检测但不要搞深度学习。无监督的意义是不需要标注好的攻击数据只要拿一段正常流量的特征训练然后对实时特征打分即可。对毕设来说可解释性和训练成本远比 AUC 高 0.01 重要。import joblib from sklearn.ensemble import IsolationForest def train_model(feature_vectors_pathnormal_features.npy): # 读入正常流量的特征矩阵每行是 [src_entropy, pkt_len_mean, ports_per_window] import numpy as np X np.load(feature_vectors_path) model IsolationForest( contamination0.1, # 假定的异常比例 n_estimators100, # 树的数量100够用调大只会增加耗时 random_state42 ) model.fit(X) joblib.dump(model, isf_model.pkl) return model def predict_anomaly(feat, model): import numpy as np # 特征格式要和训练时完全一致 vec np.array([[feat[src_entropy], feat[pkt_len_mean], feat[ports_per_window]]]) # score_samples返回负值值越小越异常decision_function小于0判异常 score model.score_samples(vec)[0] return score -0.1, scorecontamination0.1是最容易让人困惑的参数它表示模型假设训练集里有 10% 的异常样本实际上你的“正常流量”里可能一点异常都没有这时设 1% 更合适。另一个血泪经验特征向量必须和训练时完全对得上第 3 章存 SQLite 时就要同步落一份 numpy 格式的特征文件别在训练完以后才想起来特征代码改过。训练完一定要 dump 成 pkl演示时直接加载现场训练 30 秒会非常尴尬。5. 从检测到防御联动 iptables 的落地细节与五条避坑记录5.1 封禁动作怎么做才算“防御系统”检测的结果如果不能触发动作那它只能叫“监控系统”不叫“防御系统”。最常见的动作是调用 iptables 封禁源 IP封禁时长要有限制否则误报一次就把正常用户关外面了。下面这段代码同时实现了插入封禁规则和定时解封。import subprocess import threading def block_ip(ip, block_seconds300): # 先查是否已封禁避免重复插入同样规则 check_cmd fiptables -C INPUT -s {ip} -j DROP if subprocess.run(check_cmd, shellTrue).returncode 0: return False apply_cmd fiptables -I INPUT -s {ip} -j DROP subprocess.run(apply_cmd, shellTrue, checkTrue) print(f[ACTION] 已封禁 {ip}{block_seconds} 秒后自动解封) # 定时器到点后解封使用独立线程避免阻塞检测主循环 threading.Timer(block_seconds, unblock_ip, args(ip,)).start() return True def unblock_ip(ip): cmd fiptables -D INPUT -s {ip} -j DROP subprocess.run(cmd, shellTrue, checkTrue) print(f[ACTION] 已解封 {ip})这段代码里我特意先-C查规则是否存在这是个容易被忽略的幂等问题连续触发告警时如果没有查重同一个 IP 会被插入几十条 DROP 规则。-I是插到链首-D按同样条件删除正好配对。更工业化的做法是用 ipset 维护封禁集合避免规则数膨胀但毕设里 iptables 直连已经足够。运行这段代码必须有 root 权限给程序加上 sudo 执行时的日志记录很有必要。5.2 五条避坑记录现象、原因、解决现象一sniff启动后完全没有输出程序也不报错。原因多半是没权限——Linux 下用 libpcap 抓包需要 root 或CAP_NET_RAW能力普通用户调用 Scapy 时包回调一次都不触发。也可能是网卡名写错云服务器上常见ens33、eth0混用。 解决先sudo -s切到 root 再跑网卡名用ip link确认后再填。抓包丢包问题也常被当成玄学实际上多数是权限和网卡名这两个低级错误。现象二程序跑几分钟后弹出sqlite3.OperationalError: database is locked。原因肯定有多个线程在同时写同一个数据库文件。即使你只在 writer 线程里写只要别的地方调了sqlite3.connect()也可能引起锁竞争最常见的就是 Flask 展示页面里顺手查了一下数据库。 解决确保所有读写入口收敛到一个连接或者展示层用只读连接打开modero。不要用check_same_threadFalse硬顶着那只是把崩溃推迟。现象三检测引擎告警了iptables 也执行了但测试流量还是能正常访问。原因封禁规则插到了 INPUT 链可流量是从 FORWARD 链过的比如你这套系统部署在路由/网桥模式图省事只写了 INPUT。也可能是目标机器走 IPv6而 iptables 只管 IPv4。 解决先确认拓扑——是旁路镜像还是串联网关串联网关就要同时操作 FORWARD 链IPv6 环境还要同步跑ip6tables -I INPUT -s ip -j DROP。现象四正常用户被误封一天内封禁列表里全是误报。原因阈值设得太激进。统计基线里STD_MULTIPLIER设成 2.0 以下加上窗口期样本不足就开始判断偶发的浏览器预取流量就会触发告警。规则匹配那边也更危险一条宽泛的content规则能匹配所有含该字符串的正常包。 解决规则匹配的告警不要直接封禁先进入一个“观察名单”连续 3 次命中才动作统计基线把阈值回调到 3.0 以上并保证窗口内至少有 30 条样本再参与判断。毕设论文里把这两层延迟机制写上反而是个加分项。现象五离线回放 pcap 测试时告警时间戳全部乱套展板上的时间线成了乱码。原因tcpreplay默认以最快速度发包pcap 里的时间戳被压缩到几秒内打完特征里的pkt.time全部挤在一起滑动窗口统计直接失真。 解决回放时用tcpreplay --pps200限制每秒发包数或者回放前用editcap重置时间戳。回放的体积也要控制一次灌进去几 GB 的 pcap内存和队里都会承受不了分片回放更稳。6. 验证效果与进阶方向用可控重放实验给毕设一份可信结论毕设答辩时最怕的是“功能演示完但拿不出数据支撑”。正确做法是设计一个可控实验先采集 30 分钟正常办公流量建立基线然后用 Scapy 构造若干条带特征字符串的测试报文注入网络模拟一次敏感路径探测再观察系统是否能在大约 10 秒内告警并触发封禁。每次实验记录四个指标真正例数、假正例数、假负例数、封禁延迟。表格形式写进论文足够交代方法论。这套系统的进阶方向有三条值得投入把单机规则引擎改成动态热加载运行时用inotify监听规则文件变化免重启生效把 SQLite 换成 TimescaleDB 之类的时序库让特征曲线能以秒级粒度回溯在统计基线之上做关联分析比如同一源 IP 在 5 分钟内同时触发“端口扫描”和“敏感路径探测”则告警优先级自动上调。这三条任选一条深度都足够支撑毕业论文的一章“进一步工作”。我的习惯是每次调参都存一份日志把当时的阈值、窗口大小、实验数据量、结果指标写进一个 JSON 文件。答辩时直接翻日志讲“我把标准差阈值从 2.0 调到 3.0误报率降了多少”比口头说“我调过很多次”有说服力得多。这套系统真正的价值不是替你挡住所有攻击而是给你一个能反复验证、能讲清楚每次决策依据的检测闭环。希望帮到你。本文还有配套的精品资源点击获取