eggserve_core/server/
connection.rs

1//! Connection execution pipeline.
2//!
3//! This module owns the per-connection execution path from TCP accept to
4//! response completion. It is used by both the CLI accept loop and the
5//! embedded runtime.
6//!
7//! # Pipeline steps
8//!
9//! 1. Optional TLS handshake (feature-gated)
10//! 2. HTTP/1 connection setup via Hyper
11//! 3. Request conversion to canonical types
12//! 4. Request-policy validation (body rejection for body-forbidden methods)
13//! 5. Service invocation with panic containment
14//! 6. Canonical response normalization
15//! 7. Transport-body conversion
16//! 8. Permit release and connection termination
17
18use std::convert::Infallible;
19use std::sync::Arc;
20
21use hyper::body::Incoming;
22use hyper::server::conn::http1;
23use hyper::service::service_fn;
24use hyper::{Request, Response};
25use hyper_util::rt::{TokioIo, TokioTimer};
26use tokio::sync::broadcast;
27
28use crate::primitives::request_body_policy::RequestBodyPolicy;
29use crate::response::BoxBodyInner;
30use crate::server::config::RuntimeConfig;
31use crate::server::service::{Service, ServiceError};
32use crate::server::RuntimeState;
33
34/// Serve a single HTTP/1.1 connection.
35///
36/// This is the core connection executor used by both the CLI and embedded
37/// runtime. It handles:
38///
39/// - HTTP/1 connection setup with Hyper
40/// - Header-read timeout enforcement
41/// - Connection-total-timeout enforcement (maximum connection lifetime)
42/// - Graceful shutdown propagation
43///
44/// The `service` parameter provides the request handler. The built-in static
45/// path supplies [`crate::server::StaticService`]; custom services supply their own
46/// [`Service`] implementation.
47pub async fn serve_connection<I, S>(
48    io: TokioIo<I>,
49    service: S,
50    config: &RuntimeConfig,
51    shutdown_rx: &mut broadcast::Receiver<()>,
52    conn_id: u64,
53) where
54    I: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
55    S: hyper::service::Service<
56            Request<Incoming>,
57            Response = Response<BoxBodyInner>,
58            Error = Infallible,
59        > + 'static,
60{
61    let conn = http1::Builder::new()
62        .timer(TokioTimer::new())
63        .header_read_timeout(config.header_read_timeout)
64        .serve_connection(io, service)
65        .with_upgrades();
66    let mut conn = std::pin::pin!(conn);
67    tokio::select! {
68        result = tokio::time::timeout(config.connection_total_timeout, &mut conn) => {
69            match result {
70                Ok(Ok(())) => {
71                    crate::ops::Logger::global().emit(
72                        crate::ops::Event::new(
73                            crate::ops::Severity::Debug,
74                            crate::ops::EventKind::KeepAliveClosed,
75                            "connection closed",
76                        )
77                        .connection_id(conn_id),
78                    );
79                }
80                Ok(Err(e)) => {
81                    crate::ops::Logger::global().emit(
82                        crate::ops::Event::new(
83                            crate::ops::Severity::Debug,
84                            crate::ops::EventKind::ClientDisconnect,
85                            format!("connection error: {}", e),
86                        )
87                        .connection_id(conn_id),
88                    );
89                }
90                Err(_elapsed) => {
91                    crate::ops::global_counters().connection_total_timeouts.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
92                    crate::ops::Logger::global().emit(
93                        crate::ops::Event::new(
94                            crate::ops::Severity::Warn,
95                            crate::ops::EventKind::ConnectionTotalTimeout,
96                            "connection total timeout",
97                        )
98                        .connection_id(conn_id),
99                    );
100                    conn.as_mut().graceful_shutdown();
101                    let _ = conn.await;
102                }
103            }
104        }
105        _ = shutdown_rx.recv() => {
106            conn.as_mut().graceful_shutdown();
107            let _ = conn.await;
108        }
109    }
110}
111
112/// Serve a single connection with a custom [`Service`] implementation.
113///
114/// This wraps the raw Hyper service with:
115/// - Request conversion from Hyper to canonical types
116/// - Handler timeout enforcement
117/// - Service error to response conversion
118/// - Canonical response normalization
119///
120/// Panics in the service propagate to the tokio task boundary and are
121/// caught by the `JoinSet` in the accept loop. The connection is dropped
122/// and a `ConnectionPanic` event is emitted.
123#[allow(clippy::too_many_arguments)]
124pub async fn serve_connection_with_runtime_state<I, S>(
125    io: TokioIo<I>,
126    service: S,
127    config: &RuntimeConfig,
128    runtime_state: Arc<RuntimeState>,
129    shutdown_rx: &mut broadcast::Receiver<()>,
130    conn_id: u64,
131    local_addr: std::net::SocketAddr,
132    remote_addr: std::net::SocketAddr,
133    tls: bool,
134    tls_info: Option<crate::primitives::connection_info::TlsInfo>,
135) where
136    I: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
137    S: Service,
138{
139    let service = std::sync::Arc::new(service);
140    let config = Arc::new(config.clone());
141    let handler_timeout = config.handler_timeout;
142    let body_read_timeout = config.body_read_timeout;
143    let max_body_bytes = config.max_request_body_bytes;
144    let tls_info = std::sync::Arc::new(tls_info);
145    let file_stream_semaphore = runtime_state.file_stream_semaphore().clone();
146    let response_config = config.clone();
147
148    let hyper_service = service_fn(move |req: Request<Incoming>| {
149        let service = service.clone();
150        let tls_info = tls_info.clone();
151        let file_stream_semaphore = file_stream_semaphore.clone();
152        let config = response_config.clone();
153        async move {
154            // Convert Hyper request to canonical RequestHead.
155            let head = match convert_request_head(&req) {
156                Ok(h) => h,
157                Err(e) => {
158                    return Ok::<_, Infallible>(finalize_runtime_response(
159                        e.to_response(),
160                        &config,
161                    ));
162                }
163            };
164
165            // TRACE content remains a transport-level rejection. Other
166            // methods, including GET, HEAD, and DELETE, are governed by the
167            // service-declared policy below.
168            if head.method().as_str() == "TRACE"
169                && (req
170                    .headers()
171                    .get(hyper::header::CONTENT_LENGTH)
172                    .and_then(|value| value.to_str().ok())
173                    .and_then(|value| value.parse::<u64>().ok())
174                    .is_some_and(|length| length > 0)
175                    || req.headers().contains_key(hyper::header::TRANSFER_ENCODING))
176            {
177                let mut response = crate::response::bad_request(false);
178                response.headers_mut().insert(
179                    hyper::header::CONNECTION,
180                    hyper::header::HeaderValue::from_static("close"),
181                );
182                return Ok::<_, Infallible>(finalize_runtime_response(response, &config));
183            }
184
185            // Select effective body policy.
186            let service_policy = service.request_body_policy(&head);
187            let effective_policy = select_body_policy(service_policy, max_body_bytes);
188
189            // Extract body from Hyper request.
190            let (parts, body) = req.into_parts();
191
192            // Validate body framing (TE+CL conflict, duplicate CL) for all methods.
193            if let Err(e) = validate_body_framing(&parts.headers) {
194                crate::ops::global_counters()
195                    .parser_rejects
196                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
197                crate::ops::Logger::global().emit(
198                    crate::ops::Event::new(
199                        crate::ops::Severity::Debug,
200                        crate::ops::EventKind::ParserRejection,
201                        format!("parser rejection: {}", e),
202                    )
203                    .connection_id(conn_id),
204                );
205                return Ok::<_, Infallible>(finalize_runtime_response(e.to_response(), &config));
206            }
207
208            let declared_length = parts
209                .headers
210                .get(hyper::header::CONTENT_LENGTH)
211                .and_then(|v| v.to_str().ok())
212                .and_then(|s| s.parse::<u64>().ok());
213
214            // Validate Content-Length against effective limit.
215            if let Some(len) = declared_length {
216                if let Some(limit) = effective_policy.max_bytes() {
217                    if len > limit {
218                        crate::ops::global_counters()
219                            .body_rejections
220                            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
221                        crate::ops::Logger::global().emit(
222                            crate::ops::Event::new(
223                                crate::ops::Severity::Debug,
224                                crate::ops::EventKind::BodyPolicyRejection,
225                                "body too large",
226                            )
227                            .connection_id(conn_id)
228                            .field(crate::ops::Field::U64("declared_bytes".into(), len))
229                            .field(crate::ops::Field::U64("limit_bytes".into(), limit)),
230                        );
231                        let err = crate::primitives::request_body_error::RequestBodyError::DeclaredLengthTooLarge {
232                            declared: len,
233                            limit,
234                        };
235                        return Ok::<_, Infallible>(finalize_runtime_response(
236                            body_error_to_response(err, &head),
237                            &config,
238                        ));
239                    }
240                }
241            }
242
243            // Reject Expect: 100-continue early — do not send an invitation
244            // to send a body that will be rejected.
245            if effective_policy.is_reject() {
246                if let Some(expect) = parts.headers.get(hyper::header::EXPECT) {
247                    if expect == "100-continue" {
248                        crate::ops::global_counters()
249                            .body_rejections
250                            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
251                        crate::ops::Logger::global().emit(
252                            crate::ops::Event::new(
253                                crate::ops::Severity::Debug,
254                                crate::ops::EventKind::BodyPolicyRejection,
255                                "100-continue rejected by body policy",
256                            )
257                            .connection_id(conn_id),
258                        );
259                        let mut response = crate::response::payload_too_large(false);
260                        response.headers_mut().insert(
261                            hyper::header::CONNECTION,
262                            hyper::header::HeaderValue::from_static("close"),
263                        );
264                        return Ok::<_, Infallible>(finalize_runtime_response(response, &config));
265                    }
266                }
267            }
268
269            // Handle Reject policy — reject without invoking the service,
270            // but only if the request actually carries a body.
271            let has_body = declared_length.is_some_and(|len| len > 0)
272                || parts.headers.contains_key(hyper::header::TRANSFER_ENCODING);
273            if effective_policy.is_reject() && has_body {
274                crate::ops::global_counters()
275                    .body_rejections
276                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
277                crate::ops::Logger::global().emit(
278                    crate::ops::Event::new(
279                        crate::ops::Severity::Debug,
280                        crate::ops::EventKind::BodyPolicyRejection,
281                        "request body rejected by policy",
282                    )
283                    .connection_id(conn_id),
284                );
285                crate::ops::Logger::global().emit(
286                    crate::ops::Event::new(
287                        crate::ops::Severity::Debug,
288                        crate::ops::EventKind::ServiceInvocationSuppressed,
289                        "service invocation suppressed: body rejected by policy",
290                    )
291                    .connection_id(conn_id),
292                );
293                let mut response = crate::response::payload_too_large(false);
294                // Do not drain the body — drop it and close the connection to
295                // prevent unread bytes from being interpreted as a subsequent
296                // request. Hyper handles cleanup of the unconsumed body when
297                // the connection is dropped.
298                response.headers_mut().insert(
299                    hyper::header::CONNECTION,
300                    hyper::header::HeaderValue::from_static("close"),
301                );
302                return Ok::<_, Infallible>(finalize_runtime_response(response, &config));
303            }
304
305            // For Buffer/Stream policies, create RequestBody with proper limits.
306            // For Reject with no body, create an empty body (nothing to reject).
307            let body_limit = effective_policy.max_bytes().unwrap_or(u64::MAX);
308            let request_body = match &effective_policy {
309                RequestBodyPolicy::Reject => crate::primitives::request_body::RequestBody::empty(),
310                _ => crate::primitives::request_body::RequestBody::from_incoming(
311                    wrap_incoming_body(body),
312                    declared_length,
313                    body_limit,
314                ),
315            };
316
317            // Clone the consumption flag before the body is moved into Request.
318            let consumed_flag = request_body.consumed_flag();
319
320            // For Buffer policy, pre-buffer the body under timeout.
321            match &effective_policy {
322                RequestBodyPolicy::Reject => {
323                    // Reject with no body — proceed to service with empty body.
324                    let connection =
325                        build_connection_info(local_addr, remote_addr, tls, (*tls_info).clone());
326                    let request =
327                        crate::primitives::request::Request::new(head, request_body, connection);
328
329                    let result = tokio::time::timeout(handler_timeout, service.call(request)).await;
330
331                    let response = match result {
332                        Ok(Ok(canonical)) => {
333                            match crate::primitives::canonical::to_hyper_response_with_file_stream_semaphore(canonical, &file_stream_semaphore) {
334                                Ok(r) => r,
335                                Err(crate::primitives::canonical::ResponseConstructionError::FileStreamLimit) => crate::response::service_unavailable(),
336                                Err(_) => crate::response::internal_error(),
337                            }
338                        }
339                        Ok(Err(service_err)) => {
340                            let severity = if service_err.is_panic() || !service_err.is_timeout() {
341                                crate::ops::Severity::Error
342                            } else {
343                                crate::ops::Severity::Warn
344                            };
345                            crate::ops::Logger::global().emit(
346                                crate::ops::Event::new(
347                                    severity,
348                                    crate::ops::EventKind::ServiceError,
349                                    service_err.to_string(),
350                                )
351                                .connection_id(conn_id),
352                            );
353                            service_err.to_response()
354                        }
355                        Err(_elapsed) => {
356                            crate::ops::Logger::global().emit(crate::ops::Event::new(
357                                crate::ops::Severity::Warn,
358                                crate::ops::EventKind::ServiceTimeout,
359                                "handler timed out",
360                            ));
361                            ServiceError::timeout("handler timed out".to_string()).to_response()
362                        }
363                    };
364
365                    Ok::<_, Infallible>(finalize_runtime_response(response, &config))
366                }
367                RequestBodyPolicy::Buffer { .. } => {
368                    // Buffer: body is fully consumed during pre-buffering.
369                    // No incomplete body handling needed.
370                    let request_body = match tokio::time::timeout(
371                        body_read_timeout,
372                        request_body.read_all(),
373                    )
374                    .await
375                    {
376                        Ok(Ok(bytes)) => crate::primitives::request_body::RequestBody::from_bytes(
377                            bytes, body_limit,
378                        ),
379                        Ok(Err(err)) => {
380                            return Ok::<_, Infallible>(finalize_runtime_response(
381                                body_error_to_response(err, &head),
382                                &config,
383                            ));
384                        }
385                        Err(_elapsed) => {
386                            crate::ops::global_counters()
387                                .body_read_timeouts
388                                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
389                            crate::ops::Logger::global().emit(crate::ops::Event::new(
390                                crate::ops::Severity::Warn,
391                                crate::ops::EventKind::BodyReadTimeout,
392                                "body read timeout",
393                            ));
394                            let err = crate::primitives::request_body_error::RequestBodyError::ReadTimeout;
395                            return Ok::<_, Infallible>(finalize_runtime_response(
396                                body_error_to_response(err, &head),
397                                &config,
398                            ));
399                        }
400                    };
401
402                    let connection =
403                        build_connection_info(local_addr, remote_addr, tls, (*tls_info).clone());
404                    let request =
405                        crate::primitives::request::Request::new(head, request_body, connection);
406
407                    let result = tokio::time::timeout(handler_timeout, service.call(request)).await;
408
409                    let response = match result {
410                        Ok(Ok(canonical)) => {
411                            match crate::primitives::canonical::to_hyper_response_with_file_stream_semaphore(canonical, &file_stream_semaphore) {
412                                Ok(r) => r,
413                                Err(crate::primitives::canonical::ResponseConstructionError::FileStreamLimit) => crate::response::service_unavailable(),
414                                Err(_) => crate::response::internal_error(),
415                            }
416                        }
417                        Ok(Err(service_err)) => {
418                            let severity = if service_err.is_panic() || !service_err.is_timeout() {
419                                crate::ops::Severity::Error
420                            } else {
421                                crate::ops::Severity::Warn
422                            };
423                            crate::ops::Logger::global().emit(
424                                crate::ops::Event::new(
425                                    severity,
426                                    crate::ops::EventKind::ServiceError,
427                                    service_err.to_string(),
428                                )
429                                .connection_id(conn_id),
430                            );
431                            service_err.to_response()
432                        }
433                        Err(_elapsed) => {
434                            crate::ops::Logger::global().emit(crate::ops::Event::new(
435                                crate::ops::Severity::Warn,
436                                crate::ops::EventKind::ServiceTimeout,
437                                "handler timed out",
438                            ));
439                            ServiceError::timeout("handler timed out".to_string()).to_response()
440                        }
441                    };
442
443                    Ok::<_, Infallible>(finalize_runtime_response(response, &config))
444                }
445                RequestBodyPolicy::Stream { .. } => {
446                    // For Stream mode, enforce body_read_timeout as a total deadline
447                    // on the service call (which includes body consumption).
448                    let effective_timeout = body_read_timeout.min(handler_timeout);
449                    let connection =
450                        build_connection_info(local_addr, remote_addr, tls, (*tls_info).clone());
451                    let request =
452                        crate::primitives::request::Request::new(head, request_body, connection);
453
454                    let result =
455                        tokio::time::timeout(effective_timeout, service.call(request)).await;
456
457                    let response = match result {
458                        Ok(Ok(canonical)) => {
459                            match crate::primitives::canonical::to_hyper_response_with_file_stream_semaphore(canonical, &file_stream_semaphore) {
460                                Ok(r) => r,
461                                Err(crate::primitives::canonical::ResponseConstructionError::FileStreamLimit) => crate::response::service_unavailable(),
462                                Err(_) => crate::response::internal_error(),
463                            }
464                        }
465                        Ok(Err(service_err)) => {
466                            let severity = if service_err.is_panic() || !service_err.is_timeout() {
467                                crate::ops::Severity::Error
468                            } else {
469                                crate::ops::Severity::Warn
470                            };
471                            crate::ops::Logger::global().emit(
472                                crate::ops::Event::new(
473                                    severity,
474                                    crate::ops::EventKind::ServiceError,
475                                    service_err.to_string(),
476                                )
477                                .connection_id(conn_id),
478                            );
479                            service_err.to_response()
480                        }
481                        Err(_elapsed) => {
482                            crate::ops::Logger::global().emit(crate::ops::Event::new(
483                                crate::ops::Severity::Warn,
484                                crate::ops::EventKind::ServiceTimeout,
485                                "handler timed out",
486                            ));
487                            ServiceError::timeout("handler timed out".to_string()).to_response()
488                        }
489                    };
490
491                    // A stream that is not consumed to EOF cannot safely leave
492                    // unread bytes on an HTTP/1.1 connection. Close only in
493                    // that case; fully consumed streams remain reusable.
494                    let incomplete = !consumed_flag.load(std::sync::atomic::Ordering::Acquire);
495                    if incomplete {
496                        crate::ops::Logger::global().emit(
497                            crate::ops::Event::new(
498                                crate::ops::Severity::Debug,
499                                crate::ops::EventKind::IncompleteBodyClose,
500                                "service returned with unconsumed body; connection will close",
501                            )
502                            .connection_id(conn_id),
503                        );
504                    }
505
506                    let mut response = finalize_runtime_response(response, &config);
507                    if incomplete {
508                        response.headers_mut().insert(
509                            hyper::header::CONNECTION,
510                            hyper::header::HeaderValue::from_static("close"),
511                        );
512                    }
513                    Ok::<_, Infallible>(response)
514                }
515            }
516        }
517    });
518
519    serve_connection(io, hyper_service, &config, shutdown_rx, conn_id).await;
520}
521
522/// Select the effective body policy from service preference and runtime ceiling.
523fn select_body_policy(service_policy: RequestBodyPolicy, max_body_bytes: u64) -> RequestBodyPolicy {
524    match service_policy {
525        RequestBodyPolicy::Reject => RequestBodyPolicy::Reject,
526        RequestBodyPolicy::Buffer { max_bytes } => {
527            let effective = max_bytes.min(max_body_bytes);
528            if effective == 0 {
529                RequestBodyPolicy::Reject
530            } else {
531                RequestBodyPolicy::Buffer {
532                    max_bytes: effective,
533                }
534            }
535        }
536        RequestBodyPolicy::Stream { max_bytes } => {
537            let effective = max_bytes.min(max_body_bytes);
538            if effective == 0 {
539                RequestBodyPolicy::Reject
540            } else {
541                RequestBodyPolicy::Stream {
542                    max_bytes: effective,
543                }
544            }
545        }
546    }
547}
548
549/// Convert a RequestBodyError to an HTTP response.
550fn body_error_to_response(
551    err: crate::primitives::request_body_error::RequestBodyError,
552    _head: &crate::primitives::request_head::RequestHead,
553) -> hyper::Response<BoxBodyInner> {
554    let status = err.to_status_code();
555    let status =
556        hyper::StatusCode::from_u16(status).unwrap_or(hyper::StatusCode::INTERNAL_SERVER_ERROR);
557    let should_close = matches!(
558        status,
559        hyper::StatusCode::BAD_REQUEST
560            | hyper::StatusCode::REQUEST_TIMEOUT
561            | hyper::StatusCode::PAYLOAD_TOO_LARGE
562            | hyper::StatusCode::HTTP_VERSION_NOT_SUPPORTED
563    );
564    let body_text = match status.as_u16() {
565        400 => "400 Bad Request\n",
566        408 => "408 Request Timeout\n",
567        413 => "413 Payload Too Large\n",
568        501 => "501 Not Implemented\n",
569        _ => "500 Internal Server Error\n",
570    };
571    let is_head = _head.method().is_head();
572    let mut resp = crate::response::canonical_error(status, body_text, is_head);
573    if should_close {
574        resp.headers_mut().insert(
575            hyper::header::CONNECTION,
576            hyper::header::HeaderValue::from_static("close"),
577        );
578    }
579    resp
580}
581
582/// Build ConnectionInfo from real socket addresses.
583fn build_connection_info(
584    local_addr: std::net::SocketAddr,
585    remote_addr: std::net::SocketAddr,
586    tls: bool,
587    tls_info: Option<crate::primitives::connection_info::TlsInfo>,
588) -> crate::primitives::connection_info::ConnectionInfo {
589    crate::primitives::connection_info::ConnectionInfo {
590        local_addr,
591        remote_addr,
592        scheme: if tls {
593            crate::primitives::connection_info::Scheme::Https
594        } else {
595            crate::primitives::connection_info::Scheme::Http
596        },
597        tls: tls_info,
598    }
599}
600
601/// Apply runtime-owned response fields at the one final Hyper boundary.
602fn finalize_runtime_response(
603    mut response: hyper::Response<BoxBodyInner>,
604    config: &RuntimeConfig,
605) -> hyper::Response<BoxBodyInner> {
606    response.headers_mut().remove(hyper::header::SERVER);
607    if let Some(value) = &config.server_header {
608        if let Ok(value) = hyper::header::HeaderValue::from_str(value) {
609            response.headers_mut().insert(hyper::header::SERVER, value);
610        }
611    }
612    response
613}
614
615/// Validate body framing for ALL methods.
616///
617/// Rejects requests with duplicate Content-Length fields and TE+CL
618/// conflicts where both headers are visible. This is a hardened
619/// framing policy applied before body construction.
620///
621/// Note: Hyper 1.x strips the Content-Length header when
622/// Transfer-Encoding is present. In that case, Hyper's own behavior
623/// prevents request smuggling — it processes the chunked body and
624/// ignores the removed Content-Length. The duplicate-CL check remains
625/// because Hyper does not strip duplicate CL fields.
626fn validate_body_framing(headers: &hyper::HeaderMap) -> Result<(), ServiceError> {
627    let has_te = headers.contains_key(hyper::header::TRANSFER_ENCODING);
628    let cl_values: Vec<_> = headers
629        .get_all(hyper::header::CONTENT_LENGTH)
630        .iter()
631        .collect();
632    let has_cl = !cl_values.is_empty();
633    let duplicate_cl = cl_values.len() > 1;
634
635    if has_te && has_cl {
636        return Err(ServiceError::rejected(
637            400,
638            "conflicting Transfer-Encoding and Content-Length",
639        ));
640    }
641
642    if duplicate_cl {
643        return Err(ServiceError::rejected(
644            400,
645            "duplicate Content-Length headers",
646        ));
647    }
648
649    Ok(())
650}
651
652/// Wrap a Hyper `Incoming` body into a `Stream<Item = Result<Bytes, IncomingError>>`.
653///
654/// This bridges the Hyper body type to the canonical `RequestBody` type
655/// without leaking Hyper into the public API.
656fn wrap_incoming_body(
657    body: Incoming,
658) -> impl futures_util::Stream<
659    Item = Result<bytes::Bytes, crate::primitives::request_body::IncomingError>,
660> + Send
661       + 'static {
662    use futures_util::StreamExt;
663    http_body_util::BodyStream::new(body).filter_map(|result| async {
664        match result {
665            Ok(frame) => frame.into_data().ok().map(Ok),
666            Err(e) => Some(Err(crate::primitives::request_body::IncomingError(
667                e.to_string(),
668            ))),
669        }
670    })
671}
672
673/// Convert a Hyper request to a canonical [`RequestHead`].
674///
675/// This extracts method, URI, version, and headers from the Hyper request
676/// and constructs a canonical [`RequestHead`]. The body is not included —
677/// the runtime handles body rejection before service invocation.
678fn convert_request_head(
679    req: &Request<Incoming>,
680) -> Result<crate::primitives::request_head::RequestHead, ServiceError> {
681    use crate::primitives::header_block::HeaderBlock;
682    use crate::primitives::method::Method;
683    use crate::primitives::request_target::RequestTarget;
684    use crate::primitives::version::HttpVersion;
685
686    let method = match req.method().as_str() {
687        "GET" => Method::get(),
688        "HEAD" => Method::head(),
689        "POST" => Method::post(),
690        "PUT" => Method::put(),
691        "DELETE" => Method::delete(),
692        "PATCH" => Method::patch(),
693        "OPTIONS" => Method::options(),
694        "TRACE" => Method::trace(),
695        other => Method::new(other)
696            .map_err(|_| ServiceError::rejected(400, format!("invalid method: {}", other)))?,
697    };
698
699    let version = match req.version() {
700        hyper::Version::HTTP_10 => HttpVersion::Http10,
701        hyper::Version::HTTP_11 => HttpVersion::Http11,
702        other => {
703            return Err(ServiceError::rejected(
704                505,
705                format!("unsupported HTTP version: {:?}", other),
706            ))
707        }
708    };
709
710    let raw_target = req
711        .uri()
712        .path_and_query()
713        .map(|pq| pq.as_str())
714        .unwrap_or("/");
715
716    // Reject absolute-form URIs (authority present in raw target).
717    // Hyper strips scheme/authority from path_and_query, so we must check
718    // the full URI string.
719    if req.uri().scheme_str().is_some() {
720        return Err(ServiceError::rejected(
721            400,
722            "absolute-form request target not allowed",
723        ));
724    }
725
726    // Asterisk-form (`*`) is rejected as method-not-allowed (405) rather
727    // than bad-request (400) because the method check must fire before the
728    // target-form check per the release contract.
729    if raw_target == "*" {
730        return Err(ServiceError::rejected(
731            405,
732            format!("method not allowed: {}", method.as_str()),
733        ));
734    }
735
736    let target = RequestTarget::parse(raw_target)
737        .map_err(|e| ServiceError::rejected(400, format!("invalid request target: {}", e)))?;
738
739    let mut headers = HeaderBlock::new();
740    for (name, value) in req.headers().iter() {
741        let header_name = crate::primitives::header_block::HeaderName::new(name.as_str())
742            .map_err(|_| ServiceError::rejected(400, format!("invalid header name: {}", name)))?;
743        let header_value = match value.to_str() {
744            Ok(v) => crate::primitives::header_block::HeaderValue::new(v).map_err(|_| {
745                ServiceError::rejected(400, format!("invalid header value for {}", name))
746            })?,
747            Err(_) => {
748                return Err(ServiceError::rejected(
749                    400,
750                    format!("non-UTF-8 header value for {}", name),
751                ))
752            }
753        };
754        headers.push(header_name, header_value);
755    }
756
757    Ok(crate::primitives::request_head::RequestHead::new(
758        method, target, version, headers,
759    ))
760}
761
762#[cfg(test)]
763mod tests {
764    use super::*;
765    use crate::config::{ServeConfig, ServeState};
766    use crate::server::static_service::StaticService;
767    use std::sync::Arc;
768    use tempfile::TempDir;
769    use tokio::io::{AsyncReadExt, AsyncWriteExt};
770    use tokio::net::TcpListener;
771
772    fn build_state(tmp: &TempDir) -> Arc<ServeState> {
773        let config = Arc::new(ServeConfig {
774            root: tmp.path().to_path_buf(),
775            ..ServeConfig::default()
776        });
777        Arc::new(ServeState::new(config).unwrap())
778    }
779
780    #[tokio::test]
781    async fn serve_connection_handles_get() {
782        let tmp = TempDir::new().unwrap();
783        std::fs::write(tmp.path().join("hello.txt"), "hello").unwrap();
784        let state = build_state(&tmp);
785        let config = RuntimeConfig::default();
786
787        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
788        let addr = listener.local_addr().unwrap();
789        let (tx, _rx) = broadcast::channel::<()>(1);
790
791        let state_clone = state.clone();
792        let server = tokio::spawn(async move {
793            let (stream, remote_addr) = listener.accept().await.unwrap();
794            let mut shutdown_rx = tx.subscribe();
795            let runtime_state = Arc::new(RuntimeState::new(&config));
796            serve_connection_with_runtime_state(
797                TokioIo::new(stream),
798                StaticService::from_state(state_clone),
799                &config,
800                runtime_state,
801                &mut shutdown_rx,
802                1,
803                addr,
804                remote_addr,
805                false,
806                None,
807            )
808            .await;
809        });
810
811        let mut client = tokio::net::TcpStream::connect(addr).await.unwrap();
812        client
813            .write_all(b"GET /hello.txt HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
814            .await
815            .unwrap();
816        let mut buf = Vec::new();
817        client.read_to_end(&mut buf).await.unwrap();
818
819        let _ = server.await;
820
821        let response = String::from_utf8_lossy(&buf);
822        assert!(
823            response.starts_with("HTTP/1.1 200 OK"),
824            "unexpected response: {}",
825            response
826        );
827    }
828
829    #[test]
830    fn runtime_server_header_replaces_service_value() {
831        let config = RuntimeConfig::builder()
832            .server_header("eggserve-test".into())
833            .build()
834            .unwrap();
835        let mut response = crate::response::not_found(false);
836        response.headers_mut().insert(
837            hyper::header::SERVER,
838            hyper::header::HeaderValue::from_static("spoofed"),
839        );
840        let response = finalize_runtime_response(response, &config);
841        assert_eq!(
842            response.headers().get(hyper::header::SERVER).unwrap(),
843            "eggserve-test"
844        );
845        assert_eq!(
846            response
847                .headers()
848                .get_all(hyper::header::SERVER)
849                .iter()
850                .count(),
851            1
852        );
853    }
854}
eggserve_core/server/connection.rs

eggserve_core/server/
connection.rs
