CVE-2022-44900: Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal') (CWE-22)
value: A directory traversal vulnerability in the SevenZipFile.extractall() function of the python library py7zr v0.20.0 and earlier allows attackers to write arbitrary files via extracting a crafted 7z file.
CWE description: The product uses external input to construct a pathname that is intended to identify a file or directory that is located underneath a restricted parent directory, but the product does not properly neutralize special elements within the pathname that can cause the pathname to resolve to a location that is outside of the restricted directory.
CVSS v3 base score: 9.1
Published: 2022-12-06T20:15Z
Repository: https://github.com/miurahr/py7zr
Language: Python
Files changed: MANIFEST.in, py7zr/helpers.py, py7zr/py7zr.py, tests/test_extract.py, tests/test_zipslip.py
Commit date: 2022-11-01 07:04:13 +0900
Commit message: Fix sanity check for path traversal attack

- 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 <miurahr@linux.com>

Diff: diff --git a/MANIFEST.in b/MANIFEST.in
index bd9bd00..3c5d2cb 100644
--- a/MANIFEST.in
+++ b/MANIFEST.in
@@ -1,12 +1,10 @@
 include *.png
 include *.rst
 include *.svg
 include *.toml
 include LICENSE
-include .coveragerc
-include tox.ini
 include py7zr/py.typed
 recursive-include py7zr *.py
 recursive-include tests *.py
 recursive-include tests/data *.txt
 recursive-include tests/data *.7z
@@ -23,11 +21,10 @@ recursive-include docs *.py
 recursive-include docs *.rst
 recursive-include docs *.yml
 recursive-include docs *.odp
 recursive-include docs *.pdf
 recursive-include docs *.svg
-recursive-include docs *.txt
 recursive-include docs Makefile
 prune .github
 exclude .gitignore
 exclude .readthedocs.yml
 exclude codecov.yml
diff --git a/py7zr/helpers.py b/py7zr/helpers.py
index 77738f5..4e4e254 100644
--- a/py7zr/helpers.py
+++ b/py7zr/helpers.py
@@ -31,10 +31,12 @@ from datetime import datetime, timedelta, timezone, tzinfo
 from typing import BinaryIO, Optional, Union
 
 import _hashlib  # type: ignore  # noqa
 
 import py7zr.win32compat
+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 = "./"
 
 
@@ -263,11 +265,11 @@ 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:
         return is_symlink
     # special check for directory junctions which py38 does.
@@ -278,11 +280,11 @@ 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.
     """
     if sys.version_info >= (3, 9):
@@ -429,5 +431,59 @@ def remove_relative_path_marker(path: str) -> str:
 
     if path.startswith(RELATIVE_PATH_MARKER):
         processed_path = path[len(RELATIVE_PATH_MARKER) :]
 
     return processed_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 2a151f0..0d6e929 100644
--- a/py7zr/py7zr.py
+++ b/py7zr/py7zr.py
@@ -48,16 +48,17 @@ from py7zr.exceptions import Bad7zFile, CrcError, DecompressionError, InternalEr
 from py7zr.helpers import (
     ArchiveTimestamp,
     MemIO,
     NullIO,
     calculate_crc32,
+    check_archive_path,
     filetime_to_dt,
+    get_sanitized_output_path,
+    is_target_path_valid,
     readlink,
-    remove_relative_path_marker,
 )
 from py7zr.properties import DEFAULT_FILTERS, FILTER_DEFLATE64, MAGIC_7Z, get_default_blocksize, get_memory_limit
-from py7zr.win32compat import is_windows_native_python, is_windows_unc_path
 
 if sys.platform.startswith("win"):
     import _winapi
 
 FILE_ATTRIBUTE_UNIX_EXTENSION = 0x8000
@@ -565,38 +566,14 @@ class SevenZipFile(contextlib.AbstractContextManager):
                     outname = f.filename + "_%d" % i
                     if outname not in fnames:
                         break
                     i += 1
             fnames.append(outname)
-            # check f.filename has invalid directory traversals
-            if path is None:
-                # 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
-                try:
-                    pathlib.Path(os.getcwd()).joinpath(remove_relative_path_marker(outname)).relative_to(os.getcwd())
-                except ValueError:
-                    raise Bad7zFile
-                outfilename = pathlib.Path(remove_relative_path_marker(outname))
+            if path is None or path.is_absolute():
+                outfilename = get_sanitized_output_path(outname, path)
             else:
-                outfilename = path.joinpath(remove_relative_path_marker(outname))
-                try:
-                    outfilename.relative_to(path)
-                except ValueError:
-                    raise Bad7zFile
-            # 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 outfilename.is_absolute() and not is_windows_unc_path(outfilename):
-                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
             if return_dict:
                 if f.is_directory or f.is_socket:
@@ -632,16 +609,18 @@ class SevenZipFile(contextlib.AbstractContextManager):
                     raise DecompressionError("Directory {} making fails on unknown condition.".format(str(target_dir)))
 
         if callback is not None:
             self.worker.extract(
                 self.fp,
+                path,
                 parallel=(not self.password_protected and not self._filePassed),
                 q=self.q,
             )
         else:
             self.worker.extract(
                 self.fp,
+                path,
                 parallel=(not self.password_protected and not self._filePassed),
             )
 
         self.q.put(("post", None, None))
         # early return when dict specified
@@ -1038,10 +1017,15 @@ class SevenZipFile(contextlib.AbstractContextManager):
     def writed(self, targets: Dict[str, IO[Any]]) -> None:
         for target, input in targets.items():
             self.writef(input, target)
 
     def writef(self, bio: IO[Any], arcname: str):
+        if not check_archive_path(arcname):
+            raise ValueError(f"Specified path is bad: {arcname}")
+        return self._writef(bio, arcname)
+
+    def _writef(self, bio: IO[Any], arcname: str):
         if isinstance(bio, io.BytesIO):
             size = bio.getbuffer().nbytes
         elif isinstance(bio, io.TextIOBase):
             # First check whether is it Text?
             raise ValueError("Unsupported file object type: please open file with Binary mode.")
@@ -1067,16 +1051,21 @@ class SevenZipFile(contextlib.AbstractContextManager):
             self.header.files_info.files.append(file_info)
             self.header.files_info.emptyfiles.append(file_info["emptystream"])
             self.files.append(file_info)
 
     def writestr(self, data: Union[str, bytes, bytearray, memoryview], arcname: str):
+        if not check_archive_path(arcname):
+            raise ValueError(f"Specified path is bad: {arcname}")
+        return self._writestr(data, arcname)
+
+    def _writestr(self, data: Union[str, bytes, bytearray, memoryview], arcname: str):
         if not isinstance(arcname, str):
             raise ValueError("Unsupported arcname")
         if isinstance(data, str):
-            self.writef(io.BytesIO(data.encode("UTF-8")), arcname)
+            self._writef(io.BytesIO(data.encode("UTF-8")), arcname)
         elif isinstance(data, bytes) or isinstance(data, bytearray) or isinstance(data, memoryview):
-            self.writef(io.BytesIO(data), arcname)
+            self._writef(io.BytesIO(bytes(data)), arcname)
         else:
             raise ValueError("Unsupported data type.")
 
     def close(self):
         """Flush all the data into archive and close it.
@@ -1129,11 +1118,13 @@ class SevenZipFile(contextlib.AbstractContextManager):
         self.fp.seek(self.afterheader)
         self.worker = Worker(self.files, self.afterheader, self.header, self.mp)
         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 crce.args[2]
         else:
             return None
 
@@ -1198,47 +1189,49 @@ class Worker:
         if mp:
             self.concurrent: Union[Type[Thread], Type[Process]] = Process
         else:
             self.concurrent = Thread
 
-    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,
+                    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,
+                            path,
                             self.src_start + positions[i],
                             self.src_start + positions[i + 1],
                             q,
                             skip_notarget=skip_notarget,
                         )
                 else:
                     if getattr(fp, "name", None) is None:
                         raise InternalError("Caught unknown variable status error")
                     filename: str = getattr(fp, "name", "")  # do not become "" but it is for type check.
-                    self.extract_single(open(filename, "rb"), empty_files, 0, 0, q)
+                    self.extract_single(open(filename, "rb"), empty_files, path, 0, 0, q)
                     concurrent_tasks = []
                     exc_q: queue.Queue = queue.Queue()
                     for i in range(numfolders):
                         if skip_notarget:
                             if not any([self.target_filepath.get(f.id, None) for f in folders[i].files]):
@@ -1246,10 +1239,11 @@ class Worker:
                         p = self.concurrent(
                             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,
@@ -1264,16 +1258,17 @@ class Worker:
                     else:
                         exc_info = exc_q.get()
                         raise exc_info[1].with_traceback(exc_info[2])
         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,
+        path,
         src_start: int,
         src_end: int,
         q: Optional[queue.Queue],
         exc_q: Optional[queue.Queue] = None,
         skip_notarget=True,
@@ -1285,11 +1280,11 @@ class Worker:
             return
         try:
             if isinstance(fp, str):
                 fp = open(fp, "rb")
             fp.seek(src_start)
-            self._extract_single(fp, files, src_end, q, skip_notarget)
+            self._extract_single(fp, files, path, src_end, q, skip_notarget)
         except Exception as e:
             if exc_q is None:
                 raise e
             else:
                 exc_tuple = sys.exc_info()
@@ -1297,10 +1292,11 @@ class Worker:
 
     def _extract_single(
         self,
         fp: BinaryIO,
         files,
+        path,
         src_end: int,
         q: Optional[queue.Queue],
         skip_notarget=True,
     ) -> None:
         """
@@ -1329,24 +1325,32 @@ class Worker:
                 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")
-                            # 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
+                            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)
-                            sym_target = pathlib.Path(omfp.read().decode("utf-8"))
-                            # fileish.unlink(missing_ok=True) > py3.7
-                            if fileish.exists():
-                                fileish.unlink()
-                            fileish.symlink_to(sym_target)
+                            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:
                             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:
diff --git a/tests/test_extract.py b/tests/test_extract.py
index d8490d7..f08feba 100644
--- a/tests/test_extract.py
+++ b/tests/test_extract.py
@@ -330,11 +330,11 @@ def test_register_unpack_archive(tmp_path):
 def test_skip():
     archive = py7zr.SevenZipFile(testdata_path.joinpath("test_1.7z").open(mode="rb"))
     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()
 
 
 @pytest.mark.files
 def test_github_14_multi(tmp_path):
diff --git a/tests/test_zipslip.py b/tests/test_zipslip.py
new file mode 100644
index 0000000..2ddbef9
--- /dev/null
+++ b/tests/test_zipslip.py
@@ -0,0 +1,48 @@
+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")
+    good_data = b"#!/bin/sh\necho good\n"
+    good_path = "good.sh"
+    bad_data = b"!#/bin/sh\necho bad\n"
+    bad_path = "../../.../../../../../../tmp/evil.sh"
+    with SevenZipFile(target, "w", filters=my_filters) as archive:
+        archive.writestr(good_data, good_path)
+        archive._writestr(bad_data, bad_path)  # bypass a path check
+    with pytest.raises(Bad7zFile):
+        with SevenZipFile(target, "r") as archive:
+            archive.extractall(path=tmp_path)

Vulnerable code: include .coveragerc
include tox.ini
recursive-include docs *.txt
    Supports Windows NT symbolic links and reparse points.
    Supports Windows NT symbolic links and reparse points.
    remove_relative_path_marker,
from py7zr.win32compat import is_windows_native_python, is_windows_unc_path
            # check f.filename has invalid directory traversals
            if path is None:
                # 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
                try:
                    pathlib.Path(os.getcwd()).joinpath(remove_relative_path_marker(outname)).relative_to(os.getcwd())
                except ValueError:
                    raise Bad7zFile
                outfilename = pathlib.Path(remove_relative_path_marker(outname))
                outfilename = path.joinpath(remove_relative_path_marker(outname))
                try:
                    outfilename.relative_to(path)
                except ValueError:
                    raise Bad7zFile
            # 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 outfilename.is_absolute() and not is_windows_unc_path(outfilename):
                outfilename = pathlib.WindowsPath("\\\\?\\" + str(outfilename))
            self.writef(io.BytesIO(data.encode("UTF-8")), arcname)
            self.writef(io.BytesIO(data), arcname)
            self.worker.extract(self.fp, parallel=(not self.password_protected), skip_notarget=False)  # TODO: print progress
    def extract(self, fp: BinaryIO, parallel: bool, skip_notarget=True, q=None) -> None:
                    self.extract_single(fp, empty_files, 0, 0, q)
                    self.extract_single(open(filename, "rb"), empty_files, 0, 0, q)
            self.extract_single(fp, empty_files, 0, 0, q)
            self._extract_single(fp, files, src_end, q, skip_notarget)
                            # 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
                            sym_target = pathlib.Path(omfp.read().decode("utf-8"))
                            # fileish.unlink(missing_ok=True) > py3.7
                            if fileish.exists():
                                fileish.unlink()
                            fileish.symlink_to(sym_target)
    archive.worker.extract(archive.fp, parallel=True)
Fixed code: from py7zr import Bad7zFile
from py7zr.win32compat import is_windows_native_python, is_windows_unc_path
    Support Windows NT symbolic links and reparse points.
    Support Windows NT symbolic links and reparse points.


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
    check_archive_path,
    get_sanitized_output_path,
    is_target_path_valid,
            if path is None or path.is_absolute():
                outfilename = get_sanitized_output_path(outname, path)
                outfilename = get_sanitized_output_path(outname, pathlib.Path(os.getcwd()).joinpath(path))
                path,
                path,
        if not check_archive_path(arcname):
            raise ValueError(f"Specified path is bad: {arcname}")
        return self._writef(bio, arcname)

    def _writef(self, bio: IO[Any], arcname: str):
        if not check_archive_path(arcname):
            raise ValueError(f"Specified path is bad: {arcname}")
        return self._writestr(data, arcname)

    def _writestr(self, data: Union[str, bytes, bytearray, memoryview], arcname: str):
            self._writef(io.BytesIO(data.encode("UTF-8")), arcname)
            self._writef(io.BytesIO(bytes(data)), arcname)
            self.worker.extract(
                self.fp, None, parallel=(not self.password_protected), skip_notarget=False
            )  # TODO: print progress
    def extract(self, fp: BinaryIO, path: Optional[pathlib.Path], parallel: bool, skip_notarget=True, q=None) -> None:
                    path,
                    self.extract_single(fp, empty_files, path, 0, 0, q)
                            path,
                    self.extract_single(open(filename, "rb"), empty_files, path, 0, 0, q)
                                path,
            self.extract_single(fp, empty_files, path, 0, 0, q)
        path,
            self._extract_single(fp, files, path, src_end, q, skip_notarget)
        path,
                            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.")
                            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.")
    archive.worker.extract(archive.fp, None, parallel=True)
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")
    good_data = b"#!/bin/sh\necho good\n"
    good_path = "good.sh"
    bad_data = b"!#/bin/sh\necho bad\n"
    bad_path = "../../.../../../../../../tmp/evil.sh"
    with SevenZipFile(target, "w", filters=my_filters) as archive:
        archive.writestr(good_data, good_path)
        archive._writestr(bad_data, bad_path)  # bypass a path check
    with pytest.raises(Bad7zFile):
        with SevenZipFile(target, "r") as archive:
            archive.extractall(path=tmp_path)