[{'Text': '"""Permission-based authentication middleware for FastMCP.\n\nThis middleware enforces role/permission-based access control using Auth0 JWT tokens.\n\nAccess Levels:\n1. Unauthenticated → Only @public tools\n2. Authenticated (not ptAdmin) → @public + tools matching their permissions\n3. ptAdmin → All tools (full access)\n\nUsage:\n    The middleware is automatically applied when using register_tools() from\n    the auth-enabled tool registration module.\n"""\n\nimport os\n\nfrom fastmcp.server.middleware import Middleware, MiddlewareContext\nfrom fastmcp.server.dependencies import get_access_token\nfrom point_topic_mcp.auth.decorators import (\n    is_public_tool,\n    get_tool_permissions,\n    has_tool_permissions,\n)\n\n\ndef get_user_permissions() -> list[str]:\n    """Extract user permissions from Auth0 JWT token.\n\n    Returns:\n        List of permission/role strings from the JWT token claims\n        Empty list if no token or unauthenticated\n    """\n    try:\n        token = get_access_token()\n        if token and token.claims:\n            permissions = []\n\n            # Auth0 stores permissions in different claim locations\n            # Check all possible sources\n\n            # 1. \'roles\' claim (from Auth0 Actions/Rules)\n            roles = token.claims.get("roles", [])\n            if roles:\n                if isinstance(roles, list):\n                    permissions.extend(roles)\n                else:\n                    permissions.append(roles)\n\n            # 2. \'permissions\' claim (from Auth0 API Authorization)\n            perms = token.claims.get("permissions", [])\n            if perms:\n                if isinstance(perms, list):\n                    permissions.extend(perms)\n                else:\n                    permissions.append(perms)\n\n            # 3. \'scope\' claim (OAuth2 scopes, space-separated)\n            scope = token.claims.get("scope", "")\n            if scope and isinstance(scope, str):\n                permissions.extend(scope.split())\n            elif scope:\n                permissions.append(str(scope))\n\n            # Remove duplicates while preserving order\n            seen = set()\n            unique_perms = []\n            for p in permissions:\n                if p not in seen:\n                    seen.add(p)\n                    unique_perms.append(p)\n\n            return unique_perms\n    except Exception:\n        pass\n\n    # Stdio transport: always internal/backend, no JWT possible — grant full access.\n    if not os.getenv("MCP_TRANSPORT") or os.getenv("MCP_TRANSPORT") == "stdio":\n        return ["ptAdmin"]\n\n    return []\n\n\ndef is_pt_admin(user_permissions: list[str]) -> bool:\n    """Check if user has ptAdmin role.\n\n    Args:\n        user_permissions: List of user permissions/roles\n\n    Returns:\n        True if user has ptAdmin or superuser role\n    """\n    admin_roles = [\n        "Point Topic Admin",\n        "ptAdmin",\n        "admin",\n        "admin:upc-query-agent",\n    ]\n    return any(role in user_permissions for role in admin_roles)\n\n\ndef check_tool_access(tool_name: str, user_permissions: list[str]) -> tuple[bool, str]:\n    """Check if user can access a specific tool.\n\n    Args:\n        tool_name: Name of the tool to check\n        user_permissions: List of user\'s permissions/roles\n\n    Returns:\n        Tuple of (allowed: bool, reason: str)\n    """\n    # Public tools are always allowed\n    if is_public_tool(tool_name):\n        return True, "Public tool - no authentication required"\n\n    # ptAdmin has access to everything\n    if is_pt_admin(user_permissions):\n        return True, "ptAdmin access granted"\n\n    # Not authenticated and not public\n    if not user_permissions:\n        return False, "Authentication required. Please login via Auth0 OAuth."\n\n    # Get required permissions for this tool\n    required_perms = get_tool_permissions(tool_name)\n\n    # Tool with no permission requirements = accessible to all authenticated\n    if not required_perms:\n        return True, "Authenticated access granted"\n\n    # Check if user has ALL required permissions\n    missing = [p for p in required_perms if p not in user_permissions]\n    if missing:\n        return False, (\n            f"Missing permissions: {missing}. "\n            f"Required: {required_perms}. "\n            f"Your permissions: {user_permissions}"\n        )\n\n    return True, "Permission check passed"\n\n\nclass PermissionMiddleware(Middleware):\n    """Middleware that enforces permission-based access control.\n\n    This middleware filters tools based on user permissions from Auth0 JWT tokens.\n    It implements three-tier access:\n        - Unauthenticated: Only @public tools visible\n        - Authenticated: @public + matching @requires_permissions\n        - ptAdmin: All tools visible\n\n    Uses on_list_tools to filter visible tools per-user.\n    Uses on_call_tool as defense-in-depth (should rarely trigger if filtering works).\n    """\n\n    async def on_list_tools(self, context: MiddlewareContext, call_next):\n        """Filter tool list based on user\'s permissions before returning to client.\n\n        This is the PRIMARY permission check - tools not in this list won\'t be\n        visible to the user in their IDE/client.\n\n        Args:\n            context: Middleware context\n            call_next: Function to get the full tool list\n\n        Returns:\n            Filtered list of tools the user has permission to see\n        """\n        # Get all registered tools\n        all_tools = await call_next(context)\n\n        # Get current user\'s permissions\n        user_permissions = get_user_permissions()\n        is_admin = is_pt_admin(user_permissions)\n        is_authenticated = len(user_permissions) > 0\n\n        # Filter tools based on permissions\n        visible_tools = []\n        for tool in all_tools:\n            tool_name = tool.name\n\n            # Check if tool should be visible\n            if is_public_tool(tool_name):\n                # Public tools visible to everyone\n                visible_tools.append(tool)\n            elif is_admin:\n                # ptAdmin sees everything\n                visible_tools.append(tool)\n            elif is_authenticated:\n                # Authenticated user - check permissions\n                required = get_tool_permissions(tool_name)\n                if not required:\n                    # No permission requirements = visible to all authenticated\n                    visible_tools.append(tool)\n                elif has_tool_permissions(tool_name, user_permissions):\n                    # User has required permissions\n                    visible_tools.append(tool)\n                # else: user doesn\'t have permission, don\'t add to list\n            # else: unauthenticated, only public tools shown (already handled above)\n\n        return visible_tools\n\n    async def on_call_tool(self, context: MiddlewareContext, call_next):\n        """Defense-in-depth: verify permissions before executing tool.\n\n        This should rarely trigger because on_list_tools already filtered\n        the visible tools. But we check again in case of:\n        - Direct API calls bypassing the tool list\n        - Race conditions\n        - Cached tool references\n\n        Args:\n            context: Middleware context with tool name\n            call_next: Function to call the actual tool\n\n        Returns:\n            Tool result if allowed\n\n        Raises:\n            PermissionError: If user doesn\'t have required permissions\n        """\n        tool_name = context.message.name\n        user_permissions = get_user_permissions()\n\n        allowed, reason = check_tool_access(tool_name, user_permissions)\n\n        if not allowed:\n            raise PermissionError(reason)\n\n        # Access granted - call the tool\n        return await call_next(context)\n\n\n__all__ = [\n    "PermissionMiddleware",\n    "get_user_permissions",\n    "is_pt_admin",\n    "check_tool_access",\n]\n'}]