diff --git a/bobbin/crates/bobbin/src/config.rs b/bobbin/crates/bobbin/src/config.rs index a85c0537..137eff33 100644 --- a/bobbin/crates/bobbin/src/config.rs +++ b/bobbin/crates/bobbin/src/config.rs @@ -29,6 +29,7 @@ const KNOWN_KEYS: &[&str] = &[ "search.heap_bytes", "knot.allow_private", "knot.require_https", + "mirror.url", "log.format", "log.filter", ]; @@ -52,6 +53,7 @@ const KNOWN_ENVS: &[&str] = &[ "BOBBIN_SEARCH_HEAP_BYTES", "BOBBIN_KNOT_ALLOW_PRIVATE", "BOBBIN_KNOT_REQUIRE_HTTPS", + "BOBBIN_MIRROR_URL", "BOBBIN_LOG_FORMAT", "BOBBIN_LOG", ]; @@ -82,6 +84,9 @@ pub struct BobbinConfig { #[config(nested)] pub knot: KnotConfig, + #[config(nested)] + pub mirror: MirrorConfig, + #[config(nested)] pub log: LogConfig, } @@ -253,6 +258,12 @@ pub struct KnotConfig { pub require_https: bool, } +#[derive(Debug, Config)] +pub struct MirrorConfig { + #[config(env = "BOBBIN_MIRROR_URL")] + pub url: Option, +} + #[derive(Debug, Config)] pub struct LogConfig { /// Log emitter format. `text` produces human-readable output for local diff --git a/bobbin/crates/bobbin/src/main.rs b/bobbin/crates/bobbin/src/main.rs index 91961c87..61bd99d6 100644 --- a/bobbin/crates/bobbin/src/main.rs +++ b/bobbin/crates/bobbin/src/main.rs @@ -11,7 +11,7 @@ use bobbin_ingest::{ IngestConfig, IngestRuntime, RepoIdResolver, WarmingBuffer, run as run_ingest, }; use bobbin_knot_ingest::{CapabilityGate, KnotClient, KnotRegistry, Orchestrator}; -use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig, classify_ip}; +use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig, MirrorProxy, classify_ip}; use bobbin_record_lru::{CacheCapacity, LruRecordStore, RecordStore}; use bobbin_runtime::{ Clock, GuardedWs, MemoryBudget, NetworkError, OsEntropy, RuntimeHasher, SystemClock, @@ -201,8 +201,22 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { }, KnotHttpConfig::default(), clock.clone(), - hasher, + hasher.clone(), )?); + let mirror = cfg + .mirror + .url + .as_ref() + .map(|url| MirrorProxy::new(url, clock.clone(), hasher.clone()).map(Arc::new)) + .transpose() + .context("mirror.url")?; + match mirror.as_ref() { + Some(m) => tracing::info!( + mirror = %m.host().url(), + "we will forward git reads to the mirror before any knot", + ), + None => tracing::info!("we will forward git reads to knots, since mirror.url is unset"), + } let search_heap = usize::try_from(search_heap_cap) .with_context(|| format!("search heap {search_heap_cap} exceeds usize"))?; let search = Arc::new(SearchIndex::new(search_heap, clock.clone())?); @@ -323,6 +337,7 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { resolver, ) .with_limiter(limiter) + .with_mirror(mirror) .with_proxies(trusted_proxies); let app = router(state); diff --git a/bobbin/crates/xrpc/src/lib.rs b/bobbin/crates/xrpc/src/lib.rs index 78efb975..d8f4ffd9 100644 --- a/bobbin/crates/xrpc/src/lib.rs +++ b/bobbin/crates/xrpc/src/lib.rs @@ -14,8 +14,8 @@ use axum::{ HeaderMap, HeaderName, StatusCode, header::{ ACCEPT_RANGES, CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LENGTH, - CONTENT_RANGE, CONTENT_TYPE, ETAG, IF_MODIFIED_SINCE, IF_NONE_MATCH, IF_RANGE, - LAST_MODIFIED, RANGE, + CONTENT_RANGE, CONTENT_SECURITY_POLICY, CONTENT_TYPE, ETAG, IF_MODIFIED_SINCE, + IF_NONE_MATCH, IF_RANGE, LAST_MODIFIED, RANGE, X_CONTENT_TYPE_OPTIONS, }, request::Parts, }, @@ -26,7 +26,9 @@ use bobbin_edge_index::{ Coverage, CoverageWatch, CursorParseError, EdgeItem, EdgePage, EdgeStore, IssueStateKind, PageCursor, PageLimit, PageToken, PullStatusKind, SortDir, StateIndex, StateKind, }; -use bobbin_knot_proxy::{KnotHost, KnotProxy, KnotProxyError, ProxyResponse, RepoSlug}; +use bobbin_knot_proxy::{ + KnotHost, KnotProxy, KnotProxyError, MirrorNsid, MirrorProxy, ProxyResponse, RepoSlug, +}; use bobbin_record_lru::RecordStore; use bobbin_resolver::RepoIdResolver; use bobbin_search::{ @@ -118,6 +120,7 @@ pub struct AppState { pub pull_statuses: Arc>, pub coverage: Arc, pub knots: Arc, + pub mirror: Option>, pub search: Arc, pub resolver: Arc, pub limiter: Option>, @@ -145,6 +148,7 @@ impl AppState { pull_statuses, coverage, knots, + mirror: None, search, resolver, limiter: None, @@ -157,6 +161,11 @@ impl AppState { self } + pub fn with_mirror(mut self, mirror: Option>) -> Self { + self.mirror = mirror; + self + } + pub fn with_proxies(mut self, proxies: TrustedProxies) -> Self { self.client_address = Arc::new(ClientAddress::new(proxies)); self @@ -468,9 +477,11 @@ const PASSTHROUGH_HEADERS: &[&HeaderName] = &[ &CONTENT_DISPOSITION, &ACCEPT_RANGES, &CONTENT_RANGE, + &X_CONTENT_TYPE_OPTIONS, + &CONTENT_SECURITY_POLICY, ]; -const FORWARDED_REQUEST_HEADERS: &[&HeaderName] = +const RANGE_OR_CONDITIONAL_HEADERS: &[&HeaderName] = &[&RANGE, &IF_RANGE, &IF_NONE_MATCH, &IF_MODIFIED_SINCE]; const KNOT_HOST_PARAM: &str = "knot"; @@ -2637,10 +2648,16 @@ fn validate_client_supplied_knot(state: &AppState, host: &KnotHost) -> Result<() } } +struct RepoTarget { + host: KnotHost, + slug: RepoSlug, + repo_did: Option>, +} + async fn resolve_knot_target( state: &AppState, repo_uri: AtUri, -) -> Result<(KnotHost, RepoSlug), XrpcError> { +) -> Result { let rkey: Option> = repo_uri.rkey().map(|r| r.clone().into_static()); let (body, did) = resolve(state, ExpectedNsid::from_static(RepoRecord::NSID), repo_uri).await?; let value: Repo = serde_json::from_slice(&body.value) @@ -2652,7 +2669,11 @@ async fn resolve_knot_target( })?; let slug = RepoSlug::new(&did, &name) .map_err(|e| XrpcError::InvalidRecord(format!("repo slug: {e}")))?; - Ok((host, slug)) + Ok(RepoTarget { + host, + slug, + repo_did: value.repo_did, + }) } fn pick_human_slug(rkey: Option<&Rkey>, name: Option<&str>) -> Option { @@ -2670,7 +2691,7 @@ fn filter_request_headers( socket: SocketPeer, address: &ClientAddress, ) -> HeaderMap { - let forwarded = FORWARDED_REQUEST_HEADERS + let forwarded = RANGE_OR_CONDITIONAL_HEADERS .iter() .fold(HeaderMap::new(), |mut acc, name| { if let Some(value) = client.get(*name) { @@ -2704,27 +2725,70 @@ fn upstream_to_axum(resp: ProxyResponse) -> Response { response } -async fn dispatch_proxy( - state: AppState, +async fn dispatch_knot( + state: &AppState, + nsid: &Nsid, + host: &KnotHost, + query: &[(&str, &str)], headers: HeaderMap, - socket: SocketPeer, - nsid: Nsid, - host: KnotHost, - params: ProxyParams, ) -> Result { - let forward: Vec<(&str, &str)> = params - .iter() - .map(|(k, v)| (k.as_str(), v.as_str())) - .collect(); - let allowed = filter_request_headers(&headers, socket, &state.client_address); let upstream = state .knots - .forward(&host, &nsid, &forward, allowed) + .forward(host, nsid, query, headers) .await .map_err(map_proxy_error)?; Ok(upstream_to_axum(upstream)) } +async fn dispatch_mirror( + state: &AppState, + nsid: &Nsid, + target: &RepoTarget, + query: &[(&str, &str)], + headers: &HeaderMap, +) -> Option { + let mirror = state.mirror.as_ref()?; + let repo_did = target.repo_did.as_ref()?; + if RANGE_OR_CONDITIONAL_HEADERS + .iter() + .any(|name| headers.contains_key(*name)) + { + return None; + } + let mirror_nsid = MirrorNsid::route(nsid.as_ref(), query)?; + let refused = match mirror + .forward(&mirror_nsid, repo_did, query, headers.clone()) + .await + { + Ok(upstream) if !upstream.status().is_client_error() => { + return Some(upstream_to_axum(upstream)); + } + Ok(upstream) => { + let status = upstream.status(); + upstream.discard().await; + status.to_string() + } + Err(KnotProxyError::Upstream(status)) => status.to_string(), + Err(err @ KnotProxyError::CircuitOpen) => err.to_string(), + Err(err) => { + tracing::warn!( + nsid = mirror_nsid.as_str(), + repo = repo_did.as_ref(), + error = %err, + "mirror call failed, asking the knot", + ); + return None; + } + }; + tracing::debug!( + nsid = mirror_nsid.as_str(), + repo = repo_did.as_ref(), + refused, + "mirror couldn't serve this repo, asking the knot", + ); + None +} + fn extract_param( params: ProxyParams, key: &str, @@ -2751,15 +2815,17 @@ async fn proxy_repo_handler( let (repo_raw, rest) = extract_param(params, REPO_PARAM)? .ok_or_else(|| XrpcError::InvalidParams("missing repo".into()))?; let repo_uri = parse_uri(&repo_raw)?; - let (host, slug) = resolve_knot_target(&state, repo_uri).await?; - let forward = rest + let target = resolve_knot_target(&state, repo_uri).await?; + let allowed = filter_request_headers(&headers, socket, &state.client_address); + let query: Vec<(&str, &str)> = rest.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect(); + if let Some(response) = dispatch_mirror(&state, &nsid, &target, &query, &allowed).await { + return Ok(response); + } + let forward: Vec<(&str, &str)> = query .into_iter() - .chain(std::iter::once(( - REPO_PARAM.to_owned(), - slug.as_str().to_owned(), - ))) + .chain(std::iter::once((REPO_PARAM, target.slug.as_str()))) .collect(); - dispatch_proxy(state, headers, socket, nsid, host, forward).await + dispatch_knot(&state, &nsid, &target.host, &forward, allowed).await } async fn proxy_knot_handler( @@ -2769,10 +2835,12 @@ async fn proxy_knot_handler( params: ProxyParams, nsid: Nsid, ) -> Result { - let (knot_raw, forward) = extract_param(params, KNOT_HOST_PARAM)? + let (knot_raw, rest) = extract_param(params, KNOT_HOST_PARAM)? .ok_or_else(|| XrpcError::InvalidParams("missing knot".into()))?; let host = KnotHost::parse(&knot_raw).map_err(|e| XrpcError::InvalidParams(format!("knot: {e}")))?; validate_client_supplied_knot(&state, &host)?; - dispatch_proxy(state, headers, socket, nsid, host, forward).await + let allowed = filter_request_headers(&headers, socket, &state.client_address); + let forward: Vec<(&str, &str)> = rest.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect(); + dispatch_knot(&state, &nsid, &host, &forward, allowed).await } diff --git a/bobbin/crates/xrpc/tests/mirror_proxy.rs b/bobbin/crates/xrpc/tests/mirror_proxy.rs new file mode 100644 index 00000000..2a8e417a --- /dev/null +++ b/bobbin/crates/xrpc/tests/mirror_proxy.rs @@ -0,0 +1,358 @@ +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::{KnotHttpConfig, KnotProxy, KnotProxyConfig, MirrorProxy}; +use bobbin_record_lru::{CacheCapacity, LruRecordStore}; +use bobbin_resolver::RepoIdResolver; +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 serde_json::json; +use tower::ServiceExt; +use url::Url; +use url::form_urlencoded::byte_serialize; +use wiremock::matchers::{method, path, query_param}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +const CID: &str = "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i"; +const SOCKET: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4321); +const OWNER: &str = "did:plc:nel"; +const RKEY: &str = "periwinkle"; +const REPO_DID: &str = "did:plc:periwinkle"; +const REPO_URI: &str = "at://did:plc:nel/sh.tangled.repo/periwinkle"; + +const FROM_MIRROR: &str = r#"{"served_by":"mirror"}"#; +const FROM_KNOT: &str = r#"{"served_by":"knot"}"#; + +enum Mirror { + Off, + Live, + Unreachable, +} + +fn closed_port() -> String { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + drop(listener); + format!("http://{addr}") +} + +fn enc(s: &str) -> String { + byte_serialize(s.as_bytes()).collect() +} + +fn ok_from_mirror() -> ResponseTemplate { + ResponseTemplate::new(200).set_body_raw(FROM_MIRROR, "application/json") +} + +async fn paths(server: &MockServer) -> Vec { + server + .received_requests() + .await + .unwrap() + .iter() + .map(|r| r.url.path().to_owned()) + .collect() +} + +struct Harness { + _slingshot: MockServer, + knot: MockServer, + mirror: MockServer, + state: AppState, +} + +impl Harness { + async fn new(setting: Mirror, repo_did: Option<&str>) -> Self { + let slingshot = MockServer::start().await; + let knot = MockServer::start().await; + let mirror = MockServer::start().await; + let clock = Arc::new(SystemClock::new()); + let mirror_proxy = match setting { + Mirror::Off => None, + Mirror::Live => Some(mirror.uri()), + Mirror::Unreachable => Some(closed_port()), + } + .map(|url| { + Arc::new( + MirrorProxy::new( + &Url::parse(&url).unwrap(), + clock.clone(), + RuntimeHasher::default(), + ) + .unwrap(), + ) + }); + let state = AppState::new( + Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), + SlingshotClient::with_default_http(Url::parse(&slingshot.uri()).unwrap()).unwrap(), + Arc::new(EdgeStore::new(RuntimeHasher::default())), + Arc::new(StateIndex::new(RuntimeHasher::default())), + Arc::new(StateIndex::new(RuntimeHasher::default())), + Arc::new(CoverageWatch::new()), + Arc::new( + KnotProxy::new( + KnotProxyConfig { + allow_private_hosts: true, + require_https: false, + ..KnotProxyConfig::default() + }, + KnotHttpConfig { + connect_timeout: Duration::from_millis(500), + read_timeout: Duration::from_secs(2), + }, + clock.clone(), + RuntimeHasher::default(), + ) + .unwrap(), + ), + Arc::new(SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, clock).unwrap()) + as Arc, + Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), + ) + .with_mirror(mirror_proxy); + + let mut record = json!({ + "$type": "sh.tangled.repo", + "createdAt": "2026-05-01T00:00:00Z", + "knot": knot.uri(), + "name": "periwinkle", + }); + if let Some(d) = repo_did { + record["repoDid"] = json!(d); + } + Mock::given(method("GET")) + .and(path("/xrpc/com.atproto.repo.getRecord")) + .and(query_param("repo", OWNER)) + .and(query_param("collection", "sh.tangled.repo")) + .and(query_param("rkey", RKEY)) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "uri": REPO_URI, + "cid": CID, + "value": record, + }))) + .mount(&slingshot) + .await; + + Self { + _slingshot: slingshot, + knot, + mirror, + state, + } + } + + async fn with_mirror() -> Self { + Self::new(Mirror::Live, Some(REPO_DID)).await + } + + async fn mount_mirror(&self, nsid: &str, response: ResponseTemplate) { + Mock::given(method("GET")) + .and(path(format!("/xrpc/{nsid}"))) + .respond_with(response) + .mount(&self.mirror) + .await; + } + + async fn mount_knot(&self, nsid: &str) { + Mock::given(method("GET")) + .and(path(format!("/xrpc/{nsid}"))) + .respond_with(ResponseTemplate::new(200).set_body_raw(FROM_KNOT, "application/json")) + .mount(&self.knot) + .await; + } + + async fn served_by(&self, nsid: &str, query: &str) -> String { + self.served_by_with(nsid, query, &[]).await + } + + async fn served_by_with(&self, nsid: &str, query: &str, headers: &[(&str, &str)]) -> String { + let target = format!("/xrpc/{nsid}?repo={}&{query}", enc(REPO_URI)); + let request = headers.iter().fold( + Request::builder() + .uri(target) + .extension(ConnectInfo(SOCKET)), + |builder, (name, value)| builder.header(*name, *value), + ); + let resp = router(self.state.clone()) + .oneshot(request.body(Body::empty()).unwrap()) + .await + .expect("router infallible"); + assert_eq!(resp.status(), StatusCode::OK, "{nsid}?{query}"); + let body = to_bytes(resp.into_body(), 64 * 1024).await.unwrap(); + String::from_utf8(body.to_vec()).unwrap() + } +} + +#[tokio::test] +async fn tree_reads_the_mirror_keyed_on_the_repo_did() { + let h = Harness::with_mirror().await; + Mock::given(method("GET")) + .and(path("/xrpc/sh.tangled.git.temp.getTree")) + .and(query_param("repo", REPO_DID)) + .and(query_param("ref", "main")) + .respond_with(ok_from_mirror()) + .mount(&h.mirror) + .await; + h.mount_knot("sh.tangled.repo.tree").await; + + assert_eq!( + h.served_by("sh.tangled.repo.tree", "ref=main").await, + FROM_MIRROR, + "the mirror answers a read keyed on the repo did", + ); + assert!( + paths(&h.knot).await.is_empty(), + "the knot must stay untouched" + ); +} + +#[tokio::test] +async fn every_shape_compatible_request_reads_the_mirror() { + #[rustfmt::skip] + let routed = [ + ("sh.tangled.repo.branches", "sh.tangled.git.temp.listBranches", ""), + ("sh.tangled.repo.log", "sh.tangled.git.temp.listCommits", "ref=main"), + ("sh.tangled.repo.log", "sh.tangled.git.temp.listCommits", "ref=main&path="), + ("sh.tangled.repo.tag", "sh.tangled.git.temp.getTag", "tag=v1"), + ("sh.tangled.repo.tags", "sh.tangled.git.temp.listTags", ""), + ("sh.tangled.repo.tree", "sh.tangled.git.temp.getTree", "ref=main"), + ("sh.tangled.repo.tree", "sh.tangled.git.temp.getTree", "ref=main&path=crates"), + ]; + for (knot_nsid, mirror_nsid, query) in routed { + let h = Harness::with_mirror().await; + h.mount_mirror(mirror_nsid, ok_from_mirror()).await; + h.mount_knot(knot_nsid).await; + assert_eq!( + h.served_by(knot_nsid, query).await, + FROM_MIRROR, + "{knot_nsid}?{query} must read the mirror", + ); + assert_eq!(paths(&h.mirror).await, vec![format!("/xrpc/{mirror_nsid}")]); + } +} + +#[tokio::test] +async fn every_request_the_mirror_answers_in_another_shape_reads_the_knot() { + #[rustfmt::skip] + let refused = [ + ("sh.tangled.repo.blob", "ref=main&path=x", "the mirror serves content types the knot answers with 403"), + ("sh.tangled.repo.blob", "ref=main&path=x&raw=true", "raw doesn't exempt the blob"), + ("sh.tangled.repo.archive", "ref=main", "a resume would splice knot bytes onto a mirror tarball"), + ("sh.tangled.repo.log", "ref=main&path=crates/xrpc", "the mirror ignores path"), + ("sh.tangled.repo.branch", "name=main", "the mirror answers branch in another shape"), + ("sh.tangled.repo.languages", "ref=main", "the mirror answers languages in another shape"), + ("sh.tangled.repo.compare", "", "outside the routing table"), + ("sh.tangled.repo.describeRepo", "", "outside the routing table"), + ("sh.tangled.repo.diff", "", "outside the routing table"), + ("sh.tangled.repo.getDefaultBranch", "", "outside the routing table"), + ("sh.tangled.repo.listSecrets", "", "outside the routing table"), + ]; + for (nsid, query, why) in refused { + let h = Harness::with_mirror().await; + h.mount_knot(nsid).await; + assert_eq!(h.served_by(nsid, query).await, FROM_KNOT, "{nsid}: {why}"); + assert!(paths(&h.mirror).await.is_empty(), "{nsid}: {why}"); + } +} + +#[tokio::test] +async fn every_mirror_refusal_reads_the_knot() { + for status in [400, 403, 404, 503] { + let h = Harness::with_mirror().await; + h.mount_mirror( + "sh.tangled.git.temp.listBranches", + ResponseTemplate::new(status).set_body_json(json!({"error": "BadRequest"})), + ) + .await; + h.mount_knot("sh.tangled.repo.branches").await; + assert_eq!( + h.served_by("sh.tangled.repo.branches", "limit=500").await, + FROM_KNOT, + "the knot must answer after a mirror {status}", + ); + } + + let h = Harness::new(Mirror::Unreachable, Some(REPO_DID)).await; + h.mount_knot("sh.tangled.repo.branches").await; + assert_eq!( + h.served_by("sh.tangled.repo.branches", "").await, + FROM_KNOT, + "the knot must answer when the mirror is unreachable", + ); +} + +#[tokio::test] +async fn a_ranged_or_conditional_request_skips_the_mirror() { + for (header, value) in [ + ("range", "bytes=0-99"), + ("if-range", "bytes=0-99"), + ("if-none-match", "\"cafe\""), + ("if-modified-since", "Wed, 01 Jul 2026 00:00:00 GMT"), + ] { + let h = Harness::with_mirror().await; + h.mount_mirror("sh.tangled.git.temp.getTree", ok_from_mirror()) + .await; + h.mount_knot("sh.tangled.repo.tree").await; + + assert_eq!( + h.served_by_with("sh.tangled.repo.tree", "ref=main", &[(header, value)]) + .await, + FROM_KNOT, + "only the knot can answer a {header} it issued", + ); + assert!(paths(&h.mirror).await.is_empty(), "{header}"); + } +} + +#[tokio::test] +async fn a_request_reads_the_knot_when_bobbin_wont_ask_the_mirror() { + #[rustfmt::skip] + let unasked = [ + (Mirror::Live, None, "the mirror keys on a repoDid this record doesn't have"), + (Mirror::Off, Some(REPO_DID), "an unset mirror.url leaves every call on the knot"), + ]; + for (setting, repo_did, why) in unasked { + let h = Harness::new(setting, repo_did).await; + h.mount_mirror("sh.tangled.git.temp.getTree", ok_from_mirror()) + .await; + h.mount_knot("sh.tangled.repo.tree").await; + + assert_eq!( + h.served_by("sh.tangled.repo.tree", "ref=main").await, + FROM_KNOT, + "{why}", + ); + assert!(paths(&h.mirror).await.is_empty(), "{why}"); + } +} + +#[tokio::test] +async fn the_knot_keyed_endpoints_never_read_the_mirror() { + let h = Harness::with_mirror().await; + Mock::given(method("GET")) + .and(path("/xrpc/sh.tangled.knot.version")) + .respond_with(ResponseTemplate::new(200).set_body_raw(FROM_KNOT, "application/json")) + .mount(&h.knot) + .await; + + let target = format!("/xrpc/sh.tangled.knot.version?knot={}", enc(&h.knot.uri())); + let resp = router(h.state.clone()) + .oneshot( + Request::builder() + .uri(target) + .extension(ConnectInfo(SOCKET)) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + assert!(paths(&h.mirror).await.is_empty()); +} diff --git a/bobbin/example.toml b/bobbin/example.toml index e9646645..081dc5c0 100644 --- a/bobbin/example.toml +++ b/bobbin/example.toml @@ -162,6 +162,10 @@ # Default value: true #require_https = true +[mirror] +# Can also be specified via environment variable `BOBBIN_MIRROR_URL`. +#url = + [log] # Log emitter format. `text` produces human-readable output for local # development. `json` emits one structured object per line for log diff --git a/knotmirror/xrpc/proxy.go b/knotmirror/xrpc/proxy.go index 1fd99dba..b903e09b 100644 --- a/knotmirror/xrpc/proxy.go +++ b/knotmirror/xrpc/proxy.go @@ -7,6 +7,7 @@ import ( "fmt" "io" "maps" + "net" "net/http" "net/url" "path" @@ -36,6 +37,8 @@ var mirrorToKnotNSID = map[string]string{ tangled.GitTempGetBlobNSID: tangled.RepoBlobNSID, } +const forwardedForHeader = "X-Forwarded-For" + var hopByHopHeaders = map[string]bool{ "Connection": true, "Keep-Alive": true, @@ -132,6 +135,7 @@ func (x *Xrpc) proxyToKnot(w http.ResponseWriter, r *http.Request, repoDid synta x.logger.Warn("proxy: failed to build request", "target", target, "err", err) return false } + req.Header.Set(forwardedForHeader, forwardedFor(r)) resp, err := x.httpClient.Do(req) if err != nil { @@ -157,6 +161,17 @@ func (x *Xrpc) proxyToKnot(w http.ResponseWriter, r *http.Request, repoDid synta return true } +func forwardedFor(r *http.Request) string { + peer := r.RemoteAddr + if host, _, err := net.SplitHostPort(r.RemoteAddr); err == nil { + peer = host + } + chain := lo.Filter(r.Header.Values(forwardedForHeader), func(entry string, _ int) bool { + return strings.TrimSpace(entry) != "" + }) + return strings.Join(append(chain, peer), ", ") +} + func (x *Xrpc) forwardSuspended(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { repoDid, err := syntax.ParseDID(r.URL.Query().Get("repo")) diff --git a/knotmirror/xrpc/proxy_test.go b/knotmirror/xrpc/proxy_test.go new file mode 100644 index 00000000..541c64f5 --- /dev/null +++ b/knotmirror/xrpc/proxy_test.go @@ -0,0 +1,38 @@ +package xrpc + +import ( + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestForwardedForAppendsThePeerToTheChain(t *testing.T) { + cases := []struct { + name string + remote string + chain []string + want string + }{ + {"a direct caller is the whole chain", "203.0.113.7:52344", nil, "203.0.113.7"}, + {"a portless remote address passes through", "203.0.113.7", nil, "203.0.113.7"}, + {"bobbin's client address stays left of bobbin", "198.51.100.4:41000", []string{"203.0.113.7"}, "203.0.113.7, 198.51.100.4"}, + {"a chain split across header lines joins into a single value", "198.51.100.4:41000", []string{"203.0.113.7", "192.0.2.9"}, "203.0.113.7, 192.0.2.9, 198.51.100.4"}, + {"a blank entry never leaves a gap the knot has to skip", "198.51.100.4:41000", []string{"", " ", "203.0.113.7"}, "203.0.113.7, 198.51.100.4"}, + {"an ipv6 peer loses its port and keeps its colons", "[2001:db8::5]:41000", []string{"203.0.113.7"}, "203.0.113.7, 2001:db8::5"}, + {"a forged entry stays left of the address that sent it", "203.0.113.7:52344", []string{"192.0.2.9"}, "192.0.2.9, 203.0.113.7"}, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + r := httptest.NewRequest("GET", "/xrpc/sh.tangled.git.temp.getTree?repo=did:plc:limpet", nil) + r.RemoteAddr = c.remote + for _, entry := range c.chain { + r.Header.Add(forwardedForHeader, entry) + } + + assert.Equal(t, c.want, forwardedFor(r)) + assert.Equal(t, c.chain, r.Header.Values(forwardedForHeader), "the caller's own header must survive the read") + }) + } +}