# -*- coding: utf-8 -*- """JWT 实现学生与教师角色细粒度权限控制""" from datetime import datetime, timedelta from typing import Optional from jose import JWTError, jwt from fastapi import Depends, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials, APIKeyHeader from backend.config import get_settings settings = get_settings() security = HTTPBearer(auto_error=False) def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: to_encode = data.copy() expire = datetime.utcnow() + (expires_delta or timedelta(minutes=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES)) to_encode.update({"exp": expire}) return jwt.encode(to_encode, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM) def decode_token(token: str) -> Optional[dict]: try: return jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM]) except JWTError: return None async def get_current_user_id(credentials: HTTPAuthorizationCredentials = Depends(security)) -> int: if credentials is None: raise HTTPException(status_code=401, detail="需要登录") payload = decode_token(credentials.credentials) if payload is None: raise HTTPException(status_code=401, detail="token 无效") uid = payload.get("sub") if uid is None: raise HTTPException(status_code=401, detail="token 无效") return int(uid) async def get_current_user_role(credentials: HTTPAuthorizationCredentials = Depends(security)) -> str: if credentials is None: return "student" payload = decode_token(credentials.credentials) if payload is None: return "student" return payload.get("role", "student") def require_teacher(role: str = Depends(get_current_user_role)): if role != "teacher": raise HTTPException(status_code=403, detail="需要教师权限") return role