""" Connection routes for the backend API. Implements the endpoints for connection management. """ from fastapi import APIRouter, HTTPException, Depends, Body, Path, Request, Response from typing import List, Dict, Any, Optional from fastapi import status from datetime import datetime import logging import json from modules.interfaces.interfaceAppModel import User, UserConnection, AuthAuthority, ConnectionStatus from modules.security.auth import getCurrentUser, limiter from modules.interfaces.interfaceAppObjects import getInterface, getRootInterface # Configure logger logger = logging.getLogger(__name__) router = APIRouter( prefix="/api/connections", tags=["Manage Connections"], responses={404: {"description": "Not found"}} ) @router.get("/", response_model=List[UserConnection]) @limiter.limit("30/minute") async def get_connections( request: Request, currentUser: User = Depends(getCurrentUser) ) -> List[UserConnection]: """Get all connections for the current user or all connections if admin""" try: interface = getInterface(currentUser) # Clear connections cache to ensure fresh data interface.db.clearTableCache("connections") if currentUser.privilege in ['admin', 'sysadmin']: # Admins can see all connections users = interface.getAllUsers() connections = [] for user in users: connections.extend(interface.getUserConnections(user.id)) return connections else: # Regular users can only see their own connections return interface.getUserConnections(currentUser.id) except Exception as e: logger.error(f"Error getting connections: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to get connections: {str(e)}" ) @router.post("/", response_model=UserConnection) @limiter.limit("10/minute") async def create_connection( request: Request, connection_data: Dict[str, Any] = Body(...), currentUser: User = Depends(getCurrentUser) ) -> UserConnection: """Create a new connection for the current user""" try: interface = getInterface(currentUser) # Map type to authority authority_map = { 'msft': AuthAuthority.MSFT, 'google': AuthAuthority.GOOGLE } authority = authority_map.get(connection_data.get('type')) if not authority: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Unsupported connection type: {connection_data.get('type')}" ) # Get fresh copy of user from database user = interface.getUser(currentUser.id) if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) # Always create a new connection with PENDING status connection = interface.addUserConnection( userId=currentUser.id, authority=authority, externalId="", # Will be set after OAuth externalUsername="", # Will be set after OAuth status=ConnectionStatus.PENDING # Start with PENDING status ) # Convert connection to dict and ensure datetime fields are serialized connection_dict = connection.to_dict() for field in ['connectedAt', 'lastChecked', 'expiresAt']: if field in connection_dict and connection_dict[field] is not None: if isinstance(connection_dict[field], datetime): connection_dict[field] = connection_dict[field].isoformat() elif isinstance(connection_dict[field], (int, float)): connection_dict[field] = datetime.fromtimestamp(connection_dict[field]).isoformat() # Save connection record interface.db.recordModify("connections", connection.id, connection_dict) # Clear cache to ensure fresh data interface.db.clearTableCache("connections") return connection except HTTPException: raise except Exception as e: logger.error(f"Error creating connection: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to create connection: {str(e)}" ) @router.put("/{connectionId}", response_model=UserConnection) @limiter.limit("10/minute") async def update_connection( request: Request, connectionId: str = Path(..., description="The ID of the connection to update"), connection_data: Dict[str, Any] = Body(...), currentUser: User = Depends(getCurrentUser) ) -> UserConnection: """Update an existing connection""" try: interface = getInterface(currentUser) # Find the connection connection = None if currentUser.privilege in ['admin', 'sysadmin']: # Admins can update any connection users = interface.getAllUsers() for user in users: connections = interface.getUserConnections(user.id) for conn in connections: if conn.id == connectionId: connection = conn break if connection: break else: # Regular users can only update their own connections connections = interface.getUserConnections(currentUser.id) for conn in connections: if conn.id == connectionId: connection = conn break if not connection: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Connection not found" ) # Update connection fields for field, value in connection_data.items(): if hasattr(connection, field): setattr(connection, field, value) # Update lastChecked timestamp connection.lastChecked = datetime.now() # Convert connection to dict and ensure datetime fields are serialized connection_dict = connection.to_dict() for field in ['connectedAt', 'lastChecked', 'expiresAt']: if field in connection_dict and connection_dict[field] is not None: if isinstance(connection_dict[field], datetime): connection_dict[field] = connection_dict[field].isoformat() elif isinstance(connection_dict[field], (int, float)): connection_dict[field] = datetime.fromtimestamp(connection_dict[field]).isoformat() # Update connection interface.db.recordModify("connections", connectionId, connection_dict) # Clear cache to ensure fresh data interface.db.clearTableCache("connections") # Get updated connection return connection except HTTPException: raise except Exception as e: logger.error(f"Error updating connection: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to update connection: {str(e)}" ) @router.post("/{connectionId}/connect") @limiter.limit("10/minute") async def connect_service( request: Request, connectionId: str = Path(..., description="The ID of the connection to connect"), currentUser: User = Depends(getCurrentUser) ) -> Dict[str, Any]: """Connect to an external service""" try: interface = getInterface(currentUser) # Find the connection connection = None if currentUser.privilege in ['admin', 'sysadmin']: # Admins can connect any connection users = interface.getAllUsers() for user in users: connections = interface.getUserConnections(user.id) for conn in connections: if conn.id == connectionId: connection = conn break if connection: break else: # Regular users can only connect their own connections connections = interface.getUserConnections(currentUser.id) for conn in connections: if conn.id == connectionId: connection = conn break if not connection: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Connection not found" ) # Initiate OAuth flow with state=connect auth_url = None if connection.authority == AuthAuthority.MSFT: # Use the same login endpoint with state=connect to ensure account selection # Include current user ID in state state_data = { "type": "connect", "connectionId": connectionId, "userId": currentUser.id # Add current user ID } auth_url = f"/api/msft/login?state={json.dumps(state_data)}" elif connection.authority == AuthAuthority.GOOGLE: state_data = { "type": "connect", "connectionId": connectionId, "userId": currentUser.id # Add current user ID } auth_url = f"/api/google/login?state={json.dumps(state_data)}" else: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Unsupported authority: {connection.authority}" ) return {"authUrl": auth_url} except HTTPException: raise except Exception as e: logger.error(f"Error connecting service: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to connect service: {str(e)}" ) @router.post("/{connectionId}/disconnect") @limiter.limit("10/minute") async def disconnect_service( request: Request, connectionId: str = Path(..., description="The ID of the connection to disconnect"), currentUser: User = Depends(getCurrentUser) ) -> Dict[str, Any]: """Disconnect from an external service""" try: interface = getInterface(currentUser) # Find the connection connection = None if currentUser.privilege in ['admin', 'sysadmin']: # Admins can disconnect any connection users = interface.getAllUsers() for user in users: connections = interface.getUserConnections(user.id) for conn in connections: if conn.id == connectionId: connection = conn break if connection: break else: # Regular users can only disconnect their own connections connections = interface.getUserConnections(currentUser.id) for conn in connections: if conn.id == connectionId: connection = conn break if not connection: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Connection not found" ) # Update connection status connection.status = ConnectionStatus.INACTIVE connection.lastChecked = datetime.now() # Update connection record interface.db.recordModify("connections", connectionId, connection.to_dict()) # Clear cache to ensure fresh data interface.db.clearTableCache("connections") return {"message": "Service disconnected successfully"} except HTTPException: raise except Exception as e: logger.error(f"Error disconnecting service: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to disconnect service: {str(e)}" ) @router.delete("/{connectionId}") @limiter.limit("10/minute") async def delete_connection( request: Request, connectionId: str = Path(..., description="The ID of the connection to delete"), currentUser: User = Depends(getCurrentUser) ) -> Dict[str, Any]: """Delete a connection""" try: interface = getInterface(currentUser) # Find the connection connection = None if currentUser.privilege in ['admin', 'sysadmin']: # Admins can delete any connection users = interface.getAllUsers() for user in users: connections = interface.getUserConnections(user.id) for conn in connections: if conn.id == connectionId: connection = conn break if connection: break else: # Regular users can only delete their own connections connections = interface.getUserConnections(currentUser.id) for conn in connections: if conn.id == connectionId: connection = conn break if not connection: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Connection not found" ) # Remove the connection - only need connectionId since permissions are verified interface.removeUserConnection(connectionId) return {"message": "Connection deleted successfully"} except HTTPException: raise except Exception as e: logger.error(f"Error deleting connection: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to delete connection: {str(e)}" )