diff --git a/fluster/utils.py b/fluster/utils.py index 2c280a5a..be72ef89 100644 --- a/fluster/utils.py +++ b/fluster/utils.py @@ -29,6 +29,7 @@ import shutil import subprocess import sys +import tarfile import time import urllib.error import urllib.parse @@ -383,7 +384,18 @@ def _extract_zip_members(filepath: str, entries: List[Tuple[Optional[str], str]] def _extract_tar_members(filepath: str, entries: List[Tuple[Optional[str], str]]) -> List[str]: - """Run ``tar`` once per output directory; a falsy member extracts everything there.""" + """Extract tar members using Python's ``tarfile`` instead of the external ``tar`` binary. + + The external GNU tar is not usable inside strictly confined snaps: snapd's + seccomp profile denies the ``openat2`` syscall that tar relies on, so every + extraction fails with "Cannot open: Operation not permitted". The ``data`` + filter (Python 3.12+) is used when available to reject unsafe members; + on older Pythons each member path is validated with ``_safe_join`` so + members escaping ``output_dir`` are rejected regardless of the filter. + + A falsy member extracts everything into its ``output_dir``. Raises + ``ArchiveError`` on corrupt archives, missing members or unsafe members. + """ extract_all_dirs = {output_dir for member, output_dir in entries if not member} # An empty member list makes tar extract everything. members_by_dir: Dict[str, List[str]] = {} @@ -391,8 +403,27 @@ def _extract_tar_members(filepath: str, entries: List[Tuple[Optional[str], str]] members = members_by_dir.setdefault(output_dir, []) if member and output_dir not in extract_all_dirs: members.append(member) - for output_dir, members in members_by_dir.items(): - subprocess.run(["tar", "-C", output_dir, "-xf", filepath, *members], check=True) + filter_kwargs: Dict[str, str] = {"filter": "data"} if hasattr(tarfile, "data_filter") else {} + try: + with tarfile.open(filepath) as tar: + names = set(tar.getnames()) + for output_dir, members in members_by_dir.items(): + if members: + for member in members: + if member not in names: + raise ArchiveError(f"{filepath} does not contain member {member}") + # Reject members escaping output_dir even when the data + # filter is unavailable (Python < 3.12). + _safe_join(output_dir, member) + tar.extract(member, output_dir, **filter_kwargs) # type: ignore[arg-type] + else: + for info in tar.getmembers(): + _safe_join(output_dir, info.name) + tar.extractall(output_dir, **filter_kwargs) # type: ignore[arg-type] + except ArchiveError: + raise + except Exception as exc: + raise ArchiveError(f"{filepath} could not be extracted") from exc return [] diff --git a/tests/test_utils.py b/tests/test_utils.py index d60e00a9..2d9bdece 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -99,7 +99,6 @@ def test_zip_slip_rejected(self) -> None: utils.extract_archive(zp, [(None, out)]) self.assertFalse(os.path.exists(os.path.join(self.tmp, "escaped.txt"))) - @unittest.skipUnless(shutil.which("tar"), "tar not available") def test_tar_named_members_grouped_and_extract_all(self) -> None: tp = os.path.join(self.tmp, "pkg.tar.gz") with tarfile.open(tp, "w:gz") as tar_file: @@ -116,6 +115,53 @@ def test_tar_named_members_grouped_and_extract_all(self) -> None: self.assertEqual(sorted(os.listdir(out1)), ["a.bits", "b.bits"]) self.assertEqual(sorted(os.listdir(out2)), ["a.bits", "b.bits"]) + def test_tar_nested_member_creates_directories(self) -> None: + tp = os.path.join(self.tmp, "pkg.tar.gz") + with tarfile.open(tp, "w:gz") as tar_file: + info = tarfile.TarInfo("A6-GE19/ti_vga.cmp") + info.size = 4 + tar_file.addfile(info, io.BytesIO(b"bits")) + out = os.path.join(self.tmp, "out") + os.makedirs(out) + missing = utils.extract_archive(tp, [("A6-GE19/ti_vga.cmp", out)]) + self.assertEqual(missing, []) + target = os.path.join(out, "A6-GE19", "ti_vga.cmp") + self.assertTrue(os.path.exists(target)) + with open(target, "rb") as handle: + self.assertEqual(handle.read(), b"bits") + + def test_tar_missing_member_raises_archive_error(self) -> None: + tp = os.path.join(self.tmp, "pkg.tar.gz") + with tarfile.open(tp, "w:gz") as tar_file: + info = tarfile.TarInfo("a.bits") + info.size = 1 + tar_file.addfile(info, io.BytesIO(b"x")) + out = os.path.join(self.tmp, "out") + os.makedirs(out) + with self.assertRaises(utils.ArchiveError): + utils.extract_archive(tp, [("nope.bits", out)]) + + def test_tar_slip_rejected(self) -> None: + tp = os.path.join(self.tmp, "slip.tar.gz") + with tarfile.open(tp, "w:gz") as tar_file: + info = tarfile.TarInfo("../escaped.txt") + info.size = 1 + tar_file.addfile(info, io.BytesIO(b"x")) + out = os.path.join(self.tmp, "out") + os.makedirs(out) + with self.assertRaises(utils.ArchiveError): + utils.extract_archive(tp, [(None, out)]) + self.assertFalse(os.path.exists(os.path.join(self.tmp, "escaped.txt"))) + + def test_tar_corrupt_archive_raises_archive_error(self) -> None: + tp = os.path.join(self.tmp, "corrupt.tar.gz") + with open(tp, "wb") as handle: + handle.write(b"not a tar archive") + out = os.path.join(self.tmp, "out") + os.makedirs(out) + with self.assertRaises(utils.ArchiveError): + utils.extract_archive(tp, [(None, out)]) + @unittest.skipUnless(shutil.which("gunzip"), "gunzip not available") def test_gzip_decompressed_into_each_dir(self) -> None: gp = os.path.join(self.tmp, "clip.bs.gz")