member-platform/backend/app/api/auth.py

import logging
import secrets

import httpx
from fastapi import APIRouter, HTTPException, status

from app.core.config import get_settings
from app.schemas.login import LoginRequest, LoginResponse, OIDCAuthUrlResponse, OIDCCodeExchangeRequest, RefreshTokenRequest

router = APIRouter(prefix="/auth", tags=["auth"])
logger = logging.getLogger(__name__)


@router.post("/login", response_model=LoginResponse)
def login(payload: LoginRequest) -> LoginResponse:
    settings = get_settings()
    client_id = settings.idp_client_id or settings.idp_audience
    if not settings.idp_base_url or not client_id or not settings.idp_client_secret:
        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="idp_login_not_configured")

    form = {
        "grant_type": "password",
        "client_id": client_id,
        "client_secret": settings.idp_client_secret,
        "username": payload.username,
        "password": payload.password,
        "scope": "openid profile email",
    }
    try:
        resp = httpx.post(
            settings.idp_token_endpoint,
            data=form,
            timeout=10,
            verify=settings.idp_verify_tls,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
        )
    except Exception as exc:
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="idp_unreachable") from exc

    if resp.status_code >= 400:
        logger.warning("idp password grant failed: status=%s body=%s", resp.status_code, resp.text)
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid_username_or_password")

    data = resp.json()
    token = data.get("access_token")
    if not token:
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="idp_missing_access_token")
    return _build_login_response(data)


@router.get("/oidc/url", response_model=OIDCAuthUrlResponse)
def get_oidc_authorize_url(
    redirect_uri: str,
    login_hint: str | None = None,
    prompt: str = "login",
    idp_hint: str | None = None,
    code_challenge: str | None = None,
    code_challenge_method: str | None = None,
) -> OIDCAuthUrlResponse:
    settings = get_settings()
    client_id = settings.idp_client_id or settings.idp_audience
    if not settings.idp_base_url or not client_id:
        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="idp_login_not_configured")

    query = {
        "client_id": client_id,
        "response_type": "code",
        "scope": "openid profile email",
        "redirect_uri": redirect_uri,
        "state": secrets.token_urlsafe(24),
        "prompt": prompt or "login",
    }
    if login_hint:
        query["login_hint"] = login_hint
    if idp_hint:
        query["kc_idp_hint"] = idp_hint
    if code_challenge:
        query["code_challenge"] = code_challenge
        query["code_challenge_method"] = code_challenge_method or "S256"

    params = httpx.QueryParams(query)
    return OIDCAuthUrlResponse(authorize_url=f"{settings.idp_authorize_endpoint}?{params}")


@router.post("/oidc/exchange", response_model=LoginResponse)
def exchange_oidc_code(payload: OIDCCodeExchangeRequest) -> LoginResponse:
    settings = get_settings()
    client_id = settings.idp_client_id or settings.idp_audience
    if not settings.idp_base_url or not client_id or not settings.idp_client_secret:
        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="idp_login_not_configured")

    form = {
        "grant_type": "authorization_code",
        "client_id": client_id,
        "client_secret": settings.idp_client_secret,
        "code": payload.code,
        "redirect_uri": payload.redirect_uri,
    }
    if payload.code_verifier:
        form["code_verifier"] = payload.code_verifier

    try:
        resp = httpx.post(
            settings.idp_token_endpoint,
            data=form,
            timeout=10,
            verify=settings.idp_verify_tls,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
        )
    except Exception as exc:
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="idp_unreachable") from exc

    if resp.status_code >= 400:
        logger.warning("idp auth-code exchange failed: status=%s body=%s", resp.status_code, resp.text)
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="idp_code_exchange_failed")

    data = resp.json()
    token = data.get("access_token")
    if not token:
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="idp_missing_access_token")
    return _build_login_response(data)


@router.post("/refresh", response_model=LoginResponse)
def refresh_access_token(payload: RefreshTokenRequest) -> LoginResponse:
    settings = get_settings()
    client_id = settings.idp_client_id or settings.idp_audience
    if not settings.idp_base_url or not client_id or not settings.idp_client_secret:
        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="idp_login_not_configured")

    form = {
        "grant_type": "refresh_token",
        "client_id": client_id,
        "client_secret": settings.idp_client_secret,
        "refresh_token": payload.refresh_token,
    }
    try:
        resp = httpx.post(
            settings.idp_token_endpoint,
            data=form,
            timeout=10,
            verify=settings.idp_verify_tls,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
        )
    except Exception as exc:
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="idp_unreachable") from exc

    if resp.status_code >= 400:
        logger.warning("idp refresh-token grant failed: status=%s body=%s", resp.status_code, resp.text)
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid_refresh_token")

    data = resp.json()
    token = data.get("access_token")
    if not token:
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="idp_missing_access_token")
    return _build_login_response(data)


def _build_login_response(data: dict) -> LoginResponse:
    return LoginResponse(
        access_token=data.get("access_token", ""),
        refresh_token=data.get("refresh_token"),
        token_type=data.get("token_type", "Bearer"),
        expires_in=data.get("expires_in"),
        refresh_expires_in=data.get("refresh_expires_in"),
        scope=data.get("scope"),
    )