alice.pds.demo.boltless.dev
This repository has no description
bobbin/xrpc: forward client address that knot can rate-limit on
Lewis: May this revision serve well! <did:plc:3fwecdnvtcscjnrx2p4n7alz>
author
did:plc:3fwecdnvtcscjnrx2p4n7a…
date
(Jul 31, 2026, 3:48 PM +0300)
commit
f94aa020
f94aa020c1dbcffbc78219c148b157339fee9c1e
parent
5422d7ca
5422d7ca3925bba8174c00fec30076c02fba5a3d
change-id
wzktkovy
wzktkovyqmmopkpvnuuzlqptpuovqsqo
did:plc:3fwecdnvtcscjnrx2p4n7a…
date
(Jul 31, 2026, 3:48 PM +0300)
commit
f94aa020
f94aa020c1dbcffbc78219c148b157339fee9c1e
parent
5422d7ca
5422d7ca3925bba8174c00fec30076c02fba5a3d
change-id
wzktkovy
wzktkovyqmmopkpvnuuzlqptpuovqsqo
+383
-36
+2
Cargo.lock
+2
Cargo.lock
+1
bobbin/crates/bobbin/Cargo.toml
+1
bobbin/crates/bobbin/Cargo.toml
+63
-2
bobbin/crates/bobbin/src/config.rs
+63
-2
bobbin/crates/bobbin/src/config.rs
···
5
5
6
6
use anyhow::{Context, anyhow};
7
7
use confique::Config;
8
+
use trusted_proxies::{ProxyNetError, TrustedProxies};
8
9
use url::Url;
9
10
10
11
const SYSTEM_CONFIG_PATH: &str = "/etc/bobbin/config.toml";
···
14
15
"server.binds",
15
16
"server.shutdown_grace_secs",
16
17
"server.debug_bind",
18
+
"server.trusted_proxies",
17
19
"hydrant.url",
18
20
"hydrant.start_cursor",
19
21
"ingest.parallelism",
···
36
38
"BOBBIN_BIND",
37
39
"BOBBIN_SHUTDOWN_GRACE_SECS",
38
40
"BOBBIN_DEBUG_BIND",
41
+
"BOBBIN_TRUSTED_PROXIES",
39
42
"BOBBIN_HYDRANT_URL",
40
43
"BOBBIN_START_CURSOR",
41
44
"BOBBIN_INGEST_PARALLELISM",
···
103
106
/// never reachable on the public listener. Bind to loopback only.
104
107
#[config(env = "BOBBIN_DEBUG_BIND", default = "")]
105
108
pub debug_bind: String,
109
+
110
+
/// Reverse proxies in front of bobbin,
111
+
/// each a bare IP address without a port or a CIDR block such as `173.245.48.0/20`.
112
+
/// Bobbin will read the client address out of `x-forwarded-for`
113
+
/// and forward that one address to the knot
114
+
/// when a request arrives from a proxy on this list,
115
+
/// so a knot that lists bobbin under its own `xrpc.trusted_proxies`
116
+
/// can rate-limit per browser
117
+
/// instead of pooling everyone bobbin serves into a single bucket.
118
+
/// Bobbin will read the last 32 entries of the chain, at most.
119
+
/// Leave empty when bobbin takes connections directly,
120
+
/// since bobbin would otherwise believe a header any client can write.
121
+
/// When using as an env var, comma-separated.
122
+
#[config(
123
+
env = "BOBBIN_TRUSTED_PROXIES",
124
+
parse_env = trusted_proxies::comma_separated,
125
+
default = []
126
+
)]
127
+
pub trusted_proxies: Vec<String>,
128
+
}
129
+
130
+
impl ServerConfig {
131
+
pub fn trusted_proxies(&self) -> Result<TrustedProxies, ProxyNetError> {
132
+
TrustedProxies::parse(self.trusted_proxies.iter().map(String::as_str))
133
+
}
106
134
}
107
135
108
136
#[derive(Debug, thiserror::Error)]
···
269
297
if let Some(p) = path {
270
298
builder = builder.file(p);
271
299
}
272
-
builder
300
+
let config = builder
273
301
.file(SYSTEM_CONFIG_PATH)
274
302
.load()
275
-
.context("load configuration")
303
+
.context("load configuration")?;
304
+
config
305
+
.server
306
+
.trusted_proxies()
307
+
.context("server.trusted_proxies takes a bare IP address or a CIDR block")?;
308
+
Ok(config)
276
309
}
277
310
278
311
pub fn template() -> String {
···
451
484
"KNOWN_ENVS entry {name:?} must start with {ENV_PREFIX:?}"
452
485
);
453
486
});
487
+
}
488
+
489
+
#[test]
490
+
fn known_envs_matches_every_env_attribute_this_crate_declares() {
491
+
let known: HashSet<&str> = KNOWN_ENVS.iter().copied().collect();
492
+
let declared: HashSet<&str> = [include_str!("config.rs"), include_str!("main.rs")]
493
+
.into_iter()
494
+
.flat_map(|source| {
495
+
source
496
+
.split("env = \"")
497
+
.skip(1)
498
+
.filter_map(|rest| rest.split('"').next())
499
+
})
500
+
.collect();
501
+
assert!(
502
+
declared.contains("BOBBIN_BIND") && declared.contains("BOBBIN_CONFIG"),
503
+
"the scan stopped matching config.rs or main.rs and every name in it would pass unchecked, since it came back with {declared:?}"
504
+
);
505
+
let missing: Vec<&&str> = declared.difference(&known).collect();
506
+
assert!(
507
+
missing.is_empty(),
508
+
"confique reads {missing:?} but check_envs will refuse to start with them set. Add them to KNOWN_ENVS"
509
+
);
510
+
let stale: Vec<&&str> = known.difference(&declared).collect();
511
+
assert!(
512
+
stale.is_empty(),
513
+
"KNOWN_ENVS lists {stale:?}, which the fields stopped reading. Drop them, or check_envs will keep accepting a name that stopped meaning anything"
514
+
);
454
515
}
455
516
456
517
#[test]
+12
-4
bobbin/crates/bobbin/src/main.rs
+12
-4
bobbin/crates/bobbin/src/main.rs
···
300
300
format!("invalid server.debug_bind `{}`", cfg.server.debug_bind)
301
301
})?)
302
302
};
303
+
let trusted_proxies = cfg
304
+
.server
305
+
.trusted_proxies()
306
+
.context("server.trusted_proxies takes a bare IP address or a CIDR block")?;
303
307
let mem_probe = debug_bind.is_some().then(|| mem::MemProbe {
304
308
edges: edges.clone(),
305
309
search: search.clone(),
···
318
322
search as Arc<dyn SearchReader>,
319
323
resolver,
320
324
)
321
-
.with_limiter(limiter);
325
+
.with_limiter(limiter)
326
+
.with_proxies(trusted_proxies);
322
327
let app = router(state);
323
328
324
329
let _debug_server = match (debug_bind, mem_probe) {
···
428
433
let app = app.clone();
429
434
let cancel = cancel.clone();
430
435
async move {
431
-
axum::serve(listener, app)
432
-
.with_graceful_shutdown(async move { cancel.cancelled().await })
433
-
.await
436
+
axum::serve(
437
+
listener,
438
+
app.into_make_service_with_connect_info::<SocketAddr>(),
439
+
)
440
+
.with_graceful_shutdown(async move { cancel.cancelled().await })
441
+
.await
434
442
}
435
443
});
436
444
+2
bobbin/crates/xrpc/Cargo.toml
+2
bobbin/crates/xrpc/Cargo.toml
···
25
25
thiserror = { workspace = true }
26
26
tower-http = { workspace = true, features = ["trace"] }
27
27
tracing = { workspace = true }
28
+
trusted-proxies = { workspace = true }
28
29
url = { workspace = true }
29
30
30
31
[dev-dependencies]
···
32
33
http = { workspace = true }
33
34
tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
34
35
tower = { workspace = true }
36
+
trusted-proxies = { workspace = true }
35
37
url = { workspace = true }
36
38
wiremock = { workspace = true }
+150
bobbin/crates/xrpc/src/client_address.rs
+150
bobbin/crates/xrpc/src/client_address.rs
···
1
+
use std::convert::Infallible;
2
+
use std::net::{IpAddr, SocketAddr};
3
+
use std::sync::OnceLock;
4
+
5
+
use axum::extract::{ConnectInfo, FromRequestParts};
6
+
use axum::http::request::Parts;
7
+
use axum::http::{HeaderMap, HeaderName, HeaderValue};
8
+
use trusted_proxies::TrustedProxies;
9
+
10
+
pub(crate) static X_FORWARDED_FOR: HeaderName = HeaderName::from_static("x-forwarded-for");
11
+
12
+
#[derive(Default)]
13
+
pub struct ClientAddress {
14
+
proxies: TrustedProxies,
15
+
ignored_header: OnceLock<IpAddr>,
16
+
no_socket: OnceLock<()>,
17
+
}
18
+
19
+
impl ClientAddress {
20
+
pub fn new(proxies: TrustedProxies) -> Self {
21
+
Self {
22
+
proxies,
23
+
..Self::default()
24
+
}
25
+
}
26
+
27
+
pub(crate) fn of(&self, headers: &HeaderMap, socket: SocketPeer) -> Option<HeaderValue> {
28
+
match socket.0 {
29
+
None => {
30
+
if self.no_socket.set(()).is_ok() {
31
+
tracing::warn!(
32
+
"bobbin won't forward a client address to the knot for this request, and every client will share one rate-limit bucket there, because bobbin doesn't have a socket address for it. Serve the listener with `into_make_service_with_connect_info`. This warning reports the first such request only."
33
+
);
34
+
}
35
+
None
36
+
}
37
+
Some(peer) => {
38
+
let relays = self.proxies.contains(peer);
39
+
if headers.contains_key(&X_FORWARDED_FOR)
40
+
&& !relays
41
+
&& self.ignored_header.set(peer).is_ok()
42
+
{
43
+
tracing::warn!(
44
+
%peer,
45
+
"bobbin ignored x-forwarded-for and will forward the address this peer connected from, because the peer is outside server.trusted_proxies. Add this address to server.trusted_proxies if it's the reverse proxy, or every client it serves will share one rate-limit bucket on each knot. This warning reports the first such peer only."
46
+
);
47
+
}
48
+
let client = relays
49
+
.then(|| {
50
+
self.proxies.rightmost_untrusted(
51
+
headers
52
+
.get_all(&X_FORWARDED_FOR)
53
+
.iter()
54
+
.filter_map(|value| value.to_str().ok())
55
+
.flat_map(|value| value.split(',')),
56
+
)
57
+
})
58
+
.flatten()
59
+
.unwrap_or_else(|| peer.to_canonical());
60
+
HeaderValue::try_from(client.to_string()).ok()
61
+
}
62
+
}
63
+
}
64
+
}
65
+
66
+
#[derive(Debug, Clone, Copy)]
67
+
pub struct SocketPeer(Option<IpAddr>);
68
+
69
+
impl<S: Send + Sync> FromRequestParts<S> for SocketPeer {
70
+
type Rejection = Infallible;
71
+
72
+
async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
73
+
Ok(Self(
74
+
parts
75
+
.extensions
76
+
.get::<ConnectInfo<SocketAddr>>()
77
+
.map(|info| info.0.ip()),
78
+
))
79
+
}
80
+
}
81
+
82
+
#[cfg(test)]
83
+
mod tests {
84
+
use super::*;
85
+
86
+
fn ip(value: &str) -> IpAddr {
87
+
value.parse().unwrap()
88
+
}
89
+
90
+
fn relaying<'a>(entries: impl IntoIterator<Item = &'a str>) -> ClientAddress {
91
+
ClientAddress::new(TrustedProxies::parse(entries).unwrap())
92
+
}
93
+
94
+
fn chain(value: Option<&str>) -> HeaderMap {
95
+
value
96
+
.map(|value| {
97
+
let mut map = HeaderMap::new();
98
+
map.insert(&X_FORWARDED_FOR, value.parse().unwrap());
99
+
map
100
+
})
101
+
.unwrap_or_default()
102
+
}
103
+
104
+
fn forwarded(proxies: &[&str], socket: Option<&str>, claimed: Option<&str>) -> Option<String> {
105
+
relaying(proxies.iter().copied())
106
+
.of(&chain(claimed), SocketPeer(socket.map(ip)))
107
+
.map(|value| value.to_str().unwrap().to_owned())
108
+
}
109
+
110
+
#[test]
111
+
fn bobbin_forwards_the_socket_unless_a_listed_proxy_relayed_the_request() {
112
+
let listed: &[&str] = &["127.0.0.1", "173.245.48.0/20"];
113
+
[
114
+
(&["127.0.0.1"][..], Some("203.0.113.7"), Some("198.51.100.4"), Some("203.0.113.7"),
115
+
"bobbin must answer for the socket, since a client reaching it directly wrote that header itself"),
116
+
(&[], Some("203.0.113.7"), Some("198.51.100.4"), Some("203.0.113.7"),
117
+
"an operator who hasn't configured a proxy will get the socket, since honoring the header by default would hand every client its own rate-limit bucket on every knot downstream. The knot reads its own empty list the opposite way, as trusting every peer, so don't carry either default across"),
118
+
(listed, Some("127.0.0.1"), Some("198.51.100.4"), Some("198.51.100.4"),
119
+
"a listed proxy hands over the address it recorded"),
120
+
(listed, Some("127.0.0.1"), Some("198.51.100.4, 173.245.48.9"), Some("198.51.100.4"),
121
+
"and a second listed hop is stepped over with it"),
122
+
(listed, Some("127.0.0.1"), Some(" 198.51.100.4 "), Some("198.51.100.4"),
123
+
"padding around an entry won't hide it"),
124
+
(listed, Some("127.0.0.1"), Some("203.0.113.7, 198.51.100.4"), Some("198.51.100.4"),
125
+
"bobbin takes the rightmost unlisted hop"),
126
+
(listed, Some("127.0.0.1"), None, Some("127.0.0.1"),
127
+
"a listed proxy that didn't send the header leaves its own socket to forward"),
128
+
(listed, Some("127.0.0.1"), Some("not-an-ip"), Some("127.0.0.1"),
129
+
"bobbin stops at an entry it can't parse and forwards the socket, because a client can write anything left of the proxy"),
130
+
(listed, Some("127.0.0.1"), Some("127.0.0.1"), Some("127.0.0.1"),
131
+
"a chain of listed hops alone leaves the proxy's socket too"),
132
+
(&["173.245.48.0/20"], Some("173.245.48.9"), Some("198.51.100.4, 203.0.113.7"), Some("203.0.113.7"),
133
+
"bobbin stops before reaching anything a client wrote, since the proxy appends the address it saw to the right of all of it"),
134
+
(&["127.0.0.1"], Some("::ffff:203.0.113.7"), None, Some("203.0.113.7"),
135
+
"a v4-mapped socket and the plain address are one client"),
136
+
(&["127.0.0.1"], Some("::ffff:127.0.0.1"), Some("::ffff:198.51.100.4"), Some("198.51.100.4"),
137
+
"both spellings must key to one bucket, since a dual-stack listener reports the mapped form on both sides"),
138
+
(&["127.0.0.1"], None, Some("198.51.100.4"), None,
139
+
"bobbin won't identify a client it hasn't seen connect, since it can't check the header against anything"),
140
+
]
141
+
.iter()
142
+
.for_each(|&(proxies, socket, claimed, expected, why)| {
143
+
assert_eq!(
144
+
forwarded(proxies, socket, claimed).as_deref(),
145
+
expected,
146
+
"{why}: {proxies:?} saw {socket:?} claiming {claimed:?}"
147
+
);
148
+
});
149
+
}
150
+
}
+35
-16
bobbin/crates/xrpc/src/lib.rs
+35
-16
bobbin/crates/xrpc/src/lib.rs
···
95
95
use tracing::{Level, Span};
96
96
97
97
mod backpressure;
98
+
mod client_address;
98
99
mod filter;
99
100
100
101
pub use backpressure::{
101
102
HeavyLimiter, HeavyPermit, MaxInFlight, PerRequestAnonBytes, PressureVerdict, ReservedFloor,
102
103
};
104
+
use client_address::X_FORWARDED_FOR;
105
+
pub use client_address::{ClientAddress, SocketPeer};
103
106
use filter::{IssueFilter, ListFilter, NoFilter, PullFilter};
107
+
use trusted_proxies::TrustedProxies;
104
108
105
109
const DEFAULT_LIMIT: u32 = 50;
106
110
const FETCH_CONCURRENCY: usize = 8;
···
117
121
pub search: Arc<dyn SearchReader>,
118
122
pub resolver: Arc<RepoIdResolver>,
119
123
pub limiter: Option<Arc<HeavyLimiter>>,
124
+
pub client_address: Arc<ClientAddress>,
120
125
}
121
126
122
127
impl AppState {
···
143
148
search,
144
149
resolver,
145
150
limiter: None,
151
+
client_address: Arc::new(ClientAddress::default()),
146
152
}
147
153
}
148
154
149
155
pub fn with_limiter(mut self, limiter: Option<Arc<HeavyLimiter>>) -> Self {
150
156
self.limiter = limiter;
157
+
self
158
+
}
159
+
160
+
pub fn with_proxies(mut self, proxies: TrustedProxies) -> Self {
161
+
self.client_address = Arc::new(ClientAddress::new(proxies));
151
162
self
152
163
}
153
164
···
459
470
&CONTENT_RANGE,
460
471
];
461
472
462
-
static X_FORWARDED_FOR: HeaderName = HeaderName::from_static("x-forwarded-for");
463
-
464
-
const FORWARDED_REQUEST_HEADERS: &[&HeaderName] = &[
465
-
&RANGE,
466
-
&IF_RANGE,
467
-
&IF_NONE_MATCH,
468
-
&IF_MODIFIED_SINCE,
469
-
&X_FORWARDED_FOR,
470
-
];
473
+
const FORWARDED_REQUEST_HEADERS: &[&HeaderName] =
474
+
&[&RANGE, &IF_RANGE, &IF_NONE_MATCH, &IF_MODIFIED_SINCE];
471
475
472
476
const KNOT_HOST_PARAM: &str = "knot";
473
477
const REPO_PARAM: &str = "repo";
···
485
489
handler: H,
486
490
) -> Router<AppState>
487
491
where
488
-
H: Fn(AppState, HeaderMap, ProxyParams, Nsid<DefaultStr>) -> Fut
492
+
H: Fn(AppState, HeaderMap, SocketPeer, ProxyParams, Nsid<DefaultStr>) -> Fut
489
493
+ Clone
490
494
+ Send
491
495
+ Sync
···
500
504
get(
501
505
move |State(state): State<AppState>,
502
506
headers: HeaderMap,
507
+
socket: SocketPeer,
503
508
Query(params): Query<ProxyParams>| {
504
-
handler(state, headers, params, nsid.clone())
509
+
handler(state, headers, socket, params, nsid.clone())
505
510
},
506
511
),
507
512
)
···
2660
2665
}
2661
2666
}
2662
2667
2663
-
fn filter_request_headers(client: &HeaderMap) -> HeaderMap {
2664
-
FORWARDED_REQUEST_HEADERS
2668
+
fn filter_request_headers(
2669
+
client: &HeaderMap,
2670
+
socket: SocketPeer,
2671
+
address: &ClientAddress,
2672
+
) -> HeaderMap {
2673
+
let forwarded = FORWARDED_REQUEST_HEADERS
2665
2674
.iter()
2666
2675
.fold(HeaderMap::new(), |mut acc, name| {
2667
2676
if let Some(value) = client.get(*name) {
2668
2677
acc.insert((*name).clone(), value.clone());
2669
2678
}
2670
2679
acc
2680
+
});
2681
+
address
2682
+
.of(client, socket)
2683
+
.into_iter()
2684
+
.fold(forwarded, |mut acc, address| {
2685
+
acc.insert(X_FORWARDED_FOR.clone(), address);
2686
+
acc
2671
2687
})
2672
2688
}
2673
2689
···
2691
2707
async fn dispatch_proxy(
2692
2708
state: AppState,
2693
2709
headers: HeaderMap,
2710
+
socket: SocketPeer,
2694
2711
nsid: Nsid<DefaultStr>,
2695
2712
host: KnotHost,
2696
2713
params: ProxyParams,
···
2699
2716
.iter()
2700
2717
.map(|(k, v)| (k.as_str(), v.as_str()))
2701
2718
.collect();
2702
-
let allowed = filter_request_headers(&headers);
2719
+
let allowed = filter_request_headers(&headers, socket, &state.client_address);
2703
2720
let upstream = state
2704
2721
.knots
2705
2722
.forward(&host, &nsid, &forward, allowed)
···
2727
2744
async fn proxy_repo_handler(
2728
2745
state: AppState,
2729
2746
headers: HeaderMap,
2747
+
socket: SocketPeer,
2730
2748
params: ProxyParams,
2731
2749
nsid: Nsid<DefaultStr>,
2732
2750
) -> Result<Response, XrpcError> {
···
2741
2759
slug.as_str().to_owned(),
2742
2760
)))
2743
2761
.collect();
2744
-
dispatch_proxy(state, headers, nsid, host, forward).await
2762
+
dispatch_proxy(state, headers, socket, nsid, host, forward).await
2745
2763
}
2746
2764
2747
2765
async fn proxy_knot_handler(
2748
2766
state: AppState,
2749
2767
headers: HeaderMap,
2768
+
socket: SocketPeer,
2750
2769
params: ProxyParams,
2751
2770
nsid: Nsid<DefaultStr>,
2752
2771
) -> Result<Response, XrpcError> {
···
2755
2774
let host =
2756
2775
KnotHost::parse(&knot_raw).map_err(|e| XrpcError::InvalidParams(format!("knot: {e}")))?;
2757
2776
validate_client_supplied_knot(&state, &host)?;
2758
-
dispatch_proxy(state, headers, nsid, host, forward).await
2777
+
dispatch_proxy(state, headers, socket, nsid, host, forward).await
2759
2778
}
+100
-14
bobbin/crates/xrpc/tests/knot_proxy.rs
+100
-14
bobbin/crates/xrpc/tests/knot_proxy.rs
···
1
+
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
1
2
use std::sync::Arc;
2
3
use std::time::Duration;
3
4
4
5
use axum::body::{Body, to_bytes};
6
+
use axum::extract::ConnectInfo;
5
7
use bobbin_edge_index::{CoverageWatch, EdgeStore, StateIndex};
6
8
use bobbin_knot_proxy::{FailureThreshold, KnotHttpConfig, KnotProxy, KnotProxyConfig};
7
9
use bobbin_record_lru::{CacheCapacity, LruRecordStore};
···
10
12
use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader};
11
13
use bobbin_slingshot_client::SlingshotClient;
12
14
use bobbin_xrpc::{AppState, router};
13
-
use http::{Request, StatusCode};
15
+
use http::{HeaderName, HeaderValue, Request, StatusCode};
14
16
use jacquard_common::DefaultStr;
15
17
use jacquard_common::types::did::Did;
16
18
use jacquard_common::types::recordkey::Rkey;
17
19
use serde_json::{Value, json};
18
20
use tower::ServiceExt;
21
+
use trusted_proxies::TrustedProxies;
19
22
use url::Url;
20
23
use url::form_urlencoded::byte_serialize;
21
24
use wiremock::matchers::{header_exists, method, path, query_param};
···
23
26
24
27
const CID: &str = "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i";
25
28
29
+
const SOCKET: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4321);
30
+
26
31
fn did(s: &str) -> Did<DefaultStr> {
27
32
Did::new_owned(s).unwrap()
28
33
}
···
31
36
Rkey::new_owned(s).unwrap()
32
37
}
33
38
39
+
fn hdr(name: &'static str, value: &'static str) -> (HeaderName, HeaderValue) {
40
+
(
41
+
HeaderName::from_static(name),
42
+
HeaderValue::from_static(value),
43
+
)
44
+
}
45
+
34
46
fn test_config() -> KnotProxyConfig {
35
47
KnotProxyConfig {
36
48
failure_threshold: FailureThreshold::new(2).unwrap(),
···
58
70
Self::with_config(test_config()).await
59
71
}
60
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
+
61
84
async fn with_config(config: KnotProxyConfig) -> Self {
62
85
let slingshot_server = MockServer::start().await;
63
86
let knot_server = MockServer::start().await;
···
135
158
async fn call_with_headers(
136
159
&self,
137
160
path_and_query: &str,
138
-
client_headers: &[(&str, &str)],
161
+
client_headers: &[(HeaderName, HeaderValue)],
139
162
) -> http::Response<Body> {
140
-
let builder = client_headers
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()
141
189
.iter()
142
-
.fold(Request::builder().uri(path_and_query), |b, (k, v)| {
143
-
b.header(*k, *v)
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))
144
207
});
208
+
let builder = client_headers
209
+
.iter()
210
+
.fold(connected, |b, (name, value)| b.header(name, value));
145
211
router(self.state.clone())
146
212
.oneshot(builder.body(Body::empty()).unwrap())
147
213
.await
···
569
635
570
636
#[tokio::test]
571
637
async fn forwards_range_conditional_and_client_address_headers() {
572
-
let h = Harness::new().await;
638
+
let h = Harness::behind_proxy().await;
573
639
let tid = "3jzfcijpj2z2d";
574
640
h.mount_repo_record(&did("did:plc:limpet"), &rkey(tid), "kelp")
575
641
.await;
···
595
661
.call_with_headers(
596
662
&target,
597
663
&[
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"),
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"),
602
668
],
603
669
)
604
670
.await;
···
628
694
}
629
695
630
696
#[tokio::test]
697
+
async 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]
631
717
async fn drops_disallowed_client_headers() {
632
718
let h = Harness::new().await;
633
719
h.mount_repo_record(&did("did:plc:abalone"), &rkey("r1"), "barnacle")
···
647
733
.call_with_headers(
648
734
&target,
649
735
&[
650
-
("authorization", "Bearer secret"),
651
-
("cookie", "sid=evil"),
652
-
("x-custom", "should-not-pass"),
736
+
hdr("authorization", "Bearer secret"),
737
+
hdr("cookie", "sid=evil"),
738
+
hdr("x-custom", "should-not-pass"),
653
739
],
654
740
)
655
741
.await;
···
980
1066
enc("at://did:plc:limpet/sh.tangled.repo/r5"),
981
1067
);
982
1068
let resp = h
983
-
.call_with_headers(&target, &[("if-none-match", "\"v1\"")])
1069
+
.call_with_headers(&target, &[hdr("if-none-match", "\"v1\"")])
984
1070
.await;
985
1071
assert_eq!(resp.status(), StatusCode::NOT_MODIFIED);
986
1072
assert_eq!(resp.headers().get("etag").unwrap(), "\"v1\"");
+18
bobbin/example.toml
+18
bobbin/example.toml
···
23
23
# Default value: ""
24
24
#debug_bind = ""
25
25
26
+
# Reverse proxies in front of bobbin,
27
+
# each a bare IP address without a port or a CIDR block such as `173.245.48.0/20`.
28
+
# Bobbin will read the client address out of `x-forwarded-for`
29
+
# and forward that one address to the knot
30
+
# when a request arrives from a proxy on this list,
31
+
# so a knot that lists bobbin under its own `xrpc.trusted_proxies`
32
+
# can rate-limit per browser
33
+
# instead of pooling everyone bobbin serves into a single bucket.
34
+
# Bobbin will read the last 32 entries of the chain, at most.
35
+
# Leave empty when bobbin takes connections directly,
36
+
# since bobbin would otherwise believe a header any client can write.
37
+
# When using as an env var, comma-separated.
38
+
#
39
+
# Can also be specified via environment variable `BOBBIN_TRUSTED_PROXIES`.
40
+
#
41
+
# Default value: []
42
+
#trusted_proxies = []
43
+
26
44
[hydrant]
27
45
# Base URL of the hydrant instance - the cursor-replayable /stream lives
28
46
# under this. Use `ws://` or `wss://` - `http://` and `https://`