diff --git a/CHANGES.rst b/CHANGES.rst index 7b740bc7e9..461566e1f8 100644 --- a/CHANGES.rst +++ b/CHANGES.rst @@ -82,6 +82,11 @@ Compute Storage ~~~~~~~ +- [Storage] Prevent path traversal when downloading objects to a directory by + using only the object name's basename as the local filename. + (#2206) + [Miguel Caballer - @micafer] + - [Azure Blobs] Fix SAS signatures for percent-encoded object paths by using the decoded path in the canonicalized resource while preserving the encoded path in the returned URL. diff --git a/libcloud/storage/base.py b/libcloud/storage/base.py index 42b80f6fcf..0c872bb830 100644 --- a/libcloud/storage/base.py +++ b/libcloud/storage/base.py @@ -763,7 +763,7 @@ def _save_object( raise LibcloudError(value="Path %s does not exist" % (destination_path), driver=self) if not base_name: - file_path = pjoin(destination_path, obj.name) + file_path = pjoin(destination_path, os.path.basename(obj.name)) else: file_path = destination_path diff --git a/libcloud/test/storage/test_base.py b/libcloud/test/storage/test_base.py index ced1f9de74..17ff2c2a53 100644 --- a/libcloud/test/storage/test_base.py +++ b/libcloud/test/storage/test_base.py @@ -13,9 +13,11 @@ # See the License for the specific language governing permissions and # limitations under the License. +import os import sys import errno import hashlib +import tempfile from io import BytesIO from unittest import mock from unittest.mock import Mock @@ -53,6 +55,22 @@ def setUp(self): self.driver1.strict_mode = False self.driver1.strict_mode = False + def test_save_object_confines_provider_name_to_destination_directory(self): + with tempfile.TemporaryDirectory() as temp_dir: + destination_path = os.path.join(temp_dir, "downloads") + os.sep + os.mkdir(destination_path) + outside_path = os.path.join(temp_dir, "evil") + response = Mock() + response._response.iter_content.return_value = [b"PWNED"] + obj = Mock(size=5) + obj.name = "../../evil" + + self.driver1._save_object(response, obj, destination_path) + + self.assertFalse(os.path.exists(outside_path)) + with open(os.path.join(destination_path, "evil"), "rb") as file_handle: + self.assertEqual(file_handle.read(), b"PWNED") + def test__upload_object_iterator_must_have_next_method(self): valid_iterators = [BytesIO(b("134")), StringIO("bar")] invalid_iterators = ["foobar", "", False, True, 1, object()]