From: Hiroshi Miura Date: Sun, 30 Oct 2022 14:27:46 +0900 Subject: Fix sanity check for path traversal attack Forwarded: not-needed Bug-Debian: https://bugs.debian.org/cgi-bin/bugreport.cgi?bug=1032091 Origin: backport, https://github.com/miurahr/py7zr/commit/1bb43f17515c7f69673a1c88ab9cc72a7bbef406 Author: Hiroshi Miura - Previous versions do not detect the attack in some case - fixed it by call resolve() - resolve() converts "/hoge/fuga/../../../tmp/evil.sh" to be "/tmp/evil.sh" then relative_to() can detect path traversal attack. - Add path checker in writef() and writestr() methods - When pass arcname as evil path such as "../../../../tmp/evil.sh" it raises ValueError - Add test case of bad path detection - extraction: check symlink and junction is under target folder - Fix relative_path_marker removal - Don't put windows file namespace to output file path Signed-off-by: Hiroshi Miura --- MANIFEST.in | 1 - py7zr/helpers.py | 134 +++++++++++++++++++++++++++++++++++++++++++++- py7zr/py7zr.py | 145 +++++++++++++++++++++++++++----------------------- py7zr/win32compat.py | 13 +++++ tests/test_extract.py | 2 +- tests/test_zipslip.py | 54 +++++++++++++++++++ 6 files changed, 279 insertions(+), 70 deletions(-) create mode 100644 tests/test_zipslip.py diff --git a/MANIFEST.in b/MANIFEST.in index 95d2263..63b919e 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -3,7 +3,6 @@ include *.svg include *.toml include *.yml include LICENSE -include .coveragerc include tox.ini recursive-include py7zr *.py recursive-include tests *.py diff --git a/py7zr/helpers.py b/py7zr/helpers.py index 8da7218..89ec89c 100644 --- a/py7zr/helpers.py +++ b/py7zr/helpers.py @@ -37,6 +37,11 @@ import _hashlib # type: ignore # noqa import py7zr.win32compat from py7zr.properties import READ_BLOCKSIZE +from py7zr import Bad7zFile +from py7zr.win32compat import is_windows_native_python, is_windows_unc_path + +# String used at the beginning of relative paths +RELATIVE_PATH_MARKER = "./" def calculate_crc32(data: bytes, value: int = 0, blocksize: int = 1024 * 1024) -> int: @@ -249,7 +254,7 @@ class ArchiveTimestamp(int): def islink(path): """ Cross-platform islink implementation. - Supports Windows NT symbolic links and reparse points. + Support Windows NT symbolic links and reparse points. """ is_symlink = os.path.islink(str(path)) if sys.version_info >= (3, 8) or sys.platform != "win32" or sys.getwindowsversion()[0] < 6: @@ -264,7 +269,7 @@ def islink(path): def readlink(path: Union[str, pathlib.Path], *, dir_fd=None) -> Union[str, pathlib.Path]: """ Cross-platform compat implementation of os.readlink and Path.readlink(). - Supports Windows NT symbolic links and reparse points. + Support Windows NT symbolic links and reparse points. When called with path argument as pathlike(str), return result as a pathlike(str). When called with Path object, return also Path object. When called with path argument as bytes, return result as a bytes. @@ -447,3 +452,128 @@ class BufferedRW(io.BufferedIOBase): def __len__(self): return len(self._buf) + + +def remove_relative_path_marker(path: str) -> str: + """ + Removes './' from the beginning of a path-like string + """ + processed_path = path + + if path.startswith(RELATIVE_PATH_MARKER): + processed_path = path[len(RELATIVE_PATH_MARKER) :] + + return processed_path + + +def canonical_path(target: pathlib.PurePath) -> pathlib.PurePath: + """Return a canonical path of target argument.""" + stack: List[str] = [] + for p in target.parts: + if p != ".." or len(stack) == 0: + stack.append(p) + continue + # treat '..' + if stack[-1] == "..": + stack.append(p) # '../' + '../' -> '../../' + elif stack[-1] == "/": + pass # '/' + '../' -> '/' + else: + stack.pop() # 'foo/boo/' + '..' -> 'foo/' + return pathlib.PurePath(*stack) + + +def is_relative_to(my: pathlib.PurePath, *other) -> bool: + """Return True when path is relative to other path, otherwise False.""" + try: + my.relative_to(*other) + except ValueError: + return False + return True + + +def get_sanitized_output_path(fname: str, path: Optional[pathlib.Path]) -> pathlib.Path: + """ + check f.filename has invalid directory traversals + When condition is not satisfied, raise Bad7zFile + """ + if path is None: + target_path = canonical_path(pathlib.Path.cwd().joinpath(fname)) + if is_relative_to(target_path, pathlib.Path.cwd()): + return pathlib.Path(remove_relative_path_marker(fname)) + else: + outfile = canonical_path(path.joinpath(remove_relative_path_marker(fname))) + if is_relative_to(outfile, path): + return pathlib.Path(outfile) + raise Bad7zFile(f"Specified path is bad: {fname}") + + +def check_archive_path(arcname: str) -> bool: + """ + Check arcname argument is valid for archive. + It should not be absolute, if so it returns False. + It should not be evil traversal attack path. + Otherwise, returns True. + """ + if pathlib.PurePath(arcname).is_absolute(): + return False + # test against dummy parent path + if sys.platform == "win32": + path = pathlib.Path("C:/foo/boo/fuga/hoge/a90sufoiasj09/dafj08sajfa/") + else: + path = pathlib.Path("/foo/boo/fuga/hoge/a90sufoiasj09/dafj08sajfa/") + return is_path_valid(path.joinpath(arcname), path) + + +def get_sanitized_output_path(fname: str, path: Optional[pathlib.Path]) -> pathlib.Path: + """ + check f.filename has invalid directory traversals + do following but is_relative_to introduced in py 3.9, + so I replaced it with relative_to. when condition is not satisfied, raise ValueError + if not pathlib.Path(...).joinpath(remove_relative_path_marker(outname)).is_relative_to(...): + raise Bad7zFile + """ + if path is None: + try: + pathlib.Path(os.getcwd()).joinpath(fname).resolve().relative_to(os.getcwd()) + outfile = pathlib.Path(remove_relative_path_marker(fname)) + except ValueError: + raise Bad7zFile(f"Specified path is bad: {fname}") + else: + try: + outfile = path.joinpath(remove_relative_path_marker(fname)) + outfile.resolve().relative_to(path) + except ValueError: + raise Bad7zFile(f"Specified path is bad: {fname}") + return outfile + + +def check_archive_path(arcname: str) -> bool: + path = pathlib.Path("/foo/boo/fuga/hoge/a90sufoiasj09/dafj08sajfa/") # dummy path + return is_target_path_valid(path, path.joinpath(arcname)) + + +def is_target_path_valid(path: pathlib.Path, target: pathlib.Path) -> bool: + try: + if path.is_absolute(): + target.resolve().relative_to(path) + else: + target.resolve().relative_to(pathlib.Path(os.getcwd()).joinpath(path)) + except ValueError: + return False + return True + + +def check_win32_file_namespace(pathname: pathlib.Path) -> pathlib.Path: + # When python on Windows and not python on Cygwin, + # Add win32 file namespace to exceed Microsoft Windows + # path length limitation to 260 bytes + # ref. + # https://docs.microsoft.com/en-us/windows/win32/fileio/naming-a-file + # In editions of Windows before Windows 10 version 1607, + # the maximum length for a path is MAX_PATH, which is defined as + # 260 characters. In later versions of Windows, changing a registry key + # or select option when python installation is required to remove the limit. + if is_windows_native_python() and pathname.is_absolute() and not is_windows_unc_path(pathname): + pathname = pathlib.WindowsPath("\\\\?\\" + str(pathname)) + return pathname diff --git a/py7zr/py7zr.py b/py7zr/py7zr.py index 706f1f8..fbc899e 100644 --- a/py7zr/py7zr.py +++ b/py7zr/py7zr.py @@ -41,7 +41,7 @@ from py7zr.archiveinfo import Folder, Header, SignatureHeader from py7zr.callbacks import ExtractCallback from py7zr.compressor import SupportedMethods, get_methods_names_string from py7zr.exceptions import Bad7zFile, CrcError, DecompressionError, InternalError, UnsupportedCompressionMethodError -from py7zr.helpers import ArchiveTimestamp, MemIO, NullIO, calculate_crc32, filetime_to_dt, readlink +from py7zr.helpers import ArchiveTimestamp, MemIO, NullIO, calculate_crc32, check_archive_path, filetime_to_dt, get_sanitized_output_path, is_target_path_valid, readlink from py7zr.properties import ARCHIVE_DEFAULT, ENCRYPTED_ARCHIVE_DEFAULT, MAGIC_7Z, READ_BLOCKSIZE if sys.platform.startswith('win'): @@ -483,14 +483,10 @@ class SevenZipFile(contextlib.AbstractContextManager): if outname not in fnames: break fnames.append(outname) - if path is not None: - outfilename = path.joinpath(outname) + if path is None or path.is_absolute(): + outfilename = get_sanitized_output_path(outname, path) else: - outfilename = pathlib.Path(outname) - if os.name == 'nt': - if outfilename.is_absolute(): - # hack for microsoft windows path length limit < 255 - outfilename = pathlib.WindowsPath('\\\\?\\' + str(outfilename)) + outfilename = get_sanitized_output_path(outname, pathlib.Path(os.getcwd()).joinpath(path)) if targets is not None and f.filename not in targets: self.worker.register_filelike(f.id, None) continue @@ -535,52 +531,35 @@ class SevenZipFile(contextlib.AbstractContextManager): try: if callback is not None: - self.worker.extract(self.fp, parallel=(not self.password_protected and not self._filePassed), q=self.q) + self.worker.extract(self.fp, path, parallel=(not self.password_protected and not self._filePassed), q=self.q) else: - self.worker.extract(self.fp, parallel=(not self.password_protected and not self._filePassed)) + self.worker.extract(self.fp, path, parallel=(not self.password_protected and not self._filePassed)) except CrcError as ce: raise Bad7zFile("CRC32 error on archived file {}.".format(str(ce))) self.q.put(('post', None, None)) if return_dict: return self._dict - else: - # create symbolic links on target path as a working directory. - # if path is None, work on current working directory. - for t in target_sym: - sym_dst = t.resolve() - with sym_dst.open('rb') as b: - sym_src = b.read().decode(encoding='utf-8') # symlink target name stored in utf-8 - sym_dst.unlink() # unlink after close(). - sym_dst.symlink_to(pathlib.Path(sym_src)) - # create junction point only on windows platform - if sys.platform.startswith('win'): - for t in target_junction: - junction_dst = t.resolve() - with junction_dst.open('rb') as b: - junction_target = pathlib.Path(b.read().decode(encoding='utf-8')) - junction_dst.unlink() - _winapi.CreateJunction(junction_target, str(junction_dst)) # type: ignore # noqa - # set file properties - for outfilename, properties in target_files: - # mtime - lastmodified = None - try: - lastmodified = ArchiveTimestamp(properties['lastwritetime']).totimestamp() - except KeyError: - pass - if lastmodified is not None: - os.utime(str(outfilename), times=(lastmodified, lastmodified)) - if os.name == 'posix': - st_mode = properties['posix_mode'] - if st_mode is not None: - outfilename.chmod(st_mode) - continue - # fallback: only set readonly if specified - if properties['readonly'] and not properties['is_directory']: - ro_mask = 0o777 ^ (stat.S_IWRITE | stat.S_IWGRP | stat.S_IWOTH) - outfilename.chmod(outfilename.stat().st_mode & ro_mask) - return None + # set file properties + for outfilename, properties in target_files: + # mtime + lastmodified = None + try: + lastmodified = ArchiveTimestamp(properties['lastwritetime']).totimestamp() + except KeyError: + pass + if lastmodified is not None: + os.utime(str(outfilename), times=(lastmodified, lastmodified)) + if os.name == 'posix': + st_mode = properties['posix_mode'] + if st_mode is not None: + outfilename.chmod(st_mode) + continue + # fallback: only set readonly if specified + if properties['readonly'] and not properties['is_directory']: + ro_mask = 0o777 ^ (stat.S_IWRITE | stat.S_IWGRP | stat.S_IWOTH) + outfilename.chmod(outfilename.stat().st_mode & ro_mask) + return None def _prepare_append(self, filters, password): if password is not None and filters is None: @@ -940,7 +919,7 @@ class SevenZipFile(contextlib.AbstractContextManager): for f in self.files: self.worker.register_filelike(f.id, None) try: - self.worker.extract(self.fp, parallel=(not self.password_protected), skip_notarget=False) # TODO: print progress + self.worker.extract(self.fp, None, parallel=(not self.password_protected), skip_notarget=False) # TODO: print progress except CrcError as crce: return str(crce) else: @@ -998,28 +977,28 @@ class Worker: self.current_file_index = len(self.files) self.last_file_index = len(self.files) - def extract(self, fp: BinaryIO, parallel: bool, skip_notarget=True, q=None) -> None: + def extract(self, fp: BinaryIO, path: Optional[pathlib.Path], parallel: bool, skip_notarget=True, q=None) -> None: """Extract worker method to handle 7zip folder and decompress each files.""" if hasattr(self.header, 'main_streams') and self.header.main_streams is not None: src_end = self.src_start + self.header.main_streams.packinfo.packpositions[-1] numfolders = self.header.main_streams.unpackinfo.numfolders if numfolders == 1: - self.extract_single(fp, self.files, self.src_start, src_end, q, skip_notarget=skip_notarget) + self.extract_single(fp, self.files, path, self.src_start, src_end, q, skip_notarget=skip_notarget) else: folders = self.header.main_streams.unpackinfo.folders positions = self.header.main_streams.packinfo.packpositions empty_files = [f for f in self.files if f.emptystream] if not parallel: - self.extract_single(fp, empty_files, 0, 0, q) + self.extract_single(fp, empty_files, path, 0, 0, q) for i in range(numfolders): if skip_notarget: if not any([self.target_filepath.get(f.id, None) for f in folders[i].files]): continue - self.extract_single(fp, folders[i].files, self.src_start + positions[i], + self.extract_single(fp, folders[i].files, path, self.src_start + positions[i], self.src_start + positions[i + 1], q, skip_notarget=skip_notarget) else: filename = getattr(fp, 'name', None) - self.extract_single(open(filename, 'rb'), empty_files, 0, 0, q) + self.extract_single(open(filename, 'rb'), empty_files, path, 0, 0, q) extract_threads = [] exc_q = queue.Queue() # type: queue.Queue for i in range(numfolders): @@ -1028,6 +1007,7 @@ class Worker: continue p = threading.Thread(target=self.extract_single, args=(filename, folders[i].files, + path, self.src_start + positions[i], self.src_start + positions[i + 1], q, exc_q, skip_notarget)) p.start() @@ -1041,9 +1021,9 @@ class Worker: raise exc_type(exc_val).with_traceback(exc_tb) else: empty_files = [f for f in self.files if f.emptystream] - self.extract_single(fp, empty_files, 0, 0, q) + self.extract_single(fp, empty_files, path, 0, 0, q) - def extract_single(self, fp: Union[BinaryIO, str], files, src_start: int, src_end: int, + def extract_single(self, fp: Union[BinaryIO, str], files, path, src_start: int, src_end: int, q: Optional[queue.Queue], exc_q: Optional[queue.Queue] = None, skip_notarget=True) -> None: """Single thread extractor that takes file lists in single 7zip folder.""" if files is None: @@ -1057,22 +1037,55 @@ class Worker: if q is not None: q.put(('s', str(f.filename), str(f.compressed) if f.compressed is not None else '0')) fileish = self.target_filepath.get(f.id, None) - if fileish is not None: + if fileish is None: + if not f.emptystream: + just_check.append(f) + else: # delayed execution of crc check. self._check(fp, just_check, src_end) just_check = [] fileish.parent.mkdir(parents=True, exist_ok=True) - with fileish.open(mode='wb') as ofp: - if not f.emptystream: - # extract to file - crc32 = self.decompress(fp, f.folder, ofp, f.uncompressed, f.compressed, src_end) - ofp.seek(0) - if f.crc32 is not None and crc32 != f.crc32: - raise CrcError("{}".format(f.filename)) + if not f.emptystream: + if f.is_junction and not isinstance(fileish, MemIO) and sys.platform == "win32": + with io.BytesIO() as ofp: + self.decompress(fp, f.folder, ofp, f.uncompressed, f.compressed, src_end) + dst: str = ofp.read().decode("utf-8") + if is_target_path_valid(path, fileish.parent.joinpath(dst)): + # fileish.unlink(missing_ok=True) > py3.7 + if fileish.exists(): + fileish.unlink() + if sys.platform == "win32": # hint for mypy + _winapi.CreateJunction(str(fileish), dst) # noqa + else: + raise Bad7zFile("Junction point out of target directory.") + elif f.is_symlink and not isinstance(fileish, MemIO): + with io.BytesIO() as omfp: + self.decompress(fp, f.folder, omfp, f.uncompressed, f.compressed, src_end) + omfp.seek(0) + dst = omfp.read().decode("utf-8") + # check sym_target points inside an archive target? + if is_target_path_valid(path, fileish.parent.joinpath(dst)): + sym_target = pathlib.Path(dst) + # fileish.unlink(missing_ok=True) > py3.7 + if fileish.exists(): + fileish.unlink() + fileish.symlink_to(sym_target) + else: + raise Bad7zFile("Symlink point out of target directory.") + else: + with fileish.open(mode='wb') as obfp: + # extract to file + crc32 = self.decompress(fp, f.folder, obfp, f.uncompressed, f.compressed, src_end) + obfp.seek(0) + if f.crc32 is not None and crc32 != f.crc32: + raise CrcError("{}".format(f.filename)) + else: + # just create empty file + if not isinstance(fileish, MemIO): + fileish.touch() else: - pass # just create empty file - elif not f.emptystream: - just_check.append(f) + with fileish.open() as ofp: + pass if q is not None: q.put(('e', str(f.filename), str(f.uncompressed))) if not skip_notarget: diff --git a/py7zr/win32compat.py b/py7zr/win32compat.py index b2c3635..0771a65 100644 --- a/py7zr/win32compat.py +++ b/py7zr/win32compat.py @@ -169,3 +169,16 @@ if sys.platform == "win32": return pathlib.WindowsPath(rpath) else: return rpath + + +def is_windows_native_python() -> bool: + return ( + sys.platform == "win32" + and os.name == "nt" + and "cygwin" not in platform.system().lower() + and "cygwin" not in sys.platform + ) + + +def is_windows_unc_path(path) -> bool: + return sys.platform == "win32" and path.drive.startswith("\\\\") diff --git a/tests/test_extract.py b/tests/test_extract.py index ae7f2fd..633b76f 100644 --- a/tests/test_extract.py +++ b/tests/test_extract.py @@ -253,7 +253,7 @@ def test_skip(): for i, cf in enumerate(archive.files): assert cf is not None archive.worker.register_filelike(cf.id, None) - archive.worker.extract(archive.fp, parallel=True) + archive.worker.extract(archive.fp, None, parallel=True) archive.close() diff --git a/tests/test_zipslip.py b/tests/test_zipslip.py new file mode 100644 index 0000000..27a1657 --- /dev/null +++ b/tests/test_zipslip.py @@ -0,0 +1,54 @@ +import os + +import pytest + +from py7zr import SevenZipFile +from py7zr.exceptions import Bad7zFile +from py7zr.helpers import check_archive_path, get_sanitized_output_path +from py7zr.properties import FILTER_LZMA2, PRESET_DEFAULT + +testdata_path = os.path.join(os.path.dirname(__file__), "data") + + +@pytest.mark.misc +def test_check_archive_path(): + bad_path = "../../.../../../../../../tmp/evil.sh" + assert not check_archive_path(bad_path) + + +@pytest.mark.misc +def test_get_sanitized_output_path_1(tmp_path): + bad_path = "../../.../../../../../../tmp/evil.sh" + with pytest.raises(Bad7zFile): + get_sanitized_output_path(bad_path, tmp_path) + + +@pytest.mark.misc +def test_get_sanitized_output_path_2(tmp_path): + good_path = "good.sh" + expected = tmp_path.joinpath(good_path) + assert expected == get_sanitized_output_path(good_path, tmp_path) + + +@pytest.mark.misc +def test_extract_path_traversal_attack(tmp_path): + my_filters = [ + {"id": FILTER_LZMA2, "preset": PRESET_DEFAULT}, + ] + target = tmp_path.joinpath("target.7z") + + from base64 import b64decode + data = """ + N3q8ryccAATOVjH2nAAAAAAAAAAVAAAAAAAAABe26oXgACYAIV0AEYhCRj30FjRzCg2kNp3mcg12 + I+GegsuiKVIVM4p1+AQVAOAAqgBrXQAAgTMHrg/QluR8nz9HQQQEMnr/nRksOfngonWcaGZa9yk2 + MshKgENm9F8IrHzRtlxA1rG7ojA2dU0VubucO3DWfQiBaqCqtidfv5DCj8LxFZ7PbZYke+rw7nKV + bAoyU7Z/s4bc+kz6VLwXgAAXBikBCXMABwsBAAEhIQEYDICrAAA= + """ + bindata = b64decode(data) + + with open(target, "wb") as archive: + archive.write(bindata) + + with pytest.raises(Bad7zFile): + with SevenZipFile(target, "r") as archive: + archive.extractall(path=tmp_path)