[Core] pass auth token in dashboard head python client sdk

python/ray/_private/authentication/authentication_constants.py

@@ -0,0 +1,23 @@
+# Token setup instructions (used in multiple contexts)
+TOKEN_SETUP_INSTRUCTIONS = """Please provide an authentication token using one of these methods:
+  1. Set the RAY_AUTH_TOKEN environment variable
+  2. Set the RAY_AUTH_TOKEN_PATH environment variable (pointing to a token file)
+  3. Create a token file at the default location: ~/.ray/auth_token"""
+
+# When token auth is enabled but no token is found anywhere
+TOKEN_AUTH_ENABLED_BUT_NO_TOKEN_FOUND_ERROR_MESSAGE = (
+    "Token authentication is enabled but no authentication token was found. "
+    + TOKEN_SETUP_INSTRUCTIONS
+)
+
+# When HTTP request fails with 401 (Unauthorized - missing token)
+HTTP_REQUEST_MISSING_TOKEN_ERROR_MESSAGE = (
+    "The Ray cluster requires authentication, but no token was provided.\n\n"
+    + TOKEN_SETUP_INSTRUCTIONS
+)
+
+# When HTTP request fails with 403 (Forbidden - invalid token)
+HTTP_REQUEST_INVALID_TOKEN_ERROR_MESSAGE = (
+    "The authentication token you provided is invalid or incorrect.\n\n"
+    + TOKEN_SETUP_INSTRUCTIONS
+)

python/ray/_private/authentication/authentication_token_generator.py

@@ -0,0 +1,6 @@
+import uuid
+
+
+# TODO: this is a placeholder for the actual authentication token generator. Will be replaced with a proper implementation.
+def generate_new_authentication_token() -> str:
+    return uuid.uuid4().hex

python/ray/_private/authentication/authentication_token_setup.py

@@ -0,0 +1,100 @@
+"""Authentication token setup for Ray.
+
+This module provides functions to generate and save authentication tokens
+for Ray's token-based authentication system. Token loading and caching is
+handled by the C++ AuthenticationTokenLoader.
+"""
+
+import logging
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+from ray._private.authentication.authentication_constants import (
+    TOKEN_AUTH_ENABLED_BUT_NO_TOKEN_FOUND_ERROR_MESSAGE,
+)
+from ray._private.authentication.authentication_token_generator import (
+    generate_new_authentication_token,
+)
+from ray._raylet import (
+    AuthenticationMode,
+    AuthenticationTokenLoader,
+    get_authentication_mode,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def generate_and_save_token() -> None:
+    """Generate a new random token and save it in the default token path.
+
+    Returns:
+        The newly generated authentication token.
+    """
+    # Generate a UUID-based token
+    token = generate_new_authentication_token()
+
+    token_path = _get_default_token_path()
+    try:
+        # Create directory if it doesn't exist
+        token_path.parent.mkdir(parents=True, exist_ok=True)
+
+        # Write token to file with explicit flush and fsync
+        with open(token_path, "w") as f:
+            f.write(token)
+
+        logger.info(f"Generated new authentication token and saved to {token_path}")
+    except Exception:
+        raise
+
+
+def _get_default_token_path() -> Path:
+    """Get the default token file path (~/.ray/auth_token).
+
+    Returns:
+        Path object pointing to ~/.ray/auth_token
+    """
+    return Path.home() / ".ray" / "auth_token"
+
+
+def ensure_token_if_auth_enabled(
+    system_config: Optional[Dict[str, Any]] = None, create_token_if_missing: bool = True
+) -> None:
+    """Check authentication settings and set up token resources if authentication is enabled.
+
+    Ray calls this early during ray.init() to do the following for token-based authentication:
+    1. Check whether you enabled token-based authentication.
+    2. Make sure a token is available if authentication is enabled.
+    3. Generate and save a default token for new local clusters if one doesn't already exist.
+
+    Args:
+        system_config: Ray raises an error if you set auth_mode in system_config instead of the environment.
+        create_token_if_missing: Generate a new token if one doesn't already exist.
+
+    Raises:
+        RuntimeError: Ray raises this error if authentication is enabled but no token is found when connecting
+            to an existing cluster.
+    """
+
+    # Check if you enabled token authentication.
+    if get_authentication_mode() != AuthenticationMode.TOKEN:
+        if (
+            system_config
+            and "auth_mode" in system_config
+            and system_config["auth_mode"] != "disabled"
+        ):
+            raise RuntimeError(
+                "Set authentication mode can only be set with the `RAY_auth_mode` environment variable, not using the system_config."
+            )
+        return
+
+    token_loader = AuthenticationTokenLoader.instance()
+
+    if not token_loader.has_token():
+        if create_token_if_missing:
+            # Generate a new token.
+            generate_and_save_token()
+
+            # Reload the cache so subsequent calls to token_loader read the new token.
+            token_loader.reset_cache()
+        else:
+            raise RuntimeError(TOKEN_AUTH_ENABLED_BUT_NO_TOKEN_FOUND_ERROR_MESSAGE)

python/ray/_private/worker.py

@@ -62,6 +62,9 @@
 from ray._common import ray_option_utils
 from ray._common.constants import RAY_WARN_BLOCKING_GET_INSIDE_ASYNC_ENV_VAR
 from ray._common.utils import load_class
+from ray._private.authentication.authentication_token_setup import (
+    ensure_token_if_auth_enabled,
+)
 from ray._private.client_mode_hook import client_mode_hook
 from ray._private.custom_types import TensorTransportEnum
 from ray._private.function_manager import FunctionActorManager
@@ -1865,6 +1868,9 @@ def sigterm_handler(signum, frame):
     if bootstrap_address is None:
         # In this case, we need to start a new cluster.
 
+        # Setup and verify authentication for new cluster
+        ensure_token_if_auth_enabled(_system_config, create_token_if_missing=True)
+
         # Don't collect usage stats in ray.init() unless it's a nightly wheel.
         from ray._common.usage import usage_lib
 
@@ -1952,6 +1958,9 @@ def sigterm_handler(signum, frame):
                 "an existing cluster."
             )
 
+        # Setup and verify authentication for connecting to existing cluster
+        ensure_token_if_auth_enabled(_system_config, create_token_if_missing=False)
+
         # In this case, we only need to connect the node.
         ray_params = ray._private.parameter.RayParams(
             node_ip_address=_node_ip_address,

python/ray/_raylet.pyx

@@ -201,6 +201,7 @@ include "includes/metric.pxi"
 include "includes/setproctitle.pxi"
 include "includes/raylet_client.pxi"
 include "includes/gcs_subscriber.pxi"
+include "includes/rpc_token_authentication.pxi"
 
 import ray
 from ray.exceptions import (

python/ray/dashboard/authentication_utils.py

@@ -0,0 +1,31 @@
+from ray._raylet import (
+    AuthenticationMode,
+    get_authentication_mode,
+    validate_authentication_token,
+)
+
+
+def is_token_auth_enabled() -> bool:
+    """Check if token authentication is enabled.
+
+    Returns:
+        bool: True if auth_mode is set to "token", False otherwise
+    """
+    return get_authentication_mode() == AuthenticationMode.TOKEN
+
+
+def validate_request_token(auth_header: str) -> bool:
+    """Validate the Authorization header from an HTTP request.
+
+    Args:
+        auth_header: The Authorization header value (e.g., "Bearer <token>")
+
+    Returns:
+        bool: True if token is valid, False otherwise
+    """
+    if not auth_header:
+        return False
+
+    # validate_authentication_token expects full "Bearer <token>" format
+    # and performs equality comparison via C++ layer
+    return validate_authentication_token(auth_header)

python/ray/dashboard/http_server_head.py

@@ -20,6 +20,7 @@
 from ray._common.network_utils import build_address, parse_address
 from ray._common.usage.usage_lib import TagKey, record_extra_usage_tag
 from ray._common.utils import get_or_create_event_loop
+from ray.dashboard import authentication_utils as auth_utils
 from ray.dashboard.dashboard_metrics import DashboardPrometheusMetrics
 from ray.dashboard.head import DashboardHeadModule
 
@@ -163,6 +164,30 @@ def get_address(self):
         assert self.http_host and self.http_port
         return self.http_host, self.http_port
 
+    @aiohttp.web.middleware
+    async def auth_middleware(self, request, handler):
+        """Authenticate requests when token auth is enabled."""
+
+        # Skip if auth not enabled
+        if not auth_utils.is_token_auth_enabled():
+            return await handler(request)
+
+        # Extract and validate token
+        auth_header = request.headers.get("Authorization", "")
+
+        if not auth_header:
+            return aiohttp.web.Response(
+                status=401, text="Unauthorized: Missing authentication token"
+            )
+
+        # Validate token
+        if not auth_utils.validate_request_token(auth_header):
+            return aiohttp.web.Response(
+                status=403, text="Forbidden: Invalid authentication token"
+            )
+
+        return await handler(request)
+
     @aiohttp.web.middleware
     async def path_clean_middleware(self, request, handler):
         if request.path.startswith("/static") or request.path.startswith("/logs"):
@@ -252,6 +277,7 @@ async def run(
             client_max_size=ray_constants.DASHBOARD_CLIENT_MAX_SIZE,
             middlewares=[
                 self.metrics_middleware,
+                self.auth_middleware,
                 self.path_clean_middleware,
                 self.browsers_no_post_put_middleware,
                 self.cache_control_static_middleware,

python/ray/dashboard/modules/dashboard_sdk.py

@@ -12,6 +12,7 @@
 import yaml
 
 import ray
+from ray._private.authentication import authentication_constants
 from ray._private.runtime_env.packaging import (
     create_package,
     get_uri_for_directory,
@@ -20,7 +21,9 @@
 from ray._private.runtime_env.py_modules import upload_py_modules_if_needed
 from ray._private.runtime_env.working_dir import upload_working_dir_if_needed
 from ray._private.utils import split_address
+from ray._raylet import AuthenticationTokenLoader
 from ray.autoscaler._private.cli_logger import cli_logger
+from ray.dashboard.authentication_utils import is_token_auth_enabled
 from ray.dashboard.modules.job.common import uri_to_http_components
 from ray.util.annotations import DeveloperAPI, PublicAPI
 
@@ -222,7 +225,11 @@ def __init__(
         self._default_metadata = cluster_info.metadata or {}
         # Headers used for all requests sent to job server, optional and only
         # needed for cases like authentication to remote cluster.
-        self._headers = cluster_info.headers
+        self._headers = cluster_info.headers or {}
+
+        # Add authentication token if token auth is enabled
+        self._set_auth_header_if_enabled()
+
         # Set SSL verify parameter for the requests library and create an ssl_context
         # object when needed for the aiohttp library.
         self._verify = verify
@@ -242,6 +249,22 @@ def __init__(
             else:
                 self._ssl_context = None
 
+    def _set_auth_header_if_enabled(self):
+        """Add authentication token to headers if token auth is enabled."""
+        if is_token_auth_enabled():
+            token_loader = AuthenticationTokenLoader.instance()
+            token_added = token_loader.set_token_for_http_header(self._headers)
+
+            if not token_added:
+                # Token auth is enabled but no token found or Authorization already set
+                if "Authorization" not in self._headers:
+                    # No token found - log warning but don't fail yet
+                    # Let the server return 401 for a better error message
+                    logger.warning(
+                        "Token authentication is enabled but no token was found. "
+                        "Requests to authenticated clusters will fail."
+                    )
+
     def _check_connection_and_version(
         self, min_version: str = "1.9", version_error_message: str = None
     ):
@@ -293,14 +316,15 @@ def _do_request(
         json_data: Optional[dict] = None,
         **kwargs,
     ) -> "requests.Response":
-        """Perform the actual HTTP request
+        """Perform the actual HTTP request with authentication error handling.
 
         Keyword arguments other than "cookies", "headers" are forwarded to the
         `requests.request()`.
         """
         url = self._address + endpoint
         logger.debug(f"Sending request to {url} with json data: {json_data or {}}.")
-        return requests.request(
+
+        response = requests.request(
             method,
             url,
             cookies=self._cookies,
@@ -311,6 +335,22 @@ def _do_request(
             **kwargs,
         )
 
+        # Check for authentication errors and provide helpful messages
+        if response.status_code == 401:
+            # Unauthorized - missing or no token provided
+            raise RuntimeError(
+                f"Authentication required: {response.text}\n\n"
+                + authentication_constants.HTTP_REQUEST_MISSING_TOKEN_ERROR_MESSAGE
+            )
+        elif response.status_code == 403:
+            # Forbidden - invalid token
+            raise RuntimeError(
+                f"Authentication failed: {response.text}\n\n"
+                + authentication_constants.HTTP_REQUEST_INVALID_TOKEN_ERROR_MESSAGE
+            )
+
+        return response
+
     def _package_exists(
         self,
         package_uri: str,

python/ray/includes/rpc_token_authentication.pxd

@@ -0,0 +1,28 @@
+from libcpp cimport bool as c_bool
+from libcpp.string cimport string
+from ray.includes.optional cimport optional
+
+cdef extern from "ray/rpc/authentication/authentication_mode.h" namespace "ray::rpc" nogil:
+    cdef enum CAuthenticationMode "ray::rpc::AuthenticationMode":
+        DISABLED "ray::rpc::AuthenticationMode::DISABLED"
+        TOKEN "ray::rpc::AuthenticationMode::TOKEN"
+
+    CAuthenticationMode GetAuthenticationMode()
+
+cdef extern from "ray/rpc/authentication/authentication_token.h" namespace "ray::rpc" nogil:
+    cdef cppclass CAuthenticationToken "ray::rpc::AuthenticationToken":
+        CAuthenticationToken()
+        CAuthenticationToken(string value)
+        c_bool empty()
+        c_bool Equals(const CAuthenticationToken& other)
+        string ToAuthorizationHeaderValue()
+        @staticmethod
+        CAuthenticationToken FromMetadata(string metadata_value)
+
+cdef extern from "ray/rpc/authentication/authentication_token_loader.h" namespace "ray::rpc" nogil:
+    cdef cppclass CAuthenticationTokenLoader "ray::rpc::AuthenticationTokenLoader":
+        @staticmethod
+        CAuthenticationTokenLoader& instance()
+        c_bool HasToken()
+        void ResetCache()
+        optional[CAuthenticationToken] GetToken()