Merge pull request #456 from openhop-dev/merge-pr-454-455

Merge pr 454 455

manage.sh

@@ -761,7 +761,7 @@ EOF
     mkdir -p /etc/sudoers.d
     cat > /etc/sudoers.d/openhop-repeater <<'EOF'
 # Allow repeater user to manage the openhop-repeater service without password
-repeater ALL=(root) NOPASSWD: /usr/bin/systemctl restart openhop-repeater, /usr/bin/systemctl stop openhop-repeater, /usr/bin/systemctl start openhop-repeater, /usr/bin/systemctl status openhop-repeater, /usr/local/bin/pymc-do-upgrade
+repeater ALL=(root) NOPASSWD: /usr/bin/systemd-run --quiet --collect --unit=openhop-repeater-restart --on-active=2s --timer-property=AccuracySec=100ms --timer-property=RemainAfterElapse=no /usr/bin/systemctl restart openhop-repeater, /usr/bin/systemctl restart openhop-repeater, /usr/bin/systemctl stop openhop-repeater, /usr/bin/systemctl start openhop-repeater, /usr/bin/systemctl status openhop-repeater, /usr/local/bin/pymc-do-upgrade
 EOF
     chmod 0440 /etc/sudoers.d/openhop-repeater
 
@@ -1312,7 +1312,7 @@ EOF
     mkdir -p /etc/sudoers.d
     cat > /etc/sudoers.d/openhop-repeater <<'EOF'
 # Allow repeater user to manage the openhop-repeater service without password
-repeater ALL=(root) NOPASSWD: /usr/bin/systemctl restart openhop-repeater, /usr/bin/systemctl stop openhop-repeater, /usr/bin/systemctl start openhop-repeater, /usr/bin/systemctl status openhop-repeater, /usr/local/bin/pymc-do-upgrade
+repeater ALL=(root) NOPASSWD: /usr/bin/systemd-run --quiet --collect --unit=openhop-repeater-restart --on-active=2s --timer-property=AccuracySec=100ms --timer-property=RemainAfterElapse=no /usr/bin/systemctl restart openhop-repeater, /usr/bin/systemctl restart openhop-repeater, /usr/bin/systemctl stop openhop-repeater, /usr/bin/systemctl start openhop-repeater, /usr/bin/systemctl status openhop-repeater, /usr/local/bin/pymc-do-upgrade
 EOF
     chmod 0440 /etc/sudoers.d/openhop-repeater
 
@@ -1550,7 +1550,7 @@ UPGRADEEOF
         fi
         echo "✓ Configuration preserved"
         if [[ -n "$container_note" ]]; then
-            echo "$container_note"
+            printf '%b\n' "$container_note"
         fi
     else
         show_info "Upgrade Complete" "Upgrade completed successfully!\n\nVersion: $current_version → $new_version\n\n✓ Configuration preserved${container_note}"

radio-presets.json

@@ -145,7 +145,7 @@
                     "bandwidth": "250",
                     "coding_rate": "5"
                 },
-                  {
+                {
                     "title": "USA (Southern California)",
                     "description": "927.875MHz / SF7 / BW62.5 / CR7",
                     "frequency": "927.875",
@@ -153,6 +153,14 @@
                     "bandwidth": "62.5",
                     "coding_rate": "7"
                 },
+                {
+                    "title": "USA (Philadelphia Metro)",
+                    "description": "902.250MHz / SF11 / BW500 / CR5",
+                    "frequency": "902.250",
+                    "spreading_factor": "11",
+                    "bandwidth": "500",
+                    "coding_rate": "5"
+                },
                 {
                     "title": "Vietnam",
                     "description": "920.250MHz / SF11 / BW250 / CR5",

repeater/config_manager.py

@@ -1,5 +1,9 @@
+import copy
 import logging
 import os
+import stat
+import tempfile
+import threading
 from typing import Any, Dict, List, Optional
 
 import yaml
@@ -9,6 +13,9 @@ from repeater.modem_config import normalize_modem_config_in_place
 
 logger = logging.getLogger("ConfigManager")
 
+# Serialize staging, persistence and publication across managers in this process.
+_CONFIG_WRITE_LOCK = threading.RLock()
+
 
 class ConfigManager:
     """Manages configuration persistence and live updates to the daemon."""
@@ -236,32 +243,69 @@ class ConfigManager:
         return True
 
     def save_to_file(self) -> bool:
-        """
-        Save current config to YAML file.
+        """Atomically save current config, normalizing shared state only on success."""
+        with _CONFIG_WRITE_LOCK:
+            try:
+                candidate = copy.deepcopy(self.config)
+                if not self._persist_config(candidate):
+                    return False
+                normalize_modem_config_in_place(self.config)
+                return True
+            except Exception:
+                logger.exception("Failed to save config to %s", self.config_path)
+                return False
 
-        Returns:
-            True if successful, False otherwise
-        """
+    def _persist_config(self, config: dict) -> bool:
+        """Write a private candidate; the atomic replace is the commit point."""
+        temporary_path = None
         try:
-            normalize_modem_config_in_place(self.config)
-            os.makedirs(os.path.dirname(self.config_path), exist_ok=True)
-            with open(self.config_path, "w", encoding="utf-8") as f:
-                # Use safe_dump with explicit width to prevent line wrapping
-                # Setting width to a very large number prevents truncation of long strings like identity keys
+            normalize_modem_config_in_place(config)
+            # Follow existing symlinks rather than replacing the link itself.
+            path = os.path.realpath(self.config_path)
+            directory = os.path.dirname(path)
+            os.makedirs(directory, exist_ok=True)
+            try:
+                previous = os.stat(path)
+            except FileNotFoundError:
+                previous = None
+            with tempfile.NamedTemporaryFile(
+                mode="w", encoding="utf-8", dir=directory, prefix=".config-", delete=False
+            ) as f:
+                temporary_path = f.name
                 yaml.safe_dump(
-                    self.config,
+                    config,
                     f,
                     default_flow_style=False,
                     indent=2,
-                    width=1000000,  # Very large width to prevent any line wrapping
+                    width=1000000,
                     sort_keys=False,
                     allow_unicode=True,
                 )
-            logger.info(f"Configuration saved to {self.config_path}")
+                f.flush()
+                if previous is not None:
+                    current = os.fstat(f.fileno())
+                    if (current.st_uid, current.st_gid) != (previous.st_uid, previous.st_gid):
+                        os.fchown(f.fileno(), previous.st_uid, previous.st_gid)
+                    os.fchmod(f.fileno(), stat.S_IMODE(previous.st_mode))
+                os.fsync(f.fileno())
+            os.replace(temporary_path, path)
+            temporary_path = None
+            logger.info("Configuration saved to %s", self.config_path)
             return True
-        except Exception as e:
-            logger.error(f"Failed to save config to {self.config_path}: {e}", exc_info=True)
+        except Exception:
+            logger.exception("Failed to save config to %s", self.config_path)
             return False
+        finally:
+            if temporary_path is not None:
+                os.unlink(temporary_path)
+
+    @staticmethod
+    def _apply_updates(config: dict, updates: dict) -> None:
+        for section, values in updates.items():
+            if isinstance(values, dict):
+                config.setdefault(section, {}).update(values)
+            else:
+                config[section] = values
 
     def live_update_daemon(self, sections: Optional[List[str]] = None) -> bool:
         """
@@ -411,39 +455,37 @@ class ConfigManager:
         """
         result: Dict[str, Any] = {"success": False, "saved": False, "live_updated": False}
 
-        try:
-            # Apply updates to config
-            for section, values in updates.items():
-                if section not in self.config:
-                    self.config[section] = {}
+        with _CONFIG_WRITE_LOCK:
+            try:
+                # Stage privately so a failed write cannot leak into running state.
+                updates = copy.deepcopy(updates)
+                candidate = copy.deepcopy(self.config)
+                self._apply_updates(candidate, updates)
+                result["saved"] = self._persist_config(candidate)
 
-                if isinstance(values, dict):
-                    self.config[section].update(values)
-                else:
-                    self.config[section] = values
+                if not result["saved"]:
+                    result["error"] = "Failed to save config to file"
+                    return result
 
-            # Save to file
-            result["saved"] = self.save_to_file()
+                # Preserve shared section references used by daemon helpers.
+                self._apply_updates(self.config, updates)
+                normalize_modem_config_in_place(self.config)
 
-            if not result["saved"]:
-                result["error"] = "Failed to save config to file"
+                # Live update daemon if requested
+                if live_update:
+                    # Auto-detect sections if not specified
+                    if live_update_sections is None:
+                        live_update_sections = list(updates.keys())
+
+                    result["live_updated"] = self.live_update_daemon(live_update_sections)
+
+                result["success"] = result["saved"]
                 return result
 
-            # Live update daemon if requested
-            if live_update:
-                # Auto-detect sections if not specified
-                if live_update_sections is None:
-                    live_update_sections = list(updates.keys())
-
-                result["live_updated"] = self.live_update_daemon(live_update_sections)
-
-            result["success"] = result["saved"]
-            return result
-
-        except Exception as e:
-            logger.error(f"Error in update_and_save: {e}", exc_info=True)
-            result["error"] = str(e)
-            return result
+            except Exception as e:
+                logger.error(f"Error in update_and_save: {e}", exc_info=True)
+                result["error"] = str(e)
+                return result
 
     def update_nested(self, path: str, value: Any, live_update: bool = True) -> Dict[str, Any]:
         """

repeater/service_utils.py

@@ -246,9 +246,9 @@ def restart_service() -> Tuple[bool, str]:
     Restart the openhop-repeater service.
 
     On Buildroot/Luckfox, use the shipped init script directly.
-    On systemd hosts, try polkit-based restart first (plain systemctl), then
-    fall back to sudo-based restart (requires sudoers.d rule installed by
-    manage.sh).
+    On systemd hosts, submit a delayed transient timer through the exact
+    sudo command authorized by manage.sh. The restart runs outside this
+    service's cgroup, after the API can acknowledge successful scheduling.
 
     Returns:
         Tuple[bool, str]: (success, message)
@@ -277,62 +277,48 @@ def restart_service() -> Tuple[bool, str]:
             logger.error(f"Buildroot restart failed: {exc}")
             return False, f"Restart failed: {exc}"
 
-    # Try polkit-based restart first (works on bare metal / VMs with polkit running)
+    # A command launched inside this service shares its cgroup, even with
+    # --no-block or start_new_session. PID 1 must own the delayed restart so
+    # submission can return before our process group receives SIGTERM.
+    command = [
+        "/usr/bin/systemd-run",
+        "--quiet",
+        "--collect",
+        "--unit=openhop-repeater-restart",
+        "--on-active=2s",
+        "--timer-property=AccuracySec=100ms",
+        "--timer-property=RemainAfterElapse=no",
+        _SYSTEMCTL_BIN,
+        "restart",
+        "openhop-repeater",
+    ]
+    if os.geteuid() != 0:
+        command = [_SUDO_BIN, "--non-interactive", *command]
     try:
-        result = subprocess.run(
-            [_SYSTEMCTL_BIN, "restart", "openhop-repeater"],
-            capture_output=True,
-            text=True,
-            timeout=5,
-        )  # nosec B603
-
+        result = subprocess.run(command, capture_output=True, text=True, timeout=5)  # nosec B603
         if result.returncode == 0:
-            logger.info("Service restart via polkit succeeded")
-            return True, "Service restart initiated"
-
-        stderr = result.stderr or ""
-        if "Access denied" in stderr or "authorization" in stderr.lower():
-            logger.info("Polkit denied restart, trying sudo fallback...")
-        else:
-            # Some other error, still try sudo
-            logger.warning(f"systemctl restart failed ({result.returncode}): {stderr.strip()}")
-
+            logger.info("Service restart scheduled via systemd timer")
+            return True, "Service restart initiated (scheduled in 2 seconds)"
+        if result.returncode < 0:
+            logger.warning("Restart job submission interrupted (%s)", result.returncode)
+            return (
+                False,
+                "Restart unconfirmed: command interrupted; check service status before retrying",
+            )
+        error_msg = result.stderr.strip() or f"exit status {result.returncode}"
+        logger.error("Restart scheduling failed: %s", error_msg)
+        return False, f"Restart failed: {error_msg}"
     except subprocess.TimeoutExpired:
-        # Timeout likely means it's restarting - that's success
-        logger.warning("Service restart command timed out (service may be restarting)")
-        return True, "Service restart initiated (timeout - likely restarting)"
+        logger.warning("Restart scheduling timed out; acceptance is unconfirmed")
+        return (
+            False,
+            "Restart unconfirmed: submission timeout; check service status before retrying",
+        )
     except FileNotFoundError:
-        logger.error("systemctl not found")
-        return False, "systemctl not available"
-    except Exception as e:
-        logger.warning(f"Polkit restart attempt failed: {e}")
-
-    # Fallback: use sudo (requires /etc/sudoers.d/openhop-repeater rule)
-    try:
-        result = subprocess.run(
-            [_SUDO_BIN, "--non-interactive", _SYSTEMCTL_BIN, "restart", "openhop-repeater"],
-            capture_output=True,
-            text=True,
-            timeout=5,
-        )  # nosec B603
-
-        if result.returncode == 0:
-            logger.info("Service restart via sudo succeeded")
-            return True, "Service restart initiated"
-        else:
-            error_msg = result.stderr or "Unknown error"
-            logger.error(f"Service restart via sudo failed: {error_msg}")
-            return False, f"Restart failed: {error_msg}"
-
-    except subprocess.TimeoutExpired:
-        logger.warning("Sudo restart timed out (service likely restarting)")
-        return True, "Service restart initiated (timeout - likely restarting)"
-    except FileNotFoundError:
-        logger.error("sudo not found - cannot restart service")
-        return False, "Neither polkit nor sudo available for service restart"
-    except Exception as e:
-        logger.error(f"Error executing sudo restart: {e}")
-        return False, f"Restart command failed: {str(e)}"
+        return False, "Restart scheduler unavailable; install systemd-run and sudo via manage.sh"
+    except Exception as exc:
+        logger.error("Error scheduling restart: %s", exc)
+        return False, f"Restart command failed: {exc}"
 
 
 def _is_cherrypy_engine_running() -> Optional[bool]:

repeater/web/auth_endpoints.py

@@ -2,6 +2,7 @@
 Authentication endpoints for login and token management
 """
 
+import hashlib
 import logging
 import math
 import threading
@@ -35,20 +36,30 @@ class _LoginThrottle:
         self.base_backoff_sec = base_backoff_sec
         self.max_backoff_sec = max_backoff_sec
         self.window_sec = window_sec
+        self.max_states = 4096
         self._time_fn = time_fn or time.monotonic
         self._lock = threading.Lock()
         self._ip_states = {}
         self._user_states = {}
         self._global_state = {"failures": 0, "last_failure": 0.0, "blocked_until": 0.0}
 
-    def _state(self, bucket: dict, key: str):
+    @staticmethod
+    def _keys(client_ip: str, username: str) -> tuple[bytes, bytes]:
+        # Fixed-size digests also bound storage for oversized attacker input.
+        user_key = hashlib.sha256(
+            ((username or "").strip().lower() or "<unknown>").encode("utf-8")
+        ).digest()
+        ip_key = hashlib.sha256((client_ip or "<unknown>").encode("utf-8")).digest()
+        return ip_key, user_key
+
+    def _state(self, bucket: dict, key: bytes):
         if key not in bucket:
             bucket[key] = {"failures": 0, "last_failure": 0.0, "blocked_until": 0.0}
         return bucket[key]
 
     def _maybe_decay(self, state: dict, now: float) -> None:
         last = state.get("last_failure", 0.0)
-        if last and (now - last) > self.window_sec:
+        if (now - last) > self.window_sec and state.get("blocked_until", 0.0) <= now:
             state["failures"] = 0
             state["blocked_until"] = 0.0
 
@@ -68,21 +79,48 @@ class _LoginThrottle:
             return 0
         return max(1, math.ceil(blocked_until - now))
 
+    def _get_retry_after(self, ip_key: bytes, user_key: bytes, now: float) -> int:
+        # Rejected requests must not allocate attacker-controlled keys.
+        global_retry = self._retry_after(self._global_state, now)
+        if global_retry:
+            return global_retry
+        for bucket in (self._ip_states, self._user_states):
+            expired = [
+                key
+                for key, state in bucket.items()
+                if now - state["last_failure"] > self.window_sec and state["blocked_until"] <= now
+            ]
+            for key in expired:
+                del bucket[key]
+        retry = 0
+        for bucket, key in ((self._ip_states, ip_key), (self._user_states, user_key)):
+            if key in bucket:
+                retry = max(retry, self._retry_after(bucket[key], now))
+            elif len(bucket) >= self.max_states:
+                # Fail closed until space expires; evicting active failures would
+                # let a rotating-key attacker reset another client's backoff.
+                expires = min(
+                    max(state["last_failure"] + self.window_sec, state["blocked_until"])
+                    for state in bucket.values()
+                )
+                retry = max(retry, max(1, math.ceil(expires - now)))
+        return retry
+
     def get_retry_after(self, client_ip: str, username: str) -> int:
         now = self._time_fn()
-        user_key = (username or "").strip().lower() or "<unknown>"
-        ip_key = client_ip or "<unknown>"
+        ip_key, user_key = self._keys(client_ip, username)
         with self._lock:
-            ip_retry = self._retry_after(self._state(self._ip_states, ip_key), now)
-            user_retry = self._retry_after(self._state(self._user_states, user_key), now)
-            global_retry = self._retry_after(self._global_state, now)
-            return max(ip_retry, user_retry, global_retry)
+            return self._get_retry_after(ip_key, user_key, now)
 
     def register_failure(self, client_ip: str, username: str) -> int:
         now = self._time_fn()
-        user_key = (username or "").strip().lower() or "<unknown>"
-        ip_key = client_ip or "<unknown>"
+        ip_key, user_key = self._keys(client_ip, username)
         with self._lock:
+            # Recheck under the lock: other requests may have filled or blocked
+            # the store while this request was validating credentials.
+            retry = self._get_retry_after(ip_key, user_key, now)
+            if retry:
+                return retry
             self._record_failure(self._state(self._ip_states, ip_key), self.per_ip_threshold, now)
             self._record_failure(
                 self._state(self._user_states, user_key), self.per_user_threshold, now
@@ -94,8 +132,7 @@ class _LoginThrottle:
             return max(ip_retry, user_retry, global_retry)
 
     def register_success(self, client_ip: str, username: str) -> None:
-        user_key = (username or "").strip().lower() or "<unknown>"
-        ip_key = client_ip or "<unknown>"
+        ip_key, user_key = self._keys(client_ip, username)
         with self._lock:
             self._ip_states.pop(ip_key, None)
             self._user_states.pop(user_key, None)

tests/test_config_persistence_safety.py

@@ -0,0 +1,121 @@
+import copy
+import os
+import stat
+from concurrent.futures import ThreadPoolExecutor
+from threading import Event
+from unittest.mock import Mock
+
+import pytest
+import yaml
+
+from repeater.config_manager import ConfigManager
+
+
+@pytest.mark.parametrize("failure", ["dump", "fsync", "replace"])
+def test_failed_update_preserves_file_and_shared_config(tmp_path, monkeypatch, failure):
+    path = tmp_path / "config.yaml"
+    config = {"repeater": {"node_name": "before"}, "radio_type": "pymc_tcp"}
+    before = copy.deepcopy(config)
+    path.write_text(yaml.safe_dump(config))
+    original = path.read_bytes()
+    section = config["repeater"]
+    manager = ConfigManager(str(path), config)
+    live = Mock()
+    monkeypatch.setattr(manager, "live_update_daemon", live)
+
+    def fail(*args, **kwargs):
+        if failure == "dump":
+            args[1].write("partial: ")
+        assert config == before
+        raise OSError("synthetic write failure")
+
+    target = "yaml.safe_dump" if failure == "dump" else f"os.{failure}"
+    monkeypatch.setattr(f"repeater.config_manager.{target}", fail)
+    result = manager.update_and_save({"repeater": {"node_name": "after"}})
+    assert result["success"] is False
+    assert result["saved"] is False
+    assert result["live_updated"] is False
+    assert config == before
+    assert config["repeater"] is section
+    assert path.read_bytes() == original
+    assert list(tmp_path.iterdir()) == [path]
+    live.assert_not_called()
+
+
+def test_failed_direct_save_does_not_normalize_shared_config(tmp_path, monkeypatch):
+    config = {"radio_type": "pymc_tcp", "pymc_tcp": {"host": "synthetic"}}
+    before = copy.deepcopy(config)
+    path = tmp_path / "config.yaml"
+    path.write_text(yaml.safe_dump(config))
+    original = path.read_bytes()
+    monkeypatch.setattr("repeater.config_manager.yaml.safe_dump", Mock(side_effect=OSError()))
+    assert not ConfigManager(str(path), config).save_to_file()
+    assert config == before
+    assert path.read_bytes() == original
+    assert list(tmp_path.iterdir()) == [path]
+
+
+@pytest.mark.parametrize("mode", [None, 0o600, 0o640])
+def test_atomic_save_permissions_and_publication(tmp_path, monkeypatch, mode):
+    path = tmp_path / "config.yaml"
+    config = {"repeater": {"node_name": "before", "other": 1}}
+    section = config["repeater"]
+    if mode is not None:
+        path.write_text(yaml.safe_dump(config))
+        path.chmod(mode)
+    replace = os.replace
+    observed = []
+
+    def checked_replace(source, destination):
+        assert config["repeater"]["node_name"] == "before"
+        assert stat.S_IMODE(os.stat(source).st_mode) == (mode or 0o600)
+        assert os.stat(source).st_uid == os.getuid()
+        observed.append(True)
+        replace(source, destination)
+
+    monkeypatch.setattr("repeater.config_manager.os.replace", checked_replace)
+    assert ConfigManager(str(path), config).update_and_save(
+        {"repeater": {"node_name": "after"}}, live_update=False
+    )["saved"]
+    assert observed == [True]
+    assert config["repeater"] is section
+    assert yaml.safe_load(path.read_text()) == config
+    assert stat.S_IMODE(path.stat().st_mode) == (mode or 0o600)
+    assert list(tmp_path.iterdir()) == [path]
+
+
+def test_concurrent_managers_serialize_staging_and_saving(tmp_path, monkeypatch):
+    path = tmp_path / "config.yaml"
+    config = {"repeater": {"node_name": "before"}}
+    first = ConfigManager(str(path), config)
+    second = ConfigManager(str(path), config)
+    entered, release, second_started, second_dump = Event(), Event(), Event(), Event()
+    dump = yaml.safe_dump
+
+    def paused_dump(data, *args, **kwargs):
+        if data["repeater"].get("first") and not data["repeater"].get("second"):
+            entered.set()
+            assert release.wait(5)
+        else:
+            second_dump.set()
+        return dump(data, *args, **kwargs)
+
+    def second_update():
+        second_started.set()
+        return second.update_and_save({"repeater": {"second": True}}, live_update=False)
+
+    monkeypatch.setattr("repeater.config_manager.yaml.safe_dump", paused_dump)
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        a = pool.submit(first.update_and_save, {"repeater": {"first": True}}, False)
+        try:
+            assert entered.wait(5)
+            b = pool.submit(second_update)
+            assert second_started.wait(5)
+            assert not second_dump.wait(0.1)
+            assert config == {"repeater": {"node_name": "before"}}
+        finally:
+            release.set()
+        assert a.result()["saved"]
+        assert b.result()["saved"]
+    assert config["repeater"] == {"node_name": "before", "first": True, "second": True}
+    assert yaml.safe_load(path.read_text()) == config

tests/test_login_throttle_storage.py

@@ -0,0 +1,97 @@
+import pytest
+
+from repeater.web.auth_endpoints import _LoginThrottle
+
+
+@pytest.mark.parametrize("blocked", ["ip", "user", "global"])
+def test_blocked_requests_do_not_allocate_states(blocked):
+    throttle = _LoginThrottle(
+        per_ip_threshold=1 if blocked == "ip" else 99,
+        per_user_threshold=1 if blocked == "user" else 99,
+        global_threshold=1 if blocked == "global" else 99,
+        time_fn=lambda: 1000,
+    )
+    assert throttle.register_failure("ip", "user") > 0
+    for i in range(100):
+        ip = "ip" if blocked == "ip" else f"ip-{i}"
+        user = "user" if blocked == "user" else f"user-{i}"
+        assert throttle.get_retry_after(ip, user) > 0
+    assert len(throttle._ip_states) == 1
+    assert len(throttle._user_states) == 1
+
+
+def test_retained_keys_have_bounded_size_and_success_clears_them():
+    throttle = _LoginThrottle()
+    ip, user = "i" * 100000, "u" * 100000
+    assert throttle.register_failure(ip, user) == 0
+    assert max(map(len, throttle._ip_states)) <= 64
+    assert max(map(len, throttle._user_states)) <= 64
+    throttle.register_success(ip, f" {user.upper()} ")
+    assert not throttle._ip_states
+    assert not throttle._user_states
+
+
+def test_unfailed_lookups_do_not_allocate_states():
+    throttle = _LoginThrottle()
+    for i in range(100):
+        assert throttle.get_retry_after(f"ip-{i}", f"user-{i}") == 0
+    assert not throttle._ip_states
+    assert not throttle._user_states
+
+
+@pytest.mark.parametrize("start", [0, 1000])
+def test_expired_states_are_removed_without_revisiting_keys(start):
+    now = [start]
+    throttle = _LoginThrottle(time_fn=lambda: now[0])
+    throttle.register_failure("old-ip", "old-user")
+    now[0] += throttle.window_sec + 1
+    assert throttle.get_retry_after("new-ip", "new-user") == 0
+    assert not throttle._ip_states
+    assert not throttle._user_states
+    assert throttle._global_state["failures"] == 0
+
+
+def test_capacity_fails_closed_without_evicting_active_failures():
+    now = [1000]
+    throttle = _LoginThrottle(
+        per_ip_threshold=2,
+        per_user_threshold=2,
+        global_threshold=10000,
+        time_fn=lambda: now[0],
+    )
+    # A small cap exercises the same bounded-store path as the default limit.
+    throttle.max_states = 3
+    for i in range(3):
+        assert throttle.register_failure(f"ip-{i}", f"user-{i}") == 0
+    for i in range(3, 100):
+        assert throttle.get_retry_after(f"ip-{i}", f"user-{i}") > 0
+        # A failure already in flight must not bypass the capacity bound either.
+        assert throttle.register_failure(f"ip-{i}", f"user-{i}") > 0
+    assert len(throttle._ip_states) == 3
+    assert len(throttle._user_states) == 3
+    assert throttle.register_failure("ip-0", " USER-0 ") > 0
+    now[0] += throttle.window_sec + 1
+    assert throttle.get_retry_after("new-ip", "new-user") == 0
+    assert throttle.register_failure("new-ip", "new-user") == 0
+    assert len(throttle._ip_states) == 1
+    assert len(throttle._user_states) == 1
+
+
+def test_expiry_does_not_shorten_active_backoff():
+    now = [1000]
+    throttle = _LoginThrottle(
+        per_ip_threshold=1,
+        per_user_threshold=1,
+        global_threshold=1,
+        window_sec=5,
+        base_backoff_sec=20,
+        max_backoff_sec=20,
+        time_fn=lambda: now[0],
+    )
+    assert throttle.register_failure("ip", "user") == 20
+    now[0] += 6
+    assert throttle.get_retry_after("ip", "user") == 14
+    now[0] += 15
+    assert throttle.get_retry_after("ip", "user") == 0
+    assert not throttle._ip_states
+    assert not throttle._user_states

tests/test_manage_upgrade_output.py

@@ -0,0 +1,46 @@
+"""Exercise the upgrade summary without running installation/service commands."""
+
+import os
+import subprocess
+from pathlib import Path
+
+import pytest
+
+
+@pytest.mark.parametrize("service_running", [0, 1])
+def test_silent_upgrade_container_warning_renders_newlines(service_running):
+    script = (Path(__file__).resolve().parents[1] / "manage.sh").read_text()
+    start = script.index('    local container_note=""')
+    end = script.index("\n    return 0\n}", start)
+    summary = script[start:end]
+    # The nonempty container variable selects the warning without depending on
+    # the test host. Only the final output block runs, not manage.sh itself.
+    probe = (
+        "set -eu\nrender() {\n"
+        "local silent=true container=test current_version=old new_version=new\n"
+        f"local service_was_running={service_running} service_was_enabled=0\n"
+        f"{summary}\n}}\nrender\n"
+    )
+    env = os.environ.copy()
+    env.pop("BASH_ENV", None)
+    result = subprocess.run(
+        ["bash", "--noprofile", "--norc"],
+        input=probe,
+        text=True,
+        capture_output=True,
+        check=True,
+        env=env,
+    )
+    expected = "Upgrade completed successfully!\nVersion: old -> new\n"
+    if service_running:
+        expected += "✓ Service is running\n"
+    expected += (
+        "✓ Configuration preserved\n\n\n"
+        "⚠ CONTAINER DETECTED:\n"
+        "USB udev rules must be set on the HOST, not here.\n"
+        "CH341 host-side setup: "
+        "https://docs.openhop.dev/projects/openhop-repeater/hardware-setup/#ch341-usb-spi-hosts\n"
+        "=== Upgrade Complete ===\n"
+    )
+    assert result.stdout == expected
+    assert result.stderr == ""

tests/test_restart_job_submission.py

@@ -0,0 +1,69 @@
+"""The restart command must run outside the service being restarted."""
+
+import subprocess
+from pathlib import Path
+from unittest.mock import Mock
+
+import pytest
+
+from repeater import service_utils as su
+
+
+@pytest.fixture(autouse=True)
+def systemd_host(monkeypatch):
+    monkeypatch.setattr(su, "is_container", lambda: False)
+    monkeypatch.setattr(su, "is_buildroot", lambda: False)
+    monkeypatch.setattr(su.os, "geteuid", lambda: 1000)
+
+
+@pytest.mark.parametrize("root", [False, True])
+def test_restart_submits_delayed_external_job(monkeypatch, root):
+    monkeypatch.setattr(su.os, "geteuid", lambda: 0 if root else 1000)
+    run = Mock(return_value=subprocess.CompletedProcess([], 0, "", ""))
+    monkeypatch.setattr(su.subprocess, "run", run)
+    ok, message = su.restart_service()
+    assert ok and "initiated" in message
+    run.assert_called_once()
+    args = run.call_args.args[0]
+    assert "/usr/bin/systemd-run" in args
+    assert "--on-active=2s" in args
+    assert "--timer-property=AccuracySec=100ms" in args
+    assert "--timer-property=RemainAfterElapse=no" in args
+    assert "--collect" in args
+    assert args[-3:] == [su._SYSTEMCTL_BIN, "restart", "openhop-repeater"]
+    if not root:
+        assert args[:2] == [su._SUDO_BIN, "--non-interactive"]
+    else:
+        assert args[0] == "/usr/bin/systemd-run"
+
+
+@pytest.mark.parametrize("outcome", ["timeout", "interrupted", "denied"])
+def test_failed_submission_does_not_claim_success_or_retry(monkeypatch, outcome):
+    run = Mock()
+    if outcome == "timeout":
+        run.side_effect = subprocess.TimeoutExpired("systemd-run", 5)
+    else:
+        run.return_value = subprocess.CompletedProcess(
+            [], -15 if outcome == "interrupted" else 1, "", "Access denied"
+        )
+    monkeypatch.setattr(su.subprocess, "run", run)
+    ok, message = su.restart_service()
+    assert not ok
+    if outcome != "denied":
+        assert "unconfirmed" in message.lower()
+    else:
+        assert "Access denied" in message
+    run.assert_called_once()
+
+
+def test_installer_authorizes_only_fixed_restart_timer_command():
+    script = (Path(__file__).resolve().parents[1] / "manage.sh").read_text()
+    rules = [line for line in script.splitlines() if "NOPASSWD:" in line]
+    assert len(rules) == 2
+    command = (
+        "/usr/bin/systemd-run --quiet --collect --unit=openhop-repeater-restart "
+        "--on-active=2s --timer-property=AccuracySec=100ms "
+        "--timer-property=RemainAfterElapse=no /usr/bin/systemctl restart openhop-repeater"
+    )
+    assert all(command in line for line in rules)
+    assert "*" not in command

tests/test_service_utils.py

@@ -126,106 +126,24 @@ def test_restart_service_buildroot_paths(monkeypatch):
     assert "Restart failed" in msg
 
 
-def test_restart_service_systemctl_and_sudo_paths(monkeypatch):
+@pytest.mark.parametrize(
+    "error, expected",
+    [
+        (FileNotFoundError("systemd-run"), "scheduler unavailable"),
+        (RuntimeError("bad"), "Restart command failed"),
+    ],
+)
+def test_restart_scheduler_launch_errors(monkeypatch, error, expected):
     monkeypatch.setattr(su, "is_container", lambda: False)
     monkeypatch.setattr(su, "is_buildroot", lambda: False)
 
-    def _result(code=0, err=""):
-        return subprocess.CompletedProcess(args=[], returncode=code, stdout="", stderr=err)
+    def fail(*args, **kwargs):
+        raise error
 
-    # systemctl success
-    monkeypatch.setattr(su.subprocess, "run", lambda *args, **kwargs: _result(0))
-    ok, msg = su.restart_service()
-    assert ok is True
-    assert "Service restart initiated" in msg
-
-    # systemctl timeout
-    def _timeout(*_args, **_kwargs):
-        raise subprocess.TimeoutExpired(cmd="x", timeout=5)
-
-    monkeypatch.setattr(su.subprocess, "run", _timeout)
-    ok, msg = su.restart_service()
-    assert ok is True
-    assert "timeout" in msg
-
-    # systemctl missing binary
-    def _missing(*_args, **_kwargs):
-        raise FileNotFoundError("systemctl")
-
-    monkeypatch.setattr(su.subprocess, "run", _missing)
-    ok, msg = su.restart_service()
+    monkeypatch.setattr(su.subprocess, "run", fail)
+    ok, message = su.restart_service()
     assert ok is False
-    assert "systemctl not available" in msg
-
-    # systemctl denied then sudo success
-    calls = {"n": 0}
-
-    def _denied_then_sudo(*args, **kwargs):
-        calls["n"] += 1
-        if calls["n"] == 1:
-            return _result(1, "Access denied")
-        return _result(0)
-
-    monkeypatch.setattr(su.subprocess, "run", _denied_then_sudo)
-    ok, msg = su.restart_service()
-    assert ok is True
-    assert "Service restart initiated" in msg
-
-    # systemctl generic fail then sudo fail
-    calls = {"n": 0}
-
-    def _fail_then_fail(*args, **kwargs):
-        calls["n"] += 1
-        if calls["n"] == 1:
-            return _result(1, "broken")
-        return _result(2, "sudo denied")
-
-    monkeypatch.setattr(su.subprocess, "run", _fail_then_fail)
-    ok, msg = su.restart_service()
-    assert ok is False
-    assert "Restart failed" in msg
-
-    # systemctl generic fail then sudo timeout
-    calls = {"n": 0}
-
-    def _fail_then_timeout(*args, **kwargs):
-        calls["n"] += 1
-        if calls["n"] == 1:
-            return _result(1, "broken")
-        raise subprocess.TimeoutExpired(cmd="sudo", timeout=5)
-
-    monkeypatch.setattr(su.subprocess, "run", _fail_then_timeout)
-    ok, msg = su.restart_service()
-    assert ok is True
-    assert "timeout" in msg
-
-    # systemctl generic fail then sudo missing
-    calls = {"n": 0}
-
-    def _fail_then_sudo_missing(*args, **kwargs):
-        calls["n"] += 1
-        if calls["n"] == 1:
-            return _result(1, "broken")
-        raise FileNotFoundError("sudo")
-
-    monkeypatch.setattr(su.subprocess, "run", _fail_then_sudo_missing)
-    ok, msg = su.restart_service()
-    assert ok is False
-    assert "Neither polkit nor sudo" in msg
-
-    # systemctl generic fail then sudo unexpected exception
-    calls = {"n": 0}
-
-    def _fail_then_exception(*args, **kwargs):
-        calls["n"] += 1
-        if calls["n"] == 1:
-            return _result(1, "broken")
-        raise RuntimeError("bad")
-
-    monkeypatch.setattr(su.subprocess, "run", _fail_then_exception)
-    ok, msg = su.restart_service()
-    assert ok is False
-    assert "Restart command failed" in msg
+    assert expected in message
 
 
 def test_ensure_plugin_manager_service_runs_sudo_bootstrap(monkeypatch):