[Improvement][Registry] Add ZK authorization yaml control & add diges…

…t auth UT

dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/resources/application.yaml

@@ -88,7 +88,9 @@ registry:
     session-timeout: 60s
     connection-timeout: 15s
     block-until-connected: 15s
-    digest: ~
+    authorization:
+      digest: ~
+
 
 metrics:
   enabled: true

dolphinscheduler-api/src/main/resources/application.yaml

@@ -126,7 +126,8 @@ registry:
     session-timeout: 60s
     connection-timeout: 15s
     block-until-connected: 15s
-    digest: ~
+    authorization:
+      digest: ~
 
 api:
   audit-enable: false

dolphinscheduler-master/src/main/resources/application.yaml

@@ -80,7 +80,8 @@ registry:
     session-timeout: 60s
     connection-timeout: 15s
     block-until-connected: 15s
-    digest: ~
+    authorization:
+      digest: ~
 
 master:
   listen-port: 5678

dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-zookeeper/README.md

@@ -20,7 +20,8 @@ registry:
     connection-timeout: 9s
     block-until-connected: 600ms
     # The following options are set according to personal needs    
-    digest: ~
+    authorization:
+      digest: ~
 ```
 
 After do this config, you can start your DolphinScheduler cluster, your cluster will use zookeeper as registry center to

dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-zookeeper/src/main/java/org/apache/dolphinscheduler/plugin/registry/zookeeper/ZookeeperRegistry.java

@@ -80,7 +80,7 @@ final class ZookeeperRegistry implements Registry {
                         .sessionTimeoutMs(DurationUtils.toMillisInt(properties.getSessionTimeout()))
                         .connectionTimeoutMs(DurationUtils.toMillisInt(properties.getConnectionTimeout()));
 
-        final String digest = properties.getDigest();
+        final String digest = properties.getAuthorization().getDigest();
         if (!Strings.isNullOrEmpty(digest)) {
             builder.authorization("digest", digest.getBytes(StandardCharsets.UTF_8))
                     .aclProvider(new ACLProvider() {
@@ -96,6 +96,7 @@ public List<ACL> getAclForPath(final String path) {
                         }
                     });
         }
+
         client = builder.build();
     }
 

dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-zookeeper/src/main/java/org/apache/dolphinscheduler/plugin/registry/zookeeper/ZookeeperRegistryProperties.java

@@ -17,6 +17,7 @@
 
 package org.apache.dolphinscheduler.plugin.registry.zookeeper;
 
+import org.apache.commons.codec.cli.Digest;
 import org.apache.commons.lang3.StringUtils;
 
 import java.time.Duration;
@@ -86,7 +87,7 @@ private void printConfig() {
                         "\n  namespace -> " + zookeeper.getNamespace() +
                         "\n  connectString -> " + zookeeper.getConnectString() +
                         "\n  retryPolicy -> " + zookeeper.getRetryPolicy() +
-                        "\n  digest -> " + zookeeper.getDigest() +
+                        "\n  authorization -> " + zookeeper.getAuthorization() +
                         "\n  sessionTimeout -> " + zookeeper.getSessionTimeout() +
                         "\n  connectionTimeout -> " + zookeeper.getConnectionTimeout() +
                         "\n  blockUntilConnected -> " + zookeeper.getBlockUntilConnected() +
@@ -100,11 +101,17 @@ public static final class ZookeeperProperties {
         private String namespace = "dolphinscheduler";
         private String connectString;
         private RetryPolicy retryPolicy = new RetryPolicy();
-        private String digest;
+        private Authorization authorization = new Authorization();
         private Duration sessionTimeout = Duration.ofSeconds(60);
         private Duration connectionTimeout = Duration.ofSeconds(15);
         private Duration blockUntilConnected = Duration.ofSeconds(15);
 
+        @Data
+        public static final class Authorization {
+            private String digest;
+            private String x509SubjectPrincipal;
+        }
+
         @Data
         public static final class RetryPolicy {
 

dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-zookeeper/src/test/java/org/apache/dolphinscheduler/plugin/registry/zookeeper/ZookeeperRegistryDigestTestCase.java

@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dolphinscheduler.plugin.registry.zookeeper;
+
+import org.apache.dolphinscheduler.plugin.registry.RegistryTestCase;
+
+import java.util.Collections;
+import java.util.stream.Stream;
+
+import lombok.SneakyThrows;
+
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.server.DumbWatcher;
+import org.apache.zookeeper.server.admin.Commands;
+import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.utility.DockerImageName;
+
+@SpringBootTest(classes = ZookeeperRegistryProperties.class)
+@SpringBootApplication(scanBasePackageClasses = ZookeeperRegistryProperties.class)
+public class ZookeeperRegistryDigestTestCase extends RegistryTestCase<ZookeeperRegistry> {
+
+    @Autowired
+    private ZookeeperRegistryProperties zookeeperRegistryProperties;
+
+    private static GenericContainer<?> zookeeperContainer;
+
+    private static final Network NETWORK = Network.newNetwork();
+
+    private static ZooKeeper zk;
+
+    private static final String ROOT_USER = "root";
+
+    private static final String ROOT_PASSWORD = "root_passwd";
+
+    private static final String ID_PASSWORD = String.format("%s:%s", ROOT_USER, ROOT_PASSWORD);
+
+    public static void setupRootACLForDigest(final ZooKeeper zk) throws Exception  {
+        final String digest = DigestAuthenticationProvider.generateDigest(ID_PASSWORD);
+        final ACL acl = new ACL(ZooDefs.Perms.ALL, new Id("digest", digest));
+        zk.setACL("/", Collections.singletonList(acl), -1);
+    }
+
+    @SneakyThrows
+    @BeforeAll
+    public static void setUpTestingServer() {
+        zookeeperContainer = new GenericContainer<>(DockerImageName.parse("zookeeper:3.8"))
+                .withNetwork(NETWORK)
+                .withExposedPorts(2181);
+        Startables.deepStart(Stream.of(zookeeperContainer)).join();
+        System.clearProperty("registry.zookeeper.connect-string");
+        System.setProperty("registry.zookeeper.connect-string", "localhost:" + zookeeperContainer.getMappedPort(2181));
+        zk = new ZooKeeper("localhost:" + zookeeperContainer.getMappedPort(2181),
+                30000, new DumbWatcher(), new ZKClientConfig());
+        System.setProperty("registry.zookeeper.authorization.digest", ID_PASSWORD);
+        setupRootACLForDigest(zk);
+    }
+
+    @SneakyThrows
+    @Override
+    public ZookeeperRegistry createRegistry() {
+        return new ZookeeperRegistry(zookeeperRegistryProperties);
+    }
+
+    @SneakyThrows
+    @AfterAll
+    public static void tearDownTestingServer() {
+        zk.close();
+        zookeeperContainer.close();
+    }
+}
+

dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-zookeeper/src/test/resources/application.yaml

@@ -27,4 +27,5 @@ registry:
     session-timeout: 30s
     connection-timeout: 9s
     block-until-connected: 3s
-    digest: ~
+    authorization:
+      digest: ~

dolphinscheduler-worker/src/main/resources/application.yaml

@@ -37,7 +37,8 @@ registry:
     session-timeout: 60s
     connection-timeout: 15s
     block-until-connected: 15s
-    digest: ~
+    authorization:
+      digest: ~
 
 worker:
   # worker listener port