fix(nestsClient): more bugs from review — IETF parser, leaks, IPv6, URL encoding, structured concurrency

- Unknown IETF control message types now skip just the unknown frame instead of dropping the entire merge buffer (a peer sending a draft-17 message we don't enumerate — FETCH, GOAWAY, MAX_SUBSCRIBE_ID — would otherwise wedge the pump). Adds MoqUnknownTypeException carrying bytesConsumed so runControlPump can advance past it. - pumpUniStreams + pumpInboundBidis wrap their inner collect in coroutineScope so per-stream drains are children of the pump's job (was: launched on the outer scope as siblings, leaking past cancelAndJoin until the transport's own flow errored out). - parseEndpoint handles IPv6 authorities ([::1], [2001:db8::1]:4443). Naive lastIndexOf(':') used to find a colon inside the address. - buildRelayConnectTarget percent-encodes the namespace path so a malicious / careless `d` tag containing `?`, `#`, `&`, ` ` etc. can't truncate the URL and shove the JWT into the wrong slot. - Reconnect orchestrators (listener + speaker) replace `runCatching { openOnce / close / startBroadcasting }` with explicit try/catch that rethrows CancellationException, so cooperative cancellation from the parent scope dies promptly instead of running one more iteration after the cancel. Adds three regression tests: - parseEndpoint_handles_IPv6_authorities - buildRelayConnectTarget_percent_encodes_path_unsafe_chars - unknown_message_type_throws_typed_exception_with_full_frame_size 215 tests pass, 0 failures.
2026-05-01 18:01:06 +00:00
parent eb20ac2c2a
commit f3a942dbd0
9 changed files with 216 additions and 29 deletions
@@ -291,10 +291,50 @@ internal fun buildRelayConnectTarget(
    token: String,
 ): Pair<String, String> {
    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/<kind>:<host_pubkey>:<roomId>`),
+ * 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<String, String> {

    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<String, String> {
                    "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
@@ -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) }
@@ -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
@@ -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)}")
@@ -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 =
@@ -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
                    }
@@ -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
@@ -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(
@@ -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<MoqCodecException> { MoqCodec.decode(bogus.toByteArray()) }
+        bogus.writeVarint(4L)
+        bogus.writeBytes(byteArrayOf(0x01, 0x02, 0x03, 0x04))
+        val encoded = bogus.toByteArray()
+        val ex = assertFailsWith<MoqUnknownTypeException> { MoqCodec.decode(encoded) }
+        assertEquals(0xFFFFL, ex.typeCode)
+        assertEquals(encoded.size, ex.bytesConsumed)
    }

    @Test