尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

Python漏洞扫描教学系统:Flask架构与模块化实现

Python漏洞扫描教学系统:Flask架构与模块化实现 简介这是一套面向网络安全初学者与毕业设计学生的Python漏洞扫描系统实战源码聚焦Web资产探测与基础漏洞识别场景解决入门者缺乏完整项目练手的问题。资源共87个文件压缩包大小1.46MB包含27个Python核心逻辑文件如app.py、vulnscan.py、portscan.py等、13个HTML前端页面含登录、端口扫描、漏洞扫描等交互界面、6个CSS样式文件与5个JavaScript脚本共同构成前后端分离的轻量级扫描平台migrations、model、scan等目录体现模块化架构设计支持数据库迁移与功能解耦。已有335人学习下载读者可直接部署运行掌握Python网络编程、Flask框架开发、前端交互集成及常见扫描工具Nmap、Xray、Wappalyzer调用方法同时获得结构清晰的毕业设计参考范例与可扩展的二次开发基础。1. 这不是AWVS替代品而是一个可调试、可拆解的Python漏洞扫描教学系统你打开upload.zip看到86个文件、26个.py、13个HTML模板、migrations/目录下三段带哈希名的迁移脚本——这不是一个开箱即用的商业扫描器而是一套面向教学与二次开发的Python漏洞扫描骨架。它不追求每秒万级并发或绕过WAF的0day利用但把“从域名输入→端口探测→指纹识别→漏洞验证→任务持久化→Web界面展示”这条完整链路用清晰分层的代码结构摊开在你面前。app.py是Flask主入口scan/目录下每个.py文件对应一种扫描能力Nmap端口、Wappalyzer指纹、Xray/AWVS联动model/里用SQLAlchemy定义了Task、VulnResult等实体templates/中VulnScanPage.html和VulnInformationPage.html构成漏洞详情闭环。它适合两类人一是网络安全方向的毕业设计学生需要可运行、可答辩、可解释原理的完整项目二是刚接触渗透测试工具链的开发者想搞懂“为什么扫描结果能存进数据库”“前端表单如何触发后端扫描任务”“User-Agent轮换怎么集成进请求模块”。作者在readme.txt里自谦“技术有限”恰恰说明这里没有过度封装的黑盒所有randomAgent.py的UA池、decorators.py的登录鉴权装饰器、fingerprintscan.py中对apps.json的匹配逻辑都裸露在源码里等你动手改、打断点、加日志。2. Flask架构解析与核心模块职责拆解2.1 主控流程app.py如何调度扫描任务生命周期app.py是整个系统的中枢神经它不直接执行扫描而是通过路由注册、蓝图挂载、数据库初始化三步构建调度框架。关键代码如下# app.py 片段 from flask import Flask from flask_sqlalchemy import SQLAlchemy from flask_login import LoginManager from config import Config from model import db from utils import init_db def create_app(): app Flask(__name__) app.config.from_object(Config) # 加载config.py中的SECRET_KEY、SQLALCHEMY_DATABASE_URI等 db.init_app(app) # 绑定SQLAlchemy实例到Flask应用 login_manager LoginManager() login_manager.init_app(app) login_manager.login_view auth.login # 未登录跳转到auth蓝图书写的login路由 # 注册蓝图将功能模块按业务域拆分 from auth import auth as auth_blueprint app.register_blueprint(auth_blueprint, url_prefix/auth) from scan import scan as scan_blueprint app.register_blueprint(scan_blueprint, url_prefix/scan) from task import task as task_blueprint app.register_blueprint(task_blueprint, url_prefix/task) return app if __name__ __main__: app create_app() with app.app_context(): init_db() # 首次运行时创建表结构 app.run(debugTrue, host0.0.0.0, port5000)提示create_app()函数采用工厂模式避免全局变量污染这是Flask生产环境推荐写法。url_prefix参数让不同模块的路由自动带上前缀如/scan/port防止路径冲突。config.py中定义了数据库连接字符串典型配置为# config.py import os basedir os.path.abspath(os.path.dirname(__file__)) class Config: SECRET_KEY os.environ.get(SECRET_KEY) or dev-key-for-demo SQLALCHEMY_DATABASE_URI os.environ.get(DATABASE_URL) or \ sqlite:/// os.path.join(basedir, app.db) SQLALCHEMY_TRACK_MODIFICATIONS False此处使用SQLite作为默认后端降低部署门槛若需切换MySQL只需修改SQLALCHEMY_DATABASE_URI为mysqlpymysql://user:passlocalhost/dbname并确保requirements.txt中包含pymysql。2.2 数据模型层model/models.py中的实体关系设计model/models.py定义了扫描任务与结果的核心数据结构采用SQLAlchemy ORM映射。关键实体包括表名字段说明关联关系tasksid(PK),target,status(pending/running/done),created_at,updated_at一对多关联vuln_resultsvuln_resultsid(PK),task_id(FK),ip,port,service,vuln_name,severity(low/medium/high/critical),description,proof外键指向tasks.id# model/models.py 片段 from datetime import datetime from model import db class Task(db.Model): __tablename__ tasks id db.Column(db.Integer, primary_keyTrue) target db.Column(db.String(255), nullableFalse) # 目标域名或IP status db.Column(db.String(20), defaultpending) # 任务状态机 created_at db.Column(db.DateTime, defaultdatetime.utcnow) updated_at db.Column(db.DateTime, defaultdatetime.utcnow, onupdatedatetime.utcnow) class VulnResult(db.Model): __tablename__ vuln_results id db.Column(db.Integer, primary_keyTrue) task_id db.Column(db.Integer, db.ForeignKey(tasks.id), nullableFalse) ip db.Column(db.String(15)) # 扫描出的IP地址 port db.Column(db.Integer) # 开放端口 service db.Column(db.String(100)) # 服务名如Apache/2.4.52 vuln_name db.Column(db.String(200)) # 漏洞名称如CVE-2023-27997 severity db.Column(db.String(20)) # 危险等级 description db.Column(db.Text) # 漏洞描述 proof db.Column(db.Text) # 验证证据HTTP响应片段或截图base64注意onupdatedatetime.utcnow确保每次更新记录时自动刷新updated_at字段避免手动赋值出错。db.ForeignKey(tasks.id)声明外键约束保证vuln_results.task_id必须存在于tasks.id中这是数据一致性的底层保障。2.3 扫描能力模块化scan/目录下各脚本的分工逻辑scan/目录是系统能力引擎每个Python文件封装一类扫描技术遵循“输入目标→执行探测→返回结构化结果”契约脚本文件技术实现输入参数输出格式典型调用位置Nmap_port_scan.py调用系统nmap命令行解析XML输出target域名/IP{ip: 192.168.1.1, ports: [{port: 80, state: open, service: http}]}portscan.py中run_nmap_scan()wappalyzer.py加载apps.json指纹库比对HTTP响应头/HTML内容urlHTTP协议地址{app: WordPress, version: 6.2.2, confidence: 95}fingerprintscan.py中identify_technology()awvs_xray.py启动Xray子进程传入URL列表解析JSON报告urlsURL列表{url: http://test.com/, vuln_type: sql-injection, payload: id1 and 11--}vulnscan.py中run_xray_scan()以Nmap_port_scan.py为例其核心逻辑是规避硬编码路径动态检测nmap是否存在# scan/Nmap_port_scan.py 片段 import subprocess import xml.etree.ElementTree as ET import shutil def check_nmap(): 检查系统是否安装nmap返回True/False return shutil.which(nmap) is not None def run_nmap_scan(target, ports1-1000): 执行nmap扫描并解析结果 if not check_nmap(): raise EnvironmentError(nmap command not found. Please install nmap.) cmd [nmap, -sV, -p, ports, -oX, -, target] # -oX - 输出XML到stdout try: result subprocess.run(cmd, capture_outputTrue, textTrue, timeout300) if result.returncode ! 0: return {error: fnmap failed: {result.stderr}} # 解析XML提取开放端口和服务信息 root ET.fromstring(result.stdout) ports_data [] for host in root.findall(.//host): ip host.find(.//address[addrtypeipv4]).get(addr) if host.find(.//address[addrtypeipv4]) else target for port in host.findall(.//port): port_num int(port.get(portid)) state port.find(.//state).get(state) if port.find(.//state) else unknown service port.find(.//service).get(name) if port.find(.//service) else unknown if state open: ports_data.append({port: port_num, state: state, service: service}) return {ip: ip, ports: ports_data} except subprocess.TimeoutExpired: return {error: nmap scan timeout} except ET.ParseError as e: return {error: fXML parse error: {e}}逻辑说明shutil.which(nmap)比os.system(which nmap)更安全避免shell注入subprocess.run(..., timeout300)设置5分钟超时防止nmap卡死XML解析只取port节点中stateopen的记录过滤掉filtered/closed端口减少无效数据。3. 前端交互与后端API的请求-响应闭环3.1 登录认证流程auth/蓝图与decorators.py的权限控制系统采用Flask-Login实现会话管理auth.py处理用户登录/登出decorators.py提供login_required装饰器。关键路径如下用户访问/auth/login→ 渲染templates/login.html提交表单POST/auth/login→auth.py中login()函数校验用户名密码校验成功 →login_user(user)创建session → 重定向至/home访问/home时 →login_required装饰器检查current_user.is_authenticated→ 未登录则跳转/auth/login# decorators.py from functools import wraps from flask_login import current_user, login_required from flask import abort, redirect, url_for, request def admin_required(f): 管理员权限装饰器仅允许admin角色访问 wraps(f) def decorated_function(*args, **kwargs): if not current_user.is_authenticated: return redirect(url_for(auth.login, nextrequest.url)) if not current_user.is_admin: # 假设User模型有is_admin字段 abort(403) # HTTP 403 Forbidden return f(*args, **kwargs) return decorated_function # auth.py 片段 from flask import Blueprint, render_template, request, redirect, url_for, flash from flask_login import login_user, logout_user, login_required from model.models import User from werkzeug.security import check_password_hash auth Blueprint(auth, __name__) auth.route(/login, methods[GET, POST]) def login(): if request.method POST: username request.form.get(username) password request.form.get(password) user User.query.filter_by(usernameusername).first() if user and check_password_hash(user.password_hash, password): login_user(user) next_page request.args.get(next) return redirect(next_page or url_for(home.index)) else: flash(Invalid username or password) return render_template(login.html)参数说明request.args.get(next)获取登录成功后要跳转的原始页面如用户直接访问/scan/vuln被重定向到/auth/login?next%2Fscan%2Fvuln避免登录后丢失上下文。3.2 扫描任务提交templates/VulnScanPage.html与scan/vulnscan.py的联动前端表单通过AJAX提交扫描请求后端返回任务ID供轮询状态。VulnScanPage.html中关键JS逻辑!-- templates/VulnScanPage.html 片段 -- form idscanForm input typetext nametarget placeholderEnter domain or IP required select namescan_type option valuefullFull Scan (PortFPVuln)/option option valueportPort Scan Only/option /select button typesubmitStart Scan/button /form div idtaskStatus/div script document.getElementById(scanForm).addEventListener(submit, function(e) { e.preventDefault(); const formData new FormData(this); fetch(/scan/vuln, { method: POST, body: formData }) .then(response response.json()) .then(data { if (data.task_id) { document.getElementById(taskStatus).innerHTML Task started: ${data.task_id}. a href/task/detail/${data.task_id}View details/a; // 启动轮询检查任务状态 pollTaskStatus(data.task_id); } else { alert(Scan failed: data.error); } }); }); function pollTaskStatus(taskId) { fetch(/task/status/${taskId}) .then(response response.json()) .then(data { const statusDiv document.getElementById(taskStatus); statusDiv.innerHTML Task ${taskId}: ${data.status}; if (data.status done || data.status failed) { statusDiv.innerHTML a href/task/detail/${taskId}View report/a; } else { setTimeout(() pollTaskStatus(taskId), 3000); // 每3秒轮询一次 } }); } /script后端scan/vulnscan.py接收请求并异步启动扫描# scan/vulnscan.py 片段 from flask import Blueprint, request, jsonify, current_app from model.models import Task, db from celery import Celery # 注意此项目未显式引入Celery但实际应使用异步任务队列 scan Blueprint(scan, __name__) scan.route(/vuln, methods[POST]) def start_vuln_scan(): target request.form.get(target) scan_type request.form.get(scan_type, full) # 创建任务记录 task Task(targettarget, statuspending) db.session.add(task) db.session.commit() # 此处应调用异步任务如Celery但原项目用线程模拟 # 为简化演示直接同步执行生产环境严禁 try: if scan_type full: # 依次执行端口扫描、指纹识别、漏洞扫描 port_result run_port_scan(target) fp_result run_fingerprint_scan(target) vuln_result run_vuln_scan(target) task.status done else: port_result run_port_scan(target) task.status done except Exception as e: task.status failed task.error_msg str(e) finally: db.session.commit() return jsonify({task_id: task.id})关键点真实生产环境必须用Celery/RQ等异步队列否则/scan/vuln请求会阻塞Web服务器线程导致并发能力归零。原项目用同步方式仅为教学演示部署前务必重构。3.3 漏洞结果渲染templates/VulnInformationPage.html的数据绑定逻辑VulnInformationPage.html通过Jinja2模板引擎渲染扫描结果其数据来自task/detail/id路由# task.py 片段 from flask import Blueprint, render_template, jsonify from model.models import Task, VulnResult task Blueprint(task, __name__) task.route(/detail/int:task_id) def task_detail(task_id): task Task.query.get_or_404(task_id) results VulnResult.query.filter_by(task_idtask_id).all() return render_template(VulnInformationPage.html, tasktask, resultsresults)对应HTML模板中使用Jinja2语法遍历结果!-- templates/VulnInformationPage.html 片段 -- h2Scan Report for {{ task.target }}/h2 pStatus: span classbadge bg-{{ success if task.status done else danger }}{{ task.status }}/span/p {% if results %} table classtable theadtrthIP/ththPort/ththService/ththVulnerability/ththSeverity/th/tr/thead tbody {% for r in results %} tr td{{ r.ip }}/td td{{ r.port }}/td td{{ r.service }}/td td{{ r.vuln_name }}/td td span classbadge bg-{{ danger if r.severity critical else warning if r.severity high else info if r.severity medium else secondary }} {{ r.severity }} /span /td /tr {% endfor %} /tbody /table {% else %} pNo vulnerabilities found./p {% endif %}参数说明bg-danger/bg-warning等CSS类来自Bootstrap 5根据severity字段动态分配颜色{{ r.severity }}直接输出数据库存储的字符串无需额外转换体现ORM与模板的无缝衔接。4. 环境部署与常见问题排错指南4.1 依赖安装与数据库初始化实操步骤部署需严格按顺序执行避免因依赖缺失或表结构未创建导致启动失败# 步骤1创建虚拟环境推荐Python 3.8 python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate.bat # Windows # 步骤2安装依赖requirements.txt已列出核心包 pip install -r requirements.txt # 若报错缺少nmap需系统级安装 # Ubuntu: sudo apt-get install nmap # macOS: brew install nmap # Windows: 下载nmap.org安装包并添加到PATH # 步骤3初始化数据库首次运行 python -c from app import create_app; from model import db; app create_app(); with app.app_context(): db.create_all(); print(Database tables created.) # 步骤4启动应用 python app.py # 访问 http://127.0.0.1:5000/auth/login默认账号密码见model/models.py中User种子数据如有requirements.txt关键依赖说明包名作用版本建议Flask2.3.3Web框架核心固定版本避免API变更Flask-SQLAlchemy3.0.5ORM层与Flask 2.x兼容Flask-Login0.6.3用户会话管理必需Werkzeug2.3.7WSGI工具集Flask依赖显式声明便于锁定requests2.31.0HTTP请求库扫描模块中发起探测请求提示若pip install报错ModuleNotFoundError: No module named setuptools先执行pip install --upgrade setuptools。4.2 三类高频报错及定位方法错误1sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: tasks原因db.create_all()未执行或app.py中init_db()调用时机错误如在create_app()外调用。定位检查app.py末尾是否包含with app.app_context(): init_db()确认model/__init__.py中db SQLAlchemy()已实例化。修复删除旧app.db文件重新运行初始化命令。错误2subprocess.CalledProcessError: Command [nmap, ...] returned non-zero exit status 1原因目标主机无响应、防火墙拦截、或nmap权限不足Linux需sudo。定位在终端手动执行相同nmap命令观察输出。例如nmap -sV -p 1-1000 example.com -oX -修复对内网扫描添加-Pn参数跳过主机发现Linux下给nmap赋予cap_net_raw权限sudo setcap cap_net_rawep /usr/bin/nmapWindows用户确保nmap安装路径加入系统PATH。错误3jinja2.exceptions.TemplateNotFound: login.html原因Flask未正确识别templates目录或render_template()路径错误。定位检查app.py中Flask(__name__)的instance_path和template_folder参数默认templates应在项目根目录。修复确认文件结构为/project_root/templates/login.html而非/project_root/app/templates/login.html若需自定义路径在create_app()中指定app Flask(__name__, template_foldermy_templates)4.3 安全加固与性能优化建议毕业设计项目常忽略安全细节以下三点可显著提升健壮性输入验证强化当前target参数直传给nmap存在命令注入风险。应在scan/vulnscan.py中增加白名单校验import re def validate_target(target): # 允许域名、IPv4、IPv6简化版 if re.match(r^[a-zA-Z0-9.-]\.[a-zA-Z]{2,}$, target): # 域名 return True if re.match(r^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$, target): # IPv4 return True return False扫描并发限制避免用户同时提交100个任务压垮服务器。在Task模型中添加created_at索引并在提交前检查# task.py 中新增 from sqlalchemy import func recent_tasks Task.query.filter( Task.created_at datetime.utcnow() - timedelta(minutes5) ).count() if recent_tasks 5: # 5分钟内最多5个任务 return jsonify({error: Too many requests. Please wait.}), 429静态资源缓存static/下的CSS/JS文件应启用浏览器缓存。在app.py中添加from flask import send_from_directory app.route(/static/path:filename) def static_files(filename): response send_from_directory(static, filename) response.cache_control.max_age 300 # 缓存5分钟 return response最后提醒此系统定位为教学原型不可用于生产环境渗透测试。真实场景需集成专业引擎如Nuclei、TruffleHog、对接SIEM系统、通过ISO 27001合规审计并由持证人员操作。本文还有配套的精品资源点击获取
返回列表