62 lines
2.0 KiB
Python
62 lines
2.0 KiB
Python
"""JWT 认证工具。签发、校验、装饰器。"""
|
|
import functools
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import jwt
|
|
from flask import request
|
|
|
|
from insurance.config import JWT_ALGORITHM, JWT_EXPIRE_HOURS, JWT_SECRET
|
|
from insurance.utils.response import error
|
|
|
|
# 从请求中获取当前用户信息的 key
|
|
CURRENT_USER_KEY = "_insurance_current_user"
|
|
|
|
|
|
def generate_token(user_id: str, username: str, role: str, department: str = "") -> str:
|
|
payload = {
|
|
"user_id": user_id,
|
|
"username": username,
|
|
"role": role,
|
|
"department": department,
|
|
"exp": datetime.now(timezone.utc) + timedelta(hours=JWT_EXPIRE_HOURS),
|
|
"iat": datetime.now(timezone.utc),
|
|
}
|
|
return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
|
|
|
|
|
|
def decode_token(token: str) -> dict | None:
|
|
try:
|
|
return jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
|
|
except (jwt.ExpiredSignatureError, jwt.InvalidTokenError):
|
|
return None
|
|
|
|
|
|
def login_required(f):
|
|
@functools.wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
auth_header = request.headers.get("Authorization", "")
|
|
if not auth_header.startswith("Bearer "):
|
|
return error(1002, "未授权，请重新登录"), 401
|
|
|
|
token = auth_header[7:]
|
|
payload = decode_token(token)
|
|
if payload is None:
|
|
return error(1003, "登录已过期，请重新登录"), 401
|
|
|
|
# 检查 Redis 黑名单
|
|
try:
|
|
from extensions.ext_redis import redis_client
|
|
from insurance.config import REDIS_TOKEN_BLACKLIST_PREFIX
|
|
if redis_client.get(f"{REDIS_TOKEN_BLACKLIST_PREFIX}{token}"):
|
|
return error(1003, "登录已过期，请重新登录"), 401
|
|
except Exception:
|
|
pass # Redis 不可用时不阻塞
|
|
|
|
request._insurance_user = payload
|
|
return f(*args, **kwargs)
|
|
return decorated
|
|
|
|
|
|
def get_current_user() -> dict | None:
|
|
return getattr(request, "_insurance_user", None)
|