# ================================================================ # shared/auth/tenant_access.py - Complete Implementation # ================================================================ """ Tenant access control utilities for microservices Provides both gateway-level and service-level tenant access verification """ import re from typing import Dict, Any, Optional import httpx import structlog from fastapi import HTTPException, Depends import asyncio # Import FastAPI dependencies from shared.auth.decorators import get_current_user_dep # Import settings (adjust import path based on your project structure) try: from app.core.config import settings except ImportError: try: from core.config import settings except ImportError: # Fallback for different project structures import os class Settings: TENANT_SERVICE_URL = os.getenv("TENANT_SERVICE_URL", "http://tenant-service:8000") settings = Settings() # Setup logging logger = structlog.get_logger() class TenantAccessManager: """ Centralized tenant access management for both gateway and service level """ def __init__(self, redis_client=None): """ Initialize tenant access manager Args: redis_client: Optional Redis client for caching """ self.redis_client = redis_client async def verify_basic_tenant_access(self, user_id: str, tenant_id: str) -> bool: """ Gateway-level: Basic tenant access verification with caching Args: user_id: User ID to verify tenant_id: Tenant ID to check access for Returns: bool: True if user has access to tenant """ # Check cache first (5-minute TTL) cache_key = f"tenant_access:{user_id}:{tenant_id}" if self.redis_client: try: cached_result = await self.redis_client.get(cache_key) if cached_result is not None: return cached_result.decode() == "true" if isinstance(cached_result, bytes) else cached_result == "true" except Exception as cache_error: logger.warning(f"Cache lookup failed: {cache_error}") # Verify with tenant service try: async with httpx.AsyncClient(timeout=2.0) as client: # Short timeout for gateway response = await client.get( f"{settings.TENANT_SERVICE_URL}/api/v1/tenants/{tenant_id}/access/{user_id}" ) has_access = response.status_code == 200 # Cache result (5 minutes) if self.redis_client: try: await self.redis_client.setex(cache_key, 300, "true" if has_access else "false") except Exception as cache_error: logger.warning(f"Cache set failed: {cache_error}") logger.debug(f"Tenant access check", user_id=user_id, tenant_id=tenant_id, has_access=has_access) return has_access except asyncio.TimeoutError: logger.error(f"Timeout verifying tenant access: user={user_id}, tenant={tenant_id}") # Fail open for availability (let service handle detailed check) return True except httpx.RequestError as e: logger.error(f"Request error verifying tenant access: {e}") # Fail open for availability return True except Exception as e: logger.error(f"Gateway tenant access verification failed: {e}") # Fail open for availability (let service handle detailed check) return True async def get_user_role_in_tenant(self, user_id: str, tenant_id: str) -> Optional[str]: """ Get user's role within a specific tenant Args: user_id: User ID tenant_id: Tenant ID Returns: Optional[str]: User's role in tenant (owner, admin, manager, user) or None """ try: async with httpx.AsyncClient(timeout=3.0) as client: response = await client.get( f"{settings.TENANT_SERVICE_URL}/api/v1/tenants/{tenant_id}/members/{user_id}" ) if response.status_code == 200: data = response.json() role = data.get("role") logger.debug(f"User role in tenant", user_id=user_id, tenant_id=tenant_id, role=role) return role elif response.status_code == 404: logger.debug(f"User not found in tenant", user_id=user_id, tenant_id=tenant_id) return None else: logger.warning(f"Unexpected response getting user role: {response.status_code}") return None except Exception as e: logger.error(f"Failed to get user role in tenant: {e}") return None async def verify_resource_permission( self, user_id: str, tenant_id: str, resource: str, action: str ) -> bool: """ Fine-grained resource permission check (used by services) Args: user_id: User ID tenant_id: Tenant ID resource: Resource type (sales, training, forecasts, etc.) action: Action being performed (read, write, delete, etc.) Returns: bool: True if user has permission """ user_role = await self.get_user_role_in_tenant(user_id, tenant_id) if not user_role: return False # Role-based permission matrix permissions = { "owner": ["*"], # Owners can do everything "admin": ["read", "write", "delete", "manage"], "manager": ["read", "write"], "user": ["read"] } allowed_actions = permissions.get(user_role, []) has_permission = "*" in allowed_actions or action in allowed_actions logger.debug(f"Resource permission check", user_id=user_id, tenant_id=tenant_id, resource=resource, action=action, user_role=user_role, has_permission=has_permission) return has_permission async def get_user_tenants(self, user_id: str) -> list: """ Get all tenants a user has access to Args: user_id: User ID Returns: list: List of tenant dictionaries """ try: async with httpx.AsyncClient(timeout=5.0) as client: response = await client.get( f"{settings.TENANT_SERVICE_URL}/api/v1/tenants/users/{user_id}" ) if response.status_code == 200: tenants = response.json() logger.debug(f"Retrieved user tenants", user_id=user_id, tenant_count=len(tenants)) return tenants else: logger.warning(f"Failed to get user tenants: {response.status_code}") return [] except Exception as e: logger.error(f"Failed to get user tenants: {e}") return [] # Global instance for easy import tenant_access_manager = TenantAccessManager() # ================================================================ # FASTAPI DEPENDENCIES # ================================================================ async def verify_tenant_access_dep( tenant_id: str, current_user: Dict[str, Any] = Depends(get_current_user_dep) ) -> str: """ FastAPI dependency to verify tenant access and return tenant_id Args: tenant_id: Tenant ID from path parameter current_user: Current user from auth dependency Returns: str: Validated tenant_id Raises: HTTPException: If user doesn't have access to tenant """ has_access = await tenant_access_manager.verify_basic_tenant_access(current_user["user_id"], tenant_id) if not has_access: logger.warning(f"Access denied to tenant", user_id=current_user["user_id"], tenant_id=tenant_id) raise HTTPException( status_code=403, detail=f"User {current_user['user_id']} does not have access to tenant {tenant_id}" ) logger.debug(f"Tenant access verified", user_id=current_user["user_id"], tenant_id=tenant_id) return tenant_id async def verify_tenant_permission_dep( tenant_id: str, resource: str, action: str, current_user: Dict[str, Any] = Depends(get_current_user_dep) ) -> str: """ FastAPI dependency to verify tenant access AND resource permission Args: tenant_id: Tenant ID from path parameter resource: Resource type being accessed action: Action being performed current_user: Current user from auth dependency Returns: str: Validated tenant_id Raises: HTTPException: If user doesn't have access or permission """ # First verify basic tenant access has_access = await tenant_access_manager.verify_basic_tenant_access(current_user["user_id"], tenant_id) if not has_access: raise HTTPException( status_code=403, detail=f"Access denied to tenant {tenant_id}" ) # Then verify specific resource permission has_permission = await tenant_access_manager.verify_resource_permission( current_user["user_id"], tenant_id, resource, action ) if not has_permission: raise HTTPException( status_code=403, detail=f"Insufficient permissions for {action} on {resource}" ) logger.debug(f"Tenant access and permission verified", user_id=current_user["user_id"], tenant_id=tenant_id, resource=resource, action=action) return tenant_id # ================================================================ # UTILITY FUNCTIONS # ================================================================ def extract_tenant_id_from_path(path: str) -> Optional[str]: """ More robust tenant ID extraction using regex pattern matching Only matches actual tenant-scoped paths with UUID format """ # Pattern for tenant-scoped paths: /api/v1/tenants/{uuid}/... tenant_pattern = r'/api/v1/tenants/([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})/.*' match = re.match(tenant_pattern, path, re.IGNORECASE) if match: return match.group(1) def is_tenant_scoped_path(path: str) -> bool: """ Check if path is tenant-scoped (contains /tenants/{tenant_id}/) Args: path: URL path Returns: bool: True if path is tenant-scoped """ return extract_tenant_id_from_path(path) is not None # ================================================================ # EXPORTS # ================================================================ __all__ = [ # Classes "TenantAccessManager", "tenant_access_manager", # Dependencies "verify_tenant_access_dep", "verify_tenant_permission_dep", # Utilities "extract_tenant_id_from_path", "is_tenant_scoped_path" ]