rust_web_server/proxy/
grpc_web.rs

1//! gRPC-web to gRPC: [`GrpcWebLayer`] lets browsers (which can't read HTTP
2//! trailers or send HTTP/2 frames of their own) call gRPC services, as
3//! Envoy's `grpc_web` filter does.
4//!
5//! A request with `Content-Type: application/grpc-web[+proto]` or
6//! `application/grpc-web-text[+proto]` (base64) is passed on as
7//! `application/grpc[+proto]` with `te: trailers` and its body decoded; the
8//! response comes back as gRPC-web, the gRPC status and other trailers
9//! appended to the body as the final frame (flag `0x80`), base64-encoded for
10//! `-text`. Bodies stream both ways. Other requests pass through untouched.
11
12use std::io::Read;
13use std::sync::{Arc, Mutex};
14
15use crate::application::Application;
16use crate::core::base64::Base64;
17use crate::header::Header;
18use crate::middleware::Middleware;
19use crate::request::{Body, Request};
20use crate::response::{Response, Trailers};
21use crate::server::ConnectionInfo;
22
23/// Translates gRPC-web requests to gRPC for the layers after it — put it in
24/// front of a [`GrpcProxy`](super::GrpcProxy):
25///
26/// ```rust,no_run
27/// use rust_web_server::app::App;
28/// use rust_web_server::core::New;
29/// use rust_web_server::proxy::{GrpcProxy, GrpcWebLayer};
30///
31/// let app = App::new()
32///     .wrap(GrpcWebLayer::new())
33///     .wrap(GrpcProxy::new(["grpc-service:50051"]));
34/// ```
35#[derive(Default)]
36pub struct GrpcWebLayer;
37
38impl GrpcWebLayer {
39    pub fn new() -> Self {
40        GrpcWebLayer
41    }
42}
43
44/// What a gRPC-web request asked for: base64 (`-text`) or not, and the
45/// message format suffix (`+proto`, `+json` or none).
46struct Variant {
47    text: bool,
48    suffix: String,
49}
50
51fn variant(request: &Request) -> Option<Variant> {
52    let content_type = request.header(Header::_CONTENT_TYPE)?.value.as_ref().trim().to_ascii_lowercase();
53    if let Some(suffix) = content_type.strip_prefix("application/grpc-web-text") {
54        return Some(Variant { text: true, suffix: suffix.to_string() });
55    }
56    content_type.strip_prefix("application/grpc-web").map(|suffix| Variant { text: false, suffix: suffix.to_string() })
57}
58
59impl Middleware for GrpcWebLayer {
60    fn handle(&self, request: &Request, connection: &ConnectionInfo, next: &dyn Application) -> Result<Response, String> {
61        let Some(variant) = variant(request) else {
62            return next.execute(request, connection);
63        };
64        let mut grpc = request.clone();
65        grpc.headers.retain(|h| {
66            !(h.name.eq_ignore_ascii_case("content-type") || h.name.eq_ignore_ascii_case("te")
67                || (variant.text && h.name.eq_ignore_ascii_case("content-length")))
68        });
69        grpc.headers.push(Header::new(Header::_CONTENT_TYPE, format!("application/grpc{}", variant.suffix)));
70        grpc.headers.push(Header::new("te", "trailers"));
71        if variant.text {
72            grpc.body = match std::mem::take(&mut grpc.body) {
73                Body::Bytes(bytes) => Body::Bytes(decode_text(&bytes).map_err(|e| format!("grpc-web-text request body: {e}"))?),
74                stream => Body::Stream(Arc::new(Mutex::new(Box::new(Base64Decoder::new(stream.into_reader()))))),
75            };
76        }
77        let response = next.execute(&grpc, connection)?;
78        Ok(to_grpc_web(response, variant.text))
79    }
80
81    /// The body passes through (decoded on the fly for `-text`).
82    fn accepts_streamed_body(&self, _request: &Request, _connection: &ConnectionInfo) -> bool {
83        true
84    }
85}
86
87/// The gRPC response as gRPC-web: the content type renamed, the trailers as
88/// the body's last frame.
89fn to_grpc_web(mut response: Response, text: bool) -> Response {
90    let web_type = |grpc: &str| {
91        let suffix = grpc.trim().strip_prefix("application/grpc").unwrap_or("");
92        format!("application/grpc-web{}{suffix}", if text { "-text" } else { "" })
93    };
94    let mut is_grpc = false;
95    for header in response.headers.iter_mut() {
96        if header.name.eq_ignore_ascii_case("content-type") && header.value.starts_with("application/grpc") {
97            header.value = web_type(&header.value).into();
98            is_grpc = true;
99        }
100    }
101    for range in response.content_range_list.iter_mut() {
102        if range.content_type.starts_with("application/grpc") {
103            range.content_type = web_type(&range.content_type);
104            is_grpc = true;
105        }
106    }
107    if !is_grpc {
108        return response; // an error page from rws or the upstream: leave it
109    }
110    let trailers = response.trailers.take();
111    if let Some(body) = response.stream_pipe.take() {
112        let framed: Box<dyn Read + Send> = Box::new(TrailerFrame { body, trailers, frame: None });
113        response.stream_pipe = Some(if text { Box::new(Base64Encoder::new(framed)) } else { framed });
114    } else {
115        // Buffered (a small response): one body with the frame appended.
116        let mut body = Response::generate_body(std::mem::take(&mut response.content_range_list));
117        if let Some(trailers) = &trailers {
118            body.extend_from_slice(&trailer_frame(&trailers.get()));
119        }
120        if text {
121            body = Base64::encode(&body).unwrap_or_default().into_bytes();
122        }
123        let content_type = response
124            .headers
125            .iter()
126            .find(|h| h.name.eq_ignore_ascii_case("content-type"))
127            .map_or_else(|| web_type("application/grpc"), |h| h.value.to_string());
128        response.headers.retain(|h| !h.name.eq_ignore_ascii_case("content-type"));
129        if !body.is_empty() {
130            response.content_range_list = vec![crate::range::Range::get_content_range(body, content_type)];
131        } else {
132            response.headers.push(Header::new(Header::_CONTENT_TYPE, content_type));
133        }
134    }
135    response
136}
137
138/// The gRPC-web trailer frame: flag `0x80`, a 4-byte length, then
139/// `name:value\r\n` lines with lower-case names. Empty without trailers.
140pub(crate) fn trailer_frame(trailers: &[Header]) -> Vec<u8> {
141    if trailers.is_empty() {
142        return Vec::new();
143    }
144    let mut lines = Vec::new();
145    for header in trailers {
146        lines.extend(header.name.bytes().map(|b| b.to_ascii_lowercase()));
147        lines.push(b':');
148        lines.extend_from_slice(header.value.trim().as_bytes());
149        lines.extend_from_slice(b"\r\n");
150    }
151    let mut frame = Vec::with_capacity(5 + lines.len());
152    frame.push(0x80);
153    frame.extend_from_slice(&(lines.len() as u32).to_be_bytes());
154    frame.extend_from_slice(&lines);
155    frame
156}
157
158/// The response body, then the trailer frame once the body has ended (and
159/// with it the trailers are known).
160struct TrailerFrame {
161    body: Box<dyn Read + Send>,
162    trailers: Option<Trailers>,
163    frame: Option<std::io::Cursor<Vec<u8>>>,
164}
165
166impl Read for TrailerFrame {
167    fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
168        if self.frame.is_none() {
169            let n = self.body.read(out)?;
170            if n > 0 {
171                return Ok(n);
172            }
173            let fields = self.trailers.as_ref().map(|t| t.get()).unwrap_or_default();
174            self.frame = Some(std::io::Cursor::new(trailer_frame(&fields)));
175        }
176        self.frame.as_mut().map_or(Ok(0), |frame| frame.read(out))
177    }
178}
179
180/// Base64-encodes a stream: whole 3-byte groups as they arrive, the rest
181/// (padded) at the end.
182struct Base64Encoder {
183    inner: Box<dyn Read + Send>,
184    raw: Vec<u8>,
185    encoded: std::io::Cursor<Vec<u8>>,
186    ended: bool,
187}
188
189impl Base64Encoder {
190    fn new(inner: Box<dyn Read + Send>) -> Self {
191        Base64Encoder { inner, raw: Vec::new(), encoded: std::io::Cursor::new(Vec::new()), ended: false }
192    }
193}
194
195impl Read for Base64Encoder {
196    fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
197        loop {
198            let n = self.encoded.read(out)?;
199            if n > 0 || self.ended {
200                return Ok(n);
201            }
202            let mut chunk = vec![0u8; 16 * 1024];
203            let read = self.inner.read(&mut chunk)?;
204            self.raw.extend_from_slice(&chunk[..read]);
205            let take = if read == 0 {
206                self.ended = true;
207                self.raw.len()
208            } else {
209                self.raw.len() - self.raw.len() % 3
210            };
211            let group: Vec<u8> = self.raw.drain(..take).collect();
212            let text = Base64::encode(&group).map_err(std::io::Error::other)?;
213            self.encoded = std::io::Cursor::new(text.into_bytes());
214        }
215    }
216}
217
218/// Decodes a base64 request body as it arrives. A body may be several
219/// padded segments back to back (a client encoding each message on its own).
220struct Base64Decoder {
221    inner: Box<dyn Read + Send>,
222    text: Vec<u8>,
223    decoded: std::io::Cursor<Vec<u8>>,
224    ended: bool,
225}
226
227impl Base64Decoder {
228    fn new(inner: Box<dyn Read + Send>) -> Self {
229        Base64Decoder { inner, text: Vec::new(), decoded: std::io::Cursor::new(Vec::new()), ended: false }
230    }
231}
232
233impl Read for Base64Decoder {
234    fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
235        loop {
236            let n = self.decoded.read(out)?;
237            if n > 0 || self.ended {
238                return Ok(n);
239            }
240            let mut chunk = vec![0u8; 16 * 1024];
241            let read = self.inner.read(&mut chunk)?;
242            self.text.extend(chunk[..read].iter().filter(|b| !b.is_ascii_whitespace()));
243            let take = if read == 0 {
244                self.ended = true;
245                if self.text.len() % 4 != 0 {
246                    return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "grpc-web-text body isn't whole base64"));
247                }
248                self.text.len()
249            } else {
250                self.text.len() - self.text.len() % 4
251            };
252            let groups: Vec<u8> = self.text.drain(..take).collect();
253            let bytes = decode_text(&groups).map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
254            self.decoded = std::io::Cursor::new(bytes);
255        }
256    }
257}
258
259/// Decode base64 that may hold several padded segments.
260fn decode_text(text: &[u8]) -> Result<Vec<u8>, String> {
261    let text: Vec<u8> = text.iter().copied().filter(|b| !b.is_ascii_whitespace()).collect();
262    if text.len() % 4 != 0 {
263        return Err("not whole base64".to_string());
264    }
265    let mut out = Vec::with_capacity(text.len() / 4 * 3);
266    let mut start = 0;
267    for (i, group) in text.chunks(4).enumerate() {
268        // A padded group ends a segment.
269        if group.contains(&b'=') {
270            let end = (i + 1) * 4;
271            let segment = String::from_utf8(text[start..end].to_vec()).map_err(|e| e.to_string())?;
272            out.extend(Base64::decode(segment)?);
273            start = end;
274        }
275    }
276    if start < text.len() {
277        let segment = String::from_utf8(text[start..].to_vec()).map_err(|e| e.to_string())?;
278        out.extend(Base64::decode(segment)?);
279    }
280    Ok(out)
281}
282
283#[cfg(test)]
284mod tests;
rust_web_server/proxy/grpc_web.rs

rust_web_server/proxy/
grpc_web.rs
