From 551924cf5cc610df7a48fd9674506a07903d8208 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E8=99=8E=E9=B8=A3?= <hepin.p@alibaba-inc.com>
Date: Sat, 11 Jul 2026 13:46:41 +0800
Subject: [PATCH] Abort TLS transport on session verification failure #3245

Motivation:
A failed post-handshake session verifier was handled as an SSLEngine failure, causing the GraphStage TLS implementation to emit close_notify instead of aborting the transport.

Modification:
Handle verifier failures as an explicit TLS stage failure and stop handshake processing. Add a regression test shared by the GraphStage and legacy actor implementations for TLS 1.2 and TLS 1.3.

Result:
Session verification failures now fail the transport with the original exception while engine-originated TLS failures still flush their fatal alerts.

Tests:
- stream-tests / Test / testOnly org.apache.pekko.stream.io.TlsGraphStageSpec (113 passed)
- stream-tests / Test / testOnly org.apache.pekko.stream.io.TlsSpec (113 passed)
- +headerCheckAll (passed)
- checkCodeStyle (passed)
- validatePullRequest (environment failure: leveldbjni-all 1.8 has no macOS ARM64 binary; remaining unrelated tests stopped)

References:
Fixes #3245
---
 .../org/apache/pekko/stream/io/TlsSpec.scala  | 30 +++++++++++++++++--
 .../pekko/stream/impl/io/TlsGraphStage.scala  | 12 ++++----
 2 files changed, 34 insertions(+), 8 deletions(-)

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 32c152bf13d..5a7209e7c1e 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 65e3a9110cb..902d91e268c 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
         }
       }