buff3.0/backend-client/app/core/security.py

"""
安全相关功能
"""
# 临时兼容：部分 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