From b384c38dc71c12f3835f59e3fdc757a9f210a83c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Tue, 22 Sep 2026 19:35:23 +0200 Subject: [PATCH 1/3] Let a host build the rustls config and choose its crypto provider MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit turnloop_tls::ClientConfig and ServerConfig could only be built from roots or a chain, a key and ALPN, with rustls::crypto::ring hardwired as the provider. Everything else rustls can decide at config level (client certificates, SNI resolvers, custom verifiers, ticketers, protocol versions) was unreachable, so Perry's P5 lane bypassed the wrappers and P11 could not attach a client certificate to outbound requests. And a host that installs aws-lc-rs had no way to keep ring out of its binary. Wrapping an existing rustls config (#48): both configs gain `from_rustls(Arc, HostTime)` and `rustls_config()`. The wrappers' only state besides the rustls config is the host-supplied wall clock, so that clock becomes the public `HostTime`: the host passes `host_time.time_provider()` to `builder_with_details`, and certificate validity keeps following the `unix_seconds` handed to `process`, exactly as with `new`. `new` is now built on `from_rustls`. A constructor over public fields was preferred because it keeps the crate's connection handling and time plumbing in one place. Choosing the provider (#53): `ClientOptions::provider` and `ServerConfig::with_provider` take any `Arc`; `None` keeps the explicit ring provider, so current behaviour is unchanged. Selecting a provider alone would still link ring, so `ring` is now a default feature (`dep:ring` + `rustls/ring`). turnloop-tls declares rustls itself rather than inheriting the workspace entry, which enables ring for every member; the resolved rustls version and Cargo.lock are unchanged. Without `ring`, `new` uses the given provider or the rustls process default and returns an error if there is neither, and `tls_server_end_point` (which returns a ring Digest) is compiled out; the new provider-neutral `tls_server_end_point_hash` names the RFC 5929 hash for the host to compute. turnloop-http forwards the feature (`ring`, default on) and depends on turnloop-tls with default features off, since it never builds a config itself. `cargo tree -e normal -i ring` for turnloop-tls and turnloop-http with `--no-default-features` now prints nothing. Evidence (tests/tls.rs, tests/channel_binding.rs): - a client whose provider has only AES-128-GCM and a server whose provider has only ChaCha20 fail with NoCipherSuitesInCommon, and succeed when both carry ChaCha20; with ring hardwired they would have agreed on a suite; - a server built with from_rustls and a WebPkiClientVerifier receives exactly the client certificate a from_rustls client presents, and refuses a plain client with "peer sent no certificates"; the client's HostTime, created at 0, reads the time `process` supplied; - tls_server_end_point_hash agrees with the ring digest length for all 14 certificate fixtures (6 SHA-256, 4 SHA-384, 3 SHA-512, Ed25519 None). Breaking: ClientOptions has a new public field (struct literals need `..Default::default()`); turnloop-tls without default features no longer has `tls_server_end_point`. Closes #48 Closes #53 --- protocols/turnloop-http/Cargo.toml | 7 +- protocols/turnloop-tls/Cargo.toml | 11 +- protocols/turnloop-tls/README.md | 14 +- protocols/turnloop-tls/src/channel_binding.rs | 54 ++++-- protocols/turnloop-tls/src/lib.rs | 149 ++++++++++++--- .../turnloop-tls/tests/channel_binding.rs | 27 +++ protocols/turnloop-tls/tests/tls.rs | 170 +++++++++++++++++- 7 files changed, 383 insertions(+), 49 deletions(-) diff --git a/protocols/turnloop-http/Cargo.toml b/protocols/turnloop-http/Cargo.toml index 1981cd9..5e4f5e6 100644 --- a/protocols/turnloop-http/Cargo.toml +++ b/protocols/turnloop-http/Cargo.toml @@ -12,6 +12,10 @@ categories = ["network-programming"] documentation = "https://docs.rs/turnloop-http" [features] +default = ["ring"] +# turnloop-tls's default crypto provider. Disable default features to link only +# the rustls provider the host passes to turnloop-tls. +ring = ["turnloop-tls/ring"] turnloop = ["dep:turnloop-io", "turnloop-tls/turnloop"] pure-rust-zstd = ["dep:turnloop-zstd-decoder"] @@ -28,7 +32,8 @@ flate2.workspace = true brotli.workspace = true base64.workspace = true percent-encoding.workspace = true -turnloop-tls.workspace = true +# Not the workspace entry, so that the host decides whether ring is linked. +turnloop-tls = { path = "../turnloop-tls", version = "0.1.0-alpha.6", default-features = false } [dev-dependencies] rcgen.workspace = true diff --git a/protocols/turnloop-tls/Cargo.toml b/protocols/turnloop-tls/Cargo.toml index 17da587..d944296 100644 --- a/protocols/turnloop-tls/Cargo.toml +++ b/protocols/turnloop-tls/Cargo.toml @@ -12,20 +12,25 @@ categories = ["network-programming"] documentation = "https://docs.rs/turnloop-tls" [features] +default = ["ring"] turnloop = ["dep:turnloop-io"] +# The default crypto provider and the RFC 5929 hash. Disable default features and +# pass a provider (or install a rustls process default) to link only that one. +ring = ["dep:ring", "rustls/ring"] [dependencies] turnloop-io = { workspace = true, optional = true } -rustls.workspace = true -ring.workspace = true +# Not the workspace rustls entry, which enables ring for every member. +rustls = { version = "0.23", default-features = false, features = ["std", "tls12"] } +ring = { workspace = true, optional = true } webpki-roots.workspace = true [dev-dependencies] rcgen.workspace = true [target.'cfg(all(target_arch = "wasm32", target_os = "unknown"))'.dependencies] -ring = { workspace = true, features = ["wasm32_unknown_unknown_js"] } +ring = { workspace = true, optional = true, features = ["wasm32_unknown_unknown_js"] } rustls-pki-types = { workspace = true, features = ["web"] } [lints] diff --git a/protocols/turnloop-tls/README.md b/protocols/turnloop-tls/README.md index 19627e8..6a27366 100644 --- a/protocols/turnloop-tls/README.md +++ b/protocols/turnloop-tls/README.md @@ -5,13 +5,25 @@ ciphertext buffers, wall time and handshake deadlines. Handle each returned stat consume borrowed plaintext before advancing, and acknowledge encrypted writes only after their transport completion. No socket, scheduler or runtime is created. -Client/server configurations share the workspace’s ring provider (rustls defaults +By default client/server configurations use the ring provider (rustls defaults are disabled), including native Windows and wasm32. Browser entropy and PKI web features are enabled explicitly; building ring for wasm needs a wasm-capable C compiler. WASI obtains entropy through its host. No FIPS/PQ guarantee is claimed. +The provider is a host choice: `ClientOptions::provider` and +`ServerConfig::with_provider` take any rustls `CryptoProvider` (aws-lc-rs, for +example). `ring` is a default feature; with `default-features = false` the crate +links no provider of its own, and `new` uses the provider the host passes or the +rustls process default. Without `ring`, `tls_server_end_point` is unavailable and +`tls_server_end_point_hash` names the hash for the host to compute. + Mozilla roots, host-supplied extra PEM, explicit replacement CAs, SNI and ALPN are supported. See public API docs and `tests/tls.rs` for checked handshake examples. +Anything else rustls can configure (client certificates, SNI resolvers, custom +verifiers, ticketers) goes through `ClientConfig::from_rustls` / +`ServerConfig::from_rustls`: build the rustls config with +`builder_with_details(provider, host_time.time_provider())` so certificate checks +keep following the wall time passed to `process`. PEM parsing uses rustls-pki-types through rustls’s maintained re-export. ## Getting started on turnloop diff --git a/protocols/turnloop-tls/src/channel_binding.rs b/protocols/turnloop-tls/src/channel_binding.rs index 5ca6c04..28098a5 100644 --- a/protocols/turnloop-tls/src/channel_binding.rs +++ b/protocols/turnloop-tls/src/channel_binding.rs @@ -1,14 +1,36 @@ //! RFC 5929 certificate binding. This reads only the outer certificate envelope //! and signature AlgorithmIdentifier; certificate verification belongs to rustls. -use ring::digest::{self, Algorithm, Digest, SHA256, SHA384, SHA512}; +#[cfg(feature = "ring")] +use ring::digest::{self, Digest}; + +/// The hash RFC 5929 §4.1 selects for a certificate's tls-server-end-point. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum EndPointHash { + Sha256, + Sha384, + Sha512, +} +use EndPointHash::{Sha256, Sha384, Sha512}; /// Hash a verified server leaf's DER for RFC 5929 §4.1 tls-server-end-point. /// RSA/ECDSA SHA-256/384/512 and RSA-PSS parameters are supported. MD5/SHA-1 /// signatures use SHA-256. Unknown algorithms (including Ed25519), malformed DER /// and undefined bindings return `None`. This helper does not verify certificates /// and performs no allocation; call it after successful TLS authentication. +#[cfg(feature = "ring")] pub fn tls_server_end_point(certificate: &[u8]) -> Option { - Some(digest::digest(signature_hash(certificate)?, certificate)) + let algorithm = match tls_server_end_point_hash(certificate)? { + Sha256 => &digest::SHA256, + Sha384 => &digest::SHA384, + Sha512 => &digest::SHA512, + }; + Some(digest::digest(algorithm, certificate)) +} + +/// The hash `tls_server_end_point` would apply to this leaf's DER, for hosts +/// that hash with their own crypto provider. `None` in exactly the same cases. +pub fn tls_server_end_point_hash(certificate: &[u8]) -> Option { + signature_hash(certificate) } const RSA: &[u8] = b"\x2a\x86\x48\x86\xf7\x0d\x01\x01"; @@ -68,7 +90,7 @@ impl<'a> Der<'a> { } } -fn signature_hash(certificate: &[u8]) -> Option<&'static Algorithm> { +fn signature_hash(certificate: &[u8]) -> Option { let mut der = Der(certificate); let mut cert = Der(der.value(0x30)?); der.end()?; @@ -93,17 +115,17 @@ fn signature_hash(certificate: &[u8]) -> Option<&'static Algorithm> { } alg.null_or_absent()?; match suffix { - [4 | 5 | 11] => Some(&SHA256), // MD5, SHA-1, SHA-256 - [12] => Some(&SHA384), - [13] => Some(&SHA512), + [4 | 5 | 11] => Some(Sha256), // MD5, SHA-1, SHA-256 + [12] => Some(Sha384), + [13] => Some(Sha512), _ => None, } } else if let Some(suffix) = oid.strip_prefix(ECDSA) { alg.end()?; // ECDSA parameters MUST be absent. match suffix { - [2] => Some(&SHA256), - [3] => Some(&SHA384), - [4] => Some(&SHA512), + [2] => Some(Sha256), + [3] => Some(Sha384), + [4] => Some(Sha512), _ => None, } } else { @@ -111,27 +133,27 @@ fn signature_hash(certificate: &[u8]) -> Option<&'static Algorithm> { } } -fn hash_identifier(bytes: &[u8]) -> Option<&'static Algorithm> { +fn hash_identifier(bytes: &[u8]) -> Option { let mut der = Der(bytes); let mut alg = Der(der.value(0x30)?); der.end()?; let oid = alg.value(6)?; alg.null_or_absent()?; if oid == SHA1 { - return Some(&SHA256); + return Some(Sha256); } match oid.strip_prefix(SHA2)? { - [1] => Some(&SHA256), - [2] => Some(&SHA384), - [3] => Some(&SHA512), + [1] => Some(Sha256), + [2] => Some(Sha384), + [3] => Some(Sha512), _ => None, } } -fn pss_hash(parameters: &[u8]) -> Option<&'static Algorithm> { +fn pss_hash(parameters: &[u8]) -> Option { // RFC 4055: absent hashAlgorithm defaults to SHA-1 (RFC 5929 -> SHA-256). // MGF1's hash is validated separately; it does not select the binding hash. - let mut hash = &SHA256; + let mut hash = Sha256; let mut params = Der(parameters); let mut previous = 0; while let Some(&tag) = params.0.first() { diff --git a/protocols/turnloop-tls/src/lib.rs b/protocols/turnloop-tls/src/lib.rs index bd0f365..a36ed43 100644 --- a/protocols/turnloop-tls/src/lib.rs +++ b/protocols/turnloop-tls/src/lib.rs @@ -5,6 +5,25 @@ //! `TransmitTlsData` only after the host has completed all writes. Plaintext //! records borrow rustls storage; consume them before advancing the state. //! +//! # Configuration +//! [`ClientConfig::new`] and [`ServerConfig::new`] cover static TLS: roots or a +//! certificate chain, a key and ALPN. A host that needs a rustls decision those +//! constructors do not expose (client certificates, SNI resolvers, custom +//! verifiers, ticketers, protocol versions) builds the rustls config itself and +//! wraps it with [`ClientConfig::from_rustls`] / [`ServerConfig::from_rustls`], +//! passing [`HostTime::time_provider`] to `builder_with_details` so certificate +//! validity keeps following the time handed to `process`. +//! +//! # Crypto provider +//! The default `ring` feature makes `ring` the provider `new` uses when the host +//! names none. [`ClientOptions::provider`] and [`ServerConfig::with_provider`] +//! select any other rustls [`CryptoProvider`], +//! such as aws-lc-rs. With `default-features = false` this crate links no +//! provider of its own: `new` then uses the one the host passes, or the +//! process-default provider installed with rustls, and fails if there is neither. +//! Without `ring`, `tls_server_end_point` is unavailable; hash the certificate +//! with the algorithm [`tls_server_end_point_hash`] selects instead. +//! //! # Getting started on turnloop //! Enable the `turnloop` feature for the `asynchronous` module. The embedding //! host owns `LocalExecutor` and calls `turn`; adapters await its streams and @@ -15,6 +34,7 @@ use rustls::{ CertificateError, DigitallySignedStruct, Error, SignatureScheme, client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier}, + crypto::CryptoProvider, pki_types::{CertificateDer, PrivateKeyDer, ServerName, UnixTime, pem::PemObject}, }; use std::{ @@ -31,6 +51,48 @@ pub use rustls::unbuffered::{ConnectionState, UnbufferedStatus}; /// Wall time supplied by the host, never sampled from an operating-system clock. #[derive(Debug)] struct SuppliedTime(AtomicU64); + +/// A shared wall clock that only the host advances. +/// +/// Every connection made from a configuration stores the `unix_seconds` passed to +/// its `process` call here, and rustls reads it back through +/// [`time_provider`](Self::time_provider) when checking certificate validity. +/// Pass that provider to `rustls::ClientConfig::builder_with_details` (or the +/// server equivalent) before handing the result to `from_rustls`. +#[derive(Debug, Clone)] +pub struct HostTime(Arc); +impl HostTime { + pub fn new(unix_seconds: u64) -> Self { + Self(Arc::new(SuppliedTime(AtomicU64::new(unix_seconds)))) + } + pub fn set(&self, unix_seconds: u64) { + self.0.0.store(unix_seconds, Ordering::Relaxed); + } + pub fn unix_seconds(&self) -> u64 { + self.0.0.load(Ordering::Relaxed) + } + /// The rustls time provider backed by this clock. + pub fn time_provider(&self) -> Arc { + self.0.clone() + } +} + +/// The provider `new` uses when the host names none: `ring` when that feature is +/// enabled, otherwise the rustls process default. +fn default_provider() -> Result, Error> { + #[cfg(feature = "ring")] + { + Ok(Arc::new(rustls::crypto::ring::default_provider())) + } + #[cfg(not(feature = "ring"))] + { + CryptoProvider::get_default().cloned().ok_or_else(|| { + Error::General( + "no rustls crypto provider: pass one or install a process default".into(), + ) + }) + } +} impl rustls::time_provider::TimeProvider for SuppliedTime { fn current_time(&self) -> Option { Some(UnixTime::since_unix_epoch(Duration::from_secs( @@ -49,6 +111,10 @@ pub struct ClientOptions { pub extra_ca_pem: Vec, pub reject_unauthorized: bool, pub enable_sni: bool, + /// The rustls crypto provider for handshakes and signature checks. `None` + /// uses `ring` with the default `ring` feature, otherwise the process-default + /// provider. Naming one lets a host link exactly one provider. + pub provider: Option>, } impl Default for ClientOptions { fn default() -> Self { @@ -58,6 +124,7 @@ impl Default for ClientOptions { extra_ca_pem: Vec::new(), reject_unauthorized: true, enable_sni: true, + provider: None, } } } @@ -65,12 +132,15 @@ impl Default for ClientOptions { #[derive(Clone)] pub struct ClientConfig { config: Arc, - time: Arc, + time: HostTime, } impl ClientConfig { pub fn new(options: ClientOptions, unix_seconds: u64) -> Result { - let provider = Arc::new(rustls::crypto::ring::default_provider()); - let time = Arc::new(SuppliedTime(AtomicU64::new(unix_seconds))); + let provider = match options.provider { + Some(provider) => provider, + None => default_provider()?, + }; + let time = HostTime::new(unix_seconds); let mut roots = rustls::RootCertStore::empty(); if let Some(ca) = options.ca { for cert in ca { @@ -82,10 +152,11 @@ impl ClientConfig { roots.add(cert.map_err(|e| Error::General(e.to_string()))?)?; } } - let mut config = rustls::ClientConfig::builder_with_details(provider.clone(), time.clone()) - .with_safe_default_protocol_versions()? - .with_root_certificates(roots) - .with_no_client_auth(); + let mut config = + rustls::ClientConfig::builder_with_details(provider.clone(), time.time_provider()) + .with_safe_default_protocol_versions()? + .with_root_certificates(roots) + .with_no_client_auth(); if !options.reject_unauthorized { config .dangerous() @@ -93,10 +164,17 @@ impl ClientConfig { } config.alpn_protocols = options.alpn; config.enable_sni = options.enable_sni; - Ok(Self { - config: Arc::new(config), - time, - }) + Ok(Self::from_rustls(Arc::new(config), time)) + } + /// Wrap a rustls configuration the host built itself. Build it with + /// `builder_with_details(provider, time.time_provider())` so certificate + /// validity follows the time passed to `process`; with any other time + /// provider rustls keeps using that one, and `process` only updates `time`. + pub fn from_rustls(config: Arc, time: HostTime) -> Self { + Self { config, time } + } + pub fn rustls_config(&self) -> &Arc { + &self.config } /// Reuse this configuration across connections to share the session cache. pub fn connect(&self, name: ServerName<'static>) -> Result { @@ -112,29 +190,44 @@ impl ClientConfig { #[derive(Clone)] pub struct ServerConfig { config: Arc, - time: Arc, + time: HostTime, } impl ServerConfig { + /// Uses `ring` with the default `ring` feature, otherwise the process-default + /// provider; see [`with_provider`](Self::with_provider). pub fn new( chain: Vec>, key: PrivateKeyDer<'static>, alpn: Vec>, unix_seconds: u64, ) -> Result { - let time = Arc::new(SuppliedTime(AtomicU64::new(unix_seconds))); - let mut config = rustls::ServerConfig::builder_with_details( - Arc::new(rustls::crypto::ring::default_provider()), - time.clone(), - ) - .with_safe_default_protocol_versions()? - .with_no_client_auth() - .with_single_cert(chain, key)?; + Self::with_provider(default_provider()?, chain, key, alpn, unix_seconds) + } + /// [`new`](Self::new) with an explicit rustls crypto provider. + pub fn with_provider( + provider: Arc, + chain: Vec>, + key: PrivateKeyDer<'static>, + alpn: Vec>, + unix_seconds: u64, + ) -> Result { + let time = HostTime::new(unix_seconds); + let mut config = rustls::ServerConfig::builder_with_details(provider, time.time_provider()) + .with_safe_default_protocol_versions()? + .with_no_client_auth() + .with_single_cert(chain, key)?; config.alpn_protocols = alpn; // Rustls's default stateful session cache avoids a clock-reading ticketer. - Ok(Self { - config: Arc::new(config), - time, - }) + Ok(Self::from_rustls(Arc::new(config), time)) + } + /// Wrap a rustls configuration the host built itself; see + /// [`ClientConfig::from_rustls`] for how `time` reaches certificate checks. + /// A ticketer the host configures may read its own clock. + pub fn from_rustls(config: Arc, time: HostTime) -> Self { + Self { config, time } + } + pub fn rustls_config(&self) -> &Arc { + &self.config } pub fn accept(&self) -> Result { Ok(Server { @@ -147,13 +240,15 @@ impl ServerConfig { } mod channel_binding; +#[cfg(feature = "ring")] pub use channel_binding::tls_server_end_point; +pub use channel_binding::{EndPointHash, tls_server_end_point_hash}; macro_rules! endpoint { ($name:ident, $connection:ty, $data:ty) => { pub struct $name { inner: $connection, - time: Arc, + time: HostTime, deadline: Option, timed_out: bool, } @@ -164,7 +259,7 @@ macro_rules! endpoint { incoming: &'i mut [u8], unix_seconds: u64, ) -> UnbufferedStatus<'c, 'i, $data> { - self.time.0.store(unix_seconds, Ordering::Relaxed); + self.time.set(unix_seconds); if self.timed_out { return UnbufferedStatus { discard: 0, @@ -247,7 +342,7 @@ pub fn node_error_code(error: &Error) -> &'static str { #[derive(Debug)] struct Unverified { - provider: Arc, + provider: Arc, } impl ServerCertVerifier for Unverified { fn verify_server_cert( diff --git a/protocols/turnloop-tls/tests/channel_binding.rs b/protocols/turnloop-tls/tests/channel_binding.rs index 7a8b9c8..c6e1759 100644 --- a/protocols/turnloop-tls/tests/channel_binding.rs +++ b/protocols/turnloop-tls/tests/channel_binding.rs @@ -38,6 +38,33 @@ fn real_signature_algorithms_match_independent_certificate_digests() { assert_eq!(hashes, 13); } +#[test] +fn provider_neutral_hash_selection_matches_the_ring_digest() { + use turnloop_tls::{EndPointHash, tls_server_end_point_hash}; + let names: Vec<_> = include_str!("certificates/digests.txt") + .lines() + .map(|line| line.split_once(' ').expect("name and digest").0) + .collect(); + let mut selected = [0; 3]; + for (certificate, name) in CERTIFICATES.iter().zip(names) { + let hash = tls_server_end_point_hash(certificate); + let expected = match tls_server_end_point(certificate).map(|d| d.as_ref().len()) { + None => None, + Some(32) => Some(EndPointHash::Sha256), + Some(48) => Some(EndPointHash::Sha384), + Some(64) => Some(EndPointHash::Sha512), + Some(n) => panic!("{name}: unexpected digest length {n}"), + }; + assert_eq!(hash, expected, "{name}"); + if let Some(hash) = hash { + selected[hash as usize] += 1; + } + assert!(tls_server_end_point_hash(&certificate[..certificate.len() - 1]).is_none()); + } + // SHA-256: RSA MD5/SHA-1/SHA-256, ECDSA SHA-256, PSS SHA-1/SHA-256. + assert_eq!(selected, [6, 4, 3]); +} + #[test] fn truncated_and_malformed_certificates_never_produce_binding() { let mut truncated = 0; diff --git a/protocols/turnloop-tls/tests/tls.rs b/protocols/turnloop-tls/tests/tls.rs index 8963af1..90a000e 100644 --- a/protocols/turnloop-tls/tests/tls.rs +++ b/protocols/turnloop-tls/tests/tls.rs @@ -3,6 +3,7 @@ mod support; use std::{ io::{Read, Write}, net::{TcpListener, TcpStream}, + sync::Arc, thread, time::{Duration, Instant}, }; @@ -179,7 +180,6 @@ fn certificate_name_expiry_and_not_yet_valid_codes() { } #[test] fn ca_chain_and_sni_enabled_or_disabled() { - use std::sync::Arc; let mut root_params = rcgen::CertificateParams::new(vec!["Test Root".into()]).unwrap(); root_params.is_ca = rcgen::IsCa::Ca(rcgen::BasicConstraints::Unconstrained); let root_key = rcgen::KeyPair::generate().unwrap(); @@ -243,3 +243,171 @@ fn ca_chain_and_sni_enabled_or_disabled() { task.join().unwrap(); } } + +fn provider_with(suite: rustls::SupportedCipherSuite) -> Arc { + Arc::new(rustls::crypto::CryptoProvider { + cipher_suites: vec![suite], + ..rustls::crypto::ring::default_provider() + }) +} + +/// Handshake over loopback and exchange one byte each way. The server's `Ok` +/// carries the client chain it saw; each `Err` is that side's error text. +type Outcome = (Result>, String>, Result<(), String>); +fn handshake(client: ClientConfig, server: turnloop_tls::ServerConfig) -> Outcome { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let task = thread::spawn(move || { + let mut stream = Stream::new(server.accept().unwrap(), listener.accept().unwrap().0); + let mut byte = [0]; + stream.read_exact(&mut byte).map_err(|e| e.to_string())?; + assert_eq!(byte, [42]); + let peer = stream + .engine + .peer_certificates() + .unwrap_or_default() + .iter() + .map(|c| c.to_vec()) + .collect(); + stream.write_all(&[43]).map_err(|e| e.to_string())?; + Ok(peer) + }); + let mut stream = Stream::new( + client + .connect(ServerName::try_from("localhost").unwrap()) + .unwrap(), + TcpStream::connect(address).unwrap(), + ); + let mut reply = [0]; + let client = stream + .write_all(&[42]) + .and_then(|()| stream.read_exact(&mut reply)) + .map(|()| assert_eq!(reply, [43])) + .map_err(|e| e.to_string()); + drop(stream); + (task.join().unwrap(), client) +} + +#[test] +fn host_selected_crypto_provider_is_the_one_negotiating() { + use rustls::crypto::ring::cipher_suite::{ + TLS13_AES_128_GCM_SHA256, TLS13_CHACHA20_POLY1305_SHA256, + }; + let cert = certificate(); + let server = |suite| { + turnloop_tls::ServerConfig::with_provider( + provider_with(suite), + vec![cert.cert.der().clone()], + rustls::pki_types::PrivatePkcs8KeyDer::from(cert.key_pair.serialize_der()).into(), + vec![b"http/1.1".to_vec()], + NOW, + ) + .unwrap() + }; + let client = |suite| { + ClientConfig::new( + ClientOptions { + ca: Some(vec![cert.cert.der().clone()]), + provider: Some(provider_with(suite)), + ..Default::default() + }, + NOW, + ) + .unwrap() + }; + // Each side offers only its provider's one suite. Had either side kept the + // full ring provider, the two would have found a common suite. + let (server_result, client_result) = handshake( + client(TLS13_AES_128_GCM_SHA256), + server(TLS13_CHACHA20_POLY1305_SHA256), + ); + let server_error = server_result.unwrap_err(); + assert!( + server_error.contains("NoCipherSuitesInCommon"), + "{server_error}" + ); + // The test adapter drops the failed server connection without an alert. + assert!(client_result.is_err()); + let (server_result, client_result) = handshake( + client(TLS13_CHACHA20_POLY1305_SHA256), + server(TLS13_CHACHA20_POLY1305_SHA256), + ); + assert!(server_result.unwrap().is_empty()); + client_result.unwrap(); +} + +#[test] +fn from_rustls_carries_client_certificate_authentication() { + use turnloop_tls::HostTime; + let server_cert = certificate(); + let client_cert = rcgen::generate_simple_self_signed(vec!["client.test".into()]).unwrap(); + let provider = Arc::new(rustls::crypto::ring::default_provider()); + let server = || { + let time = HostTime::new(NOW); + let mut roots = rustls::RootCertStore::empty(); + roots.add(client_cert.cert.der().clone()).unwrap(); + let verifier = rustls::server::WebPkiClientVerifier::builder_with_provider( + roots.into(), + provider.clone(), + ) + .build() + .unwrap(); + let config = + rustls::ServerConfig::builder_with_details(provider.clone(), time.time_provider()) + .with_safe_default_protocol_versions() + .unwrap() + .with_client_cert_verifier(verifier) + .with_single_cert( + vec![server_cert.cert.der().clone()], + rustls::pki_types::PrivatePkcs8KeyDer::from( + server_cert.key_pair.serialize_der(), + ) + .into(), + ) + .unwrap(); + turnloop_tls::ServerConfig::from_rustls(Arc::new(config), time) + }; + // Deliberately wrong until `process` supplies the host's time. + let time = HostTime::new(0); + let mut roots = rustls::RootCertStore::empty(); + roots.add(server_cert.cert.der().clone()).unwrap(); + let config = Arc::new( + rustls::ClientConfig::builder_with_details(provider.clone(), time.time_provider()) + .with_safe_default_protocol_versions() + .unwrap() + .with_root_certificates(roots) + .with_client_auth_cert( + vec![client_cert.cert.der().clone()], + rustls::pki_types::PrivatePkcs8KeyDer::from(client_cert.key_pair.serialize_der()) + .into(), + ) + .unwrap(), + ); + let client = ClientConfig::from_rustls(config.clone(), time.clone()); + assert!(Arc::ptr_eq(client.rustls_config(), &config)); + let (server_result, client_result) = handshake(client, server()); + client_result.unwrap(); + assert_eq!( + server_result.unwrap(), + vec![client_cert.cert.der().to_vec()] + ); + // The wrapped config validated the server certificate against the clock + // that `process` fed, not the 1970 value it started with. + assert_eq!(time.unix_seconds(), NOW); + + // The same server refuses a client that has no certificate to present. + let plain = ClientConfig::new( + ClientOptions { + ca: Some(vec![server_cert.cert.der().clone()]), + ..Default::default() + }, + NOW, + ) + .unwrap(); + let (server_result, _) = handshake(plain, server()); + let server_error = server_result.unwrap_err(); + assert!( + server_error.contains("peer sent no certificates"), + "{server_error}" + ); +} From 8022ea1fe3371e2fefed39ea98701419ffb69268 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Tue, 22 Sep 2026 19:41:38 +0200 Subject: [PATCH 2/3] Give the client pool stale-id recovery and one client-wide deadline Two gaps in client::Pool left every consumer improvising. A ConnectionId from Acquire::Reuse could not be checked or retired (#51). When the host's socket table and the pool disagreed, the only recovery was a fabricated release-then-close sequence, which errors for an id the pool already dropped. The pool now has `contains(id)` and `forget(id)`: forget drops the connection whatever its state (outstanding acquisitions included, so the per-host place is free for the next acquire), clears its idle deadline and returns whether the id was live; a stale id is reported, not an error. Both were added rather than one: contains answers the question the host asks first, forget is the well-defined recovery. ConnectionId now documents that ids are never reused, so a stale id cannot alias a newer connection. There was no client-wide next_timeout (#52). Pool::next_timeout rescanned every slot, and request lifecycles were not covered at all, so a host with N requests in flight had to min() N structures per turn; Perry's P6 client skipped per-phase deadlines (connect/headers/body) for exactly that reason. The new `client::Deadlines` is a keyed deadline heap: next_timeout is a peek, set and pop_expired are O(log n). Replaced deadlines are discarded lazily when they surface and the heap is rebuilt when they outnumber live entries, so its size stays within 2n+16 even for a body deadline refreshed on every chunk. The pool owns two of them: its idle deadlines (next_timeout and handle_timeout no longer scan slots) and host-registered request deadlines (`set_request_deadline(RequestId, lifecycle.next_timeout())`, drained with `handle_request_timeout`). Pool::next_timeout is the minimum of both, so one call arms the host's timer for every connection and request. The existing handle_timeout signature and behaviour are unchanged; a heap in the pool was chosen over a timer wheel because deadlines are arbitrary Instants the host supplies and no tick granularity is imposed. Evidence (tests/client.rs, client.rs unit test): - a Reuse id the host forgets is no longer contained, a second forget and a release report it stale, the per-host place goes to a new id, and a forgotten idle connection's deadline never fires; - with 50 idle connections and 100 registered request lifecycles one next_timeout reports the earliest deadline, follows a request moving to 10s and back out when its phase drops the deadline, expires the 50 idle connections and then exactly the two requests due, each yielding UND_ERR_HEADERS_TIMEOUT from its Lifecycle; - an Http1Connection's 100-continue and headers deadlines feed the pool; - 100 000 refreshes of one key keep the heap within bound and the refreshed key still fires last. Closes #51 Closes #52 --- protocols/turnloop-http/src/client.rs | 227 ++++++++++++++++++++++-- protocols/turnloop-http/tests/client.rs | 179 +++++++++++++++++++ 2 files changed, 393 insertions(+), 13 deletions(-) create mode 100644 protocols/turnloop-http/tests/client.rs diff --git a/protocols/turnloop-http/src/client.rs b/protocols/turnloop-http/src/client.rs index 92db38f..262bf99 100644 --- a/protocols/turnloop-http/src/client.rs +++ b/protocols/turnloop-http/src/client.rs @@ -5,6 +5,9 @@ use crate::{ http1::{Head, Header}, }; use std::{ + cmp::{Ordering, Reverse}, + collections::{BinaryHeap, HashMap}, + hash::Hash, net::IpAddr, time::{Duration, Instant}, }; @@ -264,8 +267,129 @@ impl PoolKey { } } } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +/// Issued by [`Pool::acquire`]. Ids are never reused, so a stale id can never +/// name a newer connection. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub struct ConnectionId(pub u64); +/// A host-chosen key for one in-flight request's deadline in [`Pool`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct RequestId(pub u64); + +/// A deadline heap keyed by `K`, holding at most one deadline per key. +/// +/// [`next_timeout`](Self::next_timeout) is O(1) and +/// [`set`](Self::set)/[`pop_expired`](Self::pop_expired) are O(log n), so a host +/// with many connections or requests never rescans them to arm its timer. +/// Replaced deadlines stay in the heap until they surface and are discarded; +/// the heap is rebuilt whenever they outnumber the live ones, so its size stays +/// within a constant factor of [`len`](Self::len). +pub struct Deadlines { + heap: BinaryHeap>>, + live: HashMap, + sequence: u64, +} +struct Entry { + at: Instant, + sequence: u64, + key: K, +} +impl PartialEq for Entry { + fn eq(&self, other: &Self) -> bool { + self.cmp(other) == Ordering::Equal + } +} +impl Eq for Entry {} +impl PartialOrd for Entry { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} +impl Ord for Entry { + // Sequence numbers are unique, so equal deadlines fire in the order set. + fn cmp(&self, other: &Self) -> Ordering { + (self.at, self.sequence).cmp(&(other.at, other.sequence)) + } +} +impl Default for Deadlines { + fn default() -> Self { + Self::new() + } +} +impl Deadlines { + pub fn new() -> Self { + Self { + heap: BinaryHeap::new(), + live: HashMap::new(), + sequence: 0, + } + } + /// Replace `key`'s deadline; `None` removes it. + pub fn set(&mut self, key: K, deadline: Option) { + match deadline { + Some(at) if self.live.get(&key).is_some_and(|&(d, _)| d == at) => return, + Some(at) => { + self.sequence += 1; + self.live.insert(key, (at, self.sequence)); + self.heap.push(Reverse(Entry { + at, + sequence: self.sequence, + key, + })); + } + None => { + if self.live.remove(&key).is_none() { + return; + } + } + } + self.settle(); + } + pub fn get(&self, key: K) -> Option { + self.live.get(&key).map(|&(at, _)| at) + } + /// The earliest live deadline. + pub fn next_timeout(&self) -> Option { + self.heap.peek().map(|Reverse(entry)| entry.at) + } + /// Remove and return one key whose deadline is at or before `now`, earliest + /// first. Call repeatedly until `None`. + pub fn pop_expired(&mut self, now: Instant) -> Option { + if self.next_timeout()? > now { + return None; + } + let Reverse(entry) = self.heap.pop()?; + self.live.remove(&entry.key); + self.settle(); + Some(entry.key) + } + pub fn len(&self) -> usize { + self.live.len() + } + pub fn is_empty(&self) -> bool { + self.live.is_empty() + } + /// Keep the top of the heap live, so `next_timeout` can simply peek. + fn settle(&mut self) { + if self.heap.len() > 2 * self.live.len() + 16 { + self.heap = self + .live + .iter() + .map(|(&key, &(at, sequence))| Reverse(Entry { at, sequence, key })) + .collect(); + return; + } + while let Some(Reverse(top)) = self.heap.peek() { + if self.live.get(&top.key) == Some(&(top.at, top.sequence)) { + break; + } + self.heap.pop(); + } + } + #[cfg(test)] + fn heap_len(&self) -> usize { + self.heap.len() + } +} #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Acquire { Connect(ConnectionId), @@ -281,11 +405,18 @@ struct Slot { deadline: Option, closed: bool, } +/// Connection reuse plus every client-wide deadline: idle connections and, when +/// the host registers them, in-flight requests. One [`next_timeout`] covers all +/// of them without visiting each connection or request. +/// +/// [`next_timeout`]: Pool::next_timeout pub struct Pool { slots: Vec, max_per_host: usize, idle: Duration, next: u64, + idle_deadlines: Deadlines, + request_deadlines: Deadlines, } impl Pool { pub fn new(max_per_host: usize, idle: Duration) -> Self { @@ -294,6 +425,8 @@ impl Pool { max_per_host, idle, next: 1, + idle_deadlines: Deadlines::new(), + request_deadlines: Deadlines::new(), } } /// A Connect reservation counts immediately, so concurrent callers cannot overbook. @@ -307,7 +440,9 @@ impl Pool { }) { s.active += 1; s.deadline = None; - return Acquire::Reuse(s.id); + let id = s.id; + self.idle_deadlines.set(id, None); + return Acquire::Reuse(id); } if self .slots @@ -368,29 +503,63 @@ impl Pool { s.capacity = 0; } if s.active == 0 { - s.deadline = Some(if s.capacity == 0 { now } else { now + idle }); + let deadline = if s.capacity == 0 { now } else { now + idle }; + s.deadline = Some(deadline); + self.idle_deadlines.set(id, Some(deadline)); } Ok(()) } pub fn closed(&mut self, id: ConnectionId) -> Result<()> { self.slot(id)?.closed = true; + self.idle_deadlines.set(id, None); Ok(()) } + /// Whether `id` still names a live connection: acquired and neither closed, + /// forgotten nor expired by [`handle_timeout`](Self::handle_timeout). + pub fn contains(&self, id: ConnectionId) -> bool { + self.slots.iter().any(|s| s.id == id && !s.closed) + } + /// Drop `id` whatever its state, for a host whose own socket table no longer + /// has it (closed without telling the pool, or never found). Outstanding + /// acquisitions go with it, so its per-host place is free for the next + /// `acquire`. Returns whether `id` was live; a stale id is not an error. + pub fn forget(&mut self, id: ConnectionId) -> bool { + let live = self.slot(id).map(|s| s.closed = true).is_ok(); + self.idle_deadlines.set(id, None); + live + } + /// The earliest idle-connection or registered request deadline, in O(1). pub fn next_timeout(&self) -> Option { - self.slots - .iter() - .filter(|s| !s.closed) - .filter_map(|s| s.deadline) + self.idle_deadlines + .next_timeout() + .into_iter() + .chain(self.request_deadlines.next_timeout()) .min() } /// Repeatedly call to obtain all Close requests at this time. No clock or timer is owned. pub fn handle_timeout(&mut self, now: Instant) -> Option { - let s = self - .slots - .iter_mut() - .find(|s| !s.closed && s.deadline.is_some_and(|d| d <= now))?; - s.closed = true; - Some(s.id) + while let Some(id) = self.idle_deadlines.pop_expired(now) { + if let Ok(s) = self.slot(id) { + s.closed = true; + return Some(id); + } + } + None + } + /// Register (or with `None` clear) a request's next deadline, typically the + /// `next_timeout()` of its [`Lifecycle`] or [`Http1Connection`] after each + /// call that may move it. The pool never inspects the request itself. + pub fn set_request_deadline(&mut self, id: RequestId, deadline: Option) { + self.request_deadlines.set(id, deadline); + } + pub fn request_deadline(&self, id: RequestId) -> Option { + self.request_deadlines.get(id) + } + /// Repeatedly call to obtain every request whose registered deadline is at or + /// before `now`. Its registration is removed: call that request's + /// `handle_timeout(now)` and register its new `next_timeout()`, if any. + pub fn handle_request_timeout(&mut self, now: Instant) -> Option { + self.request_deadlines.pop_expired(now) } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -778,3 +947,35 @@ fn proxy_authorization(proxy: &Url) -> Option { base64::engine::general_purpose::STANDARD.encode(credential) )) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn replaced_deadlines_do_not_accumulate() { + let now = Instant::now(); + let mut deadlines = Deadlines::new(); + for key in 0..100u32 { + deadlines.set(key, Some(now + Duration::from_secs(1000))); + } + // A request refreshing its body deadline on every chunk, far from the top. + for step in 0..100_000u64 { + deadlines.set(7, Some(now + Duration::from_secs(2000 + step))); + assert!(deadlines.heap_len() <= 2 * deadlines.len() + 17); + } + assert_eq!(deadlines.len(), 100); + assert_eq!( + deadlines.next_timeout(), + Some(now + Duration::from_secs(1000)) + ); + let late = now + Duration::from_secs(1_000_000); + let mut popped = 0; + while let Some(key) = deadlines.pop_expired(late) { + assert_eq!(key == 7, popped == 99, "the refreshed key fires last"); + popped += 1; + } + assert_eq!(popped, 100); + assert_eq!(deadlines.heap_len(), 0); + } +} diff --git a/protocols/turnloop-http/tests/client.rs b/protocols/turnloop-http/tests/client.rs new file mode 100644 index 0000000..17ead05 --- /dev/null +++ b/protocols/turnloop-http/tests/client.rs @@ -0,0 +1,179 @@ +use std::time::{Duration, Instant}; +use turnloop_http::{ + client::{ + Acquire, Completion, ConnectionId, Deadlines, Http1Connection, Lifecycle, Phase, Pool, + PoolKey, Protocol, Request, RequestId, + }, + http1::{BodyLength, Header, Limits}, +}; + +fn key() -> PoolKey { + PoolKey::new(&Request::new("http://a.test/", "GET").unwrap().url, None) +} + +#[test] +fn a_stale_reuse_id_can_be_checked_and_forgotten() { + let now = Instant::now(); + let key = key(); + let mut pool = Pool::new(1, Duration::from_secs(5)); + let Acquire::Connect(id) = pool.acquire(&key, now) else { + panic!("first acquire connects") + }; + assert!(pool.contains(id)); + pool.connected(id, Protocol::Http1, 1).unwrap(); + pool.release(id, true, now).unwrap(); + assert_eq!(pool.acquire(&key, now), Acquire::Reuse(id)); + // The host looks the id up and finds no socket: the pool is told directly. + assert!(pool.forget(id)); + assert!(!pool.contains(id)); + assert!(!pool.forget(id), "a stale id is reported, not an error"); + assert_eq!( + pool.release(id, false, now).unwrap_err().code, + "UND_ERR_CLOSED" + ); + // The single per-host place is free again, under a new id. + let Acquire::Connect(next) = pool.acquire(&key, now) else { + panic!("forget frees the per-host place") + }; + assert_ne!(next, id); + assert!(pool.contains(next) && !pool.contains(id)); + // A forgotten idle connection's deadline does not fire later. + pool.connected(next, Protocol::Http1, 1).unwrap(); + pool.release(next, true, now).unwrap(); + assert_eq!(pool.next_timeout(), Some(now + Duration::from_secs(5))); + assert!(pool.forget(next)); + assert_eq!(pool.next_timeout(), None); + assert_eq!(pool.handle_timeout(now + Duration::from_secs(60)), None); + assert!(!pool.contains(ConnectionId(999))); + assert!(!pool.forget(ConnectionId(999))); +} + +#[test] +fn contains_follows_close_and_idle_expiry() { + let now = Instant::now(); + let key = key(); + let mut pool = Pool::new(2, Duration::from_secs(1)); + let Acquire::Connect(a) = pool.acquire(&key, now) else { + panic!() + }; + let Acquire::Connect(b) = pool.acquire(&key, now) else { + panic!() + }; + pool.closed(a).unwrap(); + assert!(!pool.contains(a) && pool.contains(b)); + pool.connected(b, Protocol::Http1, 1).unwrap(); + pool.release(b, true, now).unwrap(); + assert_eq!(pool.handle_timeout(now + Duration::from_secs(1)), Some(b)); + assert!(!pool.contains(b)); +} + +#[test] +fn one_next_timeout_covers_idle_connections_and_request_lifecycles() { + let now = Instant::now(); + let at = |s| now + Duration::from_secs(s); + let mut pool = Pool::new(64, Duration::from_secs(30)); + // 50 origins with one idle connection each: idle deadlines at now + 30s. + let mut idle = Vec::new(); + for n in 0..50 { + let url = Request::new(&format!("http://h{n}.test/"), "GET") + .unwrap() + .url; + let Acquire::Connect(id) = pool.acquire(&PoolKey::new(&url, None), now) else { + panic!() + }; + pool.connected(id, Protocol::Http1, 1).unwrap(); + pool.release(id, true, now).unwrap(); + idle.push(id); + } + assert_eq!(pool.next_timeout(), Some(at(30))); + // 100 in-flight requests whose lifecycles the host registers. + let mut requests: Vec = (0..100).map(|_| Lifecycle::default()).collect(); + for (n, lifecycle) in requests.iter_mut().enumerate() { + lifecycle.transition(Phase::Headers, Some(at(40 + n as u64))); + pool.set_request_deadline(RequestId(n as u64), lifecycle.next_timeout()); + } + requests[63].transition(Phase::Headers, Some(at(10))); + pool.set_request_deadline(RequestId(63), requests[63].next_timeout()); + assert_eq!(pool.next_timeout(), Some(at(10))); + assert_eq!(pool.request_deadline(RequestId(63)), Some(at(10))); + // Headers arrived: the body phase has no deadline, so the entry goes away. + requests[63].transition(Phase::Body, None); + pool.set_request_deadline(RequestId(63), requests[63].next_timeout()); + assert_eq!(pool.next_timeout(), Some(at(30))); + assert_eq!(pool.handle_request_timeout(at(35)), None); + // Idle connections expire first, then requests in deadline order. + let mut closed = Vec::new(); + while let Some(id) = pool.handle_timeout(at(41)) { + closed.push(id); + } + assert_eq!(closed.len(), 50); + assert!(idle.iter().all(|id| closed.contains(id))); + assert_eq!(pool.next_timeout(), Some(at(40))); + let mut fired = Vec::new(); + while let Some(RequestId(n)) = pool.handle_request_timeout(at(41)) { + let lifecycle = &mut requests[n as usize]; + lifecycle.handle_timeout(at(41)); + let Some(Completion::Error(error)) = lifecycle.poll() else { + panic!("request {n} should have timed out") + }; + assert_eq!(error.code, "UND_ERR_HEADERS_TIMEOUT"); + pool.set_request_deadline(RequestId(n), lifecycle.next_timeout()); + fired.push(n); + } + assert_eq!(fired, [0, 1]); + assert_eq!(pool.next_timeout(), Some(at(42))); +} + +#[test] +fn http1_connection_deadlines_feed_the_pool() { + let now = Instant::now(); + let request = Request::new("http://a.test/upload", "POST").unwrap(); + let mut head = request.head(false); + head.headers.push(Header::new("expect", "100-continue")); + let mut connection = Http1Connection::new(Limits::default()); + connection + .start( + &head, + BodyLength::Known(4), + Some(now + Duration::from_secs(10)), + Some(now + Duration::from_secs(1)), + ) + .unwrap(); + let mut pool = Pool::new(1, Duration::from_secs(30)); + pool.set_request_deadline(RequestId(1), connection.next_timeout()); + assert_eq!(pool.next_timeout(), Some(now + Duration::from_secs(1))); + assert_eq!( + pool.handle_request_timeout(now + Duration::from_secs(1)), + Some(RequestId(1)) + ); + connection.handle_timeout(now + Duration::from_secs(1)); + assert!(connection.can_send_body()); + pool.set_request_deadline(RequestId(1), connection.next_timeout()); + assert_eq!(pool.next_timeout(), Some(now + Duration::from_secs(10))); +} + +#[test] +fn deadlines_fire_in_order_and_ignore_replaced_values() { + let now = Instant::now(); + let mut deadlines = Deadlines::new(); + assert_eq!(deadlines.next_timeout(), None); + deadlines.set('a', Some(now + Duration::from_secs(3))); + deadlines.set('b', Some(now + Duration::from_secs(1))); + deadlines.set('c', Some(now + Duration::from_secs(2))); + deadlines.set('b', Some(now + Duration::from_secs(5))); + deadlines.set('c', None); + assert_eq!(deadlines.len(), 2); + assert_eq!(deadlines.get('c'), None); + assert_eq!(deadlines.next_timeout(), Some(now + Duration::from_secs(3))); + assert_eq!(deadlines.pop_expired(now + Duration::from_secs(2)), None); + assert_eq!( + deadlines.pop_expired(now + Duration::from_secs(9)), + Some('a') + ); + assert_eq!( + deadlines.pop_expired(now + Duration::from_secs(9)), + Some('b') + ); + assert_eq!(deadlines.pop_expired(now + Duration::from_secs(9)), None); + assert!(deadlines.is_empty()); +} From 1c4095cee369e57298ffadf0ecf59a668959771b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Tue, 22 Sep 2026 19:41:49 +0200 Subject: [PATCH 3/3] Add an in-memory multipart/form-data encoder turnloop_http had no multipart builder, which blocked Perry's CLI migration off reqwest: `perry publish`, `audit`, `verify` and `run --remote` all send multipart bodies, and P11 had to carry its own 280-line builder. `multipart::Form` holds ordered text and file parts (`Part::text`, `Part::file`, optional `content_type`; files default to application/octet-stream) and `encode(entropy)` returns the body plus its boundary; `Encoded::content_type()` gives the header value and `apply` installs both on a client::Request. Names and filenames are escaped as browsers do (`"`, CR, LF percent-encoded) and a content type with a control character is rejected, so no field can inject a header line. The boundary is verified, not just random. A base64 field can hold a boundary-shaped run, and a collision silently truncates the upload, so every candidate is checked against each part's bytes, name, filename and content type, and the next candidate is tried until none contains it. Entropy comes from the host as 16 bytes: the crate reads no randomness source (sans-I/O), and its dependencies offer none once ring is optional. Candidates are SplitMix64 outputs over distinct inputs and therefore pairwise distinct, so the search ends within (total part bytes + 1) tries. The format is "turnloop-" plus 32 hex digits, 41 characters with nothing to quote. In-memory encoding matches every current Perry caller (only Part::text is used; no streamed parts). Evidence (tests/multipart.rs): a four-part form (escaped name, CRLF in a value, binary file, typed file) encodes byte-for-byte to the expected body, deterministically per entropy; planting the first two candidates in a base64-looking field and a filename makes the encoder pick a third boundary whose delimiter occurs exactly three times, two openings and one closing; `apply` leaves exactly one content-type header; a CRLF content type is refused with UND_ERR_INVALID_ARG. Closes #78 --- protocols/turnloop-http/README.md | 3 +- protocols/turnloop-http/src/lib.rs | 1 + protocols/turnloop-http/src/multipart.rs | 261 +++++++++++++++++++++ protocols/turnloop-http/tests/multipart.rs | 107 +++++++++ 4 files changed, 371 insertions(+), 1 deletion(-) create mode 100644 protocols/turnloop-http/src/multipart.rs create mode 100644 protocols/turnloop-http/tests/multipart.rs diff --git a/protocols/turnloop-http/README.md b/protocols/turnloop-http/README.md index ac5f54a..dcb6bf7 100644 --- a/protocols/turnloop-http/README.md +++ b/protocols/turnloop-http/README.md @@ -6,7 +6,8 @@ Own HTTP/1.1 and HTTP/2 wire engines, with no runtime or transport dependency. - `client::Http1Connection` adds non-pipelined requests, streamed uploads, `100-continue`, output acknowledgements, deadline/abort state, and one terminal completion per accepted request. Take the completion before reusing the connection. - `http2::Connection` handles both roles. Flush and acknowledge `output`, retain incomplete input, consume one frame/event at a time. `send_data` can accept only a prefix or zero when flow-controlled; retain the remainder. Return DATA capacity with `release_capacity` after the application consumes it. Queue application responses in the host during stalls. On transport loss call `eof`, then drain `poll_failed_stream`. - `hpack` provides the independent bounded RFC 7541 codec. The encoder never indexes credentials/cookies. Decode failures poison a context. -- `client::Pool` reserves connections before DNS/connect so parallel commands respect per-origin/proxy limits. `Route` emits resolution/TLS requests and builds proxy CONNECT or absolute-form heads. `Resolver` belongs to the host. `Request::redirect` applies Fetch redirects; use `DEFAULT_MAX_REDIRECTS` (20). JS conversions and promise delivery remain with Perry. +- `client::Pool` reserves connections before DNS/connect so parallel commands respect per-origin/proxy limits. `Route` emits resolution/TLS requests and builds proxy CONNECT or absolute-form heads. `Resolver` belongs to the host. `Request::redirect` applies Fetch redirects; use `DEFAULT_MAX_REDIRECTS` (20). JS conversions and promise delivery remain with Perry. `Pool::contains`/`forget` check or drop a stale `ConnectionId`. `Pool::next_timeout` is one O(1) deadline over idle connections and every request registered with `set_request_deadline`; drain `handle_timeout` and `handle_request_timeout`. `client::Deadlines` is the same heap for host-keyed timers. +- `multipart::Form` encodes in-memory `multipart/form-data` text and file parts. The host supplies 16 bytes of entropy; the boundary is regenerated until it occurs in no part. - `compression::StreamingDecoder` accepts input and caller-owned output. Reuse it with `reset` to retain scratch buffers across bodies. Hosts may cache one per content encoding. `decode` is the convenience whole-body path and constructs algorithm state each time. No engine samples a clock. The host passes `Instant` deadlines and invokes timeout handlers. Hold output storage stable until a completion-shaped write finishes: do not mutate the engine while an I/O operation borrows its output. Error codes are transport causes; Perry creates the JS error objects and detailed OS diagnostics. diff --git a/protocols/turnloop-http/src/lib.rs b/protocols/turnloop-http/src/lib.rs index 9fec9e3..90df02b 100644 --- a/protocols/turnloop-http/src/lib.rs +++ b/protocols/turnloop-http/src/lib.rs @@ -55,6 +55,7 @@ pub mod compression; pub mod hpack; pub mod http1; pub mod http2; +pub mod multipart; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct Error { diff --git a/protocols/turnloop-http/src/multipart.rs b/protocols/turnloop-http/src/multipart.rs new file mode 100644 index 0000000..115f6d8 --- /dev/null +++ b/protocols/turnloop-http/src/multipart.rs @@ -0,0 +1,261 @@ +//! In-memory `multipart/form-data` bodies (RFC 7578 over RFC 2046 framing). +//! +//! The boundary is derived from 16 bytes of entropy the host supplies (this crate +//! reads no randomness source) and is then checked against every part: a +//! boundary that occurs anywhere in a part's name, filename, content type or +//! bytes is replaced by the next candidate, so no payload - base64 text with a +//! boundary-shaped run included - can end a part early. +//! +//! ``` +//! use turnloop_http::multipart::{Form, Part}; +//! let form = Form::new() +//! .text("version", "1.2.0") +//! .part(Part::file("tarball", "pkg.tgz", b"\x1f\x8b...".to_vec()).content_type("application/gzip")); +//! let encoded = form.encode([7; 16]).unwrap(); +//! assert!(encoded.content_type().starts_with("multipart/form-data; boundary=")); +//! ``` +use crate::{Error, Result, client::Request, http1::Header}; + +/// One form field: a text value or a named file. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Part { + name: String, + filename: Option, + content_type: Option, + body: Vec, +} +impl Part { + /// A text field. No `Content-Type` is written unless one is set. + pub fn text(name: impl Into, value: impl Into) -> Self { + Self { + name: name.into(), + filename: None, + content_type: None, + body: value.into().into_bytes(), + } + } + /// A file field. Its `Content-Type` is `application/octet-stream` unless set. + pub fn file(name: impl Into, filename: impl Into, bytes: Vec) -> Self { + Self { + name: name.into(), + filename: Some(filename.into()), + content_type: None, + body: bytes, + } + } + pub fn content_type(mut self, content_type: impl Into) -> Self { + self.content_type = Some(content_type.into()); + self + } + fn media_type(&self) -> Option<&str> { + match (&self.content_type, &self.filename) { + (Some(content_type), _) => Some(content_type), + (None, Some(_)) => Some("application/octet-stream"), + (None, None) => None, + } + } +} + +/// An ordered list of parts, encoded in one piece by [`Form::encode`]. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct Form { + parts: Vec, +} +impl Form { + pub fn new() -> Self { + Self::default() + } + pub fn text(self, name: impl Into, value: impl Into) -> Self { + self.part(Part::text(name, value)) + } + pub fn file( + self, + name: impl Into, + filename: impl Into, + bytes: Vec, + ) -> Self { + self.part(Part::file(name, filename, bytes)) + } + pub fn part(mut self, part: Part) -> Self { + self.parts.push(part); + self + } + pub fn parts(&self) -> &[Part] { + &self.parts + } + /// Serialize the form. `entropy` should come from the host's secure random + /// source; any value yields a boundary absent from every part, but only + /// unpredictable entropy keeps the boundary unguessable. + /// + /// A content type containing a control character (CR/LF included) is + /// rejected. Quotes, CR and LF in names and filenames are percent-encoded as + /// browsers do; other bytes are written as given. + pub fn encode(&self, entropy: [u8; 16]) -> Result { + for part in &self.parts { + if part + .media_type() + .is_some_and(|t| t.bytes().any(|b| b.is_ascii_control())) + { + return Err(Error::new( + "UND_ERR_INVALID_ARG", + "invalid multipart content type", + )); + } + } + let names: Vec<_> = self + .parts + .iter() + .map(|p| (escape(&p.name), p.filename.as_deref().map(escape))) + .collect(); + let boundary = self.boundary(entropy, &names); + // Framing per part is under 128 bytes beyond the boundary, name, + // filename and content type; the closing delimiter adds 6. + let mut body = Vec::with_capacity( + self.parts + .iter() + .zip(&names) + .map(|(part, (name, filename))| { + part.body.len() + + name.len() + + filename.as_ref().map_or(0, String::len) + + part.media_type().map_or(0, str::len) + + boundary.len() + + 128 + }) + .sum::() + + boundary.len() + + 6, + ); + for (part, (name, filename)) in self.parts.iter().zip(&names) { + body.extend_from_slice(b"--"); + body.extend_from_slice(boundary.as_bytes()); + body.extend_from_slice(b"\r\nContent-Disposition: form-data; name=\""); + body.extend_from_slice(name.as_bytes()); + body.push(b'"'); + if let Some(filename) = filename { + body.extend_from_slice(b"; filename=\""); + body.extend_from_slice(filename.as_bytes()); + body.push(b'"'); + } + if let Some(media_type) = part.media_type() { + body.extend_from_slice(b"\r\nContent-Type: "); + body.extend_from_slice(media_type.as_bytes()); + } + body.extend_from_slice(b"\r\n\r\n"); + body.extend_from_slice(&part.body); + body.extend_from_slice(b"\r\n"); + } + body.extend_from_slice(b"--"); + body.extend_from_slice(boundary.as_bytes()); + body.extend_from_slice(b"--\r\n"); + Ok(Encoded { boundary, body }) + } + /// The first candidate that occurs in no part. Candidates are pairwise + /// distinct and each one found in a part matches at least one of that + /// part's finitely many substrings, so the search ends within + /// (total part bytes + 1) candidates. + fn boundary(&self, entropy: [u8; 16], names: &[(String, Option)]) -> String { + let entropy = u128::from_le_bytes(entropy); + let (low, high) = (entropy as u64, (entropy >> 64) as u64); + let mut index = 0u64; + loop { + let boundary = candidate(low, high, index); + let needle = boundary.as_bytes(); + let collides = self + .parts + .iter() + .zip(names) + .any(|(part, (name, filename))| { + [ + Some(part.body.as_slice()), + Some(name.as_bytes()), + filename.as_deref().map(str::as_bytes), + part.media_type().map(str::as_bytes), + ] + .into_iter() + .flatten() + .any(|bytes| contains(bytes, needle)) + }); + if !collides { + return boundary; + } + index += 1; + } + } +} + +/// An encoded form body and the boundary its `Content-Type` must name. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Encoded { + pub boundary: String, + pub body: Vec, +} +impl Encoded { + /// The `Content-Type` header value for this body. + pub fn content_type(&self) -> String { + format!("multipart/form-data; boundary={}", self.boundary) + } + /// Make this the request body, replacing any `Content-Type` header. + pub fn apply(self, request: &mut Request) { + let content_type = self.content_type(); + request + .headers + .retain(|h| !h.name.eq_ignore_ascii_case("content-type")); + request + .headers + .push(Header::new("content-type", content_type)); + request.body = self.body; + } +} + +/// `"turnloop-"` and 32 hex digits: 41 characters, within RFC 2046's 70, and +/// no character that needs quoting in the `Content-Type` parameter. +const PREFIX: &str = "turnloop-"; +const BOUNDARY_LEN: usize = PREFIX.len() + 32; + +fn candidate(low: u64, high: u64, index: u64) -> String { + // SplitMix64: its finalizer is a bijection and the odd increment makes each + // index's input distinct, so every index yields a distinct first half. + const GAMMA: u64 = 0x9e37_79b9_7f4a_7c15; + let first = mix(low.wrapping_add(index.wrapping_mul(GAMMA))); + let second = mix(high ^ first); + let mut boundary = String::with_capacity(BOUNDARY_LEN); + boundary.push_str(PREFIX); + for word in [first, second] { + for shift in (0..16).rev() { + let digit = ((word >> (shift * 4)) & 15) as u8; + boundary.push(char::from(if digit < 10 { + b'0' + digit + } else { + b'a' + digit - 10 + })); + } + } + boundary +} + +fn mix(mut z: u64) -> u64 { + z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9); + z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb); + z ^ (z >> 31) +} + +fn contains(haystack: &[u8], needle: &[u8]) -> bool { + haystack + .windows(needle.len()) + .any(|window| window == needle) +} + +/// The WHATWG form-data escaping of a name or filename. +fn escape(value: &str) -> String { + let mut escaped = String::with_capacity(value.len()); + for c in value.chars() { + match c { + '"' => escaped.push_str("%22"), + '\r' => escaped.push_str("%0D"), + '\n' => escaped.push_str("%0A"), + c => escaped.push(c), + } + } + escaped +} diff --git a/protocols/turnloop-http/tests/multipart.rs b/protocols/turnloop-http/tests/multipart.rs new file mode 100644 index 0000000..12705f9 --- /dev/null +++ b/protocols/turnloop-http/tests/multipart.rs @@ -0,0 +1,107 @@ +use turnloop_http::{ + client::Request, + http1::Header, + multipart::{Form, Part}, +}; + +fn count(haystack: &[u8], needle: &[u8]) -> usize { + haystack + .windows(needle.len()) + .filter(|w| *w == needle) + .count() +} + +#[test] +fn multipart_body_is_framed_byte_for_byte() { + let form = Form::new() + .text("version", "1.2.0") + .text("say \"hi\"\r\n", "line1\r\nline2") + .file("tarball", "pkg.tgz", vec![0x1f, 0x8b, 0, 255]) + .part(Part::file("manifest", "p.json", b"{}".to_vec()).content_type("application/json")); + let encoded = form.encode([0x5a; 16]).unwrap(); + let b = &encoded.boundary; + assert!(b.starts_with("turnloop-") && b.len() == 41); + assert!(b[9..].bytes().all(|c| c.is_ascii_hexdigit())); + assert_eq!( + encoded.content_type(), + format!("multipart/form-data; boundary={b}") + ); + let mut expected = format!( + "--{b}\r\nContent-Disposition: form-data; name=\"version\"\r\n\r\n1.2.0\r\n\ + --{b}\r\nContent-Disposition: form-data; name=\"say %22hi%22%0D%0A\"\r\n\r\n\ + line1\r\nline2\r\n\ + --{b}\r\nContent-Disposition: form-data; name=\"tarball\"; filename=\"pkg.tgz\"\r\n\ + Content-Type: application/octet-stream\r\n\r\n" + ) + .into_bytes(); + expected.extend_from_slice(&[0x1f, 0x8b, 0, 255]); + expected.extend_from_slice( + format!( + "\r\n--{b}\r\nContent-Disposition: form-data; name=\"manifest\"; \ + filename=\"p.json\"\r\nContent-Type: application/json\r\n\r\n{{}}\r\n--{b}--\r\n" + ) + .as_bytes(), + ); + assert_eq!(encoded.body, expected); + // The same entropy gives the same body; other entropy another boundary. + assert_eq!(form.encode([0x5a; 16]).unwrap(), encoded); + assert_ne!(form.encode([0x5b; 16]).unwrap().boundary, *b); + + let mut request = Request::new("https://registry.test/publish", "POST").unwrap(); + request + .headers + .push(Header::new("Content-Type", "text/plain")); + let content_type = encoded.content_type(); + encoded.clone().apply(&mut request); + let types: Vec<_> = request + .headers + .iter() + .filter(|h| h.name == "content-type") + .map(|h| h.value.clone()) + .collect(); + assert_eq!(types, [content_type.into_bytes()]); + assert_eq!(request.body, encoded.body); +} + +#[test] +fn the_chosen_boundary_never_occurs_inside_a_part() { + let entropy = [0x42; 16]; + // With no parts nothing can collide, so this is the first candidate. + let first = Form::new().encode(entropy).unwrap().boundary; + let second = Form::new() + .file("f", "x", first.clone().into_bytes()) + .encode(entropy) + .unwrap() + .boundary; + assert_ne!(first, second); + // Plant both candidates in a base64-looking field and the first again in a + // filename: every candidate so far has to be skipped. + let payload = format!("QUJD{first}REVG{second}R0hJ"); + let form = Form::new().text("signature", payload.clone()).file( + "upload", + format!("{first}.tgz"), + b"data".to_vec(), + ); + let encoded = form.encode(entropy).unwrap(); + let chosen = encoded.boundary.as_bytes(); + assert_ne!(chosen, first.as_bytes()); + assert_ne!(chosen, second.as_bytes()); + let mut delimiter = b"--".to_vec(); + delimiter.extend_from_slice(chosen); + // One opening delimiter per part and the closing one, nothing else. + assert_eq!(count(&encoded.body, &delimiter), 3); + let mut closing = delimiter.clone(); + closing.extend_from_slice(b"--\r\n"); + assert!(encoded.body.ends_with(&closing)); + assert_eq!(count(&encoded.body, payload.as_bytes()), 1); + assert_eq!(count(&encoded.body, first.as_bytes()), 2); +} + +#[test] +fn multipart_rejects_header_injection_in_content_type() { + let error = Form::new() + .part(Part::text("a", "b").content_type("text/plain\r\nX-Evil: 1")) + .encode([0; 16]) + .unwrap_err(); + assert_eq!(error.code, "UND_ERR_INVALID_ARG"); +}