Test for unsafe files in tarfile.extractall

Closes-Bug: #1990432

Change-Id: I650fcbc8f773fad8116338f6fb0cf7b4f4f17b33
(cherry picked from commit 3d008b7)

Author: mnasiadka

kolla/image/build.py

@@ -490,6 +490,14 @@ def update_buildargs(self):
 
     def builder(self, image):
 
+        def _test_malicious_tarball(archive, path):
+            tar_file = tarfile.open(archive, 'r|gz')
+            for n in tar_file.getnames():
+                if not os.path.abspath(os.path.join(path, n)).startswith(path):
+                    tar_file.close()
+                    self.logger.error(f'Unsafe filenames in archive {archive}')
+                    raise ArchivingError
+
         def make_an_archive(items, arcname, item_child_path=None):
             if not item_child_path:
                 item_child_path = arcname
@@ -502,8 +510,9 @@ def make_an_archive(items, arcname, item_child_path=None):
                 archives.append(archive_path)
             if archives:
                 for archive in archives:
+                    _test_malicious_tarball(archive, items_path)
                     with tarfile.open(archive, 'r') as archive_tar:
-                        archive_tar.extractall(path=items_path)
+                        archive_tar.extractall(path=items_path)  # nosec
             else:
                 try:
                     os.mkdir(items_path)

kolla/tests/test_build.py

@@ -15,6 +15,8 @@
 import os
 import requests
 import sys
+import tarfile
+import tempfile
 from unittest import mock
 
 from kolla.cmd import build as build_cmd
@@ -299,6 +301,47 @@ def test_process_source(self, mock_get, mock_client,
             else:
                 self.assertIsNotNone(get_result)
 
+    @mock.patch.dict(os.environ, clear=True)
+    @mock.patch('docker.APIClient')
+    def test_malicious_tar(self, mock_client):
+        self.conf.set_override('install_type', 'source')
+        tmpdir = tempfile.mkdtemp()
+        file_name = 'test.txt'
+        archive_name = 'my_archive.tar.gz'
+        file_path = os.path.join(tmpdir, file_name)
+        archive_path = os.path.join(tmpdir, archive_name)
+        # Ensure the file is read/write by the creator only
+        saved_umask = os.umask(0o077)
+
+        try:
+            with open(file_path, 'w') as f:
+                f.write('Hello')
+
+            with tarfile.open(archive_path, 'w:gz') as tar:
+                tar.add(file_path, arcname='../test.txt')
+
+            self.dc = mock_client
+            self.image.plugins = [{
+                'name': 'fake-image-base-plugin-test',
+                'type': 'local',
+                'enabled': True,
+                'source': archive_path}
+                ]
+
+            push_queue = mock.Mock()
+            builder = build.BuildTask(self.conf, self.image, push_queue)
+            builder.run()
+            self.assertFalse(builder.success)
+
+        except IOError:
+            print('IOError')
+        else:
+            os.remove(file_path)
+            os.remove(archive_path)
+        finally:
+            os.umask(saved_umask)
+            os.rmdir(tmpdir)
+
     @mock.patch('os.path.exists')
     @mock.patch('os.utime')
     @mock.patch('shutil.rmtree')