Django渗透测试工具源码解析:从数据模型到异步扫描实战

发布时间:2026/10/30 14:20:25
Django渗透测试工具源码解析:从数据模型到异步扫描实战 简介python毕业设计之nweb渗透测试工具是基于Django框架开发的网络安全评估项目源码面向计算机相关专业毕业生、Django学习者及Web安全入门开发者。项目实现扫描器、目录遍历、CSRF防护、弱口令测试等常见功能涵盖models、views、urls、templates等Django常用模块适合课设参考或安全工具入门开发。压缩包约557.95MB共2000个文件以svg、py、pyc、css、js、html等为主py/pyc对应核心逻辑css/js/html对应界面log/sqlite3记录运行数据目录结构便于按Django分层阅读。已有497人学习下载。阅读源码可掌握Django工程组织方式、常见漏洞原理与检测思路得到一套可运行、可二次开发的渗透测试工具框架对理解Web安全与提升编码能力有实际价值。1. nweb是什么一份能跑通的Django渗透测试工具源码做毕业设计或安全测试接手一份标注为“nweb渗透测试工具”的Django源码时绝大多数人会先急着运行然后被一堆配置、依赖和数据库迁移问题劝退。我拆过不少同类项目第一反应永远是先看它的models和任务调度再决定要不要跑。这份源码本质上是一个B/S架构的Web版渗透测试工具用Python的Django框架做后端把端口扫描、子域名收集、目录爆破、SQL注入检测、XSS检测这些常见测试动作封装成Web页面操作适合毕业设计参考也适合刚接触Web安全的工程师在本地环境做实战练习。它不追求像Nessus那样庞大胜在结构清晰、代码量适中能让你从一条完整扫描链路里看懂工具类Web应用的写法。2. 项目骨架与数据建模从app划分到任务表设计拿到源码先别急着跑迁移把Django项目的目录结构和数据模型看明白后面所有排查都会省力。这类工具项目通常不会像业务系统那样划分十几个app而是按功能边界拆成目标管理、扫描执行、漏洞记录、报告展示四个模块每个模块对应一个app。源码里常见的是targets、scan_core、vulns、reports这样命名的目录配合users做登录鉴权。2.1 Django项目结构与核心app划分一个典型的nweb项目目录长这样nweb/ ├── manage.py ├── config/ # 项目配置settings、celery、urls ├── targets/ # 目标管理增删改查扫描目标 ├── scan_core/ # 扫描执行nmap、目录爆破、探测逻辑 ├── vulns/ # 漏洞检测sqlmap集成、xss检测规则 ├── reports/ # 报告生成HTML/PDF导出 ├── users/ # 用户登录与权限 └── static/ # 前端静态资源这个划分是我比较认可的方式扫描执行和漏洞检测分离意味着你可以单独扩展检测插件而不破坏整体流程目标管理和报告生成分开则是为了让一个扫描任务可以对应多份漏洞记录、多份报告。实际写的时候config里会有celery.py和settings.py两部分配置任务队列独立于Web请求进程运行。2.2 数据模型设计扫描任务与漏洞记录的表结构数据模型是整个工具的地基nweb里最核心的几张表是Target、ScanTask和VulnRecord。Target存扫描目标ScanTask记录一次完整扫描的启动时间、状态和关联目标VulnRecord存检测到的漏洞细节。看models.py时先看这三个基本就能还原整个工具的业务流程。# targets/models.py from django.db import models from django.contrib.auth.models import User class Target(models.Model): host models.CharField(max_length255, uniqueTrue) port_range models.CharField(max_length64, default1-1024) status models.CharField( max_length16, defaultpending, choices[ (pending, 待扫描), (scanning, 扫描中), (done, 已完成), (error, 扫描失败), ] ) created_by models.ForeignKey(User, on_deletemodels.CASCADE) created_at models.DateTimeField(auto_now_addTrue) def __str__(self): return self.host# scan_core/models.py from django.db import models class ScanTask(models.Model): target models.ForeignKey(targets.Target, on_deletemodels.CASCADE) scan_type models.CharField(max_length32, defaultquick) start_time models.DateTimeField(auto_now_addTrue) end_time models.DateTimeField(nullTrue, blankTrue) result_summary models.TextField(default) class Meta: ordering [-start_time]host字段设置唯一索引是为了防止同一个目标被重复添加这在多用户并发操作时很有用。ScanTask通过外键关联到Target一个目标可以发起多轮扫描但每轮扫描只属于一个目标。result_summary一般存扫描结果的JSON序列化文本比如发现的开放端口列表、子域名数量这样列表页展示时不用再去查漏洞表。漏洞记录表的设计有一个小细节要留意漏洞类型通常用vuln_type字符串字段而不是直接建外键表原因是不想为了兼容新的检测插件反复改表结构。# vulns/models.py from django.db import models class VulnRecord(models.Model): scan_task models.ForeignKey(scan_core.ScanTask, on_deletemodels.CASCADE) vuln_name models.CharField(max_length64) vuln_type models.CharField(max_length32, defaultunknown) url models.CharField(max_length512) payload models.TextField(default) risk_level models.CharField( max_length8, defaultlow, choices[ (low, 低危), (medium, 中危), (high, 高危), (critical, 严重), ] ) detail models.TextField(default) created_at models.DateTimeField(auto_now_addTrue)payload字段必须保留它记录触发漏洞的原始请求数据方便复核。detail则存响应特征比如SQL注入报错信息、XSS回显内容。这样一个表结构就能支撑报告页面的全部图表数据来源。提示拿到源码后先执行python manage.py makemigrations python manage.py migrate如果迁移报错优先检查Django版本和MySQL驱动版本是否匹配而不是直接删数据库。3. 核心扫描模块实战Nmap集成与HTTP请求封装的细节扫描模块是整个工具能不能用的关键。nweb里最常见的做法是用subprocess调用系统Nmap解析XML输出HTTP探测部分则用requests库封装处理Cookie、UA、超时和重定向。这两块代码读懂了你就能自己改扫描逻辑。3.1 用subprocess调用Nmap并解析XML结果为什么不用python-nmap库因为毕业设计场景下直接调用系统命令更容易展示你对底层命令的理解而且python-nmap在Windows和Linux下的行为差异会引入额外依赖问题。源码里通常封装一个scan_ports函数参数是主机地址和端口范围返回解析后的端口列表。# scan_core/port_scanner.py import subprocess import xml.etree.ElementTree as ET def scan_ports(host, ports21,22,23,25,80,443,3306,6379,8080): cmd [ nmap, -sS, -Pn, -p, ports, -oX, -, --open, host ] try: result subprocess.run( cmd, capture_outputTrue, textTrue, timeout120 ) except subprocess.TimeoutExpired: return [] if result.returncode ! 0: return [] root ET.fromstring(result.stdout) open_ports [] for host_elem in root.iter(host): for port_elem in host_elem.iter(port): state port_elem.find(state) if state is not None and state.get(state) open: open_ports.append(port_elem.get(portid)) return open_ports-sS是TCP SYN半开扫描速度快且日志少但在非root用户下会失败-Pn跳过主机发现直接探测端口。-oX -把结果以XML格式输出到stdout这样就不需要临时文件。解析时用ElementTree遍历port节点读state里的state属性只保留open的结果。实际使用时有一个参数要注意ports字符串是英文逗号分隔内部不要有空格。如果目标主机在线但防火墙丢弃了探测包--open参数能过滤掉大量filtered状态的结果避免列表被无意义数据刷屏。3.2 HTTP请求封装会话保持、UA伪装与响应特征提取端口扫描只解决“哪些服务在监听”的问题HTTP层探测需要封装独立的session逻辑。nweb源码里一般有一个http_client.py核心代码是对requests.Session做一层包装统一设置UA、超时、重试策略同时记录响应时间、状态码、响应头和正文片段。# scan_core/http_client.py import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry DEFAULT_HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, } def create_session(timeout10, retries2): session requests.Session() session.headers.update(DEFAULT_HEADERS) retry_strategy Retry( totalretries, backoff_factor0.5, status_forcelist[500, 502, 503, 504], ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter) session.timeout timeout return session# scan_core/probe.py def probe_url(session, url): try: resp session.get(url, allow_redirectsTrue, timeoutsession.timeout) return { url: resp.url, status: resp.status_code, headers: dict(resp.headers), body_prefix: resp.text[:500], elapsed: round(resp.elapsed.total_seconds(), 3), } except requests.RequestException as exc: return { url: url, status: 0, headers: {}, body_prefix: str(exc)[:200], elapsed: 0, }这里的Retry策略只对HTTP 5xx错误生效连接超时和读取超时不会重试原因很简单扫描场景下如果目标响应慢重试只会放大时间成本不如直接跳过并记录超时状态。DEFAULT_HEADERS里的UA不可省略很多中间件和WAF会拦截空UA或python-requests默认UA这属于入门级反爬对抗。body_prefix截取前500字节是为了后续判断指纹和漏洞特征截太长会拖慢数据库写入速度。探测结果最终会写入ScanTask.result_summary用json.dumps序列化方便前端渲染。一个常见的坑是响应头里的某些二进制值无法直接存入MySQL的utf8字段遇到这类情况要对dict(resp.headers)做一次str()转换或过滤非ASCII字符。4. 漏洞检测与报告生成从规则命中到PDF导出nweb的价值不只在于扫描端口更在于把漏洞检测结果变成可读的报告。这部分源码通常包含两类检测逻辑一类是简单的规则匹配比如检查响应中是否包含特定报错关键字另一类是主动发送Payload观察响应差异。报告生成则依赖Django模板引擎渲染HTML再用工具转成PDF。4.1 SQL注入检测基于报错与响应差异的规则实现源码里SQL注入检测的常见写法是维护一个Payload列表针对URL参数逐个注入测试Payload用“响应中是否包含数据库报错特征”或“页面响应长度与正常请求是否明显不同”来判断。这类实现虽然不如sqlmap全面但足够教学演示。# vulns/sql_check.py import re from urllib.parse import urlencode, urlparse, parse_qs, urlunparse SQL_PAYLOADS [ , OR 11, AND SLEEP(3)-- -, \, 1 UNION SELECT NULL, NULL-- -, ] ERROR_PATTERNS [ rSQL syntax.*MySQL, rWarning.*mysql_, rUnclosed quotation mark, rORA-[0-9]{5}, rMicrosoft OLE DB Provider for SQL Server, rPostgreSQL.*ERROR, ] def check_sql_injection(session, url): parsed urlparse(url) params parse_qs(parsed.query) for key in params.keys(): for payload in SQL_PAYLOADS: test_params params.copy() test_params[key] [payload] test_url urlunparse(parsed._replace(queryurlencode(test_params, doseqTrue))) resp session.get(test_url, timeout10) if resp.status_code ! 200: continue if re.search(|.join(ERROR_PATTERNS), resp.text, re.I): return { vuln: SQL注入, url: test_url, payload: payload, risk: high, } if resp.elapsed.total_seconds() 3 and SLEEP in payload: return { vuln: SQL时间盲注, url: test_url, payload: payload, risk: high, } return None这里有一个处理细节parse_qs会把参数值变成列表所以复写参数时要用test_params[key] [payload]而不是直接赋值字符串否则urlencode会报错。SLEEP(3)配合响应时长判断是时间盲注的简化做法但要设置好请求超时不然探测一个慢速目标会卡住整个扫描任务。ERROR_PATTERNS里的正则只是最基础的一组不同数据库的报错特征差异很大源码里通常允许你维护一个外部配置文件来扩充规则。4.2 XSS检测与基于Django模板的报告导出XSS检测相对简单核心思路是把Payload拼进URL参数然后检查响应中是否原样回显了Payload内容。这里重点不是检测算法而是报告模块怎么把这些结果组织起来。nweb一般用Django模板渲染一份检测详情页然后通过weasyprint或pdfkit生成PDF这里用HTML模板加weasyprint的写法最常见。# reports/views.py from django.template.loader import render_to_string import weasyprint def export_report(request, task_id): scan_task ScanTask.objects.get(idtask_id) vulns VulnRecord.objects.filter(scan_taskscan_task) html_content render_to_string(reports/report_template.html, { task: scan_task, vulns: vulns, generated_at: timezone.now(), }) pdf_file weasyprint.HTML(stringhtml_content).write_pdf() response HttpResponse(pdf_file, content_typeapplication/pdf) response[Content-Disposition] fattachment; filenamereport_{task_id}.pdf return responserender_to_string把漏洞列表渲染成全套HTML包括CSS样式weasyprint再把它转换成PDF。这个方案的优点是模板改起来方便技术栈里只要会写HTML就能定制报告格式不需要专门学报表工具。需要留意的是weasyprint依赖系统库如libpango、cairo在服务器上部署时要提前装这些依赖否则导出PDF时直接报错。5. 异步任务与并发控制Celery配置、常见问题与排查Web渗透测试工具和普通Web应用最大的区别在于扫描任务耗时动辄几十秒甚至几分钟不可能让HTTP请求一直挂在那里等结果。nweb源码里解决这个问题的方式几乎都是引入Celery做异步任务队列用Redis当broker扫描主逻辑全部丢到worker里执行。这块是代码里最容易翻车的部分也是答辩时最容易被人追问的地方。5.1 CeleryRedis异步扫描的任务流程先看config/celery.py标准的Celery初始化方式# config/celery.py import os from celery import Celery os.environ.setdefault(DJANGO_SETTINGS_MODULE, config.settings) app Celery(nweb) app.config_from_object(django.conf:settings, namespaceCELERY) app.autodiscover_tasks()# config/settings.py 部分配置 CELERY_BROKER_URL redis://localhost:6379/0 CELERY_RESULT_BACKEND redis://localhost:6379/1 CELERY_ACCEPT_CONTENT [json] CELERY_TASK_SERIALIZER json CELERY_TIMEZONE Asia/Shanghai在scan_core/tasks.py里定义扫描任务启动扫描的视图只负责创建数据库记录并调用delay()# scan_core/tasks.py from celery import shared_task from .port_scanner import scan_ports from .probe import probe_url shared_task def run_scan_task(target_id): target Target.objects.get(idtarget_id) target.status scanning target.save() open_ports scan_ports(target.host, target.port_range) http_results [] for port in open_ports: if port in (80, 443, 8080): scheme https if port 443 else http url f{scheme}://{target.host}:{port}/ session create_session() http_results.append(probe_url(session, url)) target.result_summary json.dumps({ open_ports: open_ports, http_probes: http_results, }) target.status done target.save()视图里的调用方式很简单run_scan_task.delay(target.id)HTTP请求立刻返回真正的扫描在Celery worker进程里执行。worker侧启动命令是celery -A config worker -l info生产环境还要配合celery beat做定时任务但nweb这类按需触发的工具一般用不到beat。5.2 常见问题与排查记录这部分总结了我排查这种项目时最常碰到的四类问题每一条都是实际踩过的坑。现象一任务提交后Redis里有数据但worker一直没有执行日志。原因worker没有启动或者启动时加载了错误的配置文件。很多人只启动Django开发服务器就点按钮忘了单独开Celery worker进程。解决确认celery -A config worker -l info正在运行查看worker启动日志里有没有DatabaseSynchronizer相关的warning如果worker和Django用了不同的虚拟环境检查Redis URL是否一致。现象二任务执行到一半整个worker卡死重启后恢复正常。原因扫描逻辑里某个请求没有超时控制比如DNS解析或端口握手长达几分钟不返回。nmap的subprocess.run虽然有timeout参数但底层socket连接不受它控制。解决给所有HTTP请求加timeout10给nmap调用加--host-timeout 60s参数同时用soft_time_limit修饰任务shared_task(soft_time_limit240, time_limit300) def run_scan_task(target_id): ...现象三SQLite数据库提示database is locked。原因Celery worker和Django Web进程同时写入SQLite数据库SQLite的写锁只允许单进程操作。解决要么换MySQL/PostgreSQL要么把SQLite设置为WAL模式# settings.py DATABASES { default: { ENGINE: django.db.backends.sqlite3, NAME: BASE_DIR / db.sqlite3, OPTIONS: {timeout: 30}, } }加上timeout参数只是缓解正确做法是测试用SQLite生产环境切MySQL。现象四扫描结果里中文字符显示为乱码或者写入MySQL时报编码错误。原因Linux服务器默认LANGCPython读取到非UTF-8编码的响应内容时无法正常解码。解决在settings.py里强制设置DEFAULT_CHARSET utf-8 FILE_CHARSET utf-8MySQL侧把所有表字符集改成utf8mb4并创建数据库时指定CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci否则中文和emoji字符都会出问题。6. 部署验证与进阶技巧让工具真正跑在目标环境源码能跑通和能在真实环境里稳定运行是两回事。最后一章讲两件事怎么给这个工具做安全加固以及怎么用本地靶机验证它真的能检测出漏洞。前者也是很多人在答辩时会忽略的加分项。6.1 上线前必须做的安全加固清单检查项推荐配置说明DEBUG开关DEBUGFalse忘记关会导致堆栈信息直接暴露给访问者允许主机ALLOWED_HOSTS[test.local]不配的话Django会拒绝所有非localhost请求CSRF防护保持默认启用扫描工具也属于管理后台CSRF中间件不能关认证强度启用Django自带auth至少在目标列表增删和报告下载接口上加登录限制Redis访问设置密码并绑定127.0.0.1公网裸奔的Redis很容易被人直接flushall这类工具本身就具备攻击能力上线前要格外注意不要让未授权用户访问扫描页面。源码里如果views.py只有login_required装饰器而没有额外权限控制至少把报告导出接口加上登录校验。6.2 用DVWA靶机验证完整扫描链路验证工具能不能用最稳妥的做法是在本地起一个DVWA靶场跑一次完整的“目标添加→端口探测→HTTP探测→SQL注入检测→报告导出”流程。# 启动DVWA容器 docker run -d -p 8080:80 vulnerables/web-dvwa # 启动nweb开发环境 python manage.py runserver 0.0.0.0:8000 # 启动Celery worker celery -A config worker -l info然后在工具里添加目标http://127.0.0.1:8080端口范围填80触发扫描任务。正常情况下open_ports里会出现80端口HTTP探测会返回status200如果DVWA的SQL Injection页面允许低级别权限访问SQL注入检测代码会在带有id参数的URL上命中报错特征并生成一条高危记录最终通过报告导出按钮生成PDF。验证时如果发现什么漏洞都测不出来先排查目标站的cookie配置。DVWA需要先登录才能到达注入点而源码里的probe_url默认不带会话Cookie。常见处理是手工先在浏览器登录DVWA把Cookie配置到create_session里或者改http_client.py让它自动访问login.php提交表单。这也是这种工具和商业扫描器差距最大的地方——商业产品维护了完整的登录态管理而课程设计源码往往默认目标不需要认证。从那以后我每次拿到这类扫描工具源码都强制先跑一遍靶机验证确认从任务提交到报告导出的链路真的通才敢往生产环境里放。希望帮到你。本文还有配套的精品资源点击获取