Limit the maximum number of deltas in an RRDP notification. (#961)

This PR limits the number of entries in the notification file of an RRDP
repository that Routinator is willing to consider. This can be set through
the new rrdp-max-delta-list-len configuration value which defaults to a
hopefully safe 500.

If this number is exceeded, the delta list will be considered to be empty
and ignore, resulting in snapshots to be used.

Cargo.toml

@@ -31,8 +31,7 @@ pin-project-lite = "0.2.4"
 rand            = "0.8.1"
 reqwest         = { version = "0.12.4", default-features = false, features = ["blocking", "rustls-tls" ] }
 ring            = "0.17"
-#rpki            = { version = "0.18", features = [ "repository", "rrdp", "rtr", "serde", "slurm" ] }
-rpki            = { git = "https://github.com/NLnetLabs/rpki-rs.git", features = [ "repository", "rrdp", "rtr", "serde", "slurm" ] }
+rpki            = { version = "0.18.3", features = [ "repository", "rrdp", "rtr", "serde", "slurm" ] }
 rustls-pemfile  = "2.1.2"
 serde           = { version = "1.0.95", features = [ "derive" ] }
 serde_json      = "1.0.57"

doc/manual/source/manual-page.rst

@@ -279,6 +279,13 @@ The available options are:
       larger than the value provided by this option, the snapshot is used
       instead. If the option is missing, the default of 100 is used.
 
+.. option:: --rrdp-max-delta-list-len=len
+
+      If the number of deltas included in the notification file of an RRDP
+      repository is larger than the value provided, the delta list is
+      considered empty and the snapshot is used instead. If the option is
+      missing, the default of 500 is used.
+
 .. option:: --rrdp-timeout=seconds
 
       Sets the timeout in seconds for any RRDP-related network operation,
@@ -1121,6 +1128,12 @@ All values can be overridden via the command line options.
             necessary to update an RRDP repository before using the snapshot
             instead. If the value is missing, the default of 100 is used.
 
+      rrdp-max-delta-list-len
+            An integer value that specifies the maximum number of deltas
+            listed the notification file of an RRDP repository before the
+            list is considered empty and the snapshot is used instead.
+            If the value is missing, the default of 500 is used.
+
       rrdp-timeout
             An integer value that provides a timeout in seconds for all
             individual RRDP-related network operations, i.e., connects,

src/collector/rrdp/base.rs

@@ -7,7 +7,7 @@ use bytes::Bytes;
 use log::{debug, error, info, warn};
 use rpki::uri;
 use rpki::crypto::DigestAlgorithm;
-use rpki::rrdp::{DeltaInfo, NotificationFile};
+use rpki::rrdp::{DeltaInfo, DeltaListError, NotificationFile};
 use tempfile::NamedTempFile;
 use crate::config::Config;
 use crate::error::{Fatal, RunFailed};
@@ -600,6 +600,9 @@ pub struct RrdpConfig {
 
     /// The maximum number of deltas we process before using a snapshot.
     pub max_delta_count: usize,
+
+    /// The maximum length of the delta list in a notification file.
+    pub max_delta_list_len: usize,
 }
 
 impl<'a> From<&'a Config> for RrdpConfig {
@@ -609,6 +612,7 @@ impl<'a> From<&'a Config> for RrdpConfig {
             fallback_time: FallbackTime::from_config(config),
             max_object_size: config.max_object_size,
             max_delta_count: config.rrdp_max_delta_count,
+            max_delta_list_len: config.rrdp_max_delta_list_len,
         }
     }
 }
@@ -799,6 +803,7 @@ impl<'a> RepositoryUpdate<'a> {
             &self.collector.http, self.rpki_notify,
             current.as_ref().map(|x| &x.1),
             &mut self.metrics.notify_status,
+            self.collector.config.max_delta_list_len,
         ) {
             Ok(Some(notify)) => notify,
             Ok(None) => {
@@ -910,6 +915,18 @@ impl<'a> RepositoryUpdate<'a> {
         mut archive: RrdpArchive,
         state: RepositoryState,
     ) -> Result<Option<SnapshotReason>, RunFailed> {
+        if let Err(err) = notify.content().delta_status() {
+            match err {
+                DeltaListError::Oversized => {
+                    info!(
+                        "RRDP {}: Overly large delta set in notification file",
+                        self.rpki_notify
+                    );
+                    return Ok(Some(SnapshotReason::LargeDeltaSet));
+                }
+            }
+        }
+
         if let Err(reason) = notify.check_deltas(&state) {
             return Ok(Some(reason))
         }

src/collector/rrdp/update.rs

@@ -50,6 +50,7 @@ impl Notification {
         uri: &uri::Https,
         state: Option<&RepositoryState>,
         status: &mut HttpStatus,
+        delta_list_limit: usize,
     ) -> Result<Option<Self>, Failed> {
         let response = match http.conditional_response(
             uri,
@@ -79,7 +80,9 @@ impl Notification {
             Err(Failed)
         }
         else {
-            Notification::from_response(uri.clone(), response).map(Some)
+            Notification::from_response(
+                uri.clone(), response, delta_list_limit
+            ).map(Some)
         }
     }
 
@@ -88,12 +91,12 @@ impl Notification {
     ///
     /// Assumes that the response status was 200 OK.
     fn from_response(
-        uri: uri::Https, response: HttpResponse
+        uri: uri::Https, response: HttpResponse, delta_list_limit: usize
     ) -> Result<Self, Failed> {
         let etag = response.etag();
         let last_modified = response.last_modified();
-        let mut content = NotificationFile::parse(
-            io::BufReader::new(response)
+        let mut content = NotificationFile::parse_limited(
+            io::BufReader::new(response), delta_list_limit
         ).map_err(|err| {
             warn!("RRDP {}: {}", uri, err);
             Failed
@@ -597,6 +600,9 @@ pub enum SnapshotReason {
     /// The delta set in the notification file is inconsistent.
     BadDeltaSet,
 
+    /// The delta set in the notification file was too large.
+    LargeDeltaSet,
+
     /// At least one delta hash has changed from a previous update.
     DeltaMutation,
 
@@ -625,6 +631,7 @@ impl SnapshotReason {
             NewRepository => "new-repository",
             NewSession => "new-session",
             BadDeltaSet => "inconsistent-delta-set",
+            LargeDeltaSet => "large-delta-set",
             DeltaMutation => "delta-mutation",
             LargeSerial => "large-serial",
             OutdatedLocal => "outdate-local",

src/config.rs

@@ -64,6 +64,9 @@ const DEFAULT_RRDP_FALLBACK_TIME: Duration = Duration::from_secs(3600);
 /// The default for the maximum number of deltas.
 const DEFAULT_RRDP_MAX_DELTA_COUNT: usize = 100;
 
+/// The default for the maximum number of deltas parsed.
+const DEFAULT_RRDP_MAX_DELTA_LIST_LEN: usize = 500;
+
 /// The default RRDP HTTP User Agent header value to send.
 const DEFAULT_RRDP_USER_AGENT: &str = concat!("Routinator/", crate_version!());
 
@@ -220,6 +223,12 @@ pub struct Config {
     /// The maxmimm number of deltas we allow before using snapshot.
     pub rrdp_max_delta_count: usize,
 
+    /// The maximum allowd length of the delta list in an RRDP notification.
+    ///
+    /// If this number is exceeded, the delta list will be discarded and
+    /// the snapshot will be used.
+    pub rrdp_max_delta_list_len: usize,
+
     /// RRDP timeout in seconds.
     ///
     /// If this is None, no timeout is set.
@@ -535,6 +544,11 @@ impl Config {
             self.rrdp_max_delta_count = value
         }
 
+        // rrdp_max_delta_list_len
+        if let Some(value) = args.rrdp_max_delta_list_len {
+            self.rrdp_max_delta_list_len = value
+        }
+
         // rrdp_timeout
         if let Some(value) = args.rrdp_timeout {
             self.rrdp_timeout = if value == 0 {
@@ -915,6 +929,10 @@ impl Config {
                 file.take_usize("rrdp-max-delta-count")?
                 .unwrap_or(DEFAULT_RRDP_MAX_DELTA_COUNT)
             },
+            rrdp_max_delta_list_len: {
+                file.take_usize("rrdp-max-delta-list-len")?
+                .unwrap_or(DEFAULT_RRDP_MAX_DELTA_LIST_LEN)
+            },
             rrdp_timeout: {
                 match file.take_u64("rrdp-timeout")? {
                     Some(0) => None,
@@ -1157,6 +1175,7 @@ impl Config {
             rrdp_fallback: DEFAULT_RRDP_FALLBACK,
             rrdp_fallback_time: DEFAULT_RRDP_FALLBACK_TIME,
             rrdp_max_delta_count: DEFAULT_RRDP_MAX_DELTA_COUNT,
+            rrdp_max_delta_list_len: DEFAULT_RRDP_MAX_DELTA_LIST_LEN,
             rrdp_timeout: Some(DEFAULT_RRDP_TIMEOUT), 
             rrdp_connect_timeout: None,
             rrdp_tcp_keepalive: Some(DEFAULT_RRDP_TCP_KEEPALIVE),
@@ -1344,6 +1363,9 @@ impl Config {
         insert_int(
             &mut res, "rrdp-max-delta-count", self.rrdp_max_delta_count
         );
+        insert_int(
+            &mut res, "rrdp-max-delta-list-len", self.rrdp_max_delta_list_len
+        );
         insert_int(
             &mut res, "rrdp-timeout",
             match self.rrdp_timeout {
@@ -1796,6 +1818,10 @@ struct GlobalArgs {
     #[arg(long, value_name = "COUNT")]
     rrdp_max_delta_count: Option<usize>,
 
+    /// Maximum allowed length of the delta list in a RRDP notification file.
+    #[arg(long, value_name = "LEN")]
+    rrdp_max_delta_list_len: Option<usize>,
+
     /// When to fall back to rsync if RRDP fails
     #[arg(long, value_name = "POLICY")]
     rrdp_fallback: Option<FallbackPolicy>,