fix(#4405): contain rollback checkpoint restore writes to the workspace

Restore previously used pathname-based shutil.copy2() which follows an in-tree
symlink whose target escapes the workspace. Route restore writes through the
workspace-anchored file helpers so a restored file always lands inside the
workspace tree. +regression test (escaping symlink refused, normal restore ok).

Co-authored-by: hinotoi-agent <paperlantern.agent@gmail.com>
This commit is contained in:
nesquena-hermesandhinotoi-agent committed 2026-06-18 05:12:49 +00:00
1 parent f46a447dc3
commit 330e0672b8
3 files changed
+119 -7

No files matched your search

+6
View File
@@ -9,6 +9,12 @@
- **The conversation-outline toggle button no longer overlaps the scroll-to-bottom control (#2124).** The outline FAB was `position:fixed` at a high z-index and could stack on top of the scroll-to-bottom button in the bottom-right corner. It's now anchored inside the messages shell (`position:absolute`, lower z-index) so the two controls sit cleanly without colliding. Thanks @Habib1001-m.
## [v0.51.491] — 2026-06-18 — Release RA (rollback restore writes contained to the workspace)
### Security
- **Rollback checkpoint restore writes can no longer be redirected outside the selected workspace by a symlink planted inside it (#4405).** The restore path previously used pathname-based `shutil.copy2()` writes, which would follow a symlink whose target escapes the workspace. Restore writes now go through the existing workspace-anchored file helpers, so a restored file always lands inside the workspace tree regardless of in-tree symlinks. Thanks @hinotoi-agent.
## [v0.51.490] — 2026-06-18 — Release QZ (large plain-text pastes become .md attachments)
### Added
+30 -7
View File
@@ -7,13 +7,12 @@ created by the Hermes agent's CheckpointManager. Checkpoints live at
"""
import hashlib
import json
import logging
import os
import re
import shutil
import subprocess
from datetime import datetime, timezone
from datetime import datetime
from pathlib import Path
from typing import Any
@@ -237,7 +236,7 @@ def get_checkpoint_diff(workspace: str, checkpoint: str) -> dict[str, Any]:
# File exists in checkpoint but not in workspace (deleted)
files_changed.append({"file": rel_path, "status": "deleted"})
diff_lines.append(f"--- a/{rel_path}")
diff_lines.append(f"+++ /dev/null")
diff_lines.append("+++ /dev/null")
diff_lines.append("@@ -1,{lines} +0,0 @@".format(lines=len(ckpt_content.splitlines())))
for line in ckpt_content.splitlines():
diff_lines.append(f"-{line}")
@@ -263,6 +262,31 @@ def get_checkpoint_diff(workspace: str, checkpoint: str) -> dict[str, Any]:
}
def _restore_checkpoint_file(workspace_root: Path, ckpt_file: Path, rel_path: str) -> None:
"""Restore one checkpoint file without following workspace symlinks outward."""
from api.workspace import open_anchored_create_fd, open_anchored_write_fd, safe_resolve_ws
target = safe_resolve_ws(workspace_root, rel_path)
if target.exists():
fd = open_anchored_write_fd(workspace_root, target)
else:
fd = open_anchored_create_fd(workspace_root, target)
src_stat = ckpt_file.stat()
with os.fdopen(fd, "wb") as out:
with ckpt_file.open("rb") as src:
shutil.copyfileobj(src, out)
out.flush()
try:
os.fchmod(out.fileno(), src_stat.st_mode & 0o777)
except (AttributeError, OSError):
logger.debug("Failed to apply restored mode to %s", target, exc_info=True)
try:
os.utime(out.fileno(), ns=(src_stat.st_atime_ns, src_stat.st_mtime_ns))
except OSError:
logger.debug("Failed to apply restored timestamp to %s", target, exc_info=True)
def restore_checkpoint(workspace: str, checkpoint: str) -> dict[str, Any]:
"""Restore a checkpoint by copying files back to the workspace.
@@ -274,6 +298,7 @@ def restore_checkpoint(workspace: str, checkpoint: str) -> dict[str, Any]:
files_restored: list of restored file paths
"""
resolved = _resolve_workspace(workspace)
workspace_root = Path(resolved)
checkpoint = _validate_checkpoint_id(checkpoint)
ws_hash = _workspace_hash(resolved)
ckpt_dir = _checkpoint_root() / ws_hash / checkpoint
@@ -297,16 +322,14 @@ def restore_checkpoint(workspace: str, checkpoint: str) -> dict[str, Any]:
for rel_path in ckpt_files:
ckpt_file = ckpt_dir / rel_path
ws_file = Path(resolved) / rel_path
if not ckpt_file.is_file():
continue
try:
ws_file.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(str(ckpt_file), str(ws_file))
_restore_checkpoint_file(workspace_root, ckpt_file, rel_path)
restored.append(rel_path)
except OSError as e:
except (OSError, ValueError) as e:
errors.append({"file": rel_path, "error": str(e)})
logger.warning("Failed to restore %s: %s", rel_path, e)
@@ -0,0 +1,83 @@
import subprocess
from api import rollback
def _init_checkpoint_repo(path, files):
path.mkdir(parents=True)
subprocess.run(["git", "-C", str(path), "init", "-q"], check=True)
subprocess.run(["git", "-C", str(path), "config", "user.email", "test@example.com"], check=True)
subprocess.run(["git", "-C", str(path), "config", "user.name", "Test"], check=True)
for rel_path, content in files.items():
target = path / rel_path
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(content)
subprocess.run(["git", "-C", str(path), "add", "."], check=True)
subprocess.run(["git", "-C", str(path), "commit", "-q", "-m", "checkpoint"], check=True)
def test_restore_checkpoint_refuses_symlink_escape(monkeypatch, tmp_path):
workspace = tmp_path / "workspace"
workspace.mkdir()
outside = tmp_path / "outside"
outside.mkdir()
outside_target = outside / "target.txt"
outside_target.write_text("ORIGINAL")
(workspace / "victim.txt").symlink_to(outside_target)
checkpoint_root = tmp_path / "checkpoints"
checkpoint_id = "checkpoint-1"
ckpt_dir = checkpoint_root / rollback._workspace_hash(str(workspace.resolve())) / checkpoint_id
_init_checkpoint_repo(ckpt_dir, {"victim.txt": "CHECKPOINT_MARKER"})
monkeypatch.setattr(rollback, "_resolve_workspace", lambda workspace_arg: str(workspace.resolve()))
monkeypatch.setattr(rollback, "_checkpoint_root", lambda: checkpoint_root)
result = rollback.restore_checkpoint(str(workspace), checkpoint_id)
assert result["ok"] is True
assert result["files_restored"] == []
assert result["files_restored_count"] == 0
assert result["errors"]
assert result["errors"][0]["file"] == "victim.txt"
assert outside_target.read_text() == "ORIGINAL"
def test_restore_checkpoint_restores_regular_file(monkeypatch, tmp_path):
workspace = tmp_path / "workspace"
workspace.mkdir()
(workspace / "nested").mkdir()
(workspace / "nested" / "file.txt").write_text("OLD")
checkpoint_root = tmp_path / "checkpoints"
checkpoint_id = "checkpoint-2"
ckpt_dir = checkpoint_root / rollback._workspace_hash(str(workspace.resolve())) / checkpoint_id
_init_checkpoint_repo(ckpt_dir, {"nested/file.txt": "NEW"})
monkeypatch.setattr(rollback, "_resolve_workspace", lambda workspace_arg: str(workspace.resolve()))
monkeypatch.setattr(rollback, "_checkpoint_root", lambda: checkpoint_root)
result = rollback.restore_checkpoint(str(workspace), checkpoint_id)
assert result["ok"] is True
assert result["files_restored"] == ["nested/file.txt"]
assert (workspace / "nested" / "file.txt").read_text() == "NEW"
def test_restore_checkpoint_creates_missing_regular_file(monkeypatch, tmp_path):
workspace = tmp_path / "workspace"
workspace.mkdir()
checkpoint_root = tmp_path / "checkpoints"
checkpoint_id = "checkpoint-3"
ckpt_dir = checkpoint_root / rollback._workspace_hash(str(workspace.resolve())) / checkpoint_id
_init_checkpoint_repo(ckpt_dir, {"nested/new-file.txt": "NEW"})
monkeypatch.setattr(rollback, "_resolve_workspace", lambda workspace_arg: str(workspace.resolve()))
monkeypatch.setattr(rollback, "_checkpoint_root", lambda: checkpoint_root)
result = rollback.restore_checkpoint(str(workspace), checkpoint_id)
assert result["ok"] is True
assert result["files_restored"] == ["nested/new-file.txt"]
assert (workspace / "nested" / "new-file.txt").read_text() == "NEW"