grpc_webnext/
lib.rs

1//! grpc-webnext: serve the grpc-webnext wire protocol (unary over Fetch, streaming over
2//! WebSocket, plus `+json` / REST transcoding) over any gRPC service.
3//!
4//! All the inbound protocol translation is shared; the only thing that varies is where the
5//! translated gRPC call lands — a local in-process [`tonic::service::Routes`] you own
6//! ([`serve_in_process`], the "native server" / wrap mode) or a remote upstream over an
7//! HTTP/2 channel ([`serve_proxy`], the standalone binary proxy). Both are the same
8//! [`Backend`]; the two entry points build one and run the identical handlers, so a client
9//! can't tell a wrapped response from a proxied one.
10
11use std::net::SocketAddr;
12use std::sync::Arc;
13use std::time::Duration;
14
15use bytes::Bytes;
16use http::HeaderMap;
17use http_body_util::combinators::UnsyncBoxBody;
18use hyper::body::Incoming;
19use hyper::Request;
20use hyper_util::rt::{TokioExecutor, TokioIo};
21use tokio::net::TcpListener;
22use tonic::service::Routes;
23use tonic::transport::Channel;
24
25// Inbound protocol translation.
26pub mod backend;
27mod fetch;
28mod h2ts;
29mod reflect;
30pub mod schema;
31mod ws;
32
33// Wire codec + gRPC-semantics core: generated types, frame codec, Fetch-response framing,
34// metadata, HTTP-rule transcoding.
35pub mod pb {
36    include!(concat!(env!("OUT_DIR"), "/grpc.webnext.v1.rs"));
37}
38pub mod codec;
39pub mod frame;
40mod framing;
41pub mod grpc_framing;
42pub mod httprule;
43pub mod json_frame;
44pub mod metadata;
45pub mod transcode;
46
47pub use backend::Backend;
48pub use schema::{Schema, SchemaSource};
49
50pub use codec::BytesCodec;
51pub use frame::{decode_frame, encode_frame, FrameError};
52pub use framing::{
53    decode_response_body, encode_request_body, encode_response_body, encode_trailer_block,
54    FetchError, EMPTY_MESSAGE_BLOCK, LEN_PREFIX,
55};
56pub use grpc_framing::{deframe_all, frame as grpc_frame, Deframer};
57pub use httprule::{HttpCall, HttpRouter, WsBinding};
58pub use transcode::{TranscodeError, Transcoder};
59
60// --- Wire constants ---------------------------------------------------------
61
62pub const CT_PROTO: &str = "application/grpc-webnext+proto";
63pub const CT_JSON: &str = "application/grpc-webnext+json";
64pub(crate) const CT_GRPC: &str = "application/grpc";
65
66/// Base subprotocol; a client offers it plus a codec/credential entry.
67pub const WS_SUBPROTOCOL: &str = "grpc-webnext";
68pub const WS_SUBPROTOCOL_JSON: &str = "grpc-webnext+json";
69pub const WS_SUBPROTOCOL_PROTO: &str = "grpc-webnext+proto";
70
71/// The proxy owns the client-facing deadline: it drops the call at the deadline
72/// (surfacing DEADLINE_EXCEEDED) and forwards `grpc-timeout` downstream with this grace so
73/// the callee's own enforcement is a later backstop rather than racing the local timer.
74pub const DEADLINE_GRACE: Duration = Duration::from_millis(500);
75
76pub type BoxError = Box<dyn std::error::Error + Send + Sync>;
77pub type ResBody = UnsyncBoxBody<Bytes, BoxError>;
78
79// Authorization is deliberately NOT a grpc-webnext hook — neither per-RPC nor
80// per-connection. The request carries its metadata to the router on every transport, so
81// auth belongs in a tonic interceptor (in-process) or the upstream server / mesh (proxy):
82// one uniform, per-RPC check that also covers the native/h2ts path. Connection-level app
83// auth is non-canonical (the ecosystem gates connections only on network identity, e.g.
84// mTLS) and can't be uniform (Fetch has no connection). See `tests/inproc_auth.rs`.
85
86// --- Public configs ---------------------------------------------------------
87
88/// Configuration for [`serve_in_process`] — wrap an in-process tonic service.
89#[derive(Clone)]
90pub struct ServerConfig {
91    pub max_message_bytes: usize,
92    /// Descriptor-based JSON<->proto transcoder. When set, `+json`/REST requests are
93    /// transcoded to the router's binary protobuf and back. `None` ⇒ `+json` is
94    /// `UNIMPLEMENTED`.
95    pub transcoder: Option<Arc<Transcoder>>,
96    /// Allow plain `application/json` / blank content-type to reach *main* gRPC paths
97    /// (and blank WS subprotocols to infer their codec). Off by default.
98    pub allow_implicit_codec: bool,
99    /// WebSocket keepalive ping interval (`None` disables).
100    pub ws_keepalive: Option<Duration>,
101    /// Dead-peer timeout after a keepalive ping (gRPC's `keepalive_timeout`, default 20s).
102    pub ws_keepalive_timeout: Duration,
103}
104
105impl Default for ServerConfig {
106    fn default() -> Self {
107        Self {
108            max_message_bytes: 4 * 1024 * 1024,
109            transcoder: None,
110            allow_implicit_codec: false,
111            ws_keepalive: None,
112            ws_keepalive_timeout: Duration::from_secs(20),
113        }
114    }
115}
116
117/// Configuration for [`serve_proxy`] — front a remote upstream gRPC server.
118#[derive(Clone)]
119pub struct ProxyConfig {
120    /// Upstream gRPC endpoint (e.g. `http://127.0.0.1:50051`).
121    pub upstream: http::Uri,
122    pub max_message_bytes: usize,
123    pub ws_keepalive: Option<Duration>,
124    pub ws_keepalive_timeout: Duration,
125    /// Descriptor source for `+json` termination (`None` ⇒ binary-only).
126    pub schema: SchemaSource,
127    /// Reflection snapshot refresh interval (default 4h).
128    pub reflection_ttl: Duration,
129    /// Optional management endpoint: `POST` to this exact path forces a reflection reload.
130    pub admin_reload_path: Option<String>,
131    /// Accept plain `application/json` / blank on *main* gRPC paths (off by default).
132    pub allow_implicit_codec: bool,
133}
134
135impl Default for ProxyConfig {
136    fn default() -> Self {
137        Self {
138            upstream: http::Uri::default(),
139            max_message_bytes: 4 * 1024 * 1024,
140            ws_keepalive: None,
141            ws_keepalive_timeout: Duration::from_secs(20),
142            schema: SchemaSource::None,
143            reflection_ttl: Duration::from_secs(4 * 60 * 60),
144            admin_reload_path: None,
145            allow_implicit_codec: false,
146        }
147    }
148}
149
150// --- Internal runtime -------------------------------------------------------
151
152/// The per-connection knobs the shared handlers read, independent of which surface built
153/// them.
154pub(crate) struct RunConfig {
155    pub max_message_bytes: usize,
156    pub ws_keepalive: Option<Duration>,
157    pub ws_keepalive_timeout: Duration,
158    pub allow_implicit_codec: bool,
159    pub admin_reload_path: Option<String>,
160    /// Upstream `host:port` for the h2ts byte-pump path (proxy only; `None` in-process).
161    pub upstream_authority: Option<String>,
162}
163
164/// Everything a connection needs: where to dispatch, how to transcode, and the policy.
165/// Cheap to clone.
166#[derive(Clone)]
167pub(crate) struct Runtime {
168    pub backend: Backend,
169    pub schema: Schema,
170    pub cfg: Arc<RunConfig>,
171}
172
173// --- Entry points -----------------------------------------------------------
174
175/// Serve grpc-webnext + native gRPC for an in-process `routes` on `listener`.
176pub async fn serve_in_process(
177    listener: TcpListener,
178    routes: Routes,
179    config: ServerConfig,
180) -> std::io::Result<()> {
181    let schema = Schema::from_transcoder(config.transcoder.clone());
182    let cfg = Arc::new(RunConfig {
183        max_message_bytes: config.max_message_bytes,
184        ws_keepalive: config.ws_keepalive,
185        ws_keepalive_timeout: config.ws_keepalive_timeout,
186        allow_implicit_codec: config.allow_implicit_codec,
187        admin_reload_path: None,
188        upstream_authority: None,
189    });
190    run(listener, Runtime { backend: Backend::InProcess(routes), schema, cfg }).await
191}
192
193/// Serve grpc-webnext + native gRPC passthrough in front of an upstream gRPC server.
194pub async fn serve_proxy(listener: TcpListener, config: ProxyConfig) -> std::io::Result<()> {
195    // Lazy connect: the upstream need not be up when we start.
196    let channel = Channel::builder(config.upstream.clone()).connect_lazy();
197    // Raw host:port for the h2ts byte-pump path (which bypasses the gRPC `Channel`).
198    let upstream_authority = config.upstream.authority().map(|a| match a.port_u16() {
199        Some(_) => a.to_string(),
200        None => {
201            let port = if config.upstream.scheme_str() == Some("https") { 443 } else { 80 };
202            format!("{}:{}", a.host(), port)
203        }
204    });
205    // A bad bundled descriptor set is a config error — surface it at startup.
206    let schema = Schema::build(config.schema.clone(), channel.clone(), config.reflection_ttl)
207        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidInput, e.to_string()))?;
208    // Eager reflection load + TTL refresh (no-op for None/Bundled).
209    schema.start();
210    let cfg = Arc::new(RunConfig {
211        max_message_bytes: config.max_message_bytes,
212        ws_keepalive: config.ws_keepalive,
213        ws_keepalive_timeout: config.ws_keepalive_timeout,
214        allow_implicit_codec: config.allow_implicit_codec,
215        admin_reload_path: config.admin_reload_path,
216        upstream_authority,
217    });
218    run(listener, Runtime { backend: Backend::Upstream(channel), schema, cfg }).await
219}
220
221/// The connection accept loop, shared by both surfaces.
222async fn run(listener: TcpListener, rt: Runtime) -> std::io::Result<()> {
223    loop {
224        let (stream, _peer) = listener.accept().await?;
225        let io = TokioIo::new(stream);
226        let rt = rt.clone();
227        tokio::spawn(async move {
228            let service = hyper::service::service_fn(move |req: Request<Incoming>| {
229                let rt = rt.clone();
230                async move { fetch::handle(&rt, req).await }
231            });
232            if let Err(e) = hyper_util::server::conn::auto::Builder::new(TokioExecutor::new())
233                .serve_connection_with_upgrades(io, service)
234                .await
235            {
236                tracing::debug!("connection error: {e}");
237            }
238        });
239    }
240}
241
242/// Bind an ephemeral local address, serve an in-process `routes`, and return the bound
243/// address + task handle. For tests and simple mains.
244pub async fn bind_and_serve_in_process(
245    routes: Routes,
246    config: ServerConfig,
247) -> std::io::Result<(SocketAddr, tokio::task::JoinHandle<std::io::Result<()>>)> {
248    let listener = TcpListener::bind("127.0.0.1:0").await?;
249    let addr = listener.local_addr()?;
250    let handle = tokio::spawn(serve_in_process(listener, routes, config));
251    Ok((addr, handle))
252}
253
254/// Bind an ephemeral local address, serve a proxy, and return the bound address + handle.
255pub async fn bind_and_serve_proxy(
256    config: ProxyConfig,
257) -> std::io::Result<(SocketAddr, tokio::task::JoinHandle<std::io::Result<()>>)> {
258    let listener = TcpListener::bind("127.0.0.1:0").await?;
259    let addr = listener.local_addr()?;
260    let handle = tokio::spawn(serve_proxy(listener, config));
261    Ok((addr, handle))
262}
263
264// --- WebSocket handshake helpers --------------------------------------------
265
266/// Parse the `Sec-WebSocket-Protocol` request header into its comma-separated tokens.
267/// Used to negotiate the codec / detect the `h2ts` subprotocol at handshake time.
268pub fn ws_subprotocols(headers: &HeaderMap) -> Vec<String> {
269    headers
270        .get(http::header::SEC_WEBSOCKET_PROTOCOL)
271        .and_then(|v| v.to_str().ok())
272        .map(|s| s.split(',').map(|t| t.trim().to_string()).filter(|t| !t.is_empty()).collect())
273        .unwrap_or_default()
274}
grpc_webnext/lib.rs

grpc_webnext/
lib.rs
