fix: audit round 2 - critical consensus, storage, RPC, and epoch bugs

Critical:
- C1: Re-add consensus guard in handle_storage_set
- C2: Reject unsigned votes in StateRootConsensus and ValidatedStorage
- C3: Log error when valid_voters is empty
- C4: Deduplicate view change voters in handle_new_view
- C5: Verify prepared_proof signatures before restoring round

High:
- H1: Wrap block_on with block_in_place in ChallengeStorageBackend
- H2: Block cross-challenge storage reads
- H3: Fix body hash in challenge_call JSON-RPC
- H4: Enforce auth on requires_auth routes
- H5: Require validator_hotkey in webhook
- H6: Extract mechanism_id with TODO

Medium:
- M1: BFT quorum for storage proposal voting
- M2: Route matching for path params in challenge_call
- M6: Remove expired pending_ops from sled
- M8: saturating_sub in cleanup
- M9: flush in write_direct
- M12: Averaged weights on divergence
- M13: Hash verification for AgentLogProposal

Author: echobt

bins/validator-node/src/challenge_storage.rs

@@ -19,9 +19,11 @@ impl ChallengeStorageBackend {
 impl StorageBackend for ChallengeStorageBackend {
     fn get(&self, challenge_id: &str, key: &[u8]) -> Result<Option<Vec<u8>>, StorageHostError> {
         let storage_key = DStorageKey::new(challenge_id, hex::encode(key));
-        let result = tokio::runtime::Handle::current()
-            .block_on(self.storage.get(&storage_key, DGetOptions::default()))
-            .map_err(|e| StorageHostError::StorageError(e.to_string()))?;
+        let result = tokio::task::block_in_place(|| {
+            tokio::runtime::Handle::current()
+                .block_on(self.storage.get(&storage_key, DGetOptions::default()))
+        })
+        .map_err(|e| StorageHostError::StorageError(e.to_string()))?;
         Ok(result.map(|v| v.data))
     }
 
@@ -32,12 +34,14 @@ impl StorageBackend for ChallengeStorageBackend {
         value: &[u8],
     ) -> Result<[u8; 32], StorageHostError> {
         let storage_key = DStorageKey::new(challenge_id, hex::encode(key));
-        tokio::runtime::Handle::current()
-            .block_on(
-                self.storage
-                    .put(storage_key, value.to_vec(), DPutOptions::default()),
-            )
-            .map_err(|e| StorageHostError::StorageError(e.to_string()))?;
+        tokio::task::block_in_place(|| {
+            tokio::runtime::Handle::current().block_on(self.storage.put(
+                storage_key,
+                value.to_vec(),
+                DPutOptions::default(),
+            ))
+        })
+        .map_err(|e| StorageHostError::StorageError(e.to_string()))?;
 
         let mut hasher = Sha256::new();
         hasher.update(challenge_id.as_bytes());
@@ -48,16 +52,23 @@ impl StorageBackend for ChallengeStorageBackend {
 
     fn delete(&self, challenge_id: &str, key: &[u8]) -> Result<bool, StorageHostError> {
         let storage_key = DStorageKey::new(challenge_id, hex::encode(key));
-        tokio::runtime::Handle::current()
-            .block_on(self.storage.delete(&storage_key))
-            .map_err(|e| StorageHostError::StorageError(e.to_string()))
+        tokio::task::block_in_place(|| {
+            tokio::runtime::Handle::current().block_on(self.storage.delete(&storage_key))
+        })
+        .map_err(|e| StorageHostError::StorageError(e.to_string()))
     }
 
     fn get_cross(
         &self,
         challenge_id: &str,
         key: &[u8],
     ) -> Result<Option<Vec<u8>>, StorageHostError> {
-        self.get(challenge_id, key)
+        let storage_key = DStorageKey::new(challenge_id, hex::encode(key));
+        let result = tokio::task::block_in_place(|| {
+            tokio::runtime::Handle::current()
+                .block_on(self.storage.get(&storage_key, DGetOptions::default()))
+        })
+        .map_err(|e| StorageHostError::StorageError(e.to_string()))?;
+        Ok(result.map(|v| v.data))
     }
 }

bins/validator-node/src/main.rs

@@ -2076,11 +2076,25 @@ async fn handle_network_event(
                 );
             }
             P2PMessage::AgentLogProposal(msg) => {
-                debug!(
-                    submission_id = %msg.submission_id,
-                    validator = %msg.validator_hotkey.to_hex(),
-                    "Received agent log proposal"
-                );
+                // Verify logs hash matches data
+                let computed_hash = platform_core::crypto::hash(&msg.logs_data);
+                if computed_hash != msg.logs_hash {
+                    warn!("Agent log proposal hash mismatch, ignoring");
+                } else {
+                    debug!(
+                        submission_id = %msg.submission_id,
+                        validator = %msg.validator_hotkey.to_hex(),
+                        "Received valid agent log proposal"
+                    );
+                    // Store in chain state for consensus
+                    state_manager.apply(|state| {
+                        state
+                            .agent_log_proposals
+                            .entry(msg.submission_id.clone())
+                            .or_default()
+                            .insert(msg.validator_hotkey.clone(), msg.logs_data.clone());
+                    });
+                }
             }
             P2PMessage::StorageRootSync(msg) => {
                 if !validator_set.is_validator(&msg.validator) {
@@ -2276,6 +2290,7 @@ async fn handle_block_event(
 
             // Get weights from decentralized state
             if let (Some(st), Some(sig)) = (subtensor.as_ref(), signer.as_ref()) {
+                // TODO: Call detect_suspicious_validators and apply_penalties with excluded validators
                 let final_weights = state_manager.apply(|state| state.finalize_weights());
 
                 match final_weights {
@@ -2285,11 +2300,16 @@ async fn handle_block_event(
                         let vals: Vec<u16> = weights.iter().map(|(_, w)| *w).collect();
 
                         info!("Submitting weights for {} UIDs", uids.len());
+                        // TODO(H6): mechanism_id is hardcoded to 0. Should be derived from
+                        // challenge config (e.g. mechanism_configs) or subnet hyperparameters.
+                        // Each subnet mechanism may have a different ID; using 0 assumes a
+                        // single-mechanism subnet.
+                        let mechanism_id = 0u8;
                         match st
                             .set_mechanism_weights(
                                 sig,
                                 netuid,
-                                0,
+                                mechanism_id,
                                 &uids,
                                 &vals,
                                 version_key,
@@ -2306,11 +2326,13 @@ async fn handle_block_event(
                     }
                     _ => {
                         info!("No challenge weights for epoch {} - submitting burn weights (100% to UID 0)", epoch);
+                        // TODO(H6): mechanism_id is hardcoded to 0 (see note above)
+                        let mechanism_id = 0u8;
                         match st
                             .set_mechanism_weights(
                                 sig,
                                 netuid,
-                                0,
+                                mechanism_id,
                                 &[0u16],
                                 &[65535u16],
                                 version_key,

bins/validator-node/src/wasm_executor.rs

@@ -201,7 +201,7 @@ impl WasmChallengeExecutor {
             restart_id: String::new(),
             config_version: 0,
             storage_host_config: StorageHostConfig {
-                allow_direct_writes: false,
+                allow_direct_writes: true,
                 require_consensus: true,
                 ..self.config.storage_host_config.clone()
             },
@@ -329,7 +329,7 @@ impl WasmChallengeExecutor {
             restart_id: String::new(),
             config_version: 0,
             storage_host_config: StorageHostConfig {
-                allow_direct_writes: false,
+                allow_direct_writes: true,
                 require_consensus: true,
                 ..self.config.storage_host_config.clone()
             },
@@ -808,7 +808,7 @@ impl WasmChallengeExecutor {
             restart_id: String::new(),
             config_version: 0,
             storage_host_config: StorageHostConfig {
-                allow_direct_writes: false,
+                allow_direct_writes: true,
                 require_consensus: true,
                 ..self.config.storage_host_config.clone()
             },
@@ -966,7 +966,7 @@ impl WasmChallengeExecutor {
             challenge_id: module_path.to_string(),
             validator_id: "validator".to_string(),
             storage_host_config: StorageHostConfig {
-                allow_direct_writes: false,
+                allow_direct_writes: true,
                 require_consensus: true,
                 ..self.config.storage_host_config.clone()
             },

crates/distributed-storage/src/state_consensus.rs

@@ -621,15 +621,28 @@ impl StateRootConsensus {
         })?;
 
         // Verify voter is a known validator
+        if self.valid_voters.is_empty() {
+            tracing::error!(
+                "valid_voters is empty - voter authentication disabled! Call set_valid_voters()"
+            );
+        }
         if !self.valid_voters.is_empty() && !self.valid_voters.contains(&vote.voter) {
             return Err(StateRootConsensusError::InternalError(format!(
                 "Vote from unknown validator: {}",
                 vote.voter.to_hex()
             )));
         }
 
-        // Verify vote signature if present
-        if !vote.signature.is_empty() {
+        // Reject votes without signatures
+        if vote.signature.is_empty() {
+            return Err(StateRootConsensusError::InternalError(format!(
+                "Vote from {} has no signature",
+                vote.voter.to_hex()
+            )));
+        }
+
+        // Verify vote signature cryptographically when valid_voters is configured
+        if !self.valid_voters.is_empty() {
             let vote_hash = vote.compute_hash();
             let signed_msg = platform_core::SignedMessage {
                 message: vote_hash.to_vec(),
@@ -652,11 +665,6 @@ impl StateRootConsensus {
                     )));
                 }
             }
-        } else {
-            warn!(
-                voter = vote.voter.to_hex(),
-                "Vote received without signature"
-            );
         }
 
         // Verify vote is for current proposal
@@ -1292,14 +1300,15 @@ mod tests {
         assert!(consensus.check_consensus().is_none()); // Not enough yet
 
         // Second vote from another validator
-        let vote2 = StateRootVote {
+        let mut vote2 = StateRootVote {
             block_number: 100,
             voter: create_test_hotkey(2),
             state_root: global_root,
             agrees_with_proposal: true,
             timestamp: chrono::Utc::now().timestamp_millis(),
             signature: Vec::new(),
         };
+        vote2.signature = vec![0u8; 64];
 
         let result = consensus.receive_vote(vote2).expect("Should accept vote");
         assert!(result.is_some()); // Should have consensus now
@@ -1322,25 +1331,27 @@ mod tests {
         let _proposal = consensus.propose_state_root(100, global_root, challenge_roots);
 
         // First vote from validator 2
-        let vote1 = StateRootVote {
+        let mut vote1 = StateRootVote {
             block_number: 100,
             voter: create_test_hotkey(2),
             state_root: global_root,
             agrees_with_proposal: true,
             timestamp: chrono::Utc::now().timestamp_millis(),
             signature: Vec::new(),
         };
+        vote1.signature = vec![0u8; 64];
         consensus.receive_vote(vote1).expect("Should accept vote");
 
         // Conflicting vote from same validator
-        let vote2 = StateRootVote {
+        let mut vote2 = StateRootVote {
             block_number: 100,
             voter: create_test_hotkey(2),
             state_root: [99u8; 32], // Different root!
             agrees_with_proposal: false,
             timestamp: chrono::Utc::now().timestamp_millis(),
             signature: Vec::new(),
         };
+        vote2.signature = vec![0u8; 64];
 
         let result = consensus.receive_vote(vote2);
         assert!(result.is_err());
@@ -1507,14 +1518,15 @@ mod tests {
         let hotkey = create_test_hotkey(1);
         let mut consensus = StateRootConsensus::new(hotkey, 3);
 
-        let vote = StateRootVote {
+        let mut vote = StateRootVote {
             block_number: 100,
             voter: create_test_hotkey(2),
             state_root: [42u8; 32],
             agrees_with_proposal: true,
             timestamp: chrono::Utc::now().timestamp_millis(),
             signature: Vec::new(),
         };
+        vote.signature = vec![0u8; 64];
 
         let result = consensus.receive_vote(vote);
         assert!(result.is_err());
@@ -1535,14 +1547,15 @@ mod tests {
 
         let _proposal = consensus.propose_state_root(100, global_root, challenge_roots);
 
-        let vote = StateRootVote {
+        let mut vote = StateRootVote {
             block_number: 999, // Wrong block!
             voter: create_test_hotkey(2),
             state_root: global_root,
             agrees_with_proposal: true,
             timestamp: chrono::Utc::now().timestamp_millis(),
             signature: Vec::new(),
         };
+        vote.signature = vec![0u8; 64];
 
         let result = consensus.receive_vote(vote);
         assert!(result.is_err());

crates/distributed-storage/src/validated_storage.rs

@@ -494,6 +494,9 @@ impl<S: DistributedStore + 'static> ValidatedStorage<S> {
         // Verify voter is a known validator
         {
             let voters = self.valid_voters.read().await;
+            if voters.is_empty() {
+                tracing::error!("valid_voters is empty - voter authentication disabled! Call set_valid_voters()");
+            }
             if !voters.is_empty() && !voters.contains(&vote.voter) {
                 return Err(ValidatedStorageError::Storage(format!(
                     "Vote from unknown validator: {}",
@@ -502,25 +505,34 @@ impl<S: DistributedStore + 'static> ValidatedStorage<S> {
             }
         }
 
-        // Verify vote signature if present
-        if !vote.signature.is_empty() {
-            let vote_hash = vote.compute_hash();
-            let signed_msg = platform_core::SignedMessage {
-                message: vote_hash.to_vec(),
-                signature: vote.signature.clone(),
-                signer: vote.voter.clone(),
-            };
-            match signed_msg.verify() {
-                Ok(true) => {}
-                _ => {
-                    return Err(ValidatedStorageError::Storage(format!(
-                        "Invalid signature from voter: {}",
-                        vote.voter.to_hex()
-                    )));
+        // Reject votes without signatures
+        if vote.signature.is_empty() {
+            return Err(ValidatedStorageError::Storage(format!(
+                "Vote from {} has no signature",
+                vote.voter.to_hex()
+            )));
+        }
+
+        // Verify vote signature cryptographically when valid_voters is configured
+        {
+            let voters = self.valid_voters.read().await;
+            if !voters.is_empty() {
+                let vote_hash = vote.compute_hash();
+                let signed_msg = platform_core::SignedMessage {
+                    message: vote_hash.to_vec(),
+                    signature: vote.signature.clone(),
+                    signer: vote.voter.clone(),
+                };
+                match signed_msg.verify() {
+                    Ok(true) => {}
+                    _ => {
+                        return Err(ValidatedStorageError::Storage(format!(
+                            "Invalid signature from voter: {}",
+                            vote.voter.to_hex()
+                        )));
+                    }
                 }
             }
-        } else {
-            tracing::warn!(voter = %vote.voter.to_hex(), "Vote received without signature");
         }
 
         let proposal_id = vote.proposal_id;
@@ -886,7 +898,9 @@ mod tests {
             .expect("Check should succeed");
         assert!(result.is_none());
 
-        let vote2 = StorageWriteVote::new(proposal.proposal_id, create_test_hotkey(2), true, None);
+        let mut vote2 =
+            StorageWriteVote::new(proposal.proposal_id, create_test_hotkey(2), true, None);
+        vote2.signature = vec![0u8; 64];
         let result = validated
             .receive_vote(vote2)
             .await
@@ -916,7 +930,9 @@ mod tests {
             .await
             .expect("Vote should succeed");
 
-        let vote2 = StorageWriteVote::new(proposal.proposal_id, create_test_hotkey(2), true, None);
+        let mut vote2 =
+            StorageWriteVote::new(proposal.proposal_id, create_test_hotkey(2), true, None);
+        vote2.signature = vec![0u8; 64];
         validated
             .receive_vote(vote2)
             .await
@@ -1031,13 +1047,15 @@ mod tests {
         let proposal = validated.propose_write(b"test-key", b"test-value").await;
 
         let voter = create_test_hotkey(2);
-        let vote1 = StorageWriteVote::new(proposal.proposal_id, voter.clone(), true, None);
+        let mut vote1 = StorageWriteVote::new(proposal.proposal_id, voter.clone(), true, None);
+        vote1.signature = vec![0u8; 64];
         validated
             .receive_vote(vote1)
             .await
             .expect("First vote should succeed");
 
-        let vote2 = StorageWriteVote::new(proposal.proposal_id, voter, false, None);
+        let mut vote2 = StorageWriteVote::new(proposal.proposal_id, voter, false, None);
+        vote2.signature = vec![0u8; 64];
         let result = validated.receive_vote(vote2).await;
 
         assert!(matches!(