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

import time
from collections.abc import Generator
from uuid import UUID

import jwt
from fastapi import Depends, HTTPException, Path, Query, status
from fastapi.security import OAuth2PasswordBearer
from fastapi.security.utils import get_authorization_scheme_param
from sqlalchemy import exists, select
from sqlalchemy.orm import Session
from starlette.requests import Request

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 (
    GlobalRole,
    OrgRole,
    Project,
    User,
    UserOrganization,
    association_table_commission_member,
)

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


async def get_token_flexible(
    request: Request,
    token: str | None = Query(default=None),
) -> str:
    """Same as reusable_oauth2, but also accepts ?token=... in the query
    string — needed because native EventSource cannot set custom headers."""
    auth_header = request.headers.get("Authorization")
    if auth_header:
        scheme, param = get_authorization_scheme_param(auth_header)
        if scheme.lower() == "bearer" and param:
            return param
    if token:
        return token
    raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Not authenticated")


async def get_current_user_flexible(
    session: Session = Depends(get_session),
    token: str = Depends(get_token_flexible),
) -> User:
    # identical body to get_current_user, just sourcing `token` differently
    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_org_role_sse(*allowed_roles: OrgRole):
    def dependency(
        project_id: UUID = Path(...),
        session: Session = Depends(get_session),
        current_user: User = Depends(get_current_user_flexible),
    ) -> 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)
            raise HTTPException(
                status.HTTP_403_FORBIDDEN, "Insufficient permissions for this organization"
            )
        return current_user

    return dependency


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: UUID, *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
    stmt = select(
        exists().where(
            association_table_commission_member.c.commission_id == commission_id,
            association_table_commission_member.c.user_id == user_id,
        )
    )
    return session.scalar(stmt)


def require_org_role(*allowed_roles: OrgRole):
    def dependency(
        project_id: UUID = 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 assert_commission_ownership(
    session: Session, current_user: User, project_id: UUID, 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: UUID = 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 == str(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