feat: add link checking for BinaryTar (#2283)

Co-authored-by: piglei <piglei2007@gmail.com>

Author: jamesgetx

apiserver/paasng/paasng/platform/sourcectl/package/client.py

@@ -18,6 +18,7 @@
 import abc
 import logging
 import os
+import os.path
 import re
 import subprocess
 import tarfile
@@ -97,7 +98,7 @@ def __init__(
             raise ValueError("nothing to open")
         if file_path and file_obj:
             raise ValueError("file_path and file_obj cannot be provided at the same time")
-        self.tar = tarfile.open(name=file_path, fileobj=file_obj, mode=mode)
+        self.tar = tarfile.open(name=file_path, fileobj=file_obj, mode=mode)  # noqa: SIM115
         self.relative_path = relative_path
 
     def read_file(self, file_path: str) -> bytes:
@@ -130,8 +131,8 @@ class BinaryTarClient(BasePackageClient):
     When handling big tarball(>=100mb), will faster than tarfile 10 seconds in the special testcase.
     """
 
-    def __init__(self, filepath: Union[str, Path]):
-        self.filepath = Path(filepath)
+    def __init__(self, file_path: Union[str, Path]):
+        self.filepath = Path(file_path)
 
     def read_file(self, filename) -> bytes:
         """Extract a filename from the archive as bytes.
@@ -140,6 +141,7 @@ def read_file(self, filename) -> bytes:
         :return: bytes contents of the file.
         :raises InvalidPackageFileFormatError: The file is not a valid tar file, it's content
             might be corrupt.
+        :raise RuntimeError: Raised if unexpected errors occur.
         """
         with generate_temp_dir() as temp_dir:
             p = subprocess.Popen(
@@ -157,14 +159,32 @@ def read_file(self, filename) -> bytes:
                     raise FileDoesNotExistError(f"Failed to extractfile from the tarball, error: {stderr!r}")
                 else:
                     raise RuntimeError(f"Failed to extractfile from the tarball, error: {stderr!r}")
-            return (temp_dir / filename).read_bytes()
+
+            filepath = temp_dir / filename
+
+            # Check if the file is a symbolic link and it's inside the directory
+            real_path = os.path.realpath(filepath)
+            if os.path.commonpath([temp_dir, real_path]) != str(temp_dir):
+                raise RuntimeError(f"Extracted file {filepath} is outside the target directory.")
+            return filepath.read_bytes()
 
     def export(self, local_path: str):
         """Extract all members from the archive to the current working directory
+
         :param working_dir: working directory
+        :raise RuntimeError: Raised if unexpected errors occur.
         """
         uncompress_directory(source_path=self.filepath, target_path=local_path)
 
+        # Security check: traverse the directory to check if any link files is outside the target directory
+        for dirpath, dirnames, filenames in os.walk(local_path):
+            for name in dirnames + filenames:
+                if not Path(os.path.join(dirpath, name)).is_symlink():
+                    continue
+                real_path = os.path.realpath(os.path.join(dirpath, name))
+                if os.path.commonpath([local_path, real_path]) != str(local_path):
+                    raise RuntimeError(f"Extracted file {real_path} is outside the target directory.")
+
     def close(self):
         """Nothing need to close."""
 

apiserver/paasng/tests/paasng/platform/sourcectl/packages/test_client.py

@@ -155,6 +155,75 @@ def test_export(self, client_cls, archive_maker, contents):
             assert {str(child.relative_to(working_dir)) for child in working_dir.iterdir()} == set(contents.keys())
 
 
+@pytest.mark.parametrize(
+    ("client_cls", "archive_maker"),
+    [
+        (TarClient, gen_tar),
+        (GenericLocalClient, gen_tar),
+    ],
+)
+class TestTarClientsShouldNotReadOutside:
+    @pytest.mark.parametrize(
+        ("symbolic_links", "filename"),
+        [
+            ({"passwd": "/etc/passwd"}, "./passwd"),
+            ({"passwd": "../../../../../../../../../etc/passwd"}, "./passwd"),
+        ],
+    )
+    def test_read_file_should_fail(self, client_cls, archive_maker, symbolic_links, filename):
+        with generate_temp_file() as file_path:
+            archive_maker(file_path, symbolic_links=symbolic_links)
+            cli: BasePackageClient = client_cls(file_path=file_path)
+            with pytest.raises(KeyError, match="linkname .* not found"):
+                cli.read_file(filename)
+
+    @pytest.mark.parametrize(
+        "symbolic_links",
+        [
+            ({"passwd": "/etc"}),
+            ({"passwd": "../../../../../../../../../etc/passwd"}),
+        ],
+    )
+    def test_export_should_fail(self, client_cls, archive_maker, symbolic_links):
+        with generate_temp_file() as file_path, generate_temp_dir() as working_dir:
+            archive_maker(file_path, symbolic_links=symbolic_links)
+            cli = client_cls(file_path)
+            with pytest.raises(tarfile.FilterError):  # type: ignore[attr-defined]
+                cli.export(working_dir)
+
+    # TODO: Add malformed tar file which uses bad names
+
+
+class TestBinaryTarClientsShouldNotReadOutside:
+    @pytest.mark.parametrize(
+        ("symbolic_links", "filename"),
+        [
+            ({"passwd": "/etc/passwd"}, "./passwd"),
+            ({"passwd": "../../../../../../../../../etc/passwd"}, "./passwd"),
+        ],
+    )
+    def test_read_file_should_fail(self, symbolic_links, filename):
+        with generate_temp_file() as file_path:
+            gen_tar(file_path, symbolic_links=symbolic_links)
+            cli = BinaryTarClient(file_path=file_path)
+            with pytest.raises(RuntimeError, match=".*outside the target directory."):
+                cli.read_file(filename)
+
+    @pytest.mark.parametrize(
+        "symbolic_links",
+        [
+            ({"passwd": "/etc"}),
+            ({"passwd": "../../../../../../../../../etc/passwd"}),
+        ],
+    )
+    def test_export_should_fail(self, symbolic_links):
+        with generate_temp_file() as file_path, generate_temp_dir() as working_dir:
+            gen_tar(file_path, symbolic_links=symbolic_links)
+            cli = BinaryTarClient(file_path)
+            with pytest.raises(RuntimeError):
+                cli.export(str(working_dir))
+
+
 @pytest.mark.parametrize("archive_maker", [gen_tar, gen_zip])
 class TestGenericRemoteClient:
     @pytest.mark.parametrize(

apiserver/paasng/tests/paasng/platform/sourcectl/packages/utils.py

@@ -57,30 +57,45 @@ def ensure_parent_exists(path: Path):
 
 
 @contextmanager
-def dump_contents_to_fs(contents: Dict[str, Union[str, bytes]]):
+def dump_contents_to_fs(
+    contents: Dict[str, Union[str, bytes]] | None = None,
+    symbolic_links: Dict[str, str] | None = None,
+):
     """a helper to dumps contents fo file-system
 
     :param contents Dict[str, str]: a dict, the key is filename, and value is the content.
+    :param symbolic_links: a dict, format: {link_name: target_file}.
     """
     with generate_temp_dir() as source_dir:
-        for file_name, content in contents.items():
-            target = source_dir / file_name
-            ensure_parent_exists(target)
-
-            if isinstance(content, bytes):
-                target.write_bytes(content)
-            else:
-                target.write_text(content)
+        if contents:
+            for file_name, content in contents.items():
+                target = source_dir / file_name
+                ensure_parent_exists(target)
+
+                if isinstance(content, bytes):
+                    target.write_bytes(content)
+                else:
+                    target.write_text(content)
+
+        if symbolic_links:
+            for link_name, target_file in symbolic_links.items():
+                target = source_dir / link_name
+                ensure_parent_exists(target)
+                target.symlink_to(target_file, target_is_directory=Path(target_file).is_dir())
         yield source_dir
 
 
-def gen_tar(target_path, contents: Dict[str, Union[str, bytes]]):
-    with dump_contents_to_fs(contents) as source_dir:
+def gen_tar(
+    target_path, contents: Dict[str, Union[str, bytes]] | None = None, symbolic_links: Dict[str, str] | None = None
+):
+    with dump_contents_to_fs(contents=contents, symbolic_links=symbolic_links) as source_dir:
         compress_directory(source_dir, target_path)
 
 
-def gen_zip(target_path, contents: Dict[str, Union[str, bytes]]):
-    with dump_contents_to_fs(contents) as source_dir:
+def gen_zip(
+    target_path, contents: Dict[str, Union[str, bytes]] | None = None, symbolic_links: Dict[str, str] | None = None
+):
+    with dump_contents_to_fs(contents=contents, symbolic_links=symbolic_links) as source_dir:
         filename = shutil.make_archive(target_path, format="zip", root_dir=source_dir)
         if filename != target_path:
             shutil.move(filename, target_path)