diff --git a/hermes_cli/profile_distribution.py b/hermes_cli/profile_distribution.py index f7560473d5..d6bc7d428a 100644 --- a/hermes_cli/profile_distribution.py +++ b/hermes_cli/profile_distribution.py @@ -8,6 +8,7 @@ development before the first push). from __future__ import annotations import operator +import os import re import shutil import subprocess @@ -358,10 +359,11 @@ def _owned_entries(staged: Path, manifest: DistributionManifest): def _remove_existing(path: Path) -> None: """Remove one destination entry without following a destination symlink.""" - if path.is_symlink() or path.is_file(): - path.unlink() - elif path.is_dir(): + if path.is_dir() and not path.is_symlink(): shutil.rmtree(path) + elif os.path.lexists(path): + # Covers files, dangling/any symlinks, fifos and sockets alike. + path.unlink() def _replace_entry(src: Path, dest: Path) -> None: @@ -377,12 +379,18 @@ def _replace_entry(src: Path, dest: Path) -> None: def _real_dir(base: Path, parts: Tuple[str, ...]) -> Path: """Return ``base/parts`` as a chain of real directories. - A user could have swapped any ancestor for a symlink or a file; writing through it - would land the payload outside the profile, so each is replaced by a real directory.""" + A user could have swapped an ancestor for a file; writing through it is impossible, + so a file is replaced by a real directory. A symlinked ancestor is refused rather + than silently unlinked: it is deliberate user configuration (a shared skills dir, + say) and writing through it would land the payload outside the profile.""" path = base for part in parts: path = path / part - if path.is_symlink() or (path.exists() and not path.is_dir()): + if path.is_symlink(): + raise DistributionError( + f"{path} is a symlink; refusing to replace it — remove the link or point distribution_owned elsewhere" + ) + if path.exists() and not path.is_dir(): _remove_existing(path) path.mkdir(exist_ok=True) return path diff --git a/tests/hermes_cli/test_profile_distribution.py b/tests/hermes_cli/test_profile_distribution.py index 5480e6bba4..df1f75114c 100644 --- a/tests/hermes_cli/test_profile_distribution.py +++ b/tests/hermes_cli/test_profile_distribution.py @@ -10,6 +10,7 @@ mocking git would just test the mock. from __future__ import annotations +import shutil import sys from pathlib import Path @@ -424,6 +425,28 @@ class TestUpdate: assert (plan.target_dir / "cron" / "mine.json").read_text() == '{"schedule": "* * * * *"}\n' assert (plan.target_dir / "cron" / "daily.json").read_text() == '{"schedule": "0 10 * * *"}\n' + def test_update_refuses_symlinked_owned_container(self, profile_env): + staged = _make_staging_dir(profile_env, "src") + plan = install_distribution(str(staged), name="link_safe") + + shared = profile_env / "shared-skills" + shared.mkdir() + (shared / "mine" / "SKILL.md").parent.mkdir() + (shared / "mine" / "SKILL.md").write_text("shared skill\n") + before = sorted((p.relative_to(shared), p.read_bytes()) for p in shared.rglob("*") if p.is_file()) + + skills = plan.target_dir / "skills" + shutil.rmtree(skills) + _symlink_file_or_skip(skills, shared) + (staged / "skills" / "demo" / "SKILL.md").write_text("updated demo\n") + + with pytest.raises(DistributionError, match="symlink"): + update_distribution("link_safe") + + assert skills.is_symlink() and skills.resolve() == shared.resolve() + after = sorted((p.relative_to(shared), p.read_bytes()) for p in shared.rglob("*") if p.is_file()) + assert after == before + def test_update_preserves_user_data(self, profile_env): # 1. Build staging dir, install staged = _make_staging_dir(profile_env, "src")