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_length=255, unique=True) port_range = models.CharField(max_length=64, default='1-1024') status = models.CharField( max_length=16, default='pending', choices=[ ('pending', '待扫描'), ('scanning', '扫描中'), ('done', '已完成'), ('error', '扫描失败'), ] ) created_by = models.ForeignKey(User, on_delete=models.CASCADE) created_at = models.DateTimeField(auto_now_add=True) 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_delete=models.CASCADE) scan_type = models.CharField(max_length=32, default='quick') start_time = models.DateTimeField(auto_now_add=True) end_time = models.DateTimeField(null=True, blank=True) 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_delete=models.CASCADE) vuln_name = models.CharField(max_length=64) vuln_type = models.CharField(max_length=32, default='unknown') url = models.CharField(max_length=512) payload = models.TextField(default='') risk_level = models.CharField( max_length=8, default='low', choices=[ ('low', '低危'), ('medium', '中危'), ('high', '高危'), ('critical', '严重'), ] ) detail = models.TextField(default='') created_at = models.DateTimeField(auto_now_add=True)

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, ports='21,22,23,25,80,443,3306,6379,8080'): cmd = [ 'nmap', '-sS', '-Pn', '-p', ports, '-oX', '-', '--open', host ] try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=120 ) 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/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', } def create_session(timeout=10, retries=2): session = requests.Session() session.headers.update(DEFAULT_HEADERS) retry_strategy = Retry( total=retries, backoff_factor=0.5, status_forcelist=[500, 502, 503, 504], ) adapter = HTTPAdapter(max_retries=retry_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_redirects=True, timeout=session.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 '1'='1", "' AND SLEEP(3)-- -", "\"", "1 UNION SELECT NULL, NULL-- -", ] ERROR_PATTERNS = [ r"SQL syntax.*MySQL", r"Warning.*mysql_", r"Unclosed quotation mark", r"ORA-[0-9]{5}", r"Microsoft OLE DB Provider for SQL Server", r"PostgreSQL.*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(query=urlencode(test_params, doseq=True))) resp = session.get(test_url, timeout=10) 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(id=task_id) vulns = VulnRecord.objects.filter(scan_task=scan_task) html_content = render_to_string('reports/report_template.html', { 'task': scan_task, 'vulns': vulns, 'generated_at': timezone.now(), }) pdf_file = weasyprint.HTML(string=html_content).write_pdf() response = HttpResponse(pdf_file, content_type='application/pdf') response['Content-Disposition'] = f'attachment; filename="report_{task_id}.pdf"' return response

render_to_string把漏洞列表渲染成全套HTML，包括CSS样式，weasyprint再把它转换成PDF。这个方案的优点是模板改起来方便，技术栈里只要会写HTML就能定制报告格式，不需要专门学报表工具。需要留意的是weasyprint依赖系统库（如libpango、cairo），在服务器上部署时要提前装这些依赖，否则导出PDF时直接报错。

5. 异步任务与并发控制：Celery配置、常见问题与排查

Web渗透测试工具和普通Web应用最大的区别在于：扫描任务耗时动辄几十秒甚至几分钟，不可能让HTTP请求一直挂在那里等结果。nweb源码里解决这个问题的方式几乎都是引入Celery做异步任务队列，用Redis当broker，扫描主逻辑全部丢到worker里执行。这块是代码里最容易翻车的部分，也是答辩时最容易被人追问的地方。

5.1 Celery+Redis异步扫描的任务流程

先看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', namespace='CELERY') 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(id=target_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请求加timeout=10，给nmap调用加--host-timeout 60s参数，同时用soft_time_limit修饰任务：

@shared_task(soft_time_limit=240, time_limit=300) 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服务器默认LANG=C，Python读取到非UTF-8编码的响应内容时无法正常解码。解决：在settings.py里强制设置：

DEFAULT_CHARSET = 'utf-8' FILE_CHARSET = 'utf-8'

MySQL侧把所有表字符集改成utf8mb4，并创建数据库时指定CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci，否则中文和emoji字符都会出问题。

6. 部署验证与进阶技巧：让工具真正跑在目标环境

源码能跑通和能在真实环境里稳定运行是两回事。最后一章讲两件事：怎么给这个工具做安全加固，以及怎么用本地靶机验证它真的能检测出漏洞。前者也是很多人在答辩时会忽略的加分项。

6.1 上线前必须做的安全加固清单

这类工具本身就具备攻击能力，上线前要格外注意不要让未授权用户访问扫描页面。源码里如果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探测会返回status=200，如果DVWA的SQL Injection页面允许低级别权限访问，SQL注入检测代码会在带有id参数的URL上命中报错特征并生成一条高危记录，最终通过报告导出按钮生成PDF。

验证时如果发现什么漏洞都测不出来，先排查目标站的cookie配置。DVWA需要先登录才能到达注入点，而源码里的probe_url默认不带会话Cookie。常见处理是手工先在浏览器登录DVWA，把Cookie配置到create_session里，或者改http_client.py让它自动访问login.php提交表单。这也是这种工具和商业扫描器差距最大的地方——商业产品维护了完整的登录态管理，而课程设计源码往往默认目标不需要认证。

从那以后我每次拿到这类扫描工具源码，都强制先跑一遍靶机验证，确认从任务提交到报告导出的链路真的通，才敢往生产环境里放。希望帮到你。

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