raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

=htc #20214 disabling SNI per-connection now works (passing though sslconfig) (#20621)

Browse source
This commit is contained in:
Konrad Malawski 2016-06-02 13:13:11 +02:00
parent c75504c3b6
commit b9ce2c94f6
10 changed files with 175 additions and 47 deletions
Download patch file Download diff file Expand all files Collapse all files

21
akka-http-core/src/main/scala/akka/http/javadsl/ConnectionContext.scala
View file

@ -9,6 +9,7 @@ import akka.http.scaladsl
import akka.japi.Util
import akka.stream.TLSClientAuth
import akka.http.impl.util.JavaMapping.Implicits._
import com.typesafe.sslconfig.akka.AkkaSSLConfig
import scala.compat.java8.OptionConverters
import scala.collection.JavaConverters._
@ -21,6 +22,23 @@ object ConnectionContext {
scaladsl.ConnectionContext.https(sslContext)
/** Used to serve HTTPS traffic. */
def https(sslContext: SSLContext,
sslConfig: Optional[AkkaSSLConfig],
enabledCipherSuites: Optional[JCollection[String]],
enabledProtocols: Optional[JCollection[String]],
clientAuth: Optional[TLSClientAuth],
sslParameters: Optional[SSLParameters]) =
scaladsl.ConnectionContext.https(
sslContext,
OptionConverters.toScala(sslConfig),
OptionConverters.toScala(enabledCipherSuites).map(Util.immutableSeq(_)),
OptionConverters.toScala(enabledProtocols).map(Util.immutableSeq(_)),
OptionConverters.toScala(clientAuth),
OptionConverters.toScala(sslParameters))
//#https-context-creation
/** Used to serve HTTPS traffic. */
// for binary-compatibility, since 2.4.7
def https(sslContext: SSLContext,
enabledCipherSuites: Optional[JCollection[String]],
enabledProtocols: Optional[JCollection[String]],
@ -32,7 +50,6 @@ object ConnectionContext {
OptionConverters.toScala(enabledProtocols).map(Util.immutableSeq(_)),
OptionConverters.toScala(clientAuth),
OptionConverters.toScala(sslParameters))
//#https-context-creation
/** Used to serve HTTP traffic. */
def noEncryption(): HttpConnectionContext =
@ -43,11 +60,13 @@ abstract class ConnectionContext {
def isSecure: Boolean
/** Java API */
def getDefaultPort: Int
def sslConfig: Option[AkkaSSLConfig]
}
abstract class HttpConnectionContext extends akka.http.javadsl.ConnectionContext {
override final def isSecure = false
override final def getDefaultPort = 80
override def sslConfig: Option[AkkaSSLConfig] = None
}
abstract class HttpsConnectionContext extends akka.http.javadsl.ConnectionContext {

25
akka-http-core/src/main/scala/akka/http/scaladsl/ConnectionContext.scala
View file

@ -6,6 +6,7 @@ package akka.http.scaladsl
import akka.stream.TLSClientAuth
import akka.stream.TLSProtocol._
import com.typesafe.sslconfig.akka.AkkaSSLConfig
import scala.collection.JavaConverters._
import java.util.{ Optional, Collection ⇒ JCollection }
@ -22,25 +23,43 @@ object ConnectionContext {
//#https-context-creation
// ConnectionContext
def https(sslContext: SSLContext,
sslConfig: Option[AkkaSSLConfig] = None,
enabledCipherSuites: Option[immutable.Seq[String]] = None,
enabledProtocols: Option[immutable.Seq[String]] = None,
clientAuth: Option[TLSClientAuth] = None,
sslParameters: Option[SSLParameters] = None) = {
new HttpsConnectionContext(sslContext, enabledCipherSuites, enabledProtocols, clientAuth, sslParameters)
}
sslParameters: Option[SSLParameters] = None) =
new HttpsConnectionContext(sslContext, sslConfig, enabledCipherSuites, enabledProtocols, clientAuth, sslParameters)
//#https-context-creation
// for binary-compatibility, since 2.4.7
def https(sslContext: SSLContext,
enabledCipherSuites: Option[immutable.Seq[String]],
enabledProtocols: Option[immutable.Seq[String]],
clientAuth: Option[TLSClientAuth],
sslParameters: Option[SSLParameters]) =
new HttpsConnectionContext(sslContext, None, enabledCipherSuites, enabledProtocols, clientAuth, sslParameters)
def noEncryption() = HttpConnectionContext
}
final class HttpsConnectionContext(
val sslContext: SSLContext,
val sslConfig: Option[AkkaSSLConfig] = None,
val enabledCipherSuites: Option[immutable.Seq[String]] = None,
val enabledProtocols: Option[immutable.Seq[String]] = None,
val clientAuth: Option[TLSClientAuth] = None,
val sslParameters: Option[SSLParameters] = None)
extends akka.http.javadsl.HttpsConnectionContext with ConnectionContext {
// for binary-compatibility, since 2.4.7
def this(
sslContext: SSLContext,
enabledCipherSuites: Option[immutable.Seq[String]],
enabledProtocols: Option[immutable.Seq[String]],
clientAuth: Option[TLSClientAuth],
sslParameters: Option[SSLParameters]) =
this(sslContext, None, enabledCipherSuites, enabledProtocols, clientAuth, sslParameters)
def firstSession = NegotiateNewSession(enabledCipherSuites, enabledProtocols, clientAuth, sslParameters)
override def getSslContext = sslContext

4
akka-http-core/src/main/scala/akka/http/scaladsl/Http.scala
View file

@ -587,7 +587,7 @@ class HttpExt(private val config: Config)(implicit val system: ActorSystem) exte
/** Creates real or placebo SslTls stage based on if ConnectionContext is HTTPS or not. */
private[http] def sslTlsStage(connectionContext: ConnectionContext, role: TLSRole, hostInfo: Option[(String, Int)] = None) =
connectionContext match {
case hctx: HttpsConnectionContext ⇒ TLS(hctx.sslContext, hctx.firstSession, role, hostInfo = hostInfo)
case hctx: HttpsConnectionContext ⇒ TLS(hctx.sslContext, connectionContext.sslConfig, hctx.firstSession, role, hostInfo = hostInfo)
case other ⇒ TLSPlacebo() // if it's not HTTPS, we don't enable SSL/TLS
}
@ -815,7 +815,7 @@ trait DefaultSSLContextCreation {
defaultParams.setEndpointIdentificationAlgorithm("https")
}
new HttpsConnectionContext(sslContext, Some(cipherSuites.toList), Some(defaultProtocols.toList), clientAuth, Some(defaultParams))
new HttpsConnectionContext(sslContext, Some(sslConfig), Some(cipherSuites.toList), Some(defaultProtocols.toList), clientAuth, Some(defaultParams))
}
}

9
akka-stream-tests/src/test/scala/akka/stream/io/TlsSpec.scala
View file

@ -5,6 +5,7 @@ import java.security.SecureRandom
import java.util.concurrent.TimeoutException
import akka.NotUsed
import com.typesafe.sslconfig.akka.AkkaSSLConfig
import scala.collection.immutable
import scala.concurrent.Await
@ -90,6 +91,8 @@ class TlsSpec extends AkkaSpec("akka.loglevel=INFO\nakka.actor.debug.receive=off
import GraphDSL.Implicits._
val sslConfig: Option[AkkaSSLConfig] = None // no special settings to be applied here
"SslTls" must {
val sslContext = initSslContext()
@ -103,9 +106,9 @@ class TlsSpec extends AkkaSpec("akka.loglevel=INFO\nakka.actor.debug.receive=off
}
val cipherSuites = NegotiateNewSession.withCipherSuites("TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA", "TLS_RSA_WITH_AES_128_CBC_SHA")
def clientTls(closing: TLSClosing) = TLS(sslContext, cipherSuites, Client, closing)
def badClientTls(closing: TLSClosing) = TLS(initWithTrust("/badtruststore"), cipherSuites, Client, closing)
def serverTls(closing: TLSClosing) = TLS(sslContext, cipherSuites, Server, closing)
def clientTls(closing: TLSClosing) = TLS(sslContext, None, cipherSuites, Client, closing)
def badClientTls(closing: TLSClosing) = TLS(initWithTrust("/badtruststore"), None, cipherSuites, Client, closing)
def serverTls(closing: TLSClosing) = TLS(sslContext, None, cipherSuites, Server, closing)
trait Named {
def name: String =

2
akka-stream/src/main/scala/akka/stream/impl/ActorMaterializerImpl.scala
View file

@ -121,7 +121,7 @@ private[akka] case class ActorMaterializerImpl(system: ActorSystem,
case tls: TlsModule ⇒ // TODO solve this so TlsModule doesn't need special treatment here
val es = effectiveSettings(effectiveAttributes)
val props =
TLSActor.props(es, tls.sslContext, tls.firstSession, tls.role, tls.closing, tls.hostInfo)
TLSActor.props(es, tls.sslContext, tls.sslConfig, tls.firstSession, tls.role, tls.closing, tls.hostInfo)
val impl = actorOf(props, stageName(effectiveAttributes), es.dispatcher)
def factory(id: Int) = new ActorPublisher[Any](impl) {
override val wakeUpMsg = FanOut.SubstreamSubscribePending(id)

26
akka-stream/src/main/scala/akka/stream/impl/io/TLSActor.scala
View file

@ -27,12 +27,13 @@ private[akka] object TLSActor {
def props(settings: ActorMaterializerSettings,
sslContext: SSLContext,
sslConfig: Option[AkkaSSLConfig],
firstSession: NegotiateNewSession,
role: TLSRole,
closing: TLSClosing,
hostInfo: Option[(String, Int)],
tracing: Boolean = false): Props =
Props(new TLSActor(settings, sslContext, firstSession, role, closing, hostInfo, tracing)).withDeploy(Deploy.local)
Props(new TLSActor(settings, sslContext, sslConfig, firstSession, role, closing, hostInfo, tracing)).withDeploy(Deploy.local)
final val TransportIn = 0
final val TransportOut = 0
@ -46,6 +47,7 @@ private[akka] object TLSActor {
*/
private[akka] class TLSActor(settings: ActorMaterializerSettings,
sslContext: SSLContext,
externalSslConfig: Option[AkkaSSLConfig],
firstSession: NegotiateNewSession, role: TLSRole, closing: TLSClosing,
hostInfo: Option[(String, Int)], tracing: Boolean)
extends Actor with ActorLogging with Pump {
@ -128,24 +130,23 @@ private[akka] class TLSActor(settings: ActorMaterializerSettings,
// These are Netty's default values
// 16665 + 1024 (room for compressed data) + 1024 (for OpenJDK compatibility)
val transportOutBuffer = ByteBuffer.allocate(16665 + 2048)
private val transportOutBuffer = ByteBuffer.allocate(16665 + 2048)
/*
* deviating here: chopping multiple input packets into this buffer can lead to
* an OVERFLOW signal that also is an UNDERFLOW; avoid unnecessary copying by
* increasing this buffer size to host up to two packets
*/
val userOutBuffer = ByteBuffer.allocate(16665 * 2 + 2048)
val transportInBuffer = ByteBuffer.allocate(16665 + 2048)
val userInBuffer = ByteBuffer.allocate(16665 + 2048)
private val userOutBuffer = ByteBuffer.allocate(16665 * 2 + 2048)
private val transportInBuffer = ByteBuffer.allocate(16665 + 2048)
private val userInBuffer = ByteBuffer.allocate(16665 + 2048)
val userInChoppingBlock = new ChoppingBlock(UserIn, "UserIn")
private val userInChoppingBlock = new ChoppingBlock(UserIn, "UserIn")
userInChoppingBlock.prepare(userInBuffer)
val transportInChoppingBlock = new ChoppingBlock(TransportIn, "TransportIn")
private val transportInChoppingBlock = new ChoppingBlock(TransportIn, "TransportIn")
transportInChoppingBlock.prepare(transportInBuffer)
// ssl-config
val sslConfig = AkkaSSLConfig(context.system)
val hostnameVerifier = sslConfig.hostnameVerifier
private val sslConfig = externalSslConfig.getOrElse(AkkaSSLConfig(context.system))
private val hostnameVerifier = sslConfig.hostnameVerifier
val engine: SSLEngine = {
val e = hostInfo match {
@ -473,7 +474,9 @@ private[akka] class TLSActor(settings: ActorMaterializerSettings,
// since setting a custom HostnameVerified (in JDK8, update 60 still) disables SNI
// see here: https://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html#SNIExamples
// resolves: https://github.com/akka/akka/issues/19287
private def applySNI(params: NegotiateNewSession): Unit = for {
private def applySNI(params: NegotiateNewSession): Unit = {
println("sslConfig.config.loose.disableSNI = " + sslConfig.config.loose.disableSNI)
for {
sslParams ← params.sslParameters
(hostname, _) ← hostInfo
if !sslConfig.config.loose.disableSNI
@ -493,5 +496,6 @@ private[akka] class TLSActor(settings: ActorMaterializerSettings,
clone.setServerNames(Collections.singletonList(new SNIHostName(hostname)))
engine.setSSLParameters(clone)
}
}
}

8
akka-stream/src/main/scala/akka/stream/impl/io/TlsModule.scala
View file

@ -6,6 +6,7 @@ import akka.stream._
import akka.stream.impl.StreamLayout.{ CompositeModule, AtomicModule }
import akka.stream.TLSProtocol._
import akka.util.ByteString
import com.typesafe.sslconfig.akka.AkkaSSLConfig
/**
* INTERNAL API.
@ -14,12 +15,13 @@ private[akka] final case class TlsModule(plainIn: Inlet[SslTlsOutbound], plainOu
cipherIn: Inlet[ByteString], cipherOut: Outlet[ByteString],
shape: Shape, attributes: Attributes,
sslContext: SSLContext,
sslConfig: Option[AkkaSSLConfig],
firstSession: NegotiateNewSession,
role: TLSRole, closing: TLSClosing, hostInfo: Option[(String, Int)]) extends AtomicModule {
override def withAttributes(att: Attributes): TlsModule = copy(attributes = att)
override def carbonCopy: TlsModule =
TlsModule(attributes, sslContext, firstSession, role, closing, hostInfo)
TlsModule(attributes, sslContext, sslConfig, firstSession, role, closing, hostInfo)
override def replaceShape(s: Shape) =
if (s != shape) {
@ -34,13 +36,13 @@ private[akka] final case class TlsModule(plainIn: Inlet[SslTlsOutbound], plainOu
* INTERNAL API.
*/
private[akka] object TlsModule {
def apply(attributes: Attributes, sslContext: SSLContext, firstSession: NegotiateNewSession, role: TLSRole, closing: TLSClosing, hostInfo: Option[(String, Int)]): TlsModule = {
def apply(attributes: Attributes, sslContext: SSLContext, sslConfig: Option[AkkaSSLConfig], firstSession: NegotiateNewSession, role: TLSRole, closing: TLSClosing, hostInfo: Option[(String, Int)]): TlsModule = {
val name = attributes.nameOrDefault(s"StreamTls($role)")
val cipherIn = Inlet[ByteString](s"$name.cipherIn")
val cipherOut = Outlet[ByteString](s"$name.cipherOut")
val plainIn = Inlet[SslTlsOutbound](s"$name.transportIn")
val plainOut = Outlet[SslTlsInbound](s"$name.transportOut")
val shape = new BidiShape(plainIn, cipherOut, cipherIn, plainOut)
TlsModule(plainIn, plainOut, cipherIn, cipherOut, shape, attributes, sslContext, firstSession, role, closing, hostInfo)
TlsModule(plainIn, plainOut, cipherIn, cipherOut, shape, attributes, sslContext, sslConfig, firstSession, role, closing, hostInfo)
}
}

40
akka-stream/src/main/scala/akka/stream/javadsl/TLS.scala
View file

@ -7,6 +7,7 @@ import akka.{ japi, NotUsed }
import akka.stream._
import akka.stream.TLSProtocol._
import akka.util.ByteString
import com.typesafe.sslconfig.akka.AkkaSSLConfig
import scala.compat.java8.OptionConverters
@ -48,6 +49,20 @@ import scala.compat.java8.OptionConverters
*/
object TLS {
/**
* Create a StreamTls [[akka.stream.javadsl.BidiFlow]] in client mode. The
* SSLContext will be used to create an SSLEngine to which then the
* `firstSession` parameters are applied before initiating the first
* handshake. The `role` parameter determines the SSLEngine’s role; this is
* often the same as the underlying transport’s server or client role, but
* that is not a requirement and depends entirely on the application
* protocol.
*
* This method uses the default closing behavior or [[IgnoreComplete]].
*/
def create(sslContext: SSLContext, sslConfig: Optional[AkkaSSLConfig], firstSession: NegotiateNewSession, role: TLSRole): BidiFlow[SslTlsOutbound, ByteString, ByteString, SslTlsInbound, NotUsed] =
new javadsl.BidiFlow(scaladsl.TLS.apply(sslContext, OptionConverters.toScala(sslConfig), firstSession, role))
/**
* Create a StreamTls [[akka.stream.javadsl.BidiFlow]] in client mode. The
* SSLContext will be used to create an SSLEngine to which then the
@ -60,7 +75,7 @@ object TLS {
* This method uses the default closing behavior or [[IgnoreComplete]].
*/
def create(sslContext: SSLContext, firstSession: NegotiateNewSession, role: TLSRole): BidiFlow[SslTlsOutbound, ByteString, ByteString, SslTlsInbound, NotUsed] =
new javadsl.BidiFlow(scaladsl.TLS.apply(sslContext, firstSession, role))
new javadsl.BidiFlow(scaladsl.TLS.apply(sslContext, None, firstSession, role))
/**
* Create a StreamTls [[akka.stream.javadsl.BidiFlow]] in client mode. The
@ -76,10 +91,29 @@ object TLS {
* The `hostInfo` parameter allows to optionally specify a pair of hostname and port
* that will be used when creating the SSLEngine with `sslContext.createSslEngine`.
* The SSLEngine may use this information e.g. when an endpoint identification algorithm was
* configured using [[SSLParameters.setEndpointIdentificationAlgorithm]].
* configured using [[javax.net.ssl.SSLParameters.setEndpointIdentificationAlgorithm]].
*/
def create(sslContext: SSLContext, sslConfig: Optional[AkkaSSLConfig], firstSession: NegotiateNewSession, role: TLSRole, hostInfo: Optional[japi.Pair[String, java.lang.Integer]], closing: TLSClosing): BidiFlow[SslTlsOutbound, ByteString, ByteString, SslTlsInbound, NotUsed] =
new javadsl.BidiFlow(scaladsl.TLS.apply(sslContext, OptionConverters.toScala(sslConfig), firstSession, role, closing, OptionConverters.toScala(hostInfo).map(e ⇒ (e.first, e.second))))
/**
* Create a StreamTls [[akka.stream.javadsl.BidiFlow]] in client mode. The
* SSLContext will be used to create an SSLEngine to which then the
* `firstSession` parameters are applied before initiating the first
* handshake. The `role` parameter determines the SSLEngine’s role; this is
* often the same as the underlying transport’s server or client role, but
* that is not a requirement and depends entirely on the application
* protocol.
*
* For a description of the `closing` parameter please refer to [[TLSClosing]].
*
* The `hostInfo` parameter allows to optionally specify a pair of hostname and port
* that will be used when creating the SSLEngine with `sslContext.createSslEngine`.
* The SSLEngine may use this information e.g. when an endpoint identification algorithm was
* configured using [[javax.net.ssl.SSLParameters.setEndpointIdentificationAlgorithm]].
*/
def create(sslContext: SSLContext, firstSession: NegotiateNewSession, role: TLSRole, hostInfo: Optional[japi.Pair[String, java.lang.Integer]], closing: TLSClosing): BidiFlow[SslTlsOutbound, ByteString, ByteString, SslTlsInbound, NotUsed] =
new javadsl.BidiFlow(scaladsl.TLS.apply(sslContext, firstSession, role, closing, OptionConverters.toScala(hostInfo).map(e ⇒ (e.first, e.second))))
new javadsl.BidiFlow(scaladsl.TLS.apply(sslContext, None, firstSession, role, closing, OptionConverters.toScala(hostInfo).map(e ⇒ (e.first, e.second))))
}

50
akka-stream/src/main/scala/akka/stream/scaladsl/TLS.scala
View file

@ -7,6 +7,7 @@ import akka.NotUsed
import akka.stream._
import akka.stream.TLSProtocol._
import akka.util.ByteString
import com.typesafe.sslconfig.akka.AkkaSSLConfig
/**
* Stream cipher support based upon JSSE.
@ -60,11 +61,54 @@ object TLS {
* The `hostInfo` parameter allows to optionally specify a pair of hostname and port
* that will be used when creating the SSLEngine with `sslContext.createSslEngine`.
* The SSLEngine may use this information e.g. when an endpoint identification algorithm was
* configured using [[SSLParameters.setEndpointIdentificationAlgorithm]].
* configured using [[javax.net.ssl.SSLParameters.setEndpointIdentificationAlgorithm]].
*/
def apply(sslContext: SSLContext, firstSession: NegotiateNewSession, role: TLSRole,
def apply(sslContext: SSLContext,
sslConfig: Option[AkkaSSLConfig],
firstSession: NegotiateNewSession, role: TLSRole,
closing: TLSClosing = IgnoreComplete, hostInfo: Option[(String, Int)] = None): scaladsl.BidiFlow[SslTlsOutbound, ByteString, ByteString, SslTlsInbound, NotUsed] =
new scaladsl.BidiFlow(TlsModule(Attributes.none, sslContext, firstSession, role, closing, hostInfo))
new scaladsl.BidiFlow(TlsModule(Attributes.none, sslContext, sslConfig, firstSession, role, closing, hostInfo))
/**
* Create a StreamTls [[akka.stream.scaladsl.BidiFlow]]. The
* SSLContext will be used to create an SSLEngine to which then the
* `firstSession` parameters are applied before initiating the first
* handshake. The `role` parameter determines the SSLEngine’s role; this is
* often the same as the underlying transport’s server or client role, but
* that is not a requirement and depends entirely on the application
* protocol.
*
* For a description of the `closing` parameter please refer to [[TLSClosing]].
*
* The `hostInfo` parameter allows to optionally specify a pair of hostname and port
* that will be used when creating the SSLEngine with `sslContext.createSslEngine`.
* The SSLEngine may use this information e.g. when an endpoint identification algorithm was
* configured using [[javax.net.ssl.SSLParameters.setEndpointIdentificationAlgorithm]].
*/
def apply(sslContext: SSLContext,
firstSession: NegotiateNewSession, role: TLSRole,
closing: TLSClosing, hostInfo: Option[(String, Int)]): scaladsl.BidiFlow[SslTlsOutbound, ByteString, ByteString, SslTlsInbound, NotUsed] =
new scaladsl.BidiFlow(TlsModule(Attributes.none, sslContext, None, firstSession, role, closing, hostInfo))
/**
* Create a StreamTls [[akka.stream.scaladsl.BidiFlow]]. The
* SSLContext will be used to create an SSLEngine to which then the
* `firstSession` parameters are applied before initiating the first
* handshake. The `role` parameter determines the SSLEngine’s role; this is
* often the same as the underlying transport’s server or client role, but
* that is not a requirement and depends entirely on the application
* protocol.
*
* For a description of the `closing` parameter please refer to [[TLSClosing]].
*
* The `hostInfo` parameter allows to optionally specify a pair of hostname and port
* that will be used when creating the SSLEngine with `sslContext.createSslEngine`.
* The SSLEngine may use this information e.g. when an endpoint identification algorithm was
* configured using [[javax.net.ssl.SSLParameters.setEndpointIdentificationAlgorithm]].
*/
def apply(sslContext: SSLContext,
firstSession: NegotiateNewSession, role: TLSRole): scaladsl.BidiFlow[SslTlsOutbound, ByteString, ByteString, SslTlsInbound, NotUsed] =
new scaladsl.BidiFlow(TlsModule(Attributes.none, sslContext, None, firstSession, role, IgnoreComplete, None))
}

3
project/MiMa.scala
View file

@ -850,6 +850,9 @@ object MiMa extends AutoPlugin {
// internal api
FilterAnyProblemStartingWith("akka.stream.impl"),
// #20214 SNI disabling for single connections (AkkaSSLConfig being passed around)
ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.http.javadsl.ConnectionContext.sslConfig"), // class meant only for internal extension
//#20229 migrate GroupBy to GraphStage
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.scaladsl.GraphDSL#Builder.deprecatedAndThen"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.scaladsl.Flow.deprecatedAndThen"),