structured_proxy/
upstream.rs

1//! The gRPC service the proxy calls.
2//!
3//! An upstream is any tower service that speaks gRPC over `http` types: a
4//! remote [`tonic::transport::Channel`], or an embedder's own services in
5//! process, such as [`tonic::service::Routes`]. The transcoder calls it for
6//! every transcoded request, and [`ProxyService`](crate::ProxyService) hands it
7//! native gRPC requests unchanged.
8
9use std::convert::Infallible;
10use std::future::Future;
11use std::pin::Pin;
12use std::task::{ready, Context, Poll};
13
14use bytes::Bytes;
15use pin_project_lite::pin_project;
16use tonic::client::GrpcService;
17
18/// The error type a gRPC service hands back through tonic.
19pub type BoxError = Box<dyn std::error::Error + Send + Sync>;
20
21/// A gRPC service the proxy can call: a remote
22/// [`Channel`](tonic::transport::Channel), [`tonic::service::Routes`], or any
23/// other `tower::Service<http::Request<tonic::body::Body>>` answering with a
24/// gRPC response. Implemented for every such service; there is nothing to
25/// implement by hand.
26///
27/// An in-process upstream sees a transcoded call exactly as it sees a native
28/// gRPC request: through its whole stack, interceptors and layers included.
29///
30/// # Examples
31///
32/// ```
33/// use structured_proxy::upstream::Upstream;
34///
35/// fn accepts<U: Upstream>(_upstream: U) {}
36///
37/// // An empty set of tonic services answers every call with UNIMPLEMENTED.
38/// accepts(tonic::service::Routes::default());
39/// ```
40pub trait Upstream:
41    GrpcService<
42        tonic::body::Body,
43        ResponseBody: http_body::Body<Data = Bytes, Error: Into<BoxError> + Send> + Send + 'static,
44        Error: Into<BoxError>,
45        Future: Send,
46    > + Clone
47    + Send
48    + Sync
49    + 'static
50{
51}
52
53impl<T> Upstream for T where
54    T: GrpcService<
55            tonic::body::Body,
56            ResponseBody: http_body::Body<Data = Bytes, Error: Into<BoxError> + Send>
57                              + Send
58                              + 'static,
59            Error: Into<BoxError>,
60            Future: Send,
61        > + Clone
62        + Send
63        + Sync
64        + 'static
65{
66}
67
68pin_project! {
69    /// One native gRPC request on its way through an upstream: waits for the
70    /// upstream to be ready, calls it, and answers with its response. A
71    /// failure of the upstream itself (a remote one that cannot be reached)
72    /// becomes a trailers-only gRPC error, so the client gets a status rather
73    /// than a broken connection.
74    ///
75    /// No proxy timer runs here and `grpc-timeout` travels unchanged: the
76    /// caller is a gRPC client, which enforces its own deadline (gRPC
77    /// PROTOCOL-HTTP2, "Timeout") by cancelling the stream, and that drops
78    /// this future wherever it waits, readiness included. A transcoded call is
79    /// different: there the proxy is the gRPC client and bounds the call itself.
80    #[project = PassThroughProj]
81    #[project_replace = PassThroughReplace]
82    pub(crate) enum PassThrough<U: Upstream> {
83        Ready {
84            upstream: U,
85            request: http::Request<tonic::body::Body>,
86            protocol: GrpcProtocol,
87        },
88        Call {
89            #[pin]
90            future: U::Future,
91            protocol: GrpcProtocol,
92        },
93        Done,
94    }
95}
96
97impl<U: Upstream> PassThrough<U> {
98    pub(crate) fn new(
99        upstream: U,
100        request: http::Request<tonic::body::Body>,
101        protocol: GrpcProtocol,
102    ) -> Self {
103        Self::Ready {
104            upstream,
105            request,
106            protocol,
107        }
108    }
109}
110
111impl<U: Upstream> Future for PassThrough<U> {
112    type Output = Result<http::Response<axum::body::Body>, Infallible>;
113
114    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
115        loop {
116            match self.as_mut().project() {
117                PassThroughProj::Ready {
118                    upstream, protocol, ..
119                } => {
120                    let protocol = *protocol;
121                    if let Err(e) = ready!(upstream.poll_ready(cx)) {
122                        self.set(Self::Done);
123                        return Poll::Ready(Ok(failure(e.into(), protocol)));
124                    }
125                    let PassThroughReplace::Ready {
126                        mut upstream,
127                        request,
128                        ..
129                    } = self.as_mut().project_replace(Self::Done)
130                    else {
131                        unreachable!("the state was just matched as Ready");
132                    };
133                    // The returned future owns what it needs; the service
134                    // handle is dropped, as tower's `Oneshot` does.
135                    let future = upstream.call(request);
136                    self.set(Self::Call { future, protocol });
137                }
138                PassThroughProj::Call { future, protocol } => {
139                    let protocol = *protocol;
140                    let result = ready!(future.poll(cx));
141                    self.set(Self::Done);
142                    return Poll::Ready(Ok(match result {
143                        Ok(response) => response.map(axum::body::Body::new),
144                        Err(e) => failure(e.into(), protocol),
145                    }));
146                }
147                PassThroughProj::Done => panic!("PassThrough polled after completion"),
148            }
149        }
150    }
151}
152
153/// The trailers-only answer to an upstream that failed to take the call, in
154/// the request's protocol: gRPC-Web carries a trailers-only status in its
155/// headers too, but its client reads it only under a gRPC-Web content type
156/// (gRPC PROTOCOL-WEB).
157fn failure(error: BoxError, protocol: GrpcProtocol) -> http::Response<axum::body::Body> {
158    trailers_only(tonic::Status::from_error(error), protocol)
159}
160
161/// `status` as a trailers-only gRPC answer in `protocol`.
162pub(crate) fn trailers_only(
163    status: tonic::Status,
164    protocol: GrpcProtocol,
165) -> http::Response<axum::body::Body> {
166    let mut response: http::Response<axum::body::Body> = status.into_http();
167    if protocol != GrpcProtocol::Grpc {
168        response.headers_mut().insert(
169            http::header::CONTENT_TYPE,
170            http::HeaderValue::from_static(protocol.content_type()),
171        );
172    }
173    response
174}
175
176/// The gRPC protocol a request speaks.
177#[derive(Clone, Copy, Debug, PartialEq, Eq)]
178pub(crate) enum GrpcProtocol {
179    /// gRPC over HTTP/2 (gRPC PROTOCOL-HTTP2).
180    Grpc,
181    /// Binary gRPC-Web (gRPC PROTOCOL-WEB).
182    Web,
183    /// Base64 gRPC-Web, `-text` (gRPC PROTOCOL-WEB).
184    WebText,
185}
186
187impl GrpcProtocol {
188    /// The content type of an answer the proxy writes itself: the protocol
189    /// with gRPC's default protobuf codec.
190    fn content_type(self) -> &'static str {
191        match self {
192            Self::Grpc => "application/grpc",
193            Self::Web => "application/grpc-web+proto",
194            Self::WebText => "application/grpc-web-text+proto",
195        }
196    }
197}
198
199/// The gRPC protocol `headers` announce, if any: a media type of
200/// `application/grpc` or `application/grpc+<codec>` (gRPC PROTOCOL-HTTP2,
201/// "Content-Type"), or `application/grpc-web[-text][+<codec>]` (gRPC
202/// PROTOCOL-WEB). Media types compare without case (RFC 9110 §8.3.1).
203pub(crate) fn grpc_protocol(headers: &http::HeaderMap) -> Option<GrpcProtocol> {
204    let value = headers.get(http::header::CONTENT_TYPE)?.as_bytes();
205    let media = match value.iter().position(|&b| b == b';') {
206        Some(end) => &value[..end],
207        None => value,
208    }
209    .trim_ascii();
210    let rest = strip_prefix_ignore_case(media, b"application/grpc")?;
211    match rest {
212        [] | [b'+', ..] => Some(GrpcProtocol::Grpc),
213        _ => web_protocol(strip_prefix_ignore_case(rest, b"-web")?),
214    }
215}
216
217/// Whether a request is a browser's CORS preflight for a gRPC-Web call: an
218/// `OPTIONS` with an `Origin` and an `Access-Control-Request-Method` (Fetch
219/// §3.2.2) whose `Access-Control-Request-Headers` names `x-grpc-web`, the
220/// header gRPC-Web clients send with every call (gRPC PROTOCOL-WEB). It
221/// carries no gRPC content type, so only these headers tell it apart from a
222/// REST preflight.
223pub(crate) fn is_grpc_web_preflight(method: &http::Method, headers: &http::HeaderMap) -> bool {
224    method == http::Method::OPTIONS
225        && headers.contains_key(http::header::ORIGIN)
226        && headers.contains_key(http::header::ACCESS_CONTROL_REQUEST_METHOD)
227        && headers
228            .get_all(http::header::ACCESS_CONTROL_REQUEST_HEADERS)
229            .iter()
230            .flat_map(|value| value.as_bytes().split(|&b| b == b','))
231            .any(|name| name.trim_ascii().eq_ignore_ascii_case(b"x-grpc-web"))
232}
233
234/// What follows `application/grpc-web`: nothing or `+<codec>` for binary,
235/// `-text` with an optional `+<codec>` for base64.
236fn web_protocol(rest: &[u8]) -> Option<GrpcProtocol> {
237    let (protocol, rest) = match strip_prefix_ignore_case(rest, b"-text") {
238        Some(rest) => (GrpcProtocol::WebText, rest),
239        None => (GrpcProtocol::Web, rest),
240    };
241    matches!(rest, [] | [b'+', ..]).then_some(protocol)
242}
243
244/// `bytes` after `prefix`, compared without ASCII case.
245fn strip_prefix_ignore_case<'a>(bytes: &'a [u8], prefix: &[u8]) -> Option<&'a [u8]> {
246    let (head, rest) = bytes.split_at_checked(prefix.len())?;
247    head.eq_ignore_ascii_case(prefix).then_some(rest)
248}
249
250#[cfg(test)]
251mod tests;
structured_proxy/upstream.rs

structured_proxy/
upstream.rs
