From 7045290605f1ca069e6d739987f6fd2662cc0310 Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sat, 8 Aug 2026 03:53:47 -0700 Subject: [PATCH 01/21] feat(gateway): serve the native wavekv v2 sync and push routes --- dstack/Cargo.lock | 35 +++--- dstack/Cargo.toml | 3 +- dstack/gateway/src/kv/https_client.rs | 50 +++++++++ dstack/gateway/src/kv/sync_service.rs | 93 ++++++++++++++-- dstack/gateway/src/web_routes.rs | 6 +- dstack/gateway/src/web_routes/wavekv_sync.rs | 110 ++++++++++++++++++- 6 files changed, 264 insertions(+), 33 deletions(-) diff --git a/dstack/Cargo.lock b/dstack/Cargo.lock index 7087fd08a..011f20fac 100644 --- a/dstack/Cargo.lock +++ b/dstack/Cargo.lock @@ -129,7 +129,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -140,7 +140,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1692,7 +1692,7 @@ dependencies = [ "libc", "option-ext", "redox_users 0.5.2", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -2576,7 +2576,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -3487,7 +3487,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.6.4", + "socket2 0.5.10", "tokio", "tower-service", "tracing", @@ -3817,7 +3817,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4676,7 +4676,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -4933,7 +4933,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7d8fae84b431384b68627d0f9b3b1245fcf9f46f6c0e3dc902e9dce64edd1967" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.48.0", ] [[package]] @@ -5609,7 +5609,7 @@ dependencies = [ "quinn-udp", "rustc-hash", "rustls", - "socket2 0.6.4", + "socket2 0.5.10", "thiserror 2.0.18", "tokio", "tracing", @@ -5648,7 +5648,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.6.4", + "socket2 0.5.10", "tracing", "windows-sys 0.60.2", ] @@ -6343,7 +6343,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -6411,7 +6411,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -7267,7 +7267,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -7584,7 +7584,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -8420,9 +8420,8 @@ dependencies = [ [[package]] name = "wavekv" -version = "1.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cf9b73bc556dfdb7ef33617a9d477b803198db43ea3df25463efaf43d4986fe8" +version = "2.0.0" +source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#180b5fa96c984c1a1b93fd87a5a2572cb0823bc7" dependencies = [ "anyhow", "bincode 2.0.1", @@ -8522,7 +8521,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.48.0", ] [[package]] diff --git a/dstack/Cargo.toml b/dstack/Cargo.toml index a702671e1..20b903f09 100644 --- a/dstack/Cargo.toml +++ b/dstack/Cargo.toml @@ -119,7 +119,8 @@ serde-duration = { path = "serde-duration" } dstack-mr = { path = "dstack-mr" } dstack-verifier = { path = "verifier", default-features = false } size-parser = { path = "size-parser" } -wavekv = "1.0.0" +# TODO: repoint to `wavekv = "2.0"` once Phala-Network/wavekv#3 is released to crates.io. +wavekv = { git = "https://github.com/Phala-Network/wavekv", branch = "feat/delta-state-sync" } # Core dependencies anyhow = { version = "1.0.97", default-features = false } diff --git a/dstack/gateway/src/kv/https_client.rs b/dstack/gateway/src/kv/https_client.rs index d0d034a9b..cab448bd5 100644 --- a/dstack/gateway/src/kv/https_client.rs +++ b/dstack/gateway/src/kv/https_client.rs @@ -228,6 +228,56 @@ impl HttpsClient { serde_json::from_slice(&body).context("failed to parse response") } + /// Send an already-encoded body and return the raw response bytes, or `None` when + /// the peer does not expose the route. + /// + /// `None` (rather than an error) is what lets the caller distinguish "this peer has + /// not been upgraded yet" from "the request failed", which is the basis of the + /// wavekv v1/v2 protocol negotiation. + pub async fn post_bytes_probe(&self, url: &str, body: Vec) -> Result>> { + let mut encoder = GzEncoder::new(Vec::new(), Compression::fast()); + encoder + .write_all(&body) + .context("failed to compress request")?; + let compressed = encoder.finish().context("failed to finish compression")?; + + let request = hyper::Request::builder() + .method(hyper::Method::POST) + .uri(url) + .header("content-type", "application/x-msgpack-gz") + .body(Full::new(Bytes::from(compressed))) + .context("failed to build request")?; + + let response = self + .client + .request(request) + .await + .with_context(|| format!("failed to send request to {url}"))?; + + let status = response.status(); + if status == hyper::StatusCode::NOT_FOUND || status == hyper::StatusCode::METHOD_NOT_ALLOWED + { + return Ok(None); + } + if !status.is_success() { + anyhow::bail!("request failed: {status}"); + } + + let body = response + .into_body() + .collect() + .await + .context("failed to read response body")? + .to_bytes(); + + let mut decoder = GzDecoder::new(&body[..]); + let mut decompressed = Vec::new(); + decoder + .read_to_end(&mut decompressed) + .context("failed to decompress response")?; + Ok(Some(decompressed)) + } + /// Send a POST request with msgpack + gzip encoded body and receive msgpack + gzip response pub async fn post_compressed_msg( &self, diff --git a/dstack/gateway/src/kv/sync_service.rs b/dstack/gateway/src/kv/sync_service.rs index f691595a1..1e6131beb 100644 --- a/dstack/gateway/src/kv/sync_service.rs +++ b/dstack/gateway/src/kv/sync_service.rs @@ -13,7 +13,10 @@ use anyhow::{Context, Result}; use dstack_gateway_rpc::GetPeersResponse; use tracing::{info, warn}; use wavekv::{ - sync::{ExchangeInterface, SyncConfig as KvSyncConfig, SyncManager, SyncMessage, SyncResponse}, + sync::{ + ExchangeInterface, PeerLinkStatus, SyncConfig as KvSyncConfig, SyncEnvelope, SyncManager, + SyncMessage, SyncResponse, + }, types::NodeId, Node, }; @@ -70,15 +73,7 @@ impl ExchangeInterface for HttpSyncNetwork { } async fn sync_to(&self, _node: &Node, peer: NodeId, msg: SyncMessage) -> Result { - let url = self - .get_peer_url(peer) - .ok_or_else(|| anyhow::anyhow!("peer {} address not found in DB", peer))?; - - let sync_url = format!( - "{}/wavekv/sync/{}", - url.trim_end_matches('/'), - self.store_path - ); + let sync_url = self.route_for(peer, "sync")?; // Send request with msgpack + gzip encoding // app_id verification happens during TLS handshake via AppIdVerifier @@ -93,6 +88,57 @@ impl ExchangeInterface for HttpSyncNetwork { Ok(sync_response) } + + /// Native v2 exchange. + /// + /// A peer still running a v1 gateway has no `/wavekv/sync2` route and answers 404, + /// which surfaces here as `Ok(None)`; the sync manager then records the peer as + /// v1-only, falls back to `/wavekv/sync`, and re-probes periodically so an upgraded + /// peer is picked up without a restart. + async fn sync_v2_to( + &self, + _node: &Node, + peer: NodeId, + env: SyncEnvelope, + ) -> Result> { + let sync_url = self.route_for(peer, "sync2")?; + + let Some(body) = self + .client + .post_bytes_probe(&sync_url, env.encode()?) + .await + .with_context(|| format!("failed to sync to peer {peer} at {sync_url}"))? + else { + return Ok(None); + }; + + self.kv_store.update_peer_last_seen(peer); + Ok(Some(SyncEnvelope::decode(&body)?)) + } + + /// Opportunistic push. Best-effort by design: the periodic round remains the + /// anti-entropy backstop and the only ack authority. + async fn push_to(&self, _node: &Node, peer: NodeId, env: SyncEnvelope) -> Result<()> { + let push_url = self.route_for(peer, "push")?; + self.client + .post_bytes_probe(&push_url, env.encode()?) + .await + .with_context(|| format!("failed to push to peer {peer} at {push_url}"))?; + Ok(()) + } +} + +impl HttpSyncNetwork { + fn route_for(&self, peer: NodeId, verb: &str) -> Result { + let url = self + .get_peer_url(peer) + .ok_or_else(|| anyhow::anyhow!("peer {peer} address not found in DB"))?; + Ok(format!( + "{}/wavekv/{verb}/{}", + url.trim_end_matches('/'), + self.store_path + )) + } } /// WaveKV sync service that manages synchronization for both persistent and ephemeral stores @@ -116,6 +162,7 @@ impl WaveKvSyncService { let sync_config = KvSyncConfig { interval: sync_config.interval, timeout: sync_config.timeout, + ..Default::default() }; // Both networks use the same persistent node for URL lookup, but different paths @@ -173,6 +220,32 @@ impl WaveKvSyncService { pub fn handle_ephemeral_sync(&self, msg: SyncMessage) -> Result { self.ephemeral_manager.handle_sync(msg) } + + fn manager_for(&self, store: &str) -> Option<&Arc>> { + match store { + "persistent" => Some(&self.persistent_manager), + "ephemeral" => Some(&self.ephemeral_manager), + _ => None, + } + } + + /// Handle an inbound v2 sync envelope. + pub fn handle_envelope(&self, store: &str, env: SyncEnvelope) -> Option> { + Some(self.manager_for(store)?.handle_envelope(env)) + } + + /// Handle an inbound opportunistic push (merges data only; never moves acks). + pub fn handle_push(&self, store: &str, env: SyncEnvelope) -> Option> { + Some(self.manager_for(store)?.handle_push(env)) + } + + /// Per-peer protocol and digest telemetry for both stores. + pub fn link_status(&self) -> Vec<(&'static str, Vec)> { + vec![ + ("persistent", self.persistent_manager.link_status()), + ("ephemeral", self.ephemeral_manager.link_status()), + ] + } } /// Fetch peer list from bootnode and register them in KvStore. diff --git a/dstack/gateway/src/web_routes.rs b/dstack/gateway/src/web_routes.rs index 5f72735db..6ac9b70ef 100644 --- a/dstack/gateway/src/web_routes.rs +++ b/dstack/gateway/src/web_routes.rs @@ -30,5 +30,9 @@ pub fn health_routes() -> Vec { /// WaveKV sync endpoint (for main server, requires mTLS gateway auth) pub fn wavekv_sync_routes() -> Vec { - routes![wavekv_sync::sync_store] + routes![ + wavekv_sync::sync_store, + wavekv_sync::sync_store_v2, + wavekv_sync::push_store + ] } diff --git a/dstack/gateway/src/web_routes/wavekv_sync.rs b/dstack/gateway/src/web_routes/wavekv_sync.rs index 406c45698..cd2db3b61 100644 --- a/dstack/gateway/src/web_routes/wavekv_sync.rs +++ b/dstack/gateway/src/web_routes/wavekv_sync.rs @@ -20,7 +20,7 @@ use rocket::{ }; use std::io::{Read, Write}; use tracing::warn; -use wavekv::sync::{SyncMessage, SyncResponse}; +use wavekv::sync::{SyncEnvelope, SyncMessage, SyncResponse}; /// Wrapper to implement CertExt for Rocket's Certificate struct RocketCert<'a>(&'a Certificate<'a>); @@ -57,10 +57,12 @@ fn encode_sync_response(response: &SyncResponse) -> Result, Status> { warn!("failed to encode sync response: {e}"); Status::InternalServerError })?; + gzip(&encoded) +} - // Compress +fn gzip(bytes: &[u8]) -> Result, Status> { let mut encoder = GzEncoder::new(Vec::new(), Compression::fast()); - encoder.write_all(&encoded).map_err(|e| { + encoder.write_all(bytes).map_err(|e| { warn!("failed to compress sync response: {e}"); Status::InternalServerError })?; @@ -70,6 +72,32 @@ fn encode_sync_response(response: &SyncResponse) -> Result, Status> { }) } +fn gunzip(data: &[u8]) -> Result, Status> { + let mut decoder = GzDecoder::new(data); + let mut decompressed = Vec::new(); + decoder.read_to_end(&mut decompressed).map_err(|e| { + warn!("failed to decompress sync payload: {e}"); + Status::BadRequest + })?; + Ok(decompressed) +} + +/// Read a v2 envelope from a request body, applying the same size cap as the v1 route. +async fn read_envelope(data: Data<'_>) -> Result { + let bytes = data + .open(16.mebibytes()) + .into_bytes() + .await + .map_err(|_| Status::BadRequest)?; + let decompressed = gunzip(&bytes)?; + // `SyncEnvelope::decode` enforces the schema version and rejects trailing bytes; + // it is deliberately not the generic `decode` used for KV values. + SyncEnvelope::decode(&decompressed).map_err(|e| { + warn!("failed to decode sync envelope: {e:#}"); + Status::BadRequest + }) +} + /// Verify that the request is from a gateway with the same app_id (mTLS verification) fn verify_gateway_peer(state: &Proxy, cert: Option>) -> Result<(), Status> { // Skip verification if not running in dstack (test mode) @@ -158,3 +186,79 @@ pub async fn sync_store( Ok((ContentType::new("application", "x-msgpack-gz"), encoded)) } + +/// Native v2 sync endpoint. +/// +/// A gateway still running wavekv 1.x has no route here and answers 404, which is +/// exactly the signal its peers use to fall back to `/wavekv/sync`. Mounting this route +/// is therefore the whole of the server-side protocol negotiation. +#[post("/wavekv/sync2/", data = "")] +pub async fn sync_store_v2( + state: &State, + cert: Option>, + store: &str, + data: Data<'_>, +) -> Result<(ContentType, Vec), Status> { + verify_gateway_peer(state, cert)?; + + let Some(ref wavekv_sync) = state.wavekv_sync else { + return Err(Status::ServiceUnavailable); + }; + + let env = read_envelope(data).await?; + if env.sender_id == 0 { + warn!("rejected v2 sync from invalid node_id 0"); + return Err(Status::BadRequest); + } + + let Some(result) = wavekv_sync.handle_envelope(store, env) else { + return Err(Status::NotFound); + }; + let response = result.map_err(|e| { + tracing::error!("{store} v2 sync failed: {e:#}"); + Status::InternalServerError + })?; + + let encoded = response.encode().map_err(|e| { + warn!("failed to encode sync envelope: {e:#}"); + Status::InternalServerError + })?; + Ok(( + ContentType::new("application", "x-msgpack-gz"), + gzip(&encoded)?, + )) +} + +/// Opportunistic push endpoint (wavekv RFC 0001 section 3.9). +/// +/// Entries only: the receiver merges data but never moves its ack coverage from this +/// channel, so loss, duplication and reordering here are all harmless and the periodic +/// round remains the anti-entropy backstop. +#[post("/wavekv/push/", data = "")] +pub async fn push_store( + state: &State, + cert: Option>, + store: &str, + data: Data<'_>, +) -> Result { + verify_gateway_peer(state, cert)?; + + let Some(ref wavekv_sync) = state.wavekv_sync else { + return Err(Status::ServiceUnavailable); + }; + + let env = read_envelope(data).await?; + if env.sender_id == 0 { + warn!("rejected push from invalid node_id 0"); + return Err(Status::BadRequest); + } + + let Some(result) = wavekv_sync.handle_push(store, env) else { + return Err(Status::NotFound); + }; + result.map_err(|e| { + tracing::error!("{store} push failed: {e:#}"); + Status::InternalServerError + })?; + Ok(Status::Ok) +} From 68745c22f6afe83351139c33c6ca6c65fed87dc6 Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sat, 8 Aug 2026 03:53:47 -0700 Subject: [PATCH 02/21] feat(gateway): confine replicated keys to the gateway schema --- dstack/gateway/src/kv/mod.rs | 175 +++++++++++++++++++++++++++++++- dstack/gateway/src/kv/schema.rs | 143 ++++++++++++++++++++++++++ 2 files changed, 314 insertions(+), 4 deletions(-) create mode 100644 dstack/gateway/src/kv/schema.rs diff --git a/dstack/gateway/src/kv/mod.rs b/dstack/gateway/src/kv/mod.rs index 09392798c..143857e52 100644 --- a/dstack/gateway/src/kv/mod.rs +++ b/dstack/gateway/src/kv/mod.rs @@ -29,6 +29,7 @@ //! - `last_seen/node/{node_id}/{seen_by_node_id}` → u64 (timestamp) mod https_client; +mod schema; mod sync_service; pub use https_client::{AppIdValidator, HttpsClientConfig}; @@ -374,6 +375,18 @@ pub mod keys { /// in `#[serde(default)]` fields it does not receive, so the value types below /// can gain fields without breaking gateways running an older build. Decoding /// accepts both forms, so values written by older releases stay readable. +/// wavekv configuration shared by both stores. +/// +/// The admission policy is the important part: it confines a peer to the key shapes +/// this gateway actually defines, so a compromised or buggy node in the cluster cannot +/// plant arbitrary keys that every other node would then replicate and persist forever. +fn store_config(store: schema::Store) -> wavekv::NodeConfig { + wavekv::NodeConfig { + admission: Some(std::sync::Arc::new(schema::GatewaySchema::new(store))), + ..Default::default() + } +} + pub fn encode(value: &T) -> Result> { rmp_serde::encode::to_vec_named(value).context("failed to encode value") } @@ -467,9 +480,13 @@ impl KvStore { peer_ids: Vec, data_dir: impl AsRef, ) -> Result { - let persistent = - Node::new_with_persistence(my_node_id, peer_ids.clone(), data_dir.as_ref()) - .context("failed to create persistent wavekv node")?; + let persistent = Node::with_persistence_and_config( + my_node_id, + peer_ids.clone(), + data_dir.as_ref(), + store_config(schema::Store::Persistent), + ) + .context("failed to create persistent wavekv node")?; // Get peers from persistent store (may have been restored from WAL) // and include them when creating ephemeral store @@ -481,7 +498,11 @@ impl KvStore { } } - let ephemeral = Node::new(my_node_id, all_peer_ids); + let ephemeral = Node::with_config( + my_node_id, + all_peer_ids, + store_config(schema::Store::Ephemeral), + ); Ok(Self { persistent, @@ -1355,6 +1376,152 @@ mod value_encoding_tests { } } +/// The gateway speaks two wavekv protocols during a rolling upgrade: the frozen v1 +/// `SyncMessage`/`SyncResponse` pair on `/wavekv/sync`, and the v2 `SyncEnvelope` on +/// `/wavekv/sync2`. These tests pin the wire behaviour of both at the gateway layer. +#[cfg(test)] +mod sync_wire_tests { + use super::*; + use wavekv::sync::{SyncEnvelope, SyncMessage, SyncResponse}; + + fn store(dir: &std::path::Path, id: NodeId, peers: Vec) -> KvStore { + KvStore::new(id, peers, dir).expect("failed to create kv store") + } + + /// A gateway still on wavekv 1.x encodes `SyncMessage` positionally. The v1 route + /// must keep accepting that after this upgrade. + #[test] + fn a_positionally_encoded_v1_request_is_still_accepted() { + let msg = SyncMessage { + sender_id: 2, + sender_uuid: b"uuid".to_vec(), + sender_ack: [(1u32, 5u64)].into_iter().collect(), + entries: Vec::new(), + }; + let legacy = rmp_serde::encode::to_vec(&msg).expect("legacy encode"); + assert_eq!( + legacy[0] & 0xf0, + 0x90, + "fixture must be positional to exercise the legacy path" + ); + + let decoded: SyncMessage = decode(&legacy).expect("the v1 wire format must still decode"); + assert_eq!(decoded.sender_id, 2); + assert_eq!(decoded.sender_ack.get(&1), Some(&5)); + } + + /// ...and the response this gateway sends back must decode on that older peer, + /// which uses a reader built before the named-map switch. + #[test] + fn a_v1_peer_can_decode_our_sync_response() { + let response = SyncResponse { + peer_id: 1, + entries: Vec::new(), + progress: [(1u32, 7u64)].into_iter().collect(), + is_snapshot: true, + }; + let encoded = encode(&response).expect("encode"); + let decoded: SyncResponse = + rmp_serde::decode::from_slice(&encoded).expect("a v1 peer must decode this"); + assert!(decoded.is_snapshot); + assert_eq!(decoded.progress.get(&1), Some(&7)); + } + + #[test] + fn a_v2_envelope_survives_the_transport_framing() { + use flate2::{read::GzDecoder, write::GzEncoder, Compression}; + use std::io::{Read, Write}; + + let dir = tempfile::tempdir().expect("tempdir"); + let kv = store(dir.path(), 1, vec![2]); + kv.persistent() + .write() + .put(keys::peer_addr(1), b"https://a.example".to_vec()) + .expect("put"); + + let env = kv.persistent().read().prepare_sync(2, Vec::new()); + assert!(!env.entries.is_empty()); + + let mut encoder = GzEncoder::new(Vec::new(), Compression::fast()); + encoder.write_all(&env.encode().expect("encode")).unwrap(); + let wire = encoder.finish().unwrap(); + + let mut plain = Vec::new(); + GzDecoder::new(&wire[..]).read_to_end(&mut plain).unwrap(); + let decoded = SyncEnvelope::decode(&plain).expect("decode"); + + assert_eq!(decoded.sender_id, 1); + assert_eq!(decoded.entries.len(), env.entries.len()); + assert!( + decoded.digest.is_some(), + "the digest drives divergence detection" + ); + } + + /// End-to-end through the shim: a v1-shaped exchange against this gateway's store + /// converges it with the requester's view. + #[test] + fn the_v1_shim_serves_a_complete_delta() { + let dir = tempfile::tempdir().expect("tempdir"); + let kv = store(dir.path(), 1, vec![2]); + for id in 1..=3 { + kv.persistent() + .write() + .put(keys::peer_addr(id), format!("https://n{id}").into_bytes()) + .expect("put"); + } + + let request = SyncMessage { + sender_id: 2, + sender_uuid: Vec::new(), + sender_ack: Default::default(), + entries: Vec::new(), + }; + let response = kv + .persistent() + .write() + .handle_sync_v1(request) + .expect("shim response"); + + assert_eq!(response.entries.len(), 3); + assert!( + response.is_snapshot, + "the flag is what makes a v1 client adopt our coverage before merging" + ); + assert_eq!(response.progress.get(&1), Some(&3)); + } + + /// A peer cannot plant keys outside the schema, in either store. + #[test] + fn merged_entries_outside_the_schema_are_refused() { + use wavekv::types::{Entry, Metadata}; + + let dir = tempfile::tempdir().expect("tempdir"); + let kv = store(dir.path(), 1, vec![2]); + + let mut env = SyncEnvelope::new(2, Vec::new()); + env.entries.push(Entry::new( + "not-a-gateway-key".to_string(), + Some(b"x".to_vec()), + Metadata::new(2, 1, 1), + )); + env.acks.insert(2, 1); + + let outcome = kv + .persistent() + .write() + .apply_envelope(env) + .expect("apply should not fail the whole round"); + + assert_eq!(outcome.rejected, 1); + assert!( + !outcome.acks_adopted, + "a rejection must park the round's acks so the peer keeps re-offering" + ); + assert!(kv.persistent().read().get("not-a-gateway-key").is_none()); + } +} + #[cfg(test)] mod peer_url_tests { use super::validate_peer_url; diff --git a/dstack/gateway/src/kv/schema.rs b/dstack/gateway/src/kv/schema.rs new file mode 100644 index 000000000..6f219428f --- /dev/null +++ b/dstack/gateway/src/kv/schema.rs @@ -0,0 +1,143 @@ +// SPDX-FileCopyrightText: © 2024-2025 Phala Network +// +// SPDX-License-Identifier: Apache-2.0 + +//! Key-prefix admission policy for the replicated stores. +//! +//! Every gateway in a cluster shares one app_id, so mTLS proves only that a peer is +//! *some* gateway of this deployment — not that it is well-behaved. A peer that has +//! been compromised, or that is simply running buggy code, can otherwise write any key +//! it likes into the replicated namespace, and every other node will accept and persist +//! it forever (the data map is never truncated). +//! +//! wavekv 2.0 enforces admission inside `merge`, which covers both sync directions; +//! a check on the HTTP handler would only see inbound requests, not the entries that +//! arrive in a response. Rejected entries also park the round's ack adoption (rule R1), +//! so a peer sending inadmissible data keeps re-offering it rather than having it +//! silently dropped. + +use wavekv::{types::Entry, Admission, AdmissionPolicy}; + +use super::keys; + +/// Which store a policy guards. The two stores have disjoint schemas. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Store { + Persistent, + Ephemeral, +} + +/// Accepts only the key shapes this gateway actually defines. +#[derive(Debug, Clone, Copy)] +pub struct GatewaySchema { + store: Store, +} + +impl GatewaySchema { + pub fn new(store: Store) -> Self { + Self { store } + } + + fn permits(&self, key: &str) -> bool { + match self.store { + Store::Persistent => { + key.starts_with(keys::INST_PREFIX) + || key.starts_with(keys::NODE_PREFIX) + || key.starts_with(keys::CERT_PREFIX) + || key.starts_with(keys::DNS_CRED_PREFIX) + || key.starts_with(keys::PEER_ADDR_PREFIX) + || key == keys::DNS_CRED_DEFAULT + || key == keys::GLOBAL_CERTBOT_CONFIG + || key == keys::GLOBAL_ACME_CREDENTIALS + || key == keys::GLOBAL_ACME_ATTESTATION + || key == keys::GLOBAL_ACME_ROTATION_LOCK + } + Store::Ephemeral => { + key.starts_with(keys::CONN_PREFIX) + || key.starts_with(keys::HANDSHAKE_PREFIX) + || key.starts_with(keys::LAST_SEEN_NODE_PREFIX) + || key.starts_with(keys::PEER_ADDR_PREFIX) + } + } + } +} + +impl AdmissionPolicy for GatewaySchema { + fn admit(&self, entry: &Entry) -> Admission { + if self.permits(&entry.key) { + Admission::Accept + } else { + Admission::Reject { + reason: "key is outside the gateway schema for this store", + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use wavekv::types::Metadata; + + fn entry(key: &str) -> Entry { + Entry::new(key.to_string(), Some(b"v".to_vec()), Metadata::new(1, 1, 0)) + } + + fn admits(store: Store, key: &str) -> bool { + GatewaySchema::new(store).admit(&entry(key)) == Admission::Accept + } + + #[test] + fn every_key_the_gateway_writes_is_admissible() { + for key in [ + keys::inst("abc"), + keys::node_info(1), + keys::node_status(1), + keys::zt_domain_config("example.com"), + keys::cert_data("example.com"), + keys::cert_lock("example.com"), + keys::cert_attestation_latest("example.com"), + keys::cert_attestation_history("example.com", 42), + keys::dns_cred("cred"), + keys::peer_addr(1), + keys::DNS_CRED_DEFAULT.to_string(), + keys::GLOBAL_CERTBOT_CONFIG.to_string(), + keys::GLOBAL_ACME_CREDENTIALS.to_string(), + keys::GLOBAL_ACME_ATTESTATION.to_string(), + keys::GLOBAL_ACME_ROTATION_LOCK.to_string(), + ] { + assert!( + admits(Store::Persistent, &key), + "the persistent schema must admit a key the gateway itself writes: {key}" + ); + } + + for key in [ + keys::conn("inst", 1), + keys::handshake("inst", 1), + keys::last_seen_node(1, 2), + keys::peer_addr(1), + ] { + assert!( + admits(Store::Ephemeral, &key), + "the ephemeral schema must admit a key the gateway itself writes: {key}" + ); + } + } + + #[test] + fn keys_outside_the_schema_are_refused() { + for key in ["", "random", "../escape", "global/", "certificate/x"] { + assert!(!admits(Store::Persistent, key), "accepted {key}"); + assert!(!admits(Store::Ephemeral, key), "accepted {key}"); + } + } + + #[test] + fn the_two_stores_do_not_accept_each_others_keys() { + assert!(!admits(Store::Ephemeral, &keys::inst("abc"))); + assert!(!admits(Store::Ephemeral, &keys::cert_data("example.com"))); + assert!(!admits(Store::Persistent, &keys::conn("inst", 1))); + assert!(!admits(Store::Persistent, &keys::last_seen_node(1, 2))); + } +} From 5ce9f88beb85c9def0ee5f0550cfe22744aa2688 Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sat, 8 Aug 2026 03:53:47 -0700 Subject: [PATCH 03/21] feat(gateway): report state digest and peer protocol in WaveKvStatus --- dstack/gateway/rpc/proto/gateway_rpc.proto | 18 ++++++++++++- dstack/gateway/src/admin_service.rs | 31 ++++++++++++++++++++-- 2 files changed, 46 insertions(+), 3 deletions(-) diff --git a/dstack/gateway/rpc/proto/gateway_rpc.proto b/dstack/gateway/rpc/proto/gateway_rpc.proto index 68202032d..1ba11de1c 100644 --- a/dstack/gateway/rpc/proto/gateway_rpc.proto +++ b/dstack/gateway/rpc/proto/gateway_rpc.proto @@ -339,9 +339,18 @@ message PeerSyncStatus { uint32 id = 1; uint64 local_ack = 2; uint64 peer_ack = 3; - uint64 buffered_logs = 4; + // Always 0 since wavekv 2.0, which replicates state instead of operation logs and + // keeps no per-peer log buffers. Retained so existing clients keep decoding. + uint64 buffered_logs = 4 [deprecated = true]; // Last seen timestamps: [(observer_node_id, timestamp), ...] repeated LastSeenEntry last_seen = 5; + // Whether this peer has ever reported an ack map. + bool heard_from = 6; + // Sync protocol last negotiated with this peer: "v1" or "v2". + string protocol = 7; + // Consecutive quiescent rounds whose state digests disagreed. Non-zero means the + // replicas have silently diverged; wavekv 1.x could not detect this at all. + uint32 digest_mismatches = 8; } message LastSeenEntry { @@ -358,6 +367,13 @@ message StoreSyncStatus { bool dirty = 5; bool wal_enabled = 6; repeated PeerSyncStatus peers = 7; + // Hex SHA-256 over the replicated state. Two converged replicas produce equal + // digests by construction, so comparing this across the cluster is the promotion + // gate for the wavekv v2 rollout and the standing divergence check afterwards. + string digest = 8; + uint64 entries_merged = 9; + // Entries refused by the admission policy or the ingest quotas. + uint64 entries_rejected = 10; } // WaveKV sync status response diff --git a/dstack/gateway/src/admin_service.rs b/dstack/gateway/src/admin_service.rs index c1e988d19..509655046 100644 --- a/dstack/gateway/src/admin_service.rs +++ b/dstack/gateway/src/admin_service.rs @@ -208,16 +208,33 @@ impl AdminRpc for AdminRpcHandler { .collect() }; + // Per-peer protocol/digest telemetry lives on the sync manager, not the store. + let links = self + .state + .wavekv_sync + .as_ref() + .map(|s| s.link_status()) + .unwrap_or_default(); + let links_for = |name: &str| -> Vec { + links + .iter() + .find(|(store, _)| *store == name) + .map(|(_, l)| l.clone()) + .unwrap_or_default() + }; + Ok(WaveKvStatusResponse { enabled: self.state.config.sync.enabled, persistent: Some(build_store_status( "persistent", persistent_status, + &links_for("persistent"), &get_peer_last_seen, )), ephemeral: Some(build_store_status( "ephemeral", ephemeral_status, + &links_for("ephemeral"), &get_peer_last_seen, )), }) @@ -718,6 +735,7 @@ fn port_policy_view_to_proto(view: PortPolicyView) -> GetInstancePortPolicyRespo fn build_store_status( name: &str, status: WaveKvNodeStatus, + links: &[wavekv::sync::PeerLinkStatus], get_peer_last_seen: &impl Fn(u32) -> Vec<(u32, u64)>, ) -> StoreSyncStatus { StoreSyncStatus { @@ -727,6 +745,9 @@ fn build_store_status( next_seq: status.next_seq, dirty: status.dirty, wal_enabled: status.wal, + digest: status.digest, + entries_merged: status.entries_merged, + entries_rejected: status.entries_rejected, peers: status .peers .into_iter() @@ -735,12 +756,18 @@ fn build_store_status( .into_iter() .map(|(node_id, timestamp)| LastSeenEntry { node_id, timestamp }) .collect(); + let link = links.iter().find(|l| l.id == p.id); + #[allow(deprecated)] ProtoPeerSyncStatus { id: p.id, local_ack: p.ack, - peer_ack: p.pack, - buffered_logs: p.logs as u64, + peer_ack: p.peer_ack, + // wavekv 2.0 keeps no per-peer log buffers. + buffered_logs: 0, last_seen, + heard_from: p.heard_from, + protocol: link.map(|l| l.protocol).unwrap_or_default().to_string(), + digest_mismatches: link.map(|l| l.digest_mismatches).unwrap_or(0), } }) .collect(), From 20ed09cf0a9670bb51f8f49595e686f2f759554b Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sat, 8 Aug 2026 04:47:35 -0700 Subject: [PATCH 04/21] test(gateway): pin the wavekv sync route paths --- dstack/gateway/src/web_routes.rs | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/dstack/gateway/src/web_routes.rs b/dstack/gateway/src/web_routes.rs index 6ac9b70ef..55b04f7ff 100644 --- a/dstack/gateway/src/web_routes.rs +++ b/dstack/gateway/src/web_routes.rs @@ -36,3 +36,31 @@ pub fn wavekv_sync_routes() -> Vec { wavekv_sync::push_store ] } + +#[cfg(test)] +mod tests { + use super::*; + + /// The v1/v2 negotiation is driven entirely by whether a peer answers 404 on the v2 + /// route. A typo in any of these paths would therefore not fail — every peer would + /// simply 404 forever and the whole cluster would stay silently on v1. + #[test] + fn the_sync_routes_are_mounted_where_peers_look_for_them() { + let mounted: Vec = wavekv_sync_routes() + .iter() + .map(|route| route.uri.to_string()) + .collect(); + + for expected in [ + "/wavekv/sync/", + "/wavekv/sync2/", + "/wavekv/push/", + ] { + assert!( + mounted.iter().any(|uri| uri == expected), + "{expected} is not mounted; peers would 404 and never negotiate v2. \ + mounted: {mounted:?}" + ); + } + } +} From 8de509c45a464391baad889398aaefa3eea7a91c Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sat, 8 Aug 2026 08:41:18 -0700 Subject: [PATCH 05/21] feat(gateway): surface peers that fail every sync round MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pick up the wavekv fix for the opportunistic push envelope, which was built without a `sender_uuid` and so failed `check_uuid` on every push — this gateway implements `query_uuid`, so the push channel never worked here. Writes still converged over the periodic round, but each one waited a full sync interval instead of the coalesce window and the receiver logged an error per push blaming node-id reuse. That fix also widens `link_status` to report every known peer rather than only those in the link cache. A peer whose rounds all fail was previously absent from `WaveKvStatus` entirely: a 5xx deliberately does not demote a peer to "v1", so nothing about it moved. Report the new `consecutive_failures` streak so that stall is visible. Document the one direction in which the store schema is not forward compatible: values may gain fields freely, but a new *key* is rejected by nodes that predate it, and a rejection parks ack adoption for the whole round (rule R1). The pair then re-exchanges the same batch indefinitely with no error. New keys therefore ship in two releases — widen the schema everywhere first, write the key second. Also silence a `manual_repeat_n` lint in the pp tests, unrelated but newly raised by the toolchain and enough to fail `clippy -D warnings`. --- dstack/Cargo.lock | 20 ++++++++++---------- dstack/gateway/rpc/proto/gateway_rpc.proto | 4 ++++ dstack/gateway/src/admin_service.rs | 1 + dstack/gateway/src/kv/schema.rs | 19 +++++++++++++++++++ dstack/gateway/src/pp.rs | 2 +- 5 files changed, 35 insertions(+), 11 deletions(-) diff --git a/dstack/Cargo.lock b/dstack/Cargo.lock index 011f20fac..89bb687e5 100644 --- a/dstack/Cargo.lock +++ b/dstack/Cargo.lock @@ -129,7 +129,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -140,7 +140,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -2576,7 +2576,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -3817,7 +3817,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -5650,7 +5650,7 @@ dependencies = [ "once_cell", "socket2 0.5.10", "tracing", - "windows-sys 0.60.2", + "windows-sys 0.59.0", ] [[package]] @@ -6343,7 +6343,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -6411,7 +6411,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -7267,7 +7267,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -7584,7 +7584,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -8421,7 +8421,7 @@ dependencies = [ [[package]] name = "wavekv" version = "2.0.0" -source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#180b5fa96c984c1a1b93fd87a5a2572cb0823bc7" +source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#fc56d8728906cf53395bd1e096279b0f6a8c713a" dependencies = [ "anyhow", "bincode 2.0.1", diff --git a/dstack/gateway/rpc/proto/gateway_rpc.proto b/dstack/gateway/rpc/proto/gateway_rpc.proto index 1ba11de1c..ab3262df6 100644 --- a/dstack/gateway/rpc/proto/gateway_rpc.proto +++ b/dstack/gateway/rpc/proto/gateway_rpc.proto @@ -351,6 +351,10 @@ message PeerSyncStatus { // Consecutive quiescent rounds whose state digests disagreed. Non-zero means the // replicas have silently diverged; wavekv 1.x could not detect this at all. uint32 digest_mismatches = 8; + // Consecutive sync rounds that failed outright. Only a definitive 404/405 demotes a + // peer to "v1"; a 5xx or a timeout leaves `protocol` untouched by design, so this is + // the only field that moves when a peer is failing every round. + uint32 consecutive_failures = 9; } message LastSeenEntry { diff --git a/dstack/gateway/src/admin_service.rs b/dstack/gateway/src/admin_service.rs index 509655046..078113e9b 100644 --- a/dstack/gateway/src/admin_service.rs +++ b/dstack/gateway/src/admin_service.rs @@ -768,6 +768,7 @@ fn build_store_status( heard_from: p.heard_from, protocol: link.map(|l| l.protocol).unwrap_or_default().to_string(), digest_mismatches: link.map(|l| l.digest_mismatches).unwrap_or(0), + consecutive_failures: link.map(|l| l.consecutive_failures).unwrap_or(0), } }) .collect(), diff --git a/dstack/gateway/src/kv/schema.rs b/dstack/gateway/src/kv/schema.rs index 6f219428f..958c2b68d 100644 --- a/dstack/gateway/src/kv/schema.rs +++ b/dstack/gateway/src/kv/schema.rs @@ -15,6 +15,25 @@ //! arrive in a response. Rejected entries also park the round's ack adoption (rule R1), //! so a peer sending inadmissible data keeps re-offering it rather than having it //! silently dropped. +//! +//! # Adding a key: this schema must be widened one release before it is used +//! +//! Ack parking makes the schema *forward-incompatible in one direction*. Values may gain +//! fields freely — they are named-map encoded, so an older gateway skips what it does not +//! know. Adding a **key** is different: an older gateway rejects it, which sets +//! `complete = false` for the whole round, which parks ack adoption for that pair +//! entirely. The two nodes then re-exchange the same batch forever and their digests stay +//! unequal. Nothing errors; the pair simply stops making progress, and the symptom is +//! indistinguishable from an unrelated stall such as a peer with a runaway clock. +//! +//! So a new key ships in two releases, never one: +//! +//! 1. Widen the schema to **accept** the new prefix. Do not write it yet. Roll this out +//! to every node. +//! 2. Only then start **writing** it. +//! +//! The same applies in reverse when retiring a key: stop writing it, roll that out, and +//! only afterwards narrow the schema. use wavekv::{types::Entry, Admission, AdmissionPolicy}; diff --git a/dstack/gateway/src/pp.rs b/dstack/gateway/src/pp.rs index f6c6e09f2..893e3f844 100644 --- a/dstack/gateway/src/pp.rs +++ b/dstack/gateway/src/pp.rs @@ -251,7 +251,7 @@ mod tests { // PROXY prefix matched but no \r\n terminator within V1_MAX_LENGTH bytes. let bytes = vec![b'P'; V1_MAX_LENGTH + 8]; // all 'P' — never closes let mut head = b"PROXY".to_vec(); - head.extend(std::iter::repeat(b'A').take(V1_MAX_LENGTH)); + head.extend(std::iter::repeat_n(b'A', V1_MAX_LENGTH)); let err = read_proxy_header(&head[..]).await.unwrap_err(); let msg = format!("{err:#}"); assert!( From 335fab6fc7616e86b14d292a4b6602916521a59d Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sat, 8 Aug 2026 08:51:44 -0700 Subject: [PATCH 06/21] test(gateway): drive the sync routes over a local Rocket client MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The HTTP layer was the one part of the sync path with no coverage. It was skipped on the grounds that constructing a `WaveKvSyncService` needs real TLS material; that was wrong. `rcgen` is already a dependency and already used by the cert_store tests, and `verify_gateway_peer` short-circuits under `insecure_skip_attestation`, so a self-signed CA plus a leaf written to a TempDir is enough to build a serving gateway. What this pins that nothing else did: - 503, not 404, when sync is disabled. 404 is the negotiation signal, so a sync-disabled node answering 404 would be cached as "v1" by every peer for a whole reprobe window — and sync is off, so nothing would correct it. - 404 for an unknown store, which is the same signal used deliberately. - An unstamped push is refused at the route and writes nothing. This is the server-side view of the envelope-identity bug; the sender-side view lives in the wavekv push test. - A well-formed push reaches the store, a v2 round trip returns a decodable envelope, and node id 0 is refused. Also stop reporting a 404 on the push route as a delivered push. `post_bytes_probe` maps 404/405 to `Ok(None)` so the v2 probe can read it as "not upgraded yet", but `push_to` discarded the `Option`. A mistyped push URL was therefore indistinguishable from success — the same shape of silent failure that let the unstamped-envelope bug survive, since pushes are best-effort and only debug-logged. --- dstack/gateway/src/kv/sync_service.rs | 10 +- dstack/gateway/src/web_routes/wavekv_sync.rs | 250 +++++++++++++++++++ 2 files changed, 259 insertions(+), 1 deletion(-) diff --git a/dstack/gateway/src/kv/sync_service.rs b/dstack/gateway/src/kv/sync_service.rs index 1e6131beb..aa388d72a 100644 --- a/dstack/gateway/src/kv/sync_service.rs +++ b/dstack/gateway/src/kv/sync_service.rs @@ -120,10 +120,18 @@ impl ExchangeInterface for HttpSyncNetwork { /// anti-entropy backstop and the only ack authority. async fn push_to(&self, _node: &Node, peer: NodeId, env: SyncEnvelope) -> Result<()> { let push_url = self.route_for(peer, "push")?; - self.client + let delivered = self + .client .post_bytes_probe(&push_url, env.encode()?) .await .with_context(|| format!("failed to push to peer {peer} at {push_url}"))?; + // `post_bytes_probe` maps 404/405 to `Ok(None)` so the v2 probe can read it as + // "not upgraded yet". Discarding that here would report a mistyped URL, or a + // peer with no push route, as a delivered push — and pushes are best-effort and + // debug-logged, so nothing else would ever contradict it. + if delivered.is_none() { + anyhow::bail!("peer {peer} has no push route at {push_url}"); + } Ok(()) } } diff --git a/dstack/gateway/src/web_routes/wavekv_sync.rs b/dstack/gateway/src/web_routes/wavekv_sync.rs index cd2db3b61..13389ef22 100644 --- a/dstack/gateway/src/web_routes/wavekv_sync.rs +++ b/dstack/gateway/src/web_routes/wavekv_sync.rs @@ -262,3 +262,253 @@ pub async fn push_store( })?; Ok(Status::Ok) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::{load_config_figment, Config, MutualConfig, TlsConfig}; + use crate::kv::NodeData; + use crate::main_service::{Proxy, ProxyOptions}; + use rocket::local::asynchronous::Client; + use tempfile::TempDir; + use wavekv::types::{Entry, Metadata}; + + const ME: u32 = 1; + const PEER: u32 = 2; + + fn peer_uuid() -> Vec { + b"the-real-peer-2".to_vec() + } + + /// A self-signed CA plus a leaf it signs. `HttpSyncNetwork::new` loads all three + /// from disk to build its rustls client config, and the root store only accepts a + /// trust anchor with `CA:TRUE` — so a lone self-signed leaf is not enough. + fn write_tls_material(dir: &std::path::Path) -> TlsConfig { + use ra_tls::rcgen::{BasicConstraints, CertificateParams, IsCa, KeyPair}; + + let ca_key = KeyPair::generate().expect("ca key"); + let mut ca_params = CertificateParams::new(vec![]).expect("ca params"); + ca_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained); + let ca_cert = ca_params.self_signed(&ca_key).expect("ca cert"); + + let leaf_key = KeyPair::generate().expect("leaf key"); + let leaf_params = + CertificateParams::new(vec!["gateway.test".to_string()]).expect("leaf params"); + let leaf_cert = leaf_params + .signed_by(&leaf_key, &ca_cert, &ca_key) + .expect("leaf cert"); + + let cert_path = dir.join("node.crt"); + let key_path = dir.join("node.key"); + let ca_path = dir.join("ca.crt"); + std::fs::write(&cert_path, leaf_cert.pem()).expect("write cert"); + std::fs::write(&key_path, leaf_key.serialize_pem()).expect("write key"); + std::fs::write(&ca_path, ca_cert.pem()).expect("write ca"); + + TlsConfig { + certs: cert_path.to_string_lossy().into_owned(), + key: key_path.to_string_lossy().into_owned(), + mutual: MutualConfig { + ca_certs: ca_path.to_string_lossy().into_owned(), + }, + } + } + + /// A gateway serving the real sync routes over Rocket's local client. + /// + /// `insecure_skip_attestation` stands in for the mTLS peer check, which is not what + /// these tests are about; everything below it — route dispatch, the gzip framing, + /// the store split, the uuid check — is the production path. + async fn serving_gateway(sync_enabled: bool) -> (Client, Proxy, TempDir) { + // `main` installs this once at startup; the sync client builds a rustls config, + // so a test that skips it panics inside rustls rather than failing an assertion. + let _ = rustls::crypto::ring::default_provider().install_default(); + + let figment = load_config_figment(None); + let mut config = figment.focus("core").extract::().unwrap(); + let temp_dir = TempDir::new().expect("temp dir"); + + config.sync.enabled = sync_enabled; + config.sync.node_id = ME; + config.sync.bootnode = String::new(); + config.sync.data_dir = temp_dir.path().to_string_lossy().into_owned(); + config.wg.config_path = temp_dir + .path() + .join("wg.conf") + .to_string_lossy() + .into_owned(); + config.debug.insecure_skip_attestation = true; + + let tls_config = write_tls_material(temp_dir.path()); + let proxy = Proxy::new(ProxyOptions { + config, + my_app_id: None, + tls_config, + }) + .await + .expect("failed to build gateway"); + + let rocket = rocket::build() + .manage(proxy.clone()) + .mount("/", crate::web_routes::wavekv_sync_routes()); + let client = Client::tracked(rocket).await.expect("rocket client"); + (client, proxy, temp_dir) + } + + /// Register the peer so `query_uuid` returns something: the uuid check is opt-in and + /// an unknown sender bypasses it entirely. + fn register_peer(proxy: &Proxy) { + proxy + .kv_store() + .sync_node( + PEER, + &NodeData { + uuid: peer_uuid(), + url: "https://peer.test:8011".to_string(), + wg_public_key: String::new(), + wg_endpoint: String::new(), + wg_ip: String::new(), + }, + ) + .expect("register peer"); + } + + fn push_envelope(uuid: Vec, key: &str) -> SyncEnvelope { + let mut env = SyncEnvelope::new(PEER, uuid); + env.push_only = true; + env.entries.push(Entry::new( + key.to_string(), + Some(b"v".to_vec()), + Metadata::new(PEER, 1, 1), + )); + env + } + + fn body(env: &SyncEnvelope) -> Vec { + gzip(&env.encode().expect("encode envelope")).expect("gzip") + } + + #[tokio::test] + async fn a_stamped_push_is_accepted_and_lands_in_the_store() { + let (client, proxy, _tmp) = serving_gateway(true).await; + register_peer(&proxy); + + let response = client + .post("/wavekv/push/persistent") + .body(body(&push_envelope(peer_uuid(), "node/9"))) + .dispatch() + .await; + + assert_eq!(response.status(), Status::Ok); + assert!( + proxy.kv_store().persistent().read().get("node/9").is_some(), + "a well-formed push must reach the store" + ); + } + + /// The route-level view of the bug that made every opportunistic push fail: the + /// sender built its envelope without stamping `sender_uuid`, and the receiver's + /// `check_uuid` — which only the manager runs, not `merge_push` — rejected it. + #[tokio::test] + async fn an_unstamped_push_is_refused_at_the_route() { + let (client, proxy, _tmp) = serving_gateway(true).await; + register_peer(&proxy); + + let response = client + .post("/wavekv/push/persistent") + .body(body(&push_envelope(Vec::new(), "node/9"))) + .dispatch() + .await; + + assert_eq!(response.status(), Status::InternalServerError); + assert!( + proxy.kv_store().persistent().read().get("node/9").is_none(), + "a push that fails the identity check must not write anything" + ); + } + + #[tokio::test] + async fn a_v2_round_trip_returns_a_decodable_envelope() { + let (client, proxy, _tmp) = serving_gateway(true).await; + register_peer(&proxy); + proxy + .kv_store() + .persistent() + .write() + .put("node/7".to_string(), b"v".to_vec()) + .expect("seed"); + + let request = SyncEnvelope::new(PEER, peer_uuid()); + let response = client + .post("/wavekv/sync2/persistent") + .body(body(&request)) + .dispatch() + .await; + + assert_eq!(response.status(), Status::Ok); + let bytes = response.into_bytes().await.expect("body"); + let decoded = SyncEnvelope::decode(&gunzip(&bytes).expect("gunzip")).expect("decode"); + assert_eq!(decoded.sender_id, ME); + assert!( + decoded.entries.iter().any(|e| e.key == "node/7"), + "an empty ack map must draw the whole live state" + ); + } + + /// 404 is the negotiation signal: it is what tells a peer "this node has no v2 + /// route, fall back to v1". Nothing else on these routes may produce it by accident. + #[tokio::test] + async fn an_unknown_store_is_a_404_because_that_is_the_v1_signal() { + let (client, proxy, _tmp) = serving_gateway(true).await; + register_peer(&proxy); + + let response = client + .post("/wavekv/sync2/bogus") + .body(body(&SyncEnvelope::new(PEER, peer_uuid()))) + .dispatch() + .await; + + assert_eq!(response.status(), Status::NotFound); + } + + /// ...which is why a node with sync switched off must answer 503 and not 404. A 404 + /// here would demote this node to v1 in every peer's cache for a whole reprobe + /// window — silently, and without sync being on to fix it. + #[tokio::test] + async fn a_sync_disabled_node_answers_503_rather_than_404() { + let (client, _proxy, _tmp) = serving_gateway(false).await; + + for path in [ + "/wavekv/sync/persistent", + "/wavekv/sync2/persistent", + "/wavekv/push/persistent", + ] { + let response = client + .post(path) + .body(body(&SyncEnvelope::new(PEER, peer_uuid()))) + .dispatch() + .await; + assert_eq!( + response.status(), + Status::ServiceUnavailable, + "{path} must not look like a missing v2 route" + ); + } + } + + #[tokio::test] + async fn a_push_from_node_id_zero_is_refused() { + let (client, proxy, _tmp) = serving_gateway(true).await; + register_peer(&proxy); + + let mut env = push_envelope(peer_uuid(), "node/9"); + env.sender_id = 0; + let response = client + .post("/wavekv/push/persistent") + .body(body(&env)) + .dispatch() + .await; + + assert_eq!(response.status(), Status::BadRequest); + } +} From 29c6101e174160d54179946affc85267a1126b4f Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sat, 8 Aug 2026 09:08:07 -0700 Subject: [PATCH 07/21] chore(gateway): pick up wavekv responder identity check Takes the wavekv fix that verifies the responder's uuid on a v2 response. The field was already on the wire and populated by the responder; only the initiator never read it, so node-id-reuse detection ran in one direction. --- dstack/Cargo.lock | 35 +++++++++++++---------------------- 1 file changed, 13 insertions(+), 22 deletions(-) diff --git a/dstack/Cargo.lock b/dstack/Cargo.lock index 89bb687e5..fe26ebda0 100644 --- a/dstack/Cargo.lock +++ b/dstack/Cargo.lock @@ -129,7 +129,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -140,7 +140,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1692,7 +1692,7 @@ dependencies = [ "libc", "option-ext", "redox_users 0.5.2", - "windows-sys 0.59.0", + "windows-sys 0.60.2", ] [[package]] @@ -2576,7 +2576,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -3817,7 +3817,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -4676,7 +4676,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.60.2", ] [[package]] @@ -5650,7 +5650,7 @@ dependencies = [ "once_cell", "socket2 0.5.10", "tracing", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -6330,7 +6330,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.4.15", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -6343,7 +6343,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -6411,7 +6411,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -7267,7 +7267,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -7584,7 +7584,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -8421,7 +8421,7 @@ dependencies = [ [[package]] name = "wavekv" version = "2.0.0" -source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#fc56d8728906cf53395bd1e096279b0f6a8c713a" +source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#29b2cc56d2a383b30c91a1b4d08a07e4fbc3051e" dependencies = [ "anyhow", "bincode 2.0.1", @@ -8694,15 +8694,6 @@ dependencies = [ "windows-targets 0.52.6", ] -[[package]] -name = "windows-sys" -version = "0.59.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" -dependencies = [ - "windows-targets 0.52.6", -] - [[package]] name = "windows-sys" version = "0.60.2" From b05a14641c1bc234929606c90d1a01a111ab719b Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sat, 8 Aug 2026 09:19:52 -0700 Subject: [PATCH 08/21] chore(gateway): pick up wavekv uuid-check revert MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The responder-side identity check shipped in the previous bump wedged any peer that regenerated its uuid — an ordinary CVM rebuild, since the uuid is derived from the data directory while the node id comes from config. --- dstack/Cargo.lock | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/dstack/Cargo.lock b/dstack/Cargo.lock index fe26ebda0..3429ab50a 100644 --- a/dstack/Cargo.lock +++ b/dstack/Cargo.lock @@ -129,7 +129,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -140,7 +140,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -1692,7 +1692,7 @@ dependencies = [ "libc", "option-ext", "redox_users 0.5.2", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -2576,7 +2576,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -3817,7 +3817,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -4676,7 +4676,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -5650,7 +5650,7 @@ dependencies = [ "once_cell", "socket2 0.5.10", "tracing", - "windows-sys 0.52.0", + "windows-sys 0.60.2", ] [[package]] @@ -6343,7 +6343,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -6411,7 +6411,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -7267,7 +7267,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -7584,7 +7584,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -8421,7 +8421,7 @@ dependencies = [ [[package]] name = "wavekv" version = "2.0.0" -source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#29b2cc56d2a383b30c91a1b4d08a07e4fbc3051e" +source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#9496ad1cd2505c936e124c896fa9634567d5e711" dependencies = [ "anyhow", "bincode 2.0.1", From 769f66b5129f02b6eef35ae8e9eddea006adf3ba Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sat, 8 Aug 2026 19:32:17 -0700 Subject: [PATCH 09/21] fix(gateway): bound decompression on the sync routes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sync wire is gzipped and the 16 MiB cap on a request body caps the *compressed* size, which bounds nothing on its own — gzip expands by three orders of magnitude on attacker-chosen input, so that cap admits a payload that expands into the gigabytes. Every gateway in a cluster shares one app_id, so mTLS proves only that the sender is some gateway of this deployment; it is the same trust level the key schema already treats as insufficient. All four decompression points are now bounded through one helper: both server routes and both client response paths. The client also read peer responses with `Body::collect`, which has no limit at all, so the memory was already spent before any decoding bound could apply; response bodies now go through `Limited` with the same 16 MiB the routes accept on a request. The decompressed ceiling is 128 MiB, far above any legitimate payload: a v2 delta is capped by `max_delta_bytes` at 4 MiB, and the v1 shim answers with the whole live state, which is bounded by the gateway's own key set rather than by anything a peer controls. Tested at the limit as well as past it — a fixture landing exactly on the ceiling must still decode, or the bound could tighten by a byte with only the bomb test still passing. --- dstack/Cargo.lock | 22 +++---- dstack/gateway/src/kv/https_client.rs | 59 +++++++++--------- dstack/gateway/src/kv/mod.rs | 38 ++++++++++++ dstack/gateway/src/web_routes/wavekv_sync.rs | 64 +++++++++++++++----- 4 files changed, 128 insertions(+), 55 deletions(-) diff --git a/dstack/Cargo.lock b/dstack/Cargo.lock index 3429ab50a..d9f349824 100644 --- a/dstack/Cargo.lock +++ b/dstack/Cargo.lock @@ -129,7 +129,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -140,7 +140,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1692,7 +1692,7 @@ dependencies = [ "libc", "option-ext", "redox_users 0.5.2", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -2576,7 +2576,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -3817,7 +3817,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4676,7 +4676,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -6343,7 +6343,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -6411,7 +6411,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -7267,7 +7267,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -7584,7 +7584,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -8421,7 +8421,7 @@ dependencies = [ [[package]] name = "wavekv" version = "2.0.0" -source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#9496ad1cd2505c936e124c896fa9634567d5e711" +source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#bebc431d4d2624056df22465d5affb65ca0d2d6d" dependencies = [ "anyhow", "bincode 2.0.1", diff --git a/dstack/gateway/src/kv/https_client.rs b/dstack/gateway/src/kv/https_client.rs index cab448bd5..43fb3a734 100644 --- a/dstack/gateway/src/kv/https_client.rs +++ b/dstack/gateway/src/kv/https_client.rs @@ -5,12 +5,12 @@ //! HTTPS client with mTLS and custom certificate verification during TLS handshake. use std::fmt::Debug; -use std::io::{Read, Write}; +use std::io::Write; use std::sync::Arc; use anyhow::{Context, Result}; -use flate2::{read::GzDecoder, write::GzEncoder, Compression}; -use http_body_util::{BodyExt, Full}; +use flate2::{write::GzEncoder, Compression}; +use http_body_util::{BodyExt, Full, Limited}; use hyper::body::Bytes; use hyper_rustls::HttpsConnectorBuilder; use hyper_util::{ @@ -25,6 +25,24 @@ use serde::{de::DeserializeOwned, Serialize}; use super::{decode, encode}; +/// Read a peer's response body, refusing one larger than the routes accept on a request. +/// +/// `Body::collect` reads to completion, so without this a peer could stream an unbounded +/// response and the decompression limit downstream would never be reached — the memory +/// is already gone by then. +async fn read_body_bounded(body: hyper::body::Incoming) -> Result { + Limited::new(body, super::MAX_COMPRESSED_SYNC_BYTES) + .collect() + .await + .map(|collected| collected.to_bytes()) + .map_err(|err| { + anyhow::anyhow!( + "failed to read response body (limit {} bytes): {err}", + super::MAX_COMPRESSED_SYNC_BYTES + ) + }) +} + /// Custom certificate validator trait for TLS handshake verification. /// /// Implementations can perform additional validation on the peer certificate @@ -263,19 +281,11 @@ impl HttpsClient { anyhow::bail!("request failed: {status}"); } - let body = response - .into_body() - .collect() - .await - .context("failed to read response body")? - .to_bytes(); - - let mut decoder = GzDecoder::new(&body[..]); - let mut decompressed = Vec::new(); - decoder - .read_to_end(&mut decompressed) - .context("failed to decompress response")?; - Ok(Some(decompressed)) + let body = read_body_bounded(response.into_body()).await?; + Ok(Some(crate::kv::gunzip_bounded( + &body, + crate::kv::MAX_DECOMPRESSED_SYNC_BYTES, + )?)) } /// Send a POST request with msgpack + gzip encoded body and receive msgpack + gzip response @@ -310,20 +320,9 @@ impl HttpsClient { anyhow::bail!("request failed: {}", response.status()); } - let body = response - .into_body() - .collect() - .await - .context("failed to read response body")? - .to_bytes(); - - // Decompress - let mut decoder = GzDecoder::new(body.as_ref()); - let mut decompressed = Vec::new(); - decoder - .read_to_end(&mut decompressed) - .context("failed to decompress response")?; - + let body = read_body_bounded(response.into_body()).await?; + let decompressed = + crate::kv::gunzip_bounded(&body, crate::kv::MAX_DECOMPRESSED_SYNC_BYTES)?; decode(&decompressed).context("failed to decode response") } } diff --git a/dstack/gateway/src/kv/mod.rs b/dstack/gateway/src/kv/mod.rs index 143857e52..8d7f45865 100644 --- a/dstack/gateway/src/kv/mod.rs +++ b/dstack/gateway/src/kv/mod.rs @@ -387,6 +387,44 @@ fn store_config(store: schema::Store) -> wavekv::NodeConfig { } } +/// Ceiling on a decompressed sync payload. +/// +/// The wire is gzipped, and gzip expands by three orders of magnitude on +/// attacker-chosen input: the 16 MiB cap on a request body is a cap on the *compressed* +/// size, which bounds nothing useful on its own. Every gateway in the cluster shares one +/// app_id, so mTLS proves only that a peer is *some* gateway of this deployment — the +/// same reason the key schema exists (see `schema.rs`). +/// +/// The value is far above any legitimate payload. A v2 delta is capped by +/// `max_delta_bytes` (4 MiB by default) and the v1 shim answers with the whole live +/// state, which is bounded by the gateway's own key set — instances, certificates and +/// node records — not by anything a peer controls. +pub const MAX_DECOMPRESSED_SYNC_BYTES: usize = 128 * 1024 * 1024; + +/// Ceiling on a compressed sync response, mirroring the 16 MiB the routes accept on a +/// request. Without it a peer's response body is read to completion before any decoding +/// bound applies. +pub const MAX_COMPRESSED_SYNC_BYTES: usize = 16 * 1024 * 1024; + +/// Decompress gzip, refusing anything that expands past `limit`. +/// +/// Reads one byte past the limit so a payload landing exactly on it is still accepted +/// and a larger one is rejected rather than silently truncated — `Read::take` alone +/// would hand back a short buffer that then fails to decode, reporting the wrong fault. +pub fn gunzip_bounded(data: &[u8], limit: usize) -> Result> { + use std::io::Read; + + let mut out = Vec::new(); + flate2::read::GzDecoder::new(data) + .take(limit as u64 + 1) + .read_to_end(&mut out) + .context("failed to decompress payload")?; + if out.len() > limit { + anyhow::bail!("decompressed payload exceeds {limit} bytes"); + } + Ok(out) +} + pub fn encode(value: &T) -> Result> { rmp_serde::encode::to_vec_named(value).context("failed to encode value") } diff --git a/dstack/gateway/src/web_routes/wavekv_sync.rs b/dstack/gateway/src/web_routes/wavekv_sync.rs index 13389ef22..3d9e0cdf5 100644 --- a/dstack/gateway/src/web_routes/wavekv_sync.rs +++ b/dstack/gateway/src/web_routes/wavekv_sync.rs @@ -7,10 +7,10 @@ //! Sync data is encoded using msgpack + gzip compression for efficiency. use crate::{ - kv::{decode, encode}, + kv::{decode, encode, gunzip_bounded, MAX_DECOMPRESSED_SYNC_BYTES}, main_service::Proxy, }; -use flate2::{read::GzDecoder, write::GzEncoder, Compression}; +use flate2::{write::GzEncoder, Compression}; use ra_tls::traits::CertExt; use rocket::{ data::{Data, ToByteUnit}, @@ -18,7 +18,7 @@ use rocket::{ mtls::{oid::Oid, Certificate}, post, State, }; -use std::io::{Read, Write}; +use std::io::Write; use tracing::warn; use wavekv::sync::{SyncEnvelope, SyncMessage, SyncResponse}; @@ -37,11 +37,8 @@ impl CertExt for RocketCert<'_> { /// Decode compressed msgpack data fn decode_sync_message(data: &[u8]) -> Result { - // Decompress - let mut decoder = GzDecoder::new(data); - let mut decompressed = Vec::new(); - decoder.read_to_end(&mut decompressed).map_err(|e| { - warn!("failed to decompress sync message: {e}"); + let decompressed = gunzip_bounded(data, MAX_DECOMPRESSED_SYNC_BYTES).map_err(|e| { + warn!("failed to decompress sync message: {e:#}"); Status::BadRequest })?; @@ -73,13 +70,10 @@ fn gzip(bytes: &[u8]) -> Result, Status> { } fn gunzip(data: &[u8]) -> Result, Status> { - let mut decoder = GzDecoder::new(data); - let mut decompressed = Vec::new(); - decoder.read_to_end(&mut decompressed).map_err(|e| { - warn!("failed to decompress sync payload: {e}"); + gunzip_bounded(data, MAX_DECOMPRESSED_SYNC_BYTES).map_err(|e| { + warn!("failed to decompress sync payload: {e:#}"); Status::BadRequest - })?; - Ok(decompressed) + }) } /// Read a v2 envelope from a request body, applying the same size cap as the v1 route. @@ -496,6 +490,48 @@ mod tests { } } + /// gzip expands by three orders of magnitude on attacker-chosen input, so the + /// 16 MiB cap on the request body bounds the *compressed* size and nothing else. + /// mTLS proves only that the sender is some gateway of this deployment, which is + /// the same trust level the key schema already assumes is insufficient. + #[tokio::test] + async fn a_compression_bomb_is_refused_before_it_is_decompressed() { + let (client, _proxy, _tmp) = serving_gateway(true).await; + + // ~130 MiB of zeroes compresses to well under the request cap. + let bomb = gzip(&vec![0u8; MAX_DECOMPRESSED_SYNC_BYTES + 1]).expect("gzip"); + assert!( + bomb.len() < 16 * 1024 * 1024, + "the fixture has to fit through the body cap to be testing anything: {} bytes", + bomb.len() + ); + + for path in [ + "/wavekv/sync/persistent", + "/wavekv/sync2/persistent", + "/wavekv/push/persistent", + ] { + let response = client.post(path).body(bomb.clone()).dispatch().await; + assert_eq!( + response.status(), + Status::BadRequest, + "{path} must refuse an over-sized expansion" + ); + } + } + + /// The limit is inclusive, so a payload landing exactly on it still decodes. Without + /// this the bound could tighten by a byte and only the bomb test would still pass. + #[test] + fn a_payload_exactly_on_the_limit_still_decompresses() { + let exact = gzip(&vec![7u8; MAX_DECOMPRESSED_SYNC_BYTES]).expect("gzip"); + let out = gunzip_bounded(&exact, MAX_DECOMPRESSED_SYNC_BYTES).expect("must be accepted"); + assert_eq!(out.len(), MAX_DECOMPRESSED_SYNC_BYTES); + + let one_over = gzip(&vec![7u8; MAX_DECOMPRESSED_SYNC_BYTES + 1]).expect("gzip"); + assert!(gunzip_bounded(&one_over, MAX_DECOMPRESSED_SYNC_BYTES).is_err()); + } + #[tokio::test] async fn a_push_from_node_id_zero_is_refused() { let (client, proxy, _tmp) = serving_gateway(true).await; From e128d72767042542f7f9b5674fe5a72a784ffcb1 Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sat, 8 Aug 2026 23:21:43 -0700 Subject: [PATCH 10/21] fix(gateway): allocate this node's own records after the sync bootstrap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A local write allocates a sequence number. After a data-directory loss the node keeps its id but has no record of which numbers it already spent — only its peers do — so `bootstrap` rebuilds the counter from their coverage. Anything written before that reuses numbers the peers already treat as seen, and peers filter those writes out of every delta with no error on either side. The three records written at startup were exactly the ones that must not be dropped: `node/info` carries the fresh uuid peers check us against, and `__peer_addr` carries the address they route to. A rebuilt gateway therefore wedged in both directions and stayed wedged until per-peer digest repair happened to fire. They could not simply be moved, because `HttpSyncNetwork::new` read this node's uuid back out of the store, making the `node/info` write a prerequisite of building the sync service at all. That read is the actual defect: our own uuid is local configuration, not replicated state, and routing it through the store created the ordering constraint that forced the bug. It is now passed in, and all three writes happen after the bootstrap. Also bound the response body in `post_json`. Every other response is read through `read_body_bounded`; this one collected without a limit, so a peer could stream until memory ran out. It is the bootnode GetPeers path, and the threat model does not assume a bootnode is honest. --- dstack/gateway/src/kv/https_client.rs | 9 ++---- dstack/gateway/src/kv/sync_service.rs | 19 +++++++++---- dstack/gateway/src/main_service.rs | 40 ++++++++++++++++++--------- 3 files changed, 44 insertions(+), 24 deletions(-) diff --git a/dstack/gateway/src/kv/https_client.rs b/dstack/gateway/src/kv/https_client.rs index 43fb3a734..731b73d2d 100644 --- a/dstack/gateway/src/kv/https_client.rs +++ b/dstack/gateway/src/kv/https_client.rs @@ -236,12 +236,9 @@ impl HttpsClient { anyhow::bail!("request failed: {}", response.status()); } - let body = response - .into_body() - .collect() - .await - .context("failed to read response body")? - .to_bytes(); + // Bounded like every other response: this is the bootnode GetPeers path, and + // the threat model does not assume a bootnode is honest. + let body = read_body_bounded(response.into_body()).await?; serde_json::from_slice(&body).context("failed to parse response") } diff --git a/dstack/gateway/src/kv/sync_service.rs b/dstack/gateway/src/kv/sync_service.rs index aa388d72a..7d3a48ae4 100644 --- a/dstack/gateway/src/kv/sync_service.rs +++ b/dstack/gateway/src/kv/sync_service.rs @@ -40,15 +40,21 @@ pub struct HttpSyncNetwork { } impl HttpSyncNetwork { + /// `my_uuid` is passed in rather than read back out of the store. + /// + /// Our own uuid is local configuration, not replicated state, and sourcing it from + /// the store forced this node's `node/info` record to be written before the service + /// could be built — which is to say before `bootstrap` had rebuilt the sequence + /// counter. After a data-directory loss that made the record spend a sequence number + /// the peers already consider seen, so the one record they check us against was the + /// one guaranteed to be dropped. pub fn new( kv_store: KvStore, store_path: &'static str, tls_config: &HttpsClientConfig, + my_uuid: Vec, ) -> Result { let client = HttpsClient::new(tls_config)?; - let my_uuid = kv_store - .get_peer_uuid(kv_store.my_node_id) - .context("failed to get my UUID")?; Ok(Self { client, kv_store, @@ -166,6 +172,7 @@ impl WaveKvSyncService { kv_store: &KvStore, sync_config: &GwSyncConfig, tls_config: HttpsClientConfig, + my_uuid: Vec, ) -> Result { let sync_config = KvSyncConfig { interval: sync_config.interval, @@ -174,8 +181,10 @@ impl WaveKvSyncService { }; // Both networks use the same persistent node for URL lookup, but different paths - let persistent_network = HttpSyncNetwork::new(kv_store.clone(), "persistent", &tls_config)?; - let ephemeral_network = HttpSyncNetwork::new(kv_store.clone(), "ephemeral", &tls_config)?; + let persistent_network = + HttpSyncNetwork::new(kv_store.clone(), "persistent", &tls_config, my_uuid.clone())?; + let ephemeral_network = + HttpSyncNetwork::new(kv_store.clone(), "ephemeral", &tls_config, my_uuid)?; let persistent_manager = Arc::new(SyncManager::with_config( kv_store.persistent().clone(), diff --git a/dstack/gateway/src/main_service.rs b/dstack/gateway/src/main_service.rs index 6167eaf1c..7580698af 100644 --- a/dstack/gateway/src/main_service.rs +++ b/dstack/gateway/src/main_service.rs @@ -182,7 +182,13 @@ impl ProxyInner { ); let state = build_state_from_kv_store(instances); - // Sync this node to KvStore + // This node's own records are written *after* the bootstrap below, not here. + // A local write allocates a sequence number, and after a data-directory loss + // this node has no record of which numbers it already spent — only its peers + // do. `bootstrap` rebuilds the counter from their coverage, so anything written + // before it reuses numbers the peers already treat as seen and is silently + // dropped cluster-wide. That would strand exactly the records recovery depends + // on: the fresh uuid peers check us against, and our sync address. let node_data = NodeData { uuid: config.uuid(), url: config.sync.my_url.clone(), @@ -190,17 +196,6 @@ impl ProxyInner { wg_endpoint: config.wg.endpoint.clone(), wg_ip: config.wg.ip.to_string(), }; - if let Err(err) = kv_store.sync_node(config.sync.node_id, &node_data) { - error!("Failed to sync this node to KvStore: {err:?}"); - } - // Set this node's status to Online - if let Err(err) = kv_store.set_node_status(config.sync.node_id, NodeStatus::Up) { - error!("Failed to set node status: {err:?}"); - } - // Register this node's sync URL in DB (for peer discovery) - if let Err(err) = kv_store.register_peer_url(config.sync.node_id, &config.sync.my_url) { - error!("Failed to register peer URL: {err:?}"); - } // Build HttpsClientConfig for mTLS communication let https_config = { @@ -232,7 +227,12 @@ impl ProxyInner { // Create WaveKV sync service (only if sync is enabled) let wavekv_sync = if config.sync.enabled { - match WaveKvSyncService::new(&kv_store, &config.sync, https_config.clone()) { + match WaveKvSyncService::new( + &kv_store, + &config.sync, + https_config.clone(), + node_data.uuid.clone(), + ) { Ok(sync_service) => Some(Arc::new(sync_service)), Err(err) => { error!("Failed to create WaveKV sync service: {err:?}"); @@ -267,6 +267,20 @@ impl ProxyInner { } } + // Publish this node's own records now that the sequence counter reflects + // whatever the peers already know we have spent (see the note above). + if let Err(err) = kv_store.sync_node(config.sync.node_id, &node_data) { + error!("Failed to sync this node to KvStore: {err:?}"); + } + // Set this node's status to Online + if let Err(err) = kv_store.set_node_status(config.sync.node_id, NodeStatus::Up) { + error!("Failed to set node status: {err:?}"); + } + // Register this node's sync URL in DB (for peer discovery) + if let Err(err) = kv_store.register_peer_url(config.sync.node_id, &config.sync.my_url) { + error!("Failed to register peer URL: {err:?}"); + } + // Create CertResolver and load certificates from KvStore let cert_resolver = Arc::new(CertResolver::new()); let all_cert_data = kv_store.load_all_cert_data(); From 54d789c01f7f113bcd0e192459e511c81fe0aba6 Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sat, 8 Aug 2026 23:29:18 -0700 Subject: [PATCH 11/21] chore(gateway): pin wavekv at the reviewed fixes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Picks up: WAL truncation of a damaged tail before appending (writes after a torn-tail recovery were silently lost on the next restart), the reset_acks hint the divergence repair always needed (repair reached only entries the peer itself authored), sequence-number recovery that survives an own entry losing LWW, cross-page R1 enforcement, and requests no longer disclosing our state digest. The wire test framed a *request* to check the digest survives transport. Requests no longer carry one, so it frames a response — the direction the digest actually travels — and asserts the request has none. --- dstack/Cargo.lock | 22 +++++++++++----------- dstack/gateway/src/kv/mod.rs | 15 ++++++++++++++- 2 files changed, 25 insertions(+), 12 deletions(-) diff --git a/dstack/Cargo.lock b/dstack/Cargo.lock index d9f349824..042b65eb7 100644 --- a/dstack/Cargo.lock +++ b/dstack/Cargo.lock @@ -129,7 +129,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -140,7 +140,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -1692,7 +1692,7 @@ dependencies = [ "libc", "option-ext", "redox_users 0.5.2", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -2576,7 +2576,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -3817,7 +3817,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -4676,7 +4676,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -6343,7 +6343,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -6411,7 +6411,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -7267,7 +7267,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -7584,7 +7584,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -8421,7 +8421,7 @@ dependencies = [ [[package]] name = "wavekv" version = "2.0.0" -source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#bebc431d4d2624056df22465d5affb65ca0d2d6d" +source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#179dacb81ce4c4546e19dd7fe90f3d9bf5de4716" dependencies = [ "anyhow", "bincode 2.0.1", diff --git a/dstack/gateway/src/kv/mod.rs b/dstack/gateway/src/kv/mod.rs index 8d7f45865..cc3d3c0f9 100644 --- a/dstack/gateway/src/kv/mod.rs +++ b/dstack/gateway/src/kv/mod.rs @@ -1477,7 +1477,20 @@ mod sync_wire_tests { .put(keys::peer_addr(1), b"https://a.example".to_vec()) .expect("put"); - let env = kv.persistent().read().prepare_sync(2, Vec::new()); + // Requests deliberately carry no digest: sending it would let any responder + // echo it back and forge agreement forever. So frame a *response*, which is + // the direction the digest actually travels. + assert!(kv + .persistent() + .read() + .prepare_sync(2, Vec::new()) + .digest + .is_none()); + let env = kv + .persistent() + .write() + .handle_envelope(SyncEnvelope::new(2, Vec::new()), Vec::new()) + .expect("respond"); assert!(!env.entries.is_empty()); let mut encoder = GzEncoder::new(Vec::new(), Compression::fast()); From b6d63b1d026c67215102b8ffeb8cc2b8dd716ecf Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sat, 8 Aug 2026 23:34:55 -0700 Subject: [PATCH 12/21] chore(gateway): pin wavekv at the tombstone-GC and membership-durability fixes Retiring a peer no longer discards our coverage of the entries it authored, which is what let the rest of the cluster compute a GC watermark for that origin; membership ops are now WAL-durable, so a peer lost to a crash can no longer widen the watermark and resurrect that peer's deletes. --- dstack/Cargo.lock | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/dstack/Cargo.lock b/dstack/Cargo.lock index 042b65eb7..e87ddfdb3 100644 --- a/dstack/Cargo.lock +++ b/dstack/Cargo.lock @@ -129,7 +129,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -140,7 +140,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1692,7 +1692,7 @@ dependencies = [ "libc", "option-ext", "redox_users 0.5.2", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -2576,7 +2576,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -3817,7 +3817,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4676,7 +4676,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -6343,7 +6343,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -6411,7 +6411,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -7267,7 +7267,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -7584,7 +7584,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -8421,7 +8421,7 @@ dependencies = [ [[package]] name = "wavekv" version = "2.0.0" -source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#179dacb81ce4c4546e19dd7fe90f3d9bf5de4716" +source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#82c9328b9c161e78e95eced398ee6256581a0312" dependencies = [ "anyhow", "bincode 2.0.1", From 8d04ff229f477c0e251315cc5ae4fa084be09b78 Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sun, 9 Aug 2026 20:16:17 -0700 Subject: [PATCH 13/21] test(gateway): cover the sync routes' authentication gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mutation testing found `verify_gateway_peer` replaceable with `Ok(())` without turning the suite red. The sync routes are the cluster's write surface — anything reaching them inserts entries that replicate to every gateway — and that function is the only thing in front of them. The cause was in the fixture: every route test sets `insecure_skip_attestation`, which is the function's first statement, so no test had ever executed a line of the gate. The comment claimed the flag "stands in for the mTLS peer check". It does not stand in for it; it removes it. Two gaps, so two changes. `enforcing_gateway` runs with the bypass off. Rocket's local client speaks no TLS and so presents no certificate, which is exactly the case that must be refused, and all three routes are asserted to answer 401. The app-id comparison needed a certificate, and `rocket::mtls::Certificate` has no public constructor — it exists only as the output of a real handshake. But the adapter over it only ever used `cert.extensions()`, which is public and whose element type comes straight out of `X509Certificate`. `RocketCert` now holds the extension list, so a test can build one from a certificate minted in process, and the authorization rule is split out from the Rocket plumbing it was tangled with. None of this needs a TEE or a simulator: the check reads two X.509 extensions and compares bytes. `CertRequest` adds `PHALA_RATLS_APP_ID` unconditionally, and the gateway's own app id is already a constructor parameter. Four cases now pinned: matching id accepted, foreign id forbidden, certificate without an app id refused, and a gateway with no app id of its own authorizing nobody. Each was verified to die under the mutation it targets. --- dstack/gateway/src/web_routes/wavekv_sync.rs | 184 +++++++++++++++++-- 1 file changed, 168 insertions(+), 16 deletions(-) diff --git a/dstack/gateway/src/web_routes/wavekv_sync.rs b/dstack/gateway/src/web_routes/wavekv_sync.rs index 3d9e0cdf5..2ebf37abb 100644 --- a/dstack/gateway/src/web_routes/wavekv_sync.rs +++ b/dstack/gateway/src/web_routes/wavekv_sync.rs @@ -15,20 +15,24 @@ use ra_tls::traits::CertExt; use rocket::{ data::{Data, ToByteUnit}, http::{ContentType, Status}, - mtls::{oid::Oid, Certificate}, + mtls::{oid::Oid, x509::X509Extension, Certificate}, post, State, }; use std::io::Write; use tracing::warn; use wavekv::sync::{SyncEnvelope, SyncMessage, SyncResponse}; -/// Wrapper to implement CertExt for Rocket's Certificate -struct RocketCert<'a>(&'a Certificate<'a>); +/// Adapter implementing `CertExt` over a parsed certificate's extension list. +/// +/// It holds the extensions rather than the `Certificate` so that a test can build one: +/// `rocket::mtls::Certificate` has no public constructor — it can only be produced by a +/// real mTLS handshake — while an extension list comes straight out of `X509Certificate`. +struct RocketCert<'a, 'b>(&'b [X509Extension<'a>]); -impl CertExt for RocketCert<'_> { +impl CertExt for RocketCert<'_, '_> { fn get_extension_der(&self, oid: &[u64]) -> anyhow::Result>> { let oid = Oid::from(oid).map_err(|_| anyhow::anyhow!("failed to create OID from slice"))?; - let Some(ext) = self.0.extensions().iter().find(|ext| ext.oid == oid) else { + let Some(ext) = self.0.iter().find(|ext| ext.oid == oid) else { return Ok(None); }; Ok(Some(ext.value.to_vec())) @@ -104,7 +108,15 @@ fn verify_gateway_peer(state: &Proxy, cert: Option>) -> Result<( return Err(Status::Unauthorized); }; - let cert = RocketCert(&cert); + authorize_peer(&RocketCert(cert.extensions()), state.my_app_id()) +} + +/// Decide whether a certificate's app identity is one we accept. +/// +/// Split out from `verify_gateway_peer` because that function's other half — the +/// attestation bypass and Rocket's certificate guard — cannot be exercised from a test, +/// which left this decision, the actual authorization rule, uncovered. +fn authorize_peer(cert: &impl CertExt, my_app_id: Option<&[u8]>) -> Result<(), Status> { let remote_app_id = match cert.get_app_id().map_err(|e| { warn!("WaveKV sync: failed to extract app_id from certificate: {e}"); Status::Unauthorized @@ -124,12 +136,8 @@ fn verify_gateway_peer(state: &Proxy, cert: Option>) -> Result<( return Err(Status::Unauthorized); }; - if state.my_app_id() != Some(remote_app_id.as_slice()) { - warn!( - "WaveKV sync: app_id mismatch, expected {:?}, got {:?}", - state.my_app_id(), - remote_app_id - ); + if my_app_id != Some(remote_app_id.as_slice()) { + warn!("WaveKV sync: app_id mismatch, expected {my_app_id:?}, got {remote_app_id:?}"); return Err(Status::Forbidden); } @@ -310,10 +318,25 @@ mod tests { /// A gateway serving the real sync routes over Rocket's local client. /// - /// `insecure_skip_attestation` stands in for the mTLS peer check, which is not what - /// these tests are about; everything below it — route dispatch, the gzip framing, - /// the store split, the uuid check — is the production path. + /// `insecure_skip_attestation` is on, which makes `verify_gateway_peer` return + /// immediately: these tests are about everything below it — route dispatch, the gzip + /// framing, the store split, the uuid check. `enforcing_gateway` covers the gate + /// itself, which this fixture cannot, because Rocket's local client speaks no TLS + /// and so can never present a certificate. async fn serving_gateway(sync_enabled: bool) -> (Client, Proxy, TempDir) { + serving_gateway_with(sync_enabled, true).await + } + + /// The same gateway with the attestation bypass switched off, so the peer check runs + /// for real. + async fn enforcing_gateway() -> (Client, Proxy, TempDir) { + serving_gateway_with(true, false).await + } + + async fn serving_gateway_with( + sync_enabled: bool, + skip_attestation: bool, + ) -> (Client, Proxy, TempDir) { // `main` installs this once at startup; the sync client builds a rustls config, // so a test that skips it panics inside rustls rather than failing an assertion. let _ = rustls::crypto::ring::default_provider().install_default(); @@ -331,7 +354,7 @@ mod tests { .join("wg.conf") .to_string_lossy() .into_owned(); - config.debug.insecure_skip_attestation = true; + config.debug.insecure_skip_attestation = skip_attestation; let tls_config = write_tls_material(temp_dir.path()); let proxy = Proxy::new(ProxyOptions { @@ -349,6 +372,135 @@ mod tests { (client, proxy, temp_dir) } + /// The sync routes are the cluster's write surface: anything that reaches them can + /// insert entries that replicate to every gateway. `verify_gateway_peer` is the only + /// thing standing in front of them, and with `insecure_skip_attestation` set — which + /// every other test here sets — its first statement returns `Ok(())`, so the gate + /// itself was never executed by any test. Replacing the whole function body with + /// `Ok(())` did not turn the suite red. + /// + /// Rocket's local client speaks no TLS and so presents no certificate, which is + /// exactly the case that must be refused. + #[tokio::test] + async fn every_sync_route_refuses_a_peer_it_cannot_identify() { + let (client, _proxy, _tmp) = enforcing_gateway().await; + + for route in [ + "/wavekv/sync/persistent", + "/wavekv/sync2/persistent", + "/wavekv/push/persistent", + ] { + let response = client.post(route).body(Vec::new()).dispatch().await; + assert_eq!( + response.status(), + Status::Unauthorized, + "{route} served a request from an unauthenticated caller" + ); + } + } + + /// A real certificate carrying `PHALA_RATLS_APP_ID`, minted locally. + /// + /// Nothing here needs a TEE: the extension is an ordinary X.509 extension that + /// `CertRequest` adds unconditionally, and the check under test never looks at a + /// quote — it reads two extensions and compares bytes. + fn cert_with_app_id(app_id: &[u8]) -> Vec { + use ra_tls::cert::CertRequest; + use ra_tls::rcgen::KeyPair; + + let key = KeyPair::generate().expect("key"); + let cert = CertRequest::builder() + .key(&key) + .subject("peer.test") + .app_id(app_id) + .build() + .self_signed() + .expect("self-signed cert"); + cert.der().to_vec() + } + + /// A certificate with no app identity at all. + fn cert_without_app_id() -> Vec { + use ra_tls::cert::CertRequest; + use ra_tls::rcgen::KeyPair; + + let key = KeyPair::generate().expect("key"); + let cert = CertRequest::builder() + .key(&key) + .subject("peer.test") + .build() + .self_signed() + .expect("self-signed cert"); + cert.der().to_vec() + } + + fn authorize(der: &[u8], my_app_id: Option<&[u8]>) -> Result<(), Status> { + use rocket::mtls::x509::{FromDer, X509Certificate}; + let (_, parsed) = X509Certificate::from_der(der).expect("parse cert"); + authorize_peer(&RocketCert(parsed.extensions()), my_app_id) + } + + /// The rule the sync routes are defended by: same app id or nothing. + /// + /// Every case below was previously unreachable, because the only tests that touched + /// this code set `insecure_skip_attestation` and returned before it. Inverting the + /// comparison to `==` left the suite green. + #[test] + fn a_peer_is_authorized_only_when_its_app_id_matches_ours() { + let ours = b"app-id-of-this-cluster".to_vec(); + + assert_eq!(authorize(&cert_with_app_id(&ours), Some(&ours)), Ok(())); + + assert_eq!( + authorize(&cert_with_app_id(b"a-different-app"), Some(&ours)), + Err(Status::Forbidden), + "a valid certificate from another app must not reach the sync routes" + ); + } + + /// A certificate that proves nothing about which app presented it is refused, rather + /// than falling through to a comparison against `None`. + #[test] + fn a_certificate_without_an_app_id_is_refused() { + assert_eq!( + authorize(&cert_without_app_id(), Some(b"app-id-of-this-cluster")), + Err(Status::Unauthorized) + ); + } + + /// A gateway that does not know its own app id cannot authorize anyone. Comparing + /// `None` against a present remote id must reject, never match. + #[test] + fn a_gateway_without_an_app_id_authorizes_nobody() { + assert_eq!( + authorize(&cert_with_app_id(b"anything"), None), + Err(Status::Forbidden) + ); + } + + /// The adapter must match the app-id extension by OID and no other. Returning some + /// other extension's bytes would hand `authorize_peer` a value it would happily + /// compare. + #[test] + fn the_adapter_reads_the_app_id_extension_and_not_a_neighbour() { + use ra_tls::traits::CertExt; + use rocket::mtls::x509::{FromDer, X509Certificate}; + + let der = cert_with_app_id(b"the-app-id"); + let (_, parsed) = X509Certificate::from_der(&der).expect("parse cert"); + let adapter = RocketCert(parsed.extensions()); + + assert_eq!( + adapter.get_app_id().expect("read app id"), + Some(b"the-app-id".to_vec()) + ); + assert_eq!( + adapter.get_special_usage().expect("read special usage"), + None, + "an extension that was never set must read back as absent" + ); + } + /// Register the peer so `query_uuid` returns something: the uuid check is opt-in and /// an unknown sender bypasses it entirely. fn register_peer(proxy: &Proxy) { From 6d0d20b11a6e4c3ea0504574e4ebcbe3f837d3cb Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sun, 9 Aug 2026 20:42:18 -0700 Subject: [PATCH 14/21] test(gateway): cover the v1 sync shim at the route level MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The v1 route had no round-trip test. Mutation testing could delete either store arm, invert the node-id-zero guard, or replace the whole response body with three bytes, and the suite stayed green — every route test targeted v2, because v1 is the compatibility path and attention went to the new one. Deleting the `"persistent"` arm is the sharpest of these. It falls through to `_ => 404`, and a 404 on a sync route is exactly the signal a v2 peer reads as "this node has no such route" — so a broken store dispatch would not surface as an error, it would surface as a successful protocol downgrade, cluster-wide and silently, for a whole reprobe window. The suite already documents that reasoning for the sync-disabled 503 case; the v1 route just had nothing enforcing it. Three tests: a round trip that asserts the response decodes and carries the state this node holds, the same for the ephemeral store, and a node-id-zero rejection matching the push and v2 routes. --- dstack/gateway/src/web_routes/wavekv_sync.rs | 91 ++++++++++++++++++++ 1 file changed, 91 insertions(+) diff --git a/dstack/gateway/src/web_routes/wavekv_sync.rs b/dstack/gateway/src/web_routes/wavekv_sync.rs index 2ebf37abb..e421072b8 100644 --- a/dstack/gateway/src/web_routes/wavekv_sync.rs +++ b/dstack/gateway/src/web_routes/wavekv_sync.rs @@ -617,6 +617,97 @@ mod tests { assert_eq!(response.status(), Status::NotFound); } + fn v1_body(msg: &SyncMessage) -> Vec { + gzip(&encode(msg).expect("encode v1 message")).expect("gzip") + } + + fn v1_request() -> SyncMessage { + SyncMessage { + sender_id: PEER, + sender_uuid: peer_uuid(), + // Empty coverage, so the shim answers with everything it holds. + sender_ack: Default::default(), + entries: Vec::new(), + } + } + + /// The v1 shim is how a gateway that has not been upgraded still receives state, and + /// nothing exercised it at the route level: the store dispatch could be deleted, the + /// node-id-zero guard inverted, and the response body replaced with three bytes, + /// all without turning the suite red. + /// + /// Deleting the `"persistent"` arm is the sharpest of those. It falls through to + /// `_ => 404`, and a 404 on a sync route is precisely the signal a v2 peer reads as + /// "this node does not speak that protocol" — so the failure would not look like an + /// error, it would look like a successful protocol downgrade. + #[tokio::test] + async fn a_v1_round_trip_serves_the_state_this_node_holds() { + let (client, proxy, _tmp) = serving_gateway(true).await; + register_peer(&proxy); + proxy + .kv_store() + .persistent() + .write() + .put("node/7".to_string(), b"v".to_vec()) + .expect("seed"); + + let response = client + .post("/wavekv/sync/persistent") + .body(v1_body(&v1_request())) + .dispatch() + .await; + + assert_eq!(response.status(), Status::Ok); + let bytes = response.into_bytes().await.expect("body"); + let decoded: SyncResponse = + decode(&gunzip(&bytes).expect("gunzip")).expect("decode v1 response"); + + assert_eq!(decoded.peer_id, ME); + assert!( + decoded.entries.iter().any(|e| e.key == "node/7"), + "a peer with no coverage must receive the state this node holds" + ); + } + + /// Both stores are reachable over the v1 route. The ephemeral arm carries the + /// liveness data a stale peer needs most, and losing it would read as a downgrade + /// rather than a fault, exactly as above. + #[tokio::test] + async fn the_v1_route_serves_the_ephemeral_store_as_well() { + let (client, proxy, _tmp) = serving_gateway(true).await; + register_peer(&proxy); + + let response = client + .post("/wavekv/sync/ephemeral") + .body(v1_body(&v1_request())) + .dispatch() + .await; + + assert_eq!( + response.status(), + Status::Ok, + "a 404 here would demote this node to no-such-route in the caller's cache" + ); + } + + /// Node id 0 is the unset value, so an entry authored by it collides with every + /// other unset sender. The v1 route rejects it, as the push and v2 routes do. + #[tokio::test] + async fn a_v1_sync_from_node_id_zero_is_refused() { + let (client, proxy, _tmp) = serving_gateway(true).await; + register_peer(&proxy); + + let mut msg = v1_request(); + msg.sender_id = 0; + let response = client + .post("/wavekv/sync/persistent") + .body(v1_body(&msg)) + .dispatch() + .await; + + assert_eq!(response.status(), Status::BadRequest); + } + /// ...which is why a node with sync switched off must answer 503 and not 404. A 404 /// here would demote this node to v1 in every peer's cache for a whole reprobe /// window — silently, and without sync being on to fix it. From 37f943181d6914a8e51d2c0159963b67b118a170 Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sun, 9 Aug 2026 22:23:43 -0700 Subject: [PATCH 15/21] test(gateway): pin the client-side identity check, the sync limits and the key schema MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three gaps mutation testing found, none of which needed any infrastructure — all three are pure functions over bytes. `AppIdValidator::validate` had no tests at all. It runs during the TLS handshake, so a validator that always returns `Ok(())` means this gateway completes a mutually-authenticated connection to any peer holding any certificate our CA signed, and then sends it our state. It is the client-side mirror of the route check covered in 8d04ff2, and it was equally undefended: replacing the body with `Ok(())` or inverting the comparison left the suite green. The decompression-limit test asserted a payload of exactly `MAX_DECOMPRESSED_SYNC_BYTES` is accepted — building that payload from the same constant. It therefore held for whatever the constant said, and shrinking 128 MiB to a few kilobytes kept it green while rejecting every real delta. It pinned `>` against `>=` and nothing else. The limits are now checked against what the protocol actually produces: room for wavekv's 4 MiB delta cap, and a compressed ceiling equal to what the routes accept on a request. The key namespace had no tests either. Every builder and parser survived mutation: `handshake_prefix` could return `""`, `parse_inst_key` could return `Some("xyzzy")`. These strings are how a gateway finds its own state after an upgrade, so changing one orphans every existing record — still replicated, still in the digest, unreachable by any reader. Four properties are now pinned: a prefix matches the keys it iterates, a prefix does not capture a neighbour (`inst-a` must not swallow `inst-ab`), builders and parsers round-trip, and a parser refuses a key from another namespace. --- dstack/gateway/src/kv/https_client.rs | 74 ++++++++++++++++++++ dstack/gateway/src/kv/mod.rs | 64 +++++++++++++++++ dstack/gateway/src/web_routes/wavekv_sync.rs | 31 ++++++++ 3 files changed, 169 insertions(+) diff --git a/dstack/gateway/src/kv/https_client.rs b/dstack/gateway/src/kv/https_client.rs index 731b73d2d..5ff12e9f9 100644 --- a/dstack/gateway/src/kv/https_client.rs +++ b/dstack/gateway/src/kv/https_client.rs @@ -366,3 +366,77 @@ impl CertValidator for AppIdValidator { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + use ra_tls::cert::CertRequest; + use ra_tls::rcgen::KeyPair; + + /// A certificate carrying `PHALA_RATLS_APP_ID`, minted in process. + /// + /// No TEE is involved: `CertRequest` writes the extension unconditionally, and the + /// validator below never looks at a quote — it parses DER and compares bytes. + fn cert_with_app_id(app_id: &[u8]) -> Vec { + let key = KeyPair::generate().expect("key"); + CertRequest::builder() + .key(&key) + .subject("peer.test") + .app_id(app_id) + .build() + .self_signed() + .expect("self-signed cert") + .der() + .to_vec() + } + + fn cert_without_app_id() -> Vec { + let key = KeyPair::generate().expect("key"); + CertRequest::builder() + .key(&key) + .subject("peer.test") + .build() + .self_signed() + .expect("self-signed cert") + .der() + .to_vec() + } + + /// The client half of the same rule the sync routes enforce on inbound requests. + /// + /// This runs during the TLS handshake, so a validator that always returns `Ok(())` + /// means this gateway will complete a mutually-authenticated connection to any peer + /// presenting any certificate our CA signed — and then send it our state. Replacing + /// the whole body with `Ok(())`, or inverting the comparison, left the suite green. + #[test] + fn a_peer_certificate_is_accepted_only_when_its_app_id_matches() { + let ours = b"app-id-of-this-cluster".to_vec(); + let validator = AppIdValidator::new(ours.clone()); + + assert_eq!(validator.validate(&cert_with_app_id(&ours)), Ok(())); + assert!( + validator + .validate(&cert_with_app_id(b"a-different-app")) + .is_err(), + "a certificate from another app must not complete the handshake" + ); + } + + /// A certificate that says nothing about which app holds it proves nothing, and must + /// be refused rather than treated as unconstrained. + #[test] + fn a_peer_certificate_without_an_app_id_is_refused() { + let validator = AppIdValidator::new(b"app-id-of-this-cluster".to_vec()); + let err = validator + .validate(&cert_without_app_id()) + .expect_err("a certificate with no app identity must be refused"); + assert!(err.contains("app_id"), "{err}"); + } + + /// Anything that is not a certificate is a parse failure, not a pass. + #[test] + fn a_malformed_certificate_is_refused() { + let validator = AppIdValidator::new(b"whatever".to_vec()); + assert!(validator.validate(b"not a certificate at all").is_err()); + } +} diff --git a/dstack/gateway/src/kv/mod.rs b/dstack/gateway/src/kv/mod.rs index cc3d3c0f9..2a8b4e341 100644 --- a/dstack/gateway/src/kv/mod.rs +++ b/dstack/gateway/src/kv/mod.rs @@ -1594,3 +1594,67 @@ mod peer_url_tests { } } } + +/// The key namespace is the on-disk contract between releases. +/// +/// Every builder and parser here survived mutation: `handshake_prefix` could return +/// `""`, `parse_inst_key` could return `Some("xyzzy")`, and nothing noticed. That is not +/// a cosmetic gap — these strings are what a gateway uses to find its own state after an +/// upgrade. Changing one silently orphans every existing record: the data is still +/// replicated, still in the digest, and no longer reachable by any reader. +#[cfg(test)] +mod key_schema_tests { + use super::keys; + + /// A prefix must actually be a prefix of the keys it is used to iterate, or a range + /// scan silently returns nothing and the caller reads an empty collection as "none". + #[test] + fn every_iteration_prefix_matches_the_keys_it_must_find() { + assert!(keys::handshake("inst-a", 7).starts_with(&keys::handshake_prefix("inst-a"))); + assert!(keys::last_seen_node(3, 7).starts_with(&keys::last_seen_node_prefix(3))); + assert!(keys::cert_attestation_latest("a.example") + .starts_with(&keys::cert_attestation_prefix("a.example"))); + assert!(keys::cert_attestation_history("a.example", 1234) + .starts_with(&keys::cert_attestation_prefix("a.example"))); + } + + /// A prefix must not be so short that it also matches a neighbour's keys, which + /// would make an iteration return another instance's or node's records. + #[test] + fn an_iteration_prefix_does_not_capture_a_neighbour() { + assert!(!keys::handshake("inst-b", 7).starts_with(&keys::handshake_prefix("inst-a"))); + assert!(!keys::last_seen_node(4, 7).starts_with(&keys::last_seen_node_prefix(3))); + assert!(!keys::cert_attestation_latest("b.example") + .starts_with(&keys::cert_attestation_prefix("a.example"))); + // `inst-a` must not swallow `inst-ab`. + assert!(!keys::handshake("inst-ab", 7).starts_with(&keys::handshake_prefix("inst-a"))); + } + + /// Builders and parsers must agree, or a record written by one release is invisible + /// to the next. + #[test] + fn every_key_parses_back_to_what_built_it() { + assert_eq!(keys::parse_inst_key(&keys::inst("inst-a")), Some("inst-a")); + assert_eq!(keys::parse_node_info_key(&keys::node_info(42)), Some(42)); + assert_eq!( + keys::parse_cert_domain(&keys::cert_attestation_latest("a.example")), + Some("a.example") + ); + assert_eq!( + keys::parse_cert_domain(&keys::cert_lock("a.example")), + Some("a.example") + ); + } + + /// A parser must reject a key from another namespace rather than returning a value + /// derived from it, which would cross-wire two record types. + #[test] + fn a_parser_refuses_a_key_from_another_namespace() { + assert_eq!(keys::parse_inst_key(&keys::node_info(1)), None); + assert_eq!(keys::parse_cert_domain(&keys::inst("inst-a")), None); + assert_eq!(keys::parse_node_info_key(&keys::node_status(1)), None); + assert_eq!(keys::parse_node_info_key(&keys::inst("inst-a")), None); + // `node/info/` and `node/status/` share a stem; neither may claim the other. + assert_eq!(keys::parse_node_info_key("node/info/not-a-number"), None); + } +} diff --git a/dstack/gateway/src/web_routes/wavekv_sync.rs b/dstack/gateway/src/web_routes/wavekv_sync.rs index e421072b8..337120291 100644 --- a/dstack/gateway/src/web_routes/wavekv_sync.rs +++ b/dstack/gateway/src/web_routes/wavekv_sync.rs @@ -763,6 +763,37 @@ mod tests { } } + /// The limits must leave room for the largest legitimate message. + /// + /// The boundary test below asserts a payload of exactly `MAX_DECOMPRESSED_SYNC_BYTES` + /// is accepted — but it builds that payload *from the same constant*, so it holds + /// whatever the constant says. Shrinking the limit to a few kilobytes keeps it green + /// while rejecting every real delta. Pin the values against what production sends, + /// which is the property that actually matters. + // Deliberately runtime assertions rather than `const { assert!(..) }`: a const block + // would fail the build, which mutation testing scores as "unviable" rather than + // "caught", and would lose the message explaining what the number is for. + #[allow(clippy::assertions_on_constants)] + #[test] + fn the_sync_limits_admit_the_largest_message_the_protocol_can_produce() { + // A v2 delta is capped by wavekv's `max_delta_bytes` (4 MiB by default), and the + // v1 shim answers with the whole live state. + const MAX_DELTA_BYTES: usize = 4 * 1024 * 1024; + assert!( + MAX_DECOMPRESSED_SYNC_BYTES >= 8 * MAX_DELTA_BYTES, + "a decompression limit of {MAX_DECOMPRESSED_SYNC_BYTES} bytes would reject \ + ordinary sync traffic, not just a bomb" + ); + + // The compressed ceiling mirrors what the routes accept on a request, so a peer + // cannot answer with more than it would have been allowed to ask. + assert_eq!( + crate::kv::MAX_COMPRESSED_SYNC_BYTES, + 16 * 1024 * 1024, + "this must stay equal to the 16 MiB the routes accept on a request body" + ); + } + /// The limit is inclusive, so a payload landing exactly on it still decodes. Without /// this the bound could tighten by a byte and only the bomb test would still pass. #[test] From 46fb1c5abf26fb8d70a75187b1a59a6b5a97d3ca Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sun, 9 Aug 2026 22:28:52 -0700 Subject: [PATCH 16/21] chore(gateway): pin wavekv at the bootstrap ack-adoption guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A node rebuilt from an empty data directory no longer adopts a requester's ack map while it is bootstrapping — the window in which its own coverage is unknown to it, and in which adopting would have it claim coverage of state it does not hold. --- dstack/Cargo.lock | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/dstack/Cargo.lock b/dstack/Cargo.lock index e87ddfdb3..7d6f2ddea 100644 --- a/dstack/Cargo.lock +++ b/dstack/Cargo.lock @@ -129,7 +129,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -140,7 +140,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -1692,7 +1692,7 @@ dependencies = [ "libc", "option-ext", "redox_users 0.5.2", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -2576,7 +2576,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -3817,7 +3817,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -4676,7 +4676,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -6343,7 +6343,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -6411,7 +6411,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -7267,7 +7267,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -7584,7 +7584,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -8421,7 +8421,7 @@ dependencies = [ [[package]] name = "wavekv" version = "2.0.0" -source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#82c9328b9c161e78e95eced398ee6256581a0312" +source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#688e3c12c5aaf82d7009449f1cda8624384212c0" dependencies = [ "anyhow", "bincode 2.0.1", From 4e22ee6a0e8a06c05b454881b3281df0d328ed65 Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sun, 9 Aug 2026 22:42:29 -0700 Subject: [PATCH 17/21] chore(gateway): pin wavekv at the divergence-gate and cross-page test fixes --- dstack/Cargo.lock | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/dstack/Cargo.lock b/dstack/Cargo.lock index 7d6f2ddea..beb9cb15a 100644 --- a/dstack/Cargo.lock +++ b/dstack/Cargo.lock @@ -129,7 +129,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -140,7 +140,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1692,7 +1692,7 @@ dependencies = [ "libc", "option-ext", "redox_users 0.5.2", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -2576,7 +2576,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -3817,7 +3817,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4676,7 +4676,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -6343,7 +6343,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -6411,7 +6411,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -7267,7 +7267,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -7584,7 +7584,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -8421,7 +8421,7 @@ dependencies = [ [[package]] name = "wavekv" version = "2.0.0" -source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#688e3c12c5aaf82d7009449f1cda8624384212c0" +source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#4ace7a13c838b93e96e544b948741326d4315e4e" dependencies = [ "anyhow", "bincode 2.0.1", From 338133a8ece8f2174f65a0ed3ea67142f3e746df Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sun, 9 Aug 2026 22:45:55 -0700 Subject: [PATCH 18/21] chore(gateway): pin wavekv at the remove_peer durability fix --- dstack/Cargo.lock | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/dstack/Cargo.lock b/dstack/Cargo.lock index beb9cb15a..be40e4dbf 100644 --- a/dstack/Cargo.lock +++ b/dstack/Cargo.lock @@ -129,7 +129,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -140,7 +140,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -1692,7 +1692,7 @@ dependencies = [ "libc", "option-ext", "redox_users 0.5.2", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -2576,7 +2576,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -3817,7 +3817,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -4676,7 +4676,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -6343,7 +6343,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -6411,7 +6411,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -7267,7 +7267,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -7584,7 +7584,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -8421,7 +8421,7 @@ dependencies = [ [[package]] name = "wavekv" version = "2.0.0" -source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#4ace7a13c838b93e96e544b948741326d4315e4e" +source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#655c387dfb6756a66a20e223876f1c839bca8227" dependencies = [ "anyhow", "bincode 2.0.1", From 135d46389f914fc8105cc902c92344f96e830f5b Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sun, 9 Aug 2026 22:48:03 -0700 Subject: [PATCH 19/21] chore(gateway): pin wavekv at the v1 rollback membership coverage --- dstack/Cargo.lock | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/dstack/Cargo.lock b/dstack/Cargo.lock index be40e4dbf..2c9710d94 100644 --- a/dstack/Cargo.lock +++ b/dstack/Cargo.lock @@ -129,7 +129,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -140,7 +140,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1692,7 +1692,7 @@ dependencies = [ "libc", "option-ext", "redox_users 0.5.2", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -2576,7 +2576,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -3817,7 +3817,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4676,7 +4676,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -6343,7 +6343,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -6411,7 +6411,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -7267,7 +7267,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -7584,7 +7584,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -8421,7 +8421,7 @@ dependencies = [ [[package]] name = "wavekv" version = "2.0.0" -source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#655c387dfb6756a66a20e223876f1c839bca8227" +source = "git+https://github.com/Phala-Network/wavekv?branch=feat%2Fdelta-state-sync#a95014ae79c4f95f1e869125c777fe253b12a79d" dependencies = [ "anyhow", "bincode 2.0.1", From 8b78d586fad0b5b024a1d05a55e94be16c048210 Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sun, 9 Aug 2026 23:11:03 -0700 Subject: [PATCH 20/21] test(gateway): exercise protocol negotiation against a real TLS peer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `post_bytes_probe`'s mapping of 404/405 to `Ok(None)` *is* the v1/v2 negotiation: a gateway that has not been upgraded has no `/wavekv/sync2` route, and that status is the only signal its peers get. Every mutation of the condition survived — `||` to `&&`, either `==` to `!=`, the `!` on `is_success` — because nothing exercised the function at all. It cannot be reached without a peer that speaks TLS, since the client is built `https_only()`, and that was enough friction for the whole file to sit at zero. A listener on 127.0.0.1 with a certificate minted in process is enough. No container, no simulator: nothing on this path verifies a quote. Four cases: both "no such route" statuses read as not-upgraded; a 5xx or 4xx stays an error, because reading one as not-upgraded would demote a healthy v2 peer to the v1 path for a whole reprobe window; a 200 is decompressed and returned; and an oversized body is refused. The last one initially passed with the bound removed entirely. Its payload was not valid gzip, so `gunzip_bounded` rejected it whatever the ceiling said, and the assertion measured nothing. It now sends stored-mode gzip — valid, and large enough to clear the compressed ceiling while decompressing well inside the decompressed one — so only the bound under test can reject it. --- dstack/gateway/src/kv/https_client.rs | 175 ++++++++++++++++++++++++++ 1 file changed, 175 insertions(+) diff --git a/dstack/gateway/src/kv/https_client.rs b/dstack/gateway/src/kv/https_client.rs index 5ff12e9f9..b8ea78233 100644 --- a/dstack/gateway/src/kv/https_client.rs +++ b/dstack/gateway/src/kv/https_client.rs @@ -440,3 +440,178 @@ mod tests { assert!(validator.validate(b"not a certificate at all").is_err()); } } + +/// Negotiation and response-bounding tested against a real TLS peer. +/// +/// `post_bytes_probe`'s mapping of 404/405 to `Ok(None)` *is* the v1/v2 protocol +/// negotiation: a gateway that has not been upgraded has no `/wavekv/sync2` route, and +/// that status is the only signal its peers get. Every mutation of that condition +/// survived, because nothing exercised the function at all — it needs a peer that speaks +/// TLS, and `https_only()` means a plain HTTP stub will not do. +/// +/// No container and no TEE: a local listener with a certificate minted in process. +#[cfg(test)] +mod transport_tests { + use super::*; + use hyper::service::service_fn; + use hyper::{Response, StatusCode}; + use hyper_util::rt::TokioIo; + use std::convert::Infallible; + use tokio::net::TcpListener; + use tokio_rustls::TlsAcceptor; + + /// A CA plus a leaf valid for 127.0.0.1, written where `HttpsClient::new` expects. + fn tls_material(dir: &std::path::Path) -> (HttpsClientConfig, Vec, Vec) { + use ra_tls::rcgen::{BasicConstraints, CertificateParams, IsCa, KeyPair}; + + let ca_key = KeyPair::generate().expect("ca key"); + let mut ca_params = CertificateParams::new(vec![]).expect("ca params"); + ca_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained); + let ca_cert = ca_params.self_signed(&ca_key).expect("ca cert"); + + let leaf_key = KeyPair::generate().expect("leaf key"); + let leaf_params = + CertificateParams::new(vec!["127.0.0.1".to_string()]).expect("leaf params"); + let leaf_cert = leaf_params + .signed_by(&leaf_key, &ca_cert, &ca_key) + .expect("leaf cert"); + + let cert_path = dir.join("node.crt"); + let key_path = dir.join("node.key"); + let ca_path = dir.join("ca.crt"); + std::fs::write(&cert_path, leaf_cert.pem()).expect("write cert"); + std::fs::write(&key_path, leaf_key.serialize_pem()).expect("write key"); + std::fs::write(&ca_path, ca_cert.pem()).expect("write ca"); + + ( + HttpsClientConfig { + cert_path: cert_path.to_string_lossy().into_owned(), + key_path: key_path.to_string_lossy().into_owned(), + ca_cert_path: ca_path.to_string_lossy().into_owned(), + cert_validator: None, + }, + leaf_cert.der().to_vec(), + leaf_key.serialize_der(), + ) + } + + /// Serve one fixed response over TLS and return the URL to reach it. + async fn serve(status: StatusCode, body: Vec, cert: Vec, key: Vec) -> String { + let certs = vec![rustls::pki_types::CertificateDer::from(cert)]; + let key = rustls::pki_types::PrivateKeyDer::try_from(key).expect("server key"); + let config = rustls::ServerConfig::builder() + .with_no_client_auth() + .with_single_cert(certs, key) + .expect("server config"); + let acceptor = TlsAcceptor::from(Arc::new(config)); + + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("addr"); + + tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + let acceptor = acceptor.clone(); + let body = body.clone(); + tokio::spawn(async move { + let Ok(tls) = acceptor.accept(stream).await else { + return; + }; + let _ = hyper::server::conn::http1::Builder::new() + .serve_connection( + TokioIo::new(tls), + service_fn(move |_req| { + let body = body.clone(); + async move { + Ok::<_, Infallible>( + Response::builder() + .status(status) + .body(Full::new(Bytes::from(body))) + .expect("response"), + ) + } + }), + ) + .await; + }); + } + }); + + format!("https://127.0.0.1:{}/wavekv/sync2/persistent", addr.port()) + } + + fn gzip(bytes: &[u8]) -> Vec { + let mut encoder = GzEncoder::new(Vec::new(), Compression::fast()); + encoder.write_all(bytes).expect("gzip"); + encoder.finish().expect("gzip finish") + } + + async fn probe(status: StatusCode, body: Vec) -> Result>> { + let _ = rustls::crypto::ring::default_provider().install_default(); + let dir = tempfile::tempdir().expect("tempdir"); + let (config, cert, key) = tls_material(dir.path()); + let url = serve(status, body, cert, key).await; + HttpsClient::new(&config) + .expect("client") + .post_bytes_probe(&url, b"request".to_vec()) + .await + } + + /// A peer still on v1 has no `/wavekv/sync2` route. Both statuses a router can give + /// for that must read as "not upgraded", not as a failure — a failure would be + /// retried forever instead of falling back to the v1 route. + #[tokio::test] + async fn a_missing_route_reads_as_not_upgraded() { + assert_eq!( + probe(StatusCode::NOT_FOUND, Vec::new()).await.unwrap(), + None + ); + assert_eq!( + probe(StatusCode::METHOD_NOT_ALLOWED, Vec::new()) + .await + .unwrap(), + None + ); + } + + /// ...and any other failure must stay a failure. Reading a 500 as "not upgraded" + /// would demote a healthy v2 peer to the v1 path for a whole reprobe window. + #[tokio::test] + async fn a_server_error_is_not_mistaken_for_a_missing_route() { + assert!(probe(StatusCode::INTERNAL_SERVER_ERROR, Vec::new()) + .await + .is_err()); + assert!(probe(StatusCode::BAD_REQUEST, Vec::new()).await.is_err()); + } + + /// A peer that answers gets its body decompressed and returned. + #[tokio::test] + async fn an_upgraded_peer_returns_its_decoded_body() { + let payload = b"the-envelope-bytes".to_vec(); + let got = probe(StatusCode::OK, gzip(&payload)).await.unwrap(); + assert_eq!(got, Some(payload)); + } + + /// The response body is bounded before it is decompressed, so a peer cannot spend + /// our memory ahead of any decoding limit. + /// + /// The body must be *valid* gzip that merely exceeds the compressed ceiling. A + /// malformed one is rejected by `gunzip_bounded` whatever the ceiling says, so it + /// would pass this test with the bound removed entirely — which is exactly what the + /// first version of it did. Stored-mode gzip keeps the encoded size at roughly the + /// input size, so the payload clears the ceiling while decompressing well inside it. + #[tokio::test] + async fn an_oversized_response_body_is_refused() { + let stored = { + let mut encoder = GzEncoder::new(Vec::new(), Compression::none()); + encoder + .write_all(&vec![0u8; super::super::MAX_COMPRESSED_SYNC_BYTES + 1]) + .expect("gzip"); + encoder.finish().expect("gzip finish") + }; + assert!( + stored.len() > super::super::MAX_COMPRESSED_SYNC_BYTES, + "the fixture depends on the compressed body clearing the ceiling" + ); + assert!(probe(StatusCode::OK, stored).await.is_err()); + } +} From 9a6e217438718b9a54574bfdba81193d7abd44dd Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sun, 9 Aug 2026 23:16:32 -0700 Subject: [PATCH 21/21] test(gateway): close the remaining transport gaps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three survivors were left on the TLS client after the negotiation tests, and all three sit on paths this cluster depends on. `post_compressed_msg` is the v1 sync path — how a v2 gateway talks to one that has not been upgraded. Its status check was as untested as the negotiation's, so a v1 peer answering 500 could have been decoded as a successful round. `post_json` is the bootnode GetPeers path, where the threat model does not assume the peer is honest, and a failure status must not be parsed as a peer list. The third needed the custom-verifier path. `AppIdValidator` runs inside `CustomCertVerifier`, which rustls only reaches once standard chain verification passes, so unit-testing the validator alone leaves the wiring between them untested — and the wiring is what decides whether a peer from another app can open a connection at all. It now serves a certificate carrying a foreign app id and asserts the handshake fails before any application bytes move, with the matching id as the control. Deleting the validator call turns it red. --- dstack/gateway/src/kv/https_client.rs | 113 ++++++++++++++++++++++++++ 1 file changed, 113 insertions(+) diff --git a/dstack/gateway/src/kv/https_client.rs b/dstack/gateway/src/kv/https_client.rs index b8ea78233..bf0e15afc 100644 --- a/dstack/gateway/src/kv/https_client.rs +++ b/dstack/gateway/src/kv/https_client.rs @@ -591,6 +591,119 @@ mod transport_tests { assert_eq!(got, Some(payload)); } + /// A server certificate carrying an app id, signed by the same test CA. + fn app_id_server_cert( + dir: &std::path::Path, + app_id: &[u8], + ) -> (HttpsClientConfig, Vec, Vec) { + use ra_tls::cert::CertRequest; + use ra_tls::rcgen::{BasicConstraints, CertificateParams, IsCa, KeyPair}; + + let ca_key = KeyPair::generate().expect("ca key"); + let mut ca_params = CertificateParams::new(vec![]).expect("ca params"); + ca_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained); + let ca_cert = ca_params.self_signed(&ca_key).expect("ca cert"); + + let leaf_key = KeyPair::generate().expect("leaf key"); + let alt_names = vec!["127.0.0.1".to_string()]; + let leaf_cert = CertRequest::builder() + .key(&leaf_key) + .subject("peer.test") + .alt_names(&alt_names) + .app_id(app_id) + .usage_server_auth(true) + .build() + .signed_by(&ca_cert, &ca_key) + .expect("leaf cert"); + + let cert_path = dir.join("node.crt"); + let key_path = dir.join("node.key"); + let ca_path = dir.join("ca.crt"); + std::fs::write(&cert_path, leaf_cert.pem()).expect("write cert"); + std::fs::write(&key_path, leaf_key.serialize_pem()).expect("write key"); + std::fs::write(&ca_path, ca_cert.pem()).expect("write ca"); + + ( + HttpsClientConfig { + cert_path: cert_path.to_string_lossy().into_owned(), + key_path: key_path.to_string_lossy().into_owned(), + ca_cert_path: ca_path.to_string_lossy().into_owned(), + cert_validator: None, + }, + leaf_cert.der().to_vec(), + leaf_key.serialize_der(), + ) + } + + /// The client-side identity check, over a real handshake rather than a direct call. + /// + /// `AppIdValidator` runs inside `CustomCertVerifier`, which rustls only reaches once + /// standard chain verification passes — so unit-testing the validator alone leaves + /// the wiring untested. A peer from another app must fail to connect at all, before + /// any application bytes move. + #[tokio::test] + async fn a_peer_from_another_app_cannot_complete_the_handshake() { + let _ = rustls::crypto::ring::default_provider().install_default(); + let ours = b"app-id-of-this-cluster".to_vec(); + + for (server_app_id, expect_ok) in + [(ours.clone(), true), (b"a-different-app".to_vec(), false)] + { + let dir = tempfile::tempdir().expect("tempdir"); + let (mut config, cert, key) = app_id_server_cert(dir.path(), &server_app_id); + config.cert_validator = Some(Arc::new(AppIdValidator::new(ours.clone()))); + let url = serve(StatusCode::NOT_FOUND, Vec::new(), cert, key).await; + + let got = HttpsClient::new(&config) + .expect("client") + .post_bytes_probe(&url, b"x".to_vec()) + .await; + + if expect_ok { + assert_eq!( + got.expect("a peer from our own app must connect"), + None, + "the 404 should still read as not-upgraded" + ); + } else { + assert!( + got.is_err(), + "a peer from another app completed the handshake" + ); + } + } + } + + /// `post_compressed_msg` is the v1 sync path — how a v2 gateway talks to one that + /// has not been upgraded. Its status check was as untested as the negotiation's, so + /// a v1 peer answering 500 could have been decoded as a successful round. + #[tokio::test] + async fn a_failed_v1_sync_is_not_decoded_as_a_response() { + let _ = rustls::crypto::ring::default_provider().install_default(); + let dir = tempfile::tempdir().expect("tempdir"); + let (config, cert, key) = tls_material(dir.path()); + let url = serve(StatusCode::INTERNAL_SERVER_ERROR, Vec::new(), cert, key).await; + let client = HttpsClient::new(&config).expect("client"); + let out: Result = client.post_compressed_msg(&url, &1u32).await; + assert!(out.is_err(), "a 500 from a v1 peer must not decode"); + } + + /// `post_json` is the bootnode GetPeers path, and the threat model does not assume a + /// bootnode is honest — so a failure status must not be parsed as a peer list. + #[tokio::test] + async fn a_failed_bootnode_fetch_is_not_parsed_as_peers() { + let _ = rustls::crypto::ring::default_provider().install_default(); + let dir = tempfile::tempdir().expect("tempdir"); + let (config, cert, key) = tls_material(dir.path()); + let url = serve(StatusCode::FORBIDDEN, b"null".to_vec(), cert, key).await; + let client = HttpsClient::new(&config).expect("client"); + let out: Result> = client.post_json(&url, &()).await; + assert!( + out.is_err(), + "a 403 from a bootnode must not parse as a body" + ); + } + /// The response body is bounded before it is decompressed, so a peer cannot spend /// our memory ahead of any decoding limit. ///