fix(consensus): fix the hashes-in-blocks feature implementation (dfinity#3529)

# Problem
Currently, with the feature flag enabled, when a node sends a block
proposal to their peers it first removes all the ingress messages from
the block's payload, leaving behind only the `IngressMessageId`s which
identify the ingress messages. When a node receives such a "stripped"
block proposal it _reconstructs_ the block by retrieving the referenced
ingress messages from:

1. either their local ingress pools if there exists an ingress message
with the given `IngressMessageId` there,
2. or by requesting the missing ingress messages from peers who are
advertising the block.

The problem with the approach is that the `IngressMessageId`s do _not_
uniquely identify the the ingress messages and two ingresses with the
same content _but_ with different signatures have the same id. We thus
could end up in a situation where peers reconstruct blocks different
than what the block maker created.

## Note
The feature is disabled on the Mainnet so no subnet is currently
affected by the issue

# Proposal
To address this problem we introduce a new type `SignedIngressId` (which
contains both `IngressMessageId` and the hash of a original ingress
message bytes) which uniquely identifies a `SignedIngress`, which is
used only within the the block proposal assembler module. Everything
else remains basically the same but we add several checks to make sure
we put the correct ingress message in the block:
1. when retrieving ingress messages from ingress pools by
`IngressMessageId`s we double check that the `SignedIngressId` of the
ingress message matches what we are expecting;
2. same for when we get an ingress message from a peer.

# Alternatives considered
Instead of introducing a new type, we could modify the existing
`IngressMessageId` so that it uniquely identifies ingress messages.

disadvantages over the proposed solution:
1. without any changes to the ingress pool we would end up with
duplicates of the same ingress messages (but different signatures) in
the pool & which would be further broadcasted to the other peers
2. the duplication detection in the ingress selector would have to be
modified
3. computing `IngressMessageId` would be more expensive and we currently
do it quite often
4. the change would affect multiple more components

advantages over the proposed solution:
1. in some corner-case situations in the proposed solution a node might
have to fetch more ingress message from their peers

Author: kpop-dfinity

Cargo.lock

@@ -27,6 +27,7 @@ DEPENDENCIES = [
 ]
 
 DEV_DEPENDENCIES = [
+    "//rs/canister_client/sender",
     "//rs/p2p/test_utils",
     "//rs/test_utilities/consensus",
     "//rs/test_utilities/types",

rs/p2p/artifact_downloader/BUILD.bazel

@@ -13,6 +13,7 @@ axum = { workspace = true }
 backoff = { workspace = true }
 bytes = { workspace = true }
 ic-base-types = { path = "../../types/base_types" }
+ic-canister-client-sender = { path = "../../canister_client/sender" }
 ic-consensus-manager = { path = "../consensus_manager" }
 ic-interfaces = { path = "../../interfaces" }
 ic-logger = { path = "../../monitoring/logger" }

rs/p2p/artifact_downloader/Cargo.toml

@@ -26,8 +26,11 @@ use super::{
     download::download_ingress,
     metrics::{FetchStrippedConsensusArtifactMetrics, IngressMessageSource, IngressSenderMetrics},
     stripper::Strippable,
-    types::stripped::{
-        MaybeStrippedConsensusMessage, StrippedBlockProposal, StrippedConsensusMessageId,
+    types::{
+        stripped::{
+            MaybeStrippedConsensusMessage, StrippedBlockProposal, StrippedConsensusMessageId,
+        },
+        SignedIngressId,
     },
 };
 
@@ -169,12 +172,12 @@ impl ArtifactAssembler<ConsensusMessage, MaybeStrippedConsensusMessage>
             .start_timer();
         let mut assembler = BlockProposalAssembler::new(stripped_block_proposal);
 
-        let missing_ingress_ids = assembler.missing_ingress_messages();
+        let stripped_ingress_ids = assembler.missing_ingress_messages();
         // For each stripped object in the message, try to fetch it either from the local pools
         // or from a random peer who is advertising it.
-        for missing_ingress_id in missing_ingress_ids {
+        for stripped_ingress_id in stripped_ingress_ids {
             join_set.spawn(get_or_fetch(
-                missing_ingress_id,
+                stripped_ingress_id,
                 self.ingress_pool.clone(),
                 self.transport.clone(),
                 id.as_ref().clone(),
@@ -246,7 +249,7 @@ impl ArtifactAssembler<ConsensusMessage, MaybeStrippedConsensusMessage>
 /// Tries to get the missing object either from the pool(s) or from the peers who are advertising
 /// it.
 async fn get_or_fetch<P: Peers>(
-    ingress_message_id: IngressMessageId,
+    signed_ingress_id: SignedIngressId,
     ingress_pool: ValidatedPoolReaderRef<SignedIngress>,
     transport: Arc<dyn Transport>,
     // Id of the *full* artifact which should contain the missing data
@@ -257,13 +260,21 @@ async fn get_or_fetch<P: Peers>(
     peer_rx: P,
 ) -> (SignedIngress, NodeId) {
     // First check if the ingress message exists in the Ingress Pool.
-    if let Some(ingress_message) = ingress_pool.read().unwrap().get(&ingress_message_id) {
-        return (ingress_message, node_id);
+    if let Some(ingress_message) = ingress_pool
+        .read()
+        .unwrap()
+        .get(&signed_ingress_id.ingress_message_id)
+    {
+        // Make sure that this is the correct ingress message. [`IngressMessageId`] does _not_
+        // uniquely identify ingress messages, we thus need to perform an extra check.
+        if SignedIngressId::from(&ingress_message) == signed_ingress_id {
+            return (ingress_message, node_id);
+        }
     }
 
     download_ingress(
         transport,
-        ingress_message_id,
+        signed_ingress_id,
         full_consensus_message_id,
         &log,
         &metrics,
@@ -290,7 +301,7 @@ pub(crate) enum AssemblyError {
 
 struct BlockProposalAssembler {
     stripped_block_proposal: StrippedBlockProposal,
-    ingress_messages: Vec<(IngressMessageId, Option<SignedIngress>)>,
+    ingress_messages: Vec<(SignedIngressId, Option<SignedIngress>)>,
 }
 
 impl BlockProposalAssembler {
@@ -300,19 +311,19 @@ impl BlockProposalAssembler {
                 .stripped_ingress_payload
                 .ingress_messages
                 .iter()
-                .map(|ingress_message_id| (ingress_message_id.clone(), None))
+                .map(|signed_ingress_id| (signed_ingress_id.clone(), None))
                 .collect(),
             stripped_block_proposal,
         }
     }
 
-    /// Returns the list of [`IngressMessageId`]s which have been stripped from the block.
-    pub(crate) fn missing_ingress_messages(&self) -> Vec<IngressMessageId> {
+    /// Returns the list of ingress messages which have been stripped from the block.
+    pub(crate) fn missing_ingress_messages(&self) -> Vec<SignedIngressId> {
         self.ingress_messages
             .iter()
-            .filter_map(|(ingress_message_id, maybe_ingress)| {
+            .filter_map(|(signed_ingress_id, maybe_ingress)| {
                 if maybe_ingress.is_none() {
-                    Some(ingress_message_id)
+                    Some(signed_ingress_id)
                 } else {
                     None
                 }
@@ -326,14 +337,14 @@ impl BlockProposalAssembler {
         &mut self,
         ingress_message: SignedIngress,
     ) -> Result<(), InsertionError> {
-        let ingress_message_id = IngressMessageId::from(&ingress_message);
+        let signed_ingress_id = SignedIngressId::from(&ingress_message);
 
         // We can have at most 1000 elements in the vector, so it should be reasonably fast to do a
         // linear scan here.
         let (_, ingress) = self
             .ingress_messages
             .iter_mut()
-            .find(|(id, _maybe_ingress)| *id == ingress_message_id)
+            .find(|(id, _maybe_ingress)| *id == signed_ingress_id)
             .ok_or(InsertionError::NotNeeded)?;
 
         if ingress.is_some() {
@@ -356,7 +367,9 @@ impl BlockProposalAssembler {
         let ingresses = self
             .ingress_messages
             .into_iter()
-            .map(|(id, message)| message.ok_or_else(|| AssemblyError::Missing(id)))
+            .map(|(id, message)| {
+                message.ok_or_else(|| AssemblyError::Missing(id.ingress_message_id))
+            })
             .collect::<Result<Vec<_>, _>>()?;
         let reconstructed_ingress_payload = IngressPayload::from(ingresses);
 
@@ -377,8 +390,18 @@ impl BlockProposalAssembler {
 mod tests {
     use crate::fetch_stripped_artifact::test_utils::{
         fake_block_proposal_with_ingresses, fake_ingress_message,
-        fake_ingress_message_with_arg_size, fake_stripped_block_proposal_with_ingresses,
+        fake_ingress_message_with_arg_size, fake_ingress_message_with_sig,
+        fake_stripped_block_proposal_with_ingresses,
     };
+    use crate::fetch_stripped_artifact::types::rpc::GetIngressMessageInBlockResponse;
+    use bytes::Bytes;
+    use ic_interfaces::p2p::consensus::BouncerValue;
+    use ic_logger::no_op_logger;
+    use ic_p2p_test_utils::mocks::MockBouncerFactory;
+    use ic_p2p_test_utils::mocks::MockTransport;
+    use ic_p2p_test_utils::mocks::MockValidatedPoolReader;
+    use ic_protobuf::proxy::ProtoProxy;
+    use ic_types_test_utils::ids::NODE_1;
 
     use super::*;
 
@@ -502,4 +525,94 @@ mod tests {
             Err(InsertionError::NotNeeded)
         );
     }
+
+    #[derive(Clone)]
+    struct MockPeers(NodeId);
+
+    impl Peers for MockPeers {
+        fn peers(&self) -> Vec<NodeId> {
+            vec![self.0]
+        }
+    }
+
+    fn set_up_assembler_with_fake_dependencies(
+        ingress_pool_message: Option<SignedIngress>,
+        peers_message: Option<SignedIngress>,
+    ) -> FetchStrippedConsensusArtifact {
+        let mut mock_transport = MockTransport::new();
+        let mut ingress_pool = MockValidatedPoolReader::<SignedIngress>::default();
+
+        if let Some(ingress_message) = ingress_pool_message {
+            ingress_pool.expect_get().return_const(ingress_message);
+        }
+
+        if let Some(ingress_message) = peers_message {
+            let fake_response = axum::response::Response::builder()
+                .body(Bytes::from(
+                    pb::GetIngressMessageInBlockResponse::proxy_encode(
+                        GetIngressMessageInBlockResponse {
+                            serialized_ingress_message: ingress_message.binary().clone(),
+                        },
+                    ),
+                ))
+                .unwrap();
+
+            mock_transport
+                .expect_rpc()
+                .returning(move |_, _| (Ok(fake_response.clone())));
+        }
+
+        let consensus_pool = MockValidatedPoolReader::<ConsensusMessage>::default();
+        let mut mock_bouncer_factory = MockBouncerFactory::default();
+        mock_bouncer_factory
+            .expect_new_bouncer()
+            .returning(|_| Box::new(|_| BouncerValue::Wants));
+
+        let f = FetchStrippedConsensusArtifact::new(
+            no_op_logger(),
+            tokio::runtime::Handle::current(),
+            Arc::new(RwLock::new(consensus_pool)),
+            Arc::new(RwLock::new(ingress_pool)),
+            Arc::new(mock_bouncer_factory),
+            MetricsRegistry::new(),
+            NODE_1,
+        )
+        .0;
+
+        (f)(Arc::new(mock_transport))
+    }
+
+    /// Tests whether the assembler uses the ingress message with the correct signature in the case
+    /// when the local ingress pool contains an ingress message with the same content as the one in
+    /// the stripped block proposal but with a different signature.
+    #[tokio::test]
+    async fn roundtrip_test_with_two_identical_ingress_messages_different_signatures() {
+        let (ingress_1, _ingress_1_id) = fake_ingress_message_with_sig("fake_1", vec![1, 2, 3]);
+        let (ingress_2, _ingress_2_id) = fake_ingress_message_with_sig("fake_1", vec![2, 3, 4]);
+        assert_eq!(
+            IngressMessageId::from(&ingress_1),
+            IngressMessageId::from(&ingress_2)
+        );
+        let block_proposal = fake_block_proposal_with_ingresses(vec![ingress_2.clone()]);
+
+        let assembler = set_up_assembler_with_fake_dependencies(
+            /*ingress_pool_message=*/ Some(ingress_1.clone()),
+            /*consensus_pool_message=*/ Some(ingress_2.clone()),
+        );
+        let stripped_block_proposal =
+            assembler.disassemble_message(ConsensusMessage::BlockProposal(block_proposal.clone()));
+        let reassembled_block_proposal = assembler
+            .assemble_message(
+                stripped_block_proposal.id(),
+                Some((stripped_block_proposal, NODE_1)),
+                MockPeers(NODE_1),
+            )
+            .await
+            .expect("should reassemble the message given the dependencies");
+
+        assert_eq!(
+            reassembled_block_proposal,
+            (ConsensusMessage::BlockProposal(block_proposal), NODE_1)
+        );
+    }
 }