diff --git a/tests/tools/test_file_write_safety.py b/tests/tools/test_file_write_safety.py index 5e2b5ac60c..2ae8441dc6 100644 --- a/tests/tools/test_file_write_safety.py +++ b/tests/tools/test_file_write_safety.py @@ -485,6 +485,25 @@ class TestBomHandling: assert not res.success assert target.read_bytes() == precious + def test_move_refuses_a_destination_it_could_not_read_before_any_op_applies( + self, tmp_path: Path, monkeypatch): + # Validation must keep "the read failed" apart from "the path is absent" for a Move + # destination too, and the apply must re-check it before `mv` replaces whatever is there. + from tools.file_operations import ShellFileOperations + monkeypatch.setenv("HERMES_NATIVE_FILE_READ", "0") + dst = tmp_path / "dst.txt" + dst.write_bytes(b"PRECIOUS DESTINATION\n") + ops = ShellFileOperations(self._env_without("base64", "od")(cwd=str(tmp_path)), cwd=str(tmp_path)) + + res = ops.patch_v4a( + "*** Begin Patch\n" + f"*** Add File: {tmp_path / 'new-src.txt'}\n+SOURCE\n" + f"*** Move File: {tmp_path / 'new-src.txt'} -> {dst}\n" + "*** End Patch") + assert not res.success + assert dst.read_bytes() == b"PRECIOUS DESTINATION\n" + assert not (tmp_path / "new-src.txt").exists() + def test_native_byte_exact_read_never_opens_a_non_regular_file(self, tmp_path: Path, monkeypatch): # The native fast path bypasses the backend timeout, so a blocking open there hangs the # thread with nothing to interrupt it. The shell path below has a timeout and is allowed diff --git a/tools/patch_parser.py b/tools/patch_parser.py index fd941a8f61..8e6ce92ed7 100644 --- a/tools/patch_parser.py +++ b/tools/patch_parser.py @@ -164,6 +164,20 @@ def _validate_operations(operations: List[PatchOperation], file_ops: Any) -> Lis r = file_ops.read_file_raw(path) return (None, r.error) if r.error else (r.content, None) + def _occupied(path: str) -> Optional[str]: + """Why an Add target or Move destination is not free, or None. Only a read that reports + the path absent (``not_found``) frees it: a read that FAILED (no byte transport, a + directory, an unreadable file) says nothing about what is there, and taking it as free + writes over the file the check exists to protect.""" + if path in pending_content: + return "exists" + if path in removed_paths: + return None + r = file_ops.read_file_raw(path) + if not r.error: + return "exists" + return None if getattr(r, "not_found", False) else f"could not confirm the path is free — {r.error}" + def _validate_update(op: PatchOperation) -> None: nonlocal real_change_count simulated, read_err = _read(op.file_path) @@ -222,8 +236,11 @@ def _validate_operations(operations: List[PatchOperation], file_ops: Any) -> Lis src_content, src_err = _read(op.file_path) if src_err: errors.append(f"{op.file_path}: source file not found for move") - if not _read(op.new_path)[1]: + dst_taken = _occupied(op.new_path) + if dst_taken == "exists": errors.append(f"{op.new_path}: destination already exists — move would overwrite") + elif dst_taken: + errors.append(f"{op.new_path}: {dst_taken}") elif not src_err: # only a cleanly-validated move updates the overlay pending_content[op.new_path] = src_content if src_content is not None else "" _remove(op.file_path) @@ -235,8 +252,11 @@ def _validate_operations(operations: List[PatchOperation], file_ops: Any) -> Lis # the MOVE destination guard. Overlay-aware: an Add after a Delete of the # same path in this patch stays legal, and the added content enters the # overlay so later hunks against it validate. - if not _read(op.file_path)[1]: + add_taken = _occupied(op.file_path) + if add_taken == "exists": errors.append(f"{op.file_path}: file already exists — use Update File, not Add File") + elif add_taken: + errors.append(f"{op.file_path}: {add_taken}") else: removed_paths.discard(op.file_path) pending_content[op.file_path] = '\n'.join( @@ -352,6 +372,13 @@ def _apply_delete(op: PatchOperation, file_ops: Any) -> ApplyResult: def _apply_move(op: PatchOperation, file_ops: Any) -> ApplyResult: + """Move, re-checking the destination first: validation's answer is stale once earlier ops of + this patch have applied, and ``mv`` replaces whatever is there.""" + dst = file_ops.read_file_raw(op.new_path) + if not dst.error: + return _fail(f"{op.new_path}: destination already exists — move would overwrite") + if not getattr(dst, "not_found", False): + return _fail(f"{op.new_path}: could not confirm the destination is free — {dst.error}") result = file_ops.move_file(op.file_path, op.new_path) return _fail(result.error) if result.error else ( True, f"# Moved: {op.file_path} -> {op.new_path}", None, None)