MBE 安全和优化指南

本文档介绍 MBE 系统的安全加固措施和性能优化配置。

目录


安全加固

1. CORS 配置

开发环境(允许所有源):

CORS_ORIGINS=*

生产环境(限制特定域名):

CORS_ORIGINS=https://example.com,https://app.example.com,https://admin.example.com

配置说明:

  • CORS_ORIGINS: 允许的源列表,用逗号分隔
  • 生产环境必须设置具体域名,避免 CSRF 攻击
  • 支持通配符子域名(如 https://*.example.com

2. 速率限制(Rate Limiting)

自动防止 API 滥用和 DDoS 攻击。

配置:

ENABLE_RATE_LIMIT=true

默认限制规则:

路径前缀 限制 时间窗口
/api/chat 20次 60秒
/api/users 100次 60秒
/api/ 200次 60秒
/ 500次 60秒

客户端识别优先级:

  1. 用户ID(已认证用户)
  2. API Key
  3. IP 地址(默认)

响应 Header:

X-RateLimit-Limit: 200
X-RateLimit-Remaining: 185
Retry-After: 30  # 超限时返回

自定义限制规则:

编辑 shared/src/utils/rate_limit.py:

RATE_LIMIT_RULES = {
    "/api/chat": (50, 60),  # 50次/分钟
    "/api/admin": (1000, 60),  # 管理员更高限制
}

3. 请求验证和清理

自动检测和阻止恶意输入(SQL注入、XSS等)。

配置:

ENABLE_REQUEST_VALIDATION=true
MAX_REQUEST_BODY_SIZE=10485760  # 10MB

防护内容:

  • ✅ SQL 注入(SELECT, UNION, --, etc.)
  • ✅ XSS 攻击(<script>, javascript:, etc.)
  • ✅ 路径遍历(../
  • ✅ 请求体大小限制

使用示例:

from utils.security import sanitize_input, validate_email, validate_username

# 清理用户输入
clean_text = sanitize_input(user_input)

# 严格模式(只允许字母数字)
safe_text = sanitize_input(user_input, strict=True)

# 验证邮箱
if validate_email(email):
    ...

# 验证用户名(3-20字符,字母数字下划线连字符)
if validate_username(username):
    ...

4. 环境变量敏感信息

永远不要将敏感信息硬编码到代码中!

使用环境变量管理敏感配置:

# .env (不要提交到 Git)
SECRET_KEY=your-secret-key-change-in-production
POSTGRES_PASSWORD=strong_password_here
REDIS_PASSWORD=strong_password_here

# LLM API Keys
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...

# 支付配置
WECHAT_PAY_MCH_ID=...
ALIPAY_APP_ID=...

GitHub Secrets 配置:

在 GitHub 仓库设置中添加 Secrets:

  1. Settings -> Secrets and variables -> Actions
  2. New repository secret
  3. 添加所有敏感变量

在 GitHub Actions 中使用:

env:
  SECRET_KEY: ${{ secrets.SECRET_KEY }}
  POSTGRES_PASSWORD: ${{ secrets.POSTGRES_PASSWORD }}

5. HTTPS/SSL 配置

生产环境必须使用 HTTPS。

Nginx 配置示例:

server {
    listen 80;
    server_name api.example.com;
    return 301 https://$server_name$request_uri;
}

server {
    listen 443 ssl http2;
    server_name api.example.com;

    # SSL 证书(使用 Let's Encrypt)
    ssl_certificate /etc/letsencrypt/live/api.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/api.example.com/privkey.pem;

    # SSL 优化
    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_ciphers HIGH:!aNULL:!MD5;
    ssl_prefer_server_ciphers on;
    ssl_session_cache shared:SSL:10m;

    # 安全 Header
    add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;
    add_header X-Frame-Options "SAMEORIGIN" always;
    add_header X-Content-Type-Options "nosniff" always;
    add_header X-XSS-Protection "1; mode=block" always;

    location / {
        proxy_pass http://mbe-api:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        
        # 速率限制(Nginx 层面)
        limit_req zone=api_limit burst=20 nodelay;
    }
}

# 速率限制配置
limit_req_zone $binary_remote_addr zone=api_limit:10m rate=100r/s;

性能优化

1. 数据库连接池优化

当前配置:

# shared/src/storage/database.py
engine = create_async_engine(
    db_url,
    pool_size=20,          # 基础连接池大小
    max_overflow=40,       # 最大溢出连接
    pool_timeout=30,       # 获取连接超时(秒)
    pool_recycle=3600,     # 1小时回收连接
    pool_pre_ping=True,    # 连接前测试
)

调整建议:

并发量 pool_size max_overflow
< 100 RPS 10 20
100-500 RPS 20 40
500-1000 RPS 50 100
> 1000 RPS 100 200

计算公式:

pool_size = (CPU 核心数 * 2) + 磁盘数
max_overflow = pool_size * 2

2. Redis 连接池优化

当前配置:

# shared/src/storage/redis.py
redis_pool = redis.from_url(
    settings.redis_url,
    max_connections=50,        # 连接池大小
    socket_timeout=5,          # Socket 超时
    socket_connect_timeout=5,  # 连接超时
    socket_keepalive=True,     # 保持连接
)

性能建议:

  • 高并发场景:max_connections=100+
  • 启用连接池复用
  • 使用 Redis Pipeline 批量操作

Pipeline 示例:

async def batch_set_cache(data: dict):
    r = await get_redis()
    pipe = r.pipeline()
    for key, value in data.items():
        pipe.setex(key, 3600, value)
    await pipe.execute()

3. Gzip 压缩

自动压缩 API 响应,减少网络传输。

配置:

# main.py
app.add_middleware(GZipMiddleware, minimum_size=1000)  # 大于1KB才压缩

压缩效果:

  • JSON 响应:50-80% 大小减少
  • 文本内容:60-90% 大小减少

注意:

  • 小文件(<1KB)不压缩(开销大于收益)
  • 已压缩文件(图片、视频)不再压缩

4. 缓存策略

用户画像缓存:

from storage.redis import UserCache

# 缓存用户画像(24小时)
await UserCache.set_profile(user_id, profile, expire_hours=24)

# 读取缓存
profile = await UserCache.get_profile(user_id)
if not profile:
    # 缓存未命中,从数据库加载
    profile = await load_from_db(user_id)
    await UserCache.set_profile(user_id, profile)

API 响应缓存(使用装饰器):

from functools import lru_cache

@lru_cache(maxsize=1000)
def get_expensive_data(key: str):
    # 昂贵的计算或查询
    return compute_data(key)

建议:

  • 热点数据:缓存 1-24 小时
  • 配置数据:缓存 1 周
  • 用户会话:缓存 30 分钟

5. 数据库查询优化

使用索引:

# models.py
class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, index=True)  # 索引
    created_at = Column(DateTime, index=True)         # 索引

避免 N+1 查询(使用 joinedload):

from sqlalchemy.orm import joinedload

# 错误:N+1 查询
users = await db.execute(select(User))
for user in users:
    conversations = user.conversations  # 每次都查询数据库!

# 正确:一次查询
stmt = select(User).options(joinedload(User.conversations))
users = await db.execute(stmt)

分页查询:

# 使用 limit 和 offset
stmt = select(User).limit(20).offset(page * 20)
users = await db.execute(stmt)

Docker 优化

1. 多阶段构建

使用多阶段构建减少镜像大小。

优化前: 13GB(包含编译工具)
优化后: ~2-3GB(只包含运行时)

Dockerfile.api(多阶段):

# Stage 1: Builder
FROM python:3.11-slim as builder
WORKDIR /build
RUN apt-get update && apt-get install -y build-essential gcc g++
COPY */requirements.txt ./
RUN pip install --user -r requirements.txt

# Stage 2: Runtime
FROM python:3.11-slim
WORKDIR /app
# 只安装运行时依赖
RUN apt-get update && apt-get install -y libpq5
# 从 builder 复制已安装的包
COPY --from=builder /root/.local /root/.local
COPY src/ ./src/
CMD ["uvicorn", "main:app"]

2. .dockerignore

排除不需要的文件,加快构建速度。

# .dockerignore
__pycache__
*.pyc
.git
.vscode
tests/
docs/
*.md
.env
*.log
data/

3. 缓存优化

按变化频率分层复制:

# 先复制依赖文件(变化少)
COPY requirements.txt .
RUN pip install -r requirements.txt

# 再复制源代码(变化多)
COPY src/ ./src/

4. 构建优化

使用 BuildKit:

# 启用 BuildKit
export DOCKER_BUILDKIT=1

# 并行构建多个镜像
docker build --cache-from=mbe-api:latest -t mbe-api .

使用 GitHub Actions cache:

- name: Build Docker image
  uses: docker/build-push-action@v5
  with:
    cache-from: type=gha
    cache-to: type=gha,mode=max

生产环境最佳实践

安全清单

  • 设置具体的 CORS 域名(非 *
  • 启用速率限制
  • 启用请求验证
  • 使用强密码(数据库、Redis)
  • 定期更新密钥(SECRET_KEY)
  • 配置 HTTPS/SSL
  • 使用非 root 用户运行容器
  • 限制容器资源(CPU/内存)
  • 配置防火墙规则
  • 定期更新依赖(安全补丁)

性能清单

  • 优化数据库连接池
  • 优化 Redis 连接池
  • 启用 Gzip 压缩
  • 实施缓存策略
  • 添加数据库索引
  • 使用 CDN(静态资源)
  • 配置负载均衡
  • 启用 APM 监控
  • 定期分析慢查询

监控清单

  • 配置 Prometheus 指标收集
  • 配置 Grafana 仪表板
  • 配置告警规则
  • 启用慢查询监控
  • 配置日志聚合(ELK)
  • 配置 APM 追踪
  • 定期查看监控数据

参考配置

生产环境 .env 示例

# ===== 基础配置 =====
MBE_ENV=production
MBE_DEBUG=0

# ===== 数据库 =====
DATABASE_URL=postgresql+asyncpg://mbe:STRONG_PASSWORD@db-server:5432/mbe_prod

# ===== Redis =====
REDIS_URL=redis://:STRONG_PASSWORD@redis-server:6379/0

# ===== 安全 =====
SECRET_KEY=your-64-character-random-string-here
ENABLE_RATE_LIMIT=true
ENABLE_REQUEST_VALIDATION=true
CORS_ORIGINS=https://example.com,https://app.example.com
MAX_REQUEST_BODY_SIZE=5242880  # 5MB

# ===== 日志 =====
LOG_LEVEL=INFO
ENABLE_JSON_LOGS=true
LOG_FILE=/var/log/mbe/app.log

# ===== 监控 =====
ENABLE_METRICS=true

故障排查

问题:速率限制过于严格

症状: 正常用户频繁收到 429 错误

解决: 调整限制规则

# utils/rate_limit.py
RATE_LIMIT_RULES = {
    "/api/chat": (50, 60),  # 增加到 50次/分钟
}

问题:数据库连接池耗尽

症状: 日志显示 "QueuePool limit exceeded"

解决: 增加连接池大小

# storage/database.py
pool_size=50,  # 增加到50
max_overflow=100,

问题:Docker 镜像构建慢

解决:

  1. 使用 .dockerignore 排除不需要的文件
  2. 启用 BuildKit
  3. 使用构建缓存
  4. 并行构建
DOCKER_BUILDKIT=1 docker build --cache-from=mbe-api:latest .

总结

通过以上安全和优化措施,MBE 系统现在具备:

安全加固

  • CORS 白名单配置
  • 速率限制(防 DDoS)
  • 请求验证(防注入/XSS)
  • 敏感信息管理

性能优化

  • 数据库连接池优化(20+40)
  • Redis 连接池优化(50)
  • Gzip 响应压缩
  • 缓存策略(Redis)

Docker 优化

  • 多阶段构建(减少2/3体积)
  • .dockerignore(加快构建)
  • 非 root 用户
  • 健康检查

这些措施显著提升了系统的安全性、性能和可维护性!