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