diff --git a/packages/prime/src/prime_cli/commands/env.py b/packages/prime/src/prime_cli/commands/env.py index d8a9efe1..c5d52422 100644 --- a/packages/prime/src/prime_cli/commands/env.py +++ b/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) diff --git a/packages/prime/tests/test_env_pull_path.py b/packages/prime/tests/test_env_pull_path.py index 6b45b820..f32d550d 100644 --- a/packages/prime/tests/test_env_pull_path.py +++ b/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" diff --git a/packages/prime/tests/test_env_push_archive.py b/packages/prime/tests/test_env_push_archive.py index d3d45a16..8cba89ed 100644 --- a/packages/prime/tests/test_env_push_archive.py +++ b/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())