""" Connection routes for the backend API. Implements the endpoints for connection management. SECURITY NOTE: - Regular connections endpoint (/api/connections/) only returns connections for the current user - Admin endpoint (/api/connections/admin/all) provides access to all connections for management purposes - This prevents security vulnerabilities where admin users could see other users' connections """ 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, Token from modules.security.auth import getCurrentUser, limiter from modules.interfaces.interfaceAppObjects import getInterface, getRootInterface from modules.shared.timezoneUtils import get_utc_timestamp # Configure logger logger = logging.getLogger(__name__) def get_token_status_for_connection(interface, connection_id: str) -> tuple[str, Optional[float]]: """ Get token status and expiration for a connection. Args: interface: The database interface connection_id: The connection ID to check Returns: tuple: (token_status, token_expires_at) - token_status: 'active', 'expired', or 'none' - token_expires_at: UTC timestamp or None """ try: # Query tokens table for the latest token for this connection tokens = interface.db.getRecordset( Token, recordFilter={"connectionId": connection_id} ) if not tokens: return "none", None # Find the most recent token (highest createdAt timestamp) latest_token = None latest_created_at = 0 for token_data in tokens: created_at = token_data.get("createdAt", 0) if created_at > latest_created_at: latest_created_at = created_at latest_token = token_data if not latest_token: return "none", None # Check if token is expired expires_at = latest_token.get("expiresAt") if not expires_at: return "none", None current_time = get_utc_timestamp() if expires_at <= current_time: return "expired", expires_at else: return "active", expires_at except Exception as e: logger.error(f"Error getting token status for connection {connection_id}: {str(e)}") return "none", None 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 SECURITY: This endpoint is secure - users can only see their own connections. """ try: interface = getInterface(currentUser) # SECURITY FIX: All users (including admins) can only see their own connections # This prevents admin from seeing other users' connections and causing confusion connections = interface.getUserConnections(currentUser.id) # Enhance each connection with token status information enhanced_connections = [] for connection in connections: # Get token status for this connection token_status, token_expires_at = get_token_status_for_connection(interface, connection.id) # Create enhanced connection with token status enhanced_connection = UserConnection( id=connection.id, userId=connection.userId, authority=connection.authority, externalId=connection.externalId, externalUsername=connection.externalUsername, externalEmail=connection.externalEmail, status=connection.status, connectedAt=connection.connectedAt, lastChecked=connection.lastChecked, expiresAt=connection.expiresAt, tokenStatus=token_status, tokenExpiresAt=token_expires_at ) enhanced_connections.append(enhanced_connection) return enhanced_connections 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 SECURITY: This endpoint is secure - it always creates connections for the current user and cannot be used to create connections for other users. """ 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 ) # Save connection record - models now handle timestamp serialization automatically interface.db.recordModify(UserConnection, connection.id, connection.to_dict()) 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 for the current user SECURITY: This endpoint is secure - users can only update their own connections. """ try: interface = getInterface(currentUser) # Find the connection connection = None # SECURITY FIX: All users (including admins) can only update their own connections # This prevents admin from updating other users' connections and causing confusion 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 using UTC timestamp connection.lastChecked = get_utc_timestamp() # Update connection - models now handle timestamp serialization automatically interface.db.recordModify(UserConnection, connectionId, connection.to_dict()) # Get token status for the updated connection token_status, token_expires_at = get_token_status_for_connection(interface, connectionId) # Create enhanced connection with token status enhanced_connection = UserConnection( id=connection.id, userId=connection.userId, authority=connection.authority, externalId=connection.externalId, externalUsername=connection.externalUsername, externalEmail=connection.externalEmail, status=connection.status, connectedAt=connection.connectedAt, lastChecked=connection.lastChecked, expiresAt=connection.expiresAt, tokenStatus=token_status, tokenExpiresAt=token_expires_at ) return enhanced_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 a service for the current user SECURITY: This endpoint is secure - users can only connect their own connections. """ try: interface = getInterface(currentUser) # Find the connection connection = None # SECURITY FIX: All users (including admins) can only connect their own connections # This prevents admin from connecting other users' connections and causing confusion 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 a service for the current user SECURITY: This endpoint is secure - users can only disconnect their own connections. """ try: interface = getInterface(currentUser) # Find the connection connection = None # SECURITY FIX: All users (including admins) can only disconnect their own connections # This prevents admin from disconnecting other users' connections and causing confusion 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 = get_utc_timestamp() # Update connection record - models now handle timestamp serialization automatically interface.db.recordModify(UserConnection, connectionId, connection.to_dict()) 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 for the current user SECURITY: This endpoint is secure - users can only delete their own connections. """ try: interface = getInterface(currentUser) # Find the connection connection = None # SECURITY FIX: All users (including admins) can only delete their own connections # This prevents admin from deleting other users' connections and causing confusion 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)}" )