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