企业官网建设流程全解析

简介：面向Python网络方向的毕业设计或课程设计场景，这份资源提供一套可运行的网络入侵检测与防御系统，解决从实时流量捕获、异常行为识别到自动拦截阻断与可视化监控的完整实现问题。系统基于Python语言、Flask框架与Scapy抓包库构建，具备实时流量分析、攻击检测、自动防御、告警响应及Web管理界面，可帮助网络空间安全或计算机相关专业学生快速搭建毕设原型，也便于理解入侵检测系统的典型工作流程。

资源压缩包共包含38个文件，整体大小仅92KB，以14个Python源码文件和9个编译后的pyc文件为主体，另含4个HTML页面、3个JavaScript脚本、JSON配置文件、Dockerfile与docker-compose部署编排文件，以及安装脚本和依赖清单，覆盖后端逻辑、前端展示、容器化部署、环境安装等多个环节，目录结构清晰，便于按模块阅读和定位修改。

目前已有172人学习/下载，适合需要快速完成毕业设计、课程设计或安全类实验的读者。随包附带的详细运行指南和项目说明，能有效降低环境配置与启动门槛，使使用者尽快跑通系统，并在此基础上扩展检测规则、调整防御策略或集成到现有网络监控方案中。

1. 基于Python的网络入侵检测与防御系统是什么：把实时流量分析与自动防御做成毕设

“基于Python的网络入侵检测与防御系统”这个标题看着像课程作业，但真正做过的同学会告诉你，难点不在“检测”两个字上。抓包要权限，解析要特征，封禁要联动系统防火墙，可视化还要把告警实时推到前端。这套系统能落地的关键，是先跑通一条从网卡到告警再到封禁的数据链路，而不是一上来就训练模型。

我见过不少毕设动辄上机器学习，结果训练数据全是从网上找的离线PCAP，上线后连本机流量都看得不全。这个项目老老实实走规则检测和阈值统计，用Scapy做实时流量分析，把SYN扫描、高频请求这类行为抓出来，再调用iptables做自动防御，最后用Flask-SocketIO把结果可视化。技术栈干净，每一层都能讲清楚原理，适合网络方向毕设，也适合准备安全开发岗位的在校生。

如果你手上只有一台普通笔记本或虚拟机，这套方案同样能跑。下面按“采集、检测、防御、监控、调试、验证”的顺序，把我的做法完整过一遍。

2. 实时流量分析模块：用Scapy把抓包、特征解析与数据落库跑通

整个系统里最先要打通的是数据链路，而这条链路的起点不是检测算法，是抓包。很多毕设翻车的共同点是先写检测代码，再回头补采集，结果流量根本进不来，阈值全部空转。所以第2章先把实时流量分析链路走通：从网卡到特征字典，再到数据库。

2.1 解析方案选型：为什么是Scapy，而不是dpkt、pyshark或Snort

实时流量分析首先要选一个抓包解析方案。常见选项是Scapy、dpkt、pyshark，外加Snort、Suricata这类成熟引擎。Snort功能确实完整，但对毕设来说太重，规则语法和内部架构要解释半天，容易把自己绕进去；dpkt解析效率最高，但要把以太网、IP、TCP逐层手工解开，代码量直接翻倍；pyshark解析起来方便，可前提是机器上装了tshark，部署环境多一个外部依赖。

我一般选Scapy。它底层走libpcap抓包，能力有保障，上层提供sniff()回调和现成的包对象，可以直接提取五元组、TCP标志位、载荷长度。实验室环境的包速率不大，Scapy的解析性能完全够用，而且代码量小，答辩时每一行都能讲清楚。

判断标准很简单：包速率不高，但解析字段要灵活，回调里拿到包就能转成特征字典。Scapy在这两条上恰好都占。

2.2 抓包与特征提取代码：一个队列缓冲实时流量

先看采集部分的完整代码。这个模块只做一件事：把网卡上的IP包转成扁平字典，塞进队列，不在这里写检测逻辑。

from queue import Queue from scapy.all import sniff, IP, TCP, UDP, Raw # 抓包线程和生产队列：只放特征，不放整个包对象 feature_queue = Queue(maxsize=2000) def packet_handler(pkt): # 只要IP层，非IP流量直接丢弃 if not pkt.haslayer(IP): return ip = pkt[IP] item = { "ts": pkt.time, "src": ip.src, "dst": ip.dst, "proto": ip.proto, } if pkt.haslayer(TCP): tcp = pkt[TCP] item["sport"] = tcp.sport item["dport"] = tcp.dport item["syn"] = bool(tcp.flags.S) item["ack"] = bool(tcp.flags.A) item["payload_len"] = len(tcp.payload) elif pkt.haslayer(UDP): udp = pkt[UDP] item["sport"] = udp.sport item["dport"] = udp.dport item["payload_len"] = len(udp.payload) if pkt.haslayer(Raw): # 只保留载荷前64字节的hex，用来回溯攻击特征 item["payload_head"] = pkt[Raw].load[:64].hex() try: feature_queue.put_nowait(item) except Exception: # 队列满直接丢包，优先保实时性 pass # 监听网卡，只放行IP流量；store=False表示不留原始包 sniff(iface="ens33", prn=packet_handler, store=False, filter="ip")

几个关键参数要注意。iface必须填实际网卡名，不填的话Scapy会走默认网卡，同一台机器有多个网卡时容易抓到不想要的包；store=False表示不在内存里堆积原始包，这个参数必须带上，否则跑一段时间内存就满了；filter="ip"是BPF语法，只放行IP包，把ARP等链路层噪音挡在外面。

packet_handler里提取的特征是后续检测模块直接消费的字段。tcp.flags.S和tcp.flags.A分别判断SYN、ACK标志位，这是识别端口扫描和TCP握手状态的原材料。载荷前64字节转hex，是为了给告警留证据，又不至于把整包打印出来刷屏。

这里还有一个重要的实战细节：如果这台机器同时用来SSH远程管理，建议把过滤器写成filter="ip and not tcp port 22"，把管理流量排除在采集范围外，避免检测模块把你自己的SSH会话也当成异常流量。这个动作能省掉后面大半的误封麻烦。

2.3 数据落库：用SQLite批量写入替代逐条insert

流量解析成字典之后，不能每条都立刻写数据库，否则高频流量下SQLite的commit会成为瓶颈。常见做法是采集线程只负责投递，单独开一个线程做批量落库。我用SQLite，ts字段存unixepoch秒，后面按时间聚合查询非常方便。

import sqlite3 import threading import time conn = sqlite3.connect("traffic.db", check_same_thread=False) conn.execute(""" CREATE TABLE IF NOT EXISTS packet_features ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL, src TEXT, dst TEXT, proto INTEGER, sport INTEGER, dport INTEGER, flags TEXT, payload_len INTEGER, payload_head TEXT )""") BUFFER = [] BUFFER_LOCK = threading.Lock() FLUSH_INTERVAL = 5 # 每5秒批量写一次 def append_buffer(item): with BUFFER_LOCK: BUFFER.append(item) def flush_worker(): while True: time.sleep(FLUSH_INTERVAL) with BUFFER_LOCK: if not BUFFER: continue rows = [ (i["ts"], i["src"], i["dst"], i["proto"], i.get("sport"), i.get("dport"), "", i.get("payload_len"), i.get("payload_head")) for i in BUFFER ] BUFFER.clear() # 注意：clear在锁内，executemany在锁外 conn.executemany( "INSERT INTO packet_features " "(ts, src, dst, proto, sport, dport, flags, payload_len, payload_head) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", rows ) conn.commit()

check_same_thread=False让采集线程和落库线程可以共用同一个连接，前提是同一时刻只有一个线程真正执行execute和commit。我这里的写法是把BUFFER的读写都放进锁里，而数据库操作在锁外，避免长时间持锁卡住抓包线程。

FLUSH_INTERVAL设成5秒，意思是最多丢5秒的流量数据。实验室环境这个值足够。如果追求更低的时延，改成1秒也可以，但磁盘写入会明显频繁，整机性能会往下掉。

2.4 自检方法：用一次ping验证整条链路

写检测模块之前，先跑一次最小链路验证，这一步能筛掉八成后续问题。流程是这样的：

# 终端1：启动采集脚本 sudo python3 collector.py # 终端2：发3个ICMP包 ping -c 3 192.168.1.1 # 终端3：查看库里有没有记录 sqlite3 traffic.db "SELECT src, dst, proto FROM packet_features ORDER BY id DESC LIMIT 5;"

如果查询结果里出现了本机IP и目标IP，说明抓包、解析、落库这条实时流量分析链路是通的。ping包属于ICMP协议，proto字段会显示1，能查ICMP说明BPF过滤器没把非TCP流量误过滤掉。确认这一步，后面检测模块才有数据可吃。

3. 攻击检测与自动防御：把SYN扫描、高频请求与iptables封禁做成闭环

流量采下来之后，攻击检测模块要解决的问题是“哪条流量可疑”。这块不要贪多，做两个行为检测就足够撑起整个系统：端口扫描/SYN洪泛，以及同一源IP高频新建TCP连接。防御动作做到自动封禁IP、到期自动解封，并把告警推送出去。

3.1 检测规则设计：先想清楚要抓什么行为

写检测代码前，先定义异常行为。我见过太多代码把阈值写得像随机数，自己都解释不清为什么是20而不是50。检测规则的三要素是时间窗口、阈值、白名单。

端口扫描和SYN洪泛有一个共同特征：短时间内同一个源IP发出大量SYN包，但没有完成TCP握手。正常浏览器访问网页也会建TCP连接，但密度完全达不到攻击水平。高频请求则反过来，表现为同一个源IP在几秒内完成大量TCP握手，这种更像被滥用的代理池或者刷接口脚本。设计规则时把这两类行为分开，后续调参才不糊涂。

3.2 滑动窗口计数器：识别SYN洪泛与高频建连

检测模块的核心代码我推荐用滑动窗口列表，而不是一个简单计数器。原因是计数器只统计总数，时间窗口一滚动，旧数据还留在计数里，误报率会随运行时长不断升高。看下面的实现：

from collections import defaultdict SYN_WINDOW = 10 # 统计窗口：10秒 SYN_THRESHOLD = 20 # 窗口内SYN包个数阈值 syn_memo = defaultdict(list) def detect_syn_flood(item): # 只统计纯SYN包，带ACK的TCP握手包不算 if not (item.get("syn") and not item.get("ack")): return None src = item["src"] syn_memo[src].append(item["ts"]) # 只保留当前窗口内的记录 syn_memo[src] = [t for t in syn_memo[src] if item["ts"] - t <= SYN_WINDOW] if len(syn_memo[src]) >= SYN_THRESHOLD: # 命中后清掉该源IP的窗口，避免同一波攻击刷屏告警 syn_memo[src].clear() return { "alert": "syn_flood", "src": src, "count": SYN_THRESHOLD + 1, "window": SYN_WINDOW, } return None

这段代码有两个容易被忽视的细节。第一，判断条件是syn=True且ack=False，只统计扫描或洪泛特征的SYN包，不把正常握手的第二个包算进来。第二，命中后立即clear()清楚这个源IP的窗口，否则10秒内后续每个SYN包都会再触发一次告警，告警数量会被同一波攻击刷爆。

高频连接检测的逻辑类似，但要换成对ACK包的统计。一次完整TCP连接建立的标志是收到SYN+ACK，所以判断条件是syn=True且ack=True：

CONN_INTERVAL = 5 # 统计窗口：5秒 CONN_THRESHOLD = 60 # 窗口内新建连接阈值 conn_memo = defaultdict(list) def detect_high_conn(item): # 只统计TCP握手的第二个包：SYN+ACK if not (item.get("syn") and item.get("ack")): return None src = item["src"] conn_memo[src].append(item["ts"]) recent = [t for t in conn_memo[src] if item["ts"] - t <= CONN_INTERVAL] conn_memo[src] = recent if len(recent) >= CONN_THRESHOLD: # 高频建连不清空窗口，因为这是连续行为 return { "alert": "high_conn", "src": src, "count": len(recent), "interval": CONN_INTERVAL, } return None

高频建连我没有清空窗口，因为这类攻击往往是持续性的，清空会导致下一秒又重新计数，告警节奏反而不对。具体阈值怎么调，第5章会展开。

3.3 自动防御：iptables封禁、白名单与自动解封

检测命中之后必须接上防御动作，否则检测模块只是纸上谈兵。我在Linux下用的是iptables封禁源IP，核心操作就两条：-I INPUT在规则表最前面插入一条DROP规则，-D INPUT删除这条规则实现解封。

import subprocess import time import threading BAN_SECONDS = 300 # 封禁时长：300秒，演示时可调成120 WHITELIST = {"127.0.0.1"} # 本机和内网管理地址必须放行 ban_records = {} # ip -> 解封时间戳 def do_ban(ip, alert_type): if ip in WHITELIST: return False # 已经封禁且未到期，不重复操作 if ip in ban_records and ban_records[ip] > time.time(): return False result = subprocess.run( ["iptables", "-I", "INPUT", "-s", ip, "-j", "DROP"], capture_output=True, text=True, check=False, ) if result.returncode == 0: ban_records[ip] = time.time() + BAN_SECONDS save_alert(ip, alert_type, int(ban_records[ip])) return True return False def unban_loop(): while True: time.sleep(10) now = time.time() for ip, expire in list(ban_records.items()): if now >= expire: subprocess.run( ["iptables", "-D", "INPUT", "-s", ip, "-j", "DROP"], capture_output=True, check=False, ) del ban_records[ip]

do_ban里用了check=False并捕获stderr，而不是直接让子进程抛异常。因为iptables在权限不足时返回非0退出码，如果你用check=True，脚本会当场崩溃，你只会在Traceback里看到一行Permission denied，连日志都没有。现在这个写法可以拿到报错信息，方便排查。

unban_loop每10秒扫一次封禁记录，到期的IP自动解封，这样即使写错了也不会造成永久封禁。这里要特别强调白名单，我的习惯是把127.0.0.1写死，再把本机所在网段的网关地址加进去，否则误封会把你自己也关在门外。具体踩坑记录在第5章。

3.4 告警去重与削峰：不要让抓包回调去做防御

检测和防御动作不能直接写在packet_handler里。抓包回调是高频热点，一个扫描过程可能触发几十上百个SYN包，如果每个包都去执行一次subprocess调iptables，抓包线程直接卡死，监控页面也会跟着假死。正确的做法是生产消费模型：

from queue import Queue alert_queue = Queue() def detector_loop(): while True: item = feature_queue.get() alert = detect_syn_flood(item) or detect_high_conn(item) if alert: alert_queue.put(alert) def defense_worker(): while True: alert = alert_queue.get() if do_ban(alert["src"], alert["alert"]): push_alert(alert)

detector_loop从流量队列里取特征做检测，命中后放进告警队列。defense_worker是唯一负责执行封禁和解封的线程，它拿到告警后调用do_ban，只有第一次真正执行封禁时才推送告警。这样同一波攻击不会重复刷屏，前端监控也不会被海量重复消息淹没。

4. 可视化监控：用Flask-SocketIO把检测结果实时推到Dashboard

检测和防御都接上了，最后把结果用Web页面展示出来。可视化监控这一层做得太重会被说是凑字数，完全不做又体现不出“监控”。合理的目标是一张实时告警列表、一个最近30分钟攻击趋势图，加上一张历史告警表。

4.1 监控数据链路：三个线程各管一件事

整个系统的运行结构是三条线程。抓包线程负责采集流量放进feature_queue；检测线程消费特征，产生告警后放进alert_queue并执行封禁；Web服务线程负责把告警推给浏览器，并提供历史查询接口。三个线程之间靠队列解耦，任何一环阻塞都不会拖垮其他环节。

如果只用一个进程跑全流程，Web服务要选一个支持异步的框架。普通Flask开发服务器在处理长轮询时表现并不好，所以我选了Flask-SocketIO，用WebSocket把新告警直接推到前端。实时流量分析的结果能不能“实时”看到，关键就在这一层。

4.2 Flask-SocketIO服务端：新告警主动推送

下面这段是监控服务端的骨架。它只做两件事：客户端连上来时发一条确认消息；告警产生时广播到所有已连接页面。

from flask import Flask, render_template, jsonify from flask_socketio import SocketIO, emit app = Flask(__name__) app.config["SECRET_KEY"] = "monitor" # async_mode=threading 是最稳的运维方式 socketio = SocketIO(app, async_mode="threading") MONITOR_NS = "/monitor" @socketio.on("connect", namespace=MONITOR_NS) def handle_connect(): emit("pong", {"msg": "connected"}, namespace=MONITOR_NS) def push_alert(alert): socketio.emit("alert", alert, namespace=MONITOR_NS, broadcast=True)

async_mode有三个可选值：threading、eventlet、gevent。我推荐直接用threading，它不依赖额外安装高性能并发库，兼容性最好。eventlet和gevent在Windows上经常装出问题，折腾半天只是为了让性能更好，但毕设场景根本用不到那个量级的并发。

socketio.emit的broadcast=True表示推送给所有在/monitor命名空间下的客户端。命名空间的作用是隔离监控页面和其他页面，如果以后加一个配置管理页，两边的WebSocket事件不会互相干扰。

4.3 前端接收事件与历史聚合查询

前端的核心是接收“alert”事件并即时渲染。页面用ECharts画趋势图，数据来自SQLite按分钟聚合的查询：

const socket = io("/monitor"); socket.on("alert", function (data) { const li = document.createElement("li"); li.textContent = `[${data.timestamp}] ${data.src} -> ${data.dst} | ${data.alert}`; document.getElementById("alert-list").prepend(li); refreshChart(); }); function refreshChart() { fetch("api/trend?minutes=30") .then((resp) => resp.json()) .then((rows) => myChart.setOption({ xAxis: { data: rows.map((r) => r.minute) }, series: [{ data: rows.map((r) => r.cnt) }], })); }

对应后端聚合查询的SQL语句是：

SELECT strftime('%H:%M', datetime(ts, 'unixepoch', '+8 hours')) AS minute, COUNT(*) AS cnt FROM alerts WHERE ts >= strftime('%s', 'now') - 1800 GROUP BY minute ORDER BY minute;

+8 hours这个写法是固定的东八区偏移，如果你在别的时区就改成对应偏移。更通用的写法是datetime(ts, 'unixepoch', 'localtime')，让SQLite按系统本地时区转换。WHERE ts >= strftime('%s','now') - 1800表示只取最近30分钟的告警，防止图表数据量越积越大导致页面卡顿。

4.4 线程调度经验：抓包线程不能和Web服务挤在一起

一个必须提前避开的坑是：不要把sniff()写在Flask的启动代码里，也不要放在和socketio.run()同一个线程里。sniff()是阻塞调用，一旦进入监听循环，后面的代码就不执行了，Web服务根本起不来。更隐蔽的问题是Scapy的回调函数运行在Scapy自己的线程里，你在回调里调socketio.emit()没问题，但如果直接在回调里执行SQLite写库和iptables命令，就会把抓包线程拖死。

正确做法是保持第3章的队列模型，Web服务只消费告警队列和查询数据库，不碰抓包逻辑。整个系统的实时流量分析、攻击检测、自动防御、可视化监控四个模块，通过两条队列串成一条完整的流水线。

5. 调试与避坑：5条踩坑记录和3个必调参数

这个标题听起来简单，真正跑起来后问题主要集中在环境和权限。拿到源码包后别急着安装依赖，先把第2章的采集脚本跑通，否则后面所有检测和防御代码都处于没有数据可吃的状态。这一章把我的血泪经验按售后问题记录的方式写出来，每一条都是现象、原因、解决三件套。

5.1 抓包环境和权限类问题

踩坑1：虚拟机里抓不到任何外部流量

• 现象：宿主机访问Web应用，虚拟机的采集脚本一包未收；但ping虚拟机的网关地址却能收到包。
• 原因：默认NAT模式下，虚拟网卡只能看到自己的虚拟网络流量。外部流量进不了这块虚拟网卡，自然抓不到。
• 解决：在虚拟机软件里把网卡改成桥接模式，让虚拟机直接拿到物理网段的地址；或者把演示场景收敛到局域网内部，用同一网段的两台机器互发流量。跑采集脚本必须用sudo python3 collector.py，普通用户没有libpcap的权限。

踩坑2：Scapy报“assert _ip is not None”或“kernel arp filter failed”

• 现象：启动sniff()后秒退，控制台输出assert错误；BPF过滤器不生效。
• 原因：libpcap版本与当前内核驱动不匹配，或者iface参数填成了不存在的网卡名。
• 解决：先执行sudo python3 -c "from scapy.all import conf; print(conf.ifaces)"看有哪些接口名，把iface参数改成实际存在的名字。如果问题依旧，重新安装libpcap-dev库，再升级scapy到新版本，再次启动前用sudo ldconfig刷新动态链接库。

5.2 检测与自动防御联动类问题

踩坑3：iptables把SSH管理会话误封，自己把自己踢下线

• 现象：检测到高频连接告警，随后本机SSH连接中断，再也登录不上去。
• 原因：检测规则把所有访问本机的流量都算进去，把自己正在使用的管理来源IP也封禁了。do_ban()没有先查白名单。
• 解决：把管理来源IP放进WHITELIST，并且在启动防御功能前先执行放行命令：iptables -I INPUT -p tcp --dport 22 -j ACCEPT，把它放在DROP规则之前。这样即使误封，SSH管理通道也始终可用。

踩坑4：iptables封禁规则在重启后全部消失

• 现象：重启服务器后告警还在，但攻击源IP已经能访问了，检查发现iptables -L里没有任何DROP规则。
• 原因：iptables规则只存在于运行时内存，重启即清空。没有做规则持久化。
• 解决：用iptables-save > /etc/iptables/rules.v4导出规则，在开机脚本里执行iptables-restore < /etc/iptables/rules.v4恢复。如果只是毕设演示，不需要坚持重启持久化，但要提前想清楚演示时不能重启服务器，或者让unban线程在每次启动时先清一遍旧规则。

踩坑5：SYN阈值设太低，正常网页浏览被误判为攻击

• 现象：打开一个门户首页，几十秒前端弹出告警，IP被自动封禁。
• 原因：一个大型页面会同时发起几十个TCP连接，SYN_THRESHOLD设成10，自然会把正常浏览当攻击。
• 解决：把SYN_THRESHOLD提到30以上，CONN_THRESHOLD提到60以上，窗口时间缩短到5秒。阈值调整没有公式可用，唯一可靠的方法是录制一段正常访问流量做测试，再看告警日志里有没有误报。

5.3 三个必调参数

下面这张表是我跑这个系统时认为最需要提前调好、也最影响演示效果的一组参数。参数值不是越小越灵敏，而是要和你的演示网络环境匹配。

参数改完不要直接跑主程序，先跑一段小流量采集，把告警日志打出来看一眼。这类阈值是纯经验值，不试不知道。

6. 验证与进阶：用自测流量逼出告警，再从计数特征升到状态机特征

做完上面五章，系统已经能跑，但你要在答辩前证明它真的“检测”到了攻击，而不是纸上谈兵。最直接的做法是构造一段SYN扫描流量去打自己，看系统能不能在几秒内产生告警并执行封禁。

from scapy.all import IP, TCP, send # 模拟一个扫描源，发30个不同目标端口的SYN包 pkt = IP(src="10.0.0.66", dst="192.168.1.20", ttl=64) / TCP(flags="S") send(pkt, count=30, inter=0.01)

这段代码用Scapy构造30个SYN包，目标是被检测主机，源IP伪造为10.0.0.66。跑完之后去监控页面的告警列表和iptables规则里确认，10.0.0.66应该被拉黑。这个验证有两个作用：一是确认检测逻辑没写错，二是提前给演示录制一段可控的自测视频。

再进一步，单一计数阈值误报率高，可以往状态机方向升级。现在的检测只看“SYN包个数”，更精细的规则应该把TCP三次握手状态考虑进来，比如统计“发了SYN但没有收到SYN+ACK”的组合，或者统计连续N个SYN包都发往不同端口。把特征从单纯的计数扩展到“目标端口去重数、包长分布、重传间隔”后，攻击检测会可靠很多。我现在的习惯是每加一条规则，先用自动脚本造十条正常流量和十条攻击流量去对照跑，让规则基于数据说话，而不是急着上模型。希望帮到你。

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