fix(distribution): refuse to replace a symlinked owned container
A user who points `profiles/x/skills` at a shared directory did so on purpose. The merge path replaced such a symlink with a real directory silently, discarding that configuration, where the base rmtree at least failed loudly. Raise DistributionError naming the link instead; the file->directory replacement stays because a stray file there is never deliberate configuration. `_remove_existing` now removes anything that lexists but is not a real directory via unlink, so fifos and sockets no longer slip through. Reviewer P1 on the original PR (symlink container must survive intact).
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user