# 安全架构设计

本章详细阐述系统的安全架构设计，围绕认证、授权、防护、审计四大安全领域展开，旨在帮助开发者全面理解系统的安全保障体系，确保在开发与部署过程中遵循安全规范，有效防范各类安全风险。

**认证（Authentication）**：系统采用基于 JWT（JSON Web Token）的无状态认证机制，用户通过用户名密码登录后获取 Token，后续请求通过 `Authorization: Bearer <token>` 头传递，由 `login_required` 依赖统一拦截并验证 Token 的有效性（签名校验、有效期检查、Redis 黑名单过滤）。同时，系统内置登录失败锁定策略，连续失败 5 次即锁定账号 10 分钟，有效防御暴力破解攻击。

**授权（Authorization）**：基于 RBAC（基于角色的访问控制）模型设计，权限控制粒度细化至接口级。用户通过角色继承权限节点（如 `sys:position:add`），由 `@permission_required` 装饰器在路由层统一拦截校验，权限数据采用 Redis 缓存加速查询，管理员（`userId == 1`）自动放行。权限变更时，相关缓存主动失效，确保权限更新的实时性。

**防护（Protection）**：系统从多维度构建安全防护体系。限流中间件基于 Redis 滑动窗口算法，有效防御 CC 攻击和接口滥用；参数校验通过 Pydantic Schema 自动拦截非法输入，防止 SQL 注入和 XSS 攻击；敏感密码采用 bcrypt 加盐哈希存储，即使数据库泄露也无法还原明文；演示模式下 `@check_demo` 装饰器统一拦截写操作，保护测试数据不被污染；跨域配置精确控制可信域名，防止 CSRF 攻击。

**审计（Audit）**：系统提供完备的操作审计能力。操作日志中间件自动记录所有写操作（POST/PUT/DELETE/PATCH）的请求路径、参数、操作人、IP 地址及执行结果，存入 `tornado_operation_log` 表；登录日志独立记录每次登录尝试（含 IP、User-Agent、登录状态），便于安全事件追溯与异常行为分析；定时任务每次执行记录至 `tornado_job_log` 表，确保后台任务的执行过程可审计。

:::tip 安全设计原则
纵深防御：认证、授权、限流、参数校验多层防护，单一防线失效时仍有其他机制兜底

最小权限：用户仅能访问已授权接口，角色权限按需分配，遵循"够用即可"原则

数据不落地：敏感信息（Token、密码）不在日志中明文记录，密码哈希不可逆

安全默认：框架默认配置为安全模式（如 DEBUG 关闭、限流开启），如需放宽需显式配置
:::

## 安全架构总览

```
┌─────────────────────────────────────────────────────────────────────┐
│ 安全防御层次 │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 网络层防护 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ │
│ │ │ CORS │ │ 限流 │ │ XSS 防护 │ │ 上传体积限制 │ │ │
│ │ │ 跨域控制 │ │ 滑动窗口 │ │ 输入清洗 │ │ DoS 防御 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────┘ │ │
│ └───────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 认证层 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ │
│ │ │ JWT 认证 │ │ 验证码 │ │ 登录锁定 │ │ Token 黑名单 │ │ │
│ │ │ Bearer │ │ 图形验证 │ │ 失败计数 │ │ 登出失效 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────┘ │ │
│ └───────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 授权层 │ │
│ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │ │
│ │ │ RBAC 权限 │ │ 演示模式保护 │ │ 超级管理员跳过 │ │ │
│ │ │ permission_ │ │ check_demo │ │ user_id=1 bypass │ │ │
│ │ │ required │ │ │ │ │ │ │
│ │ └──────────────┘ └──────────────┘ └──────────────────────┘ │ │
│ └───────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 数据层 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ │
│ │ │ 密码加密 │ │ 参数校验 │ │ SQL 注入 │ │ 操作日志 │ │ │
│ │ │ bcrypt │ │ Pydantic │ │ ORM 防护 │ │ 审计追踪 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────┘ │ │
│ └───────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
```

## 一、认证机制

### JWT 令牌认证

:::tip JWT 工作流程
用户登录成功后，服务端签发 JWT Token 返回给前端。前端将 Token 存储在本地，后续每次请求通过 `Authorization: Bearer <token>` 头携带。服务端验证 Token 签名和有效期，解析出用户信息。
:::

1. 工作流程时序图

![](/volume3/security/jwt_token.png)


2. 纯文本树状时序图

```
[前端]                    [后端]                    [Redis]
   │                         │                         │
   │── POST /login ─────────→│                         │
   │   {username, pwd,       │                         │
   │    captcha, uuid}       │                         │
   │                         │── 验证码校验 (uuid) ──→│
   │                         │←─ captcha_text ────────│
   │                         │                         │
   │                         │── 密码校验             │
   │                         │── 签发JWT              │
   │                         │── 存储Token ──────────→│
   │←─ {token} ─────────────│                         │
   │                         │                         │
   │── GET /api/v1/xxx ────→│                         │
   │   Authorization:        │                         │
   │   Bearer <token>        │                         │
   │                         │── 验证JWT              │
   │                         │── 检查黑名单 ─────────→│
   │                         │←─ exists ─────────────│
   │←─ {data} ──────────────│                         │
```

:::tip 流程说明
1. 前端提交用户名、密码、验证码及 UUID
2. 后端校验验证码，验证密码，生成 JWT 并缓存至 Redis
3. 前端后续请求携带 JWT，后端校验 Token 有效性并检查黑名单
4. 校验通过后返回业务数据
:::

#### JWT 配置

```python
# src/config/auth.py
# ============================================================
# JWT 配置
# ============================================================
# JWT签名密钥（必须通过环境变量设置，不能使用默认值上线）
JWT_SECRET = os.getenv('JWT_SECRET', '')
# JWT令牌过期时间（分钟）
JWT_EXPIRE_MINUTES = int(os.getenv('JWT_EXPIRE_MINUTES', '20'))
JWT_ALGORITHM = "HS256" # 签名算法
```

#### Token 签发

```python
# src/core/jwt.py
def create_token(payload, timeout=None):
    """
    生成JWT令牌
    =============================================================

    根据传入的载荷数据生成JWT访问令牌，自动添加过期时间。
    使用强密钥确保令牌安全性，符合RFC 7518安全规范。

    参数:
        payload (dict): JWT载荷数据，通常包含用户ID、用户名等信息
        timeout (int): 令牌过期时间（分钟），默认使用DEFAULT_TIMEOUT_MINUTES

    返回:
        str: 生成的JWT令牌字符串

    使用示例:
        >>> payload = {'userId': 1001, 'username': 'admin'}
        >>> token = create_token(payload, timeout=30)
        >>> print(token)
        eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...

    注意事项:
        1. 密钥长度必须至少32字节，否则会发出警告
        2. 生产环境建议通过环境变量设置JWT_SECRET
        3. 令牌过期后需要重新登录获取新令牌
    """
    try:
        # 设置过期时间
        # 说明：在载荷中添加exp字段，指定令牌的过期时间点
        # 计算方式：当前UTC时间 + 指定的分钟数
        if timeout is None:
            timeout = DEFAULT_TIMEOUT_MINUTES

        # 创建payload副本，避免修改原始数据
        payload_copy = payload.copy()

        # 添加过期时间（exp）和签发时间（iat）
        # exp: 过期时间，超过此时间令牌无效
        # iat: 签发时间，用于令牌生命周期计算
        current_time = datetime.datetime.now(tz=datetime.timezone.utc)
        payload_copy['exp'] = current_time + datetime.timedelta(minutes=timeout)
        payload_copy['iat'] = current_time

        # 声明类型和加密算法
        # 说明：定义JWT头部信息，包含令牌类型和加密算法
        headers = {
            "typ": "JWT",  # 令牌类型
            "alg": JWT_ALGORITHM  # 加密算法
        }

        # 生成JWT令牌
        # 函数：jwt.encode()
        # 参数说明：
        #   payload: JWT载荷数据
        #   key: 加密密钥（盐）
        #   algorithm: 加密算法（HS256）
        #   headers: JWT头部信息
        token = jwt.encode(
            payload=payload_copy,
            key=JWT_SECRET,
            algorithm=JWT_ALGORITHM,
            headers=headers
        )

        # 记录日志（可选，生产环境建议使用DEBUG级别）
        logger.debug(f"生成JWT令牌成功，过期时间：{timeout}分钟")

        return token

    except Exception as e:
        # 捕获并记录生成令牌时的异常
        logger.error(f"生成JWT令牌失败: {str(e)}")
        raise
```

#### Token 验证

```python
# src/common/middleware/authentication.py
# ============================================================
# 登录认证装饰器
# ============================================================
def login_required(func):
    """登录检测装饰器：校验 JWT（含黑名单）并将 userId 写入上下文。"""
    def wrapper(self, *args, **kwargs):
        request_url = self.request.path
        if request_url not in IGNORE_URL and self.request.method != "OPTIONS":
            success, access_token, _ = get_access_token(self.request)
            if not success or not access_token:
                return R.failed(self, code=401, msg="登录过期")
            result = parse_payload(access_token)
            if result["code"] != 0:
                return R.failed(self, code=401, msg="登录过期")
            if is_token_blacklisted(access_token):
                return R.failed(self, code=401, msg="登录过期")
            data = result["data"]
            current_user_id.set(int(data["userId"]))
            current_username.set(data.get("username", ""))
            current_realname.set(data.get("realname", ""))
        return func(self, *args, **kwargs)
    return wrapper
        raise AuthorizationException(code=401, msg=result['msg'])

    data = result['data']
    user_id = int(data.get('userId', 0))

    # 用户有效状态校验：优先读 Redis 缓存（状态变更/删除时已即时失效），
    # 未命中再回源 DB 并回填缓存，避免每请求查库；Redis 不可用时缓存读失败自动回源 DB
    from utils.perm_cache import get_cached_user_active, set_cached_user_active
    active = await get_cached_user_active(user_id)
    if active is None:
        from modules.user.repository import UserRepository
        user = UserRepository().get_by_id, user_id)
        active = bool(user and user.status == 1)
        await set_cached_user_active(user_id, active)

    if not active:
        raise AuthorizationException(code=401, msg="用户已被禁用或删除，请联系管理员")

    # 写入上下文变量（contextvars），供仓储/服务层读取
    current_user_id.set(user_id)
    current_username.set(data.get('username', ''))
    current_realname.set(data.get('realname', ''))

    # 滑动窗口续签：token 生命过半时自动签发新 token，通过响应头返回
    await _maybe_refresh_token(request, access_token, data)
```

#### Token 黑名单

用户登出时，将当前 Token 加入 Redis 黑名单，TTL 等于 Token 剩余有效期：

```python
# src/modules/auth/handlers.py
# ============================================================
# 退出登录
# ============================================================
class LogoutHandler(BaseHandler):
    async def get(self):
        """用户退出登录：将当前访问令牌加入黑名单。"""
        success, access_token, _ = get_access_token(self.request)
        if success and access_token:
            add_token_to_blacklist(access_token)

    # 记录退出登录日志
    username = get_username(request)
    realname = get_realname(request)
    login_log.record_login_log, request, username, 2, 0, 0, method='v1.logout',
                            request_method=request.method, create_user=realname, result="退出成功")
    return R.ok(msg="注销成功")
```

### 登录验证码

:::tip 验证码防暴力破解
每次登录前需先获取验证码，验证码存储在 Redis 中（TTL 120 秒），登录时校验验证码正确性。验证码使用 UUID 作为 key，避免被猜测。
:::

```python
# 获取验证码
GET /api/v1/captcha
Response: { "key": "xxx", "captcha": "data:image/png;base64,iVBORw0KG..." }

# 登录时校验
POST /api/v1/login
Body: { "username": "admin", "password": "123456", "code": "a3Xb", "key": "xxx" }
```

### 登录失败锁定

连续 5 次登录失败后，账号自动锁定 10 分钟：

```python
# src/modules/auth/service.py
async def login(request, data):
    redis = get_redis()

    # 检查锁定状态
    lock_key = f"login:lock:{data.username}"
    is_locked = redis.get(lock_key)
    if is_locked:
        return R.failed("账号已锁定，请10分钟后再试")

    # 验证用户名密码
    user = UserRepository().get_one, username=data.username)
    if not user or not verify_password(data.password, user.password):
        # 原子递增失败计数
        fail_key = f"login:fail:{data.username}"
        count = await incr_with_expire(redis, fail_key, 300)
        if count >= 5:
            redis.set(lock_key, 1, ex=600)  # 锁定 10 分钟
            return R.failed(f"用户名或密码错误，剩余{5-count}次机会")

    # 登录成功，清除失败计数
    redis.delete(f"login:fail:{data.username}")
```

### 密码安全

```python
# src/core/password.py
# +======================================================================
# | 模块: 密码加密工具
# | 说明: 基于 bcrypt 的双重加盐方案
# +======================================================================

"""密码加密工具：基于 bcrypt 的双重加盐方案。"""

import secrets

import bcrypt

# bcrypt 工作因子，值越大计算越慢越安全，默认12（约250ms）
BCRYPT_ROUNDS = 12


# ============================================================
# 密码工具函数
# ============================================================
def generate_salt(length: int = 10) -> str:
    """生成随机盐值"""
    return secrets.token_urlsafe(length)[:length]


def encrypt_password(password: str, salt: str) -> str:
    """使用 bcrypt 加密密码（外部 salt 双重加盐）"""
    combined = (password + salt).encode('utf-8')
    hashed = bcrypt.hashpw(combined, bcrypt.gensalt(rounds=BCRYPT_ROUNDS))
    return hashed.decode('utf-8')


def verify_password(password: str, salt: str, hashed: str) -> bool:
    """验证密码是否正确"""
    combined = (password + salt).encode('utf-8')
    return bcrypt.checkpw(combined, hashed.encode('utf-8'))
```

- 使用 bcrypt 算法，每次哈希自动生成随机盐
- 密码不可逆，数据库泄露后无法还原明文
- 默认密码在首次登录时强制修改

## 二、授权机制

### RBAC 权限模型

```
用户(User) ──N:M──→ 角色(Role) ──N:M──→ 菜单/权限(Menu)
 │ │ │
 │ user_role │ role_menu │ permission
 │ 关联表 │ 关联表 │ 权限标识
 └────────────────────┘ │
 ▼
 sys:user:add
 sys:user:page
 sys:role:update
 ...
```

权限标识格式：`sys:{module}:{action}`

- `sys`：系统前缀
- `{module}`：模块名（user / role / menu / position / level 等）
- `{action}`：操作名（page / list / detail / add / update / delete / status / batchDelete / export / import）

### permission_required 装饰器

```python
# src/common/middleware/access_decorators.py
# ============================================================
# 辅助函数
# ============================================================
def _extract_request(*args, **kwargs):
    """从函数参数中提取 Request 对象"""
    req = kwargs.get('request')
    if req is not None:
        return req
    for arg in args:
        if isinstance(arg, Request):
            return arg
    return None


# ============================================================
# 节点权限鉴权装饰器
# ============================================================
def permission_required(permission: str):
    """节点权限鉴权装饰器，userId==1（管理员）直接放行"""

    def decorator(func):
        def _check(self):
            """同步权限校验（直查DB，无Redis依赖，供线程池/回退场景使用）"""
            userId = current_user_id.get()
            if userId == 1:
                return None
            # 懒加载业务 service，避免 core → 业务模块顶层依赖
            from menu import service as menu
            permission_list = menu.get_permissions_list_sync(userId)
            if permission not in permission_list:
                return R.failed("权限不足")
            return None

        async def _check_cached(self):
            """异步权限校验（走 Redis 缓存，未命中才回源DB）"""
            userId = current_user_id.get()
            if userId == 1:
                return None
            # 懒加载业务 service，避免 core → 业务模块顶层依赖
            from menu import service as menu
            permission_list = await menu.get_permissions_list(userId)
            if permission not in permission_list:
                return R.failed("权限不足")
            return None

        if asyncio.iscoroutinefunction(func):
            @wraps(func)
            async def async_wrapper(*args, **kwargs):
                request = _extract_request(*args, **kwargs)
                denied = await _check_cached(request)
                if denied:
                    return denied
                return await func(*args, **kwargs)

            return async_wrapper

        @wraps(func)
        def sync_wrapper(*args, **kwargs):
            request = _extract_request(*args, **kwargs)
            # 同步端点由 Tornado 在线程池中执行，直接走同步直查 DB 校验；
            # 不再每请求 asyncio.run 新建事件循环（避免跨 loop 复用 Redis 连接
            # 被静默吞掉、缓存形同虚设的性能与正确性问题）
            denied = _check(request)
            if denied:
                return denied
            return func(*args, **kwargs)

        return sync_wrapper

    return decorator
```

:::tip 同步/异步双通道
装饰器根据被装饰函数的类型自动选择校验通道：
- `async def` 端点 → `_check_cached` 走 Redis 缓存（异步）
- `def` 端点 → `_check` 直查 DB（同步，在 Tornado 线程池中执行）

这样避免了在同步端点中 `asyncio.run()` 新建事件循环导致的跨 loop 复用 Redis 连接问题。
:::

### 权限缓存

权限缓存采用「用户级缓存 + 全局版本号」双层策略（`src/common/permission.py`）：

```python
# src/common/permission.py
_VERSION_KEY = 'perm:version'        # 全局版本号
_PERM_PREFIX = 'perm:list:'          # 用户权限列表前缀
_MENU_PREFIX = 'perm:menu:'          # 用户菜单树前缀
_CACHE_TTL = 7200                    # 缓存过期时间（2小时）

async def get_cached_permissions(user_id: int):
    """获取缓存的用户权限列表，未命中返回 None"""
    redis = get_redis()
    ver = await _get_version()
    key = f'{_PERM_PREFIX}{user_id}:{ver}'
    data = redis.get(key)
    if data is not None:
        return json.loads(data)
    return None

async def set_cached_permissions(user_id: int, perm_list: list):
    """缓存用户权限列表（空列表不缓存，避免缓存空壳后无法感知后续数据变更）"""
    if not perm_list:
        return
    redis = get_redis()
    ver = await _get_version()
    key = f'{_PERM_PREFIX}{user_id}:{ver}'
    redis.set(key, json.dumps(perm_list), ex=_CACHE_TTL)
```

缓存失效策略：

```python
# src/common/permission.py
async def invalidate_menu_change():
    """菜单增删改时调用，递增全局版本号使所有用户缓存自然失效"""
    await _incr_version()

async def invalidate_user_perms(user_id: int):
    """用户角色变更时调用，删除该用户的权限和菜单缓存"""
    redis = get_redis()
    ver = await _get_version()
    keys = [
        f'{_PERM_PREFIX}{user_id}:{ver}',
        f'{_MENU_PREFIX}{user_id}:{ver}',
    ]
    redis.delete(*keys)

async def invalidate_role_perms(role_id: int):
    """角色菜单变更时调用，清理该角色下所有用户的权限和菜单缓存"""
    # 查询该角色下所有用户 ID
    user_ids = _query_user_ids, role_id)
    if not user_ids:
        return
    redis = get_redis()
    ver = await _get_version()
    keys = []
    for uid in user_ids:
        keys.append(f'{_PERM_PREFIX}{uid}:{ver}')
        keys.append(f'{_MENU_PREFIX}{uid}:{ver}')
    if keys:
        redis.delete(*keys)
```

:::tip 版本号策略的优势
传统方案在菜单变更时需要遍历所有用户删除缓存，版本号方案只需递增一个全局 key，所有用户的旧版本缓存自然失效（key 中包含版本号，读取时用新版本号拼 key，旧 key 不再被访问，等 TTL 过期自动清除）。
:::

### check_demo 装饰器

```python
# src/common/middleware/access_decorators.py
def check_demo(func):
    """演示模式拦截装饰器，放在 @permission_required 之后、路由函数之前"""
    if asyncio.iscoroutinefunction(func):
        @wraps(func)
        async def async_wrapper(*args, **kwargs):
            if TORNADO_DEMO:
                return R.failed("演示环境，暂无操作权限")
            return await func(*args, **kwargs)

        return async_wrapper

    @wraps(func)
    def sync_wrapper(*args, **kwargs):
        if TORNADO_DEMO:
            return R.failed("演示环境，暂无操作权限")
        return func(*args, **kwargs)

    return sync_wrapper
```

- 保护所有写操作（add / update / delete / status / batchDelete / import）
- 在 `permission_required` 之后应用
- 根据被装饰函数类型自动选择同步/异步包装
- `TORNADO_DEMO=True` 时，所有写操作返回失败

## 三、网络层防护

### CORS 跨域控制

```python
# src/common/middleware/cors.py
from config.app import TORNADO_CORS_ORIGINS

_ALLOWED_ORIGINS = {o.strip() for o in TORNADO_CORS_ORIGINS.split(",") if o.strip()}

def cros_required(self):
    """为响应设置跨域相关头。"""
    origin = self.request.headers.get("Origin", "")
    if _ALLOWED_ORIGINS:
        if origin in _ALLOWED_ORIGINS:
            self.set_header("Access-Control-Allow-Origin", origin)
            self.set_header("Access-Control-Allow-Credentials", "true")
            self.set_header("Vary", "Origin")
    else:
        self.set_header("Access-Control-Allow-Origin", "*")
    self.set_header("Access-Control-Allow-Headers", "*")
    self.set_header("Access-Control-Expose-Headers", "*")
    self.set_header("Access-Control-Allow-Methods", "GET,POST,PUT,DELETE,OPTIONS")
    self.set_header("Access-Control-Max-Age", "3600")
```

在 `BaseHandler.set_default_headers()` 中调用，每个请求自动设置 CORS 头。

- 生产环境应配置具体的 `TORNADO_CORS_ORIGINS`，避免使用 `*`
- 配置白名单后仅对匹配来源反射来源并允许携带凭据

### XSS 防护

```python
# src/common/utils/rich_text.py
import bleach

def sanitize_html(content: str) -> str:
    """XSS 清洗：移除危险标签和属性"""
    allowed_tags = [
        'p', 'br', 'strong', 'em', 'u', 'ol', 'ul', 'li',
        'h1', 'h2', 'h3', 'h4', 'h5', 'h6',
        'a', 'img', 'table', 'tr', 'td', 'th', 'tbody', 'thead'
    ]
    allowed_attrs = {
        'a': ['href', 'title'],
        'img': ['src', 'alt', 'width', 'height'],
    }
    return bleach.clean(content, tags=allowed_tags, attributes=allowed_attrs)
```

- 富文本内容入库前进行 XSS 清洗
- 使用 `bleach` 库白名单过滤危险标签和属性
- 图片 `src` 属性只允许合法 URL

- 按 `Content-Length` 提前拒绝超大请求体
- 避免超大 multipart 落临时盘后再由业务层校验

## 四、数据层防护

### SQL 注入防护

:::tip ORM 天然防注入
SQLAlchemy ORM 使用参数化查询（Parameterized Query），所有用户输入通过参数绑定传递，不直接拼接 SQL 字符串，从根本上防止 SQL 注入。
:::

```python
# SQLAlchemy 参数化查询示例
UserRepository().get_one(username="admin")
# 实际执行：SELECT * FROM tornado_user WHERE username = :username AND is_delete = 0
# 参数绑定：{'username': 'admin'}
```

### 参数校验

```python
# Pydantic 声明式校验
class PositionForm(BaseSchema):
    name: str = Field(..., min_length=1, max_length=150)
    status: int = Field(..., ge=1, le=2)
    sort: int = Field(..., ge=0, le=99999)

    @field_validator('name')
    @classmethod
    def validate_name(cls, v):
        if not v.strip():
            raise ValueError('岗位名称不能为空')
        return v.strip()
```

- 所有请求参数通过 Pydantic 声明式校验
- 自动拒绝不符合约束的请求
- 校验错误统一返回格式

### 敏感字段保护

- 密码使用 bcrypt 哈希存储，不可逆
- `to_dict()` 序列化时自动排除 `is_delete`、`password`、`salt`
- JWT 密钥通过环境变量配置，不硬编码

## 五、审计追踪

### 操作日志

```python
# src/common/middleware/operation_log.py
def operation_log(module: str, action: str):
    """操作日志记录装饰器：自动记录写操作的模块、操作类型、执行耗时。"""
    def decorator(func):
        @wraps(func)
        async def wrapper(self, *args, **kwargs):
            start = time.time()
            result = await func(self, *args, **kwargs)
            duration = time.time() - start
            # 记录操作日志
            log_data = {
                'module': module,
                'action': action,
                'ip': get_client_ip(self.request),
                'duration': int(duration * 1000),
            }
        save_operation_log(log_data)
    return response
```

自动记录所有写操作的日志，包括：
- 操作人、操作模块、操作动作
- 请求 URL、请求参数
- 操作 IP、操作时间
- 响应状态、执行耗时

### 登录日志

每次登录/登出自动记录到 `tornado_login_log` 表，包括登录账号、IP、浏览器、操作系统、登录状态等。

## 安全配置清单

| 配置项 | 环境变量 | 默认值 | 说明 |
|:---|:---|:---|:---|
| JWT 密钥 | `JWT_SECRET` | 必须配置 | HS256 签名密钥 |
| JWT 过期时间 | `JWT_EXPIRE_MINUTES` | 20 | Token 有效期（分钟） |
| 密码错误锁定 | — | 5 次/10 分钟 | 连续失败后锁定 |
| 验证码有效期 | — | 120 秒 | 验证码 Redis TTL |
| 限流开关 | `RATE_LIMIT_ENABLED` | False | 滑动窗口限流 |
| 限流阈值 | `RATE_LIMIT_LIMIT` | 100 | 每窗口最大请求数 |
| 限流窗口 | `RATE_LIMIT_WINDOW_SECONDS` | 60 | 窗口时长（秒） |
| CORS 源 | `CORS_ORIGINS` | * | 允许的跨域源 |
| 上传限制 | `MAX_UPLOAD_SIZE` | 10MB | 最大上传体积 |
| 演示模式 | `TORNADO_DEMO` | False | 锁定写操作 |
| 数据库密码 | `DB_PASSWORD` | — | 数据库连接密码 |
| Redis 密码 | `REDIS_PASSWORD` | — | Redis 连接密码 |

:::warning 生产环境安全检查
1. `JWT_SECRET` 必须使用强随机字符串，不能使用默认值
2. `CORS_ORIGINS` 必须配置具体的域名，不能使用 `*`
3. `TORNADO_DEMO` 在生产环境应为 `False`
4. 数据库和 Redis 必须设置密码
5. 启用 HTTPS（由 Nginx/负载均衡器处理）
:::

## 总结

的安全架构涵盖认证（JWT + 验证码 + 登录锁定 + Token 黑名单）、授权（RBAC + permission_required + check_demo）、防护（CORS + 限流 + XSS + 上传限制 + SQL 注入防护）、审计（操作日志 + 登录日志）四大领域。通过纵深防御策略，在网络层、认证层、授权层、数据层分别设置安全屏障，确保系统在面对各类安全威胁时具备足够的防护能力。生产部署时需特别关注 JWT 密钥强度、CORS 配置、HTTPS 启用等关键安全项。