[SPARK-57962][PYTHON] Guard against path traversal in install_spark tar extraction

python/pyspark/install.py

@@ -159,12 +159,7 @@ def install_spark(dest: str, spark_version: str, hadoop_version: str, hive_versi
 
             print("Installing to %s" % dest)
             tar = tarfile.open(package_local_path, "r:gz")
-            for member in tar.getmembers():
-                if member.name == package_name:
-                    # Skip the root directory.
-                    continue
-                member.name = os.path.relpath(member.name, package_name + os.path.sep)
-                tar.extract(member, dest)
+            _extract_tar(tar, package_name, dest)
             return
         except Exception:
             print("Failed to download %s from %s:" % (pretty_pkg_name, url))
@@ -178,6 +173,36 @@ def install_spark(dest: str, spark_version: str, hadoop_version: str, hive_versi
     raise OSError("Unable to download %s." % pretty_pkg_name)
 
 
+def _extract_tar(tar: tarfile.TarFile, package_name: str, dest: str) -> None:
+    """
+    Extract the members of ``tar`` into ``dest``, stripping the top-level
+    ``package_name`` directory from each member path.
+
+    Guards against path traversal ("zip slip"): ``os.path.relpath`` does not
+    strip ``..`` segments, so a crafted member could otherwise resolve outside
+    ``dest``. Any member whose resolved destination escapes ``dest`` is
+    rejected instead of extracted.
+
+    Note: tarfile's ``filter="data"`` (PEP 706) rejects such members natively and
+    would replace this manual check, but it is only generally available from
+    Python 3.12.0 (backported to 3.11.4+), so we keep the explicit check while
+    Spark still supports Python 3.11.
+    """
+    dest_root = os.path.realpath(dest)
+    for member in tar.getmembers():
+        if member.name == package_name:
+            # Skip the root directory.
+            continue
+        member.name = os.path.relpath(member.name, package_name + os.path.sep)
+        resolved = os.path.realpath(os.path.join(dest, member.name))
+        if resolved != dest_root and not resolved.startswith(dest_root + os.sep):
+            raise ValueError(
+                "Archive member '%s' would extract outside of the destination "
+                "directory; refusing to extract." % member.name
+            )
+        tar.extract(member, dest)
+
+
 def get_preferred_mirrors() -> list[str]:
     mirror_urls = []
     for _ in range(3):

python/pyspark/tests/test_install_spark.py

@@ -14,15 +14,18 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+import io
 import os
 import re
+import tarfile
 import tempfile
 import unittest
 import urllib.request
 
 from pyspark.install import (
     get_preferred_mirrors,
     install_spark,
+    _extract_tar,
     DEFAULT_HADOOP,
     DEFAULT_HIVE,
     UNSUPPORTED_COMBINATIONS,
@@ -70,6 +73,40 @@ def test_install_spark(self):
             self.assertTrue(os.path.exists("%s/bin/spark-submit" % tmp_dir))
             self.assertTrue(os.path.exists("%s/RELEASE" % tmp_dir))
 
+    def test_extract_tar(self):
+        # A benign member is extracted with the top-level package directory
+        # stripped, while a member whose path escapes the destination
+        # directory (zip slip) is rejected rather than extracted.
+        package_name = "spark-4.1.1-bin-hadoop3"
+
+        def make_tar(path, member_name):
+            with tarfile.open(path, "w") as tar:
+                data = b"content"
+                info = tarfile.TarInfo(name=member_name)
+                info.size = len(data)
+                tar.addfile(info, io.BytesIO(data))
+
+        with tempfile.TemporaryDirectory(prefix="test_install_spark") as tmp_dir:
+            # Benign archive extracts into dest with the package prefix removed.
+            safe_tar = os.path.join(tmp_dir, "safe.tar")
+            make_tar(safe_tar, "%s/bin/spark-submit" % package_name)
+            safe_dest = os.path.join(tmp_dir, "safe_dest")
+            os.makedirs(safe_dest)
+            with tarfile.open(safe_tar, "r") as tar:
+                _extract_tar(tar, package_name, safe_dest)
+            self.assertTrue(os.path.exists(os.path.join(safe_dest, "bin", "spark-submit")))
+
+            # Malicious archive member escaping dest is rejected, and nothing
+            # is written into the destination directory.
+            evil_tar = os.path.join(tmp_dir, "evil.tar")
+            make_tar(evil_tar, "%s/../../evil" % package_name)
+            evil_dest = os.path.join(tmp_dir, "evil_dest")
+            os.makedirs(evil_dest)
+            with tarfile.open(evil_tar, "r") as tar:
+                with self.assertRaisesRegex(ValueError, "outside of the destination"):
+                    _extract_tar(tar, package_name, evil_dest)
+            self.assertEqual([], os.listdir(evil_dest))
+
     def test_package_name(self):
         self.assertEqual(
             "spark-3.0.0-bin-hadoop3.2", checked_package_name("spark-3.0.0", "hadoop3.2", "hive2.3")