diff --git a/CHANGELOG.md b/CHANGELOG.md index e5bdfe7..9705069 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,15 @@ All notable changes to this project are documented here. This project adheres to [Semantic Versioning](https://semver.org) and [Conventional Commits](https://www.conventionalcommits.org). +## [0.64.0] - 2026-07-27 + +### Features +- **holdings:** Real-time opcode-222 announce/retract + verified rate-limited inbound ingest (#1429) +- **holdings:** Canonicalize provider identity, bound announcement freshness, keep rejections allocation-free (#1429) + +### Tests +- **holdings:** Pin the provider map's LRU capacity and its eviction victim, the address cap at its placement, and the ingest kill switch (#1429) + ## [0.63.0] - 2026-07-27 ### Bug Fixes diff --git a/Cargo.lock b/Cargo.lock index 65dd838..def367c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2211,7 +2211,7 @@ dependencies = [ [[package]] name = "dig-node-core" -version = "0.21.0" +version = "0.22.0" dependencies = [ "async-trait", "axum", @@ -2249,6 +2249,7 @@ dependencies = [ "getrandom 0.2.17", "hex", "lru", + "rcgen", "reqwest", "ring", "rustls", @@ -2265,7 +2266,7 @@ dependencies = [ [[package]] name = "dig-node-service" -version = "0.63.0" +version = "0.64.0" dependencies = [ "async-trait", "axum", diff --git a/Cargo.toml b/Cargo.toml index 110549b..5ff5e3a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,7 +31,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.63.0" +version = "0.64.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/SPEC.md b/SPEC.md index 6ea9b6b..90baf25 100644 --- a/SPEC.md +++ b/SPEC.md @@ -3122,6 +3122,120 @@ capsule-gain path flows), guarded so an already-held capsule is a no-op (unchang re-announce). It is best-effort and a no-op on the in-process FFI path (no peer network / inventory refresher installed). +### 19.3a. Real-time holdings announce — dig-gossip opcode 222 (#1429) + +Beside the DURABLE provider records above, the node maintains a REAL-TIME holder signal. The durable +records converge only as fast as the last PUT reached and a departed holder lingers for its record's TTL; +the announce closes that gap. + +**Egress (MUST).** Every inventory reconcile MUST derive both effects from ONE delta: the DHT records +(announce gained ids, ACTIVELY retract lost ids via `retract_own_provider`, never the passive +`withdraw_provider` — a passively withdrawn record keeps answering `find_providers` with this node for +content it can no longer serve, costing each reader a wasted dial), AND a signed opcode-222 +`HoldingsAnnounce` carrying `Add`/`Remove` deltas for exactly those ids. A reconcile larger than +`HOLDINGS_MAX_CHANGES` (256) deltas MUST be SPLIT across frames, never truncated — truncation would drop +retracts and leave this node advertising content it does not hold. + +- The announcement MUST be signed by the node's own `NodeCert` leaf (ECDSA-P256), because the wire + derives `provider_peer_id` from `SHA-256(provider_spki)`; signing with any other key announces an + identity no peer can dial. +- The advertised addresses MUST be the SAME advertised candidate set the DHT provider records carry. +- `seq` MUST strictly increase per node, and MUST NOT restart from zero — a node resuming from zero has + every announcement dropped as a replay by peers that remember its previous watermark. The + implementation seeds from the wall clock and increments once per batch. Note the two limits of that + seed, neither of which a receiver can be harmed by: it does NOT guarantee the seed exceeds every value + previously announced (a node averaging more than one batch per second outruns its own clock, so the + seed after a restart can be below the counter it reached), and a backwards clock adjustment has the + same effect. The consequence is self-inflicted silence — this node's announcements are ignored until + the counter it left behind is passed — never an accepted replay. Seeding above the last value + announced requires persisting the counter, which is deferred (#1477). +- A degraded node (no signable leaf, no inbound receiver) MUST remain discoverable through the durable + DHT records — the real-time layer is additive and MUST NOT be a hard dependency of discoverability. + +**Ingress (MUST).** An inbound announcement is applied ONLY after all of the following, fail-closed, at a +single chokepoint. dig-dht is crypto-free by design, and `ingest_verified_provider` is the sole sanctioned +bypass of its mTLS self-announce check, so these are the whole of the authentication: + +1. `verify_holdings_announce` passes (batch cap, `SHA-256(provider_spki) == provider_peer_id`, P-256 SPKI, + valid signature over the `dig:holdings:v1` domain-separated message). +2. `provider_peer_id` is CANONICALIZED (decoded to 32 bytes, re-encoded lowercase) and every subsequent + comparison, map key and provider argument uses ONLY that value. This is normative, not hygiene: hex + decoding is case-insensitive and the signature covers the DECODED bytes, so one identity has many + spellings that all verify. A receiver that keys on the raw wire text gives each spelling its own replay + watermark and misses a lowercase self-identity comparison, turning gates 3 and 5 into no-ops at zero + cryptographic cost. A `provider_peer_id` that is not 64 hex characters MUST be refused before it is + compared, keyed, or logged. +3. The announcement is NOT attributed to this node's own (canonicalized) `peer_id` — the network MUST NOT + be able to tell a node what it holds. +4. `announced_at` is within **±300 s** of the receiver's clock; otherwise the announcement is refused as + stale. REQUIRED, not defence in depth: a `Remove` delta carries no expiry of its own, so without this + the only barrier to replaying a captured retract indefinitely is the in-memory watermark of gate 5 — + which a restart clears and a capacity eviction drops, letting anyone de-list an honest holder by + replaying that holder's own old retract at a freshly started peer. The signature binds WHO announced; + only this bound binds WHEN. The check is symmetric, because a future-dated frame stays replayable + longer than it should. +5. `seq` strictly advances beyond the highest already applied for that CANONICAL provider identity; + otherwise the announcement is dropped (a replayed older frame MUST NOT resurrect a retracted record). + A provider not yet seen has NO watermark, which is distinct from a watermark of zero — `seq` is not + required to start at 1. A rate window elapsing MUST NOT reset the watermark: replay protection is not + a rate limit. +6. Two token buckets, both within a 60-second window: at most **10 announcements per PROVIDER**, and at + most **1,024 (`4 × HOLDINGS_MAX_CHANGES`) deltas per TRANSPORT SENDER**. They are keyed differently on + purpose — a provider id is attacker-minted so its bucket map must be capacity-bounded and is therefore + evictable, whereas the sender key space is the connected pool and cannot be inflated from off-network, + making it the unbypassable backstop. A REJECTED announcement MUST charge neither bucket, or the + limiter itself becomes the denial of service. + +**Bounded state (MUST).** A rejected announcement MUST NOT allocate tracking state. Admission is decided +in full BEFORE any per-provider or per-sender entry is created, because every rejection path returns early +and would otherwise skip the eviction step, letting an attacker-minted key space grow the maps for the +price of one ~180-byte frame per entry. Admitted entries are capacity-bounded (1,024 senders, 8,192 +providers) with least-recently-seen eviction. + +**Untrusted fields (MUST).** `provider_peer_id` is a `u16`-length-prefixed wire string and may carry tens +of kilobytes of arbitrary UTF-8, including newlines and terminal escapes. No peer-supplied field may be +emitted to the log; only the canonicalized identity may be. Bounding log LEVEL bounds volume, not content. +An `Add` delta's address list MUST be truncated to `MAX_ADDRESSES_PER_RECORD` before it is mapped, since +its length is an attacker-declared `u16`. + +**Operator control (SHOULD).** `DIG_HOLDINGS_INGEST=0` disables the ingress while leaving egress intact; +discovery then degrades to the durable DHT records rather than failing. + +**Attribution (MUST).** The provider identity used for both `ingest_verified_provider` and +`remove_provider_record` MUST be the VERIFIED signer and nothing else. A retract therefore removes only +the signer's own record and can never de-list another holder of the same content key. The wire carries no +per-delta peer field, so this holds structurally: the receive path accepts no caller-supplied provider id. + +**No amplification (MUST).** The ingress performs NO egress — it never re-broadcasts, dials, probes or +fetches. One inbound announcement costs the receiver bounded local map work only, so a cheap anonymous +message can never make honest peers do more work than the sender did. + +**Reach is ONE HOP (normative, and deliberate).** An announcement is delivered to the announcer's directly +connected peers and is NOT re-flooded onward: dig-gossip's Plumtree eager/lazy dissemination applies to +frames a node ORIGINATES via `broadcast`, and the opcode-222 receive path does not re-broadcast. The +real-time layer is therefore a NEIGHBOURHOOD freshness signal, and the DHT provider records (PUT to the +k-closest peers and republished) remain the only network-wide discovery mechanism. This is sufficient for +the layer's purpose: + +- For an ADD, one hop covers the highest-value case. A node's direct peers are exactly who is positioned + to fetch from it, and the fetch path already prefers connected-pool holders over a DHT record. Peers + further away discover the new holder through the durable records, as they always did. +- For a REMOVE, one hop leaves peers more than one hop away holding a stale record until TTL. That is a + LATENCY gap, not a correctness gap: the local record is deleted at once by `retract_own_provider`, the + k-closest copies age out, and the cost of a stale record is one failed dial that the peer-selector + deranks. + +Re-flooding verified announcements would close the remaining gap but MUST NOT be added at the ingress: +that would make one inbound message produce N outbound, which is precisely the amplification this layer is +built to avoid. The correct shape, if wider reach is later required, is for dig-gossip to relay verified +opcode-222 frames through its EXISTING Plumtree seen-set (which already provides dedup and flood control), +never a re-broadcast from the receive path. + +**False claims.** A peer MAY announce content it does not hold. This is bounded by cost asymmetry rather +than prevented: the liar pays a signature and a flood, and buys at most one failed dial or a +`dig.getAvailability` answering "no", after which the peer-selector deranks it. No merkle- or +chain-verification decision ever rests on an announcement. + ### 19.4. Address book — durable, IPv6-first, provenance + TTL The node maintains a durable peer address book: every learned peer candidate — from PEX, `dig.getPeers`, diff --git a/crates/dig-node-core/Cargo.toml b/crates/dig-node-core/Cargo.toml index b7dd5b7..f37e595 100644 --- a/crates/dig-node-core/Cargo.toml +++ b/crates/dig-node-core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dig-node-core" -version = "0.21.0" +version = "0.22.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." @@ -272,6 +272,11 @@ futures = { version = "0.3", default-features = false, features = ["std", "async [dev-dependencies] tempfile = "3" +# Generates the P-256 leaf key pairs the opcode-222 holdings tests sign with: `public_key_der()` IS the +# leaf `SubjectPublicKeyInfo` DER the wire carries as `provider_spki`, and whose SHA-256 is the +# announcing peer_id — so a test signer is a real §5.2 identity rather than a stand-in that could hide +# an attribution bug (#1429). +rcgen = "0.13" # dig-download's in-memory harness (`MockRangeTransport`, `MockContent`, …), which the download-path # tests drive the real orchestrator over. # diff --git a/crates/dig-node-core/src/peer.rs b/crates/dig-node-core/src/peer.rs index d267f2d..d54f8a8 100644 --- a/crates/dig-node-core/src/peer.rs +++ b/crates/dig-node-core/src/peer.rs @@ -843,21 +843,57 @@ pub trait PeerRpcResponder: Send + Sync { struct DhtInventoryAnnouncer { node: Arc, dht: Arc, + /// The real-time opcode-222 flood, when this node can sign one (#1429). + holdings: Option, } #[async_trait::async_trait] impl crate::seams::dig_peer::AnnounceHolder for DhtInventoryAnnouncer { async fn announce_inventory(&self) { - let cached = self.node.cache_list_cached().await; - let (announced, withdrawn) = self.dht.refresh_inventory(&cached).await; + let delta = reconcile_and_flood(&self.node, &self.dht, self.holdings.as_ref()).await; tracing::info!( - announced, - withdrawn, + announced = delta.gained.len(), + retracted = delta.lost.len(), "capsule warm: announced this node as a holder of the newly cached capsule" ); } } +/// The signer plus the pool it floods to — everything needed to emit an opcode-222 announcement. +#[derive(Clone)] +struct HoldingsFlood { + broadcaster: Arc, + pool: dig_gossip::GossipHandle, +} + +/// The node's ONE reaction to an inventory change: reconcile the durable DHT provider records, then +/// flood the matching real-time opcode-222 announcement for exactly the ids that changed. +/// +/// Both change hooks (the #1576 reshare warm and the generic inventory refresher) route through here, +/// so the flood is derived from the SAME delta that moved the records — a capsule can never be +/// announced as gained while its provider record says otherwise, and a retract can never be skipped. +async fn reconcile_and_flood( + node: &Arc, + dht: &Arc, + holdings: Option<&HoldingsFlood>, +) -> crate::dht::InventoryDelta { + let cached = node.cache_list_cached().await; + // Reading the inventory is this shell's job; the reconcile-plus-flood composition itself lives in + // `holdings` so it can be tested against a real DhtService without a `Node`. + crate::seams::dig_peer::holdings::reconcile_and_announce( + dht, + &cached, + holdings.map(|f| { + ( + f.broadcaster.as_ref(), + &f.pool as &dyn crate::seams::dig_peer::holdings::AnnounceTransport, + ) + }), + crate::seams::dig_peer::holdings::now_unix_secs(), + ) + .await +} + /// Serve peer requests over one established, mTLS-authenticated [`dig_nat::mux::PeerSession`] (the /// SERVER role): accept inbound logical streams and answer each concurrently. Every stream is read as /// one framed request, classified by shape, and answered — a JSON-RPC request via @@ -2208,7 +2244,7 @@ async fn run_peer_network(node: Arc) -> Result<(), String> { let address_book = Arc::new(crate::address_book::AddressBook::default()); let mut dial_ranker: Option> = None; - let dht = match bring_up_dht( + let (dht, holdings) = match bring_up_dht( &node, &identity, &nat_runtime, @@ -2218,14 +2254,21 @@ async fn run_peer_network(node: Arc) -> Result<(), String> { ) .await { - Ok(dht) => Some(dht), + Ok((dht, holdings)) => (Some(dht), holdings), Err(e) => { tracing::warn!(error = %e, "dig-node DHT bring-up failed; continuing without the DHT"); status.set_error(format!("dht: {e}")); - None + (None, None) } }; + // The flood half of #1429, ready for both inventory-change hooks below: present only when the DHT + // is up AND this node can sign an announcement (the DHT records are the durable fallback if not). + let holdings_flood = holdings.map(|broadcaster| HoldingsFlood { + broadcaster, + pool: handle.clone(), + }); + // 4a. Feed the DHT routing table LIVE from the gossip pool (#1574): bring_up_dht's one-shot // bootstrap runs BEFORE any peer connects, so in a freshly-formed network routing starts empty // and find_providers finds nobody. This forwards every PoolEvent (add → insert, remove → @@ -2280,6 +2323,7 @@ async fn run_peer_network(node: Arc) -> Result<(), String> { Arc::new(DhtInventoryAnnouncer { node: node.clone(), dht: dht.clone(), + holdings: holdings_flood.clone(), }), node.cache_dir_path(), ); @@ -2298,16 +2342,17 @@ async fn run_peer_network(node: Arc) -> Result<(), String> { if let Some(dht) = dht.clone() { let node_for_hook = node.clone(); let dht_for_hook = dht.clone(); + let flood_for_hook = holdings_flood.clone(); node.set_inventory_refresher(Box::new(move || { let node = node_for_hook.clone(); let dht = dht_for_hook.clone(); + let flood = flood_for_hook.clone(); Box::pin(async move { - let cached = node.cache_list_cached().await; - let (announced, withdrawn) = dht.refresh_inventory(&cached).await; - if announced > 0 || withdrawn > 0 { + let delta = reconcile_and_flood(&node, &dht, flood.as_ref()).await; + if !delta.is_empty() { tracing::debug!( - announced, - withdrawn, + announced = delta.gained.len(), + retracted = delta.lost.len(), "dig-node DHT: refreshed provider records after an inventory change" ); } @@ -2404,7 +2449,13 @@ async fn bring_up_dht( network_id: &str, pool: &dig_gossip::GossipHandle, stun_servers: &[std::net::SocketAddr], -) -> Result, String> { +) -> Result< + ( + Arc, + Option>, + ), + String, +> { use dig_dht::{CandidateAddr, DhtConfig, DhtService}; // The single IPv6-first STUN server feeds the DHT transport's hole-punch tier (one reflexive-input @@ -2442,14 +2493,18 @@ async fn bring_up_dht( // Assemble the advertised candidate set, IPv6-first via dig_ip::Family (the reflexive leads its // family group); see `crate::net::assemble_advertised`. The wildcard bind (`[::]` / `0.0.0.0`) // is never a candidate. - let local_addresses: Vec = crate::net::advertised_socket_addrs_with_reflexive( + // Kept as `SocketAddr`s first: the SAME advertised set feeds BOTH the DHT provider records and + // the opcode-222 holdings announcements below, so the two discovery paths can never disagree + // about where this node serves. + let advertised: Vec = crate::net::advertised_socket_addrs_with_reflexive( port, crate::net::advertise_loopback_from_env(), reflexive, - ) - .into_iter() - .map(|sa| CandidateAddr::direct(sa.ip().to_string(), sa.port())) - .collect(); + ); + let local_addresses: Vec = advertised + .iter() + .map(|sa| CandidateAddr::direct(sa.ip().to_string(), sa.port())) + .collect(); let service = Arc::new(DhtService::new( node_cert.peer_id(), local_addresses, @@ -2486,6 +2541,66 @@ async fn bring_up_dht( let dht = crate::dht::DhtHandle::new(service, initial_ids); + // The real-time holdings layer (#1429): flood a signed opcode-222 announcement whenever this + // node's inventory changes, and fold every peer's verified announcement into our provider set. + // + // Signed by the node's OWN NodeCert leaf, because the wire derives `provider_peer_id` from the + // signing key's SPKI — announcing under any other key would name an identity no peer can dial. + // Advertising the SAME `local_addresses` the DHT records carry keeps the two discovery paths in + // agreement about where this node serves. + let holdings = match crate::seams::dig_peer::holdings::signer_from_node_cert(node_cert) { + Ok(signer) => { + let addresses = advertised + .iter() + .map(|sa| dig_gossip::CandidateAddr { + host: sa.ip().to_string(), + port: sa.port(), + }) + .collect(); + // Seeded from the wall clock so a restart resumes ABOVE any seq peers already remember + // (a from-zero restart would have its announcements dropped as replays until it caught + // up). Persisting the counter is #1477's durable-state work. + let broadcaster = Arc::new(crate::seams::dig_peer::holdings::HoldingsBroadcaster::new( + signer, + addresses, + crate::seams::dig_peer::holdings::now_unix_secs(), + )); + match pool.inbound_receiver() { + Ok(_) if !crate::seams::dig_peer::holdings::HoldingsIngress::ingest_enabled_from_env() => { + // Operator kill switch: keep ANNOUNCING (so this node stays discoverable in real + // time) but stop applying peers' announcements. Discovery falls back to the + // durable DHT records, which is a degradation, not an outage. + println!( + "dig-node peer network: holdings ingest DISABLED by \ + DIG_HOLDINGS_INGEST; still announcing" + ); + } + Ok(inbound) => { + let ingress = Arc::new(crate::seams::dig_peer::holdings::HoldingsIngress::new( + node_cert.peer_id().to_hex(), + )); + let sink = Arc::clone(dht.service()); + tokio::spawn(crate::seams::dig_peer::holdings::run_holdings_ingest( + inbound, ingress, sink, + )); + println!("dig-node peer network: holdings announce (opcode 222) up"); + } + Err(e) => tracing::warn!( + error = %e, + "holdings announce: no inbound receiver; this node still ANNOUNCES but will not \ + ingest peers' announcements (DHT find_providers remains the fallback)" + ), + } + Some(broadcaster) + } + Err(e) => { + // Announce-only degradation, never fatal: without a signer this node stays discoverable + // through the durable DHT provider records, just not in real time. + tracing::warn!(error = %e, "holdings announce disabled: node cert leaf is not signable"); + None + } + }; + // Spawn the maintenance loop: republish (records never lapse) + refresh buckets + gc, well inside // the provider TTL. { @@ -2496,7 +2611,7 @@ async fn bring_up_dht( }); } - Ok(dht) + Ok((dht, holdings)) } /// Run the mTLS peer-RPC accept loop over a pre-bound `listener`: accept inbound TLS connections diff --git a/crates/dig-node-core/src/seams/dig_peer/dht.rs b/crates/dig-node-core/src/seams/dig_peer/dht.rs index 51ff007..c2d9b8f 100644 --- a/crates/dig-node-core/src/seams/dig_peer/dht.rs +++ b/crates/dig-node-core/src/seams/dig_peer/dht.rs @@ -390,24 +390,56 @@ pub fn inventory_diff( (to_announce, to_withdraw) } +/// What an inventory reconcile changed — the content ids this node started and stopped providing. +/// +/// Returned (rather than just counted) so the caller can flood the matching real-time opcode-222 +/// [`HoldingsAnnounce`](dig_gossip::HoldingsAnnounce) deltas from the SAME reconcile that moved the +/// DHT records. One source of truth for "what changed" means the flood can never disagree with the +/// provider records it is announcing. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct InventoryDelta { + /// Content ids this node now provides and did not before. + pub gained: Vec, + /// Content ids this node no longer provides. + pub lost: Vec, +} + +impl InventoryDelta { + /// Whether the reconcile was a no-op (nothing to announce and nothing to retract). + #[must_use] + pub fn is_empty(&self) -> bool { + self.gained.is_empty() && self.lost.is_empty() + } +} + /// React to an inventory change: `announce_provider` newly-held content ids promptly (don't wait for -/// the periodic republish tick) and `withdraw_provider` ids the node no longer holds (they then age -/// out of the DHT via TTL). Returns `(announced, withdrawn)` counts. `previous` is the last-known -/// content-id set, `current` is derived from the fresh inventory. +/// the periodic republish tick) and ACTIVELY retract ids the node no longer holds. `previous` is the +/// last-known content-id set, `current` is derived from the fresh inventory. +/// +/// Loss uses `retract_own_provider`, NOT the passive `withdraw_provider`: the passive form leaves the +/// local record in place to lapse via TTL, so for as long as that TTL runs this node keeps answering +/// `find_providers` with itself for content it can no longer serve — every such answer costs a reader +/// a wasted dial. The active retract deletes the local record and stops republishing it immediately, +/// which is the local half of #1423's atomic evict-and-retract. (Copies already PUT at the k closest +/// peers still age out via TTL, or are removed sooner when the caller floods the signed retract +/// announce — which is why [`InventoryDelta::lost`] is returned rather than discarded.) pub async fn sync_inventory( dht: &DhtService, previous: &[dig_dht::ContentId], cached: &[CachedCapsule], -) -> (usize, usize) { +) -> InventoryDelta { let current = inventory_content_ids(cached); let (to_announce, to_withdraw) = inventory_diff(previous, ¤t); for id in &to_announce { let _ = dht.announce_provider(id).await; } for id in &to_withdraw { - dht.withdraw_provider(id).await; + dht.retract_own_provider(id).await; + } + InventoryDelta { + gained: to_announce, + lost: to_withdraw, } - (to_announce.len(), to_withdraw.len()) } // -- Bootstrap peers from the gossip pool ------------------------------------------------------------ @@ -460,14 +492,25 @@ impl DhtHandle { } /// Re-derive the inventory content-id set from `cached` and reconcile it with the DHT: announce - /// new ids, withdraw gone ids (see [`sync_inventory`]). Updates the remembered set. Call whenever - /// the node's inventory changes (a capsule cached, a root advanced, a store removed). Returns - /// `(announced, withdrawn)`. - pub async fn refresh_inventory(&self, cached: &[CachedCapsule]) -> (usize, usize) { + /// new ids, actively retract gone ids (see [`sync_inventory`]). Updates the remembered set. Call + /// whenever the node's inventory changes (a capsule cached, a root advanced, a store removed). + /// + /// Returns WHICH ids changed, so the caller can flood the matching real-time opcode-222 announce + /// from this same reconcile. This is the node's ONE inventory-reconcile chokepoint: every path + /// that gains or loses a capsule (gap-fill, explicit cache, the #1576 reshare warm) goes through + /// it, which is what keeps the DHT records and the flood in agreement. + pub async fn reconcile_inventory(&self, cached: &[CachedCapsule]) -> InventoryDelta { let mut announced = self.announced.lock().await; - let (a, w) = sync_inventory(&self.service, &announced, cached).await; + let delta = sync_inventory(&self.service, &announced, cached).await; *announced = inventory_content_ids(cached); - (a, w) + delta + } + + /// [`Self::reconcile_inventory`] projected to `(announced, retracted)` counts, for callers that + /// only log the outcome. + pub async fn refresh_inventory(&self, cached: &[CachedCapsule]) -> (usize, usize) { + let delta = self.reconcile_inventory(cached).await; + (delta.gained.len(), delta.lost.len()) } /// Locate the peers holding `content` via the DHT (`find_providers`). The returned diff --git a/crates/dig-node-core/src/seams/dig_peer/holdings.rs b/crates/dig-node-core/src/seams/dig_peer/holdings.rs new file mode 100644 index 0000000..44e7e30 --- /dev/null +++ b/crates/dig-node-core/src/seams/dig_peer/holdings.rs @@ -0,0 +1,949 @@ +//! Real-time holdings announce (dig-gossip **opcode 222**) — the flywheel's freshness signal. +//! +//! The content-replication flywheel is `read → cache → announce → others discover you → they read +//! from you → …`. Its *durable* half already exists in [`dht`](super::dht): every inventory change +//! reconciles the node's dig-dht provider records (`announce_provider` on gain, withdraw on loss) +//! and `find_providers` locates holders. That half is correct but **slow to converge** — a new +//! holder is only as discoverable as the last DHT PUT reached, and a departed holder lingers until +//! its record's TTL expires. +//! +//! This module adds the *real-time, authenticated* half: +//! +//! - **Egress** ([`announcement_for`]) — an inventory delta becomes a signed [`HoldingsAnnounce`] +//! flooded to every peer, so holder-set changes propagate in seconds rather than at TTL scale. +//! - **Ingress** ([`HoldingsIngress`]) — an inbound announce is verified, rate-limited, and folded +//! into the local provider set via `ingest_verified_provider` / `remove_provider_record`. +//! +//! # Why the ingress is the dangerous half +//! +//! dig-dht is crypto-free by design (its SPEC §15): `ingest_verified_provider` is the **sole +//! sanctioned bypass** of the DHT's own mTLS self-announce identity check, and +//! `remove_provider_record` takes `(content_key, provider_peer_id)` as two independent strings with +//! nothing binding them. Every authentication, attribution and rate guard therefore lives HERE. +//! The threat classes this module is built against, each guarded over the CLASS and not one +//! attacker action: +//! +//! | Class | Guard | +//! |---|---| +//! | **Forged attribution** — naming another peer as a holder, or as *not* a holder | An announce's provider identity is [`HoldingsAnnounce::provider_peer_id`], which `verify_holdings_announce` proves equals `SHA-256(provider_spki)` and which signed the batch. [`HoldingsIngress::accept`] takes **no** caller-supplied provider id, so no code path can name a peer the signature does not attribute. | +//! | **Identity re-spelling** — the same signer wearing many names | Hex is case-INSENSITIVE and the signature covers the DECODED bytes, so one identity has many spellings that all verify. [`HoldingsIngress::accept`] canonicalizes ONCE, up front, and every later comparison, map key and sink argument uses only that value — otherwise each spelling is a fresh provider with no replay watermark, and a lowercase self-id comparison misses all of them. | +//! | **Amplification** — one cheap inbound message causing outbound work | The ingress performs **zero egress**: it never re-broadcasts, dials, probes, or fetches. Its whole cost is bounded local map work. | +//! | **Flood / Sybil eviction** — evicting an honest holder from a full provider set | Two token buckets at ONE chokepoint ([`IngressLimits`]): announcements per *provider* and deltas per *transport sender*. They are keyed differently on purpose — see [`MAX_DELTAS_PER_SENDER`] for why each covers the other's weakness. A rejected announcement charges neither, so the limiter cannot itself be turned into the denial of service. | +//! | **Replay** — resurrecting a retracted record, or undoing a newer announce | TWO independent barriers, because either alone is escapable: a per-provider monotonic [`HoldingsAnnounce::seq`] watermark (keyed by the CANONICAL id), and a bounded-freshness check on the signed `announced_at` ([`MAX_ANNOUNCE_AGE_SECS`]). The watermark is in-memory, so a restart or a capacity eviction clears it; freshness holds regardless of watermark state, which matters most for a `Remove` — it carries no expiry of its own. | +//! | **Self-poisoning** — replaying our own announce back at us | An announce attributed to this node's own `peer_id` is dropped; only this node decides what it holds. | +//! | **Unbounded state** — the guard maps themselves becoming the DoS | A REJECTED announcement allocates nothing at all: [`IngressState::admit`] decides every gate against borrowed state and inserts only after all of them pass, so no reject path can grow a map. Admitted entries are then capacity-bounded with LRU eviction ([`MAX_TRACKED_SENDERS`], [`MAX_TRACKED_PROVIDERS`]). | +//! +//! A false claim is cheap to disprove and costs the liar, not the reader: a bogus provider record +//! only ever yields a *dial that fails* or a `dig.getAvailability` that answers "no", after which +//! the peer-selector deranks the liar. The liar paid a signature and a flood to buy one failed dial. + +use std::collections::HashMap; +use std::sync::Arc; + +use dig_dht::{CandidateAddr as DhtAddr, ContentId, Key, PeerId, ProviderRecord}; +use dig_gossip::{ + verify_holdings_announce, CandidateAddr as GossipAddr, EcdsaHoldingsSigner, HoldingsAnnounce, + HoldingsDelta, HoldingsError, HoldingsSigner, HOLDINGS_MAX_CHANGES, +}; + +/// How long an `Add` advertisement claims to stay valid, in seconds. +/// +/// Chosen to match dig-dht's own provider TTL band rather than a round number: dig-dht clamps an +/// ingested record to `min(record.expires_at, now + provider_ttl)`, so claiming *longer* than the +/// DHT's TTL is silently truncated and claiming *shorter* under-advertises a holder that is in fact +/// still serving. One hour sits inside every current provider TTL, so the clamp is a no-op and the +/// advertised lifetime is the one the announcer actually meant. +/// +/// COUPLING: this value MUST track [`dig_dht::DhtConfig::provider_ttl`] (default 2 hours) and stay +/// `<=` the smallest one any receiving node configures. Reducing that field below one hour, without a +/// matching reduction here, silently truncates every advertisement this node makes. +pub const ADVERTISED_TTL_SECS: u64 = 3_600; + +/// Announcements one **provider** may have applied per [`RATE_WINDOW_SECS`]. +/// +/// The value dig-dht's SPEC §14 delegates to the embedding node (~10 announces/holder/min). Keyed by +/// provider because that is what the number means semantically: how often a holder may revise the +/// content it advertises. An honest node announces on an inventory *change*; ten changes a minute is +/// already far above steady state. +pub const MAX_ANNOUNCES_PER_PROVIDER: u32 = 10; + +/// Deltas one **transport sender** may cause per [`RATE_WINDOW_SECS`]. +/// +/// The two buckets are deliberately keyed differently, because each covers the other's weakness: +/// +/// - The provider bucket is semantically right but its key space is attacker-minted, so its map must +/// be LRU-bounded ([`MAX_TRACKED_PROVIDERS`]) — and an attacker who overflows that map can evict +/// its own bucket and refill it. Alone, it is bypassable. +/// - This bucket is keyed by the gossip sender, whose key space is the connected pool — a set the +/// attacker cannot inflate from off-network. Alone it is coarse (it cannot distinguish one chatty +/// holder from a hundred quiet ones), but it is **not bypassable**, so it is the backstop that +/// caps total ingest work per neighbour no matter what the provider bucket lets through. +/// +/// The value is `4 × `[`HOLDINGS_MAX_CHANGES`] — a neighbour may relay four maximal signed batches, +/// or ~1,024 single-delta announcements, per window. It is deliberately far BELOW what the provider +/// bucket alone permits (`10 × 256 = 2,560` deltas per provider, times unbounded providers), which +/// is what makes it load-bearing rather than a restatement of the other bound. +pub const MAX_DELTAS_PER_SENDER: u32 = 4 * HOLDINGS_MAX_CHANGES as u32; + +/// The token-bucket refill window, in seconds. +pub const RATE_WINDOW_SECS: u64 = 60; + +/// How far `announced_at` may be from the receiver's clock, in either direction, before the +/// announcement is refused as stale. +/// +/// This bound is what makes a captured announcement stop being useful, and it is required rather +/// than merely defensive: a `Remove` delta carries NO expiry of its own, so without it the only +/// barrier to replaying a captured retract forever is the in-memory per-provider `seq` watermark — +/// which a restart clears and a capacity eviction drops. That would let anyone de-list an honest +/// holder by replaying the holder's own old retract at a freshly started peer, which is censorship, +/// not staleness. Persisting the watermark is complementary (#1477) but is NOT a substitute: the +/// signature binds WHO announced, and only this comparison binds WHEN. +/// +/// Five minutes is generous enough to absorb ordinary NTP skew and flood propagation delay while +/// keeping a captured frame useful for minutes rather than indefinitely. It is symmetric because a +/// future-dated frame is the same attack: an attacker who could post-date `announced_at` would mint a +/// retract that stays replayable long after capture. +pub const MAX_ANNOUNCE_AGE_SECS: u64 = 300; + +/// Transport senders tracked before the least-recently-seen is evicted. +/// +/// The key is a peer this node holds a live mTLS link to, but "headroom over a realistic connected +/// pool" is the wrong reading of this number: it is a HARD BOUND on unbounded growth. A `peer_id` is +/// `SHA-256(NodeCert SPKI)` — self-minted, costing one key pair — and an entry is never removed by +/// disconnection, so the key space this map follows is the set of transports that have EVER relayed +/// an accepted announcement, which connect → announce → disconnect churn grows without limit. The +/// eviction, not the pool size, is what keeps the map finite. +pub const MAX_TRACKED_SENDERS: usize = 1_024; + +/// Providers tracked (latest `seq` + announce bucket) before the least-recently-seen is evicted. +/// +/// Bounded because the provider id inside an announce is attacker-chosen. +/// +/// Losing an entry to eviction costs a provider its replay watermark. That is NOT harmless on its own +/// — an earlier version of this comment claimed it meant "bounded staleness, never cross-peer +/// censorship", which was wrong: the signature binds WHO announced but not WHEN, so replaying a +/// holder's own captured `Remove` (which carries no expiry) at a peer with no watermark de-lists an +/// honest holder. [`MAX_ANNOUNCE_AGE_SECS`] is what actually closes that class, independently of +/// watermark state. Losing an announce bucket is separately backstopped by +/// [`MAX_DELTAS_PER_SENDER`], which cannot be evicted out from under an attacker. +pub const MAX_TRACKED_PROVIDERS: usize = 8_192; + +// --------------------------------------------------------------------------------------------- +// Egress — turning an inventory change into a signed announcement +// --------------------------------------------------------------------------------------------- + +/// Build this node's holdings signer from its persistent mTLS identity. +/// +/// The wire fixes the signing key: `provider_peer_id` is `SHA-256(provider_spki)`, and the node's +/// advertised `peer_id` is `SHA-256(NodeCert SPKI DER)` (§5.2). Signing with anything other than the +/// [`NodeCert`](dig_nat::NodeCert) leaf would therefore announce holdings under an identity no peer +/// can dial. The leaf is ECDSA-P256 (dig-tls mints it with `PKCS_ECDSA_P256_SHA256`), which is +/// exactly what `verify_holdings_announce` requires. +/// +/// # Errors +/// +/// A message describing why the leaf private key could not be loaded as an ECDSA-P256 key pair — +/// the only way this fails is a NodeCert whose leaf is not P-256, which no dig-tls version mints. +pub fn signer_from_node_cert(cert: &dig_nat::NodeCert) -> Result { + use ring::signature::{EcdsaKeyPair, ECDSA_P256_SHA256_ASN1_SIGNING}; + + let key_der = cert.rustls_private_key(); + let rng = ring::rand::SystemRandom::new(); + let key_pair = + EcdsaKeyPair::from_pkcs8(&ECDSA_P256_SHA256_ASN1_SIGNING, key_der.secret_der(), &rng) + .map_err(|e| format!("node cert leaf key is not a usable ECDSA-P256 key pair: {e}"))?; + Ok(EcdsaHoldingsSigner::new(key_pair, cert.spki_der().to_vec())) +} + +/// Build the signed [`HoldingsAnnounce`] for one inventory delta, or `None` when nothing changed. +/// +/// `gained` content ids become `Add` deltas advertising `addresses` until +/// `now + `[`ADVERTISED_TTL_SECS`]; `lost` ids become `Remove` deltas. The announcement is signed +/// by `signer` — necessarily the node's own TLS leaf key, since the wire's `provider_peer_id` is +/// derived from that key's SPKI — so the flood is self-attributing. +/// +/// `seq` MUST strictly increase across a node's announcements: a receiver drops any announce whose +/// seq does not advance (see [`HoldingsIngress::accept`]), so a repeated seq silently stops +/// propagating. +/// +/// # Errors +/// +/// [`HoldingsError::TooManyChanges`] when the combined delta count exceeds +/// [`HOLDINGS_MAX_CHANGES`]. The batch is refused rather than truncated, so a caller can never +/// silently drop a retract — split with [`split_batches`] before calling. +pub fn announcement_for( + signer: &S, + seq: u64, + now: u64, + gained: &[ContentId], + lost: &[ContentId], + addresses: &[GossipAddr], +) -> Result, HoldingsError> { + let changes = deltas_for(now, gained, lost, addresses); + if changes.is_empty() { + return Ok(None); + } + HoldingsAnnounce::new_signed(signer, seq, now, changes).map(Some) +} + +/// The `Add`/`Remove` delta batch for an inventory change, adds first. +/// +/// Kept separate from [`announcement_for`] so the wire encoding of a delta set is unit-testable +/// without a signing key. +#[must_use] +pub fn deltas_for( + now: u64, + gained: &[ContentId], + lost: &[ContentId], + addresses: &[GossipAddr], +) -> Vec { + let expires_at = now.saturating_add(ADVERTISED_TTL_SECS); + let adds = gained.iter().map(|id| HoldingsDelta::Add { + content_key: *id.to_key().as_bytes(), + addresses: addresses.to_vec(), + expires_at, + }); + let removes = lost.iter().map(|id| HoldingsDelta::Remove { + content_key: *id.to_key().as_bytes(), + }); + adds.chain(removes).collect() +} + +/// Split a delta set into batches each within [`HOLDINGS_MAX_CHANGES`], preserving order. +/// +/// A node that caches or drops many capsules at once (a bulk warm, a cache clear) produces more +/// deltas than one signed frame may carry. Splitting keeps every delta — the alternative, +/// truncation, would drop retracts and leave this node advertising content it no longer serves. +#[must_use] +pub fn split_batches(deltas: Vec) -> Vec> { + if deltas.is_empty() { + return Vec::new(); + } + deltas + .chunks(HOLDINGS_MAX_CHANGES) + .map(<[HoldingsDelta]>::to_vec) + .collect() +} + +// --------------------------------------------------------------------------------------------- +// Ingress — the verified, rate-limited path into the local provider set +// --------------------------------------------------------------------------------------------- + +/// The two local provider-set mutations an accepted announce performs. +/// +/// A seam over [`dig_dht::DhtService`] so the ingress *policy* — verification, attribution, +/// rate limiting, replay rejection — is tested against an observable sink, with the live service +/// exercised separately by the real-wire integration test. +#[async_trait::async_trait] +pub trait HoldingsSink: Send + Sync { + /// Admit a third-party provider record whose attribution the caller has already verified. + /// + /// Returns whether the record was actually ADMITTED. The distinction matters: dig-dht may + /// legitimately refuse an ingest (over per-key or global capacity, or an already-expired + /// `expires_at`), and an ingress that counted attempts instead of admissions would report a + /// holder as discoverable when it is not — the kind of silent divergence that makes a "green" + /// flywheel test meaningless. + async fn ingest(&self, record: ProviderRecord) -> bool; + /// Remove exactly `(content_key, provider_peer_id)`. Returns whether a record was removed. + async fn remove(&self, content_key: &str, provider_peer_id: &str) -> bool; +} + +/// A live [`dig_dht::DhtService`] as a [`HoldingsSink`]. +#[async_trait::async_trait] +impl HoldingsSink for Arc { + async fn ingest(&self, record: ProviderRecord) -> bool { + matches!( + self.ingest_verified_provider(record).await, + dig_dht::provider_store::PutOutcome::Accepted + ) + } + + async fn remove(&self, content_key: &str, provider_peer_id: &str) -> bool { + self.remove_provider_record(content_key, provider_peer_id) + .await + } +} + +/// Why an inbound announcement was not applied. Every variant means **nothing** was ingested. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Rejected { + /// The announcement failed `verify_holdings_announce` (bad batch size, peer-id/SPKI mismatch, + /// unparseable key, or a signature that does not verify). + Unverified(HoldingsError), + /// The announcement is attributed to this node itself — only this node decides what it holds. + SelfAttributed, + /// `seq` did not advance beyond the highest already applied for this provider (replay or + /// out-of-order delivery). + StaleSeq { + /// The rejected announcement's seq. + seq: u64, + /// The highest seq already applied for that provider. + highest: u64, + }, + /// The transport sender exhausted its announce or delta budget for the current window. + RateLimited, + /// `announced_at` is further from now than [`IngressLimits::max_announce_age_secs`], in either + /// direction — the announcement is too old to act on, or dated too far ahead to be honest. + Stale { + /// The rejected announcement's signed `announced_at`. + announced_at: u64, + /// The receiver's clock when it was evaluated. + now: u64, + }, +} + +/// How many deltas of each kind an accepted announcement applied. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct Applied { + /// `Add` deltas ADMITTED as provider records (an ingest the DHT refused is not counted). + pub ingested: usize, + /// `Remove` deltas that removed an existing provider record. + pub removed: usize, +} + +/// Per-provider state: the highest applied `seq` plus that provider's announce bucket. +/// +/// `highest_seq` is an `Option` rather than a sentinel: a provider that has never been seen has NO +/// watermark, which is different from one whose watermark is zero. Seeding a first sighting at +/// `seq - 1` would reject a conforming implementation that starts numbering at `0`, and SPEC does not +/// require `seq >= 1`. +#[derive(Debug, Clone, Copy)] +struct ProviderState { + highest_seq: Option, + window_start: u64, + announces: u32, + last_seen: u64, +} + +/// Per-sender state: the neighbour's delta bucket. +#[derive(Debug, Clone, Copy)] +struct SenderState { + window_start: u64, + deltas: u32, + last_seen: u64, +} + +/// Every bound the ingress enforces: the two token buckets, the freshness window, and the capacity +/// of the two maps that hold the buckets themselves. +/// +/// The map capacities live here — beside the budgets rather than as bare constants read at the point +/// of use — for one reason: a bound reachable only by an unaffordable fixture is a bound that never +/// gets tested. Crossing `tracked_providers` at its production value costs 8,193 P-256 identities, so +/// a suite that could not lower it asserted the map's growth against the SENDER map instead and left +/// the provider-side eviction unverified. Every field is therefore tunable by a test, and every +/// default is the production constant. +#[derive(Debug, Clone, Copy)] +pub struct IngressLimits { + /// Applied announcements per **provider** per window. + pub announces_per_provider: u32, + /// Applied deltas per **transport sender** per window. + pub deltas_per_sender: u32, + /// Window length in seconds. + pub window_secs: u64, + /// Maximum absolute distance between `announced_at` and the receiver's clock. + pub max_announce_age_secs: u64, + /// Transport senders tracked before the least-recently-seen is evicted. + pub tracked_senders: usize, + /// Providers tracked before the least-recently-seen is evicted. + pub tracked_providers: usize, +} + +impl Default for IngressLimits { + fn default() -> Self { + Self { + announces_per_provider: MAX_ANNOUNCES_PER_PROVIDER, + deltas_per_sender: MAX_DELTAS_PER_SENDER, + window_secs: RATE_WINDOW_SECS, + max_announce_age_secs: MAX_ANNOUNCE_AGE_SECS, + tracked_senders: MAX_TRACKED_SENDERS, + tracked_providers: MAX_TRACKED_PROVIDERS, + } + } +} + +/// The receive path for opcode-222 announcements: verify, attribute, bound, then apply. +/// +/// One instance per node, shared across the inbound gossip task. Construct with [`Self::new`]. +pub struct HoldingsIngress { + /// This node's own 64-hex `peer_id`, so a self-attributed announce is dropped. + self_peer_id: String, + limits: IngressLimits, + state: tokio::sync::Mutex, +} + +#[derive(Default)] +struct IngressState { + /// Delta buckets keyed by 64-hex transport sender. + senders: HashMap, + /// Replay watermark + announce bucket keyed by 64-hex provider. + providers: HashMap, +} + +impl HoldingsIngress { + /// Build an ingress for a node whose own `peer_id` is `self_peer_id` (64-hex). + #[must_use] + pub fn new(self_peer_id: String) -> Self { + Self::with_limits(self_peer_id, IngressLimits::default()) + } + + /// Whether the real-time ingress is enabled for this process. + /// + /// `DIG_HOLDINGS_INGEST=0` (or `false`/`off`) disables it, leaving the node ANNOUNCING and + /// discoverable through the durable DHT provider records. An operator facing an announcement flood + /// needs a switch that does not require a downgrade; without one the only remedy is to stop the + /// node. + #[must_use] + pub fn ingest_enabled_from_env() -> bool { + ingest_enabled(std::env::var("DIG_HOLDINGS_INGEST").ok().as_deref()) + } + + /// [`Self::new`] with explicit bounds — used by tests to reach the budgets, the freshness window + /// and the map capacities deterministically, without minting thousands of identities. + #[must_use] + pub fn with_limits(self_peer_id: String, limits: IngressLimits) -> Self { + Self { + // Canonicalized so gate 3 compares like with like. A caller that passes an + // upper-case-spelled id would otherwise never match an inbound announcement, silently + // disabling the self-attribution gate. + self_peer_id: PeerId::from_hex(&self_peer_id).map_or(self_peer_id, |p| p.to_hex()), + limits, + state: tokio::sync::Mutex::new(IngressState::default()), + } + } + + /// How many senders and providers are currently tracked — the two capacity-bounded maps. + /// + /// Exists so the memory bound that keeps this ingress from becoming its own denial of service is + /// a TESTED guarantee rather than a documented intention. + pub async fn tracked_counts(&self) -> (usize, usize) { + let state = self.state.lock().await; + (state.senders.len(), state.providers.len()) + } + + /// Verify, bound and apply one inbound announcement. `now` is Unix seconds. + /// + /// `sender_peer_id` is the 64-hex `peer_id` of the **gossip peer the frame arrived from**, used + /// only to charge the rate budget. It deliberately does NOT have to equal the announcement's + /// provider: opcode 222 is a flood, so honest peers relay each other's announcements, and the + /// signature — not the transport — is the attribution. + /// + /// # Attribution invariant + /// + /// This method takes no provider identity from its caller. The only provider id it can pass to + /// [`HoldingsSink::remove`] or place in a [`ProviderRecord`] is + /// [`HoldingsAnnounce::provider_peer_id`], which `verify_holdings_announce` has proven equals + /// `SHA-256(provider_spki)` *and* signed this batch. A peer therefore cannot add or remove any + /// record but its own, whatever it puts in the deltas — the wire carries no per-delta peer + /// field to abuse. + /// + /// # Errors + /// + /// A [`Rejected`] describing the first failing gate; nothing is applied. + pub async fn accept( + &self, + sink: &dyn HoldingsSink, + sender_peer_id: &str, + announce: &HoldingsAnnounce, + now: u64, + ) -> Result { + // Gate 1 — authenticity. Fail-closed. + verify_holdings_announce(announce).map_err(Rejected::Unverified)?; + + // Gate 2 — CANONICALIZE the identity, once, before it is compared to or keyed on anything. + // + // This is not defensive tidying, it is a correctness gate. Hex decoding is case-INSENSITIVE + // and the signature covers the 32 DECODED bytes, so one identity has many valid spellings and + // every one of them verifies. Treating the field as an opaque `String` therefore gives an + // attacker a free bypass of any check built on it: a `String ==` against a lowercase self id + // misses `0xAB…`, and a map keyed by spelling makes each variant a fresh provider with no + // replay watermark. `PeerId::from_hex(..).to_hex()` collapses all of them to one value, and + // NOTHING below this line may read `announce.provider_peer_id` again. + let provider = PeerId::from_hex(&announce.provider_peer_id) + .ok_or(Rejected::Unverified(HoldingsError::BadPeerIdHex))?; + let provider_hex = provider.to_hex(); + + // Gate 3 — never let the network tell us what we hold. + if provider_hex == self.self_peer_id { + return Err(Rejected::SelfAttributed); + } + + // Gate 4 — bounded freshness. Evaluated before any state is touched, because a stale frame + // must cost the receiver nothing at all. See [`MAX_ANNOUNCE_AGE_SECS`] for why a `Remove` + // cannot be left to the `seq` watermark alone. + if now.abs_diff(announce.announced_at) > self.limits.max_announce_age_secs { + return Err(Rejected::Stale { + announced_at: announce.announced_at, + now, + }); + } + + // Gates 5, 6 and 7 — replay rejection and the two rate buckets, decided together under one + // lock so a concurrent flood cannot interleave two accepts past the same budget. The whole + // decision is committed before any `await` on the sink, so a replay of this same seq is + // already rejected while this batch is still being applied. + let delta_cost = u32::try_from(announce.changes.len()).unwrap_or(u32::MAX); + { + let mut state = self.state.lock().await; + state.admit( + sender_peer_id, + &provider_hex, + announce.seq, + delta_cost, + now, + &self.limits, + )?; + } + + Ok(self.apply(sink, &provider, &provider_hex, announce).await) + } + + /// Apply a verified, budgeted batch. No egress — purely local provider-set mutation. + /// + /// Takes the CANONICAL provider identity as both its typed and hex forms, so neither the ingest + /// nor the remove path can reach for the raw wire spelling. + async fn apply( + &self, + sink: &dyn HoldingsSink, + provider: &PeerId, + provider_hex: &str, + announce: &HoldingsAnnounce, + ) -> Applied { + let mut applied = Applied::default(); + for change in &announce.changes { + match change { + HoldingsDelta::Add { + content_key, + addresses, + expires_at, + } => { + if sink + .ingest(ProviderRecord::new( + &Key::from_bytes(*content_key), + provider, + bounded_dht_addresses(addresses), + *expires_at, + )) + .await + { + applied.ingested += 1; + } + } + HoldingsDelta::Remove { content_key } => { + // The provider id is the CANONICALIZED verified signer, never a caller- or + // wire-supplied value: this is what makes a retract unable to de-list an honest + // holder, and the canonical form is what makes it resolve to the signer's own + // record rather than to a spelling that matches nothing. + if sink + .remove(&Key::from_bytes(*content_key).to_hex(), provider_hex) + .await + { + applied.removed += 1; + } + } + } + } + applied + } +} + +impl IngressState { + /// The single chokepoint: decide replay and both rate buckets, and commit the charge. + /// + /// Every announcement that is applied passes through here exactly once. Nothing is charged + /// unless the announcement is admitted, so a rejected flood cannot exhaust an honest peer's + /// budget by proxy. + /// + /// # Errors + /// + /// [`Rejected::StaleSeq`] for a replayed or out-of-order announcement, or + /// [`Rejected::RateLimited`] when either bucket is exhausted. + fn admit( + &mut self, + sender: &str, + provider: &str, + seq: u64, + deltas: u32, + now: u64, + limits: &IngressLimits, + ) -> Result<(), Rejected> { + // DECIDE FIRST, ALLOCATE ONLY ON SUCCESS. + // + // Nothing below reads through a `HashMap::entry`, because inserting before the gates is what + // makes a REJECTED announcement grow the tracked set: the reject paths return early, so they + // would skip the eviction at the end and the map would follow an attacker-minted key space + // for the price of one ~180-byte frame per entry. Reading the current state into locals keeps + // every rejection allocation-free. + let current_provider = self.providers.get(provider).copied(); + let (provider_window, provider_announces, watermark) = match current_provider { + // A window that has fully elapsed resets the bucket but NEVER the watermark: replay + // protection is not a rate limit and must not lapse with one. + Some(p) if now.saturating_sub(p.window_start) >= limits.window_secs => { + (now, 0, p.highest_seq) + } + Some(p) => (p.window_start, p.announces, p.highest_seq), + None => (now, 0, None), + }; + if let Some(highest) = watermark { + if seq <= highest { + return Err(Rejected::StaleSeq { seq, highest }); + } + } + if provider_announces >= limits.announces_per_provider { + return Err(Rejected::RateLimited); + } + + let current_sender = self.senders.get(sender).copied(); + let (sender_window, sender_deltas) = match current_sender { + Some(s) if now.saturating_sub(s.window_start) >= limits.window_secs => (now, 0), + Some(s) => (s.window_start, s.deltas), + None => (now, 0), + }; + let charged = sender_deltas.saturating_add(deltas); + if charged > limits.deltas_per_sender { + return Err(Rejected::RateLimited); + } + + // Every gate passed — NOW commit both sides. + self.senders.insert( + sender.to_string(), + SenderState { + window_start: sender_window, + deltas: charged, + last_seen: now, + }, + ); + self.providers.insert( + provider.to_string(), + ProviderState { + highest_seq: Some(seq), + window_start: provider_window, + announces: provider_announces.saturating_add(1), + last_seen: now, + }, + ); + + // Evict AFTER committing so an entry just charged is never the victim of its own admission. + evict_lru(&mut self.senders, limits.tracked_senders, |s| s.last_seen); + evict_lru(&mut self.providers, limits.tracked_providers, |p| { + p.last_seen + }); + Ok(()) + } +} + +/// Drop least-recently-stamped entries until `map` holds at most `cap`, oldest first. +/// +/// Comparison allocates NOTHING: an earlier version built a `(stamp, String)` sort key, cloning a key +/// for every entry examined, on every removal, while holding the ingress lock — quadratic allocation +/// that turned a large tracked set into a wedged ingest task and a pinned worker thread. The tie-break +/// on key text is kept (it makes eviction deterministic for tests) but borrows instead of cloning, and +/// only the single chosen victim is cloned, which the borrow checker does require. +fn evict_lru(map: &mut HashMap, cap: usize, stamp: impl Fn(&V) -> u64) { + while map.len() > cap { + let Some(oldest) = map + .iter() + .min_by(|(a_key, a), (b_key, b)| stamp(a).cmp(&stamp(b)).then_with(|| a_key.cmp(b_key))) + .map(|(k, _)| k.clone()) + else { + return; + }; + map.remove(&oldest); + } +} + +/// A gossip-wire address as a dig-dht direct candidate. +/// +/// The wire carries `host`/`port` only; dig-dht additionally ranks candidates by kind, and an +/// announced address is by definition one the holder claims to serve on directly. +fn to_dht_addr(addr: &GossipAddr) -> DhtAddr { + DhtAddr::direct(addr.host.clone(), addr.port) +} + +/// Map a wire address list to DHT addresses, mapping AT MOST +/// [`MAX_ADDRESSES_PER_RECORD`](dig_dht::MAX_ADDRESSES_PER_RECORD) of them. +/// +/// The cap is applied HERE, before the map, and that placement is the whole point. +/// `ProviderRecord::new` also truncates, so the record this node ends up storing is correctly bounded +/// either way — but a wire announcement's address count is an attacker-declared `u16`, so mapping the +/// full list first would let a ~180-byte frame make this node allocate tens of thousands of owned +/// `String`s before dig-dht discarded all but eight. The observable property is therefore how many +/// addresses this function MAPS, not how many survive downstream; a test that can only see the stored +/// record cannot tell the two placements apart. +/// +/// Taking the leading prefix (rather than ranking first) loses nothing: every wire address becomes a +/// [`DhtAddr::direct`], so all candidates share one rank and dig-dht's own rank-then-truncate would +/// keep the same prefix. +fn bounded_dht_addresses(addresses: &[GossipAddr]) -> Vec { + addresses + .iter() + .take(dig_dht::MAX_ADDRESSES_PER_RECORD) + .map(to_dht_addr) + .collect() +} + +// --------------------------------------------------------------------------------------------- +// Composition — flooding our own changes, and ingesting everyone else's +// --------------------------------------------------------------------------------------------- + +/// How an announcement reaches the network. A seam over [`dig_gossip::GossipHandle`] so the +/// broadcaster's sequencing + batching are testable without a live pool. +#[async_trait::async_trait] +pub trait AnnounceTransport: Send + Sync { + /// Flood one opcode-222 frame to every connected peer. Returns how many peers it reached. + async fn flood(&self, announce: &HoldingsAnnounce) -> usize; +} + +#[async_trait::async_trait] +impl AnnounceTransport for dig_gossip::GossipHandle { + async fn flood(&self, announce: &HoldingsAnnounce) -> usize { + self.broadcast(dig_gossip::frame_holdings_announce(announce), None) + .await + .unwrap_or(0) + } +} + +/// This node's outbound side: signs and floods an opcode-222 announcement per inventory change. +/// +/// Owns the monotonic `seq` counter, because a receiver drops any announcement whose seq does not +/// advance — so exactly one place in the node may allocate them. +pub struct HoldingsBroadcaster { + signer: EcdsaHoldingsSigner, + /// Where this node serves the content it announces (signed, so no intermediary can repoint it). + addresses: Vec, + seq: std::sync::atomic::AtomicU64, +} + +impl HoldingsBroadcaster { + /// A broadcaster signing as `signer`, advertising `addresses` as where this node serves. + /// + /// `initial_seq` seeds the monotonic counter. It is deliberately a parameter rather than always + /// zero: seq is per-provider replay protection at every receiver, and a node that restarts and + /// resumes from 0 would have its announcements silently dropped by peers that still remember a + /// higher seq. Passing a clock-derived value makes a restart resume ABOVE anything already + /// announced. (Persisting the counter across restarts is #1477's durable-state work.) + #[must_use] + pub fn new(signer: EcdsaHoldingsSigner, addresses: Vec, initial_seq: u64) -> Self { + Self { + signer, + addresses, + seq: std::sync::atomic::AtomicU64::new(initial_seq), + } + } + + /// Flood the signed announcement(s) for one inventory reconcile. Returns how many frames were sent. + /// + /// A reconcile larger than one signed frame is split with [`split_batches`] rather than truncated, + /// so a bulk cache-clear cannot silently drop retracts and leave this node advertising content it + /// no longer serves. Each batch gets its own advancing seq. + pub async fn announce_change( + &self, + transport: &dyn AnnounceTransport, + gained: &[ContentId], + lost: &[ContentId], + now: u64, + ) -> usize { + let mut sent = 0; + for batch in split_batches(deltas_for(now, gained, lost, &self.addresses)) { + let seq = self + .seq + .fetch_add(1, std::sync::atomic::Ordering::Relaxed) + .saturating_add(1); + match HoldingsAnnounce::new_signed(&self.signer, seq, now, batch) { + Ok(announce) => { + let peers = transport.flood(&announce).await; + tracing::debug!( + seq, + peers, + changes = announce.changes.len(), + "dig-node holdings: flooded an opcode-222 announcement" + ); + sent += 1; + } + Err(e) => { + // Only reachable if `split_batches` ever produced an over-cap batch; logged + // rather than panicking so an inventory change can never abort the node. + tracing::warn!(error = %e, "dig-node holdings: refused to sign a batch"); + } + } + } + sent + } +} + +/// Reconcile the node's DHT provider records against `cached` AND flood the matching real-time +/// announcement — the composition that actually turns caching a capsule into being discovered. +/// +/// This is the node's ONE inventory-change reaction. Both halves take the SAME +/// [`InventoryDelta`](crate::dht::InventoryDelta), so the flood can never disagree with the provider +/// records it announces: a capsule cannot be announced as gained while its record says otherwise, and +/// a retract cannot be skipped. `holdings` is `None` on a node that cannot sign (it stays discoverable +/// through the durable records alone). +/// +/// It lives here, taking the pieces it needs rather than a `Node`, so the composition is testable +/// against a real `DhtService` — the two halves passing in isolation says nothing about the wiring +/// between them, which is the whole point of the feature. +pub async fn reconcile_and_announce( + dht: &crate::dht::DhtHandle, + cached: &[crate::CachedCapsule], + holdings: Option<(&HoldingsBroadcaster, &dyn AnnounceTransport)>, + now: u64, +) -> crate::dht::InventoryDelta { + let delta = dht.reconcile_inventory(cached).await; + if let (Some((broadcaster, transport)), false) = (holdings, delta.is_empty()) { + broadcaster + .announce_change(transport, &delta.gained, &delta.lost, now) + .await; + } + delta +} + +/// Consume inbound opcode-222 frames from the gossip pool forever, applying each through `ingress`. +/// +/// Spawned once at bring-up. Non-222 frames are ignored (other subscribers handle them), and a lagged +/// broadcast channel is tolerated: a missed announcement costs freshness, never correctness, because +/// the DHT's own republish/TTL cycle is the backstop. This loop performs NO egress — it is the reason +/// an inbound announcement cannot be amplified into outbound work. +pub async fn run_holdings_ingest( + mut inbound: tokio::sync::broadcast::Receiver<(dig_gossip::PeerId, dig_gossip::Message)>, + ingress: Arc, + sink: Arc, +) { + loop { + let (sender, msg) = match inbound.recv().await { + Ok(pair) => pair, + Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { + tracing::debug!( + skipped, + "dig-node holdings: inbound lagged; freshness only, DHT republish is the backstop" + ); + continue; + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => return, + }; + let Some(announce) = dig_gossip::holdings_announce_payload(&msg) else { + continue; // not an opcode-222 frame (or an undecodable one) + }; + // `provider_peer_id` is a `u16`-length-prefixed WIRE string: up to 65,535 bytes of arbitrary + // UTF-8, newlines and terminal escapes included. Normalising it here means no peer-supplied + // text can reach a log line, forge one, or drive a terminal escape. Logging at `debug` bounds + // the VOLUME an attacker can cause; it does nothing about CONTENT, so content is handled by + // never emitting the raw field at all. + let canonical_provider = + dig_dht::PeerId::from_hex(&announce.provider_peer_id).map(|p| p.to_hex()); + let now = now_unix_secs(); + match ingress + .accept(&sink, &hex::encode(sender.to_bytes()), &announce, now) + .await + { + Ok(applied) => tracing::debug!( + // Present whenever `accept` succeeded — it canonicalizes the same value. + provider = canonical_provider.as_deref().unwrap_or(""), + ingested = applied.ingested, + removed = applied.removed, + "dig-node holdings: applied a verified announcement" + ), + Err(reason) => tracing::debug!( + // `None` exactly when the id was not canonical hex, which is itself the diagnosis. + provider = canonical_provider.as_deref().unwrap_or(""), + ?reason, + "dig-node holdings: rejected an announcement" + ), + } + } +} + +/// Wall-clock Unix seconds — the clock dig-dht clamps provider expiries against. +#[must_use] +pub fn now_unix_secs() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| d.as_secs()) +} + +/// Whether an unset or given `DIG_HOLDINGS_INGEST` value leaves the ingress enabled. +/// +/// Split out from [`HoldingsIngress::ingest_enabled_from_env`] so the decision is a pure function of +/// its input: reading the process environment inside a test would make the kill switch's own coverage +/// depend on test execution order. +/// +/// FAIL-OPEN is deliberate and the safe direction here: an unrecognised value leaves the node +/// behaving exactly as it does today. The switch exists to let an operator SHED inbound work under a +/// flood, so a typo must never silently disable discovery instead. +fn ingest_enabled(raw: Option<&str>) -> bool { + !matches!( + raw.unwrap_or_default().trim().to_ascii_lowercase().as_str(), + "0" | "false" | "off" | "no" + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// PROPERTY: the address cap is applied BEFORE the map, so an attacker-declared count bounds the + /// work this node does rather than only the record it keeps. + /// + /// Asserted on the mapper's own output because that is the only place the two placements differ: + /// `ProviderRecord::new` truncates too, so any assertion made on the STORED record stays green + /// with this cap deleted. Pinned from both sides — one over the bound must be cut, exactly at the + /// bound must pass through whole — since a bound tested only from above cannot show it is the + /// right bound. + #[test] + fn the_address_cap_is_applied_where_the_mapping_happens() { + let cap = dig_dht::MAX_ADDRESSES_PER_RECORD; + let declared = |n: usize| -> Vec { + (0..n) + .map(|i| GossipAddr { + host: format!("::{i:x}"), + port: 9_257, + }) + .collect() + }; + + let over = bounded_dht_addresses(&declared(cap * 4)); + assert_eq!( + over.len(), + cap, + "an oversized wire list must be cut to {cap} AT THE MAPPING, not merely stored bounded" + ); + assert_eq!( + over[0].host, "::0", + "the kept addresses must be the declared prefix, in order" + ); + + assert_eq!( + bounded_dht_addresses(&declared(cap)).len(), + cap, + "a list exactly at the bound must pass through whole" + ); + assert!( + bounded_dht_addresses(&[]).is_empty(), + "an empty list maps to nothing" + ); + } + + /// PROPERTY: the operator kill switch reads as OFF only for an explicit falsey value, and + /// fail-opens on everything else including an unset variable. + #[test] + fn the_ingest_kill_switch_is_off_only_for_an_explicit_falsey_value() { + for off in ["0", "false", "off", "no", " OFF ", "False"] { + assert!( + !ingest_enabled(Some(off)), + "{off:?} must disable the ingress" + ); + } + for on in [ + None, + Some(""), + Some("1"), + Some("true"), + Some("yes"), + Some("x"), + ] { + assert!( + ingest_enabled(on), + "{on:?} must leave the ingress enabled — the switch fails OPEN" + ); + } + } +} 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 7f69ff3..86ff555 100644 --- a/crates/dig-node-core/src/seams/dig_peer/mod.rs +++ b/crates/dig-node-core/src/seams/dig_peer/mod.rs @@ -13,6 +13,7 @@ pub mod address_book; pub mod capsule_fallback; pub mod dht; +pub mod holdings; pub mod module_anchor; pub mod module_reshare; pub mod module_serve; diff --git a/crates/dig-node-core/tests/holdings_ingress.rs b/crates/dig-node-core/tests/holdings_ingress.rs new file mode 100644 index 0000000..66e95f5 --- /dev/null +++ b/crates/dig-node-core/tests/holdings_ingress.rs @@ -0,0 +1,1508 @@ +//! Adversarial tests for the opcode-222 holdings ingress (#1429) — the authenticated path into the +//! local dig-dht provider set. +//! +//! Each test names the property it pins and is built against the NEAREST WRONG implementation, not +//! merely against the correct one. Two fixture disciplines are held throughout: +//! +//! - **A truthful control actor is always present.** Every censorship/flood test keeps an honest +//! holder in the sink whose record must SURVIVE, because a test where every record is expendable +//! cannot tell "removed only the liar's record" from "removed every record for that key". +//! - **Time is pinned.** `NOW` is an explicit fixture constant threaded through every call, so no +//! assertion accidentally depends on wall-clock (an expiry passed as `100` through a wall-clock API +//! is already expired by ~1.8 billion seconds and silently exercises only the expired path). +//! +//! Numeric bounds are pinned from BOTH sides: at-bound must pass, one-over must fail. + +use std::collections::BTreeSet; +use std::sync::{Arc, Mutex}; + +use dig_dht::{ContentId, ProviderRecord}; +use dig_gossip::{ + frame_holdings_announce, CandidateAddr as GossipAddr, EcdsaHoldingsSigner, HoldingsAnnounce, + HoldingsDelta, HoldingsError, HOLDINGS_MAX_CHANGES, +}; +use dig_node_core::seams::dig_peer::holdings::{ + announcement_for, deltas_for, run_holdings_ingest, split_batches, AnnounceTransport, Applied, + HoldingsBroadcaster, HoldingsIngress, HoldingsSink, IngressLimits, Rejected, + ADVERTISED_TTL_SECS, MAX_ANNOUNCES_PER_PROVIDER, MAX_ANNOUNCE_AGE_SECS, MAX_DELTAS_PER_SENDER, + RATE_WINDOW_SECS, +}; + +/// The pinned fixture clock (Unix seconds, 2026-07-01T00:00:00Z). Never `SystemTime::now()`. +const NOW: u64 = 1_782_000_000; + +// --------------------------------------------------------------------------------------------- +// Fixtures +// --------------------------------------------------------------------------------------------- + +/// A real §5.2 peer identity: a P-256 leaf whose SPKI DER hashes to its `peer_id`. +struct TestPeer { + signer: EcdsaHoldingsSigner, + peer_id_hex: String, +} + +impl TestPeer { + fn new() -> Self { + let kp = rcgen::KeyPair::generate().expect("generate P-256 leaf key pair"); + let spki = kp.public_key_der(); + let rng = ring::rand::SystemRandom::new(); + let key_pair = ring::signature::EcdsaKeyPair::from_pkcs8( + &ring::signature::ECDSA_P256_SHA256_ASN1_SIGNING, + &kp.serialize_der(), + &rng, + ) + .expect("the generated key pair is a valid P-256 PKCS#8"); + let peer_id_hex = hex::encode(::digest(&spki)); + Self { + signer: EcdsaHoldingsSigner::new(key_pair, spki), + peer_id_hex, + } + } + + /// Sign `changes` as this peer at `seq`, dated at the fixture clock. + fn announce(&self, seq: u64, changes: Vec) -> HoldingsAnnounce { + self.announce_at(seq, NOW, changes) + } + + /// [`Self::announce`] with an explicit signed `announced_at`, for the freshness bound. + fn announce_at( + &self, + seq: u64, + announced_at: u64, + changes: Vec, + ) -> HoldingsAnnounce { + HoldingsAnnounce::new_signed(&self.signer, seq, announced_at, changes) + .expect("the fixture batch is within HOLDINGS_MAX_CHANGES") + } +} + +/// A deterministic content id from a single seed byte. +fn content(seed: u8) -> ContentId { + ContentId::capsule([seed; 32], [seed ^ 0xff; 32]) +} + +/// The 64-hex dig-dht provider-store key for a content id. +fn key_hex(id: &ContentId) -> String { + id.to_key().to_hex() +} + +fn add_delta(id: &ContentId) -> HoldingsDelta { + HoldingsDelta::Add { + content_key: *id.to_key().as_bytes(), + addresses: vec![GossipAddr { + host: "::1".to_string(), + port: 9_257, + }], + expires_at: NOW + ADVERTISED_TTL_SECS, + } +} + +fn remove_delta(id: &ContentId) -> HoldingsDelta { + HoldingsDelta::Remove { + content_key: *id.to_key().as_bytes(), + } +} + +/// An observable stand-in for the dig-dht provider store: a set of `(content_key, provider)` pairs. +/// +/// It models the ONE property the ingress must not violate — that a record is owned by a specific +/// provider — so a "remove every provider of this key" implementation is observably different from a +/// correct one. The live `DhtService` is exercised by the real-wire test instead. +#[derive(Default)] +struct RecordingSink { + records: Mutex>, +} + +impl RecordingSink { + /// Pre-seed a record so a later retract has a real victim to spare or destroy. + fn seed(&self, id: &ContentId, provider: &str) { + self.records + .lock() + .expect("sink mutex") + .insert((key_hex(id), provider.to_string())); + } + + fn holds(&self, id: &ContentId, provider: &str) -> bool { + self.records + .lock() + .expect("sink mutex") + .contains(&(key_hex(id), provider.to_string())) + } + + fn len(&self) -> usize { + self.records.lock().expect("sink mutex").len() + } +} + +#[async_trait::async_trait] +impl HoldingsSink for RecordingSink { + async fn ingest(&self, record: ProviderRecord) -> bool { + self.records + .lock() + .expect("sink mutex") + .insert((record.content_key, record.provider_peer_id)); + true + } + + async fn remove(&self, content_key: &str, provider_peer_id: &str) -> bool { + self.records + .lock() + .expect("sink mutex") + .remove(&(content_key.to_string(), provider_peer_id.to_string())) + } +} + +/// A sink that keeps every ingested [`ProviderRecord`] intact. +/// +/// [`RecordingSink`] projects each record to `(content_key, provider)`, which is the right narrowness +/// for the attribution tests but physically cannot express an address-count or expiry lie. Where the +/// property under test is about a field that projection discards, the double is WIDENED rather than the +/// assertion weakened. +#[derive(Default)] +struct RecordingRecords { + records: Mutex>, +} + +#[async_trait::async_trait] +impl HoldingsSink for RecordingRecords { + async fn ingest(&self, record: ProviderRecord) -> bool { + self.records.lock().expect("sink mutex").push(record); + true + } + + async fn remove(&self, _content_key: &str, _provider_peer_id: &str) -> bool { + false + } +} + +/// A local-only [`DhtService`]: its transport reaches nobody, so it serves purely as a real sink. +fn local_dht_service(port: u16) -> Arc { + struct Unreachable; + #[async_trait::async_trait] + impl dig_dht::DhtTransport for Unreachable { + async fn rpc( + &self, + _from: &dig_dht::Contact, + _target: &dig_dht::Contact, + _request: &dig_dht::DhtRequest, + ) -> Result { + Err(dig_dht::DhtError::Transport("unreachable".to_string())) + } + } + Arc::new(dig_dht::DhtService::new( + dig_dht::PeerId::from_bytes([0x5du8; 32]), + vec![dig_dht::CandidateAddr::direct("::1".to_string(), port)], + dig_dht::DhtConfig::default(), + Arc::new(Unreachable), + )) +} + +/// An ingress for a node whose own peer_id is `self_peer`, with the production limits. +fn ingress(self_peer: &str) -> HoldingsIngress { + HoldingsIngress::new(self_peer.to_string()) +} + +// --------------------------------------------------------------------------------------------- +// Egress +// --------------------------------------------------------------------------------------------- + +/// PROPERTY: an inventory delta becomes adds-then-removes over the DHT content KEYS (not the raw +/// content ids), with the advertised expiry anchored to the caller's clock. +/// +/// Nearest wrong implementation: encoding `store_id`/`root` bytes directly instead of +/// `ContentId::to_key()`. The fixture uses ids whose key differs from every constituent byte array, +/// so the wrong encoding cannot coincide with the right one. +#[test] +fn deltas_encode_dht_content_keys_and_the_pinned_expiry() { + let gained = [content(1), content(2)]; + let lost = [content(3)]; + let addrs = [GossipAddr { + host: "::1".to_string(), + port: 9_257, + }]; + + let deltas = deltas_for(NOW, &gained, &lost, &addrs); + + assert_eq!(deltas.len(), 3, "two adds then one remove"); + match &deltas[0] { + HoldingsDelta::Add { + content_key, + addresses, + expires_at, + } => { + assert_eq!( + content_key, + gained[0].to_key().as_bytes(), + "the wire carries the DHT content KEY, not the store/root bytes" + ); + assert_eq!( + addresses, &addrs, + "the announced addresses are signed as-is" + ); + assert_eq!( + *expires_at, + NOW + ADVERTISED_TTL_SECS, + "expiry is anchored to the caller's clock, not wall-clock" + ); + } + other => panic!("expected the first delta to be an Add, got {other:?}"), + } + assert_eq!( + deltas[2], + remove_delta(&lost[0]), + "a lost id becomes a Remove for the same content key" + ); +} + +/// PROPERTY: nothing to announce produces no announcement (so an idle node floods nothing). +#[test] +fn an_empty_inventory_delta_produces_no_announcement() { + let peer = TestPeer::new(); + let none = announcement_for(&peer.signer, 1, NOW, &[], &[], &[]) + .expect("an empty batch is not an error"); + assert!(none.is_none(), "an idle node must not flood an empty frame"); +} + +/// PROPERTY: a batch larger than the protocol's own cap is REFUSED, never truncated — truncation +/// would silently drop a retract and leave the node advertising content it no longer serves. +/// +/// The fixture size is taken from the protocol limit (`HOLDINGS_MAX_CHANGES + 1`), and the bound is +/// pinned from both sides: exactly at the cap must succeed. +#[test] +fn an_oversized_batch_is_refused_and_the_cap_itself_is_accepted() { + let peer = TestPeer::new(); + let at_cap: Vec<_> = (0..HOLDINGS_MAX_CHANGES) + .map(|i| add_delta(&content(u8::try_from(i % 251).unwrap_or(0)))) + .collect(); + assert!( + HoldingsAnnounce::new_signed(&peer.signer, 1, NOW, at_cap.clone()).is_ok(), + "a batch of exactly HOLDINGS_MAX_CHANGES must be accepted (at-bound passes)" + ); + + let mut one_over = at_cap; + one_over.push(remove_delta(&content(200))); + assert!( + matches!( + HoldingsAnnounce::new_signed(&peer.signer, 1, NOW, one_over), + Err(HoldingsError::TooManyChanges { .. }) + ), + "one delta over the cap must be refused, not truncated (one-over fails)" + ); +} + +/// PROPERTY: splitting preserves EVERY delta and its order across batch boundaries. +/// +/// Nearest wrong implementation: truncating to the first batch, which would silently drop retracts. +/// The fixture is `2 × HOLDINGS_MAX_CHANGES + 1` deltas — sized from the protocol limit so it +/// straddles two boundaries rather than one. +#[test] +fn splitting_preserves_every_delta_and_its_order() { + let total = 2 * HOLDINGS_MAX_CHANGES + 1; + let deltas: Vec<_> = (0..total) + .map(|i| remove_delta(&content(u8::try_from(i % 251).unwrap_or(0)))) + .collect(); + + let batches = split_batches(deltas.clone()); + + assert_eq!(batches.len(), 3, "257 deltas straddle two batch boundaries"); + assert!( + batches.iter().all(|b| b.len() <= HOLDINGS_MAX_CHANGES), + "no batch may exceed the protocol cap" + ); + let flattened: Vec<_> = batches.into_iter().flatten().collect(); + assert_eq!( + flattened, deltas, + "splitting must lose no delta and reorder none" + ); + assert!( + split_batches(Vec::new()).is_empty(), + "nothing to say produces no batch" + ); +} + +// --------------------------------------------------------------------------------------------- +// Ingress — authenticity + attribution +// --------------------------------------------------------------------------------------------- + +/// PROPERTY: a verified add is attributed to the SIGNER, and to the signer alone. +/// +/// Nearest wrong implementation: attributing the record to the transport sender. The fixture makes +/// sender and provider DIFFERENT peers (a relayed flood, the normal case), so the two cannot coincide. +#[tokio::test] +async fn a_verified_add_is_attributed_to_the_signer_not_the_relaying_sender() { + let holder = TestPeer::new(); + let relayer = TestPeer::new(); + let us = TestPeer::new(); + let sink = RecordingSink::default(); + let id = content(7); + + let applied = ingress(&us.peer_id_hex) + .accept( + &sink, + &relayer.peer_id_hex, + &holder.announce(1, vec![add_delta(&id)]), + NOW, + ) + .await + .expect("a correctly signed announce relayed by a third peer is accepted"); + + assert_eq!( + applied, + Applied { + ingested: 1, + removed: 0 + } + ); + assert!( + sink.holds(&id, &holder.peer_id_hex), + "the record must name the SIGNER as the holder" + ); + assert!( + !sink.holds(&id, &relayer.peer_id_hex), + "the relaying peer must never be recorded as a holder of content it only forwarded" + ); +} + +/// PROPERTY (H2, the censorship gate): a retract can only ever remove the SIGNER's own record. +/// +/// This is the test the whole attribution invariant rests on. The fixture keeps a truthful control +/// holder — `honest` also provides the same content key — so the assertion can distinguish +/// "removed only the liar's record" from "removed every record for that key". A single-holder +/// fixture would pass under both implementations and prove nothing. +#[tokio::test] +async fn a_retract_cannot_delist_another_peers_record() { + let attacker = TestPeer::new(); + let honest = TestPeer::new(); + let us = TestPeer::new(); + let sink = RecordingSink::default(); + let id = content(9); + + // Two holders of the SAME key: the honest control, and the attacker itself. + sink.seed(&id, &honest.peer_id_hex); + sink.seed(&id, &attacker.peer_id_hex); + + let applied = ingress(&us.peer_id_hex) + .accept( + &sink, + &attacker.peer_id_hex, + &attacker.announce(1, vec![remove_delta(&id)]), + NOW, + ) + .await + .expect( + "a validly signed retract is accepted — it just cannot reach another peer's record", + ); + + assert_eq!( + applied, + Applied { + ingested: 0, + removed: 1 + }, + "exactly one record — the attacker's own — is removed" + ); + assert!( + sink.holds(&id, &honest.peer_id_hex), + "the honest holder's record MUST survive an attacker's retract for the same content key" + ); + assert!( + !sink.holds(&id, &attacker.peer_id_hex), + "the signer's own record is the one that goes" + ); +} + +/// PROPERTY: a forged signature is rejected fail-closed and mutates nothing. +/// +/// Nearest wrong implementation: verifying but ingesting anyway (logging the error). The sink is +/// seeded with a control record so "nothing changed" is observable rather than trivially true. +#[tokio::test] +async fn a_forged_signature_is_rejected_and_mutates_nothing() { + let holder = TestPeer::new(); + let honest = TestPeer::new(); + let us = TestPeer::new(); + let sink = RecordingSink::default(); + let id = content(11); + sink.seed(&id, &honest.peer_id_hex); + + // A validly-formed announce whose signature has one flipped bit — the cheapest possible forgery. + let mut forged = holder.announce(1, vec![add_delta(&id), remove_delta(&id)]); + let last = forged.signature.len() - 1; + forged.signature[last] ^= 0x01; + + let rejected = ingress(&us.peer_id_hex) + .accept(&sink, &holder.peer_id_hex, &forged, NOW) + .await + .expect_err("a forged signature must be rejected"); + + assert_eq!( + rejected, + Rejected::Unverified(HoldingsError::InvalidSignature) + ); + assert_eq!(sink.len(), 1, "no record was added or removed"); + assert!(sink.holds(&id, &honest.peer_id_hex)); +} + +/// PROPERTY: an announce whose carried `peer_id` does not hash from its carried SPKI is rejected — +/// the impersonation attempt, where an attacker names a victim as the provider. +#[tokio::test] +async fn an_announce_claiming_another_peers_id_is_rejected() { + let attacker = TestPeer::new(); + let victim = TestPeer::new(); + let us = TestPeer::new(); + let sink = RecordingSink::default(); + let id = content(13); + + // The attacker signs with its OWN key but claims to be the victim. + let mut impersonating = attacker.announce(1, vec![add_delta(&id)]); + impersonating.provider_peer_id = victim.peer_id_hex.clone(); + + let rejected = ingress(&us.peer_id_hex) + .accept(&sink, &attacker.peer_id_hex, &impersonating, NOW) + .await + .expect_err("a peer_id that does not hash from the carried SPKI must be rejected"); + + assert_eq!( + rejected, + Rejected::Unverified(HoldingsError::PeerIdMismatch) + ); + assert_eq!(sink.len(), 0, "the victim was not named as a holder"); +} + +/// PROPERTY: the network cannot tell this node what it holds. +/// +/// An attacker who replays our own signed announce back at us must not be able to drive our local +/// provider set. The fixture uses OUR OWN valid signature, so only the self-attribution gate can +/// reject it — every authenticity check passes. +#[tokio::test] +async fn our_own_announce_replayed_back_at_us_is_ignored() { + let us = TestPeer::new(); + let attacker = TestPeer::new(); + let sink = RecordingSink::default(); + let id = content(17); + + let rejected = ingress(&us.peer_id_hex) + .accept( + &sink, + &attacker.peer_id_hex, + &us.announce(1, vec![remove_delta(&id)]), + NOW, + ) + .await + .expect_err("an announce attributed to this node itself must be ignored"); + + assert_eq!(rejected, Rejected::SelfAttributed); + assert_eq!(sink.len(), 0); +} + +// --------------------------------------------------------------------------------------------- +// Ingress — replay + rate bounds +// --------------------------------------------------------------------------------------------- + +/// PROPERTY: an announcement whose `seq` does not ADVANCE is dropped, so a captured frame cannot +/// resurrect a record its provider has since retracted. +/// +/// The fixture replays the earlier ADD after the later REMOVE — the attack that actually matters. +/// A test that merely replayed the same frame twice could pass under an implementation that +/// deduplicates on bytes while still accepting an older seq. +#[tokio::test] +async fn a_replayed_older_seq_cannot_resurrect_a_retracted_record() { + let holder = TestPeer::new(); + let us = TestPeer::new(); + let sink = RecordingSink::default(); + let id = content(19); + let ingress = ingress(&us.peer_id_hex); + + let add = holder.announce(5, vec![add_delta(&id)]); + let retract = holder.announce(6, vec![remove_delta(&id)]); + + ingress + .accept(&sink, &holder.peer_id_hex, &add, NOW) + .await + .expect("seq 5 is the provider's first announce"); + ingress + .accept(&sink, &holder.peer_id_hex, &retract, NOW) + .await + .expect("seq 6 advances"); + assert!(!sink.holds(&id, &holder.peer_id_hex), "the retract applied"); + + let rejected = ingress + .accept(&sink, &holder.peer_id_hex, &add, NOW) + .await + .expect_err("replaying the older ADD must be dropped"); + + assert_eq!(rejected, Rejected::StaleSeq { seq: 5, highest: 6 }); + assert!( + !sink.holds(&id, &holder.peer_id_hex), + "the retracted record must STAY retracted" + ); +} + +/// PROPERTY: the per-provider announce budget bounds how often one holder may revise its holdings. +/// +/// Pinned from both sides: the `MAX_ANNOUNCES_PER_PROVIDER`-th announcement in a window is accepted, +/// the next is refused. Each announcement carries ONE delta so the sender delta bucket cannot be the +/// thing that fires — otherwise this test would pass while the provider bucket did nothing. +#[tokio::test] +async fn the_per_provider_announce_budget_binds_at_its_stated_bound() { + let holder = TestPeer::new(); + let us = TestPeer::new(); + let sink = RecordingSink::default(); + let ingress = ingress(&us.peer_id_hex); + + for i in 0..MAX_ANNOUNCES_PER_PROVIDER { + let seq = u64::from(i) + 1; + ingress + .accept( + &sink, + &holder.peer_id_hex, + &holder.announce(seq, vec![add_delta(&content(u8::try_from(i).unwrap_or(0)))]), + NOW, + ) + .await + .unwrap_or_else(|e| panic!("announce {seq} is at or under the bound, got {e:?}")); + } + + let over = holder.announce( + u64::from(MAX_ANNOUNCES_PER_PROVIDER) + 1, + vec![add_delta(&content(250))], + ); + assert_eq!( + ingress + .accept(&sink, &holder.peer_id_hex, &over, NOW) + .await + .expect_err("one announcement over the bound must be refused"), + Rejected::RateLimited + ); + + // The budget is a WINDOW, not a lifetime cap — an honest long-lived holder must recover. + ingress + .accept(&sink, &holder.peer_id_hex, &over, NOW + RATE_WINDOW_SECS) + .await + .expect("the budget refills after the window elapses"); +} + +/// PROPERTY: the per-sender DELTA budget bounds total ingest work a neighbour can cause, INDEPENDENTLY +/// of the per-provider announce budget. +/// +/// This is the guard that makes the two-bucket design non-redundant, so the fixture is built so the +/// provider bucket CANNOT be what fires: each maximal batch comes from a DIFFERENT provider, so every +/// provider bucket is at 1 of 10 while the sender's delta bucket fills. Batch size is taken from the +/// protocol limit (`HOLDINGS_MAX_CHANGES`), so the bound is expressed in the units the wire allows. +#[tokio::test] +async fn the_per_sender_delta_budget_binds_independently_of_the_provider_budget() { + let relayer = TestPeer::new(); + let us = TestPeer::new(); + let sink = RecordingSink::default(); + let ingress = ingress(&us.peer_id_hex); + + let maximal: Vec<_> = (0..HOLDINGS_MAX_CHANGES) + .map(|i| add_delta(&content(u8::try_from(i % 251).unwrap_or(0)))) + .collect(); + let batches_at_bound = MAX_DELTAS_PER_SENDER / HOLDINGS_MAX_CHANGES as u32; + + for i in 0..batches_at_bound { + let holder = TestPeer::new(); // a fresh provider each time — provider bucket stays at 1/10 + ingress + .accept( + &sink, + &relayer.peer_id_hex, + &holder.announce(1, maximal.clone()), + NOW, + ) + .await + .unwrap_or_else(|e| panic!("maximal batch {i} is at or under the bound, got {e:?}")); + } + + let fresh_holder = TestPeer::new(); + assert_eq!( + ingress + .accept( + &sink, + &relayer.peer_id_hex, + &fresh_holder.announce(1, maximal.clone()), + NOW, + ) + .await + .expect_err( + "one maximal batch over the sender's delta budget must be refused even though \ + this provider has never announced before" + ), + Rejected::RateLimited + ); + + // A DIFFERENT neighbour is unaffected — the budget is per sender, not global, so one abusive + // neighbour cannot silence the rest of the network. + let other_relayer = TestPeer::new(); + ingress + .accept( + &sink, + &other_relayer.peer_id_hex, + &fresh_holder.announce(1, maximal), + NOW, + ) + .await + .expect("a different neighbour has its own budget"); +} + +/// PROPERTY: a rejected announcement charges NOTHING, so a flood of invalid frames cannot exhaust +/// the budget an honest announcement needs. +/// +/// Nearest wrong implementation: charging the bucket before verifying — which turns the rate limiter +/// itself into the denial-of-service (one bad signature per honest announce silences a neighbour). +#[tokio::test] +async fn rejected_announcements_do_not_consume_the_budget() { + let holder = TestPeer::new(); + let relayer = TestPeer::new(); + let us = TestPeer::new(); + let sink = RecordingSink::default(); + let ingress = ingress(&us.peer_id_hex); + let id = content(23); + + // Far more forgeries than either budget would allow, all from the same neighbour. + let mut forged = holder.announce(1, vec![add_delta(&id)]); + forged.signature[0] ^= 0xff; + for _ in 0..(MAX_ANNOUNCES_PER_PROVIDER * 5) { + assert!(matches!( + ingress + .accept(&sink, &relayer.peer_id_hex, &forged, NOW) + .await, + Err(Rejected::Unverified(_)) + )); + } + + ingress + .accept( + &sink, + &relayer.peer_id_hex, + &holder.announce(1, vec![add_delta(&id)]), + NOW, + ) + .await + .expect("an honest announcement still fits after a forgery flood"); + assert!(sink.holds(&id, &holder.peer_id_hex)); +} + +// --------------------------------------------------------------------------------------------- +// Egress composition — the broadcaster +// --------------------------------------------------------------------------------------------- + +/// Records every announcement handed to the transport, in order. +#[derive(Default)] +struct RecordingTransport { + sent: Mutex>, +} + +#[async_trait::async_trait] +impl AnnounceTransport for RecordingTransport { + async fn flood(&self, announce: &HoldingsAnnounce) -> usize { + self.sent + .lock() + .expect("transport mutex") + .push(announce.clone()); + 1 + } +} + +/// PROPERTY: a reconcile too large for one signed frame is SPLIT across frames with strictly +/// advancing `seq`, and every delta is flooded — never truncated. +/// +/// This is the end-to-end version of the split property, and the two halves are load-bearing +/// together: a receiver drops any announcement whose seq does not advance, so a split that reused one +/// seq would have every frame after the first silently discarded — the same data loss as truncation, +/// just further downstream. The fixture is sized from the protocol limit +/// (`HOLDINGS_MAX_CHANGES + 1` gains) so it straddles a boundary by exactly one delta. +#[tokio::test] +async fn an_oversized_reconcile_floods_every_delta_across_advancing_seqs() { + let transport = RecordingTransport::default(); + let broadcaster = HoldingsBroadcaster::new( + TestPeer::new().signer, + vec![GossipAddr { + host: "::1".to_string(), + port: 9_257, + }], + 0, + ); + + let gained: Vec<_> = (0..=HOLDINGS_MAX_CHANGES) + .map(|i| content(u8::try_from(i % 251).unwrap_or(0))) + .collect(); + let frames = broadcaster + .announce_change(&transport, &gained, &[], NOW) + .await; + + assert_eq!(frames, 2, "257 deltas need two frames"); + let sent = transport.sent.lock().expect("transport mutex"); + assert_eq!(sent.len(), 2); + assert!( + sent[1].seq > sent[0].seq, + "each frame must carry a strictly advancing seq, or a receiver drops all but the first: \ + got {} then {}", + sent[0].seq, + sent[1].seq + ); + let total: usize = sent.iter().map(|a| a.changes.len()).sum(); + assert_eq!( + total, + gained.len(), + "every delta must be flooded — a split must not lose one" + ); +} + +/// PROPERTY: nothing changed means nothing is flooded (an idle node is silent on the wire). +#[tokio::test] +async fn an_empty_reconcile_floods_nothing() { + let transport = RecordingTransport::default(); + let broadcaster = HoldingsBroadcaster::new(TestPeer::new().signer, Vec::new(), 7); + + assert_eq!( + broadcaster.announce_change(&transport, &[], &[], NOW).await, + 0 + ); + assert!(transport.sent.lock().expect("mutex").is_empty()); +} + +// --------------------------------------------------------------------------------------------- +// Ingress — hex CASE MALLEABILITY (the gate-2 / gate-3 bypass) +// --------------------------------------------------------------------------------------------- +// +// `hex::decode` is case-INSENSITIVE and dig-gossip signs over the 32 DECODED bytes, so uppercasing +// any hex digit of `provider_peer_id` yields a still-valid signature for the same identity. Every +// comparison or map key that treats the field as an opaque `String` therefore has many spellings of +// one peer, and each spelling is a free bypass. These are exploit regressions, not unit tests of a +// helper: each replays a REAL signed announcement with its identity merely re-spelled. + +/// Re-spell an announcement's `provider_peer_id` in upper case. The signature still verifies, because +/// it covers the decoded bytes rather than this text. +fn uppercase_provider(announce: &HoldingsAnnounce) -> HoldingsAnnounce { + let mut respelled = announce.clone(); + respelled.provider_peer_id = respelled.provider_peer_id.to_uppercase(); + respelled +} + +/// EXPLOIT (gate 2): our own announcement, replayed back at us with its identity uppercased, must +/// STILL be recognised as ours. +/// +/// A `String ==` against a lowercase `self_peer_id` misses every case variant, letting the network +/// drive this node's own provider set — the exact thing gate 2 exists to prevent. +#[tokio::test] +async fn an_uppercased_replay_of_our_own_announce_is_still_self_attributed() { + let us = TestPeer::new(); + let attacker = TestPeer::new(); + let sink = RecordingSink::default(); + let ours = us.announce(1, vec![add_delta(&content(31))]); + + let rejected = ingress(&us.peer_id_hex) + .accept( + &sink, + &attacker.peer_id_hex, + &uppercase_provider(&ours), + NOW, + ) + .await + .expect_err("a case-respelled replay of OUR OWN announce must be self-attributed"); + + assert_eq!(rejected, Rejected::SelfAttributed); + assert_eq!(sink.len(), 0, "the network must not drive our own holdings"); +} + +/// EXPLOIT (gate 3): a case-respelled replay of an older announcement must NOT resurrect a record +/// its provider has since retracted. +/// +/// Keying the replay watermark by hex SPELLING makes each variant a fresh provider seeded below its +/// own seq, so the stale frame is admitted; the Add path then normalises the id and writes to the +/// CANONICAL record. This also silently undoes the active-retract fix: a holder that evicted a +/// capsule is re-listed as a holder of content it cannot serve for the remaining TTL. +#[tokio::test] +async fn a_case_respelled_replay_cannot_resurrect_a_retracted_record() { + let holder = TestPeer::new(); + let us = TestPeer::new(); + let sink = RecordingSink::default(); + let id = content(37); + let ingress = ingress(&us.peer_id_hex); + + let add = holder.announce(5, vec![add_delta(&id)]); + ingress + .accept(&sink, &holder.peer_id_hex, &add, NOW) + .await + .expect("seq 5 add"); + ingress + .accept( + &sink, + &holder.peer_id_hex, + &holder.announce(6, vec![remove_delta(&id)]), + NOW, + ) + .await + .expect("seq 6 retract"); + assert!(!sink.holds(&id, &holder.peer_id_hex), "the retract applied"); + + let rejected = ingress + .accept(&sink, &holder.peer_id_hex, &uppercase_provider(&add), NOW) + .await + .expect_err("a case-respelled stale seq must still be stale"); + + assert_eq!(rejected, Rejected::StaleSeq { seq: 5, highest: 6 }); + assert!( + !sink.holds(&id, &holder.peer_id_hex), + "a retracted record must STAY retracted under any spelling of the provider id" + ); +} + +/// EXPLOIT (attribution): a case-respelled RETRACT must still resolve to the signer's own CANONICAL +/// record, and must not reach another holder's. +/// +/// The truthful control holder must survive; and the attacker's own record — stored canonically by +/// the Add path — must be the one removed, which only happens if the remove argument is normalised. +#[tokio::test] +async fn a_case_respelled_retract_resolves_to_the_signers_canonical_record() { + let attacker = TestPeer::new(); + let honest = TestPeer::new(); + let us = TestPeer::new(); + let sink = RecordingSink::default(); + let id = content(41); + sink.seed(&id, &honest.peer_id_hex); + sink.seed(&id, &attacker.peer_id_hex); + + let applied = ingress(&us.peer_id_hex) + .accept( + &sink, + &attacker.peer_id_hex, + &uppercase_provider(&attacker.announce(1, vec![remove_delta(&id)])), + NOW, + ) + .await + .expect("a validly signed retract is accepted"); + + assert_eq!( + applied.removed, 1, + "the retract must resolve to the signer's CANONICAL record, not a case variant that \ + matches nothing" + ); + assert!( + sink.holds(&id, &honest.peer_id_hex), + "the honest holder must survive a case-respelled retract" + ); + assert!(!sink.holds(&id, &attacker.peer_id_hex)); +} + +/// PROPERTY: a provider id that is not canonical 64-hex is refused before anything else, so no +/// downstream comparison, map key or log ever sees attacker-shaped text. +/// +/// The field is a `u16`-length-prefixed wire string, so it may carry tens of kilobytes of arbitrary +/// UTF-8 including newlines and terminal escapes. +#[tokio::test] +async fn a_non_canonical_provider_id_is_refused_outright() { + let holder = TestPeer::new(); + let us = TestPeer::new(); + let sink = RecordingSink::default(); + + let mut hostile = holder.announce(1, vec![add_delta(&content(53))]); + hostile.provider_peer_id = "\n\u{1b}[31mFORGED LOG LINE ".repeat(64); + + let rejected = ingress(&us.peer_id_hex) + .accept(&sink, &holder.peer_id_hex, &hostile, NOW) + .await + .expect_err("a non-hex provider id must be refused"); + + assert!( + matches!(rejected, Rejected::Unverified(_)), + "expected a verification rejection, got {rejected:?}" + ); + assert_eq!(sink.len(), 0); +} + +// --------------------------------------------------------------------------------------------- +// Ingress — bounded FRESHNESS (a captured Remove must not replay forever) +// --------------------------------------------------------------------------------------------- + +/// EXPLOIT (censorship across a restart): a captured retract, replayed at a node whose replay +/// watermark is empty, must NOT de-list an honest holder. +/// +/// `HoldingsDelta::Remove` carries no expiry, so without a freshness check the ONLY barrier to an +/// indefinite replay is the in-memory per-provider watermark — which a fresh process does not have. +/// A victim restart (or a capacity eviction of its watermark) would therefore hand an attacker a free +/// de-listing of an honest peer: censorship, not the "bounded staleness" this module used to claim. +/// The fixture models the restart as a FRESH ingress; `announced_at` is signed, so the captured +/// frame's age cannot be rewritten. +#[tokio::test] +async fn a_captured_retract_replayed_after_a_restart_cannot_delist_an_honest_holder() { + let honest = TestPeer::new(); + let attacker = TestPeer::new(); + let us = TestPeer::new(); + let sink = RecordingSink::default(); + let id = content(43); + + // The honest holder's own signed retract, captured off the wire a day earlier. + let captured = honest.announce(9, vec![remove_delta(&id)]); + // ... and since then the honest holder is serving the capsule again. + sink.seed(&id, &honest.peer_id_hex); + + // A FRESH ingress: the process restarted, so nothing remembers seq 9. + let rejected = ingress(&us.peer_id_hex) + .accept(&sink, &attacker.peer_id_hex, &captured, NOW + 86_400) + .await + .expect_err("a day-old captured retract must be refused on freshness alone"); + + assert!( + matches!(rejected, Rejected::Stale { .. }), + "expected a freshness rejection, got {rejected:?}" + ); + assert!( + sink.holds(&id, &honest.peer_id_hex), + "an honest holder must NOT be de-listable by replaying its own old retract" + ); +} + +/// PROPERTY: freshness is bounded on BOTH sides, and AT the bound it passes. +/// +/// A clock skew inside the window must not reject honest announcements; a future-dated frame must be +/// refused too, or an attacker could mint a retract that stays replayable long after it was captured. +#[tokio::test] +async fn the_freshness_window_binds_on_both_sides_of_its_bound() { + let holder = TestPeer::new(); + let us = TestPeer::new(); + let window = IngressLimits::default().max_announce_age_secs; + + for (label, at_now) in [ + ("at the bound, frame in the past", NOW + window), + ("at the bound, frame in the future", NOW - window), + ] { + let sink = RecordingSink::default(); + ingress(&us.peer_id_hex) + .accept( + &sink, + &holder.peer_id_hex, + &holder.announce(1, vec![add_delta(&content(47))]), + at_now, + ) + .await + .unwrap_or_else(|e| panic!("{label} must be accepted (at-bound passes), got {e:?}")); + } + + for (label, at_now) in [ + ("one second past the bound, in the past", NOW + window + 1), + ("one second past the bound, in the future", NOW - window - 1), + ] { + let sink = RecordingSink::default(); + let outcome = ingress(&us.peer_id_hex) + .accept( + &sink, + &holder.peer_id_hex, + &holder.announce(1, vec![add_delta(&content(47))]), + at_now, + ) + .await; + assert!( + matches!(outcome, Err(Rejected::Stale { .. })), + "{label} must be refused (one-over fails), got {outcome:?}" + ); + } +} + +// --------------------------------------------------------------------------------------------- +// Ingress — rejections must not allocate tracking state +// --------------------------------------------------------------------------------------------- + +/// PROPERTY: a REJECTED announcement leaves NO tracking state behind. +/// +/// This replaces an earlier test that drove only the sender map through ACCEPTED announcements and +/// asserted `providers == 1` — a false green, because it never entered a reject path, so deleting the +/// provider-map eviction kept it passing. The leak this pins is the real one: an entry allocated +/// BEFORE the gates, on a path that returns early and therefore skips eviction, is unbounded growth +/// bought for ~180 wire bytes per entry. +#[tokio::test] +async fn rejected_announcements_allocate_no_tracking_state() { + let us = TestPeer::new(); + let relayer = TestPeer::new(); + let sink = RecordingSink::default(); + let ingress = ingress(&us.peer_id_hex); + + // Exhaust the relaying sender's DELTA budget with maximal batches from a few real providers. + // Every announcement after this is rejected on the sender bucket — which is decided AFTER the + // provider entry would be allocated, so it is precisely the leak path. + let maximal: Vec<_> = (0..HOLDINGS_MAX_CHANGES) + .map(|i| add_delta(&content(u8::try_from(i % 251).unwrap_or(0)))) + .collect(); + let admitted = MAX_DELTAS_PER_SENDER / HOLDINGS_MAX_CHANGES as u32; + for _ in 0..admitted { + let holder = TestPeer::new(); + ingress + .accept( + &sink, + &relayer.peer_id_hex, + &holder.announce(1, maximal.clone()), + NOW, + ) + .await + .expect("a maximal batch within the sender budget is admitted"); + } + let (_, tracked_after_admits) = ingress.tracked_counts().await; + + // 50 announcements from 50 DISTINCT, freshly-minted providers — the attacker-chosen key space this + // map must not follow — all refused on the exhausted sender budget. + for i in 0..50u8 { + let stranger = TestPeer::new(); + let refused = ingress + .accept( + &sink, + &relayer.peer_id_hex, + &stranger.announce(1, vec![add_delta(&content(i))]), + NOW, + ) + .await; + assert!( + matches!(refused, Err(Rejected::RateLimited)), + "the sender budget is exhausted, so this must be refused; got {refused:?}" + ); + } + + let (senders, providers) = ingress.tracked_counts().await; + assert_eq!( + providers, tracked_after_admits, + "50 REJECTED announcements from 50 distinct providers must leave the tracked set UNCHANGED; \ + an entry allocated before the gates would show {} extra", + 50 + ); + assert_eq!(senders, 1, "one relaying sender throughout"); +} + +/// PROPERTY: the PROVIDER map is capacity-bounded, and the entry it drops at the bound is the +/// LEAST-RECENTLY-SEEN one. +/// +/// This is the guard the previous round's `the_provider_tracking_map_is_capacity_bounded` claimed and +/// did not test: that test drove the SENDER map and closed on `providers == 1`, so deleting the +/// provider-side `evict_lru` kept it green. Two fixture choices make this one able to see the guard: +/// +/// - **A reachable cap.** [`IngressLimits::tracked_providers`] is parameterised, so the bound is +/// crossed with four P-256 identities instead of 8,193. A bound that can only be reached by an +/// unaffordable fixture is a bound that never gets tested. +/// - **A second observable besides the count.** A count alone cannot distinguish LRU eviction from +/// evicting an arbitrary entry — or from evicting the entry just admitted. Eviction is therefore +/// observed through its CONSEQUENCE: losing an entry loses that provider's replay watermark, so the +/// victim's already-applied `seq` becomes admissible again while a retained provider's stays +/// `StaleSeq`. The same replay is asserted to be REFUSED before the bound is crossed, so the later +/// admission is attributable to the eviction and to nothing else. +/// +/// A rejected announcement from a fifth, never-seen provider sits in the middle of the fixture: it +/// must neither grow the map nor evict anybody, which is the composition of this bound with the +/// allocate-nothing-on-reject rule above. +#[tokio::test] +async fn the_provider_map_evicts_the_least_recently_seen_at_its_capacity() { + const CAP: usize = 3; + let us = TestPeer::new(); + let relayer = TestPeer::new(); + let sink = RecordingSink::default(); + let ingress = HoldingsIngress::with_limits( + us.peer_id_hex.clone(), + IngressLimits { + tracked_providers: CAP, + ..IngressLimits::default() + }, + ); + + // Four holders, each seen at a distinct second so the least-recently-seen order is unambiguous. + // All four stamps stay well inside MAX_ANNOUNCE_AGE_SECS of the announcements' `announced_at`. + let holders: Vec = (0..4).map(|_| TestPeer::new()).collect(); + let seen_at = |i: usize| NOW + i as u64; + let admit = |i: usize| { + let announce = holders[i].announce(7, vec![add_delta(&content(70 + i as u8))]); + let ingress = &ingress; + let sink = &sink; + let sender = &relayer.peer_id_hex; + let provider = &holders[i].peer_id_hex; + async move { + ingress + .accept(sink, sender, &announce, seen_at(i)) + .await + .unwrap_or_else(|e| panic!("holder {i} ({provider}) must be admitted, got {e:?}")); + } + }; + // A replay of holder `i`'s seq-7 announcement, judged at the clock of the last admitted holder. + let replay = |i: usize, at: u64| { + let announce = holders[i].announce(7, vec![add_delta(&content(70 + i as u8))]); + let ingress = &ingress; + let sink = &sink; + let sender = &relayer.peer_id_hex; + async move { ingress.accept(sink, sender, &announce, at).await } + }; + + for i in 0..CAP { + admit(i).await; + } + let (_, at_cap) = ingress.tracked_counts().await; + assert_eq!( + at_cap, CAP, + "exactly at the capacity nothing may be evicted — a bound tested only from above cannot \ + show it is the RIGHT bound" + ); + assert_eq!( + replay(0, seen_at(CAP)).await, + Err(Rejected::StaleSeq { + seq: 7, + highest: 7 + }), + "before the bound is crossed the oldest provider still holds its watermark; without this the \ + admission asserted below would not be attributable to eviction" + ); + + // A rejected announcement in the mix: a never-seen provider whose frame is too old to act on. It + // is judged at the live clock, so it cannot disturb the eviction order it must not affect. + let stranger = TestPeer::new(); + let stale = stranger.announce_at( + 1, + NOW - MAX_ANNOUNCE_AGE_SECS - 1, + vec![add_delta(&content(99))], + ); + assert!( + matches!( + ingress + .accept(&sink, &relayer.peer_id_hex, &stale, seen_at(CAP)) + .await, + Err(Rejected::Stale { .. }) + ), + "the fixture's rejected frame must be rejected on FRESHNESS, not on the capacity bound" + ); + let (_, after_reject) = ingress.tracked_counts().await; + assert_eq!( + after_reject, CAP, + "a rejected announcement must neither allocate an entry nor evict one" + ); + + // One over the bound. + admit(CAP).await; + let last = seen_at(CAP); + let (_, over_cap) = ingress.tracked_counts().await; + assert_eq!( + over_cap, + CAP, + "admitting a {n}th distinct provider must leave the map at its capacity", + n = CAP + 1 + ); + // The retained provider is asserted FIRST: a readmitted provider evicts a new victim, so probing + // the evicted one first would destroy the very watermark the next assertion reads. + assert_eq!( + replay(1, last).await, + Err(Rejected::StaleSeq { + seq: 7, + highest: 7 + }), + "a provider that was NOT the least-recently-seen must keep its watermark — otherwise the map \ + is being cleared, or the wrong victim chosen, rather than evicted least-recently-seen first" + ); + assert!( + replay(0, last).await.is_ok(), + "the LEAST-recently-seen provider is the one evicted, so its watermark is gone and its own \ + seq-7 frame is admissible again" + ); +} + +/// PROPERTY: the SENDER map is capacity-bounded, and the entry it drops at the bound is the +/// LEAST-RECENTLY-SEEN one. +/// +/// The sender half of `SPEC.md` §19.3a's "capacity-bounded (1,024 senders, 8,192 providers) with +/// least-recently-seen eviction" was, until this test, an untested claim: deleting +/// `evict_lru(&mut self.senders, …)` left the whole suite green. It is a real bound, not bookkeeping +/// — a `peer_id` is `SHA-256(NodeCert SPKI)` and therefore self-minted, so connect → announce → +/// disconnect churn offers an unbounded key space and nothing else ever removes a sender entry. +/// +/// The fixture mirrors the provider-side test's two disciplines, because a count alone cannot tell +/// LRU eviction from evicting an arbitrary entry: +/// +/// - **A reachable cap.** [`IngressLimits::tracked_senders`] is parameterised, so the bound is +/// crossed with four transport identities instead of 1,025. +/// - **A second observable besides the count.** Eviction is read through its CONSEQUENCE: a sender +/// entry IS that sender's delta budget, so losing the entry restores an exhausted sender's ability +/// to relay. Every sender is driven to exhaustion first, the same relay is asserted REFUSED before +/// the bound is crossed, and a RETAINED sender is probed before the evicted one — so the final +/// admission is attributable to eviction of the least-recently-seen entry and to nothing else. +/// +/// A too-old frame from a never-seen fifth sender sits in the middle: it must neither allocate a +/// sender entry nor evict one, composing this bound with the allocate-nothing-on-reject rule. +#[tokio::test] +async fn the_sender_map_evicts_the_least_recently_seen_at_its_capacity() { + const CAP: usize = 3; + /// Deltas one sender may relay per window — small enough that two relays exhaust it. + const BUDGET: u32 = 2; + + let us = TestPeer::new(); + let sink = RecordingSink::default(); + let ingress = HoldingsIngress::with_limits( + us.peer_id_hex.clone(), + IngressLimits { + tracked_senders: CAP, + deltas_per_sender: BUDGET, + ..IngressLimits::default() + }, + ); + + // Four relaying transports, each last seen at a distinct second so the least-recently-seen order + // is unambiguous. All stamps stay well inside both RATE_WINDOW_SECS and MAX_ANNOUNCE_AGE_SECS. + let relays: Vec = (0..=CAP).map(|_| TestPeer::new()).collect(); + let seen_at = |i: usize| NOW + i as u64; + + // One relay through transport `i`, carrying a single delta from a FRESHLY minted provider — so + // every provider-side gate passes and the only budget in play is the sender's. + let relay = |i: usize, seed: u8, at: u64| { + let announce = TestPeer::new().announce(1, vec![add_delta(&content(seed))]); + let ingress = &ingress; + let sink = &sink; + let sender = &relays[i].peer_id_hex; + async move { ingress.accept(sink, sender, &announce, at).await } + }; + + // Fill the map to capacity, exhausting each sender's budget as we go. + for i in 0..CAP { + for d in 0..BUDGET { + let seed = 10 + (i as u32 * BUDGET + d) as u8; + relay(i, seed, seen_at(i)) + .await + .unwrap_or_else(|e| panic!("sender {i} relay {d} must be admitted, got {e:?}")); + } + } + let (at_cap, _) = ingress.tracked_counts().await; + assert_eq!( + at_cap, CAP, + "exactly at the capacity nothing may be evicted — a bound tested only from above cannot \ + show it is the RIGHT bound" + ); + assert_eq!( + relay(0, 90, seen_at(CAP)).await, + Err(Rejected::RateLimited), + "before the bound is crossed the oldest sender still holds its exhausted budget; without \ + this the admission asserted below would not be attributable to eviction" + ); + + // A rejected frame in the mix, from a transport this ingress has never seen. + let stranger_sender = TestPeer::new(); + let stale = TestPeer::new().announce_at( + 1, + NOW - MAX_ANNOUNCE_AGE_SECS - 1, + vec![add_delta(&content(91))], + ); + assert!( + matches!( + ingress + .accept(&sink, &stranger_sender.peer_id_hex, &stale, seen_at(CAP)) + .await, + Err(Rejected::Stale { .. }) + ), + "the fixture's rejected frame must be rejected on FRESHNESS, not on the capacity bound" + ); + let (after_reject, _) = ingress.tracked_counts().await; + assert_eq!( + after_reject, CAP, + "a rejected announcement must neither allocate a sender entry nor evict one" + ); + + // One over the bound. + let last = seen_at(CAP); + relay(CAP, 92, last).await.unwrap_or_else(|e| { + panic!( + "a {n}th distinct sender must be admitted, got {e:?}", + n = CAP + 1 + ) + }); + let (over_cap, _) = ingress.tracked_counts().await; + assert_eq!( + over_cap, + CAP, + "admitting a {n}th distinct sender must leave the map at its capacity", + n = CAP + 1 + ); + + // The retained sender is asserted FIRST: readmitting the evicted one evicts a new victim, which + // would destroy the very budget the other assertion reads. + assert_eq!( + relay(1, 93, last).await, + Err(Rejected::RateLimited), + "a sender that was NOT the least-recently-seen must keep its exhausted budget — otherwise \ + the map is being cleared, or the wrong victim chosen, rather than evicted \ + least-recently-seen first" + ); + assert!( + relay(0, 94, last).await.is_ok(), + "the LEAST-recently-seen sender is the one evicted, so its budget is gone with its entry \ + and it may relay again" + ); +} + +// ============================================================================================= +// The three guards the round-2 defect-revert probe found UNVERIFIED +// ============================================================================================= + +/// Captures everything logged inside a scope, so an assertion can read exactly what an operator +/// tailing the node log would see. +#[derive(Clone, Default)] +struct CaptureBuffer(Arc>>); + +impl std::io::Write for CaptureBuffer { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0.lock().expect("capture mutex").extend_from_slice(buf); + Ok(buf.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for CaptureBuffer { + type Writer = CaptureBuffer; + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } +} + +/// PROPERTY (item 6): no peer-supplied byte ever reaches the log, on ANY path. +/// +/// `provider_peer_id` is a `u16`-length-prefixed wire string, so a hostile peer may put tens of +/// kilobytes of arbitrary UTF-8 there — newlines that forge whole log lines, ANSI escapes that drive +/// an operator's terminal. This is asserted against the REAL ingest loop's REAL emitted records +/// (captured through a `tracing` subscriber), not against a return value, because the defect is in +/// what the loop *emits*: keeping the record at `debug` bounds log VOLUME and says nothing about +/// CONTENT, so a level-based rationale cannot substitute for not emitting the field. +#[tokio::test] +async fn no_peer_supplied_bytes_ever_reach_the_log() { + let holder = TestPeer::new(); + let us = TestPeer::new(); + + // A forged log line, a terminal escape, and enough length to be obvious in a diff. + let hostile = "\n\u{1b}[2J\u{1b}[31mnode: CRITICAL forged line ".repeat(40); + let mut announce = holder.announce(1, vec![add_delta(&content(59))]); + announce.provider_peer_id = hostile.clone(); + + let (tx, rx) = tokio::sync::broadcast::channel(8); + tx.send(( + dig_gossip::PeerId::from([0x33u8; 32]), + frame_holdings_announce(&announce), + )) + .expect("the receiver is live"); + drop(tx); // closes the channel so `run_holdings_ingest` returns after draining + + let buffer = CaptureBuffer::default(); + let subscriber = tracing_subscriber::fmt() + .with_max_level(tracing::Level::TRACE) + .with_ansi(false) + .with_writer(buffer.clone()) + .finish(); + { + let _guard = tracing::subscriber::set_default(subscriber); + run_holdings_ingest( + rx, + Arc::new(HoldingsIngress::new(us.peer_id_hex.clone())), + local_dht_service(9_401), + ) + .await; + } + let logged = + String::from_utf8_lossy(&buffer.0.lock().expect("capture mutex").clone()).into_owned(); + + assert!( + !logged.contains("forged line"), + "the peer-supplied provider id must NEVER be logged; captured:\n{logged}" + ); + assert!( + !logged.contains('\u{1b}'), + "no terminal escape from a peer-supplied field may reach the log; captured:\n{logged}" + ); + assert!( + logged.contains(""), + "the rejection should still be observable, just without attacker-shaped text; captured:\n{logged}" + ); +} + +/// PROPERTY (end to end): an announcement declaring far more addresses than the cap is still ACCEPTED, +/// and the record it produces is bounded — an oversized list is trimmed, never a rejection. +/// +/// Deliberately NOT the guard test for the cap's placement. `ProviderRecord::new` truncates as well, so +/// this assertion stays green with `bounded_dht_addresses`' `take` deleted; it can only see the stored +/// record, and both placements store the same one. The placement — mapping at most the cap, so the +/// allocation and not merely the record is bounded — is pinned by the colocated unit test +/// `the_address_cap_is_applied_where_the_mapping_happens`, which observes the mapper's own output. +/// +/// The sink here records the WHOLE `ProviderRecord`: the narrower `(content_key, provider)` double used +/// elsewhere physically cannot express an address-count lie. +#[tokio::test] +async fn an_oversized_address_list_still_yields_a_bounded_record() { + let holder = TestPeer::new(); + let us = TestPeer::new(); + let sink = RecordingRecords::default(); + let cap = dig_dht::MAX_ADDRESSES_PER_RECORD; + + let declared = cap * 4; + let addresses: Vec<_> = (0..declared) + .map(|i| dig_gossip::CandidateAddr { + host: format!("::{i:x}"), + port: 9_257, + }) + .collect(); + let announce = holder.announce( + 1, + vec![HoldingsDelta::Add { + content_key: *content(61).to_key().as_bytes(), + addresses, + expires_at: NOW + 3_600, + }], + ); + + HoldingsIngress::new(us.peer_id_hex.clone()) + .accept(&sink, &holder.peer_id_hex, &announce, NOW) + .await + .expect("a validly signed announcement is accepted"); + + let records = sink.records.lock().expect("sink mutex"); + assert_eq!(records.len(), 1); + assert!( + records[0].addresses.len() <= cap, + "an attacker-declared address count must be truncated to MAX_ADDRESSES_PER_RECORD ({cap}) \ + before the list is mapped; got {} from {declared} declared", + records[0].addresses.len() + ); +} + +/// PROPERTY (watermark lifetime): a rate window elapsing must NOT reset the replay watermark. +/// +/// Replay protection is not a rate limit. Folding the two together means simply waiting out one +/// 60-second window makes every captured frame replayable again — so a retracted record can be +/// resurrected on a timer. The fixture waits exactly one window (still well inside the freshness +/// bound, so that gate cannot be what rejects the replay — otherwise this test would pass for the +/// wrong reason). +#[tokio::test] +async fn a_rate_window_elapsing_does_not_reset_the_replay_watermark() { + let holder = TestPeer::new(); + let us = TestPeer::new(); + let sink = RecordingSink::default(); + let id = content(67); + let ingress = ingress(&us.peer_id_hex); + + let add = holder.announce(5, vec![add_delta(&id)]); + ingress + .accept(&sink, &holder.peer_id_hex, &add, NOW) + .await + .expect("seq 5 add"); + ingress + .accept( + &sink, + &holder.peer_id_hex, + &holder.announce(6, vec![remove_delta(&id)]), + NOW, + ) + .await + .expect("seq 6 retract"); + + // One full rate window later — inside MAX_ANNOUNCE_AGE_SECS, so freshness still accepts it. + let later = NOW + RATE_WINDOW_SECS; + assert!( + RATE_WINDOW_SECS < IngressLimits::default().max_announce_age_secs, + "the fixture is only meaningful while a window is shorter than the freshness bound" + ); + let rejected = ingress + .accept(&sink, &holder.peer_id_hex, &add, later) + .await + .expect_err("the watermark must outlive the rate window"); + + assert_eq!(rejected, Rejected::StaleSeq { seq: 5, highest: 6 }); + assert!( + !sink.holds(&id, &holder.peer_id_hex), + "waiting out one rate window must not make a retracted record resurrectable" + ); +} diff --git a/crates/dig-node-core/tests/holdings_wire.rs b/crates/dig-node-core/tests/holdings_wire.rs new file mode 100644 index 0000000..a603014 --- /dev/null +++ b/crates/dig-node-core/tests/holdings_wire.rs @@ -0,0 +1,651 @@ +//! REAL-WIRE conformance for the opcode-222 holdings flywheel (#1429). +//! +//! Everything here is genuine: two live `dig-gossip` services connected over a loopback **mTLS** +//! link, each presenting its own persisted `dig_tls::NodeCert`; a real signature produced by node A's +//! actual TLS leaf key; the real `frame_holdings_announce` encoder; a real transmit through the +//! connected pool; the real inbound receiver and decoder on node B; and a real `dig_dht::DhtService` +//! as the provider store `find_providers` is then queried against. +//! +//! Why this test exists as a SEPARATE layer from `holdings_ingress.rs`: a struct round-trip through a +//! helper both sides share can pass while the real wire is broken — this ecosystem has shipped exactly +//! that bug (a frame-count assertion written against `tokio::io::sink()` pinned a live defect as +//! correct). The properties only a real wire can establish are: +//! +//! 1. The announcing node's signature verifies under the SPKI its **TLS handshake** presented, so +//! `provider_peer_id` is the same identity a peer would dial — not merely self-consistent. +//! 2. An announcement survives encode → transmit → decode across the opcode-222 frame intact. +//! 3. The ingested record is discoverable through the real `find_providers`, i.e. the flywheel's +//! DISCOVER stage actually sees the new holder. + +use std::collections::BTreeSet; +use std::net::SocketAddr; +use std::sync::Arc; +use std::sync::Mutex; +use std::time::Duration; + +use dig_dht::{ + CandidateAddr, ContentId, DhtConfig, DhtError, DhtRequest, DhtResponse, DhtService, + DhtTransport, PeerId, +}; +use dig_gossip::{ + frame_holdings_announce, holdings_announce_payload, GossipConfig, GossipHandle, GossipService, + HoldingsAnnounce, HoldingsDelta, PeerPoolConfig, HOLDINGS_ANNOUNCE, +}; +use dig_node_core::peer::{install_crypto_provider, load_or_generate_node_cert}; +use dig_node_core::seams::dig_peer::holdings::{ + announcement_for, reconcile_and_announce, signer_from_node_cert, AnnounceTransport, + HoldingsBroadcaster, HoldingsIngress, +}; + +/// The clock this test announces against — deliberately **real wall-clock**, unlike the pure-policy +/// suite in `holdings_ingress.rs` which pins an explicit `NOW`. +/// +/// The distinction is load-bearing and was found the hard way: `dig_dht::DhtService` reads the system +/// clock to clamp and expire provider records, so a *pinned past* `NOW` makes every announced +/// `expires_at` already expired by the time it is ingested. The first draft of this test used a pinned +/// `NOW` and `find_providers` returned an empty set — it would have "proved" the flywheel closed while +/// only ever exercising the expired-record path. Any fixture time handed to a wall-clock API must be +/// the wall clock. +fn now_secs() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("the system clock is after the Unix epoch") + .as_secs() +} + +/// A node identity plus the live gossip service presenting it. +struct WireNode { + cert: Arc, + handle: GossipHandle, + /// Kept alive so the persisted `node.crt`/`node.key` the pool listener reads are not unlinked. + _dir: tempfile::TempDir, +} + +impl WireNode { + /// Start a pool whose TLS material IS the node's persisted `NodeCert`, exactly as production does + /// (`gossip_identity_paths` points dig-gossip at `node.crt`/`node.key`). This is what makes the + /// handshake identity and the announce-signing identity the SAME key — the property under test. + async fn start(seed: [u8; 32], network: [u8; 32]) -> Self { + let dir = tempfile::tempdir().expect("cert tempdir"); + let cert = load_or_generate_node_cert(dir.path(), &seed).expect("persist a NodeCert"); + let listen: SocketAddr = "[::1]:0".parse().expect("parse [::1]:0"); + let cfg = GossipConfig { + network_id: chia_protocol::Bytes32::new(network), + cert_path: dir.path().join("node.crt").display().to_string(), + key_path: dir.path().join("node.key").display().to_string(), + peers_file_path: dir.path().join("peers.json"), + peer_pool: Some(PeerPoolConfig::default()), + listen_addr: listen, + ..Default::default() + }; + let handle = GossipService::new(cfg) + .expect("gossip config is valid") + .start() + .await + .expect("gossip service starts"); + Self { + cert, + handle, + _dir: dir, + } + } + + /// The `peer_id` this node's mTLS handshake presents, as 64-hex. + /// + /// Asserts on the way through that the handshake identity IS `SHA-256(NodeCert SPKI DER)` (§5.2), + /// because that equality is the whole reason a holdings signature can stand in for mTLS + /// attribution — if the pool ever presented a different cert, every announcement this node made + /// would name an identity no peer could dial. + fn peer_id_hex(&self) -> String { + let handshake = hex_of( + &self + .handle + .local_peer_id() + .expect("a started service has a local peer_id"), + ); + let from_spki = dig_tls::peer_id_from_tls_spki_der(self.cert.spki_der()).to_hex(); + assert_eq!( + handshake, from_spki, + "the pool must present the node's own NodeCert, so peer_id = SHA-256(SPKI DER)" + ); + handshake + } + + /// The loopback address a peer dials to reach this node's pool listener. + fn dial_addr(&self) -> SocketAddr { + let bound = self + .handle + .__listen_bound_addr_for_tests() + .expect("a started pool has resolved its ephemeral port"); + format!("[::1]:{}", bound.port()) + .parse() + .expect("loopback dial addr") + } +} + +/// Lowercase-hex a gossip `PeerId` (a chia `Bytes32`). +fn hex_of(peer_id: &dig_gossip::PeerId) -> String { + hex::encode(peer_id.to_bytes()) +} + +/// Mint a persisted `NodeCert` from `seed` (its temp dir is dropped; the cert is self-contained). +fn node_cert_for(seed: [u8; 32]) -> Arc { + let dir = tempfile::tempdir().expect("cert tempdir"); + load_or_generate_node_cert(dir.path(), &seed).expect("NodeCert") +} + +/// The pinned fixture clock for the reconcile tests — these never reach dig-dht's expiry clamp with a +/// past value, because they assert on the LOCAL provider store immediately after the reconcile. +const NOW: u64 = 1_782_000_000; + +/// Records every announcement handed to the transport, in order. +#[derive(Default)] +struct RecordingTransport { + sent: Mutex>, +} + +#[async_trait::async_trait] +impl AnnounceTransport for RecordingTransport { + async fn flood(&self, announce: &HoldingsAnnounce) -> usize { + self.sent + .lock() + .expect("transport mutex") + .push(announce.clone()); + 1 + } +} + +/// A transport that reaches nobody: the ingesting node's DHT only answers LOCAL queries here, so a +/// `find_providers` hit proves the record is in this node's own provider store rather than fetched +/// back off the network. +struct UnreachableTransport; + +#[async_trait::async_trait] +impl DhtTransport for UnreachableTransport { + async fn rpc( + &self, + _from: &dig_dht::Contact, + _target: &dig_dht::Contact, + _request: &DhtRequest, + ) -> Result { + Err(DhtError::Transport("unreachable in this test".to_string())) + } +} + +/// Poll until `handle` reports a connected pool peer, or the deadline elapses. +async fn await_connected(handle: &GossipHandle, deadline: Duration) -> usize { + let start = std::time::Instant::now(); + loop { + let n = handle.connected_pool_peers().len(); + if n >= 1 || start.elapsed() >= deadline { + return n; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } +} + +/// PROPERTY (the flywheel's RESHARE→DISCOVER hop, over the real wire): a holder's signed opcode-222 +/// announcement, transmitted over a live mTLS gossip link, is verified against the identity its TLS +/// handshake presented and lands in the receiver's real DHT provider store, where `find_providers` +/// finds it. +/// +/// The truthful control is the identity itself: node B verifies the announcement against +/// `SHA-256(SPKI)` recovered from the announcement, and the test asserts that value equals the +/// `peer_id` node A's **handshake** presented. A test that only checked internal self-consistency +/// would pass even if the announce were signed by an unrelated key. +#[tokio::test] +async fn a_signed_announcement_crosses_the_real_wire_and_becomes_discoverable() { + install_crypto_provider(); + let network = [0x5au8; 32]; + let holder = WireNode::start([0x11u8; 32], network).await; + let receiver = WireNode::start([0x22u8; 32], network).await; + assert!( + receiver.handle.connected_pool_peers().is_empty(), + "a freshly started pool has no peers yet" + ); + + // -- Connect the two pools over loopback mTLS ------------------------------------------------ + let target = receiver.dial_addr(); + holder + .handle + .connect_to(target) + .await + .expect("holder dials the receiver over loopback mTLS"); + assert_eq!( + await_connected(&receiver.handle, Duration::from_secs(10)).await, + 1, + "the receiver accepted the inbound mTLS link" + ); + + // -- Subscribe BEFORE announcing so the frame cannot be missed ------------------------------- + let mut inbound = receiver + .handle + .inbound_receiver() + .expect("a started service exposes its inbound receiver"); + + // -- The holder signs with its REAL TLS leaf key and floods opcode 222 ----------------------- + let signer = signer_from_node_cert(&holder.cert).expect("the NodeCert leaf is ECDSA-P256"); + let capsule = ContentId::capsule([0xa1u8; 32], [0xb2u8; 32]); + let serve_addr = dig_gossip::CandidateAddr { + host: "::1".to_string(), + port: 9_257, + }; + let announce = announcement_for(&signer, 1, now_secs(), &[capsule], &[], &[serve_addr]) + .expect("the batch is within the protocol cap") + .expect("a gained capsule produces an announcement"); + holder + .handle + .broadcast(frame_holdings_announce(&announce), None) + .await + .expect("the opcode-222 frame is broadcast to the connected pool"); + + // -- The receiver reads the frame off the real wire ------------------------------------------ + let (sender, decoded) = tokio::time::timeout(Duration::from_secs(10), async { + loop { + let (sender, msg) = inbound.recv().await.expect("inbound channel stays open"); + assert_eq!( + msg.msg_type as u8, HOLDINGS_ANNOUNCE, + "only the holdings frame was sent on this link" + ); + if let Some(a) = holdings_announce_payload(&msg) { + break (sender, a); + } + } + }) + .await + .expect("the opcode-222 frame arrives within the deadline"); + + assert_eq!( + decoded, announce, + "the announcement survives encode -> transmit -> decode byte-for-byte" + ); + assert_eq!( + decoded.provider_peer_id, + holder.peer_id_hex(), + "the signed provider identity IS the peer_id the holder's mTLS handshake presented — so a \ + peer that discovers this record can actually dial the announcer" + ); + + // -- Verify + ingest into a REAL DhtService -------------------------------------------------- + let receiver_id = PeerId::from_hex(&receiver.peer_id_hex()).expect("64-hex peer id"); + let dht = Arc::new(DhtService::new( + receiver_id, + vec![CandidateAddr::direct("::1".to_string(), 9_258)], + DhtConfig::default(), + Arc::new(UnreachableTransport), + )); + let ingress = HoldingsIngress::new(receiver.peer_id_hex()); + let applied = ingress + .accept(&dht, &hex_of(&sender), &decoded, now_secs()) + .await + .expect("a genuinely signed announcement off the real wire is accepted"); + assert_eq!(applied.ingested, 1, "the add was ingested"); + + // -- The flywheel's DISCOVER stage now sees the new holder ----------------------------------- + let providers = dht + .find_providers(&capsule) + .await + .expect("a local provider-store hit needs no network"); + assert!( + providers + .iter() + .any(|p| p.provider_peer_id == holder.peer_id_hex()), + "find_providers must return the announcing holder — this is the read->cache->announce->\ + discover loop closing over the real wire; got {providers:?}" + ); + + let holders = dht + .holders_of(&capsule) + .await + .expect("holders_of projects the same lookup"); + assert!( + holders.iter().any(|h| h.to_hex() == holder.peer_id_hex()), + "the holder-set query agrees with find_providers" + ); +} + +/// PROPERTY (the RETRACT half, over the real wire): the holder's signed retract, transmitted on the +/// same real link, removes its own record from the receiver's real DHT — so an evicting node stops +/// being advertised in seconds instead of at TTL scale. +/// +/// The truthful control is a SECOND, honest holder of the same capsule seeded into the receiver's DHT +/// through the ordinary serving-side path: it must still be discoverable afterwards. Without it, +/// "find_providers is now empty" would pass for a wrong implementation that wipes the whole key. +#[tokio::test] +async fn a_signed_retract_crosses_the_real_wire_and_spares_the_other_holder() { + install_crypto_provider(); + let network = [0x5au8; 32]; + let holder = WireNode::start([0x33u8; 32], network).await; + let receiver = WireNode::start([0x44u8; 32], network).await; + + let target = receiver.dial_addr(); + holder + .handle + .connect_to(target) + .await + .expect("holder dials the receiver"); + assert_eq!( + await_connected(&receiver.handle, Duration::from_secs(10)).await, + 1, + "the receiver accepted the inbound mTLS link" + ); + let mut inbound = receiver + .handle + .inbound_receiver() + .expect("inbound receiver"); + + let receiver_id = PeerId::from_hex(&receiver.peer_id_hex()).expect("64-hex peer id"); + let dht = Arc::new(DhtService::new( + receiver_id, + vec![CandidateAddr::direct("::1".to_string(), 9_259)], + DhtConfig::default(), + Arc::new(UnreachableTransport), + )); + let ingress = HoldingsIngress::new(receiver.peer_id_hex()); + let capsule = ContentId::capsule([0xc3u8; 32], [0xd4u8; 32]); + let signer = signer_from_node_cert(&holder.cert).expect("P-256 leaf"); + let serve_addr = dig_gossip::CandidateAddr { + host: "::1".to_string(), + port: 9_257, + }; + + // A truthful CONTROL holder of the same capsule, ingested through the same authenticated path. + let other_dir = tempfile::tempdir().expect("control cert dir"); + let other_cert = + load_or_generate_node_cert(other_dir.path(), &[0x55u8; 32]).expect("control NodeCert"); + let other_signer = signer_from_node_cert(&other_cert).expect("P-256 leaf"); + let other_id = dig_tls::peer_id_from_tls_spki_der(other_cert.spki_der()).to_hex(); + let control_add = announcement_for( + &other_signer, + 1, + now_secs(), + &[capsule], + &[], + std::slice::from_ref(&serve_addr), + ) + .expect("within cap") + .expect("an add"); + ingress + .accept(&dht, &holder.peer_id_hex(), &control_add, now_secs()) + .await + .expect("the control holder's add is accepted"); + + // The holder under test announces, then retracts, both across the real wire. + for (seq, gained, lost) in [(1u64, vec![capsule], vec![]), (2, vec![], vec![capsule])] { + let announce = announcement_for( + &signer, + seq, + now_secs(), + &gained, + &lost, + std::slice::from_ref(&serve_addr), + ) + .expect("within cap") + .expect("a non-empty delta"); + holder + .handle + .broadcast(frame_holdings_announce(&announce), None) + .await + .expect("frame broadcast"); + + let (sender, decoded) = tokio::time::timeout(Duration::from_secs(10), async { + loop { + let (s, msg) = inbound.recv().await.expect("channel open"); + if let Some(a) = holdings_announce_payload(&msg) { + if a.seq == seq { + break (s, a); + } + } + } + }) + .await + .expect("the frame arrives"); + + ingress + .accept(&dht, &hex_of(&sender), &decoded, now_secs()) + .await + .unwrap_or_else(|e| panic!("seq {seq} off the real wire must be accepted, got {e:?}")); + } + + let providers = dht + .find_providers(&capsule) + .await + .expect("local provider-store lookup"); + let ids: Vec<_> = providers + .iter() + .map(|p| p.provider_peer_id.clone()) + .collect(); + assert!( + !ids.contains(&holder.peer_id_hex()), + "the retracting holder must be gone from the provider set; got {ids:?}" + ); + assert!( + ids.contains(&other_id), + "the OTHER honest holder of the same capsule must survive the retract; got {ids:?}" + ); +} + +// ============================================================================================= +// Inventory reconcile — the declared behaviour change, and the reconcile->flood COMPOSITION +// ============================================================================================= +// +// These exercise a REAL `DhtService` through a real `DhtHandle`, because the two properties at stake +// are both about what the DHT ends up holding, and neither is visible from either half alone. + +/// A cached-capsule inventory entry for `(store, root)`. +fn cached_capsule(store: u8, root: u8) -> dig_node_core::CachedCapsule { + dig_node_core::CachedCapsule { + store_id: hex::encode([store; 32]), + root: hex::encode([root; 32]), + size_bytes: 4_096, + last_used_unix_ms: 1_782_000_000_000, + } +} + +/// A local-only DHT handle: its transport reaches nobody, so every `find_providers` answer comes from +/// this node's OWN provider store — which is exactly the state under test. +fn local_dht_handle(port: u16) -> Arc { + let service = Arc::new(DhtService::new( + PeerId::from_bytes([0x7eu8; 32]), + vec![CandidateAddr::direct("::1".to_string(), port)], + DhtConfig::default(), + Arc::new(UnreachableTransport), + )); + dig_node_core::dht::DhtHandle::new(service, Vec::new()) +} + +/// PROPERTY (the declared MEDIUM behaviour change): losing a capsule must make this node STOP being +/// returned by `find_providers` IMMEDIATELY, not at TTL expiry. +/// +/// This is the difference between `retract_own_provider` and the passive `withdraw_provider` it +/// replaced, and it is the whole justification for the change: `withdraw_provider` only unmarks the +/// key for republish and LEAVES the local record, so for the remainder of its TTL this node keeps +/// answering `find_providers` with itself for content it can no longer serve — one wasted dial per +/// reader. Reverting the call in `sync_inventory` reds this test on the final assertion. +#[tokio::test] +async fn losing_a_capsule_stops_this_node_being_returned_as_a_provider_at_once() { + let dht = local_dht_handle(9_301); + let capsule = ContentId::capsule([0x01u8; 32], [0x02u8; 32]); + + // GAIN: the node caches the capsule and reconciles. + let gained = dht.reconcile_inventory(&[cached_capsule(0x01, 0x02)]).await; + assert!( + gained.gained.contains(&capsule), + "the reconcile must report the capsule as gained; got {gained:?}" + ); + let providers = dht + .service() + .find_providers(&capsule) + .await + .expect("local provider-store lookup"); + assert_eq!( + providers.len(), + 1, + "after caching, this node is discoverable as a holder" + ); + + // LOSE: the capsule leaves the inventory (an eviction, a cache-remove, a store deletion). + let lost = dht.reconcile_inventory(&[]).await; + assert!( + lost.lost.contains(&capsule), + "the reconcile must report the capsule as lost; got {lost:?}" + ); + + let after = dht + .service() + .find_providers(&capsule) + .await + .expect("local provider-store lookup"); + assert!( + after.is_empty(), + "a node that no longer holds a capsule must NOT still be returned as its provider — a \ + passive withdraw leaves the record to lapse via TTL and costs every reader a failed dial; \ + got {after:?}" + ); +} + +/// PROPERTY (the COMPOSITION — the point of this feature): a reconcile that changes the inventory must +/// flood an announcement whose deltas are EXACTLY the ids the reconcile moved. +/// +/// The two halves passing in isolation says nothing about the wiring between them. The fixture makes +/// the two directions distinguishable — one capsule GAINED while a different one is LOST in the SAME +/// reconcile — so an implementation that floods only adds, only removes, or the wrong id set is +/// observably different from a correct one. +#[tokio::test] +async fn a_reconcile_floods_exactly_the_deltas_it_moved() { + let dht = local_dht_handle(9_302); + let transport = RecordingTransport::default(); + let signer = signer_from_node_cert(&node_cert_for([0x64u8; 32])).expect("P-256 leaf"); + let broadcaster = HoldingsBroadcaster::new(signer, Vec::new(), 0); + + // Establish a first capsule, then reconcile to a DIFFERENT one: one gain plus one loss at once. + let first = ContentId::capsule([0x11u8; 32], [0x12u8; 32]); + let second = ContentId::capsule([0x21u8; 32], [0x22u8; 32]); + reconcile_and_announce( + &dht, + &[cached_capsule(0x11, 0x12)], + Some((&broadcaster, &transport)), + NOW, + ) + .await; + transport.sent.lock().expect("mutex").clear(); + + let delta = reconcile_and_announce( + &dht, + &[cached_capsule(0x21, 0x22)], + Some((&broadcaster, &transport)), + NOW, + ) + .await; + + assert!(delta.gained.contains(&second), "the new capsule is gained"); + assert!(delta.lost.contains(&first), "the old capsule is lost"); + + let sent = transport.sent.lock().expect("mutex"); + assert_eq!(sent.len(), 1, "one reconcile, one frame"); + let announced_adds: BTreeSet<_> = sent[0] + .changes + .iter() + .filter_map(|c| match c { + HoldingsDelta::Add { content_key, .. } => Some(*content_key), + HoldingsDelta::Remove { .. } => None, + }) + .collect(); + let announced_removes: BTreeSet<_> = sent[0] + .changes + .iter() + .filter_map(|c| match c { + HoldingsDelta::Remove { content_key } => Some(*content_key), + HoldingsDelta::Add { .. } => None, + }) + .collect(); + let expected_adds: BTreeSet<_> = delta + .gained + .iter() + .map(|c| *c.to_key().as_bytes()) + .collect(); + let expected_removes: BTreeSet<_> = delta.lost.iter().map(|c| *c.to_key().as_bytes()).collect(); + + assert_eq!( + announced_adds, expected_adds, + "the flooded Add deltas must be exactly the ids the reconcile gained" + ); + assert_eq!( + announced_removes, expected_removes, + "the flooded Remove deltas must be exactly the ids the reconcile lost — a retract that the \ + DHT applied but the flood omitted leaves peers dialling a node that has evicted the capsule" + ); +} + +/// PROPERTY (the DEGRADED bring-up branch): a node that cannot sign still reconciles its durable DHT +/// provider records — it loses only the real-time flood, never its discoverability. +/// +/// `holdings: None` is the branch taken by a node whose leaf key cannot produce a P-256 holdings +/// signer, and the module documents it as "stays discoverable through the durable records alone". +/// That promise is the whole reason the flood is optional, and it was previously compile-checked only: +/// the two halves are guarded by ONE `if let (Some(..), false)` condition, so a mistake that makes the +/// missing signer short-circuit the RECONCILE too would silently make an unsigning node invisible to +/// `find_providers` — a discovery outage, not a freshness one. +/// +/// The fixture therefore asserts the SURVIVING half positively (the record is really in the provider +/// store) rather than only the absent half; a test that checked "nothing was flooded" alone passes +/// identically whether or not the reconcile ran, which is precisely the defect it must exclude. +#[tokio::test] +async fn a_node_that_cannot_sign_still_reconciles_its_records_without_flooding() { + let dht = local_dht_handle(9_304); + let capsule = ContentId::capsule([0x41u8; 32], [0x42u8; 32]); + + let gained = reconcile_and_announce(&dht, &[cached_capsule(0x41, 0x42)], None, NOW).await; + + assert!( + gained.gained.contains(&capsule), + "the reconcile must run and report the gain even with no signer; got {gained:?}" + ); + assert_eq!( + dht.service() + .find_providers(&capsule) + .await + .expect("local provider-store lookup") + .len(), + 1, + "a node that cannot sign an announcement must STILL be discoverable through its durable \ + provider record — losing the signer costs freshness, never discovery" + ); + + // And the loss direction, so the shared condition is exercised in both states of `delta`. + let lost = reconcile_and_announce(&dht, &[], None, NOW).await; + assert!( + lost.lost.contains(&capsule), + "the retract half must run without a signer too; got {lost:?}" + ); + assert!( + dht.service() + .find_providers(&capsule) + .await + .expect("local provider-store lookup") + .is_empty(), + "an unsigning node must still stop advertising content it no longer holds" + ); +} + +/// PROPERTY: a reconcile that changes nothing floods nothing, so a steady-state node is silent. +#[tokio::test] +async fn an_unchanged_reconcile_floods_nothing() { + let dht = local_dht_handle(9_303); + let transport = RecordingTransport::default(); + let signer = signer_from_node_cert(&node_cert_for([0x65u8; 32])).expect("P-256 leaf"); + let broadcaster = HoldingsBroadcaster::new(signer, Vec::new(), 0); + let inventory = [cached_capsule(0x31, 0x32)]; + + reconcile_and_announce(&dht, &inventory, Some((&broadcaster, &transport)), NOW).await; + transport.sent.lock().expect("mutex").clear(); + + let delta = + reconcile_and_announce(&dht, &inventory, Some((&broadcaster, &transport)), NOW).await; + + assert!(delta.is_empty(), "an identical inventory changed nothing"); + assert!( + transport.sent.lock().expect("mutex").is_empty(), + "a no-op reconcile must not put a frame on the wire" + ); +}