fix(zip): Add path validation for symlink extraction

Author: vicheey

samcli/local/lambdafn/zip.py

@@ -48,6 +48,11 @@ def _extract(file_info, output_dir, zip_ref):
     -------
     string
         Returns the target path the Zip Entry was extracted to.
+
+    Raises
+    ------
+    ValueError
+        If the extraction path would escape the output directory (path traversal attack)
     """
 
     # Handle any regular file/directory entries
@@ -57,6 +62,23 @@ def _extract(file_info, output_dir, zip_ref):
     source = zip_ref.read(file_info.filename).decode("utf8")
     link_name = os.path.normpath(os.path.join(output_dir, file_info.filename))
 
+    output_dir_abs = os.path.abspath(output_dir)
+    link_name_abs = os.path.abspath(link_name)
+
+    # Ensure the resolved path is within the output directory
+    # Check that link_name_abs starts with output_dir_abs followed by a separator
+    if not link_name_abs.startswith(output_dir_abs + os.sep) and link_name_abs != output_dir_abs:
+        LOG.warning(
+            "Entry '%s' would extract to '%s' which is outside target directory '%s'.",
+            file_info.filename,
+            link_name_abs,
+            output_dir_abs,
+        )
+        raise ValueError(
+            f"Attempted path traversal in zip file. "
+            f"Entry '{file_info.filename}' would extract outside target directory."
+        )
+
     # make leading dirs if needed
     leading_dirs = os.path.dirname(link_name)
     if not os.path.exists(leading_dirs):

tests/unit/local/lambdafn/test_zip.py

@@ -7,10 +7,10 @@
 from tempfile import NamedTemporaryFile, mkdtemp
 from unittest import TestCase
 from unittest import skipIf
-from unittest.mock import patch
+from unittest.mock import patch, Mock
 from parameterized import parameterized, param
 
-from samcli.local.lambdafn.zip import unzip, _override_permissions
+from samcli.local.lambdafn.zip import unzip, _override_permissions, _extract
 
 # On Windows, permissions do not match 1:1 with permissions on Unix systems.
 SKIP_UNZIP_PERMISSION_TESTS = platform.system() == "Windows"
@@ -153,6 +153,25 @@ def _temp_dir(self):
             if name:
                 shutil.rmtree(name)
 
+    @parameterized.expand(
+        [
+            param("../../etc/other", b"/etc/other"),
+            param("../../../etc/other", b"/etc/other"),
+            param("/etc/other", b"/etc/other"),
+            param("foo/../../../../../../etc/other", b"target"),
+        ]
+    )
+    def test_unzip_blocks_symlinks_with_invalid_paths(self, symlink_path, symlink_target):
+        """Test that unzip blocks symlinks attempting to escape the extraction directory."""
+        test_files = {symlink_path: {"file_type": 0o12, "contents": symlink_target, "permissions": 0o644}}
+
+        with self._create_zip(test_files, add_attributes=True) as zip_file_name:
+            with self._temp_dir() as extract_dir:
+                with self.assertRaises(ValueError) as context:
+                    unzip(zip_file_name, extract_dir)
+
+                self.assertIn("path traversal", str(context.exception).lower())
+
 
 class TestOverridePermissions(TestCase):
     @patch("samcli.local.lambdafn.zip.os")
@@ -166,3 +185,38 @@ def test_must_not_override_permissions(self, os_patch):
         _override_permissions(path="./home", permission=None)
 
         os_patch.chmod.assert_not_called()
+
+
+class TestSymlinkExtractionValidation(TestCase):
+    """Tests for symlink path validation during extraction.
+
+    These tests verify that symlinks with paths that would escape the target
+    directory are blocked. The key is the symlink's location (file_info.filename),
+    not what it points to (the symlink target).
+    """
+
+    # Maps symlink paths to their targets
+    # The path (key) is what we're validating - it attempts to escape the extraction directory
+    # The target (value) is what the symlink points to - this doesn't affect the validation
+    test_links = {
+        "../../etc/other": b"/etc/other",  # Relative path traversal
+        "../../../etc/other": b"/etc/other",  # Multiple levels up
+        "/etc/other": b"/etc/other",  # Absolute path
+        "foo/../../../../../../etc/other": b"target",  # Traversal with prefix
+    }
+
+    @parameterized.expand(list(test_links.keys()))
+    def test_extract_blocks_symlinks_outside_target_directory(self, invalid_path):
+        """Test that _extract blocks symlinks that would escape the target directory."""
+        file_info = zipfile.ZipInfo(invalid_path)
+        file_info.external_attr = (0o12 << 28) | (0o644 << 16)  # Mark as symlink
+
+        # Create a mock zip_ref object with a read() method
+        # In a real ZIP, read() returns the path the symlink points to, not file contents
+        mock_zip_ref = Mock()
+        mock_zip_ref.read.return_value = self.test_links[invalid_path]
+
+        with self.assertRaises(ValueError) as context:
+            _extract(file_info, "/tmp/extract", mock_zip_ref)
+
+        self.assertIn("path traversal", str(context.exception).lower())