Python FastAPI E-Learning Part 4: Enterprise Security Hardening – Stateless JWT, RBAC & 1-Click Impersonation

# backend/app/core/security.py
from datetime import datetime, timedelta
from typing import Optional, Any, Union, List
from jose import jwt, JWTError
from passlib.context import CryptContext
from app.core.config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(
    subject: Union[str, Any],
    roles: List[str],
    extra_claims: Optional[dict] = None,
    expires_delta: Optional[timedelta] = None
) -> str:
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    
    to_encode = {
        "sub": str(subject),
        "roles": roles,
        "exp": expire,
        "iat": datetime.utcnow()
    }
    if extra_claims:
        to_encode.update(extra_claims)
        
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)

def decode_access_token(token: str) -> Optional[dict]:
    try:
        return jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
    except JWTError:
        return None

# backend/app/api/deps.py
from typing import List, Optional
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from sqlalchemy.orm import selectinload

from app.core.database import get_db
from app.core.security import decode_access_token
from app.models.user import User

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login", auto_error=False)

async def get_current_user(
    token: Optional[str] = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db)
) -> User:
    if not token:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Not authenticated",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    payload = decode_access_token(token)
    if not payload or not payload.get("sub"):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired token",
            headers={"WWW-Authenticate": "Bearer"},
        )

    user_id = int(payload.get("sub"))

    # Eager load roles & profile relations in a single query
    stmt = (
        select(User)
        .options(
            selectinload(User.roles),
            selectinload(User.teacher_profile),
            selectinload(User.student_profile)
        )
        .where(User.id == user_id, User.deleted_at.is_(None))
    )
    user = (await db.execute(stmt)).scalars().first()

    if not user or not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User not found or inactive"
        )

    # Attach impersonation claims dynamically to the User entity
    setattr(user, "is_impersonating", payload.get("is_impersonating", False))
    setattr(user, "original_admin_id", payload.get("original_admin_id", None))

    return user

def require_roles(allowed_roles: List[str]):
    """Higher-order dependency enforcing strict role memberships"""
    def role_checker(current_user: User = Depends(get_current_user)) -> User:
        user_roles = [r.name for r in current_user.roles]
        if not any(role in user_roles for role in allowed_roles):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Access denied. Requires one of: {', '.join(allowed_roles)}",
            )
        return current_user
    return role_checker

@router.delete("/subjects/{id}")
async def delete_subject(
    id: int,
    db: AsyncSession = Depends(get_db),
    admin: User = Depends(require_roles(["ROLE_ADMIN"])),  # Enforces Super Admin only
):
    ...

# backend/app/api/v1/auth.py
@router.post("/admin/impersonate/{user_id}", response_model=ApiResponse[AuthResponse])
async def impersonate_user(
    user_id: int,
    db: AsyncSession = Depends(get_db),
    admin_user: User = Depends(require_roles(["ROLE_ADMIN"]))
):
    # Prevent self-impersonation loops
    if user_id == admin_user.id:
        raise HTTPException(status_code=400, detail="Cannot impersonate yourself")

    # Fetch target user with associated roles and profiles
    stmt = (
        select(User)
        .options(
            selectinload(User.roles),
            selectinload(User.teacher_profile),
            selectinload(User.student_profile)
        )
        .where(User.id == user_id, User.deleted_at.is_(None))
    )
    target_user = (await db.execute(stmt)).scalars().first()
    if not target_user:
        raise HTTPException(status_code=404, detail="Target user not found")

    target_roles = [r.name for r in target_user.roles]

    # Mint a special stateless token carrying administrative audit claims
    token = create_access_token(
        subject=target_user.id,
        roles=target_roles,
        extra_claims={
            "is_impersonating": True,
            "original_admin_id": admin_user.id,
        },
        expires_delta=timedelta(hours=2)  # Impersonation sessions expire automatically
    )

    auth_data = build_auth_response(target_user, token, admin_user.id)
    return ApiResponse.ok(
        data=auth_data,
        message=f"Now impersonating {target_user.name} ({target_roles[0]})"
    )

@router.post("/admin/stop-impersonate", response_model=ApiResponse[AuthResponse])
async def stop_impersonate(
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    # Verify that the current token was actually an impersonated session
    if not getattr(current_user, "is_impersonating", False):
        raise HTTPException(status_code=400, detail="Not currently impersonating")

    original_admin_id = getattr(current_user, "original_admin_id", None)
    if not original_admin_id:
        raise HTTPException(status_code=400, detail="Missing original administrator identity")

    # Re-fetch the original administrator
    stmt = (
        select(User)
        .options(selectinload(User.roles))
        .where(User.id == int(original_admin_id), User.deleted_at.is_(None))
    )
    admin_user = (await db.execute(stmt)).scalars().first()
    if not admin_user or "ROLE_ADMIN" not in [r.name for r in admin_user.roles]:
        raise HTTPException(status_code=403, detail="Original admin account no longer valid")

    # Mint a fresh, clean Administrator Token
    admin_token = create_access_token(
        subject=admin_user.id,
        roles=[r.name for r in admin_user.roles]
    )

    return ApiResponse.ok(
        data=build_auth_response(admin_user, admin_token),
        message="Restored administrator session"
    )

with open(f"/var/www/uploads/{file.filename}", "wb") as f:  # VULNERABLE!

# backend/app/common/storage.py
import os
import uuid
import re
from pathlib import Path
from fastapi import UploadFile, HTTPException

ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".pdf", ".docx", ".xlsx", ".mp3"}
ALLOWED_MIME_TYPES = {
    "image/jpeg", "image/png", "application/pdf",
    "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
    "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
    "audio/mpeg",
}
MAX_FILE_SIZE_BYTES = 25 * 1024 * 1024  # 25 MB limit

class StorageService:
    def __init__(self, base_dir: str = "storage/uploads"):
        self.base_dir = Path(base_dir).resolve()
        self.base_dir.mkdir(parents=True, exist_ok=True)

    def sanitize_filename(self, filename: str) -> str:
        # Strip path traversal attempts and sanitize characters
        clean_name = os.path.basename(filename)
        clean_name = re.sub(r"[^a-zA-Z0-9_.-]", "_", clean_name)
        # Unique UUID prefix prevents collision and path prediction
        return f"{uuid.uuid4().hex}_{clean_name}"

    async def save_file(self, upload_file: UploadFile) -> str:
        # 1. Validate file extension
        ext = Path(upload_file.filename).suffix.lower()
        if ext not in ALLOWED_EXTENSIONS:
            raise HTTPException(status_code=400, detail=f"File extension {ext} not permitted")

        # 2. Validate MIME type
        if upload_file.content_type not in ALLOWED_MIME_TYPES:
            raise HTTPException(status_code=400, detail=f"Content-Type {upload_file.content_type} not permitted")

        # 3. Read content with size guard
        content = await upload_file.read()
        if len(content) > MAX_FILE_SIZE_BYTES:
            raise HTTPException(status_code=400, detail="File exceeds maximum allowed size of 25MB")

        # 4. Enforce strict directory confinement
        safe_name = self.sanitize_filename(upload_file.filename)
        destination = (self.base_dir / safe_name).resolve()

        # Security check: Ensure destination is strictly inside base_dir
        if not str(destination).startswith(str(self.base_dir)):
            raise HTTPException(status_code=400, detail="Path traversal attempt detected")

        with open(destination, "wb") as f:
            f.write(content)

        return safe_name

storage_service = StorageService()