mirror of
https://github.com/Monkopedia/rouse-context.git
synced 2026-10-21 22:30:32 +00:00
Twelve broad `catch (Exception)` clauses in :core:tunnel wrapped a suspend
call and converted cancellation into a *returned* value -- a Warning, a
StorageFailed with a rollback, a NetworkError, `false` -- instead of letting
it propagate. `.claude/rules/coroutines.md` requires propagation; the
consequence is work performed after cancellation.
Fixed, lowest frame first:
MuxDemux.sendPingAwaitPong `false` -> propagate
TunnelClientImpl.runKeepaliveLoop `false` -> propagate
TunnelClientImpl.healthCheck `false` -> propagate
CtLogMonitor.check x2 Warning -> propagate
SelfCertVerifier x3 Warning -> propagate
OnboardingFlow (clear()) StorageFailed + rollback -> propagate
CertProvisioningFlow (clearCerts()) StorageFailed + rollback -> propagate
RelayApiClient.executeRequest NetworkError -> propagate
KtorWebSocketFactory.connect onFailure/onClosing -> silent teardown
The keepalive stack is the reason for the ordering. #650 added a cancellation
rethrow at WakeReconnectDecider:56 that could not fire, because both frames
below it swallowed cancellation into `false` first. A guard above an
already-swallowing callee is dead code that looks identical in a diff to one
that works, so the swallow is fixed at the lowest frame that performs it and
each guard above is covered by a test that reddens when it alone is reverted.
MuxDemux and KtorWebSocketFactory discriminate on the *job* rather than the
exception type. Not every CancellationException at those frames is the
caller's: MuxDemux.closeAllQuietly deliberately cancels each pending ping
waiter when the transport dies (#230), and that must keep reading as a failed
health check. A blanket rethrow there hangs
MuxDemuxTest.awaitPongReturnsFalseWhenTransportClosedQuietly.
KtorWebSocketFactory still fails the bind on the teardown path so an awaiter
parked in sendBinary is released (#420 finding #4); only the listener
notification is dropped.
Four sites listed in the issue are documented in place rather than guarded,
because cancellation cannot reach them:
CertProvisioningFlow:76, CertRenewalFlow:105/159/200 -- both guarded calls
(DeviceKeyManager.getOrCreateKeyPair, CsrGenerator.generate) are declared
non-suspend on their interfaces, so no implementation can suspend.
CtLogMonitor:68 -- json.decodeFromString is not a suspend function.
RelayApiClient:281 -- Ktor 3's SaveBody plugin drains the body inside
block(), so bodyAsText() reads an in-memory buffer; a body that fails
mid-read surfaces at the outer catch instead. Pinned by a test.
Two sites are deliberately untouched: SecurityCheckSources.kt:25, whose
Warning reaches a user-visible notification and so needs a ux-changes.md
decision, and TunnelSessionManager, which is #638's subject.
27 tests added, two directions per fixed site. All 12 cancellation-direction
tests are red against 54fdcd46; the 15 ordinary-direction tests pass against
it unchanged.
Closes #646
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YYSPuh9bZEJkrzvnEpY1QK
This commit is contained in:
monkopedia-coder
GitHub
parent
9b50f469d3
commit
8479cf563b
15 changed files with 1194 additions and 8 deletions
|
|
@ -1,5 +1,6 @@
|
|||
package com.rousecontext.tunnel
|
||||
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
|
||||
|
|
@ -71,6 +72,11 @@ class CertProvisioningFlow(
|
|||
// Obtain (or generate) the hardware-backed device keypair and build the CSR.
|
||||
val fqdn = "*.$subdomain.$baseDomain"
|
||||
val csrResult = try {
|
||||
// No cancellation rethrow here on purpose (issue #646):
|
||||
// `DeviceKeyManager.getOrCreateKeyPair` and `CsrGenerator.generate`
|
||||
// are both declared non-suspend on their interfaces, so no
|
||||
// implementation can suspend and no coroutine cancellation can
|
||||
// reach this clause. A guard here would be dead code.
|
||||
val keyPair = deviceKeyManager.getOrCreateKeyPair()
|
||||
csrGenerator.generate(fqdn, keyPair)
|
||||
} catch (e: Exception) {
|
||||
|
|
@ -94,6 +100,11 @@ class CertProvisioningFlow(
|
|||
certificateStore.storeClientCertificate(certResponse.data.clientCert)
|
||||
certificateStore.storeRelayCaCert(certResponse.data.relayCaCert)
|
||||
CertProvisioningResult.Success
|
||||
} catch (e: CancellationException) {
|
||||
// MUST stay above the broad catch (issue #646): a cancelled
|
||||
// write is not a failed write, and the rollback below is
|
||||
// destructive work performed after cancellation.
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
// Narrow rollback: only clear cert-related state. The onboarding
|
||||
// state (subdomain, integration secrets) is set earlier by
|
||||
|
|
|
|||
|
|
@ -100,6 +100,8 @@ class CertRenewalFlow(
|
|||
if (certInspector.inspect(currentCert).isExpired) return RenewalResult.CertExpired
|
||||
|
||||
val csrResult = try {
|
||||
// No cancellation rethrow (issue #646): both calls are declared
|
||||
// non-suspend on their interfaces, so no cancellation can reach here.
|
||||
val keyPair = deviceKeyManager.getOrCreateKeyPair()
|
||||
csrGenerator.generate("*.$subdomain.$baseDomain", keyPair)
|
||||
} catch (e: Exception) {
|
||||
|
|
@ -154,6 +156,8 @@ class CertRenewalFlow(
|
|||
val currentCert = certificateStore.getCertificate()
|
||||
|
||||
val csrResult = try {
|
||||
// No cancellation rethrow (issue #646): both calls are declared
|
||||
// non-suspend on their interfaces, so no cancellation can reach here.
|
||||
val keyPair = deviceKeyManager.getOrCreateKeyPair()
|
||||
csrGenerator.generate("*.$subdomain.$baseDomain", keyPair)
|
||||
} catch (e: Exception) {
|
||||
|
|
@ -195,6 +199,8 @@ class CertRenewalFlow(
|
|||
val currentCert = certificateStore.getCertificate()
|
||||
|
||||
val csrResult = try {
|
||||
// No cancellation rethrow (issue #646): both calls are declared
|
||||
// non-suspend on their interfaces, so no cancellation can reach here.
|
||||
val keyPair = deviceKeyManager.getOrCreateKeyPair()
|
||||
csrGenerator.generate("*.$subdomain.$baseDomain", keyPair)
|
||||
} catch (e: Exception) {
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
package com.rousecontext.tunnel
|
||||
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.serialization.SerialName
|
||||
import kotlinx.serialization.Serializable
|
||||
import kotlinx.serialization.json.Json
|
||||
|
|
@ -22,6 +23,12 @@ interface CtLogFetcher {
|
|||
*
|
||||
* Network errors or malformed responses produce [SecurityCheckResult.Warning]
|
||||
* since the check cannot be completed.
|
||||
*
|
||||
* A **cancelled** check is not a degraded one and must not produce a Warning:
|
||||
* the check was never run, so there is nothing to report. Both suspend calls
|
||||
* below rethrow cancellation ahead of their broad catch (issue #646). The layer
|
||||
* directly beneath already does this -- see [CompositeCtLogFetcher] -- and
|
||||
* swallowing it one frame up would defeat that entirely.
|
||||
*/
|
||||
class CtLogMonitor(
|
||||
private val certificateStore: CertificateStore,
|
||||
|
|
@ -39,6 +46,11 @@ class CtLogMonitor(
|
|||
suspend fun check(): SecurityCheckResult {
|
||||
val subdomain = try {
|
||||
certificateStore.getSubdomain()
|
||||
} catch (e: CancellationException) {
|
||||
// MUST stay above the broad catch: CancellationException is an
|
||||
// Exception on the JVM, so catching Exception alone would refile a
|
||||
// scope teardown as a user-facing security warning.
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
return SecurityCheckResult.Warning(
|
||||
"Could not retrieve subdomain: ${e.message}"
|
||||
|
|
@ -56,6 +68,10 @@ class CtLogMonitor(
|
|||
val responseBody: String
|
||||
try {
|
||||
responseBody = ctLogFetcher.fetch(domain)
|
||||
} catch (e: CancellationException) {
|
||||
// As above. CompositeCtLogFetcher rethrows cancellation from both
|
||||
// of its own catches; this is the frame that used to swallow it.
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
return SecurityCheckResult.Warning(
|
||||
"Could not reach CT log service: ${e.message}"
|
||||
|
|
@ -66,6 +82,9 @@ class CtLogMonitor(
|
|||
try {
|
||||
entries = json.decodeFromString<List<CtLogEntry>>(responseBody)
|
||||
} catch (e: Exception) {
|
||||
// No cancellation rethrow here on purpose: `decodeFromString` is not
|
||||
// a suspend function, so no coroutine cancellation can reach this
|
||||
// clause. A guard here would be dead code (issue #646).
|
||||
return SecurityCheckResult.Warning(
|
||||
"Malformed CT log response: ${e.message}"
|
||||
)
|
||||
|
|
|
|||
|
|
@ -7,8 +7,11 @@ import io.ktor.websocket.Frame
|
|||
import io.ktor.websocket.WebSocketSession
|
||||
import io.ktor.websocket.close
|
||||
import io.ktor.websocket.readBytes
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CompletableDeferred
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.currentCoroutineContext
|
||||
import kotlinx.coroutines.isActive
|
||||
import kotlinx.coroutines.launch
|
||||
|
||||
/**
|
||||
|
|
@ -42,21 +45,57 @@ class KtorWebSocketFactory(
|
|||
}
|
||||
// Normal close
|
||||
listener.onClosing(1000, "WebSocket closed")
|
||||
} catch (e: Exception) {
|
||||
if (handle.isBound()) {
|
||||
listener.onClosing(1006, e.message ?: "Connection lost")
|
||||
} catch (e: CancellationException) {
|
||||
// MUST stay above the broad catch (issue #646): a torn-down
|
||||
// scope is not a transport failure, so the listener is told
|
||||
// nothing -- the mirror of what #644 fixed one layer up in
|
||||
// TunnelClientImpl.connect.
|
||||
//
|
||||
// Gated on the job, not on the exception type. A cancellation
|
||||
// that arrives while this coroutine is still active did not come
|
||||
// from our teardown -- it came from the session -- and that IS a
|
||||
// transport failure the listener has to hear about, or the tunnel
|
||||
// sits CONNECTED over a dead socket. Only our own cancellation
|
||||
// takes the silent path.
|
||||
if (currentCoroutineContext().isActive) {
|
||||
reportFailure(handle, listener, e)
|
||||
} else {
|
||||
// Awaiters of sendBinary/sendText would otherwise hang
|
||||
// forever waiting for sessionDeferred. Propagate the failure
|
||||
// so they observe the connect error.
|
||||
handle.failBind(e)
|
||||
listener.onFailure(e)
|
||||
// The bind is still failed: that is the whole point of
|
||||
// failBind (#420 finding #4), since an awaiter parked in
|
||||
// sendBinary has no other way to learn the session will
|
||||
// never arrive. Completing it with the cancellation releases
|
||||
// the awaiter and tells it the truth.
|
||||
if (!handle.isBound()) handle.failBind(e)
|
||||
throw e
|
||||
}
|
||||
} catch (e: Exception) {
|
||||
reportFailure(handle, listener, e)
|
||||
}
|
||||
}
|
||||
|
||||
return handle
|
||||
}
|
||||
|
||||
/**
|
||||
* Tell [listener] the session is gone. Split out so the cancellation clause
|
||||
* can reuse it for a session-side cancellation without duplicating the
|
||||
* bound / not-yet-bound distinction.
|
||||
*/
|
||||
private fun reportFailure(
|
||||
handle: KtorWebSocketHandle,
|
||||
listener: WebSocketListener,
|
||||
e: Throwable
|
||||
) {
|
||||
if (handle.isBound()) {
|
||||
listener.onClosing(1006, e.message ?: "Connection lost")
|
||||
} else {
|
||||
// Awaiters of sendBinary/sendText would otherwise hang forever
|
||||
// waiting for sessionDeferred. Propagate the failure so they
|
||||
// observe the connect error.
|
||||
handle.failBind(e)
|
||||
listener.onFailure(e)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private class KtorWebSocketHandle : WebSocketHandle {
|
||||
|
|
|
|||
|
|
@ -1,7 +1,10 @@
|
|||
package com.rousecontext.tunnel
|
||||
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CompletableDeferred
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.currentCoroutineContext
|
||||
import kotlinx.coroutines.ensureActive
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.flow.StateFlow
|
||||
|
|
@ -155,6 +158,19 @@ class MuxDemux(private val log: (LogLevel, String) -> Unit = { _, _ -> }) {
|
|||
*
|
||||
* The optional [nonce] exists for tests; production code should leave it
|
||||
* unset so a fresh random nonce is generated.
|
||||
*
|
||||
* The **caller's** cancellation propagates rather than reading as a missed
|
||||
* Pong (issue #646). This is the lowest frame in the keepalive stack, so it
|
||||
* is the one that has to get this right: the guards above it in
|
||||
* `TunnelClientImpl.runKeepaliveLoop` / `healthCheck`, and the one #650
|
||||
* added in `WakeReconnectDecider`, are all unreachable while this frame
|
||||
* converts cancellation into `false`.
|
||||
*
|
||||
* Not every `CancellationException` here is the caller's, which is why the
|
||||
* catch below tests the job rather than the exception type. [closeAllQuietly]
|
||||
* *cancels* each pending ping waiter when the transport dies (issue #230),
|
||||
* and that must still surface as a failed health check -- a dead relay is
|
||||
* not a cancelled caller. `ensureActive` is what separates the two.
|
||||
*/
|
||||
suspend fun sendPingAwaitPong(timeoutMillis: Long, nonce: ULong = randomNonce()): Boolean {
|
||||
val send = onOutgoingFrame ?: return false
|
||||
|
|
@ -163,6 +179,16 @@ class MuxDemux(private val log: (LogLevel, String) -> Unit = { _, _ -> }) {
|
|||
return try {
|
||||
send(MuxFrame.Ping(nonce))
|
||||
withTimeoutOrNull(timeoutMillis) { waiter.await() } != null
|
||||
} catch (_: CancellationException) {
|
||||
// MUST stay above the broad catch: CancellationException is an
|
||||
// Exception on the JVM, so catching Exception alone turned a scope
|
||||
// teardown into a missed Pong (#646). But `closeAllQuietly` cancels
|
||||
// the waiter on a dead transport (#230), which must still read as a
|
||||
// missed Pong -- so propagate only when it is *this* coroutine that
|
||||
// was cancelled. `withTimeoutOrNull` handles its own timeout by
|
||||
// returning null, so it never lands here.
|
||||
currentCoroutineContext().ensureActive()
|
||||
false
|
||||
} catch (_: Exception) {
|
||||
false
|
||||
} finally {
|
||||
|
|
|
|||
|
|
@ -1,5 +1,7 @@
|
|||
package com.rousecontext.tunnel
|
||||
|
||||
import kotlinx.coroutines.CancellationException
|
||||
|
||||
/**
|
||||
* Orchestrates first-run device onboarding: registers the device with the relay
|
||||
* to get a subdomain assignment, then provisions the ACME server + relay CA
|
||||
|
|
@ -114,6 +116,13 @@ class OnboardingFlow(
|
|||
certificateStore.storeIntegrationSecrets(registerData.secrets)
|
||||
}
|
||||
RegisterOutcome.Success(registerData.subdomain)
|
||||
} catch (e: CancellationException) {
|
||||
// MUST stay above the broad catch (issue #646). A cancelled write is
|
||||
// not a failed write, and the rollback below is real destructive
|
||||
// work performed after cancellation: `clear()` drops the subdomain
|
||||
// the relay has already assigned, leaving the device registered
|
||||
// upstream and blank locally.
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
certificateStore.clear()
|
||||
RegisterOutcome.Failure(OnboardingResult.StorageFailed(e))
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ import io.ktor.client.statement.bodyAsText
|
|||
import io.ktor.http.ContentType
|
||||
import io.ktor.http.contentType
|
||||
import io.ktor.serialization.kotlinx.json.json
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.serialization.SerialName
|
||||
import kotlinx.serialization.Serializable
|
||||
import kotlinx.serialization.json.Json
|
||||
|
|
@ -279,6 +280,13 @@ class RelayApiClient(
|
|||
val body = try {
|
||||
response.bodyAsText()
|
||||
} catch (_: Exception) {
|
||||
// No cancellation rethrow (issue #646): Ktor's SaveBody
|
||||
// plugin drains the response inside `block()` above, so by
|
||||
// the time this runs it reads an in-memory buffer with no
|
||||
// cancellable suspension point, and a body that genuinely
|
||||
// fails mid-read surfaces from `block()` instead. Measured
|
||||
// by RelayApiClientCancellationTest; a guard here would be
|
||||
// dead code.
|
||||
""
|
||||
}
|
||||
RelayApiResult.Error(
|
||||
|
|
@ -287,6 +295,13 @@ class RelayApiClient(
|
|||
)
|
||||
}
|
||||
}
|
||||
} catch (e: CancellationException) {
|
||||
// MUST stay above the broad catch (issue #646): CancellationException
|
||||
// is an Exception on the JVM, and a returned NetworkError reaches
|
||||
// OnboardingFlow, which folds it into a terminal OnboardingResult with
|
||||
// no suspension point in between -- so a cancelled onboarding would
|
||||
// hand the UI a "registration failed" outcome instead of unwinding.
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
RelayApiResult.NetworkError(cause = e)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
package com.rousecontext.tunnel
|
||||
|
||||
import java.security.MessageDigest
|
||||
import kotlinx.coroutines.CancellationException
|
||||
|
||||
/**
|
||||
* Verifies that a presented certificate matches the fingerprints
|
||||
|
|
@ -30,6 +31,15 @@ import java.security.MessageDigest
|
|||
* and we MUST Alert rather than silently re-trusting whatever cert is
|
||||
* presented. The backfill is therefore a strictly one-time-per-install
|
||||
* migration, not a permanent self-heal.
|
||||
*
|
||||
* Cancellation (issue #646): every store call below is `suspend`, and each of
|
||||
* the three broad catches rethrows cancellation ahead of itself. A cancelled
|
||||
* verify is a check that never ran, not a degraded one, so it must not surface
|
||||
* a [SecurityCheckResult.Warning]. The backfill catch matters most: the
|
||||
* fingerprint write and the marker write are deliberately ordered so a failed
|
||||
* fingerprint write cannot claim bootstrap completed, and swallowing a
|
||||
* cancellation between them would report exactly the half-written state that
|
||||
* ordering exists to prevent.
|
||||
*/
|
||||
class SelfCertVerifier(private val certificateStore: CertificateStore) {
|
||||
|
||||
|
|
@ -45,6 +55,10 @@ class SelfCertVerifier(private val certificateStore: CertificateStore) {
|
|||
val knownFingerprints: Set<String>
|
||||
try {
|
||||
knownFingerprints = certificateStore.getKnownFingerprints()
|
||||
} catch (e: CancellationException) {
|
||||
// MUST stay above the broad catch: CancellationException is an
|
||||
// Exception on the JVM.
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
return SecurityCheckResult.Warning(
|
||||
"Could not retrieve known fingerprints: ${e.message}"
|
||||
|
|
@ -70,6 +84,8 @@ class SelfCertVerifier(private val certificateStore: CertificateStore) {
|
|||
private suspend fun handleEmptyFingerprints(leafFingerprint: String): SecurityCheckResult {
|
||||
val markerPresent = try {
|
||||
certificateStore.hasFingerprintBootstrapMarker()
|
||||
} catch (e: CancellationException) {
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
return SecurityCheckResult.Warning(
|
||||
"Could not check fingerprint bootstrap marker: ${e.message}"
|
||||
|
|
@ -95,6 +111,8 @@ class SelfCertVerifier(private val certificateStore: CertificateStore) {
|
|||
// surface the corruption path (empty set + marker present) incorrectly.
|
||||
certificateStore.writeFingerprintBootstrapMarker()
|
||||
SecurityCheckResult.Verified
|
||||
} catch (e: CancellationException) {
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
SecurityCheckResult.Warning(
|
||||
"Could not backfill missing fingerprint: ${e.message}"
|
||||
|
|
|
|||
|
|
@ -399,6 +399,13 @@ class TunnelClientImpl(
|
|||
if (muxDemux !== demux) return
|
||||
val alive = try {
|
||||
demux.sendPingAwaitPong(timeoutMillis = keepaliveTimeoutMillis)
|
||||
} catch (e: CancellationException) {
|
||||
// MUST stay above the broad catch (issue #646). Counting a
|
||||
// cancelled Ping as a miss lets an ordinary teardown accumulate
|
||||
// into a synthesised ConnectionFailed("Keepalive Pings missed
|
||||
// N times"). Only reachable because MuxDemux.sendPingAwaitPong
|
||||
// now propagates instead of returning false.
|
||||
throw e
|
||||
} catch (_: Exception) {
|
||||
false
|
||||
}
|
||||
|
|
@ -426,10 +433,22 @@ class TunnelClientImpl(
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A cancelled health check propagates rather than reporting a dead tunnel
|
||||
* (issue #646). `WakeReconnectDecider` reacts to `false` by calling
|
||||
* `disconnect()` and `connect()` -- work performed after cancellation --
|
||||
* and the cancellation rethrow #650 added there could not fire while this
|
||||
* frame (and `MuxDemux.sendPingAwaitPong` below it) swallowed cancellation
|
||||
* into `false` first.
|
||||
*/
|
||||
override suspend fun healthCheck(timeout: Duration): Boolean {
|
||||
val demux = muxDemux ?: return false
|
||||
return try {
|
||||
demux.sendPingAwaitPong(timeoutMillis = timeout.inWholeMilliseconds)
|
||||
} catch (e: CancellationException) {
|
||||
// MUST stay above the broad catch: CancellationException is an
|
||||
// Exception on the JVM.
|
||||
throw e
|
||||
} catch (_: Exception) {
|
||||
false
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,127 @@
|
|||
package com.rousecontext.tunnel
|
||||
|
||||
import java.io.IOException
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFailsWith
|
||||
import kotlin.test.assertIs
|
||||
import kotlin.test.assertTrue
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.runBlocking
|
||||
|
||||
/**
|
||||
* What [CtLogMonitor.check] does with a **cancellation** raised by one of the
|
||||
* suspend calls it guards (#646).
|
||||
*
|
||||
* Two of the three broad `catch (Exception)` clauses in `check` sit around a
|
||||
* suspend call, so a scope teardown mid-check came back as
|
||||
* [SecurityCheckResult.Warning] -- a security *finding* about a check that was
|
||||
* simply never run. `.claude/rules/coroutines.md` requires cancellation to
|
||||
* propagate.
|
||||
*
|
||||
* The layer directly below already gets this right: `CompositeCtLogFetcher`
|
||||
* rethrows cancellation at both of its catches. Swallowing it one frame up
|
||||
* defeats that entirely.
|
||||
*
|
||||
* No test here cancels the surrounding scope. A cancelled scope would make the
|
||||
* assertion ambiguous -- the test's own cancellation is indistinguishable from
|
||||
* the one under test -- so each collaborator raises cancellation directly while
|
||||
* the job stays active. That is also the shape a stray
|
||||
* `java.util.concurrent.CancellationException` takes (a cancelled `Future`
|
||||
* inside a store implementation), which is the one vector that survives the
|
||||
* three suspend `SecurityCheckPreferences` calls sitting between `check()`
|
||||
* returning and `SecurityCheckWorker` reaching `notifier.postInfo`.
|
||||
*
|
||||
* Each site is pinned in both directions: cancellation must come back as
|
||||
* cancellation, and an ordinary I/O failure must still come back as a Warning.
|
||||
*
|
||||
* The third catch (`json.decodeFromString`) is deliberately unguarded and has
|
||||
* no test here: `decodeFromString` is not a suspend function, so no coroutine
|
||||
* cancellation can reach that clause. A rethrow there would be dead code of
|
||||
* exactly the kind #646's comments warn about.
|
||||
*/
|
||||
class CtLogMonitorCancellationTest {
|
||||
|
||||
@Test
|
||||
fun `cancellation from getSubdomain propagates instead of becoming a Warning`() = runBlocking {
|
||||
val monitor = monitorOver(
|
||||
CancellingSubdomainStore(CancellationException("worker scope torn down"))
|
||||
)
|
||||
|
||||
val thrown = assertFailsWith<Throwable> { monitor.check() }
|
||||
|
||||
assertTrue(
|
||||
thrown is CancellationException,
|
||||
"a cancelled CT check must propagate cancellation, got " +
|
||||
"${thrown.javaClass.name}: ${thrown.message}"
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `an IO failure from getSubdomain is still reported as a Warning`(): Unit = runBlocking {
|
||||
val monitor = monitorOver(CancellingSubdomainStore(IOException("keystore unavailable")))
|
||||
|
||||
val result = monitor.check()
|
||||
|
||||
assertIs<SecurityCheckResult.Warning>(result)
|
||||
assertEquals(
|
||||
"Could not retrieve subdomain: keystore unavailable",
|
||||
result.reason,
|
||||
"the ordinary failure path must be unchanged"
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `cancellation from the CT log fetcher propagates instead of becoming a Warning`() =
|
||||
runBlocking {
|
||||
val monitor = monitorOver(
|
||||
SecurityCertificateStore(subdomain = "abc123"),
|
||||
ThrowingCtLogFetcher(CancellationException("worker scope torn down"))
|
||||
)
|
||||
|
||||
val thrown = assertFailsWith<Throwable> { monitor.check() }
|
||||
|
||||
assertTrue(
|
||||
thrown is CancellationException,
|
||||
"a cancelled CT fetch must propagate cancellation, got " +
|
||||
"${thrown.javaClass.name}: ${thrown.message}"
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `an IO failure from the CT log fetcher is still reported as a Warning`(): Unit =
|
||||
runBlocking {
|
||||
val monitor = monitorOver(
|
||||
SecurityCertificateStore(subdomain = "abc123"),
|
||||
ThrowingCtLogFetcher(IOException("crt.sh unreachable"))
|
||||
)
|
||||
|
||||
val result = monitor.check()
|
||||
|
||||
assertIs<SecurityCheckResult.Warning>(result)
|
||||
assertEquals(
|
||||
"Could not reach CT log service: crt.sh unreachable",
|
||||
result.reason,
|
||||
"the ordinary failure path must be unchanged"
|
||||
)
|
||||
}
|
||||
|
||||
private fun monitorOver(store: CertificateStore, fetcher: CtLogFetcher = FakeCtLogFetcher()) =
|
||||
CtLogMonitor(
|
||||
certificateStore = store,
|
||||
ctLogFetcher = fetcher,
|
||||
expectedIssuers = setOf("C=US, O=Google Trust Services, CN=WE1"),
|
||||
baseDomain = "rousecontext.com"
|
||||
)
|
||||
|
||||
/** Store whose `getSubdomain` raises whatever the test hands it. */
|
||||
private class CancellingSubdomainStore(private val failure: Throwable) :
|
||||
CertificateStore by SecurityCertificateStore() {
|
||||
override suspend fun getSubdomain(): String? = throw failure
|
||||
}
|
||||
|
||||
/** Fetcher that raises whatever the test hands it. */
|
||||
private class ThrowingCtLogFetcher(private val failure: Throwable) : CtLogFetcher {
|
||||
override suspend fun fetch(domain: String): String = throw failure
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,276 @@
|
|||
package com.rousecontext.tunnel
|
||||
|
||||
import java.io.IOException
|
||||
import java.util.concurrent.CopyOnWriteArrayList
|
||||
import java.util.concurrent.TimeUnit
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFalse
|
||||
import kotlin.test.assertNotNull
|
||||
import kotlin.test.assertTrue
|
||||
import kotlin.time.Duration.Companion.seconds
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.awaitCancellation
|
||||
import kotlinx.coroutines.cancel
|
||||
import kotlinx.coroutines.cancelChildren
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import org.junit.jupiter.api.Timeout
|
||||
|
||||
/**
|
||||
* What the keepalive path does when the **caller** is cancelled while a Ping is
|
||||
* in flight (#646).
|
||||
*
|
||||
* Three broad catches sit on top of each other here, and the order they are
|
||||
* fixed in is the whole point:
|
||||
*
|
||||
* ```
|
||||
* MuxDemux.sendPingAwaitPong catch (_: Exception) { false }
|
||||
* TunnelClientImpl.runKeepaliveLoop catch (_: Exception) { false } -> counts a miss
|
||||
* TunnelClientImpl.healthCheck catch (_: Exception) { false } -> "tunnel is dead"
|
||||
* ```
|
||||
*
|
||||
* #650 added a cancellation rethrow to `WakeReconnectDecider`, above
|
||||
* `healthCheck`. It could not fire: both frames below it converted cancellation
|
||||
* into `false` first. **A guard's presence is not evidence of its
|
||||
* reachability**, and the swallow has to be fixed at the lowest frame that
|
||||
* performs it or every guard above stays dead. So the [MuxDemux] test here is
|
||||
* the load-bearing one; the two [TunnelClientImpl] tests only become meaningful
|
||||
* once it passes.
|
||||
*
|
||||
* The cost of getting this wrong is concrete: a cancelled Ping increments the
|
||||
* miss counter until [TunnelClientImpl] synthesises
|
||||
* `ConnectionFailed("Keepalive Pings missed N times")` out of an ordinary
|
||||
* teardown, and a cancelled `healthCheck` reads as a dead tunnel, after which
|
||||
* the wake path proceeds to `disconnect()` and `connect()` -- work performed
|
||||
* after cancellation.
|
||||
*
|
||||
* ## Why these tests cancel a job rather than throwing a cancellation
|
||||
*
|
||||
* [MuxDemux.sendPingAwaitPong] discriminates on the *job*, not on the exception
|
||||
* type, and it has to: [MuxDemux.closeAllQuietly] **cancels** every pending ping
|
||||
* waiter when the transport dies (issue #230), and that must keep reading as a
|
||||
* failed health check -- a dead relay is not a cancelled caller. A blanket
|
||||
* rethrow there hangs `MuxDemuxTest.awaitPongReturnsFalseWhenTransportClosedQuietly`.
|
||||
* So the cancellation tests below cancel the calling job, which is the real
|
||||
* production vector, and the discriminator itself is pinned by the
|
||||
* "transport torn down under an active caller" test.
|
||||
*
|
||||
* Cancelling the caller is not ambiguous here the way it would be for a thrown
|
||||
* result: the defect returns `Result.success(false)` and the fix returns a
|
||||
* failure, so the two are distinguishable without relying on which cancellation
|
||||
* arrived.
|
||||
*/
|
||||
@Timeout(value = 60, unit = TimeUnit.SECONDS, threadMode = Timeout.ThreadMode.SEPARATE_THREAD)
|
||||
class KeepaliveCancellationTest {
|
||||
|
||||
@Test
|
||||
fun `cancelling the caller propagates out of sendPingAwaitPong`() = runBlocking {
|
||||
val demux = MuxDemux()
|
||||
demux.onOutgoingFrame = { awaitCancellation() }
|
||||
|
||||
var outcome: Result<Boolean>? = null
|
||||
val pinger = launch {
|
||||
outcome = runCatching { demux.sendPingAwaitPong(timeoutMillis = LONG_TIMEOUT_MS) }
|
||||
}
|
||||
delay(SETTLE_MS)
|
||||
pinger.cancel()
|
||||
pinger.join()
|
||||
|
||||
val result = assertNotNull(outcome, "the ping coroutine never recorded an outcome")
|
||||
assertTrue(
|
||||
result.isFailure,
|
||||
"a cancelled Ping must propagate cancellation rather than read as a missed " +
|
||||
"Pong, but came back as $result"
|
||||
)
|
||||
assertTrue(
|
||||
result.exceptionOrNull() is CancellationException,
|
||||
"expected cancellation, got ${result.exceptionOrNull()}"
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `an IO failure from the outgoing frame sink is still a missed Pong`(): Unit = runBlocking {
|
||||
val demux = MuxDemux()
|
||||
demux.onOutgoingFrame = { throw IOException("socket closed") }
|
||||
|
||||
assertFalse(
|
||||
demux.sendPingAwaitPong(timeoutMillis = LONG_TIMEOUT_MS),
|
||||
"a genuinely broken transport must still come back as a missed Pong"
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a transport torn down under an active caller is still a missed Pong`() = runBlocking {
|
||||
// Issue #230's property, re-pinned here because the cancellation guard
|
||||
// added for #646 is the code that has to preserve it: closeAllQuietly
|
||||
// cancels the waiter, and the caller is NOT cancelled, so this must stay
|
||||
// `false` rather than propagating.
|
||||
val demux = MuxDemux()
|
||||
demux.onOutgoingFrame = { /* swallow: the relay is dead */ }
|
||||
|
||||
var outcome: Result<Boolean>? = null
|
||||
val pinger = launch {
|
||||
outcome = runCatching { demux.sendPingAwaitPong(timeoutMillis = LONG_TIMEOUT_MS) }
|
||||
}
|
||||
delay(SETTLE_MS)
|
||||
demux.closeAllQuietly()
|
||||
pinger.join()
|
||||
|
||||
assertEquals(
|
||||
Result.success(false),
|
||||
outcome,
|
||||
"a dead relay is not a cancelled caller: this must stay a failed health check"
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `cancelling the caller propagates out of healthCheck`() = runBlocking {
|
||||
val client = TunnelClientImpl(
|
||||
scope = this,
|
||||
webSocketFactory = HangingSendFactory(),
|
||||
keepaliveIntervalMillis = QUIET_KEEPALIVE_MS
|
||||
)
|
||||
client.connect(URL)
|
||||
|
||||
var outcome: Result<Boolean>? = null
|
||||
val checker = launch { outcome = runCatching { client.healthCheck(LONG_TIMEOUT) } }
|
||||
delay(SETTLE_MS)
|
||||
checker.cancel()
|
||||
checker.join()
|
||||
|
||||
val result = assertNotNull(outcome, "the health check never recorded an outcome")
|
||||
assertTrue(
|
||||
result.isFailure,
|
||||
"a cancelled health check must propagate cancellation so the guard in " +
|
||||
"WakeReconnectDecider can fire, but came back as $result"
|
||||
)
|
||||
assertTrue(
|
||||
result.exceptionOrNull() is CancellationException,
|
||||
"expected cancellation, got ${result.exceptionOrNull()}"
|
||||
)
|
||||
|
||||
coroutineContext.cancelChildren()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a broken transport still makes healthCheck report false`(): Unit = runBlocking {
|
||||
val client = TunnelClientImpl(
|
||||
scope = this,
|
||||
webSocketFactory = FailingSendFactory(IOException("socket closed")),
|
||||
keepaliveIntervalMillis = QUIET_KEEPALIVE_MS
|
||||
)
|
||||
client.connect(URL)
|
||||
|
||||
assertFalse(
|
||||
client.healthCheck(SHORT_TIMEOUT),
|
||||
"a genuinely broken transport must still report the tunnel as dead"
|
||||
)
|
||||
|
||||
coroutineContext.cancelChildren()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `tearing the tunnel scope down mid-Ping is not counted as a missed Ping`() = runBlocking {
|
||||
val logs = CopyOnWriteArrayList<String>()
|
||||
// The client's own scope, so it can be torn down while the test's scope
|
||||
// stays alive to make the assertion.
|
||||
val tunnelScope = CoroutineScope(coroutineContext + Job(coroutineContext[Job]))
|
||||
val client = TunnelClientImpl(
|
||||
scope = tunnelScope,
|
||||
webSocketFactory = HangingSendFactory(),
|
||||
log = { _, message -> logs += message },
|
||||
keepaliveIntervalMillis = BRISK_KEEPALIVE_MS,
|
||||
keepaliveMaxMisses = 5
|
||||
)
|
||||
client.connect(URL)
|
||||
// The loop is now parked inside the Ping, which is where a service
|
||||
// shutdown finds it.
|
||||
delay(SETTLE_MS)
|
||||
|
||||
tunnelScope.cancel()
|
||||
delay(SETTLE_MS)
|
||||
|
||||
assertTrue(
|
||||
logs.none { it.contains("keepalive Ping missed") },
|
||||
"a cancelled Ping is teardown, not a missed Pong, but the loop logged: $logs"
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a broken transport still trips the keepalive miss counter`() = runBlocking {
|
||||
val errors = mutableListOf<TunnelError>()
|
||||
val client = TunnelClientImpl(
|
||||
scope = this,
|
||||
webSocketFactory = FailingSendFactory(IOException("socket closed")),
|
||||
keepaliveIntervalMillis = BRISK_KEEPALIVE_MS,
|
||||
keepaliveMaxMisses = 1
|
||||
)
|
||||
val collector = launch { client.errors.collect { errors += it } }
|
||||
delay(SETTLE_MS)
|
||||
client.connect(URL)
|
||||
delay(SETTLE_MS)
|
||||
|
||||
assertTrue(
|
||||
errors.any {
|
||||
it is TunnelError.ConnectionFailed && it.message?.contains("Keepalive") == true
|
||||
},
|
||||
"the ordinary dead-tunnel path must be unchanged, but errors held $errors"
|
||||
)
|
||||
assertEquals(TunnelState.DISCONNECTED, client.state.value)
|
||||
|
||||
collector.cancel()
|
||||
coroutineContext.cancelChildren()
|
||||
}
|
||||
|
||||
/**
|
||||
* Reports the handshake as open immediately, then never completes a send:
|
||||
* exactly where a scope teardown finds the keepalive.
|
||||
*/
|
||||
private class HangingSendFactory : WebSocketFactory {
|
||||
override fun connect(url: String, listener: WebSocketListener): WebSocketHandle {
|
||||
listener.onOpen()
|
||||
return object : WebSocketHandle {
|
||||
override suspend fun sendBinary(data: ByteArray): Boolean = awaitCancellation()
|
||||
|
||||
override suspend fun sendText(text: String): Boolean = awaitCancellation()
|
||||
|
||||
override suspend fun close(code: Int, reason: String) = Unit
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Reports the handshake as open, then fails every send outright. */
|
||||
private class FailingSendFactory(private val failure: Throwable) : WebSocketFactory {
|
||||
override fun connect(url: String, listener: WebSocketListener): WebSocketHandle {
|
||||
listener.onOpen()
|
||||
return object : WebSocketHandle {
|
||||
override suspend fun sendBinary(data: ByteArray): Boolean = throw failure
|
||||
|
||||
override suspend fun sendText(text: String): Boolean = throw failure
|
||||
|
||||
override suspend fun close(code: Int, reason: String) = Unit
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private companion object {
|
||||
const val URL = "ws://127.0.0.1:1/tunnel"
|
||||
|
||||
/** Longer than any test's lifetime: the Pong wait must not time out. */
|
||||
const val LONG_TIMEOUT_MS = 600_000L
|
||||
val LONG_TIMEOUT = 600.seconds
|
||||
|
||||
val SHORT_TIMEOUT = 1.seconds
|
||||
|
||||
/** Far beyond any test's lifetime: the keepalive loop must not interfere. */
|
||||
const val QUIET_KEEPALIVE_MS = 600_000L
|
||||
|
||||
const val BRISK_KEEPALIVE_MS = 30L
|
||||
|
||||
const val SETTLE_MS = 250L
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,148 @@
|
|||
package com.rousecontext.tunnel
|
||||
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.client.engine.mock.MockEngine
|
||||
import io.ktor.client.plugins.websocket.WebSockets
|
||||
import java.net.ServerSocket
|
||||
import java.util.concurrent.TimeUnit
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertTrue
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CompletableDeferred
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.cancel
|
||||
import kotlinx.coroutines.cancelChildren
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlinx.coroutines.withTimeout
|
||||
import org.junit.jupiter.api.Timeout
|
||||
|
||||
/**
|
||||
* What [KtorWebSocketFactory] does when its scope is torn down while the
|
||||
* WebSocket handshake is still in flight (#646).
|
||||
*
|
||||
* The broad `catch (Exception)` around the whole session lifecycle reported a
|
||||
* cancelled connect to the listener as `onFailure` / `onClosing(1006, …)` --
|
||||
* a transport *failure* synthesised out of a scope teardown, which is the
|
||||
* mirror image of the defect #644 fixed one layer up in
|
||||
* `TunnelClientImpl.connect`.
|
||||
*
|
||||
* The handle's `sessionDeferred` is still completed on the cancellation path.
|
||||
* That is the whole point of `failBind` (#420 finding #4): an awaiter parked in
|
||||
* `sendBinary` has no other way to learn the session will never arrive.
|
||||
* Completing it *with the cancellation* both unblocks the awaiter and tells it
|
||||
* the truth, where reporting `onFailure` did neither. Only the listener
|
||||
* notification is dropped.
|
||||
*
|
||||
* The guard is gated on the job rather than on the exception type. A
|
||||
* cancellation raised by the *session* while this coroutine is still active is
|
||||
* a transport failure the listener must still hear about -- otherwise the
|
||||
* tunnel sits CONNECTED over a dead socket, the #558 symptom. The second test
|
||||
* pins that branch so the discriminator is not just asserted in a comment.
|
||||
*
|
||||
* The peer for the teardown test is a bound-but-never-accepted [ServerSocket]: the TCP
|
||||
* connection completes out of the accept backlog and the HTTP upgrade response
|
||||
* never arrives, so the launched coroutine is reliably parked inside the `try`
|
||||
* when the cancellation lands. [KtorWebSocketFactoryFailBindTest] pins the
|
||||
* ordinary connect-failure direction against a dead port.
|
||||
*/
|
||||
@Timeout(value = 60, unit = TimeUnit.SECONDS, threadMode = Timeout.ThreadMode.SEPARATE_THREAD)
|
||||
class KtorWebSocketFactoryCancellationTest {
|
||||
|
||||
@Test
|
||||
fun `a cancelled handshake is not reported to the listener as a failure`() = runBlocking {
|
||||
val peer = ServerSocket(0)
|
||||
try {
|
||||
val listener = RecordingListener()
|
||||
val inner = CoroutineScope(coroutineContext + Job(coroutineContext[Job]))
|
||||
val factory = KtorWebSocketFactory(scope = inner)
|
||||
|
||||
factory.connect("ws://127.0.0.1:${peer.localPort}/tunnel", listener)
|
||||
// Long enough that the upgrade request is certainly sent and the
|
||||
// coroutine is parked waiting for a response that never comes.
|
||||
delay(HANDSHAKE_SETTLE_MS)
|
||||
assertTrue(
|
||||
listener.failures.isEmpty() && listener.closings.isEmpty(),
|
||||
"precondition: the handshake must still be in flight, but the listener " +
|
||||
"already saw failures=${listener.failures} closings=${listener.closings}"
|
||||
)
|
||||
|
||||
inner.cancel()
|
||||
delay(TEARDOWN_SETTLE_MS)
|
||||
|
||||
assertTrue(
|
||||
listener.failures.isEmpty(),
|
||||
"a cancelled handshake is teardown, not a transport failure, but the " +
|
||||
"listener was told: ${listener.failures}"
|
||||
)
|
||||
assertTrue(
|
||||
listener.closings.isEmpty(),
|
||||
"a cancelled handshake must not synthesise a close, but got " +
|
||||
"${listener.closings}"
|
||||
)
|
||||
} finally {
|
||||
peer.close()
|
||||
coroutineContext.cancelChildren()
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a cancellation from the session is still reported to the listener`() = runBlocking {
|
||||
// The caller is not cancelled: the session itself raises cancellation.
|
||||
// That is a broken transport, not teardown, so the listener must be told
|
||||
// -- and the awaiter must still be released.
|
||||
val listener = RecordingListener()
|
||||
val factory = KtorWebSocketFactory(
|
||||
scope = this,
|
||||
httpClient = HttpClient(MockEngine { throw CancellationException("session died") }) {
|
||||
install(WebSockets)
|
||||
}
|
||||
)
|
||||
|
||||
factory.connect("ws://relay.test/tunnel", listener)
|
||||
|
||||
withTimeout(FAILURE_TIMEOUT_MS) { listener.firstFailure.await() }
|
||||
|
||||
coroutineContext.cancelChildren()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a genuine connect failure is still reported to the listener`() = runBlocking {
|
||||
val listener = RecordingListener()
|
||||
|
||||
// Nothing is listening on port 1, so webSocketSession() throws before
|
||||
// the session is bound -- the path KtorWebSocketFactoryFailBindTest
|
||||
// covers from the awaiter's side.
|
||||
KtorWebSocketFactory(scope = this).connect("ws://127.0.0.1:1/tunnel", listener)
|
||||
|
||||
withTimeout(FAILURE_TIMEOUT_MS) { listener.firstFailure.await() }
|
||||
|
||||
coroutineContext.cancelChildren()
|
||||
}
|
||||
|
||||
private class RecordingListener : WebSocketListener {
|
||||
val failures = mutableListOf<Throwable>()
|
||||
val closings = mutableListOf<Pair<Int, String>>()
|
||||
val firstFailure = CompletableDeferred<Throwable>()
|
||||
|
||||
override fun onOpen() = Unit
|
||||
|
||||
override fun onBinaryMessage(data: ByteArray) = Unit
|
||||
|
||||
override fun onClosing(code: Int, reason: String) {
|
||||
closings += code to reason
|
||||
}
|
||||
|
||||
override fun onFailure(error: Throwable) {
|
||||
failures += error
|
||||
firstFailure.complete(error)
|
||||
}
|
||||
}
|
||||
|
||||
private companion object {
|
||||
const val HANDSHAKE_SETTLE_MS = 500L
|
||||
const val TEARDOWN_SETTLE_MS = 500L
|
||||
const val FAILURE_TIMEOUT_MS = 5_000L
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,119 @@
|
|||
package com.rousecontext.tunnel
|
||||
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.client.engine.mock.MockEngine
|
||||
import io.ktor.client.engine.mock.respond
|
||||
import io.ktor.client.plugins.contentnegotiation.ContentNegotiation
|
||||
import io.ktor.http.HttpStatusCode
|
||||
import io.ktor.serialization.kotlinx.json.json
|
||||
import io.ktor.utils.io.ByteChannel
|
||||
import java.io.IOException
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFailsWith
|
||||
import kotlin.test.assertIs
|
||||
import kotlin.test.assertTrue
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlinx.serialization.json.Json
|
||||
|
||||
/**
|
||||
* What `RelayApiClient.executeRequest` does with a **cancellation** raised by
|
||||
* the request it guards (#646).
|
||||
*
|
||||
* ```
|
||||
* try {
|
||||
* val response = block() // <- the request itself
|
||||
* ...
|
||||
* val body = try { response.bodyAsText() } // <- see below
|
||||
* catch (_: Exception) { "" }
|
||||
* } catch (e: Exception) { RelayApiResult.NetworkError(cause = e) }
|
||||
* ```
|
||||
*
|
||||
* The outer catch turned a cancelled request into a returned
|
||||
* [RelayApiResult.NetworkError], and this one reaches a caller with nothing
|
||||
* suspending in between. `OnboardingFlow.registerAndPersist` maps a
|
||||
* `NetworkError` to an [OnboardingResult] and `execute` folds and returns it
|
||||
* without another suspension point, so a cancelled onboarding hands the UI a
|
||||
* concrete "registration failed" outcome instead of unwinding -- work performed
|
||||
* after cancellation, which `.claude/rules/coroutines.md` forbids.
|
||||
*
|
||||
* ## Why the inner catch is deliberately left alone
|
||||
*
|
||||
* #646 lists `bodyAsText()` as a second site. It is not reachable by
|
||||
* cancellation, and the last test here is the measurement rather than an
|
||||
* argument: Ktor 3's `SaveBody` plugin (installed by default) drains the
|
||||
* response body inside `block()`, so by the time `bodyAsText()` runs it is
|
||||
* reading an in-memory buffer with no cancellable suspension point, and a body
|
||||
* that genuinely fails mid-read surfaces from `block()` as a
|
||||
* `ClosedByteChannelException` at the **outer** catch. A rethrow at the inner
|
||||
* catch would be exactly the dead guard #646's own comments warn about --
|
||||
* `WakeReconnectDecider:56` in the same family -- so it is documented in place
|
||||
* instead of added.
|
||||
*/
|
||||
class RelayApiClientCancellationTest {
|
||||
|
||||
@Test
|
||||
fun `cancellation raised by the request propagates instead of becoming NetworkError`() =
|
||||
runBlocking {
|
||||
val client = clientOver(
|
||||
MockEngine { throw CancellationException("onboarding scope torn down") }
|
||||
)
|
||||
|
||||
val thrown = assertFailsWith<Throwable> { client.requestSubdomain(TOKEN) }
|
||||
|
||||
assertTrue(
|
||||
thrown is CancellationException,
|
||||
"a cancelled relay request must propagate cancellation, got " +
|
||||
"${thrown.javaClass.name}: ${thrown.message}"
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a genuine transport failure is still reported as NetworkError`(): Unit = runBlocking {
|
||||
val client = clientOver(MockEngine { throw IOException("relay unreachable") })
|
||||
|
||||
val result = client.requestSubdomain(TOKEN)
|
||||
|
||||
assertIs<RelayApiResult.NetworkError>(result)
|
||||
assertEquals(
|
||||
"relay unreachable",
|
||||
result.cause.message,
|
||||
"the ordinary failure path must be unchanged"
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a body that fails mid-read surfaces from the request, not the body read`(): Unit =
|
||||
runBlocking {
|
||||
// Pins the reachability claim in the class kdoc: the inner
|
||||
// `bodyAsText()` catch never sees this failure, because SaveBody
|
||||
// already drained the channel inside `block()`. If a future Ktor
|
||||
// upgrade (or a client built without SaveBody) moves the read back
|
||||
// to `bodyAsText()`, this test flips to `RelayApiResult.Error` and
|
||||
// the inner catch becomes worth guarding.
|
||||
val client = clientOver(
|
||||
MockEngine {
|
||||
respond(
|
||||
content = ByteChannel().apply { cancel(IOException("reset mid-body")) },
|
||||
status = HttpStatusCode.InternalServerError
|
||||
)
|
||||
}
|
||||
)
|
||||
|
||||
val result = client.requestSubdomain(TOKEN)
|
||||
|
||||
assertIs<RelayApiResult.NetworkError>(result)
|
||||
}
|
||||
|
||||
private fun clientOver(engine: MockEngine) = RelayApiClient(
|
||||
baseUrl = "http://relay.test",
|
||||
httpClient = HttpClient(engine) {
|
||||
install(ContentNegotiation) { json(Json { ignoreUnknownKeys = true }) }
|
||||
}
|
||||
)
|
||||
|
||||
private companion object {
|
||||
const val TOKEN = "fake-firebase-id-token"
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,163 @@
|
|||
package com.rousecontext.tunnel
|
||||
|
||||
import java.io.IOException
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertFailsWith
|
||||
import kotlin.test.assertIs
|
||||
import kotlin.test.assertTrue
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.runBlocking
|
||||
|
||||
/**
|
||||
* What [SelfCertVerifier] does with a **cancellation** raised by one of the
|
||||
* three suspend store calls it guards (#646).
|
||||
*
|
||||
* All three catches sit around `suspend` members of [CertificateStore], so a
|
||||
* scope teardown mid-verify came back as [SecurityCheckResult.Warning] -- the
|
||||
* same "security finding about a check that never ran" shape as
|
||||
* [CtLogMonitorCancellationTest].
|
||||
*
|
||||
* The third one is the sharpest: [SelfCertVerifier] backfills a fingerprint and
|
||||
* *then* writes the one-shot bootstrap marker, deliberately in that order, so a
|
||||
* failed fingerprint write does not claim bootstrap completed (issue #210).
|
||||
* Swallowing cancellation between the two turns a teardown into exactly the
|
||||
* half-written state that ordering exists to prevent, and reports it as a
|
||||
* Warning rather than letting the caller retry.
|
||||
*
|
||||
* As in [CtLogMonitorCancellationTest], no test cancels the surrounding scope:
|
||||
* the collaborator raises cancellation directly so the assertion is about the
|
||||
* code under test and not about the test's own teardown.
|
||||
*/
|
||||
class SelfCertVerifierCancellationTest {
|
||||
|
||||
@Test
|
||||
fun `cancellation from getKnownFingerprints propagates instead of becoming a Warning`() =
|
||||
runBlocking {
|
||||
val verifier = SelfCertVerifier(
|
||||
FingerprintStore(onGetKnown = CancellationException("worker scope torn down"))
|
||||
)
|
||||
|
||||
val thrown = assertFailsWith<Throwable> { verifier.verify(listOf(LEAF)) }
|
||||
|
||||
assertTrue(
|
||||
thrown is CancellationException,
|
||||
"a cancelled verify must propagate cancellation, got " +
|
||||
"${thrown.javaClass.name}: ${thrown.message}"
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `an IO failure from getKnownFingerprints is still reported as a Warning`(): Unit =
|
||||
runBlocking {
|
||||
val verifier = SelfCertVerifier(
|
||||
FingerprintStore(onGetKnown = IOException("Storage unavailable"))
|
||||
)
|
||||
|
||||
val result = verifier.verify(listOf(LEAF))
|
||||
|
||||
assertIs<SecurityCheckResult.Warning>(result)
|
||||
assertTrue(
|
||||
result.reason.startsWith("Could not retrieve known fingerprints:"),
|
||||
"the ordinary failure path must be unchanged, got: ${result.reason}"
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `cancellation from the bootstrap marker read propagates instead of a Warning`() =
|
||||
runBlocking {
|
||||
val verifier = SelfCertVerifier(
|
||||
FingerprintStore(onMarkerRead = CancellationException("worker scope torn down"))
|
||||
)
|
||||
|
||||
val thrown = assertFailsWith<Throwable> { verifier.verify(listOf(LEAF)) }
|
||||
|
||||
assertTrue(
|
||||
thrown is CancellationException,
|
||||
"a cancelled marker read must propagate cancellation, got " +
|
||||
"${thrown.javaClass.name}: ${thrown.message}"
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `an IO failure on the bootstrap marker read is still reported as a Warning`(): Unit =
|
||||
runBlocking {
|
||||
val verifier = SelfCertVerifier(
|
||||
FingerprintStore(onMarkerRead = IOException("Storage unavailable"))
|
||||
)
|
||||
|
||||
val result = verifier.verify(listOf(LEAF))
|
||||
|
||||
assertIs<SecurityCheckResult.Warning>(result)
|
||||
assertTrue(
|
||||
result.reason.startsWith("Could not check fingerprint bootstrap marker:"),
|
||||
"the ordinary failure path must be unchanged, got: ${result.reason}"
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `cancellation during the fingerprint backfill propagates instead of a Warning`() =
|
||||
runBlocking {
|
||||
// Issue #210 ordering: the fingerprint is stored, then the marker is
|
||||
// written. A cancellation between them must not be reported as a
|
||||
// completed-but-degraded check.
|
||||
val verifier = SelfCertVerifier(
|
||||
FingerprintStore(onMarkerWrite = CancellationException("worker scope torn down"))
|
||||
)
|
||||
|
||||
val thrown = assertFailsWith<Throwable> { verifier.verify(listOf(LEAF)) }
|
||||
|
||||
assertTrue(
|
||||
thrown is CancellationException,
|
||||
"a cancelled backfill must propagate cancellation, got " +
|
||||
"${thrown.javaClass.name}: ${thrown.message}"
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `an IO failure during the fingerprint backfill is still reported as a Warning`(): Unit =
|
||||
runBlocking {
|
||||
val verifier = SelfCertVerifier(
|
||||
FingerprintStore(onMarkerWrite = IOException("Storage unavailable"))
|
||||
)
|
||||
|
||||
val result = verifier.verify(listOf(LEAF))
|
||||
|
||||
assertIs<SecurityCheckResult.Warning>(result)
|
||||
assertTrue(
|
||||
result.reason.startsWith("Could not backfill missing fingerprint:"),
|
||||
"the ordinary failure path must be unchanged, got: ${result.reason}"
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Store with an empty fingerprint set and no bootstrap marker -- the state
|
||||
* that drives [SelfCertVerifier] through all three guarded calls -- with an
|
||||
* injectable failure at each one.
|
||||
*/
|
||||
private class FingerprintStore(
|
||||
private val onGetKnown: Throwable? = null,
|
||||
private val onMarkerRead: Throwable? = null,
|
||||
private val onMarkerWrite: Throwable? = null
|
||||
) : CertificateStore by SecurityCertificateStore() {
|
||||
override suspend fun getKnownFingerprints(): Set<String> {
|
||||
onGetKnown?.let { throw it }
|
||||
return emptySet()
|
||||
}
|
||||
|
||||
override suspend fun hasFingerprintBootstrapMarker(): Boolean {
|
||||
onMarkerRead?.let { throw it }
|
||||
return false
|
||||
}
|
||||
|
||||
override suspend fun storeFingerprint(fingerprint: String) = Unit
|
||||
|
||||
override suspend fun writeFingerprintBootstrapMarker() {
|
||||
onMarkerWrite?.let { throw it }
|
||||
}
|
||||
}
|
||||
|
||||
private companion object {
|
||||
/** Any non-empty DER stand-in; only its SHA-256 is ever taken. */
|
||||
val LEAF = byteArrayOf(0x30, 0x03, 0x02, 0x01, 0x00)
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,191 @@
|
|||
package com.rousecontext.tunnel
|
||||
|
||||
import java.io.IOException
|
||||
import kotlin.test.AfterTest
|
||||
import kotlin.test.BeforeTest
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFailsWith
|
||||
import kotlin.test.assertIs
|
||||
import kotlin.test.assertTrue
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.runBlocking
|
||||
|
||||
/**
|
||||
* What the two provisioning flows do with a **cancellation** raised while they
|
||||
* are persisting freshly-issued state (#646).
|
||||
*
|
||||
* Both wrap a run of `suspend` [CertificateStore] writes in a broad
|
||||
* `catch (Exception)` whose handler performs a **rollback**:
|
||||
*
|
||||
* - [OnboardingFlow] calls `certificateStore.clear()` and returns
|
||||
* [OnboardingResult.StorageFailed];
|
||||
* - [CertProvisioningFlow] calls `certificateStore.clearCertificates()` and
|
||||
* returns [CertProvisioningResult.StorageFailed].
|
||||
*
|
||||
* A cancelled write is not a failed write. The rollback is the consequence that
|
||||
* actually lands here: it is real destructive work performed *after*
|
||||
* cancellation, which `.claude/rules/coroutines.md` forbids outright, and in
|
||||
* [OnboardingFlow]'s case `clear()` drops the subdomain the relay has already
|
||||
* assigned -- the device is registered upstream and blank locally.
|
||||
*
|
||||
* Each flow is pinned in both directions, and each cancellation test asserts
|
||||
* the rollback did **not** run, not merely that something was thrown: the
|
||||
* rollback is the damage, so its absence is the property worth holding.
|
||||
*
|
||||
* As elsewhere in this sweep the scope is never cancelled by the test -- the
|
||||
* store raises cancellation directly, so nothing about the assertion depends on
|
||||
* the test's own teardown.
|
||||
*/
|
||||
class StorageRollbackCancellationTest {
|
||||
|
||||
private val mockServer = MockRelayServer()
|
||||
|
||||
@BeforeTest
|
||||
fun setUp() = mockServer.start()
|
||||
|
||||
@AfterTest
|
||||
fun tearDown() = mockServer.stop()
|
||||
|
||||
@Test
|
||||
fun `cancelling an onboarding write propagates and does not roll back`() = runBlocking {
|
||||
val store = RollbackRecordingStore(
|
||||
onIdentityWrite = CancellationException("onboarding scope torn down")
|
||||
)
|
||||
|
||||
val thrown = assertFailsWith<Throwable> { onboardingOver(store).execute(TOKEN, FCM) }
|
||||
|
||||
assertTrue(
|
||||
thrown is CancellationException,
|
||||
"a cancelled onboarding write must propagate cancellation, got " +
|
||||
"${thrown.javaClass.name}: ${thrown.message}"
|
||||
)
|
||||
assertEquals(
|
||||
0,
|
||||
store.clearCalls,
|
||||
"cancellation must not trigger the onboarding rollback: a cancelled write is " +
|
||||
"not a failed write, and clear() drops the subdomain the relay already assigned"
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `an IO failure during an onboarding write still rolls back and reports it`(): Unit =
|
||||
runBlocking {
|
||||
val store = RollbackRecordingStore(onIdentityWrite = IOException("Disk full"))
|
||||
|
||||
val result = onboardingOver(store).execute(TOKEN, FCM)
|
||||
|
||||
assertIs<OnboardingResult.StorageFailed>(result)
|
||||
assertEquals(
|
||||
1,
|
||||
store.clearCalls,
|
||||
"the ordinary failure path must still roll back"
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `cancelling a cert-provisioning write propagates and does not roll back`() = runBlocking {
|
||||
val store = RollbackRecordingStore(
|
||||
onCertWrite = CancellationException("provisioning scope torn down")
|
||||
)
|
||||
store.storeSubdomain("abc123")
|
||||
|
||||
val thrown = assertFailsWith<Throwable> { provisioningOver(store).execute(TOKEN) }
|
||||
|
||||
assertTrue(
|
||||
thrown is CancellationException,
|
||||
"a cancelled cert write must propagate cancellation, got " +
|
||||
"${thrown.javaClass.name}: ${thrown.message}"
|
||||
)
|
||||
assertEquals(
|
||||
0,
|
||||
store.clearCertificatesCalls,
|
||||
"cancellation must not trigger the cert rollback"
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `an IO failure during a cert write still rolls back and reports it`(): Unit = runBlocking {
|
||||
val store = RollbackRecordingStore(onCertWrite = IOException("Disk full"))
|
||||
store.storeSubdomain("abc123")
|
||||
|
||||
val result = provisioningOver(store).execute(TOKEN)
|
||||
|
||||
assertIs<CertProvisioningResult.StorageFailed>(result)
|
||||
assertEquals(
|
||||
1,
|
||||
store.clearCertificatesCalls,
|
||||
"the ordinary failure path must still roll back"
|
||||
)
|
||||
}
|
||||
|
||||
private fun onboardingOver(store: CertificateStore) = OnboardingFlow(
|
||||
relayApiClient = RelayApiClient(baseUrl = mockServer.baseUrl),
|
||||
certificateStore = store
|
||||
)
|
||||
|
||||
private fun provisioningOver(store: CertificateStore) = CertProvisioningFlow(
|
||||
csrGenerator = CsrGenerator(),
|
||||
relayApiClient = RelayApiClient(baseUrl = mockServer.baseUrl),
|
||||
certificateStore = store,
|
||||
deviceKeyManager = InMemoryDeviceKeyManager()
|
||||
)
|
||||
|
||||
/**
|
||||
* Store that fails a chosen family of writes and counts rollback calls.
|
||||
* Everything else is delegated to a real in-memory store so the flows reach
|
||||
* the guarded write in their normal state.
|
||||
*/
|
||||
private class RollbackRecordingStore(
|
||||
private val onIdentityWrite: Throwable? = null,
|
||||
private val onCertWrite: Throwable? = null,
|
||||
private val delegate: InMemoryCertificateStore = InMemoryCertificateStore()
|
||||
) : CertificateStore by delegate {
|
||||
|
||||
var clearCalls = 0
|
||||
private set
|
||||
|
||||
var clearCertificatesCalls = 0
|
||||
private set
|
||||
|
||||
override suspend fun storeSubdomain(subdomain: String) {
|
||||
onIdentityWrite?.let { throw it }
|
||||
delegate.storeSubdomain(subdomain)
|
||||
}
|
||||
|
||||
override suspend fun storeIntegrationSecrets(secrets: Map<String, String>) {
|
||||
onIdentityWrite?.let { throw it }
|
||||
delegate.storeIntegrationSecrets(secrets)
|
||||
}
|
||||
|
||||
override suspend fun storeCertificate(pemChain: String) {
|
||||
onCertWrite?.let { throw it }
|
||||
delegate.storeCertificate(pemChain)
|
||||
}
|
||||
|
||||
override suspend fun storeClientCertificate(pemChain: String) {
|
||||
onCertWrite?.let { throw it }
|
||||
delegate.storeClientCertificate(pemChain)
|
||||
}
|
||||
|
||||
override suspend fun storeRelayCaCert(pem: String) {
|
||||
onCertWrite?.let { throw it }
|
||||
delegate.storeRelayCaCert(pem)
|
||||
}
|
||||
|
||||
override suspend fun clear() {
|
||||
clearCalls++
|
||||
delegate.clear()
|
||||
}
|
||||
|
||||
override suspend fun clearCertificates() {
|
||||
clearCertificatesCalls++
|
||||
delegate.clearCertificates()
|
||||
}
|
||||
}
|
||||
|
||||
private companion object {
|
||||
const val TOKEN = "fake-firebase-id-token"
|
||||
const val FCM = "fake-fcm-registration-token"
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue