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-grpc.git
The following commit(s) were added to refs/heads/main by this push:
new faabc60a trailer frame support in GrpcProtocol appears broken (#822)
faabc60a is described below
commit faabc60a155d660a4ac748f2a1bf5c0ea9e3a8f9
Author: PJ Fanning <[email protected]>
AuthorDate: Sat Aug 8 22:04:24 2026 +0100
trailer frame support in GrpcProtocol appears broken (#822)
* trailer frame support in GrpcProtocol appears broken
* Update GrpcProtocolWebSpec.scala
* support LF only splits in trailer
* Update GrpcProtocolWeb.scala
* Update GrpcProtocolWebSpec.scala
* test issue
* remove return from decodeTrailer
---
.../pekko/grpc/internal/AbstractGrpcProtocol.scala | 3 +-
.../pekko/grpc/internal/GrpcProtocolWeb.scala | 46 +++-
.../pekko/grpc/internal/GrpcProtocolWebSpec.scala | 287 +++++++++++++++++++++
3 files changed, 330 insertions(+), 6 deletions(-)
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
index 8cd4653c..3067c894 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
@@ -134,6 +134,7 @@ object AbstractGrpcProtocol {
val reader = new ByteReader(strictAdapter(bs))
val frameType = reader.readByte()
val length = reader.readIntBE()
+ if (length < 0) throw new IllegalStateException(s"Frame length must
not be negative, was $length")
val data = reader.take(length)
if (reader.hasRemaining) throw new IllegalStateException("Unexpected
data")
if ((frameType & 0x80) == 0) codec.uncompress((frameType & 1) == 1,
data)
@@ -153,8 +154,8 @@ object AbstractGrpcProtocol {
object ReadFrameHeader extends Step {
override def parse(reader: ByteReader): ParseResult[Frame] = {
val frameType = reader.readByte()
- // If we want to support > 2GB frames, this should be unsigned
val length = reader.readIntBE()
+ if (length < 0) throw new IllegalStateException(s"Frame length
must not be negative, was $length")
if (length == 0) ParseResult(Some(deframe(frameType,
ByteString.empty)), ReadFrameHeader)
else ParseResult(None, ReadFrame(frameType, length),
acceptUpstreamFinish = false)
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolWeb.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolWeb.scala
index 1995b711..1912ae77 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolWeb.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolWeb.scala
@@ -18,6 +18,7 @@ import pekko.NotUsed
import pekko.grpc.GrpcProtocol._
import pekko.http.scaladsl.model._
import pekko.http.scaladsl.model.HttpEntity.{ Chunk, ChunkStreamPart }
+import pekko.http.scaladsl.model.headers.RawHeader
import pekko.stream.scaladsl.Flow
import pekko.util.{ ByteString, ByteStringBuilder }
import io.grpc.{ Status, StatusException }
@@ -64,10 +65,10 @@ abstract class GrpcProtocolWebBase(subType: String) extends
AbstractGrpcProtocol
}
private final def decodeFrame(frameHeader: Int, data: ByteString): Frame = {
- (frameHeader & 80) match {
- case 0 => DataFrame(data)
- case 1 => TrailerFrame(decodeTrailer(data))
- case f => throw new
StatusException(Status.INTERNAL.withDescription(s"Unknown frame type [$f]"))
+ (frameHeader & 0x80) match {
+ case 0 => DataFrame(data)
+ case 0x80 => TrailerFrame(decodeTrailer(data))
+ case f => throw new
StatusException(Status.INTERNAL.withDescription(s"Unknown frame type [$f]"))
}
}
@@ -80,7 +81,42 @@ abstract class GrpcProtocolWebBase(subType: String) extends
AbstractGrpcProtocol
builder.result()
}
- private final def decodeTrailer(data: ByteString): List[HttpHeader] = ???
+ private final def decodeTrailer(data: ByteString): List[HttpHeader] = {
+ val str = data.utf8String
+ val len = str.length
+ val headers = List.newBuilder[HttpHeader]
+ var i = 0
+ while (i < len) {
+ // skip leading whitespace and blank lines
+ while (i < len &&
+ (str.charAt(i) == ' ' || str.charAt(i) == '\t' || str.charAt(i) ==
'\r' || str.charAt(i) == '\n'))
+ i += 1
+ if (i < len) {
+ // scan for colon (key:value separator), stopping at LF for malformed
lines
+ val keyStart = i
+ while (i < len && str.charAt(i) != ':' && str.charAt(i) != '\n') i += 1
+ if (i >= len || str.charAt(i) == '\n') {
+ // no colon found before end-of-line — skip malformed line
+ if (i < len) i += 1 // skip LF
+ } else {
+ val keyEnd = i
+ i += 1 // skip ':'
+ // scan for LF (line terminator)
+ val valueStart = i
+ while (i < len && str.charAt(i) != '\n') i += 1
+ var valueEnd = i
+ // strip trailing CR
+ if (valueEnd > valueStart && str.charAt(valueEnd - 1) == '\r')
valueEnd -= 1
+ i += 1 // skip LF
+ // trim and emit
+ val key = str.substring(keyStart, keyEnd).trim
+ val value = str.substring(valueStart, valueEnd).trim
+ if (key.nonEmpty) headers += RawHeader(key, value)
+ }
+ }
+ }
+ headers.result()
+ }
}
diff --git
a/runtime/src/test/scala/org/apache/pekko/grpc/internal/GrpcProtocolWebSpec.scala
b/runtime/src/test/scala/org/apache/pekko/grpc/internal/GrpcProtocolWebSpec.scala
new file mode 100644
index 00000000..670c08c5
--- /dev/null
+++
b/runtime/src/test/scala/org/apache/pekko/grpc/internal/GrpcProtocolWebSpec.scala
@@ -0,0 +1,287 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.grpc.internal
+
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.grpc.GrpcProtocol.{ DataFrame, Frame, TrailerFrame }
+import pekko.http.scaladsl.model.HttpHeader
+import pekko.http.scaladsl.model.headers.RawHeader
+import pekko.stream.scaladsl.Source
+import pekko.stream.testkit.scaladsl.TestSink
+import pekko.testkit.TestKit
+import pekko.util.ByteString
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpecLike
+
+class GrpcProtocolWebSpec extends TestKit(ActorSystem()) with AnyWordSpecLike
with Matchers {
+
+ val reader = GrpcProtocolWeb.newReader(Identity)
+ val writer = GrpcProtocolWeb.newWriter(Identity)
+
+ /** Construct a raw trailer frame: 1 byte flags (0x80) + 4 bytes length +
data */
+ private def trailerFrameBytes(data: ByteString): ByteString = {
+ val header = new Array[Byte](5)
+ header(0) = 0x80.toByte
+ header(1) = (data.length >>> 24).toByte
+ header(2) = (data.length >>> 16).toByte
+ header(3) = (data.length >>> 8).toByte
+ header(4) = data.length.toByte
+ ByteString.fromArrayUnsafe(header, 0, 5) ++ data
+ }
+
+ "GrpcProtocolWeb" should {
+
+ "encode and decode a data frame" in {
+ val data = ByteString(Array[Byte](1, 2, 3, 4))
+ val frame = DataFrame(data)
+ val chunk = writer.encodeFrame(frame)
+
+ Source
+ .single(chunk.data)
+ .via(reader.frameDecoder)
+ .runWith(TestSink[Frame]())
+ .request(1)
+ .expectNext(frame)
+ .expectComplete()
+ }
+
+ "encode and decode a trailer frame" in {
+ val trailers = List[HttpHeader](
+ RawHeader("grpc-status", "0"),
+ RawHeader("grpc-message", ""))
+ val frame = TrailerFrame(trailers)
+ val chunk = writer.encodeFrame(frame)
+
+ val probe = Source
+ .single(chunk.data)
+ .via(reader.frameDecoder)
+ .runWith(TestSink[Frame]())
+ .request(1)
+
+ probe.expectNext() match {
+ case TrailerFrame(decoded) =>
+ (decoded should have).length(2)
+ decoded.head shouldBe RawHeader("grpc-status", "0")
+ decoded(1) shouldBe RawHeader("grpc-message", "")
+ case other => fail(s"Expected TrailerFrame, got $other")
+ }
+ probe.expectComplete()
+ }
+
+ "distinguish data frames from trailer frames by type bit" in {
+ val data = ByteString(Array[Byte](1, 2, 3))
+ val dataFrame = DataFrame(data)
+ val trailers = List[HttpHeader](RawHeader("grpc-status", "0"))
+ val trailerFrame = TrailerFrame(trailers)
+
+ val encodedData = writer.encodeFrame(dataFrame)
+ val encodedTrailer = writer.encodeFrame(trailerFrame)
+
+ val probe = Source(List(encodedData.data, encodedTrailer.data))
+ .via(reader.frameDecoder)
+ .runWith(TestSink[Frame]())
+ .request(2)
+
+ probe.expectNext(dataFrame)
+ probe.expectNext() match {
+ case TrailerFrame(decoded) =>
+ (decoded should have).length(1)
+ decoded.head shouldBe RawHeader("grpc-status", "0")
+ case other => fail(s"Expected TrailerFrame, got $other")
+ }
+ probe.expectComplete()
+ }
+
+ "decode trailer with LF-only line endings" in {
+ val trailerData = ByteString("grpc-status:0\ngrpc-message:ok\n")
+ val rawFrame = trailerFrameBytes(trailerData)
+
+ val probe = Source
+ .single(rawFrame)
+ .via(reader.frameDecoder)
+ .runWith(TestSink[Frame]())
+ .request(1)
+
+ probe.expectNext() match {
+ case TrailerFrame(decoded) =>
+ (decoded should have).length(2)
+ decoded.head shouldBe RawHeader("grpc-status", "0")
+ decoded(1) shouldBe RawHeader("grpc-message", "ok")
+ case other => fail(s"Expected TrailerFrame, got $other")
+ }
+ probe.expectComplete()
+ }
+
+ "decode trailer with extra whitespace around key and value" in {
+ val trailerData = ByteString(" grpc-status : 0 \r\ngrpc-message : ok
\r\n")
+ val rawFrame = trailerFrameBytes(trailerData)
+
+ val probe = Source
+ .single(rawFrame)
+ .via(reader.frameDecoder)
+ .runWith(TestSink[Frame]())
+ .request(1)
+
+ probe.expectNext() match {
+ case TrailerFrame(decoded) =>
+ (decoded should have).length(2)
+ decoded.head shouldBe RawHeader("grpc-status", "0")
+ decoded(1) shouldBe RawHeader("grpc-message", "ok")
+ case other => fail(s"Expected TrailerFrame, got $other")
+ }
+ probe.expectComplete()
+ }
+
+ "decode trailer with empty value" in {
+ val trailerData = ByteString("grpc-status:0\r\ngrpc-message:\r\n")
+ val rawFrame = trailerFrameBytes(trailerData)
+
+ val probe = Source
+ .single(rawFrame)
+ .via(reader.frameDecoder)
+ .runWith(TestSink[Frame]())
+ .request(1)
+
+ probe.expectNext() match {
+ case TrailerFrame(decoded) =>
+ (decoded should have).length(2)
+ decoded.head shouldBe RawHeader("grpc-status", "0")
+ decoded(1) shouldBe RawHeader("grpc-message", "")
+ case other => fail(s"Expected TrailerFrame, got $other")
+ }
+ probe.expectComplete()
+ }
+
+ "decode trailer with multiple colons in value" in {
+ val trailerData = ByteString("grpc-status:0\r\ncustom-header: some:
value\r\n")
+ val rawFrame = trailerFrameBytes(trailerData)
+
+ val probe = Source
+ .single(rawFrame)
+ .via(reader.frameDecoder)
+ .runWith(TestSink[Frame]())
+ .request(1)
+
+ probe.expectNext() match {
+ case TrailerFrame(decoded) =>
+ (decoded should have).length(2)
+ decoded.head shouldBe RawHeader("grpc-status", "0")
+ decoded(1) shouldBe RawHeader("custom-header", "some: value")
+ case other => fail(s"Expected TrailerFrame, got $other")
+ }
+ probe.expectComplete()
+ }
+
+ "decode trailer with blank lines between entries" in {
+ val trailerData =
ByteString("grpc-status:0\r\n\r\n\r\ngrpc-message:ok\r\n")
+ val rawFrame = trailerFrameBytes(trailerData)
+
+ val probe = Source
+ .single(rawFrame)
+ .via(reader.frameDecoder)
+ .runWith(TestSink[Frame]())
+ .request(1)
+
+ probe.expectNext() match {
+ case TrailerFrame(decoded) =>
+ (decoded should have).length(2)
+ decoded.head shouldBe RawHeader("grpc-status", "0")
+ decoded(1) shouldBe RawHeader("grpc-message", "ok")
+ case other => fail(s"Expected TrailerFrame, got $other")
+ }
+ probe.expectComplete()
+ }
+
+ "skip malformed lines with no colon in trailer" in {
+ val trailerData = ByteString("grpc-status:0\r\nmalformed
line\r\ngrpc-message:ok\r\n")
+ val rawFrame = trailerFrameBytes(trailerData)
+
+ val probe = Source
+ .single(rawFrame)
+ .via(reader.frameDecoder)
+ .runWith(TestSink[Frame]())
+ .request(1)
+
+ probe.expectNext() match {
+ case TrailerFrame(decoded) =>
+ (decoded should have).length(2)
+ decoded.head shouldBe RawHeader("grpc-status", "0")
+ decoded(1) shouldBe RawHeader("grpc-message", "ok")
+ case other => fail(s"Expected TrailerFrame, got $other")
+ }
+ probe.expectComplete()
+ }
+
+ "decode empty trailer frame" in {
+ val trailerData = ByteString.empty
+ val rawFrame = trailerFrameBytes(trailerData)
+
+ val probe = Source
+ .single(rawFrame)
+ .via(reader.frameDecoder)
+ .runWith(TestSink[Frame]())
+ .request(1)
+
+ probe.expectNext() match {
+ case TrailerFrame(decoded) =>
+ decoded shouldBe empty
+ case other => fail(s"Expected TrailerFrame, got $other")
+ }
+ probe.expectComplete()
+ }
+
+ "decode trailer with no trailing newline" in {
+ val trailerData = ByteString("grpc-status:0")
+ val rawFrame = trailerFrameBytes(trailerData)
+
+ val probe = Source
+ .single(rawFrame)
+ .via(reader.frameDecoder)
+ .runWith(TestSink[Frame]())
+ .request(1)
+
+ probe.expectNext() match {
+ case TrailerFrame(decoded) =>
+ (decoded should have).length(1)
+ decoded.head shouldBe RawHeader("grpc-status", "0")
+ case other => fail(s"Expected TrailerFrame, got $other")
+ }
+ probe.expectComplete()
+ }
+
+ "reject frame with negative length" in {
+ // Construct a raw frame with a negative length (high bit set in length
field)
+ // Frame format: 1 byte flags + 4 bytes length (big-endian) + data
+ val header = new Array[Byte](5)
+ header(0) = 0x00 // data frame, no compression
+ header(1) = 0x80.toByte // length byte 0 (sets sign bit -> negative int)
+ header(2) = 0x00
+ header(3) = 0x00
+ header(4) = 0x00
+ val rawFrame = ByteString.fromArrayUnsafe(header, 0, 5)
+
+ Source
+ .single(rawFrame)
+ .via(reader.frameDecoder)
+ .runWith(TestSink[Frame]())
+ .request(1)
+ .expectError()
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
