13655876030

Committed 04 Mar 2025 02:26PM UTC coverage: 79.709% (-0.02%) from 79.725%

Build # 13655876030

Build Type

Pull #560

github

Committed by

web-flow

Commit Message

Merge 81f4bc826 into 2761db178

Pull Request Pull Request #560: Instantiate `v2::Reciver` only after data persistence is complete

Coverage Stats

19 of 24 new or added lines in 3 files covered. (79.17%)

157 existing lines in 6 files now uncovered.

4545 of 5702 relevant lines covered (79.71%)

809.82 hits per line

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

76.52

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

use std::sync::Arc;

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::{Receiver, UncheckedProposal};
use payjoin::receive::{Error, ImplementationError, PersistanceError, ReplyableError};
use payjoin::send::v2::{Sender, SenderBuilder};
use payjoin::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::Database;

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

#[async_trait::async_trait]
impl AppTrait for App {
    fn new(config: Config) -> Result<Self> {
        let db = Arc::new(Database::create(&config.db_path)?);
        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 };
        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 req_ctx = SenderBuilder::new(psbt, uri.clone())
                    .build_recommended(fee_rate)
                    .with_context(|| "Failed to build payjoin request")?;
                self.db.insert_send_session(&mut req_ctx, url)?;
                req_ctx
            }
        };
        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).await?;
        let ephemeral_receiver = Receiver::new(
            address,
            self.config.v2()?.pj_directory.clone(),
            ohttp_keys.clone(),
            None,
        )?;
        let session = ephemeral_receiver.persist(|r| {
            self.db
                .insert_recv_session(r.clone())
                .map_err(|e| PersistanceError::PersistError(Box::new(e)))?;
            Ok(())
        })?;
        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;
        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, &self.config.v2()?.ohttp_relay).await
                );
            }
            Err(e) => return Err(e.into()),
        };
        let (req, ohttp_ctx) = payjoin_proposal
            .extract_v2_req(&self.config.v2()?.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> {
        match req_ctx.extract_v2(self.config.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(self.config.v2()?.ohttp_relay.clone())?;
                    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> {
        loop {
            let (req, context) = session.extract_req(&self.config.v2()?.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)
    }
}

/// 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 (err_req, err_ctx) = match receiver.extract_err_req(&e, 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);
    }

    e.into()
}

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) -> 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");
        let ohttp_relay = config.v2()?.ohttp_relay.clone();
        let payjoin_directory = config.v2()?.pj_directory.clone();
        #[cfg(feature = "_danger-local-https")]
        let ohttp_keys = {
            let cert_der = crate::app::read_local_cert()?;
            payjoin::io::fetch_ohttp_keys_with_cert(ohttp_relay, payjoin_directory, cert_der)
                .await?
        };
        #[cfg(not(feature = "_danger-local-https"))]
        let ohttp_keys = payjoin::io::fetch_ohttp_keys(ohttp_relay, payjoin_directory).await?;
        Ok(ohttp_keys)
    }
}

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;
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::{Receiver, UncheckedProposal};
8	use payjoin::receive::{Error, ImplementationError, PersistanceError, ReplyableError};
9	use payjoin::send::v2::{Sender, SenderBuilder};
10	use payjoin::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::Database;
18
19	#[derive(Clone)]
20	pub(crate) struct App {
21	config: Config,
22	db: Arc<Database>,
23	wallet: BitcoindWallet,
24	interrupt: watch::Receiver<()>,
25	}
26
27	#[async_trait::async_trait]
28	impl AppTrait for App {
29	fn new(config: Config) -> Result<Self> {	4✔
30	let db = Arc::new(Database::create(&config.db_path)?);	4✔
31	let (interrupt_tx, interrupt_rx) = watch::channel(());	4✔
32	tokio::spawn(handle_interrupt(interrupt_tx));	4✔
33	let wallet = BitcoindWallet::new(&config.bitcoind)?;	4✔
34	let app = Self { config, db, wallet, interrupt: interrupt_rx };	4✔
35	app.wallet()	4✔
36	.network()	4✔
37	.context("Failed to connect to bitcoind. Check config RPC connection.")?;	4✔
38	Ok(app)	4✔
39	}	4✔
40
41	fn wallet(&self) -> BitcoindWallet { self.wallet.clone() }	10✔
42
43	async fn send_payjoin(&self, bip21: &str, fee_rate: FeeRate) -> Result<()> {	2✔
44	use payjoin::UriExt;
45	let uri =	2✔
46	Uri::try_from(bip21).map_err(\|e\| anyhow!("Failed to create URI from BIP21: {}", e))?;	2✔
47	let uri = uri.assume_checked();	2✔
48	let uri = uri.check_pj_supported().map_err(\|_\| anyhow!("URI does not support Payjoin"))?;	2✔
49	let url = uri.extras.endpoint();	2✔
50	// match bip21 to send_session public_key
51	let req_ctx = match self.db.get_send_session(url)? {	2✔
52	Some(send_session) => send_session,	1✔
53	None => {
54	let psbt = self.create_original_psbt(&uri, fee_rate)?;	1✔
55	let mut req_ctx = SenderBuilder::new(psbt, uri.clone())	1✔
56	.build_recommended(fee_rate)	1✔
57	.with_context(\|\| "Failed to build payjoin request")?;	1✔
58	self.db.insert_send_session(&mut req_ctx, url)?;	1✔
59	req_ctx	1✔
60	}
61	};
62	self.spawn_payjoin_sender(req_ctx).await	2✔
63	}	4✔
64
65	async fn receive_payjoin(&self, amount: Amount) -> Result<()> {	1✔
66	let address = self.wallet().get_new_address()?;	1✔
67	let ohttp_keys = unwrap_ohttp_keys_or_else_fetch(&self.config).await?;	1✔
68	let ephemeral_receiver = Receiver::new(	1✔
69	address,	1✔
70	self.config.v2()?.pj_directory.clone(),	1✔
71	ohttp_keys.clone(),	1✔
72	None,	1✔
UNCOV 73	)?;	×
74	let session = ephemeral_receiver.persist(\|r\| {	1✔
75	self.db	1✔
76	.insert_recv_session(r.clone())	1✔
77	.map_err(\|e\| PersistanceError::PersistError(Box::new(e)))?;	1✔
78	Ok(())	1✔
79	})?;	1✔
80	self.spawn_payjoin_receiver(session, Some(amount)).await	1✔
81	}	2✔
82
83	#[allow(clippy::incompatible_msrv)]
84	async fn resume_payjoins(&self) -> Result<()> {	1✔
85	let recv_sessions = self.db.get_recv_sessions()?;	1✔
86	let send_sessions = self.db.get_send_sessions()?;	1✔
87
88	if recv_sessions.is_empty() && send_sessions.is_empty() {	1✔
UNCOV 89	println!("No sessions to resume.");	×
UNCOV 90	return Ok(());	×
91	}	1✔
92		1✔
93	let mut tasks = Vec::new();	1✔
94
95	for session in recv_sessions {	2✔
96	let self_clone = self.clone();	1✔
97	tasks.push(tokio::spawn(async move {	1✔
98	self_clone.spawn_payjoin_receiver(session, None).await	1✔
99	}));	1✔
100	}	1✔
101
102	for session in send_sessions {	1✔
UNCOV 103	let self_clone = self.clone();	×
UNCOV 104	tasks.push(tokio::spawn(async move { self_clone.spawn_payjoin_sender(session).await }));	×
UNCOV 105	}	×
106
107	let mut interrupt = self.interrupt.clone();	1✔
108	tokio::select! {	1✔
109	_ = async {	1✔
110	for task in tasks {	1✔
111	let _ = task.await;	1✔
112	}
UNCOV 113	} => {	×
UNCOV 114	println!("All resumed sessions completed.");	×
UNCOV 115	}	×
116	_ = interrupt.changed() => {	1✔
117	println!("Resumed sessions were interrupted.");	1✔
118	}	1✔
119	}
120	Ok(())	1✔
121	}	2✔
122	}
123
124	impl App {
125	#[allow(clippy::incompatible_msrv)]
126	async fn spawn_payjoin_sender(&self, mut req_ctx: Sender) -> Result<()> {	2✔
127	let mut interrupt = self.interrupt.clone();	2✔
128	tokio::select! {	2✔
129	res = self.long_poll_post(&mut req_ctx) => {	2✔
130	self.process_pj_response(res?)?;	1✔
131	self.db.clear_send_session(req_ctx.endpoint())?;	1✔
132	}
133	_ = interrupt.changed() => {	2✔
134	println!("Interrupted. Call `send` with the same arguments to resume this session or `resume` to resume all sessions.");	1✔
135	}	1✔
136	}
137	Ok(())	2✔
138	}	2✔
139
140	#[allow(clippy::incompatible_msrv)]
141	async fn spawn_payjoin_receiver(	2✔
142	&self,	2✔
143	mut session: Receiver,	2✔
144	amount: Option<Amount>,	2✔
145	) -> Result<()> {	2✔
146	println!("Receive session established");	2✔
147	let mut pj_uri = session.pj_uri();	2✔
148	pj_uri.amount = amount;	2✔
149	println!("Request Payjoin by sharing this Payjoin Uri:");	2✔
150	println!("{}", pj_uri);	2✔
151		2✔
152	let mut interrupt = self.interrupt.clone();	2✔
153	let receiver = tokio::select! {	2✔
154	res = self.long_poll_fallback(&mut session) => res,	2✔
155	_ = interrupt.changed() => {	2✔
156	println!("Interrupted. Call the `resume` command to resume all sessions.");	1✔
157	return Ok(());	1✔
158	}
UNCOV 159	}?;	×
160
161	println!("Fallback transaction received. Consider broadcasting this to get paid if the Payjoin fails:");	1✔
162	println!("{}", serialize_hex(&receiver.extract_tx_to_schedule_broadcast()));	1✔
163	let mut payjoin_proposal = match self.process_v2_proposal(receiver.clone()) {	1✔
164	Ok(proposal) => proposal,	1✔
UNCOV 165	Err(Error::ReplyToSender(e)) => {	×
UNCOV 166	return Err(	×
UNCOV 167	handle_recoverable_error(e, receiver, &self.config.v2()?.ohttp_relay).await	×
168	);
169	}
170	Err(e) => return Err(e.into()),	×
171	};
172	let (req, ohttp_ctx) = payjoin_proposal	1✔
173	.extract_v2_req(&self.config.v2()?.ohttp_relay)	1✔
174	.map_err(\|e\| anyhow!("v2 req extraction failed {}", e))?;	1✔
175	println!("Got a request from the sender. Responding with a Payjoin proposal.");	1✔
176	let res = post_request(req).await?;	1✔
177	payjoin_proposal	1✔
178	.process_res(&res.bytes().await?, ohttp_ctx)	1✔
179	.map_err(\|e\| anyhow!("Failed to deserialize response {}", e))?;	1✔
180	let payjoin_psbt = payjoin_proposal.psbt().clone();	1✔
181	println!(	1✔
182	"Response successful. Watch mempool for successful Payjoin. TXID: {}",	1✔
183	payjoin_psbt.extract_tx_unchecked_fee_rate().clone().compute_txid()	1✔
184	);	1✔
185	self.db.clear_recv_session()?;	1✔
186	Ok(())	1✔
187	}	2✔
188
189	async fn long_poll_post(&self, req_ctx: &mut Sender) -> Result<Psbt> {	2✔
190	match req_ctx.extract_v2(self.config.v2()?.ohttp_relay.clone()) {	2✔
191	Ok((req, ctx)) => {	2✔
192	println!("Posting Original PSBT Payload request...");	2✔
193	let response = post_request(req).await?;	2✔
194	println!("Sent fallback transaction");	2✔
195	let v2_ctx = Arc::new(ctx.process_response(&response.bytes().await?)?);	2✔
196	loop {
197	let (req, ohttp_ctx) =	3✔
198	v2_ctx.extract_req(self.config.v2()?.ohttp_relay.clone())?;	3✔
199	let response = post_request(req).await?;	3✔
200	match v2_ctx.process_response(&response.bytes().await?, ohttp_ctx) {	2✔
201	Ok(Some(psbt)) => return Ok(psbt),	1✔
202	Ok(None) => {	1✔
203	println!("No response yet.");	1✔
204	}	1✔
UNCOV 205	Err(re) => {	×
UNCOV 206	println!("{}", re);	×
UNCOV 207	log::debug!("{:?}", re);	×
UNCOV 208	return Err(anyhow!("Response error").context(re));	×
209	}
210	}
211	}
212	}
213	Err(_) => {
UNCOV 214	let (req, v1_ctx) = req_ctx.extract_v1()?;	×
UNCOV 215	println!("Posting Original PSBT Payload request...");	×
UNCOV 216	let response = post_request(req).await?;	×
UNCOV 217	println!("Sent fallback transaction");	×
218	match v1_ctx.process_response(&mut response.bytes().await?.to_vec().as_slice()) {	×
219	Ok(psbt) => Ok(psbt),	×
220	Err(re) => {	×
221	println!("{}", re);	×
222	log::debug!("{:?}", re);	×
223	Err(anyhow!("Response error").context(re))	×
224	}
225	}
226	}
227	}
228	}	1✔
229
230	async fn long_poll_fallback(	2✔
231	&self,	2✔
232	session: &mut payjoin::receive::v2::Receiver,	2✔
233	) -> Result<payjoin::receive::v2::UncheckedProposal> {	2✔
234	loop {
235	let (req, context) = session.extract_req(&self.config.v2()?.ohttp_relay)?;	2✔
236	println!("Polling receive request...");	2✔
237	let ohttp_response = post_request(req).await?;	2✔
238	let proposal = session	1✔
239	.process_res(ohttp_response.bytes().await?.to_vec().as_slice(), context)	1✔
240	.map_err(\|_\| anyhow!("GET fallback failed"))?;	1✔
241	log::debug!("got response");	1✔
242	if let Some(proposal) = proposal {	1✔
243	break Ok(proposal);	1✔
UNCOV 244	}	×
245	}
246	}	1✔
247
248	fn process_v2_proposal(	1✔
249	&self,	1✔
250	proposal: payjoin::receive::v2::UncheckedProposal,	1✔
251	) -> Result<payjoin::receive::v2::PayjoinProposal, Error> {	1✔
252	let wallet = self.wallet();	1✔
253		1✔
254	// in a payment processor where the sender could go offline, this is where you schedule to broadcast the original_tx	1✔
255	let _to_broadcast_in_failure_case = proposal.extract_tx_to_schedule_broadcast();	1✔
256
257	// Receive Check 1: Can Broadcast
258	let proposal =	1✔
259	proposal.check_broadcast_suitability(None, \|tx\| Ok(wallet.can_broadcast(tx)?))?;	1✔
260	log::trace!("check1");	1✔
261
262	// Receive Check 2: receiver can't sign for proposal inputs
263	let proposal = proposal.check_inputs_not_owned(\|input\| Ok(wallet.is_mine(input)?))?;	1✔
264	log::trace!("check2");	1✔
265
266	// Receive Check 3: have we seen this input before? More of a check for non-interactive i.e. payment processor receivers.
267	let payjoin = proposal	1✔
268	.check_no_inputs_seen_before(\|input\| Ok(self.db.insert_input_seen_before(*input)?))?;	1✔
269	log::trace!("check3");	1✔
270
271	let payjoin = payjoin	1✔
272	.identify_receiver_outputs(\|output_script\| Ok(wallet.is_mine(output_script)?))?	2✔
273	.commit_outputs();	1✔
274
275	let provisional_payjoin = try_contributing_inputs(payjoin.clone(), &wallet)	1✔
276	.map_err(ReplyableError::Implementation)?;	1✔
277
278	let payjoin_proposal = provisional_payjoin.finalize_proposal(	1✔
279	\|psbt\| Ok(wallet.process_psbt(psbt)?),	1✔
280	None,	1✔
281	self.config.max_fee_rate,	1✔
282	)?;	1✔
283	let payjoin_proposal_psbt = payjoin_proposal.psbt();	1✔
284	log::debug!("Receiver's Payjoin proposal PSBT Rsponse: {:#?}", payjoin_proposal_psbt);	1✔
285	Ok(payjoin_proposal)	1✔
286	}	1✔
287	}
288
289	/// Handle request error by sending an error response over the directory
UNCOV 290	async fn handle_recoverable_error(	×
UNCOV 291	e: ReplyableError,	×
UNCOV 292	mut receiver: UncheckedProposal,	×
UNCOV 293	ohttp_relay: &payjoin::Url,	×
294	) -> anyhow::Error {	×
295	let (err_req, err_ctx) = match receiver.extract_err_req(&e, ohttp_relay) {	×
296	Ok(req_ctx) => req_ctx,	×
297	Err(e) => return anyhow!("Failed to extract error request: {}", e),	×
298	};
299
300	let err_response = match post_request(err_req).await {	×
301	Ok(response) => response,	×
UNCOV 302	Err(e) => return anyhow!("Failed to post error request: {}", e),	×
303	};
304
305	let err_bytes = match err_response.bytes().await {	×
306	Ok(bytes) => bytes,	×
UNCOV 307	Err(e) => return anyhow!("Failed to get error response bytes: {}", e),	×
308	};
309
310	if let Err(e) = receiver.process_err_res(&err_bytes, err_ctx) {	×
311	return anyhow!("Failed to process error response: {}", e);	×
UNCOV 312	}	×
UNCOV 313		×
314	e.into()	×
315	}	×
316
317	fn try_contributing_inputs(	1✔
318	payjoin: payjoin::receive::v2::WantsInputs,	1✔
319	wallet: &BitcoindWallet,	1✔
320	) -> Result<payjoin::receive::v2::ProvisionalProposal, ImplementationError> {	1✔
321	let candidate_inputs = wallet.list_unspent()?;	1✔
322
323	let selected_input =	1✔
324	payjoin.try_preserving_privacy(candidate_inputs).map_err(ImplementationError::from)?;	1✔
325
326	Ok(payjoin	1✔
327	.contribute_inputs(vec![selected_input])	1✔
328	.map_err(ImplementationError::from)?	1✔
329	.commit_inputs())	1✔
330	}	1✔
331
332	async fn unwrap_ohttp_keys_or_else_fetch(config: &Config) -> Result<payjoin::OhttpKeys> {	1✔
333	if let Some(keys) = config.v2()?.ohttp_keys.clone() {	1✔
334	println!("Using OHTTP Keys from config");	1✔
335	Ok(keys)	1✔
336	} else {
UNCOV 337	println!("Bootstrapping private network transport over Oblivious HTTP");	×
UNCOV 338	let ohttp_relay = config.v2()?.ohttp_relay.clone();	×
UNCOV 339	let payjoin_directory = config.v2()?.pj_directory.clone();	×
340	#[cfg(feature = "_danger-local-https")]
341	let ohttp_keys = {	×
342	let cert_der = crate::app::read_local_cert()?;	×
343	payjoin::io::fetch_ohttp_keys_with_cert(ohttp_relay, payjoin_directory, cert_der)	×
UNCOV 344	.await?	×
345	};
346	#[cfg(not(feature = "_danger-local-https"))]
347	let ohttp_keys = payjoin::io::fetch_ohttp_keys(ohttp_relay, payjoin_directory).await?;
348	Ok(ohttp_keys)	×
349	}
350	}	1✔
351
352	async fn post_request(req: payjoin::Request) -> Result<reqwest::Response> {	8✔
353	let http = http_agent()?;	8✔
354	http.post(req.url)	8✔
355	.header("Content-Type", req.content_type)	8✔
356	.body(req.body)	8✔
357	.send()	8✔
358	.await	8✔
359	.map_err(map_reqwest_err)	6✔
360	}	6✔
361
UNCOV 362	fn map_reqwest_err(e: reqwest::Error) -> anyhow::Error {	×
UNCOV 363	match e.status() {	×
UNCOV 364	Some(status_code) => anyhow!("HTTP request failed: {} {}", status_code, e),	×
UNCOV 365	None => anyhow!("No HTTP response: {}", e),	×
366	}
367	}	×

payjoin / rust-payjoin / 13655876030

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