From ec5974939575e558698b46289ffb234013faf063 Mon Sep 17 00:00:00 2001 From: Lewis Date: Fri, 31 Jul 2026 10:21:24 +0300 Subject: [PATCH] bobbin/xrpc: forward client address that knot can rate-limit on Lewis: May this revision serve well! --- Cargo.lock | 2 + bobbin/crates/bobbin/Cargo.toml | 1 + bobbin/crates/bobbin/src/config.rs | 65 +++++++++- bobbin/crates/bobbin/src/main.rs | 16 ++- bobbin/crates/xrpc/Cargo.toml | 2 + bobbin/crates/xrpc/src/client_address.rs | 150 +++++++++++++++++++++++ bobbin/crates/xrpc/src/lib.rs | 51 +++++--- bobbin/crates/xrpc/tests/knot_proxy.rs | 114 ++++++++++++++--- bobbin/example.toml | 18 +++ 9 files changed, 383 insertions(+), 36 deletions(-) create mode 100644 bobbin/crates/xrpc/src/client_address.rs diff --git a/Cargo.lock b/Cargo.lock index e53dfca9..9f13193b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -698,6 +698,7 @@ dependencies = [ "toml", "tracing", "tracing-subscriber", + "trusted-proxies", "url", ] @@ -930,6 +931,7 @@ dependencies = [ "tower", "tower-http 0.7.0", "tracing", + "trusted-proxies", "url", "wiremock", ] diff --git a/bobbin/crates/bobbin/Cargo.toml b/bobbin/crates/bobbin/Cargo.toml index 674ecbb0..a8dd334b 100644 --- a/bobbin/crates/bobbin/Cargo.toml +++ b/bobbin/crates/bobbin/Cargo.toml @@ -20,6 +20,7 @@ bobbin-search = { workspace = true } bobbin-slingshot-client = { workspace = true } bobbin-xrpc = { workspace = true } rustls = { workspace = true } +trusted-proxies = { workspace = true } axum = { workspace = true } serde = { workspace = true, features = ["derive"] } diff --git a/bobbin/crates/bobbin/src/config.rs b/bobbin/crates/bobbin/src/config.rs index b3f897e1..a85c0537 100644 --- a/bobbin/crates/bobbin/src/config.rs +++ b/bobbin/crates/bobbin/src/config.rs @@ -5,6 +5,7 @@ use std::str::FromStr; use anyhow::{Context, anyhow}; use confique::Config; +use trusted_proxies::{ProxyNetError, TrustedProxies}; use url::Url; const SYSTEM_CONFIG_PATH: &str = "/etc/bobbin/config.toml"; @@ -14,6 +15,7 @@ const KNOWN_KEYS: &[&str] = &[ "server.binds", "server.shutdown_grace_secs", "server.debug_bind", + "server.trusted_proxies", "hydrant.url", "hydrant.start_cursor", "ingest.parallelism", @@ -36,6 +38,7 @@ const KNOWN_ENVS: &[&str] = &[ "BOBBIN_BIND", "BOBBIN_SHUTDOWN_GRACE_SECS", "BOBBIN_DEBUG_BIND", + "BOBBIN_TRUSTED_PROXIES", "BOBBIN_HYDRANT_URL", "BOBBIN_START_CURSOR", "BOBBIN_INGEST_PARALLELISM", @@ -103,6 +106,31 @@ pub struct ServerConfig { /// never reachable on the public listener. Bind to loopback only. #[config(env = "BOBBIN_DEBUG_BIND", default = "")] pub debug_bind: String, + + /// Reverse proxies in front of bobbin, + /// each a bare IP address without a port or a CIDR block such as `173.245.48.0/20`. + /// Bobbin will read the client address out of `x-forwarded-for` + /// and forward that one address to the knot + /// when a request arrives from a proxy on this list, + /// so a knot that lists bobbin under its own `xrpc.trusted_proxies` + /// can rate-limit per browser + /// instead of pooling everyone bobbin serves into a single bucket. + /// Bobbin will read the last 32 entries of the chain, at most. + /// Leave empty when bobbin takes connections directly, + /// since bobbin would otherwise believe a header any client can write. + /// When using as an env var, comma-separated. + #[config( + env = "BOBBIN_TRUSTED_PROXIES", + parse_env = trusted_proxies::comma_separated, + default = [] + )] + pub trusted_proxies: Vec, +} + +impl ServerConfig { + pub fn trusted_proxies(&self) -> Result { + TrustedProxies::parse(self.trusted_proxies.iter().map(String::as_str)) + } } #[derive(Debug, thiserror::Error)] @@ -269,10 +297,15 @@ pub fn load(path: Option<&PathBuf>) -> anyhow::Result { if let Some(p) = path { builder = builder.file(p); } - builder + let config = builder .file(SYSTEM_CONFIG_PATH) .load() - .context("load configuration") + .context("load configuration")?; + config + .server + .trusted_proxies() + .context("server.trusted_proxies takes a bare IP address or a CIDR block")?; + Ok(config) } pub fn template() -> String { @@ -453,6 +486,34 @@ mod tests { }); } + #[test] + fn known_envs_matches_every_env_attribute_this_crate_declares() { + let known: HashSet<&str> = KNOWN_ENVS.iter().copied().collect(); + let declared: HashSet<&str> = [include_str!("config.rs"), include_str!("main.rs")] + .into_iter() + .flat_map(|source| { + source + .split("env = \"") + .skip(1) + .filter_map(|rest| rest.split('"').next()) + }) + .collect(); + assert!( + declared.contains("BOBBIN_BIND") && declared.contains("BOBBIN_CONFIG"), + "the scan stopped matching config.rs or main.rs and every name in it would pass unchecked, since it came back with {declared:?}" + ); + let missing: Vec<&&str> = declared.difference(&known).collect(); + assert!( + missing.is_empty(), + "confique reads {missing:?} but check_envs will refuse to start with them set. Add them to KNOWN_ENVS" + ); + let stale: Vec<&&str> = known.difference(&declared).collect(); + assert!( + stale.is_empty(), + "KNOWN_ENVS lists {stale:?}, which the fields stopped reading. Drop them, or check_envs will keep accepting a name that stopped meaning anything" + ); + } + #[test] fn unknown_bobbin_env_rejected() { let err = check_envs(["BOBBIN_BIDNS"]).expect_err("typo must surface"); diff --git a/bobbin/crates/bobbin/src/main.rs b/bobbin/crates/bobbin/src/main.rs index f4aec499..91961c87 100644 --- a/bobbin/crates/bobbin/src/main.rs +++ b/bobbin/crates/bobbin/src/main.rs @@ -300,6 +300,10 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { format!("invalid server.debug_bind `{}`", cfg.server.debug_bind) })?) }; + let trusted_proxies = cfg + .server + .trusted_proxies() + .context("server.trusted_proxies takes a bare IP address or a CIDR block")?; let mem_probe = debug_bind.is_some().then(|| mem::MemProbe { edges: edges.clone(), search: search.clone(), @@ -318,7 +322,8 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { search as Arc, resolver, ) - .with_limiter(limiter); + .with_limiter(limiter) + .with_proxies(trusted_proxies); let app = router(state); let _debug_server = match (debug_bind, mem_probe) { @@ -428,9 +433,12 @@ async fn serve_all( let app = app.clone(); let cancel = cancel.clone(); async move { - axum::serve(listener, app) - .with_graceful_shutdown(async move { cancel.cancelled().await }) - .await + axum::serve( + listener, + app.into_make_service_with_connect_info::(), + ) + .with_graceful_shutdown(async move { cancel.cancelled().await }) + .await } }); diff --git a/bobbin/crates/xrpc/Cargo.toml b/bobbin/crates/xrpc/Cargo.toml index 7f13094a..6b8154cd 100644 --- a/bobbin/crates/xrpc/Cargo.toml +++ b/bobbin/crates/xrpc/Cargo.toml @@ -25,6 +25,7 @@ serde_json = { workspace = true } thiserror = { workspace = true } tower-http = { workspace = true, features = ["trace"] } tracing = { workspace = true } +trusted-proxies = { workspace = true } url = { workspace = true } [dev-dependencies] @@ -32,5 +33,6 @@ bobbin-runtime = { workspace = true } http = { workspace = true } tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } tower = { workspace = true } +trusted-proxies = { workspace = true } url = { workspace = true } wiremock = { workspace = true } diff --git a/bobbin/crates/xrpc/src/client_address.rs b/bobbin/crates/xrpc/src/client_address.rs new file mode 100644 index 00000000..e1da0b9f --- /dev/null +++ b/bobbin/crates/xrpc/src/client_address.rs @@ -0,0 +1,150 @@ +use std::convert::Infallible; +use std::net::{IpAddr, SocketAddr}; +use std::sync::OnceLock; + +use axum::extract::{ConnectInfo, FromRequestParts}; +use axum::http::request::Parts; +use axum::http::{HeaderMap, HeaderName, HeaderValue}; +use trusted_proxies::TrustedProxies; + +pub(crate) static X_FORWARDED_FOR: HeaderName = HeaderName::from_static("x-forwarded-for"); + +#[derive(Default)] +pub struct ClientAddress { + proxies: TrustedProxies, + ignored_header: OnceLock, + no_socket: OnceLock<()>, +} + +impl ClientAddress { + pub fn new(proxies: TrustedProxies) -> Self { + Self { + proxies, + ..Self::default() + } + } + + pub(crate) fn of(&self, headers: &HeaderMap, socket: SocketPeer) -> Option { + match socket.0 { + None => { + if self.no_socket.set(()).is_ok() { + tracing::warn!( + "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." + ); + } + None + } + Some(peer) => { + let relays = self.proxies.contains(peer); + if headers.contains_key(&X_FORWARDED_FOR) + && !relays + && self.ignored_header.set(peer).is_ok() + { + tracing::warn!( + %peer, + "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." + ); + } + let client = relays + .then(|| { + self.proxies.rightmost_untrusted( + headers + .get_all(&X_FORWARDED_FOR) + .iter() + .filter_map(|value| value.to_str().ok()) + .flat_map(|value| value.split(',')), + ) + }) + .flatten() + .unwrap_or_else(|| peer.to_canonical()); + HeaderValue::try_from(client.to_string()).ok() + } + } + } +} + +#[derive(Debug, Clone, Copy)] +pub struct SocketPeer(Option); + +impl FromRequestParts for SocketPeer { + type Rejection = Infallible; + + async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result { + Ok(Self( + parts + .extensions + .get::>() + .map(|info| info.0.ip()), + )) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ip(value: &str) -> IpAddr { + value.parse().unwrap() + } + + fn relaying<'a>(entries: impl IntoIterator) -> ClientAddress { + ClientAddress::new(TrustedProxies::parse(entries).unwrap()) + } + + fn chain(value: Option<&str>) -> HeaderMap { + value + .map(|value| { + let mut map = HeaderMap::new(); + map.insert(&X_FORWARDED_FOR, value.parse().unwrap()); + map + }) + .unwrap_or_default() + } + + fn forwarded(proxies: &[&str], socket: Option<&str>, claimed: Option<&str>) -> Option { + relaying(proxies.iter().copied()) + .of(&chain(claimed), SocketPeer(socket.map(ip))) + .map(|value| value.to_str().unwrap().to_owned()) + } + + #[test] + fn bobbin_forwards_the_socket_unless_a_listed_proxy_relayed_the_request() { + let listed: &[&str] = &["127.0.0.1", "173.245.48.0/20"]; + [ + (&["127.0.0.1"][..], Some("203.0.113.7"), Some("198.51.100.4"), Some("203.0.113.7"), + "bobbin must answer for the socket, since a client reaching it directly wrote that header itself"), + (&[], Some("203.0.113.7"), Some("198.51.100.4"), Some("203.0.113.7"), + "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"), + (listed, Some("127.0.0.1"), Some("198.51.100.4"), Some("198.51.100.4"), + "a listed proxy hands over the address it recorded"), + (listed, Some("127.0.0.1"), Some("198.51.100.4, 173.245.48.9"), Some("198.51.100.4"), + "and a second listed hop is stepped over with it"), + (listed, Some("127.0.0.1"), Some(" 198.51.100.4 "), Some("198.51.100.4"), + "padding around an entry won't hide it"), + (listed, Some("127.0.0.1"), Some("203.0.113.7, 198.51.100.4"), Some("198.51.100.4"), + "bobbin takes the rightmost unlisted hop"), + (listed, Some("127.0.0.1"), None, Some("127.0.0.1"), + "a listed proxy that didn't send the header leaves its own socket to forward"), + (listed, Some("127.0.0.1"), Some("not-an-ip"), Some("127.0.0.1"), + "bobbin stops at an entry it can't parse and forwards the socket, because a client can write anything left of the proxy"), + (listed, Some("127.0.0.1"), Some("127.0.0.1"), Some("127.0.0.1"), + "a chain of listed hops alone leaves the proxy's socket too"), + (&["173.245.48.0/20"], Some("173.245.48.9"), Some("198.51.100.4, 203.0.113.7"), Some("203.0.113.7"), + "bobbin stops before reaching anything a client wrote, since the proxy appends the address it saw to the right of all of it"), + (&["127.0.0.1"], Some("::ffff:203.0.113.7"), None, Some("203.0.113.7"), + "a v4-mapped socket and the plain address are one client"), + (&["127.0.0.1"], Some("::ffff:127.0.0.1"), Some("::ffff:198.51.100.4"), Some("198.51.100.4"), + "both spellings must key to one bucket, since a dual-stack listener reports the mapped form on both sides"), + (&["127.0.0.1"], None, Some("198.51.100.4"), None, + "bobbin won't identify a client it hasn't seen connect, since it can't check the header against anything"), + ] + .iter() + .for_each(|&(proxies, socket, claimed, expected, why)| { + assert_eq!( + forwarded(proxies, socket, claimed).as_deref(), + expected, + "{why}: {proxies:?} saw {socket:?} claiming {claimed:?}" + ); + }); + } +} diff --git a/bobbin/crates/xrpc/src/lib.rs b/bobbin/crates/xrpc/src/lib.rs index da3becfb..78efb975 100644 --- a/bobbin/crates/xrpc/src/lib.rs +++ b/bobbin/crates/xrpc/src/lib.rs @@ -95,12 +95,16 @@ use tower_http::trace::{DefaultMakeSpan, OnFailure, OnResponse, TraceLayer}; use tracing::{Level, Span}; mod backpressure; +mod client_address; mod filter; pub use backpressure::{ HeavyLimiter, HeavyPermit, MaxInFlight, PerRequestAnonBytes, PressureVerdict, ReservedFloor, }; +use client_address::X_FORWARDED_FOR; +pub use client_address::{ClientAddress, SocketPeer}; use filter::{IssueFilter, ListFilter, NoFilter, PullFilter}; +use trusted_proxies::TrustedProxies; const DEFAULT_LIMIT: u32 = 50; const FETCH_CONCURRENCY: usize = 8; @@ -117,6 +121,7 @@ pub struct AppState { pub search: Arc, pub resolver: Arc, pub limiter: Option>, + pub client_address: Arc, } impl AppState { @@ -143,6 +148,7 @@ impl AppState { search, resolver, limiter: None, + client_address: Arc::new(ClientAddress::default()), } } @@ -151,6 +157,11 @@ impl AppState { self } + pub fn with_proxies(mut self, proxies: TrustedProxies) -> Self { + self.client_address = Arc::new(ClientAddress::new(proxies)); + self + } + fn heavy_permit(&self) -> Result, XrpcError> { self.limiter.as_ref().map(|l| l.try_enter()).transpose() } @@ -459,15 +470,8 @@ const PASSTHROUGH_HEADERS: &[&HeaderName] = &[ &CONTENT_RANGE, ]; -static X_FORWARDED_FOR: HeaderName = HeaderName::from_static("x-forwarded-for"); - -const FORWARDED_REQUEST_HEADERS: &[&HeaderName] = &[ - &RANGE, - &IF_RANGE, - &IF_NONE_MATCH, - &IF_MODIFIED_SINCE, - &X_FORWARDED_FOR, -]; +const FORWARDED_REQUEST_HEADERS: &[&HeaderName] = + &[&RANGE, &IF_RANGE, &IF_NONE_MATCH, &IF_MODIFIED_SINCE]; const KNOT_HOST_PARAM: &str = "knot"; const REPO_PARAM: &str = "repo"; @@ -485,7 +489,7 @@ fn register_proxied( handler: H, ) -> Router where - H: Fn(AppState, HeaderMap, ProxyParams, Nsid) -> Fut + H: Fn(AppState, HeaderMap, SocketPeer, ProxyParams, Nsid) -> Fut + Clone + Send + Sync @@ -500,8 +504,9 @@ where get( move |State(state): State, headers: HeaderMap, + socket: SocketPeer, Query(params): Query| { - handler(state, headers, params, nsid.clone()) + handler(state, headers, socket, params, nsid.clone()) }, ), ) @@ -2660,14 +2665,25 @@ fn pick_human_slug(rkey: Option<&Rkey>, name: Option<&str>) -> Optio } } -fn filter_request_headers(client: &HeaderMap) -> HeaderMap { - FORWARDED_REQUEST_HEADERS +fn filter_request_headers( + client: &HeaderMap, + socket: SocketPeer, + address: &ClientAddress, +) -> HeaderMap { + let forwarded = FORWARDED_REQUEST_HEADERS .iter() .fold(HeaderMap::new(), |mut acc, name| { if let Some(value) = client.get(*name) { acc.insert((*name).clone(), value.clone()); } acc + }); + address + .of(client, socket) + .into_iter() + .fold(forwarded, |mut acc, address| { + acc.insert(X_FORWARDED_FOR.clone(), address); + acc }) } @@ -2691,6 +2707,7 @@ fn upstream_to_axum(resp: ProxyResponse) -> Response { async fn dispatch_proxy( state: AppState, headers: HeaderMap, + socket: SocketPeer, nsid: Nsid, host: KnotHost, params: ProxyParams, @@ -2699,7 +2716,7 @@ async fn dispatch_proxy( .iter() .map(|(k, v)| (k.as_str(), v.as_str())) .collect(); - let allowed = filter_request_headers(&headers); + let allowed = filter_request_headers(&headers, socket, &state.client_address); let upstream = state .knots .forward(&host, &nsid, &forward, allowed) @@ -2727,6 +2744,7 @@ fn extract_param( async fn proxy_repo_handler( state: AppState, headers: HeaderMap, + socket: SocketPeer, params: ProxyParams, nsid: Nsid, ) -> Result { @@ -2741,12 +2759,13 @@ async fn proxy_repo_handler( slug.as_str().to_owned(), ))) .collect(); - dispatch_proxy(state, headers, nsid, host, forward).await + dispatch_proxy(state, headers, socket, nsid, host, forward).await } async fn proxy_knot_handler( state: AppState, headers: HeaderMap, + socket: SocketPeer, params: ProxyParams, nsid: Nsid, ) -> Result { @@ -2755,5 +2774,5 @@ async fn proxy_knot_handler( let host = KnotHost::parse(&knot_raw).map_err(|e| XrpcError::InvalidParams(format!("knot: {e}")))?; validate_client_supplied_knot(&state, &host)?; - dispatch_proxy(state, headers, nsid, host, forward).await + dispatch_proxy(state, headers, socket, nsid, host, forward).await } diff --git a/bobbin/crates/xrpc/tests/knot_proxy.rs b/bobbin/crates/xrpc/tests/knot_proxy.rs index 97601c7e..9278d123 100644 --- a/bobbin/crates/xrpc/tests/knot_proxy.rs +++ b/bobbin/crates/xrpc/tests/knot_proxy.rs @@ -1,7 +1,9 @@ +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::sync::Arc; use std::time::Duration; use axum::body::{Body, to_bytes}; +use axum::extract::ConnectInfo; use bobbin_edge_index::{CoverageWatch, EdgeStore, StateIndex}; use bobbin_knot_proxy::{FailureThreshold, KnotHttpConfig, KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; @@ -10,12 +12,13 @@ use bobbin_runtime::{RuntimeHasher, SystemClock}; use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; use bobbin_xrpc::{AppState, router}; -use http::{Request, StatusCode}; +use http::{HeaderName, HeaderValue, Request, StatusCode}; use jacquard_common::DefaultStr; use jacquard_common::types::did::Did; use jacquard_common::types::recordkey::Rkey; use serde_json::{Value, json}; use tower::ServiceExt; +use trusted_proxies::TrustedProxies; use url::Url; use url::form_urlencoded::byte_serialize; use wiremock::matchers::{header_exists, method, path, query_param}; @@ -23,6 +26,8 @@ use wiremock::{Mock, MockServer, ResponseTemplate}; const CID: &str = "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i"; +const SOCKET: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4321); + fn did(s: &str) -> Did { Did::new_owned(s).unwrap() } @@ -31,6 +36,13 @@ fn rkey(s: &str) -> Rkey { Rkey::new_owned(s).unwrap() } +fn hdr(name: &'static str, value: &'static str) -> (HeaderName, HeaderValue) { + ( + HeaderName::from_static(name), + HeaderValue::from_static(value), + ) +} + fn test_config() -> KnotProxyConfig { KnotProxyConfig { failure_threshold: FailureThreshold::new(2).unwrap(), @@ -58,6 +70,17 @@ impl Harness { Self::with_config(test_config()).await } + async fn behind_proxy() -> Self { + let harness = Self::with_config(test_config()).await; + Self { + state: harness + .state + .clone() + .with_proxies(TrustedProxies::parse(["127.0.0.1"]).unwrap()), + ..harness + } + } + async fn with_config(config: KnotProxyConfig) -> Self { let slingshot_server = MockServer::start().await; let knot_server = MockServer::start().await; @@ -135,13 +158,56 @@ impl Harness { async fn call_with_headers( &self, path_and_query: &str, - client_headers: &[(&str, &str)], + client_headers: &[(HeaderName, HeaderValue)], ) -> http::Response { - let builder = client_headers + self.call_from(path_and_query, client_headers, Some(SOCKET)) + .await + } + + async fn blob_client_address(&self, tid: &str, socket: Option) -> Option { + self.mount_repo_record(&did("did:plc:limpet"), &rkey(tid), "kelp") + .await; + Mock::given(method("GET")) + .and(path("/xrpc/sh.tangled.repo.blob")) + .respond_with( + ResponseTemplate::new(200).set_body_raw(r#"{"path":"x"}"#, "application/json"), + ) + .mount(&self.knot) + .await; + let target = format!( + "/xrpc/sh.tangled.repo.blob?repo={}&path=x", + enc(&format!("at://did:plc:limpet/sh.tangled.repo/{tid}")), + ); + let resp = self + .call_from(&target, &[hdr("x-forwarded-for", "203.0.113.42")], socket) + .await; + assert_eq!(resp.status(), StatusCode::OK); + self.knot + .received_requests() + .await + .unwrap() .iter() - .fold(Request::builder().uri(path_and_query), |b, (k, v)| { - b.header(*k, *v) + .find(|r| r.url.path() == "/xrpc/sh.tangled.repo.blob") + .expect("knot received the proxied call") + .headers + .get("x-forwarded-for") + .map(|value| value.to_str().unwrap().to_owned()) + } + + async fn call_from( + &self, + path_and_query: &str, + client_headers: &[(HeaderName, HeaderValue)], + socket: Option, + ) -> http::Response { + let connected = socket + .into_iter() + .fold(Request::builder().uri(path_and_query), |b, socket| { + b.extension(ConnectInfo(socket)) }); + let builder = client_headers + .iter() + .fold(connected, |b, (name, value)| b.header(name, value)); router(self.state.clone()) .oneshot(builder.body(Body::empty()).unwrap()) .await @@ -569,7 +635,7 @@ async fn does_not_inject_auth_or_atproto_proxy_headers() { #[tokio::test] async fn forwards_range_conditional_and_client_address_headers() { - let h = Harness::new().await; + let h = Harness::behind_proxy().await; let tid = "3jzfcijpj2z2d"; h.mount_repo_record(&did("did:plc:limpet"), &rkey(tid), "kelp") .await; @@ -595,10 +661,10 @@ async fn forwards_range_conditional_and_client_address_headers() { .call_with_headers( &target, &[ - ("range", "bytes=0-99"), - ("if-none-match", "\"old\""), - ("if-modified-since", "Wed, 01 May 2026 00:00:00 GMT"), - ("x-forwarded-for", "203.0.113.42"), + hdr("range", "bytes=0-99"), + hdr("if-none-match", "\"old\""), + hdr("if-modified-since", "Wed, 01 May 2026 00:00:00 GMT"), + hdr("x-forwarded-for", "203.0.113.42"), ], ) .await; @@ -627,6 +693,26 @@ async fn forwards_range_conditional_and_client_address_headers() { ); } +#[tokio::test] +async fn bobbin_forwards_only_a_client_address_it_can_vouch_for() { + assert_eq!( + Harness::new() + .await + .blob_client_address("3jzfcijpj2z2e", Some(SOCKET)) + .await, + Some(SOCKET.ip().to_string()), + "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" + ); + assert_eq!( + Harness::behind_proxy() + .await + .blob_client_address("3jzfcijpj2z2f", None) + .await, + None, + "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" + ); +} + #[tokio::test] async fn drops_disallowed_client_headers() { let h = Harness::new().await; @@ -647,9 +733,9 @@ async fn drops_disallowed_client_headers() { .call_with_headers( &target, &[ - ("authorization", "Bearer secret"), - ("cookie", "sid=evil"), - ("x-custom", "should-not-pass"), + hdr("authorization", "Bearer secret"), + hdr("cookie", "sid=evil"), + hdr("x-custom", "should-not-pass"), ], ) .await; @@ -980,7 +1066,7 @@ async fn knot_not_modified_passes_through() { enc("at://did:plc:limpet/sh.tangled.repo/r5"), ); let resp = h - .call_with_headers(&target, &[("if-none-match", "\"v1\"")]) + .call_with_headers(&target, &[hdr("if-none-match", "\"v1\"")]) .await; assert_eq!(resp.status(), StatusCode::NOT_MODIFIED); assert_eq!(resp.headers().get("etag").unwrap(), "\"v1\""); diff --git a/bobbin/example.toml b/bobbin/example.toml index b5cbdb9b..e9646645 100644 --- a/bobbin/example.toml +++ b/bobbin/example.toml @@ -23,6 +23,24 @@ # Default value: "" #debug_bind = "" +# Reverse proxies in front of bobbin, +# each a bare IP address without a port or a CIDR block such as `173.245.48.0/20`. +# Bobbin will read the client address out of `x-forwarded-for` +# and forward that one address to the knot +# when a request arrives from a proxy on this list, +# so a knot that lists bobbin under its own `xrpc.trusted_proxies` +# can rate-limit per browser +# instead of pooling everyone bobbin serves into a single bucket. +# Bobbin will read the last 32 entries of the chain, at most. +# Leave empty when bobbin takes connections directly, +# since bobbin would otherwise believe a header any client can write. +# When using as an env var, comma-separated. +# +# Can also be specified via environment variable `BOBBIN_TRUSTED_PROXIES`. +# +# Default value: [] +#trusted_proxies = [] + [hydrant] # Base URL of the hydrant instance - the cursor-replayable /stream lives # under this. Use `ws://` or `wss://` - `http://` and `https://` -- 2.51.2