fix(kanban): require lexical containment before scratch rmtree

The scratch-cleanup containment guard (#28818) resolved both the task's
workspace_path and the managed workspaces roots before comparing them.
When a root is itself a symlink to a broad directory (storage relocated
to another disk, or a planted link), every path inside the link target
resolves "under" the root. A legacy explicit-path scratch task naming such
a path directly then passed the guard, and task completion, deferred parent
cleanup and artifact persistence treated user data as scratch; completion
rmtree'd it.

Require the path to be strictly below the root lexically (absolute,
normalised, NFC, symlinks not followed) as well as after resolution. Tasks
created through the root are spelled through it, so relocated roots and
symlinked HERMES_HOMEs keep working. The root's lexical form is also
accepted with its anchor (kanban home, or the override's parent) resolved,
so a process that spells a symlinked home by its real path still matches;
the managed kanban/.../workspaces components are never resolved for this.

`hermes kanban gc` calls the same predicate once its own root-deletion
fix lands, so it inherits this check.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit 36b1d0453d153307e8d8d10e481e22392332fd2d)
This commit is contained in:
Kyle Caponi
2026-09-26 10:32:00 -05:00
committed by kshitij
parent 9ce7ca4ce1
commit b8a149cf48
2 changed files with 168 additions and 9 deletions

View File

@@ -64,36 +64,67 @@ def _has_active_children(conn: sqlite3.Connection, task_id: str) -> bool:
return conn.execute(_ACTIVE_CHILDREN_SQL, (task_id,)).fetchone() is not None
def _lexical_path(path: Path | str) -> Path:
"""Absolute, ``..``-collapsed, NFC form of *path* WITHOUT following symlinks."""
return Path(_path_key(os.path.abspath(path)))
def _managed_scratch_path_info(p: Path) -> tuple[bool, Optional[str]]:
"""Return whether *p* is managed scratch storage and the matching board."""
"""Return whether *p* is managed scratch storage and the matching board.
*p* must be strictly below a managed root both after resolving symlinks
AND lexically (as spelled, without resolving). Resolved containment alone
is not enough: when a root is itself a symlink to a broad directory
(relocated storage, or a planted link), every path inside the link target
would resolve "under" the root, so a scratch task naming such a path
directly would get it rmtree'd. Tasks created through the root are spelled
through it, so the lexical check keeps them managed. A root's lexical form
is accepted both as configured and with its anchor (kanban home, or the
override's parent) resolved, so a process spelling a symlinked home by its
real path still matches; the managed ``kanban/.../workspaces`` components
themselves are never resolved for the lexical check.
"""
try:
p_abs = p.resolve(strict=False)
except OSError:
return False, None
roots: list[tuple[Path, Optional[str]]] = []
p_lex = _lexical_path(p)
# (resolved root, lexical spellings of the root, board)
roots: list[tuple[Path, tuple[Path, ...], Optional[str]]] = []
def _add_root(anchor: Path, parts: tuple[str, ...], board: Optional[str]) -> None:
root = anchor.joinpath(*parts)
with contextlib.suppress(OSError):
roots.append((
root.resolve(strict=False),
(_lexical_path(root), _lexical_path(anchor.resolve(strict=False).joinpath(*parts))),
board,
))
override = os.environ.get("HERMES_KANBAN_WORKSPACES_ROOT", "").strip()
if override:
with contextlib.suppress(OSError):
roots.append((Path(override).expanduser().resolve(strict=False), None))
override_root = Path(override).expanduser()
_add_root(override_root.parent, (override_root.name,), None)
try:
home = _kb.kanban_home()
except OSError:
home = None
if home is not None:
with contextlib.suppress(OSError):
roots.append(((home / "kanban" / "workspaces").resolve(strict=False), _kb.DEFAULT_BOARD))
_add_root(home, ("kanban", "workspaces"), _kb.DEFAULT_BOARD)
entries: list[Path] = []
with contextlib.suppress(OSError):
entries = list((home / "kanban" / "boards").resolve(strict=False).iterdir())
for entry in entries:
with contextlib.suppress(OSError):
if entry.is_dir():
roots.append(((entry / "workspaces").resolve(strict=False), entry.name))
for root, board in roots:
_add_root(home, ("kanban", "boards", entry.name, "workspaces"), entry.name)
for root, lexical_roots, board in roots:
if p_abs == root:
continue
try:
if p_abs.is_relative_to(root):
if p_abs.is_relative_to(root) and any(
p_lex != lex and p_lex.is_relative_to(lex) for lex in lexical_roots
):
return True, board
except ValueError:
continue

View File

@@ -985,6 +985,134 @@ def test_is_managed_scratch_path_rejects_kanban_metadata_subtrees(kanban_home):
assert kb._is_managed_scratch_path(task_dir)
_needs_symlinks = pytest.mark.skipif(
sys.platform == "win32", reason="Symlinks require elevated privileges on Windows"
)
def _symlink_dir(link: Path, target: Path) -> None:
target.mkdir(parents=True, exist_ok=True)
if link.is_dir() and not link.is_symlink():
link.rmdir()
link.parent.mkdir(parents=True, exist_ok=True)
link.symlink_to(target, target_is_directory=True)
def _user_tree(root: Path) -> Path:
victim = root / "project"
victim.mkdir(parents=True)
(victim / "keep.txt").write_text("user data", encoding="utf-8")
return victim
def _resolve_task_workspace(conn, task_id: str) -> Path:
task = kb.get_task(conn, task_id)
assert task is not None
return kbw.resolve_workspace(task)
def _complete_scratch_task_at(conn, path: Path) -> str:
"""A legacy explicit-path scratch task pointing at *path*, then completed."""
t = kb.create_task(conn, title="scratch")
kbw.set_workspace_path(conn, t, path)
assert kb.complete_task(conn, t, result="done")
return t
@_needs_symlinks
def test_symlinked_workspaces_root_does_not_widen_scratch_cleanup(kanban_home, tmp_path):
"""A workspaces root that is a symlink to a broad directory must not make
every path inside the symlink target "managed". Only paths that are
lexically below the root (i.e. reached through it) are scratch; a path
named directly inside the target is user data (#28818)."""
broad = tmp_path / "user-data"
victim = _user_tree(broad)
_symlink_dir(kanban_home / "kanban" / "workspaces", broad)
assert not kb._is_managed_scratch_path(victim)
with kbc.connect() as conn:
_complete_scratch_task_at(conn, victim)
assert (victim / "keep.txt").read_text(encoding="utf-8") == "user data"
@_needs_symlinks
def test_symlinked_board_workspaces_root_does_not_widen_scratch_cleanup(kanban_home, tmp_path):
broad = tmp_path / "user-data"
victim = _user_tree(broad)
_symlink_dir(kanban_home / "kanban" / "boards" / "ops" / "workspaces", broad)
assert kb._managed_scratch_path_info(victim) == (False, None)
in_tree = kanban_home / "kanban" / "boards" / "ops" / "workspaces" / "t_1"
in_tree.mkdir()
assert kb._managed_scratch_path_info(in_tree) == (True, "ops")
@_needs_symlinks
def test_symlinked_workspaces_root_still_cleans_in_tree_scratch(kanban_home, tmp_path):
"""Relocating the workspaces root to another disk via a symlink keeps
scratch cleanup working for tasks created through the root."""
relocated = tmp_path / "big-disk" / "kanban-workspaces"
_symlink_dir(kanban_home / "kanban" / "workspaces", relocated)
with kbc.connect() as conn:
t = kb.create_task(conn, title="relocated scratch")
ws = _resolve_task_workspace(conn, t)
kbw.set_workspace_path(conn, t, ws)
assert (relocated / t).is_dir()
assert kb.complete_task(conn, t, result="done")
assert not (relocated / t).exists()
assert relocated.is_dir(), "the root itself is never removed"
@_needs_symlinks
def test_workspaces_root_override_symlink_is_the_lexical_anchor(kanban_home, tmp_path, monkeypatch):
"""``HERMES_KANBAN_WORKSPACES_ROOT`` (pinned into every worker) gets the
same treatment: tasks under the override are cleaned even when it is a
symlink; paths only inside its target are not."""
broad = tmp_path / "user-data"
victim = _user_tree(broad)
pinned = tmp_path / "pinned-workspaces"
_symlink_dir(pinned, broad)
monkeypatch.setenv("HERMES_KANBAN_WORKSPACES_ROOT", str(pinned))
assert not kb._is_managed_scratch_path(pinned)
assert not kb._is_managed_scratch_path(victim)
with kbc.connect() as conn:
t = kb.create_task(conn, title="pinned scratch")
ws = _resolve_task_workspace(conn, t)
assert ws == pinned / t
kbw.set_workspace_path(conn, t, ws)
assert kb.complete_task(conn, t, result="done")
_complete_scratch_task_at(conn, victim)
assert not (broad / t).exists()
assert (victim / "keep.txt").read_text(encoding="utf-8") == "user data"
@_needs_symlinks
def test_symlinked_hermes_home_scratch_cleanup(tmp_path, monkeypatch):
"""A HERMES_HOME that is a symlink (relocated storage) keeps cleanup
working, including for a path recorded by a process that spelled the
home by its real location."""
real_home = tmp_path / "storage" / "hermes"
real_home.mkdir(parents=True)
link_home = tmp_path / "hermes-link"
link_home.symlink_to(real_home, target_is_directory=True)
monkeypatch.setattr(Path, "home", lambda: tmp_path)
monkeypatch.setenv("HERMES_HOME", str(link_home))
kb.init_db()
assert kb.kanban_home() == link_home
with kbc.connect() as conn:
linked = kb.create_task(conn, title="linked spelling")
linked_ws = _resolve_task_workspace(conn, linked)
kbw.set_workspace_path(conn, linked, linked_ws)
real_ws = real_home / "kanban" / "workspaces" / "t_realpath"
real_ws.mkdir(parents=True)
assert kb.complete_task(conn, linked, result="done")
_complete_scratch_task_at(conn, real_ws)
assert not linked_ws.exists()
assert not real_ws.exists()
# ---------------------------------------------------------------------------
# Tenancy
# ---------------------------------------------------------------------------