Fix support for TLS1.3 early data (java-native-access#524)

Motivation:

How early data support was implemented was incorrect. We should only
indicate early data is ready when it really is.

Modifications:

- Fix implementation to only notify about early data readiness when it
is actually the case
- Fix unit test

Result:

Correctly handle and support early data

codec-classes-quic/src/main/java/io/netty/incubator/codec/quic/QuicheQuicChannel.java

@@ -148,6 +148,7 @@ public void operationComplete(ChannelFuture channelFuture) {
  private boolean recvStreamPending;
  private boolean streamReadable;
  private boolean handshakeCompletionNotified;
+ private boolean earlyDataReadyNotified;
 
  private int reantranceGuard = 0;
  private static final int IN_RECV = 1 << 1;
@@ -1229,7 +1230,12 @@ private static int calculateSendBufferLength(long connAddr, int maxDatagramSize)
  * {@link Channel#flush()} at some point.
  */
  private boolean connectionSend() {
- if (isConnDestroyed() || (reantranceGuard & IN_CONNECTION_SEND) != 0) {
+ if (isConnDestroyed()) {
+ return false;
+ }
+ if ((reantranceGuard & IN_CONNECTION_SEND) != 0) {
+ // Let's notify about early data if needed.
+ notifyEarlyDataReadyIfNeeded();
  return false;
  }
 
@@ -1251,13 +1257,18 @@ private boolean connectionSend() {
  // Consume all tasks
  do {
  task.run();
+ // Notify about early data ready if needed.
+ notifyEarlyDataReadyIfNeeded();
  } while ((task = connection.sslTask()) != null);
 
  // Let's try again sending after we did process all tasks.
  return packetWasWritten | connectionSend();
  } else {
  runAllTaskSend(sslTaskExecutor, task);
  }
+ } else {
+ // Notify about early data ready if needed.
+ notifyEarlyDataReadyIfNeeded();
  }
 
  if (packetWasWritten) {
@@ -1691,11 +1702,18 @@ private QuicheQuicStreamChannel addNewStreamChannel(long streamId) {
  void finishConnect() {
  assert !server;
  if (connectionSend()) {
- pipeline().fireUserEventTriggered(SslEarlyDataReadyEvent.INSTANCE);
  flushParent();
  }
  }
 
+ private void notifyEarlyDataReadyIfNeeded() {
+ if (!server && !earlyDataReadyNotified &&
+ !isConnDestroyed() && Quiche.quiche_conn_is_in_early_data(connection.address())) {
+ earlyDataReadyNotified = true;
+ pipeline().fireUserEventTriggered(SslEarlyDataReadyEvent.INSTANCE);
+ }
+ }
+
  // TODO: Come up with something better.
  static QuicheQuicChannel handleConnect(Function<QuicChannel, ? extends QuicSslEngine> sslEngineProvider,
  Executor sslTaskExecutor,

codec-classes-quic/src/main/java/io/netty/incubator/codec/quic/SslEarlyDataReadyEvent.java

@@ -17,10 +17,11 @@
 
 
 /**
- * Event which is fired once it's possible to send early data.
+ * Event which is fired once it's possible to send early data on the client-side.
+ * See <a href="https://www.rfc-editor.org/rfc/rfc8446#section-4.2.10">RFC8446 4.2.10 Early Data Indication</a>.
  * <p>
- * Users might call {@link io.netty.channel.Channel#write(Object)} to send early data. The data is automatically
- * flushed as part of the connection establishment.
+ * Users might call {@link io.netty.channel.Channel#writeAndFlush(Object)} or
+ * {@link io.netty.channel.ChannelHandlerContext#writeAndFlush(Object)} to send early data.
  * Please be aware that early data may be replay-able and so may have other security concerns then other data.
  */
 public final class SslEarlyDataReadyEvent {

codec-native-quic/src/test/java/io/netty/incubator/codec/quic/QuicChannelConnectTest.java

@@ -19,7 +19,6 @@
 import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.channel.ChannelOption;
@@ -39,6 +38,7 @@
 import org.junit.jupiter.api.Timeout;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
+import org.opentest4j.AssertionFailedError;
 
 import javax.net.ssl.SSLEngine;
 import javax.net.ssl.SSLEngineResult;
@@ -72,7 +72,6 @@
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
@@ -575,27 +574,39 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
  @ParameterizedTest
  @MethodSource("newSslTaskExecutors")
  public void testConnectWith0RTT(Executor executor) throws Throwable {
- final ChannelHandler closeHandler = new ChannelInboundHandlerAdapter() {
- @Override
- public boolean isSharable() {
- return true;
- }
-
- @Override
- public void channelActive(ChannelHandlerContext ctx) {
- // Close the connection / stream once accepted.
- ctx.close();
- }
- };
-
+ final CountDownLatch readLatch = new CountDownLatch(1);
  Channel server = QuicTestUtils.newServer(QuicTestUtils.newQuicServerBuilder(executor,
  QuicSslContextBuilder.forServer(
  QuicTestUtils.SELF_SIGNED_CERTIFICATE.privateKey(), null,
  QuicTestUtils.SELF_SIGNED_CERTIFICATE.certificate())
  .applicationProtocols(QuicTestUtils.PROTOS)
  .earlyData(true)
  .build()),
- InsecureQuicTokenHandler.INSTANCE, closeHandler, closeHandler);
+ InsecureQuicTokenHandler.INSTANCE, new ChannelInboundHandlerAdapter() {
+ @Override
+ public boolean isSharable() {
+ return true;
+ }
+ }, new ChannelInboundHandlerAdapter() {
+ @Override
+ public boolean isSharable() {
+ return true;
+ }
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) {
+ ByteBuf buffer = (ByteBuf) msg;
+ try {
+ assertEquals(4, buffer.readableBytes());
+ assertEquals(1, buffer.readInt());
+ readLatch.countDown();
+ ctx.close();
+ ctx.channel().parent().close();
+ } finally {
+ buffer.release();
+ }
+ }
+ });
  InetSocketAddress address = (InetSocketAddress) server.localAddress();
 
  QuicSslContext sslContext = QuicSslContextBuilder.forClient()
@@ -605,7 +616,6 @@ public void channelActive(ChannelHandlerContext ctx) {
  .build();
  Channel channel = QuicTestUtils.newClient(QuicTestUtils.newQuicClientBuilder(executor, sslContext)
  .sslEngineProvider(q -> sslContext.newEngine(q.alloc(), "localhost", 9999)));
- final CountDownLatch errorLatch = new CountDownLatch(1);
  final CountDownLatch streamLatch = new CountDownLatch(1);
  final AtomicReference<Throwable> errorRef = new AtomicReference<>();
 
@@ -615,20 +625,7 @@ public void channelActive(ChannelHandlerContext ctx) {
  @Override
  public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
  if (evt instanceof SslEarlyDataReadyEvent) {
- ((QuicChannel) ctx.channel()).createStream(QuicStreamType.BIDIRECTIONAL,
- new ChannelInboundHandlerAdapter()).addListener(f -> {
- try {
- QuicException exception = assertInstanceOf(QuicException.class, f.cause());
- // This is expected as we didn't receive the transport params yet.
- // Creating a new stream will only work once we create the next connection
- // as part of 0-RTT.
- assertEquals(QuicError.STREAM_LIMIT, exception.error());
- } catch (Throwable error) {
- errorRef.set(error);
- } finally {
- errorLatch.countDown();
- }
- });
+ errorRef.set(new AssertionFailedError("Shouldn't be called on the first connection"));
  }
  ctx.fireUserEventTriggered(evt);
  }
@@ -638,13 +635,12 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
  .connect()
  .get();
 
- errorLatch.await();
+ quicChannel.close().sync();
+
  if (errorRef.get() != null) {
  throw errorRef.get();
  }
 
- quicChannel.closeFuture().sync();
-
  quicChannel = QuicTestUtils.newQuicChannelBootstrap(channel)
  .handler(new ChannelInboundHandlerAdapter() {
  @Override
@@ -656,6 +652,10 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
  // This should succeed as we have the transport params cached as part of
  // the session.
  assertTrue(f.isSuccess());
+ Channel stream = (Channel) f.getNow();
+
+ // Let's write some data as part of the client hello.
+ stream.writeAndFlush(stream.alloc().buffer().writeInt(1));
  } catch (Throwable error) {
  errorRef.set(error);
  } finally {
@@ -677,6 +677,7 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
  }
 
  quicChannel.closeFuture().sync();
+ readLatch.await();
  } finally {
  server.close().sync();
  // Close the parent Datagram channel as well.