fix: don't resolve HOCON includes in config that arrived in a message

actor-tests/src/test/scala/org/apache/pekko/serialization/WireConfigSpec.scala

@@ -0,0 +1,87 @@
+/*
+ * 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.serialization
+
+import java.nio.charset.StandardCharsets
+import java.nio.file.{ Files, Path }
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.pekko.testkit.PekkoSpec
+
+import com.typesafe.config.{ ConfigFactory, ConfigRenderOptions }
+
+class WireConfigSpec extends PekkoSpec {
+
+  // a file the parser must not read when a message asks it to
+  private val secretFile: Path = {
+    val f = Files.createTempFile("wire-config-spec", ".conf")
+    Files.write(f, "secret = leaked".getBytes(StandardCharsets.UTF_8))
+    f
+  }
+  private val filePath = secretFile.toAbsolutePath.toString
+  private val fileUrl = secretFile.toUri.toString
+
+  override def afterTermination(): Unit = Files.deleteIfExists(secretFile)
+
+  "WireConfig" must {
+
+    "parse ordinary HOCON" in {
+      val config = WireConfig.parseString("a = 1\nb { c = two }")
+      config.getInt("a") should ===(1)
+      config.getString("b.c") should ===("two")
+    }
+
+    "parse what a serializer writes" in {
+      // every serializer renders config with ConfigRenderOptions.concise
+      val rendered =
+        ConfigFactory.parseString("pekko.cluster.roles = [a, b]").root.render(ConfigRenderOptions.concise())
+      WireConfig.parseString(rendered).getStringList("pekko.cluster.roles").asScala.toList should ===(List("a", "b"))
+    }
+
+    "not read a file named by an include" in {
+      val config = WireConfig.parseString(s"include file(\"$filePath\")\na = 1")
+      config.hasPath("secret") should ===(false)
+      config.getInt("a") should ===(1)
+    }
+
+    "not read a file named by a required include" in {
+      WireConfig.parseString(s"include required(file(\"$filePath\"))\na = 1").hasPath("secret") should ===(false)
+    }
+
+    "not fetch a URL named by an include" in {
+      // a file: URL stands in for an outbound request, so the test needs no network
+      val config = WireConfig.parseString(s"include url(\"$fileUrl\")\na = 1")
+      config.hasPath("secret") should ===(false)
+      config.getInt("a") should ===(1)
+    }
+
+    "not read a resource named by a classpath include" in {
+      // reference.conf is on the test classpath, so the default includer would pull it in
+      WireConfig.parseString("include classpath(\"reference.conf\")\na = 1").hasPath("pekko.version") should ===(false)
+    }
+
+    "differ from the default parser, which does resolve all three" in {
+      // guards the premise of the tests above: these directives really do resolve without the
+      // includer, so those tests are checking the change rather than an inert directive
+      ConfigFactory.parseString(s"include file(\"$filePath\")").hasPath("secret") should ===(true)
+      ConfigFactory.parseString(s"include url(\"$fileUrl\")").hasPath("secret") should ===(true)
+      ConfigFactory.parseString("include classpath(\"reference.conf\")").hasPath("pekko.version") should ===(true)
+    }
+  }
+}

actor/src/main/scala/org/apache/pekko/serialization/WireConfig.scala

@@ -0,0 +1,82 @@
+/*
+ * 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.serialization
+
+import java.io.File
+import java.net.URL
+
+import org.apache.pekko.annotation.InternalApi
+
+import com.typesafe.config.{
+  Config,
+  ConfigFactory,
+  ConfigIncludeContext,
+  ConfigIncluder,
+  ConfigIncluderClasspath,
+  ConfigIncluderFile,
+  ConfigIncluderURL,
+  ConfigObject,
+  ConfigParseOptions
+}
+
+/**
+ * INTERNAL API
+ *
+ * Parsing of HOCON that arrived in a message.
+ *
+ * HOCON `include` directives are resolved by the parser, not by `resolve()`, so parsing a
+ * string with the default includer reads whatever it names: `include file(...)` and
+ * `include classpath(...)` read from the local filesystem and classpath, and
+ * `include url(...)` performs an outbound request. None of that belongs on a path whose
+ * input came from a peer.
+ *
+ * Every serializer writes config with `ConfigRenderOptions.concise`, which renders JSON and
+ * cannot produce an `include`, so dropping them costs a well-behaved sender nothing.
+ */
+@InternalApi private[pekko] object WireConfig {
+
+  /**
+   * Resolves every form of `include` to an empty object.
+   *
+   * All four interfaces have to be implemented: the parser dispatches `include file(...)`,
+   * `include url(...)` and `include classpath(...)` to the typed methods and falls back to
+   * its own default handling — which does read the resource — when the configured includer
+   * does not implement the matching interface. Only bare `include "..."` goes to `include`.
+   */
+  private object NoIncludes
+      extends ConfigIncluder
+      with ConfigIncluderFile
+      with ConfigIncluderURL
+      with ConfigIncluderClasspath {
+
+    private def empty: ConfigObject = ConfigFactory.empty().root()
+
+    override def withFallback(fallback: ConfigIncluder): ConfigIncluder = this
+    override def include(context: ConfigIncludeContext, what: String): ConfigObject = empty
+    override def includeFile(context: ConfigIncludeContext, what: File): ConfigObject = empty
+    override def includeURL(context: ConfigIncludeContext, what: URL): ConfigObject = empty
+    override def includeResources(context: ConfigIncludeContext, what: String): ConfigObject = empty
+  }
+
+  private val parseOptions: ConfigParseOptions = ConfigParseOptions.defaults().setIncluder(NoIncludes)
+
+  /**
+   * Like `ConfigFactory.parseString`, but with `include` directives resolved to nothing.
+   */
+  def parseString(hocon: String): Config = ConfigFactory.parseString(hocon, parseOptions)
+}

cluster/src/main/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializer.scala

@@ -299,7 +299,7 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
   private def deserializeInitJoin(bytes: Array[Byte]): InternalClusterAction.InitJoin = {
     val m = cm.InitJoin.parseFrom(bytes)
     if (m.hasCurrentConfig)
-      InternalClusterAction.InitJoin(ConfigFactory.parseString(m.getCurrentConfig))
+      InternalClusterAction.InitJoin(WireConfig.parseString(m.getCurrentConfig))
     else
       InternalClusterAction.InitJoin(ConfigFactory.empty)
   }
@@ -310,7 +310,7 @@ final class ClusterMessageSerializer(val system: ExtendedActorSystem)
       val configCheck =
         i.getConfigCheck.getType match {
           case cm.ConfigCheck.Type.CompatibleConfig =>
-            CompatibleConfig(ConfigFactory.parseString(i.getConfigCheck.getClusterConfig))
+            CompatibleConfig(WireConfig.parseString(i.getConfigCheck.getClusterConfig))
           case cm.ConfigCheck.Type.IncompatibleConfig => IncompatibleConfig
           case cm.ConfigCheck.Type.UncheckedConfig    => UncheckedConfig
         }

cluster/src/test/scala/org/apache/pekko/cluster/protobuf/ClusterMessageSerializerSpec.scala

@@ -13,6 +13,9 @@
 
 package org.apache.pekko.cluster.protobuf
 
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+
 import scala.annotation.nowarn
 
 import collection.immutable.SortedSet
@@ -21,6 +24,7 @@ import org.apache.pekko
 import pekko.actor.{ Address, ExtendedActorSystem }
 import pekko.cluster._
 import pekko.cluster.InternalClusterAction.CompatibleConfig
+import pekko.cluster.protobuf.msg.{ ClusterMessages => cm }
 import pekko.cluster.routing.{ ClusterRouterPool, ClusterRouterPoolSettings }
 import pekko.routing.RoundRobinPool
 import pekko.testkit.PekkoSpec
@@ -184,6 +188,37 @@ class ClusterMessageSerializerSpec extends PekkoSpec("pekko.actor.provider = clu
       env.gossip.members.tail.head.roles should be(Set("r1", ClusterSettings.DcRolePrefix + "foo"))
     }
 
+    "not resolve includes in the config of a join message" in {
+      // The joining node renders its config with ConfigRenderOptions.concise, which is JSON and
+      // cannot carry an include, so nothing legitimate is lost by refusing to resolve one. An
+      // include that did resolve would read a local file or issue an outbound request while
+      // deserializing a message from a node that has not joined yet.
+      val secretFile = Files.createTempFile("cluster-message-serializer-spec", ".conf")
+      try {
+        Files.write(secretFile, """secret = "leaked"""".getBytes(StandardCharsets.UTF_8))
+        val hocon = s"""include file("${secretFile.toAbsolutePath}")
+          pekko.cluster.roles = []"""
+
+        val initJoin = serializer
+          .fromBinary(cm.InitJoin.newBuilder().setCurrentConfig(hocon).build().toByteArray, "IJ")
+          .asInstanceOf[InternalClusterAction.InitJoin]
+        initJoin.configOfJoiningNode.hasPath("secret") should ===(false)
+
+        val ackBytes = cm.InitJoinAck
+          .newBuilder()
+          .setAddress(serializer.addressToProto(Address("pekko", "system", "some.host.org", 4711)))
+          .setConfigCheck(
+            cm.ConfigCheck
+              .newBuilder()
+              .setType(cm.ConfigCheck.Type.CompatibleConfig)
+              .setClusterConfig(hocon))
+          .build()
+          .toByteArray
+        val ack = serializer.fromBinary(ackBytes, "IJA").asInstanceOf[InternalClusterAction.InitJoinAck]
+        ack.configCheck.asInstanceOf[CompatibleConfig].clusterConfig.hasPath("secret") should ===(false)
+      } finally Files.deleteIfExists(secretFile)
+    }
+
     "add a default data center role to internal join action if none is present" in {
       val join = roundtrip(InternalClusterAction.Join(a1.uniqueAddress, Set(), Version.Zero))
       join.roles should be(Set(ClusterSettings.DcRolePrefix + "default"))

remote/src/main/scala/org/apache/pekko/remote/serialization/MiscMessageSerializer.scala

@@ -30,7 +30,13 @@ import pekko.remote._
 import pekko.remote.WireFormats.AddressData
 import pekko.remote.routing.RemoteRouterConfig
 import pekko.routing._
-import pekko.serialization.{ BaseSerializer, Serialization, SerializationExtension, SerializerWithStringManifest }
+import pekko.serialization.{
+  BaseSerializer,
+  Serialization,
+  SerializationExtension,
+  SerializerWithStringManifest,
+  WireConfig
+}
 
 import com.typesafe.config.{ Config, ConfigFactory, ConfigRenderOptions }
 
@@ -546,7 +552,7 @@ class MiscMessageSerializer(val system: ExtendedActorSystem) extends SerializerW
 
   private def deserializeConfig(bytes: Array[Byte]): Config = {
     if (bytes.isEmpty) EmptyConfig
-    else ConfigFactory.parseString(new String(bytes, StandardCharsets.UTF_8))
+    else WireConfig.parseString(new String(bytes, StandardCharsets.UTF_8))
   }
 
   private def deserializeFromConfig(bytes: Array[Byte]): FromConfig =

remote/src/test/scala/org/apache/pekko/remote/serialization/MiscMessageSerializerSpec.scala

@@ -14,6 +14,8 @@
 package org.apache.pekko.remote.serialization
 
 import java.io.NotSerializableException
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
 import java.util.Optional
 import java.util.concurrent.TimeoutException
 
@@ -33,7 +35,7 @@ import pekko.serialization.SerializationExtension
 import pekko.testkit.JavaSerializable
 import pekko.testkit.PekkoSpec
 
-import com.typesafe.config.ConfigFactory
+import com.typesafe.config.{ Config, ConfigFactory }
 
 object MiscMessageSerializerSpec {
   val serializationTestOverrides =
@@ -158,6 +160,23 @@ class MiscMessageSerializerSpec extends PekkoSpec(MiscMessageSerializerSpec.test
         }
     }
 
+    "not resolve includes in a serialized Config" in {
+      // Config is written with ConfigRenderOptions.concise, which is JSON and cannot carry an
+      // include, so refusing to resolve one loses nothing. An include that did resolve would
+      // read a local file or issue an outbound request while deserializing a peer's message.
+      val secretFile = Files.createTempFile("misc-message-serializer-spec", ".conf")
+      try {
+        Files.write(secretFile, """secret = "leaked"""".getBytes(StandardCharsets.UTF_8))
+        val serializer = new MiscMessageSerializer(system.asInstanceOf[ExtendedActorSystem])
+        val hocon = s"""include file("${secretFile.toAbsolutePath}")
+          a = 1"""
+
+        val config = serializer.fromBinary(hocon.getBytes(StandardCharsets.UTF_8), "CF").asInstanceOf[Config]
+        config.hasPath("secret") should ===(false)
+        config.getInt("a") should ===(1)
+      } finally Files.deleteIfExists(secretFile)
+    }
+
     "reject invalid manifest" in {
       intercept[IllegalArgumentException] {
         val serializer = new MiscMessageSerializer(system.asInstanceOf[ExtendedActorSystem])