15052840071

Committed 15 May 2025 06:47PM UTC coverage: 82.13% (-0.1%) from 82.247%

Build # 15052840071

Build Type

Pull #607

github

Committed by

web-flow

Commit Message

Merge e7e0cd617 into e2b66a2b1

Pull Request Pull Request #607: Create fallback ohttp-relay logic for payjoin-cli

Coverage Stats

44 of 79 new or added lines in 4 files covered. (55.7%)

1 existing line in 1 file now uncovered.

5506 of 6704 relevant lines covered (82.13%)

690.3 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.18

/payjoin-cli/src/app/v2.rs

use std::sync::{Arc, Mutex};

use anyhow::{anyhow, Context, Result};
use payjoin::bitcoin::consensus::encode::serialize_hex;
use payjoin::bitcoin::psbt::Psbt;
use payjoin::bitcoin::{Amount, FeeRate};
use payjoin::receive::v2::{NewReceiver, Receiver, UncheckedProposal};
use payjoin::receive::{Error, ReplyableError};
use payjoin::send::v2::{Sender, SenderBuilder};
use payjoin::{ImplementationError, Uri};
use tokio::sync::watch;

use super::config::Config;
use super::wallet::BitcoindWallet;
use super::App as AppTrait;
use crate::app::{handle_interrupt, http_agent};
use crate::db::v2::{ReceiverPersister, SenderPersister};
use crate::db::Database;

#[derive(Debug, Clone)]
pub struct RelayState {
    selected_relay: Option<payjoin::Url>,
    #[cfg(not(feature = "_danger-local-https"))]
    failed_relays: Vec<payjoin::Url>,
}

impl RelayState {
    #[cfg(feature = "_danger-local-https")]
    pub fn new() -> Self { RelayState { selected_relay: None } }
    #[cfg(not(feature = "_danger-local-https"))]
    pub fn new() -> Self { RelayState { selected_relay: None, failed_relays: Vec::new() } }

    #[cfg(not(feature = "_danger-local-https"))]
    pub fn set_selected_relay(&mut self, relay: payjoin::Url) { self.selected_relay = Some(relay); }

    pub fn get_selected_relay(&self) -> Option<payjoin::Url> { self.selected_relay.clone() }

    #[cfg(not(feature = "_danger-local-https"))]
    pub fn add_failed_relay(&mut self, relay: payjoin::Url) { self.failed_relays.push(relay); }

    #[cfg(not(feature = "_danger-local-https"))]
    pub fn get_failed_relays(&self) -> Vec<payjoin::Url> { self.failed_relays.clone() }
}

#[derive(Clone)]
pub(crate) struct App {
    config: Config,
    db: Arc<Database>,
    wallet: BitcoindWallet,
    interrupt: watch::Receiver<()>,
    relay_state: Arc<Mutex<RelayState>>,
}

#[async_trait::async_trait]
impl AppTrait for App {
    fn new(config: Config) -> Result<Self> {
        let db = Arc::new(Database::create(&config.db_path)?);
        let relay_state = Arc::new(Mutex::new(RelayState::new()));
        let (interrupt_tx, interrupt_rx) = watch::channel(());
        tokio::spawn(handle_interrupt(interrupt_tx));
        let wallet = BitcoindWallet::new(&config.bitcoind)?;
        let app = Self { config, db, wallet, interrupt: interrupt_rx, relay_state };
        app.wallet()
            .network()
            .context("Failed to connect to bitcoind. Check config RPC connection.")?;
        Ok(app)
    }

    fn wallet(&self) -> BitcoindWallet { self.wallet.clone() }

    async fn send_payjoin(&self, bip21: &str, fee_rate: FeeRate) -> Result<()> {
        use payjoin::UriExt;
        let uri =
            Uri::try_from(bip21).map_err(|e| anyhow!("Failed to create URI from BIP21: {}", e))?;
        let uri = uri.assume_checked();
        let uri = uri.check_pj_supported().map_err(|_| anyhow!("URI does not support Payjoin"))?;
        let url = uri.extras.endpoint();
        // match bip21 to send_session public_key
        let req_ctx = match self.db.get_send_session(url)? {
            Some(send_session) => send_session,
            None => {
                let psbt = self.create_original_psbt(&uri, fee_rate)?;
                let mut persister = SenderPersister::new(self.db.clone());
                let new_sender = SenderBuilder::new(psbt, uri.clone())
                    .build_recommended(fee_rate)
                    .with_context(|| "Failed to build payjoin request")?;
                let storage_token = new_sender
                    .persist(&mut persister)
                    .map_err(|e| anyhow!("Failed to persist sender: {}", e))?;
                Sender::load(storage_token, &persister)
                    .map_err(|e| anyhow!("Failed to load sender: {}", e))?
            }
        };
        self.spawn_payjoin_sender(req_ctx).await
    }

    async fn receive_payjoin(&self, amount: Amount) -> Result<()> {
        let address = self.wallet().get_new_address()?;
        let ohttp_keys =
            unwrap_ohttp_keys_or_else_fetch(&self.config, self.relay_state.clone()).await?;
        let mut persister = ReceiverPersister::new(self.db.clone());
        let new_receiver = NewReceiver::new(
            address,
            self.config.v2()?.pj_directory.clone(),
            ohttp_keys.clone(),
            None,
        )?;
        let storage_token = new_receiver
            .persist(&mut persister)
            .map_err(|e| anyhow!("Failed to persist receiver: {}", e))?;
        let session = Receiver::load(storage_token, &persister)
            .map_err(|e| anyhow!("Failed to load receiver: {}", e))?;
        self.spawn_payjoin_receiver(session, Some(amount)).await
    }

    #[allow(clippy::incompatible_msrv)]
    async fn resume_payjoins(&self) -> Result<()> {
        let recv_sessions = self.db.get_recv_sessions()?;
        let send_sessions = self.db.get_send_sessions()?;

        if recv_sessions.is_empty() && send_sessions.is_empty() {
            println!("No sessions to resume.");
            return Ok(());
        }

        let mut tasks = Vec::new();

        for session in recv_sessions {
            let self_clone = self.clone();
            tasks.push(tokio::spawn(async move {
                self_clone.spawn_payjoin_receiver(session, None).await
            }));
        }

        for session in send_sessions {
            let self_clone = self.clone();
            tasks.push(tokio::spawn(async move { self_clone.spawn_payjoin_sender(session).await }));
        }

        let mut interrupt = self.interrupt.clone();
        tokio::select! {
            _ = async {
                for task in tasks {
                    let _ = task.await;
                }
            } => {
                println!("All resumed sessions completed.");
            }
            _ = interrupt.changed() => {
                println!("Resumed sessions were interrupted.");
            }
        }
        Ok(())
    }
}

impl App {
    #[allow(clippy::incompatible_msrv)]
    async fn spawn_payjoin_sender(&self, mut req_ctx: Sender) -> Result<()> {
        let mut interrupt = self.interrupt.clone();
        tokio::select! {
            res = self.long_poll_post(&mut req_ctx) => {
                self.process_pj_response(res?)?;
                self.db.clear_send_session(req_ctx.endpoint())?;
            }
            _ = interrupt.changed() => {
                println!("Interrupted. Call `send` with the same arguments to resume this session or `resume` to resume all sessions.");
            }
        }
        Ok(())
    }

    #[allow(clippy::incompatible_msrv)]
    async fn spawn_payjoin_receiver(
        &self,
        mut session: Receiver,
        amount: Option<Amount>,
    ) -> Result<()> {
        println!("Receive session established");
        let mut pj_uri = session.pj_uri();
        pj_uri.amount = amount;
        let ohttp_relay = self.unwrap_relay_or_else_fetch().await?;

        println!("Request Payjoin by sharing this Payjoin Uri:");
        println!("{pj_uri}");

        let mut interrupt = self.interrupt.clone();
        let receiver = tokio::select! {
            res = self.long_poll_fallback(&mut session) => res,
            _ = interrupt.changed() => {
                println!("Interrupted. Call the `resume` command to resume all sessions.");
                return Ok(());
            }
        }?;

        println!("Fallback transaction received. Consider broadcasting this to get paid if the Payjoin fails:");
        println!("{}", serialize_hex(&receiver.extract_tx_to_schedule_broadcast()));
        let mut payjoin_proposal = match self.process_v2_proposal(receiver.clone()) {
            Ok(proposal) => proposal,
            Err(Error::ReplyToSender(e)) => {
                return Err(handle_recoverable_error(e, receiver, &ohttp_relay).await);
            }
            Err(e) => return Err(e.into()),
        };
        let (req, ohttp_ctx) = payjoin_proposal
            .extract_req(ohttp_relay)
            .map_err(|e| anyhow!("v2 req extraction failed {}", e))?;
        println!("Got a request from the sender. Responding with a Payjoin proposal.");
        let res = post_request(req).await?;
        payjoin_proposal
            .process_res(&res.bytes().await?, ohttp_ctx)
            .map_err(|e| anyhow!("Failed to deserialize response {}", e))?;
        let payjoin_psbt = payjoin_proposal.psbt().clone();
        println!(
            "Response successful. Watch mempool for successful Payjoin. TXID: {}",
            payjoin_psbt.extract_tx_unchecked_fee_rate().clone().compute_txid()
        );
        self.db.clear_recv_session()?;
        Ok(())
    }

    async fn long_poll_post(&self, req_ctx: &mut Sender) -> Result<Psbt> {
        let ohttp_relay = self.unwrap_relay_or_else_fetch().await?;

        match req_ctx.extract_v2(ohttp_relay.clone()) {
            Ok((req, ctx)) => {
                println!("Posting Original PSBT Payload request...");
                let response = post_request(req).await?;
                println!("Sent fallback transaction");
                let v2_ctx = Arc::new(ctx.process_response(&response.bytes().await?)?);
                loop {
                    let (req, ohttp_ctx) = v2_ctx.extract_req(&ohttp_relay)?;
                    let response = post_request(req).await?;
                    match v2_ctx.process_response(&response.bytes().await?, ohttp_ctx) {
                        Ok(Some(psbt)) => return Ok(psbt),
                        Ok(None) => {
                            println!("No response yet.");
                        }
                        Err(re) => {
                            println!("{re}");
                            log::debug!("{re:?}");
                            return Err(anyhow!("Response error").context(re));
                        }
                    }
                }
            }
            Err(_) => {
                let (req, v1_ctx) = req_ctx.extract_v1();
                println!("Posting Original PSBT Payload request...");
                let response = post_request(req).await?;
                println!("Sent fallback transaction");
                match v1_ctx.process_response(&mut response.bytes().await?.to_vec().as_slice()) {
                    Ok(psbt) => Ok(psbt),
                    Err(re) => {
                        println!("{re}");
                        log::debug!("{re:?}");
                        Err(anyhow!("Response error").context(re))
                    }
                }
            }
        }
    }

    async fn long_poll_fallback(
        &self,
        session: &mut payjoin::receive::v2::Receiver,
    ) -> Result<payjoin::receive::v2::UncheckedProposal> {
        let ohttp_relay = self.unwrap_relay_or_else_fetch().await?;

        loop {
            let (req, context) = session.extract_req(&ohttp_relay)?;
            println!("Polling receive request...");
            let ohttp_response = post_request(req).await?;
            let proposal = session
                .process_res(ohttp_response.bytes().await?.to_vec().as_slice(), context)
                .map_err(|_| anyhow!("GET fallback failed"))?;
            log::debug!("got response");
            if let Some(proposal) = proposal {
                break Ok(proposal);
            }
        }
    }

    fn process_v2_proposal(
        &self,
        proposal: payjoin::receive::v2::UncheckedProposal,
    ) -> Result<payjoin::receive::v2::PayjoinProposal, Error> {
        let wallet = self.wallet();

        // in a payment processor where the sender could go offline, this is where you schedule to broadcast the original_tx
        let _to_broadcast_in_failure_case = proposal.extract_tx_to_schedule_broadcast();

        // Receive Check 1: Can Broadcast
        let proposal =
            proposal.check_broadcast_suitability(None, |tx| Ok(wallet.can_broadcast(tx)?))?;
        log::trace!("check1");

        // Receive Check 2: receiver can't sign for proposal inputs
        let proposal = proposal.check_inputs_not_owned(|input| Ok(wallet.is_mine(input)?))?;
        log::trace!("check2");

        // Receive Check 3: have we seen this input before? More of a check for non-interactive i.e. payment processor receivers.
        let payjoin = proposal
            .check_no_inputs_seen_before(|input| Ok(self.db.insert_input_seen_before(*input)?))?;
        log::trace!("check3");

        let payjoin = payjoin
            .identify_receiver_outputs(|output_script| Ok(wallet.is_mine(output_script)?))?
            .commit_outputs();

        let provisional_payjoin = try_contributing_inputs(payjoin.clone(), &wallet)
            .map_err(ReplyableError::Implementation)?;

        let payjoin_proposal = provisional_payjoin.finalize_proposal(
            |psbt| Ok(wallet.process_psbt(psbt)?),
            None,
            self.config.max_fee_rate,
        )?;
        let payjoin_proposal_psbt = payjoin_proposal.psbt();
        log::debug!("Receiver's Payjoin proposal PSBT Rsponse: {payjoin_proposal_psbt:#?}");
        Ok(payjoin_proposal)
    }

    async fn unwrap_relay_or_else_fetch(&self) -> Result<payjoin::Url> {
        let selected_relay =
            self.relay_state.lock().expect("Lock should not be poisoned").get_selected_relay();
        let ohttp_relay = match selected_relay {
            Some(relay) => relay,
            None => validate_relay(&self.config, self.relay_state.clone()).await?,
        };
        Ok(ohttp_relay)
    }
}

/// Handle request error by sending an error response over the directory
async fn handle_recoverable_error(
    e: ReplyableError,
    mut receiver: UncheckedProposal,
    ohttp_relay: &payjoin::Url,
) -> anyhow::Error {
    let to_return = anyhow!("Replied with error: {}", e);
    let (err_req, err_ctx) = match receiver.extract_err_req(&e.into(), ohttp_relay) {
        Ok(req_ctx) => req_ctx,
        Err(e) => return anyhow!("Failed to extract error request: {}", e),
    };

    let err_response = match post_request(err_req).await {
        Ok(response) => response,
        Err(e) => return anyhow!("Failed to post error request: {}", e),
    };

    let err_bytes = match err_response.bytes().await {
        Ok(bytes) => bytes,
        Err(e) => return anyhow!("Failed to get error response bytes: {}", e),
    };

    if let Err(e) = receiver.process_err_res(&err_bytes, err_ctx) {
        return anyhow!("Failed to process error response: {}", e);
    }

    to_return
}

fn try_contributing_inputs(
    payjoin: payjoin::receive::v2::WantsInputs,
    wallet: &BitcoindWallet,
) -> Result<payjoin::receive::v2::ProvisionalProposal, ImplementationError> {
    let candidate_inputs = wallet.list_unspent()?;

    let selected_input =
        payjoin.try_preserving_privacy(candidate_inputs).map_err(ImplementationError::from)?;

    Ok(payjoin
        .contribute_inputs(vec![selected_input])
        .map_err(ImplementationError::from)?
        .commit_inputs())
}

async fn unwrap_ohttp_keys_or_else_fetch(
    config: &Config,
    relay_state: Arc<Mutex<RelayState>>,
) -> Result<payjoin::OhttpKeys> {
    if let Some(keys) = config.v2()?.ohttp_keys.clone() {
        println!("Using OHTTP Keys from config");
        Ok(keys)
    } else {
        println!("Bootstrapping private network transport over Oblivious HTTP");

        fetch_keys(config, relay_state.clone())
            .await
            .and_then(|keys| keys.ok_or_else(|| anyhow::anyhow!("No OHTTP keys found")))
    }
}

#[cfg(not(feature = "_danger-local-https"))]
async fn fetch_keys(
    config: &Config,
    relay_state: Arc<Mutex<RelayState>>,
) -> Result<Option<payjoin::OhttpKeys>> {
    use payjoin::bitcoin::secp256k1::rand::prelude::SliceRandom;
    let payjoin_directory = config.v2()?.pj_directory.clone();
    let relays = config.v2()?.ohttp_relays.clone();

    loop {
        let failed_relays =
            relay_state.lock().expect("Lock should not be poisoned").get_failed_relays();

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

        let selected_relay =
            match remaining_relays.choose(&mut payjoin::bitcoin::key::rand::thread_rng()) {
                Some(relay) => relay.clone(),
                None => return Err(anyhow!("Failed to select from remaining relays")),
            };

        relay_state
            .lock()
            .expect("Lock should not be poisoned")
            .set_selected_relay(selected_relay.clone());

        let ohttp_keys = {
            payjoin::io::fetch_ohttp_keys(selected_relay.clone(), payjoin_directory.clone()).await
        };

        match ohttp_keys {
            Ok(keys) => return Ok(Some(keys)),
            Err(payjoin::io::Error::UnexpectedStatusCode(e)) => {
                return Err(payjoin::io::Error::UnexpectedStatusCode(e).into());
            }
            Err(e) => {
                log::debug!("Failed to connect to relay: {selected_relay}, {e:?}");
                relay_state
                    .lock()
                    .expect("Lock should not be poisoned")
                    .add_failed_relay(selected_relay);
            }
        }
    }
}

//Local relays are incapable of acting as proxies so we must opportunistically fetch keys from the config
#[cfg(feature = "_danger-local-https")]
async fn fetch_keys(
    config: &Config,
    _relay_state: Arc<Mutex<RelayState>>,
) -> Result<Option<payjoin::OhttpKeys>> {
    let keys = config.v2()?.ohttp_keys.clone().expect("No OHTTP keys set");

    Ok(Some(keys))
}

#[cfg(not(feature = "_danger-local-https"))]
async fn validate_relay(
    config: &Config,
    relay_state: Arc<Mutex<RelayState>>,
) -> Result<payjoin::Url> {
    use payjoin::bitcoin::secp256k1::rand::prelude::SliceRandom;
    let payjoin_directory = config.v2()?.pj_directory.clone();
    let relays = config.v2()?.ohttp_relays.clone();

    loop {
        let failed_relays =
            relay_state.lock().expect("Lock should not be poisoned").get_failed_relays();

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

        let selected_relay =
            match remaining_relays.choose(&mut payjoin::bitcoin::key::rand::thread_rng()) {
                Some(relay) => relay.clone(),
                None => return Err(anyhow!("Failed to select from remaining relays")),
            };

        relay_state
            .lock()
            .expect("Lock should not be poisoned")
            .set_selected_relay(selected_relay.clone());

        let ohttp_keys =
            payjoin::io::fetch_ohttp_keys(selected_relay.clone(), payjoin_directory.clone()).await;

        match ohttp_keys {
            Ok(_) => return Ok(selected_relay),
            Err(payjoin::io::Error::UnexpectedStatusCode(e)) => {
                return Err(payjoin::io::Error::UnexpectedStatusCode(e).into());
            }
            Err(e) => {
                log::debug!("Failed to connect to relay: {selected_relay}, {e:?}");
                relay_state
                    .lock()
                    .expect("Lock should not be poisoned")
                    .add_failed_relay(selected_relay);
            }
        }
    }
}

#[cfg(feature = "_danger-local-https")]
async fn validate_relay(
    config: &Config,
    _relay_state: Arc<Mutex<RelayState>>,
) -> Result<payjoin::Url> {
    let relay = config.v2()?.ohttp_relays.first().expect("no OHTTP relay set").clone();

    Ok(relay)
}

async fn post_request(req: payjoin::Request) -> Result<reqwest::Response> {
    let http = http_agent()?;
    http.post(req.url)
        .header("Content-Type", req.content_type)
        .body(req.body)
        .send()
        .await
        .map_err(map_reqwest_err)
}

fn map_reqwest_err(e: reqwest::Error) -> anyhow::Error {
    match e.status() {
        Some(status_code) => anyhow!("HTTP request failed: {} {}", status_code, e),
        None => anyhow!("No HTTP response: {}", e),
    }
}

1	use std::sync::{Arc, Mutex};
2
3	use anyhow::{anyhow, Context, Result};
4	use payjoin::bitcoin::consensus::encode::serialize_hex;
5	use payjoin::bitcoin::psbt::Psbt;
6	use payjoin::bitcoin::{Amount, FeeRate};
7	use payjoin::receive::v2::{NewReceiver, Receiver, UncheckedProposal};
8	use payjoin::receive::{Error, ReplyableError};
9	use payjoin::send::v2::{Sender, SenderBuilder};
10	use payjoin::{ImplementationError, Uri};
11	use tokio::sync::watch;
12
13	use super::config::Config;
14	use super::wallet::BitcoindWallet;
15	use super::App as AppTrait;
16	use crate::app::{handle_interrupt, http_agent};
17	use crate::db::v2::{ReceiverPersister, SenderPersister};
18	use crate::db::Database;
19
20	#[derive(Debug, Clone)]
21	pub struct RelayState {
22	selected_relay: Option<payjoin::Url>,
23	#[cfg(not(feature = "_danger-local-https"))]
24	failed_relays: Vec<payjoin::Url>,
25	}
26
27	impl RelayState {
28	#[cfg(feature = "_danger-local-https")]
29	pub fn new() -> Self { RelayState { selected_relay: None } }	4✔
30	#[cfg(not(feature = "_danger-local-https"))]
31	pub fn new() -> Self { RelayState { selected_relay: None, failed_relays: Vec::new() } }
32
33	#[cfg(not(feature = "_danger-local-https"))]
34	pub fn set_selected_relay(&mut self, relay: payjoin::Url) { self.selected_relay = Some(relay); }
35
36	pub fn get_selected_relay(&self) -> Option<payjoin::Url> { self.selected_relay.clone() }	6✔
37
38	#[cfg(not(feature = "_danger-local-https"))]
39	pub fn add_failed_relay(&mut self, relay: payjoin::Url) { self.failed_relays.push(relay); }
40
41	#[cfg(not(feature = "_danger-local-https"))]
42	pub fn get_failed_relays(&self) -> Vec<payjoin::Url> { self.failed_relays.clone() }
43	}
44
45	#[derive(Clone)]
46	pub(crate) struct App {
47	config: Config,
48	db: Arc<Database>,
49	wallet: BitcoindWallet,
50	interrupt: watch::Receiver<()>,
51	relay_state: Arc<Mutex<RelayState>>,
52	}
53
54	#[async_trait::async_trait]
55	impl AppTrait for App {
56	fn new(config: Config) -> Result<Self> {	4✔
57	let db = Arc::new(Database::create(&config.db_path)?);	4✔
58	let relay_state = Arc::new(Mutex::new(RelayState::new()));	4✔
59	let (interrupt_tx, interrupt_rx) = watch::channel(());	4✔
60	tokio::spawn(handle_interrupt(interrupt_tx));	4✔
61	let wallet = BitcoindWallet::new(&config.bitcoind)?;	4✔
62	let app = Self { config, db, wallet, interrupt: interrupt_rx, relay_state };	4✔
63	app.wallet()	4✔
64	.network()	4✔
65	.context("Failed to connect to bitcoind. Check config RPC connection.")?;	4✔
66	Ok(app)	4✔
67	}	4✔
68
69	fn wallet(&self) -> BitcoindWallet { self.wallet.clone() }	10✔
70
71	async fn send_payjoin(&self, bip21: &str, fee_rate: FeeRate) -> Result<()> {	2✔
72	use payjoin::UriExt;
73	let uri =	2✔
74	Uri::try_from(bip21).map_err(\|e\| anyhow!("Failed to create URI from BIP21: {}", e))?;	2✔
75	let uri = uri.assume_checked();	2✔
76	let uri = uri.check_pj_supported().map_err(\|_\| anyhow!("URI does not support Payjoin"))?;	2✔
77	let url = uri.extras.endpoint();	2✔
78	// match bip21 to send_session public_key
79	let req_ctx = match self.db.get_send_session(url)? {	2✔
80	Some(send_session) => send_session,	1✔
81	None => {
82	let psbt = self.create_original_psbt(&uri, fee_rate)?;	1✔
83	let mut persister = SenderPersister::new(self.db.clone());	1✔
84	let new_sender = SenderBuilder::new(psbt, uri.clone())	1✔
85	.build_recommended(fee_rate)	1✔
86	.with_context(\|\| "Failed to build payjoin request")?;	1✔
87	let storage_token = new_sender	1✔
88	.persist(&mut persister)	1✔
89	.map_err(\|e\| anyhow!("Failed to persist sender: {}", e))?;	1✔
90	Sender::load(storage_token, &persister)	1✔
91	.map_err(\|e\| anyhow!("Failed to load sender: {}", e))?	1✔
92	}
93	};
94	self.spawn_payjoin_sender(req_ctx).await	2✔
95	}	4✔
96
97	async fn receive_payjoin(&self, amount: Amount) -> Result<()> {	1✔
98	let address = self.wallet().get_new_address()?;	1✔
99	let ohttp_keys =	1✔
100	unwrap_ohttp_keys_or_else_fetch(&self.config, self.relay_state.clone()).await?;	1✔
101	let mut persister = ReceiverPersister::new(self.db.clone());	1✔
102	let new_receiver = NewReceiver::new(	1✔
103	address,	1✔
104	self.config.v2()?.pj_directory.clone(),	1✔
105	ohttp_keys.clone(),	1✔
106	None,	1✔
107	)?;	×
108	let storage_token = new_receiver	1✔
109	.persist(&mut persister)	1✔
110	.map_err(\|e\| anyhow!("Failed to persist receiver: {}", e))?;	1✔
111	let session = Receiver::load(storage_token, &persister)	1✔
112	.map_err(\|e\| anyhow!("Failed to load receiver: {}", e))?;	1✔
113	self.spawn_payjoin_receiver(session, Some(amount)).await	1✔
114	}	2✔
115
116	#[allow(clippy::incompatible_msrv)]
117	async fn resume_payjoins(&self) -> Result<()> {	1✔
118	let recv_sessions = self.db.get_recv_sessions()?;	1✔
119	let send_sessions = self.db.get_send_sessions()?;	1✔
120
121	if recv_sessions.is_empty() && send_sessions.is_empty() {	1✔
122	println!("No sessions to resume.");	×
123	return Ok(());	×
124	}	1✔
125		1✔
126	let mut tasks = Vec::new();	1✔
127
128	for session in recv_sessions {	2✔
129	let self_clone = self.clone();	1✔
130	tasks.push(tokio::spawn(async move {	1✔
131	self_clone.spawn_payjoin_receiver(session, None).await	1✔
132	}));	1✔
133	}	1✔
134
135	for session in send_sessions {	1✔
136	let self_clone = self.clone();	×
137	tasks.push(tokio::spawn(async move { self_clone.spawn_payjoin_sender(session).await }));	×
138	}	×
139
140	let mut interrupt = self.interrupt.clone();	1✔
141	tokio::select! {	1✔
142	_ = async {	1✔
143	for task in tasks {	1✔
144	let _ = task.await;	1✔
145	}
146	} => {	×
147	println!("All resumed sessions completed.");	×
148	}	×
149	_ = interrupt.changed() => {	1✔
150	println!("Resumed sessions were interrupted.");	1✔
151	}	1✔
152	}
153	Ok(())	1✔
154	}	2✔
155	}
156
157	impl App {
158	#[allow(clippy::incompatible_msrv)]
159	async fn spawn_payjoin_sender(&self, mut req_ctx: Sender) -> Result<()> {	2✔
160	let mut interrupt = self.interrupt.clone();	2✔
161	tokio::select! {	2✔
162	res = self.long_poll_post(&mut req_ctx) => {	2✔
163	self.process_pj_response(res?)?;	1✔
164	self.db.clear_send_session(req_ctx.endpoint())?;	1✔
165	}
166	_ = interrupt.changed() => {	2✔
167	println!("Interrupted. Call `send` with the same arguments to resume this session or `resume` to resume all sessions.");	1✔
168	}	1✔
169	}
170	Ok(())	2✔
171	}	2✔
172
173	#[allow(clippy::incompatible_msrv)]
174	async fn spawn_payjoin_receiver(	2✔
175	&self,	2✔
176	mut session: Receiver,	2✔
177	amount: Option<Amount>,	2✔
178	) -> Result<()> {	2✔
179	println!("Receive session established");	2✔
180	let mut pj_uri = session.pj_uri();	2✔
181	pj_uri.amount = amount;	2✔
182	let ohttp_relay = self.unwrap_relay_or_else_fetch().await?;	2✔
183
184	println!("Request Payjoin by sharing this Payjoin Uri:");	2✔
185	println!("{pj_uri}");	2✔
186		2✔
187	let mut interrupt = self.interrupt.clone();	2✔
188	let receiver = tokio::select! {	2✔
189	res = self.long_poll_fallback(&mut session) => res,	2✔
190	_ = interrupt.changed() => {	2✔
191	println!("Interrupted. Call the `resume` command to resume all sessions.");	1✔
192	return Ok(());	1✔
193	}
194	}?;	×
195
196	println!("Fallback transaction received. Consider broadcasting this to get paid if the Payjoin fails:");	1✔
197	println!("{}", serialize_hex(&receiver.extract_tx_to_schedule_broadcast()));	1✔
198	let mut payjoin_proposal = match self.process_v2_proposal(receiver.clone()) {	1✔
199	Ok(proposal) => proposal,	1✔
200	Err(Error::ReplyToSender(e)) => {	×
NEW 201	return Err(handle_recoverable_error(e, receiver, &ohttp_relay).await);	×
202	}
203	Err(e) => return Err(e.into()),	×
204	};
205	let (req, ohttp_ctx) = payjoin_proposal	1✔
206	.extract_req(ohttp_relay)	1✔
207	.map_err(\|e\| anyhow!("v2 req extraction failed {}", e))?;	1✔
208	println!("Got a request from the sender. Responding with a Payjoin proposal.");	1✔
209	let res = post_request(req).await?;	1✔
210	payjoin_proposal	1✔
211	.process_res(&res.bytes().await?, ohttp_ctx)	1✔
212	.map_err(\|e\| anyhow!("Failed to deserialize response {}", e))?;	1✔
213	let payjoin_psbt = payjoin_proposal.psbt().clone();	1✔
214	println!(	1✔
215	"Response successful. Watch mempool for successful Payjoin. TXID: {}",	1✔
216	payjoin_psbt.extract_tx_unchecked_fee_rate().clone().compute_txid()	1✔
217	);	1✔
218	self.db.clear_recv_session()?;	1✔
219	Ok(())	1✔
220	}	2✔
221
222	async fn long_poll_post(&self, req_ctx: &mut Sender) -> Result<Psbt> {	2✔
223	let ohttp_relay = self.unwrap_relay_or_else_fetch().await?;	2✔
224
225	match req_ctx.extract_v2(ohttp_relay.clone()) {	2✔
226	Ok((req, ctx)) => {	2✔
227	println!("Posting Original PSBT Payload request...");	2✔
228	let response = post_request(req).await?;	2✔
229	println!("Sent fallback transaction");	2✔
230	let v2_ctx = Arc::new(ctx.process_response(&response.bytes().await?)?);	2✔
231	loop {
232	let (req, ohttp_ctx) = v2_ctx.extract_req(&ohttp_relay)?;	3✔
233	let response = post_request(req).await?;	3✔
234	match v2_ctx.process_response(&response.bytes().await?, ohttp_ctx) {	2✔
235	Ok(Some(psbt)) => return Ok(psbt),	1✔
236	Ok(None) => {	1✔
237	println!("No response yet.");	1✔
238	}	1✔
239	Err(re) => {	×
240	println!("{re}");	×
241	log::debug!("{re:?}");	×
242	return Err(anyhow!("Response error").context(re));	×
243	}
244	}
245	}
246	}
247	Err(_) => {
248	let (req, v1_ctx) = req_ctx.extract_v1();	×
249	println!("Posting Original PSBT Payload request...");	×
250	let response = post_request(req).await?;	×
251	println!("Sent fallback transaction");	×
252	match v1_ctx.process_response(&mut response.bytes().await?.to_vec().as_slice()) {	×
253	Ok(psbt) => Ok(psbt),	×
254	Err(re) => {	×
255	println!("{re}");	×
256	log::debug!("{re:?}");	×
257	Err(anyhow!("Response error").context(re))	×
258	}
259	}
260	}
261	}
262	}	1✔
263
264	async fn long_poll_fallback(	2✔
265	&self,	2✔
266	session: &mut payjoin::receive::v2::Receiver,	2✔
267	) -> Result<payjoin::receive::v2::UncheckedProposal> {	2✔
268	let ohttp_relay = self.unwrap_relay_or_else_fetch().await?;	2✔
269
270	loop {
271	let (req, context) = session.extract_req(&ohttp_relay)?;	2✔
272	println!("Polling receive request...");	2✔
273	let ohttp_response = post_request(req).await?;	2✔
274	let proposal = session	1✔
275	.process_res(ohttp_response.bytes().await?.to_vec().as_slice(), context)	1✔
276	.map_err(\|_\| anyhow!("GET fallback failed"))?;	1✔
277	log::debug!("got response");	1✔
278	if let Some(proposal) = proposal {	1✔
279	break Ok(proposal);	1✔
280	}	×
281	}
282	}	1✔
283
284	fn process_v2_proposal(	1✔
285	&self,	1✔
286	proposal: payjoin::receive::v2::UncheckedProposal,	1✔
287	) -> Result<payjoin::receive::v2::PayjoinProposal, Error> {	1✔
288	let wallet = self.wallet();	1✔
289		1✔
290	// in a payment processor where the sender could go offline, this is where you schedule to broadcast the original_tx	1✔
291	let _to_broadcast_in_failure_case = proposal.extract_tx_to_schedule_broadcast();	1✔
292
293	// Receive Check 1: Can Broadcast
294	let proposal =	1✔
295	proposal.check_broadcast_suitability(None, \|tx\| Ok(wallet.can_broadcast(tx)?))?;	1✔
296	log::trace!("check1");	1✔
297
298	// Receive Check 2: receiver can't sign for proposal inputs
299	let proposal = proposal.check_inputs_not_owned(\|input\| Ok(wallet.is_mine(input)?))?;	1✔
300	log::trace!("check2");	1✔
301
302	// Receive Check 3: have we seen this input before? More of a check for non-interactive i.e. payment processor receivers.
303	let payjoin = proposal	1✔
304	.check_no_inputs_seen_before(\|input\| Ok(self.db.insert_input_seen_before(*input)?))?;	1✔
305	log::trace!("check3");	1✔
306
307	let payjoin = payjoin	1✔
308	.identify_receiver_outputs(\|output_script\| Ok(wallet.is_mine(output_script)?))?	2✔
309	.commit_outputs();	1✔
310
311	let provisional_payjoin = try_contributing_inputs(payjoin.clone(), &wallet)	1✔
312	.map_err(ReplyableError::Implementation)?;	1✔
313
314	let payjoin_proposal = provisional_payjoin.finalize_proposal(	1✔
315	\|psbt\| Ok(wallet.process_psbt(psbt)?),	1✔
316	None,	1✔
317	self.config.max_fee_rate,	1✔
318	)?;	1✔
319	let payjoin_proposal_psbt = payjoin_proposal.psbt();	1✔
320	log::debug!("Receiver's Payjoin proposal PSBT Rsponse: {payjoin_proposal_psbt:#?}");	1✔
321	Ok(payjoin_proposal)	1✔
322	}	1✔
323
324	async fn unwrap_relay_or_else_fetch(&self) -> Result<payjoin::Url> {	6✔
325	let selected_relay =	6✔
326	self.relay_state.lock().expect("Lock should not be poisoned").get_selected_relay();	6✔
327	let ohttp_relay = match selected_relay {	6✔
NEW 328	Some(relay) => relay,	×
329	None => validate_relay(&self.config, self.relay_state.clone()).await?,	6✔
330	};
331	Ok(ohttp_relay)	6✔
332	}	6✔
333	}
334
335	/// Handle request error by sending an error response over the directory
336	async fn handle_recoverable_error(	×
337	e: ReplyableError,	×
338	mut receiver: UncheckedProposal,	×
339	ohttp_relay: &payjoin::Url,	×
340	) -> anyhow::Error {	×
341	let to_return = anyhow!("Replied with error: {}", e);	×
342	let (err_req, err_ctx) = match receiver.extract_err_req(&e.into(), ohttp_relay) {	×
343	Ok(req_ctx) => req_ctx,	×
344	Err(e) => return anyhow!("Failed to extract error request: {}", e),	×
345	};
346
347	let err_response = match post_request(err_req).await {	×
348	Ok(response) => response,	×
349	Err(e) => return anyhow!("Failed to post error request: {}", e),	×
350	};
351
352	let err_bytes = match err_response.bytes().await {	×
353	Ok(bytes) => bytes,	×
354	Err(e) => return anyhow!("Failed to get error response bytes: {}", e),	×
355	};
356
357	if let Err(e) = receiver.process_err_res(&err_bytes, err_ctx) {	×
358	return anyhow!("Failed to process error response: {}", e);	×
359	}	×
360		×
361	to_return	×
362	}	×
363
364	fn try_contributing_inputs(	1✔
365	payjoin: payjoin::receive::v2::WantsInputs,	1✔
366	wallet: &BitcoindWallet,	1✔
367	) -> Result<payjoin::receive::v2::ProvisionalProposal, ImplementationError> {	1✔
368	let candidate_inputs = wallet.list_unspent()?;	1✔
369
370	let selected_input =	1✔
371	payjoin.try_preserving_privacy(candidate_inputs).map_err(ImplementationError::from)?;	1✔
372
373	Ok(payjoin	1✔
374	.contribute_inputs(vec![selected_input])	1✔
375	.map_err(ImplementationError::from)?	1✔
376	.commit_inputs())	1✔
377	}	1✔
378
379	async fn unwrap_ohttp_keys_or_else_fetch(	1✔
380	config: &Config,	1✔
381	relay_state: Arc<Mutex<RelayState>>,	1✔
382	) -> Result<payjoin::OhttpKeys> {	1✔
383	if let Some(keys) = config.v2()?.ohttp_keys.clone() {	1✔
384	println!("Using OHTTP Keys from config");	1✔
385	Ok(keys)	1✔
386	} else {
387	println!("Bootstrapping private network transport over Oblivious HTTP");	×
NEW 388		×
NEW 389	fetch_keys(config, relay_state.clone())	×
NEW 390	.await	×
NEW 391	.and_then(\|keys\| keys.ok_or_else(\|\| anyhow::anyhow!("No OHTTP keys found")))	×
392	}
393	}	1✔
394
395	#[cfg(not(feature = "_danger-local-https"))]
396	async fn fetch_keys(
397	config: &Config,
398	relay_state: Arc<Mutex<RelayState>>,
399	) -> Result<Option<payjoin::OhttpKeys>> {
400	use payjoin::bitcoin::secp256k1::rand::prelude::SliceRandom;
401	let payjoin_directory = config.v2()?.pj_directory.clone();
402	let relays = config.v2()?.ohttp_relays.clone();
403
404	loop {
405	let failed_relays =
406	relay_state.lock().expect("Lock should not be poisoned").get_failed_relays();
407
408	let remaining_relays: Vec<_> =
409	relays.iter().filter(\|r\| !failed_relays.contains(r)).cloned().collect();
410
411	if remaining_relays.is_empty() {
412	return Err(anyhow!("No valid relays available"));
413	}
414
415	let selected_relay =
416	match remaining_relays.choose(&mut payjoin::bitcoin::key::rand::thread_rng()) {
417	Some(relay) => relay.clone(),
418	None => return Err(anyhow!("Failed to select from remaining relays")),
419	};
420
421	relay_state
422	.lock()
423	.expect("Lock should not be poisoned")
424	.set_selected_relay(selected_relay.clone());
425
426	let ohttp_keys = {
427	payjoin::io::fetch_ohttp_keys(selected_relay.clone(), payjoin_directory.clone()).await
428	};
429
430	match ohttp_keys {
431	Ok(keys) => return Ok(Some(keys)),
432	Err(payjoin::io::Error::UnexpectedStatusCode(e)) => {
433	return Err(payjoin::io::Error::UnexpectedStatusCode(e).into());
434	}
435	Err(e) => {
436	log::debug!("Failed to connect to relay: {selected_relay}, {e:?}");
437	relay_state
438	.lock()
439	.expect("Lock should not be poisoned")
440	.add_failed_relay(selected_relay);
441	}
442	}
443	}
444	}
445
446	//Local relays are incapable of acting as proxies so we must opportunistically fetch keys from the config
447	#[cfg(feature = "_danger-local-https")]
NEW 448	async fn fetch_keys(	×
NEW 449	config: &Config,	×
NEW 450	_relay_state: Arc<Mutex<RelayState>>,	×
NEW 451	) -> Result<Option<payjoin::OhttpKeys>> {	×
NEW 452	let keys = config.v2()?.ohttp_keys.clone().expect("No OHTTP keys set");	×
NEW 453		×
NEW 454	Ok(Some(keys))	×
NEW 455	}	×
456
457	#[cfg(not(feature = "_danger-local-https"))]
458	async fn validate_relay(
459	config: &Config,
460	relay_state: Arc<Mutex<RelayState>>,
461	) -> Result<payjoin::Url> {
462	use payjoin::bitcoin::secp256k1::rand::prelude::SliceRandom;
463	let payjoin_directory = config.v2()?.pj_directory.clone();
464	let relays = config.v2()?.ohttp_relays.clone();
465
466	loop {
467	let failed_relays =
468	relay_state.lock().expect("Lock should not be poisoned").get_failed_relays();
469
470	let remaining_relays: Vec<_> =
471	relays.iter().filter(\|r\| !failed_relays.contains(r)).cloned().collect();
472
473	if remaining_relays.is_empty() {
474	return Err(anyhow!("No valid relays available"));
475	}
476
477	let selected_relay =
478	match remaining_relays.choose(&mut payjoin::bitcoin::key::rand::thread_rng()) {
479	Some(relay) => relay.clone(),
480	None => return Err(anyhow!("Failed to select from remaining relays")),
481	};
482
483	relay_state
484	.lock()
485	.expect("Lock should not be poisoned")
486	.set_selected_relay(selected_relay.clone());
487
488	let ohttp_keys =
489	payjoin::io::fetch_ohttp_keys(selected_relay.clone(), payjoin_directory.clone()).await;
490
491	match ohttp_keys {
492	Ok(_) => return Ok(selected_relay),
493	Err(payjoin::io::Error::UnexpectedStatusCode(e)) => {
494	return Err(payjoin::io::Error::UnexpectedStatusCode(e).into());
495	}
496	Err(e) => {
497	log::debug!("Failed to connect to relay: {selected_relay}, {e:?}");
498	relay_state
499	.lock()
500	.expect("Lock should not be poisoned")
501	.add_failed_relay(selected_relay);
502	}
503	}
504	}
505	}
506
507	#[cfg(feature = "_danger-local-https")]
508	async fn validate_relay(	6✔
509	config: &Config,	6✔
510	_relay_state: Arc<Mutex<RelayState>>,	6✔
511	) -> Result<payjoin::Url> {	6✔
512	let relay = config.v2()?.ohttp_relays.first().expect("no OHTTP relay set").clone();	6✔
513		6✔
514	Ok(relay)	6✔
515	}	6✔
516
517	async fn post_request(req: payjoin::Request) -> Result<reqwest::Response> {	8✔
518	let http = http_agent()?;	8✔
519	http.post(req.url)	8✔
520	.header("Content-Type", req.content_type)	8✔
521	.body(req.body)	8✔
522	.send()	8✔
523	.await	8✔
524	.map_err(map_reqwest_err)	6✔
525	}	6✔
526
527	fn map_reqwest_err(e: reqwest::Error) -> anyhow::Error {	×
528	match e.status() {	×
529	Some(status_code) => anyhow!("HTTP request failed: {} {}", status_code, e),	×
530	None => anyhow!("No HTTP response: {}", e),	×
531	}
532	}	×

payjoin / rust-payjoin / 15052840071

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous