JWT认证

JWT（JSON Web Token）是的核心认证机制。用户登录成功后获取 Token，后续请求通过 Authorization: Bearer <token> 头携带 Token 进行身份验证。项目内置了 Token 生成、解析、刷新和黑名单管理功能，源码位于 src/core/jwt.py。

认证流程

1. 前端发送 POST /api/v1/login（username + password + captcha）
2. 后端校验验证码、用户名密码
3. 登录成功 → create_token() 生成 JWT
4. 前端存储 Token，后续请求携带 Authorization: Bearer <token>
5. login_required 依赖解析 Token → 注入 request.state
6. 退出登录 → Token 加入 Redis 黑名单立即失效

配置项

环境变量	默认值	说明
JWT_SALT	无（必填）	JWT 签名密钥，至少32字节
JWT_EXPIRE_MINUTES	20	Token 过期时间（分钟）

密钥安全

JWT_SALT 必须通过环境变量设置，启动时会校验密钥强度（至少32字节），强度不足将拒绝启动。生成方式：

python -c "import secrets; print(secrets.token_urlsafe(48))"

Token 生成

src/core/jwt.py 中的 create_token() 函数负责生成 JWT：

from core.jwt import create_token

# 登录成功后生成 Token
access_token = create_token({
    "userId": user.id,
    "username": user.username,
    "realname": user.realname or ''
})

函数签名与核心逻辑：

def create_token(payload, timeout=None):
    """
    生成JWT令牌

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

    返回:
        str: 生成的JWT令牌字符串
    """
    if timeout is None:
        timeout = DEFAULT_TIMEOUT_MINUTES

    payload_copy = payload.copy()
    current_time = datetime.datetime.now(tz=datetime.timezone.utc)
    payload_copy['exp'] = current_time + datetime.timedelta(minutes=timeout)
    payload_copy['iat'] = current_time

    headers = {"typ": "JWT", "alg": JWT_ALGORITHM}

    token = jwt.encode(
        payload=payload_copy,
        key=JWT_SALT,
        algorithm=JWT_ALGORITHM,
        headers=headers
    )
    return token

Token 结构包含三部分：

# Header
{
    "typ": "JWT",
    "alg": "HS256"
}

# Payload（自动添加 exp 和 iat）
{
    "userId": 1,
    "username": "admin",
    "realname": "管理员",
    "exp": "2026-09-05T12:00:00Z",  # 过期时间
    "iat": "2026-09-05T11:40:00Z"  # 签发时间
}

Token 解析验证

parse_payload() 函数验证 Token 签名和过期状态，返回标准格式结果：

from core.jwt import parse_payload

result = parse_payload(token)
if result['code'] == 0:
    user_id = result['data']['userId']
    username = result['data']['username']
else:
    error_msg = result['msg']
    # "token已失效，请重新登录" / "token认证失败" / "非法的token"

核心实现：

def parse_payload(token):
    """解析验证JWT令牌，返回 {code, data, msg} 标准格式"""
    result = {"code": 0, "data": None, "msg": "操作成功"}

    if not token:
        result['code'] = 1
        result['msg'] = "token不能为空"
        return result

    try:
        verified_payload = jwt.decode(token, JWT_SALT, algorithms=[JWT_ALGORITHM])
        result['data'] = verified_payload
    except exceptions.ExpiredSignatureError:
        result['code'] = 1
        result['msg'] = "token已失效，请重新登录"
    except exceptions.DecodeError:
        result['code'] = 1
        result['msg'] = "token认证失败，无效的令牌格式"
    except exceptions.InvalidTokenError:
        result['code'] = 1
        result['msg'] = "非法的token，请检查令牌有效性"

    return result

异常处理说明：

异常类型	提示信息	说明
ExpiredSignatureError	token已失效，请重新登录	Token 过期
DecodeError	token认证失败，无效的令牌格式	格式错误或签名无效
InvalidTokenError	非法的token	算法不匹配等

登录接口

登录接口 POST /api/v1/login 接收 LoginForm 参数：

# src/modules/auth/schemas.py
class LoginForm(BaseModel):
    username: str = Field(..., min_length=1, max_length=20, description="登录账号")
    password: str = Field(..., min_length=6, max_length=128, description="登录密码")
    code: str = Field(..., min_length=CAPTCHA_LENGTH, max_length=CAPTCHA_LENGTH, description="验证码")
    key: str = Field(..., min_length=1, description="KEY值")

登录逻辑位于 src/modules/auth/service.py，完整流程包括锁定检查、验证码校验、密码比对、默认密码检测：

async def login(request, data: LoginForm):
    ip = get_client_ip(request)
    redis = get_redis()

    # 1. 检查是否被锁定（IP+用户名维度）
    locked, remaining = await _is_locked(redis, ip, data.username)
    if locked:
        return R.failed(f"登录失败次数过多，请{remaining}秒后再试")

    # 2. 验证码校验（Lua 原子消费，防重放）
    if not await verify_captcha(data.key, data.code):
        return R.failed("验证码不正确")

    # 3. 查询用户（同步DB查询放入线程池）
    user = await asyncio.to_thread(
        lambda: user_repo.get_one(username=data.username, status=1)
    )
    if not user:
        await _record_fail(redis, ip, data.username)
        return R.failed("用户名或密码不正确")

    # 4. 比对密码（bcrypt CPU密集，放入线程池）
    password_ok = await asyncio.to_thread(
        password.verify_password, data.password, user.salt or '', user.password or '')
    if not password_ok:
        await _record_fail(redis, ip, data.username)
        return R.failed("用户名或密码不正确")

    # 5. 登录成功，清除失败记录
    await _clear_fail(redis, ip, data.username)

    # 6. 检测是否为默认密码（仅提示，不阻断登录）
    is_default_pwd = bool(user.password and await asyncio.to_thread(
        password.verify_password, DEFAULT_PASSWORD, user.salt or '', user.password or ''))

    # 7. 生成 Token
    access_token = create_token({
        "userId": user.id,
        "username": user.username,
        "realname": user.realname or ''
    })

    return R.ok(msg="登录成功", data={
        "access_token": access_token,
        "must_change_password": is_default_pwd
    })

登录成功返回：

{
    "code": 0,
    "msg": "登录成功",
    "data": {
        "access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9..."
    },
    "ok": true
}

Token 刷新

Token 支持无感刷新，刷新时会执行令牌轮换（旧 Token 立即失效）：

from core.jwt import refresh_token

result = await refresh_token(old_token, extend_minutes=30)
if result['code'] == 0:
    new_token = result['data']['access_token']
    expires_in = result['data']['expires_in']  # 秒

刷新核心逻辑：

async def refresh_token(token, extend_minutes=None):
    # 1. 验证原令牌有效性
    result = parse_payload(token)
    if result['code'] != 0:
        return {"code": 1, "data": None, "msg": f"令牌无效，无法刷新: {result['msg']}"}

    # 2. 已注销的令牌禁止刷新（防止攻击者用已注销 token 无限轮换）
    if await is_token_blacklisted(token):
        return {"code": 1, "data": None, "msg": "令牌已注销，无法刷新，请重新登录"}

    # 3. 移除JWT标准字段，保留业务数据
    payload = result['data']
    clean_payload = {k: v for k, v in payload.items() if k not in ['exp', 'iat', 'nbf']}

    # 4. 生成新令牌
    new_token = create_token(clean_payload, timeout=extend_minutes)

    # 5. 令牌轮换：将旧令牌加入黑名单
    await add_token_to_blacklist(token)

    return {
        "code": 0,
        "data": {"access_token": new_token, "expires_in": extend_minutes * 60},
        "msg": "令牌刷新成功"
    }

令牌轮换

刷新后旧 Token 会被加入黑名单，防止旧 Token 在有效期内继续使用。已注销的 Token 不允许刷新，防止攻击者利用已注销令牌无限轮换出新令牌。

Token 黑名单

退出登录时将 Token 加入 Redis 黑名单，使其立即失效：

from core.jwt import add_token_to_blacklist, is_token_blacklisted

# 退出登录：加入黑名单
await add_token_to_blacklist(token)

# 中间件校验：检查是否已注销
if await is_token_blacklisted(token):
    # 拒绝访问

黑名单实现细节：

# 黑名单 Key 前缀
_TOKEN_BLACKLIST_PREFIX = 'token:blacklist:'

# SHA256 指纹存储，不将完整 Token 存入 Redis
def _token_fingerprint(token: str) -> str:
    return hashlib.sha256(token.encode('utf-8')).hexdigest()

async def add_token_to_blacklist(token: str) -> bool:
    """退出登录时调用，TTL 与 Token 剩余有效期一致"""
    expiration = get_token_expiration(token)
    ttl_seconds = int((expiration - datetime.datetime.now(tz=datetime.timezone.utc)).total_seconds())
    fingerprint = _token_fingerprint(token)
    key = f'{_TOKEN_BLACKLIST_PREFIX}{fingerprint}'
    await redis.set(key, '1', ex=ttl_seconds)

async def is_token_blacklisted(token: str) -> bool:
    """检查 Token 是否已注销，Redis 不可用时 fail-open 放行"""
    try:
        fingerprint = _token_fingerprint(token)
        key = f'{_TOKEN_BLACKLIST_PREFIX}{fingerprint}'
        return await redis.exists(key) > 0
    except Exception as e:
        # fail-open：Redis 故障时放行，避免全站不可用
        _log_blacklist_check_alert(e)
        return False

黑名单特性：

• 使用 SHA256 指纹存储，不将完整 Token 存入 Redis
• TTL 与 Token 剩余有效期一致，过期自动清理
• Redis 不可用时采用 fail-open 策略放行（避免全站不可用）
• 告警节流：Redis 故障期间首次失败记 ERROR，随后每 60 秒至多一条 ERROR，避免日志刷屏

请求中的 Token 提取

from core.jwt import get_access_token, parse_token

# 从请求头提取 Token
success, token, msg = get_access_token(request)

# 完整解析（提取 + 验证）
result = parse_token(request)
# result['code'] == 0 表示成功
# result['data']['userId'], result['data']['username']

get_access_token() 支持 Bearer 前缀大小写不敏感（RFC 7235）：

def get_access_token(request):
    """从请求头中提取 access_token，返回 (success, token, message)"""
    auth_header = request.headers.get('Authorization', '')
    stripped = auth_header.strip()
    if stripped.lower().startswith('bearer '):
        access_token = stripped[7:].strip()
    else:
        access_token = stripped
    return True, access_token, 'success'

login_required 依赖

src/api/deps.py 中的 login_required 是全局认证依赖，应用于所有 v1 路由：

# 认证成功后注入 request.state
request.state.user_id = userId
request.state.username = username
request.state.realname = realname

业务代码通过 src/core/security.py 获取当前用户信息：

from core.security import get_user_id, get_username, get_realname

user_id = get_user_id(request)
username = get_username(request)
realname = get_realname(request)

前端集成

前端登录后将 Token 存入 localStorage，每次请求自动携带：

// 请求拦截器
axios.interceptors.request.use(config => {
    const token = localStorage.getItem('access_token')
    if (token) {
        config.headers['Authorization'] = `Bearer ${token}`
    }
    return config
})

// 响应拦截器：检测 Token 过期
axios.interceptors.response.use(response => {
    if (response.data.code === 401) {
        // Token 过期，跳转登录页
        router.push('/login')
    }
    return response
})

总结

的 JWT 认证方案具备以下特点：

1. 安全性：HS256 算法 + 强密钥校验（至少32字节），启动时 fail-close
2. 令牌轮换：刷新时旧 Token 立即加入黑名单，防止重放攻击
3. 黑名单管理：SHA256 指纹存储，TTL 自动清理，Redis 故障时 fail-open 降级
4. 防暴力破解：登录失败计数 + 账户锁定（IP+用户名维度）
5. 全局统一：login_required 依赖 + request.state 注入，业务代码零侵入