"""安全工具:JWT Token 和密码哈希""" from datetime import datetime, timedelta, timezone from jose import JWTError, jwt from passlib.context import CryptContext from app.config import settings # 密码哈希上下文 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def hash_password(password: str) -> str: """密码加密""" return pwd_context.hash(password) def verify_password(plain_password: str, hashed_password: str) -> bool: """验证密码""" return pwd_context.verify(plain_password, hashed_password) def create_access_token(data: dict) -> str: """创建 Access Token""" to_encode = data.copy() expire = datetime.now(timezone.utc) + timedelta( minutes=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES ) to_encode.update({"exp": expire, "type": "access"}) return jwt.encode(to_encode, settings.JWT_SECRET_KEY, algorithm="HS256") def create_refresh_token(data: dict) -> str: """创建 Refresh Token""" to_encode = data.copy() expire = datetime.now(timezone.utc) + timedelta( days=settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS ) to_encode.update({"exp": expire, "type": "refresh"}) return jwt.encode( to_encode, settings.JWT_REFRESH_SECRET_KEY, algorithm="HS256" ) def decode_access_token(token: str) -> dict | None: """解码 Access Token""" try: payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=["HS256"]) if payload.get("type") != "access": return None return payload except JWTError: return None def decode_refresh_token(token: str) -> dict | None: """解码 Refresh Token""" try: payload = jwt.decode( token, settings.JWT_REFRESH_SECRET_KEY, algorithms=["HS256"] ) if payload.get("type") != "refresh": return None return payload except JWTError: return None