JWT认证

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

认证流程

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

配置项

环境变量	默认值	说明
JWT_SALT	内置默认密钥	JWT 签名密钥，建议至少32字节
DEFAULT_TIMEOUT_MINUTES	20	Token 过期时间（分钟）

密钥安全

生产环境务必通过 .env 设置 JWT_SALT 为强随机密钥。生成方式：

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

Token 生成

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

from utils.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

    返回:
        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 utils.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

    # 检查Token是否在黑名单中
    if is_token_blacklisted(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	算法不匹配等

Token 刷新

Token 支持无感刷新，刷新时会生成新令牌：

from utils.jwt import refresh_token

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

刷新核心逻辑：

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. 移除JWT标准字段，保留业务数据
    payload = result['data']
    clean_payload = {k: v for k, v in payload.items() if k not in ['exp', 'iat', 'nbf']}

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

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

Token 黑名单

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

from utils.jwt import add_token_to_blacklist, is_token_blacklisted

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

# 中间件校验：检查是否已注销
if 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()

def add_token_to_blacklist(token: str) -> bool:
    """退出登录时调用，TTL 与 Token 剩余有效期一致"""
    from django_redis import get_redis_connection
    redis = get_redis_connection("default")

    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}'
    redis.set(key, '1', ex=ttl_seconds)

def is_token_blacklisted(token: str) -> bool:
    """检查 Token 是否已注销，Redis 不可用时 fail-open 放行"""
    try:
        from django_redis import get_redis_connection
        redis = get_redis_connection("default")

        fingerprint = _token_fingerprint(token)
        key = f'{_TOKEN_BLACKLIST_PREFIX}{fingerprint}'
        return 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 utils.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.META.get('HTTP_AUTHORIZATION', '')
    stripped = auth_header.strip()
    if stripped.lower().startswith('bearer '):
        access_token = stripped[7:].strip()
    else:
        access_token = stripped
    return True, access_token, 'success'

check_login 装饰器

middleware/login_middleware.py 中的 check_login 是全局认证装饰器，应用于所有需要登录的视图：

from middleware.login_middleware import check_login

@method_decorator(check_login, name='dispatch')
class UserPageView(PermissionRequired, View):
    def get(self, request):
        # 只有已登录用户才能访问
        ...

认证成功后，业务代码通过 utils/security.py 获取当前用户信息：

from utils.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字节）
2. 黑名单管理：SHA256 指纹存储，TTL 自动清理，Redis 故障时 fail-open 降级
3. 全局统一：check_login 装饰器 + security 工具函数，业务代码零侵入
4. Bearer 标准：RFC 7235 规范，前缀大小写不敏感
5. Django 原生：基于 django-redis 连接池，与 Django ORM 无缝集成