256 lines
9.8 KiB
Python
256 lines
9.8 KiB
Python
# gateway/app/middleware/auth.py
|
|
"""
|
|
Enhanced Authentication Middleware for API Gateway with Tenant Access Control
|
|
"""
|
|
|
|
import structlog
|
|
from fastapi import Request, HTTPException
|
|
from fastapi.responses import JSONResponse
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
from starlette.responses import Response
|
|
from typing import Optional, Dict, Any
|
|
|
|
from app.core.config import settings
|
|
from shared.auth.jwt_handler import JWTHandler
|
|
from shared.auth.tenant_access import tenant_access_manager, extract_tenant_id_from_path, is_tenant_scoped_path
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
# JWT handler for local token validation
|
|
jwt_handler = JWTHandler(settings.JWT_SECRET_KEY, settings.JWT_ALGORITHM)
|
|
|
|
# Routes that don't require authentication
|
|
PUBLIC_ROUTES = [
|
|
"/health",
|
|
"/metrics",
|
|
"/docs",
|
|
"/redoc",
|
|
"/openapi.json",
|
|
"/api/v1/auth/login",
|
|
"/api/v1/auth/register",
|
|
"/api/v1/auth/refresh",
|
|
"/api/v1/auth/verify",
|
|
"/api/v1/nominatim/search"
|
|
]
|
|
|
|
class AuthMiddleware(BaseHTTPMiddleware):
|
|
"""
|
|
Enhanced Authentication Middleware with Tenant Access Control
|
|
"""
|
|
|
|
def __init__(self, app, redis_client=None):
|
|
super().__init__(app)
|
|
self.redis_client = redis_client # For caching and rate limiting
|
|
|
|
async def dispatch(self, request: Request, call_next) -> Response:
|
|
"""Process request with enhanced authentication and tenant access control"""
|
|
|
|
# Skip authentication for OPTIONS requests (CORS preflight)
|
|
if request.method == "OPTIONS":
|
|
return await call_next(request)
|
|
|
|
# Skip authentication for public routes
|
|
if self._is_public_route(request.url.path):
|
|
return await call_next(request)
|
|
|
|
# ✅ STEP 1: Extract and validate JWT token
|
|
token = self._extract_token(request)
|
|
if not token:
|
|
logger.warning(f"Missing token for protected route: {request.url.path}")
|
|
return JSONResponse(
|
|
status_code=401,
|
|
content={"detail": "Authentication required"}
|
|
)
|
|
|
|
# ✅ STEP 2: Verify token and get user context
|
|
# Pass self.redis_client to _verify_token to enable caching
|
|
user_context = await self._verify_token(token)
|
|
if not user_context:
|
|
logger.warning(f"Invalid token for route: {request.url.path}")
|
|
return JSONResponse(
|
|
status_code=401,
|
|
content={"detail": "Invalid or expired token"}
|
|
)
|
|
|
|
# ✅ STEP 3: Extract tenant context from URL using shared utility
|
|
tenant_id = extract_tenant_id_from_path(request.url.path)
|
|
|
|
# ✅ STEP 4: Verify tenant access if this is a tenant-scoped route
|
|
if tenant_id and is_tenant_scoped_path(request.url.path):
|
|
# Use TenantAccessManager for gateway-level verification with caching
|
|
# Ensure tenant_access_manager uses the redis_client from the middleware
|
|
if self.redis_client and tenant_access_manager.redis_client is None:
|
|
tenant_access_manager.redis_client = self.redis_client
|
|
|
|
has_access = await tenant_access_manager.verify_basic_tenant_access( # Corrected method call
|
|
user_context["user_id"],
|
|
tenant_id
|
|
)
|
|
|
|
if not has_access:
|
|
logger.warning(f"User {user_context['email']} denied access to tenant {tenant_id}")
|
|
return JSONResponse(
|
|
status_code=403,
|
|
content={"detail": f"Access denied to tenant {tenant_id}"}
|
|
)
|
|
|
|
# Set tenant context in request state
|
|
request.state.tenant_id = tenant_id
|
|
request.state.tenant_verified = True
|
|
|
|
logger.debug(f"Tenant access verified",
|
|
user_id=user_context["user_id"],
|
|
tenant_id=tenant_id,
|
|
path=request.url.path)
|
|
|
|
# ✅ STEP 5: Inject user context into request
|
|
request.state.user = user_context
|
|
request.state.authenticated = True
|
|
|
|
# ✅ STEP 6: Add context headers for downstream services
|
|
self._inject_context_headers(request, user_context, tenant_id)
|
|
|
|
logger.debug(f"Authenticated request",
|
|
user_email=user_context['email'],
|
|
tenant_id=tenant_id,
|
|
path=request.url.path)
|
|
|
|
return await call_next(request)
|
|
|
|
def _is_public_route(self, path: str) -> bool:
|
|
"""Check if route requires authentication"""
|
|
return any(path.startswith(route) for route in PUBLIC_ROUTES)
|
|
|
|
def _extract_token(self, request: Request) -> Optional[str]:
|
|
"""Extract JWT token from Authorization header"""
|
|
auth_header = request.headers.get("Authorization")
|
|
if auth_header and auth_header.startswith("Bearer "):
|
|
return auth_header.split(" ")[1]
|
|
return None
|
|
|
|
async def _verify_token(self, token: str) -> Optional[Dict[str, Any]]:
|
|
"""Verify JWT token with fallback strategy"""
|
|
|
|
# Try local JWT validation first (fast)
|
|
try:
|
|
payload = jwt_handler.verify_token(token)
|
|
if payload and self._validate_token_payload(payload):
|
|
logger.debug("Token validated locally")
|
|
return payload
|
|
except Exception as e:
|
|
logger.debug(f"Local token validation failed: {e}")
|
|
|
|
# Check cache for recently validated tokens
|
|
if self.redis_client:
|
|
try:
|
|
cached_user = await self._get_cached_user(token)
|
|
if cached_user:
|
|
logger.debug("Token found in cache")
|
|
return cached_user
|
|
except Exception as e:
|
|
logger.warning(f"Cache lookup failed: {e}")
|
|
|
|
# Verify with auth service (authoritative)
|
|
try:
|
|
user_context = await self._verify_with_auth_service(token)
|
|
if user_context:
|
|
# Cache successful validation
|
|
if self.redis_client:
|
|
await self._cache_user(token, user_context)
|
|
logger.debug("Token validated by auth service")
|
|
return user_context
|
|
except Exception as e:
|
|
logger.error(f"Auth service validation failed: {e}")
|
|
|
|
return None
|
|
|
|
def _validate_token_payload(self, payload: Dict[str, Any]) -> bool:
|
|
"""Validate JWT payload has required fields"""
|
|
required_fields = ["user_id", "email", "exp"]
|
|
return all(field in payload for field in required_fields)
|
|
|
|
async def _verify_with_auth_service(self, token: str) -> Optional[Dict[str, Any]]:
|
|
"""Verify token with auth service"""
|
|
try:
|
|
import httpx
|
|
async with httpx.AsyncClient(timeout=3.0) as client:
|
|
response = await client.post(
|
|
f"{settings.AUTH_SERVICE_URL}/api/v1/auth/verify",
|
|
headers={"Authorization": f"Bearer {token}"}
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
logger.warning(f"Auth service returned {response.status_code}")
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"Auth service error: {e}")
|
|
return None
|
|
|
|
async def _get_cached_user(self, token: str) -> Optional[Dict[str, Any]]:
|
|
"""Get user context from cache"""
|
|
if not self.redis_client:
|
|
return None
|
|
|
|
cache_key = f"auth:token:{hash(token)}"
|
|
try:
|
|
cached_data = await self.redis_client.get(cache_key)
|
|
if cached_data:
|
|
import json
|
|
if isinstance(cached_data, bytes):
|
|
cached_data = cached_data.decode()
|
|
return json.loads(cached_data)
|
|
except Exception as e:
|
|
logger.warning(f"Cache get failed: {e}")
|
|
return None
|
|
|
|
async def _cache_user(self, token: str, user_context: Dict[str, Any], ttl: int = 300):
|
|
"""Cache user context for 5 minutes"""
|
|
if not self.redis_client:
|
|
return
|
|
|
|
cache_key = f"auth:token:{hash(token)}"
|
|
try:
|
|
import json
|
|
await self.redis_client.setex(cache_key, ttl, json.dumps(user_context))
|
|
except Exception as e:
|
|
logger.warning(f"Cache set failed: {e}")
|
|
|
|
def _inject_context_headers(self, request: Request, user_context: Dict[str, Any], tenant_id: Optional[str]):
|
|
"""Inject authentication and tenant headers for downstream services"""
|
|
|
|
# Remove any existing auth headers to prevent spoofing
|
|
headers_to_remove = [
|
|
"x-user-id", "x-user-email", "x-user-role",
|
|
"x-tenant-id", "x-tenant-verified", "x-authenticated"
|
|
]
|
|
|
|
for header in headers_to_remove:
|
|
request.headers.__dict__["_list"] = [
|
|
(k, v) for k, v in request.headers.raw
|
|
if k.lower() != header.lower()
|
|
]
|
|
|
|
# Inject new headers
|
|
new_headers = [
|
|
(b"x-authenticated", b"true"),
|
|
(b"x-user-id", str(user_context.get("user_id", "")).encode()),
|
|
(b"x-user-email", str(user_context.get("email", "")).encode()),
|
|
(b"x-user-role", str(user_context.get("role", "user")).encode()),
|
|
]
|
|
|
|
# Add tenant context if verified
|
|
if tenant_id:
|
|
new_headers.extend([
|
|
(b"x-tenant-id", tenant_id.encode()),
|
|
(b"x-tenant-verified", b"true")
|
|
])
|
|
|
|
# Add headers to request
|
|
request.headers.__dict__["_list"].extend(new_headers)
|
|
|
|
logger.debug(f"Injected context headers",
|
|
user_id=user_context.get("user_id"),
|
|
tenant_id=tenant_id) |