security(net): Stop sending peer addresses from handshakes directly to the address book (#7977)

* Update Connection debug impl missed in previous PRs

* Add an initial_cached_addrs argument to Connection::new()

* Stop sending version and remote IP peers directly to the address book

* Update zebra-network/src/peer/connection.rs

Co-authored-by: teor <teor@riseup.net>

---------

Co-authored-by: Marek <mail@marek.onl>

zebra-network/src/peer/connection.rs

@@ -541,13 +541,16 @@ where
     /// other state handling.
     pub(super) request_timer: Option<Pin<Box<Sleep>>>,
 
-    /// A cached copy of the last unsolicited `addr` or `addrv2` message from this peer.
+    /// Unused peers from recent `addr` or `addrv2` messages from this peer.
+    /// Also holds the initial addresses sent in `version` messages, or guessed from the remote IP.
     ///
-    /// When Zebra requests peers, the cache is consumed and returned as a synthetic response.
-    /// This works around `zcashd`'s address response rate-limit.
+    /// When peers send solicited or unsolicited peer advertisements, Zebra puts them in this cache.
     ///
-    /// Multi-peer `addr` or `addrv2` messages replace single-peer messages in the cache.
-    /// (`zcashd` also gossips its own address at regular intervals.)
+    /// When Zebra's components request peers, some cached peers are randomly selected,
+    /// consumed, and returned as a modified response. This works around `zcashd`'s address
+    /// response rate-limit.
+    ///
+    /// The cache size is limited to avoid denial of service attacks.
     pub(super) cached_addrs: Vec<MetaAddr>,
 
     /// The `inbound` service, used to answer requests from this connection's peer.
@@ -609,6 +612,7 @@ where
             .field("error_slot", &self.error_slot)
             .field("metrics_label", &self.metrics_label)
             .field("last_metrics_state", &self.last_metrics_state)
+            .field("last_overload_time", &self.last_overload_time)
             .finish()
     }
 }
@@ -617,7 +621,7 @@ impl<S, Tx> Connection<S, Tx>
 where
     Tx: Sink<Message, Error = SerializationError> + Unpin,
 {
-    /// Return a new connection from its channels, services, and shared state.
+    /// Return a new connection from its channels, services, shared state, and metadata.
     pub(crate) fn new(
         inbound_service: S,
         client_rx: futures::channel::mpsc::Receiver<ClientRequest>,
@@ -625,6 +629,7 @@ where
         peer_tx: Tx,
         connection_tracker: ConnectionTracker,
         connection_info: Arc<ConnectionInfo>,
+        initial_cached_addrs: Vec<MetaAddr>,
     ) -> Self {
         let metrics_label = connection_info.connected_addr.get_transient_addr_label();
 
@@ -632,7 +637,7 @@ where
             connection_info,
             state: State::AwaitingRequest,
             request_timer: None,
-            cached_addrs: Vec::new(),
+            cached_addrs: initial_cached_addrs,
             svc: inbound_service,
             client_rx: client_rx.into(),
             error_slot,

zebra-network/src/peer/connection/tests.rs

@@ -83,6 +83,7 @@ fn new_test_connection<A>() -> (
         peer_tx,
         ActiveConnectionCounter::new_counter().track_connection(),
         Arc::new(connection_info),
+        Vec::new(),
     );
 
     (

zebra-network/src/peer/handshake.rs

@@ -29,7 +29,7 @@ use tracing_futures::Instrument;
 use zebra_chain::{
     chain_tip::{ChainTip, NoChainTip},
     parameters::Network,
-    serialization::SerializationError,
+    serialization::{DateTime32, SerializationError},
 };
 
 use crate::{
@@ -915,29 +915,8 @@ where
             )
             .await?;
 
-            let remote_canonical_addr = connection_info.remote.address_from.addr();
             let remote_services = connection_info.remote.services;
 
-            // If we've learned potential peer addresses from an inbound
-            // connection or handshake, add those addresses to our address book.
-            //
-            // # Security
-            //
-            // We must handle alternate addresses separately from connected
-            // addresses. Otherwise, malicious peers could interfere with the
-            // address book state of other peers by providing their addresses in
-            // `Version` messages.
-            //
-            // New alternate peer address and peer responded updates are rate-limited because:
-            // - opening connections is rate-limited
-            // - we only send these messages once per handshake
-            let alternate_addrs = connected_addr.get_alternate_addrs(remote_canonical_addr);
-            for alt_addr in alternate_addrs {
-                let alt_addr = MetaAddr::new_alternate(alt_addr, &remote_services);
-                // awaiting a local task won't hang
-                let _ = address_book_updater.send(alt_addr).await;
-            }
-
             // The handshake succeeded: update the peer status from AttemptPending to Responded,
             // and send initial connection info.
             if let Some(book_addr) = connected_addr.get_address_book_addr() {
@@ -1055,6 +1034,33 @@ where
                 })
                 .boxed();
 
+            // If we've learned potential peer addresses from the inbound connection remote address
+            // or the handshake version message, add those addresses to the peer cache for this
+            // peer.
+            //
+            // # Security
+            //
+            // We can't add these alternate addresses directly to the address book. If we did,
+            // malicious peers could interfere with the address book state of other peers by
+            // providing their addresses in `Version` messages. Or they could fill the address book
+            // with fake addresses.
+            //
+            // These peer addresses are rate-limited because:
+            // - opening connections is rate-limited
+            // - these addresses are put in the peer address cache
+            // - the peer address cache is only used when Zebra requests addresses from that peer
+            let remote_canonical_addr = connection_info.remote.address_from.addr();
+            let alternate_addrs = connected_addr
+                .get_alternate_addrs(remote_canonical_addr)
+                .map(|addr| {
+                    // Assume the connecting node is a server node, and it's available now.
+                    MetaAddr::new_gossiped_meta_addr(
+                        addr,
+                        PeerServices::NODE_NETWORK,
+                        DateTime32::now(),
+                    )
+                });
+
             let server = Connection::new(
                 inbound_service,
                 server_rx,
@@ -1062,6 +1068,7 @@ where
                 peer_tx,
                 connection_tracker,
                 connection_info.clone(),
+                alternate_addrs.collect(),
             );
 
             let connection_task = tokio::spawn(