diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt index c43b64c17..f55bd953b 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -253,6 +253,18 @@ class QuicConnection( @Volatile internal var previousReceiveProtection: PacketProtection? = null + /** + * RFC 9001 §6.4: an endpoint MUST NOT initiate a subsequent key update + * unless the previous one has been confirmed by the peer. We treat + * "confirmed" as having received any application-level packet whose + * decrypt succeeded under the post-rotation (current) live keys — + * meaning the peer has rolled forward in lockstep. Set true by + * [initiateKeyUpdate], cleared by [QuicConnectionParser] on the + * first short-header packet that AEAD-passes after the rotation. + */ + @Volatile + internal var keyUpdateInProgress: Boolean = false + /** * RFC 9001 §4.10 — 0-RTT send-side packet protection. Installed when * the TLS layer derives `client_early_traffic_secret` (right after @@ -572,6 +584,22 @@ class QuicConnection( com.vitorpamplona.quic.connection.recovery .QuicLossDetection() + /** + * Reusable scratch lists for [QuicConnectionWriter.buildApplicationPacket]. + * The function clears them at entry, fills them, encodes the + * packet, snapshots the [scratchAppTokens] via `.toList()` for + * the [SentPacket] record, and returns. The next drain (single- + * threaded by [streamsLock]) then re-clears and re-uses. + * + * Skipped for [collectHandshakeLevelFrames]: that function returns + * a [com.vitorpamplona.quic.connection.HandshakeLevelContents] + * wrapping the lists, which the caller holds across two + * `buildLongHeaderFromFrames` calls (natural-size + padded + * rebuild). Reusing those would corrupt the rebuild path. + */ + internal val scratchAppFrames: MutableList = mutableListOf() + internal val scratchAppTokens: MutableList = mutableListOf() + /** * RFC 9002 §6.2 Probe Timeout signalling. When the driver loop's * PTO timer fires (no ack-eliciting packet has been ACK'd in @@ -1845,12 +1873,14 @@ class QuicConnection( * outbound packet carries `KEY_PHASE = 1` and the peer is expected * to mirror back in the same phase. * - * RFC 9001 §6.5 says an endpoint MUST NOT initiate a key update - * before the handshake is confirmed (HANDSHAKE_DONE received). The - * caller is responsible for that check; this method just performs - * the rotation. §6.4 also forbids initiating a second update before - * the current one has been confirmed (peer responds in matching - * phase) — same caller contract. + * Spec invariants enforced here (caller no longer responsible): + * - RFC 9001 §6.5: returns false if [handshakeComplete] is not + * yet true. Initiating before handshake confirmation is a + * PROTOCOL_VIOLATION. + * - RFC 9001 §6.4: returns false if a previous rotation is still + * in flight ([keyUpdateInProgress]). The parser clears that + * flag on the first inbound packet that AEAD-decrypts under + * the post-rotation keys, signalling the peer rolled forward. * * Header-protection key is unchanged (RFC 9001 §6.1: HP key is NOT * rotated when keys are updated). @@ -1864,6 +1894,16 @@ class QuicConnection( * and server". */ fun initiateKeyUpdate(): Boolean { + // RFC 9001 §6.5: handshake MUST be confirmed before initiating + // a key update. We use [handshakeComplete] as the proxy — + // application keys are installed and the peer's HANDSHAKE_DONE + // has been processed. + if (!handshakeComplete) return false + // RFC 9001 §6.4: MUST NOT initiate a subsequent rotation until + // the previous one is confirmed. The parser clears this flag + // when it observes an inbound packet that AEAD-decrypts under + // the post-rotation live keys. + if (keyUpdateInProgress) return false val cs = appCipherSuite.takeIf { it != 0 } ?: return false val curRx = appReceiveSecret ?: return false val curTx = appSendSecret ?: return false @@ -1907,6 +1947,7 @@ class QuicConnection( appSendSecret = nextTxSecret currentReceiveKeyPhase = !currentReceiveKeyPhase currentSendKeyPhase = !currentSendKeyPhase + keyUpdateInProgress = true qlogObserver.onKeyUpdated("client", EncryptionLevel.APPLICATION) qlogObserver.onKeyUpdated("server", EncryptionLevel.APPLICATION) return true diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt index 6289747ea..337e6e612 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt @@ -436,6 +436,16 @@ private fun feedShortHeaderPacket( if (rotateOnSuccess != null && parsed.packet.packetNumber > state.pnSpace.largestReceived) { conn.commitKeyUpdate(rotateOnSuccess) } + // RFC 9001 §6.4: clear the in-flight-rotation gate once an inbound + // packet decrypts under the live (post-rotation) keys. That confirms + // the peer has rolled forward in lockstep, so it's safe to permit + // a subsequent [QuicConnection.initiateKeyUpdate]. The `rotateOnSuccess + // == null` branch above is the live-keys path (peer's key_phase bit + // matched our [currentReceiveKeyPhase]); reaching that with a + // successful parse is the confirmation signal. + if (rotateOnSuccess == null && conn.keyUpdateInProgress) { + conn.keyUpdateInProgress = false + } state.pnSpace.observeInbound(parsed.packet.packetNumber, nowMillis) if (conn.qlogObserver !== com.vitorpamplona.quic.observability.QlogObserver.NoOp) { conn.qlogObserver.onPacketReceived( diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt index 1aed5af11..1aeaf05fa 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -74,8 +74,6 @@ fun drainOutbound( conn: QuicConnection, nowMillis: Long, ): ByteArray? { - val parts = mutableListOf() - // Closing — emit a CONNECTION_CLOSE at the highest available level. The // datagram-build paths below MUST satisfy two RFC 9000 constraints we // got wrong before: @@ -601,7 +599,12 @@ private fun buildApplicationPacket( // settled mid-handshake (peer hasn't ACK'd anything yet), so the // walk is a no-op. conn.retireFullyDoneStreamsLocked() - val frames = mutableListOf() + // Reuse per-connection scratch lists (cleared at entry) instead of + // allocating fresh `mutableListOf` / `mutableListOf` + // every drain — single-writer by [streamsLock], so the next drain's + // clear() runs after this drain has already snapshotted the tokens + // via `.toList()` into the SentPacket record. + val frames = conn.scratchAppFrames.also { it.clear() } // Tokens collected in lock-step with [frames]: each retransmittable // frame contributes a [RecoveryToken] so the [SentPacket] recorded // at the bottom of this function can drive RFC 9002 loss @@ -610,7 +613,7 @@ private fun buildApplicationPacket( // are not retransmittable (DatagramFrame, StreamFrame for now) do // not contribute a token; the packet is still ack-eliciting and // tracked for loss-detection timing. - val tokens = mutableListOf() + val tokens = conn.scratchAppTokens.also { it.clear() } // RFC 9000 §17.2.3 — 0-RTT packets MUST NOT contain ACK frames. The // server cannot ACK 0-RTT-level packets because it has no way to @@ -725,7 +728,17 @@ private fun buildApplicationPacket( val sorted = when { active.size <= 1 -> active - active.all { it.priority == 0 } -> active + + // Single-pass uniform-priority check: if EVERY stream + // shares the same priority (whether default 0 or + // anything else), the sort is a no-op and we keep + // insertion order for round-robin fairness. Pre-fix + // we only short-circuited on `priority == 0`, so a + // homogeneous `priority == 7` workload paid for an + // O(N log N) sort despite the result being + // observationally identical to insertion order. + active.all { it.priority == active[0].priority } -> active + else -> active.sortedByDescending { it.priority } } val rotation = conn.streamRoundRobinStart diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/crypto/Aead.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/crypto/Aead.kt index 00d7c39c5..e0671b904 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/crypto/Aead.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/crypto/Aead.kt @@ -86,6 +86,41 @@ abstract class Aead { return seal(key, nonce, a, p) } + /** + * Range + in-place [seal]: read `aad` and `plaintext` from sub-ranges + * and write ciphertext+tag DIRECTLY into [output] at [outputOffset]. + * Returns the number of bytes written (== plaintextLength + tagLength). + * + * Saves another ByteArray allocation per outbound packet vs. + * [sealRange] — the build path can pre-allocate one final packet + * buffer and have the ciphertext land in-place rather than copying + * a fresh `seal()` result. Combined with the AAD+plaintext range + * inputs already handled by [sealRange], a complete outbound packet + * goes from ~4 allocations (headerBytes / paddedPlaintext / + * ciphertext / final concat) down to ~2 (the final packet buffer + * and the AEAD provider's internal scratch). + * + * Default impl falls back to [sealRange] + copy; the JCA-backed + * [com.vitorpamplona.quic.crypto.JcaAesGcmAead] overrides to use + * `Cipher.doFinal(input, inOff, inLen, output, outOff)`. + */ + open fun sealInto( + key: ByteArray, + nonce: ByteArray, + aad: ByteArray, + aadOffset: Int, + aadLength: Int, + plaintext: ByteArray, + plaintextOffset: Int, + plaintextLength: Int, + output: ByteArray, + outputOffset: Int, + ): Int { + val ct = sealRange(key, nonce, aad, aadOffset, aadLength, plaintext, plaintextOffset, plaintextLength) + ct.copyInto(output, outputOffset) + return ct.size + } + /** * Range-based [open] — same semantics as [open] but reads `aad` and * `ciphertext` from sub-ranges. Default impl slices and delegates. diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/LongHeaderPacket.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/LongHeaderPacket.kt index 2b476cbeb..c54d7b36a 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/LongHeaderPacket.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/LongHeaderPacket.kt @@ -109,14 +109,28 @@ object LongHeaderPacket { } val headerBytes = w.toByteArray() - // Encrypt padded payload + // Pre-allocate the final packet buffer in one shot and have the + // AEAD seal write ciphertext+tag directly into it. Pre-fix this + // path allocated `headerBytes`, `paddedPlaintext`, the + // `aead.seal` return ByteArray, and the final concat buffer — + // four ByteArrays per outbound packet. The single-buffer + + // [Aead.sealInto] form below collapses the seal output and + // concat into the same allocation. val nonce = aeadNonce(iv, plain.packetNumber) - val ciphertext = aead.seal(key, nonce, headerBytes, paddedPlaintext) - - // Concatenate header + ciphertext - val packet = ByteArray(headerBytes.size + ciphertext.size) + val packet = ByteArray(headerBytes.size + paddedPlaintext.size + aead.tagLength) headerBytes.copyInto(packet, 0) - ciphertext.copyInto(packet, headerBytes.size) + aead.sealInto( + key = key, + nonce = nonce, + aad = packet, + aadOffset = 0, + aadLength = headerBytes.size, + plaintext = paddedPlaintext, + plaintextOffset = 0, + plaintextLength = paddedPlaintext.size, + output = packet, + outputOffset = headerBytes.size, + ) // Apply header protection. Sample is 16 bytes starting 4 bytes after pnOffset. val sampleStart = pnOffset + 4 diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/ShortHeaderPacket.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/ShortHeaderPacket.kt index 99494a04b..835861bac 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/ShortHeaderPacket.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/ShortHeaderPacket.kt @@ -75,12 +75,26 @@ object ShortHeaderPacket { // Trailing 0x00 bytes decode as PADDING frames (RFC 9000 §19.1). val paddedPlaintext = padForHeaderProtectionSample(plain.payload, pnLen) + // Same single-buffer + sealInto pattern as LongHeaderPacket.build: + // pre-allocate the final packet and have the AEAD write + // ciphertext+tag directly into it instead of allocating a fresh + // `seal()` return + a concat buffer. Saves 2 ByteArrays per + // outbound short-header packet. val nonce = aeadNonce(iv, plain.packetNumber) - val ciphertext = aead.seal(key, nonce, headerBytes, paddedPlaintext) - - val packet = ByteArray(headerBytes.size + ciphertext.size) + val packet = ByteArray(headerBytes.size + paddedPlaintext.size + aead.tagLength) headerBytes.copyInto(packet, 0) - ciphertext.copyInto(packet, headerBytes.size) + aead.sealInto( + key = key, + nonce = nonce, + aad = packet, + aadOffset = 0, + aadLength = headerBytes.size, + plaintext = paddedPlaintext, + plaintextOffset = 0, + plaintextLength = paddedPlaintext.size, + output = packet, + outputOffset = headerBytes.size, + ) val sampleStart = pnOffset + 4 require(sampleStart + 16 <= packet.size) { "packet too short for HP sample" } diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt index f060ef437..cfe2988b6 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt @@ -22,7 +22,7 @@ package com.vitorpamplona.quic.stream import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.flow.Flow -import kotlinx.coroutines.flow.consumeAsFlow +import kotlinx.coroutines.flow.flow /** * One QUIC stream (bidirectional or unidirectional). Application code @@ -78,33 +78,35 @@ class QuicStream( * was discarded, leaving a hole in the stream that the application could * never know about. * - * **Single-collector contract.** [consumeAsFlow] cancels the underlying - * [Channel] when its collector terminates (either by exception or - * explicit cancellation). After that point any subsequent - * `trySend` from the parser fails, sets [overflowed] = true, and - * the parser tears the whole connection down with - * `INTERNAL_ERROR: stream … consumer overflowed`. So: - * - Application code MUST collect [incoming] **at most once** per - * stream and MUST hold the collect open until the stream - * terminates (FIN, peer RESET_STREAM, or local - * [stopSending] / [resetStream] decision). - * - Cancelling the collector early — e.g. wrapping in - * `withTimeout(...)` — is equivalent to telling the connection - * to drop. If the application wants to stop receiving without - * killing the connection, call [stopSending] FIRST, then let - * the parser's RESET_STREAM-style teardown drain the channel - * cleanly. + * **Resilience to collector cancellation.** Pre-fix this exposed + * [incomingChannel] via `consumeAsFlow()`, which cancels the + * underlying [Channel] when its collector terminates. That coupled + * "application stopped collecting" to "parser INTERNAL_ERRORs the + * whole connection on next delivery" — cancelling the collector + * early (e.g. `withTimeout(...)`) was effectively a request to + * drop the connection. Rebuilt as `flow { for (c in + * incomingChannel) emit(c) }`: a fresh emit-only Flow over the + * channel iterator, with no `consume`-style ownership transfer. + * Now collector cancellation just exits the flow without touching + * the channel; the channel stays open, the parser keeps + * delivering, and the application can re-collect later (a fresh + * collector picks up at `channel.receive()` — i.e. from the + * current head, not the start). * - * This is a fragile coupling we accept for now because (a) every - * production caller already follows the single-collector pattern, - * and (b) widening the contract would require swapping - * `consumeAsFlow` for a `MutableSharedFlow` with replay/buffer - * semantics, which has its own back-pressure pitfalls. The - * comment is here so a future refactor doesn't accidentally - * loosen the contract. + * Producer back-pressure unchanged: the channel buffer is still + * 64 chunks, [trySend] still sets [overflowed] on saturation, + * and the parser still closes the connection if the application + * fails to drain. The looser contract just permits the previously- + * disallowed pattern of "stop reading temporarily, then resume". + * + * Concurrent collectors are still NOT supported — two simultaneous + * collects would race the channel iterator and each chunk goes to + * exactly one of them non-deterministically. Sequential collects + * (one finishes / cancels, then another starts) are the new + * supported pattern. */ private val incomingChannel = Channel(capacity = 64) - val incoming: Flow get() = incomingChannel.consumeAsFlow() + val incoming: Flow = flow { for (chunk in incomingChannel) emit(chunk) } /** * True once a [deliverIncoming] call failed because the channel was diff --git a/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaAesGcmAead.kt b/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaAesGcmAead.kt index d974d9e4a..e6ddc533d 100644 --- a/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaAesGcmAead.kt +++ b/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaAesGcmAead.kt @@ -206,6 +206,56 @@ class JcaAesGcmAead( } } + /** + * In-place range [seal]. Same semantics as the parent's [sealInto], + * but takes the JCA's native `Cipher.doFinal(input, inOff, inLen, + * output, outOff)` path so ciphertext+tag land directly in + * [output] with no intermediate allocation. Caller is expected to + * have sized `output` large enough to hold + * `plaintextLength + tagLength` bytes starting at `outputOffset`. + */ + override fun sealInto( + key: ByteArray, + nonce: ByteArray, + aad: ByteArray, + aadOffset: Int, + aadLength: Int, + plaintext: ByteArray, + plaintextOffset: Int, + plaintextLength: Int, + output: ByteArray, + outputOffset: Int, + ): Int = + synchronized(this) { + val reuse = recentEncryptNonces.any { it.contentEquals(nonce) } + val cipher: Cipher + if (reuse) { + cipher = Cipher.getInstance("AES/GCM/NoPadding") + cipher.init(Cipher.ENCRYPT_MODE, keySpec, GCMParameterSpec(128, nonce)) + } else { + cipher = encryptCipher + try { + cipher.init(Cipher.ENCRYPT_MODE, keySpec, GCMParameterSpec(128, nonce)) + } catch (t: Throwable) { + if (recentEncryptNonces.lastOrNull()?.contentEquals(nonce) == true) { + recentEncryptNonces.removeLast() + } + throw t + } + } + try { + cipher.updateAAD(aad, aadOffset, aadLength) + val written = cipher.doFinal(plaintext, plaintextOffset, plaintextLength, output, outputOffset) + if (!reuse) rememberEncryptNonce(nonce) + written + } catch (t: Throwable) { + if (!reuse && recentEncryptNonces.lastOrNull()?.contentEquals(nonce) == true) { + recentEncryptNonces.removeLast() + } + throw t + } + } + private companion object { private const val NONCE_HISTORY_LIMIT = 8 }