didbot_serve/proxy/
mod.rs

1//! `atproto-proxy`: an XRPC call this server does not serve, forwarded to
2//! the service a DID document names.
3//!
4//! The XRPC spec's "Service Proxying". The header names a DID and a service
5//! id, as `did:web:api.bsky.app#bsky_appview` does. This server resolves the
6//! DID, takes the service entry with that id, and forwards the method, the
7//! query and the body there with a service-auth token signed as the calling
8//! account: `iss` the account, `aud` the header's DID and service id, `lxm`
9//! the method. The answer streams back.
10//!
11//! The call is authorized as [`crate::service_auth::authorize`] authorizes
12//! one, before anything is resolved, so a caller with no credential makes
13//! this server fetch nothing. Both fetches go through
14//! [`didbot_http::pinned_client`], and neither may reach this server itself.
15//! Which headers cross in each direction are the reference PDS's
16//! (`packages/pds/src/pipethrough.ts`).
17//!
18//! Each account's calls are bounded by [`ProxyLimits`]: a call over its
19//! rate, or past its own, its tree's or the server's share of calls in
20//! flight, is refused `429` before anything is resolved. A call takes its
21//! place once its body is read, so a slow upload holds none.
22
23mod limits;
24
25pub use limits::{ProxyLimits, ProxySettings};
26
27use std::convert::Infallible;
28use std::time::Duration;
29
30use axum::body::Body;
31use axum::extract::{Request, State};
32use axum::http::{header, HeaderMap, HeaderName, HeaderValue, Method, StatusCode};
33use axum::response::{IntoResponse, Response};
34use didbot_identity::DidDocument;
35use didbot_pds::service_auth::{ServiceAuth, DEFAULT_LIFETIME};
36use didbot_pds::EstopRefusal;
37use http_body_util::BodyExt;
38use serde_json::Value;
39use time::OffsetDateTime;
40use tower::{Layer, ServiceExt};
41use tracing::{info, warn};
42use url::Url;
43
44use crate::error::ApiError;
45use crate::routes::{AppState, MAX_REQUEST_BODY};
46use crate::service_auth::{authorize, is_protected, sign};
47
48/// The request header naming the service a call is for.
49pub const PROXY_HEADER: &str = "atproto-proxy";
50
51/// The most of one proxied answer this server passes back, in bytes.
52///
53/// **Reasonable range: 1 MiB to 64 MiB.** Ten mebibytes, the reference PDS's
54/// default. An answer that declares more is refused before any of it is
55/// read, and one that sends more is cut off where it crosses the bound.
56pub const MAX_PROXIED_RESPONSE: usize = 10 * 1024 * 1024;
57
58/// How long one proxied call may take, from connecting to the last byte of
59/// the answer.
60///
61/// **Reasonable range: 10 to 29 seconds.** Under the router's
62/// `REQUEST_TIMEOUT`, so a service that stalls is answered `UpstreamFailure`
63/// here rather than with that backstop's empty 503.
64pub const PROXY_TIMEOUT: Duration = Duration::from_secs(25);
65
66/// Request headers a proxied call carries on, beside every `x-atproto-*`
67/// header. `Accept-Encoding` is carried too, as `identity` when absent.
68const FORWARDED_REQUEST_HEADERS: [HeaderName; 3] = [
69    header::ACCEPT_LANGUAGE,
70    HeaderName::from_static("atproto-accept-labelers"),
71    HeaderName::from_static("x-bsky-topics"),
72];
73
74/// The headers that describe a procedure's body, carried with it.
75const BODY_HEADERS: [HeaderName; 2] = [header::CONTENT_TYPE, header::CONTENT_ENCODING];
76
77/// Response headers carried back, on a failure as on a success.
78const FORWARDED_RESPONSE_HEADERS: [HeaderName; 3] = [
79    HeaderName::from_static("atproto-repo-rev"),
80    HeaderName::from_static("atproto-content-labelers"),
81    header::RETRY_AFTER,
82];
83
84/// The headers that describe a successful answer's body, carried with it.
85const CONTENT_HEADERS: [HeaderName; 4] = [
86    header::CONTENT_LENGTH,
87    header::CONTENT_ENCODING,
88    header::CONTENT_TYPE,
89    header::CONTENT_LANGUAGE,
90];
91
92/// The router's fallback: a path nothing serves, or an XRPC call this
93/// server does not serve.
94///
95/// An XRPC call carrying [`PROXY_HEADER`] is forwarded. An `app.bsky.*`
96/// call without it goes to [`ProxySettings::app_view`] when one is set.
97/// Any other call is `MethodNotImplemented`. The XRPC half answers cross-origin
98/// like the methods this server serves, because a browser app calls another
99/// service's methods through the account's PDS.
100pub(crate) async fn fallback(State(state): State<AppState>, request: Request) -> Response {
101    if !request.uri().path().starts_with("/xrpc/") {
102        return crate::routes::unknown_route(request.uri());
103    }
104    let forwarding = tower::service_fn(move |request: Request| {
105        let state = state.clone();
106        async move { Ok::<_, Infallible>(forward(&state, request).await) }
107    });
108    match crate::routes::cross_origin_layer()
109        .layer(forwarding)
110        .oneshot(request)
111        .await
112    {
113        Ok(response) => response,
114        Err(never) => match never {},
115    }
116}
117
118/// Forwards one XRPC call, or answers why it is not forwarded.
119async fn forward(state: &AppState, request: Request) -> Response {
120    if !request.headers().contains_key(PROXY_HEADER)
121        && app_view(state, request.uri().path()).is_none()
122    {
123        return crate::routes::unknown_route(request.uri());
124    }
125    match proxy(state, request).await {
126        Ok(response) => response,
127        Err(err) => err.into_response(),
128    }
129}
130
131/// Where an `app.bsky.*` call at `path` goes when it names no service: the
132/// configured [`ProxySettings::app_view`]. `None` for any other method.
133fn app_view<'a>(state: &'a AppState, path: &str) -> Option<&'a str> {
134    path.strip_prefix("/xrpc/")?.strip_prefix("app.bsky.")?;
135    state.proxy.settings().app_view.as_deref()
136}
137
138/// Where a call goes: the DID and the service id an [`PROXY_HEADER`] names.
139struct Target<'a> {
140    /// The header as sent, `did#service`: the token's `aud`, and the
141    /// audience a grant's `rpc:` scope has to reach.
142    reference: &'a str,
143    did: &'a str,
144    service: &'a str,
145}
146
147impl<'a> Target<'a> {
148    /// Reads the header the way the reference PDS's `parseProxyHeader`
149    /// does: one `#`, a DID before it, a service id after it, no spaces.
150    fn parse(value: &'a str) -> Result<Self, ApiError> {
151        let Some((did, service)) = value.split_once('#') else {
152            return Err(ApiError::bad_request(
153                "the atproto-proxy header names no service id",
154            ));
155        };
156        if did.is_empty() {
157            return Err(ApiError::bad_request(
158                "the atproto-proxy header names no DID",
159            ));
160        }
161        if service.is_empty() {
162            return Err(ApiError::bad_request(
163                "the atproto-proxy header names no service id",
164            ));
165        }
166        if service.contains('#') {
167            return Err(ApiError::bad_request(
168                "the atproto-proxy header holds more than one `#`",
169            ));
170        }
171        if value.contains(' ') {
172            return Err(ApiError::bad_request(
173                "the atproto-proxy header may not contain spaces",
174            ));
175        }
176        Ok(Self {
177            reference: value,
178            did,
179            service,
180        })
181    }
182}
183
184async fn proxy(state: &AppState, request: Request) -> Result<Response, ApiError> {
185    state.estop.check_use(EstopRefusal::Operation)?;
186    let (parts, body) = request.into_parts();
187    let path = parts.uri.path();
188    let lxm = path.trim_start_matches("/xrpc/");
189    let http_method = match parts.method {
190        Method::GET => "GET",
191        Method::POST => "POST",
192        _ => {
193            return Err(ApiError::bad_request(
194                "an XRPC call is a GET or a POST, and nothing else is proxied",
195            ))
196        }
197    };
198    didbot_lexicon::nsid_syntax::validate(lxm)
199        .map_err(|_| ApiError::bad_request(format!("`{lxm}` is not an XRPC method")))?;
200    if is_protected(lxm) {
201        return Err(ApiError::new(
202            StatusCode::BAD_REQUEST,
203            "InvalidToken",
204            format!("`{lxm}` manages an account, and a call to it is not proxied"),
205        ));
206    }
207    let header = match parts.headers.get(PROXY_HEADER) {
208        Some(value) => value
209            .to_str()
210            .map_err(|_| ApiError::bad_request("the atproto-proxy header is not text"))?,
211        None => app_view(state, path).ok_or_else(|| {
212            ApiError::bad_request("the call names no service in an atproto-proxy header")
213        })?,
214    };
215    let target = Target::parse(header)?;
216
217    let caller = authorize(
218        state,
219        &parts.headers,
220        http_method,
221        path,
222        Some(lxm),
223        target.reference,
224    )?;
225    let body = axum::body::to_bytes(body, MAX_REQUEST_BODY)
226        .await
227        .map_err(|err| {
228            if std::error::Error::source(&err)
229                .is_some_and(|source| source.is::<http_body_util::LengthLimitError>())
230            {
231                crate::routes::request_too_large()
232            } else {
233                ApiError::bad_request(format!("reading the request body: {err}"))
234            }
235        })?;
236    let held = admit(state, &caller.did, &target, lxm)?;
237    let now = OffsetDateTime::now_utc();
238    let token = sign(
239        state,
240        &caller.did,
241        ServiceAuth {
242            aud: target.reference,
243            lxm: Some(lxm),
244            iat: now,
245            exp: now + DEFAULT_LIFETIME,
246        },
247    )?;
248    let (endpoint, client) = resolve(state, &target).await?;
249
250    let mut url = endpoint;
251    url.set_path(path);
252    url.set_query(parts.uri.query());
253    let mut outbound = client
254        .request(parts.method.clone(), url.clone())
255        .timeout(PROXY_TIMEOUT)
256        .headers(forwarded_request_headers(&parts.headers))
257        .header(header::AUTHORIZATION, format!("Bearer {token}"));
258    if parts.method == Method::POST {
259        for name in BODY_HEADERS {
260            if let Some(value) = parts.headers.get(&name) {
261                outbound = outbound.header(name, value.clone());
262            }
263        }
264        outbound = outbound.body(body);
265    }
266    let upstream = outbound.send().await.map_err(|error| {
267        info!(did = %caller.did, lxm, target = target.reference, %error, "proxied call failed");
268        ApiError::new(
269            StatusCode::BAD_GATEWAY,
270            "UpstreamFailure",
271            format!("{} could not be reached: {error}", target.reference),
272        )
273    })?;
274    let status = upstream.status();
275    info!(
276        did = %caller.did,
277        client = caller.client_id.as_ref().map(didbot_pds::policy::ClientId::as_str),
278        lxm,
279        target = target.reference,
280        status = status.as_u16(),
281        "proxied a call"
282    );
283    if status.is_client_error() || status.is_server_error() {
284        return Err(upstream_error(target.reference, upstream).await);
285    }
286    answer(target.reference, upstream, held)
287}
288
289/// Spends one of `did`'s proxied calls, or answers the refusal.
290///
291/// Logged at WARN, naming the account and the target: a refusal here is an
292/// account calling faster or wider than an agent does.
293fn admit(
294    state: &AppState,
295    did: &str,
296    target: &Target<'_>,
297    lxm: &str,
298) -> Result<limits::InFlight, ApiError> {
299    let tree = tree_of(state, did);
300    let refusal = match state.proxy.admit(did, &tree) {
301        Ok(held) => return Ok(held),
302        Err(refusal) => refusal,
303    };
304    let settings = state.proxy.settings();
305    match refusal {
306        limits::Refusal::Rate(wait) => {
307            warn!(
308                did,
309                target = target.reference,
310                lxm,
311                calls_per_minute = settings.calls_per_minute,
312                "refused a proxied call: the account is over its rate"
313            );
314            // Whole seconds, rounded up, so a client that waits this long
315            // finds a call in its budget.
316            let retry = wait.as_secs() + u64::from(wait.subsec_nanos() > 0);
317            Err(ApiError::rate_limited(format!(
318                "this account may make {} proxied calls a minute; retry in {retry}s",
319                settings.calls_per_minute
320            ))
321            .with_header(header::RETRY_AFTER, retry.max(1).to_string()))
322        }
323        limits::Refusal::InFlight(open) => {
324            warn!(
325                did,
326                target = target.reference,
327                lxm,
328                open,
329                "refused a proxied call: the account has as many open as it may"
330            );
331            Err(ApiError::rate_limited(format!(
332                "this account already has {open} proxied calls open; retry when one finishes"
333            )))
334        }
335        limits::Refusal::Tree(open) => {
336            warn!(
337                did,
338                tree,
339                target = target.reference,
340                lxm,
341                open,
342                "refused a proxied call: the account's tree has as many open as it may"
343            );
344            Err(ApiError::rate_limited(format!(
345                "the accounts under {tree} already have {open} proxied calls open; retry when \
346                 one finishes"
347            )))
348        }
349        limits::Refusal::Full(open) => {
350            warn!(
351                did,
352                target = target.reference,
353                lxm,
354                open,
355                "refused a proxied call: this server has as many open as it will hold"
356            );
357            Err(ApiError::rate_limited(format!(
358                "this server already has {open} proxied calls open; retry shortly"
359            )))
360        }
361    }
362}
363
364/// The account tree `did` is in, named by its topmost account below this
365/// server's own: the account the operator admitted, read up the parent
366/// chain. An account with no parent but this server is its own tree.
367fn tree_of(state: &AppState, did: &str) -> String {
368    let service = state.registry.service_did();
369    state
370        .registry
371        .boundary(did)
372        .ok()
373        .and_then(|chain| chain.into_iter().rev().find(|account| *account != service))
374        .unwrap_or_else(|| did.to_owned())
375}
376
377/// The headers a call carries on to the service; see
378/// [`FORWARDED_REQUEST_HEADERS`].
379fn forwarded_request_headers(headers: &HeaderMap) -> HeaderMap {
380    let mut forwarded = HeaderMap::new();
381    forwarded.insert(
382        header::ACCEPT_ENCODING,
383        headers
384            .get(header::ACCEPT_ENCODING)
385            .cloned()
386            .unwrap_or_else(|| HeaderValue::from_static("identity")),
387    );
388    for (name, value) in headers {
389        if FORWARDED_REQUEST_HEADERS.contains(name) || name.as_str().starts_with("x-atproto-") {
390            forwarded.append(name.clone(), value.clone());
391        }
392    }
393    forwarded
394}
395
396/// The service's endpoint origin, and a client pinned to it. The origin is
397/// the one [`ProxyLimits`] keeps for the target, or else
398/// [`resolve_endpoint`]'s; neither it nor the document is on this server.
399async fn resolve(
400    state: &AppState,
401    target: &Target<'_>,
402) -> Result<(Url, reqwest::Client), ApiError> {
403    let private = state.tree.private_addresses();
404    let endpoint = match state.proxy.cached_target(target.reference, private) {
405        Some(endpoint) => endpoint,
406        None => {
407            let endpoint = resolve_endpoint(state, target, private).await?;
408            state
409                .proxy
410                .cache_target(target.reference, private, &endpoint);
411            endpoint
412        }
413    };
414    // Checked on every call, cached or not: the names this deployment
415    // answers to, and the addresses the endpoint's name resolves to now.
416    refuse_this_server(state, target, &endpoint)?;
417    let client = didbot_http::pinned_client(&endpoint, private)
418        .await
419        .map_err(|error| unreachable_target(target, &error))?;
420    Ok((endpoint, client))
421}
422
423/// The origin the target's DID document names for its service id, fetched
424/// through a pinned client from a host that is not this server.
425async fn resolve_endpoint(
426    state: &AppState,
427    target: &Target<'_>,
428    private: didbot_http::PrivateAddresses,
429) -> Result<Url, ApiError> {
430    let unresolved = |reason: String| {
431        ApiError::bad_request(format!("could not resolve {}: {reason}", target.did))
432    };
433    let document_url = didbot_identity::document_url(target.did)
434        .map_err(|error| unresolved(error.to_string()))
435        .and_then(|url| Url::parse(&url).map_err(|error| unresolved(error.to_string())))?;
436    refuse_this_server(state, target, &document_url)?;
437    let document = match crate::foreign::fetch_document(&document_url, target.did, private).await {
438        Ok(crate::foreign::Document::Found(document)) => document,
439        Ok(crate::foreign::Document::NotFound | crate::foreign::Document::Gone) => {
440            return Err(unresolved(format!("{document_url} holds no document")))
441        }
442        Err(crate::foreign::DocumentError::Address(error)) => {
443            return Err(unreachable_target(target, &error))
444        }
445        Err(crate::foreign::DocumentError::Failed(reason)) => return Err(unresolved(reason)),
446    };
447    let document: DidDocument = serde_json::from_value(document)
448        .map_err(|error| unresolved(format!("parsing {document_url}: {error}")))?;
449    let endpoint = document
450        .service_endpoint(target.service)
451        .ok_or_else(|| unresolved(format!("its document names no #{} service", target.service)))?;
452    let mut endpoint = Url::parse(endpoint)
453        .ok()
454        .filter(|url| url.scheme() == "https" && url.host().is_some())
455        .ok_or_else(|| {
456            unresolved(format!(
457                "its #{} service is at {endpoint:?}, which is not an https URL",
458                target.service
459            ))
460        })?;
461    // The origin alone: the call's own path and query go on it.
462    endpoint.set_path("");
463    endpoint.set_query(None);
464    endpoint.set_fragment(None);
465    let _ = endpoint.set_username("");
466    let _ = endpoint.set_password(None);
467    Ok(endpoint)
468}
469
470/// Refuses a URL on a name this deployment answers to: a call proxied
471/// there is a call this server makes of itself.
472fn refuse_this_server(state: &AppState, target: &Target<'_>, url: &Url) -> Result<(), ApiError> {
473    let host = url.host_str().unwrap_or_default();
474    if !crate::foreign::serves(state.registry.as_ref(), host) {
475        return Ok(());
476    }
477    info!(
478        target = target.reference,
479        host, "refused: the proxy target is this server"
480    );
481    Err(ApiError::bad_request(format!(
482        "{} is served by this server, and a call is not proxied to itself",
483        target.reference
484    )))
485}
486
487/// The answer for a target this server will not, or cannot, reach.
488fn unreachable_target(target: &Target<'_>, error: &didbot_http::AddressError) -> ApiError {
489    info!(target = target.reference, %error, "refused a proxy target");
490    match error {
491        didbot_http::AddressError::Refused { .. } => ApiError::bad_request(format!(
492            "{} resolves to an address this server does not reach: {error}",
493            target.reference
494        )),
495        _ => ApiError::bad_request(format!("could not resolve {}: {error}", target.reference)),
496    }
497}
498
499/// A successful answer, streamed back under [`MAX_PROXIED_RESPONSE`].
500///
501/// `held` goes with the body, so the call counts as open until its last
502/// byte is sent or the body is dropped.
503fn answer(
504    target: &str,
505    upstream: reqwest::Response,
506    held: limits::InFlight,
507) -> Result<Response, ApiError> {
508    if upstream
509        .content_length()
510        .is_some_and(|length| length > MAX_PROXIED_RESPONSE as u64)
511    {
512        return Err(ApiError::new(
513            StatusCode::BAD_GATEWAY,
514            "UpstreamFailure",
515            format!("{target} answered with more than {MAX_PROXIED_RESPONSE} bytes"),
516        ));
517    }
518    let upstream = http::Response::<reqwest::Body>::from(upstream);
519    let mut response = Response::builder().status(upstream.status());
520    for name in CONTENT_HEADERS.iter().chain(&FORWARDED_RESPONSE_HEADERS) {
521        if let Some(value) = upstream.headers().get(name) {
522            response = response.header(name, value.clone());
523        }
524    }
525    let body = http_body_util::Limited::new(upstream.into_body(), MAX_PROXIED_RESPONSE).map_frame(
526        move |frame| {
527            let _held = &held;
528            frame
529        },
530    );
531    response.body(Body::new(body)).map_err(|error| {
532        ApiError::new(
533            StatusCode::BAD_GATEWAY,
534            "UpstreamFailure",
535            format!("{target} answered with a response this server cannot pass on: {error}"),
536        )
537    })
538}
539
540/// A failed answer, passed back as an XRPC error: the service's own status,
541/// `error` and `message`, except that its 500 is this server's 502.
542async fn upstream_error(target: &str, upstream: reqwest::Response) -> ApiError {
543    let status = upstream.status();
544    let forwarded: Vec<(HeaderName, String)> = FORWARDED_RESPONSE_HEADERS
545        .iter()
546        .filter_map(|name| {
547            let value = upstream.headers().get(name)?.to_str().ok()?;
548            Some((name.clone(), value.to_owned()))
549        })
550        .collect();
551    let body = didbot_http::read_bounded(upstream, didbot_http::MAX_JSON_BODY)
552        .await
553        .unwrap_or_default();
554    let named = serde_json::from_slice::<Value>(&body).ok();
555    let field = |name: &str| {
556        named
557            .as_ref()
558            .and_then(|value| value.get(name))
559            .and_then(Value::as_str)
560            .map(str::to_owned)
561    };
562    let status = if status == StatusCode::INTERNAL_SERVER_ERROR {
563        StatusCode::BAD_GATEWAY
564    } else {
565        status
566    };
567    let error = field("error").unwrap_or_else(|| status_name(status).to_owned());
568    let message = field("message").unwrap_or_else(|| format!("{target} answered {status}"));
569    forwarded.into_iter().fold(
570        ApiError::new(status, error, message),
571        |err, (name, value)| err.with_header(name, value),
572    )
573}
574
575/// The name XRPC gives a status, for a failure that names none: the
576/// reference client's `httpResponseCodeToName`.
577fn status_name(status: StatusCode) -> &'static str {
578    match status.as_u16() {
579        401 => "AuthenticationRequired",
580        403 => "Forbidden",
581        404 => "XRPCNotSupported",
582        406 => "NotAcceptable",
583        413 => "PayloadTooLarge",
584        415 => "UnsupportedMediaType",
585        429 => "RateLimitExceeded",
586        501 => "MethodNotImplemented",
587        502 => "UpstreamFailure",
588        503 => "NotEnoughResources",
589        504 => "UpstreamTimeout",
590        400..=499 => "InvalidRequest",
591        _ => "InternalServerError",
592    }
593}
didbot_serve/proxy/mod.rs

didbot_serve/proxy/
mod.rs
