Add content validation for accepted content

Author: njgheorghita

newsfragments/376.added.md

@@ -0,0 +1 @@
+Added content validation for accepted content.

trin-core/src/portalnet/overlay.rs

@@ -116,8 +116,8 @@ pub struct OverlayProtocol<TContentKey, TMetric, TValidator> {
     phantom_content_key: PhantomData<TContentKey>,
     /// Associate a metric with the overlay network.
     phantom_metric: PhantomData<TMetric>,
-    /// Declare the Validator type for a given overlay network.
-    phantom_validator: PhantomData<TValidator>,
+    /// Accepted content validator that makes requests to this/other overlay networks (or infura)
+    validator: Arc<TValidator>,
 }
 
 impl<
@@ -156,7 +156,7 @@ where
             protocol.clone(),
             utp_listener_tx.clone(),
             config.enable_metrics,
-            validator,
+            Arc::clone(&validator),
             config.query_timeout,
             config.query_peer_timeout,
             config.query_parallelism,
@@ -176,7 +176,7 @@ where
             utp_listener_tx,
             phantom_content_key: PhantomData,
             phantom_metric: PhantomData,
-            phantom_validator: PhantomData,
+            validator,
         }
     }
 
@@ -253,35 +253,54 @@ where
             ));
         }
 
-        // TODO: Verify overlay content data with an Oracle
-
-        // Temporarily store content key/value pairs to propagate here
-        let mut content_keys_values: Vec<(TContentKey, ByteList)> = Vec::new();
-
-        // Try to store the content into the database and propagate gossip the content
         for (content_key, content_value) in content_keys.into_iter().zip(content_values.to_vec()) {
             match TContentKey::try_from(content_key) {
                 Ok(key) => {
-                    // Store accepted content in DB
-                    self.store_overlay_content(&key, content_value.clone());
-                    content_keys_values.push((key, content_value))
+                    // Spawn a task that...
+                    // - Validates accepted content (this step requires a dedicated task since it
+                    // might require non-blocking requests to this/other overlay networks)
+                    // - Checks if validated content should be stored, and stores it if true
+                    // - Propagate all validated content
+                    let validator = Arc::clone(&self.validator);
+                    let storage = Arc::clone(&self.storage);
+                    let kbuckets = Arc::clone(&self.kbuckets);
+                    let request_tx = self.request_tx.clone();
+
+                    tokio::spawn(async move {
+                        // Validated received content
+                        validator
+                            .validate_content(&key, &content_value.to_vec())
+                            .await
+                            // Skip storing & propagating content if it's not valid
+                            .expect("Unable to validate received content: {err:?}");
+
+                        // Check if data should be stored, and store if true.
+                        // Ignore error since all validated content is propagated.
+                        let _ = storage
+                            .write()
+                            .store_if_should(&key, &content_value.to_vec());
+
+                        // Propagate all validated content, whether or not it was stored.
+                        Self::propagate_gossip((key, content_value), kbuckets, request_tx);
+                    });
                 }
                 Err(err) => {
-                    return Err(anyhow!(
-                        "Unexpected error while decoding overlay content key: {err}"
-                    ));
+                    warn!("Unexpected error while decoding overlay content key: {err}");
+                    continue;
                 }
             }
         }
-        // Propagate gossip accepted content
-        self.propagate_gossip(content_keys_values)?;
         Ok(())
     }
 
     /// Propagate gossip accepted content via OFFER/ACCEPT:
-    fn propagate_gossip(&self, content: Vec<(TContentKey, ByteList)>) -> anyhow::Result<()> {
+    fn propagate_gossip(
+        content_key_value: (TContentKey, ByteList),
+        kbuckets: Arc<RwLock<KBucketsTable<NodeId, Node>>>,
+        request_tx: UnboundedSender<OverlayRequest>,
+    ) {
         // Get all nodes from overlay routing table
-        let kbuckets = self.kbuckets.read();
+        let kbuckets = kbuckets.read();
         let all_nodes: Vec<&kbucket::Node<NodeId, Node>> = kbuckets
             .buckets_iter()
             .map(|kbucket| {
@@ -297,37 +316,44 @@ where
         let mut enrs_and_content: HashMap<String, Vec<RawContentKey>> = HashMap::new();
 
         // Filter all nodes from overlay routing table where XOR_distance(content_id, nodeId) < node radius
-        for content_key_value in content {
-            let interested_enrs: Vec<Enr> = all_nodes
-                .clone()
-                .into_iter()
-                .filter(|node| {
-                    XorMetric::distance(
-                        &content_key_value.0.content_id(),
-                        &node.key.preimage().raw(),
-                    ) < node.value.data_radius()
-                })
-                .map(|node| node.value.enr())
-                .collect();
-
-            // Continue if no nodes are interested in the content
-            if interested_enrs.is_empty() {
-                debug!("No nodes interested in neighborhood gossip: content_key={} num_nodes_checked={}",
-                    hex_encode(content_key_value.0.into()), all_nodes.len());
-                continue;
-            }
+        let interested_enrs: Vec<Enr> = all_nodes
+            .clone()
+            .into_iter()
+            .filter(|node| {
+                XorMetric::distance(
+                    &content_key_value.0.content_id(),
+                    &node.key.preimage().raw(),
+                ) < node.value.data_radius()
+            })
+            .map(|node| node.value.enr())
+            .collect();
 
-            // Get log2 random ENRs to gossip
-            let random_enrs = log2_random_enrs(interested_enrs)?;
+        // Continue if no nodes are interested in the content
+        if interested_enrs.is_empty() {
+            debug!(
+                "No nodes interested in neighborhood gossip: content_key={} num_nodes_checked={}",
+                hex_encode(content_key_value.0.into()),
+                all_nodes.len()
+            );
+            return;
+        }
 
-            // Temporarily store all randomly selected nodes with the content of interest.
-            // We want this so we can offer all the content to interested node in one request.
-            for enr in random_enrs {
-                enrs_and_content
-                    .entry(enr.to_base64())
-                    .or_default()
-                    .push(content_key_value.clone().0.into());
+        // Get log2 random ENRs to gossip
+        let random_enrs = match log2_random_enrs(interested_enrs) {
+            Ok(val) => val,
+            Err(msg) => {
+                debug!("No available enrs for gossip: {msg}");
+                return;
             }
+        };
+
+        // Temporarily store all randomly selected nodes with the content of interest.
+        // We want this so we can offer all the content to interested node in one request.
+        for enr in random_enrs {
+            enrs_and_content
+                .entry(enr.to_base64())
+                .or_default()
+                .push(content_key_value.clone().0.into());
         }
 
         // Create and send OFFER overlay request to the interested nodes
@@ -351,24 +377,10 @@ where
                 None,
             );
 
-            if let Err(err) = self.request_tx.send(overlay_request) {
+            if let Err(err) = request_tx.send(overlay_request) {
                 error!("Unable to send OFFER request to overlay: {err}.")
             }
         }
-        Ok(())
-    }
-
-    /// Try to store overlay content into database
-    fn store_overlay_content(&self, content_key: &TContentKey, value: ByteList) {
-        let should_store = self.storage.read().should_store(content_key);
-        match should_store {
-            Ok(_) => {
-                if let Err(err) = self.storage.write().store(content_key, &value.into()) {
-                    warn!("Unable to store accepted content: {err}");
-                }
-            }
-            Err(err) => error!("Unable to determine whether to store accepted content: {err}"),
-        }
     }
 
     /// Returns a vector of all ENR node IDs of nodes currently contained in the routing table.

trin-core/src/portalnet/storage.rs

@@ -137,6 +137,21 @@ impl PortalStorage {
         }
     }
 
+    /// Public method for automatically storing content after a `should_store` check.
+    pub fn store_if_should(
+        &mut self,
+        key: &impl OverlayContentKey,
+        value: &Vec<u8>,
+    ) -> Result<bool, PortalStorageError> {
+        match self.should_store(key)? {
+            true => {
+                let _ = self.store(key, value)?;
+                Ok(true)
+            }
+            false => Ok(false),
+        }
+    }
+
     /// Public method for storing a given value for a given content-key.
     pub fn store(
         &mut self,

trin-state/src/events.rs

@@ -15,6 +15,7 @@ impl StateEvents {
     pub async fn start(mut self) {
         loop {
             tokio::select! {
+                // blocking async!
                 Some(talk_request) = self.event_rx.recv() => {
                     self.handle_state_talk_request(talk_request).await;
                 }

trin-state/src/network.rs

@@ -5,6 +5,7 @@ use eth_trie::EthTrie;
 use log::{debug, error};
 use parking_lot::RwLock;
 use tokio::sync::mpsc::UnboundedSender;
+
 use trin_core::{
     portalnet::{
         discovery::Discovery,