""" 安全相关功能 """ # 临时兼容:部分 bcrypt 版本缺少 __about__.__version__,导致 passlib 初始化失败 # 在导入 passlib 前补充该属性,避免登录/解绑失败 try: import bcrypt # type: ignore if not hasattr(bcrypt, "__about__"): class _BcryptAbout: __version__ = getattr(bcrypt, "__version__", "unknown") bcrypt.__about__ = _BcryptAbout() # type: ignore[attr-defined] except Exception: pass try: from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") USE_PASSLIB = True except Exception: # 如果 passlib 有问题,使用 bcrypt 直接验证 import bcrypt USE_PASSLIB = False try: from jose import JWTError, jwt except ImportError: # 如果没有jose,使用简单的实现 import jwt JWTError = Exception from datetime import datetime, timedelta, timezone from typing import Optional from app.config import settings def verify_password(plain_password: str, hashed_password: str) -> bool: """验证密码""" if USE_PASSLIB: try: return pwd_context.verify(plain_password, hashed_password) except Exception: # 如果 passlib 验证失败,尝试使用 bcrypt 直接验证 import bcrypt try: return bcrypt.checkpw(plain_password.encode('utf-8'), hashed_password.encode('utf-8')) except Exception: return False else: # 直接使用 bcrypt import bcrypt try: return bcrypt.checkpw(plain_password.encode('utf-8'), hashed_password.encode('utf-8')) except Exception: return False def get_password_hash(password: str) -> str: """获取密码哈希""" if USE_PASSLIB: try: return pwd_context.hash(password) except Exception: # 如果 passlib 失败,使用 bcrypt 直接生成 import bcrypt salt = bcrypt.gensalt() hashed = bcrypt.hashpw(password.encode('utf-8'), salt) return hashed.decode('utf-8') else: # 直接使用 bcrypt import bcrypt salt = bcrypt.gensalt() hashed = bcrypt.hashpw(password.encode('utf-8'), salt) return hashed.decode('utf-8') def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: """创建访问令牌""" to_encode = data.copy() # 使用北京时间 (UTC+8) beijing_tz = timezone(timedelta(hours=8)) now = datetime.now(beijing_tz) if expires_delta: expire = now + expires_delta else: expire = now + timedelta(hours=24) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256") return encoded_jwt def verify_token(token: str) -> Optional[dict]: """验证令牌""" try: payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"]) return payload except JWTError: return None