""" OAuth Service - Handles Google and Microsoft OAuth authentication """ import secrets import logging from datetime import datetime, timedelta from typing import Optional, Dict, Any, Tuple from urllib.parse import urlencode import httpx from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, or_ from app.config import settings from app.models.user import User, UserRole from app.auth.jwt_handler import create_access_token, create_refresh_token, get_token_data logger = logging.getLogger(__name__) class OAuthProvider: """Base OAuth provider class""" def __init__(self, client_id: str, client_secret: str, redirect_uri: str): self.client_id = client_id self.client_secret = client_secret self.redirect_uri = redirect_uri def get_authorization_url(self, state: str) -> str: raise NotImplementedError async def exchange_code(self, code: str) -> Dict[str, Any]: raise NotImplementedError async def get_user_info(self, access_token: str) -> Dict[str, Any]: raise NotImplementedError class GoogleOAuthProvider(OAuthProvider): """Google OAuth 2.0 provider""" AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth" TOKEN_URL = "https://oauth2.googleapis.com/token" USER_INFO_URL = "https://www.googleapis.com/oauth2/v3/userinfo" SCOPES = ["openid", "email", "profile"] def get_authorization_url(self, state: str) -> str: params = { "client_id": self.client_id, "redirect_uri": self.redirect_uri, "response_type": "code", "scope": " ".join(self.SCOPES), "state": state, "access_type": "offline", "prompt": "consent" } return f"{self.AUTH_URL}?{urlencode(params)}" async def exchange_code(self, code: str) -> Dict[str, Any]: async with httpx.AsyncClient() as client: response = await client.post( self.TOKEN_URL, data={ "client_id": self.client_id, "client_secret": self.client_secret, "code": code, "grant_type": "authorization_code", "redirect_uri": self.redirect_uri } ) if response.status_code != 200: logger.error(f"Google token exchange failed: {response.text}") raise Exception("Failed to exchange authorization code") return response.json() async def get_user_info(self, access_token: str) -> Dict[str, Any]: async with httpx.AsyncClient() as client: response = await client.get( self.USER_INFO_URL, headers={"Authorization": f"Bearer {access_token}"} ) if response.status_code != 200: logger.error(f"Google user info failed: {response.text}") raise Exception("Failed to get user info") data = response.json() return { "id": data.get("sub"), "email": data.get("email"), "email_verified": data.get("email_verified", False), "first_name": data.get("given_name", ""), "last_name": data.get("family_name", ""), "avatar_url": data.get("picture"), "provider": "google" } class MicrosoftOAuthProvider(OAuthProvider): """Microsoft OAuth 2.0 provider""" SCOPES = ["openid", "email", "profile", "User.Read"] def __init__(self, client_id: str, client_secret: str, redirect_uri: str, tenant_id: str = "common"): super().__init__(client_id, client_secret, redirect_uri) self.tenant_id = tenant_id self.AUTH_URL = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/authorize" self.TOKEN_URL = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token" self.USER_INFO_URL = "https://graph.microsoft.com/v1.0/me" def get_authorization_url(self, state: str) -> str: params = { "client_id": self.client_id, "redirect_uri": self.redirect_uri, "response_type": "code", "scope": " ".join(self.SCOPES), "state": state, "response_mode": "query" } return f"{self.AUTH_URL}?{urlencode(params)}" async def exchange_code(self, code: str) -> Dict[str, Any]: async with httpx.AsyncClient() as client: response = await client.post( self.TOKEN_URL, data={ "client_id": self.client_id, "client_secret": self.client_secret, "code": code, "grant_type": "authorization_code", "redirect_uri": self.redirect_uri, "scope": " ".join(self.SCOPES) } ) if response.status_code != 200: logger.error(f"Microsoft token exchange failed: {response.text}") raise Exception("Failed to exchange authorization code") return response.json() async def get_user_info(self, access_token: str) -> Dict[str, Any]: async with httpx.AsyncClient() as client: response = await client.get( self.USER_INFO_URL, headers={"Authorization": f"Bearer {access_token}"} ) if response.status_code != 200: logger.error(f"Microsoft user info failed: {response.text}") raise Exception("Failed to get user info") data = response.json() return { "id": data.get("id"), "email": data.get("mail") or data.get("userPrincipalName"), "email_verified": True, # Microsoft accounts are verified "first_name": data.get("givenName", ""), "last_name": data.get("surname", ""), "avatar_url": None, # Would need separate Graph API call "provider": "microsoft" } class OAuthService: """Service for managing OAuth authentication""" def __init__(self): self.providers: Dict[str, OAuthProvider] = {} self._state_store: Dict[str, Dict[str, Any]] = {} # In production, use Redis # Initialize Google provider if settings.GOOGLE_CLIENT_ID and settings.GOOGLE_CLIENT_SECRET: self.providers["google"] = GoogleOAuthProvider( client_id=settings.GOOGLE_CLIENT_ID, client_secret=settings.GOOGLE_CLIENT_SECRET, redirect_uri=settings.GOOGLE_REDIRECT_URI ) # Initialize Microsoft provider if settings.MICROSOFT_CLIENT_ID and settings.MICROSOFT_CLIENT_SECRET: self.providers["microsoft"] = MicrosoftOAuthProvider( client_id=settings.MICROSOFT_CLIENT_ID, client_secret=settings.MICROSOFT_CLIENT_SECRET, redirect_uri=settings.MICROSOFT_REDIRECT_URI, tenant_id=settings.MICROSOFT_TENANT_ID ) def get_available_providers(self) -> list: """Get list of configured OAuth providers""" return list(self.providers.keys()) def generate_state(self, provider: str, redirect_url: Optional[str] = None) -> str: """Generate and store OAuth state parameter""" state = secrets.token_urlsafe(32) self._state_store[state] = { "provider": provider, "redirect_url": redirect_url or settings.FRONTEND_URL, "created_at": datetime.utcnow() } return state def validate_state(self, state: str) -> Optional[Dict[str, Any]]: """Validate OAuth state parameter""" state_data = self._state_store.pop(state, None) if not state_data: return None # Check if state is expired (10 minutes) if datetime.utcnow() - state_data["created_at"] > timedelta(minutes=10): return None return state_data def get_authorization_url(self, provider: str, redirect_url: Optional[str] = None) -> Tuple[str, str]: """Get OAuth authorization URL""" if provider not in self.providers: raise ValueError(f"OAuth provider '{provider}' not configured") state = self.generate_state(provider, redirect_url) auth_url = self.providers[provider].get_authorization_url(state) return auth_url, state async def handle_callback( self, provider: str, code: str, state: str, db: AsyncSession ) -> Tuple[User, Dict[str, str]]: """ Handle OAuth callback Returns (user, tokens) tuple """ # Validate state state_data = self.validate_state(state) if not state_data or state_data["provider"] != provider: raise ValueError("Invalid state parameter") if provider not in self.providers: raise ValueError(f"OAuth provider '{provider}' not configured") oauth_provider = self.providers[provider] # Exchange code for tokens token_data = await oauth_provider.exchange_code(code) access_token = token_data.get("access_token") refresh_token = token_data.get("refresh_token") expires_in = token_data.get("expires_in", 3600) # Get user info user_info = await oauth_provider.get_user_info(access_token) # Find or create user user = await self._find_or_create_user( db=db, provider=provider, oauth_id=user_info["id"], email=user_info["email"], first_name=user_info["first_name"], last_name=user_info["last_name"], avatar_url=user_info.get("avatar_url"), access_token=access_token, refresh_token=refresh_token, token_expires_in=expires_in ) # Generate JWT tokens token_data = get_token_data(user) tokens = { "access_token": create_access_token(token_data), "refresh_token": create_refresh_token(token_data), "token_type": "bearer", "redirect_url": state_data["redirect_url"] } return user, tokens async def _find_or_create_user( self, db: AsyncSession, provider: str, oauth_id: str, email: str, first_name: str, last_name: Optional[str], avatar_url: Optional[str], access_token: str, refresh_token: Optional[str], token_expires_in: int ) -> User: """Find existing user or create new one""" # First, try to find by OAuth provider and ID result = await db.execute( select(User).where( User.oauth_provider == provider, User.oauth_id == oauth_id ) ) user = result.scalar_one_or_none() if user: # Update OAuth tokens user.oauth_access_token = access_token user.oauth_refresh_token = refresh_token user.oauth_token_expires_at = datetime.utcnow() + timedelta(seconds=token_expires_in) user.last_login_at = datetime.utcnow() await db.commit() return user # Try to find by email result = await db.execute( select(User).where(User.email == email) ) user = result.scalar_one_or_none() if user: # Link existing account to OAuth user.oauth_provider = provider user.oauth_id = oauth_id user.oauth_access_token = access_token user.oauth_refresh_token = refresh_token user.oauth_token_expires_at = datetime.utcnow() + timedelta(seconds=token_expires_in) user.last_login_at = datetime.utcnow() user.is_verified = True user.email_verified_at = datetime.utcnow() if avatar_url and not user.avatar_url: user.avatar_url = avatar_url await db.commit() return user # Create new user user = User( email=email, first_name=first_name, last_name=last_name, password_hash="", # No password for OAuth users avatar_url=avatar_url, role=UserRole.CLIENT_USER, is_active=True, is_verified=True, email_verified_at=datetime.utcnow(), oauth_provider=provider, oauth_id=oauth_id, oauth_access_token=access_token, oauth_refresh_token=refresh_token, oauth_token_expires_at=datetime.utcnow() + timedelta(seconds=token_expires_in), last_login_at=datetime.utcnow() ) db.add(user) await db.commit() await db.refresh(user) logger.info(f"Created new OAuth user: {email} via {provider}") return user async def unlink_provider( self, db: AsyncSession, user: User, provider: str ) -> bool: """Unlink OAuth provider from user account""" if user.oauth_provider != provider: return False # Ensure user has a password before unlinking if not user.password_hash: raise ValueError("Cannot unlink OAuth - user has no password set") user.oauth_provider = None user.oauth_id = None user.oauth_access_token = None user.oauth_refresh_token = None user.oauth_token_expires_at = None await db.commit() return True # Singleton instance oauth_service = OAuthService()