Enforce content-length validation on sender and size limits on payjoin-cli

Author: GideonBature

payjoin-cli/Cargo.toml

@@ -29,17 +29,19 @@ v2 = ["payjoin/v2", "payjoin/io"]
 anyhow = "1.0.70"
 async-trait = "0.1"
 bitcoincore-rpc = "0.19.0"
+bytes = "1.10.1"
 clap = { version = "~4.0.32", features = ["derive"] }
 config = "0.13.3"
 env_logger = "0.9.0"
+futures = "0.3.31"
 http-body-util = { version = "0.1", optional = true }
 hyper = { version = "1", features = ["http1", "server"], optional = true }
 hyper-rustls = { version = "0.26", optional = true }
 hyper-util = { version = "0.1", optional = true }
 log = "0.4.7"
 payjoin = { version = "0.24.0", default-features = false }
 rcgen = { version = "0.11.1", optional = true }
-reqwest = { version = "0.12", default-features = false }
+reqwest = { version = "0.12", features = ["stream"], default-features = false }
 rustls = { version = "0.22.4", optional = true }
 serde = { version = "1.0.160", features = ["derive"] }
 sled = "0.34"

payjoin-cli/src/app/v1.rs

@@ -5,6 +5,8 @@ use std::sync::Arc;
 
 use anyhow::{anyhow, Context, Result};
 use bitcoincore_rpc::bitcoin::Amount;
+use bytes::{BufMut, BytesMut};
+use futures::StreamExt;
 use http_body_util::combinators::BoxBody;
 use http_body_util::{BodyExt, Full};
 use hyper::body::{Bytes, Incoming};
@@ -29,6 +31,11 @@ use crate::db::Database;
 #[cfg(feature = "_danger-local-https")]
 pub const LOCAL_CERT_FILE: &str = "localhost.der";
 
+/// 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> {
@@ -88,7 +95,29 @@ impl AppTrait for App {
             "Sent fallback transaction hex: {:#}",
             payjoin::bitcoin::consensus::encode::serialize_hex(&fallback_tx)
         );
-        let psbt = ctx.process_response(&response.bytes().await?).map_err(|e| {
+
+        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 mut body_stream = response.bytes_stream();
+        let mut body = BytesMut::with_capacity(expected_length.min(MAX_CONTENT_LENGTH));
+        while let Some(chunk) = body_stream.next().await {
+            let chunk = chunk.map_err(|e| anyhow!("Error reading response body: {}", e))?;
+            if body.len() + chunk.len() > MAX_CONTENT_LENGTH {
+                return Err(anyhow!("Response body exceeds maximum allowed size"));
+            }
+            body.put(chunk);
+        }
+
+        let psbt = ctx.process_response(&body).map_err(|e| {
             log::debug!("Error processing response: {e:?}");
             anyhow!("Failed to process response {e}")
         })?;
@@ -276,9 +305,34 @@ impl App {
     ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, ReplyableError> {
         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 {
+            log::error!("Error: Content length exceeds max allowed");
+            return Err(Implementation(
+                anyhow!("Content length too large: {expected_length}").into(),
+            ));
+        }
+
+        let mut body_stream = body.into_data_stream();
+        let mut body = BytesMut::with_capacity(expected_length.min(MAX_CONTENT_LENGTH));
+        while let Some(chunk) = body_stream.next().await {
+            let chunk = chunk.map_err(|e| Implementation(e.into()))?;
+            if body.len() + chunk.len() > MAX_CONTENT_LENGTH {
+                return Err(Implementation(anyhow!("Request body too large").into()));
+            }
+            body.put(chunk);
+        }
+
         let query_string = parts.uri.query().unwrap_or("");
-        let body = body.collect().await.map_err(|e| Implementation(e.into()))?.to_bytes();
-        let proposal = UncheckedProposal::from_request(&body, query_string, headers)?;
+        let body = body.freeze();
+        let proposal = UncheckedProposal::from_request(&body, query_string, headers.clone())?;
 
         let payjoin_proposal = self.process_v1_proposal(proposal)?;
         let psbt = payjoin_proposal.psbt();

payjoin/src/core/send/error.rs

@@ -6,7 +6,6 @@ use bitcoin::transaction::Version;
 use bitcoin::Sequence;
 
 use crate::error_codes::ErrorCode;
-use crate::MAX_CONTENT_LENGTH;
 
 /// Error building a Sender from a SenderBuilder.
 ///
@@ -95,7 +94,6 @@ pub struct ValidationError(InternalValidationError);
 #[derive(Debug)]
 pub(crate) enum InternalValidationError {
     Parse,
-    ContentTooLarge,
     Proposal(InternalProposalError),
     #[cfg(feature = "v2")]
     V2Encapsulation(crate::send::v2::EncapsulationError),
@@ -119,7 +117,6 @@ impl fmt::Display for ValidationError {
 
         match &self.0 {
             Parse => write!(f, "couldn't decode as PSBT or JSON",),
-            ContentTooLarge => write!(f, "content is larger than {MAX_CONTENT_LENGTH} bytes"),
             Proposal(e) => write!(f, "proposal PSBT error: {e}"),
             #[cfg(feature = "v2")]
             V2Encapsulation(e) => write!(f, "v2 encapsulation error: {e}"),
@@ -133,7 +130,6 @@ impl std::error::Error for ValidationError {
 
         match &self.0 {
             Parse => None,
-            ContentTooLarge => None,
             Proposal(e) => Some(e),
             #[cfg(feature = "v2")]
             V2Encapsulation(e) => Some(e),

payjoin/src/core/send/v1.rs

@@ -29,7 +29,7 @@ use url::Url;
 use super::*;
 pub use crate::output_substitution::OutputSubstitution;
 use crate::psbt::PsbtExt;
-use crate::{PjUri, Request, MAX_CONTENT_LENGTH};
+use crate::{PjUri, Request};
 
 /// A builder to construct the properties of a `Sender`.
 #[derive(Clone)]
@@ -278,10 +278,6 @@ impl V1Context {
     /// valid you will get appropriate PSBT that you should sign and broadcast.
     #[inline]
     pub fn process_response(self, response: &[u8]) -> Result<Psbt, ResponseError> {
-        if response.len() > MAX_CONTENT_LENGTH {
-            return Err(ResponseError::from(InternalValidationError::ContentTooLarge));
-        }
-
         let res_str = std::str::from_utf8(response).map_err(|_| InternalValidationError::Parse)?;
         let proposal = Psbt::from_str(res_str).map_err(|_| ResponseError::parse(res_str))?;
         self.psbt_context.process_proposal(proposal).map_err(Into::into)
@@ -425,6 +421,7 @@ mod test {
             "message": "This version of payjoin is not supported."
         })
         .to_string();
+
         match ctx.process_response(known_json_error.as_bytes()) {
             Err(ResponseError::WellKnown(WellKnownError {
                 code: ErrorCode::VersionUnsupported,
@@ -439,6 +436,7 @@ mod test {
             "message": "This version of payjoin is not supported."
         })
         .to_string();
+
         match ctx.process_response(invalid_json_error.as_bytes()) {
             Err(ResponseError::Validation(_)) => (),
             _ => panic!("Expected unrecognized JSON error"),
@@ -448,13 +446,15 @@ mod test {
     #[test]
     fn process_response_valid() {
         let ctx = create_v1_context();
+
         let response = ctx.process_response(PAYJOIN_PROPOSAL.as_bytes());
         assert!(response.is_ok())
     }
 
     #[test]
     fn process_response_invalid_psbt() {
         let ctx = create_v1_context();
+
         let response = ctx.process_response(INVALID_PSBT.as_bytes());
         match response {
             Ok(_) => panic!("Invalid PSBT should have caused an error"),
@@ -476,6 +476,7 @@ mod test {
         let invalid_utf8 = &[0xF0];
 
         let ctx = create_v1_context();
+
         let response = ctx.process_response(invalid_utf8);
         match response {
             Ok(_) => panic!("Invalid UTF-8 should have caused an error"),
@@ -490,25 +491,4 @@ mod test {
             },
         }
     }
-
-    #[test]
-    fn process_response_invalid_buffer_len() {
-        let mut data = PAYJOIN_PROPOSAL.as_bytes().to_vec();
-        data.extend(std::iter::repeat(0).take(MAX_CONTENT_LENGTH + 1));
-
-        let ctx = create_v1_context();
-        let response = ctx.process_response(&data);
-        match response {
-            Ok(_) => panic!("Invalid buffer length should have caused an error"),
-            Err(error) => match error {
-                ResponseError::Validation(e) => {
-                    assert_eq!(
-                        e.to_string(),
-                        ValidationError::from(InternalValidationError::ContentTooLarge).to_string()
-                    );
-                }
-                _ => panic!("Unexpected error type"),
-            },
-        }
-    }
 }

payjoin/tests/integration.rs

@@ -100,7 +100,8 @@ mod integration {
             // **********************
             // Inside the Sender:
             // Sender checks, signs, finalizes, extracts, and broadcasts
-            let checked_payjoin_proposal_psbt = ctx.process_response(response.as_bytes())?;
+            let response_body = response.as_bytes();
+            let checked_payjoin_proposal_psbt = ctx.process_response(response_body)?;
             let payjoin_tx = extract_pj_tx(&sender, checked_payjoin_proposal_psbt)?;
             sender.send_raw_transaction(&payjoin_tx)?;
 
@@ -585,7 +586,8 @@ mod integration {
             // **********************
             // Inside the Sender:
             // Sender checks, signs, finalizes, extracts, and broadcasts
-            let checked_payjoin_proposal_psbt = ctx.process_response(response.as_bytes())?;
+            let response_body = response.as_bytes();
+            let checked_payjoin_proposal_psbt = ctx.process_response(response_body)?;
             let payjoin_tx = extract_pj_tx(&sender, checked_payjoin_proposal_psbt)?;
             sender.send_raw_transaction(&payjoin_tx)?;
 
@@ -1211,7 +1213,8 @@ mod integration {
             // **********************
             // Inside the Sender:
             // Sender checks, signs, finalizes, extracts, and broadcasts
-            let checked_payjoin_proposal_psbt = ctx.process_response(response.as_bytes())?;
+            let response_body = response.as_bytes();
+            let checked_payjoin_proposal_psbt = ctx.process_response(response_body)?;
             let payjoin_tx = extract_pj_tx(&sender, checked_payjoin_proposal_psbt)?;
             sender.send_raw_transaction(&payjoin_tx)?;
 
@@ -1297,7 +1300,8 @@ mod integration {
             // **********************
             // Inside the Sender:
             // Sender checks, signs, finalizes, extracts, and broadcasts
-            let checked_payjoin_proposal_psbt = ctx.process_response(response.as_bytes())?;
+            let response_body = response.as_bytes();
+            let checked_payjoin_proposal_psbt = ctx.process_response(response_body)?;
             let payjoin_tx = extract_pj_tx(&sender, checked_payjoin_proposal_psbt)?;
             sender.send_raw_transaction(&payjoin_tx)?;