This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch 1.7.x
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/1.7.x by this push:
new 5923c0a81c Backport TcpFraming magic (#3444)
5923c0a81c is described below
commit 5923c0a81cb693c5330cdeb97636244d6df198f6
Author: PJ Fanning <[email protected]>
AuthorDate: Sun Aug 16 12:33:55 2026 +0100
Backport TcpFraming magic (#3444)
* support PEKK TCP magic (#3425)
* support PEKK TCP magic
* scalafmt
* change 1.x default
* Update ArterySettings.scala
* try to fix issues
---
.../src/main/paradox/additional/rolling-updates.md | 18 ++++++++
docs/src/main/paradox/remoting-artery.md | 22 ++++++++++
remote/src/main/resources/reference.conf | 11 +++++
.../pekko/remote/artery/ArterySettings.scala | 28 +++++++++++--
.../remote/artery/tcp/ArteryTcpTransport.scala | 5 ++-
.../pekko/remote/artery/tcp/TcpFraming.scala | 30 +++++++++-----
.../pekko/remote/artery/tcp/TcpFramingSpec.scala | 48 +++++++++++++++++++---
7 files changed, 141 insertions(+), 21 deletions(-)
diff --git a/docs/src/main/paradox/additional/rolling-updates.md
b/docs/src/main/paradox/additional/rolling-updates.md
index 08687000eb..78986ab2bc 100644
--- a/docs/src/main/paradox/additional/rolling-updates.md
+++ b/docs/src/main/paradox/additional/rolling-updates.md
@@ -142,6 +142,24 @@ which has a completely different protocol, a rolling
update is not supported.
Rolling update is not supported when @ref:[changing the remoting
transport](../remoting-artery.md#selecting-a-transport).
+### Changing TCP magic header
+
+The TCP magic header (`pekko.remote.artery.advanced.tcp-magic`) is used to
validate connections between nodes.
+It is an array of allowed values. The first value is used when sending
(outbound connections); all values
+are accepted when receiving (inbound connections).
+
+The magic header has evolved across versions:
+
+ * Akka and Pekko up to 1.6.x only support `"AKKA"` as the magic header.
+ * Pekko 1.7.x sends `"AKKA"` but accepts both `"AKKA"` and `"PEKK"`, enabling
future upgrades.
+ * Pekko 2.x (and above) sends `"PEKK"` by default but accepts both `"PEKK"`
and `"AKKA"`.
+
+Because Pekko 1.7.x+ and 2.x accept both values by default, rolling upgrades
between these versions
+do not require changing the `tcp-magic` configuration. Upgrading from Pekko
1.6.x or earlier to 2.x
+directly is also supported since the 2.x default accepts `"AKKA"`.
+
+If you remove `"AKKA"` from the array, nodes running older versions will be
unable to connect.
+
### Migrating from Classic Sharding to Typed Sharding
If you have been using classic sharding it is possible to do a rolling update
to typed sharding using a 3 step procedure.
diff --git a/docs/src/main/paradox/remoting-artery.md
b/docs/src/main/paradox/remoting-artery.md
index 25495a9d59..83417fd70a 100644
--- a/docs/src/main/paradox/remoting-artery.md
+++ b/docs/src/main/paradox/remoting-artery.md
@@ -153,6 +153,28 @@ officially supported. If you're on a Big Endian processor,
such as Sparc, it is
@@@
+### TCP Magic Header
+
+When using the `tcp` or `tls-tcp` transport, a 4-byte "magic header" is sent
at the start of each connection.
+This header is used to detect and reject accidental or invalid connections.
+
+The magic header is configured by `pekko.remote.artery.advanced.tcp-magic`,
which is an array of allowed values.
+The first value in the array is used when sending (outbound connections). All
values are accepted when
+receiving (inbound connections). The default is `["PEKK", "AKKA"]`.
+
+Each value must produce at least 4 UTF-8 bytes; extra bytes are ignored.
Non-ASCII characters may occupy
+multiple UTF-8 bytes (2-4 bytes each).
+
+The magic header has evolved across versions:
+
+ * Akka and Pekko up to 1.6.x only support `"AKKA"` as the magic header.
+ * Pekko 1.7.x sends `"AKKA"` but accepts both `"AKKA"` and `"PEKK"`, enabling
future upgrades.
+ * Pekko 2.x (and above) sends `"PEKK"` by default but accepts both `"PEKK"`
and `"AKKA"`.
+
+Because Pekko 1.7.x+ and 2.x accept both values by default, rolling upgrades
between these versions
+do not require changing the `tcp-magic` configuration. Once all nodes are
running Pekko 2.x, you may
+remove `"AKKA"` from the array if desired.
+
## Canonical address
In order for remoting to work properly, where each system can send messages to
any other system on the same network
diff --git a/remote/src/main/resources/reference.conf
b/remote/src/main/resources/reference.conf
index 45621ff755..3aca023281 100644
--- a/remote/src/main/resources/reference.conf
+++ b/remote/src/main/resources/reference.conf
@@ -900,6 +900,17 @@ pekko {
# collected, which is not as efficient as reusing buffers in the pool.
large-buffer-pool-size = 32
+ # The 4-byte magic header sent at the start of each TCP/TLS connection.
+ # Used to detect and reject accidental/invalid connections.
+ # This is an array of allowed magic values. The first value is used
when
+ # sending (outbound connections). All values are accepted when
receiving
+ # (inbound connections).
+ # Each value must produce at least 4 UTF-8 bytes; extra bytes are
ignored.
+ # Non-ASCII characters may occupy multiple UTF-8 bytes (e.g. 2-4 bytes
each).
+ # Pekko 1.x uses "AKKA" as the default. To support rolling upgrades to
+ # Pekko 2.x, keep "AKKA" in the array alongside "PEKK".
+ tcp-magic = ["AKKA", "PEKK"]
+
# For enabling testing features, such as blackhole in
pekko-remote-testkit.
test-mode = off
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/artery/ArterySettings.scala
b/remote/src/main/scala/org/apache/pekko/remote/artery/ArterySettings.scala
index 012483f3c9..fd847b3fe2 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/ArterySettings.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/ArterySettings.scala
@@ -14,22 +14,24 @@
package org.apache.pekko.remote.artery
import java.net.InetAddress
-
-import scala.concurrent.duration._
+import java.nio.charset.StandardCharsets
import scala.annotation.nowarn
+import scala.collection.immutable
+import scala.concurrent.duration._
import com.typesafe.config.Config
import com.typesafe.config.ConfigFactory
import org.apache.pekko
import pekko.NotUsed
+import pekko.io.dns.internal.AsyncDnsResolver
import pekko.stream.ActorMaterializerSettings
+import pekko.util.ByteString
import pekko.util.Helpers.ConfigOps
import pekko.util.Helpers.Requiring
import pekko.util.Helpers.toRootLowerCase
import pekko.util.WildcardIndex
import pekko.util.ccompat.JavaConverters._
-import pekko.io.dns.internal.AsyncDnsResolver
/** INTERNAL API */
private[pekko] final class ArterySettings private (config: Config) {
@@ -117,6 +119,26 @@ private[pekko] final class ArterySettings private (config:
Config) {
import config._
val TestMode: Boolean = getBoolean("test-mode")
+ private val tcpMagicList: immutable.Seq[String] = {
+ val list = getStringList("tcp-magic").asScala.toIndexedSeq
+ require(list.nonEmpty, "tcp-magic must not be empty")
+ list
+ }
+ val TcpMagic: ByteString = {
+ val first = tcpMagicList.head
+ val bytes = ByteString(first.getBytes(StandardCharsets.UTF_8))
+ require(bytes.length >= 4,
+ s"tcp-magic value [$first] must produce at least 4 UTF-8 bytes, but
produced [${bytes.length}] bytes")
+ bytes.take(4)
+ }
+ val TcpMagicValues: Set[ByteString] = {
+ tcpMagicList.map { s =>
+ val bytes = ByteString(s.getBytes(StandardCharsets.UTF_8))
+ require(bytes.length >= 4,
+ s"tcp-magic value [$s] must produce at least 4 UTF-8 bytes, but
produced [${bytes.length}] bytes")
+ bytes.take(4)
+ }.toSet
+ }
val Dispatcher: String = getString("use-dispatcher")
val ControlStreamDispatcher: String =
getString("use-control-stream-dispatcher")
@nowarn("msg=deprecated")
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 aadfb60472..edeb6a6ca5 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
@@ -191,7 +191,8 @@ private[remote] class ArteryTcpTransport(
if (controlIdleKillSwitch.isDefined)
outboundContext.asInstanceOf[Association].setControlIdleKillSwitch(controlIdleKillSwitch)
-
Flow[ByteString].prepend(Source.single(TcpFraming.encodeConnectionHeader(streamId))).via(connectionFlow)
+
Flow[ByteString].prepend(Source.single(TcpFraming.encodeConnectionHeader(settings.Advanced.TcpMagic,
+ streamId))).via(connectionFlow)
}))
.mapError {
case ArteryTransport.ShutdownSignal =>
ArteryTransport.ShutdownSignal
@@ -357,7 +358,7 @@ 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(flightRecorder))
+ .via(new TcpFraming(settings.Advanced.TcpMagicValues, flightRecorder))
.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 fbf72fffb0..39da7b75c6 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
@@ -35,10 +35,16 @@ import pekko.util.ByteString
val Undefined = Int.MinValue
/**
- * The first 4 bytes of a new connection must be these `0x64 0x75 0x75 0x64`
(AKKA).
+ * The legacy 4-byte magic header from Akka (AKKA).
* The purpose of the "magic" is to detect and reject weird (accidental)
accesses.
*/
- val Magic = ByteString('A'.toByte, 'K'.toByte, 'K'.toByte, 'A'.toByte)
+ val DefaultMagic = ByteString('A'.toByte, 'K'.toByte, 'K'.toByte, 'A'.toByte)
+
+ /**
+ * The default 4-byte magic header for Pekko 2.x (PEKK).
+ * The purpose of the "magic" is to detect and reject weird (accidental)
accesses.
+ */
+ val PekkoMagic = ByteString('P'.toByte, 'E'.toByte, 'K'.toByte, 'K'.toByte)
/**
* When establishing the connection this header is sent first.
@@ -46,12 +52,12 @@ import pekko.util.ByteString
* inbound streams.
*
* The purpose of the "magic" is to detect and reject weird (accidental)
accesses.
- * The magic 4 bytes are `0x64 0x75 0x75 0x64` (AKKA).
+ * The magic 4 bytes are configurable via
`pekko.remote.artery.advanced.tcp-magic`.
*
- * The streamId` is encoded as 1 byte.
+ * The `streamId` is encoded as 1 byte.
*/
- def encodeConnectionHeader(streamId: Int): ByteString =
- Magic ++ ByteString.fromArrayUnsafe(Array(streamId.toByte))
+ def encodeConnectionHeader(magic: ByteString, streamId: Int): ByteString =
+ magic ++ ByteString.fromArrayUnsafe(Array(streamId.toByte))
/**
* Each frame starts with the frame header that contains the length
@@ -69,9 +75,13 @@ import pekko.util.ByteString
/**
* INTERNAL API
*/
-@InternalApi private[pekko] class TcpFraming(flightRecorder:
RemotingFlightRecorder = NoOpRemotingFlightRecorder)
+@InternalApi private[pekko] class TcpFraming(
+ acceptedMagic: Set[ByteString] = Set(TcpFraming.DefaultMagic),
+ flightRecorder: RemotingFlightRecorder = NoOpRemotingFlightRecorder)
extends ByteStringParser[EnvelopeBuffer] {
+ private val magicLength = acceptedMagic.head.length
+
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
new ParsingLogic {
abstract class Step extends ParseStep[EnvelopeBuffer]
@@ -79,13 +89,13 @@ import pekko.util.ByteString
case object ReadMagic extends Step {
override def parse(reader: ByteReader): ParseResult[EnvelopeBuffer] = {
- val magic = reader.take(TcpFraming.Magic.length)
- if (magic == TcpFraming.Magic)
+ val receivedMagic = reader.take(magicLength)
+ if (acceptedMagic.contains(receivedMagic))
ParseResult(None, ReadStreamId)
else
throw new FramingException(
"Stream didn't start with expected magic bytes, " +
- s"got [${(magic ++
reader.remainingData).take(10).map("%02x".format(_)).mkString(" ")}] " +
+ s"got [${(receivedMagic ++
reader.remainingData).take(10).map("%02x".format(_)).mkString(" ")}] " +
"Connection is rejected. Probably invalid accidental access.")
}
}
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 df171c7e49..39f5e2ee18 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
@@ -30,7 +30,9 @@ class TcpFramingSpec extends PekkoSpec("""
""") with ImplicitSender {
import TcpFraming.encodeFrameHeader
- private val framingFlow = Flow[ByteString].via(new TcpFraming)
+ private val magic = TcpFraming.DefaultMagic
+ private val acceptedMagic = Set(magic, TcpFraming.PekkoMagic)
+ private val framingFlow = Flow[ByteString].via(new TcpFraming(acceptedMagic))
private val payload5 = ByteString((1 to 5).map(_.toByte).toArray)
@@ -57,14 +59,15 @@ class TcpFramingSpec extends PekkoSpec("""
"TcpFraming stage" must {
"grab streamId from connection header" in {
- val bytes = TcpFraming.encodeConnectionHeader(2) ++ frameBytes(1)
+ val bytes = TcpFraming.encodeConnectionHeader(magic, 2) ++ frameBytes(1)
val frames =
Source(List(bytes)).via(framingFlow).runWith(Sink.seq).futureValue
frames.head.streamId should ===(2)
}
"grab streamId from connection header in single chunk" in {
val frames =
- Source(List(TcpFraming.encodeConnectionHeader(1),
frameBytes(1))).via(framingFlow).runWith(Sink.seq).futureValue
+ Source(List(TcpFraming.encodeConnectionHeader(magic, 1),
frameBytes(1))).via(framingFlow).runWith(
+ Sink.seq).futureValue
frames.head.streamId should ===(1)
}
@@ -75,7 +78,7 @@ class TcpFramingSpec extends PekkoSpec("""
}
"include streamId in each frame" in {
- val bytes = TcpFraming.encodeConnectionHeader(3) ++ frameBytes(3)
+ val bytes = TcpFraming.encodeConnectionHeader(magic, 3) ++ frameBytes(3)
val frames =
Source(List(bytes)).via(framingFlow).runWith(Sink.seq).futureValue
frames(0).streamId should ===(3)
frames(1).streamId should ===(3)
@@ -84,7 +87,7 @@ class TcpFramingSpec extends PekkoSpec("""
"parse frames from random chunks" in {
val numberOfFrames = 100
- val bytes = TcpFraming.encodeConnectionHeader(3) ++
frameBytes(numberOfFrames)
+ val bytes = TcpFraming.encodeConnectionHeader(magic, 3) ++
frameBytes(numberOfFrames)
withClue(s"Random chunks seed: $rndSeed") {
val frames = Source.fromIterator(() =>
rechunk(bytes)).via(framingFlow).runWith(Sink.seq).futureValue
frames.size should ===(numberOfFrames)
@@ -99,7 +102,7 @@ class TcpFramingSpec extends PekkoSpec("""
}
"report truncated frames" in {
- val bytes = TcpFraming.encodeConnectionHeader(3) ++ frameBytes(3).drop(1)
+ val bytes = TcpFraming.encodeConnectionHeader(magic, 3) ++
frameBytes(3).drop(1)
Source(List(bytes)).via(framingFlow).runWith(Sink.seq).failed.futureValue
shouldBe a[FramingException]
}
@@ -108,6 +111,39 @@ class TcpFramingSpec extends PekkoSpec("""
frames.size should ===(0)
}
+ "use default AKKA magic" in {
+ TcpFraming.DefaultMagic should ===(ByteString('A'.toByte, 'K'.toByte,
'K'.toByte, 'A'.toByte))
+ }
+
+ "accept custom magic" in {
+ val customMagic = ByteString('T'.toByte, 'E'.toByte, 'S'.toByte,
'T'.toByte)
+ val customFramingFlow = Flow[ByteString].via(new
TcpFraming(Set(customMagic)))
+ val bytes = TcpFraming.encodeConnectionHeader(customMagic, 2) ++
frameBytes(1)
+ val frames =
Source(List(bytes)).via(customFramingFlow).runWith(Sink.seq).futureValue
+ frames.head.streamId should ===(2)
+ }
+
+ "reject wrong magic" in {
+ val wrongMagic = ByteString('W'.toByte, 'R'.toByte, 'O'.toByte,
'N'.toByte)
+ val bytes = TcpFraming.encodeConnectionHeader(wrongMagic, 2) ++
frameBytes(1)
+ val fail =
Source(List(bytes)).via(framingFlow).runWith(Sink.seq).failed.futureValue
+ fail shouldBe a[FramingException]
+ }
+
+ "accept default AKKA magic" in {
+ val legacyMagic = TcpFraming.DefaultMagic
+ val bytes = TcpFraming.encodeConnectionHeader(legacyMagic, 2) ++
frameBytes(1)
+ val frames =
Source(List(bytes)).via(framingFlow).runWith(Sink.seq).futureValue
+ frames.head.streamId should ===(2)
+ }
+
+ "accept legacy PEKK magic" in {
+ val magic = TcpFraming.PekkoMagic
+ val bytes = TcpFraming.encodeConnectionHeader(magic, 2) ++ frameBytes(1)
+ val frames =
Source(List(bytes)).via(framingFlow).runWith(Sink.seq).futureValue
+ frames.head.streamId should ===(2)
+ }
+
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
