Flask安全开发指南

1. 基础安全防护

1.1 HTML转义防护

Flask中任何用户提供的数据在渲染到HTML输出时都必须进行转义，以防止XSS攻击：

from markupsafe import escape

@app.route("/<path:name>")
def hello(name):
    return f"Hello, {escape(name)}!"

Jinja2模板默认会自动转义变量，但可以使用|safe过滤器标记可信内容：

from flask import render_template

@app.route("/wel/<name>")
def hello(name):
    return render_template('welcome.html', person=name)

模板文件：

<h1>Hello {{ person|safe }}!</h1>

1.2 文件上传安全

处理文件上传时，必须使用secure_filename()函数处理文件名：

from werkzeug.utils import secure_filename

@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
    if request.method == 'POST':
        file = request.files['the_file']
        file.save(f"/var/www/uploads/{secure_filename(file.filename)}")

完整文件上传处理示例：

import uuid
import os
from PIL import Image

def allowed_file(filename):
    return '.' in filename and filename.rsplit('.', 1)[1].lower() in current_app.config['ALLOWED_EXTENSIONS']

def is_image(file_path):
    try:
        Image.open(file_path).verify()
        return True
    except:
        return False

@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
    if request.method == 'POST':
        if 'file' not in request.files:
            flash('No file part')
            return redirect(request.url)
        file = request.files['file']
        if file.filename == '':
            flash('No selected file')
            return redirect(request.url)
        if file and allowed_file(file.filename):
            filename = secure_filename(file.filename)
            ext = filename.rsplit('.', 1)[1].lower()
            random_filename = f"{uuid.uuid4().hex}.{ext}"
            file_path = os.path.join(current_app.config['UPLOAD_FOLDER'], random_filename)
            file.save(file_path)
            if not is_image(file_path):
                os.remove(file_path)
                flash('Uploaded file is not a valid image')
                return redirect(request.url)
            flash('File uploaded successfully!')
            return redirect(url_for('index'))

1.3 安全文件下载

使用send_from_directory时，应设置as_attachment=True防止浏览器直接执行文件：

from flask import send_from_directory

@app.route('/download/<filename>', methods=['GET'])
def uploaded_file(filename):
    return send_from_directory("../"+current_app.config['UPLOAD_FOLDER'], filename, as_attachment=True)

2. 会话与认证安全

2.1 会话管理

Flask使用签名cookie实现会话，必须设置强密钥：

import secrets
app.secret_key = secrets.token_hex()  # 生成随机密钥

会话基本使用：

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        session['username'] = request.form['username']
        return redirect(url_for('index'))

@app.route('/logout')
def logout():
    session.pop('username', None)
    return redirect(url_for('index'))

2.2 会话伪造防护

Flask会话使用三段式结构：base64编码数据、时间戳、安全签名。防止伪造需要：

1. 使用强密钥
2. 定期轮换密钥
3. 不存储敏感信息在会话中

伪造会话示例：

import hashlib
from flask.json.tag import TaggedJSONSerializer
from itsdangerous import *

session = {"user_id":2}
secret = 'dev'
print(URLSafeSerializer(
    secret_key=secret,
    salt='cookie-session',
    serializer=TaggedJSONSerializer(),
    signer=TimestampSigner,
    signer_kwargs={
        'key_derivation': 'hmac',
        'digest_method': hashlib.sha1
    }
).dumps(session))

3. 数据库安全

3.1 SQL注入防护

使用参数化查询防止SQL注入：

db.execute(
    "INSERT INTO user (username, password) VALUES (?, ?)",
    (username, generate_password_hash(password))
)

3.2 密码存储

密码必须哈希存储，使用generate_password_hash和check_password_hash：

from werkzeug.security import generate_password_hash, check_password_hash

hashed_pw = generate_password_hash(password)
db.execute("INSERT INTO user (username, password) VALUES (?, ?)", (username, hashed_pw))

# 验证密码
user = db.execute("SELECT * FROM user WHERE username = ?", (username,)).fetchone()
if user and check_password_hash(user["password"], password):
    # 登录成功

3.3 并发控制

使用事务和锁防止并发问题：

@app.route('/<int:id>/like', methods=['POST'])
def like_post(id):
    db = get_db()
    try:
        # 检查是否已点赞
        like_exists = db.execute(
            "SELECT 1 FROM post_likes WHERE user_id = ? AND post_id = ?",
            (g.user['id'], id)
        ).fetchone()
        if like_exists:
            return jsonify(status='error', message='You have already liked this post.')
        
        # 插入点赞记录
        db.execute(
            "INSERT INTO post_likes (user_id, post_id) VALUES (?, ?)",
            (g.user['id'], id)
        )
        
        # 更新点赞数
        db.execute(
            "UPDATE post SET likes = likes + 1 WHERE id = ?",
            (id,)
        )
        db.commit()
        
        # 返回新点赞数
        likes = db.execute(
            "SELECT likes FROM post WHERE id = ?",
            (id,)
        ).fetchone()['likes']
        return jsonify(status='success', likes=likes)
    except Exception as e:
        db.rollback()
        return jsonify(status='error', message=str(e))

4. 跨域与CSRF防护

4.1 CSRF防护

使用Flask-WTF的CSRFProtect：

from flask_wtf import CSRFProtect
CSRFProtect(app)

模板中添加CSRF令牌：

<form method="post">
    <input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
</form>

AJAX请求添加CSRF令牌：

const csrfToken = document.querySelector('meta[name="csrf-token"]').getAttribute('content');
fetch(`${postId}/like`, {
    method: 'POST',
    headers: {
        'Content-Type': 'application/json',
        'X-CSRFToken': csrfToken
    }
})

4.2 CORS配置

安全配置CORS：

from flask_cors import CORS

# 全局配置
CORS(app, origins='http://example.com')

# 单个路由配置
@app.route('/api/some_endpoint')
@cross_origin(origins='https://localhost:5000', 
              methods=['GET', 'POST'],
              supports_credentials=True)
def some_endpoint():
    return jsonify({"message": f"Hello, {session['user_id']}!"})

危险配置（避免使用）：

# 不安全配置示例
CORS(app, resources={r"/api/*": {"origins": "*", "supports_credentials": True}})

5. 响应头与HTTPS

5.1 安全响应头

使用Flask-Talisman设置安全头：

from flask_talisman import Talisman

csp = {
    'default-src': "'self'",
    'script-src': "'self'",
    'style-src': "'self'"
}

Talisman(app, 
         force_https=True,
         force_https_permanent=True,
         content_security_policy=csp)

5.2 启用HTTPS

开发环境启用HTTPS：

flask --app app run --cert=cert.pem --key=key.pem

生成自签名证书：

openssl req -x509 -newkey rsa:4096 -nodes -out cert.pem -keyout key.pem -days 365

6. 模板安全

6.1 SSTI防护

避免使用render_template_string渲染用户提供的模板：

# 危险示例 - 存在SSTI漏洞
@app.route('/cc',methods=['GET', 'POST'])
def cc():
    template = '''<div>%s</div>''' %(request.url)
    return render_template_string(template)

攻击示例：

https://example.com/cc?{{7+8}}
https://example.com/cc?{{self.__init__.__globals__.__builtins__['__import__']('os').popen('ipconfig').read()}}

6.2 魔术方法利用

常见用于SSTI的Python魔术方法：

查找可用子类：

''.__class__.__base__.__subclasses__()
''.__class__.__mro__[-1].__subclasses__()

7. 对象存储安全

7.1 OSS安全配置

1. Bucket权限配置：

   • 避免公共读写权限
   • 禁用ListObject权限
   • 避免公开写权限

2. Bucket爆破防护：

   • 使用复杂Bucket名称
   • 监控异常访问

3. AccessKey保护：

   • 不在代码中硬编码AccessKey
   • 使用临时凭证
   • 定期轮换密钥

7.2 OSS接管风险

当Bucket显示NoSuchBucket时可能被接管：

1. 创建同名Bucket
2. 设置为公开
3. 上传恶意文件

防护措施：

• 及时删除不再使用的Bucket
• 监控Bucket创建行为

8. 错误处理与日志

8.1 自定义错误处理

防止信息泄露：

from flask import make_response

@app.errorhandler(404)
def not_found(error):
    resp = make_response(render_template('error.html'), 404)
    resp.headers['X-Something'] = 'A value'
    return resp

8.2 安全日志

记录安全相关事件：

• 登录尝试
• 敏感操作
• 异常请求

import logging
logging.basicConfig(level=logging.INFO, 
                   format='%(asctime)s - %(levelname)s - %(message)s')

@app.route('/login', methods=['POST'])
def login():
    logging.info(f"Login attempt from {request.remote_addr}")
    # ...

9. 生产环境部署

9.1 应用打包

使用pyproject.toml：

[project]
name = "flaskr"
version = "1.0.0"
dependencies = ["flask"]

[build-system]
requires = ["flit_core<4"]
build-backend = "flit_core.buildapi"

打包命令：

pip install build
python -m build --wheel

9.2 生产服务器

使用Waitress部署：

pip install waitress
waitress-serve --call flaskr:create_app

10. 安全对比表

11. 参考资源

1. Flask官方文档
2. Flask-Talisman文档
3. OSS安全最佳实践
4. SSTI防护指南