fix: keep callback faults out of the frame-decode fallback

Review follow-up on the previous two commits.

Widening the control-message except tuple put the callback dispatch
inside it, so an _on_new_cycle() that raised ValueError, TypeError or
AttributeError sent a perfectly good control packet to the legacy PNG
decoder -- which reported it as an image decode error and buried the
real fault. Split the two: whether the payload parses as JSON decides
frame vs control message, a second guard covers reading the fields of an
attacker-shaped body, and the callback fires outside both. It still
cannot kill the receive thread; the loop's own handler catches it, and
now says what actually went wrong.

The logo download's temp file was a fixed "<name>.part". Two plugins
asking for the same logo at once would interleave writes into it,
publish the mixture, or delete each other's partial. mkstemp gives each
download its own name in the same directory, so os.replace stays atomic.
Its descriptor is adopted by fdopen before the request runs, since a
request that raises before the write would otherwise leak the fd --
quietly, because load_logo_with_download swallows that.

Two test fixes: the oversized-frame test replaced PIL.Image.open
process-wide, the same hazard the clock helper documents, and Ruff B007
on an unused loop variable.

Full suite: 3355 passed, coverage 54%.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

src/common/logo_helper.py

@@ -6,6 +6,8 @@ Extracted from LEDMatrix core to provide reusable functionality for plugins.
 """
 
 import logging
+import os
+import tempfile
 from pathlib import Path
 from typing import Dict, List, Optional, Union
 
@@ -271,32 +273,53 @@ class LogoHelper:
         decodable image before it is left on disk: a logo URL is remote
         input, and without this an oversized or malformed response would
         be cached for every later load_logo() call to trip over.
+
+        The body is streamed and counted as it arrives rather than read
+        through response.content, which buffers the whole thing first —
+        a server that omits Content-Length and never stops sending would
+        exhaust memory before any size check could run. Nothing lands at
+        file_path until the download completes and decodes, so a failed
+        download cannot leave a truncated logo behind either.
         """
         # Ensure directory exists with proper permissions
         ensure_directory_permissions(file_path.parent, get_assets_dir_mode())
 
-        # Download with timeout
+        # A unique temp name, not a fixed "<name>.part": two plugins can
-        response = self.session.get(url, timeout=30)
+        # ask for the same logo at once, and a shared name would let them
-        response.raise_for_status()
+        # interleave writes into one file, publish the mixture, or delete
-
+        # each other's partial. Same directory, so os.replace stays atomic.
-        content = response.content
+        fd, tmp_name = tempfile.mkstemp(
-        if len(content) > MAX_LOGO_BYTES:
+            dir=str(file_path.parent), prefix=file_path.name + '.', suffix='.part')
-            raise ValueError(
+        tmp_path = Path(tmp_name)
-                f"Logo at {url} is {len(content)} bytes, over the "
-                f"{MAX_LOGO_BYTES}-byte limit; not saved")
-
-        # Save to file
-        with open(file_path, 'wb') as f:
-            f.write(content)
-
-        # Verify it decodes before leaving it on disk. PIL raises
-        # DecompressionBombError past its own pixel limit; a partial or
-        # non-image response raises UnidentifiedImageError/OSError.
         try:
-            with Image.open(file_path) as probe:
+            # fdopen outermost so the descriptor mkstemp handed back is
+            # always adopted and closed, including when the request itself
+            # raises — load_logo_with_download swallows that, so a leak
+            # here would accumulate quietly on a URL that keeps failing.
+            with os.fdopen(fd, 'wb') as f:
+                with self.session.get(url, timeout=30, stream=True) as response:
+                    response.raise_for_status()
+                    downloaded = 0
+                    for chunk in response.iter_content(chunk_size=64 * 1024):
+                        if not chunk:
+                            continue
+                        downloaded += len(chunk)
+                        if downloaded > MAX_LOGO_BYTES:
+                            raise ValueError(
+                                f"Logo at {url} exceeds the "
+                                f"{MAX_LOGO_BYTES}-byte limit; not saved")
+                        f.write(chunk)
+
+            # Verify it decodes before it becomes the cached logo. PIL
+            # raises DecompressionBombError past its own pixel limit; a
+            # partial or non-image response raises UnidentifiedImageError
+            # (an OSError subclass).
+            with Image.open(tmp_path) as probe:
                 probe.load()
-        except Exception:
+
-            file_path.unlink(missing_ok=True)
+            os.replace(tmp_path, file_path)
+        except BaseException:
+            tmp_path.unlink(missing_ok=True)
             raise
 
         # Set proper file permissions after saving

src/common/sync_manager.py

@@ -37,6 +37,13 @@ _RAW_MAGIC = b'SYNC_RAW'
 _RAW_HEADER = struct.Struct('<HH')  # width, height (uint16 LE)
 
 
+# Upper bound on a decoded frame/scroll image. Generous for any real scroll
+# image (a leader's full cycle is long but only panel-height tall), and low
+# enough that a crafted image from any host on the LAN cannot force a large
+# allocation on the render thread. Applied on both receive paths — the TCP
+# image server and the follower's legacy-PNG UDP fallback.
+_MAX_FRAME_W, _MAX_FRAME_H = 100_000, 256
+
 SYNC_PORT = 5765
 HELLO_INTERVAL = 5.0       # follower broadcasts hello every 5 s
 HEARTBEAT_INTERVAL = 2.0   # follower sends heartbeat every 2 s
@@ -278,11 +285,10 @@ class DisplaySyncManager:
                             break
                         data.extend(chunk)
                     img = Image.open(io.BytesIO(data))
-                    _MAX_W, _MAX_H = 100_000, 256  # generous for any real scroll image
+                    if img.width > _MAX_FRAME_W or img.height > _MAX_FRAME_H:
-                    if img.width > _MAX_W or img.height > _MAX_H:
                         self.logger.warning(
                             "Sync: rejected oversized scroll image %dx%d (max %dx%d) from %s",
-                            img.width, img.height, _MAX_W, _MAX_H, addr,
+                            img.width, img.height, _MAX_FRAME_W, _MAX_FRAME_H, addr,
                         )
                         continue
                     try:
@@ -489,13 +495,43 @@ class DisplaySyncManager:
                     except Exception as exc:
                         self.logger.debug("Sync: frame decode error: %s", exc)
                 else:
-                    # No magic prefix: try control-message JSON, and treat a
+                    # No magic prefix. Whether the payload parses as JSON
-                    # parse failure as a legacy (pre-magic) PNG frame. Both
+                    # decides between a control message and a legacy
-                    # wire formats are self-describing, so no size heuristic
+                    # (pre-magic) PNG frame — both wire formats are
-                    # is needed — a >512-byte control message used to be
+                    # self-describing, so no size heuristic is needed. A
-                    # misrouted into image decode and silently dropped.
+                    # >512-byte control message used to be misrouted into
+                    # image decode and silently dropped.
                     try:
                         msg = json.loads(data.decode("utf-8"))
+                    except (json.JSONDecodeError, UnicodeDecodeError):
+                        # Not JSON — try a legacy PNG frame.
+                        try:
+                            img = Image.open(io.BytesIO(data))
+                            if img.width > _MAX_FRAME_W or img.height > _MAX_FRAME_H:
+                                # Same cap the TCP image path applies: decode
+                                # is deferred until load(), so check first.
+                                self.logger.debug(
+                                    "Sync: rejected oversized legacy frame %dx%d from %s",
+                                    img.width, img.height, sender_ip,
+                                )
+                                continue
+                            img.load()
+                            self._handle_received_frame(img, sender_ip)
+                        except Exception as exc:
+                            self.logger.debug("Sync: frame decode error: %s", exc)
+                        continue
+
+                    # It parsed, so it is a control message and never a
+                    # frame. Read and validate its fields under a guard —
+                    # a UDP payload is attacker-shaped, so a non-object
+                    # body makes .get() raise AttributeError and an "sx"
+                    # carrying a non-numeric x raises ValueError/TypeError
+                    # — but dispatch the callback *outside* it. Running
+                    # the callback in here would let a fault in someone
+                    # else's code read as a malformed packet and be
+                    # logged as one.
+                    fire_new_cycle = False
+                    try:
                         t = msg.get("t")
                         if t == "hello_ack":
                             self._leader_ip = sender_ip
@@ -519,20 +555,16 @@ class DisplaySyncManager:
                                     sender_ip,
                                 )
                                 self.write_status_file()
-                                if self._on_new_cycle:
+                                fire_new_cycle = True  # build initial scroll image
-                                    self._on_new_cycle()  # build initial scroll image
                         elif t == "nc":
                             # Leader started a new scroll cycle — rebuild local image
-                            if self._on_new_cycle:
+                            fire_new_cycle = True
-                                self._on_new_cycle()
+                    except (KeyError, AttributeError, TypeError, ValueError) as exc:
-                    except (json.JSONDecodeError, UnicodeDecodeError, KeyError):
+                        self.logger.debug("Sync: malformed control message: %s", exc)
-                        # Not a control message — try legacy PNG frame.
+                        continue
-                        try:
+
-                            img = Image.open(io.BytesIO(data))
+                    if fire_new_cycle and self._on_new_cycle:
-                            img.load()
+                        self._on_new_cycle()
-                            self._handle_received_frame(img, sender_ip)
-                        except Exception as exc:
-                            self.logger.debug("Sync: frame decode error: %s", exc)
 
             except socket.timeout:
                 continue

test/test_api_v3_calendar_credentials.py

@@ -22,6 +22,7 @@ import stat
 import sys
 import time
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 
@@ -192,14 +193,25 @@ class TestBackupPruning:
         upload(api_v3_client, VALID_CREDENTIALS)
         assert len(backups(plugin_dir)) == 3  # 2 seeded + 1 new
 
-    def test_repeated_uploads_stay_bounded(self, api_v3_client, plugin_dir):
+    def test_repeated_uploads_stay_bounded(
+            self, api_v3_client, plugin_dir, api_v3_module, monkeypatch):
+        # The backup filename carries int(time.time()), so uploads inside
+        # the same second all write the same name and overwrite each other.
+        # Advance a fake clock a second per round — otherwise this never
+        # reaches six backups and the bound holds for the wrong reason.
+        clock = {"now": int(time.time())}
+        monkeypatch.setattr(
+            api_v3_module, "time", SimpleNamespace(time=lambda: clock["now"]))
         for i in range(10):
+            clock["now"] += 1
             upload(api_v3_client, {"installed": {"round": i}})
-            # Distinct mtimes so ordering is well-defined between rounds.
+            os.utime(plugin_dir / "credentials.json",
-            for path in backups(plugin_dir):
+                     (clock["now"], clock["now"]))
-                os.utime(path, (path.stat().st_mtime, path.stat().st_mtime))
+        remaining = backups(plugin_dir)
-            time.sleep(0.01)
+        assert len(remaining) == 5
-        assert len(backups(plugin_dir)) <= 5
+        # And they are the five most recent rounds, not an arbitrary five.
+        kept = sorted(int(p.name.rsplit(".", 1)[1]) for p in remaining)
+        assert kept == [clock["now"] - 4 + i for i in range(5)]
 
     def test_unremovable_backup_does_not_fail_the_upload(
             self, api_v3_client, plugin_dir, monkeypatch):

test/test_api_v3_registry_endpoints.py

@@ -166,9 +166,14 @@ class TestRegistryFromUrl:
         assert body["message"] == "An error occurred; see logs for details"
         assert "Traceback" not in str(body)
 
-    def test_non_string_repo_url_is_a_500_not_a_crash(
+    def test_non_string_repo_url_is_rejected(self, api_v3_client, api_v3_module):
-            self, api_v3_client, api_v3_module):
+        # Regression: .strip() on a non-string raised, and the catch-all
-        # .strip() on a non-string raises; the handler's catch-all turns
+        # reported the caller's own mistake as a server fault.
-        # that into a 500 rather than propagating.
         response = api_v3_client.post(self.URL, json={"repo_url": 12345})
-        assert response.status_code == 500
+        assert response.status_code == 400
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.assert_not_called()
+
+    def test_blank_repo_url_is_rejected(self, api_v3_client, api_v3_module):
+        response = api_v3_client.post(self.URL, json={"repo_url": "   "})
+        assert response.status_code == 400
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.assert_not_called()

test/test_logo_helper.py

@@ -16,12 +16,13 @@ Regression coverage for two fixed bugs:
 """
 
 import logging
+import tempfile
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
 import requests
-from PIL import Image
+from PIL import Image, UnidentifiedImageError
 
 from src.common.logo_helper import MAX_LOGO_BYTES, LogoHelper
 
@@ -46,10 +47,44 @@ def write_logo(path: Path, size=(20, 20), color=(255, 0, 0), fmt="PNG") -> Path:
     return path
 
 
-def fake_response(content: bytes):
+def fake_response(content: bytes, chunk_size: int = 64 * 1024):
+    """Stand-in for a streamed requests.Response.
+
+    _download_logo opens `with session.get(..., stream=True)` and reads
+    through iter_content(), so the fake has to be a context manager that
+    yields the body in pieces rather than exposing it as .content.
+    Chunking is the fake's own, not the caller's, so a test can dribble a
+    body out in small pieces.
+    """
     response = MagicMock()
-    response.content = content
+    response.__enter__.return_value = response
+    response.__exit__.return_value = False
     response.raise_for_status = MagicMock()
+
+    def _iter_content(*_args, **_kwargs):
+        for i in range(0, len(content), chunk_size):
+            yield content[i:i + chunk_size]
+
+    response.iter_content = _iter_content
+    return response
+
+
+def endless_response(chunk: bytes = b"\x00" * 65536):
+    """A server that declares no length and never stops sending.
+
+    This is the case response.content could not survive: it buffers to
+    completion, so the size check never got a chance to run.
+    """
+    response = MagicMock()
+    response.__enter__.return_value = response
+    response.__exit__.return_value = False
+    response.raise_for_status = MagicMock()
+
+    def _iter_content(*_args, **_kwargs):
+        while True:
+            yield chunk
+
+    response.iter_content = _iter_content
     return response
 
 
@@ -169,7 +204,10 @@ class TestLoadLogoWithDownload:
         logo = helper.load_logo_with_download("PHI", path, "http://x/logo.png")
         assert logo is not None
         assert path.exists()
-        helper.session.get.assert_called_once_with("http://x/logo.png", timeout=30)
+        # stream=True is load-bearing: it is what lets the size cap apply
+        # before the body is buffered.
+        helper.session.get.assert_called_once_with(
+            "http://x/logo.png", timeout=30, stream=True)
 
     def test_download_failure_falls_back_to_placeholder(self, helper, tmp_path):
         helper.session.get = MagicMock(
@@ -215,18 +253,88 @@ class TestDownloadLogo:
         path = tmp_path / "huge.png"
         helper.session.get = MagicMock(
             return_value=fake_response(b"\x00" * (MAX_LOGO_BYTES + 1)))
-        with pytest.raises(ValueError, match="over the"):
+        with pytest.raises(ValueError, match="exceeds the"):
             helper._download_logo("http://x/huge.png", path)
         assert not path.exists()
 
+    def test_unbounded_response_is_aborted_at_the_cap(self, helper, tmp_path):
+        # Regression: the cap used to be checked against response.content,
+        # which buffers the whole body first — so a server that omits
+        # Content-Length and never stops sending exhausted memory before
+        # the check could run. Streaming counts bytes as they arrive, so
+        # this terminates instead of hanging.
+        path = tmp_path / "endless.png"
+        helper.session.get = MagicMock(return_value=endless_response())
+        with pytest.raises(ValueError, match="exceeds the"):
+            helper._download_logo("http://x/endless.png", path)
+        assert not path.exists()
+
+    def test_no_partial_file_is_left_when_the_stream_dies(self, helper, tmp_path):
+        # A transfer that fails midway must not leave a truncated logo
+        # where the real one belongs — load_logo() would cache it.
+        path = tmp_path / "cut.png"
+        real = png_bytes()
+
+        def _dies_midway(*_args, **_kwargs):
+            yield real[:20]
+            raise OSError("connection reset")
+
+        response = MagicMock()
+        response.__enter__.return_value = response
+        response.__exit__.return_value = False
+        response.raise_for_status = MagicMock()
+        response.iter_content = _dies_midway
+        helper.session.get = MagicMock(return_value=response)
+
+        with pytest.raises(OSError):
+            helper._download_logo("http://x/cut.png", path)
+        assert not path.exists()
+        assert list(tmp_path.glob("*.part")) == []
+
+    def test_concurrent_downloads_do_not_share_a_temp_file(self, helper, tmp_path):
+        # Two plugins can ask for the same logo at once. A fixed
+        # "<name>.part" would let them interleave writes into one file and
+        # publish the mixture; each download gets its own temp name.
+        path = tmp_path / "PHI.png"
+        seen = []
+        real_mkstemp = tempfile.mkstemp
+
+        def record(*args, **kwargs):
+            fd, name = real_mkstemp(*args, **kwargs)
+            seen.append(name)
+            return fd, name
+
+        with patch("src.common.logo_helper.tempfile.mkstemp", side_effect=record):
+            helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
+            helper._download_logo("http://x/logo.png", path)
+            helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
+            helper._download_logo("http://x/logo.png", path)
+
+        assert len(seen) == 2 and seen[0] != seen[1]
+        assert path.exists()
+        assert list(tmp_path.glob("*.part")) == []  # both cleaned up
+
+    def test_request_failure_leaves_no_temp_file(self, helper, tmp_path):
+        # mkstemp creates the file up front, so an error before any bytes
+        # arrive still has something to clean up.
+        helper.session.get = MagicMock(
+            side_effect=requests.RequestException("connection reset"))
+        with pytest.raises(requests.RequestException):
+            helper._download_logo("http://x/logo.png", tmp_path / "PHI.png")
+        assert list(tmp_path.glob("*")) == []
+
     def test_non_image_response_is_deleted_and_raises(self, helper, tmp_path):
         # Regression: undecodable bytes stayed on disk, so every later
         # load_logo() call hit the corrupt file instead of re-downloading.
         path = tmp_path / "bad.png"
         helper.session.get = MagicMock(return_value=fake_response(b"<html>404</html>"))
-        with pytest.raises(Exception):
+        # Specifically Pillow's identify failure, not any OSError: the
+        # point is that the bytes did not decode, and OSError alone would
+        # also admit unrelated filesystem faults.
+        with pytest.raises(UnidentifiedImageError):
             helper._download_logo("http://x/bad.png", path)
         assert not path.exists()
+        assert list(tmp_path.glob("*.part")) == []
 
     def test_decompression_bomb_is_deleted_and_raises(self, helper, tmp_path, monkeypatch):
         path = tmp_path / "bomb.png"

test/test_sync_manager.py

@@ -31,6 +31,7 @@ import socket
 import threading
 import time
 from pathlib import Path
+from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 import numpy as np
@@ -118,11 +119,27 @@ def raise_n_then_stop(mgr, exc, count):
     return _side_effect
 
 
+def fake_clock(monkeypatch, *, time_fn=None, sleep_fn=None):
+    """Swap sync_manager's own `time` reference for a private stand-in.
+
+    sync_manager.time IS the stdlib module, so patching attributes on it
+    would freeze the clock and no-op sleep for the whole process —
+    including the daemon threads earlier tests left running, which is a
+    hard-to-trace source of cross-test flakiness. Rebinding the module's
+    reference keeps the patch scoped to the code under test. Anything not
+    overridden falls through to the real functions.
+    """
+    monkeypatch.setattr(sync_manager, "time", SimpleNamespace(
+        time=time_fn or time.time,
+        sleep=sleep_fn or time.sleep,
+    ))
+
+
 def run_watchdog_once(monkeypatch, mgr, watchdog, now):
     """Run exactly one watchdog iteration at a frozen wall-clock time."""
-    monkeypatch.setattr(sync_manager.time, "time", lambda: now)
+    fake_clock(monkeypatch,
-    monkeypatch.setattr(
+               time_fn=lambda: now,
-        sync_manager.time, "sleep", lambda _: setattr(mgr, "_running", False))
+               sleep_fn=lambda _: setattr(mgr, "_running", False))
     mgr._running = True
     watchdog()
 
@@ -334,11 +351,11 @@ class TestLeaderRecvLoop:
         assert mgr._leader_state is LeaderState.CONNECTED
         assert mgr._peer_ip == "10.0.0.8"
 
-    def test_heartbeat_from_known_peer_refreshes_timer(self):
+    def test_heartbeat_from_known_peer_refreshes_timer(self, monkeypatch):
         mgr = make_manager(role=SyncRole.LEADER)
         mgr._peer_ip = "10.0.0.8"
-        with patch.object(sync_manager.time, "time", return_value=12345.0):
+        fake_clock(monkeypatch, time_fn=lambda: 12345.0)
-            self._drive(mgr, json.dumps({"t": "hb"}).encode())
+        self._drive(mgr, json.dumps({"t": "hb"}).encode())
         assert mgr._last_heartbeat_time == 12345.0
 
     def test_heartbeat_from_stranger_is_ignored(self):
@@ -370,7 +387,7 @@ class TestLeaderRecvLoop:
         mgr._recv_sock = MagicMock()
         mgr._recv_sock.recvfrom.side_effect = raise_n_then_stop(mgr, OSError("boom"), 3)
         sleeps = MagicMock()
-        monkeypatch.setattr(sync_manager.time, "sleep", sleeps)
+        fake_clock(monkeypatch, sleep_fn=sleeps)
         mgr._running = True
         mgr._leader_recv_loop()
         assert sleeps.call_count == 3
@@ -470,6 +487,76 @@ class TestFollowerRecvLoop:
         self._drive(mgr, json.dumps({"t": "nc"}).encode())
         assert calls == [1]
 
+    def test_non_object_json_does_not_reach_the_outer_handler(self):
+        # A bare JSON scalar parses, then msg.get() raises AttributeError.
+        # That has to be caught here so the payload still gets its shot at
+        # the legacy-PNG fallback; escaping to the outer handler would also
+        # charge one malformed packet the 0.1s error backoff.
+        mgr = make_manager(role=SyncRole.FOLLOWER)
+        sleeps = MagicMock()
+        with patch.object(sync_manager, "time",
+                          SimpleNamespace(time=time.time, sleep=sleeps)):
+            self._drive(mgr, b"12345")
+        assert mgr.get_latest_frame() is None
+        sleeps.assert_not_called()
+
+    def test_non_numeric_scroll_x_does_not_reach_the_outer_handler(self):
+        # float("a") raises ValueError; {"x": null} raises TypeError.
+        for payload in ({"t": "sx", "x": "a"}, {"t": "sx", "x": None}):
+            mgr = make_manager(role=SyncRole.FOLLOWER)
+            sleeps = MagicMock()
+            with patch.object(sync_manager, "time",
+                              SimpleNamespace(time=time.time, sleep=sleeps)):
+                self._drive(mgr, json.dumps(payload).encode())
+            assert mgr.get_latest_scroll_x() is None
+            sleeps.assert_not_called()
+
+    def test_callback_failure_is_not_mistaken_for_a_malformed_packet(self, monkeypatch):
+        # A payload that parses is a control message, full stop. If the
+        # callback it triggers raises one of the types the field guard
+        # catches, that fault belongs to the callback: it must not send
+        # the packet to the image decoder, which would report it as a
+        # decode error and bury the real cause. The loop still survives
+        # it — the outer handler catches it like any other fault.
+        mgr = make_manager(role=SyncRole.FOLLOWER)
+        mgr._follower_state = FollowerState.FOLLOWER
+
+        def boom():
+            raise ValueError("callback is broken")
+
+        mgr._on_new_cycle = boom
+        fake_clock(monkeypatch, sleep_fn=MagicMock())
+        self._drive(mgr, json.dumps({"t": "nc"}).encode())
+
+        logged = " | ".join(str(c) for c in mgr.logger.debug.call_args_list)
+        assert "callback is broken" in logged
+        assert "frame decode error" not in logged
+        assert "malformed control message" not in logged
+
+    def test_oversized_legacy_frame_is_rejected_before_decode(self, monkeypatch):
+        # The UDP path is reachable by any host on the LAN, so it caps
+        # dimensions before load() just as the TCP image server does.
+        mgr = make_manager(role=SyncRole.FOLLOWER)
+
+        class Huge:
+            width, height = 10, sync_manager._MAX_FRAME_H + 1
+
+            def load(self):
+                raise AssertionError("load() must not run past the cap")
+
+        # Rebind the module's reference rather than mutating PIL.Image
+        # itself, which would hand Huge() to every caller in the process
+        # — including daemon threads earlier tests left running. Same
+        # reasoning as fake_clock above. The other names the receive loop
+        # reads off this reference pass through to the real module.
+        monkeypatch.setattr(sync_manager, "Image", SimpleNamespace(
+            open=lambda *a, **kw: Huge(),
+            frombuffer=Image.frombuffer,
+            DecompressionBombError=Image.DecompressionBombError,
+        ))
+        self._drive(mgr, b"\x89PNG not really but not JSON either")
+        assert mgr.get_latest_frame() is None
+
     def test_scroll_x_missing_key_is_swallowed(self):
         mgr = make_manager(role=SyncRole.FOLLOWER)
         self._drive(mgr, json.dumps({"t": "sx"}).encode())  # no "x"
@@ -480,7 +567,7 @@ class TestFollowerRecvLoop:
         mgr._recv_sock = MagicMock()
         mgr._recv_sock.recvfrom.side_effect = raise_n_then_stop(mgr, OSError("boom"), 3)
         sleeps = MagicMock()
-        monkeypatch.setattr(sync_manager.time, "sleep", sleeps)
+        fake_clock(monkeypatch, sleep_fn=sleeps)
         mgr._running = True
         mgr._follower_recv_loop()
         assert sleeps.call_count == 3
@@ -797,23 +884,76 @@ class TestStop:
         make_manager(role=SyncRole.STANDALONE).stop()  # all sockets None
 
 
-class TestLoopbackHandshake:
+def _broadcast_available(port):
+    """True when a UDP broadcast can be sent at all in this environment.
+
+    The handshake below depends on broadcast: the follower announces
+    itself to ("<broadcast>", port), and sync_manager swallows any sendto
+    error. Without this probe, a sandbox or CI network that refuses
+    broadcast would make the test wait out its whole deadline and then
+    fail for a reason that has nothing to do with the code.
+
+    Sending is enough to detect the case that actually occurs — a
+    refusing environment raises here. Confirming *delivery* would mean
+    binding INADDR_ANY to receive, which is a listening socket this suite
+    has no reason to open; a network that accepts the send and silently
+    drops it still reaches the assertion, exactly as before.
+    """
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    try:
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
+        sock.sendto(b"probe", ("<broadcast>", port))
+        return True
+    except OSError:
+        return False
+    finally:
+        sock.close()
+
+
+class TestRealSocketHandshake:
     def test_leader_and_follower_negotiate_over_real_sockets(self, monkeypatch):
         # One end-to-end check that the wire format actually round-trips:
         # every other test drives the loops with mocked sockets.
+        #
+        # Not loopback-only, despite the free-port probe below: the manager
+        # binds UDP and TCP on all interfaces and the follower announces by
+        # broadcast. That is the behaviour under test, so the environment
+        # has to support it.
         monkeypatch.setattr(sync_manager, "HELLO_INTERVAL", 0.02)
         monkeypatch.setattr(sync_manager, "HEARTBEAT_INTERVAL", 0.02)
 
-        # Pick a free port by binding one on loopback and releasing it.
-        # Loopback, not "", so this test never opens a port to the network.
-        probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-        probe.bind(("127.0.0.1", 0))
-        port = probe.getsockname()[1]
-        probe.close()
-
         hw = {"rows": 32, "cols": 64, "chain_length": 1}
-        leader = DisplaySyncManager("leader", {"port": port}, hw, MagicMock())
+        leader = follower = None
-        follower = DisplaySyncManager("follower", {"port": port}, hw, MagicMock())
+        # The free-port probe is inherently racy — the port can be taken
+        # between release and rebind — so retry rather than fail on it.
+        for _attempt in range(5):
+            # Probed on loopback: this only needs a port number, and the
+            # manager's own bind is what has to succeed. If the port turns
+            # out to be taken on another interface, the retry below covers
+            # it — same as for the race.
+            probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            probe.bind(("127.0.0.1", 0))
+            port = probe.getsockname()[1]
+            probe.close()
+
+            if not _broadcast_available(port):
+                pytest.skip("environment refuses UDP broadcast")
+
+            try:
+                leader = DisplaySyncManager("leader", {"port": port}, hw, MagicMock())
+                follower = DisplaySyncManager("follower", {"port": port}, hw, MagicMock())
+                break
+            except OSError:
+                # Port taken between probe and bind, or the TCP image
+                # server could not bind port+1. Tear down whichever end
+                # came up before retrying with a fresh port.
+                for mgr in (leader, follower):
+                    if mgr is not None:
+                        mgr.stop()
+                leader = follower = None
+        else:
+            pytest.skip("could not obtain a free port pair for the handshake")
+
         try:
             deadline = time.time() + 5.0
             while time.time() < deadline:

test/web_interface/test_api_v3_config_raw.py

@@ -112,6 +112,11 @@ class TestSaveRawMain:
         assert response.status_code == 400
         body = response.get_json()
         assert body["status"] == "error"
+        # A body that was sent but does not parse is a distinct mistake
+        # from sending none, and says so. Previously the handler's own
+        # json.JSONDecodeError arm was unreachable — Werkzeug raised
+        # first — so this collapsed into "No data provided".
+        assert "Invalid JSON in request body" in body["message"]
 
     def test_config_error_is_a_500_with_context(self, env, monkeypatch):
         def refuse(kind, data):

web_interface/blueprints/api_v3.py

@@ -1345,18 +1345,20 @@ def save_raw_main_config():
         if not api_v3.config_manager:
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
+        # silent=True so a malformed body returns None instead of raising
+        # Werkzeug's own BadRequest, which would answer in a different
+        # shape than this API's. Distinguish the two causes: a body that
+        # was sent but does not parse is a different mistake from no body.
         data = request.get_json(silent=True)
+        if data is None and request.get_data():
+            return jsonify({'status': 'error', 'message': 'Invalid JSON in request body'}), 400
         if not data:
             return jsonify({'status': 'error', 'message': 'No data provided'}), 400
 
-        # Validate that it's valid JSON (already parsed by request.get_json())
         # Save the raw config file
         api_v3.config_manager.save_raw_file_content('main', data)
 
         return jsonify({'status': 'success', 'message': 'Main configuration saved successfully'})
-    except json.JSONDecodeError as e:
-        logger.error('Invalid JSON', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'Invalid JSON in request body'}), 400
     except Exception as e:
         from src.exceptions import ConfigError
         logger.error("Error saving raw main config", exc_info=True)
@@ -1391,7 +1393,11 @@ def save_raw_secrets_config():
         if not api_v3.config_manager:
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
+        # See save_raw_main_config: silent parsing, with a sent-but-broken
+        # body reported separately from a missing one.
         data = request.get_json(silent=True)
+        if data is None and request.get_data():
+            return jsonify({'status': 'error', 'message': 'Invalid JSON in request body'}), 400
         if not data:
             return jsonify({'status': 'error', 'message': 'No data provided'}), 400
 
@@ -1403,9 +1409,6 @@ def save_raw_secrets_config():
             api_v3.plugin_store_manager.github_token = api_v3.plugin_store_manager._load_github_token()
 
         return jsonify({'status': 'success', 'message': 'Secrets configuration saved successfully'})
-    except json.JSONDecodeError as e:
-        logger.error('Invalid JSON', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'Invalid JSON in request body'}), 400
     except Exception as e:
         from src.exceptions import ConfigError
         logger.error("Error saving raw secrets config", exc_info=True)
@@ -3975,6 +3978,11 @@ def install_plugin_from_url():
         if not data or 'repo_url' not in data:
             return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
 
+        # A non-string repo_url is a client mistake, not a server fault:
+        # .strip() would raise and the catch-all would report it as a 500.
+        if not isinstance(data['repo_url'], str) or not data['repo_url'].strip():
+            return jsonify({'status': 'error', 'message': 'repo_url must be a non-empty string'}), 400
+
         repo_url = data['repo_url'].strip()
         plugin_id = data.get('plugin_id')  # Optional, for monorepo installations
         plugin_path = data.get('plugin_path')  # Optional, for monorepo subdirectory
@@ -4030,6 +4038,11 @@ def get_registry_from_url():
         if not data or 'repo_url' not in data:
             return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
 
+        # A non-string repo_url is a client mistake, not a server fault:
+        # .strip() would raise and the catch-all would report it as a 500.
+        if not isinstance(data['repo_url'], str) or not data['repo_url'].strip():
+            return jsonify({'status': 'error', 'message': 'repo_url must be a non-empty string'}), 400
+
         repo_url = data['repo_url'].strip()
 
         # Get registry from the URL
@@ -4075,6 +4088,11 @@ def add_saved_repository():
         if not data or 'repo_url' not in data:
             return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
 
+        # A non-string repo_url is a client mistake, not a server fault:
+        # .strip() would raise and the catch-all would report it as a 500.
+        if not isinstance(data['repo_url'], str) or not data['repo_url'].strip():
+            return jsonify({'status': 'error', 'message': 'repo_url must be a non-empty string'}), 400
+
         repo_url = data['repo_url'].strip()
         name = data.get('name')
 
@@ -7289,25 +7307,16 @@ def upload_calendar_credentials():
         try:
             file_content = file.read()
             file.seek(0)
-            json.loads(file_content)
+            creds_data = json.loads(file_content)
         except json.JSONDecodeError:
             return jsonify({'status': 'error', 'message': 'File is not valid JSON'}), 400
 
-        # Validate it looks like Google OAuth credentials. The content
+        # Validate it looks like Google OAuth credentials. A bare scalar, a
-        # already parsed as JSON above, so anything raising here means it is
+        # list, true/null — all valid JSON, none of them credentials. Reject
-        # not credentials-shaped — a bare scalar, for instance, where the
+        # rather than save: a file written as credentials.json but unusable
-        # membership test raises TypeError. Reject rather than swallow: a
+        # as credentials only fails later, somewhere less obvious.
-        # file saved as credentials.json but not usable as credentials only
+        if not isinstance(creds_data, dict) or not (
-        # fails later, somewhere less obvious.
+                'installed' in creds_data or 'web' in creds_data):
-        try:
-            file.seek(0)
-            creds_data = json.loads(file.read())
-            file.seek(0)
-            is_oauth_shaped = 'installed' in creds_data or 'web' in creds_data
-        except Exception:
-            is_oauth_shaped = False
-
-        if not is_oauth_shaped:
             return jsonify({
                 'status': 'error',
                 'message': 'File does not appear to be a valid Google OAuth credentials file'