rskit_server/
http.rs

1use std::future::Future;
2use std::net::SocketAddr;
3use std::pin::pin;
4use std::sync::Arc;
5use std::time::Duration;
6
7use async_trait::async_trait;
8use axum::body::Body;
9use axum::extract::DefaultBodyLimit;
10use axum::serve::Listener;
11use axum::{Router, http::StatusCode};
12use hyper::body::Incoming;
13use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
14use hyper_util::server::conn::auto::Builder as HyperBuilder;
15use hyper_util::service::TowerToHyperService;
16use parking_lot::Mutex;
17use rskit_bootstrap::{Component, Health, Registry};
18use rskit_errors::{AppError, AppResult, ErrorCode};
19use rskit_http::{SecurityHeadersConfig, SecurityHeadersLayer};
20use rskit_security::{TlsConfig, TlsVersion};
21use rustls::pki_types::{CertificateDer, PrivateKeyDer, pem::PemObject};
22use tokio::io::{AsyncRead, AsyncWrite};
23use tokio::net::TcpListener;
24use tokio_rustls::TlsAcceptor;
25use tokio_util::sync::CancellationToken;
26use tower::ServiceExt;
27use tower_http::{
28    request_id::{MakeRequestUuid, SetRequestIdLayer},
29    timeout::TimeoutLayer,
30    trace::TraceLayer,
31};
32
33use crate::http_config::{HttpServerConfig, validate_http_tls_config};
34use crate::middleware::HttpMiddlewareStack;
35
36/// Builder for [`HttpServer`].
37pub struct HttpServerBuilder {
38    config: HttpServerConfig,
39    cancel: CancellationToken,
40    router: Router,
41    middleware: HttpMiddlewareStack,
42    security_headers: Option<SecurityHeadersConfig>,
43}
44
45impl HttpServerBuilder {
46    /// Create a new builder.
47    pub fn new(config: HttpServerConfig, cancel: CancellationToken) -> Self {
48        Self {
49            config,
50            cancel,
51            router: Router::new(),
52            middleware: HttpMiddlewareStack::new(),
53            security_headers: None,
54        }
55    }
56
57    /// Merge an axum [`Router`] into the server.
58    #[must_use]
59    pub fn with_router(mut self, router: Router) -> Self {
60        self.router = self.router.merge(router);
61        self
62    }
63
64    /// Replace the ordered middleware stack.
65    #[must_use]
66    pub fn with_middleware_stack(mut self, middleware: HttpMiddlewareStack) -> Self {
67        self.middleware = middleware;
68        self
69    }
70
71    /// Append a logging-phase transform.
72    #[must_use]
73    pub fn with_logging_transform<F>(mut self, transform: F) -> Self
74    where
75        F: Fn(Router) -> Router + Send + Sync + 'static,
76    {
77        self.middleware = self.middleware.with_logging_transform(transform);
78        self
79    }
80
81    /// Append an auth-phase transform.
82    #[must_use]
83    pub fn with_auth_transform<F>(mut self, transform: F) -> Self
84    where
85        F: Fn(Router) -> Router + Send + Sync + 'static,
86    {
87        self.middleware = self.middleware.with_auth_transform(transform);
88        self
89    }
90
91    /// Append a validation-phase transform.
92    #[must_use]
93    pub fn with_validation_transform<F>(mut self, transform: F) -> Self
94    where
95        F: Fn(Router) -> Router + Send + Sync + 'static,
96    {
97        self.middleware = self.middleware.with_validation_transform(transform);
98        self
99    }
100
101    /// Append a metrics-phase transform.
102    #[must_use]
103    pub fn with_metrics_transform<F>(mut self, transform: F) -> Self
104    where
105        F: Fn(Router) -> Router + Send + Sync + 'static,
106    {
107        self.middleware = self.middleware.with_metrics_transform(transform);
108        self
109    }
110
111    /// Apply CORS from the server config (no-op if `cors` is `None`).
112    ///
113    /// # Errors
114    /// Returns an error when the configured CORS policy contains invalid origins,
115    /// methods, headers, or max-age values.
116    #[must_use = "builder methods return a new builder; use the returned value"]
117    pub fn with_cors(self) -> AppResult<Self> {
118        if let Some(cors_cfg) = self.config.cors.as_ref() {
119            let _ = cors_cfg.layer()?;
120        }
121        Ok(self)
122    }
123
124    /// Add automatic `X-Request-Id` injection.
125    #[must_use]
126    pub fn with_request_id(self) -> Self {
127        self
128    }
129
130    /// Add secure response headers using the default security policy.
131    ///
132    /// # Errors
133    /// Returns an error if the default security policy cannot be built (should never happen
134    /// in practice — this is a programming error guard).
135    #[must_use = "builder methods return a new builder; use the returned value"]
136    pub fn with_security_headers(self) -> AppResult<Self> {
137        self.with_security_headers_config(SecurityHeadersConfig::default())
138    }
139
140    /// Add secure response headers using an explicit security policy.
141    ///
142    /// # Errors
143    /// Returns an error when the supplied policy is invalid.
144    #[must_use = "builder methods return a new builder; use the returned value"]
145    pub fn with_security_headers_config(
146        mut self,
147        config: SecurityHeadersConfig,
148    ) -> AppResult<Self> {
149        let _ = SecurityHeadersLayer::new(&config)?;
150        self.security_headers = Some(config);
151        Ok(self)
152    }
153
154    /// Add the canonical tracing phase.
155    ///
156    /// Tracing is owned by [`build`](Self::build) so this remains a compatibility no-op.
157    #[must_use]
158    pub fn with_tracing(self) -> Self {
159        self
160    }
161
162    /// Consume the builder and produce an [`HttpServer`].
163    /// # Errors
164    /// Returns an error when baseline transport middleware configuration is invalid.
165    pub fn build(self) -> AppResult<HttpServer> {
166        let builder = self;
167        let security_headers = builder.security_headers.clone();
168        let request_timeout = builder.config.request_timeout;
169        let max_body_bytes = builder.config.max_body_bytes;
170        let cors = builder.config.cors.clone();
171        let router = builder.middleware.apply(builder.router);
172        let router = apply_canonical_tracing(router);
173        let router = apply_baseline_layers(
174            router,
175            security_headers,
176            request_timeout,
177            max_body_bytes,
178            cors,
179        )?;
180        Ok(HttpServer {
181            config: Arc::new(builder.config),
182            cancel: builder.cancel,
183            router: Arc::new(tokio::sync::Mutex::new(Some(router))),
184            local_addr: Arc::new(Mutex::new(None)),
185        })
186    }
187
188    /// Add the configured request body limit.
189    #[must_use]
190    pub fn with_body_limit(self) -> Self {
191        self
192    }
193
194    /// Add the configured request timeout.
195    #[must_use]
196    pub fn with_timeout(self) -> Self {
197        self
198    }
199}
200
201fn apply_canonical_tracing(router: Router) -> Router {
202    use http::Request;
203    use tower_http::trace::DefaultOnResponse;
204    use tracing::Level;
205
206    let trace_layer = TraceLayer::new_for_http()
207        .make_span_with(|request: &Request<_>| {
208            let path = request.uri().path();
209            tracing::info_span!(
210                "http_request",
211                method = %request.method(),
212                "http.target" = path,
213                status_code = tracing::field::Empty,
214            )
215        })
216        .on_response(DefaultOnResponse::new().level(Level::INFO));
217
218    router.layer(trace_layer)
219}
220
221fn apply_baseline_layers(
222    router: Router,
223    security_headers: Option<SecurityHeadersConfig>,
224    request_timeout: Duration,
225    max_body_bytes: usize,
226    cors: Option<crate::CorsPolicy>,
227) -> AppResult<Router> {
228    let security_headers = SecurityHeadersLayer::new(&security_headers.unwrap_or_default())?;
229    let router = router
230        .layer(TimeoutLayer::with_status_code(
231            StatusCode::REQUEST_TIMEOUT,
232            request_timeout,
233        ))
234        .layer(DefaultBodyLimit::max(max_body_bytes))
235        .layer(security_headers);
236    let router = if let Some(cors_cfg) = cors.as_ref() {
237        router.layer(cors_cfg.layer()?)
238    } else {
239        router
240    };
241    Ok(router.layer(SetRequestIdLayer::x_request_id(MakeRequestUuid)))
242}
243
244/// HTTP server that implements the [`Component`] lifecycle.
245pub struct HttpServer {
246    config: Arc<HttpServerConfig>,
247    cancel: CancellationToken,
248    router: Arc<tokio::sync::Mutex<Option<Router>>>,
249    local_addr: Arc<Mutex<Option<SocketAddr>>>,
250}
251
252impl HttpServer {
253    /// Bind address as `host:port`.
254    #[must_use]
255    pub fn bind_addr(&self) -> String {
256        self.config.bind_addr()
257    }
258
259    /// Actual local socket address after [`start`](Component::start) binds.
260    #[must_use]
261    pub fn local_addr(&self) -> Option<SocketAddr> {
262        *self.local_addr.lock()
263    }
264}
265
266#[async_trait]
267impl Component for HttpServer {
268    fn name(&self) -> &str {
269        "http-server"
270    }
271
272    async fn start(&self) -> AppResult<()> {
273        let router = self
274            .router
275            .lock()
276            .await
277            .take()
278            .ok_or_else(|| AppError::new(ErrorCode::Internal, "HTTP server already started"))?;
279
280        let addr: SocketAddr = self.config.bind_addr().parse().map_err(|error| {
281            AppError::new(
282                ErrorCode::Internal,
283                format!("invalid bind address: {error}"),
284            )
285        })?;
286        let tls_acceptor = if let Some(tls) = &self.config.tls {
287            Some(build_tls_acceptor(tls)?)
288        } else {
289            None
290        };
291
292        let listener = TcpListener::bind(addr).await.map_err(|error| {
293            AppError::new(
294                ErrorCode::Internal,
295                format!("HTTP server bind failed for {addr}: {error}"),
296            )
297        })?;
298        let actual_addr = listener.local_addr().map_err(|error| {
299            AppError::new(
300                ErrorCode::Internal,
301                format!("failed to inspect HTTP server local address: {error}"),
302            )
303        })?;
304        *self.local_addr.lock() = Some(actual_addr);
305
306        let cancel = self.cancel.clone();
307        let config = Arc::clone(&self.config);
308        tokio::spawn(async move {
309            if let Some(acceptor) = tls_acceptor {
310                tracing::info!(addr = %actual_addr, "HTTPS server listening");
311                serve_tls_listener(listener, acceptor, router, Arc::clone(&config), cancel).await;
312            } else {
313                tracing::info!(addr = %actual_addr, "HTTP server listening");
314                serve_listener(
315                    listener,
316                    router,
317                    Arc::clone(&config),
318                    cancel,
319                    config.enable_h2c,
320                )
321                .await;
322            }
323        });
324
325        Ok(())
326    }
327
328    async fn stop(&self) -> AppResult<()> {
329        self.cancel.cancel();
330        Ok(())
331    }
332
333    fn health(&self) -> Health {
334        Health::healthy("http-server")
335    }
336}
337
338async fn serve_listener<L>(
339    mut listener: L,
340    router: Router,
341    config: Arc<HttpServerConfig>,
342    cancel: CancellationToken,
343    allow_h2: bool,
344) where
345    L: Listener<Addr = SocketAddr> + Send + 'static,
346    L::Io: AsyncRead + AsyncWrite + Unpin + Send + 'static,
347{
348    loop {
349        tokio::select! {
350            () = cancel.cancelled() => break,
351            (io, addr) = listener.accept() => {
352                let router = router.clone();
353                let config = Arc::clone(&config);
354                let shutdown = cancel.clone().cancelled_owned();
355                tokio::spawn(async move {
356                    serve_connection(io, addr, router, config, shutdown, allow_h2).await;
357                });
358            }
359        }
360    }
361}
362
363async fn serve_tls_listener(
364    listener: TcpListener,
365    acceptor: TlsAcceptor,
366    router: Router,
367    config: Arc<HttpServerConfig>,
368    cancel: CancellationToken,
369) {
370    loop {
371        tokio::select! {
372            () = cancel.cancelled() => break,
373            result = listener.accept() => {
374                match result {
375                    Ok((stream, addr)) => {
376                        let acceptor = acceptor.clone();
377                        let router = router.clone();
378                        let config = Arc::clone(&config);
379                        let shutdown = cancel.clone().cancelled_owned();
380                        tokio::spawn(async move {
381                            match tokio::time::timeout(config.read_timeout, acceptor.accept(stream)).await {
382                                Ok(Ok(stream)) => {
383                                    serve_connection(stream, addr, router, config, shutdown, true).await;
384                                }
385                                Ok(Err(error)) => {
386                                    tracing::warn!(error = ?error, %addr, "TLS handshake failed");
387                                }
388                                Err(_) => {
389                                    tracing::warn!(%addr, "TLS handshake timed out");
390                                }
391                            }
392                        });
393                    }
394                    Err(error) if is_connection_accept_error(&error) => {}
395                    Err(error) => {
396                        tracing::error!(error = ?error, "HTTP TLS accept failed");
397                        tokio::time::sleep(Duration::from_secs(1)).await;
398                    }
399                }
400            }
401        }
402    }
403}
404
405async fn serve_connection<I, F>(
406    io: I,
407    addr: SocketAddr,
408    router: Router,
409    config: Arc<HttpServerConfig>,
410    shutdown: F,
411    allow_h2: bool,
412) where
413    I: AsyncRead + AsyncWrite + Unpin + Send + 'static,
414    F: Future<Output = ()> + Send + 'static,
415{
416    let service = router.map_request(|request: http::Request<Incoming>| request.map(Body::new));
417    let hyper_service = TowerToHyperService::new(service);
418    let mut builder = HyperBuilder::new(TokioExecutor::new());
419    builder
420        .http1()
421        .timer(TokioTimer::new())
422        .header_read_timeout(Some(config.read_timeout))
423        .keep_alive(true);
424    builder
425        .http2()
426        .timer(TokioTimer::new())
427        .keep_alive_interval(Some(config.idle_timeout))
428        .keep_alive_timeout(config.read_timeout)
429        .enable_connect_protocol();
430
431    let io = TokioIo::new(io);
432    if allow_h2 {
433        let mut connection = pin!(builder.serve_connection_with_upgrades(io, hyper_service));
434        let mut shutdown = pin!(shutdown);
435
436        tokio::select! {
437            result = connection.as_mut() => {
438                if let Err(error) = result {
439                    tracing::debug!(error = ?error, %addr, "HTTP connection finished with error");
440                }
441            }
442            () = &mut shutdown => {
443                connection.as_mut().graceful_shutdown();
444                if let Err(error) = connection.as_mut().await {
445                    tracing::debug!(error = ?error, %addr, "HTTP connection finished with error after graceful shutdown");
446                }
447            }
448        }
449    } else {
450        let http1 = builder.http1();
451        let mut connection = pin!(http1.serve_connection_with_upgrades(io, hyper_service));
452        let mut shutdown = pin!(shutdown);
453
454        tokio::select! {
455            result = connection.as_mut() => {
456                if let Err(error) = result {
457                    tracing::debug!(error = ?error, %addr, "HTTP/1 connection finished with error");
458                }
459            }
460            () = &mut shutdown => {
461                connection.as_mut().graceful_shutdown();
462                if let Err(error) = connection.as_mut().await {
463                    tracing::debug!(error = ?error, %addr, "HTTP/1 connection finished with error after graceful shutdown");
464                }
465            }
466        }
467    }
468}
469
470fn is_connection_accept_error(error: &std::io::Error) -> bool {
471    matches!(
472        error.kind(),
473        std::io::ErrorKind::ConnectionRefused
474            | std::io::ErrorKind::ConnectionAborted
475            | std::io::ErrorKind::ConnectionReset
476    )
477}
478
479fn build_tls_acceptor(tls: &TlsConfig) -> AppResult<TlsAcceptor> {
480    validate_http_tls_config(tls)?;
481    let cert_file = tls.cert_file.as_deref().ok_or_else(|| {
482        AppError::invalid_input("tls.cert_file", "cert_file is required for HTTPS serving")
483    })?;
484    let key_file = tls.key_file.as_deref().ok_or_else(|| {
485        AppError::invalid_input("tls.key_file", "key_file is required for HTTPS serving")
486    })?;
487
488    let certs = load_certs(cert_file)?;
489    let key = load_private_key(key_file)?;
490    let versions = match tls.min_version {
491        TlsVersion::Tls12 => vec![&rustls::version::TLS13, &rustls::version::TLS12],
492        TlsVersion::Tls13 => vec![&rustls::version::TLS13],
493        _ => vec![&rustls::version::TLS13],
494    };
495    let mut config = rustls::ServerConfig::builder_with_protocol_versions(&versions)
496        .with_no_client_auth()
497        .with_single_cert(certs, key)
498        .map_err(|error| {
499            AppError::new(
500                ErrorCode::InvalidInput,
501                format!("invalid HTTP TLS certificate/key pair: {error}"),
502            )
503            .with_cause(error)
504        })?;
505    config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
506    Ok(TlsAcceptor::from(Arc::new(config)))
507}
508
509fn load_certs(path: &str) -> AppResult<Vec<CertificateDer<'static>>> {
510    let certs = CertificateDer::pem_file_iter(path)
511        .map_err(|error| {
512            AppError::new(
513                ErrorCode::InvalidInput,
514                format!("failed to load HTTP TLS certificate file '{path}': {error}"),
515            )
516            .with_cause(error)
517        })?
518        .collect::<Result<Vec<_>, _>>()
519        .map_err(|error| {
520            AppError::new(
521                ErrorCode::InvalidInput,
522                format!("failed to parse HTTP TLS certificate file '{path}': {error}"),
523            )
524            .with_cause(error)
525        })?;
526    if certs.is_empty() {
527        return Err(AppError::invalid_input(
528            "tls.cert_file",
529            "certificate file must contain at least one certificate",
530        ));
531    }
532    Ok(certs)
533}
534
535fn load_private_key(path: &str) -> AppResult<PrivateKeyDer<'static>> {
536    PrivateKeyDer::from_pem_file(path).map_err(|error| {
537        AppError::new(
538            ErrorCode::InvalidInput,
539            format!("failed to load HTTP TLS key file '{path}': {error}"),
540        )
541        .with_cause(error)
542    })
543}
544
545/// Add a `/health` endpoint returning JSON from a [`Registry`].
546pub fn health_router(registry: Arc<Registry>) -> Router {
547    use axum::{Json, routing::get};
548
549    Router::new().route(
550        "/health",
551        get({
552            let registry = Arc::clone(&registry);
553            move || {
554                let registry = Arc::clone(&registry);
555                async move {
556                    let healths = registry.health_all();
557                    let all_ok = healths.iter().all(|health| health.is_healthy());
558                    let status = if all_ok {
559                        axum::http::StatusCode::OK
560                    } else {
561                        axum::http::StatusCode::SERVICE_UNAVAILABLE
562                    };
563                    (status, Json(healths))
564                }
565            }
566        }),
567    )
568}
569
570/// Returns a router with a `/healthz` liveness probe endpoint.
571pub fn healthz_router() -> Router {
572    use axum::{Json, routing::get};
573    use serde::Serialize;
574
575    #[derive(Serialize)]
576    struct LivenessResponse {
577        status: &'static str,
578        version: &'static str,
579    }
580
581    async fn healthz_handler() -> Json<LivenessResponse> {
582        Json(LivenessResponse {
583            status: "ok",
584            version: rskit_version::package_version(),
585        })
586    }
587
588    Router::new().route("/healthz", get(healthz_handler))
589}
590
591#[cfg(test)]
592mod tests {
593    use std::sync::Arc;
594    use std::sync::atomic::{AtomicUsize, Ordering};
595
596    use axum::{http::Request, routing::get};
597    use rskit_bootstrap::{Component, Registry};
598    use rskit_http::CorsPolicy;
599    use rskit_security::TransportSecurity;
600    use tokio::io::{AsyncReadExt, AsyncWriteExt};
601    use tower::ServiceExt;
602
603    use super::*;
604
605    fn local_config() -> HttpServerConfig {
606        HttpServerConfig {
607            host: "127.0.0.1".to_string(),
608            port: 0,
609            request_timeout: Duration::from_secs(1),
610            ..Default::default()
611        }
612    }
613
614    #[tokio::test]
615    async fn builder_applies_baseline_layers_to_application_routes() {
616        let server = HttpServerBuilder::new(local_config(), CancellationToken::new())
617            .with_router(Router::new().route("/", get(|| async { "ok" })))
618            .with_security_headers_config(
619                SecurityHeadersConfig::default()
620                    .with_transport_security(TransportSecurity::AllowInsecureLocal),
621            )
622            .expect("configure security headers")
623            .with_request_id()
624            .with_tracing()
625            .with_body_limit()
626            .with_timeout()
627            .build()
628            .expect("build server");
629        let router = server
630            .router
631            .lock()
632            .await
633            .take()
634            .expect("router is present");
635
636        let response = router
637            .oneshot(
638                Request::builder()
639                    .uri("/")
640                    .body(Body::empty())
641                    .expect("request"),
642            )
643            .await
644            .expect("route response");
645
646        assert_eq!(response.status(), StatusCode::OK);
647        assert_eq!(
648            response
649                .headers()
650                .get(http::header::X_CONTENT_TYPE_OPTIONS)
651                .expect("x-content-type-options header"),
652            "nosniff"
653        );
654    }
655
656    #[tokio::test]
657    async fn builder_applies_ordered_transform_helpers_and_cors() {
658        let calls = Arc::new(AtomicUsize::new(0));
659        let transform = |calls: Arc<AtomicUsize>| {
660            move |router: Router| {
661                calls.fetch_add(1, Ordering::SeqCst);
662                router
663            }
664        };
665
666        let server = HttpServerBuilder::new(local_config(), CancellationToken::new())
667            .with_router(Router::new().route("/", get(|| async { "ok" })))
668            .with_logging_transform(transform(Arc::clone(&calls)))
669            .with_auth_transform(transform(Arc::clone(&calls)))
670            .with_validation_transform(transform(Arc::clone(&calls)))
671            .with_metrics_transform(transform(Arc::clone(&calls)))
672            .with_cors()
673            .expect("default cors state is valid")
674            .build()
675            .expect("build server");
676
677        let router = server
678            .router
679            .lock()
680            .await
681            .take()
682            .expect("router is present");
683        let response = router
684            .oneshot(
685                Request::builder()
686                    .uri("/")
687                    .body(Body::empty())
688                    .expect("request"),
689            )
690            .await
691            .expect("route response");
692
693        assert_eq!(response.status(), StatusCode::OK);
694        assert_eq!(calls.load(Ordering::SeqCst), 4);
695        assert_eq!(calls.load(Ordering::SeqCst), 4);
696    }
697    #[test]
698    fn builder_helpers_validate_cors_security_headers_and_middleware_stack() {
699        let invalid_cors = HttpServerConfig {
700            cors: Some(CorsPolicy {
701                allow_credentials: true,
702                ..Default::default()
703            }),
704            ..local_config()
705        };
706        assert_eq!(
707            HttpServerBuilder::new(invalid_cors, CancellationToken::new())
708                .with_cors()
709                .err()
710                .unwrap()
711                .code(),
712            ErrorCode::InvalidInput
713        );
714        let invalid_cors = HttpServerConfig {
715            cors: Some(CorsPolicy {
716                allowed_origins: vec!["*".to_string()],
717                ..Default::default()
718            }),
719            ..local_config()
720        };
721        assert_eq!(
722            HttpServerBuilder::new(invalid_cors, CancellationToken::new())
723                .build()
724                .err()
725                .unwrap()
726                .code(),
727            ErrorCode::InvalidInput
728        );
729
730        let builder = HttpServerBuilder::new(local_config(), CancellationToken::new())
731            .with_middleware_stack(HttpMiddlewareStack::new())
732            .with_security_headers()
733            .expect("default security headers are valid");
734        assert!(builder.security_headers.is_some());
735    }
736
737    #[test]
738    fn builder_builds_with_valid_cors_and_exposes_server_accessors() {
739        let config = HttpServerConfig {
740            cors: Some(CorsPolicy {
741                allowed_origins: vec!["https://example.com".to_string()],
742                allowed_methods: vec!["GET".to_string()],
743                allowed_headers: vec!["x-test".to_string()],
744                ..Default::default()
745            }),
746            ..local_config()
747        };
748        let server = HttpServerBuilder::new(config, CancellationToken::new())
749            .with_router(Router::new())
750            .build()
751            .expect("valid CORS policy builds");
752
753        assert_eq!(server.name(), "http-server");
754        assert_eq!(server.bind_addr(), "127.0.0.1:0");
755        assert_eq!(server.local_addr(), None);
756    }
757
758    #[tokio::test]
759    async fn builder_applies_ordered_transform_shortcuts() {
760        let server = HttpServerBuilder::new(local_config(), CancellationToken::new())
761            .with_logging_transform(|router| router.route("/logging", get(|| async { "logging" })))
762            .with_auth_transform(|router| router.route("/auth", get(|| async { "auth" })))
763            .with_validation_transform(|router| {
764                router.route("/validation", get(|| async { "validation" }))
765            })
766            .with_metrics_transform(|router| router.route("/metrics", get(|| async { "metrics" })))
767            .with_cors()
768            .expect("empty cors config is accepted")
769            .build()
770            .expect("build server");
771        let router = server
772            .router
773            .lock()
774            .await
775            .take()
776            .expect("router is present");
777
778        for (path, expected) in [
779            ("/logging", "logging"),
780            ("/auth", "auth"),
781            ("/validation", "validation"),
782            ("/metrics", "metrics"),
783        ] {
784            let response = router
785                .clone()
786                .oneshot(
787                    Request::builder()
788                        .uri(path)
789                        .body(Body::empty())
790                        .expect("request"),
791                )
792                .await
793                .expect("route response");
794            let body = axum::body::to_bytes(response.into_body(), usize::MAX)
795                .await
796                .expect("body bytes");
797            assert_eq!(&body[..], expected.as_bytes());
798        }
799    }
800
801    #[test]
802    fn security_header_configuration_is_validated_at_build_time() {
803        let config = SecurityHeadersConfig::default()
804            .with_transport_security(TransportSecurity::AllowInsecureLocal)
805            .with_content_security_policy(None)
806            .with_permissions_policy(None)
807            .with_referrer_policy(None)
808            .with_frame_options(None)
809            .with_content_type_options(None);
810
811        let error = match HttpServerBuilder::new(local_config(), CancellationToken::new())
812            .with_security_headers_config(config)
813        {
814            Ok(_) => panic!("invalid security header policy should be rejected"),
815            Err(error) => error,
816        };
817
818        assert_eq!(error.code(), ErrorCode::InvalidInput);
819    }
820
821    #[tokio::test]
822    async fn lifecycle_sets_local_address_and_cancels_shutdown() {
823        let server = HttpServerBuilder::new(local_config(), CancellationToken::new())
824            .build()
825            .expect("build server");
826
827        assert_eq!(server.bind_addr(), "127.0.0.1:0");
828        assert!(server.local_addr().is_none());
829        assert!(server.health().is_healthy());
830
831        server.start().await.expect("start http server");
832        assert!(server.local_addr().is_some());
833        server.stop().await.expect("stop http server");
834    }
835
836    #[tokio::test]
837    async fn local_http_listener_serves_requests_and_rejects_double_start() {
838        let server = HttpServerBuilder::new(local_config(), CancellationToken::new())
839            .with_router(Router::new().route("/ping", get(|| async { "pong" })))
840            .build()
841            .expect("build server");
842
843        server.start().await.expect("start http server");
844        let second_start = server.start().await.unwrap_err();
845        assert_eq!(second_start.code(), ErrorCode::Internal);
846        assert!(second_start.message().contains("already started"));
847
848        let addr = server.local_addr().expect("local address");
849        let mut stream = tokio::net::TcpStream::connect(addr)
850            .await
851            .expect("connect to local server");
852        stream
853            .write_all(b"GET /ping HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
854            .await
855            .expect("write request");
856        let mut response = String::new();
857        tokio::time::timeout(Duration::from_secs(2), stream.read_to_string(&mut response))
858            .await
859            .expect("response read timed out")
860            .expect("read response");
861
862        assert!(response.starts_with("HTTP/1.1 200 OK"), "{response}");
863        assert!(response.contains("pong"), "{response}");
864        server.stop().await.expect("stop http server");
865    }
866
867    #[tokio::test]
868    async fn local_http_listener_serves_http1_when_h2c_disabled() {
869        let mut config = local_config();
870        config.enable_h2c = false;
871        let server = HttpServerBuilder::new(config, CancellationToken::new())
872            .with_router(Router::new().route("/http1", get(|| async { "ok" })))
873            .build()
874            .expect("build server");
875
876        server.start().await.expect("start http1 server");
877        let addr = server.local_addr().expect("local address");
878        let mut stream = tokio::net::TcpStream::connect(addr)
879            .await
880            .expect("connect to local server");
881        stream
882            .write_all(b"GET /http1 HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
883            .await
884            .expect("write request");
885        let mut response = String::new();
886        tokio::time::timeout(Duration::from_secs(2), stream.read_to_string(&mut response))
887            .await
888            .expect("response read timed out")
889            .expect("read response");
890
891        assert!(response.starts_with("HTTP/1.1 200 OK"), "{response}");
892        assert!(response.contains("ok"), "{response}");
893        server.stop().await.expect("stop http1 server");
894    }
895
896    #[tokio::test]
897    async fn health_routers_report_registry_and_liveness() {
898        let registry = Arc::new(Registry::new());
899        let health = health_router(registry);
900        let response = health
901            .oneshot(
902                Request::builder()
903                    .uri("/health")
904                    .body(Body::empty())
905                    .expect("health request"),
906            )
907            .await
908            .expect("health response");
909        assert_eq!(response.status(), StatusCode::OK);
910        let body = axum::body::to_bytes(response.into_body(), usize::MAX)
911            .await
912            .expect("health body");
913        assert_eq!(&body[..], b"[]");
914
915        let liveness = healthz_router();
916        let response = liveness
917            .oneshot(
918                Request::builder()
919                    .uri("/healthz")
920                    .body(Body::empty())
921                    .expect("healthz request"),
922            )
923            .await
924            .expect("healthz response");
925        assert_eq!(response.status(), StatusCode::OK);
926        let body = axum::body::to_bytes(response.into_body(), usize::MAX)
927            .await
928            .expect("healthz body");
929        let body = String::from_utf8_lossy(&body);
930        assert!(body.contains("\"status\":\"ok\""));
931        assert!(body.contains("\"version\""));
932    }
933
934    #[test]
935    fn tls_acceptor_rejects_missing_certificate_paths() {
936        let tls = TlsConfig::default();
937
938        let error = match build_tls_acceptor(&tls) {
939            Ok(_) => panic!("missing TLS files should be rejected"),
940            Err(error) => error,
941        };
942
943        assert_eq!(error.code(), ErrorCode::InvalidInput);
944        assert!(error.message().contains("cert_file"));
945    }
946
947    #[test]
948    fn tls_acceptor_rejects_missing_key_path_after_cert_path() {
949        let tls = TlsConfig {
950            cert_file: Some("missing-cert.pem".to_string()),
951            ..Default::default()
952        };
953
954        let error = match build_tls_acceptor(&tls) {
955            Ok(_) => panic!("missing key file should be rejected before reading files"),
956            Err(error) => error,
957        };
958
959        assert_eq!(error.code(), ErrorCode::InvalidInput);
960        assert!(error.message().contains("key_file"));
961    }
962
963    #[test]
964    fn tls_loader_reports_missing_certificate_and_key_files() {
965        let cert_error = load_certs("missing-cert.pem").unwrap_err();
966        assert_eq!(cert_error.code(), ErrorCode::InvalidInput);
967        assert!(
968            cert_error
969                .message()
970                .contains("failed to load HTTP TLS certificate")
971        );
972
973        let key_error = load_private_key("missing-key.pem").unwrap_err();
974        assert_eq!(key_error.code(), ErrorCode::InvalidInput);
975        assert!(key_error.message().contains("failed to load HTTP TLS key"));
976    }
977
978    #[test]
979    fn connection_accept_errors_are_classified_as_transient() {
980        for kind in [
981            std::io::ErrorKind::ConnectionRefused,
982            std::io::ErrorKind::ConnectionAborted,
983            std::io::ErrorKind::ConnectionReset,
984        ] {
985            let error = std::io::Error::from(kind);
986            assert!(is_connection_accept_error(&error));
987        }
988        assert!(!is_connection_accept_error(&std::io::Error::from(
989            std::io::ErrorKind::PermissionDenied
990        )));
991    }
992}
rskit_server/http.rs

rskit_server/
http.rs
