Skip to content

chore: add more test coverage #142

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

Merged
kruton merged 1 commit into from
May 9, 2026
Merged

chore: add more test coverage #142

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
236 changes: 236 additions & 0 deletions sshlib/src/test/kotlin/org/connectbot/sshlib/client/ExtInfoNegotiationTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,10 @@ import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.StandardTestDispatcher
import kotlinx.coroutines.test.runCurrent
import kotlinx.coroutines.test.runTest
import kotlinx.coroutines.withTimeout
import kotlinx.coroutines.withTimeoutOrNull
Expand All @@ -15,6 +17,7 @@ import org.connectbot.sshlib.HostKeyVerifier
import org.connectbot.sshlib.PublicKey
import org.connectbot.sshlib.protocol.SshMsgExtInfo
import org.connectbot.sshlib.transport.PipedTransport
import org.junit.jupiter.api.Assertions.assertFalse
import org.junit.jupiter.api.Assertions.assertNotNull
import org.junit.jupiter.api.Assertions.assertNull
import org.junit.jupiter.api.Test
Expand Down Expand Up @@ -166,6 +169,239 @@ class ExtInfoNegotiationTest {
}
}

// --- stripExtInfoC / appendExtInfoC boundary tests ---

@Test
fun `client strips ext-info-c from kexAlgorithms before re-advertising it`() = runTest {
val dispatcher = StandardTestDispatcher(testScheduler)
val (clientTransport, serverTransport) = PipedTransport.create()
val server = FakeSshServer(serverTransport, backgroundScope, dispatcher)
server.advertiseExtInfo = true
server.start()

val connection = SshConnection(
transport = clientTransport,
hostKeyVerifier = acceptAllVerifier,
kexAlgorithms = "curve25519-sha256,ext-info-c",
coroutineDispatcher = dispatcher,
)

try {
val result = connectInBackground(connection, backgroundScope, dispatcher)
assertIs<ConnectResult.Success>(result)
val kexInit = withTimeout(1000) { server.awaitClientKexInit() }
val kexAlgs = kexInit.kexAlgorithms().entries().data()
// ext-info-c must appear exactly once (appended), not duplicated
assertEquals(1, kexAlgs.count { it == "ext-info-c" }, "ext-info-c should appear exactly once, got: $kexAlgs")
} finally {
connection.close()
}
}

@Test
fun `client appends ext-info-c when not present in custom kex list`() = runTest {
val dispatcher = StandardTestDispatcher(testScheduler)
val (clientTransport, serverTransport) = PipedTransport.create()
val server = FakeSshServer(serverTransport, backgroundScope, dispatcher)
server.advertiseExtInfo = true
server.start()

val connection = SshConnection(
transport = clientTransport,
hostKeyVerifier = acceptAllVerifier,
kexAlgorithms = "curve25519-sha256",
coroutineDispatcher = dispatcher,
)

try {
val result = connectInBackground(connection, backgroundScope, dispatcher)
assertIs<ConnectResult.Success>(result)
val kexInit = withTimeout(1000) { server.awaitClientKexInit() }
val kexAlgs = kexInit.kexAlgorithms().entries().data()
assertTrue("ext-info-c" in kexAlgs, "ext-info-c should be appended to kex list: $kexAlgs")
} finally {
connection.close()
}
}

// --- sendClientExtInfo guard: not sent twice ---

@Test
fun `client sends EXT_INFO at most once even after rekey`() = runTest {
val dispatcher = StandardTestDispatcher(testScheduler)
val (clientTransport, serverTransport) = PipedTransport.create()
val server = FakeSshServer(serverTransport, backgroundScope, dispatcher)
server.advertiseExtInfo = true
server.start()

val connection = SshConnection(
transport = clientTransport,
hostKeyVerifier = acceptAllVerifier,
coroutineDispatcher = dispatcher,
)

try {
val result = connectInBackground(connection, backgroundScope, dispatcher)
assertIs<ConnectResult.Success>(result)

// Consume the initial EXT_INFO
val first = withTimeout(1000) { server.awaitExtInfo() }
assertNotNull(first)

// Trigger a server-initiated rekey; the client must NOT send EXT_INFO again
server.initiateRekey()
server.sendIgnore()
withTimeout(1000) { server.rekeyCount.first { it >= 1 } }

val second = withTimeoutOrNull(300) { server.awaitExtInfo() }
assertNull(second, "Client must not send EXT_INFO again after rekey")
} finally {
connection.close()
}
}

// --- processServerExtInfo: count guard (>= 2 rejects third message) ---

@Test
fun `client ignores third EXT_INFO from server`() = runTest {
val dispatcher = StandardTestDispatcher(testScheduler)
val (clientTransport, serverTransport) = PipedTransport.create()
val server = FakeSshServer(serverTransport, backgroundScope, dispatcher)
server.advertiseExtInfo = true
server.advertisePing = true
server.start()

val connection = SshConnection(
transport = clientTransport,
hostKeyVerifier = acceptAllVerifier,
coroutineDispatcher = dispatcher,
)

var authJob: kotlinx.coroutines.Job? = null
try {
val result = connectInBackground(connection, backgroundScope, dispatcher)
assertIs<ConnectResult.Success>(result)
assertTrue(connection.serverSupportsPing, "Initial EXT_INFO should set ping support")

// Second EXT_INFO (during auth) — allowed, but must not clear ping
authJob = backgroundScope.launch(dispatcher) {
connection.authenticatePassword("user", "pass")
}
withTimeout(1000) { server.awaitUserauthRequest() }
server.sendCustomExtInfo(mapOf("server-sig-algs" to "rsa-sha2-256".toByteArray(Charsets.US_ASCII)))
awaitServerSigAlgs(connection, setOf("rsa-sha2-256"))

// Third EXT_INFO — must be ignored; ping support and sig-algs must remain unchanged
server.sendCustomExtInfo(mapOf("server-sig-algs" to "ssh-ed25519".toByteArray(Charsets.US_ASCII)))
// Drain only currently scheduled packet handling. advanceUntilIdle would also
// advance virtual time far enough to fire the connection's rekey timer.
runCurrent()
assertEquals(setOf("rsa-sha2-256"), serverSigAlgs(connection), "Third EXT_INFO must be ignored")
assertTrue(connection.serverSupportsPing, "Ping support must survive ignored third EXT_INFO")
} finally {
authJob?.cancel()
connection.close()
}
}

// --- processServerExtInfo: initialExtInfo flag (== 0 check) ---

@Test
fun `second EXT_INFO does not clear initial-only fields like ping and hostbound`() = runTest {
val dispatcher = StandardTestDispatcher(testScheduler)
val (clientTransport, serverTransport) = PipedTransport.create()
val server = FakeSshServer(serverTransport, backgroundScope, dispatcher)
server.advertiseExtInfo = true
server.advertisePing = true
server.start()

val connection = SshConnection(
transport = clientTransport,
hostKeyVerifier = acceptAllVerifier,
coroutineDispatcher = dispatcher,
)

var authJob: kotlinx.coroutines.Job? = null
try {
val result = connectInBackground(connection, backgroundScope, dispatcher)
assertIs<ConnectResult.Success>(result)
assertTrue(connection.serverSupportsPing, "Ping should be set by initial EXT_INFO")

authJob = backgroundScope.launch(dispatcher) {
connection.authenticatePassword("user", "pass")
}
withTimeout(1000) { server.awaitUserauthRequest() }

// Second EXT_INFO without ping — ping must remain true (initial-only)
server.sendCustomExtInfo(mapOf("server-sig-algs" to "rsa-sha2-256".toByteArray(Charsets.US_ASCII)))
awaitServerSigAlgs(connection, setOf("rsa-sha2-256"))
assertTrue(connection.serverSupportsPing, "Ping support must not be cleared by second EXT_INFO")
} finally {
authJob?.cancel()
connection.close()
}
}

@Test
fun `initial EXT_INFO with no ping extension leaves serverSupportsPing false`() = runTest {
val dispatcher = StandardTestDispatcher(testScheduler)
val (clientTransport, serverTransport) = PipedTransport.create()
val server = FakeSshServer(serverTransport, backgroundScope, dispatcher)
server.advertiseExtInfo = true
server.advertisePing = false // no ping in initial EXT_INFO
server.start()

val connection = SshConnection(
transport = clientTransport,
hostKeyVerifier = acceptAllVerifier,
coroutineDispatcher = dispatcher,
)

try {
val result = connectInBackground(connection, backgroundScope, dispatcher)
assertIs<ConnectResult.Success>(result)
assertFalse(connection.serverSupportsPing, "Ping should be false when not in initial EXT_INFO")
} finally {
connection.close()
}
}

// --- processServerExtInfo: ignored when server did not advertise ext-info-s ---

@Test
fun `EXT_INFO from server is ignored when server did not advertise ext-info-s`() = runTest {
val dispatcher = StandardTestDispatcher(testScheduler)
val (clientTransport, serverTransport) = PipedTransport.create()
val server = FakeSshServer(serverTransport, backgroundScope, dispatcher)
server.advertiseExtInfo = false
server.start()

val connection = SshConnection(
transport = clientTransport,
hostKeyVerifier = acceptAllVerifier,
coroutineDispatcher = dispatcher,
)

var authJob: kotlinx.coroutines.Job? = null
try {
val result = connectInBackground(connection, backgroundScope, dispatcher)
assertIs<ConnectResult.Success>(result)

authJob = backgroundScope.launch(dispatcher) {
connection.authenticatePassword("user", "pass")
}
withTimeout(1000) { server.awaitUserauthRequest() }

// Send a rogue EXT_INFO even though server never advertised ext-info-s
server.sendCustomExtInfo(mapOf("server-sig-algs" to "ssh-ed25519".toByteArray(Charsets.US_ASCII)))
runCurrent()
assertNull(serverSigAlgs(connection), "server-sig-algs must remain null when EXT_INFO is ignored")
} finally {
Comment on lines +395 to +399
Copy link
Copy Markdown
Member Author

@kruton kruton May 9, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@copilot apply changes based on this feedback

Copy link
Copy Markdown

Copilot AI May 9, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done in 9e307e9 — replaced the two yield() calls with advanceUntilIdle() to drain the test scheduler before asserting serverSigAlgs remains null.

authJob?.cancel()
connection.close()
}
}

@Test
fun `server may update server-sig-algs during user authentication`() = runTest {
val dispatcher = StandardTestDispatcher(testScheduler)
Expand Down
6 changes: 6 additions & 0 deletions sshlib/src/test/kotlin/org/connectbot/sshlib/client/FakeSshServer.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,7 @@ class FakeSshServer(
private val receivedPongs = Channel<ByteArray>(Channel.UNLIMITED)
private val receivedExtInfo = Channel<SshMsgExtInfo>(Channel.UNLIMITED)
private val receivedUserauthRequests = Channel<SshMsgUserauthRequest>(Channel.UNLIMITED)
private val receivedClientKexInits = Channel<SshMsgKexinit>(Channel.UNLIMITED)

fun start() {
scope.launch(coroutineContext) { serve() }
Expand Down Expand Up @@ -231,6 +232,9 @@ class FakeSshServer(
private suspend fun doFullKex(io: PacketIO) {
val serverKexInitBytes = sendKexInit(io)
val clientKexInitRaw = readPacketRaw(io)
val clientKexInit = SshMsgKexinit(ByteBufferKaitaiStream(clientKexInitRaw.copyOfRange(1, clientKexInitRaw.size)))
clientKexInit._read()
receivedClientKexInits.trySend(clientKexInit)
val ecdhInitRaw = readPacketRaw(io)
val clientPublic = parseEcdhInit(ecdhInitRaw)
sendEcdhReply(io, clientKexInitRaw, serverKexInitBytes, clientPublic)
Expand Down Expand Up @@ -546,4 +550,6 @@ class FakeSshServer(
suspend fun awaitExtInfo(): SshMsgExtInfo = receivedExtInfo.receive()

suspend fun awaitUserauthRequest(): SshMsgUserauthRequest = receivedUserauthRequests.receive()

suspend fun awaitClientKexInit(): SshMsgKexinit = receivedClientKexInits.receive()
}
Loading
Loading