Skip to content

Abort TLS transport on session verification failure #3323

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
He-Pin wants to merge 1 commit into
base: main
Choose a base branch
from
+34 −8
Open

Abort TLS transport on session verification failure #3323

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 27 additions & 3 deletions stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,12 +21,12 @@ import javax.net.ssl._

import scala.collection.immutable
import scala.concurrent.Await
import scala.concurrent.Future
import scala.concurrent.{ Future, Promise }
import scala.concurrent.duration._
import scala.util.{ Random, Success }
import scala.util.{ Failure, Random, Success }

import org.apache.pekko
import pekko.NotUsed
import pekko.{ Done, NotUsed }
import pekko.pattern.{ after => later }
import pekko.stream._
import pekko.stream.TLSProtocol._
Expand Down Expand Up @@ -503,6 +503,30 @@ abstract class AbstractTlsSpec(useLegacyActor: Boolean)
clientErrText should include("unable to find valid certification path to requested target")
}

"abort the transport when session verification fails" in {
val verificationFailure = new SSLException("session verification failed")
val rejectingClientTls = tlsBidi(
() => createSSLEngine(sslContext, Client),
verifySession = _ => Failure(verificationFailure),
closing = IgnoreBoth)
val clientToServerDone = Promise[Done]()
val clientToServer = Flow[ByteString].watchTermination { (_, done) =>
done.onComplete(clientToServerDone.tryComplete)
NotUsed
}
val transport = BidiFlow.fromFlows(clientToServer, Flow[ByteString])
val serverApplication: Flow[SslTlsInbound, SslTlsOutbound, NotUsed] =
Flow.fromSinkAndSource(Sink.ignore, Source.maybe[SslTlsOutbound])

Source
.maybe[SslTlsOutbound]
.via(rejectingClientTls.atop(transport).atop(serverTls(IgnoreBoth).reversed).join(serverApplication))
.runWith(Sink.ignore)

val failure = intercept[SSLException](Await.result(clientToServerDone.future, 3.seconds.dilated))
(failure should be).theSameInstanceAs(verificationFailure)
}

"reliably cancel subscriptions when TransportIn fails early" in {
val ex = new Exception("hello")
val (sub, out1, out2) =
Expand Down
12 changes: 7 additions & 5 deletions stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -376,7 +376,7 @@ import pekko.util.ByteString
val result = wrapIntoTransportBuffer()
lastHandshakeStatus = result.getHandshakeStatus

if (lastHandshakeStatus == FINISHED) handshakeFinished()
if (lastHandshakeStatus == FINISHED && !handshakeFinished()) return
runDelegatedTasks()

result.getStatus match {
Expand Down Expand Up @@ -446,8 +446,8 @@ import pekko.util.ByteString

case FINISHED =>
flushToUser()
handshakeFinished()
transportInput.putBackUnreadBuffer(transportInBuffer)
if (handshakeFinished())
transportInput.putBackUnreadBuffer(transportInBuffer)

case NEED_UNWRAP
if transportInBuffer.hasRemaining &&
Expand Down Expand Up @@ -480,16 +480,18 @@ import pekko.util.ByteString
if (TlsEngineHelpers.runDelegatedTasks(engine) > 0 || lastHandshakeStatus == NEED_TASK)
lastHandshakeStatus = engine.getHandshakeStatus

private def handshakeFinished(): Unit = {
private def handshakeFinished(): Boolean = {
val session = engine.getSession
verifySession(session) match {
case Success(()) =>
currentSession = session
stateBits |= PlainDataAllowedFlag
flushToUser()
true

case Failure(ex) =>
throw ex
failTls(ex)
false
}
}

Expand Down