feat: add some validation logic to migration from-ipfs (#685)

* feat: add some validation logic to migration from-ipfs

Since ceramic-one nodes refuse to sync with nodes that share bad data
its important that the migration process ensure no bad data is migrated.

This change adds the ability to validate signatures and chain ids
skipping events that do not pass validation.

Additionally you can also filter events to only events for a specific
model.

* fix: clippy

* fix test

Author: nathanielc

event-svc/src/event/migration.rs

@@ -6,6 +6,7 @@ use std::{
 use anyhow::{anyhow, Context as _};
 use ceramic_core::{EventId, Network, StreamId};
 use ceramic_event::unvalidated::{self, signed::cacao::Capability};
+use ceramic_validation::{event_verifier::Verifier, AtTime, VerifyJwsOpts};
 use cid::Cid;
 use futures::TryStreamExt;
 use ipld_core::ipld::Ipld;
@@ -26,6 +27,10 @@ pub struct Migrator<'a, S> {
     blocks: S,
     batch: Vec<ReconItem<EventId>>,
     log_tile_docs: bool,
+    // set of sep values to import, if an event doesn't have these values it will be skipped.
+    sep_filter: Vec<Vec<u8>>,
+    validate_signatures: bool,
+    supported_chains: Option<Vec<String>>,
 
     // All unsigned init payloads we have found.
     unsigned_init_payloads: BTreeSet<Cid>,
@@ -48,12 +53,18 @@ impl<'a, S: BlockStore> Migrator<'a, S> {
         network: Network,
         blocks: S,
         log_tile_docs: bool,
+        sep_filter: Vec<Vec<u8>>,
+        validate_signatures: bool,
+        supported_chains: Option<Vec<String>>,
     ) -> Result<Self> {
         Ok(Self {
             network,
             service,
             blocks,
             log_tile_docs,
+            sep_filter,
+            validate_signatures,
+            supported_chains,
             batch: Default::default(),
             unsigned_init_payloads: Default::default(),
             referenced_unsigned_init_payloads: Default::default(),
@@ -71,7 +82,7 @@ impl<'a, S: BlockStore> Migrator<'a, S> {
             Err(Error::MissingBlock(*cid))
         }
     }
-    fn handle_error(&mut self, cid: Cid, err: &Error) {
+    fn handle_error(&mut self, cid: Cid, err: &Error, model_context: ModelContext) {
         let log = match err {
             Error::FoundInitTileDoc(_) | Error::FoundDataTileDoc(_) => {
                 self.tile_doc_count += 1;
@@ -86,12 +97,12 @@ impl<'a, S: BlockStore> Migrator<'a, S> {
                     .entry(model.to_owned())
                     .and_modify(|count| *count += 1)
                     .or_insert(1);
-                self.handle_error(cid, err);
+                self.handle_error(cid, err, model.clone());
                 false
             }
         };
         if log {
-            error!(%cid, err=format!("{err:#}"), "error processing block");
+            error!(%cid, err=format!("{err:#}"), model=%model_context, "error processing block");
         }
     }
 
@@ -105,7 +116,7 @@ impl<'a, S: BlockStore> Migrator<'a, S> {
         while let Some((cid, data)) = all_blocks.try_next().await? {
             let ret = self.process_block(cid, &data).await;
             if let Err(err) = ret {
-                self.handle_error(cid, &err)
+                self.handle_error(cid, &err, ModelContext(None))
             }
             if self.batch.len() > 1000 {
                 if let Err(err) = self.write_batch().await {
@@ -208,12 +219,8 @@ impl<'a, S: BlockStore> Migrator<'a, S> {
             );
             let event = unvalidated::init::Event::new(payload);
             let event: unvalidated::Event<Ipld> = unvalidated::Event::from(Box::new(event));
-            self.batch.push(
-                event_builder
-                    .build(&self.network, event)
-                    .await
-                    .with_model_context(&model)?,
-            );
+            self.validate_build_and_push(event_builder, event, &model)
+                .await?;
             if self.batch.len() > 1000 {
                 self.write_batch().await?
             }
@@ -296,13 +303,8 @@ impl<'a, S: BlockStore> Migrator<'a, S> {
         }
         let s = unvalidated::signed::Event::new(cid, event, link, payload, capability);
         let event = unvalidated::Event::from(s);
-        self.batch.push(
-            event_builder
-                .build(&self.network, event)
-                .await
-                .with_model_context(&model)?,
-        );
-        Ok(())
+        self.validate_build_and_push(event_builder, event, &model)
+            .await
     }
     fn is_tile_doc_init(&self, data: &[u8]) -> bool {
         // Attempt to decode the payload as a loose TileDocument.
@@ -427,12 +429,51 @@ impl<'a, S: BlockStore> Migrator<'a, S> {
         }
         let time = unvalidated::TimeEvent::new(event, proof, proof_edges);
         let event: unvalidated::Event<Ipld> = unvalidated::Event::from(Box::new(time));
-        self.batch.push(
-            event_builder
-                .build(&self.network, event)
-                .await
-                .with_model_context(&model)?,
-        );
+        self.validate_build_and_push(event_builder, event, &model)
+            .await
+    }
+
+    // Build the event and push onto the batch if its is valid.
+    async fn validate_build_and_push(
+        &mut self,
+        event_builder: EventBuilder,
+        event: unvalidated::Event<Ipld>,
+        model: &ModelContext,
+    ) -> Result<()> {
+        if self.sep_filter.is_empty() || self.sep_filter.contains(&event_builder.sep) {
+            match &event {
+                unvalidated::Event::Time(event) => {
+                    if let Some(supported_chains) = &self.supported_chains {
+                        let chain_id = event.proof().chain_id().to_owned();
+                        if !supported_chains.contains(&chain_id) {
+                            return Err(anyhow!("event has unsupported chain: {chain_id}"))
+                                .with_model_context(model);
+                        }
+                    }
+                }
+                unvalidated::Event::Signed(event) => {
+                    if self.validate_signatures {
+                        event
+                            .verify_signature(
+                                Some(&event_builder.controller),
+                                &VerifyJwsOpts {
+                                    at_time: AtTime::SkipTimeChecks,
+                                    ..Default::default()
+                                },
+                            )
+                            .await
+                            .with_model_context(model)?;
+                    }
+                }
+                unvalidated::Event::Unsigned(_) => {}
+            };
+            self.batch.push(
+                event_builder
+                    .build(&self.network, event)
+                    .await
+                    .with_model_context(model)?,
+            );
+        }
         Ok(())
     }
 }

event-svc/src/event/service.rs

@@ -201,10 +201,21 @@ impl EventService {
         network: Network,
         blocks: impl BlockStore,
         log_tile_docs: bool,
+        sep_filter: Vec<Vec<u8>>,
+        validate_signatures: bool,
+        supported_chains: Option<Vec<String>>,
     ) -> Result<()> {
-        let migrator = Migrator::new(self, network, blocks, log_tile_docs)
-            .await
-            .map_err(Error::new_fatal)?;
+        let migrator = Migrator::new(
+            self,
+            network,
+            blocks,
+            log_tile_docs,
+            sep_filter,
+            validate_signatures,
+            supported_chains,
+        )
+        .await
+        .map_err(Error::new_fatal)?;
         migrator.migrate().await.map_err(Error::new_fatal)?;
         Ok(())
     }

event-svc/src/tests/migration.rs

@@ -56,7 +56,7 @@ async fn test_migration(cars: Vec<Vec<u8>>) {
     let conn = crate::store::SqlitePool::connect_in_memory().await.unwrap();
     let service = EventService::new_with_event_validation(conn).await.unwrap();
     service
-        .migrate_from_ipfs(Network::Local(42), blocks, false)
+        .migrate_from_ipfs(Network::Local(42), blocks, false, vec![], false, None)
         .await
         .unwrap();
     let actual_events: BTreeSet<_> = ceramic_api::EventService::range_with_values(

one/src/migrations.rs

@@ -1,8 +1,9 @@
-use std::{path::PathBuf, sync::Arc};
+use std::{path::PathBuf, str::FromStr as _, sync::Arc};
 
 use anyhow::{anyhow, Result};
 use async_stream::try_stream;
 use async_trait::async_trait;
+use ceramic_core::StreamId;
 use ceramic_event::unvalidated;
 use ceramic_event_svc::{BlockStore, EventService};
 use ceramic_metrics::config::Config as MetricsConfig;
@@ -105,6 +106,21 @@ pub struct FromIpfsOpts {
     /// Number of files to process
     #[clap(long, short = 'l', env = "CERAMIC_ONE_MIGRATION_LIMIT")]
     limit: Option<u64>,
+
+    /// Optional list of model stream ids. Only events from these models will be migrated.
+    /// If the list is empty all events are migrated.
+    #[clap(long, value_delimiter = ',', env = "CERAMIC_ONE_MODEL_FILTER")]
+    model_filter: Vec<String>,
+
+    /// Whether to validate the signatures of signed events.
+    /// Events with invalid signatures will be skipped and counted as errors.
+    #[clap(long, env = "CERAMIC_ONE_VALIDATE_SIGNATURES")]
+    validate_signatures: bool,
+
+    /// Whether to validate the chain of time events.
+    /// Events with invalid chains will be skipped and counted as errors.
+    #[clap(long, env = "CERAMIC_ONE_VALIDATE_CHAINS")]
+    validate_chain: bool,
 }
 
 impl From<&FromIpfsOpts> for DBOpts {
@@ -165,7 +181,27 @@ async fn from_ipfs(opts: FromIpfsOpts) -> Result<()> {
         file_limit: opts.limit,
     };
     event_svc
-        .migrate_from_ipfs(network, blocks, opts.log_tile_docs)
+        .migrate_from_ipfs(
+            network,
+            blocks,
+            opts.log_tile_docs,
+            opts.model_filter
+                .iter()
+                .map(|model| {
+                    StreamId::from_str(model)
+                        .map_err(|err| anyhow!("model filter must be valid stream id: {err}"))
+                        .map(|s| s.to_vec())
+                })
+                .collect::<Result<Vec<Vec<u8>>, _>>()?,
+            opts.validate_signatures,
+            opts.validate_chain
+                .then(|| {
+                    opts.network
+                        .supported_chain_ids()
+                        .map(|chains| chains.into_iter().map(|chain| chain.to_string()).collect())
+                })
+                .flatten(),
+        )
         .await?;
     Ok(())
 }

pipeline/src/aggregator/model_instance_validate.rs

@@ -85,7 +85,7 @@ impl ModelInstanceValidate {
             .model_instance
             .is_valid(i)
             .then(|| serde_json::from_slice::<ModelInstance>(columns.model_instance.value(i)))
-            .ok_or_validation_internal_err("cannot validate null instance document"))
+            .ok_or_validation_failure("cannot validate null instance document"))
         .context("instance payload not a valid json documnet")
         .map_to_validation_failure());
         let patch = maybe_fail!(columns
@@ -96,16 +96,16 @@ impl ModelInstanceValidate {
             .context("instance patch is not a valid json patch document")
             // This is an error because to get to this point we know that the patch has already
             // been applied.
-            .map_to_validation_internal_err());
+            .map_to_validation_failure());
         let model_version = maybe_fail!(maybe_fail!(columns
             .model_versions
             .is_valid(i)
             .then(|| Cid::read_bytes(columns.model_versions.value(i)))
-            .ok_or_validation_internal_err(
+            .ok_or_validation_failure(
                 "cannot validate instance against an unknown model version"
             ))
         .context("model version must be a valid CID")
-        .map_to_validation_internal_err());
+        .map_to_validation_failure());
         let model_definition = maybe_fail!(maybe_fail!(columns
             .model_definitions
             .is_valid(i)
@@ -123,7 +123,7 @@ impl ModelInstanceValidate {
             .event_heights
             .is_valid(i)
             .then(|| columns.event_heights.value(i))
-            .ok_or_validation_internal_err(
+            .ok_or_validation_failure(
                 "cannot validate an instance with an unknown event event_height"
             ));