Skip to content
5 changes: 5 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -163,6 +163,11 @@ typings/
# Output of 'npm pack'
*.tgz

# Except test file
!tests/functional/testdata/lib/utils/test.tgz
!tests/functional/testdata/lib/utils/path_reversal_uxix.tgz
!tests/functional/testdata/lib/utils/path_reversal_win.tgz

# Yarn Integrity file
.yarn-integrity

Expand Down
43 changes: 43 additions & 0 deletions samcli/lib/utils/tar.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,9 @@
Tarball Archive utility
"""

import os
import tarfile
from typing import Union, IO, Optional
from tempfile import TemporaryFile
from contextlib import contextmanager

Expand Down Expand Up @@ -39,3 +41,44 @@ def create_tarball(tar_paths, tar_filter=None, mode="w"):
yield tarballfile
finally:
tarballfile.close()


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 bool(prefix == abs_directory)


def extract_tarfile(
tarfile_path: Union[str, os.PathLike] = "",
file_obj: Optional[IO[bytes]] = None,
unpack_dir: Union[str, os.PathLike] = "",
) -> None:
"""
Extracts a tarfile using the provided parameters. If file_obj is specified,
it is used instead of the file_obj opened for tarfile_path.

Parameters
----------
tarfile_path Union[str, os.PathLike]
Key representing a full path to the file or directory and the Value representing the path within the tarball

file_obj Optional[IO[bytes]]
Object for the tarfile that will be extracted

unpack_dir Union[str, os.PathLike]
The directory where the tarfile members will be extracted.
"""
with tarfile.open(name=tarfile_path, fileobj=file_obj, mode="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)
8 changes: 4 additions & 4 deletions samcli/local/docker/container.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
"""
import os
import logging
import tarfile
import tempfile
import threading
import socket
Expand All @@ -14,6 +13,7 @@

from docker.errors import NotFound as DockerNetworkNotFound
from samcli.lib.utils.retry import retry
from samcli.lib.utils.tar import extract_tarfile
from .exceptions import ContainerNotStartableException

from .utils import to_posix_path, find_free_port, NoFreePortsError
Expand Down Expand Up @@ -362,7 +362,8 @@ def _can_connect_to_socket(self) -> bool:
a_socket.close()
return connection_succeeded

def copy(self, from_container_path, to_host_path):
def copy(self, from_container_path, to_host_path) -> None:
"""Copies a path from container into host path"""

if not self.is_created():
raise RuntimeError("Container does not exist. Cannot get logs for this container")
Expand All @@ -378,8 +379,7 @@ def copy(self, from_container_path, to_host_path):
# Seek the handle back to start of file for tarfile to use
fp.seek(0)

with tarfile.open(fileobj=fp, mode="r") as tar:
tar.extractall(path=to_host_path)
extract_tarfile(file_obj=fp, unpack_dir=to_host_path)

@staticmethod
def _write_container_output(output_itr, stdout=None, stderr=None):
Expand Down
Empty file.
58 changes: 58 additions & 0 deletions tests/functional/lib/utils/test_tar.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
import os
import tempfile
import shutil
import platform
from pathlib import Path
from tarfile import ExtractError

from unittest import TestCase

from samcli.lib.utils.tar import extract_tarfile


class TestExtractTarFile(TestCase):
Comment thread
hawflau marked this conversation as resolved.
def test_extract_tarfile_arg_path_unpacks_a_tar(self):
test_tar = Path(__file__).resolve().parents[3].joinpath("functional", "testdata", "lib", "utils", "test.tgz")
test_dir = tempfile.mkdtemp()
extract_tarfile(tarfile_path=test_tar, unpack_dir=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_with_path_arg(self):
tar_filename = "path_reversal_win.tgz" if platform.system().lower() == "windows" else "path_reversal_uxix.tgz"
test_tar = Path(__file__).resolve().parents[3].joinpath("functional", "testdata", "lib", "utils", tar_filename)
test_dir = tempfile.mkdtemp()
self.assertRaisesRegex(
ExtractError,
"Attempted Path Traversal in Tar File",
extract_tarfile,
tarfile_path=test_tar,
unpack_dir=test_dir,
)
shutil.rmtree(test_dir)

def test_extract_tarfile_arg_fileobj_unpacks_a_tar(self):
test_tar = Path(__file__).resolve().parents[3].joinpath("functional", "testdata", "lib", "utils", "test.tgz")
test_dir = tempfile.mkdtemp()
with open(test_tar, mode="rb") as tar:
before_extract_output_files = set(os.listdir(test_dir))
self.assertEqual(set(), before_extract_output_files)
extract_tarfile(file_obj=tar, unpack_dir=test_dir)
after_extract_output_files = set(os.listdir(test_dir))
self.assertEqual({"test_utils.py"}, after_extract_output_files)
shutil.rmtree(test_dir)

def test_raise_exception_for_unsafe_tarfile_with_flieobj_arg(self):
tar_filename = "path_reversal_win.tgz" if platform.system().lower() == "windows" else "path_reversal_uxix.tgz"
test_tar = Path(__file__).resolve().parents[3].joinpath("functional", "testdata", "lib", "utils", tar_filename)
test_dir = tempfile.mkdtemp()
with open(test_tar, mode="rb") as tar:
self.assertRaisesRegex(
ExtractError,
"Attempted Path Traversal in Tar File",
extract_tarfile,
file_obj=tar,
unpack_dir=test_dir,
)
shutil.rmtree(test_dir)
Binary file not shown.
Binary file not shown.
Binary file added tests/functional/testdata/lib/utils/test.tgz
Binary file not shown.
73 changes: 72 additions & 1 deletion tests/unit/lib/utils/test_tar.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
import io
from unittest import TestCase
import tarfile
from unittest.mock import Mock, patch, call

from samcli.lib.utils.tar import create_tarball
from samcli.lib.utils.tar import extract_tarfile, create_tarball, _is_within_directory


class TestTar(TestCase):
Expand Down Expand Up @@ -88,3 +90,72 @@ def tar_filter(tar_info):
temp_file_mock.seek.assert_called_once_with(0)
temp_file_mock.close.assert_called_once()
tarfile_open_patch.assert_called_once_with(fileobj=temp_file_mock, mode="w")

@patch("samcli.lib.utils.tar.tarfile.open")
@patch("samcli.lib.utils.tar._is_within_directory")
def test_extract_tarfile_file_name(self, is_within_directory_patch, tarfile_open_patch):
tarfile_path = "/test_tarfile_path/"
unpack_dir = "/test_unpack_dir/"
is_within_directory_patch.return_value = True

tarfile_file_mock = Mock()
tar_file_obj_mock = Mock()
tar_file_obj_mock.name = "obj_name"
tarfile_file_mock.getmembers.return_value = [tar_file_obj_mock]
tarfile_open_patch.return_value.__enter__.return_value = tarfile_file_mock

extract_tarfile(tarfile_path=tarfile_path, unpack_dir=unpack_dir)

is_within_directory_patch.assert_called_once()
tarfile_file_mock.getmembers.assert_called_once()
tarfile_file_mock.extractall.assert_called_once_with(unpack_dir)

@patch("samcli.lib.utils.tar.tarfile.open")
@patch("samcli.lib.utils.tar._is_within_directory")
def test_extract_tarfile_fileobj(self, is_within_directory_patch, tarfile_open_patch):
stream_str = io.BytesIO(b"Hello World!")
unpack_dir = "/test_unpack_dir/"
is_within_directory_patch.return_value = True

tarfile_file_mock = Mock() # Mock tarfile
tar_file_obj_mock = Mock() # Mock member inside tarfile
tar_file_obj_mock.name = "obj_name"
tarfile_file_mock.getmembers.return_value = [tar_file_obj_mock]
tarfile_open_patch.return_value.__enter__.return_value = tarfile_file_mock

extract_tarfile(file_obj=stream_str, unpack_dir=unpack_dir)

is_within_directory_patch.assert_called_once()
tarfile_file_mock.getmembers.assert_called_once()
tarfile_file_mock.extractall.assert_called_once_with(unpack_dir)

@patch("samcli.lib.utils.tar.tarfile.open")
@patch("samcli.lib.utils.tar._is_within_directory")
def test_extract_tarfile_obj_not_within_dir(self, is_within_directory_patch, tarfile_open_patch):
tarfile_path = "/test_tarfile_path/"
unpack_dir = "/test_unpack_dir/"
is_within_directory_patch.return_value = False

tarfile_file_mock = Mock()
tar_file_obj_mock = Mock()
tar_file_obj_mock.name = "obj_name"
tarfile_file_mock.getmembers.return_value = [tar_file_obj_mock]
tarfile_open_patch.return_value.__enter__.return_value = tarfile_file_mock

with self.assertRaises(tarfile.ExtractError):
extract_tarfile(tarfile_path=tarfile_path, unpack_dir=unpack_dir)

is_within_directory_patch.assert_called_once()
tarfile_file_mock.getmembers.assert_called_once()

def test_tarfile_obj_is_within_dir(self):
directory = "/my/path"
target = "/my/path/file"

self.assertTrue(_is_within_directory(directory, target))

def test_tarfile_obj_is_not_within_dir(self):
directory = "/my/path"
target = "/another/path/file"

self.assertFalse(_is_within_directory(directory, target))
13 changes: 4 additions & 9 deletions tests/unit/local/docker/test_container.py
Original file line number Diff line number Diff line change
Expand Up @@ -839,8 +839,8 @@ def setUp(self):
self.container.id = "containerid"

@patch("samcli.local.docker.container.tempfile")
@patch("samcli.local.docker.container.tarfile")
def test_must_copy_files_from_container(self, tarfile_mock, tempfile_mock):
@patch("samcli.local.docker.container.extract_tarfile")
def test_must_copy_files_from_container(self, extract_tarfile_mock, tempfile_mock):
source = "source"
dest = "dest"

Expand All @@ -853,19 +853,14 @@ def test_must_copy_files_from_container(self, tarfile_mock, tempfile_mock):
tempfile_ctxmgr.__enter__ = Mock(return_value=fp_mock)
tempfile_ctxmgr.__exit__ = Mock()

tarfile_ctxmgr = tarfile_mock.open.return_value = Mock()
tar_mock = Mock()
tarfile_ctxmgr.return_value.__enter__ = Mock(return_value=tar_mock)
tarfile_ctxmgr.return_value.__exit__ = Mock()

self.container.copy(source, dest)

extract_tarfile_mock.assert_called_with(file_obj=fp_mock, unpack_dir=dest)

# Make sure archive data is written to the file
fp_mock.write.assert_has_calls([call(x) for x in tar_stream], any_order=False)

# Make sure we open the tarfile right and extract to right location
tarfile_mock.open.assert_called_with(fileobj=fp_mock, mode="r")
tar_mock.extractall(path=dest)

def test_raise_if_container_is_not_created(self):
source = "source"
Expand Down