Fix stream has already been operated upon or closed (#457)

* Fix stream has already been operated upon or closed * Fix stream has already been operated upon or closed * Added check number of segment of subdomain * Fix listener close using channelGroup * Set channel Group closure timeout to 10s
diennea · Feb 21, 2024 · e945c8c · e945c8c
1 parent bfe0752
commit e945c8c
Show file tree

Hide file tree

Showing 4 changed files with 31 additions and 3 deletions.
diff --git a/carapace-server/src/main/java/org/carapaceproxy/core/Listeners.java b/carapace-server/src/main/java/org/carapaceproxy/core/Listeners.java
@@ -79,6 +79,7 @@
 import org.carapaceproxy.utils.CertificatesUtils;
 import org.carapaceproxy.utils.PrometheusUtils;
 import reactor.netty.DisposableServer;
+import reactor.netty.FutureMono;
 import reactor.netty.NettyPipeline;
 import reactor.netty.http.server.HttpServer;
 
@@ -148,6 +149,7 @@ private void stopListener(HostPort hostport) throws InterruptedException {
         ListeningChannel channel = listeningChannels.remove(hostport);
         if (channel != null) {
             channel.channel.disposeNow(Duration.ofSeconds(10));
+            FutureMono.from(channel.getConfig().getGroup().close()).block(Duration.ofSeconds(10));
         }
     }
 
@@ -283,6 +285,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
                 .doOnConnection(conn -> {
                     CURRENT_CONNECTED_CLIENTS_GAUGE.inc();
                     conn.channel().closeFuture().addListener(e -> CURRENT_CONNECTED_CLIENTS_GAUGE.dec());
+                    config.getGroup().add(conn.channel());
                 })
                 .httpRequestDecoder(option -> option.maxHeaderSize(currentConfiguration.getMaxHeaderSize()))
                 .handle((request, response) -> { // Custom request-response handling

diff --git a/...ce-server/src/main/java/org/carapaceproxy/server/config/NetworkListenerConfiguration.java b/...ce-server/src/main/java/org/carapaceproxy/server/config/NetworkListenerConfiguration.java
@@ -21,6 +21,10 @@
 
 import java.util.Collections;
 import java.util.Set;
+
+import io.netty.channel.group.ChannelGroup;
+import io.netty.channel.group.DefaultChannelGroup;
+import io.netty.util.concurrent.DefaultEventExecutor;
 import lombok.Data;
 
 /**
@@ -44,6 +48,8 @@ public class NetworkListenerConfiguration {
     private int keepAliveCount;
     private int maxKeepAliveRequests;
 
+    private ChannelGroup group;
+
     public HostPort getKey() {
         return new HostPort(host, port);
     }
@@ -96,6 +102,7 @@ public NetworkListenerConfiguration(String host,
         this.keepAliveInterval = keepAliveInterval;
         this.keepAliveCount = keepAliveCount;
         this.maxKeepAliveRequests = maxKeepAliveRequests;
+        this.group = new DefaultChannelGroup(new DefaultEventExecutor());
     }
 
 }
diff --git a/...ace-server/src/main/java/org/carapaceproxy/server/config/SSLCertificateConfiguration.java b/...ace-server/src/main/java/org/carapaceproxy/server/config/SSLCertificateConfiguration.java
@@ -93,10 +93,23 @@ public boolean isMoreSpecific(SSLCertificateConfiguration other) {
         if (subjectAltNames == null || subjectAltNames.isEmpty()) {
             return hostname.length() > other.getHostname().length();
         }
-        final var otherNames = other.getNames().stream().map(CertificatesUtils::removeWildcard);
-        for (var n: getNames()) {
+
+        final int maxOtherNameLength = other.getNames().stream()
+                .map(CertificatesUtils::removeWildcard)
+                .mapToInt(String::length)
+                .max()
+                .orElse(0);
+
+        final int maxSubDomainLength = other.getNames().stream()
+                .map(name -> name.split("\\."))
+                .mapToInt(name -> name.length)
+                .max()
+                .orElse(0);
+
+        for (var n : getNames()) {
             final var name = CertificatesUtils.removeWildcard(n);
-            if (otherNames.anyMatch(on -> name.length() > on.length())) {
+            final int nameSegmentLength = n.split("\\.").length;
+            if (name.length() >= maxOtherNameLength && nameSegmentLength >= maxSubDomainLength) {
                 return true;
             }
         }

diff --git a/carapace-server/src/test/java/org/carapaceproxy/listeners/SSLSNITest.java b/carapace-server/src/test/java/org/carapaceproxy/listeners/SSLSNITest.java
@@ -96,10 +96,14 @@ public void testChooseCertificate() throws Exception {
             server.addCertificate(new SSLCertificateConfiguration("other", null, "cert", "pwd", STATIC));
             server.addCertificate(new SSLCertificateConfiguration("*.example.com", Set.of("example.com", "*.example2.com"), "cert", "pwd", STATIC));
             server.addCertificate(new SSLCertificateConfiguration("www.example.com", null, "cert", "pwd", STATIC));
+            server.addCertificate(new SSLCertificateConfiguration("*.qatest.pexample.it", Set.of("qatest.pexample.it"), "cert", "pwd", STATIC));
+            server.addCertificate(new SSLCertificateConfiguration("*.pexample.it", Set.of("qatest2.pexample.it"), "cert", "pwd", STATIC));
+
 
             // client requests bad SNI, bad default in listener
             assertNull(server.getListeners().chooseCertificate("no", "no-default"));
 
+            assertEquals("*.qatest.pexample.it", server.getListeners().chooseCertificate("test2.qatest.pexample.it", "no-default").getId());
             // client requests SNI, bad default in listener
             assertEquals("other", server.getListeners().chooseCertificate("other", "no-default").getId());
 
@@ -142,6 +146,7 @@ public void testChooseCertificate() throws Exception {
             assertEquals("*", server.getListeners().chooseCertificate("", null).getId());
             assertEquals("*", server.getListeners().chooseCertificate(null, "").getId());
         }
+
     }
 
     @Test