Add mode to support signed variant of Discovery protocol. (#19587)

## Description 

Adds a new version of discovery where all NodeInfo messages are signed
by the originator.

crates/shared-crypto/src/intent.rs

@@ -66,7 +66,8 @@ pub enum IntentScope {
     ProofOfPossession = 5, // Used as a signature representing an authority's proof of possession of its authority protocol key.
     HeaderDigest = 6,      // Used for narwhal authority signature on header digest.
     BridgeEventUnused = 7, // for bridge purposes but it's currently not included in messages.
-    ConsensusBlock = 8,    // Used for consensus authority signature on block's digest
+    ConsensusBlock = 8,    // Used for consensus authority signature on block's digest.
+    DiscoveryPeers = 9,    // Used for reporting peer addresses in discovery.
 }
 
 impl TryFrom<u8> for IntentScope {

crates/sui-config/src/p2p.rs

@@ -319,6 +319,12 @@ pub struct DiscoveryConfig {
     /// to this peer, nor advertise this peer's info to other peers in the network.
     #[serde(skip_serializing_if = "Vec::is_empty", default)]
     pub allowlisted_peers: Vec<AllowlistedPeer>,
+
+    /// If true, Discovery will require all provided peer information to be signed
+    /// by the originating peer.
+    ///
+    /// If unspecified, this will default to false.
+    pub enable_node_info_signatures: Option<bool>,
 }
 
 impl DiscoveryConfig {
@@ -345,6 +351,10 @@ impl DiscoveryConfig {
         // defaults None to Public
         self.access_type.unwrap_or(AccessType::Public)
     }
+
+    pub fn enable_node_info_signatures(&self) -> bool {
+        self.enable_node_info_signatures.unwrap_or(false)
+    }
 }
 
 #[derive(Clone, Debug, Default, Deserialize, Serialize)]

crates/sui-network/Cargo.toml

@@ -14,6 +14,7 @@ serde.workspace = true
 tonic.workspace = true
 dashmap.workspace = true
 tower.workspace = true
+shared-crypto.workspace = true
 
 sui-archival.workspace = true
 sui-macros.workspace = true
@@ -27,6 +28,7 @@ bcs.workspace = true
 bytes.workspace = true
 fastcrypto.workspace = true
 fastcrypto-tbls.workspace = true
+mysten-common.workspace = true
 mysten-network.workspace = true
 tokio = { workspace = true, features = ["full"] }
 tracing.workspace = true

crates/sui-network/build.rs

@@ -150,6 +150,15 @@ fn build_anemo_services(out_dir: &Path) {
                 .codec_path(codec_path)
                 .build(),
         )
+        .method(
+            anemo_build::manual::Method::builder()
+                .name("get_known_peers_v2")
+                .route_name("GetKnownPeersV2")
+                .request_type("()")
+                .response_type("crate::discovery::GetKnownPeersResponseV2")
+                .codec_path(codec_path)
+                .build(),
+        )
         .build();
 
     let state_sync = anemo_build::manual::Service::builder()

crates/sui-network/src/discovery/builder.rs

@@ -7,11 +7,13 @@ use super::{
 use crate::discovery::TrustedPeerChangeEvent;
 use anemo::codegen::InboundRequestLayer;
 use anemo_tower::rate_limit;
+use fastcrypto::traits::KeyPair;
 use std::{
     collections::HashMap,
     sync::{Arc, RwLock},
 };
 use sui_config::p2p::P2pConfig;
+use sui_types::crypto::NetworkKeyPair;
 use tap::Pipe;
 use tokio::{
     sync::{oneshot, watch},
@@ -117,7 +119,11 @@ pub struct UnstartedDiscovery {
 }
 
 impl UnstartedDiscovery {
-    pub(super) fn build(self, network: anemo::Network) -> (DiscoveryEventLoop, Handle) {
+    pub(super) fn build(
+        self,
+        network: anemo::Network,
+        keypair: NetworkKeyPair,
+    ) -> (DiscoveryEventLoop, Handle) {
         let Self {
             handle,
             config,
@@ -146,6 +152,7 @@ impl UnstartedDiscovery {
                 discovery_config: Arc::new(discovery_config),
                 allowlisted_peers,
                 network,
+                keypair,
                 tasks: JoinSet::new(),
                 pending_dials: Default::default(),
                 dial_seed_peers_task: None,
@@ -158,8 +165,9 @@ impl UnstartedDiscovery {
         )
     }
 
-    pub fn start(self, network: anemo::Network) -> Handle {
-        let (event_loop, handle) = self.build(network);
+    pub fn start(self, network: anemo::Network, keypair: NetworkKeyPair) -> Handle {
+        assert_eq!(network.peer_id().0, *keypair.public().0.as_bytes());
+        let (event_loop, handle) = self.build(network, keypair);
         tokio::spawn(event_loop.start());
 
         handle