CVE-2007-4559 Patch (#408)

* Adding tarfile member sanitization to extractall()

* black reformat

* Refactor: move extract_tarfile to shared module

* Add test

* Remove redundant code/tests

* Add test for safe-guarding path traversal

* add type hint

Co-authored-by: Wing Fung Lau <4760060+hawflau@users.noreply.github.com>

.gitignore

@@ -163,7 +163,9 @@ typings/
 *.tgz
 
 # Except test file
-!tests/functional/workflows/ruby_bundler/test_data/test.tgz
+!tests/functional/testdata/test.tgz
+!tests/functional/testdata/path_reversal_uxix.tgz
+!tests/functional/testdata/path_reversal_win.tgz
 
 # Yarn Integrity file
 .yarn-integrity

aws_lambda_builders/utils.py

@@ -7,6 +7,7 @@
 import os
 import logging
 from pathlib import Path
+from typing import Union
 
 from aws_lambda_builders.architecture import X86_64, ARM64
 
@@ -196,3 +197,28 @@ def create_symlink_or_copy(source: str, destination: str) -> None:
             exc_info=ex if LOG.isEnabledFor(logging.DEBUG) else None,
         )
         copytree(source, destination)
+
+
+def _is_within_directory(directory: Union[str, os.PathLike], target: Union[str, os.PathLike]) -> bool:
+    """Checks if target is located under directory"""
+    abs_directory = os.path.abspath(directory)
+    abs_target = os.path.abspath(target)
+
+    prefix = os.path.commonprefix([abs_directory, abs_target])
+
+    return prefix == abs_directory
+
+
+def extract_tarfile(tarfile_path: Union[str, os.PathLike], unpack_dir: Union[str, os.PathLike]) -> None:
+    """Extracts a tarfile"""
+    import tarfile
+
+    with tarfile.open(tarfile_path, "r:*") as tar:
+        # Makes sure the tar file is sanitized and is free of directory traversal vulnerability
+        # See: https://github.com/advisories/GHSA-gw9q-c7gh-j9vm
+        for member in tar.getmembers():
+            member_path = os.path.join(unpack_dir, member.name)
+            if not _is_within_directory(unpack_dir, member_path):
+                raise tarfile.ExtractError("Attempted Path Traversal in Tar File")
+
+        tar.extractall(unpack_dir)

aws_lambda_builders/workflows/nodejs_npm/actions.py

@@ -5,6 +5,7 @@
 import logging
 
 from aws_lambda_builders.actions import BaseAction, Purpose, ActionFailedError
+from aws_lambda_builders.utils import extract_tarfile
 from .npm import NpmExecutionError
 
 LOG = logging.getLogger(__name__)
@@ -64,7 +65,7 @@ def execute(self):
 
             LOG.debug("NODEJS extracting to %s", self.artifacts_dir)
 
-            self.osutils.extract_tarfile(tarfile_path, self.artifacts_dir)
+            extract_tarfile(tarfile_path, self.artifacts_dir)
 
         except NpmExecutionError as ex:
             raise ActionFailedError(str(ex))

aws_lambda_builders/workflows/nodejs_npm/utils.py

@@ -20,10 +20,6 @@ class OSUtils(object):
     def copy_file(self, file_path, destination_path):
         return shutil.copy2(file_path, destination_path)
 
-    def extract_tarfile(self, tarfile_path, unpack_dir):
-        with tarfile.open(tarfile_path, "r:*") as tar:
-            tar.extractall(unpack_dir)
-
     def file_exists(self, filename):
         return os.path.isfile(filename)
 

aws_lambda_builders/workflows/python_pip/packager.py

@@ -9,6 +9,7 @@
 from email.parser import FeedParser
 
 from aws_lambda_builders.architecture import ARM64, X86_64
+from aws_lambda_builders.utils import extract_tarfile
 from .compat import pip_import_string
 from .compat import pip_no_compile_c_env_vars
 from .compat import pip_no_compile_c_shim
@@ -619,7 +620,7 @@ def _unpack_sdist_into_dir(self, sdist_path, unpack_dir):
         if sdist_path.endswith(".zip"):
             self._osutils.extract_zipfile(sdist_path, unpack_dir)
         elif sdist_path.endswith((".tar.gz", ".tar.bz2")):
-            self._osutils.extract_tarfile(sdist_path, unpack_dir)
+            extract_tarfile(sdist_path, unpack_dir)
         else:
             raise InvalidSourceDistributionNameError(sdist_path)
         # There should only be one directory unpacked.

aws_lambda_builders/workflows/python_pip/utils.py

@@ -56,10 +56,6 @@ def extract_zipfile(self, zipfile_path, unpack_dir):
         with zipfile.ZipFile(zipfile_path, "r") as z:
             z.extractall(unpack_dir)
 
-    def extract_tarfile(self, tarfile_path, unpack_dir):
-        with tarfile.open(tarfile_path, "r:*") as tar:
-            tar.extractall(unpack_dir)
-
     def directory_exists(self, path):
         return os.path.isdir(path)
 

aws_lambda_builders/workflows/ruby_bundler/utils.py

@@ -16,10 +16,6 @@ class OSUtils(object):
     unit test actions in memory
     """
 
-    def extract_tarfile(self, tarfile_path, unpack_dir):
-        with tarfile.open(tarfile_path, "r:*") as tar:
-            tar.extractall(unpack_dir)
-
     def popen(self, command, stdout=None, stderr=None, env=None, cwd=None):
         p = subprocess.Popen(command, stdout=stdout, stderr=stderr, env=env, cwd=cwd)
         return p

tests/functional/test_utils.py

@@ -1,10 +1,12 @@
 import os
 import tempfile
 import shutil
+import platform
+from tarfile import ExtractError
 
 from unittest import TestCase
 
-from aws_lambda_builders.utils import copytree, get_goarch
+from aws_lambda_builders.utils import copytree, get_goarch, extract_tarfile
 
 
 class TestCopyTree(TestCase):
@@ -63,6 +65,24 @@ def test_must_return_valid_go_architecture(self):
         self.assertEqual(get_goarch(""), "amd64")
 
 
+class TestExtractTarFile(TestCase):
+    def test_extract_tarfile_unpacks_a_tar(self):
+        test_tar = os.path.join(os.path.dirname(__file__), "testdata", "test.tgz")
+        test_dir = tempfile.mkdtemp()
+        extract_tarfile(test_tar, test_dir)
+        output_files = set(os.listdir(test_dir))
+        shutil.rmtree(test_dir)
+        self.assertEqual({"test_utils.py"}, output_files)
+
+    def test_raise_exception_for_unsafe_tarfile(self):
+        tar_filename = "path_reversal_win.tgz" if platform.system().lower() == "windows" else "path_reversal_uxix.tgz"
+        test_tar = os.path.join(os.path.dirname(__file__), "testdata", tar_filename)
+        test_dir = tempfile.mkdtemp()
+        self.assertRaisesRegexp(
+            ExtractError, "Attempted Path Traversal in Tar File", extract_tarfile, test_tar, test_dir
+        )
+
+
 def file(*args):
     path = os.path.join(*args)
     basedir = os.path.dirname(path)

tests/functional/workflows/nodejs_npm/test_utils.py

@@ -65,20 +65,6 @@ def test_file_exists_checking_if_file_exists_in_a_dir(self):
 
         self.assertFalse(self.osutils.file_exists(nonexisting_file))
 
-    def test_extract_tarfile_unpacks_a_tar(self):
-
-        test_tar = os.path.join(os.path.dirname(__file__), "test_data", "test.tgz")
-
-        test_dir = tempfile.mkdtemp()
-
-        self.osutils.extract_tarfile(test_tar, test_dir)
-
-        output_files = set(os.listdir(test_dir))
-
-        shutil.rmtree(test_dir)
-
-        self.assertEqual({"test_utils.py"}, output_files)
-
     def test_dirname_returns_directory_for_path(self):
         dirname = self.osutils.dirname(sys.executable)
 

tests/functional/workflows/ruby_bundler/test_ruby_utils.py

@@ -12,14 +12,6 @@ class TestOSUtils(TestCase):
     def setUp(self):
         self.osutils = utils.OSUtils()
 
-    def test_extract_tarfile_unpacks_a_tar(self):
-        test_tar = os.path.join(os.path.dirname(__file__), "test_data", "test.tgz")
-        test_dir = tempfile.mkdtemp()
-        self.osutils.extract_tarfile(test_tar, test_dir)
-        output_files = set(os.listdir(test_dir))
-        shutil.rmtree(test_dir)
-        self.assertEqual({"test_utils.py"}, output_files)
-
     def test_dirname_returns_directory_for_path(self):
         dirname = self.osutils.dirname(sys.executable)
         self.assertEqual(dirname, os.path.dirname(sys.executable))

tests/unit/workflows/nodejs_npm/test_actions.py

@@ -15,9 +15,10 @@
 
 
 class TestNodejsNpmPackAction(TestCase):
-    @patch("aws_lambda_builders.workflows.nodejs_npm.utils.OSUtils")
+    @patch("aws_lambda_builders.workflows.nodejs_npm.actions.extract_tarfile")
     @patch("aws_lambda_builders.workflows.nodejs_npm.npm.SubprocessNpm")
-    def test_tars_and_unpacks_npm_project(self, OSUtilMock, SubprocessNpmMock):
+    @patch("aws_lambda_builders.workflows.nodejs_npm.utils.OSUtils")
+    def test_tars_and_unpacks_npm_project(self, OSUtilMock, SubprocessNpmMock, extract_tarfile_mock):
         osutils = OSUtilMock.return_value
         subprocess_npm = SubprocessNpmMock.return_value
 
@@ -34,7 +35,7 @@ def test_tars_and_unpacks_npm_project(self, OSUtilMock, SubprocessNpmMock):
         action.execute()
 
         subprocess_npm.run.assert_called_with(["pack", "-q", "file:/abs:/dir:manifest"], cwd="scratch_dir")
-        osutils.extract_tarfile.assert_called_with("scratch_dir/package.tar", "artifacts")
+        extract_tarfile_mock.assert_called_with("scratch_dir/package.tar", "artifacts")
 
     @patch("aws_lambda_builders.workflows.nodejs_npm.utils.OSUtils")
     @patch("aws_lambda_builders.workflows.nodejs_npm.npm.SubprocessNpm")