53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
from typing import Dict, Any
|
|
import httpx
|
|
from fastapi import HTTPException
|
|
from common.config.app_settings import app_settings
|
|
from common.log.log_utils import log_entry_exit_async
|
|
from common.log.module_logger import ModuleLogger
|
|
|
|
|
|
class ApiKeyIntrospectHandler:
|
|
"""
|
|
Freeleaps Auth Service API Key Introspect Handle
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.module_logger = ModuleLogger(sender_id=ApiKeyIntrospectHandler.__name__)
|
|
self.auth_service_base = app_settings.AUTH_SERVICE_ENDPOINT
|
|
|
|
|
|
@log_entry_exit_async
|
|
async def api_key_introspect(self, api_key: str) -> Dict[str, Any]:
|
|
"""
|
|
Introspect API key by calling external auth service
|
|
|
|
Args:
|
|
api_key: The API key to introspect
|
|
|
|
Returns:
|
|
Dictionary containing the API key details
|
|
|
|
Raises:
|
|
HTTPException: If the external service call fails
|
|
"""
|
|
api_url = self.auth_service_base + "introspect_api_key"
|
|
await self.module_logger.log_info(f"Starting API Key validation for key: {api_key[:8]}...")
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.post(
|
|
api_url,
|
|
json={"api_key": api_key}
|
|
)
|
|
|
|
if response.status_code != 200:
|
|
error_detail = response.json() if response.content else {"error": "Unknown error"}
|
|
await self.module_logger.log_error(f"API Key validation failed - Status: {response.status_code}, Error: {error_detail}")
|
|
raise HTTPException(
|
|
status_code=response.status_code,
|
|
detail=error_detail
|
|
)
|
|
|
|
validation_result = response.json()
|
|
await self.module_logger.log_info(f"API Key validation successful - Active: {validation_result.get('active', False)}")
|
|
return validation_result
|