This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 5378d952fb85 [SPARK-58056][CORE] Improve `NettyStreamManager` to serve
only files under the registered directory
5378d952fb85 is described below
commit 5378d952fb855c8f28677f1c2b30dd2443fe4ecd
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Jul 8 23:15:09 2026 -0700
[SPARK-58056][CORE] Improve `NettyStreamManager` to serve only files under
the registered directory
### What changes were proposed in this pull request?
This PR aims to improve `NettyStreamManager` to serve only files under the
registered directory for Apache Spark 4.3.0.
### Why are the changes needed?
To be robust and align with `addFile`/`addJar`, which already canonicalize.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Pass the CIs with the newly added `NettyStreamManagerSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
Closes #57151 from dongjoon-hyun/SPARK-58056.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 5388eba42cb16db8506fcb3691a727fc066e4252)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../spark/rpc/netty/NettyStreamManager.scala | 3 +-
.../spark/rpc/netty/NettyStreamManagerSuite.scala | 72 ++++++++++++++++++++++
2 files changed, 74 insertions(+), 1 deletion(-)
diff --git
a/core/src/main/scala/org/apache/spark/rpc/netty/NettyStreamManager.scala
b/core/src/main/scala/org/apache/spark/rpc/netty/NettyStreamManager.scala
index 9ac14f348368..b1d605b1a567 100644
--- a/core/src/main/scala/org/apache/spark/rpc/netty/NettyStreamManager.scala
+++ b/core/src/main/scala/org/apache/spark/rpc/netty/NettyStreamManager.scala
@@ -59,7 +59,8 @@ private[netty] class NettyStreamManager(rpcEnv: NettyRpcEnv)
case other =>
val dir = dirs.get(ftype)
require(dir != null, s"Invalid stream URI: $ftype not found.")
- new File(dir, fname)
+ val canonicalFile = new File(dir, fname).getCanonicalFile
+ if (canonicalFile.getPath.startsWith(dir.getPath + File.separator))
canonicalFile else null
}
if (file != null && file.isFile()) {
diff --git
a/core/src/test/scala/org/apache/spark/rpc/netty/NettyStreamManagerSuite.scala
b/core/src/test/scala/org/apache/spark/rpc/netty/NettyStreamManagerSuite.scala
new file mode 100644
index 000000000000..a5f9ef3f0d54
--- /dev/null
+++
b/core/src/test/scala/org/apache/spark/rpc/netty/NettyStreamManagerSuite.scala
@@ -0,0 +1,72 @@
+/*
+ * 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.spark.rpc.netty
+
+import java.io.File
+import java.nio.file.Files
+import java.util.UUID
+
+import org.apache.spark.{SecurityManager, SparkConf, SparkFunSuite}
+import org.apache.spark.rpc.RpcEnvConfig
+
+class NettyStreamManagerSuite extends SparkFunSuite {
+
+ private def withStreamManager(f: NettyStreamManager => Unit): Unit = {
+ val conf = new SparkConf()
+ val config = RpcEnvConfig(conf, "test", "localhost", "localhost", 0,
+ new SecurityManager(conf), 0, clientMode = false)
+ val rpcEnv = new NettyRpcEnvFactory().create(config)
+ try {
+ f(rpcEnv.fileServer.asInstanceOf[NettyStreamManager])
+ } finally {
+ rpcEnv.shutdown()
+ }
+ }
+
+ test("openStream rejects path traversal outside the registered directory") {
+ withTempDir { tempDir =>
+ val registeredDir = new File(tempDir, "registered")
+ assert(registeredDir.mkdir())
+ val realFile = new File(registeredDir, "real.txt")
+ Files.writeString(realFile.toPath, UUID.randomUUID().toString)
+
+ // A file that lives outside the registered directory and must never be
served.
+ val secretFile = new File(tempDir, "secret.txt")
+ Files.writeString(secretFile.toPath, UUID.randomUUID().toString)
+
+ withStreamManager { manager =>
+ manager.addDirectory("/classes", registeredDir)
+
+ // Normal case: a real file under the registered directory is served.
+ val buffer = manager.openStream("/classes/real.txt")
+ assert(buffer != null)
+ assert(buffer.size() === realFile.length())
+
+ // Traversal is rejected.
+ assert(manager.openStream("/classes/../secret.txt") === null)
+
+ // Normalization bypass attempts are rejected.
+ assert(manager.openStream("/classes/../../secret.txt") === null)
+ assert(manager.openStream(s"/classes/${secretFile.getAbsolutePath}")
=== null)
+
+ // A non-existent file under the registered directory is not served.
+ assert(manager.openStream("/classes/nope.txt") === null)
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
