fix(env): use safe tar extract on env pull

packages/prime/src/prime_cli/commands/env.py

@@ -581,6 +581,9 @@ def should_include_file_in_archive(file_path: Path, base_path: Path) -> bool:
     if file_path.is_symlink():
         return False
 
+    # Hardlinks are fine to include: _add_file_to_archive stores them as
+    # regular file content so pull/install safe-extract never sees LNKTYPE.
+
     rel_path = file_path.relative_to(base_path)
 
     # Skip hidden files
@@ -665,6 +668,25 @@ def is_nested_dir_ignored(dir_path: Path) -> bool:
     return [files_by_rel_path[rel_path] for rel_path in sorted(files_by_rel_path)]
 
 
+def _add_file_to_archive(tar: tarfile.TarFile, file_path: Path, arcname: str) -> None:
+    """Add a file as REGTYPE content (never symlink/hardlink members).
+
+    ``tar.add()`` records a hardlink (``LNKTYPE``) when another file with the
+    same inode is already in the archive. ``_safe_tar_extract`` rejects
+    hardlinks, so push must always store byte content as a regular file —
+    same rationale as skipping symlinks at collect time.
+    """
+    if file_path.is_symlink():
+        return
+
+    info = tar.gettarinfo(str(file_path), arcname=arcname)
+    info.type = tarfile.REGTYPE
+    info.linkname = ""
+    info.size = file_path.stat().st_size
+    with open(file_path, "rb") as handle:
+        tar.addfile(info, handle)
+
+
 def compute_content_hash(env_path: Path) -> str:
     """Compute deterministic, cross-platform content hash for environment files.
 
@@ -1419,7 +1441,7 @@ def push(
                     with tarfile.open(tmp.name, "w:gz") as tar:
                         for file_path in _collect_archive_files(env_path):
                             arcname = file_path.relative_to(env_path)
-                            tar.add(file_path, arcname=str(arcname))
+                            _add_file_to_archive(tar, file_path, str(arcname))
 
                     # Check tarball size
                     tarball_size = Path(tmp.name).stat().st_size
@@ -1833,7 +1855,11 @@ def pull(
 
                 try:
                     with tarfile.open(tmp.name, "r:gz") as tar:
-                        tar.extractall(target_dir)
+                        # Use path-traversal / symlink-safe extract (same as install)
+                        _safe_tar_extract(tar, Path(target_dir))
+                except ValueError as e:
+                    console.print(f"[red]Failed to extract archive: {e}[/red]")
+                    raise typer.Exit(1)
                 except tarfile.TarError as e:
                     console.print(f"[red]Failed to extract archive: {e}[/red]")
                     raise typer.Exit(1)

packages/prime/tests/test_env_pull_path.py

@@ -1,5 +1,10 @@
+import io
+import tarfile
+from pathlib import Path
 from typing import Any
 
+import pytest
+import typer
 from prime_cli.commands import env as env_commands
 from prime_cli.commands.env import _environment_package_download_url, _resolve_pull_environment_path
 
@@ -83,8 +88,13 @@ def __enter__(self) -> "FakeTar":
         def __exit__(self, *_args: Any) -> None:
             return None
 
+        def getmembers(self) -> list[Any]:
+            # Empty members keep _safe_tar_extract validation satisfied; extractall
+            # still writes the demo file as before.
+            return []
+
         def extractall(self, target: Any) -> None:
-            (target / "README.md").write_text("# Demo\n", encoding="utf-8")
+            (Path(target) / "README.md").write_text("# Demo\n", encoding="utf-8")
 
     def fake_stream(
         method: str,
@@ -108,3 +118,103 @@ def fake_stream(
     env_commands.pull("alice/demo", target=str(target), version="latest")
 
     assert (target / "README.md").read_text(encoding="utf-8") == "# Demo\n"
+
+
+def _gzip_tar_bytes(member_name: str, *, symlink: bool = False) -> bytes:
+    buffer = io.BytesIO()
+    with tarfile.open(fileobj=buffer, mode="w:gz") as tar:
+        info = tarfile.TarInfo(name=member_name)
+        if symlink:
+            info.type = tarfile.SYMTYPE
+            info.linkname = "/tmp"
+            tar.addfile(info)
+        else:
+            info.size = 0
+            tar.addfile(info)
+    return buffer.getvalue()
+
+
+def _stub_pull_download(
+    monkeypatch: pytest.MonkeyPatch,
+    *,
+    archive_bytes: bytes,
+    details: dict[str, Any] | None = None,
+) -> None:
+    payload = details or {
+        "id": "env-1",
+        "tracked_package_url": "https://example.test/pkg.tar.gz",
+        "semantic_version": "0.1.0",
+        "metadata": {},
+    }
+
+    class FakeAPIClient:
+        api_key = "test-token"
+
+        def __init__(self, require_auth: bool = False) -> None:
+            assert require_auth is False
+
+        def get(self, path: str) -> dict[str, Any]:
+            return {"data": payload}
+
+    class FakeStream:
+        def __enter__(self) -> "FakeStream":
+            return self
+
+        def __exit__(self, *_args: Any) -> None:
+            return None
+
+        def raise_for_status(self) -> None:
+            return None
+
+        def iter_bytes(self, chunk_size: int) -> list[bytes]:
+            return [archive_bytes]
+
+    def fake_stream(*_args: Any, **_kwargs: Any) -> FakeStream:
+        return FakeStream()
+
+    monkeypatch.setattr(env_commands, "APIClient", FakeAPIClient)
+    monkeypatch.setattr(env_commands.httpx, "stream", fake_stream)
+
+
+def test_pull_rejects_path_traversal_archive(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    _stub_pull_download(
+        monkeypatch, archive_bytes=_gzip_tar_bytes("../../../etc/malicious")
+    )
+    target = tmp_path / "safe-target"
+    outside = tmp_path / "etc"
+    outside.mkdir()
+
+    with pytest.raises(typer.Exit) as exc:
+        env_commands.pull("alice/demo", target=str(target), version="latest")
+
+    assert exc.value.exit_code == 1
+    assert not any(outside.iterdir())
+    assert not (tmp_path / "malicious").exists()
+
+
+def test_pull_rejects_symlink_archive(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    _stub_pull_download(
+        monkeypatch, archive_bytes=_gzip_tar_bytes("evil_link", symlink=True)
+    )
+    target = tmp_path / "safe-target"
+
+    with pytest.raises(typer.Exit) as exc:
+        env_commands.pull("alice/demo", target=str(target), version="latest")
+
+    assert exc.value.exit_code == 1
+    assert not (target / "evil_link").exists()
+
+
+def test_pull_extracts_safe_archive(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    buffer = io.BytesIO()
+    with tarfile.open(fileobj=buffer, mode="w:gz") as tar:
+        data = b"ok\n"
+        info = tarfile.TarInfo(name="README.md")
+        info.size = len(data)
+        tar.addfile(info, io.BytesIO(data))
+    _stub_pull_download(monkeypatch, archive_bytes=buffer.getvalue())
+    target = tmp_path / "safe-target"
+
+    env_commands.pull("alice/demo", target=str(target), version="latest")
+
+    assert (target / "README.md").read_text(encoding="utf-8") == "ok\n"

packages/prime/tests/test_env_push_archive.py

@@ -1,4 +1,14 @@
-from prime_cli.commands.env import _collect_archive_files, compute_content_hash
+import os
+import tarfile
+from pathlib import Path
+
+import pytest
+from prime_cli.commands.env import (
+    _add_file_to_archive,
+    _collect_archive_files,
+    _safe_tar_extract,
+    compute_content_hash,
+)
 
 
 def test_collect_archive_files_respects_gitignore(tmp_path):
@@ -72,3 +82,64 @@ def test_compute_content_hash_ignores_gitignored_files(tmp_path):
 
     assert second_hash == first_hash
     assert third_hash != second_hash
+
+
+@pytest.mark.skipif(not hasattr(os, "link"), reason="os.link is unavailable on this platform")
+def test_add_file_to_archive_stores_hardlinks_as_regular_files(tmp_path: Path) -> None:
+    (tmp_path / "pyproject.toml").write_text("[project]\nname='demo'\nversion='0.1.0'\n")
+    # Nested dir so files are picked up by _collect_archive_files (root *.txt is ignored)
+    payload_dir = tmp_path / "payload"
+    payload_dir.mkdir()
+    original = payload_dir / "data.txt"
+    linked = payload_dir / "data-link.txt"
+    original.write_text("shared-content\n")
+    try:
+        os.link(original, linked)
+    except OSError as exc:
+        pytest.skip(f"hardlinks unavailable: {exc}")
+
+    archive_path = tmp_path / "demo.tar.gz"
+    with tarfile.open(archive_path, "w:gz") as tar:
+        for file_path in _collect_archive_files(tmp_path):
+            _add_file_to_archive(tar, file_path, file_path.relative_to(tmp_path).as_posix())
+
+    dest = tmp_path / "extracted"
+    dest.mkdir()
+    with tarfile.open(archive_path, "r:gz") as tar:
+        members = tar.getmembers()
+        assert any(member.name == "payload/data.txt" for member in members)
+        assert any(member.name == "payload/data-link.txt" for member in members)
+        assert all(not member.islnk() and not member.issym() for member in members)
+        _safe_tar_extract(tar, dest)
+
+    assert (dest / "payload" / "data.txt").read_text(encoding="utf-8") == "shared-content\n"
+    assert (dest / "payload" / "data-link.txt").read_text(encoding="utf-8") == "shared-content\n"
+
+
+def test_tar_add_would_record_hardlink_but_helper_does_not(tmp_path: Path) -> None:
+    """Document why we cannot use tar.add for push archives."""
+    if not hasattr(os, "link"):
+        pytest.skip("os.link is unavailable on this platform")
+
+    original = tmp_path / "a.txt"
+    linked = tmp_path / "b.txt"
+    original.write_text("x\n")
+    try:
+        os.link(original, linked)
+    except OSError as exc:
+        pytest.skip(f"hardlinks unavailable: {exc}")
+
+    naive = tmp_path / "naive.tar.gz"
+    with tarfile.open(naive, "w:gz") as tar:
+        tar.add(original, arcname="a.txt")
+        tar.add(linked, arcname="b.txt")
+        naive_has_hardlink = any(member.islnk() for member in tar.getmembers())
+
+    if not naive_has_hardlink:
+        pytest.skip("this platform/filesystem does not emit hardlinks via tar.add")
+
+    safe = tmp_path / "safe.tar.gz"
+    with tarfile.open(safe, "w:gz") as tar:
+        _add_file_to_archive(tar, original, "a.txt")
+        _add_file_to_archive(tar, linked, "b.txt")
+        assert all(not member.islnk() for member in tar.getmembers())