Merge pull request #586 from nikita-khapre/fix-path-traversal-zip-extraction

Fix path traversal in zip extraction

Author: nikita-khapre

CHANGES.rst

@@ -1,5 +1,10 @@
 Changelog
 =========
+--------------------
+3.27.3 (2026-06-29)
+--------------------
+- Fix path traversal vulnerability in zip extraction for ``eb labs download`` and ``eb create``
+
 --------------------
 3.27.2 (2026-05-13)
 --------------------

ebcli/__init__.py

@@ -11,4 +11,4 @@
 # 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.
-__version__ = '3.27.2'
+__version__ = '3.27.3'

ebcli/core/fileoperations.py

@@ -40,7 +40,8 @@
 from ebcli.objects.exceptions import (
     NotInitializedError,
     InvalidSyntaxError,
-    NotFoundError
+    NotFoundError,
+    ValidationError
 )
 from ebcli.core.ebglobals import Constants
 
@@ -482,14 +483,19 @@ def unzip_folder(file_location, directory):
     if not os.path.isdir(directory):
         os.makedirs(directory)
 
-    zip = zipfile.ZipFile(file_location, 'r', allowZip64=True)
-    for cur_file in zip.namelist():
-        if not cur_file.endswith('/'):
-            root, name = os.path.split(cur_file)
-            path = os.path.normpath(os.path.join(directory, root))
-            if not os.path.isdir(path):
-                os.makedirs(path)
-            open(os.path.join(path, name), 'wb').write(zip.read(cur_file))
+    real_directory = os.path.realpath(directory)
+    with zipfile.ZipFile(file_location, 'r', allowZip64=True) as zf:
+        for cur_file in zf.namelist():
+            if not cur_file.endswith('/'):
+                root, name = os.path.split(cur_file)
+                path = os.path.normpath(os.path.join(directory, root))
+                target_file = os.path.realpath(os.path.join(path, name))
+                if not target_file.startswith(real_directory + os.sep):
+                    raise ValidationError(
+                        'Zip entry "{}" would extract outside the target directory'.format(cur_file))
+                if not os.path.isdir(path):
+                    os.makedirs(path)
+                open(os.path.join(path, name), 'wb').write(zf.read(cur_file))
 
 
 def delete_app_file(app_name):

ebcli/operations/createops.py

@@ -12,7 +12,6 @@
 # language governing permissions and limitations under the License.
 import os
 import re
-from zipfile import ZipFile
 
 from cement.utils.misc import minimal_logger
 
@@ -164,7 +163,7 @@ def download_and_extract_sample_app(env_name):
         zip_file_location = '.elasticbeanstalk/.sample_app_download.zip'
         io.echo('INFO: {}'.format(strings['create.downloading_sample_application']))
         download_application_version(url, zip_file_location)
-        ZipFile(zip_file_location, 'r', allowZip64=True).extractall()
+        fileoperations.unzip_folder(zip_file_location, os.getcwd())
         os.remove(zip_file_location)
         io.echo('INFO: {}'.format(strings['create.sample_application_download_complete']))
     except NotAuthorizedError as e:

tests/unit/core/test_fileoperations.py

@@ -25,7 +25,7 @@
 
 from ebcli.core import fileoperations
 from ebcli.objects.buildconfiguration import BuildConfiguration
-from ebcli.objects.exceptions import NotInitializedError, NotFoundError
+from ebcli.objects.exceptions import NotInitializedError, NotFoundError, ValidationError
 
 
 class TestFileOperations(unittest.TestCase):
@@ -1060,3 +1060,63 @@ def test___validate_file_for_archive__ignores_socket_files(
         actual = fileoperations._validate_file_for_archive(filepath)
 
         self.assertFalse(actual)
+
+
+class TestUnzipFolderPathTraversal(unittest.TestCase):
+    def setUp(self):
+        self.test_dir = os.path.join(os.path.dirname(__file__), '_test_unzip_traversal')
+        self.extract_dir = os.path.join(self.test_dir, 'extract')
+        os.makedirs(self.extract_dir, exist_ok=True)
+
+    def tearDown(self):
+        shutil.rmtree(self.test_dir, ignore_errors=True)
+
+    def _make_zip(self, entries, zip_path=None):
+        if zip_path is None:
+            zip_path = os.path.join(self.test_dir, 'test.zip')
+        with zipfile.ZipFile(zip_path, 'w') as zf:
+            for name, content in entries.items():
+                zf.writestr(name, content)
+        return zip_path
+
+    def test_unzip_folder__normal_entries_extract_correctly(self):
+        zip_path = self._make_zip({
+            'app.py': 'print("hello")',
+            'subdir/config.yml': 'key: value',
+            'a/b/c/deep.txt': 'deep content',
+            './foo/bar.txt': 'content',
+        })
+        fileoperations.unzip_folder(zip_path, self.extract_dir)
+
+        self.assertTrue(os.path.exists(os.path.join(self.extract_dir, 'app.py')))
+        self.assertTrue(os.path.exists(os.path.join(self.extract_dir, 'subdir', 'config.yml')))
+        self.assertTrue(os.path.exists(os.path.join(self.extract_dir, 'a', 'b', 'c', 'deep.txt')))
+        self.assertTrue(os.path.exists(os.path.join(self.extract_dir, 'foo', 'bar.txt')))
+
+    def test_unzip_folder__path_traversal_raises_error(self):
+        zip_path = self._make_zip({
+            '../../../tmp/pwned.txt': 'PWNED',
+        })
+        with self.assertRaises(ValidationError):
+            fileoperations.unzip_folder(zip_path, self.extract_dir)
+
+        self.assertFalse(os.path.exists(os.path.join(self.test_dir, 'pwned.txt')))
+
+    def test_unzip_folder__absolute_path_entry_raises_error(self):
+        zip_path = self._make_zip({
+            '/tmp/absolute_pwned.txt': 'PWNED',
+        })
+        with self.assertRaises(ValidationError):
+            fileoperations.unzip_folder(zip_path, self.extract_dir)
+
+    def test_unzip_folder__legitimate_deep_and_redundant_paths(self):
+        zip_path = self._make_zip({
+            'a/b/c/d/e/deep.txt': 'deep content',
+            './foo/bar.txt': 'content1',
+            'foo//baz.txt': 'content2',
+        })
+        fileoperations.unzip_folder(zip_path, self.extract_dir)
+
+        self.assertTrue(os.path.exists(os.path.join(self.extract_dir, 'a', 'b', 'c', 'd', 'e', 'deep.txt')))
+        self.assertTrue(os.path.exists(os.path.join(self.extract_dir, 'foo', 'bar.txt')))
+        self.assertTrue(os.path.exists(os.path.join(self.extract_dir, 'foo', 'baz.txt')))

tests/unit/operations/test_createops.py

@@ -425,19 +425,17 @@ def test_download_application_version(
 
     @mock.patch('ebcli.operations.createops.retrieve_application_version_url')
     @mock.patch('ebcli.operations.createops.download_application_version')
-    @mock.patch('ebcli.operations.createops.ZipFile')
+    @mock.patch('ebcli.operations.createops.fileoperations.unzip_folder')
     @mock.patch('ebcli.operations.createops.os.remove')
     @mock.patch('ebcli.operations.createops.io.echo')
     def test_download_and_extract_sample_app__successfully_downloads_and_extracts(
             self,
             echo_mock,
             remove_mock,
-            ZipFile_mock,
+            unzip_folder_mock,
             download_application_version_mock,
             retrieve_application_version_url_mock
     ):
-        zipfile_mock = mock.MagicMock()
-        ZipFile_mock.return_value = zipfile_mock
         retrieve_application_version_url_mock.return_value = 'http://app-server.com'
 
         createops.download_and_extract_sample_app('my-environment')
@@ -447,12 +445,7 @@ def test_download_and_extract_sample_app__successfully_downloads_and_extracts(
             'http://app-server.com',
             '.elasticbeanstalk/.sample_app_download.zip'
         )
-        ZipFile_mock.assert_called_once_with(
-            '.elasticbeanstalk/.sample_app_download.zip',
-            'r',
-            allowZip64=True
-        )
-        zipfile_mock.extractallassert_called_once_with()
+        unzip_folder_mock.assert_called_once()
         remove_mock.assert_called_once_with('.elasticbeanstalk/.sample_app_download.zip')
         echo_mock.assert_has_calls(
             [
@@ -463,7 +456,7 @@ def test_download_and_extract_sample_app__successfully_downloads_and_extracts(
 
     @mock.patch('ebcli.operations.createops.retrieve_application_version_url')
     @mock.patch('ebcli.operations.createops.download_application_version')
-    @mock.patch('ebcli.operations.createops.ZipFile')
+    @mock.patch('ebcli.operations.createops.fileoperations.unzip_folder')
     @mock.patch('ebcli.operations.createops.os.remove')
     @mock.patch('ebcli.operations.createops.io.echo')
     @mock.patch('ebcli.operations.createops.io.log_warning')
@@ -472,7 +465,7 @@ def test_download_and_extract_sample_app__lacks_permissions_to_query_cloudformat
             lgo_warning_mock,
             echo_mock,
             remove_mock,
-            ZipFile_mock,
+            unzip_folder_mock,
             download_application_version_mock,
             retrieve_application_version_url_mock
     ):
@@ -487,7 +480,7 @@ def test_download_and_extract_sample_app__lacks_permissions_to_query_cloudformat
             'http://app-server.com',
             '.elasticbeanstalk/.sample_app_download.zip'
         )
-        ZipFile_mock.assert_not_called()
+        unzip_folder_mock.assert_not_called()
         remove_mock.assert_not_called()
         echo_mock.assert_called_once_with(
             'INFO: Downloading sample application to the current directory.'