Merge pull request #6042 from eclipse-vertx/http-tls-test-improvements

HTTP/QUIC improvements

Author: vietj

vertx-core/src/main/java/io/vertx/core/http/HttpServer.java

@@ -92,7 +92,7 @@ public interface HttpServer extends Measured {
 
   /**
    * Set an exception handler called for socket errors happening before the HTTP connection
-   * is established, e.g during the TLS handshake.
+   * is established, e.g. during the TLS handshake.
    *
    * @param handler the handler to set
    * @return a reference to this, so the API can be used fluently

vertx-core/src/main/java/io/vertx/core/http/impl/quic/QuicHttpServer.java

@@ -45,6 +45,7 @@ public class QuicHttpServer implements HttpServerInternal {
   private final boolean manageMetrics;
   private volatile Handler<HttpServerRequest> requestHandler;
   private Handler<HttpConnection> connectionHandler;
+  private Handler<Throwable> exceptionHandler;
   private QuicServerImpl quicServer;
   private HttpServerMetrics<?, ?> httpMetrics;
   private volatile int actualPort;
@@ -108,6 +109,7 @@ public HttpServer exceptionHandler(Handler<Throwable> handler) {
     if (actualPort > 0) {
       throw new IllegalStateException("Server already bound");
     }
+    this.exceptionHandler = handler;
     return this;
   }
 
@@ -241,6 +243,7 @@ public Future<HttpServer> listen(ContextInternal current, SocketAddress address)
     quicServer.connectHandler(new ConnectionHandler(quicServer, httpMetrics, requestHandler, connectionHandler,
       config.isHandle100ContinueAutomatically(), config.getMaxFormAttributeSize(), config.getMaxFormFields(), config.getMaxFormBufferedBytes(),
       http3Config.getInitialSettings() != null ? http3Config.getInitialSettings().copy() : new Http3Settings(), logEnabled));
+    quicServer.exceptionHandler(exceptionHandler);
     return quicServer
       .bind(current, address)
       .map(addr -> {

vertx-core/src/main/java/io/vertx/core/internal/tls/ServerSslContextProvider.java

@@ -106,7 +106,7 @@ public SslContext createServerContext(KeyManagerFactory keyManagerFactory,
     }
     try {
       SslContextFactory factory = provider.get()
-        .forServer(SslContextManager.CLIENT_AUTH_MAPPING.get(clientAuth))
+        .forServer(SslContextManager.mapClientAuth(clientAuth))
         .enabledProtocols(enabledProtocols)
         .enabledCipherSuites(enabledCipherSuites)
         .useAlpn(applicationProtocols != null)

vertx-core/src/main/java/io/vertx/core/internal/tls/SslContextManager.java

@@ -36,8 +36,12 @@
  */
 public abstract class SslContextManager<P extends SslContextProvider> {
 
+  public static io.netty.handler.ssl.ClientAuth mapClientAuth(ClientAuth auth) {
+    return CLIENT_AUTH_MAPPING.get(auth);
+  }
+
   private static final Config NULL_CONFIG = new Config(null, null, null, null, null);
-  static final EnumMap<ClientAuth, io.netty.handler.ssl.ClientAuth> CLIENT_AUTH_MAPPING = new EnumMap<>(ClientAuth.class);
+  private static final EnumMap<ClientAuth, io.netty.handler.ssl.ClientAuth> CLIENT_AUTH_MAPPING = new EnumMap<>(ClientAuth.class);
 
   static {
     CLIENT_AUTH_MAPPING.put(ClientAuth.REQUIRED, io.netty.handler.ssl.ClientAuth.REQUIRE);

vertx-core/src/main/java/io/vertx/core/net/QuicServer.java

@@ -10,9 +10,11 @@
  */
 package io.vertx.core.net;
 
+import io.vertx.codegen.annotations.Fluent;
 import io.vertx.codegen.annotations.VertxGen;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
+import io.vertx.core.http.HttpServer;
 import io.vertx.core.net.impl.SocketAddressImpl;
 
 /**
@@ -28,6 +30,7 @@ public interface QuicServer extends QuicEndpoint {
    * @param handler the connection handler
    * @return this object instance
    */
+  @Fluent
   QuicServer connectHandler(Handler<QuicConnection> handler);
 
   /**
@@ -37,6 +40,7 @@ public interface QuicServer extends QuicEndpoint {
    * @param handler the handler processing streams
    * @return this object instance
    */
+  @Fluent
   default QuicServer streamHandler(Handler<QuicStream> handler) {
     if (handler != null) {
       return connectHandler(connection -> {
@@ -47,6 +51,16 @@ default QuicServer streamHandler(Handler<QuicStream> handler) {
     }
   }
 
+  /**
+   * Set an exception handler called for socket errors happening before the QUIC connection
+   * is established, e.g. during the TLS handshake.
+   *
+   * @param handler the handler to set
+   * @return a reference to this, so the API can be used fluently
+   */
+  @Fluent
+  QuicServer exceptionHandler(Handler<Throwable> handler);
+
   /**
    * Start listening on the {@code port} and {@code host} as configured in the {@link io.vertx.core.net.QuicServerConfig} used when
    * creating the server.

vertx-core/src/main/java/io/vertx/core/net/impl/quic/QuicClientImpl.java

@@ -23,6 +23,7 @@
 import io.netty.handler.logging.ByteBufFormat;
 import io.netty.util.Attribute;
 import io.netty.util.AttributeKey;
+import io.vertx.core.Completable;
 import io.vertx.core.Future;
 import io.vertx.core.internal.ContextInternal;
 import io.vertx.core.internal.PromiseInternal;
@@ -32,8 +33,6 @@
 import io.vertx.core.internal.resolver.NameResolver;
 import io.vertx.core.internal.tls.ClientSslContextManager;
 import io.vertx.core.internal.tls.ClientSslContextProvider;
-import io.vertx.core.internal.tls.SslContextManager;
-import io.vertx.core.internal.tls.SslContextProvider;
 import io.vertx.core.net.*;
 import io.vertx.core.net.impl.ConnectRetry;
 import io.vertx.core.spi.metrics.Metrics;
@@ -93,8 +92,7 @@ protected Future<QuicCodecBuilder<?>> codecBuilder(ContextInternal context, Tran
         Attribute<HostAndPort> peerAttr = q.attr(SSL_PEER_KEY);
         HostAndPort peer = peerAttr.get();
         return sslContext.newEngine(q.alloc(), peer.host(), peer.port());
-      })
-      .maxIdleTimeout(5000, TimeUnit.MILLISECONDS));
+      }));
   }
 
   @Override
@@ -207,9 +205,14 @@ protected void initChannel(Channel ch) {
           connectionGroup.add(ch);
           LogConfig logConfig = config.getLogConfig();
           ByteBufFormat activityLogging = logConfig != null && logConfig.isEnabled()? logConfig.getDataFormat() : null;
+          Completable<QuicConnection> adapter = (result, failure) -> {
+            if (failure == null) {
+              promise.tryComplete(result);
+            }
+          };
           QuicConnectionHandler handler = new QuicConnectionHandler(context, metrics, config.getIdleTimeout(),
             config.getReadIdleTimeout(), config.getWriteIdleTimeout(), activityLogging, config.getMaxStreamBidiRequests(),
-            config.getMaxStreamUniRequests(), remoteAddress, false, promise::tryComplete);
+            config.getMaxStreamUniRequests(), remoteAddress, false, adapter);
           ch.pipeline().addLast("handler", handler);
         }
       })

vertx-core/src/main/java/io/vertx/core/net/impl/quic/QuicConnectionHandler.java

@@ -20,7 +20,8 @@
 import io.netty.handler.codec.quic.QuicStreamLimitChangedEvent;
 import io.netty.handler.logging.ByteBufFormat;
 import io.netty.handler.ssl.SniCompletionEvent;
-import io.vertx.core.Handler;
+import io.netty.handler.ssl.SslHandshakeCompletionEvent;
+import io.vertx.core.Completable;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.internal.ContextInternal;
 import io.vertx.core.internal.net.SslHandshakeCompletionHandler;
@@ -51,15 +52,15 @@ private static long timeoutMillis(Duration timeout) {
   private final int maxStreamBidiRequests;
   private final int maxStreamUniRequests;
   private final boolean server;
-  private Handler<QuicConnection> handler;
-  private QuicChannel channel;
+  private final SocketAddress remoteAddress;
+  private Completable<QuicConnection> handler;
   private QuicConnectionImpl connection;
-  private SocketAddress remoteAddress;
+  private int maxDatagramLength;
 
   public QuicConnectionHandler(ContextInternal context, TransportMetrics<?> metrics, Duration idleTimeout,
                                Duration readIdleTimeout, Duration writeIdleTimeout, ByteBufFormat activityLogging,
                                int maxStreamBidiRequests, int maxStreamUniRequests, SocketAddress remoteAddress,
-                               boolean server, Handler<QuicConnection> handler) {
+                               boolean server, Completable<QuicConnection> handler) {
     this.context = context;
     this.metrics = metrics;
     this.idleTimeout = timeoutMillis(idleTimeout);
@@ -73,29 +74,24 @@ public QuicConnectionHandler(ContextInternal context, TransportMetrics<?> metric
     this.server = server;
   }
 
-  @Override
-  public void handlerAdded(ChannelHandlerContext ctx) {
-    QuicChannel ch = (QuicChannel) ctx.channel();
-    channel = ch;
-    connection = new QuicConnectionImpl(context, metrics, idleTimeout, readIdleTimeout, writeIdleTimeout, activityLogging,
-      maxStreamBidiRequests, maxStreamUniRequests, ch, remoteAddress, ctx, server);
-  }
-
   @Override
   public void channelActive(ChannelHandlerContext ctx) throws Exception {
-    super.channelActive(ctx);
-    activate();
+    activate(ctx);
   }
 
-  private void activate() {
+  private void activate(ChannelHandlerContext ctx) {
+    QuicChannel ch = (QuicChannel) ctx.channel();
+    QuicConnectionImpl c = new QuicConnectionImpl(context, metrics, idleTimeout, readIdleTimeout, writeIdleTimeout, activityLogging,
+      maxStreamBidiRequests, maxStreamUniRequests, maxDatagramLength, ch, remoteAddress, ctx, server);
+    connection = c;
     if (metrics != null) {
-      Object metric = metrics.connected(connection.remoteAddress(), connection.remoteName());
-      connection.metric(metric);
+      Object metric = metrics.connected(c.remoteAddress(), c.remoteName());
+      c.metric(metric);
     }
-    Handler<QuicConnection> h = handler;
+    Completable<QuicConnection> h = handler;
     if (h != null) {
       handler = null;
-      context.dispatch(connection, h);
+      h.succeed(c);
     }
   }
 
@@ -131,22 +127,29 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
         c.shutdown(shutdown.timeout());
       }
     } else if (evt instanceof QuicStreamLimitChangedEvent) {
-      connection.handleQuicStreamLimitChanged();
+      QuicConnectionImpl c = connection;
+      if (c != null) {
+        c.handleQuicStreamLimitChanged();
+      }
     } else if (evt instanceof QuicDatagramExtensionEvent) {
       QuicDatagramExtensionEvent datagramExtensionEvent = (QuicDatagramExtensionEvent) evt;
-      QuicConnectionImpl c = connection;
-      c.enableDatagramExtension(datagramExtensionEvent.maxLength());
-
-      // Activate at this moment, since the handler activation happens before we get relevant information
-      // datagram extension event is last, see QuicheQuicChannel
-      activate();
+      this.maxDatagramLength = datagramExtensionEvent.maxLength();
+    } else if (evt instanceof SslHandshakeCompletionEvent) {
+      SslHandshakeCompletionEvent handshakeEvt = (SslHandshakeCompletionEvent)evt;
+      if (!handshakeEvt.isSuccess()) {
+        Completable<QuicConnection> h = handler;
+        if (h != null) {
+          h.fail(handshakeEvt.cause());
+        }
+      }
     }
     super.userEventTriggered(ctx, evt);
   }
 
   @Override
   public void exceptionCaught(ChannelHandlerContext chctx, final Throwable t) {
-    if (connection.handleException(t)) {
+    QuicConnectionImpl c = connection;
+    if (c != null && c.handleException(t)) {
       chctx.close();
     }
   }

vertx-core/src/main/java/io/vertx/core/net/impl/quic/QuicConnectionImpl.java

@@ -78,8 +78,8 @@ public class QuicConnectionImpl extends ConnectionBase implements QuicConnection
 
   public QuicConnectionImpl(ContextInternal context, TransportMetrics metrics, long idleTimeout,
                             long readIdleTimeout,  long writeIdleTimeout, ByteBufFormat activityLogging, int maxStreamBidiRequests,
-                            int maxStreamUniRequests, QuicChannel channel, SocketAddress remoteAddress, ChannelHandlerContext chctx,
-                            boolean server) {
+                            int maxStreamUniRequests, int maxDatagramLength, QuicChannel channel, SocketAddress remoteAddress,
+                            ChannelHandlerContext chctx, boolean server) {
     super(context, chctx);
 
     Map<QuicStreamType, StreamOpenRequestQueue> pendingStreamRequestsMap = new EnumMap<>(QuicStreamType.class);
@@ -96,7 +96,7 @@ public QuicConnectionImpl(ContextInternal context, TransportMetrics metrics, lon
     this.remoteAddress = remoteAddress;
     this.server = server;
     this.pendingStreamOpenRequestsMap = pendingStreamRequestsMap;
-    this.maxDatagramLength = 0;
+    this.maxDatagramLength = maxDatagramLength;
     this.streamGroup = new ConnectionGroup(context.nettyEventLoop()) {
       @Override
       protected void handleShutdown(Duration timeout, Completable<Void> completion) {
@@ -200,10 +200,6 @@ void handleDatagram(ByteBuf byteBuf) {
     }
   }
 
-  void enableDatagramExtension(int maxDatagramLength) {
-    this.maxDatagramLength = Math.max(0, maxDatagramLength);
-  }
-
   void handleQuicStreamLimitChanged() {
     trySatisfyPendingStreamOpenRequests(QuicStreamType.BIDIRECTIONAL);
     trySatisfyPendingStreamOpenRequests(QuicStreamType.UNIDIRECTIONAL);

vertx-core/src/main/java/io/vertx/core/net/impl/quic/QuicEndpointImpl.java

@@ -58,6 +58,7 @@ public abstract class QuicEndpointImpl implements QuicEndpointInternal, MetricsP
   private final QuicEndpointConfig config;
   private final String protocol;
   protected final SslContextManager<?> manager;
+  protected final BoringSSLKeylog keylog;
   protected final VertxInternal vertx;
   private TransportMetrics<?> metrics;
   private Channel channel;
@@ -100,6 +101,7 @@ public QuicEndpointImpl(VertxInternal vertx, QuicEndpointConfig config, String p
     this.protocol = protocol;
     this.vertx = Objects.requireNonNull(vertx);
     this.manager = sslContextManager(new BoringSslEngineOptions(keylog));
+    this.keylog = keylog;
   }
 
   abstract SslContextManager<?> sslContextManager(BoringSslEngineOptions engine);