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