Merge pull request #623 from clj-commons/only-yield-tcp-ssl-client-stream-after-successful-handshake

Only yield client streams after successful SSL handshake

Author: DerGuteMoritz

src/aleph/http/client.clj

@@ -680,9 +680,12 @@
 
        :channel-active
        ([_ ctx]
-        (let [ch (.channel ctx)]
-          (reset! in (netty/buffered-source ch (constantly 1) 16))
-          (.handshake handshaker ch))
+        (-> (.channel ctx)
+            netty/maybe-ssl-handshake-future
+            (d/on-realized (fn [ch]
+                             (reset! in (netty/buffered-source ch (constantly 1) 16))
+                             (.handshake handshaker ch))
+                           netty/ignore-ssl-handshake-errors))
         (.fireChannelActive ctx))
 
        :user-event-triggered

src/aleph/netty.clj

@@ -1131,6 +1131,25 @@ initialize an DnsAddressResolverGroup instance.
   [dns-options]
   (DnsAddressResolverGroup. (dns-resolver-group-builder dns-options)))
 
+(defn ^:nodoc maybe-ssl-handshake-future
+  "Returns a deferred which resolves to the channel after a potential
+  SSL handshake has completed successfully. If no `SslHandler` is
+  present on the associated pipeline, resolves immediately."
+  [^Channel ch]
+  (if-let [^SslHandler ssl-handler (-> ch .pipeline (.get SslHandler))]
+    (-> ssl-handler
+        .handshakeFuture
+        wrap-future)
+    (d/success-deferred ch)))
+
+(defn ^:nodoc ignore-ssl-handshake-errors
+  "Intended for use as error callback on a `maybe-ssl-handshake-future`
+  within a `:channel-active` handler. In this context, SSL handshake
+  errors don't need to be handled since the SSL handler will terminate
+  the whole pipeline by throwing `javax.net.ssl.SSLHandshakeException`
+  anyway."
+  [_])
+
 (defn create-client
   ([pipeline-builder
     ssl-context
@@ -1197,7 +1216,7 @@ initialize an DnsAddressResolverGroup instance.
           (d/chain' (wrap-future f)
             (fn [_]
               (let [ch (.channel ^ChannelFuture f)]
-                ch))))))))
+                (maybe-ssl-handshake-future ch)))))))))
 
 (defn start-server
   [pipeline-builder

src/aleph/tcp.clj

@@ -28,16 +28,7 @@
 
 (defn- ^ChannelHandler server-channel-handler
   [handler {:keys [raw-stream?] :as options}]
-  (let [in (atom nil)
-        call-handler (fn [^ChannelHandlerContext ctx]
-                       (let [ch (.channel ctx)]
-                         (handler
-                          (doto
-                              (s/splice
-                               (netty/sink ch true netty/to-byte-buf)
-                               (reset! in (netty/source ch)))
-                            (reset-meta! {:aleph/channel ch}))
-                          (->TcpConnection ch))))]
+  (let [in (atom nil)]
     (netty/channel-inbound-handler
 
       :exception-caught
@@ -58,19 +49,16 @@
 
       :channel-active
       ([_ ctx]
-       (if-let [^SslHandler ssl-handler (-> ctx .pipeline (.get SslHandler))]
-         (-> ssl-handler
-             .handshakeFuture
-             netty/wrap-future
-             (d/on-realized (fn [_]
-                              (call-handler ctx))
-                            ;; No need to handle errors here since
-                            ;; the SSL handler will terminate the
-                            ;; whole pipeline by throwing a
-                            ;; `javax.net.ssl.SSLHandshakeException`
-                            ;; in this case anyway.
-                            (fn [_])))
-         (call-handler ctx))
+       (-> (.channel ctx)
+           netty/maybe-ssl-handshake-future
+           (d/on-realized (fn [ch]
+                            (handler
+                             (doto (s/splice
+                                    (netty/sink ch true netty/to-byte-buf)
+                                    (reset! in (netty/source ch)))
+                               (reset-meta! {:aleph/channel ch}))
+                             (->TcpConnection ch)))
+                          netty/ignore-ssl-handshake-errors))
        (.fireChannelActive ctx))
 
       :channel-read
@@ -129,19 +117,20 @@
 
        :channel-inactive
        ([_ ctx]
-         (s/close! @in)
+         (some-> @in s/close!)
          (.fireChannelInactive ctx))
 
        :channel-active
        ([_ ctx]
-         (let [ch (.channel ctx)]
-           (d/success! d
-             (doto
-               (s/splice
-                 (netty/sink ch true netty/to-byte-buf)
-                 (reset! in (netty/source ch)))
-               (reset-meta! {:aleph/channel ch}))))
-         (.fireChannelActive ctx))
+        (-> (.channel ctx)
+            netty/maybe-ssl-handshake-future
+            (d/on-realized (fn [ch]
+                             (d/success! d (doto (s/splice
+                                                  (netty/sink ch true netty/to-byte-buf)
+                                                  (reset! in (netty/source ch)))
+                                             (reset-meta! {:aleph/channel ch}))))
+                           netty/ignore-ssl-handshake-errors))
+        (.fireChannelActive ctx))
 
        :channel-read
        ([_ ctx msg]

test/aleph/tcp_ssl_test.clj

@@ -6,9 +6,11 @@
    [aleph.tcp-test :refer [with-server]]
    [clj-commons.byte-streams :as bs]
    [clojure.test :refer [deftest is]]
+   [manifold.deferred :as d]
    [manifold.stream :as s])
   (:import
-   (java.security.cert X509Certificate)))
+   (java.security.cert X509Certificate)
+   (java.util.concurrent TimeoutException)))
 
 (netty/leak-detector-level! :paranoid)
 
@@ -80,3 +82,26 @@
         (s/close! c)
         (is (deref connection-closed 1000 false))
         (is (nil? @ssl-session) "SSL session should be undefined")))))
+
+(deftest test-client-yields-stream-only-after-successful-handshake
+  (let [connection-active (promise)
+        continue-handshake (promise)
+        notify-connection-active #_:clj-kondo/ignore (netty/channel-inbound-handler
+                                                      :channel-active
+                                                      ([_ ctx]
+                                                       (deliver connection-active true)
+                                                       (when-not (deref continue-handshake 5000 false)
+                                                         (throw (TimeoutException.)))
+                                                       (.fireChannelActive ctx)))]
+    (with-server (tcp/start-server (fn [s _])
+                                   {:port 10001
+                                    :ssl-context ssl/server-ssl-context})
+      (let [c (tcp/client {:host "localhost"
+                           :port 10001
+                           :ssl-context ssl/client-ssl-context-opts
+                           :pipeline-transform (fn [p]
+                                                 (.addAfter p "handler" "notify-active" notify-connection-active))})]
+        (is (deref connection-active 1000 false))
+        (is (not (d/realized? c)))
+        (deliver continue-handshake true)
+        (is (deref c 1000 false))))))