solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4from __future__ import annotations
5
6import json
7import logging
8from pathlib import Path
9from typing import Any
10
11import pytest
12
13from solstone.think.backup import rotation, state
14from solstone.think.backup.destination import DestinationStatus
15from solstone.think.backup.repo import ResticKeyError
16
17OLD_KEY = "A" * 64
18NEW_KEY = "B" * 64
19
20
21def _config_path(journal: Path) -> Path:
22 return journal / "config" / "journal.json"
23
24
25def _write_config(journal: Path, payload: dict[str, Any]) -> None:
26 config_path = _config_path(journal)
27 config_path.parent.mkdir(parents=True, exist_ok=True)
28 config_path.write_text(json.dumps(payload), encoding="utf-8")
29
30
31def _read_config(journal: Path) -> dict[str, Any]:
32 return json.loads(_config_path(journal).read_text(encoding="utf-8"))
33
34
35def _configured_backup() -> dict[str, Any]:
36 return {
37 "backup": {
38 "enabled": True,
39 "destination": {
40 "repository": "s3:safe-bucket/path",
41 "backend": "s3",
42 "credentials": {
43 "access_key_id": "access-key",
44 "secret_access_key": "secret-key",
45 },
46 },
47 "daily_key": "daily-secret",
48 "recovery_key": OLD_KEY,
49 "confirmed_recovery_key": True,
50 }
51 }
52
53
54def _repo_exists() -> DestinationStatus:
55 return DestinationStatus(True, True, "repo_exists", "exists")
56
57
58def test_rotation_skips_when_unconfigured(
59 tmp_path: Path,
60 monkeypatch: pytest.MonkeyPatch,
61) -> None:
62 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
63 _write_config(tmp_path, {})
64 monkeypatch.setattr(
65 rotation,
66 "ensure_restic",
67 lambda: pytest.fail("restic should not be resolved"),
68 )
69
70 result = rotation.rotate_recovery_key()
71
72 assert result == rotation.RotationResult(
73 status="skipped",
74 reason_code=None,
75 recovery_key=None,
76 recovery_key_display=None,
77 )
78
79
80def test_rotation_restic_unavailable(
81 tmp_path: Path,
82 monkeypatch: pytest.MonkeyPatch,
83) -> None:
84 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
85 _write_config(tmp_path, _configured_backup())
86 monkeypatch.setattr(
87 rotation,
88 "ensure_restic",
89 lambda: (_ for _ in ()).throw(RuntimeError("missing")),
90 )
91
92 result = rotation.rotate_recovery_key()
93
94 assert result.reason_code == "restic_unavailable"
95
96
97def test_rotation_backend_invalid(
98 tmp_path: Path,
99 monkeypatch: pytest.MonkeyPatch,
100) -> None:
101 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
102 config = _configured_backup()
103 del config["backup"]["destination"]["credentials"]["secret_access_key"]
104 _write_config(tmp_path, config)
105 monkeypatch.setattr(rotation, "ensure_restic", lambda: Path("/restic"))
106
107 result = rotation.rotate_recovery_key()
108
109 assert result.reason_code == "failed"
110
111
112def test_rotation_success_order_persists_new_key_and_scrubs_logs(
113 tmp_path: Path,
114 monkeypatch: pytest.MonkeyPatch,
115 caplog: pytest.LogCaptureFixture,
116) -> None:
117 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
118 _write_config(tmp_path, _configured_backup())
119 order: list[tuple[str, Any]] = []
120 original_set_recovery_key = state.set_recovery_key
121 original_set_recovery_key_confirmed = state.set_recovery_key_confirmed
122
123 def fake_capture(destination, *, password, restic_path, timeout=None):
124 order.append(("capture", password))
125 return "old-key-id"
126
127 def fake_add(destination, *, daily_key, recovery_key, restic_path, timeout=None):
128 order.append(("add", daily_key, recovery_key))
129
130 def fake_validate(destination, password, *, restic_path, timeout=None):
131 order.append(("verify", password))
132 return _repo_exists()
133
134 def fake_remove(destination, *, password, key_id, restic_path, timeout=None):
135 order.append(("remove", password, key_id))
136
137 def wrapped_set_recovery_key(recovery_key: str) -> None:
138 order.append(("persist-key", recovery_key))
139 original_set_recovery_key(recovery_key)
140
141 def wrapped_set_recovery_key_confirmed(confirmed: bool) -> None:
142 order.append(("persist-confirmed", confirmed))
143 original_set_recovery_key_confirmed(confirmed)
144
145 monkeypatch.setattr(rotation, "ensure_restic", lambda: Path("/restic"))
146 monkeypatch.setattr(rotation, "_capture_current_key_id", fake_capture)
147 monkeypatch.setattr(rotation, "_add_recovery_key", fake_add)
148 monkeypatch.setattr(rotation, "validate_destination", fake_validate)
149 monkeypatch.setattr(rotation, "_remove_key", fake_remove)
150 monkeypatch.setattr(rotation, "generate_recovery_key", lambda: NEW_KEY)
151 monkeypatch.setattr(rotation, "set_recovery_key", wrapped_set_recovery_key)
152 monkeypatch.setattr(
153 rotation,
154 "set_recovery_key_confirmed",
155 wrapped_set_recovery_key_confirmed,
156 )
157 caplog.set_level(logging.INFO, logger="solstone.backup.rotation")
158
159 result = rotation.rotate_recovery_key()
160
161 assert result.status == "ok"
162 assert result.reason_code is None
163 assert result.recovery_key == NEW_KEY
164 assert result.recovery_key_display == rotation.format_recovery_key_display(NEW_KEY)
165 assert order == [
166 ("capture", OLD_KEY),
167 ("add", "daily-secret", NEW_KEY),
168 ("verify", NEW_KEY),
169 ("remove", "daily-secret", "old-key-id"),
170 ("persist-key", NEW_KEY),
171 ("persist-confirmed", False),
172 ]
173 backup = _read_config(tmp_path)["backup"]
174 assert backup["daily_key"] == "daily-secret"
175 assert backup["recovery_key"] == NEW_KEY
176 assert backup["confirmed_recovery_key"] is False
177 serialized = json.dumps(result.__dict__)
178 for secret in ("daily-secret", OLD_KEY, "access-key", "secret-key"):
179 assert secret not in serialized
180 assert secret not in caplog.text
181 assert NEW_KEY not in caplog.text
182
183
184def test_rotation_capture_auth_failure_maps_reason_and_writes_nothing(
185 tmp_path: Path,
186 monkeypatch: pytest.MonkeyPatch,
187) -> None:
188 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
189 original_config = _configured_backup()
190 _write_config(tmp_path, original_config)
191 monkeypatch.setattr(rotation, "ensure_restic", lambda: Path("/restic"))
192 monkeypatch.setattr(
193 rotation,
194 "_capture_current_key_id",
195 lambda *args, **kwargs: (_ for _ in ()).throw(ResticKeyError("key list", 12)),
196 )
197
198 result = rotation.rotate_recovery_key()
199
200 assert result.reason_code == "auth_failed"
201 assert _read_config(tmp_path) == original_config
202
203
204def test_rotation_confirm_failure_aborts_before_verify_or_remove(
205 tmp_path: Path,
206 monkeypatch: pytest.MonkeyPatch,
207) -> None:
208 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
209 original_config = _configured_backup()
210 _write_config(tmp_path, original_config)
211 calls: list[str] = []
212 monkeypatch.setattr(rotation, "ensure_restic", lambda: Path("/restic"))
213 monkeypatch.setattr(
214 rotation,
215 "_capture_current_key_id",
216 lambda *args, **kwargs: calls.append("capture") or "old-key-id",
217 )
218 monkeypatch.setattr(
219 rotation,
220 "_add_recovery_key",
221 lambda *args, **kwargs: calls.append("add"),
222 )
223 monkeypatch.setattr(rotation, "generate_recovery_key", lambda: NEW_KEY)
224 monkeypatch.setattr(rotation, "confirm_recovery_key", lambda *args: False)
225 monkeypatch.setattr(
226 rotation,
227 "validate_destination",
228 lambda *args, **kwargs: pytest.fail("must not verify"),
229 )
230 monkeypatch.setattr(
231 rotation,
232 "_remove_key",
233 lambda *args, **kwargs: pytest.fail("must not remove"),
234 )
235
236 result = rotation.rotate_recovery_key()
237
238 assert result.reason_code == "failed"
239 assert calls == ["capture", "add"]
240 assert _read_config(tmp_path) == original_config
241
242
243def test_rotation_verify_failure_aborts_before_remove_and_writes_nothing(
244 tmp_path: Path,
245 monkeypatch: pytest.MonkeyPatch,
246) -> None:
247 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
248 original_config = _configured_backup()
249 _write_config(tmp_path, original_config)
250 monkeypatch.setattr(rotation, "ensure_restic", lambda: Path("/restic"))
251 monkeypatch.setattr(rotation, "_capture_current_key_id", lambda *a, **k: "old-id")
252 monkeypatch.setattr(rotation, "_add_recovery_key", lambda *a, **k: None)
253 monkeypatch.setattr(rotation, "generate_recovery_key", lambda: NEW_KEY)
254 monkeypatch.setattr(
255 rotation,
256 "validate_destination",
257 lambda *a, **k: DestinationStatus(True, True, "auth_failed", "bad"),
258 )
259 monkeypatch.setattr(
260 rotation,
261 "_remove_key",
262 lambda *args, **kwargs: pytest.fail("must not remove"),
263 )
264
265 result = rotation.rotate_recovery_key()
266
267 assert result.reason_code == "auth_failed"
268 assert _read_config(tmp_path) == original_config
269
270
271def test_rotation_remove_failure_leaves_config_unchanged(
272 tmp_path: Path,
273 monkeypatch: pytest.MonkeyPatch,
274) -> None:
275 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
276 original_config = _configured_backup()
277 _write_config(tmp_path, original_config)
278 monkeypatch.setattr(rotation, "ensure_restic", lambda: Path("/restic"))
279 monkeypatch.setattr(rotation, "_capture_current_key_id", lambda *a, **k: "old-id")
280 monkeypatch.setattr(rotation, "_add_recovery_key", lambda *a, **k: None)
281 monkeypatch.setattr(rotation, "generate_recovery_key", lambda: NEW_KEY)
282 monkeypatch.setattr(
283 rotation, "validate_destination", lambda *a, **k: _repo_exists()
284 )
285 monkeypatch.setattr(
286 rotation,
287 "_remove_key",
288 lambda *args, **kwargs: (_ for _ in ()).throw(ResticKeyError("key remove", 11)),
289 )
290
291 result = rotation.rotate_recovery_key()
292
293 assert result.reason_code == "locked"
294 assert _read_config(tmp_path) == original_config