请求协助理解并修复Flask应用中的SQL注入安全漏洞

一、什么是SQL注入？ #

SQL注入是一种常见的Web安全漏洞：攻击者通过在用户输入中插入特殊SQL语法字符，让应用执行原本预期之外的数据库操作。比如你的登录接口，如果攻击者输入用户名 ' OR '1'='1，原本的SQL语句会被拼接成：

SELECT * FROM users WHERE username='' OR '1'='1' AND password='xxx'

这条SQL会直接匹配所有用户，攻击者无需正确密码就能登录成功，甚至可能篡改、删除数据库数据。

二、防范SQL注入的核心方法 #

最有效的防御手段是使用参数化查询（预编译语句）：把SQL语句的结构和用户输入的参数分开处理。数据库会先编译固定的SQL模板，再安全地代入参数，参数里的特殊字符会被当作普通字符串处理，不会被解析成SQL语法。

对于Python的sqlite3库，用?作为参数占位符，将参数以元组形式传递给execute()方法即可。

三、你的代码漏洞修复 #

1. 风险代码问题点 #

所有使用f-string拼接SQL语句的地方都存在注入风险，比如：

cur.execute(f"SELECT * FROM users WHERE username='{username}'")

这种写法会直接把用户输入代入SQL逻辑，完全没有防护。

2. 修复后的完整代码 #

以下是安全优化后的版本，重点修改了数据库操作的参数传递方式，同时修复了连接泄漏、输入校验等问题：

from flask import Flask, request, jsonify, render_template
import hashlib
import sqlite3
import jwt
import os

app = Flask(__name__)

# 从环境变量读取JWT密钥，避免硬编码泄露
secret = os.getenv('JWT_SECRET', "$up3r_$3cur4_t0k3n")

def check_username(username):
    conn = sqlite3.connect('users.db')
    cur = conn.cursor()
    # 使用参数化查询，?为占位符，参数以元组传入
    cur.execute("SELECT * FROM users WHERE username=?", (username,))
    res = cur.fetchone()
    # 关闭连接，避免数据库连接泄漏
    conn.close()
    return res is not None

@app.route('/')
def home():
    return render_template('index.html')

@app.route('/api/v3/signup', methods=['POST'])
def signup():
    username = request.json.get('uname')
    password = request.json.get('pwd')
    
    # 基础输入校验，拦截无效或恶意输入
    if not username or len(username) > 20:
        return jsonify({"message":"Invalid username"}), 400
    if not password or len(password) < 6:
        return jsonify({"message":"Password must be at least 6 characters"}), 400
    
    if check_username(username):
        return jsonify({"message":"User already exists"}), 200
    
    conn = sqlite3.connect('users.db')
    cur = conn.cursor()
    # 参数化插入，避免注入风险
    hashed_pwd = hashlib.sha256(password.encode()).hexdigest()
    cur.execute(
        "INSERT INTO users (username, password, isAdmin) VALUES (?, ?, 0)",
        (username, hashed_pwd)
    )
    conn.commit()
    conn.close()
    return jsonify({"message":"User Created Successfully."}), 200

@app.route('/api/v3/login', methods=['POST'])
def login():
    username = request.json.get('uname')
    password = request.json.get('pwd')
    
    if not username or not password:
        return jsonify({"message":"Missing username or password"}), 400
    
    conn = sqlite3.connect('users.db')
    cur = conn.cursor()
    # 参数化查询，避免注入
    hashed_pwd = hashlib.sha256(password.encode()).hexdigest()
    cur.execute(
        "SELECT * FROM users WHERE username=? AND password=?",
        (username, hashed_pwd)
    )
    res = cur.fetchone()
    conn.close()
    
    if not res:
        return jsonify({"message":"Wrong Username/Password"}), 401
    
    data = {"username":res[0], "isAdmin":res[2]}
    jwt_data = jwt.encode(payload=data, key=secret, algorithm='HS256')
    return jsonify({"auth":jwt_data.decode()}), 200

@app.route('/api/v3/validate', methods=['POST'])
def validate():
    auth = request.json.get('auth')
    if not auth:
        return jsonify({"message":"Missing auth token"}), 400
    
    try:
        data = jwt.decode(auth, key=secret, algorithms=['HS256'])
    except jwt.InvalidTokenError:
        return jsonify({"message":"Invalid auth token"}), 401
    
    if data["username"] == 'admin' and data['isAdmin'] is True:
        return jsonify({"message":f"Welcome admin! Hope you're okay."})
    else:
        return jsonify({"message":f"Welcome {data['username']}!"})

if __name__ == '__main__':
    # 生产环境强制关闭debug模式，避免暴露敏感错误信息
    debug_mode = os.getenv('FLASK_DEBUG', 'False').lower() == 'true'
    app.run('0.0.0.0', 8090, debug=debug_mode)

3. 额外安全优化建议 #

• 关闭Debug模式：生产环境绝对不能开启debug=True，否则会暴露详细错误信息（包括SQL语句），给攻击者提供漏洞线索。
• 密码哈希加盐：单纯的SHA256哈希容易被彩虹表破解，建议使用带盐的哈希算法，比如werkzeug.security.generate_password_hash和check_password_hash，更安全且易用。
• 输入验证强化：可以对用户名、密码的格式做更严格的校验（比如禁止特殊字符），进一步降低恶意输入的可能性。

内容的提问来源于stack exchange，提问作者Devang Solanki