From 959ef191e8213c578d912c524c6537a61a881f12 Mon Sep 17 00:00:00 2001 From: Kenny Root Date: Tue, 4 Aug 2026 21:46:26 -0700 Subject: [PATCH] fix(channels): preserve unread data on remote close CHANNEL_CLOSE can arrive while a delivery pump is suspended on a "rendezvous" send or while ingress data remains queued. Cancelling the pump then drops stdout, stderr, extended-data, or forwarding tails (depending on the channel type) and reports a misleading clean EOF. Gracefully drain accepted data after remote closure while keeping caller-requested close and disconnect as an abort. Explicit close also releases pumps retained after remote closure, delayed delivery no longer returns stale window credit, and the state-machine model checks valid inbound-stream closure effects. Add unit and FakeServer coverage for delayed session and forwarding reads, explicit aborts, direct TCP forwarding, and a deterministic 1 MiB multi-window transfer whose final packet remains parked until CLOSE is processed. This should exercise the issues that were being seen without relying on probabalistic tests. Fixes #245 --- .../sshlib/client/ForwardingChannel.kt | 36 ++++-- .../sshlib/client/SessionChannel.kt | 67 +++++++----- .../sshlib/protocol/SshChannelStateMachine.kt | 8 +- .../connectbot/sshlib/client/FakeSshServer.kt | 11 ++ .../sshlib/client/ForwardingChannelTest.kt | 27 +++++ .../sshlib/client/SessionChannelTest.kt | 57 ++++++++++ .../sshlib/client/SshConnectionFlowTest.kt | 103 ++++++++++++++++++ .../protocol/SshChannelStateMachineTest.kt | 6 +- .../protocol/SshStateMachineTlaGenerator.kt | 11 ++ .../resources/tla/SshClientStateMachine.cfg | 2 + .../resources/tla/SshClientStateMachine.tla | 17 +++ .../tla/SshClientStateMachineGenerated.tla | 18 +-- 12 files changed, 315 insertions(+), 48 deletions(-) diff --git a/sshlib/src/main/kotlin/org/connectbot/sshlib/client/ForwardingChannel.kt b/sshlib/src/main/kotlin/org/connectbot/sshlib/client/ForwardingChannel.kt index c359e78..97eeceb 100644 --- a/sshlib/src/main/kotlin/org/connectbot/sshlib/client/ForwardingChannel.kt +++ b/sshlib/src/main/kotlin/org/connectbot/sshlib/client/ForwardingChannel.kt @@ -49,13 +49,17 @@ internal class ForwardingChannel( private val incomingIngress = Channel(Channel.UNLIMITED) private val _incomingData = Channel(Channel.RENDEZVOUS) + + @Volatile private var inboundDeliveryOpen = true val incomingData: ReceiveChannel get() = _incomingData private val incomingDeliveryJob = connectionScope.launch { try { for (data in incomingIngress) { _incomingData.send(data) val adjust = window.releaseLocal(data.size) - connection.sendWindowAdjust(remoteChannelNumber, adjust) + if (inboundDeliveryOpen) { + connection.sendWindowAdjust(remoteChannelNumber, adjust) + } } } finally { _incomingData.close() @@ -90,7 +94,7 @@ internal class ForwardingChannel( internal suspend fun onEof() { if (!lifecycle.receiveEof { logger.debug("Forwarding channel $localChannelNumber received EOF") - incomingIngress.close() + finishInboundDelivery() } ) { throw SshException("Received duplicate EOF or EOF after CLOSE on forwarding channel $localChannelNumber") @@ -100,6 +104,7 @@ internal class ForwardingChannel( internal suspend fun onClose() { if (!lifecycle.receiveClose { transition -> logger.debug("Forwarding channel $localChannelNumber closed") + finishInboundDelivery() if (SshChannelEffect.SEND_CLOSE in transition.effects) { try { connection.sendChannelClose(remoteChannelNumber) @@ -107,9 +112,6 @@ internal class ForwardingChannel( logger.debug("Failed to send CHANNEL_CLOSE reply", e) } } - incomingIngress.close() - incomingDeliveryJob.cancel() - _incomingData.close() windowAvailable.close() if (SshChannelEffect.CLOSE_CHANNEL in transition.effects) { connection.notifyChannelClosed(localChannelNumber) @@ -122,9 +124,7 @@ internal class ForwardingChannel( internal suspend fun onDisconnected() { lifecycle.disconnect { transition -> - incomingIngress.close() - incomingDeliveryJob.cancel() - _incomingData.close() + abortInboundDelivery() windowAvailable.close() if (SshChannelEffect.CLOSE_CHANNEL in transition.effects) { connection.notifyChannelClosed(localChannelNumber) @@ -134,6 +134,17 @@ internal class ForwardingChannel( internal suspend fun receiveRequest(action: suspend () -> Unit): Boolean = lifecycle.receiveRequest { action() } + private fun finishInboundDelivery() { + inboundDeliveryOpen = false + incomingIngress.close() + } + + private fun abortInboundDelivery() { + finishInboundDelivery() + incomingDeliveryJob.cancel() + _incomingData.close() + } + suspend fun sendData(data: ByteArray) { var offset = 0 while (offset < data.size) { @@ -159,13 +170,16 @@ internal class ForwardingChannel( suspend fun close() { lifecycle.sendClose { transition -> - incomingIngress.close() - incomingDeliveryJob.cancel() - _incomingData.close() + abortInboundDelivery() + windowAvailable.close() connection.sendChannelClose(remoteChannelNumber) if (SshChannelEffect.CLOSE_CHANNEL in transition.effects) { connection.notifyChannelClosed(localChannelNumber) } } + // A remote CLOSE makes sendClose a no-op, but close() still owns + // releasing any unread delivery job retained for graceful draining. + abortInboundDelivery() + windowAvailable.close() } } diff --git a/sshlib/src/main/kotlin/org/connectbot/sshlib/client/SessionChannel.kt b/sshlib/src/main/kotlin/org/connectbot/sshlib/client/SessionChannel.kt index 40ba22d..93fc98c 100644 --- a/sshlib/src/main/kotlin/org/connectbot/sshlib/client/SessionChannel.kt +++ b/sshlib/src/main/kotlin/org/connectbot/sshlib/client/SessionChannel.kt @@ -71,6 +71,8 @@ class SessionChannel internal constructor( private val _stderr = Channel(Channel.RENDEZVOUS) private val _extendedData = Channel>(Channel.RENDEZVOUS) + @Volatile private var inboundDeliveryOpen = true + private val stdoutDeliveryJob = connectionScope.launch { deliverData(stdoutIngress, _stdout) { it.size } } @@ -150,7 +152,9 @@ class SessionChannel internal constructor( for (value in ingress) { output.send(value) val adjust = window.releaseLocal(sizeOf(value)) - connection.sendWindowAdjust(_remoteChannelNumber, adjust) + if (inboundDeliveryOpen) { + connection.sendWindowAdjust(_remoteChannelNumber, adjust) + } } } finally { output.close() @@ -171,9 +175,7 @@ class SessionChannel internal constructor( internal suspend fun onEof() { if (!lifecycle.receiveEof { logger.debug("Received EOF on channel $localChannelNumber") - stdoutIngress.close() - stderrIngress.close() - extendedDataIngress.close() + finishInboundDelivery() } ) { throw org.connectbot.sshlib.SshException("Received duplicate EOF or EOF after CLOSE on channel $localChannelNumber") @@ -182,7 +184,11 @@ class SessionChannel internal constructor( internal suspend fun onClose() { if (!lifecycle.receiveClose { transition -> - closeResources(SshChannelEffect.SEND_CLOSE in transition.effects, "Received CLOSE") + closeResources( + replyRequired = SshChannelEffect.SEND_CLOSE in transition.effects, + preserveInbound = SshChannelEffect.CLOSE_INBOUND_STREAMS in transition.effects, + reason = "Received CLOSE", + ) if (SshChannelEffect.CLOSE_CHANNEL in transition.effects) { connection.notifyChannelClosed(localChannelNumber) } @@ -192,12 +198,34 @@ class SessionChannel internal constructor( } } - private suspend fun closeResources(replyRequired: Boolean, reason: String) { + private fun finishInboundDelivery() { + inboundDeliveryOpen = false + stdoutIngress.close() + stderrIngress.close() + extendedDataIngress.close() + } + + private fun abortInboundDelivery() { + finishInboundDelivery() + stdoutDeliveryJob.cancel() + stderrDeliveryJob.cancel() + extendedDeliveryJob.cancel() + _stdout.close() + _stderr.close() + _extendedData.close() + } + + private suspend fun closeResources(replyRequired: Boolean, preserveInbound: Boolean, reason: String) { logger.debug("$reason on channel $localChannelNumber") obfuscatorMutex.withLock { obfuscator?.stop() } chaffJob?.cancel() + if (preserveInbound) { + finishInboundDelivery() + } else { + abortInboundDelivery() + } if (replyRequired) { try { connection.sendChannelClose(_remoteChannelNumber) @@ -205,15 +233,6 @@ class SessionChannel internal constructor( logger.debug("Failed to send CHANNEL_CLOSE reply", e) } } - stdoutIngress.close() - stderrIngress.close() - extendedDataIngress.close() - stdoutDeliveryJob.cancel() - stderrDeliveryJob.cancel() - extendedDeliveryJob.cancel() - _stdout.close() - _stderr.close() - _extendedData.close() windowAvailable.close() // Channel is gone; if the server never reported an exit, resolve // waiters with "unknown" rather than leaving them suspended. @@ -222,7 +241,7 @@ class SessionChannel internal constructor( internal suspend fun onDisconnected() { lifecycle.disconnect { transition -> - closeResources(replyRequired = false, reason = "Disconnected") + closeResources(replyRequired = false, preserveInbound = false, reason = "Disconnected") if (SshChannelEffect.CLOSE_CHANNEL in transition.effects) { connection.notifyChannelClosed(localChannelNumber) } @@ -461,15 +480,8 @@ class SessionChannel internal constructor( logger.debug("Closing channel $localChannelNumber") obfuscatorMutex.withLock { obfuscator?.stop() } chaffJob?.cancel() - stdoutIngress.close() - stderrIngress.close() - extendedDataIngress.close() - stdoutDeliveryJob.cancel() - stderrDeliveryJob.cancel() - extendedDeliveryJob.cancel() - _stdout.close() - _stderr.close() - _extendedData.close() + abortInboundDelivery() + windowAvailable.close() _exitInfo.complete(null) try { connection.sendChannelClose(_remoteChannelNumber) @@ -480,6 +492,11 @@ class SessionChannel internal constructor( connection.notifyChannelClosed(localChannelNumber) } } + // A remote CLOSE makes sendClose a no-op, but close() still owns + // releasing any unread delivery job retained for graceful draining. + abortInboundDelivery() + windowAvailable.close() + _exitInfo.complete(null) } } } diff --git a/sshlib/src/main/kotlin/org/connectbot/sshlib/protocol/SshChannelStateMachine.kt b/sshlib/src/main/kotlin/org/connectbot/sshlib/protocol/SshChannelStateMachine.kt index 7fa0d42..7cf917d 100644 --- a/sshlib/src/main/kotlin/org/connectbot/sshlib/protocol/SshChannelStateMachine.kt +++ b/sshlib/src/main/kotlin/org/connectbot/sshlib/protocol/SshChannelStateMachine.kt @@ -308,14 +308,18 @@ internal class SshChannelStateMachine( class SendClose(action: suspend (SshChannelAcceptedTransition) -> Unit) : ChannelEvent( SshChannelEventId.SEND_CLOSE, - setOf(SshChannelEffect.SEND_CLOSE), + setOf(SshChannelEffect.SEND_CLOSE, SshChannelEffect.CLOSE_INBOUND_STREAMS), SshChannelEventOrigin.LOCAL_COMMAND, action, ) class ReceiveClose(action: suspend (SshChannelAcceptedTransition) -> Unit) : ChannelEvent( SshChannelEventId.RECEIVE_CLOSE, - setOf(SshChannelEffect.SEND_CLOSE, SshChannelEffect.CLOSE_CHANNEL), + setOf( + SshChannelEffect.SEND_CLOSE, + SshChannelEffect.CLOSE_INBOUND_STREAMS, + SshChannelEffect.CLOSE_CHANNEL, + ), SshChannelEventOrigin.PARSED_PACKET, action, ) diff --git a/sshlib/src/test/kotlin/org/connectbot/sshlib/client/FakeSshServer.kt b/sshlib/src/test/kotlin/org/connectbot/sshlib/client/FakeSshServer.kt index 54993f5..4e949fe 100644 --- a/sshlib/src/test/kotlin/org/connectbot/sshlib/client/FakeSshServer.kt +++ b/sshlib/src/test/kotlin/org/connectbot/sshlib/client/FakeSshServer.kt @@ -45,6 +45,7 @@ import org.connectbot.sshlib.protocol.SshMsgChannelOpenConfirmation import org.connectbot.sshlib.protocol.SshMsgChannelOpenFailure import org.connectbot.sshlib.protocol.SshMsgChannelRequest import org.connectbot.sshlib.protocol.SshMsgChannelSuccess +import org.connectbot.sshlib.protocol.SshMsgChannelWindowAdjust import org.connectbot.sshlib.protocol.SshMsgDisconnect import org.connectbot.sshlib.protocol.SshMsgExtInfo import org.connectbot.sshlib.protocol.SshMsgIgnore @@ -118,6 +119,7 @@ class FakeSshServer( private val receivedChannelOpenConfirmations = Channel(Channel.UNLIMITED) private val receivedChannelOpenFailures = Channel(Channel.UNLIMITED) private val receivedChannelData = Channel(Channel.UNLIMITED) + private val receivedChannelWindowAdjusts = Channel(Channel.UNLIMITED) private val receivedUnimplemented = Channel(Channel.UNLIMITED) fun start(ignoreTransportErrors: Boolean = false) { @@ -270,6 +272,13 @@ class FakeSshServer( receivedChannelData.trySend(data) } + SshEnums.MessageType.SSH_MSG_CHANNEL_WINDOW_ADJUST -> { + val bodyBytes = rawBytes.copyOfRange(1, rawBytes.size) + val adjust = SshMsgChannelWindowAdjust(ByteBufferKaitaiStream(bodyBytes)) + adjust._read() + receivedChannelWindowAdjusts.trySend(adjust) + } + SshEnums.MessageType.SSH_MSG_PING -> { val bodyBytes = rawBytes.copyOfRange(1, rawBytes.size) val pingMsg = SshMsgPing(ByteBufferKaitaiStream(bodyBytes)) @@ -952,4 +961,6 @@ class FakeSshServer( suspend fun awaitChannelOpenFailure(): SshMsgChannelOpenFailure = receivedChannelOpenFailures.receive() suspend fun awaitChannelData(): SshMsgChannelData = receivedChannelData.receive() + + suspend fun awaitChannelWindowAdjust(): SshMsgChannelWindowAdjust = receivedChannelWindowAdjusts.receive() } diff --git a/sshlib/src/test/kotlin/org/connectbot/sshlib/client/ForwardingChannelTest.kt b/sshlib/src/test/kotlin/org/connectbot/sshlib/client/ForwardingChannelTest.kt index acf1061..84e7051 100644 --- a/sshlib/src/test/kotlin/org/connectbot/sshlib/client/ForwardingChannelTest.kt +++ b/sshlib/src/test/kotlin/org/connectbot/sshlib/client/ForwardingChannelTest.kt @@ -108,6 +108,33 @@ class ForwardingChannelTest { assertTrue(channel.incomingData.isClosedForReceive) } + @Test + fun `remote close preserves unread incoming data until consumed`() = runTest { + val (channel, conn) = createChannel() + val first = "first".toByteArray() + val second = "second".toByteArray() + + channel.onData(first) + channel.onData(second) + channel.onClose() + + assertArrayEquals(first, channel.incomingData.receive()) + assertArrayEquals(second, channel.incomingData.receive()) + assertTrue(channel.incomingData.receiveCatching().isClosed) + coVerify(exactly = 0) { conn.sendWindowAdjust(any(), any()) } + } + + @Test + fun `explicit close after remote close discards abandoned incoming data`() = runTest { + val (channel, _) = createChannel() + + channel.onData("abandoned".toByteArray()) + channel.onClose() + channel.close() + + assertTrue(channel.incomingData.receiveCatching().isClosed) + } + @Test fun `sendData chunks data by maxPacketSize`() = runTest { val conn = mockk(relaxed = true) diff --git a/sshlib/src/test/kotlin/org/connectbot/sshlib/client/SessionChannelTest.kt b/sshlib/src/test/kotlin/org/connectbot/sshlib/client/SessionChannelTest.kt index 2bb93c6..eabff87 100644 --- a/sshlib/src/test/kotlin/org/connectbot/sshlib/client/SessionChannelTest.kt +++ b/sshlib/src/test/kotlin/org/connectbot/sshlib/client/SessionChannelTest.kt @@ -153,6 +153,63 @@ class SessionChannelTest { assertArrayEquals(testData, received) } + @Test + fun `remote close preserves unread stdout until consumed`() = runTest { + val (channel, conn) = createChannel() + val first = "first".toByteArray() + val second = "second".toByteArray() + + channel.onData(first) + channel.onData(second) + channel.onClose() + + assertArrayEquals(first, channel.stdout.receive()) + assertArrayEquals(second, channel.stdout.receive()) + assertTrue(channel.stdout.receiveCatching().isClosed) + coVerify(exactly = 0) { conn.sendWindowAdjust(any(), any()) } + } + + @Test + fun `remote close preserves unread stderr and extended data`() = runTest { + val (channel, _) = createChannel() + val stderr = "error".toByteArray() + val extended = "extended".toByteArray() + + channel.onExtendedData(1, stderr) + channel.onExtendedData(7, extended) + channel.onClose() + + assertArrayEquals(stderr, channel.stderr.receive()) + val receivedExtended = channel.readExtended() + assertEquals(7, receivedExtended?.first) + assertArrayEquals(extended, receivedExtended?.second) + assertTrue(channel.stderr.receiveCatching().isClosed) + assertEquals(null, channel.readExtended()) + } + + @Test + fun `remote EOF preserves unread stdout until consumed`() = runTest { + val (channel, _) = createChannel() + val data = "tail".toByteArray() + + channel.onData(data) + channel.onEof() + + assertArrayEquals(data, channel.stdout.receive()) + assertTrue(channel.stdout.receiveCatching().isClosed) + } + + @Test + fun `explicit close after remote close discards abandoned output`() = runTest { + val (channel, _) = createChannel() + + channel.onData("abandoned".toByteArray()) + channel.onClose() + channel.close() + + assertTrue(channel.stdout.receiveCatching().isClosed) + } + @Test fun `close marks channel not open and sends channel close`() = runTest { val (channel, conn) = createChannel() diff --git a/sshlib/src/test/kotlin/org/connectbot/sshlib/client/SshConnectionFlowTest.kt b/sshlib/src/test/kotlin/org/connectbot/sshlib/client/SshConnectionFlowTest.kt index 4fc0e63..bf278d0 100644 --- a/sshlib/src/test/kotlin/org/connectbot/sshlib/client/SshConnectionFlowTest.kt +++ b/sshlib/src/test/kotlin/org/connectbot/sshlib/client/SshConnectionFlowTest.kt @@ -47,6 +47,7 @@ import org.connectbot.sshlib.crypto.PrivateKeyReader import org.connectbot.sshlib.crypto.SshPublicKeyEncoder import org.connectbot.sshlib.transport.PipedTransport import org.junit.jupiter.api.Test +import java.io.ByteArrayOutputStream import java.nio.ByteBuffer import java.nio.file.Files import java.nio.file.Paths @@ -452,6 +453,82 @@ class SshConnectionFlowTest { } } + @Test + fun `session channel preserves data received before close until consumer reads`() = runTest { + connectedFixture { connection, server, dispatcher -> + connection.autoDisconnectOnLastChannelClose = false + authenticate(connection, server, dispatcher) + val session = openSession(connection, server, dispatcher) + val localChannel = session.localChannelNumber + + server.sendChannelData(localChannel, byteArrayOf(1, 2, 3)) + server.sendChannelExtendedData(localChannel, 1, byteArrayOf(4, 5)) + server.sendChannelExtendedData(localChannel, 7, byteArrayOf(6, 7)) + server.sendChannelClose(localChannel) + withTimeout(5_000) { + while (session.isOpen) yield() + } + + assertContentEquals(byteArrayOf(1, 2, 3), withTimeout(5_000) { session.stdout.receive() }) + assertContentEquals(byteArrayOf(4, 5), withTimeout(5_000) { session.stderr.receive() }) + val extended = withTimeout(5_000) { session.readExtended() } + assertEquals(7, extended?.first) + assertContentEquals(byteArrayOf(6, 7), extended?.second) + assertNull(withTimeout(5_000) { session.read() }) + assertNull(withTimeout(5_000) { session.stderr.receiveCatching().getOrNull() }) + assertNull(withTimeout(5_000) { session.readExtended() }) + } + } + + @Test + fun `session channel preserves a multi-window stdout tail when close follows data`() = runTest { + connectedFixture { connection, server, dispatcher -> + connection.autoDisconnectOnLastChannelClose = false + authenticate(connection, server, dispatcher) + val session = openSession(connection, server, dispatcher) + val localChannel = session.localChannelNumber + val packetSize = 32 * 1024 + val expected = ByteArray(1024 * 1024) { index -> (index % 251).toByte() } + val tailReceiverBlocked = CompletableDeferred() + val releaseTailReceiver = CompletableDeferred() + val received = async(dispatcher) { + val output = ByteArrayOutputStream(expected.size) + for (chunk in session.stdout) { + output.write(chunk) + if (output.size() == expected.size - packetSize) { + tailReceiverBlocked.complete(Unit) + releaseTailReceiver.await() + } + yield() + } + output.toByteArray() + } + + var offset = 0 + var availableWindow = 64 * 1024L + while (offset < expected.size) { + if (availableWindow == 0L) { + val adjust = withTimeout(5_000) { server.awaitChannelWindowAdjust() } + assertEquals(100L, adjust.recipientChannel()) + availableWindow += adjust.bytesToAdd() + } + val chunkSize = minOf(packetSize.toLong(), availableWindow, (expected.size - offset).toLong()).toInt() + server.sendChannelData(localChannel, expected.copyOfRange(offset, offset + chunkSize)) + offset += chunkSize + availableWindow -= chunkSize + } + server.sendChannelClose(localChannel) + + withTimeout(5_000) { + while (session.isOpen) yield() + } + assertTrue(tailReceiverBlocked.isCompleted) + releaseTailReceiver.complete(Unit) + assertContentEquals(expected, withTimeout(5_000) { received.await() }) + assertNull(withTimeout(5_000) { session.read() }) + } + } + @Test fun `exit-status channel request completes exitInfo`() = runTest { connectedFixture { connection, server, dispatcher -> @@ -577,6 +654,32 @@ class SshConnectionFlowTest { } } + @Test + fun `direct tcpip channel preserves data received before close until consumer reads`() = runTest { + connectedFixture { connection, server, dispatcher -> + authenticate(connection, server, dispatcher) + + val open = async(dispatcher) { + connection.openDirectTcpipChannel("target", 22, "127.0.0.1", 12345) + } + val openRequest = withTimeout(5_000) { server.awaitChannelOpen() } + server.sendChannelOpenConfirmation(openRequest.senderChannel().toInt(), senderChannel = 200) + val channel = assertNotNull(withTimeout(5_000) { open.await() }) + + server.sendChannelData(channel.localChannelNumber, byteArrayOf(9, 8, 7)) + server.sendChannelClose(channel.localChannelNumber) + withTimeout(5_000) { + while (channel.isOpen) yield() + } + + assertContentEquals( + byteArrayOf(9, 8, 7), + withTimeout(5_000) { channel.incomingData.receive() }, + ) + assertNull(withTimeout(5_000) { channel.incomingData.receiveCatching().getOrNull() }) + } + } + @Test fun `incoming agent channel is rejected without provider and accepted with provider`() = runTest { connectedFixture { connection, server, dispatcher -> diff --git a/sshlib/src/test/kotlin/org/connectbot/sshlib/protocol/SshChannelStateMachineTest.kt b/sshlib/src/test/kotlin/org/connectbot/sshlib/protocol/SshChannelStateMachineTest.kt index 21bfa81..ce27fde 100644 --- a/sshlib/src/test/kotlin/org/connectbot/sshlib/protocol/SshChannelStateMachineTest.kt +++ b/sshlib/src/test/kotlin/org/connectbot/sshlib/protocol/SshChannelStateMachineTest.kt @@ -109,14 +109,18 @@ class SshChannelStateMachineTest { val remotelyClosed = SshChannelStateMachine(SshChannelState.OPEN) val closeSent = SshChannelStateMachine(SshChannelState.OPEN) var remoteCloseEffects = emptySet() + var localCloseEffects = emptySet() var closeReplyEffects = emptySet() remotelyClosed.receiveClose { remoteCloseEffects = it.effects } - closeSent.sendClose {} + closeSent.sendClose { localCloseEffects = it.effects } closeSent.receiveClose { closeReplyEffects = it.effects } assertTrue(SshChannelEffect.SEND_CLOSE in remoteCloseEffects) + assertTrue(SshChannelEffect.CLOSE_INBOUND_STREAMS in remoteCloseEffects) + assertTrue(SshChannelEffect.CLOSE_INBOUND_STREAMS in localCloseEffects) assertFalse(SshChannelEffect.SEND_CLOSE in closeReplyEffects) + assertFalse(SshChannelEffect.CLOSE_INBOUND_STREAMS in closeReplyEffects) assertEquals( SshChannelEventOrigin.PARSED_PACKET, closeSent.formalModel().transitions.first { diff --git a/sshlib/src/test/kotlin/org/connectbot/sshlib/protocol/SshStateMachineTlaGenerator.kt b/sshlib/src/test/kotlin/org/connectbot/sshlib/protocol/SshStateMachineTlaGenerator.kt index 2ae8fea..018120f 100644 --- a/sshlib/src/test/kotlin/org/connectbot/sshlib/protocol/SshStateMachineTlaGenerator.kt +++ b/sshlib/src/test/kotlin/org/connectbot/sshlib/protocol/SshStateMachineTlaGenerator.kt @@ -250,6 +250,17 @@ class SshStateMachineFormalModelTest { assertTrue("INVARIANT GlobalChannelMutationIsDisconnectCascade" in config) } + @Test + fun `TLC checks inbound stream closure effects`() { + val handwritten = Files.readString(Path.of("src/test/resources/tla/SshClientStateMachine.tla")) + val config = Files.readString(Path.of("src/test/resources/tla/SshClientStateMachine.cfg")) + + assertTrue("InboundStreamsCloseOnlyOnTermination ==" in handwritten) + assertTrue("InboundTerminationClosesStreams ==" in handwritten) + assertTrue("INVARIANT InboundStreamsCloseOnlyOnTermination" in config) + assertTrue("INVARIANT InboundTerminationClosesStreams" in config) + } + @Test fun `TLC explores strict and non-strict key exchange`() { val rendered = createFormalModel().renderTla() diff --git a/sshlib/src/test/resources/tla/SshClientStateMachine.cfg b/sshlib/src/test/resources/tla/SshClientStateMachine.cfg index 508a6bd..5f98234 100644 --- a/sshlib/src/test/resources/tla/SshClientStateMachine.cfg +++ b/sshlib/src/test/resources/tla/SshClientStateMachine.cfg @@ -9,6 +9,8 @@ VIEW ModelView INVARIANT TypeOK INVARIANT NoInvalidChannelSideEffects +INVARIANT InboundStreamsCloseOnlyOnTermination +INVARIANT InboundTerminationClosesStreams INVARIANT ChannelIsolation INVARIANT GlobalOperationsPreserveChannels INVARIANT GlobalChannelMutationIsDisconnectCascade diff --git a/sshlib/src/test/resources/tla/SshClientStateMachine.tla b/sshlib/src/test/resources/tla/SshClientStateMachine.tla index bc413ad..dc9a804 100644 --- a/sshlib/src/test/resources/tla/SshClientStateMachine.tla +++ b/sshlib/src/test/resources/tla/SshClientStateMachine.tla @@ -85,6 +85,23 @@ NoInvalidChannelSideEffects == ) /\ channelOrigin = ChannelOriginFor(channelEvent) +InboundStreamsCloseOnlyOnTermination == + "CLOSE_INBOUND_STREAMS" \in channelEffects => + channelEvent \in {"ReceiveEof", "ReceiveClose", "SendClose"} + +InboundTerminationClosesStreams == + activeChannel \in ChannelIDs /\ + ChannelOperationAllowed( + authenticationEstablished, + state, + previousChannels[activeChannel], + channelEvent + ) /\ + (\/ channelEvent \in {"ReceiveEof", "SendClose"} + \/ /\ channelEvent = "ReceiveClose" + /\ previousChannels[activeChannel] # "CLOSE_SENT") => + "CLOSE_INBOUND_STREAMS" \in channelEffects + ChannelIsolation == activeChannel \in ChannelIDs => \A other \in ChannelIDs \ {activeChannel} : diff --git a/sshlib/src/test/resources/tla/SshClientStateMachineGenerated.tla b/sshlib/src/test/resources/tla/SshClientStateMachineGenerated.tla index 2336016..9249440 100644 --- a/sshlib/src/test/resources/tla/SshClientStateMachineGenerated.tla +++ b/sshlib/src/test/resources/tla/SshClientStateMachineGenerated.tla @@ -1,6 +1,6 @@ ---- MODULE SshClientStateMachineGenerated ---- \* Generated from SshClientStateMachine. Do not edit. -\* Model SHA-256: 04f78793763d07d60314395341b04de0a29ebc5a024845d80d27599ad75f0214 +\* Model SHA-256: f93b71171b62816926e550b59ad72fa55b567872e0ad9aea52e90d72bbc78501 \* Lifecycle states: 11; transitions: 43. \* TLC distinct states count full variable valuations, not lifecycle nodes. EXTENDS Naturals @@ -114,38 +114,38 @@ ChannelTransitionTarget(channelState, operation) == CHOOSE target \in ChannelStates : <> \in ChannelTransitions ChannelEffectsFor(channelState, operation) == - CASE /\ channelState = "BOTH_EOF" /\ operation = "ReceiveClose" -> {"SEND_CLOSE", "CLOSE_CHANNEL"} + CASE /\ channelState = "BOTH_EOF" /\ operation = "ReceiveClose" -> {"SEND_CLOSE", "CLOSE_INBOUND_STREAMS", "CLOSE_CHANNEL"} [] /\ channelState = "BOTH_EOF" /\ operation = "ReceiveRequest" -> {"DELIVER_REQUEST"} [] /\ channelState = "BOTH_EOF" /\ operation = "ReceiveWindowAdjust" -> {"ADJUST_WINDOW"} - [] /\ channelState = "BOTH_EOF" /\ operation = "SendClose" -> {"SEND_CLOSE"} + [] /\ channelState = "BOTH_EOF" /\ operation = "SendClose" -> {"SEND_CLOSE", "CLOSE_INBOUND_STREAMS"} [] /\ channelState = "BOTH_EOF" /\ operation = "SendRequest" -> {"SEND_REQUEST"} [] /\ channelState = "CLOSE_SENT" /\ operation = "ReceiveClose" -> {"CLOSE_CHANNEL"} [] /\ channelState = "CLOSE_SENT" /\ operation = "ReceiveData" -> {} [] /\ channelState = "CLOSE_SENT" /\ operation = "ReceiveEof" -> {} [] /\ channelState = "CLOSE_SENT" /\ operation = "ReceiveRequest" -> {} [] /\ channelState = "CLOSE_SENT" /\ operation = "ReceiveWindowAdjust" -> {} - [] /\ channelState = "LOCAL_EOF" /\ operation = "ReceiveClose" -> {"SEND_CLOSE", "CLOSE_CHANNEL"} + [] /\ channelState = "LOCAL_EOF" /\ operation = "ReceiveClose" -> {"SEND_CLOSE", "CLOSE_INBOUND_STREAMS", "CLOSE_CHANNEL"} [] /\ channelState = "LOCAL_EOF" /\ operation = "ReceiveData" -> {"DELIVER_DATA"} [] /\ channelState = "LOCAL_EOF" /\ operation = "ReceiveEof" -> {"CLOSE_INBOUND_STREAMS"} [] /\ channelState = "LOCAL_EOF" /\ operation = "ReceiveRequest" -> {"DELIVER_REQUEST"} [] /\ channelState = "LOCAL_EOF" /\ operation = "ReceiveWindowAdjust" -> {"ADJUST_WINDOW"} - [] /\ channelState = "LOCAL_EOF" /\ operation = "SendClose" -> {"SEND_CLOSE"} + [] /\ channelState = "LOCAL_EOF" /\ operation = "SendClose" -> {"SEND_CLOSE", "CLOSE_INBOUND_STREAMS"} [] /\ channelState = "LOCAL_EOF" /\ operation = "SendRequest" -> {"SEND_REQUEST"} - [] /\ channelState = "OPEN" /\ operation = "ReceiveClose" -> {"SEND_CLOSE", "CLOSE_CHANNEL"} + [] /\ channelState = "OPEN" /\ operation = "ReceiveClose" -> {"SEND_CLOSE", "CLOSE_INBOUND_STREAMS", "CLOSE_CHANNEL"} [] /\ channelState = "OPEN" /\ operation = "ReceiveData" -> {"DELIVER_DATA"} [] /\ channelState = "OPEN" /\ operation = "ReceiveEof" -> {"CLOSE_INBOUND_STREAMS"} [] /\ channelState = "OPEN" /\ operation = "ReceiveRequest" -> {"DELIVER_REQUEST"} [] /\ channelState = "OPEN" /\ operation = "ReceiveWindowAdjust" -> {"ADJUST_WINDOW"} - [] /\ channelState = "OPEN" /\ operation = "SendClose" -> {"SEND_CLOSE"} + [] /\ channelState = "OPEN" /\ operation = "SendClose" -> {"SEND_CLOSE", "CLOSE_INBOUND_STREAMS"} [] /\ channelState = "OPEN" /\ operation = "SendData" -> {"SEND_DATA"} [] /\ channelState = "OPEN" /\ operation = "SendEof" -> {"SEND_EOF"} [] /\ channelState = "OPEN" /\ operation = "SendRequest" -> {"SEND_REQUEST"} [] /\ channelState = "OPENING" /\ operation = "OpenConfirmed" -> {"COMPLETE_OPEN"} [] /\ channelState = "OPENING" /\ operation = "OpenFailed" -> {"FAIL_OPEN"} - [] /\ channelState = "REMOTE_EOF" /\ operation = "ReceiveClose" -> {"SEND_CLOSE", "CLOSE_CHANNEL"} + [] /\ channelState = "REMOTE_EOF" /\ operation = "ReceiveClose" -> {"SEND_CLOSE", "CLOSE_INBOUND_STREAMS", "CLOSE_CHANNEL"} [] /\ channelState = "REMOTE_EOF" /\ operation = "ReceiveRequest" -> {"DELIVER_REQUEST"} [] /\ channelState = "REMOTE_EOF" /\ operation = "ReceiveWindowAdjust" -> {"ADJUST_WINDOW"} - [] /\ channelState = "REMOTE_EOF" /\ operation = "SendClose" -> {"SEND_CLOSE"} + [] /\ channelState = "REMOTE_EOF" /\ operation = "SendClose" -> {"SEND_CLOSE", "CLOSE_INBOUND_STREAMS"} [] /\ channelState = "REMOTE_EOF" /\ operation = "SendData" -> {"SEND_DATA"} [] /\ channelState = "REMOTE_EOF" /\ operation = "SendEof" -> {"SEND_EOF"} [] /\ channelState = "REMOTE_EOF" /\ operation = "SendRequest" -> {"SEND_REQUEST"}