diff --git a/clipsync/_syncthing_hashes.py b/clipsync/_syncthing_hashes.py new file mode 100644 index 0000000..8d69e81 --- /dev/null +++ b/clipsync/_syncthing_hashes.py @@ -0,0 +1,39 @@ +"""SHA-256 hashes of the pinned Syncthing release archives. + +Generated by ``tools/refresh_syncthing_hashes.py``, which fetches +Syncthing's ``sha256sum.txt.asc`` and refuses to emit anything unless the +PGP signature verifies against the pinned release key in +``_release_key.py``. Do not hand-edit: regenerate. + +Why these live in the source tree rather than being fetched at runtime: +verifying the signature at runtime needs gpg on PATH, and most end users +do not have it (stock Windows and macOS ship without it), so that check +silently degraded to hash-only. Pinning moves the trust anchor into code +that goes through review and git history, and removes the network from +the verification path entirely for the pinned version. +""" + +from __future__ import annotations + +SYNCTHING_PINNED_SHA256: dict[str, dict[str, str]] = { + "v2.0.16": { + "syncthing-linux-386-v2.0.16.tar.gz": "407076caead4eec3c1ecacb5e3466958784220b5d646c83f5e178ee85d62b4a7", + "syncthing-linux-amd64-v2.0.16.tar.gz": "d5ca379993844b0e6e4fced05e3ac4a6c4513dee916ab65516c6d07d5e53e317", + "syncthing-linux-arm-v2.0.16.tar.gz": "93a52ffd06b1627810a229c1e26d242cbd267b3d1902ed918819a897f0724237", + "syncthing-linux-arm64-v2.0.16.tar.gz": "0254442af9be1886a7b55f63ad3c24c32fc2e294a3a2efd75a3b8b00335e5bd7", + "syncthing-linux-loong64-v2.0.16.tar.gz": "0d4682079380b3abce25ba588765e40660e39fead6bb0601835d9df389078599", + "syncthing-linux-mips-v2.0.16.tar.gz": "adc3cadf33df284d4d5c91d8edfd4f8aaf3876d0d435c0eea540383a7a29d78a", + "syncthing-linux-mips64-v2.0.16.tar.gz": "6c89cb7ec39a247e0c63e737ec8dd7d3d8f4f81bb80730e5053d0e7d3a0e87f9", + "syncthing-linux-mips64le-v2.0.16.tar.gz": "f5be69b95d5927213d52aee81f069ff16bb965918531e4630cb4ced8cc2b6623", + "syncthing-linux-mipsle-v2.0.16.tar.gz": "855b481ce19d0c25c044511284b3357630e9929c550c8d88e50839f8dc4f4753", + "syncthing-linux-ppc64le-v2.0.16.tar.gz": "0762d74b2a4d1ea167956c393b3ea7e639699eee78446b085a63e412d7aabbc0", + "syncthing-linux-riscv64-v2.0.16.tar.gz": "6d53cb3870e78a1c27dc68e4fe401de072bbbe64b72b3ae39d761511c7bf252a", + "syncthing-linux-s390x-v2.0.16.tar.gz": "cb9a5efb1b07c22f14264cf47f785702f1aad7fb85499b633fc05987d8059bf7", + "syncthing-macos-amd64-v2.0.16.zip": "2b5fe419de35c26354843ae567b2ae5c1bf82b151e3aea3dcfb620ca590999d4", + "syncthing-macos-arm64-v2.0.16.zip": "a90318e32ffab04561c7ececc3be79be3db478dabff3e3e3ee1183e38a4bfa3b", + "syncthing-macos-universal-v2.0.16.zip": "a767bf73035af2cb7755abc23666778a7edbf6f42ef3f319f5804ff87159f8fa", + "syncthing-windows-386-v2.0.16.zip": "c8e17cb290290f78823e0900b46a3b6020a1afbf8760533534f3a6fc8a6e2538", + "syncthing-windows-amd64-v2.0.16.zip": "5b519408c11e69e712702911caa399077e3fc602d8d70d6147f620e67bd83037", + "syncthing-windows-arm64-v2.0.16.zip": "0d0ea5ced8d900ec861628ff8fa4d7ec529c8a662f5183d7c49d5333d0cae71e", + }, +} diff --git a/clipsync/clipboard.py b/clipsync/clipboard.py index 309e548..400f06e 100644 --- a/clipsync/clipboard.py +++ b/clipsync/clipboard.py @@ -46,6 +46,15 @@ log = logging.getLogger(__name__) +# The OUT loop reads the system clipboard while the IN loop writes it, from two +# different threads. The native clipboard is a single shared object on every +# platform and is not thread-safe: on macOS, pyperclip's PyObjC backend (chosen +# over pbcopy/pbpaste whenever AppKit is importable, as it is in the bundled +# app) drives NSPasteboard directly, and a read racing a write segfaults inside +# -[_NSPasteboardOwnersCollection handleOwnershipChange]. Serialize every native +# clipboard touch through this lock. +_CLIPBOARD_LOCK = threading.RLock() + _HOSTNAME = _safe_hostname() @@ -339,6 +348,19 @@ def reply(p: int) -> None: pass +def _truncate_for_log(value: object, limit: int = 40) -> str: + """repr *value* for a log line, truncating long payloads. + + bytes needs this as much as str does: _last_synced holds the whole + clipboard, so once an image is synced it is megabytes of PNG. The + previous guard only tested str, so every heartbeat repr()'d the full + image into the log and churned the rotating handler. + """ + if isinstance(value, str | bytes) and len(value) > limit: + return repr(value[:limit]) + "..." + return repr(value) + + def _normalize_newlines(s: str) -> str: """Collapse CRLF/CR to LF so Windows's clipboard normalization does not look like a real change to the OUT loop after a remote update.""" @@ -715,7 +737,8 @@ def _is_paused(self) -> bool: def _read_clipboard(self) -> str | None: try: - value = pyperclip.paste() + with _CLIPBOARD_LOCK: + value = pyperclip.paste() except Exception as exc: msg = f"{type(exc).__name__}: {exc}" if msg != self._last_read_error: @@ -745,7 +768,8 @@ def _write_clipboard(self, value: str) -> bool: self._last_write_error = msg # fall through to pyperclip try: - pyperclip.copy(value) + with _CLIPBOARD_LOCK: + pyperclip.copy(value) log.debug("clipboard write (pyperclip) (%d chars)", len(value)) except Exception as exc: msg = f"{type(exc).__name__}: {exc}" @@ -760,14 +784,16 @@ def _write_clipboard(self, value: str) -> bool: def _read_clipboard_image(self) -> bytes | None: try: - return _read_image_from_system_clipboard() + with _CLIPBOARD_LOCK: + return _read_image_from_system_clipboard() except Exception as exc: log.debug("Image clipboard read failed: %s", exc) return None def _write_clipboard_image(self, png_bytes: bytes) -> bool: try: - return _write_image_to_system_clipboard(png_bytes) + with _CLIPBOARD_LOCK: + return _write_image_to_system_clipboard(png_bytes) except Exception as exc: log.debug("Image clipboard write failed: %s", exc) return False @@ -831,7 +857,7 @@ def _out_loop(self) -> None: log.debug( "HEARTBEAT (host=%s): last_synced=%s, paused=%s", _HOSTNAME, - (repr(last[:40]) + "...") if isinstance(last, str) and len(last) > 40 else repr(last), + _truncate_for_log(last), self._is_paused(), ) @@ -855,6 +881,11 @@ def _out_tick(self) -> None: log.warning("OUT [%s]: %s", _HOSTNAME, reason) self._last_decrypt_error = reason except OSError: + # Roll back too: _last_synced is the "already sent" guard, so + # leaving it set after a failed write means every later tick + # sees this image as synced and it is never retried. + with self._lock: + self._last_synced = previous_last_synced log.exception("OUT [%s]: Failed to write image file", _HOSTNAME) return @@ -878,6 +909,8 @@ def _out_tick(self) -> None: log.warning("OUT [%s]: %s", _HOSTNAME, reason) self._last_decrypt_error = reason except OSError: + with self._lock: + self._last_synced = previous_last_synced log.exception("OUT [%s]: Failed to write clipboard file", _HOSTNAME) def _in_loop(self) -> None: @@ -967,7 +1000,10 @@ class _ClipboardFileHandler(FileSystemEventHandler): def __init__(self, sync: ClipboardSync) -> None: super().__init__() self._sync = sync - self._debounce_until = 0.0 + # Per-path deadlines. A single shared deadline let a clipboard.txt and + # a clipboard.png update arriving within the debounce window suppress + # each other, so only one of the two was ever applied. + self._debounce_until: dict[str, float] = {} # Fast name-based pre-filter to avoid Path.resolve() on every event. # Syncthing generates many temp-file events; most are irrelevant. self._target_names = {config.CLIPBOARD_FILENAME, config.CLIPBOARD_IMAGE_FILENAME} @@ -988,9 +1024,10 @@ def _dispatch(self, path: str) -> None: if not self._matches(path): return now = time.monotonic() - if now < self._debounce_until: + key = str(Path(path).name) + if now < self._debounce_until.get(key, 0.0): return - self._debounce_until = now + 0.1 + self._debounce_until[key] = now + 0.1 # Non-blocking: hand off to _in_loop so the watchdog thread pool # is never held by clipboard I/O (avoids pool exhaustion on Windows). self._sync._in_queue.put(path) diff --git a/clipsync/config.py b/clipsync/config.py index b4f1372..36fe2d7 100644 --- a/clipsync/config.py +++ b/clipsync/config.py @@ -21,8 +21,40 @@ APP_NAME = "ClipSync" APP_ID = "clipsync" -ACCENT_COLOR = "#1A6B8A" -ACCENT_HOVER = "#145670" +# Legacy accent aliases kept for backwards compatibility. +ACCENT_COLOR = "#5A6BFF" +ACCENT_HOVER = "#4654CC" + +# --------------------------------------------------------------------------- +# Pro theme palette (slate + indigo). These values are used directly by the +# UI module so changes stay centralized here instead of scattered through UI +# constructors. +# --------------------------------------------------------------------------- + +COLOR_PRIMARY = "#5A6BFF" # indigo action accent +COLOR_PRIMARY_HOVER = "#4654CC" +COLOR_PRIMARY_MUTED = (228, 231, 255) # light-mode card tint, RGB tuple + +COLOR_SUCCESS = "#2DD36F" +COLOR_DANGER = "#FF4D4D" +COLOR_DANGER_HOVER = "#CC3D3D" +COLOR_WARNING = "#FFB020" + +# Light mode +COLOR_BG_LIGHT = "#F5F6F8" +COLOR_CARD_LIGHT = "#FFFFFF" +COLOR_TEXT_LIGHT = "#11131A" +COLOR_TEXT_MUTED_LIGHT = "#6B7280" +COLOR_BORDER_LIGHT = "#E2E4E9" +COLOR_ROW_BG_LIGHT = "#F0F1F5" + +# Dark mode +COLOR_BG_DARK = "#0F1117" +COLOR_CARD_DARK = "#181A21" +COLOR_TEXT_DARK = "#F0F1F5" +COLOR_TEXT_MUTED_DARK = "#8B92A5" +COLOR_BORDER_DARK = "#2A2D38" +COLOR_ROW_BG_DARK = "#1E212B" SYNCTHING_VERSION = "v2.0.16" SYNCTHING_API_HOST = "127.0.0.1" @@ -85,6 +117,13 @@ def assets_dir() -> Path: "history_max_items": 50, "history_auto_clear_minutes": 0, "theme": "System", + # Mirror this device's log into the shared folder so peers can see it. + # Off by default: it is a debugging aid, and the sync folder is replicated + # to every paired device, so leaving it on ships your log (hostnames, + # device IDs, file names, error traces) to all of them forever. No + # clipboard text is ever logged, but none of that is obvious from the + # tray, and it was previously always on with no way to turn it off. + "debug_log_mirror": False, } HISTORY_FILE = APP_DATA_DIR / "clipsync_history.json" diff --git a/clipsync/crypto.py b/clipsync/crypto.py index 85e8c16..9ac2145 100644 --- a/clipsync/crypto.py +++ b/clipsync/crypto.py @@ -19,6 +19,7 @@ import base64 import logging import os +from pathlib import Path from typing import Final from cryptography.fernet import Fernet, InvalidToken @@ -105,6 +106,112 @@ def decrypt(data: bytes, passphrase: str) -> bytes | None: return None +# --------------------------------------------------------------------------- +# Streaming file encryption +# --------------------------------------------------------------------------- +# +# Fernet holds the whole payload in memory, so the clipboard helpers above are +# unusable for file transfer: a multi-GB send would exhaust RAM. Files are +# therefore written as a sequence of independently-authenticated chunks. +# +# Layout: +# CSENCF\x01 magic + version (distinct from the CSENC clipboard +# payloads so the two can never be confused) +# salt 16 random bytes, PBKDF2 as v2 (600k iterations) +# repeated: +# 4-byte big-endian token length +# Fernet token over (8-byte big-endian chunk index || chunk bytes) +# +# The index is inside the authenticated plaintext, so chunks cannot be +# reordered or dropped without detection. The stream ends with an +# empty-payload chunk acting as an EOF marker, so truncation is caught too: +# without it, a cut-short file would otherwise decrypt cleanly to a prefix. +_FILE_MAGIC: Final = b"CSENCF\x01" +_FILE_CHUNK_SIZE: Final = 1024 * 1024 + + +class StreamDecryptError(Exception): + """Raised when an encrypted file cannot be decrypted or fails integrity.""" + + +def encrypt_file(src: Path, dst: Path, passphrase: str) -> None: + """Encrypt *src* into *dst* in bounded memory.""" + salt = os.urandom(_SALT_LEN) + fernet = Fernet(_derive_key(passphrase, salt, _PBKDF2_ITERATIONS_V2)) + with open(src, "rb") as fin, open(dst, "wb") as fout: + fout.write(_FILE_MAGIC) + fout.write(salt) + index = 0 + while True: + chunk = fin.read(_FILE_CHUNK_SIZE) + if not chunk: + break + token = fernet.encrypt(index.to_bytes(8, "big") + chunk) + fout.write(len(token).to_bytes(4, "big")) + fout.write(token) + index += 1 + # EOF marker: an authenticated empty chunk at the next index. + token = fernet.encrypt(index.to_bytes(8, "big")) + fout.write(len(token).to_bytes(4, "big")) + fout.write(token) + + +def decrypt_file(src: Path, dst: Path, passphrase: str) -> None: + """Decrypt *src* into *dst*. Raises StreamDecryptError on any failure. + + *dst* is removed on failure so a partial plaintext is never left behind + for the user to mistake for a complete file. + """ + try: + with open(src, "rb") as fin, open(dst, "wb") as fout: + if fin.read(len(_FILE_MAGIC)) != _FILE_MAGIC: + raise StreamDecryptError("not a clipsync encrypted file") + salt = fin.read(_SALT_LEN) + if len(salt) != _SALT_LEN: + raise StreamDecryptError("truncated header") + fernet = Fernet(_derive_key(passphrase, salt, _PBKDF2_ITERATIONS_V2)) + expected = 0 + saw_eof = False + while True: + raw_len = fin.read(4) + if not raw_len: + break + if len(raw_len) != 4: + raise StreamDecryptError("truncated chunk length") + token = fin.read(int.from_bytes(raw_len, "big")) + try: + plain = fernet.decrypt(token) + except (InvalidToken, ValueError) as exc: + raise StreamDecryptError("wrong passphrase or corrupt data") from exc + if len(plain) < 8: + raise StreamDecryptError("corrupt chunk") + if int.from_bytes(plain[:8], "big") != expected: + raise StreamDecryptError("chunks reordered or dropped") + body = plain[8:] + if not body: + saw_eof = True + break + fout.write(body) + expected += 1 + if not saw_eof: + raise StreamDecryptError("file is truncated") + except StreamDecryptError: + dst.unlink(missing_ok=True) + raise + except OSError: + dst.unlink(missing_ok=True) + raise + + +def is_encrypted_file(path: Path) -> bool: + """True if *path* begins with the streaming-file magic.""" + try: + with open(path, "rb") as fh: + return fh.read(len(_FILE_MAGIC)) == _FILE_MAGIC + except OSError: + return False + + def is_encrypted(data: bytes) -> bool: """Return True if *data* looks like a CSENC payload of ANY version. diff --git a/clipsync/debug.py b/clipsync/debug.py index f0f0e94..bcd20c6 100644 --- a/clipsync/debug.py +++ b/clipsync/debug.py @@ -52,12 +52,23 @@ def __init__(self, settings: config.Settings) -> None: self._stop = threading.Event() self._thread: threading.Thread | None = None self._hostname = _safe_hostname() + # Cleared whenever mirroring is on, so switching off retracts what is + # published. Starts False so the first disabled tick also cleans up a + # file left behind by an older build, where mirroring was always on + # and there was no way to turn it off. + self._retracted = False def start(self) -> None: self._stop.clear() self._thread = threading.Thread(target=self._loop, name="clipsync-logmirror", daemon=True) self._thread.start() - log.info("LogMirror started (host=%s)", self._hostname) + # The thread self-gates on the setting each tick; it stays running so + # the toggle applies live. + log.info( + "LogMirror thread started (host=%s, mirroring=%s)", + self._hostname, + self._enabled(), + ) def stop(self) -> None: self._stop.set() @@ -74,7 +85,33 @@ def _loop(self) -> None: if self._stop.wait(_MIRROR_INTERVAL_SEC): return + def _enabled(self) -> bool: + return bool(self._settings.get("debug_log_mirror", False)) + + def _remove_own_mirror(self) -> None: + """Delete this host's mirrored log from the shared folder. + + Turning the mirror off has to retract what is already published: + otherwise the last copy sits in the synced folder and keeps being + replicated to every peer indefinitely. + """ + dest = _debug_dir(self._settings) / f"{self._hostname}.log" + try: + if dest.exists(): + dest.unlink() + log.info("Log mirror disabled; removed %s from the shared folder", dest.name) + except OSError: + log.debug("Could not remove mirrored log", exc_info=True) + def _tick(self) -> None: + if not self._enabled(): + # Checked every tick rather than at start, so toggling the setting + # takes effect without a restart. + if not self._retracted: + self._retracted = True + self._remove_own_mirror() + return + self._retracted = False src = config.LOG_FILE if not src.exists(): return diff --git a/clipsync/file_transfer.py b/clipsync/file_transfer.py index 87fc4e7..fb28661 100644 --- a/clipsync/file_transfer.py +++ b/clipsync/file_transfer.py @@ -15,6 +15,7 @@ import logging import os import shutil +import threading import time from collections.abc import Callable from pathlib import Path @@ -24,11 +25,16 @@ from watchdog.observers.api import BaseObserver from . import config +from .crypto import encrypt_file from .debug import _safe_hostname log = logging.getLogger(__name__) _HOSTNAME = _safe_hostname() +# Marks a file in the shared folder as encrypted. Stripped when writing the +# plaintext out to the receiver's Downloads folder. +ENCRYPTED_SUFFIX = ".csenc" + class FileTransfer: """Send files to the sync folder and notify on incoming files from peers.""" @@ -47,17 +53,46 @@ def files_dir(self) -> Path: folder = Path(self._settings.get("sync_folder") or config.SYNC_FOLDER) return folder / "files" + def _passphrase(self) -> str: + val = self._settings.get("encryption_passphrase") or "" + return val if isinstance(val, str) else "" + def send(self, source: Path) -> Path: """Copy *source* into the shared folder under this host's subdirectory. - Returns the destination path. Raises OSError on failure. + When a passphrase is configured the file is encrypted on the way in and + gains a ``.csenc`` suffix. Previously it was copied verbatim, so + enabling at-rest encryption protected the clipboard but left every sent + file sitting in the synced folder as plaintext -- readable by anything + with access to that directory, and replicated that way to each peer. + + Returns the destination path. Raises OSError on failure. """ dest_dir = self.files_dir / _HOSTNAME dest_dir.mkdir(parents=True, exist_ok=True) timestamp = time.strftime("%Y%m%d_%H%M%S") + passphrase = self._passphrase() + size = source.stat().st_size + + if passphrase: + dest = dest_dir / f"{timestamp}_{source.name}{ENCRYPTED_SUFFIX}" + tmp = dest.with_name(dest.name + ".part") + try: + encrypt_file(source, tmp, passphrase) + config.set_file_permissions(tmp) + tmp.replace(dest) + except BaseException: + tmp.unlink(missing_ok=True) + raise + log.info("FILE OUT [%s]: %s (%d bytes, encrypted)", _HOSTNAME, source.name, size) + return dest + dest = dest_dir / f"{timestamp}_{source.name}" shutil.copy2(source, dest) - log.info("FILE OUT [%s]: %s (%d bytes)", _HOSTNAME, source.name, source.stat().st_size) + # copy2 preserves the source mode, so a world-readable original stayed + # world-readable inside the shared folder. + config.set_file_permissions(dest) + log.info("FILE OUT [%s]: %s (%d bytes)", _HOSTNAME, source.name, size) return dest def start(self) -> None: @@ -84,7 +119,11 @@ def __init__(self, on_received: Callable[[Path, str], None]) -> None: self._on_received = on_received # Guard against duplicate events (watchdog can fire multiple times for # a single file, e.g. created + modified during Syncthing's atomic write). + # watchdog dispatches from a thread pool on Windows, so the + # check-then-add below has to be atomic or two events for the same + # file can both pass it and deliver the file twice. self._seen: set[str] = set() + self._seen_lock = threading.Lock() def _handle(self, path: Path) -> None: # Expected layout: files// @@ -96,9 +135,10 @@ def _handle(self, path: Path) -> None: if path.name.startswith(".syncthing.") and path.name.endswith(".tmp"): return key = str(path) - if key in self._seen: - return - self._seen.add(key) + with self._seen_lock: + if key in self._seen: + return + self._seen.add(key) log.info("FILE IN [%s]: %s from %s", _HOSTNAME, path.name, sender) try: self._on_received(path, sender) diff --git a/clipsync/history.py b/clipsync/history.py index 626d1b7..4f2ad05 100644 --- a/clipsync/history.py +++ b/clipsync/history.py @@ -56,8 +56,15 @@ def __init__(self, settings: config.Settings | None = None) -> None: self._lock = threading.RLock() self._entries: list[HistoryEntry] = [] self._settings = settings - self._max_items: int = 50 if settings is None else int(settings.get("history_max_items", 50) or 50) - self._enabled: bool = True if settings is None else bool(settings.get("history_enabled", True)) + # Parsed defensively: a malformed value in settings.json used to raise + # straight out of __init__, and ClipboardHistory is built during + # startup, so the whole app died before the tray appeared. A bad value + # should cost the default, not the application. + self._max_items: int = _coerce_int(None if settings is None else settings.get("history_max_items", 50), 50) + self._enabled: bool = _coerce_bool(None if settings is None else settings.get("history_enabled", True), True) + # Set when the on-disk file holds data we could not read and could not + # move aside. Persisting would destroy it, so we stay in memory only. + self._readonly: bool = False self._load() def _passphrase(self) -> str: @@ -69,8 +76,10 @@ def _passphrase(self) -> str: def _auto_clear_minutes(self) -> int: if self._settings is None: return 0 - val = self._settings.get("history_auto_clear_minutes") - return int(val) if isinstance(val, int) and val > 0 else 0 + # 0 means "never auto-clear", so an unparseable value must fall back to + # 0 rather than silently keeping sensitive entries forever under the + # user's belief that they expire. A stringified "30" is accepted. + return _coerce_int(self._settings.get("history_auto_clear_minutes"), 0) def _prune_old(self) -> None: minutes = self._auto_clear_minutes() @@ -79,23 +88,58 @@ def _prune_old(self) -> None: cutoff = time.time() - (minutes * 60) self._entries = [e for e in self._entries if e.timestamp > cutoff] + def _quarantine_unreadable(self, reason: str) -> None: + """Move an undecryptable history file aside instead of overwriting it. + + Without this, an unreadable file was simply left in place with an + empty in-memory list, and the very next add_entry() persisted that + list straight over it — silently destroying every stored entry. The + sync file is protected from exactly this by + ClipboardSync._refuse_if_unreadable_ciphertext; the history file was + not. Renaming keeps the ciphertext recoverable if the passphrase + turns up, while letting history keep working from now on. + """ + stamp = time.strftime("%Y%m%d-%H%M%S") + backup = self._path.with_name(f"{self._path.stem}.unreadable-{stamp}{self._path.suffix}") + try: + self._path.replace(backup) + except OSError as exc: + # Could not move it aside, so we must not overwrite it either. + self._readonly = True + log.error( + "Clipboard history is unreadable (%s) and could not be moved aside (%s); " + "history will not be written this session so the existing file is preserved", + reason, + exc, + ) + return + log.warning( + "Clipboard history could not be read (%s). The old file was kept as %s " + "and a fresh history started. If you recover the passphrase, that file " + "can still be decrypted.", + reason, + backup.name, + ) + def _load(self) -> None: if not self._path.exists(): return try: raw = self._path.read_bytes() except OSError as exc: + # Unread, so its contents are unknown: refuse to overwrite it. + self._readonly = True log.warning("Failed to read clipboard history: %s", exc) return if is_encrypted(raw): passphrase = self._passphrase() if not passphrase: - log.warning("History file is encrypted but no passphrase is configured") + self._quarantine_unreadable("encrypted but no passphrase is configured") return decrypted = decrypt(raw, passphrase) if decrypted is None: - log.warning("Failed to decrypt clipboard history (passphrase mismatch?)") + self._quarantine_unreadable("passphrase mismatch, or written by a newer build") return try: data = json.loads(decrypted.decode("utf-8")) @@ -122,6 +166,10 @@ 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 + if self._readonly: + # The file holds data we could not read; overwriting it would + # destroy it. Keep this session's entries in memory only. + return tmp = self._path.with_name(f"{self._path.name}.{os.getpid()}.tmp") try: self._path.parent.mkdir(parents=True, exist_ok=True) @@ -130,8 +178,11 @@ def _persist_locked(self) -> None: if passphrase: payload = encrypt(payload, passphrase) tmp.write_bytes(payload) + # Tighten before the rename, not after: this file holds clipboard + # text. Chmod'ing the final name afterwards leaves a window where + # it is readable by other local users under its real name. + config.set_file_permissions(tmp) tmp.replace(self._path) - config.set_file_permissions(self._path) except OSError as exc: log.warning("Failed to persist clipboard history: %s", exc) finally: @@ -184,5 +235,42 @@ def count(self) -> int: return len(self._entries) +def _coerce_int(value: object, default: int) -> int: + """Best-effort int, falling back to *default* rather than raising. + + Accepts the numeric strings a hand-edited settings.json can easily end up + holding ("50"), and refuses values that are not positive. + """ + if isinstance(value, bool): # bool is an int subclass; not a count + return default + if isinstance(value, int): + return value if value > 0 else default + if isinstance(value, str): + try: + parsed = int(value.strip()) + except ValueError: + return default + return parsed if parsed > 0 else default + return default + + +def _coerce_bool(value: object, default: bool) -> bool: + """Best-effort bool. ``bool("false")`` is True, which is exactly the trap + a JSON-stringified setting falls into, so strings are matched explicitly. + """ + if isinstance(value, bool): + return value + if isinstance(value, str): + lowered = value.strip().lower() + if lowered in {"true", "1", "yes", "on"}: + return True + if lowered in {"false", "0", "no", "off", ""}: + return False + return default + if isinstance(value, int): + return bool(value) + return default + + def _normalize(s: str) -> str: return s.replace("\r\n", "\n").replace("\r", "\n") if isinstance(s, str) else "" diff --git a/clipsync/main.py b/clipsync/main.py index 2ab70f2..62c52b6 100644 --- a/clipsync/main.py +++ b/clipsync/main.py @@ -30,8 +30,9 @@ from . import config, update from .clipboard import ClipboardSync +from .crypto import StreamDecryptError, decrypt_file, is_encrypted_file from .debug import LogMirror -from .file_transfer import FileTransfer +from .file_transfer import ENCRYPTED_SUFFIX, FileTransfer from .pairing import PendingDeviceWatcher, accept_pending_device from .single_instance import AlreadyRunning, SingleInstance from .syncthing import SyncthingError, SyncthingService @@ -416,11 +417,43 @@ def _on_device_accepted(self, device_id: str) -> None: self._notify("Device connected", f"Now syncing clipboard with {device_id[:7]}") def _on_folder_changed(self, new_path: str) -> None: + """Repoint every consumer of the sync folder, not just the clipboard. + + Syncthing reads the folder path once, when prepare_home() patches + config.xml, and FileTransfer's observer is scheduled on the old + directory at construction. Restarting only ClipboardSync left both + pointing at the previous folder: clipboard.txt was written where no + peer was replicating it, so sync silently stopped working. + """ Path(new_path).mkdir(parents=True, exist_ok=True) + # The settings UI runs in its own process and persists this before + # emitting the event, and Settings reloads on mtime change, so this is + # normally a no-op. Set it anyway: prepare_home() below reads the + # folder back out of settings, and that read must not depend on + # another process having already written it. + self.settings.set("sync_folder", new_path) + if self.clipboard is not None: self.clipboard.stop() - self.clipboard = ClipboardSync(self.settings) - self.clipboard.start() + self.clipboard = None + if self.file_transfer is not None: + self.file_transfer.stop() + self.file_transfer = None + + # Re-patches config.xml with the new folder path, then restarts the + # daemon so it actually picks the change up. prepare_home() is + # idempotent: it regenerates nothing when a home already exists. + try: + self.syncthing.stop() + self._start_syncthing_with_retry() + except Exception: + log.exception("Failed to restart Syncthing for new sync folder %s", new_path) + + self.clipboard = ClipboardSync(self.settings) + self.clipboard.start() + self.file_transfer = FileTransfer(self.settings, on_received=self._on_file_received) + self.file_transfer.start() + log.info("Sync folder changed to %s; Syncthing, clipboard and file transfer restarted", new_path) def _send_file_worker(self, source: Path) -> None: if self.file_transfer is None: @@ -435,13 +468,32 @@ def _send_file_worker(self, source: Path) -> None: def _on_file_received(self, path: Path, sender: str) -> None: downloads = Path.home() / "Downloads" downloads.mkdir(parents=True, exist_ok=True) - stem = path.stem - suffix = path.suffix + + # Decryption happens here, on the way OUT of the sync folder, and never + # in place: writing plaintext back inside the folder would hand it + # straight to Syncthing and undo the encryption for every peer. + encrypted = is_encrypted_file(path) + display_name = path.name + if encrypted and display_name.endswith(ENCRYPTED_SUFFIX): + display_name = display_name[: -len(ENCRYPTED_SUFFIX)] + passphrase = self.settings.get("encryption_passphrase") or "" + if encrypted and not isinstance(passphrase, str): + passphrase = "" + if encrypted and not passphrase: + log.warning("Received encrypted file %s but no passphrase is configured", path.name) + self._notify( + f"File from {sender}", + f"{display_name} is encrypted and no passphrase is set; not saved.", + ) + return + + stem = Path(display_name).stem + suffix = Path(display_name).suffix # Atomically claim the destination filename with O_EXCL to close # the TOCTOU window between two concurrent receives of same-named # files from different senders: previously the exists() check + # copy2 could race and clobber each other. - dest = downloads / path.name + dest = downloads / display_name fd = -1 attempt = 0 while attempt < 1000: @@ -455,12 +507,30 @@ def _on_file_received(self, path: Path, sender: str) -> None: log.warning("Could not find free filename for received file %s", path.name) return try: - with os.fdopen(fd, "wb") as out, path.open("rb") as src: - shutil.copyfileobj(src, out) - try: - shutil.copystat(path, dest) - except OSError: - pass + if encrypted: + # decrypt_file owns the destination, so release the claim we + # took with O_EXCL while keeping the name reserved on disk. + os.close(fd) + fd = -1 + try: + decrypt_file(path, dest, passphrase) + except StreamDecryptError as exc: + log.warning("Could not decrypt received file %s: %s", path.name, exc) + dest.unlink(missing_ok=True) + self._notify( + f"File from {sender}", + f"{display_name} could not be decrypted ({exc}).", + ) + return + config.set_file_permissions(dest) + else: + with os.fdopen(fd, "wb") as out, path.open("rb") as src: + shutil.copyfileobj(src, out) + fd = -1 + try: + shutil.copystat(path, dest) + except OSError: + pass except OSError: log.exception("Failed to save received file %s", path) try: diff --git a/clipsync/syncthing.py b/clipsync/syncthing.py index 4364d82..06c8aaf 100644 --- a/clipsync/syncthing.py +++ b/clipsync/syncthing.py @@ -35,6 +35,7 @@ from . import config from ._release_key import SYNCTHING_RELEASE_FINGERPRINT, SYNCTHING_RELEASE_KEY +from ._syncthing_hashes import SYNCTHING_PINNED_SHA256 log = logging.getLogger(__name__) @@ -214,14 +215,21 @@ def _fetch_official_sha256sums(version: str) -> dict[str, str]: The signature is verified first (see :func:`_verify_release_signature`); on a signature failure this raises SyncthingError rather than returning hashes, so a tampered sums file can never reach the hash - comparison. Returns an empty dict only if the file cannot be fetched - (network error), in which case archive verification is skipped. + comparison. + + A fetch failure also raises. We are about to execute the archive we + downloaded, so "could not check" has to be fatal: an attacker able to + drop or poison this one request would otherwise get an unverified + binary run, which is precisely the threat the hash check exists to + close. Availability is the cost, and it is the correct trade here. """ try: data = _download(_asc_url(version)) except URLError as exc: - log.warning("Failed to fetch Syncthing sha256sum.txt.asc: %s", exc) - return {} + raise SyncthingError( + f"Failed to fetch Syncthing sha256sum.txt.asc: {exc}. Refusing to " + "extract an unverified Syncthing binary. Check your network and retry." + ) from exc _verify_release_signature(data) return _parse_sha256sums(data) @@ -233,27 +241,52 @@ def _archive_filename(version: str) -> str: return f"syncthing-{os_name}-{arch}-{v}.{ext}" -def _verify_archive_hash(data: bytes, version: str) -> None: - """Raise SyncthingError if *data* (the downloaded archive bytes) does - not match the official Syncthing sha256sum.txt.asc entry for this - platform, or if that sums file's PGP signature is invalid. Falls back - to a logged warning (not an error) only when the sums file cannot be - fetched or the platform entry is absent. - """ - try: - archive_name = _archive_filename(version) - except SyncthingError: - log.warning("Cannot determine platform for archive hash verification; skipping") - return +def _expected_archive_hash(archive_name: str, version: str) -> str: + """Return the trusted SHA-256 for *archive_name*. + + Prefers the manifest pinned in the source tree. Those hashes come from a + sha256sum.txt.asc whose PGP signature was verified against the pinned + release key at release-prep time, by + ``tools/refresh_syncthing_hashes.py``. + + Pinning matters because the runtime signature check needs gpg on PATH, + and stock Windows and macOS do not ship it, so for most users that check + degraded to hash-only and the hash arrived from the same origin as the + download it was meant to vouch for. A pinned manifest moves the trust + anchor into reviewed source and takes the network out of the decision. - sums = _fetch_official_sha256sums(version) - expected = sums.get(archive_name) + The live fetch remains only as a fallback for a version we have not + pinned, which normally means an explicitly requested non-default one. + """ + v = version if version.startswith("v") else f"v{version}" + pinned = SYNCTHING_PINNED_SHA256.get(v, {}).get(archive_name) + if pinned is not None: + log.debug("Using pinned hash for %s", archive_name) + return pinned + + log.warning( + "No pinned hash for %s; falling back to fetching sha256sum.txt.asc. " + "Run tools/refresh_syncthing_hashes.py to pin this version.", + archive_name, + ) + expected = _fetch_official_sha256sums(version).get(archive_name) if expected is None: - log.warning( - "No hash for %s in sha256sum.txt.asc; skipping archive verification", - archive_name, + raise SyncthingError( + f"No hash for {archive_name} in sha256sum.txt.asc. Refusing to extract an unverified Syncthing binary." ) - return + return expected + + +def _verify_archive_hash(data: bytes, version: str) -> None: + """Raise SyncthingError unless *data* (the downloaded archive bytes) + matches the trusted SHA-256 for this platform's release archive. + + Every failure path is fatal. The archive is about to be extracted and + executed, so an unverifiable download must never be trusted; a missing + hash is indistinguishable from one an attacker stripped. + """ + archive_name = _archive_filename(version) + expected = _expected_archive_hash(archive_name, version) import hashlib @@ -283,6 +316,11 @@ def _extract_binary(data: bytes, ext: str, dest_dir: Path) -> Path: members = [m for m in zf.namelist() if m.endswith(f"/{target_name}") or m.endswith(target_name)] if not members: raise SyncthingError("Syncthing binary not found in archive") + # The release archive also ships helper files (e.g. + # etc/firewall-ufw/syncthing) that share the binary's basename. + # The real binary sits directly under the top-level release dir, + # so prefer the shallowest matching path. + members.sort(key=lambda m: m.count("/")) with zf.open(members[0]) as src, target.open("wb") as dst: shutil.copyfileobj(src, dst) else: @@ -292,6 +330,7 @@ def _extract_binary(data: bytes, ext: str, dest_dir: Path) -> Path: ] if not tar_members: raise SyncthingError("Syncthing binary not found in archive") + tar_members.sort(key=lambda m: m.name.count("/")) extracted = tf.extractfile(tar_members[0]) if extracted is None: raise SyncthingError("Failed to extract syncthing binary") @@ -321,24 +360,87 @@ def _binary_version(binary: Path) -> str: return "" +def _file_sha256(path: Path) -> str: + import hashlib + + digest = hashlib.sha256() + with path.open("rb") as fh: + for chunk in iter(lambda: fh.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _binary_digest_path() -> Path: + return config.SYNCTHING_BIN_DIR / "syncthing.sha256" + + +def _record_binary_digest(binary: Path, version: str) -> None: + """Pin the hash of a binary we just extracted from a verified archive.""" + want = version if version.startswith("v") else f"v{version}" + path = _binary_digest_path() + try: + path.write_text(f"{want} {_file_sha256(binary)}\n", encoding="utf-8") + config.set_file_permissions(path) + except OSError: + log.warning("Could not record Syncthing binary digest at %s", path) + + +def _binary_digest_matches(binary: Path, version: str) -> bool: + """True if *binary* still hashes to what we recorded at install time. + + This is deliberately weaker than it looks, and the limit is worth being + explicit about: the digest file lives beside the binary, so anyone who + can overwrite one can overwrite the other. What it does buy is detection + of the cases that actually happen -- Syncthing self-upgrading over its + own binary, a partial or corrupted extraction, and any tampering that + does not also know to rewrite the digest. Checking `--version` alone + buys none of that, since a replaced binary can simply print the string + we want to see. + """ + want = version if version.startswith("v") else f"v{version}" + try: + recorded = _binary_digest_path().read_text(encoding="utf-8").split() + except OSError: + return False + if len(recorded) != 2 or recorded[0] != want: + return False + try: + return recorded[1] == _file_sha256(binary) + except OSError: + return False + + def ensure_binary(version: str = config.SYNCTHING_VERSION) -> Path: """Return the path to a working Syncthing binary at exactly *version*. - Re-downloads if the binary is missing, empty, or was self-upgraded to a + Re-downloads if the binary is missing, empty, was self-upgraded to a different version (Syncthing replaces its own binary on upgrade, which - would otherwise silently drift from the pinned version). + would otherwise silently drift from the pinned version), or no longer + matches the digest recorded when we installed it. + + An offline start with a good binary stays working: the digest check is + local, so no network is touched on the happy path. An offline start with + a bad binary fails, which is intended -- running an unverifiable + Syncthing is worse than not starting. """ binary = config.syncthing_binary_path() + want = version if version.startswith("v") else f"v{version}" if binary.exists() and binary.stat().st_size > 0: on_disk = _binary_version(binary) - want = version if version.startswith("v") else f"v{version}" if on_disk == want: - return binary - log.info( - "Syncthing binary is %s but pinned version is %s; re-downloading", - on_disk, - want, - ) + if _binary_digest_matches(binary, version): + return binary + log.warning( + "Syncthing binary at %s reports %s but does not match the digest recorded at install; re-downloading", + binary, + on_disk, + ) + else: + log.info( + "Syncthing binary is %s but pinned version is %s; re-downloading", + on_disk, + want, + ) _, _, ext = _platform_archive_info() url = _release_asset_url(version) try: @@ -347,6 +449,7 @@ def ensure_binary(version: str = config.SYNCTHING_VERSION) -> Path: raise SyncthingError(f"Failed to download Syncthing: {exc}") from exc _verify_archive_hash(data, version) extracted = _extract_binary(data, ext, config.SYNCTHING_BIN_DIR) + _record_binary_digest(extracted, version) log.info("Installed syncthing binary at %s", extracted) return extracted diff --git a/clipsync/ui.py b/clipsync/ui.py index 8f5abbd..7de5018 100644 --- a/clipsync/ui.py +++ b/clipsync/ui.py @@ -9,11 +9,14 @@ from __future__ import annotations +import base64 +import io import json import logging import subprocess import sys import threading +import tkinter from collections.abc import Callable from pathlib import Path from typing import Any, cast @@ -31,6 +34,58 @@ _WINDOWS = ("pairing", "devices", "settings", "logs", "incoming", "tabbed", "history", "file_picker") +# --------------------------------------------------------------------------- +# Theme helpers +# --------------------------------------------------------------------------- + + +class Theme: + """Resolved light/dark color values for the pro theme.""" + + def __init__(self, mode: str) -> None: + self.set_mode(mode) + + def set_mode(self, mode: str) -> None: + self.mode = mode + is_dark = mode == "Dark" + self.bg = config.COLOR_BG_DARK if is_dark else config.COLOR_BG_LIGHT + self.card = config.COLOR_CARD_DARK if is_dark else config.COLOR_CARD_LIGHT + self.text = config.COLOR_TEXT_DARK if is_dark else config.COLOR_TEXT_LIGHT + self.muted = config.COLOR_TEXT_MUTED_DARK if is_dark else config.COLOR_TEXT_MUTED_LIGHT + self.border = config.COLOR_BORDER_DARK if is_dark else config.COLOR_BORDER_LIGHT + self.row_bg = config.COLOR_ROW_BG_DARK if is_dark else config.COLOR_ROW_BG_LIGHT + + def card_fg(self) -> tuple[str, str] | str: + # CustomTkinter tuple format: (light, dark) + return (config.COLOR_CARD_LIGHT, config.COLOR_CARD_DARK) + + def transparent(self) -> str: + return "transparent" + + +# Global resolved theme; set by _run_child() before any windows are built. +THEME = Theme("System") + + +def _tk_image(img: Image.Image) -> tkinter.PhotoImage: + """Build a Tk image from a PIL image without going through PIL.ImageTk. + + ImageTk needs _imagingtk to resolve Tcl/Tk symbols at runtime, which fails + on interpreters that compile _tkinter directly into the binary (the + python-build-standalone builds uv installs). The failure is a bare + "bad argument type for built-in operation" that kills the whole window + process. Tk 8.6 decodes PNG natively, so round-trip through PNG bytes and + skip Pillow's Tk bridge entirely. + + CTkLabel.configure(image=...) accepts a plain Tk image; it only loses + CTkImage's HiDPI rescaling, and every caller here already renders at a + fixed pixel size. + """ + buf = io.BytesIO() + img.save(buf, format="PNG") + return tkinter.PhotoImage(data=base64.b64encode(buf.getvalue()).decode("ascii")) + + def _center_window(window: ctk.CTkToplevel | ctk.CTk, width: int, height: int) -> None: window.update_idletasks() sw = window.winfo_screenwidth() @@ -40,6 +95,124 @@ def _center_window(window: ctk.CTkToplevel | ctk.CTk, width: int, height: int) - window.geometry(f"{width}x{height}+{x}+{y}") +def _fonts() -> dict[str, ctk.CTkFont]: + """Typography scale used across all windows.""" + return { + "headline": ctk.CTkFont(size=22, weight="bold"), + "title": ctk.CTkFont(size=16, weight="bold"), + "subtitle": ctk.CTkFont(size=12, weight="bold"), + "body": ctk.CTkFont(size=12), + "small": ctk.CTkFont(size=11), + "tiny": ctk.CTkFont(size=10), + "mono": ctk.CTkFont(family="Menlo", size=11), + } + + +def _section_header(parent: ctk.CTkBaseClass, text: str) -> ctk.CTkLabel: + """Bold subsection title with consistent top spacing.""" + return ctk.CTkLabel( + parent, + text=text, + font=_fonts()["subtitle"], + text_color=THEME.text, + anchor="w", + ) + + +def _card_frame(parent: ctk.CTkBaseClass, **kwargs: Any) -> ctk.CTkFrame: + """Rounded card container with the current theme background.""" + defaults: dict[str, Any] = dict( + master=parent, + fg_color=THEME.card_fg(), + border_width=1, + border_color=(config.COLOR_BORDER_LIGHT, config.COLOR_BORDER_DARK), + corner_radius=12, + ) + defaults.update(kwargs) + return ctk.CTkFrame(**defaults) + + +def _primary_button(parent: ctk.CTkBaseClass, text: str, command: Callable[[], None], **kwargs: Any) -> ctk.CTkButton: + """Main call-to-action button (filled indigo).""" + defaults: dict[str, Any] = dict( + master=parent, + text=text, + command=command, + fg_color=config.COLOR_PRIMARY, + hover_color=config.COLOR_PRIMARY_HOVER, + text_color="white", + height=32, + corner_radius=8, + font=_fonts()["body"], + ) + defaults.update(kwargs) + return ctk.CTkButton(**defaults) + + +def _secondary_button(parent: ctk.CTkBaseClass, text: str, command: Callable[[], None], **kwargs: Any) -> ctk.CTkButton: + """Secondary outline button on the card background.""" + defaults: dict[str, Any] = dict( + master=parent, + text=text, + command=command, + fg_color="transparent", + hover_color=(config.COLOR_ROW_BG_LIGHT, config.COLOR_ROW_BG_DARK), + text_color=THEME.text, + border_width=1, + border_color=(config.COLOR_BORDER_LIGHT, config.COLOR_BORDER_DARK), + height=32, + corner_radius=8, + font=_fonts()["body"], + ) + defaults.update(kwargs) + return ctk.CTkButton(**defaults) + + +def _ghost_button(parent: ctk.CTkBaseClass, text: str, command: Callable[[], None], **kwargs: Any) -> ctk.CTkButton: + """Low-emphasis text-like button colored with the primary accent.""" + defaults: dict[str, Any] = dict( + master=parent, + text=text, + command=command, + fg_color="transparent", + hover_color=(config.COLOR_ROW_BG_LIGHT, config.COLOR_ROW_BG_DARK), + text_color=config.COLOR_PRIMARY, + border_width=0, + height=28, + corner_radius=8, + font=_fonts()["small"], + ) + defaults.update(kwargs) + return ctk.CTkButton(**defaults) + + +def _entry(parent: ctk.CTkBaseClass, **kwargs: Any) -> ctk.CTkEntry: + """Standard input with themed border and rounded corners.""" + defaults: dict[str, Any] = dict( + master=parent, + border_width=1, + border_color=(config.COLOR_BORDER_LIGHT, config.COLOR_BORDER_DARK), + corner_radius=8, + font=_fonts()["body"], + ) + defaults.update(kwargs) + return ctk.CTkEntry(**defaults) + + +def _switch(parent: ctk.CTkBaseClass, text: str, variable: ctk.Variable, command: Callable[[], None]) -> ctk.CTkSwitch: + return ctk.CTkSwitch( + parent, + text=text, + variable=variable, + command=command, + progress_color=config.COLOR_PRIMARY, + button_color=config.COLOR_PRIMARY, + button_hover_color=config.COLOR_PRIMARY_HOVER, + text_color=THEME.text, + font=_fonts()["body"], + ) + + # --------------------------------------------------------------------------- # Parent-side controller # --------------------------------------------------------------------------- @@ -221,6 +394,7 @@ def __init__( self.window.resizable(False, False) self.window.protocol("WM_DELETE_WINDOW", self.close) _center_window(self.window, *size) + self.window.configure(fg_color=THEME.bg) self.window.lift() self.window.focus_force() self.window.bind("", lambda _e: self.close()) @@ -276,86 +450,60 @@ def __init__( self._preview_size = (300, 225) self._scan_container = container - ctk.CTkLabel(container, text="Pair a device", font=ctk.CTkFont(size=18, weight="bold")).pack(pady=(0, 8)) + ctk.CTkLabel(container, text="Pair a device", font=_fonts()["headline"], text_color=THEME.text).pack( + pady=(0, 18) + ) - ctk.CTkLabel( + _section_header(container, "Nearby devices").pack(fill="x") + self._nearby_frame = ctk.CTkScrollableFrame( container, - text="Nearby devices on your network", - font=ctk.CTkFont(size=12, weight="bold"), - anchor="w", - ).pack(fill="x") - self._nearby_frame = ctk.CTkScrollableFrame(container, fg_color=("gray90", "gray17"), height=100) - self._nearby_frame.pack(fill="x", pady=(2, 10)) + fg_color=THEME.card_fg(), + border_width=1, + border_color=(config.COLOR_BORDER_LIGHT, config.COLOR_BORDER_DARK), + corner_radius=12, + height=100, + ) + self._nearby_frame.pack(fill="x", pady=(6, 14)) self._nearby_seen: set[str] = set() self._render_nearby([]) self._schedule_nearby_refresh() - ctk.CTkLabel(container, text="Or paste a device ID", font=ctk.CTkFont(size=12, weight="bold"), anchor="w").pack( - fill="x" - ) - entry_row = ctk.CTkFrame(container, fg_color="transparent") - entry_row.pack(fill="x", pady=(2, 8)) - self._entry = ctk.CTkEntry(entry_row, placeholder_text="XXXXXXX-XXXXXXX-…") - self._entry.pack(side="left", fill="x", expand=True, padx=(0, 6)) + _section_header(container, "Or paste a device ID").pack(fill="x", pady=(0, 6)) + entry_row = _card_frame(container) + entry_row.pack(fill="x", pady=(0, 14)) + self._entry = _entry(entry_row, placeholder_text="XXXXXXX-XXXXXXX-…") + self._entry.pack(side="left", fill="x", expand=True, padx=(12, 6), pady=10) self._entry.bind("", lambda _e: self._on_add_clicked()) - ctk.CTkButton( - entry_row, - text="Paste", - width=60, - height=28, - fg_color="transparent", - border_width=1, - border_color=config.ACCENT_COLOR, - text_color=config.ACCENT_COLOR, - hover_color=config.ACCENT_HOVER, - command=self._on_paste_clicked, - ).pack(side="left", padx=(0, 6)) - self._add_btn = ctk.CTkButton( - entry_row, - text="Add", - width=60, - height=28, - fg_color=config.ACCENT_COLOR, - hover_color=config.ACCENT_HOVER, - command=self._on_add_clicked, - ) - self._add_btn.pack(side="left") + _ghost_button(entry_row, text="Paste", command=self._on_paste_clicked, width=60).pack(side="left", padx=(0, 6)) + self._add_btn = _primary_button(entry_row, text="Add", command=self._on_add_clicked, width=60) + self._add_btn.pack(side="left", padx=(0, 12)) - ctk.CTkLabel( - container, text="Your device ID (for the other side)", font=ctk.CTkFont(size=12, weight="bold"), anchor="w" - ).pack(fill="x", pady=(4, 2)) - own_row = ctk.CTkFrame(container, fg_color=("gray90", "gray17")) - own_row.pack(fill="x", pady=(0, 8)) + _section_header(container, "Your device ID").pack(fill="x", pady=(0, 6)) + own_row = _card_frame(container) + own_row.pack(fill="x", pady=(0, 14)) self._qr_label = ctk.CTkLabel(own_row, text="") - self._qr_label.pack(side="left", padx=8, pady=8) + self._qr_label.pack(side="left", padx=14, pady=14) self._render_qr(app.device_id) own_right = ctk.CTkFrame(own_row, fg_color="transparent") - own_right.pack(side="left", fill="both", expand=True, padx=(0, 8), pady=8) + own_right.pack(side="left", fill="both", expand=True, padx=(0, 14), pady=14) ctk.CTkLabel( own_right, text=app.device_id, - font=ctk.CTkFont(size=9), + font=_fonts()["small"], wraplength=180, justify="left", anchor="w", + text_color=THEME.text, ).pack(fill="x") - ctk.CTkButton( + _primary_button( own_right, - text="Copy", - height=28, - fg_color=config.ACCENT_COLOR, - hover_color=config.ACCENT_HOVER, + text="Copy ID", command=lambda: self._copy_to_clipboard(app.device_id), - ).pack(fill="x", pady=(6, 0)) + ).pack(fill="x", pady=(10, 0)) - ctk.CTkButton( + _secondary_button( container, - text="Scan QR with webcam (slower)", - height=28, - fg_color="transparent", - border_width=1, - border_color=("gray70", "gray40"), - text_color=("gray30", "gray80"), + text="Scan QR with webcam", command=self._on_scan_clicked, ).pack(fill="x") @@ -364,8 +512,9 @@ def __init__( textvariable=self._status_var, wraplength=360, justify="center", - font=ctk.CTkFont(size=11), - ).pack(pady=(10, 0)) + font=_fonts()["small"], + text_color=THEME.muted, + ).pack(pady=(14, 0)) def _exists(self) -> bool: try: @@ -376,9 +525,9 @@ def _exists(self) -> bool: def _render_qr(self, device_id: str) -> None: qr_img = pairing.generate_qr(device_id, box_size=4, border=2) qr_img = qr_img.resize((110, 110), Image.Resampling.NEAREST) - ctk_img = ctk.CTkImage(light_image=qr_img, dark_image=qr_img, size=(110, 110)) - self._qr_label.configure(image=ctk_img) - self._qr_label.image = ctk_img # keep reference + tk_img = _tk_image(qr_img) + self._qr_label.configure(image=tk_img) + self._qr_label.image = tk_img # keep reference def _copy_to_clipboard(self, value: str) -> None: try: @@ -426,30 +575,39 @@ def _render_nearby(self, device_ids: list[str]) -> None: for child in self._nearby_frame.winfo_children(): child.destroy() if not device_ids: + empty = ctk.CTkFrame(self._nearby_frame, fg_color="transparent") + empty.pack(pady=20) ctk.CTkLabel( - self._nearby_frame, - text="Searching… make sure the other device is running ClipSync on the same network.", - font=ctk.CTkFont(size=11), - wraplength=320, + empty, + text="No nearby devices found", + font=_fonts()["body"], + text_color=THEME.muted, + ).pack() + ctk.CTkLabel( + empty, + text="Make sure the other device is running ClipSync on the same network.", + font=_fonts()["small"], + wraplength=300, justify="center", - ).pack(pady=16) + text_color=THEME.muted, + ).pack(pady=(4, 0)) return for did in device_ids: - row = ctk.CTkFrame(self._nearby_frame, fg_color=("gray85", "gray22")) - row.pack(fill="x", padx=4, pady=3) + row = ctk.CTkFrame(self._nearby_frame, fg_color=(config.COLOR_ROW_BG_LIGHT, config.COLOR_ROW_BG_DARK), corner_radius=8) + row.pack(fill="x", padx=8, pady=4) row.grid_columnconfigure(0, weight=1) - ctk.CTkLabel(row, text=did[:24] + "…", font=ctk.CTkFont(size=11), anchor="w").grid( - row=0, column=0, sticky="we", padx=10, pady=6 + ctk.CTkLabel(row, text=did[:24] + "…", font=_fonts()["small"], anchor="w", text_color=THEME.text).grid( + row=0, column=0, sticky="we", padx=12, pady=10 ) - ctk.CTkButton( + def _pair_handler(d: str = did) -> None: + self._pair_from_nearby(d) + + _primary_button( row, text="Pair", width=60, - height=28, - fg_color=config.ACCENT_COLOR, - hover_color=config.ACCENT_HOVER, - command=lambda d=did: self._pair_from_nearby(d), - ).grid(row=0, column=1, padx=(0, 8), pady=4) + command=_pair_handler, + ).grid(row=0, column=1, padx=(0, 10), pady=6) def _pair_from_nearby(self, device_id: str) -> None: self._set_pending(device_id) @@ -544,9 +702,9 @@ def _on_frame(self, frame: object) -> None: def update() -> None: if self._preview_label is None or not self._exists(): return - ctk_img = ctk.CTkImage(light_image=img, dark_image=img, size=(nw, nh)) - self._preview_label.configure(image=ctk_img, text="") - self._preview_label.image = ctk_img # keep reference + tk_img = _tk_image(img) + self._preview_label.configure(image=tk_img, text="") + self._preview_label.image = tk_img # keep reference try: self._win.after(0, update) @@ -586,18 +744,20 @@ def __init__( self._app = app self._refreshing = False - ctk.CTkLabel(container, text="Connected devices", font=ctk.CTkFont(size=18, weight="bold")).pack(pady=(0, 10)) + ctk.CTkLabel(container, text="Connected devices", font=_fonts()["headline"], text_color=THEME.text).pack( + pady=(0, 18) + ) - self._list_frame = ctk.CTkScrollableFrame(container, fg_color=("gray90", "gray17")) + self._list_frame = ctk.CTkScrollableFrame( + container, + fg_color=THEME.card_fg(), + border_width=1, + border_color=(config.COLOR_BORDER_LIGHT, config.COLOR_BORDER_DARK), + corner_radius=12, + ) self._list_frame.pack(fill="both", expand=True) - ctk.CTkButton( - container, - text="Refresh", - fg_color=config.ACCENT_COLOR, - hover_color=config.ACCENT_HOVER, - command=self._refresh, - ).pack(fill="x", pady=(12, 0)) + _primary_button(container, text="Refresh", command=self._refresh).pack(fill="x", pady=(16, 0)) self._refresh() self._schedule_refresh() @@ -648,70 +808,68 @@ def _apply_refresh(self, devices: list[dict], error: str | None) -> None: for child in self._list_frame.winfo_children(): child.destroy() if error: - ctk.CTkLabel(self._list_frame, text=error, text_color="red").pack(pady=10) + ctk.CTkLabel(self._list_frame, text=error, text_color=config.COLOR_DANGER, font=_fonts()["body"]).pack(pady=14) return if not devices: empty = ctk.CTkFrame(self._list_frame, fg_color="transparent") - empty.pack(pady=30) + empty.pack(pady=40) ctk.CTkLabel( empty, text="No devices paired yet.", - font=ctk.CTkFont(size=14, weight="bold"), - text_color=("gray30", "gray70"), + font=_fonts()["title"], + text_color=THEME.muted, ).pack() ctk.CTkLabel( empty, text="Go to the Pair tab to connect a device.", - font=ctk.CTkFont(size=11), - text_color=("gray30", "gray70"), + font=_fonts()["small"], + text_color=THEME.muted, ).pack(pady=(4, 0)) return for d in devices: self._build_row(d) def _build_row(self, device: dict) -> None: - row = ctk.CTkFrame(self._list_frame, fg_color=("gray85", "gray22")) - row.pack(fill="x", padx=4, pady=4) + row = ctk.CTkFrame(self._list_frame, fg_color=(config.COLOR_ROW_BG_LIGHT, config.COLOR_ROW_BG_DARK), corner_radius=10) + row.pack(fill="x", padx=8, pady=5) row.grid_columnconfigure(0, weight=1) name_text = device.get("name") or device["deviceID"][:7] - ctk.CTkLabel(row, text=name_text, font=ctk.CTkFont(size=13, weight="bold"), anchor="w").grid( - row=0, column=0, sticky="we", padx=10, pady=(8, 0) + ctk.CTkLabel(row, text=name_text, font=_fonts()["subtitle"], text_color=THEME.text, anchor="w").grid( + row=0, column=0, sticky="we", padx=12, pady=(10, 0) ) - ctk.CTkLabel(row, text=device["deviceID"][:24] + "…", font=ctk.CTkFont(size=10), anchor="w").grid( - row=1, column=0, sticky="we", padx=10, pady=(0, 8) + ctk.CTkLabel(row, text=device["deviceID"][:24] + "…", font=_fonts()["tiny"], text_color=THEME.muted, anchor="w").grid( + row=1, column=0, sticky="we", padx=12, pady=(0, 10) ) - status_color = "#2E8B57" if device["connected"] else ("gray50", "gray60") + status_color = config.COLOR_SUCCESS if device["connected"] else THEME.muted status_text = "● Connected" if device["connected"] else "○ Offline" - ctk.CTkLabel(row, text=status_text, text_color=status_color, font=ctk.CTkFont(size=11)).grid( - row=0, column=1, rowspan=2, padx=10 + ctk.CTkLabel(row, text=status_text, text_color=status_color, font=_fonts()["small"]).grid( + row=0, column=1, rowspan=2, padx=12 ) - ctk.CTkButton( + def _rename_handler( + did: str = device["deviceID"], + nm: str = name_text, + ) -> None: + self._rename_device(did, nm) + + def _remove_handler(did: str = device["deviceID"]) -> None: + self._remove_device(did) + + _ghost_button( row, text="Rename", width=70, - height=28, - fg_color="transparent", - border_width=1, - text_color=config.ACCENT_COLOR, - border_color=config.ACCENT_COLOR, - hover_color=config.ACCENT_HOVER, - command=lambda did=device["deviceID"], nm=name_text: self._rename_device(did, nm), - ).grid(row=0, column=2, rowspan=2, padx=(0, 6)) + command=_rename_handler, + ).grid(row=0, column=2, rowspan=2, padx=(0, 4)) - ctk.CTkButton( + _secondary_button( row, text="Remove", width=70, - height=28, - fg_color="transparent", - border_width=1, - text_color=("gray30", "gray80"), - hover_color=("gray75", "gray30"), - command=lambda did=device["deviceID"]: self._remove_device(did), - ).grid(row=0, column=3, rowspan=2, padx=(0, 10)) + command=_remove_handler, + ).grid(row=0, column=3, rowspan=2, padx=(0, 12)) def _remove_device(self, device_id: str) -> None: try: @@ -728,17 +886,31 @@ def _rename_device(self, device_id: str, current_name: str) -> None: dialog.transient(self._win) dialog.grab_set() - ctk.CTkLabel(dialog, text=f"New name for {device_id[:7]}:", font=ctk.CTkFont(size=12)).pack( - padx=20, pady=(18, 6) - ) - entry = ctk.CTkEntry(dialog) + dialog.configure(fg_color=THEME.bg) + container = ctk.CTkFrame(dialog, fg_color="transparent") + container.pack(fill="both", expand=True, padx=20, pady=20) + + ctk.CTkLabel( + container, + text="Rename device", + font=_fonts()["title"], + text_color=THEME.text, + ).pack(anchor="w") + ctk.CTkLabel( + container, + text=f"Choose a new name for {device_id[:7]}.", + font=_fonts()["small"], + text_color=THEME.muted, + ).pack(anchor="w", pady=(4, 12)) + + entry = _entry(container) entry.insert(0, current_name) - entry.pack(fill="x", padx=20) + entry.pack(fill="x") entry.select_range(0, "end") entry.focus_set() - btns = ctk.CTkFrame(dialog, fg_color="transparent") - btns.pack(fill="x", padx=20, pady=(12, 16)) + btns = ctk.CTkFrame(container, fg_color="transparent") + btns.pack(fill="x", pady=(16, 0)) def do_save() -> None: new_name = entry.get().strip() @@ -751,12 +923,10 @@ def do_save() -> None: dialog.destroy() self._refresh() - ctk.CTkButton(btns, text="Cancel", fg_color="transparent", border_width=1, command=dialog.destroy).pack( - side="left", expand=True, fill="x", padx=(0, 4) + _secondary_button(btns, text="Cancel", command=dialog.destroy).pack( + side="left", expand=True, fill="x", padx=(0, 6) ) - ctk.CTkButton( - btns, text="Save", fg_color=config.ACCENT_COLOR, hover_color=config.ACCENT_HOVER, command=do_save - ).pack(side="left", expand=True, fill="x", padx=(4, 0)) + _primary_button(btns, text="Save", command=do_save).pack(side="left", expand=True, fill="x", padx=(6, 0)) entry.bind("", lambda _e: do_save()) dialog.bind("", lambda _e: dialog.destroy()) @@ -774,60 +944,88 @@ def __init__( self._app = app self._logs_window: LogsWindow | None = None - ctk.CTkLabel(container, text="Settings", font=ctk.CTkFont(size=18, weight="bold")).pack( - anchor="w", pady=(0, 12) + ctk.CTkLabel(container, text="Settings", font=_fonts()["headline"], text_color=THEME.text).pack( + anchor="w", pady=(0, 18) ) + general_card = _card_frame(container) + general_card.pack(fill="x", pady=(0, 16)) + _section_header(general_card, "General").pack(anchor="w", padx=16, pady=(14, 8)) + self._autostart_var = ctk.BooleanVar(value=is_autostart_enabled()) - ctk.CTkSwitch( - container, - text="Start on login", - variable=self._autostart_var, - command=self._on_autostart_toggle, - progress_color=config.ACCENT_COLOR, - ).pack(anchor="w", pady=4) + _switch(general_card, "Start on login", self._autostart_var, self._on_autostart_toggle).pack( + anchor="w", padx=16, pady=4 + ) self._notify_var = ctk.BooleanVar(value=bool(app.settings.get("show_notifications"))) - ctk.CTkSwitch( - container, - text="Show notifications on sync", - variable=self._notify_var, - command=self._on_notify_toggle, - progress_color=config.ACCENT_COLOR, - ).pack(anchor="w", pady=4) + _switch(general_card, "Show notifications on sync", self._notify_var, self._on_notify_toggle).pack( + anchor="w", padx=16, pady=4 + ) self._sync_enabled_var = ctk.BooleanVar(value=not bool(app.settings.get("sync_paused"))) - ctk.CTkSwitch( - container, - text="Sync enabled", - variable=self._sync_enabled_var, - command=self._on_sync_enabled_toggle, - progress_color=config.ACCENT_COLOR, - ).pack(anchor="w", pady=4) + _switch(general_card, "Sync enabled", self._sync_enabled_var, self._on_sync_enabled_toggle).pack( + anchor="w", padx=16, pady=4 + ) self._auto_accept_var = ctk.BooleanVar(value=bool(app.settings.get("auto_accept_incoming"))) - ctk.CTkSwitch( - container, - text="Auto-accept incoming requests (no prompt)", - variable=self._auto_accept_var, - command=self._on_auto_accept_toggle, - progress_color=config.ACCENT_COLOR, - ).pack(anchor="w", pady=4) - - ctk.CTkLabel(container, text="Appearance", font=ctk.CTkFont(size=11)).pack(anchor="w", pady=(14, 2)) - theme_row = ctk.CTkFrame(container, fg_color="transparent") - theme_row.pack(fill="x", pady=(2, 0)) + _switch( + general_card, + "Auto-accept incoming requests (no prompt)", + self._auto_accept_var, + self._on_auto_accept_toggle, + ).pack(anchor="w", padx=16, pady=(4, 14)) + + privacy_card = _card_frame(container) + privacy_card.pack(fill="x", pady=(0, 16)) + _section_header(privacy_card, "Privacy & security").pack(anchor="w", padx=16, pady=(14, 8)) + + self._log_mirror_var = ctk.BooleanVar(value=bool(app.settings.get("debug_log_mirror"))) + _switch( + privacy_card, + "Share my log with paired devices (for debugging)", + self._log_mirror_var, + self._on_log_mirror_toggle, + ).pack(anchor="w", padx=16, pady=(4, 2)) + ctk.CTkLabel( + privacy_card, + text="Copies this device's log into the synced folder. No clipboard text is logged.", + font=_fonts()["tiny"], + justify="left", + text_color=THEME.muted, + ).pack(anchor="w", padx=(52, 16), pady=(0, 14)) + + _section_header(privacy_card, "Encryption passphrase (optional)").pack( + anchor="w", padx=16, pady=(4, 2) + ) + ctk.CTkLabel( + privacy_card, + text="Same passphrase on every device. Empty = no encryption.", + font=_fonts()["tiny"], + text_color=THEME.muted, + ).pack(anchor="w", padx=16) + passphrase_row = ctk.CTkFrame(privacy_card, fg_color="transparent") + passphrase_row.pack(fill="x", padx=16, pady=(6, 16)) + self._passphrase_entry = _entry(passphrase_row, show="•") + self._passphrase_entry.insert(0, str(app.settings.get("encryption_passphrase") or "")) + self._passphrase_entry.pack(side="left", fill="x", expand=True, padx=(0, 8)) + _primary_button(passphrase_row, text="Save", command=self._on_save_passphrase, width=70).pack(side="left") + + appearance_card = _card_frame(container) + appearance_card.pack(fill="x", pady=(0, 16)) + _section_header(appearance_card, "Appearance").pack(anchor="w", padx=16, pady=(14, 8)) + theme_row = ctk.CTkFrame(appearance_card, fg_color="transparent") + theme_row.pack(fill="x", padx=16, pady=(0, 14)) self._theme_seg = ctk.CTkSegmentedButton( theme_row, values=["Light", "Dark", "System"], command=self._on_theme_changed, - fg_color=("gray85", "gray25"), - selected_color=config.ACCENT_COLOR, - selected_hover_color=config.ACCENT_HOVER, - unselected_color=("gray90", "gray20"), - unselected_hover_color=("gray80", "gray30"), - text_color="white", - text_color_disabled=("gray50", "gray60"), + fg_color=(config.COLOR_BORDER_LIGHT, config.COLOR_BORDER_DARK), + selected_color=config.COLOR_PRIMARY, + selected_hover_color=config.COLOR_PRIMARY_HOVER, + unselected_color=(config.COLOR_CARD_LIGHT, config.COLOR_CARD_DARK), + unselected_hover_color=(config.COLOR_ROW_BG_LIGHT, config.COLOR_ROW_BG_DARK), + text_color=THEME.text, + text_color_disabled=THEME.muted, ) self._theme_seg.pack(side="left") current_theme = str(app.settings.get("theme") or "System") @@ -836,9 +1034,9 @@ def __init__( else: self._theme_seg.set("System") - ctk.CTkLabel(container, text="Clipboard history auto-clear", font=ctk.CTkFont(size=11)).pack( - anchor="w", pady=(14, 2) - ) + history_card = _card_frame(container) + history_card.pack(fill="x", pady=(0, 16)) + _section_header(history_card, "Clipboard history auto-clear").pack(anchor="w", padx=16, pady=(14, 8)) self._auto_clear_options: dict[str, int] = { "Never": 0, "5 minutes": 5, @@ -848,103 +1046,76 @@ def __init__( "4 hours": 240, "24 hours": 1440, } - auto_clear_row = ctk.CTkFrame(container, fg_color="transparent") - auto_clear_row.pack(fill="x", pady=(2, 0)) + auto_clear_row = ctk.CTkFrame(history_card, fg_color="transparent") + auto_clear_row.pack(fill="x", padx=16, pady=(0, 14)) current_auto_clear = int(app.settings.get("history_auto_clear_minutes") or 0) auto_clear_label = {v: k for k, v in self._auto_clear_options.items()}.get(current_auto_clear, "Never") self._auto_clear_menu = ctk.CTkOptionMenu( auto_clear_row, values=list(self._auto_clear_options.keys()), command=self._on_auto_clear_changed, - fg_color=("gray85", "gray25"), - button_color=config.ACCENT_COLOR, - button_hover_color=config.ACCENT_HOVER, - text_color="white", - dropdown_fg_color=("gray90", "gray20"), - dropdown_hover_color=("gray80", "gray30"), - dropdown_text_color="white", + fg_color=(config.COLOR_ROW_BG_LIGHT, config.COLOR_ROW_BG_DARK), + button_color=config.COLOR_PRIMARY, + button_hover_color=config.COLOR_PRIMARY_HOVER, + text_color=THEME.text, + dropdown_fg_color=(config.COLOR_CARD_LIGHT, config.COLOR_CARD_DARK), + dropdown_hover_color=(config.COLOR_ROW_BG_LIGHT, config.COLOR_ROW_BG_DARK), + dropdown_text_color=THEME.text, + corner_radius=8, ) self._auto_clear_menu.pack(side="left") self._auto_clear_menu.set(auto_clear_label) - ctk.CTkLabel(container, text="Encryption passphrase (optional)", font=ctk.CTkFont(size=11)).pack( - anchor="w", pady=(14, 2) - ) - ctk.CTkLabel( - container, - text="Same passphrase on every device. Empty = no encryption.", - font=ctk.CTkFont(size=10), - text_color=("gray30", "gray70"), - ).pack(anchor="w") - passphrase_row = ctk.CTkFrame(container, fg_color="transparent") - passphrase_row.pack(fill="x", pady=(2, 0)) - self._passphrase_entry = ctk.CTkEntry( - passphrase_row, show="•", border_width=1, border_color=("gray70", "gray40") - ) - self._passphrase_entry.insert(0, str(app.settings.get("encryption_passphrase") or "")) - self._passphrase_entry.pack(side="left", fill="x", expand=True, padx=(0, 6)) - ctk.CTkButton( - passphrase_row, - text="Save", - width=70, - fg_color=config.ACCENT_COLOR, - hover_color=config.ACCENT_HOVER, - command=self._on_save_passphrase, - ).pack(side="left") + advanced_card = _card_frame(container) + advanced_card.pack(fill="x", pady=(0, 16)) + _section_header(advanced_card, "Advanced").pack(anchor="w", padx=16, pady=(14, 8)) - ctk.CTkLabel(container, text="Sync folder path (advanced)", font=ctk.CTkFont(size=11)).pack( - anchor="w", pady=(14, 2) - ) - folder_row = ctk.CTkFrame(container, fg_color="transparent") - folder_row.pack(fill="x") - self._folder_entry = ctk.CTkEntry(folder_row, border_width=1, border_color=("gray70", "gray40")) + _section_header(advanced_card, "Sync folder path").pack(anchor="w", padx=16) + ctk.CTkLabel( + advanced_card, + text="Changing this requires a restart to take effect.", + font=_fonts()["tiny"], + text_color=THEME.muted, + ).pack(anchor="w", padx=16) + folder_row = ctk.CTkFrame(advanced_card, fg_color="transparent") + folder_row.pack(fill="x", padx=16, pady=(6, 14)) + self._folder_entry = _entry(folder_row) self._folder_entry.insert(0, str(app.settings.get("sync_folder") or config.SYNC_FOLDER)) - self._folder_entry.pack(side="left", fill="x", expand=True, padx=(0, 6)) - ctk.CTkButton( - folder_row, - text="Save", - width=70, - fg_color=config.ACCENT_COLOR, - hover_color=config.ACCENT_HOVER, - command=self._on_save_folder, - ).pack(side="left") + self._folder_entry.pack(side="left", fill="x", expand=True, padx=(0, 8)) + _primary_button(folder_row, text="Save", command=self._on_save_folder, width=70).pack(side="left") - ctk.CTkButton( - container, + _secondary_button( + advanced_card, text="View Syncthing logs", - fg_color="transparent", - border_width=1, - text_color=config.ACCENT_COLOR, - border_color=config.ACCENT_COLOR, - hover_color=config.ACCENT_HOVER, command=self._on_view_logs, - ).pack(fill="x", pady=(18, 6)) + ).pack(fill="x", padx=16, pady=(0, 14)) - ctk.CTkButton( - container, + danger_card = _card_frame(container) + danger_card.pack(fill="x", pady=(0, 16)) + _section_header(danger_card, "Danger zone").pack(anchor="w", padx=16, pady=(14, 8)) + _primary_button( + danger_card, text="Reset / unpair all devices", - fg_color="#9b2c2c", - hover_color="#7a2222", + fg_color=config.COLOR_DANGER, + hover_color=config.COLOR_DANGER_HOVER, command=self._on_reset, - ).pack(fill="x", pady=(0, 6)) - - update_row = ctk.CTkFrame(container, fg_color="transparent") - update_row.pack(fill="x", pady=(8, 0)) - self._update_btn = ctk.CTkButton( + ).pack(fill="x", padx=16, pady=(0, 14)) + + update_card = _card_frame(container) + update_card.pack(fill="x", pady=(0, 16)) + _section_header(update_card, "Updates").pack(anchor="w", padx=16, pady=(14, 8)) + update_row = ctk.CTkFrame(update_card, fg_color="transparent") + update_row.pack(fill="x", padx=16, pady=(0, 14)) + self._update_btn = _secondary_button( update_row, text=f"Check for updates (v{__version__})", - fg_color="transparent", - border_width=1, - text_color=config.ACCENT_COLOR, - border_color=config.ACCENT_COLOR, - hover_color=config.ACCENT_HOVER, command=self._on_check_update, ) self._update_btn.pack(fill="x") self._download_btn: ctk.CTkButton | None = None self._update_url = update.RELEASES_HTML_URL - self._status = ctk.CTkLabel(container, text="", font=ctk.CTkFont(size=11)) + self._status = ctk.CTkLabel(container, text="", font=_fonts()["small"], text_color=THEME.muted) self._status.pack(pady=(8, 0)) def _exists(self) -> bool: @@ -969,6 +1140,18 @@ def _on_sync_enabled_toggle(self) -> None: self._app.on_pause_changed(paused) self._status.configure(text=f"Sync {'enabled' if enabled else 'paused'}.") + def _on_log_mirror_toggle(self) -> None: + enabled = bool(self._log_mirror_var.get()) + self._app.settings.set("debug_log_mirror", enabled) + self._app.on_settings_changed() + self._status.configure( + text=( + "Log sharing on. Paired devices will receive this device's log." + if enabled + else "Log sharing off. Your published log will be removed shortly." + ) + ) + def _on_auto_accept_toggle(self) -> None: enabled = bool(self._auto_accept_var.get()) self._app.settings.set("auto_accept_incoming", enabled) @@ -1060,14 +1243,12 @@ def _finish_update_check(self, info: update.UpdateInfo | None, error: str | None self._update_url = info.release_url self._status.configure(text=f"Update available: v{info.latest_version} (you have v{info.current_version}).") if self._download_btn is None: - self._download_btn = ctk.CTkButton( + self._download_btn = _primary_button( self._update_btn.master, text="Download update", - fg_color=config.ACCENT_COLOR, - hover_color=config.ACCENT_HOVER, command=self._on_download_clicked, ) - self._download_btn.pack(fill="x", pady=(6, 0)) + self._download_btn.pack(fill="x", pady=(10, 0)) def _on_download_clicked(self) -> None: if update.open_download_page(self._update_url): @@ -1078,38 +1259,46 @@ def _on_download_clicked(self) -> None: def _on_reset(self) -> None: confirm = ctk.CTkToplevel(self._win) confirm.title("Confirm reset") + confirm.configure(fg_color=THEME.bg) confirm.resizable(False, False) - _center_window(confirm, 320, 140) + _center_window(confirm, 340, 170) confirm.bind("", lambda _e: confirm.destroy()) + container = ctk.CTkFrame(confirm, fg_color="transparent") + container.pack(fill="both", expand=True, padx=24, pady=24) + ctk.CTkLabel( - confirm, - text="Remove all paired devices?\nYou will need to re-pair them.", - justify="center", - ).pack(padx=20, pady=(20, 10)) - btns = ctk.CTkFrame(confirm, fg_color="transparent") - btns.pack(fill="x", padx=20, pady=(0, 16)) + container, + text="Reset everything?", + font=_fonts()["title"], + text_color=THEME.text, + ).pack(anchor="w") + ctk.CTkLabel( + container, + text="This removes all paired devices. You will need to re-pair them.", + font=_fonts()["small"], + text_color=THEME.muted, + justify="left", + wraplength=280, + ).pack(anchor="w", pady=(6, 18)) + + btns = ctk.CTkFrame(container, fg_color="transparent") + btns.pack(fill="x") def do_reset() -> None: confirm.destroy() self._app.on_reset() self._status.configure(text="All devices removed.") - ctk.CTkButton( - btns, - text="Cancel", - height=28, - fg_color="transparent", - border_width=1, - command=confirm.destroy, - ).pack(side="left", expand=True, fill="x", padx=(0, 4)) - ctk.CTkButton( + _secondary_button(btns, text="Cancel", command=confirm.destroy).pack( + side="left", expand=True, fill="x", padx=(0, 6) + ) + _primary_button( btns, text="Reset", - height=28, - fg_color="#9b2c2c", - hover_color="#7a2222", + fg_color=config.COLOR_DANGER, + hover_color=config.COLOR_DANGER_HOVER, command=do_reset, - ).pack(side="left", expand=True, fill="x", padx=(4, 0)) + ).pack(side="left", expand=True, fill="x", padx=(6, 0)) # --------------------------------------------------------------------------- @@ -1145,9 +1334,13 @@ class SettingsWindow(_BaseWindow): """Toggles for autostart, notifications, pause, sync folder, reset.""" def __init__(self, parent: ctk.CTk, app: AppContext, on_close: Callable[[], None]) -> None: - super().__init__(parent, f"{config.APP_NAME} — Settings", config.SETTINGS_WINDOW_SIZE, on_close) - container = ctk.CTkScrollableFrame(self.window, fg_color=("gray95", "gray13")) - container.pack(fill="both", expand=True, padx=20, pady=20) + super().__init__(parent, f"{config.APP_NAME} — Settings", (440, 560), on_close) + container = ctk.CTkScrollableFrame( + self.window, + fg_color=THEME.bg, + scrollbar_button_color=(config.COLOR_BORDER_LIGHT, config.COLOR_BORDER_DARK), + ) + container.pack(fill="both", expand=True, padx=16, pady=16) _SettingsContent(self.window, container, app) @@ -1164,24 +1357,37 @@ def __init__( on_close: Callable[[], None], initial_tab: str = "Devices", ) -> None: - super().__init__(parent, config.APP_NAME, (520, 580), on_close) + super().__init__(parent, config.APP_NAME, (560, 620), on_close) self.window.resizable(True, True) - tabs = ctk.CTkTabview(self.window) - tabs.pack(fill="both", expand=True, padx=8, pady=8) + tabs = ctk.CTkTabview( + self.window, + fg_color=THEME.bg, + segmented_button_fg_color=(config.COLOR_BORDER_LIGHT, config.COLOR_BORDER_DARK), + segmented_button_selected_color=config.COLOR_PRIMARY, + segmented_button_selected_hover_color=config.COLOR_PRIMARY_HOVER, + segmented_button_unselected_color=THEME.card_fg(), + segmented_button_unselected_hover_color=(config.COLOR_ROW_BG_LIGHT, config.COLOR_ROW_BG_DARK), + text_color=THEME.text, + ) + tabs.pack(fill="both", expand=True, padx=12, pady=12) for name in self._TAB_NAMES: tabs.add(name) dev_frame = ctk.CTkFrame(tabs.tab("Devices"), fg_color="transparent") - dev_frame.pack(fill="both", expand=True, padx=16, pady=12) + dev_frame.pack(fill="both", expand=True, padx=14, pady=10) _DevicesContent(self.window, dev_frame, app) pair_frame = ctk.CTkFrame(tabs.tab("Pair"), fg_color="transparent") - pair_frame.pack(fill="both", expand=True, padx=16, pady=12) + pair_frame.pack(fill="both", expand=True, padx=14, pady=10) self._pairing = _PairingContent(self.window, pair_frame, app) - settings_frame = ctk.CTkScrollableFrame(tabs.tab("Settings"), fg_color=("gray95", "gray13")) - settings_frame.pack(fill="both", expand=True, padx=16, pady=12) + settings_frame = ctk.CTkScrollableFrame( + tabs.tab("Settings"), + fg_color=THEME.bg, + scrollbar_button_color=(config.COLOR_BORDER_LIGHT, config.COLOR_BORDER_DARK), + ) + settings_frame.pack(fill="both", expand=True, padx=14, pady=10) _SettingsContent(self.window, settings_frame, app) if initial_tab in self._TAB_NAMES: @@ -1196,19 +1402,22 @@ class LogsWindow(_BaseWindow): """Read-only tail of the ClipSync log file.""" def __init__(self, parent: ctk.CTk, on_close: Callable[[], None]) -> None: - super().__init__(parent, f"{config.APP_NAME} — Logs", (600, 400), on_close) + super().__init__(parent, f"{config.APP_NAME} — Logs", (640, 440), on_close) container = ctk.CTkFrame(self.window, fg_color="transparent") - container.pack(fill="both", expand=True, padx=16, pady=16) - self._textbox = ctk.CTkTextbox(container, wrap="none", font=ctk.CTkFont(family="Menlo", size=11)) + container.pack(fill="both", expand=True, padx=20, pady=20) + ctk.CTkLabel(container, text="Logs", font=_fonts()["headline"], text_color=THEME.text).pack(anchor="w", pady=(0, 12)) + self._textbox = ctk.CTkTextbox( + container, + wrap="none", + font=_fonts()["mono"], + fg_color=(config.COLOR_CARD_LIGHT, config.COLOR_CARD_DARK), + border_color=(config.COLOR_BORDER_LIGHT, config.COLOR_BORDER_DARK), + border_width=1, + corner_radius=10, + ) self._textbox.pack(fill="both", expand=True) self._refresh() - ctk.CTkButton( - container, - text="Refresh", - fg_color=config.ACCENT_COLOR, - hover_color=config.ACCENT_HOVER, - command=self._refresh, - ).pack(fill="x", pady=(10, 0)) + _primary_button(container, text="Refresh", command=self._refresh).pack(fill="x", pady=(12, 0)) def _refresh(self) -> None: try: @@ -1240,21 +1449,27 @@ def __init__(self, parent: ctk.CTk, app: AppContext, on_close: Callable[[], None container = ctk.CTkFrame(self.window, fg_color="transparent") container.pack(fill="both", expand=True, padx=20, pady=20) - ctk.CTkLabel(container, text="Incoming device requests", font=ctk.CTkFont(size=18, weight="bold")).pack( - pady=(0, 8) - ) + ctk.CTkLabel( + container, text="Incoming device requests", font=_fonts()["headline"], text_color=THEME.text + ).pack(pady=(0, 6)) ctk.CTkLabel( container, text="Accept a device to start syncing clipboard with it.", - font=ctk.CTkFont(size=11), - text_color=("gray30", "gray70"), - ).pack(pady=(0, 10)) + font=_fonts()["small"], + text_color=THEME.muted, + ).pack(pady=(0, 16)) - self._list_frame = ctk.CTkScrollableFrame(container, fg_color=("gray90", "gray17")) + self._list_frame = ctk.CTkScrollableFrame( + container, + fg_color=THEME.card_fg(), + border_width=1, + border_color=(config.COLOR_BORDER_LIGHT, config.COLOR_BORDER_DARK), + corner_radius=12, + ) self._list_frame.pack(fill="both", expand=True) - self._status = ctk.CTkLabel(container, text="", font=ctk.CTkFont(size=11)) - self._status.pack(pady=(8, 0)) + self._status = ctk.CTkLabel(container, text="", font=_fonts()["small"], text_color=THEME.muted) + self._status.pack(pady=(10, 0)) self._refresh() self._schedule_refresh() @@ -1299,7 +1514,7 @@ def _apply_refresh(self, pending: dict, error: str | None) -> None: for child in self._list_frame.winfo_children(): child.destroy() if error: - ctk.CTkLabel(self._list_frame, text=error, text_color="red").pack(pady=10) + ctk.CTkLabel(self._list_frame, text=error, text_color=config.COLOR_DANGER, font=_fonts()["body"]).pack(pady=14) return rejected = set(self._app.settings.get("rejected_device_ids") or []) visible = [ @@ -1309,57 +1524,55 @@ def _apply_refresh(self, pending: dict, error: str | None) -> None: ] if not visible: empty = ctk.CTkFrame(self._list_frame, fg_color="transparent") - empty.pack(pady=30) + empty.pack(pady=40) ctk.CTkLabel( empty, text="No pending requests.", - font=ctk.CTkFont(size=14, weight="bold"), - text_color=("gray30", "gray70"), + font=_fonts()["title"], + text_color=THEME.muted, ).pack() ctk.CTkLabel( empty, text="Ask the other device to pair with this one.", - font=ctk.CTkFont(size=11), - text_color=("gray30", "gray70"), + font=_fonts()["small"], + text_color=THEME.muted, ).pack(pady=(4, 0)) return for device_id, info in visible: self._build_row(device_id, info) def _build_row(self, device_id: str, info: dict) -> None: - row = ctk.CTkFrame(self._list_frame, fg_color=("gray85", "gray22")) - row.pack(fill="x", padx=4, pady=4) + row = ctk.CTkFrame(self._list_frame, fg_color=(config.COLOR_ROW_BG_LIGHT, config.COLOR_ROW_BG_DARK), corner_radius=10) + row.pack(fill="x", padx=8, pady=5) row.grid_columnconfigure(0, weight=1) name = info.get("name") or device_id[:7] - ctk.CTkLabel(row, text=str(name), font=ctk.CTkFont(size=13, weight="bold"), anchor="w").grid( - row=0, column=0, sticky="we", padx=10, pady=(8, 0) + ctk.CTkLabel(row, text=str(name), font=_fonts()["subtitle"], text_color=THEME.text, anchor="w").grid( + row=0, column=0, sticky="we", padx=12, pady=(10, 0) ) - ctk.CTkLabel(row, text=device_id[:24] + "…", font=ctk.CTkFont(size=10), anchor="w").grid( - row=1, column=0, sticky="we", padx=10, pady=(0, 8) + ctk.CTkLabel(row, text=device_id[:24] + "…", font=_fonts()["tiny"], text_color=THEME.muted, anchor="w").grid( + row=1, column=0, sticky="we", padx=12, pady=(0, 10) ) - ctk.CTkButton( + def _accept_handler(did: str = device_id) -> None: + self._accept(did) + + def _reject_handler(did: str = device_id) -> None: + self._reject(did) + + _primary_button( row, text="Accept", width=70, - height=28, - fg_color=config.ACCENT_COLOR, - hover_color=config.ACCENT_HOVER, - command=lambda did=device_id: self._accept(did), - ).grid(row=0, column=1, rowspan=2, padx=(0, 6)) + command=_accept_handler, + ).grid(row=0, column=1, rowspan=2, padx=(0, 4)) - ctk.CTkButton( + _secondary_button( row, text="Reject", width=70, - height=28, - fg_color="transparent", - border_width=1, - text_color=("gray30", "gray80"), - hover_color=("gray75", "gray30"), - command=lambda did=device_id: self._reject(did), - ).grid(row=0, column=2, rowspan=2, padx=(0, 10)) + command=_reject_handler, + ).grid(row=0, column=2, rowspan=2, padx=(0, 12)) def _accept(self, device_id: str) -> None: self._handled.add(device_id) @@ -1390,45 +1603,35 @@ def __init__(self, parent: ctk.CTk, app: AppContext, on_close: Callable[[], None header = ctk.CTkFrame(container, fg_color="transparent") header.pack(fill="x", pady=(0, 10)) - ctk.CTkLabel(header, text="Clipboard History", font=ctk.CTkFont(size=18, weight="bold")).pack(side="left") - self._status = ctk.CTkLabel(header, text="", font=ctk.CTkFont(size=11), text_color=("gray30", "gray70")) + ctk.CTkLabel(header, text="Clipboard History", font=_fonts()["headline"], text_color=THEME.text).pack(side="left") + self._status = ctk.CTkLabel(header, text="", font=_fonts()["small"], text_color=THEME.muted) self._status.pack(side="right") - search_row = ctk.CTkFrame(container, fg_color="transparent") - search_row.pack(fill="x", pady=(0, 8)) + search_row = _card_frame(container) + search_row.pack(fill="x", pady=(0, 12)) self._search_var = ctk.StringVar() self._search_var.trace_add("write", lambda *_: self._refresh()) - self._search_entry = ctk.CTkEntry( + self._search_entry = _entry( search_row, placeholder_text="Search history…", textvariable=self._search_var, ) - self._search_entry.pack(side="left", fill="x", expand=True) + self._search_entry.pack(side="left", fill="x", expand=True, padx=10, pady=10) self._search_entry.bind("", lambda _e: (self._search_var.set(""), self._search_entry.focus_set())) - self._list_frame = ctk.CTkScrollableFrame(container, fg_color=("gray90", "gray17")) + self._list_frame = ctk.CTkScrollableFrame( + container, + fg_color=THEME.card_fg(), + border_width=1, + border_color=(config.COLOR_BORDER_LIGHT, config.COLOR_BORDER_DARK), + corner_radius=12, + ) self._list_frame.pack(fill="both", expand=True) btn_row = ctk.CTkFrame(container, fg_color="transparent") - btn_row.pack(fill="x", pady=(10, 0)) - ctk.CTkButton( - btn_row, - text="Refresh", - height=28, - fg_color=config.ACCENT_COLOR, - hover_color=config.ACCENT_HOVER, - width=100, - command=self._refresh, - ).pack(side="left") - ctk.CTkButton( - btn_row, - text="Clear All", - height=28, - fg_color=("gray75", "gray30"), - hover_color=("gray65", "gray40"), - width=100, - command=self._confirm_clear, - ).pack(side="right") + btn_row.pack(fill="x", pady=(14, 0)) + _primary_button(btn_row, text="Refresh", command=self._refresh, width=100).pack(side="left") + _secondary_button(btn_row, text="Clear All", command=self._confirm_clear, width=100).pack(side="right") self._all_entries: list[object] = [] self._refresh() @@ -1503,40 +1706,47 @@ def _build_row(self, entry: object, now: float) -> None: if len(preview) > 72: preview = preview[:72] + "..." - row = ctk.CTkFrame(self._list_frame, fg_color=("white", "gray20"), corner_radius=6) - row.pack(fill="x", padx=4, pady=3) + row = ctk.CTkFrame( + self._list_frame, + fg_color=(config.COLOR_CARD_LIGHT, config.COLOR_CARD_DARK), + border_width=1, + border_color=(config.COLOR_BORDER_LIGHT, config.COLOR_BORDER_DARK), + corner_radius=10, + ) + row.pack(fill="x", padx=6, pady=4) row.grid_columnconfigure(1, weight=1) - source_label = "[Remote]" if source == "remote" else "[Local]" + source_label = "Remote" if source == "remote" else "Local" + source_color = config.COLOR_PRIMARY if source == "remote" else THEME.muted meta = ctk.CTkLabel( row, - text=f"{time_str} {source_label}", - font=ctk.CTkFont(size=10), - text_color=("gray30", "gray70"), + text=f"{time_str} • {source_label}", + font=_fonts()["tiny"], + text_color=source_color, anchor="w", ) - meta.grid(row=0, column=0, columnspan=2, sticky="ew", padx=10, pady=(6, 0)) + meta.grid(row=0, column=0, columnspan=2, sticky="ew", padx=12, pady=(8, 0)) preview_label = ctk.CTkLabel( row, text=preview or "(empty)", - font=ctk.CTkFont(size=12), + font=_fonts()["small"], + text_color=THEME.text, anchor="w", justify="left", ) - preview_label.grid(row=1, column=0, columnspan=2, sticky="ew", padx=10, pady=(2, 6)) + preview_label.grid(row=1, column=0, columnspan=2, sticky="ew", padx=12, pady=(2, 8)) - copy_btn = ctk.CTkButton( + def _copy_handler(t: str = text) -> None: + self._copy_entry(t) + + copy_btn = _primary_button( row, text="Copy", width=56, - height=28, - font=ctk.CTkFont(size=11), - fg_color=config.ACCENT_COLOR, - hover_color=config.ACCENT_HOVER, - command=lambda t=text, b=None: self._copy_entry(t), + command=_copy_handler, ) - copy_btn.grid(row=0, column=2, rowspan=2, padx=(0, 8), pady=6) + copy_btn.grid(row=0, column=2, rowspan=2, padx=(0, 10), pady=6) def _copy_entry(self, text: str) -> None: try: @@ -1551,23 +1761,31 @@ def _copy_entry(self, text: str) -> None: def _confirm_clear(self) -> None: dialog = ctk.CTkToplevel(self.window) dialog.title("Clear History") + dialog.configure(fg_color=THEME.bg) dialog.resizable(False, False) - _center_window(dialog, 320, 140) + _center_window(dialog, 340, 170) dialog.lift() dialog.focus_force() dialog.grab_set() dialog.bind("", lambda _e: dialog.destroy()) - ctk.CTkLabel(dialog, text="Clear all clipboard history?", font=ctk.CTkFont(size=13)).pack(pady=(24, 4)) + container = ctk.CTkFrame(dialog, fg_color="transparent") + container.pack(fill="both", expand=True, padx=24, pady=24) + ctk.CTkLabel( + container, + text="Clear all clipboard history?", + font=_fonts()["title"], + text_color=THEME.text, + ).pack(anchor="w") ctk.CTkLabel( - dialog, + container, text="This cannot be undone.", - font=ctk.CTkFont(size=11), - text_color=("gray30", "gray70"), - ).pack() + font=_fonts()["small"], + text_color=THEME.muted, + ).pack(anchor="w", pady=(6, 18)) - btn_row = ctk.CTkFrame(dialog, fg_color="transparent") - btn_row.pack(pady=14) + btn_row = ctk.CTkFrame(container, fg_color="transparent") + btn_row.pack(fill="x") def do_clear() -> None: _emit("clear_history") @@ -1577,24 +1795,16 @@ def do_clear() -> None: dialog.destroy() self._refresh() - ctk.CTkButton( - btn_row, - text="Cancel", - width=90, - height=28, - fg_color=("gray75", "gray30"), - hover_color=("gray65", "gray40"), - command=dialog.destroy, - ).pack(side="left", padx=6) - ctk.CTkButton( + _secondary_button(btn_row, text="Cancel", command=dialog.destroy).pack( + side="left", expand=True, fill="x", padx=(0, 6) + ) + _primary_button( btn_row, text="Clear All", - width=90, - height=28, - fg_color=config.ACCENT_COLOR, - hover_color=config.ACCENT_HOVER, + fg_color=config.COLOR_DANGER, + hover_color=config.COLOR_DANGER_HOVER, command=do_clear, - ).pack(side="left", padx=6) + ).pack(side="left", expand=True, fill="x", padx=(6, 0)) def _run_child(window_name: str) -> int: @@ -1627,8 +1837,19 @@ def _run_child(window_name: str) -> int: theme = "System" ctk.set_appearance_mode(theme) ctk.set_default_color_theme("dark-blue") + THEME.set_mode(theme) root = ctk.CTk() root.withdraw() + # The root is intentionally hidden on every platform. Configuring its + # background color on Windows, especially with CTk's titlebar + # manipulation disabled, can trigger internal state updates that leave the + # real Toplevel window hidden or unresponsive. Keep the cosmetics off the + # root and apply them to the actual windows instead. + if sys.platform != "win32": + try: + root.configure(fg_color=THEME.bg) + except Exception: + pass def _quit() -> None: try: diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..22f5d2b --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,49 @@ +"""Global test isolation. + +ClipboardSync builds a ClipboardHistory bound to ``config.HISTORY_FILE`` at +construction, so any test that constructs one writes to whatever that module +global points at. Several suites also set an encryption passphrase +("shared-secret", "mac-pass", ...) before syncing, so an unisolated run does +not merely touch the developer's real clipboard history: it overwrites it, +encrypted with a throwaway key, and the running app can then never read its +own history again. + +That is not hypothetical. It happened: running the suite on a machine with +ClipSync installed replaced a 76KB real history with a 1698-byte file +encrypted under "shared-secret", repeatedly. + +Individual modules used to patch HISTORY_FILE one by one, and four of them +(test_cross_os_sync, test_image_sync, test_linux_paste_freeze, +test_mac_windows_sync) did not. Doing it here instead makes isolation the +default for every test, present and future, rather than something each new +file has to remember. +""" + +from __future__ import annotations + +import pytest + +from clipsync import config + + +@pytest.fixture(autouse=True) +def _isolate_user_data(tmp_path, monkeypatch): + """Point every user-data path at a per-test temp directory. + + Autouse and unconditional: a test that wants the real paths would have to + opt in explicitly, which nothing should ever do. + """ + data_dir = tmp_path / "_clipsync_data" + data_dir.mkdir(parents=True, exist_ok=True) + + monkeypatch.setattr(config, "HISTORY_FILE", data_dir / "clipsync_history.json", raising=False) + for name, filename in ( + ("SETTINGS_FILE", "settings.json"), + ("LOG_FILE", "clipsync.log"), + ): + if hasattr(config, name): + monkeypatch.setattr(config, name, data_dir / filename, raising=False) + for name in ("APP_DATA_DIR", "SYNC_FOLDER"): + if hasattr(config, name): + monkeypatch.setattr(config, name, data_dir / name.lower(), raising=False) + return data_dir diff --git a/tests/test_audit_fixes.py b/tests/test_audit_fixes.py new file mode 100644 index 0000000..36a7e57 --- /dev/null +++ b/tests/test_audit_fixes.py @@ -0,0 +1,408 @@ +"""Regression tests for the bugs found in the July 2026 audit pass. + +Each test fails against the pre-fix code. Grouped by the defect they pin +down rather than by module, so a future reader can see what behaviour is +being protected and why it mattered. +""" + +from __future__ import annotations + +import time +from pathlib import Path + +import pytest + +from clipsync import clipboard as clipboard_mod +from clipsync import config, syncthing +from clipsync.clipboard import ClipboardSync, _ClipboardFileHandler +from clipsync.syncthing import SyncthingError + + +@pytest.fixture(autouse=True) +def _isolate_history(tmp_path, monkeypatch): + monkeypatch.setattr(config, "HISTORY_FILE", tmp_path / "clipsync_history.json") + + +def _make_sync(tmp_path): + 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)) + return ClipboardSync(settings) + + +# --------------------------------------------------------------------------- +# A transient write failure must not strand a clipboard value forever. +# --------------------------------------------------------------------------- + + +def test_out_tick_retries_text_after_write_oserror(tmp_path, monkeypatch): + """_last_synced is the "already sent" guard. Leaving it set after a failed + write meant every later tick treated the value as synced, so a single + transient OSError silently dropped that clipboard entry for good.""" + sync = _make_sync(tmp_path) + monkeypatch.setattr(sync, "_read_clipboard_image", lambda: None) + monkeypatch.setattr(sync, "_read_clipboard", lambda: "important text") + + calls: list[str] = [] + + def failing_write(text: str) -> None: + calls.append(text) + raise OSError("disk busy") + + monkeypatch.setattr(sync, "_write_file", failing_write) + + sync._out_tick() + assert calls == ["important text"] + # The retry is the whole point: without the rollback this second tick + # returns early because _last_synced still holds the failed value. + sync._out_tick() + assert calls == ["important text", "important text"], "value was never retried after OSError" + + +def test_out_tick_retries_image_after_write_oserror(tmp_path, monkeypatch): + sync = _make_sync(tmp_path) + monkeypatch.setattr(sync, "_read_clipboard_image", lambda: b"\x89PNGfake") + + calls: list[bytes] = [] + + def failing_write(png: bytes) -> None: + calls.append(png) + raise OSError("disk busy") + + monkeypatch.setattr(sync, "_write_image_file", failing_write) + + sync._out_tick() + sync._out_tick() + assert len(calls) == 2, "image was never retried after OSError" + + +def test_out_tick_still_keeps_guard_when_write_succeeds(tmp_path, monkeypatch): + """The rollback must not defeat the ping-pong guard on the happy path.""" + sync = _make_sync(tmp_path) + monkeypatch.setattr(sync, "_read_clipboard_image", lambda: None) + monkeypatch.setattr(sync, "_read_clipboard", lambda: "stable text") + + calls: list[str] = [] + monkeypatch.setattr(sync, "_write_file", lambda t: calls.append(t)) + + sync._out_tick() + sync._out_tick() + assert calls == ["stable text"], "unchanged clipboard was re-sent" + + +# --------------------------------------------------------------------------- +# Text and image updates must not debounce each other. +# --------------------------------------------------------------------------- + + +def test_debounce_is_per_path(tmp_path, monkeypatch): + """One shared deadline meant a clipboard.txt and clipboard.png update + landing within 100ms suppressed each other, so only one was applied.""" + sync = _make_sync(tmp_path) + handler = _ClipboardFileHandler(sync) + + text_path = str(sync.clipboard_file) + image_path = str(sync.clipboard_image_file) + monkeypatch.setattr(handler, "_matches", lambda _p: True) + + handler._dispatch(text_path) + handler._dispatch(image_path) + + queued = [] + while not sync._in_queue.empty(): + queued.append(sync._in_queue.get_nowait()) + + assert text_path in queued, "text update was dropped" + assert image_path in queued, "image update was suppressed by the text debounce" + + +def test_debounce_still_suppresses_repeats_of_same_path(tmp_path, monkeypatch): + """Per-path deadlines must still collapse Syncthing's event storms.""" + sync = _make_sync(tmp_path) + handler = _ClipboardFileHandler(sync) + text_path = str(sync.clipboard_file) + monkeypatch.setattr(handler, "_matches", lambda _p: True) + + for _ in range(5): + handler._dispatch(text_path) + + queued = [] + while not sync._in_queue.empty(): + queued.append(sync._in_queue.get_nowait()) + assert queued == [text_path], "debounce no longer collapses repeat events" + + +def test_debounce_expires(tmp_path, monkeypatch): + sync = _make_sync(tmp_path) + handler = _ClipboardFileHandler(sync) + text_path = str(sync.clipboard_file) + monkeypatch.setattr(handler, "_matches", lambda _p: True) + + handler._dispatch(text_path) + # Reach back past the window rather than sleeping through it. + handler._debounce_until = {k: v - 1.0 for k, v in handler._debounce_until.items()} + handler._dispatch(text_path) + + queued = [] + while not sync._in_queue.empty(): + queued.append(sync._in_queue.get_nowait()) + assert len(queued) == 2, "debounce never expires" + + +# --------------------------------------------------------------------------- +# The heartbeat must not dump a whole image into the log. +# --------------------------------------------------------------------------- + + +def test_heartbeat_truncates_image_bytes(): + """The truncation guard tested isinstance(last, str), so a multi-megabyte + PNG was repr()'d in full into the log on every heartbeat.""" + big_png = b"\x89PNG" + b"A" * 500_000 + rendered = clipboard_mod._truncate_for_log(big_png) + assert len(rendered) < 200, f"image bytes rendered to {len(rendered)} chars" + assert rendered.endswith("...") + + +def test_truncate_for_log_still_truncates_str(): + rendered = clipboard_mod._truncate_for_log("x" * 5000) + assert len(rendered) < 200 + assert rendered.endswith("...") + + +def test_truncate_for_log_leaves_short_values_intact(): + assert clipboard_mod._truncate_for_log("hi") == "'hi'" + assert clipboard_mod._truncate_for_log(None) == "None" + assert clipboard_mod._truncate_for_log(b"hi") == "b'hi'" + + +# --------------------------------------------------------------------------- +# Supply chain: never run a binary we could not verify. +# --------------------------------------------------------------------------- + + +def test_ensure_binary_rejects_binary_that_fails_digest(tmp_path, monkeypatch): + """A replaced binary can print whatever --version string we want to see, + so the version check alone is not evidence of anything.""" + bin_dir = tmp_path / "syncthing" + bin_dir.mkdir(parents=True) + binary = bin_dir / "syncthing" + binary.write_bytes(b"tampered binary") + + monkeypatch.setattr(config, "SYNCTHING_BIN_DIR", bin_dir) + monkeypatch.setattr(config, "syncthing_binary_path", lambda: binary) + monkeypatch.setattr(syncthing, "_binary_version", lambda _b: "v2.0.16") + # Digest file records a different binary. + (bin_dir / "syncthing.sha256").write_text("v2.0.16 " + "0" * 64 + "\n", encoding="utf-8") + + redownloaded = {"hit": False} + + def fake_download(_url): + redownloaded["hit"] = True + raise syncthing.URLError("offline") + + monkeypatch.setattr(syncthing, "_download", fake_download) + + with pytest.raises(SyncthingError): + syncthing.ensure_binary("v2.0.16") + assert redownloaded["hit"], "tampered binary was trusted instead of re-downloaded" + + +def test_ensure_binary_accepts_binary_matching_recorded_digest(tmp_path, monkeypatch): + """The happy path must stay offline-safe: a good binary means no network.""" + bin_dir = tmp_path / "syncthing" + bin_dir.mkdir(parents=True) + binary = bin_dir / "syncthing" + binary.write_bytes(b"the real binary") + + monkeypatch.setattr(config, "SYNCTHING_BIN_DIR", bin_dir) + monkeypatch.setattr(config, "syncthing_binary_path", lambda: binary) + monkeypatch.setattr(syncthing, "_binary_version", lambda _b: "v2.0.16") + digest = syncthing._file_sha256(binary) + (bin_dir / "syncthing.sha256").write_text(f"v2.0.16 {digest}\n", encoding="utf-8") + + def explode(_url): + raise AssertionError("network touched despite a verified binary on disk") + + monkeypatch.setattr(syncthing, "_download", explode) + + assert syncthing.ensure_binary("v2.0.16") == binary + + +def test_ensure_binary_redownloads_when_digest_missing(tmp_path, monkeypatch): + """Binaries installed before digests were recorded must be re-verified, + not grandfathered in on trust.""" + bin_dir = tmp_path / "syncthing" + bin_dir.mkdir(parents=True) + binary = bin_dir / "syncthing" + binary.write_bytes(b"unknown provenance") + + monkeypatch.setattr(config, "SYNCTHING_BIN_DIR", bin_dir) + monkeypatch.setattr(config, "syncthing_binary_path", lambda: binary) + monkeypatch.setattr(syncthing, "_binary_version", lambda _b: "v2.0.16") + + hit = {"n": 0} + + def fake_download(_url): + hit["n"] += 1 + raise syncthing.URLError("offline") + + monkeypatch.setattr(syncthing, "_download", fake_download) + + with pytest.raises(SyncthingError): + syncthing.ensure_binary("v2.0.16") + assert hit["n"] == 1 + + +def test_record_binary_digest_roundtrip(tmp_path, monkeypatch): + bin_dir = tmp_path / "syncthing" + bin_dir.mkdir(parents=True) + binary = bin_dir / "syncthing" + binary.write_bytes(b"payload") + monkeypatch.setattr(config, "SYNCTHING_BIN_DIR", bin_dir) + + syncthing._record_binary_digest(binary, "v2.0.16") + assert syncthing._binary_digest_matches(binary, "v2.0.16") + + binary.write_bytes(b"payload tampered") + assert not syncthing._binary_digest_matches(binary, "v2.0.16") + + +def test_binary_digest_rejects_version_mismatch(tmp_path, monkeypatch): + bin_dir = tmp_path / "syncthing" + bin_dir.mkdir(parents=True) + binary = bin_dir / "syncthing" + binary.write_bytes(b"payload") + monkeypatch.setattr(config, "SYNCTHING_BIN_DIR", bin_dir) + + syncthing._record_binary_digest(binary, "v2.0.16") + assert not syncthing._binary_digest_matches(binary, "v2.0.17") + + +# --------------------------------------------------------------------------- +# Changing the sync folder must repoint every consumer of it. +# --------------------------------------------------------------------------- + + +def test_folder_change_restarts_syncthing_and_file_transfer(tmp_path, monkeypatch): + """Syncthing reads the folder path once, at prepare_home() time, and + FileTransfer schedules its observer at construction. Restarting only + ClipboardSync left both on the old folder, so clipboard.txt was written + where no peer replicated it and sync silently stopped.""" + from clipsync import main as main_mod + + app = object.__new__(main_mod.ClipSyncApp) + app.settings = config.Settings(path=tmp_path / "settings.json") + + events: list[str] = [] + + class _Stub: + def __init__(self, name): + self._name = name + + def stop(self): + events.append(f"{self._name}.stop") + + def start(self): + events.append(f"{self._name}.start") + + app.clipboard = _Stub("clipboard") + app.file_transfer = _Stub("file_transfer") + app.syncthing = _Stub("syncthing") + app._start_syncthing_with_retry = lambda: events.append("syncthing.start") + app._on_file_received = lambda *_a: None + + monkeypatch.setattr(main_mod, "ClipboardSync", lambda _s: _Stub("clipboard")) + monkeypatch.setattr(main_mod, "FileTransfer", lambda _s, on_received=None: _Stub("file_transfer")) + + new_folder = tmp_path / "new_sync" + app._on_folder_changed(str(new_folder)) + + assert new_folder.exists() + assert "syncthing.stop" in events and "syncthing.start" in events, ( + "Syncthing was not restarted, so it still replicates the old folder" + ) + assert "file_transfer.stop" in events and "file_transfer.start" in events, ( + "FileTransfer still watches the old directory" + ) + assert events.index("syncthing.start") < events.index("clipboard.start"), ( + "clipboard restarted before Syncthing was reconfigured" + ) + # prepare_home() reads the folder back out of settings when it re-patches + # config.xml, so the handler must not rely on the UI process having + # persisted it first. + assert app.settings.get("sync_folder") == str(new_folder) + + +# --------------------------------------------------------------------------- +# Concurrency: sets touched from more than one thread. +# --------------------------------------------------------------------------- + + +def test_file_transfer_delivers_each_file_once_under_concurrency(tmp_path): + """watchdog dispatches from a thread pool on Windows, so an unguarded + check-then-add on _seen let two events for one file both pass, delivering + it twice.""" + import threading + + from clipsync import file_transfer as ft_mod + + delivered: list[Path] = [] + deliver_lock = threading.Lock() + barrier = threading.Barrier(8) + + def on_received(path, _sender): + with deliver_lock: + delivered.append(path) + + class _SlowSet(set): + """The real check-then-add is two adjacent bytecodes, so the GIL + almost never splits it and the race will not reproduce by chance. + Sleeping inside the membership test widens the window to what a + thread-pool dispatch can actually hit. Under a correct lock only one + thread is ever inside this at a time, so the result is unchanged.""" + + def __contains__(self, item): + # Resolve membership FIRST, then sleep. Sleeping before the real + # lookup would let the winner add the key while the others are + # parked, so they would see it present and the race would hide + # itself. This ordering parks every thread on the same "not + # present" answer, which is the interleaving being guarded. + result = super().__contains__(item) + time.sleep(0.005) + return result + + handler = ft_mod._FileReceiveHandler(on_received=on_received) + handler._seen = _SlowSet() + incoming = tmp_path / "files" / "peer-host" / "report.pdf" + incoming.parent.mkdir(parents=True) + incoming.write_bytes(b"data") + + def fire(): + barrier.wait() + handler._handle(incoming) + + threads = [threading.Thread(target=fire) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=10) + + assert not any(t.is_alive() for t in threads) + assert len(delivered) == 1, f"file delivered {len(delivered)} times; _seen check-then-add is not atomic" + + +def test_debounce_dict_does_not_grow_unbounded(tmp_path, monkeypatch): + """Per-path deadlines are keyed by filename, and only two filenames ever + match, so the dict cannot grow with event volume.""" + sync = _make_sync(tmp_path) + handler = _ClipboardFileHandler(sync) + monkeypatch.setattr(handler, "_matches", lambda _p: True) + + for i in range(500): + handler._dispatch(str(sync.clipboard_file)) + handler._dispatch(str(sync.clipboard_image_file)) + if i % 100 == 0: + handler._debounce_until = {k: v - 1.0 for k, v in handler._debounce_until.items()} + + assert len(handler._debounce_until) <= 2, f"debounce map grew to {len(handler._debounce_until)} entries" diff --git a/tests/test_clipboard_thread_safety.py b/tests/test_clipboard_thread_safety.py new file mode 100644 index 0000000..5db41e6 --- /dev/null +++ b/tests/test_clipboard_thread_safety.py @@ -0,0 +1,117 @@ +"""The native clipboard is one shared object and is not thread-safe. + +The OUT loop reads it while the IN loop writes it, from two different threads. +On macOS pyperclip's PyObjC backend drives NSPasteboard directly, and a read +overlapping a write crashes the process inside +-[_NSPasteboardOwnersCollection handleOwnershipChange] (SIGSEGV), taking the +tray down with it. These tests fail if the serializing lock is ever dropped. + +The instrumented clipboard calls sleep rather than spin: sleeping releases the +GIL, so an unsynchronized second thread reliably lands inside the window. A +busy loop does not yield often enough to expose the race. +""" + +from __future__ import annotations + +import threading +import time + +import pytest + +from clipsync import clipboard as clipboard_mod +from clipsync import config +from clipsync.clipboard import ClipboardSync + +# Long enough to force a GIL handoff, short enough to keep the suite fast. +_WINDOW = 0.002 + + +@pytest.fixture(autouse=True) +def _isolate_history(tmp_path, monkeypatch): + """ClipboardHistory binds to config.HISTORY_FILE at construction. Without + this the tests would read and rewrite the real clipboard history.""" + monkeypatch.setattr(config, "HISTORY_FILE", tmp_path / "clipsync_history.json") + + +def _make_sync(tmp_path): + 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)) + return ClipboardSync(settings) + + +class _OverlapDetector: + """Records whether two instrumented clipboard calls were ever in flight at + the same time.""" + + def __init__(self) -> None: + self.inside = 0 + self.overlapped = False + self._guard = threading.Lock() + + def instrument(self, result): + def fn(*_args, **_kwargs): + with self._guard: + self.inside += 1 + if self.inside > 1: + self.overlapped = True + try: + time.sleep(_WINDOW) + return result + finally: + with self._guard: + self.inside -= 1 + + return fn + + +def _run(targets, timeout=30): + threads = [threading.Thread(target=t) for t in targets] + for t in threads: + t.start() + for t in threads: + t.join(timeout=timeout) + assert not any(t.is_alive() for t in threads), "clipboard access deadlocked" + + +def test_clipboard_reads_and_writes_never_overlap(tmp_path, monkeypatch): + sync = _make_sync(tmp_path) + detector = _OverlapDetector() + + monkeypatch.setattr(clipboard_mod.pyperclip, "paste", detector.instrument("text")) + monkeypatch.setattr(clipboard_mod.pyperclip, "copy", detector.instrument(None)) + # The Linux in-process owner would bypass the pyperclip write path. + sync._clipboard_owner = None + + _run( + [ + lambda: [sync._read_clipboard() for _ in range(40)], + lambda: [sync._write_clipboard("value") for _ in range(40)], + ] + ) + + assert not detector.overlapped, ( + "a clipboard read overlapped a write; the native clipboard is not thread-safe and this segfaults on macOS" + ) + + +def test_image_clipboard_access_is_serialized_with_text(tmp_path, monkeypatch): + """Image and text paths touch the same pasteboard, so they must share one + lock rather than each holding a private one.""" + sync = _make_sync(tmp_path) + detector = _OverlapDetector() + + monkeypatch.setattr(clipboard_mod.pyperclip, "paste", detector.instrument("text")) + monkeypatch.setattr(clipboard_mod, "_read_image_from_system_clipboard", detector.instrument(None)) + monkeypatch.setattr(clipboard_mod, "_write_image_to_system_clipboard", detector.instrument(True)) + + _run( + [ + lambda: [sync._read_clipboard() for _ in range(30)], + lambda: [sync._read_clipboard_image() for _ in range(30)], + lambda: [sync._write_clipboard_image(b"png") for _ in range(30)], + ] + ) + + assert not detector.overlapped, "image and text clipboard access must share one lock" diff --git a/tests/test_file_transfer_encryption.py b/tests/test_file_transfer_encryption.py new file mode 100644 index 0000000..dda5a24 --- /dev/null +++ b/tests/test_file_transfer_encryption.py @@ -0,0 +1,223 @@ +"""Sent files must honour the at-rest passphrase, like the clipboard does. + +send() was a bare shutil.copy2, so enabling encryption protected clipboard +text but left every sent file sitting in the shared folder as plaintext -- +readable by anything with access to that directory and replicated that way to +every peer. copy2 also preserves the source mode, so a world-readable original +stayed world-readable inside the folder. + +Fernet is all-in-memory, so files use a chunked streaming format instead: +CSENCF magic, salt, then length-prefixed Fernet tokens over +(chunk index || data), terminated by an authenticated empty chunk. The index +blocks reordering and dropping; the terminator catches truncation, which would +otherwise decrypt cleanly to a prefix. + +Tests use random bytes as file payloads. Nothing here reads real user data. +""" + +from __future__ import annotations + +import hashlib +import os +import sys + +import pytest + +from clipsync import config +from clipsync.crypto import ( + StreamDecryptError, + decrypt_file, + encrypt_file, + is_encrypted_file, +) +from clipsync.file_transfer import ENCRYPTED_SUFFIX, FileTransfer + + +class _Settings: + def __init__(self, **kw): + self._d = dict(kw) + + def get(self, key, default=None): + return self._d.get(key, default) + + +def _digest(p) -> str: + return hashlib.sha256(p.read_bytes()).hexdigest() + + +# --------------------------------------------------------------------------- +# Streaming primitive +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "size", + [0, 1, 1024, 1024 * 1024, 1024 * 1024 + 1, 3_500_000], + ids=["empty", "tiny", "1k", "exactly-one-chunk", "chunk-plus-one", "multi-chunk"], +) +def test_encrypt_decrypt_roundtrip_is_exact(tmp_path, size): + src = tmp_path / "in.bin" + src.write_bytes(os.urandom(size)) + enc, out = tmp_path / "e", tmp_path / "o" + + encrypt_file(src, enc, "pw") + assert is_encrypted_file(enc) + decrypt_file(enc, out, "pw") + assert _digest(out) == _digest(src) + + +def test_ciphertext_does_not_contain_the_plaintext(tmp_path): + src = tmp_path / "in.bin" + marker = b"TOP-SECRET-MARKER-" + os.urandom(16) + src.write_bytes(marker * 100) + enc = tmp_path / "e" + encrypt_file(src, enc, "pw") + assert marker not in enc.read_bytes(), "plaintext leaked into the encrypted file" + + +def test_wrong_passphrase_is_rejected_and_leaves_no_partial(tmp_path): + src = tmp_path / "in.bin" + src.write_bytes(os.urandom(2_000_000)) + enc, out = tmp_path / "e", tmp_path / "o" + encrypt_file(src, enc, "right") + + with pytest.raises(StreamDecryptError): + decrypt_file(enc, out, "wrong") + assert not out.exists(), "a partial plaintext was left behind" + + +def test_truncation_at_a_chunk_boundary_is_detected(tmp_path): + """Without the EOF marker a truncated file decrypts cleanly to a prefix, + so the user silently gets a corrupt file that looks complete.""" + src = tmp_path / "in.bin" + src.write_bytes(os.urandom(2_500_000)) + enc, out = tmp_path / "e", tmp_path / "o" + encrypt_file(src, enc, "pw") + + raw = enc.read_bytes() + # Drop the trailing EOF chunk: walk the length prefixes and stop early. + pos = 7 + 16 # magic + salt + boundaries = [] + while pos + 4 <= len(raw): + n = int.from_bytes(raw[pos : pos + 4], "big") + pos += 4 + n + boundaries.append(pos) + assert len(boundaries) >= 2 + (tmp_path / "trunc").write_bytes(raw[: boundaries[-2]]) + + with pytest.raises(StreamDecryptError): + decrypt_file(tmp_path / "trunc", out, "pw") + assert not out.exists() + + +def test_reordered_chunks_are_detected(tmp_path): + src = tmp_path / "in.bin" + src.write_bytes(os.urandom(2_500_000)) + enc, out = tmp_path / "e", tmp_path / "o" + encrypt_file(src, enc, "pw") + + raw = enc.read_bytes() + header, pos, chunks = raw[: 7 + 16], 7 + 16, [] + while pos + 4 <= len(raw): + n = int.from_bytes(raw[pos : pos + 4], "big") + chunks.append(raw[pos : pos + 4 + n]) + pos += 4 + n + assert len(chunks) >= 3 + chunks[0], chunks[1] = chunks[1], chunks[0] + (tmp_path / "reordered").write_bytes(header + b"".join(chunks)) + + with pytest.raises(StreamDecryptError, match="reordered|dropped|corrupt"): + decrypt_file(tmp_path / "reordered", out, "pw") + + +def test_plaintext_file_is_not_mistaken_for_encrypted(tmp_path): + p = tmp_path / "plain.txt" + p.write_bytes(b"just a normal file") + assert not is_encrypted_file(p) + + +# --------------------------------------------------------------------------- +# send() integration +# --------------------------------------------------------------------------- + + +def _transfer(tmp_path, passphrase): + settings = _Settings(sync_folder=str(tmp_path / "sync"), encryption_passphrase=passphrase) + return FileTransfer(settings, on_received=lambda *_a: None) + + +def test_send_encrypts_when_a_passphrase_is_set(tmp_path): + payload = os.urandom(50_000) + src = tmp_path / "report.pdf" + src.write_bytes(payload) + + dest = _transfer(tmp_path, "secret").send(src) + + assert dest.name.endswith(ENCRYPTED_SUFFIX), f"sent file is not marked encrypted: {dest.name}" + assert is_encrypted_file(dest) + assert payload[:64] not in dest.read_bytes(), "plaintext reached the shared folder" + + out = tmp_path / "recovered.pdf" + decrypt_file(dest, out, "secret") + assert _digest(out) == _digest(src) + + +def test_send_stays_plaintext_when_no_passphrase(tmp_path): + """Without encryption configured the old behaviour is preserved, so peers + on older builds keep working.""" + src = tmp_path / "note.txt" + src.write_bytes(b"hello") + dest = _transfer(tmp_path, "").send(src) + + assert not dest.name.endswith(ENCRYPTED_SUFFIX) + assert dest.read_bytes() == b"hello" + + +def test_sent_file_is_not_world_readable(tmp_path): + """copy2 preserved the source mode, so a 0644 original stayed 0644 in the + shared folder.""" + if sys.platform == "win32": + pytest.skip("POSIX permission bits do not apply on Windows") + src = tmp_path / "open.txt" + src.write_bytes(b"data") + os.chmod(src, 0o644) + + for passphrase in ("", "secret"): + dest = _transfer(tmp_path, passphrase).send(src) + mode = dest.stat().st_mode & 0o777 + assert mode == 0o600, f"passphrase={passphrase!r}: shared copy is {oct(mode)}" + + +def test_failed_encryption_leaves_no_partial_in_the_shared_folder(tmp_path, monkeypatch): + src = tmp_path / "big.bin" + src.write_bytes(os.urandom(1000)) + + import clipsync.file_transfer as ft + + def boom(*_a, **_k): + raise OSError("disk full") + + monkeypatch.setattr(ft, "encrypt_file", boom) + ftr = _transfer(tmp_path, "secret") + with pytest.raises(OSError): + ftr.send(src) + + leftovers = list((ftr.files_dir).rglob("*")) + files = [p for p in leftovers if p.is_file()] + assert not files, f"partial file left in the shared folder: {files}" + + +def test_encrypted_send_roundtrips_a_large_file_in_bounded_memory(tmp_path): + """The whole reason for the chunked format: Fernet in one shot would hold + the entire file in RAM.""" + src = tmp_path / "big.bin" + src.write_bytes(os.urandom(5_000_000)) + dest = _transfer(tmp_path, "secret").send(src) + out = tmp_path / "out.bin" + decrypt_file(dest, out, "secret") + assert _digest(out) == _digest(src) + + +def test_set_file_permissions_is_available_to_file_transfer(): + """send() relies on it for both paths.""" + assert callable(config.set_file_permissions) diff --git a/tests/test_history_no_clobber.py b/tests/test_history_no_clobber.py new file mode 100644 index 0000000..2202914 --- /dev/null +++ b/tests/test_history_no_clobber.py @@ -0,0 +1,203 @@ +"""An unreadable history file must never be silently overwritten. + +_load() logged "Failed to decrypt clipboard history" and returned with an +empty in-memory list. Nothing marked the file as untouchable, so the very +next add_entry() persisted that empty list straight over it and every stored +entry was gone. Measured on the pre-fix code: a 5-entry encrypted history +became a 1-entry file, unrecoverable even with the correct passphrase. + +The sync file has been protected from exactly this since the #21 audit, by +ClipboardSync._refuse_if_unreadable_ciphertext. The history file, which holds +the same clipboard text, was not. + +These tests only ever count entries. They never assert on clipboard text. +""" + +from __future__ import annotations + +import json + +import pytest + +from clipsync import config +from clipsync.crypto import encrypt, is_encrypted +from clipsync.history import ClipboardHistory + + +class _Settings: + def __init__(self, **kw): + self._d = {"history_enabled": True, "history_max_items": 50} + self._d.update(kw) + + def get(self, key, default=None): + return self._d.get(key, default) + + def set(self, key, value): + self._d[key] = value + + +@pytest.fixture(autouse=True) +def _isolate(tmp_path, monkeypatch): + monkeypatch.setattr(config, "HISTORY_FILE", tmp_path / "hist.json") + return tmp_path + + +def _seed(passphrase: str, count: int) -> ClipboardHistory: + h = ClipboardHistory(_Settings(encryption_passphrase=passphrase)) + for i in range(count): + h.add_entry(f"entry-{i}") + return h + + +def test_passphrase_mismatch_does_not_destroy_history(_isolate): + original = None + _seed("right", 5) + original = config.HISTORY_FILE.read_bytes() + assert is_encrypted(original) + + # Reopen with the wrong passphrase, then take a new clipboard entry — + # which is what clipboard.py does on every OUT tick. + h2 = ClipboardHistory(_Settings(encryption_passphrase="wrong")) + h2.add_entry("something new") + + preserved = [p for p in _isolate.iterdir() if "unreadable" in p.name] + assert preserved, "the unreadable history was not preserved anywhere" + assert preserved[0].read_bytes() == original, "preserved copy is not byte-identical" + + +def test_preserved_history_is_recoverable_with_the_right_passphrase(_isolate, monkeypatch): + _seed("right", 5) + ClipboardHistory(_Settings(encryption_passphrase="wrong")).add_entry("new") + + preserved = [p for p in _isolate.iterdir() if "unreadable" in p.name][0] + monkeypatch.setattr(config, "HISTORY_FILE", preserved) + recovered = ClipboardHistory(_Settings(encryption_passphrase="right")) + assert recovered.count() == 5, "the preserved file did not decrypt back to the original entries" + + +def test_encrypted_history_with_no_passphrase_is_not_overwritten(_isolate): + """Clearing the passphrase in settings must not wipe an encrypted history.""" + _seed("right", 3) + original = config.HISTORY_FILE.read_bytes() + + h = ClipboardHistory(_Settings(encryption_passphrase="")) + h.add_entry("plaintext era") + + preserved = [p for p in _isolate.iterdir() if "unreadable" in p.name] + assert preserved, "encrypted history was overwritten once the passphrase was cleared" + assert preserved[0].read_bytes() == original + + +def test_unreadable_and_unmovable_file_is_left_alone(_isolate, monkeypatch): + """If it cannot even be moved aside, refuse to write rather than clobber.""" + _seed("right", 4) + original = config.HISTORY_FILE.read_bytes() + + def refuse_move(self, target): + raise OSError("read-only filesystem") + + monkeypatch.setattr(type(config.HISTORY_FILE), "replace", refuse_move) + h = ClipboardHistory(_Settings(encryption_passphrase="wrong")) + h.add_entry("new entry") + + assert config.HISTORY_FILE.read_bytes() == original, "clobbered a file it could not move aside" + assert h.count() == 1, "in-memory history should still work for this session" + + +def test_normal_operation_still_persists(_isolate): + """The guard must not break the ordinary path.""" + h = _seed("right", 3) + assert h.count() == 3 + reopened = ClipboardHistory(_Settings(encryption_passphrase="right")) + assert reopened.count() == 3 + + +def test_plaintext_history_still_loads_and_persists(_isolate): + h = ClipboardHistory(_Settings(encryption_passphrase="")) + h.add_entry("a") + h.add_entry("b") + assert not is_encrypted(config.HISTORY_FILE.read_bytes()) + assert ClipboardHistory(_Settings(encryption_passphrase="")).count() == 2 + + +def test_corrupt_plaintext_json_is_not_treated_as_encrypted(_isolate): + """A truncated plaintext file is a different failure: it carries no CSENC + marker, so it is not quarantined, but it must not crash the app either.""" + config.HISTORY_FILE.write_bytes(b'{"entries": [') + h = ClipboardHistory(_Settings(encryption_passphrase="")) + assert h.count() == 0 + h.add_entry("recovered") + assert h.count() == 1 + + +def test_history_file_is_not_world_readable(_isolate): + """It holds clipboard text, so it must not be readable by other local users + at any point — including between the write and the chmod.""" + import sys + + if sys.platform == "win32": + pytest.skip("POSIX permission bits do not apply on Windows") + _seed("right", 2) + mode = config.HISTORY_FILE.stat().st_mode & 0o777 + assert mode == 0o600, f"history file is {oct(mode)}" + + +def test_future_version_ciphertext_is_preserved(_isolate): + """A payload written by a NEWER build carries CSENC but an unknown version + byte. is_encrypted() matches it deliberately; it must be preserved, not + overwritten, so downgrading does not destroy the newer machine's data.""" + future = b"CSENC\x09" + b"\x00" * 32 + config.HISTORY_FILE.write_bytes(future) + + h = ClipboardHistory(_Settings(encryption_passphrase="right")) + h.add_entry("new") + + preserved = [p for p in _isolate.iterdir() if "unreadable" in p.name] + assert preserved, "future-version ciphertext was destroyed" + assert preserved[0].read_bytes() == future + + +def test_disabled_history_does_not_clobber_existing_file(_isolate): + _seed("right", 3) + original = config.HISTORY_FILE.read_bytes() + h = ClipboardHistory(_Settings(encryption_passphrase="wrong", history_enabled=False)) + h.add_entry("ignored") + surviving = config.HISTORY_FILE.exists() and config.HISTORY_FILE.read_bytes() == original + preserved = [p for p in _isolate.iterdir() if "unreadable" in p.name] + assert surviving or preserved, "history was destroyed while disabled" + + +def test_plaintext_history_is_encrypted_once_a_passphrase_is_set(_isolate): + """Turning encryption on must actually protect what is already stored.""" + h = ClipboardHistory(_Settings(encryption_passphrase="")) + h.add_entry("before") + assert not is_encrypted(config.HISTORY_FILE.read_bytes()) + + h2 = ClipboardHistory(_Settings(encryption_passphrase="now-secret")) + assert h2.count() == 1, "existing plaintext entries were lost when encryption was enabled" + h2.add_entry("after") + assert is_encrypted(config.HISTORY_FILE.read_bytes()), "history stayed plaintext after enabling encryption" + + +def test_entry_cap_is_honoured(_isolate): + h = ClipboardHistory(_Settings(encryption_passphrase="right", history_max_items=5)) + for i in range(12): + h.add_entry(f"e{i}") + assert h.count() == 5 + raw = config.HISTORY_FILE.read_bytes() + assert is_encrypted(raw) + + +def test_load_of_valid_but_empty_file(_isolate): + config.HISTORY_FILE.write_bytes(json.dumps({"entries": []}).encode()) + assert ClipboardHistory(_Settings(encryption_passphrase="")).count() == 0 + + +def test_encrypt_roundtrip_used_by_history(_isolate): + """Sanity check on the primitive the guard depends on.""" + from clipsync.crypto import decrypt + + blob = encrypt(b'{"entries": []}', "pw") + assert is_encrypted(blob) + assert decrypt(blob, "pw") == b'{"entries": []}' + assert decrypt(blob, "other") is None diff --git a/tests/test_log_mirror_and_settings.py b/tests/test_log_mirror_and_settings.py new file mode 100644 index 0000000..5ea8345 --- /dev/null +++ b/tests/test_log_mirror_and_settings.py @@ -0,0 +1,200 @@ +"""Log mirroring is opt-in, and malformed settings must not kill startup. + +LogMirror copied this device's log into the shared folder every 10s. The +folder is replicated to every paired device, so the log (hostnames, device +IDs, file names, error traces) went to all of them. It was always on, had no +setting, and was not mentioned in the README. No clipboard text is ever +logged -- every clipboard log line records a character count -- but none of +that is visible from the tray. + +Separately, ClipboardHistory parsed settings with a bare int(), and it is +constructed during startup, so one malformed value in settings.json took the +whole app down before the tray appeared. +""" + +from __future__ import annotations + +import pytest + +from clipsync import config +from clipsync.debug import LogMirror +from clipsync.history import ClipboardHistory, _coerce_bool, _coerce_int + + +class _Settings: + def __init__(self, **kw): + self._d = dict(kw) + + def get(self, key, default=None): + return self._d.get(key, default) + + def set(self, key, value): + self._d[key] = value + + +@pytest.fixture +def synced(tmp_path, monkeypatch): + log_file = tmp_path / "clipsync.log" + log_file.write_text("2026-08-03 10:00:00 [INFO] clipsync.main: started\n") + monkeypatch.setattr(config, "LOG_FILE", log_file) + return tmp_path + + +def _mirror(synced, **kw): + settings = _Settings(sync_folder=str(synced / "sync"), **kw) + return LogMirror(settings), synced / "sync" / "debug" + + +# --------------------------------------------------------------------------- +# Mirroring is opt-in +# --------------------------------------------------------------------------- + + +def test_mirror_is_off_by_default(synced): + """The default must not publish anything to the shared folder.""" + mirror, debug_dir = _mirror(synced) + mirror._tick() + assert not debug_dir.exists() or not list(debug_dir.glob("*.log")), ( + "log was mirrored into the shared folder without being asked" + ) + + +def test_mirror_publishes_when_explicitly_enabled(synced): + mirror, debug_dir = _mirror(synced, debug_log_mirror=True) + mirror._tick() + published = list(debug_dir.glob("*.log")) + assert published, "opting in did not publish the log" + assert "started" in published[0].read_text() + + +def test_disabling_retracts_an_already_published_log(synced): + """Switching it off has to remove what is already in the folder, or the + last copy keeps being replicated to every peer indefinitely.""" + settings = _Settings(sync_folder=str(synced / "sync"), debug_log_mirror=True) + mirror = LogMirror(settings) + debug_dir = synced / "sync" / "debug" + + mirror._tick() + assert list(debug_dir.glob("*.log")) + + settings.set("debug_log_mirror", False) + mirror._tick() + assert not list(debug_dir.glob("*.log")), "disabling left the published log behind" + + +def test_first_tick_cleans_up_a_log_left_by_an_older_build(synced): + """Older builds always mirrored. After upgrading, the stale file must be + retracted rather than left replicating forever.""" + debug_dir = synced / "sync" / "debug" + debug_dir.mkdir(parents=True) + from clipsync.debug import _safe_hostname + + stale = debug_dir / f"{_safe_hostname()}.log" + stale.write_text("old always-on mirror output\n") + + mirror, _ = _mirror(synced) # default: disabled + mirror._tick() + assert not stale.exists(), "stale mirror from an older build was not cleaned up" + + +def test_toggle_applies_without_restart(synced): + settings = _Settings(sync_folder=str(synced / "sync"), debug_log_mirror=False) + mirror = LogMirror(settings) + debug_dir = synced / "sync" / "debug" + + mirror._tick() + assert not list(debug_dir.glob("*.log")) + settings.set("debug_log_mirror", True) + mirror._tick() + assert list(debug_dir.glob("*.log")), "enabling did not take effect until restart" + + +def test_peer_logs_are_never_removed(synced): + """Retraction must only touch our own file.""" + settings = _Settings(sync_folder=str(synced / "sync"), debug_log_mirror=False) + debug_dir = synced / "sync" / "debug" + debug_dir.mkdir(parents=True) + peer = debug_dir / "someone-elses-laptop.log" + peer.write_text("peer output\n") + + LogMirror(settings)._tick() + assert peer.exists(), "retraction deleted a peer's mirrored log" + + +# --------------------------------------------------------------------------- +# Settings coercion +# --------------------------------------------------------------------------- + + +def test_malformed_max_items_does_not_crash_startup(tmp_path, monkeypatch): + monkeypatch.setattr(config, "HISTORY_FILE", tmp_path / "h.json") + h = ClipboardHistory(_Settings(history_max_items="not-a-number")) + assert h.get_max_items() == 50, "bad value should fall back to the default" + + +def test_stringified_numbers_are_accepted(tmp_path, monkeypatch): + monkeypatch.setattr(config, "HISTORY_FILE", tmp_path / "h.json") + h = ClipboardHistory(_Settings(history_max_items="25")) + assert h.get_max_items() == 25 + + +def test_stringified_false_disables_history(tmp_path, monkeypatch): + """bool("false") is True, which is exactly the trap a JSON-stringified + setting falls into: history would stay on after the user turned it off.""" + monkeypatch.setattr(config, "HISTORY_FILE", tmp_path / "h.json") + h = ClipboardHistory(_Settings(history_enabled="false")) + assert h.is_enabled() is False + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + (10, 10), + ("10", 10), + (" 10 ", 10), + (0, 5), + (-3, 5), + ("abc", 5), + (None, 5), + (True, 5), + (3.7, 5), + ], +) +def test_coerce_int(value, expected): + assert _coerce_int(value, 5) == expected + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + (True, True), + (False, False), + ("true", True), + ("TRUE", True), + ("false", False), + ("0", False), + ("", False), + ("nonsense", True), + (None, True), + ], +) +def test_coerce_bool(value, expected): + assert _coerce_bool(value, True) is expected + + +def test_malformed_auto_clear_does_not_silently_keep_entries_forever(tmp_path, monkeypatch): + """0 means never expire, so an unparseable value must fall back to 0 rather + than to some arbitrary retention the user did not choose.""" + monkeypatch.setattr(config, "HISTORY_FILE", tmp_path / "h.json") + h = ClipboardHistory(_Settings(history_auto_clear_minutes="oops")) + assert h._auto_clear_minutes() == 0 + h2 = ClipboardHistory(_Settings(history_auto_clear_minutes="30")) + assert h2._auto_clear_minutes() == 30 + + +def test_history_still_works_with_sane_settings(tmp_path, monkeypatch): + monkeypatch.setattr(config, "HISTORY_FILE", tmp_path / "h.json") + h = ClipboardHistory(_Settings(history_enabled=True, history_max_items=3)) + for i in range(6): + h.add_entry(f"e{i}") + assert h.count() == 3 diff --git a/tests/test_pinned_hashes.py b/tests/test_pinned_hashes.py new file mode 100644 index 0000000..2275a3f --- /dev/null +++ b/tests/test_pinned_hashes.py @@ -0,0 +1,107 @@ +"""The pinned Syncthing hash manifest is the trust anchor for the binary. + +Runtime signature verification needs gpg on PATH, which stock Windows and +macOS do not have, so for most users it degraded to hash-only with the hash +coming from the same origin as the download. These tests pin down the +manifest's role: it must be used, it must be complete for the version we +ship, and it must never silently disappear. +""" + +from __future__ import annotations + +import hashlib + +import pytest + +from clipsync import config, syncthing +from clipsync._syncthing_hashes import SYNCTHING_PINNED_SHA256 +from clipsync.syncthing import SyncthingError + + +def test_pinned_manifest_covers_the_shipped_version(): + """A version bump that forgets to regenerate the manifest would silently + fall back to the network path this change exists to remove.""" + version = config.SYNCTHING_VERSION + v = version if version.startswith("v") else f"v{version}" + assert v in SYNCTHING_PINNED_SHA256, f"No pinned hashes for {v}. Run: python tools/refresh_syncthing_hashes.py {v}" + + +def test_pinned_manifest_covers_this_platform(): + name = syncthing._archive_filename(config.SYNCTHING_VERSION) + v = config.SYNCTHING_VERSION + v = v if v.startswith("v") else f"v{v}" + assert name in SYNCTHING_PINNED_SHA256[v], f"{name} missing from the pinned manifest" + + +def test_pinned_hashes_are_wellformed_sha256(): + for version, archives in SYNCTHING_PINNED_SHA256.items(): + assert archives, f"{version} has an empty archive map" + for name, digest in archives.items(): + assert len(digest) == 64, f"{name}: {digest!r} is not a sha256" + assert all(c in "0123456789abcdef" for c in digest), f"{name}: {digest!r} not lowercase hex" + + +def test_verify_uses_pinned_hash_without_touching_the_network(monkeypatch): + """The whole point: for the shipped version, verification must not depend + on a request that can be blocked or poisoned.""" + name = syncthing._archive_filename(config.SYNCTHING_VERSION) + v = config.SYNCTHING_VERSION + v = v if v.startswith("v") else f"v{v}" + expected = SYNCTHING_PINNED_SHA256[v][name] + + def explode(*_a, **_k): + raise AssertionError("network was used despite a pinned hash being available") + + monkeypatch.setattr(syncthing, "_fetch_official_sha256sums", explode) + monkeypatch.setattr(syncthing, "_download", explode) + + class _FakeDigest: + def hexdigest(self): + return expected + + monkeypatch.setattr(hashlib, "sha256", lambda _d: _FakeDigest()) + syncthing._verify_archive_hash(b"pretend archive", config.SYNCTHING_VERSION) + + +def test_tampered_archive_still_rejected_against_pinned_hash(monkeypatch): + monkeypatch.setattr(syncthing, "_fetch_official_sha256sums", lambda _v: {}) + with pytest.raises(SyncthingError, match="hash mismatch"): + syncthing._verify_archive_hash(b"tampered bytes", config.SYNCTHING_VERSION) + + +def test_unpinned_version_falls_back_to_signed_fetch(monkeypatch): + """A non-default version has no pinned entry, so it must still go through + the signed-sums path rather than being waved through.""" + archive = b"archive bytes" + name = syncthing._archive_filename("v9.9.9") + monkeypatch.setattr( + syncthing, + "_fetch_official_sha256sums", + lambda _v: {name: hashlib.sha256(archive).hexdigest()}, + ) + syncthing._verify_archive_hash(archive, "v9.9.9") + + +def test_unpinned_version_with_unreachable_sums_is_fatal(monkeypatch): + def boom(_v): + raise SyncthingError("Refusing to extract an unverified Syncthing binary") + + monkeypatch.setattr(syncthing, "_fetch_official_sha256sums", boom) + with pytest.raises(SyncthingError, match="Refusing to extract"): + syncthing._verify_archive_hash(b"bytes", "v9.9.9") + + +def test_manifest_covers_every_platform_we_build_for(): + """A partial manifest would send some platforms down the network fallback + while leaving this machine's tests green.""" + v = config.SYNCTHING_VERSION + v = v if v.startswith("v") else f"v{v}" + names = set(SYNCTHING_PINNED_SHA256[v]) + for expected in ( + f"syncthing-macos-arm64-{v}.zip", + f"syncthing-macos-amd64-{v}.zip", + f"syncthing-windows-amd64-{v}.zip", + f"syncthing-linux-amd64-{v}.tar.gz", + f"syncthing-linux-arm64-{v}.tar.gz", + ): + assert expected in names, f"{expected} missing from the pinned manifest" diff --git a/tests/test_syncthing_hash.py b/tests/test_syncthing_hash.py index 6561a23..5451645 100644 --- a/tests/test_syncthing_hash.py +++ b/tests/test_syncthing_hash.py @@ -52,12 +52,18 @@ def fake_download(url: str) -> bytes: assert seen_urls == ["https://github.com/syncthing/syncthing/releases/download/v2.0.16/sha256sum.txt.asc"] -def test_fetch_official_sha256sums_returns_empty_on_network_error() -> None: +def test_fetch_official_sha256sums_raises_on_network_error() -> None: + """Fail closed. The archive is executed after extraction, so a fetch we + could not complete must abort rather than silently skip verification: + an attacker able to drop this one request would otherwise downgrade us + to running an unverified binary.""" from urllib.error import URLError - with patch("clipsync.syncthing._download", side_effect=URLError("offline")): - sums = syncthing._fetch_official_sha256sums("v2.0.16") - assert sums == {} + with ( + patch("clipsync.syncthing._download", side_effect=URLError("offline")), + pytest.raises(SyncthingError, match="Refusing to extract"), + ): + syncthing._fetch_official_sha256sums("v2.0.16") def test_fetch_official_sha256sums_ignores_malformed_lines(monkeypatch) -> None: @@ -146,36 +152,48 @@ def test_verify_release_signature_falls_back_when_gpg_absent(monkeypatch, caplog def test_verify_archive_hash_succeeds_on_match(monkeypatch) -> None: archive = b"the bytes of a syncthing archive" expected = hashlib.sha256(archive).hexdigest() - name = "syncthing-linux-amd64-v2.0.16.tar.gz" + # Derive the name for the platform the test is running on. Hardcoding the + # Linux asset made this pass vacuously everywhere else: the lookup missed, + # verification was skipped, and the test asserted nothing. + # + # Uses a version with no pinned manifest entry, so this still exercises + # the fetched-sums path. The pinned path is covered in test_pinned_hashes. + name = syncthing._archive_filename("v9.9.9") monkeypatch.setattr(syncthing, "_fetch_official_sha256sums", lambda _v: {name: expected}) # Should not raise. - syncthing._verify_archive_hash(archive, "v2.0.16") + syncthing._verify_archive_hash(archive, "v9.9.9") def test_verify_archive_hash_raises_on_mismatch(monkeypatch) -> None: archive = b"the bytes of a syncthing archive" - name = "syncthing-linux-amd64-v2.0.16.tar.gz" + name = syncthing._archive_filename("v2.0.16") monkeypatch.setattr(syncthing, "_fetch_official_sha256sums", lambda _v: {name: "0" * 64}) with pytest.raises(SyncthingError, match="hash mismatch"): syncthing._verify_archive_hash(archive, "v2.0.16") -def test_verify_archive_hash_skips_when_platform_absent(monkeypatch) -> None: - """If sha256sum.txt has no entry for our platform, verification must - be skipped (logged) rather than fail. Otherwise an unusual platform - would be unable to install even when Syncthing ships a binary for it.""" +def test_verify_archive_hash_raises_when_platform_absent(monkeypatch) -> None: + """A missing entry for our platform is indistinguishable from one an + attacker stripped, so it must abort rather than skip verification.""" archive = b"some bytes" monkeypatch.setattr(syncthing, "_fetch_official_sha256sums", lambda _v: {}) - # Should not raise. - syncthing._verify_archive_hash(archive, "v2.0.16") + with pytest.raises(SyncthingError, match="Refusing to extract"): + syncthing._verify_archive_hash(archive, "v2.0.16") -def test_verify_archive_hash_skips_when_fetch_fails(monkeypatch) -> None: +def test_verify_archive_hash_propagates_fetch_failure(monkeypatch) -> None: + """A fetch failure inside _fetch_official_sha256sums must reach the + caller, not be swallowed into a skipped check.""" archive = b"some bytes" - monkeypatch.setattr(syncthing, "_fetch_official_sha256sums", lambda _v: {}) - syncthing._verify_archive_hash(archive, "v2.0.16") + + def _boom(_v): + raise SyncthingError("Refusing to extract an unverified Syncthing binary") + + monkeypatch.setattr(syncthing, "_fetch_official_sha256sums", _boom) + with pytest.raises(SyncthingError, match="Refusing to extract"): + syncthing._verify_archive_hash(archive, "v2.0.16") def test_archive_filename_matches_release_naming() -> None: diff --git a/tools/refresh_syncthing_hashes.py b/tools/refresh_syncthing_hashes.py new file mode 100644 index 0000000..a405bdc --- /dev/null +++ b/tools/refresh_syncthing_hashes.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python3 +"""Regenerate ``clipsync/_syncthing_hashes.py`` for a Syncthing version. + +Run this whenever ``config.SYNCTHING_VERSION`` is bumped: + + python tools/refresh_syncthing_hashes.py v2.0.17 + +It fetches Syncthing's ``sha256sum.txt.asc`` and hands it to the app's own +:func:`clipsync.syncthing._verify_release_signature`, so it writes nothing +unless the PGP signature verifies against the release key pinned in +``clipsync/_release_key.py``. + +This needs gpg on PATH. That is the point of doing it here: verification +happens once, on a maintainer's machine, at release-prep time, instead of on +every end user's machine where gpg usually does not exist and the check +quietly degraded to hash-only. +""" + +from __future__ import annotations + +import json +import re +import shutil +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from clipsync import syncthing # noqa: E402 +from clipsync._release_key import SYNCTHING_RELEASE_FINGERPRINT # noqa: E402 + +_ARCHIVE_RE = re.compile(r"^syncthing-(windows|macos|linux)-[a-z0-9]+-v[\d.]+\.(zip|tar\.gz)$") +_TARGET = Path(__file__).resolve().parent.parent / "clipsync" / "_syncthing_hashes.py" + +_HEADER = '''"""SHA-256 hashes of the pinned Syncthing release archives. + +Generated by ``tools/refresh_syncthing_hashes.py``, which fetches +Syncthing's ``sha256sum.txt.asc`` and refuses to emit anything unless the +PGP signature verifies against the pinned release key in +``_release_key.py``. Do not hand-edit: regenerate. + +Why these live in the source tree rather than being fetched at runtime: +verifying the signature at runtime needs gpg on PATH, and most end users +do not have it (stock Windows and macOS ship without it), so that check +silently degraded to hash-only. Pinning moves the trust anchor into code +that goes through review and git history, and removes the network from +the verification path entirely for the pinned version. +""" + +from __future__ import annotations + +SYNCTHING_PINNED_SHA256: dict[str, dict[str, str]] = { +''' + + +def main(argv: list[str]) -> int: + if len(argv) != 2: + print(__doc__) + return 2 + version = argv[1] if argv[1].startswith("v") else f"v{argv[1]}" + + if not (shutil.which("gpg") or shutil.which("gpg2")): + print( + "ERROR: gpg not found on PATH. This tool exists to verify the " + "release signature, so it will not run without one.\n" + "Install it (macOS: brew install gnupg, Debian: apt install gnupg).", + file=sys.stderr, + ) + return 1 + + print(f"Fetching sha256sum.txt.asc for {version} ...") + asc = syncthing._download(syncthing._asc_url(version)) + + # Raises unless there is a VALIDSIG from exactly the pinned fingerprint. + syncthing._verify_release_signature(asc) + print(f"Signature verified against {SYNCTHING_RELEASE_FINGERPRINT}") + + sums = syncthing._parse_sha256sums(asc) + archives = {k: v for k, v in sums.items() if _ARCHIVE_RE.match(k)} + if not archives: + print("ERROR: no release archives found in the signed sums file", file=sys.stderr) + return 1 + + existing: dict[str, dict[str, str]] = {} + if _TARGET.exists(): + namespace: dict[str, object] = {} + exec(compile(_TARGET.read_text(), str(_TARGET), "exec"), namespace) # noqa: S102 + existing = dict(namespace.get("SYNCTHING_PINNED_SHA256", {})) # type: ignore[arg-type] + existing[version] = archives + + body = [] + for ver in sorted(existing): + body.append(f' "{ver}": {{') + for name in sorted(existing[ver]): + body.append(f' "{name}": "{existing[ver][name]}",') + body.append(" },") + _TARGET.write_text(_HEADER + "\n".join(body) + "\n}\n") + print(f"Wrote {len(archives)} hashes for {version} to {_TARGET}") + print(json.dumps(sorted(archives), indent=2)[:200] + " ...") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv))