support ssl for graph

connector/src/main/java/org.apache.flink/connector/nebula/connection/NebulaClientOptions.java

@@ -6,11 +6,14 @@
 
 package org.apache.flink.connector.nebula.connection;
 
+import com.vesoft.nebula.client.graph.data.CASignedSSLParam;
 import com.vesoft.nebula.client.graph.data.HostAddress;
+import com.vesoft.nebula.client.graph.data.SelfSignedSSLParam;
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.List;
 import org.apache.flink.connector.nebula.utils.NebulaConstant;
+import org.apache.flink.connector.nebula.utils.SslSighType;
 
 public class NebulaClientOptions implements Serializable {
 
@@ -28,14 +31,33 @@ public class NebulaClientOptions implements Serializable {
 
     private final int connectRetry;
 
+    private final boolean enableGraphSsl;
+
+    private final boolean enableMetaSsl;
+
+    private final SslSighType sslSighType;
+
+    private final CASignedSSLParam caSignParam;
+
+    private final SelfSignedSSLParam selfSignParam;
+
+
     private NebulaClientOptions(String metaAddress, String graphAddress, String username,
-                                String password, int timeout, int connectRetry) {
+                                String password, int timeout, int connectRetry,
+                                boolean enableGraphSsl, boolean enableMetaSsl,
+                                SslSighType sslSighType, CASignedSSLParam caSignParam,
+                                SelfSignedSSLParam selfSignParam) {
         this.metaAddress = metaAddress;
         this.graphAddress = graphAddress;
         this.username = username;
         this.password = password;
         this.timeout = timeout;
         this.connectRetry = connectRetry;
+        this.enableGraphSsl = enableGraphSsl;
+        this.enableMetaSsl = enableMetaSsl;
+        this.sslSighType = sslSighType;
+        this.caSignParam = caSignParam;
+        this.selfSignParam = selfSignParam;
     }
 
     public List<HostAddress> getMetaAddress() {
@@ -67,6 +89,26 @@ public int getConnectRetry() {
         return connectRetry;
     }
 
+    public boolean isEnableGraphSsl() {
+        return enableGraphSsl;
+    }
+
+    public boolean isEnableMetaSsl() {
+        return enableMetaSsl;
+    }
+
+    public SslSighType getSslSighType() {
+        return sslSighType;
+    }
+
+    public CASignedSSLParam getCaSignParam() {
+        return caSignParam;
+    }
+
+    public SelfSignedSSLParam getSelfSignParam() {
+        return selfSignParam;
+    }
+
     /**
      * Builder for {@link NebulaClientOptions}
      */
@@ -78,6 +120,13 @@ public static class NebulaClientOptionsBuilder {
         private int timeout = 6000;
         private int connectRetry = 1;
 
+        // ssl options
+        private boolean enableGraphSsl = false;
+        private boolean enableMetaSsl = false;
+        private SslSighType sslSighType = null;
+        private CASignedSSLParam caSignParam = null;
+        private SelfSignedSSLParam selfSignParam = null;
+
         public NebulaClientOptionsBuilder setMetaAddress(String metaAddress) {
             this.metaAddress = metaAddress;
             return this;
@@ -108,18 +157,76 @@ public NebulaClientOptionsBuilder setConnectRetry(int connectRetry) {
             return this;
         }
 
+        public NebulaClientOptionsBuilder setEnableGraphSsl(boolean enableGraphSsl) {
+            this.enableGraphSsl = enableGraphSsl;
+            return this;
+        }
+
+        public NebulaClientOptionsBuilder setEnableMetaSsl(boolean enableMetaSsl) {
+            this.enableMetaSsl = enableMetaSsl;
+            return this;
+        }
+
+        public NebulaClientOptionsBuilder setSslSignType(SslSighType sslSighType) {
+            this.sslSighType = sslSighType;
+            return this;
+        }
+
+        public NebulaClientOptionsBuilder setCaSignParam(String caCrtFilePath, String crtFilePath,
+                                                         String keyFilePath) {
+            this.caSignParam = new CASignedSSLParam(caCrtFilePath, crtFilePath,
+                    keyFilePath);
+            return this;
+        }
+
+        public NebulaClientOptionsBuilder setSelfSignParam(String crtFilePath, String keyFilePath,
+                                                           String password) {
+            this.selfSignParam = new SelfSignedSSLParam(crtFilePath, keyFilePath, password);
+            return this;
+        }
+
         public NebulaClientOptions build() {
             if (metaAddress == null || metaAddress.trim().isEmpty()) {
                 throw new IllegalArgumentException("meta address can not be empty.");
             }
+            if (enableMetaSsl || enableGraphSsl) {
+                // if meta is set to open ssl, then graph must be set to open ssl
+                enableGraphSsl = true;
+                if (sslSighType == null) {
+                    throw new IllegalArgumentException("ssl is enable, ssl sign type must not be "
+                            + "null");
+                }
+                switch (sslSighType) {
+                    case CA:
+                        if (caSignParam == null) {
+                            throw new IllegalArgumentException("ssl is enable and sign type is "
+                                    + "CA, caSignParam must not be null");
+                        }
+                        break;
+                    case SELF:
+                        if (selfSignParam == null) {
+                            throw new IllegalArgumentException("ssl is enable and sign type is "
+                                    + "CA, selfSignParam must not be null");
+                        }
+                        break;
+                    default:
+                        throw new IllegalArgumentException("ssl sign type is not supported.");
+                }
+
+            }
 
             return new NebulaClientOptions(
                     metaAddress,
                     graphAddress,
                     username,
                     password,
                     timeout,
-                    connectRetry);
+                    connectRetry,
+                    enableGraphSsl,
+                    enableMetaSsl,
+                    sslSighType,
+                    caSignParam,
+                    selfSignParam);
         }
     }
 }

connector/src/main/java/org.apache.flink/connector/nebula/connection/NebulaGraphConnectionProvider.java

@@ -47,7 +47,22 @@ public Session getSession() throws NotValidConnectionException, IOErrorException
         }
 
         try {
-            nebulaPool.init(addresses, new NebulaPoolConfig());
+            NebulaPoolConfig poolConfig = new NebulaPoolConfig();
+            poolConfig.setTimeout(nebulaClientOptions.getTimeout());
+            if (nebulaClientOptions.isEnableGraphSsl()) {
+                poolConfig.setEnableSsl(true);
+                switch (nebulaClientOptions.getSslSighType()) {
+                    case CA:
+                        poolConfig.setSslParam(nebulaClientOptions.getCaSignParam());
+                        break;
+                    case SELF:
+                        poolConfig.setSslParam(nebulaClientOptions.getSelfSignParam());
+                        break;
+                    default:
+                        throw new IllegalArgumentException("ssl sign type is not supported.");
+                }
+            }
+            nebulaPool.init(addresses, poolConfig);
         } catch (UnknownHostException e) {
             LOG.error("NebulaPool init error, ", e);
         }

connector/src/main/java/org.apache.flink/connector/nebula/utils/SslSighType.java

@@ -0,0 +1,25 @@
+/* Copyright (c) 2021 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+package org.apache.flink.connector.nebula.utils;
+
+public enum SslSighType {
+    /**
+     * CA sign
+     */
+    CA("ca"),
+
+    /**
+     * SELF sign
+     */
+    SELF("self");
+
+    private String type;
+
+    SslSighType(String type) {
+        this.type = type;
+    }
+}

connector/src/test/java/org/apache/flink/connector/nebula/connection/NebulaClientOptionsTest.java

@@ -0,0 +1,114 @@
+/* Copyright (c) 2021 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+package org.apache.flink.connector.nebula.connection;
+
+import org.apache.flink.connector.nebula.utils.SslSighType;
+import org.junit.Test;
+
+public class NebulaClientOptionsTest {
+    @Test
+    public void testConfigAddress() {
+        NebulaClientOptions nebulaClientOptions =
+                new NebulaClientOptions.NebulaClientOptionsBuilder()
+                        .setGraphAddress("127.0.0.1:9669")
+                        .setMetaAddress("127.0.0.1:9559")
+                        .setUsername("root")
+                        .setPassword("nebula")
+                        .setConnectRetry(1)
+                        .setTimeout(1000)
+                        .setEnableGraphSsl(true)
+                        .setEnableMetaSsl(true)
+                        .setSslSignType(SslSighType.CA)
+                        .setCaSignParam("caCrtFile", "crtFile", "keyFile")
+                        .setSelfSignParam("crtFile", "keyFile", "password")
+                        .build();
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testMetaAddressWithEmptyAddress() {
+        new NebulaClientOptions.NebulaClientOptionsBuilder()
+                .setGraphAddress("127.0.0.1:9669")
+                .setMetaAddress(null)
+                .build();
+    }
+
+    @Test
+    public void testGraphAddressWithEmptyAddress() {
+        new NebulaClientOptions.NebulaClientOptionsBuilder()
+                .setGraphAddress(null)
+                .setMetaAddress("127.0.0.1:9559")
+                .build();
+    }
+
+
+    @Test
+    public void testIsEnableGraphSsl() {
+        NebulaClientOptions nebulaClientOptions = new NebulaClientOptions
+                .NebulaClientOptionsBuilder()
+                .setGraphAddress(null)
+                .setMetaAddress("127.0.0.1:9559")
+                .setEnableGraphSsl(false)
+                .setEnableMetaSsl(true)
+                .setSslSignType(SslSighType.CA)
+                .setCaSignParam("caCrtFile", "crtFile", "keyFile")
+                .build();
+        assert (nebulaClientOptions.isEnableGraphSsl());
+    }
+
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testNullSslSighType() {
+        NebulaClientOptions nebulaClientOptions = new NebulaClientOptions
+                .NebulaClientOptionsBuilder()
+                .setGraphAddress(null)
+                .setMetaAddress("127.0.0.1:9559")
+                .setEnableGraphSsl(false)
+                .setEnableMetaSsl(true)
+                .setSslSignType(null)
+                .setCaSignParam("caCrtFile", "crtFile", "keyFile")
+                .build();
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testNoSslSighType() {
+        NebulaClientOptions nebulaClientOptions = new NebulaClientOptions
+                .NebulaClientOptionsBuilder()
+                .setGraphAddress(null)
+                .setMetaAddress("127.0.0.1:9559")
+                .setEnableGraphSsl(false)
+                .setEnableMetaSsl(true)
+                .setCaSignParam("caCrtFile", "crtFile", "keyFile")
+                .build();
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testGetCaSignParam() {
+        NebulaClientOptions nebulaClientOptions = new NebulaClientOptions
+                .NebulaClientOptionsBuilder()
+                .setGraphAddress(null)
+                .setMetaAddress("127.0.0.1:9559")
+                .setEnableGraphSsl(false)
+                .setEnableMetaSsl(true)
+                .setSslSignType(SslSighType.CA)
+                .setSelfSignParam("crtFile", "keyFile", "password")
+                .build();
+
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testGetSelfSignParam() {
+        NebulaClientOptions nebulaClientOptions = new NebulaClientOptions
+                .NebulaClientOptionsBuilder()
+                .setGraphAddress(null)
+                .setMetaAddress("127.0.0.1:9559")
+                .setEnableGraphSsl(false)
+                .setEnableMetaSsl(true)
+                .setSslSignType(SslSighType.SELF)
+                .setCaSignParam("caCrtFile", "crtFile", "keyFile")
+                .build();
+    }
+}