backend/app/auth/rbac.py

"""
Role-Based Access Control (RBAC) — Permission enforcement for API endpoints.
Three roles: admin, analyst, viewer with hierarchical permissions.
"""

from enum import Enum
from fastapi import HTTPException, status
import structlog

logger = structlog.get_logger()


class Role(str, Enum):
    """User roles with increasing privilege levels."""
    VIEWER = "viewer"       # Can view data only
    ANALYST = "analyst"     # Can execute queries and view analytics
    ADMIN = "admin"         # Full access including user management


class Permission(str, Enum):
    """Granular permissions mapped to roles."""
    READ = "read"                   # View query results
    EXECUTE = "execute"             # Execute SQL queries
    VIEW_ANALYTICS = "view_analytics"  # View usage analytics
    MANAGE_SCHEMA = "manage_schema"    # Refresh schema index
    MANAGE_USERS = "manage_users"      # Create/delete users
    MANAGE_API_KEYS = "manage_api_keys"  # Create API keys
    VIEW_LOGS = "view_logs"            # View system logs


# Role → Permission mapping
ROLE_PERMISSIONS: dict[Role, set[Permission]] = {
    Role.VIEWER: {
        Permission.READ,
    },
    Role.ANALYST: {
        Permission.READ,
        Permission.EXECUTE,
        Permission.VIEW_ANALYTICS,
    },
    Role.ADMIN: {
        Permission.READ,
        Permission.EXECUTE,
        Permission.VIEW_ANALYTICS,
        Permission.MANAGE_SCHEMA,
        Permission.MANAGE_USERS,
        Permission.MANAGE_API_KEYS,
        Permission.VIEW_LOGS,
    },
}


def check_permission(user_role: str, required_permission: Permission) -> bool:
    """Check if a role has a specific permission."""
    try:
        role = Role(user_role)
    except ValueError:
        return False
    return required_permission in ROLE_PERMISSIONS.get(role, set())


def require_role(minimum_role: Role):
    """
    FastAPI dependency that enforces minimum role level.
    Usage: Depends(require_role(Role.ANALYST))
    """
    role_hierarchy = {Role.VIEWER: 0, Role.ANALYST: 1, Role.ADMIN: 2}

    def checker(current_user: dict):
        user_role = current_user.get("role", "viewer")
        try:
            user_level = role_hierarchy.get(Role(user_role), -1)
            required_level = role_hierarchy.get(minimum_role, 99)
        except ValueError:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Invalid role: {user_role}",
            )

        if user_level < required_level:
            logger.warning(
                "access_denied",
                user_role=user_role,
                required_role=minimum_role.value,
                user=current_user.get("sub"),
            )
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Insufficient permissions. Required: {minimum_role.value}, Current: {user_role}",
            )
        return current_user

    return checker