Harden CloudFormation artifact zipping

awscli/customizations/cloudformation/artifact_exporter.py

@@ -182,19 +182,39 @@ def zip_folder(folder_path):
 def make_zip(filename, source_root):
     zipfile_name = "{0}.zip".format(filename)
     source_root = os.path.abspath(source_root)
+    source_root_real = os.path.realpath(source_root)
     with open(zipfile_name, 'wb') as f:
         zip_file = zipfile.ZipFile(f, 'w', zipfile.ZIP_DEFLATED)
         with contextlib.closing(zip_file) as zf:
             for root, dirs, files in os.walk(source_root, followlinks=True):
+                dirs[:] = [
+                    d for d in dirs
+                    if is_path_within_directory(
+                        os.path.realpath(os.path.join(root, d)),
+                        source_root_real
+                    )
+                ]
                 for filename in files:
                     full_path = os.path.join(root, filename)
+                    if not is_path_within_directory(
+                        os.path.realpath(full_path),
+                        source_root_real
+                    ):
+                        continue
                     relative_path = os.path.relpath(
                         full_path, source_root)
                     zf.write(full_path, relative_path)
 
     return zipfile_name
 
 
+def is_path_within_directory(path, directory):
+    try:
+        return os.path.commonpath([path, directory]) == directory
+    except ValueError:
+        return False
+
+
 @contextmanager
 def mktempfile():
     directory = tempfile.gettempdir()

tests/unit/customizations/cloudformation/test_artifact_exporter.py

@@ -1353,6 +1353,39 @@ def test_make_zip(self):
                 os.remove(zipfile_name)
             test_file_creator.remove_all()
 
+    def test_make_zip_skips_symlinked_directory_outside_root(self):
+        with tempfile.TemporaryDirectory() as rootdir:
+            source_root = os.path.join(rootdir, 'source')
+            outside_root = os.path.join(rootdir, 'outside')
+            os.mkdir(source_root)
+            os.mkdir(outside_root)
+            with open(os.path.join(source_root, 'index.js'), 'w') as f:
+                f.write('exports.handler = () => {};')
+            with open(os.path.join(outside_root, 'secret.txt'), 'w') as f:
+                f.write('secret')
+            try:
+                os.symlink(
+                    outside_root,
+                    os.path.join(source_root, 'linked-outside'),
+                    target_is_directory=True,
+                )
+                os.symlink(
+                    os.path.join(outside_root, 'secret.txt'),
+                    os.path.join(source_root, 'secret-link.txt'),
+                )
+            except (AttributeError, NotImplementedError, OSError) as e:
+                pytest.skip('Symlink creation is not available: %s' % e)
+
+            outfile = os.path.join(rootdir, 'artifact')
+            zipfile_name = make_zip(outfile, source_root)
+            try:
+                with closing(zipfile.ZipFile(zipfile_name, 'r')) as zf:
+                    files_in_zip = {info.filename for info in zf.infolist()}
+
+                self.assertEqual(files_in_zip, {'index.js'})
+            finally:
+                os.remove(zipfile_name)
+
     @mock.patch("shutil.copy")
     @mock.patch("tempfile.mkdtemp")
     def test_copy_to_temp_dir(self, mkdtemp_mock, copyfile_mock):