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 6574499688 support PEKK TCP magic (#3425)
6574499688 is described below
commit 6574499688bbc55b7e9f4028b13a1065bb96509a
Author: PJ Fanning <[email protected]>
AuthorDate: Sat Aug 15 10:33:23 2026 +0100
support PEKK TCP magic (#3425)
* support PEKK TCP magic
* scalafmt
---
.../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 | 23 ++++++++++++
.../remote/artery/tcp/ArteryTcpTransport.scala | 5 +--
.../pekko/remote/artery/tcp/TcpFraming.scala | 30 ++++++++++------
.../pekko/remote/artery/tcp/TcpFramingSpec.scala | 41 ++++++++++++++++++----
7 files changed, 132 insertions(+), 18 deletions(-)
diff --git a/docs/src/main/paradox/additional/rolling-updates.md
b/docs/src/main/paradox/additional/rolling-updates.md
index d6044557aa..2cf4a643f6 100644
--- a/docs/src/main/paradox/additional/rolling-updates.md
+++ b/docs/src/main/paradox/additional/rolling-updates.md
@@ -146,6 +146,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 5733c02451..48f00fa4e7 100644
--- a/docs/src/main/paradox/remoting-artery.md
+++ b/docs/src/main/paradox/remoting-artery.md
@@ -161,6 +161,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 160bf59d27..5f32f714b7 100644
--- a/remote/src/main/resources/reference.conf
+++ b/remote/src/main/resources/reference.conf
@@ -906,6 +906,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 used "AKKA" as the default. To support rolling upgrades
from
+ # Pekko 1.x, keep "AKKA" in the array alongside "PEKK".
+ tcp-magic = ["PEKK", "AKKA"]
+
# 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 add2d75c82..60c632ebb8 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,7 +14,9 @@
package org.apache.pekko.remote.artery
import java.net.InetAddress
+import java.nio.charset.StandardCharsets
+import scala.collection.immutable
import scala.annotation.nowarn
import scala.concurrent.duration._
import scala.jdk.CollectionConverters._
@@ -22,6 +24,7 @@ import scala.jdk.CollectionConverters._
import org.apache.pekko
import pekko.NotUsed
import pekko.io.dns.internal.AsyncDnsResolver
+import pekko.util.ByteString
import pekko.stream.ActorMaterializerSettings
import pekko.util.Helpers.ConfigOps
import pekko.util.Helpers.Requiring
@@ -117,6 +120,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.toSeq
+ 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 10069b2168..55ebafa7cd 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
@@ -188,7 +188,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
@@ -354,7 +355,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 6aca0ddd95..26d86c58ef 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 default 4-byte magic header for Pekko (PEKK).
* 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('P'.toByte, 'E'.toByte, 'K'.toByte, 'K'.toByte)
+
+ /**
+ * The legacy 4-byte magic header from Akka (AKKA).
+ * Used for backward compatibility with Pekko 1.x.
+ */
+ val LegacyMagic = ByteString('A'.toByte, 'K'.toByte, 'K'.toByte, 'A'.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 [${java.util.HexFormat.ofDelimiter(" ").formatHex((magic ++
reader.remainingData).take(10).toArray)}] " +
+ s"got [${java.util.HexFormat.ofDelimiter("
").formatHex((receivedMagic ++ reader.remainingData).take(10).toArray)}] " +
"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 c927853dfb..4a1073070a 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.LegacyMagic)
+ 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,32 @@ class TcpFramingSpec extends PekkoSpec("""
frames.size should ===(0)
}
+ "use default PEKK magic" in {
+ TcpFraming.DefaultMagic should ===(ByteString('P'.toByte, 'E'.toByte,
'K'.toByte, 'K'.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 legacy AKKA magic" in {
+ val legacyMagic = TcpFraming.LegacyMagic
+ val bytes = TcpFraming.encodeConnectionHeader(legacyMagic, 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]
