fix: Only dereference file paths if they all exist

samcli/lib/utils/tar.py

@@ -2,12 +2,15 @@
 Tarball Archive utility
 """
 
+import logging
 import os
 import tarfile
 from contextlib import contextmanager
 from pathlib import Path
 from tempfile import TemporaryFile
-from typing import IO, Callable, Dict, Optional, Union
+from typing import IO, Callable, Dict, List, Optional, Union
+
+LOG = logging.getLogger(__name__)
 
 
 @contextmanager
@@ -38,7 +41,14 @@ def create_tarball(
     """
     tarballfile = TemporaryFile()
 
-    with tarfile.open(fileobj=tarballfile, mode=mode, dereference=dereference) as archive:
+    do_dereferece = dereference
+
+    # validate that the destinations for the symlink targets exist
+    if do_dereferece and not _validate_destinations_exists(list(tar_paths.keys())):
+        LOG.warning("Falling back to not resolving symlinks to create a tarball.")
+        do_dereferece = False
+
+    with tarfile.open(fileobj=tarballfile, mode=mode, dereference=do_dereferece) as archive:
         for path_on_system, path_in_tarball in tar_paths.items():
             archive.add(path_on_system, arcname=path_in_tarball, filter=tar_filter)
 
@@ -52,6 +62,32 @@ def create_tarball(
         tarballfile.close()
 
 
+def _validate_destinations_exists(tar_paths: List[Union[str, Path]]) -> bool:
+    """
+    Validates whether the destination of a symlink exists by resolving the link
+    and checking the resolved path.
+
+    Parameters
+    ----------
+    tar_paths: List[Union[str, Path]]
+        A list of Paths to check
+
+    Return
+    ------
+    bool:
+        True all the checked paths exist, otherwise returns false
+    """
+    for file in tar_paths:
+        file_path_obj = Path(file)
+        resolved_path = file_path_obj.resolve()
+
+        if file_path_obj.is_symlink() and not resolved_path.exists():
+            LOG.warning(f"Symlinked file {file_path_obj} -> {resolved_path} does not exist!")
+            return False
+
+    return True
+
+
 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)

tests/unit/lib/utils/test_tar.py

@@ -1,9 +1,10 @@
 import io
 from unittest import TestCase
 import tarfile
-from unittest.mock import Mock, patch, call
+from unittest.mock import MagicMock, Mock, patch, call
+from parameterized import parameterized
 
-from samcli.lib.utils.tar import extract_tarfile, create_tarball, _is_within_directory
+from samcli.lib.utils.tar import _validate_destinations_exists, extract_tarfile, create_tarball, _is_within_directory
 
 
 class TestTar(TestCase):
@@ -159,3 +160,45 @@ def test_tarfile_obj_is_not_within_dir(self):
         target = "/another/path/file"
 
         self.assertFalse(_is_within_directory(directory, target))
+
+    @patch("samcli.lib.utils.tar.tarfile.open")
+    @patch("samcli.lib.utils.tar.TemporaryFile")
+    @patch("samcli.lib.utils.tar._validate_destinations_exists")
+    def test_generating_tarball_revert_false_derefernce(self, validate_mock, temporary_file_patch, tarfile_open_patch):
+        temp_file_mock = Mock()
+        temporary_file_patch.return_value = temp_file_mock
+
+        tarfile_file_mock = Mock()
+        tarfile_open_patch.return_value.__enter__.return_value = tarfile_file_mock
+
+        validate_mock.return_value = False
+
+        # pass in dereference is True
+        with create_tarball({"/some/path": "/layer1"}, tar_filter=None, dereference=True) as tarball:
+            self.assertEqual(tarball, temp_file_mock)
+
+        # validate that deference was changed back to False
+        tarfile_open_patch.assert_called_once_with(fileobj=temp_file_mock, mode="w", dereference=False)
+
+    @parameterized.expand(
+        [
+            (True,),
+            (False,),
+        ]
+    )
+    @patch("samcli.lib.utils.tar.Path")
+    def test_validating_symlinked_tar_path(self, does_resolved_exist, path_mock):
+        mock_resolved_object = Mock()
+        mock_resolved_object.exists.return_value = does_resolved_exist
+
+        mock_path_object = Mock()
+        mock_path_object.resolve = Mock()
+        mock_path_object.resolve.return_value = mock_resolved_object
+        mock_path_object.is_symlink = Mock()
+        mock_path_object.is_symlink.return_value = True
+
+        path_mock.return_value = mock_path_object
+
+        result = _validate_destinations_exists(["mock_path"])
+
+        self.assertEqual(result, does_resolved_exist)