ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

Python+Django构建漏洞数据治理系统

Python+Django构建漏洞数据治理系统 简介这是一套面向计算机专业本科生与网络安全初学者的毕业设计级漏洞扫描系统实战源码基于Python与Django框架构建聚焦Web应用安全审计场景可支撑课程设计、毕设开发及渗透测试工具二次开发。资源包共376个文件涵盖26个核心Python模块含扫描引擎、任务调度与漏洞分析逻辑、53个前端JS脚本与26个CSS样式文件支撑LayuiBootstrap响应式管理界面以及大量预置字典库如nikto.conf、db_subdomains、db_dir_traversal等共20类db文件完整实现目标管理、多线程扫描、结果可视化与权限控制功能压缩包大小为82.58MB。目前已有403人学习下载提供开箱即用的Django项目结构、数据库迁移脚本、配置化扫描策略及典型漏洞检测能力端口探测、服务识别、路径遍历等适合希望深入理解安全工具工程化落地的开发者快速上手并拓展功能。1. 这不是个“扫描器”而是一套可落地的漏洞数据治理闭环当你在 GitHub 或技术论坛看到“基于 Python Django 的漏洞扫描系统源码数据库.zip”这个标题时第一反应可能是又一个带 Web 界面的 nmap 封装但实际拆开这类项目会发现它极少直接调用 masscan 或 nuclei——它的核心价值不在“扫”而在“存、管、查、溯”。典型场景是安全团队每周收到 200 条 Nessus/OpenVAS 扫描报告 CSV人工导入 Excel 标记重复项、比对历史、写整改跟踪表耗时且易漏而这个 Django 系统把扫描结果无论来自哪种工具统一解析为结构化模型自动去重、关联资产、标记状态、生成趋势图表并支持按 CVE 编号、影响组件、修复状态、所属业务线多维筛选。它面向的是中型企业的安全运营岗、DevSecOps 工程师或高校信息安全课程设计者——你需要的不是“再写一遍端口扫描逻辑”而是让每次扫描产出的数据真正沉淀为可复用、可审计、可驱动修复的知识资产。Python 提供灵活的解析与调度能力Django 提供开箱即用的权限管理、API 接口和 Admin 后台数据库通常是 PostgreSQL 或 MySQL则承载漏洞生命周期的完整状态流转。2. 用 Django Model 建模漏洞全生命周期而非简单存 raw JSON2.1 为什么必须放弃“一条记录 一份扫描原始输出”的粗放设计很多初版漏洞系统直接把nuclei -o report.json的输出整条存进TextField看似省事实则彻底丧失查询与分析能力。例如无法快速统计“近 30 天 Apache Tomcat 暴露在公网的 CVE-2023-22745 数量”也无法筛选“状态为‘待验证’且影响资产属于‘支付网关集群’的所有高危漏洞”。Django 的 ORM 能力在此刻成为刚需——必须将扫描结果解耦为原子化、可索引、可关联的模型。2.2 核心模型设计从资产到漏洞再到处置动作以下是最小可行模型集已通过 Django 4.2 实测全部定义在vuln/models.py中# vuln/models.py from django.db import models from django.contrib.auth.models import User class Asset(models.Model): ip models.GenericIPAddressField() hostname models.CharField(max_length255, blankTrue) port models.PositiveIntegerField() service models.CharField(max_length100, blankTrue) # e.g., http, ssh product models.CharField(max_length200, blankTrue) # e.g., nginx/1.22.1 version models.CharField(max_length50, blankTrue) created_at models.DateTimeField(auto_now_addTrue) class Meta: unique_together (ip, port) # 防止同一端口重复录入 indexes [ models.Index(fields[ip]), models.Index(fields[service, product]), ] class Vulnerability(models.Model): CVE_ID models.CharField(max_length20, db_indexTrue, blankTrue) # 可为空因部分漏洞无 CVE title models.CharField(max_length500) severity models.CharField( max_length10, choices[(CRITICAL, Critical), (HIGH, High), (MEDIUM, Medium), (LOW, Low), (INFO, Info)] ) description models.TextField() cvss_score models.DecimalField(max_digits3, decimal_places1, nullTrue, blankTrue) references models.JSONField(defaultlist) # 存 URL 列表如 [https://nvd.nist.gov/vuln/detail/CVE-2023-22745] published_date models.DateField(nullTrue, blankTrue) created_at models.DateTimeField(auto_now_addTrue) class ScanResult(models.Model): asset models.ForeignKey(Asset, on_deletemodels.CASCADE, related_namescan_results) vulnerability models.ForeignKey(Vulnerability, on_deletemodels.CASCADE, related_namescan_results) scanner models.CharField(max_length50) # e.g., nuclei, nessus, custom_script raw_data models.JSONField() # 存原始扫描器返回的字段如 nuclei 的 matched-at, template-id status models.CharField( max_length20, choices[ (NEW, New), (CONFIRMED, Confirmed), (FALSE_POSITIVE, False Positive), (FIXED, Fixed), (IGNORED, Ignored) ], defaultNEW ) reporter models.ForeignKey(User, on_deletemodels.SET_NULL, nullTrue, blankTrue) reported_at models.DateTimeField(auto_now_addTrue) verified_at models.DateTimeField(nullTrue, blankTrue) fixed_at models.DateTimeField(nullTrue, blankTrue) class Meta: unique_together (asset, vulnerability, scanner) # 同一资产漏洞扫描器组合唯一提示unique_together是防重复录入的关键。若某次扫描发现资产10.1.1.5:8080存在CVE-2023-22745且扫描器为nuclei则后续相同组合的扫描结果将被忽略或触发更新逻辑而非新增记录。2.3 数据库迁移与初始数据加载执行迁移前确保settings.py中DATABASES配置指向真实数据库非 SQLite生产环境推荐 PostgreSQL# 创建迁移文件 python manage.py makemigrations vuln # 应用迁移生成表结构 python manage.py migrate # 加载初始漏洞库可选提升演示效果 python manage.py loaddata initial_vulns.json其中initial_vulns.json是预置的常见漏洞数据如 OWASP Top 10 模板格式为标准 Django fixture[ { model: vuln.vulnerability, pk: 1, fields: { CVE_ID: CVE-2023-22745, title: Apache Tomcat Remote Code Execution, severity: CRITICAL, description: Unauthenticated remote code execution via crafted HTTP request..., cvss_score: 9.8, references: [https://nvd.nist.gov/vuln/detail/CVE-2023-22745], published_date: 2023-03-15 } } ]3. 解析不同扫描器输出用 Python 写适配器而非硬编码规则3.1 为什么不能只支持一种扫描器现实中的混合输入场景企业安全流程中边界扫描用 Nessus容器镜像用 TrivyWeb 应用用 NucleiAPI 测试用 Burp Suite Exporter —— 它们的输出格式天差地别Nessus 是 XMLTrivy 是 JSON with nested arraysNuclei 是每行一个 JSON 对象ndjsonBurp 是 HTML 表格。强行用正则或固定 key 解析必然崩溃。正确做法是为每种扫描器编写独立的 Parser 类统一实现parse()方法返回标准化字典列表。3.2 实现一个可扩展的 Parser 架构在vuln/parsers/目录下创建基类与具体实现# vuln/parsers/base.py from abc import ABC, abstractmethod class BaseParser(ABC): abstractmethod def parse(self, raw_content: str) - list[dict]: 输入原始扫描输出内容字符串 输出标准化字典列表每个 dict 必须含 - ip: str - port: int - cve_id: str (可空) - title: str - severity: str in [CRITICAL, HIGH, ...] - description: str - scanner: str pass# vuln/parsers/nuclei_parser.py import json from .base import BaseParser class NucleiParser(BaseParser): def parse(self, raw_content: str) - list[dict]: results [] for line in raw_content.strip().split(\n): if not line.strip(): continue try: data json.loads(line) # nuclei v3 输出结构示例{matched-at:http://10.1.1.5:8080/,host:10.1.1.5,port:8080,template-id:tomcat-cve-2023-22745,...} ip data.get(host, ).split(:)[0] if data.get(host) else port data.get(port, 0) if not ip or port 0: continue results.append({ ip: ip, port: port, cve_id: data.get(cve-id, ), title: data.get(info, {}).get(name, Unknown), severity: data.get(info, {}).get(severity, INFO).upper(), description: data.get(info, {}).get(description, ), scanner: nuclei }) except json.JSONDecodeError: continue # 跳过非法 JSON 行 return results# vuln/parsers/nessus_parser.py from xml.etree import ElementTree as ET from .base import BaseParser class NessusParser(BaseParser): def parse(self, raw_content: str) - list[dict]: root ET.fromstring(raw_content) results [] for report_host in root.findall(.//ReportHost): ip report_host.get(name, ) for item in report_host.findall(.//ReportItem): port item.get(port, 0) if not port.isdigit(): continue plugin_output item.find(.//plugin_output) if plugin_output is None: continue # 提取 CVENessus XML 中通常在 cve 标签或描述文本中 cve_id cve_elem item.find(.//cve) if cve_elem is not None and cve_elem.text: cve_id cve_elem.text.strip() results.append({ ip: ip, port: int(port), cve_id: cve_id, title: item.get(pluginName, Unknown), severity: self._map_nessus_severity(item.get(severity, 0)), description: plugin_output.text[:500] if plugin_output.text else , scanner: nessus }) return results def _map_nessus_severity(self, nessus_code: str) - str: mapping {0: INFO, 1: LOW, 2: MEDIUM, 3: HIGH, 4: CRITICAL} return mapping.get(nessus_code, INFO)3.3 在视图中调用解析器并入库vuln/views.py中的上传处理逻辑# vuln/views.py from django.shortcuts import render, redirect from django.contrib import messages from django.core.files.storage import default_storage from django.http import HttpResponse from .parsers import NucleiParser, NessusParser from .models import Asset, Vulnerability, ScanResult def upload_scan_report(request): if request.method POST and request.FILES.get(report_file): file request.FILES[report_file] content file.read().decode(utf-8) filename file.name.lower() # 自动选择解析器 parser None if filename.endswith(.json) and nuclei in filename: parser NucleiParser() elif filename.endswith(.xml) and nessus in filename: parser NessusParser() else: messages.error(request, 不支持的文件类型请上传 nuclei JSON 或 Nessus XML) return redirect(upload) if parser: try: parsed_data parser.parse(content) processed_count 0 for item in parsed_data: # 1. 获取或创建 Asset asset, _ Asset.objects.get_or_create( ipitem[ip], portitem[port], defaults{ service: unknown, product: unknown, version: unknown } ) # 2. 获取或创建 Vulnerability vuln, _ Vulnerability.objects.get_or_create( CVE_IDitem[cve_id], defaults{ title: item[title][:499], severity: item[severity], description: item[description][:999], cvss_score: None } ) # 3. 创建 ScanResult去重 ScanResult.objects.get_or_create( assetasset, vulnerabilityvuln, scanneritem[scanner], defaults{status: NEW} ) processed_count 1 messages.success(request, f成功解析并入库 {processed_count} 条漏洞记录) except Exception as e: messages.error(request, f解析失败{str(e)}) return redirect(upload) return render(request, vuln/upload.html)注意get_or_create()是避免重复的核心。它基于unique_together约束确保同一资产漏洞扫描器组合仅存一条记录后续导入相同结果时不会新增而是返回已有对象。4. 在 Django Admin 中实现漏洞状态流转与批量操作4.1 重写 ScanResultAdmin让安全员用鼠标完成 80% 的日常操作默认 Admin 界面仅支持增删改查但漏洞运营需要一键标记多个漏洞为“已确认”、导出当前筛选结果为 Excel、查看某资产所有历史漏洞。这些必须通过自定义 Admin 实现# vuln/admin.py from django.contrib import admin from django.http import HttpResponse from django.urls import path from openpyxl import Workbook from openpyxl.styles import Font, PatternFill from .models import ScanResult, Asset, Vulnerability admin.register(ScanResult) class ScanResultAdmin(admin.ModelAdmin): list_display (asset, vulnerability, scanner, status, reported_at, verified_at, fixed_at) list_filter (status, scanner, vulnerability__severity, asset__service) search_fields (asset__ip, vulnerability__title, vulnerability__CVE_ID) actions [mark_as_confirmed, mark_as_fixed, export_to_excel] def get_urls(self): urls super().get_urls() custom_urls [ path(export-excel/, self.export_excel_view, namevuln_scanresult_export_excel), ] return custom_urls urls def mark_as_confirmed(self, request, queryset): queryset.update(statusCONFIRMED, verified_attimezone.now()) mark_as_confirmed.short_description 标记为【已确认】 def mark_as_fixed(self, request, queryset): queryset.update(statusFIXED, fixed_attimezone.now()) mark_as_fixed.short_description 标记为【已修复】 def export_to_excel(self, request, queryset): response HttpResponse( content_typeapplication/vnd.openxmlformats-officedocument.spreadsheetml.sheet, ) response[Content-Disposition] attachment; filenamevuln_export.xlsx wb Workbook() ws wb.active ws.title 漏洞清单 # 表头 headers [IP, 端口, 服务, 漏洞标题, CVE编号, 严重等级, 状态, 发现时间, 确认时间, 修复时间] for col_num, header in enumerate(headers, 1): cell ws.cell(row1, columncol_num, valueheader) cell.font Font(boldTrue) cell.fill PatternFill(start_colorD3D3D3, end_colorD3D3D3, fill_typesolid) # 数据行 for row_num, obj in enumerate(queryset, 2): ws.cell(rowrow_num, column1, valueobj.asset.ip) ws.cell(rowrow_num, column2, valueobj.asset.port) ws.cell(rowrow_num, column3, valueobj.asset.service) ws.cell(rowrow_num, column4, valueobj.vulnerability.title[:50]) ws.cell(rowrow_num, column5, valueobj.vulnerability.CVE_ID) ws.cell(rowrow_num, column6, valueobj.vulnerability.severity) ws.cell(rowrow_num, column7, valueobj.status) ws.cell(rowrow_num, column8, valueobj.reported_at.strftime(%Y-%m-%d %H:%M)) ws.cell(rowrow_num, column9, valueobj.verified_at.strftime(%Y-%m-%d %H:%M) if obj.verified_at else ) ws.cell(rowrow_num, column10, valueobj.fixed_at.strftime(%Y-%m-%d %H:%M) if obj.fixed_at else ) wb.save(response) return response export_to_excel.short_description 导出选中项为 Excel def export_excel_view(self, request): # 此方法用于支持 URL 直接导出如从列表页按钮跳转 queryset self.get_queryset(request) return self.export_to_excel(request, queryset)4.2 添加资产关联视图一眼看清某 IP 的全部风险在vuln/admin.py中为Asset模型添加内联编辑直观展示该资产关联的漏洞class ScanResultInline(admin.TabularInline): model ScanResult extra 0 fields (vulnerability, scanner, status, reported_at, verified_at, fixed_at) readonly_fields (reported_at, verified_at, fixed_at) show_change_link True admin.register(Asset) class AssetAdmin(admin.ModelAdmin): list_display (ip, hostname, port, service, product, created_at) list_filter (service, product, created_at) search_fields (ip, hostname, service) inlines [ScanResultInline]提示TabularInline让管理员在查看10.1.1.5资产时直接在其页面下方看到所有关联漏洞及当前状态无需跳转大幅提升排查效率。5. 生产部署关键配置与性能优化点5.1 数据库层面索引与查询优化漏洞表数据增长极快未加索引的SELECT * FROM scanresult WHERE statusNEW AND asset_id IN (SELECT id FROM asset WHERE servicehttp)会全表扫描。必须在models.py中显式声明索引# vuln/models.py续 class ScanResult(models.Model): # ... 字段定义不变 ... class Meta: unique_together (asset, vulnerability, scanner) indexes [ models.Index(fields[status]), # 频繁按状态筛选 models.Index(fields[asset, status]), # 资产状态联合查询 models.Index(fields[vulnerability, status]), # 漏洞状态联合查询 models.Index(fields[reported_at]), # 按时间排序 ]执行python manage.py sqlmigrate vuln 0002可查看生成的 SQL确认CREATE INDEX语句已包含。5.2 Django 设置关闭 DEBUG启用连接池限制上传大小settings.py关键生产配置DEBUG False ALLOWED_HOSTS [vuln.yourcompany.com, 10.10.10.5] # 严禁 [*] # 数据库连接池以 PostgreSQL 为例需安装 django-db-geventpool DATABASES { default: { ENGINE: django_db_geventpool.backends.postgresql_psycopg2, NAME: vuln_db, USER: vuln_user, PASSWORD: os.environ.get(DB_PASSWORD), HOST: 127.0.0.1, PORT: 5432, OPTIONS: { MAX_CONNS: 20, MIN_CONNS: 5, } } } # 限制上传文件大小防止恶意大文件 DATA_UPLOAD_MAX_MEMORY_SIZE 26214400 # 25MB FILE_UPLOAD_MAX_MEMORY_SIZE 262144005.3 使用 Celery 异步处理扫描报告解析解析大型 Nessus XML100MB会阻塞 Web 请求。应解耦为异步任务# vuln/tasks.py from celery import shared_task from .parsers import NessusParser, NucleiParser from .models import ScanResult, Asset, Vulnerability shared_task def async_parse_and_import(file_path: str, scanner_type: str): with open(file_path, r, encodingutf-8) as f: content f.read() parser NessusParser() if scanner_type nessus else NucleiParser() parsed_data parser.parse(content) for item in parsed_data: asset, _ Asset.objects.get_or_create(ipitem[ip], portitem[port]) vuln, _ Vulnerability.objects.get_or_create(CVE_IDitem[cve_id], defaults{title: item[title]}) ScanResult.objects.get_or_create( assetasset, vulnerabilityvuln, scanneritem[scanner], defaults{status: NEW} )在视图中触发# vuln/views.py修改 upload_scan_report def upload_scan_report(request): if request.method POST and request.FILES.get(report_file): file request.FILES[report_file] # 保存文件到临时目录 temp_path f/tmp/{file.name} with open(temp_path, wb) as destination: for chunk in file.chunks(): destination.write(chunk) # 异步启动解析 async_parse_and_import.delay(temp_path, nuclei if nuclei in file.name else nessus) messages.info(request, 报告已提交后台解析稍后可在列表中查看结果) return redirect(upload)注意Celery 需配合 Redis 或 RabbitMQ 作为 Broker此配置超出本文范围但它是支撑日均万级扫描报告入库的必备基础设施。5.4 最小化部署验证命令三步确认系统就绪部署完成后用以下命令链快速验证核心链路是否通畅# 1. 检查数据库连接与迁移状态 python manage.py dbshell # 进入 psql/mysql执行 \dt 查看表是否存在 # 2. 创建超级用户用于登录 Admin python manage.py createsuperuser # 3. 启动开发服务器仅验证生产用 Gunicorn python manage.py runserver 0.0.0.0:8000 # 4. 手动插入一条测试数据curl 模拟 API 或 Admin 页面操作 curl -X POST http://localhost:8000/admin/login/ \ -F usernameadmin \ -F passwordyourpass \ -F next/admin/vuln/scanresult/add/ \ --cookie-jar cookies.txt # 5. 查看日志确认无 WARNING/ERROR tail -f ./logs/django.log | grep -E (ERROR|WARNING)只要ScanResult表能成功写入记录且 Admin 界面可正常筛选、导出、状态变更即证明这套基于 Python Django 的漏洞扫描系统源码数据库已具备生产可用基础。后续迭代可聚焦于集成更多扫描器解析器、增加漏洞修复建议知识库、对接 Jira/禅道工单系统、构建资产指纹自动识别模块。本文还有配套的精品资源点击获取
返回列表