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