☰
Python网络入侵检测与防御系统:从抓包到自动封禁的毕业设计实战
2026/9/25 9:40:41 网站建设 项目流程

简介:这是一套面向计算机相关专业学生的网络入侵检测与防御系统毕业设计项目源码,适合正在做大作业、课程设计或期末大作业的学习者,也可作为项目实战练习的参考。项目经导师指导并认可通过,评审分99分,代码完整可运行,对新手较为友好。资源包共38个文件,以14个Python源码文件为核心,辅以9个编译缓存文件、4个HTML页面、3个JavaScript脚本及JSON、YML、Markdown、CSS、Dockerfile、Shell、TXT等配置与说明文件,整体约97KB,涵盖应用主程序、功能模块、前端模板、静态资源与容器化部署脚本,目录结构清晰。已有62人学习关注。通过该资源,读者可获得完整的入侵检测与防御实现方案、模块划分思路与部署配置参考,便于快速理解系统架构、对照调试运行,并在此基础上完成自己的毕业设计或课程实践。

1. 从零读懂网络入侵检测与防御系统:Python 毕业设计到底在做什么

很多人第一次拿到「基于 Python 的网络入侵检测与防御系统」这个毕业设计题目时,脑子里冒出来的画面是电影里那种绿色代码瀑布,或者一个能自动挡住黑客的炫酷界面。真动手做的时候才发现,核心问题其实特别朴素:怎么让一台机器盯着网络流量,判断哪些是正常的、哪些是可疑的,然后在可疑流量造成破坏之前把它拦下来。这件事拆开就是三个动作——采集、判断、处置,Python 在里面扮演的是「胶水 + 大脑」的角色,负责把抓包库、特征提取、检测模型和防火墙规则串成一条流水线。

这个方向适合两类人:一类是计算机、网络工程、信息安全专业的毕业生,需要一个既有工程量又不至于完全做不出来的题目;另一类是已经工作、想补一块安全方向实操经验的开发者。它不需要你从零训练一个深度学习模型,也不需要你有一台企业级防火墙,一台能跑 Linux 的虚拟机加 Python 环境就够了。真正决定这个项目能不能拿得出手的,不是模型多先进,而是你有没有把「检测到之后怎么办」这条链路走通——很多人的系统只能报警,不能防御,答辩时一问就露馅。

2. 检测引擎怎么选:从规则匹配到轻量机器学习

2.1 三种主流检测路线的取舍

做入侵检测,绕不开的第一个决策是检测引擎用什么。常见做法有三条路线,各自适合不同的答辩场景和实现成本。

第一条是基于规则的匹配,代表工具是 Snort 和 Suricata。它们的思路是把已知攻击写成规则,比如「某个 IP 在 1 秒内发起了 50 次 SSH 连接」就判定为暴力破解。优点是解释性强,每条告警都能说清楚触发了哪条规则,答辩时老师问「为什么判定这是攻击」你能直接指出来。缺点是只能抓已知模式,遇到变形攻击就瞎了。

第二条是基于统计与阈值的异常检测,比如用滑动窗口统计单位时间内的连接数、包大小分布,偏离基线就报警。实现简单,纯 Python 就能写,但阈值特别难调,网络稍微抖一下误报就上来了。

第三条是基于机器学习分类,用 KDD Cup 99、NSL-KDD 或 CICIDS2017 这类公开数据集训练一个分类器,把流量特征喂进去判断是否异常。这是目前毕业设计里最讨喜的路线,因为「用了机器学习」听起来有技术含量,而且数据集是现成的。

我的建议是规则 + 机器学习混合:用规则处理高频、明确的攻击类型(端口扫描、暴力破解),用机器学习模型兜底那些规则覆盖不到的异常。这样既有可解释性,又有「智能」的卖点。

2.2 用 Python 搭一个最小可跑的检测原型

先不急着上模型,把「抓包 → 提取特征 → 判断」这条链路跑通最重要。下面这段代码用 scapy 抓包,提取每条流的基础特征,先用一个简单阈值做演示。

from scapy.all import sniff, IP, TCP from collections import defaultdict import time # 记录每个源 IP 在时间窗口内的连接次数 conn_counter = defaultdict(list) WINDOW = 10 # 时间窗口,单位秒 THRESHOLD = 30 # 窗口内连接数超过该值判定为可疑 def extract_features(pkt): """从单个数据包提取基础特征""" if IP not in pkt: return None feat = { "src": pkt[IP].src, "dst": pkt[IP].dst, "len": len(pkt), "proto": pkt[IP].proto, "time": time.time() } # TCP 包额外提取标志位和端口 if TCP in pkt: feat["sport"] = pkt[TCP].sport feat["dport"] = pkt[TCP].dport feat["flags"] = str(pkt[TCP].flags) return feat def detect(pkt): feat = extract_features(pkt) if not feat: return now = feat["time"] src = feat["src"] # 清理窗口外的历史记录 conn_counter[src] = [t for t in conn_counter[src] if now - t < WINDOW] conn_counter[src].append(now) # 阈值判定 if len(conn_counter[src]) > THRESHOLD: print(f"[告警] {src} 在 {WINDOW}s 内连接 {len(conn_counter[src])} 次,疑似扫描或暴力破解") if __name__ == "__main__": # 只抓 TCP 包,避免噪音;实际部署时可加过滤条件 sniff(filter="tcp", prn=detect, store=False)

这段代码的逻辑很直白:每来一个包,先提取源 IP、目的 IP、包长、协议、时间戳这几个特征,然后按源 IP 维护一个时间窗口内的连接时间列表,窗口内连接数超过阈值就告警。WINDOW和THRESHOLD是两个必须调的参数——窗口太小会漏掉慢速扫描,太大又会让正常的高频服务误报。我一般会先用WINDOW=10、THRESHOLD=30跑一遍,观察正常业务的基线,再往上调。

提示:scapy 抓包需要 root 权限,Linux 下用sudo python detect.py,Windows 下需要先装 Npcap 并勾选 WinPcap 兼容模式。

2.3 把机器学习模型接进来

规则跑通之后,把阈值判定换成模型预测。用 NSL-KDD 数据集训练一个随机森林,特征维度是 41 维,包括连接时长、协议类型、服务类型、错误率等。训练脚本大概长这样:

import pandas as pd from sklearn.ensemble import RandomForestClassifier from sklearn.model_selection import train_test_split from sklearn.metrics import classification_report import joblib # NSL-KDD 的列名,官方文档里有,这里按顺序定义 COLUMNS = ["duration","protocol_type","service","flag","src_bytes","dst_bytes", "land","wrong_fragment","urgent","hot","num_failed_logins","logged_in", "num_compromised","root_shell","su_attempted","num_root","num_file_creations", "num_shells","num_access_files","num_outbound_cmds","is_host_login", "is_guest_login","count","srv_count","serror_rate","srv_serror_rate", "rerror_rate","srv_rerror_rate","same_srv_rate","diff_srv_rate", "srv_diff_host_rate","dst_host_count","dst_host_srv_count", "dst_host_same_srv_rate","dst_host_diff_srv_rate","dst_host_same_src_port_rate", "dst_host_srv_diff_host_rate","dst_host_serror_rate","dst_host_srv_serror_rate", "dst_host_rerror_rate","dst_host_srv_rerror_rate","label","difficulty"] def load_data(path): df = pd.read_csv(path, names=COLUMNS) # 把多分类标签二值化:normal 为 0,其余攻击类型为 1 df["label"] = df["label"].apply(lambda x: 0 if x == "normal" else 1) # 类别型特征做 one-hot df = pd.get_dummies(df, columns=["protocol_type", "service", "flag"]) return df df = load_data("KDDTrain+.txt") X = df.drop(["label", "difficulty"], axis=1) y = df["label"] X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42) clf = RandomForestClassifier(n_estimators=100, max_depth=20, random_state=42) clf.fit(X_train, y_train) print(classification_report(y_test, clf.predict(X_test))) joblib.dump(clf, "ids_model.pkl")

这里有几个参数值得说清楚。n_estimators=100是树的数量,太少欠拟合,太多训练慢且收益递减,100 到 200 之间比较稳。max_depth=20控制单棵树深度,防止过拟合,NSL-KDD 这种规模的数据集 15 到 25 都合理。test_size=0.3是留出法划分比例,如果数据量小可以改成 0.2。训练完用joblib存模型,线上检测时加载这个 pkl 文件做推理。

注意:NSL-KDD 的特征顺序和列名必须严格对应,网上很多版本列名对不上,训练出来的模型准确率虚高但实际不可用。建议直接从官方文档核对列定义。

3. 防御模块怎么做:从告警到真正拦下来

3.1 用 iptables 做动态封禁

检测出攻击源之后,防御动作最常见的是封 IP。Linux 下用 iptables 加一条 DROP 规则就行,Python 里通过 subprocess 调用:

import subprocess import time BLOCK_LIST = {} # 记录被封 IP 和封禁时间 BLOCK_DURATION = 300 # 封禁时长,单位秒 def block_ip(ip): """封禁指定 IP,已封禁则跳过""" if ip in BLOCK_LIST: return try: subprocess.run( ["iptables", "-I", "INPUT", "-s", ip, "-j", "DROP"], check=True ) BLOCK_LIST[ip] = time.time() print(f"[防御] 已封禁 {ip},时长 {BLOCK_DURATION}s") except subprocess.CalledProcessError as e: print(f"[错误] 封禁 {ip} 失败: {e}") def unblock_expired(): """定期清理过期封禁,避免规则表无限增长""" now = time.time() for ip in list(BLOCK_LIST.keys()): if now - BLOCK_LIST[ip] > BLOCK_DURATION: subprocess.run( ["iptables", "-D", "INPUT", "-s", ip, "-j", "DROP"], check=True ) del BLOCK_LIST[ip] print(f"[防御] 已解封 {ip}")

-I INPUT是插入到规则链头部,保证优先匹配;-s指定源 IP;-j DROP直接丢包。BLOCK_DURATION设 300 秒是个折中,太短攻击者换个时间又来,太长可能误封正常用户。实际部署时建议把BLOCK_LIST持久化到 Redis 或 SQLite,进程重启后不丢状态。

3.2 告警与日志:别让系统变成黑匣子

防御动作必须留痕,否则出了问题没法回溯。日志至少记录四样东西:时间、源 IP、判定类型、处置动作。用 Python 的 logging 模块写到文件,同时可以推一份到控制台方便调试。

import logging import json logger = logging.getLogger("ids") logger.setLevel(logging.INFO) handler = logging.FileHandler("ids_alert.log", encoding="utf-8") handler.setFormatter(logging.Formatter("%(asctime)s %(message)s")) logger.addHandler(handler) def log_alert(src_ip, attack_type, action): record = { "src_ip": src_ip, "attack_type": attack_type, "action": action } logger.info(json.dumps(record, ensure_ascii=False))

用 JSON 格式写日志的好处是后续可以用脚本直接解析做统计,比如「今天封了多少个 IP」「哪类攻击最多」。答辩时如果能拿出一张攻击类型分布的统计图,比干讲代码有说服力得多。

3.3 检测与防御的联动架构

把前面几块拼起来,整个系统的数据流是这样的:scapy 持续抓包,特征提取模块把包转成特征向量,检测模块(规则或模型)给出判定,如果判定为攻击就调用防御模块封 IP 并写日志。这里有个工程上的坑:抓包和检测如果在同一个线程里,检测耗时长了会丢包。常见做法是抓包线程只负责把包塞进队列,检测和防御在另一个线程消费队列。

import queue import threading pkt_queue = queue.Queue(maxsize=10000) def capture_thread(): """抓包线程:只负责入队""" sniff(filter="tcp", prn=lambda p: pkt_queue.put(p), store=False) def process_thread(): """处理线程:出队、检测、防御""" while True: pkt = pkt_queue.get() feat = extract_features(pkt) if feat and is_attack(feat): block_ip(feat["src"]) log_alert(feat["src"], "suspicious", "block") threading.Thread(target=capture_thread, daemon=True).start() threading.Thread(target=process_thread, daemon=True).start()

maxsize=10000是队列上限,防止内存被撑爆。队列满了之后put会阻塞,抓包线程会短暂卡住,这是可接受的——总比丢包强。daemon=True让线程随主进程退出,避免程序关不掉。

4. 避坑指南:那些让我熬夜排查的翻车现场

4.1 抓不到包或抓到一堆无关流量

现象:程序跑起来但一个包都抓不到,或者抓到的全是自己的 SSH 流量。

原因:网卡没开混杂模式,或者 filter 写得太宽。虚拟机环境下还可能是网卡类型选错了(NAT 模式抓不到宿主机和其他虚拟机的流量)。

解决:确认用sudo运行;filter 精确到端口或协议,比如tcp port 80;虚拟机网络改成桥接模式,这样能抓到局域网内其他机器的流量。

4.2 模型在测试集上准确率 99%,实际跑起来全是误报

现象:离线评估指标漂亮,一上线正常业务被大量封禁。

原因:训练数据(NSL-KDD)是十几年前的数据集,和当前真实网络流量分布差异巨大;另外 one-hot 编码时训练集和线上推理的特征列顺序不一致,导致模型输入错位。

解决:线上推理前必须保证特征列和训练时完全一致,把训练时的列顺序存下来,推理时按同样顺序对齐;如果条件允许,用自己抓的正常流量做一次微调或至少做阈值校准。

4.3 iptables 规则越加越多,最后把正常用户也封了

现象:系统跑了一天,规则表里几百条 DROP,正常访问也被拦。

原因:只封不解,或者解封逻辑没跑起来;误报的 IP 也被封了。

解决:必须有定时清理任务,BLOCK_DURATION到期自动解封;对封禁操作加白名单,内网 IP 和已知可信 IP 不封;封禁前先记录不直接 DROP,观察一段时间确认是攻击再封。

4.4 程序跑一会儿就内存暴涨

现象:进程内存持续增长,几小时后被系统 kill。

原因:conn_counter这类字典只增不减,或者队列无上限导致积压。

解决:所有按 IP 维护的状态都要有清理机制,定期删除过期条目;队列设maxsize;用tracemalloc或memory_profiler定位泄漏点。

4.5 答辩时被问「你的系统和 Snort 有什么区别」

现象:答不上来,显得项目没有价值。

原因:只做了功能,没想过定位。

解决:提前想清楚差异化——比如「Snort 是纯规则,我加了机器学习兜底未知攻击」「我做了自动封禁联动,Snort 默认只告警不处置」「我针对毕业设计场景做了轻量化,单机可跑不依赖专用硬件」。把「轻量、可解释、检测防御一体」这三点讲透就够了。

5. 让系统更耐打的三个进阶技巧

5.1 用滑动窗口替代固定阈值

固定阈值最大的问题是扛不住突发流量。比如公司早上九点大家集中登录,连接数天然就高,固定阈值必然误报。改成滑动窗口加动态基线会稳很多:维护最近 N 个窗口的连接数均值,当前窗口超过均值 3 倍标准差才告警。

import numpy as np from collections import deque history = deque(maxlen=60) # 保留最近 60 个窗口的统计 def dynamic_threshold(current_count): if len(history) < 10: history.append(current_count) return False mean = np.mean(history) std = np.std(history) + 1e-6 # 防止除零 history.append(current_count) # 超过均值 3 倍标准差判定异常 return current_count > mean + 3 * std

maxlen=60控制历史长度,太长反应迟钝,太短基线不稳。3 * std是经验值,2 倍更敏感但误报多,4 倍更保守但可能漏报,根据自己场景调。

5.2 特征工程比换模型更有效

很多人一上来就想换 XGBoost、换深度学习,其实在入侵检测这个场景,特征质量比模型复杂度重要得多。除了 NSL-KDD 自带的 41 维特征,可以自己加几个:单位时间内的唯一目的端口数(扫描行为特征)、SYN 包占比(SYN Flood 特征)、平均包长(某些攻击的包长分布和正常流量差异明显)。加特征之后用同样的随机森林跑一遍,往往比换模型提升更明显。

5.3 用混淆矩阵定位误报来源

不要只看准确率,把混淆矩阵打出来,看清楚是误报多还是漏报多。误报多就调高判定阈值或加白名单,漏报多就降低阈值或补充规则。答辩时如果能展示「我通过分析混淆矩阵把误报率从 15% 降到 5%」这个过程,比单纯说「准确率 95%」有说服力得多。

from sklearn.metrics import confusion_matrix import seaborn as sns import matplotlib.pyplot as plt cm = confusion_matrix(y_test, clf.predict(X_test)) sns.heatmap(cm, annot=True, fmt="d", cmap="Blues") plt.xlabel("预测") plt.ylabel("真实") plt.savefig("confusion_matrix.png")

最后说个我自己的习惯:这个项目我前后重构过三次,前两次都栽在「检测和防御耦合太紧」上,改一处逻辑要动三个文件。第三次把抓包、检测、防御拆成三个独立模块,用队列通信,后面加新检测规则或者换防御策略都只动一个文件。如果你也在做这个题目,建议一开始就把模块边界划清楚,别图省事全写在一个脚本里,后期改起来会感谢自己。希望帮到你。

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

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询