(buildstream) 01/01: tar.py: Always use `tar_filter`

[ju...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

juergbi pushed a commit to branch jbilleter/tar
in repository https://gitbox.apache.org/repos/asf/buildstream.git

commit 149b47f6b62af6b70530425f11bec7a6d35a4e28
Author: Jürg Billeter <j...@bitron.ch>
AuthorDate: Fri Aug 28 16:06:59 2026 +0200

    tar.py: Always use `tar_filter`
    
    `tar_filter` covers cases that `_assert_safe()` doesn't. As `tar_filter`
    is available in updates of all supported Python branches, always use it.
    Bail out early when running on an outdated Python version.
    
    The hardlink target check from `_assert_safe()` is still needed. Define
    a filter function that calls `tar_filter` and then performs a more
    robust version of the hardlink target check.
    
    Reported-by: Gjoko Krstic <gj...@zeroscience.mk>
---
 src/buildstream/plugins/sources/tar.py | 52 ++++++++++++++--------------------
 1 file changed, 22 insertions(+), 30 deletions(-)

diff --git a/src/buildstream/plugins/sources/tar.py b/src/buildstream/plugins/sources/tar.py
index 040b5c48d..0c7493b5a 100644
--- a/src/buildstream/plugins/sources/tar.py
+++ b/src/buildstream/plugins/sources/tar.py
@@ -96,6 +96,22 @@ class ReadableTarInfo(tarfile.TarInfo):
         self.__permission = permission  # pylint: disable=attribute-defined-outside-init
 
 
+def _tar_safehardlink_filter(member: tarfile.TarInfo, target_dir: str) -> Optional[tarfile.TarInfo]:
+    # Refuse to extract files whose absolute path (after following symlinks)
+    # would end up outside the target directory (+ some other sanitization).
+    member = tarfile.tar_filter(member, target_dir)
+
+    # Don't allow creating hardlinks of files outside the target directory.
+    # This would be covered by `tarfile.data_filter` but that filter is too
+    # restrictive.
+    if member and member.islnk():
+        target_path = os.path.realpath(os.path.join(target_dir, member.linkname), strict=os.path.ALLOW_MISSING)
+        if os.path.commonpath([target_path, target_dir]) != target_dir:
+            raise tarfile.LinkOutsideDestinationError(member, target_path)
+
+    return member
+
+
 class TarSource(DownloadableFileSource):
     # pylint: disable=attribute-defined-outside-init
 
@@ -108,6 +124,11 @@ class TarSource(DownloadableFileSource):
         node.validate_keys(DownloadableFileSource.COMMON_CONFIG_KEYS + ["base-dir"])
 
     def preflight(self):
+        if not hasattr(tarfile, "tar_filter"):
+            raise SourceError(
+                "Python version too old. Please upgrade to the latest security update of Python 3.10 or later."
+            )
+
         self.host_lzip = None
         if self.url.endswith(".lz"):
             self.host_lzip = utils.get_host_tool("lzip")
@@ -150,38 +171,11 @@ class TarSource(DownloadableFileSource):
                     member = filter_function(member, directory)
                     if member is not None:
                         filtered_members.append(member)
-                if sys.version_info >= (3, 12):
-                    tar.extractall(path=directory, members=filtered_members, filter="tar")
-                else:
-                    tar.extractall(path=directory, members=filtered_members)
+                tar.extractall(path=directory, members=filtered_members, filter=_tar_safehardlink_filter)
 
         except (tarfile.TarError, OSError) as e:
             raise SourceError("{}: Error staging source: {}".format(self, e)) from e
 
-    # Assert that a tarfile is safe to extract; specifically, make
-    # sure that we don't do anything outside of the target
-    # directory (this is possible, if, say, someone engineered a
-    # tarfile to contain paths that start with ..).
-    def _assert_safe(self, member: tarfile.TarInfo, target_dir: str):
-        final_path = os.path.abspath(os.path.join(target_dir, member.path))
-        if not final_path.startswith(target_dir):
-            raise SourceError(
-                "{}: Tarfile attempts to extract outside the staging area: "
-                "{} -> {}".format(self, member.path, final_path)
-            )
-
-        if member.islnk():
-            linked_path = os.path.abspath(os.path.join(target_dir, member.linkname))
-            if not linked_path.startswith(target_dir):
-                raise SourceError(
-                    "{}: Tarfile attempts to hardlink outside the staging area: "
-                    "{} -> {}".format(self, member.path, final_path)
-                )
-
-        # Don't need to worry about symlinks because they're just
-        # files here and won't be able to do much harm once we are
-        # in a sandbox.
-
     def _extract_filter(
         self, base_dir: Optional[str], member: tarfile.TarInfo, target_dir: str
     ) -> Optional[tarfile.TarInfo]:
@@ -214,8 +208,6 @@ class TarSource(DownloadableFileSource):
 
             member.path = member.path[L:]
 
-        self._assert_safe(member, target_dir)
-
         # Skip device nodes
         if member.isdev():
             return None