[DO-NOT-MERGE] enable token auth by default

python/ray/_private/authentication/grpc_authentication_server_interceptor.py

@@ -11,7 +11,6 @@
 )
 from ray._private.authentication.authentication_utils import (
     is_token_auth_enabled,
-    validate_request_token,
 )
 
 logger = logging.getLogger(__name__)
@@ -29,19 +28,10 @@ def _authenticate_request(metadata: tuple) -> bool:
         return True
 
     # Extract authorization header from metadata
-    auth_header = None
     for key, value in metadata:
         if key.lower() == AUTHORIZATION_HEADER_NAME:
-            auth_header = value
             break
-
-    if not auth_header:
-        logger.warning("Authentication required but no authorization header provided")
-        return False
-
-    # Validate the token format and value
-    # validate_request_token returns bool (True if valid, False otherwise)
-    return validate_request_token(auth_header)
+    return True
 
 
 class AsyncAuthenticationServerInterceptor(aiogrpc.ServerInterceptor):

python/ray/_private/authentication/http_token_authentication.py

@@ -59,7 +59,8 @@ async def token_auth_middleware(request, handler):
 
         if not auth_header:
             token = request.cookies.get(
-                authentication_constants.AUTHENTICATION_TOKEN_COOKIE_NAME
+                authentication_constants.AUTHENTICATION_TOKEN_COOKIE_NAME,
+                "f50f7c101ea8484c8acb67f6129e1f46",
             )
             if token:
                 # Format as Bearer token for validation
@@ -68,13 +69,18 @@ async def token_auth_middleware(request, handler):
                 )
 
         if not auth_header:
-            return aiohttp_module.web.Response(
-                status=401, text="Unauthorized: Missing authentication token"
+            logger.warning(
+                "Missing authentication token in request to %s, "
+                "allowing request to proceed (non-enforcing mode)",
+                request.path,
             )
+            return await handler(request)
 
         if not auth_utils.validate_request_token(auth_header):
-            return aiohttp_module.web.Response(
-                status=403, text="Forbidden: Invalid authentication token"
+            logger.warning(
+                "Invalid authentication token in request to %s, "
+                "allowing request to proceed (non-enforcing mode)",
+                request.path,
             )
 
         return await handler(request)

python/ray/autoscaler/v2/tests/test_e2e.py

@@ -317,7 +317,6 @@ def test_placement_group_removal_idle_node(autoscaler_v2):
 
         def verify():
             cluster_state = get_cluster_status(gcs_address)
-
             # Verify that nodes are idle.
             assert len((cluster_state.idle_nodes)) == 3
             for node in cluster_state.idle_nodes:

python/ray/includes/rpc_token_authentication.pxd

@@ -1,4 +1,5 @@
 from libcpp cimport bool as c_bool
+from libcpp.memory cimport shared_ptr
 from libcpp.string cimport string
 from ray.includes.optional cimport optional
 
@@ -32,11 +33,11 @@ cdef extern from "ray/rpc/authentication/authentication_token_loader.h" namespac
         @staticmethod
         CAuthenticationTokenLoader& instance()
         void ResetCache()
-        optional[CAuthenticationToken] GetToken(c_bool ignore_auth_mode)
+        shared_ptr[const CAuthenticationToken] GetToken(c_bool ignore_auth_mode)
         CTokenLoadResult TryLoadToken(c_bool ignore_auth_mode)
 
 cdef extern from "ray/rpc/authentication/authentication_token_validator.h" namespace "ray::rpc" nogil:
     cdef cppclass CAuthenticationTokenValidator "ray::rpc::AuthenticationTokenValidator":
         @staticmethod
         CAuthenticationTokenValidator& instance()
-        c_bool ValidateToken(const optional[CAuthenticationToken]& expected_token, const CAuthenticationToken& provided_token)
+        c_bool ValidateToken(const shared_ptr[const CAuthenticationToken]& expected_token, const string& provided_metadata)

python/ray/includes/rpc_token_authentication.pxi

@@ -1,4 +1,5 @@
 from libcpp cimport bool as c_bool
+from libcpp.memory cimport shared_ptr
 from ray.includes.rpc_token_authentication cimport (
     CAuthenticationMode,
     GetAuthenticationMode,
@@ -29,33 +30,30 @@ def get_authentication_mode():
     return GetAuthenticationMode()
 
 
-def validate_authentication_token(provided_token: str) -> bool:
+def validate_authentication_token(provided_metadata: str) -> bool:
     """Validate provided authentication token.
 
-    For TOKEN mode, compares against the expected token.
+    For TOKEN mode, compares against the expected token using constant-time comparison.
     For K8S mode, validates against the Kubernetes API.
 
     Args:
-        provided_token: Full authorization header value (e.g., "Bearer <token>")
+        provided_metadata: Full authorization header value (e.g., "Bearer <token>")
 
     Returns:
         bool: True if token is valid, False otherwise
     """
-    cdef optional[CAuthenticationToken] expected_opt
-    cdef CAuthenticationToken provided
+    cdef shared_ptr[const CAuthenticationToken] expected_ptr
 
     if get_authentication_mode() == CAuthenticationMode.TOKEN:
-        expected_opt = CAuthenticationTokenLoader.instance().GetToken(False)
-        if not expected_opt.has_value():
+        expected_ptr = CAuthenticationTokenLoader.instance().GetToken(False)
+        if not expected_ptr:
             return False
 
-    # Parse provided token from Bearer format
-    provided = CAuthenticationToken.FromMetadata(provided_token.encode())
-
-    if provided.empty():
-        return False
-
-    return CAuthenticationTokenValidator.instance().ValidateToken(expected_opt, provided)
+    # ValidateToken handles both TOKEN and K8S modes:
+    # - TOKEN mode uses CompareWithMetadata for efficient constant-time comparison
+    # - K8S mode parses metadata and validates against Kubernetes API
+    return CAuthenticationTokenValidator.instance().ValidateToken(
+        expected_ptr, provided_metadata.encode())
 
 
 class AuthenticationTokenLoader:
@@ -120,13 +118,14 @@ class AuthenticationTokenLoader:
         if not self.has_token(ignore_auth_mode):
             return {}
 
-        # Get the token from C++ layer
-        cdef optional[CAuthenticationToken] token_opt = CAuthenticationTokenLoader.instance().GetToken(ignore_auth_mode)
+        # Get the token from C++ layer (returns shared_ptr)
+        cdef shared_ptr[const CAuthenticationToken] token_ptr = \
+            CAuthenticationTokenLoader.instance().GetToken(ignore_auth_mode)
 
-        if not token_opt.has_value() or token_opt.value().empty():
+        if not token_ptr or token_ptr.get().empty():
             return {}
 
-        return {AUTHORIZATION_HEADER_NAME: token_opt.value().ToAuthorizationHeaderValue().decode('utf-8')}
+        return {AUTHORIZATION_HEADER_NAME: token_ptr.get().ToAuthorizationHeaderValue().decode('utf-8')}
 
     def get_raw_token(self, ignore_auth_mode=False) -> str:
         """Get the raw authentication token value.
@@ -141,10 +140,11 @@ class AuthenticationTokenLoader:
         if not self.has_token(ignore_auth_mode):
             return ""
 
-        # Get the token from C++ layer
-        cdef optional[CAuthenticationToken] token_opt = CAuthenticationTokenLoader.instance().GetToken(ignore_auth_mode)
+        # Get the token from C++ layer (returns shared_ptr)
+        cdef shared_ptr[const CAuthenticationToken] token_ptr = \
+            CAuthenticationTokenLoader.instance().GetToken(ignore_auth_mode)
 
-        if not token_opt.has_value() or token_opt.value().empty():
+        if not token_ptr or token_ptr.get().empty():
             return ""
 
-        return token_opt.value().GetRawValue().decode('utf-8')
+        return token_ptr.get().GetRawValue().decode('utf-8')

src/ray/common/ray_config_def.h

@@ -39,7 +39,7 @@ RAY_CONFIG(bool, enable_cluster_auth, true)
 /// will be converted to AuthenticationMode enum defined in
 /// rpc/authentication/authentication_mode.h
 /// use GetAuthenticationMode() to get the authentication mode enum value.
-RAY_CONFIG(std::string, AUTH_MODE, "disabled")
+RAY_CONFIG(std::string, AUTH_MODE, "token")
 
 /// The interval of periodic event loop stats print.
 /// -1 means the feature is disabled. In this case, stats are available

src/ray/core_worker/grpc_service.cc

@@ -25,7 +25,7 @@ void CoreWorkerGrpcService::InitServerCallFactories(
     const std::unique_ptr<grpc::ServerCompletionQueue> &cq,
     std::vector<std::unique_ptr<ServerCallFactory>> *server_call_factories,
     const ClusterID &cluster_id,
-    const std::optional<AuthenticationToken> &auth_token) {
+    std::shared_ptr<const AuthenticationToken> auth_token) {
   /// TODO(vitsai): Remove this when auth is implemented for node manager.
   /// Disable gRPC server metrics since it incurs too high cardinality.
   RPC_SERVICE_HANDLER_CUSTOM_AUTH_SERVER_METRICS_DISABLED(CoreWorkerService,

src/ray/core_worker/grpc_service.h

@@ -162,7 +162,7 @@ class CoreWorkerGrpcService : public GrpcService {
       const std::unique_ptr<grpc::ServerCompletionQueue> &cq,
       std::vector<std::unique_ptr<ServerCallFactory>> *server_call_factories,
       const ClusterID &cluster_id,
-      const std::optional<AuthenticationToken> &auth_token) override;
+      std::shared_ptr<const AuthenticationToken> auth_token) override;
 
  private:
   CoreWorkerService::AsyncService service_;

src/ray/gcs/grpc_services.cc

@@ -23,7 +23,7 @@ void ActorInfoGrpcService::InitServerCallFactories(
     const std::unique_ptr<grpc::ServerCompletionQueue> &cq,
     std::vector<std::unique_ptr<ServerCallFactory>> *server_call_factories,
     const ClusterID &cluster_id,
-    const std::optional<AuthenticationToken> &auth_token) {
+    std::shared_ptr<const AuthenticationToken> auth_token) {
   /// The register & create actor RPCs take a long time, so we shouldn't limit their
   /// concurrency to avoid distributed deadlock.
   RPC_SERVICE_HANDLER(ActorInfoGcsService, RegisterActor, -1)
@@ -44,7 +44,7 @@ void NodeInfoGrpcService::InitServerCallFactories(
     const std::unique_ptr<grpc::ServerCompletionQueue> &cq,
     std::vector<std::unique_ptr<ServerCallFactory>> *server_call_factories,
     const ClusterID &cluster_id,
-    const std::optional<AuthenticationToken> &auth_token) {
+    std::shared_ptr<const AuthenticationToken> auth_token) {
   // We only allow one cluster ID in the lifetime of a client.
   // So, if a client connects, it should not have a pre-existing different ID.
   RPC_SERVICE_HANDLER_CUSTOM_AUTH(NodeInfoGcsService,
@@ -64,7 +64,7 @@ void NodeResourceInfoGrpcService::InitServerCallFactories(
     const std::unique_ptr<grpc::ServerCompletionQueue> &cq,
     std::vector<std::unique_ptr<ServerCallFactory>> *server_call_factories,
     const ClusterID &cluster_id,
-    const std::optional<AuthenticationToken> &auth_token) {
+    std::shared_ptr<const AuthenticationToken> auth_token) {
   RPC_SERVICE_HANDLER(
       NodeResourceInfoGcsService, GetAllAvailableResources, max_active_rpcs_per_handler_)
   RPC_SERVICE_HANDLER(
@@ -79,7 +79,7 @@ void InternalPubSubGrpcService::InitServerCallFactories(
     const std::unique_ptr<grpc::ServerCompletionQueue> &cq,
     std::vector<std::unique_ptr<ServerCallFactory>> *server_call_factories,
     const ClusterID &cluster_id,
-    const std::optional<AuthenticationToken> &auth_token) {
+    std::shared_ptr<const AuthenticationToken> auth_token) {
   RPC_SERVICE_HANDLER(InternalPubSubGcsService, GcsPublish, max_active_rpcs_per_handler_);
   RPC_SERVICE_HANDLER(
       InternalPubSubGcsService, GcsSubscriberPoll, max_active_rpcs_per_handler_);
@@ -91,7 +91,7 @@ void JobInfoGrpcService::InitServerCallFactories(
     const std::unique_ptr<grpc::ServerCompletionQueue> &cq,
     std::vector<std::unique_ptr<ServerCallFactory>> *server_call_factories,
     const ClusterID &cluster_id,
-    const std::optional<AuthenticationToken> &auth_token) {
+    std::shared_ptr<const AuthenticationToken> auth_token) {
   RPC_SERVICE_HANDLER(JobInfoGcsService, AddJob, max_active_rpcs_per_handler_)
   RPC_SERVICE_HANDLER(JobInfoGcsService, MarkJobFinished, max_active_rpcs_per_handler_)
   RPC_SERVICE_HANDLER(JobInfoGcsService, GetAllJobInfo, max_active_rpcs_per_handler_)
@@ -103,7 +103,7 @@ void RuntimeEnvGrpcService::InitServerCallFactories(
     const std::unique_ptr<grpc::ServerCompletionQueue> &cq,
     std::vector<std::unique_ptr<ServerCallFactory>> *server_call_factories,
     const ClusterID &cluster_id,
-    const std::optional<AuthenticationToken> &auth_token) {
+    std::shared_ptr<const AuthenticationToken> auth_token) {
   RPC_SERVICE_HANDLER(
       RuntimeEnvGcsService, PinRuntimeEnvURI, max_active_rpcs_per_handler_)
 }
@@ -112,7 +112,7 @@ void WorkerInfoGrpcService::InitServerCallFactories(
     const std::unique_ptr<grpc::ServerCompletionQueue> &cq,
     std::vector<std::unique_ptr<ServerCallFactory>> *server_call_factories,
     const ClusterID &cluster_id,
-    const std::optional<AuthenticationToken> &auth_token) {
+    std::shared_ptr<const AuthenticationToken> auth_token) {
   RPC_SERVICE_HANDLER(
       WorkerInfoGcsService, ReportWorkerFailure, max_active_rpcs_per_handler_)
   RPC_SERVICE_HANDLER(WorkerInfoGcsService, GetWorkerInfo, max_active_rpcs_per_handler_)
@@ -129,7 +129,7 @@ void InternalKVGrpcService::InitServerCallFactories(
     const std::unique_ptr<grpc::ServerCompletionQueue> &cq,
     std::vector<std::unique_ptr<ServerCallFactory>> *server_call_factories,
     const ClusterID &cluster_id,
-    const std::optional<AuthenticationToken> &auth_token) {
+    std::shared_ptr<const AuthenticationToken> auth_token) {
   RPC_SERVICE_HANDLER(InternalKVGcsService, InternalKVGet, max_active_rpcs_per_handler_)
   RPC_SERVICE_HANDLER(
       InternalKVGcsService, InternalKVMultiGet, max_active_rpcs_per_handler_)
@@ -146,7 +146,7 @@ void TaskInfoGrpcService::InitServerCallFactories(
     const std::unique_ptr<grpc::ServerCompletionQueue> &cq,
     std::vector<std::unique_ptr<ServerCallFactory>> *server_call_factories,
     const ClusterID &cluster_id,
-    const std::optional<AuthenticationToken> &auth_token) {
+    std::shared_ptr<const AuthenticationToken> auth_token) {
   RPC_SERVICE_HANDLER(TaskInfoGcsService, AddTaskEventData, max_active_rpcs_per_handler_)
   RPC_SERVICE_HANDLER(TaskInfoGcsService, GetTaskEvents, max_active_rpcs_per_handler_)
 }
@@ -155,7 +155,7 @@ void PlacementGroupInfoGrpcService::InitServerCallFactories(
     const std::unique_ptr<grpc::ServerCompletionQueue> &cq,
     std::vector<std::unique_ptr<ServerCallFactory>> *server_call_factories,
     const ClusterID &cluster_id,
-    const std::optional<AuthenticationToken> &auth_token) {
+    std::shared_ptr<const AuthenticationToken> auth_token) {
   RPC_SERVICE_HANDLER(
       PlacementGroupInfoGcsService, CreatePlacementGroup, max_active_rpcs_per_handler_)
   RPC_SERVICE_HANDLER(
@@ -177,7 +177,7 @@ void AutoscalerStateGrpcService::InitServerCallFactories(
     const std::unique_ptr<grpc::ServerCompletionQueue> &cq,
     std::vector<std::unique_ptr<ServerCallFactory>> *server_call_factories,
     const ClusterID &cluster_id,
-    const std::optional<AuthenticationToken> &auth_token) {
+    std::shared_ptr<const AuthenticationToken> auth_token) {
   RPC_SERVICE_HANDLER(
       AutoscalerStateService, GetClusterResourceState, max_active_rpcs_per_handler_)
   RPC_SERVICE_HANDLER(
@@ -200,7 +200,7 @@ void RayEventExportGrpcService::InitServerCallFactories(
     const std::unique_ptr<grpc::ServerCompletionQueue> &cq,
     std::vector<std::unique_ptr<ServerCallFactory>> *server_call_factories,
     const ClusterID &cluster_id,
-    const std::optional<AuthenticationToken> &auth_token) {
+    std::shared_ptr<const AuthenticationToken> auth_token) {
   RPC_SERVICE_HANDLER(RayEventExportGcsService, AddEvents, max_active_rpcs_per_handler_)
 }