diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt index ee67aad6e..7c3368c9e 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt @@ -291,10 +291,50 @@ internal fun buildRelayConnectTarget( token: String, ): Pair { val (authority, _) = parseEndpoint(endpoint) - val path = "/" + namespace + "?jwt=" + token + // Percent-encode any character in [namespace] that would terminate the + // URL path (`?`, `#`, ` `) or otherwise break parsing — `roomId` comes + // from the kind-30312 `d` tag, which NIP-53 does NOT constrain to a + // safe charset, so a `d` tag containing `?` or `#` would otherwise + // truncate the path and split the JWT into the wrong slot. + val path = "/" + percentEncodePath(namespace) + "?jwt=" + token return authority to path } +/** + * Percent-encode the bytes of [s] that aren't legal in an RFC 3986 URI + * `pchar`. We preserve `:` and `/` literally because the namespace uses + * them as structural separators (`nests/::`), + * and the relay compares the path against its claim root using the same + * canonical form. Anything else — including `?`, `#`, `&`, ` `, control + * bytes, and any non-ASCII — is encoded. + */ +private fun percentEncodePath(s: String): String { + val bytes = s.encodeToByteArray() + val out = StringBuilder(bytes.size) + for (raw in bytes) { + val b = raw.toInt() and 0xFF + val c = b.toChar() + val safe = + (c in 'A'..'Z') || + (c in 'a'..'z') || + (c in '0'..'9') || + c == '-' || c == '_' || c == '.' || c == '~' || + c == '!' || c == '$' || c == '\'' || c == '(' || c == ')' || + c == '*' || c == '+' || c == ',' || c == ';' || c == '=' || + c == ':' || c == '@' || c == '/' + if (safe) { + out.append(c) + } else { + out.append('%') + out.append(HEX[b ushr 4]) + out.append(HEX[b and 0x0F]) + } + } + return out.toString() +} + +private val HEX = charArrayOf('0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'A', 'B', 'C', 'D', 'E', 'F') + /** * Split a typical nests endpoint URL such as `https://relay.example.com/moq` * or `https://relay.example.com:4443/api/v1/moq?room=abc` into the @@ -322,16 +362,37 @@ internal fun parseEndpoint(endpoint: String): Pair { require(authorityRaw.isNotEmpty()) { "endpoint must include an authority (got '$endpoint')" } - val portSep = authorityRaw.lastIndexOf(':') val hasUserInfo = authorityRaw.contains('@') require(!hasUserInfo) { "endpoint must not include userinfo (got '$endpoint')" } + // IPv6 literal authorities use `[host]:port` form (RFC 3986 §3.2.2); + // a naive `lastIndexOf(':')` finds a colon *inside* the address and + // breaks parsing. Detect the bracketed form first and split on the + // colon after `]`. + val (hostPart, portStr) = + if (authorityRaw.startsWith('[')) { + val closeBracket = authorityRaw.indexOf(']') + require(closeBracket > 0) { "malformed IPv6 authority: '$authorityRaw'" } + val host = authorityRaw.substring(0, closeBracket + 1) + val tail = authorityRaw.substring(closeBracket + 1) + when { + tail.isEmpty() -> host to null + tail.startsWith(':') -> host to tail.substring(1) + else -> error("malformed IPv6 authority tail '$tail' in '$endpoint'") + } + } else { + val portSep = authorityRaw.lastIndexOf(':') + if (portSep >= 0) { + authorityRaw.substring(0, portSep) to authorityRaw.substring(portSep + 1) + } else { + authorityRaw to null + } + } + // Strip the port if it's the scheme default so the on-the-wire authority is // canonical. val authority = - if (portSep >= 0) { - val host = authorityRaw.substring(0, portSep) - val portStr = authorityRaw.substring(portSep + 1) + if (portStr != null) { val port = portStr.toIntOrNull() ?: error("malformed port '$portStr' in '$endpoint'") val defaultPort = when (scheme) { @@ -339,9 +400,9 @@ internal fun parseEndpoint(endpoint: String): Pair { "http", "ws" -> 80 else -> -1 } - if (port == defaultPort) host else "$host:$port" + if (port == defaultPort) hostPart else "$hostPart:$port" } else { - authorityRaw + hostPart } return authority to pathRaw diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt index 9a5363731..7fcaf1a15 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt @@ -114,8 +114,18 @@ suspend fun connectReconnectingNestsListener( var attempt = 0 while (true) { val listener = - runCatching { openOnce() }.getOrElse { - state.value = NestsListenerState.Failed("connect failed: ${it.message}", it) + try { + openOnce() + } catch (ce: kotlinx.coroutines.CancellationException) { + // Orchestrator scope was cancelled mid-connect — + // propagate so the reconnect loop dies promptly. + // `runCatching` would have swallowed this and + // run one more loop iteration (potentially + // re-opening a doomed listener) before the next + // suspending call re-checked the cancel. + throw ce + } catch (t: Throwable) { + state.value = NestsListenerState.Failed("connect failed: ${t.message}", t) null } var refreshTriggered = false @@ -151,7 +161,13 @@ suspend fun connectReconnectingNestsListener( // listener; don't bump `attempt` (it's not a // backoff event) so the next openOnce() runs // immediately. - runCatching { listener.close() } + try { + listener.close() + } catch (ce: kotlinx.coroutines.CancellationException) { + throw ce + } catch (_: Throwable) { + // Best-effort — the listener will GC either way. + } attempt = 0 refreshTriggered = true } else if (terminal is NestsListenerState.Failed && !isUserCancelled(terminal)) { @@ -325,8 +341,19 @@ private class ReconnectingHandle( while (currentCoroutineContext().isActive) { val handle = - runCatching { opener(listener) } - .getOrNull() ?: break + try { + opener(listener) + } catch (ce: kotlinx.coroutines.CancellationException) { + // Don't `break` on cancel — let it propagate + // so the launched pumpJob actually dies on + // unsubscribe / scope cancellation. The old + // `runCatching` shape ate the cancel and ran + // one extra iteration before the next + // suspend re-checked active state. + throw ce + } catch (_: Throwable) { + null + } ?: break liveHandleRef.set(handle) try { handle.objects.collect { frames.emit(it) } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt index 9ccb6332d..797466f6e 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt @@ -142,8 +142,15 @@ suspend fun connectReconnectingNestsSpeaker( var attempt = 0 while (true) { val speaker = - runCatching { openOnce() }.getOrElse { - state.value = NestsSpeakerState.Failed("connect failed: ${it.message}", it) + try { + openOnce() + } catch (ce: kotlinx.coroutines.CancellationException) { + // Propagate so the orchestrator dies promptly on + // scope cancellation; `runCatching` would have + // eaten the cancel and run one more iteration. + throw ce + } catch (t: Throwable) { + state.value = NestsSpeakerState.Failed("connect failed: ${t.message}", t) null } var refreshTriggered = false @@ -183,7 +190,13 @@ suspend fun connectReconnectingNestsSpeaker( // speaker; don't bump `attempt` (it's not a // backoff event) so the next openOnce() runs // immediately. - runCatching { speaker.close() } + try { + speaker.close() + } catch (ce: kotlinx.coroutines.CancellationException) { + throw ce + } catch (_: Throwable) { + // Best-effort. + } attempt = 0 refreshTriggered = true } else if (terminal is NestsSpeakerState.Failed && !isUserCancelledSpeaker(terminal)) { @@ -348,8 +361,17 @@ private class ReissuingBroadcastHandle( } if (closed) return@collectLatest val handle = - runCatching { sp.startBroadcasting() } - .getOrNull() ?: return@collectLatest + try { + sp.startBroadcasting() + } catch (ce: kotlinx.coroutines.CancellationException) { + // Don't `return@collectLatest` on cancel — + // propagate so the launched pumpJob actually + // dies on close/scope cancellation. The old + // `runCatching` shape ate the cancel. + throw ce + } catch (_: Throwable) { + null + } ?: return@collectLatest if (closed) { runCatching { handle.close() } return@collectLatest diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqBuffer.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqBuffer.kt index d1b9e9f72..80ada64b6 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqBuffer.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqBuffer.kt @@ -126,7 +126,19 @@ class MoqReader( } /** Thrown when an encoded MoQ frame is malformed or truncated. */ -class MoqCodecException( +open class MoqCodecException( message: String, cause: Throwable? = null, ) : RuntimeException(message, cause) + +/** + * Thrown when a MoQ control frame uses a type code we don't recognise but + * is otherwise well-framed (length-prefixed, fully buffered). Carries the + * full frame size so the caller's pump can skip just this message and + * keep parsing the next one — versus a generic [MoqCodecException] which + * means "this byte stream is corrupted, drop everything". + */ +class MoqUnknownTypeException( + val typeCode: Long, + val bytesConsumed: Int, +) : MoqCodecException("unknown MoQ message type: 0x${typeCode.toString(16)}") diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodec.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodec.kt index 23f85254f..816c4af73 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodec.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodec.kt @@ -87,7 +87,10 @@ object MoqCodec { val type = MoqMessageType.fromCode(typeDec.value) - ?: throw MoqCodecException("unknown MoQ message type: 0x${typeDec.value.toString(16)}") + ?: throw MoqUnknownTypeException( + typeCode = typeDec.value, + bytesConsumed = payloadEnd - offset, + ) val reader = MoqReader(src, payloadStart, payloadEnd) val message = diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt index a12d83b9a..26c197a7b 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt @@ -412,9 +412,18 @@ class MoqSession private constructor( val decoded = try { MoqCodec.decode(buffer) ?: break + } catch (e: MoqUnknownTypeException) { + // Forward-compat: a peer sending a draft-17 control + // message we don't enumerate (FETCH, GOAWAY, + // MAX_SUBSCRIBE_ID, …) shouldn't poison the rest of + // the buffer. Skip just this frame and parse on. + buffer = buffer.copyOfRange(e.bytesConsumed, buffer.size) + continue } catch (e: MoqCodecException) { - // Drop the corrupted buffer; keep the pump alive so the - // next valid frame can recover the session. + // Genuinely-corrupted bytes — we have no idea where + // the next valid frame starts. Drop the buffer; the + // pump stays alive so the next valid frame can + // recover the session. buffer = ByteArray(0) break } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt index 44a6ad55e..f56a26824 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt @@ -412,8 +412,15 @@ class MoqLiteSession internal constructor( */ private suspend fun pumpUniStreams() { try { - transport.incomingUniStreams().collect { stream -> - scope.launch { drainOneGroup(stream) } + // coroutineScope binds each per-stream drain to this pump's + // job — when the pump is cancelled (session close), every + // drain is cancelled with it instead of leaking as a + // sibling on the outer [scope] until the transport's flow + // independently errors out. + kotlinx.coroutines.coroutineScope { + transport.incomingUniStreams().collect { stream -> + launch { drainOneGroup(stream) } + } } } catch (ce: CancellationException) { throw ce @@ -531,8 +538,13 @@ class MoqLiteSession internal constructor( */ private suspend fun pumpInboundBidis() { try { - transport.incomingBidiStreams().collect { bidi -> - scope.launch { handleInboundBidi(bidi) } + // Bind each per-bidi handler to this pump's job (see + // [pumpUniStreams]'s identical comment) so they don't outlive + // bidiPump.cancelAndJoin() in [close]. + kotlinx.coroutines.coroutineScope { + transport.incomingBidiStreams().collect { bidi -> + launch { handleInboundBidi(bidi) } + } } } catch (ce: CancellationException) { throw ce diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsConnectTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsConnectTest.kt index c81bbc5ce..9232f9634 100644 --- a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsConnectTest.kt +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsConnectTest.kt @@ -167,6 +167,41 @@ class NestsConnectTest { ) } + @Test + fun parseEndpoint_handles_IPv6_authorities() { + // No port — naive `lastIndexOf(':')` would find a colon inside + // the address; the bracket-aware path keeps `[::1]` whole. + assertEquals( + "[::1]" to "/moq", + parseEndpoint("https://[::1]/moq"), + ) + assertEquals( + "[2001:db8::1]:4443" to "/moq", + parseEndpoint("https://[2001:db8::1]:4443/moq"), + ) + // Default-port stripping still applies on bracketed authorities. + assertEquals( + "[::1]" to "/", + parseEndpoint("https://[::1]:443/"), + ) + } + + @Test + fun buildRelayConnectTarget_percent_encodes_path_unsafe_chars() { + // A malicious or careless `d` tag containing `?`, `#`, ` ` etc. + // would otherwise truncate the URL path and shove the JWT into + // the wrong slot. Encoding keeps the namespace literal in path + // position; the relay URL-decodes before comparing to claim root. + val (authority, path) = + buildRelayConnectTarget( + endpoint = "https://relay.example.com/moq", + namespace = "nests/30312:0123:room?weird#frag", + token = "tok-abc", + ) + assertEquals("relay.example.com", authority) + assertEquals("/nests/30312:0123:room%3Fweird%23frag?jwt=tok-abc", path) + } + // ---------------------------------------------------------- fakes private class FakeNestsClient( diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodecTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodecTest.kt index 9fd2f06f7..3f95eb148 100644 --- a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodecTest.kt +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodecTest.kt @@ -100,12 +100,18 @@ class MoqCodecTest { } @Test - fun unknown_message_type_is_rejected() { - // Craft a frame with message type 0xFFFF (large varint), length 0, no payload. + fun unknown_message_type_throws_typed_exception_with_full_frame_size() { + // Craft a frame with message type 0xFFFF (large varint), length 4, + // four bytes of payload. The pump uses bytesConsumed to skip past + // the unknown frame instead of dropping the whole buffer. val bogus = MoqWriter() bogus.writeVarint(0xFFFFL) - bogus.writeVarint(0L) - assertFailsWith { MoqCodec.decode(bogus.toByteArray()) } + bogus.writeVarint(4L) + bogus.writeBytes(byteArrayOf(0x01, 0x02, 0x03, 0x04)) + val encoded = bogus.toByteArray() + val ex = assertFailsWith { MoqCodec.decode(encoded) } + assertEquals(0xFFFFL, ex.typeCode) + assertEquals(encoded.size, ex.bytesConsumed) } @Test