feat: add ssl support

Author: NguyenHoangSon96

src/main/java/com/influxdb/v3/client/config/ClientConfig.java

@@ -35,8 +35,10 @@
 import java.util.function.BiFunction;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
+import javax.net.ssl.SSLContext;
 
 import io.grpc.ProxyDetector;
+import io.netty.handler.ssl.SslContext;
 
 import com.influxdb.v3.client.write.WritePrecision;
 
@@ -103,6 +105,8 @@ public final class ClientConfig {
     private final ProxyDetector queryApiProxy;
     private final Authenticator authenticator;
     private final Map<String, String> headers;
+    private final SslContext grpcSslContext;
+    private final SSLContext sslContext;
 
     /**
      * Gets URL of the InfluxDB server.
@@ -253,6 +257,17 @@ public Map<String, String> getHeaders() {
         return headers;
     }
 
+
+    @Nullable
+    public SslContext getGrpcSslContext() {
+        return grpcSslContext;
+    }
+
+    @Nullable
+    public SSLContext getSslContext() {
+        return sslContext;
+    }
+
     /**
      * Validates the configuration properties.
      */
@@ -285,7 +300,9 @@ public boolean equals(final Object o) {
                 && Objects.equals(proxy, that.proxy)
                 && Objects.equals(queryApiProxy, that.queryApiProxy)
                 && Objects.equals(authenticator, that.authenticator)
-                && Objects.equals(headers, that.headers);
+                && Objects.equals(headers, that.headers)
+                && Objects.equals(grpcSslContext, that.grpcSslContext)
+                && Objects.equals(sslContext, that.sslContext);
     }
 
     @Override
@@ -294,7 +311,7 @@ public int hashCode() {
           database, writePrecision, gzipThreshold,
           timeout, allowHttpRedirects, disableServerCertificateValidation,
           proxy, queryApiProxy, authenticator, headers,
-          defaultTags);
+          defaultTags, grpcSslContext, sslContext);
     }
 
     @Override
@@ -313,6 +330,8 @@ public String toString() {
                 .add("authenticator=" + authenticator)
                 .add("headers=" + headers)
                 .add("defaultTags=" + defaultTags)
+                .add("grpcSslContext=" + grpcSslContext)
+                .add("sslContext=" + sslContext)
                 .toString();
     }
 
@@ -337,6 +356,8 @@ public static final class Builder {
         private ProxyDetector queryApiProxy;
         private Authenticator authenticator;
         private Map<String, String> headers;
+        private SslContext grpcSslContext;
+        private SSLContext sslContext;
 
         /**
          * Sets the URL of the InfluxDB server.
@@ -553,6 +574,20 @@ public Builder headers(@Nullable final Map<String, String> headers) {
             return this;
         }
 
+        @Nonnull
+        public Builder grpcSslContext(@Nullable final SslContext grpcSslContext) {
+
+            this.grpcSslContext = grpcSslContext;
+            return this;
+        }
+
+        @Nonnull
+        public Builder sslContext(@Nullable final SSLContext sslContext) {
+
+            this.sslContext = sslContext;
+            return this;
+        }
+
         /**
          * Build an instance of {@code ClientConfig}.
          *
@@ -691,5 +726,7 @@ private ClientConfig(@Nonnull final Builder builder) {
         queryApiProxy = builder.queryApiProxy;
         authenticator = builder.authenticator;
         headers = builder.headers;
+        grpcSslContext = builder.grpcSslContext;
+        sslContext = builder.sslContext;
     }
 }

src/main/java/com/influxdb/v3/client/internal/FlightSqlClient.java

@@ -47,6 +47,7 @@
 import io.grpc.netty.NettyChannelBuilder;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.ServerChannel;
+import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
 import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
 import org.apache.arrow.flight.FlightClient;
@@ -94,11 +95,7 @@ final class FlightSqlClient implements AutoCloseable {
             defaultHeaders.putAll(config.getHeaders());
         }
 
-        this.client = Objects.requireNonNullElseGet(client, () -> config.getQueryApiProxy() != null
-                ?
-                createFlightWithQueryProxy(config)
-                :
-                createFlightClient(config));
+        this.client = createFlightClient(config);
     }
 
     @Nonnull
@@ -142,17 +139,8 @@ public void close() throws Exception {
     @Nonnull
     private FlightClient createFlightClient(@Nonnull final ClientConfig config) {
         Location location = createLocation(config);
-
-        return FlightClient.builder()
-                .location(location)
-                .allocator(new RootAllocator(Long.MAX_VALUE))
-                .verifyServer(!config.getDisableServerCertificateValidation())
-                .build();
-    }
-
-    public FlightClient createFlightWithQueryProxy(@Nonnull final ClientConfig config) {
         final NettyChannelBuilder nettyChannelBuilder;
-        Location location = createLocation(config);
+
         switch (location.getUri().getScheme()) {
             case LocationSchemes.GRPC:
             case LocationSchemes.GRPC_INSECURE:
@@ -205,17 +193,24 @@ public FlightClient createFlightWithQueryProxy(@Nonnull final ClientConfig confi
         if (LocationSchemes.GRPC_TLS.equals(location.getUri().getScheme())) {
             nettyChannelBuilder.useTransportSecurity();
 
-            final SslContextBuilder sslContextBuilder = GrpcSslContexts.forClient();
-
-            if (config.getDisableServerCertificateValidation()) {
-                sslContextBuilder.trustManager(InsecureTrustManagerFactory.INSTANCE);
-            }
-
-            try {
-                nettyChannelBuilder.sslContext(sslContextBuilder.build());
-            } catch (SSLException e) {
-                throw new RuntimeException(e);
+            SslContextBuilder sslContextBuilder;
+            SslContext sslContext;
+            if(config.getGrpcSslContext() != null) {
+                sslContext = config.getGrpcSslContext();
+                nettyChannelBuilder.sslContext(sslContext);
+            } else {
+                sslContextBuilder = GrpcSslContexts.forClient();
+                if (config.getDisableServerCertificateValidation()) {
+                    sslContextBuilder.trustManager(InsecureTrustManagerFactory.INSTANCE);
+                }
+                try {
+                    nettyChannelBuilder.sslContext(sslContextBuilder.build());
+                } catch (SSLException e) {
+                    throw new RuntimeException(e);
+                }
             }
+        } else {
+            nettyChannelBuilder.usePlaintext();
         }
 
         if (config.getQueryApiProxy() != null) {
@@ -229,6 +224,7 @@ public FlightClient createFlightWithQueryProxy(@Nonnull final ClientConfig confi
         return FlightGrpcUtils.createFlightClient(new RootAllocator(Long.MAX_VALUE), nettyChannelBuilder.build());
     }
 
+
     @Nonnull
     private Location createLocation(@Nonnull final ClientConfig config) {
         try {

src/main/java/com/influxdb/v3/client/internal/RestClient.java

@@ -107,10 +107,15 @@ public void checkServerTrusted(
 
         if (baseUrl.startsWith("https")) {
             try {
-                if (config.getDisableServerCertificateValidation()) {
-                    SSLContext sslContext = SSLContext.getInstance("TLS");
+                SSLContext sslContext;
+                sslContext = SSLContext.getInstance("TLS");
+                SSLContext customeSslContext = config.getSslContext();
+                if (customeSslContext == null && config.getDisableServerCertificateValidation()) {
                     sslContext.init(null, TRUST_ALL_CERTS, new SecureRandom());
                     builder.sslContext(sslContext);
+                } else {
+                    sslContext.init(null, null, new SecureRandom());
+                    builder.sslContext(customeSslContext != null ? customeSslContext : sslContext);
                 }
             } catch (Exception e) {
                 throw new RuntimeException(e);