Python FastAPI E-Learning Part 4: Penguatan Keamanan Enterprise – 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="Autentikasi dibutuhkan",
            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="Token tidak valid atau telah kedaluwarsa",
            headers={"WWW-Authenticate": "Bearer"},
        )

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

    # Ambil role dan profil pengguna dalam 1 query terindeks
    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="Pengguna tidak ditemukan atau akun dinonaktifkan"
        )

    # Sematkan klaim impersonation secara dinamis pada entitas User
    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 untuk menegakkan izin hak akses peran (RBAC)"""
    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"Akses ditolak. Memerlukan salah satu peran: {', '.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"])),  # Hanya Super Admin yang berhak
):
    ...

# 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"]))
):
    # Cegah impersonasi diri sendiri
    if user_id == admin_user.id:
        raise HTTPException(status_code=400, detail="Tidak dapat meng-impersonate diri sendiri")

    # Ambil data pengguna target
    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="Pengguna target tidak ditemukan")

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

    # Terbitkan token stateless khusus dengan audit metadata
    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)  # Sesi impersonasi otomatis kedaluwarsa dalam 2 jam
    )

    auth_data = build_auth_response(target_user, token, admin_user.id)
    return ApiResponse.ok(
        data=auth_data,
        message=f"Beralih identitas ke {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)
):
    # Validasi bahwa token yang aktif memang sesi impersonasi
    if not getattr(current_user, "is_impersonating", False):
        raise HTTPException(status_code=400, detail="Saat ini tidak dalam sesi impersonasi")

    original_admin_id = getattr(current_user, "original_admin_id", None)
    if not original_admin_id:
        raise HTTPException(status_code=400, detail="Identitas administrator asli tidak ditemukan")

    # Ambil kembali data administrator asli
    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="Akun administrator asli sudah tidak valid")

    # Terbitkan kembali token administrator bersih
    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="Sesi administrator berhasil dipulihkan"
    )

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

# 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  # Batas maksimal 25 MB

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:
        # Bersihkan karakter manipulasi path dan karakter liar
        clean_name = os.path.basename(filename)
        clean_name = re.sub(r"[^a-zA-Z0-9_.-]", "_", clean_name)
        # Tambahkan prefix UUID untuk mencegah tabrakan nama dan tebakan nama berkas
        return f"{uuid.uuid4().hex}_{clean_name}"

    async def save_file(self, upload_file: UploadFile) -> str:
        # 1. Validasi ekstensi berkas
        ext = Path(upload_file.filename).suffix.lower()
        if ext not in ALLOWED_EXTENSIONS:
            raise HTTPException(status_code=400, detail=f"Ekstensi berkas {ext} tidak diizinkan")

        # 2. Validasi MIME type
        if upload_file.content_type not in ALLOWED_MIME_TYPES:
            raise HTTPException(status_code=400, detail=f"Tipe konten {upload_file.content_type} tidak diizinkan")

        # 3. Baca berkas dengan proteksi ukuran
        content = await upload_file.read()
        if len(content) > MAX_FILE_SIZE_BYTES:
            raise HTTPException(status_code=400, detail="Ukuran berkas melebihi batas 25MB")

        # 4. Batasi penyimpanan mutlak hanya di dalam base_dir
        safe_name = self.sanitize_filename(upload_file.filename)
        destination = (self.base_dir / safe_name).resolve()

        # Proteksi Path Traversal: Pastikan path tujuan tidak keluar dari base_dir
        if not str(destination).startswith(str(self.base_dir)):
            raise HTTPException(status_code=400, detail="Percobaan path traversal terdeteksi")

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

        return safe_name

storage_service = StorageService()