简介：这是一套面向高校计算机相关专业学生与安全初学者的网络入侵检测与防御系统完整源码，可作为毕业设计或课程设计项目直接使用。项目基于Python构建，围绕实时流量分析、攻击检测、自动防御与可视化监控四大能力展开，涵盖流量捕获与异常识别、恶意流量拦截、威胁告警与自动响应、网络状态与攻击统计可视化等模块，并提供直观的Web操作界面。资源包共38个文件，以14个py源码文件为核心，辅以html、js、css等前端页面与静态资源，另有json配置、yml与Dockerfile容器化文件及安装脚本，整体约92KB，结构紧凑、便于部署。目前已有172人学习关注。读者可获得完整可运行的项目代码、详细运行指南与Docker Compose部署方案，快速理解Scapy抓包、Flask-SocketIO实时通信与MongoDB日志存储的协作方式，并在此基础上完成功能扩展与答辩演示。
1. 从一份毕设源码说起：Python 网络入侵检测与防御系统到底在做什么
很多同学做毕设时都会搜「Python 网络入侵检测与防御系统源码」，但真正拿到一份能跑起来的代码之前，得先想清楚这套系统在链路里站在哪。它本质上是一个跑在主机或网关上的流量哨兵：抓包拿到原始数据帧，解析出 IP、端口、协议和载荷特征，用规则或模型判断这条流量是不是攻击，命中之后触发阻断动作，同时把结果推到前端做可视化监控。整条链路是「采集 → 解析 → 检测 → 防御 → 展示」，缺一环都只能算半成品。
这套东西适合两类人：一类是安全方向毕设需要完整闭环、能演示攻击检测和自动防御效果的学生；另一类是想把 Python 网络编程、多线程、抓包和简单机器学习串起来练手的工程师。它不追求工业级 IDS 的吞吐，但胜在链路完整、每一层都能自己改。下面按真实落地顺序拆开讲，从环境到抓包、从检测到防御、再到可视化，最后给几个能直接抄的调参和排错习惯。
2. 环境与依赖：把 Python 抓包和检测的地基打稳
2.1 为什么选 Python 而不是别的语言做这套系统
选 Python 做网络入侵检测与防御系统，核心原因是生态。抓包有 Scapy 和 pcapy，数据处理有 NumPy 和 Pandas，检测侧可以接 scikit-learn 做轻量分类，前端可视化用 Flask 加 ECharts 就能出图。C 或 Go 性能更好，但毕设周期内把抓包、解析、检测、阻断、展示全链路用 C 写一遍，时间成本不现实。Python 的短板是 GIL 导致多线程吃不满多核，但抓包和检测本身是 IO 密集加轻计算，用多进程或异步队列就能绕开。
常见做法是：抓包主线程只负责收包入队，解析和检测放到独立进程或线程池，防御动作异步执行，避免阻塞抓包。这样即使检测逻辑稍慢，也不会丢包太严重。选型上，Scapy 适合做协议解析和构造测试包，性能要求高的抓包环节可以用 pcapy 或直接调 libpcap 的 Python 绑定。
2.2 依赖安装与 Python 版本选择
建议用 Python 3.8 及以上，3.8 是很多安全库兼容性最稳的版本。安装依赖时不要一个个 pip install，直接写 requirements.txt 一次装完，方便复现。
# requirements.txt scapy==2.5.0 # 抓包与协议解析 flask==2.3.2 # 可视化后端 flask-socketio==5.3.4 # 实时推送检测结果 scikit-learn==1.3.0 # 轻量检测模型 numpy==1.24.3 # 特征数值计算 pandas==2.0.3 # 流量特征整理 joblib==1.3.1 # 模型持久化# 安装命令，建议在虚拟环境里执行 python -m venv venv source venv/bin/activate # Linux/macOS venv\Scripts\activate # Windows pip install -r requirements.txt逻辑说明：Scapy 负责 sniff 抓包和解析 IP/TCP/UDP 层；Flask 加 SocketIO 把检测事件实时推到浏览器；scikit-learn 用来跑一个简单的异常检测或分类模型。参数上，Scapy 的 sniff 函数关键参数是 iface（网卡名）、prn（回调函数）、store（是否缓存包，设为 0 省内存）。Linux 下抓包需要 root 或给 Python 加 CAP_NET_RAW 能力，Windows 下需要装 Npcap 并勾选 WinPcap 兼容模式。
提示：如果 pip 装 Scapy 后 import 报缺少 pcap 相关模块，先确认系统里有没有 libpcap，Linux 用 apt install libpcap-dev，Windows 装 Npcap 后重启终端。
2.3 目录结构怎么摆才不乱
一份能维护的源码，目录要按职责分，不要所有 py 文件堆在根目录。常见做法是：
ids_project/ ├── capture/ # 抓包与解析 │ └── sniffer.py ├── detect/ # 检测规则与模型 │ ├── rules.py │ └── model.py ├── defense/ # 自动防御动作 │ └── blocker.py ├── web/ # 可视化后端与前端 │ ├── app.py │ └── templates/ ├── config.py # 全局参数 └── main.py # 入口，串联各模块这样拆的好处是，检测逻辑改动不影响抓包，防御策略调整不用动前端。config.py 里集中放网卡名、检测阈值、阻断开关、数据库路径这些容易变的参数，避免硬编码散落各处。
3. 实时流量分析：用 Scapy 抓包并解析出可用特征
3.1 抓包回调里到底该提取哪些字段
实时流量分析的第一步不是检测，是把原始包变成结构化特征。一个 TCP 包能提取的字段很多，但检测侧真正有用的通常是：源 IP、目的 IP、源端口、目的端口、协议号、包长度、TCP 标志位、时间戳。如果是 UDP 或 ICMP，标志位字段换成对应类型码。这些字段拼成一条记录，后续规则和模型都基于它判断。
from scapy.all import sniff, IP, TCP, UDP, ICMP import time def extract_features(pkt): """从原始包中提取检测所需字段，返回 dict""" if IP not in pkt: return None feat = { "timestamp": time.time(), "src_ip": pkt[IP].src, "dst_ip": pkt[IP].dst, "proto": pkt[IP].proto, "length": len(pkt), } if TCP in pkt: feat["src_port"] = pkt[TCP].sport feat["dst_port"] = pkt[TCP].dport feat["flags"] = int(pkt[TCP].flags) # SYN/FIN/ACK 等标志位 elif UDP in pkt: feat["src_port"] = pkt[UDP].sport feat["dst_port"] = pkt[UDP].dport feat["flags"] = 0 elif ICMP in pkt: feat["src_port"] = 0 feat["dst_port"] = 0 feat["flags"] = int(pkt[ICMP].type) else: return None return feat逻辑说明：先判断有没有 IP 层，非 IP 包直接丢弃；再按 TCP/UDP/ICMP 分支取端口和标志位。参数上，pkt[TCP].flags 是 FlagValue 类型，转 int 后才能进模型或做位运算。时间戳用 time.time() 而不是包内时间，方便和防御动作对齐。
3.2 抓包性能与丢包控制
Scapy 的 sniff 默认在单线程里跑回调，回调里做重活会丢包。常见优化是回调里只做最轻的提取，然后把特征塞进 queue.Queue，由消费者线程或进程做检测。队列要设 maxsize，满了就丢最旧的包并计数，避免内存涨爆。
from queue import Queue from threading import Thread pkt_queue = Queue(maxsize=5000) drop_count = 0 def on_packet(pkt): global drop_count feat = extract_features(pkt) if feat is None: return try: pkt_queue.put_nowait(feat) except Exception: drop_count += 1 # 队列满，记录丢包 def consumer(): while True: feat = pkt_queue.get() # 这里调用检测逻辑 detect(feat) Thread(target=consumer, daemon=True).start() sniff(iface="eth0", prn=on_packet, store=0)逻辑说明：put_nowait 不阻塞抓包回调，队列满时直接丢并计数，drop_count 可以暴露到监控面板。参数上，maxsize 根据内存和流量大小调，千兆网卡满速时 5000 可能偏小，可以到 20000。store=0 让 Scapy 不缓存原始包，省内存。iface 在 Linux 用 ip link 查，Windows 用 get_windows_if_list() 查。
注意：如果发现丢包率很高，先看是不是回调里做了字符串拼接或日志打印，这些操作比想象中慢，挪到消费者线程里做。
3.3 特征窗口与流聚合
单包特征只能做简单规则，要检测扫描、爆破、DDoS 这类行为，得按时间窗口聚合。常见做法是维护一个滑动窗口，按源 IP 或五元组统计单位时间内的包数、SYN 数、目的端口数。窗口大小一般 1 秒到 10 秒，太短噪声大，太长响应慢。
from collections import defaultdict, deque import time # 每个源 IP 维护一个时间戳队列 syn_window = defaultdict(deque) WINDOW = 5 # 秒 SYN_THRESHOLD = 100 # 5 秒内 SYN 超过这个数判定为扫描 def check_syn_flood(feat): if feat.get("flags") is None: return False # TCP SYN 标志位是 0x02 if not (feat["flags"] & 0x02): return False now = feat["timestamp"] q = syn_window[feat["src_ip"]] q.append(now) while q and now - q[0] > WINDOW: q.popleft() return len(q) > SYN_THRESHOLD逻辑说明：用 deque 存时间戳，每次新包进来先剔除窗口外的旧记录，再看数量是否超阈值。参数上，WINDOW 和 SYN_THRESHOLD 要按自己网络基线调，内网正常业务也可能有突发 SYN，阈值太低会误报。这段逻辑放在消费者线程里跑，不要放抓包回调。
4. 攻击检测：规则引擎和轻量模型怎么配合
4.1 规则检测覆盖哪些高频攻击
规则引擎负责已知攻击特征，落地快、可解释。常见规则包括：端口扫描（短时间内大量不同目的端口）、SYN Flood（窗口内 SYN 数超阈值）、ICMP Flood、暴力破解（同一目的端口短时间大量连接尝试）、异常大包或畸形包。规则用 Python 字典或列表配置，每条规则一个函数，命中返回攻击类型和置信度。
RULES = [] def rule(attack_type): def wrapper(fn): RULES.append((attack_type, fn)) return fn return wrapper @rule("port_scan") def detect_port_scan(feat, state): """同一源 IP 在窗口内访问不同目的端口数超阈值""" key = feat["src_ip"] ports = state.setdefault(key, set()) ports.add(feat.get("dst_port", 0)) return len(ports) > 50 def run_rules(feat, state): hits = [] for attack_type, fn in RULES: try: if fn(feat, state): hits.append(attack_type) except Exception: continue return hits逻辑说明：用装饰器注册规则，新增规则不用改主流程。state 是跨包状态字典，按源 IP 存端口集合等。参数上，端口数阈值 50 是经验值，内网服务器正常也可能连很多端口，建议先跑观察模式统计基线再定。
4.2 轻量模型补规则覆盖不到的异常
规则只能覆盖已知模式，遇到变种或未知攻击会漏。补一个轻量异常检测模型，用正常流量训练，推理时偏离正常分布就告警。毕设场景常用 IsolationForest 或 OneClassSVM，特征用窗口统计量：包速率、平均包长、SYN 比例、目的端口熵。
from sklearn.ensemble import IsolationForest import numpy as np import joblib def train_model(normal_features, path="model.pkl"): """normal_features: 二维数组，每行一个窗口的特征""" model = IsolationForest( n_estimators=100, contamination=0.05, # 预期异常比例 random_state=42 ) model.fit(normal_features) joblib.dump(model, path) return model def load_and_predict(features, path="model.pkl"): model = joblib.load(path) # 返回 1 正常，-1 异常 pred = model.predict(np.array([features])) return pred[0] == -1逻辑说明：IsolationForest 不需要标注攻击样本，只用正常流量训练，适合毕设数据难获取的情况。contamination 控制告警灵敏度，0.05 表示假设 5% 是异常，调大误报多，调小漏报多。特征要做归一化，否则包长这种大数值会主导距离。
4.3 规则和模型的融合策略
两路结果不能简单或起来，否则误报叠加。常见做法是分级：规则命中高危攻击直接告警并触发防御；模型异常但规则没命中，先记低危事件，连续多个窗口异常再升级。这样既保留规则的确定性，又用模型兜底未知威胁。
def fuse_detection(rule_hits, model_anomaly, recent_anomaly_count): if rule_hits: return {"level": "high", "types": rule_hits} if model_anomaly: recent_anomaly_count += 1 if recent_anomaly_count >= 3: return {"level": "medium", "types": ["anomaly"]} return {"level": "low", "types": ["anomaly"]} return {"level": "none", "types": []}逻辑说明：规则命中直接高危；模型异常累计三次才升中危，避免单次抖动误报。参数上，连续次数 3 和窗口长度相关，窗口 5 秒时相当于 15 秒内持续异常。这个计数要按源 IP 维护，不能全局共用一个。
5. 自动防御与可视化监控：从告警到动作再到面板
5.1 自动防御动作怎么设计才安全
自动防御是这套系统里最容易翻车的部分。直接调 iptables 封 IP 风险很大，万一误封网关或 DNS 会导致整网断掉。稳妥做法是：防御动作分级，低危只记录，中危限速，高危才封禁，且封禁有白名单和自动解封时间。
import subprocess import time WHITELIST = {"192.168.1.1", "8.8.8.8"} BAN_TIME = 300 # 秒 def block_ip(ip): if ip in WHITELIST: return False, "whitelisted" # 用 iptables 封禁，注意用 -I 插到前面 cmd = ["iptables", "-I", "INPUT", "-s", ip, "-j", "DROP"] subprocess.run(cmd, check=True) # 记录解封时间，由定时任务处理 schedule_unblock(ip, BAN_TIME) return True, "blocked" def schedule_unblock(ip, delay): def _unblock(): time.sleep(delay) subprocess.run(["iptables", "-D", "INPUT", "-s", ip, "-j", "DROP"]) from threading import Thread Thread(target=_unblock, daemon=True).start()逻辑说明：白名单先过滤，避免封禁关键设备；用 -I 插入规则到链首，确保生效；解封用后台线程延时执行。参数上，BAN_TIME 按场景定，演示可以 60 秒，真实环境建议 300 秒以上并加人工确认。Windows 下没有 iptables，可以用 netsh advfirewall 或直接在应用层拒绝该 IP 的连接。
注意：防御动作一定要有开关，config 里加 DEFENSE_ENABLED，调试阶段关掉，只记录不封禁，确认检测准确率后再打开。
5.2 可视化监控面板要展示什么
可视化监控不是把日志堆到页面上，要让人一眼看出当前风险。核心展示四块：实时流量速率曲线、攻击类型分布饼图、最近告警列表、被封禁 IP 列表。后端用 Flask-SocketIO 把检测事件推给前端，前端用 ECharts 渲染。
from flask import Flask, render_template from flask_socketio import SocketIO app = Flask(__name__) socketio = SocketIO(app, cors_allowed_origins="*") def push_event(event): """检测到事件后调用，推送到前端""" socketio.emit("detection", event) @app.route("/") def index(): return render_template("dashboard.html") if __name__ == "__main__": socketio.run(app, host="0.0.0.0", port=5000)逻辑说明：socketio.emit 把事件推给所有连接的浏览器，前端监听 detection 事件更新图表。参数上，cors_allowed_origins 演示时设 *，生产要收紧。端口 5000 按需改。前端 dashboard.html 里用 setInterval 或 socket.on 更新数据，流量曲线保留最近 60 个点即可，太多会卡。
5.3 数据存储与查询
告警和流量统计要落库，方便回溯。毕设场景用 SQLite 就够，表结构简单：events 表存时间、源 IP、目的 IP、攻击类型、级别、处理动作。写入用独立线程或批量提交，避免每条告警都开事务拖慢检测。
import sqlite3 import threading db_lock = threading.Lock() def init_db(path="ids.db"): conn = sqlite3.connect(path) conn.execute(""" CREATE TABLE IF NOT EXISTS events ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL, src_ip TEXT, dst_ip TEXT, attack_type TEXT, level TEXT, action TEXT ) """) conn.commit() conn.close() def save_event(evt, path="ids.db"): with db_lock: conn = sqlite3.connect(path) conn.execute( "INSERT INTO events (ts,src_ip,dst_ip,attack_type,level,action) VALUES (?,?,?,?,?,?)", (evt["ts"], evt["src_ip"], evt["dst_ip"], evt["attack_type"], evt["level"], evt["action"]) ) conn.commit() conn.close()逻辑说明：用锁保证多线程写 SQLite 不冲突，批量场景可以攒 100 条再提交。参数上，ts 用 Unix 时间戳，查询时转换。表加索引在 src_ip 和 ts 上，回溯特定 IP 时快很多。
6. 避坑与排查：这套系统最容易翻车的 5 个地方
6.1 抓不到包或只能抓到本机流量
现象：sniff 跑起来但队列里没数据，或者只看到自己机器的包。原因通常是网卡选错，或者交换机端口没做镜像，普通模式下网卡只能收到发给自己的包。解决：Linux 用 ip link 确认网卡名，抓包前用 tcpdump -i eth0 验证能看到流量；需要全量流量时，在交换机上配端口镜像，或者把系统部署在网关位置。Windows 下确认 Npcap 安装时勾了监控模式兼容。
6.2 检测误报把正常业务封了
现象：防御开启后，正常用户访问被阻断，业务方找上门。原因一般是阈值太激进，或者白名单没覆盖关键 IP。解决：先跑一周观察模式，只记录不封禁，统计规则命中情况，把高频正常 IP 加白名单，阈值按基线调高。封禁时间从短开始，确认准确后再延长。
6.3 多线程下状态字典数据错乱
现象：端口扫描检测时有时无，或者计数明显不对。原因是多个消费者线程同时读写同一个 state 字典，set 的 add 和 len 不是原子操作。解决：给每个源 IP 的状态加锁，或者把同一源 IP 的包哈希到固定线程处理，避免并发写。简单做法是用 threading.Lock 包住状态更新。
6.4 模型推理拖慢整条链路
现象：开启模型检测后丢包率飙升，面板延迟大。原因是模型推理在消费者线程里同步跑，单次几十毫秒，流量大时队列积压。解决：模型推理单独放一个进程，用 multiprocessing.Queue 传特征；或者降低推理频率，每 N 个窗口跑一次，不用每个包都过模型。
6.5 可视化面板数据不更新
现象：后端日志有检测事件，但前端图表不动。原因通常是 SocketIO 版本和前端客户端不匹配，或者 emit 在请求上下文外调用报错。解决：确认 flask-socketio 和前端 socket.io.js 版本兼容，emit 用 socketio.emit 而不是 flask 的 emit；浏览器控制台看 WebSocket 是否连接成功，跨域问题加 cors_allowed_origins。
7. 进阶技巧：把检测准确率再往上提一档
规则加模型的组合跑通之后，想再提准确率，重点不在换更复杂的模型，而在特征工程和反馈闭环。我一般会做三件事。第一，把窗口特征从「计数」扩展到「分布」，比如目的端口熵、源 IP 熵、包长方差，扫描和正常访问在这些分布上差异明显。第二，加一个误报反馈入口，面板上每条告警可以标记「误报」，标记后自动把对应特征加入负样本，每周重训一次模型，contamination 根据误报率微调。第三，对高频攻击类型单独训二分类模型，比如 SYN Flood 和端口扫描各一个，比通用异常检测准得多。
验证方法上，不要只看准确率，要看漏报和误报的绝对数量。毕设演示可以用 Scapy 构造攻击包自测：
from scapy.all import IP, TCP, send import random # 模拟端口扫描，向目标发不同目的端口的 SYN target = "192.168.1.100" for port in random.sample(range(1, 1024), 100): pkt = IP(dst=target)/TCP(dport=port, flags="S") send(pkt, verbose=0)跑完看系统有没有在预期时间内告警、有没有触发防御、面板有没有更新。参数上，发包速率别太快，否则本机网卡先扛不住。自测通过后，再用真实流量跑观察模式对比。
几个我踩过的习惯：config 里永远留一个 DEBUG_MODE，打开后所有防御动作只打印不执行；每次改阈值前先备份当前配置，出问题能回滚；日志按天切分，别让单个文件涨到几个 G。这套系统不难，难的是让检测准、防御稳、面板实时，三者平衡好，毕设演示和实际用都不会太难看。希望帮到你。
本文还有配套的精品资源，点击获取
