diff --git a/packages/paperclip-runner/runner/Cargo.lock b/packages/paperclip-runner/runner/Cargo.lock index d4a5af60b48..e609c2e5bd9 100644 --- a/packages/paperclip-runner/runner/Cargo.lock +++ b/packages/paperclip-runner/runner/Cargo.lock @@ -2,24 +2,200 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "aead" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d122413f284cf2d62fb1b7db97e02edb8cda96d769b16e443a4f6195e35662b0" +dependencies = [ + "crypto-common", + "generic-array", +] + +[[package]] +name = "aes" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" +dependencies = [ + "cfg-if", + "cipher", + "cpufeatures", +] + +[[package]] +name = "aes-gcm" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "831010a0f742e1209b3bcea8fab6a8e149051ba6099432c8cb2cc117dec3ead1" +dependencies = [ + "aead", + "aes", + "cipher", + "ctr", + "ghash", + "subtle", +] + +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "cipher" +version = "0.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" +dependencies = [ + "crypto-common", + "inout", +] + +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "rand_core", + "typenum", +] + +[[package]] +name = "ctr" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0369ee1ad671834580515889b80f2ea915f23b8be8d0daa4bbaf2ac5c7590835" +dependencies = [ + "cipher", +] + +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", + "subtle", +] + +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "libc", + "wasi", +] + +[[package]] +name = "ghash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0d8a4362ccb29cb0b265253fb0a2728f592895ee6854fd9bc13f2ffda266ff1" +dependencies = [ + "opaque-debug", + "polyval", +] + +[[package]] +name = "hmac" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" +dependencies = [ + "digest", +] + +[[package]] +name = "inout" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01" +dependencies = [ + "generic-array", +] + [[package]] name = "itoa" version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "libc" +version = "0.2.189" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" + [[package]] name = "memchr" version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" +[[package]] +name = "opaque-debug" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" + [[package]] name = "paperclip-runner-core" version = "0.0.0" dependencies = [ + "aes-gcm", + "hmac", "serde", "serde_json", + "sha2", +] + +[[package]] +name = "polyval" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1fe60d06143b2430aa532c94cfe9e29783047f06c0d7fd359a9a51b729fa25" +dependencies = [ + "cfg-if", + "cpufeatures", + "opaque-debug", + "universal-hash", ] [[package]] @@ -40,6 +216,15 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "rand_core" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" +dependencies = [ + "getrandom", +] + [[package]] name = "serde" version = "1.0.229" @@ -83,6 +268,23 @@ dependencies = [ "zmij", ] +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + [[package]] name = "syn" version = "3.0.3" @@ -94,12 +296,40 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "unicode-ident" version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "universal-hash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc1de2c688dc15305988b563c3854064043356019f97a4b46276fe734c4f07ea" +dependencies = [ + "crypto-common", + "subtle", +] + +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + [[package]] name = "zmij" version = "1.0.23" diff --git a/packages/paperclip-runner/runner/Cargo.toml b/packages/paperclip-runner/runner/Cargo.toml index 47b88d455c1..58aeff2a863 100644 --- a/packages/paperclip-runner/runner/Cargo.toml +++ b/packages/paperclip-runner/runner/Cargo.toml @@ -9,5 +9,8 @@ license = "MIT" publish = false [workspace.dependencies] +aes-gcm = "0.10" +hmac = "0.12" serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" +sha2 = "0.10" diff --git a/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml b/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml index e3292759a2f..d5c8dc3c6cc 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml +++ b/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml @@ -6,8 +6,11 @@ license.workspace = true publish.workspace = true [dependencies] +aes-gcm.workspace = true +hmac.workspace = true serde.workspace = true serde_json.workspace = true +sha2.workspace = true [[bin]] name = "phase0-tracer" diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/phase3.rs b/packages/paperclip-runner/runner/crates/runner-core/src/phase3.rs index 74871b3b9c1..9ce6c6f0311 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/phase3.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/phase3.rs @@ -6,26 +6,36 @@ use std::io::{self, Read, Write}; use std::net::{SocketAddr, TcpStream, ToSocketAddrs}; use std::path::{Path, PathBuf}; use std::thread; -use std::time::{Duration, Instant}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; #[cfg(unix)] use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt}; +use aes_gcm::aead::{Aead, Payload}; +use aes_gcm::{Aes256Gcm, KeyInit, Nonce}; +use hmac::{Hmac, Mac}; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; use crate::phase2::SupervisedProcess; const STATE_SCHEMA: &str = "paperclip.runner.phase3.state.v1"; const PROTOCOL: &str = "paperclip.runner"; const PROTOCOL_VERSION: u64 = 1; +const SECURE_FRAME_SCHEMA: &str = "paperclip.runner.secure-frame.v1"; const STATIC_WEBSOCKET_KEY: &str = "dGhlIHNhbXBsZSBub25jZQ=="; const STATIC_WEBSOCKET_ACCEPT: &str = "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="; const MAX_HTTP_HEADER_BYTES: usize = 16 * 1024; const MAX_DIAGNOSTICS: usize = 32; +const MAX_RECENT_PROCESSED_COMMANDS: usize = 128; +const COMPACTED_COMMAND_FILTER_BYTES: usize = 4096; +const COMPACTED_COMMAND_FILTER_HASHES: usize = 7; const REVOKE_FLUSH_TIMEOUT: Duration = Duration::from_secs(2); const TEMP_FILE_ATTEMPTS: usize = 16; +type HmacSha256 = Hmac; + #[derive(Clone, Debug, PartialEq, Eq)] pub struct Phase3Error(String); @@ -150,6 +160,10 @@ pub struct DurableRunnerState { pub peak_outbox_bytes: usize, pub outbox: Vec, pub processed_commands: BTreeMap, + #[serde(default = "empty_compacted_command_filter")] + pub compacted_command_filter: String, + #[serde(default)] + pub compacted_command_count: u64, pub diagnostics: Vec, pub backpressure: bool, pub recoverable_failure: Option, @@ -177,6 +191,8 @@ impl DurableRunnerState { peak_outbox_bytes: 0, outbox: Vec::new(), processed_commands: BTreeMap::new(), + compacted_command_filter: empty_compacted_command_filter(), + compacted_command_count: 0, diagnostics: Vec::new(), backpressure: false, recoverable_failure: None, @@ -284,12 +300,17 @@ impl DurableStateStore { ))) } }; - let state: DurableRunnerState = serde_json::from_slice(&bytes).map_err(|error| { + let mut state: DurableRunnerState = serde_json::from_slice(&bytes).map_err(|error| { Phase3Error::invalid(format!( "durable runner state is malformed and cannot be recovered: {error}" )) })?; validate_state_binding(&state, config)?; + let previous_command_count = state.processed_commands.len(); + compact_processed_commands(&mut state)?; + if state.processed_commands.len() != previous_command_count { + self.save(&state)?; + } Ok((state, true)) } @@ -568,6 +589,7 @@ fn validate_state_binding( ))); } } + decode_compacted_command_filter(&state.compacted_command_filter)?; Ok(()) } @@ -646,13 +668,173 @@ fn canonical_json(value: &Value) -> String { } } -fn fnv1a_digest(input: &[u8]) -> String { - let mut digest = 0xcbf29ce484222325_u64; - for byte in input { - digest ^= u64::from(*byte); - digest = digest.wrapping_mul(0x100000001b3); +fn hex_encode(input: &[u8]) -> String { + input.iter().map(|byte| format!("{byte:02x}")).collect() +} + +fn hex_decode(input: &str) -> Result, Phase3Error> { + if input.len() % 2 != 0 { + return Err(Phase3Error::invalid("hex value has an odd length")); } - format!("fnv1a64:{digest:016x}") + input + .as_bytes() + .chunks_exact(2) + .map(|pair| { + let pair = std::str::from_utf8(pair) + .map_err(|_| Phase3Error::invalid("hex value is not valid UTF-8"))?; + u8::from_str_radix(pair, 16) + .map_err(|_| Phase3Error::invalid("hex value contains a non-hex character")) + }) + .collect() +} + +fn sha256_bytes(input: &[u8]) -> [u8; 32] { + Sha256::digest(input).into() +} + +fn sha256_digest(input: &[u8]) -> String { + format!("sha256:{}", hex_encode(&sha256_bytes(input))) +} + +fn digest_domain(domain: &str, parts: &[&[u8]]) -> [u8; 32] { + let mut digest = Sha256::new(); + digest.update(domain.as_bytes()); + digest.update([0]); + for part in parts { + digest.update((part.len() as u64).to_be_bytes()); + digest.update(part); + } + digest.finalize().into() +} + +fn hmac_domain(key: &[u8], domain: &str, parts: &[&[u8]]) -> [u8; 32] { + let mut mac = + ::new_from_slice(key).expect("HMAC-SHA256 accepts keys of every size"); + mac.update(domain.as_bytes()); + mac.update(&[0]); + for part in parts { + mac.update(&(part.len() as u64).to_be_bytes()); + mac.update(part); + } + mac.finalize().into_bytes().into() +} + +fn verify_hmac_hex( + key: &[u8], + domain: &str, + parts: &[&[u8]], + supplied: &str, +) -> Result<(), Phase3Error> { + let supplied = hex_decode(supplied)?; + let mut mac = + ::new_from_slice(key).expect("HMAC-SHA256 accepts keys of every size"); + mac.update(domain.as_bytes()); + mac.update(&[0]); + for part in parts { + mac.update(&(part.len() as u64).to_be_bytes()); + mac.update(part); + } + mac.verify_slice(&supplied) + .map_err(|_| Phase3Error::invalid("transport authentication proof is invalid")) +} + +fn current_unix_ms() -> Result { + let millis = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|_| Phase3Error::invalid("system clock precedes the Unix epoch"))? + .as_millis(); + u64::try_from(millis).map_err(|_| Phase3Error::invalid("system clock value overflowed")) +} + +struct CredentialMaterial { + credential_id: String, + auth_key: [u8; 32], +} + +impl CredentialMaterial { + fn from_token(token: &str) -> Self { + Self { + credential_id: format!( + "sha256:{}", + hex_encode(&digest_domain( + "paperclip-runner-credential-id-v1", + &[token.as_bytes()] + )) + ), + auth_key: digest_domain("paperclip-runner-auth-key-v1", &[token.as_bytes()]), + } + } +} + +impl Drop for CredentialMaterial { + fn drop(&mut self) { + self.auth_key.fill(0); + } +} + +fn empty_compacted_command_filter() -> String { + "00".repeat(COMPACTED_COMMAND_FILTER_BYTES) +} + +fn command_filter_positions(command_id: &str) -> [usize; COMPACTED_COMMAND_FILTER_HASHES] { + let digest = digest_domain( + "paperclip-runner-command-replay-filter-v1", + &[command_id.as_bytes()], + ); + let mut positions = [0_usize; COMPACTED_COMMAND_FILTER_HASHES]; + for (index, position) in positions.iter_mut().enumerate() { + let offset = index * 4; + let word = u32::from_be_bytes(digest[offset..offset + 4].try_into().unwrap()); + *position = word as usize % (COMPACTED_COMMAND_FILTER_BYTES * 8); + } + positions +} + +fn decode_compacted_command_filter(encoded: &str) -> Result, Phase3Error> { + let filter = hex_decode(encoded)?; + if filter.len() != COMPACTED_COMMAND_FILTER_BYTES { + return Err(Phase3Error::invalid( + "durable compacted command filter has an invalid size", + )); + } + Ok(filter) +} + +fn compacted_command_maybe_seen( + state: &DurableRunnerState, + command_id: &str, +) -> Result { + let filter = decode_compacted_command_filter(&state.compacted_command_filter)?; + Ok(command_filter_positions(command_id) + .iter() + .all(|position| filter[position / 8] & (1 << (position % 8)) != 0)) +} + +fn add_compacted_command( + state: &mut DurableRunnerState, + command_id: &str, +) -> Result<(), Phase3Error> { + let mut filter = decode_compacted_command_filter(&state.compacted_command_filter)?; + for position in command_filter_positions(command_id) { + filter[position / 8] |= 1 << (position % 8); + } + state.compacted_command_filter = hex_encode(&filter); + state.compacted_command_count = state.compacted_command_count.saturating_add(1); + Ok(()) +} + +fn compact_processed_commands(state: &mut DurableRunnerState) -> Result<(), Phase3Error> { + while state.processed_commands.len() > MAX_RECENT_PROCESSED_COMMANDS { + let oldest = state + .processed_commands + .values() + .min_by_key(|command| command.controller_seq) + .map(|command| command.command_id.clone()) + .ok_or_else(|| Phase3Error::invalid("processed command compaction lost its cursor"))?; + state.processed_commands.remove(&oldest); + add_compacted_command(state, &oldest)?; + } + Ok(()) } fn deterministic_time(sequence: u64) -> String { @@ -697,9 +879,11 @@ fn event_envelope( "envelopeId": format!("envelope_event_{source_seq:06}"), "kind": "event", "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, "sentAt": deterministic_time(source_seq), "payload": event, }) @@ -729,6 +913,14 @@ fn mark_backpressure( ) } +fn mark_p0_storage_exhausted(state: &mut DurableRunnerState) { + state.lifecycle = "unrecoverable".to_owned(); + state.unrecoverable_outcome = Some("p0_storage_exhausted".to_owned()); + state.record_diagnostic( + "P0 storage reserve is exhausted; the durable session requires operator recovery", + ); +} + fn enqueue_event( state: &mut DurableRunnerState, config: &DurableRunnerConfig, @@ -742,7 +934,7 @@ fn enqueue_event( } if priority == 2 { - if let Some(last) = state.outbox.last_mut() { + if let Some(last) = state.outbox.last() { if last.priority == 2 && last.event_type == event_type && last.item_id.as_deref() == item_id @@ -756,18 +948,26 @@ fn enqueue_event( "coalescedCount": previous_count + 1, "latest": sanitize_value(&payload), }); - last.envelope["payload"]["payload"] = compacted; - last.byte_size = serde_json::to_vec(&last.envelope) + let mut replacement = last.clone(); + replacement.envelope["payload"]["payload"] = compacted; + replacement.byte_size = serde_json::to_vec(&replacement.envelope) .map_err(|error| Phase3Error::invalid(error.to_string()))? .len(); - state.peak_outbox_bytes = state.peak_outbox_bytes.max(state.outbox_bytes()); - if state.outbox_bytes() - > config - .max_outbox_bytes - .saturating_sub(config.p0_reserve_bytes) - { + let projected = state + .outbox_bytes() + .saturating_sub(last.byte_size) + .saturating_add(replacement.byte_size); + let non_p0_limit = config + .max_outbox_bytes + .saturating_sub(config.p0_reserve_bytes); + if projected > non_p0_limit { return mark_backpressure(state, config); } + *state + .outbox + .last_mut() + .expect("coalesced event was just read from the outbox tail") = replacement; + state.peak_outbox_bytes = state.peak_outbox_bytes.max(projected); return Ok(()); } } @@ -793,11 +993,7 @@ fn enqueue_event( )); } if projected > config.max_outbox_bytes { - state.lifecycle = "unrecoverable".to_owned(); - state.unrecoverable_outcome = Some("p0_storage_exhausted".to_owned()); - state.record_diagnostic( - "P0 storage reserve is exhausted; the durable session requires operator recovery", - ); + mark_p0_storage_exhausted(state); return Err(Phase3Error::invalid( "P0 storage reserve exhausted; durable recovery is explicit", )); @@ -824,8 +1020,11 @@ fn command_result_envelope(state: &DurableRunnerState, processed: &ProcessedComm "envelopeId": format!("envelope_result_{}", processed.command_id), "kind": "command_result", "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, "sentAt": deterministic_time(processed.controller_seq), "payload": processed.result, }) @@ -1113,7 +1312,7 @@ fn process_command( .and_then(Value::as_str) .ok_or_else(|| Phase3Error::invalid("command type is required"))?; let canonical = canonical_json(command); - let command_digest = fnv1a_digest(canonical.as_bytes()); + let command_digest = sha256_digest(canonical.as_bytes()); if let Some(previous) = state.processed_commands.get(command_id) { if previous.command_digest != command_digest { @@ -1123,6 +1322,40 @@ fn process_command( } return Ok(previous.clone()); } + if compacted_command_maybe_seen(state, command_id)? { + if controller_seq > state.last_controller_command_seq + 1 { + return Err(Phase3Error::invalid(format!( + "controllerSeq must be at most {}; received {controller_seq}", + state.last_controller_command_seq + 1 + ))); + } + let result = sanitize_value(&json!({ + "commandId": command_id, + "controllerSeq": controller_seq, + "status": "rejected", + "logicalEffectCount": 0, + "detail": "command identity may have been processed before ledger compaction; replay rejected fail-closed", + })); + let processed = ProcessedCommand { + command_id: command_id.to_owned(), + controller_seq, + command_digest, + status: "rejected".to_owned(), + logical_effect_count: 0, + result, + }; + if controller_seq == state.last_controller_command_seq + 1 { + let mut candidate = state.clone(); + candidate.last_controller_command_seq = controller_seq; + candidate + .processed_commands + .insert(command_id.to_owned(), processed.clone()); + compact_processed_commands(&mut candidate)?; + store.save(&candidate)?; + *state = candidate; + } + return Ok(processed); + } if controller_seq != state.last_controller_command_seq + 1 { return Err(Phase3Error::invalid(format!( "controllerSeq must be {}; received {controller_seq}", @@ -1131,11 +1364,20 @@ fn process_command( } let payload = command.get("payload").cloned().unwrap_or(Value::Null); - let effect = execute_command_effect(state, config, command_type, &payload); + let mut candidate = state.clone(); + let effect = execute_command_effect(&mut candidate, config, command_type, &payload); let (status, logical_effect_count, detail) = match effect { Ok(effect) => effect, - Err(error) if state.unrecoverable_outcome.is_some() => { - ("failed".to_owned(), 1, error.to_string()) + Err(error) + if state.unrecoverable_outcome.as_deref() != Some("p0_storage_exhausted") + && candidate.unrecoverable_outcome.as_deref() == Some("p0_storage_exhausted") => + { + // The attempted command may have staged earlier events before discovering that a + // mandatory P0 event cannot be stored. Discard every staged logical effect and keep + // only the truthful terminal storage outcome plus its durable command result. + candidate = state.clone(); + mark_p0_storage_exhausted(&mut candidate); + ("failed".to_owned(), 0, error.to_string()) } Err(error) => return Err(error), }; @@ -1154,11 +1396,13 @@ fn process_command( logical_effect_count, result, }; - state.last_controller_command_seq = controller_seq; - state + candidate.last_controller_command_seq = controller_seq; + candidate .processed_commands .insert(command_id.to_owned(), processed.clone()); - store.save(state)?; + compact_processed_commands(&mut candidate)?; + store.save(&candidate)?; + *state = candidate; Ok(processed) } @@ -1276,6 +1520,129 @@ struct WsClient { stream: TcpStream, mask_counter: u32, max_frame_bytes: usize, + secure_channel: Option, +} + +struct SecureChannel { + send_cipher: Aes256Gcm, + receive_cipher: Aes256Gcm, + send_counter: u64, + receive_counter: u64, + session_id: String, +} + +impl SecureChannel { + fn new( + auth_key: &[u8], + challenge: &[u8], + server_proof: &[u8], + client_proof: &[u8], + ) -> Result { + let session_binding = digest_domain( + "paperclip-runner-session-binding-v1", + &[challenge, server_proof, client_proof], + ); + let send_key = hmac_domain( + auth_key, + "paperclip-runner-client-to-core-key-v1", + &[&session_binding], + ); + let receive_key = hmac_domain( + auth_key, + "paperclip-runner-core-to-client-key-v1", + &[&session_binding], + ); + Ok(Self { + send_cipher: Aes256Gcm::new_from_slice(&send_key) + .map_err(|_| Phase3Error::invalid("failed to initialize transport encryption"))?, + receive_cipher: Aes256Gcm::new_from_slice(&receive_key) + .map_err(|_| Phase3Error::invalid("failed to initialize transport decryption"))?, + send_counter: 0, + receive_counter: 0, + session_id: format!("sha256:{}", hex_encode(&session_binding)), + }) + } + + fn nonce(direction: &[u8; 4], counter: u64) -> [u8; 12] { + let mut nonce = [0_u8; 12]; + nonce[..4].copy_from_slice(direction); + nonce[4..].copy_from_slice(&counter.to_be_bytes()); + nonce + } + + fn aad(&self, direction: &str, counter: u64) -> Vec { + format!( + "{SECURE_FRAME_SCHEMA}\0{}\0{direction}\0{counter}", + self.session_id + ) + .into_bytes() + } + + fn encrypt(&mut self, plaintext: &[u8]) -> Result { + let counter = self.send_counter; + let nonce = Self::nonce(b"P3C1", counter); + let aad = self.aad("client_to_core", counter); + let ciphertext = self + .send_cipher + .encrypt( + Nonce::from_slice(&nonce), + Payload { + msg: plaintext, + aad: &aad, + }, + ) + .map_err(|_| Phase3Error::invalid("secure transport encryption failed"))?; + self.send_counter = self + .send_counter + .checked_add(1) + .ok_or_else(|| Phase3Error::invalid("secure transport send counter exhausted"))?; + Ok(json!({ + "schema": SECURE_FRAME_SCHEMA, + "counter": counter, + "ciphertext": hex_encode(&ciphertext), + })) + } + + fn decrypt(&mut self, frame: &Value) -> Result { + if frame.get("schema").and_then(Value::as_str) != Some(SECURE_FRAME_SCHEMA) { + return Err(Phase3Error::invalid( + "unauthenticated plaintext control frame was rejected", + )); + } + let counter = frame + .get("counter") + .and_then(Value::as_u64) + .ok_or_else(|| Phase3Error::invalid("secure frame counter is required"))?; + if counter != self.receive_counter { + return Err(Phase3Error::invalid( + "secure frame counter was replayed or arrived out of order", + )); + } + let ciphertext = frame + .get("ciphertext") + .and_then(Value::as_str) + .ok_or_else(|| Phase3Error::invalid("secure frame ciphertext is required"))?; + let ciphertext = hex_decode(ciphertext)?; + let nonce = Self::nonce(b"P3S1", counter); + let aad = self.aad("core_to_client", counter); + let plaintext = self + .receive_cipher + .decrypt( + Nonce::from_slice(&nonce), + Payload { + msg: &ciphertext, + aad: &aad, + }, + ) + .map_err(|_| Phase3Error::invalid("secure frame authentication failed"))?; + self.receive_counter = self + .receive_counter + .checked_add(1) + .ok_or_else(|| Phase3Error::invalid("secure transport receive counter exhausted"))?; + serde_json::from_slice(&plaintext).map_err(|error| { + Phase3Error::invalid(format!("secure frame JSON is malformed: {error}")) + }) + } } fn encode_masked_frame( @@ -1323,11 +1690,7 @@ fn checked_inbound_frame_length(length: u64, max_frame_bytes: usize) -> Result Result { + fn connect(target: &ResolvedWsTarget, max_frame_bytes: usize) -> Result { let mut stream = TcpStream::connect(target.addresses.as_slice()) .map_err(|error| Phase3Error::invalid(format!("WebSocket connect failed: {error}")))?; stream @@ -1337,16 +1700,15 @@ impl WsClient { .set_write_timeout(Some(Duration::from_secs(2))) .map_err(|error| Phase3Error::invalid(error.to_string()))?; let mut request = format!( - "GET {} HTTP/1.1\r\nHost: {}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: {}\r\nSec-WebSocket-Version: 13\r\nAuthorization: Bearer {}\r\n\r\n", + "GET {} HTTP/1.1\r\nHost: {}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: {}\r\nSec-WebSocket-Version: 13\r\n\r\n", target.path, target.authority, - STATIC_WEBSOCKET_KEY, - authorization + STATIC_WEBSOCKET_KEY ) .into_bytes(); let write_result = stream.write_all(&request).and_then(|_| stream.flush()); - // The HTTP upgrade request contains the bearer capability. Overwrite it before - // propagating success or failure so it does not linger in daemon memory. + // Keep the request buffer short-lived even though it contains only public data. + // Authentication capabilities never cross the socket. request.fill(0); request.clear(); write_result.map_err(|error| { @@ -1395,10 +1757,23 @@ impl WsClient { stream, mask_counter: 1, max_frame_bytes, + secure_channel: None, }) } fn send_json(&mut self, value: &Value) -> Result<(), Phase3Error> { + let bytes = serde_json::to_vec(value).map_err(|error| { + Phase3Error::invalid(format!("frame serialization failed: {error}")) + })?; + let frame = self + .secure_channel + .as_mut() + .ok_or_else(|| Phase3Error::invalid("secure transport is not authenticated"))? + .encrypt(&bytes)?; + self.send_plain_json(&frame) + } + + fn send_plain_json(&mut self, value: &Value) -> Result<(), Phase3Error> { let bytes = serde_json::to_vec(value).map_err(|error| { Phase3Error::invalid(format!("frame serialization failed: {error}")) })?; @@ -1416,6 +1791,18 @@ impl WsClient { } fn receive_json(&mut self) -> Result, Phase3Error> { + let Some(frame) = self.receive_plain_json()? else { + return Ok(None); + }; + let value = self + .secure_channel + .as_mut() + .ok_or_else(|| Phase3Error::invalid("secure transport is not authenticated"))? + .decrypt(&frame)?; + Ok(Some(value)) + } + + fn receive_plain_json(&mut self) -> Result, Phase3Error> { loop { let mut header = [0_u8; 2]; match self.stream.read_exact(&mut header) { @@ -1480,9 +1867,18 @@ impl WsClient { } } } + + fn enable_secure_channel(&mut self, channel: SecureChannel) { + self.secure_channel = Some(channel); + } } -fn hello_envelope(state: &DurableRunnerState, config: &DurableRunnerConfig) -> Value { +fn authentication_hello_envelope( + state: &DurableRunnerState, + config: &DurableRunnerConfig, + credential_id: &str, + client_nonce: &str, +) -> Value { let unacked_range = state .outbox .first() @@ -1491,16 +1887,20 @@ fn hello_envelope(state: &DurableRunnerState, config: &DurableRunnerConfig) -> V json!({ "protocol": PROTOCOL, "version": PROTOCOL_VERSION, - "envelopeId": format!("envelope_hello_{}", state.reconnect_count), - "kind": "hello", - "runnerInstanceId": state.runner_instance_id, - "sentAt": deterministic_time(state.reconnect_count), + "kind": "auth_hello", "payload": { + "credentialId": credential_id, + "clientNonce": client_nonce, "protocolMin": 1, "protocolMax": 1, + "runnerInstanceId": state.runner_instance_id, "runnerVersion": config.runner_version, "runnerDigest": config.runner_digest, "environmentLeaseId": state.environment_lease_id, + "runId": state.run_id, + "normalizedSessionId": state.normalized_session_id, + "turnId": state.turn_id, + "itemId": state.item_id, "sandboxProvider": "standalone_mock", "platform": { "os": std::env::consts::OS, @@ -1522,6 +1922,152 @@ fn hello_envelope(state: &DurableRunnerState, config: &DurableRunnerConfig) -> V }) } +fn required_string<'a>(value: &'a Value, field: &str) -> Result<&'a str, Phase3Error> { + value + .get(field) + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + .ok_or_else(|| Phase3Error::invalid(format!("{field} is required"))) +} + +fn authenticate_transport( + client: &mut WsClient, + state: &DurableRunnerState, + config: &DurableRunnerConfig, + credential: &CredentialMaterial, + credential_kind: &str, + expected_lease_id: Option<&str>, + expected_expires_at_unix_ms: Option, + expected_revocation_epoch: Option, +) -> Result<(), Phase3Error> { + let client_nonce = random_suffix()?; + client.send_plain_json(&authentication_hello_envelope( + state, + config, + &credential.credential_id, + &client_nonce, + ))?; + let deadline = Instant::now() + .checked_add(Duration::from_secs(2)) + .ok_or_else(|| Phase3Error::invalid("transport authentication deadline overflowed"))?; + let challenge = loop { + match client.receive_plain_json()? { + Some(value) => break value, + None if Instant::now() < deadline => continue, + None => return Err(Phase3Error::invalid("transport authentication timed out")), + } + }; + if challenge.get("protocol").and_then(Value::as_str) != Some(PROTOCOL) + || challenge.get("version").and_then(Value::as_u64) != Some(PROTOCOL_VERSION) + || challenge.get("kind").and_then(Value::as_str) != Some("auth_challenge") + { + return Err(Phase3Error::invalid( + "core did not return an authenticated transport challenge", + )); + } + let payload = challenge + .get("payload") + .ok_or_else(|| Phase3Error::invalid("authentication challenge payload is required"))?; + for (field, expected) in [ + ("credentialId", credential.credential_id.as_str()), + ("credentialKind", credential_kind), + ("clientNonce", client_nonce.as_str()), + ("runnerInstanceId", state.runner_instance_id.as_str()), + ("environmentLeaseId", state.environment_lease_id.as_str()), + ("runId", state.run_id.as_str()), + ("normalizedSessionId", state.normalized_session_id.as_str()), + ("turnId", state.turn_id.as_str()), + ("itemId", state.item_id.as_str()), + ("runnerVersion", config.runner_version.as_str()), + ("runnerDigest", config.runner_digest.as_str()), + ] { + if required_string(payload, field)? != expected { + return Err(Phase3Error::invalid(format!( + "authentication challenge {field} does not match the requested session" + ))); + } + } + required_string(payload, "serverNonce")?; + required_string(payload, "credentialExpiresAt")?; + if payload.get("selectedVersion").and_then(Value::as_u64) != Some(PROTOCOL_VERSION) { + return Err(Phase3Error::invalid( + "authentication challenge selected an unsupported protocol", + )); + } + let expires_at_unix_ms = payload + .get("credentialExpiresAtUnixMs") + .and_then(Value::as_u64) + .ok_or_else(|| { + Phase3Error::invalid("authentication challenge credential expiry is required") + })?; + if expires_at_unix_ms <= current_unix_ms()? { + return Err(Phase3Error::invalid( + "transport credential expired before authentication completed", + )); + } + if expected_expires_at_unix_ms.is_some_and(|expected| expected != expires_at_unix_ms) { + return Err(Phase3Error::invalid( + "authentication challenge changed the connection lease expiry", + )); + } + let revocation_epoch = payload + .get("revocationEpoch") + .and_then(Value::as_u64) + .ok_or_else(|| Phase3Error::invalid("authentication revocation epoch is required"))?; + if expected_revocation_epoch.is_some_and(|expected| expected != revocation_epoch) { + return Err(Phase3Error::invalid( + "authentication challenge changed the connection lease revocation epoch", + )); + } + match (expected_lease_id, payload.get("credentialLeaseId")) { + (Some(expected), Some(Value::String(actual))) if actual == expected => {} + (None, Some(Value::Null)) => {} + _ => { + return Err(Phase3Error::invalid( + "authentication challenge lease identity does not match the credential", + )) + } + } + + let server_proof = required_string(payload, "serverProof")?.to_owned(); + let mut authenticated_payload = payload.clone(); + authenticated_payload + .as_object_mut() + .ok_or_else(|| Phase3Error::invalid("authentication challenge payload must be an object"))? + .remove("serverProof"); + let canonical_challenge = canonical_json(&authenticated_payload); + verify_hmac_hex( + &credential.auth_key, + "paperclip-runner-server-proof-v1", + &[canonical_challenge.as_bytes()], + &server_proof, + )?; + let client_proof = hex_encode(&hmac_domain( + &credential.auth_key, + "paperclip-runner-client-proof-v1", + &[canonical_challenge.as_bytes(), server_proof.as_bytes()], + )); + client.send_plain_json(&json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": "auth_response", + "payload": { + "credentialId": credential.credential_id, + "clientNonce": client_nonce, + "serverNonce": required_string(payload, "serverNonce")?, + "clientProof": client_proof, + }, + }))?; + let secure_channel = SecureChannel::new( + &credential.auth_key, + canonical_challenge.as_bytes(), + server_proof.as_bytes(), + client_proof.as_bytes(), + )?; + client.enable_secure_channel(secure_channel); + Ok(()) +} + fn send_outbox(client: &mut WsClient, state: &DurableRunnerState) -> Result<(), Phase3Error> { for event in &state.outbox { client.send_json(&event.envelope)?; @@ -1529,16 +2075,122 @@ fn send_outbox(client: &mut WsClient, state: &DurableRunnerState) -> Result<(), Ok(()) } -fn welcome_payload(value: &Value) -> Result<&Value, Phase3Error> { +struct ConnectionMetadata { + connection_id: String, + lease_id: String, + lease_expires_at_unix_ms: u64, + revocation_epoch: u64, +} + +fn validate_control_identity( + value: &Value, + state: &DurableRunnerState, + connection: Option<&ConnectionMetadata>, +) -> Result<(), Phase3Error> { if value.get("protocol").and_then(Value::as_str) != Some(PROTOCOL) || value.get("version").and_then(Value::as_u64) != Some(PROTOCOL_VERSION) - || value.get("kind").and_then(Value::as_str) != Some("welcome") { + return Err(Phase3Error::invalid( + "control envelope protocol identity is invalid", + )); + } + for (field, expected) in [ + ("runnerInstanceId", state.runner_instance_id.as_str()), + ("environmentLeaseId", state.environment_lease_id.as_str()), + ("runId", state.run_id.as_str()), + ("normalizedSessionId", state.normalized_session_id.as_str()), + ("turnId", state.turn_id.as_str()), + ("itemId", state.item_id.as_str()), + ] { + if required_string(value, field)? != expected { + return Err(Phase3Error::invalid(format!( + "control envelope {field} does not match the authenticated session" + ))); + } + } + if let Some(connection) = connection { + if required_string(value, "connectionId")? != connection.connection_id + || required_string(value, "connectionLeaseId")? != connection.lease_id + { + return Err(Phase3Error::invalid( + "control envelope connection lease identity does not match the authenticated session", + )); + } + if current_unix_ms()? >= connection.lease_expires_at_unix_ms { + return Err(Phase3Error::invalid( + "connection lease expired before the control envelope was applied", + )); + } + } + Ok(()) +} + +fn validate_welcome<'a>( + value: &'a Value, + state: &DurableRunnerState, +) -> Result<(&'a Value, ConnectionMetadata), Phase3Error> { + validate_control_identity(value, state, None)?; + if value.get("kind").and_then(Value::as_str) != Some("welcome") { return Err(Phase3Error::invalid("expected a PRP v1 welcome envelope")); } - value + let connection_id = required_string(value, "connectionId")?.to_owned(); + let lease_id = required_string(value, "connectionLeaseId")?.to_owned(); + let payload = value .get("payload") - .ok_or_else(|| Phase3Error::invalid("welcome payload is required")) + .ok_or_else(|| Phase3Error::invalid("welcome payload is required"))?; + if payload.get("selectedVersion").and_then(Value::as_u64) != Some(PROTOCOL_VERSION) { + return Err(Phase3Error::invalid( + "core selected an unsupported protocol version", + )); + } + if required_string(payload, "connectionLeaseId")? != lease_id { + return Err(Phase3Error::invalid( + "welcome lease identity is internally inconsistent", + )); + } + required_string(payload, "connectionLeaseExpiresAt")?; + let lease_expires_at_unix_ms = payload + .get("connectionLeaseExpiresAtUnixMs") + .and_then(Value::as_u64) + .ok_or_else(|| Phase3Error::invalid("welcome lease expiry is required"))?; + if lease_expires_at_unix_ms <= current_unix_ms()? { + return Err(Phase3Error::invalid( + "welcome carried an already-expired connection lease", + )); + } + let revocation_epoch = payload + .get("connectionLeaseRevocationEpoch") + .and_then(Value::as_u64) + .ok_or_else(|| Phase3Error::invalid("welcome revocation epoch is required"))?; + let binding = payload + .get("leaseBinding") + .ok_or_else(|| Phase3Error::invalid("welcome lease binding is required"))?; + for (field, expected) in [ + ("runnerInstanceId", state.runner_instance_id.as_str()), + ("environmentLeaseId", state.environment_lease_id.as_str()), + ("runId", state.run_id.as_str()), + ("normalizedSessionId", state.normalized_session_id.as_str()), + ] { + if required_string(binding, field)? != expected { + return Err(Phase3Error::invalid(format!( + "welcome lease binding {field} does not match the authenticated session" + ))); + } + } + if binding.get("protocolVersion").and_then(Value::as_u64) != Some(PROTOCOL_VERSION) { + return Err(Phase3Error::invalid( + "welcome lease binding protocol is invalid", + )); + } + Ok(( + payload, + ConnectionMetadata { + connection_id, + lease_id, + lease_expires_at_unix_ms, + revocation_epoch, + }, + )) } pub fn run_durable_runner( @@ -1550,11 +2202,18 @@ pub fn run_durable_runner( "P0 reserve must be smaller than the outbox limit", )); } - // Resolve once before a bearer can be sent. Reconnects use only this validated, - // concrete address set, so DNS cannot redirect an authenticated retry. - let target = ResolvedWsTarget::resolve(&config.connect_url)?; let store = DurableStateStore::new(&config.state_dir)?; let (mut state, recovered) = store.load_or_create(&config)?; + if recovered && state.lifecycle == "revoked" { + // Revocation is durable and terminal. Dropping the fresh bootstrap capability before + // destination resolution guarantees a restarted daemon cannot authenticate, reconnect, + // advance cursors, or rewrite the preserved operator-recovery state. + drop(bootstrap_ticket); + return Ok(()); + } + // Resolve once before authentication. Reconnects use only this validated, + // concrete address set, so DNS cannot redirect a retry. + let target = ResolvedWsTarget::resolve(&config.connect_url)?; if recovered { state.reconnect_count += 1; state.record_diagnostic("runner process restored the same durable identity"); @@ -1576,6 +2235,9 @@ pub fn run_durable_runner( } let mut bootstrap_ticket = Some(bootstrap_ticket); let mut connection_lease_token: Option = None; + let mut connection_lease_id: Option = None; + let mut connection_lease_expires_at_unix_ms: Option = None; + let mut connection_lease_revocation_epoch: Option = None; let started = Instant::now(); loop { @@ -1588,38 +2250,61 @@ pub fn run_durable_runner( "transport reconnect deadline exceeded; durable state is preserved", )); } - let authorization = connection_lease_token + if connection_lease_expires_at_unix_ms + .is_some_and(|expires_at| current_unix_ms().is_ok_and(|now| now >= expires_at)) + { + state.recoverable_failure = Some("lease_expired_requires_bootstrap".to_owned()); + state.lifecycle = "recoverable_failure".to_owned(); + state.record_diagnostic( + "connection lease expired; a fresh bootstrap may resume this state", + ); + store.save(&state)?; + connection_lease_token.take(); + return Err(Phase3Error::invalid( + "connection lease expired; durable state requires a fresh bootstrap", + )); + } + let (credential_token, credential_kind) = connection_lease_token .as_ref() - .map(SensitiveString::expose) - .or_else(|| bootstrap_ticket.as_ref().map(BootstrapTicket::expose)) + .map(|token| (token.expose(), "lease")) + .or_else(|| { + bootstrap_ticket + .as_ref() + .map(|ticket| (ticket.expose(), "bootstrap")) + }) .ok_or_else(|| { Phase3Error::invalid( "transport capability is unavailable; a fresh runner bootstrap is required", ) })?; - let mut client = match WsClient::connect(&target, authorization, config.max_frame_bytes) { + let credential = CredentialMaterial::from_token(credential_token); + let mut client = match WsClient::connect(&target, config.max_frame_bytes) { Ok(client) => client, Err(error) => { let text = error.to_string(); - if text.contains("401") || text.contains("403") { - state.recoverable_failure = Some("lease_expired_requires_bootstrap".to_owned()); - state.lifecycle = "recoverable_failure".to_owned(); - state.record_diagnostic( - "connection lease expired; a fresh bootstrap may resume this state", - ); - store.save(&state)?; - return Err(Phase3Error::invalid( - "connection lease expired; durable state requires a fresh bootstrap", - )); - } state.record_diagnostic(format!("transport reconnect scheduled: {text}")); store.save(&state)?; thread::sleep(config.reconnect_delay); continue; } }; - - client.send_json(&hello_envelope(&state, &config))?; + if let Err(error) = authenticate_transport( + &mut client, + &state, + &config, + &credential, + credential_kind, + connection_lease_id.as_deref(), + connection_lease_expires_at_unix_ms, + connection_lease_revocation_epoch, + ) { + state.record_diagnostic(format!( + "transport peer authentication failed closed: {error}" + )); + store.save(&state)?; + thread::sleep(config.reconnect_delay); + continue; + } let mut welcome = loop { match client.receive_json() { Ok(Some(value)) => break value, @@ -1635,6 +2320,7 @@ pub fn run_durable_runner( } } }; + let (_, connection) = validate_welcome(&welcome, &state)?; let next_lease_token = match welcome.pointer_mut("/payload/connectionLeaseToken") { Some(Value::String(value)) => { let token = std::mem::take(value); @@ -1643,22 +2329,23 @@ pub fn run_durable_runner( } _ => None, }; - let payload = welcome_payload(&welcome)?; - let selected_version = payload - .get("selectedVersion") - .and_then(Value::as_u64) - .ok_or_else(|| Phase3Error::invalid("welcome selectedVersion is required"))?; - if selected_version != PROTOCOL_VERSION { + if credential_kind == "bootstrap" && next_lease_token.is_none() { return Err(Phase3Error::invalid( - "mock core selected an unsupported version", + "authenticated bootstrap welcome did not issue a connection lease", )); } if let Some(token) = next_lease_token { connection_lease_token = Some(token); } - // A bootstrap ticket is one-use. Clear it as soon as the authenticated welcome - // has exchanged it for a connection lease. + connection_lease_id = Some(connection.lease_id.clone()); + connection_lease_expires_at_unix_ms = Some(connection.lease_expires_at_unix_ms); + connection_lease_revocation_epoch = Some(connection.revocation_epoch); + // A bootstrap ticket is one-use. Clear it only after the mutually authenticated, + // encrypted welcome has exchanged it for a bound connection lease. bootstrap_ticket.take(); + let payload = welcome + .get("payload") + .ok_or_else(|| Phase3Error::invalid("welcome payload is required"))?; if let Some(acked_source_seq) = payload.get("ackedSourceSeq").and_then(Value::as_u64) { state.apply_ack(acked_source_seq)?; } @@ -1717,6 +2404,18 @@ pub fn run_durable_runner( bootstrap_ticket.take(); return Ok(()); } + if current_unix_ms()? >= connection.lease_expires_at_unix_ms { + state.recoverable_failure = Some("lease_expired_requires_bootstrap".to_owned()); + state.lifecycle = "recoverable_failure".to_owned(); + state.record_diagnostic( + "connection lease expired before more control data could be accepted", + ); + store.save(&state)?; + connection_lease_token.take(); + return Err(Phase3Error::invalid( + "connection lease expired; durable state requires a fresh bootstrap", + )); + } match client.receive_json() { Ok(None) => continue, Err(error) => { @@ -1731,100 +2430,124 @@ pub fn run_durable_runner( store.save(&state)?; break true; } - Ok(Some(message)) => match message.get("kind").and_then(Value::as_str) { - Some("ack") => { - let acked = message - .pointer("/payload/ackedSourceSeq") - .and_then(Value::as_u64) - .ok_or_else(|| Phase3Error::invalid("ACK cursor is required"))?; - state.apply_ack(acked)?; + Ok(Some(message)) => { + if let Err(error) = + validate_control_identity(&message, &state, Some(&connection)) + { + state.record_diagnostic(format!( + "control envelope identity mismatch failed closed: {error}" + )); + state.reconnect_count += 1; store.save(&state)?; + break true; } - Some("command") => { - let command = message - .get("payload") - .ok_or_else(|| Phase3Error::invalid("command payload is required"))?; - match process_command(&mut state, &store, &config, command) { - Ok(processed) => { - let delivery = client - .send_json(&command_result_envelope(&state, &processed)) - .and_then(|()| send_outbox(&mut client, &state)); - if let Err(error) = delivery { - state.record_diagnostic(error.to_string()); - if state.lifecycle == "revoked" { + match message.get("kind").and_then(Value::as_str) { + Some("ack") => { + let acked = message + .pointer("/payload/ackedSourceSeq") + .and_then(Value::as_u64) + .ok_or_else(|| Phase3Error::invalid("ACK cursor is required"))?; + state.apply_ack(acked)?; + store.save(&state)?; + } + Some("command") => { + let command = message.get("payload").ok_or_else(|| { + Phase3Error::invalid("command payload is required") + })?; + match process_command(&mut state, &store, &config, command) { + Ok(processed) => { + let delivery = client + .send_json(&command_result_envelope(&state, &processed)) + .and_then(|()| send_outbox(&mut client, &state)); + if let Err(error) = delivery { + state.record_diagnostic(error.to_string()); + if state.lifecycle == "revoked" { + store.save(&state)?; + connection_lease_token.take(); + bootstrap_ticket.take(); + return Ok(()); + } + state.reconnect_count += 1; store.save(&state)?; - connection_lease_token.take(); - bootstrap_ticket.take(); - return Ok(()); + break true; } - state.reconnect_count += 1; + } + Err(error) => { + state.record_diagnostic(error.to_string()); store.save(&state)?; - break true; } } - Err(error) => { - state.record_diagnostic(error.to_string()); - store.save(&state)?; + } + Some("revoke") => { + let revocation_epoch = message + .pointer("/payload/revocationEpoch") + .and_then(Value::as_u64) + .ok_or_else(|| { + Phase3Error::invalid("revoke revocation epoch is required") + })?; + if revocation_epoch <= connection.revocation_epoch { + return Err(Phase3Error::invalid( + "revoke did not advance the authenticated revocation epoch", + )); } - } - } - Some("revoke") => { - state.lifecycle = "revoked".to_owned(); - state.stop_after_flush = true; - revoke_deadline.get_or_insert_with(|| { - Instant::now() - .checked_add(REVOKE_FLUSH_TIMEOUT) - .unwrap_or_else(Instant::now) - }); - state - .record_diagnostic("connection lease revoked; flushing durable events"); - store.save(&state)?; - if let Err(error) = send_outbox(&mut client, &state) { - state.record_diagnostic(error.to_string()); + state.lifecycle = "revoked".to_owned(); + state.stop_after_flush = true; + revoke_deadline.get_or_insert_with(|| { + Instant::now() + .checked_add(REVOKE_FLUSH_TIMEOUT) + .unwrap_or_else(Instant::now) + }); + state.record_diagnostic( + "connection lease revoked; flushing durable events", + ); store.save(&state)?; - connection_lease_token.take(); - bootstrap_ticket.take(); - return Ok(()); - } - } - Some("ping") => { - let pong = client.send_json(&json!({ - "protocol": PROTOCOL, - "version": PROTOCOL_VERSION, - "kind": "pong", - "runnerInstanceId": state.runner_instance_id, - "payload": { - "lifecycle": state.lifecycle, - "outboxBytes": state.outbox_bytes(), - "ackedSourceSeq": state.acked_source_seq, - }, - })); - if let Err(error) = pong { - state.record_diagnostic(error.to_string()); - if state.lifecycle == "revoked" { + if let Err(error) = send_outbox(&mut client, &state) { + state.record_diagnostic(error.to_string()); store.save(&state)?; connection_lease_token.take(); bootstrap_ticket.take(); return Ok(()); } - state.reconnect_count += 1; + } + Some("ping") => { + let pong = client.send_json(&json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": "pong", + "runnerInstanceId": state.runner_instance_id, + "payload": { + "lifecycle": state.lifecycle, + "outboxBytes": state.outbox_bytes(), + "ackedSourceSeq": state.acked_source_seq, + }, + })); + if let Err(error) = pong { + state.record_diagnostic(error.to_string()); + if state.lifecycle == "revoked" { + store.save(&state)?; + connection_lease_token.take(); + bootstrap_ticket.take(); + return Ok(()); + } + state.reconnect_count += 1; + store.save(&state)?; + break true; + } + } + _ => { + state.record_diagnostic( + "malformed or unsupported control frame closed the connection", + ); store.save(&state)?; + if state.lifecycle == "revoked" { + connection_lease_token.take(); + bootstrap_ticket.take(); + return Ok(()); + } break true; } } - _ => { - state.record_diagnostic( - "malformed or unsupported control frame closed the connection", - ); - store.save(&state)?; - if state.lifecycle == "revoked" { - connection_lease_token.take(); - bootstrap_ticket.take(); - return Ok(()); - } - break true; - } - }, + } } }; if disconnected { @@ -1836,7 +2559,8 @@ pub fn run_durable_runner( #[cfg(test)] mod tests { use super::*; - use std::sync::Mutex; + use std::net::TcpListener; + use std::sync::{mpsc, Mutex}; static ENVIRONMENT_TEST_LOCK: Mutex<()> = Mutex::new(()); @@ -1883,6 +2607,53 @@ mod tests { }) } + fn read_masked_client_text(stream: &mut TcpStream) -> (Value, Vec) { + let mut header = [0_u8; 2]; + stream.read_exact(&mut header).unwrap(); + assert_eq!(header[0] & 0x0f, 0x1); + assert_ne!(header[1] & 0x80, 0); + let mut captured = header.to_vec(); + let mut length = usize::from(header[1] & 0x7f); + if length == 126 { + let mut extended = [0_u8; 2]; + stream.read_exact(&mut extended).unwrap(); + captured.extend_from_slice(&extended); + length = usize::from(u16::from_be_bytes(extended)); + } else if length == 127 { + let mut extended = [0_u8; 8]; + stream.read_exact(&mut extended).unwrap(); + captured.extend_from_slice(&extended); + length = usize::try_from(u64::from_be_bytes(extended)).unwrap(); + } + let mut mask = [0_u8; 4]; + stream.read_exact(&mut mask).unwrap(); + captured.extend_from_slice(&mask); + let mut payload = vec![0_u8; length]; + stream.read_exact(&mut payload).unwrap(); + captured.extend_from_slice(&payload); + for (index, byte) in payload.iter_mut().enumerate() { + *byte ^= mask[index % mask.len()]; + } + (serde_json::from_slice(&payload).unwrap(), captured) + } + + fn send_server_json(stream: &mut TcpStream, value: &Value) { + let payload = serde_json::to_vec(value).unwrap(); + let mut frame = vec![0x81]; + if payload.len() <= 125 { + frame.push(payload.len() as u8); + } else if payload.len() <= u16::MAX as usize { + frame.push(126); + frame.extend_from_slice(&(payload.len() as u16).to_be_bytes()); + } else { + frame.push(127); + frame.extend_from_slice(&(payload.len() as u64).to_be_bytes()); + } + frame.extend_from_slice(&payload); + stream.write_all(&frame).unwrap(); + stream.flush().unwrap(); + } + #[test] fn durable_state_restores_identity_outbox_and_cumulative_ack() { let root = temporary_root("restore"); @@ -1933,6 +2704,8 @@ mod tests { let (mut state, _) = store.load_or_create(&config).unwrap(); let value = command("command_prepare", 1, "run.prepare", json!({})); let first = process_command(&mut state, &store, &config, &value).unwrap(); + assert!(first.command_digest.starts_with("sha256:")); + assert_eq!(first.command_digest.len(), "sha256:".len() + 64); let event_count = state.outbox.len(); let second = process_command(&mut state, &store, &config, &value).unwrap(); assert_eq!(first.result, second.result); @@ -1954,6 +2727,270 @@ mod tests { let _ = fs::remove_dir_all(root); } + #[test] + fn processed_command_ledger_compacts_to_a_fixed_bound_without_reenabling_effects() { + let root = temporary_root("command-compaction"); + let config = config(&root); + let store = DurableStateStore::new(&root).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + for sequence in 1..=256_u64 { + let command_id = format!("command_compaction_{sequence:04}"); + process_command( + &mut state, + &store, + &config, + &command(&command_id, sequence, "unsupported.compaction", json!({})), + ) + .unwrap(); + } + let size_after_256 = fs::metadata(store.path()).unwrap().len(); + for sequence in 257..=512_u64 { + let command_id = format!("command_compaction_{sequence:04}"); + process_command( + &mut state, + &store, + &config, + &command(&command_id, sequence, "unsupported.compaction", json!({})), + ) + .unwrap(); + } + let size_after_512 = fs::metadata(store.path()).unwrap().len(); + assert_eq!( + state.processed_commands.len(), + MAX_RECENT_PROCESSED_COMMANDS + ); + assert_eq!(state.compacted_command_count, 384); + assert_eq!( + state.compacted_command_filter.len(), + COMPACTED_COMMAND_FILTER_BYTES * 2 + ); + assert!(size_after_512 <= size_after_256 + 1024); + + let replay = process_command( + &mut state, + &store, + &config, + &command( + "command_compaction_0001", + 513, + "run.prepare", + json!({ "changed": true }), + ), + ) + .unwrap(); + assert_eq!(replay.status, "rejected"); + assert_eq!(replay.logical_effect_count, 0); + assert_eq!(state.last_controller_command_seq, 513); + assert!(!state + .outbox + .iter() + .any(|event| event.event_type == "workspace.ready")); + let (restored, recovered) = store.load_or_create(&config).unwrap(); + assert!(recovered); + assert_eq!( + restored.processed_commands.len(), + MAX_RECENT_PROCESSED_COMMANDS + ); + let _ = fs::remove_dir_all(root); + } + + #[test] + fn malicious_loopback_listener_gets_no_capability_and_cannot_forge_control() { + let root = temporary_root("malicious-loopback"); + let config = config(&root); + let store = DurableStateStore::new(&root).unwrap(); + let (state, _) = store.load_or_create(&config).unwrap(); + let secret = "bootstrap-malicious-listener-regression"; + let credential = CredentialMaterial::from_token(secret); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let (wire_sender, wire_receiver) = mpsc::channel(); + let server = thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut request = Vec::new(); + let mut byte = [0_u8; 1]; + while !request.ends_with(b"\r\n\r\n") { + stream.read_exact(&mut byte).unwrap(); + request.push(byte[0]); + } + stream + .write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + ) + .unwrap(); + stream.flush().unwrap(); + let (hello, frame) = read_masked_client_text(&mut stream); + let hello_payload = hello.get("payload").unwrap(); + let challenge_payload = json!({ + "credentialId": hello_payload.get("credentialId").unwrap(), + "credentialKind": "bootstrap", + "clientNonce": hello_payload.get("clientNonce").unwrap(), + "serverNonce": "malicious-server-nonce", + "runnerInstanceId": hello_payload.get("runnerInstanceId").unwrap(), + "environmentLeaseId": hello_payload.get("environmentLeaseId").unwrap(), + "runId": hello_payload.get("runId").unwrap(), + "normalizedSessionId": hello_payload.get("normalizedSessionId").unwrap(), + "turnId": hello_payload.get("turnId").unwrap(), + "itemId": hello_payload.get("itemId").unwrap(), + "runnerVersion": hello_payload.get("runnerVersion").unwrap(), + "runnerDigest": hello_payload.get("runnerDigest").unwrap(), + "selectedVersion": PROTOCOL_VERSION, + "credentialLeaseId": Value::Null, + "credentialExpiresAt": "2099-01-01T00:00:00.000Z", + "credentialExpiresAtUnixMs": 4_070_908_800_000_u64, + "revocationEpoch": 0, + "serverProof": "00".repeat(32), + }); + send_server_json( + &mut stream, + &json!({ + "protocol": PROTOCOL, + "version": PROTOCOL_VERSION, + "kind": "auth_challenge", + "payload": challenge_payload, + }), + ); + request.extend_from_slice(&frame); + wire_sender.send(request).unwrap(); + }); + let target = + resolve_ws_target_with(&format!("ws://{address}/phase3/connect"), |_host, _port| { + Ok(vec![address]) + }) + .unwrap(); + let mut client = WsClient::connect(&target, config.max_frame_bytes).unwrap(); + let error = authenticate_transport( + &mut client, + &state, + &config, + &credential, + "bootstrap", + None, + None, + None, + ) + .unwrap_err(); + assert!(error + .to_string() + .contains("transport authentication proof is invalid")); + assert!(client.secure_channel.is_none()); + let wire = wire_receiver.recv().unwrap(); + assert!(!wire + .windows(secret.len()) + .any(|window| window == secret.as_bytes())); + assert_eq!(state.acked_source_seq, 0); + assert!(state.processed_commands.is_empty()); + server.join().unwrap(); + let _ = fs::remove_dir_all(root); + } + + #[test] + fn non_terminal_command_failure_restores_exact_prior_state_on_every_retry() { + let root = temporary_root("command-atomic-p1"); + let mut config = config(&root); + let sizing_state = DurableRunnerState::new(&config); + let accepted_size = serde_json::to_vec(&event_envelope( + &sizing_state, + 1, + "turn.accepted", + 0, + json!({ "turnId": sizing_state.turn_id, "sameSession": true }), + None, + )) + .unwrap() + .len(); + let started_size = serde_json::to_vec(&event_envelope( + &sizing_state, + 2, + "item.started", + 1, + json!({ "kind": "assistant_message" }), + Some(&sizing_state.item_id), + )) + .unwrap() + .len(); + let non_p0_limit = accepted_size + started_size - 1; + config.p0_reserve_bytes = 8 * 1024; + config.max_outbox_bytes = non_p0_limit + config.p0_reserve_bytes; + + let store = DurableStateStore::new(&root).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + let before_state = serde_json::to_vec(&state).unwrap(); + let before_file = fs::read(store.path()).unwrap(); + let value = command("command_tight_p1", 1, "turn.start", json!({})); + + let first_error = process_command(&mut state, &store, &config, &value).unwrap_err(); + assert_eq!(serde_json::to_vec(&state).unwrap(), before_state); + assert_eq!(fs::read(store.path()).unwrap(), before_file); + let second_error = process_command(&mut state, &store, &config, &value).unwrap_err(); + assert_eq!(second_error, first_error); + assert_eq!(serde_json::to_vec(&state).unwrap(), before_state); + assert_eq!(fs::read(store.path()).unwrap(), before_file); + assert!(state.outbox.is_empty()); + assert!(state.processed_commands.is_empty()); + let _ = fs::remove_dir_all(root); + } + + #[test] + fn p0_exhaustion_commits_only_unrecoverable_result_and_retry_is_stable() { + let root = temporary_root("command-atomic-p0"); + let mut probe_config = config(&root); + probe_config.max_outbox_bytes = 128 * 1024; + probe_config.p0_reserve_bytes = 1; + let mut probe = DurableRunnerState::new(&probe_config); + execute_command_effect( + &mut probe, + &probe_config, + "turn.start", + &json!({ "text": "Phase 3" }), + ) + .unwrap(); + assert!(probe.outbox.len() >= 5); + let bytes_before_mandatory_p0 = probe + .outbox + .iter() + .take(4) + .map(|event| event.byte_size) + .sum::(); + + let mut config = config(&root); + config.p0_reserve_bytes = 1; + config.max_outbox_bytes = bytes_before_mandatory_p0 + 1; + let store = DurableStateStore::new(&root).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + let value = command( + "command_tight_p0", + 1, + "turn.start", + json!({ "text": "Phase 3" }), + ); + + let first = process_command(&mut state, &store, &config, &value).unwrap(); + assert_eq!(first.status, "failed"); + assert_eq!(first.logical_effect_count, 0); + assert_eq!(state.lifecycle, "unrecoverable"); + assert_eq!( + state.unrecoverable_outcome.as_deref(), + Some("p0_storage_exhausted") + ); + assert_eq!(state.next_source_seq, 1); + assert!(state.outbox.is_empty()); + assert_eq!(state.processed_commands.len(), 1); + let after_first = serde_json::to_vec(&state).unwrap(); + let second = process_command(&mut state, &store, &config, &value).unwrap(); + assert_eq!(second.result, first.result); + assert_eq!(serde_json::to_vec(&state).unwrap(), after_first); + let (persisted, recovered) = store.load_or_create(&config).unwrap(); + assert!(recovered); + assert!(persisted.outbox.is_empty()); + assert_eq!(persisted.next_source_seq, 1); + assert_eq!(persisted.processed_commands.len(), 1); + let _ = fs::remove_dir_all(root); + } + #[test] fn pressure_coalesces_p2_preserves_p0_and_stays_bounded() { let root = temporary_root("pressure"); @@ -1989,6 +3026,51 @@ mod tests { let _ = fs::remove_dir_all(root); } + #[test] + fn oversized_coalesced_p2_retains_prior_event_and_never_exceeds_bounds() { + let root = temporary_root("oversized-p2-coalesce"); + let mut config = config(&root); + config.max_outbox_bytes = 8 * 1024; + config.p0_reserve_bytes = 4 * 1024; + let mut state = DurableRunnerState::new(&config); + let item_id = state.item_id.clone(); + enqueue_event( + &mut state, + &config, + "item.delta", + 2, + json!({ "text": "small durable delta" }), + Some(&item_id), + ) + .unwrap(); + let prior = state.outbox[0].clone(); + + enqueue_event( + &mut state, + &config, + "item.delta", + 2, + json!({ "text": "x".repeat(16 * 1024) }), + Some(&item_id), + ) + .unwrap(); + + let retained = state + .outbox + .iter() + .find(|event| event.event_type == "item.delta") + .unwrap(); + assert_eq!(retained, &prior); + assert!(state.backpressure); + assert!(state + .outbox + .iter() + .any(|event| event.priority == 0 && event.event_type == "runner.backpressure")); + assert!(state.outbox_bytes() <= config.max_outbox_bytes); + assert!(state.peak_outbox_bytes <= config.max_outbox_bytes); + let _ = fs::remove_dir_all(root); + } + #[test] fn drain_rejects_new_turn_and_revoke_preserves_unacked_events() { let root = temporary_root("drain"); diff --git a/packages/paperclip-runner/src/contracts/phase3.ts b/packages/paperclip-runner/src/contracts/phase3.ts index 0dd50d00d83..0a3e6b5a2bf 100644 --- a/packages/paperclip-runner/src/contracts/phase3.ts +++ b/packages/paperclip-runner/src/contracts/phase3.ts @@ -62,6 +62,8 @@ export interface Phase3RunnerState extends Phase3Identity { peakOutboxBytes: number; outbox: Phase3StoredEvent[]; processedCommands: Record; + compactedCommandFilter: string; + compactedCommandCount: number; diagnostics: string[]; backpressure: boolean; recoverableFailure: string | null; diff --git a/packages/paperclip-runner/src/mock-core/phase3-recovery.test.ts b/packages/paperclip-runner/src/mock-core/phase3-recovery.test.ts index cd2731a8c37..e79bdf8c208 100644 --- a/packages/paperclip-runner/src/mock-core/phase3-recovery.test.ts +++ b/packages/paperclip-runner/src/mock-core/phase3-recovery.test.ts @@ -1,3 +1,4 @@ +import { spawn } from "node:child_process"; import { chmodSync, lstatSync, @@ -20,38 +21,7 @@ import { runPhase3Recovery, } from "./phase3-recovery.js"; -async function upgradeStatus(url: string, ticket: string): Promise { - const parsed = new URL(url); - return new Promise((resolveStatus, rejectStatus) => { - const socket = connect(Number(parsed.port), parsed.hostname); - let response = ""; - socket.once("error", rejectStatus); - socket.once("connect", () => { - socket.write( - [ - `GET ${parsed.pathname} HTTP/1.1`, - `Host: ${parsed.host}`, - "Upgrade: websocket", - "Connection: Upgrade", - "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==", - "Sec-WebSocket-Version: 13", - `Authorization: Bearer ${ticket}`, - "\r\n", - ].join("\r\n"), - ); - }); - socket.on("data", (chunk) => { - response += chunk.toString("utf8"); - if (response.includes("\r\n\r\n")) { - const status = Number(response.match(/^HTTP\/1\.1 (\d{3})/)?.[1]); - socket.destroy(); - resolveStatus(status); - } - }); - }); -} - -async function upgradeSocket(url: string, ticket: string): Promise { +async function upgradeSocket(url: string): Promise { const parsed = new URL(url); return new Promise((resolveSocket, rejectSocket) => { const socket = connect(Number(parsed.port), parsed.hostname); @@ -81,7 +51,6 @@ async function upgradeSocket(url: string, ticket: string): Promise { "Connection: Upgrade", "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==", "Sec-WebSocket-Version: 13", - `Authorization: Bearer ${ticket}`, "\r\n", ].join("\r\n"), ); @@ -90,6 +59,39 @@ async function upgradeSocket(url: string, ticket: string): Promise { }); } +async function receiveServerJson(socket: Socket): Promise> { + return new Promise((resolveValue, rejectValue) => { + let buffer = Buffer.alloc(0); + const timer = setTimeout(() => rejectValue(new Error("Server frame timed out.")), 2_000); + const onData = (chunk: Buffer): void => { + buffer = Buffer.concat([buffer, chunk]); + if (buffer.length < 2) return; + let length = buffer[1]! & 0x7f; + let cursor = 2; + if (length === 126) { + if (buffer.length < 4) return; + length = buffer.readUInt16BE(2); + cursor = 4; + } else if (length === 127) { + if (buffer.length < 10) return; + length = Number(buffer.readBigUInt64BE(2)); + cursor = 10; + } + if (buffer.length < cursor + length) return; + clearTimeout(timer); + socket.off("data", onData); + resolveValue( + JSON.parse(buffer.subarray(cursor, cursor + length).toString("utf8")) as Record< + string, + unknown + >, + ); + }; + socket.on("data", onData); + socket.once("error", rejectValue); + }); +} + function sendMaskedJson(socket: Socket, value: unknown): void { const payload = Buffer.from(JSON.stringify(value)); const mask = Buffer.from([0x11, 0x22, 0x33, 0x44]); @@ -116,23 +118,93 @@ async function expectSocketClosed(socket: Socket): Promise { }); } -function hello(identity: ReturnType): Record { +function authHello( + identity: ReturnType, + credentialId: string, +): Record { return { protocol: "paperclip.runner", version: 1, - envelopeId: "hello_auth_regression", - kind: "hello", - runnerInstanceId: identity.runnerInstanceId, + kind: "auth_hello", payload: { + credentialId, + clientNonce: "client_nonce_auth_regression", protocolMin: 1, protocolMax: 1, + runnerInstanceId: identity.runnerInstanceId, runnerVersion: "0.3.0", runnerDigest: "sha256:phase3-approved", environmentLeaseId: identity.environmentLeaseId, + runId: identity.runId, + normalizedSessionId: identity.normalizedSessionId, + turnId: identity.turnId, + itemId: identity.itemId, }, }; } +async function runRunnerProcess(options: { + connectUrl: string; + stateDirectory: string; + identity: ReturnType; + ticket: string; + maxRuntimeMs: number; +}): Promise<{ code: number | null; stderr: string }> { + const child = spawn( + phase3Internals.runnerBinary, + [ + "--connect-url", + options.connectUrl, + "--state-dir", + options.stateDirectory, + "--runner-id", + options.identity.runnerInstanceId, + "--environment-lease-id", + options.identity.environmentLeaseId, + "--run-id", + options.identity.runId, + "--session-id", + options.identity.normalizedSessionId, + "--turn-id", + options.identity.turnId, + "--item-id", + options.identity.itemId, + "--runner-version", + "0.3.0", + "--runner-digest", + "sha256:phase3-approved", + "--max-outbox-bytes", + String(64 * 1024), + "--p0-reserve-bytes", + String(32 * 1024), + "--reconnect-delay-ms", + "10", + "--max-runtime-ms", + String(options.maxRuntimeMs), + ], + { + env: phase3Internals.runnerEnvironment(options.ticket), + stdio: "pipe", + }, + ); + let stderr = ""; + child.stderr.setEncoding("utf8").on("data", (chunk: string) => { + stderr += chunk; + }); + const code = await new Promise((resolveExit, rejectExit) => { + const timer = setTimeout(() => { + child.kill("SIGKILL"); + rejectExit(new Error("Phase 3 runner process did not terminate.")); + }, options.maxRuntimeMs + 3_000); + child.once("error", rejectExit); + child.once("exit", (exitCode) => { + clearTimeout(timer); + resolveExit(exitCode); + }); + }); + return { code, stderr }; +} + describe.sequential("Phase 3 durable transport and recovery", () => { it("accepts a short-lived bootstrap ticket once and rejects its reuse", async () => { const scratch = @@ -147,19 +219,50 @@ describe.sequential("Phase 3 durable transport and recovery", () => { }); try { await core.start(); + core.queueCommand("runner.shutdown", {}); const ticket = core.issueBootstrapTicket(); + const stateDirectory = resolve(root, "runner"); + + const first = await runRunnerProcess({ + connectUrl: core.connectUrl, + stateDirectory, + identity: core.identity, + ticket, + maxRuntimeMs: 2_000, + }); + expect(first.code, first.stderr).toBe(0); + expect(core.store.state.connectionCount).toBe(1); + + const reused = await runRunnerProcess({ + connectUrl: core.connectUrl, + stateDirectory, + identity: core.identity, + ticket, + maxRuntimeMs: 100, + }); + expect(reused.code).not.toBe(0); + expect(core.store.state.connectionCount).toBe(1); - await expect(upgradeStatus(core.connectUrl, ticket)).resolves.toBe(101); - await expect(upgradeStatus(core.connectUrl, ticket)).resolves.toBe(401); const expiredTicket = core.issueBootstrapTicket(-1); - await expect(upgradeStatus(core.connectUrl, expiredTicket)).resolves.toBe(401); + const expired = await runRunnerProcess({ + connectUrl: core.connectUrl, + stateDirectory: resolve(root, "expired-runner"), + identity: core.identity, + ticket: expiredTicket, + maxRuntimeMs: 100, + }); + expect(expired.code).not.toBe(0); + expect(core.store.state.connectionCount).toBe(1); + expect( + Object.values(core.store.state.tickets).filter((record) => record.usedAt !== null), + ).toHaveLength(1); } finally { await core.stop(); rmSync(root, { recursive: true, force: true }); } }); - it("rejects bootstrap and resumed lease credentials bound to a different runner identity", async () => { + it("requires proof before consuming a ticket and rejects mismatched session identity", async () => { const scratch = process.env.PAPERCLIP_RUN_SCRATCH_DIR ?? process.env.PAPERCLIP_SCRATCH_DIR ?? @@ -175,26 +278,31 @@ describe.sequential("Phase 3 durable transport and recovery", () => { await core.start(); const bootstrap = core.issueBootstrapTicket(); - const bootstrapSocket = await upgradeSocket(core.connectUrl, bootstrap); - sendMaskedJson(bootstrapSocket, { - ...hello(identity), - runnerInstanceId: "runner_wrong_identity", - }); - await expectSocketClosed(bootstrapSocket); + const credentialId = phase3Internals.credentialMaterial(bootstrap).credentialId; + const wrongIdentitySocket = await upgradeSocket(core.connectUrl); + const wrongHello = authHello(identity, credentialId); + (wrongHello.payload as Record).runId = "run_wrong_identity"; + sendMaskedJson(wrongIdentitySocket, wrongHello); + await expectSocketClosed(wrongIdentitySocket); + expect(core.store.state.tickets[credentialId]?.usedAt).toBeNull(); - const leaseToken = "lease_wrong_identity_regression"; - const digest = phase3Internals.tokenDigest(leaseToken); - core.store.state.leases[digest] = { - digest, - leaseId: "lease_wrong_identity", - runnerInstanceId: "runner_other_identity", - expiresAt: new Date(Date.now() + 30_000).toISOString(), - revokedAt: null, - }; - core.store.save(); - const leaseSocket = await upgradeSocket(core.connectUrl, leaseToken); - sendMaskedJson(leaseSocket, hello(identity)); - await expectSocketClosed(leaseSocket); + const forgedProofSocket = await upgradeSocket(core.connectUrl); + sendMaskedJson(forgedProofSocket, authHello(identity, credentialId)); + const challenge = await receiveServerJson(forgedProofSocket); + const challengePayload = challenge.payload as Record; + sendMaskedJson(forgedProofSocket, { + protocol: "paperclip.runner", + version: 1, + kind: "auth_response", + payload: { + credentialId, + clientNonce: challengePayload.clientNonce, + serverNonce: challengePayload.serverNonce, + clientProof: "00".repeat(32), + }, + }); + await expectSocketClosed(forgedProofSocket); + expect(core.store.state.tickets[credentialId]?.usedAt).toBeNull(); } finally { await core.stop(); rmSync(root, { recursive: true, force: true }); @@ -295,6 +403,100 @@ describe.sequential("Phase 3 durable transport and recovery", () => { expect(trace.assertions.sourceCursorContinuous).toBe(true); }); + it("keeps a restarted revoked runner network-silent and byte-stable with a fresh ticket", async () => { + const scratch = + process.env.PAPERCLIP_RUN_SCRATCH_DIR ?? + process.env.PAPERCLIP_SCRATCH_DIR ?? + tmpdir(); + const root = mkdtempSync(resolve(scratch, "phase3-revoked-restart-test-")); + const runnerStateDirectory = resolve(root, "runner"); + const identity = phase3Internals.phase3Identity(); + mkdirSync(runnerStateDirectory, { recursive: true, mode: 0o700 }); + const envelope = { + protocol: "paperclip.runner", + version: 1, + envelopeId: "envelope_event_000004", + kind: "event", + runnerInstanceId: identity.runnerInstanceId, + payload: { + sourceEventId: `event_${identity.runnerInstanceId}_000004`, + sourceSeq: 4, + eventType: "run.terminal", + priority: 0, + payload: { status: "succeeded" }, + }, + }; + const byteSize = Buffer.byteLength(JSON.stringify(envelope)); + const statePath = resolve(runnerStateDirectory, "runner-state.json"); + writeFileSync( + statePath, + JSON.stringify( + { + schema: "paperclip.runner.phase3.state.v1", + ...identity, + lifecycle: "revoked", + nextSourceSeq: 5, + ackedSourceSeq: 3, + lastControllerCommandSeq: 2, + reconnectCount: 7, + maxOutboxBytes: 64 * 1024, + peakOutboxBytes: byteSize, + outbox: [ + { + sourceSeq: 4, + sourceEventId: `event_${identity.runnerInstanceId}_000004`, + priority: 0, + eventType: "run.terminal", + itemId: null, + envelope, + byteSize, + }, + ], + processedCommands: {}, + compactedCommandFilter: "00".repeat(4_096), + compactedCommandCount: 0, + diagnostics: ["connection lease revoked; preserving durable events"], + backpressure: false, + recoverableFailure: null, + unrecoverableOutcome: null, + harnessGeneration: 1, + stopAfterFlush: true, + }, + null, + 2, + ), + { mode: 0o600 }, + ); + const before = readFileSync(statePath); + const core = new Phase3MockCore({ + stateDirectory: resolve(root, "mock-core"), + identity, + fault: "none", + }); + try { + await core.start(); + const ticket = core.issueBootstrapTicket(); + const connectionsBefore = core.store.state.connectionCount; + const result = await runRunnerProcess({ + connectUrl: core.connectUrl, + stateDirectory: runnerStateDirectory, + identity, + ticket, + maxRuntimeMs: 1_000, + }); + + expect(result.code, result.stderr).toBe(0); + expect(core.store.state.connectionCount).toBe(connectionsBefore); + expect(Object.values(core.store.state.tickets).map((record) => record.usedAt)).toEqual([ + null, + ]); + expect(readFileSync(statePath)).toEqual(before); + } finally { + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } + }); + it("reconciles a restarted harness without replacing session, turn, or item IDs", async () => { const trace = await runPhase3Recovery({ fault: "harness-restart" }); const reconciled = trace.diagnostics.committedEvents.find( diff --git a/packages/paperclip-runner/src/mock-core/phase3-recovery.ts b/packages/paperclip-runner/src/mock-core/phase3-recovery.ts index 62d1594dc44..b5021b607f8 100644 --- a/packages/paperclip-runner/src/mock-core/phase3-recovery.ts +++ b/packages/paperclip-runner/src/mock-core/phase3-recovery.ts @@ -1,5 +1,12 @@ import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; -import { createHash, randomUUID } from "node:crypto"; +import { + createCipheriv, + createDecipheriv, + createHash, + createHmac, + randomUUID, + timingSafeEqual, +} from "node:crypto"; import { chmodSync, closeSync, @@ -51,22 +58,31 @@ const fakeHarnessScript = resolve( ); const protocol = "paperclip.runner"; const protocolVersion = 1; +const secureFrameSchema = "paperclip.runner.secure-frame.v1"; const websocketGuid = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; const coreStateSchema = "paperclip.runner.phase3.mock-core-state.v1"; const maxFrameBytes = 1024 * 1024; interface BootstrapTicketRecord { - digest: string; - runnerInstanceId: string; + credentialId: string; + authKeyDigest: string; + identity: Phase3Identity; + runnerVersion: string; + runnerDigest: string; expiresAt: string; + expiresAtUnixMs: number; usedAt: string | null; } interface ConnectionLeaseRecord { - digest: string; + credentialId: string; + authKeyDigest: string; leaseId: string; - runnerInstanceId: string; + identity: Phase3Identity; + protocolVersion: number; expiresAt: string; + expiresAtUnixMs: number; + revocationEpoch: number; revokedAt: string | null; } @@ -88,20 +104,34 @@ interface StoredCoreState { lastLeaseExpiresAt: string | null; } -type Authorization = +type PendingAuthorization = | { kind: "bootstrap"; - digest: string; - rawToken: string; + authKey: Buffer; ticket: BootstrapTicketRecord; } | { kind: "lease"; - digest: string; - rawToken: string; + authKey: Buffer; lease: ConnectionLeaseRecord; }; +interface PendingChallenge { + authorization: PendingAuthorization; + canonicalChallenge: string; + serverProof: string; + clientNonce: string; + serverNonce: string; +} + +interface SecureChannel { + sendKey: Buffer; + receiveKey: Buffer; + sendCounter: bigint; + receiveCounter: bigint; + sessionId: string; +} + interface MockCoreOptions { stateDirectory: string; identity: Phase3Identity; @@ -122,8 +152,36 @@ interface RunnerProcessHandle { completion: Promise; } +function domainDigest(domain: string, parts: readonly Buffer[]): Buffer { + const digest = createHash("sha256").update(domain).update(Buffer.from([0])); + for (const part of parts) { + const length = Buffer.alloc(8); + length.writeBigUInt64BE(BigInt(part.length)); + digest.update(length).update(part); + } + return digest.digest(); +} + +function domainHmac(key: Buffer, domain: string, parts: readonly Buffer[]): Buffer { + const digest = createHmac("sha256", key).update(domain).update(Buffer.from([0])); + for (const part of parts) { + const length = Buffer.alloc(8); + length.writeBigUInt64BE(BigInt(part.length)); + digest.update(length).update(part); + } + return digest.digest(); +} + +function credentialMaterial(token: string): { credentialId: string; authKey: Buffer } { + const bytes = Buffer.from(token); + return { + credentialId: `sha256:${domainDigest("paperclip-runner-credential-id-v1", [bytes]).toString("hex")}`, + authKey: domainDigest("paperclip-runner-auth-key-v1", [bytes]), + }; +} + function tokenDigest(token: string): string { - return `sha256:${createHash("sha256").update(token).digest("hex")}`; + return `sha256:${credentialMaterial(token).authKey.toString("hex")}`; } function canonicalJson(value: unknown): string { @@ -144,6 +202,103 @@ function safeDate(offsetMs: number): string { return new Date(Date.now() + offsetMs).toISOString(); } +function authKeyFromDigest(digest: string): Buffer { + const hex = digest.match(/^sha256:([0-9a-f]{64})$/)?.[1]; + if (hex === undefined) throw new Error("Stored transport authentication key is malformed."); + return Buffer.from(hex, "hex"); +} + +function proofMatches(expected: Buffer, supplied: unknown): boolean { + if (typeof supplied !== "string" || !/^[0-9a-f]{64}$/.test(supplied)) return false; + return timingSafeEqual(expected, Buffer.from(supplied, "hex")); +} + +function createSecureChannel( + authKey: Buffer, + canonicalChallenge: string, + serverProof: string, + clientProof: string, +): SecureChannel { + const parts = [ + Buffer.from(canonicalChallenge), + Buffer.from(serverProof), + Buffer.from(clientProof), + ]; + const binding = domainDigest("paperclip-runner-session-binding-v1", parts); + return { + sendKey: domainHmac(authKey, "paperclip-runner-core-to-client-key-v1", [binding]), + receiveKey: domainHmac(authKey, "paperclip-runner-client-to-core-key-v1", [binding]), + sendCounter: 0n, + receiveCounter: 0n, + sessionId: `sha256:${binding.toString("hex")}`, + }; +} + +function secureNonce(prefix: "P3C1" | "P3S1", counter: bigint): Buffer { + const nonce = Buffer.alloc(12); + nonce.write(prefix, 0, "ascii"); + nonce.writeBigUInt64BE(counter, 4); + return nonce; +} + +function secureAad( + channel: SecureChannel, + direction: "client_to_core" | "core_to_client", + counter: bigint, +): Buffer { + return Buffer.from(`${secureFrameSchema}\0${channel.sessionId}\0${direction}\0${counter}`); +} + +function encryptSecureJson(channel: SecureChannel, value: unknown): Record { + const counter = channel.sendCounter; + const cipher = createCipheriv("aes-256-gcm", channel.sendKey, secureNonce("P3S1", counter)); + cipher.setAAD(secureAad(channel, "core_to_client", counter)); + const ciphertext = Buffer.concat([ + cipher.update(Buffer.from(JSON.stringify(value))), + cipher.final(), + cipher.getAuthTag(), + ]); + channel.sendCounter += 1n; + return { + schema: secureFrameSchema, + counter: Number(counter), + ciphertext: ciphertext.toString("hex"), + }; +} + +function decryptSecureJson(channel: SecureChannel, value: unknown): Record { + if (typeof value !== "object" || value === null) { + throw new Error("Secure frame must be an object."); + } + const frame = value as Record; + if ( + frame.schema !== secureFrameSchema || + typeof frame.counter !== "number" || + !Number.isSafeInteger(frame.counter) || + BigInt(frame.counter) !== channel.receiveCounter || + typeof frame.ciphertext !== "string" || + !/^[0-9a-f]+$/.test(frame.ciphertext) || + frame.ciphertext.length % 2 !== 0 + ) { + throw new Error("Secure frame metadata or counter is invalid."); + } + const sealed = Buffer.from(frame.ciphertext, "hex"); + if (sealed.length < 16) throw new Error("Secure frame authentication tag is missing."); + const ciphertext = sealed.subarray(0, -16); + const tag = sealed.subarray(-16); + const counter = channel.receiveCounter; + const decipher = createDecipheriv( + "aes-256-gcm", + channel.receiveKey, + secureNonce("P3C1", counter), + ); + decipher.setAAD(secureAad(channel, "client_to_core", counter)); + decipher.setAuthTag(tag); + const plaintext = Buffer.concat([decipher.update(ciphertext), decipher.final()]); + channel.receiveCounter += 1n; + return JSON.parse(plaintext.toString("utf8")) as Record; +} + function initialCoreState(identity: Phase3Identity): StoredCoreState { return { schema: coreStateSchema, @@ -304,8 +459,10 @@ class DurableCoreStore { class MockWebSocketConnection { readonly socket: Duplex; - readonly authorization: Authorization; + pendingChallenge: PendingChallenge | null = null; + secureChannel: SecureChannel | null = null; lease: ConnectionLeaseRecord | null = null; + connectionId: string | null = null; #buffer = Buffer.alloc(0); #closed = false; #onText: (text: string) => void; @@ -313,12 +470,10 @@ class MockWebSocketConnection { constructor( socket: Duplex, - authorization: Authorization, onText: (text: string) => void, onClose: () => void, ) { this.socket = socket; - this.authorization = authorization; this.#onText = onText; this.#onClose = onClose; socket.on("data", (chunk: Buffer) => this.#consume(chunk)); @@ -332,7 +487,8 @@ class MockWebSocketConnection { } sendJson(value: unknown): void { - this.sendText(JSON.stringify(value)); + const wire = this.secureChannel === null ? value : encryptSecureJson(this.secureChannel, value); + this.sendText(JSON.stringify(wire)); } sendText(text: string): void { @@ -492,11 +648,16 @@ export class Phase3MockCore { issueBootstrapTicket(ttlMs = 5_000): string { const ticket = `bootstrap_${randomUUID()}`; - const digest = tokenDigest(ticket); - this.store.state.tickets[digest] = { - digest, - runnerInstanceId: this.identity.runnerInstanceId, - expiresAt: safeDate(ttlMs), + const material = credentialMaterial(ticket); + const expiresAtUnixMs = Date.now() + ttlMs; + this.store.state.tickets[material.credentialId] = { + credentialId: material.credentialId, + authKeyDigest: `sha256:${material.authKey.toString("hex")}`, + identity: structuredClone(this.identity), + runnerVersion: this.#expectedRunnerVersion, + runnerDigest: this.#expectedRunnerDigest, + expiresAt: new Date(expiresAtUnixMs).toISOString(), + expiresAtUnixMs, usedAt: null, }; this.store.state.freshBootstraps += 1; @@ -542,12 +703,6 @@ export class Phase3MockCore { socket.destroy(); return; } - const authorization = this.#authorize(request.headers.authorization); - if (authorization === null) { - socket.write("HTTP/1.1 401 Unauthorized\r\nConnection: close\r\n\r\n"); - socket.destroy(); - return; - } const websocketKey = request.headers["sec-websocket-key"]; if (typeof websocketKey !== "string") { socket.write("HTTP/1.1 400 Bad Request\r\nConnection: close\r\n\r\n"); @@ -568,38 +723,20 @@ export class Phase3MockCore { ); const connection = new MockWebSocketConnection( socket, - authorization, (text) => this.#handleText(connection, text), () => this.#connections.delete(connection), ); this.#connections.add(connection); } - #authorize(header: string | undefined): Authorization | null { - const rawToken = header?.match(/^Bearer (.+)$/i)?.[1]; - if (rawToken === undefined) return null; - const digest = tokenDigest(rawToken); - const ticket = this.store.state.tickets[digest]; - if (ticket !== undefined && ticket.usedAt === null && Date.parse(ticket.expiresAt) > Date.now()) { - ticket.usedAt = new Date().toISOString(); - this.store.save(); - return { kind: "bootstrap", digest, rawToken, ticket }; - } - const lease = this.store.state.leases[digest]; - if ( - lease !== undefined && - lease.revokedAt === null && - Date.parse(lease.expiresAt) > Date.now() - ) { - return { kind: "lease", digest, rawToken, lease }; - } - return null; - } - #handleText(connection: MockWebSocketConnection, text: string): void { let envelope: Record; try { - envelope = JSON.parse(text) as Record; + const wire = JSON.parse(text) as unknown; + envelope = + connection.secureChannel === null + ? (wire as Record) + : decryptSecureJson(connection.secureChannel, wire); } catch { this.store.state.malformedFrames += 1; this.store.save(); @@ -611,11 +748,15 @@ export class Phase3MockCore { return; } const kind = envelope.kind; - if (kind === "hello") { - this.#hello(connection, envelope); + if (connection.secureChannel === null && kind === "auth_hello") { + this.#authHello(connection, envelope); return; } - if (connection.lease === null) { + if (connection.secureChannel === null && kind === "auth_response") { + this.#authResponse(connection, envelope); + return; + } + if (connection.secureChannel === null || connection.lease === null) { connection.close(); return; } @@ -632,42 +773,165 @@ export class Phase3MockCore { } } - #hello(connection: MockWebSocketConnection, envelope: Record): void { - const payload = envelope.payload as Record | undefined; - const authorizedRunnerInstanceId = - connection.authorization.kind === "bootstrap" - ? connection.authorization.ticket.runnerInstanceId - : connection.authorization.lease.runnerInstanceId; + #authorizeHello(payload: Record): PendingAuthorization | null { + const credentialId = payload.credentialId; + if (typeof credentialId !== "string") return null; + const ticket = this.store.state.tickets[credentialId]; + const lease = this.store.state.leases[credentialId]; + const authorization: PendingAuthorization | null = + ticket !== undefined && ticket.usedAt === null && ticket.expiresAtUnixMs > Date.now() + ? { kind: "bootstrap", authKey: authKeyFromDigest(ticket.authKeyDigest), ticket } + : lease !== undefined && + lease.revokedAt === null && + lease.expiresAtUnixMs > Date.now() + ? { kind: "lease", authKey: authKeyFromDigest(lease.authKeyDigest), lease } + : null; + if (authorization === null) return null; + const binding = authorization.kind === "bootstrap" ? authorization.ticket : authorization.lease; + const identity = binding.identity; if ( - envelope.runnerInstanceId !== this.identity.runnerInstanceId || - envelope.runnerInstanceId !== authorizedRunnerInstanceId || - payload?.environmentLeaseId !== this.identity.environmentLeaseId || - payload?.runnerVersion !== this.#expectedRunnerVersion || - payload?.runnerDigest !== this.#expectedRunnerDigest || - payload?.protocolMin !== 1 || - payload.protocolMax !== 1 + payload.runnerInstanceId !== identity.runnerInstanceId || + payload.environmentLeaseId !== identity.environmentLeaseId || + payload.runId !== identity.runId || + payload.normalizedSessionId !== identity.normalizedSessionId || + payload.turnId !== identity.turnId || + payload.itemId !== identity.itemId || + payload.runnerVersion !== this.#expectedRunnerVersion || + payload.runnerDigest !== this.#expectedRunnerDigest || + payload.protocolMin !== 1 || + payload.protocolMax !== 1 || + (authorization.kind === "lease" && authorization.lease.protocolVersion !== protocolVersion) + ) { + return null; + } + return authorization; + } + + #authHello(connection: MockWebSocketConnection, envelope: Record): void { + if (connection.pendingChallenge !== null) { + connection.close(); + return; + } + const payload = envelope.payload as Record | undefined; + if (payload === undefined || typeof payload.clientNonce !== "string") { + connection.close(); + return; + } + const authorization = this.#authorizeHello(payload); + if (authorization === null) { + connection.close(); + return; + } + const credential = authorization.kind === "bootstrap" ? authorization.ticket : authorization.lease; + const serverNonce = randomUUID(); + const challengePayload: Record = { + credentialId: credential.credentialId, + credentialKind: authorization.kind, + clientNonce: payload.clientNonce, + serverNonce, + runnerInstanceId: payload.runnerInstanceId, + environmentLeaseId: payload.environmentLeaseId, + runId: payload.runId, + normalizedSessionId: payload.normalizedSessionId, + turnId: payload.turnId, + itemId: payload.itemId, + runnerVersion: payload.runnerVersion, + runnerDigest: payload.runnerDigest, + selectedVersion: protocolVersion, + credentialLeaseId: authorization.kind === "lease" ? authorization.lease.leaseId : null, + credentialExpiresAt: credential.expiresAt, + credentialExpiresAtUnixMs: credential.expiresAtUnixMs, + revocationEpoch: authorization.kind === "lease" ? authorization.lease.revocationEpoch : 0, + }; + const canonicalChallenge = canonicalJson(challengePayload); + const serverProof = domainHmac( + authorization.authKey, + "paperclip-runner-server-proof-v1", + [Buffer.from(canonicalChallenge)], + ).toString("hex"); + connection.pendingChallenge = { + authorization, + canonicalChallenge, + serverProof, + clientNonce: payload.clientNonce, + serverNonce, + }; + connection.sendJson({ + protocol, + version: protocolVersion, + kind: "auth_challenge", + payload: { ...challengePayload, serverProof }, + }); + } + + #authResponse(connection: MockWebSocketConnection, envelope: Record): void { + const pending = connection.pendingChallenge; + const payload = envelope.payload as Record | undefined; + if ( + pending === null || + payload === undefined || + payload.credentialId !== + (pending.authorization.kind === "bootstrap" + ? pending.authorization.ticket.credentialId + : pending.authorization.lease.credentialId) || + payload.clientNonce !== pending.clientNonce || + payload.serverNonce !== pending.serverNonce ) { connection.close(); return; } - - let leaseToken = connection.authorization.rawToken; + const expectedClientProof = domainHmac( + pending.authorization.authKey, + "paperclip-runner-client-proof-v1", + [Buffer.from(pending.canonicalChallenge), Buffer.from(pending.serverProof)], + ); + if (!proofMatches(expectedClientProof, payload.clientProof)) { + connection.close(); + return; + } + const clientProof = expectedClientProof.toString("hex"); + let leaseToken: string | null = null; let lease: ConnectionLeaseRecord; - if (connection.authorization.kind === "bootstrap") { + if (pending.authorization.kind === "bootstrap") { + pending.authorization.ticket.usedAt = new Date().toISOString(); leaseToken = `lease_${randomUUID()}`; - const digest = tokenDigest(leaseToken); + const material = credentialMaterial(leaseToken); + const ttlMs = this.fault === "lease-expiry" && !this.#faultTriggered ? 50 : 30_000; + const expiresAtUnixMs = Date.now() + ttlMs; lease = { - digest, + credentialId: material.credentialId, + authKeyDigest: `sha256:${material.authKey.toString("hex")}`, leaseId: `connection_lease_${randomUUID()}`, - runnerInstanceId: this.identity.runnerInstanceId, - expiresAt: safeDate(this.fault === "lease-expiry" && !this.#faultTriggered ? 50 : 30_000), + identity: structuredClone(this.identity), + protocolVersion, + expiresAt: new Date(expiresAtUnixMs).toISOString(), + expiresAtUnixMs, + revocationEpoch: 0, revokedAt: null, }; - this.store.state.leases[digest] = lease; + this.store.state.leases[material.credentialId] = lease; } else { - lease = connection.authorization.lease; + lease = pending.authorization.lease; } + connection.pendingChallenge = null; connection.lease = lease; + connection.connectionId = `connection_${this.store.state.connectionCount + 1}`; + connection.secureChannel = createSecureChannel( + pending.authorization.authKey, + pending.canonicalChallenge, + pending.serverProof, + clientProof, + ); + this.#welcome(connection, leaseToken); + } + + #welcome(connection: MockWebSocketConnection, leaseToken: string | null): void { + const lease = connection.lease; + if (lease === null || connection.connectionId === null) { + connection.close(); + return; + } + this.store.state.connectionCount += 1; this.store.state.lastLeaseId = lease.leaseId; this.store.state.lastLeaseExpiresAt = lease.expiresAt; @@ -691,14 +955,29 @@ export class Phase3MockCore { envelopeId: `welcome_${this.store.state.connectionCount}`, kind: "welcome", runnerInstanceId: this.identity.runnerInstanceId, - connectionId: `connection_${this.store.state.connectionCount}`, + environmentLeaseId: this.identity.environmentLeaseId, + runId: this.identity.runId, + normalizedSessionId: this.identity.normalizedSessionId, + turnId: this.identity.turnId, + itemId: this.identity.itemId, + connectionId: connection.connectionId, + connectionLeaseId: lease.leaseId, sentAt: new Date().toISOString(), payload: { selectedVersion: 1, heartbeatIntervalMs: 250, connectionLeaseId: lease.leaseId, - connectionLeaseToken: leaseToken, + ...(leaseToken === null ? {} : { connectionLeaseToken: leaseToken }), connectionLeaseExpiresAt: lease.expiresAt, + connectionLeaseExpiresAtUnixMs: lease.expiresAtUnixMs, + connectionLeaseRevocationEpoch: lease.revocationEpoch, + leaseBinding: { + runnerInstanceId: this.identity.runnerInstanceId, + environmentLeaseId: this.identity.environmentLeaseId, + runId: this.identity.runId, + normalizedSessionId: this.identity.normalizedSessionId, + protocolVersion, + }, maxFrameBytes, maxBatchEvents: 100, ackedSourceSeq: reportedAck, @@ -715,6 +994,7 @@ export class Phase3MockCore { } if (this.fault === "lease-expiry") { lease.expiresAt = safeDate(-1); + lease.expiresAtUnixMs = Date.now() - 1; this.store.state.lastLeaseExpiresAt = lease.expiresAt; this.store.save(); } @@ -732,23 +1012,47 @@ export class Phase3MockCore { return command === undefined ? [] : [command]; } + #controlEnvelope( + connection: MockWebSocketConnection, + envelopeId: string, + kind: string, + payload: Record, + ): Record { + if (connection.lease === null || connection.connectionId === null) { + throw new Error("Cannot send control data before transport authentication."); + } + return { + protocol, + version: protocolVersion, + envelopeId, + kind, + runnerInstanceId: this.identity.runnerInstanceId, + environmentLeaseId: this.identity.environmentLeaseId, + runId: this.identity.runId, + normalizedSessionId: this.identity.normalizedSessionId, + turnId: this.identity.turnId, + itemId: this.identity.itemId, + connectionId: connection.connectionId, + connectionLeaseId: connection.lease.leaseId, + sentAt: new Date().toISOString(), + payload, + }; + } + #sendNextCommand(connection: MockWebSocketConnection): void { const [command] = this.#nextPendingCommand(); if (command === undefined) return; this.store.state.commandDeliveryCounts[command.commandId] = (this.store.state.commandDeliveryCounts[command.commandId] ?? 0) + 1; this.store.save(); - connection.sendJson({ - protocol, - version: protocolVersion, - envelopeId: `command_${command.commandId}_${this.store.state.commandDeliveryCounts[command.commandId]}`, - kind: "command", - runnerInstanceId: this.identity.runnerInstanceId, - runId: this.identity.runId, - normalizedSessionId: this.identity.normalizedSessionId, - sentAt: new Date().toISOString(), - payload: this.#wireCommand(command), - }); + connection.sendJson( + this.#controlEnvelope( + connection, + `command_${command.commandId}_${this.store.state.commandDeliveryCounts[command.commandId]}`, + "command", + this.#wireCommand(command), + ), + ); } #commandResult( @@ -834,6 +1138,7 @@ export class Phase3MockCore { if (this.fault === "revoke" && eventType === "run.terminal" && !this.#faultTriggered) { if (connection.lease !== null) { connection.lease.revokedAt = new Date().toISOString(); + connection.lease.revocationEpoch += 1; } this.queueCommand( "turn.start", @@ -843,15 +1148,13 @@ export class Phase3MockCore { this.#triggerFault(); this.store.save(); // Revoke before ACK so the runner must flush a genuinely non-empty durable outbox. - connection.sendJson({ - protocol, - version: protocolVersion, - envelopeId: "revoke_phase3", - kind: "revoke", - runnerInstanceId: this.identity.runnerInstanceId, - sentAt: new Date().toISOString(), - payload: { reason: "fault_injection_revoke", drain: true }, - }); + connection.sendJson( + this.#controlEnvelope(connection, "revoke_phase3", "revoke", { + reason: "fault_injection_revoke", + drain: true, + revocationEpoch: connection.lease?.revocationEpoch, + }), + ); this.#sendNextCommand(connection); return; } @@ -867,15 +1170,11 @@ export class Phase3MockCore { return; } - connection.sendJson({ - protocol, - version: protocolVersion, - envelopeId: `ack_${this.store.state.ackedSourceSeq}`, - kind: "ack", - runnerInstanceId: this.identity.runnerInstanceId, - sentAt: new Date().toISOString(), - payload: { ackedSourceSeq: this.store.state.ackedSourceSeq }, - }); + connection.sendJson( + this.#controlEnvelope(connection, `ack_${this.store.state.ackedSourceSeq}`, "ack", { + ackedSourceSeq: this.store.state.ackedSourceSeq, + }), + ); } } @@ -1249,6 +1548,7 @@ export async function runPhase3Recovery( export const phase3Internals = { runnerBinary, runnerEnvironment, + credentialMaterial, tokenDigest, canonicalJson, phase3Identity,