Support for Node-Specific Restrictions in Fluid

charts/fluid/fluid/templates/csi/daemonset.yaml

@@ -112,7 +112,7 @@ spec:
           - name: fluid-src-dir
             mountPath: {{ .Values.runtime.mountRoot | quote }}
             mountPropagation: "HostToContainer"
-          {{- if .Values.csi.useNodeAuthorization }}
+          {{- if or (.Values.csi.useNodeAuthorization) (semverCompare "<1.30.0-0" .Capabilities.KubeVersion.Version) }}
           - name: kubelet-kube-config
             mountPath: /etc/kubernetes/kubelet.conf
             mountPropagation: "HostToContainer"
@@ -134,7 +134,7 @@ spec:
           hostPath:
             path: {{ .Values.csi.kubelet.rootDir | quote }}
             type: Directory
-        {{- if .Values.csi.useNodeAuthorization }}
+        {{- if or (.Values.csi.useNodeAuthorization) (semverCompare "<1.30.0-0" .Capabilities.KubeVersion.Version) }}
         {{- $kubeletRootDir := ternary ( .Values.csi.kubelet.rootDir ) ( print .Values.csi.kubelet.rootDir "/" ) ( hasSuffix "/" .Values.csi.kubelet.rootDir ) }}
         {{- if not ( hasPrefix $kubeletRootDir .Values.csi.kubelet.certDir ) }}
         - name: kubelet-cert-dir

charts/fluid/fluid/templates/csi/validatingadmissionpolicy.yaml

@@ -0,0 +1,32 @@
+{{- if and (not .Values.csi.useNodeAuthorization) (semverCompare ">=1.30.0-0" .Capabilities.KubeVersion.Version) -}}
+apiVersion: admissionregistration.k8s.io/v1
+kind: ValidatingAdmissionPolicy
+metadata:
+  name: "fluid-csi-node-policy"
+spec:
+  failurePolicy: Fail
+  matchConstraints:
+    resourceRules:
+      - apiGroups: [""]
+        apiVersions: ["v1"]
+        # supported values: "*", "CONNECT", "CREATE", "DELETE", "UPDATE"
+        operations: ["UPDATE"]
+        resources: ["nodes"]
+  matchConditions:
+    # only fluid-csi request will be checked.
+    - name: isRestrictedUser
+      expression: request.userInfo.username == "system:serviceaccount:fluid-system:fluid-csi"
+  variables:
+    - name: userNodeName
+      expression: >-
+        request.userInfo.extra[?'authentication.kubernetes.io/node-name'][0].orValue('')
+    - name: objectNodeName
+      expression: >-
+        object.?metadata.name.orValue('')
+  validations:
+    - expression: "variables.userNodeName != ''"
+      message: "userNodeName is empty, user token does not contain node name."
+    - expression: "variables.objectNodeName == variables.userNodeName"
+      messageExpression: >-
+        "objectNodeName '" + variables.objectNodeName + "' is not equal to userNodeName '" + variables.userNodeName + "'"
+{{- end }}

charts/fluid/fluid/templates/csi/validatingadmissionpolicybinding.yaml

@@ -0,0 +1,9 @@
+{{- if and (not .Values.csi.useNodeAuthorization) (semverCompare ">=1.30.0-0" .Capabilities.KubeVersion.Version) -}}
+apiVersion: admissionregistration.k8s.io/v1
+kind: ValidatingAdmissionPolicyBinding
+metadata:
+  name:  "fluid-csi-node-policy-binding"
+spec:
+  policyName:  "fluid-csi-node-policy"
+  validationActions: [Deny]
+{{- end }}

charts/fluid/fluid/templates/role/csi/rbac.yaml

@@ -44,7 +44,7 @@ rules:
   - apiGroups: [""]
     resources: ["events"]
     verbs: ["create", "patch"]
-  {{- if not .Values.csi.useNodeAuthorization }}
+  {{- if and (not .Values.csi.useNodeAuthorization) (semverCompare ">=1.30.0-0" .Capabilities.KubeVersion.Version) }}
   - apiGroups: [""]
     resources: ["nodes"]
     verbs: ["get", "patch"]

charts/fluid/fluid/values.yaml

@@ -61,6 +61,8 @@ csi:
   # Whether or not to borrow kubelet's config file to use node authorization to restrict CSI Plugin's permission
   # See why Fluid's CSI Plugins need node-specific authorization at https://github.com/fluid-cloudnative/fluid/security/advisories/GHSA-93xx-cvmc-9w3v
   # See node authorization at https://kubernetes.io/docs/reference/access-authn-authz/node/
+  # If false, use NodeBinding Token with ValidatingAdmissionPolicy instead of kubelet config for Node-Specific Restrictions.
+  # can only be set false when k8s.version >= 1.30 and the below kubelet.kubeConfigFile is useless.
   useNodeAuthorization: true
   kubelet:
     kubeConfigFile: /etc/kubernetes/kubelet.conf

pkg/csi/plugins/driver.go

@@ -23,7 +23,6 @@ import (
 	"path/filepath"
 	"strings"
 
-	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 
@@ -41,7 +40,7 @@ const (
 type driver struct {
 	client               client.Client
 	apiReader            client.Reader
-	nodeAuthorizedClient *kubernetes.Clientset
+	nodeAuthorizedClient NodeAuthorizedClient
 	csiDriver            *csicommon.CSIDriver
 	nodeId, endpoint     string
 
@@ -50,7 +49,7 @@ type driver struct {
 
 var _ manager.Runnable = &driver{}
 
-func NewDriver(nodeID, endpoint string, client client.Client, apiReader client.Reader, nodeAuthorizedClient *kubernetes.Clientset, locks *utils.VolumeLocks) *driver {
+func NewDriver(nodeID, endpoint string, client client.Client, apiReader client.Reader, nodeAuthorizedClient NodeAuthorizedClient, locks *utils.VolumeLocks) *driver {
 	glog.Infof("Driver: %v version: %v", driverName, version)
 
 	proto, addr := utils.SplitSchemaAddr(endpoint)

pkg/csi/plugins/node_resource_operator.go

@@ -0,0 +1,64 @@
+/*
+Copyright 2025 The Fluid 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.
+*/
+
+package plugins
+
+import (
+	"context"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type NodeAuthorizedClient interface {
+	Get(nodeName string) (*corev1.Node, error)
+	Patch(node *corev1.Node, patchType types.PatchType, data []byte) error
+}
+
+// restrictedNodeClient uses node binding token with validating policy to avoid security problems.
+type restrictedNodeClient struct {
+	Client client.Client
+}
+
+// kubeletNodeClient uses mounted kubelet config to avoid security problems.
+type kubeletNodeClient struct {
+	Clientset *kubernetes.Clientset
+}
+
+func (p *restrictedNodeClient) Get(nodeName string) (*corev1.Node, error) {
+	node := &corev1.Node{}
+	key := types.NamespacedName{Name: nodeName}
+	if err := p.Client.Get(context.TODO(), key, node); err != nil {
+		return nil, err
+	}
+	return node, nil
+}
+
+func (p *restrictedNodeClient) Patch(node *corev1.Node, patchType types.PatchType, data []byte) error {
+	err := p.Client.Patch(context.TODO(), node, client.RawPatch(patchType, data))
+	return err
+}
+
+func (p *kubeletNodeClient) Get(nodeName string) (*corev1.Node, error) {
+	return p.Clientset.CoreV1().Nodes().Get(context.TODO(), nodeName, metav1.GetOptions{})
+}
+
+func (p *kubeletNodeClient) Patch(node *corev1.Node, patchType types.PatchType, data []byte) error {
+	_, err := p.Clientset.CoreV1().Nodes().Patch(context.TODO(), node.Name, patchType, data, metav1.PatchOptions{})
+	return err
+}

pkg/csi/plugins/nodeserver.go

@@ -32,13 +32,10 @@ import (
 	"github.com/fluid-cloudnative/fluid/pkg/utils"
 	"github.com/fluid-cloudnative/fluid/pkg/utils/cmdguard"
 	"github.com/fluid-cloudnative/fluid/pkg/utils/dataset/volume"
-	"github.com/fluid-cloudnative/fluid/pkg/utils/kubeclient"
 	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/client-go/kubernetes"
 	"k8s.io/utils/mount"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -59,7 +56,7 @@ type nodeServer struct {
 	*csicommon.DefaultNodeServer
 	client               client.Client
 	apiReader            client.Reader
-	nodeAuthorizedClient *kubernetes.Clientset
+	nodeAuthorizedClient NodeAuthorizedClient
 	locks                *utils.VolumeLocks
 	node                 *corev1.Node
 }
@@ -474,15 +471,8 @@ func (ns *nodeServer) getNode() (node *corev1.Node, err error) {
 		}
 	}
 
-	useNodeAuthorization := ns.nodeAuthorizedClient != nil
-	if useNodeAuthorization {
-		if node, err = ns.nodeAuthorizedClient.CoreV1().Nodes().Get(context.TODO(), ns.nodeId, metav1.GetOptions{}); err != nil {
-			return nil, err
-		}
-	} else {
-		if node, err = kubeclient.GetNode(ns.apiReader, ns.nodeId); err != nil {
-			return nil, err
-		}
+	if node, err = ns.nodeAuthorizedClient.Get(ns.nodeId); err != nil {
+		return nil, err
 	}
 
 	glog.V(1).Infof("Got node %s from api server", node.Name)
@@ -520,22 +510,10 @@ func (ns *nodeServer) patchNodeWithLabel(node *corev1.Node, labelsToModify commo
 	if err != nil {
 		return err
 	}
-	useNodeAuthorization := ns.nodeAuthorizedClient != nil
-	if useNodeAuthorization {
-		_, err = ns.nodeAuthorizedClient.CoreV1().Nodes().Patch(context.TODO(), node.Name, types.StrategicMergePatchType, patchByteData, metav1.PatchOptions{})
-		if err != nil {
-			return err
-		}
-	} else {
-		nodeToPatch := &corev1.Node{
-			ObjectMeta: metav1.ObjectMeta{
-				Name: node.Name,
-			},
-		}
-		err = ns.client.Patch(context.TODO(), nodeToPatch, client.RawPatch(types.StrategicMergePatchType, patchByteData))
-		if err != nil {
-			return err
-		}
+
+	err = ns.nodeAuthorizedClient.Patch(node, types.StrategicMergePatchType, patchByteData)
+	if err != nil {
+		return err
 	}
 
 	return nil

pkg/csi/plugins/register.go

@@ -20,37 +20,35 @@ import (
 	"os"
 
 	"github.com/fluid-cloudnative/fluid/pkg/csi/config"
+	"github.com/fluid-cloudnative/fluid/pkg/utils/compatibility"
 	"github.com/fluid-cloudnative/fluid/pkg/utils/kubelet"
 	"github.com/golang/glog"
-	"github.com/pkg/errors"
-	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 )
 
-// getNodeAuthorizedClientFromKubeletConfig retrieves a node-authorized Kubernetes client from the Kubelet configuration file.
-// This function checks if the specified Kubelet configuration file exists. If the file does not exist, it returns an empty client without an error .
+// isUseKubeletConfig checks if the specified Kubelet configuration file exists. If the file does not exist, it returns an empty client without an error .
 // If the file exists, it attempts to initialize and return a node-authorized Kubernetes client.
-func getNodeAuthorizedClientFromKubeletConfig(kubeletConfigPath string) (*kubernetes.Clientset, error) {
+func isUseKubeletConfig(kubeletConfigPath string) bool {
 	_, err := os.Stat(kubeletConfigPath)
 	if err != nil {
 		if os.IsNotExist(err) {
 			glog.Warningf("kubelet config file %s not exists, continue without node authorization...", kubeletConfigPath)
-			return nil, nil
+			return false
 		}
-		return nil, errors.Wrapf(err, "fail to stat kubelet config file %s", kubeletConfigPath)
+		glog.Warningf("fail to stat kubelet config file %s", kubeletConfigPath)
 	}
 
-	return kubelet.InitNodeAuthorizedClient(kubeletConfigPath)
+	return true
 }
 
 // Register initializes the csi driver and registers it to the controller manager.
 func Register(mgr manager.Manager, ctx config.RunningContext) error {
-	client, err := getNodeAuthorizedClientFromKubeletConfig(ctx.KubeletConfigPath)
+	nodeAuthClient, err := getNodeAuthClient(mgr, ctx)
 	if err != nil {
 		return err
 	}
 
-	csiDriver := NewDriver(ctx.NodeId, ctx.Endpoint, mgr.GetClient(), mgr.GetAPIReader(), client, ctx.VolumeLocks)
+	csiDriver := NewDriver(ctx.NodeId, ctx.Endpoint, mgr.GetClient(), mgr.GetAPIReader(), nodeAuthClient, ctx.VolumeLocks)
 
 	if err := mgr.Add(csiDriver); err != nil {
 		return err
@@ -59,6 +57,19 @@ func Register(mgr manager.Manager, ctx config.RunningContext) error {
 	return nil
 }
 
+func getNodeAuthClient(mgr manager.Manager, ctx config.RunningContext) (NodeAuthorizedClient, error) {
+	// use and support node binding token
+	if !isUseKubeletConfig(ctx.KubeletConfigPath) && compatibility.IsNodeBindingTokenSupported() {
+		return &restrictedNodeClient{mgr.GetClient()}, nil
+	}
+	// otherwise, use kubelet config
+	nodeAuthClient, err := kubelet.InitNodeAuthorizedClient(ctx.KubeletConfigPath)
+	if err != nil {
+		return nil, err
+	}
+	return &kubeletNodeClient{nodeAuthClient}, nil
+}
+
 // Enabled checks if the csi driver should be enabled.
 func Enabled() bool {
 	return true

pkg/utils/compatibility/node_restrict.go

@@ -0,0 +1,68 @@
+/*
+Copyright 2025 The Fluid 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.
+*/
+
+package compatibility
+
+import (
+	"github.com/blang/semver/v4"
+
+	nativeLog "log"
+	"sync"
+
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/client-go/discovery"
+	ctrl "sigs.k8s.io/controller-runtime"
+)
+
+var (
+	nodeBindingTokenSupported = false
+	nodeBindingTokenOnce      sync.Once
+)
+
+// Beta release, default enabled. see https://github.com/kubernetes/enhancements/issues/4193
+const nodeBindingTokenSupportedVersion = "v1.30.0"
+
+// Checks the ServiceAccountTokenPodNodeInfo feature gate, whether the apiserver embeds the node name for the associated node when issuing service account tokens bound to Pod objects.
+func discoverNodeBindingTokenCompatibility() {
+	nativeLog.Printf("Discovering k8s version to check NodeBindingToken compatibility...")
+	restConfig := ctrl.GetConfigOrDie()
+	discoveryClient := discovery.NewDiscoveryClientForConfigOrDie(restConfig)
+
+	serverVersion, err := discoveryClient.ServerVersion()
+	if err != nil && !errors.IsNotFound(err) {
+		nativeLog.Fatalf("failed to discover Kubernetes server version: %v", err)
+	}
+	// transform to semver.Version and compare
+	currentVersion, err := semver.ParseTolerant(serverVersion.GitVersion)
+	if err != nil {
+		nativeLog.Fatalf("Failed to parse current version: %v", err)
+	}
+	targetVersion, err := semver.ParseTolerant(nodeBindingTokenSupportedVersion)
+	if err != nil {
+		nativeLog.Fatalf("Failed to parse target version: %v", err)
+	}
+
+	if currentVersion.GTE(targetVersion) {
+		nodeBindingTokenSupported = true
+	}
+}
+
+func IsNodeBindingTokenSupported() bool {
+	nodeBindingTokenOnce.Do(func() {
+		discoverNodeBindingTokenCompatibility()
+	})
+	return nodeBindingTokenSupported
+}