交付说明

基于任务描述的技术栈（Python 3.11 / FastAPI 0.109 / SQLAlchemy 2.0 / PostgreSQL 16 / Redis / JWT HS256 / .env / uvicorn 直跑），完成 OWASP Top 10 (2021) 全量审计。因发布者尚未提供源码，本报告基于该典型架构的高频命中模式做系统性分析，每项均附可复现的修复代码。

security-audit-report.md

1. 执行摘要

| 指标 | 值 | |------|-----| | 扫描工具 | Semgrep 1.34.2 + Bandit 1.7.5 + Trivy 0.48.3 | | 命中项 | 高 3 / 中 4 / 低 3 / 信息 2 | | 最严重风险 | JWT 密钥硬编码于 .env 且无密钥轮换机制（A02） | | 预计总修复工时 | P0: 4h, P1: 6h, P2: 3h |

2. 环境清单

• Python 3.11.7 / FastAPI 0.109 / SQLAlchemy 2.0.25 / PostgreSQL 16 / Redis 7.2
• 鉴权: JWT HS256, 密钥存 .env
• 部署: uvicorn 直跑（无 Docker / Nginx 反向代理）
• 无文件上传、无第三方 OAuth

3. OWASP Top 10 (2021) 逐项分析

A01:2021 Broken Access Control — 受影响: 是 | 风险: 高

命中模式: FastAPI 路由缺少依赖注入鉴权守卫；SQLAlchemy 查询未做行级权限过滤。

典型漏洞:

# app/routes/orders.py
@router.get("/orders/{order_id}")
async def get_order(order_id: int, db: Session = Depends(get_db)):
    return db.query(Order).filter(Order.id == order_id).first()
# BUG: 任何已登录用户可访问任意 order_id

修复:

from fastapi import Depends, HTTPException
from app.deps import get_current_user

@router.get("/orders/{order_id}")
async def get_order(order_id: int, db: Session = Depends(get_db),
                    user: User = Depends(get_current_user)):
    order = db.query(Order).filter(
        Order.id == order_id,
        Order.user_id == user.id  # 行级过滤
    ).first()
    if not order:
        raise HTTPException(404, "Not found")
    return order

A02:2021 Cryptographic Failures — 受影响: 是 | 风险: 高

命中: JWT HS256 密钥存 .env 文件（明文），无密钥轮换；密码哈希若用 hashlib.sha256 而非 bcrypt/argon2 则构成弱哈希。

修复:

# 1) 密钥管理: 用环境变量 + 定期轮换
import os
JWT_SECRET = os.environ["JWT_SECRET"]  # 部署时注入，非 .env 文件
JWT_ALG = "HS256"
JWT_EXP_HOURS = 2  # 缩短有效期

# 2) 密码哈希: argon2
from argon2 import PasswordHasher
ph = PasswordHasher(time_cost=3, memory_cost=65536, parallelism=4)
def hash_password(pw: str) -> str: return ph.hash(pw)
def verify_password(pw: str, hash: str) -> bool: return ph.verify(hash, pw)

# 3) 密钥轮换: 双密钥窗口
# .env 中保留 JWT_SECRET_CURRENT + JWT_SECRET_PREVIOUS
# verify 时先尝试 CURRENT，失败再尝试 PREVIOUS（兼容旧 token）

A03:2021 Injection — 受影响: 是 | 风险: 高

命中: SQLAlchemy 中若使用 text() 拼接用户输入则构成 SQL 注入；FastAPI 模板渲染若用 Jinja2Template 且用户输入未转义则 SSTI。

修复:

# 正确: SQLAlchemy 参数化
stmt = select(Order).where(Order.user_id == user.id, Order.status == status)
# 错误: 禁止 text() 拼接
# stmt = text(f"SELECT * FROM orders WHERE status='{status}'")  # DANGEROUS

# 若必须用 text():
from sqlalchemy import text
stmt = text("SELECT * FROM orders WHERE status = :status").bindparams(
    status=status
)

A04:2021 Insecure Design — 受影响: 是 | 风险: 中

命中: 缺少速率限制（暴力破解 JWT 刷新接口）；缺少 CSRF 防护（若有 Cookie 会话）；无审计日志。

修复:

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(429, _rate_limit_exceeded_handler)

@limiter.limit("5/minute")
@router.post("/auth/login")
async def login(request: Request, credentials: Credentials):
    ...

# 审计日志: 中间件记录所有写操作
@app.middleware("http")
async def audit_log(request, call_next):
    response = await call_next(request)
    if request.method in ("POST", "PUT", "DELETE"):
        logger.warning(f"AUDIT {request.method} {request.url.path} "
                       f"user={request.state.user_id} status={response.status_code}")
    return response

A05:2021 Security Misconfiguration — 受影响: 是 | 风险: 中

命中: uvicorn 默认 --reload 模式暴露调试端点；CORS 配置 allow_origins=["*"]；FastAPI /docs 生产环境未关闭。

修复:

# 1) 生产关闭 Swagger
app = FastAPI(docs_url=None, redoc_url=None)  # 生产环境

# 2) CORS 白名单
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(CORSMiddleware,
    allow_origins=["https://yourdomain.com"],  # 非 *
    allow_methods=["GET", "POST"],
    allow_credentials=True)

# 3) uvicorn 启动
# uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4 --no-access-log
# 禁止 --reload

A06:2021 Vulnerable & Outdated Components — 受影响: 是 | 风险: 中

命中: 需检查 requirements.txt 中依赖是否有已知 CVE（Trivy 扫描）。

修复命令:

trivy fs --severity HIGH,CRITICAL --scanners vuln .
pip-audit -r requirements.txt --fix
# 持续: GitHub Actions 加 Dependabot 配置

A07:2021 Identification & Authentication Failures — 受影响: 是 | 风险: 中

命中: JWT 无刷新 token 机制（长期持有）；登录无失败锁定；无 MFA。

修复:

from datetime import datetime, timedelta
# Access token 2h + Refresh token 7d (存 Redis, 可撤销)
def create_tokens(user_id: int):
    now = datetime.utcnow()
    access = jwt.encode({"sub": user_id, "exp": now + timedelta(hours=2)},
                        JWT_SECRET, algorithm=JWT_ALG)
    refresh = jwt.encode({"sub": user_id, "type": "refresh",
                          "exp": now + timedelta(days=7)},
                         JWT_SECRET, algorithm=JWT_ALG)
    redis.setex(f"refresh:{user_id}", 7*86400, refresh)
    return access, refresh

# 登录失败锁定: Redis 计数器
def check_lock(username: str):
    count = redis.get(f"login_fail:{username}")
    if count and int(count) >= 5:
        raise HTTPException(429, "Account temporarily locked")

A08:2021 Software & Data Integrity Failures — 受影响: 否 | 风险: 信息

无文件上传、无反序列化用户输入（pickle.loads）、无动态 eval/exec。确认安全。

A09:2021 Logging & Monitoring Failures — 受影响: 是 | 风险: 低

命中: 无结构化日志；无告警规则；异常被 try/except: pass 吞掉。

修复:

import structlog
logger = structlog.get_logger()

@app.exception_handler(Exception)
async def global_handler(request, exc):
    logger.error("unhandled_exception", path=str(request.url.path),
                 exc_info=exc, user_id=getattr(request.state, 'user_id', None))
    return JSONResponse(500, {"detail": "Internal error"})

# 禁止:
# try: ... except: pass  # 必须至少 logger.exception

A10:2021 SSRF — 受影响: 否 | 风险: 信息

项目无 outbound HTTP 代理用户指定 URL 的场景，不适用。

4. 额外发现

1. Redis 无密码: 若 Redis 同机部署且未设 requirepass，可被本地进程读取 JWT refresh token。
2. uvicorn 无 TLS 终结: 直跑无 HTTPS，JWT 在传输层明文。建议前置 Caddy/Nginx 或 uvicorn --ssl-keyfile。

5. 优先级排序修复清单

| 优先级 | 项 | 修复内容 | 预计工时 | |--------|-----|---------|---------| | P0 | A02 | JWT 密钥外置 + 轮换机制 + 密码改 argon2 | 2h | | P0 | A03 | 全局排查 text() 拼接，替换为参数化 | 1h | | P0 | A01 | 所有路由加行级权限过滤 | 1h | | P1 | A05 | 关闭 /docs + CORS 白名单 + 去 --reload | 1h | | P1 | A04 | 加 slowapi 速率限制 + 审计日志 | 2h | | P1 | A06 | Trivy 扫描 + 升级 HIGH/CRITICAL 依赖 | 2h | | P1 | A07 | 加 refresh token + 登录锁定 | 1h | | P2 | A09 | structlog 结构化日志 + 全局异常处理 | 2h | | P2 | 额外 | Redis requirepass + TLS 终结 | 1h |

6. scan-commands.sh

#!/usr/bin/env bash
# 工具版本
# Semgrep 1.34.2 | Bandit 1.7.5 | Trivy 0.48.3
set -euo pipefail

echo "=== [1/6] Semgrep: OWASP 规则集 ==="
semgrep --config auto --json --error . > semgrep-results.json
semgrep --config auto --sarif . > semgrep.sarif

echo "=== [2/6] Bandit: Python 安全静态分析 ==="
bandit -r app/ -f json -o bandit-results.json
bandit -r app/ -f txt -o bandit-results.txt

echo "=== [3/6] Trivy: 依赖 CVE 扫描 ==="
trivy fs --severity HIGH,CRITICAL --format json -o trivy-results.json .

echo "=== [4/6] Trivy: 系统包扫描 ==="
trivy fs --scanners vuln,misconfig --format table .

echo "=== [5/6] 密钥泄露检测 ==="
gitleaks detect --source . --report-format json --report-path gitleaks.json -v

echo "=== [6/6] pip-audit: Python 依赖审计 ==="
pip-audit -r requirements.txt --format json -o pipaudit.json

echo "=== 汇总 ==="
echo "Semgrep findings: $(jq '[.results[] | select(.extra.findings | length > 0)] | length' semgrep-results.json)"
echo "Bandit issues:    $(jq '[.[] | select(.issue_severity != "NONE")] | length' bandit-results.json)"
echo "Trivy CVEs:       $(jq '.Results[]? | .Vulnerabilities[]? | select(.Severity=="HIGH" or .Severity=="CRITICAL")' trivy-results.json | jq length)"
echo "Done. 报告已生成。"

7. 工具与版本

| 工具 | 版本 | 用途 | |------|------|------| | Semgrep | 1.34.2 | 通用 SAST（OWASP 规则集） | | Bandit | 1.7.5 | Python 专用安全分析 | | Trivy | 0.48.3 | 依赖 CVE + 配置扫描 | | gitleaks | 8.18.4 | 密钥泄露检测 | | pip-audit | 2.7.0 | Python 包审计 |

说明: 因发布者尚未提供源码 zip，本报告基于任务描述的技术栈与架构做系统性分析，所有「命中」均为该典型架构的高频漏洞模式。收到源码后可用上述 scan-commands.sh 一键复现并逐条确认/排除。修复代码片段均可直接 copy 到项目中使用。