From 6d6b9481c5336a74ebbd12ed89f484463b218d56 Mon Sep 17 00:00:00 2001 From: forkwright Date: Fri, 7 Aug 2026 15:47:02 +0000 Subject: [PATCH 1/2] fix(akroasis-server): bound request timeout, concurrency, and body size The router attached only CORS and trace layers. A caller-supplied slow operation (e.g. an unresponsive serial device via /radio/detect?port=) occupied its worker indefinitely, and the number of simultaneously in-flight requests was unbounded. tower::limit::ConcurrencyLimitLayer applied via Router::layer does not actually bound anything -- axum's router clones and calls the matched route's service directly per request rather than propagating poll_ready backpressure through the layered stack, so the semaphore that layer relies on is never contended (caught by the concurrency regression test against a naive first attempt: 256 handlers ran simultaneously against a configured cap of 64). Concurrency is now gated with an explicit Semaphore-based middleware instead. Refs #194 --- Cargo.toml | 4 +- crates/akroasis-server/Cargo.toml | 6 ++ crates/akroasis-server/src/router.rs | 93 ++++++++++++++++++- crates/akroasis-server/tests/smoke.rs | 129 ++++++++++++++++++++++++++ 4 files changed, 228 insertions(+), 4 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 2d3cf57..a770b80 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -124,8 +124,8 @@ rusb = "0.9.4" # HTTP server axum = { version = "0.8", features = ["macros"] } -tower = "0.5" -tower-http = { version = "0.7", features = ["cors", "trace"] } +tower = { version = "0.5", features = ["util", "timeout"] } +tower-http = { version = "0.7", features = ["cors", "trace", "limit"] } # Testing proptest = "1" diff --git a/crates/akroasis-server/Cargo.toml b/crates/akroasis-server/Cargo.toml index 5772853..1cc653b 100644 --- a/crates/akroasis-server/Cargo.toml +++ b/crates/akroasis-server/Cargo.toml @@ -23,5 +23,11 @@ tower-http = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } +[dev-dependencies] +# WHY(#194): `start_paused = true` on the timeout-layer regression test needs +# tokio's virtual clock, which "full" deliberately excludes from production +# builds. +tokio = { workspace = true, features = ["test-util"] } + [lints] workspace = true diff --git a/crates/akroasis-server/src/router.rs b/crates/akroasis-server/src/router.rs index e1a48df..c6d9377 100644 --- a/crates/akroasis-server/src/router.rs +++ b/crates/akroasis-server/src/router.rs @@ -1,12 +1,102 @@ //! Axum router — assembles all API routes. +use std::sync::Arc; +use std::time::Duration; + use axum::Router; +use axum::error_handling::HandleErrorLayer; +use axum::extract::Request; +use axum::http::StatusCode; +use axum::middleware::{self, Next}; +use axum::response::Response; use axum::routing::get; +use tokio::sync::Semaphore; +use tower::ServiceBuilder; use tower_http::cors::CorsLayer; +use tower_http::limit::RequestBodyLimitLayer; use tower_http::trace::TraceLayer; +use crate::error::ApiError; use crate::{mesh, radio}; +/// Maximum accepted request body size, in bytes. +/// +/// WHY(#194): no current route reads a request body, but the limit is +/// applied once at the router level rather than per-route so any future +/// body-reading route inherits the bound automatically instead of relying on +/// each handler to remember it. +pub const MAX_BODY_BYTES: usize = 10 * 1024 * 1024; + +/// Maximum number of requests the router services concurrently. +/// +/// WHY(#194): the API is unauthenticated and reachable on the LAN. Without a +/// cap, an attacker can open unbounded simultaneous connections against a +/// slow handler (e.g. `/radio/detect?port=` probing an unresponsive serial +/// device) and exhaust server resources. +pub const MAX_CONCURRENT_REQUESTS: usize = 64; + +/// Per-request timeout applied to every route. +/// +/// WHY(#194): generous enough for USB radio detection, but bounds any +/// handler that would otherwise occupy its worker indefinitely. +pub const REQUEST_TIMEOUT: Duration = Duration::from_secs(30); + +/// Apply the request-hardening middleware common to every akroasis API router. +/// +/// Layered outermost to innermost: a request body-size limit (reject before +/// doing any other work), a per-request timeout, and a global concurrency cap +/// (gates admission to a route handler). +/// +/// Exposed separately from [`build`] so tests can exercise the middleware +/// stack against a synthetic handler without needing radio hardware. +pub fn harden(router: Router) -> Router { + // WHY(#194): `tower::limit::ConcurrencyLimitLayer` bounds nothing when + // applied via `Router::layer` — axum's router clones and calls the + // matched route's service directly per request rather than routing + // `poll_ready` backpressure through the layered stack, so the semaphore + // that layer relies on is never actually contended. An explicit + // `Semaphore`-gated `from_fn` middleware enforces the bound directly, + // independent of that propagation gap. + let concurrency = Arc::new(Semaphore::new(MAX_CONCURRENT_REQUESTS)); + + router + .layer(middleware::from_fn(move |request, next| { + let concurrency = Arc::clone(&concurrency); + limit_concurrency(concurrency, request, next) + })) + .layer( + ServiceBuilder::new() + .layer(HandleErrorLayer::new(handle_middleware_error)) + .timeout(REQUEST_TIMEOUT), + ) + .layer(RequestBodyLimitLayer::new(MAX_BODY_BYTES)) +} + +/// Hold a permit from `concurrency` for the duration of `next`, blocking +/// admission once [`MAX_CONCURRENT_REQUESTS`] handlers are already running. +async fn limit_concurrency(concurrency: Arc, request: Request, next: Next) -> Response { + #[expect( + clippy::expect_used, + reason = "the only closer is `Semaphore::close`, which `harden` never calls; the \ + `Arc` this closure holds keeps the semaphore itself alive for the router's lifetime" + )] + let _permit = concurrency + .acquire_owned() + .await + .expect("semaphore is never closed while `harden` holds the strong reference to it"); // SAFETY: harden() never calls Semaphore::close(); acquire_owned() can only fail after close() + next.run(request).await +} + +/// Convert a middleware failure into the same JSON envelope every other API +/// error uses. +async fn handle_middleware_error(err: tower::BoxError) -> ApiError { + if err.is::() { + ApiError::client(StatusCode::REQUEST_TIMEOUT, "request timed out") + } else { + ApiError::internal(err.to_string()) + } +} + /// Build the complete akroasis API router. /// /// Attach this to an axum `serve()` call or embed in a larger router. @@ -18,8 +108,7 @@ pub fn build() -> Router { .route("/mesh/nodes", get(mesh::nodes)) .route("/mesh/topology", get(mesh::topology)); - Router::new() - .nest("/api/v1", api) + harden(Router::new().nest("/api/v1", api)) .layer(CorsLayer::permissive()) .layer(TraceLayer::new_for_http()) } diff --git a/crates/akroasis-server/tests/smoke.rs b/crates/akroasis-server/tests/smoke.rs index d2b789d..67393d4 100644 --- a/crates/akroasis-server/tests/smoke.rs +++ b/crates/akroasis-server/tests/smoke.rs @@ -10,10 +10,16 @@ reason = "test code: panics and unwraps acceptable in assertions" )] +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; + use akroasis_server::error::ApiError; +use axum::Router; use axum::body::Body; use axum::http::{Request, StatusCode}; use axum::response::IntoResponse; +use axum::routing::{get, post}; use tower::ServiceExt as _; #[tokio::test] @@ -42,3 +48,126 @@ fn api_error_internal_reports_500_and_hides_detail() { let response = ApiError::internal("disk full at /var/lib/akroasis").into_response(); assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); } + +// ── request hardening (#194) ──────────────────────────────────────────────── +// +// These exercise `router::harden` directly against synthetic handlers rather +// than the production routes, since none of the real handlers offer a way to +// deterministically hang or accept an oversized body from a test. + +#[tokio::test(start_paused = true)] +async fn timeout_layer_bounds_a_handler_that_never_returns() { + let router = akroasis_server::router::harden(Router::new().route( + "/slow", + get(|| async { + // WHY: any duration well past REQUEST_TIMEOUT proves the + // layer -- not the handler -- ends the request. Paused tokio + // time makes this resolve without real wall-clock delay. + tokio::time::sleep(Duration::from_secs(3600)).await; + }), + )); + + let request = Request::builder() + .uri("/slow") + .body(Body::empty()) + .expect("request is well-formed"); + + let response = router + .oneshot(request) + .await + .expect("the timeout layer must convert an elapsed request into a response, not an error"); + assert_eq!(response.status(), StatusCode::REQUEST_TIMEOUT); +} + +#[tokio::test] +async fn body_limit_layer_rejects_oversized_bodies() { + let router = akroasis_server::router::harden(Router::new().route( + "/echo", + post(|body: axum::body::Bytes| async move { body.len().to_string() }), + )); + + let oversized = vec![0_u8; akroasis_server::router::MAX_BODY_BYTES + 1]; + let request = Request::builder() + .method("POST") + .uri("/echo") + .body(Body::from(oversized)) + .expect("request is well-formed"); + + let response = router + .oneshot(request) + .await + .expect("router must produce a response"); + assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE); +} + +#[tokio::test] +async fn body_limit_layer_accepts_bodies_within_the_cap() { + let router = akroasis_server::router::harden(Router::new().route( + "/echo", + post(|body: axum::body::Bytes| async move { body.len().to_string() }), + )); + + let within_cap = vec![0_u8; 1024]; + let request = Request::builder() + .method("POST") + .uri("/echo") + .body(Body::from(within_cap)) + .expect("request is well-formed"); + + let response = router + .oneshot(request) + .await + .expect("router must produce a response"); + assert_eq!(response.status(), StatusCode::OK); +} + +#[tokio::test] +async fn concurrency_limit_bounds_simultaneously_executing_handlers() { + let in_flight = Arc::new(AtomicUsize::new(0)); + let max_observed = Arc::new(AtomicUsize::new(0)); + let in_flight_handler = Arc::clone(&in_flight); + let max_observed_handler = Arc::clone(&max_observed); + + let router = akroasis_server::router::harden(Router::new().route( + "/probe", + get(move || { + let in_flight = Arc::clone(&in_flight_handler); + let max_observed = Arc::clone(&max_observed_handler); + async move { + let now = in_flight.fetch_add(1, Ordering::SeqCst) + 1; + max_observed.fetch_max(now, Ordering::SeqCst); + tokio::time::sleep(Duration::from_millis(30)).await; // kanon:ignore TESTING/sleep-in-test -- real concurrent scheduling must be observed; virtual/paused time collapses the overlap this test measures + in_flight.fetch_sub(1, Ordering::SeqCst); + } + }), + )); + + // WHY: fire well beyond MAX_CONCURRENT_REQUESTS so the cap -- not simple + // request volume -- is what this test proves. + let total_requests = akroasis_server::router::MAX_CONCURRENT_REQUESTS * 4; + let mut handles = Vec::with_capacity(total_requests); + for _ in 0..total_requests { + let router = router.clone(); + handles.push(tokio::spawn(async move { + let request = Request::builder() + .uri("/probe") + .body(Body::empty()) + .expect("request is well-formed"); + router + .oneshot(request) + .await + .expect("router must produce a response") + })); + } + for handle in handles { + let response = handle.await.expect("handler task must not panic"); + assert_eq!(response.status(), StatusCode::OK); + } + + assert!( + max_observed.load(Ordering::SeqCst) <= akroasis_server::router::MAX_CONCURRENT_REQUESTS, + "observed {} simultaneously executing handlers, more than the configured cap of {}", + max_observed.load(Ordering::SeqCst), + akroasis_server::router::MAX_CONCURRENT_REQUESTS, + ); +} From 4f0c7b4ad7c1d7972cd3cb9fd40094dfe739b027 Mon Sep 17 00:00:00 2001 From: forkwright Date: Fri, 7 Aug 2026 16:30:29 +0000 Subject: [PATCH 2/2] chore(akroasis-server): annotate the paused-clock sleep for kanon lint The timeout regression test's tokio::time::sleep runs under start_paused = true (virtual clock, no real wall-clock delay), which is exactly the deterministic time control TESTING/sleep-in-test wants -- annotate it so the lint stops flagging a sleep that is already non-blocking. Refs #194 --- Cargo.lock | 1 + crates/akroasis-server/tests/smoke.rs | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index 28150fa..7dc4dac 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2697,6 +2697,7 @@ dependencies = [ "bytes", "http", "http-body", + "http-body-util", "percent-encoding", "pin-project-lite", "tower-layer", diff --git a/crates/akroasis-server/tests/smoke.rs b/crates/akroasis-server/tests/smoke.rs index 67393d4..eaaab10 100644 --- a/crates/akroasis-server/tests/smoke.rs +++ b/crates/akroasis-server/tests/smoke.rs @@ -63,7 +63,7 @@ async fn timeout_layer_bounds_a_handler_that_never_returns() { // WHY: any duration well past REQUEST_TIMEOUT proves the // layer -- not the handler -- ends the request. Paused tokio // time makes this resolve without real wall-clock delay. - tokio::time::sleep(Duration::from_secs(3600)).await; + tokio::time::sleep(Duration::from_secs(3600)).await; // kanon:ignore TESTING/sleep-in-test -- runs under start_paused = true; tokio's virtual clock resolves this without real wall-clock delay, which is the deterministic time control this rule wants }), ));