[ISSUE #XXXXX] Add SNI multi-domain certificate support for Proxy TLS

proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java

@@ -45,6 +45,7 @@
 import org.apache.rocketmq.proxy.processor.MessagingProcessor;
 import org.apache.rocketmq.proxy.remoting.RemotingProtocolServer;
 import org.apache.rocketmq.proxy.service.cert.TlsCertificateManager;
+import org.apache.rocketmq.proxy.service.cert.TlsSniManager;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.srvutil.ServerUtil;
 
@@ -78,7 +79,9 @@ public static void main(String[] args) {
             MessagingProcessor messagingProcessor = createMessagingProcessor();
 
             // tls cert update
-            TlsCertificateManager tlsCertificateManager = new TlsCertificateManager();
+            TlsSniManager tlsSniManager = new TlsSniManager();
+            tlsSniManager.initialize(ConfigurationManager.getProxyConfig());
+            TlsCertificateManager tlsCertificateManager = new TlsCertificateManager(tlsSniManager);
             PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(tlsCertificateManager);
 
             // create grpcServer

proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java

@@ -84,6 +84,7 @@ public class ProxyConfig implements ConfigFile {
     private String tlsKeyPassword = "";
     private String tlsCertPath = ConfigurationManager.getProxyHome() + "/conf/tls/rocketmq.crt";
     private int tlsCertWatchIntervalMs = 60 * 60 * 1000; // 1 hour
+    private Map<String, TlsDomainConfig> tlsDomainConfigs = new HashMap<>();
     /**
      * gRPC
      */
@@ -529,6 +530,14 @@ public void setTlsCertPath(String tlsCertPath) {
         this.tlsCertPath = tlsCertPath;
     }
 
+    public Map<String, TlsDomainConfig> getTlsDomainConfigs() {
+        return tlsDomainConfigs;
+    }
+
+    public void setTlsDomainConfigs(Map<String, TlsDomainConfig> tlsDomainConfigs) {
+        this.tlsDomainConfigs = tlsDomainConfigs;
+    }
+
     public int getGrpcBossLoopNum() {
         return grpcBossLoopNum;
     }

proxy/src/main/java/org/apache/rocketmq/proxy/config/TlsDomainConfig.java

@@ -0,0 +1,50 @@
+/*
+ * 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.rocketmq.proxy.config;
+
+public class TlsDomainConfig {
+    private String certPath;
+    private String keyPath;
+    private String keyPassword;
+
+    public TlsDomainConfig() {
+    }
+
+    public String getCertPath() {
+        return certPath;
+    }
+
+    public void setCertPath(String certPath) {
+        this.certPath = certPath;
+    }
+
+    public String getKeyPath() {
+        return keyPath;
+    }
+
+    public void setKeyPath(String keyPath) {
+        this.keyPath = keyPath;
+    }
+
+    public String getKeyPassword() {
+        return keyPassword;
+    }
+
+    public void setKeyPassword(String keyPassword) {
+        this.keyPassword = keyPassword;
+    }
+}

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServer.java

@@ -77,9 +77,9 @@ class GrpcTlsReloadHandler implements TlsCertificateManager.TlsContextReloadList
         @Override
         public void onTlsContextReload() {
             try {
-                ProxyAndTlsProtocolNegotiator.loadSslContext();
+                ProxyAndTlsProtocolNegotiator.loadAllSslContexts();
                 log.info("SslContext reloaded for grpc server");
-            } catch (CertificateException | IOException e) {
+            } catch (Exception e) {
                 log.error("Failed to reload SslContext for server", e);
             }
         }

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/ProxyAndTlsProtocolNegotiator.java

@@ -18,7 +18,6 @@
 
 import io.grpc.Attributes;
 import io.grpc.netty.shaded.io.grpc.netty.GrpcHttp2ConnectionHandler;
-import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
 import io.grpc.netty.shaded.io.grpc.netty.InternalProtocolNegotiationEvent;
 import io.grpc.netty.shaded.io.grpc.netty.InternalProtocolNegotiator;
 import io.grpc.netty.shaded.io.grpc.netty.InternalProtocolNegotiators;
@@ -35,22 +34,14 @@
 import io.grpc.netty.shaded.io.netty.handler.codec.haproxy.HAProxyMessageDecoder;
 import io.grpc.netty.shaded.io.netty.handler.codec.haproxy.HAProxyProtocolVersion;
 import io.grpc.netty.shaded.io.netty.handler.codec.haproxy.HAProxyTLV;
-import io.grpc.netty.shaded.io.netty.handler.ssl.ClientAuth;
-import io.grpc.netty.shaded.io.netty.handler.ssl.OpenSsl;
 import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
 import io.grpc.netty.shaded.io.netty.handler.ssl.SslHandler;
-import io.grpc.netty.shaded.io.netty.handler.ssl.SslProvider;
-
-import io.grpc.netty.shaded.io.netty.handler.ssl.util.InsecureTrustManagerFactory;
-import io.grpc.netty.shaded.io.netty.handler.ssl.util.SelfSignedCertificate;
+import io.grpc.netty.shaded.io.netty.handler.ssl.SniHandler;
 import io.grpc.netty.shaded.io.netty.util.AsciiString;
 import io.grpc.netty.shaded.io.netty.util.CharsetUtil;
+import io.grpc.netty.shaded.io.netty.util.GlobalEventExecutor;
+import io.grpc.netty.shaded.io.netty.util.concurrent.Promise;
 
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.file.Files;
-import java.nio.file.Paths;
-import java.security.cert.CertificateException;
 import java.util.List;
 
 import org.apache.commons.collections.CollectionUtils;
@@ -63,6 +54,7 @@
 import org.apache.rocketmq.proxy.config.ConfigurationManager;
 import org.apache.rocketmq.proxy.config.ProxyConfig;
 import org.apache.rocketmq.proxy.grpc.constant.AttributeKeys;
+import org.apache.rocketmq.proxy.service.cert.TlsSniManager;
 import org.apache.rocketmq.remoting.common.TlsMode;
 import org.apache.rocketmq.remoting.netty.TlsSystemConfig;
 
@@ -72,18 +64,19 @@ public class ProxyAndTlsProtocolNegotiator implements InternalProtocolNegotiator
     private static final String HA_PROXY_DECODER = "HAProxyDecoder";
     private static final String HA_PROXY_HANDLER = "HAProxyHandler";
     private static final String TLS_MODE_HANDLER = "TlsModeHandler";
+    private static final String SNI_HANDLER = "SniHandler";
     /**
      * the length of the ssl record header (in bytes)
      */
     private static final int SSL_RECORD_HEADER_LENGTH = 5;
 
-    private static SslContext sslContext;
+    private static volatile TlsSniManager tlsSniManager;
 
     public ProxyAndTlsProtocolNegotiator() {
         try {
-            loadSslContext();
-            log.info("SslContext created for proxy server");
-        } catch (IOException | CertificateException e) {
+            loadAllSslContexts();
+            log.info("SslContext created for proxy server with SNI support");
+        } catch (Exception e) {
             log.error("SslContext init error", e);
             throw new RuntimeException(e);
         }
@@ -103,39 +96,24 @@ public ChannelHandler newHandler(GrpcHttp2ConnectionHandler grpcHandler) {
     public void close() {
     }
 
-    public static void loadSslContext() throws CertificateException, IOException {
-        ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
-        SslProvider provider;
-        if (OpenSsl.isAvailable()) {
-            provider = SslProvider.OPENSSL;
-            log.info("Using OpenSSL provider");
-        } else {
-            provider = SslProvider.JDK;
-            log.info("Using JDK SSL provider");
-        }
-        if (proxyConfig.isTlsTestModeEnable()) {
-            SelfSignedCertificate selfSignedCertificate = new SelfSignedCertificate();
-            sslContext = GrpcSslContexts.forServer(selfSignedCertificate.certificate(), selfSignedCertificate.privateKey())
-                .sslProvider(provider)
-                .trustManager(InsecureTrustManagerFactory.INSTANCE)
-                .clientAuth(ClientAuth.NONE)
-                .build();
-        } else {
-            String tlsCertPath = ConfigurationManager.getProxyConfig().getTlsCertPath();
-            String tlsKeyPath = ConfigurationManager.getProxyConfig().getTlsKeyPath();
-            String tlsKeyPassword = ConfigurationManager.getProxyConfig().getTlsKeyPassword();
-            try (InputStream serverKeyInputStream = Files.newInputStream(
-                Paths.get(tlsKeyPath));
-                 InputStream serverCertificateStream = Files.newInputStream(
-                     Paths.get(tlsCertPath))) {
-                sslContext = GrpcSslContexts.forServer(serverCertificateStream,
-                        serverKeyInputStream,
-                        StringUtils.isNotBlank(tlsKeyPassword) ? tlsKeyPassword : null)
-                    .trustManager(InsecureTrustManagerFactory.INSTANCE)
-                    .clientAuth(ClientAuth.NONE)
-                    .build();
+    private static TlsSniManager getTlsSniManager() {
+        if (tlsSniManager == null) {
+            synchronized (ProxyAndTlsProtocolNegotiator.class) {
+                if (tlsSniManager == null) {
+                    tlsSniManager = new TlsSniManager();
+                    tlsSniManager.initialize(ConfigurationManager.getProxyConfig());
+                }
             }
         }
+        return tlsSniManager;
+    }
+
+    public static void loadAllSslContexts() {
+        getTlsSniManager();
+    }
+
+    public static TlsSniManager getManager() {
+        return getTlsSniManager();
     }
 
     private class ProxyAndTlsProtocolHandler extends ByteToMessageDecoder {
@@ -199,12 +177,6 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
             ctx.pipeline().remove(this);
         }
 
-        /**
-         * The definition of key refers to the implementation of nginx
-         * <a href="https://nginx.org/en/docs/http/ngx_http_core_module.html#var_proxy_protocol_addr">ngx_http_core_module</a>
-         *
-         * @param msg
-         */
         private void handleWithMessage(HAProxyMessage msg) {
             try {
                 Attributes.Builder builder = InternalProtocolNegotiationEvent.getAttributes(pne).toBuilder();
@@ -254,33 +226,28 @@ private class TlsModeHandler extends ByteToMessageDecoder {
 
         private ProtocolNegotiationEvent pne = InternalProtocolNegotiationEvent.getDefault();
 
-        private final ChannelHandler ssl;
-        private final ChannelHandler plaintext;
+        private final GrpcHttp2ConnectionHandler grpcHandler;
 
         public TlsModeHandler(GrpcHttp2ConnectionHandler grpcHandler) {
-            this.ssl = InternalProtocolNegotiators.serverTls(sslContext)
-                .newHandler(grpcHandler);
-            this.plaintext = InternalProtocolNegotiators.serverPlaintext()
-                .newHandler(grpcHandler);
+            this.grpcHandler = grpcHandler;
         }
 
         @Override
         protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
             try {
                 TlsMode tlsMode = TlsSystemConfig.tlsMode;
                 if (TlsMode.ENFORCING.equals(tlsMode)) {
-                    ctx.pipeline().addAfter(ctx.name(), null, this.ssl);
+                    addSniHandler(ctx);
                 } else if (TlsMode.DISABLED.equals(tlsMode)) {
-                    ctx.pipeline().addAfter(ctx.name(), null, this.plaintext);
+                    addPlaintextHandler(ctx);
                 } else {
-                    // in SslHandler.isEncrypted, it needs at least 5 bytes to judge is encrypted or not
                     if (in.readableBytes() < SSL_RECORD_HEADER_LENGTH) {
                         return;
                     }
                     if (SslHandler.isEncrypted(in)) {
-                        ctx.pipeline().addAfter(ctx.name(), null, this.ssl);
+                        addSniHandler(ctx);
                     } else {
-                        ctx.pipeline().addAfter(ctx.name(), null, this.plaintext);
+                        addPlaintextHandler(ctx);
                     }
                 }
                 ctx.fireUserEventTriggered(pne);
@@ -291,6 +258,25 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
             }
         }
 
+        private void addSniHandler(ChannelHandlerContext ctx) {
+            TlsSniManager sniManager = getTlsSniManager();
+            SniHandler sniHandler = new SniHandler((hostname, promise) -> {
+                SslContext sslCtx = sniManager.getSslContext(hostname);
+                if (sslCtx != null) {
+                    promise.setSuccess(sslCtx);
+                } else {
+                    promise.setSuccess(sniManager.getDefaultContext());
+                }
+            }, GlobalEventExecutor.INSTANCE);
+            ctx.pipeline().addAfter(ctx.name(), SNI_HANDLER, sniHandler);
+        }
+
+        private void addPlaintextHandler(ChannelHandlerContext ctx) {
+            ChannelHandler plaintext = InternalProtocolNegotiators.serverPlaintext()
+                .newHandler(grpcHandler);
+            ctx.pipeline().addAfter(ctx.name(), null, plaintext);
+        }
+
         @Override
         public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
             if (evt instanceof ProtocolNegotiationEvent) {
@@ -300,4 +286,4 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
             }
         }
     }
-}
+}