322 lines
No EOL
12 KiB
Python
322 lines
No EOL
12 KiB
Python
"""
|
|
Routes for Google authentication.
|
|
"""
|
|
|
|
from fastapi import APIRouter, HTTPException, Depends, Request, Response, status, Cookie, Body
|
|
from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse
|
|
import logging
|
|
import json
|
|
from typing import Dict, Any, Optional, List
|
|
from datetime import datetime, timedelta
|
|
|
|
# Import auth module
|
|
import modules.security.auth as auth
|
|
|
|
# Import interfaces
|
|
import modules.interfaces.googleInterface as googleInterface
|
|
import modules.interfaces.gatewayInterface as gatewayInterface
|
|
from modules.interfaces.googleModel import (
|
|
GoogleToken,
|
|
GoogleUserInfo,
|
|
GoogleAuthStatus,
|
|
GoogleTokenResponse,
|
|
GoogleSaveTokenResponse
|
|
)
|
|
|
|
# Configure logger
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Create router for Google Auth endpoints
|
|
router = APIRouter(
|
|
prefix="/api/google",
|
|
tags=["Google"],
|
|
responses={
|
|
404: {"description": "Not found"},
|
|
400: {"description": "Bad request"},
|
|
401: {"description": "Unauthorized"},
|
|
403: {"description": "Forbidden"},
|
|
500: {"description": "Internal server error"}
|
|
}
|
|
)
|
|
|
|
@router.get("/login")
|
|
async def login():
|
|
"""Initiate Google login for the current user"""
|
|
try:
|
|
# Get Google interface with root context for initial setup
|
|
google = googleInterface.getRootInterface()
|
|
|
|
# Get login URL
|
|
auth_url = google.initiateLogin()
|
|
if not auth_url:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to initiate Google login"
|
|
)
|
|
|
|
logger.info("Redirecting to Google login")
|
|
return RedirectResponse(auth_url)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error initiating Google login: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Failed to initiate Google login: {str(e)}"
|
|
)
|
|
|
|
@router.get("/auth/callback")
|
|
async def auth_callback(code: str, state: str, request: Request):
|
|
"""Handle Google OAuth callback"""
|
|
try:
|
|
# Get Google interface with root context for initial setup
|
|
google = googleInterface.getRootInterface()
|
|
|
|
# Handle auth callback
|
|
token_response = google.handleAuthCallback(code)
|
|
if not token_response:
|
|
return HTMLResponse(
|
|
content="""
|
|
<html>
|
|
<head>
|
|
<title>Authentication Failed</title>
|
|
<style>
|
|
body { font-family: Arial, sans-serif; text-align: center; margin-top: 50px; }
|
|
.error { color: red; }
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<h1 class="error">Authentication Failed</h1>
|
|
<p>Could not acquire access token.</p>
|
|
<script>
|
|
setTimeout(() => window.close(), 3000);
|
|
</script>
|
|
</body>
|
|
</html>
|
|
""",
|
|
status_code=400
|
|
)
|
|
|
|
# Get gateway interface for user operations
|
|
gateway = gatewayInterface.getRootInterface()
|
|
|
|
# Check if user exists
|
|
user = gateway.getUserByUsername(token_response.user_info["email"])
|
|
|
|
# If user doesn't exist, create a new user in the default mandate
|
|
if not user:
|
|
try:
|
|
# Get the root mandate ID
|
|
rootMandateId = gateway.getInitialId("mandates")
|
|
if not rootMandateId:
|
|
raise ValueError("Root mandate not found")
|
|
|
|
# Create new user with Google authentication
|
|
user = gateway.createUser(
|
|
username=token_response.user_info["email"],
|
|
email=token_response.user_info["email"],
|
|
fullName=token_response.user_info.get("name", token_response.user_info["email"]),
|
|
mandateId=rootMandateId,
|
|
authenticationAuthority="google"
|
|
)
|
|
logger.info(f"Created new user for Google account: {token_response.user_info['email']}")
|
|
|
|
# Verify user was created by retrieving it
|
|
user = gateway.getUserByUsername(token_response.user_info["email"])
|
|
if not user:
|
|
raise ValueError("Failed to retrieve created user")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to create user for Google account: {str(e)}")
|
|
return HTMLResponse(
|
|
content="""
|
|
<html>
|
|
<head>
|
|
<title>Registration Failed</title>
|
|
<style>
|
|
body { font-family: Arial, sans-serif; text-align: center; margin-top: 50px; }
|
|
.error { color: red; }
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<h1 class="error">Registration Failed</h1>
|
|
<p>Could not create user account.</p>
|
|
<script>
|
|
setTimeout(() => window.close(), 3000);
|
|
</script>
|
|
</body>
|
|
</html>
|
|
""",
|
|
status_code=400
|
|
)
|
|
|
|
# Create backend token
|
|
access_token_expires = timedelta(minutes=auth.ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
access_token = auth.createAccessToken(
|
|
data={
|
|
"sub": user["username"],
|
|
"mandateId": str(user["mandateId"]),
|
|
"userId": str(user["id"]),
|
|
"authenticationAuthority": "google"
|
|
},
|
|
expiresDelta=access_token_expires
|
|
)
|
|
|
|
# Store tokens in session storage for the frontend to pick up
|
|
response = HTMLResponse(
|
|
content=f"""
|
|
<html>
|
|
<head>
|
|
<title>Authentication Successful</title>
|
|
<style>
|
|
body {{ font-family: Arial, sans-serif; text-align: center; margin-top: 50px; }}
|
|
.success {{ color: green; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<h1 class="success">Authentication Successful</h1>
|
|
<p>Welcome, {token_response.user_info.get('name', 'User')}!</p>
|
|
<p>This window will close automatically.</p>
|
|
<script>
|
|
// Store token data in session storage
|
|
sessionStorage.setItem('google_token_data', JSON.stringify({json.dumps(token_response.model_dump())}));
|
|
|
|
// Notify parent window of success
|
|
if (window.opener) {{
|
|
window.opener.postMessage({{
|
|
type: 'google_auth_success',
|
|
user: {json.dumps(token_response.user_info)},
|
|
token_data: {json.dumps(token_response.model_dump())},
|
|
access_token: "{access_token}"
|
|
}}, '*');
|
|
}}
|
|
// Close window after 3 seconds
|
|
setTimeout(() => window.close(), 3000);
|
|
</script>
|
|
</body>
|
|
</html>
|
|
"""
|
|
)
|
|
|
|
return response
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in auth callback: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Authentication failed: {str(e)}"
|
|
)
|
|
|
|
@router.get("/status", response_model=GoogleAuthStatus)
|
|
async def auth_status(currentUser: Dict[str, Any] = Depends(auth.getCurrentActiveUser)):
|
|
"""Check Google authentication status"""
|
|
try:
|
|
# For authenticated endpoints, use the current user's context
|
|
google = googleInterface.getInterface(currentUser)
|
|
|
|
# Get current user token and info
|
|
user_info, access_token = google.getCurrentUserToken()
|
|
|
|
if not user_info or not access_token:
|
|
return GoogleAuthStatus(
|
|
authenticated=False,
|
|
message="Not authenticated with Google"
|
|
)
|
|
|
|
# Convert user_info to GoogleUserInfo model
|
|
user_info_model = GoogleUserInfo(**user_info)
|
|
|
|
return GoogleAuthStatus(
|
|
authenticated=True,
|
|
user=user_info_model
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error checking authentication status: {str(e)}")
|
|
return GoogleAuthStatus(
|
|
authenticated=False,
|
|
message=f"Error checking authentication status: {str(e)}"
|
|
)
|
|
|
|
@router.get("/token", response_model=GoogleTokenResponse)
|
|
async def get_token(currentUser: Dict[str, Any] = Depends(auth.getCurrentActiveUser)):
|
|
"""Get Google token for current user."""
|
|
try:
|
|
# For authenticated endpoints, use the current user's context
|
|
google = googleInterface.getInterface(currentUser)
|
|
|
|
# Get token
|
|
token_data = google.getGoogleToken()
|
|
if not token_data:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="No token found"
|
|
)
|
|
|
|
# Convert to GoogleToken model
|
|
token = GoogleToken(**token_data)
|
|
return GoogleTokenResponse(token=token)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting token: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(e)
|
|
)
|
|
|
|
@router.post("/save-token", response_model=GoogleSaveTokenResponse)
|
|
async def save_token(
|
|
token_data: GoogleToken,
|
|
currentUser: Dict[str, Any] = Depends(auth.getCurrentActiveUser)
|
|
):
|
|
"""Save Google token data from frontend"""
|
|
try:
|
|
# For authenticated endpoints, use the current user's context
|
|
google = googleInterface.getInterface(currentUser)
|
|
|
|
# Save token
|
|
success = google.saveGoogleToken(token_data.model_dump())
|
|
|
|
if success:
|
|
return GoogleSaveTokenResponse(
|
|
success=True,
|
|
message="Token saved successfully",
|
|
token=token_data
|
|
)
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to save token"
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error saving token: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Error saving token: {str(e)}"
|
|
)
|
|
|
|
@router.post("/logout")
|
|
async def logout(currentUser: Dict[str, Any] = Depends(auth.getCurrentActiveUser)):
|
|
"""Logout from Google"""
|
|
try:
|
|
# For authenticated endpoints, use the current user's context
|
|
google = googleInterface.getInterface(currentUser)
|
|
|
|
# Delete token
|
|
success = google.deleteGoogleToken()
|
|
|
|
if success:
|
|
return JSONResponse({
|
|
"message": "Successfully logged out from Google"
|
|
})
|
|
else:
|
|
return JSONResponse({
|
|
"message": "Failed to logout from Google"
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error during logout: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Logout failed: {str(e)}"
|
|
) |