34 lines
1.0 KiB
Python
34 lines
1.0 KiB
Python
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import jwt
|
|||
|
|
from datetime import datetime, timedelta, timezone
|
|||
|
|
|
|||
|
|
from app.core.config import get_settings
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _now() -> datetime:
|
|||
|
|
return datetime.now(timezone.utc)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def create_access_token(user_id: str, expires_minutes: int | None = None) -> str:
|
|||
|
|
"""为指定用户签发 JWT 访问令牌。"""
|
|||
|
|
settings = get_settings()
|
|||
|
|
expire = _now() + timedelta(minutes=expires_minutes or settings.access_token_expire_minutes)
|
|||
|
|
payload = {
|
|||
|
|
"sub": user_id,
|
|||
|
|
"iat": int(_now().timestamp()),
|
|||
|
|
"exp": int(expire.timestamp()),
|
|||
|
|
}
|
|||
|
|
return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def decode_access_token(token: str) -> str | None:
|
|||
|
|
"""校验并返回令牌中的用户 ID；无效/过期返回 None。"""
|
|||
|
|
settings = get_settings()
|
|||
|
|
try:
|
|||
|
|
payload = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm])
|
|||
|
|
except jwt.PyJWTError:
|
|||
|
|
return None
|
|||
|
|
sub = payload.get("sub")
|
|||
|
|
return sub if isinstance(sub, str) else None
|