diff --git a/Cargo.lock b/Cargo.lock index 71aa8d7b..d9824d3c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2969,7 +2969,7 @@ dependencies = [ [[package]] name = "dig-node-core" -version = "0.65.0" +version = "0.67.0" dependencies = [ "async-trait", "axum", @@ -3031,7 +3031,7 @@ dependencies = [ [[package]] name = "dig-node-service" -version = "0.222.0" +version = "0.223.0" dependencies = [ "async-trait", "axum", diff --git a/Cargo.toml b/Cargo.toml index 63b51984..14e4da07 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,7 +32,7 @@ edition = "2021" # the ROOT manifest (`[workspace.package].version`), so it MUST be set here for a # release to fire (§3.6). The library crates (dig-node-core/dig-runtime/dig-wallet) # keep their own independent versions — only the released binary tracks the workspace version. -version = "0.222.0" +version = "0.223.0" # Release hardening, matching digstore: keep integer-overflow checks ON in release. # The node parses untrusted serialized input and does offset/length arithmetic over diff --git a/crates/dig-node-core/Cargo.toml b/crates/dig-node-core/Cargo.toml index 10e7f31a..66e40940 100644 --- a/crates/dig-node-core/Cargo.toml +++ b/crates/dig-node-core/Cargo.toml @@ -30,7 +30,7 @@ name = "dig-node-core" # dig-node#276/#296). Changing a public return type is BREAKING for an out-of-workspace implementor; # this crate is consumed in-workspace only and is pre-1.0, so it is a MINOR bump under SemVer's 0.x # rule -- recorded here rather than letting the number imply the locator surface held still. -version = "0.65.0" +version = "0.67.0" edition = "2021" license = "GPL-2.0-only" description = "The canonical DIG node ENGINE library (crate `dig_node_core`): the JSON-RPC dispatch (`handle_rpc`, the same contract as rpc.dig.net), local-first content serve/fetch/redirect from LOCAL .dig store modules (via digstore_host::serve_blind), chain-anchored-root resolution, chain-watch + subscriptions + generation gap-fill, the LRU cache, and the full P2P stack. Shared UNCHANGED by both host shells: the `dig-node` OS-service binary (dig-node-service) and the DIG Browser's in-process cdylib (dig-runtime). Native Rust so the compiled-module serve path works." diff --git a/crates/dig-node-core/src/download.rs b/crates/dig-node-core/src/download.rs index 6c5c4e84..2c795c6e 100644 --- a/crates/dig-node-core/src/download.rs +++ b/crates/dig-node-core/src/download.rs @@ -856,6 +856,15 @@ pub struct NodeContent { /// at all, because pool membership is its liveness gate. Folding them together would make one /// structure whose key, lifetime and eviction rule all mean two different things at once. ask_routing: AskRoutingState, + /// What this node has observed each pool peer DO, and the dial share that earns them (#268). + /// + /// Held beside [`Self::ask_routing`] because the two read the same verified identity and are + /// bounded by the same pool: routing ranks the peers worth asking FIRST, conduct decides which are + /// worth asking AT ALL. Node-local and never gossiped — a shared reputation channel would be a + /// defamation primitive. + conduct: crate::seams::dig_peer::conduct::ConductState, + /// The instant conduct ticks are measured from. Wall-clock-free and monotonic. + conduct_epoch: std::time::Instant, } /// What a holder search ESTABLISHED — the records it found AND whether an empty result is a fact. @@ -1357,6 +1366,8 @@ impl NodeContent { ask_seen: AskSeenSet::new(), onion_relay: std::sync::atomic::AtomicBool::new(onion_relay_from_env()), ask_routing, + conduct: crate::seams::dig_peer::conduct::ConductState::new(), + conduct_epoch: std::time::Instant::now(), }) } @@ -1842,15 +1853,22 @@ impl NodeContent { .filter_map(|(peer, _)| RoutedPeer::from_pool_key(peer)) .collect(); + // CONDUCT gates who is dialable at all (#268, SPEC 8.3); routing then ranks what remains. + // The order matters: ranking a peer this node has PROVEN dishonest would still spend a dial on + // it whenever the ranking happened to favour it. A peer excluded here has a verifiable fault + // against it — a lie or a self-contradiction — and never merely a slow or silent history, + // which `dial_share` floors above zero precisely so distress cannot evict an honest holder. + let dialable = self.conduct.dialable(&routable, self.conduct_ticks()); + // Ranked by what THIS node has observed, not by the pool `HashMap`'s arbitrary order — and the - // observations of peers no longer in `routable` are dropped in the same call, so a cycled-away + // observations of peers no longer in `dialable` are dropped in the same call, so a cycled-away // peer leaves this node's memory when it leaves the pool. let decision = self.ask_routing.decide( &config, asker, budget.remaining(config.hop_cap), me, - &routable, + &dialable, self.relay_rate_limiter.check(requestor), ); @@ -1938,6 +1956,30 @@ impl NodeContent { // The ONLY writer of this node's routing memory, fed an outcome this node classified from // an exchange it issued and saw complete (dig_ecosystem#3129). self.ask_routing.record(routed, &outcome, started.elapsed()); + // The SAME exchange, classified for conduct (#268). An outcome where the peer answered — + // including an honest "I do not have it" — is an `HonestAnswer`, because SPEC 8.2A + // requires that answering is never worse than staying silent. A refusal, a timeout and an + // unreachable peer are `NonPerformance`: unverifiable, decaying, and floored, since none + // of them can be distinguished from distress an attacker induced in an honest peer. + // + // Neither VERIFIABLE class is produced here, and deliberately so. A `ProvenLie` needs + // bytes that failed verification against the anchor, attributed to the peer that supplied + // them; that attribution happens inside `dig-download`'s engine against `chunk_hashes` and + // is not surfaced per-peer to this node (see the report on #268). Claiming one from a + // transport error would brand an honest peer on unverifiable evidence, which is the exact + // conflation SPEC 8.2A exists to prevent. + self.conduct.observe( + routed, + match outcome { + AskOutcome::Answered(_) | AskOutcome::AnsweredInconclusive(_) => { + dig_sex::ConductEvidence::HonestAnswer + } + AskOutcome::Refused | AskOutcome::TimedOut | AskOutcome::Unreachable => { + dig_sex::ConductEvidence::NonPerformance + } + }, + self.conduct_ticks(), + ); match outcome { AskOutcome::Answered(records) => answers.records.extend(records), // The peer answered and told us its OWN subtree did not finish. Its records are @@ -1974,6 +2016,16 @@ impl NodeContent { .collect() } + /// This node's monotonic conduct clock, in seconds since process start. + /// + /// Conduct decay is measured in elapsed ticks and nothing else, so the clock must advance on its + /// own — a counter incremented per exchange would mean a peer nobody dials never ages out of its + /// penalty, and the recovery SPEC 8.2A requires would be unreachable for exactly the peer being + /// punished. + fn conduct_ticks(&self) -> u64 { + self.conduct_epoch.elapsed().as_secs() + } + /// This node's routing memory, so a test can drive the forwarded ask and then ask what the ask /// LEFT BEHIND. Without it the recording leg would only be observable through its effect on a /// later round, and a test that could not see the write directly could not tell a missing write diff --git a/crates/dig-node-core/src/lib.rs b/crates/dig-node-core/src/lib.rs index 12bb7058..f52bd3db 100644 --- a/crates/dig-node-core/src/lib.rs +++ b/crates/dig-node-core/src/lib.rs @@ -491,6 +491,12 @@ pub struct Node { /// registry. The registry's distinct-generation cap ([`crate::seams::dig_peer::DEFAULT_MAX_CONCURRENT_WARMS`]) /// therefore bounds concurrent acquisitions across BOTH legs, not each in isolation. capsule_acquisition: Arc, + /// Inbound admission for the mTLS peer surface (dig-sex SPEC 8.5, #269). + /// + /// One meter for the whole NODE, not one per connection or per responder: the ceiling it enforces + /// is node-wide, and a per-connection meter would let a peer buy more allowance simply by opening + /// more connections. + peer_admission: Arc, /// A WEAK self-reference, installed by the standalone peer-network bring-up (which holds the /// `Arc`), so a `&self` read handler can spawn a detached background task that needs an owned /// `Arc` — the capsule backfill (§14.3). `Weak` (not `Arc`) so the node's refcount is @@ -4057,7 +4063,11 @@ impl Node { /// a per-request walk of the whole cache is a cost amplifier a peer controls. /// /// The batch is CAPPED at [`MAX_AVAILABILITY_ITEMS`] — the item count is caller-controlled — with - /// the excess simply not answered (the result array is aligned to the answered prefix). + /// the excess simply not answered (the result array is aligned to the answered prefix). That + /// truncation is the LAST line of defence, reached only by in-process callers: on the peer + /// surface an oversized batch never gets here, because + /// `NodeResponder::handle_availability` (peer.rs) meters the requested item + /// count against the same limit and refuses the whole request at the boundary (#269). /// /// `requestor` keys the per-item not-held → DHT `find_providers` enrichment against its /// per-requestor miss-lookup budget (dig_ecosystem#2007), so a large batch of not-held items from @@ -4562,6 +4572,11 @@ impl Node { self.capsule_acquisition.clone() } + /// The node-wide inbound admission meter (dig-sex SPEC 8.5, #269). + pub(crate) fn peer_admission(&self) -> &Arc { + &self.peer_admission + } + /// Build a node from the environment (cache dir/cap, §21 identity, upstream). /// Used by both the standalone bin's [`run`] and the in-process `dig-runtime`. pub fn from_env() -> Arc { @@ -4640,6 +4655,7 @@ impl Node { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), + peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -4953,6 +4969,7 @@ pub(crate) mod test_support { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), + peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -5742,6 +5759,7 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), + peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -5875,6 +5893,7 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), + peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -5942,6 +5961,7 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), + peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -6034,6 +6054,9 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), + peer_admission: Arc::new( + crate::seams::dig_peer::admission::PeerAdmission::default(), + ), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -6108,6 +6131,9 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), + peer_admission: Arc::new( + crate::seams::dig_peer::admission::PeerAdmission::default(), + ), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -8935,6 +8961,7 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), + peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -10715,6 +10742,10 @@ mod tests { assert_eq!(arr[2]["available"], false, "unknown capsule is a miss"); } + /// Pins the IN-PROCESS truncation only. A peer-surface batch this size is refused whole at + /// admission long before it reaches here, and this test cannot see that: it calls the batch + /// BELOW the responder that decides. The responder-level pair lives in `peer.rs` + /// (`the_responder_serves_a_batch_at_the_advertised_limit_and_refuses_one_past_it`). #[tokio::test] async fn availability_batch_caps_the_item_count() { let (node, _td) = test_node(None); diff --git a/crates/dig-node-core/src/peer.rs b/crates/dig-node-core/src/peer.rs index b80d0c88..629344ef 100644 --- a/crates/dig-node-core/src/peer.rs +++ b/crates/dig-node-core/src/peer.rs @@ -897,6 +897,11 @@ pub trait PeerRpcResponder: Send + Sync { /// by the SAME per-requestor miss-lookup budget as the single-item legs (dig_ecosystem#2007). A /// batch is the LARGEST amplification vector (up to `MAX_AVAILABILITY_ITEMS` lookups per request), /// so it must key by the ASKING peer, never one shared peer bucket every peer could exhaust. + /// + /// A batch LARGER than `MAX_AVAILABILITY_ITEMS` is refused whole at the admission boundary + /// (`-32000`, reason `request too large`) rather than answered as a truncated prefix: the node + /// admits exactly the batch size it advertises, and a caller that asked for more is told so + /// instead of being handed a short answer it must diff against its request to notice. async fn handle_availability(&self, items: Value, conn_key: &str) -> Value; /// Stream a `dig.fetchRange` response for `req` (the RangeRequest value) by writing framed @@ -1035,13 +1040,22 @@ async fn reconcile_and_flood( /// [`PeerRpcResponder::handle_availability`], a range fetch via [`PeerRpcResponder::stream_range`]. /// Returns when the peer closes the connection. The caller has already verified the remote `peer_id` /// (dig-nat enforces it during the mTLS handshake), so every stream here is from an authenticated peer. +/// +/// This entry point threads NO caller identity through to the responder, and since #269 that is +/// observable: the JSON-RPC and availability paths meter against the mTLS-verified `peer_id` and +/// refuse a session without one (`-32000`, reason `unauthenticated`). Callers that have the verified +/// identity — every production listener does — MUST use [`serve_peer_session_from`], which is what +/// keeps those two paths served. pub async fn serve_peer_session( mut session: dig_nat::mux::PeerSession, responder: Arc, ) { // No authenticated caller threaded here (the mTLS-verified caller is supplied by the listener via - // `serve_peer_session_from`); a caller-less session still serves the JSON-RPC/range/availability - // paths — only DHT routing-table population needs the caller. + // `serve_peer_session_from`). Since #269 a caller-less session serves the RANGE and module-range + // paths only: JSON-RPC and availability now meter against the mTLS-verified peer_id, and an + // absent one is refused `-32000` (`unauthenticated`) rather than admitted unmetered — otherwise + // presenting no identity would be the cheapest way out of the meter. DHT routing-table + // population likewise needs the caller. Every production listener supplies one. serve_peer_session_from(None, &mut session, responder).await } @@ -1226,6 +1240,30 @@ where } } +/// The JSON-RPC error body for inbound work refused at the admission boundary (#269). +/// +/// `-32000` (server error) rather than a method/param error: the request was well-formed and the node +/// simply declined to spend on it now. The reason names the LIMIT, never the peer's standing, so shed +/// load stays distinguishable from a ban by the peer and from an outage by the operator. +fn admission_refused( + id: Value, + conn_key: &str, + refusal: crate::seams::dig_peer::admission::AdmissionRefusal, +) -> Value { + // The refused PEER is named, because a refusal log that omits it cannot answer the only question + // an operator has when the node starts shedding: is one caller taking the allowance, or is the + // node simply loaded? A count without identities looks identical in both cases (gate S3 on + // dig-node#456). Truncated to the 16-hex prefix — enough to tell callers apart in a log, short + // enough not to turn every refusal into a full identity dump. + tracing::debug!( + reason = refusal.reason(), + peer = %conn_key.get(..16).unwrap_or(conn_key), + "peer serve: inbound work refused at admission" + ); + json!({"jsonrpc":"2.0","id":id, + "error":{"code":-32000,"message":"request refused","data":{"reason":refusal.reason()}}}) +} + /// Whether `method` may be answered over the **mTLS peer surface** (other DIG nodes). /// /// The allowlist itself lives in ONE place — [`dig_rpc_protocol::Method::is_peer_reachable`], @@ -1391,6 +1429,18 @@ impl PeerRpcResponder for NodeResponder { async fn handle_json_rpc(&self, req: Value, conn_key: &str) -> Value { let method = req.get("method").and_then(Value::as_str).unwrap_or(""); let id = req.get("id").cloned().unwrap_or(json!(1)); + // ADMISSION (dig-sex SPEC 8.5, #269) — ahead of the allowlist and every dispatch below, so a + // refused request costs a hex decode and a counter bump rather than a read, a decode or a DHT + // lookup. The guard is held for the whole method: it releases on `Drop`, including on the + // early `return`s below, which is why no path here needs to remember to. + let _admitted = match self + .node + .peer_admission() + .admit(conn_key, dig_sex::WorkKind::Own, 1) + { + Ok(guard) => guard, + Err(refusal) => return admission_refused(id, conn_key, refusal), + }; // PEER-SURFACE ALLOWLIST (audit #179 CRITICAL). The mTLS verifier accepts any self-signed // leaf, so an "authenticated" peer is merely "some peer_id", NOT an authorized admin. Route // ONLY the intended L7 read/discovery/announce methods to the shared dispatch; return -32601 @@ -1454,6 +1504,19 @@ impl PeerRpcResponder for NodeResponder { } async fn handle_availability(&self, items: Value, conn_key: &str) -> Value { + // ADMISSION (dig-sex SPEC 8.5, #269) — before the items are even read. `items.len()` is the + // attacker-chosen quantity this request asks for, so it is what gets clamped at the boundary + // (`AdmissionLimits::max_request_units`); clamping it deeper in would already have paid for it. + let requested_units = + u32::try_from(items.as_array().map_or(0, Vec::len)).unwrap_or(u32::MAX); + let _admitted = match self.node.peer_admission().admit( + conn_key, + dig_sex::WorkKind::Own, + requested_units, + ) { + Ok(guard) => guard, + Err(refusal) => return admission_refused(json!(1), conn_key, refusal), + }; let items = items.as_array().cloned().unwrap_or_default(); // The verified mTLS peer_id (`conn_key`) keys the per-requestor miss-lookup budget, identical // to the range-stream miss on this same peer surface (dig_ecosystem#2007). @@ -5410,8 +5473,17 @@ pub(crate) mod tests { // End-to-end over the responder: a peer JSON-RPC frame naming a management/mutation // method is answered with -32601 (method not found) WITHOUT ever reaching the // node's `handle_rpc` dispatch (which would run the mutation). getPeers still works. + // + // The conn_key is a well-formed 64-hex peer id — the shape every production session supplies, + // since both listeners derive it from the verified client leaf. It became load-bearing when + // admission landed (#269): the meter runs AHEAD of this allowlist and refuses a session with + // no verified identity, so an empty key would now answer -32000 and this test would stop + // exercising the allowlist at all. Using a real one keeps the property under test unchanged — + // an AUTHENTICATED peer is still merely "some peer_id", never an authorized admin, which is + // the whole point of audit #179. let (node, _td) = crate::test_support::test_node_for_peer_surface(); let responder = NodeResponder::without_pool(node); + let authenticated = "ab".repeat(32); for m in [ "cache.clear", "cache.setCapBytes", @@ -5420,7 +5492,7 @@ pub(crate) mod tests { "dig.stage", ] { let req = json!({"jsonrpc":"2.0","id":1,"method":m,"params":{}}); - let resp = responder.handle_json_rpc(req, "").await; + let resp = responder.handle_json_rpc(req, &authenticated).await; assert_eq!( resp["error"]["code"], json!(-32601), @@ -5435,7 +5507,7 @@ pub(crate) mod tests { let ok = responder .handle_json_rpc( json!({"jsonrpc":"2.0","id":1,"method":"dig.getNetworkInfo"}), - "", + &authenticated, ) .await; assert!( @@ -5444,11 +5516,97 @@ pub(crate) mod tests { ); // getPeers is answered from the (empty) pool view, not -32601. let peers = responder - .handle_json_rpc(json!({"jsonrpc":"2.0","id":1,"method":"dig.getPeers"}), "") + .handle_json_rpc( + json!({"jsonrpc":"2.0","id":1,"method":"dig.getPeers"}), + &authenticated, + ) .await; assert!(peers["result"]["peers"].is_array()); } + /// **Proves (#269):** the admission meter is LIVE on the responder — not merely implemented + /// beside it — and it refuses a session that carries no verified identity BEFORE the method + /// allowlist is consulted. + /// + /// This is the wiring assertion. `admission.rs`'s unit tests prove the meter behaves; they cannot + /// prove anything calls it, and a meter nothing calls is the exact defect this ticket describes. + /// The authenticated control is what makes the refusal meaningful: without it, a responder that + /// refused EVERY request would pass. + #[tokio::test] + async fn the_responder_refuses_an_unauthenticated_session_before_consulting_the_allowlist() { + let (node, _td) = crate::test_support::test_node_for_peer_surface(); + let responder = NodeResponder::without_pool(node); + let method = json!({"jsonrpc":"2.0","id":1,"method":"dig.getNetworkInfo"}); + + // No verified peer id: refused at admission, and NOT with the allowlist's -32601 — which is + // how we know the meter ran first rather than the request simply failing later. + let refused = responder.handle_json_rpc(method.clone(), "").await; + assert_eq!( + refused["error"]["code"], + json!(-32000), + "an unauthenticated peer session must be refused at admission" + ); + assert!( + refused.get("result").is_none(), + "a refused request must produce no result — the work must not have been done" + ); + + // CONTROL: the same method, from an authenticated session, is served. + let served = responder.handle_json_rpc(method, &"cd".repeat(32)).await; + assert!( + served.get("result").is_some(), + "an authenticated peer must still be served, or the meter is refusing everybody" + ); + } + /// **Proves (#269):** the admission clamp and the availability batch's own advertised limit are + /// the SAME number, measured through the responder that actually decides. + /// + /// The clamp lives in [`NodeResponder::handle_availability`], above + /// [`crate::Node::availability_batch`]. `lib.rs`'s cap test calls the batch directly, BELOW that + /// decision, so it passes identically whether the clamp admits this batch, refuses it, or is not + /// wired at all — a test below the decision passes under the defect. This one asks the responder. + /// + /// Both sides are pinned, because a bound tested only from below can only confirm itself: a batch + /// AT [`crate::MAX_AVAILABILITY_ITEMS`] must be ANSWERED (a clamp set lower than the advertised + /// limit would refuse work this node says it serves), and one item past it must be REFUSED at + /// admission (a clamp set higher would let the meter admit work the batch will not answer). + #[tokio::test] + async fn the_responder_serves_a_batch_at_the_advertised_limit_and_refuses_one_past_it() { + let (node, _td) = crate::test_support::test_node_for_peer_surface(); + let responder = NodeResponder::without_pool(node); + let authenticated = "ef".repeat(32); + let batch = |n: usize| -> Value { + (0..n) + .map(|_| json!({ "store_id": "ee".repeat(32) })) + .collect::>() + .into() + }; + + let at_bound = responder + .handle_availability(batch(crate::MAX_AVAILABILITY_ITEMS), &authenticated) + .await; + assert_eq!( + at_bound["items"].as_array().map(Vec::len), + Some(crate::MAX_AVAILABILITY_ITEMS), + "a batch at the advertised limit must be ANSWERED in full — the node states it serves \ + this many items, so refusing it denies work of its own contract" + ); + + let past_bound = responder + .handle_availability(batch(crate::MAX_AVAILABILITY_ITEMS + 1), &authenticated) + .await; + assert_eq!( + past_bound["error"]["data"]["reason"], + json!("request too large"), + "one item past the advertised limit must be refused AT admission, before the batch is \ + read — not silently truncated after the cost is committed" + ); + assert!( + past_bound.get("items").is_none(), + "a refused batch must produce no answers at all" + ); + } + // -- OUTGOING-BANDWIDTH THROTTLE on the peer range-stream (dig_ecosystem #30) -------------------- // // `stream_range` is the busiest node-to-node egress path (multi-source downloaders fan ranges diff --git a/crates/dig-node-core/src/seams/dig_peer/admission.rs b/crates/dig-node-core/src/seams/dig_peer/admission.rs new file mode 100644 index 00000000..2d18c508 --- /dev/null +++ b/crates/dig-node-core/src/seams/dig_peer/admission.rs @@ -0,0 +1,696 @@ +//! Inbound admission for the mTLS peer surface (dig-sex SPEC §8.5, dig-node#269). +//! +//! `dig_sex::admission` decides whether inbound work is admitted; this module is the node's half — +//! it derives the authenticated identity to meter against, holds the meter, and makes the paired +//! release structural. +//! +//! # Admit BEFORE the work, not after +//! +//! The value of an admission meter is that it refuses before spending anything. A check placed after +//! the read/decode/fetch has already paid the cost it exists to avoid, so [`PeerAdmission::admit`] is +//! called at the top of each responder method, ahead of every dispatch. +//! +//! # Metered by the AUTHENTICATED identity, never by anything the caller chooses +//! +//! The meter key is the mTLS-verified `peer_id` the session derived — never a wire field, never a +//! connection counter, never a constant. A key a caller can choose freely turns a per-peer limit into +//! a limit on whichever bucket the caller picks, and a CONSTANT key collapses every requestor into one +//! shared bucket, so a single peer exhausts the allowance for everybody. That failure looks exactly +//! like working DoS protection until the moment it matters, which is why +//! [`dig_sex::AuthenticatedPeer`] is a newtype and why this module refuses rather than substituting a +//! placeholder when no verified identity exists. +//! +//! # An unauthenticated peer-surface request is REFUSED, not admitted unmetered +//! +//! In production every session reaching the responder carries a verified `peer_id` (the peer surface is +//! mTLS-only), so an absent one means the session carried no authenticated caller at all. Admitting it +//! unmetered would make "present no identity" the cheapest way to escape the meter — the guard would +//! be optional at the attacker's discretion. It is therefore refused as [`Refusal::MeterFull`]'s +//! sibling case, [`AdmissionRefusal::Unauthenticated`]. Nothing on the loopback admin / in-process FFI +//! path routes through here (that is `crate::handle_rpc`), so refusing costs no legitimate caller. +//! +//! # Release is structural, not remembered +//! +//! [`AdmissionGuard`] releases on `Drop`. A `release` skipped on an error path leaks allowance until +//! the node refuses everything, and an error path is precisely the one a hand-written release is +//! forgotten on — so the guard makes the omission unrepresentable rather than reviewable. + +use std::sync::Mutex; + +use dig_sex::{AdmissionLimits, AdmissionMeter, AuthenticatedPeer, Refusal, WorkKind}; + +use super::dht::hex64; + +/// Why inbound work was refused at the boundary. +/// +/// Wraps the crate's [`Refusal`] so the node can name the one case the crate cannot: a request with no +/// authenticated identity to meter against. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AdmissionRefusal { + /// The session carried no mTLS-verified `peer_id`, so there is no identity to meter against. + Unauthenticated, + /// The crate's meter refused; the variant says which limit was reached. + Limited(Refusal), +} + +impl AdmissionRefusal { + /// A short, stable reason string for the JSON-RPC error body and the serve log. + /// + /// Deliberately names the LIMIT rather than the peer's standing: shed load must be + /// distinguishable from an outage by an operator reading a log, and from a ban by the peer + /// reading the response. + #[must_use] + pub const fn reason(self) -> &'static str { + match self { + AdmissionRefusal::Unauthenticated => "unauthenticated", + AdmissionRefusal::Limited(Refusal::GlobalCeiling) => "node at capacity", + AdmissionRefusal::Limited(Refusal::PeerShare) => "peer at capacity", + AdmissionRefusal::Limited(Refusal::RelayBudget) => "relay budget exhausted", + AdmissionRefusal::Limited(Refusal::MeterFull) => "meter full", + AdmissionRefusal::Limited(Refusal::RequestTooLarge) => "request too large", + } + } +} + +/// The authenticated identity to meter against, derived from the session's verified `peer_id`. +/// +/// `conn_key` is the mTLS-verified peer id as lowercase 64-hex, empty on a caller-less session. Only a +/// well-formed 64-hex value yields an identity: anything else is not a verified peer id, and coercing +/// it into one would meter distinct callers into whatever bucket the malformed value happened to hash +/// to. +#[must_use] +pub fn authenticated_peer(conn_key: &str) -> Option { + hex64(conn_key).map(AuthenticatedPeer::from_verified_session) +} + +/// How many peers are each guaranteed ONE concurrent slot, regardless of node-wide load. +/// +/// Equal to [`crate::peer::MAX_INFLIGHT_PEER_CONNECTIONS`], and derived from it rather than restated, +/// because the property it buys is a comparison against that number: a peer that holds a connection +/// holds a reserved slot to go with it, so **admission can never be the scarcer resource**. +/// +/// # Why a reserve exists at all (gate G1 on dig-node#456) +/// +/// [`AdmissionMeter::admit`] tests the node-wide ceiling BEFORE the per-peer share, and the node-wide +/// counter is shared by every peer. With a single pool, the identities needed to deny the whole +/// network is `global_ceiling / per_peer_share` — at the dig-sex defaults, **eight**. A `peer_id` is +/// SHA-256 of a self-signed TLS SPKI, so eight identities cost eight keypairs, each one staying +/// INSIDE its own share so the per-peer limiter never fires. That is 8:1 amplification: eight free +/// identities silence the whole discovery, availability and content-read surface of the node. +/// +/// Raising the ceiling only raises that price; it does not change the shape. What changes the shape is +/// spending the FIRST concurrent unit of each peer from a pool whose per-peer share is exactly one. +/// Denying an honest peer then costs one identity AND one held connection per slot — linear, 1:1, and +/// bounded by the connection cap the node already enforces rather than by a number 8x below it. +/// +/// The reserve is not extra allowance: the total concurrent share of a peer is unchanged, only the +/// pool its first unit is drawn from. See [`PeerAdmission::with_reserved_first_slots`]. +pub const RESERVED_FIRST_SLOTS: u32 = crate::peer::MAX_INFLIGHT_PEER_CONNECTIONS as u32; + +/// Which pool the slot of a guard came from, so `Drop` returns it to the meter that issued it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Tier { + /// The first concurrent unit of a peer, from the per-peer-guaranteed reserve. + Reserved, + /// Any further concurrent unit, from the shared node-wide pool. + Burst, +} + +/// One admitted unit of inbound work. Releases its slot on `Drop`, on every exit path. +/// +/// Held by value across the work it admits; the borrow checker then makes "return before releasing" +/// impossible rather than merely discouraged. +#[derive(Debug)] +pub struct AdmissionGuard<'a> { + admission: &'a PeerAdmission, + peer: AuthenticatedPeer, + kind: WorkKind, + tier: Tier, +} + +impl Drop for AdmissionGuard<'_> { + fn drop(&mut self) { + // Returned to the pool it was TAKEN from. Releasing into the other one would credit allowance + // that was never spent there and leak the pool that was — the same permanent leak a forgotten + // release causes, only harder to see. + let meter = match self.tier { + Tier::Reserved => &self.admission.reserved, + Tier::Burst => &self.admission.burst, + }; + // A poisoned meter is recovered rather than propagated: refusing to RELEASE on the panic path + // would leak the very allowance this guard exists to return, permanently. + let mut meter = meter + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + meter.release(self.peer, self.kind); + } +} + +/// The inbound admission meter of the node, shared across every peer session. +/// +/// Node-wide rather than per-connection, because a per-connection meter would let a peer buy more +/// allowance by opening more connections — the same collapse as a caller-chosen key wearing a +/// different shape. +/// +/// Two pools, not one. See [`RESERVED_FIRST_SLOTS`] for why the first concurrent unit of each peer is +/// metered separately from the rest. +#[derive(Debug)] +pub struct PeerAdmission { + /// The FIRST concurrent unit of own work of every peer. `per_peer_share` is 1 here, so one + /// identity takes exactly one slot and the denial cost is linear in identities. + reserved: Mutex, + /// Everything beyond the first concurrent unit of a peer, plus all relayed work. Shared + /// node-wide, and therefore the pool a busy node sheds from first — which is what shedding load + /// should mean. + burst: Mutex, +} + +/// The limits this node admits under. +/// +/// The dig-sex defaults are used for every dimension EXCEPT `max_request_units`, which is raised +/// to [`crate::MAX_AVAILABILITY_ITEMS`] — the largest batch this node advertises it answers. The +/// crate default of 256 is half that, and a clamp set below the advertised limit refuses work the +/// contract of the node says it serves: a 257-512 item `dig.getAvailability` batch would have been +/// answered `-32000 "request too large"` while [`crate::Node::availability_batch`] stood ready to +/// answer all 512. The clamp and the limit it clamps to must be the SAME number, so this derives one +/// from the other rather than restating it. +/// +/// These are the BURST pool limits. They no longer describe the whole surface a peer can reach: +/// [`RESERVED_FIRST_SLOTS`] peers hold a guaranteed slot outside them. +/// +/// # `relay_ceiling` is VACUOUS on this node today (gate S4 on dig-node#456) +/// +/// Nothing in this crate constructs [`WorkKind::Relayed`] — every production `admit` call site passes +/// [`WorkKind::Own`] (`crate::peer::NodeResponder`). The separate relay budget of dig-sex SPEC 6.1.8 +/// is therefore configured and never consulted: it is satisfied because the case it governs never +/// occurs, not because it is enforced. Stated here rather than left to read as an active rule, since +/// a limit nobody reaches and a limit nobody applies are indistinguishable from the number alone. The +/// ceiling is kept, not removed, so that the first producer of relayed work inherits a budget instead +/// of an omission. +#[must_use] +pub fn node_limits() -> AdmissionLimits { + AdmissionLimits { + max_request_units: crate::MAX_AVAILABILITY_ITEMS as u32, + ..AdmissionLimits::default() + } +} + +impl Default for PeerAdmission { + fn default() -> Self { + Self::new(node_limits()) + } +} + +impl PeerAdmission { + /// A meter with no work in flight, admitting under `limits` with the production reserve. + #[must_use] + pub fn new(limits: AdmissionLimits) -> Self { + Self::with_reserved_first_slots(limits, RESERVED_FIRST_SLOTS) + } + + /// As [`PeerAdmission::new`], with the reserve sized explicitly so a test can exhaust it. + /// + /// `limits.per_peer_share` remains the TOTAL concurrent share of a peer: the first of those units + /// is drawn from the reserve and the remainder from the burst pool, so the reserve grants no peer + /// any extra concurrency — it only decides which pool the first unit is charged to. + #[must_use] + pub fn with_reserved_first_slots(limits: AdmissionLimits, reserved_first_slots: u32) -> Self { + Self { + reserved: Mutex::new(AdmissionMeter::new(AdmissionLimits { + global_ceiling: reserved_first_slots, + per_peer_share: 1, + // Relayed work never reaches this pool (see `admit`), so a ceiling here would be + // vacuous; zero states that rather than implying a budget nothing consults. + relay_ceiling: 0, + max_tracked_peers: limits.max_tracked_peers, + max_request_units: limits.max_request_units, + })), + burst: Mutex::new(AdmissionMeter::new(AdmissionLimits { + per_peer_share: limits.per_peer_share.saturating_sub(1), + ..limits + })), + } + } + + /// Admit one unit of inbound work for the session identified by `conn_key`, BEFORE performing it. + /// + /// `requested_units` is the attacker-chosen quantity the request asks for, clamped here at the + /// boundary rather than deeper in where the cost would already be committed. + /// + /// Own work tries the reserve first and falls back to the burst pool; relayed work goes straight + /// to the burst pool, where the separate relay ceiling of SPEC 6.1.8 applies unchanged. Work done + /// on behalf of another node is exactly the work a loaded node should shed, so it is deliberately + /// not given a guaranteed slot. + /// + /// # Errors + /// + /// [`AdmissionRefusal::Unauthenticated`] when the session carried no verified peer id, or + /// [`AdmissionRefusal::Limited`] when a limit is reached. In both cases NO work has been done. + pub fn admit( + &self, + conn_key: &str, + kind: WorkKind, + requested_units: u32, + ) -> Result, AdmissionRefusal> { + let peer = authenticated_peer(conn_key).ok_or(AdmissionRefusal::Unauthenticated)?; + if kind == WorkKind::Own { + match Self::charge(&self.reserved, peer, kind, requested_units) { + Ok(()) => return Ok(self.guard(peer, kind, Tier::Reserved)), + // Both pools clamp on the SAME `max_request_units`, so retrying the burst pool could + // only reach the identical refusal one lock later. Returning it here keeps the answer + // to an over-sized request independent of how loaded the node happens to be. + Err(Refusal::RequestTooLarge) => { + return Err(AdmissionRefusal::Limited(Refusal::RequestTooLarge)); + } + // PeerShare (this peer already holds its reserved slot) or GlobalCeiling (every + // reserved slot is held): both mean "not from the reserve", never "refuse". + Err(_) => {} + } + } + Self::charge(&self.burst, peer, kind, requested_units) + .map_err(AdmissionRefusal::Limited)?; + Ok(self.guard(peer, kind, Tier::Burst)) + } + + /// Take one unit from `meter`, holding its lock for no longer than the accounting. + fn charge( + meter: &Mutex, + peer: AuthenticatedPeer, + kind: WorkKind, + requested_units: u32, + ) -> Result<(), Refusal> { + meter + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .admit(peer, kind, requested_units) + } + + fn guard(&self, peer: AuthenticatedPeer, kind: WorkKind, tier: Tier) -> AdmissionGuard<'_> { + AdmissionGuard { + admission: self, + peer, + kind, + tier, + } + } + + /// Units of work currently in flight node-wide, across both pools. + /// + /// Used by the tests of this module. It is deliberately NOT described as an operator surface: + /// nothing renders it today, and a doc-comment promising a status readout that does not exist is + /// the kind of claim that gets believed (gate S3 on dig-node#456). + #[must_use] + pub fn in_flight_total(&self) -> u32 { + Self::pool_in_flight(&self.reserved) + Self::pool_in_flight(&self.burst) + } + + /// Units in flight in the reserve pool — i.e. how many DISTINCT peers currently hold their + /// guaranteed slot, since the per-peer share of that pool is one. + #[must_use] + pub fn reserved_in_flight(&self) -> u32 { + Self::pool_in_flight(&self.reserved) + } + + fn pool_in_flight(meter: &Mutex) -> u32 { + meter + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .in_flight_total() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A 64-hex conn_key that is distinct per `n` — the shape a real mTLS session yields. + fn conn_key(n: u8) -> String { + hex::encode([n; 32]) + } + + /// Limits with a per-peer share small enough to exhaust in a test, and a global ceiling well + /// ABOVE it — so exhausting one peer's share can only be the PER-PEER limit, never the node-wide + /// one. A test whose two ceilings are equal cannot tell those apart and would pass on a meter that + /// had collapsed every peer into one bucket, which is the exact defect this file exists to prevent. + fn limits() -> AdmissionLimits { + AdmissionLimits { + global_ceiling: 64, + per_peer_share: 2, + relay_ceiling: 16, + max_tracked_peers: 1024, + max_request_units: 8, + } + } + + /// **Proves (#269):** the meter is keyed on the AUTHENTICATED identity, so exhausting one peer's + /// share refuses that peer and leaves a second peer entirely unaffected. + /// + /// The second peer is the load-bearing half. Asserting only that an over-quota peer is refused + /// passes identically on a meter keyed by a CONSTANT — the shared-bucket mistake — because that + /// meter also refuses at the limit. It just refuses everybody. Varying one actor while keeping an + /// honest control is what distinguishes a per-peer limit from a global one wearing its clothes. + #[test] + fn exhausting_one_peers_share_does_not_refuse_a_different_peer() { + let admission = PeerAdmission::new(limits()); + let noisy = conn_key(0xaa); + let quiet = conn_key(0xbb); + + let _first = admission + .admit(&noisy, WorkKind::Own, 1) + .expect("first unit is within the share"); + let _second = admission + .admit(&noisy, WorkKind::Own, 1) + .expect("second unit is within the share"); + + assert_eq!( + admission.admit(&noisy, WorkKind::Own, 1).unwrap_err(), + AdmissionRefusal::Limited(Refusal::PeerShare), + "a third unit exceeds this peer's share and must be refused" + ); + assert!( + admission.admit(&quiet, WorkKind::Own, 1).is_ok(), + "a DIFFERENT peer must be unaffected — if it is refused, every requestor shares one \ + bucket and a single peer can exhaust the allowance for everybody" + ); + } + + /// **Proves (#269):** the guard releases on `Drop`, so allowance returns on every exit path. + /// + /// A `release` skipped on an error path leaks allowance until the node refuses everything, and the + /// error path is the one a hand-written release is forgotten on. Dropping the guard here stands in + /// for every such path, because `Drop` cannot distinguish them. + #[test] + fn a_dropped_guard_returns_its_allowance() { + let admission = PeerAdmission::new(limits()); + let peer = conn_key(0xcc); + + { + let _a = admission.admit(&peer, WorkKind::Own, 1).expect("first"); + let _b = admission.admit(&peer, WorkKind::Own, 1).expect("second"); + assert_eq!(admission.in_flight_total(), 2); + assert!( + admission.admit(&peer, WorkKind::Own, 1).is_err(), + "the share must be exhausted while the guards are held, or the release below \ + proves nothing" + ); + } + + assert_eq!( + admission.in_flight_total(), + 0, + "both guards went out of scope; neither slot may still be held" + ); + assert!( + admission.admit(&peer, WorkKind::Own, 1).is_ok(), + "the same peer must be admissible again once its work finished" + ); + } + + /// **Proves (#269):** a session with no verified peer id is REFUSED, not admitted unmetered. + /// + /// Admitting it would make "present no identity" the cheapest way to escape the meter, and the + /// alternative mistake — substituting a placeholder identity — is the shared bucket again. Both + /// non-64-hex shapes are covered: absent, and present-but-malformed. + #[test] + fn an_unauthenticated_session_is_refused_rather_than_metered_under_a_placeholder() { + let admission = PeerAdmission::new(limits()); + for bad in ["", "not-hex", &"ab".repeat(31), &"zz".repeat(32)] { + assert_eq!( + admission.admit(bad, WorkKind::Own, 1).unwrap_err(), + AdmissionRefusal::Unauthenticated, + "{bad:?} is not a verified peer id and must yield no admission" + ); + } + assert_eq!( + admission.in_flight_total(), + 0, + "a refused request must consume no allowance" + ); + } + + /// **Proves (#269):** the attacker-chosen request size is clamped AT the boundary. + /// + /// `max_request_units` is 8 here, so 9 is over and 8 is at the bound. Pinning both sides matters: + /// a bound tested only from below can only confirm itself, and a clamp that refused everything + /// would pass a one-sided test while denying all legitimate work. + #[test] + fn an_oversized_request_is_refused_and_the_at_bound_request_is_admitted() { + let admission = PeerAdmission::new(limits()); + let peer = conn_key(0xdd); + + assert_eq!( + admission.admit(&peer, WorkKind::Own, 9).unwrap_err(), + AdmissionRefusal::Limited(Refusal::RequestTooLarge), + "one unit over max_request_units must be refused" + ); + assert!( + admission.admit(&peer, WorkKind::Own, 8).is_ok(), + "the at-bound request must still be admitted, or the clamp denies legitimate work" + ); + } + + /// **Proves (#269):** the node's admitted request size is the availability batch's OWN advertised + /// limit, not the crate's smaller default. + /// + /// Asserting a literal 512 here would pass on a hard-coded constant that had drifted from + /// `MAX_AVAILABILITY_ITEMS`; asserting the derivation is what keeps the two numbers one number. + #[test] + fn the_node_admits_a_request_as_large_as_the_availability_batch_it_advertises() { + assert_eq!( + node_limits().max_request_units, + crate::MAX_AVAILABILITY_ITEMS as u32, + "the admission clamp must equal the advertised batch limit, or the node refuses batches it says it serves" + ); + assert!( + node_limits().max_request_units > AdmissionLimits::default().max_request_units, + "the crate default is the smaller of the two — if this ever stops holding, the override is doing nothing and the comment above it is false" + ); + } + + /// **Proves (#269):** relayed work draws on its own ceiling, so work done on other nodes' behalf + /// cannot consume the whole node-wide allowance (SPEC 6.1.8). + /// + /// **This test is the ONLY producer of [`WorkKind::Relayed`] in the crate** (gate S4 on #456). It + /// proves the meter would enforce the budget; it does not show the budget being enforced in + /// production, because no production call site asks for relayed work. See [`node_limits`]. + #[test] + fn relayed_work_exhausts_the_relay_ceiling_while_own_work_still_admits() { + let admission = PeerAdmission::new(AdmissionLimits { + global_ceiling: 64, + per_peer_share: 8, + relay_ceiling: 1, + max_tracked_peers: 1024, + max_request_units: 8, + }); + let peer = conn_key(0xee); + + let _relayed = admission + .admit(&peer, WorkKind::Relayed, 1) + .expect("the first relayed unit is within the relay ceiling"); + assert_eq!( + admission.admit(&peer, WorkKind::Relayed, 1).unwrap_err(), + AdmissionRefusal::Limited(Refusal::RelayBudget), + "the relay ceiling is 1 and must be reached" + ); + assert!( + admission.admit(&peer, WorkKind::Own, 1).is_ok(), + "OWN work draws on a separate budget — a spent relay allowance must not stop this node \ + serving its own callers" + ); + } + + /// Hold as many concurrent units as `peer` can take, returning the guards so they stay held. + /// + /// Loops until the meter refuses rather than counting to a literal, so it measures the allowance + /// the node actually grants instead of restating the number the test hoped for. + fn hold_until_refused<'a>(admission: &'a PeerAdmission, peer: &str) -> Vec> { + let mut held = Vec::new(); + while let Ok(guard) = admission.admit(peer, WorkKind::Own, 1) { + held.push(guard); + assert!(held.len() < 1024, "a peer must not hold unbounded work"); + } + held + } + + /// **Proves (gate G1 on #456):** under the SHIPPED limits, eight free identities holding their + /// full share each cannot refuse an honest peer that holds nothing. + /// + /// This is the exploit the security gate executed against the previous single-pool meter, run + /// here against the configuration the node really ships (`PeerAdmission::default()`), not against + /// hand-picked limits. Before the reserve it failed: 8 identities x `per_peer_share` = 64 = + /// `global_ceiling`, every unit inside its own share so the per-peer limiter never fired, and the + /// ninth peer was answered `-32000 "node at capacity"`. + /// + /// The honest ninth peer is the load-bearing half. Asserting only that the sybils were eventually + /// refused would pass identically on the defective meter, because that meter also refuses at the + /// limit — it just refuses everybody. + #[test] + fn eight_sybil_identities_cannot_deny_an_honest_peer_under_the_shipped_limits() { + let admission = PeerAdmission::default(); + let sybils: Vec = (0..8u8).map(conn_key).collect(); + + let _held: Vec<_> = sybils + .iter() + .map(|peer| hold_until_refused(&admission, peer)) + .collect(); + + let honest = conn_key(0xf0); + assert!( + admission.admit(&honest, WorkKind::Own, 1).is_ok(), + "a peer holding ZERO work was refused while 8 free identities held theirs — the node-wide \ + pool is deniable at a Sybil cost of eight keypairs" + ); + } + + /// **Proves (gate G1 on #456):** the same holds when the sybils spend everything they can, not + /// merely eight of them. + /// + /// The single-pool meter needed `global_ceiling / per_peer_share` identities. This walks distinct + /// identities until an honest newcomer is finally refused and asserts that the count reached is + /// bounded BELOW by the reserve — i.e. denial is linear in identities held, not amplified by the + /// per-peer share. + #[test] + fn denying_a_newcomer_costs_at_least_one_identity_per_reserved_slot() { + // A small reserve so the walk terminates quickly. The single-pool meter would have needed + // global_ceiling / per_peer_share = 4 / 2 = 2 identities; the reserve raises the floor. + let limits = AdmissionLimits { + global_ceiling: 4, + per_peer_share: 2, + relay_ceiling: 16, + max_tracked_peers: 1024, + max_request_units: 8, + }; + let reserved_first_slots = 3; + let admission = PeerAdmission::with_reserved_first_slots(limits, reserved_first_slots); + + let mut held = Vec::new(); + let mut identities = 0u32; + loop { + let peer = conn_key(u8::try_from(identities).expect("fewer than 256 identities")); + let guards = hold_until_refused(&admission, &peer); + if guards.is_empty() { + break; + } + identities += 1; + held.push(guards); + assert!( + identities < 64, + "the meter must refuse a newcomer eventually" + ); + } + + assert!( + identities > limits.global_ceiling / limits.per_peer_share, + "denial took {identities} identities; a single shared pool needs only {} — the reserve is \ + not raising the floor", + limits.global_ceiling / limits.per_peer_share + ); + assert!( + identities >= reserved_first_slots, + "every reserved slot must be individually occupied before a newcomer can be refused; \ + {identities} identities is fewer than the {reserved_first_slots} reserved slots" + ); + } + + /// **Proves (gate G1 on #456):** a peer holding work is shed BEFORE a peer holding none. + /// + /// This is the property the reserve exists to establish, stated directly and separately from the + /// Sybil count: once the shared pool is spent, a busy peer is refused while a quiet one is served. + /// A test that only checked the Sybil count would pass on a meter that had merely raised its + /// ceiling, which is a price change rather than a shape change. + #[test] + fn a_spent_shared_pool_sheds_a_busy_peer_while_a_quiet_one_is_still_served() { + // A shared ceiling two busy peers can spend entirely, and a reserve with room left over — + // so the two outcomes below are distinguishable. With one pool they are not: the busy peer + // and the quiet one receive the identical GlobalCeiling refusal. + let limits = AdmissionLimits { + global_ceiling: 8, + per_peer_share: 5, + relay_ceiling: 16, + max_tracked_peers: 1024, + max_request_units: 8, + }; + let admission = PeerAdmission::with_reserved_first_slots(limits, 8); + let busy = conn_key(0x11); + let also_busy = conn_key(0x12); + + let _held = ( + hold_until_refused(&admission, &busy), + hold_until_refused(&admission, &also_busy), + ); + + assert!( + admission.admit(&busy, WorkKind::Own, 1).is_err(), + "a peer that already holds its share must be refused once the shared pool is spent" + ); + assert!( + admission.admit(&conn_key(0xfe), WorkKind::Own, 1).is_ok(), + "a peer holding NOTHING must still be served from the reserve — shedding must fall on \ + the peers holding work, not on whoever arrives next" + ); + } + + /// **Proves (gate G2 on #456):** the SHIPPED configuration is pinned, every dimension of it. + /// + /// The gate set `global_ceiling: 1` — a node that serves nobody — and the full 1061-test suite + /// stayed green, because `node_limits()` was asserted for one of its five dimensions and every + /// other meter test supplied hand-picked limits. A configuration nothing measures is how G1 + /// landed green in the first place, so all five are pinned here and the reserve with them. + #[test] + fn the_shipped_admission_configuration_is_pinned() { + let shipped = node_limits(); + assert_eq!(shipped.global_ceiling, 64, "burst pool node-wide ceiling"); + assert_eq!(shipped.per_peer_share, 8, "total concurrent share per peer"); + assert_eq!(shipped.relay_ceiling, 16, "SPEC 6.1.8 relay budget"); + assert_eq!(shipped.max_tracked_peers, 1024, "meter table size"); + assert_eq!( + shipped.max_request_units, + crate::MAX_AVAILABILITY_ITEMS as u32, + "the admission clamp must equal the advertised batch limit" + ); + } + + /// **Proves (gate G1/G2 on #456):** the reserve is at least as large as the connection cap, so + /// admission is never the scarcer of the two resources. + /// + /// Asserted as the RELATION rather than as the literal 512: a literal would pass on a reserve that + /// had drifted away from `MAX_INFLIGHT_PEER_CONNECTIONS`, which is the drift the derivation exists + /// to prevent. The comparison against the burst ceiling is the one that would have failed before + /// this change, when the whole surface was 64 slots wide. + #[test] + fn the_reserve_is_never_scarcer_than_the_connections_it_serves() { + assert_eq!( + RESERVED_FIRST_SLOTS, + crate::peer::MAX_INFLIGHT_PEER_CONNECTIONS as u32, + "every peer that can hold a connection must hold a guaranteed slot with it" + ); + assert!( + RESERVED_FIRST_SLOTS > node_limits().global_ceiling, + "a reserve no larger than the shared ceiling reserves nothing: the shared pool would run \ + out first and the node would be deniable at global_ceiling / per_peer_share identities" + ); + } + + /// **Proves (gate G1 on #456):** the reserve grants no peer EXTRA concurrency. + /// + /// The two-pool split must not become a share increase by accident — that would silently double + /// the work one peer can pin. The total a single peer can hold is still `per_peer_share`, with the + /// first unit charged to the reserve and the rest to the burst pool. + #[test] + fn the_reserve_does_not_widen_the_share_of_any_single_peer() { + let admission = PeerAdmission::new(limits()); + let peer = conn_key(0x77); + + let held = hold_until_refused(&admission, &peer); + assert_eq!( + u32::try_from(held.len()).expect("small"), + limits().per_peer_share, + "one peer must still hold exactly per_peer_share units in total across both pools" + ); + assert_eq!( + admission.reserved_in_flight(), + 1, + "exactly one of those units is the reserved slot" + ); + } +} diff --git a/crates/dig-node-core/src/seams/dig_peer/ask_routing.rs b/crates/dig-node-core/src/seams/dig_peer/ask_routing.rs index 15717bee..628c903e 100644 --- a/crates/dig-node-core/src/seams/dig_peer/ask_routing.rs +++ b/crates/dig-node-core/src/seams/dig_peer/ask_routing.rs @@ -69,7 +69,7 @@ const TICK: Duration = Duration::from_secs(1); /// /// See the module docs: this type existing at all is the security boundary, because it is the only /// way a `dig-sex` routing key can be minted and it can only be minted from a verified identity. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub(crate) struct RoutedPeer([u8; 32]); impl RoutedPeer { diff --git a/crates/dig-node-core/src/seams/dig_peer/conduct.rs b/crates/dig-node-core/src/seams/dig_peer/conduct.rs new file mode 100644 index 00000000..c5c579cc --- /dev/null +++ b/crates/dig-node-core/src/seams/dig_peer/conduct.rs @@ -0,0 +1,320 @@ +//! Peer conduct, and the dial share it earns (dig-sex SPEC §8.2A, dig-node#268). +//! +//! `dig_sex::conduct` classifies what this node observed a peer do and answers how much of the dial +//! budget that peer has earned. It was implemented, tested, and received not one observation from +//! dig-node. This module is the node's half: it holds the per-peer records, feeds them the outcomes +//! the ask loop already classifies, and exposes [`ConductState::dial_share`] so the ranking actually +//! spends fewer dials on peers that do not answer. +//! +//! # Why the classes are kept apart, and why that is a security property +//! +//! A **proven lie** (bytes that fail verification against the chain-anchored root) and a +//! **self-contradiction** (a peer that claimed to hold content, then answered absence) are both +//! *verifiable*: arithmetic and the peer's own words, checkable without trusting anyone. They carry a +//! durable penalty. +//! +//! **Non-performance** — a timeout, a reset, silence, a truncation — is *not* verifiable. It is +//! indistinguishable from genuine distress, and an attacker can manufacture it in an honest third +//! party by loading it up. So it decays on elapsed time alone, is capped, and MUST NOT reduce a dial +//! share to zero: if it could, inducing distress would become a way to evict an honest holder from +//! everybody's routing table. The crate enforces all three of those; this module's job is not to +//! undo them by feeding the wrong class. +//! +//! # Reputation is node-LOCAL and is never gossiped +//! +//! Nothing here is advertised, exchanged, or written to a peer-visible surface. A shared reputation +//! channel is a defamation primitive — one node's claim that a peer lied becomes every node's belief, +//! with no way to check it — so a record earned here influences only this node's own dialling. +//! +//! # Bounds: the pool is the liveness gate, so there is no TTL +//! +//! [`ConductState::retain`] is called with the CURRENT pool before every read, exactly as +//! [`AskRoutingState`](super::ask_routing) does for its observations. The map is therefore keyed only +//! by peers this node holds a verified session to and cannot grow while the pool does not — an +//! attacker cannot inflate it by minting identities it never connects with. + +use std::collections::HashMap; +use std::sync::Mutex; + +use dig_sex::{ConductEvidence, ConductRecord}; + +use super::ask_routing::RoutedPeer; + +/// This node's memory of how its pool peers have behaved. +/// +/// Keyed by [`RoutedPeer`] — the mTLS-verified `peer_id`, the same identity the ask router ranks on — +/// so a peer's conduct cannot be attributed to, or escaped by, an identity it chose for itself. +#[derive(Debug, Default)] +pub(crate) struct ConductState { + records: Mutex>, +} + +impl ConductState { + /// An empty conduct memory. + pub(crate) fn new() -> Self { + Self::default() + } + + fn lock(&self) -> std::sync::MutexGuard<'_, HashMap> { + self.records + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + /// Fold one observation about `peer` into its record. + /// + /// An unseen peer starts from [`ConductRecord::neutral`], never from a penalised one: a peer this + /// node has no history with must be indistinguishable from one that has behaved, or a fresh + /// identity would be *better* than an honest long-lived one and the ranking would reward churn. + pub(crate) fn observe(&self, peer: RoutedPeer, evidence: ConductEvidence, now_ticks: u64) { + let mut records = self.lock(); + let record = records + .get(&peer) + .copied() + .unwrap_or_else(ConductRecord::neutral); + records.insert(peer, dig_sex::observe(record, evidence, now_ticks)); + } + + /// The share of the dial budget `peer` has earned, in `0.0..=1.0`. + /// + /// Decay is applied at READ time rather than on a timer, because the crate's decay is a pure + /// function of elapsed ticks — so a peer whose non-performance has aged out recovers **without + /// this node having to talk to it**. That direction is the point: a peer punished for silence + /// that could only be forgiven by answering could never recover, because it is not being asked. + pub(crate) fn dial_share(&self, peer: RoutedPeer, now_ticks: u64) -> f64 { + let record = self + .lock() + .get(&peer) + .copied() + .unwrap_or_else(ConductRecord::neutral); + dig_sex::dial_share(dig_sex::decay(record, now_ticks)) + } + + /// The subset of `pool` this node will still spend a dial on, worst conduct excluded. + /// + /// The threshold is "a share above zero", and it is not an arbitrary cut: `dig_sex::dial_share` + /// returns exactly `0.0` for a peer with a PROVEN fault and floors every non-performance penalty + /// at [`MIN_NON_PERFORMANCE_DIAL_SHARE`](dig_sex::MIN_NON_PERFORMANCE_DIAL_SHARE). So this + /// excludes precisely the peers that lied or contradicted themselves — SPEC 8.3's durable + /// exclusion, earned by verifiable evidence — and can never exclude a peer that was merely slow, + /// however slow it was. Picking any threshold above the floor instead would hand an attacker the + /// eviction primitive the floor exists to deny. + /// + /// `retain` runs first, so a peer that left the pool is neither ranked nor remembered. + pub(crate) fn dialable(&self, pool: &[RoutedPeer], now_ticks: u64) -> Vec { + self.retain(pool); + pool.iter() + .copied() + .filter(|peer| self.dial_share(*peer, now_ticks) > 0.0) + .collect() + } + + /// Drop the records of peers no longer in `pool`, so this map is bounded by pool membership. + pub(crate) fn retain(&self, pool: &[RoutedPeer]) { + self.lock().retain(|peer, _| pool.contains(peer)); + } + + /// How many peers this node currently holds conduct for. Test-only: the bound it proves is the + /// pool's, not a number this code chooses. + #[cfg(test)] + pub(crate) fn len(&self) -> usize { + self.lock().len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn peer(n: u8) -> RoutedPeer { + RoutedPeer::from_pool_key(&hex::encode([n; 32])).expect("64-hex is a pool key") + } + + /// **Proves (#268):** a peer that delivers bytes failing verification loses dial share, while a + /// peer observed answering honestly at the same moment keeps its full share. + /// + /// The honest peer is the load-bearing half. Asserting only that the liar drops passes on an + /// implementation that penalises EVERY peer — including the one that behaved — because that + /// implementation also drops the liar. Varying one actor against a truthful control is what + /// separates "conduct is read" from "a number went down". + #[test] + fn a_proven_lie_costs_dial_share_while_an_honest_peer_keeps_its_own() { + let conduct = ConductState::new(); + let liar = peer(0xa1); + let honest = peer(0xb2); + + conduct.observe(liar, ConductEvidence::ProvenLie, 0); + conduct.observe(honest, ConductEvidence::HonestAnswer, 0); + + let liar_share = conduct.dial_share(liar, 0); + let honest_share = conduct.dial_share(honest, 0); + + assert!( + liar_share < honest_share, + "a proven lie must cost dial share (liar {liar_share}, honest {honest_share})" + ); + assert!( + (honest_share - 1.0).abs() < f64::EPSILON, + "the honest peer must be unpenalised, or every peer is being penalised alike \ + (got {honest_share})" + ); + } + + /// **Proves (#268):** a proven lie is DURABLE — it does not decay, however much time passes. + /// + /// This is the direction that distinguishes the two verifiable classes from non-performance, and + /// it is asserted at a tick far beyond the non-performance decay window so a decay applied to the + /// wrong field would show up here rather than passing silently. + #[test] + fn a_proven_lie_does_not_decay_with_elapsed_time() { + let conduct = ConductState::new(); + let liar = peer(0xa2); + conduct.observe(liar, ConductEvidence::ProvenLie, 0); + + let immediately = conduct.dial_share(liar, 0); + let much_later = conduct.dial_share(liar, dig_sex::NON_PERFORMANCE_DECAY_TICKS * 100); + + assert!( + (immediately - much_later).abs() < f64::EPSILON, + "a verifiable fault is a fact about what the peer did and must not age away \ + ({immediately} then {much_later})" + ); + } + + /// **Proves (#268):** non-performance costs some share and then recovers on ELAPSED TIME ALONE — + /// the peer is never spoken to between the two reads. + /// + /// That is the property that keeps an attacker from evicting an honest holder by manufacturing + /// distress in it. A recovery that required a fresh successful exchange would be unreachable for a + /// peer nobody is dialling any more, which is precisely the peer being punished. + #[test] + fn non_performance_decays_on_elapsed_time_without_the_peer_being_talked_to() { + let conduct = ConductState::new(); + let distressed = peer(0xc3); + + conduct.observe(distressed, ConductEvidence::NonPerformance, 0); + let penalised = conduct.dial_share(distressed, 0); + assert!( + penalised < 1.0, + "a timeout must cost something, or the observation is not being read ({penalised})" + ); + + // No `observe` between these two reads: nothing happened except the clock moving. + let recovered = conduct.dial_share(distressed, dig_sex::NON_PERFORMANCE_DECAY_TICKS * 4); + assert!( + recovered > penalised, + "non-performance must decay on elapsed time alone ({penalised} then {recovered})" + ); + } + + /// **Proves (#268):** sustained non-performance never reduces a dial share to zero. + /// + /// Far more observations than the crate's ceiling, so an implementation that let the penalty run + /// unbounded would reach zero here. A zero share means an honest peer that a hostile third party + /// merely made SLOW can be removed from this node's dialling entirely — the eviction the floor + /// exists to prevent. + #[test] + fn sustained_non_performance_never_silences_a_peer_completely() { + let conduct = ConductState::new(); + let distressed = peer(0xd4); + for _ in 0..(dig_sex::NON_PERFORMANCE_CEILING * 10) { + conduct.observe(distressed, ConductEvidence::NonPerformance, 0); + } + + let share = conduct.dial_share(distressed, 0); + assert!( + share >= dig_sex::MIN_NON_PERFORMANCE_DIAL_SHARE, + "unverifiable distress must never drive a share below the floor (got {share})" + ); + assert!(share > 0.0, "a distressed honest peer must remain dialable"); + } + + /// **Proves (#268):** an unobserved peer is treated as neutral, not as suspect. + /// + /// Guards the direction where a missing record is read as a bad one — which would make every newly + /// connected peer worse than a known-mediocre one and quietly freeze the routing table. + #[test] + fn an_unobserved_peer_starts_neutral_rather_than_penalised() { + let conduct = ConductState::new(); + let share = conduct.dial_share(peer(0xe5), 0); + assert!( + (share - 1.0).abs() < f64::EPSILON, + "a peer with no history must be indistinguishable from one that behaved (got {share})" + ); + } + + /// **Proves (#268):** conduct is bounded by pool membership — a departed peer leaves this node's + /// memory when it leaves the pool, so the map cannot grow while the pool does not. + #[test] + fn retain_drops_peers_that_left_the_pool() { + let conduct = ConductState::new(); + let stays = peer(0x01); + let goes = peer(0x02); + conduct.observe(stays, ConductEvidence::NonPerformance, 0); + conduct.observe(goes, ConductEvidence::ProvenLie, 0); + assert_eq!(conduct.len(), 2); + + conduct.retain(&[stays]); + + assert_eq!(conduct.len(), 1, "the departed peer must be forgotten"); + assert!( + (conduct.dial_share(goes, 0) - 1.0).abs() < f64::EPSILON, + "a forgotten peer returns to neutral — its durable fault was dropped WITH its session, \ + which is the bound's cost and is deliberate: the alternative is an unbounded map keyed \ + by identities an attacker mints for free" + ); + } + + /// **Proves (#268):** the dial filter excludes a proven liar from the peers this node will spend + /// a dial on, and leaves a merely-distressed peer in. + /// + /// This is the assertion that makes the whole wiring load-bearing rather than merely reachable: + /// it is about the SET the node dials, not about a number a function returned. The distressed + /// peer is the control, and it is the half that catches an over-eager filter — a threshold set + /// anywhere above the non-performance floor would drop it too, handing an attacker exactly the + /// eviction primitive the floor exists to deny. + #[test] + fn a_proven_liar_leaves_the_dial_set_while_a_merely_slow_peer_stays_in_it() { + let conduct = ConductState::new(); + let liar = peer(0x11); + let slow = peer(0x22); + let quiet = peer(0x33); + let pool = [liar, slow, quiet]; + + conduct.observe(liar, ConductEvidence::ProvenLie, 0); + // Far past the ceiling: as much non-performance as an attacker could ever manufacture. + for _ in 0..(dig_sex::NON_PERFORMANCE_CEILING * 10) { + conduct.observe(slow, ConductEvidence::NonPerformance, 0); + } + + let dialable = conduct.dialable(&pool, 0); + + assert!( + !dialable.contains(&liar), + "a peer with a verifiable fault must not be dialled" + ); + assert!( + dialable.contains(&slow), + "unverifiable distress must NEVER remove a peer from the dial set, however sustained — otherwise loading an honest holder is enough to evict it" + ); + assert!( + dialable.contains(&quiet), + "an unobserved peer must remain dialable" + ); + } + + /// **Proves (#268):** a self-contradiction is treated as verifiable, like a lie and unlike a + /// timeout — the peer contradicted its OWN claim, which needs no trust to check. + #[test] + fn a_self_contradiction_is_durable_like_a_lie_not_transient_like_a_timeout() { + let conduct = ConductState::new(); + let contradictor = peer(0x44); + conduct.observe(contradictor, ConductEvidence::SelfContradiction, 0); + + assert_eq!( + conduct.dial_share(contradictor, dig_sex::NON_PERFORMANCE_DECAY_TICKS * 100), + 0.0, + "a peer that announced content then denied holding it earned a durable exclusion" + ); + } +} diff --git a/crates/dig-node-core/src/seams/dig_peer/mod.rs b/crates/dig-node-core/src/seams/dig_peer/mod.rs index 9531af35..5e4263dd 100644 --- a/crates/dig-node-core/src/seams/dig_peer/mod.rs +++ b/crates/dig-node-core/src/seams/dig_peer/mod.rs @@ -11,9 +11,11 @@ //! (matching W1b-0's pattern) but is out of scope for this trait carve. pub mod address_book; +pub mod admission; pub mod ask_routing; pub mod bootstrap; pub mod capsule_fallback; +pub mod conduct; // The self-verifying tier-0 preimage resolver (#2033, PR-2). Its surface is exercised by its own // tests but not yet CALLED by production code — the tier-0 fetch loop wires it in PR-3 — so the // as-yet-unconsumed client + resolver would otherwise trip `dead_code`. The allow is removed when