[helm] Enable SASL authenticated connection to Zookeeper nodes

helm/templates/_security.tpl

@@ -29,6 +29,28 @@ Usage:
 {{- $mechanism -}}
 {{- end -}}
 
+{{/*
+Returns the ZooKeeper SASL authentication mechanism value.
+Allowed mechanism values: '', 'plain'
+Usage:
+  include "fluss.security.zookeeper.sasl.mechanism" .
+*/}}
+{{- define "fluss.security.zookeeper.sasl.mechanism" -}}
+{{- $sasl := .Values.security.zookeeper.sasl | default (dict) -}}
+{{- $mechanism := lower (default "" $sasl.mechanism) -}}
+{{- $mechanism -}}
+{{- end -}}
+
+{{/*
+Returns true if ZooKeeper SASL authentication is enabled (mechanism is non-empty).
+Usage:
+  include "fluss.security.zookeeper.sasl.enabled" .
+*/}}
+{{- define "fluss.security.zookeeper.sasl.enabled" -}}
+{{- $mechanism := include "fluss.security.zookeeper.sasl.mechanism" . -}}
+{{- if ne $mechanism "" -}}true{{- end -}}
+{{- end -}}
+
 {{/*
 Returns true if any of the listeners uses SASL based authentication mechanism ('plain' for now).
 Usage:
@@ -117,6 +139,56 @@ Usage:
 {{- end -}}
 {{- end -}}
 
+{{/*
+Validates that ZooKeeper SASL mechanism is valid.
+Returns an error message if invalid, empty string otherwise.
+Usage:
+  include "fluss.security.zookeeper.sasl.validateMechanism" .
+*/}}
+{{- define "fluss.security.zookeeper.sasl.validateMechanism" -}}
+{{- $allowedMechanisms := list "" "plain" -}}
+{{- $mechanism := include "fluss.security.zookeeper.sasl.mechanism" . -}}
+{{- if not (has $mechanism $allowedMechanisms) -}}
+  {{- print "security.zookeeper.sasl.mechanism must be empty or: plain" -}}
+{{- end -}}
+{{- end -}}
+
+{{/*
+Validates that ZooKeeper SASL loginModuleClass is not empty when ZK SASL is enabled.
+Returns an error message if invalid, empty string otherwise.
+Usage:
+  include "fluss.security.zookeeper.sasl.validateLoginModuleClass" .
+*/}}
+{{- define "fluss.security.zookeeper.sasl.validateLoginModuleClass" -}}
+{{- if and (include "fluss.security.zookeeper.sasl.enabled" .) (not .Values.security.zookeeper.sasl.plain.loginModuleClass) -}}
+  {{- print "security.zookeeper.sasl.plain.loginModuleClass must not be empty when security.zookeeper.sasl.mechanism is plain" -}}
+{{- end -}}
+{{- end -}}
+
+{{/*
+Validates that ZooKeeper SASL username is not empty when ZK SASL is enabled.
+Returns an error message if invalid, empty string otherwise.
+Usage:
+  include "fluss.security.zookeeper.sasl.validateUsername" .
+*/}}
+{{- define "fluss.security.zookeeper.sasl.validateUsername" -}}
+{{- if and (include "fluss.security.zookeeper.sasl.enabled" .) (not .Values.security.zookeeper.sasl.plain.username) -}}
+  {{- print "security.zookeeper.sasl.plain.username must not be empty when security.zookeeper.sasl.mechanism is plain" -}}
+{{- end -}}
+{{- end -}}
+
+{{/*
+Validates that ZooKeeper SASL password is not empty when ZK SASL is enabled.
+Returns an error message if invalid, empty string otherwise.
+Usage:
+  include "fluss.security.zookeeper.sasl.validatePassword" .
+*/}}
+{{- define "fluss.security.zookeeper.sasl.validatePassword" -}}
+{{- if and (include "fluss.security.zookeeper.sasl.enabled" .) (not .Values.security.zookeeper.sasl.plain.password) -}}
+  {{- print "security.zookeeper.sasl.plain.password must not be empty when security.zookeeper.sasl.mechanism is plain" -}}
+{{- end -}}
+{{- end -}}
+
 {{/*
 Returns the default internal SASL username based on the release name.
 Usage:
@@ -153,6 +225,17 @@ Usage:
 {{- .Values.security.internal.sasl.plain.password | default (include "fluss.security.sasl.plain.internal.defaultPassword" .) -}}
 {{- end -}}
 
+{{/*
+Returns true if JAAS configuration is required, either by listeners using SASL protocol or ZooKeeper SASL enablement.
+Usage:
+  include "fluss.security.jaas.required" .
+*/}}
+{{- define "fluss.security.jaas.required" -}}
+{{- if or (include "fluss.security.sasl.enabled" .) (include "fluss.security.zookeeper.sasl.enabled" .) -}}
+{{- true -}}
+{{- end -}}
+{{- end -}}
+
 {{/*
 Returns a warning if the internal SASL user is using auto-generated credentials.
 Usage:
@@ -179,6 +262,10 @@ Usage:
 {{- $errMessages := list -}}
 {{- $errMessages = append $errMessages (include "fluss.security.sasl.validateMechanisms" .) -}}
 {{- $errMessages = append $errMessages (include "fluss.security.sasl.validateClientPlainUsers" .) -}}
+{{- $errMessages = append $errMessages (include "fluss.security.zookeeper.sasl.validateMechanism" .) -}}
+{{- $errMessages = append $errMessages (include "fluss.security.zookeeper.sasl.validateLoginModuleClass" .) -}}
+{{- $errMessages = append $errMessages (include "fluss.security.zookeeper.sasl.validateUsername" .) -}}
+{{- $errMessages = append $errMessages (include "fluss.security.zookeeper.sasl.validatePassword" .) -}}
 
 {{- $errMessages = without $errMessages "" -}}
 {{- $errMessage := join "\n" $errMessages -}}
@@ -202,8 +289,8 @@ Usage:
 {{/*
 Returns the SASL JAAS config name.
 Usage:
-  include "fluss.security.sasl.configName" .
+  include "fluss.security.jaas.configName" .
 */}}
-{{- define "fluss.security.sasl.configName" -}}
+{{- define "fluss.security.jaas.configName" -}}
 {{ include "fluss.fullname" . }}-sasl-jaas-config
 {{- end -}}

helm/templates/configmap.yaml

@@ -53,3 +53,9 @@ data:
     {{- end }}
 
     {{- end }}
+
+    ### Zookeeper
+
+    {{- if (include "fluss.security.zookeeper.sasl.enabled" .) }}
+    zookeeper.client.config.path: /etc/fluss/conf/zookeeper-client.properties
+    {{- end }}

helm/templates/secret-jaas-config.yaml

@@ -16,20 +16,21 @@
 # limitations under the License.
 #
 
-{{ if (include "fluss.security.sasl.plain.enabled" .) }}
+{{ if (include "fluss.security.jaas.required" .) }}
 {{- $internalMechanism := include "fluss.security.listener.mechanism" (dict "context" .Values "listener" "internal") -}}
 {{- $clientMechanism := include "fluss.security.listener.mechanism" (dict "context" .Values "listener" "client") -}}
 {{- $internalUsername := include "fluss.security.sasl.plain.internal.username" . -}}
 {{- $internalPassword := include "fluss.security.sasl.plain.internal.password" . -}}
 apiVersion: v1
 kind: Secret
 metadata:
-  name: {{ include "fluss.security.sasl.configName" . }}
+  name: {{ include "fluss.security.jaas.configName" . }}
   labels:
     {{- include "fluss.labels" . | nindent 4 }}
 type: Opaque
 stringData:
   jaas.conf: |
+{{- if (include "fluss.security.sasl.plain.enabled" .) }}
 {{- if eq $internalMechanism "plain" }}
     internal.FlussServer {
        org.apache.fluss.security.auth.sasl.plain.PlainLoginModule required
@@ -49,4 +50,16 @@ stringData:
 {{- end }};
     };
 {{- end }}
+{{- end }}
+{{- if (include "fluss.security.zookeeper.sasl.enabled" .) }}
+    ZookeeperClient {
+       {{ .Values.security.zookeeper.sasl.plain.loginModuleClass }} required
+       username="{{ .Values.security.zookeeper.sasl.plain.username }}"
+       password="{{ .Values.security.zookeeper.sasl.plain.password }}";
+    };
+{{- end }}
+{{- if (include "fluss.security.zookeeper.sasl.enabled" .) }}
+  zookeeper-client.properties: |
+    zookeeper.sasl.clientconfig=ZookeeperClient
+{{- end }}
 {{- end -}}

helm/templates/sts-coordinator.yaml

@@ -60,7 +60,7 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: status.hostIP
-            {{- if (include "fluss.security.sasl.plain.enabled" .) }}
+            {{- if (include "fluss.security.jaas.required" .) }}
             - name: FLUSS_ENV_JAVA_OPTS
               value: "-Djava.security.auth.login.config=/etc/fluss/conf/jaas.conf"
             {{- end }}
@@ -105,7 +105,7 @@ spec:
               mountPath: /opt/conf
             - name: data
               mountPath: /tmp/fluss/data
-            {{- if (include "fluss.security.sasl.plain.enabled" .) }}
+            {{- if (include "fluss.security.jaas.required" .) }}
             - name: sasl-config
               mountPath: /etc/fluss/conf
               readOnly: true
@@ -118,10 +118,10 @@ spec:
         - name: data
           emptyDir: {}
         {{- end }}
-        {{- if (include "fluss.security.sasl.plain.enabled" .) }}
+        {{- if (include "fluss.security.jaas.required" .) }}
         - name: sasl-config
           secret:
-            secretName: {{ include "fluss.security.sasl.configName" . }}
+            secretName: {{ include "fluss.security.jaas.configName" . }}
         {{- end }}
   {{- if .Values.coordinator.storage.enabled }}
   volumeClaimTemplates:

helm/templates/sts-tablet.yaml

@@ -56,7 +56,7 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: metadata.namespace
-            {{- if (include "fluss.security.sasl.plain.enabled" .) }}
+            {{- if (include "fluss.security.jaas.required" .) }}
             - name: FLUSS_ENV_JAVA_OPTS
               value: "-Djava.security.auth.login.config=/etc/fluss/conf/jaas.conf"
             {{- end }}
@@ -102,7 +102,7 @@ spec:
               mountPath: /opt/conf
             - name: data
               mountPath: /tmp/fluss/data
-            {{- if (include "fluss.security.sasl.plain.enabled" .) }}
+            {{- if (include "fluss.security.jaas.required" .) }}
             - name: sasl-config
               mountPath: /etc/fluss/conf
               readOnly: true
@@ -115,10 +115,10 @@ spec:
         - name: data
           emptyDir: {}
         {{- end }}
-        {{- if (include "fluss.security.sasl.plain.enabled" .) }}
+        {{- if (include "fluss.security.jaas.required" .) }}
         - name: sasl-config
           secret:
-            secretName: {{ include "fluss.security.sasl.configName" . }}
+            secretName: {{ include "fluss.security.jaas.configName" . }}
         {{- end }}
   {{- if .Values.tablet.storage.enabled }}
   volumeClaimTemplates: