From 7add22424b4dcde40974510032f71d9a35ddd84f Mon Sep 17 00:00:00 2001 From: Giles Cope Date: Fri, 31 Jul 2026 11:47:30 +0100 Subject: [PATCH 1/5] fix(warp-sync): bound per-peer arena serialization on the ledger-sync server Serializing the ledger arena is the most expensive thing this node does for a remote peer, and the request that triggers it is ~44 bytes. The handler had no rate limiting and no reputation reporting, and memoized exactly one target block, so alternating two finalized hashes evicted the memo on every request and forced a full serialize + compress each time. One peer could pin the handler thread indefinitely and starve honest warp-syncing peers. Substrate's state_request_handler -- which this is patterned on -- already carries seen_requests plus rep::SAME_REQUEST for the same class of abuse, at a fraction of the per-request cost. Carry that over and add the bound the memo cannot provide on its own: - seen_requests LRU keyed (peer, target, offset), penalising a peer that replays a byte-identical range. An honest client pages each offset once. - snapshot memo widened to a 3-entry LRU, so alternation is a hit and nearby targets share work. This bounds memory, not CPU. - per-peer serialization budget, charged *before* the work: a peer cycling target blocks to defeat a memo of any fixed size is refused rather than served-then-penalised. This is the actual resource bound; the reputation change only accelerates eviction. Cheap rejections (unknown block, not finalized, undecodable request) stay unpenalised -- an honest peer racing finality or a reorg produces those, and banning for our own timing would cost us good peers. Assisted-by: Claude:claude-opus-5 claude-code --- Cargo.lock | 1 + node/Cargo.toml | 1 + node/src/warp_ledger_sync/server.rs | 350 +++++++++++++++++++++++++--- 3 files changed, 319 insertions(+), 33 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 820f0ba85..7cb61e2fa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7906,6 +7906,7 @@ dependencies = [ "jsonrpsee", "local-ip-address", "log", + "lru 0.18.2", "midnight-node-ledger", "midnight-node-ledger-helpers", "midnight-node-res", diff --git a/node/Cargo.toml b/node/Cargo.toml index 400fcd4e6..9ab2234ab 100644 --- a/node/Cargo.toml +++ b/node/Cargo.toml @@ -27,6 +27,7 @@ async-trait.workspace = true futures.workspace = true # Must match the version sc-network's `request_response_config` inbound queue expects (1.9). async-channel = "1.9" +lru.workspace = true sc-cli.workspace = true sc-chain-spec.workspace = true diff --git a/node/src/warp_ledger_sync/server.rs b/node/src/warp_ledger_sync/server.rs index 9bfa61032..a4827f456 100644 --- a/node/src/warp_ledger_sync/server.rs +++ b/node/src/warp_ledger_sync/server.rs @@ -21,14 +21,33 @@ //! //! Verification is the *client's* job: the server is untrusted, so it performs no crypto — //! it only serves bytes whose recomputed root the client checks against the on-chain `StateKey`. +//! +//! ## Abuse resistance +//! +//! Serializing the arena is by far the most expensive thing this node does on behalf of a remote +//! peer, and the request that triggers it is ~44 bytes. Three bounds keep that asymmetry from being +//! a remote CPU amplifier, in increasing order of importance: +//! +//! 1. **Snapshot memo** (`SNAPSHOT_CACHE_ENTRIES`) — the many range requests a client makes while +//! paging one blob cost one serialization, not one each. Holding several entries also means +//! concurrent clients converging on *nearby* finalized targets share the work. +//! 2. **Replay penalty** (`MAX_NUMBER_OF_SAME_REQUESTS_PER_PEER`) — an honest client asks for each +//! `(target, offset)` exactly once. Mirrors substrate's `state_request_handler`. +//! 3. **Per-peer serialization budget** (`MAX_SERIALIZATIONS_PER_PEER`) — the load-bearing one. +//! A memo of any fixed size is defeated by cycling `target_hash` across more distinct blocks +//! than it holds, so the budget bounds how much expensive work a single peer can ever induce, +//! independent of the memo size. It is charged *before* the work, so an over-budget peer is +//! refused rather than served-and-then-penalised — the reputation change only accelerates +//! eviction; the refusal is the actual resource bound. -use std::{marker::PhantomData, sync::Arc, time::Duration}; +use std::{marker::PhantomData, num::NonZeroUsize, sync::Arc, time::Duration}; use futures::StreamExt; +use lru::LruCache; use parity_scale_codec::{Decode, Encode}; use sc_client_api::{Backend, StorageProvider}; use sc_network::{ - MAX_RESPONSE_SIZE, NetworkBackend, + MAX_RESPONSE_SIZE, NetworkBackend, PeerId, ReputationChange, request_responses::{IncomingRequest, OutgoingResponse}, }; use sp_blockchain::HeaderBackend; @@ -45,19 +64,108 @@ const MAX_REQUEST_SIZE: u64 = 1024; /// Request timeout, matching substrate's state protocol. const REQUEST_TIMEOUT: Duration = Duration::from_secs(40); +/// How many times a peer may replay a byte-identical `(target, offset)` request before being +/// penalised. Matches substrate's `state_request_handler`. +const MAX_NUMBER_OF_SAME_REQUESTS_PER_PEER: usize = 2; + +/// How many distinct target blocks one peer may make us serialize the arena for. +/// +/// An honest warp client needs exactly **one**: its warp target. Every subsequent range request for +/// that target is a memo hit, and a client that fails verification and retries against us re-uses +/// the same target. The headroom covers a peer that legitimately re-warps to a newer target over a +/// long-lived connection (e.g. after its own restart). +const MAX_SERIALIZATIONS_PER_PEER: usize = 4; + +/// How many distinct target blocks' compressed snapshots to memoize. +/// +/// Deliberately small: this bounds *memory* (each entry is a full compressed arena), while +/// [`MAX_SERIALIZATIONS_PER_PEER`] — not this — bounds the *CPU* an attacker can induce. Three +/// entries let a few clients on slightly different finalized targets share serializations without +/// holding several arena-sized blobs resident indefinitely. +const SNAPSHOT_CACHE_ENTRIES: usize = 3; + +mod rep { + use sc_network::ReputationChange as Rep; + + /// Peer replayed a byte-identical `(target, offset)` request. Same penalty substrate applies + /// for the same behaviour on the state protocol. + pub const SAME_REQUEST: Rep = Rep::new(i32::MIN, "Same ledger-sync request multiple times"); + + /// Peer induced more full-arena serializations than any honest client needs, by cycling + /// `target_hash` to defeat the snapshot memo. Heavy but not an instant ban — the refusal in + /// [`super::LedgerSyncRequestHandler::blob_for`] is the resource bound; this just gets the + /// peer evicted sooner. + pub const TARGET_CYCLING: Rep = Rep::new(-(1 << 20), "Ledger-sync target cycling"); +} + +/// Key of [`LedgerSyncRequestHandler::seen_requests`]. +/// +/// `Hash`/`Eq` are written by hand rather than derived so the impls don't pick up a spurious +/// `B: Hash + Eq` bound from the generic parameter (as substrate's equivalent does). +struct SeenRequestsKey { + peer: PeerId, + target: B::Hash, + offset: u64, +} + +impl Clone for SeenRequestsKey { + fn clone(&self) -> Self { + Self { peer: self.peer, target: self.target, offset: self.offset } + } +} + +impl PartialEq for SeenRequestsKey { + fn eq(&self, other: &Self) -> bool { + self.peer == other.peer && self.target == other.target && self.offset == other.offset + } +} + +impl Eq for SeenRequestsKey {} + +impl std::fmt::Debug for SeenRequestsKey { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SeenRequestsKey") + .field("peer", &self.peer) + .field("target", &self.target) + .field("offset", &self.offset) + .finish() + } +} + +impl std::hash::Hash for SeenRequestsKey { + fn hash(&self, state: &mut H) { + self.peer.hash(state); + self.target.hash(state); + self.offset.hash(state); + } +} + +/// Value of [`LedgerSyncRequestHandler::seen_requests`]. +enum SeenRequestsValue { + /// Seen once, not yet answered. + First, + /// Answered `n` times. + Fulfilled(usize), +} + /// Handler for incoming ledger-sync requests from warp-syncing peers. /// -/// Memoizes the serialized blob for the most-recently-served target block so that the many -/// byte-range requests a single client makes while paging the blob do not each re-serialize the -/// (multi-million node) arena. +/// Memoizes serialized blobs so that the many byte-range requests a single client makes while +/// paging a blob do not each re-serialize the (multi-million node) arena, and tracks per-peer +/// request patterns so that neither replaying a range nor cycling target blocks can turn a 44-byte +/// request into unbounded CPU. See the module docs. pub struct LedgerSyncRequestHandler { client: Arc, /// Whether the ledger arena uses the unified ParityDb layout (selects the DB instantiation the /// serializer dispatches to — see [`midnight_node_ledger::serialize_ledger_snapshot`]). unified: bool, request_receiver: async_channel::Receiver, - /// `(target_block, compressed serialized blob)` memo for the last block served. - cache: Option<(B::Hash, CachedSnapshot)>, + /// Compressed serialized blobs, keyed by target block. + snapshot_cache: LruCache, + /// Replay detector: how many times each `(peer, target, offset)` has been answered. + seen_requests: LruCache, SeenRequestsValue>, + /// How many arena serializations each peer has already cost us. + serializations_per_peer: LruCache, _phantom: PhantomData, } @@ -77,12 +185,13 @@ where /// `serve` is true — the handler to spawn via [`run`](Self::run). /// /// `serve` gates the **server** side only. Validators pass `serve = false` unless they opt in - /// via `--serve-warp-ledger-sync`: serializing the multi-million-node arena is this - /// protocol's most CPU-expensive operation, and it must never compete with a validator's - /// authoring/finality duties (an easy remote DoS vector). A non-serving node advertises no - /// inbound queue, so the network routes no requests to it — but the protocol is still - /// registered, so the node can act as a warp-sync *client* and recover its own arena. Returns - /// `None` for the handler when not serving. + /// via `--serve-warp-ledger-sync`, and any node can opt out with `--no-serve-warp-ledger-sync`: + /// serializing the multi-million-node arena is this protocol's most CPU-expensive operation, and + /// it must never compete with a validator's authoring/finality duties (an easy remote DoS + /// vector). A non-serving node advertises no inbound queue, so the network layer marks the + /// protocol `Outbound`-only and routes no requests to it — but the protocol is still registered, + /// so the node can act as a warp-sync *client* and recover its own arena. Returns `None` for the + /// handler when not serving. pub fn new::Hash>>( genesis_hash: B::Hash, fork_id: Option<&str>, @@ -98,8 +207,18 @@ where // Reserve one in-flight request slot per peer. let capacity = std::cmp::max(num_peer_hint, 1); let (tx, request_receiver) = async_channel::bounded(capacity); - let handler = - Self { client, unified, request_receiver, cache: None, _phantom: PhantomData }; + // Two in-flight `(target, offset)` keys per peer, matching substrate's sizing of the + // same structure; one budget entry per peer. + let seen_capacity = nonzero(capacity.saturating_mul(2)); + let handler = Self { + client, + unified, + request_receiver, + snapshot_cache: LruCache::new(nonzero(SNAPSHOT_CACHE_ENTRIES)), + seen_requests: LruCache::new(seen_capacity), + serializations_per_peer: LruCache::new(nonzero(capacity)), + _phantom: PhantomData, + }; (Some(tx), Some(handler)) } else { (None, None) @@ -122,45 +241,82 @@ where while let Some(IncomingRequest { peer, payload, pending_response }) = self.request_receiver.next().await { - let result = match self.handle_request(&payload) { - Ok(bytes) => Ok(bytes), + let (result, reputation_changes) = match self.handle_request(&peer, &payload) { + Ok(bytes) => (Ok(bytes), Vec::new()), Err(e) => { log::debug!(target: LOG_TARGET, "ledger-sync request from {peer} failed: {e}"); - Err(()) + (Err(()), e.reputation_change().into_iter().collect()) }, }; // A failed send just means the peer disconnected; nothing to do. let _ = pending_response.send(OutgoingResponse { result, - reputation_changes: Vec::new(), + reputation_changes, sent_feedback: None, }); } } - fn handle_request(&mut self, payload: &[u8]) -> Result, HandleError> { + fn handle_request(&mut self, peer: &PeerId, payload: &[u8]) -> Result, HandleError> { let req = LedgerSyncRequest::::decode(&mut &payload[..])?; - let snapshot = self.blob_for(req.target_hash)?; - Ok(build_response(&snapshot.compressed_blob, snapshot.raw_len, req.offset, req.max_len) - .encode()) + + // Replay detection, mirroring substrate's `state_request_handler`: an honest client pages + // each `(target, offset)` exactly once, so a replay is a broken client or deliberate load. + let key = SeenRequestsKey { peer: *peer, target: req.target_hash, offset: req.offset }; + match self.seen_requests.get_mut(&key) { + Some(SeenRequestsValue::First) => {}, + Some(SeenRequestsValue::Fulfilled(requests)) => { + *requests = requests.saturating_add(1); + if *requests > MAX_NUMBER_OF_SAME_REQUESTS_PER_PEER { + return Err(HandleError::SameRequest); + } + }, + None => { + self.seen_requests.put(key.clone(), SeenRequestsValue::First); + }, + } + + let snapshot = self.blob_for(peer, req.target_hash)?; + let bytes = + build_response(&snapshot.compressed_blob, snapshot.raw_len, req.offset, req.max_len) + .encode(); + + // Only now does this count as answered, so a request refused above is not held against the + // peer as a fulfilment. + if let Some(value) = self.seen_requests.get_mut(&key) + && matches!(value, SeenRequestsValue::First) + { + *value = SeenRequestsValue::Fulfilled(1); + } + + Ok(bytes) } /// Return the compressed serialized `Ledger`-rooted blob for `target`, building and memoizing it - /// on a cache miss. Rejects unknown or not-yet-finalized blocks. - fn blob_for(&mut self, target: B::Hash) -> Result { - if let Some((cached, snapshot)) = &self.cache - && *cached == target - { + /// on a cache miss. Rejects unknown or not-yet-finalized blocks, and refuses peers that have + /// exhausted their serialization budget. + fn blob_for(&mut self, peer: &PeerId, target: B::Hash) -> Result { + if let Some(snapshot) = self.snapshot_cache.get(&target) { return Ok(snapshot.clone()); } - // Only serve finalized blocks whose state we hold: an unknown hash or a block beyond our - // finalized number is rejected (the warp target is always finalized). + // Cheap rejections first. An unknown hash or a block beyond our finalized number costs a + // header lookup, not a serialization, so it consumes no budget and earns no reputation + // penalty — an honest peer can race finality or a reorg and ask for a block we can't serve. let header = self.client.header(target)?.ok_or(HandleError::UnknownBlock)?; if *header.number() > self.client.info().finalized_number { return Err(HandleError::NotFinalized); } + // Everything past here is the expensive path, so charge it to the peer *before* doing the + // work: an over-budget peer is refused, not served-then-penalised. A peer cycling targets + // to defeat the snapshot memo hits this rather than the memo size. + let spent = self.serializations_per_peer.get(peer).copied().unwrap_or(0); + if spent >= MAX_SERIALIZATIONS_PER_PEER { + return Err(HandleError::TargetCycling); + } + self.serializations_per_peer.put(*peer, spent + 1); + // Read the raw `pallet_midnight::StateKey` at the target block. let state_key = read_state_key::(&self.client, target)? .ok_or(HandleError::NoStateKey)?; @@ -171,17 +327,26 @@ where let compressed_blob = compress_snapshot(&blob).map_err(HandleError::Compress)?; log::debug!( target: LOG_TARGET, - "Serialized ledger snapshot for {target:?}: {} bytes raw, {} bytes compressed", + "Serialized ledger snapshot for {target:?}: {} bytes raw, {} bytes compressed \ + (serialization {} of {} for {peer})", raw_len, - compressed_blob.len() + compressed_blob.len(), + spent + 1, + MAX_SERIALIZATIONS_PER_PEER, ); let snapshot = CachedSnapshot { compressed_blob: Arc::new(compressed_blob), raw_len }; - self.cache = Some((target, snapshot.clone())); + self.snapshot_cache.put(target, snapshot.clone()); Ok(snapshot) } } +/// `NonZeroUsize` from a capacity that is only zero if a caller passed nonsense; clamp rather than +/// panic, since these are all "how much to remember" knobs where 1 is a valid answer. +fn nonzero(n: usize) -> NonZeroUsize { + NonZeroUsize::new(n.max(1)).expect("max(1) is non-zero; qed") +} + #[derive(Debug, thiserror::Error)] enum HandleError { #[error("failed to decode request / state key: {0}")] @@ -198,4 +363,123 @@ enum HandleError { Serialize(String), #[error("failed to compress ledger snapshot: {0}")] Compress(snap::Error), + #[error( + "peer replayed an identical ledger-sync request more than {MAX_NUMBER_OF_SAME_REQUESTS_PER_PEER} times" + )] + SameRequest, + #[error( + "peer exhausted its budget of {MAX_SERIALIZATIONS_PER_PEER} arena serializations by cycling target blocks" + )] + TargetCycling, +} + +impl HandleError { + /// Reputation change to attach when refusing for this reason. + /// + /// Only the two abuse patterns are penalised. A malformed, unknown-block, or unfinalized-block + /// request is cheap to reject and an honest peer racing finality or a reorg can produce one, so + /// penalising those would ban good peers for our own timing. + fn reputation_change(&self) -> Option { + match self { + HandleError::SameRequest => Some(rep::SAME_REQUEST), + HandleError::TargetCycling => Some(rep::TARGET_CYCLING), + HandleError::Decode(_) + | HandleError::Client(_) + | HandleError::UnknownBlock + | HandleError::NotFinalized + | HandleError::NoStateKey + | HandleError::Serialize(_) + | HandleError::Compress(_) => None, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use sc_network::peer_store::BANNED_THRESHOLD; + + #[test] + fn only_abuse_patterns_are_penalised() { + // The two abuse patterns carry a penalty... + assert!(HandleError::SameRequest.reputation_change().is_some()); + assert!(HandleError::TargetCycling.reputation_change().is_some()); + + // ...and the reasons an honest peer can hit — racing finality, a reorg, a truncated + // request — carry none. Penalising these would ban good peers for our own timing. + assert!(HandleError::UnknownBlock.reputation_change().is_none()); + assert!(HandleError::NotFinalized.reputation_change().is_none()); + assert!(HandleError::NoStateKey.reputation_change().is_none()); + assert!( + HandleError::Decode(parity_scale_codec::Error::from("truncated")) + .reputation_change() + .is_none() + ); + } + + /// These are all compile-time constants; the assertions exist to catch a future edit that + /// silently inverts the intended severity ordering, not to test runtime behaviour. + #[allow(clippy::assertions_on_constants)] + #[test] + fn penalty_severities_are_ordered_as_intended() { + // Target cycling is refused *before* the work happens, so its reputation change only needs + // to accelerate eviction. Keeping it above the ban threshold means a peer that trips it + // through some benign pattern we haven't foreseen can recover, while a persistent offender + // still accumulates its way to a ban. + let cycling = rep::TARGET_CYCLING.value; + assert!(cycling < 0, "must be a penalty"); + assert!( + cycling > BANNED_THRESHOLD, + "{cycling} should not ban on a single occurrence (threshold {BANNED_THRESHOLD})" + ); + + // Replaying a byte-identical request has no benign explanation, so it is strictly harsher + // than target cycling and bans on sight, matching substrate's state protocol. + assert!(rep::SAME_REQUEST.value < cycling); + } + + /// Constant guard-rails, as above. + #[allow(clippy::assertions_on_constants)] + #[test] + fn snapshot_cache_holds_more_than_one_target() { + // A single-entry memo is defeated by alternating two targets: each request evicts the + // other's blob and forces a fresh arena serialization. The LRU must hold at least two so + // that alternation is a hit, and the per-peer budget — not the memo — bounds a peer that + // cycles more targets than the memo can hold. + assert!(SNAPSHOT_CACHE_ENTRIES >= 2); + assert!(MAX_SERIALIZATIONS_PER_PEER >= 1, "an honest client needs one serialization"); + } + + #[test] + fn seen_requests_key_distinguishes_peer_target_and_offset() { + // Guards the assumption `handle_request` relies on: the replay counter is keyed by + // (peer, target, offset), so paging distinct offsets never trips the replay penalty. + let peer = PeerId::random(); + let target = sp_core::H256::repeat_byte(1); + let mut lru: LruCache< + SeenRequestsKey, + SeenRequestsValue, + > = LruCache::new(nonzero(2)); + + let key_at = |offset| SeenRequestsKey:: { + peer, + target, + offset, + }; + lru.put(key_at(0), SeenRequestsValue::First); + lru.put(key_at(1), SeenRequestsValue::First); + assert!(lru.get(&key_at(0)).is_some()); + assert!(lru.get(&key_at(1)).is_some()); + + // Distinct offsets are distinct keys, so a client paging forward is never a "same request". + assert_ne!(key_at(0), key_at(1)); + + // A different peer asking for the same range is also a distinct key. + let other = SeenRequestsKey:: { + peer: PeerId::random(), + target, + offset: 0, + }; + assert_ne!(key_at(0), other); + } } From d6f3fd804850464751e97168c82b6823fc91e284 Mon Sep 17 00:00:00 2001 From: Giles Cope Date: Fri, 31 Jul 2026 11:47:39 +0100 Subject: [PATCH 2/5] feat(warp-sync): add --no-serve-warp-ledger-sync opt-out Non-validators served ledger snapshots unconditionally, with no way to turn it off. That is exactly backwards for the nodes whose exposure to arbitrary peers is highest -- public RPC endpoints and bootnodes -- which are non-validators and so had no opt-out at all. Add the counterpart to --serve-warp-ledger-sync. Passing both is rejected by clap rather than resolved by silent precedence, so an operator who sets both is told to pick one instead of quietly getting whichever the code happened to check first. Serving off still leaves the protocol registered as Outbound, so the node can warp-sync as a client. Assisted-by: Claude:claude-opus-5 claude-code --- node/src/cli.rs | 54 +++++++++++++++++++++++++++++++++++++++++++-- node/src/command.rs | 1 + node/src/service.rs | 16 +++++++++++--- 3 files changed, 66 insertions(+), 5 deletions(-) diff --git a/node/src/cli.rs b/node/src/cli.rs index dab6a6236..348a487f8 100644 --- a/node/src/cli.rs +++ b/node/src/cli.rs @@ -50,12 +50,21 @@ pub struct RunMidnight { /// Serve Midnight ledger snapshots to warp-syncing peers even when running as a validator. /// - /// Non-validator nodes always serve. Validators don't by default — serializing the ledger - /// arena is the warp ledger-sync protocol's most CPU-expensive operation and competes with + /// Non-validator nodes serve by default. Validators don't — serializing the ledger arena is the + /// warp ledger-sync protocol's most CPU-expensive operation and competes with /// authoring/finality duties — but can opt in with this flag, e.g. on small or local networks /// that have no non-validator nodes. Nodes can warp-sync as clients regardless of this flag. #[arg(long)] pub serve_warp_ledger_sync: bool, + + /// Never serve Midnight ledger snapshots to warp-syncing peers. + /// + /// The opt-out counterpart of `--serve-warp-ledger-sync`, for non-validators that must not + /// spend CPU on other nodes' warp sync — public RPC endpoints and bootnodes, whose exposure to + /// arbitrary peers is highest. Overrides the serve-by-default for non-validators; passing both + /// flags is rejected rather than silently resolved. The node can still warp-sync as a client. + #[arg(long, conflicts_with = "serve_warp_ledger_sync")] + pub no_serve_warp_ledger_sync: bool, } #[derive(Debug, clap::Parser)] @@ -538,3 +547,44 @@ impl std::fmt::Display for NotImplementedError { } } impl core::error::Error for NotImplementedError {} + +#[cfg(test)] +mod tests { + use super::*; + use clap::Parser; + + /// Serving is off by default for validators (they must opt in) and the opt-out is separate, so + /// neither flag may be set unless the operator asked for it. + #[test] + fn warp_ledger_sync_flags_default_off() { + let run = RunMidnight::try_parse_from(["midnight-node"]).expect("bare run parses"); + assert!(!run.serve_warp_ledger_sync); + assert!(!run.no_serve_warp_ledger_sync); + } + + #[test] + fn warp_ledger_sync_flags_parse_individually() { + let opt_in = RunMidnight::try_parse_from(["midnight-node", "--serve-warp-ledger-sync"]) + .expect("opt-in parses"); + assert!(opt_in.serve_warp_ledger_sync); + assert!(!opt_in.no_serve_warp_ledger_sync); + + let opt_out = RunMidnight::try_parse_from(["midnight-node", "--no-serve-warp-ledger-sync"]) + .expect("opt-out parses"); + assert!(opt_out.no_serve_warp_ledger_sync); + assert!(!opt_out.serve_warp_ledger_sync); + } + + /// "Serve" and "never serve" together is operator error, not something to resolve by silent + /// precedence — the operator should be told which one they meant. + #[test] + fn warp_ledger_sync_flags_conflict() { + let err = RunMidnight::try_parse_from([ + "midnight-node", + "--serve-warp-ledger-sync", + "--no-serve-warp-ledger-sync", + ]) + .expect_err("both flags together must be rejected"); + assert_eq!(err.kind(), clap::error::ErrorKind::ArgumentConflict); + } +} diff --git a/node/src/command.rs b/node/src/command.rs index a52ff4a2e..e7bf00157 100644 --- a/node/src/command.rs +++ b/node/src/command.rs @@ -336,6 +336,7 @@ fn run_node(cfg: Cfg) -> sc_cli::Result<()> { tx_filter_config, run_midnight.rpc_max_finality_subscriptions, run_midnight.serve_warp_ledger_sync, + run_midnight.no_serve_warp_ledger_sync, ) .await .map_err(sc_cli::Error::Service)?; diff --git a/node/src/service.rs b/node/src/service.rs index 1c1765c70..30889e261 100644 --- a/node/src/service.rs +++ b/node/src/service.rs @@ -540,6 +540,7 @@ pub async fn new_full Result<(TaskManager, Arc), ServiceError> { let database_source = config.database.clone(); // Captured before `storage_config` is moved into `new_partial`: selects the ParityDb layout the @@ -617,9 +618,12 @@ pub async fn new_full Date: Fri, 31 Jul 2026 11:47:51 +0100 Subject: [PATCH 3/5] fix(warp-sync): bound single-peer ledger transfer time required_chunk_len bounds the *number* of requests spent on one peer, not the wall clock. With ~1 MiB ranges and the protocol's 40s request timeout, a peer answering each range just inside that timeout holds arena recovery open for roughly thirteen hours while remaining, technically, responsive. Recovery is sequential across peers and the block-import gate is held throughout, so this is a stall of the whole node, not just of one fetch. Add the time bound the size bound cannot give: - a per-chunk timeout well under the protocol timeout, so a peer that simply goes quiet is dropped in seconds rather than tying up a slot for 40s - a whole-transfer deadline derived from the advertised size and a minimum throughput, so a large arena over a slow link still completes but a peer that stops making real progress loses its turn The first range keeps the full protocol timeout: on a cold server memo it legitimately pays for the entire arena serialization. The budget is measured from the start of the transfer, so a peer cannot buy extra time by being slow to answer that first request. Assisted-by: Claude:claude-opus-5 claude-code --- node/src/warp_ledger_sync/client.rs | 139 ++++++++++++++++++++++++++-- 1 file changed, 133 insertions(+), 6 deletions(-) diff --git a/node/src/warp_ledger_sync/client.rs b/node/src/warp_ledger_sync/client.rs index 201dc6981..21aae6d32 100644 --- a/node/src/warp_ledger_sync/client.rs +++ b/node/src/warp_ledger_sync/client.rs @@ -24,8 +24,16 @@ //! //! Verification + persistence live in the ledger crate (next to the arena); this module is pure //! network orchestration. No peer is trusted: a bad blob fails the root check and is discarded. +//! +//! Untrusted peers are bounded in time as well as in content — see `fetch_blob_from` for the +//! per-chunk timeout and whole-transfer throughput floor that stop one peer from holding recovery +//! open indefinitely. -use std::{marker::PhantomData, sync::Arc}; +use std::{ + marker::PhantomData, + sync::Arc, + time::{Duration, Instant}, +}; use parity_scale_codec::{Decode, Encode}; use sc_client_api::{Backend, StorageProvider}; @@ -33,6 +41,7 @@ use sc_network::{ IfDisconnected, NetworkRequest, PeerId, ProtocolName, request_responses::RequestFailure, }; use sp_runtime::traits::Block as BlockT; +use tokio::time::timeout; use super::{ LOG_TARGET, @@ -163,14 +172,26 @@ where /// Fetch the full compressed blob from a single peer by paging contiguous byte ranges in order, /// then decompress it to the canonical `Ledger`-rooted blob. /// - /// Every chunk must be full-size ([`required_chunk_len`]) — an honest server always fills the - /// requested range — so a peer drip-feeding tiny (or empty) chunks fails immediately instead of - /// tying the client up in an unbounded request loop. + /// Two independent bounds keep one peer from monopolising recovery: + /// - **Size**: every chunk must be full-size ([`required_chunk_len`]) — an honest server always + /// fills the requested range — so a peer drip-feeding tiny (or empty) chunks fails + /// immediately instead of tying the client up in an unbounded *number* of requests. + /// - **Time**: [`CHUNK_TIMEOUT`] per range plus a whole-transfer deadline from + /// [`transfer_budget`]. Full-size chunks bound the request *count*, not the wall clock: a peer + /// answering each 1 MiB range just inside the protocol's 40 s timeout would otherwise hold + /// recovery open for hours (a slowloris). The deadline scales with the advertised size, so a + /// genuinely large arena over a slow link still completes, but a peer must sustain + /// [`MIN_THROUGHPUT_BYTES_PER_SEC`] to keep its turn. /// /// (Parallel / multi-peer range fetch is a possible future optimization; the /// `ChunkAssembler` already supports resume by `next_offset`.) async fn fetch_blob_from(&self, peer: PeerId, target: B::Hash) -> Result, ClientError> { - // First range establishes the compressed transfer length and expected raw size. + let started = Instant::now(); + + // First range establishes the compressed transfer length and expected raw size. It is the + // one request that may legitimately be slow — on a cold server cache it pays for the whole + // arena serialization — so it gets the protocol's own request timeout rather than the + // tighter per-chunk one applied to the rest of the transfer. let first = self.request_range(peer, target, 0).await?; let compressed_total_len = first.compressed_total_len; let raw_total_len = first.raw_total_len; @@ -179,8 +200,27 @@ where ensure_full_chunk(&first)?; assembler.accept(first.offset, &first.bytes)?; + // Budget starts from the size the peer just advertised, measured from the start of the + // transfer so the peer cannot buy extra time by having been slow to answer the first range. + let budget = transfer_budget(compressed_total_len); + let deadline = started + budget; + while !assembler.is_complete() { - let next = self.request_range(peer, target, assembler.next_offset()).await?; + let elapsed = started.elapsed(); + if elapsed >= budget { + return Err(ClientError::TransferTooSlow { + got: assembler.next_offset(), + total: compressed_total_len, + elapsed, + }); + } + // Cap each range at the shorter of the per-chunk timeout and the remaining budget, so + // the last request cannot overrun the deadline it was checked against. + let chunk_timeout = CHUNK_TIMEOUT.min(deadline - Instant::now()); + let offset = assembler.next_offset(); + let next = timeout(chunk_timeout, self.request_range(peer, target, offset)) + .await + .map_err(|_| ClientError::ChunkTimeout { offset, waited: chunk_timeout })??; if next.compressed_total_len != compressed_total_len || next.raw_total_len != raw_total_len { @@ -216,6 +256,36 @@ where } } +/// Wall clock a peer gets to answer a single range request once the transfer is under way. +/// +/// Deliberately well under the protocol's 40 s request timeout: by this point the server is serving +/// from its memoized blob, so a range is a memcpy and a send. A peer that has simply gone quiet is +/// dropped in seconds rather than tying up a slot for the full protocol timeout. +const CHUNK_TIMEOUT: Duration = Duration::from_secs(10); + +/// Throughput a peer must sustain across the whole transfer to keep its turn. +/// +/// Any node healthy enough to be worth recovering from serves state sync far faster than this; the +/// floor exists to convert "technically still responding" into a failure the driver can act on. +const MIN_THROUGHPUT_BYTES_PER_SEC: u64 = 256 * 1024; + +/// Fixed allowance added to the throughput-derived budget, covering per-request round trips and a +/// server whose first response had to serialize the arena. +const TRANSFER_GRACE: Duration = Duration::from_secs(60); + +/// Wall clock allowed for a whole single-peer transfer of `compressed_total_len` bytes. +/// +/// Scales with the advertised size rather than being a flat timeout, so a large arena over a slow +/// link is still recoverable while a peer that stops making real progress is dropped. +/// +/// The argument is the *compressed* length, so the worst case a peer can advertise is +/// `max_compress_len(MAX_LEDGER_SYNC_RAW_BYTES)` (~1.17 GiB), not the 1 GiB raw ceiling — about 81 +/// minutes. For a realistic arena, minutes. +fn transfer_budget(compressed_total_len: u64) -> Duration { + TRANSFER_GRACE + .saturating_add(Duration::from_secs(compressed_total_len / MIN_THROUGHPUT_BYTES_PER_SEC)) +} + /// Require a response chunk to be full-size for its offset (see [`required_chunk_len`]). Oversized /// chunks are fine (more progress than required); the assembler's overflow check still bounds them. fn ensure_full_chunk(response: &LedgerSyncResponse) -> Result<(), ClientError> { @@ -254,6 +324,63 @@ pub enum ClientError { UndersizedChunk { offset: u64, got: u64, required: u64 }, #[error("failed to decompress ledger snapshot: {0}")] Decompress(#[from] DecompressError), + #[error("peer did not answer the range at offset {offset} within {waited:?}")] + ChunkTimeout { offset: u64, waited: Duration }, + #[error( + "peer served {got} of {total} bytes in {elapsed:?}, below the required \ + {MIN_THROUGHPUT_BYTES_PER_SEC} B/s" + )] + TransferTooSlow { got: u64, total: u64, elapsed: Duration }, #[error("all peers failed to provide a verifiable snapshot")] AllPeersFailed, } + +#[cfg(test)] +mod tests { + use super::*; + use crate::warp_ledger_sync::protocol::MAX_LEDGER_SYNC_RAW_BYTES; + + #[test] + fn transfer_budget_scales_with_size_and_is_always_positive() { + // An empty/tiny transfer still gets the fixed grace, so a fast small blob is never + // failed for being quick. + assert_eq!(transfer_budget(0), TRANSFER_GRACE); + assert_eq!(transfer_budget(1), TRANSFER_GRACE); + + // The budget is grace + size/throughput. + let ten_mib = 10 * 1024 * 1024; + assert_eq!( + transfer_budget(ten_mib), + TRANSFER_GRACE + Duration::from_secs(ten_mib / MIN_THROUGHPUT_BYTES_PER_SEC) + ); + + // Monotonic: a bigger advertised blob never buys less time. + assert!(transfer_budget(ten_mib) > transfer_budget(0)); + } + + #[test] + fn worst_case_transfer_budget_is_bounded() { + // `transfer_budget` takes the *compressed* length, so the largest value a peer can get past + // `validate_snapshot_lengths` is snappy's worst-case expansion of the raw ceiling (~1.17 + // GiB), not the 1 GiB raw ceiling itself. Budget against that, or the bound is asserted + // against an underestimate of what an attacker can actually claim. + let worst_compressed = + snap::raw::max_compress_len(MAX_LEDGER_SYNC_RAW_BYTES as usize) as u64; + assert!(worst_compressed > MAX_LEDGER_SYNC_RAW_BYTES, "compressed ceiling is the larger"); + + // Even there, one peer's turn is capped well inside two hours rather than the ~13 hours + // that `MAX_LEDGER_SYNC_CHUNK`-sized ranges at the 40s protocol timeout would allow. + let worst = transfer_budget(worst_compressed); + assert!( + worst < Duration::from_secs(2 * 60 * 60), + "worst-case single-peer budget {worst:?} should stay well under 2h" + ); + } + + #[test] + fn chunk_timeout_is_tighter_than_the_protocol_timeout() { + // The point of the per-chunk bound is to drop a quiet peer faster than the 40s + // request-response timeout would. If this ever inverts, the bound is dead code. + assert!(CHUNK_TIMEOUT < Duration::from_secs(40)); + } +} From 0f0ce1d6b20e9e7d771cdeffb3431eb43da4fe0f Mon Sep 17 00:00:00 2001 From: Giles Cope Date: Fri, 31 Jul 2026 11:47:58 +0100 Subject: [PATCH 4/5] docs(warp-sync): record the serving bounds in the change file The change file described only --serve-warp-ledger-sync. Note the opt-out and the abuse bounds on both sides, since both are operator-visible: serving can now refuse a peer, and recovery can now give up on one. Assisted-by: Claude:claude-opus-5 claude-code --- changes/added/ledger-warp-sync.md | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/changes/added/ledger-warp-sync.md b/changes/added/ledger-warp-sync.md index a80118a3a..eceae46be 100644 --- a/changes/added/ledger-warp-sync.md +++ b/changes/added/ledger-warp-sync.md @@ -2,8 +2,14 @@ Adds support for syncing ledger state, while warp syncing. Should enable Substrate warp sync. Non-validator nodes serve ledger snapshots to warp-syncing peers; validators don't by default, -but can opt in via the new `--serve-warp-ledger-sync` flag (off by default). Nodes can warp-sync -as clients regardless of the flag. +but can opt in via the new `--serve-warp-ledger-sync` flag (off by default), and any node can opt +out with `--no-serve-warp-ledger-sync`. Nodes can warp-sync as clients regardless of either flag. + +Serving is bounded against abuse: snapshots are memoized in a small LRU, peers that replay an +identical byte range are penalised, and each peer has a budget for how many full-arena +serializations it may induce, charged before the work rather than after. On the client side each +range request has its own timeout and the whole per-peer transfer has a throughput floor, so one +slow peer cannot hold arena recovery open indefinitely. PR: https://github.com/midnightntwrk/midnight-node/pull/1650 -Issue: https://github.com/midnightntwrk/midnight-node/issues/1648 \ No newline at end of file +Issue: https://github.com/midnightntwrk/midnight-node/issues/1648 From b52d224f086a868011472df1f26644e7d0b753fc Mon Sep 17 00:00:00 2001 From: Giles Cope Date: Tue, 11 Aug 2026 09:27:33 +0100 Subject: [PATCH 5/5] fix(warp-sync): retain GRANDPA justifications when serving warp sync A warp proof is built from the blocks carrying GRANDPA justifications at authority-set changes. Under `--blocks-pruning ` those bodies are pruned like any other, so a block-pruned node silently stops being a viable warp-sync server -- it still advertises the protocol, it just can no longer answer. substrate provides `GrandpaPruningFilter` for exactly this, and `DatabaseSettings::pruning_filters` to install it. Midnight wired neither, so the field was left empty. Not a live fault today, because `--blocks-pruning` defaults to `archive-canonical` and nothing is pruned -- but this PR is what makes Midnight a warp-serving network, so the trap is newly reachable by any operator reclaiming disk. Installed only when this node serves, so a node that never serves does not retain blocks it has no use for. That makes the decision a `new_partial` input, which also removes the duplicated serve/no-serve expression: it is now computed once in `new_full` and shared with the protocol registration. Note the filter only protects blocks pruned from here on. Enabling serving later on a node that has already run block-pruned leaves holes in its justification history that this cannot repair -- see the comment at the call site. Assisted-by: Claude:claude-opus-5 claude-code --- node/src/command.rs | 20 +++++++++++++++++ node/src/service.rs | 53 +++++++++++++++++++++++++++++++++++---------- 2 files changed, 61 insertions(+), 12 deletions(-) diff --git a/node/src/command.rs b/node/src/command.rs index e7bf00157..93c97ab68 100644 --- a/node/src/command.rs +++ b/node/src/command.rs @@ -423,6 +423,8 @@ fn run_subcommand(subcommand: Subcommand, cfg: Cfg) -> sc_cli::Result<()> { midnight_cfg, storage_config, tx_filter_config, + // Subcommands never serve warp sync. + false, )?; Ok((client, task_manager, other.5.authority_selection)) }; @@ -444,6 +446,8 @@ fn run_subcommand(subcommand: Subcommand, cfg: Cfg) -> sc_cli::Result<()> { cfg.midnight_cfg.clone(), storage_config, tx_filter_config, + // Subcommands never serve warp sync. + false, )?; Ok((cmd.run(client, import_queue), task_manager)) }) @@ -459,6 +463,8 @@ fn run_subcommand(subcommand: Subcommand, cfg: Cfg) -> sc_cli::Result<()> { cfg.midnight_cfg.clone(), storage_config, tx_filter_config, + // Subcommands never serve warp sync. + false, )?; Ok((cmd.run(client, config.database), task_manager)) }) @@ -473,6 +479,8 @@ fn run_subcommand(subcommand: Subcommand, cfg: Cfg) -> sc_cli::Result<()> { cfg.midnight_cfg.clone(), storage_config, tx_filter_config, + // Subcommands never serve warp sync. + false, )?; Ok((cmd.run(client, config.chain_spec), task_manager)) }) @@ -489,6 +497,8 @@ fn run_subcommand(subcommand: Subcommand, cfg: Cfg) -> sc_cli::Result<()> { cfg.midnight_cfg.clone(), storage_config, tx_filter_config, + // Subcommands never serve warp sync. + false, )?; Ok((cmd.run(client, import_queue), task_manager)) }) @@ -508,6 +518,8 @@ fn run_subcommand(subcommand: Subcommand, cfg: Cfg) -> sc_cli::Result<()> { cfg.midnight_cfg.clone(), storage_config, tx_filter_config, + // Subcommands never serve warp sync. + false, )?; let aux_revert = Box::new(|client, _, blocks| { sc_consensus_grandpa::revert(client, blocks)?; @@ -548,6 +560,8 @@ fn run_subcommand(subcommand: Subcommand, cfg: Cfg) -> sc_cli::Result<()> { cfg.midnight_cfg.clone(), storage_config, tx_filter_config, + // Subcommands never serve warp sync. + false, )?; cmd.run(partial.client) @@ -568,6 +582,8 @@ fn run_subcommand(subcommand: Subcommand, cfg: Cfg) -> sc_cli::Result<()> { cfg.midnight_cfg.clone(), storage_config, tx_filter_config, + // Subcommands never serve warp sync. + false, )?; let db = partial.backend.expose_db(); let storage = partial.backend.expose_storage(); @@ -584,6 +600,8 @@ fn run_subcommand(subcommand: Subcommand, cfg: Cfg) -> sc_cli::Result<()> { cfg.midnight_cfg.clone(), storage_config, tx_filter_config, + // Subcommands never serve warp sync. + false, )?; let ext_builder = RemarkBuilder::new(partial.client.clone()); @@ -606,6 +624,8 @@ fn run_subcommand(subcommand: Subcommand, cfg: Cfg) -> sc_cli::Result<()> { cfg.midnight_cfg.clone(), storage_config, tx_filter_config, + // Subcommands never serve warp sync. + false, )?; // Register the *Remark* and *TKA* builders. let ext_factory = ExtrinsicFactory(vec![Box::new(RemarkBuilder::new( diff --git a/node/src/service.rs b/node/src/service.rs index 30889e261..375b25f47 100644 --- a/node/src/service.rs +++ b/node/src/service.rs @@ -288,6 +288,7 @@ pub fn new_partial( midnight_cfg: MidnightCfg, storage_config: StorageInit, tx_filter_config: TxFilterConfig, + serve_warp_ledger_sync: bool, ) -> Result { let mc_follower_metrics = register_metrics_warn_errors(config.prometheus_registry()); let midnight_metrics = @@ -327,6 +328,22 @@ pub fn new_partial( let executor = sc_service::new_wasm_executor(&config.executor); let mut db_config = config.db_config(); + + // Serving warp sync needs the GRANDPA justifications at authority-set changes to survive + // `--blocks-pruning`; without the filter a block-pruned node silently stops being a viable + // warp-sync server, because the warp proof is built from exactly those blocks. Costs nothing + // under the default `archive-canonical`, where no bodies are pruned at all. + // + // Conditional on serving, so a node that never serves does not retain blocks it has no use + // for. NOTE: the filter only protects blocks pruned *from here on*. Enabling serving later on + // a node that has already run block-pruned leaves holes in its justification history that + // this cannot repair. + if serve_warp_ledger_sync { + db_config + .pruning_filters + .push(std::sync::Arc::new(sc_consensus_grandpa::GrandpaPruningFilter)); + } + let DatabaseSource::ParityDb { path: db_path } = db_config.source else { panic!("Midnight node support only parity-db as a backend"); }; @@ -546,8 +563,29 @@ pub async fn new_full