From 03b6b4ae92a01c739a464d30e1df5fc2cf3c3ce8 Mon Sep 17 00:00:00 2001 From: Alexandre Esse Date: Sat, 10 Oct 2026 01:14:09 +0200 Subject: [PATCH 1/2] fix: extract tar archives with Python tarfile instead of external tar The external GNU tar relies on the openat2 syscall, which snapd's seccomp profile denies inside strictly confined snaps, making every tar.gz/tgz extraction fail with "Cannot open: Operation not permitted". Extracting with the standard library tarfile avoids the syscall and works in any environment. The tarfile "data" filter is used on Python 3.12+ to reject unsafe members; older Pythons keep the previous behaviour. Signed-off-by: Alexandre Esse --- fluster/utils.py | 32 +++++++++++++++++++++++++++++--- tests/test_utils.py | 36 +++++++++++++++++++++++++++++++++++- 2 files changed, 64 insertions(+), 4 deletions(-) diff --git a/fluster/utils.py b/fluster/utils.py index 2c280a5a..14b518d6 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,17 @@ 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; + older Pythons fall back to the default extraction. + + 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 +402,23 @@ 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(): + os.makedirs(output_dir, exist_ok=True) + if members: + for member in members: + if member not in names: + raise ArchiveError(f"{filepath} does not contain member {member}") + tar.extract(member, output_dir, **filter_kwargs) # type: ignore[arg-type] + else: + 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..3cadc88f 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,41 @@ 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_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") From c375f18b610d94b57c1a5c16933063b9243fdcf8 Mon Sep 17 00:00:00 2001 From: Alexandre Esse Date: Mon, 26 Oct 2026 22:52:04 +0100 Subject: [PATCH 2/2] fix: reject unsafe tar members via _safe_join on Python < 3.12 The tarfile data filter only guards Python 3.12+, so on older Pythons (e.g. 3.7 used by min_linux CI) members escaping output_dir were extracted silently. Validate each member path with _safe_join before extraction and drop the redundant os.makedirs. Add a tar-slip test. Signed-off-by: Alexandre Esse --- fluster/utils.py | 9 +++++++-- tests/test_utils.py | 12 ++++++++++++ 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/fluster/utils.py b/fluster/utils.py index 14b518d6..be72ef89 100644 --- a/fluster/utils.py +++ b/fluster/utils.py @@ -390,7 +390,8 @@ def _extract_tar_members(filepath: str, entries: List[Tuple[Optional[str], str]] 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; - older Pythons fall back to the default extraction. + 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. @@ -407,13 +408,17 @@ def _extract_tar_members(filepath: str, entries: List[Tuple[Optional[str], str]] with tarfile.open(filepath) as tar: names = set(tar.getnames()) for output_dir, members in members_by_dir.items(): - os.makedirs(output_dir, exist_ok=True) 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 diff --git a/tests/test_utils.py b/tests/test_utils.py index 3cadc88f..2d9bdece 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -141,6 +141,18 @@ def test_tar_missing_member_raises_archive_error(self) -> None: 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: