fix: Introduce a byte limit when fetching CUPs (#10675)

Before this PR there was no limit on the number of bytes downloaded by
the orchestrator when fetching a CUP. It was only limited indirectly by
the exponential timeout.

Instead, we introduce a byte limit similar to what is already done for
NNS delegations:

https://github.com/dfinity/ic/blob/38556a3a0f8445310a9fd7323e28e6fb2eae2d30/rs/http_endpoints/nns_delegation_manager/src/nns_delegation_manager.rs#L338-L341

We chose the same byte limit (128 MB) that already exists for consensus
artifacts as part of the P2P layer.

---------

Co-authored-by: IDX GitHub Automation <infra+github-automation@dfinity.org>

Author: eichhorl

Cargo.lock

@@ -97,6 +97,15 @@ pub const DKG_DEALINGS_PER_BLOCK: usize = 1;
 /// The replica code should be designed in such a way that if we put a channel of size 1, the protocol should still work.
 pub const MAX_P2P_IO_CHANNEL_SIZE: usize = 100_000;
 
+/// The maximum size of a single message exchanged between nodes of a subnet.
+///
+/// On purpose the value is big, otherwise there is risk of not processing important consensus messages.
+/// E.g. summary blocks generated by the consensus protocol for 40 node subnet can be bigger than 5MB.
+///
+/// This is used both by the P2P/QUIC transport layer and by the orchestrator when fetching a
+/// CatchUpPackage from a peer over HTTPS.
+pub const MAX_MESSAGE_SIZE_BYTES: usize = 128 * 1024 * 1024;
+
 /// The maximum number of pre-signatures that may be paired with signature requests,
 /// per key ID.
 pub const MAX_PAIRED_PRE_SIGNATURES: usize = 100;

rs/limits/src/lib.rs

@@ -36,6 +36,7 @@ rust_library(
         "//rs/ic_os/sev/guest",
         "//rs/interfaces",
         "//rs/interfaces/registry",
+        "//rs/limits",
         "//rs/monitoring/logger",
         "//rs/monitoring/metrics",
         "//rs/nns/constants",

rs/orchestrator/BUILD.bazel

@@ -43,6 +43,7 @@ ic-http-utils = { path = "../http_utils" }
 ic-image-upgrader = { path = "./image_upgrader" }
 ic-interfaces = { path = "../interfaces" }
 ic-interfaces-registry = { path = "../interfaces/registry" }
+ic-limits = { path = "../limits" }
 ic-logger = { path = "../monitoring/logger" }
 ic-management-canister-types-private = { path = "../types/management_canister_types" }
 ic-metrics = { path = "../monitoring/metrics" }

rs/orchestrator/Cargo.toml

@@ -35,12 +35,13 @@ use crate::{
     registry_helper::RegistryHelper,
     utils::https_endpoint_to_url,
 };
-use http_body_util::{BodyExt, Full};
+use http_body_util::{BodyExt, Full, Limited};
 use hyper::{Method, Request, StatusCode, body::Bytes};
 use hyper_rustls::HttpsConnectorBuilder;
 use hyper_util::{client::legacy::Client, rt::TokioExecutor};
 use ic_crypto_tls_interfaces::TlsConfig;
 use ic_interfaces::crypto::ThresholdSigVerifierByPublicKey;
+use ic_limits::MAX_MESSAGE_SIZE_BYTES;
 use ic_logger::{ReplicaLogger, info, warn};
 use ic_protobuf::{registry::node::v1::NodeRecord, types::v1 as pb};
 use ic_sys::fs::write_protobuf_using_tmp_file;
@@ -113,6 +114,7 @@ pub(crate) struct CatchUpPackageProvider {
     node_id: NodeId,
     backoff: Duration,
     initial_backoff: Duration,
+    max_response_size_bytes: usize,
     local_cup_reader: LocalCUPReader,
 }
 
@@ -154,6 +156,7 @@ impl CatchUpPackageProvider {
             logger,
             backoff: initial_backoff,
             initial_backoff,
+            max_response_size_bytes: MAX_MESSAGE_SIZE_BYTES,
             local_cup_reader,
         }
     }
@@ -346,7 +349,10 @@ impl CatchUpPackageProvider {
             .map_err(|e| format!("Failed to query CUP endpoint at {url}: {e:?}"))?;
 
         let status = res.status();
-        let body_req = timeout(self.backoff, res.into_body().collect());
+        let body_req = timeout(
+            self.backoff,
+            Limited::new(res.into_body(), self.max_response_size_bytes).collect(),
+        );
 
         let bytes = match body_req.await {
             Ok(result) => {
@@ -831,6 +837,32 @@ pub(crate) mod tests {
         assert_eq!(cup_provider.backoff, initial_backoff);
     }
 
+    #[tokio::test]
+    async fn test_fetch_catch_up_package_body_exceeds_size_limit() {
+        // The server responds with a full CUP body.
+        let server_addr =
+            start_server(TestService::SendBodyOrStall(Arc::new(Mutex::new(true)))).await;
+        let url = format!("https://{server_addr}");
+        let tmp_dir = tempfile::tempdir().unwrap();
+        let node_id = node_test_id(1);
+
+        let mut cup_provider = make_cup_provider(
+            tmp_dir.path().to_path_buf(),
+            node_id,
+            Duration::from_secs(5),
+        );
+        // Set a size limit far below the size of the CUP body the server sends, so that reading
+        // the body is aborted rather than buffered in full.
+        cup_provider.max_response_size_bytes = 4;
+
+        let err = cup_provider
+            .fetch_catch_up_package(&node_id, url, None)
+            .await
+            .expect_err("Expected an error when the CUP body exceeds the size limit");
+
+        assert!(err.contains("LengthLimitError"), "Unexpected error: {err}");
+    }
+
     #[tokio::test]
     async fn test_fetch_catch_up_package_unresponsive_times_out() {
         let server_addr = start_server(TestService::Unresponsive).await;

rs/orchestrator/src/catch_up_package_provider.rs

@@ -22,6 +22,7 @@ rust_library(
         "//rs/crypto/tls_interfaces",
         "//rs/crypto/utils/tls",
         "//rs/interfaces/registry",
+        "//rs/limits",
         "//rs/monitoring/logger",
         "//rs/monitoring/metrics",
         "//rs/phantom_newtype",

rs/p2p/quic_transport/BUILD.bazel

@@ -19,6 +19,7 @@ ic-base-types = { path = "../../types/base_types" }
 ic-crypto-tls-interfaces = { path = "../../crypto/tls_interfaces" }
 ic-crypto-utils-tls = { path = "../../crypto/utils/tls" }
 ic-interfaces-registry = { path = "../../interfaces/registry" }
+ic-limits = { path = "../../limits" }
 ic-logger = { path = "../../monitoring/logger" }
 ic-metrics = { path = "../../monitoring/metrics" }
 ic-protobuf = { path = "../../protobuf" }

rs/p2p/quic_transport/Cargo.toml

@@ -4,12 +4,13 @@ use std::sync::atomic::{AtomicU64, Ordering};
 
 use bytes::Bytes;
 use http::{Method, Request, Response, Version};
+use ic_limits::MAX_MESSAGE_SIZE_BYTES;
 use ic_protobuf::transport::v1 as pb;
 use prost::Message;
 use quinn::Connection;
 
 use crate::{
-    ConnId, MAX_MESSAGE_SIZE_BYTES, MessagePriority, P2PError, ResetStreamOnDrop,
+    ConnId, MessagePriority, P2PError, ResetStreamOnDrop,
     metrics::{
         INFALIBBLE, QuicTransportMetrics, observe_conn_error, observe_read_to_end_error,
         observe_stopped_error, observe_write_error,

rs/p2p/quic_transport/src/connection_handle.rs

@@ -69,10 +69,6 @@ mod metrics;
 mod request_handler;
 pub use crate::connection_manager::create_udp_socket;
 
-/// On purpose the value is big, otherwise there is risk of not processing important consensus messages.
-/// E.g. summary blocks generated by the consensus protocol for 40 node subnet can be bigger than 5MB.
-pub(crate) const MAX_MESSAGE_SIZE_BYTES: usize = 128 * 1024 * 1024;
-
 /// The shutdown primitive is useful if futures should be cancelled at places different than '.await' points.
 /// Such functionality is needed to have explicit control on the state when exiting.
 pub struct Shutdown {

rs/p2p/quic_transport/src/lib.rs

@@ -16,6 +16,7 @@ use axum::{Router, body::Body}; // TODO: try to remove the axum dep here
 use bytes::Bytes;
 use http::{Method, Request, Response, Version};
 use ic_base_types::NodeId;
+use ic_limits::MAX_MESSAGE_SIZE_BYTES;
 use ic_logger::{ReplicaLogger, info};
 use ic_protobuf::transport::v1 as pb;
 use prost::Message;
@@ -24,7 +25,7 @@ use tokio::task::JoinSet;
 use tower::ServiceExt;
 
 use crate::{
-    ConnId, MAX_MESSAGE_SIZE_BYTES, P2PError, ResetStreamOnDrop,
+    ConnId, P2PError, ResetStreamOnDrop,
     connection_handle::ConnectionHandle,
     metrics::{
         ERROR_TYPE_APP, INFALIBBLE, QuicTransportMetrics, STREAM_TYPE_BIDI, observe_conn_error,