from datetime import datetime, timedelta from typing import Optional from jose import JWTError, jwt from passlib.context import CryptContext from .config import settings import hashlib # 密码加密上下文 - 禁用bcrypt的wrap bug检测以避免72字节限制错误 pwd_context = CryptContext( schemes=["bcrypt"], deprecated="auto", bcrypt__ident="2b", # 使用2b版本,跳过wrap bug检测 ) def _truncate_password(password: str) -> str: """ 截断密码以适应bcrypt的72字节限制 对于超长密码,使用SHA256哈希确保唯一性 """ password_bytes = password.encode('utf-8') if len(password_bytes) <= 72: return password # 对超长密码进行hash,确保唯一性且在72字节以内 return hashlib.sha256(password_bytes).hexdigest() def verify_password(plain_password: str, hashed_password: str) -> bool: """验证密码""" safe_password = _truncate_password(plain_password) return pwd_context.verify(safe_password, hashed_password) def get_password_hash(password: str) -> str: """加密密码(bcrypt限制最多72字节)""" safe_password = _truncate_password(password) return pwd_context.hash(safe_password) def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: """创建 JWT Token""" to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) return encoded_jwt def decode_access_token(token: str) -> Optional[dict]: """解析 JWT Token""" try: payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) return payload except JWTError: return None