diff --git a/docs/PRICE_PROVIDERS.md b/docs/PRICE_PROVIDERS.md index 39e5bd1e..b277dc3e 100644 --- a/docs/PRICE_PROVIDERS.md +++ b/docs/PRICE_PROVIDERS.md @@ -480,6 +480,14 @@ only = ["CUP", "MLC"] # El Toque is only meaningful for these (§6.6) Phases 3 and 4 both depend on Phase 2 and can land in either order. +> **Out-of-band addition — Nostr *subscribe* (issue #697, §11.7).** The +> five phases above only ever covered *publishing* rates to Nostr (Phase 5). +> A separate, independently-landing addition — the `nostr` provider — makes +> `mostrod` able to *consume* another Mostro node's published rates instead +> of an HTTP API, for operators in regions where the price APIs are +> network-blocked. It slots into the existing registry (§5.4) like any +> other provider and does not otherwise depend on Phase 5. + --- ## 9. Phase details @@ -838,6 +846,52 @@ BTCPayServer solves the same problem; contrasting choices clarify ours: is not reintroduced by accident when porting a provider idea from BTCPay. +### 11.7 Nostr (subscribe, trusted-node relay mode) — issue #697 + +Every source above is an HTTP API. This one is not: it sources quotes from +**other Mostro nodes**, over Nostr, for operators whose network blocks the +HTTP price APIs outright (the motivating case: `api.yadio.io` is DNS-blocked +for Venezuelan ISPs) but who can still reach a Nostr relay. + +- **Not a new publish path.** Mostro nodes already publish their aggregated + rates as a kind-30078 NIP-33 event (`d = "mostro-rates"`, content + `{"BTC": {ccy: price}}`) — see `docs/NOSTR_EXCHANGE_RATES.md` and + `nip33::new_exchange_rates_event`. The `nostr` provider is a **consumer** + of that exact same event, from other operators' nodes instead of its own. +- **Config (`[price.providers.nostr]`, §7):** no `url` — `trusted_nodes` (a + list of hex pubkeys) is the provider's only required field. Enabled + without at least one valid `trusted_nodes` pubkey is a startup error + (mirrors El Toque's missing-token fail-fast, §7). +- **Relays are not reconfigured.** The provider reuses the process-wide + Nostr client already connected to the node's own `[nostr]` relays — it + does not open a second set of connections or accept a separate relay list. +- **One-shot query per tick, not a persistent subscription.** `fetch()` is + called once per `update_interval_seconds` tick like every other provider + (spec §5.3); it issues a single `kind=30078, authors=trusted_nodes, + #d=mostro-rates` relay query bounded by a short fixed timeout, rather than + keeping an open `REQ` subscription running between ticks. +- **Freshest wins — no cross-node combine.** With several `trusted_nodes` + configured, the provider does **not** run the §6.2 median+outlier combine + across their individual events; it takes the single event with the + highest `created_at` among the trusted, valid ones. Rationale (confirmed + with the maintainer): the issue frames multiple `trusted_nodes` as + *redundancy* (if one is down or stale, use another), not as independent + samples to statistically combine, and skipping a second aggregation layer + inside a single provider keeps the adapter's contract identical to every + HTTP provider (one `ProviderQuotes` map out of `fetch()`, no new state). + A compromised single trusted node can therefore still push a bad price + with no in-provider outlier guard; operators mitigate by choosing + `trusted_nodes` they actually trust, same as any other single-sourced + provider. +- **Client-side pubkey verification.** Even though the relay-side `authors` + filter already restricts the query, the adapter re-checks each returned + event's `pubkey` against `trusted_nodes` before trusting its content — + the same verification `docs/NOSTR_EXCHANGE_RATES.md` requires of mobile + clients, applied here on the operator side. +- **Currency codes are re-upper-cased** (§6.6) even though a Mostro + publisher already emits uppercase codes — a third-party trusted node is + outside this codebase's control. + --- ## 12. Tracking diff --git a/settings.tpl.toml b/settings.tpl.toml index dfe67b2d..ee49cd9a 100644 --- a/settings.tpl.toml +++ b/settings.tpl.toml @@ -180,6 +180,17 @@ port = 50051 # url = "https://tasas.eltoque.com" # # token = "xxxx" # REQUIRED when enabled; provider refuses to start otherwise # only = ["CUP", "MLC"] +# +# # Subscribe to prices published (kind 30078) by trusted Mostro nodes over +# # Nostr instead of an HTTP API — for operators in regions where price APIs +# # are DNS/IP blocked (docs/PRICE_PROVIDERS.md §11.7). Reuses the relays +# # already configured in [nostr]; no url needed. With several trusted_nodes, +# # the freshest valid event wins — not a cross-node combine. +# [price.providers.nostr] +# enabled = false +# trusted_nodes = [ +# # hex pubkeys of Mostro nodes you trust to publish accurate rates +# ] # Anti-abuse bond (issue #711). Opt-in, disabled by default. Uncomment to # require a Lightning hold-invoice bond from takers and/or makers. See diff --git a/src/price/aggregate.rs b/src/price/aggregate.rs index 89809c79..d8d909ce 100644 --- a/src/price/aggregate.rs +++ b/src/price/aggregate.rs @@ -14,11 +14,20 @@ use super::provider::{ProviderId, ProviderQuotes, Quote}; /// value actually **survived** [`combine`]'s outlier filter — what the /// Nostr `source` tag really means. The two can differ: with three /// providers and one outlier the count is 3 but contributors is 2. +/// +/// `nostr_anchor_dependent` is true when a fiat-cross (`PerBase`) path for +/// this currency resolved against an anchor that itself included `Nostr` +/// among its surviving direct contributors. The absolute per-BTC figure +/// then embeds a relayed rate even if `contributors` only names the +/// cross provider (e.g. El Toque × Nostr USD → CUP tagged `eltoque`). +/// [`crate::price::manager`] uses this so republication cannot re-stamp +/// that hybrid as a fresh local observation (PR #841 re-review). #[derive(Debug, Clone, PartialEq)] pub struct AggregateResult { pub value: f64, pub sources: u8, pub contributors: Vec, + pub nostr_anchor_dependent: bool, } /// Combine a currency's candidate per-BTC prices into one figure (spec §6.2). @@ -111,8 +120,9 @@ pub fn resolve_per_base( /// their provider id**. /// 2. Per-currency **anchors** are the [`combine`]d direct quotes; fiat-cross /// (`PerBase`) quotes are resolved against those anchors and attributed -/// to the fiat-cross provider (the anchor's own contributors are an -/// intermediate, not a contributor to the resolved currency). +/// to the fiat-cross provider (the anchor's own contributors stay out of +/// `contributors` — they are an intermediate of the cross math — but a +/// Nostr-touched anchor sets `nostr_anchor_dependent` on the result). /// 3. Each currency's final value is the [`combine`] of its direct **and** /// resolved candidates. `contributors` lists the providers whose value /// survived the outlier filter ([`kept_contributors`]) — what the Nostr @@ -138,20 +148,23 @@ pub fn aggregate_tick( } } - // Step 2: anchors = aggregated direct quotes (values only), then - // resolve cross quotes — attributing each resolved candidate to the - // fiat-cross provider that emitted it. Anchor contributors are *not* - // propagated into resolved-currency contributors: they are an - // intermediate of the cross math, not an upstream of the cross - // currency. + // Step 2: anchors = aggregated direct quotes, then resolve cross quotes. + // Track whether each anchor's surviving contributors include Nostr so a + // later PerBase resolution can mark the cross currency as + // `nostr_anchor_dependent` without putting Nostr into `contributors` + // (which would change the published `source` tag's meaning). let mut anchors: HashMap = HashMap::new(); + let mut anchor_uses_nostr: HashMap = HashMap::new(); for (currency, pairs) in &direct { let values: Vec = pairs.iter().map(|(_, v)| *v).collect(); if let Some(v) = combine(&values, outlier_pct) { anchors.insert(currency.clone(), v); + let kept = kept_contributors(pairs, outlier_pct); + anchor_uses_nostr.insert(currency.clone(), kept.contains(&ProviderId::Nostr)); } } let mut resolved: HashMap> = HashMap::new(); + let mut resolved_nostr_anchor: HashMap = HashMap::new(); for (id, currency, base, value) in &per_base { if let Some(anchor) = anchors.get(base) { let candidate = value * anchor; @@ -160,6 +173,9 @@ pub fn aggregate_tick( .entry(currency.clone()) .or_default() .push((*id, candidate)); + if *anchor_uses_nostr.get(base).unwrap_or(&false) { + resolved_nostr_anchor.insert(currency.clone(), true); + } } } } @@ -191,6 +207,7 @@ pub fn aggregate_tick( value, sources, contributors, + nostr_anchor_dependent: *resolved_nostr_anchor.get(currency).unwrap_or(&false), }, ); } @@ -386,6 +403,44 @@ mod tests { out["CUP"].contributors, vec![ProviderId::Yadio, ProviderId::ElToque] ); + assert!( + !out["CUP"].nostr_anchor_dependent, + "Yadio/CoinGecko USD anchor is local — CUP cross is not Nostr-tainted" + ); + } + + #[test] + fn aggregate_tick_marks_cross_resolved_via_nostr_usd_as_anchor_dependent() { + // Nostr is the only USD source; El Toque CUP/USD resolves against it. + // contributors stay [ElToque] (anchor providers are intermediates), + // but nostr_anchor_dependent must be set so republication cannot + // re-stamp the hybrid as a fresh local observation. + let mut nostr = ProviderQuotes::new(); + nostr.insert("USD".into(), Quote::PerBtc(50_000.0)); + let mut eltoque = ProviderQuotes::new(); + eltoque.insert( + "CUP".into(), + Quote::PerBase { + base: "USD".into(), + value: 400.0, + }, + ); + + let out = aggregate_tick( + &[(ProviderId::Nostr, nostr), (ProviderId::ElToque, eltoque)], + PCT, + ); + + approx(out["CUP"].value, 20_000_000.0); + assert_eq!(out["CUP"].contributors, vec![ProviderId::ElToque]); + assert!( + out["CUP"].nostr_anchor_dependent, + "CUP absolute value embeds Nostr USD and must be flagged" + ); + assert!( + !out["USD"].nostr_anchor_dependent, + "direct Nostr USD is Nostr-only via contributors, not via anchor taint" + ); } #[test] diff --git a/src/price/config.rs b/src/price/config.rs index 6e77e76e..3676598d 100644 --- a/src/price/config.rs +++ b/src/price/config.rs @@ -47,7 +47,10 @@ pub struct ProviderConfig { /// Whether this provider participates in aggregation. #[serde(default)] pub enabled: bool, - /// Primary base URL. + /// Primary base URL. Not needed by a provider sourcing quotes over Nostr + /// instead of HTTP (see `trusted_nodes`) — defaults to empty so those + /// providers can omit it. + #[serde(default)] pub url: String, /// Ordered mirrors tried when `url` fails this tick (spec §7). #[serde(default)] @@ -65,6 +68,10 @@ pub struct ProviderConfig { /// Exclude these currencies from this provider (spec §6.6). #[serde(default)] pub except: Option>, + /// Trusted node pubkeys (hex) to source quotes from over Nostr, instead + /// of an HTTP `url` (the `nostr` provider; see §11.7). + #[serde(default)] + pub trusted_nodes: Vec, } impl ProviderConfig { @@ -78,11 +85,25 @@ impl ProviderConfig { (see docs/PRICE_PROVIDERS.md §7)" )); } + // `trusted_nodes` only excuses a missing `url` for the `nostr` + // provider — it's the only adapter that reads it (§11.7). Any other + // id is still HTTP-only, so exempting it here would let a typo'd or + // copy-pasted `trusted_nodes` mask a missing `url` past startup and + // into a confusing per-tick HTTP failure instead. if self.enabled && self.url.trim().is_empty() { - return Err(format!( - "price provider '{id}': enabled provider must have a non-empty `url` \ - (see docs/PRICE_PROVIDERS.md §7)" - )); + if id == "nostr" { + if self.trusted_nodes.is_empty() { + return Err(format!( + "price provider '{id}': enabled provider must have a non-empty \ + `trusted_nodes` (see docs/PRICE_PROVIDERS.md §11.7)" + )); + } + } else { + return Err(format!( + "price provider '{id}': enabled provider must have a non-empty `url` \ + (see docs/PRICE_PROVIDERS.md §7)" + )); + } } Ok(()) } @@ -217,13 +238,21 @@ enabled = false url = "https://tasas.eltoque.com" token = "secret" only = ["CUP", "MLC"] + +[price.providers.nostr] +enabled = false +trusted_nodes = ["82fa8cb978b43c79b2156585bac2c011176a21d2aead6d9f7c575c005be88390"] "#; let parsed: Stub = toml::from_str(toml_str).unwrap(); let p = parsed.price; // Overridden value + defaulted siblings. assert_eq!(p.update_interval_seconds, 60); assert_eq!(p.max_price_staleness_seconds, 1800); - assert_eq!(p.providers.len(), 3); + assert_eq!(p.providers.len(), 4); + + let nostr = &p.providers["nostr"]; + assert!(nostr.url.is_empty()); + assert_eq!(nostr.trusted_nodes.len(), 1); let ca = &p.providers["currency_api"]; assert!(ca.enabled); @@ -251,6 +280,7 @@ only = ["CUP", "MLC"] token: None, only: Some(vec!["CUP".into()]), except: Some(vec!["MLC".into()]), + trusted_nodes: vec![], }; assert!(cfg.validate("eltoque").is_err()); } @@ -298,6 +328,7 @@ only = ["CUP", "MLC"] token: None, only: None, except: None, + trusted_nodes: vec![], }; assert!(blank.validate("yadio").is_err()); // A disabled provider with a blank url is allowed (inert). @@ -308,6 +339,55 @@ only = ["CUP", "MLC"] disabled.validate("yadio").unwrap(); } + #[test] + fn enabled_provider_with_trusted_nodes_needs_no_url() { + let cfg = ProviderConfig { + enabled: true, + url: String::new(), + fallback_urls: vec![], + api_key: None, + token: None, + only: None, + except: None, + trusted_nodes: vec!["a".repeat(64)], + }; + cfg.validate("nostr").unwrap(); + } + + #[test] + fn trusted_nodes_does_not_excuse_a_missing_url_on_a_non_nostr_provider() { + // Regression: `trusted_nodes` only means anything to the `nostr` + // adapter. A copy-paste onto e.g. `yadio` must not silently pass + // validation and defer the failure to a confusing per-tick HTTP + // error against an empty URL. + let cfg = ProviderConfig { + enabled: true, + url: String::new(), + fallback_urls: vec![], + api_key: None, + token: None, + only: None, + except: None, + trusted_nodes: vec!["a".repeat(64)], + }; + assert!(cfg.validate("yadio").is_err()); + } + + #[test] + fn enabled_provider_without_url_or_trusted_nodes_is_rejected() { + let cfg = ProviderConfig { + enabled: true, + url: String::new(), + fallback_urls: vec![], + api_key: None, + token: None, + only: None, + except: None, + trusted_nodes: vec![], + }; + assert!(cfg.validate("nostr").is_err()); + } + #[test] fn no_scoping_allows_everything() { let cfg = ProviderConfig { @@ -318,6 +398,7 @@ only = ["CUP", "MLC"] token: None, only: None, except: None, + trusted_nodes: vec![], }; assert!(cfg.allows_currency("USD")); assert!(cfg.allows_currency("CUP")); diff --git a/src/price/manager.rs b/src/price/manager.rs index 423cf308..5e19666f 100644 --- a/src/price/manager.rs +++ b/src/price/manager.rs @@ -34,14 +34,17 @@ use mostro_core::error::{MostroError, ServiceError}; use nostr_sdk::prelude::*; use tracing::{debug, error, info, warn}; -use super::aggregate::{aggregate_tick, AggregateResult}; +use super::aggregate::{aggregate_tick, combine, AggregateResult}; use super::config::{PriceSettings, ProviderConfig}; use super::fiat::is_known_fiat; -use super::provider::{PriceProvider, ProviderError, ProviderHealth, ProviderId, ProviderQuotes}; +use super::provider::{ + PriceProvider, ProviderError, ProviderHealth, ProviderId, ProviderQuotes, Quote, +}; use super::providers::blockchain::BlockchainProvider; use super::providers::coingecko::CoinGeckoProvider; use super::providers::currency_api::CurrencyApiProvider; use super::providers::eltoque::ElToqueProvider; +use super::providers::nostr::NostrProvider; use super::providers::yadio::YadioProvider; use super::store::{PriceError, PriceStore}; @@ -107,7 +110,12 @@ impl PriceManager { } match id_str.parse::() { Ok(id) => { - let provider = build_provider(id, cfg)?; + let provider = build_provider( + id, + cfg, + settings.provider_timeout_seconds, + settings.max_price_staleness_seconds, + )?; providers.push(EnabledProvider { id, provider, @@ -282,6 +290,15 @@ impl PriceManager { (id, self.scope_quotes(id, fiat_only)) }) .collect(); + // Nostr is a relay of another node's *already-aggregated* rate, not + // an independent peer observation (re-review, PR #841): that node's + // own direct sources already fed into the value it published, so + // blending it into the median/mean here alongside this node's own + // direct sources would double-count correlated information and skew + // the result. Keep it strictly fallback — only for currencies no + // other provider covered this tick. + let filtered_with_ids = + restrict_nostr_to_fallback(filtered_with_ids, self.settings.outlier_threshold_pct); let aggregates = aggregate_tick(&filtered_with_ids, self.settings.outlier_threshold_pct); if aggregates.is_empty() { @@ -309,8 +326,7 @@ impl PriceManager { report.contributors = contributors; if self.settings.publish_to_nostr { - self.publish_rates_to_nostr(&aggregates, &report.contributors) - .await; + self.publish_rates_to_nostr(&aggregates).await; } report @@ -485,16 +501,19 @@ impl PriceManager { /// **contributing** provider ids (spec §9 Phase 1: still effectively /// one source, but the multi-source shape is in place). Publishing is /// best-effort and never fails the tick. - async fn publish_rates_to_nostr( - &self, - aggregates: &HashMap, - successes: &[ProviderId], - ) { + async fn publish_rates_to_nostr(&self, aggregates: &HashMap) { // Build the `{"BTC": {ccy: value}}` body the legacy format used. - let rates: HashMap = aggregates - .iter() - .map(|(c, a)| (c.clone(), a.value)) - .collect(); + let rates = republishable_rates(aggregates); + if rates.is_empty() { + debug!( + "price: nothing to republish to Nostr this tick \ + (every fresh currency was Nostr-only sourced)" + ); + return; + } + // Derived before `rates` is moved into the body it describes. + let source_tag = sources_to_tag(&republished_contributors(aggregates, &rates)); + let mut wrapper: HashMap> = HashMap::new(); wrapper.insert("BTC".to_string(), rates); @@ -518,7 +537,6 @@ impl PriceManager { // Match legacy bitcoin_price.rs: 2× the interval, capped at 1h. let expiration_seconds = std::cmp::min(self.settings.update_interval_seconds * 2, 3600); let expiration = timestamp + expiration_seconds as i64; - let source_tag = sources_to_tag(successes); let tags = Tags::from_list(vec![ Tag::custom( TagKind::Custom("published_at".into()), @@ -548,7 +566,7 @@ impl PriceManager { match tokio::time::timeout(timeout_duration, client.send_event(&event)).await { Ok(Ok(output)) => info!( "price: published exchange rates to Nostr ({} currencies). Output: {:?}", - aggregates.len(), + wrapper["BTC"].len(), output ), Ok(Err(e)) => error!("price: send_event to relays failed: {e}"), @@ -557,6 +575,142 @@ impl PriceManager { } } +/// Drop the `Nostr` provider's quotes for any currency another provider +/// can actually produce a value for this tick, leaving every other +/// provider untouched. +/// +/// A trusted-node `nostr` quote is a relayed *aggregate*, not an +/// independent local observation — the remote node already ran its own +/// direct sources through `combine` to produce it. Feeding it into this +/// node's `aggregate_tick` alongside those same kinds of direct sources +/// would let one upstream signal count twice (once locally, once via the +/// remote's combine) and skew the median/mean. Restricting it to +/// currencies nobody else can resolve keeps it strictly a gap-filler — +/// its documented purpose (spec §11.7) — while a fully-covered currency's +/// value stays whatever the direct sources agree on. +/// +/// Coverage is decided **after** fiat-cross resolution, not from raw quote +/// keys. El Toque reports CUP/MLC only as `PerBase { base: USD }`; a key +/// presence check would treat that as covering CUP even when the USD +/// anchor is missing, drop Nostr's usable `PerBtc` CUP, then lose CUP +/// entirely when the cross fails to resolve (ermeme, PR #841). Anchors +/// are built from *all* direct `PerBtc` quotes (including Nostr) so a +/// local El Toque cross that needs Nostr's USD still counts as covering +/// CUP — Nostr's direct CUP is then correctly suppressed as redundant. +/// +/// Currency codes are folded to uppercase because `aggregate_tick` does +/// the same before grouping; Yadio forwards API codes verbatim. +fn restrict_nostr_to_fallback( + results: Vec<(ProviderId, ProviderQuotes)>, + outlier_pct: f64, +) -> Vec<(ProviderId, ProviderQuotes)> { + let covered_elsewhere = currencies_covered_by_non_nostr(&results, outlier_pct); + results + .into_iter() + .map(|(id, quotes)| { + if id != ProviderId::Nostr { + return (id, quotes); + } + let fallback_only: ProviderQuotes = quotes + .into_iter() + .filter(|(currency, _)| !covered_elsewhere.contains(¤cy.to_uppercase())) + .collect(); + (id, fallback_only) + }) + .collect() +} + +/// Currencies a non-Nostr provider can actually yield this tick: either a +/// usable direct `PerBtc`, or a `PerBase` that resolves against the +/// tick's anchors (anchors include Nostr `PerBtc` so local crosses can +/// still cover a currency when only the relayed USD is available). +fn currencies_covered_by_non_nostr( + results: &[(ProviderId, ProviderQuotes)], + outlier_pct: f64, +) -> HashSet { + let mut direct: HashMap> = HashMap::new(); + for (_id, quotes) in results { + for (currency, quote) in quotes { + if let Quote::PerBtc(v) = quote { + if v.is_finite() && *v > 0.0 { + direct.entry(currency.to_uppercase()).or_default().push(*v); + } + } + } + } + let mut anchors: HashMap = HashMap::new(); + for (currency, values) in &direct { + if let Some(v) = combine(values, outlier_pct) { + anchors.insert(currency.clone(), v); + } + } + + let mut covered: HashSet = HashSet::new(); + for (id, quotes) in results { + if *id == ProviderId::Nostr { + continue; + } + for (currency, quote) in quotes { + let code = currency.to_uppercase(); + match quote { + Quote::PerBtc(v) if v.is_finite() && *v > 0.0 => { + covered.insert(code); + } + Quote::PerBase { base, value } => { + if let Some(anchor) = anchors.get(&base.to_uppercase()) { + let candidate = value * anchor; + if candidate.is_finite() && candidate > 0.0 { + covered.insert(code); + } + } + } + _ => {} + } + } + } + covered +} + +/// Currencies fit to republish to Nostr this tick: everything **except** +/// those that still depend on a Nostr-sourced rate. +/// +/// Two cases must not be re-stamped with a fresh `created_at`/`expiration` +/// (re-review, PR #841): +/// - sole surviving contributor is `Nostr` itself; +/// - a fiat-cross resolved against a Nostr-touched USD (or other) anchor +/// (`nostr_anchor_dependent`) — e.g. El Toque CUP × Nostr USD would +/// otherwise publish as `source=eltoque` while embedding a relayed rate. +/// +/// A currency this node independently corroborated with a non-Nostr +/// absolute path (and no Nostr-anchor cross) republishes as before. +fn republishable_rates(aggregates: &HashMap) -> HashMap { + aggregates + .iter() + .filter(|(_, a)| a.contributors != [ProviderId::Nostr] && !a.nostr_anchor_dependent) + .map(|(c, a)| (c.clone(), a.value)) + .collect() +} + +/// Providers that actually contributed to the republished payload. +/// +/// The tick-wide contributor list covers every aggregate, but +/// `republishable_rates` can drop currencies — so a provider whose every +/// contribution was filtered out would still be named in the `source` tag +/// of a body carrying none of its data. Deriving the tag from the +/// surviving currencies keeps the tag honest about what actually ships. +fn republished_contributors( + aggregates: &HashMap, + rates: &HashMap, +) -> Vec { + aggregates + .iter() + .filter(|(currency, _)| rates.contains_key(*currency)) + .flat_map(|(_, a)| a.contributors.iter().copied()) + .collect::>() + .into_iter() + .collect() +} + /// Joined list of contributing provider ids for the Nostr `source` tag. /// Sorted so the tag is deterministic across ticks with the same provider /// set, regardless of map-iteration order. @@ -569,7 +723,12 @@ fn sources_to_tag(ids: &[ProviderId]) -> String { /// Single designated extension point (spec §5.4 Step 3). Adding a new /// provider adds exactly one match arm here — the aggregation core, the /// store, the scheduler, and every order handler stay untouched. -fn build_provider(id: ProviderId, cfg: &ProviderConfig) -> Result, String> { +fn build_provider( + id: ProviderId, + cfg: &ProviderConfig, + provider_timeout_seconds: u64, + max_price_staleness_seconds: i64, +) -> Result, String> { match id { ProviderId::Yadio => Ok(Box::new(YadioProvider::new(cfg))), ProviderId::CoinGecko => Ok(Box::new(CoinGeckoProvider::new(cfg))), @@ -579,6 +738,16 @@ fn build_provider(id: ProviderId, cfg: &ProviderConfig) -> Result Ok(Box::new(ElToqueProvider::new(cfg)?)), + // Nostr trusted-node relay mode (§11.7). `new` returns `Err` when + // `trusted_nodes` is empty or contains an unparsable hex pubkey. + // `provider_timeout_seconds` sizes its one-shot relay query; + // `max_price_staleness_seconds` is the freshness gate on event + // `created_at` so zombie relay data cannot refresh the store clock. + ProviderId::Nostr => Ok(Box::new(NostrProvider::new( + cfg, + provider_timeout_seconds, + max_price_staleness_seconds, + )?)), } } @@ -640,6 +809,7 @@ pub fn synthesise_legacy_price_settings( token: None, only: None, except: None, + trusted_nodes: vec![], }, ); PriceSettings { @@ -712,6 +882,7 @@ mod tests { token: None, only: None, except: None, + trusted_nodes: vec![], }, ); providers.push(EnabledProvider { @@ -854,6 +1025,7 @@ mod tests { token: token.map(String::from), only: Some(vec!["CUP".into(), "MLC".into()]), except: None, + trusted_nodes: vec![], } } @@ -881,6 +1053,52 @@ mod tests { assert!(PriceManager::from_settings(settings).is_err()); } + #[test] + fn from_settings_builds_nostr_with_trusted_nodes() { + // §11.7: Nostr trusted-node relay mode builds into the registry like + // any other provider once it has at least one valid pubkey. + let mut settings = PriceSettings::default(); + settings.providers.insert( + ProviderId::Nostr.to_string(), + ProviderConfig { + enabled: true, + url: String::new(), + fallback_urls: vec![], + api_key: None, + token: None, + only: None, + except: None, + trusted_nodes: vec![ + "82fa8cb978b43c79b2156585bac2c011176a21d2aead6d9f7c575c005be88390".into(), + ], + }, + ); + let m = PriceManager::from_settings(settings).expect("nostr builds with trusted_nodes"); + assert_eq!(m.providers.len(), 1); + assert_eq!(m.providers[0].id, ProviderId::Nostr); + } + + #[test] + fn from_settings_rejects_nostr_with_invalid_pubkey() { + // §11.7 fail-fast: an enabled Nostr provider with an unparsable + // trusted-node pubkey must not silently produce no quotes. + let mut settings = PriceSettings::default(); + settings.providers.insert( + ProviderId::Nostr.to_string(), + ProviderConfig { + enabled: true, + url: String::new(), + fallback_urls: vec![], + api_key: None, + token: None, + only: None, + except: None, + trusted_nodes: vec!["not-a-pubkey".into()], + }, + ); + assert!(PriceManager::from_settings(settings).is_err()); + } + #[test] fn from_settings_builds_all_phase2_providers() { // Spec §9 Phase 2: the three keyless backups join Yadio in the @@ -902,6 +1120,7 @@ mod tests { token: None, only: None, except: None, + trusted_nodes: vec![], }, ); } @@ -925,6 +1144,7 @@ mod tests { token: None, only: None, except: None, + trusted_nodes: vec![], }, ); let m = PriceManager::from_settings(settings).expect("unknown id is non-fatal"); @@ -944,6 +1164,7 @@ mod tests { token: None, only: None, except: None, + trusted_nodes: vec![], }, ); let m = PriceManager::from_settings(settings).unwrap(); @@ -956,6 +1177,286 @@ mod tests { assert_eq!(tag, "coingecko,yadio"); } + #[test] + fn restrict_nostr_to_fallback_drops_currencies_another_provider_covers() { + let mut yadio = ProviderQuotes::new(); + yadio.insert("USD".into(), Quote::PerBtc(50_000.0)); + let mut nostr = ProviderQuotes::new(); + nostr.insert("USD".into(), Quote::PerBtc(50_500.0)); // same currency, different value + nostr.insert("ARS".into(), Quote::PerBtc(105_000_000.0)); // nobody else has this + + let out = restrict_nostr_to_fallback( + vec![(ProviderId::Yadio, yadio), (ProviderId::Nostr, nostr)], + 5.0, + ); + + let nostr_out = out + .iter() + .find(|(id, _)| *id == ProviderId::Nostr) + .map(|(_, q)| q) + .unwrap(); + assert!( + !nostr_out.contains_key("USD"), + "USD is covered by Yadio, Nostr's USD must be dropped to avoid double-counting" + ); + assert_eq!( + nostr_out.get("ARS"), + Some(&Quote::PerBtc(105_000_000.0)), + "ARS has no other source, Nostr must still fill the gap" + ); + + let yadio_out = out + .iter() + .find(|(id, _)| *id == ProviderId::Yadio) + .map(|(_, q)| q) + .unwrap(); + assert_eq!( + yadio_out.get("USD"), + Some(&Quote::PerBtc(50_000.0)), + "non-Nostr providers are untouched" + ); + } + + #[test] + fn restrict_nostr_to_fallback_matches_coverage_case_insensitively() { + // Yadio forwards the API's codes verbatim, so a lowercase code is + // reachable in production. `aggregate_tick` upper-cases before + // grouping, so `usd` and `USD` are the same currency by the time it + // matters — coverage has to be matched the same way. + let mut yadio = ProviderQuotes::new(); + yadio.insert("usd".into(), Quote::PerBtc(50_000.0)); + let mut nostr = ProviderQuotes::new(); + nostr.insert("USD".into(), Quote::PerBtc(50_500.0)); + nostr.insert("ARS".into(), Quote::PerBtc(105_000_000.0)); + + let out = restrict_nostr_to_fallback( + vec![(ProviderId::Yadio, yadio), (ProviderId::Nostr, nostr)], + 5.0, + ); + + let nostr_out = out + .iter() + .find(|(id, _)| *id == ProviderId::Nostr) + .map(|(_, q)| q) + .unwrap(); + assert!( + !nostr_out.contains_key("USD"), + "Yadio's lowercase `usd` covers Nostr's `USD` — both fold to the \ + same aggregate, so Nostr's quote must be dropped" + ); + assert_eq!( + nostr_out.get("ARS"), + Some(&Quote::PerBtc(105_000_000.0)), + "ARS is still uncovered, Nostr must fill the gap" + ); + } + + #[test] + fn restrict_nostr_to_fallback_is_a_noop_without_a_nostr_provider() { + let mut yadio = ProviderQuotes::new(); + yadio.insert("USD".into(), Quote::PerBtc(50_000.0)); + + let out = restrict_nostr_to_fallback(vec![(ProviderId::Yadio, yadio.clone())], 5.0); + assert_eq!(out, vec![(ProviderId::Yadio, yadio)]); + } + + #[test] + fn restrict_nostr_to_fallback_keeps_nostr_when_per_base_cannot_resolve() { + // El Toque "has" CUP as a key, but without a USD anchor the cross + // cannot resolve — treating the raw key as coverage would drop + // Nostr's usable PerBtc CUP and leave the tick with no CUP at all + // (ermeme, PR #841). + let mut eltoque = ProviderQuotes::new(); + eltoque.insert( + "CUP".into(), + Quote::PerBase { + base: "USD".into(), + value: 300.0, + }, + ); + let mut nostr = ProviderQuotes::new(); + nostr.insert("CUP".into(), Quote::PerBtc(30_000_000.0)); + + let out = restrict_nostr_to_fallback( + vec![(ProviderId::ElToque, eltoque), (ProviderId::Nostr, nostr)], + 5.0, + ); + let nostr_out = out + .iter() + .find(|(id, _)| *id == ProviderId::Nostr) + .map(|(_, q)| q) + .unwrap(); + assert_eq!( + nostr_out.get("CUP"), + Some(&Quote::PerBtc(30_000_000.0)), + "unresolvable El Toque PerBase must not suppress Nostr's PerBtc CUP" + ); + } + + #[test] + fn restrict_nostr_to_fallback_drops_nostr_when_per_base_resolves_via_nostr_usd() { + // Nostr supplies USD; El Toque's CUP/USD then resolves. CUP is + // covered by the local cross, so Nostr's direct CUP must not also + // vote (would double-count correlated remote aggregate data). + let mut eltoque = ProviderQuotes::new(); + eltoque.insert( + "CUP".into(), + Quote::PerBase { + base: "USD".into(), + value: 400.0, + }, + ); + let mut nostr = ProviderQuotes::new(); + nostr.insert("USD".into(), Quote::PerBtc(50_000.0)); + nostr.insert("CUP".into(), Quote::PerBtc(30_000_000.0)); + + let out = restrict_nostr_to_fallback( + vec![(ProviderId::ElToque, eltoque), (ProviderId::Nostr, nostr)], + 5.0, + ); + let nostr_out = out + .iter() + .find(|(id, _)| *id == ProviderId::Nostr) + .map(|(_, q)| q) + .unwrap(); + assert!( + !nostr_out.contains_key("CUP"), + "El Toque CUP that resolves against Nostr USD covers the currency" + ); + assert_eq!( + nostr_out.get("USD"), + Some(&Quote::PerBtc(50_000.0)), + "Nostr USD is still needed as the anchor" + ); + } + + #[test] + fn republishable_rates_drops_nostr_only_sourced_currencies() { + let mut aggregates = HashMap::new(); + aggregates.insert( + "USD".to_string(), + AggregateResult { + value: 50_000.0, + sources: 2, + contributors: vec![ProviderId::Yadio, ProviderId::CoinGecko], + nostr_anchor_dependent: false, + }, + ); + aggregates.insert( + "ARS".to_string(), + AggregateResult { + value: 105_000_000.0, + sources: 1, + contributors: vec![ProviderId::Nostr], + nostr_anchor_dependent: false, + }, + ); + + let out = republishable_rates(&aggregates); + assert_eq!(out.get("USD"), Some(&50_000.0)); + assert!( + !out.contains_key("ARS"), + "Nostr-only-sourced ARS must not be republished with a fresh timestamp" + ); + } + + #[test] + fn republished_contributors_omits_a_provider_whose_currencies_were_all_dropped() { + // Nostr's only contribution is the ARS aggregate, which + // `republishable_rates` drops as Nostr-only — so the published body + // carries nothing from Nostr and the `source` tag must not claim it. + let mut aggregates = HashMap::new(); + aggregates.insert( + "USD".to_string(), + AggregateResult { + value: 50_000.0, + sources: 2, + contributors: vec![ProviderId::Yadio, ProviderId::CoinGecko], + nostr_anchor_dependent: false, + }, + ); + aggregates.insert( + "ARS".to_string(), + AggregateResult { + value: 105_000_000.0, + sources: 1, + contributors: vec![ProviderId::Nostr], + nostr_anchor_dependent: false, + }, + ); + + let rates = republishable_rates(&aggregates); + let contributors = republished_contributors(&aggregates, &rates); + + assert_eq!( + contributors, + vec![ProviderId::Yadio, ProviderId::CoinGecko] + .into_iter() + .collect::>() + .into_iter() + .collect::>(), + "only the providers behind the surviving USD rate belong in the tag" + ); + assert!( + !sources_to_tag(&contributors).contains("nostr"), + "the `source` tag must not name a provider absent from the body" + ); + } + + #[test] + fn republishable_rates_keeps_a_currency_nostr_only_partly_helped_with() { + // Nostr alongside another contributor (e.g. it filled in after a + // fallback scenario in a prior tick, or n>=2 combine) is not + // "Nostr-only" — still safe to republish since it wasn't the sole + // source. + let mut aggregates = HashMap::new(); + aggregates.insert( + "EUR".to_string(), + AggregateResult { + value: 45_000.0, + sources: 2, + contributors: vec![ProviderId::Nostr, ProviderId::Yadio], + nostr_anchor_dependent: false, + }, + ); + + let out = republishable_rates(&aggregates); + assert_eq!(out.get("EUR"), Some(&45_000.0)); + } + + #[test] + fn republishable_rates_drops_nostr_anchor_dependent_cross() { + // El Toque CUP resolved via Nostr USD: contributors name only + // eltoque, but the absolute value embeds a relayed rate — must not + // be re-stamped as a fresh local observation (ermeme, PR #841). + let mut aggregates = HashMap::new(); + aggregates.insert( + "CUP".to_string(), + AggregateResult { + value: 20_000_000.0, + sources: 1, + contributors: vec![ProviderId::ElToque], + nostr_anchor_dependent: true, + }, + ); + aggregates.insert( + "EUR".to_string(), + AggregateResult { + value: 45_000.0, + sources: 1, + contributors: vec![ProviderId::Yadio], + nostr_anchor_dependent: false, + }, + ); + + let out = republishable_rates(&aggregates); + assert!( + !out.contains_key("CUP"), + "Nostr-anchor-dependent CUP must not be republished" + ); + assert_eq!(out.get("EUR"), Some(&45_000.0)); + } + #[tokio::test] async fn scoped_out_provider_is_success_but_not_contributor() { // A successful poll whose every currency is filtered by `only` @@ -1258,6 +1759,7 @@ mod tests { value: 50_000.0, sources: 1, contributors: vec![ProviderId::Yadio], + nostr_anchor_dependent: false, }, ); // 1_000_000s ago: well past any plausible TTL. @@ -1285,6 +1787,7 @@ mod tests { value: 50_000.0, sources: 1, contributors: vec![ProviderId::Yadio], + nostr_anchor_dependent: false, }, ); let fresh_now = Utc::now().timestamp(); @@ -1320,6 +1823,7 @@ mod tests { value: 50_000.0, sources: 1, contributors: vec![ProviderId::Yadio], + nostr_anchor_dependent: false, }, ); let now = Utc::now().timestamp(); @@ -1367,6 +1871,7 @@ mod tests { value: v, sources: 1, contributors: vec![ProviderId::Yadio], + nostr_anchor_dependent: false, }, ); agg @@ -1493,6 +1998,7 @@ mod coverage_tests { token: None, only: None, except: None, + trusted_nodes: vec![], }, ); let manager = bare_manager( @@ -1571,11 +2077,10 @@ mod coverage_tests { value: 50_000.0, sources: 1, contributors: vec![ProviderId::Yadio], + nostr_anchor_dependent: false, }, ); - manager - .publish_rates_to_nostr(&aggregates, &[ProviderId::Yadio]) - .await; + manager.publish_rates_to_nostr(&aggregates).await; } #[tokio::test] @@ -1596,6 +2101,7 @@ mod coverage_tests { token: None, only: None, except: None, + trusted_nodes: vec![], }, ); diff --git a/src/price/provider.rs b/src/price/provider.rs index 80de63b8..902a3015 100644 --- a/src/price/provider.rs +++ b/src/price/provider.rs @@ -39,6 +39,7 @@ pub enum ProviderId { CurrencyApi, Blockchain, ElToque, + Nostr, } impl fmt::Display for ProviderId { @@ -49,6 +50,7 @@ impl fmt::Display for ProviderId { ProviderId::CurrencyApi => "currency_api", ProviderId::Blockchain => "blockchain", ProviderId::ElToque => "eltoque", + ProviderId::Nostr => "nostr", }; f.write_str(s) } @@ -64,6 +66,7 @@ impl FromStr for ProviderId { "currency_api" => Ok(ProviderId::CurrencyApi), "blockchain" => Ok(ProviderId::Blockchain), "eltoque" => Ok(ProviderId::ElToque), + "nostr" => Ok(ProviderId::Nostr), other => Err(format!("unknown price provider id: {other}")), } } @@ -205,6 +208,7 @@ mod tests { ProviderId::CurrencyApi, ProviderId::Blockchain, ProviderId::ElToque, + ProviderId::Nostr, ] { let s = id.to_string(); assert_eq!(ProviderId::from_str(&s).unwrap(), id, "roundtrip {s}"); diff --git a/src/price/providers/blockchain.rs b/src/price/providers/blockchain.rs index db41f02c..8d06124c 100644 --- a/src/price/providers/blockchain.rs +++ b/src/price/providers/blockchain.rs @@ -146,6 +146,7 @@ mod tests { token: None, only: None, except: None, + trusted_nodes: vec![], }; assert_eq!(BlockchainProvider::new(&cfg).url, "https://blockchain.info"); } diff --git a/src/price/providers/coingecko.rs b/src/price/providers/coingecko.rs index 66dec104..2e52aed4 100644 --- a/src/price/providers/coingecko.rs +++ b/src/price/providers/coingecko.rs @@ -144,6 +144,7 @@ mod tests { token: None, only: None, except: None, + trusted_nodes: vec![], } } diff --git a/src/price/providers/currency_api.rs b/src/price/providers/currency_api.rs index 6e3264f3..217006fb 100644 --- a/src/price/providers/currency_api.rs +++ b/src/price/providers/currency_api.rs @@ -139,6 +139,7 @@ mod tests { token: None, only: None, except: None, + trusted_nodes: vec![], } } diff --git a/src/price/providers/eltoque.rs b/src/price/providers/eltoque.rs index a1e87e80..f522d60b 100644 --- a/src/price/providers/eltoque.rs +++ b/src/price/providers/eltoque.rs @@ -223,6 +223,7 @@ mod tests { token: token.map(String::from), only: Some(vec!["CUP".into(), "MLC".into()]), except: None, + trusted_nodes: vec![], } } diff --git a/src/price/providers/mod.rs b/src/price/providers/mod.rs index d1a796f7..e6bfd79a 100644 --- a/src/price/providers/mod.rs +++ b/src/price/providers/mod.rs @@ -10,4 +10,5 @@ pub mod blockchain; pub mod coingecko; pub mod currency_api; pub mod eltoque; +pub mod nostr; pub mod yadio; diff --git a/src/price/providers/nostr.rs b/src/price/providers/nostr.rs new file mode 100644 index 00000000..17923f4a --- /dev/null +++ b/src/price/providers/nostr.rs @@ -0,0 +1,733 @@ +//! Nostr trusted-node relay-mode quoter (spec §11.7). +//! +//! Instead of an HTTP API, this provider sources BTC/fiat quotes from the +//! same kind-30078 NIP-33 rate event Mostro nodes already publish +//! (`nip33::new_exchange_rates_event`, `docs/NOSTR_EXCHANGE_RATES.md`) — for +//! operators in regions where price APIs are DNS/IP blocked but Nostr relays +//! are reachable (issue #697). It queries the process-wide Nostr client +//! (already connected to the `[nostr]` relays) for the latest such event +//! from each configured `trusted_nodes` pubkey and takes the **freshest** +//! one as this tick's source — no cross-node statistical combine; a stale or +//! unreachable trusted node is simply outrun by a fresher one. Events whose +//! `created_at` is older than `[price].max_price_staleness_seconds` are +//! discarded so a relay that still serves an expired kind-30078 cannot keep +//! refreshing the local cache clock with zombie rates. + +use std::collections::HashMap; +use std::time::Duration; + +use async_trait::async_trait; +use futures::StreamExt; +use nostr_sdk::prelude::*; +use serde::Deserialize; +use tracing::debug; + +use crate::config::constants::NOSTR_EXCHANGE_RATES_EVENT_KIND; +use crate::price::config::ProviderConfig; +use crate::price::provider::{PriceProvider, ProviderError, ProviderId, ProviderQuotes, Quote}; + +/// Hard cap on events retained from one relay query before local ranking. +/// +/// `nostr-relay-pool`'s `fetch_events` uses `Events::force_insert`, which +/// grows without bound when a noncompliant relay ignores the subscription +/// filter (`verify_subscriptions` defaults to false). Streaming with an +/// early envelope/author check and this ceiling bounds memory/work within +/// `query_timeout` (ermeme, PR #841). Well above any realistic number of +/// trusted-node candidates for one tick. +const MAX_FETCHED_RATE_EVENTS: usize = 64; + +/// Same wrapper the daemon publishes: `{"BTC": {ccy: price}}`. +#[derive(Debug, Deserialize)] +struct RatesContent { + #[serde(rename = "BTC")] + btc: HashMap, +} + +/// Trusted-node Nostr quoter. +pub struct NostrProvider { + trusted_nodes: Vec, + /// One-shot relay query bound, derived from the shared + /// `provider_timeout_seconds` (see `new`) rather than a fixed constant — + /// an operator who lowers that setting must not have this provider's + /// queries consistently swallowed by `PriceManager::poll_budget`'s outer + /// timeout before they can complete or fail on their own (CodeRabbit, + /// PR #841). + query_timeout: Duration, + /// Maximum age of a trusted-node rate event (`created_at`), taken from + /// the shared `[price].max_price_staleness_seconds` so upstream Nostr + /// freshness uses the same TTL the store enforces on cached quotes. + max_age: Duration, +} + +impl NostrProvider { + /// Build the provider from its `[price.providers.nostr]` sub-table. + /// + /// Fails fast (mirrors `ElToqueProvider::new`'s missing-token check) when + /// `trusted_nodes` is empty or contains a pubkey that isn't valid hex — + /// `ProviderConfig::validate` only checks the list is non-empty, not + /// that its entries parse (spec §7). + /// + /// `provider_timeout_seconds` is the shared `[price]` setting (not + /// per-provider config): `query_timeout` is set to exactly that value, + /// so it fits under `poll_budget`'s `provider_timeout_seconds + 1s` + /// (no `fallback_urls` for this provider ⇒ one attempt) with the same + /// 1s of slack every other provider gets from the shared `reqwest` + /// client's own per-attempt timeout. + /// + /// `max_price_staleness_seconds` is the same shared TTL used by + /// [`crate::price::store::PriceStore`]: events older than that are not + /// eligible as this tick's source. + pub fn new( + cfg: &ProviderConfig, + provider_timeout_seconds: u64, + max_price_staleness_seconds: i64, + ) -> Result { + if cfg.trusted_nodes.is_empty() { + return Err( + "price provider 'nostr': enabled provider requires at least one \ + `trusted_nodes` pubkey (see docs/PRICE_PROVIDERS.md §11.7)" + .to_string(), + ); + } + let trusted_nodes = cfg + .trusted_nodes + .iter() + .map(|hex| { + PublicKey::from_hex(hex).map_err(|e| { + format!( + "price provider 'nostr': invalid `trusted_nodes` pubkey \ + '{hex}': {e}" + ) + }) + }) + .collect::, _>>()?; + Ok(Self { + trusted_nodes, + query_timeout: Duration::from_secs(provider_timeout_seconds.max(1)), + // `PriceSettings::validate` already rejects non-positive values; + // clamp here so a zero can never make every event look fresh. + max_age: Duration::from_secs(max_price_staleness_seconds.max(1) as u64), + }) + } + + /// The relay query for this tick: the latest `mostro-rates` (kind + /// 30078) event from any configured trusted node. Split out from + /// [`PriceProvider::fetch`] so its shape is unit-testable without a + /// relay (spec §10.5) — a wrong `kind`/`identifier` here would + /// otherwise only surface via the `#[ignore]`d live-relay test. + pub(crate) fn build_filter(&self) -> Filter { + Filter::new() + .kind(Kind::Custom(NOSTR_EXCHANGE_RATES_EVENT_KIND)) + .authors(self.trusted_nodes.clone()) + .identifier("mostro-rates") + } + + /// Parse a rate event's `content` into [`ProviderQuotes`]. Split out so + /// it is unit-testable without a relay (spec §10.5). + pub(crate) fn parse_content(body: &str) -> Result { + let parsed: RatesContent = + serde_json::from_str(body).map_err(|e| ProviderError::Parse(format!("nostr: {e}")))?; + Ok(parsed + .btc + .into_iter() + .filter_map(|(code, v)| match v { + v if v.is_finite() && v > 0.0 => Some((code.to_uppercase(), Quote::PerBtc(v))), + _ => None, + }) + .collect()) + } + + /// Every event from `events` that passes every event-level trust and + /// freshness gate, newest first: + /// + /// - authored by a trusted pubkey — a second client-side check even + /// though the relay-side `authors` filter should already guarantee it + /// (spec §10.3 / `NOSTR_EXCHANGE_RATES.md` "clients MUST verify + /// pubkey"); + /// - the exact expected envelope (`kind` 30078, `d` = `mostro-rates`) — + /// `build_filter`'s relay-side query asks for this, but + /// `nostr-relay-pool`'s `verify_subscriptions` defaults to `false` (not + /// enabled by `connect_nostr`), so a noncompliant relay could still + /// hand back a validly-signed event from a trusted pubkey of a + /// different kind/`d` (CodeRabbit / re-review, PR #841); + /// - not future-dated (a forged or clock-skewed `created_at` must not + /// look artificially fresh); + /// - `created_at` within `max_age` of `now`, so a relay that ignores + /// NIP-40 expiration cannot keep serving a zombie rate; + /// - not expired per its own NIP-40 `expiration` tag, if any + /// (`Event::is_expired_at`) — a trusted node's shorter self-declared + /// expiry is honored even within `max_age`. + /// + /// Ranking (rather than returning a single winner) lets [`Self::fetch`] + /// fall through to the next-freshest candidate when the newest one turns + /// out to be unparsable or empty — see `parse_content` — instead of + /// letting one bad event from a trusted node shadow a perfectly good + /// older one from the same or another trusted node. Pure and + /// unit-testable without a relay. + pub(crate) fn rank_candidates<'a>( + events: &'a [Event], + trusted: &[PublicKey], + now: Timestamp, + max_age: Duration, + ) -> Vec<&'a Event> { + let max_age_secs = max_age.as_secs(); + let mut candidates: Vec<&Event> = events + .iter() + .filter(|e| trusted.contains(&e.pubkey)) + .filter(|e| e.kind == Kind::Custom(NOSTR_EXCHANGE_RATES_EVENT_KIND)) + .filter(|e| e.tags.identifier() == Some("mostro-rates")) + // A future-dated `created_at` (forged or clock-skewed relay) would + // otherwise saturate the age check to 0 and win the ranking outright. + .filter(|e| e.created_at <= now) + .filter(|e| now.as_secs().saturating_sub(e.created_at.as_secs()) <= max_age_secs) + .filter(|e| !e.is_expired_at(&now)) + .collect(); + candidates.sort_unstable_by(|a, b| b.created_at.cmp(&a.created_at)); + candidates + } + + /// Try `candidates` newest-first, returning the first one whose content + /// parses into a non-empty rate map. A malformed body or a `{"BTC":{}}` + /// empty one from the freshest candidate must not shadow a good, still- + /// fresh body from an older candidate — that would fail the whole tick + /// even though a usable redundant trusted node is right there. Split out + /// from [`Self::fetch`] so the fallback behavior is unit-testable + /// without a relay. + pub(crate) fn pick_first_usable( + candidates: &[&Event], + ) -> Result { + for event in candidates { + match Self::parse_content(&event.content) { + Ok(quotes) if !quotes.is_empty() => return Ok(quotes), + Ok(_) => debug!( + "price: nostr: candidate event {} parsed to zero usable rates, trying next", + event.id + ), + Err(e) => debug!( + "price: nostr: candidate event {} failed to parse ({e}), trying next", + event.id + ), + } + } + Err(ProviderError::Parse( + "nostr: every fresh trusted-node candidate failed to parse or yielded no usable rates" + .to_string(), + )) + } + + /// Whether a relay-delivered event is worth buffering for ranking. + /// Applied while streaming so irrelevant junk never accumulates toward + /// [`MAX_FETCHED_RATE_EVENTS`]. + pub(crate) fn is_plausible_candidate(event: &Event, trusted: &[PublicKey]) -> bool { + trusted.contains(&event.pubkey) + && event.kind == Kind::Custom(NOSTR_EXCHANGE_RATES_EVENT_KIND) + && event.tags.identifier() == Some("mostro-rates") + } + + /// Keep at most `limit` plausible candidates from `incoming`, dropping + /// everything else before it is stored. Pure helper so the adversarial + /// bound is unit-testable without a relay; production `fetch` applies + /// the same filter/cap while streaming. + #[cfg(test)] + pub(crate) fn collect_bounded_candidates( + incoming: impl IntoIterator, + trusted: &[PublicKey], + limit: usize, + ) -> Vec { + let mut out = Vec::with_capacity(limit.min(16)); + for event in incoming { + if !Self::is_plausible_candidate(&event, trusted) { + continue; + } + out.push(event); + if out.len() >= limit { + break; + } + } + out + } +} + +#[async_trait] +impl PriceProvider for NostrProvider { + fn id(&self) -> ProviderId { + ProviderId::Nostr + } + + async fn fetch(&self, _http: &reqwest::Client) -> Result { + let client = crate::util::get_nostr_client() + .map_err(|e| ProviderError::Http(format!("nostr: {e}")))?; + + // Stream rather than `fetch_events`: the pool's collector + // `force_insert`s every delivered event (ignoring Filter::limit), + // so a noncompliant relay can OOM the process within query_timeout. + // Early-filter + hard cap while receiving (ermeme, PR #841). + let filter = self.build_filter().limit(MAX_FETCHED_RATE_EVENTS); + let mut stream = client + .stream_events(filter, self.query_timeout) + .await + .map_err(|e| ProviderError::Http(format!("nostr: relay query failed: {e}")))?; + + let mut events = Vec::new(); + let mut inspected = 0usize; + while let Some(event) = stream.next().await { + inspected += 1; + if Self::is_plausible_candidate(&event, &self.trusted_nodes) { + events.push(event); + if events.len() >= MAX_FETCHED_RATE_EVENTS { + break; + } + } + // Bound work even when the relay floods irrelevant frames. + if inspected >= MAX_FETCHED_RATE_EVENTS.saturating_mul(4) { + break; + } + } + + let candidates = + Self::rank_candidates(&events, &self.trusted_nodes, Timestamp::now(), self.max_age); + if candidates.is_empty() { + return Err(ProviderError::Http( + "nostr: no fresh trusted-node rate event found \ + (all missing, untrusted, or older than max_price_staleness_seconds)" + .to_string(), + )); + } + + Self::pick_first_usable(&candidates) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const SAMPLE_CONTENT: &str = r#"{"BTC": {"USD": 50000.0, "eur": 45000.0, "ARS": 105000000.0}}"#; + + #[test] + fn parse_content_upper_cases_codes() { + let quotes = NostrProvider::parse_content(SAMPLE_CONTENT).unwrap(); + assert_eq!(quotes.get("USD"), Some(&Quote::PerBtc(50_000.0))); + assert_eq!(quotes.get("EUR"), Some(&Quote::PerBtc(45_000.0))); + assert_eq!(quotes.get("ARS"), Some(&Quote::PerBtc(105_000_000.0))); + } + + #[test] + fn parse_content_drops_non_finite_and_non_positive() { + let body = r#"{"BTC": {"USD": 0, "EUR": -1, "GBP": 50000.0}}"#; + let quotes = NostrProvider::parse_content(body).unwrap(); + assert_eq!(quotes.len(), 1, "only GBP is a usable rate"); + assert_eq!(quotes.get("GBP"), Some(&Quote::PerBtc(50_000.0))); + } + + #[test] + fn parse_content_error_is_returned() { + let err = NostrProvider::parse_content("not json").unwrap_err(); + assert!(matches!(err, ProviderError::Parse(_))); + } + + fn signed_event(keys: &Keys, content: &str, created_at: u64) -> Event { + EventBuilder::new(Kind::Custom(NOSTR_EXCHANGE_RATES_EVENT_KIND), content) + .tags(vec![Tag::identifier("mostro-rates")]) + .custom_created_at(Timestamp::from(created_at)) + .sign_with_keys(keys) + .expect("event must sign") + } + + /// Like `signed_event`, but carrying a NIP-40 `expiration` tag — for + /// exercising `rank_candidates`'s `is_expired_at` gate independently of + /// the `max_age` gate. + fn signed_event_expiring_at( + keys: &Keys, + content: &str, + created_at: u64, + expiration: u64, + ) -> Event { + EventBuilder::new(Kind::Custom(NOSTR_EXCHANGE_RATES_EVENT_KIND), content) + .tags(vec![ + Tag::identifier("mostro-rates"), + Tag::expiration(Timestamp::from(expiration)), + ]) + .custom_created_at(Timestamp::from(created_at)) + .sign_with_keys(keys) + .expect("event must sign") + } + + /// Relative clock for `rank_candidates` unit tests — events use small + /// synthetic `created_at` values, so tests pass an explicit `now` + /// instead of wall-clock time. + const NOW: u64 = 10_000; + const MAX_AGE: Duration = Duration::from_secs(5_000); + + /// The single freshest candidate, if any — these tests only care about + /// the winner, not the full fallback order `rank_candidates` exposes for + /// `pick_first_usable`. + fn freshest<'a>( + events: &'a [Event], + trusted: &[PublicKey], + now: Timestamp, + max_age: Duration, + ) -> Option<&'a Event> { + NostrProvider::rank_candidates(events, trusted, now, max_age) + .into_iter() + .next() + } + + #[test] + fn rank_candidates_returns_none_for_empty_events() { + let trusted = vec![Keys::generate().public_key()]; + assert!(freshest(&[], &trusted, Timestamp::from(NOW), MAX_AGE).is_none()); + } + + #[test] + fn rank_candidates_ignores_untrusted_authors() { + let trusted_keys = Keys::generate(); + let untrusted_keys = Keys::generate(); + let trusted = vec![trusted_keys.public_key()]; + + let untrusted_event = signed_event(&untrusted_keys, SAMPLE_CONTENT, NOW - 100); + let events = vec![untrusted_event]; + + assert!(freshest(&events, &trusted, Timestamp::from(NOW), MAX_AGE).is_none()); + } + + #[test] + fn rank_candidates_picks_the_newest_trusted_event() { + let older_keys = Keys::generate(); + let newer_keys = Keys::generate(); + let trusted = vec![older_keys.public_key(), newer_keys.public_key()]; + + let older = signed_event(&older_keys, SAMPLE_CONTENT, NOW - 2_000); + let newer = signed_event(&newer_keys, SAMPLE_CONTENT, NOW - 100); + let events = vec![older, newer.clone()]; + + let picked = freshest(&events, &trusted, Timestamp::from(NOW), MAX_AGE).unwrap(); + assert_eq!(picked.id, newer.id); + } + + #[test] + fn rank_candidates_discards_events_older_than_max_age() { + let keys = Keys::generate(); + let trusted = vec![keys.public_key()]; + + let stale = signed_event(&keys, SAMPLE_CONTENT, NOW - MAX_AGE.as_secs() - 1); + let fresh = signed_event(&keys, SAMPLE_CONTENT, NOW - 100); + let events = vec![stale, fresh.clone()]; + + let picked = freshest(&events, &trusted, Timestamp::from(NOW), MAX_AGE).unwrap(); + assert_eq!(picked.id, fresh.id); + } + + #[test] + fn rank_candidates_returns_none_when_all_trusted_are_stale() { + let keys = Keys::generate(); + let trusted = vec![keys.public_key()]; + + let stale = signed_event(&keys, SAMPLE_CONTENT, NOW - MAX_AGE.as_secs() - 1); + let events = vec![stale]; + + assert!(freshest(&events, &trusted, Timestamp::from(NOW), MAX_AGE).is_none()); + } + + #[test] + fn rank_candidates_keeps_event_exactly_at_max_age_boundary() { + let keys = Keys::generate(); + let trusted = vec![keys.public_key()]; + + let at_boundary = signed_event(&keys, SAMPLE_CONTENT, NOW - MAX_AGE.as_secs()); + let events = vec![at_boundary.clone()]; + + let picked = freshest(&events, &trusted, Timestamp::from(NOW), MAX_AGE).unwrap(); + assert_eq!(picked.id, at_boundary.id); + } + + #[test] + fn rank_candidates_rejects_future_dated_events() { + let keys = Keys::generate(); + let trusted = vec![keys.public_key()]; + + // A forged or clock-skewed `created_at` in the future must not win + // `max_by_key` over a legitimately current event. + let future = signed_event(&keys, SAMPLE_CONTENT, NOW + 1_000); + let current = signed_event(&keys, SAMPLE_CONTENT, NOW - 100); + let events = vec![future, current.clone()]; + + let picked = freshest(&events, &trusted, Timestamp::from(NOW), MAX_AGE).unwrap(); + assert_eq!(picked.id, current.id); + } + + #[test] + fn rank_candidates_rejects_an_expired_event_even_if_it_is_the_newest() { + let keys = Keys::generate(); + let trusted = vec![keys.public_key()]; + + // Newer by created_at, but self-declared expired via NIP-40 — a + // relay that ignores expiration and still serves it must not let it + // win over an older, still-valid event. + let expired = signed_event_expiring_at(&keys, SAMPLE_CONTENT, NOW - 100, NOW - 50); + let valid = signed_event(&keys, SAMPLE_CONTENT, NOW - 2_000); + let events = vec![expired, valid.clone()]; + + let picked = freshest(&events, &trusted, Timestamp::from(NOW), MAX_AGE).unwrap(); + assert_eq!(picked.id, valid.id); + } + + #[test] + fn rank_candidates_returns_none_when_every_trusted_event_is_expired() { + let keys = Keys::generate(); + let trusted = vec![keys.public_key()]; + + let expired = signed_event_expiring_at(&keys, SAMPLE_CONTENT, NOW - 100, NOW - 50); + let events = vec![expired]; + + assert!(freshest(&events, &trusted, Timestamp::from(NOW), MAX_AGE).is_none()); + } + + /// Like `signed_event`, but with an envelope (`kind`/`d`) that does not + /// match what `mostro-rates` events use — for exercising + /// `rank_candidates`'s client-side envelope revalidation. + fn signed_event_with_envelope( + keys: &Keys, + content: &str, + created_at: u64, + kind: u16, + identifier: &str, + ) -> Event { + EventBuilder::new(Kind::Custom(kind), content) + .tags(vec![Tag::identifier(identifier)]) + .custom_created_at(Timestamp::from(created_at)) + .sign_with_keys(keys) + .expect("event must sign") + } + + #[test] + fn rank_candidates_rejects_a_trusted_event_of_the_wrong_kind() { + let keys = Keys::generate(); + let trusted = vec![keys.public_key()]; + + // Correctly signed by a trusted node, correct `d` tag, but not the + // mostro-rates kind (30078) — e.g. some other event kind that + // pubkey happens to also publish. A relay ignoring the `authors` + // *and* `kind` filter (verify_subscriptions defaults to false) + // could still hand this back. + let wrong_kind = + signed_event_with_envelope(&keys, SAMPLE_CONTENT, NOW - 100, 1, "mostro-rates"); + let events = vec![wrong_kind]; + + assert!( + NostrProvider::rank_candidates(&events, &trusted, Timestamp::from(NOW), MAX_AGE) + .is_empty() + ); + } + + #[test] + fn rank_candidates_rejects_a_trusted_event_with_the_wrong_identifier() { + let keys = Keys::generate(); + let trusted = vec![keys.public_key()]; + + // Right kind, right pubkey, but a `d` tag that isn't "mostro-rates" + // — some other NIP-33 replaceable event this trusted node happens + // to also publish under the same kind. + let wrong_identifier = signed_event_with_envelope( + &keys, + SAMPLE_CONTENT, + NOW - 100, + NOSTR_EXCHANGE_RATES_EVENT_KIND, + "something-else", + ); + let events = vec![wrong_identifier]; + + assert!( + NostrProvider::rank_candidates(&events, &trusted, Timestamp::from(NOW), MAX_AGE) + .is_empty() + ); + } + + #[test] + fn pick_first_usable_skips_a_malformed_newest_candidate_for_an_older_valid_one() { + let keys = Keys::generate(); + let malformed = signed_event(&keys, "not json", NOW - 100); + let valid = signed_event(&keys, SAMPLE_CONTENT, NOW - 2_000); + let candidates = vec![&malformed, &valid]; + + let quotes = NostrProvider::pick_first_usable(&candidates).unwrap(); + assert_eq!(quotes.get("USD"), Some(&Quote::PerBtc(50_000.0))); + } + + #[test] + fn pick_first_usable_skips_an_empty_newest_candidate_for_an_older_valid_one() { + let keys = Keys::generate(); + let empty = signed_event(&keys, r#"{"BTC": {}}"#, NOW - 100); + let valid = signed_event(&keys, SAMPLE_CONTENT, NOW - 2_000); + let candidates = vec![&empty, &valid]; + + let quotes = NostrProvider::pick_first_usable(&candidates).unwrap(); + assert_eq!(quotes.get("USD"), Some(&Quote::PerBtc(50_000.0))); + } + + #[test] + fn pick_first_usable_errs_when_every_candidate_is_unusable() { + let keys = Keys::generate(); + let malformed = signed_event(&keys, "not json", NOW - 100); + let empty = signed_event(&keys, r#"{"BTC": {}}"#, NOW - 200); + let candidates = vec![&malformed, &empty]; + + assert!(NostrProvider::pick_first_usable(&candidates).is_err()); + } + + #[test] + fn collect_bounded_candidates_drops_junk_and_caps_trusted_envelope() { + let trusted_keys = Keys::generate(); + let junk_keys = Keys::generate(); + let trusted = vec![trusted_keys.public_key()]; + + // Flood of wrong-author / wrong-kind frames mixed with more + // trusted mostro-rates events than the cap — only `limit` + // plausible ones may be retained (ermeme adversarial bound). + let mut incoming = Vec::new(); + for i in 0..20 { + incoming.push(signed_event(&junk_keys, SAMPLE_CONTENT, NOW - i)); + incoming.push(signed_event_with_envelope( + &trusted_keys, + SAMPLE_CONTENT, + NOW - i, + 1, + "mostro-rates", + )); + } + for i in 0..10 { + incoming.push(signed_event(&trusted_keys, SAMPLE_CONTENT, NOW - 100 - i)); + } + + let kept = NostrProvider::collect_bounded_candidates(incoming, &trusted, 3); + assert_eq!(kept.len(), 3); + assert!(kept + .iter() + .all(|e| NostrProvider::is_plausible_candidate(e, &trusted))); + } + + fn sample_cfg(trusted_hex: String) -> ProviderConfig { + ProviderConfig { + enabled: true, + url: String::new(), + fallback_urls: vec![], + api_key: None, + token: None, + only: None, + except: None, + trusted_nodes: vec![trusted_hex], + } + } + + #[test] + fn new_parses_valid_hex_trusted_nodes() { + let cfg = sample_cfg(Keys::generate().public_key().to_hex()); + assert!(NostrProvider::new(&cfg, 10, 1_800).is_ok()); + } + + #[test] + fn new_derives_query_timeout_and_max_age_from_shared_settings() { + let cfg = sample_cfg(Keys::generate().public_key().to_hex()); + let provider = NostrProvider::new(&cfg, 7, 1_800).unwrap(); + assert_eq!(provider.query_timeout, Duration::from_secs(7)); + assert_eq!(provider.max_age, Duration::from_secs(1_800)); + + // A misconfigured 0 must not produce a zero-duration timeout/age. + let provider = NostrProvider::new(&cfg, 0, 0).unwrap(); + assert_eq!(provider.query_timeout, Duration::from_secs(1)); + assert_eq!(provider.max_age, Duration::from_secs(1)); + } + + #[test] + fn build_filter_has_kind_authors_and_identifier() { + let node_a = Keys::generate().public_key(); + let node_b = Keys::generate().public_key(); + let cfg = ProviderConfig { + enabled: true, + url: String::new(), + fallback_urls: vec![], + api_key: None, + token: None, + only: None, + except: None, + trusted_nodes: vec![node_a.to_hex(), node_b.to_hex()], + }; + let provider = NostrProvider::new(&cfg, 10, 1_800).unwrap(); + + let expected = Filter::new() + .kind(Kind::Custom(NOSTR_EXCHANGE_RATES_EVENT_KIND)) + .authors([node_a, node_b]) + .identifier("mostro-rates"); + assert_eq!(provider.build_filter(), expected); + } + + #[test] + fn new_rejects_empty_trusted_nodes() { + let cfg = ProviderConfig { + enabled: true, + url: String::new(), + fallback_urls: vec![], + api_key: None, + token: None, + only: None, + except: None, + trusted_nodes: vec![], + }; + assert!(NostrProvider::new(&cfg, 10, 1_800).is_err()); + } + + #[test] + fn new_rejects_invalid_hex_pubkey() { + let cfg = sample_cfg("not-a-pubkey".to_string()); + assert!(NostrProvider::new(&cfg, 10, 1_800).is_err()); + } + + /// Live-relay evidence for issue #697: exercises the real `fetch()` path + /// (not a fixture) against `wss://relay.mostro.network` and the two + /// example `trusted_nodes` pubkeys the issue names. Inherently flaky + /// against third-party infra outside this repo's control, so it's + /// `#[ignore]`d (never runs in CI) — run explicitly, alone, so no other + /// test races it to set the process-wide `NOSTR_CLIENT`: + /// + /// `cargo test price::providers::nostr::tests::live_relay_fetch_returns_real_rates -- --ignored --exact` + #[tokio::test] + #[ignore = "hits a real Nostr relay; run explicitly for manual verification"] + async fn live_relay_fetch_returns_real_rates() { + let client = Client::default(); + client + .add_relay("wss://relay.mostro.network") + .await + .expect("add_relay"); + client.connect().await; + crate::NOSTR_CLIENT + .set(client) + .expect("NOSTR_CLIENT must be unset — run this test alone (see doc comment)"); + + // Both pubkeys issue #697 names as example trusted_nodes. + let cfg = ProviderConfig { + enabled: true, + url: String::new(), + fallback_urls: vec![], + api_key: None, + token: None, + only: None, + except: None, + trusted_nodes: vec![ + "82fa8cb978b43c79b2156585bac2c011176a21d2aead6d9f7c575c005be88390".to_string(), + "00000235a3e904cfe1213a8a54d6f1ec1bef7cc6bfaabd6193e82931ccf1366a".to_string(), + ], + }; + let provider = NostrProvider::new(&cfg, 10, 1_800).expect("valid hex pubkeys"); + let http = reqwest::Client::new(); + + let quotes = provider.fetch(&http).await.expect("live relay fetch"); + println!( + "nostr provider live fetch: {} currencies — {quotes:?}", + quotes.len() + ); + assert!( + !quotes.is_empty(), + "expected at least one live currency quote" + ); + } +} diff --git a/src/price/providers/yadio.rs b/src/price/providers/yadio.rs index 2b99546c..6d2e49b1 100644 --- a/src/price/providers/yadio.rs +++ b/src/price/providers/yadio.rs @@ -127,6 +127,7 @@ mod tests { token: None, only: None, except: None, + trusted_nodes: vec![], }; let p = YadioProvider::new(&cfg); // We rebuild the request URL by appending `/exrates/BTC`; without diff --git a/src/price/store.rs b/src/price/store.rs index 8ae5906f..3ce4e866 100644 --- a/src/price/store.rs +++ b/src/price/store.rs @@ -131,6 +131,7 @@ mod tests { // exercise TTL/last-known-good semantics. Empty // is fine since the store never reads this field. contributors: Vec::new(), + nostr_anchor_dependent: false, }, ) })