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/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 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/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..93c97ab68 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)?; @@ -422,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)) }; @@ -443,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)) }) @@ -458,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)) }) @@ -472,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)) }) @@ -488,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)) }) @@ -507,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)?; @@ -547,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) @@ -567,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(); @@ -583,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()); @@ -605,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 1c1765c70..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"); }; @@ -540,13 +557,35 @@ 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 // warp ledger-sync server/importer dispatch to. let warp_ledger_unified = matches!(storage_config.separation, StorageSeparation::Unified); - let new_partial_components = - new_partial(&config, epoch_config.clone(), midnight_cfg, storage_config, tx_filter_config)?; + + // Decided here rather than at the point of use: `new_partial` needs it too, to know whether to + // install the GRANDPA pruning filter that keeps this node able to serve warp proofs. One + // computation, both consumers. + // + // Non-validators serve by default. Validators don't — arena serialization is the protocol's + // most CPU-expensive operation and must not compete with authoring/finality (a remote DoS + // vector) — unless the operator opts in via `--serve-warp-ledger-sync` (for small or local + // networks with no non-validator nodes to serve). `--no-serve-warp-ledger-sync` opts out + // outright, for non-validators whose CPU must not be spent on other nodes' warp sync (public + // RPC endpoints, bootnodes). Every node still registers the protocol as a client, so it can + // warp-sync and recover its own arena regardless. Clap rejects both flags together. + let serve_ledger_sync = + !no_serve_warp_ledger_sync && (!config.role.is_authority() || serve_warp_ledger_sync); + + let new_partial_components = new_partial( + &config, + epoch_config.clone(), + midnight_cfg, + storage_config, + tx_filter_config, + serve_ledger_sync, + )?; let sc_service::PartialComponents { client, @@ -613,13 +652,7 @@ pub async fn new_full 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)); + } +} 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); + } }