Enforce response body size limits in payjoin-cli

Cargo-minimal.lock

@@ -2787,6 +2787,7 @@ dependencies = [
  "clap 4.5.45",
  "config",
  "dirs",
+ "futures",
  "http-body-util",
  "hyper",
  "hyper-util",
@@ -3410,12 +3411,14 @@ dependencies = [
  "sync_wrapper",
  "tokio",
  "tokio-rustls",
+ "tokio-util",
  "tower",
  "tower-http",
  "tower-service",
  "url",
  "wasm-bindgen",
  "wasm-bindgen-futures",
+ "wasm-streams",
  "web-sys",
  "webpki-roots 1.0.7",
 ]
@@ -4421,9 +4424,9 @@ dependencies = [
 
 [[package]]
 name = "tokio-util"
-version = "0.7.1"
+version = "0.7.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0edfdeb067411dba2044da6d1cb2df793dd35add7888d73c16e3381ded401764"
+checksum = "1d68074620f57a0b21594d9735eb2e98ab38b17f80d3fcb189fca266771ca60d"
 dependencies = [
  "bytes",
  "futures-core",
@@ -5111,9 +5114,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-futures"
-version = "0.4.19"
+version = "0.4.37"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1fe9756085a84584ee9457a002b7cdfe0bfff169f45d2591d8be1345a6780e35"
+checksum = "c02dbc21516f9f1f04f187958890d7e6026df8d16540b7ad9492bc34a67cea03"
 dependencies = [
  "cfg-if 1.0.0",
  "js-sys",
@@ -5153,11 +5156,24 @@ dependencies = [
  "unicode-ident",
 ]
 
+[[package]]
+name = "wasm-streams"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b65dc4c90b63b118468cf747d8bf3566c1913ef60be765b5730ead9e0a3ba129"
+dependencies = [
+ "futures-util",
+ "js-sys",
+ "wasm-bindgen",
+ "wasm-bindgen-futures",
+ "web-sys",
+]
+
 [[package]]
 name = "web-sys"
-version = "0.3.37"
+version = "0.3.64"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2d6f51648d8c56c366144378a33290049eafdd784071077f6fe37dae64c1c4cb"
+checksum = "9b85cbef8c220a6abc02aefd892dfc0fc23afb1c6a426316ec33253a3877249b"
 dependencies = [
  "js-sys",
  "wasm-bindgen",

Cargo-recent.lock

@@ -2881,6 +2881,7 @@ dependencies = [
  "clap 4.6.1",
  "config",
  "dirs",
+ "futures",
  "http-body-util",
  "hyper",
  "hyper-util",
@@ -3548,12 +3549,14 @@ dependencies = [
  "sync_wrapper",
  "tokio",
  "tokio-rustls",
+ "tokio-util",
  "tower",
  "tower-http",
  "tower-service",
  "url",
  "wasm-bindgen",
  "wasm-bindgen-futures",
+ "wasm-streams",
  "web-sys",
  "webpki-roots 1.0.8",
 ]
@@ -5277,6 +5280,19 @@ dependencies = [
  "unicode-ident",
 ]
 
+[[package]]
+name = "wasm-streams"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65"
+dependencies = [
+ "futures-util",
+ "js-sys",
+ "wasm-bindgen",
+ "wasm-bindgen-futures",
+ "web-sys",
+]
+
 [[package]]
 name = "web-sys"
 version = "0.3.102"

payjoin-cli/Cargo.toml

@@ -21,7 +21,7 @@ path = "src/main.rs"
 default = ["v2"]
 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"]
+v1 = ["payjoin/v1", "futures", "hyper", "hyper-util", "http-body-util"]
 v2 = ["payjoin/v2", "payjoin/io"]
 
 [dependencies]
@@ -32,6 +32,7 @@ bitcoind-async-client = "0.14.0"
 clap = { version = "4.5.45", features = ["derive"] }
 config = "0.15.17"
 dirs = "6.0.0"
+futures = { version = "0.3.21", optional = true }
 http-body-util = { version = "0.1.3", optional = true }
 hyper = { version = "1.8.0", features = ["http1", "server"], optional = true }
 hyper-util = { version = "0.1.18", optional = true }
@@ -42,6 +43,7 @@ r2d2_sqlite = "0.22.0"
 reqwest = { version = "0.12.23", default-features = false, features = [
   "json",
   "rustls-tls",
+  "stream",
 ] }
 rusqlite = { version = "0.29.0", features = ["bundled"] }
 serde = { version = "1.0.228", features = ["derive"] }

payjoin-cli/src/app/mod.rs

@@ -1,6 +1,12 @@
 use std::collections::HashMap;
 
+#[cfg(feature = "v1")]
+use anyhow::anyhow;
 use anyhow::Result;
+#[cfg(feature = "v1")]
+use futures::{Stream, StreamExt};
+#[cfg(feature = "v1")]
+use hyper::body::Bytes;
 use payjoin::bitcoin::psbt::Psbt;
 use payjoin::bitcoin::{self, Address, Amount, FeeRate};
 use tokio::signal;
@@ -103,3 +109,22 @@ async fn handle_interrupt(tx: watch::Sender<()>) {
     }
     let _ = tx.send(());
 }
+
+#[cfg(feature = "v1")]
+pub async fn read_limited_body<S, E>(mut stream: S, expected_len: usize) -> Result<Vec<u8>>
+where
+    S: Stream<Item = Result<Bytes, E>> + Unpin,
+    E: std::error::Error + Send + Sync + 'static,
+{
+    let mut body = Vec::with_capacity(expected_len);
+
+    while let Some(chunk) = stream.next().await {
+        let chunk = chunk.map_err(|e| anyhow!("Error reading body chunk: {}", e))?;
+        if body.len() + chunk.len() > expected_len {
+            return Err(anyhow!("Body exceeds expected size of {expected_len} bytes"));
+        }
+        body.extend_from_slice(&chunk);
+    }
+
+    Ok(body)
+}

payjoin-cli/src/app/v1.rs

@@ -22,9 +22,14 @@ 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::app::{handle_interrupt, http_agent, read_limited_body};
 use crate::db::Database;
 
+/// 4M block size limit with base64 encoding overhead => maximum reasonable size of content-length
+/// 4_000_000 * 4 / 3 fits in u32
+const MAX_CONTENT_LENGTH: usize = 4_000_000 * 4 / 3;
+
+#[derive(Clone)]
 struct Headers<'a>(&'a hyper::HeaderMap);
 impl payjoin::receive::v1::Headers for Headers<'_> {
     fn get_header(&self, key: &str) -> Option<&str> {
@@ -71,36 +76,33 @@ impl AppTrait for App {
         let http = http_agent(&self.config)?;
         let body = String::from_utf8(req.body.clone()).unwrap();
         println!("Sending Original PSBT to {}", req.url);
-        let response = match http
+        let response = http
             .post(req.url)
             .header("Content-Type", req.content_type)
             .body(body.clone())
             .send()
             .await
-        {
-            Ok(response) => response,
-            Err(e) => {
-                tracing::error!("HTTP request failed: {e}");
-                println!("Payjoin failed. To broadcast the fallback transaction, run:");
-                println!(
-                    "  bitcoin-cli -rpcwallet=<wallet> sendrawtransaction {:#}",
-                    serialize_hex(&fallback_tx)
-                );
-                return Err(anyhow!("HTTP request failed: {e}"));
-            }
-        };
-        let psbt = match ctx.process_response(&response.bytes().await?) {
-            Ok(psbt) => psbt,
-            Err(e) => {
-                tracing::error!("Error processing response: {e:?}");
-                println!("Payjoin failed. To broadcast the fallback transaction, run:");
-                println!(
-                    "  bitcoin-cli -rpcwallet=<wallet> sendrawtransaction {:#}",
-                    serialize_hex(&fallback_tx)
-                );
-                return Err(anyhow!("Failed to process response {e}"));
-            }
-        };
+            .with_context(|| "HTTP request failed")?;
+        println!("Sent fallback transaction txid: {}", fallback_tx.compute_txid());
+        println!("Sent fallback transaction hex: {:#}", serialize_hex(&fallback_tx));
+
+        let expected_length = response
+            .headers()
+            .get("Content-Length")
+            .and_then(|val| val.to_str().ok())
+            .and_then(|s| s.parse::<usize>().ok())
+            .unwrap_or(MAX_CONTENT_LENGTH);
+
+        if expected_length > MAX_CONTENT_LENGTH {
+            return Err(anyhow!("Response body is too large: {} bytes", expected_length));
+        }
+
+        let body = read_limited_body(response.bytes_stream(), MAX_CONTENT_LENGTH).await?;
+
+        let psbt = ctx.process_response(&body).map_err(|e| {
+            tracing::debug!("Error processing response: {e:?}");
+            anyhow!("Failed to process response {e}")
+        })?;
 
         self.process_pj_response(psbt)?;
         Ok(())
@@ -323,12 +325,27 @@ impl App {
     ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, Error> {
         let (parts, body) = req.into_parts();
         let headers = Headers(&parts.headers);
+
+        let expected_length = headers
+            .0
+            .get("Content-Length")
+            .and_then(|val| val.to_str().ok())
+            .and_then(|s| s.parse::<usize>().ok())
+            .unwrap_or(MAX_CONTENT_LENGTH);
+
+        if expected_length > MAX_CONTENT_LENGTH {
+            tracing::error!("Error: Content length exceeds max allowed");
+            return Err(Error::Implementation(ImplementationError::from(
+                anyhow!("Content length too large: {expected_length}").into_boxed_dyn_error(),
+            )));
+        }
+
+        let body =
+            read_limited_body(body.into_data_stream(), expected_length).await.map_err(|e| {
+                Error::Implementation(ImplementationError::from(e.into_boxed_dyn_error()))
+            })?;
+
         let query_string = parts.uri.query().unwrap_or("");
-        let body = body
-            .collect()
-            .await
-            .map_err(|e| Error::Implementation(ImplementationError::new(e)))?
-            .to_bytes();
         let proposal = UncheckedOriginalPayload::from_request(&body, query_string, headers)?;
 
         let payjoin_proposal = self.process_v1_proposal(proposal)?;

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

@@ -172,7 +172,7 @@ impl AppTrait for App {
                 let fallback_tx = psbt.clone().extract_tx()?;
                 let (req, ctx) = payjoin::send::v1::SenderBuilder::from_parts(
                     psbt,
-                    pj_param,
+                    &PjParam::V1(pj_param.clone()),
                     &address,
                     Some(amount),
                 )

payjoin/src/core/send/error.rs

@@ -120,10 +120,7 @@ impl fmt::Display for ValidationError {
         match &self.0 {
             Parse => write!(f, "couldn't decode as PSBT or JSON",),
             #[cfg(feature = "v1")]
-            ContentTooLarge => {
-                use crate::MAX_CONTENT_LENGTH;
-                write!(f, "content is larger than {MAX_CONTENT_LENGTH} bytes")
-            }
+            ContentTooLarge => write!(f, "The response body is too large"),
             Proposal(e) => write!(f, "proposal PSBT error: {e}"),
             #[cfg(feature = "v2")]
             V2Decapsulation(e) => write!(f, "v2 encapsulation error: {e}"),