event_graph: Validate events before RLN proofs

src/event_graph/mod.rs

@@ -1641,42 +1641,74 @@ impl EventGraph {
             return Ok(vec![])
         }
 
-        // Pre-flight RLN verification. Done BEFORE acquiring the
-        // DAG-store write lock so a slow proof verification doesn't
-        // hold up other inserts.
+        // Pre-flight structural validation and RLN verification. Done
+        // BEFORE acquiring the DAG-store write lock so slow proof work does
+        // not hold up other inserts. Cheap structural checks run first, so
+        // malformed events cannot force proof verification or mutate RLN
+        // metadata.
         //
-        // Events we already have are skipped without verification.
-        // This matters because `rln_verify_signal` records the share
-        // on `Accepted`, and re-running it for an already-seen event
-        // would trip its duplicate-share check (returning `Rejected`)
-        // - which would be incorrect: the event is legitimate, we
-        // just already know about it.
+        // Events we already have are skipped without verification. This
+        // matters because `rln_verify_signal` records the share on `Accepted`,
+        // and re-running it for an already-seen event would trip its
+        // duplicate-share check.
         let dag_ts = u64::from_str(dag_name)?;
-        let already_have: Vec<bool> = {
+        let (already_have, structurally_valid): (Vec<bool>, Vec<bool>) = {
             let store = self.dag_store.read().await;
             let slot = store.get_slot(&dag_ts);
-            events
-                .iter()
-                .map(|ev| match slot {
-                    Some(s) => s.main_tree.contains_key(ev.id().as_bytes()).unwrap_or(false),
+            let mut already_have = Vec::with_capacity(events.len());
+            let mut structurally_valid = Vec::with_capacity(events.len());
+
+            for ev in events {
+                let eid = ev.id();
+                let have = match slot {
+                    Some(s) => s.main_tree.contains_key(eid.as_bytes())?,
                     None => false,
-                })
-                .collect()
+                };
+                already_have.push(have);
+
+                if have || ev.header.parents == NULL_PARENTS {
+                    structurally_valid.push(true);
+                    continue
+                }
+
+                let Some(slot) = slot else {
+                    structurally_valid.push(false);
+                    continue
+                };
+
+                if !slot.header_tree.contains_key(eid.as_bytes())? {
+                    structurally_valid.push(false);
+                    continue
+                }
+
+                structurally_valid
+                    .push(ev.dag_validate(&slot.header_tree, &self.config, dag_ts).await?);
+            }
+
+            (already_have, structurally_valid)
         };
 
         let mut accepted: Vec<usize> = Vec::with_capacity(events.len());
         for (i, ev) in events.iter().enumerate() {
-            // Already-known events go through structurally (the
-            // downstream `contains_key` check will skip them) but
-            // skip the RLN verifier to avoid double-recording the
-            // share for the same (epoch, internal_nullifier, x, y)
-            // tuple.
+            if !structurally_valid[i] {
+                error!(
+                    target: "event_graph::dag_insert",
+                    "[DAG_INSERT] event {} failed structural validation before RLN verification; skipping",
+                    ev.id(),
+                );
+                continue
+            }
+
+            // Already-known events go through structurally (the downstream
+            // `contains_key` check will skip them) but skip the RLN verifier to
+            // avoid double-recording the share for the same
+            // (epoch, internal_nullifier, x, y) tuple.
             if already_have[i] {
                 accepted.push(i);
                 continue
             }
-            // Genesis-shaped events have no blob and no proof -
-            // they're consensus inputs, not user signals.
+            // Genesis-shaped events have no blob and no proof - they're
+            // consensus inputs, not user signals.
             if ev.header.parents == NULL_PARENTS {
                 accepted.push(i);
                 continue
@@ -1692,9 +1724,8 @@ impl EventGraph {
                     );
                     continue
                 }
-                // Lenient path: caller pre-verified. Accept the
-                // event structurally without running the RLN
-                // verifier on it.
+                // Lenient path: caller pre-verified. Accept the event
+                // structurally without running the RLN verifier on it.
                 accepted.push(i);
                 continue
             }
@@ -1709,12 +1740,11 @@ impl EventGraph {
                 }
                 rln::SignalCheck::Slashable(_) => {
                     // The conflicting share is recorded inside
-                    // `rln_verify_signal` ONLY on `Accepted`. On
-                    // `Slashable` it returns the conflicting shares
-                    // *without* mutating metadata, so we don't
-                    // double-record. We don't broadcast a slash
-                    // here - that's the live broadcast handler's
-                    // job. We just skip the event.
+                    // `rln_verify_signal` ONLY on `Accepted`. On `Slashable` it
+                    // returns the conflicting shares without mutating metadata,
+                    // so we don't double-record. We don't broadcast a slash
+                    // here - that's the live broadcast handler's job. We just
+                    // skip the event.
                     error!(
                         target: "event_graph::dag_insert",
                         "[DAG_INSERT] sync event {} is slashable (slot reuse); skipping",

src/event_graph/proto.rs

@@ -408,32 +408,9 @@ impl ProtocolEventGraph {
                 continue
             }
 
-            // RLN: every non-genesis event MUST carry a valid signal
-            // proof. The only exception is genesis-shaped events
-            // (parents == NULL_PARENTS), which are produced by
-            // `dag_prune` on rotation and don't represent user
-            // signals. An empty blob on a non-genesis event is an
-            // unauthenticated injection attempt - strike the peer
-            // and drop the event.
-            if event.header.parents != NULL_PARENTS {
-                if blob.is_empty() {
-                    self.clone().strike().await?;
-                    continue
-                }
-                if self.verify_rln_signal(&event, &blob).await {
-                    continue
-                }
-            } else if !blob.is_empty() {
-                // A genesis-shaped event with a non-empty blob is
-                // also misbehavior - genesis events are deterministic
-                // and don't carry signals. Strike.
-                self.clone().strike().await?;
-                continue
-            }
-
             _ = self.ev_rep_sub.clean().await;
 
-            // Extract genesis info and immediately release the lock
+            // Extract genesis info and immediately release the lock.
             let genesis_ts = self.event_graph.current_genesis.read().await.header.timestamp;
             let dag_name = genesis_ts.to_string();
             let eid = event.id();
@@ -448,24 +425,39 @@ impl ProtocolEventGraph {
                 }
             }
 
-            // Flood protection
+            // Flood protection.
             bantimes.ticktock();
             if bantimes.count() > WINDOW_MAXSIZE {
                 self.channel.ban().await;
                 return Err(Error::MaliciousFlood)
             }
 
-            // Reject events from before the current rotation period
+            // Reject events from before the current rotation period.
             if event.header.timestamp < genesis_ts {
                 continue
             }
 
-            // Quick structural validation
+            // Cheap structural validation happens before RLN proof verification
+            // so malformed content, timestamps, or parent sets cannot force
+            // expensive proof work or mutate RLN metadata.
             if !event.validate_new() {
                 self.clone().strike().await?;
                 continue
             }
 
+            // RLN: every non-genesis event MUST carry a signal proof. The only
+            // exception is genesis-shaped events (parents == NULL_PARENTS),
+            // which are deterministic consensus inputs and never carry blobs.
+            if event.header.parents != NULL_PARENTS {
+                if blob.is_empty() {
+                    self.clone().strike().await?;
+                    continue
+                }
+            } else if !blob.is_empty() {
+                self.clone().strike().await?;
+                continue
+            }
+
             // Fetch missing parents (depth-bounded)
             // See MAX_PARENT_FETCH_DEPTH doc for why this is bounded.
             let mut missing = HashSet::new();
@@ -489,6 +481,35 @@ impl ProtocolEventGraph {
                 continue
             }
 
+            let structurally_valid = {
+                let store = self.event_graph.dag_store.read().await;
+                match store.get_slot(&genesis_ts) {
+                    Some(slot) => match event
+                        .dag_validate(&slot.header_tree, &self.event_graph.config, genesis_ts)
+                        .await
+                    {
+                        Ok(valid) => valid,
+                        Err(e) => {
+                            error!(
+                                target: "event_graph::protocol",
+                                "[EVENTGRAPH] Failed validating event {} before RLN verification: {e}",
+                                event.id(),
+                            );
+                            false
+                        }
+                    },
+                    None => false,
+                }
+            };
+            if !structurally_valid {
+                self.clone().strike().await?;
+                continue
+            }
+
+            if event.header.parents != NULL_PARENTS && self.verify_rln_signal(&event, &blob).await {
+                continue
+            }
+
             // Commit the already-verified signal without re-running RLN proof
             // verification. `verify_rln_signal` above recorded the share, so
             // the post-verification helper stores the blob, inserts the header,

src/event_graph/tests.rs

@@ -34,6 +34,7 @@ use crate::{
         event::Header,
         filter_requested_event_rep, merge_static_sync_event_rep,
         proto::{cap_layer_tips, count_layer_tips, EventPut, SyncDirection, MAX_RANGE_PAGE_SIZE},
+        rln::epoch_of,
         test_helpers::{
             archive_config, bounded_dag_store_config, init_logger, make_eg, make_eg_with_config,
             make_network, run_multi_node_test, shutdown_network, test_config, TestIdentity,
@@ -834,6 +835,53 @@ async fn empty_blob_rejected(ex: Arc<Executor<'static>>) {
     shutdown_network(&nodes).await;
 }
 
+#[test]
+fn evgr_multi_node_malformed_event_rejected_before_rln() {
+    init_logger();
+    run_multi_node_test(malformed_event_rejected_before_rln);
+}
+async fn malformed_event_rejected_before_rln(ex: Arc<Executor<'static>>) {
+    let nodes = make_network(ex).await;
+
+    let mut alice = TestIdentity::new();
+    for eg in &nodes {
+        alice.register_directly(eg).await.unwrap();
+    }
+
+    let dag_ts = nodes[0].current_genesis.read().await.header.timestamp;
+    let event = Event::new(b"preflight-live".to_vec(), &nodes[0]).await;
+    let message_id = alice.next_message_id(event.header.timestamp).expect("budget");
+    let blob = alice.create_signal(&event, message_id, &nodes[0]).await.unwrap();
+    let internal_nullifier = blob.internal_nullifier;
+    let blob = serialize_async(&blob).await;
+
+    let mut malformed = event.clone();
+    malformed.content.extend_from_slice(b"-tampered");
+    assert!(!malformed.content_matches_header());
+
+    nodes[0].p2p.broadcast(&EventPut(malformed.clone(), blob)).await;
+    sleep(5).await;
+
+    let epoch = epoch_of(malformed.header.timestamp);
+    for (i, eg) in nodes.iter().enumerate().skip(1) {
+        let store = eg.dag_store.read().await;
+        let slot = store.get_slot(&dag_ts).unwrap();
+        assert!(
+            !slot.main_tree.contains_key(malformed.id().as_bytes()).unwrap(),
+            "node {i} accepted a structurally invalid event",
+        );
+        drop(store);
+
+        let state = eg.rln_state.read().await;
+        assert!(
+            !state.metadata.is_reused(epoch, &internal_nullifier),
+            "node {i} ran RLN verification before structural rejection",
+        );
+    }
+
+    shutdown_network(&nodes).await;
+}
+
 #[test]
 fn evgr_multi_node_genesis_with_blob_rejected() {
     init_logger();

src/event_graph/tests_rln.rs

@@ -1434,6 +1434,48 @@ fn rln_dag_insert_with_blobs_rejects_missing_blob_on_non_genesis() {
     })
 }
 
+#[test]
+fn rln_dag_insert_with_blobs_rejects_bad_content_before_verification() {
+    // RLN proofs bind to the event header ID. If the body no longer matches
+    // the header content hash, the event is structurally invalid even though a
+    // proof for that header would verify. This must be rejected before RLN
+    // verification so the share is never recorded.
+    smol::block_on(async {
+        let eg = make_eg().await;
+        let mut alice = TestIdentity::new();
+        alice.register_directly(&eg).await.unwrap();
+
+        let dag_ts = eg.current_genesis.read().await.header.timestamp;
+        let dag_name = dag_ts.to_string();
+        let event = Event::new(b"preflight-sync".to_vec(), &eg).await;
+        let message_id = alice.next_message_id(event.header.timestamp).expect("budget");
+        let blob = alice.create_signal(&event, message_id, &eg).await.unwrap();
+        let internal_nullifier = blob.internal_nullifier;
+        let blob = serialize_async(&blob).await;
+
+        let mut malformed = event.clone();
+        malformed.content.extend_from_slice(b"-tampered");
+        assert!(!malformed.content_matches_header());
+
+        eg.header_dag_insert(vec![malformed.header.clone()], &dag_name).await.unwrap();
+        let result = eg
+            .dag_insert_with_blobs(
+                std::slice::from_ref(&malformed),
+                std::slice::from_ref(&blob),
+                &dag_name,
+            )
+            .await
+            .unwrap();
+        assert!(result.is_empty(), "malformed event must not be inserted");
+
+        let state = eg.rln_state.read().await;
+        assert!(
+            !state.metadata.is_reused(epoch_of(malformed.header.timestamp), &internal_nullifier),
+            "structural rejection must happen before RLN metadata is recorded",
+        );
+    })
+}
+
 #[test]
 fn rln_insert_signal_with_blob_rejects_missing_blob_on_non_genesis() {
     // The public insertion API must not expose the internal