Web 安全基础

官方文档：https://owasp.org/www-project-top-ten/
适用版本：OWASP Top 10 2021（2026-05-07 核实）

OWASP Top 10 概览

XSS（跨站脚本攻击）

攻击类型

防御措施

1. HTML 转义（服务端渲染必须）：

import html

user_input = '<script>alert("xss")</script>'
safe_output = html.escape(user_input)
# 输出：&lt;script&gt;alert(&quot;xss&quot;)&lt;/script&gt;

2. CSP（Content Security Policy）：

# FastAPI 添加 CSP 头
from fastapi import Response

@app.middleware("http")
async def add_security_headers(request, call_next):
    response = await call_next(request)
    response.headers["Content-Security-Policy"] = (
        "default-src 'self'; "
        "script-src 'self' 'nonce-{nonce}'; "
        "style-src 'self' 'unsafe-inline';"
    )
    return response

3. HttpOnly Cookie：防止 JS 读取会话 Cookie：

response.set_cookie(
    key="session_id",
    value=session_token,
    httponly=True,   # 禁止 JS 访问
    secure=True,     # 仅 HTTPS 传输
    samesite="strict"
)

4. 前端框架防护：

• React：dangerouslySetInnerHTML 要经过 DOMPurify.sanitize()
• Vue：v-html 同上，避免直接渲染用户内容

SQL 注入

攻击原理

# 危险写法：字符串拼接
user_id = request.query_params["id"]
query = f"SELECT * FROM users WHERE id = {user_id}"
# 攻击者传入：1 OR 1=1
# 实际执行：SELECT * FROM users WHERE id = 1 OR 1=1

防御：参数化查询

# SQLAlchemy ORM（天然安全）
user = await User.get(id=user_id)

# SQLAlchemy Core 参数化
from sqlalchemy import text
result = await db.execute(
    text("SELECT * FROM users WHERE id = :id"),
    {"id": user_id}
)

# aiomysql 参数化
await cursor.execute(
    "SELECT * FROM users WHERE id = %s",
    (user_id,)
)

检测工具

• sqlmap：自动化 SQL 注入检测
• 手工测试：输入 '、1 OR 1=1、1; DROP TABLE users-- 观察响应

CSRF（跨站请求伪造）

攻击原理

攻击者在第三方网站构造一个表单，诱导已登录用户提交，浏览器会自动携带目标站的 Cookie。

触发条件：

1. 用户已登录目标网站（Cookie 有效）
2. 目标 API 仅凭 Cookie 识别用户身份
3. 请求来源未验证

防御方案

1. CSRF Token（推荐）：

import secrets
from fastapi import Request, HTTPException

def generate_csrf_token() -> str:
    return secrets.token_urlsafe(32)

async def verify_csrf_token(request: Request):
    """依赖注入：验证 CSRF Token"""
    token_from_header = request.headers.get("X-CSRF-Token")
    token_from_session = request.session.get("csrf_token")
    if not token_from_header or token_from_header != token_from_session:
        raise HTTPException(status_code=403, detail="CSRF token invalid")

2. SameSite Cookie：

response.set_cookie(
    key="session_id",
    value=session_token,
    samesite="strict",  # 第三方站点请求不携带此 Cookie
)

3. 验证 Origin/Referer 头：

ALLOWED_ORIGINS = {"https://example.com", "https://www.example.com"}

async def check_origin(request: Request):
    origin = request.headers.get("Origin") or request.headers.get("Referer", "")
    if not any(origin.startswith(allowed) for allowed in ALLOWED_ORIGINS):
        raise HTTPException(status_code=403)

认证与授权

密码存储

永远不要明文存储密码，不要用 MD5/SHA1：

# 推荐：bcrypt 或 argon2
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(plain: str) -> str:
    return pwd_context.hash(plain)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

bcrypt vs argon2 对比：

JWT 安全使用

参考 JWT完全指南，关键安全要点：

OAuth2 授权码流程

1. 用户点击"第三方登录"→ 跳转到授权服务器（带 client_id + redirect_uri + state）
2. 用户在授权服务器登录并同意授权
3. 授权服务器重定向回 redirect_uri，携带 code（授权码，一次性）
4. 后端用 code + client_secret 向授权服务器换取 access_token
5. 用 access_token 调用资源服务器获取用户信息

state 参数：防止 CSRF 攻击，客户端生成随机值，回调时验证是否一致。

SSRF（服务端请求伪造）

危害

攻击者控制服务端发起的请求，可访问：

• 内网服务（数据库、Redis、内部 API）
• 云服务商元数据接口（http://169.254.169.254，获取 IAM 凭证）

防御

import ipaddress
import socket
from urllib.parse import urlparse

BLOCKED_NETWORKS = [
    ipaddress.ip_network("10.0.0.0/8"),
    ipaddress.ip_network("172.16.0.0/12"),
    ipaddress.ip_network("192.168.0.0/16"),
    ipaddress.ip_network("127.0.0.0/8"),
    ipaddress.ip_network("169.254.0.0/16"),  # 云元数据
]

def is_safe_url(url: str) -> bool:
    parsed = urlparse(url)
    if parsed.scheme not in ("http", "https"):
        return False
    try:
        ip = ipaddress.ip_address(socket.gethostbyname(parsed.hostname))
    except Exception:
        return False
    return not any(ip in net for net in BLOCKED_NETWORKS)

async def fetch_external(url: str) -> bytes:
    if not is_safe_url(url):
        raise ValueError("URL not allowed")
    async with httpx.AsyncClient() as client:
        response = await client.get(url, timeout=5)
        return response.content

文件上传安全

import uuid
import magic  # python-magic
from pathlib import Path

ALLOWED_MIME_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"}
MAX_SIZE = 5 * 1024 * 1024  # 5MB

async def save_upload(file: UploadFile) -> str:
    content = await file.read()

    # 检查大小
    if len(content) > MAX_SIZE:
        raise ValueError("File too large")

    # 检查真实 MIME 类型（读魔术字节）
    mime = magic.from_buffer(content, mime=True)
    if mime not in ALLOWED_MIME_TYPES:
        raise ValueError(f"File type not allowed: {mime}")

    # 随机文件名
    ext = {"image/jpeg": ".jpg", "image/png": ".png",
           "image/gif": ".gif", "image/webp": ".webp"}[mime]
    filename = f"{uuid.uuid4().hex}{ext}"

    upload_dir = Path("/var/uploads")  # Web 根目录之外
    upload_dir.mkdir(parents=True, exist_ok=True)
    (upload_dir / filename).write_bytes(content)

    return filename

HTTPS 配置安全

HSTS（强制 HTTPS）

# Nginx 配置
add_header Strict-Transport-Security "max-age=31536000; includeSubDomains; preload" always;

# FastAPI 中间件
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
app.add_middleware(HTTPSRedirectMiddleware)

禁用弱密码套件（Nginx）

ssl_protocols TLSv1.2 TLSv1.3;
ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384;
ssl_prefer_server_ciphers off;

安全响应头速查

最佳实践

纵深防御原则

安全不能依赖单一防线。每个层次都要设置独立防御：

输入验证三原则

1. 在服务端验证：永远不要仅依赖前端校验，所有用户输入都必须在服务端再次验证
2. 白名单而非黑名单：明确允许什么格式，而不是过滤已知危险字符
3. 最小化信任：即使是内部服务之间的请求也要验证身份

from pydantic import BaseModel, constr, validator
import re

class UserCreateRequest(BaseModel):
    username: constr(min_length=3, max_length=32, pattern=r'^[a-zA-Z0-9_]+$')
    email: str
    age: int

    @validator('email')
    def validate_email(cls, v):
        if not re.match(r'^[^@]+@[^@]+\.[^@]+$', v):
            raise ValueError('Invalid email format')
        return v.lower()

密钥与凭证管理

# 错误：硬编码密钥
SECRET_KEY = "mysecretkey123"

# 正确：从环境变量读取
import os
SECRET_KEY = os.environ["SECRET_KEY"]  # 启动时若缺失会立即报错

# 生产环境推荐：HashiCorp Vault / AWS Secrets Manager / GCP Secret Manager

• 使用 .env 文件存储本地开发密钥，.env 加入 .gitignore
• 定期轮换密钥（JWT secret、API key、数据库密码）
• 不同环境（dev/staging/prod）使用不同密钥

安全编码检查清单

常见陷阱

陷阱：SQL 拼接字符串导致注入漏洞

现象： 用户输入 ' OR '1'='1 绕过登录验证，或输入 '; DROP TABLE users; -- 删除数据。

原因： 将用户输入直接拼接进 SQL 字符串，攻击者可以改变 SQL 语义。

解决： 始终使用参数化查询（ORM 或 cursor.execute(sql, params)），永远不要用字符串格式化拼接 SQL。

# 错误：字符串拼接，存在注入风险
cursor.execute(f"SELECT * FROM users WHERE name = '{username}'")

# 正确：参数化查询
cursor.execute("SELECT * FROM users WHERE name = %s", (username,))

陷阱：生产环境开启 debug 模式暴露内部信息

现象： 接口报错时，响应体包含完整的 Python 堆栈、数据库连接字符串、本地文件路径等敏感信息。

原因： 框架的 debug 模式（FastAPI debug=True、Django DEBUG=True）会在响应中返回详细错误信息，方便开发调试，但在生产环境会暴露内部实现。

解决： 生产环境通过环境变量控制 debug 开关，错误响应只返回通用错误码和 ID，详细错误记录到日志（不包含敏感字段）。

陷阱：JWT Secret 使用弱密钥或硬编码

现象： 攻击者通过暴力破解或代码泄露获得 JWT Secret，伪造任意用户的 Token，绕过认证。

原因： 使用简短密钥（如 secret、123456）容易被字典攻击；硬编码在源码中一旦仓库泄露立即失效。

解决： JWT Secret 至少 256 位随机字节（openssl rand -hex 32），通过环境变量注入，定期轮换；对高价值接口添加额外的时效校验（不依赖 JWT exp 单点失效）。

参见

• JWT完全指南 — JWT 认证的安全实现与常见漏洞
• HTTP协议深度指南 — HTTPS、Cookie 安全属性、安全响应头详解
• FastAPI完全指南 — FastAPI 安全最佳实践（认证、权限、依赖注入）