43 lines
1.5 KiB
Python
43 lines
1.5 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
from typing import Literal
|
|
|
|
import jwt
|
|
from fastapi import HTTPException, status
|
|
|
|
from app.core.config import get_settings
|
|
|
|
TokenType = Literal["staff", "member"]
|
|
|
|
|
|
def create_access_token(subject: str, token_type: TokenType) -> str:
|
|
settings = get_settings()
|
|
if len(settings.jwt_secret) < 32:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="鉴权未配置",
|
|
)
|
|
minutes = (
|
|
settings.jwt_member_expire_minutes
|
|
if token_type == "member"
|
|
else settings.jwt_expire_minutes
|
|
)
|
|
expire = datetime.now(timezone.utc) + timedelta(minutes=minutes)
|
|
payload = {"sub": subject, "typ": token_type, "exp": expire}
|
|
return jwt.encode(payload, settings.jwt_secret, algorithm="HS256")
|
|
|
|
|
|
def decode_access_token(token: str, expected_type: TokenType) -> dict:
|
|
settings = get_settings()
|
|
if len(settings.jwt_secret) < 32:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="鉴权未配置",
|
|
)
|
|
try:
|
|
payload = jwt.decode(token, settings.jwt_secret, algorithms=["HS256"])
|
|
except jwt.PyJWTError:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="登录已失效")
|
|
if payload.get("typ") != expected_type or not payload.get("sub"):
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="无权访问该接口")
|
|
return payload
|