fix(p2p): judge ipv4-mapped addresses by their embedded ipv4 form

p2p/src/peer_manager/address_groups.rs

@@ -13,9 +13,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::net::{Ipv4Addr, Ipv6Addr};
-
-use crate::types::peer_address::PeerAddress;
+use crate::types::{global_ip::IsGlobalIp, peer_address::PeerAddress};
 
 // IPv4 addresses grouped into /16 subnets
 pub const IPV4_GROUP_BYTES: usize = 2;
@@ -39,20 +37,30 @@ impl AddressGroup {
     ///
     /// See `NetGroupManager::GetGroup` in Bitcoin Core for a reference.
     pub fn from_peer_address(address: &PeerAddress) -> AddressGroup {
-        if address.is_global_unicast_ip() {
-            match address {
-                PeerAddress::Ip4(addr) => AddressGroup::PublicV4(
-                    Ipv4Addr::from(addr.ip).octets()[0..IPV4_GROUP_BYTES]
+        // PeerAddress is not canonicalized at construction, so the reduction to the
+        // embedded IPv4 form of an IPv4-mapped IPv6 address happens here. All checks run
+        // on the canonical form. Note that other IPv4-embedded encodings (6to4, Teredo,
+        // NAT64, IPv4-translated) are not reduced and are grouped as plain IPv6: 6to4
+        // keeps per-/16 granularity of the embedded IPv4 (the key holds the 6to4 prefix
+        // plus its first two octets), while all Teredo (2001:0::/32), all NAT64
+        // (64:ff9b::/96), and all IPv4-translated (::ffff:0:0:0/96) addresses collapse
+        // into one group bucket each, unlike the Teredo-aware grouping of Bitcoin
+        // Core's GetGroup.
+        let std_ip = address.canonical_ip();
+        if std_ip.is_global_unicast_ip() {
+            match std_ip {
+                std::net::IpAddr::V4(ip) => AddressGroup::PublicV4(
+                    ip.octets()[0..IPV4_GROUP_BYTES]
                         .try_into()
-                        .expect("must be valid"),
+                        .expect("group size must not exceed the IPv4 octet count"),
                 ),
-                PeerAddress::Ip6(addr) => AddressGroup::PublicV6(
-                    Ipv6Addr::from(addr.ip).octets()[0..IPV6_GROUP_BYTES]
+                std::net::IpAddr::V6(ip) => AddressGroup::PublicV6(
+                    ip.octets()[0..IPV6_GROUP_BYTES]
                         .try_into()
-                        .expect("must be valid"),
+                        .expect("group size must not exceed the IPv6 octet count"),
                 ),
             }
-        } else if address.is_loopback() {
+        } else if std_ip.is_loopback() {
             AddressGroup::Local
         } else {
             AddressGroup::Private
@@ -91,3 +99,42 @@ mod tests {
         );
     }
 }
+
+#[cfg(test)]
+mod mapped_address_tests {
+    use super::*;
+
+    #[test]
+    fn mapped_addresses_are_grouped_like_the_embedded_ipv4() {
+        use crate::peer_manager::tests::utils::mapped_v6_peer_address;
+
+        assert_eq!(
+            AddressGroup::from_peer_address(&mapped_v6_peer_address(
+                std::net::Ipv4Addr::LOCALHOST,
+                12345
+            )),
+            AddressGroup::Local
+        );
+        assert_eq!(
+            AddressGroup::from_peer_address(&mapped_v6_peer_address(
+                std::net::Ipv4Addr::new(192, 168, 1, 1),
+                12345
+            )),
+            AddressGroup::Private
+        );
+        assert_eq!(
+            AddressGroup::from_peer_address(&mapped_v6_peer_address(
+                std::net::Ipv4Addr::new(10, 0, 0, 1),
+                12345
+            )),
+            AddressGroup::Private
+        );
+        assert_eq!(
+            AddressGroup::from_peer_address(&mapped_v6_peer_address(
+                std::net::Ipv4Addr::new(1, 2, 3, 4),
+                12345
+            )),
+            AddressGroup::PublicV4([1, 2])
+        );
+    }
+}

p2p/src/peer_manager/mod.rs

@@ -1445,6 +1445,9 @@ where
             }
 
             let socket_addr = addr.socket_addr();
+            // SocketAddress is canonical by construction, so a mapped form
+            // (::ffff:a.b.c.d) counts towards the same per-ip connection limit as the
+            // plain ipv4 form.
             cur_conn_ip_port_to_role_map.insert((socket_addr.ip(), socket_addr.port()), role);
         }
 
@@ -2078,8 +2081,9 @@ where
     ) -> bool {
         assert!(new_peer_role.is_outbound());
 
-        let new_peer_ip = new_peer_addr.socket_addr().ip();
-        let new_peer_port = new_peer_addr.socket_addr().port();
+        let new_peer_addr_socket = new_peer_addr.socket_addr();
+        let new_peer_ip = new_peer_addr_socket.ip();
+        let new_peer_port = new_peer_addr_socket.port();
         existing_connections.range((new_peer_ip, 0)..=(new_peer_ip, u16::MAX)).all(
             |((_, existing_peer_port), existing_peer_role)| {
                 self.may_allow_outbound_connection_to_existing_ip(

p2p/src/peer_manager/peerdb/storage.rs

@@ -71,6 +71,15 @@ pub trait PeerDbStorageWrite {
 
     fn add_anchor_address(&mut self, address: &SocketAddress) -> crate::Result<()>;
     fn del_anchor_address(&mut self, address: &SocketAddress) -> crate::Result<()>;
+
+    /// Rewrite rows keyed by the mapped spelling (`::ffff:a.b.c.d`) of an IPv4 address so
+    /// that they are keyed by the canonical form; without this, removals would miss the
+    /// legacy rows and bans could resurrect after a restart.
+    ///
+    /// Returns whether any rows were rewritten. A `false` result guarantees no writes were
+    /// staged on this transaction, so callers may skip committing it; a `true` result means
+    /// the transaction must be committed for the rewrites to take effect.
+    fn normalize_legacy_mapped_address_keys(&mut self) -> crate::Result<bool>;
 }
 
 // Note: here we want to say something like:

p2p/src/peer_manager/peerdb/storage_impl.rs

@@ -124,6 +124,172 @@ impl<B: storage::SharedBackend> PeerDbStorageWrite for PeerDbStoreTxRw<'_, B> {
     fn del_anchor_address(&mut self, address: &SocketAddress) -> crate::Result<()> {
         Ok(self.storage().get_mut::<DBAnchorAddresses, _>().del(address.to_string())?)
     }
+
+    fn normalize_legacy_mapped_address_keys(&mut self) -> crate::Result<bool> {
+        fn canonical_spelling<T: std::str::FromStr + std::fmt::Display>(
+            table: &'static str,
+            key: &str,
+            prefix: &str,
+        ) -> LegacyKeyAction
+        where
+            <T as std::str::FromStr>::Err: std::fmt::Display,
+        {
+            // Legacy mapped spellings always start with the prefix, so the parse can
+            // be skipped for the vast majority of keys on every startup. Socket
+            // addresses were persisted through their Display form, which brackets
+            // IPv6 addresses.
+            if !key.starts_with(prefix) {
+                return LegacyKeyAction::Keep;
+            }
+            let address = match key.parse::<T>() {
+                Ok(address) => address,
+                Err(err) => {
+                    logging::log::warn!(
+                        "Dropping unparseable legacy {table} key {key}: {err}; the row \
+                         cannot be read by this version"
+                    );
+                    return LegacyKeyAction::Drop;
+                }
+            };
+            let canonical = address.to_string();
+            if canonical == key {
+                LegacyKeyAction::Keep
+            } else {
+                LegacyKeyAction::Rewrite(canonical)
+            }
+        }
+
+        const MAPPED_SOCKET_ADDRESS_PREFIX: &str = "[::ffff:";
+        const MAPPED_BANNABLE_ADDRESS_PREFIX: &str = "::ffff:";
+
+        let storage = self.storage();
+        let mut rewrote_any = false;
+
+        rewrote_any |= rekey_mapped_rows(
+            &mut storage.get_mut::<DBKnownAddresses, _>(),
+            |key| {
+                canonical_spelling::<SocketAddress>(
+                    "known addresses",
+                    key,
+                    MAPPED_SOCKET_ADDRESS_PREFIX,
+                )
+            },
+            // A legacy row marked as tried keeps its selection preference.
+            |canonical, legacy| match legacy {
+                KnownAddressState::Tried => legacy,
+                KnownAddressState::New => canonical,
+            },
+        )?;
+        rewrote_any |= rekey_mapped_rows(
+            &mut storage.get_mut::<DBAnchorAddresses, _>(),
+            |key| {
+                canonical_spelling::<SocketAddress>(
+                    "anchor addresses",
+                    key,
+                    MAPPED_SOCKET_ADDRESS_PREFIX,
+                )
+            },
+            |_canonical, _legacy| (),
+        )?;
+        rewrote_any |= rekey_mapped_rows(
+            &mut storage.get_mut::<DBBannedAddresses, _>(),
+            |key| {
+                canonical_spelling::<BannableAddress>(
+                    "banned addresses",
+                    key,
+                    MAPPED_BANNABLE_ADDRESS_PREFIX,
+                )
+            },
+            // The later unban time wins so the migration cannot shorten a ban.
+            std::cmp::max,
+        )?;
+        rewrote_any |= rekey_mapped_rows(
+            &mut storage.get_mut::<DBDiscouragedAddresses, _>(),
+            |key| {
+                canonical_spelling::<BannableAddress>(
+                    "discouraged addresses",
+                    key,
+                    MAPPED_BANNABLE_ADDRESS_PREFIX,
+                )
+            },
+            std::cmp::max,
+        )?;
+
+        Ok(rewrote_any)
+    }
+}
+
+/// What to do with a key during the legacy fold.
+enum LegacyKeyAction {
+    // Not a legacy mapped-form key; leave the row alone.
+    Keep,
+    // A legacy mapped-form key that no longer parses; drop the row. It cannot
+    // be loaded by the regular read path either, so keeping it would fail the
+    // startup load on every run.
+    Drop,
+    // Rewrite the row under the canonical spelling of the same address.
+    Rewrite(String),
+}
+
+/// Rewrite rows keyed by a non-canonical spelling of an IPv4 address (the mapped form
+/// `::ffff:a.b.c.d` written by an older version) so that the key matches the canonical
+/// spelling used at runtime, merging the value with an existing canonical row via
+/// `merge` when one is already present. Rows whose key no longer parses are dropped.
+fn rekey_mapped_rows<B, DBM>(
+    map: &mut storage::MapMut<'_, storage::TransactionRw<'_, B, Schema>, DBM>,
+    key_action: impl Fn(&str) -> LegacyKeyAction,
+    merge: impl Fn(DBM::Value, DBM::Value) -> DBM::Value,
+) -> crate::Result<bool>
+where
+    B: storage::SharedBackend,
+    DBM: storage::schema::DbMap<Key = String>,
+    DBM::Value: serialization::Encode + serialization::EncodeLike,
+{
+    let mut rewrites = Vec::new();
+    let mut drops = Vec::new();
+    for (key, encoded) in map.prefix_iter(&())? {
+        match key_action(&key) {
+            LegacyKeyAction::Keep => {}
+            LegacyKeyAction::Drop => drops.push(key),
+            LegacyKeyAction::Rewrite(canonical) => {
+                // Decode explicitly so a corrupt row is reported instead of panicking.
+                let value =
+                    <DBM::Value as serialization::DecodeAll>::decode_all(&mut encoded.as_ref())
+                        .map_err(|err| {
+                            P2pError::InvalidStorageState(format!(
+                                "Failed to decode the value of the row keyed by {key}: {err}"
+                            ))
+                        })?;
+                rewrites.push((key, canonical, value));
+            }
+        }
+    }
+    let mut rewrote_any = false;
+    for legacy_key in drops {
+        map.del(&legacy_key)?;
+        rewrote_any = true;
+    }
+    for (legacy_key, canonical_key, legacy_value) in rewrites {
+        map.del(&legacy_key)?;
+        let value = match map.get(&canonical_key)? {
+            Some(existing) => {
+                // Decode explicitly so a corrupt canonical row is reported too.
+                let existing_value = <DBM::Value as serialization::DecodeAll>::decode_all(
+                    &mut existing.as_ref(),
+                )
+                .map_err(|err| {
+                    P2pError::InvalidStorageState(format!(
+                        "Failed to decode the value of the row keyed by {canonical_key}: {err}"
+                    ))
+                })?;
+                merge(existing_value, legacy_value)
+            }
+            None => legacy_value,
+        };
+        map.put(&canonical_key, value)?;
+        rewrote_any = true;
+    }
+    Ok(rewrote_any)
 }
 
 impl<B: storage::SharedBackend> PeerDbStorageRead for PeerDbStoreTxRo<'_, B> {

p2p/src/peer_manager/peerdb/storage_load.rs

@@ -83,6 +83,19 @@ impl LoadedStorage {
     }
 
     fn load_storage_v3<S: PeerDbStorage>(storage: &S) -> crate::Result<LoadedStorage> {
+        {
+            // Rewrite any rows written by an older version under the mapped spelling of an
+            // IPv4 address, so the stored keys match the canonical form used at runtime.
+            let mut write_tx = storage.transaction_rw()?;
+            // Commit only when rows were actually rewritten, so the common case of an
+            // already-canonical database stays read-only.
+            if write_tx.normalize_legacy_mapped_address_keys()? {
+                write_tx.commit()?;
+            }
+            // TODO: once older databases are no longer expected, replace this
+            // every-startup scan with a one-time versioned migration or a stored marker.
+        }
+
         let tx = storage.transaction_ro()?;
 
         let known_addresses = tx.get_known_addresses()?.into_iter().collect::<BTreeMap<_, _>>();