Backport TcpFraming magic

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.

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

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
 

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")

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

remote/src/main/scala/org/apache/pekko/remote/artery/tcp/TcpFraming.scala

@@ -35,23 +35,29 @@ 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.
    * It contains a "magic" and the stream identifier for selecting control, ordinary, large
    * 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,23 +75,27 @@ 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]
     startWith(ReadMagic)
 
     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.")
       }
     }

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)
+    }
+
   }
 
 }