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 socket
8import stat
9import threading
10import time
11import urllib.error
12import urllib.parse
13import urllib.request
14from pathlib import Path
15
16import pytest
17
18from solstone.think.importers.oura_auth import (
19 CALLBACK_URL,
20 OAUTH_SCOPES,
21 OuraAuthError,
22 OuraTokens,
23 load_oura_client_secret,
24 load_oura_tokens,
25 refresh_tokens,
26 run_owner_present_auth,
27 save_oura_tokens,
28)
29
30pytestmark = pytest.mark.xdist_group("oura_auth_loopback")
31
32
33def _make_journal(tmp_path: Path, config: dict | None = None) -> Path:
34 journal = tmp_path / "journal"
35 (journal / "config").mkdir(parents=True)
36 (journal / "config" / "journal.json").write_text(
37 json.dumps(config if config is not None else {}), encoding="utf-8"
38 )
39 return journal
40
41
42def _request_callback(url: str) -> None:
43 try:
44 urllib.request.urlopen(url, timeout=2).read()
45 except urllib.error.HTTPError:
46 pass
47
48
49def _state_from_auth_url(auth_url: str) -> str:
50 parsed = urllib.parse.urlparse(auth_url)
51 params = urllib.parse.parse_qs(parsed.query)
52 return params["state"][0]
53
54
55def test_owner_present_auth_uses_fixed_redirect_pkce_and_private_token_exchange():
56 opened_urls: list[str] = []
57 exchanges: list[dict[str, str]] = []
58
59 def browser_open(url: str) -> bool:
60 opened_urls.append(url)
61 state = _state_from_auth_url(url)
62 threading.Thread(
63 target=lambda: (
64 time.sleep(0.05),
65 _request_callback(
66 f"{CALLBACK_URL}?code=owner-code-sensitive&state={state}"
67 ),
68 ),
69 daemon=True,
70 ).start()
71 return True
72
73 def transport(
74 url: str,
75 data: dict[str, str],
76 headers: dict[str, str],
77 timeout_s: float,
78 ) -> dict[str, object]:
79 exchanges.append(data)
80 assert url.endswith("/oauth/token")
81 assert headers["Content-Type"] == "application/x-www-form-urlencoded"
82 assert data["grant_type"] == "authorization_code"
83 assert data["client_id"] == "client-public-id"
84 assert data["redirect_uri"] == CALLBACK_URL
85 assert data["code"] == "owner-code-sensitive"
86 assert data["code_verifier"]
87 assert timeout_s > 0
88 return {
89 "access_token": "access-token-sensitive",
90 "refresh_token": "refresh-token-sensitive",
91 "token_type": "Bearer",
92 "expires_in": 3600,
93 }
94
95 tokens = run_owner_present_auth(
96 client_id="client-public-id",
97 timeout_s=2,
98 http_transport=transport,
99 browser_open=browser_open,
100 )
101
102 assert tokens.access_token == "access-token-sensitive"
103 assert tokens.refresh_token == "refresh-token-sensitive"
104 assert tokens.token_type == "Bearer"
105 assert tokens.expires_at > time.time()
106 assert len(exchanges) == 1
107
108 opened = urllib.parse.urlparse(opened_urls[0])
109 params = urllib.parse.parse_qs(opened.query)
110 assert opened.scheme == "https"
111 assert params["client_id"] == ["client-public-id"]
112 assert params["redirect_uri"] == [CALLBACK_URL]
113 assert params["response_type"] == ["code"]
114 assert params["code_challenge_method"] == ["S256"]
115 assert params["state"][0]
116 assert params["code_challenge"][0]
117 # The full researched scope set is requested by default — including
118 # the blood-glucose (metabolic) scope the default grant lacks.
119 assert params["scope"] == [" ".join(OAUTH_SCOPES)]
120 requested_scopes = params["scope"][0].split()
121 assert len(requested_scopes) == 9
122 assert "metabolic" in requested_scopes
123 assert "heart_health" in requested_scopes
124 assert "email" not in requested_scopes
125 assert "personal" not in requested_scopes
126
127
128def test_wrong_state_and_wrong_path_do_not_exchange_tokens():
129 opened_urls: list[str] = []
130 exchange_called = False
131
132 def browser_open(url: str) -> bool:
133 opened_urls.append(url)
134 state = _state_from_auth_url(url)
135
136 def send_bad_callbacks() -> None:
137 time.sleep(0.05)
138 _request_callback(
139 f"http://localhost:8765/not-callback?code=x&state={state}"
140 )
141 _request_callback(f"{CALLBACK_URL}?code=x&state=wrong-state")
142
143 threading.Thread(target=send_bad_callbacks, daemon=True).start()
144 return True
145
146 def transport(
147 url: str,
148 data: dict[str, str],
149 headers: dict[str, str],
150 timeout_s: float,
151 ) -> dict[str, object]:
152 nonlocal exchange_called
153 exchange_called = True
154 return {}
155
156 with pytest.raises(OuraAuthError, match="timed out"):
157 run_owner_present_auth(
158 client_id="client-public-id",
159 timeout_s=0.3,
160 http_transport=transport,
161 browser_open=browser_open,
162 )
163
164 assert opened_urls
165 assert exchange_called is False
166
167
168def test_owner_present_auth_timeout_is_sanitized():
169 with pytest.raises(OuraAuthError) as excinfo:
170 run_owner_present_auth(
171 client_id="client-public-id",
172 timeout_s=0.05,
173 http_transport=lambda url, data, headers, timeout_s: {},
174 browser_open=lambda url: True,
175 )
176
177 message = str(excinfo.value)
178 assert "timed out" in message
179 assert "client-public-id" not in message
180 assert "access_token" not in message
181 assert "refresh_token" not in message
182
183
184def test_owner_present_auth_aborts_if_fixed_redirect_port_is_in_use():
185 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
186 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
187 sock.bind(("127.0.0.1", 8765))
188 sock.listen(1)
189 try:
190 with pytest.raises(OuraAuthError, match="port 8765 is unavailable"):
191 run_owner_present_auth(
192 client_id="client-public-id",
193 timeout_s=0.05,
194 http_transport=lambda url, data, headers, timeout_s: {},
195 browser_open=lambda url: True,
196 )
197 finally:
198 sock.close()
199
200
201def test_token_exchange_errors_do_not_expose_token_substrings():
202 def transport(
203 url: str,
204 data: dict[str, str],
205 headers: dict[str, str],
206 timeout_s: float,
207 ) -> dict[str, object]:
208 raise RuntimeError(
209 "bad token response access-token-sensitive refresh-token-sensitive"
210 )
211
212 def browser_open(url: str) -> bool:
213 state = _state_from_auth_url(url)
214 threading.Thread(
215 target=lambda: (
216 time.sleep(0.05),
217 _request_callback(
218 f"{CALLBACK_URL}?code=owner-code-sensitive&state={state}"
219 ),
220 ),
221 daemon=True,
222 ).start()
223 return True
224
225 with pytest.raises(OuraAuthError) as excinfo:
226 run_owner_present_auth(
227 client_id="client-public-id",
228 timeout_s=2,
229 http_transport=transport,
230 browser_open=browser_open,
231 )
232
233 message = str(excinfo.value)
234 assert "access-token-sensitive" not in message
235 assert "refresh-token-sensitive" not in message
236 assert "owner-code-sensitive" not in message
237 assert excinfo.value.__cause__ is None
238
239
240def test_refresh_tokens_uses_refresh_grant_without_exposing_existing_token():
241 captured: dict[str, str] = {}
242
243 def transport(
244 url: str,
245 data: dict[str, str],
246 headers: dict[str, str],
247 timeout_s: float,
248 ) -> dict[str, object]:
249 captured.update(data)
250 return {
251 "access_token": "new-access-sensitive",
252 "refresh_token": "new-refresh-sensitive",
253 "token_type": "Bearer",
254 "expires_in": 7200,
255 }
256
257 refreshed = refresh_tokens(
258 OuraTokens(
259 access_token="old-access-sensitive",
260 refresh_token="old-refresh-sensitive",
261 expires_at=1700000000.0,
262 ),
263 client_id="client-public-id",
264 http_transport=transport,
265 )
266
267 assert captured["grant_type"] == "refresh_token"
268 assert captured["refresh_token"] == "old-refresh-sensitive"
269 assert captured["client_id"] == "client-public-id"
270 assert refreshed.access_token == "new-access-sensitive"
271 assert refreshed.refresh_token == "new-refresh-sensitive"
272 assert refreshed.expires_at > time.time()
273
274
275def test_token_grants_attach_client_secret_only_when_present(tmp_path: Path):
276 # Server-side-flow apps (Jack's registration) require the secret at
277 # exchange and refresh; public PKCE clients must send requests
278 # unchanged. The secret lives in journal config (oura.client_secret)
279 # alongside the tokens — the journal is the one trusted store.
280 captured: list[dict] = []
281
282 def transport(url, data, headers, timeout_s):
283 captured.append(dict(data))
284 return {
285 "access_token": "at",
286 "refresh_token": "rt",
287 "expires_at": 4102444800.0,
288 "token_type": "Bearer",
289 }
290
291 journal = _make_journal(tmp_path, {"oura": {"client_id": "cid"}})
292 tokens = OuraTokens("old-at", "old-rt", 4102444800.0)
293 refresh_tokens(
294 tokens, client_id="cid", http_transport=transport, journal_root=journal
295 )
296 assert "client_secret" not in captured[-1]
297
298 config_path = journal / "config" / "journal.json"
299 config = json.loads(config_path.read_text(encoding="utf-8"))
300 config["oura"]["client_secret"] = "shh-42"
301 config_path.write_text(json.dumps(config), encoding="utf-8")
302 refresh_tokens(
303 tokens, client_id="cid", http_transport=transport, journal_root=journal
304 )
305 assert captured[-1]["client_secret"] == "shh-42"
306 assert captured[-1]["client_id"] == "cid"
307
308
309def test_load_oura_client_secret_reads_config_and_strips(tmp_path: Path):
310 journal = _make_journal(tmp_path, {"oura": {"client_secret": " shh-42 \n"}})
311
312 assert load_oura_client_secret(journal) == "shh-42"
313 assert load_oura_client_secret(_make_journal(tmp_path / "bare")) is None
314
315
316# ---------------------------------------------------------------------------
317# Token storage — journal config is the one trusted store (owner ruling
318# 2026-07-07): tokens live under oura.tokens.*, written only through the
319# config owner, preserving every other config key.
320# ---------------------------------------------------------------------------
321
322
323def test_save_and_load_oura_tokens_round_trip_through_journal_config(
324 tmp_path: Path,
325):
326 journal = _make_journal(
327 tmp_path,
328 {
329 "identity": {"timezone": "America/Denver"},
330 "oura": {"client_id": "cid", "client_secret": "shh-42"},
331 },
332 )
333 tokens = OuraTokens(
334 access_token="access-token-sensitive",
335 refresh_token="refresh-token-sensitive",
336 expires_at=1800000000.0,
337 token_type="bearer",
338 )
339
340 save_oura_tokens(tokens, journal)
341
342 assert load_oura_tokens(journal) == tokens
343 config_path = journal / "config" / "journal.json"
344 config = json.loads(config_path.read_text(encoding="utf-8"))
345 assert config["oura"]["tokens"] == {
346 "access_token": "access-token-sensitive",
347 "refresh_token": "refresh-token-sensitive",
348 "expires_at": 1800000000.0,
349 "token_type": "bearer",
350 }
351 # Read-modify-write preserves every other key in the file.
352 assert config["oura"]["client_id"] == "cid"
353 assert config["oura"]["client_secret"] == "shh-42"
354 assert config["identity"]["timezone"] == "America/Denver"
355 # The config owner writes the file with private permissions.
356 assert stat.S_IMODE(config_path.stat().st_mode) == 0o600
357 # Nothing token-shaped lands outside the journal config file (the
358 # config lock sidecar is the only other artifact, and stays empty).
359 others = [p for p in journal.rglob("*") if p.is_file() and p != config_path]
360 assert [p.name for p in others] == ["journal.json.lock"]
361 assert others[0].read_text(encoding="utf-8") == ""
362
363
364def test_load_oura_tokens_missing_returns_none(tmp_path: Path):
365 assert load_oura_tokens(_make_journal(tmp_path)) is None
366 assert (
367 load_oura_tokens(_make_journal(tmp_path / "with-section", {"oura": {}})) is None
368 )
369
370
371def test_load_oura_tokens_malformed_fails_loud(tmp_path: Path):
372 journal = _make_journal(
373 tmp_path, {"oura": {"tokens": {"access_token": "only-this"}}}
374 )
375
376 with pytest.raises(OuraAuthError, match="missing required fields") as excinfo:
377 load_oura_tokens(journal)
378
379 assert "only-this" not in str(excinfo.value)
380
381 wrong_shape = _make_journal(
382 tmp_path / "wrong-shape", {"oura": {"tokens": "not-an-object"}}
383 )
384 with pytest.raises(OuraAuthError, match="malformed"):
385 load_oura_tokens(wrong_shape)
386
387
388def test_save_oura_tokens_creates_section_when_absent(tmp_path: Path):
389 journal = _make_journal(tmp_path, {"identity": {"timezone": "UTC"}})
390 tokens = OuraTokens("at", "rt", 4102444800.0)
391
392 save_oura_tokens(tokens, journal)
393
394 loaded = load_oura_tokens(journal)
395 assert loaded is not None
396 assert loaded.access_token == "at"
397 assert loaded.token_type == "Bearer"