From 330e0672b857e52e3f1512deaefdcd0beecf533d Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Thu, 18 Jun 2026 04:46:50 +0000 Subject: [PATCH] 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 --- CHANGELOG.md | 6 ++ api/rollback.py | 37 +++++++-- ...st_rollback_restore_symlink_containment.py | 83 +++++++++++++++++++ 3 files changed, 119 insertions(+), 7 deletions(-) create mode 100644 tests/test_rollback_restore_symlink_containment.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1cf0a43f..cf104f46 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/api/rollback.py b/api/rollback.py index 18d888fd..d026a9fd 100644 --- a/api/rollback.py +++ b/api/rollback.py @@ -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) diff --git a/tests/test_rollback_restore_symlink_containment.py b/tests/test_rollback_restore_symlink_containment.py new file mode 100644 index 00000000..397b879d --- /dev/null +++ b/tests/test_rollback_restore_symlink_containment.py @@ -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"