diff --git a/libs/cua-driver/docs/private-envelope-http.md b/libs/cua-driver/docs/private-envelope-http.md new file mode 100644 index 000000000..fb0805942 --- /dev/null +++ b/libs/cua-driver/docs/private-envelope-http.md @@ -0,0 +1,73 @@ +# Private typed envelope HTTP carrier + +This carrier connects unary HTTP requests to `DriverEnvelopeReceiver`. It is +disabled unless the trusted launcher sets `CUA_DRIVER_ENVELOPE_HTTP_PORT` to a +nonzero port. It binds only to `127.0.0.1` and fails startup if binding fails. +No non-loopback binding option is provided. + +This is an implementation contract, not a released Fleet setup guide. A later +image/service change must connect the loopback endpoint to Fleet's existing +authorized private named-service route. Do not expose this endpoint directly: +it does not authenticate callers. IDs and generations are lifecycle markers, +not credentials. The carrier introduces no bearer token or pairing scheme. + +## Wire contract + +All requests use HTTP/1.1. POST bodies require one valid `Content-Length`. +Responses are JSON with `Connection: close`; pipelined requests are not executed. +Browser-origin requests, transfer encoding, duplicate lengths, and query strings +are refused. There is no CORS support. + +| Request | Body | Result | +| ------------------------------------ | --------------------------------- | ------------------------------------------------------------------- | +| `POST /v1/connections` | `{}` | `connection_id`, `generation`, `public_session`, and `capabilities` | +| `POST /v1/connections/{id}/exchange` | Canonical `DriverRequestEnvelope` | Canonical `DriverResponseEnvelope` | +| `POST /v1/connections/{id}/cancel` | `{"request_id":"REQUEST_ID"}` | `{"ok":true}` | +| `DELETE /v1/connections/{id}` | Empty or `{}` | `{"ok":true}` | + +Every connection-specific request requires `X-Cua-Driver-Generation` with the +value returned at creation. Missing generations fail with HTTP 400, absent +connections with 404, and mismatches with 409. A client must not reopen a +connection automatically after those failures. + +Creation binds a Standard session with a one-hour maximum lifetime and a +five-minute idle lifetime. The immutable runtime ceiling still applies: +incompatible runtimes refuse creation. This slice does not inherit unrestricted +mode or accept permission modes, manifests, or arbitrary session options from +the wire. Ordinary typed calls use `session=None`; operations requiring a +session label use the returned `public_session`. + +`capabilities` contains `minimum_envelope_version`, `maximum_envelope_version`, +and `supports_cancellation`. This carrier supports envelope version 1 and +cancellation. Independent `bind_session` requests are not implemented; the +returned canonical Driver root already owns one bound session. + +## Resource and cleanup limits + +- Request headers: 16 KiB; request body: 1 MiB. +- Response body: 16 MiB. An oversized exchange closes its session and returns + `response_too_large` with unknown completion; it must not be replayed. +- Live connection entries: 64, reaped after five idle minutes. +- HTTP tasks: 64; exchanges: 32, admitted without an unbounded wait queue. + Remaining task capacity permits cancellation and close during action load. +- Request reading and response writing: ten-second limits. Exchange deadlines + and per-connection replay prevention are enforced by the receiver. + +Close is idempotent and retains the closed request ledger until idle removal. +Idle reaping skips active exchanges. Shutdown cancels the listener and its +connection tasks and closes owned sessions. It does not shut down the shared +Driver runtime or computer-server, release a Fleet claim, or delete a pool. + +## Verification boundary + +Synthetic parser/routing tests cover malformed input, limits, generation +checks, raw results, close, active-session reaping, cancellation under full +exchange load, and oversized responses. Run them from `libs/cua-driver/rust`: + +```sh +cargo test -p cua-driver --bin cua-driver driver_service_http --locked +``` + +These tests do not establish guest network isolation, Fleet authorization, +image packaging, real desktop effects, or compatibility with a released Python +client. Those remain explicit integration and qualification gates. diff --git a/libs/cua-driver/rust/Cargo.lock b/libs/cua-driver/rust/Cargo.lock index a0b00be36..7736dbca4 100644 --- a/libs/cua-driver/rust/Cargo.lock +++ b/libs/cua-driver/rust/Cargo.lock @@ -1153,6 +1153,7 @@ dependencies = [ "embed-resource", "flate2", "futures-util", + "httparse", "image", "libc", "pip-preview", diff --git a/libs/cua-driver/rust/crates/cua-driver/Cargo.toml b/libs/cua-driver/rust/crates/cua-driver/Cargo.toml index 3b76b0271..bfcf45650 100644 --- a/libs/cua-driver/rust/crates/cua-driver/Cargo.toml +++ b/libs/cua-driver/rust/crates/cua-driver/Cargo.toml @@ -21,6 +21,7 @@ cua-driver-sdk = { path = "../cua-driver-sdk" } cursor-overlay = { path = "../cursor-overlay" } pip-preview = { path = "../pip-preview" } async-trait = "0.1" +httparse = "1.10.1" base64 = { workspace = true } uuid = { workspace = true } zeroize = { workspace = true } diff --git a/libs/cua-driver/rust/crates/cua-driver/src/driver_service_http.rs b/libs/cua-driver/rust/crates/cua-driver/src/driver_service_http.rs new file mode 100644 index 000000000..886a18dc0 --- /dev/null +++ b/libs/cua-driver/rust/crates/cua-driver/src/driver_service_http.rs @@ -0,0 +1,769 @@ +//! Private unary carrier, disabled unless CUA_DRIVER_ENVELOPE_HTTP_PORT is set. +//! +//! This loopback endpoint is NOT an authentication boundary. A trusted upstream +//! service must authorize and forward requests inside the guest before any later +//! Fleet exposure; packaging and exposure require a separate explicit change. +//! Connection IDs and generations are routing/lifecycle markers, not credentials. +//! Each connection already has a host-bound root session; independent bound +//! sessions are unsupported. No request supplies permission options or paths. +//! This first slice requests Standard sessions only; incompatible runtime +//! ceilings refuse creation. It does not inherit or widen the runtime mode. + +use cua_driver_sdk::remote::DriverRequestEnvelope; +use cua_driver_sdk::remote_receiver::DriverEnvelopeReceiver; +use serde::Deserialize; +use serde_json::{json, Value}; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; + +const MAX_HEADERS: usize = 16 * 1024; +const MAX_BODY: usize = 1024 * 1024; +const MAX_RESPONSE: usize = 16 * 1024 * 1024; +const MAX_CONNECTIONS: usize = 64; +const MAX_EXCHANGES: usize = 32; +const IDLE: Duration = Duration::from_secs(300); +const IO_TIMEOUT: Duration = Duration::from_secs(10); +type Factory = dyn Fn() -> Result<(Arc, String), String> + Send + Sync; +type HttpResult = Result; + +struct Entry { + receiver: Arc, + touched: Instant, + active: usize, +} + +struct Service { + entries: Mutex>, + factory: Arc, + exchanges: tokio::sync::Semaphore, +} + +struct Active { + service: Arc, + id: String, + receiver: Arc, +} + +impl Drop for Active { + fn drop(&mut self) { + if let Some(entry) = self.service.entries.lock().unwrap().get_mut(&self.id) { + entry.active -= 1; + entry.touched = Instant::now(); + } + } +} + +impl Drop for Service { + fn drop(&mut self) { + for entry in self.entries.get_mut().unwrap().values() { + entry.receiver.close(); + } + } +} + +impl Service { + fn reap(&self) { + self.entries.lock().unwrap().retain(|_, entry| { + if entry.active == 0 && entry.touched.elapsed() >= IDLE { + entry.receiver.close(); + false + } else { + true + } + }); + } + + fn lookup(self: &Arc, id: &str, generation: &str) -> HttpResult { + let mut entries = self.entries.lock().unwrap(); + let entry = entries.get_mut(id).ok_or((404, "connection_not_found"))?; + if entry.receiver.generation() != generation { + return Err((409, "stale_connection")); + } + entry.active += 1; + entry.touched = Instant::now(); + Ok(Active { + service: self.clone(), + id: id.into(), + receiver: entry.receiver.clone(), + }) + } + + async fn route(self: &Arc, request: Request) -> HttpResult { + self.reap(); + if request.method == "POST" && request.path == "/v1/connections" { + let _: Empty = decode(&request.body)?; + let mut entries = self.entries.lock().unwrap(); + if entries.len() >= MAX_CONNECTIONS { + return Err((503, "connection_limit")); + } + let (receiver, public_session) = + (self.factory)().map_err(|_| (503, "session_unavailable"))?; + let id = uuid::Uuid::new_v4().to_string(); + let response = json!({"connection_id": id, "generation": receiver.generation(), "capabilities": receiver.capabilities(), "public_session": public_session}); + entries.insert( + id, + Entry { + receiver, + touched: Instant::now(), + active: 0, + }, + ); + return Ok(response); + } + let suffix = request + .path + .strip_prefix("/v1/connections/") + .ok_or((404, "unknown_route"))?; + let parts: Vec<_> = suffix.split('/').collect(); + let id = parts[0]; + if uuid::Uuid::parse_str(id).is_err() { + return Err((404, "unknown_route")); + } + let valid_route = matches!( + (request.method.as_str(), parts.as_slice()), + ("POST", [_, "exchange"]) | ("POST", [_, "cancel"]) | ("DELETE", [_]) + ); + if !valid_route { + return Err((404, "unknown_route")); + } + let generation = request + .generation + .as_deref() + .ok_or((400, "generation_required"))?; + let active = self.lookup(id, generation)?; + match (request.method.as_str(), parts.as_slice()) { + ("POST", [_, "exchange"]) => { + // Do not let queued actions consume every HTTP task: control + // requests must still be admitted while exchanges are waiting. + let _permit = self + .exchanges + .try_acquire() + .map_err(|_| (429, "exchange_limit"))?; + let envelope: DriverRequestEnvelope = decode(&request.body)?; + let mut response = active.receiver.exchange(generation, envelope).await; + if !response_fits(&response) { + active.receiver.close(); + response.ok = false; + response.result = None; + response.error = Some( + "Driver response exceeded carrier limit; completion is unknown".into(), + ); + response.error_code = Some("response_too_large".into()); + response.completion_known = false; + } + serde_json::to_value(response).map_err(|_| (500, "serialization_failed")) + } + ("POST", [_, "cancel"]) => { + let cancel: Cancel = decode(&request.body)?; + if cancel.request_id.is_empty() || cancel.request_id.len() > 256 { + return Err((400, "invalid_request_id")); + } + active + .receiver + .cancel(generation, &cancel.request_id) + .map_err(|_| (409, "cancel_failed"))?; + Ok(json!({"ok": true})) + } + ("DELETE", [_]) => { + if !request.body.is_empty() { + let _: Empty = decode(&request.body)?; + } + // Retain the receiver's closed ledger until idle removal. + active.receiver.close(); + Ok(json!({"ok": true})) + } + _ => unreachable!(), + } + } +} + +fn response_fits(value: &impl serde::Serialize) -> bool { + // Count serialized bytes without allocating another copy of a potentially + // large native result, stopping as soon as the wire limit is exceeded. + struct Budget(usize); + impl std::io::Write for Budget { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + if bytes.len() > self.0 { + return Err(std::io::Error::other("response_too_large")); + } + self.0 -= bytes.len(); + Ok(bytes.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + serde_json::to_writer(Budget(MAX_RESPONSE), value).is_ok() +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct Empty {} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct Cancel { + request_id: String, +} + +fn decode(body: &[u8]) -> HttpResult { + serde_json::from_slice(body).map_err(|_| (400, "invalid_json")) +} + +#[derive(Debug)] +struct Request { + method: String, + path: String, + generation: Option, + body: Vec, +} + +fn parse_headers(bytes: &[u8]) -> HttpResult<(Request, usize)> { + let mut headers = [httparse::EMPTY_HEADER; 64]; + let mut parsed = httparse::Request::new(&mut headers); + if !parsed + .parse(bytes) + .map_err(|_| (400, "invalid_http"))? + .is_complete() + { + return Err((400, "incomplete_headers")); + } + if parsed.version != Some(1) { + return Err((400, "http_1_1_required")); + } + let method = parsed.method.ok_or((400, "method_required"))?; + if !matches!(method, "POST" | "DELETE") { + return Err((405, "unsupported_method")); + } + let path = parsed.path.ok_or((400, "path_required"))?; + if path.contains(['?', '#', '%']) || !path.starts_with('/') { + return Err((400, "invalid_path")); + } + let mut length = None; + let mut generation = None; + for header in parsed.headers.iter() { + if header.name.eq_ignore_ascii_case("origin") { + return Err((403, "browser_origin_forbidden")); + } + if header.name.eq_ignore_ascii_case("transfer-encoding") { + return Err((400, "transfer_encoding_forbidden")); + } + if header.name.eq_ignore_ascii_case("content-length") { + if length.is_some() + || header.value.is_empty() + || !header.value.iter().all(u8::is_ascii_digit) + { + return Err((400, "invalid_content_length")); + } + let value = + std::str::from_utf8(header.value).map_err(|_| (400, "invalid_content_length"))?; + length = Some( + value + .parse::() + .map_err(|_| (413, "body_too_large"))?, + ); + } + if header.name.eq_ignore_ascii_case("x-cua-driver-generation") { + if generation.is_some() || header.value.is_empty() || header.value.len() > 128 { + return Err((400, "invalid_generation")); + } + generation = Some( + std::str::from_utf8(header.value) + .map_err(|_| (400, "invalid_generation"))? + .to_owned(), + ); + } + } + let length = match length { + Some(n) => n, + None if method == "DELETE" => 0, + None => return Err((411, "content_length_required")), + }; + if length > MAX_BODY { + return Err((413, "body_too_large")); + } + Ok(( + Request { + method: method.into(), + path: path.into(), + generation, + body: Vec::new(), + }, + length, + )) +} + +async fn read_request(reader: &mut (impl AsyncRead + Unpin)) -> HttpResult { + let mut bytes = Vec::new(); + let header_end = loop { + if let Some(index) = bytes.windows(4).position(|w| w == b"\r\n\r\n") { + break index + 4; + } + if bytes.len() >= MAX_HEADERS { + return Err((431, "headers_too_large")); + } + let mut chunk = [0; 1024]; + let remaining = (MAX_HEADERS - bytes.len()).min(chunk.len()); + let n = reader + .read(&mut chunk[..remaining]) + .await + .map_err(|_| (400, "read_failed"))?; + if n == 0 { + return Err((400, "incomplete_headers")); + } + bytes.extend_from_slice(&chunk[..n]); + }; + let (mut request, length) = parse_headers(&bytes[..header_end])?; + request + .body + .extend_from_slice(&bytes[header_end..bytes.len().min(header_end + length)]); + let received = request.body.len(); + request.body.resize(length, 0); + reader + .read_exact(&mut request.body[received..]) + .await + .map_err(|_| (400, "incomplete_body"))?; + Ok(request) +} + +pub fn configured_port() -> anyhow::Result> { + match std::env::var("CUA_DRIVER_ENVELOPE_HTTP_PORT") { + Err(std::env::VarError::NotPresent) => Ok(None), + Ok(value) => { + let port = value.parse::()?; + anyhow::ensure!(port != 0, "CUA_DRIVER_ENVELOPE_HTTP_PORT must be nonzero"); + Ok(Some(port)) + } + Err(error) => Err(error.into()), + } +} + +pub struct Server(tokio::task::JoinHandle<()>); +impl Drop for Server { + fn drop(&mut self) { + self.0.abort(); + } +} + +pub async fn start(sdk: Arc, port: u16) -> anyhow::Result { + let listener = TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, port)).await?; + let service = Arc::new(Service { + entries: Mutex::new(HashMap::new()), + factory: Arc::new(move || sdk.create_envelope_receiver()), + exchanges: tokio::sync::Semaphore::new(MAX_EXCHANGES), + }); + let task = tokio::spawn(async move { + let permits = Arc::new(tokio::sync::Semaphore::new(64)); + let mut tasks = tokio::task::JoinSet::new(); + let mut reaper = tokio::time::interval(Duration::from_secs(30)); + loop { + tokio::select! { + accepted = listener.accept() => { + let Ok((mut stream, _)) = accepted else { break }; + let Ok(permit) = permits.clone().try_acquire_owned() else { continue }; + let service = service.clone(); + tasks.spawn(async move { + let _permit = permit; + let result = match tokio::time::timeout(IO_TIMEOUT, read_request(&mut stream)).await { + Ok(Ok(request)) => service.route(request).await, + Ok(Err(error)) => Err(error), + Err(_) => Err((408, "read_timeout")), + }; + let (status, body) = match result { Ok(body) => (200, body), Err((status, error)) => (status, json!({"error": error})) }; + let body = body.to_string(); + let response = format!("HTTP/1.1 {status} Response\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()); + let _ = tokio::time::timeout(IO_TIMEOUT, stream.write_all(response.as_bytes())).await; + }); + } + _ = reaper.tick() => service.reap(), + _ = tasks.join_next(), if !tasks.is_empty() => {} + } + } + }); + eprintln!("Private Driver envelope HTTP listening on 127.0.0.1:{port}"); + Ok(Server(task)) +} + +#[cfg(test)] +mod tests { + use super::*; + use cua_driver_sdk::{remote_receiver::DriverEnvelopeExecutor, DriverError}; + use std::sync::atomic::{AtomicUsize, Ordering}; + + struct Fake(Arc); + #[async_trait::async_trait] + impl DriverEnvelopeExecutor for Fake { + async fn metadata(&self) -> Result { + Ok(json!({"raw": [1, {"x": true}]})) + } + async fn list_tools(&self) -> Result { + Ok(json!({"tools": []})) + } + async fn call(&self, _: String, _: Value) -> Result { + Ok(json!({"native": true})) + } + fn close(&self) { + self.0.fetch_add(1, Ordering::SeqCst); + } + } + fn service() -> (Arc, Arc) { + let closed = Arc::new(AtomicUsize::new(0)); + let count = closed.clone(); + ( + Arc::new(Service { + entries: Mutex::new(HashMap::new()), + exchanges: tokio::sync::Semaphore::new(MAX_EXCHANGES), + factory: Arc::new(move || { + Ok(( + DriverEnvelopeReceiver::new(Arc::new(Fake(count.clone()))), + "test-session".into(), + )) + }), + }), + closed, + ) + } + fn request(method: &str, path: &str, generation: Option<&str>, body: Value) -> Request { + Request { + method: method.into(), + path: path.into(), + generation: generation.map(str::to_owned), + body: body.to_string().into_bytes(), + } + } + async fn create(service: &Arc) -> Value { + service + .route(request("POST", "/v1/connections", None, json!({}))) + .await + .unwrap() + } + + struct SpecialFake { + oversized: bool, + } + #[async_trait::async_trait] + impl DriverEnvelopeExecutor for SpecialFake { + async fn metadata(&self) -> Result { + if self.oversized { + Ok(json!({"data": "x".repeat(MAX_RESPONSE)})) + } else { + std::future::pending().await + } + } + async fn list_tools(&self) -> Result { + unreachable!() + } + async fn call(&self, _: String, _: Value) -> Result { + unreachable!() + } + fn close(&self) {} + } + fn special_service(oversized: bool) -> Arc { + Arc::new(Service { + entries: Mutex::new(HashMap::new()), + exchanges: tokio::sync::Semaphore::new(MAX_EXCHANGES), + factory: Arc::new(move || { + Ok(( + DriverEnvelopeReceiver::new(Arc::new(SpecialFake { oversized })), + "test-session".into(), + )) + }), + }) + } + fn metadata_envelope(id: &str) -> Value { + json!({"envelope_version":1,"request_id":id,"operation":"metadata","deadline_unix_ms": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_millis()+10000}) + } + + #[tokio::test] + async fn saturated_exchanges_leave_cancel_and_close_admitted() { + let service = special_service(false); + let mut tasks = tokio::task::JoinSet::new(); + let mut connections = Vec::new(); + for _ in 0..MAX_EXCHANGES { + let connection = create(&service).await; + let path = format!( + "/v1/connections/{}/exchange", + connection["connection_id"].as_str().unwrap() + ); + let request = request( + "POST", + &path, + connection["generation"].as_str(), + metadata_envelope("pending"), + ); + let service = service.clone(); + tasks.spawn(async move { service.route(request).await }); + connections.push(connection); + } + tokio::time::timeout(Duration::from_secs(1), async { + while service.exchanges.available_permits() != 0 { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + let first = &connections[0]; + let path = format!( + "/v1/connections/{}", + first["connection_id"].as_str().unwrap() + ); + assert_eq!( + service + .route(request( + "POST", + &format!("{path}/exchange"), + first["generation"].as_str(), + metadata_envelope("extra") + )) + .await + .unwrap_err(), + (429, "exchange_limit") + ); + service + .route(request( + "POST", + &format!("{path}/cancel"), + first["generation"].as_str(), + json!({"request_id":"pending"}), + )) + .await + .unwrap(); + for connection in &connections { + let path = format!( + "/v1/connections/{}", + connection["connection_id"].as_str().unwrap() + ); + service + .route(request( + "DELETE", + &path, + connection["generation"].as_str(), + json!({}), + )) + .await + .unwrap(); + } + tokio::time::timeout(Duration::from_secs(1), async { + while let Some(result) = tasks.join_next().await { + assert!(result.unwrap().is_ok()); + } + }) + .await + .unwrap(); + assert_eq!(service.exchanges.available_permits(), MAX_EXCHANGES); + } + + #[tokio::test] + async fn oversized_response_closes_connection_with_unknown_completion() { + let service = special_service(true); + let connection = create(&service).await; + let path = format!( + "/v1/connections/{}/exchange", + connection["connection_id"].as_str().unwrap() + ); + let response = service + .route(request( + "POST", + &path, + connection["generation"].as_str(), + metadata_envelope("large"), + )) + .await + .unwrap(); + assert_eq!(response["error_code"], "response_too_large"); + assert_eq!(response["completion_known"], false); + assert_eq!(response["ok"], false); + assert!(response.get("result").is_none()); + assert!(response_fits(&response)); + let response = service + .route(request( + "POST", + &path, + connection["generation"].as_str(), + metadata_envelope("after"), + )) + .await + .unwrap(); + assert_eq!(response["error_code"], "connection_closed"); + assert!(response_fits(&"x".repeat(MAX_RESPONSE - 2))); + assert!(!response_fits(&"x".repeat(MAX_RESPONSE - 1))); + } + + #[test] + fn strict_headers() { + for header in [ + "Origin: null\r\n", + "Transfer-Encoding: chunked\r\n", + "Content-Length: 0\r\n", + "X-Cua-Driver-Generation: a\r\nX-Cua-Driver-Generation: b\r\n", + ] { + let bytes = + format!("POST /v1/connections HTTP/1.1\r\nContent-Length: 2\r\n{header}\r\n"); + assert!(parse_headers(bytes.as_bytes()).is_err(), "{header}"); + } + for line in [ + "GET /v1/connections HTTP/1.1", + "POST /v1/connections?x HTTP/1.1", + "POST /v1/connections HTTP/1.0", + ] { + assert!( + parse_headers(format!("{line}\r\nContent-Length: 2\r\n\r\n").as_bytes()).is_err() + ); + } + for length in ["-1", "+1", "1,1", "1048577", "999999999999999999999999"] { + assert!(parse_headers( + format!("POST /v1/connections HTTP/1.1\r\nContent-Length: {length}\r\n\r\n") + .as_bytes() + ) + .is_err()); + } + } + + #[tokio::test] + async fn bounded_read_and_unary_pipelining() { + let data = b"POST /v1/connections HTTP/1.1\r\nContent-Length: 2\r\n\r\n{}POST /ignored HTTP/1.1\r\n\r\n"; + let result = read_request(&mut &data[..]).await.unwrap(); + assert_eq!(result.body, b"{}"); + assert_eq!(result.path, "/v1/connections"); + assert_eq!( + read_request(&mut &vec![b'x'; MAX_HEADERS + 1][..]) + .await + .unwrap_err() + .0, + 431 + ); + assert!(read_request( + &mut &b"POST /v1/connections HTTP/1.1\r\nContent-Length: 2\r\n\r\n{"[..] + ) + .await + .is_err()); + } + + #[tokio::test] + async fn generation_required_on_every_connection_route() { + let (service, closed) = service(); + let connection = create(&service).await; + let id = connection["connection_id"].as_str().unwrap(); + for (method, suffix, body) in [ + ("POST", "/exchange", json!({})), + ("POST", "/cancel", json!({"request_id":"r"})), + ("DELETE", "", json!({})), + ] { + let path = format!("/v1/connections/{id}{suffix}"); + assert_eq!( + service + .route(request(method, &path, None, body.clone())) + .await + .unwrap_err() + .0, + 400 + ); + assert_eq!( + service + .route(request(method, &path, Some("stale"), body)) + .await + .unwrap_err() + .0, + 409 + ); + } + assert_eq!(closed.load(Ordering::SeqCst), 0); + assert!(service + .route(request( + "POST", + &format!("/v1/connections/{id}/bound_session"), + None, + json!({}) + )) + .await + .is_err()); + } + + #[tokio::test] + async fn raw_result_and_closed_ledger() { + let (service, closed) = service(); + let connection = create(&service).await; + assert_eq!(connection["public_session"], "test-session"); + let path = format!( + "/v1/connections/{}", + connection["connection_id"].as_str().unwrap() + ); + let generation = connection["generation"].as_str(); + let envelope = json!({"envelope_version":1,"request_id":"first","operation":"metadata","deadline_unix_ms": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_millis()+10000}); + let result = service + .route(request( + "POST", + &format!("{path}/exchange"), + generation, + envelope.clone(), + )) + .await + .unwrap(); + assert_eq!(result["result"], json!({"raw": [1, {"x": true}]})); + service + .route(request("DELETE", &path, generation, json!({}))) + .await + .unwrap(); + let result = service + .route(request( + "POST", + &format!("{path}/exchange"), + generation, + envelope, + )) + .await + .unwrap(); + assert_eq!(result["error_code"], "connection_closed"); + assert_eq!(closed.load(Ordering::SeqCst), 1); + assert_eq!(service.entries.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn capacity_and_active_reaping() { + let (service, closed) = service(); + let first = create(&service).await; + for _ in 1..MAX_CONNECTIONS { + create(&service).await; + } + assert_eq!( + service + .route(request("POST", "/v1/connections", None, json!({}))) + .await + .unwrap_err() + .0, + 503 + ); + let active = service + .lookup( + first["connection_id"].as_str().unwrap(), + first["generation"].as_str().unwrap(), + ) + .unwrap(); + for entry in service.entries.lock().unwrap().values_mut() { + entry.touched = Instant::now() - IDLE; + } + service.reap(); + assert_eq!(service.entries.lock().unwrap().len(), 1); + assert_eq!(closed.load(Ordering::SeqCst), MAX_CONNECTIONS - 1); + drop(active); + assert_eq!(service.entries.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn creation_rejects_wire_session_options() { + let (service, _) = service(); + assert!(service + .route(request( + "POST", + "/v1/connections", + None, + json!({"mode":"unrestricted"}) + )) + .await + .is_err()); + assert!(service.entries.lock().unwrap().is_empty()); + } +} diff --git a/libs/cua-driver/rust/crates/cua-driver/src/main.rs b/libs/cua-driver/rust/crates/cua-driver/src/main.rs index 7ce3e7c07..bd836f49e 100644 --- a/libs/cua-driver/rust/crates/cua-driver/src/main.rs +++ b/libs/cua-driver/rust/crates/cua-driver/src/main.rs @@ -22,6 +22,7 @@ mod bundle; mod check_update_tool; mod cli; mod doctor; +mod driver_service_http; mod history_runtime; mod mcp_http; mod private_worker; diff --git a/libs/cua-driver/rust/crates/cua-driver/src/sdk_adapter.rs b/libs/cua-driver/rust/crates/cua-driver/src/sdk_adapter.rs index 99dea9815..9d1283331 100644 --- a/libs/cua-driver/rust/crates/cua-driver/src/sdk_adapter.rs +++ b/libs/cua-driver/rust/crates/cua-driver/src/sdk_adapter.rs @@ -57,6 +57,34 @@ pub struct SdkAdapter { } impl SdkAdapter { + pub fn create_envelope_receiver( + &self, + ) -> Result< + ( + Arc, + String, + ), + String, + > { + // The private HTTP slice requests only Standard; the runtime's immutable + // ceiling rejects incompatible hosts rather than widening their policy. + let public_session = format!("http-{}", uuid::Uuid::new_v4()); + let options = TrustedSessionOptions { + public_session: public_session.clone(), + mode: cua_driver_sdk::SessionPermissionMode::Standard, + ttl_seconds: 3600, + idle_ttl_seconds: 300, + capability_manifest_path: None, + bounded_manifest_path: None, + }; + cua_driver_sdk::remote_receiver::DriverEnvelopeReceiver::for_driver( + self.driver.clone(), + options, + ) + .map(|receiver| (receiver, public_session)) + .map_err(|error| error.to_string()) + } + pub async fn load(driver: Arc) -> anyhow::Result> { let tools_json = driver .list_tools_json() diff --git a/libs/cua-driver/rust/crates/cua-driver/src/serve.rs b/libs/cua-driver/rust/crates/cua-driver/src/serve.rs index d9e7f3953..738427804 100644 --- a/libs/cua-driver/rust/crates/cua-driver/src/serve.rs +++ b/libs/cua-driver/rust/crates/cua-driver/src/serve.rs @@ -1019,6 +1019,11 @@ pub async fn run_serve( crate::mcp_http::spawn(sdk.clone(), port)?; } + let _envelope_http = match crate::driver_service_http::configured_port()? { + Some(port) => Some(crate::driver_service_http::start(sdk.clone(), port).await?), + None => None, + }; + loop { tokio::select! { result = listener.accept() => { @@ -1731,6 +1736,11 @@ pub async fn run_serve( crate::mcp_http::spawn(sdk.clone(), port)?; } + let _envelope_http = match crate::driver_service_http::configured_port()? { + Some(port) => Some(crate::driver_service_http::start(sdk.clone(), port).await?), + None => None, + }; + let mut first_pipe = true; loop { // All daemons use the current-user descriptor. Embedded daemons also