This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-http.git
The following commit(s) were added to refs/heads/main by this push:
new 524a55751 hpack: decode header blocks from any InputStream, without
copying (#1251)
524a55751 is described below
commit 524a55751cbfb131aeb531144b0d37de406b3595
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Sep 11 12:16:32 2026 +0100
hpack: decode header blocks from any InputStream, without copying (#1251)
* hpack: decode header blocks from any InputStream, without copying
Motivation:
HeaderDecompression compacted every header block before decoding it, because
the decoder needed an InputStream with mark/reset and an available()
covering
the whole block - which of ByteString's implementations only the
array-backed
ones provide. A HEADERS frame followed by CONTINUATION frames is assembled
with
`++`, so it is a rope whose asInputStream is a SequenceInputStream, and
even a
single-frame payload is usually a slice of a network buffer that compact()
copies in full.
#1231 removed the decoder's assumption that one read() fills the buffer.
Three
dependencies on ByteArrayInputStream semantics remained:
- decodeULE128 used mark(5)/reset() to rewind a partially read varint
- the main loop was driven by `while (in.available() > 0)`
- the literal name and value states waited for `available() >= length`
Modification:
Read forward only. The main loop now ends when read() reports the end of the
stream between representations, which is the normal end of a header block,
and
every other state treats the end of the stream as truncated input.
decodeULE128
reads without marking, readByte reports the end of the stream as a
decompression failure, and skipFully skips a run in one go while coping with
skip() returning zero. The available() guards before readStringLiteral are
gone
because readNBytes already reports a short literal.
HeaderDecompression then hands the payload straight to the decoder, so a
header
block is no longer copied.
Result:
No functional change for a well-formed block. A block that ends mid
representation is now reported as a decompression failure -
COMPRESSION_ERROR,
per RFC 9113 section 4.3 - where it previously decoded to however many
headers
had been read so far. The decoder no longer supports being fed a block
incrementally across calls; parseAndEmit is its only caller and always
passes a
complete block.
Tests:
- HpackDecoderSpec gains coverage over a SequenceInputStream that supports
neither mark/reset nor a whole-block available(), at chunk sizes 4, 3 and
1,
plus two truncated blocks. The chunked cases fail before this change with
a
decompression failure - 8 passed
- sbt "http-core/testOnly org.apache.pekko.http.impl.engine.http2.*" - 18
passed
- sbt http2-tests/test - 352 passed, 25 ignored, 26 pending
- scalafmtCheckAll, javafmtCheckAll, headerCheck,
http-core/mimaReportBinaryIssues - clean
References:
Follows #1231
Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
* fix: keep the HPACK decoder usable after a header fails to parse
Motivation:
HeaderDecompression's HeaderListener threw ParsingException straight
through the
HPACK decoder, which left the connection unable to decode any later HEADERS
frame:
- Decoder.insertHeader calls the listener before adding the entry to the
dynamic
table, so the entry for the offending header was never added
- decode() unwound at that point, so every representation after it in the
block
was never read and never added either
- endHeaderBlock() was called inside the try, so it was skipped and the
decoder
kept the state and headerSize of the abandoned block
The first two desynchronise the decoder's dynamic table from the peer's
encoder's, which HPACK cannot recover from; the third resumes the next block
part way through a representation. HeaderDecompression answers a parse
failure
with a bad request and keeps the connection open, so this is reachable with
a
single malformed header - an unknown method is enough.
Modification:
Catch ParsingException in the listener, remember the first ErrorInfo and
return
null so that decoding runs to the end of the block and the dynamic table
keeps
tracking the peer's. Report the remembered failure once the block is
decoded.
Call endHeaderBlock() in a finally as well, so anything else that unwinds -
a
malformed pseudo header raises Http2ProtocolException - still resets the
decoder. The outer ParsingException handler stays as a fallback.
Result:
A request with an unparseable header still gets a bad request response, and
subsequent requests on the same connection are decoded correctly.
Tests:
- New "keep the connection usable after a header parsing failure" in
Http2ClientServerSpec sends a request with an unknown method, expects the
bad
request, then sends a valid request on the same connection. Without the
fix
the second request never reaches the handler at all - the spec times out
waiting for it - and with the fix it is served normally
- sbt "http2-tests/testOnly ...Http2ClientServerSpec" - 8 passed
- sbt "http-core/testOnly org.apache.pekko.http.impl.engine.http2.*" - 13
passed
- sbt http2-tests/test - 353 passed, 25 ignored, 26 pending
- scalafmtCheckAll, headerCheck, http-core/mimaReportBinaryIssues - clean
References:
Noticed while working on #1251
Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
---------
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
.../http/shaded/com/twitter/hpack/Decoder.java | 106 ++++++++++-----------
.../engine/http2/hpack/HeaderDecompression.scala | 38 ++++++--
.../impl/engine/http2/hpack/HpackDecoderSpec.scala | 35 ++++++-
.../impl/engine/http2/Http2ClientServerSpec.scala | 16 ++++
4 files changed, 134 insertions(+), 61 deletions(-)
diff --git
a/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/Decoder.java
b/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/Decoder.java
index 0356a6690..63bae0b75 100644
---
a/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/Decoder.java
+++
b/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/Decoder.java
@@ -98,12 +98,25 @@ public final class Decoder {
indexType = IndexType.NONE;
}
- /** Decode the header block into header fields. */
+ /**
+ * Decode the header block into header fields.
+ *
+ * <p>{@code in} must hold the complete header block. The decoder reads
forward until the stream
+ * ends and reports a block that stops in the middle of a representation as
a decompression
+ * failure, so the stream needs to support neither mark/reset nor an {@code
available()} that
+ * covers the whole remaining block.
+ */
public void decode(InputStream in, HeaderListener headerListener) throws
IOException {
- while (in.available() > 0) {
+ while (true) {
switch (state) {
case READ_HEADER_REPRESENTATION:
- byte b = (byte) in.read();
+ int nextByte = in.read();
+ if (nextByte < 0) {
+ // between representations is the one place where running out of
input is the expected
+ // end of the header block rather than a truncated one
+ return;
+ }
+ byte b = (byte) nextByte;
if (maxDynamicTableSizeChangeRequired && (b & 0xE0) != 0x20) {
// Encoder MUST signal maximum dynamic table size change
throw MAX_DYNAMIC_TABLE_SIZE_CHANGE_REQUIRED;
@@ -158,10 +171,6 @@ public final class Decoder {
case READ_MAX_DYNAMIC_TABLE_SIZE:
int maxSize = decodeULE128(in);
- if (maxSize == -1) {
- return;
- }
-
// Check for numerical overflow
if (maxSize > Integer.MAX_VALUE - index) {
throw DECOMPRESSION_EXCEPTION;
@@ -173,10 +182,6 @@ public final class Decoder {
case READ_INDEXED_HEADER:
int headerIndex = decodeULE128(in);
- if (headerIndex == -1) {
- return;
- }
-
// Check for numerical overflow
if (headerIndex > Integer.MAX_VALUE - index) {
throw DECOMPRESSION_EXCEPTION;
@@ -189,10 +194,6 @@ public final class Decoder {
case READ_INDEXED_HEADER_NAME:
// Header Name matches an entry in the Header Table
int nameIndex = decodeULE128(in);
- if (nameIndex == -1) {
- return;
- }
-
// Check for numerical overflow
if (nameIndex > Integer.MAX_VALUE - index) {
throw DECOMPRESSION_EXCEPTION;
@@ -203,7 +204,7 @@ public final class Decoder {
break;
case READ_LITERAL_HEADER_NAME_LENGTH_PREFIX:
- b = (byte) in.read();
+ b = readByte(in);
huffmanEncoded = (b & 0x80) == 0x80;
index = b & 0x7F;
if (index == 0x7f) {
@@ -243,10 +244,6 @@ public final class Decoder {
case READ_LITERAL_HEADER_NAME_LENGTH:
// Header Name is a Literal String
nameLength = decodeULE128(in);
- if (nameLength == -1) {
- return;
- }
-
// Check for numerical overflow
if (nameLength > Integer.MAX_VALUE - index) {
throw DECOMPRESSION_EXCEPTION;
@@ -276,26 +273,19 @@ public final class Decoder {
break;
case READ_LITERAL_HEADER_NAME:
- // Wait until entire name is readable
- if (in.available() < nameLength) {
- return;
- }
-
name = readStringLiteral(in, nameLength);
state = State.READ_LITERAL_HEADER_VALUE_LENGTH_PREFIX;
break;
case SKIP_LITERAL_HEADER_NAME:
- skipLength -= in.skip(skipLength);
-
- if (skipLength == 0) {
- state = State.READ_LITERAL_HEADER_VALUE_LENGTH_PREFIX;
- }
+ skipFully(in, skipLength);
+ skipLength = 0;
+ state = State.READ_LITERAL_HEADER_VALUE_LENGTH_PREFIX;
break;
case READ_LITERAL_HEADER_VALUE_LENGTH_PREFIX:
- b = (byte) in.read();
+ b = readByte(in);
huffmanEncoded = (b & 0x80) == 0x80;
index = b & 0x7F;
if (index == 0x7f) {
@@ -336,10 +326,6 @@ public final class Decoder {
case READ_LITERAL_HEADER_VALUE_LENGTH:
// Header Value is a Literal String
valueLength = decodeULE128(in);
- if (valueLength == -1) {
- return;
- }
-
// Check for numerical overflow
if (valueLength > Integer.MAX_VALUE - index) {
throw DECOMPRESSION_EXCEPTION;
@@ -369,22 +355,15 @@ public final class Decoder {
break;
case READ_LITERAL_HEADER_VALUE:
- // Wait until entire value is readable
- if (in.available() < valueLength) {
- return;
- }
-
String value = readStringLiteral(in, valueLength);
insertHeader(headerListener, name, value, indexType);
state = State.READ_HEADER_REPRESENTATION;
break;
case SKIP_LITERAL_HEADER_VALUE:
- valueLength -= in.skip(valueLength);
-
- if (valueLength == 0) {
- state = State.READ_HEADER_REPRESENTATION;
- }
+ skipFully(in, valueLength);
+ valueLength = 0;
+ state = State.READ_HEADER_REPRESENTATION;
break;
default:
@@ -546,19 +525,39 @@ public final class Decoder {
return StringTools.asciiStringFromBytes(result);
}
+ private static byte readByte(InputStream in) throws IOException {
+ int b = in.read();
+ if (b < 0) {
+ // the header block stopped in the middle of a representation
+ throw DECOMPRESSION_EXCEPTION;
+ }
+ return (byte) b;
+ }
+
+ private static void skipFully(InputStream in, int length) throws IOException
{
+ long remaining = length;
+ while (remaining > 0) {
+ long skipped = in.skip(remaining);
+ if (skipped <= 0) {
+ // skip() is free to skip nothing; read a byte to make progress and to
notice the end
+ if (in.read() < 0) {
+ throw DECOMPRESSION_EXCEPTION;
+ }
+ remaining--;
+ } else {
+ remaining -= skipped;
+ }
+ }
+ }
+
// Unsigned Little Endian Base 128 Variable-Length Integer Encoding
private static int decodeULE128(InputStream in) throws IOException {
- in.mark(5);
int result = 0;
int shift = 0;
while (shift < 32) {
- if (in.available() == 0) {
- // Buffer does not contain entire integer,
- // reset reader index and return -1.
- in.reset();
- return -1;
- }
- byte b = (byte) in.read();
+ // reading forward only: a block that ends inside an integer is
truncated, which used to be
+ // rewound with mark/reset and reported to the caller as "come back with
more data"
+ byte b = readByte(in);
if (shift == 28 && (b & 0xF8) != 0) {
break;
}
@@ -569,7 +568,6 @@ public final class Decoder {
shift += 7;
}
// Value exceeds Integer.MAX_VALUE
- in.reset();
throw DECOMPRESSION_EXCEPTION;
}
}
diff --git
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/hpack/HeaderDecompression.scala
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/hpack/HeaderDecompression.scala
index 6aad6e085..d51dd85e8 100644
---
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/hpack/HeaderDecompression.scala
+++
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/hpack/HeaderDecompression.scala
@@ -21,7 +21,7 @@ import pekko.http.impl.engine.http2.Http2Protocol.ErrorCode
import pekko.http.impl.engine.http2.RequestParsing.parseHeaderPair
import pekko.http.impl.engine.http2._
import pekko.http.impl.engine.parsing.HttpHeaderParser
-import pekko.http.scaladsl.model.ParsingException
+import pekko.http.scaladsl.model.{ ErrorInfo, ParsingException }
import pekko.http.scaladsl.settings.ParserSettings
import pekko.http.shaded.com.twitter.hpack.HeaderListener
import pekko.stream._
@@ -73,8 +73,14 @@ private[http2] final class
HeaderDecompression(masterHeaderParser: HttpHeaderPar
def parseAndEmit(
streamId: Int, endStream: Boolean, payload: ByteString, prioInfo:
Option[PriorityFrame]): Unit = {
val headers = new VectorBuilder[(String, AnyRef)]
+ // A header that fails to parse must not unwind out of the decoder.
Decoding has to run to the end of
+ // the block so that the HPACK dynamic table keeps tracking the peer's
- insertHeader calls this
+ // listener before adding to the table, and the representations after
this one would not be read at
+ // all - and so that endHeaderBlock resets the state machine. Both
would otherwise stay wrong for
+ // every later HEADERS frame on the connection. Remember the first
failure and report it afterwards.
+ var parsingError: Option[ErrorInfo] = None
object Receiver extends HeaderListener {
- def addHeader(name: String, value: String, parsed: AnyRef,
sensitive: Boolean): AnyRef = {
+ def addHeader(name: String, value: String, parsed: AnyRef,
sensitive: Boolean): AnyRef = try {
if (parsed ne null) {
headers += name -> parsed
parsed
@@ -104,25 +110,45 @@ private[http2] final class
HeaderDecompression(masterHeaderParser: HttpHeaderPar
handle(header)
}
}
+ } catch {
+ case ex: ParsingException =>
+ if (parsingError.isEmpty) parsingError = Some(ex.info)
+ // nothing usable to cache against the table entry, so the value
is parsed again if it is
+ // referenced again - and fails again, consistently
+ null
}
}
- val stream = payload.compact.asInputStream
+ // no compact() needed: the decoder only reads forward, so the
SequenceInputStream that a
+ // multi-fragment ByteString hands out is enough and the header block
is not copied
+ val stream = payload.asInputStream
try {
- decoder.decode(stream, Receiver) // only compact ByteString supports
InputStream with mark/reset
+ decoder.decode(stream, Receiver)
// the decoder stops emitting headers as soon as the limit is
exceeded and reports that here
val truncated = decoder.endHeaderBlock()
if (truncated) headerListSizeExceeded(streamId)
- else push(eventsOut, ParsedHeadersFrame(streamId, endStream,
headers.result(), prioInfo, None))
+ else
+ parsingError match {
+ // push details further and let RequestErrorFlow handle
responding with bad request
+ case Some(info) =>
+ push(eventsOut, ParsedHeadersFrame(streamId, endStream,
Seq.empty, prioInfo, Some(info)))
+ case None =>
+ push(eventsOut, ParsedHeadersFrame(streamId, endStream,
headers.result(), prioInfo, None))
+ }
} catch {
case ex: ParsingException =>
- // push details further and let RequestErrorFlow handle responding
with bad request
+ // not expected any more now that the listener catches them, kept
so that one thrown from
+ // somewhere else still answers with a bad request rather than
tearing down the connection
push(eventsOut, ParsedHeadersFrame(streamId, endStream, Seq.empty,
prioInfo, Some(ex.info)))
case _: IOException =>
// this is signalled by the decoder when it failed, we want to
react to this by rendering a GOAWAY frame
fail(eventsOut,
new
Http2Compliance.Http2ProtocolException(ErrorCode.COMPRESSION_ERROR,
"Decompression failed."))
} finally {
+ // endHeaderBlock is what resets the decoder for the next block, so
it has to run even when decode
+ // unwound anyway - a malformed pseudo header raises an
Http2ProtocolException, for one. Running it
+ // a second time after the call above is a no-op.
+ decoder.endHeaderBlock()
stream.close()
}
}
diff --git
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/http2/hpack/HpackDecoderSpec.scala
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/http2/hpack/HpackDecoderSpec.scala
index fa5643797..2e0d90711 100644
---
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/http2/hpack/HpackDecoderSpec.scala
+++
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/http2/hpack/HpackDecoderSpec.scala
@@ -17,7 +17,8 @@
package org.apache.pekko.http.impl.engine.http2.hpack
-import java.io.{ ByteArrayInputStream, ByteArrayOutputStream, InputStream }
+import java.io.{ ByteArrayInputStream, ByteArrayOutputStream, IOException,
InputStream, SequenceInputStream }
+import scala.jdk.CollectionConverters._
import scala.collection.mutable.ListBuffer
@@ -47,6 +48,14 @@ class HpackDecoderSpec extends AnyWordSpec with Matchers {
override def skip(n: Long): Long = underlying.skip(n)
}
+ /**
+ * The stream a multi-fragment `ByteString` hands out: no mark/reset, and
`available()` only ever
+ * reports what is left in the current chunk rather than the whole block.
+ */
+ private def chunkedStream(bytes: Array[Byte], chunkSize: Int): InputStream =
+ new SequenceInputStream(
+ bytes.grouped(chunkSize).map(chunk => new ByteArrayInputStream(chunk):
InputStream).asJavaEnumeration)
+
private def encode(headers: (String, String)*): Array[Byte] = {
val out = new ByteArrayOutputStream
val encoder = new Encoder(maxHeaderTableSize)
@@ -83,5 +92,29 @@ class HpackDecoderSpec extends AnyWordSpec with Matchers {
"decode a header block from a stream that returns a few bytes per read" in
{
decode(new TricklingInputStream(encode(headers: _*), bytesPerRead = 7))
shouldEqual headers
}
+
+ "decode a header block from a stream that supports neither mark/reset nor
a whole-block available()" in {
+ val stream = chunkedStream(encode(headers: _*), chunkSize = 4)
+ stream.markSupported() shouldEqual false
+ decode(stream) shouldEqual headers
+ }
+
+ "decode a header block split so that a string literal straddles a chunk
boundary" in {
+ decode(chunkedStream(encode(headers: _*), chunkSize = 3)) shouldEqual
headers
+ }
+
+ "decode a header block split into single-byte chunks" in {
+ decode(chunkedStream(encode(headers: _*), chunkSize = 1)) shouldEqual
headers
+ }
+
+ "report a block that ends in the middle of a string literal as a
decompression failure" in {
+ val full = encode(headers: _*)
+ a[IOException] should be thrownBy decode(new
ByteArrayInputStream(full.dropRight(5)))
+ }
+
+ "report a block that ends in the middle of a length prefix as a
decompression failure" in {
+ // a literal header field with incremental indexing, new name, whose
name length never arrives
+ a[IOException] should be thrownBy decode(new
ByteArrayInputStream(Array[Byte](0x40)))
+ }
}
}
diff --git
a/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ClientServerSpec.scala
b/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ClientServerSpec.scala
index 6f718ec56..eb3d2d721 100644
---
a/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ClientServerSpec.scala
+++
b/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ClientServerSpec.scala
@@ -146,6 +146,22 @@ class Http2ClientServerSpec extends
PekkoSpecWithMaterializer(
response.status should be(StatusCodes.BadRequest)
}
+ "keep the connection usable after a header parsing failure" in new
TestSetup {
+ sendClientRequest(HttpRequest(
+ method = HttpMethod.custom("UNKNOWN_TO_SERVER"),
+ uri = "http://www.example.com/test";).addAttribute(requestIdAttr,
RequestId("bad")))
+ expectClientResponse().status should be(StatusCodes.BadRequest)
+
+ // the failing header must not have left the HPACK dynamic table out of
step with the client's, nor the
+ // decoder's state machine part way through the previous block
+ sendClientRequest(
+ HttpRequest(uri =
"http://www.example.com/afterwards";).addAttribute(requestIdAttr,
RequestId("good")))
+ val serverRequest = expectServerRequest()
+ serverRequest.request.uri.path.toString shouldBe "/afterwards"
+ serverRequest.sendResponse(HttpResponse(entity = "pong"))
+ expectClientResponse().status should be(StatusCodes.OK)
+ }
+
"return internal server error when handler future fails" in new TestSetup {
sendClientRequest()
val serverRequest = expectServerRequest()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
