fix(spoolbuddy): sync SSH key over heartbeat to survive Bambuddy keypair rotation

  Bambuddy's SSH keypair under <DATA_DIR>/spoolbuddy/ssh/ regenerates whenever
  the data dir is recreated (volume remount, container recreate, fresh deploy).
  The daemon previously only fetched the pubkey at registration, so any
  rotation after a successful boot left ~/.ssh/authorized_keys pointing at
  a stale public half — every Update click then failed with "Connection
  closed by authenticating user spoolbuddy [preauth]" until the daemon was
  restarted by hand. Each prior registration also appended a fresh entry
  without pruning, accumulating stale Bambuddy-tagged keys indefinitely.

  - HeartbeatResponse now carries ssh_public_key; the heartbeat route reads
    it via the same try/except shape as the register route so a missing or
    unreadable backend key doesn't break telemetry.
  - _deploy_ssh_key() strips lines tagged bambuddy-spoolbuddy and writes
    the current key once. No-op when already in sync (no mtime churn on
    every heartbeat). User-managed entries are preserved.
  - Daemon heartbeat handler calls _deploy_ssh_key when the response
    carries a key, so rotations propagate within one heartbeat instead
    of requiring a service restart.

  Tests: 5 unit (creates-when-missing, replace-stale-pileup, preserve-user-keys,
  idempotent, swallows-write-errors) + 2 backend integration (heartbeat carries
  the key; backend key-read failure leaves ssh_public_key None but the
  heartbeat still 200s).

backend/app/api/routes/spoolbuddy.py

@@ -279,6 +279,17 @@ async def device_heartbeat(
     if was_offline:
         logger.info("SpoolBuddy device back online: %s", device.device_id)
 
+    # Include current SSH public key so the daemon can re-deploy it whenever
+    # Bambuddy's keypair rotates (data dir wiped, container recreated, etc.) —
+    # otherwise SSH updates fail until the daemon restarts.
+    ssh_public_key: str | None = None
+    try:
+        from backend.app.services.spoolbuddy_ssh import get_public_key
+
+        ssh_public_key = await get_public_key()
+    except Exception:
+        pass
+
     return HeartbeatResponse(
         pending_command=pending,
         pending_write_payload=pending_write,
@@ -287,6 +298,7 @@ async def device_heartbeat(
         calibration_factor=device.calibration_factor,
         display_brightness=device.display_brightness,
         display_blank_timeout=device.display_blank_timeout,
+        ssh_public_key=ssh_public_key,
     )
 
 

backend/app/schemas/spoolbuddy.py

@@ -74,6 +74,7 @@ class HeartbeatResponse(BaseModel):
     calibration_factor: float
     display_brightness: int = 100
     display_blank_timeout: int = 0
+    ssh_public_key: str | None = None
 
 
 # --- NFC schemas ---

backend/tests/integration/test_spoolbuddy.py

@@ -203,6 +203,54 @@ class TestDeviceEndpoints:
         assert msg["type"] == "spoolbuddy_online"
         assert msg["device_id"] == "sb-hb"
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_heartbeat_returns_ssh_public_key(self, async_client: AsyncClient, device_factory):
+        """Heartbeat response carries the current SSH public key so the daemon
+        can re-deploy it whenever Bambuddy's keypair rotates without waiting
+        for a service restart."""
+        await device_factory(device_id="sb-ssh-hb")
+
+        fake_key = "ssh-ed25519 AAAATESTKEY bambuddy-spoolbuddy"
+        with (
+            patch("backend.app.api.routes.spoolbuddy.ws_manager") as mock_ws,
+            patch(
+                "backend.app.services.spoolbuddy_ssh.get_public_key",
+                AsyncMock(return_value=fake_key),
+            ),
+        ):
+            mock_ws.broadcast = AsyncMock()
+            resp = await async_client.post(
+                f"{API}/devices/sb-ssh-hb/heartbeat",
+                json={"nfc_ok": True, "scale_ok": True, "uptime_s": 5},
+            )
+
+        assert resp.status_code == 200
+        assert resp.json()["ssh_public_key"] == fake_key
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_heartbeat_ssh_key_failure_does_not_break_heartbeat(self, async_client: AsyncClient, device_factory):
+        """If the backend can't read its own SSH key, the heartbeat must still
+        succeed — telemetry/commands are far more critical than key sync."""
+        await device_factory(device_id="sb-ssh-fail")
+
+        with (
+            patch("backend.app.api.routes.spoolbuddy.ws_manager") as mock_ws,
+            patch(
+                "backend.app.services.spoolbuddy_ssh.get_public_key",
+                AsyncMock(side_effect=OSError("disk full")),
+            ),
+        ):
+            mock_ws.broadcast = AsyncMock()
+            resp = await async_client.post(
+                f"{API}/devices/sb-ssh-fail/heartbeat",
+                json={"nfc_ok": True, "scale_ok": True, "uptime_s": 5},
+            )
+
+        assert resp.status_code == 200
+        assert resp.json()["ssh_public_key"] is None
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_heartbeat_returns_pending_command(self, async_client: AsyncClient, device_factory):

spoolbuddy/daemon/main.py

@@ -66,8 +66,18 @@ def _get_ip() -> str:
         return "unknown"
 
 
+SSH_KEY_TAG = "bambuddy-spoolbuddy"
+
+
 def _deploy_ssh_key(public_key: str) -> None:
-    """Write Bambuddy's SSH public key to authorized_keys if not already present."""
+    """Sync Bambuddy's SSH public key into authorized_keys.
+
+    Replaces any prior key tagged ``bambuddy-spoolbuddy`` so the file always
+    reflects Bambuddy's *current* keypair. Without this, every Bambuddy key
+    rotation (data dir wipe, container recreate, etc.) leaves a stale entry
+    behind and the file grows unbounded.
+    """
+    target = public_key.strip()
     home = Path.home()
     ssh_dir = home / ".ssh"
     auth_keys = ssh_dir / "authorized_keys"
@@ -75,17 +85,24 @@ def _deploy_ssh_key(public_key: str) -> None:
     try:
         ssh_dir.mkdir(mode=0o700, exist_ok=True)
 
-        # Check if key already deployed
+        existing_lines: list[str] = []
         if auth_keys.exists():
-            existing = auth_keys.read_text()
-            if public_key.strip() in existing:
-                return
+            existing_lines = auth_keys.read_text().splitlines()
+
+        kept = [line for line in existing_lines if SSH_KEY_TAG not in line]
+        new_lines = kept + [target]
 
-        # Append key
-        with auth_keys.open("a") as f:
-            f.write(public_key.strip() + "\n")
+        # Already in sync — current key present and no stale Bambuddy entries.
+        if existing_lines == new_lines:
+            return
+
+        auth_keys.write_text("\n".join(new_lines) + "\n")
         auth_keys.chmod(0o600)
-        logger.info("SSH public key deployed to %s", auth_keys)
+        removed = len(existing_lines) - len(kept)
+        if removed:
+            logger.info("SSH public key updated in %s (replaced %d stale entries)", auth_keys, removed)
+        else:
+            logger.info("SSH public key deployed to %s", auth_keys)
     except Exception as e:
         logger.warning("Failed to deploy SSH key: %s", e)
 
@@ -233,6 +250,10 @@ async def heartbeat_loop(config: Config, api: APIClient, start_time: float, shar
         )
 
         if result:
+            ssh_key = result.get("ssh_public_key")
+            if ssh_key:
+                _deploy_ssh_key(ssh_key)
+
             cmd = result.get("pending_command")
             if cmd == "tare":
                 scale = shared.get("scale")

spoolbuddy/tests/test_deploy_ssh_key.py

@@ -0,0 +1,86 @@
+"""Tests for daemon.main._deploy_ssh_key — Bambuddy key sync.
+
+Background: Bambuddy generates an ed25519 keypair under its data dir and ships
+the public half to the SpoolBuddy daemon over the registration/heartbeat
+response. The daemon writes that key into ~/.ssh/authorized_keys so Bambuddy
+can SSH in to drive remote updates. Whenever Bambuddy's keypair rotates (data
+volume wiped, container recreated, fresh deploy) the device's authorized_keys
+must drop the old entries and pick up the new one — otherwise:
+
+  1. SSH updates start failing silently with permission-denied
+  2. Stale Bambuddy-tagged keys pile up over time, eroding the security
+     boundary (any prior keypair Bambuddy held is permanently authorized).
+
+These tests pin the replace-not-append semantics of the deploy helper.
+"""
+
+from unittest.mock import patch
+
+from daemon.main import _deploy_ssh_key
+
+CURRENT_KEY = "ssh-ed25519 AAAACURRENT bambuddy-spoolbuddy"
+STALE_KEY_1 = "ssh-ed25519 AAAASTALE1 bambuddy-spoolbuddy"
+STALE_KEY_2 = "ssh-ed25519 AAAASTALE2 bambuddy-spoolbuddy"
+USER_KEY = "ssh-ed25519 AAAAUSER alice@laptop"
+
+
+class TestDeploySshKey:
+    def test_creates_authorized_keys_when_missing(self, tmp_path):
+        with patch("daemon.main.Path.home", return_value=tmp_path):
+            _deploy_ssh_key(CURRENT_KEY)
+
+        auth_keys = tmp_path / ".ssh" / "authorized_keys"
+        assert auth_keys.exists()
+        assert auth_keys.read_text().strip() == CURRENT_KEY
+        assert auth_keys.stat().st_mode & 0o777 == 0o600
+
+    def test_replaces_all_prior_bambuddy_tagged_keys(self, tmp_path):
+        """The pile-up scenario: 6+ stale keys accumulated over rotations.
+        After deploy, only the current key remains — no growth."""
+        ssh_dir = tmp_path / ".ssh"
+        ssh_dir.mkdir()
+        auth_keys = ssh_dir / "authorized_keys"
+        auth_keys.write_text(f"{STALE_KEY_1}\n{STALE_KEY_2}\n")
+
+        with patch("daemon.main.Path.home", return_value=tmp_path):
+            _deploy_ssh_key(CURRENT_KEY)
+
+        lines = auth_keys.read_text().strip().splitlines()
+        assert lines == [CURRENT_KEY]
+
+    def test_preserves_unrelated_user_keys(self, tmp_path):
+        """Only Bambuddy-tagged keys get replaced — user's own keys stay."""
+        ssh_dir = tmp_path / ".ssh"
+        ssh_dir.mkdir()
+        auth_keys = ssh_dir / "authorized_keys"
+        auth_keys.write_text(f"{USER_KEY}\n{STALE_KEY_1}\n")
+
+        with patch("daemon.main.Path.home", return_value=tmp_path):
+            _deploy_ssh_key(CURRENT_KEY)
+
+        lines = auth_keys.read_text().strip().splitlines()
+        assert USER_KEY in lines
+        assert STALE_KEY_1 not in lines
+        assert CURRENT_KEY in lines
+
+    def test_idempotent_when_already_in_sync(self, tmp_path):
+        """No-op when authorized_keys already matches the desired state —
+        avoids needless writes on every heartbeat."""
+        ssh_dir = tmp_path / ".ssh"
+        ssh_dir.mkdir()
+        auth_keys = ssh_dir / "authorized_keys"
+        auth_keys.write_text(f"{USER_KEY}\n{CURRENT_KEY}\n")
+        original_mtime = auth_keys.stat().st_mtime_ns
+
+        with patch("daemon.main.Path.home", return_value=tmp_path):
+            _deploy_ssh_key(CURRENT_KEY)
+
+        assert auth_keys.stat().st_mtime_ns == original_mtime
+
+    def test_swallows_write_errors(self, tmp_path):
+        """A failed deploy must not crash the heartbeat loop."""
+        with (
+            patch("daemon.main.Path.home", return_value=tmp_path),
+            patch("daemon.main.Path.mkdir", side_effect=PermissionError("readonly fs")),
+        ):
+            _deploy_ssh_key(CURRENT_KEY)  # should not raise