diff --git a/docs/manual/src/config-reference.md b/docs/manual/src/config-reference.md index a3f64644a..75cadd6f4 100644 --- a/docs/manual/src/config-reference.md +++ b/docs/manual/src/config-reference.md @@ -189,7 +189,7 @@ startup error. | `v2transport` | true | hot | core | Offer/accept BIP 324 v2 encrypted transport (Core default since v26). | | `v2only` | false | hot | satd | Refuse peers that do not speak BIP 324 v2 (privacy hardening). | | `externalip` | none | hot | core | External address to advertise to peers (repeatable). | -| `whitelist` | none | hot | core | Grant net permissions to peers by source subnet (repeatable). `[@]`. **Inbound only unless the permission list carries `out`**, as in Core; an `out` entry applies only to *manual* outbound connections (`connect`, `addnode`, and `seednode`, which satd keeps as a manual connection), never to automatic ones such as the peers the node finds through gossip, `peers.dat` or a DNS seed. `@` with an empty list matches the range and grants nothing. A peer arriving over the Tor hidden service is never matched against this — it reaches the node on a loopback socket Tor forwards to, so matching would grant an anonymous remote peer whatever you granted your own machine. | +| `whitelist` | none | hot | core | Grant net permissions to peers by source subnet (repeatable). `[@]`. **Inbound only unless the permission list carries `out`**, as in Core; an `out` entry applies only to *manual* outbound connections (`connect`, `addnode`, and `seednode`, which satd keeps as a manual connection), never to automatic ones such as the peers the node finds through gossip, `peers.dat` or a DNS seed. `@` with an empty list matches the range and grants nothing. A peer arriving over the Tor hidden service is never matched against this — it reaches the node on a loopback socket Tor forwards to, so matching would grant an anonymous remote peer whatever you granted your own machine. The `addr` permission exempts a peer from Core's address rate limit: otherwise the node takes one address from a peer's `addr`/`addrv2` messages to start and one more every 10 seconds (a thousand more after it sends the peer `getaddr`), and `getpeerinfo` reports the rest as `addr_rate_limited`. | | `whitelistrelay` | on | hot | core | Grant `relay` to whitelisted peers with default permissions (relay their txes even under `-blocksonly`). Entries with an explicit `perms@` prefix are unaffected. | | `whitelistforcerelay` | off | hot | core | Grant `forcerelay` to whitelisted peers with default permissions. Entries with an explicit `perms@` prefix are unaffected. | | `whitebind` | none | restart | core | Bind an extra permissioned P2P listener (repeatable). | @@ -197,10 +197,10 @@ startup error. | `port` | network default | restart | core | P2P listen port. | | `bind` | `0.0.0.0` | restart | core | Bind P2P to this address; **repeatable**. Accepts `addr`, `addr:port`, and `addr[:port]=onion`. An entry with a port uses it; a bare address takes `port`, and a bare `=onion` entry takes `port` + 1 (as in Core). IPv6 literals may be plain (`::1`) or bracketed. Any explicit `bind` replaces the default listener. Cannot be combined with `listen=0`. | | `connect` | none | hot | core | Connect only to specific peer(s) (repeatable). An entry with no port takes the network's default P2P port. Any `connect` stops the node dialling addresses it learns from gossip and soft-sets `listen=0` (see `listen`); `connect=0` — and the command-line negation `-noconnect` — is Core's spelling for "open no outbound connections at all"; a literal `connect=0` mixed with a peer address is refused, since Core keeps the peer and satd will not dial the `0`. Connect-only *exclusivity* is a startup decision (restart to change). | -| `addnode` | none | hot | core | Add a node to connect to (does not disable DNS seeding, and does not affect `listen`). An entry with no port takes the network's default P2P port. | +| `addnode` | none | hot | core | Add a node to connect to (does not disable DNS seeding, and does not affect `listen`). An entry with no port takes the network's default P2P port. As in Core, added nodes are counted apart from the automatic outbound peers and `maxconnections`: at most 8 are connected at a time, and one that is not connected is dialled again at most a minute later, however many automatic peers the node has. A ban does not stop the dial; it does stop automatic dials to that address. | | `uacomment` | none | restart | core | Append a comment to the advertised user agent (**repeatable**; command-line and config-file values accumulate, command line first). Renders as `/satd:(c1; c2)/`. A comment may contain only alphanumerics and `` .,;-_?@`` — the user agent's own delimiters `/`, `:`, `(` and `)` are refused — and the whole user agent may not exceed 256 bytes. Either violation is a startup error, as in Core. | | `seednode` | none | hot | core | One-shot seed peer connected at startup to bootstrap discovery. | -| `maxconnections` | 125 | hot | core | Maximum total connections. `0` (or any value ≤ 0) soft-sets `listen=0`, as in Core — see `listen`. That half is a startup decision: changing `maxconnections` over SIGHUP applies the new cap but does not start or stop the listener. | +| `maxconnections` | 125 | hot | core | Maximum total connections. Manual connections (`connect`, `addnode`, `seednode`, `addnode onetry`) do not count against the automatic outbound peers, as in Core. `0` (or any value ≤ 0) soft-sets `listen=0`, as in Core — see `listen`. That half is a startup decision: changing `maxconnections` over SIGHUP applies the new cap but does not start or stop the listener. | | `maxinboundperip` | 3 | hot | satd | Max simultaneous inbound peers from one source IP (Core-style flood guard; no Core flag). | | `cmpctblockprefill` | false | restart | satd | Announce a new block as a `cmpctblock` with the transactions this node lacked when it arrived prefilled, so a peer that lacks them too rebuilds the block without a `getblocktxn` round trip. Bitcoin Core's proposed design (#35558, not yet merged in Core); off until measurement says otherwise. | | `cmpctblockprefillbytes` | 8192 | restart | satd | Transaction bytes a `cmpctblockprefill` announcement may carry beyond the coinbase. Transactions nobody relayed to this node go first, then replaced or policy-refused ones; one that does not fit is skipped. `0` prefills nothing. | diff --git a/node/src/net/addr_rate.rs b/node/src/net/addr_rate.rs new file mode 100644 index 000000000..28d468115 --- /dev/null +++ b/node/src/net/addr_rate.rs @@ -0,0 +1,142 @@ +//! Bitcoin Core's rate limit on the addresses a peer announces +//! (`net_processing.cpp` `ProcessAddrs`). +//! +//! Every peer has a token bucket. It starts with one token and gains +//! [`MAX_ADDR_RATE_PER_SECOND`] tokens a second, up to +//! [`MAX_ADDR_PROCESSING_TOKEN_BUCKET`]. Each address taken from the peer's +//! `addr` and `addrv2` messages spends one token; an address that finds the +//! bucket empty is dropped, unless the peer has the `addr` permission. +//! Sending the peer a `getaddr` adds [`MAX_ADDR_TO_SEND`] tokens on top, +//! past that limit, so its answer is taken whole. + +use crate::net::limits::MAX_ADDR_TO_SEND; + +/// Core's `MAX_ADDR_RATE_PER_SECOND`: tokens gained per second. +pub const MAX_ADDR_RATE_PER_SECOND: f64 = 0.1; + +/// Core's `MAX_ADDR_PROCESSING_TOKEN_BUCKET`: the bucket refills no +/// further than this. +pub const MAX_ADDR_PROCESSING_TOKEN_BUCKET: f64 = MAX_ADDR_TO_SEND as f64; + +/// One peer's address token bucket and its counters: Core's +/// `Peer::m_addr_token_bucket`, `m_addr_token_timestamp`, +/// `m_addr_processed` and `m_addr_rate_limited`. +#[derive(Debug, Clone)] +pub struct AddrTokenBucket { + tokens: f64, + /// Node time of the last refill, in microseconds + /// (`crate::time::now_micros`). + refilled_at_micros: u64, + /// Addresses from this peer that passed the limit and were looked at + /// (`getpeerinfo.addr_processed`). + pub processed: u64, + /// Addresses from this peer dropped by the limit + /// (`getpeerinfo.addr_rate_limited`). + pub rate_limited: u64, +} + +impl AddrTokenBucket { + /// A new peer's bucket: one token, as in Core. + pub fn new(now_micros: u64) -> Self { + Self { + tokens: 1.0, + refilled_at_micros: now_micros, + processed: 0, + rate_limited: 0, + } + } + + /// Add the tokens earned since the last refill, unless the bucket is + /// already at [`MAX_ADDR_PROCESSING_TOKEN_BUCKET`], and restart the clock + /// either way. Core does this once per message, before taking any of it. + /// A clock that went backwards earns nothing. + pub fn refill(&mut self, now_micros: u64) { + if self.tokens < MAX_ADDR_PROCESSING_TOKEN_BUCKET { + let elapsed_secs = now_micros.saturating_sub(self.refilled_at_micros) as f64 / 1_000_000.0; + self.tokens = (self.tokens + elapsed_secs * MAX_ADDR_RATE_PER_SECOND) + .min(MAX_ADDR_PROCESSING_TOKEN_BUCKET); + } + self.refilled_at_micros = now_micros; + } + + /// Spend a token on one address. Without one the address is dropped + /// and counted as rate limited, unless `exempt` (the `addr` + /// permission), in which case it is taken for free. + pub fn admit(&mut self, exempt: bool) -> bool { + if self.tokens < 1.0 { + if !exempt { + self.rate_limited += 1; + return false; + } + } else { + self.tokens -= 1.0; + } + true + } + + /// A `getaddr` was sent to the peer: room for its answer, past the + /// refill limit. + pub fn getaddr_sent(&mut self) { + self.tokens += MAX_ADDR_TO_SEND as f64; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const SEC: u64 = 1_000_000; + + fn take(bucket: &mut AddrTokenBucket, n: usize, exempt: bool) -> usize { + (0..n).filter(|_| bucket.admit(exempt)).count() + } + + #[test] + fn a_new_peer_gets_one_address_then_one_every_ten_seconds() { + let mut b = AddrTokenBucket::new(0); + b.refill(0); + assert_eq!(take(&mut b, 1000, false), 1); + assert_eq!(b.rate_limited, 999); + + b.refill(100 * SEC); + assert_eq!(take(&mut b, 1000, false), 10, "0.1 a second"); + + b.refill(105 * SEC); + assert_eq!(take(&mut b, 1, false), 0, "half a token is not one"); + b.refill(110 * SEC); + assert_eq!(take(&mut b, 1, false), 1, "the half carried over"); + } + + #[test] + fn the_bucket_refills_to_a_thousand_and_no_further() { + let mut b = AddrTokenBucket::new(0); + b.refill(1_000_000 * SEC); + assert_eq!(take(&mut b, 2000, false), 1000); + } + + #[test] + fn a_getaddr_makes_room_for_its_answer_past_the_limit() { + let mut b = AddrTokenBucket::new(0); + b.refill(1_000_000 * SEC); + b.getaddr_sent(); + assert_eq!(take(&mut b, 3000, false), 2000); + } + + #[test] + fn the_addr_permission_takes_every_address() { + let mut b = AddrTokenBucket::new(0); + b.refill(0); + assert_eq!(take(&mut b, 1000, true), 1000); + assert_eq!(b.rate_limited, 0); + } + + #[test] + fn a_clock_that_goes_backwards_earns_nothing() { + let mut b = AddrTokenBucket::new(100 * SEC); + b.refill(0); + assert_eq!(take(&mut b, 10, false), 1, "only the first token"); + // And the clock restarted from the earlier time, as Core's does. + b.refill(10 * SEC); + assert_eq!(take(&mut b, 10, false), 1); + } +} diff --git a/node/src/net/addrman.rs b/node/src/net/addrman.rs index 6e62d618c..0dd963226 100644 --- a/node/src/net/addrman.rs +++ b/node/src/net/addrman.rs @@ -38,7 +38,7 @@ use std::path::Path; /// Hard cap on total stored addresses (new + tried). Keeps `peers.dat` /// and memory bounded; well above what a single node needs to bootstrap. -const MAX_ENTRIES: usize = 16_384; +pub const MAX_ENTRIES: usize = 16_384; /// Per-network-group cap in the *new* table (eclipse resistance: stop one /// `/16`/ASN from flooding the book with addresses). const MAX_NEW_PER_GROUP: usize = 64; @@ -92,6 +92,29 @@ pub struct AddrEntry { group: Vec, } +/// Core's `ADDRMAN_HORIZON`: an address not seen for this long is terrible. +const ADDRMAN_HORIZON_SECS: u64 = 30 * 24 * 60 * 60; +/// Core's `ADDRMAN_RETRIES`: failed attempts, with no success ever, that +/// make an address terrible. +const ADDRMAN_RETRIES: u32 = 3; +/// Core's `ADDRMAN_MAX_FAILURES` within `ADDRMAN_MIN_FAIL` (7 days). +const ADDRMAN_MAX_FAILURES: u32 = 10; +const ADDRMAN_MIN_FAIL_SECS: u64 = 7 * 24 * 60 * 60; + +impl AddrEntry { + /// Core's `AddrInfo::IsTerrible`, without its one-minute grace after an + /// attempt (satd does not record when it last tried): seen in the + /// future, not seen for 30 days, tried three times and never reached, + /// or ten failures with no success in a week. + pub fn is_terrible(&self, now: u64) -> bool { + self.last_seen > now + 10 * 60 + || now.saturating_sub(self.last_seen) > ADDRMAN_HORIZON_SECS + || (self.last_success == 0 && self.attempts >= ADDRMAN_RETRIES) + || (now.saturating_sub(self.last_success) > ADDRMAN_MIN_FAIL_SECS + && self.attempts >= ADDRMAN_MAX_FAILURES) + } +} + /// Network group key used for bucketing. IPv4 → the `/16`; IPv6 → the /// `/32`. A custom grouping (e.g. `-asmap` ASN) can be installed via /// [`AddrMan::set_group_fn`]. @@ -251,19 +274,62 @@ impl AddrMan { /// of the peer that announced it. A refresh of a known address updates /// its last-seen time only, as before. pub fn add_from(&mut self, addr: SocketAddr, now: u64, services: u64, source: IpAddr) -> bool { + self.add_from_replacing(addr, now, services, source).0 + } + + /// [`add_from`](Self::add_from), and when the book or the address's + /// network group is full, make room for the new address by dropping one + /// entry (of that group and new, when the group is full; new before + /// tried otherwise) that is terrible + /// ([`AddrEntry::is_terrible`]), as Core's address manager replaces + /// terrible entries. Returns whether `addr` was inserted, and the + /// address dropped for it. + /// + /// Without this a full book refused every address it did not already + /// hold, for good: nothing ever left it. + pub fn add_from_replacing( + &mut self, + addr: SocketAddr, + now: u64, + services: u64, + source: IpAddr, + ) -> (bool, Option) { if !self.admits(&addr) { - return false; + return (false, None); } if let Some(e) = self.entries.get_mut(&addr) { e.last_seen = now; - return false; - } - if self.entries.len() >= MAX_ENTRIES { - return false; + return (false, None); } let group = self.group_of(&addr); + let mut evicted = None; if self.new_count_in_group(&group) >= MAX_NEW_PER_GROUP { - return false; + // The group is full: make room in it, or refuse. + let Some(victim) = self + .entries + .values() + .find(|e| !e.tried && e.group == group && e.is_terrible(now)) + .map(|e| e.addr) + else { + return (false, None); + }; + self.entries.remove(&victim); + evicted = Some(victim); + } else if self.entries.len() >= MAX_ENTRIES { + // A terrible new entry first, then a terrible tried one: Core + // keeps the two tables apart, so a full tried table never shuts + // new addresses out there; one shared cap must not either. + let terrible = |tried: bool| { + self.entries + .values() + .find(|e| e.tried == tried && e.is_terrible(now)) + .map(|e| e.addr) + }; + let Some(victim) = terrible(false).or_else(|| terrible(true)) else { + return (false, None); + }; + self.entries.remove(&victim); + evicted = Some(victim); } self.entries.insert( addr, @@ -278,7 +344,7 @@ impl AddrMan { group, }, ); - true + (true, evicted) } /// Promote an address to the *tried* table after a successful connect. @@ -317,6 +383,11 @@ impl AddrMan { entry.attempts = 0; } + /// Whether the book holds `addr`, in either table. + pub fn contains(&self, addr: &SocketAddr) -> bool { + self.entries.contains_key(addr) + } + /// Record a failed connection attempt. pub fn mark_attempt(&mut self, addr: SocketAddr) { if let Some(e) = self.entries.get_mut(&addr) { diff --git a/node/src/net/manager.rs b/node/src/net/manager.rs index 96b6e83cf..42ba23abc 100644 --- a/node/src/net/manager.rs +++ b/node/src/net/manager.rs @@ -97,6 +97,23 @@ const MAX_ONION_DIALS_PER_TICK: usize = 16; /// attempts once a minute (`ThreadOpenAddedConnections`). const MANUAL_TARGET_RELOOKUP: Duration = Duration::from_secs(60); +/// Core's `MAX_ADDNODE_CONNECTIONS` (`net.h`): added nodes (`-addnode`, +/// `addnode add`) connected or being connected at once. Counted apart from +/// the automatic outbound target (`semAddnode`). +const MAX_ADDNODE_CONNECTIONS: usize = 8; + +/// The longest a peer the operator named waits between two dials. Core's +/// `ThreadOpenAddedConnections` tries its added nodes again every 60 +/// seconds; automatic candidates back off to 300 seconds. +const MANUAL_RECONNECT_MAX_BACKOFF: Duration = Duration::from_secs(60); + +/// The longest an automatic dial candidate waits between two dials. +const AUTOMATIC_RECONNECT_MAX_BACKOFF: Duration = Duration::from_secs(300); + +/// Cap on the automatic dial list. Only an address the address book holds +/// gets on it (seeds aside), so this is the book's own size. +const MAX_LEARNED_ADDRS: usize = crate::net::addrman::MAX_ENTRIES; + /// A peer the operator named (`-addnode`, `addnode add`, or a `-connect` /// host name), with the address it currently resolves to. #[derive(Debug, Clone)] @@ -394,15 +411,17 @@ impl ReconnectState { } } - /// Backoff delay: 10s, 20s, 40s, 80s, 160s, capped at 300s. - fn backoff_duration(&self) -> Duration { + /// Backoff delay: 10s, 20s, 40s, 80s, 160s, capped at `max` + /// ([`MANUAL_RECONNECT_MAX_BACKOFF`] or + /// [`AUTOMATIC_RECONNECT_MAX_BACKOFF`]). + fn backoff_duration(&self, max: Duration) -> Duration { let secs = 10u64.saturating_mul(1u64 << self.attempts.min(5)); - Duration::from_secs(secs.min(300)) + Duration::from_secs(secs).min(max) } - fn record_failure(&mut self) { + fn record_failure(&mut self, max: Duration) { self.attempts = self.attempts.saturating_add(1); - self.next_attempt = Instant::now() + self.backoff_duration(); + self.next_attempt = Instant::now() + self.backoff_duration(max); } fn reset(&mut self) { @@ -411,6 +430,88 @@ impl ReconnectState { } } +/// The addresses the node learned for itself, as the reconnect loop dials +/// them: in the order they arrived, each once, at most +/// [`MAX_LEARNED_ADDRS`]. This was a plain `Vec` searched end to end on +/// every insert, with no cap. +#[derive(Debug, Default)] +struct LearnedAddrs { + order: Vec, + members: HashSet, +} + +impl LearnedAddrs { + /// Append `addr`. False, changing nothing, if it is already on the + /// list or the list is full. + fn push(&mut self, addr: SocketAddr) -> bool { + if self.members.len() >= MAX_LEARNED_ADDRS || !self.members.insert(addr) { + return false; + } + self.order.push(addr); + true + } + + /// Drop `addr` from the list. + fn remove(&mut self, addr: &SocketAddr) { + if self.members.remove(addr) { + self.order.retain(|a| a != addr); + } + } + + fn contains(&self, addr: &SocketAddr) -> bool { + self.members.contains(addr) + } + + fn iter(&self) -> impl Iterator { + self.order.iter() + } +} + +/// The limit an outbound dial counts against, from the moment it passes the +/// capacity check until its peer is in the peer map, where it is counted +/// instead. Core's counting semaphores: a grant is taken before the dial. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum DialBudget { + /// The node's own connections: the reconnect loop's and the seeds'. + /// Together with the automatic peers already connected, at most the + /// automatic outbound target (8, or 64 during IBD). Core's + /// `semOutbound`. + Automatic, + /// An added node (`-addnode`, `addnode add`): at most + /// [`MAX_ADDNODE_CONNECTIONS`] at once, apart from the automatic target. + /// Core's `semAddnode`. + AddedNode, + /// No limit: the other manual connections (`-connect`, `-seednode`), + /// which Core opens without a grant. + Unlimited, +} + +/// Outbound dials in flight, per [`DialBudget`]. +#[derive(Debug, Default)] +struct DialsInFlight { + automatic: usize, + added_node: usize, +} + +/// A dial's place in its [`DialBudget`], given back when dropped. Owned, so +/// a reservation taken in the reconnect loop can travel with the dial it +/// spawns. +struct DialSlot { + pm: Arc, + budget: DialBudget, +} + +impl Drop for DialSlot { + fn drop(&mut self) { + let mut in_flight = self.pm.dials_in_flight.lock(); + match self.budget { + DialBudget::Automatic => in_flight.automatic = in_flight.automatic.saturating_sub(1), + DialBudget::AddedNode => in_flight.added_node = in_flight.added_node.saturating_sub(1), + DialBudget::Unlimited => {} + } + } +} + /// Event sent from peer tasks to the central manager loop. pub enum NetEvent { PeerConnected { @@ -534,6 +635,10 @@ mod p2pbounds_tests; #[path = "manager_peerpunish_tests.rs"] mod peerpunish_tests; +#[cfg(test)] +#[path = "manager_addrdials_tests.rs"] +mod addrdials_tests; + /// How a transaction is announced to a peer: `MSG_WTX` carrying the wtxid /// to a peer that negotiated BIP 339 wtxid relay, the txid to the rest. fn tx_announcement(wtxid_relay: bool, txid: bitcoin::Txid, wtxid: bitcoin::Wtxid) -> Inventory { @@ -770,13 +875,13 @@ pub struct PeerManager { /// Addresses the node learned for itself: `addr`/`addrv2` gossip, the /// address book at startup, DNS seeds. The reconnect loop dials them only /// while `automatic_outbound` is on, as `outbound-full-relay` - /// connections. Only addresses the address book would take get here - /// (`AddrMan::admits`). + /// connections. A gossiped address gets here only if the address book + /// holds it, a seed only if the book would take it (`AddrMan::admits`). /// /// Kept apart from `manual_addrs`. This list used to hold both, and /// `spawn_peer` called any peer on it `manual`, so every peer reached /// through gossip was reported, whitelisted and counted as one (#866). - learned_addrs: RwLock>, + learned_addrs: RwLock, /// Bitcoin Core's `m_use_addrman_outgoing`. False under `-connect`. automatic_outbound: std::sync::atomic::AtomicBool, /// The clearnet peers the operator named: `-connect`, `-addnode` / @@ -927,6 +1032,10 @@ pub struct PeerManager { /// onion equivalent, stopping the reconnect loop from opening a second /// connection to an onion peer it's already dialing/connected to. pending_onion_dials: RwLock>, + /// Outbound dials in flight, per [`DialBudget`]: counted against the + /// automatic outbound target and the added-node limit along with the + /// peers already connected. See [`Self::reserve_dial`]. + dials_in_flight: parking_lot::Mutex, /// Ban duration in seconds (default: 86400). Atomic for live SIGHUP /// reload (applies to bans created after the change). ban_duration_secs: AtomicU64, @@ -1150,7 +1259,7 @@ impl PeerManager { compact_in_progress: RwLock::new(HashMap::new()), block_refetch: RwLock::new(HashMap::new()), unreadable_refetch_at: parking_lot::Mutex::new(HashMap::new()), - learned_addrs: RwLock::new(Vec::new()), + learned_addrs: RwLock::new(LearnedAddrs::default()), automatic_outbound: std::sync::atomic::AtomicBool::new(true), dns_enabled: std::sync::atomic::AtomicBool::new(true), manual_addrs: RwLock::new(HashSet::new()), @@ -1194,6 +1303,7 @@ impl PeerManager { max_inbound_per_ip: AtomicUsize::new(max_inbound_per_ip), pending_connections: RwLock::new(HashSet::new()), pending_onion_dials: RwLock::new(HashSet::new()), + dials_in_flight: parking_lot::Mutex::new(DialsInFlight::default()), ban_duration_secs: AtomicU64::new(ban_duration_secs), connect_timeout_ms: AtomicU64::new(DEFAULT_CONNECT_TIMEOUT_MS), peer_connect_timeout_secs: AtomicU64::new(DEFAULT_PEER_CONNECT_TIMEOUT_SECS as u64), @@ -1822,13 +1932,29 @@ impl PeerManager { // Linux dials the local host, or, off regtest, one that is not // publicly routable. Core refuses both in `AddrManImpl::AddSingle` // (v31.1 `src/addrman.cpp:554`) and dials only from its book. + // + // Nor does an address the book admits but has no room for (its + // per-group or total cap). Pushing that too put every gossiped + // address on the dial list, past the bounds the book keeps. An + // address the book already holds is pushed again; the list holds + // each address once. { let mut book = self.addrman.write(); if !book.admits(&addr) { tracing::debug!(%addr, "Ignoring a gossiped address the address book does not take"); return; } - book.add_from(addr, now_unix_secs(), services, source); + let (inserted, evicted) = book.add_from_replacing(addr, now_unix_secs(), services, source); + if !inserted && !book.contains(&addr) { + tracing::debug!(%addr, "Not dialling a gossiped address the address book has no room for"); + return; + } + // A full book makes room by dropping a terrible entry; it + // leaves the dial list with it. + if let Some(gone) = evicted { + self.learned_addrs.write().remove(&gone); + self.reconnect_backoff.write().remove(&gone); + } } self.push_learned(addr); } @@ -1862,10 +1988,7 @@ impl PeerManager { if !self.automatic_outbound.load(Ordering::Relaxed) { return; } - let mut addrs = self.learned_addrs.write(); - if !addrs.contains(&addr) { - addrs.push(addr); - } + self.learned_addrs.write().push(addr); } /// Bitcoin Core's `m_use_addrman_outgoing`: whether the node may dial @@ -1909,6 +2032,11 @@ impl PeerManager { if self.is_onion_connected(&host) { return; } + // Core looks no further at a banned address it is told about, as the + // clearnet arms do; the reconnect loop would not dial it either. + if self.ban_list.read().is_onion_banned(&host, crate::time::now_secs()) { + return; + } let mut addrs = self.connect_peer_addrs.write(); let already = addrs .iter() @@ -1935,9 +2063,7 @@ impl PeerManager { // not been applied yet, and `may_dial` holds these back under it. let mut addrs = self.learned_addrs.write(); for a in picks { - if !addrs.contains(&a) { - addrs.push(a); - } + addrs.push(a); } tracing::info!(loaded = self.addrman.read().len(), "Loaded persistent address book"); } @@ -2398,6 +2524,50 @@ impl PeerManager { .count() } + /// Core's address rate limit (`ProcessAddrs`; see + /// [`crate::net::addr_rate`]): refill the peer's token bucket, then keep + /// the entries of its `addr` / `addrv2` that find a token, taken in a + /// random order so the sender does not choose which ones get through. A + /// peer with the `addr` permission keeps them all. + /// + /// satd took every entry of every message: any peer could put up to a + /// thousand addresses per message into the address book and onto the + /// dial list, as often as it liked. + fn rate_limit_addrs(&self, id: PeerId, mut addrs: Vec) -> Vec { + use rand::seq::SliceRandom; + addrs.shuffle(&mut rand::thread_rng()); + let mut peers = self.peers.write(); + let Some(handle) = peers.get_mut(&id) else { + return Vec::new(); + }; + let exempt = handle.info.permissions.addr; + let bucket = &mut handle.info.addr_rate; + bucket.refill(crate::time::now_micros()); + let before = bucket.rate_limited; + addrs.retain(|_| bucket.admit(exempt)); + let limited = bucket.rate_limited - before; + if limited > 0 { + tracing::debug!("Rate-limited {limited} addresses from peer={id}"); + } + addrs + } + + /// Count `n` addresses from peer `id` as processed + /// (`getpeerinfo.addr_processed`). + fn note_addrs_processed(&self, id: PeerId, n: u64) { + if let Some(handle) = self.peers.write().get_mut(&id) { + handle.info.addr_rate.processed += n; + } + } + + /// Note a `getaddr` sent to peer `id`: its answer may carry another + /// `MAX_ADDR_TO_SEND` addresses past the rate limit, as in Core. + fn note_getaddr_sent(&self, id: PeerId) { + if let Some(handle) = self.peers.write().get_mut(&id) { + handle.info.addr_rate.getaddr_sent(); + } + } + /// An addr-fetch connection exists to collect one batch of addresses and /// go. Core requires more than one entry before treating the answer as /// complete, so a peer that only announces itself does not end the @@ -2510,18 +2680,122 @@ impl PeerManager { .min(max_connections.load(Ordering::Relaxed)) } - /// Check outbound connection limit. - fn check_outbound_limit(&self) -> Result<(), String> { - let max_outbound = if self.is_ibd() { + /// The automatic outbound target: 8, capped by `-maxconnections`, or 64 + /// during IBD. + fn automatic_outbound_target(&self) -> usize { + if self.is_ibd() { MAX_OUTBOUND_IBD } else { self.max_connections.load(Ordering::Relaxed).min(MAX_OUTBOUND) + } + } + + /// Outbound peers that count against the automatic target: every one + /// but the manual connections, which Core counts apart (`semAddnode`, + /// or no limit at all), connected or still in the handshake. + fn automatic_outbound_peers(&self) -> usize { + self.peers + .read() + .values() + .filter(|h| { + h.info.direction == Direction::Outbound + && h.info.conn_type != ConnType::Manual + && h.info.state != PeerState::Disconnected + }) + .count() + } + + /// The added nodes the operator listed, at the addresses they resolve to. + fn added_node_addrs(&self) -> Vec { + self.addnode_entries + .read() + .iter() + .filter(|e| e.listed) + .filter_map(|e| e.resolved.clone()) + .collect() + } + + /// Manual peers that are added nodes, connected or still in the + /// handshake: what [`MAX_ADDNODE_CONNECTIONS`] counts. + fn added_node_peers(&self) -> usize { + let added = self.added_node_addrs(); + self.peers + .read() + .values() + .filter(|h| h.info.conn_type == ConnType::Manual && h.info.state != PeerState::Disconnected) + .filter(|h| { + added.iter().any(|a| match (a, h.info.onion_host.as_deref()) { + (PeerAddr::Socket(sa), None) => *sa == h.info.addr, + (PeerAddr::Onion { host, .. }, Some(peer_host)) => host == peer_host, + _ => false, + }) + }) + .count() + } + + /// The [`DialBudget`] a dial to `target` counts against. `conn_type` is + /// the dial's: `None` for an automatic dial, which is manual after all + /// when the operator named the peer (as `spawn_peer` will type it), + /// `Some(Manual)` for a dial the operator asked for. A typed + /// `addconnection` dial holds its own reservation instead. + fn dial_budget(&self, target: &PeerAddr, conn_type: Option) -> DialBudget { + let manual = match (conn_type, target) { + (Some(ConnType::Manual), _) => true, + (Some(_), _) => return DialBudget::Unlimited, + (None, PeerAddr::Socket(sa)) => self.untyped_outbound_conn_type(sa, None) == ConnType::Manual, + (None, PeerAddr::Onion { host, port }) => { + let placeholder = SocketAddr::from(([0, 0, 0, 0], *port)); + self.untyped_outbound_conn_type(&placeholder, Some(host)) == ConnType::Manual + } }; - let outbound = self.outbound_count(); - if outbound >= max_outbound { - return Err("max outbound connections reached".to_string()); + if !manual { + DialBudget::Automatic + } else if self.added_node_addrs().contains(target) { + DialBudget::AddedNode + } else { + DialBudget::Unlimited } - Ok(()) + } + + /// Take a place for one dial in `budget`, or refuse when the budget is + /// spent. The place is held until the returned slot drops, which a + /// dial that connected does only after its peer is in `peers`. + /// + /// The check counts the dials already in flight along with the peers, + /// and it and the reservation happen under one lock. satd used to count + /// connected outbound peers only, manual ones included: each reconnect + /// tick started a dial for every eligible address at once (the limit + /// could not see them until they connected), and a full set of + /// automatic peers left no room for an added node or `addnode onetry`. + /// Core takes a grant from a counting semaphore before each dial: + /// `semOutbound` for automatic connections, `semAddnode` for added + /// nodes, none for `-connect` and `onetry` (`net.cpp` + /// `ThreadOpenConnections`, `ThreadOpenAddedConnections`; + /// `rpc/net.cpp` `addnode`). + fn reserve_dial(self: &Arc, budget: DialBudget) -> Result { + let mut in_flight = self.dials_in_flight.lock(); + match budget { + DialBudget::Automatic => { + // `addconnection`'s reservations take automatic slots too. + let open = self.automatic_outbound_peers() + + in_flight.automatic + + self.pending_typed_dials.read().len(); + if open >= self.automatic_outbound_target() { + return Err("max outbound connections reached".to_string()); + } + in_flight.automatic += 1; + } + DialBudget::AddedNode => { + if self.added_node_peers() + in_flight.added_node >= MAX_ADDNODE_CONNECTIONS { + return Err(format!( + "addnode connections are limited to {MAX_ADDNODE_CONNECTIONS} at a time" + )); + } + in_flight.added_node += 1; + } + DialBudget::Unlimited => {} + } + Ok(DialSlot { pm: Arc::clone(self), budget }) } /// Connect to an outbound peer with no caller-chosen type, as the @@ -2536,8 +2810,8 @@ impl PeerManager { /// `conn_type` is `Some(Manual)` for a dial the operator asked for /// ([`Self::connect_peer_addr_with`]), and one of `addconnection`'s four /// types for that RPC. `None` is an automatic dial, classified by - /// `spawn_peer`. A manual or automatic dial takes the historical single - /// outbound cap; `addconnection`'s types take Core's per-type ones. + /// `spawn_peer`. A manual or automatic dial counts against its + /// [`DialBudget`]; `addconnection`'s types take Core's per-type limits. pub async fn connect_outbound_as( self: &Arc, addr: SocketAddr, @@ -2626,9 +2900,11 @@ impl PeerManager { Ok(()) } - /// `connect_outbound_as`, with the per-type reservation optional: - /// `add_connection` has already taken it and holds the grant across the - /// spawned dial. + /// `connect_outbound_as`, with the reservation optional. `reserve` false + /// takes no place in any limit here: `add_connection` has already taken + /// its per-type one and the reconnect loop its [`DialSlot`], each held + /// across the spawned dial, and an `addnode onetry` has none, as Core + /// opens one without a grant. async fn connect_outbound_inner( self: &Arc, addr: SocketAddr, @@ -2651,15 +2927,16 @@ impl PeerManager { if !crate::net::is_valid(addr.ip()) { return Err(format!("{addr} is not a valid address to connect to")); } - // Take the capacity check and the per-type reservation together, so - // two callers cannot both pass the same free slot. + // Take the capacity check and the reservation together, so two + // callers cannot both pass the same free slot. + let manual_or_automatic = matches!(conn_type, None | Some(ConnType::Manual)); + let _dial_slot = if manual_or_automatic && reserve { + Some(self.reserve_dial(self.dial_budget(&PeerAddr::Socket(addr), conn_type))?) + } else { + None + }; let _typed_slot = match conn_type { - // A manual dial is capped as an automatic one is, as it always - // was; it has no per-type reservation to take. - None | Some(ConnType::Manual) => { - self.check_outbound_limit()?; - None - } + None | Some(ConnType::Manual) => None, Some(t) if reserve => { let mut pending = self.pending_typed_dials.write(); self.check_outbound_limit_for(t, &pending)?; @@ -2747,11 +3024,28 @@ impl PeerManager { host: &str, port: u16, conn_type: Option, + ) -> Result<(), String> { + self.connect_outbound_onion_inner(host, port, conn_type, true).await + } + + /// `connect_outbound_onion`, with the [`DialSlot`] optional, as in + /// [`Self::connect_outbound_inner`]. + async fn connect_outbound_onion_inner( + self: &Arc, + host: &str, + port: u16, + conn_type: Option, + reserve: bool, ) -> Result<(), String> { if !self.is_network_active() { return Err("networking disabled (networkactive=false)".to_string()); } - self.check_outbound_limit()?; + let _dial_slot = if reserve { + let target = PeerAddr::Onion { host: host.to_string(), port }; + Some(self.reserve_dial(self.dial_budget(&target, conn_type))?) + } else { + None + }; // Dedupe by onion host BEFORE any network I/O. Onion peers all share // the `0.0.0.0` placeholder socket, so `is_addr_connected` / @@ -2819,8 +3113,10 @@ impl PeerManager { /// Dial a peer the operator asked for (either socket or .onion), there /// and then: `-connect`, `-addnode` and `-seednode` at startup and on - /// reload, `addnode add`, `addnode onetry`, and a named peer whose - /// name resolved. The connection is `manual`. + /// reload, `addnode add`, and a named peer whose name resolved. The + /// connection is `manual`; an added node counts against + /// [`MAX_ADDNODE_CONNECTIONS`]. `addnode onetry` is + /// [`Self::connect_peer_addr_onetry`]. pub async fn connect_peer_addr(self: &Arc, addr: &PeerAddr) -> Result<(), String> { self.connect_peer_addr_with(addr, None).await } @@ -2847,11 +3143,53 @@ impl PeerManager { } } + /// `addnode onetry`: one manual dial, outside every limit, as + /// Core's `onetry` opens its connection without a grant (`rpc/net.cpp` + /// `addnode`). A peer already connected, or being dialled, is left as + /// it is and the call succeeds: Core's `OpenNetworkConnection` returns + /// without dialling a host it is connected to, and `onetry` answers + /// null either way. A dial that fails is reported. + /// + /// satd refused `onetry` with "max outbound connections reached" once + /// the automatic outbound peers were all connected, and with "already + /// connected" for a connected peer. + pub async fn connect_peer_addr_onetry( + self: &Arc, + addr: &PeerAddr, + use_v2: Option, + ) -> Result<(), String> { + if self.is_peer_addr_connected(addr) || self.is_dial_in_flight(addr) { + tracing::debug!(%addr, "addnode onetry: already connected or connecting"); + return Ok(()); + } + match addr { + PeerAddr::Socket(sa) => { + self.connect_outbound_inner(*sa, Some(ConnType::Manual), use_v2, false).await + } + PeerAddr::Onion { host, port } => { + self.connect_outbound_onion_inner(host, *port, Some(ConnType::Manual), false).await + } + } + } + + /// Whether a dial to `addr` is under way. + fn is_dial_in_flight(&self, addr: &PeerAddr) -> bool { + match addr { + PeerAddr::Socket(sa) => self.pending_connections.read().contains(sa), + PeerAddr::Onion { host, .. } => self.pending_onion_dials.read().contains(host), + } + } + /// Dial `addr` as one of the node's own automatic dials: the reconnect /// loop's, and a seed's. No type is chosen here; `spawn_peer` reports /// the connection `manual` if the operator named the peer and /// `outbound-full-relay` otherwise. pub async fn connect_peer_addr_automatic(self: &Arc, addr: &PeerAddr) -> Result<(), String> { + // An automatic connection is not opened to a banned address, as in + // Core; a peer the operator named is dialled regardless. + if !self.is_manual_target(addr) && self.is_peer_addr_banned(addr) { + return Err(format!("{addr} is banned")); + } match addr { PeerAddr::Socket(sa) => self.connect_outbound(*sa).await, PeerAddr::Onion { host, port } => self.connect_outbound_onion(host, *port, None).await, @@ -3346,14 +3684,25 @@ impl PeerManager { .already_banned(target, is_subnet, crate::time::now_secs()) } - /// Disconnect every peer whose IP falls within `target`. + /// Disconnect every peer `target` covers: an onion peer by its onion + /// host, any other by its IP. An onion peer's IP is the `0.0.0.0` + /// placeholder they all share, so matching it said nothing about the + /// peer: `setban .onion` left that peer connected. fn disconnect_banned_peers(&self, target: &crate::net::ban::BanTarget) { let ids_to_disconnect: Vec = { let peers = self.peers.read(); peers .iter() .filter_map(|(id, handle)| { - if target.contains_addr(&handle.info.addr.ip()) { + let covered = match (target, handle.info.onion_host.as_deref()) { + (crate::net::ban::BanTarget::Onion(host), Some(peer_host)) => host == peer_host, + (crate::net::ban::BanTarget::Onion(_), None) => false, + (crate::net::ban::BanTarget::Net(_), Some(_)) => false, + (crate::net::ban::BanTarget::Net(_), None) => { + target.contains_addr(&handle.info.addr.ip()) + } + }; + if covered { Some(*id) } else { None @@ -3662,11 +4011,204 @@ impl PeerManager { Self::onion_connected_in(&self.peers.read(), host) } + /// One pass of the reconnect loop, every 10 seconds of the run loop: + /// redial the peers the operator named, then fill the free automatic + /// outbound slots from the addresses the node learned. + /// + /// The two halves are apart, as in Core, where + /// `ThreadOpenAddedConnections` and the `-connect` loop run separately + /// from `ThreadOpenConnections` (`net.cpp`). satd ran one pass, and only + /// while the outbound count, manual peers included, was below the + /// automatic target: on a node with its automatic peers connected, an + /// added node that dropped, or whose first dial failed, was not dialled + /// again. + fn reconnect_tick(self: &Arc) { + let now = Instant::now(); + // Clean expired bans + { + let now_secs = crate::time::now_secs(); + self.ban_list.write().prune_expired(now_secs); + self.flush_banlist(); + } + self.redial_manual_peers(now); + self.dial_automatic_peers(now); + } + + /// Whether `target` is a peer the operator named. + fn is_manual_target(&self, target: &PeerAddr) -> bool { + match target { + PeerAddr::Socket(sa) => self.manual_addrs.read().contains(sa), + PeerAddr::Onion { host, .. } => self.manual_onion_hosts.read().contains(host), + } + } + + /// The manual half of [`Self::reconnect_tick`]: every peer the operator + /// named (`-connect`, `-addnode` / `addnode add`, `-seednode`) that is + /// not connected, whatever the automatic outbound count. Added nodes stop + /// at [`MAX_ADDNODE_CONNECTIONS`], and the one past it waits for a later + /// tick, as Core's added-node thread waits for a grant; the others have + /// no limit. A failed dial backs off to at most + /// [`MANUAL_RECONNECT_MAX_BACKOFF`], Core's added-node retry interval. + /// + /// A ban does not stop these dials. Core opens a manual connection by + /// the name the operator gave and checks bans only for the automatic + /// ones (`OpenNetworkConnection`, `net.cpp`); satd skipped a banned + /// manual peer here. + fn redial_manual_peers(self: &Arc, now: Instant) { + let mut targets: Vec = + self.manual_addrs.read().iter().map(|a| PeerAddr::Socket(*a)).collect(); + let onion: Vec = self.connect_peer_addrs.read().clone(); + targets.extend(onion.into_iter().filter(|a| self.is_manual_target(a))); + for target in targets { + if !self.redial_due(&target, now) { + continue; + } + let Ok(slot) = self.reserve_dial(self.dial_budget(&target, None)) else { + continue; + }; + self.spawn_redial(target, slot, MANUAL_RECONNECT_MAX_BACKOFF); + } + } + + /// The automatic half of [`Self::reconnect_tick`]: learned addresses, + /// then onion peers learned from `addrv2` gossip, one dial each until the + /// automatic outbound target is reached, the dials in flight counted + /// (`reserve_dial`). Under `-connect` the node dials none of them; see + /// [`Self::may_dial`]. A banned address is skipped, an onion host + /// included (`setban .onion`, or a ban for misbehaviour), as Core + /// skips a banned address for an automatic connection; satd checked + /// the bans of clearnet addresses only. + fn dial_automatic_peers(self: &Arc, now: Instant) { + // An address the operator also named is offered by the manual half, + // and `spawn_peer` makes it the manual connection it is: Core makes + // no automatic connection to an added node (`AddedNodesContain`, + // v31.1 `src/net.cpp:2883`). + let learned: Vec = self.learned_addrs.read().iter().copied().collect(); + for addr in learned { + if self.is_manual_target(&PeerAddr::Socket(addr)) || !self.may_dial(&addr) { + continue; + } + let target = PeerAddr::Socket(addr); + if self.is_peer_addr_banned(&target) || !self.redial_due(&target, now) { + continue; + } + match self.reserve_dial(DialBudget::Automatic) { + Ok(slot) => self.spawn_redial(target, slot, AUTOMATIC_RECONNECT_MAX_BACKOFF), + Err(_) => return, + } + } + + // `connect_peer_addrs` is fillable from untrusted addrv2 gossip (up + // to MAX_ONION_CONNECT_ADDRS), and every onion dial holds a circuit + // through the one proxy, so on top of the free slots a tick starts + // no more than a burst of them. + let onion: Vec = self.connect_peer_addrs.read().clone(); + let mut budget = Self::onion_dial_budget( + self.automatic_outbound_target(), + self.automatic_outbound_peers(), + self.dials_in_flight.lock().automatic, + ); + for target in onion.into_iter().filter(|a| !self.is_manual_target(a)) { + // Under `-connect`, only the peers the operator named. + if budget == 0 || !self.automatic_outbound.load(Ordering::Relaxed) { + break; + } + if self.is_peer_addr_banned(&target) || !self.redial_due(&target, now) { + continue; + } + match self.reserve_dial(DialBudget::Automatic) { + Ok(slot) => { + budget -= 1; + self.spawn_redial(target, slot, AUTOMATIC_RECONNECT_MAX_BACKOFF); + } + Err(_) => return, + } + } + } + + /// Whether the reconnect loop should dial `target` now: it is neither + /// connected nor being dialled, and not backing off. Bans are the + /// caller's to check: they hold for automatic dials only. + fn redial_due(&self, target: &PeerAddr, now: Instant) -> bool { + match target { + PeerAddr::Socket(sa) => { + !self.is_addr_connected(sa) + && !self.pending_connections.read().contains(sa) + && self + .reconnect_backoff + .read() + .get(sa) + .is_none_or(|state| now >= state.next_attempt) + } + PeerAddr::Onion { host, .. } => { + !self.is_onion_connected(host) + && !self.pending_onion_dials.read().contains(host) + && self + .onion_reconnect_backoff + .read() + .get(&target.to_string()) + .is_none_or(|state| now >= state.next_attempt) + } + } + } + + /// Dial `target` for the reconnect loop in a task of its own, holding the + /// place `slot` took for it, and keep its backoff: cleared when the dial + /// connects, longer (up to `max_backoff`) when it fails. Only a + /// connection clears it. + fn spawn_redial(self: &Arc, target: PeerAddr, slot: DialSlot, max_backoff: Duration) { + let pm = Arc::clone(self); + tokio::spawn(async move { + let _slot = slot; + let result = match &target { + PeerAddr::Socket(sa) => pm.connect_outbound_inner(*sa, None, None, false).await, + PeerAddr::Onion { host, port } => { + pm.connect_outbound_onion_inner(host, *port, None, false).await + } + }; + if let Err(e) = &result { + tracing::debug!(%target, "Reconnect failed: {e}"); + // What makes an address terrible, and so replaceable once + // the book is full (Core's `Attempt`). + if let PeerAddr::Socket(sa) = &target { + pm.addrman.write().mark_attempt(*sa); + } + } + let connected = result.is_ok(); + let update = |state: &mut ReconnectState| { + if connected { + state.reset(); + } else { + state.record_failure(max_backoff); + } + }; + match &target { + // Not for an address that left both lists while the dial ran + // (the address book replaced it): its state would never go. + // Checked with the backoff map locked, so a replacement that + // removes the address's state either sees this entry (and + // removes it after) or has already taken the address off the + // list (and this adds nothing). + PeerAddr::Socket(sa) => { + let mut backoff = pm.reconnect_backoff.write(); + if pm.learned_addrs.read().contains(sa) || pm.manual_addrs.read().contains(sa) { + update(backoff.entry(*sa).or_insert_with(ReconnectState::new)); + } + } + PeerAddr::Onion { .. } => update( + pm.onion_reconnect_backoff + .write() + .entry(target.to_string()) + .or_insert_with(ReconnectState::new), + ), + } + }); + } + /// How many new onion dials the reconnect loop may start this tick: the - /// open outbound slots (counting in-flight dials, which `outbound_count` - /// excludes because they aren't `Connected` yet) capped by the per-tick - /// burst limit. Without counting `in_flight`, a single tick could spawn a - /// dial for every gossip-discovered onion at once. + /// open automatic outbound slots (counting dials in flight) capped by the + /// per-tick burst limit. Without counting `in_flight`, a single tick + /// could spawn a dial for every gossip-discovered onion at once. fn onion_dial_budget(target: usize, outbound: usize, in_flight: usize) -> usize { target .saturating_sub(outbound + in_flight) @@ -3687,6 +4229,17 @@ impl PeerManager { self.ban_list.read().is_banned(&addr.ip(), now) } + /// Whether `addr` is banned: a socket address by its IP (and any subnet + /// holding it), an onion one by its host. + fn is_peer_addr_banned(&self, addr: &PeerAddr) -> bool { + match addr { + PeerAddr::Socket(sa) => self.is_addr_banned(sa), + PeerAddr::Onion { host, .. } => { + self.ban_list.read().is_onion_banned(host, crate::time::now_secs()) + } + } + } + /// Add ban score to a peer. Once the score reaches [`BAN_THRESHOLD`] the /// peer has misbehaved (Core's `Misbehaving`), and it is dealt with as /// [`Self::misbehaviour_action`] says. @@ -3880,14 +4433,13 @@ impl PeerManager { let _htip = self.headers_tip.load(Ordering::Relaxed) as u32; // When chain advances, immediately request more blocks (don't wait for timer) + // Reconnect backoff is not reset here: a dial's backoff ends when + // a dial to that address succeeds. Resetting every address's on + // each new tip, which during IBD is every tick, meant no address + // ever backed off. let tip_advanced = tip != last_tip; if tip_advanced { last_tip = tip; - // Reset reconnect backoff on chain progress - let mut backoff = self.reconnect_backoff.write(); - for state in backoff.values_mut() { - state.reset(); - } } // IBD scheduler maintenance @@ -4069,13 +4621,13 @@ impl PeerManager { } } - // Every 20 ticks (10 seconds), reconnect if below outbound target. - // Skipped entirely while networking is paused (`networkactive=0` / - // `setnetworkactive false`): otherwise each tick would spawn dials - // that immediately fail the gate, churning backoff state and logs. - // Named peers are looked up again on the same tick, whatever the - // outbound count: an added node is a manual connection, which - // Core dials outside the outbound target. + // Every 20 ticks (10 seconds), redial the peers the operator + // named and fill the automatic outbound slots + // (`reconnect_tick`). Skipped entirely while networking is paused + // (`networkactive=0` / `setnetworkactive false`): otherwise each + // tick would spawn dials that immediately fail the gate, churning + // backoff state and logs. Named peers are looked up again on the + // same tick. if ticks.is_multiple_of(20) && self.is_network_active() && self.manual_targets_due(Instant::now()) @@ -4083,157 +4635,7 @@ impl PeerManager { tokio::spawn(Arc::clone(self).refresh_manual_targets()); } if ticks.is_multiple_of(20) && self.is_network_active() { - let outbound = self.outbound_count(); - let target = if self.is_ibd() { MAX_OUTBOUND_IBD } else { MAX_OUTBOUND }; - let need_peers = outbound < target; - if need_peers { - // The peers the operator named first, then those the node - // learned. An address on both lists is offered once, and - // `spawn_peer` makes it the manual connection it is: Core - // makes no automatic connection to an added node - // (`AddedNodesContain`, v31.1 `src/net.cpp:2883`). - let manual: HashSet = self.manual_addrs.read().clone(); - let learned: Vec = self - .learned_addrs - .read() - .iter() - .filter(|a| !manual.contains(a)) - .copied() - .collect(); - let addrs: Vec = manual.into_iter().chain(learned).collect(); - - let now = Instant::now(); - - // Clean expired bans - { - let now_secs = crate::time::now_secs(); - self.ban_list.write().prune_expired(now_secs); - self.flush_banlist(); - } - - for addr in addrs { - // Under `-connect`, only the peers the operator named. - if !self.may_dial(&addr) { - continue; - } - // Skip if already connected - if self.is_addr_connected(&addr) { - continue; - } - // Skip if banned - if self.is_addr_banned(&addr) { - continue; - } - // Check backoff timer - { - let backoff = self.reconnect_backoff.read(); - if let Some(state) = backoff.get(&addr) - && now < state.next_attempt { - continue; - } - } - - // Don't exceed target - if self.check_outbound_limit().is_err() { - break; - } - - let pm = Arc::clone(self); - tokio::spawn(async move { - match pm.connect_outbound(addr).await { - Ok(_) => { - let mut backoff = pm.reconnect_backoff.write(); - backoff - .entry(addr) - .or_insert_with(ReconnectState::new) - .reset(); - } - Err(e) => { - tracing::debug!(%addr, "Reconnect failed: {}", e); - let mut backoff = pm.reconnect_backoff.write(); - backoff - .entry(addr) - .or_insert_with(ReconnectState::new) - .record_failure(); - } - } - }); - } - - // Also reconnect .onion peers — with the same discipline the - // clearnet loop above has. `connect_peer_addrs` is fillable - // from untrusted addrv2 gossip (up to MAX_ONION_CONNECT_ADDRS), - // so spawning a dial per entry with no cap would let one peer - // drive hundreds of concurrent SOCKS dials through the single - // proxy every tick. Bound new dials this tick by the open - // slot budget (counting in-flight dials, which - // `outbound_count()` excludes) and a per-tick burst cap; skip - // already-connected / in-flight hosts; and back off failures - // (onion-host-keyed, since `reconnect_backoff` is SocketAddr- - // keyed). - let onion_addrs = self.connect_peer_addrs.read().clone(); - let mut budget = Self::onion_dial_budget( - target, - self.outbound_count(), - self.pending_onion_dials.read().len(), - ); - for peer_addr in onion_addrs { - if budget == 0 { - break; - } - // Under `-connect`, only the peers the operator named. - let manual = match &peer_addr { - PeerAddr::Onion { host, .. } => { - self.manual_onion_hosts.read().contains(host) - } - PeerAddr::Socket(sa) => self.manual_addrs.read().contains(sa), - }; - if !manual && !self.automatic_outbound.load(Ordering::Relaxed) { - continue; - } - let already = match &peer_addr { - PeerAddr::Onion { host, .. } => { - self.is_onion_connected(host) - || self.pending_onion_dials.read().contains(host) - } - PeerAddr::Socket(sa) => self.is_addr_connected(sa), - }; - if already { - continue; - } - let key = peer_addr.to_string(); - { - let backoff = self.onion_reconnect_backoff.read(); - if let Some(state) = backoff.get(&key) - && now < state.next_attempt - { - continue; - } - } - budget -= 1; - let pm = Arc::clone(self); - tokio::spawn(async move { - let key = peer_addr.to_string(); - match pm.connect_peer_addr_automatic(&peer_addr).await { - Ok(_) => { - pm.onion_reconnect_backoff - .write() - .entry(key) - .or_insert_with(ReconnectState::new) - .reset(); - } - Err(e) => { - tracing::debug!(%peer_addr, "Onion reconnect failed: {}", e); - pm.onion_reconnect_backoff - .write() - .entry(key) - .or_insert_with(ReconnectState::new) - .record_failure(); - } - } - }); - } - } + self.reconnect_tick(); } // Drop addr-fetch connections whose peer never sent a usable @@ -4441,12 +4843,20 @@ impl PeerManager { self.add_ban_score(id, 100, &format!("addr message size = {}", addrs.len())); return; } + let count = addrs.len(); + let admitted = if relay_addrs { self.rate_limit_addrs(id, addrs) } else { Vec::new() }; let source = self.peer_ip(id); - for (_, addr) in &addrs { - if relay_addrs - && let Ok(sock_addr) = addr.socket_addr() + let mut processed = 0u64; + for (_, addr) in &admitted { + let sock_addr = addr.socket_addr().ok(); + // Core looks no further at a banned address, and does + // not count it processed. + if sock_addr.is_some_and(|sa| self.is_addr_banned(&sa)) { + continue; + } + processed += 1; + if let Some(sock_addr) = sock_addr && !self.is_addr_connected(&sock_addr) - && !self.is_addr_banned(&sock_addr) { self.add_learned_addr_from( sock_addr, @@ -4455,7 +4865,8 @@ impl PeerManager { ); } } - self.note_addr_fetch_answered(id, addrs.len()); + self.note_addrs_processed(id, processed); + self.note_addr_fetch_answered(id, count); } NetworkMessage::GetAddr => { // Respond with our declared external addresses (-externalip) @@ -4595,8 +5006,11 @@ impl PeerManager { self.add_ban_score(id, 100, &format!("addrv2 message size = {}", addrs.len())); return; } + let count = addrs.len(); + let admitted = if relay_addrs { self.rate_limit_addrs(id, addrs) } else { Vec::new() }; let source = self.peer_ip(id); - for addr_msg in addrs.iter().filter(|_| relay_addrs) { + let mut processed = 0u64; + for addr_msg in &admitted { match &addr_msg.addr { // BIP 155 TorV3: `socket_addr()` can't represent these, // so derive the .onion host and queue it for dialing. @@ -4604,14 +5018,23 @@ impl PeerManager { if onion_routing && addr_msg.port != 0 => { let host = crate::net::peer::torv3_to_onion_host(pubkey); + if self.ban_list.read().is_onion_banned(&host, crate::time::now_secs()) { + continue; + } + processed += 1; self.add_onion_connect_addr(host, addr_msg.port); } // IPv4 / IPv6 (and, under a proxy, anything else with a // socket form) flow through the clearnet address book. _ => { - if let Ok(sock_addr) = addr_msg.socket_addr() + let sock_addr = addr_msg.socket_addr().ok(); + // As for `addr`. + if sock_addr.is_some_and(|sa| self.is_addr_banned(&sa)) { + continue; + } + processed += 1; + if let Some(sock_addr) = sock_addr && !self.is_addr_connected(&sock_addr) - && !self.is_addr_banned(&sock_addr) { self.add_learned_addr_from( sock_addr, @@ -4622,7 +5045,8 @@ impl PeerManager { } } } - self.note_addr_fetch_answered(id, addrs.len()); + self.note_addrs_processed(id, processed); + self.note_addr_fetch_answered(id, count); } NetworkMessage::SendAddrV2 => { let mut peers = self.peers.write(); @@ -9901,6 +10325,8 @@ impl PeerManager { // connection is about to be dropped. let conn_type = self.conn_type_of(id); if conn_type == ConnType::AddrFetch { + // Room for the answer, before it can arrive. + self.note_getaddr_sent(id); writer .send(NetworkMessage::GetAddr) .await @@ -9957,6 +10383,7 @@ impl PeerManager { && conn_type != ConnType::AddrFetch && (self.proxy.is_some() || self.onion_proxy.is_some()) { + self.note_getaddr_sent(id); writer.send(NetworkMessage::GetAddr) .await .map_err(|e| format!("send getaddr: {}", e))?; @@ -14731,6 +15158,9 @@ mod tests { pm.peers .write() .insert(1, mk_handle(1, announcer, Direction::Inbound, PeerState::Connected)); + // Two addresses from one peer: past the address rate limit's first + // token, so the peer has the `addr` permission. + pm.peers.write().get_mut(&1).unwrap().info.permissions.addr = true; let v1: SocketAddr = "198.51.100.1:8333".parse().unwrap(); let v1_services = ServiceFlags::NETWORK | ServiceFlags::WITNESS | ServiceFlags::COMPACT_FILTERS; diff --git a/node/src/net/manager_addrdials_tests.rs b/node/src/net/manager_addrdials_tests.rs new file mode 100644 index 000000000..ef7ddbc00 --- /dev/null +++ b/node/src/net/manager_addrdials_tests.rs @@ -0,0 +1,704 @@ +//! The addresses a peer may hand the node, and the outbound dials the node +//! starts from them: Core's address rate limit (`ProcessAddrs`), and the +//! separate limits on automatic and manual connections (`semOutbound`, +//! `semAddnode`; `net.cpp` `ThreadOpenConnections`, +//! `ThreadOpenAddedConnections`), against a live `PeerManager`. + +use super::*; +use crate::net::flow::InFlight; +use std::sync::atomic::AtomicUsize; + +fn mk_pm() -> (Arc, tempfile::TempDir) { + use crate::chain::state::AssumeValid; + use crate::storage::db::InMemoryStore; + use crate::storage::flatfile::FlatFileManager; + use crate::validation::script::NoopVerifier; + + let dir = tempfile::TempDir::new().unwrap(); + let store = Box::new(InMemoryStore::new()); + let flat_files = FlatFileManager::new(&dir.path().join("blocks")).unwrap(); + let chain_state = Arc::new( + ChainState::new( + store, + flat_files, + Network::Regtest, + Box::new(NoopVerifier), + AssumeValid::Disabled, + 450, + 4, + Default::default(), + Default::default(), + Default::default(), + ) + .unwrap(), + ); + (over(chain_state), dir) +} + +fn over(chain_state: Arc) -> Arc { + let mempool = Arc::new(Mempool::new(1_000_000, 0)); + let fee_estimator = Arc::new(FeeEstimator::new()); + let (_shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false); + PeerManager::new(chain_state, mempool, fee_estimator, Network::Regtest, shutdown_rx) +} + +/// Out of IBD, so the automatic outbound target is 8 rather than 64: a +/// peer has announced a headers tip the node's tip is within 24 of. +fn leave_ibd(pm: &PeerManager) { + pm.headers_tip.store(1, Ordering::Relaxed); + assert!(!pm.is_ibd(), "fixture: out of IBD"); +} + +fn add_peer(pm: &PeerManager, info: PeerInfo) { + let (tx, _rx) = mpsc::channel::(8); + pm.peers.write().insert( + info.id, + PeerHandle { + info, + msg_tx: tx.into(), + disconnect: Arc::new(tokio::sync::Notify::new()), + flow: Arc::new(crate::net::flow::PeerFlow::new()), + last_getheaders_sent: None, + last_mempool_served: None, + fee_filter_sent: None, + stats: PeerStats::new(NetTotals::new()), + }, + ); +} + +/// A connected peer. +fn peer(id: PeerId, addr: SocketAddr, direction: Direction) -> PeerInfo { + let mut info = PeerInfo::new(id, addr, direction); + info.state = PeerState::Connected; + info +} + +/// `n` connected automatic (`outbound-full-relay`) peers, ids from 1000. +fn fill_automatic(pm: &PeerManager, n: usize) { + for i in 0..n { + let addr = SocketAddr::from(([203, 0, 113, i as u8 + 1], 8333)); + add_peer(pm, peer(1000 + i as PeerId, addr, Direction::Outbound)); + } +} + +/// `n` addresses, each in a /16 of its own so the address book's +/// per-group limit does not apply, starting at first octet `base`. +fn spread(base: u8, n: usize) -> Vec { + (0..n) + .map(|i| SocketAddr::from(([base + (i / 250) as u8, (i % 250) as u8, 7, 7], 8333))) + .collect() +} + +fn addr_msg(addrs: &[SocketAddr]) -> NetworkMessage { + NetworkMessage::Addr( + addrs + .iter() + .map(|a| (1_700_000_000, Address::new(a, ServiceFlags::NETWORK | ServiceFlags::WITNESS))) + .collect(), + ) +} + +fn addr_counts(pm: &PeerManager, id: PeerId) -> (u64, u64) { + let peers = pm.peers.read(); + let rate = &peers.get(&id).expect("peer").info.addr_rate; + (rate.processed, rate.rate_limited) +} + +/// A listener on loopback that accepts every connection and keeps it open, +/// counting the accepts. +struct Listener { + addr: SocketAddr, + accepts: Arc, + task: tokio::task::JoinHandle<()>, +} + +impl Listener { + async fn bind() -> Self { + let socket = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = socket.local_addr().unwrap(); + let accepts = Arc::new(AtomicUsize::new(0)); + let counter = accepts.clone(); + let task = tokio::spawn(async move { + let mut held = Vec::new(); + while let Ok((stream, _)) = socket.accept().await { + counter.fetch_add(1, Ordering::SeqCst); + held.push(stream); + } + }); + Self { addr, accepts, task } + } + + fn accepts(&self) -> usize { + self.accepts.load(Ordering::SeqCst) + } +} + +impl Drop for Listener { + fn drop(&mut self) { + self.task.abort(); + } +} + +async fn listeners(n: usize) -> Vec { + let mut out = Vec::new(); + for _ in 0..n { + out.push(Listener::bind().await); + } + out +} + +fn total_accepts(ls: &[Listener]) -> usize { + ls.iter().map(Listener::accepts).sum() +} + +/// Wait until `cond` holds. +async fn wait_for(pm: &PeerManager, cond: impl Fn() -> bool, what: &str) { + let deadline = Instant::now() + Duration::from_secs(30); + while !cond() { + assert!( + Instant::now() < deadline, + "timed out: {what} (in flight {:?}, dialling {:?}, backoff {:?})", + pm.dials_in_flight.lock(), + pm.pending_connections.read(), + pm.reconnect_backoff.read().iter().map(|(a, s)| (*a, s.attempts)).collect::>(), + ); + tokio::time::sleep(Duration::from_millis(20)).await; + } +} + +/// Wait for the dials a reconnect tick started to finish, one way or the +/// other: nothing is in flight any more. +async fn settle(pm: &PeerManager) { + let deadline = Instant::now() + Duration::from_secs(20); + loop { + let in_flight = { + let d = pm.dials_in_flight.lock(); + d.automatic + d.added_node + }; + if in_flight == 0 && pm.pending_connections.read().is_empty() && pm.pending_onion_dials.read().is_empty() { + break; + } + assert!(Instant::now() < deadline, "dials still in flight"); + tokio::time::sleep(Duration::from_millis(20)).await; + } + // Accepts land on the listener tasks a moment after the dial returns. + tokio::time::sleep(Duration::from_millis(200)).await; +} + +// ---- Address rate limit ---- + +/// Core gives each peer a token bucket for the addresses it announces: one +/// token to start, 0.1 a second after that. satd took every entry of every +/// `addr`, so one peer could put a thousand addresses into the address book +/// and onto the dial list per message. A peer with the `addr` permission is +/// not limited, as in Core. +#[test] +fn a_peers_addresses_are_rate_limited_unless_it_has_the_addr_permission() { + let (pm, _dir) = mk_pm(); + let limited = SocketAddr::from(([127, 0, 0, 1], 40001)); + add_peer(&pm, peer(1, limited, Direction::Inbound)); + pm.handle_message(1, addr_msg(&spread(1, 1000)), InFlight::adopt(None)); + assert_eq!(addr_counts(&pm, 1), (1, 999), "one token: one address processed, the rest rate-limited"); + assert_eq!(pm.addrman.read().len(), 1, "one address reached the address book"); + assert_eq!(pm.learned_addrs.read().iter().count(), 1, "and the dial list"); + + let mut exempt = peer(2, SocketAddr::from(([127, 0, 0, 1], 40002)), Direction::Inbound); + exempt.permissions.addr = true; + add_peer(&pm, exempt); + pm.handle_message(2, addr_msg(&spread(20, 1000)), InFlight::adopt(None)); + assert_eq!(addr_counts(&pm, 2), (1000, 0), "the addr permission takes them all"); + assert_eq!(pm.addrman.read().len(), 1001); +} + +/// Core adds `MAX_ADDR_TO_SEND` tokens when it sends a peer `getaddr`, so +/// the answer is taken whole. satd sends one to an addr-fetch peer, and to +/// outbound peers under a proxy. +#[test] +fn the_answer_to_a_getaddr_is_taken_whole() { + let (pm, _dir) = mk_pm(); + add_peer(&pm, peer(1, SocketAddr::from(([198, 51, 100, 1], 8333)), Direction::Outbound)); + pm.note_getaddr_sent(1); + pm.handle_message(1, addr_msg(&spread(1, 1000)), InFlight::adopt(None)); + let (processed, limited) = addr_counts(&pm, 1); + assert_eq!((processed, limited), (1000, 0)); +} + +/// The same limit applies to `addrv2`. +#[test] +fn addrv2_is_rate_limited_too() { + use bitcoin::p2p::address::{AddrV2, AddrV2Message}; + let (pm, _dir) = mk_pm(); + add_peer(&pm, peer(1, SocketAddr::from(([127, 0, 0, 1], 40001)), Direction::Inbound)); + let msg = NetworkMessage::AddrV2( + spread(1, 500) + .into_iter() + .map(|a| AddrV2Message { + time: 1_700_000_000, + services: ServiceFlags::NETWORK, + addr: match a.ip() { + IpAddr::V4(v4) => AddrV2::Ipv4(v4), + IpAddr::V6(v6) => AddrV2::Ipv6(v6), + }, + port: a.port(), + }) + .collect(), + ); + pm.handle_message(1, msg, InFlight::adopt(None)); + assert_eq!(addr_counts(&pm, 1), (1, 499)); +} + +// ---- The automatic dial list ---- + +/// Only an address the address book holds goes on the dial list. The book +/// keeps at most 64 new addresses per network group; satd put every +/// gossiped address on the dial list whether the book took it or not. +#[test] +fn an_address_the_book_has_no_room_for_is_not_dialled() { + let (pm, _dir) = mk_pm(); + let source = IpAddr::from([198, 51, 100, 1]); + let one_group: Vec = + (0..100).map(|i| SocketAddr::from(([44, 1, (i / 250) as u8, (i % 250) as u8 + 1], 8333))).collect(); + for a in &one_group { + pm.add_learned_addr_from(*a, 9, source); + } + let book = pm.addrman.read().len(); + assert_eq!(book, 64, "fixture: the book's per-group limit"); + assert_eq!(pm.learned_addrs.read().iter().count(), 64, "the dial list holds what the book holds"); + for a in &one_group[..64] { + assert!(pm.learned_addrs.read().contains(a)); + } + // An address the book already holds, announced again, stays listed once. + pm.add_learned_addr_from(one_group[0], 9, source); + assert_eq!(pm.learned_addrs.read().iter().count(), 64); +} + +/// The list holds each address once and stops at its cap. +#[test] +fn the_dial_list_stops_at_its_cap() { + let mut list = LearnedAddrs::default(); + let all: Vec = (0..=MAX_LEARNED_ADDRS as u32) + .map(|i| SocketAddr::from((std::net::Ipv4Addr::from(0x0b00_0000 + i), 8333))) + .collect(); + for a in &all[..MAX_LEARNED_ADDRS] { + assert!(list.push(*a)); + } + assert!(!list.push(all[0]), "a duplicate is refused"); + assert!(!list.push(all[MAX_LEARNED_ADDRS]), "past the cap is refused"); + assert_eq!(list.iter().count(), MAX_LEARNED_ADDRS); +} + +// ---- Outbound dials ---- + +/// The automatic limit counts the dials in flight along with the peers. +/// satd counted connected peers only, so a reconnect tick started a dial +/// for every eligible learned address at once: twenty listeners, three +/// automatic peers connected and a target of eight got twenty dials. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_reconnect_tick_starts_no_more_automatic_dials_than_free_slots() { + let (pm, _dir) = mk_pm(); + leave_ibd(&pm); + fill_automatic(&pm, 3); + let ls = listeners(20).await; + for l in &ls { + pm.learned_addrs.write().push(l.addr); + } + + pm.reconnect_tick(); + wait_for(&pm, || total_accepts(&ls) >= 5, "five dials reach a listener").await; + settle(&pm).await; + assert_eq!(total_accepts(&ls), 5, "five free slots, five dials"); + assert_eq!(pm.automatic_outbound_peers(), 8, "the target is reached"); + + // Full now: the next tick starts nothing. + pm.reconnect_tick(); + settle(&pm).await; + assert_eq!(total_accepts(&ls), 5); +} + +/// Manual connections are outside the automatic target, as in Core, where +/// added nodes have their own limit (`semAddnode`). satd counted them +/// against it, and redialled named peers only while the automatic count +/// was below the target: with eight automatic peers, an added node was not +/// dialled at all. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn an_added_node_is_dialled_with_the_automatic_slots_full() { + let (pm, _dir) = mk_pm(); + leave_ibd(&pm); + fill_automatic(&pm, 8); + let l = Listener::bind().await; + assert!(pm.addnode_add(&l.addr.to_string(), PeerAddr::Socket(l.addr))); + + pm.reconnect_tick(); + wait_for(&pm, || l.accepts() == 1, "the added node is dialled").await; + settle(&pm).await; + let conn_type = pm.peers.read().values().find(|h| h.info.addr == l.addr).map(|h| h.info.conn_type); + assert_eq!(conn_type, Some(ConnType::Manual)); + + // The automatic limit still holds for an automatic dial. + let spare = Listener::bind().await; + let err = pm.connect_outbound(spare.addr).await.expect_err("the automatic slots are full"); + assert_eq!(err, "max outbound connections reached"); +} + +/// Core's `MAX_ADDNODE_CONNECTIONS`: eight added nodes at a time. The other +/// manual connections (`-connect`, `-seednode`) have no limit. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn added_nodes_stop_at_eight_and_other_manual_peers_do_not() { + let (pm, _dir) = mk_pm(); + leave_ibd(&pm); + let added = listeners(10).await; + for l in &added { + assert!(pm.addnode_add(&l.addr.to_string(), PeerAddr::Socket(l.addr))); + } + let connect = listeners(10).await; + for l in &connect { + assert!(pm.add_peer_addr(PeerAddr::Socket(l.addr))); + } + + pm.reconnect_tick(); + wait_for( + &pm, + || total_accepts(&added) >= MAX_ADDNODE_CONNECTIONS && total_accepts(&connect) >= 10, + "the dials reach their listeners", + ) + .await; + settle(&pm).await; + assert_eq!(total_accepts(&added), MAX_ADDNODE_CONNECTIONS, "eight added nodes"); + assert_eq!(total_accepts(&connect), 10, "every other manual peer"); + assert_eq!(pm.added_node_peers(), MAX_ADDNODE_CONNECTIONS); + + // One added node goes away: the next tick dials one more. + let gone = pm + .peers + .read() + .iter() + .find(|(_, h)| added.iter().any(|l| l.addr == h.info.addr)) + .map(|(id, _)| *id) + .unwrap(); + pm.peers.write().remove(&gone); + pm.reconnect_tick(); + wait_for(&pm, || total_accepts(&added) > MAX_ADDNODE_CONNECTIONS, "a ninth dial").await; + settle(&pm).await; + assert_eq!(total_accepts(&added), MAX_ADDNODE_CONNECTIONS + 1); +} + +/// `addnode onetry` is outside every limit, as Core's takes no grant, and +/// answers success for a peer already connected, as Core's answers null. +/// satd refused it with "max outbound connections reached" once the +/// automatic peers were all connected, and with "already connected". +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn onetry_is_outside_the_limits_and_succeeds_for_a_connected_peer() { + let (pm, _dir) = mk_pm(); + leave_ibd(&pm); + fill_automatic(&pm, 8); + let l = Listener::bind().await; + let target = PeerAddr::Socket(l.addr); + + pm.connect_peer_addr_onetry(&target, None).await.expect("onetry with the automatic slots full"); + wait_for(&pm, || l.accepts() == 1, "the onetry dial reaches the listener").await; + + pm.connect_peer_addr_onetry(&target, None).await.expect("onetry to a connected peer"); + settle(&pm).await; + assert_eq!(l.accepts(), 1, "no second connection"); + + // A dial that fails is still reported. + let closed = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let closed_addr = closed.local_addr().unwrap(); + drop(closed); + assert!(pm.connect_peer_addr_onetry(&PeerAddr::Socket(closed_addr), None).await.is_err()); +} + +/// A peer the operator named waits at most a minute between dials, Core's +/// added-node retry interval. satd backed it off to five minutes, like a +/// learned address. +#[test] +fn a_manual_peers_backoff_stops_at_a_minute() { + let mut manual = ReconnectState::new(); + let mut automatic = ReconnectState::new(); + for _ in 0..10 { + manual.record_failure(MANUAL_RECONNECT_MAX_BACKOFF); + automatic.record_failure(AUTOMATIC_RECONNECT_MAX_BACKOFF); + } + assert_eq!(manual.backoff_duration(MANUAL_RECONNECT_MAX_BACKOFF), Duration::from_secs(60)); + assert_eq!(automatic.backoff_duration(AUTOMATIC_RECONNECT_MAX_BACKOFF), Duration::from_secs(300)); +} + +/// A failed dial backs off, and in the reconnect loop that is per address: +/// a manual peer's failure uses the manual cap. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_failed_manual_redial_backs_off_by_the_manual_cap() { + let (pm, _dir) = mk_pm(); + leave_ibd(&pm); + let closed = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = closed.local_addr().unwrap(); + drop(closed); + assert!(pm.addnode_add(&addr.to_string(), PeerAddr::Socket(addr))); + for _ in 0..10 { + pm.reconnect_backoff.write().entry(addr).or_insert_with(ReconnectState::new).next_attempt = + Instant::now(); + pm.reconnect_tick(); + settle(&pm).await; + } + let backoff = pm.reconnect_backoff.read(); + let state = backoff.get(&addr).expect("a backoff entry"); + assert_eq!(state.attempts, 10); + assert!( + state.next_attempt <= Instant::now() + MANUAL_RECONNECT_MAX_BACKOFF, + "the next dial is at most a minute away" + ); +} + +/// Backoff ends when a dial to that address connects, and not otherwise. +/// satd cleared every address's backoff whenever the tip moved, which +/// during IBD is every tick, so nothing ever backed off. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_new_tip_leaves_reconnect_backoff_alone() { + use crate::chain::state::tests::{build_test_block, make_chain_state}; + + let (cs, dir) = make_chain_state(); + let chain_state = Arc::new(cs); + let pm = over(chain_state.clone()); + let waiting = SocketAddr::from(([198, 51, 100, 9], 8333)); + { + let mut backoff = pm.reconnect_backoff.write(); + let state = backoff.entry(waiting).or_insert_with(ReconnectState::new); + for _ in 0..3 { + state.record_failure(AUTOMATIC_RECONNECT_MAX_BACKOFF); + } + } + let genesis = bitcoin::constants::genesis_block(Network::Regtest).block_hash(); + chain_state + .accept_block(&build_test_block(genesis, 1, 1_707_000_000)) + .expect("connect block 1"); + + let pm_run = pm.clone(); + let run = tokio::spawn(async move { pm_run.run().await }); + // The run loop sees the new tip on its first maintenance pass, 500 ms in. + tokio::time::sleep(Duration::from_secs(2)).await; + let attempts = pm.reconnect_backoff.read().get(&waiting).map(|s| s.attempts); + run.abort(); + *pm.ibd.write() = None; + let _ = std::fs::remove_dir_all(&dir); + assert_eq!(attempts, Some(3), "a new tip does not clear backoff"); +} + +/// The other half: a dial that connects clears its address's backoff. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_dial_that_connects_clears_its_backoff() { + let (pm, _dir) = mk_pm(); + leave_ibd(&pm); + let l = Listener::bind().await; + pm.learned_addrs.write().push(l.addr); + { + let mut backoff = pm.reconnect_backoff.write(); + let state = backoff.entry(l.addr).or_insert_with(ReconnectState::new); + state.record_failure(AUTOMATIC_RECONNECT_MAX_BACKOFF); + state.record_failure(AUTOMATIC_RECONNECT_MAX_BACKOFF); + state.next_attempt = Instant::now(); + } + pm.reconnect_tick(); + wait_for(&pm, || l.accepts() == 1, "the address is dialled").await; + settle(&pm).await; + assert_eq!(pm.reconnect_backoff.read().get(&l.addr).map(|s| s.attempts), Some(0)); +} + +// ---- Bans and dials ---- + +fn ban(pm: &PeerManager, target: crate::net::ban::BanTarget) { + let now = crate::time::now_secs(); + pm.ban_list.write().add(&target, now, now + 3600).expect("ban"); +} + +/// Core checks bans for automatic connections only: a manual one is opened +/// by the name the operator gave, banned or not (`OpenNetworkConnection`, +/// `net.cpp`). satd skipped a banned added node in the reconnect loop. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_banned_manual_peer_is_dialled_and_a_banned_learned_one_is_not() { + let (pm, _dir) = mk_pm(); + leave_ibd(&pm); + let manual = Listener::bind().await; + let learned = Listener::bind().await; + assert!(pm.addnode_add(&manual.addr.to_string(), PeerAddr::Socket(manual.addr))); + pm.learned_addrs.write().push(learned.addr); + ban(&pm, crate::net::ban::BanTarget::Net(ipnet::IpNet::from(IpAddr::from([127, 0, 0, 1])))); + + pm.reconnect_tick(); + wait_for(&pm, || manual.accepts() == 1, "the banned added node is dialled").await; + settle(&pm).await; + assert_eq!(learned.accepts(), 0, "the banned learned address is not"); + let err = pm + .connect_peer_addr_automatic(&PeerAddr::Socket(learned.addr)) + .await + .expect_err("nor does a seed dial reach it"); + assert!(err.contains("banned"), "{err}"); + assert_eq!(learned.accepts(), 0); +} + +/// The same for onion peers. satd never looked at the ban of an onion host +/// in the reconnect loop or when gossip named one, so `setban .onion`, +/// or a ban for misbehaviour, did not stop automatic dials to it. With no +/// proxy here every onion dial fails, which records a backoff entry: that +/// entry is the trace of a dial attempted. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_banned_onion_host_is_neither_learned_nor_dialled_automatically() { + let (pm, _dir) = mk_pm(); + leave_ibd(&pm); + let host = |c: char| format!("{}.onion", c.to_string().repeat(56)); + let (banned, open, named) = (host('a'), host('b'), host('c')); + + // Learned before the ban, so only the reconnect loop can hold it back. + pm.add_onion_connect_addr(banned.clone(), 8333); + pm.add_onion_connect_addr(open.clone(), 8333); + assert!(pm.add_peer_addr(PeerAddr::Onion { host: named.clone(), port: 8333 })); + for h in [&banned, &named] { + ban(&pm, crate::net::ban::BanTarget::Onion(h.clone())); + } + + pm.reconnect_tick(); + let tried = |h: &str| pm.onion_reconnect_backoff.read().contains_key(&format!("{h}:8333")); + wait_for(&pm, || tried(&open) && tried(&named), "the open host and the named one are dialled").await; + settle(&pm).await; + assert!(!tried(&banned), "the banned learned host is not"); + + // Gossip naming a banned host does not add it. + let later = host('d'); + ban(&pm, crate::net::ban::BanTarget::Onion(later.clone())); + pm.add_onion_connect_addr(later.clone(), 8333); + pm.add_onion_connect_addr(host('e'), 8333); + let listed = |h: &str| { + pm.connect_peer_addrs.read().iter().any(|a| matches!(a, PeerAddr::Onion { host, .. } if host == h)) + }; + assert!(!listed(&later), "a banned onion host is not learned"); + assert!(listed(&host('e')), "an open one is"); +} + +/// `setban .onion` disconnects a connected peer at that onion host, as +/// Core's `setban` disconnects the peers the ban covers. satd matched a ban +/// against peers' IP addresses only, and an onion peer's is the `0.0.0.0` +/// placeholder, so the peer stayed. +#[test] +fn setban_of_an_onion_host_disconnects_that_peer_only() { + let (pm, _dir) = mk_pm(); + let placeholder = SocketAddr::from(([0, 0, 0, 0], 8333)); + let host = |c: char| format!("{}.onion", c.to_string().repeat(56)); + for (id, h) in [(1, host('a')), (2, host('b'))] { + let mut info = peer(id, placeholder, Direction::Outbound); + info.onion_host = Some(h); + add_peer(&pm, info); + } + add_peer(&pm, peer(3, SocketAddr::from(([198, 51, 100, 3], 8333)), Direction::Outbound)); + let now = crate::time::now_secs(); + pm.set_ban_subnet(&crate::net::ban::BanTarget::Onion(host('a')), true, now, now + 3600).unwrap(); + let ids: std::collections::BTreeSet = pm.peers.read().keys().copied().collect(); + assert_eq!(ids, [2, 3].into_iter().collect(), "only the banned onion peer is disconnected"); + + // A ban of an IP range does not reach onion peers through their placeholder. + pm.set_ban_subnet( + &crate::net::ban::BanTarget::Net("0.0.0.0/8".parse().unwrap()), + true, + now, + now + 3600, + ) + .unwrap(); + assert!(pm.peers.read().contains_key(&2), "the other onion peer stays"); +} + +/// A full address book makes room for a new address by dropping a terrible +/// entry (tried three times, never reached), as Core's does, and the dropped +/// address leaves the dial list. satd's book refused every new address once +/// full, for good. With nothing terrible to drop it still refuses. +#[test] +fn a_full_book_replaces_a_terrible_entry() { + let (pm, _dir) = mk_pm(); + let source = IpAddr::from([198, 51, 100, 1]); + let full = crate::net::addrman::MAX_ENTRIES; + // 64 per /16, across enough groups to fill the book. + let addr_at = |i: usize| SocketAddr::from(([11 + (i / 64 / 256) as u8, ((i / 64) % 256) as u8, 1, (i % 64) as u8 + 1], 8333)); + for i in 0..full { + pm.add_learned_addr_from(addr_at(i), 9, source); + } + assert_eq!(pm.addrman.read().len(), full, "fixture: the book is full"); + + let fresh = SocketAddr::from(([99, 1, 1, 1], 8333)); + pm.add_learned_addr_from(fresh, 9, source); + assert!(!pm.addrman.read().contains(&fresh), "nothing terrible: refused"); + + let stale = addr_at(0); + for _ in 0..3 { + pm.addrman.write().mark_attempt(stale); + } + pm.add_learned_addr_from(fresh, 9, source); + assert!(pm.addrman.read().contains(&fresh), "the new address takes the terrible one's place"); + assert!(!pm.addrman.read().contains(&stale)); + assert_eq!(pm.addrman.read().len(), full); + assert!(pm.learned_addrs.read().contains(&fresh), "and is dialled"); + assert!(!pm.learned_addrs.read().contains(&stale), "the dropped one is not"); +} + +/// A failed redial counts as an attempt in the address book: what makes an +/// unreachable address terrible. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_failed_redial_counts_as_an_attempt() { + let (pm, _dir) = mk_pm(); + leave_ibd(&pm); + let closed = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = closed.local_addr().unwrap(); + drop(closed); + pm.add_learned_addr_from(addr, 9, IpAddr::from([198, 51, 100, 1])); + pm.reconnect_tick(); + wait_for(&pm, || pm.reconnect_backoff.read().contains_key(&addr), "the dial fails").await; + settle(&pm).await; + let attempts = pm.addrman.read().iter().find(|e| e.addr == addr).map(|e| e.attempts); + assert_eq!(attempts, Some(1)); +} + +/// The same within a network group: a full group (64 new addresses) makes +/// room by dropping a terrible entry of that group, and the dropped +/// address's reconnect backoff goes with it. +#[test] +fn a_full_group_replaces_a_terrible_entry_and_its_backoff() { + let (pm, _dir) = mk_pm(); + let source = IpAddr::from([198, 51, 100, 1]); + let in_group = |i: u8| SocketAddr::from(([44, 1, 0, i + 1], 8333)); + for i in 0..64 { + pm.add_learned_addr_from(in_group(i), 9, source); + } + let fresh = in_group(100); + pm.add_learned_addr_from(fresh, 9, source); + assert!(!pm.addrman.read().contains(&fresh), "nothing terrible: refused"); + + let stale = in_group(0); + for _ in 0..3 { + pm.addrman.write().mark_attempt(stale); + } + pm.reconnect_backoff.write().insert(stale, ReconnectState::new()); + pm.add_learned_addr_from(fresh, 9, source); + assert!(pm.addrman.read().contains(&fresh)); + assert!(!pm.addrman.read().contains(&stale)); + assert!(pm.learned_addrs.read().contains(&fresh)); + assert!(!pm.learned_addrs.read().contains(&stale)); + assert!(!pm.reconnect_backoff.read().contains_key(&stale), "its backoff is dropped"); +} + +/// A book full of tried addresses, all unseen for over 30 days, still makes +/// room for a new one: Core keeps its tried table apart from its new one, so +/// tried entries never shut new addresses out. +#[test] +fn a_book_full_of_stale_tried_entries_takes_a_new_address() { + let (pm, _dir) = mk_pm(); + let full = crate::net::addrman::MAX_ENTRIES; + let long_ago = crate::time::now_secs() - 31 * 24 * 60 * 60; + { + let mut book = pm.addrman.write(); + for i in 0..full { + let a = SocketAddr::from(([11 + (i / 65536) as u8, ((i / 256) % 256) as u8, (i % 256) as u8, 1], 8333)); + book.mark_good(a, long_ago); + } + assert_eq!(book.len(), full, "fixture: the book is full of tried entries"); + } + let fresh = SocketAddr::from(([99, 1, 1, 1], 8333)); + pm.add_learned_addr_from(fresh, 9, IpAddr::from([198, 51, 100, 1])); + assert!(pm.addrman.read().contains(&fresh)); + assert!(pm.learned_addrs.read().contains(&fresh)); + assert_eq!(pm.addrman.read().len(), full); +} diff --git a/node/src/net/mod.rs b/node/src/net/mod.rs index 75062f31a..2102537bb 100644 --- a/node/src/net/mod.rs +++ b/node/src/net/mod.rs @@ -1,3 +1,4 @@ +pub mod addr_rate; pub mod addrman; pub mod ban; pub mod asmap; diff --git a/node/src/net/peer.rs b/node/src/net/peer.rs index f92ce064e..64f8256ed 100644 --- a/node/src/net/peer.rs +++ b/node/src/net/peer.rs @@ -263,6 +263,9 @@ pub struct PeerInfo { /// lazily, on the first addr-related message the peer sends, so a peer /// that never participates in addr relay is never counted as doing so. pub addr_relay_enabled: bool, + /// The rate limit on the addresses this peer announces, and its + /// `getpeerinfo` counters: Core's address token bucket. + pub addr_rate: crate::net::addr_rate::AddrTokenBucket, /// Peer's minimum fee rate for tx relay (BIP 133 feefilter), in sat/kvB. pub fee_filter: u64, pub conn_time: SystemTime, @@ -330,6 +333,7 @@ impl PeerInfo { wants_addrv2: false, wtxid_relay: false, addr_relay_enabled: false, + addr_rate: crate::net::addr_rate::AddrTokenBucket::new(crate::time::now_micros()), fee_filter: 0, conn_time: std::time::UNIX_EPOCH + std::time::Duration::from_secs(crate::time::now_secs()), permissions: crate::net::permissions::NetPermissions::NONE, @@ -503,8 +507,8 @@ impl PeerInfo { // permissions granted", not "not implemented", which is the // opposite claim for a `-whitelist`ed peer. "permissions": self.permissions.to_strings(), - "addr_processed": 0, - "addr_rate_limited": 0, + "addr_processed": self.addr_rate.processed, + "addr_rate_limited": self.addr_rate.rate_limited, // Core withholds address relay from block-relay-only peers -- // that is the whole point of the connection type. For everyone // else this is latched by `SetupAddressRelay`, not derived from diff --git a/node/src/rpc/server.rs b/node/src/rpc/server.rs index 9b7a7cf24..13ad75b85 100644 --- a/node/src/rpc/server.rs +++ b/node/src/rpc/server.rs @@ -2922,14 +2922,16 @@ pub async fn start( } } "onetry" => { - // A single, un-remembered attempt — block on it and surface the - // result, matching the prior satd behavior (now onion-capable). + // A single, un-remembered attempt — block on it and surface a + // failed dial, as satd always has (now onion-capable). Like + // Core's, it is outside the outbound limits, and a peer + // already connected answers null. let addr = ctx.peer_manager .resolve_peer_target(&addr_str, crate::net::peer::default_p2p_port(ctx.chain_state.network)) .await .map_err(|e| ErrorObjectOwned::owned(-1, e, None::<()>))?; ctx.peer_manager - .connect_peer_addr_with(&addr, v2transport) + .connect_peer_addr_onetry(&addr, v2transport) .await .map_err(|e| ErrorObjectOwned::owned(-1, e, None::<()>))?; } diff --git a/satd/tests/p2p_addr_dials.rs b/satd/tests/p2p_addr_dials.rs new file mode 100644 index 000000000..fa25059f2 --- /dev/null +++ b/satd/tests/p2p_addr_dials.rs @@ -0,0 +1,315 @@ +//! Addresses a peer hands a regtest node, and the manual connections it +//! opens with its automatic outbound slots full: Bitcoin Core's address rate +//! limit (`net_processing.cpp` `ProcessAddrs`) and its separate limits for +//! manual connections (`net.cpp` `ThreadOpenAddedConnections`, `rpc/net.cpp` +//! `addnode`), exercised over raw connections. + +mod common; + +use bitcoin::consensus::{deserialize, serialize}; +use bitcoin::p2p::message::{NetworkMessage, RawNetworkMessage}; +use bitcoin::p2p::message_network::VersionMessage; +use bitcoin::p2p::{Address, Magic, ServiceFlags}; +use common::{TestNode, find_available_port, poll_until, test_timeout}; +use serde_json::json; +use std::io::{self, Read, Write}; +use std::net::{SocketAddr, TcpListener, TcpStream}; +use std::time::{Duration, Instant}; + +const ADDR: &str = "bcrt1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqdku202"; + +/// A raw peer, read on the test's own thread. +struct Peer { + stream: TcpStream, +} + +impl Peer { + /// Connect to `port` and complete the handshake. + fn connect(port: u16) -> Self { + let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap(); + let deadline = Instant::now() + Duration::from_secs(30); + let stream = loop { + match TcpStream::connect_timeout(&addr, Duration::from_secs(2)) { + Ok(s) => break s, + Err(_) if Instant::now() < deadline => std::thread::sleep(Duration::from_millis(100)), + Err(e) => panic!("connect to {addr}: {e}"), + } + }; + Self::handshake(stream).expect("handshake with the node") + } + + /// Accept the connection the node dials to `listener` within `timeout`. + fn accept(listener: &TcpListener, timeout: Duration) -> Option { + listener.set_nonblocking(true).unwrap(); + let deadline = Instant::now() + timeout; + let stream = loop { + match listener.accept() { + Ok((s, _)) => break s, + Err(e) if e.kind() == io::ErrorKind::WouldBlock && Instant::now() < deadline => { + std::thread::sleep(Duration::from_millis(20)) + } + Err(_) => return None, + } + }; + stream.set_nonblocking(false).unwrap(); + Some(Self::handshake(stream).expect("handshake with the dialling node")) + } + + fn handshake(stream: TcpStream) -> io::Result { + stream.set_nodelay(true).unwrap(); + stream.set_write_timeout(Some(Duration::from_secs(30))).unwrap(); + let mut peer = Peer { stream }; + peer.send(NetworkMessage::Version(version())); + let (mut got_version, mut got_verack) = (false, false); + while !(got_version && got_verack) { + match peer.recv(Duration::from_secs(30))? { + NetworkMessage::Version(_) => { + got_version = true; + peer.send(NetworkMessage::Verack); + } + NetworkMessage::Verack => got_verack = true, + _ => {} + } + } + Ok(peer) + } + + fn send(&mut self, msg: NetworkMessage) { + let _ = self.stream.write_all(&serialize(&RawNetworkMessage::new(Magic::REGTEST, msg))); + } + + fn recv(&mut self, timeout: Duration) -> io::Result { + self.stream.set_read_timeout(Some(timeout)).unwrap(); + let mut header = [0u8; 24]; + self.stream.read_exact(&mut header)?; + let len = u32::from_le_bytes(header[16..20].try_into().unwrap()) as usize; + let mut buf = header.to_vec(); + buf.resize(24 + len, 0); + self.stream.read_exact(&mut buf[24..])?; + let raw: RawNetworkMessage = + deserialize(&buf).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))?; + Ok(raw.into_payload()) + } + + /// Answer pings until the node sends a message `pred` accepts. + fn recv_until(&mut self, timeout: Duration, pred: impl Fn(&NetworkMessage) -> bool) -> Option { + let deadline = Instant::now() + timeout; + loop { + let left = deadline.saturating_duration_since(Instant::now()).max(Duration::from_millis(1)); + match self.recv(left) { + Ok(NetworkMessage::Ping(n)) => self.send(NetworkMessage::Pong(n)), + Ok(m) if pred(&m) => return Some(m), + Ok(_) => {} + Err(_) => return None, + } + } + } + + /// Send `msg` and wait until the node has handled it: the node handles + /// one peer's messages in order, so the `pong` to a `ping` sent after it + /// comes once `msg` is done. + fn send_and_sync(&mut self, msg: NetworkMessage) { + self.send(msg); + let nonce = rand_nonce(); + self.send(NetworkMessage::Ping(nonce)); + assert!( + self.recv_until(test_timeout(30), |m| matches!(m, NetworkMessage::Pong(n) if *n == nonce)) + .is_some(), + "the node answers the ping" + ); + } +} + +fn version() -> VersionMessage { + let services = ServiceFlags::NETWORK | ServiceFlags::WITNESS; + let zero: SocketAddr = "0.0.0.0:0".parse().unwrap(); + VersionMessage { + version: 70016, + services, + timestamp: 1_760_000_000, + receiver: Address::new(&zero, ServiceFlags::NONE), + sender: Address::new(&zero, services), + nonce: rand_nonce(), + user_agent: "/p2p-addr-dials-test/".into(), + start_height: 0, + relay: true, + } +} + +/// A distinct nonce per call, so the node's self-connection check never +/// mistakes one raw peer for another. +fn rand_nonce() -> u64 { + use std::hash::{BuildHasher, Hasher}; + static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + let mut h = std::collections::hash_map::RandomState::new().build_hasher(); + h.write_u64(COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed)); + h.finish() +} + +/// An `addr` of `n` full-node addresses, each in a /16 of its own (so the +/// address book's per-group limit does not apply), from first octet `base`. +fn addr_msg(base: u8, n: usize) -> NetworkMessage { + let services = ServiceFlags::NETWORK | ServiceFlags::WITNESS; + NetworkMessage::Addr( + (0..n) + .map(|i| { + let a = SocketAddr::from(([base + (i / 250) as u8, (i % 250) as u8, 7, 7], 8333)); + (1_760_000_000, Address::new(&a, services)) + }) + .collect(), + ) +} + +/// The one `getpeerinfo` entry for a peer that reached the node on `port`. +fn peer_on(node: &TestNode, port: u16) -> serde_json::Value { + let suffix = format!(":{port}"); + let peers: Vec = node + .rpc_ok("getpeerinfo", vec![]) + .as_array() + .unwrap() + .iter() + .filter(|p| p["addrbind"].as_str().is_some_and(|b| b.ends_with(&suffix))) + .cloned() + .collect(); + assert_eq!(peers.len(), 1, "one peer on {port}: {peers:?}"); + peers[0].clone() +} + +fn addr_counts(node: &TestNode, port: u16) -> (u64, u64) { + let p = peer_on(node, port); + (p["addr_processed"].as_u64().unwrap(), p["addr_rate_limited"].as_u64().unwrap()) +} + +/// The addresses `addr_msg` announced that the address book holds (the +/// book also holds the outbound peers the node connected to). +fn book_size(node: &TestNode) -> usize { + node.rpc_ok("getnodeaddresses", vec![json!(0)]) + .as_array() + .unwrap() + .iter() + .filter(|e| e["address"].as_str().is_some_and(|a| a.ends_with(".7.7"))) + .count() +} + +fn now_secs() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs() +} + +/// Core meters the addresses each peer announces with a token bucket: one +/// token to start, then 0.1 a second, and a peer with the `addr` permission +/// is not metered (`ProcessAddrs`; Core's `p2p_addr_relay.py` +/// `rate_limit_tests` makes the same steps). satd took every entry of every +/// message and reported both `getpeerinfo` counters as 0. +#[test] +fn a_peers_addresses_are_rate_limited_unless_it_has_the_addr_permission() { + let p2p = find_available_port(); + let exempt_port = find_available_port(); + let node = TestNode::start(&[ + &format!("--port={p2p}"), + &format!("--bind=127.0.0.1:{p2p}"), + &format!("--whitebind=addr@127.0.0.1:{exempt_port}"), + ]); + let start = now_secs() - 1000; + node.rpc_ok("setmocktime", vec![json!(start)]); + + let mut limited = Peer::connect(p2p); + limited.send_and_sync(addr_msg(1, 1000)); + assert_eq!(addr_counts(&node, p2p), (1, 999), "one token: one address"); + assert_eq!(book_size(&node), 1); + + // A hundred seconds earn ten more. + node.rpc_ok("setmocktime", vec![json!(start + 100)]); + limited.send_and_sync(addr_msg(20, 1000)); + assert_eq!(addr_counts(&node, p2p), (11, 1989)); + assert_eq!(book_size(&node), 11); + + let mut exempt = Peer::connect(exempt_port); + exempt.send_and_sync(addr_msg(40, 1000)); + assert_eq!(addr_counts(&node, exempt_port), (1000, 0), "the addr permission takes them all"); + assert_eq!(book_size(&node), 1011); +} + +/// Core adds a thousand tokens when it sends a peer `getaddr`, so the +/// answer is taken whole. An addr-fetch connection exists for that answer. +#[test] +fn an_addr_fetch_peers_answer_is_taken_whole() { + let node = TestNode::start(&["--v2transport=0"]); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let target = listener.local_addr().unwrap().to_string(); + node.rpc_ok("addconnection", vec![json!(target), json!("addr-fetch"), json!(false)]); + let mut peer = Peer::accept(&listener, test_timeout(30)).expect("the node dials the addr-fetch peer"); + assert!( + peer.recv_until(test_timeout(30), |m| matches!(m, NetworkMessage::GetAddr)).is_some(), + "the node asks for addresses" + ); + peer.send(addr_msg(1, 600)); + poll_until(|| book_size(&node) == 600, test_timeout(30), "all 600 addresses are learned"); +} + +/// Core counts manual connections apart from the automatic outbound target: +/// `addnode onetry` takes no grant at all, and added nodes have their own +/// limit of eight (`semAddnode`), redialled every minute however many +/// automatic peers there are. satd counted manual connections against the +/// automatic target, so once it was full `onetry` failed with "max outbound +/// connections reached" and an added node whose first dial failed was never +/// dialled again; an `onetry` to a connected peer failed with "already +/// connected", where Core answers null. +#[test] +fn manual_connections_are_outside_the_automatic_outbound_target() { + // `-maxconnections=3` caps the automatic target at three. + let mut node = TestNode::start(&["--v2transport=0", "--maxconnections=3"]); + // Out of IBD: the target is 64 there. + node.mine_blocks(1, ADDR); + + let mut automatic = Vec::new(); + for _ in 0..3 { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let target = listener.local_addr().unwrap().to_string(); + node.rpc_ok("addconnection", vec![json!(target), json!("outbound-full-relay"), json!(false)]); + automatic.push(Peer::accept(&listener, test_timeout(30)).expect("an automatic peer")); + } + let outbound = |node: &TestNode| { + node.rpc_ok("getpeerinfo", vec![]) + .as_array() + .unwrap() + .iter() + .filter(|p| p["inbound"] == json!(false)) + .map(|p| (p["addr"].as_str().unwrap().to_string(), p["connection_type"].as_str().unwrap().to_string())) + .collect::>() + }; + poll_until(|| outbound(&node).len() == 3, test_timeout(20), "three automatic peers"); + + // `onetry`, with the automatic slots full. + let one = TcpListener::bind("127.0.0.1:0").unwrap(); + let one_addr = one.local_addr().unwrap().to_string(); + let out = node.rpc_call_with_params("addnode", vec![json!(one_addr), json!("onetry")]).unwrap(); + assert!(out["error"].is_null(), "onetry is outside the automatic target: {out}"); + let _one = Peer::accept(&one, test_timeout(30)).expect("the onetry dial"); + poll_until( + || outbound(&node).contains(&(one_addr.clone(), "manual".to_string())), + test_timeout(20), + "the onetry peer is a manual connection", + ); + // And again, to the connected peer: null, as in Core. + let out = node.rpc_call_with_params("addnode", vec![json!(one_addr), json!("onetry")]).unwrap(); + assert!(out["error"].is_null(), "onetry to a connected peer answers null: {out}"); + assert_eq!(out["result"], json!(null)); + + // An added node whose first dial fails is dialled again. + let port = find_available_port(); + let added = format!("127.0.0.1:{port}"); + node.rpc_ok("addnode", vec![json!(added), json!("add")]); + std::thread::sleep(Duration::from_secs(1)); + let listener = TcpListener::bind(&added).unwrap(); + let _added = Peer::accept(&listener, test_timeout(40)).expect("the added node is dialled again"); + poll_until( + || outbound(&node).contains(&(added.clone(), "manual".to_string())), + test_timeout(20), + "the added node is a manual connection", + ); + assert_eq!(outbound(&node).len(), 5, "three automatic peers and two manual ones"); + node.stop(); +} diff --git a/satd/tests/regtest.rs b/satd/tests/regtest.rs index 3cea39e4c..2f8043682 100644 --- a/satd/tests/regtest.rs +++ b/satd/tests/regtest.rs @@ -25392,7 +25392,9 @@ fn an_address_that_names_no_host_is_neither_learned_nor_dialled() { use serde_json::json; use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr}; - let node = TestNode::start(&["--v2transport=0"]); + // `addr`: the gossiper's addresses are not rate limited (Core's + // address token bucket), so all six entries are looked at. + let node = TestNode::start(&["--v2transport=0", "--whitelist=addr@127.0.0.1"]); // Where a dial to `0.0.0.0:` lands. let local = Inbound::bind(); let port = local.addr.port(); @@ -25545,6 +25547,8 @@ fn off_regtest_a_gossiped_private_address_is_neither_learned_nor_dialled() { "--dnsseed=0", "--fixedseeds=0", &format!("--addnode={}", named.addr), + // Both announced addresses are looked at, past the rate limit. + "--whitelist=addr@127.0.0.1", ], ); let stream = gossip::take_dial(&named, Magic::SIGNET);