Add support for RAY_AUTH_MODE=k8s

python/ray/_private/authentication/authentication_utils.py

@@ -15,11 +15,15 @@ def is_token_auth_enabled() -> bool:
     """Check if token authentication is enabled.
 
     Returns:
-        bool: True if AUTH_MODE is set to "token", False otherwise
+        bool: True if AUTH_MODE is set to "token" or "k8s", False otherwise
     """
     if not _RAYLET_AVAILABLE:
         return False
-    return get_authentication_mode() == AuthenticationMode.TOKEN
+
+    return get_authentication_mode() in {
+        AuthenticationMode.TOKEN,
+        AuthenticationMode.K8S,
+    }
 
 
 def validate_request_token(auth_header: str) -> bool:

python/ray/dashboard/http_server_head.py

@@ -173,6 +173,8 @@ async def get_authentication_mode(self, req) -> aiohttp.web.Response:
             mode = get_authentication_mode()
             if mode == AuthenticationMode.TOKEN:
                 mode_str = "token"
+            elif mode == AuthenticationMode.K8S:
+                mode_str = "k8s"
             else:
                 mode_str = "disabled"
 

python/ray/includes/rpc_token_authentication.pxd

@@ -7,6 +7,7 @@ cdef extern from "ray/rpc/authentication/authentication_mode.h" namespace "ray::
     cdef enum CAuthenticationMode "ray::rpc::AuthenticationMode":
         DISABLED "ray::rpc::AuthenticationMode::DISABLED"
         TOKEN "ray::rpc::AuthenticationMode::TOKEN"
+        K8S "ray::rpc::AuthenticationMode::K8S"
 
     CAuthenticationMode GetAuthenticationMode()
 
@@ -28,3 +29,9 @@ cdef extern from "ray/rpc/authentication/authentication_token_loader.h" namespac
         c_bool HasToken()
         void ResetCache()
         optional[CAuthenticationToken] GetToken()
+
+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)

python/ray/includes/rpc_token_authentication.pxi

@@ -3,48 +3,57 @@ from ray.includes.rpc_token_authentication cimport (
     GetAuthenticationMode,
     CAuthenticationToken,
     CAuthenticationTokenLoader,
+    CAuthenticationTokenValidator,
 )
 from ray._private.authentication.authentication_constants import AUTHORIZATION_HEADER_NAME
+import logging
+
+logger = logging.getLogger(__name__)
 
 
 # Authentication mode enum exposed to Python
 class AuthenticationMode:
     DISABLED = CAuthenticationMode.DISABLED
     TOKEN = CAuthenticationMode.TOKEN
+    K8S = CAuthenticationMode.K8S
 
 
 def get_authentication_mode():
     """Get the current authentication mode.
 
     Returns:
-        AuthenticationMode enum value (DISABLED or TOKEN)
+        AuthenticationMode enum value (DISABLED or TOKEN or K8S)
     """
     return GetAuthenticationMode()
 
 
 def validate_authentication_token(provided_token: str) -> bool:
-    """Validate provided authentication token against expected token.
+    """Validate provided authentication token.
+
+    For TOKEN mode, compares against the expected token.
+    For K8S mode, validates against the Kubernetes API.
 
     Args:
         provided_token: Full authorization header value (e.g., "Bearer <token>")
 
     Returns:
-        bool: True if tokens match, False otherwise
+        bool: True if token is valid, False otherwise
     """
-    # Get expected token from loader
-    cdef optional[CAuthenticationToken] expected_opt = CAuthenticationTokenLoader.instance().GetToken()
+    cdef optional[CAuthenticationToken] expected_opt
+    cdef CAuthenticationToken provided
 
-    if not expected_opt.has_value():
-        return False
+    if get_authentication_mode() == CAuthenticationMode.TOKEN:
+        expected_opt = CAuthenticationTokenLoader.instance().GetToken()
+        if not expected_opt.has_value():
+            return False
 
     # Parse provided token from Bearer format
-    cdef CAuthenticationToken provided = CAuthenticationToken.FromMetadata(provided_token.encode())
+    provided = CAuthenticationToken.FromMetadata(provided_token.encode())
 
     if provided.empty():
         return False
 
-    # Use constant-time comparison from C++
-    return expected_opt.value().Equals(provided)
+    return CAuthenticationTokenValidator.instance().ValidateToken(expected_opt, provided)
 
 
 class AuthenticationTokenLoader:

src/ray/rpc/BUILD.bazel

@@ -110,7 +110,10 @@ ray_cc_library(
         "//src/ray/common:id",
         "//src/ray/common:ray_config",
         "//src/ray/common:status",
+        "//src/ray/rpc/authentication:authentication_mode",
         "//src/ray/rpc/authentication:authentication_token",
+        "//src/ray/rpc/authentication:authentication_token_loader",
+        "//src/ray/rpc/authentication:authentication_token_validator",
         "//src/ray/stats:stats_metric",
         "@com_github_grpc_grpc//:grpc++",
     ],

src/ray/rpc/authentication/BUILD.bazel

@@ -21,6 +21,27 @@ ray_cc_library(
     ],
 )
 
+ray_cc_library(
+    name = "k8s_constants",
+    hdrs = ["k8s_constants.h"],
+    visibility = ["//visibility:public"],
+)
+
+ray_cc_library(
+    name = "k8s_util",
+    srcs = ["k8s_util.cc"],
+    hdrs = ["k8s_util.h"],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":authentication_token",
+        ":k8s_constants",
+        "//src/ray/util:logging",
+        "@boost//:asio_ssl",
+        "@boost//:beast",
+        "@nlohmann_json",
+    ],
+)
+
 ray_cc_library(
     name = "authentication_token_loader",
     srcs = ["authentication_token_loader.cc"],
@@ -29,6 +50,7 @@ ray_cc_library(
     deps = [
         ":authentication_mode",
         ":authentication_token",
+        ":k8s_constants",
         "//src/ray/util:logging",
     ],
 )
@@ -45,3 +67,15 @@ ray_cc_library(
         "@com_github_grpc_grpc//:grpc++",
     ],
 )
+
+ray_cc_library(
+    name = "authentication_token_validator",
+    srcs = ["authentication_token_validator.cc"],
+    hdrs = ["authentication_token_validator.h"],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":authentication_mode",
+        ":authentication_token",
+        ":k8s_util",
+    ],
+)

src/ray/rpc/authentication/authentication_mode.cc

@@ -28,10 +28,17 @@ AuthenticationMode GetAuthenticationMode() {
 
   if (auth_mode_lower == "token") {
     return AuthenticationMode::TOKEN;
+  } else if (auth_mode_lower == "k8s") {
+    return AuthenticationMode::K8S;
   } else {
     return AuthenticationMode::DISABLED;
   }
 }
 
+bool RequiresTokenAuthentication() {
+  AuthenticationMode auth_mode = GetAuthenticationMode();
+  return (auth_mode == AuthenticationMode::TOKEN || auth_mode == AuthenticationMode::K8S);
+}
+
 }  // namespace rpc
 }  // namespace ray

src/ray/rpc/authentication/authentication_mode.h

@@ -22,12 +22,17 @@ namespace rpc {
 enum class AuthenticationMode {
   DISABLED,
   TOKEN,
+  K8S,
 };
 
 /// Get the authentication mode from the RayConfig.
 /// \return The authentication mode enum value. returns AuthenticationMode::DISABLED if
 /// the authentication mode is not set or is invalid.
 AuthenticationMode GetAuthenticationMode();
 
+// Checks authentication mode and returns whether token authentication is required.
+// \return bool
+bool RequiresTokenAuthentication();
+
 }  // namespace rpc
 }  // namespace ray

src/ray/rpc/authentication/authentication_token.h

@@ -107,6 +107,14 @@ class AuthenticationToken {
     return std::string(secret_.begin(), secret_.end());
   }
 
+  /// Get token hash
+  /// @return Hash of the token value
+  std::size_t ToHash() const {
+    // TODO(andrewsykim): consider using a more secure hashing algorithm like SHA256
+    // before documenting this feature in Ray docs.
+    return std::hash<std::string>()(std::string(secret_.begin(), secret_.end()));
+  }
+
   /// Create AuthenticationToken from gRPC metadata value
   /// Strips "Bearer " prefix and creates token object
   /// @param metadata_value The raw value from server metadata (should include "Bearer "
@@ -173,5 +181,12 @@ class AuthenticationToken {
   }
 };
 
+// Hash function for AuthenticationToken
+struct AuthenticationTokenHash {
+  std::size_t operator()(const AuthenticationToken &token) const {
+    return token.ToHash();
+  }
+};
+
 }  // namespace rpc
 }  // namespace ray

src/ray/rpc/authentication/authentication_token_loader.cc

@@ -18,6 +18,7 @@
 #include <string>
 #include <utility>
 
+#include "ray/rpc/authentication/k8s_constants.h"
 #include "ray/util/logging.h"
 
 #ifdef _WIN32
@@ -47,8 +48,8 @@ std::optional<AuthenticationToken> AuthenticationTokenLoader::GetToken() {
     return cached_token_;
   }
 
-  // If token auth is not enabled, return std::nullopt
-  if (GetAuthenticationMode() != AuthenticationMode::TOKEN) {
+  // If token or k8s auth is not enabled, return std::nullopt
+  if (!RequiresTokenAuthentication()) {
     cached_token_ = std::nullopt;
     return std::nullopt;
   }
@@ -77,8 +78,8 @@ bool AuthenticationTokenLoader::HasToken() {
     return !cached_token_->empty();
   }
 
-  // If token auth is not enabled, no token needed
-  if (GetAuthenticationMode() != AuthenticationMode::TOKEN) {
+  // If token or k8s auth is not enabled, no token needed
+  if (!RequiresTokenAuthentication()) {
     cached_token_ = std::nullopt;
     return false;
   }
@@ -135,7 +136,20 @@ AuthenticationToken AuthenticationTokenLoader::LoadTokenFromSources() {
     }
   }
 
-  // Precedence 3: Default token path ~/.ray/auth_token
+  // Precedence 3 (auth_mode=k8s only): Load Kubernetes service account token
+  if (GetAuthenticationMode() == AuthenticationMode::K8S) {
+    std::string token_str = TrimWhitespace(ReadTokenFromFile(k8s::kK8sSaTokenPath));
+    if (!token_str.empty()) {
+      RAY_LOG(DEBUG)
+          << "Loaded authentication token from Kubernetes service account path: "
+          << k8s::kK8sSaTokenPath;
+      return AuthenticationToken(token_str);
+    }
+    RAY_LOG(DEBUG) << "Kubernetes service account token not found or empty at: "
+                   << k8s::kK8sSaTokenPath;
+  }
+
+  // Precedence 4: Default token path ~/.ray/auth_token
   std::string default_path = GetDefaultTokenPath();
   std::string token_str = TrimWhitespace(ReadTokenFromFile(default_path));
   if (!token_str.empty()) {

src/ray/rpc/authentication/authentication_token_validator.cc

@@ -0,0 +1,86 @@
+// Copyright 2025 The Ray Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions
+// and limitations under the License.
+
+#include "ray/rpc/authentication/authentication_token_validator.h"
+
+#include "ray/rpc/authentication/authentication_mode.h"
+#include "ray/rpc/authentication/k8s_util.h"
+#include "ray/util/logging.h"
+
+namespace ray {
+namespace rpc {
+
+const std::chrono::minutes kCacheTTL(5);
+
+AuthenticationTokenValidator &AuthenticationTokenValidator::instance() {
+  static AuthenticationTokenValidator instance;
+  return instance;
+}
+
+bool AuthenticationTokenValidator::ValidateToken(
+    const std::optional<AuthenticationToken> &expected_token,
+    const AuthenticationToken &provided_token) {
+  if (GetAuthenticationMode() == AuthenticationMode::TOKEN) {
+    RAY_CHECK(expected_token.has_value() && !expected_token->empty())
+        << "Ray token authentication is enabled but expected token is empty";
+
+    return expected_token->Equals(provided_token);
+  }
+
+  if (GetAuthenticationMode() == AuthenticationMode::K8S) {
+    std::call_once(k8s::k8s_client_config_flag, k8s::InitK8sClientConfig);
+    if (!k8s::k8s_client_initialized) {
+      RAY_LOG(WARNING) << "Kubernetes client not initialized, K8s authentication failed.";
+      return false;
+    }
+
+    // Check cache first.
+    {
+      std::lock_guard<std::mutex> lock(k8s_token_cache_mutex_);
+      auto it = k8s_token_cache_.find(provided_token);
+      if (it != k8s_token_cache_.end()) {
+        if (std::chrono::steady_clock::now() < it->second.expiration) {
+          RAY_LOG(DEBUG) << "K8s token found in cache and is valid.";
+          return it->second.allowed;
+        } else {
+          RAY_LOG(DEBUG) << "K8s token in cache expired, removing from cache.";
+          k8s_token_cache_.erase(it);
+        }
+      }
+    }
+
+    bool is_allowed = false;
+    is_allowed = k8s::ValidateToken(provided_token);
+
+    // Only cache validated tokens for now. We don't want to invalidate a token
+    // due to unrelated errors from Kubernetes API server. This has the downside of
+    // causing more load if an unauthenticated client continues to make calls.
+    // TODO(andrewsykim): cache invalid tokens once k8s::ValidateToken can distinguish
+    // between invalid token errors and server errors.
+    if (is_allowed) {
+      std::lock_guard<std::mutex> lock(k8s_token_cache_mutex_);
+      k8s_token_cache_[provided_token] = {is_allowed,
+                                          std::chrono::steady_clock::now() + kCacheTTL};
+      RAY_LOG(DEBUG) << "K8s token validated and saved to cache.";
+    }
+
+    return is_allowed;
+  }
+
+  RAY_LOG(DEBUG) << "Authentication mode is disabled, token considered valid.";
+  return true;
+}
+
+}  // namespace rpc
+}  // namespace ray