Merge pull request #6205 from MervinPraison/claude/issue-6204-20261103

fix: crash-safe writes in praisonai.code edit tools

src/praisonai/praisonai/code/tools/apply_diff.py

@@ -10,6 +10,7 @@ from typing import Optional, Dict, Any
 
 from ..diff.diff_strategy import apply_search_replace_diff
 from ..utils.file_utils import (
+    atomic_write,
     file_exists,
     is_path_within_directory,
 )
@@ -142,8 +143,7 @@ def apply_diff(
                 f.write(original_content)
         
         # Write the modified content
-        with open(abs_path, 'w', encoding=encoding) as f:
-            f.write(result.content)
+        atomic_write(abs_path, result.content, encoding=encoding)
         
         return {
             'success': True,

src/praisonai/praisonai/code/tools/search_replace.py

@@ -10,6 +10,7 @@ import re
 from typing import Optional, List, Dict, Any
 
 from ..utils.file_utils import (
+    atomic_write,
     file_exists,
     is_path_within_directory,
 )
@@ -191,8 +192,7 @@ def search_replace(
                 f.write(original_content)
         
         # Write the modified content
-        with open(abs_path, 'w', encoding=encoding) as f:
-            f.write(content)
+        atomic_write(abs_path, content, encoding=encoding)
         
         return {
             'success': True,

src/praisonai/praisonai/code/tools/write_file.py

@@ -8,6 +8,7 @@ import os
 from typing import Optional, Dict, Any
 
 from ..utils.file_utils import (
+    atomic_write,
     file_exists,
     create_directories_for_file,
     is_path_within_directory,
@@ -131,8 +132,7 @@ def write_file(
                 pass  # If we can't read, just use the content as-is
         
         # Write the file
-        with open(abs_path, 'w', encoding=encoding) as f:
-            f.write(processed_content)
+        atomic_write(abs_path, processed_content, encoding=encoding)
         
         return {
             'success': True,

src/praisonai/praisonai/code/utils/file_utils.py

@@ -8,8 +8,10 @@ Provides utilities for file operations including:
 - Line ending normalization
 """
 
+import errno
 import os
 import re
+import uuid
 from typing import Optional, Tuple
 
 
@@ -305,3 +307,68 @@ def count_file_lines(file_path: str) -> int:
         for _ in f:
             count += 1
     return count
+
+
+def atomic_write(path: str, content: str, *, encoding: Optional[str] = "utf-8") -> None:
+    """Write ``content`` to ``path`` atomically (temp file + fsync + os.replace).
+
+    ``open(path, 'w')`` truncates the target immediately, so any interrupt
+    (cancellation, SIGTERM, disk-full) between open and write leaves the file
+    zero-byte or half-written and the original content lost. The code editing
+    tools run on LLM-driven tool calls that regularly hit timeouts/cancellations,
+    so they need the same crash-safe write the wrapper already uses for configs.
+    """
+    # Resolve symlinks so we replace the link target (like open(path, 'w')
+    # writes through the link) instead of turning the symlink into a plain file.
+    path = os.path.realpath(path)
+    target_dir = os.path.dirname(path) or "."
+    # Atomic replace swaps in a fresh inode, so carry over the metadata that
+    # matters for a replaced file: permission bits and ownership (uid/gid).
+    # Hard-link identity and xattrs/ACLs cannot be preserved through an inode
+    # swap and are out of scope for this tool.
+    try:
+        existing_stat = os.stat(path)
+    except OSError:
+        existing_stat = None
+    # An inode swap would bypass a read-only target's permissions, so keep the
+    # old open(path, 'w') behaviour and refuse to replace a file we can't write.
+    if existing_stat is not None and not os.access(path, os.W_OK):
+        raise PermissionError(errno.EACCES, os.strerror(errno.EACCES), path)
+    # Create the temp file with the target's permission bits (setuid/setgid
+    # dropped) so private files (e.g. 0600 .env) are never readable by others
+    # while the new content is written. New files use 0o666 so the kernel
+    # applies the umask, like open(path, 'w'). Never call os.umask(): it is
+    # process-wide and would race with other threads.
+    mode = (existing_stat.st_mode & 0o777) if existing_stat is not None else 0o666
+    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0)
+    while True:
+        tmp_path = os.path.join(target_dir, f".tmp_{uuid.uuid4().hex}")
+        try:
+            fd = os.open(tmp_path, flags, mode)
+            break
+        except FileExistsError:
+            continue
+    try:
+        with os.fdopen(fd, "w", encoding=encoding) as f:
+            f.write(content)
+            f.flush()
+            os.fsync(f.fileno())  # durability before rename
+        if existing_stat is not None:
+            # Restore bits the umask may have removed; never widens beyond the
+            # target. If this fails the temp file is at most as open as the target.
+            try:
+                os.chmod(tmp_path, mode)
+            except OSError:
+                pass  # best-effort
+            if hasattr(os, "chown"):
+                try:
+                    os.chown(tmp_path, existing_stat.st_uid, existing_stat.st_gid)
+                except (OSError, AttributeError):
+                    pass  # best-effort (needs privilege; no-op on Windows)
+        os.replace(tmp_path, path)  # atomic on POSIX + Windows
+    except BaseException:
+        try:
+            os.unlink(tmp_path)
+        except OSError:
+            pass
+        raise

src/praisonai/praisonai/tools/multiedit.py

@@ -8,55 +8,12 @@ without multiple file read/write cycles.
 import contextlib
 import difflib
 import os
-import tempfile
 from typing import Any, Dict, List, Optional
 
+from praisonai.code.utils.file_utils import atomic_write as _atomic_write
 from praisonai.code.utils.file_utils import is_path_within_directory
 
 
-def _atomic_write(path: str, content: str) -> None:
-    """Write ``content`` to ``path`` atomically (temp file + fsync + os.replace).
-
-    ``open(path, 'w')`` truncates the target immediately, so any interrupt
-    (cancellation, SIGTERM, disk-full) between open and write leaves the file
-    zero-byte or half-written and the original content lost. multiedit runs on
-    LLM-driven tool calls that regularly hit timeouts/cancellations, so it needs
-    the same crash-safe write the rest of the wrapper already uses for configs.
-    """
-    target_dir = os.path.dirname(os.path.abspath(path)) or "."
-    fd, tmp_path = tempfile.mkstemp(prefix=".tmp_", dir=target_dir)
-    try:
-        # Atomic replace swaps in a fresh inode, so carry over the metadata that
-        # matters for a replaced file: mode (mkstemp creates 0600 by default) and
-        # ownership (uid/gid) so editing a group-/shared-owned file does not
-        # silently change its owner. Hard-link identity and xattrs/ACLs cannot be
-        # preserved through an inode swap and are out of scope for this tool.
-        try:
-            existing_stat = os.stat(path)
-        except OSError:
-            existing_stat = None
-        with os.fdopen(fd, "w") as f:
-            f.write(content)
-            f.flush()
-            os.fsync(f.fileno())  # durability before rename
-        if existing_stat is not None:
-            try:
-                os.chmod(tmp_path, existing_stat.st_mode)
-            except OSError:
-                pass  # best-effort
-            if hasattr(os, "chown"):
-                try:
-                    os.chown(tmp_path, existing_stat.st_uid, existing_stat.st_gid)
-                except (OSError, AttributeError):
-                    pass  # best-effort (needs privilege; no-op on Windows)
-        os.replace(tmp_path, path)  # atomic on POSIX + Windows
-    except BaseException:
-        try:
-            os.unlink(tmp_path)
-        except OSError:
-            pass
-        raise
-
 
 @contextlib.contextmanager
 def _file_lock(path: str):
@@ -250,7 +207,7 @@ def _run_edits(
 
     # Write file atomically if not dry run (crash-safe: no zero-byte truncation)
     if not dry_run and result["edits_applied"] > 0:
-        _atomic_write(filepath, content)
+        _atomic_write(filepath, content, encoding=None)  # match locale-default read above
 
     result["success"] = result["edits_failed"] == 0
 

src/praisonai/tests/unit/code/test_tools.py

@@ -4,6 +4,7 @@ Tests for praisonai.code.tools module.
 
 import os
 import tempfile
+import pytest
 
 
 class TestReadFile:
@@ -286,7 +287,129 @@ class TestSearchReplace:
             
             with open(file_path, 'r') as f:
                 assert f.read() == "Hello, new world!"
-    
+
+    def test_interrupted_write_keeps_original(self, monkeypatch):
+        """A failure mid-write must not truncate the original file."""
+        from praisonai.code.tools.search_replace import search_replace
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            file_path = os.path.join(temp_dir, "test.txt")
+            with open(file_path, 'w') as f:
+                f.write("Hello, old world!")
+
+            def boom(*args, **kwargs):
+                raise KeyboardInterrupt
+
+            monkeypatch.setattr(os, "fsync", boom)
+            try:
+                search_replace(file_path, [
+                    {'search': 'old', 'replace': 'new'}
+                ], workspace=temp_dir)
+            except KeyboardInterrupt:
+                pass
+
+            with open(file_path, 'r') as f:
+                assert f.read() == "Hello, old world!"
+            assert os.listdir(temp_dir) == ["test.txt"]
+
+    @pytest.mark.skipif(os.name == "nt", reason="POSIX permissions")
+    def test_atomic_write_new_file_respects_umask(self):
+        """New files get 0666 & ~umask like open(path, 'w'), not mkstemp's 0600."""
+        from praisonai.code.utils.file_utils import atomic_write
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            old = os.umask(0o022)
+            try:
+                path = os.path.join(temp_dir, "new.txt")
+                atomic_write(path, "x")
+            finally:
+                os.umask(old)
+            assert (os.stat(path).st_mode & 0o777) == 0o644
+
+    @pytest.mark.skipif(os.name == "nt", reason="POSIX permissions")
+    def test_atomic_write_does_not_touch_process_umask(self, monkeypatch):
+        """os.umask is process-wide; changing it races with other threads."""
+        from praisonai.code.utils.file_utils import atomic_write
+
+        def no_umask(*args, **kwargs):
+            raise AssertionError("atomic_write must not call os.umask")
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            monkeypatch.setattr(os, "umask", no_umask)
+            new_path = os.path.join(temp_dir, "new.txt")
+            atomic_write(new_path, "a")
+            atomic_write(new_path, "b")
+            with open(new_path) as f:
+                assert f.read() == "b"
+            assert os.listdir(temp_dir) == ["new.txt"]
+
+    @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks unsupported")
+    def test_atomic_write_writes_through_symlink(self):
+        """Like open(path, 'w'), editing a symlink must update its target and keep the link."""
+        from praisonai.code.utils.file_utils import atomic_write
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            real = os.path.join(temp_dir, "real.txt")
+            link = os.path.join(temp_dir, "link.txt")
+            with open(real, "w") as f:
+                f.write("orig")
+            try:
+                os.symlink(real, link)
+            except OSError:
+                pytest.skip("cannot create symlink")
+            atomic_write(link, "new")
+            assert os.path.islink(link)
+            with open(real) as f:
+                assert f.read() == "new"
+            assert sorted(os.listdir(temp_dir)) == ["link.txt", "real.txt"]
+
+    @pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits")
+    def test_atomic_write_temp_file_never_wider_than_private_target(self, monkeypatch):
+        """A 0600 target's new content must never sit in a more readable temp file."""
+        import stat
+        from praisonai.code.utils import file_utils
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            path = os.path.join(temp_dir, ".env")
+            with open(path, "w") as f:
+                f.write("OLD=1")
+            os.chmod(path, 0o600)
+
+            seen = []
+            real_fdopen = os.fdopen
+
+            def spy_fdopen(fd, *a, **kw):
+                seen.append(stat.S_IMODE(os.fstat(fd).st_mode))
+                return real_fdopen(fd, *a, **kw)
+
+            monkeypatch.setattr(file_utils.os, "fdopen", spy_fdopen)
+            file_utils.atomic_write(path, "NEW=2")
+
+            assert seen and all(m & 0o077 == 0 for m in seen), seen
+            assert stat.S_IMODE(os.stat(path).st_mode) == 0o600
+            with open(path) as f:
+                assert f.read() == "NEW=2"
+
+    @pytest.mark.skipif(os.name == "nt" or (hasattr(os, "geteuid") and os.geteuid() == 0),
+                        reason="POSIX permissions; root bypasses them")
+    def test_atomic_write_refuses_read_only_target(self):
+        """Like open(path, 'w'), a read-only file must not be replaced via inode swap."""
+        from praisonai.code.utils.file_utils import atomic_write
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            path = os.path.join(temp_dir, "locked.txt")
+            with open(path, "w") as f:
+                f.write("orig")
+            os.chmod(path, 0o444)
+            try:
+                with pytest.raises(PermissionError):
+                    atomic_write(path, "new")
+                with open(path) as f:
+                    assert f.read() == "orig"
+                assert os.listdir(temp_dir) == ["locked.txt"]
+            finally:
+                os.chmod(path, 0o644)
+
     def test_multiple_operations(self):
         """Test multiple search/replace operations."""
         from praisonai.code.tools.search_replace import search_replace