diff --git a/.loop/WIP-276.md b/.loop/WIP-276.md new file mode 100644 index 00000000..76cb7277 --- /dev/null +++ b/.loop/WIP-276.md @@ -0,0 +1,2 @@ +WIP: module-granularity onion relay leg (#276) + infallible-locator type guard (#296). +Branch: loop/276-module-relay diff --git a/Cargo.lock b/Cargo.lock index bea17113..5ee20515 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3011,7 +3011,7 @@ dependencies = [ [[package]] name = "dig-node-core" -version = "0.51.0" +version = "0.52.0" dependencies = [ "async-trait", "axum", @@ -3073,7 +3073,7 @@ dependencies = [ [[package]] name = "dig-node-service" -version = "0.138.1" +version = "0.139.0" dependencies = [ "async-trait", "axum", diff --git a/Cargo.toml b/Cargo.toml index 65d4be44..76d64569 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,7 +32,7 @@ edition = "2021" # the ROOT manifest (`[workspace.package].version`), so it MUST be set here for a # release to fire (§3.6). The library crates (dig-node-core/dig-runtime/dig-wallet) # keep their own independent versions — only the released binary tracks the workspace version. -version = "0.138.1" +version = "0.139.0" # Release hardening, matching digstore: keep integer-overflow checks ON in release. # The node parses untrusted serialized input and does offset/length arithmetic over diff --git a/crates/dig-node-core/Cargo.toml b/crates/dig-node-core/Cargo.toml index 911eb13b..e57d7bd9 100644 --- a/crates/dig-node-core/Cargo.toml +++ b/crates/dig-node-core/Cargo.toml @@ -23,7 +23,14 @@ name = "dig-node-core" # out-of-workspace implementor; this crate is consumed in-workspace only and is pre-1.0, so it is a # MINOR bump under SemVer's 0.x rule -- recorded here rather than letting the number imply the trait # surface held still. -version = "0.51.0" +# 0.52.0 adds the module-granularity capsule relay leg (`seams/dig_peer/module_relay.rs`, gated OFF +# by default) and makes the provider/candidate locator readers absence-safe: `NodeContent::find_providers` +# and `NatModuleTransport::discovered_candidates` now return `BestEffort` rather than a bare `Vec`, +# so a source that could not be consulted can no longer be read as a proven absence (dig_ecosystem#3128, +# dig-node#276/#296). Changing a public return type is BREAKING for an out-of-workspace implementor; +# this crate is consumed in-workspace only and is pre-1.0, so it is a MINOR bump under SemVer's 0.x +# rule -- recorded here rather than letting the number imply the locator surface held still. +version = "0.52.0" edition = "2021" license = "GPL-2.0-only" description = "The canonical DIG node ENGINE library (crate `dig_node_core`): the JSON-RPC dispatch (`handle_rpc`, the same contract as rpc.dig.net), local-first content serve/fetch/redirect from LOCAL .dig store modules (via digstore_host::serve_blind), chain-anchored-root resolution, chain-watch + subscriptions + generation gap-fill, the LRU cache, and the full P2P stack. Shared UNCHANGED by both host shells: the `dig-node` OS-service binary (dig-node-service) and the DIG Browser's in-process cdylib (dig-runtime). Native Rust so the compiled-module serve path works." diff --git a/crates/dig-node-core/src/download.rs b/crates/dig-node-core/src/download.rs index 26bae93a..b8526e20 100644 --- a/crates/dig-node-core/src/download.rs +++ b/crates/dig-node-core/src/download.rs @@ -179,6 +179,38 @@ fn resolve_miss_mode(v: Option<&str>) -> MissMode { } } +/// The environment variable an operator opts the MODULE-GRANULARITY RELAY in with (dig-node#276). +pub const ONION_RELAY_ENV: &str = "DIG_NODE_ONION_RELAY"; + +/// Whether this node is willing to RELAY a whole `.dig` capsule on a requestor's behalf — pull it +/// from a holder the requestor cannot reach, then serve the requestor's module windows from the +/// result (dig-node#276). Resolved from [`ONION_RELAY_ENV`]; **unset means OFF**. +/// +/// Off by default because this leg spends OTHER nodes' bandwidth, not only this one's: an admitted +/// relay makes this node pull an entire capsule from a third party for a stranger. A path with that +/// reach is not gated more loosely than the ones that spend only local egress, so the operator opts +/// in explicitly or the leg does not exist. +pub fn onion_relay_from_env() -> bool { + resolve_onion_relay(std::env::var(ONION_RELAY_ENV).ok().as_deref()) +} + +/// Pure core of [`onion_relay_from_env`]: `1` / `true` / `on` / `yes` (case-insensitive, trimmed) +/// enables the relay; **everything else, including unset and anything unreadable, is OFF**. +/// +/// The failure direction is deliberate and is the whole point of the gate: an unreadable value must +/// never be read as consent to spend the network's bandwidth. Pure so the policy is tested without +/// process-global env. +fn resolve_onion_relay(v: Option<&str>) -> bool { + matches!( + v.map(str::trim), + Some(s) + if s.eq_ignore_ascii_case("1") + || s.eq_ignore_ascii_case("true") + || s.eq_ignore_ascii_case("on") + || s.eq_ignore_ascii_case("yes") + ) +} + /// The [`BackfillPolicy`] this node acquires under (§5.6): when a resource read is satisfied FROM /// ANOTHER NODE (a redirect or a fetch-through miss for a concrete `(store, root)`), the node ALSO /// pulls the whole `.dig` capsule for that generation in the background and caches it, so the NEXT @@ -786,6 +818,11 @@ pub struct NodeContent { /// The asks already walked here, so the same question arriving by two paths through the graph is /// forwarded once (dig-node#273). See [`AskSeenSet`]. ask_seen: AskSeenSet, + /// Whether the operator opted this node into RELAYING whole capsules for requestors that cannot + /// reach a holder themselves (dig-node#276). Read from [`ONION_RELAY_ENV`] ONCE, at construction, + /// for the same reason the recursion config is passed in rather than re-read at decision time: the + /// amplification posture of a running node must not change underneath an in-flight request. + onion_relay: std::sync::atomic::AtomicBool, } /// What a holder search ESTABLISHED — the records it found AND whether an empty result is a fact. @@ -1064,6 +1101,66 @@ impl RangeTransport for PoolConfirmTransport { } } +/// A best-effort lookup result: the items a source managed to produce, plus whether that source +/// FAILED while producing them (dig-node#296). +/// +/// # Why this is a type and not a doc comment +/// +/// The readers below are deliberately infallible — a dial candidate list must not be lost because a +/// DHT walk errored, and a debug counter must not propagate an error. The hazard is that their +/// `Vec` answers an emptiness question they were never entitled to answer: a walk that FAILED and a +/// walk that found NOBODY collapse into the same empty vector, and a caller reading that as "this +/// content does not exist" asserts a proven absence on the strength of its own broken network. That +/// is not hypothetical — it is exactly the defect PR #292 took three rounds to see, because two +/// correct `Err` arms were written above the layer that had already swallowed the failure. +/// +/// So the failure travels WITH the items, and the only way to conclude an absence is +/// [`Self::absence_established`], which consults it. `for_finding` names what it is for at every +/// call site, and consumes `self` so the absence question cannot be asked afterwards. +/// +/// This is construction, not care — but it is honest about its limit: a caller determined to read +/// `for_finding().is_empty()` as an absence can still do so. What it makes impossible is doing it +/// *accidentally*, which is how the original defect arrived. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BestEffort { + items: Vec, + source_failed: bool, +} + +impl BestEffort { + /// The source answered. An empty `items` here is a real "nobody holds this". + pub fn found(items: Vec) -> Self { + Self { + items, + source_failed: false, + } + } + + /// The source could not be consulted. Empty, and NOT an absence. + pub fn source_failed() -> Self { + Self { + items: Vec::new(), + source_failed: true, + } + } + + /// The items, for a caller that is FINDING — dialling, counting, enriching a candidate set. + /// + /// Best-effort is right for finding: a partial answer beats no answer, and a lost DHT hint costs + /// one unused dial candidate rather than a wrong conclusion. + pub fn for_finding(self) -> Vec { + self.items + } + + /// Whether this result licenses the claim that the content is held by NOBODY. + /// + /// **Strict for absence.** True only when the source actually answered AND answered empty; a + /// failed source is unknown, never absent. + pub fn absence_established(&self) -> bool { + self.items.is_empty() && !self.source_failed + } +} + impl NodeContent { /// Build the engine from injected locate + transport seams (the constructor tests use with the /// dig-download [`testkit`](dig_download::testkit) mocks; production goes through @@ -1179,6 +1276,7 @@ impl NodeContent { discovery_cache: std::sync::OnceLock::new(), holder_cache: FirstHandHolderCache::new(), ask_seen: AskSeenSet::new(), + onion_relay: std::sync::atomic::AtomicBool::new(onion_relay_from_env()), }) } @@ -1363,6 +1461,33 @@ impl NodeContent { self.miss_mode } + /// Whether this node will RELAY a whole capsule on a requestor's behalf (dig-node#276) — the + /// operator gate resolved once at construction. See [`onion_relay_from_env`]. + pub(crate) fn onion_relay_enabled(&self) -> bool { + self.onion_relay.load(std::sync::atomic::Ordering::Relaxed) + } + + /// Flip the relay gate. **Test-only**: production resolves it once from the environment at + /// construction, so a running node's amplification posture is fixed. Exposing the switch here — + /// rather than having the relay leg read `std::env` at decision time — is what lets the leg be + /// exercised in BOTH directions without a process-global mutation two parallel tests would race on. + #[cfg(test)] + pub(crate) fn set_onion_relay(&self, enabled: bool) { + self.onion_relay + .store(enabled, std::sync::atomic::Ordering::Relaxed); + } + + /// Admit one PROXY-class fetch for `requestor` against its separate, tighter allowance + /// (dig_ecosystem#2189), or refuse it. The relay leg (dig-node#276) admits through this rather + /// than the cheap-lookup budget: relaying a whole capsule is the costliest thing a stranger can + /// ask this node to do, so it draws from the bucket sized for expensive egress. + /// + /// A LOCAL requestor is exempt, mirroring [`Self::miss_outcome`]'s proxy leg — the bound targets + /// remote callers, and the operator's own reads are not an amplification vector. + pub(crate) fn allow_proxy_fetch(&self, requestor: &crate::rate_limit::RequestorId) -> bool { + requestor.is_local() || self.proxy_rate_limiter.check(requestor) + } + /// The staging directory downloads run in (`/downloads`). pub fn downloads_dir(&self) -> &Path { &self.downloads_dir @@ -1791,14 +1916,17 @@ impl NodeContent { Ok(found) } - /// [`Self::walk_for_providers`] for the callers that genuinely cannot act on the difference — a - /// failed walk reads as an empty set. + /// [`Self::walk_for_providers`] for the callers that genuinely cannot act on a failure — a dial + /// candidate set, a debug counter. /// - /// **Not for anything that reports an ABSENCE.** The only production caller is the debug log in - /// [`Self::fetch_resource`], which is counting candidates rather than deciding whether content - /// exists; a caller that must tell "found nobody" from "could not look" uses the fallible form. - pub async fn find_providers(&self, content: &ContentId) -> Vec { - self.walk_for_providers(content).await.unwrap_or_default() + /// **A failed walk does not become an empty one.** The failure rides along in the returned + /// [`BestEffort`], so a caller gets its best-effort items via `for_finding()` and can only claim + /// an absence through `absence_established()`, which is false when the walk failed (dig-node#296). + pub async fn find_providers(&self, content: &ContentId) -> BestEffort { + match self.walk_for_providers(content).await { + Ok(found) => BestEffort::found(found), + Err(_) => BestEffort::source_failed(), + } } /// The #164 content-acquisition path: multi-source download `content` (locate → confirm → fan @@ -1832,7 +1960,7 @@ impl NodeContent { // re-serve — the common case — never pays this locate's cost; only an actual cache-miss // download does, and only when someone is watching at DEBUG. if tracing::enabled!(tracing::Level::DEBUG) { - let located = self.find_providers(content).await.len(); + let located = self.find_providers(content).await.for_finding().len(); let pool_size = self .connected_pool .lock() @@ -2972,6 +3100,27 @@ pub(crate) mod tests { // -- miss-mode resolution -------------------------------------------------------------------- + /// **Proves (dig-node#276):** the capsule-relay operator gate is OFF unless the operator wrote a + /// recognised affirmative, and is ON for each of the four accepted spellings. + /// **Catches:** the failure direction being inverted — an unset, empty, or unreadable value read + /// as consent. This leg spends OTHER nodes' bandwidth, so "I could not tell" must mean OFF; a + /// default-on relay would make every stock node a capsule-scale amplifier for strangers. + #[test] + fn the_capsule_relay_gate_is_off_unless_explicitly_enabled() { + assert!(!resolve_onion_relay(None), "unset → OFF (opt-in)"); + assert!(!resolve_onion_relay(Some("")), "empty → OFF"); + assert!(!resolve_onion_relay(Some("0"))); + assert!(!resolve_onion_relay(Some("false"))); + assert!(!resolve_onion_relay(Some("off"))); + assert!( + !resolve_onion_relay(Some("maybe")), + "an unreadable value is not consent" + ); + for enabled in ["1", "true", "TRUE", " on ", "Yes"] { + assert!(resolve_onion_relay(Some(enabled)), "{enabled}"); + } + } + #[test] fn miss_mode_defaults_to_redirect_and_opts_into_fetch_through() { assert_eq!( @@ -3919,6 +4068,75 @@ pub(crate) mod tests { .is_err()); } + /// A locator whose walk cannot be performed at all — the network is down, not the content absent. + struct UnreachableLocator; + + #[async_trait::async_trait] + impl dig_download::ProviderLocator for UnreachableLocator { + async fn find_providers( + &self, + _content: &ContentId, + ) -> Result, DownloadError> { + Err(DownloadError::NotDownloadable) + } + } + + /// **Proves (dig-node#296):** a locator whose walk FAILS and a locator that genuinely holds + /// nothing produce results that are equal as candidate lists and OPPOSITE as absence claims. The + /// reader stays infallible — a dial candidate set must never be lost to a DHT error — but it no + /// longer silently licenses the absence. + /// + /// **Catches:** the exact shape of the defect PR #292 took three rounds to see. A failed DHT walk + /// was swallowed into `Ok(vec![])` two layers below the decision, so the node reported content + /// that exists as proven-absent, with no forged message and no red anywhere. Two correct `Err` + /// arms were written above the swallowing layer and BOTH were unreachable — which is why the + /// answer is a type that carries the failure, not a third `Err` arm. + /// + /// **Why the fixture varies exactly one thing:** both halves ask about the same content and both + /// come back with zero candidates. Asserting only `for_finding()` would pass with a broken flag; + /// asserting `absence_established()` only on the failing locator would pass with a flag hardwired + /// to `false`. The empty-but-honest locator is the control that rules that out — it MUST answer + /// `true`, or the flag carries no information at all. + #[tokio::test] + async fn a_failed_walk_yields_no_candidates_and_establishes_no_absence() { + let td = tempfile::tempdir().unwrap(); + let cid = mock_content_id(); + + let engine = |locator: Arc| { + NodeContent::new( + locator, + Arc::new(MockRangeTransport::new(MockContent::even(10, 1))), + MissMode::Redirect, + None, + td.path(), + ) + }; + + // The source could not be consulted at all. + let unreachable = engine(Arc::new(UnreachableLocator)) + .find_providers(&cid) + .await; + // The source answered, and the answer is that nobody holds it. + let empty = engine(Arc::new(MockProviderLocator::fixed(Vec::new()))) + .find_providers(&cid) + .await; + + assert!( + !unreachable.absence_established(), + "a walk that FAILED proves nothing about whether the content exists" + ); + assert!( + empty.absence_established(), + "a walk that COMPLETED and found nobody is a genuine absence — without this half the flag \ + could be hardwired to `false` and still pass" + ); + assert!( + unreachable.for_finding().is_empty() && empty.for_finding().is_empty(), + "both are empty as CANDIDATE lists — precisely why a bare `Vec` could not tell them \ + apart, and why the distinction had to move into the type" + ); + } + #[tokio::test] async fn find_providers_excludes_self() { let td = tempfile::tempdir().unwrap(); @@ -3934,6 +4152,7 @@ pub(crate) mod tests { td.path(), ); let got = pc.find_providers(&cid).await; + let got = got.for_finding(); assert_eq!(got.len(), 1, "own record excluded"); assert_eq!(got[0].provider_peer_id, mock_peer_hex(2)); } diff --git a/crates/dig-node-core/src/lib.rs b/crates/dig-node-core/src/lib.rs index 7ea066e0..e4c993fa 100644 --- a/crates/dig-node-core/src/lib.rs +++ b/crates/dig-node-core/src/lib.rs @@ -2723,6 +2723,47 @@ impl Node { } } + /// The transfer descriptor for a `.dig` module at the CACHE path, or `None` if this node does not + /// hold it. The blocking read + per-chunk hashing runs on a `spawn_blocking` thread (a `.dig` is + /// large; hashing it must never stall the async runtime). + /// + /// Extracted because the relay leg (dig-node#276) reads the SAME cache TWICE — once to discover + /// the miss, once after the relayed pull has landed — and a relayed answer that came from a second + /// reader would be a second thing to keep byte-identical with the held one. + async fn describe_held_module( + &self, + store_hex: &str, + root_hex: &str, + ) -> Option { + let cache_dir = self.cache_dir.clone(); + let (store, root) = (store_hex.to_string(), root_hex.to_string()); + tokio::task::spawn_blocking(move || { + seams::dig_peer::module_serve::describe_module(&cache_dir, &store, &root) + }) + .await + .unwrap_or(None) + } + + /// One window of a `.dig` module at the CACHE path, or `None` if this node does not hold it. The + /// companion of [`Self::describe_held_module`]; see there for why it is its own function. + async fn read_held_module_window( + &self, + store_hex: &str, + root_hex: &str, + offset: u64, + length: u64, + ) -> Option> { + let cache_dir = self.cache_dir.clone(); + let (store, root) = (store_hex.to_string(), root_hex.to_string()); + tokio::task::spawn_blocking(move || { + seams::dig_peer::module_serve::read_module_window( + &cache_dir, &store, &root, offset, length, + ) + }) + .await + .unwrap_or(None) + } + /// `dig.getModuleInfo` (#1576, the reshare leg): the transfer descriptor for a whole `.dig` module /// this node HOLDS — the handshake a peer reads before range-pulling the entire capsule so it can /// become a resharer of it. @@ -2735,7 +2776,12 @@ impl Node { /// - Module NOT held (or a 0-byte file, which is not a module) → the same /// `RESOURCE_UNAVAILABLE` code `dig.fetchRange` reports on a miss. Declining is the honest answer: /// describing a module this node cannot serve would advertise a capsule it does not have. - async fn get_module_info(&self, params: &Value, id: Value) -> Value { + async fn get_module_info( + &self, + params: &Value, + id: Value, + requestor: &crate::rate_limit::RequestorId, + ) -> Value { let store_hex = params .get("store_id") .and_then(Value::as_str) @@ -2753,13 +2799,19 @@ impl Node { "dig.getModuleInfo requires store_id + root (64-hex each)", ); } - let cache_dir = self.cache_dir.clone(); - let (store, root) = (store_hex.clone(), root_hex.clone()); - let info = tokio::task::spawn_blocking(move || { - seams::dig_peer::module_serve::describe_module(&cache_dir, &store, &root) - }) - .await - .unwrap_or(None); + let mut info = self.describe_held_module(&store_hex, &root_hex).await; + // MISS -> the RELAY leg (dig-node#276). A requestor that asked for a relay, on a node whose + // operator opted in and within its proxy allowance, makes this node pull the whole capsule + // from a holder and describe it from its own cache. Every gate is inside `relay_capsule`; a + // refusal simply leaves `info` as `None` and the honest not-held answer below stands. + if info.is_none() + && seams::dig_peer::module_relay::relay_capsule( + self, &store_hex, &root_hex, params, requestor, + ) + .await + { + info = self.describe_held_module(&store_hex, &root_hex).await; + } // The serve log records both outcomes with sentinelled ids, so "was this holder asked for the // descriptor, and did it have it?" is answerable from the log alone (#1595). seams::dig_peer::module_serve::module_info_answered( @@ -2792,7 +2844,12 @@ impl Node { /// /// Params `{store_id, root, offset?, length}`; the window is clamped to the serve cap. A module this /// node does not hold answers the same `RESOURCE_UNAVAILABLE` code the streaming form reports. - async fn fetch_module_range_frame(&self, params: &Value, id: Value) -> Value { + async fn fetch_module_range_frame( + &self, + params: &Value, + id: Value, + requestor: &crate::rate_limit::RequestorId, + ) -> Value { use seams::dig_peer::module_serve; let store_hex = params @@ -2818,13 +2875,22 @@ impl Node { .and_then(Value::as_u64) .unwrap_or(module_serve::MAX_MODULE_WINDOW); - let cache_dir = self.cache_dir.clone(); - let (store, root) = (store_hex.clone(), root_hex.clone()); - let window = tokio::task::spawn_blocking(move || { - module_serve::read_module_window(&cache_dir, &store, &root, offset, length) - }) - .await - .unwrap_or(None); + let mut window = self + .read_held_module_window(&store_hex, &root_hex, offset, length) + .await; + // MISS -> the RELAY leg (dig-node#276), exactly as on the descriptor above: a relayed window + // is read from the same cache, through the same reader, so it is byte-identical to the one a + // genuine holder would have served and the requestor needs no second code path. + if window.is_none() + && seams::dig_peer::module_relay::relay_capsule( + self, &store_hex, &root_hex, params, requestor, + ) + .await + { + window = self + .read_held_module_window(&store_hex, &root_hex, offset, length) + .await; + } match window { Some(bytes) => { @@ -6992,6 +7058,150 @@ mod tests { assert_eq!(decoded.bytes, bytes[100..150]); } + /// Wire node `b` as a RELAY for `(store, root)`: a P2P content engine with the relay gate open, + /// plus a capsule warmer whose only holder serves `module`. + /// + /// `b` holds nothing of its own — the capsule can reach its cache only through the relay leg, so + /// every assertion below about what `b` serves is an assertion about the relay and nothing else. + fn wire_relay_hop( + b: &Node, + store: &str, + root: &str, + module: Vec, + td: &tempfile::TempDir, + ) -> Arc { + let pc = NodeContent::new( + Arc::new(dig_download::testkit::MockProviderLocator::fixed(Vec::new())), + Arc::new(dig_download::testkit::MockRangeTransport::new( + dig_download::testkit::MockContent::even(8, 1), + )), + MissMode::Redirect, + None, + td.path(), + ); + let content = + dig_download::module_content_id(store, root).expect("canonical ids yield a content id"); + pc.set_capsule_warmer(crate::seams::dig_peer::CapsuleWarmer::new( + Arc::new(dig_download::testkit::MockProviderLocator::fixed( + dig_download::testkit::mock_providers(1, &content), + )), + Arc::new(dig_download::testkit::MockModuleTransport::serving( + store, root, module, 8, + )), + Arc::new(dig_download::InMemoryStateStore::new()), + MockResolver::one(store, Bytes32::from_hex(root).expect("64-hex root")), + crate::seams::dig_peer::WarmPaths { + staging_dir: td.path().join("relay-staging"), + cache_dir: b.cache_dir.clone(), + }, + Arc::new(SilentAnnounce), + Arc::new(crate::seams::dig_peer::WarmRegistry::new()), + dig_download::ModuleDownloadConfig::default(), + Arc::new(crate::tier0_live::NoopModulesEvictor), + )); + b.set_p2p_content(Arc::clone(&pc)); + pc + } + + /// One `dig.getModuleInfo` against `node`, with or without the relay opt-in. + async fn ask_module_info(node: &Node, store: &str, root: &str, proxy: bool) -> Value { + let mut params = json!({"store_id": store, "root": root}); + if proxy { + params["proxy"] = json!(true); + } + handle_rpc( + node, + json!({"jsonrpc":"2.0","id":1,"method":"dig.getModuleInfo","params":params}), + crate::download::ReadOrigin::Local, + crate::download::RequestProvenance::FirstParty, + ) + .await + } + + /// **Proves (dig-node#276, units 1-3):** a node that holds NOTHING answers a relay-flagged + /// `dig.getModuleInfo` + `dig.fetchModuleRange` with the real capsule — pulled through the hop from + /// the only holder — and the served window is byte-identical to what that holder would have served. + /// And it refuses, exactly as it did before this leg existed, when EITHER gate is shut: the operator + /// has not opted in, or the requestor did not ask for a relay. + /// + /// **Catches:** the two failure directions that matter in opposite ways. A relay that never fires + /// leaves requirement 4 of the recursive-download epic unimplemented at the granularity a `.dig` + /// download actually uses — which is the state this ticket found. A relay that fires REGARDLESS of + /// the gates turns every stock node into a capsule-scale amplifier a stranger can aim at a third + /// party, which is strictly worse than the gap. + /// + /// **Why all three cases run against the SAME node and the SAME wiring:** a refusal asserted on a + /// separately-built node is satisfied identically by a working gate and by a fixture whose holder + /// never answers, whose chain never confirms, or whose warmer was never wired — and each of those + /// would also silence a legitimate relay. Here the ONLY thing that varies between the refusals and + /// the success is one boolean on the request and one boolean on the node, and the success case runs + /// LAST, so it proves the same fixture that just refused twice is fully capable of answering. + #[tokio::test] + async fn a_relay_serves_a_capsule_it_does_not_hold_only_when_both_gates_are_open() { + let (b, _bd) = test_node(None); + let staging = tempfile::tempdir().unwrap(); + let store_raw = [0x7au8; 32]; + let (module, root) = chain_anchored_module(store_raw, [0x7bu8; 32]); + let (store, root) = (hex::encode(store_raw), root.to_hex()); + let pc = wire_relay_hop(&b, &store, &root, module.clone(), &staging); + + let held_path = module_path(&b.cache_dir, &store, &root); + assert!( + !held_path.exists(), + "the relay hop must start out holding nothing, or it is not relaying anything" + ); + + // GATE 1 SHUT — the operator has not opted in. The default posture of every stock node. + pc.set_onion_relay(false); + assert_eq!( + ask_module_info(&b, &store, &root, true).await["error"]["code"], + json!(download::RESOURCE_UNAVAILABLE), + "a node whose operator did not opt in must answer exactly as it did before this leg" + ); + assert!( + !held_path.exists(), + "a refused relay must not have pulled the capsule anyway" + ); + + // GATE 2 SHUT — the operator opted in, but this requestor did not ask for a relay. Relaying is + // never automatic: a requestor that can reach holders itself should, and does. + pc.set_onion_relay(true); + assert_eq!( + ask_module_info(&b, &store, &root, false).await["error"]["code"], + json!(download::RESOURCE_UNAVAILABLE), + "an unflagged request is answered from local holdings only" + ); + assert!( + !held_path.exists(), + "an unflagged request must not provoke a capsule pull" + ); + + // BOTH OPEN — the capsule arrives THROUGH the hop. + let described = ask_module_info(&b, &store, &root, true).await; + assert_eq!( + described["result"]["total_size"], + json!(module.len() as u64), + "the relayed descriptor describes the real capsule, not a placeholder" + ); + + let framed = handle_rpc( + &b, + json!({"jsonrpc":"2.0","id":2,"method":"dig.fetchModuleRange", + "params":{"store_id":store,"root":root,"offset":16,"length":32,"proxy":true}}), + crate::download::ReadOrigin::Local, + crate::download::RequestProvenance::FirstParty, + ) + .await; + let decoded: dig_nat::RangeFrame = serde_json::from_value(framed["result"].clone()) + .expect("a relayed window decodes as a RangeFrame, exactly like a held one"); + assert_eq!( + decoded.bytes, + module[16..48], + "the relayed window is byte-identical to the holder's own bytes — the requestor cannot \ + tell it was relayed, and so needs no second code path" + ); + } + /// **Proves:** a non-canonical id on either module method is a -32602 that never reaches the /// filesystem — a store id concatenated into a path would be a traversal primitive. #[tokio::test] diff --git a/crates/dig-node-core/src/peer.rs b/crates/dig-node-core/src/peer.rs index 9ee23f49..79c3497c 100644 --- a/crates/dig-node-core/src/peer.rs +++ b/crates/dig-node-core/src/peer.rs @@ -1426,13 +1426,37 @@ impl PeerRpcResponder for NodeResponder { // node's neighbourhood wants the store — tag it Tier1Demand + (opt-in) trigger a tier-1 cache. self.node.note_inbound_demand(&store, &root); - let cache = self.node.cache_dir_path().to_path_buf(); - let (s, r) = (store.clone(), root.clone()); - let window = tokio::task::spawn_blocking(move || { - module_serve::read_module_window(&cache, &s, &r, offset, length) - }) - .await - .unwrap_or(None); + let read_window = || { + let cache = self.node.cache_dir_path().to_path_buf(); + let (s, r) = (store.clone(), root.clone()); + async move { + tokio::task::spawn_blocking(move || { + module_serve::read_module_window(&cache, &s, &r, offset, length) + }) + .await + .unwrap_or(None) + } + }; + let mut window = read_window().await; + + // MISS -> the RELAY leg (dig-node#276). This is the peer-facing half: a requestor that cannot + // reach the holder itself asks US, with `proxy: true`, and — if the operator opted in and the + // requestor is inside its proxy allowance — this node pulls the whole capsule from a holder + // and serves the window from its own cache. All three gates live in `relay_capsule`; a refusal + // leaves the not-held frame below exactly as it was, so the requestor stays free to ask + // another hop (NC-12: a hop's "not found" may be a lie, including ours). + if window.is_none() + && crate::seams::dig_peer::module_relay::relay_capsule( + &self.node, + &store, + &root, + ¶ms, + &crate::rate_limit::RequestorId::Peer(conn_key.to_string()), + ) + .await + { + window = read_window().await; + } let Some(window) = window else { module_serve::module_range_outcome(conn_key, &store, &root, offset, None); diff --git a/crates/dig-node-core/src/seams/content/bandwidth.rs b/crates/dig-node-core/src/seams/content/bandwidth.rs index f4ef0002..752e9da5 100644 --- a/crates/dig-node-core/src/seams/content/bandwidth.rs +++ b/crates/dig-node-core/src/seams/content/bandwidth.rs @@ -200,7 +200,10 @@ impl crate::Node { if depth >= crate::download::REDIRECT_HOP_CAP { return None; } - let providers = pc.find_providers(content).await; + // FINDING, not deciding: this picks somebody else to serve from. An unreachable DHT means + // "no alternate known", which falls through to serving over budget rather than dropping the + // caller — the graceful direction. It must never read as "this content exists nowhere". + let providers = pc.find_providers(content).await.for_finding(); if providers.is_empty() { return None; } 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 934ad30e..bff3038f 100644 --- a/crates/dig-node-core/src/seams/dig_peer/mod.rs +++ b/crates/dig-node-core/src/seams/dig_peer/mod.rs @@ -24,6 +24,7 @@ pub mod forwarded_ask; pub mod holder_cache; pub mod holdings; pub mod module_anchor; +pub mod module_relay; pub mod module_reshare; pub mod module_serve; pub mod module_transport; diff --git a/crates/dig-node-core/src/seams/dig_peer/module_relay.rs b/crates/dig-node-core/src/seams/dig_peer/module_relay.rs new file mode 100644 index 00000000..4bffb3d2 --- /dev/null +++ b/crates/dig-node-core/src/seams/dig_peer/module_relay.rs @@ -0,0 +1,106 @@ +//! The MODULE-GRANULARITY relay leg (dig-node#276): serving a whole-`.dig` window to a requestor +//! that asked this node to fetch the capsule on its behalf. +//! +//! # What this is +//! +//! A requestor A wants a capsule held by C but cannot reach C's socket — C is behind a NAT A cannot +//! traverse, or A only ever learned of C through B. A asks B, with `proxy: true`. B does not hold the +//! capsule, so B pulls the WHOLE capsule from C over the ordinary chain-anchored, merkle-verified +//! whole-capsule path ([`CapsuleWarmer`](super::CapsuleWarmer)) and then answers A's module windows +//! from its own cache, byte-identically to a genuine holder. Bytes flow `A <- B <- C`, and **A never +//! learns C's address**. +//! +//! This is the whole-`.dig` twin of the resource-granularity relay that already ships on +//! `dig.fetchRange` ([`crate::download::NodeContent::miss_outcome`], leg 2). The module path had no +//! miss branch at all: it answered `RESOURCE_UNAVAILABLE` unconditionally, which is why requirement 4 +//! of the recursive-download epic — content streaming back THROUGH a hop — did not exist at the +//! granularity a `.dig` download actually uses. +//! +//! # What this deliberately is NOT +//! +//! * **One hop, not a circuit.** `A -> B -> C`, and no further. B knows exactly who asked and for +//! what. This buys the requestor HOLDER-ADDRESS privacy, not sender anonymity. Multi-hop layered +//! circuits are `dig-onion`'s job; nothing here discharges that crate's SPEC. +//! * **Store-and-forward, not pass-through.** B completes and VERIFIES its pull before serving the +//! first byte. That costs first-byte latency on a cold hop and buys reuse of the one whole-capsule +//! path that is already chain-anchored and already audited. +//! * **Capsule bytes ONLY.** A `.dig` is public-by-content-address and the hop is handed +//! `(store_id, root)` and never a retrieval key, so it relays ciphertext it cannot read — exactly +//! what a DHT-discovered holder sees. No directed message, no chat, no mail and no +//! recipient-specific request may ever be routed through this path; those stay end-to-end sealed to +//! the recipient key (NC-1 / §5.4). Widening this path's payload is an NC-1 review, not an +//! extension. +//! +//! # The three gates, each independently sufficient to refuse +//! +//! 1. **The requestor asked.** `params.proxy == true`. Automatic relaying is off; a requestor that +//! can reach holders itself should, and does. +//! 2. **The operator opted in.** [`crate::download::ONION_RELAY_ENV`], default OFF. This leg spends a +//! THIRD party's bandwidth, so it is not gated more loosely than the legs that spend only this +//! node's. +//! 3. **The requestor is within its PROXY allowance** — the separate, tighter bucket +//! (dig_ecosystem#2189), never the cheap-lookup one. Relaying a whole capsule is the costliest +//! thing a stranger can ask this node to do. +//! +//! A refusal at any gate leaves the pre-existing `RESOURCE_UNAVAILABLE` answer exactly as it was, so +//! the requestor stays free to ask a different hop — a hop's "not found" may always be a lie (NC-12). + +use serde_json::Value; + +use crate::download::proxy_requested; +use crate::rate_limit::RequestorId; +use crate::seams::dig_peer::module_reshare::WarmOutcome; +use crate::Node; + +/// Try to make this node able to serve `(store_hex, root_hex)` on `requestor`'s behalf, returning +/// whether the capsule is now in the local cache and may be read from. +/// +/// Awaited, not spawned: the caller has a module window to answer RIGHT NOW, and the answer depends +/// on the pull. This is the store-and-forward cost, paid once per capsule — a second window of the +/// same capsule finds it cached and returns immediately. +/// +/// Every refusal is silent and returns `false`; the caller's own `RESOURCE_UNAVAILABLE` then stands. +/// Never a silent success, and never an unbounded fetch: the pull is the ordinary +/// [`CapsuleWarmer`](super::CapsuleWarmer) one, byte-capped and chain-anchored, and it does NOT make +/// this node a holder ([`HolderClaim::Suppress`](super::module_reshare::HolderClaim)). +pub(crate) async fn relay_capsule( + node: &Node, + store_hex: &str, + root_hex: &str, + params: &Value, + requestor: &RequestorId, +) -> bool { + // (1) The requestor must ASK. Checked first because it is free and because it is the only gate + // whose absence means "this request never wanted a relay" rather than "this node refuses". + if !proxy_requested(params) { + return false; + } + let Some(content) = node.p2p_content() else { + return false; + }; + // (2) The OPERATOR must have opted in. + if !content.onion_relay_enabled() { + return false; + } + // (3) The requestor must be inside its PROXY-class allowance — the expensive-egress bucket. + if !content.allow_proxy_fetch(requestor) { + return false; + } + let Some(warmer) = content.capsule_warmer() else { + // No warmer wired (the FFI/base path): there is no whole-capsule pull to drive, so there is + // no relay. A read behaves identically with or without the leg. + return false; + }; + tracing::debug!( + store = %super::serve_log::SafeId::new(store_hex), + root = %super::serve_log::SafeId::new(root_hex), + "module relay: pulling a capsule this node does not hold, on a requestor's behalf" + ); + // `AlreadyHeld` is admitted alongside `Held` because a concurrent warm may have landed the + // capsule between the caller's miss and this call — the question this function answers is "can the + // window be read now?", not "did I personally pull it?". + matches!( + warmer.warm_relayed(store_hex, root_hex).await, + WarmOutcome::Held { .. } | WarmOutcome::AlreadyHeld + ) +} diff --git a/crates/dig-node-core/src/seams/dig_peer/module_reshare.rs b/crates/dig-node-core/src/seams/dig_peer/module_reshare.rs index 36527a22..fc6aba39 100644 --- a/crates/dig-node-core/src/seams/dig_peer/module_reshare.rs +++ b/crates/dig-node-core/src/seams/dig_peer/module_reshare.rs @@ -216,6 +216,31 @@ pub enum WarmOutcome { AlreadyHeld, } +/// Whether a completed warm makes this node a DISCOVERABLE holder of what it just pulled. +/// +/// The two callers of a warm want opposite answers, and the difference is a security boundary rather +/// than a preference (dig-node#276): +/// +/// * A warm this node's OWN operator provoked — a local read — SHOULD announce. That is the reshare +/// flywheel: every read leaves the content more available than it found it. +/// * A warm a STRANGER provoked, by asking this node to relay a capsule it does not hold, MUST NOT. +/// Announcing it would let any peer drive this node into advertising capsules of the ATTACKER's +/// choosing — a few hundred request bytes in, an attacker-shaped holder inventory out, and eviction +/// pressure on the operator's own content. That is precisely the hole +/// [`crate::download::NodeContent`]'s `origin != Local` reshare refusal exists to close, and +/// relaying reopens it one level up unless the announce is suppressed here. +/// +/// An enum rather than a `bool` so the call site names which of the two it is, and so a future third +/// caller has to CHOOSE rather than inherit whichever default was in the signature. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HolderClaim { + /// Announce the capsule: this node pulled it for ITSELF and is a genuine, willing holder. + Announce, + /// Cache the capsule but announce NOTHING: this node pulled it on a stranger's behalf and is a + /// relay, not a holder. + Suppress, +} + /// Where a capsule warm stages + promotes to, and how it announces. /// /// A struct rather than a long argument list so the call site reads as one intention, and so the @@ -376,8 +401,31 @@ impl CapsuleWarmer { /// Callers on the read path use [`spawn_capsule_warm`] instead; this is the awaitable core so the /// behaviour is testable without a background task. pub async fn warm(self: &Arc, store_hex: &str, root_hex: &str) -> WarmOutcome { + self.warm_claiming(store_hex, root_hex, HolderClaim::Announce) + .await + } + + /// [`warm`](Self::warm) for a capsule pulled on ANOTHER node's behalf (dig-node#276): identical in + /// every trust step — chain anchor, merkle verification, promote-recheck, cache bound — except that + /// this node does **not** announce itself as a holder of the result. + /// + /// The capsule still lands in the cache, because that is what lets the relayed module windows be + /// served from the same code path a genuine holder serves from, byte-identically. What it does not + /// do is make a stranger's choice of content into this node's advertised inventory. + pub async fn warm_relayed(self: &Arc, store_hex: &str, root_hex: &str) -> WarmOutcome { + self.warm_claiming(store_hex, root_hex, HolderClaim::Suppress) + .await + } + + /// The shared body of [`warm`](Self::warm) and [`warm_relayed`](Self::warm_relayed). + async fn warm_claiming( + self: &Arc, + store_hex: &str, + root_hex: &str, + claim: HolderClaim, + ) -> WarmOutcome { let outcome = self - .warm_with_config(store_hex, root_hex, self.config.clone()) + .warm_with_config(store_hex, root_hex, self.config.clone(), claim) .await; // #2053: the tier-aware `/modules` size-cap sweep, run ONLY after a land that actually // grew the cache (`Held`) — a refusal wrote nothing, so there is nothing new to bound. This @@ -414,7 +462,8 @@ impl CapsuleWarmer { // tier-0 round can never pull more than its remaining sub-budget even if the node default is // larger. config.max_module_size = config.max_module_size.min(max_bytes); - self.warm_with_config(store_hex, root_hex, config).await + self.warm_with_config(store_hex, root_hex, config, HolderClaim::Announce) + .await } /// The awaitable core of [`warm`](Self::warm) / [`warm_capped`](Self::warm_capped), parameterized by @@ -424,6 +473,7 @@ impl CapsuleWarmer { store_hex: &str, root_hex: &str, config: dig_download::ModuleDownloadConfig, + claim: HolderClaim, ) -> WarmOutcome { // Already a holder → nothing to pull, nothing to announce again. Checked BEFORE claiming a // registry slot: a burst of reads across an already-cached capsule should cost one stat call @@ -492,13 +542,19 @@ impl CapsuleWarmer { match promote_into_cache(&staged, &cached, &verifier) { Ok(promoted) => { discard_staging(&staged); - self.announce.announce_inventory().await; + // The ONE step a relayed warm skips. Everything above it — the chain anchor, the + // merkle verification, the promote-recheck — ran identically, so the bytes are equally + // trustworthy; what differs is whether this node CLAIMS them (see [`HolderClaim`]). + if claim == HolderClaim::Announce { + self.announce.announce_inventory().await; + } tracing::info!( store = %super::serve_log::SafeId::new(store_hex), root = %super::serve_log::SafeId::new(root_hex), outcome = "held", bytes = promoted, - "capsule warm: whole capsule verified + cached; announced as a holder" + announced = claim == HolderClaim::Announce, + "capsule warm: whole capsule verified + cached" ); WarmOutcome::Held { bytes: promoted } } @@ -1021,6 +1077,118 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + /// Build a warmer over a REAL, answering holder for `(STORE, chain_root())`, staging + caching + /// under `dir`, announcing through `spy`. + /// + /// Extracted so the relay pair below can hold every input constant and vary exactly ONE thing — + /// which entry point is called. Two independently-constructed warmers would leave "the relayed one + /// simply had no working holder" as an untested explanation for its silence. + fn serving_warmer(dir: &Path, spy: &Arc, module: Vec) -> Arc { + let (store_hex, root_hex) = (hex32(STORE), hex32(chain_root())); + let content = dig_download::module_content_id(&store_hex, &root_hex) + .expect("canonical ids yield a content id"); + CapsuleWarmer::new( + Arc::new(dig_download::testkit::MockProviderLocator::fixed( + dig_download::testkit::mock_providers(1, &content), + )), + Arc::new(dig_download::testkit::MockModuleTransport::serving( + &store_hex, &root_hex, module, 8, + )), + Arc::new(dig_download::InMemoryStateStore::new()), + Arc::new(ConfirmingResolver), + WarmPaths { + staging_dir: dir.join("staging"), + cache_dir: dir.join("cache"), + }, + Arc::clone(spy) as Arc, + Arc::new(WarmRegistry::new()), + dig_download::ModuleDownloadConfig::default(), + Arc::new(crate::tier0_live::NoopModulesEvictor), + ) + } + + /// The cache path whose EXISTENCE is this node's holder claim, under `dir`. + fn cached_module_path(dir: &Path) -> std::path::PathBuf { + let (store_hex, root_hex) = (hex32(STORE), hex32(chain_root())); + dir.join("cache") + .join("modules") + .join(&store_hex) + .join(format!("{root_hex}.dig")) + } + + /// **Proves (dig-node#276, unit 4):** a capsule pulled ON A STRANGER'S BEHALF lands in the cache — + /// so the relayed windows can be served from it — and announces NOTHING, while the *same pull, + /// through the same holder, of the same bytes*, driven for this node's OWN sake announces exactly + /// once. + /// + /// **Catches:** the amplification hole the relay leg would otherwise reopen one level up. The + /// `origin != Local` reshare refusal exists so a stranger cannot drive this node into caching AND + /// DHT-announcing capsules of the attacker's choosing; a relay that pulls a whole capsule for a + /// stranger and then announces it hands that exact primitive back, with no forged message and no + /// privileged access required. + /// + /// **Why BOTH halves, and why they share `serving_warmer`:** an assertion that the relayed pull + /// announces zero times is satisfied identically by a suppression that works and by a warmer whose + /// announce is broken, whose holder never answers, or whose chain never confirms — every one of + /// which would also make a legitimate reshare silent. The `Announce` half is the truthful control + /// that distinguishes them: it is the same code, the same fixture and the same holder, differing + /// only in the [`HolderClaim`] the entry point names. + #[tokio::test] + async fn a_relayed_capsule_is_cached_without_announcing_while_a_local_one_announces() { + let (store_hex, root_hex) = (hex32(STORE), hex32(chain_root())); + let module = module_committing(STORE, chain_root()); + + // RELAYED — pulled for a stranger. + let relay_dir = temp_dir("relayed-warm"); + let relay_spy = Arc::new(AnnounceSpy::default()); + let relayed = serving_warmer(&relay_dir, &relay_spy, module.clone()) + .warm_relayed(&store_hex, &root_hex) + .await; + + // LOCAL — the identical pull, for this node's own sake. The control. + let local_dir = temp_dir("local-warm"); + let local_spy = Arc::new(AnnounceSpy::default()); + let local = serving_warmer(&local_dir, &local_spy, module.clone()) + .warm(&store_hex, &root_hex) + .await; + + let held = WarmOutcome::Held { + bytes: module.len() as u64, + }; + assert_eq!( + local, held, + "the control must genuinely succeed, or its announce count proves nothing" + ); + assert_eq!( + relayed, held, + "a relayed pull still verifies and caches — it is the holder CLAIM that is withheld" + ); + + assert_eq!( + local_spy.calls.load(Ordering::SeqCst), + 1, + "a warm this node drove for itself announces exactly once" + ); + assert_eq!( + relay_spy.calls.load(Ordering::SeqCst), + 0, + "a warm driven by a stranger must never advertise this node as a holder of it" + ); + + // The bytes ARE cached in both cases: the relay serves its requestor's windows from the same + // artifact a holder serves from, byte-identically, so the requestor needs no second code path. + for dir in [&relay_dir, &local_dir] { + assert_eq!( + std::fs::read(cached_module_path(dir)).expect("module is at the cache path"), + module, + "the verified capsule is cached whether or not it was announced" + ); + } + + let _ = std::fs::remove_dir_all(&relay_dir); + let _ = std::fs::remove_dir_all(&local_dir); + } + /// A [`ModulesCacheEvictor`](crate::tier0_live::ModulesCacheEvictor) that counts sweeps, so /// "the reshare-warm land triggered exactly one sweep" (and "a refusal triggered none") are /// assertable properties without a Node. diff --git a/crates/dig-node-core/src/seams/dig_peer/module_transport.rs b/crates/dig-node-core/src/seams/dig_peer/module_transport.rs index c1eb538e..648d45da 100644 --- a/crates/dig-node-core/src/seams/dig_peer/module_transport.rs +++ b/crates/dig-node-core/src/seams/dig_peer/module_transport.rs @@ -42,6 +42,7 @@ use dig_peer::DigPeer; use dig_rpc_protocol::types::{FetchModuleRangeParams, GetModuleInfoParams, ModuleInfo}; use super::pool_locator::ConnectedPool; +use crate::download::BestEffort; /// The peer-RPC transport the module pull rides. /// @@ -103,7 +104,11 @@ impl NatModuleTransport { })?; let mut record_addrs: Vec = self.pool_candidates(peer_hex); - record_addrs.extend(self.discovered_candidates(peer_hex, store_id, root).await); + record_addrs.extend( + self.discovered_candidates(peer_hex, store_id, root) + .await + .for_finding(), + ); // Order + cap the merged candidate set through dig-download's ONE resolver: IPv6 before IPv4 // (§5.2), and each socket CONSTRUCTED from a parsed IpAddr rather than a formatted string @@ -153,25 +158,34 @@ impl NatModuleTransport { .unwrap_or_default() } - /// The peer's advertised addresses from capsule-granularity discovery (empty on any failure — - /// discovery is best-effort; the pool addresses above must never be lost to a DHT error). + /// The peer's advertised addresses from capsule-granularity discovery. + /// + /// Best-effort by design: the live pool addresses this enriches must never be lost to a DHT + /// error. But the emptiness it can produce is ambiguous — a walk that failed and a walk that + /// found no advertisement look identical as a bare `Vec` — so the failure is carried out in a + /// [`BestEffort`] rather than discarded, and a caller can only read an absence from it through + /// `absence_established()` (dig-node#296). async fn discovered_candidates( &self, peer_hex: &str, store_id: &str, root: &str, - ) -> Vec { + ) -> BestEffort { let Some(content) = module_content_id(store_id, root) else { - return Vec::new(); + // A non-canonical id was never asked about, so nothing was found AND nothing failed: + // this is a genuine "no advertisement", not an unreachable source. + return BestEffort::found(Vec::new()); }; let Ok(records) = self.locator.find_providers(&content).await else { - return Vec::new(); + return BestEffort::source_failed(); }; - records - .into_iter() - .find(|r| r.provider_peer_id == peer_hex) - .map(|r| r.addresses) - .unwrap_or_default() + BestEffort::found( + records + .into_iter() + .find(|r| r.provider_peer_id == peer_hex) + .map(|r| r.addresses) + .unwrap_or_default(), + ) } /// Dial `peer_hex`, trying every candidate in order and reporting the LAST failure with the address @@ -211,6 +225,118 @@ impl NatModuleTransport { } } +/// The largest framed JSON body accepted for a module DESCRIPTOR answer. +/// +/// The generic peer-request reader ([`crate::peer::read_framed`]) caps at 64 KiB, which is right for +/// a REQUEST and too small for this RESPONSE: a descriptor declares one 32-byte hash and one length +/// per chunk, so the largest permitted module runs to a few hundred kilobytes of JSON. This is still +/// a hard bound — a peer cannot make this node buffer an arbitrary body by declaring one. +const MAX_DESCRIPTOR_FRAME: usize = 8 * 1024 * 1024; + +/// Ask `stream` a whole-`.dig` question as a framed JSON-RPC request carrying the RELAY opt-in +/// (dig-node#276). +/// +/// # Why this node frames the request itself instead of calling dig-peer's typed method +/// +/// `GetModuleInfoParams` / `FetchModuleRangeParams` live in `dig-rpc-protocol` and carry no `proxy` +/// field. Adding one is a level-00 crate change and a release-first cascade through `dig-peer` -> +/// `dig-download` -> this repo, for a single boolean on a request this repo both sends and serves. +/// dig-peer's own [`DigPeer::open_stream`] is the documented escape hatch for exactly this — a +/// consumer carrying its own wire shape over the authenticated mux — and the typed method it replaces +/// is itself only a `build_request` plus a framed write over that same stream. +/// +/// The flag is ADDITIVE: a peer that does not implement the relay ignores an unknown params key and +/// answers precisely as it does today, so this is safe to send to every holder unconditionally. +async fn ask_with_relay( + stream: &mut dig_nat::PeerStream, + method: dig_rpc_protocol::Method, + mut params: serde_json::Value, +) -> std::io::Result<()> { + if let Some(object) = params.as_object_mut() { + // A whole-`.dig` download defaults to ONION mode per the recursive-download epic: if the + // holder we reached does not hold it, we would rather it fetched the capsule for us than tell + // us "not found" while sitting one hop from someone who has it. Individual RESOURCE requests + // are unaffected and still default to DIRECT (NC-4). + object.insert("proxy".to_string(), serde_json::Value::Bool(true)); + } + crate::peer::write_framed( + stream, + &serde_json::json!({ + "jsonrpc": "2.0", + "id": 1, + "method": method.name(), + "params": params, + }), + ) + .await +} + +/// Read one framed JSON-RPC response body from `stream`, bounded by [`MAX_DESCRIPTOR_FRAME`]. +async fn read_response_frame( + stream: &mut dig_nat::PeerStream, +) -> std::io::Result { + use tokio::io::AsyncReadExt; + + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf).await?; + let len = u32::from_be_bytes(len_buf) as usize; + if len > MAX_DESCRIPTOR_FRAME { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "module descriptor frame too large", + )); + } + let mut body = vec![0u8; len]; + stream.read_exact(&mut body).await?; + serde_json::from_slice(&body) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e)) +} + +/// One `dig.getModuleInfo` over `peer`, carrying the relay opt-in, decoded into a [`ModuleInfo`]. +/// +/// `None` for every failure — a refused stream, an unwritable request, an unreadable frame, an error +/// envelope, or a body that is not a descriptor. The caller turns that into one transport error whose +/// text is this node's own, so a peer can never author what this node logs (#1603). +async fn descriptor_over(peer: &mut DigPeer, store_id: &str, root: &str) -> Option { + let params = serde_json::to_value(GetModuleInfoParams { + store_id: store_id.to_string(), + root: root.to_string(), + }) + .ok()?; + let mut stream = peer.open_stream().await.ok()?; + ask_with_relay(&mut stream, dig_rpc_protocol::Method::GetModuleInfo, params) + .await + .ok()?; + let response = read_response_frame(&mut stream).await.ok()?; + serde_json::from_value(response.get("result")?.clone()).ok() +} + +/// Open a `dig.fetchModuleRange` frame stream over `peer`, carrying the relay opt-in. +async fn window_stream_over( + peer: &mut DigPeer, + store_id: &str, + root: &str, + offset: u64, + length: u64, +) -> Option { + let params = serde_json::to_value(FetchModuleRangeParams { + store_id: store_id.to_string(), + root: root.to_string(), + offset: Some(offset), + length, + }) + .ok()?; + let mut stream = peer.open_stream().await.ok()?; + ask_with_relay( + &mut stream, + dig_rpc_protocol::Method::FetchModuleRange, + params, + ) + .await + .ok()?; + Some(stream) +} + #[async_trait] impl ModuleTransport for NatModuleTransport { async fn get_module_info( @@ -220,16 +346,11 @@ impl ModuleTransport for NatModuleTransport { root: &str, ) -> Result { let mut peer = self.connect(provider_peer_id, store_id, root).await?; - let result = peer - .get_module_info(&GetModuleInfoParams { - store_id: store_id.to_string(), - root: root.to_string(), - }) - .await; + let result = descriptor_over(&mut peer, store_id, root).await; peer.disconnect().await; // The reason names the STEP and the sentinelled peer; the peer's own answer text is never // embedded (#1603) — the crate sanitizes at its Display layer and upstream must not defeat it. - result.map_err(|_| DownloadError::transport(provider_peer_id, "getModuleInfo failed")) + result.ok_or_else(|| DownloadError::transport(provider_peer_id, "getModuleInfo failed")) } async fn fetch_module_range( @@ -241,17 +362,9 @@ impl ModuleTransport for NatModuleTransport { length: u64, ) -> Result, DownloadError> { let mut peer = self.connect(provider_peer_id, store_id, root).await?; - let stream = peer - .fetch_module_range(&FetchModuleRangeParams { - store_id: store_id.to_string(), - root: root.to_string(), - offset: Some(offset), - length, - }) - .await; - let bytes = match stream { - Ok(mut stream) => read_module_window(&mut stream, provider_peer_id, length).await, - Err(_) => Err(DownloadError::transport( + let bytes = match window_stream_over(&mut peer, store_id, root, offset, length).await { + Some(mut stream) => read_module_window(&mut stream, provider_peer_id, length).await, + None => Err(DownloadError::transport( provider_peer_id, "fetchModuleRange stream refused", )), diff --git a/crates/dig-node-core/src/seams/dig_rpc/dispatch.rs b/crates/dig-node-core/src/seams/dig_rpc/dispatch.rs index 598fb6eb..d6e19d06 100644 --- a/crates/dig-node-core/src/seams/dig_rpc/dispatch.rs +++ b/crates/dig-node-core/src/seams/dig_rpc/dispatch.rs @@ -230,7 +230,7 @@ impl RpcDispatch for Node { } Some(Method::GetModuleInfo) => { let params = req.get("params").cloned().unwrap_or(json!({})); - return node.get_module_info(¶ms, id).await; + return node.get_module_info(¶ms, id, &requestor).await; } // dig.fetchModuleRange (#1576): one window of a held `.dig` module. // @@ -242,7 +242,7 @@ impl RpcDispatch for Node { // (§6.2) without implementing the frame protocol. Some(Method::FetchModuleRange) => { let params = req.get("params").cloned().unwrap_or(json!({})); - return node.fetch_module_range_frame(¶ms, id).await; + return node.fetch_module_range_frame(¶ms, id, &requestor).await; } // dig.stage (#95 Pass C): turn a local folder into a capsule (.dig module) IN // PROCESS — the staging/compile half of a local deploy. The DIG Browser's