fix(quic): SETTINGS validation, sensitive headers, peer-stream count, drain alloc

Round 7 — closing out the audit's spec/correctness/perf quick wins. * WtPeerStreamDemux: validate peer SETTINGS includes ENABLE_WEBTRANSPORT=1 AND ENABLE_CONNECT_PROTOCOL=1 (draft-ietf-webtrans-http3 §3 + RFC 8441). Pre-fix we accepted any SETTINGS and proceeded to Extended CONNECT, which the server then rejects with no diagnostic for the application. Now surfaces as a typed protocol error on peerH3ProtocolError so the QUIC layer closes deliberately. * QpackEncoder: set N=1 (never-indexed) on literal field lines for authorization / cookie / set-cookie / proxy-authorization per RFC 9204 §4.5.4. Pre-fix sensitive credentials could be cached by intermediate QPACK encoder caches. * QuicConnection.getOrCreatePeerStreamLocked: track peerInitiated*Count via max(current, peerIndex+1) instead of += 1. The counter now derives from the stream id's index field (RFC 9000 §2.1) and is idempotent against retransmits-after-eviction. Pre-fix a peer retransmit on a stream id that aged out of the retired-IDs FIFO bumped the counter again, eventually triggering spurious MAX_STREAMS_* emissions. * applyPeerRetireConnectionIdLocked: reclassify the close-on-seq=0 case as PROTOCOL_VIOLATION (peer fault) instead of INTERNAL_ERROR (our fault). The diagnostic was misleading — the peer IS misbehaving (asking us to retire our only SCID with no replacement available), not us. * QuicConnectionWriter.drainOutbound: skip the `streamsView.filter { !it.isClosed }` allocation when no streams are closed. Quick `any` scan first; only allocate the filtered list when at least one stream is actually closed. Saves an N-sized ArrayList per drain in the common case (~50 drains/sec × N up to ~2000 streams under multiplex load). All 269 :quic:jvmTest tests pass. BUILD SUCCESSFUL in 40s. https://claude.ai/code/session_01AhGvbMV8uPRse3TmAGaddM
2026-05-09 00:14:07 +00:00
parent 5421b81569
commit f987b3dfe2
5 changed files with 132 additions and 18 deletions
@@ -1586,14 +1586,28 @@ class QuicConnection(
        streams[id] = 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
-        // currently-advertised cap. Without this, our initial cap of
-        // [config.initialMaxStreams*] is the lifetime maximum the peer
-        // can open and any longer broadcast silently truncates.
+        // Track the peer's own stream-counter, derived from the stream
+        // id's index field (bits 2..63 hold the per-direction sequence
+        // — see RFC 9000 §2.1). Pre-fix we incremented on every
+        // create-event, which double-counted when the peer retransmits
+        // a STREAM frame on a stream id we've retired AND aged out of
+        // [retiredStreamIdSet] (FIFO ring of bounded size). The
+        // phantom-stream guard in the parser drops most retransmits,
+        // but a sufficiently long broadcast can age the id out of the
+        // ring and the next retransmit lands here, bumping the
+        // counter and eventually triggering a spurious MAX_STREAMS_*
+        // emission. Using max(current, peerIndex + 1) makes the
+        // counter idempotent: re-creating the same id is a no-op,
+        // matching the peer's view exactly.
+        val peerIndexPlusOne = (id ushr 2) + 1L
        when (kind) {
-            StreamId.Kind.SERVER_UNI, StreamId.Kind.CLIENT_UNI -> peerInitiatedUniCount += 1
-            StreamId.Kind.SERVER_BIDI, StreamId.Kind.CLIENT_BIDI -> peerInitiatedBidiCount += 1
+            StreamId.Kind.SERVER_UNI, StreamId.Kind.CLIENT_UNI -> {
+                peerInitiatedUniCount = maxOf(peerInitiatedUniCount, peerIndexPlusOne)
+            }
+
+            StreamId.Kind.SERVER_BIDI, StreamId.Kind.CLIENT_BIDI -> {
+                peerInitiatedBidiCount = maxOf(peerInitiatedBidiCount, peerIndexPlusOne)
+            }
        }
        // Wake any awaitIncomingPeerStream caller. trySend on a CONFLATED
        // channel can never fail in steady state.
@@ -2183,10 +2197,16 @@ class QuicConnection(
            )
            return
        }
-        // Sequence 0: peer wants us off our initial SCID, but we
-        // haven't issued any replacements. Close — see kdoc above.
+        // Sequence 0: peer wants us off our initial SCID, but we never
+        // advertised additional SCIDs (we don't issue NEW_CONNECTION_ID
+        // frames yet). RFC 9000 §19.16 lets us close — and we MUST,
+        // because there's no replacement SCID to switch to. Reclassify
+        // as PROTOCOL_VIOLATION (peer-side fault) rather than
+        // INTERNAL_ERROR (our-side fault) — the original diagnostic
+        // misdescribed which side made the mistake.
        markClosedExternally(
-            "INTERNAL_ERROR: peer asked to retire our initial SCID but we have no replacement to offer",
+            "PROTOCOL_VIOLATION: peer asked to retire our initial SCID (seq=0) but we never " +
+                "advertised additional SCIDs to switch to",
        )
    }

@@ -712,7 +712,16 @@ private fun buildApplicationPacket(
        // The combination drops drainOutbound's per-call cost from
        // O(N log N) where N=total-streams to roughly O(active) under
        // realistic loads.
-        val active = streamsView.filter { !it.isClosed }
+        // Allocation hot-path: under realistic load most streams are
+        // open, so the `filter` allocates an N-sized ArrayList only to
+        // hand back the same N elements. Quick `any` scan first; only
+        // pay for the filter list when at least one stream is closed.
+        val active =
+            if (streamsView.any { it.isClosed }) {
+                streamsView.filter { !it.isClosed }
+            } else {
+                streamsView
+            }
        val sorted =
            when {
                active.size <= 1 -> active
@@ -54,6 +54,13 @@ class QpackEncoder {
        name: String,
        value: String,
    ) {
+        // RFC 9204 §4.5.4 / §7.1: sensitive headers (credentials and
+        // session cookies) MUST be marked never-indexed (N=1) so an
+        // intermediate cache cannot retain them. Indexed-field-line
+        // form (no literal value) doesn't need this since no value
+        // appears literally on the wire.
+        val sensitive = isSensitive(name)
+
        val pairIdx = QpackStaticTable.pairToIndex[name to value]
        if (pairIdx != null) {
            // Indexed field line, static. Pattern: 1|T(=1)|index(6-bit prefix)
@@ -62,20 +69,38 @@ class QpackEncoder {
        }
        val nameIdx = QpackStaticTable.nameToIndex[name]
        if (nameIdx != null) {
-            // Literal Field Line With Name Reference, static. Pattern: 0|1|N(=0)|T(=1)|index(4-bit prefix)
-            QpackInteger.encode(nameIdx.toLong(), 4, 0x50, w)
+            // Literal Field Line With Name Reference, static. Pattern:
+            //   0|1|N|T(=1)|index(4-bit prefix)
+            // N=1 for sensitive headers per RFC 9204 §4.5.4.
+            val firstByte = if (sensitive) 0x70 else 0x50
+            QpackInteger.encode(nameIdx.toLong(), 4, firstByte, w)
            // Then value: H=0|len(7-bit prefix)
            val valueBytes = value.encodeToByteArray()
            QpackInteger.encode(valueBytes.size.toLong(), 7, 0x00, w)
            w.writeBytes(valueBytes)
            return
        }
-        // Literal field line with literal name. Pattern: 0|0|1|N(=0)|H(=0)|len(3-bit prefix)
+        // Literal field line with literal name. Pattern:
+        //   0|0|1|N|H(=0)|len(3-bit prefix)
+        // N=1 for sensitive headers.
+        val firstByte = if (sensitive) 0x30 else 0x20
        val nameBytes = name.encodeToByteArray()
-        QpackInteger.encode(nameBytes.size.toLong(), 3, 0x20, w)
+        QpackInteger.encode(nameBytes.size.toLong(), 3, firstByte, w)
        w.writeBytes(nameBytes)
        val valueBytes = value.encodeToByteArray()
        QpackInteger.encode(valueBytes.size.toLong(), 7, 0x00, w)
        w.writeBytes(valueBytes)
    }
+
+    /**
+     * Headers whose values are credentials or session-bound and MUST
+     * NOT be cached by an intermediary per RFC 9204 §4.5.4. Names are
+     * compared after the encoder's lower-case normalization, so we
+     * test against lower-case forms here.
+     */
+    private fun isSensitive(name: String): Boolean =
+        when (name) {
+            "authorization", "proxy-authorization", "cookie", "set-cookie" -> true
+            else -> false
+        }
 }
@@ -24,6 +24,7 @@ import com.vitorpamplona.quic.Varint
 import com.vitorpamplona.quic.http3.Http3Frame
 import com.vitorpamplona.quic.http3.Http3FrameReader
 import com.vitorpamplona.quic.http3.Http3Settings
+import com.vitorpamplona.quic.http3.Http3SettingsId
 import com.vitorpamplona.quic.http3.Http3StreamType
 import com.vitorpamplona.quic.stream.QuicStream
 import com.vitorpamplona.quic.stream.StreamId
@@ -288,6 +289,28 @@ class WtPeerStreamDemux(
            val frame = reader.next() ?: return
            when (frame) {
                is Http3Frame.Settings -> {
+                    // draft-ietf-webtrans-http3 §3 + RFC 8441: a server that
+                    // accepts WebTransport MUST advertise both
+                    // ENABLE_WEBTRANSPORT=1 and ENABLE_CONNECT_PROTOCOL=1.
+                    // If either is missing/zero, the WT session can't
+                    // proceed — surface as an H3 protocol error so
+                    // [drainControlStream]'s catch records it on
+                    // [peerH3ProtocolError] and the application closes
+                    // the connection instead of issuing an Extended
+                    // CONNECT the server will reject anyway.
+                    val s = frame.settings.settings
+                    val enableWt = s[Http3SettingsId.ENABLE_WEBTRANSPORT] ?: 0L
+                    val enableConnect = s[Http3SettingsId.ENABLE_CONNECT_PROTOCOL] ?: 0L
+                    if (enableWt != 1L) {
+                        throw com.vitorpamplona.quic.QuicCodecException(
+                            "peer SETTINGS missing ENABLE_WEBTRANSPORT=1 (got $enableWt)",
+                        )
+                    }
+                    if (enableConnect != 1L) {
+                        throw com.vitorpamplona.quic.QuicCodecException(
+                            "peer SETTINGS missing ENABLE_CONNECT_PROTOCOL=1 (got $enableConnect)",
+                        )
+                    }
                    peerSettings = frame.settings
                }

@@ -23,6 +23,7 @@ package com.vitorpamplona.quic.webtransport
 import com.vitorpamplona.quic.QuicWriter
 import com.vitorpamplona.quic.http3.Http3FrameType
 import com.vitorpamplona.quic.http3.Http3Settings
+import com.vitorpamplona.quic.http3.Http3SettingsId
 import com.vitorpamplona.quic.http3.Http3StreamType
 import com.vitorpamplona.quic.stream.QuicStream
 import kotlinx.coroutines.CoroutineScope
@@ -68,7 +69,19 @@ class WtPeerStreamDemuxTest {
            //   2. SETTINGS frame with a single QPACK_MAX_TABLE_CAPACITY=0.
            //   3. GOAWAY frame with stream id 12.
            val typePrefix = QuicWriter().also { it.writeVarint(Http3StreamType.CONTROL) }.toByteArray()
-            val settingsFrame = Http3Settings(emptyMap()).encodeFrame()
+            // Realistic server SETTINGS — both ENABLE_CONNECT_PROTOCOL=1
+            // and ENABLE_WEBTRANSPORT=1 are required for the demux to
+            // accept the SETTINGS frame (draft-ietf-webtrans-http3 §3 +
+            // RFC 8441). Pre-validation tests used emptyMap() as a
+            // stand-in; with the validation in place we mirror what a
+            // real WT-capable server sends.
+            val settingsFrame =
+                Http3Settings(
+                    mapOf(
+                        Http3SettingsId.ENABLE_CONNECT_PROTOCOL to 1L,
+                        Http3SettingsId.ENABLE_WEBTRANSPORT to 1L,
+                    ),
+                ).encodeFrame()
            val goawayBody = QuicWriter().also { it.writeVarint(12L) }.toByteArray()
            val goawayFrame =
                QuicWriter()
@@ -113,7 +126,19 @@ class WtPeerStreamDemuxTest {
            demux.process(stream)

            val typePrefix = QuicWriter().also { it.writeVarint(Http3StreamType.CONTROL) }.toByteArray()
-            val settingsFrame = Http3Settings(emptyMap()).encodeFrame()
+            // Realistic server SETTINGS — both ENABLE_CONNECT_PROTOCOL=1
+            // and ENABLE_WEBTRANSPORT=1 are required for the demux to
+            // accept the SETTINGS frame (draft-ietf-webtrans-http3 §3 +
+            // RFC 8441). Pre-validation tests used emptyMap() as a
+            // stand-in; with the validation in place we mirror what a
+            // real WT-capable server sends.
+            val settingsFrame =
+                Http3Settings(
+                    mapOf(
+                        Http3SettingsId.ENABLE_CONNECT_PROTOCOL to 1L,
+                        Http3SettingsId.ENABLE_WEBTRANSPORT to 1L,
+                    ),
+                ).encodeFrame()
            // First GOAWAY = 8.
            val ga1Body = QuicWriter().also { it.writeVarint(8L) }.toByteArray()
            val ga1 =
@@ -166,7 +191,19 @@ class WtPeerStreamDemuxTest {
            demux.process(stream)

            val typePrefix = QuicWriter().also { it.writeVarint(Http3StreamType.CONTROL) }.toByteArray()
-            val settingsFrame = Http3Settings(emptyMap()).encodeFrame()
+            // Realistic server SETTINGS — both ENABLE_CONNECT_PROTOCOL=1
+            // and ENABLE_WEBTRANSPORT=1 are required for the demux to
+            // accept the SETTINGS frame (draft-ietf-webtrans-http3 §3 +
+            // RFC 8441). Pre-validation tests used emptyMap() as a
+            // stand-in; with the validation in place we mirror what a
+            // real WT-capable server sends.
+            val settingsFrame =
+                Http3Settings(
+                    mapOf(
+                        Http3SettingsId.ENABLE_CONNECT_PROTOCOL to 1L,
+                        Http3SettingsId.ENABLE_WEBTRANSPORT to 1L,
+                    ),
+                ).encodeFrame()
            stream.deliverIncoming(typePrefix)
            stream.deliverIncoming(settingsFrame)
            stream.closeIncoming()