""" Conditional Authentication Middleware This middleware only enables authentication when running in Hugging Face Spaces. Local development bypasses authentication entirely. """ import os import logging from typing import Optional, Dict, Any, Callable, Awaitable from fastapi import Request, Response, HTTPException from starlette.responses import RedirectResponse, JSONResponse from starlette.types import ASGIApp, Receive, Scope, Send from utils.environment import should_enable_auth, get_oauth_config, is_huggingface_space logger = logging.getLogger(__name__) class ConditionalAuthMiddleware: """ ASGI middleware that conditionally enables authentication based on deployment environment. - In HF Spaces: Full OAuth authentication required - In local development: Authentication bypassed """ def __init__(self, app: ASGIApp, excluded_paths: Optional[list] = None): self.app = app # Paths that don't require authentication even in HF Spaces self.excluded_paths = excluded_paths or [ "/docs", "/redoc", "/openapi.json", "/api/observability/health-check", "/api/observability/environment", "/auth/login", "/auth/callback", "/auth/oauth-callback", # OAuth callback from HF "/auth/logout", "/auth/login-page", "/auth/status", "/assets/", "/static/", # Note: Removed "/" and "/agentgraph" to force authentication ] # Check if auth should be enabled self.auth_enabled = should_enable_auth() self.oauth_config = get_oauth_config() if self.auth_enabled else None # Log auth status if self.auth_enabled: logger.info("🔐 Authentication middleware ENABLED (HF Spaces environment)") if not self.oauth_config: logger.warning("⚠️ OAuth configuration not found in HF Spaces environment") else: logger.info("🏠 Authentication middleware DISABLED (Local development)") async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: """ ASGI callable that processes HTTP requests. """ if scope["type"] != "http": await self.app(scope, receive, send) return request = Request(scope, receive) # If auth is disabled (local dev), bypass all authentication if not self.auth_enabled: logger.debug(f"🏠 Auth disabled - allowing {request.url.path}") await self.app(scope, receive, send) return # If auth is enabled but OAuth not properly configured, log warning and continue if not self.oauth_config: logger.warning("OAuth not configured properly, bypassing auth") await self.app(scope, receive, send) return # Check if path is excluded from authentication if self._is_excluded_path(request.url.path): logger.debug(f"🚪 Excluded path - allowing {request.url.path}") await self.app(scope, receive, send) return # Log the authentication check logger.info(f"🔐 Checking authentication for {request.url.path}") # Check user authentication user = await self._get_current_user(request) if not user: # 🔐 MANDATORY AUTHENTICATION: Protect OpenAI API usage # All users must be authenticated to prevent abuse of OpenAI resources logger.warning(f"🚫 Unauthorized access attempt to {request.url.path} from {request.client.host if request.client else 'unknown'}") # For API calls, return JSON error with login instructions if request.url.path.startswith("/api/"): response = JSONResponse( status_code=401, content={ "error": "Authentication required to access OpenAI-powered features", "message": "Please log in with your Hugging Face account to use this service", "login_url": "/auth/login", "reason": "API access requires user authentication for security and usage tracking" } ) else: # For web requests, redirect to login page response = RedirectResponse(url="/auth/login-page", status_code=302) await response(scope, receive, send) return # Add user info to request state scope["state"] = scope.get("state", {}) scope["state"]["user"] = user await self.app(scope, receive, send) def _is_excluded_path(self, path: str) -> bool: """Check if the request path should bypass authentication.""" return any( path.startswith(excluded_path) for excluded_path in self.excluded_paths ) async def _get_current_user(self, request: Request) -> Optional[Dict[str, Any]]: """ Get current user from session or token. For HF Spaces, we check: 1. Session cookies (our own auth) 2. HF __sign parameter (HF pre-auth) 3. Authorization header """ # Check if user info is in session (our own auth) user = None try: user = request.session.get("user") if user: logger.info(f"🔓 Found authenticated user in session: {user.get('username', 'unknown')}") else: logger.debug(f"🔍 No user found in session for {request.url.path}") except (AttributeError, AssertionError) as e: # Session middleware not available or not configured logger.error(f"Session access failed: {e}") user = None # In HF Spaces, check for __sign parameter which indicates HF has pre-authenticated the user if not user and is_huggingface_space(): sign_param = request.query_params.get("__sign") if sign_param: # HF has authenticated the user via __sign parameter # We'll create a basic user object to indicate authentication user = { "id": "hf_authenticated", "name": "HF User", "username": "hf_user", "email": None, "avatar_url": None, "auth_method": "hf_sign" } # Store in session for future requests try: request.session["user"] = user except (AttributeError, AssertionError): logger.debug("Cannot store user in session - session middleware not available") logger.info("User authenticated via HF __sign parameter") # Check Authorization header as fallback if not user: auth_header = request.headers.get("Authorization") if auth_header and auth_header.startswith("Bearer "): # In a full implementation, validate this token with HF API # For now, we'll assume it's valid if present in HF environment if is_huggingface_space(): user = { "id": "hf_bearer_auth", "name": "HF Bearer User", "username": "hf_bearer_user", "email": None, "avatar_url": None, "auth_method": "bearer_token" } return user def get_auth_status(self) -> Dict[str, Any]: """Get current authentication configuration status.""" return { "auth_enabled": self.auth_enabled, "environment": "huggingface_spaces" if is_huggingface_space() else "local_development", "oauth_configured": bool(self.oauth_config), "excluded_paths": self.excluded_paths, }