diff --git a/CHANGES.rst b/CHANGES.rst index 2b50a0fbb..3080173f8 100644 --- a/CHANGES.rst +++ b/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) -------------------- diff --git a/ebcli/__init__.py b/ebcli/__init__.py index 3e0792f74..92b273298 100644 --- a/ebcli/__init__.py +++ b/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' diff --git a/ebcli/core/fileoperations.py b/ebcli/core/fileoperations.py index 043329da6..255b45b8b 100644 --- a/ebcli/core/fileoperations.py +++ b/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): diff --git a/ebcli/operations/createops.py b/ebcli/operations/createops.py index 40c98bc68..d60f163df 100644 --- a/ebcli/operations/createops.py +++ b/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: diff --git a/tests/unit/core/test_fileoperations.py b/tests/unit/core/test_fileoperations.py index 0372a8029..c59d90003 100644 --- a/tests/unit/core/test_fileoperations.py +++ b/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'))) diff --git a/tests/unit/operations/test_createops.py b/tests/unit/operations/test_createops.py index a960444a9..f1b1bfe3e 100644 --- a/tests/unit/operations/test_createops.py +++ b/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.'