尧图网站建设 
简介本资源是一套完整可运行的Python网络入侵检测与防御系统毕业设计项目面向计算机相关专业本科生及网络安全初学者解决课程设计、期末大作业与毕业设计选题难、环境搭建复杂、代码调试无从下手等实际问题。压缩包共38个文件含14个核心Python源码如app.py、routes.py、modules模块、Docker容器化部署文件Dockerfile、docker-compose.yml、前端静态资源HTML/JS/CSS、配置文件requirements.txt、README.md、install.sh及说明文档MD/YML/TXT整体仅97KB轻量易解压、零依赖障碍。已有63人下载学习项目经导师指导并获99分高分评价代码结构清晰、注释充分配套文档涵盖系统架构、模块功能、运行步骤与常见问题排错指引小白可直接运行验证适合快速掌握IDS基础原理与工程实践闭环。1. 这不是又一个“用Scikit-learn跑个准确率”的毕业设计它真能抓到SYN Flood、端口扫描和HTTP慢速攻击且所有检测逻辑可调试、可替换、可部署到树莓派你见过多少份标着“网络入侵检测系统”的Python毕业设计点开一看训练集是KDD Cup 99——那个连IP协议栈都没模拟的合成数据模型是RandomForestClassifier.fit(X, y)predict后直接print(检测完成)防御动作是os.system(iptables -D INPUT -p tcp --dport 22 -j DROP)——但没人告诉你这条命令在Windows上根本跑不通也没人告诉你iptables规则没加-s指定源IP删错规则会导致SSH彻底失联。这份基于Python的网络入侵检测与防御系统从抓包开始就走真实路径用scapy实时捕获本机网卡原始流量用pcap_filter语法过滤出可疑SYN包用滑动窗口统计3秒内TCP连接请求数超过阈值触发告警并调用subprocess执行iptables或ufw动态封禁HTTP层则用http.parser解析请求头识别Slowloris类慢速攻击特征如只发HEAD、间隔超长、Connection: keep-alive但不发Body。它不依赖预训练模型所有规则逻辑写在rule_engine.py里变量名全是real_time_conn_count、http_header_field_count、payload_entropy_threshold——你能一眼看懂每行代码在防什么、为什么设这个阈值、改哪个数就能适配你实验室的千兆交换机。适合需要答辩时现场演示“我刚发起一次nmap -sS扫描系统0.8秒后自动封了你的IP”也适合想把毕业设计真正部署在校内服务器上做前置WAF的同学。2. 抓包→解析→规则匹配三层流水线拆解与可复现配置2.1 抓包层绕过libpcap绑定限制在Windows/Linux/macOS统一用scapy.RawSocket项目默认使用scapy的sniff()函数监听网卡但很多同学在Windows下遇到WinPcap/Npcap安装失败、权限不足或抓不到回环流量的问题。这不是bug是scapy默认行为它优先尝试libpcap失败后才fallback到RawSocket。而RawSocket在Windows需管理员权限在Linux需CAP_NET_RAW能力。解决方案是显式指定接口和socket类型from scapy.all import * import os # 强制使用RawSocket避免libpcap依赖 conf.L3socket L3RawSocket # 发包用 conf.L2socket L2RawSocket # 抓包用 # 获取可用网卡列表避免硬编码eth0 interfaces get_if_list() print(可用网卡:, interfaces) # 选择第一个非lo接口如ens33、en0、Wi-Fi target_iface [i for i in interfaces if i not in [lo, Loopback]][0] # 关键设置混杂模式为True否则只能看到发给本机的包 sniff(ifacetarget_iface, prnpacket_callback, store0, promiscTrue, filtertcp or udp or icmp) # BPF过滤语法比全量抓包省90%CPU提示filter参数必须用标准BPF语法不是Wireshark显示过滤器例如tcp and dst port 80合法tcp.port 80非法。常见误写是把and写成或漏掉dst/src限定导致抓到大量ARP包拖慢处理。2.2 解析层从Raw Packet到结构化特征避开scapy字段解析陷阱scapy对TCP选项、HTTP头部的解析是懒加载的——你不访问pkt[TCP].options它就不解析。但毕业设计常犯的错误是直接if pkt[TCP].flags 0x02:判断SYN标志却没检查TCP层是否存在。更隐蔽的坑是IPv6包里pkt[IP]会报KeyError必须先判断IP in pkt or IPv6 in pkt。以下是安全解析模板def extract_features(pkt): features { src_ip: None, dst_ip: None, proto: None, src_port: None, dst_port: None, tcp_flags: 0, payload_len: 0, http_method: None, http_path: None } # 1. IP层统一处理兼容IPv4/IPv6 if IP in pkt: ip_layer pkt[IP] features[src_ip] ip_layer.src features[dst_ip] ip_layer.dst features[proto] ip_layer.proto elif IPv6 in pkt: ip_layer pkt[IPv6] features[src_ip] ip_layer.src features[dst_ip] ip_layer.dst features[proto] ip_layer.nh # IPv6的next header # 2. TCP/UDP层提取端口和标志 if TCP in pkt: tcp_layer pkt[TCP] features[src_port] tcp_layer.sport features[dst_port] tcp_layer.dport features[tcp_flags] tcp_layer.flags # 直接取int值不用位运算 features[payload_len] len(tcp_layer.payload) # 注意不是len(pkt[TCP]) elif UDP in pkt: udp_layer pkt[UDP] features[src_port] udp_layer.sport features[dst_port] udp_layer.dport features[payload_len] len(udp_layer.payload) # 3. HTTP层深度解析仅当TCP且端口80/443时触发 if TCP in pkt and (pkt[TCP].dport in [80, 443] or pkt[TCP].sport in [80, 443]): try: # scapy的HTTP解析不稳定改用手动切分 raw_payload bytes(pkt[TCP].payload) if len(raw_payload) 0 and raw_payload[0] in [bG[0], bP[0], bH[0]]: # GET/POST/HEAD # 简单HTTP头解析取第一行 first_line raw_payload.split(b\n)[0] parts first_line.split() if len(parts) 2: features[http_method] parts[0].decode(utf-8, errorsignore) features[http_path] parts[1].decode(utf-8, errorsignore) except Exception as e: pass # 忽略解析失败不影响主流程 return features关键参数说明len(tcp_layer.payload)获取应用层载荷长度不是整个TCP包长度含headererrorsignore防止HTTP路径含乱码导致decode崩溃raw_payload[0] in [bG[0], ...]用字节ASCII值判断HTTP方法比正则快10倍且不依赖re模块。2.3 规则引擎层状态化检测逻辑告别无状态的“单包打分”多数毕业设计把每个包独立打分但真实攻击是有状态的SYN Flood是短时间内大量SYN包Slowloris是同一IP持续发送不完整的HTTP请求。本项目用collections.defaultdict维护IP级状态from collections import defaultdict, deque import time # 全局状态字典{ip: {conn_count: int, last_seen: float, http_requests: deque}} ip_states defaultdict(lambda: { conn_count: 0, last_seen: 0.0, http_requests: deque(maxlen100) # 只存最近100个HTTP请求时间戳 }) def update_state(features): ip features[src_ip] now time.time() # 1. TCP连接计数3秒滑动窗口 if features[proto] 6 and features[tcp_flags] 0x02: # SYN flag # 清理过期记录 if now - ip_states[ip][last_seen] 3.0: ip_states[ip][conn_count] 0 ip_states[ip][conn_count] 1 ip_states[ip][last_seen] now # 2. HTTP请求频率统计10秒窗口 if features[http_method]: ip_states[ip][http_requests].append(now) def check_attack(ip): # SYN Flood检测3秒内50个SYN包 if ip_states[ip][conn_count] 50: return SYN_FLOOD, ip_states[ip][conn_count] # Slowloris检测10秒内HTTP请求间隔5秒且数量3 reqs list(ip_states[ip][http_requests]) if len(reqs) 3: intervals [reqs[i] - reqs[i-1] for i in range(1, len(reqs))] if all(interval 5.0 for interval in intervals[-3:]): # 最近3次间隔都5秒 return SLOWLORIS, len(reqs) return None, 0为什么用deque(maxlenN)内存占用恒定O(N)比list.append切片更高效maxlen100足够覆盖10秒内高频请求又不会撑爆内存。3. 防御动作闭环从封禁IP到日志审计拒绝“检测完就结束”的假闭环3.1 动态防火墙封禁跨平台命令生成器适配iptables/ufw/firewall-cmd不同系统防火墙命令差异极大Ubuntu用ufw deny from 192.168.1.100CentOS用firewall-cmd --add-rich-rulerule familyipv4 source address192.168.1.100 reject树莓派可能只有iptables。项目提供firewall_manager.py统一接口import subprocess import platform import os class FirewallManager: def __init__(self): self.os_name platform.system().lower() self.distro self._detect_distro() def _detect_distro(self): if os.path.exists(/etc/os-release): with open(/etc/os-release) as f: for line in f: if line.startswith(ID): return line.strip().split()[1].strip() return unknown def block_ip(self, ip): if self.os_name linux: if self.distro in [ubuntu, debian]: cmd [ufw, deny, from, ip] elif self.distro in [centos, rhel, fedora]: cmd [firewall-cmd, --add-rich-rule, frule familyipv4 source address{ip} reject] else: # fallback to iptables cmd [iptables, -A, INPUT, -s, ip, -j, DROP] elif self.os_name darwin: # macOS cmd [pfctl, -t, blocked_ips, -T, add, ip] else: raise OSError(fUnsupported OS: {self.os_name}) try: result subprocess.run(cmd, capture_outputTrue, textTrue, timeout5) if result.returncode 0: print(f[✓] Blocked {ip} via {cmd[0]}) return True else: print(f[✗] Block failed: {result.stderr}) return False except subprocess.TimeoutExpired: print(f[✗] Timeout blocking {ip}) return False # 使用示例 fw FirewallManager() fw.block_ip(192.168.1.100)注意ufw需提前启用sudo ufw enablefirewall-cmd需重启firewalld服务。脚本不自动启用避免学生误操作锁死SSH。3.2 日志审计与可视化ELK替代方案——用SQLiteFlask轻量级实现毕设常忽略日志留存导致答辩时无法证明“确实封禁了”。本项目用SQLite存储每条告警含时间戳、IP、攻击类型、封禁状态CREATE TABLE alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, src_ip TEXT NOT NULL, attack_type TEXT NOT NULL, severity INTEGER CHECK(severity IN (1,2,3)), -- 1low, 2medium, 3high blocked BOOLEAN DEFAULT 0, rule_id TEXT );配套web_dashboard.py启动Flask服务提供实时告警表格和TOP10攻击IP柱状图from flask import Flask, render_template, jsonify import sqlite3 from datetime import datetime, timedelta app Flask(__name__) app.route(/api/alerts) def get_alerts(): conn sqlite3.connect(ids.db) cursor conn.cursor() # 取最近1小时告警按时间倒序 one_hour_ago (datetime.now() - timedelta(hours1)).strftime(%Y-%m-%d %H:%M:%S) cursor.execute( SELECT * FROM alerts WHERE timestamp ? ORDER BY timestamp DESC LIMIT 50 , (one_hour_ago,)) rows cursor.fetchall() conn.close() return jsonify([{ id: r[0], time: r[1], ip: r[2], type: r[3], severity: r[4], blocked: bool(r[5]) } for r in rows]) app.route(/) def dashboard(): return render_template(dashboard.html)提示dashboard.html用Chart.js渲染柱状图数据来自/api/top_ips接口SQL为SELECT src_ip, COUNT(*) as cnt FROM alerts GROUP BY src_ip ORDER BY cnt DESC LIMIT 10。无需部署Elasticsearch单文件SQLite即可支撑百台设备日志。4. 避坑指南答辩前必测的五个血泪问题与当场修复方案4.1 现象抓包时CPU飙升到100%系统卡死原因sniff(filter)未加BPF过滤scapy抓取所有链路层帧含广播、ARP、LLDP每秒数万包压垮Python GIL。解决严格限定filter如ip and (tcp or udp or icmp)在prn回调函数内加if time.time() - last_process_time 0.01: return限频。4.2 现象Windows下scapy.sniff()报错“No module named winpcapy”原因scapy 2.4.5默认依赖winpcapy但pip install scapy不自动装它。解决pip install winpcapy或降级pip install scapy2.4.4兼容性更好。4.3 现象封禁IP后被封机器仍能访问HTTP服务原因iptables规则插入位置错误如插在OUTPUT链而非INPUT链或未指定协议/端口导致规则不匹配。解决用sudo iptables -L INPUT -n --line-numbers查规则序号确认封禁规则在ACCEPT之前添加-p tcp --dport 80明确端口。4.4 现象HTTP慢速攻击检测失效日志里全是“GET / HTTP/1.1”原因scapy解析HTTP时自动补全了缺失的Header把不完整请求当成合法GET。解决禁用scapy的HTTP层解析改用bytes(pkt[TCP].payload)手动分析原始字节检查是否含\r\n\r\n分隔符。4.5 现象多线程运行时ip_states字典出现KeyError原因defaultdict线程不安全多个线程同时访问同一IP键时可能触发竞态。解决用threading.Lock()包裹状态更新state_lock threading.Lock() with state_lock: ip_states[ip][conn_count] 15. 毕业答辩现场演示技巧三步让老师信服“这真是你写的”5.1 第一步环境隔离——用Docker一键还原答辩机环境别让老师用他那台装了Anaconda、PyTorch、CUDA的笔记本跑你的代码。提供docker-compose.ymlversion: 3.8 services: ids: build: . cap_add: - NET_ADMIN - SYS_ADMIN network_mode: host # 必须host模式才能抓本机网卡 volumes: - ./config:/app/config - ./logs:/app/logs environment: - PYTHONUNBUFFERED1Dockerfile里明确指定Python 3.8避免3.11新特性兼容问题并用apt-get install -y libpcap-dev预装依赖。答辩前在老师电脑上docker-compose up -d5分钟搞定纯净环境。5.2 第二步攻击复现——用预录pcap精准触发特定规则别现场nmap -sS等运气用tcpreplay重放已知攻击流量# 下载预置pcap含SYN Flood、Slowloris、SQLi请求 wget https://example.com/attack_samples.pcap # 以1/10速度重放避免冲击生产网络 sudo tcpreplay -i eth0 --rate0.1 attack_samples.pcap在rule_engine.py里加print(f[DEBUG] Triggered {attack_type} on {ip})答辩时终端实时滚动日志比PPT截图有力十倍。5.3 第三步代码溯源——指着rule_engine.py第47行说“这里我调大了阈值”老师问“这个50是怎么来的”别答“网上抄的”。打开test_thresholds.py# 在实验室真实流量中统计正常SYN速率连续10分钟 normal_rates [] for _ in range(10): count count_syn_in_3s() # 实际测量函数 normal_rates.append(count) time.sleep(60) print(正常峰值SYN速率:, max(normal_rates)) # 输出32 # 所以设阈值为32*1.5≈50留50%余量这才是答辩高光时刻你不是调参侠你是用实测数据定义安全边界的工程师。从那以后我每次改阈值都强制走一遍test_thresholds.pytcpreplay验证哪怕只是把50改成55——因为安全没有“差不多”只有“测过”和“没测过”。希望帮到你。本文还有配套的精品资源点击获取 基于Scapy的轻量级网络入侵检测与防御系统
简介本资源是一套完整可运行的Python网络入侵检测与防御系统毕业设计项目面向计算机相关专业本科生及网络安全初学者解决课程设计、期末大作业与毕业设计选题难、环境搭建复杂、代码调试无从下手等实际问题。压缩包共38个文件含14个核心Python源码如app.py、routes.py、modules模块、Docker容器化部署文件Dockerfile、docker-compose.yml、前端静态资源HTML/JS/CSS、配置文件requirements.txt、README.md、install.sh及说明文档MD/YML/TXT整体仅97KB轻量易解压、零依赖障碍。已有63人下载学习项目经导师指导并获99分高分评价代码结构清晰、注释充分配套文档涵盖系统架构、模块功能、运行步骤与常见问题排错指引小白可直接运行验证适合快速掌握IDS基础原理与工程实践闭环。1. 这不是又一个“用Scikit-learn跑个准确率”的毕业设计它真能抓到SYN Flood、端口扫描和HTTP慢速攻击且所有检测逻辑可调试、可替换、可部署到树莓派你见过多少份标着“网络入侵检测系统”的Python毕业设计点开一看训练集是KDD Cup 99——那个连IP协议栈都没模拟的合成数据模型是RandomForestClassifier.fit(X, y)predict后直接print(检测完成)防御动作是os.system(iptables -D INPUT -p tcp --dport 22 -j DROP)——但没人告诉你这条命令在Windows上根本跑不通也没人告诉你iptables规则没加-s指定源IP删错规则会导致SSH彻底失联。这份基于Python的网络入侵检测与防御系统从抓包开始就走真实路径用scapy实时捕获本机网卡原始流量用pcap_filter语法过滤出可疑SYN包用滑动窗口统计3秒内TCP连接请求数超过阈值触发告警并调用subprocess执行iptables或ufw动态封禁HTTP层则用http.parser解析请求头识别Slowloris类慢速攻击特征如只发HEAD、间隔超长、Connection: keep-alive但不发Body。它不依赖预训练模型所有规则逻辑写在rule_engine.py里变量名全是real_time_conn_count、http_header_field_count、payload_entropy_threshold——你能一眼看懂每行代码在防什么、为什么设这个阈值、改哪个数就能适配你实验室的千兆交换机。适合需要答辩时现场演示“我刚发起一次nmap -sS扫描系统0.8秒后自动封了你的IP”也适合想把毕业设计真正部署在校内服务器上做前置WAF的同学。2. 抓包→解析→规则匹配三层流水线拆解与可复现配置2.1 抓包层绕过libpcap绑定限制在Windows/Linux/macOS统一用scapy.RawSocket项目默认使用scapy的sniff()函数监听网卡但很多同学在Windows下遇到WinPcap/Npcap安装失败、权限不足或抓不到回环流量的问题。这不是bug是scapy默认行为它优先尝试libpcap失败后才fallback到RawSocket。而RawSocket在Windows需管理员权限在Linux需CAP_NET_RAW能力。解决方案是显式指定接口和socket类型from scapy.all import * import os # 强制使用RawSocket避免libpcap依赖 conf.L3socket L3RawSocket # 发包用 conf.L2socket L2RawSocket # 抓包用 # 获取可用网卡列表避免硬编码eth0 interfaces get_if_list() print(可用网卡:, interfaces) # 选择第一个非lo接口如ens33、en0、Wi-Fi target_iface [i for i in interfaces if i not in [lo, Loopback]][0] # 关键设置混杂模式为True否则只能看到发给本机的包 sniff(ifacetarget_iface, prnpacket_callback, store0, promiscTrue, filtertcp or udp or icmp) # BPF过滤语法比全量抓包省90%CPU提示filter参数必须用标准BPF语法不是Wireshark显示过滤器例如tcp and dst port 80合法tcp.port 80非法。常见误写是把and写成或漏掉dst/src限定导致抓到大量ARP包拖慢处理。2.2 解析层从Raw Packet到结构化特征避开scapy字段解析陷阱scapy对TCP选项、HTTP头部的解析是懒加载的——你不访问pkt[TCP].options它就不解析。但毕业设计常犯的错误是直接if pkt[TCP].flags 0x02:判断SYN标志却没检查TCP层是否存在。更隐蔽的坑是IPv6包里pkt[IP]会报KeyError必须先判断IP in pkt or IPv6 in pkt。以下是安全解析模板def extract_features(pkt): features { src_ip: None, dst_ip: None, proto: None, src_port: None, dst_port: None, tcp_flags: 0, payload_len: 0, http_method: None, http_path: None } # 1. IP层统一处理兼容IPv4/IPv6 if IP in pkt: ip_layer pkt[IP] features[src_ip] ip_layer.src features[dst_ip] ip_layer.dst features[proto] ip_layer.proto elif IPv6 in pkt: ip_layer pkt[IPv6] features[src_ip] ip_layer.src features[dst_ip] ip_layer.dst features[proto] ip_layer.nh # IPv6的next header # 2. TCP/UDP层提取端口和标志 if TCP in pkt: tcp_layer pkt[TCP] features[src_port] tcp_layer.sport features[dst_port] tcp_layer.dport features[tcp_flags] tcp_layer.flags # 直接取int值不用位运算 features[payload_len] len(tcp_layer.payload) # 注意不是len(pkt[TCP]) elif UDP in pkt: udp_layer pkt[UDP] features[src_port] udp_layer.sport features[dst_port] udp_layer.dport features[payload_len] len(udp_layer.payload) # 3. HTTP层深度解析仅当TCP且端口80/443时触发 if TCP in pkt and (pkt[TCP].dport in [80, 443] or pkt[TCP].sport in [80, 443]): try: # scapy的HTTP解析不稳定改用手动切分 raw_payload bytes(pkt[TCP].payload) if len(raw_payload) 0 and raw_payload[0] in [bG[0], bP[0], bH[0]]: # GET/POST/HEAD # 简单HTTP头解析取第一行 first_line raw_payload.split(b\n)[0] parts first_line.split() if len(parts) 2: features[http_method] parts[0].decode(utf-8, errorsignore) features[http_path] parts[1].decode(utf-8, errorsignore) except Exception as e: pass # 忽略解析失败不影响主流程 return features关键参数说明len(tcp_layer.payload)获取应用层载荷长度不是整个TCP包长度含headererrorsignore防止HTTP路径含乱码导致decode崩溃raw_payload[0] in [bG[0], ...]用字节ASCII值判断HTTP方法比正则快10倍且不依赖re模块。2.3 规则引擎层状态化检测逻辑告别无状态的“单包打分”多数毕业设计把每个包独立打分但真实攻击是有状态的SYN Flood是短时间内大量SYN包Slowloris是同一IP持续发送不完整的HTTP请求。本项目用collections.defaultdict维护IP级状态from collections import defaultdict, deque import time # 全局状态字典{ip: {conn_count: int, last_seen: float, http_requests: deque}} ip_states defaultdict(lambda: { conn_count: 0, last_seen: 0.0, http_requests: deque(maxlen100) # 只存最近100个HTTP请求时间戳 }) def update_state(features): ip features[src_ip] now time.time() # 1. TCP连接计数3秒滑动窗口 if features[proto] 6 and features[tcp_flags] 0x02: # SYN flag # 清理过期记录 if now - ip_states[ip][last_seen] 3.0: ip_states[ip][conn_count] 0 ip_states[ip][conn_count] 1 ip_states[ip][last_seen] now # 2. HTTP请求频率统计10秒窗口 if features[http_method]: ip_states[ip][http_requests].append(now) def check_attack(ip): # SYN Flood检测3秒内50个SYN包 if ip_states[ip][conn_count] 50: return SYN_FLOOD, ip_states[ip][conn_count] # Slowloris检测10秒内HTTP请求间隔5秒且数量3 reqs list(ip_states[ip][http_requests]) if len(reqs) 3: intervals [reqs[i] - reqs[i-1] for i in range(1, len(reqs))] if all(interval 5.0 for interval in intervals[-3:]): # 最近3次间隔都5秒 return SLOWLORIS, len(reqs) return None, 0为什么用deque(maxlenN)内存占用恒定O(N)比list.append切片更高效maxlen100足够覆盖10秒内高频请求又不会撑爆内存。3. 防御动作闭环从封禁IP到日志审计拒绝“检测完就结束”的假闭环3.1 动态防火墙封禁跨平台命令生成器适配iptables/ufw/firewall-cmd不同系统防火墙命令差异极大Ubuntu用ufw deny from 192.168.1.100CentOS用firewall-cmd --add-rich-rulerule familyipv4 source address192.168.1.100 reject树莓派可能只有iptables。项目提供firewall_manager.py统一接口import subprocess import platform import os class FirewallManager: def __init__(self): self.os_name platform.system().lower() self.distro self._detect_distro() def _detect_distro(self): if os.path.exists(/etc/os-release): with open(/etc/os-release) as f: for line in f: if line.startswith(ID): return line.strip().split()[1].strip() return unknown def block_ip(self, ip): if self.os_name linux: if self.distro in [ubuntu, debian]: cmd [ufw, deny, from, ip] elif self.distro in [centos, rhel, fedora]: cmd [firewall-cmd, --add-rich-rule, frule familyipv4 source address{ip} reject] else: # fallback to iptables cmd [iptables, -A, INPUT, -s, ip, -j, DROP] elif self.os_name darwin: # macOS cmd [pfctl, -t, blocked_ips, -T, add, ip] else: raise OSError(fUnsupported OS: {self.os_name}) try: result subprocess.run(cmd, capture_outputTrue, textTrue, timeout5) if result.returncode 0: print(f[✓] Blocked {ip} via {cmd[0]}) return True else: print(f[✗] Block failed: {result.stderr}) return False except subprocess.TimeoutExpired: print(f[✗] Timeout blocking {ip}) return False # 使用示例 fw FirewallManager() fw.block_ip(192.168.1.100)注意ufw需提前启用sudo ufw enablefirewall-cmd需重启firewalld服务。脚本不自动启用避免学生误操作锁死SSH。3.2 日志审计与可视化ELK替代方案——用SQLiteFlask轻量级实现毕设常忽略日志留存导致答辩时无法证明“确实封禁了”。本项目用SQLite存储每条告警含时间戳、IP、攻击类型、封禁状态CREATE TABLE alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, src_ip TEXT NOT NULL, attack_type TEXT NOT NULL, severity INTEGER CHECK(severity IN (1,2,3)), -- 1low, 2medium, 3high blocked BOOLEAN DEFAULT 0, rule_id TEXT );配套web_dashboard.py启动Flask服务提供实时告警表格和TOP10攻击IP柱状图from flask import Flask, render_template, jsonify import sqlite3 from datetime import datetime, timedelta app Flask(__name__) app.route(/api/alerts) def get_alerts(): conn sqlite3.connect(ids.db) cursor conn.cursor() # 取最近1小时告警按时间倒序 one_hour_ago (datetime.now() - timedelta(hours1)).strftime(%Y-%m-%d %H:%M:%S) cursor.execute( SELECT * FROM alerts WHERE timestamp ? ORDER BY timestamp DESC LIMIT 50 , (one_hour_ago,)) rows cursor.fetchall() conn.close() return jsonify([{ id: r[0], time: r[1], ip: r[2], type: r[3], severity: r[4], blocked: bool(r[5]) } for r in rows]) app.route(/) def dashboard(): return render_template(dashboard.html)提示dashboard.html用Chart.js渲染柱状图数据来自/api/top_ips接口SQL为SELECT src_ip, COUNT(*) as cnt FROM alerts GROUP BY src_ip ORDER BY cnt DESC LIMIT 10。无需部署Elasticsearch单文件SQLite即可支撑百台设备日志。4. 避坑指南答辩前必测的五个血泪问题与当场修复方案4.1 现象抓包时CPU飙升到100%系统卡死原因sniff(filter)未加BPF过滤scapy抓取所有链路层帧含广播、ARP、LLDP每秒数万包压垮Python GIL。解决严格限定filter如ip and (tcp or udp or icmp)在prn回调函数内加if time.time() - last_process_time 0.01: return限频。4.2 现象Windows下scapy.sniff()报错“No module named winpcapy”原因scapy 2.4.5默认依赖winpcapy但pip install scapy不自动装它。解决pip install winpcapy或降级pip install scapy2.4.4兼容性更好。4.3 现象封禁IP后被封机器仍能访问HTTP服务原因iptables规则插入位置错误如插在OUTPUT链而非INPUT链或未指定协议/端口导致规则不匹配。解决用sudo iptables -L INPUT -n --line-numbers查规则序号确认封禁规则在ACCEPT之前添加-p tcp --dport 80明确端口。4.4 现象HTTP慢速攻击检测失效日志里全是“GET / HTTP/1.1”原因scapy解析HTTP时自动补全了缺失的Header把不完整请求当成合法GET。解决禁用scapy的HTTP层解析改用bytes(pkt[TCP].payload)手动分析原始字节检查是否含\r\n\r\n分隔符。4.5 现象多线程运行时ip_states字典出现KeyError原因defaultdict线程不安全多个线程同时访问同一IP键时可能触发竞态。解决用threading.Lock()包裹状态更新state_lock threading.Lock() with state_lock: ip_states[ip][conn_count] 15. 毕业答辩现场演示技巧三步让老师信服“这真是你写的”5.1 第一步环境隔离——用Docker一键还原答辩机环境别让老师用他那台装了Anaconda、PyTorch、CUDA的笔记本跑你的代码。提供docker-compose.ymlversion: 3.8 services: ids: build: . cap_add: - NET_ADMIN - SYS_ADMIN network_mode: host # 必须host模式才能抓本机网卡 volumes: - ./config:/app/config - ./logs:/app/logs environment: - PYTHONUNBUFFERED1Dockerfile里明确指定Python 3.8避免3.11新特性兼容问题并用apt-get install -y libpcap-dev预装依赖。答辩前在老师电脑上docker-compose up -d5分钟搞定纯净环境。5.2 第二步攻击复现——用预录pcap精准触发特定规则别现场nmap -sS等运气用tcpreplay重放已知攻击流量# 下载预置pcap含SYN Flood、Slowloris、SQLi请求 wget https://example.com/attack_samples.pcap # 以1/10速度重放避免冲击生产网络 sudo tcpreplay -i eth0 --rate0.1 attack_samples.pcap在rule_engine.py里加print(f[DEBUG] Triggered {attack_type} on {ip})答辩时终端实时滚动日志比PPT截图有力十倍。5.3 第三步代码溯源——指着rule_engine.py第47行说“这里我调大了阈值”老师问“这个50是怎么来的”别答“网上抄的”。打开test_thresholds.py# 在实验室真实流量中统计正常SYN速率连续10分钟 normal_rates [] for _ in range(10): count count_syn_in_3s() # 实际测量函数 normal_rates.append(count) time.sleep(60) print(正常峰值SYN速率:, max(normal_rates)) # 输出32 # 所以设阈值为32*1.5≈50留50%余量这才是答辩高光时刻你不是调参侠你是用实测数据定义安全边界的工程师。从那以后我每次改阈值都强制走一遍test_thresholds.pytcpreplay验证哪怕只是把50改成55——因为安全没有“差不多”只有“测过”和“没测过”。希望帮到你。本文还有配套的精品资源点击获取