diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/Buffer.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/Buffer.kt index 58c8f1aa53..1d94428fd6 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/Buffer.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/Buffer.kt @@ -203,6 +203,9 @@ class QuicReader( } fun skip(n: Int) { + if (n < 0) { + throw QuicCodecException("skip with negative count: $n") + } require(n) pos += n } @@ -248,6 +251,14 @@ class QuicReader( } fun readBytes(n: Int): ByteArray { + // Translate a negative `n` (e.g. an attacker-controlled length + // that wrapped through `.toInt()`) into a typed + // [QuicCodecException] instead of `IllegalArgumentException` + // from `copyOfRange`. Bounds-checking via [require] still kicks + // in for positive `n` past `end`. + if (n < 0) { + throw QuicCodecException("readBytes with negative count: $n") + } require(n) val out = src.copyOfRange(pos, pos + n) pos += n @@ -280,3 +291,17 @@ class QuicCodecException( message: String, cause: Throwable? = null, ) : RuntimeException(message, cause) + +/** + * Peer protocol violation that mandates connection close per RFC 9000 / 9001. + * Distinct from [QuicCodecException] (which is also "drop the packet" for + * AEAD-failed inputs) — a [QuicProtocolViolationException] means the peer + * sent something well-formed enough to AEAD-decrypt but inconsistent with + * the wire spec, so the connection MUST be closed with PROTOCOL_VIOLATION. + * + * Typical sources: reserved-bit-set in the unmasked QUIC header + * (RFC 9000 §17.2 / §17.3.1). + */ +class QuicProtocolViolationException( + message: String, +) : RuntimeException(message) diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/PathValidator.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/PathValidator.kt index 8ed2a7289d..eb6534bbf9 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/PathValidator.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/PathValidator.kt @@ -430,7 +430,15 @@ class PathValidator( currentPtoMillis: Long, ): PathMigrationResult { if (state is PathValidationState.Validating) return PathMigrationResult.AlreadyInProgress - val (seq, entry) = unusedCids.entries.firstOrNull() ?: return PathMigrationResult.NoSpareCid + // Pick the SMALLEST sequence number, not insertion order. The + // peer is allowed (RFC 9000 §19.15) to issue NEW_CONNECTION_ID + // out of sequence — e.g. retransmits arriving after newer + // higher-seq offers. LinkedHashMap iteration is by insertion, + // so the prior `firstOrNull` would pick whichever offer landed + // first, not the lowest seq. Picking the smallest preserves + // RFC's expected ordering (lower seq retired first) and lines + // up with what other clients (quicly, neqo) do. + val (seq, entry) = unusedCids.entries.minByOrNull { it.key } ?: return PathMigrationResult.NoSpareCid unusedCids.remove(seq) val payload = challengePayloadFactory().also { @@ -580,7 +588,16 @@ class PathValidator( */ fun forceRotateToHigherSequence(): ForcedRotationResult? { if (activeCidSequence >= retirePriorToWatermark) return null - val (seq, entry) = unusedCids.entries.firstOrNull() ?: return ForcedRotationResult.NoSpareCid + // Pick the smallest seq above the watermark — see + // [tryStartValidation]'s rationale. LinkedHashMap is insertion- + // ordered, not seq-ordered, so a peer that retransmits an old + // offer can shift the "first" entry away from the actual + // smallest. + val (seq, entry) = + unusedCids.entries + .filter { it.key >= retirePriorToWatermark } + .minByOrNull { it.key } + ?: return ForcedRotationResult.NoSpareCid unusedCids.remove(seq) val priorSeq = activeCidSequence queueRetireSequence(priorSeq) 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 3fbeff6661..457cfef982 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -69,11 +69,19 @@ class QuicConnection( * of letting null silently disable MITM protection. */ val tlsCertificateValidator: com.vitorpamplona.quic.tls.CertificateValidator, - val nowMillis: () -> Long = { - kotlin.time.Clock.System - .now() - .toEpochMilliseconds() - }, + /** + * Monotonic clock used by ACK-delay encoding, RTT samples, PTO scheduling, + * loss detection, and path-validation timeouts. Default uses + * [kotlin.time.TimeSource.Monotonic] so an NTP step / suspend-resume + * doesn't poison RTT estimates or trigger spurious losses. The returned + * value is "milliseconds since this connection was constructed" — only + * differences are meaningful. + * + * Tests inject a virtual clock; production callers should leave the + * default unless they have a specific reason (e.g. recording wallclock + * timestamps in qlog) and understand the wallclock pitfalls. + */ + val nowMillis: () -> Long = defaultMonotonicNowMillis(), val alpnList: List = listOf(TlsConstants.ALPN_H3), /** * Optional second listener invoked after the connection's own @@ -283,13 +291,27 @@ class QuicConnection( /** * Lock-split refactor (2026-05-08): @Volatile so concurrent loops can * read the status without a lock — coarse "are we still alive?" checks. - * Mutating transitions still go through [lifecycleLock] for atomicity + * Mutating transitions still go through [closeStateMonitor] for atomicity * with [closeReason]/[closeErrorCode] updates. */ @Volatile var status: Status = Status.HANDSHAKING internal set + /** + * Non-suspend monitor protecting the atomic transition of + * [status] / [closeReason] / [closeErrorCode] from a non-CLOSED to a + * CLOSED/CLOSING state. We intentionally do NOT use [lifecycleLock] + * for this — that's a `kotlinx.coroutines.sync.Mutex` which only + * works from suspend contexts, and [markClosedExternally] is invoked + * from the parser inside a `streamsLock.withLock { }` block where + * suspending again on a different mutex is awkward and risks lock + * inversion. A plain monitor avoids both problems while still + * giving us a true compare-and-set for "first caller wins the + * close-reason and gets to fire the qlog event". + */ + private val closeStateMonitor = Any() + /** App-level error code for graceful close. */ var closeReason: String? = null private set @@ -299,27 +321,36 @@ class QuicConnection( private val streams = mutableMapOf() /** - * Round-4 perf #10: parallel insertion-ordered list of streams so the - * writer's round-robin scan can index by position without - * `streams.entries.toList()` allocating per drain. Mutated in lockstep - * with [streams] under [streamsLock]: + * Per-connection insertion-ordered list of live streams. Mutated only + * under [streamsLock] (single-writer): * - [openBidiStreamLocked] / [openUniStreamLocked] / * [getOrCreatePeerStreamLocked] append. * - [retireFullyDoneStreamsLocked] removes entries whose stream has - * flipped [QuicStream.isFullyRetired] = true. The retire pass - * folds the receive-side high-water mark into - * [retiredStreamsRecvBytes] so the writer's MAX_DATA accounting - * in `appendFlowControlUpdates` doesn't regress when a retired - * stream's `receive.contiguousEnd()` drops out of the iteration. + * flipped [QuicStream.isFullyRetired] = true, folding the + * receive-side high-water mark into [retiredStreamsRecvBytes] so + * the writer's MAX_DATA accounting in `appendFlowControlUpdates` + * doesn't regress when a retired stream's `receive.contiguousEnd()` + * drops out of the iteration. + * + * Read by both lock-holding paths (writer drain, parser dispatch) and + * by lock-free observers ([closeAllSignals] after teardown). To make + * lock-free reads safe we use an immutable snapshot pattern: every + * mutation publishes a fresh `List` via the `@Volatile` reference, + * and readers see either the pre- or post-mutation snapshot — never + * a half-modified ArrayList that can raise + * `ConcurrentModificationException`. The cost is one shallow copy + * per add / retire; steady-state churn stays under ~100/sec even at + * peak audio-room load (see + * `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`, + * which pegs ~50 streams/sec at peak). * * Without retirement, the moq-lite audio-rooms path leaks one - * QuicStream per Opus frame for the lifetime of the session — the - * stream-cliff investigation in - * `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md` - * pegs steady-state churn at ~50 streams/sec, so a 3-hour room - * accumulates ~540 000 entries before retirement was wired. + * QuicStream per Opus frame for the lifetime of the session — a + * 3-hour room accumulates ~540 000 entries before retirement was + * wired. */ - private val streamsList = mutableListOf() + @Volatile + private var streamsList: List = emptyList() private var nextLocalBidiIndex: Long = 0L private var nextLocalUniIndex: Long = 0L @@ -926,6 +957,15 @@ class QuicConnection( originalPacketBytes: ByteArray, ): Boolean { if (retryConsumed) return false + // RFC 9000 §17.2.5.2: "the client MUST discard a Retry packet + // that contains a Source Connection ID field that is identical + // to the Destination Connection ID field of its Initial packet". + // Without this guard a self-loop / off-path attacker could feed + // us a Retry that nominally validates but pins our state to the + // attacker's chosen DCID forever. + if (retryPacket.scid.bytes.contentEquals(originalDestinationConnectionId.bytes)) { + return false + } if (!retryPacket.verifyIntegrityTag(originalPacketBytes, originalDestinationConnectionId.bytes)) { return false } @@ -1184,7 +1224,7 @@ class QuicConnection( stream.sendCredit = peerTransportParameters?.initialMaxStreamDataBidiRemote ?: config.initialMaxStreamDataBidiRemote stream.receiveLimit = config.initialMaxStreamDataBidiLocal streams[id] = stream - streamsList += stream + streamsList = streamsList + stream return stream } @@ -1218,7 +1258,7 @@ class QuicConnection( stream.sendCredit = peerTransportParameters?.initialMaxStreamDataUni ?: config.initialMaxStreamDataUni stream.receiveLimit = 0L // can't receive streams[id] = stream - streamsList += stream + streamsList = streamsList + stream return stream } @@ -1412,15 +1452,21 @@ class QuicConnection( errorCode: Long, reason: String, ) { - var firedQlog = false - lifecycleLock.withLock { - if (status == Status.CLOSED || status == Status.CLOSING) return@withLock - closeErrorCode = errorCode - closeReason = reason - status = Status.CLOSING - firedQlog = true - } - if (firedQlog) qlogObserver.onConnectionClosed("local", errorCode, reason) + // Atomic CAS via [closeStateMonitor] so two concurrent + // close()/markClosedExternally callers can't both observe a + // non-CLOSED state and both proceed to fire the qlog event / + // overwrite [closeReason]. First caller wins; subsequent + // callers no-op silently. + val firedQlog = + synchronized(closeStateMonitor) { + if (status == Status.CLOSED || status == Status.CLOSING) return@synchronized false + closeErrorCode = errorCode + closeReason = reason + status = Status.CLOSING + true + } + if (!firedQlog) return + qlogObserver.onConnectionClosed("local", errorCode, reason) // If a caller is suspended on awaitHandshake() and we're tearing down // before completion, fail the deferred so the caller throws instead // of hanging forever. @@ -1432,17 +1478,23 @@ class QuicConnection( /** Called by the parser on inbound CONNECTION_CLOSE or by the driver on read-loop death. */ internal fun markClosedExternally(reason: String) { - val wasClosed = status == Status.CLOSED - if (status != Status.CLOSED) status = Status.CLOSED - if (!wasClosed) { - // First-call wins for [closeReason] so the highest-quality - // diagnostic is preserved when several teardown paths race - // (e.g. read loop's `socket.receive() == null` finally fires - // a moment before the send loop's `socket.send` throw catch - // block does). Without this, downstream observers like - // `ReconnectingNestsListener.terminalAwait` see a closed - // connection but no human-readable cause for the failure. - closeReason = reason + // First-call wins for [closeReason] so the highest-quality + // diagnostic is preserved when several teardown paths race + // (e.g. read loop's `socket.receive() == null` finally fires + // a moment before the send loop's `socket.send` throw catch + // block does). Pre-fix, the "did we win the race?" check was a + // non-atomic read-then-write on [status], so two callers could + // both observe `status != CLOSED`, both fire the qlog event, + // and both stomp on [closeReason] in unpredictable order. The + // monitor below makes the transition truly atomic. + val firstClose = + synchronized(closeStateMonitor) { + if (status == Status.CLOSED) return@synchronized false + status = Status.CLOSED + closeReason = reason + true + } + if (firstClose) { // "remote" covers both peer-initiated CONNECTION_CLOSE and // local invariant violations (CID mismatch, frame decode // failure) that the parser surfaces as markClosedExternally. @@ -1480,7 +1532,12 @@ class QuicConnection( closedSignal.close() peerStreamSignal.close() incomingDatagramSignal.close() - // Iterate the snapshot list (safe: we never remove from it). + // [streamsList] is now an immutable List published via @Volatile, + // so reading it here without [streamsLock] yields a consistent + // snapshot — either the pre-mutation or post-mutation view — + // and never a half-mutated ArrayList raising CME. Mutators + // (open*Locked, retireFullyDoneStreamsLocked) all run under + // [streamsLock] and publish a fresh List on each change. // closeIncoming is idempotent on the underlying Channel.close(). for (stream in streamsList) { stream.closeIncoming() @@ -1527,7 +1584,7 @@ class QuicConnection( StreamId.Kind.CLIENT_BIDI -> config.initialMaxStreamDataBidiLocal } streams[id] = stream - streamsList += stream + streamsList = streamsList + stream newPeerStreams.addLast(stream) // Track lifetime peer-stream counts so the writer can emit a // refreshed MAX_STREAMS_* once the peer's usage approaches the @@ -1882,12 +1939,20 @@ class QuicConnection( * stream that delivers duplicate bytes to the application. */ internal fun retireFullyDoneStreamsLocked(): Int { - if (streamsList.isEmpty()) return 0 + val current = streamsList + if (current.isEmpty()) return 0 + // Build the post-retire snapshot in one pass. Mutating the live + // [streamsList] in-place would force readers (closeAllSignals, + // diagnostic accessors) to take [streamsLock] just to iterate; + // building a fresh list and publishing it via the @Volatile ref + // makes those reads lock-free and CME-free. var removed = 0 - val it = streamsList.iterator() - while (it.hasNext()) { - val stream = it.next() - if (!stream.isFullyRetired) continue + val kept = ArrayList(current.size) + for (stream in current) { + if (!stream.isFullyRetired) { + kept.add(stream) + continue + } // Fold the per-stream receive high-water into the cumulative // counter BEFORE we drop it — once we lose the reference the // writer can no longer reconstruct the contribution. @@ -1905,10 +1970,10 @@ class QuicConnection( // peer's plausible retransmit horizon. recordRetiredStreamIdLocked(stream.streamId) streams.remove(stream.streamId) - it.remove() removed++ } if (removed > 0) { + streamsList = kept retiredStreamsCount += removed.toLong() // The writer's round-robin cursor is a position in // [streamsList], so a removal that crossed the cursor would @@ -2491,6 +2556,19 @@ class QuicConnection( } } +/** + * Default monotonic clock supplier for [QuicConnection.nowMillis]. Returns a + * lambda that yields milliseconds-elapsed-since-construction, anchored on a + * fresh [kotlin.time.TimeSource.Monotonic.markNow] mark. Only differences are + * meaningful; an NTP step / suspend-resume does not affect the values. + */ +private fun defaultMonotonicNowMillis(): () -> Long { + val anchor = + kotlin.time.TimeSource.Monotonic + .markNow() + return { anchor.elapsedNow().inWholeMilliseconds } +} + /** Connection was closed (locally or by peer) before reaching CONNECTED. */ class QuicConnectionClosedException( message: String, diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt index b2fc47b4b6..8efca138ae 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt @@ -52,8 +52,16 @@ class QuicConnectionDriver( val connection: QuicConnection, private val socket: UdpSocket, private val parentScope: CoroutineScope, - private val nowMillis: () -> Long = { System.currentTimeMillis() }, ) { + /** + * Single time source: defer to the connection's [QuicConnection.nowMillis] + * (monotonic by default). Pre-fix the driver had its own + * `System.currentTimeMillis()` default which silently disagreed with the + * connection's wallclock-derived clock; under NTP step the two could + * report different values for the SAME logical "now", leading to RTT + * samples computed against drifted timestamps. + */ + private val nowMillis: () -> Long get() = connection.nowMillis private val job = SupervisorJob(parentScope.coroutineContext[Job]) private val scope = CoroutineScope(parentScope.coroutineContext + job + Dispatchers.IO) private val sendWakeup = Channel(Channel.CONFLATED) @@ -390,6 +398,13 @@ class QuicConnectionDriver( * (or the fresh one — both are valid) and is at worst a no-op. */ internal suspend fun handlePtoFired(conn: QuicConnection) { + // Increment FIRST so [requeueInflightForProbe]'s threshold check + // sees the post-increment value. The increment must happen exactly + // once per PTO event — not per call to [requeueInflightForProbe], + // because the send loop calls that helper again between the first + // and second probe (RFC 9002 §6.2.4) and that re-requeue is part + // of the SAME PTO event. + conn.consecutivePtoCount = (conn.consecutivePtoCount + 1).coerceAtMost(6) conn.pendingPing = true requeueInflightForProbe(conn) // RFC 9002 §6.2.4: the spec allows up to 2 ack-eliciting packets @@ -400,7 +415,6 @@ internal suspend fun handlePtoFired(conn: QuicConnection) { // nothing requeued yet" — the requeue is what makes the budget // meaningful. conn.pendingProbePackets = 2 - conn.consecutivePtoCount = (conn.consecutivePtoCount + 1).coerceAtMost(6) } /** @@ -426,12 +440,13 @@ internal suspend fun requeueInflightForProbe(conn: QuicConnection) { if (conn.initial.sendProtection != null && !conn.initial.keysDiscarded) { conn.requeueAllInflightCrypto(EncryptionLevel.INITIAL) } - // Bug-6 fix: increment BEFORE the threshold check. With the - // pre-fix ordering and threshold=2, the rotation actually fired - // on the 3rd PTO (count went 0→1→2 before check). Now the count - // matches the constant's natural reading: "after 2 consecutive - // PTOs with no progress, trigger migration on the 2nd PTO firing." - conn.consecutivePtoCount = (conn.consecutivePtoCount + 1).coerceAtMost(6) + // The PTO count is incremented in [handlePtoFired] BEFORE this + // helper runs, so the threshold check below sees the post-increment + // value (matching the constant's natural reading: "after N + // consecutive PTOs with no progress, trigger migration on the Nth + // PTO firing"). The send loop's between-probe re-requeue calls + // this helper without bumping the count again — that's the same + // PTO event, not a new one. // Once 1-RTT keys are installed, PTO must also retransmit application // data — STREAM bytes that were sent but never ACK'd. Without this, // a single corrupted/lost 1-RTT packet (especially the first one 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 f4927721cb..92648006b0 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt @@ -71,6 +71,22 @@ fun feedDatagram( conn: QuicConnection, datagram: ByteArray, nowMillis: Long, +) { + try { + feedDatagramInner(conn, datagram, nowMillis) + } catch (e: com.vitorpamplona.quic.QuicProtocolViolationException) { + // RFC 9000 §17.2 / §17.3.1: peer set reserved bits in the + // header after HP unmask (or a similar invariant violation + // bubbled out of the parse path). Spec MUST close with + // PROTOCOL_VIOLATION. + conn.markClosedExternally(e.message ?: "PROTOCOL_VIOLATION") + } +} + +private fun feedDatagramInner( + conn: QuicConnection, + datagram: ByteArray, + nowMillis: Long, ) { var offset = 0 while (offset < datagram.size) { @@ -408,7 +424,16 @@ private fun feedShortHeaderPacket( // to [previousReceiveProtection], and rolls the send side forward so // the next outbound carries the matching KEY_PHASE bit (peer uses // that to confirm the rotation completed). - if (rotateOnSuccess != null) { + // + // RFC 9001 §6.1: a peer that initiated a key update MUST NOT send a + // post-rotation packet with a smaller PN than any pre-rotation + // packet. So a "next-phase" packet with PN <= largestReceived is + // either a misbehaving peer or an off-path attempt to flip our key + // state with a captured/forged packet. We refuse the commit in that + // case — AEAD already cleared, so the frames are still delivered, + // but we don't promote next-phase keys to current. Aligns with + // quicly / quiche / s2n-quic behaviour. + if (rotateOnSuccess != null && parsed.packet.packetNumber > state.pnSpace.largestReceived) { conn.commitKeyUpdate(rotateOnSuccess) } state.pnSpace.observeInbound(parsed.packet.packetNumber, nowMillis) @@ -499,7 +524,19 @@ private fun dispatchFrames( state.largestAckedSentTimeMs = largestSentTime } if (advancedLargest && largestSentTime != null && drained.any { it.ackEliciting }) { - val ackDelayUs = frame.ackDelay shl conn.config.ackDelayExponent.toInt() + // Peer-controlled `frame.ackDelay` is a varint up to 2^62-1. + // Naive `shl ackDelayExponent` overflows Long for crafted + // values (negative result → poisoned RTT sample inflating + // smoothed_rtt). Clamp the exponent and the shift result + // before consuming. Per RFC 9000 §18.2 the exponent is + // 0..20; we further clamp ackDelay so the shift never + // overflows (`ackDelay <= Long.MAX_VALUE >>> exponent`). + val rawExponent = conn.config.ackDelayExponent + val exponent = rawExponent.coerceIn(0L, 20L).toInt() + val maxAckDelayPreShift = + if (exponent == 0) Long.MAX_VALUE else Long.MAX_VALUE ushr exponent + val safeAckDelay = frame.ackDelay.coerceIn(0L, maxAckDelayPreShift) + val ackDelayUs = safeAckDelay shl exponent val ackDelayMs = ackDelayUs / 1_000L conn.lossDetection.onRttSample( largestAckedSentTimeMs = largestSentTime, @@ -602,7 +639,33 @@ private fun dispatchFrames( ) return } - stream.receive.insert(frame.offset, frame.data, frame.fin) + // RFC 9000 §4.5: enforce final-size invariants. The + // [com.vitorpamplona.quic.stream.ReceiveBuffer.insert] + // surface returns a typed result; map any non-OK result + // to a connection close with FINAL_SIZE_ERROR so the + // peer knows it just violated the spec instead of + // having its bytes silently dropped. + when (stream.receive.insert(frame.offset, frame.data, frame.fin)) { + com.vitorpamplona.quic.stream.ReceiveBuffer.InsertResult.OK -> { + Unit + } + + com.vitorpamplona.quic.stream.ReceiveBuffer.InsertResult.OFFSET_PAST_FIN -> { + conn.markClosedExternally( + "FINAL_SIZE_ERROR: stream ${frame.streamId} frame ends at " + + "${frame.offset + frame.data.size} past final size ${stream.receive.finOffset}", + ) + return + } + + com.vitorpamplona.quic.stream.ReceiveBuffer.InsertResult.FIN_CONFLICTS_WITH_PRIOR_FIN -> { + conn.markClosedExternally( + "FINAL_SIZE_ERROR: stream ${frame.streamId} second FIN at " + + "${frame.offset + frame.data.size} disagrees with prior final size ${stream.receive.finOffset}", + ) + return + } + } val data = stream.receive.readContiguous() if (data.isNotEmpty()) { // Round-4 perf #9: mark the stream as needing a flow- @@ -698,17 +761,61 @@ private fun dispatchFrames( ) return } + // RFC 9000 §4.5: the [finalSize] in RESET_STREAM MUST agree + // with any final size implied by previously-received STREAM + // frames AND MUST be ≥ the highest offset already observed. + // A peer that violates this is closed with FINAL_SIZE_ERROR + // — pre-fix we accepted any value silently, letting a buggy + // peer drift our state. + val target = conn.streamByIdLocked(frame.streamId) + if (target != null) { + val priorFin = target.receive.finOffset + val highestSeen = target.receive.highestObservedOffset() + if (priorFin != null && frame.finalSize != priorFin) { + conn.markClosedExternally( + "FINAL_SIZE_ERROR: stream ${frame.streamId} RESET_STREAM finalSize " + + "${frame.finalSize} disagrees with prior FIN size $priorFin", + ) + return + } + if (frame.finalSize < highestSeen) { + conn.markClosedExternally( + "FINAL_SIZE_ERROR: stream ${frame.streamId} RESET_STREAM finalSize " + + "${frame.finalSize} below already-observed offset $highestSeen", + ) + return + } + } // Mark the peer's stream aborted and close our read side; the // application sees a truncated incoming flow. - conn.streamByIdLocked(frame.streamId)?.closeIncoming() + target?.closeIncoming() } is StopSendingFrame -> { - // Round-4 #2: peer asks us to stop sending on its read side. - // We don't model an outbound abort yet — this is acknowledged - // and dropped. A future enhancement should emit RESET_STREAM - // back per RFC 9000 §3.5. + // RFC 9000 §3.5: peer asks us to stop sending on its read side. + // We MUST respond with RESET_STREAM carrying the same + // application error code so the peer can free its receive + // resources. Pre-fix this was silently dropped, so the peer + // would keep buffering bytes we kept emitting and the + // application kept paying CPU on a stream the peer no + // longer cared about. + // + // resetStream() is "first-call wins" so a duplicate + // STOP_SENDING (peer retransmit) doesn't redundantly mutate + // state. Only triggers for streams with an outgoing side — + // peer-uni (CLIENT_UNI from peer's perspective = SERVER_UNI + // here) has none, so the call is a defensive no-op. ackEliciting = true + conn.streamByIdLocked(frame.streamId)?.let { stream -> + val kind = StreamId.kindOf(frame.streamId) + val hasLocalSend = + kind == StreamId.Kind.CLIENT_BIDI || + kind == StreamId.Kind.SERVER_BIDI || + kind == StreamId.Kind.CLIENT_UNI + if (hasLocalSend) { + stream.resetStream(frame.applicationErrorCode) + } + } } is NewTokenFrame -> { 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 bc27225abb..689a630e44 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -762,8 +762,15 @@ private fun buildApplicationPacket( fin = chunk.fin, ) packetBudget -= chunk.data.size + 32 - connBudget -= chunk.data.size - conn.sendConnectionFlowConsumed += chunk.data.size + // RFC 9000 §4.1: connection-level flow credit caps + // cumulative *new* bytes only — retransmits MUST + // NOT debit further. Pre-fix every retransmit + // chunk re-debited credit, eventually starving the + // connection on lossy paths after a few PTO rounds. + if (!chunk.isRetransmit) { + connBudget -= chunk.data.size + conn.sendConnectionFlowConsumed += chunk.data.size + } } } tierStart = tierEnd diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/AckedPackets.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/AckedPackets.kt index ec98d7b413..5028276840 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/AckedPackets.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/AckedPackets.kt @@ -34,8 +34,14 @@ import com.vitorpamplona.quic.frame.AckFrame * from blowing up the iterator. * * The function does not allocate per-PN; it iterates by primitive - * `Long` and invokes [block] for each ACK'd PN. Hot path on every - * inbound ACK frame, so allocation matters. + * `Long` and invokes [block] for each ACK'd PN. + * + * NOTE: this primitive walks every PN in the range. Production code + * MUST use [drainAckedSentPackets] (which is bounded by the in-flight + * map) — a hostile peer with `firstAckRange = 2^62-1` would otherwise + * pin a core forever inside this loop. This entry point is retained + * for tests that intentionally inspect the on-wire range expansion + * with small, well-formed inputs. */ inline fun forEachAckedPacketNumber( ack: AckFrame, @@ -71,14 +77,62 @@ inline fun forEachAckedPacketNumber( * `quic/plans/2026-05-04-control-frame-retransmit.md`). * * Caller must hold the connection lock. + * + * DoS hardening: instead of walking every PN inside each ACK range + * (up to 2^62 with a hostile peer — single 8-byte varint pinning a + * core forever), we walk the [sentPackets] keys (bounded by the + * congestion window, typically ≤ a few thousand) and check + * membership against the parsed range list. O(N·M) where N = sent + * packets in flight, M = ACK ranges (bounded by packet payload + * size). A peer with `firstAckRange = 2^62-1` simply matches every + * legitimate in-flight PN once, then returns. */ fun drainAckedSentPackets( sentPackets: MutableMap, ack: AckFrame, ): List { + if (sentPackets.isEmpty()) return emptyList() + val ranges = parseAckRanges(ack) + if (ranges.isEmpty()) return emptyList() val drained = mutableListOf() - forEachAckedPacketNumber(ack) { pn -> - sentPackets.remove(pn)?.let { drained += it } + val it = sentPackets.entries.iterator() + while (it.hasNext()) { + val entry = it.next() + val pn = entry.key + for (i in ranges.indices) { + val r = ranges[i] + if (pn in r) { + drained += entry.value + it.remove() + break + } + } } return drained } + +/** + * Parse an [AckFrame]'s on-wire ranges into a list of `[smallest, largest]` + * `LongRange` entries, descending. Each range is clamped to non-negative + * PNs and bounded against [Long] underflow on the additional-ranges + * walk. Stops early when the descending walk crosses zero (per RFC 9000 + * §19.3.1, all PNs are non-negative). The result is small (bounded by + * the ACK frame's payload size). + */ +private fun parseAckRanges(ack: AckFrame): List { + val largest = ack.largestAcknowledged + if (largest < 0L) return emptyList() + val firstSmallest = (largest - ack.firstAckRange).coerceAtLeast(0L) + val out = ArrayList(ack.additionalRanges.size + 1) + out += firstSmallest..largest + var prevSmallest = firstSmallest + for (range in ack.additionalRanges) { + // RFC 9000 §19.3.1: nextLargest = prevSmallest - gap - 2. + val nextLargest = prevSmallest - range.gap - 2L + if (nextLargest < 0L) break + val nextSmallest = (nextLargest - range.ackRangeLength).coerceAtLeast(0L) + out += nextSmallest..nextLargest + prevSmallest = nextSmallest + } + return out +} diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/crypto/InitialSecrets.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/crypto/InitialSecrets.kt index 66051349b3..cd5b30dd6d 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/crypto/InitialSecrets.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/crypto/InitialSecrets.kt @@ -37,7 +37,14 @@ package com.vitorpamplona.quic.crypto * protection epoch. */ object InitialSecrets { - val V1_INITIAL_SALT: ByteArray = + /** + * RFC 9001 §5.2 fixed Initial-secret salt for QUIC v1. Private + only + * read internally by [derive] — pre-fix the constant was a public + * mutable [ByteArray] that any caller could stomp on (or that any + * stack trace / `toString()` could leak). Crypto material doesn't + * need to be reachable outside the derive path. + */ + private val V1_INITIAL_SALT: ByteArray = byteArrayOf( 0x38.toByte(), 0x76.toByte(), diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/http3/Http3FrameReader.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/http3/Http3FrameReader.kt index b7689d9f8f..f00816f0d1 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/http3/Http3FrameReader.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/http3/Http3FrameReader.kt @@ -33,13 +33,71 @@ import com.vitorpamplona.quic.Varint * Use one [Http3FrameReader] per HTTP/3 stream. Feed bytes via [push]; drain * frames via [next] until it returns null. * - * Unknown frame types (per RFC 9114 §9 — "frame types in the format and intent - * not recognized SHOULD be ignored") are surfaced as [Http3Frame.Unknown] so - * the caller can decide policy. SETTINGS, HEADERS, DATA are typed. + * Stream-context enforcement (RFC 9114 §7.2): + * * On the [StreamContext.CONTROL] stream, the FIRST frame MUST be SETTINGS + * (else H3_MISSING_SETTINGS); DATA / HEADERS / PUSH_PROMISE are not + * allowed (H3_FRAME_UNEXPECTED). + * * On a [StreamContext.REQUEST] stream, SETTINGS / GOAWAY / MAX_PUSH_ID / + * CANCEL_PUSH are not allowed (H3_FRAME_UNEXPECTED). + * * On a [StreamContext.PUSH] stream, SETTINGS / GOAWAY / MAX_PUSH_ID / + * PUSH_PROMISE / CANCEL_PUSH are not allowed. + * * Reserved frame types `0x02, 0x06, 0x08, 0x09` are explicitly rejected + * (H3_FRAME_UNEXPECTED) per RFC 9114 §7.2.8 — only the + * `0x21 + 0x1f * N` reserved-greasing pattern is ignored as Unknown. + * * On [StreamContext.WT_BIDI_DATA] / [StreamContext.WT_UNI_DATA] + * (post-prefix WebTransport stream payload), no HTTP/3 framing applies + * — the caller never feeds bytes through here. Provided for symmetry. + * + * Memory safety: + * * Pending buffered bytes are capped at [maxPendingBytes]; a peer that + * streams a partial-frame prefix without ever delivering the body + * cannot pin unbounded heap. Exceeding the cap throws + * [QuicCodecException], which the caller surfaces as a connection + * error. + * * Per-frame body length is capped by [maxFrameBodyBytes]. A peer that + * advertises a 2 GiB length on the wire is rejected before the + * `ByteArray(len.toInt())` allocation. */ -class Http3FrameReader { +class Http3FrameReader( + private val context: StreamContext = StreamContext.UNCHECKED, + private val maxPendingBytes: Int = DEFAULT_MAX_PENDING_BYTES, + private val maxFrameBodyBytes: Int = DEFAULT_MAX_FRAME_BODY_BYTES, +) { private var buf: ByteArray = ByteArray(0) private var pos: Int = 0 + private var framesEmitted: Int = 0 + + /** Bytes currently buffered awaiting a complete frame. Diagnostic; tests use this. */ + val bufferedBytes: Int get() = buf.size - pos + + /** + * Stream context for [Http3FrameReader] frame validation. The reader is + * stateless about which kind of stream it's running on unless the + * caller tells it. [UNCHECKED] preserves pre-fix behaviour for tests + * that intentionally feed mixed frames; production callers should + * specify the actual context. + */ + enum class StreamContext { + /** No per-context validation (pre-fix behaviour, tests). */ + UNCHECKED, + + /** RFC 9114 §6.2.1 server CONTROL unidi stream. */ + CONTROL, + + /** RFC 9114 §6.1 client/server REQUEST bidi stream. */ + REQUEST, + + /** RFC 9114 §6.2.2 PUSH unidi stream. */ + PUSH, + + /** + * Post-prefix WT bidi/uni stream payload. WebTransport bytes don't + * use HTTP/3 framing — this exists so the type system can prevent + * accidental use of [Http3FrameReader] on raw WT data. + */ + WT_BIDI_DATA, + WT_UNI_DATA, + } fun push(bytes: ByteArray) { if (bytes.isEmpty()) return @@ -50,6 +108,14 @@ class Http3FrameReader { buf = buf.copyOfRange(pos, buf.size) pos = 0 } + val newPending = (buf.size - pos).toLong() + bytes.size.toLong() + if (newPending > maxPendingBytes) { + throw QuicCodecException( + "HTTP/3 frame reader buffer would exceed cap " + + "($newPending > $maxPendingBytes); peer is streaming a partial " + + "frame without delivering the body — likely H3_EXCESSIVE_LOAD", + ) + } val combined = ByteArray(buf.size + bytes.size) buf.copyInto(combined, 0) bytes.copyInto(combined, buf.size) @@ -63,21 +129,132 @@ class Http3FrameReader { val lenRes = Varint.decode(buf, typeEnd) ?: return null val bodyStart = typeEnd + lenRes.bytesConsumed val len = lenRes.value - if (len < 0 || len > Int.MAX_VALUE.toLong()) { - throw QuicCodecException("HTTP/3 frame length out of range: $len") + if (len < 0 || len > maxFrameBodyBytes.toLong()) { + throw QuicCodecException( + "HTTP/3 frame length out of range: $len (cap $maxFrameBodyBytes)", + ) } val bodyEnd = bodyStart + len.toInt() if (bodyEnd > buf.size) return null // not all body bytes present yet + val type = typeRes.value + validateFrameType(type) val body = buf.copyOfRange(bodyStart, bodyEnd) pos = bodyEnd - return when (typeRes.value) { + framesEmitted++ + return when (type) { Http3FrameType.DATA -> Http3Frame.Data(body) Http3FrameType.HEADERS -> Http3Frame.Headers(body) Http3FrameType.SETTINGS -> Http3Frame.Settings(Http3Settings.decodeBody(body)) Http3FrameType.GOAWAY -> Http3Frame.Goaway(body) - else -> Http3Frame.Unknown(typeRes.value, body) + else -> Http3Frame.Unknown(type, body) } } + + /** + * Enforce RFC 9114 §7.2 per-stream-context rules. Throws + * [QuicCodecException] (mapped upstream to H3_FRAME_UNEXPECTED / + * H3_MISSING_SETTINGS) on violation. + */ + private fun validateFrameType(type: Long) { + // RFC 9114 §7.2.8: explicit reserved-and-forbidden frame types. + // Distinct from the reserved-greasing pattern `0x21 + 0x1f * N` + // (which is allowed and falls through as Unknown). + if (type == 0x02L || type == 0x06L || type == 0x08L || type == 0x09L) { + throw QuicCodecException( + "H3_FRAME_UNEXPECTED: reserved HTTP/3 frame type 0x${type.toString(16)}", + ) + } + when (context) { + StreamContext.UNCHECKED -> { + Unit + } + + StreamContext.CONTROL -> { + // §7.2.4: SETTINGS MUST be the first frame on the + // control stream; any non-SETTINGS first frame => + // H3_MISSING_SETTINGS. + if (framesEmitted == 0 && type != Http3FrameType.SETTINGS) { + throw QuicCodecException( + "H3_MISSING_SETTINGS: first CONTROL-stream frame was 0x${type.toString(16)}", + ) + } + // §7.2.4: a second SETTINGS is also forbidden. + if (framesEmitted > 0 && type == Http3FrameType.SETTINGS) { + throw QuicCodecException( + "H3_FRAME_UNEXPECTED: duplicate SETTINGS on CONTROL stream", + ) + } + // §7.2.1, §7.2.2, §7.2.5: DATA / HEADERS / PUSH_PROMISE + // are forbidden on the control stream. + if (type == Http3FrameType.DATA || + type == Http3FrameType.HEADERS || + type == Http3FrameType.PUSH_PROMISE + ) { + throw QuicCodecException( + "H3_FRAME_UNEXPECTED: 0x${type.toString(16)} on CONTROL stream", + ) + } + } + + StreamContext.REQUEST -> { + // §7.2.4 / §7.2.6 / §7.2.7: control-only frames are + // forbidden on request streams. + if (type == Http3FrameType.SETTINGS || + type == Http3FrameType.GOAWAY || + type == Http3FrameType.MAX_PUSH_ID || + type == Http3FrameType.CANCEL_PUSH + ) { + throw QuicCodecException( + "H3_FRAME_UNEXPECTED: 0x${type.toString(16)} on REQUEST stream", + ) + } + } + + StreamContext.PUSH -> { + if (type == Http3FrameType.SETTINGS || + type == Http3FrameType.GOAWAY || + type == Http3FrameType.MAX_PUSH_ID || + type == Http3FrameType.PUSH_PROMISE || + type == Http3FrameType.CANCEL_PUSH + ) { + throw QuicCodecException( + "H3_FRAME_UNEXPECTED: 0x${type.toString(16)} on PUSH stream", + ) + } + } + + StreamContext.WT_BIDI_DATA, StreamContext.WT_UNI_DATA -> { + // WebTransport stream payload doesn't use HTTP/3 framing. + // The caller shouldn't be feeding bytes through this + // reader for those contexts; flag any attempt loudly. + throw QuicCodecException( + "WebTransport stream payload routed through Http3FrameReader", + ) + } + } + } + + companion object { + /** + * Default cap on pending unparsed bytes (1 MiB). This is the + * "frame in flight but body not yet complete" headroom — more + * than enough for a legitimate HEADERS / SETTINGS / WT capsule + * preamble plus a single in-progress DATA chunk, far less than + * the heap a hostile peer could pin. + */ + const val DEFAULT_MAX_PENDING_BYTES: Int = 1 shl 20 + + /** + * Default cap on a single HTTP/3 frame body (16 MiB). Exceeds + * any plausible HEADERS payload (RFC 9114 only ever expects + * tens of KiB) and any legitimate per-frame DATA chunk that + * fits inside QUIC's per-packet payload budget. Setting this + * lower than [DEFAULT_MAX_PENDING_BYTES] would require + * splitting; setting it equal keeps the two caps aligned for + * single-frame stalls. + */ + const val DEFAULT_MAX_FRAME_BODY_BYTES: Int = 16 shl 20 + } } /** A parsed HTTP/3 frame. */ diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/http3/Http3Settings.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/http3/Http3Settings.kt index 212e4dbc37..cab63e806f 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/http3/Http3Settings.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/http3/Http3Settings.kt @@ -63,10 +63,62 @@ data class Http3Settings( "duplicate HTTP/3 SETTINGS id 0x${id.toString(16)}", ) } + // RFC 9114 §7.2.4.1 / RFC 9204 §5: per-id range checks. + // A peer that advertises e.g. `MAX_FIELD_SECTION_SIZE = + // 2^60` could otherwise drive the encoder into + // unbounded heap (we'd emit headers up to that cap in a + // single allocation). Bounds chosen to comfortably + // exceed any legitimate value while staying inside + // [Int.MAX_VALUE] for safe `.toInt()` casts at use + // sites. + validateValue(id, value) map[id] = value } return Http3Settings(map) } + + /** + * Reject obviously-malicious SETTINGS values per the relevant + * RFCs. Negative varints are already impossible (varint is + * unsigned), but we double-check defensively. + */ + private fun validateValue( + id: Long, + value: Long, + ) { + if (value < 0L) { + throw com.vitorpamplona.quic.QuicCodecException( + "negative HTTP/3 SETTINGS value for id 0x${id.toString(16)}: $value", + ) + } + // Per-id sanity caps. Unknown ids fall through (RFC 9114 + // §7.2.4.1 says unknown SETTINGS MUST be ignored, but we + // still bound the value to avoid attacker-controlled + // long-tail allocation if any consumer ever uses the + // unknown id directly). + val cap: Long = + when (id) { + Http3SettingsId.QPACK_MAX_TABLE_CAPACITY -> 1L shl 30 + + // 1 GiB + Http3SettingsId.MAX_FIELD_SECTION_SIZE -> 1L shl 30 + + Http3SettingsId.QPACK_BLOCKED_STREAMS -> 65535L + + Http3SettingsId.ENABLE_CONNECT_PROTOCOL -> 1L + + Http3SettingsId.H3_DATAGRAM -> 1L + + Http3SettingsId.ENABLE_WEBTRANSPORT -> 1L + + else -> 1L shl 32 // generic cap for unknown ids + } + if (value > cap) { + throw com.vitorpamplona.quic.QuicCodecException( + "HTTP/3 SETTINGS id 0x${id.toString(16)} value $value exceeds cap $cap", + ) + } + } } } 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 4554cd1f38..5f5416baf6 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/LongHeaderPacket.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/LongHeaderPacket.kt @@ -183,10 +183,27 @@ object LongHeaderPacket { val packet = bytes.copyOfRange(packetStart, packetEnd) val localPnOffset = pnOffset - packetStart - // Step 1: unmask the first byte so we can read pnLen. - val firstByteMask = if ((first and 0x80) != 0) 0x0F else 0x1F - packet[0] = (first xor (mask[0].toInt() and firstByteMask)).toByte() - val pnLen = ((packet[0].toInt() and 0xFF) and 0x03) + 1 + // Step 1: unmask the first byte so we can read pnLen. The + // long-header form bit (0x80) was already validated above (we + // wouldn't be in this function otherwise), so the first-byte + // mask is fixed at 0x0F — pre-fix the conditional `if (form == 1)` + // path was dead code. + packet[0] = (first xor (mask[0].toInt() and 0x0F)).toByte() + val unmaskedFirst = packet[0].toInt() and 0xFF + // RFC 9000 §17.2: long header layout is `1|1|T|T|R|R|P|P` — + // the two reserved bits at 0x0C MUST be zero after HP unmasking. + // A peer that sets either bit is in protocol violation. The + // form-bit (0x80) and fixed-bit (0x40) are not header-protected, + // so they're already correct; we skip the AEAD here on + // reserved-bit set rather than letting a malformed-but-AEAD-OK + // packet drift our state. + if ((unmaskedFirst and 0x0C) != 0) { + throw com.vitorpamplona.quic.QuicProtocolViolationException( + "PROTOCOL_VIOLATION: long-header reserved bits set " + + "(0x${unmaskedFirst.toString(16)})", + ) + } + val pnLen = (unmaskedFirst and 0x03) + 1 // Step 2: unmask exactly `pnLen` packet-number bytes. for (i in 0 until pnLen) { diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/RetryPacket.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/RetryPacket.kt index b3e78a76ea..5a3bdf5119 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/RetryPacket.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/RetryPacket.kt @@ -107,8 +107,13 @@ data class RetryPacket( ) } - /** RFC 9001 §5.8 — fixed retry-integrity AES-128-GCM key for QUIC v1. */ - val V1_RETRY_KEY: ByteArray = + /** + * RFC 9001 §5.8 — fixed retry-integrity AES-128-GCM key for QUIC v1. + * Private — only read by [computeIntegrityTag] inside this file. + * Pre-fix exposing the public mutable [ByteArray] let any caller + * stomp on it or surface it via reflection / `toString()`. + */ + private val V1_RETRY_KEY: ByteArray = byteArrayOf( 0xbe.toByte(), 0x0c.toByte(), @@ -128,8 +133,11 @@ data class RetryPacket( 0x4e.toByte(), ) - /** RFC 9001 §5.8 — fixed retry-integrity AES-128-GCM nonce for QUIC v1. */ - val V1_RETRY_NONCE: ByteArray = + /** + * RFC 9001 §5.8 — fixed retry-integrity AES-128-GCM nonce for QUIC v1. + * Private; same rationale as [V1_RETRY_KEY]. + */ + private val V1_RETRY_NONCE: ByteArray = byteArrayOf( 0x46.toByte(), 0x15.toByte(), 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 c14d3a845d..c5d4ca0737 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/ShortHeaderPacket.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/ShortHeaderPacket.kt @@ -158,7 +158,20 @@ object ShortHeaderPacket { val localPnOffset = pnOffset - offset val firstByteMask = 0x1F packet[0] = (first xor (mask[0].toInt() and firstByteMask)).toByte() - val pnLen = ((packet[0].toInt() and 0xFF) and 0x03) + 1 + val unmaskedFirst = packet[0].toInt() and 0xFF + // RFC 9000 §17.3.1: short header layout is `0|1|S|R|R|K|P|P` — + // the two reserved bits at 0x18 MUST be zero after HP unmasking. + // A peer that sets either bit is in protocol violation. Pre-fix + // we silently accepted the packet and let the AEAD pass; an + // off-spec server (or a fuzzer) could then push us to interop + // bugs that don't reproduce against well-behaved peers. + if ((unmaskedFirst and 0x18) != 0) { + throw com.vitorpamplona.quic.QuicProtocolViolationException( + "PROTOCOL_VIOLATION: short-header reserved bits set " + + "(0x${unmaskedFirst.toString(16)})", + ) + } + val pnLen = (unmaskedFirst and 0x03) + 1 for (i in 0 until pnLen) { packet[localPnOffset + i] = (packet[localPnOffset + i].toInt() xor mask[1 + i].toInt()).toByte() } diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/qpack/QpackHuffman.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/qpack/QpackHuffman.kt index 58687b26a6..9fda41e726 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/qpack/QpackHuffman.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/qpack/QpackHuffman.kt @@ -293,30 +293,54 @@ object QpackHuffman { ) /** - * Lookup tables grouped by code length. `byLength[L]` is a HashMap from - * `code` (an Int up to 30 bits) to `symbol` (0..255) for all symbols - * whose Huffman code is exactly L bits long. Lengths used in the table - * range from 5 to 30. We omit the EOS (length 30, code 0x3FFFFFFF) since - * it must never appear in valid input. + * Lookup tables grouped by code length. For each length L in 5..30, + * [codesByLen]\[L\] holds the codes ascending and [symsByLen]\[L\] + * the matching symbol indices in lock-step — binary-search by the + * candidate Int gives the symbol with no boxing. Pre-fix this was + * `Array>` keyed by boxed `Integer`: every + * decoded character allocated a fresh `Integer` for the + * `candidate` plus a fresh `Integer` return wrapper, dominating + * GC pressure on every QPACK header decode. We omit the EOS + * (length 30, code 0x3FFFFFFF) since it must never appear in + * valid input. */ - private val byLength: Array> = buildLookupByLength() + private val codesByLen: Array + private val symsByLen: Array /** Sorted ascending list of distinct code lengths actually used by the table. */ - private val lengths: IntArray = byLength.indices.filter { byLength[it].isNotEmpty() }.toIntArray() + private val lengths: IntArray - private fun buildLookupByLength(): Array> { - val out = Array(31) { HashMap() } - for (sym in 0..255) { - val code = table[sym][0] - val len = table[sym][1] - out[len][code] = sym + init { + // Allocate slots for lengths up to and including 30 (RFC 7541's + // longest non-EOS code is 30 bits — symbols 10/13/22 — plus EOS + // itself which we exclude by iterating 0..255 below). + val codes = Array(31) { IntArray(0) } + val syms = Array(31) { IntArray(0) } + for (len in 5..30) { + // Collect all symbols whose code length equals `len`, sorted + // by code so we can binary-search at decode time. Iterating + // 0..255 (not 0..256) silently excludes EOS, the only + // length-30 entry that must never appear in valid input. + val pairs = (0..255).filter { table[it][1] == len }.map { it to table[it][0] } + val sorted = pairs.sortedBy { it.second } + codes[len] = IntArray(sorted.size) { sorted[it].second } + syms[len] = IntArray(sorted.size) { sorted[it].first } } - return out + codesByLen = codes + symsByLen = syms + lengths = (5..30).filter { codesByLen[it].isNotEmpty() }.toIntArray() } /** Decode a Huffman-encoded byte sequence into a UTF-8 string. */ fun decode(encoded: ByteArray): ByteArray { - val result = ArrayList(encoded.size * 2) // rough upper bound + // Output is a growable ByteArray rather than ArrayList: + // the latter boxes every emitted byte through `java.lang.Byte`, + // which on a 64-bit JVM is ~16 bytes of wrapper per output byte. + // Pre-grow to ~2× input as a rough upper bound; ASCII-heavy + // headers compress to ~62% with HPACK Huffman, so 2× input is + // an overestimate and we rarely need to grow. + var out = ByteArray(maxOf(8, encoded.size * 2)) + var outPos = 0 var bitBuf = 0L var bitsAvailable = 0 var i = 0 @@ -333,9 +357,12 @@ object QpackHuffman { for (len in lengths) { if (len > bitsAvailable) break val candidate = ((bitBuf ushr (bitsAvailable - len)) and ((1L shl len) - 1)).toInt() - val sym = byLength[len][candidate] - if (sym != null) { - result.add(sym.toByte()) + val codes = codesByLen[len] + val pos = codes.binarySearch(candidate) + if (pos >= 0) { + val sym = symsByLen[len][pos] + if (outPos == out.size) out = out.copyOf(out.size * 2) + out[outPos++] = sym.toByte() bitsAvailable -= len bitBuf = bitBuf and ((1L shl bitsAvailable) - 1) matched = true @@ -354,6 +381,6 @@ object QpackHuffman { throw QuicCodecException("invalid Huffman bit stream") } } - return result.toByteArray() + return if (outPos == out.size) out else out.copyOf(outPos) } } diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/recovery/AckTracker.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/recovery/AckTracker.kt index 90e76d9edf..095cc9a755 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/recovery/AckTracker.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/recovery/AckTracker.kt @@ -55,6 +55,7 @@ class AckTracker { val r = ranges[i] if (packetNumber > r.endInclusive + 1) { ranges.add(i, LongRange(packetNumber, packetNumber)) + trimToMaxRanges() return } if (packetNumber == r.endInclusive + 1) { @@ -80,6 +81,25 @@ class AckTracker { } } ranges += LongRange(packetNumber, packetNumber) + trimToMaxRanges() + } + + /** + * Cap [ranges] at [MAX_STORED_RANGES] by dropping the oldest entries + * (the smallest PNs, at the tail of the descending list). Without + * this a peer that sends a continuous stream of sparse PNs (every + * other one, alternating-bit pattern) inflates the range list + * unboundedly until the encoded ACK frame no longer fits in a + * packet. The trade-off is that the very oldest ranges become + * implicitly "unknown to us"; if the peer retransmits a packet from + * an evicted range we'll re-deliver it and the deduplication is + * downstream's problem (per-stream offset bookkeeping). Spurious + * retransmits on truly ancient PNs are negligible in practice. + */ + private fun trimToMaxRanges() { + while (ranges.size > MAX_STORED_RANGES) { + ranges.removeAt(ranges.lastIndex) + } } fun hasUnackedAckEliciting(): Boolean = ackElicitingPending @@ -143,8 +163,15 @@ class AckTracker { val len = ranges[i].endInclusive - ranges[i].start rest += AckRange(gap, len) } - val ackDelayMicros = (nowMillis - largestRecvTimeMillis) * 1000L - val ackDelay = (ackDelayMicros ushr ackDelayExponent).coerceAtLeast(0L) + // Clamp ≥ 0 BEFORE the shift, not after: `ushr` on a negative + // Long produces a giant positive value that the peer's RTT + // estimator would interpret as a multi-hour delay, poisoning + // its smoothed_rtt below min_rtt. Clock can move backwards if + // [nowMillis] is wallclock-derived (NTP step) — even though we + // default to a monotonic source, the ctor allows a custom + // supplier and tests inject virtual clocks. + val rawDelayMicros = (nowMillis - largestRecvTimeMillis).coerceAtLeast(0L) * 1000L + val ackDelay = rawDelayMicros ushr ackDelayExponent ackElicitingPending = false return AckFrame( largestAcknowledged = largest, @@ -153,4 +180,14 @@ class AckTracker { additionalRanges = rest, ) } + + companion object { + /** + * Cap on stored disjoint ranges. 64 is far above any healthy-link + * count (a single in-order receive needs 1 range; tens of + * reorders per second push that to 4–8). Above this we evict + * oldest first. + */ + const val MAX_STORED_RANGES: Int = 64 + } } 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 46d1cd8881..ef410099c1 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt @@ -220,13 +220,26 @@ class QuicStream( * (two app threads racing the writer's clear-after-emit). */ fun resetStream(errorCode: Long) { - if (resetState != null) return - resetState = - ResetState( - errorCode = errorCode, - finalSize = send.nextOffset, - ) - resetEmitPending = true + // Synchronized atomic compare-and-set: pre-fix the + // `if (resetState != null) return` plus the assignment was + // racy. Two concurrent callers (e.g. the application aborting + // a request while STOP_SENDING from the peer triggers our own + // resetStream from the parser) could both observe null and + // both write — the second write would clobber the first + // errorCode while [resetEmitPending] was already set. The + // writer would then emit a RESET_STREAM with whichever + // errorCode landed last, possibly different from what the + // application asked for. The synchronized block makes + // first-call-wins genuinely first-call-wins. + synchronized(this) { + if (resetState != null) return + resetState = + ResetState( + errorCode = errorCode, + finalSize = send.nextOffset, + ) + resetEmitPending = true + } } /** @@ -244,9 +257,12 @@ class QuicStream( * original frame already on the wire). */ fun stopSending(errorCode: Long) { - if (stopSendingState != null) return - stopSendingState = StopSendingState(errorCode = errorCode) - stopSendingEmitPending = true + // Same atomic-CAS rationale as [resetStream]. + synchronized(this) { + if (stopSendingState != null) return + stopSendingState = StopSendingState(errorCode = errorCode) + stopSendingEmitPending = true + } } /** diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/ReceiveBuffer.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/ReceiveBuffer.kt index 702807c8bd..7f7ff9188a 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/ReceiveBuffer.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/ReceiveBuffer.kt @@ -25,19 +25,31 @@ package com.vitorpamplona.quic.stream * (or for the per-encryption-level CRYPTO offset stream). * * Chunks may arrive in any order with possibly overlapping ranges. The buffer - * coalesces them into a single contiguous prefix that's available to the - * consumer via [readContiguous]. We never expose bytes past the contiguous - * frontier — gaps stall further reads until the missing offsets arrive. + * stores them as **non-overlapping, sorted-by-offset segments** without + * eager coalescing — the contiguous prefix is collected at [readContiguous] + * time. Pre-fix the buffer coalesced overlapping segments on EVERY [insert] + * by allocating a fresh merged `ByteArray((hi - lo).toInt())` and copying + * each existing segment into it; under a 200-chunk reorder burst that's + * O(N²) bytes copied. The new layout is O(1) amortized per [insert] for + * the common cases (adjacent or non-overlapping) and only allocates at + * read time, when the consumer is ready to pull the bytes. * - * For a fully streamed CRYPTO transcript, the consumer just reads contiguous - * bytes whenever new data arrives; the lookup is O(log N) on the gap tree. + * Invariants on [chunks]: + * * Sorted strictly ascending by `offset`. + * * No two segments overlap, but adjacent segments (one ending at the + * next's start) are allowed. + * * No segment lives below [readOffset]. * - * Implementation: we store raw chunks sorted by offset. Calls to [insert] - * coalesce adjacent and overlapping ranges. [readContiguous] returns the - * range from the current cursor up to the first gap. + * For a fully streamed CRYPTO transcript the consumer just reads contiguous + * bytes whenever new data arrives. */ class ReceiveBuffer { - private val chunks = mutableListOf() // sorted by offset, non-overlapping after insert + /** Sorted ascending by [Chunk.offset]; non-overlapping. See class kdoc. */ + private val chunks = ArrayDeque() + + /** Cached `chunks.sumOf { data.size }` so [bufferedAhead] is O(1). */ + private var bufferedAheadBytes: Long = 0L + var readOffset: Long = 0L private set @@ -54,27 +66,42 @@ class ReceiveBuffer { var finOffset: Long? = null private set - /** Insert a chunk at [offset] of size [data.size]. Idempotent on overlap. */ + /** + * Insert a chunk at [offset] of size [data.size]. Idempotent on overlap. + * + * Returns [InsertResult.OK] for a well-formed frame, or one of the + * RFC 9000 §4.5 final-size error variants when the peer's frame is + * inconsistent with state we've already accepted. The caller (the + * QUIC parser) is expected to translate these into a connection + * close with `FINAL_SIZE_ERROR`. Pre-fix the buffer silently + * dropped conflicting FINs and accepted post-FIN extension data, + * letting a buggy peer drift state without termination. + */ fun insert( offset: Long, data: ByteArray, fin: Boolean = false, - ) { - if (data.isEmpty() && !fin) return + ): InsertResult { + if (data.isEmpty() && !fin) return InsertResult.OK + val frameEnd = offset + data.size + // RFC 9000 §4.5: once a final size is established, no STREAM + // frame may extend past it, and a second FIN must agree. + finOffset?.let { existing -> + if (fin && frameEnd != existing) { + return InsertResult.FIN_CONFLICTS_WITH_PRIOR_FIN + } + if (frameEnd > existing) { + return InsertResult.OFFSET_PAST_FIN + } + } if (fin) { finReceived = true - // The FIN flag carries an implicit final offset = offset + data.size. - // RFC 9000 §4.5: once set, this MUST NOT change; ignore subsequent - // FIN frames whose final size disagrees (they should already have - // been rejected at the stream-state level, but be defensive here). - val finalSize = offset + data.size - if (finOffset == null) finOffset = finalSize + if (finOffset == null) finOffset = frameEnd } - if (data.isEmpty()) return + if (data.isEmpty()) return InsertResult.OK - val end = offset + data.size - // Drop chunk parts already consumed. - if (end <= readOffset) return + // Drop chunk parts already consumed by the reader. + if (frameEnd <= readOffset) return InsertResult.OK val effOffset: Long val effData: ByteArray if (offset < readOffset) { @@ -86,54 +113,170 @@ class ReceiveBuffer { effData = data } - // Find the first chunk that's not strictly before the new range. The - // boundary is `<=` so a perfectly adjacent prior chunk (its endOffset - // equals our offset) is included in the merge — otherwise it would - // stay as a separate adjacent chunk and bufferedAhead() would - // overcount on perfectly-sequential receives starting at offset > 0. - var startIdx = 0 - while (startIdx < chunks.size && chunks[startIdx].endOffset() < effOffset) startIdx++ - var endIdx = startIdx - while (endIdx < chunks.size && chunks[endIdx].offset <= effOffset + effData.size) endIdx++ - // Also pull in the prior chunk if it's exactly adjacent on the lower end. - if (startIdx > 0 && chunks[startIdx - 1].endOffset() == effOffset) startIdx -= 1 - - if (startIdx == endIdx) { - // No overlap — just insert. - chunks.add(startIdx, Chunk(effOffset, effData)) - return - } - - // Coalesce [startIdx, endIdx) plus the new chunk. - var lo = effOffset - var hi = effOffset + effData.size - for (i in startIdx until endIdx) { - lo = minOf(lo, chunks[i].offset) - hi = maxOf(hi, chunks[i].endOffset()) - } - val merged = ByteArray((hi - lo).toInt()) - for (i in startIdx until endIdx) { - chunks[i].data.copyInto(merged, (chunks[i].offset - lo).toInt()) - } - effData.copyInto(merged, (effOffset - lo).toInt()) - // Replace - for (i in 1..(endIdx - startIdx)) chunks.removeAt(startIdx) - chunks.add(startIdx, Chunk(lo, merged)) + insertNonOverlapping(effOffset, effData) + return InsertResult.OK } - /** Returns and consumes the contiguous bytes available starting from [readOffset]. */ + /** + * Insert [data] starting at [start] into [chunks], trimming portions + * that already exist (retransmit / overlap). The result preserves the + * non-overlapping, sorted invariants. No allocation when the new + * range is fully covered (retransmit no-op) or when it doesn't + * touch any existing segment (single insert). + */ + private fun insertNonOverlapping( + start: Long, + data: ByteArray, + ) { + val end = start + data.size + // Locate the first existing chunk whose `endOffset > start` — + // anything before it is strictly to the left of [start, end). + var idx = lowerBoundEnd(start) + var cursor = start + + while (cursor < end) { + val existing = if (idx < chunks.size) chunks[idx] else null + val existingStart = existing?.offset ?: Long.MAX_VALUE + if (existingStart >= end) { + // Tail (or whole new range) sits past every existing + // segment that could overlap. Emit the remainder. + emitSegment(start, data, cursor, end) + cursor = end + } else if (existingStart > cursor) { + // Gap from [cursor, existingStart) — emit it, then skip + // over the existing segment (it covers some of [start,end)). + emitSegment(start, data, cursor, existingStart) + cursor = minOf(existing!!.endOffset, end) + idx++ + } else { + // existingStart <= cursor — the existing segment covers + // [cursor, existing.endOffset). Skip ahead. + cursor = minOf(existing!!.endOffset, end) + idx++ + } + } + } + + /** + * Emit the slice of [src] covering absolute range `[segStart, segEnd)` + * as a new chunk, inserted in sorted order. [src] starts at absolute + * offset [srcStart], so the slice index is `(segStart - srcStart)`. + */ + private fun emitSegment( + srcStart: Long, + src: ByteArray, + segStart: Long, + segEnd: Long, + ) { + if (segStart >= segEnd) return + val from = (segStart - srcStart).toInt() + val to = (segEnd - srcStart).toInt() + val piece = + if (from == 0 && to == src.size) { + // Whole src is the segment; avoid the copyOfRange. + src + } else { + src.copyOfRange(from, to) + } + // Insert in sorted position. The index from [lowerBoundOffset] + // is monotonically non-decreasing across the walk in + // [insertNonOverlapping] but we recompute defensively to keep + // this helper self-contained. + val insertAt = lowerBoundOffset(segStart) + chunks.add(insertAt, Chunk(segStart, piece)) + bufferedAheadBytes += piece.size + } + + /** Index of first chunk whose `endOffset > target`, or `chunks.size`. */ + private fun lowerBoundEnd(target: Long): Int { + var lo = 0 + var hi = chunks.size + while (lo < hi) { + val mid = (lo + hi) ushr 1 + if (chunks[mid].endOffset > target) hi = mid else lo = mid + 1 + } + return lo + } + + /** Index of first chunk whose `offset >= target`, or `chunks.size`. */ + private fun lowerBoundOffset(target: Long): Int { + var lo = 0 + var hi = chunks.size + while (lo < hi) { + val mid = (lo + hi) ushr 1 + if (chunks[mid].offset >= target) hi = mid else lo = mid + 1 + } + return lo + } + + /** Highest `offset + data.size` ever seen on this stream (whether or not contiguous). */ + fun highestObservedOffset(): Long { + val finEnd = finOffset + val bufEnd = if (chunks.isEmpty()) readOffset else chunks.last().endOffset + return if (finEnd != null) maxOf(finEnd, bufEnd) else bufEnd + } + + /** Result of an [insert] call — see RFC 9000 §4.5. */ + enum class InsertResult { + /** Frame accepted (or harmlessly redundant). */ + OK, + + /** + * The frame's `offset + data.size` exceeds the previously-established + * final size. Caller MUST close with FINAL_SIZE_ERROR. + */ + OFFSET_PAST_FIN, + + /** + * A second FIN-bearing frame disagreed with the established final + * size. Caller MUST close with FINAL_SIZE_ERROR. + */ + FIN_CONFLICTS_WITH_PRIOR_FIN, + } + + /** + * Returns and consumes the contiguous bytes available starting from + * [readOffset]. Walks every consecutive segment whose left edge meets + * the current read frontier and returns them as a single concatenated + * [ByteArray]. Returns an empty array if the next pending segment is + * beyond the current frontier (gap) or the buffer is empty. + * + * Allocation: zero allocations when only one segment is consecutive + * (the most common case — one inserted segment per call); a single + * concat allocation when multiple segments stack up after a gap fill. + */ fun readContiguous(): ByteArray { - if (chunks.isEmpty()) return ByteArray(0) - val first = chunks[0] - if (first.offset != readOffset) return ByteArray(0) - val data = first.data - readOffset += data.size - chunks.removeAt(0) - return data + if (chunks.isEmpty() || chunks.first().offset != readOffset) return EMPTY + val first = chunks.removeFirst() + bufferedAheadBytes -= first.data.size + var cursor = first.endOffset + // Fast-path: only one consecutive segment. + if (chunks.isEmpty() || chunks.first().offset != cursor) { + readOffset = cursor + return first.data + } + // Multiple consecutive segments — collect, then concat once. + val collected = mutableListOf(first) + var total = first.data.size.toLong() + while (chunks.isNotEmpty() && chunks.first().offset == cursor) { + val next = chunks.removeFirst() + bufferedAheadBytes -= next.data.size + cursor = next.endOffset + collected += next + total += next.data.size + } + val out = ByteArray(total.toInt()) + var pos = 0 + for (c in collected) { + c.data.copyInto(out, pos) + pos += c.data.size + } + readOffset = cursor + return out } /** Bytes already buffered and held back due to gaps. */ - fun bufferedAhead(): Long = chunks.sumOf { it.data.size.toLong() } + fun bufferedAhead(): Long = bufferedAheadBytes /** Highest contiguous offset received so far. */ fun contiguousEnd(): Long = readOffset @@ -150,6 +293,10 @@ class ReceiveBuffer { val offset: Long, val data: ByteArray, ) { - fun endOffset() = offset + data.size + val endOffset: Long get() = offset + data.size + } + + private companion object { + private val EMPTY = ByteArray(0) } } diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt index ec56ee9ada..f4100c5093 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt @@ -223,7 +223,7 @@ class SendBuffer( } addToInFlight(Range(retransmitHead.offset, take, fin)) if (fin) _finSent = true - return@synchronized Chunk(retransmitHead.offset, payload, fin) + return@synchronized Chunk(retransmitHead.offset, payload, fin, isRetransmit = true) } // 2. Fresh bytes. @@ -237,14 +237,14 @@ class SendBuffer( val finForThis = _finPending && !_finSent && nextSendOffset == _nextOffset addToInFlight(Range(offset, take, finForThis)) if (finForThis) _finSent = true - return@synchronized Chunk(offset, payload, finForThis) + return@synchronized Chunk(offset, payload, finForThis, isRetransmit = false) } // 3. FIN-only. if (_finPending && !_finSent) { _finSent = true addToInFlight(Range(nextSendOffset, 0L, true)) - return@synchronized Chunk(nextSendOffset, ByteArray(0), true) + return@synchronized Chunk(nextSendOffset, ByteArray(0), true, isRetransmit = false) } null } @@ -591,6 +591,23 @@ class SendBuffer( } dataLen -= advanceInt flushedFloor += advance + // Shrink the backing buffer when a transient burst has been fully + // drained. Without this, a stream that ever held N bytes pins + // `data.size = N` for the rest of the connection — long-tail + // memory retention. Trigger when: + // * the buffer is large enough to bother (above the doubling + // floor of 64 bytes) AND + // * live data fits in 1/4 of the allocation (capacity is + // ≥ 4× live size). + // Shrink to twice the live size (or [SHRINK_FLOOR_BYTES], + // whichever is larger) to leave headroom for the next push + // without re-doubling immediately. + if (data.size > SHRINK_FLOOR_BYTES && dataLen * 4 < data.size) { + val newCap = maxOf(SHRINK_FLOOR_BYTES, dataLen * 2) + if (newCap < data.size) { + data = data.copyOf(newCap) + } + } } /** @@ -614,6 +631,17 @@ class SendBuffer( } } + private companion object { + /** + * Backing-buffer floor below which [advanceFlushedFloorIfPossible] + * does NOT shrink. Below this size the doubling-on-grow is so + * cheap that shrinking would be a wash; above it, transient + * bursts that bloat the buffer to MiBs leave a long-tail + * memory footprint we want to release. + */ + const val SHRINK_FLOOR_BYTES: Int = 4096 + } + /** * One contiguous offset range tracked by the buffer's bookkeeping. * [length] is `Long` to match QUIC's offset arithmetic — practical @@ -625,21 +653,36 @@ class SendBuffer( val fin: Boolean, ) + /** + * One emit-able STREAM chunk. [isRetransmit] distinguishes a chunk + * pulled off the retransmit queue (path 1 of [takeChunk]) from a + * fresh-bytes emission (path 2/3). Connection-level flow control + * (RFC 9000 §4.1) caps cumulative *new* bytes — retransmits MUST + * NOT consume additional credit, so the writer reads this flag and + * skips `sendConnectionFlowConsumed += data.size` when it's true. + * Without the distinction a single round of loss recovery on a + * long stream eventually exhausts credit and stalls the connection. + */ data class Chunk( val offset: Long, val data: ByteArray, val fin: Boolean, + val isRetransmit: Boolean = false, ) { // ByteArray needs explicit equality. override fun equals(other: Any?): Boolean { if (this === other) return true if (other !is Chunk) return false - return offset == other.offset && fin == other.fin && data.contentEquals(other.data) + return offset == other.offset && + fin == other.fin && + isRetransmit == other.isRetransmit && + data.contentEquals(other.data) } override fun hashCode(): Int { var result = offset.hashCode() result = 31 * result + fin.hashCode() + result = 31 * result + isRetransmit.hashCode() result = 31 * result + data.contentHashCode() return result } diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/tls/TlsHandshakeMessages.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/tls/TlsHandshakeMessages.kt index 1a2e45adaa..618d317883 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/tls/TlsHandshakeMessages.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/tls/TlsHandshakeMessages.kt @@ -77,6 +77,13 @@ data class TlsServerHello( val cipherSuite = r.readUint16() r.readByte() // legacy_compression_method = 0 val extensions = TlsExtension.decodeList(r) + // RFC 8446 §4.1.3: ServerHello body ends with the extensions + // block. Any trailing bytes are a malformed handshake message. + if (r.remaining != 0) { + throw QuicCodecException( + "ServerHello has ${r.remaining} trailing bytes after extensions", + ) + } return TlsServerHello(random, sessionId, cipherSuite, extensions) } } @@ -92,14 +99,38 @@ data class TlsEncryptedExtensions( val alpn: ByteArray? get() = extensions.firstOrNull { it.type == TlsConstants.EXT_ALPN }?.data?.let { - // ALPN response carries a single protocol_name<1..2^8-1> inside protocols<3..2^16-1> + // ALPN response carries EXACTLY one protocol_name<1..2^8-1> + // inside protocols<3..2^16-1> per RFC 7301 §3.1. Reject + // multi-name responses — a server that returns more than + // one is in protocol violation, and silently picking the + // first masks an interop bug. val r = QuicReader(it) - r.skip(2) // outer length - r.readTlsOpaque1() + val outerLen = r.readUint16() + if (outerLen != r.remaining) { + throw QuicCodecException( + "ALPN extension outerLen=$outerLen does not match remaining ${r.remaining}", + ) + } + val name = r.readTlsOpaque1() + if (r.remaining != 0) { + throw QuicCodecException( + "ALPN response carries multiple protocol names (RFC 7301 §3.1 forbids)", + ) + } + name } companion object { - fun decodeBody(r: QuicReader): TlsEncryptedExtensions = TlsEncryptedExtensions(TlsExtension.decodeList(r)) + fun decodeBody(r: QuicReader): TlsEncryptedExtensions { + val extensions = TlsExtension.decodeList(r) + // RFC 8446 §4.3.1: EE body is exactly the extensions block. + if (r.remaining != 0) { + throw QuicCodecException( + "EncryptedExtensions has ${r.remaining} trailing bytes", + ) + } + return TlsEncryptedExtensions(extensions) + } } } @@ -115,6 +146,17 @@ data class TlsCertificateChain( fun decodeBody(r: QuicReader): TlsCertificateChain { val ctx = r.readTlsOpaque1() val listLen = r.readUint24() + // Pre-fix: a malformed peer setting `listLen` larger than the + // remaining body bytes caused `end = r.position + listLen` to + // point past `r.limit`. The while-loop then ran on garbage + // until each `readTlsOpaque*` ran off the end and threw — + // delivering corrupt cert blobs to certs[] before that + // happened. Bound `end` against `r.limit` up front. + if (listLen > r.remaining) { + throw QuicCodecException( + "Certificate listLen=$listLen exceeds remaining body ${r.remaining}", + ) + } val end = r.position + listLen val certs = mutableListOf() while (r.position < end) { @@ -123,6 +165,14 @@ data class TlsCertificateChain( r.readTlsOpaque2() certs += cert } + // RFC 8446 §4.4.2: the certificate_list ends at exactly `end`; + // if a per-cert extensions length pushed us past it, the cert + // chain is malformed and we MUST close. + if (r.position != end) { + throw QuicCodecException( + "Certificate listLen mismatch: position=${r.position} expected=$end", + ) + } return TlsCertificateChain(ctx, certs) } } diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/QuicWebTransportSessionState.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/QuicWebTransportSessionState.kt index e9eac528c2..7f1fe16388 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/QuicWebTransportSessionState.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/QuicWebTransportSessionState.kt @@ -223,6 +223,13 @@ class QuicWebTransportSessionState( connection.streamById(connectStreamId)?.let { it.send.enqueue(encodeCloseSessionCapsule(errorCode, reason)) it.send.finish() + // Wake the driver BEFORE asking it to close — otherwise + // [driver.close] may short-circuit the send loop before our + // newly-enqueued WT_CLOSE_SESSION capsule + FIN drain to + // the wire. The peer would then see an abrupt UDP-level + // tear-down instead of the graceful WT close, and any + // application-error-code we wanted to deliver is lost. + driver.wakeup() } driver.close() // Round-5 concurrency #1: cancel the WT scope so the demux pump diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtCapsule.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtCapsule.kt index fb036a8b44..59521f0ce8 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtCapsule.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtCapsule.kt @@ -48,12 +48,22 @@ fun encodeCloseSessionCapsule( errorCode: Int = 0, reason: String = "", ): ByteArray { + val reasonBytes = reason.encodeToByteArray() + // Symmetric with the decoder: draft-ietf-webtrans-http3 §5 caps the + // reason string at 8192 bytes. Catching this on encode means a local + // bug doesn't ship a capsule the peer must reject. + require(reasonBytes.size <= MAX_WT_CLOSE_REASON_BYTES) { + "WT_CLOSE_SESSION reason exceeds $MAX_WT_CLOSE_REASON_BYTES bytes (${reasonBytes.size})" + } val body = QuicWriter() body.writeUint32(errorCode) - body.writeBytes(reason.encodeToByteArray()) + body.writeBytes(reasonBytes) return encodeCapsule(WtCapsuleType.WT_CLOSE_SESSION, body.toByteArray()) } +/** draft-ietf-webtrans-http3 §5 cap on the reason string in WT_CLOSE_SESSION. */ +const val MAX_WT_CLOSE_REASON_BYTES: Int = 8192 + /** Parsed WT_CLOSE_SESSION capsule. */ data class WtCloseSession( val errorCode: Int, @@ -64,11 +74,22 @@ data class WtCloseSession( * Stateful capsule reader. Capsules on the WT CONNECT bidi stream are * `(varint type)(varint length)(body)`. Feed bytes via [push]; drain * complete capsules via [next]. + * + * Memory safety: pending unparsed bytes are capped at [maxPendingBytes] + * and per-capsule body size is capped at [maxCapsuleBodyBytes]. A peer + * that streams a partial capsule prefix (or advertises a 4 GiB length) + * cannot pin unbounded heap. */ -class CapsuleReader { +class CapsuleReader( + private val maxPendingBytes: Int = DEFAULT_MAX_PENDING_BYTES, + private val maxCapsuleBodyBytes: Int = DEFAULT_MAX_CAPSULE_BODY_BYTES, +) { private var buf: ByteArray = ByteArray(0) private var pos: Int = 0 + /** Bytes currently buffered awaiting a complete capsule. Diagnostic. */ + val bufferedBytes: Int get() = buf.size - pos + fun push(bytes: ByteArray) { if (bytes.isEmpty()) return // Amortized compaction (same pattern as Http3FrameReader). @@ -76,6 +97,14 @@ class CapsuleReader { buf = buf.copyOfRange(pos, buf.size) pos = 0 } + val newPending = (buf.size - pos).toLong() + bytes.size.toLong() + if (newPending > maxPendingBytes) { + throw com.vitorpamplona.quic.QuicCodecException( + "WT capsule reader buffer would exceed cap " + + "($newPending > $maxPendingBytes); peer is streaming a partial " + + "capsule without delivering the body", + ) + } val combined = ByteArray(buf.size + bytes.size) buf.copyInto(combined, 0) bytes.copyInto(combined, buf.size) @@ -97,8 +126,10 @@ class CapsuleReader { .decode(buf, typeEnd) ?: return null val bodyStart = typeEnd + lenRes.bytesConsumed val len = lenRes.value - if (len < 0 || len > Int.MAX_VALUE.toLong()) { - throw com.vitorpamplona.quic.QuicCodecException("capsule length out of range: $len") + if (len < 0 || len > maxCapsuleBodyBytes.toLong()) { + throw com.vitorpamplona.quic.QuicCodecException( + "capsule length out of range: $len (cap $maxCapsuleBodyBytes)", + ) } val bodyEnd = bodyStart + len.toInt() if (bodyEnd > buf.size) return null @@ -137,4 +168,16 @@ class CapsuleReader { } } } + + companion object { + /** Per-stream pending-capsule cap (1 MiB). */ + const val DEFAULT_MAX_PENDING_BYTES: Int = 1 shl 20 + + /** + * Per-capsule body cap (64 KiB). The largest legitimate capsule + * we expect is WT_CLOSE_SESSION (≤ [MAX_WT_CLOSE_REASON_BYTES] + + * 4 bytes); 64 KiB is generous headroom for future capsule types. + */ + const val DEFAULT_MAX_CAPSULE_BODY_BYTES: Int = 64 * 1024 + } } diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtDatagram.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtDatagram.kt index 0d40f7a537..66b0d6ae02 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtDatagram.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtDatagram.kt @@ -45,15 +45,28 @@ object WtDatagram { return w.toByteArray() } - /** Returns (quarterStreamId * 4, payload). Returns null on truncation. */ + /** Returns (quarterStreamId * 4, payload). Returns null on truncation or out-of-range quarter id. */ fun decode(bytes: ByteArray): Decoded? { val r = Varint.decode(bytes, 0) ?: return null if (r.bytesConsumed > bytes.size) return null + // QUIC stream ids are bounded by 2^62 - 1 (RFC 9000 §16). The + // quarter id is the actual stream id divided by 4, so a valid + // wire value is at most (2^62 - 1) / 4. Anything larger is + // either malformed or a deliberate attempt to overflow `r.value + // * 4` past `Long.MAX_VALUE` and have it wrap into a small + // signed value that matches our expected session id — which + // would let an off-path peer feed forged datagrams to the + // application bypassing the session-id check in + // [QuicWebTransportSessionState.pollIncomingDatagram]. + if (r.value < 0 || r.value > MAX_VALID_QUARTER_ID) return null val sessionId = r.value * 4 val payload = bytes.copyOfRange(r.bytesConsumed, bytes.size) return Decoded(sessionId, payload) } + /** (2^62 - 1) / 4 — see [decode]. */ + internal const val MAX_VALID_QUARTER_ID: Long = ((1L shl 62) - 1L) / 4L + data class Decoded( val sessionStreamId: Long, val payload: ByteArray, diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtPeerStreamDemux.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtPeerStreamDemux.kt index a2f48a19ee..700c24276b 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtPeerStreamDemux.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtPeerStreamDemux.kt @@ -82,8 +82,20 @@ class WtPeerStreamDemux( * can leave this null. */ private val driver: com.vitorpamplona.quic.connection.QuicConnectionDriver? = null, + /** + * Cap on the number of peer-initiated streams that may queue here + * waiting for the application to consume them. Pre-fix this was + * `Channel.UNLIMITED` — a peer that opens streams faster than the + * application drains could pin one [StrippedWtStream] (and its + * captured `chunkChannel`) per stream id, indefinitely. Bounded + * buffer + suspending `send` propagates backpressure: when the app + * is slow, [route] suspends inside [emitStripped] before launching + * any further per-stream work, which in turn lets QUIC's per-conn + * `MAX_STREAMS_*` accounting throttle the peer. + */ + private val readyStreamsBuffer: Int = DEFAULT_READY_STREAMS_BUFFER, ) { - private val readyStreams = Channel(Channel.UNLIMITED) + private val readyStreams = Channel(readyStreamsBuffer) @Volatile var peerSettings: Http3Settings? = null @@ -113,6 +125,21 @@ class WtPeerStreamDemux( var peerGoawayProtocolError: String? = null private set + /** + * Surface RFC 9114 §7.2 frame-validation violations the + * [Http3FrameReader] catches (H3_FRAME_UNEXPECTED, H3_MISSING_SETTINGS, + * forbidden reserved types). The route()-level `catch (_: Throwable)` + * cancels the per-stream collector but doesn't propagate the + * underlying [com.vitorpamplona.quic.QuicCodecException]'s message + * upstream — without this field a buggy server would just see + * "stream stops being read" with no diagnostic. Stays null until a + * violation is observed; the QUIC-layer caller polls it and closes + * the connection on non-null. + */ + @Volatile + var peerH3ProtocolError: String? = null + private set + val incomingStrippedStreams: Flow = readyStreams.consumeAsFlow() /** @@ -134,7 +161,16 @@ class WtPeerStreamDemux( kotlinx.coroutines.coroutineScope { val pending = ArrayDeque() val flowIterator = stream.incoming - val chunkChannel = Channel(Channel.UNLIMITED) + // Bounded suspending channel — pre-fix `UNLIMITED` let a + // single misbehaving peer stream pin gigabytes of heap when + // the consumer (the application) couldn't keep up. With a + // bounded buffer + suspending `send`, the collector below + // back-pressures the QuicStream's incoming flow — which in + // turn delays our outbound MAX_STREAM_DATA grants and + // throttles the peer at the QUIC layer. The fixed size is + // small (64 chunks ≈ ~64 KiB at typical packet payloads): + // anything bigger just delays the back-pressure signal. + val chunkChannel = Channel(CHUNK_CHANNEL_BUFFER) val collector = launch { try { @@ -225,13 +261,25 @@ class WtPeerStreamDemux( pending: ArrayDeque, chunkChannel: Channel, ) { - val reader = Http3FrameReader() + val reader = Http3FrameReader(context = Http3FrameReader.StreamContext.CONTROL) // Push whatever we already buffered. - while (pending.isNotEmpty()) reader.push(pending.removeFirst()) - consumeFrames(reader) - for (chunk in chunkChannel) { - reader.push(chunk) + try { + while (pending.isNotEmpty()) reader.push(pending.removeFirst()) consumeFrames(reader) + for (chunk in chunkChannel) { + reader.push(chunk) + consumeFrames(reader) + } + } catch (e: com.vitorpamplona.quic.QuicCodecException) { + // RFC 9114 §7.2 frame-validation throws (H3_FRAME_UNEXPECTED / + // H3_MISSING_SETTINGS / reserved type) land here. Record the + // diagnostic so the QUIC layer / application can close the + // connection deliberately instead of having the route() catch + // silently swallow the message. Idempotent on duplicate hits. + if (peerH3ProtocolError == null) { + peerH3ProtocolError = e.message ?: "HTTP/3 protocol violation on CONTROL stream" + } + throw e } } @@ -289,7 +337,7 @@ class WtPeerStreamDemux( } } - private fun emitStripped( + private suspend fun emitStripped( stream: QuicStream, pending: ArrayDeque, chunkChannel: Channel, @@ -324,7 +372,14 @@ class WtPeerStreamDemux( driver?.wakeup() } } - readyStreams.trySend( + // Suspending send instead of `trySend` so a slow application + // back-pressures the demux instead of silently dropping the + // stream. With a bounded buffer ([readyStreamsBuffer]), + // [readyStreams.send] suspends when the app hasn't drained, + // which keeps `route()` busy and (combined with [process]'s + // `scope.launch`) lets the next peer stream wait its turn + // rather than spawning a coroutine for every offered id. + readyStreams.send( StrippedWtStream( streamId = stream.streamId, isUnidirectional = isUni, @@ -334,6 +389,14 @@ class WtPeerStreamDemux( ), ) } + + companion object { + /** Default cap on queued peer-initiated streams (1024). */ + const val DEFAULT_READY_STREAMS_BUFFER: Int = 1024 + + /** Per-stream chunk-channel capacity (64). */ + internal const val CHUNK_CHANNEL_BUFFER: Int = 64 + } } private fun flatten(chunks: ArrayDeque): ByteArray { diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/http3/Http3FrameReaderTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/http3/Http3FrameReaderTest.kt index 54af94bb9b..c09a265106 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/http3/Http3FrameReaderTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/http3/Http3FrameReaderTest.kt @@ -20,10 +20,13 @@ */ package com.vitorpamplona.quic.http3 +import com.vitorpamplona.quic.QuicCodecException +import com.vitorpamplona.quic.QuicWriter import com.vitorpamplona.quic.webtransport.encodeHeadersFrame import kotlin.test.Test import kotlin.test.assertContentEquals import kotlin.test.assertEquals +import kotlin.test.assertFailsWith import kotlin.test.assertNull import kotlin.test.assertTrue @@ -90,4 +93,55 @@ class Http3FrameReaderTest { assertContentEquals(byteArrayOf(0x01, 0x02, 0x03, 0x04), second.body) assertNull(r.next()) } + + @Test + fun control_stream_first_frame_must_be_settings() { + // RFC 9114 §7.2.4: first CONTROL-stream frame is SETTINGS or + // H3_MISSING_SETTINGS. A peer that opens a CONTROL stream and + // immediately sends GOAWAY is malformed. + val r = Http3FrameReader(context = Http3FrameReader.StreamContext.CONTROL) + val w = QuicWriter() + w.writeVarint(Http3FrameType.GOAWAY) + w.writeVarint(0L) + r.push(w.toByteArray()) + assertFailsWith { r.next() } + } + + @Test + fun control_stream_rejects_data_frame() { + // RFC 9114 §7.2.1: DATA frames are forbidden on the control stream. + val r = Http3FrameReader(context = Http3FrameReader.StreamContext.CONTROL) + // Settings first (legal), then DATA (illegal). + r.push(buildClientWebTransportSettings().encodeFrame()) + assertTrue(r.next() is Http3Frame.Settings) + val w = QuicWriter() + w.writeVarint(Http3FrameType.DATA) + w.writeVarint(0L) + r.push(w.toByteArray()) + assertFailsWith { r.next() } + } + + @Test + fun request_stream_rejects_settings_frame() { + // RFC 9114 §7.2.4: SETTINGS forbidden on request streams. + val r = Http3FrameReader(context = Http3FrameReader.StreamContext.REQUEST) + r.push(buildClientWebTransportSettings().encodeFrame()) + assertFailsWith { r.next() } + } + + @Test + fun reserved_frame_types_are_rejected() { + // RFC 9114 §7.2.8: types 0x02, 0x06, 0x08, 0x09 are reserved + // and MUST be treated as connection error. + for (forbidden in longArrayOf(0x02L, 0x06L, 0x08L, 0x09L)) { + val r = Http3FrameReader(context = Http3FrameReader.StreamContext.REQUEST) + val w = QuicWriter() + w.writeVarint(forbidden) + w.writeVarint(0L) + r.push(w.toByteArray()) + assertFailsWith("type 0x${forbidden.toString(16)} must be rejected") { + r.next() + } + } + } } diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/qpack/HuffmanRfc7541Test.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/qpack/HuffmanRfc7541Test.kt index c90f5bf7ff..f474d92e92 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/qpack/HuffmanRfc7541Test.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/qpack/HuffmanRfc7541Test.kt @@ -94,4 +94,27 @@ class HuffmanRfc7541Test { val decoded = QpackHuffman.decode(encoded).decodeToString() assertEquals("https://www.example.com", decoded) } + + @Test + fun length_30_codes_decode_correctly() { + // RFC 7541 Appendix B: symbols 10 (LF), 13 (CR), 22 (DC2) all use + // 30-bit codes. Hand-encoded with two trailing pad-1 bits so the + // total spans exactly 4 bytes. Pre-fix the [codesByLen] range + // accidentally excluded length 30 and these three bytes were + // silently rejected as "invalid Huffman bit stream". + // sym 10 (LF): code 0x3FFFFFFC | (pad 11) → FF FF FF F3 + // sym 13 (CR): code 0x3FFFFFFD | (pad 11) → FF FF FF F7 + // sym 22 (DC2): code 0x3FFFFFFE | (pad 11) → FF FF FF FB + val cases = + listOf( + 10 to byteArrayOf(0xFF.toByte(), 0xFF.toByte(), 0xFF.toByte(), 0xF3.toByte()), + 13 to byteArrayOf(0xFF.toByte(), 0xFF.toByte(), 0xFF.toByte(), 0xF7.toByte()), + 22 to byteArrayOf(0xFF.toByte(), 0xFF.toByte(), 0xFF.toByte(), 0xFB.toByte()), + ) + for ((sym, encoded) in cases) { + val decoded = QpackHuffman.decode(encoded) + assertEquals(1, decoded.size, "sym=$sym") + assertEquals(sym.toByte(), decoded[0], "sym=$sym") + } + } } 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 3e9087bc1d..592d0b6d49 100644 --- a/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaAesGcmAead.kt +++ b/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaAesGcmAead.kt @@ -34,8 +34,12 @@ import javax.crypto.spec.SecretKeySpec * (which is much cheaper than `getInstance`) plus the AEAD math itself. * * Single-thread per direction: one PacketProtection feeds either the read - * loop OR the send loop, never both. Locking would be needed if that ever - * changes. + * loop OR the send loop, never both. The class still synchronizes on a + * private monitor as a defence-in-depth: the lock-split refactor in + * `QuicConnectionDriver` keeps each side single-threaded by design, but + * a future caller (test harness, key-update path) sharing the instance + * across coroutines would otherwise corrupt the cached `Cipher` state + * silently. The JCA `Cipher` itself is documented as not thread-safe. */ class JcaAesGcmAead( key: ByteArray, @@ -58,29 +62,63 @@ class JcaAesGcmAead( // Cipher.getInstance — slow but rare (once per Initial datagram). private val encryptCipher: Cipher = Cipher.getInstance("AES/GCM/NoPadding") private val decryptCipher: Cipher = Cipher.getInstance("AES/GCM/NoPadding") - private var lastEncryptNonce: ByteArray? = null + + /** + * Last nonce successfully consumed by [seal]. We use a fresh + * [Cipher.getInstance] when the caller asks us to seal under the + * SAME nonce a second time (RFC 9000 §14 Initial-padding rebuild). + * + * Recent-history set rather than just the most-recent nonce: a + * single intermediate seal between two rebuilds could otherwise + * mask a duplicate against the SECOND-most-recent nonce, which + * some JCA providers (Conscrypt) reject with + * `InvalidAlgorithmParameterException` while others (SunJCE) + * silently allow. Bounded at [NONCE_HISTORY_LIMIT] entries — far + * more than any legitimate rebuild path needs (typically 1–2), + * but cheap to keep. + */ + private val recentEncryptNonces = ArrayDeque() override fun seal( key: ByteArray, nonce: ByteArray, aad: ByteArray, plaintext: ByteArray, - ): ByteArray { - // Detect IV reuse — happens on the Initial-padding rebuild path. Fall - // back to a one-shot fresh Cipher for that case rather than fighting - // JCA's safety check. - val reuse = lastEncryptNonce?.contentEquals(nonce) == true - return if (reuse) { - val fresh = Cipher.getInstance("AES/GCM/NoPadding") - fresh.init(Cipher.ENCRYPT_MODE, keySpec, GCMParameterSpec(128, nonce)) - fresh.updateAAD(aad) - fresh.doFinal(plaintext) - } else { - encryptCipher.init(Cipher.ENCRYPT_MODE, keySpec, GCMParameterSpec(128, nonce)) - encryptCipher.updateAAD(aad) - val out = encryptCipher.doFinal(plaintext) - lastEncryptNonce = nonce - out + ): ByteArray = + synchronized(this) { + val reuse = recentEncryptNonces.any { it.contentEquals(nonce) } + if (reuse) { + val fresh = Cipher.getInstance("AES/GCM/NoPadding") + fresh.init(Cipher.ENCRYPT_MODE, keySpec, GCMParameterSpec(128, nonce)) + fresh.updateAAD(aad) + fresh.doFinal(plaintext) + } else { + try { + encryptCipher.init(Cipher.ENCRYPT_MODE, keySpec, GCMParameterSpec(128, nonce)) + encryptCipher.updateAAD(aad) + val out = encryptCipher.doFinal(plaintext) + rememberEncryptNonce(nonce) + out + } catch (t: Throwable) { + // Any throw mid-`init`/`updateAAD`/`doFinal` leaves the + // cached cipher in a provider-defined state. The next + // legitimate call's `init` should reset it, but if a + // crafted input triggers a partial init the next seal + // could conceivably reuse residual state. Drop the + // most recent nonce from the history so a retry with + // the same nonce takes the safe fresh-Cipher path. + if (recentEncryptNonces.lastOrNull()?.contentEquals(nonce) == true) { + recentEncryptNonces.removeLast() + } + throw t + } + } + } + + private fun rememberEncryptNonce(nonce: ByteArray) { + recentEncryptNonces.addLast(nonce) + while (recentEncryptNonces.size > NONCE_HISTORY_LIMIT) { + recentEncryptNonces.removeFirst() } } @@ -90,11 +128,17 @@ class JcaAesGcmAead( aad: ByteArray, ciphertext: ByteArray, ): ByteArray? = - try { - decryptCipher.init(Cipher.DECRYPT_MODE, keySpec, GCMParameterSpec(128, nonce)) - decryptCipher.updateAAD(aad) - decryptCipher.doFinal(ciphertext) - } catch (_: GeneralSecurityException) { - null + synchronized(this) { + try { + decryptCipher.init(Cipher.DECRYPT_MODE, keySpec, GCMParameterSpec(128, nonce)) + decryptCipher.updateAAD(aad) + decryptCipher.doFinal(ciphertext) + } catch (_: GeneralSecurityException) { + null + } } + + private companion object { + private const val NONCE_HISTORY_LIMIT = 8 + } } diff --git a/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt b/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt index 6612013c91..72a1bd4568 100644 --- a/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt +++ b/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt @@ -165,10 +165,23 @@ actual class UdpSocket private constructor( channel.setOption(StandardSocketOptions.IP_TOS, ECT0_TOS_BITS) } channel.bind(InetSocketAddress(0)) // ephemeral - // We use receive()/send(addr) instead of channel.connect() so that - // sendDatagram-style flows can still be implemented on the same - // socket if we ever need them. For the pure client use-case this - // is identical in latency. + // RFC 9000 §9 / defence-in-depth: connect() asks the kernel + // to filter inbound datagrams to those from [remote]. Without + // this any host on the network can spoof our 4-tuple and + // force us to attempt AEAD decryption on garbage — each + // failed packet costs a `Cipher.init` + AAD/decrypt + tag + // check, and a successful unauthenticated stateless reset + // forgery would terminate our connection. With connect(), + // the kernel rejects mismatched-source datagrams before + // they reach userspace. + // + // We connect AFTER bind so the ephemeral local port is + // chosen first, then the destination is associated. Pure + // client semantics — `quic` doesn't currently support + // server-side or connection migration to a new remote IP + // (path validation rotates DCIDs on the SAME 4-tuple), so + // pinning the socket to one remote is a strict win. + channel.connect(remote) UdpSocket(channel, remote) } }