from datetime import datetime, timedelta from typing import Optional from jose import JWTError, jwt from fastapi import HTTPException, Depends from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials import db SECRET_KEY = "your-secret-key-change-in-production" ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 security = HTTPBearer() optional_security = HTTPBearer(auto_error=False) import bcrypt # 使用原生 bcrypt 进行密码校验,解决 passlib 与 bcrypt 5.x 的兼容性问题 def verify_password(plain_password: str, hashed_password: str) -> bool: try: return bcrypt.checkpw(plain_password.encode('utf-8'), hashed_password.encode('utf-8')) except Exception: return False # 使用原生 bcrypt 生成密码哈希 def get_password_hash(password: str) -> str: salt = bcrypt.gensalt() return bcrypt.hashpw(password.encode('utf-8'), salt).decode('utf-8') def create_access_token(data: dict): to_encode = data.copy() expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)): token = credentials.credentials try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) user_id: int = payload.get("id") if user_id is None: raise HTTPException(status_code=401, detail="无效的认证信息") except JWTError: raise HTTPException(status_code=403, detail="登录已过期,请重新登录") user = db.get_user_by_id(user_id) if user is None: raise HTTPException(status_code=404, detail="用户不存在") return user def get_optional_user(credentials: Optional[HTTPAuthorizationCredentials] = Depends(optional_security)): if not credentials: return None try: payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM]) user_id = payload.get("id") if user_id: return db.get_user_by_id(user_id) except JWTError: return None return None