Harden Apptainer bundle extraction against tar path traversal

plugins/providers/apptainer/agentix/provider/apptainer.py

@@ -42,10 +42,12 @@
 import json
 import logging
 import os
+import posixpath
 import shutil
 import socket
 import tarfile
 from pathlib import Path
+from tempfile import TemporaryDirectory
 from uuid import uuid4
 
 from agentix.provider.base import Sandbox, SandboxConfig, SandboxId, SandboxInfo, SandboxProvider
@@ -151,13 +153,87 @@ def _extract_bundle(bundle_tar: Path, target: Path) -> Path:
     nix_root = target / "nix"
     if (nix_root / "runtime" / "bootstrap.sh").exists():
         return nix_root
-    target.mkdir(parents=True, exist_ok=True)
-    with tarfile.open(bundle_tar, "r:*") as tar:
-        for member in tar:
-            name = member.name
-            if not (name == "nix" or name.startswith("nix/")):
-                continue
-            tar.extract(member, target)
+
+    def checked_nix_member_name(name: str) -> str:
+        normalized = posixpath.normpath(name)
+        if normalized in {"", "."} or normalized.startswith("/") or normalized == ".." or normalized.startswith("../"):
+            raise RuntimeError(f"bundle tar produced unsafe member: {name!r}")
+        if normalized != "nix" and not normalized.startswith("nix/"):
+            raise RuntimeError(f"bundle tar produced non-/nix member: {name!r}")
+        return normalized
+
+    def ensure_safe_parent(root: Path, path: Path) -> None:
+        try:
+            relative_parent = path.parent.relative_to(root)
+        except ValueError as exc:
+            raise RuntimeError(f"bundle tar member escaped extraction root: {path}") from exc
+        current = root
+        for part in relative_parent.parts:
+            current = current / part
+            if os.path.lexists(current):
+                if current.is_symlink() or not current.is_dir():
+                    raise RuntimeError(f"bundle tar member parent is not a directory: {current}")
+            else:
+                current.mkdir()
+
+    def remove_existing_path(path: Path) -> None:
+        if path.is_dir() and not path.is_symlink():
+            shutil.rmtree(path)
+        elif os.path.lexists(path):
+            path.unlink()
+
+    def extract_nix_member(tar: tarfile.TarFile, member: tarfile.TarInfo, root: Path) -> None:
+        name = checked_nix_member_name(member.name)
+        member_path = root / name
+        ensure_safe_parent(root, member_path)
+
+        if member.isdir():
+            if os.path.lexists(member_path):
+                if member_path.is_symlink() or not member_path.is_dir():
+                    raise RuntimeError(
+                        f"bundle tar cannot replace non-directory with directory: {name}"
+                    )
+            else:
+                member_path.mkdir()
+            return
+
+        remove_existing_path(member_path)
+        if member.issym():
+            os.symlink(member.linkname, member_path)
+            return
+        if member.islnk():
+            try:
+                link_name = checked_nix_member_name(member.linkname)
+            except RuntimeError as exc:
+                raise RuntimeError(f"bundle tar produced unsafe hard-link target: {member.linkname!r}") from exc
+            link_target = root / link_name
+            if not os.path.lexists(link_target):
+                raise RuntimeError(f"bundle tar hard-link target does not exist: {member.linkname!r}")
+            os.link(link_target, member_path)
+            return
+        if member.isfile():
+            source = tar.extractfile(member)
+            if source is None:
+                raise RuntimeError(f"bundle tar has unreadable file member: {name}")
+            with source, member_path.open("wb") as f:
+                shutil.copyfileobj(source, f)
+            os.chmod(member_path, member.mode & 0o7777)
+            return
+
+        raise RuntimeError(f"bundle tar contains unsupported member type: {name}")
+
+    target.parent.mkdir(parents=True, exist_ok=True)
+    with TemporaryDirectory(prefix=f".{target.name}.", dir=target.parent) as tmp:
+        tmp_root = Path(tmp)
+        with tarfile.open(bundle_tar, "r:*") as tar:
+            for member in tar:
+                name = member.name
+                if not (name == "nix" or name.startswith("nix/")):
+                    continue
+                extract_nix_member(tar, member, tmp_root)
+        remove_existing_path(target)
+        tmp_root.replace(target)
+
     if not nix_root.is_dir():
         raise RuntimeError(f"bundle {bundle_tar} did not contain a `nix/` tree")
     return nix_root

plugins/providers/apptainer/tests/test_apptainer_provider.py

@@ -13,6 +13,7 @@
 
 from __future__ import annotations
 
+import io
 import json
 import os
 import stat
@@ -206,6 +207,20 @@ def test_extract_bundle_skips_when_runtime_already_present(tmp_path: Path) -> No
     assert sentinel.read_text() == "MUTATED"
 
 
+def test_extract_bundle_rejects_path_traversal_member(tmp_path: Path) -> None:
+    bundle = tmp_path / "bundle.tar"
+    with tarfile.open(bundle, "w") as tar:
+        data = b"owned"
+        info = tarfile.TarInfo("nix/../../escape.txt")
+        info.size = len(data)
+        tar.addfile(info, io.BytesIO(data))
+
+    target = tmp_path / "cache" / "bundle"
+    with pytest.raises(RuntimeError, match="unsafe member"):
+        _extract_bundle(bundle, target)
+    assert not (tmp_path / "cache" / "escape.txt").exists()
+
+
 @pytest.mark.anyio
 async def test_create_and_delete_roundtrip(env, tmp_path: Path) -> None:
     bundle = _write_bundle_tar(tmp_path)