Spaces:
Running
Running
File size: 8,810 Bytes
c2ea5ed 97068b2 c2ea5ed 7bc750c c2ea5ed 97068b2 041fce6 97068b2 be074f1 1c1facd 2431039 1c1facd 97068b2 188f8a3 97068b2 188f8a3 97068b2 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 |
"""
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)
|