diff --git a/Cargo.lock b/Cargo.lock index dc2314e..d26117d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -91,10 +91,13 @@ dependencies = [ "axum", "eventsource-stream", "futures", + "h2", + "rcgen", "reqwest", "serde_json", "thiserror", "tokio", + "tokio-rustls", "tower", "tower-http 0.7.0", "tracing", @@ -268,6 +271,45 @@ version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" +[[package]] +name = "asn1-rs" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7f43a50ac4fdca5df8e885c21b835997f0a1cdee65494a6847694a98652d9d8" +dependencies = [ + "asn1-rs-derive", + "asn1-rs-impl", + "displaydoc", + "nom", + "num-traits", + "rusticata-macros", + "thiserror", + "time", +] + +[[package]] +name = "asn1-rs-derive" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3109e49b1e4909e9db6515a30c633684d68cdeaa252f215214cb4fa1a5bfee2c" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "synstructure", +] + +[[package]] +name = "asn1-rs-impl" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b18050c2cd6fe86c3a76584ef5e0baf286d038cda203eb6223df2cc413565f7" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "async-channel" version = "2.5.0" @@ -422,6 +464,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ce2b2dcc879c3bae0d371e77c99f2238400ef24ec001394befa67b6e543add9e" dependencies = [ "aws-lc-sys", + "untrusted 0.7.1", "zeroize", ] @@ -511,6 +554,15 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "bit-vec" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b71798fca2c1fe1086445a7258a4bc81e6e49dcd24c8d0dd9a1e57395b603f51" +dependencies = [ + "serde", +] + [[package]] name = "bitflags" version = "1.3.2" @@ -843,6 +895,20 @@ dependencies = [ "thiserror", ] +[[package]] +name = "der-parser" +version = "10.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07da5016415d5a3c4dd39b11ed26f915f52fc4e0dc197d87908bc916e51bc1a6" +dependencies = [ + "asn1-rs", + "displaydoc", + "nom", + "num-bigint", + "num-traits", + "rusticata-macros", +] + [[package]] name = "deranged" version = "0.5.8" @@ -1000,6 +1066,12 @@ version = "0.5.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99" +[[package]] +name = "fnv" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" + [[package]] name = "form_urlencoded" version = "1.2.2" @@ -1178,6 +1250,25 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "h2" +version = "0.4.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef8e5e5a340588f4452631496976cf8636d4a7ecf600239fdc27615d2530bc16" +dependencies = [ + "atomic-waker", + "bytes", + "fnv", + "futures-core", + "futures-sink", + "http", + "indexmap 2.14.0", + "slab", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "hashbrown" version = "0.12.3" @@ -1272,6 +1363,7 @@ dependencies = [ "bytes", "futures-channel", "futures-core", + "h2", "http", "http-body", "httparse", @@ -1749,12 +1841,31 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "num-bigint" +version = "0.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c89e69e7e0f03bea5ef08013795c25018e101932225a656383bd384495ecc367" +dependencies = [ + "num-integer", + "num-traits", +] + [[package]] name = "num-conv" version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" +[[package]] +name = "num-integer" +version = "0.1.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ce2d95d4b3734dc35aa2f45e1aa22cd416814592a4f9d9205e11affd5b8e10b" +dependencies = [ + "num-traits", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -1764,6 +1875,15 @@ dependencies = [ "autocfg", ] +[[package]] +name = "oid-registry" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12f40cff3dde1b6087cc5d5f5d4d65712f34016a03ed60e9c08dcc392736b5b7" +dependencies = [ + "asn1-rs", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -2062,6 +2182,19 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "rcgen" +version = "0.14.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8774e05a7d0de114588e6a28fe7e71694b82614ed569d86d8b389dfbc98b8ad8" +dependencies = [ + "aws-lc-rs", + "rustls-pki-types", + "time", + "x509-parser", + "yasna", +] + [[package]] name = "ref-cast" version = "1.0.26" @@ -2121,6 +2254,7 @@ dependencies = [ "bytes", "futures-core", "futures-util", + "h2", "http", "http-body", "http-body-util", @@ -2161,7 +2295,7 @@ dependencies = [ "cfg-if", "getrandom 0.2.17", "libc", - "untrusted", + "untrusted 0.9.0", "windows-sys 0.52.0", ] @@ -2220,6 +2354,15 @@ dependencies = [ "semver", ] +[[package]] +name = "rusticata-macros" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "faf0c4a6ece9950b9abdb62b1cfcf2a68b3b67a10ba445b3bb85be2a293d0632" +dependencies = [ + "nom", +] + [[package]] name = "rustix" version = "1.1.4" @@ -2305,7 +2448,7 @@ dependencies = [ "aws-lc-rs", "ring", "rustls-pki-types", - "untrusted", + "untrusted 0.9.0", ] [[package]] @@ -3069,6 +3212,12 @@ version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" +[[package]] +name = "untrusted" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a156c684c91ea7d62626509bce3cb4e1d9ed5c4d978f7b4352658f96a4c26b4a" + [[package]] name = "untrusted" version = "0.9.0" @@ -3484,6 +3633,34 @@ version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" +[[package]] +name = "x509-parser" +version = "0.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d43b0f71ce057da06bc0851b23ee24f3f86190b07203dd8f567d0b706a185202" +dependencies = [ + "asn1-rs", + "aws-lc-rs", + "data-encoding", + "der-parser", + "lazy_static", + "nom", + "oid-registry", + "rusticata-macros", + "thiserror", + "time", +] + +[[package]] +name = "yasna" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5f6765e852b9b4dc8e2a76843e4d64d1cea8e79bcde0b6901aea8e7c7f08282" +dependencies = [ + "bit-vec", + "time", +] + [[package]] name = "yoke" version = "0.8.3" diff --git a/Cargo.toml b/Cargo.toml index c5b8a21..2c46c02 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -94,6 +94,9 @@ syn = "3" # Testing expect-test = "1.5" regex = "1.12.2" +h2 = "0.4" +rcgen = { version = "0.14", default-features = false, features = ["crypto", "aws_lc_rs"] } +tokio-rustls = { version = "0.26", default-features = false, features = ["aws_lc_rs", "tls12"] } [workspace.lints.rust] future_incompatible = { level = "warn", priority = -1 } diff --git a/md/http-transport.md b/md/http-transport.md index c8584e7..acc349e 100644 --- a/md/http-transport.md +++ b/md/http-transport.md @@ -135,3 +135,101 @@ and it will open a single bidirectional connection instead of using POST + SSE: let transport = HttpClient::new("ws://127.0.0.1:8080")?; my_client().connect_to(transport).await?; ``` + +The client validates the WebSocket handshake before sending any queued ACP +messages. This transport does not negotiate WebSocket subprotocols or extensions, +including compression. Setting `Sec-WebSocket-Protocol` or +`Sec-WebSocket-Extensions` in custom default headers does not enable support for +them; a server response selecting either is rejected. + +### Client Configuration + +Use the client builder to configure headers, proxies, DNS, TLS, and timeouts. +The configuration is used for both HTTP/SSE requests and WebSocket handshakes: + +```rust +use std::time::Duration; +use agent_client_protocol_http::HttpClient; + +let transport = HttpClient::builder("wss://agent.example") + .configure_http(|http| { + http.user_agent("my-acp-client") + .connect_timeout(Duration::from_secs(5)) + .timeout(Duration::from_secs(10)) + }) + .build()?; +``` + +`configure_http` transforms the current `reqwest::ClientBuilder`; repeated calls +retain earlier configuration. Use it for options such as `default_headers`, +`proxy`, `resolve`, `tls_certs_merge`, and `identity`. It does not open a +connection. URL validation and HTTP-client construction errors are returned by +`build()`, before connecting the transport. + +`builder(base_url)` uses the same path rule as `new`: append `/acp` unless the +path already ends with it, after removing trailing slashes. For an exact +endpoint, including a custom server path or query string, use +`builder_with_endpoint(endpoint)` instead. `with_endpoint(endpoint)` remains +the unconfigured convenience constructor. + +For WebSocket URLs, the SDK applies HTTP/1.1 and disables redirects **after** +the configuration callback. These are transport requirements, not defaults a +callback can override. A redirect fails at the original endpoint without sending +handshake headers to its destination; resolve the intended WebSocket endpoint +before connecting. HTTP/SSE retains the caller's HTTP-version and redirect +settings. + +Reqwest's connection timeout covers connection establishment. Its request/read +timeouts also cover the opening WebSocket handshake but do not impose a lifetime +or idle timeout on the upgraded socket. For HTTP/SSE, request/read timeouts retain +their normal reqwest semantics, including the long-lived SSE response body. + +For custom trust roots and client certificates, prefer reqwest's TLS options. +If using `tls_backend_preconfigured`, configure that backend's ALPN for +HTTP/1.1 yourself: reqwest does not rewrite a preconfigured TLS backend's ALPN, +even when HTTP/1.1 is selected on its builder. For rustls, set +`ClientConfig::alpn_protocols` to `vec![b"http/1.1".to_vec()]`. Incompatible +negotiation is rejected before any ACP messages are sent; HTTP/2 and HTTP/3 +WebSocket negotiation is not implemented. + +### Reusing an HTTP Client + +To share an already-built reqwest client and its connection pool, use the +HTTP/SSE-only constructor with an **exact endpoint**, including `/acp` or the +server's custom path: + +```rust +let http = reqwest::Client::builder().build()?; +let transport = HttpClient::from_http_client( + "https://agent.example/acp", + http.clone(), +)?; +``` + +`from_http_client` rejects `ws://` and `wss://` URLs before any network request. +An already-built reqwest client cannot be reconfigured to enforce the +WebSocket connection policies. Use the SDK builder for those URLs. + +`HttpClient` itself can also be cloned to reuse its endpoint and underlying +HTTP client. Each connection has independent ACP transport state. + +### Migrating Custom Client Construction + +`HttpClient::with_client(base_url, client)` and +`with_endpoint_and_client(endpoint, client)` remain available but are deprecated. +Existing HTTP/SSE calls keep their path handling, supplied client, and connection +pool. They emit a deprecation warning so applications can migrate incrementally. + +For `ws://` and `wss://`, both deprecated constructors now return +`HttpClientError::WebSocketRequiresBuilder` before any network I/O. Those callers +must migrate to the builder: retaining the old signatures cannot make an opaque, +already-built reqwest client enforce the WebSocket connection policies. + +- For HTTP/SSE or WebSocket configuration, move the reqwest builder settings into + `HttpClient::builder(base_url).configure_http(|http| ...).build()`. Do not call + reqwest's `build()` inside the callback. +- For an exact endpoint, use `builder_with_endpoint(endpoint)` instead. +- To retain a shared, already-built HTTP/SSE client, use + `from_http_client(exact_endpoint, client)`. Unlike the old base-URL constructor, + this does not append `/acp`. +- Code using `new` or `with_endpoint` needs no changes. diff --git a/src/agent-client-protocol-http/CHANGELOG.md b/src/agent-client-protocol-http/CHANGELOG.md index c58d6bd..7e008c5 100644 --- a/src/agent-client-protocol-http/CHANGELOG.md +++ b/src/agent-client-protocol-http/CHANGELOG.md @@ -2,6 +2,29 @@ ## [Unreleased] +### Breaking changes + +- WebSocket URLs passed to `HttpClient::with_client` or + `with_endpoint_and_client` now return `WebSocketRequiresBuilder` before network + I/O. Migrate to `builder(...).configure_http(...).build()` or + `builder_with_endpoint(...)` so the transport can enforce connection policies. + See the + [migration guide](https://agentclientprotocol.github.io/rust-sdk/http-transport.html#migrating-custom-client-construction). + +### Deprecated + +- Retain `HttpClient::with_client` and `with_endpoint_and_client` as deprecated + HTTP/SSE compatibility wrappers with unchanged path handling. + `from_http_client(exact_endpoint, client)` preserves shared reqwest clients for + HTTP/SSE; its endpoint is exact and does not append `/acp`. + +### Fixed + +- Apply custom headers, TLS, proxies, DNS, and timeouts to WebSocket handshakes. + Enforce HTTP/1.1 and disable redirects for WebSockets without changing HTTP/SSE + policies. Validate the upgrade response before sending queued ACP data, and + reject unsupported subprotocols and extensions. + ## [2.0.0](https://github.com/agentclientprotocol/rust-sdk/compare/agent-client-protocol-http-v1.3.0...agent-client-protocol-http-v2.0.0) - 2026-07-23 ### Breaking changes diff --git a/src/agent-client-protocol-http/Cargo.toml b/src/agent-client-protocol-http/Cargo.toml index 799cc6c..2cba212 100644 --- a/src/agent-client-protocol-http/Cargo.toml +++ b/src/agent-client-protocol-http/Cargo.toml @@ -73,6 +73,11 @@ tokio = { workspace = true, features = ["macros", "net", "rt", "sync", "time"] } async-tungstenite.workspace = true tracing-subscriber.workspace = true tower.workspace = true +h2.workspace = true +rcgen.workspace = true +tokio-rustls.workspace = true +# Reproduce downstream feature unification, even in client-only test runs. +reqwest = { workspace = true, features = ["http2"] } [lints] workspace = true diff --git a/src/agent-client-protocol-http/README.md b/src/agent-client-protocol-http/README.md index b659472..8303790 100644 --- a/src/agent-client-protocol-http/README.md +++ b/src/agent-client-protocol-http/README.md @@ -3,7 +3,7 @@ HTTP/WebSocket transport for ACP agents. - **Server**: `AcpHttpServer` exposes agents over HTTP + SSE with optional WebSocket upgrade -- **Client**: `HttpClient` connects to remote agents over HTTP + SSE +- **Client**: `HttpClient` connects over HTTP + SSE or WebSocket, selected by the URL scheme The crate does not enable either transport side by default. Opt into the surface you need: @@ -18,4 +18,16 @@ with `CorsOptions::allow_origins(...)` to allow specific browser origins. Core SDK request cancellation support is forwarded through this transport. +Use `HttpClient::builder(url).configure_http(|http| ...).build()` to customize +reqwest headers, TLS, proxies, DNS, and timeouts. WebSocket handshakes use HTTP/1.1, +do not follow redirects, and are validated before sending ACP data. A raw +preconfigured TLS backend must itself use HTTP/1.1 ALPN. + +`HttpClient::from_http_client(exact_endpoint, client)` reuses an existing reqwest +client for HTTP/SSE only. The old `with_client` and `with_endpoint_and_client` +constructors remain as deprecated HTTP/SSE compatibility wrappers. WebSocket +calls to those constructors return a migration error; use the builder instead. + +See the [client configuration and migration guide](https://agentclientprotocol.github.io/rust-sdk/http-transport.html#client-configuration). + See the [documentation](https://docs.rs/agent-client-protocol-http) for usage examples. diff --git a/src/agent-client-protocol-http/src/client.rs b/src/agent-client-protocol-http/src/client.rs index 1c89987..8abf5d6 100644 --- a/src/agent-client-protocol-http/src/client.rs +++ b/src/agent-client-protocol-http/src/client.rs @@ -28,15 +28,36 @@ use crate::protocol::{ pub enum HttpClientError { #[error("invalid URL: {0}")] InvalidUrl(#[from] url::ParseError), + #[error("unsupported URL scheme: {0}; expected http, https, ws, or wss")] + UnsupportedScheme(String), + #[error( + "WebSocket URLs require HttpClient::builder or builder_with_endpoint; a prebuilt reqwest client cannot enforce WebSocket connection policies" + )] + WebSocketRequiresBuilder, #[error("failed to build HTTP client: {0}")] Reqwest(#[from] reqwest::Error), } +/// An endpoint-bound ACP transport using HTTP/SSE or WebSocket. +/// +/// Cloning shares the underlying HTTP connection pool. Each connection has +/// independent ACP transport state. +#[derive(Clone)] pub struct HttpClient { endpoint: url::Url, http: reqwest::Client, } +/// Configures an [`HttpClient`] before its underlying HTTP client is built. +/// +/// Use [`HttpClient::builder`] for a base URL or +/// [`HttpClient::builder_with_endpoint`] for an exact endpoint. +#[must_use = "the builder must be built to create an HTTP client"] +pub struct HttpClientBuilder { + endpoint: Result, + http: reqwest::ClientBuilder, +} + impl std::fmt::Debug for HttpClient { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("HttpClient") @@ -45,61 +66,201 @@ impl std::fmt::Debug for HttpClient { } } +impl std::fmt::Debug for HttpClientBuilder { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("HttpClientBuilder") + .field("endpoint", &self.endpoint) + .finish_non_exhaustive() + } +} + impl HttpClient { /// Create a client from a base URL and target the standard ACP endpoint. /// /// If the URL path is empty, `/acp` is used. Otherwise `/acp` is appended /// unless the path already ends with `/acp`. + /// + /// Use [`Self::builder`] to customize headers, TLS, proxies, or timeouts. pub fn new(base_url: impl AsRef) -> Result { - Self::with_client(base_url, reqwest::Client::new()) + Self::builder(base_url).build() } /// Create a client that targets the exact endpoint URL. /// /// Use this when connecting to a server configured with a custom - /// `ServerOptions::path`. + /// `ServerOptions::path`. Use [`Self::builder_with_endpoint`] to also + /// customize the HTTP client configuration. pub fn with_endpoint(endpoint: impl AsRef) -> Result { - Self::with_endpoint_and_client(endpoint, reqwest::Client::new()) + Self::builder_with_endpoint(endpoint).build() } - /// Create a client with a custom HTTP client and the standard ACP endpoint. + /// Configure a client from a base URL targeting the standard ACP endpoint. /// - /// If the URL path is empty, `/acp` is used. Otherwise `/acp` is appended - /// unless the path already ends with `/acp`. + /// Uses the same path normalization as [`Self::new`]. Invalid URLs and + /// unsupported schemes are reported by [`HttpClientBuilder::build`]. + /// + /// ``` + /// use std::time::Duration; + /// use agent_client_protocol_http::HttpClient; + /// + /// let transport = HttpClient::builder("wss://agent.example") + /// .configure_http(|http| { + /// http.connect_timeout(Duration::from_secs(5)) + /// .timeout(Duration::from_secs(10)) + /// }) + /// .build()?; + /// # Ok::<(), agent_client_protocol_http::HttpClientError>(()) + /// ``` + pub fn builder(base_url: impl AsRef) -> HttpClientBuilder { + HttpClientBuilder { + endpoint: parse_base_url(base_url.as_ref()), + http: reqwest::Client::builder(), + } + } + + /// Configure a client targeting an exact endpoint URL without changing its path. + /// + /// Use this when connecting to a server configured with a custom + /// `ServerOptions::path`. Invalid URLs and unsupported schemes are reported + /// by [`HttpClientBuilder::build`]. + pub fn builder_with_endpoint(endpoint: impl AsRef) -> HttpClientBuilder { + HttpClientBuilder { + endpoint: parse_endpoint(endpoint.as_ref()), + http: reqwest::Client::builder(), + } + } + + /// Reuse an existing reqwest client for HTTP/SSE at an exact endpoint URL. + /// + /// The path is not changed: include `/acp` or the server's custom path. + /// This preserves the supplied client's configuration and connection pool. + /// + /// Only `http://` and `https://` URLs are accepted. For WebSockets, use + /// [`Self::builder`] or [`Self::builder_with_endpoint`] so this transport can + /// configure HTTP/1.1 and disable handshake redirects before building the + /// underlying client. + pub fn from_http_client( + endpoint: impl AsRef, + http: reqwest::Client, + ) -> Result { + let endpoint = parse_endpoint(endpoint.as_ref())?; + if is_websocket_url(&endpoint) { + return Err(HttpClientError::WebSocketRequiresBuilder); + } + Ok(Self { endpoint, http }) + } + + /// Reuse an existing HTTP/SSE client with the standard ACP endpoint. + /// + /// Preserves the same base-URL path normalization as [`Self::new`]. + /// WebSocket URLs return [`HttpClientError::WebSocketRequiresBuilder`] + /// before any network I/O; migrate those calls to [`Self::builder`]. + #[deprecated( + note = "Use builder(...).configure_http(...).build(), or from_http_client with an exact HTTP/SSE endpoint" + )] pub fn with_client( base_url: impl AsRef, http: reqwest::Client, ) -> Result { - let mut endpoint = url::Url::parse(base_url.as_ref())?; - let path = endpoint.path().trim_end_matches('/').to_string(); - let path = if path.is_empty() { - "/acp".to_string() - } else if path.ends_with("/acp") { - path - } else { - format!("{path}/acp") - }; - endpoint.set_path(&path); - Ok(Self { endpoint, http }) + Self::from_http_client(parse_base_url(base_url.as_ref())?, http) } - /// Create a client with a custom HTTP client and exact endpoint URL. + /// Reuse an existing HTTP/SSE client at an exact endpoint. /// - /// Use this when connecting to a server configured with a custom - /// `ServerOptions::path`. + /// WebSocket URLs return [`HttpClientError::WebSocketRequiresBuilder`] + /// before any network I/O; migrate those calls to + /// [`Self::builder_with_endpoint`]. + #[deprecated( + note = "Use builder_with_endpoint(...).configure_http(...).build(), or from_http_client for an existing HTTP/SSE client" + )] pub fn with_endpoint_and_client( endpoint: impl AsRef, http: reqwest::Client, ) -> Result { - let endpoint = url::Url::parse(endpoint.as_ref())?; - Ok(Self { endpoint, http }) + Self::from_http_client(endpoint, http) } fn is_websocket(&self) -> bool { - matches!(self.endpoint.scheme(), "ws" | "wss") + is_websocket_url(&self.endpoint) } } +impl HttpClientBuilder { + /// Customize the HTTP client used for HTTP/SSE or the WebSocket handshake. + /// + /// Each call transforms the current configuration, retaining previous changes. + /// Configure default headers, proxies, DNS, trust roots, client certificates, + /// and timeouts through reqwest's builder rather than building a client first. + /// + /// For WebSockets, [`Self::build`] overrides the HTTP version preference with + /// HTTP/1.1 and disables redirects. HTTP/SSE retains the supplied settings. + /// Handshake headers are transport-owned; subprotocols and extensions are not + /// negotiated, even if custom default headers request them. + /// + /// # Timeouts + /// + /// Reqwest request/read timeouts apply to the WebSocket opening handshake, + /// not the lifetime of the upgraded socket. For HTTP/SSE, they retain their + /// normal reqwest request/body semantics, including long-lived SSE bodies. + /// + /// # Preconfigured TLS + /// + /// Prefer reqwest's TLS options for custom roots and identities. If using + /// `tls_backend_preconfigured`, its ALPN configuration must itself use + /// HTTP/1.1 for WebSockets: reqwest cannot rewrite a preconfigured backend's + /// ALPN. Incompatible negotiation is rejected before transmitting ACP data. + pub fn configure_http( + mut self, + configure: impl FnOnce(reqwest::ClientBuilder) -> reqwest::ClientBuilder, + ) -> Self { + self.http = configure(self.http); + self + } + + /// Build a client, applying the selected transport's connection policies. + /// + /// Accepts `http`, `https`, `ws`, and `wss` URLs. No connection is opened + /// until the resulting [`HttpClient`] is connected through [`ConnectTo`]. + pub fn build(self) -> Result { + let endpoint = self.endpoint?; + let http = if is_websocket_url(&endpoint) { + self.http + .http1_only() + .redirect(reqwest::redirect::Policy::none()) + } else { + self.http + } + .build()?; + Ok(HttpClient { endpoint, http }) + } +} + +fn parse_base_url(base_url: &str) -> Result { + let mut endpoint = parse_endpoint(base_url)?; + let path = endpoint.path().trim_end_matches('/'); + let path = if path.is_empty() { + "/acp".to_string() + } else if path.ends_with("/acp") { + path.to_string() + } else { + format!("{path}/acp") + }; + endpoint.set_path(&path); + Ok(endpoint) +} + +fn parse_endpoint(endpoint: &str) -> Result { + let endpoint = url::Url::parse(endpoint)?; + match endpoint.scheme() { + "http" | "https" | "ws" | "wss" => Ok(endpoint), + scheme => Err(HttpClientError::UnsupportedScheme(scheme.to_string())), + } +} + +fn is_websocket_url(endpoint: &url::Url) -> bool { + matches!(endpoint.scheme(), "ws" | "wss") +} + impl ConnectTo for HttpClient { async fn connect_to(self, client: impl ConnectTo) -> Result<(), AcpError> { let (channel, transport) = ConnectTo::::into_channel_and_future(self); @@ -1148,20 +1309,132 @@ fn pending_request_key(id: &RequestId) -> Option { } async fn run_ws(client: HttpClient, channel: Channel) -> Result<(), AcpError> { - let HttpClient { endpoint, .. } = client; + let HttpClient { endpoint, http } = client; - let (ws_stream, response) = async_tungstenite::tokio::connect_async(endpoint.as_str()) - .await - .map_err(|e| AcpError::internal_error().data(format!("WebSocket connect failed: {e}")))?; - trace!( - status = %response.status(), - "WebSocket connection established" - ); + let (ws_stream, status) = connect_ws(&http, endpoint).await?; + trace!(status = %status, "WebSocket connection established"); let (ws_tx, ws_rx) = ws_stream.split(); drive_ws(ws_tx, ws_rx, channel).await } +fn websocket_http_url(mut endpoint: url::Url) -> Result { + let scheme = match endpoint.scheme() { + "ws" => "http", + "wss" => "https", + other => { + return Err( + AcpError::internal_error().data(format!("unsupported WebSocket scheme: {other}")) + ); + } + }; + endpoint + .set_scheme(scheme) + .map_err(|()| AcpError::internal_error().data("failed to convert WebSocket URL"))?; + Ok(endpoint) +} + +async fn connect_ws( + http: &reqwest::Client, + endpoint: url::Url, +) -> Result< + ( + async_tungstenite::WebSocketStream< + async_tungstenite::tokio::TokioAdapter, + >, + reqwest::StatusCode, + ), + AcpError, +> { + let http_url = websocket_http_url(endpoint)?; + let key = async_tungstenite::tungstenite::handshake::client::generate_key(); + let expected_accept = + async_tungstenite::tungstenite::handshake::derive_accept_key(key.as_bytes()); + + let response = http + .get(http_url) + .version(reqwest::Version::HTTP_11) + .header("Connection", "Upgrade") + .header("Upgrade", "websocket") + .header("Sec-WebSocket-Version", "13") + .header("Sec-WebSocket-Key", &key) + .send() + .await + .map_err(|e| AcpError::internal_error().data(format!("WebSocket connect failed: {e}")))?; + let status = response.status(); + validate_ws_response( + response.version(), + status, + response.headers(), + &expected_accept, + )?; + + let upgraded = response + .upgrade() + .await + .map_err(|e| AcpError::internal_error().data(format!("WebSocket connect failed: {e}")))?; + let ws_stream = async_tungstenite::WebSocketStream::from_raw_socket( + async_tungstenite::tokio::TokioAdapter::new(upgraded), + async_tungstenite::tungstenite::protocol::Role::Client, + None, + ) + .await; + Ok((ws_stream, status)) +} + +fn validate_ws_response( + version: reqwest::Version, + status: reqwest::StatusCode, + headers: &reqwest::header::HeaderMap, + expected_accept: &str, +) -> Result<(), AcpError> { + let invalid = + |reason| AcpError::internal_error().data(format!("WebSocket connect failed: {reason}")); + if version != reqwest::Version::HTTP_11 { + return Err(invalid(format!( + "expected HTTP/1.1, received {version:?}; preconfigured TLS must use HTTP/1.1 ALPN" + ))); + } + if status != reqwest::StatusCode::SWITCHING_PROTOCOLS { + return Err(invalid(format!("unexpected status {status}"))); + } + let mut upgrades = headers.get_all("upgrade").iter(); + if !upgrades + .next() + .is_some_and(|value| value.as_bytes().eq_ignore_ascii_case(b"websocket")) + || upgrades.next().is_some() + { + return Err(invalid("invalid upgrade header".to_string())); + } + let connection_upgrade = headers.get_all("connection").iter().any(|value| { + value.to_str().is_ok_and(|value| { + value.split(',').any(|part| { + part.trim_matches([' ', '\t']) + .eq_ignore_ascii_case("upgrade") + }) + }) + }); + if !connection_upgrade { + return Err(invalid("invalid connection header".to_string())); + } + let mut accepts = headers.get_all("sec-websocket-accept").iter(); + if accepts.next().map(reqwest::header::HeaderValue::as_bytes) + != Some(expected_accept.as_bytes()) + || accepts.next().is_some() + { + return Err(invalid("invalid Sec-WebSocket-Accept".to_string())); + } + // ACP does not negotiate subprotocols or extensions. In particular, passing + // an extension through to from_raw_socket does not enable support for it + // (e.g. compression). + for header in ["sec-websocket-protocol", "sec-websocket-extensions"] { + if headers.contains_key(header) { + return Err(invalid(format!("unsupported {header}"))); + } + } + Ok(()) +} + trait WsSink { fn send( &mut self, @@ -1665,17 +1938,126 @@ mod tests { "http://example.com/agent" ); assert_eq!( - HttpClient::with_endpoint_and_client( - "ws://example.com/custom/acp?token=abc", - reqwest::Client::new(), - ) - .unwrap() - .endpoint - .as_str(), + HttpClient::builder_with_endpoint("ws://example.com/custom/acp?token=abc") + .build() + .unwrap() + .endpoint + .as_str(), "ws://example.com/custom/acp?token=abc" ); } + #[test] + fn builder_uses_the_same_base_url_rule_for_all_transports() { + for scheme in ["http", "https", "ws", "wss"] { + for (path, expected) in [ + ("", "/acp"), + ("/", "/acp"), + ("/proxy/", "/proxy/acp"), + ("/proxy/acp/", "/proxy/acp"), + ("/proxy/acp/nested", "/proxy/acp/nested/acp"), + ] { + let url = format!("{scheme}://example.com{path}?key=value"); + let client = HttpClient::builder(&url).build().unwrap(); + assert_eq!( + client.endpoint.as_str(), + format!("{scheme}://example.com{expected}?key=value") + ); + let exact = HttpClient::builder_with_endpoint(&url).build().unwrap(); + assert_eq!(exact.endpoint, url::Url::parse(&url).unwrap()); + } + } + } + + #[test] + fn constructors_reject_invalid_urls_and_unsupported_schemes() { + for build in [HttpClient::builder, HttpClient::builder_with_endpoint] { + assert!(matches!( + build("not a URL".to_string()).build(), + Err(HttpClientError::InvalidUrl(_)) + )); + for scheme in ["file", "ftp", "custom"] { + assert!(matches!( + build(format!("{scheme}://example.com/acp")).build(), + Err(HttpClientError::UnsupportedScheme(actual)) if actual == scheme + )); + } + } + } + + #[test] + fn prebuilt_http_client_requires_an_http_endpoint() { + let http = reqwest::Client::new(); + for scheme in ["http", "https"] { + let endpoint = format!("{scheme}://example.com/custom?key=value"); + let client = HttpClient::from_http_client(&endpoint, http.clone()).unwrap(); + assert_eq!(client.endpoint.as_str(), endpoint); + } + for scheme in ["ws", "wss"] { + assert!(matches!( + HttpClient::from_http_client(format!("{scheme}://example.com/acp"), http.clone()), + Err(HttpClientError::WebSocketRequiresBuilder) + )); + } + assert!(matches!( + HttpClient::from_http_client("ftp://example.com/acp", http), + Err(HttpClientError::UnsupportedScheme(_)) + )); + } + + #[test] + #[allow(deprecated)] + fn deprecated_constructors_preserve_http_paths_and_reject_websockets() { + let http = reqwest::Client::new(); + for scheme in ["http", "https"] { + for path in ["", "/proxy", "/proxy/acp/"] { + let url = format!("{scheme}://example.com{path}?key=value"); + let legacy = HttpClient::with_client(&url, http.clone()).unwrap(); + assert_eq!(legacy.endpoint, HttpClient::new(&url).unwrap().endpoint); + let exact = HttpClient::with_endpoint_and_client(&url, http.clone()).unwrap(); + assert_eq!( + exact.endpoint, + HttpClient::with_endpoint(&url).unwrap().endpoint + ); + } + } + for scheme in ["ws", "wss"] { + let url = format!("{scheme}://example.com/custom"); + assert!(matches!( + HttpClient::with_client(&url, http.clone()), + Err(HttpClientError::WebSocketRequiresBuilder) + )); + assert!(matches!( + HttpClient::with_endpoint_and_client(&url, http.clone()), + Err(HttpClientError::WebSocketRequiresBuilder) + )); + } + } + + #[test] + fn builder_propagates_http_configuration_errors_without_panicking() { + let error = HttpClient::builder("ws://example.com") + .configure_http(|http| http.user_agent("\n")) + .build() + .unwrap_err(); + assert!(matches!(error, HttpClientError::Reqwest(_))); + } + + #[test] + fn client_and_builder_debug_do_not_expose_default_headers() { + let mut headers = HeaderMap::new(); + headers.insert( + "x-api-key", + HeaderValue::from_static("private-header-value"), + ); + let builder = HttpClient::builder("ws://example.com") + .configure_http(|http| http.default_headers(headers)); + assert!(!format!("{builder:?}").contains("private-header-value")); + let client = builder.build().unwrap(); + assert!(!format!("{client:?}").contains("private-header-value")); + assert_eq!(client.clone().endpoint, client.endpoint); + } + #[tokio::test] async fn post_sends_cancel_request_without_session_header() { let (capture_tx, mut capture_rx) = tokio::sync::mpsc::unbounded_channel(); @@ -3234,6 +3616,162 @@ mod tests { server.abort(); } + fn valid_ws_response_headers() -> HeaderMap { + HeaderMap::from_iter([ + ( + reqwest::header::UPGRADE, + HeaderValue::from_static("websocket"), + ), + ( + reqwest::header::CONNECTION, + HeaderValue::from_static("Upgrade"), + ), + ( + reqwest::header::SEC_WEBSOCKET_ACCEPT, + HeaderValue::from_static("s3pPLMBiTxaQ9kYGzzhZRbK+xOo="), + ), + ]) + } + + #[test] + fn websocket_response_validation() { + let valid = valid_ws_response_headers(); + let validate = |version, status, headers: &HeaderMap| { + validate_ws_response(version, status, headers, "s3pPLMBiTxaQ9kYGzzhZRbK+xOo=") + }; + let version = reqwest::Version::HTTP_11; + let status = StatusCode::SWITCHING_PROTOCOLS; + validate(version, status, &valid).unwrap(); + for version in [ + reqwest::Version::HTTP_10, + reqwest::Version::HTTP_2, + reqwest::Version::HTTP_3, + ] { + assert!(validate(version, status, &valid).is_err()); + } + for status in [StatusCode::OK, StatusCode::BAD_REQUEST, StatusCode::FOUND] { + assert!(validate(version, status, &valid).is_err()); + } + for (header, invalid_values) in [ + ( + "upgrade", + vec!["", "h2c", "websocket/13", "notwebsocket", "websocket, h2c"], + ), + ("connection", vec!["", "keep-alive", "notupgrade"]), + ( + "sec-websocket-accept", + vec!["", "wrong", "s3pPLMBiTxaQ9kYGzzhZRbK+xOo=, wrong"], + ), + ] { + let mut headers = valid.clone(); + headers.remove(header); + assert!(validate(version, status, &headers).is_err(), "{header}"); + for value in invalid_values { + headers.insert(header, HeaderValue::from_str(value).unwrap()); + assert!( + validate(version, status, &headers).is_err(), + "{header}: {value}" + ); + } + headers.insert(header, HeaderValue::from_bytes(b"\xff").unwrap()); + assert!(validate(version, status, &headers).is_err(), "{header}"); + } + for header in ["upgrade", "sec-websocket-accept"] { + let mut duplicate = valid.clone(); + duplicate.append(header, valid[header].clone()); + assert!(validate(version, status, &duplicate).is_err(), "{header}"); + } + + for header in ["sec-websocket-protocol", "sec-websocket-extensions"] { + for value in ["", "acp", "permessage-deflate"] { + let mut headers = valid.clone(); + headers.insert(header, HeaderValue::from_str(value).unwrap()); + assert!(validate(version, status, &headers).is_err(), "{header}"); + } + } + + let mut token_lists = valid; + token_lists.insert("upgrade", HeaderValue::from_static("WebSocket")); + token_lists.insert("connection", HeaderValue::from_static("keep-alive")); + token_lists.append("connection", HeaderValue::from_static("other, uPgRaDe\t ")); + validate(version, status, &token_lists).unwrap(); + } + + #[tokio::test] + async fn websocket_public_transport_validates_before_sending_acp() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + for upgrade in ["websocket", "not-websocket"] { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let fixture = async { + let (mut socket, _) = listener.accept().await.unwrap(); + // Read exactly the handshake, preserving any subsequent ACP + // bytes so the assertion also detects premature writes. + let mut request = Vec::new(); + while !request.ends_with(b"\r\n\r\n") { + request.push(socket.read_u8().await.unwrap()); + assert!(request.len() < 16 * 1024); + } + let request = String::from_utf8(request).unwrap(); + let key = request + .lines() + .filter_map(|line| line.split_once(':')) + .find(|(name, _)| name.eq_ignore_ascii_case("sec-websocket-key")) + .unwrap() + .1 + .trim(); + let accept = + async_tungstenite::tungstenite::handshake::derive_accept_key(key.as_bytes()); + socket + .write_all( + format!( + "HTTP/1.1 101 Switching Protocols\r\nUpgrade: {upgrade}\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: {accept}\r\n\r\n" + ) + .as_bytes(), + ) + .await + .unwrap(); + let mut received = Vec::new(); + socket.read_to_end(&mut received).await.unwrap(); + received + }; + let client = HttpClient::new(format!("ws://{addr}")).unwrap(); + let (caller, transport) = ConnectTo::::into_channel_and_future(client); + caller + .tx + .unbounded_send(single_frame( + RawJsonRpcMessage::notification("custom/queued".to_string(), json!({})) + .unwrap(), + )) + .unwrap(); + drop(caller); + + // Neither future is spawned: timeout drops both fixtures and + // sockets together, without detached tasks or synchronization sleeps. + let (result, received) = timeout(Duration::from_secs(2), async { + futures::join!(transport, fixture) + }) + .await + .expect("handshake fixture should complete"); + if upgrade == "websocket" { + result.unwrap(); + assert!(!received.is_empty(), "valid handshake must send queued ACP"); + assert_eq!(received[0], 0x81, "first frame must be WebSocket text"); + } else { + let error = result.unwrap_err(); + assert!( + error.to_string().contains("invalid upgrade header"), + "{error}" + ); + assert!( + received.is_empty(), + "ACP escaped before handshake validation" + ); + } + } + } + #[tokio::test] async fn websocket_serializes_batch_as_one_text_frame() { let (caller, transport) = Channel::duplex(); @@ -3391,6 +3929,89 @@ mod tests { server.abort(); } + #[tokio::test] + async fn websocket_builder_sends_default_headers() { + let (header_tx, mut header_rx) = tokio::sync::mpsc::unbounded_channel(); + let app = Router::new().route( + "/acp", + get(move |headers: HeaderMap, ws: WebSocketUpgrade| { + let header_tx = header_tx.clone(); + async move { + header_tx.send(headers).unwrap(); + ws.on_upgrade(|_socket| async {}) + } + }), + ); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let mut default_headers = reqwest::header::HeaderMap::new(); + default_headers.insert( + reqwest::header::HeaderName::from_static("x-acp-test-client"), + reqwest::header::HeaderValue::from_static("from-reqwest"), + ); + let client = HttpClient::builder(format!("ws://{addr}")) + .configure_http(|http| http.default_headers(default_headers)) + .configure_http(reqwest::ClientBuilder::no_proxy) + .build() + .unwrap(); + let (_caller, transport) = Channel::duplex(); + let transport = tokio::spawn(run(client, transport)); + + let headers = timeout(Duration::from_secs(1), header_rx.recv()) + .await + .expect("WebSocket handshake should reach the server") + .expect("handshake headers were not captured"); + assert_eq!( + headers.get("x-acp-test-client").map(HeaderValue::as_bytes), + Some(&b"from-reqwest"[..]), + "default headers must be retained across configure_http calls and sent on the handshake" + ); + + transport.abort(); + drop(transport.await); + server.abort(); + drop(server.await); + } + + async fn assert_websocket_handshake_times_out( + configure: impl FnOnce(reqwest::ClientBuilder) -> reqwest::ClientBuilder, + ) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let client = HttpClient::builder(format!("ws://{addr}")) + .configure_http(reqwest::ClientBuilder::no_proxy) + .configure_http(configure) + .build() + .unwrap(); + let (_caller, transport) = Channel::duplex(); + + let error = timeout(Duration::from_secs(1), run(client, transport)) + .await + .expect("custom reqwest timeout should fail the WebSocket handshake") + .expect_err("handshake should not succeed while the listener never accepts"); + assert!( + error.to_string().contains("WebSocket connect failed"), + "{error}" + ); + + drop(listener); + } + + #[tokio::test] + async fn websocket_builder_honors_request_timeout() { + assert_websocket_handshake_times_out(|http| http.timeout(Duration::from_millis(200))).await; + } + + #[tokio::test] + async fn websocket_builder_honors_read_timeout() { + assert_websocket_handshake_times_out(|http| http.read_timeout(Duration::from_millis(200))) + .await; + } + #[tokio::test] async fn dropped_transport_future_deletes_initialized_connection() { let delete_count = Arc::new(AtomicUsize::new(0)); diff --git a/src/agent-client-protocol-http/src/lib.rs b/src/agent-client-protocol-http/src/lib.rs index 854c12a..ae2be40 100644 --- a/src/agent-client-protocol-http/src/lib.rs +++ b/src/agent-client-protocol-http/src/lib.rs @@ -14,6 +14,6 @@ mod server; mod websocket_server; #[cfg(feature = "client")] -pub use client::{HttpClient, HttpClientError}; +pub use client::{HttpClient, HttpClientBuilder, HttpClientError}; #[cfg(feature = "server")] pub use server::{AcpHttpServer, CorsOptions, ServerOptions}; diff --git a/src/agent-client-protocol-http/tests/websocket_client.rs b/src/agent-client-protocol-http/tests/websocket_client.rs new file mode 100644 index 0000000..b622a2b --- /dev/null +++ b/src/agent-client-protocol-http/tests/websocket_client.rs @@ -0,0 +1,512 @@ +#![cfg(feature = "client")] + +use std::{future::Future, sync::Arc, time::Duration}; + +use agent_client_protocol::{Channel, Client, ConnectTo, RawJsonRpcMessage, TransportFrame}; +use agent_client_protocol_http::HttpClient; +use async_tungstenite::{tokio::accept_hdr_async, tungstenite::handshake::server::Request}; +use futures::{StreamExt, future::BoxFuture}; +use serde_json::json; +use tokio::{ + io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}, + net::{TcpListener, TcpStream}, + time::timeout, +}; +use tokio_rustls::{TlsAcceptor, rustls}; + +const DEADLINE: Duration = Duration::from_secs(10); + +struct Tls { + certificate: rustls::pki_types::CertificateDer<'static>, + acceptor: TlsAcceptor, +} + +impl Tls { + fn new() -> Self { + let rcgen::CertifiedKey { cert, signing_key } = + rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap(); + let certificate = cert.der().clone(); + let key = rustls::pki_types::PrivatePkcs8KeyDer::from(signing_key.serialize_der()); + let mut config = rustls::ServerConfig::builder_with_provider(Arc::new( + rustls::crypto::aws_lc_rs::default_provider(), + )) + .with_safe_default_protocol_versions() + .unwrap() + .with_no_client_auth() + .with_single_cert(vec![certificate.clone()], key.into()) + .unwrap(); + // Put h2 first: simply enabling HTTP/2 in a downstream crate used to + // make reqwest select it for this HTTP/1.1-only upgrade. + config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()]; + Self { + certificate, + acceptor: TlsAcceptor::from(Arc::new(config)), + } + } + + fn configure(&self, builder: reqwest::ClientBuilder) -> reqwest::ClientBuilder { + builder + .no_proxy() + .tls_certs_merge([reqwest::Certificate::from_der(&self.certificate).unwrap()]) + } + + fn raw_config(&self, protocol: &[u8]) -> rustls::ClientConfig { + let mut roots = rustls::RootCertStore::empty(); + roots.add(self.certificate.clone()).unwrap(); + let mut config = rustls::ClientConfig::builder_with_provider(Arc::new( + rustls::crypto::aws_lc_rs::default_provider(), + )) + .with_safe_default_protocol_versions() + .unwrap() + .with_root_certificates(roots) + .with_no_client_auth(); + config.alpn_protocols = vec![protocol.to_vec()]; + config + } +} + +async fn listen() -> TcpListener { + TcpListener::bind("127.0.0.1:0").await.unwrap() +} + +fn queued( + client: HttpClient, +) -> ( + Channel, + BoxFuture<'static, Result<(), agent_client_protocol::Error>>, +) { + let (caller, transport) = ConnectTo::::into_channel_and_future(client); + caller + .tx + .unbounded_send(TransportFrame::Single( + RawJsonRpcMessage::notification("custom/queued".into(), json!({"probe": 333})).unwrap(), + )) + .unwrap(); + (caller, transport) +} + +async fn headers(socket: &mut (impl AsyncRead + Unpin)) -> String { + let mut bytes = Vec::new(); + while !bytes.ends_with(b"\r\n\r\n") { + bytes.push(socket.read_u8().await.unwrap()); + assert!(bytes.len() < 16 * 1024, "oversized handshake"); + } + String::from_utf8(bytes).unwrap() +} + +// Tungstenite fixes the callback's error type to an unboxed HTTP response. +#[allow(clippy::result_large_err)] +async fn exchange(socket: impl AsyncRead + AsyncWrite + Unpin, path: &str, delay: Duration) { + let mut ws = accept_hdr_async(socket, |request: &Request, response| { + assert_eq!(request.uri().path(), path); + Ok(response) + }) + .await + .unwrap(); + let message = ws.next().await.unwrap().unwrap(); + let value: serde_json::Value = serde_json::from_str(message.to_text().unwrap()).unwrap(); + assert_eq!(value["method"], "custom/queued"); + assert_eq!(value["params"]["probe"], 333); + // Deliberately cross the configured request/read deadline after upgrade. + // This is the behavior under test, not a fixture synchronization sleep. + if !delay.is_zero() { + tokio::time::sleep(delay).await; + } + ws.send(message).await.unwrap(); + // The observer drops its ACP channel after receiving the echo, allowing + // the writer to finish successfully before the server closes its socket. + assert!(ws.next().await.unwrap().unwrap().is_close()); +} + +fn successful_exchange( + client: HttpClient, + fixture: impl Future, +) -> impl Future { + // Box before constructing the runner future so nested TLS state machines + // do not inflate every caller's future. + let fixture = Box::pin(fixture); + async move { + let (mut caller, transport) = queued(client); + let observe = async { + let message = caller.rx.next().await.expect("echo from upgraded stream"); + let value: serde_json::Value = + serde_json::from_str(&message.to_json().unwrap()).unwrap(); + assert_eq!(value["method"], "custom/queued"); + assert_eq!(value["params"]["probe"], 333); + drop(caller); + }; + let (result, (), ()) = timeout(DEADLINE, async { + futures::join!(transport, fixture, observe) + }) + .await + .expect("client and fixture must finish"); + result.unwrap(); + } +} + +#[tokio::test] +async fn custom_roots_and_http2_feature_unification_use_http1_for_wss() { + for exact in [false, true] { + let tls = Tls::new(); + let listener = listen().await; + let url = format!( + "wss://localhost:{}/custom", + listener.local_addr().unwrap().port() + ); + let builder = if exact { + HttpClient::builder_with_endpoint(&url) + } else { + HttpClient::builder(&url) + }; + let client = builder + .configure_http(|builder| tls.configure(builder).http2_prior_knowledge()) + .build() + .unwrap(); + successful_exchange(client, async { + let (socket, _) = listener.accept().await.unwrap(); + let socket = tls.acceptor.accept(socket).await.unwrap(); + assert_eq!( + socket.get_ref().1.alpn_protocol(), + Some(b"http/1.1".as_slice()) + ); + exchange( + socket, + if exact { "/custom" } else { "/custom/acp" }, + Duration::ZERO, + ) + .await; + }) + .await; + } +} + +#[tokio::test] +async fn websocket_redirects_never_reach_destination_or_send_acp() { + for (secure, destination_scheme) in [(false, "http"), (true, "https"), (true, "http")] { + let tls = Tls::new(); + let origin = listen().await; + let destination = listen().await; + let location = format!( + "{destination_scheme}://localhost:{}/stolen", + destination.local_addr().unwrap().port() + ); + let scheme = if secure { "wss" } else { "ws" }; + let client = HttpClient::builder(format!( + "{scheme}://localhost:{}", + origin.local_addr().unwrap().port() + )) + .configure_http(|builder| { + tls.configure(builder) + .redirect(reqwest::redirect::Policy::limited(5)) + }) + .build() + .unwrap(); + let (caller, transport) = queued(client); + let fixture = async { + let (socket, _) = origin.accept().await.unwrap(); + if secure { + redirect(tls.acceptor.accept(socket).await.unwrap(), &location).await; + } else { + redirect(socket, &location).await; + } + }; + timeout(DEADLINE, async { + tokio::select! { + biased; + connection = destination.accept() => panic!("redirect destination contacted: {connection:?}"), + (result, ()) = async { futures::join!(transport, fixture) } => { + assert!(result.is_err(), "redirect must fail closed"); + } + } + // Catch a connection already queued when the transport completed. + assert!(futures::poll!(Box::pin(destination.accept())).is_pending()); + }).await.unwrap(); + drop(caller); + } +} + +async fn redirect(mut socket: impl AsyncRead + AsyncWrite + Unpin, location: &str) { + assert!( + headers(&mut socket) + .await + .starts_with("GET /acp HTTP/1.1\r\n") + ); + socket.write_all(format!("HTTP/1.1 302 Found\r\nLocation: {location}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").as_bytes()).await.unwrap(); + let mut received = Vec::new(); + if let Err(error) = socket.read_to_end(&mut received).await { + // A rejected handshake may drop TLS without sending close_notify. + assert!( + matches!( + error.kind(), + std::io::ErrorKind::UnexpectedEof | std::io::ErrorKind::ConnectionReset + ), + "{error}" + ); + } + assert!(received.is_empty(), "ACP sent to redirecting origin"); +} + +#[tokio::test] +async fn configured_proxy_tunnels_wss_with_authentication() { + let tls = Tls::new(); + let origin = listen().await; + let proxy = listen().await; + let target = format!("localhost:{}", origin.local_addr().unwrap().port()); + let client = HttpClient::builder(format!("wss://{target}")) + .configure_http(|builder| { + tls.configure(builder).proxy( + reqwest::Proxy::all(format!("http://{}", proxy.local_addr().unwrap())) + .unwrap() + .basic_auth("acp", "secret"), + ) + }) + .build() + .unwrap(); + successful_exchange(client, async { + let tunnel = async { + let (mut downstream, _) = proxy.accept().await.unwrap(); + let request = headers(&mut downstream).await; + assert!(request.starts_with(&format!("CONNECT {target} HTTP/1.1\r\n"))); + assert_eq!( + request + .lines() + .filter_map(|line| line.split_once(':')) + .find(|(name, _)| name.eq_ignore_ascii_case("proxy-authorization")) + .map(|(_, value)| value.trim()), + Some("Basic YWNwOnNlY3JldA==") + ); + let mut upstream = TcpStream::connect(origin.local_addr().unwrap()) + .await + .unwrap(); + downstream + .write_all(b"HTTP/1.1 200 Connection Established\r\n\r\n") + .await + .unwrap(); + // Ends when the client and TLS server close their upgraded sockets. + drop(tokio::io::copy_bidirectional(&mut downstream, &mut upstream).await); + }; + let server = async { + let (socket, _) = origin.accept().await.unwrap(); + let socket = tls.acceptor.accept(socket).await.unwrap(); + exchange(socket, "/acp", Duration::ZERO).await; + }; + futures::join!(tunnel, server); + }) + .await; +} + +#[tokio::test] +async fn request_and_read_timeouts_do_not_expire_upgraded_stream() { + let listener = listen().await; + let client = HttpClient::builder(format!("ws://{}", listener.local_addr().unwrap())) + .configure_http(|builder| { + builder + .no_proxy() + .timeout(Duration::from_secs(1)) + .read_timeout(Duration::from_secs(1)) + }) + .build() + .unwrap(); + successful_exchange(client, async { + let (socket, _) = listener.accept().await.unwrap(); + exchange(socket, "/acp", Duration::from_secs(2)).await; + }) + .await; +} + +#[tokio::test] +async fn preconfigured_rustls_http1_supports_websocket_exchange() { + let tls = Tls::new(); + let listener = listen().await; + let client = HttpClient::builder(format!( + "wss://localhost:{}", + listener.local_addr().unwrap().port() + )) + .configure_http(|builder| { + builder + .no_proxy() + .tls_backend_preconfigured(tls.raw_config(b"http/1.1")) + }) + .build() + .unwrap(); + successful_exchange(client, async { + let (socket, _) = listener.accept().await.unwrap(); + let socket = tls.acceptor.accept(socket).await.unwrap(); + assert_eq!( + socket.get_ref().1.alpn_protocol(), + Some(b"http/1.1".as_slice()) + ); + exchange(socket, "/acp", Duration::ZERO).await; + }) + .await; +} + +#[tokio::test] +async fn incompatible_preconfigured_alpn_fails_without_acp_frames() { + let tls = Tls::new(); + let listener = listen().await; + let client = HttpClient::builder(format!( + "wss://localhost:{}", + listener.local_addr().unwrap().port() + )) + .configure_http(|builder| { + builder + .no_proxy() + .tls_backend_preconfigured(tls.raw_config(b"h2")) + }) + .build() + .unwrap(); + let (caller, transport) = queued(client); + let fixture = async { + let (socket, _) = listener.accept().await.unwrap(); + let socket = tls.acceptor.accept(socket).await.unwrap(); + assert_eq!(socket.get_ref().1.alpn_protocol(), Some(b"h2".as_slice())); + // Complete an actual h2 exchange, rather than merely provoking a TLS + // or HTTP parsing error. The response cannot authorize a WS upgrade. + let mut connection = h2::server::handshake(socket).await.unwrap(); + let (request, mut respond) = connection.accept().await.unwrap().unwrap(); + assert_eq!(request.method(), "GET"); + assert_eq!(request.uri().path(), "/acp"); + assert!(request.body().is_end_stream(), "ACP in handshake request"); + respond + .send_response( + async_tungstenite::tungstenite::http::Response::builder() + .status(200) + .body(()) + .unwrap(), + true, + ) + .unwrap(); + // Driving the connection flushes the response. The failed transport + // must close it without creating any further streams. + assert!(!matches!(connection.accept().await, Some(Ok(_)))); + }; + let (result, ()) = timeout(DEADLINE, async { futures::join!(transport, fixture) }) + .await + .unwrap(); + let error = result.unwrap_err(); + assert!(error.to_string().contains("expected HTTP/1.1"), "{error}"); + drop(caller); +} + +fn initialize( + client: HttpClient, +) -> ( + Channel, + BoxFuture<'static, Result<(), agent_client_protocol::Error>>, +) { + let (caller, transport) = ConnectTo::::into_channel_and_future(client); + caller + .tx + .unbounded_send(TransportFrame::Single( + RawJsonRpcMessage::request( + "initialize".into(), + json!({}), + agent_client_protocol::schema::v1::RequestId::Number(1), + ) + .unwrap(), + )) + .unwrap(); + (caller, transport) +} + +#[tokio::test] +async fn http_builder_retains_redirect_policy() { + let origin = listen().await; + let destination = listen().await; + let client = HttpClient::builder(format!("http://{}", origin.local_addr().unwrap())) + .configure_http(|builder| { + builder + .no_proxy() + .redirect(reqwest::redirect::Policy::limited(1)) + }) + .build() + .unwrap(); + let (caller, transport) = initialize(client); + let fixture = async { + let (mut socket, _) = origin.accept().await.unwrap(); + observe_initialize(&mut socket).await; + socket.write_all(format!( + "HTTP/1.1 307 Temporary Redirect\r\nLocation: http://{}/acp\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + destination.local_addr().unwrap(), + ).as_bytes()).await.unwrap(); + drop(socket); + let (mut socket, _) = destination.accept().await.unwrap(); + observe_initialize(&mut socket).await; + socket + .write_all( + b"HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + ) + .await + .unwrap(); + }; + let (result, ()) = timeout(DEADLINE, async { futures::join!(transport, fixture) }) + .await + .unwrap(); + assert!(result.is_err(), "dummy initialize response should fail"); + drop(caller); +} + +async fn observe_initialize(socket: &mut TcpStream) { + let request = headers(socket).await; + assert!(request.starts_with("POST /acp HTTP/1.1\r\n")); + let length: usize = request + .lines() + .filter_map(|line| line.split_once(':')) + .find(|(name, _)| name.eq_ignore_ascii_case("content-length")) + .unwrap() + .1 + .trim() + .parse() + .unwrap(); + assert!(length < 16 * 1024); + let mut body = vec![0; length]; + socket.read_exact(&mut body).await.unwrap(); + let message: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(message["method"], "initialize"); +} + +#[tokio::test] +async fn http_builder_retains_http2_prior_knowledge() { + let listener = listen().await; + let client = HttpClient::builder(format!("http://{}", listener.local_addr().unwrap())) + .configure_http(|builder| builder.no_proxy().http2_prior_knowledge()) + .build() + .unwrap(); + let (caller, transport) = initialize(client); + let fixture = async { + let (socket, _) = listener.accept().await.unwrap(); + let mut connection = h2::server::handshake(socket).await.unwrap(); + let (request, mut respond) = connection.accept().await.unwrap().unwrap(); + assert_eq!(request.method(), "POST"); + assert_eq!(request.uri().path(), "/acp"); + let body = async { + let mut stream = request.into_body(); + let mut body = Vec::new(); + while let Some(bytes) = stream.data().await { + body.extend_from_slice(&bytes.unwrap()); + assert!(body.len() < 16 * 1024); + } + let message: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(message["method"], "initialize"); + respond + .send_response( + async_tungstenite::tungstenite::http::Response::builder() + .status(400) + .body(()) + .unwrap(), + true, + ) + .unwrap(); + }; + let drive = async { + assert!(!matches!(connection.accept().await, Some(Ok(_)))); + }; + futures::join!(body, drive); + }; + let (result, ()) = timeout(DEADLINE, async { futures::join!(transport, fixture) }) + .await + .unwrap(); + assert!(result.is_err(), "dummy initialize response should fail"); + drop(caller); +}