diff --git a/doc/architecture/paperclip-runner-daytona-networking.md b/doc/architecture/paperclip-runner-daytona-networking.md index a2a3d72945..8ce5bcf9e1 100644 --- a/doc/architecture/paperclip-runner-daytona-networking.md +++ b/doc/architecture/paperclip-runner-daytona-networking.md @@ -20,6 +20,8 @@ The PRP identity binds company, issue, agent, run, environment lease, runner ins - Allow the sandbox provider's private preview proxy to reach runnerd on TCP 43127. Do not expose the port with a public sandbox or signed URL. - Allow the sandbox outbound access only to destinations explicitly required by the provider runtime. - Use Daytona's `wss://` preview URL with normal certificate and hostname validation. The preview token is sent only as the `X-Daytona-Preview-Token` header by Paperclip. +- Runnerd rejects public plaintext `ws://` destinations. Public dial targets require `wss://` with hostname validation and either the platform trust store or an explicitly staged private CA bundle. +- Listener mode binds only `0.0.0.0:43127`, accepts only the run-specific `/api/runner/v1/connect/:runId` path, and rejects WebSocket extension negotiation. PRP authentication and application-layer secure frames remain mandatory above the preview-proxy hop. - Do not put the bootstrap ticket in argv, files, provider environment, logs, or model context. Inject it into runnerd's initial environment/secret channel; runnerd already removes it from its environment immediately. - Runnerd is the only process allowed to reach PRP. The provider communicates with runnerd over inherited pipes. diff --git a/packages/paperclip-runner/docs/protocol-compatibility.md b/packages/paperclip-runner/docs/protocol-compatibility.md index 8fa3497d02..27dba20b7f 100644 --- a/packages/paperclip-runner/docs/protocol-compatibility.md +++ b/packages/paperclip-runner/docs/protocol-compatibility.md @@ -187,8 +187,9 @@ These envelopes are local Local runner implementation contracts. ## Durable wire rules -- The runner opens an outbound WebSocket and sends PRP v1 `hello` before any - command result or event. +- The runner opens loopback `ws://` or hostname-verified `wss://`, or accepts a + preview-proxy connection on its fixed listener, and completes the PRP v1 + authenticated handshake before any command result or event. - A one-use bootstrap bearer capability returns a short-lived connection lease in `welcome`. Later connections use that lease. Neither raw capability is durable state. @@ -205,8 +206,16 @@ These envelopes are local Local runner implementation contracts. - Frames are bounded at 1 MiB and upgrade headers at 16 KiB. Unknown or invalid required protocol data fails closed; malformed JSON is a bounded diagnostic. -These are package-local Durable recovery rules. Production TLS, control-plane admission, -and deployment policy remain separately reviewed work. +Runnerd build-metadata contract v2 advertises the exact transport inventory: +`dial_ws_loopback`, `dial_wss`, and `listen_ws`. Plaintext dial destinations +must resolve entirely to loopback. Public dial targets require TLS trust and +hostname validation; a private CA bundle augments the platform roots and must +be a bounded, private, regular file. Listener mode is fixed to port 43127 and a +single run-bound path. All modes retain the same message/frame bounds and PRP +authentication. + +These are package-local Durable recovery and transport rules. Control-plane +admission and deployment policy remain separately reviewed work. ## Change policy diff --git a/packages/paperclip-runner/runner/Cargo.lock b/packages/paperclip-runner/runner/Cargo.lock index c429d5a3c2..654e59b517 100644 --- a/packages/paperclip-runner/runner/Cargo.lock +++ b/packages/paperclip-runner/runner/Cargo.lock @@ -126,6 +126,16 @@ version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" +[[package]] +name = "cc" +version = "1.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273" +dependencies = [ + "find-msvc-tools", + "shlex", +] + [[package]] name = "cfg-if" version = "1.0.4" @@ -142,6 +152,22 @@ dependencies = [ "inout", ] +[[package]] +name = "core-foundation" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b2a6cd9ae233e7f62ba4e9353e81a88df7fc8a5987b8d445b4d90c879bd156f6" +dependencies = [ + "core-foundation-sys", + "libc", +] + +[[package]] +name = "core-foundation-sys" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" + [[package]] name = "cpufeatures" version = "0.2.17" @@ -225,6 +251,12 @@ dependencies = [ "regex-syntax", ] +[[package]] +name = "find-msvc-tools" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890" + [[package]] name = "fluent-uri" version = "0.4.1" @@ -649,6 +681,12 @@ version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" +[[package]] +name = "openssl-probe" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" + [[package]] name = "outref" version = "0.5.2" @@ -665,6 +703,9 @@ dependencies = [ "jsonschema", "num-bigint", "num-traits", + "rustls", + "rustls-native-certs", + "rustls-pemfile", "serde", "serde_json", "sha2", @@ -867,18 +908,119 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.17", + "libc", + "untrusted", + "windows-sys 0.52.0", +] + +[[package]] +name = "rustls" +version = "0.23.43" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" +dependencies = [ + "once_cell", + "ring", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-native-certs" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d" +dependencies = [ + "openssl-probe", + "rustls-pki-types", + "schannel", + "security-framework", +] + +[[package]] +name = "rustls-pemfile" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dce314e5fee3f39953d46bb63bb8a46d40c2f8fb7cc5a3b6cab2bde9721d6e50" +dependencies = [ + "rustls-pki-types", +] + +[[package]] +name = "rustls-pki-types" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" +dependencies = [ + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustversion" version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" +[[package]] +name = "schannel" +version = "0.1.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91c1b7e4904c873ef0710c1f407dde2e6287de2bebc1bbbf7d430bb7cbffd939" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "scopeguard" version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" +[[package]] +name = "security-framework" +version = "3.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d" +dependencies = [ + "bitflags", + "core-foundation", + "core-foundation-sys", + "libc", + "security-framework-sys", +] + +[[package]] +name = "security-framework-sys" +version = "2.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2691df843ecc5d231c0b14ece2acc3efb62c0a398c7e1d875f3983ce020e3" +dependencies = [ + "core-foundation-sys", + "libc", +] + [[package]] name = "serde" version = "1.0.229" @@ -944,6 +1086,12 @@ dependencies = [ "digest", ] +[[package]] +name = "shlex" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + [[package]] name = "smallvec" version = "1.15.2" @@ -1058,6 +1206,9 @@ dependencies = [ "httparse", "log", "rand", + "rustls", + "rustls-native-certs", + "rustls-pki-types", "sha1", "thiserror", "utf-8", @@ -1091,6 +1242,12 @@ dependencies = [ "subtle", ] +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + [[package]] name = "utf-8" version = "0.7.6" @@ -1191,6 +1348,88 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-sys" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" +dependencies = [ + "windows-targets", +] + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + [[package]] name = "wit-bindgen" version = "0.57.1" @@ -1267,6 +1506,12 @@ dependencies = [ "synstructure", ] +[[package]] +name = "zeroize" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" + [[package]] name = "zerotrie" version = "0.2.5" diff --git a/packages/paperclip-runner/runner/Cargo.toml b/packages/paperclip-runner/runner/Cargo.toml index d2d049bda4..8178d281c2 100644 --- a/packages/paperclip-runner/runner/Cargo.toml +++ b/packages/paperclip-runner/runner/Cargo.toml @@ -18,4 +18,7 @@ num-traits = "0.2" serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" sha2 = "0.10" -tungstenite = { version = "0.28", default-features = false, features = ["handshake"] } +rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12"] } +rustls-native-certs = "0.8" +rustls-pemfile = "2.2" +tungstenite = { version = "0.28", default-features = false, features = ["handshake", "rustls-tls-native-roots"] } diff --git a/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml b/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml index d34f35ef48..fe670825c0 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml +++ b/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml @@ -15,6 +15,9 @@ num-traits.workspace = true serde.workspace = true serde_json.workspace = true sha2.workspace = true +rustls.workspace = true +rustls-native-certs.workspace = true +rustls-pemfile.workspace = true tungstenite.workspace = true [[bin]] diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs index 62aed56ea1..acca704dd9 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs @@ -7,6 +7,27 @@ use paperclip_runner_core::durable::{ }; use paperclip_runner_core::local_runner::{run_local_runner, LocalRunnerError, RunnerConfig}; use paperclip_runner_core::provider_backend::CodexCommandExecutor; +use serde_json::json; + +const RUNNERD_BUILD_METADATA_SCHEMA: &str = "paperclip-runner/runnerd-build-metadata/v1"; + +fn build_metadata() -> serde_json::Value { + json!({ + "schema": RUNNERD_BUILD_METADATA_SCHEMA, + "binaryName": "paperclip-runnerd", + "packageName": "@paperclipai/paperclip-runner", + "packageVersion": env!("CARGO_PKG_VERSION"), + "binaryContractVersion": 2, + "nativeExecutionVersion": 1, + "harnessDriverVersion": 1, + "prp": { + "name": "paperclip.runner", + "minimumVersion": 1, + "maximumVersion": 1 + }, + "prpTransportModes": ["dial_ws_loopback", "dial_wss", "listen_ws"] + }) +} fn value(args: &[String], name: &str) -> Result { let index = args @@ -50,12 +71,52 @@ fn run_durable(args: &[String]) -> Result<(), LocalRunnerError> { optional_u64(args, name).map(|value| Duration::from_millis(value.unwrap_or(default))) }; let state_dir = PathBuf::from(value(args, "--state-dir")?); + let run_id = value(args, "--run-id")?; + let has_connect = args.iter().any(|argument| argument == "--connect-url"); + let has_listener = ["--listen-address", "--listen-port", "--listen-path"] + .iter() + .any(|name| args.iter().any(|argument| argument == name)); + let connect_url = match (has_connect, has_listener) { + (true, false) => value(args, "--connect-url")?, + (false, true) => { + let address = value(args, "--listen-address")?; + let port = value(args, "--listen-port")?; + let path = value(args, "--listen-path")?; + if address != "0.0.0.0" || port != "43127" { + return Err(LocalRunnerError::invalid( + "runner listener requires --listen-address 0.0.0.0 and --listen-port 43127", + )); + } + if path != format!("/api/runner/v1/connect/{run_id}") { + return Err(LocalRunnerError::invalid( + "runner listener path must exactly match the configured run", + )); + } + format!("listen://{address}:{port}{path}") + } + _ => { + return Err(LocalRunnerError::invalid( + "durable runner requires exactly one connect URL or complete listener group", + )) + } + }; + let ca_bundle_path = args + .iter() + .any(|argument| argument == "--ca-bundle-path") + .then(|| value(args, "--ca-bundle-path").map(PathBuf::from)) + .transpose()?; + if ca_bundle_path.is_some() && !connect_url.starts_with("wss://") { + return Err(LocalRunnerError::invalid( + "--ca-bundle-path is accepted only with wss://", + )); + } let config = DurableRunnerConfig { - connect_url: value(args, "--connect-url")?, + connect_url, + ca_bundle_path, state_dir: state_dir.clone(), runner_instance_id: value(args, "--runner-id")?, environment_lease_id: value(args, "--environment-lease-id")?, - run_id: value(args, "--run-id")?, + run_id, normalized_session_id: value(args, "--session-id")?, turn_id: value(args, "--turn-id")?, item_id: value(args, "--item-id")?, @@ -65,6 +126,7 @@ fn run_durable(args: &[String]) -> Result<(), LocalRunnerError> { p0_reserve_bytes: usize_value(args, "--p0-reserve-bytes", 1024 * 1024)?, max_frame_bytes: usize_value(args, "--max-frame-bytes", 1024 * 1024)?, reconnect_delay: duration("--reconnect-delay-ms", 250)?, + reconnect_grace: optional_u64(args, "--reconnect-grace-ms")?.map(Duration::from_millis), max_runtime: duration("--max-runtime-ms", 60 * 60 * 1000)?, }; let executor = CodexCommandExecutor::with_runner_config(state_dir, &config); @@ -74,7 +136,15 @@ fn run_durable(args: &[String]) -> Result<(), LocalRunnerError> { fn run() -> Result<(), LocalRunnerError> { let args = std::env::args().skip(1).collect::>(); - if args.iter().any(|argument| argument == "--connect-url") { + if args.as_slice() == ["--build-metadata"] { + println!("{}", build_metadata()); + return Ok(()); + } + if args.iter().any(|argument| argument == "--connect-url") + || args.iter().any(|argument| argument == "--listen-address") + || args.iter().any(|argument| argument == "--listen-port") + || args.iter().any(|argument| argument == "--listen-path") + { return run_durable(&args); } run_local_runner(RunnerConfig { @@ -95,6 +165,22 @@ fn run() -> Result<(), LocalRunnerError> { }) } +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn build_metadata_advertises_the_remote_transport_contract() { + let metadata = build_metadata(); + assert_eq!(metadata["schema"], RUNNERD_BUILD_METADATA_SCHEMA); + assert_eq!(metadata["binaryContractVersion"], 2); + assert_eq!( + metadata["prpTransportModes"], + json!(["dial_ws_loopback", "dial_wss", "listen_ws"]) + ); + } +} + fn main() -> ExitCode { match run() { Ok(()) => ExitCode::SUCCESS, diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs index 920a75770d..539ef5e944 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs @@ -98,6 +98,7 @@ pub fn capture_bootstrap_ticket() -> Result, DurableRunn #[derive(Clone, Debug, PartialEq, Eq)] pub struct DurableRunnerConfig { pub connect_url: String, + pub ca_bundle_path: Option, pub state_dir: PathBuf, pub runner_instance_id: String, pub environment_lease_id: String, @@ -111,6 +112,7 @@ pub struct DurableRunnerConfig { pub p0_reserve_bytes: usize, pub max_frame_bytes: usize, pub reconnect_delay: Duration, + pub reconnect_grace: Option, pub max_runtime: Duration, } @@ -172,6 +174,11 @@ impl DurableRunnerConfig { "reconnect delay must be between one millisecond and 60 seconds", )); } + if self.reconnect_grace.is_some_and(|grace| grace.is_zero()) { + return Err(DurableRunnerError::invalid( + "reconnect grace must be non-zero when configured", + )); + } if self.max_runtime > Duration::from_secs(7 * 24 * 60 * 60) { return Err(DurableRunnerError::invalid( "durable runner max runtime must not exceed seven days", @@ -188,6 +195,7 @@ mod tests { fn config() -> DurableRunnerConfig { DurableRunnerConfig { connect_url: "ws://127.0.0.1/runner".to_owned(), + ca_bundle_path: None, state_dir: PathBuf::from("state"), runner_instance_id: "runner-1".to_owned(), environment_lease_id: "lease-1".to_owned(), @@ -201,6 +209,7 @@ mod tests { p0_reserve_bytes: 64 * 1024, max_frame_bytes: 64 * 1024, reconnect_delay: Duration::from_millis(1), + reconnect_grace: None, max_runtime: Duration::from_secs(60), } } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs index 68587db845..190bd62185 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs @@ -1,5 +1,5 @@ use std::thread; -use std::time::Instant; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; @@ -10,7 +10,7 @@ use super::state::{ }; use super::transport::{ current_unix_ms, validate_control_identity, AuthenticatedTransport, ConnectionMetadata, - LeaseCredential, ResolvedWsTarget, + LeaseCredential, RunnerTransportEndpoint, }; use super::{BootstrapTicket, DurableRunnerConfig, DurableRunnerError, PROTOCOL, PROTOCOL_VERSION}; @@ -45,6 +45,49 @@ enum CommandLifecycle { Shutdown, } +fn sleep_for_reconnect(base: Duration, max_delay: Duration, attempt: &mut u32) { + let multiplier = 1_u128 << (*attempt).min(5); + let uncapped = base.as_millis().saturating_mul(multiplier); + let capped = uncapped.clamp(1, 5_000) as u64; + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |duration| u64::from(duration.subsec_nanos())); + let jitter_percent = 75 + nanos % 51; + *attempt = attempt.saturating_add(1); + let delay = Duration::from_millis(capped.saturating_mul(jitter_percent) / 100).min(max_delay); + if !delay.is_zero() { + thread::sleep(delay); + } +} + +fn sleep_before_deadline(delay: Duration, deadline: Instant) { + let bounded_delay = delay.min(deadline.saturating_duration_since(Instant::now())); + if !bounded_delay.is_zero() { + thread::sleep(bounded_delay); + } +} + +fn connection_attempt_deadline( + config: &DurableRunnerConfig, + started: Instant, + disconnected_since: Option, +) -> Instant { + let now = Instant::now(); + let runtime_remaining = config + .max_runtime + .saturating_sub(now.saturating_duration_since(started)); + let remaining = disconnected_since.zip(config.reconnect_grace).map_or( + runtime_remaining, + |(disconnected_at, grace)| { + runtime_remaining + .min(grace.saturating_sub(now.saturating_duration_since(disconnected_at))) + }, + ); + // Validation caps max_runtime at seven days, and reconnect grace can only + // shorten this budget, so adding it to a current Instant cannot overflow. + now + remaining +} + impl CommandLifecycle { fn for_completed(command: &Command) -> Self { match command.command_type.as_str() { @@ -112,14 +155,36 @@ pub fn run_durable_runner( )?; store.save(&state)?; } - // Resolve once. Every reconnect uses the same validated concrete addresses, - // so DNS cannot redirect a retry after the trust decision. - let target = ResolvedWsTarget::resolve(&config.connect_url)?; + // Bind listener mode or resolve dial mode before processing commands. Dial + // reconnects retain the same validated addresses so DNS cannot redirect a + // retry after the trust decision. + let endpoint = RunnerTransportEndpoint::new(&config.connect_url, &config.run_id)?; let started = Instant::now(); let mut bootstrap_ticket = Some(bootstrap_ticket); let mut lease: Option = None; + let mut authenticated_once = false; + let mut disconnected_since: Option = None; + let mut reconnect_attempt = 0_u32; loop { + if authenticated_once { + let disconnected_at = disconnected_since.get_or_insert_with(Instant::now); + if config + .reconnect_grace + .is_some_and(|grace| disconnected_at.elapsed() >= grace) + { + let _ = executor.shutdown(); + state.lifecycle = "recoverable_failure".to_owned(); + state.recoverable_failure = Some("transport_reconnect_grace_exceeded".to_owned()); + state.record_diagnostic( + "transport reconnect grace exceeded; durable state is preserved", + ); + store.save(&state)?; + return Err(DurableRunnerError::invalid( + "transport reconnect grace exceeded; durable state is preserved", + )); + } + } if started.elapsed() >= config.max_runtime { let _ = executor.shutdown(); state.lifecycle = "recoverable_failure".to_owned(); @@ -146,27 +211,51 @@ pub fn run_durable_runner( } let using_bootstrap = lease.is_none(); + let connect_deadline = connection_attempt_deadline(&config, started, disconnected_since); let connection = AuthenticatedTransport::connect( - &target, + &endpoint, &config, &state, bootstrap_ticket.as_ref(), lease.as_ref(), + connect_deadline, ); let (mut transport, welcome) = match connection { - Ok(connection) => connection, + Ok(Some(connection)) => connection, + Ok(None) => { + sleep_before_deadline(config.reconnect_delay, connect_deadline); + continue; + } Err(error) => { state.record_diagnostic(format!("transport reconnect scheduled: {error}")); store.save(&state)?; + if Instant::now() >= connect_deadline { + // Re-enter the lifecycle checks immediately so an auth + // timeout cannot be misreported as a reusable-bootstrap + // failure or delayed by reconnect backoff. + continue; + } if using_bootstrap && error.bootstrap_maybe_consumed { return Err(DurableRunnerError::invalid( "bootstrap connection failed closed; provide a fresh one-use ticket", )); } - thread::sleep(config.reconnect_delay); + sleep_for_reconnect( + config.reconnect_delay, + connect_deadline.saturating_duration_since(Instant::now()), + &mut reconnect_attempt, + ); continue; } }; + if Instant::now() >= connect_deadline { + // A transport that authenticated after its lifecycle deadline is + // never allowed to clear reconnect state or process commands. + continue; + } + authenticated_once = true; + disconnected_since = None; + reconnect_attempt = 0; if let Some(next_lease) = welcome.lease { lease = Some(next_lease); // A bootstrap capability is one-use. It is destroyed only after a @@ -207,9 +296,12 @@ pub fn run_durable_runner( return Ok(()); } if disconnected { + disconnected_since.get_or_insert_with(Instant::now); state.reconnect_count = state.reconnect_count.saturating_add(1); store.save(&state)?; - thread::sleep(config.reconnect_delay); + let reconnect_deadline = + connection_attempt_deadline(&config, started, disconnected_since); + sleep_before_deadline(config.reconnect_delay, reconnect_deadline); continue; } @@ -220,6 +312,7 @@ pub fn run_durable_runner( } poll_executor_events(&mut state, &store, &config, &mut executor)?; if let Err(error) = send_outbox(&mut transport, &state, &mut sent_source_seq) { + disconnected_since.get_or_insert_with(Instant::now); state.record_diagnostic(error.to_string()); state.reconnect_count = state.reconnect_count.saturating_add(1); store.save(&state)?; @@ -239,6 +332,7 @@ pub fn run_durable_runner( Ok(Some(message)) => message, Ok(None) => continue, Err(error) => { + disconnected_since.get_or_insert_with(Instant::now); state.record_diagnostic(error.to_string()); state.reconnect_count = state.reconnect_count.saturating_add(1); store.save(&state)?; @@ -246,6 +340,7 @@ pub fn run_durable_runner( } }; if let Err(error) = validate_control_identity(&message, &state, Some(&connection)) { + disconnected_since.get_or_insert_with(Instant::now); state.record_diagnostic(format!( "control identity mismatch closed the connection: {error}" )); @@ -276,6 +371,7 @@ pub fn run_durable_runner( .send_json(&command_result_envelope(&state, &result)) .and_then(|()| send_outbox(&mut transport, &state, &mut sent_source_seq)); if let Err(error) = delivery { + disconnected_since.get_or_insert_with(Instant::now); state.record_diagnostic(error.to_string()); state.reconnect_count = state.reconnect_count.saturating_add(1); store.save(&state)?; @@ -319,6 +415,7 @@ pub fn run_durable_runner( ))?; } _ => { + disconnected_since.get_or_insert_with(Instant::now); state.record_diagnostic( "malformed or unsupported control frame closed the connection", ); @@ -328,7 +425,9 @@ pub fn run_durable_runner( } } } - thread::sleep(config.reconnect_delay); + disconnected_since.get_or_insert_with(Instant::now); + let reconnect_deadline = connection_attempt_deadline(&config, started, disconnected_since); + sleep_before_deadline(config.reconnect_delay, reconnect_deadline); } } @@ -507,6 +606,7 @@ mod tests { fn config(directory: PathBuf) -> DurableRunnerConfig { DurableRunnerConfig { connect_url: "ws://127.0.0.1:3000/path".to_owned(), + ca_bundle_path: None, state_dir: directory, runner_instance_id: "runner_1".to_owned(), environment_lease_id: "environment_1".to_owned(), @@ -520,6 +620,7 @@ mod tests { p0_reserve_bytes: 4096, max_frame_bytes: 64 * 1024, reconnect_delay: Duration::from_millis(1), + reconnect_grace: None, max_runtime: Duration::from_secs(1), } } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs index 0e27f28f3c..febfef7cbe 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs @@ -1197,6 +1197,7 @@ mod tests { fn config(state_dir: PathBuf) -> DurableRunnerConfig { DurableRunnerConfig { connect_url: "ws://127.0.0.1:3000/api/runner/v1/connect/run_1".to_owned(), + ca_bundle_path: None, state_dir, runner_instance_id: "runner_1".to_owned(), environment_lease_id: "environment_1".to_owned(), @@ -1210,6 +1211,7 @@ mod tests { p0_reserve_bytes: 4096, max_frame_bytes: 65_536, reconnect_delay: Duration::from_millis(1), + reconnect_grace: None, max_runtime: Duration::from_secs(1), } } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs index 5d7a9d37d0..6e42875ce6 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs @@ -1,6 +1,9 @@ -use std::io; -use std::net::{SocketAddr, TcpStream, ToSocketAddrs}; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::io::{self, BufReader, Read, Write}; +use std::net::{SocketAddr, TcpListener, TcpStream, ToSocketAddrs}; +use std::path::Path; +use std::sync::Arc; +use std::thread; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use aes_gcm::aead::{Aead, Payload}; use aes_gcm::{Aes256Gcm, KeyInit, Nonce}; @@ -8,22 +11,31 @@ use hmac::{Hmac, Mac}; use serde::Deserialize; use serde_json::{json, Value}; use sha2::{Digest, Sha256}; -use tungstenite::client::{client_with_config, IntoClientRequest}; +use tungstenite::client::IntoClientRequest; +use tungstenite::handshake::server::ErrorResponse; +use tungstenite::handshake::HandshakeError; use tungstenite::protocol::WebSocketConfig; -use tungstenite::{Message, WebSocket}; +use tungstenite::stream::MaybeTlsStream; +use tungstenite::{accept_hdr_with_config, client_tls_with_config, Connector, Message, WebSocket}; -use super::state::{Command, DurableState}; +use super::state::{open_private_regular_file, Command, DurableState}; use super::{ BootstrapTicket, DurableRunnerConfig, DurableRunnerError, Secret, PROTOCOL, PROTOCOL_VERSION, }; const SECURE_FRAME_SCHEMA: &str = "paperclip.runner.secure-frame.v1"; const AUTH_TIMEOUT: Duration = Duration::from_secs(2); +const WELCOME_TIMEOUT: Duration = Duration::from_secs(2); +const RUNTIME_READ_TIMEOUT: Duration = Duration::from_millis(250); +const CONNECT_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(3); +const CONNECT_TOTAL_TIMEOUT: Duration = Duration::from_secs(10); +const MAX_CA_BUNDLE_BYTES: u64 = 4 * 1024 * 1024; type HmacSha256 = Hmac; #[derive(Clone, Debug)] struct ParsedWsUrl { + secure: bool, host: String, authority: String, port: u16, @@ -32,6 +44,7 @@ struct ParsedWsUrl { #[derive(Clone, Debug)] pub(crate) struct ResolvedWsTarget { + secure: bool, authority: String, path: String, addresses: Vec, @@ -47,14 +60,21 @@ impl ResolvedWsTarget { } fn request_url(&self) -> String { - format!("ws://{}{}", self.authority, self.path) + let scheme = if self.secure { "wss" } else { "ws" }; + format!("{scheme}://{}{}", self.authority, self.path) } } fn parse_ws_url(input: &str) -> Result { - let remainder = input - .strip_prefix("ws://") - .ok_or_else(|| DurableRunnerError::invalid("runner transport accepts exactly ws://"))?; + let (secure, remainder, default_port) = if let Some(value) = input.strip_prefix("ws://") { + (false, value, 80) + } else if let Some(value) = input.strip_prefix("wss://") { + (true, value, 443) + } else { + return Err(DurableRunnerError::invalid( + "runner connect URL must use ws:// or wss://", + )); + }; if remainder.is_empty() || remainder .chars() @@ -80,16 +100,26 @@ fn parse_ws_url(input: &str) -> Result { .find(']') .ok_or_else(|| DurableRunnerError::invalid("bracketed IPv6 authority is malformed"))?; let host = &authority[1..closing]; - let port = authority[closing + 1..].strip_prefix(':').ok_or_else(|| { - DurableRunnerError::invalid("bracketed IPv6 authority requires a port") - })?; + let suffix = &authority[closing + 1..]; + let port = if suffix.is_empty() { + default_port.to_string() + } else { + suffix + .strip_prefix(':') + .ok_or_else(|| { + DurableRunnerError::invalid("bracketed IPv6 authority is malformed") + })? + .to_owned() + }; host.parse::() .map_err(|_| DurableRunnerError::invalid("bracketed WebSocket host must be IPv6"))?; (host, port) } else { - let (host, port) = authority.rsplit_once(':').ok_or_else(|| { - DurableRunnerError::invalid("WebSocket URL requires an explicit port") - })?; + let (host, port) = authority + .rsplit_once(':') + .map_or((authority, default_port.to_string()), |(host, port)| { + (host, port.to_owned()) + }); if host.is_empty() || host.contains(':') { return Err(DurableRunnerError::invalid( "WebSocket host is empty or contains unbracketed IPv6", @@ -106,6 +136,7 @@ fn parse_ws_url(input: &str) -> Result { )); } Ok(ParsedWsUrl { + secure, host: host.to_owned(), authority: authority.to_owned(), port, @@ -131,18 +162,487 @@ where "WebSocket destination resolved to no addresses", )); } - if addresses.iter().any(|address| !address.ip().is_loopback()) { + if !parsed.secure && addresses.iter().any(|address| !address.ip().is_loopback()) { return Err(DurableRunnerError::invalid( - "every WebSocket destination must resolve to loopback", + "plaintext WebSocket destinations must all resolve to loopback", )); } Ok(ResolvedWsTarget { + secure: parsed.secure, authority: parsed.authority, path: parsed.path, addresses, }) } +pub(crate) enum RunnerTransportEndpoint { + Dial(ResolvedWsTarget), + Listen { listener: TcpListener, path: String }, +} + +impl RunnerTransportEndpoint { + pub(crate) fn new(input: &str, run_id: &str) -> Result { + if let Some(remainder) = input.strip_prefix("listen://") { + let (authority, path) = remainder.split_once('/').ok_or_else(|| { + DurableRunnerError::invalid( + "runner_ingress_bind_conflict: listener path is required", + ) + })?; + if authority != "0.0.0.0:43127" { + return Err(DurableRunnerError::invalid( + "runner_ingress_bind_conflict: listener must bind 0.0.0.0:43127", + )); + } + let path = format!("/{path}"); + validate_listener_path(&path)?; + if path != format!("/api/runner/v1/connect/{run_id}") { + return Err(DurableRunnerError::invalid( + "runner listener path does not match the configured run", + )); + } + let listener = TcpListener::bind(authority).map_err(|error| { + DurableRunnerError::invalid(format!( + "runner_ingress_bind_conflict: failed to bind fixed listener: {error}" + )) + })?; + listener.set_nonblocking(true).map_err(|error| { + DurableRunnerError::invalid(format!( + "runner_ingress_bind_conflict: failed to configure listener: {error}" + )) + })?; + return Ok(Self::Listen { listener, path }); + } + Ok(Self::Dial(ResolvedWsTarget::resolve(input)?)) + } + + fn open( + &self, + max_frame_bytes: usize, + ca_bundle_path: Option<&Path>, + connect_deadline: Instant, + ) -> Result, DurableRunnerError> { + ensure_connection_deadline(connect_deadline)?; + let websocket_config = || { + WebSocketConfig::default() + .max_message_size(Some(max_frame_bytes)) + .max_frame_size(Some(max_frame_bytes)) + }; + match self { + Self::Dial(target) => { + if ca_bundle_path.is_some() && !target.secure { + return Err(DurableRunnerError::invalid( + "--ca-bundle-path is accepted only with wss://", + )); + } + let stream = connect_pinned_addresses(&target.addresses, connect_deadline)?; + let request = target + .request_url() + .into_client_request() + .map_err(|error| { + DurableRunnerError::invalid(format!("invalid WebSocket request: {error}")) + })?; + let deadline = bounded_operation_deadline(connect_deadline, AUTH_TIMEOUT)?; + let connector = ca_bundle_path.map(custom_tls_connector).transpose()?; + ensure_connection_deadline(deadline)?; + stream.set_nonblocking(true).map_err(|error| { + DurableRunnerError::invalid(format!( + "WebSocket stream configuration failed: {error}" + )) + })?; + let mut handshake = + client_tls_with_config(request, stream, Some(websocket_config()), connector); + let mut socket = loop { + match handshake { + Ok((socket, _)) => break socket, + Err(HandshakeError::Interrupted(mid_handshake)) => { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(DurableRunnerError::invalid( + "WebSocket upgrade deadline elapsed", + )); + } + thread::sleep(Duration::from_millis(5).min(remaining)); + handshake = mid_handshake.handshake(); + } + Err(error) => { + return Err(DurableRunnerError::invalid(format!( + "WebSocket upgrade failed: {error}" + ))) + } + } + }; + set_dial_stream_nonblocking(socket.get_mut(), false)?; + let mut socket = RunnerSocket::Dial(socket); + socket.configure_auth_timeouts(connect_deadline)?; + Ok(Some(socket)) + } + Self::Listen { listener, path } => { + if ca_bundle_path.is_some() { + return Err(DurableRunnerError::invalid( + "--ca-bundle-path is not accepted in listener mode", + )); + } + let (stream, _) = match listener.accept() { + Ok(connection) => connection, + Err(error) if error.kind() == io::ErrorKind::WouldBlock => return Ok(None), + Err(error) => { + return Err(DurableRunnerError::invalid(format!( + "runner ingress listener accept failed: {error}" + ))) + } + }; + stream.set_nonblocking(true).map_err(|error| { + DurableRunnerError::invalid(format!( + "runner ingress stream configuration failed: {error}" + )) + })?; + let expected_path = path.clone(); + let mut handshake = accept_hdr_with_config( + stream, + move |request: &tungstenite::handshake::server::Request, response| { + if request.uri().path() != expected_path + || request.uri().query().is_some() + || request.headers().contains_key("sec-websocket-extensions") + { + let mut rejection = ErrorResponse::new(Some( + "runner listener requires the configured path without extensions" + .to_owned(), + )); + *rejection.status_mut() = tungstenite::http::StatusCode::BAD_REQUEST; + return Err(rejection); + } + Ok(response) + }, + Some(websocket_config()), + ); + let deadline = bounded_operation_deadline(connect_deadline, AUTH_TIMEOUT)?; + let mut socket = loop { + match handshake { + Ok(socket) => break socket, + Err(HandshakeError::Interrupted(mid_handshake)) => { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(DurableRunnerError::invalid( + "runner ingress WebSocket upgrade deadline elapsed", + )); + } + thread::sleep(Duration::from_millis(5).min(remaining)); + handshake = mid_handshake.handshake(); + } + Err(error) => { + return Err(DurableRunnerError::invalid(format!( + "runner ingress WebSocket upgrade failed: {error}" + ))) + } + } + }; + socket.get_mut().set_nonblocking(false).map_err(|error| { + DurableRunnerError::invalid(format!( + "runner ingress stream configuration failed: {error}" + )) + })?; + configure_auth_timeouts(socket.get_ref(), connect_deadline)?; + Ok(Some(RunnerSocket::Listen(socket))) + } + } + } +} + +fn validate_listener_path(path: &str) -> Result<(), DurableRunnerError> { + let run_id = path + .strip_prefix("/api/runner/v1/connect/") + .filter(|run_id| !run_id.is_empty() && !run_id.contains('/')); + if run_id.is_none() + || path.contains(['?', '#', '\\', '%']) + || path + .chars() + .any(|character| character.is_ascii_control() || character.is_ascii_whitespace()) + { + return Err(DurableRunnerError::invalid( + "runner listener path is invalid", + )); + } + Ok(()) +} + +fn ensure_connection_deadline(deadline: Instant) -> Result<(), DurableRunnerError> { + if Instant::now() >= deadline { + return Err(DurableRunnerError::invalid( + "transport connection lifecycle deadline elapsed", + )); + } + Ok(()) +} + +fn bounded_operation_deadline( + lifecycle_deadline: Instant, + operation_budget: Duration, +) -> Result { + ensure_connection_deadline(lifecycle_deadline)?; + Ok(lifecycle_deadline.min(Instant::now() + operation_budget)) +} + +fn configure_auth_timeouts( + stream: &TcpStream, + lifecycle_deadline: Instant, +) -> Result<(), DurableRunnerError> { + let deadline = bounded_operation_deadline(lifecycle_deadline, AUTH_TIMEOUT)?; + let timeout = deadline.saturating_duration_since(Instant::now()); + if timeout.is_zero() { + return Err(DurableRunnerError::invalid( + "transport connection lifecycle deadline elapsed", + )); + } + stream + .set_read_timeout(Some(timeout)) + .and_then(|()| stream.set_write_timeout(Some(timeout))) + .map_err(|error| DurableRunnerError::invalid(error.to_string())) +} + +fn set_dial_stream_nonblocking( + stream: &mut MaybeTlsStream, + nonblocking: bool, +) -> Result<(), DurableRunnerError> { + let result = match stream { + MaybeTlsStream::Plain(stream) => stream.set_nonblocking(nonblocking), + MaybeTlsStream::Rustls(stream) => stream.sock.set_nonblocking(nonblocking), + _ => Err(io::Error::new( + io::ErrorKind::Unsupported, + "unsupported TLS stream", + )), + }; + result.map_err(|error| { + DurableRunnerError::invalid(format!("WebSocket stream configuration failed: {error}")) + }) +} + +fn connect_pinned_addresses( + addresses: &[SocketAddr], + lifecycle_deadline: Instant, +) -> Result { + connect_pinned_addresses_with( + addresses, + lifecycle_deadline, + CONNECT_TOTAL_TIMEOUT, + CONNECT_ATTEMPT_TIMEOUT, + TcpStream::connect_timeout, + ) +} + +fn connect_pinned_addresses_with( + addresses: &[SocketAddr], + lifecycle_deadline: Instant, + total_budget: Duration, + attempt_budget: Duration, + mut connect: F, +) -> Result +where + F: FnMut(&SocketAddr, Duration) -> io::Result, +{ + if addresses.is_empty() || total_budget.is_zero() || attempt_budget.is_zero() { + return Err(DurableRunnerError::invalid( + "WebSocket connect requires pinned addresses and non-zero timeout budgets", + )); + } + let started = Instant::now(); + let transport_deadline = started + total_budget; + let deadline = lifecycle_deadline.min(transport_deadline); + if deadline <= started { + return Err(DurableRunnerError::invalid( + "WebSocket connect lifecycle deadline elapsed before dialing", + )); + } + let effective_budget = deadline.saturating_duration_since(started); + let mut ordered = addresses.to_vec(); + ordered.sort_unstable(); + ordered.dedup(); + let mut attempted = 0_usize; + let mut last_failure: Option<(SocketAddr, io::Error)> = None; + for address in &ordered { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + break; + } + attempted += 1; + match connect(address, attempt_budget.min(remaining)) { + Ok(stream) => return Ok(stream), + Err(error) => last_failure = Some((*address, error)), + } + } + let suffix = last_failure.map_or_else( + || "the total connection budget elapsed before an attempt".to_owned(), + |(address, error)| format!("last attempt {address} failed: {error}"), + ); + Err(DurableRunnerError::invalid(format!( + "WebSocket connect failed after {attempted}/{} pinned address attempts within {} ms; {suffix}", + ordered.len(), + effective_budget.as_millis(), + ))) +} + +fn custom_tls_connector(path: &Path) -> Result { + let file = open_private_regular_file(path).map_err(|error| { + DurableRunnerError::invalid(format!("failed to open private CA bundle: {error}")) + })?; + if file + .metadata() + .map_err(|error| DurableRunnerError::invalid(error.to_string()))? + .len() + > MAX_CA_BUNDLE_BYTES + { + return Err(DurableRunnerError::invalid( + "private CA bundle exceeds the 4 MiB limit", + )); + } + let native = rustls_native_certs::load_native_certs(); + let mut roots = rustls::RootCertStore::empty(); + roots.add_parsable_certificates(native.certs); + let mut reader = BufReader::new(file); + let certificates = rustls_pemfile::certs(&mut reader) + .collect::, _>>() + .map_err(|error| DurableRunnerError::invalid(format!("invalid CA bundle: {error}")))?; + if certificates.is_empty() { + return Err(DurableRunnerError::invalid( + "private CA bundle contains no certificates", + )); + } + let (added, _) = roots.add_parsable_certificates(certificates); + if added == 0 { + return Err(DurableRunnerError::invalid( + "private CA bundle contains no usable certificates", + )); + } + let config = rustls::ClientConfig::builder() + .with_root_certificates(roots) + .with_no_client_auth(); + Ok(Connector::Rustls(Arc::new(config))) +} + +enum RunnerSocket { + Dial(WebSocket>), + Listen(WebSocket), +} + +trait WebSocketMessages { + fn send_message(&mut self, message: Message) -> Result<(), tungstenite::Error>; + fn read_message(&mut self) -> Result; +} + +impl WebSocketMessages for WebSocket { + fn send_message(&mut self, message: Message) -> Result<(), tungstenite::Error> { + self.send(message) + } + + fn read_message(&mut self) -> Result { + self.read() + } +} + +impl WebSocketMessages for RunnerSocket { + fn send_message(&mut self, message: Message) -> Result<(), tungstenite::Error> { + match self { + Self::Dial(socket) => socket.send(message), + Self::Listen(socket) => socket.send(message), + } + } + + fn read_message(&mut self) -> Result { + match self { + Self::Dial(socket) => socket.read(), + Self::Listen(socket) => socket.read(), + } + } +} + +impl RunnerSocket { + fn set_read_timeout(&mut self, timeout: Option) -> io::Result<()> { + match self { + Self::Dial(socket) => match socket.get_mut() { + MaybeTlsStream::Plain(stream) => stream.set_read_timeout(timeout), + MaybeTlsStream::Rustls(stream) => stream.sock.set_read_timeout(timeout), + _ => Err(io::Error::new( + io::ErrorKind::Unsupported, + "unsupported TLS stream", + )), + }, + Self::Listen(socket) => socket.get_mut().set_read_timeout(timeout), + } + } + + fn set_write_timeout(&mut self, timeout: Option) -> io::Result<()> { + match self { + Self::Dial(socket) => match socket.get_mut() { + MaybeTlsStream::Plain(stream) => stream.set_write_timeout(timeout), + MaybeTlsStream::Rustls(stream) => stream.sock.set_write_timeout(timeout), + _ => Err(io::Error::new( + io::ErrorKind::Unsupported, + "unsupported TLS stream", + )), + }, + Self::Listen(socket) => socket.get_mut().set_write_timeout(timeout), + } + } + + #[cfg(test)] + fn read_timeout(&self) -> io::Result> { + match self { + Self::Dial(socket) => match socket.get_ref() { + MaybeTlsStream::Plain(stream) => stream.read_timeout(), + MaybeTlsStream::Rustls(stream) => stream.sock.read_timeout(), + _ => Err(io::Error::new( + io::ErrorKind::Unsupported, + "unsupported TLS stream", + )), + }, + Self::Listen(socket) => socket.get_ref().read_timeout(), + } + } + + fn configure_auth_timeouts( + &mut self, + lifecycle_deadline: Instant, + ) -> Result { + let deadline = bounded_operation_deadline(lifecycle_deadline, AUTH_TIMEOUT)?; + let timeout = deadline.saturating_duration_since(Instant::now()); + if timeout.is_zero() { + return Err(DurableRunnerError::invalid( + "transport connection lifecycle deadline elapsed", + )); + } + self.set_read_timeout(Some(timeout)) + .and_then(|()| self.set_write_timeout(Some(timeout))) + .map_err(|error| DurableRunnerError::invalid(error.to_string()))?; + Ok(deadline) + } +} + +fn send_auth_plain( + socket: &mut RunnerSocket, + value: &Value, + max_frame_bytes: usize, + lifecycle_deadline: Instant, +) -> Result<(), DurableRunnerError> { + socket.configure_auth_timeouts(lifecycle_deadline)?; + send_plain(socket, value, max_frame_bytes)?; + ensure_connection_deadline(lifecycle_deadline) +} + +fn send_auth_response_plain( + socket: &mut RunnerSocket, + value: &Value, + max_frame_bytes: usize, + lifecycle_deadline: Instant, + send_started: &mut bool, +) -> Result<(), DurableRunnerError> { + socket.configure_auth_timeouts(lifecycle_deadline)?; + let message = encode_plain_message(value, max_frame_bytes)?; + // A write error can occur after the peer receives the complete response, + // so the one-use ticket becomes ambiguous immediately before this call. + *send_started = true; + socket.send_message(message).map_err(map_websocket_error)?; + ensure_connection_deadline(lifecycle_deadline) +} + #[derive(Debug)] struct CredentialMaterial { credential_id: String, @@ -368,7 +868,7 @@ impl SecureChannel { } pub(crate) struct AuthenticatedTransport { - socket: WebSocket, + socket: RunnerSocket, secure_channel: SecureChannel, max_frame_bytes: usize, } @@ -393,22 +893,28 @@ impl ConnectFailure { } } - fn after_auth_started(error: DurableRunnerError, credential_kind: &str) -> Self { + fn after_auth_started( + error: DurableRunnerError, + credential_kind: &str, + bootstrap_may_have_been_consumed: bool, + ) -> Self { Self { error, - bootstrap_maybe_consumed: credential_kind == "bootstrap", + bootstrap_maybe_consumed: credential_kind == "bootstrap" + && bootstrap_may_have_been_consumed, } } } impl AuthenticatedTransport { pub(crate) fn connect( - target: &ResolvedWsTarget, + endpoint: &RunnerTransportEndpoint, config: &DurableRunnerConfig, state: &DurableState, bootstrap: Option<&BootstrapTicket>, lease: Option<&LeaseCredential>, - ) -> Result<(Self, Welcome), ConnectFailure> { + connect_deadline: Instant, + ) -> Result, ConnectFailure> { let (credential_token, credential_kind, expected_lease) = match (lease, bootstrap) { (Some(lease), _) => ( lease.expose().map_err(ConnectFailure::retryable)?, @@ -427,134 +933,146 @@ impl AuthenticatedTransport { } }; let credential = CredentialMaterial::from_token(credential_token); - let stream = TcpStream::connect(target.addresses.as_slice()) - .map_err(|error| { - DurableRunnerError::invalid(format!("WebSocket connect failed: {error}")) - }) - .map_err(ConnectFailure::retryable)?; - stream - .set_read_timeout(Some(AUTH_TIMEOUT)) - .and_then(|()| stream.set_write_timeout(Some(AUTH_TIMEOUT))) - .map_err(|error| DurableRunnerError::invalid(error.to_string())) - .map_err(ConnectFailure::retryable)?; - let request = target - .request_url() - .into_client_request() - .map_err(|error| { - DurableRunnerError::invalid(format!("invalid WebSocket request: {error}")) - }) - .map_err(ConnectFailure::retryable)?; - let websocket_config = WebSocketConfig::default() - .max_message_size(Some(config.max_frame_bytes)) - .max_frame_size(Some(config.max_frame_bytes)); - let (mut socket, _) = client_with_config(request, stream, Some(websocket_config)) - .map_err(|error| { - DurableRunnerError::invalid(format!("WebSocket upgrade failed: {error}")) - }) - .map_err(ConnectFailure::retryable)?; + let Some(mut socket) = endpoint + .open( + config.max_frame_bytes, + config.ca_bundle_path.as_deref(), + connect_deadline, + ) + .map_err(ConnectFailure::retryable)? + else { + return Ok(None); + }; + let mut auth_response_send_started = false; - let authenticate = - || -> Result<(Self, Welcome), DurableRunnerError> { - let client_nonce = random_nonce()?; - send_plain( - &mut socket, - &json!({ - "protocol": PROTOCOL, - "version": PROTOCOL_VERSION, - "kind": "auth_hello", - "payload": { - "credentialId": credential.credential_id, - "credentialKind": credential_kind, - "clientNonce": client_nonce, - "protocolMin": PROTOCOL_VERSION, - "protocolMax": PROTOCOL_VERSION, - "runnerInstanceId": state.runner_instance_id, - "environmentLeaseId": state.environment_lease_id, - "runId": state.run_id, - "normalizedSessionId": state.normalized_session_id, - "turnId": state.turn_id, - "itemId": state.item_id, - "runnerVersion": config.runner_version, - "runnerDigest": config.runner_digest, - "resume": { - "lastControllerCommandSeq": state.last_controller_command_seq, - "nextSourceEventSeq": state.next_source_seq, - "ackedSourceSeq": state.acked_source_seq, - }, + let authenticate = || -> Result<(Self, Welcome), DurableRunnerError> { + let client_nonce = random_nonce()?; + send_auth_plain( + &mut socket, + &json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": "auth_hello", + "payload": { + "credentialId": credential.credential_id, + "credentialKind": credential_kind, + "clientNonce": client_nonce, + "protocolMin": PROTOCOL_VERSION, + "protocolMax": PROTOCOL_VERSION, + "runnerInstanceId": state.runner_instance_id, + "environmentLeaseId": state.environment_lease_id, + "runId": state.run_id, + "normalizedSessionId": state.normalized_session_id, + "turnId": state.turn_id, + "itemId": state.item_id, + "runnerVersion": config.runner_version, + "runnerDigest": config.runner_digest, + "resume": { + "lastControllerCommandSeq": state.last_controller_command_seq, + "nextSourceEventSeq": state.next_source_seq, + "ackedSourceSeq": state.acked_source_seq, }, - }), - config.max_frame_bytes, - )?; + }, + }), + config.max_frame_bytes, + connect_deadline, + )?; - let challenge_value = receive_plain(&mut socket, config.max_frame_bytes)?; - validate_envelope_kind(&challenge_value, "auth_challenge")?; - let challenge: AuthChallenge = - serde_json::from_value(challenge_value.get("payload").cloned().ok_or_else( - || DurableRunnerError::invalid("challenge payload is required"), - )?) - .map_err(|error| { - DurableRunnerError::invalid(format!("invalid auth challenge: {error}")) - })?; - validate_challenge( - &challenge, - state, - config, - &credential, - credential_kind, - &client_nonce, - expected_lease, - )?; - let signing_bytes = challenge_signing_bytes(&challenge); - verify_hmac_hex( - &credential.auth_key, - "paperclip-runner-server-proof-v1", - &[&signing_bytes], - &challenge.server_proof, - )?; - let client_proof = hex_encode(&hmac_domain( - &credential.auth_key, - "paperclip-runner-client-proof-v1", - &[&signing_bytes, challenge.server_proof.as_bytes()], - )); - send_plain( - &mut socket, - &json!({ - "protocol": PROTOCOL, - "version": PROTOCOL_VERSION, - "kind": "auth_response", - "payload": { - "credentialId": credential.credential_id, - "clientNonce": client_nonce, - "serverNonce": challenge.server_nonce, - "clientProof": client_proof, - }, - }), - config.max_frame_bytes, - )?; - let secure_channel = SecureChannel::client( - &credential.auth_key, - &signing_bytes, - challenge.server_proof.as_bytes(), - client_proof.as_bytes(), - )?; - let mut transport = Self { - socket, - secure_channel, - max_frame_bytes: config.max_frame_bytes, - }; - transport - .socket - .get_mut() - .set_read_timeout(Some(Duration::from_millis(250))) - .map_err(|error| DurableRunnerError::invalid(error.to_string()))?; - let mut welcome_value = transport.receive_json()?.ok_or_else(|| { - DurableRunnerError::invalid("authenticated welcome timed out") - })?; - let welcome = - validate_welcome(&mut welcome_value, state, credential_kind, expected_lease)?; - Ok((transport, welcome)) + let challenge_deadline = socket.configure_auth_timeouts(connect_deadline)?; + let challenge_value = + receive_plain_until(&mut socket, config.max_frame_bytes, challenge_deadline)?; + validate_envelope_kind(&challenge_value, "auth_challenge")?; + let challenge: AuthChallenge = serde_json::from_value( + challenge_value + .get("payload") + .cloned() + .ok_or_else(|| DurableRunnerError::invalid("challenge payload is required"))?, + ) + .map_err(|error| { + DurableRunnerError::invalid(format!("invalid auth challenge: {error}")) + })?; + validate_challenge( + &challenge, + state, + config, + &credential, + credential_kind, + &client_nonce, + expected_lease, + )?; + let signing_bytes = challenge_signing_bytes(&challenge); + verify_hmac_hex( + &credential.auth_key, + "paperclip-runner-server-proof-v1", + &[&signing_bytes], + &challenge.server_proof, + )?; + let client_proof = hex_encode(&hmac_domain( + &credential.auth_key, + "paperclip-runner-client-proof-v1", + &[&signing_bytes, challenge.server_proof.as_bytes()], + )); + send_auth_response_plain( + &mut socket, + &json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": "auth_response", + "payload": { + "credentialId": credential.credential_id, + "clientNonce": client_nonce, + "serverNonce": challenge.server_nonce, + "clientProof": client_proof, + }, + }), + config.max_frame_bytes, + connect_deadline, + &mut auth_response_send_started, + )?; + let secure_channel = SecureChannel::client( + &credential.auth_key, + &signing_bytes, + challenge.server_proof.as_bytes(), + client_proof.as_bytes(), + )?; + let mut transport = Self { + socket, + secure_channel, + max_frame_bytes: config.max_frame_bytes, }; - authenticate().map_err(|error| ConnectFailure::after_auth_started(error, credential_kind)) + // Ticket validation and lease persistence happen before the server + // emits welcome. Keep that work bounded without assuming a loaded + // control-plane event loop can always respond within 250 ms. + let welcome_deadline = bounded_operation_deadline(connect_deadline, WELCOME_TIMEOUT)?; + let welcome_timeout = welcome_deadline.saturating_duration_since(Instant::now()); + if welcome_timeout.is_zero() { + return Err(DurableRunnerError::invalid( + "transport connection lifecycle deadline elapsed", + )); + } + transport + .socket + .set_read_timeout(Some(welcome_timeout)) + .map_err(|error| DurableRunnerError::invalid(error.to_string()))?; + let mut welcome_value = transport + .receive_json_until(Some(welcome_deadline))? + .ok_or_else(|| DurableRunnerError::invalid("authenticated welcome timed out"))?; + let welcome = + validate_welcome(&mut welcome_value, state, credential_kind, expected_lease)?; + // Authentication can wait longer for control-plane validation, but + // the steady-state runner loop must return to provider polling + // promptly when no control message is available. + transport + .socket + .set_read_timeout(Some(RUNTIME_READ_TIMEOUT)) + .map_err(|error| DurableRunnerError::invalid(error.to_string()))?; + ensure_connection_deadline(connect_deadline)?; + Ok((transport, welcome)) + }; + let result = authenticate(); + result.map(Some).map_err(|error| { + ConnectFailure::after_auth_started(error, credential_kind, auth_response_send_started) + }) } pub(crate) fn send_json(&mut self, value: &Value) -> Result<(), DurableRunnerError> { @@ -570,10 +1088,18 @@ impl AuthenticatedTransport { } pub(crate) fn receive_json(&mut self) -> Result, DurableRunnerError> { - let frame = match receive_plain_optional(&mut self.socket, self.max_frame_bytes)? { - Some(frame) => frame, - None => return Ok(None), - }; + self.receive_json_until(None) + } + + fn receive_json_until( + &mut self, + deadline: Option, + ) -> Result, DurableRunnerError> { + let frame = + match receive_plain_optional_until(&mut self.socket, self.max_frame_bytes, deadline)? { + Some(frame) => frame, + None => return Ok(None), + }; self.secure_channel.decrypt(&frame, false).map(Some) } } @@ -885,10 +1411,18 @@ fn required_string<'a>(value: &'a Value, field: &str) -> Result<&'a str, Durable } fn send_plain( - socket: &mut WebSocket, + socket: &mut (impl WebSocketMessages + ?Sized), value: &Value, max_frame_bytes: usize, ) -> Result<(), DurableRunnerError> { + let message = encode_plain_message(value, max_frame_bytes)?; + socket.send_message(message).map_err(map_websocket_error) +} + +fn encode_plain_message( + value: &Value, + max_frame_bytes: usize, +) -> Result { let bytes = serde_json::to_vec(value) .map_err(|error| DurableRunnerError::invalid(error.to_string()))?; if bytes.len() > max_frame_bytes { @@ -898,25 +1432,44 @@ fn send_plain( } let text = String::from_utf8(bytes).map_err(|error| DurableRunnerError::invalid(error.to_string()))?; - socket - .send(Message::Text(text.into())) - .map_err(map_websocket_error) + Ok(Message::Text(text.into())) } fn receive_plain( - socket: &mut WebSocket, + socket: &mut (impl WebSocketMessages + ?Sized), max_frame_bytes: usize, ) -> Result { - receive_plain_optional(socket, max_frame_bytes)? + receive_plain_optional_until(socket, max_frame_bytes, Some(Instant::now() + AUTH_TIMEOUT))? .ok_or_else(|| DurableRunnerError::invalid("WebSocket message timed out")) } -fn receive_plain_optional( - socket: &mut WebSocket, +fn receive_plain_until( + socket: &mut (impl WebSocketMessages + ?Sized), max_frame_bytes: usize, + deadline: Instant, +) -> Result { + receive_plain_optional_until(socket, max_frame_bytes, Some(deadline))? + .ok_or_else(|| DurableRunnerError::invalid("WebSocket message timed out")) +} + +fn receive_plain_optional_until( + socket: &mut (impl WebSocketMessages + ?Sized), + max_frame_bytes: usize, + deadline: Option, ) -> Result, DurableRunnerError> { loop { - match socket.read() { + if deadline.is_some_and(|deadline| Instant::now() >= deadline) { + return Err(DurableRunnerError::invalid( + "WebSocket authentication message deadline elapsed", + )); + } + let message = socket.read_message(); + if deadline.is_some_and(|deadline| Instant::now() >= deadline) { + return Err(DurableRunnerError::invalid( + "WebSocket authentication message deadline elapsed", + )); + } + match message { Ok(Message::Text(text)) => { if text.len() > max_frame_bytes { return Err(DurableRunnerError::invalid( @@ -930,7 +1483,7 @@ fn receive_plain_optional( }); } Ok(Message::Ping(payload)) => socket - .send(Message::Pong(payload)) + .send_message(Message::Pong(payload)) .map_err(map_websocket_error)?, Ok(Message::Pong(_)) => {} Ok(Message::Close(_)) => { @@ -1053,16 +1606,18 @@ mod tests { use std::net::{IpAddr, Ipv4Addr, TcpListener}; use std::path::PathBuf; use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::mpsc; use std::sync::Arc; use std::thread; - use tungstenite::accept; + use tungstenite::{accept, client}; use super::*; fn config(port: u16) -> DurableRunnerConfig { DurableRunnerConfig { connect_url: format!("ws://127.0.0.1:{port}/api/runner/v1/connect/run_1"), + ca_bundle_path: None, state_dir: PathBuf::from("unused"), runner_instance_id: "runner_1".to_owned(), environment_lease_id: "environment_1".to_owned(), @@ -1076,6 +1631,7 @@ mod tests { p0_reserve_bytes: 4096, max_frame_bytes: 64 * 1024, reconnect_delay: Duration::from_millis(1), + reconnect_grace: None, max_runtime: Duration::from_secs(1), } } @@ -1256,11 +1812,18 @@ mod tests { assert!( resolve_ws_target_with("ws://example.test:80/path", |_, _| Ok(vec![public])).is_err() ); + let secure = resolve_ws_target_with("wss://example.test/path", |host, port| { + assert_eq!(host, "example.test"); + assert_eq!(port, 443); + Ok(vec![public]) + }) + .unwrap(); + assert!(secure.secure); + assert_eq!(secure.request_url(), "wss://example.test/path"); for input in [ - "wss://127.0.0.1:80/path", "ws://user@127.0.0.1:80/path", - "ws://127.0.0.1/path", "ws://127.0.0.1:80/path?ticket=secret", + "https://127.0.0.1/path", ] { assert!( resolve_ws_target_with(input, |_, _| Ok(vec![])).is_err(), @@ -1269,6 +1832,310 @@ mod tests { } } + #[test] + fn pinned_connect_falls_back_in_deterministic_order_with_bounded_attempts() { + let first = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 3001); + let second = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 3002); + let mut attempts = Vec::new(); + let connected = connect_pinned_addresses_with( + &[second, first], + Instant::now() + Duration::from_secs(2), + Duration::from_secs(2), + Duration::from_millis(250), + |address, timeout| { + attempts.push((*address, timeout)); + if *address == first { + Err(io::Error::new( + io::ErrorKind::ConnectionRefused, + "first address refused", + )) + } else { + Ok("connected") + } + }, + ) + .unwrap(); + assert_eq!(connected, "connected"); + assert_eq!( + attempts.iter().map(|attempt| attempt.0).collect::>(), + vec![first, second] + ); + assert!(attempts + .iter() + .all(|attempt| attempt.1 <= Duration::from_millis(250))); + + let failure = connect_pinned_addresses_with( + &[second, first], + Instant::now() + Duration::from_secs(2), + Duration::from_secs(2), + Duration::from_millis(250), + |_, _| -> io::Result<()> { + Err(io::Error::new(io::ErrorKind::TimedOut, "simulated timeout")) + }, + ) + .unwrap_err(); + assert!(failure.to_string().contains(&second.to_string())); + assert!(failure.to_string().contains("2/2 pinned address attempts")); + + let lifecycle_budget = Duration::from_millis(40); + let mut observed_timeout = None; + connect_pinned_addresses_with( + &[first], + Instant::now() + lifecycle_budget, + Duration::from_secs(2), + Duration::from_millis(250), + |_, timeout| -> io::Result<()> { + observed_timeout = Some(timeout); + Err(io::Error::new(io::ErrorKind::TimedOut, "simulated timeout")) + }, + ) + .unwrap_err(); + assert!(observed_timeout.is_some_and(|timeout| timeout <= lifecycle_budget)); + } + + #[test] + fn authentication_io_rejects_work_past_the_lifecycle_deadline() { + struct DelayedSocket { + delay: Duration, + message: Option, + } + + impl WebSocketMessages for DelayedSocket { + fn send_message(&mut self, _message: Message) -> Result<(), tungstenite::Error> { + Ok(()) + } + + fn read_message(&mut self) -> Result { + thread::sleep(self.delay); + Ok(self.message.take().unwrap()) + } + } + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let client = TcpStream::connect(listener.local_addr().unwrap()).unwrap(); + let (stream, _) = listener.accept().unwrap(); + let lifecycle_budget = Duration::from_secs(1); + configure_auth_timeouts(&stream, Instant::now() + lifecycle_budget).unwrap(); + assert!(stream + .read_timeout() + .unwrap() + .is_some_and(|timeout| !timeout.is_zero() && timeout <= lifecycle_budget)); + assert!(stream + .write_timeout() + .unwrap() + .is_some_and(|timeout| !timeout.is_zero() && timeout <= lifecycle_budget)); + drop(client); + + let mut delayed = DelayedSocket { + delay: Duration::from_millis(5), + message: Some(Message::Text("{}".into())), + }; + let error = receive_plain_until( + &mut delayed, + 1024, + Instant::now() + Duration::from_millis(1), + ) + .unwrap_err(); + assert!(error.to_string().contains("deadline elapsed")); + } + + #[test] + fn listener_path_is_exact_and_unambiguous() { + validate_listener_path("/api/runner/v1/connect/run_1").unwrap(); + for path in [ + "/api/runner/v1/connect/", + "/api/runner/v1/connect/run_1/extra", + "/api/runner/v1/connect/run_1?ticket=secret", + "/api/runner/v1/connect/run%5f1", + ] { + assert!(validate_listener_path(path).is_err(), "{path}"); + } + } + + #[test] + fn dial_bootstrap_failure_before_auth_response_is_retryable() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let config = config(port); + let state = test_state(&config); + let server_config = config.clone(); + let server = thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + let mut socket = accept(stream).unwrap(); + receive_plain(&mut socket, server_config.max_frame_bytes).unwrap(); + send_plain( + &mut socket, + &json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": "auth_challenge", + "payload": {}, + }), + server_config.max_frame_bytes, + ) + .unwrap(); + }); + + let endpoint = RunnerTransportEndpoint::new(&config.connect_url, &config.run_id).unwrap(); + let ticket = BootstrapTicket::new("bootstrap-secret".to_owned()).unwrap(); + let failure = match AuthenticatedTransport::connect( + &endpoint, + &config, + &state, + Some(&ticket), + None, + Instant::now() + config.max_runtime, + ) { + Err(error) => error, + Ok(_) => panic!("invalid challenge unexpectedly authenticated"), + }; + assert!(!failure.bootstrap_maybe_consumed); + server.join().unwrap(); + } + + #[test] + fn dial_bootstrap_failure_after_auth_response_is_fail_closed() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let config = config(port); + let state = test_state(&config); + let server_config = config.clone(); + let server_state = state.clone(); + let server = thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + let mut socket = accept(stream).unwrap(); + server_authenticate( + &mut socket, + &server_config, + &server_state, + ServerCredential { + token: "bootstrap-secret", + kind: "bootstrap", + lease_id: None, + expires_at_unix_ms: current_unix_ms().unwrap() + 60_000, + revocation_epoch: 0, + }, + ); + // Disconnect after receiving the authenticated response but before + // welcome, when the authority may already have consumed the ticket. + }); + + let endpoint = RunnerTransportEndpoint::new(&config.connect_url, &config.run_id).unwrap(); + let ticket = BootstrapTicket::new("bootstrap-secret".to_owned()).unwrap(); + let failure = match AuthenticatedTransport::connect( + &endpoint, + &config, + &state, + Some(&ticket), + None, + Instant::now() + config.max_runtime, + ) { + Err(error) => error, + Ok(_) => panic!("connection without welcome unexpectedly authenticated"), + }; + assert!(failure.bootstrap_maybe_consumed); + server.join().unwrap(); + } + + #[test] + fn listener_rejects_invalid_peer_before_accepting_valid_bootstrap_peer() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + let address = listener.local_addr().unwrap(); + let path = "/api/runner/v1/connect/run_1".to_owned(); + let endpoint = RunnerTransportEndpoint::Listen { + listener, + path: path.clone(), + }; + let config = config(address.port()); + let state = test_state(&config); + let ticket = BootstrapTicket::new("bootstrap-secret".to_owned()).unwrap(); + + let (invalid_ready, invalid_connected) = mpsc::channel(); + let invalid_path = path.clone(); + let invalid_config = config.clone(); + let invalid_peer = thread::spawn(move || { + let stream = TcpStream::connect(address).unwrap(); + invalid_ready.send(()).unwrap(); + let (mut socket, _) = client(format!("ws://{address}{invalid_path}"), stream).unwrap(); + receive_plain(&mut socket, invalid_config.max_frame_bytes).unwrap(); + send_plain( + &mut socket, + &json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": "auth_challenge", + "payload": {}, + }), + invalid_config.max_frame_bytes, + ) + .unwrap(); + }); + invalid_connected.recv().unwrap(); + let invalid_failure = match AuthenticatedTransport::connect( + &endpoint, + &config, + &state, + Some(&ticket), + None, + Instant::now() + config.max_runtime, + ) { + Err(error) => error, + Ok(_) => panic!("invalid listener peer unexpectedly authenticated"), + }; + assert!(!invalid_failure.bootstrap_maybe_consumed); + invalid_peer.join().unwrap(); + + let (valid_ready, valid_connected) = mpsc::channel(); + let valid_config = config.clone(); + let valid_state = state.clone(); + let valid_peer = thread::spawn(move || { + let stream = TcpStream::connect(address).unwrap(); + valid_ready.send(()).unwrap(); + let (mut socket, _) = client(format!("ws://{address}{path}"), stream).unwrap(); + let expires = current_unix_ms().unwrap() + 60_000; + let mut secure = server_authenticate( + &mut socket, + &valid_config, + &valid_state, + ServerCredential { + token: "bootstrap-secret", + kind: "bootstrap", + lease_id: None, + expires_at_unix_ms: expires, + revocation_epoch: 0, + }, + ); + send_secure( + &mut socket, + &mut secure, + &valid_config, + &welcome( + &valid_state, + "connection_1", + Some("lease-secret"), + expires, + 0, + vec![], + ), + ); + }); + valid_connected.recv().unwrap(); + let (_, accepted) = AuthenticatedTransport::connect( + &endpoint, + &config, + &state, + Some(&ticket), + None, + Instant::now() + config.max_runtime, + ) + .unwrap() + .unwrap(); + assert_eq!(accepted.connection.connection_id, "connection_1"); + assert_eq!(accepted.lease.unwrap().expose().unwrap(), "lease-secret"); + valid_peer.join().unwrap(); + } + #[test] fn secure_frames_reject_replay_and_tampering() { let key = [7_u8; 32]; @@ -1419,10 +2286,22 @@ mod tests { send_plain(&mut socket, &encrypted, server_config.max_frame_bytes).unwrap(); }); - let target = ResolvedWsTarget::resolve(&config.connect_url).unwrap(); + let endpoint = RunnerTransportEndpoint::new(&config.connect_url, &config.run_id).unwrap(); let ticket = BootstrapTicket::new("bootstrap-secret".to_owned()).unwrap(); - let (_, welcome) = - AuthenticatedTransport::connect(&target, &config, &state, Some(&ticket), None).unwrap(); + let (transport, welcome) = AuthenticatedTransport::connect( + &endpoint, + &config, + &state, + Some(&ticket), + None, + Instant::now() + config.max_runtime, + ) + .unwrap() + .unwrap(); + assert_eq!( + transport.socket.read_timeout().unwrap(), + Some(RUNTIME_READ_TIMEOUT) + ); assert_eq!(welcome.connection.lease_id, "lease_1"); assert_eq!(welcome.lease.unwrap().expose().unwrap(), "lease-secret"); handle.join().unwrap(); diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs index 6fae3bd0dc..3d9fcab756 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs @@ -80,6 +80,7 @@ fn task_context_tool_set() -> AuthorizedToolSet { fn durable_config(directory: &Path) -> DurableRunnerConfig { DurableRunnerConfig { connect_url: "ws://127.0.0.1:3000/runner".to_owned(), + ca_bundle_path: None, state_dir: directory.to_path_buf(), runner_instance_id: "runner-1".to_owned(), environment_lease_id: "lease-1".to_owned(), @@ -93,6 +94,7 @@ fn durable_config(directory: &Path) -> DurableRunnerConfig { p0_reserve_bytes: 1024 * 1024, max_frame_bytes: 1024 * 1024, reconnect_delay: std::time::Duration::from_millis(1), + reconnect_grace: None, max_runtime: std::time::Duration::from_secs(5), } } diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/durable_recovery.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/durable_recovery.rs index 6ac1f3f914..30293f39e3 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/durable_recovery.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/durable_recovery.rs @@ -13,6 +13,7 @@ static NEXT_TEMPORARY_DIRECTORY: AtomicU64 = AtomicU64::new(0); fn config(state_dir: PathBuf) -> DurableRunnerConfig { DurableRunnerConfig { connect_url: "ws://127.0.0.1:3000/api/runner/v1/connect/run_1".to_owned(), + ca_bundle_path: None, state_dir, runner_instance_id: "runner_1".to_owned(), environment_lease_id: "environment_1".to_owned(), @@ -26,6 +27,7 @@ fn config(state_dir: PathBuf) -> DurableRunnerConfig { p0_reserve_bytes: 4096, max_frame_bytes: 65_536, reconnect_delay: Duration::from_millis(1), + reconnect_grace: None, max_runtime: Duration::from_secs(1), } }