From 2e8364f80ec79fcecffdf71c1553fed6f82ee9c0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=99=8E=E9=B8=A3?= Date: Sat, 31 Oct 2026 01:36:30 +0800 Subject: [PATCH] Report TLS truncation on transport-output cancellation #3739 Motivation: Cancelling a ciphertext output can make received TLS data unreachable. Completing the plaintext output silently hides the unclean shutdown. Modification: - Handle transport-output cancellation in both TLS engines, cancel the plaintext input immediately, and allow at most one final inbound round. - Report SessionTruncated for established sessions without a peer close_notify. Preserve pending plaintext and session validity, including during renegotiation, and skip depleted transport input. - Preserve an existing wait for the peer's close_notify after the local close_notify has been sent. End that wait if both outputs are cancelled. - Share cancellation predicates, preserve actor error publication order, and make actor failure cleanup safe against pump reentrancy. - Add directional tests and document the 2.0.0 behavior and the TCP ByteString APIs' filtering of SessionTruncated. Result: Direct TLS BidiFlow consumers can observe an inbound session abandoned by cancellation. The final signal obeys backpressure; incomplete input and data beyond the final inbound round can still be discarded. Tests: - JDK 17: sbt "stream-tests/Test/testOnly *Tls*": 379 passed, 0 failed. - Eight upstream-cancellation and close-handshake regression cases fail on the previous PR head and pass with the fix, across both engines and TLS 1.2/1.3. Depleted-input and renegotiation regressions were also verified to fail before their fixes. - sbt +mimaReportBinaryIssues: passed on Scala 2.13.18 and 3.3.9. - sbt headerCreateAll +headerCheckAll checkCodeStyle: passed. - scalafmt diff check and git diff --check: passed. - sortImports was blocked by the existing scala.meta.package$.Input() NoSuchMethodError; unrelated partial edits were removed. - An earlier JDK 25 run was interrupted after a ScalaTest SocketReporter Broken pipe error; final local validation uses JDK 17. - docs/paradox and validatePullRequest are delegated to remote CI; an earlier local docs/paradox attempt was interrupted. References: Refs #3739 --- .../paradox/release-notes/releases-2.0.md | 10 + docs/src/main/paradox/stream/stream-io.md | 31 ++ .../stream/io/TlsGraphStageIsolatedSpec.scala | 135 ++++- .../io/TlsTransportOutputCancelSpec.scala | 522 ++++++++++++++++++ .../pekko/stream/impl/io/TLSActor.scala | 81 ++- .../stream/impl/io/TlsEngineHelpers.scala | 15 + .../pekko/stream/impl/io/TlsGraphStage.scala | 53 +- 7 files changed, 819 insertions(+), 28 deletions(-) create mode 100644 stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsTransportOutputCancelSpec.scala diff --git a/docs/src/main/paradox/release-notes/releases-2.0.md b/docs/src/main/paradox/release-notes/releases-2.0.md index d79d519a3c9..93f13aea333 100644 --- a/docs/src/main/paradox/release-notes/releases-2.0.md +++ b/docs/src/main/paradox/release-notes/releases-2.0.md @@ -48,6 +48,16 @@ Please read the @ref:[Migration Guide](../migration/migration-guide-1.x-2.x.md) breaking API changes, dependency changes (such as the Agrona 2.x `--add-opens` requirement) and the configuration defaults that have changed in Pekko 2.0.0. +Unless already waiting for the peer's `close_notify`, cancelling a TLS operator's transport output +now reports `SessionTruncated` on its plaintext output +if a session was established and the peer's `close_notify` has not been processed, including when +no inbound data is buffered or a renegotiation is in progress ([PR3740](https://github.com/apache/pekko/pull/3740)). +Previously this path could complete silently after discarding received records. Consumers using +the TLS `BidiFlow` directly must handle `SessionTruncated` as well as `SessionBytes`. The Scala and +Java `Tcp.bindWithTls` and `Tcp.outgoingConnectionWithTls` APIs filter out this signal and continue +to expose `ByteString` streams. The final signal obeys backpressure; cancel the plaintext output +too if it will no longer be consumed. See @ref:[Stream IO](../stream/stream-io.md) for details. + ### Bug Fixes * Replaying a replicated event sourced actor from a snapshot caused a bug ([PR2372](https://github.com/apache/pekko/pull/2372)) diff --git a/docs/src/main/paradox/stream/stream-io.md b/docs/src/main/paradox/stream/stream-io.md index b780254ddef..d05aef27f09 100644 --- a/docs/src/main/paradox/stream/stream-io.md +++ b/docs/src/main/paradox/stream/stream-io.md @@ -203,6 +203,37 @@ The drain also continues after a TLS 1.3 peer sends `close_notify`, because that peer's sending direction. A TLS 1.2 `close_notify` closes both directions of the `SSLEngine`, so it ends the drain and accepted plaintext that has not been sent is discarded. +Unless a clean shutdown is already waiting for the peer's reply, cancelling the transport output +ends the session, because nothing more can be sent to the peer. That +teardown is not silent: once the session is established, the operator still delivers the inbound +data it can decrypt in the round that observes the cancellation, and then reports +`SessionTruncated` before the plaintext output completes if the peer's `close_notify` has not been +processed. This also applies with no buffered inbound records and during renegotiation of an +established session: truncation means that a clean peer shutdown was not observed, rather than proof +that application bytes were lost. Inbound records still buffered beyond that round are discarded. +The operator does not wait for more ciphertext to complete a partial TLS record: cancellation can +discard fragments still in flight, and the truncation signal makes that abandonment observable. +An already pending plaintext element is delivered before the signal. A plaintext output that is +already closed, or a session cancelled before its initial handshake finished, completes without that +signal. The transport-output cancellation itself does not call `SSLEngine.closeInbound()` or +invalidate the established `SSLSession`. If transport input has already completed and is depleted, +the cancellation path skips the final inbound round and reports truncation directly. + +If the operator has already sent its own `close_notify` and is waiting for the peer's reply, +transport-output cancellation preserves that existing wait while the plaintext output is open. +Cancelling the plaintext output as well ends the wait. An operator still flushing outbound data +cannot continue sending after transport-output cancellation. + +The truncation signal obeys plaintext backpressure. The plaintext input is cancelled immediately +when the transport output is cancelled, even if the final inbound round is waiting for plaintext +demand. The transport input is cancelled after that round; the operator then stays alive until +the signal is consumed or the plaintext output is cancelled. +Applications that stop reading must cancel that output as well; applications requiring a time bound +can apply `completionTimeout` to the plaintext stream. Consumers using the TLS `BidiFlow` directly +must handle `SessionTruncated` explicitly rather than using a partial `map` that only matches +`SessionBytes`. The Scala and Java `Tcp.bindWithTls` and `Tcp.outgoingConnectionWithTls` APIs expose +`ByteString` streams and filter out this signal. + ## Streaming File IO Pekko Streams provide simple Sources and Sinks that can work with @apidoc[util.ByteString] instances to perform IO operations diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsGraphStageIsolatedSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsGraphStageIsolatedSpec.scala index 9328481aded..2c68a1ee375 100644 --- a/stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsGraphStageIsolatedSpec.scala +++ b/stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsGraphStageIsolatedSpec.scala @@ -37,7 +37,7 @@ import pekko.actor.Cancellable import pekko.event.LoggingAdapter import pekko.stream._ import pekko.stream.TLSProtocol._ -import pekko.stream.impl.io.TlsGraphStage +import pekko.stream.impl.io.{ TlsGraphStage, TlsModule } import pekko.stream.scaladsl._ import pekko.stream.scaladsl.GraphDSL.Implicits._ import pekko.stream.testkit.{ StreamSpec, TestPublisher, TestSubscriber } @@ -314,12 +314,15 @@ abstract class TlsGraphStageBufferTestBase(config: Config) extends StreamSpec(co private var handshakeStatus: SSLEngineResult.HandshakeStatus = NEED_UNWRAP @volatile var unwrapCount = 0 private var inboundDone = false - private var outboundDone = false + @volatile private var outboundDone = false private var handshakePending = true @volatile var applicationWraps = 0 @volatile var applicationBytes = 0 @volatile var handshakesStarted = 0 + @volatile var closeInboundCalls = 0 + @volatile var unwrapFailure: SSLException = _ + @volatile var failedUnwraps = 0 @volatile var overflowWraps = 0 @volatile var largestOutputCapacity = 0 @volatile var inputArray: Array[Byte] = _ @@ -399,6 +402,10 @@ abstract class TlsGraphStageBufferTestBase(config: Config) extends StreamSpec(co dsts: Array[ByteBuffer], offset: Int, length: Int): SSLEngineResult = { + if (unwrapFailure ne null) { + failedUnwraps += 1 + throw unwrapFailure + } if (src.hasArray) inputArray = src.array() if (!handshakePending && negotiatedPacketSize > 0) { reportedPacketSize = negotiatedPacketSize @@ -437,6 +444,7 @@ abstract class TlsGraphStageBufferTestBase(config: Config) extends StreamSpec(co } override def closeInbound(): Unit = { + closeInboundCalls += 1 inboundDone = true handshakeStatus = if (wrapOnCloseInbound) NEED_WRAP else NOT_HANDSHAKING } @@ -487,13 +495,23 @@ abstract class TlsGraphStageBufferTestBase(config: Config) extends StreamSpec(co negotiatedPacketSize: Int = 0, negotiatedApplicationSize: Int = 0, sessionProtocol: String = null, - wrapHandshakeOnNeedWrap: Boolean = false): DeferredCloseHarness = { + wrapHandshakeOnNeedWrap: Boolean = false, + useLegacyActor: Boolean = false): DeferredCloseHarness = { val engine = new ScriptedSSLEngine(TlsSpec.initSslContext("TLSv1.2").createSSLEngine(), wrapOnCloseInbound, requestHandshakeAfterApplication, requestWrapOnSubsequentUnwrap, requiredOutputSpace, negotiatedPacketSize, negotiatedApplicationSize, sessionProtocol, wrapHandshakeOnNeedWrap) val sessionVerified = Promise[Unit]() + val verifySession = (_: SSLSession) => { + sessionVerified.trySuccess(()) + Success(()) + } + val tlsFlow = + if (useLegacyActor) + BidiFlow.fromGraph(TlsModule(Attributes.none, () => engine, verifySession, closing)) + else + BidiFlow.fromGraph(new TlsGraphStage(() => engine, verifySession, closing)) val probes = RunnableGraph .fromGraph( @@ -503,15 +521,7 @@ abstract class TlsGraphStageBufferTestBase(config: Config) extends StreamSpec(co TestSink[ByteString](), TestSink[SslTlsInbound]())(Tuple4.apply) { implicit b => (plainIn, cipherIn, cipherOut, plainOut) => val tls = b.add( - BidiFlow - .fromGraph(new TlsGraphStage( - () => engine, - _ => { - sessionVerified.trySuccess(()) - Success(()) - }, - closing)) - .withAttributes(Attributes.inputBuffer(1, 1).and(extraAttributes))) + tlsFlow.withAttributes(Attributes.inputBuffer(1, 1).and(extraAttributes))) plainIn ~> tls.in1 tls.out1 ~> cipherOut cipherIn ~> tls.in2 @@ -693,6 +703,107 @@ class TlsGraphStageDeferredCloseSpec extends TlsGraphStageBufferTestBase(TlsGrap harness.engine.applicationWraps shouldBe 0 } + "complete without truncation when the transport output is cancelled before the initial handshake" in { + val harness = newHarness() + + // No port has received an element or demand that would otherwise start the pump. + harness.cipherOut.cancel() + + harness.plainIn.expectCancellation() + harness.cipherIn.expectCancellation() + harness.plainOut.expectComplete() + harness.engine.unwrapCount shouldBe 0 + harness.engine.closeInboundCalls shouldBe 0 + } + + "report transport output cancellation without closing the established engine inbound" in { + val harness = newHarness() + finishInitialHandshake(harness) + + harness.cipherOut.cancel() + + harness.plainIn.expectCancellation() + harness.cipherIn.expectCancellation() + harness.plainOut.expectNext(SessionTruncated) + harness.plainOut.expectComplete() + harness.engine.closeInboundCalls shouldBe 0 + } + + "report transport output cancellation after the engine outbound has closed" in { + val harness = newHarness(closing = EagerClose) + finishInitialHandshake(harness) + + harness.plainIn.sendComplete() + harness.cipherOut.request(1) + awaitAssert(harness.engine.isOutboundDone shouldBe true) + harness.cipherOut.cancel() + + harness.cipherIn.expectCancellation() + harness.plainOut.expectNext(SessionTruncated) + harness.plainOut.expectComplete() + harness.engine.closeInboundCalls shouldBe 0 + } + + for { + useLegacyActor <- List(false, true) + bufferedInput <- List(false, true) + } { + val engine = if (useLegacyActor) "legacy actor" else "graph stage" + s"report cancellation after transport input completes without closing inbound ($engine, buffered=$bufferedInput)" in { + assertAllStagesStopped { + val harness = newHarness( + negotiatedPacketSize = TlsRecord.size, + negotiatedApplicationSize = 1, + useLegacyActor = useLegacyActor) + finishInitialHandshake(harness) + + // Consume the handshake's remaining plaintext demand before completing transport input. + harness.cipherIn.sendNext(TlsRecord) + harness.plainOut.expectNext().asInstanceOf[SessionBytes].bytes shouldBe ByteString("A") + if (bufferedInput) harness.cipherIn.sendNext(TlsRecord) + // These probe calls enqueue completion, cancellation, then demand in that order. + harness.cipherIn.sendComplete() + harness.cipherOut.cancel() + // With nothing left to decrypt, teardown releases the plaintext input without waiting for demand. + if (!bufferedInput) harness.plainIn.expectCancellation() + harness.plainOut.request(2) + + if (bufferedInput) + harness.plainOut.expectNext().asInstanceOf[SessionBytes].bytes shouldBe ByteString("A") + harness.plainOut.expectNext(SessionTruncated) + harness.plainOut.expectComplete() + if (bufferedInput) harness.plainIn.expectCancellation() + harness.engine.closeInboundCalls shouldBe 0 + } + } + } + + for (useLegacyActor <- List(false, true)) { + val engine = if (useLegacyActor) "legacy actor" else "graph stage" + s"preserve an unwrap failure during transport output cancellation ($engine)" in { + assertAllStagesStopped { + val harness = newHarness( + negotiatedPacketSize = TlsRecord.size, + negotiatedApplicationSize = 1, + useLegacyActor = useLegacyActor) + finishInitialHandshake(harness) + harness.cipherIn.sendNext(TlsRecord) + harness.plainOut.expectNext().asInstanceOf[SessionBytes].bytes shouldBe ByteString("A") + + val failure = new SSLException("failed final inbound round") + harness.engine.unwrapFailure = failure + harness.cipherIn.sendNext(TlsRecord) + harness.cipherOut.cancel() + harness.plainOut.request(1) + + harness.plainOut.expectError() shouldBe failure + harness.plainIn.expectCancellation() + harness.cipherIn.expectCancellation() + harness.engine.failedUnwraps shouldBe 1 + } + } + } + for (closing <- Seq(IgnoreComplete, EagerClose)) { s"drain accepted plaintext when transport input completes during cancellation with $closing" in { val harness = newHarness(closing = closing) diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsTransportOutputCancelSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsTransportOutputCancelSpec.scala new file mode 100644 index 00000000000..a214174f99d --- /dev/null +++ b/stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsTransportOutputCancelSpec.scala @@ -0,0 +1,522 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.pekko.stream.io + +import java.nio.ByteBuffer +import java.nio.charset.StandardCharsets +import javax.net.ssl.{ SSLContext, SSLEngine, SSLEngineResult, SSLSession } +import javax.net.ssl.SSLEngineResult.HandshakeStatus._ + +import scala.concurrent.duration._ +import scala.util.Success + +import org.apache.pekko +import pekko.NotUsed +import pekko.stream._ +import pekko.stream.TLSProtocol._ +import pekko.stream.impl.io.{ TlsGraphStage, TlsModule } +import pekko.stream.scaladsl._ +import pekko.stream.testkit.{ StreamSpec, TestPublisher, TestSubscriber } +import pekko.stream.testkit.scaladsl.{ TestSink, TestSource } +import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped +import pekko.testkit.WithLogCapturing +import pekko.util.ByteString + +/** + * Pins the inbound contract of both TLS engines when the transport (ciphertext) output is cancelled: + * records already received from the peer either reach the plaintext output or are reported as a + * truncated session. Silently dropping them is what these tests must keep catching. + */ +class TlsTransportOutputCancelSpec extends StreamSpec(TlsSpec.configOverrides) with WithLogCapturing { + + private val Protocols = List("TLSv1.2", "TLSv1.3") + + private def tlsBidi( + context: SSLContext, + role: TLSRole, + closing: TLSClosing, + useLegacyActor: Boolean): BidiFlow[SslTlsOutbound, ByteString, ByteString, SslTlsInbound, NotUsed] = { + def createEngine(): SSLEngine = { + val engine = context.createSSLEngine() + engine.setUseClientMode(role == Client) + engine.setEnabledProtocols(Array(context.getProtocol)) + engine + } + val verifySession: SSLSession => scala.util.Try[Unit] = _ => Success(()) + if (useLegacyActor) + BidiFlow.fromGraph(TlsModule(Attributes.none, createEngine _, verifySession, closing)) + else + BidiFlow + .fromGraph(new TlsGraphStage(createEngine _, verifySession, closing)) + .withAttributes(TlsGraphStage.StreamTlsAttributes) + } + + /** A plain `SSLEngine` driving the ciphertext ports of the operator under test. */ + private final class Peer( + context: SSLContext, + cipherIn: TestPublisher.Probe[ByteString], + cipherOut: TestSubscriber.Probe[ByteString]) { + private val engine: SSLEngine = context.createSSLEngine() + engine.setUseClientMode(true) + engine.setEnabledProtocols(Array(context.getProtocol)) + + private var cipherBuffer = ByteBuffer.allocate(engine.getSession.getPacketBufferSize) + private var plainBuffer = ByteBuffer.allocate(engine.getSession.getApplicationBufferSize) + private var decoded = ByteString.empty + private var outstandingDemand = 0L + + def handshake(): Unit = { + engine.beginHandshake() + var status = engine.getHandshakeStatus + while (status != NOT_HANDSHAKING && status != FINISHED) { + status match { + case NEED_WRAP => send(ByteString.empty) + case NEED_UNWRAP => if (!decodeBuffered()) receive() + case NEED_TASK => runDelegatedTasks() + case other => fail(s"unexpected handshake status $other") + } + status = engine.getHandshakeStatus + } + } + + /** Ciphertext elements the operator may push before this peer has to ask again. */ + def receiveDemand(elements: Long): Unit = { + cipherOut.request(elements) + outstandingDemand += elements + } + + def cancelCipherOut(): Unit = cipherOut.cancel() + + def expectHandshakeOutput(): Unit = { + cipherOut.expectNext().nonEmpty shouldBe true + outstandingDemand -= 1 + } + + def expectCipherInCancellation(): Unit = cipherIn.expectCancellation() + + def receiveCloseNotify(): Unit = + while (!engine.isInboundDone) + if (!decodeBuffered()) receive() + + def sendCloseNotify(): Unit = { + engine.closeOutbound() + send(ByteString.empty) + } + + def sendPartialRecord(text: String): ByteString = { + val ciphertext = encrypt(ByteString(text, StandardCharsets.US_ASCII)) + sendCiphertext(ciphertext.take(3)) + ciphertext.drop(3) + } + + def sendCiphertext(bytes: ByteString): Unit = cipherIn.sendNext(bytes) + + def send(plaintext: ByteString): Unit = sendCiphertext(encrypt(plaintext)) + + private def encrypt(plaintext: ByteString): ByteString = { + val packet = ByteBuffer.allocate(engine.getSession.getPacketBufferSize) + val source = ByteBuffer.allocate(plaintext.size) + plaintext.copyToBuffer(source) + source.flip() + val result = engine.wrap(source, packet) + result.getStatus shouldBe + (if (engine.isOutboundDone) SSLEngineResult.Status.CLOSED else SSLEngineResult.Status.OK) + packet.flip() + ByteString(packet) + } + + def sendText(text: String): Unit = send(ByteString(text, StandardCharsets.US_ASCII)) + + /** Reads ciphertext one element at a time until `expected` plaintext bytes arrived. */ + def receiveText(expected: Int): ByteString = { + while (decoded.size < expected) + if (!decodeBuffered()) receive() + val result = decoded + decoded = ByteString.empty + result + } + + private def receive(): Unit = { + if (outstandingDemand == 0) receiveDemand(1) + outstandingDemand -= 1 + append(cipherOut.expectNext()) + } + + private def runDelegatedTasks(): Unit = { + var task = engine.getDelegatedTask + while (task ne null) { + task.run() + task = engine.getDelegatedTask + } + } + + private def append(bytes: ByteString): Unit = { + if (cipherBuffer.remaining < bytes.size) { + val grown = ByteBuffer.allocate(cipherBuffer.position() + bytes.size) + cipherBuffer.flip() + grown.put(cipherBuffer) + cipherBuffer = grown + } + bytes.copyToBuffer(cipherBuffer) + } + + private def decodeBuffered(): Boolean = { + cipherBuffer.flip() + val result = engine.unwrap(cipherBuffer, plainBuffer) + cipherBuffer.compact() + result.getStatus match { + case SSLEngineResult.Status.BUFFER_OVERFLOW => + val grown = ByteBuffer.allocate(plainBuffer.capacity * 2) + plainBuffer.flip() + grown.put(plainBuffer) + plainBuffer = grown + true + case SSLEngineResult.Status.BUFFER_UNDERFLOW => false + case _ => + plainBuffer.flip() + decoded ++= ByteString(plainBuffer) + plainBuffer.clear() + result.bytesConsumed > 0 || result.bytesProduced > 0 + } + } + } + + private final class Harness( + val plainIn: TestPublisher.Probe[SslTlsOutbound], + val plainOut: TestSubscriber.Probe[SslTlsInbound], + val peer: Peer) + + private def newHarness( + protocol: String, + useLegacyActor: Boolean, + closing: TLSClosing = IgnoreComplete, + handshake: Boolean = true): Harness = { + val context = TlsSpec.initSslContext(protocol) + val probes = RunnableGraph + .fromGraph(GraphDSL.createGraph( + TestSource[SslTlsOutbound](), + TestSink[SslTlsInbound](), + TestSource[ByteString](), + TestSink[ByteString]())(Tuple4.apply) { implicit b => (plainIn, plainOut, cipherIn, cipherOut) => + import GraphDSL.Implicits._ + val tls = b.add(tlsBidi(context, Server, closing, useLegacyActor)) + plainIn ~> tls.in1 + tls.out1 ~> cipherOut + cipherIn ~> tls.in2 + tls.out2 ~> plainOut + ClosedShape + }) + .run() + + val peer = new Peer(context, probes._3, probes._4) + val harness = new Harness(probes._1, probes._2, peer) + + // Inbound decryption needs plaintext demand, so the handshake consumes one request. + harness.plainOut.ensureSubscription() + probes._4.ensureSubscription() + if (handshake) { + harness.plainOut.request(1) + peer.receiveDemand(32) + peer.handshake() + } + harness + } + + /** Establishes the session and leaves the plaintext output with no outstanding demand. */ + private def establishIdleSession(harness: Harness): SSLSession = { + harness.plainIn.sendNext(SendBytes(ByteString("ready"))) + harness.peer.receiveText(5) shouldBe ByteString("ready") + harness.peer.sendText("ping") + val received = harness.plainOut.expectNext().asInstanceOf[SessionBytes] + received.bytes shouldBe ByteString("ping") + received.session + } + + private def expectSessionBytes(inbound: SslTlsInbound, text: String): Unit = inbound match { + case SessionBytes(_, bytes) => bytes shouldBe ByteString(text, StandardCharsets.US_ASCII) + case other => fail(s"expected SessionBytes($text) but got $other") + } + + "A TLS stage whose transport output is cancelled" must { + for { + protocol <- Protocols + useLegacyActor <- List(false, true) + } { + val engine = if (useLegacyActor) "legacy actor" else "graph stage" + + s"complete without truncation before the initial handshake ($engine, $protocol)" in { + assertAllStagesStopped { + val harness = newHarness(protocol, useLegacyActor, handshake = false) + harness.peer.cancelCipherOut() + harness.peer.expectCipherInCancellation() + harness.plainIn.expectCancellation() + harness.plainOut.expectComplete() + } + } + + for (closing <- List(IgnoreComplete, IgnoreCancel, IgnoreBoth, EagerClose)) { + s"report cancellation with no buffered input and preserve session validity ($engine, $protocol, $closing)" in { + assertAllStagesStopped { + val harness = newHarness(protocol, useLegacyActor, closing) + val session = establishIdleSession(harness) + session.isValid shouldBe true + harness.peer.cancelCipherOut() + harness.plainOut.request(1) + harness.plainOut.expectNext() shouldBe SessionTruncated + harness.plainOut.expectComplete() + harness.peer.expectCipherInCancellation() + harness.plainIn.expectCancellation() + session.isValid shouldBe true + } + } + + s"stop when the plaintext output cancels while the truncation report waits ($engine, $protocol, $closing)" in { + assertAllStagesStopped { + val harness = newHarness(protocol, useLegacyActor, closing) + establishIdleSession(harness) + harness.peer.cancelCipherOut() + harness.peer.expectCipherInCancellation() + harness.plainIn.expectCancellation() + harness.plainOut.cancel() + } + } + } + + if (protocol == "TLSv1.2") { + s"report cancellation during renegotiation of an established session ($engine)" in { + assertAllStagesStopped { + val harness = newHarness(protocol, useLegacyActor) + establishIdleSession(harness) + harness.plainIn.sendNext(NegotiateNewSession) + // Observe the HelloRequest without answering it: the second handshake remains in flight. + harness.peer.expectHandshakeOutput() + harness.peer.sendText("first") + harness.peer.sendText("second") + harness.plainOut.expectNoMessage(200.millis) + harness.peer.cancelCipherOut() + harness.plainOut.request(3) + expectSessionBytes(harness.plainOut.expectNext(), "first") + harness.plainOut.expectNext() shouldBe SessionTruncated + harness.plainOut.expectComplete() + } + } + } + + s"report cancellation after the plaintext input completes ($engine, $protocol)" in { + assertAllStagesStopped { + val harness = newHarness(protocol, useLegacyActor) + establishIdleSession(harness) + harness.plainIn.sendComplete() + // An inbound round moves the pump to outboundClosed after observing input completion. + harness.peer.sendText("last") + harness.plainOut.request(1) + expectSessionBytes(harness.plainOut.expectNext(), "last") + harness.peer.cancelCipherOut() + harness.peer.expectCipherInCancellation() + harness.plainOut.request(1) + harness.plainOut.expectNext() shouldBe SessionTruncated + harness.plainOut.expectComplete() + } + } + + s"report a truncated session when buffered inbound records cannot all be delivered ($engine, $protocol)" in { + val harness = newHarness(protocol, useLegacyActor) + within(10.seconds) { + establishIdleSession(harness) + + // Two records arrive while the plaintext output has no demand, so both stay buffered. + harness.peer.sendText("first") + harness.plainOut.expectNoMessage(200.millis) + harness.peer.sendText("second") + harness.plainOut.expectNoMessage(200.millis) + + harness.peer.cancelCipherOut() + harness.plainOut.request(2) + + expectSessionBytes(harness.plainOut.expectNext(), "first") + // The remaining record is unreachable, so the abandonment must be observable. + harness.plainOut.expectNext() shouldBe SessionTruncated + harness.plainOut.expectComplete() + harness.peer.expectCipherInCancellation() + } + } + + s"still deliver a single buffered record before reporting truncation ($engine, $protocol)" in { + val harness = newHarness(protocol, useLegacyActor) + within(10.seconds) { + establishIdleSession(harness) + harness.peer.sendText("response") + harness.plainOut.expectNoMessage(200.millis) + + harness.peer.cancelCipherOut() + // Cancellation alone must not complete the plaintext output. + harness.plainOut.expectNoMessage(200.millis) + + harness.plainOut.request(2) + expectSessionBytes(harness.plainOut.expectNext(), "response") + harness.plainOut.expectNext() shouldBe SessionTruncated + harness.plainOut.expectComplete() + } + } + + s"cancel plaintext input before waiting to deliver buffered input ($engine, $protocol)" in { + assertAllStagesStopped { + val harness = newHarness(protocol, useLegacyActor) + establishIdleSession(harness) + harness.peer.sendText("response") + harness.peer.cancelCipherOut() + + harness.plainIn.expectCancellation() + harness.plainOut.request(2) + expectSessionBytes(harness.plainOut.expectNext(), "response") + harness.plainOut.expectNext(SessionTruncated) + harness.plainOut.expectComplete() + harness.peer.expectCipherInCancellation() + } + } + + s"finish an already initiated close handshake after transport output cancellation ($engine, $protocol)" in { + assertAllStagesStopped { + val harness = newHarness(protocol, useLegacyActor, EagerClose) + establishIdleSession(harness) + harness.plainIn.sendComplete() + harness.peer.receiveCloseNotify() + harness.peer.cancelCipherOut() + harness.peer.sendCloseNotify() + + harness.plainOut.request(1) + harness.plainOut.expectComplete() + harness.peer.expectCipherInCancellation() + } + } + + s"stop waiting for the peer close when both outputs are cancelled ($engine, $protocol)" in { + assertAllStagesStopped { + val harness = newHarness(protocol, useLegacyActor, EagerClose) + establishIdleSession(harness) + harness.plainIn.sendComplete() + harness.peer.receiveCloseNotify() + harness.peer.cancelCipherOut() + harness.plainOut.cancel() + harness.peer.expectCipherInCancellation() + } + } + + s"report cancellation of a partial record without waiting for more ciphertext ($engine, $protocol)" in { + assertAllStagesStopped { + val harness = newHarness(protocol, useLegacyActor) + establishIdleSession(harness) + harness.peer.sendPartialRecord("response") + harness.peer.cancelCipherOut() + harness.plainOut.request(1) + + harness.plainOut.expectNext(SessionTruncated) + harness.plainOut.expectComplete() + harness.peer.expectCipherInCancellation() + harness.plainIn.expectCancellation() + } + } + + s"assemble a partial record while the transport output stays open ($engine, $protocol)" in { + assertAllStagesStopped { + val harness = newHarness(protocol, useLegacyActor) + establishIdleSession(harness) + val remainder = harness.peer.sendPartialRecord("response") + harness.plainOut.request(1) + harness.plainOut.expectNoMessage(100.millis) + harness.peer.sendCiphertext(remainder) + expectSessionBytes(harness.plainOut.expectNext(), "response") + + harness.peer.cancelCipherOut() + harness.plainOut.request(1) + harness.plainOut.expectNext(SessionTruncated) + harness.plainOut.expectComplete() + harness.peer.expectCipherInCancellation() + } + } + + s"report a truncated session when the buffered records outnumber the plaintext demand ($engine, $protocol)" in { + val harness = newHarness(protocol, useLegacyActor) + within(10.seconds) { + establishIdleSession(harness) + List("first", "second", "third").foreach { text => + harness.peer.sendText(text) + harness.plainOut.expectNoMessage(200.millis) + } + + harness.peer.cancelCipherOut() + + // One element at a time: the truncation report must survive demand that is only granted + // after the last buffered record was already delivered. + harness.plainOut.request(1) + expectSessionBytes(harness.plainOut.expectNext(), "first") + harness.plainOut.request(1) + harness.plainOut.expectNext() shouldBe SessionTruncated + harness.plainOut.expectComplete() + harness.peer.expectCipherInCancellation() + } + } + + s"deliver every buffered record while the transport output stays open ($engine, $protocol)" in { + val harness = newHarness(protocol, useLegacyActor) + within(10.seconds) { + establishIdleSession(harness) + harness.peer.sendText("first") + harness.plainOut.expectNoMessage(200.millis) + harness.peer.sendText("second") + harness.plainOut.expectNoMessage(200.millis) + + harness.plainOut.request(2) + expectSessionBytes(harness.plainOut.expectNext(), "first") + expectSessionBytes(harness.plainOut.expectNext(), "second") + // Nothing was abandoned, so no truncation may be reported and the stage stays open. + harness.plainOut.expectNoMessage(200.millis) + + harness.peer.cancelCipherOut() + harness.plainOut.cancel() + harness.peer.expectCipherInCancellation() + } + } + } + } + + "The legacy actor TLS stage" must { + Protocols.foreach { protocol => + s"cancel its read side when plaintext arrives after the transport output was cancelled ($protocol)" in { + val harness = newHarness(protocol, useLegacyActor = true) + within(10.seconds) { + establishIdleSession(harness) + + // Cancellation leaves the transport output's demand flag set, so plaintext arriving + // afterwards still makes the outbound side runnable and reaches the teardown path while + // the plaintext output has no demand at all. + harness.peer.cancelCipherOut() + harness.plainIn.sendNext(SendBytes(ByteString("tail"))) + + // Parking the pump to report a truncation must release both upstream subscriptions. + harness.peer.expectCipherInCancellation() + harness.plainIn.expectCancellation() + + harness.plainOut.request(1) + harness.plainOut.expectNext() shouldBe SessionTruncated + harness.plainOut.expectComplete() + } + } + } + } +} diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/io/TLSActor.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/io/TLSActor.scala index 7871dd95354..cf2cfc89a1d 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/io/TLSActor.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/io/TLSActor.scala @@ -71,10 +71,21 @@ import pekko.util.ByteString import TLSActor._ private var stopped = false + private var terminationCompleted = false + private var awaitingPeerClose = false private val maxTLSIterations = 1000 private var unwrapPutBackCounter: Int = 0 - protected val outputBunch = new OutputBunch(outputCount = 2, self, this) + protected val outputBunch: OutputBunch = new OutputBunch(outputCount = 2, self, this) { + override def onCancel(output: Int): Unit = + if (output == TransportOut) { + inputBunch.cancel(UserIn) + // Once close_notify has been sent, only the peer's reply is needed for a clean shutdown. + if (!awaitingPeerClose || userOutDemand.isCompleted) nextPhase(transportOutputCancelled) + } else if (output == UserOut && outputBunch.isCancelled(TransportOut)) { + nextPhase(completedPhase) + } + } outputBunch.markAllOutputs() protected val inputBunch = new InputBunch(inputCount = 2, maxInputBufferSize, this) { @@ -165,6 +176,7 @@ import pekko.util.ByteString var lastHandshakeStatus: HandshakeStatus = null var corkUser = true + private var sessionEstablished = false // The engine could also be instantiated in ActorMaterializerImpl but if creation fails // during materialization it would be worse than failing later on. @@ -238,9 +250,9 @@ import pekko.util.ByteString } // bidirectional case + private val userOutDemand = outputBunch.demandAvailableFor(UserOut) val outbound = (userHasData || engineNeedsWrap) && outputBunch.demandAvailableFor(TransportOut) - val inbound = - (transportInChoppingBlock && outputBunch.demandAvailableFor(UserOut) && engineNotBlockedOnWrap) || userOutCancelled + val inbound = (transportInChoppingBlock && userOutDemand && engineNotBlockedOnWrap) || userOutCancelled // half-closed val outboundHalfClosed = engineNeedsWrap && outputBunch.demandAvailableFor(TransportOut) @@ -287,13 +299,46 @@ import pekko.util.ByteString } } + private val signallingInboundTruncation = TransferPhase(userOutDemand) { () => + if (tracing) log.debug("signalling SessionTruncated after TransportOut cancellation") + outputBunch.enqueue(UserOut, SessionTruncated) + nextPhase(completedPhase) + } + + private def transportInDepleted: Boolean = + inputBunch.isDepleted(TransportIn) && transportInChoppingBlock.isEmpty + + // Cancellation must also wake an idle pump, unless it is already waiting for a peer close_notify. + private val transportOutputCancelled = TransferPhase(new TransferState { + def isReady = TlsEngineHelpers.transportOutputCancellationReady( + userOutputDemand = userOutDemand.isReady, + transportInputPending = transportInChoppingBlock.isReady && !transportInDepleted, + sessionEstablished = sessionEstablished, + inboundClosed = engine.isInboundDone, + userOutputClosed = userOutDemand.isCompleted) + // fail() can re-enter the pump; do not execute another cancellation round after failure. + def isCompleted = stopped + }) { () => + // Depletion is also ready, but cancellation must not run doInbound's closeInbound path. + if (inbound.isReady && !userOutDemand.isCompleted && !transportInDepleted) + doInbound(isOutboundClosed = true, inbound) + if (!stopped) { + // Do not closeInbound here: some providers invalidate a session missing close_notify. + val reportTruncation = TlsEngineHelpers.shouldReportInboundTruncation( + sessionEstablished, engine.isInboundDone, userOutDemand.isCompleted) + inputBunch.cancel() + if (reportTruncation) nextPhase(signallingInboundTruncation) + else nextPhase(completedPhase) + } + } + def completeOrFlush(): Unit = if (engine.isOutboundDone || (engine.isInboundDone && userInChoppingBlock.isEmpty && !userInBuffer.hasRemaining)) nextPhase(completedPhase) else nextPhase(flushingOutbound) private def doInbound(isOutboundClosed: Boolean, inboundState: TransferState): Boolean = - if (inputBunch.isDepleted(TransportIn) && transportInChoppingBlock.isEmpty) { + if (transportInDepleted) { if (tracing) log.debug("closing inbound") try engine.closeInbound() catch { case _: SSLException => outputBunch.enqueue(UserOut, SessionTruncated) } @@ -410,7 +455,10 @@ import pekko.util.ByteString case CLOSED => flushToTransport() if (engine.isInboundDone) nextPhase(completedPhase) - else nextPhase(awaitingClose) + else { + awaitingPeerClose = true + nextPhase(awaitingClose) + } case s => fail(new IllegalStateException(s"unexpected status $s in doWrap()")) } } @@ -482,6 +530,7 @@ import pekko.util.ByteString case Success(()) => currentSession = session corkUser = false + sessionEstablished = true flushToUser() case Failure(ex) => fail(ex, closeTransport = true) @@ -497,12 +546,14 @@ import pekko.util.ByteString protected def fail(e: Throwable, closeTransport: Boolean = true): Unit = { if (tracing) log.debug("fail {} due to: {}", self, e.getMessage) - inputBunch.cancel() + // Publish the failure before cancelling the peer's output: its prompt cancellation handling + // can otherwise cancel our transport output before the error reaches its subscriber. if (closeTransport) { log.debug("closing output") outputBunch.error(TransportOut, e) } outputBunch.error(UserOut, e) + inputBunch.cancel() stopped = true pump() } @@ -512,9 +563,9 @@ import pekko.util.ByteString val e = new RuntimeException( s"Unexpected termination of TLS actor for connection to ${engine.getPeerHost}:${engine.getPeerPort}") log.warning(e.getMessage) - inputBunch.cancel() outputBunch.error(TransportOut, e) outputBunch.error(UserOut, e) + inputBunch.cancel() } else if (tracing) { log.debug("postStop") } @@ -523,13 +574,15 @@ import pekko.util.ByteString override protected def pumpFailed(e: Throwable): Unit = fail(e) - override protected def pumpFinished(): Unit = { - inputBunch.cancel() - outputBunch.complete() - if (tracing) log.debug("STOP Outbound Closed: {} Inbound closed: {}", engine.isOutboundDone, engine.isInboundDone) - stopped = true - context.stop(self) - } + override protected def pumpFinished(): Unit = + if (!terminationCompleted) { + terminationCompleted = true + inputBunch.cancel() + outputBunch.complete() + if (tracing) log.debug("STOP Outbound Closed: {} Inbound closed: {}", engine.isOutboundDone, engine.isInboundDone) + stopped = true + context.stop(self) + } } /** diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsEngineHelpers.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsEngineHelpers.scala index 339cf12caf3..a3f9c109cae 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsEngineHelpers.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsEngineHelpers.scala @@ -28,6 +28,21 @@ import pekko.annotation.InternalApi */ @InternalApi private[stream] object TlsEngineHelpers { + def shouldReportInboundTruncation( + sessionEstablished: Boolean, + inboundClosed: Boolean, + userOutputClosed: Boolean): Boolean = + sessionEstablished && !inboundClosed && !userOutputClosed + + def transportOutputCancellationReady( + userOutputDemand: Boolean, + transportInputPending: Boolean, + sessionEstablished: Boolean, + inboundClosed: Boolean, + userOutputClosed: Boolean): Boolean = + userOutputDemand || !transportInputPending || + !shouldReportInboundTruncation(sessionEstablished, inboundClosed, userOutputClosed) + @inline def emptyReadBuffer(buffer: ByteBuffer): Unit = { buffer.clear().flip() 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 7107c634804..2dd4dc6ada3 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 @@ -73,6 +73,8 @@ import pekko.util.ByteString private var phase: Byte = BidirectionalPhase.toByte private var deferredTransportFlushes = 0 private var stateBits = 0 + private var sessionEstablished = false + private var inboundTruncationPending = false private var transportOutBuffer: ByteBuffer = _ private var userOutBuffer: ByteBuffer = _ @@ -205,6 +207,13 @@ import pekko.util.ByteString case OutboundClosedPhase => (outboundHalfClosedReady || inboundReady) && !outboundClosedCompleted case InboundClosedPhase => (outboundReady || inboundHalfClosedReady || mustCloseOutboundOnCancel) && !inboundClosedCompleted + case TransportOutputCancelledPhase => + TlsEngineHelpers.transportOutputCancellationReady( + userOutputDemand = userOutput.isDemandAvailable, + transportInputPending = transportInput.isReady && !transportInput.isDepleted, + sessionEstablished = sessionEstablished, + inboundClosed = engine.isInboundDone, + userOutputClosed = userOutput.isClosed) case _ => false } @@ -216,11 +225,20 @@ import pekko.util.ByteString case OutboundClosedPhase => outboundClosedCompleted case InboundClosedPhase => inboundClosedCompleted case CompletedPhase => true - case _ => false + // Only runPhase completes cancellation, after the final inbound round and truncation decision. + case TransportOutputCancelledPhase => false + case _ => false } private def runPhase(): Unit = (phase.toInt: @switch) match { + case TransportOutputCancelledPhase => + // Depletion is also ready, but cancellation must not run doInbound's closeInbound path. + if (inboundReady && !userOutput.isClosed && !transportInput.isDepleted) + doInbound(isOutboundClosed = true, inboundHalfClosedMode = false) + signalInboundTruncation() + nextPhase(CompletedPhase) + case BidirectionalPhase => val continue = doInbound(isOutboundClosed = false, inboundHalfClosedMode = false) if (continue) doOutbound(isInboundClosed = false) @@ -575,6 +593,7 @@ import pekko.util.ByteString verifySession(session) match { case Success(()) => currentSession = session + sessionEstablished = true stateBits |= PlainDataAllowedFlag transportPacketSize = math.max(1, session.getPacketBufferSize) applicationRecordSize = math.max(1, session.getApplicationBufferSize) @@ -733,12 +752,28 @@ import pekko.util.ByteString case NonFatal(_) => () } + // Do not closeInbound here: a missing + // close_notify can invalidate the SSLSession. Record the signal separately so an already + // pending plaintext element is delivered first, including during renegotiation. + private def signalInboundTruncation(): Unit = + if (TlsEngineHelpers.shouldReportInboundTruncation( + sessionEstablished, engine.isInboundDone, userOutput.isClosed)) + inboundTruncationPending = true + + private def completeUserOutput(): Unit = { + if (inboundTruncationPending && (userOutput.isClosed || !userOutput.hasPending)) { + inboundTruncationPending = false + userOutput.enqueue(SessionTruncated) + } + if (!inboundTruncationPending) userOutput.completeOutput() + } + private def pumpFinished(): Unit = if ((stateBits & StageClosedFlag) == 0) { userInput.cancel() transportInput.cancel() transportOutput.completeOutput() - userOutput.completeOutput() + completeUserOutput() stateBits |= StageCompletionPendingFlag completeStageIfReady() } @@ -746,6 +781,7 @@ import pekko.util.ByteString private def completeStageIfReady(): Unit = if ((stateBits & StageCompletionPendingFlag) != 0 && (stateBits & StageClosedFlag) == 0 && + !inboundTruncationPending && !userOutput.hasPending && !transportOutput.hasPending) { stateBits |= StageClosedFlag @@ -1152,6 +1188,7 @@ import pekko.util.ByteString scheduleOnce(InitialOutputFlushTimer, Duration.Zero) } else { pushPending() + if ((stateBits & StageCompletionPendingFlag) != 0) completeUserOutput() completeStageIfReady() runPump() } @@ -1171,6 +1208,7 @@ import pekko.util.ByteString override def onDownstreamFinish(cause: Throwable): Unit = { onCancel() + if ((stateBits & StageCompletionPendingFlag) != 0) completeUserOutput() completeStageIfReady() runPump() } @@ -1188,6 +1226,16 @@ import pekko.util.ByteString def onCancel(): Unit = { stateBits |= cancelledFlag pending = null.asInstanceOf[T] + if (cancelledFlag == TransportOutputCancelledFlag && + (stateBits & (StageClosedFlag | StageCompletionPendingFlag)) == 0) { + userInput.cancel() + stateBits |= InitialPumpEnabledFlag + // Once close_notify has been sent, only the peer's reply is needed for a clean shutdown. + if (phase != AwaitingClosePhase || userOutput.isClosed) nextPhase(TransportOutputCancelledPhase) + } else if (cancelledFlag == UserOutputCancelledFlag && transportOutput.isCancelled && + phase == AwaitingClosePhase) { + nextPhase(CompletedPhase) + } } def completeOutput(): Unit = @@ -1249,4 +1297,5 @@ import pekko.util.ByteString private[pekko] final val OutboundClosedPhase = 4 private[pekko] final val InboundClosedPhase = 5 private[pekko] final val CompletedPhase = 6 + private[pekko] final val TransportOutputCancelledPhase = 7 }