mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-08-11 00:30:12 -04:00
2aabbe5d37
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).
86 lines
3.6 KiB
Python
86 lines
3.6 KiB
Python
"""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
|