[core] Configure an interceptor to pass auth token in python direct g… (#58395)

## Description

there are places in the python code where we use the raw grpc library to
make grpc calls (eg: pub-sub, some calls to gcs etc). In the long term
we want to fully deprecate grpc library usage in our python code base
but as that can take more effort and testing, in this pr I am
introducing an interceptor to add auth headers (this will take effect
for all grpc calls made from python).

## Testing

### case 1: submitting a job using CLI
```
export RAY_auth_mode="token"
export RAY_AUTH_TOKEN="abcdef1234567890abcdef123456789"
ray start --head
ray job submit -- echo "hi"
```

output
``` 
ray job submit -- echo "hi"
2025-11-04 06:28:09,122 - INFO - NumExpr defaulting to 4 threads.
Job submission server address: http://127.0.0.1:8265

-------------------------------------------------------
Job 'raysubmit_1EV8q86uKM24nHmH' submitted successfully
-------------------------------------------------------

Next steps
  Query the logs of the job:
    ray job logs raysubmit_1EV8q86uKM24nHmH
  Query the status of the job:
    ray job status raysubmit_1EV8q86uKM24nHmH
  Request the job to be stopped:
    ray job stop raysubmit_1EV8q86uKM24nHmH

Tailing logs until the job exits (disable with --no-wait):
2025-11-04 06:28:10,363 INFO job_manager.py:568 -- Runtime env is setting up.
hi
Running entrypoint for job raysubmit_1EV8q86uKM24nHmH: echo hi

------------------------------------------
Job 'raysubmit_1EV8q86uKM24nHmH' succeeded
------------------------------------------
```


### case 2: submitting a job with actors and tasks and verifying on
dashboard
test.py
```python
import time
import ray
from ray._raylet import Config

ray.init()

@ray.remote
def print_hi():
    print("Hi")
    time.sleep(2)

@ray.remote
class SimpleActor:
    def __init__(self):
        self.value = 0

    def increment(self):
        self.value += 1
        return self.value

actor = SimpleActor.remote()
result = ray.get(actor.increment.remote())

for i in range(100):
    ray.get(print_hi.remote())
    time.sleep(20)

ray.shutdown()
```

``` 
export RAY_auth_mode="token"
export RAY_AUTH_TOKEN="abcdef1234567890abcdef123456789"
python test.py
```
### dashboard screenshots:

#### promts user to input the token
<img width="1720" height="1073" alt="image"
src="https://github.com/user-attachments/assets/008829d8-51b6-445a-b135-5f76b6ccf292"
/>

### on passing the right token:
overview page
<img width="1720" height="1073" alt="image"
src="https://github.com/user-attachments/assets/cece0da7-0edd-4438-9d60-776526b49762"
/>

job page: tasks are listed
<img width="1720" height="1073" alt="image"
src="https://github.com/user-attachments/assets/b98eb1d9-cacc-45ea-b0e2-07ce8922202a"
/>

task page
<img width="1720" height="1073" alt="image"
src="https://github.com/user-attachments/assets/09ff38e1-e151-4e34-8651-d206eb8b5136"
/>

actors page
<img width="1720" height="1073" alt="image"
src="https://github.com/user-attachments/assets/10a30b3d-3f7e-4f3d-b669-962056579459"
/>

specific actor page
<img width="1720" height="1073" alt="image"
src="https://github.com/user-attachments/assets/ab1915bd-3d1b-4813-8101-a219432a55c0"
/>

---------

Signed-off-by: sampan <[email protected]>
Co-authored-by: sampan <[email protected]>

Author: sampan-s-nayak

python/ray/dashboard/authentication_utils.py renamed to python/ray/_private/authentication/authentication_utils.py

@@ -0,0 +1,122 @@
+"""gRPC client interceptor for token-based authentication."""
+
+import logging
+from collections import namedtuple
+from typing import Tuple
+
+import grpc
+from grpc import aio as aiogrpc
+
+from ray._raylet import AuthenticationTokenLoader
+
+logger = logging.getLogger(__name__)
+
+
+# Named tuple to hold client call details
+_ClientCallDetails = namedtuple(
+    "_ClientCallDetails",
+    ("method", "timeout", "metadata", "credentials", "wait_for_ready", "compression"),
+)
+
+
+def _get_authentication_metadata_tuple() -> Tuple[Tuple[str, str], ...]:
+    """Get gRPC metadata tuple for authentication. Currently only supported for token authentication.
+
+    Returns:
+        tuple: Empty tuple or ((AUTHORIZATION_HEADER_NAME, "Bearer <token>"),)
+    """
+    token_loader = AuthenticationTokenLoader.instance()
+    if not token_loader.has_token():
+        return ()
+
+    headers = token_loader.get_token_for_http_header()
+
+    # Convert HTTP header dict to gRPC metadata tuple
+    # gRPC expects: (("key", "value"), ...)
+    return tuple((k, v) for k, v in headers.items())
+
+
+class AuthenticationMetadataClientInterceptor(
+    grpc.UnaryUnaryClientInterceptor,
+    grpc.UnaryStreamClientInterceptor,
+    grpc.StreamUnaryClientInterceptor,
+    grpc.StreamStreamClientInterceptor,
+):
+    """Synchronous gRPC client interceptor that adds authentication metadata."""
+
+    def _intercept_call_details(self, client_call_details):
+        """Helper method to add authentication metadata to client call details."""
+        metadata = list(client_call_details.metadata or [])
+        metadata.extend(_get_authentication_metadata_tuple())
+
+        return _ClientCallDetails(
+            method=client_call_details.method,
+            timeout=client_call_details.timeout,
+            metadata=metadata,
+            credentials=client_call_details.credentials,
+            wait_for_ready=getattr(client_call_details, "wait_for_ready", None),
+            compression=getattr(client_call_details, "compression", None),
+        )
+
+    def intercept_unary_unary(self, continuation, client_call_details, request):
+        new_details = self._intercept_call_details(client_call_details)
+        return continuation(new_details, request)
+
+    def intercept_unary_stream(self, continuation, client_call_details, request):
+        new_details = self._intercept_call_details(client_call_details)
+        return continuation(new_details, request)
+
+    def intercept_stream_unary(
+        self, continuation, client_call_details, request_iterator
+    ):
+        new_details = self._intercept_call_details(client_call_details)
+        return continuation(new_details, request_iterator)
+
+    def intercept_stream_stream(
+        self, continuation, client_call_details, request_iterator
+    ):
+        new_details = self._intercept_call_details(client_call_details)
+        return continuation(new_details, request_iterator)
+
+
+class AsyncAuthenticationMetadataClientInterceptor(
+    aiogrpc.UnaryUnaryClientInterceptor,
+    aiogrpc.UnaryStreamClientInterceptor,
+    aiogrpc.StreamUnaryClientInterceptor,
+    aiogrpc.StreamStreamClientInterceptor,
+):
+    """Async gRPC client interceptor that adds authentication metadata."""
+
+    def _intercept_call_details(self, client_call_details):
+        """Helper method to add authentication metadata to client call details."""
+        metadata = list(client_call_details.metadata or [])
+        metadata.extend(_get_authentication_metadata_tuple())
+
+        return _ClientCallDetails(
+            method=client_call_details.method,
+            timeout=client_call_details.timeout,
+            metadata=metadata,
+            credentials=client_call_details.credentials,
+            wait_for_ready=getattr(client_call_details, "wait_for_ready", None),
+            compression=getattr(client_call_details, "compression", None),
+        )
+
+    async def intercept_unary_unary(self, continuation, client_call_details, request):
+        new_details = self._intercept_call_details(client_call_details)
+        return await continuation(new_details, request)
+
+    async def intercept_unary_stream(self, continuation, client_call_details, request):
+        new_details = self._intercept_call_details(client_call_details)
+        return await continuation(new_details, request)
+
+    async def intercept_stream_unary(
+        self, continuation, client_call_details, request_iterator
+    ):
+        new_details = self._intercept_call_details(client_call_details)
+        return await continuation(new_details, request_iterator)
+
+    async def intercept_stream_stream(
+        self, continuation, client_call_details, request_iterator
+    ):
+        new_details = self._intercept_call_details(client_call_details)
+        return await continuation(new_details, request_iterator)

python/ray/_private/authentication/grpc_authentication_client_interceptor.py

@@ -2,8 +2,10 @@
 from types import ModuleType
 from typing import Dict, List, Optional
 
-from ray._private.authentication import authentication_constants
-from ray.dashboard import authentication_utils as auth_utils
+from ray._private.authentication import (
+    authentication_constants,
+    authentication_utils as auth_utils,
+)
 
 logger = logging.getLogger(__name__)
 

python/ray/_private/authentication/http_token_authentication.py

@@ -1026,6 +1026,7 @@ def init_grpc_channel(
     import grpc
     from grpc import aio as aiogrpc
 
+    from ray._private.authentication import authentication_utils
     from ray._private.tls_utils import load_certs_from_env
 
     grpc_module = aiogrpc if asynchronous else grpc
@@ -1040,16 +1041,43 @@ def init_grpc_channel(
     )
     options = options_dict.items()
 
-    if os.environ.get("RAY_USE_TLS", "0").lower() in ("1", "true"):
+    # Build interceptors list
+    interceptors = []
+    if authentication_utils.is_token_auth_enabled():
+        from ray._private.authentication.grpc_authentication_client_interceptor import (
+            AsyncAuthenticationMetadataClientInterceptor,
+            AuthenticationMetadataClientInterceptor,
+        )
+
+        if asynchronous:
+            interceptors.append(AsyncAuthenticationMetadataClientInterceptor())
+        else:
+            interceptors.append(AuthenticationMetadataClientInterceptor())
+
+    # Create channel with TLS if enabled
+    use_tls = os.environ.get("RAY_USE_TLS", "0").lower() in ("1", "true")
+    if use_tls:
         server_cert_chain, private_key, ca_cert = load_certs_from_env()
         credentials = grpc.ssl_channel_credentials(
             certificate_chain=server_cert_chain,
             private_key=private_key,
             root_certificates=ca_cert,
         )
-        channel = grpc_module.secure_channel(address, credentials, options=options)
+        channel_creator = grpc_module.secure_channel
+        base_args = (address, credentials)
+    else:
+        channel_creator = grpc_module.insecure_channel
+        base_args = (address,)
+
+    # Create channel (async channels get interceptors in constructor, sync via intercept_channel)
+    if asynchronous:
+        channel = channel_creator(
+            *base_args, options=options, interceptors=interceptors
+        )
     else:
-        channel = grpc_module.insecure_channel(address, options=options)
+        channel = channel_creator(*base_args, options=options)
+        if interceptors:
+            channel = grpc.intercept_channel(channel, *interceptors)
 
     return channel
 

python/ray/_private/utils.py

@@ -342,5 +342,61 @@ def worker_joined():
             _cleanup_ray_start(env)
 
 
+def test_e2e_operations_with_token_auth(setup_cluster_with_token_auth):
+    """Test that e2e operations work with token authentication enabled.
+
+    This verifies that with token auth enabled:
+    1. Job submission works
+    2. Tasks execute successfully
+    3. Actors can be created and called
+    """
+    cluster_info = setup_cluster_with_token_auth
+
+    # Test 1: Submit a simple task
+    @ray.remote
+    def simple_task(x):
+        return x + 1
+
+    result = ray.get(simple_task.remote(41))
+    assert result == 42, f"Task should return 42, got {result}"
+
+    # Test 2: Create and use an actor
+    @ray.remote
+    class SimpleActor:
+        def __init__(self):
+            self.value = 0
+
+        def increment(self):
+            self.value += 1
+            return self.value
+
+    actor = SimpleActor.remote()
+    result = ray.get(actor.increment.remote())
+    assert result == 1, f"Actor method should return 1, got {result}"
+
+    # Test 3: Submit a job and wait for completion
+    from ray.job_submission import JobSubmissionClient
+
+    # Create job submission client (uses HTTP with auth headers)
+    client = JobSubmissionClient(address=cluster_info["dashboard_url"])
+
+    # Submit a simple job
+    job_id = client.submit_job(
+        entrypoint="echo 'Hello from job'",
+    )
+
+    # Wait for job to complete
+    def job_finished():
+        status = client.get_job_status(job_id)
+        return status in ["SUCCEEDED", "FAILED", "STOPPED"]
+
+    wait_for_condition(job_finished, timeout=30)
+
+    final_status = client.get_job_status(job_id)
+    assert (
+        final_status == "SUCCEEDED"
+    ), f"Job should succeed, got status: {final_status}"
+
+
 if __name__ == "__main__":
     sys.exit(pytest.main(["-vv", __file__]))