From 37e24ce4f3ab08642d0784a2aafdaae9b1847056 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 22 Apr 2026 03:21:15 +0000 Subject: [PATCH] feat(nestsClient): MoQ varint + SETUP handshake codec MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 3c-1 of the Clubhouse/nests integration. Lands a MoQ-transport control-plane codec (draft-ietf-moq-transport) and a session wrapper that runs the SETUP handshake over any WebTransportSession. Purely commonMain + fully tested end-to-end against FakeWebTransport — no network or Kwik dependency required. commonMain (nestsclient.moq): - `Varint` — QUIC variable-length integer codec per RFC 9000 §16. Encode/decode/size, with typed truncation handling (decode returns null so callers can buffer more and retry). - `MoqWriter` / `MoqReader` — append-only byte writer + bounds-checked reader used by the message codec. - `MoqCodecException` — typed error for malformed frames. - `MoqMessage` sealed class + `MoqMessageType` enum + `SetupParameter`. Phase 3c-1 covers just `ClientSetup` / `ServerSetup` (message types 0x40 / 0x41). - `MoqCodec.encode/decode` — wraps payload with `type (varint) + length (varint)`. Rejects unknown types and trailing bytes inside a declared payload window. - `MoqSession.client/server` — attaches to a WebTransportSession and runs the CLIENT_SETUP / SERVER_SETUP handshake with version negotiation + configurable timeout. - `MoqVersion` constants for draft-11 and draft-17. Tests: - `VarintTest` — all four RFC 9000 §A.1 sample vectors (1/2/4/8 byte), boundary round-trips, negative/overflow rejection, short-buffer returns null, bytesConsumed accuracy. - `MoqCodecTest` — CLIENT_SETUP / SERVER_SETUP round-trip (empty and multi-param), multi-version negotiation, exact wire layout for ServerSetup(1L), truncated-frame returns null, concatenated frames decode with offset, unknown type rejected, trailing bytes rejected, zero-length parameter values allowed. - `MoqSessionTest` — full SETUP exchange over FakeWebTransport (both sides happy path, client falling back to older version, server rejecting when no version overlap + client timing out cleanly). SUBSCRIBE / ANNOUNCE / OBJECT_STREAM / OBJECT_DATAGRAM land in Phase 3c-2 on top of the same MoqSession. https://claude.ai/code/session_013nVLALALKaHVgHm9u5Cg8D --- .../nestsclient/moq/MoqBuffer.kt | 130 ++++++++++++ .../vitorpamplona/nestsclient/moq/MoqCodec.kt | 144 +++++++++++++ .../nestsclient/moq/MoqMessage.kt | 108 ++++++++++ .../nestsclient/moq/MoqSession.kt | 190 ++++++++++++++++++ .../vitorpamplona/nestsclient/moq/Varint.kt | 122 +++++++++++ .../nestsclient/moq/MoqCodecTest.kt | 133 ++++++++++++ .../nestsclient/moq/MoqSessionTest.kt | 117 +++++++++++ .../nestsclient/moq/VarintTest.kt | 114 +++++++++++ 8 files changed, 1058 insertions(+) create mode 100644 nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqBuffer.kt create mode 100644 nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodec.kt create mode 100644 nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqMessage.kt create mode 100644 nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt create mode 100644 nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/Varint.kt create mode 100644 nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodecTest.kt create mode 100644 nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqSessionTest.kt create mode 100644 nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/VarintTest.kt diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqBuffer.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqBuffer.kt new file mode 100644 index 000000000..9786c88a2 --- /dev/null +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqBuffer.kt @@ -0,0 +1,130 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.nestsclient.moq + +/** + * Append-only byte buffer used by the MoQ encoders. Doubles in capacity when + * full. Kept intentionally minimal so this module has no buffer-library + * dependency — helps keep the KMP surface thin. + */ +class MoqWriter( + initialCapacity: Int = 64, +) { + private var buf: ByteArray = ByteArray(initialCapacity) + private var pos: Int = 0 + + val size: Int get() = pos + + fun toByteArray(): ByteArray = buf.copyOf(pos) + + fun writeByte(value: Int) { + ensure(1) + buf[pos++] = value.toByte() + } + + fun writeBytes(bytes: ByteArray) { + ensure(bytes.size) + bytes.copyInto(buf, pos) + pos += bytes.size + } + + fun writeVarint(value: Long) { + ensure(Varint.size(value)) + pos += Varint.writeTo(value, buf, pos) + } + + fun writeVarint(value: Int) = writeVarint(value.toLong()) + + /** Write a length-prefixed byte array (varint length + contents). */ + fun writeLengthPrefixedBytes(bytes: ByteArray) { + writeVarint(bytes.size.toLong()) + writeBytes(bytes) + } + + /** Write a length-prefixed UTF-8 string. */ + fun writeLengthPrefixedString(s: String) = writeLengthPrefixedBytes(s.encodeToByteArray()) + + private fun ensure(more: Int) { + if (pos + more > buf.size) { + var newSize = buf.size * 2 + while (newSize < pos + more) newSize *= 2 + buf = buf.copyOf(newSize) + } + } +} + +/** + * Read cursor over a fully-buffered frame payload. Throws [MoqCodecException] + * on under-read so callers get a typed error surface instead of opaque + * [IndexOutOfBoundsException]s. + */ +class MoqReader( + private val src: ByteArray, + private var pos: Int = 0, + private val end: Int = src.size, +) { + val remaining: Int get() = end - pos + + fun hasMore(): Boolean = pos < end + + fun readByte(): Int { + require(1) + return src[pos++].toInt() and 0xFF + } + + fun readBytes(n: Int): ByteArray { + require(n) + val out = src.copyOfRange(pos, pos + n) + pos += n + return out + } + + fun readVarint(): Long { + val dec = + Varint.decode(src, pos) + ?: throw MoqCodecException("truncated varint at offset=$pos (remaining=$remaining)") + require(dec.bytesConsumed) + pos += dec.bytesConsumed + return dec.value + } + + fun readLengthPrefixedBytes(): ByteArray { + val len = readVarint() + if (len < 0 || len > Int.MAX_VALUE) throw MoqCodecException("absurd length prefix: $len") + return readBytes(len.toInt()) + } + + fun readLengthPrefixedString(): String = readLengthPrefixedBytes().decodeToString() + + private fun require(n: Int) { + if (pos + n > end) { + throw MoqCodecException( + "short read at offset=$pos: wanted $n bytes, have $remaining", + ) + } + } +} + +/** Thrown when an encoded MoQ frame is malformed or truncated. */ +class MoqCodecException( + message: String, + cause: Throwable? = null, +) : RuntimeException(message, cause) diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodec.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodec.kt new file mode 100644 index 000000000..bfb5faa04 --- /dev/null +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodec.kt @@ -0,0 +1,144 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.nestsclient.moq + +/** + * Encode/decode MoQ control-stream messages per draft-ietf-moq-transport. + * + * Wire format for every control message: + * + * message_type (varint) | message_length (varint) | payload bytes + * + * The payload layout is per-message. Payload encoders here produce only the + * payload; [encode] wraps it with type+length for on-the-wire use. + */ +object MoqCodec { + /** + * Encode a full control-stream frame (type + length + payload). + */ + fun encode(message: MoqMessage): ByteArray { + val payload = encodePayload(message) + val frame = MoqWriter(payload.size + 8) + frame.writeVarint(message.type.code) + frame.writeVarint(payload.size.toLong()) + frame.writeBytes(payload) + return frame.toByteArray() + } + + /** + * Decode one control-stream message from [src] starting at [offset]. + * + * Returns the message plus the number of bytes consumed, or null if [src] + * doesn't yet contain a whole frame (caller should buffer more data from + * the control stream and retry). + */ + fun decode( + src: ByteArray, + offset: Int = 0, + ): DecodeResult? { + val typeDec = Varint.decode(src, offset) ?: return null + val lenDec = Varint.decode(src, offset + typeDec.bytesConsumed) ?: return null + val payloadStart = offset + typeDec.bytesConsumed + lenDec.bytesConsumed + val payloadEnd = payloadStart + lenDec.value.toInt() + if (payloadEnd > src.size) return null + + val type = + MoqMessageType.fromCode(typeDec.value) + ?: throw MoqCodecException("unknown MoQ message type: 0x${typeDec.value.toString(16)}") + + val reader = MoqReader(src, payloadStart, payloadEnd) + val message = + when (type) { + MoqMessageType.ClientSetup -> decodeClientSetup(reader) + MoqMessageType.ServerSetup -> decodeServerSetup(reader) + } + if (reader.hasMore()) { + throw MoqCodecException( + "trailing bytes in ${type.name} payload (consumed=${payloadEnd - payloadStart - reader.remaining}, total=${payloadEnd - payloadStart})", + ) + } + return DecodeResult(message, payloadEnd - offset) + } + + private fun encodePayload(message: MoqMessage): ByteArray = + when (message) { + is ClientSetup -> encodeClientSetup(message) + is ServerSetup -> encodeServerSetup(message) + } + + private fun encodeClientSetup(message: ClientSetup): ByteArray { + val w = MoqWriter() + w.writeVarint(message.supportedVersions.size.toLong()) + for (v in message.supportedVersions) w.writeVarint(v) + encodeParameters(w, message.parameters) + return w.toByteArray() + } + + private fun decodeClientSetup(r: MoqReader): ClientSetup { + val nVersions = r.readVarint().toInt() + if (nVersions < 0) throw MoqCodecException("negative version count: $nVersions") + val versions = ArrayList(nVersions) + repeat(nVersions) { versions.add(r.readVarint()) } + val params = decodeParameters(r) + return ClientSetup(versions, params) + } + + private fun encodeServerSetup(message: ServerSetup): ByteArray { + val w = MoqWriter() + w.writeVarint(message.selectedVersion) + encodeParameters(w, message.parameters) + return w.toByteArray() + } + + private fun decodeServerSetup(r: MoqReader): ServerSetup { + val version = r.readVarint() + val params = decodeParameters(r) + return ServerSetup(version, params) + } + + private fun encodeParameters( + w: MoqWriter, + params: List, + ) { + w.writeVarint(params.size.toLong()) + for (p in params) { + w.writeVarint(p.key) + w.writeLengthPrefixedBytes(p.value) + } + } + + private fun decodeParameters(r: MoqReader): List { + val n = r.readVarint().toInt() + if (n < 0) throw MoqCodecException("negative parameter count: $n") + val out = ArrayList(n) + repeat(n) { + val key = r.readVarint() + val value = r.readLengthPrefixedBytes() + out.add(SetupParameter(key, value)) + } + return out + } + + data class DecodeResult( + val message: MoqMessage, + val bytesConsumed: Int, + ) +} diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqMessage.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqMessage.kt new file mode 100644 index 000000000..6af9bb801 --- /dev/null +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqMessage.kt @@ -0,0 +1,108 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.nestsclient.moq + +/** + * Subset of the MoQ-transport control-plane messages needed for a listener + * talking to a nests server. Per draft-ietf-moq-transport, control messages + * on the control stream share the wire layout: + * + * message_type (varint) | message_length (varint) | payload... + * + * This phase (3c-1) covers only the setup handshake. SUBSCRIBE / ANNOUNCE / + * OBJECT messages arrive in Phase 3c-2. + */ +sealed class MoqMessage { + abstract val type: MoqMessageType +} + +/** + * Message type codes per draft-ietf-moq-transport-17. Held as enum so future + * draft revisions can extend without breaking call sites. + */ +enum class MoqMessageType( + val code: Long, +) { + ClientSetup(0x40), + ServerSetup(0x41), + ; + + companion object { + private val byCode = entries.associateBy { it.code } + + fun fromCode(code: Long): MoqMessageType? = byCode[code] + } +} + +/** + * Wire-level setup parameter: key is a varint, value is an opaque byte string + * (MoQ doesn't prescribe a single type for values — some draft revisions use + * varints, others length-prefixed bytes; nests uses bytes in either case). + */ +data class SetupParameter( + val key: Long, + val value: ByteArray, +) { + override fun equals(other: Any?): Boolean { + if (this === other) return true + if (other !is SetupParameter) return false + return key == other.key && value.contentEquals(other.value) + } + + override fun hashCode(): Int = 31 * key.hashCode() + value.contentHashCode() + + companion object { + /** ROLE parameter (key 0x00), draft-specific values. Kept as bytes. */ + const val KEY_ROLE: Long = 0x00L + + /** PATH parameter (key 0x01). */ + const val KEY_PATH: Long = 0x01L + + /** MAX_SUBSCRIBE_ID (key 0x02 in most drafts). */ + const val KEY_MAX_SUBSCRIBE_ID: Long = 0x02L + } +} + +/** + * CLIENT_SETUP (0x40): sent by the client immediately after the WebTransport + * session opens, on the first bidi stream (the control stream). + */ +data class ClientSetup( + val supportedVersions: List, + val parameters: List = emptyList(), +) : MoqMessage() { + override val type: MoqMessageType = MoqMessageType.ClientSetup + + init { + require(supportedVersions.isNotEmpty()) { "must offer at least one version" } + } +} + +/** + * SERVER_SETUP (0x41): response to CLIENT_SETUP. Contains the one version the + * server selected from [ClientSetup.supportedVersions]. + */ +data class ServerSetup( + val selectedVersion: Long, + val parameters: List = emptyList(), +) : MoqMessage() { + override val type: MoqMessageType = MoqMessageType.ServerSetup +} diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt new file mode 100644 index 000000000..745be13ce --- /dev/null +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt @@ -0,0 +1,190 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.nestsclient.moq + +import com.vitorpamplona.nestsclient.transport.WebTransportBidiStream +import com.vitorpamplona.nestsclient.transport.WebTransportSession +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.withTimeout + +/** + * MoQ-transport draft version constants. Value shape: `0xff00000000 | draft` + * is how some draft tests label versions; draft-ietf-moq-transport-17 is + * commonly represented as 0xff000011 but implementations vary. The actual + * version used on the wire is negotiated in [MoqSession.setup]. + * + * Listed as `Long` to survive the MSB-set encoding on JVM. + */ +object MoqVersion { + /** draft-ietf-moq-transport-11 (a popular stable draft). */ + const val DRAFT_11: Long = 0xff00000bL + + /** draft-ietf-moq-transport-17 (newer draft). */ + const val DRAFT_17: Long = 0xff000011L +} + +/** + * Session wrapper over a [WebTransportSession] that speaks MoQ-transport. + * + * Phase 3c-1 ships only [setup] — the CLIENT_SETUP / SERVER_SETUP handshake. + * Phase 3c-2 adds SUBSCRIBE + ANNOUNCE + object-stream multiplexing. + * + * Usage: + * ``` + * val session = MoqSession.client(webTransport).apply { + * setup(listOf(MoqVersion.DRAFT_17)) + * } + * // session.selectedVersion is now the version the server picked. + * ``` + */ +class MoqSession private constructor( + private val transport: WebTransportSession, + /** The single bidi stream every MoQ exchange uses for control. */ + private val controlStream: WebTransportBidiStream, + private val role: Role, +) { + enum class Role { Client, Server } + + /** Server's selected version; null until [setup] returns. */ + var selectedVersion: Long? = null + private set + + /** Parameters the server sent back in SERVER_SETUP. Empty until [setup] returns. */ + var serverParameters: List = emptyList() + private set + + /** + * Run the SETUP handshake. + * + * Client side: writes CLIENT_SETUP with [supportedVersions] + [clientParameters], + * then reads exactly one SERVER_SETUP, stores the result, and returns. + * Server side: reads exactly one CLIENT_SETUP, selects the first mutually + * supported version from [supportedVersions] (where the local list is the + * set of versions the server is willing to accept), writes SERVER_SETUP, + * and returns. + * + * @throws MoqProtocolException if the peer sent an unexpected message, or + * if no version overlap exists (server side). + */ + suspend fun setup( + supportedVersions: List, + clientParameters: List = emptyList(), + handshakeTimeoutMs: Long = 10_000, + ) { + withTimeout(handshakeTimeoutMs) { + when (role) { + Role.Client -> runClientSetup(supportedVersions, clientParameters) + Role.Server -> runServerSetup(supportedVersions, clientParameters) + } + } + } + + private suspend fun runClientSetup( + supportedVersions: List, + clientParameters: List, + ) { + controlStream.write(MoqCodec.encode(ClientSetup(supportedVersions, clientParameters))) + val reply = readOneMessage() + val server = + reply as? ServerSetup + ?: throw MoqProtocolException("expected SERVER_SETUP, got ${reply.type.name}") + if (server.selectedVersion !in supportedVersions) { + throw MoqProtocolException( + "server picked version 0x${server.selectedVersion.toString(16)} which we did not offer", + ) + } + selectedVersion = server.selectedVersion + serverParameters = server.parameters + } + + private suspend fun runServerSetup( + acceptedVersions: List, + serverParameters: List, + ) { + val incoming = readOneMessage() + val client = + incoming as? ClientSetup + ?: throw MoqProtocolException("expected CLIENT_SETUP, got ${incoming.type.name}") + val overlap = + client.supportedVersions.firstOrNull { it in acceptedVersions } + ?: throw MoqProtocolException( + "no mutually-supported MoQ version (client offered ${client.supportedVersions})", + ) + controlStream.write(MoqCodec.encode(ServerSetup(overlap, serverParameters))) + this.selectedVersion = overlap + this.serverParameters = serverParameters + } + + /** + * Read exactly one full MoQ message from the control stream. + * + * Phase 3c-1 assumes a whole message fits in a single transport write + * (SETUP frames are only a handful of bytes and the peer always writes + * them atomically). Phase 3c-2 will replace this with a buffer-and-retry + * loop for messages that may fragment across chunks. + */ + private suspend fun readOneMessage(): MoqMessage { + val chunk = controlStream.incoming().first() + val decoded = + MoqCodec.decode(chunk) + ?: throw MoqProtocolException( + "control-stream chunk did not contain a complete MoQ frame (size=${chunk.size})", + ) + return decoded.message + } + + /** Close the underlying transport. */ + suspend fun close( + code: Int = 0, + reason: String = "", + ) { + controlStream.finish() + transport.close(code, reason) + } + + companion object { + /** + * Attach to a [WebTransportSession] in the client role. Opens the + * control stream eagerly so [setup] doesn't need to manage stream + * acquisition. + */ + suspend fun client(transport: WebTransportSession): MoqSession { + val control = transport.openBidiStream() + return MoqSession(transport, control, Role.Client) + } + + /** + * Attach to a [WebTransportSession] in the server role over an + * already-accepted control stream (usually the first bidi stream the + * peer opened). Used in tests — a real server accepts the first bidi. + */ + fun server( + transport: WebTransportSession, + control: WebTransportBidiStream, + ): MoqSession = MoqSession(transport, control, Role.Server) + } +} + +/** Thrown when the peer violates the MoQ-transport state machine. */ +class MoqProtocolException( + message: String, + cause: Throwable? = null, +) : RuntimeException(message, cause) diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/Varint.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/Varint.kt new file mode 100644 index 000000000..ed1529683 --- /dev/null +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/Varint.kt @@ -0,0 +1,122 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.nestsclient.moq + +/** + * QUIC variable-length integer codec, per RFC 9000 §16. + * + * The two most significant bits of the first byte encode the total length: + * 00 → 1 byte, 6-bit value (0 .. 63) + * 01 → 2 bytes, 14-bit value (0 .. 16_383) + * 10 → 4 bytes, 30-bit value (0 .. 1_073_741_823) + * 11 → 8 bytes, 62-bit value (0 .. 4_611_686_018_427_387_903) + * + * MoQ messages, parameters, and most length prefixes use this encoding. + */ +object Varint { + const val MAX_VALUE: Long = (1L shl 62) - 1 + + /** Number of bytes required to encode [value]. */ + fun size(value: Long): Int { + require(value >= 0) { "varint must be non-negative: $value" } + require(value <= MAX_VALUE) { "varint overflow: $value" } + return when { + value < (1L shl 6) -> 1 + value < (1L shl 14) -> 2 + value < (1L shl 30) -> 4 + else -> 8 + } + } + + /** Encode [value] into a fresh ByteArray. */ + fun encode(value: Long): ByteArray { + val buf = ByteArray(size(value)) + writeTo(value, buf, 0) + return buf + } + + /** + * Write [value] to [dst] starting at [offset]. Returns the number of bytes + * written. [dst] must have room for [size] bytes starting at [offset]. + */ + fun writeTo( + value: Long, + dst: ByteArray, + offset: Int, + ): Int { + val size = size(value) + when (size) { + 1 -> { + dst[offset] = value.toByte() + } + + 2 -> { + dst[offset] = (0x40 or ((value ushr 8).toInt() and 0x3F)).toByte() + dst[offset + 1] = (value and 0xFF).toByte() + } + + 4 -> { + dst[offset] = (0x80 or ((value ushr 24).toInt() and 0x3F)).toByte() + dst[offset + 1] = ((value ushr 16) and 0xFF).toByte() + dst[offset + 2] = ((value ushr 8) and 0xFF).toByte() + dst[offset + 3] = (value and 0xFF).toByte() + } + + 8 -> { + dst[offset] = (0xC0 or ((value ushr 56).toInt() and 0x3F)).toByte() + dst[offset + 1] = ((value ushr 48) and 0xFF).toByte() + dst[offset + 2] = ((value ushr 40) and 0xFF).toByte() + dst[offset + 3] = ((value ushr 32) and 0xFF).toByte() + dst[offset + 4] = ((value ushr 24) and 0xFF).toByte() + dst[offset + 5] = ((value ushr 16) and 0xFF).toByte() + dst[offset + 6] = ((value ushr 8) and 0xFF).toByte() + dst[offset + 7] = (value and 0xFF).toByte() + } + } + return size + } + + /** + * Decode a varint starting at [offset] in [src]. Returns [DecodeResult] with + * the value and the number of bytes consumed, or null if [src] doesn't + * contain enough bytes (caller should buffer more and retry). + */ + fun decode( + src: ByteArray, + offset: Int = 0, + ): DecodeResult? { + if (offset >= src.size) return null + val first = src[offset].toInt() and 0xFF + val length = 1 shl ((first ushr 6) and 0x03) + if (offset + length > src.size) return null + + var value = (first and 0x3F).toLong() + for (i in 1 until length) { + value = (value shl 8) or ((src[offset + i].toInt() and 0xFF).toLong()) + } + return DecodeResult(value, length) + } + + data class DecodeResult( + val value: Long, + val bytesConsumed: Int, + ) +} diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodecTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodecTest.kt new file mode 100644 index 000000000..9fd2f06f7 --- /dev/null +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodecTest.kt @@ -0,0 +1,133 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.nestsclient.moq + +import kotlin.test.Test +import kotlin.test.assertContentEquals +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertIs +import kotlin.test.assertNull + +class MoqCodecTest { + @Test + fun client_setup_round_trip_no_parameters() { + val msg = ClientSetup(supportedVersions = listOf(MoqVersion.DRAFT_17)) + val encoded = MoqCodec.encode(msg) + val decoded = MoqCodec.decode(encoded)!! + assertEquals(encoded.size, decoded.bytesConsumed) + assertEquals(msg, decoded.message) + } + + @Test + fun client_setup_round_trip_multi_version_and_parameters() { + val msg = + ClientSetup( + supportedVersions = listOf(MoqVersion.DRAFT_11, MoqVersion.DRAFT_17), + parameters = + listOf( + SetupParameter(SetupParameter.KEY_ROLE, byteArrayOf(0x03)), + SetupParameter(SetupParameter.KEY_PATH, "/moq".encodeToByteArray()), + ), + ) + val encoded = MoqCodec.encode(msg) + val decoded = MoqCodec.decode(encoded)!! + val roundTripped = assertIs(decoded.message) + assertEquals(msg.supportedVersions, roundTripped.supportedVersions) + assertEquals(msg.parameters, roundTripped.parameters) + } + + @Test + fun server_setup_round_trip() { + val msg = + ServerSetup( + selectedVersion = MoqVersion.DRAFT_17, + parameters = listOf(SetupParameter(SetupParameter.KEY_MAX_SUBSCRIBE_ID, byteArrayOf(0x40, 0x10))), + ) + val encoded = MoqCodec.encode(msg) + val decoded = MoqCodec.decode(encoded)!! + assertEquals(msg, decoded.message) + } + + @Test + fun encode_prefixes_with_type_and_length() { + val encoded = MoqCodec.encode(ServerSetup(selectedVersion = 1L)) + // ServerSetup's type code is 0x41 (65). That's > 63 so it needs a + // 2-byte varint: 0x40 0x41. Length of the payload (2 bytes: one byte + // for varint(version=1), one byte for varint(0 params)) fits in a + // single-byte varint → 0x02. Payload: 0x01, 0x00. Total 5 bytes. + assertContentEquals(byteArrayOf(0x40, 0x41, 0x02, 0x01, 0x00), encoded) + } + + @Test + fun decode_returns_null_when_frame_is_truncated() { + val full = MoqCodec.encode(ClientSetup(listOf(MoqVersion.DRAFT_17))) + // Lose the last byte → caller should buffer more and retry. + assertNull(MoqCodec.decode(full.copyOf(full.size - 1))) + } + + @Test + fun decode_with_offset_reads_from_middle_of_buffer() { + val a = MoqCodec.encode(ClientSetup(listOf(MoqVersion.DRAFT_17))) + val b = MoqCodec.encode(ServerSetup(selectedVersion = MoqVersion.DRAFT_17)) + val concatenated = a + b + + val first = MoqCodec.decode(concatenated, offset = 0)!! + assertEquals(a.size, first.bytesConsumed) + assertIs(first.message) + + val second = MoqCodec.decode(concatenated, offset = first.bytesConsumed)!! + assertEquals(b.size, second.bytesConsumed) + assertIs(second.message) + } + + @Test + fun unknown_message_type_is_rejected() { + // Craft a frame with message type 0xFFFF (large varint), length 0, no payload. + val bogus = MoqWriter() + bogus.writeVarint(0xFFFFL) + bogus.writeVarint(0L) + assertFailsWith { MoqCodec.decode(bogus.toByteArray()) } + } + + @Test + fun trailing_bytes_in_payload_are_rejected() { + // Valid ServerSetup payload of (selectedVersion=1, 0 params) fits in 2 bytes + // (varint 0x01 + varint 0x00). Craft a frame that declares a 4-byte + // payload and pads with 2 junk bytes inside the declared window so the + // decoder sees extra data after the last field. + val w = MoqWriter() + w.writeVarint(MoqMessageType.ServerSetup.code) + w.writeVarint(4L) + w.writeVarint(1L) // selected version (1 byte) + w.writeVarint(0L) // 0 parameters (1 byte) + w.writeByte(0xAA) // trailing junk inside declared payload + w.writeByte(0xBB) + assertFailsWith { MoqCodec.decode(w.toByteArray()) } + } + + @Test + fun parameters_with_zero_length_value_are_allowed() { + val msg = ClientSetup(listOf(1L), parameters = listOf(SetupParameter(0x10L, ByteArray(0)))) + val decoded = MoqCodec.decode(MoqCodec.encode(msg))!! + assertEquals(msg, decoded.message) + } +} diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqSessionTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqSessionTest.kt new file mode 100644 index 000000000..353e7a60d --- /dev/null +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqSessionTest.kt @@ -0,0 +1,117 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.nestsclient.moq + +import com.vitorpamplona.nestsclient.transport.FakeWebTransport +import kotlinx.coroutines.async +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.test.runTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith + +class MoqSessionTest { + @Test + fun client_and_server_complete_setup_handshake_over_fake_transport() = + runTest { + val (clientSide, serverSide) = FakeWebTransport.pair() + + val clientJob = + async { + val session = MoqSession.client(clientSide) + session.setup(listOf(MoqVersion.DRAFT_17)) + session.selectedVersion + } + + val serverJob = + async { + val control = serverSide.peerOpenedBidiStreams().first() + val session = MoqSession.server(serverSide, control) + session.setup( + supportedVersions = listOf(MoqVersion.DRAFT_17, MoqVersion.DRAFT_11), + clientParameters = + listOf( + SetupParameter( + SetupParameter.KEY_MAX_SUBSCRIBE_ID, + byteArrayOf(0x10), + ), + ), + ) + session.selectedVersion + } + + assertEquals(MoqVersion.DRAFT_17, clientJob.await()) + assertEquals(MoqVersion.DRAFT_17, serverJob.await()) + } + + @Test + fun server_picks_first_mutually_supported_version_from_its_own_list() = + runTest { + val (clientSide, serverSide) = FakeWebTransport.pair() + + val clientJob = + async { + val session = MoqSession.client(clientSide) + // Client prefers 17, falls back to 11. + session.setup(listOf(MoqVersion.DRAFT_17, MoqVersion.DRAFT_11)) + session.selectedVersion + } + + val serverJob = + async { + val control = serverSide.peerOpenedBidiStreams().first() + val session = MoqSession.server(serverSide, control) + // Server only speaks 11 → overlap is 11. + session.setup(supportedVersions = listOf(MoqVersion.DRAFT_11)) + session.selectedVersion + } + + assertEquals(MoqVersion.DRAFT_11, clientJob.await()) + assertEquals(MoqVersion.DRAFT_11, serverJob.await()) + } + + @Test + fun server_rejects_when_no_version_overlap() = + runTest { + val (clientSide, serverSide) = FakeWebTransport.pair() + + val clientJob = + async { + val session = MoqSession.client(clientSide) + runCatching { session.setup(listOf(MoqVersion.DRAFT_17)) } + } + + val serverJob = + async { + val control = serverSide.peerOpenedBidiStreams().first() + val session = MoqSession.server(serverSide, control) + assertFailsWith { + session.setup(supportedVersions = listOf(MoqVersion.DRAFT_11)) + } + } + + serverJob.await() + // Client's result: its control stream closes with no reply, which throws. + // We just verify that some throwable propagated. + val clientOutcome = clientJob.await() + assert(clientOutcome.isFailure) { "client setup should have failed, got: $clientOutcome" } + } +} diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/VarintTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/VarintTest.kt new file mode 100644 index 000000000..e92d0fdcd --- /dev/null +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/VarintTest.kt @@ -0,0 +1,114 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.nestsclient.moq + +import kotlin.test.Test +import kotlin.test.assertContentEquals +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertNull + +class VarintTest { + /** + * RFC 9000 §A.1 sample encodings, which every QUIC varint implementation + * must reproduce bit-for-bit. + */ + @Test + fun rfc9000_sample_151288809941952652() { + // 62-bit: 151288809941952652 == 0x2136 0x0000 0000 8c4c (encoded) + val value = 151288809941952652L + val encoded = Varint.encode(value) + assertContentEquals( + byteArrayOf(0xC2.toByte(), 0x19, 0x7C, 0x5E, 0xFF.toByte(), 0x14, 0xE8.toByte(), 0x8C.toByte()), + encoded, + ) + assertEquals(value, Varint.decode(encoded)!!.value) + } + + @Test + fun rfc9000_sample_494878333() { + val value = 494878333L + val encoded = Varint.encode(value) + assertContentEquals( + byteArrayOf(0x9D.toByte(), 0x7F.toByte(), 0x3E, 0x7D.toByte()), + encoded, + ) + assertEquals(value, Varint.decode(encoded)!!.value) + } + + @Test + fun rfc9000_sample_15293() { + val value = 15293L + val encoded = Varint.encode(value) + assertContentEquals(byteArrayOf(0x7B.toByte(), 0xBD.toByte()), encoded) + assertEquals(value, Varint.decode(encoded)!!.value) + } + + @Test + fun rfc9000_sample_37() { + val encoded = Varint.encode(37L) + assertContentEquals(byteArrayOf(0x25), encoded) + assertEquals(37L, Varint.decode(encoded)!!.value) + } + + @Test + fun boundary_values_round_trip() { + for (v in listOf(0L, 63L, 64L, 16_383L, 16_384L, 1_073_741_823L, 1_073_741_824L, Varint.MAX_VALUE)) { + val encoded = Varint.encode(v) + assertEquals( + v, + Varint.decode(encoded)!!.value, + "round-trip for $v", + ) + } + } + + @Test + fun size_matches_encoding_length() { + for (v in listOf(0L, 63L, 64L, 16_383L, 16_384L, 1_073_741_823L, 1_073_741_824L, Varint.MAX_VALUE)) { + assertEquals(Varint.size(v), Varint.encode(v).size) + } + } + + @Test + fun negative_value_is_rejected() { + assertFailsWith { Varint.encode(-1L) } + } + + @Test + fun overflow_value_is_rejected() { + assertFailsWith { Varint.encode(Varint.MAX_VALUE + 1) } + } + + @Test + fun short_buffer_returns_null_so_caller_can_buffer_more() { + assertNull(Varint.decode(ByteArray(0))) + // 4-byte varint with only 3 bytes available: + assertNull(Varint.decode(byteArrayOf(0x9D.toByte(), 0x7F.toByte(), 0x3E), 0)) + } + + @Test + fun bytesConsumed_reflects_actual_length() { + assertEquals(1, Varint.decode(byteArrayOf(0x25))!!.bytesConsumed) + assertEquals(2, Varint.decode(byteArrayOf(0x7B.toByte(), 0xBD.toByte()))!!.bytesConsumed) + assertEquals(4, Varint.decode(byteArrayOf(0x9D.toByte(), 0x7F.toByte(), 0x3E, 0x7D.toByte()))!!.bytesConsumed) + } +}