knai/school-system

1from datetime import datetime, timedelta2from typing import Optional3from jose import JWTError, jwt4from passlib.context import CryptContext5from fastapi import Depends, HTTPException, status6from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials7from config import settings8from sqlalchemy.orm import Session9from database import get_db10 11pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")12bearer_scheme = HTTPBearer(auto_error=False)13 14def verify_password(plain_password: str, hashed_password: str) -> bool:15    return pwd_context.verify(plain_password, hashed_password)16 17def hash_password(password: str) -> str:18    return pwd_context.hash(password)19 20def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:21    to_encode = data.copy()22    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES))23    to_encode.update({"exp": expire})24    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)25 26def decode_token(token: str) -> dict:27    try:28        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])29        return payload30    except JWTError:31        return {}32 33def add_token_to_blacklist(db: Session, token: str, user_id: int, expires_at: datetime):34    from models import TokenBlacklist35    entry = TokenBlacklist(token=token, user_id=user_id, expires_at=expires_at)36    db.add(entry)37    db.commit()38 39 40def is_token_blacklisted(db: Session, token: str) -> bool:41    from models import TokenBlacklist42    return db.query(TokenBlacklist).filter(TokenBlacklist.token == token).first() is not None43 44 45async def get_current_user(46    credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),47    db: Session = Depends(get_db),48):49    if not credentials:50        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")51 52    payload = decode_token(credentials.credentials)53    if not payload.get("user_id"):54        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")55 56    if is_token_blacklisted(db, credentials.credentials):57        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token has been revoked")58 59    return payload60 61async def require_admin(current_user: dict = Depends(get_current_user)):62    if current_user.get("role") != "admin":63        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")64    return current_user65 66async def require_teacher_or_admin(current_user: dict = Depends(get_current_user)):67    if current_user.get("role") not in ["admin", "teacher"]:68        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Teacher or Admin access required")69    return current_user70 71 72async def require_super_admin(current_user: dict = Depends(get_current_user)):73    """Any super admin (global or school-level) — school_id filter still applies."""74    if not current_user.get("is_super_admin"):75        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Super Admin access required.")76    return current_user77 78 79async def require_global_super_admin(current_user: dict = Depends(get_current_user)):80    """Only the GLOBAL super admin (school_id=None) — manages schools themselves."""81    if not current_user.get("is_super_admin") or current_user.get("school_id") is not None:82        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Global Super Admin access required.")83    return current_user84