diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsSpec.scala
index 32c152bf13..5a7209e7c1 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsSpec.scala
@@ -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._
@@ -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) =
diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala
index 65e3a9110c..902d91e268 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala
@@ -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 {
@@ -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 &&
@@ -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
         }
       }