Fixed config and packet buffer size limits

Dockerfile

@@ -33,6 +33,9 @@ RUN adduser \
 
 FROM docker.io/debian:bookworm-slim
 
+RUN apt update
+RUN apt install -y openssl
+
 COPY --from=build /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt
 
 COPY --from=build /etc/passwd /etc/passwd

ext_federation/src/client/federation_client.rs

@@ -1,8 +1,8 @@
 use async_stream::stream;
-use futures_util::{FutureExt, select, Stream, stream::StreamExt, TryStreamExt};
+use futures_util::{select, stream::StreamExt, FutureExt, Stream, TryStreamExt};
 use headers::UserAgent;
 use hyper::body::Bytes;
-use reqwest::{Client, redirect::Policy, RequestBuilder};
+use reqwest::{redirect::Policy, Client, RequestBuilder};
 use serde_json::Value;
 use thiserror::Error;
 use tokio::pin;
@@ -119,10 +119,8 @@ impl FederationRequestBuilder<'_> {
 
     async fn send_stream(
         self,
-    ) -> Result<impl Stream<Item=Result<Bytes, FederationClientError>>, FederationClientError>
+    ) -> Result<impl Stream<Item = Result<Bytes, FederationClientError>>, FederationClientError>
     {
-        eprintln!("{:?}", self.builder);
-
         let mut body = self
             .builder
             .send()
@@ -149,7 +147,7 @@ impl FederationRequestBuilder<'_> {
         let sleep = tokio::time::sleep(tokio::time::Duration::from_secs(
             self.client.timeout_seconds,
         ))
-            .fuse();
+        .fuse();
         tokio::pin!(sleep);
 
         let body = async move {
@@ -161,7 +159,7 @@ impl FederationRequestBuilder<'_> {
                 })
                 .await
         }
-            .fuse();
+        .fuse();
 
         pin!(body);
 

magnetar_common/src/config.rs

@@ -111,6 +111,7 @@ pub enum MagnetarRpcSocketKind {
 #[derive(Deserialize, Debug)]
 #[non_exhaustive]
 pub struct MagnetarRpcConfig {
+    #[serde(flatten)]
     pub connection_settings: MagnetarRpcSocketKind,
 }
 

src/rpc_v1/proto.rs

@@ -401,6 +401,14 @@ impl RpcCallDecoder {
                     let serial = buf_read.read_u64().await.into_diagnostic()?;
 
                     let name_len = buf_read.read_u32().await.into_diagnostic()? as usize;
+
+                    if name_len > 10 * 1024 {
+                        return Err(std::io::Error::new(
+                            std::io::ErrorKind::Other,
+                            "Method name too large!"
+                        )).into_diagnostic();
+                    }
+
                     if name_len > name_buf.capacity() {
                         name_buf.reserve(name_len - name_buf.capacity());
                     }
@@ -419,6 +427,14 @@ impl RpcCallDecoder {
                     }
 
                     let payload_len = buf_read.read_u32().await.into_diagnostic()? as usize;
+
+                    if payload_len > 10 * 1024 * 1024 {
+                        return Err(std::io::Error::new(
+                            std::io::ErrorKind::Other,
+                            "Payload size too large!"
+                        )).into_diagnostic();
+                    }
+
                     if payload_len > buf.capacity() {
                         buf.reserve(payload_len - buf.capacity());
                     }