Something went wrong. Try again.
A downstream fork of Tranquil PDS for use in margin.cafe
Something went wrong. Try again.
19 kB · 475 lines
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476use std::collections::HashSet;use std::convert::Infallible;use std::sync::LazyLock;
use crate::api::error::ApiError;use crate::api::proxy_client::{is_ssrf_safe, proxy_client};use crate::state::AppState;use crate::types::{Did, Nsid};use crate::util::get_header_str;use axum::{ body::Bytes, extract::{RawQuery, Request, State}, handler::Handler, http::{HeaderMap, Method}, response::{IntoResponse, Response},};use futures_util::future::Either;use tower::{Service, util::BoxCloneSyncService};use tracing::{error, info, warn};
static PROTECTED_METHODS: LazyLock<HashSet<&'static str>> = LazyLock::new(|| { let mut methods: HashSet<&str> = [ "com.atproto.admin.deleteAccount", "com.atproto.admin.disableAccountInvites", "com.atproto.admin.disableInviteCodes", "com.atproto.admin.enableAccountInvites", "com.atproto.admin.getAccountInfo", "com.atproto.admin.getAccountInfos", "com.atproto.admin.getInviteCodes", "com.atproto.admin.getSubjectStatus", "com.atproto.admin.searchAccounts", "com.atproto.admin.sendEmail", "com.atproto.admin.updateAccountEmail", "com.atproto.admin.updateAccountHandle", "com.atproto.admin.updateAccountPassword", "com.atproto.admin.updateSubjectStatus", "com.atproto.identity.getRecommendedDidCredentials", "com.atproto.identity.requestPlcOperationSignature", "com.atproto.identity.signPlcOperation", "com.atproto.identity.submitPlcOperation", "com.atproto.identity.updateHandle", "com.atproto.repo.applyWrites", "com.atproto.repo.createRecord", "com.atproto.repo.deleteRecord", "com.atproto.repo.importRepo", "com.atproto.repo.putRecord", "com.atproto.repo.uploadBlob", "com.atproto.server.activateAccount", "com.atproto.server.checkAccountStatus", "com.atproto.server.confirmEmail", "com.atproto.server.confirmSignup", "com.atproto.server.createAccount", "com.atproto.server.createAppPassword", "com.atproto.server.createInviteCode", "com.atproto.server.createInviteCodes", "com.atproto.server.createSession", "com.atproto.server.createTotpSecret", "com.atproto.server.deactivateAccount", "com.atproto.server.deleteAccount", "com.atproto.server.deletePasskey", "com.atproto.server.deleteSession", "com.atproto.server.describeServer", "com.atproto.server.disableTotp", "com.atproto.server.enableTotp", "com.atproto.server.finishPasskeyRegistration", "com.atproto.server.getAccountInviteCodes", "com.atproto.server.getServiceAuth", "com.atproto.server.getSession", "com.atproto.server.getTotpStatus", "com.atproto.server.listAppPasswords", "com.atproto.server.listPasskeys", "com.atproto.server.refreshSession", "com.atproto.server.regenerateBackupCodes", "com.atproto.server.requestAccountDelete", "com.atproto.server.requestEmailConfirmation", "com.atproto.server.requestEmailUpdate", "com.atproto.server.requestPasswordReset", "com.atproto.server.resendMigrationVerification", "com.atproto.server.resendVerification", "com.atproto.server.reserveSigningKey", "com.atproto.server.resetPassword", "com.atproto.server.revokeAppPassword", "com.atproto.server.startPasskeyRegistration", "com.atproto.server.updateEmail", "com.atproto.server.updatePasskey", "com.atproto.server.verifyMigrationEmail", "com.atproto.sync.getBlob", "com.atproto.sync.getBlocks", "com.atproto.sync.getCheckout", "com.atproto.sync.getHead", "com.atproto.sync.getLatestCommit", "com.atproto.sync.getRecord", "com.atproto.sync.getRepo", "com.atproto.sync.getRepoStatus", "com.atproto.sync.listBlobs", "com.atproto.sync.listRepos", "com.atproto.sync.notifyOfUpdate", "com.atproto.sync.requestCrawl", "com.atproto.sync.subscribeRepos", "com.atproto.temp.checkSignupQueue", "com.atproto.temp.dereferenceScope", ] .into_iter() .collect(); // BSKY: the Bluesky preferences API must be implemented by PDSs if cfg!(feature = "bsky-support") { methods.insert("app.bsky.actor.getPreferences"); methods.insert("app.bsky.actor.putPreferences"); }; methods});
fn target_origin(target_url: &str) -> Option<String> { let url = reqwest::Url::parse(target_url).ok()?; let origin = url.origin().ascii_serialization(); (origin != "null").then_some(origin)}
fn is_protected_method(method: &str) -> bool { PROTECTED_METHODS.contains(method)}
/// Fetch the `feed` generator record from the AppView and return its `did`.#[cfg(feature = "bsky-support")]async fn resolve_feed_generator_did(appview_url: &str, query: Option<&str>) -> Option<Did> { #[derive(serde::Deserialize)] struct GetFeedQuery { feed: String, }
let feed = serde_urlencoded::from_str::<GetFeedQuery>(query?) .ok()? .feed; let at_uri = crate::types::AtUri::new(feed).ok()?; let repo = at_uri.did()?; let collection = at_uri.collection()?; let rkey = at_uri.rkey()?; let target_url = format!("{appview_url}/xrpc/com.atproto.repo.getRecord");
is_ssrf_safe(&target_url).await.ok()?; let resp = proxy_client() .get(target_url) .query(&[("repo", repo), ("collection", collection), ("rkey", rkey)]) .send() .await .ok()?; if !resp.status().is_success() { warn!(status = %resp.status(), "getFeed proxy: getRecord for feed generator failed"); return None; } let body: serde_json::Value = resp.json().await.ok()?; body.get("value")? .get("did")? .as_str() .and_then(|s| Did::new(s).ok())}
pub struct XrpcProxyLayer { state: AppState,}
impl XrpcProxyLayer { pub fn new(state: AppState) -> Self { XrpcProxyLayer { state } }}
impl<S> tower_layer::Layer<S> for XrpcProxyLayer { type Service = XrpcProxyingService<S>;
fn layer(&self, inner: S) -> Self::Service { XrpcProxyingService { inner, // TODO(nel): make our own service here instead of boxing a HandlerService handler: BoxCloneSyncService::new(proxy_handler.with_state(self.state.clone())), } }}
#[derive(Clone)]pub struct XrpcProxyingService<S> { inner: S, handler: BoxCloneSyncService<Request, Response, Infallible>,}
impl<S: Service<Request, Response = Response, Error = Infallible>> Service<Request> for XrpcProxyingService<S>{ type Response = Response;
type Error = Infallible;
type Future = Either< <BoxCloneSyncService<Request, Response, Infallible> as Service<Request>>::Future, S::Future, >;
fn poll_ready( &mut self, cx: &mut std::task::Context<'_>, ) -> std::task::Poll<Result<(), Self::Error>> { self.inner.poll_ready(cx) }
fn call(&mut self, req: Request) -> Self::Future { if req.headers().contains_key(http::HeaderName::from( jacquard_common::xrpc::Header::AtprotoProxy, )) { let path = req.uri().path(); let method = path.trim_start_matches("/");
if is_protected_method(method) { return Either::Right(self.inner.call(req)); }
// If the age assurance override is set and this is an age assurance call then we dont want to proxy even if the client requests it #[cfg(feature = "bsky")] if tranquil_config::get().server.age_assurance_override && (path.ends_with("app.bsky.ageassurance.getState") || path.ends_with("app.bsky.unspecced.getAgeAssuranceState")) { return Either::Right(self.inner.call(req)); }
Either::Left(self.handler.call(req)) } else { Either::Right(self.inner.call(req)) } }}
async fn proxy_handler( State(state): State<AppState>, uri: http::Uri, method_verb: Method, headers: HeaderMap, RawQuery(query): RawQuery, body: Bytes,) -> Response { // This layer is nested under /xrpc in an axum router so the extracted uri will look like /<method> and thus we can just strip the / let method = uri.path().trim_start_matches("/"); if is_protected_method(method) { warn!(method = %method, "Attempted to proxy protected method"); return ApiError::InvalidRequest(format!("Cannot proxy protected method: {}", method)) .into_response(); }
let Some(proxy_header) = get_header_str(&headers, crate::util::HEADER_ATPROTO_PROXY).map(String::from) else { return ApiError::InvalidRequest("Missing required atproto-proxy header".into()) .into_response(); };
let Some((did, service_id)) = proxy_header.split_once("#") else { return ApiError::InvalidRequest( "Invalid atproto-proxy header. Missing service identifier.".into(), ) .into_response(); }; let Ok(did) = did.parse::<Did>() else { return ApiError::InvalidRequest("Invalid DID in atproto-proxy header".into()) .into_response(); }; let Ok(resolved) = state.did_resolver.resolve_service(&did, service_id).await else { error!(did = %did, service_id = %service_id, "Could not resolve service DID"); return ApiError::UpstreamFailure.into_response(); };
let target_url = match &query { Some(q) => format!("{}/xrpc/{}?{}", resolved.url, method, q), None => format!("{}/xrpc/{}", resolved.url, method), }; let mut auth_header_val = headers.get(http::header::AUTHORIZATION).cloned(); if let Some(extracted) = crate::auth::extract_auth_token_from_header( crate::util::get_header_str(&headers, http::header::AUTHORIZATION), ) { let token = extracted.token; let dpop_proof = crate::util::get_header_str(&headers, crate::util::HEADER_DPOP); let http_uri = crate::util::build_full_url(&format!("/xrpc{}", uri));
match crate::auth::validate_token_with_dpop( state.repos.user.as_ref(), state.repos.oauth.as_ref(), &token, extracted.scheme, dpop_proof, method_verb.as_str(), &http_uri, crate::auth::AccountRequirement::Active, ) .await { Ok(auth_user) => { let Ok(method_nsid) = method.parse::<Nsid>() else { return ApiError::InvalidRequest(format!("Invalid XRPC method: {}", method)) .into_response(); }; if let Err(e) = crate::auth::scope_check::check_rpc_scope( &auth_user.auth_source, auth_user.scope.as_deref(), &resolved.did, &method_nsid, ) { return e.into_response(); }
let key_bytes = match auth_user.key_bytes { Some(kb) => kb, None => match state.repos.user.get_user_info_by_did(&auth_user.did).await { Ok(Some(info)) => match info.key_bytes { Some(key_bytes_enc) => { match crate::config::decrypt_key( &key_bytes_enc, info.encryption_version, ) { Ok(key) => key, Err(e) => { error!(error = ?e, "Failed to decrypt user key for proxy"); return ApiError::UpstreamFailure.into_response(); } } } None => { warn!(did = %auth_user.did, "User has no signing key for proxy"); return ApiError::UpstreamFailure.into_response(); } }, Ok(None) => { warn!(did = %auth_user.did, "User not found for proxy service auth"); return ApiError::UpstreamFailure.into_response(); } Err(e) => { error!(error = ?e, "DB error fetching user key for proxy"); return ApiError::UpstreamFailure.into_response(); } }, };
// BSKY: getFeed must be audienced to the feed generator, not the AppView. let (token_aud, token_lxm) = if cfg!(feature = "bsky-support") && method == "app.bsky.feed.getFeed" { match resolve_feed_generator_did(&resolved.url, query.as_deref()).await { Some(feed_did) => ( feed_did, "app.bsky.feed.getFeedSkeleton" .parse::<Nsid>() .expect("getFeedSkeleton is a valid NSID"), ), None => { warn!( "getFeed proxy: could not resolve feed generator DID; refusing \ to mint an AppView-audienced token" ); return ApiError::InvalidRequest("Could not resolve feed".into()) .into_response(); } } } else { (resolved.did.clone(), method_nsid.clone()) };
match crate::auth::create_service_token( &auth_user.did, &token_aud, Some(&token_lxm), &key_bytes, ) { Ok(new_token) => { if let Ok(val) = axum::http::HeaderValue::from_str(&format!("Bearer {}", new_token)) { auth_header_val = Some(val); } } Err(e) => { error!("Failed to create service token: {:?}", e); return ApiError::UpstreamFailure.into_response(); } } } Err(e) => { info!(error = ?e, "Proxy token validation failed, returning error to client"); let mut response = ApiError::from(e).into_response(); if let Ok(nonce_val) = crate::oauth::verify::generate_dpop_nonce().parse() { response .headers_mut() .insert(crate::util::HEADER_DPOP_NONCE, nonce_val); } return response; } } }
if let Err(e) = is_ssrf_safe(&target_url).await { warn!(did = %did, service_id = %service_id, error = %e, "Refusing unsafe proxy target"); return ApiError::UpstreamFailure.into_response(); } info!("Proxying {} request to {}", method_verb, target_url);
let client = proxy_client(); let mut request_builder = client.request(method_verb.clone(), &target_url); if !matches!(method_verb, Method::GET | Method::HEAD) && let Some(origin) = target_origin(&target_url) { request_builder = request_builder.header(http::header::ORIGIN, origin); } if let Some(val) = auth_header_val { request_builder = request_builder.header(http::header::AUTHORIZATION, val); } request_builder = crate::api::proxy_client::HEADERS_TO_FORWARD .iter() .filter_map(|name| headers.get(name).map(|val| (name, val))) .fold(request_builder, |builder, (name, val)| { builder.header(name.as_str(), val) }); if !body.is_empty() { request_builder = request_builder.body(body); }
match request_builder.send().await { Ok(resp) => { let status = resp.status(); let headers = resp.headers().clone(); let body = match resp.bytes().await { Ok(b) => b, Err(e) => { error!("Error reading proxy response body: {:?}", e); return ApiError::UpstreamUnavailable("Error reading upstream response".into()) .into_response(); } }; let mut response_builder = Response::builder().status(status); response_builder = crate::api::proxy_client::RESPONSE_HEADERS_TO_FORWARD .iter() .filter_map(|name| headers.get(name).map(|val| (name, val))) .fold(response_builder, |builder, (name, val)| { builder.header(name, val) }); match response_builder.body(axum::body::Body::from(body)) { Ok(r) => r, Err(e) => { error!("Error building proxy response: {:?}", e); ApiError::InternalError(None).into_response() } } } Err(e) => { error!("Error sending proxy request: {:?}", e); if e.is_timeout() { ApiError::UpstreamTimeout.into_response() } else { ApiError::UpstreamFailure.into_response() } } }}
#[cfg(test)]mod tests { use super::target_origin;
#[test] fn proxy_origin_matches_resolved_service_target() { assert_eq!( target_origin("https://wisp.place/xrpc/place.wisp.v2.domain.claim"), Some("https://wisp.place".to_string()) ); assert_eq!( target_origin("https://example.com:8443/xrpc/example.method"), Some("https://example.com:8443".to_string()) ); assert_eq!(target_origin("not a URL"), None); }}