mirror of
https://github.com/usestrix/strix.git
synced 2026-08-16 09:26:39 +02:00
fix(runtime): stage symlink-safe copies for LocalDir uploads (#766)
The sandbox SDK's LocalDir walker rejects any symlink outright (LocalDirReadError, reason=symlink_not_supported), so uploading a cloned repository that commits symlinks (common in JS/TS monorepos) aborts before the agent starts. Stage such trees into a temp copy first: in-tree links are dereferenced; out-of-tree, dangling, and cyclic links are dropped and never followed, preserving the walker's path-escape safety. Symlink-free trees are uploaded as-is.
This commit is contained in:
@@ -0,0 +1,108 @@
|
||||
"""Tests for symlink-safe LocalDir staging."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from strix.runtime.local_dir_staging import stage_symlink_safe_dir, tree_has_symlink
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _make_repo(tmp_path: Path) -> Path:
|
||||
repo = tmp_path / "repo"
|
||||
(repo / "pkg").mkdir(parents=True)
|
||||
(repo / "pkg" / "mod.py").write_text("x = 1\n")
|
||||
(repo / "README.md").write_text("readme\n")
|
||||
return repo
|
||||
|
||||
|
||||
def test_tree_without_symlinks_used_as_is(tmp_path: Path) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
|
||||
upload_path, staged = stage_symlink_safe_dir(repo)
|
||||
|
||||
assert staged is None
|
||||
assert upload_path == repo.resolve()
|
||||
assert not tree_has_symlink(repo)
|
||||
|
||||
|
||||
def test_in_tree_file_symlink_is_dereferenced(tmp_path: Path) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
(repo / "link.py").symlink_to(repo / "pkg" / "mod.py")
|
||||
|
||||
upload_path, staged = stage_symlink_safe_dir(repo)
|
||||
|
||||
assert staged is not None
|
||||
assert upload_path == staged
|
||||
assert not (staged / "link.py").is_symlink()
|
||||
assert (staged / "link.py").read_text() == "x = 1\n"
|
||||
assert (staged / "pkg" / "mod.py").read_text() == "x = 1\n"
|
||||
assert not tree_has_symlink(staged)
|
||||
|
||||
|
||||
def test_in_tree_relative_dir_symlink_is_dereferenced(tmp_path: Path) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
(repo / "pkg_alias").symlink_to("pkg")
|
||||
|
||||
_upload, staged = stage_symlink_safe_dir(repo)
|
||||
|
||||
assert staged is not None
|
||||
assert (staged / "pkg_alias" / "mod.py").read_text() == "x = 1\n"
|
||||
assert not tree_has_symlink(staged)
|
||||
|
||||
|
||||
def test_out_of_tree_symlink_is_dropped(tmp_path: Path) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
outside = tmp_path / "outside.txt"
|
||||
outside.write_text("secret\n")
|
||||
(repo / "escape.txt").symlink_to(outside)
|
||||
(repo / "abs_escape").symlink_to("/etc")
|
||||
|
||||
_upload, staged = stage_symlink_safe_dir(repo)
|
||||
|
||||
assert staged is not None
|
||||
assert not (staged / "escape.txt").exists()
|
||||
assert not (staged / "abs_escape").exists()
|
||||
assert (staged / "README.md").exists()
|
||||
|
||||
|
||||
def test_dangling_symlink_is_dropped(tmp_path: Path) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
(repo / "dangling").symlink_to(repo / "does-not-exist")
|
||||
|
||||
_upload, staged = stage_symlink_safe_dir(repo)
|
||||
|
||||
assert staged is not None
|
||||
assert not (staged / "dangling").exists()
|
||||
assert not (staged / "dangling").is_symlink()
|
||||
|
||||
|
||||
def test_cyclic_symlink_terminates(tmp_path: Path) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
(repo / "self").symlink_to(repo)
|
||||
(repo / "pkg" / "up").symlink_to("..")
|
||||
|
||||
_upload, staged = stage_symlink_safe_dir(repo)
|
||||
|
||||
assert staged is not None
|
||||
assert (staged / "README.md").exists()
|
||||
assert not tree_has_symlink(staged)
|
||||
|
||||
|
||||
def test_nested_symlinks_inside_linked_dir(tmp_path: Path) -> None:
|
||||
repo = _make_repo(tmp_path)
|
||||
shared = repo / "shared"
|
||||
shared.mkdir()
|
||||
(shared / "conf.json").write_text("{}\n")
|
||||
(shared / "escape").symlink_to("/etc/passwd")
|
||||
(repo / "pkg" / "shared_link").symlink_to(shared)
|
||||
|
||||
_upload, staged = stage_symlink_safe_dir(repo)
|
||||
|
||||
assert staged is not None
|
||||
assert (staged / "pkg" / "shared_link" / "conf.json").read_text() == "{}\n"
|
||||
assert not (staged / "pkg" / "shared_link" / "escape").exists()
|
||||
assert not (staged / "shared" / "escape").exists()
|
||||
@@ -18,15 +18,18 @@ def _source(subdir: str, path: str, *, mount: bool = False) -> dict[str, Any]:
|
||||
|
||||
|
||||
def test_copied_source_becomes_localdir_entry(tmp_path: Path) -> None:
|
||||
entries, bind_mounts = build_session_entries([_source("repo", str(tmp_path))])
|
||||
entries, bind_mounts, staged_dirs = build_session_entries([_source("repo", str(tmp_path))])
|
||||
|
||||
assert bind_mounts == []
|
||||
assert staged_dirs == []
|
||||
assert isinstance(entries["repo"], LocalDir)
|
||||
assert entries["repo"].src == tmp_path.resolve()
|
||||
|
||||
|
||||
def test_mounted_source_becomes_bind_mount(tmp_path: Path) -> None:
|
||||
entries, bind_mounts = build_session_entries([_source("repo", str(tmp_path), mount=True)])
|
||||
entries, bind_mounts, _staged = build_session_entries(
|
||||
[_source("repo", str(tmp_path), mount=True)]
|
||||
)
|
||||
|
||||
assert entries == {}
|
||||
assert bind_mounts == [
|
||||
@@ -44,7 +47,7 @@ def test_mixed_sources_split_correctly(tmp_path: Path) -> None:
|
||||
copied.mkdir()
|
||||
mounted.mkdir()
|
||||
|
||||
entries, bind_mounts = build_session_entries(
|
||||
entries, bind_mounts, _staged = build_session_entries(
|
||||
[
|
||||
_source("copied", str(copied)),
|
||||
_source("mounted", str(mounted), mount=True),
|
||||
@@ -57,7 +60,7 @@ def test_mixed_sources_split_correctly(tmp_path: Path) -> None:
|
||||
|
||||
|
||||
def test_incomplete_sources_are_skipped() -> None:
|
||||
entries, bind_mounts = build_session_entries(
|
||||
entries, bind_mounts, staged_dirs = build_session_entries(
|
||||
[
|
||||
{"source_path": "", "workspace_subdir": "x"},
|
||||
{"source_path": "/p", "workspace_subdir": ""},
|
||||
@@ -65,3 +68,20 @@ def test_incomplete_sources_are_skipped() -> None:
|
||||
)
|
||||
assert entries == {}
|
||||
assert bind_mounts == []
|
||||
assert staged_dirs == []
|
||||
|
||||
|
||||
def test_symlink_tree_is_staged(tmp_path: Path) -> None:
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir()
|
||||
(repo / "real.txt").write_text("content")
|
||||
(repo / "link.txt").symlink_to(repo / "real.txt")
|
||||
|
||||
entries, _mounts, staged_dirs = build_session_entries([_source("repo", str(repo))])
|
||||
|
||||
assert len(staged_dirs) == 1
|
||||
entry = entries["repo"]
|
||||
assert isinstance(entry, LocalDir)
|
||||
assert entry.src == staged_dirs[0]
|
||||
assert not (staged_dirs[0] / "link.txt").is_symlink()
|
||||
assert (staged_dirs[0] / "link.txt").read_text() == "content"
|
||||
|
||||
Reference in New Issue
Block a user