From 32491b435c5a12332e90c3593d0c76a1f6200b51 Mon Sep 17 00:00:00 2001 From: Mshehu5 Date: Tue, 14 Jul 2026 14:32:26 +0100 Subject: [PATCH 1/6] Add stateless relay selection Add a deterministic relay-selection module to the main crate. Applications provide prepared relay candidates. The selector groups them by privacy bucket, orders buckets and relays from the receiver key and time window, and separates POST from POLL when enough buckets are available. This keeps the reusable selection logic independent from DNS resolution, ASMap lookup, directory trust policy, and HTTP transport. --- payjoin/src/core/mod.rs | 2 + payjoin/src/core/relay_selection.rs | 515 ++++++++++++++++++++++++++++ 2 files changed, 517 insertions(+) create mode 100644 payjoin/src/core/relay_selection.rs diff --git a/payjoin/src/core/mod.rs b/payjoin/src/core/mod.rs index ff1fdb2af..bc0659d38 100644 --- a/payjoin/src/core/mod.rs +++ b/payjoin/src/core/mod.rs @@ -32,6 +32,8 @@ pub(crate) mod hpke; #[cfg(feature = "v2")] pub mod persist; #[cfg(feature = "v2")] +pub mod relay_selection; +#[cfg(feature = "v2")] pub use crate::hpke::{HpkeKeyPair, HpkePublicKey}; #[cfg(feature = "v2")] pub(crate) mod ohttp; diff --git a/payjoin/src/core/relay_selection.rs b/payjoin/src/core/relay_selection.rs new file mode 100644 index 000000000..af038a429 --- /dev/null +++ b/payjoin/src/core/relay_selection.rs @@ -0,0 +1,515 @@ +//! Deterministic OHTTP relay ordering for BIP 77 sessions. +//! +//! Given the same relay candidates, receiver public key, request kind, and time +//! window, both sender and receiver derive the same relay order without storing +//! relay-selection state. POST requests use the preferred bucket for the current +//! window; POLL requests avoid POST buckets from the previous, current, and next +//! windows so the two mailbox directions do not use the same relay bucket when +//! enough buckets are available. +//! +//! Relay candidates are grouped into privacy buckets by the application. With +//! ASMap, a bucket is an ASN; without ASMap, each relay URL is its own bucket. +//! This module only orders those prepared candidates. DNS resolution, ASMap +//! lookup, directory filtering, and HTTP requests stay with the application. + +use std::collections::{BTreeMap, BTreeSet, VecDeque}; + +use bitcoin::hashes::{sha256, Hash, HashEngine}; + +use crate::{HpkePublicKey, Url}; + +const RELAY_SELECTION_TAG: &[u8] = b"payjoin-relay-selection"; +const CLOCK_SKEW_WINDOWS: i64 = 1; +const POST_RESERVED_COUNT: usize = 1; + +/// Number of seconds in one relay-selection window. +pub const WINDOW_SECS: u64 = 30; + +/// A wallet-specific relay value with a URL used for deterministic selection. +/// +/// This keeps transport data generic without storing the URL twice. +pub trait Relay { + /// Return the relay URL used for hashing and URL-based fallback grouping. + fn url(&self) -> &Url; +} + +/// Whether the current OHTTP request posts or polls a BIP 77 message. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RequestKind { + /// Post a message to the directory. + Post, + /// Poll the directory for a message. + Poll, +} + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +enum RelayBucket { + Asn(u32), + Url(String), +} + +impl RelayBucket { + /// Return the stable bytes hashed when ordering this privacy bucket. + fn identifier_bytes(&self) -> Vec { + match self { + Self::Asn(asn) => asn.to_be_bytes().to_vec(), + Self::Url(url) => url.as_bytes().to_vec(), + } + } +} + +/// A relay together with the identifiers needed for deterministic selection. +/// +/// `T` is owned by the application. It may be a URL, a DNS-pinned transport +/// target, or another wallet-specific relay representation. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RelayCandidate { + relay: T, + bucket: RelayBucket, +} + +impl RelayCandidate { + /// Return the application-owned relay value. + pub fn relay(&self) -> &T { &self.relay } + + /// Consume the candidate and return its application-owned relay value. + pub fn into_relay(self) -> T { self.relay } +} + +impl RelayCandidate { + /// Construct a relay grouped with other relays in the same autonomous system. + pub fn with_asn(relay: T, asn: u32) -> Self { Self { relay, bucket: RelayBucket::Asn(asn) } } + + /// Construct a relay treated as its own URL bucket when its ASN is unknown. + pub fn individual(relay: T) -> Self { + let url = relay.url().as_str().to_owned(); + Self { relay, bucket: RelayBucket::Url(url) } + } +} + +/// A deterministic relay-selection time window. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub struct TimeWindow(u64); + +impl TimeWindow { + /// Derive the selection window from Unix time and the receiver key. + /// + /// The key-derived offset prevents every session from changing its relay + /// preference at the same wall-clock boundary. + pub fn from_unix_seconds(unix_seconds: u64, receiver_pubkey: &HpkePublicKey) -> Self { + let offset = receiver_key_offset(receiver_pubkey); + Self((unix_seconds + offset) / WINDOW_SECS) + } + + fn saturating_offset(self, offset: i64) -> Self { + if offset.is_negative() { + Self(self.0.saturating_sub(offset.unsigned_abs())) + } else { + Self(self.0.saturating_add(offset as u64)) + } + } +} + +/// Return the deterministic relay order for one POST or POLL request. +/// +/// POST requests try one relay from each preferred privacy bucket. POLL +/// requests avoid buckets reserved for matching POST requests in nearby time +/// windows. If too few buckets exist, POLL falls back to the complete ordering. +pub fn select_relay_candidates<'a, T: Relay>( + candidates: &'a [RelayCandidate], + request_kind: RequestKind, + receiver_pubkey: &HpkePublicKey, + window: TimeWindow, +) -> Vec<&'a RelayCandidate> { + match request_kind { + RequestKind::Post => select_post_candidates(candidates, receiver_pubkey, window), + RequestKind::Poll => select_poll_candidates(candidates, receiver_pubkey, window), + } +} + +// POST uses the first bucket for this window. If several relays are in that +// bucket, keep only the first hash-ordered relay so POST does not consume every +// relay in one ASN. +fn select_post_candidates<'a, T: Relay>( + candidates: &'a [RelayCandidate], + receiver_pubkey: &HpkePublicKey, + window: TimeWindow, +) -> Vec<&'a RelayCandidate> { + let reserved = ranked_relay_buckets_for_post(candidates, receiver_pubkey, window) + .into_iter() + .take(POST_RESERVED_COUNT) + .collect::>(); + let mut selected_buckets = BTreeSet::new(); + + bucket_round_robin_candidates(candidates, receiver_pubkey, window) + .into_iter() + .filter(|candidate| { + reserved.contains(&candidate.bucket) + && selected_buckets.insert(candidate.bucket.clone()) + }) + .collect() +} + +// POLL avoids buckets that POST may use in the previous, current, or next +// window. If that leaves no candidate, fall back so the session can still make +// progress with degraded separation. +fn select_poll_candidates<'a, T: Relay>( + candidates: &'a [RelayCandidate], + receiver_pubkey: &HpkePublicKey, + window: TimeWindow, +) -> Vec<&'a RelayCandidate> { + let reserved = reserved_relay_buckets_for_poll(candidates, receiver_pubkey, window); + let poll_candidates = + candidates.iter().filter(|candidate| !reserved.contains(&candidate.bucket)); + + let ordered_poll_candidates = + bucket_round_robin_candidates(poll_candidates, receiver_pubkey, window); + if !ordered_poll_candidates.is_empty() { + return ordered_poll_candidates; + } + + tracing::warn!( + "Not enough relay buckets to keep POLL separate from POST, POLL may reuse POST-reserved buckets" + ); + bucket_round_robin_candidates(candidates, receiver_pubkey, window) +} + +// Reserve the POST bucket from nearby windows to tolerate small clock skew +// between sender and receiver. +fn reserved_relay_buckets_for_poll( + candidates: &[RelayCandidate], + receiver_pubkey: &HpkePublicKey, + window: TimeWindow, +) -> BTreeSet { + let mut reserved = BTreeSet::new(); + for offset in -CLOCK_SKEW_WINDOWS..=CLOCK_SKEW_WINDOWS { + let adjacent_window = window.saturating_offset(offset); + reserved.extend( + ranked_relay_buckets_for_post(candidates, receiver_pubkey, adjacent_window) + .into_iter() + .take(POST_RESERVED_COUNT), + ); + } + reserved +} + +/// Return candidates in hash order, taking one relay from each bucket per round. +/// +/// This reduces to simple hash ordering when each bucket has one relay. When +/// ASMap groups several relays into the same ASN bucket, round-robin keeps that +/// bucket from dominating the front of the order before other ASNs are tried. +fn bucket_round_robin_candidates<'a, T: Relay>( + candidates: impl IntoIterator>, + receiver_pubkey: &HpkePublicKey, + window: TimeWindow, +) -> Vec<&'a RelayCandidate> { + let mut buckets = BTreeMap::>>::new(); + for candidate in candidates { + buckets.entry(candidate.bucket.clone()).or_default().push(candidate); + } + + let mut bucket_entries = buckets + .into_iter() + .map(|(bucket, mut relays)| { + relays.sort_by_key(|candidate| relay_score(receiver_pubkey, window, candidate)); + let bucket_score = bucket_score(receiver_pubkey, window, &bucket); + (bucket_score, VecDeque::from(relays)) + }) + .collect::>(); + bucket_entries.sort_by_key(|(score, _)| *score); + + let rounds = bucket_entries.iter().map(|(_, bucket)| bucket.len()).max().unwrap_or(0); + let mut ordered = Vec::new(); + for _ in 0..rounds { + for (_, bucket) in &mut bucket_entries { + if let Some(candidate) = bucket.pop_front() { + ordered.push(candidate); + } + } + } + ordered +} + +// Deduplicate buckets first: POST reservation is by privacy bucket, not by +// individual relay. With ASMap, multiple relays may share the same ASN. +fn ranked_relay_buckets_for_post( + candidates: &[RelayCandidate], + receiver_pubkey: &HpkePublicKey, + window: TimeWindow, +) -> Vec { + let mut buckets = candidates + .iter() + .map(|candidate| candidate.bucket.clone()) + .collect::>() + .into_iter() + .collect::>(); + buckets.sort_by_key(|bucket| bucket_score(receiver_pubkey, window, bucket)); + buckets +} + +// Stagger window boundaries per session so all sessions do not rotate relay +// preferences at the same wall-clock second. +fn receiver_key_offset(receiver_pubkey: &HpkePublicKey) -> u64 { + let hash = sha256::Hash::hash(&receiver_pubkey.to_compressed_bytes()); + let mut bytes = [0u8; 8]; + bytes.copy_from_slice(&Hash::as_byte_array(&hash)[..8]); + u64::from_be_bytes(bytes) % WINDOW_SECS +} + +// Order privacy buckets, e.g. ASNs when ASMap is available or relay URLs when +// ASMap is unavailable. +fn bucket_score( + receiver_pubkey: &HpkePublicKey, + window: TimeWindow, + bucket: &RelayBucket, +) -> [u8; 32] { + let bucket_identifier = bucket.identifier_bytes(); + selection_hash(receiver_pubkey, window, b"bucket", &bucket_identifier) +} + +// Order relays inside a bucket. This is separate from bucket ordering so relay +// URLs cannot change which bucket is preferred for POST. +fn relay_score( + receiver_pubkey: &HpkePublicKey, + window: TimeWindow, + candidate: &RelayCandidate, +) -> [u8; 32] { + selection_hash(receiver_pubkey, window, b"relay", candidate.relay.url().as_str().as_bytes()) +} + +// Scope all relay-selection hashes with a tag and use labels to separate bucket +// ordering from relay ordering. +fn selection_hash( + receiver_pubkey: &HpkePublicKey, + window: TimeWindow, + label: &[u8], + payload: &[u8], +) -> [u8; 32] { + let mut engine = sha256::Hash::engine(); + engine.input(RELAY_SELECTION_TAG); + engine.input(&receiver_pubkey.to_compressed_bytes()); + engine.input(&window.0.to_be_bytes()); + engine.input(label); + engine.input(payload); + *sha256::Hash::from_engine(engine).as_byte_array() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[derive(Debug, Clone, PartialEq, Eq)] + struct TestRelay(Url); + + impl Relay for TestRelay { + fn url(&self) -> &Url { &self.0 } + } + + const RECEIVER_PUBKEY: [u8; 33] = [ + 0x02, 0x79, 0xbe, 0x66, 0x7e, 0xf9, 0xdc, 0xbb, 0xac, 0x55, 0xa0, 0x62, 0x95, 0xce, 0x87, + 0x0b, 0x07, 0x02, 0x9b, 0xfc, 0xdb, 0x2d, 0xce, 0x28, 0xd9, 0x59, 0xf2, 0x81, 0x5b, 0x16, + 0xf8, 0x17, 0x98, + ]; + + fn receiver_pubkey() -> HpkePublicKey { + HpkePublicKey::from_compressed_bytes(&RECEIVER_PUBKEY).unwrap() + } + + fn candidate(relay: &str, asn: u32) -> RelayCandidate { + let url = Url::parse(&format!("https://{relay}.example")).unwrap(); + RelayCandidate::with_asn(TestRelay(url), asn) + } + + fn individual_candidate(relay: &str) -> RelayCandidate { + let url = Url::parse(&format!("https://{relay}.example")).unwrap(); + RelayCandidate::individual(TestRelay(url)) + } + + #[test] + fn time_window_matches_fixed_vector() { + let receiver_pubkey = receiver_pubkey(); + + assert_eq!(receiver_key_offset(&receiver_pubkey), 25); + assert_eq!(TimeWindow::from_unix_seconds(0, &receiver_pubkey), TimeWindow(0)); + assert_eq!(TimeWindow::from_unix_seconds(4, &receiver_pubkey), TimeWindow(0)); + assert_eq!(TimeWindow::from_unix_seconds(5, &receiver_pubkey), TimeWindow(1)); + assert_eq!(TimeWindow::from_unix_seconds(34, &receiver_pubkey), TimeWindow(1)); + assert_eq!(TimeWindow::from_unix_seconds(35, &receiver_pubkey), TimeWindow(2)); + } + + #[test] + fn scores_depend_on_bucket_and_relay_identifiers() { + let receiver_pubkey = receiver_pubkey(); + let window = TimeWindow(100); + let relay_a = individual_candidate("relay-a"); + let relay_b = individual_candidate("relay-b"); + + assert_ne!( + bucket_score(&receiver_pubkey, window, &RelayBucket::Asn(1)), + bucket_score(&receiver_pubkey, window, &RelayBucket::Asn(2)) + ); + assert_ne!( + bucket_score(&receiver_pubkey, window, &relay_a.bucket), + bucket_score(&receiver_pubkey, window, &relay_b.bucket) + ); + assert_ne!( + relay_score(&receiver_pubkey, window, &relay_a), + relay_score(&receiver_pubkey, window, &relay_b) + ); + } + + #[test] + fn bucket_round_robin_candidates_matches_fixed_vector() { + let candidates = vec![ + candidate("relay-a", 1), + candidate("relay-b", 1), + candidate("relay-c", 2), + candidate("relay-d", 2), + candidate("relay-e", 3), + candidate("relay-f", 4), + ]; + let receiver_pubkey = receiver_pubkey(); + + let ordered = bucket_round_robin_candidates(&candidates, &receiver_pubkey, TimeWindow(100)); + + // Buckets are hash-ordered, relays are hash-ordered within each + // bucket, and round-robin selection eventually returns every relay. + assert_eq!( + ordered, + vec![ + &candidates[4], + &candidates[5], + &candidates[3], + &candidates[0], + &candidates[2], + &candidates[1], + ] + ); + } + + #[test] + fn selection_is_deterministic() { + let candidates = vec![ + candidate("relay-a", 1), + candidate("relay-b", 2), + candidate("relay-c", 3), + candidate("relay-d", 4), + ]; + let receiver_pubkey = receiver_pubkey(); + let window = TimeWindow::from_unix_seconds(1_700_000_000, &receiver_pubkey); + + let first = + select_relay_candidates(&candidates, RequestKind::Post, &receiver_pubkey, window); + let second = + select_relay_candidates(&candidates, RequestKind::Post, &receiver_pubkey, window); + + assert_eq!(first, second); + } + + #[test] + fn relays_in_same_asn_share_one_bucket() { + let candidates = vec![ + candidate("relay-a", 1), + candidate("relay-b", 1), + candidate("relay-c", 2), + candidate("relay-d", 3), + ]; + let receiver_pubkey = receiver_pubkey(); + let window = TimeWindow::from_unix_seconds(1_700_000_000, &receiver_pubkey); + + let buckets = ranked_relay_buckets_for_post(&candidates, &receiver_pubkey, window); + + assert_eq!(buckets.len(), 3); + } + + #[test] + fn individual_relays_use_their_urls_as_buckets() { + let candidates = vec![individual_candidate("relay-a"), individual_candidate("relay-b")]; + let receiver_pubkey = receiver_pubkey(); + let window = TimeWindow::from_unix_seconds(1_700_000_000, &receiver_pubkey); + + let buckets = ranked_relay_buckets_for_post(&candidates, &receiver_pubkey, window); + + assert_eq!(buckets.len(), 2); + assert_ne!(buckets[0], buckets[1]); + } + + #[test] + fn post_uses_first_bucket_for_current_window() { + let candidates = vec![ + candidate("relay-a", 1), + candidate("relay-b", 2), + candidate("relay-c", 3), + candidate("relay-d", 4), + ]; + let receiver_pubkey = receiver_pubkey(); + let window = TimeWindow(100); + + let selected = + select_relay_candidates(&candidates, RequestKind::Post, &receiver_pubkey, window); + let first_bucket = ranked_relay_buckets_for_post(&candidates, &receiver_pubkey, window) + .into_iter() + .next() + .expect("candidate buckets are not empty"); + + assert_eq!(selected.len(), 1); + assert_eq!(selected[0].bucket, first_bucket); + } + + #[test] + fn four_buckets_keep_poll_separate_from_nearby_posts() { + let candidates = vec![ + candidate("relay-a", 1), + candidate("relay-b", 2), + candidate("relay-c", 3), + candidate("relay-d", 4), + ]; + let receiver_pubkey = receiver_pubkey(); + let window = TimeWindow(100); + let nearby_post_buckets = (-CLOCK_SKEW_WINDOWS..=CLOCK_SKEW_WINDOWS) + .flat_map(|offset| { + select_relay_candidates( + &candidates, + RequestKind::Post, + &receiver_pubkey, + window.saturating_offset(offset), + ) + }) + .map(|candidate| candidate.bucket.clone()) + .collect::>(); + + let poll = + select_relay_candidates(&candidates, RequestKind::Poll, &receiver_pubkey, window); + + assert!(!poll.is_empty()); + assert!(poll.iter().all(|candidate| !nearby_post_buckets.contains(&candidate.bucket))); + } + + #[test] + fn eight_buckets_leave_several_poll_candidates() { + let candidates = + (1..=8).map(|asn| candidate(&format!("relay-{asn}"), asn)).collect::>(); + let receiver_pubkey = receiver_pubkey(); + let window = TimeWindow(100); + + let poll = + select_relay_candidates(&candidates, RequestKind::Poll, &receiver_pubkey, window); + + // At most one distinct bucket is reserved in each of three nearby windows. + assert!(poll.len() >= 5); + } + + #[test] + fn one_bucket_uses_degraded_poll_fallback() { + let candidates = vec![candidate("relay-a", 1)]; + let receiver_pubkey = receiver_pubkey(); + let window = TimeWindow(100); + + let post = + select_relay_candidates(&candidates, RequestKind::Post, &receiver_pubkey, window); + let poll = + select_relay_candidates(&candidates, RequestKind::Poll, &receiver_pubkey, window); + + assert_eq!(post, poll); + } +} From 07354b8d42ddf61223dde5f5ca7d109306c3adea Mon Sep 17 00:00:00 2001 From: Mshehu5 Date: Tue, 14 Jul 2026 14:32:26 +0100 Subject: [PATCH 2/6] Add ASMap configuration Add optional ASMap configuration for v2 relay selection. The CLI can load an ASMap file and accept user ASNs or user public IPs. Later commits use this information to avoid relays and directories that share an ASN with the user. --- Cargo-minimal.lock | 7 ++ Cargo-recent.lock | 7 ++ payjoin-cli/Cargo.toml | 2 + payjoin-cli/src/app/config.rs | 136 ++++++++++++++++++++++++++++++---- 4 files changed, 136 insertions(+), 16 deletions(-) diff --git a/Cargo-minimal.lock b/Cargo-minimal.lock index 0d287e53f..48db1c694 100644 --- a/Cargo-minimal.lock +++ b/Cargo-minimal.lock @@ -268,6 +268,12 @@ dependencies = [ "winnow 0.7.13", ] +[[package]] +name = "asmap" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "996e3818c450a9497e2f1aff7306d1c56b4198a07880222ee0aa85b6c42ac81f" + [[package]] name = "asn1-rs" version = "0.7.0" @@ -2596,6 +2602,7 @@ version = "1.0.0-rc.0" dependencies = [ "ahash 0.7.8", "anyhow", + "asmap", "async-trait", "bitcoind-async-client", "clap 4.5.45", diff --git a/Cargo-recent.lock b/Cargo-recent.lock index a4b99c41c..13d719b85 100644 --- a/Cargo-recent.lock +++ b/Cargo-recent.lock @@ -268,6 +268,12 @@ dependencies = [ "winnow 0.7.15", ] +[[package]] +name = "asmap" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "996e3818c450a9497e2f1aff7306d1c56b4198a07880222ee0aa85b6c42ac81f" + [[package]] name = "asn1-rs" version = "0.7.2" @@ -2727,6 +2733,7 @@ version = "1.0.0-rc.0" dependencies = [ "ahash 0.7.8", "anyhow", + "asmap", "async-trait", "bitcoind-async-client", "clap 4.6.1", diff --git a/payjoin-cli/Cargo.toml b/payjoin-cli/Cargo.toml index 512f0483d..f4c640e72 100644 --- a/payjoin-cli/Cargo.toml +++ b/payjoin-cli/Cargo.toml @@ -19,6 +19,7 @@ path = "src/main.rs" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [features] default = ["v2"] +asmap = ["dep:asmap"] native-certs = ["reqwest/rustls-tls-native-roots"] _manual-tls = ["reqwest/rustls-tls", "payjoin/_manual-tls", "tokio-rustls"] v1 = ["payjoin/v1", "hyper", "hyper-util", "http-body-util"] @@ -27,6 +28,7 @@ v2 = ["payjoin/v2", "payjoin/io"] [dependencies] ahash = "0.7.8" anyhow = "1.0.99" +asmap = { version = "0.1.0", optional = true } async-trait = "0.1.89" bitcoind-async-client = "0.14.0" clap = { version = "4.5.45", features = ["derive"] } diff --git a/payjoin-cli/src/app/config.rs b/payjoin-cli/src/app/config.rs index 460732e74..21366a90b 100644 --- a/payjoin-cli/src/app/config.rs +++ b/payjoin-cli/src/app/config.rs @@ -1,4 +1,10 @@ +#[cfg(all(feature = "v2", feature = "asmap"))] +use std::fmt; +#[cfg(all(feature = "v2", feature = "asmap"))] +use std::net::IpAddr; use std::path::PathBuf; +#[cfg(all(feature = "v2", feature = "asmap"))] +use std::sync::Arc; use anyhow::Result; use config::builder::DefaultState; @@ -29,6 +35,66 @@ pub struct V1Config { pub pj_endpoint: Url, } +#[cfg(all(feature = "v2", feature = "asmap"))] +#[derive(Clone)] +pub struct LoadedAsmap { + map: Arc<::asmap::Asmap>, +} + +#[cfg(all(feature = "v2", feature = "asmap"))] +impl LoadedAsmap { + pub fn lookup(&self, ip: IpAddr) -> u32 { self.map.lookup(ip) } + + pub fn as_bytes(&self) -> &[u8] { self.map.as_bytes() } +} + +#[cfg(all(feature = "v2", feature = "asmap"))] +impl<'de> Deserialize<'de> for LoadedAsmap { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + let path = PathBuf::deserialize(deserializer)?; + let map = ::asmap::Asmap::from_file(&path).map_err(|e| { + serde::de::Error::custom(format!( + "Failed to load v2.asmap.asmap_file {}: {e}", + path.display() + )) + })?; + Ok(LoadedAsmap { map: Arc::new(map) }) + } +} + +#[cfg(all(feature = "v2", feature = "asmap"))] +impl fmt::Debug for LoadedAsmap { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("LoadedAsmap").field("bytes", &self.as_bytes().len()).finish() + } +} + +#[cfg(all(feature = "v2", feature = "asmap"))] +#[derive(Debug, Clone, Deserialize)] +pub struct AsmapConfig { + #[serde(rename = "asmap_file")] + pub asmap: LoadedAsmap, + #[serde(default)] + pub user_public_ips: Vec, + #[serde(default)] + pub user_asns: Vec, +} + +#[cfg(all(feature = "v2", feature = "asmap"))] +impl AsmapConfig { + fn validate(&self) -> Result<(), ConfigError> { + if self.user_public_ips.is_empty() && self.user_asns.is_empty() { + return Err(ConfigError::Message( + "v2.asmap requires at least one of user_public_ips or user_asns".into(), + )); + } + Ok(()) + } +} + #[cfg(feature = "v2")] #[derive(Debug, Clone, Deserialize)] pub struct V2Config { @@ -36,17 +102,35 @@ pub struct V2Config { pub ohttp_keys: Option, pub ohttp_relays: Vec, pub pj_directories: Vec, + #[cfg(feature = "asmap")] + #[serde(default)] + pub asmap: Option, +} + +#[cfg(feature = "v2")] +impl V2Config { + fn validate(&self) -> Result<(), ConfigError> { + if self.pj_directories.is_empty() { + return Err(ConfigError::Message( + "At least one v2 trusted directory is required".to_owned(), + )); + } + + #[cfg(feature = "asmap")] + if let Some(asmap) = &self.asmap { + asmap.validate()?; + } + + Ok(()) + } } #[allow(clippy::large_enum_variant)] -#[derive(Debug, Clone, Deserialize)] -#[serde(tag = "version")] +#[derive(Debug, Clone)] pub enum VersionConfig { #[cfg(feature = "v1")] - #[serde(rename = "v1")] V1(V1Config), #[cfg(feature = "v2")] - #[serde(rename = "v2")] V2(V2Config), } @@ -206,7 +290,7 @@ impl Config { Version::Two => { #[cfg(feature = "v2")] { - match built_config.get::("v2") { + match load_v2_config(&built_config) { Ok(v2) => { if v2.ohttp_relays.len() < 2 { tracing::warn!( @@ -402,6 +486,20 @@ fn handle_subcommands(config: Builder, cli: &Cli) -> Result Result { + #[cfg(not(feature = "asmap"))] + if built_config.get_table("v2.asmap").is_ok() { + return Err(ConfigError::Message( + "This build does not include ASMap support. Recompile with --features asmap".to_owned(), + )); + } + + let v2 = built_config.get::("v2")?; + v2.validate()?; + Ok(v2) +} + #[cfg(feature = "v2")] fn deserialize_ohttp_keys_from_path<'de, D>( deserializer: D, @@ -409,18 +507,24 @@ fn deserialize_ohttp_keys_from_path<'de, D>( where D: serde::Deserializer<'de>, { - let path_str: Option = Option::deserialize(deserializer)?; - - match path_str { + let path: Option = Option::deserialize(deserializer)?; + match path { None => Ok(None), - Some(path) => std::fs::read(path) - .map_err(|e| serde::de::Error::custom(format!("Failed to read ohttp_keys file: {e}"))) - .and_then(|bytes| { - payjoin::OhttpKeys::decode(&bytes).map_err(|e| { - serde::de::Error::custom(format!("Failed to decode ohttp keys: {e}")) - }) - }) - .map(Some), + Some(path) => { + let bytes = std::fs::read(&path).map_err(|e| { + serde::de::Error::custom(format!( + "Failed to read ohttp_keys file {}: {e}", + path.display() + )) + })?; + let keys = payjoin::OhttpKeys::decode(&bytes).map_err(|e| { + serde::de::Error::custom(format!( + "Failed to decode ohttp keys from {}: {e}", + path.display() + )) + })?; + Ok(Some(keys)) + } } } From 5910eb6c3f1958c6e6683f4ca1eac6c624b8589c Mon Sep 17 00:00:00 2001 From: Mshehu5 Date: Tue, 14 Jul 2026 14:32:26 +0100 Subject: [PATCH 3/6] Fetch OHTTP keys via resolved relays Expose OHTTP key fetching through a relay with known socket addresses. ASMap-aware callers resolve and classify relay IPs before making HTTP requests. Passing those addresses into reqwest prevents a second DNS lookup from using a different address than the one that was checked. Move the CLI HTTP client setup behind a builder helper so v2 code can add resolve_to_addrs before building the client. The v1 path still builds a client through http_agent. --- payjoin-cli/src/app/mod.rs | 34 ++++++------- payjoin/src/core/io.rs | 102 +++++++++++++++++++++++++++++-------- 2 files changed, 99 insertions(+), 37 deletions(-) diff --git a/payjoin-cli/src/app/mod.rs b/payjoin-cli/src/app/mod.rs index a0174e187..431d41fbd 100644 --- a/payjoin-cli/src/app/mod.rs +++ b/payjoin-cli/src/app/mod.rs @@ -71,28 +71,28 @@ pub trait App: Send + Sync { } } -#[cfg(feature = "_manual-tls")] +#[cfg(feature = "v1")] fn http_agent(config: &Config) -> Result { - Ok(http_agent_builder(config.root_certificate.as_ref())?.build()?) -} - -#[cfg(not(feature = "_manual-tls"))] -fn http_agent(_config: &Config) -> Result { - Ok(reqwest::Client::builder().http1_only().build()?) + Ok(http_client_builder(config)?.build()?) } -#[cfg(feature = "_manual-tls")] -fn http_agent_builder( - root_cert_path: Option<&std::path::PathBuf>, -) -> Result { - let mut builder = reqwest::ClientBuilder::new().use_rustls_tls().http1_only(); +pub(crate) fn http_client_builder(config: &Config) -> Result { + #[cfg(feature = "_manual-tls")] + { + let mut builder = reqwest::ClientBuilder::new().use_rustls_tls().http1_only(); + if let Some(root_cert_path) = config.root_certificate.as_ref() { + let cert_der = std::fs::read(root_cert_path)?; + builder = builder + .add_root_certificate(reqwest::tls::Certificate::from_der(cert_der.as_slice())?); + } + Ok(builder) + } - if let Some(root_cert_path) = root_cert_path { - let cert_der = std::fs::read(root_cert_path)?; - builder = - builder.add_root_certificate(reqwest::tls::Certificate::from_der(cert_der.as_slice())?) + #[cfg(not(feature = "_manual-tls"))] + { + let _ = config; + Ok(reqwest::Client::builder().http1_only()) } - Ok(builder) } async fn handle_interrupt(tx: watch::Sender<()>) { diff --git a/payjoin/src/core/io.rs b/payjoin/src/core/io.rs index 05994c1a6..3a6977aa7 100644 --- a/payjoin/src/core/io.rs +++ b/payjoin/src/core/io.rs @@ -1,11 +1,12 @@ //! IO-related types and functions. Specifically, fetching OHTTP keys from a payjoin directory. +use std::net::SocketAddr; use std::time::Duration; use http::header::ACCEPT; -use reqwest::{Client, Proxy}; +use reqwest::{Client, ClientBuilder, Proxy}; use crate::into_url::IntoUrl; -use crate::OhttpKeys; +use crate::{OhttpKeys, Url}; /// Fetch the ohttp keys from the specified payjoin directory via proxy. /// @@ -19,16 +20,31 @@ pub async fn fetch_ohttp_keys( ohttp_relay: impl IntoUrl, payjoin_directory: impl IntoUrl, ) -> Result { - let ohttp_keys_url = payjoin_directory.into_url()?.join("/.well-known/ohttp-gateway")?; - let proxy = Proxy::all(ohttp_relay.into_url()?.as_str())?; - let client = Client::builder().proxy(proxy).http1_only().build()?; - let res = client - .get(ohttp_keys_url.as_str()) - .timeout(Duration::from_secs(10)) - .header(ACCEPT, "application/ohttp-keys") - .send() - .await?; - parse_ohttp_keys_response(res).await + fetch_ohttp_keys_inner( + ohttp_relay.into_url()?, + payjoin_directory.into_url()?, + Client::builder(), + None, + ) + .await +} + +/// Fetch OHTTP keys through a relay using previously resolved relay addresses. +/// +/// The relay URL remains unchanged for proxy and TLS hostname validation while +/// its DNS resolution is overridden with `relay_addresses`. +pub async fn fetch_ohttp_keys_with_relay_addresses( + ohttp_relay: impl IntoUrl, + payjoin_directory: impl IntoUrl, + relay_addresses: &[SocketAddr], +) -> Result { + fetch_ohttp_keys_inner( + ohttp_relay.into_url()?, + payjoin_directory.into_url()?, + Client::builder(), + Some(relay_addresses), + ) + .await } /// Fetch the ohttp keys from the specified payjoin directory via proxy. @@ -47,14 +63,49 @@ pub async fn fetch_ohttp_keys_with_cert( payjoin_directory: impl IntoUrl, cert_der: &[u8], ) -> Result { - let ohttp_keys_url = payjoin_directory.into_url()?.join("/.well-known/ohttp-gateway")?; - let proxy = Proxy::all(ohttp_relay.into_url()?.as_str())?; - let client = Client::builder() - .use_rustls_tls() - .add_root_certificate(reqwest::tls::Certificate::from_der(cert_der)?) - .proxy(proxy) - .http1_only() - .build()?; + fetch_ohttp_keys_inner( + ohttp_relay.into_url()?, + payjoin_directory.into_url()?, + Client::builder() + .use_rustls_tls() + .add_root_certificate(reqwest::tls::Certificate::from_der(cert_der)?), + None, + ) + .await +} + +/// Fetch OHTTP keys through a resolved relay using a custom TLS certificate. +#[cfg(feature = "_manual-tls")] +pub async fn fetch_ohttp_keys_with_cert_and_relay_addresses( + ohttp_relay: impl IntoUrl, + payjoin_directory: impl IntoUrl, + cert_der: &[u8], + relay_addresses: &[SocketAddr], +) -> Result { + fetch_ohttp_keys_inner( + ohttp_relay.into_url()?, + payjoin_directory.into_url()?, + Client::builder() + .use_rustls_tls() + .add_root_certificate(reqwest::tls::Certificate::from_der(cert_der)?), + Some(relay_addresses), + ) + .await +} + +async fn fetch_ohttp_keys_inner( + ohttp_relay: Url, + payjoin_directory: Url, + mut builder: ClientBuilder, + relay_addresses: Option<&[SocketAddr]>, +) -> Result { + let ohttp_keys_url = payjoin_directory.join("/.well-known/ohttp-gateway")?; + let proxy = Proxy::all(ohttp_relay.as_str())?; + builder = builder.proxy(proxy).http1_only(); + if let (Some(domain), Some(addresses)) = (ohttp_relay.domain(), relay_addresses) { + builder = builder.resolve_to_addrs(domain, addresses); + } + let client = builder.build()?; let res = client .get(ohttp_keys_url.as_str()) .timeout(Duration::from_secs(10)) @@ -85,6 +136,17 @@ pub enum Error { Internal(InternalError), } +impl Error { + /// Whether retrying the request through another relay may succeed. + pub fn is_retryable(&self) -> bool { + matches!( + self, + Self::Internal(InternalError(InternalErrorInner::Reqwest(error))) + if error.is_timeout() || error.is_connect() || error.is_request() + ) + } +} + #[derive(Debug)] pub struct InternalError(InternalErrorInner); From 76a480e733ea686d39dcd8064fd080bec8491d3b Mon Sep 17 00:00:00 2001 From: Mshehu5 Date: Tue, 14 Jul 2026 14:32:27 +0100 Subject: [PATCH 4/6] Prepare AS-aware relays in the CLI Resolve configured v2 directories and OHTTP relays before selection. When ASMap is configured, the CLI maps resolved IPs to ASNs, rejects mixed-ASN hosts, skips directories that share the user ASN, and skips relays that share the user or selected directory ASN. The CLI then converts the remaining relays into main-crate RelayCandidate values. Without ASMap, relays are still resolved and treated as individual URL buckets. --- payjoin-cli/src/app/v2/network.rs | 208 +++++++++++ payjoin-cli/src/app/v2/relay_selection.rs | 415 ++++++++++++++++++++++ 2 files changed, 623 insertions(+) create mode 100644 payjoin-cli/src/app/v2/network.rs create mode 100644 payjoin-cli/src/app/v2/relay_selection.rs diff --git a/payjoin-cli/src/app/v2/network.rs b/payjoin-cli/src/app/v2/network.rs new file mode 100644 index 000000000..61ae60fc7 --- /dev/null +++ b/payjoin-cli/src/app/v2/network.rs @@ -0,0 +1,208 @@ +#[cfg(feature = "asmap")] +use std::collections::BTreeSet; +use std::net::{IpAddr, SocketAddr, ToSocketAddrs}; + +use anyhow::{anyhow, bail, Context, Result}; +use payjoin::relay_selection::Relay as SelectionRelay; +use payjoin::Url; + +use crate::app::config::V2Config; +#[cfg(feature = "asmap")] +use crate::app::config::{AsmapConfig, LoadedAsmap}; + +#[cfg(feature = "asmap")] +pub(crate) type Asn = u32; + +/// A URL together with the socket addresses resolved for its host. +/// +/// Requests keep using `url` for the HTTP target and TLS hostname validation, +/// while `socket_addrs` pins reqwest to the addresses that were already +/// resolved and, when ASMap is enabled, ASN-checked. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct ResolvedUrl { + pub(crate) url: Url, + /// Addresses later passed to reqwest so it does not re-resolve the host. + pub(crate) socket_addrs: Vec, +} + +impl SelectionRelay for ResolvedUrl { + fn url(&self) -> &Url { &self.url } +} + +/// A resolved relay or directory, optionally annotated with its ASN. +/// +/// The ASN is present only after ASMap lookup succeeds. The resolved URL is +/// kept in both ASMap and non-ASMap paths so later HTTP requests can reuse the +/// DNS result. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct ResolvedServer { + /// URL plus resolved socket addresses. + pub(crate) resolved: ResolvedUrl, + /// ASN is present only when the server was resolved through ASMap. + #[cfg(feature = "asmap")] + pub(crate) asn: Option, +} + +/// Supplies DNS and optional ASMap lookups for relay and directory URLs. +/// +/// Tests implement this trait with fixed answers. The runtime implementation +/// uses system DNS and the configured ASMap. +pub(crate) trait UrlResolver { + /// Return IP addresses for a host and port. + fn resolve_host(&self, host: &str, port: u16) -> Result>; + #[cfg(feature = "asmap")] + /// Return the ASN for an IP, or `None` when ASMap has no mapping. + fn lookup_asn(&self, ip: IpAddr) -> Result>; +} + +/// Resolves mailroom relay and directory hosts using system DNS. +/// +/// When ASMap is configured, it also maps resolved IP addresses to ASNs. +#[derive(Debug, Clone)] +pub(crate) struct MailroomUrlResolver { + #[cfg(feature = "asmap")] + asmap: Option, +} + +impl MailroomUrlResolver { + #[cfg(feature = "asmap")] + pub(crate) fn new(v2: &V2Config) -> Self { + Self { asmap: v2.asmap.as_ref().map(|cfg| cfg.asmap.clone()) } + } + + #[cfg(not(feature = "asmap"))] + pub(crate) fn new(_v2: &V2Config) -> Self { Self {} } +} + +impl UrlResolver for MailroomUrlResolver { + fn resolve_host(&self, host: &str, port: u16) -> Result> { + if let Ok(ip) = host.parse::() { + return Ok(vec![ip]); + } + + let resolved = (host, port) + .to_socket_addrs() + .with_context(|| format!("Failed to resolve host {host}:{port}"))? + .map(|addr| addr.ip()) + .collect::>(); + Ok(resolved) + } + + #[cfg(feature = "asmap")] + fn lookup_asn(&self, ip: IpAddr) -> Result> { + let Some(asmap) = &self.asmap else { + return Ok(None); + }; + let asn = asmap.lookup(ip); + Ok((asn != 0).then_some(asn)) + } +} + +/// Return the user's configured ASNs plus ASNs found from configured public IPs. +#[cfg(feature = "asmap")] +pub(crate) fn user_asns(asmap: &AsmapConfig, network: &impl UrlResolver) -> Result> { + let mut user_asns = asmap.user_asns.iter().copied().collect::>(); + for ip in &asmap.user_public_ips { + let asn = network.lookup_asn(*ip)?.ok_or_else(|| { + anyhow!("Failed to map user public IP {ip} to an ASN using the ASMap") + })?; + user_asns.insert(asn); + } + Ok(user_asns) +} + +/// Resolve a URL to socket addresses without assigning an ASN. +/// +/// This is used when ASMap is unavailable or disabled. The returned +/// `ResolvedServer` can still pin later HTTP requests to the resolved +/// addresses. +pub(crate) fn resolve_url(network: &impl UrlResolver, url: &Url) -> Result { + Ok(ResolvedServer { + resolved: resolved_url_from_ips(url, &resolve_url_ips(network, url)?)?, + #[cfg(feature = "asmap")] + asn: None, + }) +} + +/// Resolve a URL and require all resolved IPs to map to one ASN. +/// +/// Mixed-ASN hostnames are rejected because relay selection treats each resolved +/// server as belonging to one privacy bucket. +#[cfg(feature = "asmap")] +pub(crate) fn resolve_url_with_asn( + network: &impl UrlResolver, + url: &Url, +) -> Result { + let ips = resolve_url_ips(network, url)?; + + let mut asns = BTreeSet::new(); + for ip in &ips { + let asn = network.lookup_asn(*ip)?.ok_or_else(|| { + anyhow!("{} resolved to {ip}, which could not be mapped to an ASN", url.as_str()) + })?; + asns.insert(asn); + } + + match asns.len() { + 1 => Ok(ResolvedServer { + resolved: resolved_url_from_ips(url, &ips)?, + asn: Some(*asns.first().expect("checked len")), + }), + 0 => bail!("{} resolved to no ASN-mapped addresses", url.as_str()), + _ => bail!( + "{} resolves to multiple ASNs {:?}; mixed-ASN hostnames are rejected", + url.as_str(), + asns + ), + } +} + +/// Resolve the URL host to a sorted, deduplicated list of IP addresses. +/// +/// IP literals are returned directly. Domain names are resolved through the +/// supplied resolver. +fn resolve_url_ips(network: &impl UrlResolver, url: &Url) -> Result> { + let port = relay_port(url)?; + let host = url.host_str(); + let mut ips = if let Some(ip) = parse_ip_literal(&host) { + vec![ip] + } else { + network.resolve_host(&host, port)? + }; + if ips.is_empty() { + bail!("{} resolved to no IP addresses", url.as_str()); + } + ips.sort(); + ips.dedup(); + Ok(ips) +} + +/// Build a `ResolvedUrl` from IPs that were already resolved for `url`. +/// +/// The IPs are paired with the URL port so reqwest can later use them with +/// `resolve_to_addrs`. +fn resolved_url_from_ips(url: &Url, ips: &[IpAddr]) -> Result { + let port = relay_port(url)?; + let socket_addrs = ips.iter().copied().map(|ip| SocketAddr::new(ip, port)).collect(); + Ok(ResolvedUrl { url: url.clone(), socket_addrs }) +} + +fn relay_port(url: &Url) -> Result { + url.port().or_else(|| known_default_port(url)).ok_or_else(|| { + anyhow!("Unsupported scheme {} for relay/directory URL {}", url.scheme(), url.as_str()) + }) +} + +pub(crate) fn known_default_port(url: &Url) -> Option { + match url.scheme() { + "https" => Some(443), + "http" => Some(80), + _ => None, + } +} + +fn parse_ip_literal(host: &str) -> Option { + host.parse::() + .ok() + .or_else(|| host.strip_prefix('[')?.strip_suffix(']')?.parse::().ok()) +} diff --git a/payjoin-cli/src/app/v2/relay_selection.rs b/payjoin-cli/src/app/v2/relay_selection.rs new file mode 100644 index 000000000..93baf2738 --- /dev/null +++ b/payjoin-cli/src/app/v2/relay_selection.rs @@ -0,0 +1,415 @@ +//! Stateless OHTTP relay selection for BIP77 sessions. +//! +//! This module has two jobs: +//! - choose the receiver's directory and usable relay set before the receiver key exists +//! - use the receiver key, request kind, and time window to choose relays without +//! storing a current relay index or failed-relay state +//! +#[cfg(feature = "asmap")] +use std::collections::BTreeSet; +use std::time::{SystemTime, UNIX_EPOCH}; + +#[cfg(feature = "asmap")] +use anyhow::anyhow; +use anyhow::{bail, Context, Result}; +use payjoin::bitcoin::key::rand::seq::SliceRandom; +use payjoin::bitcoin::key::rand::thread_rng; +use payjoin::relay_selection::{ + select_relay_candidates, RelayCandidate as SelectionCandidate, RequestKind, TimeWindow, +}; +use payjoin::uri::v2::PjParam as V2PjParam; +use payjoin::{HpkePublicKey, PjParam, Url}; + +use super::network::{known_default_port, resolve_url, ResolvedServer, ResolvedUrl, UrlResolver}; +#[cfg(feature = "asmap")] +use super::network::{resolve_url_with_asn, user_asns, Asn}; +use crate::app::config::V2Config; + +/// Directory and relay set chosen before the receiver key is available. +/// +/// The receiver needs this during session creation: it must choose a directory +/// and fetch OHTTP keys before the receiver pubkey can be read from the endpoint. +/// The resolved addresses are kept so later HTTP requests use the same DNS +/// results that were checked during selection. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct ReceiverNetworkSelection { + pub(crate) directory: ResolvedUrl, + pub(crate) relays: Vec, +} + +/// Chooses which OHTTP relays to try for a request. +/// +/// It stores the usable relay candidates and the receiver public key. For each +/// POST or POLL request, it combines those values with the current time window +/// to compute a fresh relay order. It does not remember the last relay used or +/// keep a cursor into the previous ordering. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct RelaySelector { + relays: Vec, + receiver_pubkey: HpkePublicKey, +} + +type RelayCandidate = SelectionCandidate; + +impl RelaySelector { + /// Return the relay order to try for one POST or POLL request. + /// + /// The ordering is recomputed from the receiver key and current time, so no + /// relay-selection progress needs to be stored between requests. + pub(crate) fn select_relays_for_request( + &self, + request_kind: RequestKind, + ) -> Result> { + let selected = select_relay_candidates( + &self.relays, + request_kind, + &self.receiver_pubkey, + current_time_window(&self.receiver_pubkey), + ); + if selected.is_empty() { + bail!("No valid relays available"); + } + Ok(selected.into_iter().map(|candidate| candidate.relay().clone()).collect()) + } +} + +fn current_time_window(receiver_pubkey: &HpkePublicKey) -> TimeWindow { + let unix_seconds = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(); + TimeWindow::from_unix_seconds(unix_seconds, receiver_pubkey) +} + +/// Convert the receiver network selection into request-time relay selection after +/// the receiver endpoint exists and contains the receiver pubkey. +pub(crate) fn relay_selector_from_network_selection( + endpoint: &str, + network_selection: ReceiverNetworkSelection, +) -> Result { + let pj_param = parse_v2_pj_param(endpoint)?; + let endpoint_directory = directory_from_endpoint(endpoint)?; + if normalized_url(&endpoint_directory) != normalized_url(&network_selection.directory.url) { + bail!( + "Receiver endpoint directory {} does not match selected directory {}", + endpoint_directory.as_str(), + network_selection.directory.url.as_str() + ); + } + Ok(RelaySelector { + relays: network_selection.relays, + receiver_pubkey: pj_param.receiver_pubkey().clone(), + }) +} + +/// Choose the receiver directory and usable relay set for a new session. +/// +/// This runs before the receiver pubkey is available. It can filter directories +/// and relays by DNS and ASMap data, but request-time relay ordering waits until +/// the endpoint has been created and the receiver pubkey is known. +pub(crate) fn choose_receiver_network_selection( + v2: &V2Config, + network: &impl UrlResolver, + excluded_directories: &[Url], +) -> Result { + let chosen_directory = choose_directory(v2, network, excluded_directories)?; + #[cfg(feature = "asmap")] + let directory_asn = chosen_directory.asn; + let directory = chosen_directory.resolved; + let mut relays = relay_candidates( + v2, + network, + #[cfg(feature = "asmap")] + directory_asn, + )?; + + if relays.is_empty() { + bail!("No valid relays available for the selected directory {}", directory.url.as_str()); + } + + relays.shuffle(&mut thread_rng()); + Ok(ReceiverNetworkSelection { directory, relays }) +} + +pub(crate) fn relay_selector_from_endpoint( + v2: &V2Config, + endpoint: &str, + network: &impl UrlResolver, +) -> Result { + let pj_param = parse_v2_pj_param(endpoint)?; + let directory_url = directory_from_endpoint(endpoint)?; + ensure_directory_trusted(v2, &directory_url)?; + #[cfg(feature = "asmap")] + let (directory, directory_asn) = if let Some(asmap) = &v2.asmap { + let user_asns = user_asns(asmap, network)?; + let directory = resolve_url_with_asn(network, &directory_url)?; + let directory_asn = directory.asn.expect("ASMap directory candidates carry a resolved ASN"); + if user_asns.contains(&directory_asn) { + bail!( + "Endpoint directory {} shares ASN {} with the user", + directory.resolved.url.as_str(), + directory_asn + ); + } + (directory.resolved, Some(directory_asn)) + } else { + (resolve_url(network, &directory_url)?.resolved, None) + }; + #[cfg(not(feature = "asmap"))] + let directory = resolve_url(network, &directory_url)?.resolved; + + let receiver_pubkey = pj_param.receiver_pubkey(); + let relays = relay_candidates( + v2, + network, + #[cfg(feature = "asmap")] + directory_asn, + )?; + if relays.is_empty() { + bail!( + "No valid relays available after filtering user and directory ASNs for {}", + directory.url.as_str() + ); + } + + Ok(RelaySelector { relays, receiver_pubkey: receiver_pubkey.clone() }) +} + +/// Choose a resolvable trusted directory. +/// +/// With ASMap enabled, directories sharing an ASN with the user are skipped. +/// Without ASMap, the first resolvable trusted directory is used. +fn choose_directory( + v2: &V2Config, + network: &impl UrlResolver, + excluded_directories: &[Url], +) -> Result { + let mut directories = v2 + .pj_directories + .iter() + .filter(|candidate| { + !excluded_directories + .iter() + .any(|excluded| normalized_url(excluded) == normalized_url(candidate)) + }) + .cloned() + .collect::>(); + directories.shuffle(&mut thread_rng()); + if directories.is_empty() { + bail!("No trusted directories remain after excluding failed directories"); + } + + #[cfg(feature = "asmap")] + if let Some(asmap) = &v2.asmap { + let user_asns = user_asns(asmap, network)?; + for directory in directories { + match resolve_url_with_asn(network, &directory) { + Ok(candidate) + if candidate.asn.map(|asn| !user_asns.contains(&asn)).unwrap_or(false) => + { + return Ok(candidate); + } + Ok(candidate) => tracing::debug!( + "Skipping directory {} because it shares an ASN with the user", + candidate.resolved.url + ), + Err(error) => tracing::debug!( + "Skipping directory {} because resolution failed: {error:#}", + directory + ), + } + } + bail!("No trusted directories remain after resolution and ASMap filtering"); + } + + for directory in directories { + match resolve_url(network, &directory) { + Ok(candidate) => return Ok(candidate), + Err(error) => tracing::debug!( + "Skipping directory {} because resolution failed: {error:#}", + directory + ), + } + } + bail!("No trusted directories could be resolved") +} + +/// Build relay candidates for request-time selection. +/// +/// With ASMap enabled, relays sharing the user ASN or chosen directory ASN are +/// skipped and the remaining relays are bucketed by ASN. Without ASMap, each +/// resolved relay is kept as its own URL bucket. +fn relay_candidates( + v2: &V2Config, + network: &impl UrlResolver, + #[cfg(feature = "asmap")] directory_asn: Option, +) -> Result> { + #[cfg(feature = "asmap")] + if let Some(asmap) = &v2.asmap { + let user_asns = user_asns(asmap, network)?; + let directory_asn = directory_asn.expect("ASMap directory candidates carry a resolved ASN"); + let candidates = v2 + .ohttp_relays + .iter() + .filter_map(|url| match resolve_url_with_asn(network, url) { + Ok(candidate) => Some(candidate), + Err(error) => { + tracing::debug!("Skipping relay {url} because resolution failed: {error:#}"); + None + } + }) + .collect(); + return asn_relay_candidates(candidates, user_asns, directory_asn); + } + let relays = v2 + .ohttp_relays + .iter() + .filter_map(|url| match resolve_url(network, url) { + Ok(target) => Some(RelayCandidate::individual(target.resolved)), + Err(error) => { + tracing::debug!("Skipping relay {url} because resolution failed: {error:#}"); + None + } + }) + .collect(); + Ok(relays) +} + +/// Convert ASN-resolved relays into library relay candidates. +/// +/// Relays in the user ASN or selected directory ASN are removed. The remaining +/// relays carry their ASN so the library selector can group them by AS. +#[cfg(feature = "asmap")] +fn asn_relay_candidates( + candidates: Vec, + user_asns: BTreeSet, + directory_asn: Asn, +) -> Result> { + let mut filtered = vec![]; + for candidate in candidates { + let asn = + candidate.asn.ok_or_else(|| anyhow!("ASMap relay candidate lacks a resolved ASN"))?; + if asn != directory_asn && !user_asns.contains(&asn) { + filtered.push((candidate.resolved, asn)); + } + } + Ok(filtered + .into_iter() + .map(|(resolved, asn)| RelayCandidate::with_asn(resolved, asn)) + .collect()) +} + +pub(crate) fn ensure_directory_trusted(v2: &V2Config, directory: &Url) -> Result<()> { + if v2 + .pj_directories + .iter() + .any(|candidate| normalized_url(candidate) == normalized_url(directory)) + { + return Ok(()); + } + + bail!( + "The directory embedded in the BIP21 URI is not in the configured trusted directory set: {}", + directory.as_str() + ); +} + +fn normalized_url(url: &Url) -> String { + let scheme = url.scheme().to_ascii_lowercase(); + let host = url.host_str().to_ascii_lowercase(); + let default_port = known_default_port(url); + + let mut normalized = format!("{scheme}://{host}"); + if let Some(port) = url.port() { + if Some(port) != default_port { + normalized.push(':'); + normalized.push_str(&port.to_string()); + } + } + + let path = url.path().trim_end_matches('/'); + if !path.is_empty() && path != "/" { + normalized.push_str(path); + } + + normalized +} + +fn parse_v2_pj_param(endpoint: &str) -> Result { + match PjParam::parse(endpoint)? { + PjParam::V2(pj_param) => Ok(pj_param), + #[cfg(feature = "v1")] + PjParam::V1(_) => bail!("Expected a BIP77 endpoint, got a BIP78 endpoint"), + _ => bail!("Expected a BIP77 endpoint"), + } +} + +pub(crate) fn directory_from_endpoint(endpoint: &str) -> Result { + let endpoint = Url::parse(endpoint)?; + let mut raw = format!("{}://{}", endpoint.scheme(), endpoint.host_str()); + if let Some(port) = endpoint.port() { + raw.push(':'); + raw.push_str(&port.to_string()); + } + + let mut segments = endpoint + .path_segments() + .expect("payjoin::Url path_segments() is always available") + .collect::>(); + if segments.is_empty() { + bail!("The BIP77 endpoint has no session path segment"); + } + segments.pop(); + + if segments.is_empty() { + raw.push('/'); + } else { + raw.push('/'); + raw.push_str(&segments.join("/")); + } + + Url::parse(&raw) + .with_context(|| format!("Failed to derive the directory from endpoint {endpoint}")) +} + +#[cfg(test)] +mod tests { + use std::net::IpAddr; + + use super::*; + + struct TestNetwork; + + impl UrlResolver for TestNetwork { + fn resolve_host(&self, host: &str, _port: u16) -> Result> { + if host.starts_with("down-") { + bail!("simulated DNS failure for {host}"); + } + Ok(vec!["192.0.2.1".parse().expect("valid test IP")]) + } + + #[cfg(feature = "asmap")] + fn lookup_asn(&self, _ip: IpAddr) -> Result> { Ok(Some(64500)) } + } + + #[test] + fn skips_unresolvable_directories_and_relays() { + let config = V2Config { + ohttp_keys: None, + ohttp_relays: vec![ + Url::parse("https://down-relay.example").expect("valid URL"), + Url::parse("https://relay.example").expect("valid URL"), + ], + pj_directories: vec![ + Url::parse("https://down-directory.example").expect("valid URL"), + Url::parse("https://directory.example").expect("valid URL"), + ], + #[cfg(feature = "asmap")] + asmap: None, + }; + + let selection = + choose_receiver_network_selection(&config, &TestNetwork, &[]).expect("fallback works"); + + assert_eq!(selection.directory.url.as_str(), "https://directory.example/"); + assert_eq!(selection.relays.len(), 1); + assert_eq!(selection.relays[0].relay().url.as_str(), "https://relay.example/"); + } +} From 76ba401e13e429e7956b7b1cc0b3dc996f9a90d3 Mon Sep 17 00:00:00 2001 From: Mshehu5 Date: Tue, 14 Jul 2026 14:32:27 +0100 Subject: [PATCH 5/6] Use AS-aware relays for v2 OHTTP Use the prepared relay selector for v2 OHTTP POST and POLL requests. Sender and receiver flows now derive relay order from the receiver key, request kind, and current time window. OHTTP key fetching also uses resolved relay addresses so the relay used for bootstrapping is the one that was checked. Update e2e child process invocations to pass the local test directory as a trusted directory. Sender and resume paths validate the directory embedded in the BIP21 URI, so the local test directory must be configured explicitly. --- payjoin-cli/src/app/v2/mod.rs | 294 ++++++++++++++++++++++---------- payjoin-cli/src/app/v2/ohttp.rs | 250 +++++++++++++++------------ payjoin-cli/tests/e2e.rs | 18 ++ 3 files changed, 364 insertions(+), 198 deletions(-) diff --git a/payjoin-cli/src/app/v2/mod.rs b/payjoin-cli/src/app/v2/mod.rs index 73247b7e8..7d1239922 100644 --- a/payjoin-cli/src/app/v2/mod.rs +++ b/payjoin-cli/src/app/v2/mod.rs @@ -12,6 +12,7 @@ use payjoin::receive::v2::{ ReceiverBuilder, SessionOutcome as ReceiverSessionOutcome, UncheckedOriginalPayload, WantsFeeRange, WantsInputs, WantsOutputs, }; +use payjoin::relay_selection::RequestKind; use payjoin::send::v2::{ replay_event_log as replay_sender_event_log, PendingFallback as SenderPendingFallback, PollingForProposal, SendSession, Sender, SenderBuilder, SessionOutcome as SenderSessionOutcome, @@ -23,13 +24,18 @@ use tokio::sync::watch; use super::config::Config; use super::wallet::BitcoindWallet; use super::App as AppTrait; -use crate::app::v2::ohttp::MailroomManager; -use crate::app::{handle_interrupt, http_agent}; +#[cfg(feature = "v1")] +use crate::app::http_agent; +use crate::app::v2::ohttp::{classify_reqwest_error, MailroomManager, RelayAttemptError}; +use crate::app::v2::relay_selection::RelaySelector; +use crate::app::{handle_interrupt, http_client_builder}; use crate::cli::Role as CliRole; use crate::db::v2::{ReceiverPersister, SenderPersister, SessionId}; use crate::db::Database; +pub(crate) mod network; mod ohttp; +pub(crate) mod relay_selection; const W_ID: usize = 36; const W_ROLE: usize = 15; @@ -41,8 +47,9 @@ const TRANSIENT_RETRY_DELAY: std::time::Duration = std::time::Duration::from_sec /// A request-construction error that can report whether it was caused by /// session expiry. Implemented by the sender/receiver request-building error -/// types so `post_via_relay` can hand expiry back to the caller (which owns the -/// typestate needed to react) instead of flattening it into `anyhow::Error`. +/// types so the relay-send helper can hand expiry back to the caller (which +/// owns the typestate needed to react) instead of flattening it into +/// `anyhow::Error`. trait RequestExpiry { fn expired(&self) -> bool; } @@ -55,9 +62,9 @@ impl RequestExpiry for payjoin::receive::v2::CreateRequestError { fn expired(&self) -> bool { self.is_expired() } } -/// Outcome of building and posting a request via `post_via_relay`. HTTP -/// failures are retried against other relays inside the helper; only session -/// expiry and fatal build errors escape to the caller. +/// Outcome of building and posting a request via selected OHTTP relays. HTTP +/// failures are retried against other selected relays inside the helper; only +/// session expiry and fatal build errors escape to the caller. enum RelayPost { Posted(reqwest::Response, T), Expired, @@ -214,6 +221,7 @@ impl AppTrait for App { .assume_checked() .check_pj_supported() .map_err(|_| anyhow!("URI does not support Payjoin"))?; + let pj_endpoint = uri.extras.endpoint(); let address = uri.address; let amount = uri.amount.ok_or_else(|| anyhow!("please specify the amount in the Uri"))?; match uri.extras.pj_param() { @@ -269,6 +277,8 @@ impl AppTrait for App { Ok(()) } PjParam::V2(pj_param) => { + let directory = relay_selection::directory_from_endpoint(&pj_endpoint)?; + relay_selection::ensure_directory_trusted(self.config.v2()?, &directory)?; let receiver_pubkey = pj_param.receiver_pubkey(); let sender_state = self .db @@ -324,23 +334,12 @@ impl AppTrait for App { async fn receive_payjoin(&self, amount: Amount) -> Result<()> { let address = self.wallet().get_new_address()?; + let v2_config = self.config.v2()?; + let network = network::MailroomUrlResolver::new(v2_config); + let (network_selection, ohttp_keys) = + self.mailroom_manager.bootstrap_receiver(&network).await?; let persister = ReceiverPersister::new(self.db.clone())?; - let (directory, ohttp_keys) = loop { - let directory = self.mailroom_manager.choose_directory()?; - match self - .mailroom_manager - .unwrap_ohttp_keys_or_else_fetch_from_directory(&directory) - .await - { - Ok(keys) => break (directory, keys.ohttp_keys), - Err(e) => { - tracing::debug!("Directory {directory} failed: {e:#}"); - self.mailroom_manager.add_failed_directory(directory); - self.mailroom_manager.clear_failed_relays(); - continue; - } - } - }; + let directory = network_selection.directory.url.clone(); let mut receiver_builder = ReceiverBuilder::new(address, directory.as_str(), ohttp_keys)?.with_amount(amount); if let Some(max_fee_rate) = self.config.max_fee_rate { @@ -354,6 +353,11 @@ impl AppTrait for App { persister.print("Session established"); let pj_uri = session.pj_uri(); + let receiver_endpoint = pj_uri.extras.endpoint(); + let relay_selector = relay_selection::relay_selector_from_network_selection( + &receiver_endpoint, + network_selection, + )?; persister.print("Request Payjoin by sharing this Payjoin Uri:"); println!("{pj_uri}"); @@ -365,6 +369,7 @@ impl AppTrait for App { res = self.process_receiver_session( ReceiveSession::Initialized(session.clone()), &persister, + &relay_selector, ) => res?, _ = interrupt.changed() => { let session_id = persister.session_id(); @@ -401,12 +406,39 @@ impl AppTrait for App { let self_clone = self.clone(); let recv_persister = ReceiverPersister::from_id(self.db.clone(), session_id.clone()); match replay_receiver_event_log(&recv_persister) { - Ok((receiver_state, _)) => { + Ok((receiver_state, history)) => { + let receiver_endpoint = history.pj_uri().extras.endpoint(); + let v2_config = self_clone.config.v2()?; + let network = network::MailroomUrlResolver::new(v2_config); + let relay_selector = match relay_selection::relay_selector_from_endpoint( + v2_config, + &receiver_endpoint, + &network, + ) { + Ok(relay_selector) => relay_selector, + Err(error) => { + tracing::error!( + "Failed to derive relay selector for receiver session {}: {:?}", + session_id, + error + ); + Self::close_failed_session( + &recv_persister, + &session_id, + Role::Receiver, + ); + continue; + } + }; tasks.push(( (Role::Receiver, session_id), tokio::spawn(async move { self_clone - .process_receiver_session(receiver_state, &recv_persister) + .process_receiver_session( + receiver_state, + &recv_persister, + &relay_selector, + ) .await }), )); @@ -837,10 +869,26 @@ impl App { ) -> Result<()> { loop { session = match session { - SendSession::WithReplyKey(context) => - self.post_original_proposal(context, persister).await?, - SendSession::PollingForProposal(context) => - self.get_proposed_payjoin_psbt(context, persister).await?, + SendSession::WithReplyKey(context) => { + let v2_config = self.config.v2()?; + let network = network::MailroomUrlResolver::new(v2_config); + let relay_selector = relay_selection::relay_selector_from_endpoint( + v2_config, + &context.endpoint(), + &network, + )?; + self.post_original_proposal(context, persister, &relay_selector).await? + } + SendSession::PollingForProposal(context) => { + let v2_config = self.config.v2()?; + let network = network::MailroomUrlResolver::new(v2_config); + let relay_selector = relay_selection::relay_selector_from_endpoint( + v2_config, + &context.endpoint(), + &network, + )?; + self.get_proposed_payjoin_psbt(context, persister, &relay_selector).await? + } SendSession::Closed(SenderSessionOutcome::Success(proposal)) => { let txid = self.process_pj_response(proposal)?; persister.print(format_args!("Payjoin sent. TXID: {txid}")); @@ -862,15 +910,20 @@ impl App { &self, sender: Sender, persister: &SenderPersister, + relay_selector: &RelaySelector, ) -> Result { - let (response, ctx) = - match self.post_via_relay(|relay| sender.create_v2_post_request(relay)).await? { - RelayPost::Posted(resp, ctx) => (resp, ctx), - RelayPost::Expired => { - self.cancel_sender_session(persister.session_id(), true)?; - return Ok(SendSession::Closed(SenderSessionOutcome::Aborted)); - } - }; + let (response, ctx) = match self + .send_ohttp_request_with_relay_selector(relay_selector, RequestKind::Post, |relay| { + sender.create_v2_post_request(relay.as_str()) + }) + .await? + { + RelayPost::Posted(resp, ctx) => (resp, ctx), + RelayPost::Expired => { + self.cancel_sender_session(persister.session_id(), true)?; + return Ok(SendSession::Closed(SenderSessionOutcome::Aborted)); + } + }; match sender.process_response(&response.bytes().await?, ctx).save(persister) { Ok(sender) => { persister.print("Posted Original PSBT..."); @@ -890,15 +943,20 @@ impl App { &self, sender: Sender, persister: &SenderPersister, + relay_selector: &RelaySelector, ) -> Result { - let (response, ctx) = - match self.post_via_relay(|relay| sender.create_poll_request(relay)).await? { - RelayPost::Posted(resp, ctx) => (resp, ctx), - RelayPost::Expired => { - self.cancel_sender_session(persister.session_id(), true)?; - return Ok(SendSession::Closed(SenderSessionOutcome::Aborted)); - } - }; + let (response, ctx) = match self + .send_ohttp_request_with_relay_selector(relay_selector, RequestKind::Poll, |relay| { + sender.create_poll_request(relay.as_str()) + }) + .await? + { + RelayPost::Posted(resp, ctx) => (resp, ctx), + RelayPost::Expired => { + self.cancel_sender_session(persister.session_id(), true)?; + return Ok(SendSession::Closed(SenderSessionOutcome::Aborted)); + } + }; let res = sender.clone().process_response(&response.bytes().await?, ctx).save(persister); match res { Ok(OptionalTransitionOutcome::Progress(psbt)) => { @@ -931,11 +989,12 @@ impl App { &self, mut session: ReceiveSession, persister: &ReceiverPersister, + relay_selector: &RelaySelector, ) -> Result<()> { loop { session = match session { ReceiveSession::Initialized(proposal) => - self.read_from_directory(proposal, persister).await?, + self.read_from_directory(proposal, persister, relay_selector).await?, ReceiveSession::UncheckedOriginalPayload(proposal) => self.check_proposal(proposal, persister)?, ReceiveSession::MaybeInputsOwned(proposal) => @@ -953,9 +1012,9 @@ impl App { ReceiveSession::ProvisionalProposal(proposal) => self.finalize_proposal(proposal, persister)?, ReceiveSession::PayjoinProposal(proposal) => - self.send_payjoin_proposal(proposal, persister).await?, + self.send_payjoin_proposal(proposal, persister, relay_selector).await?, ReceiveSession::HasReplyableError(error) => - self.handle_error(error, persister).await?, + self.handle_error(error, persister, relay_selector).await?, ReceiveSession::Monitor(proposal) => { self.monitor_payjoin_proposal(proposal, persister).await?; return Ok(()); @@ -977,16 +1036,21 @@ impl App { &self, session: Receiver, persister: &ReceiverPersister, + relay_selector: &RelaySelector, ) -> Result { persister.print("Polling receive request..."); - let (ohttp_response, context) = - match self.post_via_relay(|relay| session.create_poll_request(relay)).await? { - RelayPost::Posted(resp, ctx) => (resp, ctx), - RelayPost::Expired => { - self.cancel_receiver_session(persister.session_id(), true)?; - return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); - } - }; + let (ohttp_response, context) = match self + .send_ohttp_request_with_relay_selector(relay_selector, RequestKind::Poll, |relay| { + session.create_poll_request(relay.as_str()) + }) + .await? + { + RelayPost::Posted(resp, ctx) => (resp, ctx), + RelayPost::Expired => { + self.cancel_receiver_session(persister.session_id(), true)?; + return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); + } + }; let state_transition = session .process_response(ohttp_response.bytes().await?.to_vec().as_slice(), context) .save(persister); @@ -1132,15 +1196,20 @@ impl App { &self, proposal: Receiver, persister: &ReceiverPersister, + relay_selector: &RelaySelector, ) -> Result { - let (res, ohttp_ctx) = - match self.post_via_relay(|relay| proposal.create_post_request(relay)).await? { - RelayPost::Posted(resp, ctx) => (resp, ctx), - RelayPost::Expired => { - self.cancel_receiver_session(persister.session_id(), true)?; - return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); - } - }; + let (res, ohttp_ctx) = match self + .send_ohttp_request_with_relay_selector(relay_selector, RequestKind::Post, |relay| { + proposal.create_post_request(relay.as_str()) + }) + .await? + { + RelayPost::Posted(resp, ctx) => (resp, ctx), + RelayPost::Expired => { + self.cancel_receiver_session(persister.session_id(), true)?; + return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); + } + }; let payjoin_psbt = proposal.psbt().clone(); match proposal.process_response(&res.bytes().await?, ohttp_ctx).save(persister) { Ok(session) => { @@ -1222,15 +1291,20 @@ impl App { &self, session: Receiver, persister: &ReceiverPersister, + relay_selector: &RelaySelector, ) -> Result { - let (err_response, err_ctx) = - match self.post_via_relay(|relay| session.create_error_request(relay)).await? { - RelayPost::Posted(resp, ctx) => (resp, ctx), - RelayPost::Expired => { - self.cancel_receiver_session(persister.session_id(), true)?; - return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); - } - }; + let (err_response, err_ctx) = match self + .send_ohttp_request_with_relay_selector(relay_selector, RequestKind::Post, |relay| { + session.create_error_request(relay.as_str()) + }) + .await? + { + RelayPost::Posted(resp, ctx) => (resp, ctx), + RelayPost::Expired => { + self.cancel_receiver_session(persister.session_id(), true)?; + return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); + } + }; let err_bytes = match err_response.bytes().await { Ok(bytes) => bytes, Err(e) => return Err(anyhow!("Failed to get error response bytes: {}", e)), @@ -1269,36 +1343,72 @@ impl App { } } - async fn post_request(&self, req: payjoin::Request) -> Result { - let http = http_agent(&self.config)?; - http.post(req.url) - .header("Content-Type", req.content_type) - .body(req.body) - .send() - .await - .and_then(|r| r.error_for_status()) - .context("HTTP request failed") - } - - async fn post_via_relay(&self, mut build: F) -> Result> + async fn send_ohttp_request_with_relay_selector( + &self, + relay_selector: &RelaySelector, + request_kind: RequestKind, + mut build: F, + ) -> Result> where - F: FnMut(&str) -> std::result::Result<(payjoin::Request, T), E>, + F: FnMut(&payjoin::Url) -> std::result::Result<(payjoin::Request, T), E>, E: RequestExpiry + Into, { - loop { - let relay = self.mailroom_manager.choose_relay()?; - let (req, ctx) = match build(relay.as_str()) { + let request_label = match request_kind { + RequestKind::Post => "POST", + RequestKind::Poll => "POLL", + }; + let mut relays = + relay_selector.select_relays_for_request(request_kind)?.into_iter().peekable(); + + while let Some(relay) = relays.next() { + let is_last_relay = relays.peek().is_none(); + let (req, ctx) = match build(&relay.url) { Ok(r) => r, Err(e) if e.expired() => return Ok(RelayPost::Expired), Err(e) => return Err(e.into()), }; - match self.post_request(req).await { + match self.post_request(req, &relay).await { Ok(resp) => return Ok(RelayPost::Posted(resp, ctx)), - Err(e) => { - tracing::debug!("Request to relay {relay} failed: {e:?}"); - self.mailroom_manager.add_failed_relay(relay); + Err(RelayAttemptError::Retryable(error)) => { + tracing::debug!( + "Retryable OHTTP {request_label} failure via relay {}: {error:?}", + relay.url + ); + if is_last_relay { + return Err(error); + } } + Err(RelayAttemptError::Terminal(error)) => return Err(error), } } + + Err(anyhow!("No valid relays available")) + } + + async fn post_request( + &self, + req: payjoin::Request, + relay: &network::ResolvedUrl, + ) -> std::result::Result { + let mut builder = http_client_builder(&self.config).map_err(|err| { + RelayAttemptError::Terminal(anyhow!("Failed to build HTTP client: {err}")) + })?; + if let Some(domain) = relay.url.domain() { + builder = builder.resolve_to_addrs(domain, &relay.socket_addrs); + } + let http = builder.build().map_err(|err| { + RelayAttemptError::Terminal(anyhow!("Failed to build HTTP client: {err}")) + })?; + let response = http + .post(req.url) + .header("Content-Type", req.content_type) + .body(req.body) + .send() + .await + .map_err(|err| classify_reqwest_error(err, "HTTP request failed"))?; + + response + .error_for_status() + .map_err(|err| classify_reqwest_error(err, "HTTP request failed")) } } diff --git a/payjoin-cli/src/app/v2/ohttp.rs b/payjoin-cli/src/app/v2/ohttp.rs index e438540ec..7e552e9a6 100644 --- a/payjoin-cli/src/app/v2/ohttp.rs +++ b/payjoin-cli/src/app/v2/ohttp.rs @@ -1,138 +1,176 @@ -//! OHTTP relay and payjoin directory selection / key bootstrapping for the payjoin-cli. +//! OHTTP relay selection and key bootstrapping for the payjoin-cli. //! -//! [`MailroomManager`] tracks relays and directories that have failed, -//! excluding them from future selections for the lifetime of the [`MailroomManager`]. -//! -//! `unwrap_ohttp_keys_or_else_fetch_from_directory` returns user-supplied keys -//! when present, otherwise selects a relay at random from the configured list -//! (excluding failed relays) to fetch OHTTP keys from the given directory. -//! -//! `fetch_ohttp_keys_from_directory` retries on relay failures (e.g. connection -//! errors) by selecting another relay. Once a directory is chosen for a session -//! it must not change — the directory is embedded in the BIP21 URI at session -//! creation and recovered from the session event log on resume. +//! Bootstrap key fetching uses temporary relay failover. Protocol requests use +//! stateless relay selection from the receiver network selection. use std::sync::{Arc, Mutex}; use anyhow::{anyhow, Result}; use payjoin::Url; +use super::network::{ResolvedUrl, UrlResolver}; +use super::relay_selection::{choose_receiver_network_selection, ReceiverNetworkSelection}; use super::Config; +/// Coordinates receiver bootstrap across configured directories. +/// +/// Relay ordering for protocol requests remains stateless. This manager only +/// remembers directories that failed while this application instance is alive. #[derive(Debug, Clone)] -pub struct MailroomManager { +pub(crate) struct MailroomManager { config: Config, - failed_relays: Arc>>, failed_directories: Arc>>, } impl MailroomManager { - pub fn new(config: Config) -> Self { - MailroomManager { - config, - failed_relays: Arc::new(Mutex::new(Vec::new())), - failed_directories: Arc::new(Mutex::new(Vec::new())), - } - } - - pub fn add_failed_relay(&self, relay: Url) { - self.failed_relays.lock().expect("Lock should not be poisoned").push(relay); - } - - pub fn clear_failed_relays(&self) { - self.failed_relays.lock().expect("Lock should not be poisoned").clear(); - } - - pub fn add_failed_directory(&self, directory: Url) { - self.failed_directories.lock().expect("Lock should not be poisoned").push(directory); + pub(crate) fn new(config: Config) -> Self { + Self { config, failed_directories: Arc::new(Mutex::new(Vec::new())) } } - pub fn choose_relay(&self) -> Result { - use payjoin::bitcoin::secp256k1::rand::prelude::SliceRandom; - let relays = &self.config.v2()?.ohttp_relays; - let failed_relays = self.failed_relays.lock().expect("Lock should not be poisoned"); - let remaining_relays: Vec<_> = - relays.iter().filter(|r| !failed_relays.contains(r)).cloned().collect(); - - if remaining_relays.is_empty() { - return Err(anyhow!("No valid relays available")); + pub(crate) async fn bootstrap_receiver( + &self, + network: &impl UrlResolver, + ) -> Result<(ReceiverNetworkSelection, payjoin::OhttpKeys)> { + loop { + let failed_directories = + self.failed_directories.lock().expect("Lock should not be poisoned").clone(); + let network_selection = + choose_receiver_network_selection(self.config.v2()?, network, &failed_directories)?; + + match unwrap_ohttp_keys_or_else_fetch(&self.config, &network_selection).await { + Ok(ohttp_keys) => return Ok((network_selection, ohttp_keys)), + Err(error) => { + tracing::debug!( + "Directory {} failed: {error:#}", + network_selection.directory.url + ); + self.failed_directories + .lock() + .expect("Lock should not be poisoned") + .push(network_selection.directory.url); + } + } } - - remaining_relays - .choose(&mut payjoin::bitcoin::key::rand::thread_rng()) - .cloned() - .ok_or_else(|| anyhow!("Failed to select from remaining relays")) } +} - pub fn choose_directory(&self) -> Result { - use payjoin::bitcoin::secp256k1::rand::prelude::SliceRandom; - let directories = &self.config.v2()?.pj_directories; - let failed_directories = - self.failed_directories.lock().expect("Lock should not be poisoned"); - let remaining_directories: Vec<_> = - directories.iter().filter(|d| !failed_directories.contains(d)).cloned().collect(); - - if remaining_directories.is_empty() { - return Err(anyhow!("No valid directories available")); - } +#[derive(Debug)] +pub(crate) enum RelayAttemptError { + /// Network-shaped failures can try the next relay candidate. + Retryable(anyhow::Error), + /// Protocol/configuration-shaped failures should stop immediately. + Terminal(anyhow::Error), +} - remaining_directories - .choose(&mut payjoin::bitcoin::key::rand::thread_rng()) - .cloned() - .ok_or_else(|| anyhow!("Failed to select from remaining directories")) +/// Decide whether a reqwest failure should fail over to another relay. +pub(crate) fn classify_reqwest_error( + err: reqwest::Error, + context: &'static str, +) -> RelayAttemptError { + let error = anyhow!("{context}: {err}"); + if err.is_timeout() || err.is_connect() || err.is_request() { + RelayAttemptError::Retryable(error) + } else { + RelayAttemptError::Terminal(error) } +} - pub(crate) async fn unwrap_ohttp_keys_or_else_fetch_from_directory( - &self, - directory: &Url, - ) -> Result { - if let Some(ohttp_keys) = self.config.v2()?.ohttp_keys.clone() { - return Ok(ValidatedOhttpKeys { ohttp_keys }); - } - self.fetch_ohttp_keys_from_directory(directory).await +pub(crate) async fn unwrap_ohttp_keys_or_else_fetch( + config: &Config, + network_selection: &ReceiverNetworkSelection, +) -> Result { + if let Some(ohttp_keys) = config.v2()?.ohttp_keys.clone() { + println!("Using OHTTP Keys from config"); + Ok(ohttp_keys) + } else { + println!("Bootstrapping private network transport over Oblivious HTTP"); + fetch_ohttp_keys(config, network_selection).await } +} - async fn fetch_ohttp_keys_from_directory(&self, directory: &Url) -> Result { - loop { - let selected_relay = self.choose_relay()?; - - let ohttp_keys = { - #[cfg(feature = "_manual-tls")] - { - if let Some(cert_path) = self.config.root_certificate.as_ref() { - let cert_der = std::fs::read(cert_path)?; - payjoin::io::fetch_ohttp_keys_with_cert( - selected_relay.as_str(), - directory.as_str(), - &cert_der, - ) - .await - } else { - payjoin::io::fetch_ohttp_keys(selected_relay.as_str(), directory.as_str()) - .await - } - } - #[cfg(not(feature = "_manual-tls"))] - payjoin::io::fetch_ohttp_keys(selected_relay.as_str(), directory.as_str()).await - }; +// Fetch directory OHTTP keys through the already chosen receiver network selection. +// This happens before the receiver key exists, so it cannot use RelaySelector. +async fn fetch_ohttp_keys( + config: &Config, + network_selection: &ReceiverNetworkSelection, +) -> Result { + if network_selection.relays.is_empty() { + return Err(anyhow!( + "No valid relays available for {}", + network_selection.directory.url.as_str() + )); + } - match ohttp_keys { - Ok(keys) => return Ok(ValidatedOhttpKeys { ohttp_keys: keys }), - Err(payjoin::io::Error::UnexpectedStatusCode(e)) => { - tracing::debug!( - "Directory {directory} returned unexpected status via relay {selected_relay}: {e:?}" - ); - self.add_failed_directory(directory.clone()); - return Err(anyhow!("Directory {directory} returned unexpected status: {e}")); - } - Err(e) => { - tracing::debug!("Failed to connect to relay: {selected_relay}, {e:?}"); - self.add_failed_relay(selected_relay); + let last_relay_index = network_selection.relays.len() - 1; + for (index, relay) in network_selection.relays.iter().enumerate() { + match fetch_directory_ohttp_keys_via_resolved_relay_url( + config, + relay.relay(), + &network_selection.directory, + ) + .await + { + Ok(keys) => return Ok(keys), + Err(RelayAttemptError::Retryable(error)) => { + tracing::debug!( + "Failed to fetch OHTTP keys via relay {}: {error:?}", + relay.relay().url + ); + if index == last_relay_index { + return Err(error); } } + Err(RelayAttemptError::Terminal(error)) => return Err(error), } } + + unreachable!( + "empty relay selections return before the loop and successful key fetches return inside it" + ) } -pub(crate) struct ValidatedOhttpKeys { - pub(crate) ohttp_keys: payjoin::OhttpKeys, +// Fetch through one relay using addresses checked by relay_selection. The +// library owns the request and decoding while the CLI owns relay failover. +async fn fetch_directory_ohttp_keys_via_resolved_relay_url( + _config: &Config, + relay: &ResolvedUrl, + directory: &ResolvedUrl, +) -> std::result::Result { + #[cfg(feature = "_manual-tls")] + let result = if let Some(cert_path) = _config.root_certificate.as_ref() { + let cert_der = std::fs::read(cert_path).map_err(|error| { + RelayAttemptError::Terminal(anyhow!("Failed to read root certificate: {error}")) + })?; + payjoin::io::fetch_ohttp_keys_with_cert_and_relay_addresses( + relay.url.as_str(), + directory.url.as_str(), + &cert_der, + &relay.socket_addrs, + ) + .await + } else { + payjoin::io::fetch_ohttp_keys_with_relay_addresses( + relay.url.as_str(), + directory.url.as_str(), + &relay.socket_addrs, + ) + .await + }; + + #[cfg(not(feature = "_manual-tls"))] + let result = payjoin::io::fetch_ohttp_keys_with_relay_addresses( + relay.url.as_str(), + directory.url.as_str(), + &relay.socket_addrs, + ) + .await; + + result.map_err(|error| { + let retryable = error.is_retryable(); + let error = anyhow!("Failed to fetch OHTTP keys: {error}"); + if retryable { + RelayAttemptError::Retryable(error) + } else { + RelayAttemptError::Terminal(error) + } + }) } diff --git a/payjoin-cli/tests/e2e.rs b/payjoin-cli/tests/e2e.rs index 9cebbb9b8..0ce12489d 100644 --- a/payjoin-cli/tests/e2e.rs +++ b/payjoin-cli/tests/e2e.rs @@ -323,6 +323,8 @@ mod e2e { .arg(&sender_db_path) .arg("--ohttp-relays") .arg(ohttp_relays) + .arg("--pj-directories") + .arg(directory) .arg("send") .arg(&bip21) .arg("--fee-rate") @@ -344,6 +346,8 @@ mod e2e { .arg(&receiver_db_path) .arg("--ohttp-relays") .arg(ohttp_relays) + .arg("--pj-directories") + .arg(directory) .arg("resume") .stdout(Stdio::piped()) .stderr(Stdio::inherit()) @@ -362,6 +366,8 @@ mod e2e { .arg(&receiver_db_path) .arg("--ohttp-relays") .arg(ohttp_relays) + .arg("--pj-directories") + .arg(directory) .arg("resume") .stdout(Stdio::piped()) .stderr(Stdio::inherit()) @@ -380,6 +386,8 @@ mod e2e { .arg(&sender_db_path) .arg("--ohttp-relays") .arg(ohttp_relays) + .arg("--pj-directories") + .arg(directory) .arg("send") .arg(&bip21) .arg("--fee-rate") @@ -411,6 +419,8 @@ mod e2e { .arg(&receiver_db_path) .arg("--ohttp-relays") .arg(ohttp_relays) + .arg("--pj-directories") + .arg(directory) .arg("resume") .stdout(Stdio::piped()) .stderr(Stdio::inherit()) @@ -430,6 +440,8 @@ mod e2e { .arg(&receiver_db_path) .arg("--ohttp-relays") .arg(ohttp_relays) + .arg("--pj-directories") + .arg(directory) .arg("resume") .stdout(Stdio::piped()) .stderr(Stdio::inherit()) @@ -447,6 +459,8 @@ mod e2e { .arg(&sender_db_path) .arg("--ohttp-relays") .arg(ohttp_relays) + .arg("--pj-directories") + .arg(directory) .arg("resume") .stdout(Stdio::piped()) .stderr(Stdio::inherit()) @@ -765,6 +779,8 @@ mod e2e { .arg(&sender_db_path) .arg("--ohttp-relays") .arg(ohttp_relays) + .arg("--pj-directories") + .arg(directory) .arg("send") .arg(&bip21) .arg("--fee-rate") @@ -1089,6 +1105,8 @@ mod e2e { .arg(&sender_db_path) .arg("--ohttp-relays") .arg(ohttp_relays) + .arg("--pj-directories") + .arg(directory) .arg("send") .arg(&bip21) .arg("--fee-rate") From acf609f7b66bc209177777d17856c61d668d8240 Mon Sep 17 00:00:00 2001 From: Mshehu5 Date: Tue, 14 Jul 2026 14:32:27 +0100 Subject: [PATCH 6/6] Document AS-aware relay selection Document AS-aware v2 relay selection configuration. Describe how to configure trusted directories, OHTTP relays, ASMap input, and user ASN information so downstream implementations can understand the reference CLI behavior. --- payjoin-cli/README.md | 24 ++++++++++++++++++++++++ payjoin-cli/example.config.toml | 10 ++++++++++ 2 files changed, 34 insertions(+) diff --git a/payjoin-cli/README.md b/payjoin-cli/README.md index 81e91ce77..05ca9aa9b 100644 --- a/payjoin-cli/README.md +++ b/payjoin-cli/README.md @@ -140,6 +140,30 @@ See the [example.config.toml](https://github.com/payjoin/rust-payjoin/blob/fde867b93ede767c9a50913432a73782a94ef40b/payjoin-cli/example.config.toml) for inspiration. +`payjoin-cli` also supports optional AS-aware filtering for BIP77 relay +selection: + +```toml +[v2] +pj_directories = ["https://payjo.in", "https://backup.example"] +ohttp_relays = ["https://relay-1.example", "https://relay-2.example"] + +[v2.asmap] +asmap_file = "./ip_asn.dat" +user_public_ips = ["198.51.100.10"] +user_asns = [64512] +``` + +Build `payjoin-cli` with `--features asmap` to enable the `[v2.asmap]` +configuration block. + +When enabled, directories and relays that resolve into the same ASN as the +configured user identity are excluded, mixed-ASN hostnames are rejected, and +relay ordering becomes deterministic from the receiver key embedded in the +BIP77 URI. This mitigates some AS-level correlation risks, but it does not +eliminate traffic analysis when sender and receiver already share the same +network. + ### Asynchronous Operation Sender and receiver state is saved to a database in the directory from which `payjoin-cli` is run, called `payjoin.sqlite`. Once a send or receive session is started, it may resume using the `resume` argument if prior payjoin sessions have not yet complete. diff --git a/payjoin-cli/example.config.toml b/payjoin-cli/example.config.toml index af1d7f094..122d71c5d 100644 --- a/payjoin-cli/example.config.toml +++ b/payjoin-cli/example.config.toml @@ -54,3 +54,13 @@ rpcpassword = "password" # # for the payjoin packets to be encrypted. # # These can now be fetched and no longer need to be configured. # ohttp_keys = "./path/to/ohttp_keys" +# +# # Optional AS-aware relay and directory filtering. +# # When enabled, payjoin-cli will reject directories/relays that share an +# # ASN with the configured user identity, and will deterministically order +# # relay selection from the remaining candidates. +# # Requires building payjoin-cli with `--features asmap`. +# [v2.asmap] +# asmap_file = "./ip_asn.dat" +# user_public_ips = ["198.51.100.10", "2001:db8::10"] +# user_asns = [64512]