Merge branch 'develop'

Author: aws-sam-cli-bot

.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

CODEOWNERS

@@ -0,0 +1,3 @@
+# https://docs.github.com/en/repositories/managing-your-repositorys-settings-and-features/customizing-your-repository/about-code-owners
+
+* @aws/serverless-application-experience-sbt

aws_lambda_builders/__init__.py

@@ -4,5 +4,5 @@
 
 # Changing version will trigger a new release!
 # Please make the version change as the last step of your development.
-__version__ = "1.23.0"
+__version__ = "1.23.1"
 RPC_PROTOCOL_VERSION = "0.3"

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)