""" Dependencies for FastAPI routes """ import logging from typing import Generator, Optional, Any, Dict from fastapi import Depends, HTTPException, status, Request from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials # Initialize global testers knowledge_graph_tester = None prompt_reconstructor = None # Import knowledge graph tester conditionally try: from agentgraph.prompt_tester import KnowledgeGraphTester from agentgraph.prompt_reconstructor import PromptReconstructor TESTER_AVAILABLE = True # Update type annotations after imports knowledge_graph_tester: Optional[KnowledgeGraphTester] = None prompt_reconstructor: Optional[PromptReconstructor] = None except ImportError: TESTER_AVAILABLE = False # Import database utilities from backend.database import get_db from sqlalchemy.orm import Session # Setup logger logger = logging.getLogger("agent_monitoring_server") def get_db_session() -> Generator[Session, None, None]: """Get a database session""" session = next(get_db()) try: yield session finally: session.close() def get_knowledge_graph_tester(kg_path: Optional[str] = None) -> Any: """Get or create a KnowledgeGraphTester instance""" global knowledge_graph_tester if not TESTER_AVAILABLE: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Prompt tester module not available" ) if knowledge_graph_tester is None: try: from backend.server_config import DEFAULT_KNOWLEDGE_GRAPH # Use provided path or default knowledge_graph_path = kg_path or DEFAULT_KNOWLEDGE_GRAPH knowledge_graph_tester = KnowledgeGraphTester( knowledge_graph_path=knowledge_graph_path, model="gpt-5-mini" # Default model ) logger.info(f"Initialized KnowledgeGraphTester with {knowledge_graph_path}") except Exception as e: logger.error(f"Failed to initialize KnowledgeGraphTester: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error initializing tester: {str(e)}" ) return knowledge_graph_tester def get_prompt_reconstructor() -> Any: """Get or create a PromptReconstructor instance""" global prompt_reconstructor if not TESTER_AVAILABLE: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Prompt reconstructor module not available" ) if prompt_reconstructor is None: try: prompt_reconstructor = PromptReconstructor() logger.info("Initialized PromptReconstructor") except Exception as e: logger.error(f"Failed to initialize PromptReconstructor: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error initializing reconstructor: {str(e)}" ) return prompt_reconstructor # ===== AUTHENTICATION DEPENDENCIES ===== # Optional security scheme for Bearer tokens security = HTTPBearer(auto_error=False) def get_current_user_optional(request: Request) -> Optional[Dict[str, Any]]: """ Get current user from session, but don't raise error if not found. Used for endpoints where authentication is optional. """ from utils.environment import should_enable_auth if not should_enable_auth(): logger.debug("🏠 Auth disabled - no user required") return None # Try to get user from session try: user = request.session.get("user") if user: logger.info(f"🔓 Found authenticated user: {user.get('username', 'unknown')}") return user else: # Add detailed debugging for session contents session_keys = list(request.session.keys()) if hasattr(request.session, 'keys') else [] logger.warning(f"🔍 No user found in session for {request.url.path}. Session keys: {session_keys}") # HF Spaces specific debugging from utils.environment import is_huggingface_space if is_huggingface_space(): # Log HF-specific session debugging info cookies = request.cookies logger.warning(f"🍪 HF Spaces cookies: {list(cookies.keys())}") logger.warning(f"🍪 Session cookie present: {'session' in cookies}") # Default SessionMiddleware cookie name # Check if session middleware is properly initialized session_data = dict(request.session) if hasattr(request.session, 'keys') else {} logger.warning(f"🔍 HF Session data: {session_data}") # Check request headers that might affect session relevant_headers = { 'host': request.headers.get('host'), 'x-forwarded-for': request.headers.get('x-forwarded-for'), 'x-forwarded-proto': request.headers.get('x-forwarded-proto'), 'user-agent': request.headers.get('user-agent', '')[:50] + '...', } logger.warning(f"🔍 HF Request headers: {relevant_headers}") else: logger.debug(f"🔍 Full session contents: {dict(request.session) if hasattr(request.session, 'keys') else 'no session'}") return None except Exception as e: logger.error(f"Session access failed: {e}") return None def get_current_user_required(request: Request) -> Dict[str, Any]: """ Get current user from session, raise HTTPException if not authenticated. Used for endpoints that require authentication. """ from utils.environment import should_enable_auth if not should_enable_auth(): logger.debug("🏠 Auth disabled - returning mock user") return { "id": "local_dev", "username": "local_user", "name": "Local Development User", "auth_method": "local_dev" } user = get_current_user_optional(request) if not user: logger.warning(f"🚫 Authentication required for {request.url.path}") raise HTTPException( status_code=401, detail={ "error": "Authentication required", "message": "Please log in with your Hugging Face account", "login_url": "/auth/login" } ) return user def require_auth_in_hf_spaces(request: Request) -> None: """ Dependency that enforces authentication only in HF Spaces. Raises 401 if in HF Spaces and user is not authenticated. """ from utils.environment import should_enable_auth if should_enable_auth(): user = get_current_user_optional(request) if not user: logger.warning(f"🚫 HF Spaces requires authentication for {request.url.path}") # Check if this is an API request or web request is_api_request = ( request.url.path.startswith("/api/") or request.headers.get("accept", "").startswith("application/json") or request.headers.get("content-type", "").startswith("application/json") ) if is_api_request: # For API requests, return JSON error raise HTTPException( status_code=401, detail={ "error": "Authentication required in Hugging Face Spaces", "message": "Please log in to access this service", "login_url": "/auth/login-page", "reason": "This prevents abuse of OpenAI resources" } ) else: # For web requests, include redirect info in the error # This will be handled by our custom exception handler raise HTTPException( status_code=401, detail={ "error": "Authentication required in Hugging Face Spaces", "message": "Please log in to access this service", "redirect_to": "/auth/login-page", "reason": "This prevents abuse of OpenAI resources" } ) # Convenience aliases for common use cases CurrentUser = Depends(get_current_user_required) OptionalUser = Depends(get_current_user_optional) RequireHFAuth = Depends(require_auth_in_hf_spaces)