fix: three high-severity audit bugs (#21, #22, #23)

clipsync/clipboard.py

@@ -51,6 +51,16 @@
 
 _PNG_HEADER = b"\x89PNG\r\n\x1a\n"
 
+
+class EncryptedPayloadError(RuntimeError):
+    """Refused to overwrite a ciphertext payload we cannot decrypt.
+
+    Raised when the shared file holds a CSENC payload this process cannot
+    read (no passphrase, wrong passphrase, or a payload version written by a
+    newer build). Overwriting it would destroy the peer's data.
+    """
+
+
 # Sentinel pushed onto the XFixes queue by stop() to unblock the OUT loop.
 _STOP_SENTINEL = object()
 
@@ -564,9 +574,27 @@ def _read_file(self) -> str | None:
             log.warning("Clipboard file is not valid UTF-8 and not encrypted; ignoring")
             return None
 
+    def _refuse_if_unreadable_ciphertext(self, path: Path) -> None:
+        """Raise EncryptedPayloadError if *path* holds a CSENC payload that this
+        process cannot decrypt. Overwriting it would destroy a peer's data."""
+        if not path.exists():
+            return
+        try:
+            data = path.read_bytes()
+        except OSError:
+            return
+        if not is_encrypted(data):
+            return
+        passphrase = self._passphrase()
+        if not passphrase:
+            raise EncryptedPayloadError(path)
+        if decrypt(data, passphrase) is None:
+            raise EncryptedPayloadError(path)
+
     def _write_file(self, text: str) -> None:
         """Atomic write of the shared file, encrypting if a passphrase is set."""
         path = self.clipboard_file
+        self._refuse_if_unreadable_ciphertext(path)
         path.parent.mkdir(parents=True, exist_ok=True)
         passphrase = self._passphrase()
         encoded = text.encode("utf-8")
@@ -620,6 +648,7 @@ def _read_image_file(self) -> bytes | None:
     def _write_image_file(self, png_bytes: bytes) -> None:
         """Atomic write of the shared image file, encrypting if a passphrase is set."""
         path = self.clipboard_image_file
+        self._refuse_if_unreadable_ciphertext(path)
         path.parent.mkdir(parents=True, exist_ok=True)
         passphrase = self._passphrase()
         payload = encrypt(png_bytes, passphrase) if passphrase else png_bytes
@@ -653,13 +682,34 @@ def _seed_from_file(self) -> None:
                     with self._lock:
                         self._last_synced = image
                     return
+                self._warn_if_locked_ciphertext(img_path)
             content = self._read_file()
             if content is not None:
                 with self._lock:
                     self._last_synced = content
+            else:
+                self._warn_if_locked_ciphertext(text_path)
         except Exception:
             log.warning("Could not read existing clipboard file")
 
+    def _warn_if_locked_ciphertext(self, path: Path) -> None:
+        """Log once if *path* holds a CSENC payload we cannot read.
+
+        Outbound sync to this file is refused while that is true (see
+        _refuse_if_unreadable_ciphertext), so say so plainly rather than
+        failing silently.
+        """
+        try:
+            if not path.exists() or not is_encrypted(path.read_bytes()):
+                return
+        except OSError:
+            return
+        log.warning(
+            "%s holds an encrypted payload this build cannot decrypt; outbound sync "
+            "for it is paused until a matching passphrase is configured",
+            path.name,
+        )
+
     def _is_paused(self) -> bool:
         return bool(self._settings.get("sync_paused"))
 
@@ -792,10 +842,18 @@ def _out_tick(self) -> None:
             with self._lock:
                 if image == self._last_synced:
                     return
+                previous_last_synced = self._last_synced
                 self._last_synced = image
             try:
                 self._write_image_file(image)
                 log.info("OUT [%s]: %d bytes image written", _HOSTNAME, len(image))
+            except EncryptedPayloadError:
+                with self._lock:
+                    self._last_synced = previous_last_synced
+                reason = "Refusing to overwrite encrypted clipboard image file (cannot decrypt)"
+                if reason != self._last_decrypt_error:
+                    log.warning("OUT [%s]: %s", _HOSTNAME, reason)
+                    self._last_decrypt_error = reason
             except OSError:
                 log.exception("OUT [%s]: Failed to write image file", _HOSTNAME)
             return
@@ -806,11 +864,19 @@ def _out_tick(self) -> None:
         with self._lock:
             if current == self._last_synced:
                 return
+            previous_last_synced = self._last_synced
             self._last_synced = current
         try:
             self._write_file(current)
             log.info("OUT [%s]: %d chars written", _HOSTNAME, len(current))
             self._history.add_entry(current, "local")
+        except EncryptedPayloadError:
+            with self._lock:
+                self._last_synced = previous_last_synced
+            reason = "Refusing to overwrite encrypted clipboard file (cannot decrypt)"
+            if reason != self._last_decrypt_error:
+                log.warning("OUT [%s]: %s", _HOSTNAME, reason)
+                self._last_decrypt_error = reason
         except OSError:
             log.exception("OUT [%s]: Failed to write clipboard file", _HOSTNAME)
 

clipsync/config.py

@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import contextlib
 import json
 import logging
 import os
@@ -146,10 +147,16 @@ def _load(self) -> None:
 
     def _persist_locked(self) -> None:
         self._path.parent.mkdir(parents=True, exist_ok=True)
-        tmp = self._path.with_suffix(".json.tmp")
-        with tmp.open("w", encoding="utf-8") as fh:
-            json.dump(self._data, fh, indent=2)
-        os.replace(tmp, self._path)
+        tmp = self._path.with_name(f"{self._path.name}.{os.getpid()}.tmp")
+        try:
+            with tmp.open("w", encoding="utf-8") as fh:
+                json.dump(self._data, fh, indent=2)
+            set_file_permissions(tmp)
+            os.replace(tmp, self._path)
+        finally:
+            if tmp.exists():
+                with contextlib.suppress(OSError):
+                    tmp.unlink(missing_ok=True)
         set_file_permissions(self._path)
         try:
             self._mtime_ns = self._path.stat().st_mtime_ns

clipsync/crypto.py

@@ -27,9 +27,15 @@
 
 log = logging.getLogger(__name__)
 
-_ENC_MAGIC_V0: Final = b"CSENC\x00"
-_ENC_MAGIC_V1: Final = b"CSENC\x01"
-_ENC_MAGIC_V2: Final = b"CSENC\x02"
+# Version-agnostic prefix. is_encrypted() matches on this rather than on the
+# known magics so that a payload written by a NEWER build is still recognized
+# as ciphertext by an older one. decrypt() will return None for a version it
+# does not know, and the caller must then refuse to overwrite the file rather
+# than treating it as plaintext and destroying the peer's data.
+_ENC_MAGIC_PREFIX: Final = b"CSENC"
+_ENC_MAGIC_V0: Final = _ENC_MAGIC_PREFIX + b"\x00"
+_ENC_MAGIC_V1: Final = _ENC_MAGIC_PREFIX + b"\x01"
+_ENC_MAGIC_V2: Final = _ENC_MAGIC_PREFIX + b"\x02"
 _SALT_LEN: Final = 16
 _LEGACY_SALT: Final = b"clipsync-v1-salt"
 
@@ -100,5 +106,11 @@ def decrypt(data: bytes, passphrase: str) -> bytes | None:
 
 
 def is_encrypted(data: bytes) -> bool:
-    """Return True if *data* starts with a recognized encryption header."""
-    return data.startswith(_ENC_MAGIC_V0) or data.startswith(_ENC_MAGIC_V1) or data.startswith(_ENC_MAGIC_V2)
+    """Return True if *data* looks like a CSENC payload of ANY version.
+
+    Deliberately matches the version-agnostic prefix, including versions this
+    build cannot decrypt. Callers use this to decide whether a file is safe to
+    overwrite, and an unknown future version is exactly the case where it is
+    not.
+    """
+    return data.startswith(_ENC_MAGIC_PREFIX)

clipsync/history.py

@@ -10,8 +10,10 @@
 
 from __future__ import annotations
 
+import contextlib
 import json
 import logging
+import os
 import threading
 import time
 from dataclasses import dataclass
@@ -116,10 +118,11 @@ def _load(self) -> None:
         except (KeyError, ValueError) as exc:
             log.warning("Failed to load clipboard history: %s", exc)
 
-    def _persist(self) -> None:
+    def _persist_locked(self) -> None:
+        """Persist history to disk. Caller MUST already hold ``self._lock``."""
         if not self._enabled and len(self._entries) == 0:
             return
-        tmp = self._path.with_suffix(".json.tmp")
+        tmp = self._path.with_name(f"{self._path.name}.{os.getpid()}.tmp")
         try:
             self._path.parent.mkdir(parents=True, exist_ok=True)
             payload = json.dumps({"entries": [e.to_dict() for e in self._entries]}, indent=2).encode("utf-8")
@@ -131,6 +134,10 @@ def _persist(self) -> None:
             config.set_file_permissions(self._path)
         except OSError as exc:
             log.warning("Failed to persist clipboard history: %s", exc)
+        finally:
+            if tmp.exists():
+                with contextlib.suppress(OSError):
+                    tmp.unlink(missing_ok=True)
 
     def add_entry(self, text: str, source: str = "local") -> None:
         if not self._enabled or not text:
@@ -144,7 +151,7 @@ def add_entry(self, text: str, source: str = "local") -> None:
             while len(self._entries) > self._max_items:
                 self._entries.pop(0)
             self._prune_old()
-        self._persist()
+            self._persist_locked()
 
     def get_entries(self) -> list[HistoryEntry]:
         with self._lock:
@@ -153,7 +160,7 @@ def get_entries(self) -> list[HistoryEntry]:
     def clear(self) -> None:
         with self._lock:
             self._entries.clear()
-        self._persist()
+            self._persist_locked()
 
     def get_max_items(self) -> int:
         return self._max_items
@@ -164,7 +171,7 @@ def set_max_items(self, value: int) -> None:
                 self._max_items = value
                 while len(self._entries) > self._max_items:
                     self._entries.pop(0)
-            self._persist()
+                self._persist_locked()
 
     def is_enabled(self) -> bool:
         return self._enabled

tests/test_encrypted_overwrite.py

@@ -0,0 +1,112 @@
+"""Guard against destroying a peer's ciphertext (issue #21).
+
+These are synchronous unit tests: no start(), no Observer, no threads. They
+poke _write_file / _write_image_file / _out_tick directly.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from clipsync import config, crypto
+from clipsync.clipboard import ClipboardSync, EncryptedPayloadError
+
+
+@pytest.fixture(autouse=True)
+def _isolate_history(tmp_path, monkeypatch):
+    """ClipboardSync builds a ClipboardHistory bound to config.HISTORY_FILE at
+    import-time module scope. Without this the tests read and rewrite the
+    developer's real clipboard history."""
+    monkeypatch.setattr(config, "HISTORY_FILE", tmp_path / "clipsync_history.json")
+
+
+def _make_sync(tmp_path, passphrase=None):
+    sync_folder = tmp_path / "sync"
+    sync_folder.mkdir(parents=True, exist_ok=True)
+    settings = config.Settings(path=tmp_path / "settings.json")
+    settings.set("sync_folder", str(sync_folder))
+    if passphrase is not None:
+        settings.set("encryption_passphrase", passphrase)
+    return ClipboardSync(settings)
+
+
+def test_write_file_refuses_ciphertext_without_passphrase(tmp_path):
+    sync = _make_sync(tmp_path)
+    ciphertext = crypto.encrypt(b"peer text", "peer-secret")
+    sync.clipboard_file.write_bytes(ciphertext)
+    original = sync.clipboard_file.read_bytes()
+
+    with pytest.raises(EncryptedPayloadError):
+        sync._write_file("local text")
+
+    assert sync.clipboard_file.read_bytes() == original
+
+
+def test_write_image_file_refuses_ciphertext_without_passphrase(tmp_path):
+    sync = _make_sync(tmp_path)
+    ciphertext = crypto.encrypt(b"peer image", "peer-secret")
+    sync.clipboard_image_file.write_bytes(ciphertext)
+    original = sync.clipboard_image_file.read_bytes()
+
+    with pytest.raises(EncryptedPayloadError):
+        sync._write_image_file(b"local image")
+
+    assert sync.clipboard_image_file.read_bytes() == original
+
+
+def test_write_file_refuses_ciphertext_with_wrong_passphrase(tmp_path):
+    sync = _make_sync(tmp_path, passphrase="local-secret")
+    ciphertext = crypto.encrypt(b"peer text", "peer-secret")
+    sync.clipboard_file.write_bytes(ciphertext)
+    original = sync.clipboard_file.read_bytes()
+
+    with pytest.raises(EncryptedPayloadError):
+        sync._write_file("local text")
+
+    assert sync.clipboard_file.read_bytes() == original
+
+
+def test_write_file_succeeds_on_decryptable_ciphertext(tmp_path):
+    passphrase = "shared-secret"
+    sync = _make_sync(tmp_path, passphrase=passphrase)
+    ciphertext = crypto.encrypt(b"peer text", passphrase)
+    sync.clipboard_file.write_bytes(ciphertext)
+
+    sync._write_file("local text")
+
+    data = sync.clipboard_file.read_bytes()
+    assert crypto.decrypt(data, passphrase) == b"local text"
+
+
+def test_write_file_succeeds_on_plaintext_or_missing_file(tmp_path):
+    sync = _make_sync(tmp_path)
+    sync._write_file("first text")
+    assert sync.clipboard_file.read_bytes() == b"first text"
+
+    sync._write_file("second text")
+    assert sync.clipboard_file.read_bytes() == b"second text"
+
+
+def test_out_tick_restores_last_synced_on_encrypted_payload_error(tmp_path):
+    sync = _make_sync(tmp_path)
+    ciphertext = crypto.encrypt(b"peer text", "peer-secret")
+    sync.clipboard_file.write_bytes(ciphertext)
+    sync._last_synced = "prior value"
+    sync._read_clipboard = lambda: "local text"
+    sync._read_clipboard_image = lambda: None
+
+    sync._out_tick()
+
+    assert sync._last_synced == "prior value"
+    assert sync.clipboard_file.read_bytes() == ciphertext
+
+
+def test_write_file_refuses_future_ciphertext_version(tmp_path):
+    sync = _make_sync(tmp_path)
+    sync.clipboard_file.write_bytes(b"CSENC\xff")
+    original = sync.clipboard_file.read_bytes()
+
+    with pytest.raises(EncryptedPayloadError):
+        sync._write_file("local text")
+
+    assert sync.clipboard_file.read_bytes() == original