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

rws_proxy/proxy/
grpc_web.rs
