netget/server/grpc_web/
mod.rs

1//! Separate binary protobuf HTTP/1.1 binding; the HTTP connection owns every RPC worker.
2pub mod actions;
3pub mod wire;
4use crate::server::grpc::{
5    self,
6    streaming::{ConnectionTasks, OwnedExecutor, Registry},
7    HttpCore,
8};
9use crate::{
10    protocol::SpawnContext,
11    state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo},
12};
13use anyhow::{ensure, Context, Result};
14use bytes::Bytes;
15use http_body_util::{BodyExt, Full, Limited, StreamBody};
16use hyper::{
17    body::{Frame, Incoming},
18    HeaderMap, Method, Request, Response, StatusCode,
19};
20use std::{convert::Infallible, sync::Arc, time::Duration};
21use tonic::{body::BoxBody, Status};
22use tower::{Layer, Service};
23pub const DEFAULT_RPC_TIMEOUT_SECS: u64 = 300;
24pub const FIRST_BYTE_TIMEOUT: Duration = Duration::from_secs(30);
25pub const IDLE_TIMEOUT: Duration = Duration::from_secs(120);
26pub const MAX_CONNECTIONS: usize = 256;
27const REFUSAL: &[u8] = b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nRetry-After: 5\r\nConnection: close\r\n\r\n";
28
29fn empty(status: StatusCode) -> Response<BoxBody> {
30    let mut response = Response::new(tonic::body::empty_body());
31    *response.status_mut() = status;
32    if status != StatusCode::NO_CONTENT {
33        response
34            .headers_mut()
35            .insert("connection", "close".parse().unwrap());
36    }
37    response
38}
39fn closing_status(status: Status) -> Response<BoxBody> {
40    let mut response = status_in_body(status.into_http());
41    response
42        .headers_mut()
43        .insert("connection", "close".parse().unwrap());
44    response
45}
46/// Always put application status in a Web trailer envelope, including failures.
47fn status_in_body(mut response: Response<BoxBody>) -> Response<BoxBody> {
48    if response.headers().contains_key("grpc-status") {
49        let mut trailers = HeaderMap::new();
50        for key in ["grpc-status", "grpc-message", "grpc-status-details-bin"] {
51            if let Some(value) = response.headers_mut().remove(key) {
52                trailers.insert(key, value);
53            }
54        }
55        *response.body_mut() =
56            StreamBody::new(futures::stream::iter([Ok::<_, Status>(
57                Frame::<Bytes>::trailers(trailers),
58            )]))
59            .boxed_unsync();
60    }
61    response
62}
63fn cors(response: &mut Response<BoxBody>, origin: Option<&str>) {
64    if let Some(origin) = origin {
65        response
66            .headers_mut()
67            .insert("access-control-allow-origin", origin.parse().unwrap());
68        response
69            .headers_mut()
70            .insert("vary", "Origin".parse().unwrap());
71        response.headers_mut().insert(
72            "access-control-expose-headers",
73            "grpc-status,grpc-message,grpc-status-details-bin"
74                .parse()
75                .unwrap(),
76        );
77    }
78}
79fn allowed_origin(
80    request: &Request<Incoming>,
81    configured: Option<&str>,
82) -> Result<Option<String>, StatusCode> {
83    let mut origins = request.headers().get_all("origin").iter();
84    let Some(origin) = origins.next() else {
85        return Ok(None);
86    };
87    let origin = origin.to_str().map_err(|_| StatusCode::FORBIDDEN)?;
88    if origins.next().is_some() || configured != Some(origin) {
89        return Err(StatusCode::FORBIDDEN);
90    }
91    Ok(Some(origin.into()))
92}
93async fn request(
94    mut request: Request<Incoming>,
95    core: HttpCore,
96    connection: crate::server::connection::ConnectionId,
97    activity: Arc<crate::server::accept_bounded::ConnectionActivity>,
98    registry: Registry,
99    executor: OwnedExecutor,
100    allow_origin: Option<String>,
101) -> Response<BoxBody> {
102    if !wire::bounded_headers(request.headers()) {
103        return empty(StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE);
104    }
105    let origin = match allowed_origin(&request, allow_origin.as_deref()) {
106        Ok(origin) => origin,
107        Err(code) => return empty(code),
108    };
109    if request.method() == Method::OPTIONS {
110        let allowed = [
111            "content-type",
112            "x-grpc-web",
113            "x-user-agent",
114            "grpc-timeout",
115            "grpc-encoding",
116            "grpc-accept-encoding",
117        ];
118        let headers_ok = request
119            .headers()
120            .get("access-control-request-headers")
121            .map(|value| {
122                value.to_str().is_ok_and(|value| {
123                    value.len() <= 1024
124                        && value.split(',').count() <= 32
125                        && value.split(',').all(|header| {
126                            allowed.contains(&header.trim().to_ascii_lowercase().as_str())
127                        })
128                })
129            })
130            .unwrap_or(true);
131        if origin.is_none()
132            || request
133                .headers()
134                .get("access-control-request-method")
135                .is_none_or(|value| value != "POST")
136            || request
137                .headers()
138                .get_all("access-control-request-method")
139                .iter()
140                .count()
141                != 1
142            || request
143                .headers()
144                .get_all("access-control-request-headers")
145                .iter()
146                .count()
147                > 1
148            || !headers_ok
149        {
150            return empty(StatusCode::FORBIDDEN);
151        }
152        let mut response = empty(StatusCode::NO_CONTENT);
153        cors(&mut response, origin.as_deref());
154        response
155            .headers_mut()
156            .insert("access-control-allow-methods", "POST".parse().unwrap());
157        response.headers_mut().insert(
158            "access-control-allow-headers",
159            allowed.join(",").parse().unwrap(),
160        );
161        return response;
162    }
163    if request.method() != Method::POST {
164        return empty(StatusCode::METHOD_NOT_ALLOWED);
165    }
166    if !wire::binary_content_type(request.headers())
167        || request.headers().get_all("accept").iter().any(|value| {
168            value
169                .to_str()
170                .map_or(true, |value| value.contains("grpc-web-text"))
171        })
172        || request.headers().contains_key("content-encoding")
173    {
174        return empty(StatusCode::UNSUPPORTED_MEDIA_TYPE);
175    }
176    if request.headers().get_all("grpc-encoding").iter().count() > 1 {
177        return empty(StatusCode::BAD_REQUEST);
178    }
179    if request.uri().path().len() > 2048 || request.uri().query().is_some() {
180        return empty(StatusCode::BAD_REQUEST);
181    }
182    // tonic-web chooses response encoding from Accept, so pin the selected binary subset.
183    request
184        .headers_mut()
185        .insert("accept", "application/grpc-web+proto".parse().unwrap());
186    let busy = activity.busy();
187    let body = request.map(|body| {
188        body.map_err(|_| Status::unavailable("request transport failed"))
189            .boxed_unsync()
190    });
191    let mut busy = Some(busy);
192    let mut service = tonic_web::GrpcWebLayer::new().layer(tower::service_fn(
193        move |request: Request<BoxBody>| {
194            let core = core.clone();
195            let registry = registry.clone();
196            let executor = executor.clone();
197            // HTTP/1.1 admits one live response per connection; this invocation owns its busy guard.
198            let busy = busy.take().expect("single request service invoked once");
199            async move {
200                let admission = match core.prepare(&request, executor) {
201                    Ok(admission) => admission,
202                    Err(status) => return Ok::<_, Infallible>(closing_status(status)),
203                };
204                let (parts, body) = request.into_parts();
205                // The selected methods have exactly one request. Admit before buffering, retain
206                // the same hard deadline, and read only framed bound+1 through EOF. This permits
207                // deterministic limit+1 refusal without racing an HTTP/1 client's upload or
208                // attempting an unlimited drain after an oversized prefix.
209                let input = tokio::time::timeout_at(
210                    admission.deadline,
211                    Limited::new(wire::request_body(body), wire::MAX_MESSAGE_BYTES + 6).collect(),
212                )
213                .await;
214                let bytes = match input {
215                    Ok(Ok(body)) => body.to_bytes(),
216                    Ok(Err(_)) => {
217                        return Ok(admission.guard(
218                            closing_status(Status::resource_exhausted(
219                                "request exceeds framed bound+1 or is malformed",
220                            )),
221                            busy,
222                        ))
223                    }
224                    Err(_) => {
225                        return Ok(admission.guard(
226                            closing_status(Status::deadline_exceeded(
227                                "RPC request deadline exceeded",
228                            )),
229                            busy,
230                        ))
231                    }
232                };
233                if bytes.len() > wire::MAX_MESSAGE_BYTES + 5 {
234                    return Ok(admission.guard(
235                        status_in_body(
236                            Status::resource_exhausted("request exceeds 4 MiB framed message")
237                                .into_http(),
238                        ),
239                        busy,
240                    ));
241                }
242                let request = Request::from_parts(
243                    parts,
244                    Full::new(bytes)
245                        .map_err(|never| match never {})
246                        .boxed_unsync(),
247                );
248                let response = core
249                    .request(request, connection, registry, &admission)
250                    .await;
251                Ok(admission.guard(status_in_body(response), busy))
252            }
253        },
254    ));
255    // The service is invoked exactly once. Using oneshot avoids a reusable closure that
256    // would incorrectly attempt to clone the affine admission/activity guard.
257    let mut response = match service.call(body).await {
258        Ok(response) => response,
259        Err(never) => match never {},
260    };
261    cors(&mut response, origin.as_deref());
262    response
263}
264
265pub struct GrpcWebServer;
266impl GrpcWebServer {
267    pub async fn spawn(ctx: SpawnContext) -> Result<std::net::SocketAddr> {
268        let p = ctx.startup_params.as_ref();
269        let schema = p
270            .context("proto_schema required")?
271            .get_string("proto_schema")?;
272        let timeout = p
273            .map(|p| p.get_optional_u64("rpc_timeout_secs"))
274            .transpose()?
275            .flatten()
276            .unwrap_or(DEFAULT_RPC_TIMEOUT_SECS);
277        ensure!(
278            (1..=3600).contains(&timeout),
279            "rpc_timeout_secs must be 1..3600"
280        );
281        let allow_origin = p
282            .map(|p| p.get_optional_string("allow_origin"))
283            .transpose()?
284            .flatten();
285        if let Some(origin) = &allow_origin {
286            ensure!(origin.len() <= 2048, "allow_origin exceeds 2048 bytes");
287            let uri: hyper::Uri = origin.parse()?;
288            ensure!(
289                matches!(uri.scheme_str(), Some("http" | "https"))
290                    && uri.host().is_some()
291                    && uri.path() == "/"
292                    && uri.query().is_none()
293                    && !origin.contains('@')
294                    && !origin.ends_with('/'),
295                "allow_origin must be one exact HTTP(S) origin without a trailing slash"
296            );
297        }
298        let pool = grpc::schema::load(&schema).await?;
299        ensure!(
300            pool.services().next().is_some(),
301            "schema must define services"
302        );
303        let core = HttpCore::new(
304            &ctx,
305            pool,
306            Arc::new(actions::GrpcWebProtocol),
307            Duration::from_secs(timeout),
308        );
309        let listener =
310            crate::server::socket_helpers::create_reusable_tcp_listener(ctx.legacy_listen_addr())
311                .await?;
312        let address = listener.local_addr()?;
313        let registrar = ctx.state.clone();
314        let server_id = ctx.server_id;
315        let task = tokio::spawn(async move {
316            let limiter = crate::server::accept_bounded::ConnectionLimiter::new(MAX_CONNECTIONS);
317            loop {
318                let (stream, remote, permit) = match crate::server::accept_bounded::accept_bounded(
319                    &listener,
320                    &limiter,
321                    REFUSAL,
322                    "gRPC-Web",
323                    Some(&ctx.status_tx),
324                )
325                .await
326                {
327                    Ok(value) => value,
328                    Err(error) => {
329                        crate::console_error!(ctx.status_tx, "gRPC-Web accept failed: {error}");
330                        break;
331                    }
332                };
333                let id = crate::server::connection::ConnectionId::new(
334                    ctx.state.get_next_unified_id().await,
335                );
336                let now = crate::utils::clock::Instant::now();
337                ctx.state
338                    .add_connection_to_server(
339                        ctx.server_id,
340                        ConnectionState {
341                            id,
342                            remote_addr: remote,
343                            local_addr: address,
344                            bytes_sent: 0,
345                            bytes_received: 0,
346                            packets_sent: 0,
347                            packets_received: 0,
348                            last_activity: now,
349                            status: ConnectionStatus::Active,
350                            status_changed_at: now,
351                            protocol_info: ProtocolConnectionInfo::empty(),
352                        },
353                    )
354                    .await;
355                let state = ctx.state.clone();
356                let owner = state.clone();
357                let core = core.clone();
358                let origin = allow_origin.clone();
359                let server = ctx.server_id;
360                owner.spawn_server_task(server, async move {
361                    let _permit = permit;
362                    if matches!(tokio::time::timeout(FIRST_BYTE_TIMEOUT, stream.peek(&mut [0u8;1])).await, Ok(Ok(n)) if n > 0) {
363                        let activity = Arc::new(crate::server::accept_bounded::ConnectionActivity::new());
364                        let executor = OwnedExecutor::default(); let _children = ConnectionTasks(executor.clone());
365                        let registry = Registry::default();
366                        let commands = crate::server::peer_support::register_peer_channel(&state, server, id.as_u32()).await;
367                        let _ = executor.spawn(core.clone().commands(commands, registry.clone(), id));
368                        let execute = executor.clone(); let active = activity.clone();
369                        let (done, ended) = tokio::sync::oneshot::channel();
370                        // Tracking the HTTP/1 owner in the same executor lets the core's hard
371                        // RPC timer cancel a body stalled behind a peer that never reads.
372                        let _ = executor.spawn(async move {
373                            let service = hyper::service::service_fn(move |request| {
374                                let core = core.clone(); let activity = active.clone(); let registry = registry.clone();
375                                let executor = execute.clone(); let origin = origin.clone();
376                                async move { Ok::<_, Infallible>(self::request(request, core, id, activity, registry, executor, origin).await) }
377                            });
378                            let _ = hyper::server::conn::http1::Builder::new().max_headers(64).max_buf_size(32768)
379                                .serve_connection(hyper_util::rt::TokioIo::new(stream), service).await;
380                            let _ = done.send(());
381                        });
382                        tokio::select! { _ = ended => {}, _ = crate::server::accept_bounded::watch_idle(activity, IDLE_TIMEOUT) => {} }
383                    }
384                    state.remove_peer_handle(server, id.as_u32()).await;
385                    state.remove_connection_from_server(server, id).await;
386                }).await;
387            }
388        });
389        registrar.register_server_task(server_id, task).await;
390        Ok(address)
391    }
392}
netget/server/grpc_web/mod.rs

netget/server/grpc_web/
mod.rs
