Repository: clj-commons/aleph (clojure)

Issue:
TCP server: Spurious NPE on bogus TLS connection establishment
Using `b41471529f023ef82595353dc2e9f3db548e3abf`, the following errors occur when attempting to establish a bogus TLS connection to an `aleph.tcp` server with TLS enabled:

<pre>
 WARN [aleph.tcp:326] - error in TCP server
                                                      java.lang.Thread.run                         Thread.java:  829
                      io.netty.util.concurrent.FastThreadLocalRunnable.run        FastThreadLocalRunnable.java:   30
                                                                       ...                                          
                                  manifold.executor/thread-factory/reify/f                        executor.clj:   70
                            io.netty.util.internal.ThreadExecutorMap$2.run              ThreadExecutorMap.java:   74
                  io.netty.util.concurrent.SingleThreadEventExecutor$4.run      SingleThreadEventExecutor.java:  997
                                     io.netty.channel.nio.NioEventLoop.run                   NioEventLoop.java:  496
                     io.netty.channel.nio.NioEventLoop.processSelectedKeys                   NioEventLoop.java:  584
            io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized                   NioEventLoop.java:  658
                      io.netty.channel.nio.NioEventLoop.processSelectedKey                   NioEventLoop.java:  722
            io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read         AbstractNioByteChannel.java:  166
                   io.netty.channel.DefaultChannelPipeline.fireChannelRead         DefaultChannelPipeline.java:  919
          io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead  AbstractChannelHandlerContext.java:  365
          io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead  AbstractChannelHandlerContext.java:  379
           io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead         DefaultChannelPipeline.java: 1410
            io.netty.channel.AbstractChannelHandlerContext.fireChannelRead  AbstractChannelHandlerContext.java:  357
          io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead  AbstractChannelHandlerContext.java:  365
          io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead  AbstractChannelHandlerContext.java:  379
                   io.netty.handler.codec.ByteToMessageDecoder.channelRead           ByteToMessageDecoder.java:  279
                    io.netty.handler.codec.ByteToMessageDecoder.callDecode           ByteToMessageDecoder.java:  449
io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection           ByteToMessageDecoder.java:  510
                                    io.netty.handler.ssl.SslHandler.decode                     SslHandler.java: 1285
                       io.netty.handler.ssl.SslHandler.decodeJdkCompatible                     SslHandler.java: 1215
io.netty.handler.ssl.NotSslRecordException: not an SSL/TLS record: 666f6f626172
   io.netty.handler.codec.DecoderException: io.netty.handler.ssl.NotSslRecordException: not an SSL/TLS record: 666f6f626172

 WARN [aleph.tcp:326] - error in TCP server
                                                java.lang.Thread.run                         Thread.java:  829
                io.netty.util.concurrent.FastThreadLocalRunnable.run        FastThreadLocalRunnable.java:   30
                                                                 ...                                          
                            manifold.executor/thread-factory/reify/f                        executor.clj:   70
                      io.netty.util.internal.ThreadExecutorMap$2.run              ThreadExecutorMap.java:   74
            io.netty.util.concurrent.SingleThreadEventExecutor$4.run      SingleThreadEventExecutor.java:  997
                               io.netty.channel.nio.NioEventLoop.run                   NioEventLoop.java:  500
      io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks      SingleThreadEventExecutor.java:  470
          io.netty.util.concurrent.AbstractEventExecutor.safeExecute          AbstractEventExecutor.java:  167
              io.netty.util.concurrent.AbstractEventExecutor.runTask          AbstractEventExecutor.java:  174
               io.netty.channel.AbstractChannel$AbstractUnsafe$7.run                AbstractChannel.java:  813
         io.netty.channel.DefaultChannelPipeline.fireChannelInactive         DefaultChannelPipeline.java:  901
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive  AbstractChannelHandlerContext.java:  248
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive  AbstractChannelHandlerContext.java:  262
 io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive         DefaultChannelPipeline.java: 1405
  io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive  AbstractChannelHandlerContext.java:  241
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive  AbstractChannelHandlerContext.java:  248
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive  AbstractChannelHandlerContext.java:  262
                     io.netty.handler.ssl.SslHandler.channelInactive                     SslHandler.java: 1075
         io.netty.handler.codec.ByteToMessageDecoder.channelInactive           ByteToMessageDecoder.java:  357
      io.netty.handler.codec.ByteToMessageDecoder.channelInputClosed           ByteToMessageDecoder.java:  392
  io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive  AbstractChannelHandlerContext.java:  241
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive  AbstractChannelHandlerContext.java:  248
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive  AbstractChannelHandlerContext.java:  262
       aleph.tcp$server_channel_handler$reify__66154.channelInactive                      NO_SOURCE_FILE:   56
java.lang.NullPointerException: 
</pre>

The second one of these is spurious and should be fixed. 

Reproducer:

```clojure
(aleph.tcp/start-server
 (fn [_ _])
 {:port 12345
  :ssl-context (aleph.netty/self-signed-ssl-context)})
(s/put! @(aleph.tcp/client {:host "localhost" :port 12345}) "foobar")
```

Might also affect the HTTP server with TLS enabled. Needs to be determined.

Pull request:
Went for both because I had to fix #617 for being able to construct the regression test case for #618. Without it, the `:channel-inactive` event would not be propagated to the `notify-connection-closed` handler.

Patch:
diff --git a/src/aleph/netty.clj b/src/aleph/netty.clj
index de1d83e7..d205f9f2 100644
--- a/src/aleph/netty.clj
+++ b/src/aleph/netty.clj
@@ -232,9 +232,7 @@
      (d/error! d (CancellationException. "future is cancelled."))
 
      (some? (.cause f))
-     (if (instance? ClosedChannelException (.cause f))
-       (d/success! d false)
-       (d/error! d (.cause f)))
+     (d/error! d (.cause f))
 
      :else
      (d/error! d (IllegalStateException. "future in unknown state"))))
diff --git a/src/aleph/tcp.clj b/src/aleph/tcp.clj
index 2becd633..53759e26 100644
--- a/src/aleph/tcp.clj
+++ b/src/aleph/tcp.clj
@@ -53,8 +53,8 @@
 
       :channel-inactive
       ([_ ctx]
-        (s/close! @in)
-        (.fireChannelInactive ctx))
+       (some-> @in s/close!)
+       (.fireChannelInactive ctx))
 
       :channel-active
       ([_ ctx]


Test patch:
diff --git a/test/aleph/netty_test.clj b/test/aleph/netty_test.clj
new file mode 100644
index 00000000..e916b1b5
--- /dev/null
+++ b/test/aleph/netty_test.clj
@@ -0,0 +1,14 @@
+(ns aleph.netty-test
+  (:require
+   [aleph.netty :as netty]
+   [clojure.test :refer [deftest is]]
+   [manifold.stream :as s])
+  (:import
+   (io.netty.channel.embedded EmbeddedChannel)))
+
+(deftest closing-a-channel-sink
+  (let [ch (EmbeddedChannel.)
+        s (netty/sink ch)]
+    (is (= true @(s/put! s "foo")))
+    (is (nil? @(netty/wrap-future (netty/close ch))))
+    (is (= false @(s/put! s "foo")))))
diff --git a/test/aleph/ssl.clj b/test/aleph/ssl.clj
index 14ee2ad6..fae9d0fa 100644
--- a/test/aleph/ssl.clj
+++ b/test/aleph/ssl.clj
@@ -2,7 +2,6 @@
   (:require
    [aleph.netty :as netty])
   (:import
-   (io.netty.handler.ssl ClientAuth SslContextBuilder)
    (java.io ByteArrayInputStream)
    (java.security KeyFactory PrivateKey)
    (java.security.cert CertificateFactory X509Certificate)
diff --git a/test/aleph/tcp_ssl_test.clj b/test/aleph/tcp_ssl_test.clj
index 785352f2..0ae2d8b0 100644
--- a/test/aleph/tcp_ssl_test.clj
+++ b/test/aleph/tcp_ssl_test.clj
@@ -61,3 +61,22 @@
                                            :trust-store (into-array X509Certificate [ssl/ca-cert])})})]
         (is (nil? @(s/take! c)))
         (is (nil? @ssl-session) "SSL session should be undefined")))))
+
+(deftest test-connection-close-during-ssl-handshake
+  (let [ssl-session (atom nil)
+        connection-closed (promise)
+        notify-connection-closed #_:clj-kondo/ignore (netty/channel-handler
+                                                      :channel-inactive
+                                                      ([_ ctx]
+                                                       (deliver connection-closed true)
+                                                       (.fireChannelInactive ctx)))]
+    (with-server (tcp/start-server (ssl-echo-handler ssl-session)
+                                   {:port 10001
+                                    :ssl-context ssl/server-ssl-context
+                                    :pipeline-transform (fn [p]
+                                                          (.addLast p notify-connection-closed))})
+      (let [c @(tcp/client {:host "localhost"
+                            :port 10001})]
+        (s/close! c)
+        (is (deref connection-closed 1000 false))
+        (is (nil? @ssl-session) "SSL session should be undefined")))))