Skip to content

安全架构设计

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

认证(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 地址及执行结果,存入 fastapi_operation_log 表;登录日志独立记录每次登录尝试(含 IP、User-Agent、登录状态),便于安全事件追溯与异常行为分析;定时任务每次执行记录至 fastapi_job_log 表,确保后台任务的执行过程可审计。

安全设计原则

纵深防御:认证、授权、限流、参数校验多层防护,单一防线失效时仍有其他机制兜底

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

数据不落地:敏感信息(Token、密码)不在日志中明文记录,密码哈希不可逆

安全默认:框架默认配置为安全模式(如 DEBUG 关闭、限流开启),如需放宽需显式配置

安全架构总览

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

一、认证机制

JWT 令牌认证

JWT 工作流程

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

  1. 工作流程时序图

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

流程说明

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

JWT 配置

python
# src/core/config/auth.py
# ============================================================
# JWT 配置
# ============================================================
# JWT签名密钥(必须通过环境变量设置,不能使用默认值上线)
JWT_SALT = os.getenv('JWT_SALT', '')
# 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_SALT
        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_SALT,
            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/api/deps.py
# ============================================================
# 登录认证依赖
# ============================================================
async def login_required(request: Request):
    """登录验证依赖:提取并校验 Token,滑动窗口续签,校验用户有效,写入 request.state"""
    success, access_token, msg = get_access_token(request)
    if not success:
        raise AuthorizationException(code=401, msg=msg)

    if await is_token_blacklisted(access_token):
        raise AuthorizationException(code=401, msg="登录已注销")

    result = parse_payload(access_token)
    if result['code'] != 0:
        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.system.user.repository import user_repo
        user = await asyncio.to_thread(user_repo.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="用户已被禁用或删除,请联系管理员")

    request.state.user_id = user_id
    request.state.username = data.get('username', '')
    request.state.realname = data.get('realname', '')

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

Token 黑名单

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

python
# src/api/v1/endpoints/auth.py
# ============================================================
# 退出登录
# ============================================================
@router.get('/logout', summary='退出登录', dependencies=[Depends(login_required)])
async def logout(request: Request):
    """
    用户退出登录
    将当前访问令牌加入黑名单,并记录退出日志
    :param request: FastAPI请求对象
    :return: 退出结果响应
    """
    # 将当前令牌加入黑名单
    success, access_token, _ = get_access_token(request)
    if success and access_token:
        await add_token_to_blacklist(access_token)

    # 记录退出登录日志
    username = get_username(request)
    realname = get_realname(request)
    await asyncio.to_thread(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="注销成功")

登录验证码

验证码防暴力破解

每次登录前需先获取验证码,验证码存储在 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 = await redis.get(lock_key)
    if is_locked:
        return R.failed("账号已锁定,请10分钟后再试")

    # 验证用户名密码
    user = await asyncio.to_thread(user_repo.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:
            await redis.set(lock_key, 1, ex=600)  # 锁定 10 分钟
            return R.failed(f"用户名或密码错误,剩余{5-count}次机会")

    # 登录成功,清除失败计数
    await 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/core/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(request):
            """同步权限校验(直查DB,无Redis依赖,供线程池/回退场景使用)"""
            userId = get_user_id(request)
            if userId == 1:
                return None
            # 懒加载业务 service,避免 core → 业务模块顶层依赖
            from modules.system.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(request):
            """异步权限校验(走 Redis 缓存,未命中才回源DB)"""
            userId = get_user_id(request)
            if userId == 1:
                return None
            # 懒加载业务 service,避免 core → 业务模块顶层依赖
            from modules.system.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)
            # 同步端点由 FastAPI 在线程池中执行,直接走同步直查 DB 校验;
            # 不再每请求 asyncio.run 新建事件循环(避免跨 loop 复用 Redis 连接
            # 被静默吞掉、缓存形同虚设的性能与正确性问题)
            denied = _check(request)
            if denied:
                return denied
            return func(*args, **kwargs)

        return sync_wrapper

    return decorator

同步/异步双通道

装饰器根据被装饰函数的类型自动选择校验通道:

  • async def 端点 → _check_cached 走 Redis 缓存(异步)
  • def 端点 → _check 直查 DB(同步,在 FastAPI 线程池中执行)

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

权限缓存

python
# src/modules/system/menu/service.py
async def get_permissions_list(user_id: int) -> list:
    """获取用户权限列表(异步,走 Redis 缓存)"""
    cache_key = f"perm:user:{user_id}"
    redis = get_redis()
    cached = await redis.get(cache_key)
    if cached:
        return json.loads(cached)
    # 从数据库查询
    permissions = query_user_permissions(user_id)
    await redis.set(cache_key, json.dumps(permissions), ex=3600)
    return permissions


def get_permissions_list_sync(user_id: int) -> list:
    """获取用户权限列表(同步,直查DB,供线程池场景使用)"""
    return query_user_permissions(user_id)


def invalidate_perm_cache(user_id: int):
    """失效用户权限缓存(角色/菜单变更时调用)"""
    from core.redis_client import get_redis
    import asyncio

    redis = get_redis()
    asyncio.get_event_loop().run_until_complete(
        redis.delete(f"perm:user:{user_id}")
    )

check_demo 装饰器

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

        return async_wrapper

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

    return sync_wrapper
  • 保护所有写操作(add / update / delete / status / batchDelete / import)
  • permission_required 之后应用
  • 根据被装饰函数类型自动选择同步/异步包装
  • FASTAPI_DEMO=True 时,所有写操作返回失败

三、网络层防护

CORS 跨域控制

python
# src/middleware/cors.py
from fastapi.middleware.cors import CORSMiddleware

def register_cors(app):
    app.add_middleware(
        CORSMiddleware,
        allow_origins=CORS_ORIGINS,       # 允许的源
        allow_credentials=True,           # 允许 Cookie
        allow_methods=["*"],              # 允许的 HTTP 方法
        allow_headers=["*"],              # 允许的请求头
    )
  • 生产环境应配置具体的 allow_origins,避免使用 *
  • 开发环境可配置为 ["http://localhost:3000"]

请求限流

python
# src/middleware/rate_limit.py
async def rate_limit_middleware(request: Request, call_next):
    # 基于客户端 IP 的滑动窗口限流
    client_ip = get_client_ip(request)
    count = await sliding_window_incr(redis, current_key, prev_key, ...)
    if count > RATE_LIMIT_LIMIT:
        return JSONResponse(
            status_code=200,
            content={"code": 1, "msg": "请求过于频繁,请稍后再试"}
        )
    return await call_next(request)
  • 默认关闭,通过 RATE_LIMIT_ENABLED=True 启用
  • 使用 Redis 滑动窗口计数器,消除固定窗口边界 2x 突发
  • Redis 不可用时降级放行(fail-open)
  • CORS 预检请求不计数

XSS 防护

python
# src/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

上传体积限制

python
# src/middleware/upload_size.py
async def upload_size_limit_middleware(request: Request, call_next):
    content_length = request.headers.get('content-length')
    if content_length and int(content_length) > MAX_UPLOAD_SIZE:
        return JSONResponse(
            status_code=200,
            content={"code": 1, "msg": "上传文件超出大小限制"}
        )
    return await call_next(request)
  • Content-Length 提前拒绝超大请求体
  • 避免超大 multipart 落临时盘后再由业务层校验

四、数据层防护

SQL 注入防护

ORM 天然防注入

SQLAlchemy ORM 使用参数化查询(Parameterized Query),所有用户输入通过参数绑定传递,不直接拼接 SQL 字符串,从根本上防止 SQL 注入。

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

参数校验

python
# Pydantic 声明式校验
class PositionForm(BaseForm):
    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_deletepasswordsalt
  • JWT 密钥通过环境变量配置,不硬编码

五、审计追踪

操作日志

python
# src/middleware/operation_log.py
async def operation_log_middleware(request: Request, call_next):
    response = await call_next(request)
    if request.method in ('POST', 'PUT', 'DELETE', 'PATCH'):
        # 记录操作日志
        log_data = {
            'username': request.state.username,
            'module': extract_module(request.url.path),
            'action': request.method,
            'url': str(request.url),
            'params': await get_request_params(request),
            'ip': get_client_ip(request),
            'status': 1 if is_success else 2,
            'duration': int(duration * 1000),
        }
        save_operation_log(log_data)
    return response

自动记录所有写操作的日志,包括:

  • 操作人、操作模块、操作动作
  • 请求 URL、请求参数
  • 操作 IP、操作时间
  • 响应状态、执行耗时

登录日志

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

安全配置清单

配置项环境变量默认值说明
JWT 密钥JWT_SALT必须配置HS256 签名密钥
JWT 过期时间JWT_EXPIRE_MINUTES120Token 有效期(分钟)
密码错误锁定5 次/10 分钟连续失败后锁定
验证码有效期120 秒验证码 Redis TTL
限流开关RATE_LIMIT_ENABLEDFalse滑动窗口限流
限流阈值RATE_LIMIT_LIMIT100每窗口最大请求数
限流窗口RATE_LIMIT_WINDOW_SECONDS60窗口时长(秒)
CORS 源CORS_ORIGINS*允许的跨域源
上传限制MAX_UPLOAD_SIZE10MB最大上传体积
演示模式FASTAPI_DEMOFalse锁定写操作
数据库密码DB_PASSWORD数据库连接密码
Redis 密码REDIS_PASSWORDRedis 连接密码

生产环境安全检查

  1. JWT_SALT 必须使用强随机字符串,不能使用默认值
  2. CORS_ORIGINS 必须配置具体的域名,不能使用 *
  3. FASTAPI_DEMO 在生产环境应为 False
  4. 数据库和 Redis 必须设置密码
  5. 启用 HTTPS(由 Nginx/负载均衡器处理)

总结

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

小蚂蚁云团队 · 提供技术支持