[{'Text': '"""Auth0 authentication helpers for role-based access control.\n\nThis module provides utilities to check user roles from Auth0 JWT tokens.\nWorks with both the official MCP SDK and FastMCP.\n\nExample:\n    from point_topic_mcp.auth import require_admin_role\n\n    @mcp.tool()\n    async def admin_only_tool(token_claims: dict | None = None):\n        require_admin_role(token_claims)  # Raises PermissionError if not admin\n        return {"message": "Admin access granted"}\n"""\n\nimport logging\nfrom typing import Any\n\nimport httpx\nfrom jose import jwt, JWTError\n\nlogger = logging.getLogger(__name__)\n\n# Thread-local storage for current token context\n# This allows token to be passed without changing function signatures everywhere\n_current_token_claims: dict[str, Any] = {}\n\n\ndef set_current_token_claims(claims: dict | None):\n    """Set the current token claims for this request context.\n\n    This is used internally by the server to make token claims\n    available to tools without changing their signatures.\n    """\n    global _current_token_claims\n    _current_token_claims = claims or {}\n\n\ndef get_current_token_claims() -> dict | None:\n    """Get the current token claims for this request context.\n\n    Returns:\n        Token claims dict if available, None otherwise\n    """\n    return _current_token_claims if _current_token_claims else None\n\n\ndef get_user_roles_from_claims(claims: dict | None = None) -> list[str]:\n    """Get roles from JWT token claims.\n\n    Extracts roles from Auth0 JWT token claims. Auth0 can store roles\n    in different claim locations depending on the token configuration.\n\n    Args:\n        claims: Token claims dict. If None, uses current context claims.\n\n    Returns:\n        List of role/permission strings from the token\n        Empty list if no token or no roles found\n    """\n    if claims is None:\n        claims = get_current_token_claims()\n\n    if not claims:\n        return []\n\n    # Auth0 stores roles in different places depending on configuration\n    # Try common locations in order of preference\n\n    # 1. Direct \'roles\' claim (configured in Auth0 Actions/rules)\n    roles = claims.get("roles", [])\n    if roles:\n        return roles if isinstance(roles, list) else [roles]\n\n    # 2. \'permissions\' claim (from Auth0 API Authorization)\n    permissions = claims.get("permissions", [])\n    if permissions:\n        return permissions if isinstance(permissions, list) else [permissions]\n\n    # 3. \'scope\' claim (OAuth2 scope, space-separated)\n    scope = claims.get("scope", "")\n    if scope:\n        return scope.split() if isinstance(scope, str) else [scope]\n\n    return []\n\n\n# Backwards compatibility alias\nget_user_roles = get_user_roles_from_claims\n\n\ndef has_role(role_name: str, claims: dict | None = None) -> bool:\n    """Check if user has a specific role.\n\n    Args:\n        role_name: The role to check for\n        claims: Token claims dict. If None, uses current context claims.\n\n    Returns:\n        True if user has the role, False otherwise\n    """\n    roles = get_user_roles_from_claims(claims)\n    return role_name in roles\n\n\ndef has_any_role(role_names: list[str], claims: dict | None = None) -> bool:\n    """Check if user has any of the specified roles.\n\n    Args:\n        role_names: List of roles to check for\n        claims: Token claims dict. If None, uses current context claims.\n\n    Returns:\n        True if user has at least one of the roles\n    """\n    roles = get_user_roles_from_claims(claims)\n    return any(role in roles for role in role_names)\n\n\ndef require_admin_role(claims: dict | None = None):\n    """Require user to have Point Topic Admin role.\n\n    Checks for admin roles. Raises PermissionError if user\n    doesn\'t have appropriate admin access.\n\n    Args:\n        claims: Token claims dict. If None, uses current context claims.\n\n    Raises:\n        PermissionError: If user doesn\'t have admin role\n    """\n    roles = get_user_roles_from_claims(claims)\n\n    # Admin role names to check for\n    admin_roles = [\n        "Point Topic Admin",\n        "ptAdmin",\n        "admin",\n        "user:upc-query-agent",  # From Auth0 permissions\n        "admin:upc-query-agent",  # From Auth0 permissions\n    ]\n\n    if not any(role in roles for role in admin_roles):\n        raise PermissionError(\n            f"This tool requires admin role. "\n            f"Your roles: {roles if roles else \'None\'}. "\n            f"Required: one of {admin_roles}"\n        )\n\n\ndef require_role(role_name: str, claims: dict | None = None):\n    """Require user to have a specific role.\n\n    Generic role checker. Raises PermissionError if user\n    doesn\'t have the specified role.\n\n    Args:\n        role_name: The required role name\n        claims: Token claims dict. If None, uses current context claims.\n\n    Raises:\n        PermissionError: If user doesn\'t have the role\n    """\n    roles = get_user_roles_from_claims(claims)\n\n    if role_name not in roles:\n        raise PermissionError(\n            f"This tool requires \'{role_name}\' role. "\n            f"Your roles: {roles if roles else \'None\'}"\n        )\n\n\ndef get_user_info(claims: dict | None = None) -> dict:\n    """Get user information from the JWT token.\n\n    Args:\n        claims: Token claims dict. If None, uses current context claims.\n\n    Returns:\n        Dictionary with user claims (sub, email, name, etc.)\n        Empty dict if no token available\n    """\n    if claims is None:\n        claims = get_current_token_claims()\n\n    if not claims:\n        return {}\n\n    return {\n        "user_id": claims.get("sub"),\n        "email": claims.get("email"),\n        "name": claims.get("name"),\n        "roles": get_user_roles_from_claims(claims),\n        "permissions": claims.get("permissions", []),\n    }\n\n\nclass Auth0TokenVerifier:\n    """JWT token verifier using Auth0 JWKS endpoint.\n\n    This class validates JWT tokens issued by Auth0 using the\n    JSON Web Key Set (JWKS) endpoint for signature verification.\n    """\n\n    def __init__(\n        self,\n        auth0_domain: str,\n        audience: str,\n        jwks_url: str | None = None,\n    ):\n        """Initialize the token verifier.\n\n        Args:\n            auth0_domain: Auth0 tenant domain (e.g., \'point-topic.eu.auth0.com\')\n            audience: Expected audience/resource URL (e.g., \'https://point-topic-mcp.fastmcp.app/\')\n            jwks_url: Optional JWKS URL. Defaults to https://{domain}/.well-known/jwks.json\n        """\n        self.auth0_domain = auth0_domain\n        self.audience = audience\n        self.jwks_url = jwks_url or f"https://{auth0_domain}/.well-known/jwks.json"\n        self._jwks: dict | None = None\n\n    async def _get_jwks(self) -> dict | None:\n        """Fetch JWKS from Auth0 if not already cached."""\n        if self._jwks is None:\n            async with httpx.AsyncClient() as client:\n                response = await client.get(self.jwks_url)\n                response.raise_for_status()\n                self._jwks = response.json()\n        return self._jwks\n\n    async def verify_token(self, token: str) -> dict | None:\n        """Verify and decode a JWT token.\n\n        Args:\n            token: The JWT token string to verify\n\n        Returns:\n            Decoded token claims dict if valid, None otherwise\n        """\n        try:\n            jwks = await self._get_jwks()\n            if jwks is None:\n                logger.warning("Failed to fetch JWKS")\n                return None\n            signing_key = self._get_signing_key(token, jwks)\n\n            if not signing_key:\n                logger.warning("No signing key found for token")\n                return None\n\n            # Decode and verify the token\n            payload = jwt.decode(\n                token,\n                signing_key,\n                audience=self.audience,\n                issuer=f"https://{self.auth0_domain}/",\n            )\n\n            return payload\n\n        except jwt.ExpiredSignatureError:  # type: ignore\n            logger.warning("Token has expired")\n        except jwt.JWTClaimsError as e:  # type: ignore\n            logger.warning(f"Token claims error: {e}")\n        except JWTError as e:\n            logger.warning(f"JWT validation error: {e}")\n        except Exception as e:\n            logger.error(f"Unexpected error verifying token: {e}")\n\n        return None\n\n    def _get_signing_key(self, token: str, jwks: dict) -> dict | None:\n        """Extract the signing key from JWKS matching the token\'s kid."""\n        try:\n            headers = jwt.get_unverified_header(token)\n            kid = headers.get("kid")\n            if not kid:\n                return None\n\n            for key in jwks.get("keys", []):\n                if key.get("kid") == kid:\n                    return key\n        except JWTError as e:\n            logger.warning(f"Failed to extract token header: {e}")\n\n        return None\n\n    async def get_token_claims(self, token: str) -> dict | None:\n        """Get claims from a token (wrapper around verify_token).\n\n        Args:\n            token: The JWT token string\n\n        Returns:\n            Token claims dict if valid, None otherwise\n        """\n        return await self.verify_token(token)\n\n\n__all__ = [\n    # Token context management\n    "set_current_token_claims",\n    "get_current_token_claims",\n    # Role checking functions\n    "get_user_roles",\n    "get_user_roles_from_claims",\n    "has_role",\n    "has_any_role",\n    "require_admin_role",\n    "require_role",\n    "get_user_info",\n    # Token verification\n    "Auth0TokenVerifier",\n]\n'}]