From 7ea260f96311cbbff52244f1c2836fa9b54976c6 Mon Sep 17 00:00:00 2001 From: Lewis Date: Mon, 4 May 2026 18:29:01 +0300 Subject: [PATCH] feat(xrpc): knot proxy in AppState + proxied routes Lewis: May this revision serve well! --- crates/bobbin/Cargo.toml | 1 + crates/bobbin/src/main.rs | 16 +- crates/edge-index/src/lib.rs | 4 +- crates/ingest/src/lib.rs | 37 ++++- crates/xrpc/Cargo.toml | 2 + crates/xrpc/src/lib.rs | 275 +++++++++++++++++++++++++++++-- crates/xrpc/tests/aggregation.rs | 5 +- crates/xrpc/tests/cold_start.rs | 2 + 8 files changed, 307 insertions(+), 35 deletions(-) diff --git a/crates/bobbin/Cargo.toml b/crates/bobbin/Cargo.toml index 8cc6fd4..55b76ed 100644 --- a/crates/bobbin/Cargo.toml +++ b/crates/bobbin/Cargo.toml @@ -12,6 +12,7 @@ path = "src/main.rs" [dependencies] bobbin-edge-index = { workspace = true } bobbin-ingest = { workspace = true } +bobbin-knot-proxy = { workspace = true } bobbin-record-lru = { workspace = true } bobbin-slingshot-client = { workspace = true } bobbin-types = { workspace = true } diff --git a/crates/bobbin/src/main.rs b/crates/bobbin/src/main.rs index b36882d..026ea57 100644 --- a/crates/bobbin/src/main.rs +++ b/crates/bobbin/src/main.rs @@ -5,6 +5,7 @@ use std::sync::Arc; use anyhow::{Context, anyhow}; use bobbin_edge_index::{CoverageWatch, EdgeStore, HydrantCursor}; use bobbin_ingest::{IngestConfig, RepoDidResolver, ResolveError, run as run_ingest}; +use bobbin_knot_proxy::{KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore, RecordStore}; use bobbin_slingshot_client::{SlingshotClient, SlingshotError}; use bobbin_types::record::RecordBody; @@ -22,13 +23,15 @@ const REPO_NSID: &str = "sh.tangled.repo"; #[tokio::main] async fn main() -> anyhow::Result<()> { tracing_subscriber::fmt() - .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"))) + .with_env_filter( + EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")), + ) .init(); - let hydrant_url = env::var("BOBBIN_HYDRANT_URL") - .unwrap_or_else(|_| "http://127.0.0.1:13010".into()); - let slingshot_url = env::var("BOBBIN_SLINGSHOT_URL") - .unwrap_or_else(|_| "http://127.0.0.1:13011".into()); + let hydrant_url = + env::var("BOBBIN_HYDRANT_URL").unwrap_or_else(|_| "http://127.0.0.1:13010".into()); + let slingshot_url = + env::var("BOBBIN_SLINGSHOT_URL").unwrap_or_else(|_| "http://127.0.0.1:13011".into()); let bind: SocketAddr = env::var("BOBBIN_BIND") .unwrap_or_else(|_| "127.0.0.1:8090".into()) .parse()?; @@ -46,6 +49,7 @@ async fn main() -> anyhow::Result<()> { let slingshot = SlingshotClient::new(Url::parse(&slingshot_url)?)?; let edges = Arc::new(EdgeStore::new()); let coverage = Arc::new(CoverageWatch::new()); + let knots = Arc::new(KnotProxy::new(KnotProxyConfig::default())?); let ingest_cfg = IngestConfig { hydrant_base: Url::parse(&hydrant_url)?, @@ -62,7 +66,7 @@ async fn main() -> anyhow::Result<()> { run_ingest(ingest_cfg, ingest_edges, ingest_coverage, ingest_resolver).await }); - let state = AppState::new(records, slingshot, edges, coverage); + let state = AppState::new(records, slingshot, edges, coverage, knots); let app = router(state); tracing::info!(%bind, %hydrant_url, %slingshot_url, "bobbin listening"); diff --git a/crates/edge-index/src/lib.rs b/crates/edge-index/src/lib.rs index d5e9151..2dc89e8 100644 --- a/crates/edge-index/src/lib.rs +++ b/crates/edge-index/src/lib.rs @@ -76,7 +76,9 @@ impl PageCursor { } pub fn from_token(raw: Option<&str>) -> Result { - raw.map_or(Ok(Self::Start), |t| SourceId::decode_token(t).map(Self::After)) + raw.map_or(Ok(Self::Start), |t| { + SourceId::decode_token(t).map(Self::After) + }) } } diff --git a/crates/ingest/src/lib.rs b/crates/ingest/src/lib.rs index 676c9a3..751bb47 100644 --- a/crates/ingest/src/lib.rs +++ b/crates/ingest/src/lib.rs @@ -3,8 +3,8 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use bobbin_edge_index::{Coverage, CoverageWatch, EdgeStore, HydrantCursor, PromotionSignal}; use bobbin_types::edges::{Edge, ExtractError, Record}; -use futures::stream::{self, StreamExt, TryStreamExt}; use futures::SinkExt; +use futures::stream::{self, StreamExt, TryStreamExt}; use jacquard_common::DefaultStr; use jacquard_common::types::did::Did; use jacquard_common::types::ident::AtIdentifier; @@ -129,8 +129,14 @@ pub async fn run( let mut backoff = RECONNECT_INITIAL_DELAY; loop { let cursor = next_connect_cursor(coverage.snapshot(), config.start_cursor); - let SessionEnd { outcome, error } = - run_session(&config, cursor, store.clone(), coverage.clone(), resolver.clone()).await; + let SessionEnd { outcome, error } = run_session( + &config, + cursor, + store.clone(), + coverage.clone(), + resolver.clone(), + ) + .await; match (outcome, &error) { (SessionOutcome::Progressed, None) => { info!("hydrant stream closed after delivering frames, reconnecting") @@ -176,15 +182,22 @@ async fn run_session( ) -> SessionEnd { let url = match config.stream_url(cursor) { Ok(u) => u, - Err(e) => return SessionEnd { outcome: SessionOutcome::Empty, error: Some(e) }, + Err(e) => { + return SessionEnd { + outcome: SessionOutcome::Empty, + error: Some(e), + }; + } }; info!(%url, "connecting to hydrant /stream"); let (mut ws, _resp) = match tokio_tungstenite::connect_async(url.as_str()).await { Ok(pair) => pair, - Err(e) => return SessionEnd { - outcome: SessionOutcome::Empty, - error: Some(IngestError::Transport(e)), - }, + Err(e) => { + return SessionEnd { + outcome: SessionOutcome::Empty, + error: Some(IngestError::Transport(e)), + }; + } }; let mut outcome = SessionOutcome::Empty; let mut pinger = interval(PING_INTERVAL); @@ -198,7 +211,13 @@ async fn run_session( let processor_resolver = resolver.clone(); let processor = tokio::spawn(async move { while let Some(frame) = frame_rx.recv().await { - handle_frame(frame, &processor_store, &processor_coverage, &*processor_resolver).await; + handle_frame( + frame, + &processor_store, + &processor_coverage, + &*processor_resolver, + ) + .await; } }); diff --git a/crates/xrpc/Cargo.toml b/crates/xrpc/Cargo.toml index 34997b9..2f30625 100644 --- a/crates/xrpc/Cargo.toml +++ b/crates/xrpc/Cargo.toml @@ -10,10 +10,12 @@ bobbin-types = { workspace = true } bobbin-edge-index = { workspace = true } bobbin-record-lru = { workspace = true } bobbin-slingshot-client = { workspace = true } +bobbin-knot-proxy = { workspace = true } jacquard-common = { workspace = true } axum = { workspace = true } futures = { workspace = true } +http = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } diff --git a/crates/xrpc/src/lib.rs b/crates/xrpc/src/lib.rs index 304da6e..ac78e74 100644 --- a/crates/xrpc/src/lib.rs +++ b/crates/xrpc/src/lib.rs @@ -1,15 +1,26 @@ +use std::future::Future; use std::sync::Arc; use axum::{ Router, + body::Body, extract::{FromRequestParts, Query, State, rejection::QueryRejection}, - http::{StatusCode, request::Parts}, + http::{ + 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, + }, + request::Parts, + }, response::{IntoResponse, Json, Response}, routing::get, }; use bobbin_edge_index::{ Coverage, CoverageWatch, CursorParseError, EdgePage, EdgeStore, PageCursor, PageLimit, SourceId, }; +use bobbin_knot_proxy::{KnotHost, KnotProxy, KnotProxyError, ProxyResponse, RepoSlug}; use bobbin_record_lru::RecordStore; use bobbin_slingshot_client::{SlingshotClient, SlingshotError}; use bobbin_types::ids::{EdgeKey, nsid_static}; @@ -24,10 +35,11 @@ use bobbin_types::sh_tangled::repo::issue::{Issue, IssueGetRecordOutput, IssueRe use bobbin_types::sh_tangled::repo::pull::{Pull, PullGetRecordOutput, PullRecord}; use bobbin_types::sh_tangled::repo::{Repo, RepoGetRecordOutput, RepoRecord}; use futures::stream::{self, StreamExt, TryStreamExt}; -use jacquard_common::DefaultStr; +use jacquard_common::types::did::Did; use jacquard_common::types::ident::AtIdentifier; use jacquard_common::types::string::{AtUri, Cid}; use jacquard_common::xrpc::XrpcResp; +use jacquard_common::{DefaultStr, IntoStatic}; use serde::{Deserialize, Serialize}; use thiserror::Error; @@ -40,6 +52,7 @@ pub struct AppState { pub slingshot: SlingshotClient, pub edges: Arc, pub coverage: Arc, + pub knots: Arc, } impl AppState { @@ -48,12 +61,14 @@ impl AppState { slingshot: SlingshotClient, edges: Arc, coverage: Arc, + knots: Arc, ) -> Self { Self { records, slingshot, edges, coverage, + knots, } } } @@ -80,9 +95,82 @@ pub fn router(state: AppState) -> Router { "/xrpc/sh.tangled.repo.issue.countComments", get(count_issue_comments), ) + .merge(knot_proxied_routes()) .with_state(state) } +const REPO_PROXIED_NSIDS: &[&str] = &[ + "sh.tangled.repo.archive", + "sh.tangled.repo.blob", + "sh.tangled.repo.branch", + "sh.tangled.repo.branches", + "sh.tangled.repo.compare", + "sh.tangled.repo.describeRepo", + "sh.tangled.repo.diff", + "sh.tangled.repo.getDefaultBranch", + "sh.tangled.repo.languages", + "sh.tangled.repo.listSecrets", + "sh.tangled.repo.log", + "sh.tangled.repo.tag", + "sh.tangled.repo.tags", + "sh.tangled.repo.tree", +]; + +const KNOT_PROXIED_NSIDS: &[&str] = &[ + "sh.tangled.owner", + "sh.tangled.knot.version", + "sh.tangled.knot.listKeys", +]; + +const PASSTHROUGH_HEADERS: &[&HeaderName] = &[ + &CONTENT_TYPE, + &CONTENT_LENGTH, + &CONTENT_ENCODING, + &ETAG, + &CACHE_CONTROL, + &LAST_MODIFIED, + &CONTENT_DISPOSITION, + &ACCEPT_RANGES, + &CONTENT_RANGE, +]; + +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"; + +type ProxyParams = Vec<(String, String)>; + +fn knot_proxied_routes() -> Router { + let with_repo = register_proxied(Router::new(), REPO_PROXIED_NSIDS, proxy_repo_handler); + register_proxied(with_repo, KNOT_PROXIED_NSIDS, proxy_knot_handler) +} + +fn register_proxied( + router: Router, + nsids: &[&'static str], + handler: H, +) -> Router +where + H: Fn(AppState, HeaderMap, ProxyParams, &'static str) -> Fut + Clone + Send + Sync + 'static, + Fut: Future> + Send + 'static, +{ + nsids.iter().fold(router, |router, &nsid| { + let handler = handler.clone(); + router.route( + &format!("/xrpc/{nsid}"), + get( + move |State(state): State, + headers: HeaderMap, + Query(params): Query| { + handler(state, headers, params, nsid) + }, + ), + ) + }) +} + #[derive(Clone, Debug, Deserialize)] #[serde(transparent)] pub struct RawAtUriParam(String); @@ -249,9 +337,8 @@ fn map_slingshot(err: SlingshotError) -> XrpcError { } } -fn parse_uri(raw: &RawAtUriParam) -> Result, XrpcError> { - AtUri::::new_owned(raw.as_str()) - .map_err(|e| XrpcError::InvalidParams(format!("uri: {e}"))) +fn parse_uri(raw: &str) -> Result, XrpcError> { + AtUri::::new_owned(raw).map_err(|e| XrpcError::InvalidParams(format!("uri: {e}"))) } #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -281,11 +368,8 @@ impl HasSubject for IssueCommentRecord { const SHAPE: SubjectShape = SubjectShape::Collection("sh.tangled.repo.issue"); } -fn parse_subject( - raw: &RawAtUriParam, - shape: SubjectShape, -) -> Result, XrpcError> { - let uri = parse_uri(raw)?; +fn parse_subject(raw: &RawAtUriParam, shape: SubjectShape) -> Result, XrpcError> { + let uri = parse_uri(raw.as_str())?; if matches!(uri.authority(), AtIdentifier::Handle(_)) { return Err(XrpcError::InvalidParams( "subject authority must be a did, not a handle".into(), @@ -339,7 +423,7 @@ async fn resolve( state: &AppState, expected: ExpectedNsid, uri: AtUri, -) -> Result, XrpcError> { +) -> Result<(Arc, Did), XrpcError> { let collection = uri .collection() .ok_or_else(|| XrpcError::InvalidParams("uri missing collection".into()))?; @@ -353,7 +437,7 @@ async fn resolve( let rkey = uri .rkey() .ok_or_else(|| XrpcError::InvalidParams("uri missing rkey".into()))?; - let did = match uri.authority() { + let did_ref = match uri.authority() { AtIdentifier::Did(d) => d, AtIdentifier::Handle(_) => { return Err(XrpcError::InvalidParams( @@ -361,18 +445,19 @@ async fn resolve( )); } }; + let did: Did = did_ref.clone().into_static(); if let Some(hit) = state.records.get(&uri) { - return Ok(hit); + return Ok((hit, did)); } let body = state .slingshot - .get_record(&did, &collection, &rkey) + .get_record(&did_ref, &collection, &rkey) .await .map_err(map_slingshot)?; verify_type_tag(&body, expected)?; state.records.put(uri, body.clone()); - Ok(body) + Ok((body, did)) } #[derive(Deserialize)] @@ -402,8 +487,8 @@ where R: XrpcResp, V: serde::de::DeserializeOwned, { - let parsed = parse_uri(uri)?; - let body = resolve(state, ExpectedNsid::new(R::NSID), parsed).await?; + let parsed = parse_uri(uri.as_str())?; + let (body, _did) = resolve(state, ExpectedNsid::new(R::NSID), parsed).await?; let value: V = serde_json::from_slice(&body.value).map_err(|e| XrpcError::InvalidRecord(e.to_string()))?; Ok((body, value)) @@ -470,7 +555,7 @@ where let EdgePage { items, next } = state.edges.list(&key, cursor, limit); let items = stream::iter(items) .map(|uri| async move { - let body = resolve(state, ExpectedNsid::new(R::NSID), uri.clone()) + let (body, _did) = resolve(state, ExpectedNsid::new(R::NSID), uri.clone()) .await .map_err(|e| match e { XrpcError::NotFound => XrpcError::UpstreamGone(uri.as_ref().to_owned()), @@ -579,3 +664,157 @@ async fn count_issue_comments( ) -> Result, XrpcError> { count_for::(&state, q).map(Json) } + +fn map_proxy_error(err: KnotProxyError) -> XrpcError { + match err { + KnotProxyError::CircuitOpen => { + XrpcError::UpstreamUnavailable("knot circuit breaker open".into()) + } + KnotProxyError::BlockedHost { host, reason } => { + XrpcError::InvalidRecord(format!("knot host {host} is {reason} address space")) + } + KnotProxyError::PlaintextHttp { host } => { + XrpcError::InvalidRecord(format!("knot host {host} requires https")) + } + KnotProxyError::Connect(e) => XrpcError::UpstreamUnavailable(format!("connect: {e}")), + KnotProxyError::Timeout(e) => { + XrpcError::UpstreamUnavailable(format!("upstream timeout: {e}")) + } + KnotProxyError::Redirect(e) => XrpcError::UpstreamUnavailable(format!("redirect: {e}")), + KnotProxyError::Transport(e) => XrpcError::UpstreamUnavailable(format!("transport: {e}")), + KnotProxyError::Upstream(s) => XrpcError::UpstreamUnavailable(format!("status {s}")), + } +} + +fn validate_client_supplied_knot(state: &AppState, host: &KnotHost) -> Result<(), XrpcError> { + let host_str = || host.url().host_str().unwrap_or_default().to_owned(); + if state.knots.requires_https() && host.url().scheme() != "https" { + return Err(XrpcError::InvalidParams(format!( + "knot host {} must be https", + host_str(), + ))); + } + if state.knots.allows_private_hosts() { + return Ok(()); + } + match host.private_literal_reason() { + None => Ok(()), + Some(reason) => Err(XrpcError::InvalidParams(format!( + "knot host {} blocked: {} address space", + host_str(), + reason, + ))), + } +} + +async fn resolve_knot_target( + state: &AppState, + repo_uri_raw: &str, +) -> Result<(KnotHost, RepoSlug), XrpcError> { + let repo_uri = parse_uri(repo_uri_raw)?; + let (body, did) = resolve(state, ExpectedNsid::new(RepoRecord::NSID), repo_uri).await?; + let value: Repo = serde_json::from_slice(&body.value) + .map_err(|e| XrpcError::InvalidRecord(format!("decode repo record: {e}")))?; + let host = KnotHost::parse(value.knot.as_ref()) + .map_err(|e| XrpcError::InvalidRecord(format!("knot field: {e}")))?; + let slug = RepoSlug::new(did.as_ref(), value.name.as_ref()) + .map_err(|e| XrpcError::InvalidRecord(format!("repo slug: {e}")))?; + Ok((host, slug)) +} + +fn filter_request_headers(client: &HeaderMap) -> HeaderMap { + FORWARDED_REQUEST_HEADERS + .iter() + .fold(HeaderMap::new(), |mut acc, name| { + if let Some(value) = client.get(*name) { + acc.insert((*name).clone(), value.clone()); + } + acc + }) +} + +fn upstream_to_axum(resp: ProxyResponse) -> Response { + let status = resp.status(); + let upstream_headers = resp.headers().clone(); + let body = Body::from_stream(resp.into_body_stream()); + let mut response = Response::builder() + .status(status) + .body(body) + .expect("response body construction must succeed"); + let response_headers = response.headers_mut(); + PASSTHROUGH_HEADERS.iter().for_each(|name| { + if let Some(value) = upstream_headers.get(*name) { + response_headers.insert((*name).clone(), value.clone()); + } + }); + response +} + +async fn dispatch_proxy( + state: AppState, + headers: HeaderMap, + nsid: &'static str, + 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); + let upstream = state + .knots + .forward(&host, nsid, &forward, allowed) + .await + .map_err(map_proxy_error)?; + Ok(upstream_to_axum(upstream)) +} + +fn extract_param( + params: ProxyParams, + key: &str, +) -> Result, XrpcError> { + let (matching, rest): (ProxyParams, ProxyParams) = + params.into_iter().partition(|(k, _)| k == key); + match matching.as_slice() { + [] => Ok(None), + [_] => Ok(matching.into_iter().next().map(|(_, v)| (v, rest))), + _ => Err(XrpcError::InvalidParams(format!( + "{key} parameter must appear at most once, got {}", + matching.len(), + ))), + } +} + +async fn proxy_repo_handler( + state: AppState, + headers: HeaderMap, + params: ProxyParams, + nsid: &'static str, +) -> Result { + let (repo_raw, rest) = extract_param(params, REPO_PARAM)? + .ok_or_else(|| XrpcError::InvalidParams("missing repo".into()))?; + let (host, slug) = resolve_knot_target(&state, &repo_raw).await?; + let forward = rest + .into_iter() + .chain(std::iter::once(( + REPO_PARAM.to_owned(), + slug.as_str().to_owned(), + ))) + .collect(); + dispatch_proxy(state, headers, nsid, host, forward).await +} + +async fn proxy_knot_handler( + state: AppState, + headers: HeaderMap, + params: ProxyParams, + nsid: &'static str, +) -> Result { + let (knot_raw, forward) = 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, nsid, host, forward).await +} diff --git a/crates/xrpc/tests/aggregation.rs b/crates/xrpc/tests/aggregation.rs index 32771d7..610fefe 100644 --- a/crates/xrpc/tests/aggregation.rs +++ b/crates/xrpc/tests/aggregation.rs @@ -2,6 +2,7 @@ use std::sync::Arc; use axum::body::{Body, to_bytes}; use bobbin_edge_index::{Coverage, CoverageWatch, EdgeStore, HydrantCursor, SourceId}; +use bobbin_knot_proxy::{KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; use bobbin_slingshot_client::SlingshotClient; use bobbin_types::edges::Edge; @@ -45,6 +46,7 @@ impl Harness { SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(), edges.clone(), coverage.clone(), + Arc::new(KnotProxy::new(KnotProxyConfig::default()).unwrap()), ); Self { server, @@ -1075,7 +1077,8 @@ async fn extractor_to_xrpc_round_trip_for_star() { .expect("extract") .into_iter() .for_each(|e| h.edges.add(e)); - h.mount(source_did, "sh.tangled.feed.star", rkey, body).await; + h.mount(source_did, "sh.tangled.feed.star", rkey, body) + .await; let app = router(h.state.clone()); let (status, json) = json_response( diff --git a/crates/xrpc/tests/cold_start.rs b/crates/xrpc/tests/cold_start.rs index 15c9c0d..4474267 100644 --- a/crates/xrpc/tests/cold_start.rs +++ b/crates/xrpc/tests/cold_start.rs @@ -2,6 +2,7 @@ use std::sync::Arc; use axum::body::{Body, to_bytes}; use bobbin_edge_index::{CoverageWatch, EdgeStore}; +use bobbin_knot_proxy::{KnotProxy, KnotProxyConfig}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; use bobbin_slingshot_client::SlingshotClient; use bobbin_xrpc::{AppState, router}; @@ -22,6 +23,7 @@ async fn fresh_app(server_uri: &str) -> AppState { SlingshotClient::new(Url::parse(server_uri).unwrap()).unwrap(), Arc::new(EdgeStore::new()), Arc::new(CoverageWatch::new()), + Arc::new(KnotProxy::new(KnotProxyConfig::default()).unwrap()), ) } -- 2.51.2