Fix test malicious tarball fail

Since I650fcbc8f773fad8116338f6fb0cf7b4f4f17b33 builds from git fails
on plugins with an exception: 'tarfile.ReadError: not a gzip file'
because the test checks only gzip compressed archives but plugins
created as plain tar files. This change fixes the issue using
transparent compression support and also adds some debug info.

Closes-Bug: #1990432
Change-Id: If0f9b4dd058a257d0653332d1b663e150c717304
Signed-off-by: Maksim Malchuk <maksim.malchuk@gmail.com>
Co-Authored-by: Michal Nasiadka <mnasiadka@gmail.com>
(cherry picked from commit 143765fb67221cc51f1dc56a41ac2b67dddc453f)

2 files changed, 73 insertions, 1 deletions

diff --git a/kolla/image/tasks.py b/kolla/image/tasks.py
index 53ae4b705..64df3d386 100644
--- a/kolla/image/tasks.py
+++ b/kolla/image/tasks.py
@@ -265,7 +265,7 @@ class BuildTask(DockerTask):
     def builder(self, image):
 
         def _test_malicious_tarball(archive, path):
-            tar_file = tarfile.open(archive, 'r|gz')
+            tar_file = tarfile.open(archive, 'r|*')
             for n in tar_file.getnames():
                 if not os.path.abspath(os.path.join(path, n)).startswith(path):
                     tar_file.close()
diff --git a/kolla/tests/test_build.py b/kolla/tests/test_build.py
index 078714583..6b373ea2f 100644
--- a/kolla/tests/test_build.py
+++ b/kolla/tests/test_build.py
@@ -293,9 +293,81 @@ class TasksTest(base.TestCase):
 
     @mock.patch.dict(os.environ, clear=True)
     @mock.patch('docker.APIClient')
+    def test_local_directory(self, mock_client):
+        tmpdir = tempfile.mkdtemp()
+        file_name = 'test.txt'
+        file_path = os.path.join(tmpdir, file_name)
+        saved_umask = os.umask(0o077)
+
+        try:
+            with open(file_path, 'w') as f:
+                f.write('Hello')
+
+            self.dc = mock_client
+            self.image.plugins = [{
+                'name': 'fake-image-base-plugin-test',
+                'type': 'local',
+                'enabled': True,
+                'source': tmpdir}
+                ]
+            push_queue = mock.Mock()
+            builder = tasks.BuildTask(self.conf, self.image, push_queue)
+            builder.run()
+            self.assertTrue(builder.success)
+
+        except IOError:
+            print('IOError')
+        else:
+            os.remove(file_path)
+        finally:
+            os.umask(saved_umask)
+            os.rmdir(tmpdir)
+
+    @mock.patch.dict(os.environ, clear=True)
+    @mock.patch('docker.APIClient')
     def test_malicious_tar(self, mock_client):
         tmpdir = tempfile.mkdtemp()
         file_name = 'test.txt'
+        archive_name = 'my_archive.tar'
+        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') 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 = tasks.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.dict(os.environ, clear=True)
+    @mock.patch('docker.APIClient')
+    def test_malicious_tar_gz(self, mock_client):
+        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)