""" Authentication router — Microsoft Entra ID (Azure AD) OAuth2 + internal JWT. Flow: 1. GET /auth/microsoft/login → returns the Microsoft authorization URL 2. POST /auth/microsoft/callback → exchanges code for Microsoft token, validates @creditfair.in domain, issues internal JWT 3. GET /auth/me → verifies current JWT, returns user info 4. POST /auth/token → legacy password login (emergency/API access only) """ import urllib.parse from datetime import timedelta import httpx from fastapi import APIRouter, Depends, HTTPException, status from fastapi.security import OAuth2PasswordRequestForm from pydantic import BaseModel from app.config.config import settings from app.middleware.auth import create_access_token, get_current_user router = APIRouter(prefix="/auth", tags=["Authentication"]) # ─── Schema models ──────────────────────────────────────────────────────────── class Token(BaseModel): access_token: str token_type: str expires_in: int # seconds user_email: str | None = None user_name: str | None = None class UserInfo(BaseModel): username: str email: str | None = None expires_in: int class MicrosoftCallbackRequest(BaseModel): code: str redirect_uri: str class MicrosoftLoginUrlResponse(BaseModel): authorization_url: str # ─── Microsoft OAuth helpers ────────────────────────────────────────────────── def _build_microsoft_auth_url(redirect_uri: str) -> str: """Constructs the Microsoft Entra ID authorization URL.""" scopes = " ".join(settings.oauth_scopes_list) params = { "client_id": settings.OAUTH_APP_ID, "response_type": "code", "redirect_uri": redirect_uri, "response_mode": "query", "scope": scopes, "prompt": "select_account", # always show account picker } base = f"https://login.microsoftonline.com/{settings.OAUTH_TENANT_ID}/oauth2/v2.0/authorize" return base + "?" + urllib.parse.urlencode(params) async def _exchange_code_for_token(code: str, redirect_uri: str) -> dict: """Exchanges an auth code for Microsoft tokens via the token endpoint.""" token_url = ( f"https://login.microsoftonline.com/{settings.OAUTH_TENANT_ID}/oauth2/v2.0/token" ) payload = { "client_id": settings.OAUTH_APP_ID, "client_secret": settings.OAUTH_APP_SECRET, "grant_type": "authorization_code", "code": code, "redirect_uri": redirect_uri, "scope": " ".join(settings.oauth_scopes_list), } async with httpx.AsyncClient(timeout=15.0) as client: resp = await client.post(token_url, data=payload) if resp.status_code != 200: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Failed to exchange authorization code with Microsoft. Code may be expired or invalid.", ) return resp.json() async def _get_microsoft_user_info(access_token: str) -> dict: """Fetches the logged-in user's profile from Microsoft Graph.""" async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.get( "https://graph.microsoft.com/v1.0/me", headers={"Authorization": f"Bearer {access_token}"}, ) if resp.status_code != 200: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not retrieve user profile from Microsoft Graph.", ) return resp.json() # ─── Endpoints ──────────────────────────────────────────────────────────────── @router.get( "/microsoft/login", response_model=MicrosoftLoginUrlResponse, summary="Get Microsoft OAuth2 authorization URL", ) async def microsoft_login(): """ Returns the Microsoft Entra ID authorization URL. The frontend should redirect the user to this URL to begin the OAuth flow. """ if not settings.OAUTH_APP_ID or not settings.OAUTH_TENANT_ID: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Microsoft OAuth is not configured on this server. Set OAUTH_APP_ID and OAUTH_TENANT_ID.", ) url = _build_microsoft_auth_url(settings.OAUTH_REDIRECT_URI) return MicrosoftLoginUrlResponse(authorization_url=url) @router.post( "/microsoft/callback", response_model=Token, summary="Exchange Microsoft auth code for internal JWT", ) async def microsoft_callback(body: MicrosoftCallbackRequest): """ Called by the frontend after the Microsoft redirect. 1. Exchanges the authorization code for a Microsoft access token. 2. Fetches the user's email from Microsoft Graph. 3. Validates that the email belongs to @creditfair.in. 4. Issues an internal JWT for the LMS portal session. """ # 1. Exchange code for Microsoft token ms_tokens = await _exchange_code_for_token(body.code, body.redirect_uri) # 2. Fetch user profile ms_access_token = ms_tokens.get("access_token") user_info = await _get_microsoft_user_info(ms_access_token) # Prefer userPrincipalName (always an email), fall back to mail email: str = user_info.get("userPrincipalName") or user_info.get("mail") or "" display_name: str = user_info.get("displayName") or email # 3. Domain enforcement — only @creditfair.in email_lower = email.lower().strip() allowed_domain = settings.OAUTH_ALLOWED_DOMAIN.lower().strip() if not email_lower.endswith(f"@{allowed_domain}"): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=( f"Access denied. Only @{settings.OAUTH_ALLOWED_DOMAIN} accounts " "are permitted to access this portal." ), ) # 4. Issue internal JWT access_token = create_access_token( data={"sub": email_lower, "name": display_name}, expires_delta=timedelta(minutes=settings.JWT_EXPIRE_MINUTES), ) return Token( access_token=access_token, token_type="bearer", expires_in=settings.JWT_EXPIRE_MINUTES * 60, user_email=email_lower, user_name=display_name, ) @router.get( "/me", response_model=UserInfo, summary="Verify token and return current user info", ) async def whoami(current_user: dict = Depends(get_current_user)): return UserInfo( username=current_user["username"], email=current_user.get("email"), expires_in=settings.JWT_EXPIRE_MINUTES * 60, ) @router.post( "/token", response_model=Token, summary="[Emergency] Static admin login — not for regular use", include_in_schema=not settings.is_production, # hide from Swagger in production ) # Duplicate token endpoint decorator removed async def legacy_login(form_data: OAuth2PasswordRequestForm = Depends()): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Endpoint disabled", )