from fastapi import Depends, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from typing import Optional, Dict, Any import httpx import sys import os # Add shared modules to path sys.path.append('/app') from shared.config import settings from shared.redis_client import redis_client from app.models.user_role import UserRole from app.models.role import Role from bson import ObjectId import json security = HTTPBearer(auto_error=False) class CurrentUser: def __init__(self, user_id: str, email: str, username: str, token: str): self.id = user_id self.email = email self.username = username self.token = token async def verify_auth_token(token: str) -> Optional[Dict[str, Any]]: """Verify token with auth service""" try: # First check Redis cache cache_key = f"token_validation:{token}" cached_result = await redis_client.get_json(cache_key) if cached_result and cached_result.get("valid"): return cached_result # Call auth service for validation auth_service_url = os.getenv("AUTH_SERVICE_URL", "http://auth-service:8000") async with httpx.AsyncClient() as client: response = await client.post( f"{auth_service_url}/api/v1/tokens/validate", json={"token": token}, timeout=5.0 ) if response.status_code == 200: result = response.json() # Cache valid tokens for 5 minutes if result.get("valid"): await redis_client.set(cache_key, result, expire=300) return result return None except Exception as e: print(f"Token verification error: {e}") return None async def get_current_user( credentials: Optional[HTTPAuthorizationCredentials] = Depends(security) ) -> Optional[CurrentUser]: """Get current user from token""" if not credentials: return None token = credentials.credentials token_data = await verify_auth_token(token) if not token_data or not token_data.get("valid"): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token" ) return CurrentUser( user_id=token_data.get("user_id"), email=token_data.get("email"), username=token_data.get("username"), token=token ) async def get_current_active_user( current_user: Optional[CurrentUser] = Depends(get_current_user) ) -> CurrentUser: """Get current active user (required)""" if not current_user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required" ) return current_user async def get_optional_current_user( credentials: Optional[HTTPAuthorizationCredentials] = Depends(security) ) -> Optional[CurrentUser]: """Get current user without raising error if not authenticated""" try: return await get_current_user(credentials) except HTTPException: return None async def get_user_permissions(user_id: str) -> list[str]: """Get user permissions from roles""" try: # Check Redis cache first cache_key = f"user_permissions:{user_id}" cached_permissions = await redis_client.get(cache_key) if cached_permissions: try: return json.loads(cached_permissions) except json.JSONDecodeError: pass # Get user roles user_roles = await UserRole.get_user_roles(ObjectId(user_id), active_only=True) all_permissions = set() for user_role in user_roles: if user_role.is_valid(): role = await Role.find_one(Role.name == user_role.role_name) if role and role.is_active: role_permissions = await role.get_all_permissions() all_permissions.update(role_permissions) permissions_list = list(all_permissions) # Cache for 10 minutes await redis_client.set( cache_key, json.dumps(permissions_list), expire=600 ) return permissions_list except Exception as e: print(f"Error getting user permissions: {e}") return [] async def get_user_roles(user_id: str) -> list[str]: """Get user role names""" try: # Check cache first cache_key = f"user_roles:{user_id}" cached_roles = await redis_client.get(cache_key) if cached_roles: try: return json.loads(cached_roles) except json.JSONDecodeError: pass user_roles = await UserRole.get_user_roles(ObjectId(user_id), active_only=True) role_names = [ur.role_name for ur in user_roles if ur.is_valid()] # Cache for 10 minutes await redis_client.set( cache_key, json.dumps(role_names), expire=600 ) return role_names except Exception: return [] async def check_permission(user_id: str, resource: str, action: str, scope: str = None) -> bool: """Check if user has specific permission""" permissions = await get_user_permissions(user_id) # Check exact permission match permission_name = f"{resource}.{action}" if scope: permission_name += f".{scope}" if permission_name in permissions: return True # Check broader permissions broader_permission = f"{resource}.admin" if broader_permission in permissions: return True # Check system admin if "system.admin" in permissions: return True return False def require_permissions(*required_permissions: str): """Dependency to require specific permissions""" async def permission_checker(current_user: CurrentUser = Depends(get_current_active_user)): user_permissions = await get_user_permissions(current_user.id) for permission in required_permissions: if not await check_permission(current_user.id, *permission.split('.')): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"Permission '{permission}' required" ) return current_user return permission_checker def require_roles(*required_roles: str): """Dependency to require specific roles""" async def role_checker(current_user: CurrentUser = Depends(get_current_active_user)): user_roles = await get_user_roles(current_user.id) has_required_role = any(role in user_roles for role in required_roles) if not has_required_role: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"One of these roles required: {', '.join(required_roles)}" ) return current_user return role_checker async def require_own_resource_or_permission( resource_user_id: str, permission: str, current_user: CurrentUser = Depends(get_current_active_user) ): """Check if user owns resource or has permission""" # User can access their own resources if resource_user_id == current_user.id: return current_user # Check if user has the required permission if not await check_permission(current_user.id, *permission.split('.')): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied" ) return current_user async def clear_user_cache(user_id: str): """Clear cached user data""" await redis_client.delete(f"user_permissions:{user_id}") await redis_client.delete(f"user_roles:{user_id}")