bdlg2023-sms-back/app/api/deps.py

import time
from collections.abc import Generator

import jwt
from fastapi import Depends, HTTPException, Path, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy import UUID, exists, select
from sqlalchemy.orm import Session

from app.api.utils import get_project_organization_id
from app.core import config, security
from app.core.session import session
from app.models import (
    CommissionMember,
    GlobalRole,
    OrgRole,
    Project,
    User,
    UserOrganization,
)

reusable_oauth2 = OAuth2PasswordBearer(tokenUrl="auth/access-token")

FULL_ACCESS_ROLES = (OrgRole.ORG_ADMIN, OrgRole.RESPO_BENEVOLE)


def get_session() -> Generator[Session, None, None]:
    with session() as db:
        yield db


async def get_current_user(
    session: Session = Depends(get_session), token: str = Depends(reusable_oauth2)
) -> User:
    try:
        payload = jwt.decode(token, config.settings.SECRET_KEY, algorithms=[security.JWT_ALGORITHM])
    except jwt.DecodeError:
        raise HTTPException(status.HTTP_403_FORBIDDEN, "Could not validate credentials.")

    token_data = security.JWTTokenPayload(**payload)

    if token_data.refresh:
        raise HTTPException(
            status.HTTP_403_FORBIDDEN, "Could not validate credentials, cannot use refresh token"
        )

    now = int(time.time())
    if now < token_data.issued_at or now > token_data.expires_at:
        raise HTTPException(
            status.HTTP_403_FORBIDDEN,
            "Could not validate credentials, token expired or not yet valid",
        )

    result = session.execute(select(User).where(User.id == token_data.sub))
    user = result.scalars().first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found.")
    return user


def require_super_admin(current_user: User = Depends(get_current_user)) -> User:
    if current_user.global_role != GlobalRole.SUPER_ADMIN:
        raise HTTPException(status.HTTP_403_FORBIDDEN, "Requires super_admin")
    return current_user


def _has_org_role(
    session: Session, user_id: UUID, project_id: str, *allowed_roles: OrgRole
) -> bool:
    """Single query: does user_id hold one of allowed_roles in the
    organization that owns project_id? Replaces loading the user's full
    organizations list and scanning it in Python."""
    return (
        session.execute(
            select(
                exists().where(
                    UserOrganization.user_id == user_id,
                    UserOrganization.role.in_(allowed_roles),
                    UserOrganization.organization_id
                    == (
                        select(Project.organization_id)
                        .where(Project.id == project_id)
                        .scalar_subquery()
                    ),
                )
            )
        ).scalar()
        is True
    )


def _is_commission_member(session: Session, user_id: str, commission_id: str | None) -> bool:
    if commission_id is None:
        return False
    return session.get(CommissionMember, (commission_id, user_id)) is not None


def require_org_role(*allowed_roles: OrgRole):
    def dependency(
        project_id: str = Path(...),
        session: Session = Depends(get_session),
        current_user: User = Depends(get_current_user),
    ) -> User:
        if current_user.global_role == GlobalRole.SUPER_ADMIN:
            return current_user
        if not _has_org_role(session, current_user.id, project_id, *allowed_roles):
            get_project_organization_id(session, project_id)  # raises 404 if project missing
            raise HTTPException(
                status.HTTP_403_FORBIDDEN, "Insufficient permissions for this organization"
            )
        return current_user

    return dependency


def require_commission_membership():
    """For CREATE endpoints (template/slot): confirms org membership only.
    The finer 'does this respo_commission own the commission_id in the
    payload' check happens inside the handler via assert_commission_ownership,
    since the dependency layer can't see the parsed body cheaply."""
    return require_org_role(OrgRole.ORG_ADMIN, OrgRole.RESPO_BENEVOLE, OrgRole.RESPO_COMMISSION)


def assert_commission_ownership(
    session: Session, current_user: User, project_id: str, commission_id: str | None
) -> None:
    """Called inside create/update handlers once the payload's commission_id
    is known. org_admin/respo_benevole: any commission is fine. respo_commission:
    must be a member of commission_id, which must be set."""
    if current_user.global_role == GlobalRole.SUPER_ADMIN:
        return
    if _has_org_role(session, current_user.id, project_id, *FULL_ACCESS_ROLES):
        return
    if commission_id is None or not _is_commission_member(
        session, current_user.id, str(commission_id)
    ):
        raise HTTPException(403, "Cannot assign to a commission you are not a member of")


def require_organization_role(*allowed_roles: OrgRole):
    """Authorize based on the caller's role within the organization named
    directly in the path (organization_id), rather than resolved through a
    project. Super admins always pass."""

    def dependency(
        organization_id: str = Path(...),
        current_user: User = Depends(get_current_user),
    ) -> User:
        if current_user.global_role == GlobalRole.SUPER_ADMIN:
            return current_user

        role = next(
            (m.role for m in current_user.organizations if m.organization_id == organization_id),
            None,
        )
        if role is None or role not in allowed_roles:
            raise HTTPException(
                status.HTTP_403_FORBIDDEN, "Insufficient permissions for this organization"
            )
        return current_user

    return dependency