From 32987c8704254db02ba65852a11f1cc5a36c586e Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Johan=20Andr=C3=A9n?= <johan@markatta.com>
Date: Tue, 16 Jan 2018 18:05:08 +0100
Subject: [PATCH] Simpler tls over tcp #24153

---
 .../src/main/paradox/stream/stream-io.md      |  15 +++
 .../java/akka/stream/javadsl/TcpTest.java     |  97 ++++++++++++---
 .../src/test/resources/tcp-spec-keystore.p12  | Bin 0 -> 3384 bytes
 .../test/scala/akka/stream/io/TcpSpec.scala   |  96 ++++++++++++++-
 .../main/scala/akka/stream/javadsl/Tcp.scala  |  82 ++++++++++++-
 .../main/scala/akka/stream/scaladsl/Tcp.scala | 116 +++++++++++++++++-
 6 files changed, 382 insertions(+), 24 deletions(-)
 create mode 100644 akka-stream-tests/src/test/resources/tcp-spec-keystore.p12

diff --git a/akka-docs/src/main/paradox/stream/stream-io.md b/akka-docs/src/main/paradox/stream/stream-io.md
index 6d92d4bb4c..c7f13ad864 100644
--- a/akka-docs/src/main/paradox/stream/stream-io.md
+++ b/akka-docs/src/main/paradox/stream/stream-io.md
@@ -123,6 +123,21 @@ see
 @java[[Javadoc](http://doc.akka.io/japi/akka/current/akka/stream/javadsl/Framing.html#simpleFramingProtocol-int-)]
 for more information.
 
+### TLS
+
+Similar factories as shown above for raw TCP but where the data is encrypted using TLS are available from `Tcp` through `outgoingTlsConnection`, `bindTls` and `bindAndHandleTls`, see the @scala[@scaladoc[`Tcp Scaladoc`](akka.stream.scaladsl.Tcp)]@java[@javadoc[`Tcp Javadoc`](akka.stream.javadsl.Tcp)]  for details.
+
+Using TLS requires a keystore and a truststore and then a somewhat involved dance of configuring the SSLContext and the details for how the session should be negotiated:
+
+Scala
+:  @@snip [TcpSpec.scala]($akka$akka-stream-tests/src/test/scala/akka/stream/io/TcpSpec.scala) { #setting-up-ssl-context }
+
+Java
+:  @@snip [TcpSpec.scala]($akka$akka-stream-tests/src/test/java/akka/stream/javadsl/TcpTest.java) { #setting-up-ssl-context }
+
+
+The `SslContext` and `NegotiateFirstSession` instances can then be used with the binding or outgoing connection factory methods.
+
 ## Streaming File IO
 
 Akka Streams provide simple Sources and Sinks that can work with `ByteString` instances to perform IO operations
diff --git a/akka-stream-tests/src/test/java/akka/stream/javadsl/TcpTest.java b/akka-stream-tests/src/test/java/akka/stream/javadsl/TcpTest.java
index aefbb49ff1..0763d1dbb7 100644
--- a/akka-stream-tests/src/test/java/akka/stream/javadsl/TcpTest.java
+++ b/akka-stream-tests/src/test/java/akka/stream/javadsl/TcpTest.java
@@ -3,34 +3,44 @@
  */
 package akka.stream.javadsl;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import akka.Done;
+import akka.actor.ActorSystem;
+import akka.japi.function.Function2;
+import akka.japi.function.Procedure;
+import akka.stream.BindFailedException;
+import akka.stream.StreamTcpException;
+import akka.stream.StreamTest;
+import akka.stream.javadsl.Tcp.IncomingConnection;
+import akka.stream.javadsl.Tcp.ServerBinding;
+import akka.testkit.AkkaJUnitActorSystemResource;
+import akka.testkit.AkkaSpec;
+import akka.testkit.SocketUtil;
+import akka.testkit.javadsl.EventFilter;
+import akka.testkit.javadsl.TestKit;
+import akka.util.ByteString;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import java.net.BindException;
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
-import java.net.BindException;
 
-import akka.Done;
-import akka.NotUsed;
-import akka.testkit.javadsl.EventFilter;
-import akka.testkit.javadsl.TestKit;
-import org.junit.ClassRule;
-import org.junit.Test;
-import scala.concurrent.Await;
-import scala.concurrent.Future;
-import scala.concurrent.duration.FiniteDuration;
-import scala.runtime.BoxedUnit;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
-import akka.stream.*;
-import akka.stream.javadsl.Tcp.*;
-import akka.japi.function.*;
-import akka.testkit.AkkaSpec;
-import akka.testkit.SocketUtil;
-import akka.util.ByteString;
-import akka.testkit.AkkaJUnitActorSystemResource;
+// #setting-up-ssl-context
+// imports
+import akka.stream.TLSClientAuth;
+import akka.stream.TLSProtocol;
+import com.typesafe.sslconfig.akka.AkkaSSLConfig;
+import java.security.KeyStore;
+import javax.net.ssl.*;
+import java.security.SecureRandom;
+// #setting-up-ssl-context
 
 public class TcpTest extends StreamTest {
   public TcpTest() {
@@ -126,4 +136,51 @@ public class TcpTest extends StreamTest {
     }
   }
 
+  // compile only sample
+  public void constructSslContext() throws Exception {
+    ActorSystem system = null;
+
+    // #setting-up-ssl-context
+
+    // -- setup logic ---
+
+    AkkaSSLConfig sslConfig = AkkaSSLConfig.get(system);
+
+    // Don't hardcode your password in actual code
+    char[] password = "abcdef".toCharArray();
+
+    // trust store and keys in one keystore
+    KeyStore keyStore = KeyStore.getInstance("PKCS12");
+    keyStore.load(getClass().getResourceAsStream("/tcp-spec-keystore.p12"), password);
+
+    TrustManagerFactory tmf = TrustManagerFactory.getInstance("SunX509");
+    tmf.init(keyStore);
+
+    KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance("SunX509");
+    keyManagerFactory.init(keyStore, password);
+
+    // initial ssl context
+    SSLContext sslContext = SSLContext.getInstance("TLS");
+    sslContext.init(keyManagerFactory.getKeyManagers(), tmf.getTrustManagers(), new SecureRandom());
+
+    // protocols
+    SSLParameters defaultParams = sslContext.getDefaultSSLParameters();
+    String[] defaultProtocols = defaultParams.getProtocols();
+    String[] protocols = sslConfig.configureProtocols(defaultProtocols, sslConfig.config());
+    defaultParams.setProtocols(protocols);
+
+    // ciphers
+    String[] defaultCiphers = defaultParams.getCipherSuites();
+    String[] cipherSuites = sslConfig.configureCipherSuites(defaultCiphers, sslConfig.config());
+    defaultParams.setCipherSuites(cipherSuites);
+
+    TLSProtocol.NegotiateNewSession negotiateNewSession = TLSProtocol.negotiateNewSession()
+        .withCipherSuites(cipherSuites)
+        .withProtocols(protocols)
+        .withParameters(defaultParams)
+        .withClientAuth(TLSClientAuth.none());
+
+    // #setting-up-ssl-context
+  }
+
 }
diff --git a/akka-stream-tests/src/test/resources/tcp-spec-keystore.p12 b/akka-stream-tests/src/test/resources/tcp-spec-keystore.p12
new file mode 100644
index 0000000000000000000000000000000000000000..d72cc9f3d4ecb9ee14918145b1a506a06b642054
GIT binary patch
literal 3384
zcmXqL;x%DnWHxBx`NhVm)#lOmotKfFaX}N$YnCRS=LStYj}4mG%Mns+3!2z744T+e
zm>3xhn%H9zGK>Z?Y+O(ico^9X_*gi03#?`7iZSbEV&Y(EXks@o%-VmOSKT$$_dw{`
zhHE#b8?BvkWX?xfPKG0AoK*H-yK=Q*@|}|8`j}4w%=P#3-aVLnVg1X+nLOr;?##3L
zbKgd~C27XDm{q+Q)33Wf?=tzWRGr8)>9Pp-2Ie!2M`x{Xh`jfOvtU}^2Gt#3)0)J2
zdRFtb7%6OAI8QxS!BkWE&b)x#vHv;N&uhuu>Ha3}R@US{&%6F}6jjum-!CTFmS~@I
z?yB=oAKAH+-NWB62rNw&xzW0FYt-!|)r?i`8og!*I<7XUMLWeklQ1`u`O3cALF|84
zeCv~x3r@Nd_O0AD>1)X;<Cc{oQst9PoBMNSN9hF>J#xPHVCe&gmG>-jZU+C}CH3f@
z`~s8DYtofpDXmIy&wg8MzD~Sq*|J?P?G7sygb7bOtvvbK%(#QE9hN)U=hb`s54m;D
z{>*ias)xZYrxzOLKU{l0^z7X9{XX7H@5Mf0z2<d5VMEZo%rc+j_xu|Vl^^MDGjVch
ze;sPs9uWR9wX<jAA=@cxdK(|ai1qS+>{8czySMQ{!n>w39npRFJDP9gn<y1~6<Y4x
zz59<~bFa*A5$&ZH7M^YQZU0tfdy=7=B}vO_%^mq?AD0~7weaez4f;yAH8-}ne_th>
z_17d}<)6tj!;3ugEuLsPe#m`qt3KIu)BlJ5Yz=9?R?h`=W;<K_nSAozO)qr~W@qh-
zUn2Gg-Fv*bU+?bmM<!Emhg@P>U4C!Efjhh$*7?F-<*)r8Yo@X(uPeH)caSmUPnuuz
ze@XqSqWyCwZwq*KG=}B#(zuJ~#1`dvL=^})zMfrNVQ@-3nTcP`$S-d9nu*NCb_Mgz
zE&m*HUB68x<m{)7vD~r0!czDBGg7jb{ruwWpC6^pi^O=WIQVv-Pv5|PtlFqPTJi23
ztxrmVmOT-lHiR-5uzY;E=l#*yTST>5e!I>uU$;W8Wa_f*MFksr|DXEn`^w_pj56jo
zd{s>owLiqgw+iVzabi*1P{O9qqS92irgwR^)=suB-Q_DM^)3@nINW}4a~E%L@Qy8#
z0%sO2+@<Umw$S@mX6mvz<;Nbo*4%DCw(6!W*X)z!A1`gu4&U@N*H>pM^P+3FxUZej
zH=Ne)sI%zlvYMHpEWLJ}$DaS)=6xwC^7%R25W8guHZT7wwJhNOZ2?smS0*>+zR2$M
zSuIKrOqEaE$*fi1Y#~xDBO=^ex8c}Ht;ZK16!U$Z<2fZfrB3E!M_4MiXt?}`m4`*<
zML9%Ve)d0G*n+2iYjNzhb6>?Hlbw%TQxrD5R<9wkPVa5mwteU87O3(56f<frv##3z
ze}B(ciF-S@udm#-<<0wWe%{|FZoHqVQM<UfEor03!OJK9R6S&WAF|@igCE@Lyuo*}
zzf9@Qy*o?r!?C?GJuBxVeu^^9ef#B2{KbN}+d_sF0wKa~oAe|+gSj{_|6S|%EJRD>
z?9OEwmtWlYq0~^RbF<pRraRyoi(5vyno!akO_9TXhXvU<k^|J$wrHAoO?)V@Cxmgq
zQq8*ut-TMQu3pE)w40}WdVTku1CKBKEIWPq3%_&b*SFmUT``kV;~1?~4)%XaKibj}
zv&E`6|7Dj~*}Rj2=T@8XFt};H3p#jHfT1Y)%%T-98%n;a#JnzAbi-rHF-y&vJ=y6l
zF<xe6bN6t~*<~rBnim+%{PnxZ$r3|Pix5Rs-RAUzV*9pU{Z_o$?au#F(JS|+n2P>R
z=MS!1H0_tyUzHTO_OC$)-`B_0)F|9|#3kVH?Y&miEj!zz+>J_hyiC)tXfK|bZtY&W
zCq!s|h1hBjBgd2^`yKK}Z%o{t|MyV+`(XYJJ2or*etl2vxcQb}ij2MjS|17?#;`5g
z>w58`Sku2Qi+e9733qI3;hXdG^8RoAVs*<e%t%NrQB0GXz41jt^^Oup(F@b01Rn_f
zTzELpDDv~_>+c(sngsT*;hbNU6K8vP!-`XP8H_vml^wh`nyH=aGW@qpzIe^Nrs*}@
z5y7dEm2<xxk2?5*aY?-A^94KCDqWSBvnp3>VbgUlC!bU^kB5apeVhj#uw^gMm9h<f
z6)}bLhv31_lC3{pJBArA=;z5QmTKMdO6{9_Ngj`A%(a>KOW4)?KPT;IZgB0lvf;7i
z<`Op#ljYfcm!+^?O?W{p->vYk%ipA2TFi8+AYtA?o5KtBPJWf$qPD8$y78usr+<9i
zc-2m>>f+p-)$>%_k9s}0@F{MQLHC61{5Fqjj`}v+och0Q!!4nec7|&c&Oa$@i+#qH
zb+07B^S$!JmA&bEYS+Bq^0$k>L}d5H6|75ooLarUymY-T*3VNFuUecXWYrSoWPjhX
zY0cmD>DO~NsOewle>3;a3dQEQU1{CRPO7eFo${#Y4zGH5mGYw26#|SB?K~H%zu#}@
z|JO3d|3W;YqyLAzPKl6M-++MIXW5qi{_;fX_xZQ&(Js2Tb{_3~^t)xbOZ~|?ytf2O
zJJtznD7^PM8SZWPCftbY$O4()G3&fkysylvZ2x@xr;@7K&Ns`>zF%`C??zg&MQp+E
zR!4&-))b@)c0m(sC`%J-pg|L>uR#;5CmT1cy5(YITF}Ju&Y+3q6-q744623ytDd-i
zC%G^fR132-v7DH6+g2=M)71t1<)tMl)1nlQ)qI$nxUO6CY4kzIkLB}w`d>{ucGaeR
zu|k)X>kFliXJ?5&IbM}##>lK(oVv%`b*ZA+a<2#p^9OIZtCwCbf8;)6Qqjb$k`D^)
zo6>j~m({i=t==K6^qY&<v~;42C)d^v^OsF??amz&(%qDLB&IaA?&jI~PTy>0`L8!=
z_Pw}O+E8@YHZ-L$Wuy4^lSZ3F&ooFmB{a-cEQ!2r{#hjVVMI8mTB*48Um4}5uwI?D
z&X51NtozCS?+5EMkB*gVl(%tnFJHn@D`YKyj@36;?V<1X_tzwaPHC#oaq?KGber$L
zV1RAO!Vg!jdiU0ew^sJ8+u`3?z#j8_?+aUYnI4AS+PpcOZr5gJSgtqH{&0j{examQ
zZe)Vcqm+rY(FZ=}tFFvY2@0S0sYHqSPSn4hmu_uj6gs=7Jp4|z(#@y*bv&=1u3#|R
z^i)Ll<NcklId^rO{U+2h)nwzFu&M9EE=m0onq(l6(5)J(zq(9V=;^%!`=<S=dZD1W
z&qYeEQuw|5l7E*DN3Hq)@NQU|&Ro5fMU}ZMxsxVG?|jzR>>{zZ)YDLK6{nQ_Ki%Zu
z!VaZI|CD&Q%75C~Y7HxQFEyUHwQ(o^`n}Cp6jnxya;UMcn<ARiJ#*oqY)?PC(8Scp
z!e;HLy=yL;mh_ezUsC-ns(<Fpd26d<4-;y?u38Y;{pja<#*`0AaZv~37kb4pU3swK
z0n?Xd6S6(z-^RCni9h>iN_$a6-sc~`EqC)MvAkr>N$lOL_sC4hQ|{8~?>T3mTnc|z
zc5|IuzS+l?_krm<@6TnF;yL{0`{8Fc^A6=7uDQu@ciB4DjZ2QkU%6oPYE`lPx{v_=
z$&1SWo|T;%eceYV*g<{s-qy4vjx4>|;YzjH?6+bjzInA$>^^UQq()(uQ^n`*l!a|+
zTdL1aJGUoJ?m@>b>-3B3KCSar|C;mt^sd~?A6BlNxjo2j=XT2lDX~V2IX-VsV?Wi`
zWVim>Q7);6D~w*8`|Xo?cDE}F<EQkMf7dMXNvjJw=*d3AX3m%P|LToT<KqjL-(S7S
zmd{j}Ph=|7#E;A~gAymS?mntyY`=kZbIFreOAkt&p0e+1#FU(SzqZCalzTU6@tkLI
z2iN=$xSb!Jdo{dLEXn59wCSHU@87={d-l$~%@+11eSVJ*{@i@;fO*d~rnx3F9F4u-
z$d*l9kdqK~S6)PI-#@N{4>Mzo3zjsmsgt+y-ui6jrkKSuf5<q-MkIvo&3JqK;w!22
z+3UorgCAF%lYF_d*wi{>-ZaJH)!7%dShPO+3JFbEalYo7#LB>3+yCvJAQG<I?zbmp
ztzn0G)R+FCOSRfwE*2@ADYo;Z%&s|nPteVO<5F;M>w(yWYjruwS>4YbR4bP)xV5EP
z_(4<hnNnS!iP!Zr@9Gr#$vDha<9=57?2K*qoHtuc&NkV&>fK@AVEp*$hr_*Is{@xu
z{>{kxxo)M%RCTFzow75zea@<nigK!0*0G-7Kjp38S2x+J-u7P!iIzn(gTK~Ic5;3A
z_prScw~Nl{g}h2@O9i;L9eC(>E-Y@nXl;@89Vd=l%NyH1DL>3<K9~D^JtMQuwG-}#
z?>!8)Te@5>%<Fh0UrVXBfa^5Q_BqSLnf9*{i1KAD&*FR{|NH+w3*$qAMoo5KnnjNo
zdKf6eOGi!-LlHS4hGK?Ph9ZVCFqzCy#86_Oj8GzKD8?d^nY@fke5FWbpuX~oMD50!
ztVTy~149Ev15P$pZ9ZluDOLs+5fi=Y25+06mvp)!#dQSshkST&E}ezLL9|r)sd-B!
Hs2~LZ7uZ55

literal 0
HcmV?d00001

diff --git a/akka-stream-tests/src/test/scala/akka/stream/io/TcpSpec.scala b/akka-stream-tests/src/test/scala/akka/stream/io/TcpSpec.scala
index 7171e51c15..e2dea0b4a7 100644
--- a/akka-stream-tests/src/test/scala/akka/stream/io/TcpSpec.scala
+++ b/akka-stream-tests/src/test/scala/akka/stream/io/TcpSpec.scala
@@ -4,7 +4,9 @@
 package akka.stream.io
 
 import java.net._
+import java.security.SecureRandom
 import java.util.concurrent.atomic.AtomicInteger
+import javax.net.ssl.{ KeyManagerFactory, SSLContext, TrustManagerFactory }
 
 import akka.actor.{ ActorIdentity, ActorSystem, ExtendedActorSystem, Identify, Kill }
 import akka.io.Tcp._
@@ -18,6 +20,7 @@ import akka.testkit.SocketUtil.temporaryServerAddress
 import akka.util.ByteString
 import akka.{ Done, NotUsed }
 import com.typesafe.config.ConfigFactory
+import org.scalatest.concurrent.PatienceConfiguration
 import org.scalatest.concurrent.PatienceConfiguration.Timeout
 
 import scala.collection.immutable
@@ -25,7 +28,10 @@ import scala.concurrent.duration._
 import scala.concurrent.{ Await, Future, Promise }
 import scala.util.control.NonFatal
 
-class TcpSpec extends StreamSpec("akka.stream.materializer.subscription-timeout.timeout = 2s") with TcpHelper {
+class TcpSpec extends StreamSpec("""
+    akka.loglevel = info
+    akka.stream.materializer.subscription-timeout.timeout = 2s
+  """) with TcpHelper {
 
   "Outgoing TCP stream" must {
 
@@ -692,6 +698,94 @@ class TcpSpec extends StreamSpec("akka.stream.materializer.subscription-timeout.
     }
   }
 
+  "TLS client and server convenience methods" should {
+
+    "allow for 'simple' TLS" in {
+      // cert is valid until 2025, so if this tests starts failing after that you need to create a new one
+      val (sslContext, firstSession) = initSslMess()
+      val address = temporaryServerAddress()
+
+      Tcp().bindAndHandleTls(
+        // just echo charactes until we reach '\n', then complete stream
+        // also - byte is our framing
+        Flow[ByteString].mapConcat(_.utf8String.toList)
+          .takeWhile(_ != '\n')
+          .map(c ⇒ ByteString(c)),
+        address.getHostName,
+        address.getPort,
+        sslContext,
+        firstSession
+      ).futureValue
+      system.log.info(s"Server bound to ${address.getHostString}:${address.getPort}")
+
+      val connectionFlow = Tcp().outgoingTlsConnection(address.getHostName, address.getPort, sslContext, firstSession)
+
+      val chars = "hello\n".toList.map(_.toString)
+      val (connectionF, result) =
+        Source(chars).map(c ⇒ ByteString(c))
+          .concat(Source.maybe) // do not complete it from our side
+          .viaMat(connectionFlow)(Keep.right)
+          .map(_.utf8String)
+          .toMat(Sink.fold("")(_ + _))(Keep.both)
+          .run()
+
+      connectionF.futureValue
+      system.log.info(s"Client connected to ${address.getHostString}:${address.getPort}")
+
+      result.futureValue(PatienceConfiguration.Timeout(10.seconds)) should ===("hello")
+    }
+
+    def initSslMess() = {
+      // #setting-up-ssl-context
+      import akka.stream.TLSClientAuth
+      import akka.stream.TLSProtocol
+      import com.typesafe.sslconfig.akka.AkkaSSLConfig
+      import java.security.KeyStore
+      import javax.net.ssl._
+
+      val sslConfig = AkkaSSLConfig()
+
+      // Don't hardcode your password in actual code
+      val password = "abcdef".toCharArray
+
+      // trust store and keys in one keystore
+      val keyStore = KeyStore.getInstance("PKCS12")
+      keyStore.load(classOf[TcpSpec].getResourceAsStream("/tcp-spec-keystore.p12"), password)
+
+      val tmf = TrustManagerFactory.getInstance("SunX509")
+      tmf.init(keyStore)
+
+      val keyManagerFactory = KeyManagerFactory.getInstance("SunX509")
+      keyManagerFactory.init(keyStore, password)
+
+      // initial ssl context
+      val sslContext = SSLContext.getInstance("TLS")
+      sslContext.init(keyManagerFactory.getKeyManagers, tmf.getTrustManagers, new SecureRandom)
+
+      // protocols
+      val defaultParams = sslContext.getDefaultSSLParameters
+      val defaultProtocols = defaultParams.getProtocols
+      val protocols = sslConfig.configureProtocols(defaultProtocols, sslConfig.config)
+      defaultParams.setProtocols(protocols)
+
+      // ciphers
+      val defaultCiphers = defaultParams.getCipherSuites
+      val cipherSuites = sslConfig.configureCipherSuites(defaultCiphers, sslConfig.config)
+      defaultParams.setCipherSuites(cipherSuites)
+
+      val negotiateNewSession = TLSProtocol.NegotiateNewSession
+        .withCipherSuites(cipherSuites: _*)
+        .withProtocols(protocols: _*)
+        .withParameters(defaultParams)
+        .withClientAuth(TLSClientAuth.None)
+
+      // #setting-up-ssl-context
+
+      (sslContext, negotiateNewSession)
+    }
+
+  }
+
   def validateServerClientCommunication(
     testData:         ByteString,
     serverConnection: ServerConnection,
diff --git a/akka-stream/src/main/scala/akka/stream/javadsl/Tcp.scala b/akka-stream/src/main/scala/akka/stream/javadsl/Tcp.scala
index 93d5be32d4..46bf553dc3 100644
--- a/akka-stream/src/main/scala/akka/stream/javadsl/Tcp.scala
+++ b/akka-stream/src/main/scala/akka/stream/javadsl/Tcp.scala
@@ -24,8 +24,10 @@ import akka.io.Inet.SocketOption
 import scala.compat.java8.OptionConverters._
 import scala.compat.java8.FutureConverters._
 import java.util.concurrent.CompletionStage
+import javax.net.ssl.SSLContext
 
-import akka.annotation.InternalApi
+import akka.annotation.{ ApiMayChange, InternalApi }
+import akka.stream.TLSProtocol.NegotiateNewSession
 
 object Tcp extends ExtensionId[Tcp] with ExtensionIdProvider {
 
@@ -199,4 +201,82 @@ class Tcp(system: ExtendedActorSystem) extends akka.actor.Extension {
     Flow.fromGraph(delegate.outgoingConnection(new InetSocketAddress(host, port))
       .mapMaterializedValue(_.map(new OutgoingConnection(_))(ec).toJava))
 
+  /**
+   * Creates an [[Tcp.OutgoingConnection]] with TLS.
+   * The returned flow represents a TCP client connection to the given endpoint where all bytes in and
+   * out go through TLS.
+   *
+   * @see [[Tcp.outgoingConnection()]]
+   */
+  def outgoingTlsConnection(host: String, port: Int, sslContext: SSLContext, negotiateNewSession: NegotiateNewSession): Flow[ByteString, ByteString, CompletionStage[OutgoingConnection]] =
+    Flow.fromGraph(delegate.outgoingTlsConnection(host, port, sslContext, negotiateNewSession)
+      .mapMaterializedValue(_.map(new OutgoingConnection(_))(ec).toJava))
+
+  /**
+   * Creates an [[Tcp.OutgoingConnection]] with TLS.
+   * The returned flow represents a TCP client connection to the given endpoint where all bytes in and
+   * out go through TLS.
+   *
+   * @see [[Tcp.outgoingConnection()]]
+   *
+   * Marked API-may-change to leave room for an improvement around the very long parameter list.
+   */
+  @ApiMayChange
+  def outgoingTlsConnection(
+    remoteAddress:       InetSocketAddress,
+    sslContext:          SSLContext,
+    negotiateNewSession: NegotiateNewSession,
+    localAddress:        Optional[InetSocketAddress],
+    options:             JIterable[SocketOption],
+    connectTimeout:      Duration,
+    idleTimeout:         Duration
+  ): Flow[ByteString, ByteString, CompletionStage[OutgoingConnection]] =
+    Flow.fromGraph(delegate.outgoingTlsConnection(
+      remoteAddress,
+      sslContext,
+      negotiateNewSession,
+      localAddress.asScala,
+      immutableSeq(options),
+      connectTimeout,
+      idleTimeout)
+      .mapMaterializedValue(_.map(new OutgoingConnection(_))(ec).toJava))
+
+  /**
+   * Creates a [[Tcp.ServerBinding]] instance which represents a prospective TCP server binding on the given `endpoint`
+   * where all incoming and outgoing bytes are passed through TLS.
+   *
+   * @see [[Tcp.bind()]]
+   * Marked API-may-change to leave room for an improvement around the very long parameter list.
+   */
+  @ApiMayChange
+  def bindTls(
+    interface:           String,
+    port:                Int,
+    sslContext:          SSLContext,
+    negotiateNewSession: NegotiateNewSession,
+    backlog:             Int,
+    options:             JIterable[SocketOption],
+    halfClose:           Boolean,
+    idleTimeout:         Duration
+  ): Source[IncomingConnection, CompletionStage[ServerBinding]] =
+    Source.fromGraph(delegate.bindTls(interface, port, sslContext, negotiateNewSession, backlog, immutableSeq(options), idleTimeout)
+      .map(new IncomingConnection(_))
+      .mapMaterializedValue(_.map(new ServerBinding(_))(ec).toJava))
+
+  /**
+   * Creates a [[Tcp.ServerBinding]] instance which represents a prospective TCP server binding on the given `endpoint`
+   * where all incoming and outgoing bytes are passed through TLS.
+   *
+   * @see [[Tcp.bind()]]
+   */
+  def bindTls(
+    interface:           String,
+    port:                Int,
+    sslContext:          SSLContext,
+    negotiateNewSession: NegotiateNewSession
+  ): Source[IncomingConnection, CompletionStage[ServerBinding]] =
+    Source.fromGraph(delegate.bindTls(interface, port, sslContext, negotiateNewSession)
+      .map(new IncomingConnection(_))
+      .mapMaterializedValue(_.map(new ServerBinding(_))(ec).toJava))
+
 }
diff --git a/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala b/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala
index 6ce003e763..47f72e2d13 100644
--- a/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala
+++ b/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala
@@ -5,16 +5,18 @@ package akka.stream.scaladsl
 
 import java.net.InetSocketAddress
 import java.util.concurrent.TimeoutException
+import javax.net.ssl.SSLContext
 
-import akka.{ Done, NotUsed }
 import akka.actor._
-import akka.annotation.InternalApi
+import akka.annotation.{ ApiMayChange, InternalApi }
 import akka.io.Inet.SocketOption
 import akka.io.{ IO, Tcp ⇒ IoTcp }
+import akka.stream.TLSProtocol.NegotiateNewSession
 import akka.stream._
 import akka.stream.impl.fusing.GraphStages.detacher
 import akka.stream.impl.io.{ ConnectionSourceStage, OutgoingConnectionStage, TcpIdleTimeout }
 import akka.util.ByteString
+import akka.{ Done, NotUsed }
 
 import scala.collection.immutable
 import scala.concurrent.Future
@@ -70,6 +72,15 @@ object Tcp extends ExtensionId[Tcp] with ExtensionIdProvider {
   def lookup() = Tcp
 
   def createExtension(system: ExtendedActorSystem): Tcp = new Tcp(system)
+
+  // just wraps/unwraps the TLS byte events to provide ByteString, ByteString flows
+  private val tlsWrapping: BidiFlow[ByteString, TLSProtocol.SendBytes, TLSProtocol.SslTlsInbound, ByteString, NotUsed] = BidiFlow.fromFlows(
+    Flow[ByteString].map(TLSProtocol.SendBytes),
+    Flow[TLSProtocol.SslTlsInbound].collect {
+      case sb: TLSProtocol.SessionBytes ⇒ sb.bytes
+      // ignore other kinds of inbounds (currently only Truncated)
+    }
+  )
 }
 
 final class Tcp(system: ExtendedActorSystem) extends akka.actor.Extension {
@@ -208,6 +219,107 @@ final class Tcp(system: ExtendedActorSystem) extends akka.actor.Extension {
    */
   def outgoingConnection(host: String, port: Int): Flow[ByteString, ByteString, Future[OutgoingConnection]] =
     outgoingConnection(InetSocketAddress.createUnresolved(host, port))
+
+  /**
+   * Creates an [[Tcp.OutgoingConnection]] with TLS.
+   * The returned flow represents a TCP client connection to the given endpoint where all bytes in and
+   * out go through TLS.
+   *
+   * For more advanced use cases you can manually combine [[Tcp.outgoingConnection()]] and [[TLS]]
+   *
+   * @param negotiateNewSession Details about what to require when negotiating the connection with the server
+   * @param sslContext Context containing details such as the trust and keystore
+   *
+   * @see [[Tcp.outgoingConnection()]]
+   */
+  def outgoingTlsConnection(
+    host:                String,
+    port:                Int,
+    sslContext:          SSLContext,
+    negotiateNewSession: NegotiateNewSession): Flow[ByteString, ByteString, Future[OutgoingConnection]] =
+    outgoingTlsConnection(InetSocketAddress.createUnresolved(host, port), sslContext, negotiateNewSession)
+
+  /**
+   * Creates an [[Tcp.OutgoingConnection]] with TLS.
+   * The returned flow represents a TCP client connection to the given endpoint where all bytes in and
+   * out go through TLS.
+   *
+   * @see [[Tcp.outgoingConnection()]]
+   * @param negotiateNewSession Details about what to require when negotiating the connection with the server
+   * @param sslContext Context containing details such as the trust and keystore
+   *
+   * Marked API-may-change to leave room for an improvement around the very long parameter list.
+   */
+  @ApiMayChange
+  def outgoingTlsConnection(
+    remoteAddress:       InetSocketAddress,
+    sslContext:          SSLContext,
+    negotiateNewSession: NegotiateNewSession,
+    localAddress:        Option[InetSocketAddress]           = None,
+    options:             immutable.Traversable[SocketOption] = Nil,
+    connectTimeout:      Duration                            = Duration.Inf,
+    idleTimeout:         Duration                            = Duration.Inf): Flow[ByteString, ByteString, Future[OutgoingConnection]] = {
+
+    val connection = outgoingConnection(remoteAddress, localAddress, options, true, connectTimeout, idleTimeout)
+    val tls = TLS(sslContext, negotiateNewSession, TLSRole.client)
+    connection.join(tlsWrapping.atop(tls).reversed)
+  }
+
+  /**
+   * Creates a [[Tcp.ServerBinding]] instance which represents a prospective TCP server binding on the given `endpoint`
+   * where all incoming and outgoing bytes are passed through TLS.
+   *
+   * @param negotiateNewSession Details about what to require when negotiating the connection with the server
+   * @param sslContext Context containing details such as the trust and keystore
+   * @see [[Tcp.bind]]
+   *
+   * Marked API-may-change to leave room for an improvement around the very long parameter list.
+   */
+  @ApiMayChange
+  def bindTls(
+    interface:           String,
+    port:                Int,
+    sslContext:          SSLContext,
+    negotiateNewSession: NegotiateNewSession,
+    backlog:             Int                                 = 100,
+    options:             immutable.Traversable[SocketOption] = Nil,
+    idleTimeout:         Duration                            = Duration.Inf): Source[IncomingConnection, Future[ServerBinding]] = {
+
+    val tls = tlsWrapping.atop(TLS(sslContext, negotiateNewSession, TLSRole.server)).reversed
+
+    bind(interface, port, backlog, options, true, idleTimeout).map { incomingConnection ⇒
+      incomingConnection.copy(
+        flow = incomingConnection.flow.join(tls)
+      )
+    }
+  }
+
+  /**
+   * Creates a [[Tcp.ServerBinding]] instance which represents a prospective TCP server binding on the given `endpoint`
+   * handling the incoming connections through TLS and then run using the provided Flow.
+   *
+   * @param negotiateNewSession Details about what to require when negotiating the connection with the server
+   * @param sslContext Context containing details such as the trust and keystore
+   * @see [[Tcp.bindAndHandle()]]
+   *
+   * Marked API-may-change to leave room for an improvement around the very long parameter list.
+   */
+  @ApiMayChange
+  def bindAndHandleTls(
+    handler:             Flow[ByteString, ByteString, _],
+    interface:           String,
+    port:                Int,
+    sslContext:          SSLContext,
+    negotiateNewSession: NegotiateNewSession,
+    backlog:             Int                                 = 100,
+    options:             immutable.Traversable[SocketOption] = Nil,
+    idleTimeout:         Duration                            = Duration.Inf)(implicit m: Materializer): Future[ServerBinding] = {
+    bindTls(interface, port, sslContext, negotiateNewSession, backlog, options, idleTimeout)
+      .to(Sink.foreach { conn: IncomingConnection ⇒
+        conn.handleWith(handler)
+      }).run()
+  }
+
 }
 
 final class TcpIdleTimeoutException(msg: String, timeout: Duration)