This repository has no description
test(xrpc): knot proxy integration suite
Lewis: May this revision serve well! <lu5a@proton.me>
+826
+826
crates/xrpc/tests/knot_proxy.rs
+826
crates/xrpc/tests/knot_proxy.rs
···
1
+
use std::sync::Arc;
2
+
use std::time::Duration;
3
+
4
+
use axum::body::{Body, to_bytes};
5
+
use bobbin_edge_index::{CoverageWatch, EdgeStore};
6
+
use bobbin_knot_proxy::{FailureThreshold, KnotProxy, KnotProxyConfig};
7
+
use bobbin_record_lru::{CacheCapacity, LruRecordStore};
8
+
use bobbin_slingshot_client::SlingshotClient;
9
+
use bobbin_xrpc::{AppState, router};
10
+
use http::{Request, StatusCode};
11
+
use serde_json::{Value, json};
12
+
use tower::ServiceExt;
13
+
use url::Url;
14
+
use url::form_urlencoded::byte_serialize;
15
+
use wiremock::matchers::{header_exists, method, path, query_param};
16
+
use wiremock::{Mock, MockServer, ResponseTemplate};
17
+
18
+
const CID: &str = "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i";
19
+
20
+
fn test_config() -> KnotProxyConfig {
21
+
KnotProxyConfig {
22
+
failure_threshold: FailureThreshold::new(2).unwrap(),
23
+
cooldown: Duration::from_millis(80),
24
+
connect_timeout: Duration::from_millis(500),
25
+
read_timeout: Duration::from_secs(2),
26
+
allow_private_hosts: true,
27
+
require_https: false,
28
+
}
29
+
}
30
+
31
+
struct Harness {
32
+
slingshot: MockServer,
33
+
knot: MockServer,
34
+
state: AppState,
35
+
}
36
+
37
+
impl Harness {
38
+
async fn new() -> Self {
39
+
Self::with_config(test_config()).await
40
+
}
41
+
42
+
async fn with_config(config: KnotProxyConfig) -> Self {
43
+
let slingshot_server = MockServer::start().await;
44
+
let knot_server = MockServer::start().await;
45
+
let state = AppState::new(
46
+
Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))),
47
+
SlingshotClient::new(Url::parse(&slingshot_server.uri()).unwrap()).unwrap(),
48
+
Arc::new(EdgeStore::new()),
49
+
Arc::new(CoverageWatch::new()),
50
+
Arc::new(KnotProxy::new(config).unwrap()),
51
+
);
52
+
Self {
53
+
slingshot: slingshot_server,
54
+
knot: knot_server,
55
+
state,
56
+
}
57
+
}
58
+
59
+
async fn mount_repo_record(&self, did: &str, rkey: &str, name: &str) {
60
+
let knot_value = self.knot.uri();
61
+
let record = json!({
62
+
"$type": "sh.tangled.repo",
63
+
"createdAt": "2026-05-01T00:00:00Z",
64
+
"knot": knot_value,
65
+
"name": name,
66
+
});
67
+
let uri = format!("at://{did}/sh.tangled.repo/{rkey}");
68
+
Mock::given(method("GET"))
69
+
.and(path("/xrpc/com.atproto.repo.getRecord"))
70
+
.and(query_param("repo", did))
71
+
.and(query_param("collection", "sh.tangled.repo"))
72
+
.and(query_param("rkey", rkey))
73
+
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
74
+
"uri": uri,
75
+
"cid": CID,
76
+
"value": record,
77
+
})))
78
+
.mount(&self.slingshot)
79
+
.await;
80
+
}
81
+
82
+
async fn call(&self, path_and_query: &str) -> http::Response<Body> {
83
+
self.call_with_headers(path_and_query, &[]).await
84
+
}
85
+
86
+
async fn call_with_headers(
87
+
&self,
88
+
path_and_query: &str,
89
+
client_headers: &[(&str, &str)],
90
+
) -> http::Response<Body> {
91
+
let builder = client_headers
92
+
.iter()
93
+
.fold(Request::builder().uri(path_and_query), |b, (k, v)| {
94
+
b.header(*k, *v)
95
+
});
96
+
router(self.state.clone())
97
+
.oneshot(builder.body(Body::empty()).unwrap())
98
+
.await
99
+
.expect("router infallible")
100
+
}
101
+
}
102
+
103
+
fn enc(s: &str) -> String {
104
+
byte_serialize(s.as_bytes()).collect()
105
+
}
106
+
107
+
async fn body_string(resp: http::Response<Body>) -> String {
108
+
let body = to_bytes(resp.into_body(), 64 * 1024).await.unwrap();
109
+
String::from_utf8(body.to_vec()).expect("response body is utf-8")
110
+
}
111
+
112
+
async fn body_value(resp: http::Response<Body>) -> Value {
113
+
let s = body_string(resp).await;
114
+
serde_json::from_str(&s).unwrap_or_else(|e| panic!("body not json: {e}: {s}"))
115
+
}
116
+
117
+
#[tokio::test]
118
+
async fn proxies_repo_blob_with_did_slash_name_repo_param() {
119
+
let h = Harness::new().await;
120
+
h.mount_repo_record("did:plc:abalone", "r1", "barnacle")
121
+
.await;
122
+
Mock::given(method("GET"))
123
+
.and(path("/xrpc/sh.tangled.repo.blob"))
124
+
.and(query_param("repo", "did:plc:abalone/barnacle"))
125
+
.and(query_param("ref", "main"))
126
+
.and(query_param("path", "README.md"))
127
+
.respond_with(
128
+
ResponseTemplate::new(200)
129
+
.set_body_raw(r#"{"path":"README.md","content":"hi"}"#, "application/json"),
130
+
)
131
+
.mount(&h.knot)
132
+
.await;
133
+
134
+
let target = format!(
135
+
"/xrpc/sh.tangled.repo.blob?repo={}&ref=main&path=README.md",
136
+
enc("at://did:plc:abalone/sh.tangled.repo/r1"),
137
+
);
138
+
let resp = h.call(&target).await;
139
+
assert_eq!(resp.status(), StatusCode::OK);
140
+
assert_eq!(
141
+
resp.headers().get("content-type").unwrap(),
142
+
"application/json",
143
+
);
144
+
let v = body_value(resp).await;
145
+
assert_eq!(v["path"], "README.md");
146
+
assert_eq!(v["content"], "hi");
147
+
}
148
+
149
+
#[tokio::test]
150
+
async fn streams_binary_archive_through_proxy() {
151
+
let h = Harness::new().await;
152
+
h.mount_repo_record("did:plc:limpet", "r2", "kelp").await;
153
+
let payload: Vec<u8> = (0u8..=255).collect();
154
+
Mock::given(method("GET"))
155
+
.and(path("/xrpc/sh.tangled.repo.archive"))
156
+
.and(query_param("repo", "did:plc:limpet/kelp"))
157
+
.and(query_param("ref", "v1"))
158
+
.respond_with(
159
+
ResponseTemplate::new(200)
160
+
.insert_header("content-type", "application/gzip")
161
+
.set_body_bytes(payload.clone()),
162
+
)
163
+
.mount(&h.knot)
164
+
.await;
165
+
166
+
let target = format!(
167
+
"/xrpc/sh.tangled.repo.archive?repo={}&ref=v1",
168
+
enc("at://did:plc:limpet/sh.tangled.repo/r2"),
169
+
);
170
+
let resp = h.call(&target).await;
171
+
assert_eq!(resp.status(), StatusCode::OK);
172
+
assert_eq!(
173
+
resp.headers().get("content-type").unwrap(),
174
+
"application/gzip",
175
+
);
176
+
let body = to_bytes(resp.into_body(), 4 * 1024).await.unwrap();
177
+
assert_eq!(body.as_ref(), payload.as_slice());
178
+
}
179
+
180
+
#[tokio::test]
181
+
async fn missing_repo_param_returns_400() {
182
+
let h = Harness::new().await;
183
+
let resp = h.call("/xrpc/sh.tangled.repo.blob?ref=main").await;
184
+
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
185
+
let v = body_value(resp).await;
186
+
assert_eq!(v["error"], "InvalidRequest");
187
+
}
188
+
189
+
#[tokio::test]
190
+
async fn unknown_repo_propagates_404_from_slingshot() {
191
+
let h = Harness::new().await;
192
+
Mock::given(method("GET"))
193
+
.and(path("/xrpc/com.atproto.repo.getRecord"))
194
+
.respond_with(ResponseTemplate::new(404).set_body_string("not found"))
195
+
.mount(&h.slingshot)
196
+
.await;
197
+
let target = format!(
198
+
"/xrpc/sh.tangled.repo.blob?repo={}&ref=main",
199
+
enc("at://did:plc:abalone/sh.tangled.repo/missing"),
200
+
);
201
+
let resp = h.call(&target).await;
202
+
assert_eq!(resp.status(), StatusCode::NOT_FOUND);
203
+
let v = body_value(resp).await;
204
+
assert_eq!(v["error"], "RecordNotFound");
205
+
}
206
+
207
+
#[tokio::test]
208
+
async fn knot_5xx_routes_to_upstream_failed() {
209
+
let h = Harness::new().await;
210
+
h.mount_repo_record("did:plc:abalone", "r1", "barnacle")
211
+
.await;
212
+
Mock::given(method("GET"))
213
+
.and(path("/xrpc/sh.tangled.repo.blob"))
214
+
.respond_with(ResponseTemplate::new(503))
215
+
.mount(&h.knot)
216
+
.await;
217
+
let target = format!(
218
+
"/xrpc/sh.tangled.repo.blob?repo={}&ref=main&path=x",
219
+
enc("at://did:plc:abalone/sh.tangled.repo/r1"),
220
+
);
221
+
let resp = h.call(&target).await;
222
+
assert_eq!(resp.status(), StatusCode::BAD_GATEWAY);
223
+
let v = body_value(resp).await;
224
+
assert_eq!(v["error"], "UpstreamFailed");
225
+
}
226
+
227
+
#[tokio::test]
228
+
async fn knot_4xx_passes_through_unchanged() {
229
+
let h = Harness::new().await;
230
+
h.mount_repo_record("did:plc:abalone", "r1", "barnacle")
231
+
.await;
232
+
Mock::given(method("GET"))
233
+
.and(path("/xrpc/sh.tangled.repo.blob"))
234
+
.respond_with(ResponseTemplate::new(404).set_body_raw(
235
+
r#"{"error":"FileNotFound","message":"nope"}"#,
236
+
"application/json",
237
+
))
238
+
.mount(&h.knot)
239
+
.await;
240
+
let target = format!(
241
+
"/xrpc/sh.tangled.repo.blob?repo={}&ref=main&path=missing",
242
+
enc("at://did:plc:abalone/sh.tangled.repo/r1"),
243
+
);
244
+
let resp = h.call(&target).await;
245
+
assert_eq!(resp.status(), StatusCode::NOT_FOUND);
246
+
let v = body_value(resp).await;
247
+
assert_eq!(v["error"], "FileNotFound");
248
+
}
249
+
250
+
#[tokio::test]
251
+
async fn breaker_opens_after_threshold_then_short_circuits() {
252
+
let h = Harness::new().await;
253
+
h.mount_repo_record("did:plc:abalone", "r1", "barnacle")
254
+
.await;
255
+
Mock::given(method("GET"))
256
+
.and(path("/xrpc/sh.tangled.repo.blob"))
257
+
.respond_with(ResponseTemplate::new(503))
258
+
.mount(&h.knot)
259
+
.await;
260
+
let target = format!(
261
+
"/xrpc/sh.tangled.repo.blob?repo={}&ref=main&path=x",
262
+
enc("at://did:plc:abalone/sh.tangled.repo/r1"),
263
+
);
264
+
let r1 = h.call(&target).await;
265
+
assert_eq!(r1.status(), StatusCode::BAD_GATEWAY);
266
+
let _ = body_string(r1).await;
267
+
let r2 = h.call(&target).await;
268
+
assert_eq!(r2.status(), StatusCode::BAD_GATEWAY);
269
+
let _ = body_string(r2).await;
270
+
let r3 = h.call(&target).await;
271
+
assert_eq!(r3.status(), StatusCode::BAD_GATEWAY);
272
+
let v = body_value(r3).await;
273
+
assert!(
274
+
v["message"]
275
+
.as_str()
276
+
.unwrap_or_default()
277
+
.contains("circuit breaker open"),
278
+
"third call must be short-circuited by breaker, got {v}",
279
+
);
280
+
}
281
+
282
+
#[tokio::test]
283
+
async fn proxy_owner_uses_knot_query_param() {
284
+
let h = Harness::new().await;
285
+
Mock::given(method("GET"))
286
+
.and(path("/xrpc/sh.tangled.owner"))
287
+
.respond_with(
288
+
ResponseTemplate::new(200)
289
+
.set_body_raw(r#"{"owner":"did:plc:nautilus"}"#, "application/json"),
290
+
)
291
+
.mount(&h.knot)
292
+
.await;
293
+
let target = format!("/xrpc/sh.tangled.owner?knot={}", enc(&h.knot.uri()));
294
+
let resp = h.call(&target).await;
295
+
assert_eq!(resp.status(), StatusCode::OK);
296
+
let v = body_value(resp).await;
297
+
assert_eq!(v["owner"], "did:plc:nautilus");
298
+
}
299
+
300
+
#[tokio::test]
301
+
async fn proxy_knot_version_uses_knot_query_param() {
302
+
let h = Harness::new().await;
303
+
Mock::given(method("GET"))
304
+
.and(path("/xrpc/sh.tangled.knot.version"))
305
+
.respond_with(
306
+
ResponseTemplate::new(200).set_body_raw(r#"{"version":"0.42"}"#, "application/json"),
307
+
)
308
+
.mount(&h.knot)
309
+
.await;
310
+
let target = format!("/xrpc/sh.tangled.knot.version?knot={}", enc(&h.knot.uri()));
311
+
let resp = h.call(&target).await;
312
+
assert_eq!(resp.status(), StatusCode::OK);
313
+
let v = body_value(resp).await;
314
+
assert_eq!(v["version"], "0.42");
315
+
}
316
+
317
+
#[tokio::test]
318
+
async fn proxy_knot_list_keys_forwards_pagination_params() {
319
+
let h = Harness::new().await;
320
+
Mock::given(method("GET"))
321
+
.and(path("/xrpc/sh.tangled.knot.listKeys"))
322
+
.and(query_param("limit", "5"))
323
+
.and(query_param("cursor", "abc"))
324
+
.respond_with(ResponseTemplate::new(200).set_body_raw(r#"{"keys":[]}"#, "application/json"))
325
+
.mount(&h.knot)
326
+
.await;
327
+
let target = format!(
328
+
"/xrpc/sh.tangled.knot.listKeys?knot={}&limit=5&cursor=abc",
329
+
enc(&h.knot.uri()),
330
+
);
331
+
let resp = h.call(&target).await;
332
+
assert_eq!(resp.status(), StatusCode::OK);
333
+
}
334
+
335
+
#[tokio::test]
336
+
async fn missing_knot_param_on_knot_route_returns_400() {
337
+
let h = Harness::new().await;
338
+
let resp = h.call("/xrpc/sh.tangled.knot.version").await;
339
+
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
340
+
let v = body_value(resp).await;
341
+
assert_eq!(v["error"], "InvalidRequest");
342
+
}
343
+
344
+
#[tokio::test]
345
+
async fn second_proxy_call_skips_slingshot_via_lru() {
346
+
let h = Harness::new().await;
347
+
h.mount_repo_record("did:plc:abalone", "r1", "barnacle")
348
+
.await;
349
+
Mock::given(method("GET"))
350
+
.and(path("/xrpc/sh.tangled.repo.tree"))
351
+
.and(query_param("repo", "did:plc:abalone/barnacle"))
352
+
.and(query_param("ref", "main"))
353
+
.respond_with(
354
+
ResponseTemplate::new(200)
355
+
.set_body_raw(r#"{"ref":"main","files":[]}"#, "application/json"),
356
+
)
357
+
.mount(&h.knot)
358
+
.await;
359
+
let target = format!(
360
+
"/xrpc/sh.tangled.repo.tree?repo={}&ref=main",
361
+
enc("at://did:plc:abalone/sh.tangled.repo/r1"),
362
+
);
363
+
let r1 = h.call(&target).await;
364
+
assert_eq!(r1.status(), StatusCode::OK);
365
+
let _ = body_string(r1).await;
366
+
let r2 = h.call(&target).await;
367
+
assert_eq!(r2.status(), StatusCode::OK);
368
+
let _ = body_string(r2).await;
369
+
let received = h.slingshot.received_requests().await.unwrap();
370
+
let getrecord = received
371
+
.iter()
372
+
.filter(|r| r.url.path() == "/xrpc/com.atproto.repo.getRecord")
373
+
.count();
374
+
assert_eq!(
375
+
getrecord, 1,
376
+
"slingshot must be hit exactly once; LRU serves the second proxy call",
377
+
);
378
+
}
379
+
380
+
#[tokio::test]
381
+
async fn does_not_inject_auth_or_atproto_proxy_headers() {
382
+
let h = Harness::new().await;
383
+
h.mount_repo_record("did:plc:abalone", "r1", "barnacle")
384
+
.await;
385
+
Mock::given(method("GET"))
386
+
.and(path("/xrpc/sh.tangled.repo.blob"))
387
+
.and(header_exists("user-agent"))
388
+
.respond_with(
389
+
ResponseTemplate::new(200).set_body_raw(r#"{"path":"x"}"#, "application/json"),
390
+
)
391
+
.mount(&h.knot)
392
+
.await;
393
+
let target = format!(
394
+
"/xrpc/sh.tangled.repo.blob?repo={}&ref=main&path=x",
395
+
enc("at://did:plc:abalone/sh.tangled.repo/r1"),
396
+
);
397
+
let resp = h.call(&target).await;
398
+
assert_eq!(resp.status(), StatusCode::OK);
399
+
let received = h.knot.received_requests().await.unwrap();
400
+
let knot_call = received
401
+
.iter()
402
+
.find(|r| r.url.path() == "/xrpc/sh.tangled.repo.blob")
403
+
.expect("knot received the proxied call");
404
+
assert!(
405
+
knot_call.headers.get("authorization").is_none(),
406
+
"bobbin must not inject auth, anonymous read by design",
407
+
);
408
+
assert!(
409
+
knot_call.headers.get("atproto-proxy").is_none(),
410
+
"bobbin is not an atproto-proxy chain",
411
+
);
412
+
assert!(
413
+
knot_call.headers.get("atproto-accept-labelers").is_none(),
414
+
"bobbin does not negotiate labelers with knots",
415
+
);
416
+
}
417
+
418
+
#[tokio::test]
419
+
async fn forwards_range_and_conditional_request_headers() {
420
+
let h = Harness::new().await;
421
+
h.mount_repo_record("did:plc:limpet", "r3", "kelp").await;
422
+
Mock::given(method("GET"))
423
+
.and(path("/xrpc/sh.tangled.repo.archive"))
424
+
.and(query_param("repo", "did:plc:limpet/kelp"))
425
+
.respond_with(
426
+
ResponseTemplate::new(206)
427
+
.insert_header("content-type", "application/octet-stream")
428
+
.insert_header("content-range", "bytes 0-99/2048")
429
+
.insert_header("accept-ranges", "bytes")
430
+
.insert_header("etag", "\"v1\"")
431
+
.set_body_bytes(vec![0u8; 100]),
432
+
)
433
+
.mount(&h.knot)
434
+
.await;
435
+
436
+
let target = format!(
437
+
"/xrpc/sh.tangled.repo.archive?repo={}&ref=v1",
438
+
enc("at://did:plc:limpet/sh.tangled.repo/r3"),
439
+
);
440
+
let resp = h
441
+
.call_with_headers(
442
+
&target,
443
+
&[
444
+
("range", "bytes=0-99"),
445
+
("if-none-match", "\"old\""),
446
+
("if-modified-since", "Wed, 01 May 2026 00:00:00 GMT"),
447
+
],
448
+
)
449
+
.await;
450
+
assert_eq!(resp.status(), StatusCode::PARTIAL_CONTENT);
451
+
assert_eq!(
452
+
resp.headers().get("content-range").unwrap(),
453
+
"bytes 0-99/2048"
454
+
);
455
+
assert_eq!(resp.headers().get("accept-ranges").unwrap(), "bytes");
456
+
assert_eq!(resp.headers().get("etag").unwrap(), "\"v1\"");
457
+
458
+
let received = h.knot.received_requests().await.unwrap();
459
+
let knot_call = received
460
+
.iter()
461
+
.find(|r| r.url.path() == "/xrpc/sh.tangled.repo.archive")
462
+
.expect("knot received the proxied call");
463
+
assert_eq!(knot_call.headers.get("range").unwrap(), "bytes=0-99");
464
+
assert_eq!(knot_call.headers.get("if-none-match").unwrap(), "\"old\"");
465
+
assert_eq!(
466
+
knot_call.headers.get("if-modified-since").unwrap(),
467
+
"Wed, 01 May 2026 00:00:00 GMT",
468
+
);
469
+
}
470
+
471
+
#[tokio::test]
472
+
async fn drops_disallowed_client_headers() {
473
+
let h = Harness::new().await;
474
+
h.mount_repo_record("did:plc:abalone", "r1", "barnacle")
475
+
.await;
476
+
Mock::given(method("GET"))
477
+
.and(path("/xrpc/sh.tangled.repo.blob"))
478
+
.respond_with(
479
+
ResponseTemplate::new(200).set_body_raw(r#"{"path":"x"}"#, "application/json"),
480
+
)
481
+
.mount(&h.knot)
482
+
.await;
483
+
let target = format!(
484
+
"/xrpc/sh.tangled.repo.blob?repo={}&path=x",
485
+
enc("at://did:plc:abalone/sh.tangled.repo/r1"),
486
+
);
487
+
let resp = h
488
+
.call_with_headers(
489
+
&target,
490
+
&[
491
+
("authorization", "Bearer secret"),
492
+
("cookie", "sid=evil"),
493
+
("x-custom", "should-not-pass"),
494
+
],
495
+
)
496
+
.await;
497
+
assert_eq!(resp.status(), StatusCode::OK);
498
+
let received = h.knot.received_requests().await.unwrap();
499
+
let knot_call = received
500
+
.iter()
501
+
.find(|r| r.url.path() == "/xrpc/sh.tangled.repo.blob")
502
+
.expect("knot received the proxied call");
503
+
assert!(knot_call.headers.get("authorization").is_none());
504
+
assert!(knot_call.headers.get("cookie").is_none());
505
+
assert!(knot_call.headers.get("x-custom").is_none());
506
+
}
507
+
508
+
#[tokio::test]
509
+
async fn rejects_client_supplied_loopback_under_strict_config() {
510
+
let strict = KnotProxyConfig {
511
+
allow_private_hosts: false,
512
+
..test_config()
513
+
};
514
+
let h = Harness::with_config(strict).await;
515
+
let resp = h
516
+
.call(&format!(
517
+
"/xrpc/sh.tangled.knot.version?knot={}",
518
+
enc("http://127.0.0.1:9"),
519
+
))
520
+
.await;
521
+
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
522
+
let v = body_value(resp).await;
523
+
assert_eq!(v["error"], "InvalidRequest");
524
+
let msg = v["message"].as_str().unwrap_or_default().to_owned();
525
+
assert!(
526
+
msg.contains("loopback") || msg.contains("blocked"),
527
+
"message should explain block reason, got {msg}",
528
+
);
529
+
}
530
+
531
+
#[tokio::test]
532
+
async fn rejects_client_supplied_link_local_metadata_endpoint() {
533
+
let strict = KnotProxyConfig {
534
+
allow_private_hosts: false,
535
+
..test_config()
536
+
};
537
+
let h = Harness::with_config(strict).await;
538
+
let resp = h
539
+
.call(&format!(
540
+
"/xrpc/sh.tangled.knot.version?knot={}",
541
+
enc("http://169.254.169.254"),
542
+
))
543
+
.await;
544
+
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
545
+
let v = body_value(resp).await;
546
+
assert_eq!(v["error"], "InvalidRequest");
547
+
}
548
+
549
+
#[tokio::test]
550
+
async fn record_with_private_knot_returns_invalid_record() {
551
+
let strict = KnotProxyConfig {
552
+
allow_private_hosts: false,
553
+
..test_config()
554
+
};
555
+
let h = Harness::with_config(strict).await;
556
+
let did = "did:plc:abalone";
557
+
let rkey = "r1";
558
+
let record = json!({
559
+
"$type": "sh.tangled.repo",
560
+
"createdAt": "2026-05-01T00:00:00Z",
561
+
"knot": "http://10.0.0.5:3000",
562
+
"name": "barnacle",
563
+
});
564
+
let uri = format!("at://{did}/sh.tangled.repo/{rkey}");
565
+
Mock::given(method("GET"))
566
+
.and(path("/xrpc/com.atproto.repo.getRecord"))
567
+
.and(query_param("repo", did))
568
+
.and(query_param("collection", "sh.tangled.repo"))
569
+
.and(query_param("rkey", rkey))
570
+
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
571
+
"uri": uri,
572
+
"cid": CID,
573
+
"value": record,
574
+
})))
575
+
.mount(&h.slingshot)
576
+
.await;
577
+
let resp = h
578
+
.call(&format!(
579
+
"/xrpc/sh.tangled.repo.blob?repo={}&path=x",
580
+
enc(&uri),
581
+
))
582
+
.await;
583
+
assert_eq!(resp.status(), StatusCode::BAD_GATEWAY);
584
+
let v = body_value(resp).await;
585
+
assert_eq!(v["error"], "InvalidRecord");
586
+
}
587
+
588
+
#[tokio::test]
589
+
async fn strips_basic_auth_from_credentialed_knot_url() {
590
+
let h = Harness::new().await;
591
+
let parsed = Url::parse(&h.knot.uri()).unwrap();
592
+
let knot_with_creds = format!(
593
+
"{}://attacker:secret@{}:{}/",
594
+
parsed.scheme(),
595
+
parsed.host_str().unwrap(),
596
+
parsed.port().unwrap(),
597
+
);
598
+
let did = "did:plc:abalone";
599
+
let rkey = "r1";
600
+
let record = json!({
601
+
"$type": "sh.tangled.repo",
602
+
"createdAt": "2026-05-01T00:00:00Z",
603
+
"knot": knot_with_creds,
604
+
"name": "barnacle",
605
+
});
606
+
let uri = format!("at://{did}/sh.tangled.repo/{rkey}");
607
+
Mock::given(method("GET"))
608
+
.and(path("/xrpc/com.atproto.repo.getRecord"))
609
+
.and(query_param("repo", did))
610
+
.and(query_param("collection", "sh.tangled.repo"))
611
+
.and(query_param("rkey", rkey))
612
+
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
613
+
"uri": uri,
614
+
"cid": CID,
615
+
"value": record,
616
+
})))
617
+
.mount(&h.slingshot)
618
+
.await;
619
+
Mock::given(method("GET"))
620
+
.and(path("/xrpc/sh.tangled.repo.blob"))
621
+
.respond_with(
622
+
ResponseTemplate::new(200).set_body_raw(r#"{"path":"x"}"#, "application/json"),
623
+
)
624
+
.mount(&h.knot)
625
+
.await;
626
+
let target = format!("/xrpc/sh.tangled.repo.blob?repo={}&path=x", enc(&uri));
627
+
let resp = h.call(&target).await;
628
+
assert_eq!(resp.status(), StatusCode::OK);
629
+
let received = h.knot.received_requests().await.unwrap();
630
+
let knot_call = received
631
+
.iter()
632
+
.find(|r| r.url.path() == "/xrpc/sh.tangled.repo.blob")
633
+
.expect("knot received the proxied call");
634
+
assert!(
635
+
knot_call.headers.get("authorization").is_none(),
636
+
"userinfo in knot field must not become an Authorization header",
637
+
);
638
+
}
639
+
640
+
#[tokio::test]
641
+
async fn knot_redirect_surfaces_as_upstream_failed() {
642
+
let h = Harness::new().await;
643
+
let secondary = MockServer::start().await;
644
+
h.mount_repo_record("did:plc:abalone", "r1", "barnacle")
645
+
.await;
646
+
Mock::given(method("GET"))
647
+
.and(path("/xrpc/sh.tangled.repo.blob"))
648
+
.respond_with(
649
+
ResponseTemplate::new(302)
650
+
.insert_header("location", &format!("{}/secret", secondary.uri())),
651
+
)
652
+
.mount(&h.knot)
653
+
.await;
654
+
Mock::given(method("GET"))
655
+
.and(path("/secret"))
656
+
.respond_with(ResponseTemplate::new(200).set_body_string("leaked"))
657
+
.mount(&secondary)
658
+
.await;
659
+
let resp = h
660
+
.call(&format!(
661
+
"/xrpc/sh.tangled.repo.blob?repo={}&path=x",
662
+
enc("at://did:plc:abalone/sh.tangled.repo/r1"),
663
+
))
664
+
.await;
665
+
assert_eq!(resp.status(), StatusCode::BAD_GATEWAY);
666
+
let v = body_value(resp).await;
667
+
assert_eq!(v["error"], "UpstreamFailed");
668
+
let received = secondary.received_requests().await.unwrap();
669
+
assert!(received.is_empty(), "redirect target must not be dialled");
670
+
}
671
+
672
+
#[tokio::test]
673
+
async fn forwards_repeated_query_params() {
674
+
let h = Harness::new().await;
675
+
h.mount_repo_record("did:plc:limpet", "r4", "kelp").await;
676
+
Mock::given(method("GET"))
677
+
.and(path("/xrpc/sh.tangled.repo.tags"))
678
+
.respond_with(ResponseTemplate::new(200).set_body_raw(r#"{"tags":[]}"#, "application/json"))
679
+
.mount(&h.knot)
680
+
.await;
681
+
let target = format!(
682
+
"/xrpc/sh.tangled.repo.tags?repo={}&filter=alpha&filter=beta",
683
+
enc("at://did:plc:limpet/sh.tangled.repo/r4"),
684
+
);
685
+
let resp = h.call(&target).await;
686
+
assert_eq!(resp.status(), StatusCode::OK);
687
+
let received = h.knot.received_requests().await.unwrap();
688
+
let knot_call = received
689
+
.iter()
690
+
.find(|r| r.url.path() == "/xrpc/sh.tangled.repo.tags")
691
+
.expect("knot received the proxied call");
692
+
let filters: Vec<String> = knot_call
693
+
.url
694
+
.query_pairs()
695
+
.filter(|(k, _)| k == "filter")
696
+
.map(|(_, v)| v.into_owned())
697
+
.collect();
698
+
assert_eq!(filters, vec!["alpha".to_owned(), "beta".to_owned()]);
699
+
}
700
+
701
+
#[tokio::test]
702
+
async fn duplicate_repo_param_rejected_as_invalid_request() {
703
+
let h = Harness::new().await;
704
+
let target = format!(
705
+
"/xrpc/sh.tangled.repo.blob?repo={}&repo={}",
706
+
enc("at://did:plc:abalone/sh.tangled.repo/r1"),
707
+
enc("at://did:plc:limpet/sh.tangled.repo/r2"),
708
+
);
709
+
let resp = h.call(&target).await;
710
+
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
711
+
let v = body_value(resp).await;
712
+
assert_eq!(v["error"], "InvalidRequest");
713
+
assert!(
714
+
v["message"]
715
+
.as_str()
716
+
.unwrap_or_default()
717
+
.contains("repo parameter must appear at most once"),
718
+
"got {v}",
719
+
);
720
+
}
721
+
722
+
#[tokio::test]
723
+
async fn duplicate_knot_param_rejected_as_invalid_request() {
724
+
let h = Harness::new().await;
725
+
let target = format!(
726
+
"/xrpc/sh.tangled.knot.version?knot={}&knot={}",
727
+
enc("https://oyster.cafe"),
728
+
enc("https://nel.pet"),
729
+
);
730
+
let resp = h.call(&target).await;
731
+
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
732
+
let v = body_value(resp).await;
733
+
assert_eq!(v["error"], "InvalidRequest");
734
+
}
735
+
736
+
#[tokio::test]
737
+
async fn rejects_client_supplied_plaintext_when_https_required() {
738
+
let strict = KnotProxyConfig {
739
+
require_https: true,
740
+
..test_config()
741
+
};
742
+
let h = Harness::with_config(strict).await;
743
+
let resp = h
744
+
.call(&format!(
745
+
"/xrpc/sh.tangled.knot.version?knot={}",
746
+
enc("http://oyster.cafe"),
747
+
))
748
+
.await;
749
+
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
750
+
let v = body_value(resp).await;
751
+
assert_eq!(v["error"], "InvalidRequest");
752
+
assert!(
753
+
v["message"]
754
+
.as_str()
755
+
.unwrap_or_default()
756
+
.contains("must be https"),
757
+
"got {v}",
758
+
);
759
+
}
760
+
761
+
#[tokio::test]
762
+
async fn record_with_plaintext_knot_returns_invalid_record_when_https_required() {
763
+
let strict = KnotProxyConfig {
764
+
require_https: true,
765
+
allow_private_hosts: true,
766
+
..test_config()
767
+
};
768
+
let h = Harness::with_config(strict).await;
769
+
let did = "did:plc:abalone";
770
+
let rkey = "r1";
771
+
let record = json!({
772
+
"$type": "sh.tangled.repo",
773
+
"createdAt": "2026-05-01T00:00:00Z",
774
+
"knot": "http://oyster.cafe",
775
+
"name": "barnacle",
776
+
});
777
+
let uri = format!("at://{did}/sh.tangled.repo/{rkey}");
778
+
Mock::given(method("GET"))
779
+
.and(path("/xrpc/com.atproto.repo.getRecord"))
780
+
.and(query_param("repo", did))
781
+
.and(query_param("collection", "sh.tangled.repo"))
782
+
.and(query_param("rkey", rkey))
783
+
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
784
+
"uri": uri,
785
+
"cid": CID,
786
+
"value": record,
787
+
})))
788
+
.mount(&h.slingshot)
789
+
.await;
790
+
let resp = h
791
+
.call(&format!(
792
+
"/xrpc/sh.tangled.repo.blob?repo={}&path=x",
793
+
enc(&uri),
794
+
))
795
+
.await;
796
+
assert_eq!(resp.status(), StatusCode::BAD_GATEWAY);
797
+
let v = body_value(resp).await;
798
+
assert_eq!(v["error"], "InvalidRecord");
799
+
assert!(
800
+
v["message"]
801
+
.as_str()
802
+
.unwrap_or_default()
803
+
.contains("requires https"),
804
+
"got {v}",
805
+
);
806
+
}
807
+
808
+
#[tokio::test]
809
+
async fn knot_not_modified_passes_through() {
810
+
let h = Harness::new().await;
811
+
h.mount_repo_record("did:plc:limpet", "r5", "kelp").await;
812
+
Mock::given(method("GET"))
813
+
.and(path("/xrpc/sh.tangled.repo.archive"))
814
+
.respond_with(ResponseTemplate::new(304).insert_header("etag", "\"v1\""))
815
+
.mount(&h.knot)
816
+
.await;
817
+
let target = format!(
818
+
"/xrpc/sh.tangled.repo.archive?repo={}&ref=v1",
819
+
enc("at://did:plc:limpet/sh.tangled.repo/r5"),
820
+
);
821
+
let resp = h
822
+
.call_with_headers(&target, &[("if-none-match", "\"v1\"")])
823
+
.await;
824
+
assert_eq!(resp.status(), StatusCode::NOT_MODIFIED);
825
+
assert_eq!(resp.headers().get("etag").unwrap(), "\"v1\"");
826
+
}