diff --git a/Cargo.lock b/Cargo.lock index 2d71b13..3b28d18 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -49,7 +49,7 @@ dependencies = [ "nom", "num-traits", "rusticata-macros", - "thiserror 2.0.18", + "thiserror 2.0.19", "time", ] @@ -61,7 +61,7 @@ checksum = "3109e49b1e4909e9db6515a30c633684d68cdeaa252f215214cb4fa1a5bfee2c" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.119", "synstructure", ] @@ -73,20 +73,26 @@ checksum = "7b18050c2cd6fe86c3a76584ef5e0baf286d038cda203eb6223df2cc413565f7" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.119", ] [[package]] name = "async-trait" -version = "0.1.89" +version = "0.1.91" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" +checksum = "ae36dc4177970ef04fde5178d3e2429882def40e57a451f919c098f72baa6cec" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 3.0.2", ] +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + [[package]] name = "autocfg" version = "1.5.1" @@ -188,9 +194,9 @@ checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" [[package]] name = "cc" -version = "1.2.67" +version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e17dd265a7d0f31ef544e1b20e03add05d3b45b491b633b10d67145d2acc1a38" +checksum = "c89588d05638b5b4594a3348a2d6c20277e43a7f5c5202b05cc56888475a47b8" dependencies = [ "find-msvc-tools", "jobserver", @@ -401,7 +407,7 @@ checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.119", ] [[package]] @@ -481,7 +487,7 @@ checksum = "1ac70aa55017e108007fbaf5aa0f54b021c98f92ff8af59d42eda9da96e3dd4f" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.119", ] [[package]] @@ -544,9 +550,9 @@ dependencies = [ [[package]] name = "fastrand" -version = "2.4.1" +version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6" +checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" [[package]] name = "fiat-crypto" @@ -560,6 +566,12 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "fnv" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" + [[package]] name = "foldhash" version = "0.2.0" @@ -637,7 +649,7 @@ checksum = "2d6d3cde68c518367be28956066ddfef33813991b77a55005a69dae04bf3b10b" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.119", ] [[package]] @@ -716,6 +728,25 @@ dependencies = [ "polyval", ] +[[package]] +name = "h2" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155" +dependencies = [ + "atomic-waker", + "bytes", + "fnv", + "futures-core", + "futures-sink", + "http", + "indexmap", + "slab", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "h3" version = "0.0.8" @@ -818,6 +849,41 @@ dependencies = [ "itoa", ] +[[package]] +name = "http-body" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e9f41fd6a08e4d4ec69df65976da761afd5ad5e58a9d4acb46bd1c953a9e3ff2" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "hybrid-array" version = "0.4.13" @@ -828,6 +894,42 @@ dependencies = [ "typenum", ] +[[package]] +name = "hyper" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d22053281f852e11534f5198498373cbb59295120a20771d90f7ed1897490a72" +dependencies = [ + "atomic-waker", + "bytes", + "futures-channel", + "futures-core", + "h2", + "http", + "http-body", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", + "want", +] + +[[package]] +name = "hyper-util" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" +dependencies = [ + "bytes", + "http", + "http-body", + "hyper", + "pin-project-lite", + "tokio", +] + [[package]] name = "icu_collections" version = "2.2.0" @@ -968,7 +1070,7 @@ dependencies = [ "jni-sys", "log", "simd_cesu8", - "thiserror 2.0.18", + "thiserror 2.0.19", "walkdir", "windows-link", ] @@ -983,7 +1085,7 @@ dependencies = [ "quote", "rustc_version", "simd_cesu8", - "syn", + "syn 2.0.119", ] [[package]] @@ -1002,7 +1104,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "38c0b942f458fe50cdac086d2f946512305e5631e720728f2a61aabcd47a6264" dependencies = [ "quote", - "syn", + "syn 2.0.119", ] [[package]] @@ -1226,7 +1328,7 @@ version = "0.2.0" dependencies = [ "quinn", "rustls", - "thiserror 2.0.18", + "thiserror 2.0.19", "wtransport", ] @@ -1336,6 +1438,9 @@ dependencies = [ "h3-quinn", "h3-webtransport", "http", + "http-body-util", + "hyper", + "hyper-util", "mtp-codec", "mtp-common", "mtp-crypto", @@ -1345,8 +1450,10 @@ dependencies = [ "rand 0.10.2", "rcgen", "rustls", - "thiserror 2.0.18", + "thiserror 2.0.19", "tokio", + "tokio-rustls", + "tokio-stream", "tracing", ] @@ -1578,9 +1685,9 @@ dependencies = [ [[package]] name = "proc-macro2" -version = "1.0.106" +version = "1.0.107" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" dependencies = [ "unicode-ident", ] @@ -1600,7 +1707,7 @@ dependencies = [ "rustc-hash", "rustls", "socket2", - "thiserror 2.0.18", + "thiserror 2.0.19", "tokio", "tracing", "web-time", @@ -1625,7 +1732,7 @@ dependencies = [ "rustls-pki-types", "rustls-platform-verifier", "slab", - "thiserror 2.0.18", + "thiserror 2.0.19", "tinyvec", "tracing", "web-time", @@ -1647,9 +1754,9 @@ dependencies = [ [[package]] name = "quote" -version = "1.0.46" +version = "1.0.47" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" dependencies = [ "proc-macro2", ] @@ -1922,9 +2029,9 @@ checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" [[package]] name = "serde" -version = "1.0.228" +version = "1.0.229" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" dependencies = [ "serde_core", "serde_derive", @@ -1932,29 +2039,29 @@ dependencies = [ [[package]] name = "serde_core" -version = "1.0.228" +version = "1.0.229" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" dependencies = [ "serde_derive", ] [[package]] name = "serde_derive" -version = "1.0.228" +version = "1.0.229" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 3.0.2", ] [[package]] name = "serde_json" -version = "1.0.150" +version = "1.0.151" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" dependencies = [ "itoa", "memchr", @@ -2156,6 +2263,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a207d6d6a2b7fc470b80443726053f18a2481b7e1eee970597051596567987a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "synstructure" version = "0.13.2" @@ -2164,7 +2282,7 @@ checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.119", ] [[package]] @@ -2178,11 +2296,11 @@ dependencies = [ [[package]] name = "thiserror" -version = "2.0.18" +version = "2.0.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" +checksum = "09a43598840e33d5b0331f38c5e30d13bb11c11210a4b58f0d9b18a5a5eefcd9" dependencies = [ - "thiserror-impl 2.0.18", + "thiserror-impl 2.0.19", ] [[package]] @@ -2193,18 +2311,18 @@ checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.119", ] [[package]] name = "thiserror-impl" -version = "2.0.18" +version = "2.0.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" +checksum = "43cbfe0cf76104d42a574802844187e84a305e531ed54455f11fbde0f10541cd" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 3.0.2", ] [[package]] @@ -2218,9 +2336,9 @@ dependencies = [ [[package]] name = "time" -version = "0.3.53" +version = "0.3.54" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "18dfaaeddcb932337b5e7866ee7d0ce9b76d2fd092997146f187ec09b4558a50" +checksum = "3e1d5e639ff6bab73cb6885cc7e7b1de96c3f32c68ec55f3952614bec1092244" dependencies = [ "deranged", "num-conv", @@ -2238,9 +2356,9 @@ checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109" [[package]] name = "time-macros" -version = "0.2.31" +version = "0.2.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c431b87111666e491a90baa837f914fb45cd5dc3c268591b0220ff5057f2085f" +checksum = "7e689342a48d2ea927c87ea50cabf8594854bf940e9310208848d680d668ed85" dependencies = [ "num-conv", "time-core", @@ -2273,9 +2391,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.53.0" +version = "1.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d988bcd52dbe076d3d46903332f58c912b87a2c49b1428419a5845154762ffee" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" dependencies = [ "bytes", "libc", @@ -2296,7 +2414,28 @@ checksum = "6328af13490e73a9b4694030fafd93f8c8c6a9dede33e821c3fc63eddf8042ba" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.119", +] + +[[package]] +name = "tokio-rustls" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" +dependencies = [ + "rustls", + "tokio", +] + +[[package]] +name = "tokio-stream" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", ] [[package]] @@ -2332,7 +2471,7 @@ checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.119", ] [[package]] @@ -2355,6 +2494,12 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "try-lock" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" + [[package]] name = "typenum" version = "1.20.1" @@ -2429,6 +2574,15 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "want" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bfa7760aed19e106de2c7c0b581b509f2f25d3dacaf737cb82ac61bc6d760b0e" +dependencies = [ + "try-lock", +] + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" @@ -2477,7 +2631,7 @@ dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn", + "syn 2.0.119", "wasm-bindgen-shared", ] @@ -2520,7 +2674,7 @@ checksum = "94eb68555b95bcea5e8cf4abe280b529049479fa995bfc23734af96a6aedc120" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.119", ] [[package]] @@ -2677,7 +2831,7 @@ dependencies = [ "rustls-pki-types", "sha2 0.11.0", "socket2", - "thiserror 2.0.18", + "thiserror 2.0.19", "time", "tokio", "tracing", @@ -2694,7 +2848,7 @@ checksum = "d5867c629e4252f7439d82315923daaf27f4fa442410d51b78ab93ef4c432a11" dependencies = [ "httlib-huffman", "octets", - "thiserror 2.0.18", + "thiserror 2.0.19", "url", ] @@ -2725,7 +2879,7 @@ dependencies = [ "oid-registry", "ring", "rusticata-macros", - "thiserror 2.0.18", + "thiserror 2.0.19", "time", ] @@ -2758,28 +2912,28 @@ checksum = "de844c262c8848816172cef550288e7dc6c7b7814b4ee56b3e1553f275f1858e" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.119", "synstructure", ] [[package]] name = "zerocopy" -version = "0.8.54" +version = "0.8.55" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b7cbbc0a705a0fd05cc3676525980d2bf5a9bc4adac6d6475209a7887cf59d19" +checksum = "b5a105cd7b140f6eeec8acff2ea38135d3cab283ada58540f629fe51e46696eb" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.54" +version = "0.8.55" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e2e817b7b52d0c7358d3246da9d69935ebb18116b2b102b4230dac079b4862f5" +checksum = "0fe976fb70c78cd64cccfe3a6fc142244e8a77b70959b30faf9d0ac37ee228eb" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.119", ] [[package]] @@ -2799,7 +2953,7 @@ checksum = "11532158c46691caf0f2593ea8358fed6bbf68a0315e80aae9bd41fbade684a1" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.119", "synstructure", ] @@ -2820,7 +2974,7 @@ checksum = "3c50655cbb0fe3fc43170059e702f1ce5e19b84cec58dc87b037a09935c2f328" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.119", ] [[package]] @@ -2853,7 +3007,7 @@ checksum = "625dc425cab0dca6dc3c3319506e6593dcb08a9f387ea3b284dbd52a92c40555" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.119", ] [[package]] diff --git a/README.md b/README.md index f1b787b..a7634df 100644 --- a/README.md +++ b/README.md @@ -19,7 +19,7 @@ Area-specific guides: - [Troubleshooting](./docs/TROUBLESHOOTING.md) - [Operations](./docs/OPERATIONS.md) -`MTPWebServer` owns its UDP endpoint and must not bind to the same address and port as `MTPHost`. The two structs use incompatible QUIC ALPN protocols (`h3` for the web server, native MTP for the host), so a single socket cannot service both. The `host` and `webserver` Cargo features are not designed to be enabled simultaneously in the same binary; choose the one that matches your client type. +`MTPWebServer` owns TCP TLS (HTTP/1.1 and HTTP/2) plus UDP QUIC (HTTP/3 and WebTransport) on one numeric port. It must not bind its UDP address and port as `MTPHost`: their QUIC ALPN protocols remain incompatible (`h3` for the web server, native MTP for the host). ## Browser SDK @@ -50,7 +50,7 @@ Feature summary: | `crypto` | `mtp::crypto` | AEAD, signatures, KEM, KDF, hashing | | `host` | `mtp::host`, codec registry | QUIC host and version negotiation | | `client` | `mtp::client` | QUIC client connections | -| `webserver` | `mtp::webserver` | HTTP/3 server with WebTransport MTP sessions | +| `webserver` | `mtp::webserver` | HTTPS server with HTTP/1.1, HTTP/2, HTTP/3, and WebTransport MTP sessions | The core crates are always available: `codec`, `transport`, `common`, and `type_map`. See the [native client](./docs/NATIVE-CLIENT.md) and [native host](./docs/NATIVE-HOST.md) guides for configuration and usage. See [Security](./docs/SECURITY.md) for security boundaries. diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index dcbfa0f..5d60133 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -24,7 +24,7 @@ MTP separates wire encoding, QUIC transport, connection policy, protocol negotia ┌─────────────────┴─────────────────┐ │ │ MTPHost MTPWebServer - native QUIC HTTP/3 + WebTransport + native QUIC HTTPS + HTTP/3 + WebTransport │ │ └──────────────┬────────────────────┘ │ @@ -40,10 +40,10 @@ Both clients exchange the same MTP frames with a host. The middle row is shared protocol machinery. The type map determines numeric IDs, the codec serializes values, and transport framing places each serialized frame on a QUIC stream. This is why a type-map change must be compiled into both peers before the new message can be exchanged. -The bottom row shows the two server entry points. `MTPHost` is a native QUIC endpoint for native MTP clients. `MTPWebServer` is an HTTP/3 server that reuses `HostConfig` and provides the same `accept()`-based MTP session API, adding web routing and WebTransport support for browser clients. Because they rely on different QUIC ALPN protocols (native MTP vs. `h3`), they must bind to different IP/port pairs and should not be enabled as Cargo features in the same binary. Choose `MTPHost` when you only serve native clients; choose `MTPWebServer` when you need HTTP/3 routes or browser-based MTP clients. +The bottom row shows the two server entry points. `MTPHost` is a native QUIC endpoint for native MTP clients. `MTPWebServer` owns TCP HTTPS and UDP HTTP/3/WebTransport listeners on the same numeric port, reuses one `HostConfig` and router, and provides the same `accept()`-based MTP session API. Its QUIC listener still uses only the `h3` ALPN, so it cannot share its UDP address with the native MTP ALPN endpoint. Choose `MTPHost` for native clients and `MTPWebServer` for browser-facing HTTP and WebTransport. `mtp-crypto` is an optional cross-cutting layer used by authenticated native connections, WebTransport connections, and browser E2EE; TLS remains the transport security layer in both paths. -`mtp-host` performs version negotiation and native authentication before returning an `MTPConnection`. `mtp-webserver` routes HTTP/3 requests and WebTransport sessions through its endpoint. WebTransport MTP sessions support the same optional cryptographic authentication as native hosts when the `crypto` feature is enabled. +`mtp-host` performs version negotiation and native authentication before returning an `MTPConnection`. `mtp-webserver` routes HTTP/1.1, HTTP/2, and HTTP/3 requests through one route table and surfaces WebTransport sessions through `accept()`. WebTransport MTP sessions support the same optional cryptographic authentication as native hosts when the `crypto` feature is enabled. The [native client](NATIVE-CLIENT.md), [WASM client](WASM-CLIENT.md), [native host](NATIVE-HOST.md), and [web server](NATIVE-HOST-WEB-SERVER.md) guides cover the public APIs for each boundary. The web server guide should be read as the host API for browser-facing deployments; it accepts the same `HostConfig` and authentication callbacks as the native host. diff --git a/docs/CONNECTIONS.md b/docs/CONNECTIONS.md index a52b9b4..34eb24a 100644 --- a/docs/CONNECTIONS.md +++ b/docs/CONNECTIONS.md @@ -16,7 +16,7 @@ Native clients and hosts share the same connection shape after the opening hands `WebMTPConnection`, returned by `MTPWebServer::accept()`, exposes the same members as the native host connection plus `request_path`, which contains the HTTP/3 path used for the WebTransport extended CONNECT request. Server-side MTP connections expose `remote_addr`, the peer address observed by -QUIC. HTTP/3 route handlers receive the same address as `Http3Request::remote_addr`. +QUIC. HTTP route handlers receive the peer address as `HttpRequest::remote_addr`. It is transport metadata and should not be treated as an authenticated identity; behind a proxy, use the proxy's trusted forwarding mechanism separately. The host connection also exposes a version-scoped `codec` and, for an authenticated client, its `client_public_key`. The native client connection also exposes these methods: diff --git a/docs/NATIVE-HOST-WEB-SERVER.md b/docs/NATIVE-HOST-WEB-SERVER.md index 130e6f9..a1758c6 100644 --- a/docs/NATIVE-HOST-WEB-SERVER.md +++ b/docs/NATIVE-HOST-WEB-SERVER.md @@ -1,27 +1,28 @@ # MTP Web Server -`MTPWebServer` serves ordinary HTTP/3 routes and WebTransport MTP sessions through one QUIC endpoint. HTTP/3 requests are handled inside the server; -WebTransport sessions are returned by `accept()` for application messages. -`MTPWebServer` and `MTPHost` cannot bind the same IP and port. +`MTPWebServer` is a complete browser-facing HTTPS server. TCP TLS serves HTTP/1.1 and HTTP/2, while UDP QUIC serves HTTP/3 and WebTransport. Both listeners use the same certificate, router, IP address, and numeric port. Ordinary HTTP requests are handled inside the server; WebTransport MTP sessions are returned by `accept()` for application messages. -The repository's combined server example registers `/` on `MTPWebServer` and -returns `OK` while the process is running. The route is served over HTTP/3 at -`https://localhost:8080/` on the same QUIC endpoint as WebTransport MTP sessions. +`MTPWebServer` and the native `MTPHost` cannot bind the same IP and port. The TCP integration does not add the native MTP QUIC ALPN protocol to `MTPWebServer`. + +The repository's server example serves the compiled web client at `/`, exposes status at `/health`, and accepts WebTransport sessions at the same origin. No second TCP server is required. ## WebServerConfig | Builder | Default | Purpose | | --- | --- | --- | -| `route(path, handler)` | None | Register an exact-path HTTP/3 handler. | +| `route(path, handler)` | None | Register an exact-path HTTP handler. | | `route_method(method, path, handler)` | None | Register a method-specific handler. | | `route_pattern(pattern, handler)` | None | Register a route with `{name}` single-segment parameters. | | `route_pattern_method(method, pattern, handler)` | None | Register a method-specific parameterized route. | | `fallback(handler)` | None | Handle requests that match no route. | | `mtp_path(path)` | `/` | Path for WebTransport extended CONNECT. | -| `max_request_body(bytes)` | 4 MiB | Maximum buffered HTTP/3 request body. | -| `max_connections(count)` | 256 | Maximum concurrent HTTP/3 connections. | -| `request_timeout(duration)` | 30 seconds | HTTP/3 request handling timeout. | -| `drain_timeout(duration)` | 10 seconds | Shutdown drain period. | +| `serve_tcp_https(enabled)` | `true` | Enable the TCP TLS listener for HTTP/1.1 and HTTP/2. | +| `max_tcp_connections(count)` | 256 | Maximum concurrent TCP TLS connections. | +| `tls_handshake_timeout(duration)` | 10 seconds | Maximum TCP TLS handshake duration. | +| `max_request_body(bytes)` | 4 MiB | Maximum request body across all HTTP versions. | +| `max_connections(count)` | 256 | Maximum concurrent QUIC/HTTP/3 connections. | +| `request_timeout(duration)` | 30 seconds | Handler timeout across all HTTP versions. | +| `drain_timeout(duration)` | 5 seconds | Graceful shutdown period across both transports. | | `with_metrics(metrics)` | None | Receive connection, request, and error callbacks. | The route and fallback builders return `Result` because duplicate routes and duplicate fallback handlers are rejected. @@ -33,13 +34,13 @@ method-specific and more-specific routes take precedence. ```rust use http::{Method, StatusCode}; -use mtp::webserver::{Http3Request, Http3Response, RouteParams, WebServerConfig}; +use mtp::webserver::{HttpRequest, HttpResponse, RouteParams, WebServerConfig}; async fn profile( - _request: Http3Request, - response: Http3Response, + _request: HttpRequest, + response: HttpResponse, params: RouteParams, -) -> Http3Response { +) -> HttpResponse { let Some(userid) = params.get("userid") else { return response.status(StatusCode::BAD_REQUEST); }; @@ -64,25 +65,25 @@ UTF-8 decoded before being passed to the handler. Malformed encoded values do not match the route. Query strings remain available through `request.uri.query()` and are not part of route matching. -## HTTP/3 Requests and Responses +## HTTP Requests and Responses -`Http3Request` contains `method`, `uri`, `headers`, the connecting `remote_addr`, and an optional buffered `body` represented by `bytes::Bytes`. `Http3Response::status`, `header`, and `body` build a buffered response. `try_header` returns an error for invalid header names or values. `stream` takes a `tokio::sync::mpsc::Receiver` for incremental response chunks. +`HttpRequest` contains `method`, `uri`, `headers`, the connecting `remote_addr`, and an optional buffered `body` represented by `bytes::Bytes`. `HttpResponse::status`, `header`, and `body` build a buffered response. `try_header` returns an error for invalid header names or values. `stream` takes a `tokio::sync::mpsc::Receiver` for incremental response chunks. The deprecated `Http3Request` and `Http3Response` aliases remain available for source compatibility. ```rust use bytes::Bytes; use http::{Method, StatusCode}; use tokio::sync::mpsc; -use mtp::webserver::{Http3Request, Http3Response, WebServerConfig}; +use mtp::webserver::{HttpRequest, HttpResponse, WebServerConfig}; -async fn health(_request: Http3Request, response: Http3Response) -> Http3Response { +async fn health(_request: HttpRequest, response: HttpResponse) -> HttpResponse { response.status(StatusCode::OK).body("ok") } -async fn whoami(request: Http3Request, response: Http3Response) -> Http3Response { +async fn whoami(request: HttpRequest, response: HttpResponse) -> HttpResponse { response.body(format!("client: {}", request.remote_addr)) } -async fn stream_numbers(_request: Http3Request, response: Http3Response) -> Http3Response { +async fn stream_numbers(_request: HttpRequest, response: HttpResponse) -> HttpResponse { let (tx, rx) = mpsc::channel::(10); tokio::spawn(async move { for number in 0..10 { @@ -129,7 +130,15 @@ while let Some(connection) = server.accept().await? { ``` > `MTPWebServer::new` consumes a `HostConfig` (not an `MTPHost` instance). It creates its own QUIC endpoint and does not share a port with a running `MTPHost`. -`server.accept()` returns `Option` for each WebTransport session. HTTP/3 routes do not surface through `accept()` because the server dispatches them internally. `WebMTPConnection` retains the negotiated version, codec, request path, remote address, description, sender, and receiver used by native MTP connections. +`server.accept()` returns `Option` for each WebTransport session. Ordinary HTTP routes do not surface through `accept()` because the server dispatches them internally. `WebMTPConnection` retains the negotiated version, codec, request path, remote address, description, sender, and receiver used by native MTP connections. + +## Deployment + +For direct browser access, leave `serve_tcp_https(true)` enabled. The server advertises `h2` and `http/1.1` on TCP TLS and `h3` on UDP QUIC; WebTransport extended CONNECT is available only over HTTP/3. Both transports must present the certificate supplied by the same `HostConfig` and use the same origin port. + +When a reverse proxy or another process owns TCP, use `WebServerConfig::new().serve_tcp_https(false)`. This retains the UDP HTTP/3/WebTransport endpoint and its shared router without claiming the TCP port. + +With port `0` and TCP enabled, construction binds TCP first and binds UDP to the selected TCP port, so `local_addr()` reports the common address. With TCP disabled, Quinn selects the UDP port as before. `shutdown()` stops both accept loops, gracefully finishes active HTTP requests until `drain_timeout`, closes Quinn, and then aborts remaining work. `close()` and dropping the server stop both listeners immediately. ### Authentication @@ -166,4 +175,4 @@ fn request_completed(&self, path: &str, status: u16, duration: Duration) fn error_occurred(&self, error: &WebServerError) ``` -Errors include route misses, invalid requests, body-limit failures, handler timeouts, response construction failures, and transport failures. Supply the metrics object with `WebServerConfig::with_metrics`. +Errors include invalid requests, body-limit failures, handler timeouts, response write failures, TLS failures, and transport failures. Completion callbacks include the final HTTP status for every supported HTTP version. Supply the metrics object with `WebServerConfig::with_metrics`. diff --git a/docs/NATIVE-HOST.md b/docs/NATIVE-HOST.md index 0ae16ad..c71f3ac 100644 --- a/docs/NATIVE-HOST.md +++ b/docs/NATIVE-HOST.md @@ -2,7 +2,7 @@ The native host is a Rust library (`mtp-host`) that runs a QUIC server, accepts MTP client connections, negotiates protocol versions, and optionally performs a mutual-authentication handshake (login/register) using Ed25519 and ML-DSA-65 signatures. -> **Note:** `MTPHost` serves native MTP clients over raw QUIC. If you need to serve HTTP/3 routes on the same endpoint, use [`MTPWebServer`](NATIVE-HOST-WEB-SERVER.md) instead. `MTPWebServer` accepts the same `HostConfig` but binds an HTTP/3 endpoint rather than a native QUIC endpoint. +> **Note:** `MTPHost` serves native MTP clients over raw QUIC. For browser-facing HTTP/1.1, HTTP/2, HTTP/3, and WebTransport, use [`MTPWebServer`](NATIVE-HOST-WEB-SERVER.md). It accepts the same `HostConfig` but its UDP endpoint uses HTTP/3 rather than the native MTP QUIC ALPN. ## Cargo Dependency diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md index 4c29908..4ada886 100644 --- a/docs/OPERATIONS.md +++ b/docs/OPERATIONS.md @@ -15,7 +15,7 @@ Expose counters and gauges around the host and transport callbacks: | Ping round-trip time and missed pings | Peer reachability and path latency. | | Pipe accept, reject, EOF, and reset counts | Application admission and stream completion behavior. | -Implement `WebServerMetrics` for HTTP/3 request and error callbacks. Record the request path, status, duration, and `WebServerError` category without logging credentials, private keys, or message contents. Export host callback results through the application's metrics system for native deployments. +Implement `WebServerMetrics` for HTTP/1.1, HTTP/2, HTTP/3, TLS, and WebTransport callbacks. Record the request path, status, duration, and `WebServerError` category without logging credentials, private keys, or message contents. Export host callback results through the application's metrics system for native deployments. ## Tuning @@ -39,4 +39,4 @@ Back up host keyrings and client keyrings as protected secrets. Test restoring a ### Graceful Shutdown -Stop accepting new connections, reject new work at the application layer, and allow active requests and pipe writers to finish. Send a normal connection close, wait for the configured drain period, then force-close remaining QUIC sessions. For `MTPWebServer`, call `shutdown()` after the accept loop stops; Headits `drain_timeout` controls the drain period. +Stop accepting new connections, reject new work at the application layer, and allow active requests and pipe writers to finish. For `MTPWebServer`, call `shutdown()`; its `drain_timeout` controls graceful TCP HTTP completion and the QUIC drain period before remaining connection tasks are terminated. diff --git a/example/Cargo.lock b/example/Cargo.lock index 23258d9..c84365a 100644 --- a/example/Cargo.lock +++ b/example/Cargo.lock @@ -62,6 +62,12 @@ dependencies = [ "syn", ] +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + [[package]] name = "autocfg" version = "1.5.1" @@ -520,6 +526,12 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "fnv" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" + [[package]] name = "foldhash" version = "0.2.0" @@ -666,6 +678,25 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "h2" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155" +dependencies = [ + "atomic-waker", + "bytes", + "fnv", + "futures-core", + "futures-sink", + "http", + "indexmap", + "slab", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "h3" version = "0.0.8" @@ -768,6 +799,41 @@ dependencies = [ "itoa", ] +[[package]] +name = "http-body" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e9f41fd6a08e4d4ec69df65976da761afd5ad5e58a9d4acb46bd1c953a9e3ff2" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "hybrid-array" version = "0.4.13" @@ -778,6 +844,41 @@ dependencies = [ "typenum", ] +[[package]] +name = "hyper" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d22053281f852e11534f5198498373cbb59295120a20771d90f7ed1897490a72" +dependencies = [ + "atomic-waker", + "bytes", + "futures-channel", + "futures-core", + "h2", + "http", + "http-body", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", +] + +[[package]] +name = "hyper-util" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" +dependencies = [ + "bytes", + "http", + "http-body", + "hyper", + "pin-project-lite", + "tokio", +] + [[package]] name = "icu_collections" version = "2.2.0" @@ -1254,6 +1355,9 @@ dependencies = [ "h3-quinn", "h3-webtransport", "http", + "http-body-util", + "hyper", + "hyper-util", "mtp-codec", "mtp-common", "mtp-crypto", @@ -1264,6 +1368,8 @@ dependencies = [ "rustls", "thiserror 2.0.18", "tokio", + "tokio-rustls", + "tokio-stream", "tracing", ] @@ -1880,11 +1986,10 @@ version = "0.2.0" dependencies = [ "base64", "hex", + "http", "mtp", - "rustls", "serde_json", "tokio", - "tokio-rustls", "tracing-subscriber", ] @@ -2221,6 +2326,17 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-stream" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", +] + [[package]] name = "tokio-util" version = "0.7.18" diff --git a/example/server/Cargo.toml b/example/server/Cargo.toml index 890fa91..dcca160 100644 --- a/example/server/Cargo.toml +++ b/example/server/Cargo.toml @@ -10,8 +10,7 @@ path = "src/main.rs" [dependencies] mtp = { version = "0.2.0", path = "../../", features = ["crypto", "tls", "web-server", "files", "pipes"] } tokio = { version = "1", features = ["full"] } -tokio-rustls = "0.26" -rustls = "0.23" +http = "1" serde_json = { version = "1" } hex = "0.4" base64 = "0.22" diff --git a/example/server/src/main.rs b/example/server/src/main.rs index fc06960..13f87a7 100644 --- a/example/server/src/main.rs +++ b/example/server/src/main.rs @@ -133,17 +133,10 @@ async fn main() -> Result<(), Box> { Box::new(complete_register), ); - let _https = web_server::spawn_https( - std::net::SocketAddr::new(config.ip, config.port), - &config.tls_fullchain, - &config.tls_key, - ) - .await?; let mut host = mtp::webserver::MTPWebServer::new(config, web_server::config()?).await?; - println!( - "Server listening on https://{} (HTTPS + UDP WebTransport)", - host.local_addr() - ); + println!("Server listening on https://{}", host.local_addr()); + println!("TCP: HTTP/1.1 and HTTP/2"); + println!("UDP: HTTP/3 and WebTransport"); while let Some(conn) = host.accept().await? { let decrypt_keyring = Arc::clone(&decrypt_keyring); diff --git a/example/server/src/web-server.rs b/example/server/src/web-server.rs index 9736909..5d4cf1c 100644 --- a/example/server/src/web-server.rs +++ b/example/server/src/web-server.rs @@ -1,98 +1,47 @@ -use mtp::webserver::{Http3Request, Http3Response, RouteParams, WebServerConfig}; -use rustls::pki_types::{PrivateKeyDer, pem::PemObject}; +use mtp::webserver::{HttpRequest, HttpResponse, RouteParams, WebServerConfig}; use std::{ - io, - net::SocketAddr, - path::{Path, PathBuf}, + path::{Component, Path, PathBuf}, sync::Arc, }; -use tokio::{ - io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}, - net::TcpListener, - task::JoinHandle, -}; -use tokio_rustls::TlsAcceptor; -async fn ok(request: Http3Request, response: Http3Response) -> Http3Response { +async fn health(request: HttpRequest, response: HttpResponse) -> HttpResponse { response .header("content-type", "text/plain; charset=utf-8") .body(format!("OK\nclient: {}\n", request.remote_addr)) } async fn profile( - request: Http3Request, - response: Http3Response, + request: HttpRequest, + response: HttpResponse, params: RouteParams, -) -> Http3Response { +) -> HttpResponse { let Some(user) = params.get("user") else { return response.body("missing user"); }; - let body = serde_json::json!({ "user": user, "remote_addr": request.remote_addr.to_string(), - "profile": { - "display_name": format!("Example user {user}"), - "status": "active" - } + "profile": { "display_name": format!("Example user {user}"), "status": "active" } }); - response .header("content-type", "application/json; charset=utf-8") .body(body.to_string()) } pub fn config() -> Result { - WebServerConfig::new() - .route("/", ok)? - .route_pattern("/api/get/{user}/profile", profile) -} - -/// Starts the conventional HTTPS side of the example host. WebTransport uses -/// UDP/QUIC on the same port; browsers still need TCP/TLS to navigate to a URL. -pub async fn spawn_https( - address: SocketAddr, - certificate_pem: &[u8], - key_pem: &[u8], -) -> io::Result> { - // The TCP listener is created before the QUIC endpoint, so it must select - // rustls' process-wide provider itself. - mtp::crypto::ensure_crypto_provider(); - let certificates = rustls::pki_types::CertificateDer::pem_slice_iter(certificate_pem) - .collect::, _>>() - .map_err(io::Error::other)?; - let key = PrivateKeyDer::from_pem_slice(key_pem).map_err(io::Error::other)?; - let tls = rustls::ServerConfig::builder() - .with_no_client_auth() - .with_single_cert(certificates, key) - .map_err(io::Error::other)?; - let listener = TcpListener::bind(address).await?; - let acceptor = TlsAcceptor::from(Arc::new(tls)); - let asset_root = web_client_dist(); - match &asset_root { - Some(_) => println!( - "HTTPS web client available at https://localhost:{}", - address.port() - ), - None => eprintln!( - "Web client build not found; HTTPS will show setup instructions. Run `pnpm --dir example/web-client build`." - ), + let root = Arc::new(web_client_dist()); + if root.is_none() { + eprintln!( + "Web client build not found; requests will show setup instructions. Run `pnpm --dir example/web-client build`." + ); } - Ok(tokio::spawn(async move { - loop { - let Ok((stream, _)) = listener.accept().await else { - break; - }; - let acceptor = acceptor.clone(); - let asset_root = asset_root.clone(); - tokio::spawn(async move { - let Ok(mut stream) = acceptor.accept(stream).await else { - return; - }; - let _ = serve_https_request(&mut stream, &asset_root).await; - }); - } - })) + WebServerConfig::new() + .route("/health", health)? + .route_pattern("/api/get/{user}/profile", profile)? + .fallback(move |request, response| { + let root = Arc::clone(&root); + async move { static_assets(request, response, root).await } + }) } fn web_client_dist() -> Option { @@ -104,60 +53,78 @@ fn web_client_dist() -> Option { .find(|path| path.join("index.html").is_file()) } -async fn serve_https_request(stream: &mut S, asset_root: &Option) -> io::Result<()> -where - S: AsyncRead + AsyncWrite + Unpin, -{ - let mut request = [0; 16 * 1024]; - let size = stream.read(&mut request).await?; - let request = std::str::from_utf8(&request[..size]).unwrap_or_default(); - let path = request - .lines() - .next() - .and_then(|line| line.split_whitespace().nth(1)) - .unwrap_or("/") - .split('?') - .next() - .unwrap_or("/"); - let (status, content_type, body) = match asset_root { - Some(asset_root) => { - let relative = path.trim_start_matches('/'); - let candidate = asset_root.join(relative); - let file = if relative.is_empty() || !candidate.is_file() || relative.contains("..") { - asset_root.join("index.html") +async fn static_assets( + request: HttpRequest, + response: HttpResponse, + root: Arc>, +) -> HttpResponse { + if request.method != http::Method::GET && request.method != http::Method::HEAD { + return response.status(http::StatusCode::METHOD_NOT_ALLOWED); + } + let Some(root) = root.as_ref() else { + return response + .status(http::StatusCode::SERVICE_UNAVAILABLE) + .header("content-type", "text/html; charset=utf-8") + .body("MTP web client not built

Run pnpm --dir example/web-client build.

"); + }; + let relative = request.uri.path().trim_start_matches('/'); + let path = Path::new(relative); + if relative.contains('\\') + || path.components().any(|part| { + matches!( + part, + Component::ParentDir | Component::RootDir | Component::Prefix(_) + ) + }) + { + return response + .status(http::StatusCode::BAD_REQUEST) + .body("Invalid path"); + } + let requested = if relative.is_empty() { + root.join("index.html") + } else { + root.join(path) + }; + let file = if requested.is_file() { + requested + } else if path.extension().is_none() { + root.join("index.html") + } else { + return response + .status(http::StatusCode::NOT_FOUND) + .body("Not found"); + }; + match tokio::fs::read(&file).await { + Ok(body) => { + let response = response.header("content-type", content_type(&file)); + if request.method == http::Method::HEAD { + response.header("content-length", &body.len().to_string()) } else { - candidate - }; - let content_type = content_type(&file); - match tokio::fs::read(&file).await { - Ok(body) => ("200 OK", content_type, body), - Err(_) => ("404 Not Found", "text/plain; charset=utf-8", b"Not found".to_vec()), + response.body(body) } } - None => ( - "503 Service Unavailable", - "text/html; charset=utf-8", - b"MTP web client not built

Run pnpm --dir example/web-client build.

".to_vec(), - ), - }; - - let response = format!( - "HTTP/1.1 {status}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", - body.len() - ); - stream.write_all(response.as_bytes()).await?; - stream.write_all(&body).await?; - stream.shutdown().await + Err(_) => response + .status(http::StatusCode::NOT_FOUND) + .body("Not found"), + } } fn content_type(file: &Path) -> &'static str { match file.extension().and_then(|extension| extension.to_str()) { Some("html") => "text/html; charset=utf-8", - Some("js") => "text/javascript; charset=utf-8", + Some("js" | "mjs") => "text/javascript; charset=utf-8", Some("css") => "text/css; charset=utf-8", Some("wasm") => "application/wasm", Some("svg") => "image/svg+xml", - Some("json") => "application/json", + Some("json" | "map") => "application/json", + Some("png") => "image/png", + Some("jpg" | "jpeg") => "image/jpeg", + Some("gif") => "image/gif", + Some("webp") => "image/webp", + Some("ico") => "image/x-icon", + Some("woff") => "font/woff", + Some("woff2") => "font/woff2", _ => "application/octet-stream", } } diff --git a/mtp-webserver/Cargo.toml b/mtp-webserver/Cargo.toml index d7d9777..cbbe72f 100644 --- a/mtp-webserver/Cargo.toml +++ b/mtp-webserver/Cargo.toml @@ -12,6 +12,11 @@ mtp-crypto = { version = "0.2.0", path = "../crypto" } bytes = "1" http = "1" tokio = { version = "1", features = ["io-util", "macros", "net", "rt", "sync", "time"] } +hyper = { version = "1", features = ["server", "http1", "http2"] } +hyper-util = { version = "0.1", features = ["server", "http1", "http2", "tokio"] } +http-body-util = "0.1" +tokio-rustls = "0.26" +tokio-stream = "0.1" h3 = "0.0.8" h3-quinn = { version = "0.0.10", features = ["datagram"] } h3-webtransport = "0.1.2" @@ -24,6 +29,7 @@ rand = { version = "0.10.1", optional = true } [dev-dependencies] rcgen = "0.14" +hyper = { version = "1", features = ["client", "http2"] } [features] default = [] diff --git a/mtp-webserver/src/error.rs b/mtp-webserver/src/error.rs index 72833fa..0b02479 100644 --- a/mtp-webserver/src/error.rs +++ b/mtp-webserver/src/error.rs @@ -6,7 +6,9 @@ use std::fmt; pub enum WebServerError { Transport(CommunicationError), WebTransport(String), + Tls(String), Http(String), + PayloadTooLarge, NotFound(String), } @@ -15,7 +17,9 @@ impl fmt::Display for WebServerError { match self { Self::Transport(e) => write!(f, "transport error: {e}"), Self::WebTransport(msg) => write!(f, "webtransport error: {msg}"), + Self::Tls(msg) => write!(f, "TLS error: {msg}"), Self::Http(msg) => write!(f, "HTTP error: {msg}"), + Self::PayloadTooLarge => write!(f, "HTTP request body is too large"), Self::NotFound(route) => write!(f, "route not found: {route}"), } } diff --git a/mtp-webserver/src/h3.rs b/mtp-webserver/src/h3.rs new file mode 100644 index 0000000..6fce209 --- /dev/null +++ b/mtp-webserver/src/h3.rs @@ -0,0 +1,341 @@ +use crate::{ + HttpRequest, HttpResponse, Router, WebMTPConnection, WebServerError, WebServerMetrics, + transport::accept_web_connection, +}; +use bytes::{Buf, Bytes}; +use http::{Request, Response, StatusCode}; +use mtp_host::HostConfig; +use std::{net::SocketAddr, sync::Arc, time::Duration}; +use tokio::sync::{Semaphore, watch}; + +pub(crate) struct DriverConfig { + pub(crate) router: Router, + pub(crate) mtp_path: String, + pub(crate) max_request_body: usize, + pub(crate) request_timeout: Duration, + pub(crate) drain_timeout: Duration, + pub(crate) send_pongs: bool, + pub(crate) policy: mtp_transport::Policy, + pub(crate) host_config: Arc, + pub(crate) metrics: Option>, +} + +pub(crate) async fn run_driver( + endpoint: quinn::Endpoint, + config: DriverConfig, + mtp_tx: tokio::sync::mpsc::Sender>, + connection_semaphore: Arc, + mut shutdown_rx: watch::Receiver<()>, +) { + let DriverConfig { + router, + mtp_path, + max_request_body, + request_timeout, + drain_timeout, + send_pongs, + policy, + host_config, + metrics, + } = config; + let mut connection_tasks = tokio::task::JoinSet::new(); + loop { + tokio::select! { + biased; + _ = shutdown_rx.changed() => { + break; + } + incoming = endpoint.accept() => { + let Some(incoming) = incoming else { + break; + }; + // Do not await capacity here: doing so would prevent this loop + // from observing shutdown while all connection slots are in use. + let permit = match connection_semaphore.clone().try_acquire_owned() { + Ok(permit) => permit, + Err(_) => { + tracing::debug!("rejecting QUIC connection at configured connection limit"); + continue; + } + }; + let router = router.clone(); + let mtp_path = mtp_path.clone(); + let mtp_tx = mtp_tx.clone(); + let metrics = metrics.clone(); + let host_config = host_config.clone(); + connection_tasks.spawn(async move { + let _permit = permit; + let connect_start = std::time::Instant::now(); + let connection = match incoming.await { + Ok(connection) => connection, + Err(error) => { + tracing::debug!(%error, "QUIC connection failed during handshake"); + return; + } + }; + let mut builder = h3::server::builder(); + builder.enable_extended_connect(true); + builder.enable_webtransport(true); + builder.enable_datagram(true); + builder.max_webtransport_sessions(16); + let mut h3 = match builder + .build(h3_quinn::Connection::new(connection.clone())) + .await + { + Ok(connection) => connection, + Err(error) => { + tracing::debug!(%error, "HTTP/3 connection setup failed"); + return; + } + }; + + if let Some(ref m) = metrics { + m.connection_accepted(); + } + let remote_addr = connection.remote_address(); + + let mut tasks = tokio::task::JoinSet::new(); + loop { + let resolver = match h3.accept().await { + Ok(Some(resolver)) => resolver, + Ok(None) => break, + Err(error) => { + tracing::debug!(%error, "HTTP/3 request accept failed"); + break; + } + }; + let (request, mut stream) = match resolver.resolve_request().await { + Ok(request) => request, + Err(error) => { + tracing::debug!(%error, "HTTP/3 request parse failed"); + continue; + } + }; + if request.method() == http::Method::CONNECT && request.uri().path() == mtp_path { + if request.extensions().get::() + != Some(&h3::ext::Protocol::WEB_TRANSPORT) + { + let _ = stream + .send_response( + Response::builder() + .status(StatusCode::METHOD_NOT_ALLOWED) + .body(()) + .unwrap(), + ) + .await; + let _ = stream.finish().await; + continue; + } + let session = match h3_webtransport::server::WebTransportSession::accept( + request, stream, h3, + ) + .await + { + Ok(session) => Arc::new(session), + Err(error) => { + tracing::debug!(%error, "WebTransport session accept failed"); + return; + } + }; + // The WebTransport session request driver must outlive this + // endpoint request task. Keep it detached so handing the MTP + // connection to the application does not wait for the session + // (which is intentionally an open-ended accept loop). + tokio::spawn(run_session_requests( + session.clone(), + router.clone(), + max_request_body, + request_timeout, + metrics.clone(), + remote_addr, + )); + let result = + accept_web_connection(session, mtp_path, connection, send_pongs, policy, host_config.clone()) + .await; + match mtp_tx.try_send(result) { + Ok(()) => { + // The detached session driver remains active while the + // delivered MTP connection keeps the session alive. + } + Err(tokio::sync::mpsc::error::TrySendError::Full(result)) => { + tracing::warn!("MTP connection backlog is full; dropping connection"); + if let Ok(connection) = result { + connection.sender.close(); + } + } + Err(tokio::sync::mpsc::error::TrySendError::Closed(result)) => { + if let Ok(connection) = result { + connection.sender.close(); + } + } + } + return; + } + let router = router.clone(); + let metrics = metrics.clone(); + tasks.spawn(async move { + let path = request.uri().path().to_string(); + let response = crate::http::run_request( + &path, + request_timeout, + metrics.as_ref(), + handle_http_request(request, &mut stream, &router, max_request_body, remote_addr), + ).await; + if let Err(error) = write_response(&mut stream, response).await + && let Some(metrics) = &metrics + { + metrics.error_occurred(&WebServerError::Http(format!("HTTP/3 response write failed: {error}"))); + } + }); + } + tasks.join_all().await; + if let Some(ref m) = metrics { + m.connection_closed(connect_start.elapsed(), "normal"); + } + }); + } + } + } + + // --- Drain phase: wait for in-flight connections --- + + endpoint.close(quinn::VarInt::from_u32(0), b"mtp-webserver shutdown"); + + let drain_start = std::time::Instant::now(); + while !connection_tasks.is_empty() { + tokio::select! { + Some(result) = connection_tasks.join_next() => { + if let Err(e) = result { + tracing::warn!("Connection task panicked: {}", e); + } + } + _ = tokio::time::sleep(drain_timeout.saturating_sub(drain_start.elapsed())) => { + tracing::warn!( + "Drain timeout expired with {} connections still in flight", + connection_tasks.len() + ); + break; + } + } + } + + connection_tasks.shutdown().await; +} + +async fn handle_http_request( + request: Request<()>, + stream: &mut h3::server::RequestStream, + router: &Router, + max_request_body: usize, + remote_addr: SocketAddr, +) -> Result +where + S: h3::quic::BidiStream, +{ + let (request, too_large) = read_request(request, stream, max_request_body, remote_addr) + .await + .map_err(|e| WebServerError::Http(format!("request body read failed: {e}")))?; + if too_large { + return Err(WebServerError::PayloadTooLarge); + } + Ok(crate::http::dispatch_request(request, router).await) +} + +async fn run_session_requests( + session: Arc>, + router: Router, + max_request_body: usize, + request_timeout: Duration, + metrics: Option>, + remote_addr: SocketAddr, +) { + loop { + match session.accept_bi().await { + Ok(Some(h3_webtransport::server::AcceptedBi::Request(request, mut stream))) => { + let router = router.clone(); + let metrics = metrics.clone(); + tokio::spawn(async move { + let path = request.uri().path().to_string(); + let response = crate::http::run_request( + &path, + request_timeout, + metrics.as_ref(), + handle_http_request( + request, + &mut stream, + &router, + max_request_body, + remote_addr, + ), + ) + .await; + if let Err(error) = write_response(&mut stream, response).await + && let Some(metrics) = &metrics + { + metrics.error_occurred(&WebServerError::Http(format!( + "HTTP/3 response write failed: {error}" + ))); + } + }); + } + Ok(Some(h3_webtransport::server::AcceptedBi::BidiStream(_, _))) => {} + Ok(None) | Err(_) => break, + } + } +} + +async fn read_request( + request: Request<()>, + stream: &mut h3::server::RequestStream, + max_body: usize, + remote_addr: SocketAddr, +) -> Result<(HttpRequest, bool), h3::error::StreamError> +where + S: h3::quic::BidiStream, +{ + let (parts, _) = request.into_parts(); + let mut body = Vec::new(); + let mut too_large = false; + while let Some(chunk) = stream.recv_data().await? { + if body.len().saturating_add(chunk.remaining()) > max_body { + too_large = true; + break; + } + body.extend_from_slice(chunk.chunk()); + } + Ok(( + HttpRequest { + method: parts.method, + uri: parts.uri, + headers: parts.headers, + body: (!body.is_empty()).then(|| Bytes::from(body)), + remote_addr, + }, + too_large, + )) +} + +async fn write_response( + stream: &mut h3::server::RequestStream, + response: HttpResponse, +) -> Result<(), h3::error::StreamError> +where + S: h3::quic::BidiStream, +{ + let mut builder = Response::builder().status(response.status); + for (name, value) in &response.headers { + builder = builder.header(name, value); + } + stream + .send_response(builder.body(()).expect("valid HTTP response")) + .await?; + for chunk in response.body { + stream.send_data(chunk).await?; + } + if let Some(mut chunks) = response.stream { + while let Some(chunk) = chunks.recv().await { + stream.send_data(chunk).await?; + } + } + stream.finish().await +} diff --git a/mtp-webserver/src/http.rs b/mtp-webserver/src/http.rs new file mode 100644 index 0000000..f7c5a64 --- /dev/null +++ b/mtp-webserver/src/http.rs @@ -0,0 +1,100 @@ +use crate::{HttpRequest, HttpResponse, Router, WebServerError, WebServerMetrics}; +use http::StatusCode; +use std::{future::Future, sync::Arc, time::Duration}; + +pub(crate) async fn dispatch_request(request: HttpRequest, router: &Router) -> HttpResponse { + let path = request.uri.path(); + if let Some(handler) = router.handler(&request.method, path) { + return handler(request, HttpResponse::default()).await; + } + if let Some((handler, params)) = router.pattern_handler(&request.method, path) { + return handler(request, HttpResponse::default(), params).await; + } + if let Some(handler) = router.fallback_handler() { + return handler(request, HttpResponse::default()).await; + } + HttpResponse::new(StatusCode::NOT_FOUND) +} + +pub(crate) async fn run_request( + path: &str, + timeout: Duration, + metrics: Option<&Arc>, + future: F, +) -> HttpResponse +where + F: Future>, +{ + if let Some(metrics) = metrics { + metrics.request_started(path); + } + let started = std::time::Instant::now(); + let response = match tokio::time::timeout(timeout, future).await { + Ok(Ok(response)) => response, + Ok(Err(error)) => { + let status = if matches!(error, WebServerError::PayloadTooLarge) { + StatusCode::PAYLOAD_TOO_LARGE + } else { + StatusCode::BAD_REQUEST + }; + if let Some(metrics) = metrics { + metrics.error_occurred(&error); + } + HttpResponse::new(status) + } + Err(_) => { + let error = WebServerError::Http("request handler timed out".into()); + if let Some(metrics) = metrics { + metrics.error_occurred(&error); + } + HttpResponse::new(StatusCode::REQUEST_TIMEOUT) + } + }; + if let Some(metrics) = metrics { + metrics.request_completed(path, response.status.as_u16(), started.elapsed()); + } + response +} + +#[cfg(test)] +mod tests { + use super::*; + use http::{Method, Uri}; + + fn request(method: Method, uri: &'static str) -> HttpRequest { + HttpRequest { + method, + uri: Uri::from_static(uri), + headers: Default::default(), + body: None, + remote_addr: "127.0.0.1:1".parse().unwrap(), + } + } + + #[tokio::test] + async fn shared_dispatch_preserves_precedence_and_ignores_query() { + let router = Router::new() + .route("/items", |_, response| async move { + response.status(StatusCode::ACCEPTED) + }) + .unwrap() + .route_pattern("/{name}", |_, response, _| async move { + response.status(StatusCode::CREATED) + }) + .unwrap() + .fallback(|_, response| async move { response.status(StatusCode::IM_A_TEAPOT) }) + .unwrap(); + assert_eq!( + dispatch_request(request(Method::GET, "/items?q=1"), &router) + .await + .status, + StatusCode::ACCEPTED + ); + assert_eq!( + dispatch_request(request(Method::GET, "/other"), &router) + .await + .status, + StatusCode::CREATED + ); + } +} diff --git a/mtp-webserver/src/lib.rs b/mtp-webserver/src/lib.rs index a4b65ba..b7c044d 100644 --- a/mtp-webserver/src/lib.rs +++ b/mtp-webserver/src/lib.rs @@ -1,13 +1,16 @@ -//! HTTP/3 routing primitives and the combined MTP web-server API. +//! HTTP routing primitives and the combined MTP web-server API. //! //! The public routing API is transport-independent. The HTTP/3 driver is //! intentionally kept behind the crate's implementation boundary so callers //! do not need to depend on a particular QUIC implementation. mod error; +mod h3; +mod http; mod router; mod server; mod stream; +mod tcp; mod transport; pub use error::WebServerError; @@ -15,7 +18,8 @@ pub use error::WebServerError; pub use mtp_transport::TransportEvent; pub use router::{DynamicHttpHandler, HttpHandler, RouteParams, Router, RouterError}; pub use server::{MTPWebServer, WebServerConfig, WebServerMetrics}; -pub use stream::{Http3Request, Http3Response}; +#[allow(deprecated)] +pub use stream::{Http3Request, Http3Response, HttpRequest, HttpResponse}; pub use transport::{ H3TransportConnection, H3TransportReceiver, H3TransportSender, WebMTPConnection, WebMtpReceiver, WebMtpSender, diff --git a/mtp-webserver/src/router.rs b/mtp-webserver/src/router.rs index 540e554..9c1cc4a 100644 --- a/mtp-webserver/src/router.rs +++ b/mtp-webserver/src/router.rs @@ -1,24 +1,24 @@ -use crate::{Http3Request, Http3Response}; +use crate::{HttpRequest, HttpResponse}; use http::Method; use std::{collections::HashMap, future::Future, pin::Pin, sync::Arc}; /// Values captured from a parameterized route. pub type RouteParams = HashMap; -/// An asynchronous HTTP/3 route handler. +/// An asynchronous HTTP route handler. pub type HttpHandler = Arc< - dyn Fn(Http3Request, Http3Response) -> Pin + Send>> + dyn Fn(HttpRequest, HttpResponse) -> Pin + Send>> + Send + Sync, >; -/// An asynchronous handler for a parameterized HTTP/3 route. +/// An asynchronous handler for a parameterized HTTP route. pub type DynamicHttpHandler = Arc< dyn Fn( - Http3Request, - Http3Response, + HttpRequest, + HttpResponse, RouteParams, - ) -> Pin + Send>> + ) -> Pin + Send>> + Send + Sync, >; @@ -66,8 +66,8 @@ impl Router { pub fn route(self, path: impl Into, handler: F) -> Result where - F: Fn(Http3Request, Http3Response) -> Fut + Send + Sync + 'static, - Fut: Future + Send + 'static, + F: Fn(HttpRequest, HttpResponse) -> Fut + Send + Sync + 'static, + Fut: Future + Send + 'static, { self.route_inner( None, @@ -83,8 +83,8 @@ impl Router { handler: F, ) -> Result where - F: Fn(Http3Request, Http3Response) -> Fut + Send + Sync + 'static, - Fut: Future + Send + 'static, + F: Fn(HttpRequest, HttpResponse) -> Fut + Send + Sync + 'static, + Fut: Future + Send + 'static, { self.route_inner( Some(method), @@ -101,8 +101,8 @@ impl Router { handler: F, ) -> Result where - F: Fn(Http3Request, Http3Response, RouteParams) -> Fut + Send + Sync + 'static, - Fut: Future + Send + 'static, + F: Fn(HttpRequest, HttpResponse, RouteParams) -> Fut + Send + Sync + 'static, + Fut: Future + Send + 'static, { self.route_pattern_inner( None, @@ -120,8 +120,8 @@ impl Router { handler: F, ) -> Result where - F: Fn(Http3Request, Http3Response, RouteParams) -> Fut + Send + Sync + 'static, - Fut: Future + Send + 'static, + F: Fn(HttpRequest, HttpResponse, RouteParams) -> Fut + Send + Sync + 'static, + Fut: Future + Send + 'static, { self.route_pattern_inner( Some(method), @@ -132,8 +132,8 @@ impl Router { pub fn fallback(mut self, handler: F) -> Result where - F: Fn(Http3Request, Http3Response) -> Fut + Send + Sync + 'static, - Fut: Future + Send + 'static, + F: Fn(HttpRequest, HttpResponse) -> Fut + Send + Sync + 'static, + Fut: Future + Send + 'static, { if self.fallback.is_some() { return Err(RouterError::DuplicateFallback); @@ -307,7 +307,7 @@ mod tests { response.status(StatusCode::NO_CONTENT) }) .unwrap(); - let request = Http3Request { + let request = HttpRequest { method: Method::GET, uri: Uri::from_static("/health"), headers: Default::default(), @@ -315,7 +315,7 @@ mod tests { remote_addr: "127.0.0.1:4433".parse().unwrap(), }; let response = - router.handler(&Method::GET, "/health").unwrap()(request, Http3Response::default()) + router.handler(&Method::GET, "/health").unwrap()(request, HttpResponse::default()) .await; assert_eq!(response.status, StatusCode::NO_CONTENT); assert!(router.handler(&Method::GET, "/missing").is_none()); @@ -335,14 +335,14 @@ mod tests { let (handler, params) = router .pattern_handler(&Method::GET, "/api/get/user%2D123/profile.json") .unwrap(); - let request = Http3Request { + let request = HttpRequest { method: Method::GET, uri: Uri::from_static("/api/get/user%2D123/profile.json"), headers: Default::default(), body: Some(Bytes::new()), remote_addr: "127.0.0.1:4433".parse().unwrap(), }; - let response = handler(request, Http3Response::default(), params).await; + let response = handler(request, HttpResponse::default(), params).await; assert_eq!(response.body, vec![Bytes::from("user-123")]); } @@ -380,14 +380,14 @@ mod tests { let (handler, params) = router .pattern_handler(&Method::GET, "/api/users/profile.json") .unwrap(); - let request = Http3Request { + let request = HttpRequest { method: Method::GET, uri: Uri::from_static("/api/users/profile.json"), headers: Default::default(), body: None, remote_addr: "127.0.0.1:4433".parse().unwrap(), }; - let response = handler(request, Http3Response::default(), params).await; + let response = handler(request, HttpResponse::default(), params).await; assert_eq!(response.status, StatusCode::CREATED); } } diff --git a/mtp-webserver/src/server.rs b/mtp-webserver/src/server.rs index ddf28bc..0dd27ea 100644 --- a/mtp-webserver/src/server.rs +++ b/mtp-webserver/src/server.rs @@ -1,14 +1,17 @@ use crate::{ - Http3Request, Http3Response, Router, RouterError, WebMTPConnection, WebServerError, - transport::accept_web_connection, + HttpRequest, HttpResponse, Router, RouterError, WebMTPConnection, WebServerError, + h3::{DriverConfig, run_driver}, + tcp::{TcpDriverConfig, run_driver as run_tcp_driver}, }; -use bytes::{Buf, Bytes}; -use http::{Method, Request, Response, StatusCode}; +use http::Method; use mtp_common::CommunicationError; use mtp_host::HostConfig; use rustls::pki_types::{PrivateKeyDer, pem::PemObject}; use std::{net::SocketAddr, sync::Arc, time::Duration}; -use tokio::sync::{Semaphore, watch}; +use tokio::{ + net::TcpListener, + sync::{Semaphore, watch}, +}; /// Observability hooks for the web server. /// @@ -23,7 +26,7 @@ pub trait WebServerMetrics: Send + Sync { fn error_occurred(&self, _error: &WebServerError) {} } -/// Configuration for the HTTP/3 server and MTP routing. +/// Configuration for the HTTPS/HTTP/3 server and MTP routing. /// /// Use the builder methods to customise behaviour. All fields have sensible /// defaults so `WebServerConfig::new()` gives a usable production-ready @@ -34,6 +37,9 @@ pub struct WebServerConfig { pub(crate) mtp_path: String, pub max_request_body: usize, pub max_connections: usize, + pub serve_tcp_https: bool, + pub max_tcp_connections: usize, + pub tls_handshake_timeout: Duration, pub request_timeout: Duration, pub drain_timeout: Duration, pub(crate) metrics: Option>, @@ -52,6 +58,9 @@ impl WebServerConfig { mtp_path: "/".to_string(), max_request_body: 4 * 1024 * 1024, max_connections: 256, + serve_tcp_https: true, + max_tcp_connections: 256, + tls_handshake_timeout: Duration::from_secs(10), request_timeout: Duration::from_secs(30), drain_timeout: Duration::from_secs(5), metrics: None, @@ -60,8 +69,8 @@ impl WebServerConfig { pub fn route(mut self, path: impl Into, handler: F) -> Result where - F: Fn(Http3Request, Http3Response) -> Fut + Send + Sync + 'static, - Fut: std::future::Future + Send + 'static, + F: Fn(HttpRequest, HttpResponse) -> Fut + Send + Sync + 'static, + Fut: std::future::Future + Send + 'static, { self.router = self.router.route(path, handler)?; Ok(self) @@ -74,8 +83,8 @@ impl WebServerConfig { handler: F, ) -> Result where - F: Fn(Http3Request, Http3Response) -> Fut + Send + Sync + 'static, - Fut: std::future::Future + Send + 'static, + F: Fn(HttpRequest, HttpResponse) -> Fut + Send + Sync + 'static, + Fut: std::future::Future + Send + 'static, { self.router = self.router.route_method(method, path, handler)?; Ok(self) @@ -89,8 +98,8 @@ impl WebServerConfig { handler: F, ) -> Result where - F: Fn(Http3Request, Http3Response, crate::RouteParams) -> Fut + Send + Sync + 'static, - Fut: std::future::Future + Send + 'static, + F: Fn(HttpRequest, HttpResponse, crate::RouteParams) -> Fut + Send + Sync + 'static, + Fut: std::future::Future + Send + 'static, { self.router = self.router.route_pattern(pattern, handler)?; Ok(self) @@ -104,8 +113,8 @@ impl WebServerConfig { handler: F, ) -> Result where - F: Fn(Http3Request, Http3Response, crate::RouteParams) -> Fut + Send + Sync + 'static, - Fut: std::future::Future + Send + 'static, + F: Fn(HttpRequest, HttpResponse, crate::RouteParams) -> Fut + Send + Sync + 'static, + Fut: std::future::Future + Send + 'static, { self.router = self.router.route_pattern_method(method, pattern, handler)?; Ok(self) @@ -113,8 +122,8 @@ impl WebServerConfig { pub fn fallback(mut self, handler: F) -> Result where - F: Fn(Http3Request, Http3Response) -> Fut + Send + Sync + 'static, - Fut: std::future::Future + Send + 'static, + F: Fn(HttpRequest, HttpResponse) -> Fut + Send + Sync + 'static, + Fut: std::future::Future + Send + 'static, { self.router = self.router.fallback(handler)?; Ok(self) @@ -135,6 +144,21 @@ impl WebServerConfig { self } + pub fn serve_tcp_https(mut self, enabled: bool) -> Self { + self.serve_tcp_https = enabled; + self + } + + pub fn max_tcp_connections(mut self, max: usize) -> Self { + self.max_tcp_connections = max; + self + } + + pub fn tls_handshake_timeout(mut self, timeout: Duration) -> Self { + self.tls_handshake_timeout = timeout; + self + } + pub fn request_timeout(mut self, timeout: Duration) -> Self { self.request_timeout = timeout; self @@ -151,7 +175,7 @@ impl WebServerConfig { } } -/// An HTTP/3 server bound to MTP's configured address and certificate. +/// A combined HTTPS, HTTP/3, and WebTransport server. /// /// One task owns the Quinn endpoint and dispatches all HTTP/3 requests. This /// is the required ownership model for adding WebTransport MTP sessions on the @@ -160,52 +184,98 @@ pub struct MTPWebServer { endpoint: quinn::Endpoint, mtp_incoming: tokio::sync::mpsc::Receiver>, shutdown_tx: watch::Sender<()>, - _driver: Option>, + quic_driver: Option>, + tcp_driver: Option>, + local_addr: SocketAddr, } impl MTPWebServer { pub async fn new( - host_config: HostConfig, + mut host_config: HostConfig, web_config: WebServerConfig, ) -> Result { + mtp_crypto::ensure_crypto_provider(); + let certificates = + rustls::pki_types::CertificateDer::pem_slice_iter(&host_config.tls_fullchain) + .collect::, _>>() + .map_err(|_| CommunicationError::CertificateLoadFailed)?; + let key = PrivateKeyDer::from_pem_slice(&host_config.tls_key) + .map_err(|_| CommunicationError::CertificateParseFailed)?; + let tcp_listener = if web_config.serve_tcp_https { + let listener = TcpListener::bind(SocketAddr::new(host_config.ip, host_config.port)) + .await + .map_err(|error| CommunicationError::Other(error.to_string()))?; + host_config.port = listener + .local_addr() + .map_err(|error| CommunicationError::Other(error.to_string()))? + .port(); + Some(listener) + } else { + None + }; + let tcp_tls = tcp_listener + .as_ref() + .map(|_| build_tcp_tls(&certificates, key.clone_key())) + .transpose()?; + let endpoint = build_endpoint(&host_config, certificates, key)?; + let local_addr = endpoint + .local_addr() + .map_err(|error| CommunicationError::Other(error.to_string()))?; let host_config = Arc::new(host_config); - let endpoint = build_endpoint(&host_config)?; let driver_endpoint = endpoint.clone(); // A completed MTP handshake must never block the endpoint driver just // because the application is briefly slow to call `accept()`. let (mtp_tx, mtp_incoming) = tokio::sync::mpsc::channel(web_config.max_connections.max(1)); let (shutdown_tx, shutdown_rx) = watch::channel(()); let connection_semaphore = Arc::new(Semaphore::new(web_config.max_connections)); + let router = web_config.router.clone(); + let metrics = web_config.metrics.clone(); let driver_config = DriverConfig { - router: web_config.router, - mtp_path: web_config.mtp_path, + router: web_config.router.clone(), + mtp_path: web_config.mtp_path.clone(), max_request_body: web_config.max_request_body, request_timeout: web_config.request_timeout, drain_timeout: web_config.drain_timeout, send_pongs: host_config.send_pongs, policy: host_config.policy, host_config, - metrics: web_config.metrics, + metrics: web_config.metrics.clone(), }; - let driver = tokio::spawn(run_driver( + let quic_driver = tokio::spawn(run_driver( driver_endpoint, driver_config, mtp_tx, connection_semaphore, - shutdown_rx, + shutdown_rx.clone(), )); + let tcp_driver = tcp_listener.zip(tcp_tls).map(|(listener, tls)| { + tokio::spawn(run_tcp_driver( + listener, + tls, + TcpDriverConfig { + router, + max_request_body: web_config.max_request_body, + request_timeout: web_config.request_timeout, + tls_handshake_timeout: web_config.tls_handshake_timeout, + drain_timeout: web_config.drain_timeout, + max_connections: web_config.max_tcp_connections, + metrics, + }, + shutdown_rx, + )) + }); Ok(Self { endpoint, mtp_incoming, shutdown_tx, - _driver: Some(driver), + quic_driver: Some(quic_driver), + tcp_driver, + local_addr, }) } pub fn local_addr(&self) -> SocketAddr { - self.endpoint - .local_addr() - .expect("endpoint has a local address") + self.local_addr } pub async fn accept(&mut self) -> Result, mtp_host::AcceptError> { @@ -222,7 +292,10 @@ impl MTPWebServer { /// in-flight requests to complete before closing the endpoint. pub async fn shutdown(mut self) { let _ = self.shutdown_tx.send(()); - if let Some(driver) = self._driver.take() { + if let Some(driver) = self.quic_driver.take() { + let _ = driver.await; + } + if let Some(driver) = self.tcp_driver.take() { let _ = driver.await; } self.endpoint @@ -233,40 +306,33 @@ impl MTPWebServer { pub async fn close(mut self) { self.endpoint .close(quinn::VarInt::from_u32(0), b"mtp-webserver shutdown"); - if let Some(driver) = self._driver.take() { + if let Some(driver) = self.quic_driver.take() { driver.abort(); + let _ = driver.await; + } + if let Some(driver) = self.tcp_driver.take() { + driver.abort(); + let _ = driver.await; } } } impl Drop for MTPWebServer { fn drop(&mut self) { - if let Some(driver) = self._driver.take() { + if let Some(driver) = self.quic_driver.take() { + driver.abort(); + } + if let Some(driver) = self.tcp_driver.take() { driver.abort(); } } } -struct DriverConfig { - router: Router, - mtp_path: String, - max_request_body: usize, - request_timeout: Duration, - drain_timeout: Duration, - send_pongs: bool, - policy: mtp_transport::Policy, - host_config: Arc, - metrics: Option>, -} - -fn build_endpoint(config: &HostConfig) -> Result { - mtp_crypto::ensure_crypto_provider(); - let certificates = rustls::pki_types::CertificateDer::pem_slice_iter(&config.tls_fullchain) - .collect::, _>>() - .map_err(|_| CommunicationError::CertificateLoadFailed)?; - let key = PrivateKeyDer::from_pem_slice(&config.tls_key) - .map_err(|_| CommunicationError::CertificateParseFailed)?; - +fn build_endpoint( + config: &HostConfig, + certificates: Vec>, + key: PrivateKeyDer<'static>, +) -> Result { let mut tls = rustls::ServerConfig::builder() .with_no_client_auth() .with_single_cert(certificates, key) @@ -281,373 +347,14 @@ fn build_endpoint(config: &HostConfig) -> Result>, - connection_semaphore: Arc, - mut shutdown_rx: watch::Receiver<()>, -) { - let DriverConfig { - router, - mtp_path, - max_request_body, - request_timeout, - drain_timeout, - send_pongs, - policy, - host_config, - metrics, - } = config; - let mut connection_tasks = tokio::task::JoinSet::new(); - loop { - tokio::select! { - biased; - _ = shutdown_rx.changed() => { - break; - } - incoming = endpoint.accept() => { - let Some(incoming) = incoming else { - break; - }; - // Do not await capacity here: doing so would prevent this loop - // from observing shutdown while all connection slots are in use. - let permit = match connection_semaphore.clone().try_acquire_owned() { - Ok(permit) => permit, - Err(_) => { - tracing::debug!("rejecting QUIC connection at configured connection limit"); - continue; - } - }; - let router = router.clone(); - let mtp_path = mtp_path.clone(); - let mtp_tx = mtp_tx.clone(); - let metrics = metrics.clone(); - let host_config = host_config.clone(); - connection_tasks.spawn(async move { - let _permit = permit; - let connect_start = std::time::Instant::now(); - let connection = match incoming.await { - Ok(connection) => connection, - Err(error) => { - tracing::debug!(%error, "QUIC connection failed during handshake"); - return; - } - }; - let mut builder = h3::server::builder(); - builder.enable_extended_connect(true); - builder.enable_webtransport(true); - builder.enable_datagram(true); - builder.max_webtransport_sessions(16); - let mut h3 = match builder - .build(h3_quinn::Connection::new(connection.clone())) - .await - { - Ok(connection) => connection, - Err(error) => { - tracing::debug!(%error, "HTTP/3 connection setup failed"); - return; - } - }; - - if let Some(ref m) = metrics { - m.connection_accepted(); - } - let remote_addr = connection.remote_address(); - - let mut tasks = tokio::task::JoinSet::new(); - loop { - let resolver = match h3.accept().await { - Ok(Some(resolver)) => resolver, - Ok(None) => break, - Err(error) => { - tracing::debug!(%error, "HTTP/3 request accept failed"); - break; - } - }; - let (request, mut stream) = match resolver.resolve_request().await { - Ok(request) => request, - Err(error) => { - tracing::debug!(%error, "HTTP/3 request parse failed"); - continue; - } - }; - if request.method() == http::Method::CONNECT && request.uri().path() == mtp_path { - if request.extensions().get::() - != Some(&h3::ext::Protocol::WEB_TRANSPORT) - { - let _ = stream - .send_response( - Response::builder() - .status(StatusCode::METHOD_NOT_ALLOWED) - .body(()) - .unwrap(), - ) - .await; - let _ = stream.finish().await; - continue; - } - let session = match h3_webtransport::server::WebTransportSession::accept( - request, stream, h3, - ) - .await - { - Ok(session) => Arc::new(session), - Err(error) => { - tracing::debug!(%error, "WebTransport session accept failed"); - return; - } - }; - // The WebTransport session request driver must outlive this - // endpoint request task. Keep it detached so handing the MTP - // connection to the application does not wait for the session - // (which is intentionally an open-ended accept loop). - tokio::spawn(run_session_requests( - session.clone(), - router.clone(), - max_request_body, - request_timeout, - metrics.clone(), - remote_addr, - )); - let result = - accept_web_connection(session, mtp_path, connection, send_pongs, policy, host_config.clone()) - .await; - match mtp_tx.try_send(result) { - Ok(()) => { - // The detached session driver remains active while the - // delivered MTP connection keeps the session alive. - } - Err(tokio::sync::mpsc::error::TrySendError::Full(result)) => { - tracing::warn!("MTP connection backlog is full; dropping connection"); - if let Ok(connection) = result { - connection.sender.close(); - } - } - Err(tokio::sync::mpsc::error::TrySendError::Closed(result)) => { - if let Ok(connection) = result { - connection.sender.close(); - } - } - } - return; - } - let router = router.clone(); - let metrics = metrics.clone(); - tasks.spawn(async move { - let path = request.uri().path().to_string(); - if let Some(ref m) = metrics { - m.request_started(&path); - } - let req_start = std::time::Instant::now(); - let (response, status) = match tokio::time::timeout( - request_timeout, - handle_http_request(request, &mut stream, &router, max_request_body, remote_addr), - ) - .await - { - Ok(Ok((resp, status))) => (resp, status), - Ok(Err(_)) => ( - Http3Response::new(StatusCode::BAD_GATEWAY), - StatusCode::BAD_GATEWAY, - ), - Err(_) => ( - Http3Response::new(StatusCode::REQUEST_TIMEOUT), - StatusCode::REQUEST_TIMEOUT, - ), - }; - let _ = write_response(&mut stream, response).await; - if let Some(ref m) = metrics { - m.request_completed(&path, status.as_u16(), req_start.elapsed()); - } - }); - } - tasks.join_all().await; - if let Some(ref m) = metrics { - m.connection_closed(connect_start.elapsed(), "normal"); - } - }); - } - } - } - - // --- Drain phase: wait for in-flight connections --- - - endpoint.close(quinn::VarInt::from_u32(0), b"mtp-webserver shutdown"); - - let drain_start = std::time::Instant::now(); - while !connection_tasks.is_empty() { - tokio::select! { - Some(result) = connection_tasks.join_next() => { - if let Err(e) = result { - tracing::warn!("Connection task panicked: {}", e); - } - } - _ = tokio::time::sleep(drain_timeout.saturating_sub(drain_start.elapsed())) => { - tracing::warn!( - "Drain timeout expired with {} connections still in flight", - connection_tasks.len() - ); - break; - } - } - } - - connection_tasks.shutdown().await; -} - -async fn handle_http_request( - request: Request<()>, - stream: &mut h3::server::RequestStream, - router: &Router, - max_request_body: usize, - remote_addr: SocketAddr, -) -> Result<(Http3Response, StatusCode), WebServerError> -where - S: h3::quic::BidiStream, -{ - let (request, too_large) = read_request(request, stream, max_request_body, remote_addr) - .await - .map_err(|e| WebServerError::Http(format!("request body read failed: {e}")))?; - if too_large { - return Ok(( - Http3Response::new(StatusCode::PAYLOAD_TOO_LARGE), - StatusCode::PAYLOAD_TOO_LARGE, - )); - } - let path = request.uri.path().to_string(); - match router.handler(&request.method, &path) { - Some(handler) => { - let response = handler(request, Http3Response::default()).await; - let status = response.status; - Ok((response, status)) - } - None => match router.pattern_handler(&request.method, &path) { - Some((handler, params)) => { - let response = handler(request, Http3Response::default(), params).await; - let status = response.status; - Ok((response, status)) - } - None => match router.fallback_handler() { - Some(handler) => { - let response = handler(request, Http3Response::default()).await; - let status = response.status; - Ok((response, status)) - } - None => Ok(( - Http3Response::new(StatusCode::NOT_FOUND), - StatusCode::NOT_FOUND, - )), - }, - }, - } -} - -async fn run_session_requests( - session: Arc>, - router: Router, - max_request_body: usize, - request_timeout: Duration, - metrics: Option>, - remote_addr: SocketAddr, -) { - loop { - match session.accept_bi().await { - Ok(Some(h3_webtransport::server::AcceptedBi::Request(request, mut stream))) => { - let router = router.clone(); - let metrics = metrics.clone(); - tokio::spawn(async move { - let path = request.uri().path().to_string(); - if let Some(ref m) = metrics { - m.request_started(&path); - } - let req_start = std::time::Instant::now(); - let (response, status) = match tokio::time::timeout( - request_timeout, - handle_http_request( - request, - &mut stream, - &router, - max_request_body, - remote_addr, - ), - ) - .await - { - Ok(Ok((resp, status))) => (resp, status), - Ok(Err(_)) => ( - Http3Response::new(StatusCode::BAD_GATEWAY), - StatusCode::BAD_GATEWAY, - ), - Err(_) => ( - Http3Response::new(StatusCode::REQUEST_TIMEOUT), - StatusCode::REQUEST_TIMEOUT, - ), - }; - let _ = write_response(&mut stream, response).await; - if let Some(ref m) = metrics { - m.request_completed(&path, status.as_u16(), req_start.elapsed()); - } - }); - } - Ok(Some(h3_webtransport::server::AcceptedBi::BidiStream(_, _))) => {} - Ok(None) | Err(_) => break, - } - } -} - -async fn read_request( - request: Request<()>, - stream: &mut h3::server::RequestStream, - max_body: usize, - remote_addr: SocketAddr, -) -> Result<(Http3Request, bool), h3::error::StreamError> -where - S: h3::quic::BidiStream, -{ - let (parts, _) = request.into_parts(); - let mut body = Vec::new(); - let mut too_large = false; - while let Some(chunk) = stream.recv_data().await? { - if body.len().saturating_add(chunk.remaining()) > max_body { - too_large = true; - break; - } - body.extend_from_slice(chunk.chunk()); - } - Ok(( - Http3Request { - method: parts.method, - uri: parts.uri, - headers: parts.headers, - body: (!body.is_empty()).then(|| Bytes::from(body)), - remote_addr, - }, - too_large, - )) -} - -async fn write_response( - stream: &mut h3::server::RequestStream, - response: Http3Response, -) -> Result<(), h3::error::StreamError> -where - S: h3::quic::BidiStream, -{ - let mut builder = Response::builder().status(response.status); - for (name, value) in &response.headers { - builder = builder.header(name, value); - } - stream - .send_response(builder.body(()).expect("valid HTTP response")) - .await?; - for chunk in response.body { - stream.send_data(chunk).await?; - } - if let Some(mut chunks) = response.stream { - while let Some(chunk) = chunks.recv().await { - stream.send_data(chunk).await?; - } - } - stream.finish().await +fn build_tcp_tls( + certificates: &[rustls::pki_types::CertificateDer<'static>], + key: PrivateKeyDer<'static>, +) -> Result, CommunicationError> { + let mut tls = rustls::ServerConfig::builder() + .with_no_client_auth() + .with_single_cert(certificates.to_vec(), key) + .map_err(|_| CommunicationError::CertificateLoadFailed)?; + tls.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()]; + Ok(Arc::new(tls)) } diff --git a/mtp-webserver/src/stream.rs b/mtp-webserver/src/stream.rs index 833e99f..7e87f17 100644 --- a/mtp-webserver/src/stream.rs +++ b/mtp-webserver/src/stream.rs @@ -3,9 +3,9 @@ use http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode, Uri}; use std::net::SocketAddr; use tokio::sync::mpsc; -/// An owned HTTP/3 request passed to a route handler. +/// An owned HTTP request passed to a route handler. #[derive(Clone, Debug)] -pub struct Http3Request { +pub struct HttpRequest { pub method: Method, pub uri: Uri, pub headers: HeaderMap, @@ -13,15 +13,15 @@ pub struct Http3Request { pub remote_addr: SocketAddr, } -/// A buffered HTTP/3 response returned from a route handler. -pub struct Http3Response { +/// An HTTP response returned from a route handler. +pub struct HttpResponse { pub status: StatusCode, pub headers: HeaderMap, pub body: Vec, pub(crate) stream: Option>, } -impl Http3Response { +impl HttpResponse { pub fn new(status: StatusCode) -> Self { Self { status, @@ -68,19 +68,25 @@ impl Http3Response { } } -impl Default for Http3Response { +impl Default for HttpResponse { fn default() -> Self { Self::new(StatusCode::OK) } } +#[deprecated(note = "use HttpRequest")] +pub type Http3Request = HttpRequest; + +#[deprecated(note = "use HttpResponse")] +pub type Http3Response = HttpResponse; + #[cfg(test)] mod tests { use super::*; #[test] fn response_collects_headers_and_body_chunks() { - let response = Http3Response::new(StatusCode::CREATED) + let response = HttpResponse::new(StatusCode::CREATED) .header("content-type", "text/plain") .body("hello") .body(" world"); diff --git a/mtp-webserver/src/tcp.rs b/mtp-webserver/src/tcp.rs new file mode 100644 index 0000000..2812ab1 --- /dev/null +++ b/mtp-webserver/src/tcp.rs @@ -0,0 +1,202 @@ +use crate::{HttpRequest, HttpResponse, Router, WebServerError, WebServerMetrics}; +use bytes::Bytes; +use http::{Request, Response, header::CONTENT_LENGTH}; +use http_body_util::{BodyExt, Full, StreamBody, combinators::BoxBody}; +use hyper::{ + body::{Frame, Incoming}, + service::service_fn, +}; +use hyper_util::{ + rt::{TokioExecutor, TokioIo}, + server::conn::auto::Builder, +}; +use std::{convert::Infallible, sync::Arc, time::Duration}; +use tokio::{ + net::TcpListener, + sync::{Semaphore, watch}, + task::JoinSet, +}; +use tokio_rustls::TlsAcceptor; +use tokio_stream::{StreamExt, wrappers::ReceiverStream}; + +pub(crate) struct TcpDriverConfig { + pub router: Router, + pub max_request_body: usize, + pub request_timeout: Duration, + pub tls_handshake_timeout: Duration, + pub drain_timeout: Duration, + pub max_connections: usize, + pub metrics: Option>, +} + +pub(crate) async fn run_driver( + listener: TcpListener, + tls: Arc, + config: TcpDriverConfig, + mut shutdown_rx: watch::Receiver<()>, +) { + let permits = Arc::new(Semaphore::new(config.max_connections)); + let mut tasks = JoinSet::new(); + loop { + tokio::select! { + biased; + _ = shutdown_rx.changed() => break, + accepted = listener.accept() => { + let Ok((stream, remote_addr)) = accepted else { break }; + let Ok(permit) = permits.clone().try_acquire_owned() else { + tracing::debug!(%remote_addr, "rejecting TCP connection at configured connection limit"); + continue; + }; + let acceptor = TlsAcceptor::from(tls.clone()); + let router = config.router.clone(); + let metrics = config.metrics.clone(); + let timeout = config.request_timeout; + let body_limit = config.max_request_body; + let handshake_timeout = config.tls_handshake_timeout; + let connection_shutdown = shutdown_rx.clone(); + tasks.spawn(async move { + let _permit = permit; + let started = std::time::Instant::now(); + let tls_stream = match tokio::time::timeout(handshake_timeout, acceptor.accept(stream)).await { + Ok(Ok(stream)) => stream, + Ok(Err(error)) => { + report_error(metrics.as_ref(), WebServerError::Tls(error.to_string())); + return; + } + Err(_) => { + report_error(metrics.as_ref(), WebServerError::Tls("TLS handshake timed out".into())); + return; + } + }; + if let Some(metrics) = &metrics { metrics.connection_accepted(); } + let service_metrics = metrics.clone(); + let service = service_fn(move |request| { + serve_request(request, remote_addr, router.clone(), body_limit, timeout, service_metrics.clone()) + }); + let builder = Builder::new(TokioExecutor::new()); + let connection = builder.serve_connection_with_upgrades(TokioIo::new(tls_stream), service); + tokio::pin!(connection); + tokio::select! { + result = &mut connection => { + if let Err(error) = result { + report_error(metrics.as_ref(), WebServerError::Http(format!("TCP HTTP connection failed: {error}"))); + } + } + _ = wait_for_shutdown(connection_shutdown) => { + connection.as_mut().graceful_shutdown(); + if let Err(error) = connection.await { + tracing::debug!(%error, "TCP HTTP connection ended during shutdown"); + } + } + } + if let Some(metrics) = &metrics { metrics.connection_closed(started.elapsed(), "normal"); } + }); + } + } + } + + let drain = async { while tasks.join_next().await.is_some() {} }; + if tokio::time::timeout(config.drain_timeout, drain) + .await + .is_err() + { + tasks.shutdown().await; + } +} + +async fn wait_for_shutdown(mut shutdown_rx: watch::Receiver<()>) { + let _ = shutdown_rx.changed().await; +} + +async fn serve_request( + request: Request, + remote_addr: std::net::SocketAddr, + router: Router, + max_body: usize, + timeout: Duration, + metrics: Option>, +) -> Result>, Infallible> { + let path = request.uri().path().to_string(); + let response = crate::http::run_request(&path, timeout, metrics.as_ref(), async move { + let request = read_request(request, remote_addr, max_body).await?; + Ok(crate::http::dispatch_request(request, &router).await) + }) + .await; + Ok(into_hyper_response(response)) +} + +async fn read_request( + request: Request, + remote_addr: std::net::SocketAddr, + max_body: usize, +) -> Result { + let (parts, mut body) = request.into_parts(); + if parts + .headers + .get(CONTENT_LENGTH) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()) + .is_some_and(|length| length > max_body) + { + return Err(WebServerError::PayloadTooLarge); + } + let mut chunks = Vec::new(); + let mut size = 0usize; + while let Some(frame) = body.frame().await { + let frame = frame + .map_err(|error| WebServerError::Http(format!("request body read failed: {error}")))?; + if let Ok(data) = frame.into_data() { + size = size.saturating_add(data.len()); + if size > max_body { + return Err(WebServerError::PayloadTooLarge); + } + chunks.push(data); + } + } + let body = if chunks.is_empty() { + None + } else { + let mut combined = bytes::BytesMut::with_capacity(size); + for chunk in chunks { + combined.extend_from_slice(&chunk); + } + Some(combined.freeze()) + }; + Ok(HttpRequest { + method: parts.method, + uri: parts.uri, + headers: parts.headers, + body, + remote_addr, + }) +} + +fn into_hyper_response(mut response: HttpResponse) -> Response> { + let mut builder = Response::builder().status(response.status); + if let Some(headers) = builder.headers_mut() { + *headers = response.headers; + } + let body = if let Some(stream) = response.stream.take() { + let stream = ReceiverStream::new(stream).map(|chunk| Ok(Frame::data(chunk))); + BodyExt::boxed(StreamBody::new(stream)) + } else { + let length: usize = response.body.iter().map(Bytes::len).sum(); + if let Some(headers) = builder.headers_mut() + && !headers.contains_key(CONTENT_LENGTH) + { + headers.insert(CONTENT_LENGTH, length.into()); + } + let mut body = bytes::BytesMut::with_capacity(length); + for chunk in response.body { + body.extend_from_slice(&chunk); + } + Full::new(body.freeze()).boxed() + }; + builder.body(body).expect("valid HTTP response") +} + +fn report_error(metrics: Option<&Arc>, error: WebServerError) { + if let Some(metrics) = metrics { + metrics.error_occurred(&error); + } +} diff --git a/mtp-webserver/tests/integration.rs b/mtp-webserver/tests/integration.rs index 2c376cf..0848b12 100644 --- a/mtp-webserver/tests/integration.rs +++ b/mtp-webserver/tests/integration.rs @@ -1,9 +1,11 @@ -use http::StatusCode; +use http::{Method, StatusCode}; use mtp_webserver::{MTPWebServer, WebServerConfig, WebServerError, WebServerMetrics}; -use std::net::IpAddr; +use rustls::pki_types::{CertificateDer, ServerName, pem::PemObject}; +use std::net::{IpAddr, SocketAddr}; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; fn generate_self_signed_cert() -> (Vec, Vec) { let key_pair = rcgen::KeyPair::generate().expect("failed to generate self-signed key pair"); @@ -18,11 +20,52 @@ fn generate_self_signed_cert() -> (Vec, Vec) { ) } +fn host_config(port: u16, cert: Vec, key: Vec) -> mtp_host::HostConfig { + mtp_host::HostConfig::new(IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), port, cert, key) +} + +async fn tls_connect( + addr: SocketAddr, + cert_pem: &[u8], + alpn: Vec>, +) -> tokio_rustls::client::TlsStream { + let certs = CertificateDer::pem_slice_iter(cert_pem) + .collect::, _>>() + .unwrap(); + let mut roots = rustls::RootCertStore::empty(); + for cert in certs { + roots.add(cert).unwrap(); + } + let mut config = rustls::ClientConfig::builder() + .with_root_certificates(roots) + .with_no_client_auth(); + config.alpn_protocols = alpn; + let connector = tokio_rustls::TlsConnector::from(Arc::new(config)); + connector + .connect( + ServerName::try_from("localhost").unwrap().to_owned(), + tokio::net::TcpStream::connect(addr).await.unwrap(), + ) + .await + .unwrap() +} + +async fn http1_request(addr: SocketAddr, cert_pem: &[u8], request: &str) -> Vec { + let mut stream = tls_connect(addr, cert_pem, vec![b"http/1.1".to_vec()]).await; + stream.write_all(request.as_bytes()).await.unwrap(); + let mut response = Vec::new(); + stream.read_to_end(&mut response).await.unwrap(); + response +} + #[test] fn config_builder_defaults() { let config = WebServerConfig::new(); assert_eq!(config.max_request_body, 4 * 1024 * 1024); assert_eq!(config.max_connections, 256); + assert!(config.serve_tcp_https); + assert_eq!(config.max_tcp_connections, 256); + assert_eq!(config.tls_handshake_timeout, Duration::from_secs(10)); assert_eq!(config.request_timeout, Duration::from_secs(30)); } @@ -30,10 +73,13 @@ fn config_builder_defaults() { fn config_builder_chain() { let config = WebServerConfig::new() .max_connections(64) + .max_tcp_connections(32) + .tls_handshake_timeout(Duration::from_secs(2)) .max_request_body(1024) .request_timeout(Duration::from_secs(5)) .mtp_path("/ws"); assert_eq!(config.max_connections, 64); + assert_eq!(config.max_tcp_connections, 32); assert_eq!(config.max_request_body, 1024); assert_eq!(config.request_timeout, Duration::from_secs(5)); } @@ -201,5 +247,301 @@ async fn graceful_shutdown_completes() { let server = MTPWebServer::new(host_config, WebServerConfig::new()) .await .unwrap(); + let addr = server.local_addr(); server.shutdown().await; + assert!(tokio::net::TcpStream::connect(addr).await.is_err()); + assert!(std::net::UdpSocket::bind(addr).is_ok()); +} + +#[tokio::test] +async fn tcp_and_udp_share_port_zero_assignment() { + let (cert, key) = generate_self_signed_cert(); + let server = MTPWebServer::new(host_config(0, cert, key), WebServerConfig::new()) + .await + .unwrap(); + let addr = server.local_addr(); + assert!(addr.port() > 0); + assert!(tokio::net::TcpStream::connect(addr).await.is_ok()); + assert!(std::net::UdpSocket::bind(addr).is_err()); + server.close().await; +} + +#[tokio::test] +async fn tcp_conflict_fails_and_udp_only_does_not_claim_tcp() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let (cert, key) = generate_self_signed_cert(); + assert!( + MTPWebServer::new(host_config(addr.port(), cert, key), WebServerConfig::new()) + .await + .is_err() + ); + drop(listener); + + let (cert, key) = generate_self_signed_cert(); + let server = MTPWebServer::new( + host_config(addr.port(), cert, key), + WebServerConfig::new().serve_tcp_https(false), + ) + .await + .unwrap(); + let tcp = tokio::net::TcpListener::bind(addr).await.unwrap(); + drop(tcp); + server.close().await; +} + +#[tokio::test] +async fn close_and_drop_release_tcp_listener() { + let (cert, key) = generate_self_signed_cert(); + let server = MTPWebServer::new(host_config(0, cert, key), WebServerConfig::new()) + .await + .unwrap(); + let addr = server.local_addr(); + server.close().await; + assert!(tokio::net::TcpListener::bind(addr).await.is_ok()); + + let (cert, key) = generate_self_signed_cert(); + let server = MTPWebServer::new(host_config(0, cert, key), WebServerConfig::new()) + .await + .unwrap(); + let addr = server.local_addr(); + drop(server); + tokio::task::yield_now().await; + assert!(tokio::net::TcpListener::bind(addr).await.is_ok()); +} + +#[tokio::test] +async fn http1_routes_bodies_chunks_and_timeout() { + let (cert, key) = generate_self_signed_cert(); + let config = WebServerConfig::new() + .max_request_body(8) + .request_timeout(Duration::from_millis(20)) + .route_method(Method::POST, "/exact", |request, response| async move { + response.body(request.body.unwrap()).body("-tail") + }) + .unwrap() + .route_pattern("/users/{user}", |request, response, params| async move { + response.body(format!( + "{}:{}", + params["user"], + request.uri.query().unwrap_or("") + )) + }) + .unwrap() + .route("/slow", |_, response| async move { + tokio::time::sleep(Duration::from_millis(100)).await; + response + }) + .unwrap() + .fallback(|_, response| async move { response.status(StatusCode::IM_A_TEAPOT) }) + .unwrap(); + let server = MTPWebServer::new(host_config(0, cert.clone(), key), config) + .await + .unwrap(); + let addr = server.local_addr(); + + let response = http1_request(addr, &cert, "POST /exact HTTP/1.1\r\nHost: localhost\r\nContent-Length: 4\r\nConnection: close\r\n\r\ndata").await; + assert!(String::from_utf8_lossy(&response).ends_with("data-tail")); + let response = http1_request( + addr, + &cert, + "GET /users/alice%2Dsmith?full=1 HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", + ) + .await; + assert!(String::from_utf8_lossy(&response).ends_with("alice-smith:full=1")); + let response = http1_request( + addr, + &cert, + "GET /exact HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", + ) + .await; + assert!(String::from_utf8_lossy(&response).starts_with("HTTP/1.1 418")); + let response = http1_request(addr, &cert, "POST /exact HTTP/1.1\r\nHost: localhost\r\nContent-Length: 9\r\nConnection: close\r\n\r\n123456789").await; + assert!(String::from_utf8_lossy(&response).starts_with("HTTP/1.1 413")); + let response = http1_request( + addr, + &cert, + "GET /slow HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", + ) + .await; + assert!(String::from_utf8_lossy(&response).starts_with("HTTP/1.1 408")); + server.close().await; +} + +#[tokio::test] +async fn tls_negotiates_h2() { + let (cert, key) = generate_self_signed_cert(); + let config = WebServerConfig::new() + .route("/h2", |_, response| async move { response.body("over-h2") }) + .unwrap(); + let server = MTPWebServer::new(host_config(0, cert.clone(), key), config) + .await + .unwrap(); + let stream = tls_connect(server.local_addr(), &cert, vec![b"h2".to_vec()]).await; + assert_eq!(stream.get_ref().1.alpn_protocol(), Some(b"h2".as_slice())); + let (mut sender, connection) = hyper::client::conn::http2::handshake( + hyper_util::rt::TokioExecutor::new(), + hyper_util::rt::TokioIo::new(stream), + ) + .await + .unwrap(); + tokio::spawn(async move { + let _ = connection.await; + }); + let request = http::Request::builder() + .uri("https://localhost/h2") + .body(http_body_util::Empty::::new()) + .unwrap(); + let response = sender.send_request(request).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = http_body_util::BodyExt::collect(response.into_body()) + .await + .unwrap() + .to_bytes(); + assert_eq!(body, "over-h2"); + server.close().await; +} + +#[tokio::test] +async fn tcp_metrics_report_completion_and_errors() { + let (cert, key) = generate_self_signed_cert(); + let metrics = Arc::new(TestMetrics::new()); + let config = WebServerConfig::new() + .request_timeout(Duration::from_millis(10)) + .with_metrics(metrics.clone()) + .route("/slow", |_, response| async move { + tokio::time::sleep(Duration::from_millis(50)).await; + response + }) + .unwrap(); + let server = MTPWebServer::new(host_config(0, cert.clone(), key), config) + .await + .unwrap(); + let response = http1_request( + server.local_addr(), + &cert, + "GET /slow HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", + ) + .await; + assert!(String::from_utf8_lossy(&response).starts_with("HTTP/1.1 408")); + assert_eq!(metrics.requests_started.load(Ordering::SeqCst), 1); + assert_eq!(metrics.requests_completed.load(Ordering::SeqCst), 1); + assert_eq!(metrics.errors.load(Ordering::SeqCst), 1); + server.close().await; +} + +#[tokio::test] +async fn http1_keep_alive_and_streaming_work() { + let (cert, key) = generate_self_signed_cert(); + let config = WebServerConfig::new() + .route("/one", |_, response| async move { response.body("one") }) + .unwrap() + .route("/stream", |_, response| async move { + let (tx, rx) = tokio::sync::mpsc::channel(1); + tokio::spawn(async move { + tx.send(bytes::Bytes::from_static(b"first-")).await.unwrap(); + tokio::time::sleep(Duration::from_millis(10)).await; + let _ = tx.send(bytes::Bytes::from_static(b"second")).await; + }); + response.stream(rx) + }) + .unwrap(); + let server = MTPWebServer::new(host_config(0, cert.clone(), key), config) + .await + .unwrap(); + let addr = server.local_addr(); + let mut stream = tls_connect(addr, &cert, vec![b"http/1.1".to_vec()]).await; + stream.write_all(b"GET /one HTTP/1.1\r\nHost: localhost\r\n\r\nGET /one HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n").await.unwrap(); + let mut response = Vec::new(); + stream.read_to_end(&mut response).await.unwrap(); + assert_eq!( + String::from_utf8_lossy(&response) + .matches("HTTP/1.1 200") + .count(), + 2 + ); + + let response = http1_request( + addr, + &cert, + "GET /stream HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", + ) + .await; + let response = String::from_utf8_lossy(&response); + assert!(response.contains("first-")); + assert!(response.contains("second")); + assert!(!response.to_ascii_lowercase().contains("content-length:")); + server.close().await; +} + +#[tokio::test] +async fn shutdown_allows_active_request_within_drain_period() { + let (cert, key) = generate_self_signed_cert(); + let entered = Arc::new(tokio::sync::Notify::new()); + let handler_entered = Arc::clone(&entered); + let config = WebServerConfig::new() + .drain_timeout(Duration::from_millis(250)) + .route("/work", move |_, response| { + let entered = Arc::clone(&handler_entered); + async move { + entered.notify_one(); + tokio::time::sleep(Duration::from_millis(30)).await; + response.body("done") + } + }) + .unwrap(); + let server = MTPWebServer::new(host_config(0, cert.clone(), key), config) + .await + .unwrap(); + let addr = server.local_addr(); + let request = tokio::spawn(async move { + http1_request( + addr, + &cert, + "GET /work HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", + ) + .await + }); + entered.notified().await; + server.shutdown().await; + assert!(String::from_utf8_lossy(&request.await.unwrap()).ends_with("done")); +} + +#[tokio::test] +async fn shutdown_terminates_request_after_drain_period() { + let (cert, key) = generate_self_signed_cert(); + let entered = Arc::new(tokio::sync::Notify::new()); + let handler_entered = Arc::clone(&entered); + let config = WebServerConfig::new() + .drain_timeout(Duration::from_millis(20)) + .route("/stuck", move |_, response| { + let entered = Arc::clone(&handler_entered); + async move { + entered.notify_one(); + std::future::pending::<()>().await; + response + } + }) + .unwrap(); + let server = MTPWebServer::new(host_config(0, cert.clone(), key), config) + .await + .unwrap(); + let addr = server.local_addr(); + let request = tokio::spawn(async move { + let mut stream = tls_connect(addr, &cert, vec![b"http/1.1".to_vec()]).await; + stream + .write_all(b"GET /stuck HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + .await + .unwrap(); + let mut response = Vec::new(); + let _ = stream.read_to_end(&mut response).await; + response + }); + entered.notified().await; + server.shutdown().await; + let response = tokio::time::timeout(Duration::from_millis(250), request) + .await + .expect("connection task survived drain timeout") + .unwrap(); + assert!(!String::from_utf8_lossy(&response).contains("HTTP/1.1 200")); }