Add default response body size limit

Cargo.lock

@@ -14,11 +14,13 @@ dependencies = [
  "axum",
  "background-jobs",
  "base64",
+ "bytes",
  "chrono",
  "derive_builder",
  "dyn-clone",
  "enum_delegate",
  "env_logger",
+ "futures-util",
  "http",
  "http-signature-normalization",
  "http-signature-normalization-reqwest",
@@ -417,9 +419,9 @@ checksum = "572f695136211188308f16ad2ca5c851a712c464060ae6974944458eb83880ba"
 
 [[package]]
 name = "bytes"
-version = "1.2.1"
+version = "1.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ec8a7b6a70fde80372154c65702f00a0f56f3e1c36abbc6c440484be248856db"
+checksum = "dfb24e866b15a1af2a1b663f10c6b6b8f397a84aadb828f12e5b289ec23a3a3c"
 
 [[package]]
 name = "bytestring"
@@ -1481,6 +1483,7 @@ dependencies = [
  "serde_urlencoded",
  "tokio",
  "tokio-native-tls",
+ "tokio-util",
  "tower-service",
  "url",
  "wasm-bindgen",

Cargo.toml

@@ -15,7 +15,7 @@ async-trait = "0.1.58"
 url = { version = "2.3.1", features = ["serde"] }
 serde_json = { version = "1.0.87", features = ["preserve_order"] }
 anyhow = "1.0.66"
-reqwest = { version = "0.11.12", features = ["json"] }
+reqwest = { version = "0.11.12", features = ["json", "stream"] }
 reqwest-middleware = "0.2.0"
 tracing = "0.1.37"
 base64 = "0.13.1"
@@ -33,6 +33,8 @@ httpdate = "1.0.2"
 http-signature-normalization-reqwest = { version = "0.7.1", default-features = false, features = ["sha-2", "middleware"] }
 http-signature-normalization = "0.6.0"
 actix-rt = { version = "2.7.0" }
+bytes = "1.3.0"
+futures-util = { version = "0.3.25", default-features = false }
 
 actix-web = { version = "4.2.1", default-features = false, optional = true }
 axum = { version = "0.6.0", features = ["json", "headers", "macros", "original-uri"], optional = true }

src/lib.rs

@@ -111,6 +111,12 @@ pub struct InstanceSettings {
     /// use the same as timeout when sending
     #[builder(default = "Duration::from_secs(10)")]
     request_timeout: Duration,
+    /// Maximum size of the HTTP response body in bytes.
+    /// This is limited to prevent potential memory exhaustion DoS attacks.
+    ///
+    /// Defaults to 2MB
+    #[builder(default = "2 * 1024 * 1024")]
+    response_body_size: usize,
     /// Function used to verify that urls are valid, used when receiving activities or fetching remote
     /// objects. Use this to implement functionality like federation blocklists. In case verification
     /// fails, it should return an error message.
@@ -173,6 +179,8 @@ pub enum Error {
     NotFound,
     #[error("Request limit was reached during fetch")]
     RequestLimit,
+    #[error("Response body limit was reached during fetch")]
+    ResponseBodyLimit,
     #[error("Object to be fetched was deleted")]
     ObjectDeleted,
     #[error("{0}")]

src/utils.rs

@@ -1,4 +1,6 @@
 use crate::{Error, LocalInstance, APUB_JSON_CONTENT_TYPE};
+use bytes::{BufMut, BytesMut};
+use futures_util::stream::StreamExt;
 use http::{header::HeaderName, HeaderValue, StatusCode};
 use serde::de::DeserializeOwned;
 use std::collections::BTreeMap;
@@ -33,7 +35,17 @@ pub async fn fetch_object_http<Kind: DeserializeOwned>(
         return Err(Error::ObjectDeleted);
     }
 
-    res.json().await.map_err(Error::conv)
+    let mut bytes_stream = res.bytes_stream();
+    let mut body = BytesMut::new();
+    while let Some(chunk) = bytes_stream.next().await.transpose().map_err(Error::conv)? {
+        body.put(chunk);
+
+        if body.len() > instance.settings.response_body_size {
+            return Err(Error::ResponseBodyLimit);
+        }
+    }
+
+    serde_json::from_slice(&body).map_err(Error::conv)
 }
 
 /// Check that both urls have the same domain. If not, return UrlVerificationError.