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.git
The following commit(s) were added to refs/heads/main by this push:
new fcc340ce9a fix: bound inbound Artery TCP frame length at framing
(#3492)
fcc340ce9a is described below
commit fcc340ce9a271fdcadda1f425024f4d60ae65eb0
Author: PJ Fanning <[email protected]>
AuthorDate: Mon Aug 31 12:34:20 2026 +0100
fix: bound inbound Artery TCP frame length at framing (#3492)
Motivation:
TcpFraming read the 4-byte frame length from the wire and passed it straight
to reader.take / ByteBuffer allocation with no bound. A peer -
unauthenticated
on the default tcp transport - could declare an oversized or negative frame
and
drive a large allocation. Because the decoder and deserializer stages
downstream
of the MergeHub are shared by every inbound connection, an OutOfMemoryError
there
is fatal to the shared stream and, after inbound-max-restarts, terminates
the
whole ActorSystem - turning one malformed connection into node loss.
Modification:
Pass the configured maximum-frame-size and maximum-large-frame-size into
TcpFraming and reject a frame length that is negative or exceeds the
maximum for
the connection's stream, before any data is buffered. The large-message
stream
keeps its larger bound. Rejection is a FramingException, which tears down
only
that connection.
Result:
A malformed or oversized frame is rejected per-connection instead of
allocating
without limit and risking a fatal error on the shared inbound stream.
Tests:
- sbt "remote/testOnly org.apache.pekko.remote.artery.tcp.TcpFramingSpec" -
14 passed, incl. oversized, negative and at-limit frame cases
- sbt "remote/testOnly
org.apache.pekko.remote.artery.LargeMessagesStreamSpec" - 4 passed (legitimate
large frames still delivered)
- sbt "remote/mimaReportBinaryIssues" - no issues (TcpFraming is
@InternalApi; new params are defaulted)
References:
None - found while reviewing the draft threat model in #3478
---
.../remote/artery/tcp/ArteryTcpTransport.scala | 3 ++-
.../pekko/remote/artery/tcp/TcpFraming.scala | 26 ++++++++++++++++++----
.../pekko/remote/artery/tcp/TcpFramingSpec.scala | 22 ++++++++++++++++++
3 files changed, 46 insertions(+), 5 deletions(-)
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/artery/tcp/ArteryTcpTransport.scala
b/remote/src/main/scala/org/apache/pekko/remote/artery/tcp/ArteryTcpTransport.scala
index 55ebafa7cd..1d168ee06c 100644
---
a/remote/src/main/scala/org/apache/pekko/remote/artery/tcp/ArteryTcpTransport.scala
+++
b/remote/src/main/scala/org/apache/pekko/remote/artery/tcp/ArteryTcpTransport.scala
@@ -355,7 +355,8 @@ private[remote] class ArteryTcpTransport(
Flow[ByteString]
.via(inboundKillSwitch.flow)
// must create new FlightRecorder event sink for each connection
because they can't be shared
- .via(new TcpFraming(settings.Advanced.TcpMagicValues, flightRecorder))
+ .via(new TcpFraming(settings.Advanced.TcpMagicValues, flightRecorder,
+ settings.Advanced.MaximumFrameSize,
settings.Advanced.MaximumLargeFrameSize))
.alsoTo(inboundStream)
.filter(_ => false) // don't send back anything in this TCP socket
.map(_ => ByteString.empty) // make it a Flow[ByteString] again
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/artery/tcp/TcpFraming.scala
b/remote/src/main/scala/org/apache/pekko/remote/artery/tcp/TcpFraming.scala
index e69fa91c79..c495688c98 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/tcp/TcpFraming.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/tcp/TcpFraming.scala
@@ -79,7 +79,9 @@ import pekko.util.ByteString
*/
@InternalApi private[pekko] class TcpFraming(
acceptedMagic: immutable.Seq[ByteString] = List(TcpFraming.DefaultMagic),
- flightRecorder: RemotingFlightRecorder = NoOpRemotingFlightRecorder)
+ flightRecorder: RemotingFlightRecorder = NoOpRemotingFlightRecorder,
+ maximumFrameSize: Int = Int.MaxValue,
+ maximumLargeFrameSize: Int = Int.MaxValue)
extends ByteStringParser[EnvelopeBuffer] {
private val magicLength = acceptedMagic.head.length
@@ -102,15 +104,31 @@ import pekko.util.ByteString
}
}
case object ReadStreamId extends Step {
- override def parse(reader: ByteReader): ParseResult[EnvelopeBuffer] =
- ParseResult(None, ReadFrame(reader.readByte()))
+ override def parse(reader: ByteReader): ParseResult[EnvelopeBuffer] = {
+ val streamId = reader.readByte()
+ // A connection carries a single stream for its lifetime; the
large-message
+ // stream is allowed bigger frames than the control and ordinary
streams.
+ val maxFrameSize =
+ if (streamId == ArteryTransport.LargeStreamId) maximumLargeFrameSize
else maximumFrameSize
+ ParseResult(None, ReadFrame(streamId, maxFrameSize))
+ }
}
- case class ReadFrame(streamId: Int) extends Step {
+ case class ReadFrame(streamId: Int, maxFrameSize: Int) extends Step {
override def onTruncation(): Unit =
failStage(new FramingException("Stream finished but there was a
truncated final frame in the buffer"))
override def parse(reader: ByteReader): ParseResult[EnvelopeBuffer] = {
val frameLength = reader.readIntLE()
+ // frameLength is read from the wire before any data is buffered;
reject an
+ // out-of-range value here so a peer cannot drive a huge allocation
(which
+ // would be a fatal OutOfMemoryError on the shared inbound stream) by
+ // declaring an oversized or negative frame. FramingException tears
down
+ // only this connection.
+ if (frameLength < 0 || frameLength > maxFrameSize)
+ throw new FramingException(
+ s"Frame length [$frameLength] for stream [$streamId] is out of
range, " +
+ s"must be between 0 and the maximum frame size [$maxFrameSize]. " +
+ "Connection is rejected.")
val buffer = createBuffer(reader.take(frameLength))
ParseResult(Some(buffer), this)
}
diff --git
a/remote/src/test/scala/org/apache/pekko/remote/artery/tcp/TcpFramingSpec.scala
b/remote/src/test/scala/org/apache/pekko/remote/artery/tcp/TcpFramingSpec.scala
index 49c0efc588..795f50876d 100644
---
a/remote/src/test/scala/org/apache/pekko/remote/artery/tcp/TcpFramingSpec.scala
+++
b/remote/src/test/scala/org/apache/pekko/remote/artery/tcp/TcpFramingSpec.scala
@@ -33,6 +33,9 @@ class TcpFramingSpec extends PekkoSpec("""
private val magic = TcpFraming.DefaultMagic
private val acceptedMagic = List(magic, TcpFraming.LegacyMagic)
private val framingFlow = Flow[ByteString].via(new TcpFraming(acceptedMagic))
+ private val maxFrameSize = 256 * 1024
+ private val boundedFramingFlow =
+ Flow[ByteString].via(new TcpFraming(acceptedMagic, maximumFrameSize =
maxFrameSize))
private val payload5 = ByteString((1 to 5).map(_.toByte).toArray)
@@ -101,6 +104,25 @@ class TcpFramingSpec extends PekkoSpec("""
}
}
+ "reject a frame that exceeds the maximum frame size" in {
+ val bytes = TcpFraming.encodeConnectionHeader(magic, 2) ++
encodeFrameHeader(maxFrameSize + 1)
+ val fail =
Source(List(bytes)).via(boundedFramingFlow).runWith(Sink.seq).failed.futureValue
+ fail shouldBe a[FramingException]
+ }
+
+ "reject a frame with a negative declared length" in {
+ val bytes = TcpFraming.encodeConnectionHeader(magic, 2) ++
encodeFrameHeader(-1)
+ val fail =
Source(List(bytes)).via(boundedFramingFlow).runWith(Sink.seq).failed.futureValue
+ fail shouldBe a[FramingException]
+ }
+
+ "accept a frame at exactly the maximum frame size" in {
+ val payload = ByteString(Array.fill(maxFrameSize)(7.toByte))
+ val bytes = TcpFraming.encodeConnectionHeader(magic, 2) ++
encodeFrameHeader(maxFrameSize) ++ payload
+ val frames =
Source(List(bytes)).via(boundedFramingFlow).runWith(Sink.seq).futureValue
+ frames.head.byteBuffer.limit() should ===(maxFrameSize)
+ }
+
"report truncated frames" in {
val bytes = TcpFraming.encodeConnectionHeader(magic, 3) ++
frameBytes(3).drop(1)
Source(List(bytes)).via(framingFlow).runWith(Sink.seq).failed.futureValue
shouldBe a[FramingException]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
