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 io
7import json
8import socket
9import ssl
10import urllib.error
11import urllib.request
12from pathlib import Path
13from typing import Any
14
15import pytest
16
17from solstone.think.backup.hosted import (
18 HostedBinding,
19 HostedCredentials,
20 HostedCredsUnavailable,
21 fetch_hosted_credentials,
22 hosted_binding_path,
23 load_hosted_binding,
24 operated_destination,
25 operated_repository,
26 save_hosted_binding,
27)
28from solstone.think.backup.hosted_provider import (
29 hosted_append_only_restic_session,
30 hosted_restic_session,
31)
32
33
34class _FakeResponse:
35 def __init__(self, body: bytes, *, status: int = 200):
36 self._body = body
37 self.status = status
38
39 def __enter__(self) -> _FakeResponse:
40 return self
41
42 def __exit__(self, *_args: object) -> None:
43 return None
44
45 def getcode(self) -> int:
46 return self.status
47
48 def read(self) -> bytes:
49 return self._body
50
51
52def _binding(
53 *, broker_token: str = "broker-token", prefix: str = "prefix"
54) -> HostedBinding:
55 return HostedBinding(
56 broker_endpoint="https://broker.example",
57 account_id="acct",
58 instance_id="inst",
59 bucket="bkt",
60 prefix=prefix,
61 broker_token=broker_token,
62 )
63
64
65def _credentials() -> HostedCredentials:
66 return HostedCredentials(
67 access_key_id="AKID",
68 secret_access_key="SAK",
69 session_token="SESS",
70 endpoint="https://acct.r2.cloudflarestorage.com/",
71 expires_at="2026-07-13T12:00:00Z",
72 )
73
74
75def _http_error(status: int, body: bytes = b"") -> urllib.error.HTTPError:
76 return urllib.error.HTTPError(
77 "https://broker.example/backup/credentials",
78 status,
79 "error",
80 {},
81 io.BytesIO(body),
82 )
83
84
85def test_binding_round_trip_private_file_mode(
86 tmp_path: Path,
87 monkeypatch: pytest.MonkeyPatch,
88) -> None:
89 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
90 binding = _binding(prefix="users/acct/inst/")
91
92 save_hosted_binding(binding)
93
94 assert load_hosted_binding() == binding
95 assert hosted_binding_path().stat().st_mode & 0o777 == 0o600
96
97
98def test_load_hosted_binding_returns_none_on_missing_or_invalid(
99 tmp_path: Path,
100 monkeypatch: pytest.MonkeyPatch,
101) -> None:
102 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
103
104 assert load_hosted_binding() is None
105
106 path = hosted_binding_path()
107 path.write_text("{", encoding="utf-8")
108 assert load_hosted_binding() is None
109
110 path.write_text(
111 json.dumps(
112 {
113 "broker_endpoint": "https://broker.example",
114 "account_id": "acct",
115 "instance_id": "inst",
116 "bucket": "bkt",
117 "prefix": " ",
118 }
119 ),
120 encoding="utf-8",
121 )
122 assert load_hosted_binding() is None
123
124
125def test_operated_repository_preserves_prefix_and_omits_credentials() -> None:
126 binding = _binding(prefix="users/acct/inst/")
127 creds = _credentials()
128
129 repo = operated_repository(binding, creds)
130
131 assert repo == "s3:https://acct.r2.cloudflarestorage.com/bkt/users/acct/inst/"
132 for secret in ("AKID", "SAK", "SESS"):
133 assert secret not in repo
134
135
136def test_operated_destination_includes_s3_session_credentials() -> None:
137 destination = operated_destination(_binding(), _credentials())
138
139 assert destination.backend == "s3"
140 assert destination.credentials == {
141 "access_key_id": "AKID",
142 "secret_access_key": "SAK",
143 "session_token": "SESS",
144 }
145
146
147def test_fetch_hosted_credentials_happy_path(
148 monkeypatch: pytest.MonkeyPatch,
149) -> None:
150 captured: dict[str, Any] = {}
151 binding = _binding(broker_token="the-token")
152
153 def fake_urlopen(
154 request: urllib.request.Request,
155 timeout: float | None = None,
156 ) -> _FakeResponse:
157 captured["request"] = request
158 captured["timeout"] = timeout
159 return _FakeResponse(
160 json.dumps(
161 {
162 "access_key_id": "AKID",
163 "secret_access_key": "SAK",
164 "session_token": "SESS",
165 "endpoint": "https://acct.r2.cloudflarestorage.com",
166 "expires_at": "2026-07-13T12:00:00Z",
167 }
168 ).encode("utf-8")
169 )
170
171 monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
172
173 creds = fetch_hosted_credentials(binding, scope="backup")
174
175 request = captured["request"]
176 assert request.get_header("Authorization") == "Bearer the-token"
177 assert json.loads((request.data or b"").decode("utf-8")) == {"scope": "backup"}
178 assert creds == HostedCredentials(
179 access_key_id="AKID",
180 secret_access_key="SAK",
181 session_token="SESS",
182 endpoint="https://acct.r2.cloudflarestorage.com",
183 expires_at="2026-07-13T12:00:00Z",
184 )
185
186
187def test_fetch_hosted_credentials_402_entitlement_inactive(
188 monkeypatch: pytest.MonkeyPatch,
189) -> None:
190 def fake_urlopen(*_args: object, **_kwargs: object) -> _FakeResponse:
191 raise _http_error(402, b"")
192
193 monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
194
195 with pytest.raises(HostedCredsUnavailable) as exc_info:
196 fetch_hosted_credentials(_binding(), scope="backup")
197
198 assert exc_info.value.reason_code == "hosted_entitlement_inactive"
199
200
201@pytest.mark.parametrize(
202 "failure",
203 [
204 _http_error(403, b'{"error":"needs_subscription"}'),
205 _FakeResponse(b'{"needs_subscription": true}'),
206 ],
207)
208def test_fetch_hosted_credentials_needs_subscription_marker(
209 monkeypatch: pytest.MonkeyPatch,
210 failure: Exception | _FakeResponse,
211) -> None:
212 def fake_urlopen(*_args: object, **_kwargs: object) -> _FakeResponse:
213 if isinstance(failure, Exception):
214 raise failure
215 return failure
216
217 monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
218
219 with pytest.raises(HostedCredsUnavailable) as exc_info:
220 fetch_hosted_credentials(_binding(), scope="backup")
221
222 assert exc_info.value.reason_code == "hosted_entitlement_inactive"
223
224
225@pytest.mark.parametrize(
226 "exc",
227 [
228 urllib.error.URLError("x"),
229 socket.timeout(),
230 ssl.SSLError("tls failed"),
231 ],
232)
233def test_fetch_hosted_credentials_network_failures(
234 monkeypatch: pytest.MonkeyPatch,
235 exc: Exception,
236) -> None:
237 def fake_urlopen(*_args: object, **_kwargs: object) -> _FakeResponse:
238 raise exc
239
240 monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
241
242 with pytest.raises(HostedCredsUnavailable) as exc_info:
243 fetch_hosted_credentials(_binding(), scope="backup")
244
245 assert exc_info.value.reason_code == "broker_unreachable"
246
247
248@pytest.mark.parametrize(
249 "failure",
250 [
251 _http_error(401, b'{"error":"unauthorized"}'),
252 _http_error(500, b"server error"),
253 _FakeResponse(b"not-json"),
254 _FakeResponse(
255 b'{"access_key_id":"AKID","secret_access_key":"SAK",'
256 b'"endpoint":"https://acct.r2.cloudflarestorage.com"}'
257 ),
258 ],
259)
260def test_fetch_hosted_credentials_broker_error(
261 monkeypatch: pytest.MonkeyPatch,
262 failure: Exception | _FakeResponse,
263) -> None:
264 def fake_urlopen(*_args: object, **_kwargs: object) -> _FakeResponse:
265 if isinstance(failure, Exception):
266 raise failure
267 return failure
268
269 monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
270
271 with pytest.raises(HostedCredsUnavailable) as exc_info:
272 fetch_hosted_credentials(_binding(), scope="backup")
273
274 assert exc_info.value.reason_code == "broker_error"
275
276
277def test_repr_redacts_secrets() -> None:
278 assert "the-token" not in repr(_binding(broker_token="the-token"))
279
280 creds = HostedCredentials(
281 access_key_id="AKID-SECRET",
282 secret_access_key="SAK-SECRET",
283 session_token="SESS-SECRET",
284 endpoint="https://acct.r2.cloudflarestorage.com",
285 expires_at="2026-07-13T12:00:00Z",
286 )
287 rendered = repr(creds)
288 for secret in ("AKID-SECRET", "SAK-SECRET", "SESS-SECRET"):
289 assert secret not in rendered
290 assert "https://acct.r2.cloudflarestorage.com" in rendered
291
292
293def test_broker_token_not_logged_on_degrade(
294 monkeypatch: pytest.MonkeyPatch,
295 caplog: pytest.LogCaptureFixture,
296) -> None:
297 def fake_urlopen(*_args: object, **_kwargs: object) -> _FakeResponse:
298 return _FakeResponse(b"not-json")
299
300 monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
301 caplog.set_level("WARNING", logger="solstone.backup.hosted")
302
303 with pytest.raises(HostedCredsUnavailable):
304 fetch_hosted_credentials(_binding(broker_token="the-token"), scope="backup")
305
306 assert "the-token" not in caplog.text
307
308
309def test_hosted_restic_session_uses_one_operation_scoped_credential() -> None:
310 initial = _credentials()
311
312 with hosted_restic_session(
313 _binding(prefix="users/acct/inst/"),
314 initial_credentials=initial,
315 ) as session:
316 assert session.destination.repository == (
317 "s3:https://acct.r2.cloudflarestorage.com/bkt/users/acct/inst/"
318 )
319 assert session.backend_env == {
320 "AWS_ACCESS_KEY_ID": "AKID",
321 "AWS_SECRET_ACCESS_KEY": "SAK",
322 "AWS_SESSION_TOKEN": "SESS",
323 }
324 assert "AWS_CONTAINER_CREDENTIALS_FULL_URI" not in session.backend_env
325 assert "AWS_CONTAINER_AUTHORIZATION_TOKEN" not in session.backend_env
326
327
328def test_append_only_session_uses_one_operation_scoped_credential() -> None:
329 initial = _credentials()
330 binding = _binding(prefix="users/acct/inst/")
331
332 with hosted_append_only_restic_session(
333 binding,
334 rclone_path=Path("/opt/solstone/rclone"),
335 initial_credentials=initial,
336 ) as session:
337 assert session.destination.repository == "rclone:spb:bkt/users/acct/inst/"
338 assert session.destination.credentials == {}
339 assert session.global_options == (
340 "-o",
341 "rclone.program=/opt/solstone/rclone",
342 "-o",
343 "rclone.args=serve restic --stdio --append-only --config /dev/null",
344 )
345 assert session.backend_env["RCLONE_CONFIG_SPB_TYPE"] == "s3"
346 assert session.backend_env["RCLONE_CONFIG_SPB_PROVIDER"] == "Cloudflare"
347 assert session.backend_env["RCLONE_CONFIG_SPB_ENV_AUTH"] == "false"
348 assert session.backend_env["RCLONE_CONFIG_SPB_ACCESS_KEY_ID"] == "AKID"
349 assert session.backend_env["RCLONE_CONFIG_SPB_SECRET_ACCESS_KEY"] == "SAK"
350 assert session.backend_env["RCLONE_CONFIG_SPB_SESSION_TOKEN"] == "SESS"
351 assert session.backend_env["RCLONE_CONFIG_SPB_ENDPOINT"] == initial.endpoint
352 for name in (
353 "AWS_ACCESS_KEY_ID",
354 "AWS_SECRET_ACCESS_KEY",
355 "AWS_SESSION_TOKEN",
356 "AWS_CONTAINER_CREDENTIALS_FULL_URI",
357 "AWS_CONTAINER_AUTHORIZATION_TOKEN",
358 ):
359 assert name not in session.backend_env