🐛 FIX: Support rooted tar members

archive_path/tar_path.py

@@ -6,6 +6,7 @@
 # For further information on the license, see the LICENSE file            #
 ###########################################################################
 """A implementation of the ``pathlib.Path`` interface for ``tarfile.TarFile``."""
+import copy
 from contextlib import contextmanager, suppress
 import io
 import itertools
@@ -32,6 +33,11 @@
 __all__ = ("TarPath", "open_file_in_tar", "read_file_in_tar")
 
 
+def _normalize_at(path: str) -> str:
+    """Normalize an archive member name to an internal relative path."""
+    return path.strip(posixpath.sep)
+
+
 class TarPath:
     """A wrapper around ``tarfile.TarPath``,
     to provide an interface equivalent to ``pathlib.Path``
@@ -82,7 +88,9 @@ def __init__(
         :param dereference: If true, add content of linked file to the tar file, else the link.
 
         """
-        if posixpath.isabs(at):
+        if at == posixpath.sep:
+            at = ""
+        elif posixpath.isabs(at):
             raise ValueError(f"'at' cannot be an absolute path: {at}")
         assert not any(
             p == ".." for p in at.split(posixpath.sep)
@@ -148,18 +156,43 @@ def _all_at_set(self) -> Set[str]:
         if read_mode:
             with suppress(AttributeError):
                 return self._tarfile.__all_at  # type: ignore
-        names = self._tarfile.getnames()
+        names = [_normalize_at(name) for name in self._tarfile.getnames()]
         parents = itertools.chain.from_iterable(map(_parents, names))
         all_set = {p.rstrip("/") for p in itertools.chain([""], names, parents)}
         if read_mode:
             self._tarfile.__all_at = all_set  # type: ignore
         return all_set
 
+    def _member_names(self, path: Optional[str] = None) -> Tuple[str, ...]:
+        """Return possible tar member names for an internal relative path."""
+        path = self.at if path is None else path
+        if path == "":
+            return ("", posixpath.sep)
+        return (
+            path,
+            f"{path}{posixpath.sep}",
+            f"{posixpath.sep}{path}",
+            f"{posixpath.sep}{path}{posixpath.sep}",
+        )
+
+    def _getmember(self, path: Optional[str] = None) -> tarfile.TarInfo:
+        """Return a member by internal path, accepting rooted tar names."""
+        for member_name in self._member_names(path):
+            with suppress(KeyError):
+                return self._tarfile.getmember(member_name)
+        raise KeyError(self.at if path is None else path)
+
+    def _getmember_relative(self, path: Optional[str] = None) -> tarfile.TarInfo:
+        """Return a member whose name is safe to extract below an output path."""
+        member = copy.copy(self._getmember(path))
+        member.name = _normalize_at(member.name)
+        return member
+
     def _has_member(self) -> bool:
-        info = None
         with suppress(KeyError):
-            info = self._tarfile.getmember(self.at)
-        return info is not None
+            self._getmember()
+            return True
+        return False
 
     def close(self):
         """Close the tarfile."""
@@ -203,7 +236,7 @@ def is_dir(self):
     def is_file(self):
         """Whether this path is an existing regular file."""
         try:
-            info = self._tarfile.getmember(self.at)
+            info = self._getmember()
         except KeyError:
             return False
         return not info.isdir()
@@ -213,10 +246,7 @@ def exists(self) -> bool:
         if self.at == "":
             return True
         with suppress(KeyError):
-            self._tarfile.getmember(self.at)
-            return True
-        with suppress(KeyError):
-            self._tarfile.getmember(self.at + "/")
+            self._getmember()
             return True
         # note, we could just check this, but it can takes time/memory to construct
         return self.at in self._all_at_set()
@@ -238,7 +268,7 @@ def open(self, mode: str = "rb"):  # noqa: A003
         if mode == "rb":
             handle = None
             with suppress(KeyError):
-                handle = self._tarfile.extractfile(self.at)
+                handle = self._tarfile.extractfile(self._getmember())
             if handle is None:
                 raise FileNotFoundError(f"No such file: '{self.at}'")
             yield handle
@@ -271,7 +301,7 @@ def read_bytes(self) -> bytes:
         """Read bytes from the file."""
         handle = None
         with suppress(KeyError):
-            handle = self._tarfile.extractfile(self.at)
+            handle = self._tarfile.extractfile(self._getmember())
         if handle is None:
             raise FileNotFoundError(f"No such file: '{self.at}'")
         return handle.read()
@@ -307,7 +337,7 @@ def glob(self, pattern: str, include_virtual: bool = True):
         iterator = (
             self._all_at_set()
             if include_virtual
-            else {name.rstrip("/") for name in self._tarfile.getnames()}
+            else {_normalize_at(name).rstrip("/") for name in self._tarfile.getnames()}
         )
         for name in match_glob(self.at, pattern, iterator):
             yield self.__class__(self, at=name)
@@ -440,7 +470,7 @@ def extract_tree(
 
         for path in self.glob(pattern, include_virtual=False):
             callback("update", 1)
-            info = self._tarfile.getmember(path.at)
+            info = self._getmember_relative(path.at)
             if (not allow_dev) and info.isdev():
                 continue
             if (not allow_symlink) and (info.islnk() or info.issym()):
@@ -467,12 +497,13 @@ def open_file_in_tar(
     :raises FileNotFoundError: If the path in the zip file does not exist
     """
     assert mode.startswith("r")
+    normalized_path = _normalize_at(path)
     try:
         with tarfile.open(filepath, mode, format=tarfile.PAX_FORMAT) as tar_handle:
             tarinfo = None
             while True:
                 tarinfo = tar_handle.next()  # noqa: B305
-                if tarinfo is None or tarinfo.name == path:
+                if tarinfo is None or _normalize_at(tarinfo.name) == normalized_path:
                     break
                 # flush stored members
                 tar_handle.members = []  # type: ignore

tests/test_basic.py

@@ -6,6 +6,8 @@
 # For further information on the license, see the LICENSE file            #
 ###########################################################################
 """Test compression utilities"""
+import io
+import tarfile
 from typing import Type, Union
 import zipfile
 
@@ -192,6 +194,40 @@ def test_path(
     }
 
 
+def test_tar_rooted_members_are_exposed_as_relative_paths(tmp_path):
+    """Test that rooted tar member names are exposed as relative archive paths."""
+    content = b"garbage"
+    archive = tmp_path / "rooted.tar"
+
+    with tarfile.open(archive, "w") as handle:
+        info = tarfile.TarInfo("/bin/garbage.txt")
+        info.size = len(content)
+        handle.addfile(info, io.BytesIO(content))
+
+    with TarPath(archive, mode="r", at="/") as path:
+        assert path.at == ""
+        assert path._all_at_set() == {  # pylint: disable=protected-access
+            "",
+            "bin",
+            "bin/garbage.txt",
+        }
+        assert {child.at for child in path.iterdir()} == {"bin"}
+
+        bin_path = path / "bin"
+        garbage_path = bin_path / "garbage.txt"
+        assert bin_path.is_dir()
+        assert garbage_path.exists()
+        assert garbage_path.is_file()
+        assert garbage_path.read_bytes() == content
+        assert read_file_in_tar(archive, "bin/garbage.txt") == "garbage"
+
+        path.extract_tree(tmp_path / "rooted_extract")
+
+    assert (
+        tmp_path / "rooted_extract" / "bin" / "garbage.txt"
+    ).read_bytes() == content
+
+
 def test_zip_write(tmp_path):
     """Test setting compression and comment options for write."""
     zipinfos: dict = {}