Stop array-item secrets being wiped, and logging them

Three review findings from #485 that I missed when addressing that PR;
it has since merged, so they land here.

1. Array-item secrets destroyed by any unrelated save (data loss).

remove_empty_secrets recursed into dicts but let a list fall through to
the scalar branch and kept it verbatim. Lists merge by *replacement*, so
the blanks the masked form posts back went straight over the stored
array:

    stored   [{"name":"a","token":"REAL-A"}, {"name":"b","token":"REAL-B"}]
    posted   [{"name":"a","token":""},       {"name":"b","token":""}]
    merged   [{"name":"a","token":""},       {"name":"b","token":""}]
             -> both credentials gone

Same failure as the scalar api_key case fixed earlier, one container
deeper. Lists now prune element-wise, and a list with nothing real in it
is dropped so the stored one is left alone. Where one entry does change,
the new merge_secrets merges by index instead of replacing.

Two details the first attempt got wrong, both caught by existing tests:

- An emptied dict item must stay {}, not None. ConfigManager's
  _strip_secrets_recursive treats a secrets list as *parallel* to the
  regular one ({} = "item i has no secrets"); a None makes it stop
  looking parallel, and it then drops the whole key from the main config
  -- silently deleting the items' non-secret fields too.
- The incoming list's length wins. The regular config's list is
  authoritative about how many items exist, so preserving surplus stored
  entries would let the two fall out of step and make deleting an entry
  impossible.

2. Submitted credentials written to the journal (security).

save_plugin_config logged `Full config: {plugin_config}` at INFO and
`Config that failed: {plugin_config}` at ERROR. Both run before
separate_secrets, so plugin_config still held the values just typed into
the form. Now keys only. Swept the rest of web_interface/ and src/ for
the same shape -- these were the only two.

3. Restart banner kept stale wording.

showRestartPending() cleared the stored custom text but left the DOM
element alone, so a config save could show the previous update's
message. The default is read back from the server-rendered copy rather
than duplicated in JS, so the template stays the one owner of the string.

Verified: 556 passed, 1 skipped across the web suite. Mutation-checked --
reverting api_v3 fails the logging guard and the array-merge test;
reverting either half of the secret_helpers change fails the unit tests.
New end-to-end coverage drives the real endpoint, not just the helpers.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01STMbQE4YctTacQXfbYqKuW

.github/workflows/test.yml

@@ -72,4 +72,4 @@ jobs:
             --ignore=test/plugins \
             --cov=src --cov=web_interface \
             --cov-report=term \
-            --cov-fail-under=48
+            --cov-fail-under=52

first_time_install.sh

@@ -1419,9 +1419,16 @@ $ACTUAL_USER ALL=(ALL) NOPASSWD: $BASH_PATH $PROJECT_ROOT_DIR/scripts/fix_perms/
 EOF
 if [ -n "$JOURNALCTL_PATH" ]; then
     cat >> /tmp/ledmatrix_web_sudoers << EOF
-$ACTUAL_USER ALL=(ALL) NOPASSWD: $JOURNALCTL_PATH -u ledmatrix.service *
+# NOEXEC, because these rules end in a wildcard and journalctl starts a pager
-$ACTUAL_USER ALL=(ALL) NOPASSWD: $JOURNALCTL_PATH -u ledmatrix *
+# when its output is a terminal. From that pager (less) a "!sh" is a root
-$ACTUAL_USER ALL=(ALL) NOPASSWD: $JOURNALCTL_PATH -t ledmatrix *
+# shell -- the standard journalctl escalation. The web interface always passes
+# --no-pager, so nothing here needs it, but the rule cannot require a flag that
+# sits in the middle of the command line. NOEXEC stops the command executing
+# another program at all, which closes the hole without depending on wildcard
+# matching subtleties.
+$ACTUAL_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -u ledmatrix.service *
+$ACTUAL_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -u ledmatrix *
+$ACTUAL_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -t ledmatrix *
 EOF
 fi
 

scripts/install/configure_web_sudo.sh

@@ -100,10 +100,15 @@ TEMP_SUDOERS="/tmp/ledmatrix_web_sudoers_$$"
     echo "$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH restart ledmatrix-web.service"
 
     # Optional: journalctl (non-critical — skip if not found)
+    #
+    # NOEXEC, matching first_time_install.sh. These rules end in a wildcard and
+    # journalctl starts a pager, so without it the caller can reach a shell:
+    # less runs "!command" as the user the pager belongs to, which here is
+    # root. NOEXEC stops the granted command executing anything of its own.
     if [ -n "$JOURNALCTL_PATH" ]; then
-        echo "$WEB_USER ALL=(ALL) NOPASSWD: $JOURNALCTL_PATH -u ledmatrix.service *"
+        echo "$WEB_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -u ledmatrix.service *"
-        echo "$WEB_USER ALL=(ALL) NOPASSWD: $JOURNALCTL_PATH -u ledmatrix *"
+        echo "$WEB_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -u ledmatrix *"
-        echo "$WEB_USER ALL=(ALL) NOPASSWD: $JOURNALCTL_PATH -t ledmatrix *"
+        echo "$WEB_USER ALL=(ALL) NOPASSWD:NOEXEC: $JOURNALCTL_PATH -t ledmatrix *"
     fi
 
     # Required: python3, bash

scripts/install/configure_wifi_permissions.sh

@@ -25,9 +25,44 @@ if [ "$EUID" -eq 0 ]; then
     exit 1
 fi
 
+# Resolve command paths against a fixed PATH, and check what we resolved.
+#
+# Every path found here is written into a sudoers file as a NOPASSWD grant, so
+# whoever controls the binary at that path controls root. first_time_install.sh
+# re-execs itself with `sudo -E`, which preserves the invoking user's
+# environment -- PATH included -- so without pinning it, `which nmcli` can
+# resolve to anything on that PATH: a writable directory early in it turns a
+# compromise of the low-privilege web user into permanent root.
+PATH=/usr/sbin:/usr/bin:/sbin:/bin
+export PATH
+
+# A binary named in a sudoers rule must be root-owned and writable by nobody
+# else, or the grant hands root to whoever can rewrite it.
+require_trusted_binary() {
+    local label="$1" path="$2"
+    if [ ! -x "$path" ]; then
+        echo "✗ $label: $path is not an executable file"
+        exit 1
+    fi
+    local owner perms
+    owner=$(stat -c '%u' "$path") || exit 1
+    perms=$(stat -c '%a' "$path") || exit 1
+    if [ "$owner" != "0" ]; then
+        echo "✗ $label: $path is not owned by root (uid $owner); refusing to"
+        echo "  grant it NOPASSWD sudo."
+        exit 1
+    fi
+    # Group- or world-writable means someone other than root can replace it.
+    case "$perms" in
+        *[2367]) echo "✗ $label: $path is writable by group or other ($perms);"
+                 echo "  refusing to grant it NOPASSWD sudo."
+                 exit 1 ;;
+    esac
+}
+
 # Get the full paths to commands
-NMCLI_PATH=$(which nmcli || echo "/usr/bin/nmcli")
+NMCLI_PATH=$(command -v nmcli || echo "/usr/bin/nmcli")
-SYSTEMCTL_PATH=$(which systemctl)
+SYSTEMCTL_PATH=$(command -v systemctl)
 
 echo "Command paths:"
 echo "  nmcli: $NMCLI_PATH"
@@ -37,6 +72,18 @@ echo "  systemctl: $SYSTEMCTL_PATH"
 echo ""
 echo "Step 1: Configuring sudo permissions for nmcli..."
 SUDOERS_FILE="/etc/sudoers.d/ledmatrix_wifi"
+SYSCTL_PATH=$(command -v sysctl || echo /usr/sbin/sysctl)
+NFT_PATH=$(command -v nft || echo /usr/sbin/nft)
+RFKILL_PATH=$(command -v rfkill || echo /usr/sbin/rfkill)
+MKDIR_PATH=$(command -v mkdir || echo /usr/bin/mkdir)
+
+# Checked before any of them reaches the sudoers file.
+require_trusted_binary "nmcli"     "$NMCLI_PATH"
+require_trusted_binary "systemctl" "$SYSTEMCTL_PATH"
+require_trusted_binary "sysctl"    "$SYSCTL_PATH"
+require_trusted_binary "nft"       "$NFT_PATH"
+require_trusted_binary "rfkill"    "$RFKILL_PATH"
+require_trusted_binary "mkdir"     "$MKDIR_PATH"
 
 # Create a temporary sudoers file using mktemp (handles permissions better)
 TEMP_SUDOERS=$(mktemp) || {
@@ -62,6 +109,36 @@ $WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH start dnsmasq
 $WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH stop dnsmasq
 $WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH restart dnsmasq
 $WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH restart NetworkManager
+# The captive portal turns IP forwarding on while the access point is up and
+# restores the previous value when it comes down (wifi_manager._setup_iptables_
+# redirect / _teardown_iptables_redirect). Without this rule that sudo call
+# needs a password, so forwarding stays off and clients associate to the AP but
+# cannot route. It goes unnoticed on a stock Raspberry Pi image, where
+# /etc/sudoers.d/010_pi-nopasswd grants the default user blanket NOPASSWD and
+# masks every gap in this file -- it only bites once that blanket rule is
+# removed.
+$WEB_USER ALL=(ALL) NOPASSWD: $SYSCTL_PATH -w net.ipv4.ip_forward=0
+$WEB_USER ALL=(ALL) NOPASSWD: $SYSCTL_PATH -w net.ipv4.ip_forward=1
+# The portal's redirect lives in its own nftables table, created when the AP
+# comes up and deleted when it goes down, and the radio has to be unblocked
+# before the AP can start at all. Same story as the sysctl rules above: called
+# with sudo, never granted here, and invisible on a stock Pi image.
+$WEB_USER ALL=(ALL) NOPASSWD: $NFT_PATH add table ip ledmatrix
+$WEB_USER ALL=(ALL) NOPASSWD: $NFT_PATH delete table ip ledmatrix
+$WEB_USER ALL=(ALL) NOPASSWD: $RFKILL_PATH unblock wifi
+# NetworkManager's dnsmasq drop-in directory, exact path.
+$WEB_USER ALL=(ALL) NOPASSWD: $MKDIR_PATH -p /etc/NetworkManager/dnsmasq-shared.d
+#
+# iptables is deliberately NOT granted here. Its rules are built from the live
+# interface name and port, so a rule covering them needs a trailing wildcard --
+# and `iptables --modprobe=/path/to/anything` runs that path as root, so
+# `NOPASSWD: iptables *` is a root shell for the web user by another name. That
+# is a worse outcome than the gap it would close, which today is masked anyway
+# by the blanket NOPASSWD rule on stock Pi images.
+#
+# Closing it safely means a wrapper script that builds the rules itself and
+# takes only an interface and a port, granted the way safe_plugin_rm.sh already
+# is. That belongs in its own change rather than being smuggled into this one.
 
 # Allow copying hostapd and dnsmasq config files into place
 $WEB_USER ALL=(ALL) NOPASSWD: /usr/bin/cp /tmp/hostapd.conf /etc/hostapd/hostapd.conf

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
 
@@ -19,6 +21,10 @@ from src.common.permission_utils import (
 )
 
 
+# Well above any real team logo; bounds what a remote URL can write to disk.
+MAX_LOGO_BYTES = 10 * 1024 * 1024
+
+
 class LogoHelper:
     """
     Helper class for logo loading, caching, and resizing.
@@ -226,7 +232,10 @@ class LogoHelper:
         return {
             'cached_logos': len(self._logo_cache),
             'cache_size_limit': self.cache_size,
-            'cache_usage_percent': (len(self._logo_cache) / self.cache_size) * 100
+            'cache_usage_percent': (
+                (len(self._logo_cache) / self.cache_size) * 100
+                if self.cache_size else 0
+            ),
         }
     
     def _resize_logo(self, logo: Image.Image, max_width: Optional[int] = None, 
@@ -258,21 +267,64 @@ class LogoHelper:
         self._cache_order.append(cache_key)
     
     def _download_logo(self, url: str, file_path: Path) -> None:
-        """Download logo from URL."""
+        """Download logo from URL.
+
+        The response size is capped and the saved file is verified as a
+        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.
-        # Save to file
+        fd, tmp_name = tempfile.mkstemp(
-        with open(file_path, 'wb') as f:
+            dir=str(file_path.parent), prefix=file_path.name + '.', suffix='.part')
-            f.write(response.content)
+        tmp_path = Path(tmp_name)
-        
+        try:
+            # 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()
+
+            os.replace(tmp_path, file_path)
+        except BaseException:
+            tmp_path.unlink(missing_ok=True)
+            raise
+
         # Set proper file permissions after saving
         ensure_file_permissions(file_path, get_assets_file_mode())
-        
+
         self.logger.debug(f"Downloaded logo to {file_path}")
     
     def _create_placeholder_logo(self, team_abbr: str, 

src/common/sync_manager.py

@@ -19,6 +19,7 @@ Port default: 5765 (UDP). Open this port on both Pis if ufw is active:
 
 import io
 import json
+import math
 import os
 import socket
 import struct
@@ -37,6 +38,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
@@ -101,6 +109,7 @@ class DisplaySyncManager:
         self._peer_chain: int = 0
         self._last_heartbeat_time: float = 0.0
         self._leader_width: int = 0  # set by display_controller after init
+        self._oversized_frame_warned: bool = False
 
         # Follower state
         self._follower_state = FollowerState.STANDALONE
@@ -174,6 +183,10 @@ class DisplaySyncManager:
                 continue
             except Exception as exc:
                 self.logger.debug("Sync leader recv error: %s", exc)
+                # Brief backoff: a socket left in a bad state raises
+                # immediately, which would otherwise spin this thread at
+                # 100% CPU logging the same error.
+                time.sleep(0.1)
 
     def _handle_hello(self, msg: dict, sender_ip: str) -> None:
         hw = self._hw_config
@@ -273,11 +286,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:
@@ -396,7 +408,7 @@ class DisplaySyncManager:
             data = header + arr.tobytes()
             if len(data) <= 65000:
                 self._send_sock.sendto(data, (self._peer_ip, self.port))
-            elif not getattr(self, '_oversized_frame_warned', False):
+            elif not self._oversized_frame_warned:
                 self._oversized_frame_warned = True
                 self.logger.warning(
                     "Sync: frame too large for UDP (%d bytes, max 65000) — "
@@ -451,43 +463,76 @@ class DisplaySyncManager:
         )
         self.write_status_file()
 
+    def _handle_received_frame(self, img: Image.Image, sender_ip: str) -> None:
+        """Record a decoded leader frame and enter follower mode if needed."""
+        with self._frame_lock:
+            self._latest_frame = img
+        self._last_leader_frame_time = time.time()
+        self._leader_ip = sender_ip
+
+        if self._follower_state == FollowerState.STANDALONE:
+            self._follower_state = FollowerState.FOLLOWER
+            self.logger.info(
+                "Sync: leader active at %s — switching to follower mode",
+                sender_ip,
+            )
+            self.write_status_file()
+
     def _follower_recv_loop(self) -> None:
         while self._running:
             try:
                 data, addr = self._recv_sock.recvfrom(65535)
                 sender_ip = addr[0]
 
-                if data[:8] == _RAW_MAGIC or len(data) > 512:
+                if data[:8] == _RAW_MAGIC:
-                    # Frame data: prefer magic-tagged raw RGB; fall back to legacy PNG
+                    # Magic-tagged raw RGB frame — self-describing, no guessing.
                     try:
-                        if data[:8] == _RAW_MAGIC:
+                        w, h = _RAW_HEADER.unpack(data[8:12])
-                            w, h = _RAW_HEADER.unpack(data[8:12])
+                        raw = data[12:]
-                            raw = data[12:]
+                        img = Image.frombuffer(
-                            img = Image.frombuffer(
+                            "RGB", (w, h), raw, "raw", "RGB", 0, 1
-                                "RGB", (w, h), raw, "raw", "RGB", 0, 1
+                        )
-                            )
+                        self._handle_received_frame(img, sender_ip)
-                        else:
-                            # Fallback: try legacy PNG
-                            img = Image.open(io.BytesIO(data))
-                            img.load()
-                        with self._frame_lock:
-                            self._latest_frame = img
-                        self._last_leader_frame_time = time.time()
-                        self._leader_ip = sender_ip
-
-                        if self._follower_state == FollowerState.STANDALONE:
-                            self._follower_state = FollowerState.FOLLOWER
-                            self.logger.info(
-                                "Sync: leader active at %s — switching to follower mode",
-                                sender_ip,
-                            )
-                            self.write_status_file()
                     except Exception as exc:
                         self.logger.debug("Sync: frame decode error: %s", exc)
                 else:
-                    # Control message
+                    # No magic prefix. Whether the payload parses as JSON
+                    # decides between a control message and a legacy
+                    # (pre-magic) PNG frame — both wire formats are
+                    # self-describing, so no size heuristic is needed. A
+                    # >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
@@ -501,7 +546,17 @@ class DisplaySyncManager:
                             self.write_status_file()
                         elif t == "sx":
                             # Vegas scroll-position sync — tiny message, renders locally
-                            self._latest_scroll_x = float(msg["x"])
+                            scroll_x = float(msg["x"])
+                            if not math.isfinite(scroll_x):
+                                # json.loads accepts the NaN/Infinity literals,
+                                # and float("nan") accepts the strings, so a
+                                # non-finite x reaches here intact. Left alone
+                                # it poisons every offset computed from it —
+                                # NaN comparisons are all false, so the
+                                # follower renders a frame it can never scroll
+                                # back from. Treat it as malformed.
+                                raise ValueError(f"non-finite scroll x: {msg['x']!r}")
+                            self._latest_scroll_x = scroll_x
                             self._last_leader_frame_time = time.time()
                             self._leader_ip = sender_ip
                             if self._follower_state == FollowerState.STANDALONE:
@@ -511,19 +566,22 @@ 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)
-                        pass
+                        continue
+
+                    if fire_new_cycle and self._on_new_cycle:
+                        self._on_new_cycle()
 
             except socket.timeout:
                 continue
             except Exception as exc:
                 self.logger.debug("Sync follower recv error: %s", exc)
+                time.sleep(0.1)
 
     def _follower_announce_loop(self) -> None:
         hw = self._hw_config

src/startup_validator.py

@@ -62,6 +62,9 @@ class StartupValidator:
         # Validate plugins if plugin manager is available
         if self.plugin_manager:
             self._validate_plugins()
+
+        # Warn when the running systemd unit no longer matches the repo's
+        self._validate_systemd_units()
         
         is_valid = len(self.errors) == 0
         
@@ -74,6 +77,80 @@ class StartupValidator:
         
         return (is_valid, self.errors.copy(), self.warnings.copy())
     
+    #: Units this project installs, and where each is installed to.
+    _UNITS = (
+        ("systemd/ledmatrix.service", "/etc/systemd/system/ledmatrix.service"),
+        ("systemd/ledmatrix-web.service", "/etc/systemd/system/ledmatrix-web.service"),
+    )
+
+    def _validate_systemd_units(self) -> None:
+        """Warn when an installed unit has drifted from the repo's template.
+
+        Nothing re-applies these after the first install. `git pull` -- which is
+        what the web UI's update button runs -- brings a new template into the
+        checkout, but nothing copies it to /etc/systemd/system and nothing runs
+        `systemctl daemon-reload`, so the unit that actually runs is whatever
+        first_time_install.sh wrote on day one.
+
+        That makes every hardening added to a unit inert on existing installs.
+        Measured on one rig: the installed unit was thirteen days older than the
+        repo's and differed in content, so a MemoryMax the repo had specified
+        was not being enforced at all -- `systemctl show` reported
+        MemoryMax=infinity.
+
+        A warning rather than an error, and certainly not a silent rewrite:
+        editing files under /etc and restarting services is the installer's job,
+        not something a display process should do to a machine while it boots.
+        The remedy is to re-run scripts/install/install_service.sh.
+        """
+        try:
+            project_root = Path(__file__).resolve().parent.parent
+            for template_rel, installed_path in self._UNITS:
+                template = project_root / template_rel
+                installed = Path(installed_path)
+                if not template.is_file() or not installed.is_file():
+                    continue
+
+                # The template carries placeholders the installer substitutes,
+                # so compare the substituted form rather than the raw file.
+                expected = template.read_text(encoding="utf-8")
+                expected = expected.replace("__PROJECT_ROOT_DIR__", str(project_root))
+                expected = expected.replace("__USER__", "root")
+
+                try:
+                    actual = installed.read_text(encoding="utf-8")
+                except PermissionError:
+                    continue
+
+                if self._unit_body(expected) != self._unit_body(actual):
+                    self.warnings.append(
+                        f"{installed.name} differs from {template_rel}; the "
+                        "installed unit is not refreshed by an update, so "
+                        "settings added to the template are not in effect. "
+                        "Re-run scripts/install/install_service.sh to apply them."
+                    )
+        except OSError as e:
+            self.logger.debug("Could not compare systemd units: %s", e)
+
+    @staticmethod
+    def _unit_body(text: str) -> str:
+        """A unit's meaningful lines, in order: no comments, no blanks.
+
+        Order is preserved deliberately. This used to sort, which made the
+        comparison insensitive to two changes that matter in a systemd unit:
+        repeated directives such as ExecStartPre= and ExecStartPost= run in
+        the order they appear, and a directive that moves between [Unit],
+        [Service] and [Install] means something different -- or nothing --
+        where it lands. A drift check that normalises those away reports no
+        drift for a unit that has genuinely changed.
+        """
+        lines = []
+        for line in text.splitlines():
+            line = line.strip()
+            if line and not line.startswith("#"):
+                lines.append(line)
+        return "\n".join(lines)
+
     def _validate_config(self) -> None:
         """Validate configuration files."""
         try:

src/web_interface/api_helpers.py

@@ -29,18 +29,16 @@ def success_response(
         Flask jsonify response
     """
     response_data = create_success_response(data, message, metadata)
-    
+
-    # Add request metadata if available
+    # Timing is merged into whatever the caller passed, without inventing a
-    if metadata is None:
+    # metadata block for responses that have neither.
-        metadata = {}
+    enriched = dict(metadata) if metadata is not None else {}
-    
-    # Add timing if request start time is available
     if hasattr(request, 'start_time'):
-        metadata['response_time_ms'] = int((time.time() - request.start_time) * 1000)
+        enriched['response_time_ms'] = int((time.time() - request.start_time) * 1000)
-    
+
-    if metadata:
+    if metadata is not None or enriched:
-        response_data['metadata'] = metadata
+        response_data['metadata'] = enriched
-    
+
     return jsonify(response_data)
 
 

src/web_interface/error_handler.py

@@ -161,14 +161,17 @@ def create_success_response(
         "status": "success"
     }
     
+    # All three use `is not None` rather than truthiness: "" and {} are
+    # values a caller chose to send, and dropping them silently would make
+    # the response shape depend on the data.
     if data is not None:
         response["data"] = data
-    
+
-    if message:
+    if message is not None:
         response["message"] = message
-    
+
-    if metadata:
+    if metadata is not None:
         response["metadata"] = metadata
-    
+
     return response
 

src/web_interface/errors.py

@@ -89,7 +89,11 @@ class WebInterfaceError:
         self.category = category or self._infer_category(error_code)
         self.details = details
         self.context = context or {}
-        self.suggested_fixes = suggested_fixes or self._get_default_suggestions(error_code)
+        # `is None`, not truthiness: an explicit [] means "this caller has
+        # no suggestions to offer", which the default list would override.
+        self.suggested_fixes = (
+            suggested_fixes if suggested_fixes is not None
+            else self._get_default_suggestions(error_code))
         self.original_error = original_error
     
     def _infer_category(self, error_code: ErrorCode) -> ErrorCategory:

src/web_interface/secret_helpers.py

@@ -5,7 +5,7 @@ Provides functions for identifying, masking, separating, and filtering
 secret fields in plugin configurations based on JSON Schema x-secret markers.
 """
 
-from typing import Any, Dict, Set, Tuple
+from typing import Any, Dict, Optional, Set, Tuple
 
 
 def find_secret_fields(properties: Dict[str, Any], prefix: str = '') -> Set[str]:
@@ -143,6 +143,12 @@ def mask_secret_fields(config: Dict[str, Any], schema_properties: Dict[str, Any]
     return result
 
 
+#: What a masked secret looks like on the wire. Named because the write path
+#: has to recognise it coming back: a client that renders the mask and posts
+#: it unchanged must not store the mask as if it were the secret.
+SECRET_MASK = '\u2022' * 8
+
+
 def mask_all_secret_values(config: Dict[str, Any]) -> Dict[str, Any]:
     """Blanket-mask every non-empty value in a secrets config dict.
 
@@ -156,15 +162,25 @@ def mask_all_secret_values(config: Dict[str, Any]) -> Dict[str, Any]:
     Returns:
         A copy with all real values replaced by ``'••••••••'``.
     """
-    masked: Dict[str, Any] = {}
+    return {k: _mask_value(v) for k, v in config.items()}
-    for k, v in config.items():
+
-        if isinstance(v, dict):
+
-            masked[k] = mask_all_secret_values(v)
+def _mask_value(value: Any) -> Any:
-        elif v not in (None, '') and not (isinstance(v, str) and v.startswith('YOUR_')):
+    """Mask one value, recursing through dicts and lists.
-            masked[k] = '••••••••'
+
-        else:
+    A list used to be masked as though it were a scalar, so
-            masked[k] = v
+    ``accounts: [{"name": "a", "token": "..."}]`` came back as a single
-    return masked
+    ``'••••••••'``. Nothing leaked, but the caller could no longer see how
+    many entries there were or any of their non-secret fields, and the raw
+    editor was shown a string where the file holds an array.
+    """
+    if isinstance(value, dict):
+        return {k: _mask_value(v) for k, v in value.items()}
+    if isinstance(value, list):
+        return [_mask_value(item) for item in value]
+    if value in (None, '') or (isinstance(value, str) and value.startswith('YOUR_')):
+        return value
+    return SECRET_MASK
 
 
 def remove_empty_secrets(secrets: Dict[str, Any]) -> Dict[str, Any]:
@@ -186,6 +202,133 @@ def remove_empty_secrets(secrets: Dict[str, Any]) -> Dict[str, Any]:
             nested = remove_empty_secrets(v)
             if nested:
                 result[k] = nested
+        elif isinstance(v, list):
+            # Lists used to fall through to the scalar branch below and be
+            # kept verbatim, blanks and all. Because lists merge by
+            # *replacement*, saving any unrelated setting then wrote
+            # [{"token": ""}, ...] straight over the stored list and
+            # destroyed every credential in it.
+            pruned = _prune_secret_list(v)
+            if pruned is not None:
+                result[k] = pruned
         elif v is not None and not (isinstance(v, str) and v.strip() == ''):
             result[k] = v
     return result
+
+
+def _prune_secret_list(items: list) -> Optional[list]:
+    """Strip blanks from inside a list of secrets, preserving every index.
+
+    The rest of the system treats a secrets list as *parallel* to the regular
+    one -- ``sec[i]`` holds the secret fields of item ``i``, and ``{}`` means
+    "item i has none" (see ConfigManager._strip_secrets_recursive). So an
+    emptied dict item stays ``{}``: putting ``None`` there makes that list stop
+    looking parallel, and the stripper then drops the whole key from the main
+    config, taking the non-secret fields with it.
+
+    A blank *scalar* becomes ``None``, meaning "no update at this index" --
+    :func:`merge_secrets` substitutes whatever is stored there. Returns
+    ``None`` when nothing in the list carries a real value, so the caller drops
+    the key and leaves the stored list untouched.
+    """
+    pruned: list = []
+    has_real_value = False
+    for item in items:
+        if isinstance(item, dict):
+            kept = remove_empty_secrets(item)
+            pruned.append(kept)
+            has_real_value = has_real_value or bool(kept)
+        elif isinstance(item, list):
+            sub = _prune_secret_list(item)
+            pruned.append(sub if sub is not None else [])
+            has_real_value = has_real_value or sub is not None
+        elif item is not None and not (isinstance(item, str) and item.strip() == ''):
+            pruned.append(item)
+            has_real_value = True
+        else:
+            pruned.append(None)
+    return pruned if has_real_value else None
+
+
+def merge_secrets(stored: Any, incoming: Any) -> Any:
+    """Merge submitted secrets over stored ones, element-wise inside lists.
+
+    ``deep_merge`` replaces a list wholesale. For secrets that is destructive:
+    an incoming list that carries a real value for one entry and ``None`` for
+    the rest would drop the stored credentials of every other entry. Here a
+    list merges by index, and ``None`` means "keep what is stored".
+
+    Entries are matched by *position*, which is what the config form gives us
+    -- there is no schema-declared identity to key on, and it is the same
+    contract ConfigManager._strip_secrets_recursive already relies on. The
+    incoming list's length wins, so deleting an item deletes its secrets;
+    an item the client left blank keeps whatever is stored at that index.
+    """
+    if isinstance(stored, dict) and isinstance(incoming, dict):
+        merged = dict(stored)
+        for key, value in incoming.items():
+            merged[key] = (merge_secrets(stored[key], value)
+                           if key in stored else value)
+        return merged
+    if isinstance(stored, list) and isinstance(incoming, list):
+        # The incoming list sets the length -- the regular config's list is
+        # authoritative about how many items exist, and this one runs parallel
+        # to it. Removing an entry must therefore remove its secrets too.
+        merged_list = []
+        for index, item in enumerate(incoming):
+            stored_item = stored[index] if index < len(stored) else None
+            merged_list.append(stored_item if item is None
+                               else merge_secrets(stored_item, item))
+        return merged_list
+    if incoming is None:
+        return stored
+    return incoming
+
+
+def strip_masked_values(secrets: Dict[str, Any]) -> Dict[str, Any]:
+    """Remove values a client echoed back rather than changed.
+
+    The counterpart to :func:`mask_all_secret_values`. A client that GETs the
+    masked secrets, edits one field and POSTs the whole object back is sending
+    ``SECRET_MASK`` for every field it did not touch. Storing those would
+    replace each untouched credential with eight bullet characters.
+
+    Drops the mask and, like :func:`remove_empty_secrets`, blank values -- so
+    the caller can merge the result onto what is already stored and have
+    "unchanged" mean unchanged. Empty nested dicts are pruned.
+    """
+    result: Dict[str, Any] = {}
+    for k, v in secrets.items():
+        if isinstance(v, dict):
+            nested = strip_masked_values(v)
+            if nested:
+                result[k] = nested
+        elif isinstance(v, list):
+            # A list is merged by replacement, not element by element -- there
+            # is no identity to match entries on -- so a list that still holds
+            # a mask cannot be merged safely: keeping it would store bullets,
+            # and keeping the submitted entries alone would drop whichever the
+            # client did not send back. Dropping the key leaves the stored
+            # list untouched, which is what an untouched list should do.
+            #
+            # The consequence, deliberately: editing one secret inside a list
+            # through this endpoint requires sending real values for all of
+            # them. Sending some masks leaves the whole list as it was.
+            if not _contains_mask(v):
+                result[k] = v
+        elif v is None:
+            continue
+        elif isinstance(v, str) and (v.strip() == '' or v == SECRET_MASK):
+            continue
+        else:
+            result[k] = v
+    return result
+
+
+def _contains_mask(value: Any) -> bool:
+    """True when a mask sentinel survives anywhere inside ``value``."""
+    if isinstance(value, dict):
+        return any(_contains_mask(v) for v in value.values())
+    if isinstance(value, list):
+        return any(_contains_mask(item) for item in value)
+    return value == SECRET_MASK

src/web_interface/validators.py

@@ -43,10 +43,15 @@ def validate_image_url(url: str) -> Tuple[bool, Optional[str]]:
     if any(handler in url_lower for handler in ['onerror=', 'onload=', 'onclick=']):
         return False, "Event handlers not allowed in URLs"
     
+    # Reject directory traversal anywhere, not only in relative paths:
+    # http://host/../secret is as much a traversal attempt as /../secret.
+    if '..' in url:
+        return False, "Invalid path: directory traversal not allowed"
+
     # Allow relative paths starting with /
     if url.startswith('/'):
-        # Validate it's a safe relative path (no directory traversal)
+        # // would be a protocol-relative URL, not a local path
-        if '..' in url or url.startswith('//'):
+        if url.startswith('//'):
             return False, "Invalid relative path"
         return True, None
     
@@ -104,10 +109,11 @@ def validate_file_upload(filename: str, max_size_mb: int = 10,
     if '..' in filename or '/' in filename or '\\' in filename:
         return False, "Filename contains invalid characters"
     
-    # Check extension if specified
+    # Check extension if specified. Both sides are lowercased: the caller's
+    # list is as likely to hold '.TTF' as the filename is.
     if allowed_extensions:
         file_ext = Path(filename).suffix.lower()
-        if file_ext not in allowed_extensions:
+        if file_ext not in [ext.lower() for ext in allowed_extensions]:
             return False, f"File extension must be one of: {', '.join(allowed_extensions)}"
     
     return True, None
@@ -147,7 +153,8 @@ def validate_numeric_range(value: float, min_val: Optional[float] = None,
     Returns:
         Tuple of (is_valid, error_message)
     """
-    if not isinstance(value, (int, float)):
+    # bool is an int subclass, so True would otherwise validate as 1.
+    if not isinstance(value, (int, float)) or isinstance(value, bool):
         return False, "Value must be a number"
     
     if min_val is not None and value < min_val:
@@ -183,11 +190,19 @@ def validate_string_length(text: str, min_length: Optional[int] = None,
 
 def sanitize_plugin_config(config: dict) -> dict:
     """
-    Sanitize plugin configuration input to prevent injection.
+    Restrict a plugin config to safe key names and value types.
-    
+
+    Drops keys that are not plain identifiers and values that are not
+    JSON-ish scalars, lists, or dicts, recursing into the latter two.
+
+    String values are returned **unescaped**: output escaping is the
+    template layer's job, and escaping here would store the escaped form
+    in config.json. Do not read this function as XSS protection for
+    rendered output.
+
     Args:
         config: Configuration dictionary
-    
+
     Returns:
         Sanitized configuration dictionary
     """

systemd/ledmatrix.service

@@ -8,6 +8,18 @@ Type=simple
 User=root
 WorkingDirectory=__PROJECT_ROOT_DIR__
 Environment=PYTHONDONTWRITEBYTECODE=1
+# glibc gives each allocating thread its own malloc arena, up to 8 x CPU count,
+# and an arena that has grown is never handed back to the OS. This process runs
+# 9 threads on a 3-core Pi, so the ceiling is 24 arenas -- and a rig measured at
+# 1030 MB resident held 23 large anonymous mappings on 64 MB-aligned addresses,
+# 920 MB of them, while the live data it was actually holding (widest scroll
+# strip seen: 35,746 x 64) accounts for roughly 15 MB. That gap is arena bloat,
+# not leaked objects: RSS was flat across repeated sampling, not climbing.
+#
+# Capping the arenas trades a little allocator concurrency for a large amount of
+# resident memory on a device that has neither to spare. 2 is the usual value;
+# raise it if frame times regress.
+Environment=MALLOC_ARENA_MAX=2
 ExecStart=/usr/bin/python3 __PROJECT_ROOT_DIR__/run.py
 # Restart=always, not on-failure: run.py exiting 0 (a clean shutdown path taken
 # for a reason that no longer applies, e.g. a config reload) would otherwise leave

test/_api_v3_test_helpers.py

@@ -0,0 +1,75 @@
+"""
+Shared scaffolding for api_v3 blueprint tests.
+
+Not a test module (the leading underscore keeps pytest from collecting
+it). It is the pytest-fixture equivalent of ``_make_client()`` in
+test_uninstall_and_reconcile_endpoint.py, which is unittest-style and
+requires ``self.addCleanup``.
+
+The api_v3 blueprint keeps its managers as attributes on a module-level
+singleton, not in Flask app state, so replacing them with mocks leaks
+into every later test that imports api_v3 unless the originals are put
+back. ``api_v3_client`` snapshots and restores them around each test.
+"""
+
+from unittest.mock import MagicMock
+
+import pytest
+from flask import Flask
+
+
+# Every manager attribute the blueprint reads. Anything missing here keeps
+# whatever a previously-run test left on the singleton.
+API_V3_MANAGER_ATTRS = (
+    'config_manager', 'plugin_manager', 'plugin_store_manager',
+    'plugin_state_manager', 'saved_repositories_manager', 'schema_manager',
+    'operation_queue', 'operation_history', 'cache_manager',
+)
+
+_SENTINEL = object()
+
+
+def build_app(blueprint):
+    app = Flask(__name__)
+    app.config['TESTING'] = True
+    app.config['SECRET_KEY'] = 'test'
+    app.register_blueprint(blueprint, url_prefix='/api/v3')
+    return app
+
+
+@pytest.fixture
+def api_v3_module():
+    """The api_v3 module with every manager replaced by a MagicMock.
+
+    Restores the original attributes afterwards. Tests point individual
+    managers at real objects (a ConfigManager over tmp_path, say) or set
+    them to None to exercise the not-initialized branches.
+    """
+    from web_interface.blueprints import api_v3 as module
+
+    originals = {
+        name: getattr(module.api_v3, name, _SENTINEL)
+        for name in API_V3_MANAGER_ATTRS
+    }
+    for name in API_V3_MANAGER_ATTRS:
+        setattr(module.api_v3, name, MagicMock())
+    # Default to the direct path; queue tests opt in explicitly.
+    module.api_v3.operation_queue = None
+
+    yield module
+
+    for name, original in originals.items():
+        if original is _SENTINEL:
+            if hasattr(module.api_v3, name):
+                try:
+                    delattr(module.api_v3, name)
+                except AttributeError:
+                    pass
+        else:
+            setattr(module.api_v3, name, original)
+
+
+@pytest.fixture
+def api_v3_client(api_v3_module):
+    """Flask test client wired to the mocked blueprint."""
+    return build_app(api_v3_module.api_v3).test_client()

test/test_api_v3_calendar_credentials.py

@@ -0,0 +1,226 @@
+"""
+Endpoint tests for POST /plugins/calendar/upload-credentials.
+
+The endpoint takes an uploaded Google OAuth credentials file, writes it
+into the calendar plugin's directory as credentials.json at mode 0600, and
+copies any previous file aside first. It had no tests.
+
+Regression coverage for two fixed bugs:
+- The OAuth-shape check sat inside `except Exception: pass`, so a valid
+  JSON document that is not an object — a bare `42`, a list, a string —
+  raised TypeError on the membership test, was swallowed, and got saved
+  as credentials.json anyway.
+- Each overwrite created a timestamped backup and nothing ever removed
+  them, so every re-upload left another complete copy of the user's OAuth
+  client credentials in the plugin directory, indefinitely.
+"""
+
+import io
+import json
+import os
+import stat
+import sys
+import time
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
+
+URL = "/api/v3/plugins/calendar/upload-credentials"
+
+VALID_CREDENTIALS = {
+    "installed": {
+        "client_id": "abc.apps.googleusercontent.com",
+        "client_secret": "shh",
+        "redirect_uris": ["http://localhost"],
+    }
+}
+
+
+@pytest.fixture
+def plugin_dir(tmp_path, api_v3_module):
+    directory = tmp_path / "plugins" / "calendar"
+    directory.mkdir(parents=True)
+    api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(directory)
+    return directory
+
+
+def upload(client, content, filename="credentials.json"):
+    # bytes are sent verbatim (to exercise malformed input); anything else
+    # is serialized, so None becomes the JSON literal null rather than an
+    # empty body.
+    payload = content if isinstance(content, bytes) else json.dumps(content).encode()
+    return client.post(
+        URL,
+        data={"file": (io.BytesIO(payload), filename)},
+        content_type="multipart/form-data",
+    )
+
+
+def backups(plugin_dir):
+    return sorted(plugin_dir.glob("credentials.json.backup.*"))
+
+
+class TestRequestValidation:
+    def test_no_file_part_is_a_400(self, api_v3_client, plugin_dir):
+        response = api_v3_client.post(URL, data={}, content_type="multipart/form-data")
+        assert response.status_code == 400
+        assert "No file provided" in response.get_json()["message"]
+
+    def test_empty_filename_is_a_400(self, api_v3_client, plugin_dir):
+        response = upload(api_v3_client, VALID_CREDENTIALS, filename="")
+        assert response.status_code == 400
+
+    @pytest.mark.parametrize("filename", ["creds.txt", "creds.pem", "creds"])
+    def test_non_json_extension_is_a_400(self, api_v3_client, plugin_dir, filename):
+        response = upload(api_v3_client, VALID_CREDENTIALS, filename=filename)
+        assert response.status_code == 400
+        assert "JSON file" in response.get_json()["message"]
+
+    def test_uppercase_json_extension_accepted(self, api_v3_client, plugin_dir):
+        assert upload(api_v3_client, VALID_CREDENTIALS,
+                      filename="CREDENTIALS.JSON").status_code == 200
+
+    def test_oversized_file_is_a_400(self, api_v3_client, plugin_dir):
+        response = upload(api_v3_client, b"x" * (1024 * 1024 + 1))
+        assert response.status_code == 400
+        assert "1MB" in response.get_json()["message"]
+        assert not (plugin_dir / "credentials.json").exists()
+
+    def test_invalid_json_is_a_400(self, api_v3_client, plugin_dir):
+        response = upload(api_v3_client, b"{not json")
+        assert response.status_code == 400
+        assert "not valid JSON" in response.get_json()["message"]
+        assert not (plugin_dir / "credentials.json").exists()
+
+    def test_missing_plugin_directory_is_a_404(self, api_v3_client, api_v3_module, tmp_path):
+        api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(
+            tmp_path / "not-installed")
+        assert upload(api_v3_client, VALID_CREDENTIALS).status_code == 404
+
+
+class TestOAuthShapeValidation:
+    def test_installed_key_accepted(self, api_v3_client, plugin_dir):
+        assert upload(api_v3_client, VALID_CREDENTIALS).status_code == 200
+
+    def test_web_key_accepted(self, api_v3_client, plugin_dir):
+        assert upload(api_v3_client, {"web": {"client_id": "x"}}).status_code == 200
+
+    def test_object_without_oauth_keys_is_a_400(self, api_v3_client, plugin_dir):
+        response = upload(api_v3_client, {"something": "else"})
+        assert response.status_code == 400
+        assert "valid Google OAuth" in response.get_json()["message"]
+        assert not (plugin_dir / "credentials.json").exists()
+
+    @pytest.mark.parametrize("content", [42, "a string", [1, 2, 3], True, None])
+    def test_valid_json_that_is_not_an_object_is_rejected(
+            self, api_v3_client, plugin_dir, content):
+        # Regression: `'installed' not in 42` raises TypeError, which the
+        # bare `except Exception: pass` swallowed — the file was then saved
+        # as credentials.json despite being unusable as credentials.
+        response = upload(api_v3_client, content)
+        assert response.status_code == 400
+        assert "valid Google OAuth" in response.get_json()["message"]
+        assert not (plugin_dir / "credentials.json").exists()
+
+
+class TestSaving:
+    def test_file_written_with_contents_intact(self, api_v3_client, plugin_dir):
+        response = upload(api_v3_client, VALID_CREDENTIALS)
+        assert response.status_code == 200
+        saved = json.loads((plugin_dir / "credentials.json").read_text())
+        assert saved == VALID_CREDENTIALS
+
+    def test_response_reports_the_path(self, api_v3_client, plugin_dir):
+        body = upload(api_v3_client, VALID_CREDENTIALS).get_json()
+        assert body["path"].endswith("credentials.json")
+
+    def test_permissions_are_owner_only(self, api_v3_client, plugin_dir):
+        upload(api_v3_client, VALID_CREDENTIALS)
+        mode = stat.S_IMODE((plugin_dir / "credentials.json").stat().st_mode)
+        assert mode == 0o600
+
+    def test_first_upload_creates_no_backup(self, api_v3_client, plugin_dir):
+        upload(api_v3_client, VALID_CREDENTIALS)
+        assert backups(plugin_dir) == []
+
+    def test_overwrite_backs_up_the_previous_file(self, api_v3_client, plugin_dir):
+        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"old": 1}}))
+        upload(api_v3_client, VALID_CREDENTIALS)
+        assert len(backups(plugin_dir)) == 1
+        assert json.loads(backups(plugin_dir)[0].read_text()) == {"installed": {"old": 1}}
+        assert json.loads((plugin_dir / "credentials.json").read_text()) == VALID_CREDENTIALS
+
+
+class TestBackupPruning:
+    def _seed(self, plugin_dir, count):
+        """Create `count` backups with distinct, increasing mtimes."""
+        now = int(time.time())
+        for i in range(count):
+            path = plugin_dir / f"credentials.json.backup.{now - (count - i) * 10}"
+            path.write_text(json.dumps({"installed": {"gen": i}}))
+            os.utime(path, (now - (count - i) * 10, now - (count - i) * 10))
+
+    def test_old_backups_are_pruned(self, api_v3_client, plugin_dir):
+        # Regression: nothing ever removed these, so a plugin directory
+        # accumulated one full copy of the user's OAuth credentials per
+        # re-upload, forever.
+        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"cur": 1}}))
+        self._seed(plugin_dir, 7)
+        assert len(backups(plugin_dir)) == 7
+
+        upload(api_v3_client, VALID_CREDENTIALS)
+        assert len(backups(plugin_dir)) == 5
+
+    def test_the_newest_backups_are_the_ones_kept(self, api_v3_client, plugin_dir):
+        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"cur": 1}}))
+        self._seed(plugin_dir, 7)
+
+        upload(api_v3_client, VALID_CREDENTIALS)
+        remaining = backups(plugin_dir)
+        # The just-created backup (of "cur") plus the four newest seeds.
+        contents = [json.loads(p.read_text()) for p in remaining]
+        assert {"installed": {"cur": 1}} in contents
+        assert {"installed": {"gen": 0}} not in contents  # oldest seed gone
+
+    def test_under_the_limit_nothing_is_removed(self, api_v3_client, plugin_dir):
+        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"cur": 1}}))
+        self._seed(plugin_dir, 2)
+        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, 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}})
+            os.utime(plugin_dir / "credentials.json",
+                     (clock["now"], clock["now"]))
+        remaining = backups(plugin_dir)
+        assert len(remaining) == 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):
+        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"cur": 1}}))
+        self._seed(plugin_dir, 7)
+
+        def refuse(self):
+            raise OSError("read-only filesystem")
+        monkeypatch.setattr(Path, "unlink", refuse)
+
+        # Pruning is housekeeping; failing it must not lose the upload.
+        assert upload(api_v3_client, VALID_CREDENTIALS).status_code == 200

test/test_api_v3_music_auth_endpoints.py

@@ -0,0 +1,302 @@
+"""
+Endpoint tests for /plugins/authenticate/spotify and .../ytm.
+
+The Spotify step-2 handler writes a Python wrapper script to a temp file
+with the user's redirect URL embedded in it, then runs that file through
+subprocess. That is the most dangerous shape in the blueprint and had no
+tests: the URL is user input reaching generated source code.
+
+The two endpoints are NOT symmetrical, despite the matching names. Only
+Spotify has a two-step flow, a wrapper script, and a redirect_url; YTM
+just runs its script directly.
+
+Regression coverage for one fixed bug: the wrapper file was unlinked in
+the success/failure branch and again in the TimeoutExpired handler, so
+any other failure from subprocess.run — the interpreter missing, a fork
+failure, an interrupted call — left a temp file containing the user's
+redirect URL behind.
+"""
+
+import ast
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
+
+
+@pytest.fixture
+def plugin_dir(tmp_path, api_v3_module):
+    """A plugin directory containing both auth scripts."""
+    directory = tmp_path / "plugins" / "ledmatrix-music"
+    directory.mkdir(parents=True)
+    (directory / "authenticate_spotify.py").write_text("print('spotify')\n")
+    (directory / "authenticate_ytm.py").write_text("print('ytm')\n")
+    api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(directory)
+    return directory
+
+
+def completed(returncode=0, stdout="ok", stderr=""):
+    return subprocess.CompletedProcess(
+        args=["python3"], returncode=returncode, stdout=stdout, stderr=stderr)
+
+
+class TestSpotifyPreconditions:
+    URL = "/api/v3/plugins/authenticate/spotify"
+
+    def test_missing_plugin_directory_is_404(self, api_v3_client, api_v3_module, tmp_path):
+        api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(
+            tmp_path / "not-installed")
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 404
+        assert response.get_json()["message"] == "Plugin not found"
+
+    def test_none_plugin_directory_is_404(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = None
+        assert api_v3_client.post(self.URL, json={}).status_code == 404
+
+    def test_missing_auth_script_is_404(self, api_v3_client, plugin_dir):
+        (plugin_dir / "authenticate_spotify.py").unlink()
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 404
+        assert "script not found" in response.get_json()["message"]
+
+
+class TestSpotifyStepTwo:
+    """redirect_url present — the wrapper-script path."""
+
+    URL = "/api/v3/plugins/authenticate/spotify"
+
+    def test_success(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run", return_value=completed(0, "done")):
+            response = api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
+        assert response.status_code == 200
+        body = response.get_json()
+        assert body["status"] == "success"
+        assert body["output"] == "done"
+
+    def test_script_failure_is_a_400_with_combined_output(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run", return_value=completed(1, "out", "err")):
+            response = api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
+        assert response.status_code == 400
+        assert response.get_json()["output"] == "outerr"
+
+    def test_timeout_is_a_408(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run",
+                          side_effect=subprocess.TimeoutExpired("python3", 120)):
+            response = api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
+        assert response.status_code == 408
+        assert "timed out" in response.get_json()["message"]
+
+    def test_runs_a_list_argv_never_a_shell(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run", return_value=completed()) as run:
+            api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
+        args, kwargs = run.call_args
+        assert isinstance(args[0], list)
+        assert args[0][0] == "python3"
+        assert kwargs.get("shell") in (None, False)
+
+    def test_timeout_is_bounded(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run", return_value=completed()) as run:
+            api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
+        assert run.call_args.kwargs["timeout"] == 120
+
+
+class TestSpotifyWrapperCleanup:
+    URL = "/api/v3/plugins/authenticate/spotify"
+
+    def _wrapper_paths_after(self, api_v3_client, run_mock):
+        """Run the endpoint and return the wrapper path subprocess saw."""
+        seen = {}
+
+        def capture(args, **kwargs):
+            seen["path"] = args[1]
+            return run_mock(args, **kwargs)
+
+        with patch.object(subprocess, "run", side_effect=capture):
+            api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
+        return seen["path"]
+
+    def test_removed_after_success(self, api_v3_client, plugin_dir):
+        path = self._wrapper_paths_after(api_v3_client, lambda *a, **kw: completed())
+        assert not os.path.exists(path)
+
+    def test_removed_after_script_failure(self, api_v3_client, plugin_dir):
+        path = self._wrapper_paths_after(
+            api_v3_client, lambda *a, **kw: completed(1, "out", "err"))
+        assert not os.path.exists(path)
+
+    def test_removed_after_timeout(self, api_v3_client, plugin_dir):
+        def raise_timeout(*a, **kw):
+            raise subprocess.TimeoutExpired("python3", 120)
+        path = self._wrapper_paths_after(api_v3_client, raise_timeout)
+        assert not os.path.exists(path)
+
+    def test_removed_when_subprocess_cannot_start(self, api_v3_client, plugin_dir):
+        # Regression: cleanup lived in the success/failure branch and in the
+        # TimeoutExpired handler only. An OSError from subprocess.run itself
+        # — no interpreter, fork failure — skipped both and left the wrapper,
+        # which contains the user's redirect URL, on disk.
+        def raise_oserror(*a, **kw):
+            raise OSError("[Errno 12] Cannot allocate memory")
+        path = self._wrapper_paths_after(api_v3_client, raise_oserror)
+        assert not os.path.exists(path)
+
+
+class TestSpotifyRedirectUrlIsNotInjectable:
+    """The wrapper embeds redirect_url into generated Python source."""
+
+    URL = "/api/v3/plugins/authenticate/spotify"
+
+    ADVERSARIAL = [
+        '''http://cb/?code=x"''',
+        """http://cb/?code=x'""",
+        'http://cb/?code=x\\',
+        'http://cb/?code=x\nimport os; os.system("id")',
+        'http://cb/?code=x"""\nimport os\n"""',
+        "http://cb/?code=x'''",
+        'http://cb/?code=x\\"\\n',
+        '"; import os; os.system("id"); "',
+    ]
+
+    def _wrapper_source(self, api_v3_client, redirect_url):
+        captured = {}
+
+        def capture(args, **kwargs):
+            captured["source"] = Path(args[1]).read_text()
+            return completed()
+
+        with patch.object(subprocess, "run", side_effect=capture):
+            api_v3_client.post(self.URL, json={"redirect_url": redirect_url})
+        return captured["source"]
+
+    @pytest.mark.parametrize("redirect_url", ADVERSARIAL)
+    def test_wrapper_is_still_valid_python(self, api_v3_client, plugin_dir, redirect_url):
+        # If escaping failed, the generated file would not parse at all.
+        source = self._wrapper_source(api_v3_client, redirect_url)
+        ast.parse(source)
+
+    @pytest.mark.parametrize("redirect_url", ADVERSARIAL)
+    def test_url_survives_as_one_string_literal(
+            self, api_v3_client, plugin_dir, redirect_url):
+        # Stronger than "it parses": the URL must still be a single string
+        # assigned to redirect_url, not code that escaped into statements.
+        source = self._wrapper_source(api_v3_client, redirect_url)
+        tree = ast.parse(source)
+        assigned = [
+            node.value.value for node in ast.walk(tree)
+            if isinstance(node, ast.Assign)
+            and isinstance(node.value, ast.Constant)
+            and any(getattr(t, "id", None) == "redirect_url" for t in node.targets)
+        ]
+        assert assigned == [redirect_url.strip()]
+
+    def test_injected_call_does_not_become_a_statement(self, api_v3_client, plugin_dir):
+        source = self._wrapper_source(
+            api_v3_client, 'http://cb/\nimport os; os.system("id")')
+        tree = ast.parse(source)
+        imported = {
+            alias.name for node in ast.walk(tree)
+            if isinstance(node, ast.Import) for alias in node.names
+        }
+        # The wrapper legitimately imports sys, subprocess and os; what it
+        # must not gain is a *call* smuggled in through the URL.
+        calls = [
+            node for node in ast.walk(tree)
+            if isinstance(node, ast.Call)
+            and isinstance(node.func, ast.Attribute)
+            and node.func.attr == "system"
+        ]
+        assert calls == []
+
+
+class TestSpotifyStepOne:
+    """No redirect_url — the OAuth-URL path, which imports the script."""
+
+    URL = "/api/v3/plugins/authenticate/spotify"
+
+    def test_script_without_credentials_helper_is_an_error(
+            self, api_v3_client, plugin_dir):
+        # The stub script defines neither get_auth_url nor
+        # load_spotify_credentials, so no URL can be produced.
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code in (400, 500)
+        assert response.get_json()["status"] == "error"
+
+    def test_unusable_credentials_do_not_leak_into_the_response(
+            self, api_v3_client, plugin_dir):
+        (plugin_dir / "authenticate_spotify.py").write_text(
+            "def load_spotify_credentials():\n"
+            "    return ('id-abc', 'super-secret-value', None)\n"
+        )
+        response = api_v3_client.post(self.URL, json={})
+        assert "super-secret-value" not in response.get_data(as_text=True)
+
+    def test_script_raising_on_import_is_handled(self, api_v3_client, plugin_dir):
+        (plugin_dir / "authenticate_spotify.py").write_text("raise RuntimeError('boom')\n")
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 500
+        assert response.get_json()["status"] == "error"
+
+    def test_bodyless_post_reaches_step_one(self, api_v3_client, plugin_dir):
+        # Covered by the silent=True fix: previously a 500 from body parsing.
+        response = api_v3_client.post(self.URL)
+        assert response.status_code in (400, 500)
+        assert response.get_json()["status"] == "error"
+
+    def test_whitespace_redirect_url_is_treated_as_absent(
+            self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run", return_value=completed()) as run:
+            api_v3_client.post(self.URL, json={"redirect_url": "   "})
+        # Step 2 never runs, so no wrapper is executed.
+        run.assert_not_called()
+
+
+class TestYouTubeMusic:
+    """No wrapper script and no redirect_url — deliberately not symmetric."""
+
+    URL = "/api/v3/plugins/authenticate/ytm"
+
+    def test_missing_plugin_directory_is_404(self, api_v3_client, api_v3_module, tmp_path):
+        api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(
+            tmp_path / "not-installed")
+        assert api_v3_client.post(self.URL).status_code == 404
+
+    def test_missing_script_is_404(self, api_v3_client, plugin_dir):
+        (plugin_dir / "authenticate_ytm.py").unlink()
+        response = api_v3_client.post(self.URL)
+        assert response.status_code == 404
+        assert "script not found" in response.get_json()["message"]
+
+    def test_success(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run", return_value=completed(0, "authorized")):
+            response = api_v3_client.post(self.URL)
+        assert response.status_code == 200
+        assert response.get_json()["output"] == "authorized"
+
+    def test_failure_is_a_400_with_combined_output(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run", return_value=completed(1, "out", "err")):
+            response = api_v3_client.post(self.URL)
+        assert response.status_code == 400
+        assert response.get_json()["output"] == "outerr"
+
+    def test_timeout_is_a_408(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run",
+                          side_effect=subprocess.TimeoutExpired("python3", 60)):
+            assert api_v3_client.post(self.URL).status_code == 408
+
+    def test_runs_the_script_directly_without_a_shell(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run", return_value=completed()) as run:
+            api_v3_client.post(self.URL)
+        args, kwargs = run.call_args
+        assert args[0][0] == "python3"
+        assert args[0][1].endswith("authenticate_ytm.py")
+        assert kwargs.get("shell") in (None, False)
+        assert kwargs["timeout"] == 60

test/test_api_v3_optional_body.py

@@ -0,0 +1,136 @@
+"""
+Regression tests: POST endpoints whose body is optional must accept a
+request that has no body at all.
+
+Six handlers in api_v3 read their body as ``request.get_json() or {}``.
+The ``or {}`` states the intent plainly — every field is optional, so a
+bodyless POST should fall back to defaults. But ``get_json()`` without
+``silent=True`` raises ``UnsupportedMediaType`` when the request carries
+no JSON Content-Type, and it raises *before* ``or {}`` is evaluated. Each
+handler's catch-all then turned that into a 500.
+
+So the natural way to call these endpoints — a POST with no body, which
+is what curl, a fetch() without options, and most HTTP clients send by
+default — failed on every one of them. The shipped UI always sends a JSON
+object, which is why this went unnoticed.
+
+This file covers the endpoints whose bodyless behaviour is not already
+tested in their own suite.
+"""
+
+import re
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
+
+
+class TestOnDemandStart:
+    URL = "/api/v3/display/on-demand/start"
+
+    def test_bodyless_post_is_not_a_server_error(self, api_v3_client, api_v3_module):
+        response = api_v3_client.post(self.URL)
+        # The endpoint may still reject the request on its own terms (no
+        # plugin_id, nothing to display); what it must not do is fail with
+        # a 500 raised out of body parsing.
+        assert response.status_code != 500
+
+    def test_json_body_still_works(self, api_v3_client, api_v3_module):
+        assert api_v3_client.post(self.URL, json={}).status_code != 500
+
+
+class TestResetPluginConfig:
+    URL = "/api/v3/plugins/config/reset"
+
+    def test_bodyless_post_is_not_a_server_error(self, api_v3_client, api_v3_module):
+        assert api_v3_client.post(self.URL).status_code != 500
+
+    def test_json_body_still_works(self, api_v3_client, api_v3_module):
+        assert api_v3_client.post(self.URL, json={}).status_code != 500
+
+
+class TestDeleteOfTheDayJson:
+    URL = "/api/v3/plugins/of-the-day/json/delete"
+
+    def test_bodyless_post_is_not_a_server_error(self, api_v3_client, api_v3_module):
+        assert api_v3_client.post(self.URL).status_code != 500
+
+    def test_json_body_still_works(self, api_v3_client, api_v3_module):
+        assert api_v3_client.post(self.URL, json={}).status_code != 500
+
+
+class TestPluginLimits:
+    URL = "/api/v3/plugins/clock/limits"
+
+    def test_bodyless_post_is_not_a_server_error(self, api_v3_client, api_v3_module):
+        assert api_v3_client.post(self.URL).status_code != 500
+
+
+class TestMissingBodyGivesTheDeclaredError:
+    """Handlers that answer "No data provided" must actually be able to.
+
+    A second group of handlers reads `data = request.get_json()` and then
+    guards with `if not data: return 400`. That guard is unreachable for a
+    request with no JSON body, because get_json() raises first — so the
+    caller got a 500 "an error occurred; see logs for details" instead of
+    the 400 the handler plainly intends to send.
+    """
+
+    @pytest.mark.parametrize("url", [
+        "/api/v3/plugins/install",
+        "/api/v3/plugins/install-from-url",
+        "/api/v3/plugins/registry-from-url",
+        "/api/v3/config/raw/main",
+        "/api/v3/config/raw/secrets",
+        "/api/v3/cache/delete",
+    ])
+    def test_bodyless_post_gets_a_400_not_a_500(self, api_v3_client, api_v3_module, url):
+        response = api_v3_client.post(url)
+        assert response.status_code == 400, (
+            f"{url} answered {response.status_code}: "
+            f"{response.get_data(as_text=True)[:200]}")
+
+    @pytest.mark.parametrize("url", [
+        "/api/v3/plugins/install",
+        "/api/v3/config/raw/main",
+    ])
+    def test_malformed_json_gets_a_400_not_a_500(self, api_v3_client, api_v3_module, url):
+        response = api_v3_client.post(
+            url, data="{not json", content_type="application/json")
+        assert response.status_code == 400
+
+
+class TestNoBodyReadContradictsItsOwnGuard:
+    SOURCE = Path(__file__).parent.parent / "web_interface/blueprints/api_v3.py"
+
+    def test_no_or_default_read_is_unguarded(self):
+        """`get_json() or <default>` is a contradiction without silent=True.
+
+        Writing `or {}` declares the body optional; omitting silent=True
+        means the call raises before the default can apply.
+        """
+        offenders = [
+            line.strip() for line in self.SOURCE.read_text().splitlines()
+            if "request.get_json()" in line and " or " in line
+        ]
+        assert offenders == [], (
+            "these reads declare a default but raise before reaching it; "
+            f"use get_json(silent=True): {offenders}")
+
+    def test_no_not_data_guard_is_unreachable(self):
+        """A `if not data:` guard needs a read that can actually return None."""
+        lines = self.SOURCE.read_text().splitlines()
+        offenders = []
+        for i, line in enumerate(lines):
+            if re.search(r"=\s*request\.get_json\(\)\s*$", line):
+                window = "\n".join(lines[i + 1:i + 3])
+                if re.search(r"if\s+(not\s+data\b|data\s+is\s+None)", window):
+                    offenders.append(f"line {i + 1}: {line.strip()}")
+        assert offenders == [], (
+            "these handlers guard on a missing body but raise before the "
+            f"guard runs; use get_json(silent=True): {offenders}")

test/test_api_v3_plugin_install_endpoints.py

@@ -0,0 +1,302 @@
+"""
+Endpoint tests for POST /plugins/install and POST /plugins/install-from-url.
+
+Both were only ever tested at the PluginStoreManager layer, so the route
+logic — the queue-vs-direct branch, schema invalidation, plugin discovery,
+state and history recording — was unexercised.
+
+/plugins/install carries the same install logic twice: once inside the
+operation-queue callback and once in the direct fallback. The paired
+tests below assert both branches produce the same side effects, so the
+duplication cannot quietly drift.
+"""
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
+
+INSTALL = "/api/v3/plugins/install"
+FROM_URL = "/api/v3/plugins/install-from-url"
+
+
+@pytest.fixture
+def queued(api_v3_module):
+    """Enable the operation queue and run its callback synchronously."""
+    queue = MagicMock()
+
+    def enqueue(operation_type, plugin_id, operation_callback=None):
+        queue.callback_result = operation_callback(MagicMock())
+        return "op-123"
+
+    queue.enqueue_operation.side_effect = enqueue
+    api_v3_module.api_v3.operation_queue = queue
+    return queue
+
+
+def side_effects(module):
+    """The manager calls a successful install is expected to make."""
+    api = module.api_v3
+    return {
+        "schema_invalidated": api.schema_manager.invalidate_cache.call_args_list,
+        "discovered": api.plugin_manager.discover_plugins.call_count,
+        "loaded": api.plugin_manager.load_plugin.call_args_list,
+        "state_set": api.plugin_state_manager.set_plugin_installed.call_args_list,
+        "history": api.operation_history.record_operation.call_args_list,
+    }
+
+
+class TestInstallValidation:
+    def test_uninitialized_store_manager_is_a_500(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager = None
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        assert response.status_code == 500
+        assert "not initialized" in response.get_json()["message"]
+
+    def test_missing_plugin_id_is_a_400(self, api_v3_client, api_v3_module):
+        response = api_v3_client.post(INSTALL, json={})
+        assert response.status_code == 400
+        assert "plugin_id required" in response.get_json()["message"]
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.assert_not_called()
+
+    def test_empty_body_is_a_400(self, api_v3_client, api_v3_module):
+        assert api_v3_client.post(INSTALL, json=None).status_code == 400
+
+
+class TestInstallDirectPath:
+    """operation_queue is None — the fallback branch."""
+
+    def test_success(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        assert response.status_code == 200
+        assert response.get_json()["status"] == "success"
+
+    def test_success_side_effects(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
+        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        effects = side_effects(api_v3_module)
+        assert effects["schema_invalidated"] == [(("clock",), {})]
+        assert effects["discovered"] == 1
+        assert effects["loaded"] == [(("clock",), {})]
+        assert effects["state_set"] == [(("clock",), {})]
+        assert effects["history"][0].kwargs["status"] == "success"
+
+    def test_branch_forwarded_to_the_manager(self, api_v3_client, api_v3_module):
+        manager = api_v3_module.api_v3.plugin_store_manager
+        manager.install_plugin.return_value = True
+        api_v3_client.post(INSTALL, json={"plugin_id": "clock", "branch": "dev"})
+        manager.install_plugin.assert_called_once_with("clock", branch="dev")
+
+    def test_branch_named_in_the_message(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock", "branch": "dev"})
+        assert "(branch: dev)" in response.get_json()["message"]
+
+    def test_failure_is_a_500(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        assert response.status_code == 500
+        assert "Failed to install" in response.get_json()["message"]
+
+    def test_failure_mentions_missing_registry_entry(self, api_v3_client, api_v3_module):
+        manager = api_v3_module.api_v3.plugin_store_manager
+        manager.install_plugin.return_value = False
+        manager.get_plugin_info.return_value = None
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "ghost"})
+        assert "not found in registry" in response.get_json()["message"]
+
+    def test_failure_omits_registry_note_when_plugin_is_known(
+            self, api_v3_client, api_v3_module):
+        manager = api_v3_module.api_v3.plugin_store_manager
+        manager.install_plugin.return_value = False
+        manager.get_plugin_info.return_value = {"id": "clock"}
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        assert "not found in registry" not in response.get_json()["message"]
+
+    def test_failure_recorded_in_history(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
+        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        record = api_v3_module.api_v3.operation_history.record_operation.call_args
+        assert record.kwargs["status"] == "failed"
+
+    def test_no_side_effects_on_failure(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
+        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        effects = side_effects(api_v3_module)
+        assert effects["schema_invalidated"] == []
+        assert effects["loaded"] == []
+        assert effects["state_set"] == []
+
+
+class TestInstallQueuedPath:
+    """operation_queue present — the callback branch."""
+
+    def test_returns_an_operation_id(self, api_v3_client, api_v3_module, queued):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        assert response.status_code == 200
+        assert response.get_json()["data"]["operation_id"] == "op-123"
+
+    def test_message_says_queued(self, api_v3_client, api_v3_module, queued):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        assert "queued" in response.get_json()["message"]
+
+    def test_callback_success_side_effects(self, api_v3_client, api_v3_module, queued):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
+        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        effects = side_effects(api_v3_module)
+        assert effects["schema_invalidated"] == [(("clock",), {})]
+        assert effects["discovered"] == 1
+        assert effects["loaded"] == [(("clock",), {})]
+        assert effects["state_set"] == [(("clock",), {})]
+        assert effects["history"][0].kwargs["status"] == "success"
+
+    def test_callback_reports_success(self, api_v3_client, api_v3_module, queued):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
+        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        assert queued.callback_result["success"] is True
+
+    def test_callback_failure_raises_for_the_queue(self, api_v3_client, api_v3_module, queued):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
+        # The callback signals failure by raising, so the queue can mark the
+        # operation failed; the route's catch-all turns it into a 500.
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        assert response.status_code == 500
+
+    def test_callback_failure_recorded_in_history(self, api_v3_client, api_v3_module, queued):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
+        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        record = api_v3_module.api_v3.operation_history.record_operation.call_args
+        assert record.kwargs["status"] == "failed"
+
+    def test_branch_forwarded_from_the_callback(self, api_v3_client, api_v3_module, queued):
+        manager = api_v3_module.api_v3.plugin_store_manager
+        manager.install_plugin.return_value = True
+        api_v3_client.post(INSTALL, json={"plugin_id": "clock", "branch": "dev"})
+        manager.install_plugin.assert_called_once_with("clock", branch="dev")
+
+
+class TestInstallPathsAgree:
+    """The queue callback and the direct fallback duplicate the same logic."""
+
+    def _run(self, client, module, install_ok, queue):
+        module.api_v3.plugin_store_manager.install_plugin.return_value = install_ok
+        client.post(INSTALL, json={"plugin_id": "clock", "branch": "dev"})
+        return side_effects(module)
+
+    def test_success_side_effects_match(self, api_v3_client, api_v3_module):
+        direct = self._run(api_v3_client, api_v3_module, True, None)
+
+        # Reset and re-run through the queue.
+        for mock in (api_v3_module.api_v3.schema_manager,
+                     api_v3_module.api_v3.plugin_manager,
+                     api_v3_module.api_v3.plugin_state_manager,
+                     api_v3_module.api_v3.operation_history):
+            mock.reset_mock()
+        queue = MagicMock()
+        queue.enqueue_operation.side_effect = (
+            lambda t, p, operation_callback=None: operation_callback(MagicMock()) and "op")
+        api_v3_module.api_v3.operation_queue = queue
+        queued = self._run(api_v3_client, api_v3_module, True, queue)
+
+        assert direct["schema_invalidated"] == queued["schema_invalidated"]
+        assert direct["discovered"] == queued["discovered"]
+        assert direct["loaded"] == queued["loaded"]
+        assert direct["state_set"] == queued["state_set"]
+        assert (direct["history"][0].kwargs["status"]
+                == queued["history"][0].kwargs["status"])
+        assert (direct["history"][0].kwargs["details"]
+                == queued["history"][0].kwargs["details"])
+
+    def test_only_the_message_wording_differs(self, api_v3_client, api_v3_module):
+        # Characterized: the direct path says "Plugin installed
+        # successfully" while the queue callback says "Plugin clock
+        # installed successfully". Cosmetic, and the queue's text is
+        # internal to the operation record rather than the HTTP response.
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
+        direct = api_v3_client.post(INSTALL, json={"plugin_id": "clock"}).get_json()
+        assert direct["message"] == "Plugin installed successfully"
+
+
+class TestInstallFromUrl:
+    def test_uninitialized_store_manager_is_a_500(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager = None
+        assert api_v3_client.post(FROM_URL, json={"repo_url": "http://x"}).status_code == 500
+
+    def test_missing_repo_url_is_a_400(self, api_v3_client, api_v3_module):
+        response = api_v3_client.post(FROM_URL, json={})
+        assert response.status_code == 400
+        assert "repo_url required" in response.get_json()["message"]
+
+    def test_success(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
+            "success": True, "plugin_id": "clock", "name": "Clock"}
+        response = api_v3_client.post(FROM_URL, json={"repo_url": "https://github.com/o/r"})
+        assert response.status_code == 200
+        body = response.get_json()
+        assert body["plugin_id"] == "clock"
+        assert body["name"] == "Clock"
+
+    def test_all_optional_arguments_forwarded(self, api_v3_client, api_v3_module):
+        manager = api_v3_module.api_v3.plugin_store_manager
+        manager.install_from_url.return_value = {"success": True, "plugin_id": "clock"}
+        api_v3_client.post(FROM_URL, json={
+            "repo_url": "  https://github.com/o/r  ",
+            "plugin_id": "clock",
+            "plugin_path": "plugins/clock",
+            "branch": "dev",
+        })
+        manager.install_from_url.assert_called_once_with(
+            repo_url="https://github.com/o/r",
+            plugin_id="clock",
+            plugin_path="plugins/clock",
+            branch="dev",
+        )
+
+    def test_success_invalidates_schema_and_loads_plugin(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
+            "success": True, "plugin_id": "clock"}
+        api_v3_client.post(FROM_URL, json={"repo_url": "http://x"})
+        api_v3_module.api_v3.schema_manager.invalidate_cache.assert_called_once_with("clock")
+        api_v3_module.api_v3.plugin_manager.load_plugin.assert_called_once_with("clock")
+
+    def test_success_without_plugin_id_skips_discovery(self, api_v3_client, api_v3_module):
+        # install_from_url can succeed without naming the plugin; there is
+        # then nothing to invalidate or load.
+        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
+            "success": True, "plugin_id": None}
+        api_v3_client.post(FROM_URL, json={"repo_url": "http://x"})
+        api_v3_module.api_v3.schema_manager.invalidate_cache.assert_not_called()
+        api_v3_module.api_v3.plugin_manager.load_plugin.assert_not_called()
+
+    def test_branch_from_result_included(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
+            "success": True, "plugin_id": "clock", "branch": "dev"}
+        body = api_v3_client.post(FROM_URL, json={"repo_url": "http://x"}).get_json()
+        assert body["branch"] == "dev"
+        assert "(branch: dev)" in body["message"]
+
+    def test_failure_reports_the_managers_error(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
+            "success": False, "error": "repo not found"}
+        response = api_v3_client.post(FROM_URL, json={"repo_url": "http://x"})
+        assert response.status_code == 500
+        assert response.get_json()["message"] == "repo not found"
+
+    def test_failure_without_error_uses_fallback_text(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
+            "success": False}
+        response = api_v3_client.post(FROM_URL, json={"repo_url": "http://x"})
+        assert "Failed to install plugin from URL" in response.get_json()["message"]
+
+    def test_manager_exception_is_a_500(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_from_url.side_effect = (
+            RuntimeError("boom"))
+        assert api_v3_client.post(FROM_URL, json={"repo_url": "http://x"}).status_code == 500

test/test_api_v3_registry_endpoints.py

@@ -0,0 +1,179 @@
+"""
+Endpoint tests for the plugin-registry routes in api_v3:
+POST /plugins/store/refresh and POST /plugins/registry-from-url.
+
+Both reach out to the network through PluginStoreManager (mocked here) and
+had no endpoint-level coverage; registry-from-url in particular takes a
+user-supplied URL and hands it straight to the manager.
+"""
+
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
+
+
+class TestRefreshPluginStore:
+    URL = "/api/v3/plugins/store/refresh"
+
+    def test_uninitialized_manager_is_a_500(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager = None
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 500
+        assert "not initialized" in response.get_json()["message"]
+
+    def test_success_reports_plugin_count(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {
+            "plugins": [{"id": "a"}, {"id": "b"}, {"id": "c"}]}
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 200
+        assert response.get_json()["plugin_count"] == 3
+
+    def test_forces_a_refresh_rather_than_using_cache(self, api_v3_client, api_v3_module):
+        manager = api_v3_module.api_v3.plugin_store_manager
+        manager.fetch_registry.return_value = {"plugins": []}
+        api_v3_client.post(self.URL, json={})
+        manager.fetch_registry.assert_called_once_with(force_refresh=True)
+
+    def test_empty_registry_reports_zero(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {}
+        response = api_v3_client.post(self.URL, json={})
+        assert response.get_json()["plugin_count"] == 0
+
+    def test_no_body_is_accepted(self, api_v3_client, api_v3_module):
+        # Regression: `request.get_json() or {}` says a missing body is
+        # fine, but get_json() raises UnsupportedMediaType before `or {}`
+        # is reached, so a bodyless POST — the natural way to call a
+        # refresh endpoint — came back 500.
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []}
+        assert api_v3_client.post(self.URL).status_code == 200
+
+    def test_body_without_json_content_type_is_accepted(
+            self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []}
+        response = api_v3_client.post(self.URL, data="", content_type="text/plain")
+        assert response.status_code == 200
+
+    def test_malformed_json_body_falls_back_to_defaults(
+            self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []}
+        response = api_v3_client.post(
+            self.URL, data="{not json", content_type="application/json")
+        assert response.status_code == 200
+
+    @pytest.mark.parametrize("key", ["fetch_commit_info", "fetch_latest_versions"])
+    def test_either_commit_info_key_extends_the_message(
+            self, api_v3_client, api_v3_module, key):
+        # fetch_latest_versions is the older spelling; both must work.
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []}
+        response = api_v3_client.post(self.URL, json={key: True})
+        assert "commit metadata" in response.get_json()["message"]
+
+    def test_message_stays_plain_without_the_flag(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []}
+        response = api_v3_client.post(self.URL, json={})
+        assert response.get_json()["message"] == "Plugin store refreshed"
+
+    def test_network_failure_is_a_500(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.side_effect = (
+            ConnectionError("github unreachable"))
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 500
+        assert response.get_json()["message"] == "An error occurred; see logs for details"
+
+    def test_failure_body_carries_no_traceback_or_paths(
+            self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.side_effect = (
+            RuntimeError("failed at /home/user/LEDMatrix/src/secret.py line 42"))
+        body = api_v3_client.post(self.URL, json={}).get_json()
+        assert "Traceback" not in str(body)
+        # `details` is describe_exception output: one line, type-named,
+        # credential-redacted. It may quote the message, but never a stack.
+        assert body["details"].startswith("RuntimeError:")
+        assert "\n" not in body["details"]
+
+
+class TestRegistryFromUrl:
+    URL = "/api/v3/plugins/registry-from-url"
+
+    def test_uninitialized_manager_is_a_500(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager = None
+        response = api_v3_client.post(self.URL, json={"repo_url": "http://x"})
+        assert response.status_code == 500
+
+    def test_missing_repo_url_is_a_400(self, api_v3_client, api_v3_module):
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 400
+        assert "repo_url required" in response.get_json()["message"]
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.assert_not_called()
+
+    def test_success_returns_the_plugin_list(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.return_value = {
+            "plugins": [{"id": "clock"}]}
+        response = api_v3_client.post(
+            self.URL, json={"repo_url": "https://github.com/o/r"})
+        assert response.status_code == 200
+        body = response.get_json()
+        assert body["plugins"] == [{"id": "clock"}]
+        assert body["registry_url"] == "https://github.com/o/r"
+
+    def test_url_is_trimmed_before_use(self, api_v3_client, api_v3_module):
+        manager = api_v3_module.api_v3.plugin_store_manager
+        manager.fetch_registry_from_url.return_value = {"plugins": []}
+        api_v3_client.post(self.URL, json={"repo_url": "  https://github.com/o/r  "})
+        manager.fetch_registry_from_url.assert_called_once_with("https://github.com/o/r")
+
+    def test_registry_without_plugins_key_returns_empty_list(
+            self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.return_value = {
+            "other": 1}
+        response = api_v3_client.post(self.URL, json={"repo_url": "http://x"})
+        assert response.get_json()["plugins"] == []
+
+    def test_no_registry_found_is_a_400(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.return_value = None
+        response = api_v3_client.post(self.URL, json={"repo_url": "http://x/not-a-registry"})
+        assert response.status_code == 400
+        assert "Failed to fetch registry" in response.get_json()["message"]
+
+    @pytest.mark.parametrize("url", [
+        "not a url",
+        "javascript:alert(1)",
+        "file:///etc/passwd",
+        "http://localhost:8080/admin",
+    ])
+    def test_unusable_urls_fail_cleanly(self, api_v3_client, api_v3_module, url):
+        # Characterization: the handler performs no URL validation of its
+        # own — whatever the manager makes of the URL decides the outcome.
+        # What is pinned here is that a rejected URL produces a clean 400
+        # rather than a traceback or a 500.
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.return_value = None
+        response = api_v3_client.post(self.URL, json={"repo_url": url})
+        assert response.status_code == 400
+        assert "Traceback" not in str(response.get_json())
+
+    def test_fetch_exception_is_a_500_without_internals(
+            self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.side_effect = (
+            ValueError("parse failed in /srv/app/internal.py"))
+        response = api_v3_client.post(self.URL, json={"repo_url": "http://x"})
+        assert response.status_code == 500
+        body = response.get_json()
+        assert body["message"] == "An error occurred; see logs for details"
+        assert "Traceback" not in str(body)
+
+    def test_non_string_repo_url_is_rejected(self, api_v3_client, api_v3_module):
+        # Regression: .strip() on a non-string raised, and the catch-all
+        # reported the caller's own mistake as a server fault.
+        response = api_v3_client.post(self.URL, json={"repo_url": 12345})
+        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_api_v3_wifi_endpoints.py

@@ -0,0 +1,240 @@
+"""
+Endpoint tests for the /wifi/* routes in api_v3.
+
+These routes drive the host's actual networking — connecting, dropping a
+connection, switching the radio off — and had no endpoint-level tests at
+all. WiFiManager is mocked throughout; nothing here may touch real
+networking.
+
+Each handler does `from src.wifi_manager import WiFiManager` inside the
+function body, so the patch target is the class at its definition site.
+"""
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
+
+
+@pytest.fixture
+def wifi_manager():
+    """Patch WiFiManager where it is defined; yield the instance mock."""
+    with patch("src.wifi_manager.WiFiManager") as cls:
+        instance = MagicMock()
+        cls.return_value = instance
+        yield instance
+
+
+class TestConnect:
+    URL = "/api/v3/wifi/connect"
+
+    def test_success(self, api_v3_client, wifi_manager):
+        wifi_manager.connect_to_network.return_value = (True, "Connected to HomeNet")
+        response = api_v3_client.post(self.URL, json={"ssid": "HomeNet", "password": "pw"})
+        assert response.status_code == 200
+        assert response.get_json()["message"] == "Connected to HomeNet"
+        wifi_manager.connect_to_network.assert_called_once_with("HomeNet", "pw")
+
+    def test_missing_body_rejected(self, api_v3_client, wifi_manager):
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 400
+        wifi_manager.connect_to_network.assert_not_called()
+
+    def test_missing_ssid_rejected(self, api_v3_client, wifi_manager):
+        response = api_v3_client.post(self.URL, json={"password": "pw"})
+        assert response.status_code == 400
+        assert "SSID is required" in response.get_json()["message"]
+        wifi_manager.connect_to_network.assert_not_called()
+
+    @pytest.mark.parametrize("ssid", ["", "   ", "\t"])
+    def test_blank_ssid_rejected(self, api_v3_client, wifi_manager, ssid):
+        response = api_v3_client.post(self.URL, json={"ssid": ssid})
+        assert response.status_code == 400
+        wifi_manager.connect_to_network.assert_not_called()
+
+    def test_ssid_is_trimmed(self, api_v3_client, wifi_manager):
+        wifi_manager.connect_to_network.return_value = (True, "ok")
+        api_v3_client.post(self.URL, json={"ssid": "  HomeNet  "})
+        wifi_manager.connect_to_network.assert_called_once_with("HomeNet", "")
+
+    def test_missing_password_becomes_empty_string(self, api_v3_client, wifi_manager):
+        wifi_manager.connect_to_network.return_value = (True, "ok")
+        api_v3_client.post(self.URL, json={"ssid": "OpenNet"})
+        wifi_manager.connect_to_network.assert_called_once_with("OpenNet", "")
+
+    def test_null_password_becomes_empty_string(self, api_v3_client, wifi_manager):
+        wifi_manager.connect_to_network.return_value = (True, "ok")
+        api_v3_client.post(self.URL, json={"ssid": "OpenNet", "password": None})
+        wifi_manager.connect_to_network.assert_called_once_with("OpenNet", "")
+
+    def test_failure_reports_the_managers_reason(self, api_v3_client, wifi_manager):
+        wifi_manager.connect_to_network.return_value = (False, "Bad password")
+        response = api_v3_client.post(self.URL, json={"ssid": "HomeNet"})
+        assert response.status_code == 400
+        assert response.get_json()["message"] == "Bad password"
+
+    def test_failure_without_reason_uses_fallback_text(self, api_v3_client, wifi_manager):
+        wifi_manager.connect_to_network.return_value = (False, None)
+        response = api_v3_client.post(self.URL, json={"ssid": "HomeNet"})
+        assert response.status_code == 400
+        assert response.get_json()["message"] == "Failed to connect to network"
+
+    def test_manager_exception_is_a_500_without_leaking_internals(
+            self, api_v3_client, wifi_manager):
+        wifi_manager.connect_to_network.side_effect = RuntimeError(
+            "/usr/lib/secret/path blew up")
+        response = api_v3_client.post(self.URL, json={"ssid": "HomeNet"})
+        assert response.status_code == 500
+        body = response.get_json()
+        assert body["message"] == "An error occurred; see logs for details"
+        # `details` comes from describe_exception, which is deliberately
+        # safe to return (redacted, capped) — it names the type.
+        assert "RuntimeError" in body["details"]
+
+
+class TestDisconnect:
+    URL = "/api/v3/wifi/disconnect"
+
+    def test_success(self, api_v3_client, wifi_manager):
+        wifi_manager.disconnect_from_network.return_value = (True, "Disconnected")
+        response = api_v3_client.post(self.URL)
+        assert response.status_code == 200
+        assert response.get_json()["message"] == "Disconnected"
+
+    def test_failure(self, api_v3_client, wifi_manager):
+        wifi_manager.disconnect_from_network.return_value = (False, "Not connected")
+        response = api_v3_client.post(self.URL)
+        assert response.status_code == 400
+        assert response.get_json()["message"] == "Not connected"
+
+    def test_failure_without_reason_uses_fallback(self, api_v3_client, wifi_manager):
+        wifi_manager.disconnect_from_network.return_value = (False, "")
+        response = api_v3_client.post(self.URL)
+        assert response.get_json()["message"] == "Failed to disconnect from network"
+
+    def test_exception_is_a_500(self, api_v3_client, wifi_manager):
+        wifi_manager.disconnect_from_network.side_effect = OSError("nmcli missing")
+        assert api_v3_client.post(self.URL).status_code == 500
+
+
+class TestApMode:
+    ENABLE = "/api/v3/wifi/ap/enable"
+    DISABLE = "/api/v3/wifi/ap/disable"
+
+    def test_enable_success(self, api_v3_client, wifi_manager):
+        wifi_manager.enable_ap_mode.return_value = (True, "AP enabled")
+        response = api_v3_client.post(self.ENABLE, json={})
+        assert response.status_code == 200
+        wifi_manager.enable_ap_mode.assert_called_once_with(force=False)
+
+    @pytest.mark.parametrize("raw,expected", [
+        (True, True), (False, False),
+        ("true", True), ("TRUE", True), ("1", True),
+        ("false", False), ("no", False), ("yes", False),
+        (1, False),  # only real True or the listed strings count
+    ])
+    def test_force_coercion(self, api_v3_client, wifi_manager, raw, expected):
+        wifi_manager.enable_ap_mode.return_value = (True, "ok")
+        api_v3_client.post(self.ENABLE, json={"force": raw})
+        wifi_manager.enable_ap_mode.assert_called_once_with(force=expected)
+
+    def test_enable_without_body(self, api_v3_client, wifi_manager):
+        wifi_manager.enable_ap_mode.return_value = (True, "ok")
+        assert api_v3_client.post(self.ENABLE).status_code == 200
+
+    def test_enable_failure(self, api_v3_client, wifi_manager):
+        wifi_manager.enable_ap_mode.return_value = (False, "hostapd missing")
+        response = api_v3_client.post(self.ENABLE, json={})
+        assert response.status_code == 400
+        assert response.get_json()["message"] == "hostapd missing"
+
+    def test_disable_success(self, api_v3_client, wifi_manager):
+        wifi_manager.disable_ap_mode.return_value = (True, "AP disabled")
+        assert api_v3_client.post(self.DISABLE).status_code == 200
+
+    def test_disable_failure(self, api_v3_client, wifi_manager):
+        wifi_manager.disable_ap_mode.return_value = (False, "not running")
+        assert api_v3_client.post(self.DISABLE).status_code == 400
+
+    def test_enable_exception_is_a_500(self, api_v3_client, wifi_manager):
+        wifi_manager.enable_ap_mode.side_effect = RuntimeError("boom")
+        assert api_v3_client.post(self.ENABLE, json={}).status_code == 500
+
+
+class TestRadio:
+    URL = "/api/v3/wifi/radio"
+
+    def test_get_state(self, api_v3_client, wifi_manager):
+        wifi_manager.get_wifi_radio_state.return_value = {
+            "enabled": True, "ethernet_connected": False}
+        response = api_v3_client.get(self.URL)
+        assert response.status_code == 200
+        assert response.get_json()["data"]["enabled"] is True
+
+    def test_get_state_exception_is_a_500(self, api_v3_client, wifi_manager):
+        wifi_manager.get_wifi_radio_state.side_effect = OSError("rfkill missing")
+        assert api_v3_client.get(self.URL).status_code == 500
+
+    def test_enabled_is_required(self, api_v3_client, wifi_manager):
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 400
+        assert "enabled is required" in response.get_json()["message"]
+        wifi_manager.set_wifi_radio.assert_not_called()
+
+    def test_enable_success(self, api_v3_client, wifi_manager):
+        wifi_manager.set_wifi_radio.return_value = (True, "Radio on", None)
+        wifi_manager.get_wifi_radio_state.return_value = {"enabled": True}
+        response = api_v3_client.post(self.URL, json={"enabled": True})
+        assert response.status_code == 200
+        wifi_manager.set_wifi_radio.assert_called_once_with(True, force=False)
+
+    @pytest.mark.parametrize("raw,expected", [
+        (True, True), ("true", True), ("1", True), ("yes", True),
+        (False, False), ("false", False), ("off", False), (0, False),
+    ])
+    def test_enabled_coercion_is_string_aware(
+            self, api_v3_client, wifi_manager, raw, expected):
+        # bool("false") is True, so the endpoint parses strings explicitly
+        # rather than trusting truthiness — it is a public contract, not
+        # only the shipped UI which always sends real JSON booleans.
+        wifi_manager.set_wifi_radio.return_value = (True, "ok", None)
+        wifi_manager.get_wifi_radio_state.return_value = {}
+        api_v3_client.post(self.URL, json={"enabled": raw})
+        wifi_manager.set_wifi_radio.assert_called_once_with(expected, force=False)
+
+    def test_force_passed_through(self, api_v3_client, wifi_manager):
+        wifi_manager.set_wifi_radio.return_value = (True, "ok", None)
+        wifi_manager.get_wifi_radio_state.return_value = {}
+        api_v3_client.post(self.URL, json={"enabled": False, "force": "true"})
+        wifi_manager.set_wifi_radio.assert_called_once_with(False, force=True)
+
+    def test_refusal_reports_reason(self, api_v3_client, wifi_manager):
+        # Disabling the radio without Ethernet would lock the user out of
+        # this very interface, so the manager can refuse with a reason.
+        wifi_manager.set_wifi_radio.return_value = (
+            False, "Refusing: no wired fallback", "no_ethernet")
+        response = api_v3_client.post(self.URL, json={"enabled": False})
+        assert response.status_code == 400
+        body = response.get_json()
+        assert body["reason"] == "no_ethernet"
+        assert "Refusing" in body["message"]
+
+    def test_exception_is_a_500(self, api_v3_client, wifi_manager):
+        wifi_manager.set_wifi_radio.side_effect = RuntimeError("boom")
+        assert api_v3_client.post(self.URL, json={"enabled": True}).status_code == 500
+
+
+class TestNoRealNetworking:
+    def test_wifi_manager_is_never_constructed_for_real(self, api_v3_client):
+        # Guard against a future refactor moving the import to module level,
+        # where the fixture's patch of the definition site would stop
+        # applying and the tests would start driving real networking.
+        with patch("src.wifi_manager.WiFiManager") as cls:
+            cls.return_value.disconnect_from_network.return_value = (True, "ok")
+            api_v3_client.post("/api/v3/wifi/disconnect")
+            assert cls.called

test/test_git_pull_resolution.py

@@ -58,7 +58,7 @@ def repos(tmp_path):
 def test_branch_with_upstream_uses_a_plain_pull(repos):
     args, note, error = resolve_pull_command(str(repos))
     assert error is None
-    assert args == ['git', 'pull', '--rebase']
+    assert args == ['git', 'pull', '--rebase', '--autostash']
     assert note == ''
 
 
@@ -73,7 +73,7 @@ def test_branch_without_upstream_falls_back_to_origin_branch(repos):
 
     args, note, error = resolve_pull_command(str(repos))
     assert error is None
-    assert args == ['git', 'pull', '--rebase', 'origin', 'audit']
+    assert args == ['git', 'pull', '--rebase', '--autostash', 'origin', 'audit']
     assert 'audit' in note
 
 
@@ -155,7 +155,7 @@ def test_switching_attaches_tracking_so_pull_needs_no_fallback(repos):
 
     args, note, error = resolve_pull_command(str(repos))
     assert error is None
-    assert args == ['git', 'pull', '--rebase']
+    assert args == ['git', 'pull', '--rebase', '--autostash']
     assert note == ''
 
 
@@ -200,3 +200,44 @@ def test_stash_option_lets_the_switch_through_and_keeps_the_work(repos):
     assert _git('branch', '--show-current', cwd=repos).stdout.strip() == 'other'
     # The edit is not lost — it is on the stash.
     assert 'switch to other' in _git('stash', 'list', cwd=repos).stdout
+
+
+class TestInstallerDoesNotBlockTheUpdateButton:
+    """first_time_install.sh chmods scripts that git tracked as 644.
+
+    With core.fileMode true -- the default on Linux -- that leaves five
+    permanently modified tracked files on every machine that ran the
+    installer, and `git pull --rebase` refuses to start:
+
+        error: cannot pull with rebase: You have unstaged changes.
+
+    Tracking them as executable makes the installer's chmod a no-op.
+    """
+
+    CHMODDED = [
+        'first_time_install.sh',
+        'start_display.sh',
+        'stop_display.sh',
+        'scripts/install/install_service.sh',
+        'scripts/install/install_web_service.sh',
+    ]
+
+    def test_scripts_the_installer_chmods_are_tracked_executable(self):
+        import subprocess
+        from pathlib import Path
+        root = Path(__file__).resolve().parent.parent
+        out = subprocess.run(['git', 'ls-files', '-s', *self.CHMODDED],
+                             capture_output=True, text=True, cwd=str(root)).stdout
+        modes = {line.split()[3]: line.split()[0] for line in out.strip().split('\n') if line}
+        # git ls-files says nothing about a path it does not track, so a
+        # renamed or deleted script would simply be absent here and the mode
+        # check below would pass over it silently.
+        untracked = sorted(set(self.CHMODDED) - set(modes))
+        assert not untracked, (
+            f"{untracked} are chmodded by the installer but not tracked by "
+            "git, so their mode cannot be asserted at all")
+        non_exec = sorted(f for f, m in modes.items() if m != '100755')
+        assert not non_exec, (
+            f"{non_exec} are chmodded by the installer but tracked non-executable, "
+            "so every install leaves the working tree dirty and the update "
+            "button cannot pull")

test/test_logo_helper.py

@@ -0,0 +1,423 @@
+"""
+Tests for src/common/logo_helper.py — logo loading, LRU caching, resizing,
+and download-with-fallback. Previously untested: nothing in test/ referenced
+this module at all.
+
+Real PIL images under tmp_path are used rather than mocked ones, since
+load_logo() does real Path.exists() and Image.open() calls; only the HTTP
+session and the permission helpers are patched.
+
+Regression coverage for two fixed bugs:
+- _download_logo wrote response.content to disk with no size cap and no
+  check that the bytes decoded as an image, so a hostile or broken URL
+  could leave arbitrary/oversized content cached in the assets directory.
+- get_cache_stats() divided by self.cache_size unguarded, raising
+  ZeroDivisionError for a helper constructed with cache_size=0.
+"""
+
+import logging
+import tempfile
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+import requests
+from PIL import Image, UnidentifiedImageError
+
+from src.common.logo_helper import MAX_LOGO_BYTES, LogoHelper
+
+
+@pytest.fixture(autouse=True)
+def _no_real_chmod(monkeypatch):
+    # Keep the permission helpers out of the way: their own env detection
+    # is not what these tests are about.
+    monkeypatch.setattr("src.common.logo_helper.ensure_directory_permissions", MagicMock())
+    monkeypatch.setattr("src.common.logo_helper.ensure_file_permissions", MagicMock())
+
+
+@pytest.fixture
+def helper():
+    return LogoHelper(display_width=64, display_height=32,
+                      logger=logging.getLogger("test.logo_helper"))
+
+
+def write_logo(path: Path, size=(20, 20), color=(255, 0, 0), fmt="PNG") -> Path:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    Image.new("RGB", size, color).save(path, format=fmt)
+    return path
+
+
+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.__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
+
+
+def png_bytes(size=(20, 20), color=(0, 128, 0)) -> bytes:
+    import io
+    buf = io.BytesIO()
+    Image.new("RGB", size, color).save(buf, format="PNG")
+    return buf.getvalue()
+
+
+class TestLoadLogo:
+    def test_loads_and_converts_to_rgba(self, helper, tmp_path):
+        path = write_logo(tmp_path / "PHI.png")
+        logo = helper.load_logo("PHI", path)
+        assert logo is not None
+        assert logo.mode == "RGBA"
+
+    def test_missing_file_returns_none(self, helper, tmp_path, caplog):
+        with caplog.at_level(logging.WARNING):
+            assert helper.load_logo("NOPE", tmp_path / "missing.png") is None
+        assert "Logo not found" in caplog.text
+
+    def test_second_load_is_served_from_cache(self, helper, tmp_path):
+        path = write_logo(tmp_path / "PHI.png")
+        first = helper.load_logo("PHI", path)
+        path.unlink()  # cache hit must not touch the filesystem
+        assert helper.load_logo("PHI", path) is first
+
+    def test_cache_key_includes_requested_size(self, helper, tmp_path):
+        # A panel-size change must not hand back a logo sized for the old
+        # dimensions, so the two sizes get separate cache entries.
+        path = write_logo(tmp_path / "PHI.png", size=(100, 100))
+        small = helper.load_logo("PHI", path, max_width=10, max_height=10)
+        large = helper.load_logo("PHI", path, max_width=50, max_height=50)
+        assert small is not large
+        assert small.size != large.size
+        assert len(helper._logo_cache) == 2
+
+    def test_default_size_is_one_and_a_half_display(self, helper, tmp_path):
+        path = write_logo(tmp_path / "PHI.png", size=(500, 500))
+        logo = helper.load_logo("PHI", path)
+        assert logo.width <= int(64 * 1.5)
+        assert logo.height <= int(32 * 1.5)
+
+    def test_smaller_image_is_not_upscaled(self, helper, tmp_path):
+        path = write_logo(tmp_path / "PHI.png", size=(8, 8))
+        assert helper.load_logo("PHI", path, max_width=64, max_height=64).size == (8, 8)
+
+    def test_larger_image_is_downscaled_preserving_aspect(self, helper, tmp_path):
+        path = write_logo(tmp_path / "PHI.png", size=(200, 100))
+        logo = helper.load_logo("PHI", path, max_width=50, max_height=50)
+        assert logo.width <= 50 and logo.height <= 50
+        assert logo.width == 50 and logo.height == 25  # 2:1 preserved
+
+    def test_string_path_accepted(self, helper, tmp_path):
+        path = write_logo(tmp_path / "PHI.png")
+        assert helper.load_logo("PHI", str(path)) is not None
+
+    def test_corrupt_file_returns_none(self, helper, tmp_path, caplog):
+        bad = tmp_path / "bad.png"
+        bad.write_bytes(b"not an image")
+        with caplog.at_level(logging.ERROR):
+            assert helper.load_logo("BAD", bad) is None
+        assert "Error loading logo" in caplog.text
+
+
+class TestCacheManagement:
+    def test_lru_evicts_oldest(self, tmp_path):
+        helper = LogoHelper(64, 32, cache_size=2, logger=MagicMock())
+        paths = [write_logo(tmp_path / f"T{i}.png") for i in range(3)]
+        for i, path in enumerate(paths):
+            helper.load_logo(f"T{i}", path)
+        assert len(helper._logo_cache) == 2
+        assert not any(k.startswith("T0_") for k in helper._logo_cache)
+
+    def test_cache_hit_refreshes_lru_position(self, tmp_path):
+        helper = LogoHelper(64, 32, cache_size=2, logger=MagicMock())
+        a, b, c = [write_logo(tmp_path / f"{n}.png") for n in ("A", "B", "C")]
+        helper.load_logo("A", a)
+        helper.load_logo("B", b)
+        helper.load_logo("A", a)   # A is now most-recently used
+        helper.load_logo("C", c)   # evicts B, not A
+        assert any(k.startswith("A_") for k in helper._logo_cache)
+        assert not any(k.startswith("B_") for k in helper._logo_cache)
+
+    def test_clear_cache_empties_both_structures(self, helper, tmp_path):
+        helper.load_logo("PHI", write_logo(tmp_path / "PHI.png"))
+        helper.clear_cache()
+        assert helper._logo_cache == {}
+        assert helper._cache_order == []
+
+    def test_cache_stats(self, tmp_path):
+        helper = LogoHelper(64, 32, cache_size=4, logger=MagicMock())
+        helper.load_logo("PHI", write_logo(tmp_path / "PHI.png"))
+        stats = helper.get_cache_stats()
+        assert stats["cached_logos"] == 1
+        assert stats["cache_size_limit"] == 4
+        assert stats["cache_usage_percent"] == 25
+
+    def test_zero_cache_size_does_not_divide_by_zero(self):
+        # Regression: this raised ZeroDivisionError.
+        stats = LogoHelper(64, 32, cache_size=0, logger=MagicMock()).get_cache_stats()
+        assert stats["cache_usage_percent"] == 0
+        assert stats["cache_size_limit"] == 0
+
+
+class TestLoadLogoWithDownload:
+    def test_existing_file_skips_download(self, helper, tmp_path):
+        path = write_logo(tmp_path / "PHI.png")
+        helper.session.get = MagicMock()
+        assert helper.load_logo_with_download("PHI", path, "http://x/logo.png") is not None
+        helper.session.get.assert_not_called()
+
+    def test_downloads_then_loads(self, helper, tmp_path):
+        path = tmp_path / "PHI.png"
+        helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
+        logo = helper.load_logo_with_download("PHI", path, "http://x/logo.png")
+        assert logo is not None
+        assert path.exists()
+        # 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(
+            side_effect=requests.RequestException("connection reset"))
+        logo = helper.load_logo_with_download(
+            "PHI", tmp_path / "PHI.png", "http://x/logo.png",
+            max_width=20, max_height=20)
+        assert logo is not None and logo.size == (20, 20)  # placeholder
+
+    def test_http_error_falls_back_to_placeholder(self, helper, tmp_path):
+        response = fake_response(b"")
+        response.raise_for_status.side_effect = requests.HTTPError("404")
+        helper.session.get = MagicMock(return_value=response)
+        logo = helper.load_logo_with_download(
+            "PHI", tmp_path / "PHI.png", "http://x/logo.png",
+            max_width=20, max_height=20)
+        assert logo is not None and logo.size == (20, 20)
+
+    def test_no_url_and_no_file_gives_placeholder(self, helper, tmp_path):
+        logo = helper.load_logo_with_download(
+            "PHI", tmp_path / "missing.png", None, max_width=16, max_height=16)
+        assert logo is not None and logo.size == (16, 16)
+
+
+class TestDownloadLogo:
+    def test_writes_file_and_sets_permissions(self, helper, tmp_path):
+        path = tmp_path / "assets" / "PHI.png"
+        # Directory creation is ensure_directory_permissions' job, and the
+        # autouse fixture stubs it out — so make the directory here.
+        path.parent.mkdir()
+        helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
+        with patch("src.common.logo_helper.ensure_directory_permissions") as dirs, \
+             patch("src.common.logo_helper.ensure_file_permissions") as files:
+            helper._download_logo("http://x/logo.png", path)
+        assert path.exists()
+        dirs.assert_called_once()
+        files.assert_called_once()
+        assert dirs.call_args[0][0] == path.parent
+
+    def test_oversized_response_is_rejected_without_writing(self, helper, tmp_path):
+        # Regression: an unbounded response.content was written straight to
+        # disk, so a hostile URL chose how many bytes landed in assets/.
+        path = tmp_path / "huge.png"
+        helper.session.get = MagicMock(
+            return_value=fake_response(b"\x00" * (MAX_LOGO_BYTES + 1)))
+        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>"))
+        # 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"
+        helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
+
+        class Bomb:
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *a):
+                return False
+
+            def load(self):
+                raise Image.DecompressionBombError("too many pixels")
+
+        monkeypatch.setattr("src.common.logo_helper.Image.open", lambda *a, **kw: Bomb())
+        with pytest.raises(Image.DecompressionBombError):
+            helper._download_logo("http://x/bomb.png", path)
+        assert not path.exists()
+
+    def test_bad_download_surfaces_as_placeholder_not_crash(self, helper, tmp_path):
+        # The new guards raise, and load_logo_with_download's existing
+        # broad except turns that into the placeholder path.
+        helper.session.get = MagicMock(return_value=fake_response(b"garbage"))
+        logo = helper.load_logo_with_download(
+            "PHI", tmp_path / "PHI.png", "http://x/bad.png",
+            max_width=12, max_height=12)
+        assert logo is not None and logo.size == (12, 12)
+
+
+class TestLogoVariations:
+    def test_plain_abbreviation_returns_itself(self, helper):
+        assert helper.get_logo_variations("PHI") == ["PHI"]
+
+    def test_ampersand_expanded(self, helper):
+        assert "TAAND M" in helper.get_logo_variations("TA& M")
+
+    def test_and_contracted(self, helper):
+        assert "T&M" in helper.get_logo_variations("TANDM")
+
+    def test_special_case_appends_known_aliases(self, helper):
+        variations = helper.get_logo_variations("TA&M")
+        assert "TAMU" in variations and "TEXASAM" in variations
+        assert "TAANDM" in variations  # the generic & rule still applies
+
+
+class TestNormalizeAbbreviation:
+    def test_uppercases_and_strips(self, helper):
+        assert helper.normalize_abbreviation("  phi  ") == "PHI"
+
+    def test_ampersand_becomes_and(self, helper):
+        assert helper.normalize_abbreviation("TA&M") == "TAANDM"
+
+    def test_internal_spaces_removed(self, helper):
+        assert helper.normalize_abbreviation("New York") == "NEWYORK"
+
+    def test_deliberately_differs_from_logo_downloader(self, helper):
+        # Pinned, not a bug: LogoDownloader.normalize_abbreviation replaces
+        # filesystem-unsafe characters but keeps spaces, and plugins call
+        # that one. Changing either changes which logo filenames resolve on
+        # existing installs. Both docstrings say so explicitly.
+        from src.logo_downloader import LogoDownloader
+        assert helper.normalize_abbreviation("New York") == "NEWYORK"
+        assert LogoDownloader.normalize_abbreviation("New York") == "NEW YORK"
+
+
+class TestPlaceholderLogo:
+    def test_uses_requested_dimensions(self, helper):
+        assert helper._create_placeholder_logo("PHI", 30, 20).size == (30, 20)
+
+    def test_defaults_to_one_and_a_half_display(self, helper):
+        assert helper._create_placeholder_logo("PHI").size == (96, 48)
+
+    def test_is_rgba(self, helper):
+        assert helper._create_placeholder_logo("PHI", 10, 10).mode == "RGBA"
+
+    def test_invalid_dimensions_return_none(self, helper, caplog):
+        with caplog.at_level(logging.ERROR):
+            assert helper._create_placeholder_logo("PHI", -5, -5) is None
+        assert "Error creating placeholder" in caplog.text
+
+
+class TestSessionConfiguration:
+    def test_user_agent_and_accept_headers(self, helper):
+        assert helper.session.headers["User-Agent"] == "LEDMatrix-Common/1.0"
+        assert helper.session.headers["Accept"] == "image/*"

test/test_sudo_allowlist_covers_calls.py

@@ -0,0 +1,147 @@
+"""The captive portal's fixed-argument sudo calls must be granted.
+
+The installers write two allow-lists, /etc/sudoers.d/ledmatrix_web and
+ledmatrix_wifi. A sudo call absent from both needs a password, which a service
+cannot supply, so it fails.
+
+Four such calls were ungranted, all of them captive-portal teardown/setup:
+
+    sysctl -w net.ipv4.ip_forward=0|1     wifi_manager.py:788, 883
+    nft add|delete table ip ledmatrix     wifi_manager.py:835, 895
+    rfkill unblock wifi                   wifi_manager.py:1811
+    mkdir -p .../dnsmasq-shared.d         wifi_manager.py:922
+
+It goes unnoticed because a stock Raspberry Pi image ships
+/etc/sudoers.d/010_pi-nopasswd granting the default user
+`ALL=(ALL) NOPASSWD: ALL`, which satisfies every gap in both files. It only
+bites once that blanket rule is removed or the service runs as another user.
+
+Scope, deliberately narrow: this pins the four commands above, each of which
+can be written out literally. The portal makes further sudo calls whose
+arguments are built at runtime -- iptables and nft rules carrying an interface
+name and a port, `ip addr`, `ip link` -- and those cannot be granted safely
+here. A rule covering them needs a trailing wildcard, and
+`iptables --modprobe=/path/to/anything` runs that path as root, so
+`NOPASSWD: iptables *` is a root shell for the web user by another name.
+Closing that half needs a privileged helper that builds the rules itself and
+takes only an interface and a port, granted the way safe_plugin_rm.sh already
+is. That is a design decision, not a one-line grant, and belongs in its own
+change.
+"""
+import re
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parent.parent
+INSTALLERS = (
+    ROOT / "first_time_install.sh",
+    ROOT / "scripts" / "install" / "configure_wifi_permissions.sh",
+)
+
+#: Commands this change grants, each fully literal in the source.
+REQUIRED = (
+    ("sysctl", "-w", "net.ipv4.ip_forward=0"),
+    ("sysctl", "-w", "net.ipv4.ip_forward=1"),
+    ("nft", "add", "table", "ip", "ledmatrix"),
+    ("nft", "delete", "table", "ip", "ledmatrix"),
+    ("rfkill", "unblock", "wifi"),
+    ("mkdir", "-p", "/etc/NetworkManager/dnsmasq-shared.d"),
+)
+
+#: Tools with an option that executes a program of the caller's choosing.
+#: A trailing wildcard on any of these is a privilege escalation.
+EXEC_CAPABLE = ("iptables", "ip6tables", "nft", "tcpdump", "find", "awk",
+                "sed", "perl", "python", "python3", "env")
+
+
+def _grant_lines():
+    lines = []
+    for installer in INSTALLERS:
+        if not installer.is_file():
+            continue
+        for line in installer.read_text(encoding="utf-8", errors="replace").splitlines():
+            if "NOPASSWD:" in line:
+                lines.append(line.split("NOPASSWD:", 1)[1])
+    return lines
+
+
+def _normalise(rule):
+    """One rule with binary-path variables reduced to bare tool names.
+
+    Rules are written as `$SYSCTL_PATH -w ...` or `${NFT_PATH} ...`, so
+    matching the literal "sysctl" finds nothing and every rule looks absent --
+    which is exactly how an earlier version of this test reported six gaps
+    that did not exist. Both spellings are handled: shell expands them
+    identically, and a check that understood only one silently skipped the
+    other.
+    """
+    rule = re.sub(r"\$\{?([A-Z][A-Z0-9_]*)_PATH\}?",
+                  lambda m: m.group(1).lower(), rule)
+    return re.sub(r"/usr/(?:s?bin)/", "", rule)
+
+
+def _granted_commands():
+    """The command each NOPASSWD rule actually grants, normalised.
+
+    _grant_lines() already returns everything after "NOPASSWD:", so what
+    arrives here is the command, possibly preceded by the NOEXEC tag and
+    possibly still carrying the closing quote of an `echo "..."` that wrote
+    it. Both are stripped so the result is comparable to a plain command.
+    """
+    commands = []
+    for rule in _grant_lines():
+        command = _normalise(rule).strip()
+        command = re.sub(r"^NOEXEC:\s*", "", command)
+        command = command.rstrip('"').rstrip("'").strip()
+        if command:
+            commands.append(" ".join(command.split()))
+    return commands
+
+
+def test_the_installers_are_present():
+    missing = [str(p.relative_to(ROOT)) for p in INSTALLERS if not p.is_file()]
+    assert not missing, f"installer(s) missing: {missing}"
+
+
+@pytest.mark.parametrize("command", REQUIRED, ids=lambda c: " ".join(c))
+def test_the_command_is_granted(command):
+    """Whole command, not just the binary.
+
+    Checking only the binary made this far weaker than it looked: with
+    `sysctl` present anywhere, deleting the ip_forward=0 grant still passed,
+    and the portal would then be unable to restore forwarding on teardown.
+    """
+    wanted = " ".join(command)
+    granted = _granted_commands()
+    # Exact match, not a prefix. A substring search was satisfied by
+    # `sysctl -w net.ipv4.ip_forward=0 *`, and that trailing wildcard lets the
+    # caller append whatever they like to a command running as root -- a far
+    # wider grant than the one this test is meant to be confirming.
+    assert wanted in granted, (
+        f"no installer grants exactly `{wanted}`; closest matches: "
+        + str([g for g in granted if g.startswith(command[0])])[:200])
+
+
+def test_no_wildcard_on_a_tool_that_can_exec():
+    """`NOPASSWD: iptables *` hands the web user root.
+
+    iptables --modprobe=/path runs that path as root. This caught a grant added
+    in this very change, which is why it is here.
+    """
+    offenders = []
+    for rule in _grant_lines():
+        rule = rule.strip()
+        if not rule.endswith("*"):
+            continue
+        # Normalised the same way as everything else: `${NFT_PATH} *` left a
+        # brace before the tool name, and the word-boundary check below does
+        # not treat "{" as a boundary, so that spelling slipped through.
+        haystack = _normalise(rule).lower()
+        for tool in EXEC_CAPABLE:
+            if re.search(rf"(^|/|\s|\$){tool}(\s|$)", haystack):
+                offenders.append(rule)
+                break
+    assert not offenders, (
+        "wildcard grant on a tool that can execute another program:\n  "
+        + "\n  ".join(offenders))

test/test_sudoers_noexec_on_pagers.py

@@ -0,0 +1,99 @@
+"""Wildcard grants to commands that start a pager must carry NOEXEC.
+
+`journalctl` runs a pager when its output is a terminal, and from `less` a
+`!sh` is a shell with the privileges journalctl was given. That is the standard
+journalctl privilege escalation, and the installer's rules end in a wildcard:
+
+    <user> ALL=(ALL) NOPASSWD: /usr/bin/journalctl -u ledmatrix *
+
+The web interface always passes --no-pager -- both call sites do, in app.py and
+api_v3.py -- so nothing the project runs needs the pager. But a sudoers rule
+cannot require a flag that sits in the middle of the command line, and reasoning
+about what a trailing `*` does or does not admit is exactly the kind of
+subtlety that produces a hole.
+
+sudo's NOEXEC tag stops the command executing another program at all, which
+closes it without depending on that reasoning. It works by LD_PRELOAD, so it
+applies to dynamically linked binaries; journalctl is one.
+
+On a stock Raspberry Pi image none of this is reachable, because
+/etc/sudoers.d/010_pi-nopasswd already grants the default user
+`ALL=(ALL) NOPASSWD: ALL`. It matters on a hardened install, or where the
+service runs as a user without that blanket rule.
+"""
+import re
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parent.parent
+INSTALLERS = (
+    ROOT / "first_time_install.sh",
+    ROOT / "scripts" / "install" / "configure_wifi_permissions.sh",
+    # Writes the same journalctl grants as first_time_install.sh. It was
+    # missing here, and because of that this suite passed while three
+    # ungranted wildcard rules sat in it.
+    ROOT / "scripts" / "install" / "configure_web_sudo.sh",
+)
+
+#: Commands that will start another program of their own accord -- a pager, an
+#: editor, a shell -- and so must not be granted the ability to do so.
+SPAWNS_A_PROGRAM = ("journalctl", "systemctl", "less", "more", "man", "git")
+
+
+def _grant_lines():
+    lines = []
+    for installer in INSTALLERS:
+        if not installer.is_file():
+            continue
+        for line in installer.read_text(encoding="utf-8", errors="replace").splitlines():
+            stripped = line.strip()
+            if "NOPASSWD" not in stripped or stripped.startswith("#"):
+                continue
+            # Installers emit rules two ways: written literally into a heredoc,
+            # or echoed into a file. An echoed rule ends in a quote, so the
+            # trailing-wildcard check below would skip it and the rule would
+            # never be examined at all.
+            echoed = re.fullmatch(r"""echo\s+(['"])(.*)\1""", stripped)
+            lines.append(echoed.group(2) if echoed else stripped)
+    return lines
+
+
+def test_the_installers_are_present():
+    missing = [str(p.relative_to(ROOT)) for p in INSTALLERS if not p.is_file()]
+    assert not missing, f"installer(s) missing: {missing}"
+
+
+def test_wildcard_pager_grants_carry_noexec():
+    offenders = []
+    for rule in _grant_lines():
+        command = rule.split("NOPASSWD", 1)[1]
+        if not command.rstrip().endswith("*"):
+            continue
+        tool = command.replace("_PATH", "").replace("$", "").lower()
+        for name in SPAWNS_A_PROGRAM:
+            if re.search(rf"(^|/|\s){name}(\s|$)", tool):
+                if "NOEXEC" not in rule:
+                    offenders.append(rule)
+                break
+    assert not offenders, (
+        "wildcard grant to a command that can start a pager or shell, without "
+        "NOEXEC:\n  " + "\n  ".join(offenders))
+
+
+def test_journalctl_is_granted_at_all():
+    """Guard against 'fixing' the above by deleting the rules."""
+    text = "\n".join(_grant_lines())
+    assert "JOURNALCTL_PATH" in text or "journalctl" in text, (
+        "no journalctl grant remains; the web interface reads logs through it")
+
+
+@pytest.mark.parametrize("selector", ["-u ledmatrix.service", "-u ledmatrix",
+                                      "-t ledmatrix"])
+def test_each_journalctl_rule_is_tagged(selector):
+    """Every selector, so removing one cannot pass by the others' presence."""
+    matching = [r for r in _grant_lines()
+                if "JOURNALCTL_PATH" in r and f"{selector} " in r]
+    assert matching, f"no journalctl rule for {selector}"
+    untagged = [r for r in matching if "NOEXEC" not in r]
+    assert not untagged, f"untagged journalctl rule(s): {untagged}"

test/test_systemd_malloc_arenas.py

@@ -0,0 +1,95 @@
+"""The display unit must cap glibc's malloc arenas.
+
+glibc hands each allocating thread its own malloc arena, up to 8 x CPU count,
+and an arena that has grown is never returned to the OS. This process runs
+threads for the render loop, the update workers and the background fetchers, so
+on a 3-core Pi the ceiling is 24 arenas.
+
+Measured on a live rig, 2.5 hours in:
+
+    RSS                 1030 MB
+    Private_Dirty        988 MB
+    anonymous mappings > 10 MB     23     (ceiling is 8 x 3 = 24)
+    largest few          104, 79, 66, 63, 63 MB, on 64 MB-aligned addresses
+
+against live data that accounts for perhaps 15 MB -- the widest scroll strip
+observed was 35,746 x 64, about 7 MB as RGB and the same again for its numpy
+mirror. Repeated sampling showed RSS flat between 990 and 1030 MB rather than
+climbing, so this is arena bloat rather than a leak: memory Python has freed
+but glibc is holding per-arena.
+
+The device had 59 MB free at the time.
+
+Capping the arena count trades a little allocator concurrency for that resident
+memory. The render loop is latency-sensitive, so if p99 frame time regresses the
+right response is to raise this rather than remove it.
+"""
+import re
+from pathlib import Path
+
+import pytest
+
+UNIT = (Path(__file__).resolve().parent.parent / "systemd" / "ledmatrix.service")
+
+#: The value the unit is expected to carry. 2 is the usual choice for a
+#: threaded Python process; 1-4 all keep some of the saving, but only one of
+#: them is what this project ships.
+EXPECTED_ARENA_MAX = 2
+
+
+def _environment(unit_text):
+    return dict(
+        line.split("=", 2)[1:3] if line.count("=") >= 2 else (line.split("=", 1)[1], "")
+        for line in unit_text.splitlines()
+        if line.startswith("Environment=")
+    )
+
+
+def test_the_unit_exists():
+    assert UNIT.is_file(), f"{UNIT} is missing"
+
+
+def test_malloc_arena_max_is_capped():
+    env = _environment(UNIT.read_text(encoding="utf-8"))
+    assert "MALLOC_ARENA_MAX" in env, (
+        "the display unit does not cap glibc arenas; on a 3-core Pi the default "
+        "ceiling is 24 and a measured rig held 23 of them, 920 MB"
+    )
+    value = int(env["MALLOC_ARENA_MAX"])
+    # Pinned, not a range. A range let a change to 4 -- which hands most of the
+    # saving back -- pass unnoticed, which was the point of the finding that
+    # prompted this. Raising it is a legitimate response to a frame-time
+    # regression, but it should be a visible edit here rather than a silent
+    # drift, so the number lives in one place and changing it shows up in
+    # review.
+    assert value == EXPECTED_ARENA_MAX, (
+        f"MALLOC_ARENA_MAX={value}, expected {EXPECTED_ARENA_MAX}. If this was "
+        "raised deliberately because frame times regressed, update "
+        "EXPECTED_ARENA_MAX here and say so in the commit."
+    )
+
+
+def test_the_reason_is_recorded_next_to_it():
+    """A bare tuning knob invites removal by whoever meets it next."""
+    text = UNIT.read_text(encoding="utf-8")
+    index = text.index("Environment=MALLOC_ARENA_MAX")
+    preamble = text[:index].splitlines()[-12:]
+    comment = "\n".join(line for line in preamble if line.startswith("#"))
+    assert "arena" in comment.lower(), "no explanation precedes the setting"
+    assert re.search(r"\d", comment), (
+        "the explanation cites no measurement, so a reader cannot tell whether "
+        "it still applies to their hardware"
+    )
+
+
+@pytest.mark.parametrize("unit", ["ledmatrix.service"])
+def test_the_unit_still_parses_as_ini(unit):
+    """systemd will refuse a malformed unit, and the panel stays dark."""
+    import configparser
+
+    path = UNIT.parent / unit
+    parser = configparser.ConfigParser(strict=False)
+    # systemd allows repeated keys; ConfigParser needs them merged, not rejected.
+    parser.read_string(path.read_text(encoding="utf-8"))
+    assert parser.has_section("Service")
+    assert parser.has_option("Service", "ExecStart")

test/test_systemd_unit_drift.py

@@ -0,0 +1,152 @@
+"""An installed unit that no longer matches the repo's must be reported.
+
+Nothing re-applies systemd units after the first install. `git pull` -- what
+the web UI's update button runs -- brings a new template into the checkout, but
+no code in web_interface/ or src/ copies it to /etc/systemd/system or runs
+`systemctl daemon-reload`. The unit that actually runs is whatever
+first_time_install.sh wrote on day one.
+
+So every hardening added to a unit is inert on existing installs. Measured on a
+live rig: the installed unit was dated 2026-08-06 and the repo's 2026-08-19,
+and they differed -- with the result that a MemoryMax=85% present in the repo's
+template was not being enforced at all. `systemctl show` reported
+MemoryMax=infinity.
+
+This is a warning, not an error, and deliberately not a silent rewrite:
+editing files under /etc and restarting services is the installer's job, not
+something a display process should do to a machine while it boots.
+"""
+import logging
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from src.startup_validator import StartupValidator
+
+
+@pytest.fixture
+def validator():
+    v = StartupValidator(config_manager=MagicMock())
+    v.logger = logging.getLogger("test")
+    v.warnings = []
+    v.errors = []
+    return v
+
+
+def test_a_matching_unit_produces_no_warning(validator, tmp_path):
+    """The installed unit, substituted exactly as the installer would."""
+    project_root = Path("src/startup_validator.py").resolve().parent.parent
+    template_rel = "systemd/ledmatrix.service"
+    template = project_root / template_rel
+    if not template.is_file():
+        pytest.skip("repo unit template not present")
+
+    installed = tmp_path / "ledmatrix.service"
+    installed.write_text(
+        template.read_text(encoding="utf-8")
+        .replace("__PROJECT_ROOT_DIR__", str(project_root))
+        .replace("__USER__", "root"),
+        encoding="utf-8")
+
+    validator._UNITS = ((template_rel, str(installed)),)
+    validator._validate_systemd_units()
+    assert not validator.warnings, f"a matching unit warned: {validator.warnings}"
+    assert not validator.errors
+
+
+def test_comments_and_blank_lines_are_not_drift():
+    """Otherwise every comment the repo adds would look like a changed unit."""
+    a = "[Service]\n# explains a setting\nExecStart=/x\nRestart=always\n"
+    b = "[Service]\nExecStart=/x\n\nRestart=always\n"
+    assert StartupValidator._unit_body(a) == StartupValidator._unit_body(b)
+
+
+def test_a_changed_directive_is_drift():
+    a = "[Service]\nExecStart=/x\nMemoryMax=85%\n"
+    b = "[Service]\nExecStart=/x\n"
+    assert StartupValidator._unit_body(a) != StartupValidator._unit_body(b)
+
+
+def test_reordered_directives_are_drift():
+    """Order is not noise in a systemd unit.
+
+    Repeated directives -- ExecStartPre=, ExecStartPost= -- run in the order
+    they appear, and a directive that moves between [Unit], [Service] and
+    [Install] means something different, or nothing, where it lands. This
+    check used to sort the lines before comparing, which reported no drift for
+    a unit that had genuinely changed.
+    """
+    a = "[Service]\nExecStartPre=/first\nExecStartPre=/second\n"
+    b = "[Service]\nExecStartPre=/second\nExecStartPre=/first\n"
+    assert StartupValidator._unit_body(a) != StartupValidator._unit_body(b), (
+        "swapping two ExecStartPre= lines changes what runs first, and was "
+        "being normalised away")
+
+
+def test_a_directive_moved_between_sections_is_drift():
+    a = "[Unit]\nDescription=x\n[Service]\nExecStart=/x\n"
+    b = "[Unit]\nDescription=x\nExecStart=/x\n[Service]\n"
+    assert StartupValidator._unit_body(a) != StartupValidator._unit_body(b), (
+        "ExecStart= in [Unit] is not the same unit, and sorting hid it")
+
+
+def test_cosmetic_differences_do_not_warn(validator, tmp_path):
+    """Through the real comparison, not the helper.
+
+    The repo's template carries explanatory comments the installed copy may not
+    have, and the installer does not preserve ordering or blank lines. If those
+    counted as drift, every boot would warn and the warning would be ignored.
+    Asserting this on _unit_body alone would not catch a comparison that stopped
+    calling it -- which is exactly what a careless edit does.
+    """
+    project_root = Path("src/startup_validator.py").resolve().parent.parent
+    template_rel = "systemd/ledmatrix.service"
+    template = project_root / template_rel
+    if not template.is_file():
+        pytest.skip("repo unit template not present")
+
+    substituted = (template.read_text(encoding="utf-8")
+                   .replace("__PROJECT_ROOT_DIR__", str(project_root))
+                   .replace("__USER__", "root"))
+    # Cosmetic means comments, blank lines and stray indentation -- the things
+    # the installer really does drop. Not reordering: that changes the unit,
+    # and is asserted as drift above.
+    directives = [line.strip() for line in substituted.splitlines()
+                  if line.strip() and not line.strip().startswith("#")]
+    installed = tmp_path / "ledmatrix.service"
+    installed.write_text(
+        "\n\n".join("   " + d for d in directives) + "\n", encoding="utf-8")
+
+    validator._UNITS = ((template_rel, str(installed)),)
+    validator._validate_systemd_units()
+    assert not validator.warnings, (
+        f"cosmetic-only difference reported as drift: {validator.warnings}")
+
+
+def test_drift_is_reported_as_a_warning(validator, tmp_path):
+    """The whole point: a real difference must surface, and only as a warning."""
+    installed = tmp_path / "ledmatrix.service"
+    installed.write_text("[Service]\nExecStart=/usr/bin/python3 /x/run.py\n")
+
+    project_root = Path("src/startup_validator.py").resolve().parent.parent
+    template_rel = "systemd/ledmatrix.service"
+    template = project_root / template_rel
+    if not template.is_file():
+        pytest.skip("repo unit template not present")
+
+    validator._UNITS = ((template_rel, str(installed)),)
+    validator._validate_systemd_units()
+
+    assert validator.warnings, "a differing unit produced no warning"
+    assert "install_service.sh" in validator.warnings[0], (
+        "the warning does not tell the user how to fix it")
+    assert not validator.errors, "drift must not be fatal at startup"
+
+
+def test_a_missing_installed_unit_is_silent(validator, tmp_path):
+    """Development checkouts have no /etc/systemd unit; that is not drift."""
+    validator._UNITS = (("systemd/ledmatrix.service", str(tmp_path / "absent.service")),)
+    validator._validate_systemd_units()
+    assert not validator.warnings
+    assert not validator.errors

test/test_update_check_reports_failure.py

@@ -0,0 +1,85 @@
+"""A check that could not run must not be reported as "up to date".
+
+check-update returned update_available=False whenever git failed. The banner
+is the only route to the update button, so a checkout git refuses to touch
+looked exactly like a current one -- permanently, and with nothing for the
+user to act on. The usual cause is an install performed as root, after which
+every git command fails with "detected dubious ownership".
+"""
+import subprocess
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+from flask import Flask
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from web_interface.blueprints import api_v3 as mod  # noqa: E402
+from web_interface.blueprints.api_v3 import api_v3  # noqa: E402
+
+DUBIOUS = ("fatal: detected dubious ownership in repository at "
+           "'/home/pi/LEDMatrix'\nTo add an exception for this directory, call:\n"
+           "\tgit config --global --add safe.directory /home/pi/LEDMatrix\n")
+
+
+@pytest.fixture
+def client():
+    app = Flask(__name__)
+    app.config['TESTING'] = True
+    app.register_blueprint(api_v3, url_prefix='/api/v3')
+    mod._update_check_cache['result'] = None
+    mod._update_check_cache['ts'] = 0
+    return app.test_client()
+
+
+def _fetch_fails(stderr: bytes):
+    def fake_run(args, **kwargs):
+        if args[:2] == ['git', 'fetch']:
+            return subprocess.CompletedProcess(args, 1, stdout=b'', stderr=stderr)
+        return subprocess.CompletedProcess(args, 0, stdout='', stderr='')
+    return fake_run
+
+
+class TestFailedCheckIsNotSilence:
+    def test_dubious_ownership_is_reported_not_swallowed(self, client):
+        with patch.object(mod.subprocess, 'run', _fetch_fails(DUBIOUS.encode())):
+            data = client.get('/api/v3/system/check-update').get_json()
+        assert data['check_failed'] is True, (
+            "a git failure was reported as a successful 'no update' check")
+        assert data['update_available'] is False
+
+    def test_the_message_tells_the_user_what_to_do(self, client):
+        with patch.object(mod.subprocess, 'run', _fetch_fails(DUBIOUS.encode())):
+            data = client.get('/api/v3/system/check-update').get_json()
+        assert 'chown' in data['error'], (
+            "dubious ownership is unactionable without the fix command")
+        assert 'root' in data['error']
+
+    def test_an_ordinary_git_failure_still_surfaces(self, client):
+        with patch.object(mod.subprocess, 'run',
+                          _fetch_fails(b'fatal: some other git problem\n')):
+            data = client.get('/api/v3/system/check-update').get_json()
+        assert data['check_failed'] is True
+        assert 'some other git problem' in data['error']
+
+    def test_offline_reads_as_offline(self, client):
+        with patch.object(mod.subprocess, 'run',
+                          _fetch_fails(b'fatal: could not resolve host: github.com\n')):
+            data = client.get('/api/v3/system/check-update').get_json()
+        assert 'Could not reach GitHub' in data['error']
+
+
+class TestSuccessPathUnchanged:
+    def test_up_to_date_carries_no_failure_flag(self, client):
+        def fake_run(args, **kwargs):
+            if args[:2] == ['git', 'fetch']:
+                return subprocess.CompletedProcess(args, 0, stdout=b'', stderr=b'')
+            if args[:2] == ['git', 'rev-parse']:
+                return subprocess.CompletedProcess(args, 0, stdout='abc123\n', stderr='')
+            return subprocess.CompletedProcess(args, 0, stdout='0\n', stderr='')
+        with patch.object(mod.subprocess, 'run', fake_run):
+            data = client.get('/api/v3/system/check-update').get_json()
+        assert data['update_available'] is False
+        assert not data.get('check_failed'), "a healthy check must not look like a failure"

test/test_update_prompts_restart.py

@@ -0,0 +1,84 @@
+"""A pull that changed nothing on the running system is not an applied update.
+
+git_pull replaces files on disk and restarts nothing -- there is no systemctl
+call anywhere in the handler. The display and web services keep running the
+code they loaded at boot, so the user is told "Code updated successfully" and
+sees no change until they happen to reboot. The response now says whether a
+restart is owed, and the UI raises the existing restart-pending banner.
+"""
+import subprocess
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+from flask import Flask
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from web_interface.blueprints import api_v3 as mod  # noqa: E402
+from web_interface.blueprints.api_v3 import api_v3  # noqa: E402
+
+
+@pytest.fixture
+def client():
+    app = Flask(__name__)
+    app.config['TESTING'] = True
+    app.register_blueprint(api_v3, url_prefix='/api/v3')
+    # The handler consults these after a successful pull; None is the
+    # "not wired up" case it already guards for.
+    api_v3.plugin_store_manager = None
+    api_v3.config_manager = None
+    return app.test_client()
+
+
+def _git(heads, pull_rc=0, pull_out='Updating a1b2c3..d4e5f6\n'):
+    """Fake git. `heads` are the successive answers to rev-parse HEAD."""
+    seq = list(heads)
+
+    def run(args, **kwargs):
+        def ok(stdout='', rc=0, b=False):
+            return subprocess.CompletedProcess(
+                args, rc, stdout=(stdout.encode() if b else stdout),
+                stderr=(b'' if b else ''))
+        if args[:2] == ['git', 'rev-parse'] and args[-1] == 'HEAD':
+            return ok(seq.pop(0) + '\n' if seq else 'deadbeef\n')
+        if 'symbolic-full-name' in args or '@{u}' in args:
+            return ok('origin/main\n')
+        if args[:2] == ['git', 'status']:
+            return ok('')
+        if args[:2] == ['git', 'diff']:
+            return ok('')
+        if args[:2] == ['git', 'pull']:
+            return ok(pull_out, pull_rc)
+        return ok('')
+    return run
+
+
+def _pull(client):
+    return client.post('/api/v3/system/action',
+                       json={'action': 'git_pull'}).get_json()
+
+
+class TestRestartIsRequestedWhenCodeChanged:
+    def test_a_pull_that_moved_head_asks_for_a_restart(self, client):
+        with patch.object(mod.subprocess, 'run', _git(['aaa111', 'bbb222'])):
+            data = _pull(client)
+        assert data['status'] == 'success'
+        assert data['restart_required'] is True, (
+            "new code on disk, services still running the old code, and "
+            "nothing told the user to restart")
+
+    def test_already_up_to_date_does_not(self, client):
+        with patch.object(mod.subprocess, 'run',
+                          _git(['aaa111', 'aaa111'], pull_out='Already up to date.\n')):
+            data = _pull(client)
+        assert data['status'] == 'success'
+        assert data['restart_required'] is False, (
+            "prompting after a no-op update trains users to ignore the prompt")
+
+    def test_a_failed_pull_does_not(self, client):
+        with patch.object(mod.subprocess, 'run', _git(['aaa111'], pull_rc=1)):
+            data = _pull(client)
+        assert data['status'] == 'error'
+        assert data['restart_required'] is False

test/web_interface/test_api_v3_backup_paths.py

@@ -0,0 +1,220 @@
+"""
+Path-containment tests for the backup file routes:
+GET /backup/download/<filename>, DELETE /backup/<filename>, and the
+listing/validation routes alongside them.
+
+Both filename routes take user input straight from the URL and turn it
+into a filesystem path, one to read and one to unlink. `_safe_backup_path`
+is what stops that from reaching outside the export directory, and it had
+no tests.
+
+This is verification of existing containment, not a fix: no bypass was
+found. The tests exist so that a later "just let dots through" change has
+to argue with something.
+"""
+
+import io
+import json
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+from flask import Flask
+
+project_root = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(project_root))
+
+from web_interface.blueprints import api_v3 as api_v3_module  # noqa: E402
+from web_interface.blueprints.api_v3 import api_v3  # noqa: E402
+
+_MANAGER_ATTRS = (
+    'config_manager', 'plugin_manager', 'plugin_store_manager',
+    'plugin_state_manager', 'saved_repositories_manager', 'schema_manager',
+    'operation_queue', 'operation_history', 'cache_manager',
+)
+_SENTINEL = object()
+
+# Anything that tries to name a file outside the export directory, or that
+# is not a plain <name>.zip.
+TRAVERSAL_ATTEMPTS = [
+    "../../etc/passwd",
+    "../config.json",
+    "..%2f..%2fetc%2fpasswd",
+    "....//....//etc/passwd",
+    "/etc/passwd",
+    "..\\..\\config.json",
+    "backup.zip/../../../etc/passwd",
+    ".hidden.zip",
+    "backup.txt",
+    "backup.zip.exe",
+    "",
+    ".",
+    "..",
+]
+
+
+@pytest.fixture
+def env(tmp_path, monkeypatch):
+    export_dir = tmp_path / "backups"
+    export_dir.mkdir()
+    monkeypatch.setattr(api_v3_module, "_BACKUP_EXPORT_DIR", export_dir)
+
+    # A file outside the export dir that a traversal would be reaching for.
+    secret = tmp_path / "config.json"
+    secret.write_text(json.dumps({"secret": "do not touch"}))
+
+    originals = {name: getattr(api_v3, name, _SENTINEL) for name in _MANAGER_ATTRS}
+    for name in _MANAGER_ATTRS:
+        setattr(api_v3, name, MagicMock())
+
+    app = Flask(__name__)
+    app.config["TESTING"] = True
+    app.register_blueprint(api_v3, url_prefix="/api/v3")
+
+    class Env:
+        pass
+
+    e = Env()
+    e.client = app.test_client()
+    e.export_dir = export_dir
+    e.secret = secret
+    yield e
+
+    for name, original in originals.items():
+        if original is _SENTINEL:
+            if hasattr(api_v3, name):
+                delattr(api_v3, name)
+        else:
+            setattr(api_v3, name, original)
+
+
+def make_backup(export_dir, name="backup-2026-01-01.zip"):
+    path = export_dir / name
+    path.write_bytes(b"PK\x03\x04fake zip")
+    return path
+
+
+class TestSafeBackupPath:
+    """The containment helper itself."""
+
+    @pytest.mark.parametrize("filename", TRAVERSAL_ATTEMPTS)
+    def test_rejects_unsafe_names(self, env, filename):
+        assert api_v3_module._safe_backup_path(filename) is None
+
+    def test_rejects_none(self, env):
+        assert api_v3_module._safe_backup_path(None) is None
+
+    @pytest.mark.parametrize("filename", [
+        "backup.zip",
+        "backup-2026-01-01.zip",
+        "backup_2026.01.01-v2.zip",
+        "a.zip",
+    ])
+    def test_accepts_plain_zip_names(self, env, filename):
+        resolved = api_v3_module._safe_backup_path(filename)
+        assert resolved is not None
+        assert resolved.parent == env.export_dir.resolve()
+
+    def test_result_is_always_inside_the_export_dir(self, env):
+        resolved = api_v3_module._safe_backup_path("backup.zip")
+        resolved.relative_to(env.export_dir.resolve())  # raises if outside
+
+    def test_overlong_name_rejected(self, env):
+        assert api_v3_module._safe_backup_path("a" * 250 + ".zip") is None
+
+
+class TestDownload:
+    def test_downloads_an_existing_backup(self, env):
+        make_backup(env.export_dir)
+        response = env.client.get("/api/v3/backup/download/backup-2026-01-01.zip")
+        assert response.status_code == 200
+        assert response.data == b"PK\x03\x04fake zip"
+
+    def test_missing_file_is_a_404(self, env):
+        response = env.client.get("/api/v3/backup/download/never-made.zip")
+        assert response.status_code == 404
+
+    @pytest.mark.parametrize("filename", TRAVERSAL_ATTEMPTS)
+    def test_traversal_attempts_are_refused(self, env, filename):
+        response = env.client.get(f"/api/v3/backup/download/{filename}")
+        # However the request is turned away — 404 from the containment
+        # check, or 308/405 from routing never matching at all — what
+        # matters is that no file outside the export directory is served.
+        assert response.status_code != 200
+        assert b"do not touch" not in response.data
+
+
+class TestDelete:
+    def test_deletes_an_existing_backup(self, env):
+        path = make_backup(env.export_dir)
+        response = env.client.delete("/api/v3/backup/backup-2026-01-01.zip")
+        assert response.status_code == 200
+        assert not path.exists()
+
+    def test_missing_file_is_a_404(self, env):
+        response = env.client.delete("/api/v3/backup/never-made.zip")
+        assert response.status_code == 404
+
+    @pytest.mark.parametrize("filename", TRAVERSAL_ATTEMPTS)
+    def test_traversal_attempts_delete_nothing(self, env, filename):
+        response = env.client.delete(f"/api/v3/backup/{filename}")
+        assert response.status_code != 200
+        assert env.secret.exists()  # the file a traversal was aiming at
+
+    def test_only_the_named_backup_is_removed(self, env):
+        keep = make_backup(env.export_dir, "keep.zip")
+        drop = make_backup(env.export_dir, "drop.zip")
+        env.client.delete("/api/v3/backup/drop.zip")
+        assert keep.exists()
+        assert not drop.exists()
+
+    def test_directory_with_a_matching_name_is_not_removed(self, env):
+        # The delete loop matches by name but requires a regular file.
+        (env.export_dir / "sneaky.zip").mkdir()
+        response = env.client.delete("/api/v3/backup/sneaky.zip")
+        assert response.status_code == 404
+        assert (env.export_dir / "sneaky.zip").is_dir()
+
+
+class TestList:
+    def test_lists_only_zip_files(self, env):
+        make_backup(env.export_dir, "one.zip")
+        (env.export_dir / "notes.txt").write_text("ignore me")
+        response = env.client.get("/api/v3/backup/list")
+        assert response.status_code == 200
+        names = [entry["filename"] for entry in response.get_json()["data"]]
+        assert names == ["one.zip"]
+
+    def test_empty_directory_lists_nothing(self, env):
+        response = env.client.get("/api/v3/backup/list")
+        assert response.get_json()["data"] == []
+
+    def test_entries_carry_size_and_timestamp(self, env):
+        make_backup(env.export_dir, "one.zip")
+        entry = env.client.get("/api/v3/backup/list").get_json()["data"][0]
+        assert entry["size"] == len(b"PK\x03\x04fake zip")
+        assert entry["created_at"]
+
+
+class TestValidate:
+    def test_missing_file_is_a_400(self, env):
+        response = env.client.post("/api/v3/backup/validate", data={},
+                                   content_type="multipart/form-data")
+        assert response.status_code == 400
+        assert "No backup_file" in response.get_json()["message"]
+
+    def test_invalid_archive_is_a_400(self, env):
+        response = env.client.post(
+            "/api/v3/backup/validate",
+            data={"backup_file": (io.BytesIO(b"not a zip"), "bad.zip")},
+            content_type="multipart/form-data")
+        assert response.status_code == 400
+        assert "Invalid or corrupted" in response.get_json()["message"]
+
+    def test_validation_does_not_leave_temp_files_in_the_export_dir(self, env):
+        env.client.post(
+            "/api/v3/backup/validate",
+            data={"backup_file": (io.BytesIO(b"not a zip"), "bad.zip")},
+            content_type="multipart/form-data")
+        assert list(env.export_dir.iterdir()) == []

test/web_interface/test_api_v3_backup_restore.py

@@ -0,0 +1,262 @@
+"""
+Endpoint tests for POST /backup/restore.
+
+Restore is the most destructive operation the web interface exposes: it
+overwrites config, secrets, WiFi settings and fonts, and reinstalls
+plugins. It had no tests.
+
+restore_backup itself is mocked — this file is about what the route does
+with the request and with the result, not about ZIP handling, which
+belongs to backup_manager's own tests.
+
+Regression coverage for one fixed bug: a malformed `options` field fell
+back to {}, and since every RestoreOptions flag defaults to True, that
+turned a mis-serialized narrow restore into a full one — secrets
+included — with no indication anything had been ignored.
+"""
+
+import io
+import json
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+from flask import Flask
+
+project_root = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(project_root))
+
+from web_interface.blueprints.api_v3 import api_v3  # noqa: E402
+
+URL = "/api/v3/backup/restore"
+
+_MANAGER_ATTRS = (
+    'config_manager', 'plugin_manager', 'plugin_store_manager',
+    'plugin_state_manager', 'saved_repositories_manager', 'schema_manager',
+    'operation_queue', 'operation_history', 'cache_manager',
+)
+_SENTINEL = object()
+
+
+class FakeResult:
+    """Stand-in for backup_manager.RestoreResult."""
+
+    def __init__(self, success=True, restored=None, errors=None,
+                 plugins_to_install=None):
+        self.success = success
+        self.restored = restored if restored is not None else ["config"]
+        self.errors = errors or []
+        self.plugins_to_install = plugins_to_install or []
+        self.plugins_installed = []
+        self.plugins_failed = []
+
+    def to_dict(self):
+        return {
+            "success": self.success,
+            "restored": self.restored,
+            "errors": self.errors,
+            "plugins_installed": self.plugins_installed,
+            "plugins_failed": self.plugins_failed,
+        }
+
+
+@pytest.fixture
+def client():
+    originals = {name: getattr(api_v3, name, _SENTINEL) for name in _MANAGER_ATTRS}
+    for name in _MANAGER_ATTRS:
+        setattr(api_v3, name, MagicMock())
+
+    app = Flask(__name__)
+    app.config["TESTING"] = True
+    app.register_blueprint(api_v3, url_prefix="/api/v3")
+    yield app.test_client()
+
+    for name, original in originals.items():
+        if original is _SENTINEL:
+            if hasattr(api_v3, name):
+                delattr(api_v3, name)
+        else:
+            setattr(api_v3, name, original)
+
+
+@pytest.fixture
+def restore():
+    """Patch backup_manager.restore_backup (imported inside the handler)."""
+    with patch("src.backup_manager.restore_backup") as mock:
+        mock.return_value = FakeResult()
+        yield mock
+
+
+def post(client, options=None, filename="backup.zip", content=b"PK\x03\x04fake"):
+    data = {"backup_file": (io.BytesIO(content), filename)}
+    if options is not None:
+        data["options"] = options
+    return client.post(URL, data=data, content_type="multipart/form-data")
+
+
+class TestRequestValidation:
+    def test_missing_file_is_a_400(self, client, restore):
+        response = client.post(URL, data={}, content_type="multipart/form-data")
+        assert response.status_code == 400
+        assert "No backup_file" in response.get_json()["message"]
+        restore.assert_not_called()
+
+    def test_absent_options_defaults_to_a_full_restore(self, client, restore):
+        # Documented default, not the bug: omitting options entirely means
+        # "restore everything".
+        post(client)
+        options = restore.call_args[0][2]
+        assert options.restore_config is True
+        assert options.restore_secrets is True
+        assert options.reinstall_plugins is True
+
+    def test_partial_options_are_honoured(self, client, restore):
+        post(client, options=json.dumps({
+            "restore_secrets": False, "reinstall_plugins": False}))
+        options = restore.call_args[0][2]
+        assert options.restore_secrets is False
+        assert options.reinstall_plugins is False
+        assert options.restore_config is True  # unspecified stays default
+
+    @pytest.mark.parametrize("raw", ["{not json", "", "{'single': 'quotes'}"])
+    def test_malformed_options_are_refused(self, client, restore, raw):
+        # Regression: this fell back to {}, and every flag defaults to
+        # True, so a caller asking for a narrow restore and mis-serializing
+        # it got a full one — secrets overwritten — and no warning.
+        response = post(client, options=raw)
+        assert response.status_code == 400
+        assert "Invalid options" in response.get_json()["message"]
+        restore.assert_not_called()
+
+    @pytest.mark.parametrize("raw", ["[1,2,3]", '"a string"', "42", "true", "null"])
+    def test_options_that_are_not_an_object_are_refused(self, client, restore, raw):
+        response = post(client, options=raw)
+        assert response.status_code == 400
+        restore.assert_not_called()
+
+    def test_empty_object_is_accepted_as_all_defaults(self, client, restore):
+        assert post(client, options="{}").status_code == 200
+        assert restore.call_args[0][2].restore_config is True
+
+
+class TestSuccess:
+    def test_success_returns_the_result(self, client, restore):
+        restore.return_value = FakeResult(success=True, restored=["config", "secrets"])
+        response = post(client)
+        assert response.status_code == 200
+        body = response.get_json()
+        assert body["status"] == "success"
+        assert body["data"]["restored"] == ["config", "secrets"]
+
+    def test_temp_file_is_cleaned_up(self, client, restore):
+        seen = {}
+
+        def capture(path, project_root, options):
+            seen["path"] = Path(path)
+            assert seen["path"].exists()  # present while restoring
+            return FakeResult()
+
+        restore.side_effect = capture
+        post(client)
+        assert not seen["path"].exists()
+
+    def test_temp_file_cleaned_up_even_when_restore_raises(self, client, restore):
+        seen = {}
+
+        def blow_up(path, project_root, options):
+            seen["path"] = Path(path)
+            raise RuntimeError("corrupt archive")
+
+        restore.side_effect = blow_up
+        response = post(client)
+        assert response.status_code == 500
+        assert not seen["path"].exists()
+
+
+class TestPluginReinstall:
+    def test_plugins_are_reinstalled_when_requested(self, client, restore):
+        restore.return_value = FakeResult(
+            plugins_to_install=[{"plugin_id": "clock"}, {"plugin_id": "weather"}])
+        api_v3.plugin_store_manager.install_plugin.return_value = True
+        response = post(client)
+        assert response.status_code == 200
+        assert response.get_json()["data"]["plugins_installed"] == ["clock", "weather"]
+
+    def test_reinstall_skipped_when_not_requested(self, client, restore):
+        restore.return_value = FakeResult(plugins_to_install=[{"plugin_id": "clock"}])
+        post(client, options=json.dumps({"reinstall_plugins": False}))
+        api_v3.plugin_store_manager.install_plugin.assert_not_called()
+
+    def test_entries_without_a_plugin_id_are_skipped(self, client, restore):
+        restore.return_value = FakeResult(plugins_to_install=[{}, {"plugin_id": "clock"}])
+        api_v3.plugin_store_manager.install_plugin.return_value = True
+        post(client)
+        assert api_v3.plugin_store_manager.install_plugin.call_count == 1
+
+    def test_failed_reinstall_turns_the_whole_restore_into_an_error(
+            self, client, restore):
+        # Pinned as intentional: file restoration succeeded and does not
+        # touch result.errors, but a user whose plugins did not come back
+        # should not be told the restore was a success.
+        restore.return_value = FakeResult(
+            success=True, plugins_to_install=[{"plugin_id": "clock"}])
+        api_v3.plugin_store_manager.install_plugin.return_value = False
+        response = post(client)
+        assert response.status_code == 500
+        body = response.get_json()
+        assert body["status"] == "error"
+        assert "clock" in body["message"]
+
+    def test_message_names_what_landed_and_what_did_not(self, client, restore):
+        restore.return_value = FakeResult(
+            success=True, restored=["config", "fonts"],
+            plugins_to_install=[{"plugin_id": "clock"}])
+        api_v3.plugin_store_manager.install_plugin.return_value = False
+        message = post(client).get_json()["message"]
+        assert "restored: config, fonts" in message
+        assert "plugins not reinstalled: clock" in message
+
+    def test_install_exception_is_recorded_without_leaking_details(
+            self, client, restore):
+        restore.return_value = FakeResult(plugins_to_install=[{"plugin_id": "clock"}])
+        api_v3.plugin_store_manager.install_plugin.side_effect = RuntimeError(
+            "/srv/internal/path exploded")
+        body = post(client).get_json()
+        failures = body["data"]["plugins_failed"]
+        assert failures[0]["plugin_id"] == "clock"
+        assert "/srv/internal/path" not in json.dumps(body)
+
+    def test_missing_store_manager_is_reported_per_plugin(self, client, restore):
+        restore.return_value = FakeResult(plugins_to_install=[{"plugin_id": "clock"}])
+        api_v3.plugin_store_manager = None
+        with patch("web_interface.blueprints.api_v3.plugin_store_manager", None):
+            body = post(client).get_json()
+        assert body["data"]["plugins_failed"][0]["error"] == "Store manager unavailable"
+
+
+class TestFailureReporting:
+    def test_restore_errors_produce_a_500(self, client, restore):
+        restore.return_value = FakeResult(
+            success=False, restored=[], errors=["config: permission denied"])
+        response = post(client)
+        assert response.status_code == 500
+        assert "permission denied" in response.get_json()["message"]
+
+    def test_partial_restore_names_both_sides(self, client, restore):
+        restore.return_value = FakeResult(
+            success=False, restored=["config"], errors=["secrets: unwritable"])
+        message = post(client).get_json()["message"]
+        assert "restored: config" in message
+        assert "failed: secrets: unwritable" in message
+
+    def test_failure_without_detail_still_says_something(self, client, restore):
+        restore.return_value = FakeResult(success=False, restored=[], errors=[])
+        message = post(client).get_json()["message"]
+        assert "Restore incomplete" in message
+
+    def test_unexpected_exception_is_a_500(self, client, restore):
+        restore.side_effect = RuntimeError("boom")
+        response = post(client)
+        assert response.status_code == 500
+        assert response.get_json()["status"] == "error"

test/web_interface/test_api_v3_config_raw.py

@@ -0,0 +1,204 @@
+"""
+Endpoint tests for POST /config/raw/main and POST /config/raw/secrets.
+
+These write whatever JSON they are given straight to config.json and
+config_secrets.json, bypassing the secret-separation path that
+/config/main and the plugin-config endpoints go through. Given how much
+care the rest of the config surface takes to keep secrets out of
+config.json, an untested pair of endpoints that writes it verbatim is
+worth pinning precisely.
+
+Like test_api_v3_secret_roundtrip.py, these run a REAL ConfigManager over
+tmp_path so the assertions are against files on disk rather than mock
+calls.
+"""
+
+import json
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+from flask import Flask
+
+project_root = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(project_root))
+
+from src.config_manager import ConfigManager  # noqa: E402
+from src.exceptions import ConfigError  # noqa: E402
+from web_interface.blueprints.api_v3 import api_v3  # noqa: E402
+
+MAIN = "/api/v3/config/raw/main"
+SECRETS = "/api/v3/config/raw/secrets"
+
+
+@pytest.fixture
+def env(tmp_path):
+    config_file = tmp_path / "config.json"
+    config_file.write_text(json.dumps({"timezone": "UTC"}))
+    secrets_file = tmp_path / "config_secrets.json"
+
+    config_manager = ConfigManager(
+        config_path=str(config_file), secrets_path=str(secrets_file))
+    config_manager.template_path = str(tmp_path / "no-template.json")
+
+    _SENTINEL = object()
+    attrs = ('config_manager', 'plugin_manager', 'plugin_store_manager',
+             'plugin_state_manager', 'saved_repositories_manager',
+             'schema_manager', 'operation_queue', 'operation_history',
+             'cache_manager')
+    originals = {name: getattr(api_v3, name, _SENTINEL) for name in attrs}
+
+    for name in attrs:
+        setattr(api_v3, name, MagicMock())
+    api_v3.config_manager = config_manager
+
+    app = Flask(__name__)
+    app.config["TESTING"] = True
+    app.register_blueprint(api_v3, url_prefix="/api/v3")
+
+    class Env:
+        pass
+
+    e = Env()
+    e.client = app.test_client()
+    e.config_manager = config_manager
+    e.config_file = config_file
+    e.secrets_file = secrets_file
+    yield e
+
+    for name, original in originals.items():
+        if original is _SENTINEL:
+            if hasattr(api_v3, name):
+                delattr(api_v3, name)
+        else:
+            setattr(api_v3, name, original)
+
+
+class TestSaveRawMain:
+    def test_writes_the_body_to_config_json(self, env):
+        response = env.client.post(MAIN, json={"timezone": "America/Chicago"})
+        assert response.status_code == 200
+        assert json.loads(env.config_file.read_text()) == {"timezone": "America/Chicago"}
+
+    def test_replaces_rather_than_merges(self, env):
+        env.client.post(MAIN, json={"only": "this"})
+        assert json.loads(env.config_file.read_text()) == {"only": "this"}
+
+    def test_does_not_touch_the_secrets_file(self, env):
+        env.secrets_file.write_text(json.dumps({"weather": {"api_key": "k"}}))
+        env.client.post(MAIN, json={"timezone": "UTC"})
+        assert json.loads(env.secrets_file.read_text()) == {"weather": {"api_key": "k"}}
+
+    def test_uninitialized_manager_is_a_500(self, env):
+        api_v3.config_manager = None
+        response = env.client.post(MAIN, json={"timezone": "UTC"})
+        assert response.status_code == 500
+        assert "not initialized" in response.get_json()["message"]
+
+    def test_empty_object_is_a_400(self, env):
+        response = env.client.post(MAIN, json={})
+        assert response.status_code == 400
+        assert "No data provided" in response.get_json()["message"]
+
+    def test_bodyless_post_is_a_400(self, env):
+        response = env.client.post(MAIN)
+        assert response.status_code == 400
+        assert "No data provided" in response.get_json()["message"]
+
+    def test_malformed_json_is_a_400_in_the_app_shape(self, env):
+        response = env.client.post(MAIN, data="{not json",
+                                   content_type="application/json")
+        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):
+            raise ConfigError("cannot write", config_path="/etc/x.json")
+        monkeypatch.setattr(env.config_manager, "save_raw_file_content", refuse)
+        response = env.client.post(MAIN, json={"timezone": "UTC"})
+        assert response.status_code == 500
+        assert "/etc/x.json" in json.dumps(response.get_json())
+
+    def test_unexpected_error_is_a_500(self, env, monkeypatch):
+        def boom(kind, data):
+            raise RuntimeError("disk on fire")
+        monkeypatch.setattr(env.config_manager, "save_raw_file_content", boom)
+        response = env.client.post(MAIN, json={"timezone": "UTC"})
+        assert response.status_code == 500
+        assert response.get_json()["status"] == "error"
+
+
+class TestSaveRawSecrets:
+    def test_writes_only_to_the_secrets_file(self, env):
+        response = env.client.post(SECRETS, json={"weather": {"api_key": "s3cret"}})
+        assert response.status_code == 200
+        assert json.loads(env.secrets_file.read_text()) == {"weather": {"api_key": "s3cret"}}
+
+    def test_secret_values_never_reach_config_json(self, env):
+        env.client.post(SECRETS, json={"weather": {"api_key": "s3cret"}})
+        assert "s3cret" not in env.config_file.read_text()
+
+    def test_existing_main_config_is_untouched(self, env):
+        before = env.config_file.read_text()
+        env.client.post(SECRETS, json={"weather": {"api_key": "k"}})
+        assert env.config_file.read_text() == before
+
+    def test_github_token_is_reloaded_for_the_store_manager(self, env):
+        store = MagicMock()
+        store._load_github_token.return_value = "ghp_new"
+        api_v3.plugin_store_manager = store
+        env.client.post(SECRETS, json={"github": {"token": "ghp_new"}})
+        store._load_github_token.assert_called_once()
+        assert store.github_token == "ghp_new"
+
+    def test_absent_store_manager_is_fine(self, env):
+        api_v3.plugin_store_manager = None
+        assert env.client.post(SECRETS, json={"a": 1}).status_code == 200
+
+    def test_uninitialized_manager_is_a_500(self, env):
+        api_v3.config_manager = None
+        assert env.client.post(SECRETS, json={"a": 1}).status_code == 500
+
+    def test_empty_object_is_a_400(self, env):
+        assert env.client.post(SECRETS, json={}).status_code == 400
+
+    def test_bodyless_post_is_a_400(self, env):
+        assert env.client.post(SECRETS).status_code == 400
+
+    def test_error_is_a_500(self, env, monkeypatch):
+        def boom(kind, data):
+            raise RuntimeError("nope")
+        monkeypatch.setattr(env.config_manager, "save_raw_file_content", boom)
+        assert env.client.post(SECRETS, json={"a": 1}).status_code == 500
+
+
+class TestRawEndpointsBypassSecretSeparation:
+    """Pinned behaviour, deliberately not "fixed".
+
+    These endpoints are the escape hatch for editing the config files
+    directly from the web UI's raw JSON editor. They write what they are
+    given, so a secret typed into the main-config editor lands in
+    config.json in plain text — unlike /config/main and the plugin-config
+    endpoints, which route x-secret fields into config_secrets.json.
+
+    That is the point of a raw editor, but it is a sharp edge worth
+    stating out loud: anyone adding a "convenience" that posts plugin
+    config through this endpoint would silently lose secret separation.
+    """
+
+    def test_secret_shaped_keys_are_written_verbatim_to_main(self, env):
+        env.client.post(MAIN, json={"weather": {"api_key": "PLAINTEXT-KEY"}})
+        on_disk = json.loads(env.config_file.read_text())
+        assert on_disk["weather"]["api_key"] == "PLAINTEXT-KEY"
+
+    def test_no_separation_happens_on_the_raw_path(self, env):
+        env.client.post(MAIN, json={"weather": {"api_key": "PLAINTEXT-KEY"}})
+        # Nothing was moved aside into the secrets file.
+        assert not env.secrets_file.exists() or "PLAINTEXT-KEY" not in env.secrets_file.read_text()

test/web_interface/test_api_v3_secret_roundtrip.py

@@ -194,15 +194,84 @@ class TestSavePluginConfig:
 
     def test_secret_count_message_counts_top_level_keys(self, env):
         # Pinned: the "(N secret field(s))" message counts TOP-LEVEL keys of
-        # the separated secrets dict. Here that is 2: the posted accounts
+        # the separated secrets dict. Here that is 1: the posted accounts
-        # array (all its item tokens count as ONE key) plus the schema's
+        # array, whose item tokens all count as ONE key.
-        # api_key default ("") that merge_with_defaults adds before
+        #
-        # separation.
+        # It was 2 before blank secrets were dropped, the second being the
+        # schema's api_key default (""), which merge_with_defaults adds to
+        # every save. Counting it was the visible edge of a real bug: that
+        # injected blank was merged over the stored api_key, so saving any
+        # unrelated field destroyed the credential. See
+        # test_an_unrelated_edit_does_not_erase_a_stored_secret.
         resp = self._save(env, {
             "accounts": [{"name": "a", "token": "t"}],
         })
         message = resp.get_json()["message"]
-        assert "(2 secret field(s) saved to config_secrets.json)" in message
+        assert "(1 secret field(s) saved to config_secrets.json)" in message
+
+    def test_an_unrelated_edit_does_not_erase_a_stored_secret(self, env):
+        """Editing one field must not wipe the plugin's API key.
+
+        The config form renders secrets masked, so the browser posts them
+        back blank; merge_with_defaults injects a blank api_key even when
+        the client omits it entirely. Either way a "" reached the secrets
+        file and deep_merge wrote it over the stored credential.
+        """
+        assert self._save(env, {"api_key": "REAL-KEY-0123456789",
+                                "city": "Austin"}).status_code == 200
+        assert _on_disk(env.secrets_file)[PLUGIN_ID]["api_key"] == \
+            "REAL-KEY-0123456789"
+
+        # the user changes the city; the masked api_key rides along blank
+        assert self._save(env, {"api_key": "", "city": "Dallas"}).status_code == 200
+
+        assert _on_disk(env.secrets_file)[PLUGIN_ID]["api_key"] == \
+            "REAL-KEY-0123456789", "an unrelated edit destroyed the API key"
+        assert env.fresh_load()[PLUGIN_ID]["city"] == "Dallas"
+
+    def test_an_unrelated_edit_does_not_erase_array_item_secrets(self, env):
+        """The scalar api_key case above, but for a list of credentials.
+
+        remove_empty_secrets recursed into dicts only, so a list went into
+        deep_merge untouched -- and lists merge by *replacement*. Saving any
+        unrelated field posted [{"token": ""}, ...] straight over the stored
+        array and destroyed every token in it at once.
+        """
+        assert self._save(env, {"accounts": [
+            {"name": "a", "token": "REAL-A"},
+            {"name": "b", "token": "REAL-B"},
+        ], "city": "Austin"}).status_code == 200
+
+        # the user changes the city; both masked tokens ride along blank
+        assert self._save(env, {"accounts": [
+            {"name": "a", "token": ""},
+            {"name": "b", "token": ""},
+        ], "city": "Dallas"}).status_code == 200
+
+        merged = env.fresh_load()[PLUGIN_ID]
+        assert [a.get("token") for a in merged["accounts"]] == \
+            ["REAL-A", "REAL-B"], "an unrelated edit destroyed the array secrets"
+        assert [a["name"] for a in merged["accounts"]] == ["a", "b"]
+        assert merged["city"] == "Dallas"
+
+    def test_one_array_secret_can_be_changed_without_losing_the_rest(self, env):
+        assert self._save(env, {"accounts": [
+            {"name": "a", "token": "REAL-A"},
+            {"name": "b", "token": "REAL-B"},
+        ]}).status_code == 200
+        assert self._save(env, {"accounts": [
+            {"name": "a", "token": ""},
+            {"name": "b", "token": "NEW-B"},
+        ]}).status_code == 200
+
+        merged = env.fresh_load()[PLUGIN_ID]
+        assert [a.get("token") for a in merged["accounts"]] == ["REAL-A", "NEW-B"]
+
+    def test_a_secret_can_still_be_changed(self, env):
+        """Dropping blanks must not stop a real new value from being saved."""
+        self._save(env, {"api_key": "first-key"})
+        self._save(env, {"api_key": "second-key"})
+        assert _on_disk(env.secrets_file)[PLUGIN_ID]["api_key"] == "second-key"
 
     def test_resave_replaces_stored_secrets_list_wholesale(self, env):
         # Characterized: api_v3's deep_merge intentionally replaces lists,

test/web_interface/test_config_logging_omits_secrets.py

@@ -0,0 +1,45 @@
+"""The validation logging ran before separate_secrets, so it logged credentials.
+
+api_v3's plugin-config save logged `Full config: {plugin_config}` at INFO and
+`Config that failed: {plugin_config}` at ERROR. Both run *before*
+separate_secrets(), so plugin_config still held the values the user just typed
+into the form -- API keys and tokens went to the journal in clear text.
+"""
+import re
+from pathlib import Path
+
+import pytest
+
+SOURCE = (Path(__file__).resolve().parents[2]
+          / "web_interface" / "blueprints" / "api_v3.py")
+
+#: Objects that still hold submitted secret values at the point these log
+#: calls run. Interpolating one whole into a log message leaks credentials.
+UNREDACTED = ("plugin_config", "secrets_config", "current_secrets")
+
+
+def _logging_lines():
+    for number, line in enumerate(SOURCE.read_text(encoding="utf-8").splitlines(), 1):
+        stripped = line.strip()
+        if stripped.startswith("#"):
+            continue
+        if re.match(r"logger\.(debug|info|warning|error|critical|exception)\(", stripped):
+            yield number, stripped
+
+
+@pytest.mark.parametrize("name", UNREDACTED)
+def test_no_log_call_interpolates_a_whole_secret_bearing_object(name):
+    # {name} or {name['k']} leaks; {list(name.keys())} and {len(name)} do not.
+    bare = re.compile(r"\{" + re.escape(name) + r"(\[[^\]]*\])*\}")
+    offenders = [f"{n}: {text}" for n, text in _logging_lines() if bare.search(text)]
+    assert not offenders, (
+        f"{name} still holds submitted secrets where these log calls run:\n  "
+        + "\n  ".join(offenders))
+
+
+def test_the_guard_would_notice_a_reintroduced_leak():
+    """Pin the detector itself, so a rewrite cannot silently stop matching."""
+    bare = re.compile(r"\{" + re.escape("plugin_config") + r"(\[[^\]]*\])*\}")
+    assert bare.search('logger.info(f"Full config: {plugin_config}")')
+    assert bare.search("logger.error(f\"{plugin_config['api_key']}\")")
+    assert not bare.search('logger.info(f"{list(plugin_config.keys())}")')

test/web_interface/test_config_secrets_masking.py

@@ -0,0 +1,123 @@
+"""GET /config/secrets must not hand out credentials, and the client's
+read-modify-write cycle must not destroy them.
+
+This interface has no authentication. The endpoint returned the whole
+config_secrets.json to anyone who could reach the port; on one rig that was a
+40-character GitHub token, a 183-character Home Assistant token and three API
+keys. Masking it alone is not enough: the only client fetches every secret,
+edits one field and posts all of them back, so the write path has to treat an
+echoed mask as "unchanged".
+"""
+import json
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+
+from test_api_v3_secret_roundtrip import env, _on_disk  # noqa: F401,E402
+from src.web_interface.secret_helpers import SECRET_MASK  # noqa: E402
+
+STORED = {
+    "github": {"api_token": "ghp_" + "x" * 36},
+    "ledmatrix-weather": {"api_key": "w" * 32},
+    "incoming-packages": {"ha_token": "h" * 183},
+    "unset-plugin": {"api_key": ""},
+    "placeholder-plugin": {"api_key": "YOUR_API_KEY_HERE"},
+}
+
+
+def _seed(env):
+    env.secrets_file.write_text(json.dumps(STORED))
+
+
+def _get(env):
+    r = env.client.get("/api/v3/config/secrets")
+    assert r.status_code == 200, r.get_data(as_text=True)[:200]
+    return r.get_json()["data"]
+
+
+def test_no_credential_leaves_the_process(env):
+    _seed(env)
+    body = json.dumps(_get(env))
+    for secret in ("ghp_" + "x" * 36, "w" * 32, "h" * 183):
+        assert secret not in body, "endpoint returned a stored credential"
+
+
+def test_set_and_unset_remain_distinguishable(env):
+    _seed(env)
+    data = _get(env)
+    assert data["github"]["api_token"] == SECRET_MASK
+    assert data["unset-plugin"]["api_key"] == ""
+    assert data["placeholder-plugin"]["api_key"] == "YOUR_API_KEY_HERE"
+
+
+def test_the_clients_read_modify_write_preserves_every_other_secret(env):
+    """What the GitHub-token save button actually does."""
+    _seed(env)
+    secrets = _get(env)                      # everything arrives masked
+    secrets["github"]["api_token"] = "ghp_" + "n" * 36   # user changes one
+    r = env.client.post("/api/v3/config/raw/secrets", json=secrets)
+    assert r.status_code == 200, r.get_data(as_text=True)[:200]
+
+    on_disk = _on_disk(env.secrets_file)
+    assert on_disk["github"]["api_token"] == "ghp_" + "n" * 36, "new token not saved"
+    assert on_disk["ledmatrix-weather"]["api_key"] == "w" * 32
+    assert on_disk["incoming-packages"]["ha_token"] == "h" * 183
+
+
+def test_a_mask_echoed_back_is_never_stored(env):
+    _seed(env)
+    # Assert the write succeeded. A 500 leaves the old file in place, so the
+    # assertions below would hold without the write path running at all.
+    resp = env.client.post("/api/v3/config/raw/secrets", json=_get(env))
+    assert resp.status_code == 200, resp.get_data(as_text=True)[:200]
+    on_disk = _on_disk(env.secrets_file)
+    assert SECRET_MASK not in json.dumps(on_disk), "the mask was stored as a secret"
+    assert on_disk["github"]["api_token"] == "ghp_" + "x" * 36
+
+
+def test_a_brand_new_secret_can_still_be_added(env):
+    _seed(env)
+    env.client.post("/api/v3/config/raw/secrets",
+                    json={"new-plugin": {"api_key": "brand-new"}})
+    on_disk = _on_disk(env.secrets_file)
+    assert on_disk["new-plugin"]["api_key"] == "brand-new"
+    assert on_disk["github"]["api_token"] == "ghp_" + "x" * 36
+
+
+def test_a_list_of_secrets_keeps_its_shape(env):
+    """A list must not be masked as though it were one scalar.
+
+    accounts: [{...}, {...}] came back as a single '••••••••', so a caller
+    could not see how many entries existed, and the raw editor was shown a
+    string where the file holds an array.
+    """
+    env.secrets_file.write_text(json.dumps({
+        "myplugin": {"accounts": [{"name": "a", "token": "tok-a"},
+                                  {"name": "b", "token": "tok-b"}]}}))
+    accounts = _get(env)["myplugin"]["accounts"]
+    assert isinstance(accounts, list), "the list was flattened to a scalar"
+    assert len(accounts) == 2, "entries were lost"
+    assert all(isinstance(a, dict) for a in accounts), "entry shape was lost"
+    assert "tok-a" not in json.dumps(accounts), "a token survived masking"
+
+
+def test_a_list_posted_back_unchanged_is_left_alone(env):
+    """Lists merge by replacement, so a half-masked list must not be stored."""
+    original = {"myplugin": {"accounts": [{"name": "a", "token": "tok-a"},
+                                          {"name": "b", "token": "tok-b"}]}}
+    env.secrets_file.write_text(json.dumps(original))
+    resp = env.client.post("/api/v3/config/raw/secrets", json=_get(env))
+    assert resp.status_code == 200, resp.get_data(as_text=True)[:200]
+    assert _on_disk(env.secrets_file)["myplugin"]["accounts"] == \
+        original["myplugin"]["accounts"], "round-tripping the mask damaged the list"
+
+
+def test_a_fully_supplied_list_still_saves(env):
+    env.secrets_file.write_text(json.dumps(
+        {"myplugin": {"accounts": [{"name": "a", "token": "old"}]}}))
+    body = _get(env)
+    body["myplugin"]["accounts"] = [{"name": "a", "token": "new"}]
+    resp = env.client.post("/api/v3/config/raw/secrets", json=body)
+    assert resp.status_code == 200, resp.get_data(as_text=True)[:200]
+    assert _on_disk(env.secrets_file)["myplugin"]["accounts"][0]["token"] == "new"

test/web_interface/test_error_handler.py

@@ -0,0 +1,149 @@
+"""
+Tests for the response builders in src/web_interface/error_handler.py and
+the success path in src/web_interface/api_helpers.py.
+
+describe_exception() in the same module is already covered by
+test/test_web_error_detail.py and is not duplicated here.
+
+Regression coverage for one fixed bug: create_success_response used
+truthiness for `message` and `metadata` while using `is not None` for
+`data`, so an explicitly-passed "" or {} was silently dropped —
+api_helpers.success_response() repeated the same gate, which is the path
+every api_v3 endpoint actually calls.
+"""
+
+import pytest
+from flask import Flask
+
+from src.web_interface.api_helpers import success_response
+from src.web_interface.error_handler import (
+    create_error_response,
+    create_success_response,
+)
+from src.web_interface.errors import ErrorCode, WebInterfaceError
+
+
+@pytest.fixture
+def app():
+    return Flask(__name__)
+
+
+class TestCreateErrorResponse:
+    def test_returns_response_and_status_tuple(self, app):
+        with app.test_request_context():
+            response, status = create_error_response(
+                ErrorCode.CONFIG_SAVE_FAILED, "could not save")
+        assert status == 500
+        assert response.get_json()["message"] == "could not save"
+
+    def test_status_code_passthrough(self, app):
+        with app.test_request_context():
+            _, status = create_error_response(
+                ErrorCode.INVALID_INPUT, "bad", status_code=400)
+        assert status == 400
+
+    def test_body_matches_the_error_dataclass(self, app):
+        with app.test_request_context():
+            response, _ = create_error_response(
+                ErrorCode.NETWORK_ERROR, "offline",
+                details="connection refused", context={"url": "http://x"})
+        expected = WebInterfaceError(
+            error_code=ErrorCode.NETWORK_ERROR, message="offline",
+            details="connection refused", context={"url": "http://x"}).to_dict()
+        assert response.get_json() == expected
+
+    def test_none_context_produces_no_context_key(self, app):
+        with app.test_request_context():
+            response, _ = create_error_response(ErrorCode.SYSTEM_ERROR, "boom")
+        assert "context" not in response.get_json()
+
+    def test_suggested_fixes_passed_through(self, app):
+        with app.test_request_context():
+            response, _ = create_error_response(
+                ErrorCode.SYSTEM_ERROR, "boom", suggested_fixes=["Try again"])
+        assert response.get_json()["suggested_fixes"] == ["Try again"]
+
+
+class TestCreateSuccessResponse:
+    def test_bare_success(self):
+        assert create_success_response() == {"status": "success"}
+
+    def test_data_included(self):
+        assert create_success_response(data={"a": 1})["data"] == {"a": 1}
+
+    @pytest.mark.parametrize("falsy", [0, "", False, {}, []])
+    def test_falsy_data_is_still_included(self, falsy):
+        assert create_success_response(data=falsy)["data"] == falsy
+
+    def test_none_data_omitted(self):
+        assert "data" not in create_success_response(data=None)
+
+    def test_message_included(self):
+        assert create_success_response(message="done")["message"] == "done"
+
+    def test_empty_message_is_still_included(self):
+        # Regression: `if message:` dropped an explicitly-passed "".
+        assert create_success_response(message="")["message"] == ""
+
+    def test_none_message_omitted(self):
+        assert "message" not in create_success_response(message=None)
+
+    def test_metadata_included(self):
+        assert create_success_response(metadata={"v": 1})["metadata"] == {"v": 1}
+
+    def test_empty_metadata_is_still_included(self):
+        # Regression: `if metadata:` dropped an explicitly-passed {}.
+        assert create_success_response(metadata={})["metadata"] == {}
+
+    def test_none_metadata_omitted(self):
+        assert "metadata" not in create_success_response(metadata=None)
+
+
+class TestSuccessResponseHelper:
+    """api_helpers.success_response — the wrapper every endpoint calls."""
+
+    def test_plain_response_has_no_metadata_block(self, app):
+        with app.test_request_context():
+            body = success_response(data={"a": 1}).get_json()
+        assert body == {"status": "success", "data": {"a": 1}}
+
+    def test_explicit_empty_metadata_survives_the_wrapper(self, app):
+        # Regression: the wrapper re-gated metadata on truthiness after
+        # create_success_response had already included it, so {} was
+        # dropped again on the way out.
+        with app.test_request_context():
+            body = success_response(data=None, metadata={}).get_json()
+        assert body["metadata"] == {}
+
+    def test_caller_metadata_preserved(self, app):
+        with app.test_request_context():
+            body = success_response(metadata={"version": "1.2"}).get_json()
+        assert body["metadata"]["version"] == "1.2"
+
+    def test_timing_added_when_request_has_start_time(self, app):
+        with app.test_request_context() as ctx:
+            ctx.request.start_time = 0.0
+            body = success_response(data={"a": 1}).get_json()
+        assert "response_time_ms" in body["metadata"]
+
+    def test_timing_merges_with_caller_metadata(self, app):
+        with app.test_request_context() as ctx:
+            ctx.request.start_time = 0.0
+            body = success_response(metadata={"version": "1.2"}).get_json()
+        assert body["metadata"]["version"] == "1.2"
+        assert "response_time_ms" in body["metadata"]
+
+    def test_caller_metadata_dict_is_not_mutated(self, app):
+        # The helper used to add response_time_ms straight into the dict the
+        # caller passed, so a module-level or reused metadata dict would
+        # accumulate timings from previous requests.
+        caller_metadata = {"version": "1.2"}
+        with app.test_request_context() as ctx:
+            ctx.request.start_time = 0.0
+            success_response(metadata=caller_metadata)
+        assert caller_metadata == {"version": "1.2"}
+
+    def test_message_passed_through(self, app):
+        with app.test_request_context():
+            body = success_response(message="saved").get_json()
+        assert body["message"] == "saved"

test/web_interface/test_errors.py

@@ -0,0 +1,208 @@
+"""
+Tests for src/web_interface/errors.py — the structured error type behind
+every API error response (category inference, default suggestions, the
+JSON shape, and exception conversion).
+
+Pure logic; no Flask context needed.
+
+Regression coverage for one fixed bug: suggested_fixes used `or`, so a
+caller passing [] to mean "no suggestions" silently got the default list.
+"""
+
+import pytest
+
+from src.web_interface.errors import ErrorCategory, ErrorCode, WebInterfaceError
+
+
+class TestCategoryInference:
+    @pytest.mark.parametrize("code,expected", [
+        (ErrorCode.CONFIG_SAVE_FAILED, ErrorCategory.CONFIGURATION),
+        (ErrorCode.CONFIG_ROLLBACK_FAILED, ErrorCategory.CONFIGURATION),
+        (ErrorCode.PLUGIN_NOT_FOUND, ErrorCategory.PLUGIN),
+        (ErrorCode.PLUGIN_OPERATION_CONFLICT, ErrorCategory.PLUGIN),
+        (ErrorCode.VALIDATION_ERROR, ErrorCategory.VALIDATION),
+        (ErrorCode.SCHEMA_VALIDATION_FAILED, ErrorCategory.VALIDATION),
+        (ErrorCode.INVALID_INPUT, ErrorCategory.VALIDATION),
+        (ErrorCode.NETWORK_ERROR, ErrorCategory.NETWORK),
+        (ErrorCode.API_ERROR, ErrorCategory.NETWORK),
+        (ErrorCode.TIMEOUT, ErrorCategory.NETWORK),
+        (ErrorCode.PERMISSION_DENIED, ErrorCategory.PERMISSION),
+        (ErrorCode.FILE_PERMISSION_ERROR, ErrorCategory.PERMISSION),
+        (ErrorCode.SYSTEM_ERROR, ErrorCategory.SYSTEM),
+        (ErrorCode.SERVICE_UNAVAILABLE, ErrorCategory.SYSTEM),
+        (ErrorCode.UNKNOWN_ERROR, ErrorCategory.UNKNOWN),
+    ])
+    def test_every_code_prefix_maps_to_its_category(self, code, expected):
+        assert WebInterfaceError(code, "msg").category is expected
+
+    def test_explicit_category_overrides_inference(self):
+        error = WebInterfaceError(
+            ErrorCode.CONFIG_SAVE_FAILED, "msg", category=ErrorCategory.SYSTEM)
+        assert error.category is ErrorCategory.SYSTEM
+
+    def test_every_error_code_gets_a_category(self):
+        # No code may fall through uncategorized as the enum grows.
+        for code in ErrorCode:
+            assert isinstance(WebInterfaceError(code, "msg").category, ErrorCategory)
+
+
+class TestDefaultSuggestions:
+    def test_mapped_code_gets_specific_suggestions(self):
+        fixes = WebInterfaceError(ErrorCode.CONFIG_SAVE_FAILED, "msg").suggested_fixes
+        assert "Check available disk space" in fixes
+
+    def test_unmapped_code_gets_generic_fallback(self):
+        # PLUGIN_UPDATE_FAILED has no entry in suggestions_map.
+        fixes = WebInterfaceError(ErrorCode.PLUGIN_UPDATE_FAILED, "msg").suggested_fixes
+        assert fixes == ["Review error details and try again"]
+
+    def test_explicit_suggestions_win(self):
+        error = WebInterfaceError(
+            ErrorCode.CONFIG_SAVE_FAILED, "msg", suggested_fixes=["Do the thing"])
+        assert error.suggested_fixes == ["Do the thing"]
+
+    def test_explicit_empty_list_is_respected(self):
+        # Regression: `suggested_fixes or default` treated [] as "unset",
+        # so a caller could not express "I have no suggestions".
+        error = WebInterfaceError(
+            ErrorCode.CONFIG_SAVE_FAILED, "msg", suggested_fixes=[])
+        assert error.suggested_fixes == []
+
+    def test_none_still_gets_defaults(self):
+        error = WebInterfaceError(
+            ErrorCode.CONFIG_SAVE_FAILED, "msg", suggested_fixes=None)
+        assert len(error.suggested_fixes) > 0
+
+
+class TestToDict:
+    def test_base_keys_always_present(self):
+        result = WebInterfaceError(ErrorCode.SYSTEM_ERROR, "boom").to_dict()
+        assert result["status"] == "error"
+        assert result["error_code"] == "SYSTEM_ERROR"
+        assert result["error_category"] == "system"
+        assert result["message"] == "boom"
+
+    def test_details_included_when_set(self):
+        result = WebInterfaceError(
+            ErrorCode.SYSTEM_ERROR, "boom", details="disk full").to_dict()
+        assert result["details"] == "disk full"
+
+    def test_details_omitted_when_absent(self):
+        assert "details" not in WebInterfaceError(ErrorCode.SYSTEM_ERROR, "boom").to_dict()
+
+    def test_context_included_when_non_empty(self):
+        result = WebInterfaceError(
+            ErrorCode.SYSTEM_ERROR, "boom", context={"path": "/tmp/x"}).to_dict()
+        assert result["context"] == {"path": "/tmp/x"}
+
+    def test_empty_context_is_omitted(self):
+        # Pinned as intentional, not a bug: __init__ normalizes context to
+        # {}, and an empty context carries no information, so it is left out
+        # rather than padding every error body with "context": {}.
+        result = WebInterfaceError(ErrorCode.SYSTEM_ERROR, "boom", context={}).to_dict()
+        assert "context" not in result
+
+    def test_empty_suggestions_omitted(self):
+        result = WebInterfaceError(
+            ErrorCode.SYSTEM_ERROR, "boom", suggested_fixes=[]).to_dict()
+        assert "suggested_fixes" not in result
+
+    def test_is_json_serializable(self):
+        import json
+        error = WebInterfaceError(
+            ErrorCode.NETWORK_ERROR, "boom",
+            details="timeout", context={"url": "http://x"})
+        assert json.loads(json.dumps(error.to_dict()))["error_code"] == "NETWORK_ERROR"
+
+
+class TestFromException:
+    @pytest.mark.parametrize("exc_name,expected", [
+        ("ConfigError", ErrorCode.CONFIG_LOAD_FAILED),
+        ("PluginError", ErrorCode.PLUGIN_LOAD_FAILED),
+        ("PermissionError", ErrorCode.PERMISSION_DENIED),
+        ("AccessDenied", ErrorCode.PERMISSION_DENIED),
+        ("ValidationError", ErrorCode.VALIDATION_ERROR),
+        ("SchemaError", ErrorCode.VALIDATION_ERROR),
+        ("NetworkError", ErrorCode.NETWORK_ERROR),
+        ("ConnectionError", ErrorCode.NETWORK_ERROR),
+        ("TimeoutError", ErrorCode.TIMEOUT),
+        ("SomethingElse", ErrorCode.UNKNOWN_ERROR),
+    ])
+    def test_code_inferred_from_exception_class_name(self, exc_name, expected):
+        exc = type(exc_name, (Exception,), {})("boom")
+        assert WebInterfaceError.from_exception(exc).error_code is expected
+
+    def test_explicit_code_skips_inference(self):
+        error = WebInterfaceError.from_exception(
+            ValueError("boom"), error_code=ErrorCode.PLUGIN_NOT_FOUND)
+        assert error.error_code is ErrorCode.PLUGIN_NOT_FOUND
+
+    def test_message_is_the_safe_one_not_the_exception_text(self):
+        # The raw exception text is not echoed into `message`; that field is
+        # a fixed, user-facing string per code.
+        error = WebInterfaceError.from_exception(ValueError("secret-ish detail"))
+        assert error.message == "An unexpected error occurred"
+        assert "secret-ish" not in error.message
+
+    def test_exception_type_recorded_in_context(self):
+        error = WebInterfaceError.from_exception(ValueError("boom"))
+        assert error.context["exception_type"] == "ValueError"
+
+    def test_caller_context_is_preserved_alongside_type(self):
+        error = WebInterfaceError.from_exception(
+            ValueError("boom"), context={"plugin_id": "clock"})
+        assert error.context["plugin_id"] == "clock"
+        assert error.context["exception_type"] == "ValueError"
+
+    def test_caller_supplied_exception_type_is_overwritten(self):
+        error = WebInterfaceError.from_exception(
+            ValueError("boom"), context={"exception_type": "Fake"})
+        assert error.context["exception_type"] == "ValueError"
+
+    def test_original_error_retained(self):
+        exc = ValueError("boom")
+        assert WebInterfaceError.from_exception(exc).original_error is exc
+
+    def test_every_code_has_a_safe_message(self):
+        for code in ErrorCode:
+            assert WebInterfaceError._safe_message(code)
+
+
+class TestExceptionDetails:
+    def test_context_dict_is_flattened(self):
+        exc = ValueError("boom")
+        exc.context = {"config_path": "/etc/x.json", "line": 4}
+        details = WebInterfaceError._get_exception_details(exc)
+        assert "config_path: /etc/x.json" in details
+        assert "line: 4" in details
+        assert "; " in details
+
+    def test_exception_type_key_excluded(self):
+        exc = ValueError("boom")
+        exc.context = {"exception_type": "ValueError", "path": "/tmp/x"}
+        details = WebInterfaceError._get_exception_details(exc)
+        assert "exception_type" not in details
+        assert details == "path: /tmp/x"
+
+    def test_context_with_only_exception_type_gives_none(self):
+        exc = ValueError("boom")
+        exc.context = {"exception_type": "ValueError"}
+        assert WebInterfaceError._get_exception_details(exc) is None
+
+    def test_no_context_attribute_gives_none(self):
+        assert WebInterfaceError._get_exception_details(ValueError("boom")) is None
+
+    def test_non_dict_context_gives_none(self):
+        exc = ValueError("boom")
+        exc.context = "not a dict"
+        assert WebInterfaceError._get_exception_details(exc) is None
+
+    def test_empty_context_gives_none(self):
+        exc = ValueError("boom")
+        exc.context = {}
+        assert WebInterfaceError._get_exception_details(exc) is None
+
+    def test_details_flow_into_from_exception(self):
+        exc = ValueError("boom")
+        exc.context = {"config_path": "/etc/x.json"}
+        assert "config_path" in WebInterfaceError.from_exception(exc).details

test/web_interface/test_secret_helpers.py

@@ -17,6 +17,7 @@ from src.web_interface.secret_helpers import (
     separate_secrets,
     mask_secret_fields,
     mask_all_secret_values,
+    merge_secrets,
     remove_empty_secrets,
 )
 
@@ -239,3 +240,67 @@ class TestRemoveEmptySecrets:
     def test_keeps_falsey_non_string_values(self):
         # 0 and False are neither None nor blank strings — they are kept.
         assert remove_empty_secrets({"a": 0, "b": False}) == {"a": 0, "b": False}
+
+
+class TestArrayItemSecrets:
+    """Lists merge by replacement, so a blanked array wipes stored credentials.
+
+    remove_empty_secrets recursed into dicts but let a list through untouched,
+    so [{"token": ""}] went straight into deep_merge and overwrote the stored
+    list. Saving any unrelated setting destroyed every token in the array.
+    """
+
+    STORED = {"accounts": [{"name": "a", "token": "REAL-A"},
+                           {"name": "b", "token": "REAL-B"}]}
+
+    def test_an_unrelated_save_keeps_every_stored_token(self):
+        posted = {"accounts": [{"name": "a", "token": ""},
+                               {"name": "b", "token": ""}]}
+        merged = merge_secrets(self.STORED, remove_empty_secrets(posted))
+        assert [a["token"] for a in merged["accounts"]] == ["REAL-A", "REAL-B"]
+
+    def test_editing_one_entry_leaves_the_others_alone(self):
+        posted = {"accounts": [{"name": "a", "token": ""},
+                               {"name": "b", "token": "NEW-B"}]}
+        merged = merge_secrets(self.STORED, remove_empty_secrets(posted))
+        assert [a["token"] for a in merged["accounts"]] == ["REAL-A", "NEW-B"]
+
+    def test_a_new_entry_is_appended(self):
+        posted = {"accounts": [{"name": "a", "token": ""},
+                               {"name": "b", "token": ""},
+                               {"name": "c", "token": "NEW-C"}]}
+        merged = merge_secrets(self.STORED, remove_empty_secrets(posted))
+        assert [a["token"] for a in merged["accounts"]] == \
+            ["REAL-A", "REAL-B", "NEW-C"]
+
+    def test_a_list_of_bare_strings_merges_by_index(self):
+        merged = merge_secrets({"keys": ["K1", "K2", "K3"]},
+                               remove_empty_secrets({"keys": ["", "K2-NEW", ""]}))
+        assert merged["keys"] == ["K1", "K2-NEW", "K3"]
+
+    def test_an_all_blank_list_is_dropped_entirely(self):
+        posted = {"accounts": [{"token": ""}, {"token": ""}]}
+        assert "accounts" not in remove_empty_secrets(posted)
+
+    def test_plain_dict_secrets_are_unaffected(self):
+        merged = merge_secrets({"api_key": "OLD", "other": "keep"},
+                               remove_empty_secrets({"api_key": "", "other": "changed"}))
+        assert merged == {"api_key": "OLD", "other": "changed"}
+
+    def test_a_removed_entry_takes_its_secret_with_it(self):
+        """The regular config's list is authoritative about how many items
+        exist, and the secrets list runs parallel to it -- see
+        ConfigManager._strip_secrets_recursive. So a shorter incoming list
+        must shorten the stored secrets too, or the two fall out of step."""
+        posted = {"accounts": [{"name": "a", "token": "NEW-A"}]}
+        merged = merge_secrets(self.STORED, remove_empty_secrets(posted))
+        assert [a["token"] for a in merged["accounts"]] == ["NEW-A"]
+
+    def test_an_emptied_item_stays_a_dict_not_none(self):
+        """None there stops the list looking parallel, and
+        _strip_secrets_recursive then drops the whole key from the main
+        config -- deleting the item's non-secret fields as well."""
+        pruned = remove_empty_secrets(
+            {"accounts": [{"token": "real"}, {"token": ""}]})
+        assert pruned["accounts"] == [{"token": "real"}, {}]
+        assert None not in pruned["accounts"]

test/web_interface/test_validators.py

@@ -0,0 +1,284 @@
+"""
+Tests for src/web_interface/validators.py.
+
+dedup_unique_arrays is already covered by test_dedup_unique_arrays.py and
+is not repeated here; this file covers the other eight functions, none of
+which had any tests.
+
+Regression coverage for three fixed bugs:
+- validate_numeric_range accepted True/False, since bool subclasses int.
+- validate_file_upload lowercased the filename's extension but not the
+  caller's allowed_extensions list, so ['.TTF'] rejected 'font.ttf'.
+- validate_image_url only checked for '..' inside the relative-path
+  branch, so http://host/../secret passed validation untouched.
+"""
+
+import pytest
+
+from src.web_interface.validators import (
+    escape_html,
+    sanitize_plugin_config,
+    validate_file_upload,
+    validate_font_awesome_class,
+    validate_image_url,
+    validate_mime_type,
+    validate_numeric_range,
+    validate_string_length,
+)
+
+
+class TestEscapeHtml:
+    def test_escapes_all_five_entities(self):
+        assert escape_html("""<a href="x">O'Neill & co</a>""") == (
+            "&lt;a href=&quot;x&quot;&gt;O&#x27;Neill &amp; co&lt;/a&gt;")
+
+    def test_ampersand_is_escaped_first_so_nothing_double_escapes(self):
+        # If '<' were replaced before '&', the '&' of '&lt;' would be
+        # escaped again into '&amp;lt;'.
+        assert escape_html("<") == "&lt;"
+        assert escape_html("&") == "&amp;"
+        assert escape_html("&<") == "&amp;&lt;"
+
+    def test_plain_text_unchanged(self):
+        assert escape_html("hello world") == "hello world"
+
+    def test_non_string_is_coerced(self):
+        assert escape_html(42) == "42"
+        assert escape_html(None) == "None"
+
+    def test_script_tag_neutralized(self):
+        assert "<script>" not in escape_html("<script>alert(1)</script>")
+
+
+class TestValidateImageUrl:
+    @pytest.mark.parametrize("url", [
+        "javascript:alert(1)",
+        "JavaScript:alert(1)",
+        "JAVASCRIPT:alert(1)",
+        "data:text/html;base64,PHNjcmlwdD4=",
+        "vbscript:msgbox(1)",
+        "file:///etc/passwd",
+    ])
+    def test_dangerous_protocols_rejected(self, url):
+        valid, error = validate_image_url(url)
+        assert valid is False and "protocol" in error.lower()
+
+    @pytest.mark.parametrize("url", [
+        "http://x/a.png?onerror=alert(1)",
+        "http://x/a.png#onload=alert(1)",
+        "http://x/onclick=alert(1).png",
+    ])
+    def test_event_handlers_rejected(self, url):
+        valid, error = validate_image_url(url)
+        assert valid is False and "Event handlers" in error
+
+    @pytest.mark.parametrize("url", ["", None, 123, []])
+    def test_empty_or_non_string_rejected(self, url):
+        assert validate_image_url(url)[0] is False
+
+    def test_http_and_https_allowed(self):
+        assert validate_image_url("http://example.com/logo.png") == (True, None)
+        assert validate_image_url("https://example.com/logo.png") == (True, None)
+
+    def test_other_schemes_rejected(self):
+        valid, error = validate_image_url("ftp://example.com/logo.png")
+        assert valid is False and "http://" in error
+
+    def test_relative_path_allowed(self):
+        assert validate_image_url("/static/logo.png") == (True, None)
+
+    def test_protocol_relative_url_rejected(self):
+        assert validate_image_url("//evil.com/logo.png")[0] is False
+
+    def test_relative_traversal_rejected(self):
+        assert validate_image_url("/static/../../etc/passwd")[0] is False
+
+    def test_absolute_url_traversal_rejected(self):
+        # Regression: the '..' check used to sit inside the leading-slash
+        # branch, so an absolute URL skipped it entirely.
+        valid, error = validate_image_url("http://example.com/../secret")
+        assert valid is False and "traversal" in error.lower()
+
+    def test_bare_traversal_rejected(self):
+        assert validate_image_url("../../etc/passwd")[0] is False
+
+
+class TestValidateFontAwesomeClass:
+    @pytest.mark.parametrize("cls", ["fa-star", "fas fa-star", "fa-solid fa-house"])
+    def test_valid_classes_accepted(self, cls):
+        assert validate_font_awesome_class(cls) == (True, None)
+
+    @pytest.mark.parametrize("cls", ["star", "glyphicon-star", ""])
+    def test_classes_without_fa_prefix_rejected(self, cls):
+        assert validate_font_awesome_class(cls)[0] is False
+
+    def test_injection_attempt_rejected(self):
+        assert validate_font_awesome_class('fa-star" onload="alert(1)')[0] is False
+
+    def test_angle_brackets_rejected(self):
+        assert validate_font_awesome_class("<script>fa-star</script>")[0] is False
+
+    def test_non_string_rejected(self):
+        valid, error = validate_font_awesome_class(None)
+        assert valid is False and "string" in error
+
+    def test_explicit_fa_check_is_unreachable_but_harmless(self):
+        # Characterized, not fixed: the regex already requires 'fa-', so the
+        # follow-up `if 'fa-' not in class_name` can never fire. Anything
+        # lacking 'fa-' is rejected by the pattern first, with the pattern's
+        # own message.
+        valid, error = validate_font_awesome_class("star")
+        assert valid is False
+        assert error == "Invalid Font Awesome class name format"
+
+
+class TestValidateFileUpload:
+    def test_plain_filename_accepted(self):
+        assert validate_file_upload("logo.png") == (True, None)
+
+    @pytest.mark.parametrize("filename", [
+        "../etc/passwd", "dir/file.png", "dir\\file.png", "..\\..\\secrets",
+    ])
+    def test_traversal_characters_rejected(self, filename):
+        valid, error = validate_file_upload(filename)
+        assert valid is False and "invalid characters" in error
+
+    @pytest.mark.parametrize("filename", ["", None, 123])
+    def test_empty_or_non_string_rejected(self, filename):
+        assert validate_file_upload(filename)[0] is False
+
+    def test_allowed_extension_accepted(self):
+        assert validate_file_upload("font.ttf", allowed_extensions=[".ttf", ".otf"]) == (True, None)
+
+    def test_disallowed_extension_rejected(self):
+        valid, error = validate_file_upload("evil.exe", allowed_extensions=[".ttf"])
+        assert valid is False and "extension" in error
+
+    def test_uppercase_filename_extension_matches(self):
+        assert validate_file_upload("FONT.TTF", allowed_extensions=[".ttf"]) == (True, None)
+
+    def test_uppercase_allowed_list_matches(self):
+        # Regression: only the filename side was lowercased, so a caller
+        # passing ['.TTF'] rejected every valid .ttf upload.
+        assert validate_file_upload("font.ttf", allowed_extensions=[".TTF"]) == (True, None)
+
+    def test_no_extension_list_skips_the_check(self):
+        assert validate_file_upload("anything.xyz") == (True, None)
+
+
+class TestValidateMimeType:
+    def test_known_type_accepted(self):
+        assert validate_mime_type("logo.png", ["image/png"]) == (True, None)
+
+    def test_mismatched_type_rejected(self):
+        valid, error = validate_mime_type("logo.png", ["image/jpeg"])
+        assert valid is False and "not allowed" in error
+
+    def test_undeterminable_type_rejected(self):
+        valid, error = validate_mime_type("mystery.zzz", ["image/png"])
+        assert valid is False and "Could not determine" in error
+
+    def test_guess_type_failure_is_caught(self, monkeypatch):
+        import mimetypes
+        monkeypatch.setattr(mimetypes, "guess_type",
+                            lambda *a, **kw: (_ for _ in ()).throw(RuntimeError("boom")))
+        valid, error = validate_mime_type("logo.png", ["image/png"])
+        assert valid is False and "Error validating MIME type" in error
+
+
+class TestValidateNumericRange:
+    def test_value_in_range(self):
+        assert validate_numeric_range(5, min_val=0, max_val=10) == (True, None)
+
+    def test_boundaries_are_inclusive(self):
+        assert validate_numeric_range(0, min_val=0, max_val=10) == (True, None)
+        assert validate_numeric_range(10, min_val=0, max_val=10) == (True, None)
+
+    def test_below_minimum_rejected(self):
+        valid, error = validate_numeric_range(-1, min_val=0)
+        assert valid is False and "at least" in error
+
+    def test_above_maximum_rejected(self):
+        valid, error = validate_numeric_range(11, max_val=10)
+        assert valid is False and "at most" in error
+
+    def test_floats_accepted(self):
+        assert validate_numeric_range(2.5, min_val=0, max_val=10) == (True, None)
+
+    def test_no_bounds_accepts_any_number(self):
+        assert validate_numeric_range(-9999) == (True, None)
+
+    @pytest.mark.parametrize("value", ["5", None, [], {}])
+    def test_non_numeric_rejected(self, value):
+        valid, error = validate_numeric_range(value, min_val=0, max_val=10)
+        assert valid is False and error == "Value must be a number"
+
+    @pytest.mark.parametrize("value", [True, False])
+    def test_booleans_rejected(self, value):
+        # Regression: bool subclasses int, so True passed the isinstance
+        # check and then compared as 1 against the range.
+        valid, error = validate_numeric_range(value, min_val=0, max_val=10)
+        assert valid is False and error == "Value must be a number"
+
+
+class TestValidateStringLength:
+    def test_within_range(self):
+        assert validate_string_length("hello", min_length=1, max_length=10) == (True, None)
+
+    def test_boundaries_are_inclusive(self):
+        assert validate_string_length("abc", min_length=3, max_length=3) == (True, None)
+
+    def test_too_short_rejected(self):
+        valid, error = validate_string_length("", min_length=1)
+        assert valid is False and "at least" in error
+
+    def test_too_long_rejected(self):
+        valid, error = validate_string_length("abcdef", max_length=3)
+        assert valid is False and "at most" in error
+
+    def test_non_string_rejected(self):
+        valid, error = validate_string_length(123, max_length=10)
+        assert valid is False and "must be a string" in error
+
+    def test_no_bounds_accepts_anything(self):
+        assert validate_string_length("") == (True, None)
+
+
+class TestSanitizePluginConfig:
+    def test_valid_keys_and_scalars_kept(self):
+        config = {"enabled": True, "count": 3, "ratio": 1.5, "name": "clock"}
+        assert sanitize_plugin_config(config) == config
+
+    @pytest.mark.parametrize("key", ["has space", "has-dash", "has.dot", "has/slash", ""])
+    def test_invalid_key_names_dropped(self, key):
+        assert sanitize_plugin_config({key: "value", "good": 1}) == {"good": 1}
+
+    def test_non_string_keys_dropped(self):
+        assert sanitize_plugin_config({1: "a", "good": 2}) == {"good": 2}
+
+    def test_nested_dicts_recursed(self):
+        result = sanitize_plugin_config({"outer": {"inner": 1, "bad key": 2}})
+        assert result == {"outer": {"inner": 1}}
+
+    def test_list_of_scalars_preserved(self):
+        assert sanitize_plugin_config({"teams": ["PHI", "NYG"]})["teams"] == ["PHI", "NYG"]
+
+    def test_list_of_dicts_recursed(self):
+        result = sanitize_plugin_config({"items": [{"ok": 1, "bad key": 2}]})
+        assert result["items"] == [{"ok": 1}]
+
+    def test_unknown_value_types_dropped(self):
+        assert sanitize_plugin_config({"weird": {1, 2, 3}, "good": 1}) == {"good": 1}
+
+    def test_none_values_dropped(self):
+        assert sanitize_plugin_config({"nothing": None, "good": 1}) == {"good": 1}
+
+    def test_strings_are_not_html_escaped(self):
+        # Pinned, not a bug: escaping here would persist the escaped form in
+        # config.json. Output escaping belongs to the template layer, which
+        # the function's docstring now says explicitly.
+        payload = "<script>alert(1)</script>"
+        assert sanitize_plugin_config({"title": payload})["title"] == payload
+
+    def test_empty_config(self):
+        assert sanitize_plugin_config({}) == {}

web_interface/blueprints/api_v3.py

@@ -21,7 +21,10 @@ logger = logging.getLogger(__name__)
 # Import new infrastructure
 from src.web_interface.api_helpers import success_response, error_response, validate_request_json
 from src.web_interface.errors import ErrorCode
-from src.web_interface.secret_helpers import find_secret_fields, separate_secrets
+from src.web_interface.secret_helpers import (find_secret_fields, mask_all_secret_values,
+                                              merge_secrets, remove_empty_secrets,
+                                              separate_secrets,
+                                              strip_masked_values)
 from src.web_interface.error_handler import describe_exception, redact_text
 from src.plugin_system.operation_types import OperationType
 from src.web_interface.validators import (
@@ -367,7 +370,7 @@ def save_schedule_config():
         if not api_v3.config_manager:
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if not data:
             return jsonify({'status': 'error', 'message': 'No data provided'}), 400
 
@@ -575,7 +578,7 @@ def save_dim_schedule_config():
         if not api_v3.config_manager:
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if not data:
             return jsonify({'status': 'error', 'message': 'No data provided'}), 400
 
@@ -754,10 +757,12 @@ def save_main_config():
         if not data:
             return jsonify({'status': 'error', 'message': 'No data provided'}), 400
 
-        import logging
+        # What arrives here is the config itself, and the headers carry the
-        logging.error(f"DEBUG: save_main_config received data: {data}")
+        # session cookie -- neither belongs in the journal, least of all at
-        logging.error(f"DEBUG: Content-Type header: {request.content_type}")
+        # ERROR on every save. The shape of the request is the part with
-        logging.error(f"DEBUG: Headers: {dict(request.headers)}")
+        # diagnostic value, so log that, at the level it deserves.
+        logger.debug("save_main_config: %s, %d top-level key(s)",
+                     request.content_type or 'no content-type', len(data))
 
         # Merge with existing config (similar to original implementation)
         current_config = api_v3.config_manager.load_config()
@@ -1255,6 +1260,11 @@ def save_main_config():
 
                 # Separate secrets from regular config (same logic as save_plugin_config)
                 regular_config, secrets_config = separate_secrets(plugin_config, secret_fields)
+                # The config form renders secrets masked, so every save posts
+                # them back blank. Without this the blank is merged over the
+                # stored value and the credential is destroyed by the act of
+                # changing an unrelated setting. A blank means "unchanged".
+                secrets_config = remove_empty_secrets(secrets_config)
 
                 # PRE-PROCESSING: Preserve 'enabled' state if not in regular_config
                 # This prevents overwriting the enabled state when saving config from a form that doesn't include the toggle
@@ -1287,7 +1297,10 @@ def save_main_config():
                 if secrets_config:
                     if plugin_id not in current_secrets:
                         current_secrets[plugin_id] = {}
-                    current_secrets[plugin_id] = deep_merge(current_secrets[plugin_id], secrets_config)
+                    # Lists merge by replacement, so deep_merge here wrote a
+                    # blanked array straight over the stored credentials.
+                    current_secrets[plugin_id] = merge_secrets(
+                        current_secrets[plugin_id], secrets_config)
                     # Save secrets file
                     api_v3.config_manager.save_raw_file_content('secrets', current_secrets)
 
@@ -1372,7 +1385,12 @@ def get_secrets_config():
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
         config = api_v3.config_manager.get_raw_file_content('secrets')
-        return jsonify({'status': 'success', 'data': config})
+        # This interface has no authentication, and this file is nothing but
+        # credentials. It was handing all of them to anyone who could reach
+        # the port. Values are masked; empty and YOUR_* placeholders are left
+        # alone so a client can still tell "set" from "not set".
+        return jsonify({'status': 'success',
+                        'data': mask_all_secret_values(config)})
     except Exception as e:
         logger.error('Unhandled exception', exc_info=True)
         return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
@@ -1384,18 +1402,20 @@ def save_raw_main_config():
         if not api_v3.config_manager:
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
-        data = request.get_json()
+        # 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)
@@ -1430,21 +1450,33 @@ def save_raw_secrets_config():
         if not api_v3.config_manager:
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
-        data = request.get_json()
+        # 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
 
-        # Save the secrets config
+        # The GET above masks what it returns, and this endpoint's only client
-        api_v3.config_manager.save_raw_file_content('secrets', data)
+        # reads the whole file, edits one field and posts all of it back. So
+        # most of what arrives here is the mask, echoed rather than changed --
+        # storing it verbatim would replace every untouched credential with
+        # eight bullets. Strip those, then merge onto what is already stored,
+        # which makes "unchanged" mean unchanged.
+        #
+        # The cost is that a secret can no longer be cleared by blanking it.
+        # That needs its own affordance; a control that erases credentials as
+        # a side effect of saving an unrelated one is not it.
+        current = api_v3.config_manager.get_raw_file_content('secrets') or {}
+        merged = deep_merge(current, strip_masked_values(data))
+        api_v3.config_manager.save_raw_file_content('secrets', merged)
 
         # Reload GitHub token in plugin store manager if it exists
         if api_v3.plugin_store_manager:
             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)
@@ -1696,13 +1728,22 @@ def resolve_pull_command(project_dir):
     backup, or following an install guide that names one. The update button
     then reports a failure the user cannot act on.
 
+    ``--autostash`` is passed for the same reason. Rebase refuses to start
+    when any tracked file is modified, and on these installs something always
+    is: first_time_install.sh chmods five scripts that git tracked as 644, so
+    every machine that ran the installer carries five permanent mode changes
+    and the update button reports "cannot pull with rebase: You have unstaged
+    changes". Those modes are corrected in this commit, but a user cannot pull
+    the correction while the pull is what is blocked, and any other local edit
+    would reproduce it anyway. Autostash reapplies the changes afterwards.
+
     Returns ``(args, note, error)``. When ``origin/<branch>`` exists the pull
     is made explicit against it, so the update proceeds and the branch is
     given tracking information afterwards.
     """
     upstream = _git_upstream(project_dir)
     if upstream:
-        return ['git', 'pull', '--rebase'], '', None
+        return ['git', 'pull', '--rebase', '--autostash'], '', None
 
     branch = _git_current_branch(project_dir)
     if not branch:
@@ -1712,7 +1753,7 @@ def resolve_pull_command(project_dir):
         )
     if _git_remote_branch_exists(project_dir, branch):
         return (
-            ['git', 'pull', '--rebase', 'origin', branch],
+            ['git', 'pull', '--rebase', '--autostash', 'origin', branch],
             f"Branch '{branch}' had no upstream; pulled from origin/{branch} and set it as the upstream.",
             None,
         )
@@ -1860,6 +1901,33 @@ def get_system_version():
 _update_check_cache: Dict[str, Any] = {'result': None, 'ts': 0.0}
 _UPDATE_CHECK_TTL = 300  # 5 minutes — avoids a git fetch on every page load
 
+def _update_check_failed(detail: str) -> Dict[str, Any]:
+    """A check that could not run is not the same as being up to date.
+
+    Reporting update_available=False on a git failure hides the banner, and
+    the banner is the only route to the update button -- so a checkout git
+    refuses to touch looks exactly like a current one, permanently. The most
+    common cause is an install performed as root: git then reports "dubious
+    ownership" and every command fails, including the fetch here.
+    """
+    return {'update_available': False, 'remote_sha': 'unknown',
+            'commits_behind': 0, 'check_failed': True, 'error': detail}
+
+
+def _describe_git_failure(stderr: str) -> str:
+    """Turn git's stderr into something the user can act on."""
+    text = (stderr or '').strip()
+    if 'dubious ownership' in text or 'detected dubious ownership' in text:
+        return ("This checkout is owned by a different user than the one "
+                "running the web interface, so git refuses to use it. It is "
+                "usually the result of installing as root. Fix the ownership "
+                "and the update will work: sudo chown -R $USER:$USER "
+                + str(PROJECT_ROOT))
+    if 'could not resolve host' in text.lower() or 'network is unreachable' in text.lower():
+        return "Could not reach GitHub to check for updates."
+    return "Could not check for updates: " + (text.splitlines()[0] if text else "git failed")
+
+
 @api_v3.route('/system/check-update', methods=['GET'])
 def check_for_update():
     """Check whether a newer LEDMatrix commit is available on origin/main."""
@@ -1875,12 +1943,13 @@ def check_for_update():
             capture_output=True, timeout=10, cwd=cwd,
         )
         if fetch_result.returncode != 0:
+            stderr = fetch_result.stderr.decode(errors='replace').strip()
             logger.warning("check-update: git fetch failed (rc=%d): %s",
-                           fetch_result.returncode,
+                           fetch_result.returncode, stderr)
-                           fetch_result.stderr.decode(errors='replace').strip())
+            failed = _update_check_failed(_describe_git_failure(stderr))
-            _update_check_cache['result'] = _safe
+            _update_check_cache['result'] = failed
             _update_check_cache['ts'] = now
-            return jsonify(_safe)
+            return jsonify(failed)
         local = subprocess.run(
             ['git', 'rev-parse', 'HEAD'],
             capture_output=True, text=True, timeout=5, cwd=cwd,
@@ -1908,7 +1977,8 @@ def check_for_update():
         return jsonify(result)
     except Exception as e:
         logger.warning("check-update failed: %s", e)
-        return jsonify(_safe)
+        return jsonify(_update_check_failed(
+            "Could not check for updates; see logs for details."))
 
 @api_v3.route('/system/action', methods=['POST'])
 def execute_system_action():
@@ -2035,6 +2105,11 @@ def execute_system_action():
             except subprocess.TimeoutExpired:
                 logger.warning("git rev-parse timed out before pull")
 
+            # Whether the pull actually brought new code in. "Already up to
+            # date" is a success too, and prompting for a restart then would
+            # train users to ignore the prompt.
+            code_changed = False
+
             # Perform the git pull. Branches without an upstream were given
             # an explicit "origin <branch>" above so the update still works.
             result = subprocess.run(
@@ -2078,6 +2153,7 @@ def execute_system_action():
                                            capture_output=True, text=True, timeout=10, cwd=project_dir)
                     new_head = _post.stdout.strip() if _post.returncode == 0 else None
                     if old_head and new_head and old_head != new_head:
+                        code_changed = True
                         diff = subprocess.run(
                             ['git', 'diff', '--name-only', f'{old_head}..{new_head}'],
                             capture_output=True, text=True, timeout=15, cwd=project_dir)
@@ -2137,9 +2213,14 @@ def execute_system_action():
                                if ln.strip()), '')
                 pull_message = f"Update failed: {detail}" if detail else "Update failed; check logs for details"
 
+            # Nothing here restarts anything: the pull replaces files on
+            # disk while the display and web services keep running the code
+            # they loaded at boot. Without this the user is told the update
+            # succeeded and sees no change until they happen to reboot.
             return jsonify({
                 'status': 'success' if result.returncode == 0 else 'error',
                 'message': pull_message,
+                'restart_required': bool(result.returncode == 0 and code_changed),
             })
         elif action == 'checkout_branch':
             # Switch branches from the Tools tab. Needed because a checkout
@@ -2450,7 +2531,7 @@ def get_on_demand_status():
 def start_on_demand_display():
     """Request the display controller to run a specific plugin on-demand."""
     try:
-        data = request.get_json() or {}
+        data = request.get_json(silent=True) or {}
         plugin_id = data.get('plugin_id')
         mode = data.get('mode')
         duration = data.get('duration')
@@ -2974,7 +3055,7 @@ def manage_plugin_limits(plugin_id):
                 })
         else:
             # POST - Set limits
-            data = request.get_json() or {}
+            data = request.get_json(silent=True) or {}
             from src.plugin_system.resource_monitor import ResourceLimits
 
             limits = ResourceLimits(
@@ -3005,7 +3086,7 @@ def toggle_plugin():
         content_type = request.content_type or ''
 
         if 'application/json' in content_type:
-            data = request.get_json()
+            data = request.get_json(silent=True)
             if not data or 'plugin_id' not in data or 'enabled' not in data:
                 return jsonify({'status': 'error', 'message': 'plugin_id and enabled required'}), 400
             plugin_id = data['plugin_id']
@@ -3876,7 +3957,7 @@ def install_plugin():
         if not api_v3.plugin_store_manager:
             return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if not data or 'plugin_id' not in data:
             return jsonify({'status': 'error', 'message': 'plugin_id required'}), 400
 
@@ -4010,10 +4091,15 @@ def install_plugin_from_url():
         if not api_v3.plugin_store_manager:
             return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         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
@@ -4065,10 +4151,15 @@ def get_registry_from_url():
         if not api_v3.plugin_store_manager:
             return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         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
@@ -4110,10 +4201,15 @@ def add_saved_repository():
         if not api_v3.saved_repositories_manager:
             return jsonify({'status': 'error', 'message': 'Saved repositories manager not initialized'}), 500
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         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')
 
@@ -4141,7 +4237,7 @@ def remove_saved_repository():
         if not api_v3.saved_repositories_manager:
             return jsonify({'status': 'error', 'message': 'Saved repositories manager not initialized'}), 500
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if not data or 'repo_url' not in data:
             return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
 
@@ -4275,7 +4371,7 @@ def refresh_plugin_store():
         if not api_v3.plugin_store_manager:
             return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
 
-        data = request.get_json() or {}
+        data = request.get_json(silent=True) or {}
         fetch_commit_info = data.get('fetch_commit_info', data.get('fetch_latest_versions', False))
 
         # Force refresh the registry
@@ -5583,8 +5679,10 @@ def save_plugin_config():
         if schema:
             # Log what we're validating for debugging
             logger.info(f"Validating config for {plugin_id}")
+            # Only the shape. plugin_config still holds the submitted secret
+            # values at this point -- separate_secrets does not run until
+            # below -- so logging it wrote live credentials to the journal.
             logger.info(f"Config keys being validated: {list(plugin_config.keys())}")
-            logger.info(f"Full config: {plugin_config}")
 
             # Get enhanced schema keys (including injected core properties)
             # We need to create an enhanced schema to get the actual allowed keys
@@ -5607,7 +5705,8 @@ def save_plugin_config():
                 # Log validation errors for debugging
                 logger.error(f"Config validation failed for {plugin_id}")
                 logger.error(f"Validation errors: {validation_errors}")
-                logger.error(f"Config that failed: {plugin_config}")
+                # Keys only, for the same reason as above.
+                logger.error(f"Config keys that failed: {list(plugin_config.keys())}")
                 logger.error(f"Schema properties: {list(enhanced_schema.get('properties', {}).keys())}")
 
                 # Also print to console for immediate visibility
@@ -5638,6 +5737,11 @@ def save_plugin_config():
         # Separate secrets from regular config (handles nested configs and
         # array-item secrets — see src/web_interface/secret_helpers.py)
         regular_config, secrets_config = separate_secrets(plugin_config, secret_fields)
+        # The config form renders secrets masked, so every save posts
+        # them back blank. Without this the blank is merged over the
+        # stored value and the credential is destroyed by the act of
+        # changing an unrelated setting. A blank means "unchanged".
+        secrets_config = remove_empty_secrets(secrets_config)
 
         # Get current configs
         current_config = api_v3.config_manager.load_config()
@@ -5653,7 +5757,9 @@ def save_plugin_config():
         if secrets_config:
             if plugin_id not in current_secrets:
                 current_secrets[plugin_id] = {}
-            current_secrets[plugin_id] = deep_merge(current_secrets[plugin_id], secrets_config)
+            # See above -- secrets lists must merge element-wise.
+            current_secrets[plugin_id] = merge_secrets(
+                current_secrets[plugin_id], secrets_config)
             # Save secrets file
             try:
                 api_v3.config_manager.save_raw_file_content('secrets', current_secrets)
@@ -5861,7 +5967,7 @@ def reset_plugin_config():
         if not api_v3.config_manager:
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
-        data = request.get_json() or {}
+        data = request.get_json(silent=True) or {}
         plugin_id = data.get('plugin_id')
         preserve_secrets = data.get('preserve_secrets', True)
 
@@ -6248,7 +6354,7 @@ sys.exit(proc.returncode)
 def authenticate_spotify():
     """Run Spotify authentication script"""
     try:
-        data = request.get_json() or {}
+        data = request.get_json(silent=True) or {}
         redirect_url = data.get('redirect_url', '').strip()
 
         # Get plugin directory
@@ -6311,7 +6417,6 @@ sys.exit(proc.returncode)
                     timeout=120,
                     env=env
                 )
-                os.unlink(wrapper_path)
 
                 if result.returncode == 0:
                     return jsonify({
@@ -6326,9 +6431,13 @@ sys.exit(proc.returncode)
                         'output': result.stdout + result.stderr
                     }), 400
             except subprocess.TimeoutExpired:
+                return jsonify({'status': 'error', 'message': 'Authentication timed out'}), 408
+            finally:
+                # The wrapper carries the user's redirect URL, so it must not
+                # survive the request on any path — including a failure to
+                # launch, which the previous per-branch unlinks missed.
                 if os.path.exists(wrapper_path):
                     os.unlink(wrapper_path)
-                return jsonify({'status': 'error', 'message': 'Authentication timed out'}), 408
         else:
             # Step 1: Get authorization URL
             # Import the script's functions directly to get the auth URL
@@ -6565,7 +6674,7 @@ def get_fonts_overrides():
 def save_fonts_overrides():
     """Save font overrides"""
     try:
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if not data:
             return jsonify({'status': 'error', 'message': 'No data provided'}), 400
 
@@ -7185,7 +7294,7 @@ def upload_of_the_day_json():
 def delete_of_the_day_json():
     """Delete a JSON file from of-the-day plugin"""
     try:
-        data = request.get_json() or {}
+        data = request.get_json(silent=True) or {}
         file_id = data.get('file_id')  # This is the category_name
 
         if not file_id:
@@ -7275,6 +7384,29 @@ def serve_plugin_static(plugin_id, file_path):
         return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
 
 
+_MAX_CREDENTIAL_BACKUPS = 5
+
+
+def _prune_credential_backups(plugin_dir: Path) -> None:
+    """Keep only the newest _MAX_CREDENTIAL_BACKUPS credential backups.
+
+    Every re-upload copies the previous credentials.json aside. Without
+    pruning those accumulate for the life of the install — each one a
+    complete set of OAuth client credentials sitting in the plugin
+    directory.
+    """
+    backups = sorted(
+        plugin_dir.glob('credentials.json.backup.*'),
+        key=lambda p: p.stat().st_mtime,
+        reverse=True,
+    )
+    for stale in backups[_MAX_CREDENTIAL_BACKUPS:]:
+        try:
+            stale.unlink()
+        except OSError:
+            logger.warning("Could not remove old credential backup %s", stale.name)
+
+
 @api_v3.route('/plugins/calendar/upload-credentials', methods=['POST'])
 def upload_calendar_credentials():
     """Upload credentials.json file for calendar plugin"""
@@ -7302,24 +7434,20 @@ 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
+        # Validate it looks like Google OAuth credentials. A bare scalar, a
-        try:
+        # list, true/null — all valid JSON, none of them credentials. Reject
-            file.seek(0)
+        # rather than save: a file written as credentials.json but unusable
-            creds_data = json.loads(file.read())
+        # as credentials only fails later, somewhere less obvious.
-            file.seek(0)
+        if not isinstance(creds_data, dict) or not (
-
+                'installed' in creds_data or 'web' in creds_data):
-            # Check for required Google OAuth fields
+            return jsonify({
-            if 'installed' not in creds_data and 'web' not in creds_data:
+                'status': 'error',
-                return jsonify({
+                'message': 'File does not appear to be a valid Google OAuth credentials file'
-                    'status': 'error',
+            }), 400
-                    'message': 'File does not appear to be a valid Google OAuth credentials file'
-                }), 400
-        except Exception:
-            pass  # Continue even if validation fails
 
         # Get plugin directory
         plugin_id = 'calendar'
@@ -7339,6 +7467,7 @@ def upload_calendar_credentials():
             backup_path = Path(plugin_dir) / f'credentials.json.backup.{int(time.time())}'
             import shutil
             shutil.copy2(credentials_path, backup_path)
+            _prune_credential_backups(Path(plugin_dir))
 
         # Save new file
         file.save(str(credentials_path))
@@ -7863,7 +7992,7 @@ def connect_wifi():
     try:
         from src.wifi_manager import WiFiManager
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if not data:
             return jsonify({
                 'status': 'error',
@@ -8017,7 +8146,7 @@ def set_auto_enable_ap_mode():
     try:
         from src.wifi_manager import WiFiManager
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if data is None or 'auto_enable_ap_mode' not in data:
             return jsonify({
                 'status': 'error',
@@ -8146,7 +8275,7 @@ def delete_cache_file():
             from src.cache_manager import CacheManager
             api_v3.cache_manager = CacheManager()
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if not data or 'key' not in data:
             return jsonify({'status': 'error', 'message': 'cache key is required'}), 400
 
@@ -8409,7 +8538,16 @@ def backup_restore():
         try:
             opts_dict = json.loads(options_raw)
         except json.JSONDecodeError:
-            opts_dict = {}
+            opts_dict = None
+        if not isinstance(opts_dict, dict):
+            # Every option defaults to True, so falling back to {} on a
+            # parse failure would silently perform a FULL restore —
+            # secrets and all — for a caller who asked for a narrow one
+            # and mis-serialized it. Refuse instead of guessing.
+            return jsonify({
+                'status': 'error',
+                'message': 'Invalid options: expected a JSON object',
+            }), 400
         options = RestoreOptions(
             restore_config=bool(opts_dict.get('restore_config', True)),
             restore_secrets=bool(opts_dict.get('restore_secrets', True)),

web_interface/static/v3/app.js

@@ -116,14 +116,35 @@ document.body.addEventListener('htmx:afterRequest', function(event) {
 // ===== Restart-pending banner =====
 // Shown after restart-requiring saves; persists across tab switches (and
 // reloads, via sessionStorage) until the display restarts or it's dismissed.
-window.showRestartPending = function() {
+window.showRestartPending = function(message) {
-    try { sessionStorage.setItem('ledmatrix-restart-pending', '1'); } catch { /* private browsing */ }
+    try {
+        sessionStorage.setItem('ledmatrix-restart-pending', '1');
+        // Persisted alongside the flag: a code update and a config save want
+        // different wording, and the banner outlives the page that raised it.
+        if (message) sessionStorage.setItem('ledmatrix-restart-pending-text', message);
+        else sessionStorage.removeItem('ledmatrix-restart-pending-text');
+    } catch { /* private browsing */ }
     const banner = document.getElementById('restart-pending-banner');
+    const text = document.getElementById('restart-pending-text');
+    if (text) {
+        // Without the else-branch a config save inherited whatever wording the
+        // previous update left in the DOM: showRestartPending() clears the
+        // stored text but used to leave the element itself alone. The default
+        // is read back from the server-rendered copy rather than duplicated
+        // here, so the template stays the one place that owns the string.
+        if (text.dataset.defaultText === undefined) {
+            text.dataset.defaultText = text.textContent.trim();
+        }
+        text.textContent = message || text.dataset.defaultText;
+    }
     if (banner) banner.style.display = 'block';
 };
 
 window.dismissRestartPending = function() {
-    try { sessionStorage.removeItem('ledmatrix-restart-pending'); } catch { /* no-op */ }
+    try {
+        sessionStorage.removeItem('ledmatrix-restart-pending');
+        sessionStorage.removeItem('ledmatrix-restart-pending-text');
+    } catch { /* no-op */ }
     const banner = document.getElementById('restart-pending-banner');
     if (banner) banner.style.display = 'none';
 };
@@ -151,6 +172,9 @@ document.addEventListener('DOMContentLoaded', function() {
     try {
         if (sessionStorage.getItem('ledmatrix-restart-pending') === '1') {
             const banner = document.getElementById('restart-pending-banner');
+            const saved = sessionStorage.getItem('ledmatrix-restart-pending-text');
+            const text = document.getElementById('restart-pending-text');
+            if (text && saved) text.textContent = saved;
             if (banner) banner.style.display = 'block';
         }
     } catch { /* no-op */ }

web_interface/static/v3/plugins_manager.js

@@ -4622,15 +4622,17 @@ window.loadGithubToken = function() {
                 // Handle empty data (secrets file doesn't exist) - API returns {} in this case
                 const secrets = data.data || {};
                 const token = secrets.github?.api_token || '';
+                const configured = token && token !== 'YOUR_GITHUB_PERSONAL_ACCESS_TOKEN';
 
                 if (input) {
-                    if (token && token !== 'YOUR_GITHUB_PERSONAL_ACCESS_TOKEN') {
+                    // The endpoint masks what it returns, so this never holds
-                        // Token exists and is valid
+                    // the real token -- and the field is deliberately left
-                        input.value = token;
+                    // empty rather than filled with the mask, which would be
-                        showNotification('GitHub token loaded successfully', 'success');
+                    // saved verbatim the next time the user pressed Save.
+                    input.value = '';
+                    if (configured) {
+                        showNotification('A GitHub token is saved. Enter a new one to replace it.', 'success');
                     } else {
-                        // No token configured or placeholder value
-                        input.value = '';
                         showNotification('No GitHub token configured. Enter a new token to save.', 'info');
                     }
                 }

web_interface/templates/v3/base.html

@@ -413,7 +413,8 @@
             <div class="flex items-center justify-between">
                 <div class="flex items-center space-x-3">
                     <i class="fas fa-rotate text-lg"></i>
-                    <span class="text-sm font-medium" aria-live="polite">
+                    <span class="text-sm font-medium" aria-live="polite"
+                          id="restart-pending-text">
                         Configuration saved &mdash; restart the display to apply the changes
                     </span>
                 </div>
@@ -1107,15 +1108,29 @@
         fetch('/api/v3/system/check-update')
             .then(function(r) { return r.json(); })
             .then(function(data) {
+                var banner = document.getElementById('update-banner');
+                var btn = document.getElementById('update-banner-btn');
+                if (data.check_failed) {
+                    // A check that could not run is not the same as being up
+                    // to date. Hiding the banner here made a checkout git
+                    // refuses to touch look permanently current, with no
+                    // route to the update button and nothing to act on.
+                    document.getElementById('update-banner-text').textContent =
+                        data.error || 'Could not check for updates.';
+                    if (btn) btn.style.display = 'none';
+                    banner.style.display = '';
+                    return;
+                }
+                if (btn) btn.style.display = '';
                 if (data.update_available && getDismissedSha() !== data.remote_sha) {
                     var n = data.commits_behind || 0;
                     var msg = 'A new LEDMatrix update is available';
                     if (n > 0) msg += ' (' + n + ' commit' + (n > 1 ? 's' : '') + ')';
                     document.getElementById('update-banner-text').textContent = msg;
-                    document.getElementById('update-banner').style.display = '';
+                    banner.style.display = '';
                     try { sessionStorage.setItem('update-sha', data.remote_sha); } catch(e) {}
                 } else {
-                    document.getElementById('update-banner').style.display = 'none';
+                    banner.style.display = 'none';
                 }
             })
             .catch(function() {});
@@ -1146,6 +1161,13 @@
             if (data.status === 'success') {
                 document.getElementById('update-banner').style.display = 'none';
                 try { sessionStorage.removeItem('update-sha-dismissed'); } catch(e) {}
+                // The pull replaced files on disk; the running services still
+                // hold the code they loaded at boot. Ask for the restart that
+                // makes the update actually take effect.
+                if (data.restart_required && typeof window.showRestartPending === 'function') {
+                    window.showRestartPending(
+                        'Update installed \u2014 restart the display to run the new code');
+                }
             }
             if (typeof showNotification === 'function') {
                 showNotification(data.message || 'Update complete', data.status || 'success');