hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

fix(patch): V4A Delete and Move act on a symlink itself, not the file it points to

Browse Source 
patch_tool rewrites every V4A header to the path _resolve_path_for_task
returns, and on a host backend that is Path.resolve(), which follows a
symlink in the last component. For Update/Add that is harmless: the shell
layer reads and writes the target through the link either way. Delete and
Move act on the directory entry, so "*** Delete File: config/local.yaml"
(a link to base.yaml) deleted base.yaml and left the link dangling, and
"*** Move File: current.txt -> previous.txt" renamed the link's target,
both reported as success.

Delete headers and both Move endpoints now resolve their parent directory
only (_resolve_entry_for_task), keeping the final component, so the link
is removed or renamed. The same paths are locked and reported in
files_modified. Update/Add headers are unchanged.

(cherry picked from commit 7b1fe43d30db6015155349b67ca814b340df0e18)
This commit is contained in:
John Paul Soliva
2026-09-26 19:16:09 +09:00
committed by kshitij
parent d915436657
commit 0a99750128
3 changed files with 92 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
tools/file_tools.py
View File

@@ -28,7 +28,8 @@ from tools.file_operations_common import DEFAULT_READ_LIMIT, count_conflict_bloc
from tools import file_state
from agent.redact import _is_secret_file_arg, redact_sensitive_text
from tools.file_tools_paths import (
_expand_tilde, _path_resolution_warning, _resolve_base_dir, _resolve_path_for_task)
_expand_tilde, _path_resolution_warning, _resolve_base_dir, _resolve_entry_for_task,
_resolve_path_for_task)
from tools.file_tools_write_guards import (
_READ_DEDUP_STATUS_MESSAGE, _check_approval_required_write, _check_binary_document_write,
_check_cross_profile_path, _check_protected_instruction_write, _check_sensitive_path,
@@ -150,22 +151,26 @@ _V4A_SINGLE_HEADER_RE = re.compile(r'^(\*\*\*\s*(Update|Add|Delete)\s+File:\s*)(
_V4A_MOVE_HEADER_RE = re.compile(r'^(\*\*\*\s*Move\s+File:\s*)(.+?)\s*->\s*(.+)$', re.MULTILINE)
def _rewrite_v4a_patch_paths_for_host(patch: str, path_to_resolved: dict, file_ops) -> str:
def _rewrite_v4a_patch_paths_for_host(patch: str, path_to_resolved: dict, path_to_entry: dict,
file_ops) -> str:
"""Rewrite V4A file headers to the resolved host paths (host backends only).
The shell layer must patch the SAME files ``patch_tool`` resolved for
locking/staleness, not re-resolve a relative header against its own cwd
(which can differ — the git-worktree cwd bug).
(which can differ — the git-worktree cwd bug). Delete and Move headers take
``path_to_entry``: they act on a symlink itself, never on its target.
"""
if not _file_ops_uses_host_paths(file_ops):
return patch
def _res(raw: str) -> str:
def _res(raw: str, entry: bool = False) -> str:
raw = raw.strip()
return path_to_resolved.get(raw) or raw
return (path_to_entry if entry else path_to_resolved).get(raw) or raw
patch = _V4A_SINGLE_HEADER_RE.sub(lambda m: f"{m.group(1)}{_res(m.group(3))}", patch)
return _V4A_MOVE_HEADER_RE.sub(lambda m: f"{m.group(1)}{_res(m.group(2))} -> {_res(m.group(3))}", patch)
patch = _V4A_SINGLE_HEADER_RE.sub(
lambda m: f"{m.group(1)}{_res(m.group(3), entry=m.group(2) == 'Delete')}", patch)
return _V4A_MOVE_HEADER_RE.sub(
lambda m: f"{m.group(1)}{_res(m.group(2), entry=True)} -> {_res(m.group(3), entry=True)}", patch)
def _is_blocked_device_path(path: str) -> bool:
@@ -750,10 +755,11 @@ def read_file_tool(path: str, offset: int = 1, limit: int = DEFAULT_READ_LIMIT,
# ── Shared write/patch plumbing ──────────────────────────────────────────
def _resolve_or_none(filepath: str, task_id: str) -> str | None:
"""Task-resolved path string, or None when resolution fails for any reason."""
def _resolve_or_none(filepath: str, task_id: str, *, entry: bool = False) -> str | None:
"""Task-resolved path string, or None when resolution fails for any reason.
``entry``: keep a symlink in the last component (``_resolve_entry_for_task``)."""
try:
return str(_resolve_path_for_task(filepath, task_id))
return str((_resolve_entry_for_task if entry else _resolve_path_for_task)(filepath, task_id))
except Exception:
return None
@@ -917,13 +923,14 @@ def write_file_tool(path: str, content: str, task_id: str = "default",
return tool_error(str(e))
def _collect_v4a_header_paths(patch: str) -> tuple[list[str], list[str]] | str:
def _collect_v4a_header_paths(patch: str) -> tuple[list[str], list[str], list[str]] | str:
"""Extract every path named in V4A headers, rejecting ``..`` traversal.
Returns ``(all_paths, content_write_paths)`` or a tool_error string. Header
paths come from patch CONTENT (more attacker-influenceable than ``path=``,
which keeps its legitimate ``..`` use). Move headers check BOTH endpoints;
only Update/Add write text and feed the binary-document guard.
Returns ``(all_paths, content_write_paths, entry_paths)`` or a tool_error
string. Header paths come from patch CONTENT (more attacker-influenceable
than ``path=``, which keeps its legitimate ``..`` use). Move headers check
BOTH endpoints; only Update/Add write text and feed the binary-document
guard; Delete and Move act on the directory entry (``entry_paths``).
"""
from tools.path_security import has_traversal_component
@@ -931,6 +938,7 @@ def _collect_v4a_header_paths(patch: str) -> tuple[list[str], list[str]] | str:
headers += [(g, False) for m in _V4A_MOVE_HEADER_RE.finditer(patch) for g in (m.group(2), m.group(3))]
paths: list[str] = []
content_paths: list[str] = []
entry_paths: list[str] = []
for raw, writes_text in headers:
v4a_path = raw.strip()
if has_traversal_component(v4a_path):
@@ -940,9 +948,8 @@ def _collect_v4a_header_paths(patch: str) -> tuple[list[str], list[str]] | str:
"path in '*** Update File:' / '*** Add File:' / "
"'*** Delete File:' / '*** Move File:' headers.")
paths.append(v4a_path)
if writes_text:
content_paths.append(v4a_path)
return paths, content_paths
(content_paths if writes_text else entry_paths).append(v4a_path)
return paths, content_paths, entry_paths
def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
@@ -956,12 +963,14 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
"""
_paths_to_check = [path] if path else []
_content_write_paths = list(_paths_to_check)
_entry_paths: list[str] = []
if mode == "patch" and patch:
collected = _collect_v4a_header_paths(patch)
if isinstance(collected, str):
return collected
_paths_to_check += collected[0]
_content_write_paths += collected[1]
_entry_paths = collected[2]
precheck_err = _write_precheck_error(_paths_to_check, _content_write_paths, task_id, cross_profile)
if precheck_err:
return tool_error(precheck_err)
@@ -970,8 +979,9 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
# overlapping multi-file patches can't deadlock (every caller locks in
# the same order). An unresolvable path is simply not locked.
_path_to_resolved: dict[str, str] = {_p: _resolve_or_none(_p, task_id) for _p in _paths_to_check}
_path_to_entry: dict[str, str] = {_p: _resolve_or_none(_p, task_id, entry=True) for _p in _entry_paths}
with ExitStack() as _locks:
for _r in sorted({_r for _r in _path_to_resolved.values() if _r}):
for _r in sorted({_r for _r in (*_path_to_resolved.values(), *_path_to_entry.values()) if _r}):
_locks.enter_context(file_state.lock_path(_r))
stale_warnings = _edit_warnings(_paths_to_check, _path_to_resolved, task_id)
file_ops = _get_file_ops(task_id)
@@ -988,7 +998,8 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
elif mode == "patch":
if not patch:
return tool_error("patch content required")
result = file_ops.patch_v4a(_rewrite_v4a_patch_paths_for_host(patch, _path_to_resolved, file_ops))
result = file_ops.patch_v4a(
_rewrite_v4a_patch_paths_for_host(patch, _path_to_resolved, _path_to_entry, file_ops))
else:
return tool_error(f"Unknown mode: {mode}")
@@ -998,7 +1009,8 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
if not result_dict.get("error"):
# Report the ABSOLUTE path(s) actually patched so a wrong-cwd
# mismatch is visible instead of silently landing elsewhere.
_resolved_modified = [_path_to_resolved.get(_p) or _p for _p in _paths_to_check]
_resolved_modified = [_path_to_entry.get(_p) or _path_to_resolved.get(_p) or _p
for _p in _paths_to_check]
result_dict["files_modified"] = _resolved_modified
if len(_resolved_modified) == 1:
result_dict["resolved_path"] = _resolved_modified[0]

10
tools/file_tools_paths.py
View File

@@ -272,6 +272,16 @@ def _resolve_path_for_task(filepath: str, task_id: str = "default") -> Path | Pu
lambda: _resolve_base_dir(task_id, container_paths=container_paths), container_paths)
def _resolve_entry_for_task(filepath: str, task_id: str = "default") -> Path | PurePosixPath:
"""``_resolve_path_for_task`` for an operation on the directory entry itself (delete,
rename): the parent is resolved, but a symlink in the last component is kept, since
resolving it aims the operation at the file the link points to."""
parent, name = os.path.split(filepath)
if name in ("", ".", "..") or (not parent and name.startswith("~")):
return _resolve_path_for_task(filepath, task_id)
return _resolve_path_for_task(parent or ".", task_id) / name
def _path_resolution_warning(filepath: str, resolved: Path | PurePosixPath, task_id: str = "default") -> str | None:
"""Warn when a RELATIVE path resolved OUTSIDE the task's workspace root (the