47 lines
1.5 KiB
Python
47 lines
1.5 KiB
Python
"""Authentication middleware for Heicode integration."""
|
|
from fastapi import Request, HTTPException, status, Depends
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
from config.settings import settings
|
|
from config.error_codes import ErrorCode
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
security = HTTPBearer()
|
|
|
|
|
|
async def verify_service_token(
|
|
credentials: HTTPAuthorizationCredentials = Depends(security)
|
|
) -> str:
|
|
"""Verify service token from mcp-server.
|
|
|
|
Phase 1-4: Simple pre-shared token validation.
|
|
Phase 5: Migrate to AKS Workload Identity.
|
|
"""
|
|
token = credentials.credentials
|
|
|
|
if token != settings.HEICODE_SERVICE_TOKEN:
|
|
logger.warning("Invalid service token attempt")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={
|
|
"success": False,
|
|
"error": {
|
|
"code": ErrorCode.INVALID_TOKEN,
|
|
"message": "Invalid service token",
|
|
"request_id": None
|
|
}
|
|
}
|
|
)
|
|
|
|
return token
|
|
|
|
|
|
def extract_headers(request: Request) -> dict:
|
|
"""Extract required headers for correlation and audit."""
|
|
return {
|
|
"correlation_id": request.headers.get("X-Correlation-Id"),
|
|
"user_id": request.headers.get("X-User-Id"),
|
|
"binding_scope": request.headers.get("X-Binding-Scope"),
|
|
"idempotency_key": request.headers.get("X-Idempotency-Key") or request.headers.get("Idempotency-Key"),
|
|
}
|