Verify consensus message author matches with the sender

consensus/consensus-types/src/proof_of_store.rs

@@ -211,6 +211,10 @@ impl SignedBatchInfoMsg {
     pub fn take(self) -> Vec<SignedBatchInfo> {
         self.signed_infos
     }
+
+    pub fn author(&self) -> PeerId {
+        self.signed_infos[0].signer()
+    }
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]

consensus/src/epoch_manager.rs

@@ -1446,6 +1446,7 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
                 BlockStage::EPOCH_MANAGER_RECEIVED,
             );
         }
+        self.check_author(peer_id, &consensus_msg)?;
         // we can't verify signatures from a different epoch
         let maybe_unverified_event = self.check_epoch(peer_id, consensus_msg).await?;
 
@@ -1512,6 +1513,44 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
         Ok(())
     }
 
+    fn check_author(&mut self, peer_id: AccountAddress, msg: &ConsensusMsg) -> anyhow::Result<()> {
+        let author = match msg {
+            ConsensusMsg::CommitMessage(commit) => commit.author(),
+            ConsensusMsg::ProposalMsg(proposal) => proposal.proposal().author(),
+            ConsensusMsg::VoteMsg(vote) => Some(vote.vote().author()),
+            ConsensusMsg::OrderVoteMsg(order_vote) => Some(order_vote.order_vote().author()),
+            ConsensusMsg::CommitVoteMsg(commit_vote) => Some(commit_vote.author()),
+            ConsensusMsg::BatchMsg(batch) => Some(batch.author()),
+            ConsensusMsg::RoundTimeoutMsg(round_timeout) => Some(round_timeout.author()),
+            ConsensusMsg::BatchResponse(batch_response) => Some(batch_response.author()),
+            ConsensusMsg::BatchRequestMsg(batch_request) => Some(batch_request.source()),
+            ConsensusMsg::SignedBatchInfo(sign_batch_info) => Some(sign_batch_info.author()),
+
+            ConsensusMsg::CommitDecisionMsg(_)
+            | ConsensusMsg::DAGMessage(_)
+            | ConsensusMsg::EpochChangeProof(_)
+            | ConsensusMsg::EpochRetrievalRequest(_)
+            | ConsensusMsg::ProofOfStoreMsg(_)
+            | ConsensusMsg::SyncInfo(_)
+            | ConsensusMsg::RandGenMessage(_)
+            | ConsensusMsg::BatchResponseV2(_)
+            | ConsensusMsg::BlockRetrievalRequest(_)
+            | ConsensusMsg::BlockRetrievalResponse(_) => None,
+        };
+
+        if let Some(author) = author {
+            ensure!(
+                author == peer_id,
+                "Received {:?} message from peer {} with different author {}",
+                discriminant(msg),
+                peer_id,
+                author
+            );
+        }
+
+        Ok(())
+    }
+
     async fn check_epoch(
         &mut self,
         peer_id: AccountAddress,

consensus/src/pipeline/commit_reliable_broadcast.rs

@@ -60,6 +60,13 @@ impl CommitMessage {
             _ => None,
         }
     }
+
+    pub fn author(&self) -> Option<PeerId> {
+        match self {
+            CommitMessage::Vote(vote) => Some(vote.author()),
+            _ => None,
+        }
+    }
 }
 
 impl RBMessage for CommitMessage {}