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/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] 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)) + } } } 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-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/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/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/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/"); + } +} 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") 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); 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); + } +}