diff --git a/pm/store.py b/pm/store.py index 5305490c07..7dba809be4 100644 --- a/pm/store.py +++ b/pm/store.py @@ -222,8 +222,8 @@ def tree_digest(root: Path) -> str: """Deterministic sha256 over a directory tree: walk every file, sort by posix relpath, hash `relpath\\0` per entry. No mtimes, no mode bits. Symlinks contribute their LINK TARGET TEXT (os.readlink), - not the target's bytes — the link is the data. Directory symlinks are - not followed. + not the target's bytes — the link is the data. Directory symlinks and + junctions are not followed. ``__pycache__`` directories are skipped: CPython writes .pyc caches into them the first time the staged interpreter runs (uv venv/uv sync @@ -233,7 +233,14 @@ def tree_digest(root: Path) -> str: files: list[tuple[str, Path]] = [] for dirpath, dirnames, filenames in os.walk(root): - dirnames[:] = [d for d in sorted(dirnames) if d != "__pycache__"] + descend = [] + for name in sorted(dirnames): + path = Path(dirpath) / name + if path.is_symlink() or path.is_junction(): + files.append((path.relative_to(root).as_posix(), path)) + elif name != "__pycache__": + descend.append(name) + dirnames[:] = descend for fname in filenames: path = Path(dirpath) / fname files.append((path.relative_to(root).as_posix(), path)) @@ -243,7 +250,7 @@ def tree_digest(root: Path) -> str: for rel, path in files: digest.update(rel.encode("utf-8")) digest.update(b"\0") - if path.is_symlink(): + if path.is_symlink() or path.is_junction(): digest.update(os.readlink(path).encode("utf-8")) else: with open(path, "rb") as f: diff --git a/tests/pm/test_tree_digest_links.py b/tests/pm/test_tree_digest_links.py new file mode 100644 index 0000000000..44d1cbb0f2 --- /dev/null +++ b/tests/pm/test_tree_digest_links.py @@ -0,0 +1,73 @@ +"""Realized digests bind link text without reading the linked directory.""" +import os +from pathlib import Path + +import pytest + +from pm.store import tree_digest + + +@pytest.mark.platforms("windows", "posix") +def test_directory_links_bind_only_their_target_text(tmp_path): + root = tmp_path / "tree" + root.mkdir() + (root / "payload").write_bytes(b"unchanged") + left, right = tmp_path / "left", tmp_path / "right" + for target in (left, right): + target.mkdir() + (target / "content").write_bytes(b"same bytes") + plain = tree_digest(root) + link = root / "directory" + try: + link.symlink_to(Path("..") / "left", target_is_directory=True) + except OSError as exc: + if getattr(exc, "winerror", None) == 1314: + pytest.skip("directory symlinks require Windows Developer Mode or privilege") + raise + first = tree_digest(root) + assert first != plain + (left / "content").write_bytes(b"changed outside the tree") + assert tree_digest(root) == first + link.unlink() + link.symlink_to(Path("..") / "right", target_is_directory=True) + assert tree_digest(root) != first + link.unlink() + link.symlink_to(".", target_is_directory=True) + cycle = tree_digest(root) + assert tree_digest(root) == cycle + assert cycle not in (plain, first) + + +@pytest.mark.platforms("windows") +def test_junctions_bind_target_text_without_walking_outside(tmp_path): + import subprocess + + root = tmp_path / "tree" + root.mkdir() + left, right = tmp_path / "left", tmp_path / "right" + for target in (left, right): + target.mkdir() + (target / "content").write_bytes(b"same bytes") + junction = root / "junction" + command = str(Path(os.environ["SystemRoot"]) / "System32" / "cmd.exe") + + def point_at(target): + result = subprocess.run( + [command, "/d", "/c", "mklink", "/J", str(junction), str(target)], + capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=15, + ) + assert result.returncode == 0, result.stdout + result.stderr + assert junction.is_junction() + + point_at(left) + try: + first = tree_digest(root) + (left / "content").write_bytes(b"changed outside the tree") + assert tree_digest(root) == first + junction.rmdir() + point_at(right) + assert tree_digest(root) != first + finally: + junction.rmdir() + assert (left / "content").read_bytes() == b"changed outside the tree" + assert (right / "content").read_bytes() == b"same bytes"