solpbc.org
personal memory agent
1from __future__ import annotations
2
3import subprocess
4from pathlib import Path
5
6import pytest
7
8from scripts.release_candidate_driver import DriverError
9from scripts.transparency_transport import (
10 CurlResult,
11 CurlTransparencyTransport,
12 DirectoryTransparencyTransport,
13 parse_curl_write_out,
14)
15
16
17def test_directory_transport_create_only_conflict_returns_412_exit_zero(
18 tmp_path: Path,
19) -> None:
20 transport = DirectoryTransparencyTransport(tmp_path)
21 first = transport.put_object(
22 "releases/solstone-journal/v/0.0.1/ledger-entry.json",
23 b"first",
24 content_type="application/json",
25 cache_control="immutable",
26 if_none_match=True,
27 )
28 second = transport.put_object(
29 "releases/solstone-journal/v/0.0.1/ledger-entry.json",
30 b"second",
31 content_type="application/json",
32 cache_control="immutable",
33 if_none_match=True,
34 )
35 assert first.status == 200
36 assert second.status == 412
37 assert second.exit_code == 0
38 assert second.body == b"precondition failed"
39 assert (
40 transport.get_object("releases/solstone-journal/v/0.0.1/ledger-entry.json").body
41 == b"first"
42 )
43
44
45def test_directory_transport_if_match_uses_captured_etag(tmp_path: Path) -> None:
46 transport = DirectoryTransparencyTransport(tmp_path)
47 key = "releases/solstone-journal/latest.json"
48 first = transport.put_object(
49 key,
50 b"old",
51 content_type="application/json",
52 cache_control="no-cache",
53 )
54 rejected = transport.put_object(
55 key,
56 b"new",
57 content_type="application/json",
58 cache_control="no-cache",
59 if_match='"wrong"',
60 )
61 accepted = transport.put_object(
62 key,
63 b"new",
64 content_type="application/json",
65 cache_control="no-cache",
66 if_match=first.etag,
67 )
68 assert rejected.status == 412
69 assert rejected.exit_code == 0
70 assert accepted.status == 200
71 assert transport.get_object(key).body == b"new"
72
73
74def test_directory_transport_forbidden_returns_403_with_body_exit_zero(
75 tmp_path: Path,
76) -> None:
77 transport = DirectoryTransparencyTransport(tmp_path)
78 key = "releases/solstone-journal/latest.json"
79 transport.forbidden.add(key)
80 result = transport.put_object(
81 key,
82 b"body",
83 content_type="application/json",
84 cache_control="no-cache",
85 )
86 assert result.status == 403
87 assert result.body == b"forbidden"
88 assert result.exit_code == 0
89
90
91def test_directory_transport_public_get_reads_same_object_store(tmp_path: Path) -> None:
92 transport = DirectoryTransparencyTransport(tmp_path)
93 key = "releases/solstone-journal/v/0.0.1/ledger-entry.json"
94 transport.put_object(
95 key,
96 b"entry",
97 content_type="application/json",
98 cache_control="immutable",
99 )
100 public = transport.get_public(key, cache_bypass=True)
101 assert public.status == 200
102 assert public.body == b"entry"
103
104
105def test_directory_transport_records_exact_destination_set(tmp_path: Path) -> None:
106 transport = DirectoryTransparencyTransport(
107 tmp_path,
108 endpoint="https://r2.example.invalid",
109 bucket="transparency-test",
110 public_base_url="https://transparency.solstone.app",
111 )
112 key = "releases/solstone-journal/v/0.0.1/ledger-entry.json"
113 transport.put_object(
114 key,
115 b"entry",
116 content_type="application/json",
117 cache_control="immutable",
118 )
119 transport.get_public(key)
120 assert transport.destination_set == {
121 "https://r2.example.invalid/transparency-test",
122 "https://transparency.solstone.app",
123 }
124 assert [
125 {"plane": call["plane"], "op": call["op"], "key": call["key"]}
126 for call in transport.call_log
127 ] == [
128 {"plane": "s3", "op": "PUT", "key": key},
129 {"plane": "public", "op": "GET", "key": key},
130 ]
131
132
133def test_curl_check_rejects_too_old_version(
134 monkeypatch: pytest.MonkeyPatch,
135) -> None:
136 def run(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]:
137 return subprocess.CompletedProcess(
138 args=["curl", "--version"],
139 returncode=0,
140 stdout="curl 7.83.1 (x86_64-redhat-linux-gnu)\n",
141 stderr="",
142 )
143
144 monkeypatch.setattr("scripts.transparency_transport.subprocess.run", run)
145 transport = CurlTransparencyTransport(
146 endpoint="https://r2.example.invalid",
147 bucket="transparency-test",
148 access_key_id="key",
149 secret_access_key="secret",
150 base_url="https://transparency.solstone.app",
151 )
152
153 with pytest.raises(DriverError) as error:
154 transport.check()
155
156 failure = error.value.failures[0]
157 assert failure.error == "transparency curl version is too old"
158 assert failure.expected == "curl >= 7.84"
159 assert failure.actual == "curl 7.83.1"
160
161
162def test_curl_check_missing_binary_fails_as_driver_error(
163 monkeypatch: pytest.MonkeyPatch,
164) -> None:
165 def run(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]:
166 raise FileNotFoundError("curl")
167
168 monkeypatch.setattr("scripts.transparency_transport.subprocess.run", run)
169 transport = CurlTransparencyTransport(
170 endpoint="https://r2.example.invalid",
171 bucket="transparency-test",
172 access_key_id="key",
173 secret_access_key="secret",
174 base_url="https://transparency.solstone.app",
175 )
176
177 with pytest.raises(DriverError) as error:
178 transport.check()
179
180 assert error.value.failures[0].error == "transparency curl preflight failed"
181
182
183def test_curl_check_accepts_new_enough_version(
184 monkeypatch: pytest.MonkeyPatch,
185) -> None:
186 def run(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]:
187 return subprocess.CompletedProcess(
188 args=["curl", "--version"],
189 returncode=0,
190 stdout="curl 8.15.0 (x86_64-redhat-linux-gnu)\n",
191 stderr="",
192 )
193
194 monkeypatch.setattr("scripts.transparency_transport.subprocess.run", run)
195 transport = CurlTransparencyTransport(
196 endpoint="https://r2.example.invalid",
197 bucket="transparency-test",
198 access_key_id="key",
199 secret_access_key="secret",
200 base_url="https://transparency.solstone.app",
201 )
202
203 transport.check()
204
205
206def test_curl_write_out_parser_extracts_status_and_etag() -> None:
207 assert parse_curl_write_out('200\t"abc123"\n') == (200, '"abc123"')
208 assert parse_curl_write_out("204\t\n") == (204, None)
209
210
211def test_curl_run_preserves_exit_code_without_http_status(
212 tmp_path: Path,
213 monkeypatch: pytest.MonkeyPatch,
214) -> None:
215 def run(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]:
216 return subprocess.CompletedProcess(
217 args=["curl"],
218 returncode=7,
219 stdout="000\t\n",
220 stderr="curl: failed to connect",
221 )
222
223 monkeypatch.setattr("scripts.transparency_transport.subprocess.run", run)
224 transport = CurlTransparencyTransport(
225 endpoint="https://r2.example.invalid",
226 bucket="transparency-test",
227 access_key_id="key",
228 secret_access_key="secret",
229 base_url="https://transparency.solstone.app",
230 )
231 result = transport._run_curl([], body_output=tmp_path / "body")
232 assert result.status == 0
233 assert result.exit_code == 7
234 assert result.body == b"curl: failed to connect"
235
236
237def test_curl_transport_repr_does_not_expose_secret() -> None:
238 transport = CurlTransparencyTransport(
239 endpoint="https://r2.example.invalid",
240 bucket="transparency-test",
241 access_key_id="key",
242 secret_access_key="SECRET",
243 base_url="https://transparency.solstone.app",
244 )
245 assert "SECRET" not in repr(transport)
246
247
248def test_curl_list_prefix_parses_namespaced_s3_xml(
249 monkeypatch: pytest.MonkeyPatch,
250) -> None:
251 body = b"""<?xml version="1.0" encoding="UTF-8"?>
252<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
253 <Name>transparency-test</Name>
254 <Prefix>releases/solstone-journal/v/</Prefix>
255 <IsTruncated>false</IsTruncated>
256 <Contents><Key>releases/solstone-journal/v/0.0.1/ledger-entry.json</Key></Contents>
257 <Contents><Key>releases/solstone-journal/v/0.0.1/ledger-entry.json.minisig</Key></Contents>
258</ListBucketResult>
259"""
260
261 transport = CurlTransparencyTransport(
262 endpoint="https://r2.example.invalid",
263 bucket="transparency-test",
264 access_key_id="key",
265 secret_access_key="secret",
266 base_url="https://transparency.solstone.app",
267 )
268
269 def run(
270 _self: CurlTransparencyTransport,
271 _args: object,
272 *,
273 body_output: Path,
274 ) -> CurlResult:
275 return CurlResult(status=200, body=body, etag=None, exit_code=0)
276
277 monkeypatch.setattr(CurlTransparencyTransport, "_run_curl", run)
278 result = transport.list_prefix("releases/solstone-journal/v/")
279 assert result.status == 200
280 assert result.keys == (
281 "releases/solstone-journal/v/0.0.1/ledger-entry.json",
282 "releases/solstone-journal/v/0.0.1/ledger-entry.json.minisig",
283 )
284
285
286def test_curl_list_prefix_paginates_list_objects_v2(
287 monkeypatch: pytest.MonkeyPatch,
288) -> None:
289 pages = [
290 b"""<?xml version="1.0" encoding="UTF-8"?>
291<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
292 <IsTruncated>true</IsTruncated>
293 <NextContinuationToken>page-2</NextContinuationToken>
294 <Contents><Key>releases/solstone-journal/v/0.0.1/ledger-entry.json</Key></Contents>
295</ListBucketResult>
296""",
297 b"""<?xml version="1.0" encoding="UTF-8"?>
298<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
299 <IsTruncated>false</IsTruncated>
300 <Contents><Key>releases/solstone-journal/v/0.0.2/ledger-entry.json</Key></Contents>
301</ListBucketResult>
302""",
303 ]
304 calls: list[list[str]] = []
305 transport = CurlTransparencyTransport(
306 endpoint="https://r2.example.invalid",
307 bucket="transparency-test",
308 access_key_id="key",
309 secret_access_key="secret",
310 base_url="https://transparency.solstone.app",
311 )
312
313 def run(
314 _self: CurlTransparencyTransport,
315 args: list[str],
316 *,
317 body_output: Path,
318 ) -> CurlResult:
319 calls.append(args)
320 return CurlResult(
321 status=200,
322 body=pages[len(calls) - 1],
323 etag=None,
324 exit_code=0,
325 )
326
327 monkeypatch.setattr(CurlTransparencyTransport, "_run_curl", run)
328 result = transport.list_prefix("releases/solstone-journal/v/")
329 assert result.status == 200
330 assert result.keys == (
331 "releases/solstone-journal/v/0.0.1/ledger-entry.json",
332 "releases/solstone-journal/v/0.0.2/ledger-entry.json",
333 )
334 assert all("continuation-token=page-2" not in arg for arg in calls[0])
335 assert "continuation-token=page-2" in calls[1]
336
337
338def test_curl_list_prefix_fails_closed_on_truncated_xml_without_token(
339 monkeypatch: pytest.MonkeyPatch,
340) -> None:
341 body = b"""<?xml version="1.0" encoding="UTF-8"?>
342<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
343 <IsTruncated>true</IsTruncated>
344 <Contents><Key>releases/solstone-journal/v/0.0.1/ledger-entry.json</Key></Contents>
345</ListBucketResult>
346"""
347 transport = CurlTransparencyTransport(
348 endpoint="https://r2.example.invalid",
349 bucket="transparency-test",
350 access_key_id="key",
351 secret_access_key="secret",
352 base_url="https://transparency.solstone.app",
353 )
354
355 def run(
356 _self: CurlTransparencyTransport,
357 _args: object,
358 *,
359 body_output: Path,
360 ) -> CurlResult:
361 return CurlResult(status=200, body=body, etag=None, exit_code=0)
362
363 monkeypatch.setattr(CurlTransparencyTransport, "_run_curl", run)
364 result = transport.list_prefix("releases/solstone-journal/v/")
365 assert result.status == 0
366 assert result.keys == ()
367 assert (
368 result.body
369 == b"S3 ListObjectsV2 response is truncated without NextContinuationToken"
370 )