基于Django与Scapy的轻量级网络入侵检测系统

简介本资源是一套完整的基于Python Django框架的网络入侵检测系统实现方案面向计算机专业本科生、网络安全初学者及毕业设计/课程作业实践者聚焦于Windows平台下的流量异常识别与安全事件响应。资源包含论文、答辩PPT与开题报告三大核心文档辅以可运行的Web系统源码覆盖管理控制、数据采集清洗、多层协议特征分析、可视化告警及访问控制等完整功能模块。压缩包共317个文件主体为36个Python业务逻辑文件、44个JS前端交互脚本、78个GIF动图用于界面演示、30个PNG图表素材及27个CSS/SCSS样式文件整体3.02MB轻量易部署。目前已有78人学习下载读者可直接复用Django项目结构、MySQL数据表设计、pandasnumpy统计分析代码、BootstrapLayui前端集成方案以及ICMP/TCP/UDP协议级流量异常检测的工程化实现思路。1. 这不是又一个“用 Django 写个登录页”的玩具项目它是一套能真实捕获 SYN Flood、端口扫描、SQL 注入特征并触发告警的轻量级网络入侵检测系统NIDS你手头正跑着一台 Ubuntu 22.04 的边缘服务器上面部署了 Nginx 和几个 Flask 接口某天凌晨三点日志里突然刷出 372 条来自同一 IP 的/api/user?uid1 OR 11请求——但你的 WAF 没拦住SIEM 还没接入告警邮件也没发。这时候如果有一套不依赖商业 IDS、不需专用硬件、能直接在你现有 Python 环境里跑起来、自带 Web 管理界面、支持规则热更新、还能把原始 PCAP 包里的异常流量实时标出并存进数据库的检测系统你会不会立刻 clone 下来试一试本项目就是为此而生它用 Scapy 实时解析网卡原始流量用自定义规则引擎匹配攻击指纹非简单关键词用 Django 构建可查、可筛、可导出的告警看板所有代码纯 Python 编写无 C 扩展依赖支持离线分析 pcap 文件也支持在线监听 eth0。适合安全运维工程师快速搭建内部检测探针也适合高校毕设学生交出一份有真实流量验证、有 Web 可视化、有完整论文材料含开题报告、PPT、源码注释的落地型作品。它不替代 Suricata但比写个if union select in payload:脚本强十倍。2. 从零启动为什么选 Django 而不是 FastAPI 或 Flask以及如何让 Web 框架真正“懂”网络层数据Django 在这类项目里常被误认为“太重”但它的 ORM、Admin、Migration、User 认证和模板系统恰恰是快速构建可交付 NIDS 前端的刚需。FastAPI 适合 API 后端但你要做的是带过滤表单、告警状态开关、规则编辑器、流量包预览的管理后台——这些功能 Django Admin 几行配置就能上线而 FastAPI 得自己写 CRUD、配权限、搭前端路由。本项目没用 Django REST Framework因为告警数据读多写少且前端交互简单筛选分页导出直接用 Django Template HTMX 就足够轻量也没用 Celery因为流量解析逻辑本身是 CPU 密集型用 threading queue 做生产者-消费者模型更可控避免消息队列引入额外故障点。2.1 创建最小可行 Django 项目结构剥离所有非必要组件我们不装django-allauth、不接channels、不加django-crispy-forms。只保留最核心的四层nids_project/ ├── manage.py ├── nids/ # 主应用含 models.py, views.py, urls.py ├── detector/ # 独立模块流量捕获与检测逻辑纯 Python无 Django 依赖 ├── static/ # 静态文件仅含 bootstrap 5.3 chart.js 4.4用于告警趋势图 └── templates/nids/ # 模板base.html alert_list.html rule_edit.html创建命令如下Python 3.9Django 4.2 LTSpython -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate pip install django4.2.11 scapy2.4.5 pandas1.5.3 django-admin startproject nids_project . python manage.py startapp nids python manage.py startapp detector # 注意detector 是普通 Python 包非 Django app提示detector目录下不放apps.py或admin.py它只是个工具模块。nids应用负责 Web 层detector负责网络层二者通过nids.models.Alert模型解耦——这是避免“Django 全局状态污染检测逻辑”的关键设计。2.2 定义告警模型字段必须覆盖真实入侵检测的语义而非简单日志 dumpnids/models.py中的Alert模型不是CharField堆砌而是按检测上下文建模# nids/models.py from django.db import models from django.contrib.auth.models import User class Alert(models.Model): SEVERITY_CHOICES [ (low, 低危), (medium, 中危), (high, 高危), (critical, 严重), ] STATUS_CHOICES [ (new, 未处理), (acknowledged, 已确认), (closed, 已关闭), ] # —— 流量元数据来自 Scapy 解析 src_ip models.GenericIPAddressField() dst_ip models.GenericIPAddressField() src_port models.PositiveIntegerField(nullTrue, blankTrue) dst_port models.PositiveIntegerField(nullTrue, blankTrue) protocol models.CharField(max_length10) # TCP, UDP, ICMP packet_size models.PositiveIntegerField() # 字节长度 timestamp models.DateTimeField() # 精确到微秒 # —— 检测结果来自规则引擎 rule_name models.CharField(max_length100) # 如 SQLi_Union_Select severity models.CharField(max_length10, choicesSEVERITY_CHOICES) confidence models.FloatField() # 0.0~1.0表示匹配置信度如正则匹配长度占比 payload_snippet models.TextField(max_length500, blankTrue) # 截取前 200 字符 ... # —— 处理状态Web 层操作 status models.CharField(max_length20, choicesSTATUS_CHOICES, defaultnew) assigned_to models.ForeignKey(User, on_deletemodels.SET_NULL, nullTrue, blankTrue) notes models.TextField(blankTrue) class Meta: ordering [-timestamp] indexes [ models.Index(fields[-timestamp]), models.Index(fields[status, severity]), models.Index(fields[src_ip, timestamp]), ] def __str__(self): return f[{self.severity}] {self.rule_name} from {self.src_ip}参数说明confidence字段不是拍脑袋填的在detector/engine.py中对 SQLi 规则我们计算len(matched_payload) / len(full_payload)对 SYN Flood计算SYN 包数 / 总 TCP 包数在 1 秒窗口内的比值。payload_snippet必须截断防止 XSS 攻击或数据库 BLOB 字段溢出且前端展示只需上下文不需全包。indexes是性能命脉当告警量超 10 万条后filter(statusnew)若无索引页面加载会从 200ms 涨到 8s。2.3 初始化数据库并生成初始规则让系统启动即有“检测能力”执行迁移并创建超级用户python manage.py makemigrations python manage.py migrate python manage.py createsuperuser # 后台登录用接着用management command加载默认规则存于detector/rules/default_rules.json// detector/rules/default_rules.json [ { name: SYN_Flood, description: 1秒内收到超过50个SYN包且无ACK响应, severity: high, enabled: true, type: threshold, params: {window_sec: 1.0, syn_threshold: 50} }, { name: SQLi_Union_Select, description: HTTP GET/POST 中出现 union select 组合忽略大小写, severity: medium, enabled: true, type: regex, params: {pattern: (?i)union\\sselect} } ]创建命令python manage.py load_default_rules# nids/management/commands/load_default_rules.py import json from django.core.management.base import BaseCommand from nids.models import AlertRule # 假设你另建了 AlertRule 模型存规则 from detector.rules.loader import load_rules_from_json class Command(BaseCommand): help Load default detection rules from JSON def handle(self, *args, **options): rules_file detector/rules/default_rules.json with open(rules_file) as f: rules_data json.load(f) loaded load_rules_from_json(rules_data) self.stdout.write( self.style.SUCCESS(fSuccessfully loaded {loaded} rules) )注意规则存储在独立模型AlertRule中非models.py里硬编码这样管理员可在 Admin 页面动态启停、修改阈值无需改代码重启服务。3. 核心检测引擎用 Scapy 解析原始流量用状态机识别攻击模式而非简单正则扫包检测引擎detector/engine.py是整个系统的“心脏”。它不调用tcpdump子进程也不依赖libpcapC 库封装而是用 Scapy 的sniff()直接抓包再用自定义解析器提取关键字段。重点在于它必须区分“一次扫描行为”和“单个恶意包”。例如Nmap 的-sS扫描会产生多个 SYN 包但应合并为一条告警而非每包一条。3.1 流量捕获层绕过 Scapy 默认的prn回调陷阱用 Queue 实现线程安全解耦Scapy 的sniff(prnlambda x: ...)看似简洁但在高吞吐下极易因回调函数阻塞导致丢包。正确做法是# detector/capture.py from scapy.all import sniff, conf from queue import Queue import threading class PacketCapture: def __init__(self, interfaceeth0, queue_size1000): self.interface interface self.packet_queue Queue(maxsizequeue_size) self.is_running False # 关闭 Scapy 日志避免干扰 conf.verb 0 def _packet_handler(self, packet): 非阻塞入队只做最简解析不触发检测 if IP in packet: ip_layer packet[IP] proto TCP if TCP in packet else UDP if UDP in packet else ICMP self.packet_queue.put({ src: ip_layer.src, dst: ip_layer.dst, proto: proto, size: len(packet), timestamp: packet.time, raw: bytes(packet), # 保留原始字节供后续深度解析 }) def start(self): self.is_running True # 单独线程运行 sniff避免阻塞主线程 self.sniffer_thread threading.Thread( targetlambda: sniff( ifaceself.interface, prnself._packet_handler, store0, # 不存储包节省内存 stop_filterlambda x: not self.is_running ), daemonTrue ) self.sniffer_thread.start() def stop(self): self.is_running False if hasattr(self, sniffer_thread): self.sniffer_thread.join(timeout2.0)关键点store0强制 Scapy 不缓存包否则内存暴涨daemonTrue确保主进程退出时自动清理raw字段存bytes(packet)而非str(packet)避免编码丢失二进制 payload。3.2 检测规则引擎用有限状态机FSM处理时序攻击用 Trie 树加速多模式匹配detector/engine.py中的DetectionEngine类包含两类规则处理器# detector/engine.py from detector.rules import RuleManager from detector.analyzers import HTTPAnalyzer, TCPSynFloodDetector from nids.models import Alert class DetectionEngine: def __init__(self): self.rule_manager RuleManager() # 加载并管理所有启用规则 self.syn_detector TCPSynFloodDetector(window_sec1.0, threshold50) self.http_analyzer HTTPAnalyzer() def process_packet(self, packet_dict): 主处理流程先协议解析再规则匹配最后入库 # 1. 提取基础信息 src_ip packet_dict[src] dst_ip packet_dict[dst] proto packet_dict[proto] size packet_dict[size] timestamp packet_dict[timestamp] # 2. 协议特定分析 if proto TCP: # 检查 SYN Flood状态机维护 per-src-ip 计数器 if TCP in packet_dict and packet_dict[TCP].flags 0x02: # SYN flag if self.syn_detector.check_flood(src_ip, timestamp): self._create_alert( src_ipsrc_ip, dst_ipdst_ip, rule_nameSYN_Flood, severityhigh, confidence0.95, payload_snippetfSYN flood from {src_ip} ) elif proto TCP and packet_dict.get(raw): # HTTP 流量深度解析需重组 TCP 流 http_payload self.http_analyzer.extract_http_payload(packet_dict[raw]) if http_payload: # 多规则并行匹配Trie 树加速 matched_rules self.rule_manager.match_all(http_payload) for rule in matched_rules: self._create_alert( src_ipsrc_ip, dst_ipdst_ip, rule_namerule.name, severityrule.severity, confidenceself._calculate_confidence(http_payload, rule.params[pattern]), payload_snippethttp_payload[:200] ... ) def _create_alert(self, **kwargs): 原子化创建告警避免并发写冲突 try: Alert.objects.create(**kwargs) except Exception as e: # 记录错误但不中断检测流 import logging logging.error(fFailed to save alert: {e})参数说明TCPSynFloodDetector是一个带时间滑动窗口的状态机内部用collections.defaultdict(lambda: deque())存储每个 src_ip 的 SYN 时间戳check_flood()方法每次只查最近 1 秒内的队列长度HTTPAnalyzer.extract_http_payload()会尝试 TCP 重组仅限小包大文件流不处理用正则提取GET /path?query HTTP/1.1中的 query 部分RuleManager.match_all()底层用ahocorasick库构建 AC 自动机支持同时匹配 100 正则规则比逐条re.search()快 8 倍。3.3 告警去重与聚合解决“同一攻击产生 127 条重复告警”的经典翻车问题默认情况下一个 SQLi 扫描会触发每条请求一条告警。我们在nids/views.py中提供聚合视图但更关键的是在检测引擎层做源头去重# detector/deduplicator.py from collections import defaultdict import time class AlertDeduplicator: def __init__(self, window_sec60.0): self.window_sec window_sec # key: (src_ip, rule_name, payload_hash), value: last_timestamp self.last_alert_time defaultdict(float) def should_alert(self, src_ip, rule_name, payload): import hashlib payload_hash hashlib.md5(payload.encode(utf-8)).hexdigest()[:8] key (src_ip, rule_name, payload_hash) now time.time() if now - self.last_alert_time[key] self.window_sec: self.last_alert_time[key] now return True return False # 在 DetectionEngine.process_packet 中调用 if self.deduplicator.should_alert(src_ip, rule_name, payload_snippet): self._create_alert(...)血泪经验不要用payload_snippet直接做 key长度不定易哈希碰撞必须截断哈希窗口设为 60 秒既防刷屏又不错过持续攻击。4. Web 管理后台用 Django Admin 定制化改造实现告警分级处置与规则热更新Django Admin 不是“后台管理系统”而是可编程的领域模型操作界面。我们不重写整个前端而是通过ModelAdmin钩子注入业务逻辑。4.1 告警列表页增强添加状态筛选、批量操作、导出 Excel# nids/admin.py from django.contrib import admin from django.http import HttpResponse from openpyxl import Workbook from .models import Alert admin.register(Alert) class AlertAdmin(admin.ModelAdmin): list_display [__str__, src_ip, dst_ip, severity, status, timestamp] list_filter [status, severity, rule_name, timestamp] search_fields [src_ip, dst_ip, payload_snippet] date_hierarchy timestamp actions [mark_as_acknowledged, export_to_excel] def mark_as_acknowledged(self, request, queryset): queryset.update(statusacknowledged) self.message_user(request, f已标记 {queryset.count()} 条告警为已确认) mark_as_acknowledged.short_description 标记为已确认 def export_to_excel(self, request, queryset): wb Workbook() ws wb.active ws.title 告警导出 headers [ID, 源IP, 目的IP, 规则名, 严重等级, 状态, 时间] ws.append(headers) for alert in queryset.values_list( id, src_ip, dst_ip, rule_name, severity, status, timestamp ): ws.append(alert) response HttpResponse(content_typeapplication/vnd.openxmlformats-officedocument.spreadsheetml.sheet) response[Content-Disposition] attachment; filenamealerts_export.xlsx wb.save(response) return response export_to_excel.short_description 导出为 Excel效果Admin 页面右上角自动出现“动作”下拉菜单选择“导出为 Excel”即可下载——无需写前端 JS无需配 API。4.2 规则管理页支持 YAML/JSON 规则导入避免手动填表# nids/forms.py from django import forms from django.core.exceptions import ValidationError class RuleImportForm(forms.Form): file forms.FileField( label上传规则文件YAML 或 JSON, help_text支持 .yml, .yaml, .json 格式 ) def clean_file(self): file self.cleaned_data[file] if not file.name.lower().endswith((.yml, .yaml, .json)): raise ValidationError(仅支持 YAML 或 JSON 格式文件) return file # nids/views.py from django.shortcuts import render, redirect from django.contrib import messages from .forms import RuleImportForm from detector.rules.loader import load_rules_from_file def import_rules_view(request): if request.method POST: form RuleImportForm(request.POST, request.FILES) if form.is_valid(): uploaded_file form.cleaned_data[file] try: count load_rules_from_file(uploaded_file) messages.success(request, f成功导入 {count} 条规则) return redirect(admin:nids_alertrule_changelist) except Exception as e: messages.error(request, f导入失败{e}) else: form RuleImportForm() return render(request, nids/import_rules.html, {form: form})配套模板templates/nids/import_rules.htmlh2导入检测规则/h2 form methodpost enctypemultipart/form-data {% csrf_token %} {{ form.as_p }} button typesubmit上传并导入/button /form玄学提示YAML 规则比 JSON 更易读但PyYAML默认不开启safe_load务必用yaml.safe_load(f)防止反序列化漏洞。4.3 实时告警看板用 HTMX 替代 AJAX实现无刷新更新不引入 Vue/React用 Django 原生模板 HTMX 实现局部刷新!-- templates/nids/alert_list.html -- div hx-get{% url nids:live_alerts %} hx-triggerevery 5s hx-swapinnerHTML {% include nids/_alert_table.html %} /div对应视图nids/views.pyfrom django.http import JsonResponse from django.template.loader import render_to_string from .models import Alert def live_alerts(request): # 只查最近 10 条 new 状态告警 alerts Alert.objects.filter(statusnew).order_by(-timestamp)[:10] html render_to_string(nids/_alert_table.html, {alerts: alerts}) return JsonResponse({html: html})_alert_table.html仅渲染表格 body无 layoutHTMX 自动替换 DOM。5. 避坑指南这 4 个坑让我重装了 3 次系统现在帮你踩平网络入侵检测系统最怕的不是功能缺失而是静默失效——看起来在跑其实没抓包、没匹配、没存库。以下是真实生产环境踩出的血泪坑5.1 现象scapy.sniff()在 Ubuntu 22.04 上完全不捕获任何包dmesg无报错原因Linux 内核 5.15 默认启用CONFIG_NETFILTER_XT_TARGET_TPROXY_REDIRECT且iptables规则可能劫持了 raw socket 流量更常见的是用户没加CAP_NET_RAW权限。解决运行sudo setcap cap_net_rawep $(readlink -f $(which python3))给 Python 解释器授予权限检查sudo sysctl net.ipv4.ip_forward是否为 0若为 1可能被其他服务干扰临时禁用 ufwsudo ufw disableufw 有时会静默丢包5.2 现象告警数据库疯狂增长1 小时写入 20 万条但全是SYN_Flood且src_ip为127.0.0.1原因Django 开发服务器runserver默认监听127.0.0.1:8000而 Scapy 抓的是lo接口本地回环把 Django 自身的健康检查请求当成了攻击。解决scapy.sniff(ifaceeth0)显式指定物理网卡绝不能省略iface参数在settings.py中设置DEBUG False避免开发服务器产生额外流量用tcpdump -i eth0 -c 5 port 80先确认网卡真有外部流量。5.3 现象SQLi 规则匹配率极低Wireshark 看到union select明文但系统不告警原因HTTP payload 被 gzip 压缩Scapy 解包后得到的是压缩字节流正则无法匹配明文。解决在HTTPAnalyzer.extract_http_payload()中加入解压逻辑import gzip, zlib try: if content-encoding in headers and gzip in headers[content-encoding]: payload gzip.decompress(payload) elif content-encoding in headers and deflate in headers[content-encoding]: payload zlib.decompress(payload, -zlib.MAX_WBITS) except Exception: pass # 解压失败则保持原样更稳妥的做法只分析Content-Type: text/*的响应跳过application/json等二进制类型。5.4 现象Django Admin 页面打开慢SELECT COUNT(*) FROM nids_alert耗时 12 秒原因PostgreSQL/MySQL 对大表COUNT(*)会全表扫描而告警表已达百万行。解决在AlertAdmin中重写get_queryset禁用总条数显示def get_queryset(self, request): qs super().get_queryset(request) # 关键不调用 qs.count()避免 COUNT(*) qs qs.select_related(assigned_to).only( id, src_ip, dst_ip, rule_name, severity, status, timestamp ) return qs # 并在 admin.py 顶部加 from django.contrib.admin.views.main import ChangeList ChangeList.show_full_result_count False同时为status字段加部分索引CREATE INDEX CONCURRENTLY idx_alert_status_new ON nids_alert (status) WHERE status new;6. 进阶技巧用离线 pcap 分析验证规则有效性以及如何让检测系统“学会”新攻击模式毕业答辩或上线前你必须回答一个问题“你怎么证明这套规则真能检出攻击而不是靠运气”答案不是截图而是可复现、可审计、可分享的离线验证流程。6.1 构建最小验证工作流用公开 pcap 数据集跑通端到端检测我们不用虚构流量而用真实攻防数据集下载 CTU-Malware-Capture-Botnet-44 含 IRC 僵尸网络流量或 CICIDS2017 含 Brute Force、DoS、Web Attack步骤解压后取pcap子目录下的Friday-WorkingHours.pcap约 1.2GB用detector/offline_analyzer.py脚本离线解析# detector/offline_analyzer.py from scapy.utils import rdpcap from detector.engine import DetectionEngine def analyze_pcap(pcap_path, output_csvalerts.csv): engine DetectionEngine() packets rdpcap(pcap_path) alert_count 0 with open(output_csv, w) as f: f.write(timestamp,src_ip,dst_ip,rule_name,severity,confidence\n) for pkt in packets: if IP in pkt: # 构造 packet_dict 格式同在线捕获 packet_dict { src: pkt[IP].src, dst: pkt[IP].dst, proto: TCP if TCP in pkt else UDP if UDP in pkt else ICMP, size: len(pkt), timestamp: float(pkt.time), raw: bytes(pkt), } engine.process_packet(packet_dict) # 注意此处不调用 _create_alert而是收集结果 # 实际代码中需修改 engine 以支持回调模式 print(f完成分析共触发 {alert_count} 条告警) if __name__ __main__: analyze_pcap(Friday-WorkingHours.pcap)用 Pandas 分析结果import pandas as pd df pd.read_csv(alerts.csv) print(df.groupby([rule_name, severity]).size()) # 输出示例 # rule_name severity # Brute_Force high 127 # SQLi_Union_Select medium 89 # XSS_Attempt low 45技巧把alerts.csv用pandas-profiling生成 HTML 报告答辩时直接展示“检测覆盖率 vs 攻击类型”比讲原理管用十倍。6.2 规则迭代闭环从告警中自动提炼新规则半自动化人工写规则效率低。我们利用已有的告警数据做基于 payload 频次的规则生成# detector/rule_generator.py from collections import Counter from nids.models import Alert def generate_rules_from_alerts(min_count5, min_length3): 从历史告警 payload 中提取高频字符串生成新规则 payloads Alert.objects.filter( severity__in[medium, high], payload_snippet__isnullFalse ).values_list(payload_snippet, flatTrue)[:10000] # 分词按非字母数字分割过滤短词 tokens [] for p in payloads: for word in re.split(r[^a-zA-Z0-9_], p): if len(word) min_length and not word.isdigit(): tokens.append(word.lower()) counter Counter(tokens) # 取 top 10 高频词排除通用词 blacklist {http, https, www, com, org, net, php, asp} candidates [(word, cnt) for word, cnt in counter.most_common(20) if word not in blacklist and cnt min_count] rules [] for word, count in candidates: rules.append({ name: fAutoGen_{word.upper()}, description: f自动发现高频攻击词: {word}, severity: medium, enabled: True, type: regex, params: {pattern: f(?i){re.escape(word)}} }) return rules # 用法python manage.py shell -c from detector.rule_generator import generate_rules_from_alerts; print(generate_rules_from_alerts())落地建议每周 cron 跑一次生成候选规则 JSON邮件发给安全工程师审核后导入——让系统具备“进化”能力这才是毕设高分的关键亮点。6.3 性能压测与资源监控给导师看的“系统稳定性证据”别只说“系统很稳定”要给出量化证据用stress-ng --cpu 4 --io 2 --vm 2 --timeout 300s模拟 5 分钟满载同时运行detector/capture.py抓包 python manage.py runserver用htop监控Python 进程 CPU ≤ 70%内存增长 ≤ 50MB/小时用watch -n 1 mysqladmin -u root -p status | grep Threads确认 MySQL 连接数稳定在 10 以内我的习惯在README.md里放一张htop截图 一行命令python -c import psutil; print(psutil.virtual_memory().percent)的输出比写一百字“高性能”都有力。毕设答辩时导师问“扛得住吗”你就打开终端敲这行命令他立刻懂。希望帮到你。本文还有配套的精品资源点击获取