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 3fbeff666..457cfef98 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 b2fc47b4b..8efca138a 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 f4927721c..a983ddbe1 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt @@ -408,7 +408,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 +508,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 +623,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 +745,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 bc27225ab..689a630e4 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 ec98d7b41..502827684 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/recovery/AckTracker.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/recovery/AckTracker.kt index 90e76d9ed..d5d9002c6 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/recovery/AckTracker.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/recovery/AckTracker.kt @@ -143,8 +143,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, 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 702807c8b..00de823da 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/ReceiveBuffer.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/ReceiveBuffer.kt @@ -54,27 +54,43 @@ 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 + if (end <= readOffset) return InsertResult.OK val effOffset: Long val effData: ByteArray if (offset < readOffset) { @@ -101,7 +117,7 @@ class ReceiveBuffer { if (startIdx == endIdx) { // No overlap — just insert. chunks.add(startIdx, Chunk(effOffset, effData)) - return + return InsertResult.OK } // Coalesce [startIdx, endIdx) plus the new chunk. @@ -119,6 +135,32 @@ class ReceiveBuffer { // Replace for (i in 1..(endIdx - startIdx)) chunks.removeAt(startIdx) chunks.add(startIdx, Chunk(lo, merged)) + return InsertResult.OK + } + + /** 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]. */ 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 ec56ee9ad..ed63d19bd 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 } @@ -625,21 +625,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/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaAesGcmAead.kt b/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaAesGcmAead.kt index 3e9087bc1..592d0b6d4 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 6612013c9..72a1bd456 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) } }