Skip to content

Commit 69b0424

Browse files
committed
refactor: Replace custom copytree() with stdlib shutil.copytree
Replace hand-rolled copytree implementations with shutil.copytree(..., symlinks=True, dirs_exist_ok=True) in both osutils.py and copy_terraform_built_artifacts.py. The custom implementations were originally needed for Python 3.7 compatibility (dirs_exist_ok was added in 3.8), but SAM CLI now requires Python 3.10+, making them unnecessary. Using the stdlib version with symlinks=True correctly preserves both file and directory symlinks without following them. Changes: - osutils.copytree: replaced ~40-line manual implementation with single shutil.copytree call; removed unused errno import; replaced os.remove try/except with Path.unlink(missing_ok=True) in remove() - Terraform copy_terraform_built_artifacts.copytree: same replacement - Updated unit tests to match new implementation - Added integration-style tests verifying file and directory symlink preservation on disk - Audited os.walk(followlinks=True) in package/utils.py — intentional and correct for zip packaging (zip files need dereferenced content) This is a code quality / defense-in-depth improvement, not a security fix. We do not believe this warrants a new CVE assignment.
1 parent fdf769d commit 69b0424

3 files changed

Lines changed: 74 additions & 107 deletions

File tree

samcli/hook_packages/terraform/copy_terraform_built_artifacts.py

Lines changed: 4 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -195,20 +195,11 @@ def search(self, data):
195195

196196

197197
def copytree(src, dst):
198-
"""Modified copytree method
199-
Note: before python3.8 there is no `dir_exists_ok` argument, therefore
200-
this function explicitly creates one if it does not exist.
198+
"""Copy a directory tree from src to dst, merging into dst if it already exists.
199+
200+
Symbolic links are preserved as symbolic links in the destination.
201201
"""
202-
if not os.path.exists(dst):
203-
os.makedirs(dst)
204-
for item in os.listdir(src):
205-
src_item = os.path.join(src, item)
206-
dst_item = os.path.join(dst, item)
207-
if os.path.isdir(src_item):
208-
# recursively call itself.
209-
copytree(src_item, dst_item)
210-
else:
211-
shutil.copy2(src_item, dst_item)
202+
shutil.copytree(src, dst, symlinks=True, dirs_exist_ok=True)
212203

213204

214205
def cli_exit():

samcli/lib/utils/osutils.py

Lines changed: 14 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@
22
Common OS utilities
33
"""
44

5-
import errno
65
import io
76
import logging
87
import os
@@ -117,7 +116,7 @@ def stderr() -> io.TextIOWrapper:
117116
def remove(path):
118117
if path:
119118
try:
120-
os.remove(path)
119+
Path(path).unlink(missing_ok=True)
121120
except OSError:
122121
pass
123122

@@ -134,60 +133,27 @@ def tempfile_platform_independent():
134133
remove(_tempfile.name)
135134

136135

137-
# NOTE: Py3.8 or higher has a ``dir_exist_ok=True`` parameter to provide this functionality.
138-
# This method can be removed if we stop supporting Py37
139136
def copytree(source, destination, ignore=None):
140137
"""
141-
Similar to shutil.copytree except that it removes the limitation that the destination directory should
142-
be present.
138+
Recursively copy a directory tree from source to destination, allowing the destination
139+
to already exist. Symbolic links in the source tree are preserved as symbolic links
140+
in the destination.
141+
142+
Delegates to ``shutil.copytree`` with ``dirs_exist_ok=True`` and ``symlinks=True``.
143+
143144
:type source: str
144145
:param source:
145-
Path to the source folder to copy
146+
Path to the source directory to copy
146147
:type destination: str
147148
:param destination:
148-
Path to destination folder
149-
:type ignore: function
149+
Path to the destination directory. Will be created if it does not exist;
150+
existing files will be overwritten by corresponding files from source.
151+
:type ignore: callable, optional
150152
:param ignore:
151-
A function that returns a set of file names to ignore, given a list of available file names. Similar to the
152-
``ignore`` property of ``shutils.copytree`` method
153+
A callable that receives the directory being visited and a list of its contents,
154+
and returns a set of names to ignore. See ``shutil.ignore_patterns`` for a helper.
153155
"""
154-
155-
if not os.path.exists(destination):
156-
os.makedirs(destination)
157-
158-
try:
159-
# Let's try to copy the directory metadata from source to destination
160-
shutil.copystat(source, destination)
161-
except OSError as ex:
162-
# Can't copy file access times in Windows
163-
LOG.debug("Unable to copy file access times from %s to %s", source, destination, exc_info=ex)
164-
165-
names = os.listdir(source)
166-
if ignore is not None:
167-
ignored_names = ignore(source, names)
168-
else:
169-
ignored_names = set()
170-
171-
for name in names:
172-
# Skip ignored names
173-
if name in ignored_names:
174-
continue
175-
176-
new_source = os.path.join(source, name)
177-
new_destination = os.path.join(destination, name)
178-
179-
if os.path.isdir(new_source):
180-
copytree(new_source, new_destination, ignore=ignore)
181-
else:
182-
try:
183-
shutil.copy2(new_source, new_destination, follow_symlinks=False)
184-
except OSError as e:
185-
if e.errno != errno.EINVAL:
186-
raise e
187-
188-
# Symlinks do not get copied for Windows using shutil.copy2, which is why
189-
# they are handled separately here.
190-
create_symlink_or_copy(new_source, new_destination)
156+
shutil.copytree(source, destination, symlinks=True, ignore=ignore, dirs_exist_ok=True)
191157

192158

193159
def convert_files_to_unix_line_endings(path: str, target_files: Optional[List[str]] = None) -> None:

tests/unit/lib/utils/test_osutils.py

Lines changed: 56 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,9 @@
33
"""
44

55
import os
6+
import shutil
67
import sys
8+
import tempfile
79

810
from unittest import TestCase
911
from unittest.mock import patch, Mock
@@ -91,66 +93,74 @@ def test_must_delete_if_path_exist(self, patched_rmtree, patched_path):
9193

9294

9395
class Test_copytree(TestCase):
94-
@patch("samcli.lib.utils.osutils.Path")
95-
@patch("samcli.lib.utils.osutils.os")
96-
@patch("samcli.lib.utils.osutils.shutil.copy2")
97-
def test_must_copytree(self, patched_copy2, patched_os, patched_path):
96+
@patch("samcli.lib.utils.osutils.shutil.copytree")
97+
def test_must_copytree(self, patched_copytree):
9898
source_path = "mock-source/path"
9999
destination_path = "mock-destination/path"
100-
mock_path_obj = Mock()
101-
patched_path.exists.return_value = True
102-
patched_os.path.return_value = mock_path_obj
103100

104-
patched_os.path.join.side_effect = [source_path, destination_path]
105-
patched_os.path.isdir.return_value = False
106-
patched_os.listdir.return_value = ["mock-source-file1"]
107101
osutils.copytree(source_path, destination_path)
108102

109-
patched_os.path.join.assert_called()
110-
patched_copy2.assert_called_with(source_path, destination_path, follow_symlinks=False)
103+
patched_copytree.assert_called_once_with(
104+
source_path, destination_path, symlinks=True, ignore=None, dirs_exist_ok=True
105+
)
111106

112-
@patch("samcli.lib.utils.osutils.Path")
113-
@patch("samcli.lib.utils.osutils.os")
114-
@patch("samcli.lib.utils.osutils.shutil.copy2")
115-
def test_copytree_throws_oserror_path_exists(self, patched_copy2, patched_os, patched_path):
107+
@patch("samcli.lib.utils.osutils.shutil.copytree")
108+
def test_copytree_with_ignore(self, patched_copytree):
116109
source_path = "mock-source/path"
117110
destination_path = "mock-destination/path"
118-
mock_path_obj = Mock()
119-
patched_path.exists.return_value = True
120-
patched_os.path.return_value = mock_path_obj
121-
patched_copy2.side_effect = OSError("mock-os-error")
111+
mock_ignore = Mock()
112+
113+
osutils.copytree(source_path, destination_path, ignore=mock_ignore)
114+
115+
patched_copytree.assert_called_once_with(
116+
source_path, destination_path, symlinks=True, ignore=mock_ignore, dirs_exist_ok=True
117+
)
118+
119+
@patch("samcli.lib.utils.osutils.shutil.copytree")
120+
def test_copytree_throws_oserror(self, patched_copytree):
121+
patched_copytree.side_effect = OSError("mock-os-error")
122122

123-
patched_os.path.join.side_effect = [source_path, destination_path]
124-
patched_os.path.isdir.return_value = False
125-
patched_os.listdir.return_value = ["mock-source-file1"]
126123
with self.assertRaises(OSError):
127-
osutils.copytree(source_path, destination_path)
124+
osutils.copytree("mock-source/path", "mock-destination/path")
128125

129-
patched_os.path.join.assert_called()
130-
patched_copy2.assert_called_with(source_path, destination_path, follow_symlinks=False)
131126

132-
@patch("samcli.lib.utils.osutils.create_symlink_or_copy")
133-
@patch("samcli.lib.utils.osutils.Path")
134-
@patch("samcli.lib.utils.osutils.os")
135-
@patch("samcli.lib.utils.osutils.shutil.copy2")
136-
def test_copytree_symlink_copy_error_handling(
137-
self, patched_copy2, patched_os, patched_path, patched_create_symlink_or_copy
138-
):
139-
source_path = "mock-source/path"
140-
destination_path = "mock-destination/path"
141-
mock_path_obj = Mock()
142-
patched_path.exists.return_value = True
143-
patched_os.path.return_value = mock_path_obj
144-
patched_copy2.side_effect = OSError(22, "mock-os-error")
127+
class Test_copytree_symlinks(TestCase):
128+
"""Integration-style tests verifying symlinks=True preserves symlinks on disk."""
145129

146-
patched_os.path.join.side_effect = [source_path, destination_path]
147-
patched_os.path.isdir.return_value = False
148-
patched_os.listdir.return_value = ["mock-source-file1"]
149-
osutils.copytree(source_path, destination_path)
130+
def setUp(self):
131+
self.src = tempfile.mkdtemp()
132+
self.dst = tempfile.mkdtemp()
133+
# clean dst so copytree creates it
134+
shutil.rmtree(self.dst)
135+
136+
def tearDown(self):
137+
shutil.rmtree(self.src, ignore_errors=True)
138+
shutil.rmtree(self.dst, ignore_errors=True)
139+
140+
def test_preserves_file_symlink(self):
141+
real_file = os.path.join(self.src, "real.txt")
142+
with open(real_file, "w") as f:
143+
f.write("content")
144+
os.symlink("real.txt", os.path.join(self.src, "link.txt"))
145+
146+
osutils.copytree(self.src, self.dst)
147+
148+
link = os.path.join(self.dst, "link.txt")
149+
self.assertTrue(os.path.islink(link))
150+
self.assertEqual(os.readlink(link), "real.txt")
151+
152+
def test_preserves_directory_symlink(self):
153+
subdir = os.path.join(self.src, "subdir")
154+
os.makedirs(subdir)
155+
with open(os.path.join(subdir, "file.txt"), "w") as f:
156+
f.write("content")
157+
os.symlink("subdir", os.path.join(self.src, "link_dir"))
158+
159+
osutils.copytree(self.src, self.dst)
150160

151-
patched_os.path.join.assert_called()
152-
patched_copy2.assert_called_with(source_path, destination_path, follow_symlinks=False)
153-
patched_create_symlink_or_copy.assert_called_with(source_path, destination_path)
161+
link = os.path.join(self.dst, "link_dir")
162+
self.assertTrue(os.path.islink(link))
163+
self.assertEqual(os.readlink(link), "subdir")
154164

155165

156166
class Test_create_symlink_or_copy(TestCase):

0 commit comments

Comments
 (0)