企业官网建设流程全解析

简介：这份资源是面向高校计算机、网络安全相关专业学生及课程设计学习者的完整项目源码包，围绕网络入侵检测与防御系统展开，可用于毕业设计、课程设计或安全方向练手。项目基于Python构建，后端采用Flask、Flask-SocketIO与Scapy实现实时流量捕获与分析，前端借助HTML5、Bootstrap与Chart.js完成可视化监控，数据由MongoDB存储，并通过Docker Compose实现容器化部署，覆盖流量检测、入侵防御、告警响应与网络监控等核心模块。压缩包共38个文件，约92KB，包含14个py源码、9个pyc编译文件、4个html页面、3个js脚本及json、yml、Dockerfile、sh等配置与部署文件，结构清晰，便于按模块阅读与二次开发。目前已有172人学习关注。资源附带详细运行指南，读者可据此快速搭建环境、理解检测与防御逻辑，并参考Web界面与日志设计完成自己的安全类项目。

1. 从一份毕设源码说起：Python 网络入侵检测与防御系统到底在做什么

很多人第一次接触「网络入侵检测与防御系统」是在毕设选题表上，看到「Python 实现、实时流量分析、攻击检测、自动防御、可视化监控」这几个词，觉得高大上，真拿到源码却发现跑不起来：抓包没权限、依赖装不上、Web 面板一片空白。这篇笔记就围绕这套典型架构，把从环境搭建到实时流量分析、攻击检测规则、自动防御联动、可视化监控的完整链路拆开讲清楚，让你既能照着复现，也能看懂每一层在干什么、参数怎么调、哪里最容易翻车。

它解决的核心问题是：在单机或小型内网里，用 Python 把「抓包 → 特征提取 → 规则/阈值判定 → 触发防御动作 → 前端展示」串成一条实时流水线。适合做毕设的学生、想入门安全开发的后端工程师，以及需要给内网加一层轻量监控的运维。下面按落地顺序推进，先讲清抓包与流量分析，再讲检测、防御和可视化，最后给一套可验证的进阶技巧。

2. 实时流量分析：用 Scapy 抓包并提取五元组特征

实时流量分析是整个系统的地基。没有稳定的数据源，后面的攻击检测和可视化都是空中楼阁。这一章先把抓包环境跑通，再把原始数据包变成可判定的结构化特征。

2.1 抓包选型：Scapy 与原始套接字的取舍

Python 抓包常见三条路：Scapy、pcapy/winpcap 绑定、以及直接读/proc/net或调用系统命令。做毕设最稳的是 Scapy，原因是它纯 Python、跨平台、能直接解析到 TCP/UDP/ICMP 各层字段，改起来不用碰 C 扩展。代价是性能一般，千兆满速下会丢包，但对毕设演示和中小流量监控完全够用。

抓包前必须解决权限问题。Linux 下普通用户没有原始套接字权限，直接跑会报PermissionError。常见做法是给 Python 解释器加CAP_NET_RAW能力，或者干脆用 root 跑；Windows 下则需要装 Npcap 并勾选 WinPcap 兼容模式。这一步不解决，后面所有代码都是白搭。

# Linux：给 python 解释器授予抓包能力，避免每次 sudo sudo setcap cap_net_raw,cap_net_admin=eip $(readlink -f $(which python3)) # 验证是否生效 getcap $(readlink -f $(which python3))

setcap给的是二进制能力，eip分别代表有效、可继承、允许提升。执行后普通用户即可抓包。注意：如果你用的是虚拟环境，which python3要指向虚拟环境里的解释器，否则能力加到了系统 Python 上，跑起来仍然报权限错误。

2.2 用 Scapy 写一个最小可用的抓包与特征提取脚本

下面这段代码是整套系统的数据入口：持续抓包，按五元组（源 IP、目的 IP、源端口、目的端口、协议）聚合，并统计单位时间内的包数和字节数。这两个统计量是后面检测 SYN Flood、端口扫描的基础。

from scapy.all import sniff, IP, TCP, UDP from collections import defaultdict import time # 五元组 -> [包数, 字节数, 首次时间, 末次时间] flow_stats = defaultdict(lambda: [0, 0, 0.0, 0.0]) WINDOW = 10 # 统计窗口，单位秒 def extract(pkt): if IP not in pkt: return proto = "TCP" if TCP in pkt else ("UDP" if UDP in pkt else "OTHER") sport = pkt[TCP].sport if TCP in pkt else (pkt[UDP].sport if UDP in pkt else 0) dport = pkt[TCP].dport if TCP in pkt else (pkt[UDP].dport if UDP in pkt else 0) key = (pkt[IP].src, pkt[IP].dst, sport, dport, proto) now = time.time() rec = flow_stats[key] if rec[0] == 0: rec[2] = now rec[0] += 1 rec[1] += len(pkt) rec[3] = now def report(): now = time.time() for key, rec in list(flow_stats.items()): if now - rec[3] > WINDOW: # 窗口内无新包，输出并清理 pps = rec[0] / max(rec[3] - rec[2], 1e-6) print(f"{key} 包数={rec[0]} 字节={rec[1]} pps={pps:.1f}") del flow_stats[key] if __name__ == "__main__": # store=0 表示不把包留在内存，避免长时间运行 OOM sniff(prn=extract, store=0, filter="ip")

逻辑说明：sniff的prn回调每来一个包就执行一次extract，把包归入对应五元组。store=0是关键参数，默认 Scapy 会把所有包存进内存列表，跑几分钟就吃满内存，这是新手最常见的翻车点。filter="ip"用 BPF 语法在底层过滤，只把 IP 包交给 Python，能显著降低 CPU 占用。

参数说明：WINDOW决定统计粒度，设太小会频繁输出、噪声大，设太大则攻击响应迟钝，一般 5～10 秒。pps（每秒包数）是后续阈值判定的核心指标，正常业务流量通常个位数到几百，SYN Flood 能轻松上千。

2.3 把特征落到可查询的结构里

抓到的数据要能被检测模块和前端同时消费，常见做法是写进一个带时间戳的队列或轻量数据库。毕设里我一般用queue.Queue做进程内缓冲，再用 SQLite 落盘历史记录，前端轮询最近 N 条。这样检测线程和 Web 线程解耦，不会因为前端卡顿拖慢抓包。

import queue, sqlite3, threading pkt_queue = queue.Queue(maxsize=10000) # 有界队列，防止内存无限增长 def init_db(path="ids.db"): conn = sqlite3.connect(path) conn.execute("""CREATE TABLE IF NOT EXISTS flows( ts REAL, src TEXT, dst TEXT, sport INT, dport INT, proto TEXT, pkts INT, bytes INT)""") conn.commit() return conn

maxsize必须设，否则生产快于消费时队列会撑爆内存。消费端用get(timeout=1)配合queue.Empty异常，避免线程永久阻塞。SQLite 写入建议批量提交，每条都 commit 会让磁盘 IO 成为瓶颈。

3. 攻击检测：规则引擎与阈值判定怎么落地

有了流量特征，下一步是判定「这是不是攻击」。这一章讲三类最常见攻击的检测逻辑：SYN Flood、端口扫描、ICMP 洪水，并给出可调参数和误报控制思路。

3.1 三类攻击的特征与判定条件

不同攻击在流量特征上留下的痕迹不一样，硬套一个阈值必然误报。下面这张表是我实际调参后总结的判定条件，可以直接作为规则引擎的初始配置。

判定条件要组合多个维度，单看 pps 太粗糙。比如 SYN Flood 必须同时满足「速率高」和「SYN 占比高」，否则正常的高并发业务也会被拦。端口扫描则要看「目的端口去重数量」，而不是总包数。

3.2 规则引擎的代码实现

下面是一个可扩展的规则判定函数，输入是聚合后的流特征，输出是命中的攻击类型。用字典注册规则，新增攻击类型只需加一个函数，不用改主流程。

from collections import defaultdict # 记录每个源 IP 在窗口内访问过的目的端口集合 port_scan_tracker = defaultdict(set) def detect_syn_flood(flow, syn_ratio, pps): # 速率高 + SYN 占比高，两个条件同时满足才判定 return pps > 500 and syn_ratio > 0.8 def detect_port_scan(src_ip, dport, window=10): port_scan_tracker[src_ip].add(dport) # 简化版：超过阈值即判定，实际应带时间窗口清理 return len(port_scan_tracker[src_ip]) > 50 def detect_icmp_flood(flow, pps): return flow[4] == "OTHER" and pps > 200 RULES = { "SYN_FLOOD": detect_syn_flood, "PORT_SCAN": detect_port_scan, "ICMP_FLOOD": detect_icmp_flood, }

逻辑说明：detect_syn_flood接收速率和 SYN 占比两个参数，避免单一指标误判。port_scan_tracker用字典记录每个源 IP 访问过的端口，实际生产里要加时间窗口和定期清理，否则字典会无限增长——这是内存泄漏的经典来源。RULES字典让规则可插拔，检测主循环只需遍历它。

参数说明：syn_ratio需要在抓包层额外统计 SYN 标志位数量，不能只靠五元组。window参数控制端口扫描的观察周期，设太短会漏判慢速扫描，设太长会误伤正常的多端口访问。

3.3 误报控制：白名单与阈值自适应

规则引擎上线后最大的敌人是误报。血泪经验是：一定要有白名单机制，把网关、监控服务器、已知压测机的 IP 排除。其次阈值不要写死，可以按基线动态调整——统计过去一小时的正常 pps 均值，超过均值 5 倍才告警。

WHITELIST = {"192.168.1.1", "192.168.1.100"} def is_whitelisted(src_ip): return src_ip in WHITELIST # 基线自适应：维护滑动均值 from collections import deque baseline = deque(maxlen=360) # 一小时，每 10 秒一个点 def adaptive_threshold(current_pps): baseline.append(current_pps) avg = sum(baseline) / len(baseline) return max(avg * 5, 100) # 至少 100，避免基线过低导致误报

maxlen让 deque 自动淘汰旧数据，不用手动清理。max(avg*5, 100)里的下限很重要：凌晨流量接近零时，avg*5可能只有个位数，任何正常访问都会触发告警。这个下限值要根据你的网络规模调，内网一般 100 起步。

4. 自动防御：从检测到封禁的联动链路

检测出攻击只是第一步，自动防御才是「防御系统」区别于「检测系统」的地方。这一章讲怎么把检测结果变成实际的封禁动作，以及怎么保证封禁本身不出事。

4.1 防御动作的三种实现方式

自动防御常见三种手段：调用系统防火墙（iptables/firewalld）、修改应用层黑名单、发送告警通知。毕设里最直观的是调 iptables 直接封 IP，效果立竿见影，但风险也最大——封错了可能把自己关在门外。

# 封禁一个源 IP，限制其新建连接 sudo iptables -I INPUT -s 192.168.1.200 -j DROP # 查看当前封禁列表 sudo iptables -L INPUT -n --line-numbers # 解封（按行号删除） sudo iptables -D INPUT 1

-I是插入到链首，优先级最高；-D按规则或行号删除。注意：iptables 规则重启后会丢失，生产环境要配合iptables-persistent或自己写恢复脚本。毕设演示时建议加一个「封禁时长」机制，到期自动解封，避免演示完忘了清理。

4.2 用 Python 封装防御动作并加安全阀

直接让检测线程调subprocess执行 iptables 很危险，一旦规则写错或 IP 解析出错，可能封掉整个网段。我一般会封装一层，加三个安全阀：白名单校验、封禁频率限制、自动解封定时器。

import subprocess, time, threading WHITELIST = {"192.168.1.1", "192.168.1.100"} BAN_DURATION = 300 # 封禁 5 分钟后自动解封 ban_history = {} # ip -> 解封时间戳 def ban_ip(ip): if ip in WHITELIST: print(f"[跳过] {ip} 在白名单中") return False if ip in ban_history and ban_history[ip] > time.time(): return False # 已在封禁中，避免重复加规则 # 校验 IP 格式，防止命令注入 if not all(p.isdigit() and 0 <= int(p) <= 255 for p in ip.split(".")): return False subprocess.run(["iptables", "-I", "INPUT", "-s", ip, "-j", "DROP"], check=True) ban_history[ip] = time.time() + BAN_DURATION threading.Timer(BAN_DURATION, unban_ip, args=[ip]).start() return True def unban_ip(ip): subprocess.run(["iptables", "-D", "INPUT", "-s", ip, "-j", "DROP"], check=False) ban_history.pop(ip, None)

逻辑说明：ban_ip先过白名单，再查是否已在封禁期，然后做 IP 格式校验——这一步是防命令注入的关键，绝不能把未校验的字符串拼进 shell 命令。subprocess.run用列表传参而不是shell=True，同样是为了安全。threading.Timer实现自动解封，ban_history防止同一 IP 被重复加规则导致 iptables 链膨胀。

参数说明：BAN_DURATION是封禁时长，演示场景 300 秒够用，生产环境可以按攻击严重程度分级，比如扫描封 10 分钟、洪水封 1 小时。check=True让 iptables 执行失败时抛异常，便于日志记录；解封时用check=False，因为规则可能已被手动删除。

4.3 防御动作的可观测性

封禁必须留痕，否则出了问题无法回溯。每次封禁/解封都要写日志，记录时间、IP、触发规则、操作结果。前端监控面板上要能看到「当前封禁列表」和「历史封禁记录」，这是毕设答辩时的加分项，也是实际运维的刚需。

import logging logging.basicConfig(filename="defense.log", level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") def log_ban(ip, rule, success): logging.info(f"ban ip={ip} rule={rule} success={success}")

日志格式里带上规则名，方便统计哪类攻击最频繁。日志文件要配轮转，否则长期运行会撑满磁盘，可以用logging.handlers.RotatingFileHandler。

5. 可视化监控：用 Flask + ECharts 做实时面板

可视化是毕设的门面，也是把前面所有模块串起来的窗口。这一章讲怎么用 Flask 提供数据接口，用 ECharts 做实时刷新，以及怎么避免前端拖垮后端。

5.1 后端接口设计：轮询还是推送

实时监控有两种数据通道：前端定时轮询 REST 接口，或者用 WebSocket 推送。毕设里轮询实现简单、调试方便，推荐先用轮询，刷新间隔 2～3 秒。WebSocket 虽然实时性更好，但 Flask 原生不支持，要引入 Flask-SocketIO，复杂度上升，除非答辩明确要求，否则没必要。

from flask import Flask, jsonify import sqlite3 app = Flask(__name__) @app.route("/api/recent_flows") def recent_flows(): conn = sqlite3.connect("ids.db") rows = conn.execute( "SELECT ts, src, dst, proto, pkts FROM flows ORDER BY ts DESC LIMIT 50" ).fetchall() conn.close() return jsonify([{"ts": r[0], "src": r[1], "dst": r[2], "proto": r[3], "pkts": r[4]} for r in rows]) @app.route("/api/ban_list") def ban_list(): return jsonify(list(ban_history.keys()))

逻辑说明：接口只返回最近 50 条，避免一次传太多数据拖慢前端。jsonify自动处理中文和序列化。数据库连接每次请求新建、用完关闭，简单可靠；高并发场景才需要连接池，毕设规模用不上。

参数说明：LIMIT 50控制返回量，前端表格一屏能显示的行数有限，传太多没意义。刷新间隔由前端setInterval控制，建议 2000～3000 毫秒，太快会给后端和数据库压力。

5.2 前端实时图表与表格

前端用 ECharts 画流量趋势折线图，用表格展示最近流量和封禁列表。关键是每次刷新只更新数据，不重建图表实例，否则会有内存泄漏和闪烁。

// 初始化一次图表实例 const chart = echarts.init(document.getElementById('traffic')); const option = { xAxis: { type: 'category', data: [] }, yAxis: { type: 'value' }, series: [{ type: 'line', data: [], smooth: true }] }; chart.setOption(option); // 定时拉取数据并更新，不重建实例 setInterval(async () => { const res = await fetch('/api/recent_flows'); const data = await res.json(); const times = data.map(d => new Date(d.ts * 1000).toLocaleTimeString()); const pkts = data.map(d => d.pkts); chart.setOption({ xAxis: { data: times }, series: [{ data: pkts }] }); }, 3000);

逻辑说明：echarts.init只调一次，后续用setOption增量更新，这是 ECharts 的正确用法。fetch拉数据后直接映射成坐标轴和序列数据。时间戳乘 1000 是因为 ECharts 和 JS 用毫秒，而 Python 的time.time()是秒。

参数说明：smooth: true让折线平滑，视觉更好但会轻微失真，追求精确可以关掉。刷新间隔 3000 毫秒和前端轮询保持一致，避免请求堆积。

5.3 面板要展示哪些指标

一个能打的监控面板至少要有四块：实时流量趋势、攻击告警列表、当前封禁 IP、系统运行状态（抓包速率、队列长度）。队列长度尤其重要，如果pkt_queue持续接近maxsize，说明消费跟不上生产，要么加消费线程，要么降低抓包过滤粒度。

6. 避坑与排查：这套系统最容易翻车的五个地方

前面讲的都是「应该怎么做」，这一章专门讲「实际会怎么坏」。以下五条都是我在复现这类系统时真实踩过的坑，按「现象 → 原因 → 解决」写，照着排查能省大量时间。

坑一：抓包脚本跑几分钟就内存爆满。现象是进程 RSS 持续上涨直到被 OOM Killer 杀掉。原因是sniff默认store=1，把所有包存在内存里。解决是显式传store=0，并且聚合用的字典要加时间窗口清理，不能只增不减。

坑二：iptables 封禁后自己 SSH 断连。现象是执行封禁脚本后终端卡死，再也连不上。原因是白名单没配，或者封禁规则误伤了管理 IP。解决是封禁前强制校验白名单，并且永远保留一条允许管理网段的规则在最前面。演示环境建议先在虚拟机里试。

坑三：Flask 面板打开一片空白，控制台报跨域。现象是接口单独访问正常，页面里请求失败。原因是前端页面和接口不同源，浏览器拦截。解决是 Flask 加flask-cors扩展，或者把前端页面直接由 Flask 托管，同源就没问题。

坑四：检测规则频繁误报，正常业务被当成攻击。现象是告警列表刷屏，封禁列表里全是正常 IP。原因是阈值写死且偏低，没有基线自适应。解决是引入滑动均值做动态阈值，并给阈值设下限，同时把已知的正常高频 IP 加白名单。

坑五：SQLite 写入报database is locked。现象是抓包线程和 Web 线程同时写库时随机报错。原因是 SQLite 默认单写锁，多线程并发写会冲突。解决是写操作集中到一个线程，或者用WAL模式提升并发，最简单的办法是给写操作加线程锁。

提示：以上五个坑里，内存和权限问题占了大半。上线前先在低流量环境跑够 30 分钟，观察内存曲线和日志，比直接上生产靠谱得多。

7. 进阶技巧：用基线对比验证检测效果

系统跑起来只是及格，能证明它「真的检测到了」才算过关。这一章给一个可复现的验证方法：用基线流量对比攻击流量，量化检测的准确率和响应时间。这也是答辩时最有说服力的一页。

7.1 构造可控的攻击流量做验证

不要用真实攻击工具去打别人的网络，在自己搭的隔离环境里用 Python 构造流量即可。下面这段代码模拟一次 SYN 扫描，用来验证端口扫描规则是否触发。

from scapy.all import IP, TCP, send import random target = "192.168.1.50" # 向目标的不同端口发送 SYN，模拟端口扫描 for port in random.sample(range(1, 1000), 80): pkt = IP(dst=target) / TCP(dport=port, flags="S") send(pkt, verbose=0) print("扫描流量发送完成")

逻辑说明：random.sample从 1～1000 里取 80 个不重复端口，超过规则里 50 的阈值，应该触发PORT_SCAN。flags="S"只发 SYN，不完成握手，符合扫描特征。verbose=0关掉 Scapy 的发送日志，输出干净。

参数说明：端口数量要略高于规则阈值，才能验证「刚好触发」的边界。发送速率用send的inter参数控制，默认不限速，验证慢速扫描时可以设inter=0.1。

7.2 用基线对比量化误报和漏报

验证要成对做：先跑一段纯正常流量记录基线，再叠加攻击流量，看系统是否只在攻击时段告警。下面这张表是我验证时用的记录模板，把每次测试的结果填进去，准确率和响应时间一目了然。

响应延迟从攻击开始到封禁生效，用日志时间戳相减即可。误报数统计非攻击时段触发的告警。如果「正常压测」这一行出现告警，说明白名单或阈值需要调整。

7.3 我自己的习惯

我做完这类系统一定会做一件事：把检测规则和阈值单独抽成一个配置文件，不写死在代码里。因为调参是反复的过程，每次改阈值都重新改代码、重启服务，效率太低。配置文件用 YAML 或 JSON，启动时加载，改完重启即可生效。另一个习惯是给每个封禁动作加一个「干跑模式」，只记日志不真正执行 iptables，先在干跑模式下观察一天，确认没有误封再开真封禁。这个后悔药机制救过我很多次，希望你也能用上。希望帮到你。

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