Line data Source code
1 : // Copyright (c) 2009-2010 Satoshi Nakamoto
2 : // Copyright (c) 2009-2021 The Bitcoin Core developers
3 : // Copyright (c) 2014-2025 The Dash Core developers
4 : // Distributed under the MIT software license, see the accompanying
5 : // file COPYING or http://www.opensource.org/licenses/mit-license.php.
6 :
7 : #if defined(HAVE_CONFIG_H)
8 : #include <config/bitcoin-config.h>
9 : #endif
10 :
11 : #include <net.h>
12 : #include <netmessagemaker.h>
13 :
14 : #include <addrdb.h>
15 : #include <addrman.h>
16 : #include <banman.h>
17 : #include <clientversion.h>
18 : #include <compat/compat.h>
19 : #include <consensus/consensus.h>
20 : #include <crypto/sha256.h>
21 : #include <node/eviction.h>
22 : #include <fs.h>
23 : #include <i2p.h>
24 : #include <key.h>
25 : #include <memusage.h>
26 : #include <net_permissions.h>
27 : #include <netaddress.h>
28 : #include <netbase.h>
29 : #include <node/interface_ui.h>
30 : #include <protocol.h>
31 : #include <random.h>
32 : #include <scheduler.h>
33 : #include <util/sock.h>
34 : #include <util/strencodings.h>
35 : #include <util/system.h>
36 : #include <util/thread.h>
37 : #include <util/threadinterrupt.h>
38 : #include <util/time.h>
39 : #include <util/trace.h>
40 : #include <util/translation.h>
41 :
42 : #include <evo/deterministicmns.h>
43 : #include <masternode/meta.h>
44 : #include <masternode/sync.h>
45 : #include <stats/client.h>
46 : #include <util/std23.h>
47 : #include <util/vector.h>
48 : #include <util/wpipe.h>
49 :
50 : #ifdef WIN32
51 : #include <string.h>
52 : #endif
53 :
54 : #if HAVE_DECL_GETIFADDRS && HAVE_DECL_FREEIFADDRS
55 : #include <ifaddrs.h>
56 : #endif
57 :
58 : #include <algorithm>
59 : #include <array>
60 : #include <cstdint>
61 : #include <functional>
62 : #include <unordered_map>
63 :
64 : #include <math.h>
65 :
66 : /** Maximum number of block-relay-only anchor connections */
67 : static constexpr size_t MAX_BLOCK_RELAY_ONLY_ANCHORS = 2;
68 : static_assert (MAX_BLOCK_RELAY_ONLY_ANCHORS <= static_cast<size_t>(MAX_BLOCK_RELAY_ONLY_CONNECTIONS), "MAX_BLOCK_RELAY_ONLY_ANCHORS must not exceed MAX_BLOCK_RELAY_ONLY_CONNECTIONS.");
69 : /** Anchor IP address database file name */
70 : const char* const ANCHORS_DATABASE_FILENAME = "anchors.dat";
71 :
72 : // How often to dump addresses to peers.dat
73 : static constexpr std::chrono::minutes DUMP_PEERS_INTERVAL{15};
74 :
75 : /** Number of DNS seeds to query when the number of connections is low. */
76 : static constexpr int DNSSEEDS_TO_QUERY_AT_ONCE = 3;
77 :
78 : /** How long to delay before querying DNS seeds
79 : *
80 : * If we have more than THRESHOLD entries in addrman, then it's likely
81 : * that we got those addresses from having previously connected to the P2P
82 : * network, and that we'll be able to successfully reconnect to the P2P
83 : * network via contacting one of them. So if that's the case, spend a
84 : * little longer trying to connect to known peers before querying the
85 : * DNS seeds.
86 : */
87 : static constexpr std::chrono::seconds DNSSEEDS_DELAY_FEW_PEERS{11};
88 : static constexpr std::chrono::minutes DNSSEEDS_DELAY_MANY_PEERS{5};
89 : static constexpr int DNSSEEDS_DELAY_PEER_THRESHOLD = 1000; // "many" vs "few" peers
90 :
91 : /** The default timeframe for -maxuploadtarget. 1 day. */
92 : static constexpr std::chrono::seconds MAX_UPLOAD_TIMEFRAME{60 * 60 * 24};
93 :
94 : // A random time period (0 to 1 seconds) is added to feeler connections to prevent synchronization.
95 : static constexpr auto FEELER_SLEEP_WINDOW{1s};
96 :
97 : /** Frequency to attempt extra connections to reachable networks we're not connected to yet **/
98 : static constexpr auto EXTRA_NETWORK_PEER_INTERVAL{5min};
99 :
100 : /** Used to pass flags to the Bind() function */
101 : enum BindFlags {
102 : BF_NONE = 0,
103 : BF_REPORT_ERROR = (1U << 0),
104 : /**
105 : * Do not call AddLocal() for our special addresses, e.g., for incoming
106 : * Tor connections, to prevent gossiping them over the network.
107 : */
108 : BF_DONT_ADVERTISE = (1U << 1),
109 : };
110 :
111 : #ifndef USE_WAKEUP_PIPE
112 : // The set of sockets cannot be modified while waiting
113 : // The sleep time needs to be small to avoid new sockets stalling
114 : static const uint64_t SELECT_TIMEOUT_MILLISECONDS = 50;
115 : #else
116 : // select() is woken up through the wakeup pipe whenever a new node is added, so we can wait much longer.
117 : // We are however still somewhat limited in how long we can sleep as there is periodic work (cleanup) to be done in
118 : // the socket handler thread
119 : static const uint64_t SELECT_TIMEOUT_MILLISECONDS = 500;
120 : #endif /* USE_WAKEUP_PIPE */
121 :
122 : const std::string NET_MESSAGE_TYPE_OTHER = "*other*";
123 :
124 : static const uint64_t RANDOMIZER_ID_NETGROUP = 0x6c0edd8036ef4036ULL; // SHA256("netgroup")[0:8]
125 : static const uint64_t RANDOMIZER_ID_LOCALHOSTNONCE = 0xd93e69e2bbfa5735ULL; // SHA256("localhostnonce")[0:8]
126 : static const uint64_t RANDOMIZER_ID_ADDRCACHE = 0x1cf2e4ddd306dda9ULL; // SHA256("addrcache")[0:8]
127 : //
128 : // Global state variables
129 : //
130 : bool fDiscover = true;
131 : bool fListen = true;
132 : GlobalMutex g_maplocalhost_mutex;
133 146 : std::map<CNetAddr, LocalServiceInfo> mapLocalHost GUARDED_BY(g_maplocalhost_mutex);
134 : std::string strSubVersion;
135 :
136 80 : size_t CSerializedNetMsg::GetMemoryUsage() const noexcept
137 : {
138 : // Don't count the dynamic memory used for the m_type string, by assuming it fits in the
139 : // "small string" optimization area (which stores data inside the object itself, up to some
140 : // size; 15 bytes in modern libstdc++).
141 80 : return sizeof(*this) + memusage::DynamicUsage(data);
142 : }
143 :
144 0 : void CConnman::AddAddrFetch(const std::string& strDest)
145 : {
146 0 : LOCK(m_addr_fetches_mutex);
147 0 : m_addr_fetches.push_back(strDest);
148 0 : }
149 :
150 27 : uint16_t GetListenPort()
151 : {
152 : // If -bind= is provided with ":port" part, use that (first one if multiple are provided).
153 51 : for (const std::string& bind_arg : gArgs.GetArgs("-bind")) {
154 24 : constexpr uint16_t dummy_port = 0;
155 :
156 24 : const std::optional<CService> bind_addr{Lookup(bind_arg, dummy_port, /*fAllowLookup=*/false)};
157 24 : if (bind_addr.has_value() && bind_addr->GetPort() != dummy_port) return bind_addr->GetPort();
158 24 : }
159 :
160 : // Otherwise, if -whitebind= without NetPermissionFlags::NoBan is provided, use that
161 : // (-whitebind= is required to have ":port").
162 24 : for (const std::string& whitebind_arg : gArgs.GetArgs("-whitebind")) {
163 0 : NetWhitebindPermissions whitebind;
164 0 : bilingual_str error;
165 0 : if (NetWhitebindPermissions::TryParse(whitebind_arg, whitebind, error)) {
166 0 : if (!NetPermissions::HasFlag(whitebind.m_flags, NetPermissionFlags::NoBan)) {
167 0 : return whitebind.m_service.GetPort();
168 : }
169 0 : }
170 0 : }
171 :
172 : // Otherwise, if -port= is provided, use that. Otherwise use the default port.
173 24 : return static_cast<uint16_t>(gArgs.GetIntArg("-port", Params().GetDefaultPort()));
174 27 : }
175 :
176 : // Determine the "best" local address for a particular peer.
177 25 : [[nodiscard]] static std::optional<CService> GetLocal(const CNode& peer)
178 : {
179 25 : if (!fListen) return std::nullopt;
180 :
181 25 : std::optional<CService> addr;
182 25 : int nBestScore = -1;
183 25 : int nBestReachability = -1;
184 : {
185 25 : LOCK(g_maplocalhost_mutex);
186 241 : for (const auto& [local_addr, local_service_info] : mapLocalHost) {
187 : // For privacy reasons, don't advertise our privacy-network address
188 : // to other networks and don't advertise our other-network address
189 : // to privacy networks.
190 104 : if (local_addr.GetNetwork() != peer.ConnectedThroughNetwork()
191 71 : && (local_addr.IsPrivacyNet() || peer.IsConnectedThroughPrivacyNet())) {
192 36 : continue;
193 : }
194 35 : const int nScore{local_service_info.nScore};
195 35 : const int nReachability{local_addr.GetReachabilityFrom(peer.addr)};
196 35 : if (nReachability > nBestReachability || (nReachability == nBestReachability && nScore > nBestScore)) {
197 18 : addr.emplace(CService{local_addr, local_service_info.nPort});
198 18 : nBestReachability = nReachability;
199 18 : nBestScore = nScore;
200 18 : }
201 : }
202 25 : }
203 25 : return addr;
204 50 : }
205 :
206 : //! Convert the serialized seeds into usable address objects.
207 0 : static std::vector<CAddress> ConvertSeeds(const std::vector<uint8_t> &vSeedsIn)
208 : {
209 : // It'll only connect to one or two seed nodes because once it connects,
210 : // it'll get a pile of addresses with newer timestamps.
211 : // Seed nodes are given a random 'last seen time' of between one and two
212 : // weeks ago.
213 0 : const auto one_week{7 * 24h};
214 0 : std::vector<CAddress> vSeedsOut;
215 0 : FastRandomContext rng;
216 0 : CDataStream s(vSeedsIn, SER_NETWORK, PROTOCOL_VERSION | ADDRV2_FORMAT);
217 0 : while (!s.eof()) {
218 0 : CService endpoint;
219 0 : s >> endpoint;
220 0 : CAddress addr{endpoint, GetDesirableServiceFlags(NODE_NONE)};
221 0 : addr.nTime = rng.rand_uniform_delay(Now<NodeSeconds>() - one_week, -one_week);
222 0 : LogPrint(BCLog::NET, "Added hardcoded seed: %s\n", addr.ToStringAddrPort());
223 0 : vSeedsOut.push_back(addr);
224 0 : }
225 0 : return vSeedsOut;
226 0 : }
227 :
228 : // Determine the "best" local address for a particular peer.
229 : // If none, return the unroutable 0.0.0.0 but filled in with
230 : // the normal parameters, since the IP may be changed to a useful
231 : // one by discovery.
232 25 : CService GetLocalAddress(const CNode& peer)
233 : {
234 25 : return GetLocal(peer).value_or(CService{CNetAddr(), GetListenPort()});
235 0 : }
236 :
237 0 : static int GetnScore(const CService& addr)
238 : {
239 0 : LOCK(g_maplocalhost_mutex);
240 0 : const auto it = mapLocalHost.find(addr);
241 0 : return (it != mapLocalHost.end()) ? it->second.nScore : 0;
242 0 : }
243 :
244 : // Is our peer's addrLocal potentially useful as an external IP source?
245 4 : [[nodiscard]] static bool IsPeerAddrLocalGood(CNode *pnode)
246 : {
247 4 : CService addrLocal = pnode->GetAddrLocal();
248 4 : return fDiscover && pnode->addr.IsRoutable() && addrLocal.IsRoutable() &&
249 4 : g_reachable_nets.Contains(addrLocal);
250 4 : }
251 :
252 4 : std::optional<CService> GetLocalAddrForPeer(CNode& node)
253 : {
254 4 : CService addrLocal{GetLocalAddress(node)};
255 : // If discovery is enabled, sometimes give our peer the address it
256 : // tells us that it sees us as in case it has a better idea of our
257 : // address than we do.
258 4 : FastRandomContext rng;
259 4 : if (IsPeerAddrLocalGood(&node) && (!addrLocal.IsRoutable() ||
260 0 : rng.randbits((GetnScore(addrLocal) > LOCAL_MANUAL) ? 3 : 1) == 0))
261 : {
262 4 : if (node.IsInboundConn()) {
263 : // For inbound connections, assume both the address and the port
264 : // as seen from the peer.
265 1 : addrLocal = CService{node.GetAddrLocal()};
266 1 : } else {
267 : // For outbound connections, assume just the address as seen from
268 : // the peer and leave the port in `addrLocal` as returned by
269 : // `GetLocalAddress()` above. The peer has no way to observe our
270 : // listening port when we have initiated the connection.
271 3 : addrLocal.SetIP(node.GetAddrLocal());
272 : }
273 4 : }
274 4 : if (addrLocal.IsRoutable())
275 : {
276 4 : LogPrint(BCLog::NET, "Advertising address %s to peer=%d\n", addrLocal.ToStringAddrPort(), node.GetId());
277 4 : return addrLocal;
278 : }
279 : // Address is unroutable. Don't advertise.
280 0 : return std::nullopt;
281 4 : }
282 :
283 : // learn a new local address
284 11 : bool AddLocal(const CService& addr_, int nScore)
285 : {
286 11 : CService addr{MaybeFlipIPv6toCJDNS(addr_)};
287 :
288 11 : if (!addr.IsRoutable() && Params().RequireRoutableExternalIP())
289 0 : return false;
290 :
291 11 : if (!fDiscover && nScore < LOCAL_MANUAL)
292 0 : return false;
293 :
294 11 : if (!g_reachable_nets.Contains(addr))
295 0 : return false;
296 :
297 11 : LogPrintf("AddLocal(%s,%i)\n", addr.ToStringAddrPort(), nScore);
298 :
299 : {
300 11 : LOCK(g_maplocalhost_mutex);
301 11 : const auto [it, is_newly_added] = mapLocalHost.emplace(addr, LocalServiceInfo());
302 11 : LocalServiceInfo &info = it->second;
303 11 : if (is_newly_added || nScore >= info.nScore) {
304 22 : info.nScore = nScore + (is_newly_added ? 0 : 1);
305 11 : info.nPort = addr.GetPort();
306 11 : }
307 11 : }
308 :
309 11 : return true;
310 11 : }
311 :
312 0 : bool AddLocal(const CNetAddr &addr, int nScore)
313 : {
314 0 : return AddLocal(CService(addr, GetListenPort()), nScore);
315 0 : }
316 :
317 11 : void RemoveLocal(const CService& addr)
318 : {
319 11 : LOCK(g_maplocalhost_mutex);
320 11 : LogPrintf("RemoveLocal(%s)\n", addr.ToStringAddrPort());
321 11 : mapLocalHost.erase(addr);
322 11 : }
323 :
324 : /** vote for a local address */
325 0 : bool SeenLocal(const CService& addr)
326 : {
327 0 : LOCK(g_maplocalhost_mutex);
328 0 : const auto it = mapLocalHost.find(addr);
329 0 : if (it == mapLocalHost.end()) return false;
330 0 : ++it->second.nScore;
331 0 : return true;
332 0 : }
333 :
334 :
335 : /** check whether a given address is potentially local */
336 3 : bool IsLocal(const CService& addr)
337 : {
338 3 : LOCK(g_maplocalhost_mutex);
339 3 : return mapLocalHost.count(addr) > 0;
340 3 : }
341 :
342 6 : bool CConnman::AlreadyConnectedToAddress(const CAddress& addr) const
343 : {
344 6 : READ_LOCK(m_nodes_mutex);
345 6 : return FindNode(static_cast<CService>(addr)) != nullptr;
346 6 : }
347 :
348 0 : bool CConnman::CheckIncomingNonce(uint64_t nonce) const
349 : {
350 0 : READ_LOCK(m_nodes_mutex);
351 0 : for (const CNode* pnode : m_nodes) {
352 0 : if (!pnode->fSuccessfullyConnected && !pnode->IsInboundConn() && pnode->GetLocalNonce() == nonce)
353 0 : return false;
354 : }
355 0 : return true;
356 0 : }
357 :
358 : /** Get the bind address for a socket as CAddress */
359 0 : static CAddress GetBindAddress(const Sock& sock)
360 : {
361 0 : CAddress addr_bind;
362 : struct sockaddr_storage sockaddr_bind;
363 0 : socklen_t sockaddr_bind_len = sizeof(sockaddr_bind);
364 0 : if (!sock.GetSockName((struct sockaddr*)&sockaddr_bind, &sockaddr_bind_len)) {
365 0 : addr_bind.SetSockAddr((const struct sockaddr*)&sockaddr_bind);
366 0 : } else {
367 0 : LogPrintLevel(BCLog::NET, BCLog::Level::Warning, "getsockname failed\n");
368 : }
369 0 : return addr_bind;
370 0 : }
371 :
372 10 : CNode* CConnman::ConnectNode(CAddress addrConnect, const char *pszDest, bool fCountFailure, ConnectionType conn_type, bool use_v2transport)
373 : {
374 10 : AssertLockNotHeld(m_unused_i2p_sessions_mutex);
375 10 : assert(conn_type != ConnectionType::INBOUND);
376 :
377 10 : if (pszDest == nullptr) {
378 0 : if (addrConnect.GetPort() == GetListenPort() && IsLocal(addrConnect)) {
379 0 : return nullptr;
380 : }
381 :
382 : // Look for an existing connection
383 0 : if (ExistsNode(static_cast<CService>(addrConnect))) {
384 0 : LogPrintf("Failed to open new connection, already connected\n");
385 0 : return nullptr;
386 : }
387 0 : }
388 :
389 : /// debug print
390 10 : if (fLogIPs) {
391 10 : LogPrintLevel(BCLog::NET, BCLog::Level::Debug, "trying %s connection %s lastseen=%.1fhrs\n",
392 : use_v2transport ? "v2" : "v1",
393 : pszDest ? pszDest : addrConnect.ToStringAddrPort(),
394 : Ticks<HoursDouble>(pszDest ? 0h : Now<NodeSeconds>() - addrConnect.nTime));
395 10 : } else {
396 0 : LogPrintLevel(BCLog::NET, BCLog::Level::Debug, "trying %s connection lastseen=%.1fhrs\n",
397 : use_v2transport ? "v2" : "v1",
398 : Ticks<HoursDouble>(pszDest ? 0h : Now<NodeSeconds>() - addrConnect.nTime));
399 : }
400 :
401 : // Resolve
402 10 : const uint16_t default_port{pszDest != nullptr ? Params().GetDefaultPort(pszDest) :
403 0 : Params().GetDefaultPort()};
404 10 : if (pszDest) {
405 10 : std::vector<CService> resolved{Lookup(pszDest, default_port, fNameLookup && !HaveNameProxy(), 256)};
406 10 : if (!resolved.empty()) {
407 10 : Shuffle(resolved.begin(), resolved.end(), FastRandomContext());
408 : // If the connection is made by name, it can be the case that the name resolves to more than one address.
409 : // We don't want to connect any more of them if we are already connected to one
410 16 : for (const auto& r : resolved) {
411 16 : addrConnect = CAddress{MaybeFlipIPv6toCJDNS(r), NODE_NONE};
412 16 : if (!addrConnect.IsValid()) {
413 0 : LogPrint(BCLog::NET, "Resolver returned invalid address %s for %s\n", addrConnect.ToStringAddrPort(), pszDest);
414 0 : return nullptr;
415 : }
416 : // It is possible that we already have a connection to the IP/port pszDest resolved to.
417 : // In that case, drop the connection that was just created.
418 16 : if (ExistsNode(static_cast<CService>(addrConnect))) {
419 10 : LogPrintf("Not opening a connection to %s, already connected to %s\n", pszDest, addrConnect.ToStringAddrPort());
420 10 : return nullptr;
421 : }
422 : }
423 0 : }
424 10 : }
425 :
426 : // Connect
427 0 : std::unique_ptr<Sock> sock;
428 0 : Proxy proxy;
429 0 : CAddress addr_bind;
430 0 : assert(!addr_bind.IsValid());
431 0 : std::unique_ptr<i2p::sam::Session> i2p_transient_session;
432 :
433 0 : if (addrConnect.IsValid()) {
434 0 : const bool use_proxy{GetProxy(addrConnect.GetNetwork(), proxy)};
435 0 : bool proxyConnectionFailed = false;
436 :
437 0 : if (addrConnect.IsI2P() && use_proxy) {
438 0 : i2p::Connection conn;
439 0 : bool connected{false};
440 :
441 0 : if (m_i2p_sam_session) {
442 0 : connected = m_i2p_sam_session->Connect(addrConnect, conn, proxyConnectionFailed);
443 0 : } else {
444 : {
445 0 : LOCK(m_unused_i2p_sessions_mutex);
446 0 : if (m_unused_i2p_sessions.empty()) {
447 0 : i2p_transient_session =
448 0 : std::make_unique<i2p::sam::Session>(proxy, &interruptNet);
449 0 : } else {
450 0 : i2p_transient_session.swap(m_unused_i2p_sessions.front());
451 0 : m_unused_i2p_sessions.pop();
452 : }
453 0 : }
454 0 : connected = i2p_transient_session->Connect(addrConnect, conn, proxyConnectionFailed);
455 0 : if (!connected) {
456 0 : LOCK(m_unused_i2p_sessions_mutex);
457 0 : if (m_unused_i2p_sessions.size() < MAX_UNUSED_I2P_SESSIONS_SIZE) {
458 0 : m_unused_i2p_sessions.emplace(i2p_transient_session.release());
459 0 : }
460 0 : }
461 : }
462 :
463 0 : if (connected) {
464 0 : sock = std::move(conn.sock);
465 0 : addr_bind = CAddress{conn.me, NODE_NONE};
466 0 : }
467 0 : } else if (use_proxy) {
468 0 : LogPrintLevel(BCLog::PROXY, BCLog::Level::Debug, "Using proxy: %s to connect to %s:%s\n", proxy.ToString(), addrConnect.ToStringAddr(), addrConnect.GetPort());
469 0 : sock = ConnectThroughProxy(proxy, addrConnect.ToStringAddr(), addrConnect.GetPort(), proxyConnectionFailed);
470 0 : } else {
471 : // no proxy needed (none set for target network)
472 0 : sock = ConnectDirectly(addrConnect, conn_type == ConnectionType::MANUAL);
473 : }
474 0 : if (!proxyConnectionFailed) {
475 : // If a connection to the node was attempted, and failure (if any) is not caused by a problem connecting to
476 : // the proxy, mark this as an attempt.
477 0 : addrman.Attempt(addrConnect, fCountFailure);
478 0 : }
479 0 : } else if (pszDest && GetNameProxy(proxy)) {
480 0 : std::string host;
481 0 : uint16_t port{default_port};
482 0 : SplitHostPort(std::string(pszDest), port, host);
483 : bool proxyConnectionFailed;
484 0 : sock = ConnectThroughProxy(proxy, host, port, proxyConnectionFailed);
485 0 : }
486 0 : if (!sock) {
487 0 : return nullptr;
488 : }
489 :
490 : // Add node
491 0 : NodeId id = GetNewNodeId();
492 0 : uint64_t nonce = GetDeterministicRandomizer(RANDOMIZER_ID_LOCALHOSTNONCE).Write(id).Finalize();
493 0 : if (!addr_bind.IsValid()) {
494 0 : addr_bind = GetBindAddress(*sock);
495 0 : }
496 0 : CNode* pnode = new CNode(id,
497 0 : std::move(sock),
498 : addrConnect,
499 0 : CalculateKeyedNetGroup(addrConnect),
500 0 : nonce,
501 : addr_bind,
502 0 : pszDest ? pszDest : "",
503 0 : conn_type,
504 : /*inbound_onion=*/false,
505 0 : CNodeOptions{
506 : .i2p_sam_session = std::move(i2p_transient_session),
507 : .recv_flood_size = nReceiveFloodSize,
508 : .use_v2transport = use_v2transport,
509 : });
510 : pnode->AddRef();
511 : ::g_stats_client->inc("peers.connect", 1.0f);
512 :
513 : // We're making a new connection, harvest entropy from the time (and our peer count)
514 : RandAddEvent((uint32_t)id);
515 :
516 : return pnode;
517 0 : }
518 :
519 0 : void CNode::CloseSocketDisconnect(CConnman* connman)
520 : {
521 0 : fDisconnect = true;
522 0 : LOCK2(connman->cs_mapSocketToNode, m_sock_mutex);
523 0 : if (!m_sock) {
524 0 : return;
525 : }
526 :
527 0 : fHasRecvData = false;
528 0 : fCanSendData = false;
529 :
530 0 : connman->mapSocketToNode.erase(m_sock->Get());
531 : {
532 0 : LOCK(connman->cs_sendable_receivable_nodes);
533 0 : connman->mapReceivableNodes.erase(GetId());
534 0 : connman->mapSendableNodes.erase(GetId());
535 0 : }
536 :
537 0 : if (connman->m_edge_trig_events) {
538 0 : connman->m_edge_trig_events->UnregisterEvents(m_sock->Get());
539 0 : }
540 :
541 0 : LogPrint(BCLog::NET, "disconnecting peer=%d\n", id);
542 0 : m_sock.reset();
543 0 : m_i2p_sam_session.reset();
544 :
545 0 : ::g_stats_client->inc("peers.disconnect", 1.0f);
546 0 : }
547 :
548 0 : void CConnman::AddWhitelistPermissionFlags(NetPermissionFlags& flags, const CNetAddr &addr) const {
549 0 : for (const auto& subnet : vWhitelistedRange) {
550 0 : if (subnet.m_subnet.Match(addr)) NetPermissions::AddFlag(flags, subnet.m_flags);
551 : }
552 0 : }
553 :
554 5 : bool CNode::IsBlockRelayOnly() const {
555 5 : bool ignores_incoming_txs{gArgs.GetBoolArg("-blocksonly", DEFAULT_BLOCKSONLY)};
556 : // Stop processing non-block data early if
557 : // 1) We are in blocks only mode and peer has no relay permission
558 : // 2) This peer is a block-relay-only peer
559 5 : return (ignores_incoming_txs && !HasPermission(NetPermissionFlags::Relay)) || IsBlockOnlyConn();
560 0 : }
561 :
562 8 : CService CNode::GetAddrLocal() const
563 : {
564 8 : AssertLockNotHeld(m_addr_local_mutex);
565 8 : LOCK(m_addr_local_mutex);
566 8 : return addrLocal;
567 8 : }
568 :
569 5 : void CNode::SetAddrLocal(const CService& addrLocalIn) {
570 5 : AssertLockNotHeld(m_addr_local_mutex);
571 5 : LOCK(m_addr_local_mutex);
572 5 : if (addrLocal.IsValid()) {
573 0 : error("Addr local already set for node: %i. Refusing to change from %s to %s", id, addrLocal.ToStringAddrPort(), addrLocalIn.ToStringAddrPort());
574 0 : } else {
575 5 : addrLocal = addrLocalIn;
576 : }
577 5 : }
578 :
579 0 : std::string CNode::GetLogString() const
580 : {
581 0 : return fLogIPs ? addr.ToStringAddrPort() : strprintf("%d", id);
582 : }
583 :
584 76 : Network CNode::ConnectedThroughNetwork() const
585 : {
586 76 : return m_inbound_onion ? NET_ONION : addr.GetNetClass();
587 : }
588 :
589 33 : bool CNode::IsConnectedThroughPrivacyNet() const
590 : {
591 33 : return m_inbound_onion || addr.IsPrivacyNet();
592 : }
593 :
594 : #undef X
595 : #define X(name) stats.name = name
596 0 : void CNode::CopyStats(CNodeStats& stats) const
597 : {
598 0 : stats.nodeid = this->GetId();
599 0 : X(addr);
600 0 : X(addrBind);
601 0 : stats.m_network = ConnectedThroughNetwork();
602 0 : X(m_last_send);
603 0 : X(m_last_recv);
604 0 : X(m_last_tx_time);
605 0 : X(m_last_block_time);
606 0 : X(m_connected);
607 0 : X(nTimeOffset);
608 0 : X(m_addr_name);
609 0 : X(nVersion);
610 : {
611 0 : LOCK(m_subver_mutex);
612 0 : X(cleanSubVer);
613 0 : }
614 0 : stats.fInbound = IsInboundConn();
615 0 : X(m_bip152_highbandwidth_to);
616 0 : X(m_bip152_highbandwidth_from);
617 : {
618 0 : LOCK(cs_vSend);
619 0 : X(mapSendBytesPerMsgType);
620 0 : X(nSendBytes);
621 0 : }
622 : {
623 0 : LOCK(cs_vRecv);
624 0 : X(mapRecvBytesPerMsgType);
625 0 : X(nRecvBytes);
626 0 : Transport::Info info = m_transport->GetInfo();
627 0 : stats.m_transport_type = info.transport_type;
628 0 : if (info.session_id) stats.m_session_id = HexStr(*info.session_id);
629 0 : }
630 0 : X(m_permission_flags);
631 :
632 0 : X(m_last_ping_time);
633 0 : X(m_min_ping_time);
634 :
635 : // Leave string empty if addrLocal invalid (not filled in yet)
636 0 : CService addrLocalUnlocked = GetAddrLocal();
637 0 : stats.addrLocal = addrLocalUnlocked.IsValid() ? addrLocalUnlocked.ToStringAddrPort() : "";
638 :
639 : {
640 0 : LOCK(cs_mnauth);
641 0 : X(verifiedProRegTxHash);
642 0 : X(verifiedPubKeyHash);
643 0 : }
644 0 : X(m_masternode_connection);
645 0 : X(m_conn_type);
646 0 : }
647 : #undef X
648 :
649 3 : bool CNode::ReceiveMsgBytes(Span<const uint8_t> msg_bytes, bool& complete)
650 : {
651 3 : complete = false;
652 3 : const auto time = GetTime<std::chrono::microseconds>();
653 3 : LOCK(cs_vRecv);
654 3 : m_last_recv = std::chrono::duration_cast<std::chrono::seconds>(time);
655 3 : nRecvBytes += msg_bytes.size();
656 6 : while (msg_bytes.size() > 0) {
657 : // absorb network data
658 3 : if (!m_transport->ReceivedBytes(msg_bytes)) {
659 : // Serious transport problem, disconnect from the peer.
660 0 : return false;
661 : }
662 :
663 3 : if (m_transport->ReceivedMessageComplete()) {
664 : // decompose a transport agnostic CNetMessage from the deserializer
665 2 : bool reject_message{false};
666 2 : CNetMessage msg = m_transport->GetReceivedMessage(time, reject_message);
667 2 : if (reject_message) {
668 : // Message deserialization failed. Drop the message but don't disconnect the peer.
669 : // store the size of the corrupt message
670 0 : mapRecvBytesPerMsgType.at(NET_MESSAGE_TYPE_OTHER) += msg.m_raw_message_size;
671 0 : continue;
672 : }
673 :
674 : // Store received bytes per message type.
675 : // To prevent a memory DOS, only allow known message types.
676 2 : auto i = mapRecvBytesPerMsgType.find(msg.m_type);
677 2 : if (i == mapRecvBytesPerMsgType.end()) {
678 0 : i = mapRecvBytesPerMsgType.find(NET_MESSAGE_TYPE_OTHER);
679 0 : }
680 2 : assert(i != mapRecvBytesPerMsgType.end());
681 2 : i->second += msg.m_raw_message_size;
682 2 : ::g_stats_client->count("bandwidth.message." + std::string(msg.m_type) + ".bytesReceived", msg.m_raw_message_size, 1.0f);
683 :
684 : // push the message to the process queue,
685 2 : vRecvMsg.push_back(std::move(msg));
686 :
687 2 : complete = true;
688 2 : }
689 : }
690 :
691 3 : return true;
692 3 : }
693 :
694 472 : V1Transport::V1Transport(const NodeId node_id, int nTypeIn, int nVersionIn) noexcept :
695 236 : m_node_id(node_id), hdrbuf(nTypeIn, nVersionIn), vRecv(nTypeIn, nVersionIn)
696 236 : {
697 : assert(std::size(Params().MessageStart()) == std::size(m_magic_bytes));
698 : std::copy(std::begin(Params().MessageStart()), std::end(Params().MessageStart()), m_magic_bytes);
699 : LOCK(m_recv_mutex);
700 : Reset();
701 118 : }
702 :
703 2 : Transport::Info V1Transport::GetInfo() const noexcept
704 : {
705 2 : return {.transport_type = TransportProtocolType::V1, .session_id = {}};
706 : }
707 :
708 7 : int V1Transport::readHeader(Span<const uint8_t> msg_bytes)
709 : {
710 7 : AssertLockHeld(m_recv_mutex);
711 : // copy data to temporary parsing buffer
712 7 : unsigned int nRemaining = CMessageHeader::HEADER_SIZE - nHdrPos;
713 7 : unsigned int nCopy = std::min<unsigned int>(nRemaining, msg_bytes.size());
714 :
715 7 : memcpy(&hdrbuf[nHdrPos], msg_bytes.data(), nCopy);
716 7 : nHdrPos += nCopy;
717 :
718 : // if header incomplete, exit
719 7 : if (nHdrPos < CMessageHeader::HEADER_SIZE)
720 4 : return nCopy;
721 :
722 : // deserialize to CMessageHeader
723 : try {
724 3 : hdrbuf >> hdr;
725 3 : }
726 : catch (const std::exception&) {
727 0 : LogPrint(BCLog::NET, "Header error: Unable to deserialize, peer=%d\n", m_node_id);
728 0 : return -1;
729 0 : }
730 :
731 : // Check start string, network magic
732 3 : if (memcmp(hdr.pchMessageStart, m_magic_bytes, CMessageHeader::MESSAGE_START_SIZE) != 0) {
733 0 : LogPrint(BCLog::NET, "Header error: Wrong MessageStart %s received, peer=%d\n", HexStr(hdr.pchMessageStart), m_node_id);
734 0 : return -1;
735 : }
736 :
737 : // reject messages larger than MAX_SIZE or MAX_PROTOCOL_MESSAGE_LENGTH
738 3 : if (hdr.nMessageSize > MAX_SIZE || hdr.nMessageSize > MAX_PROTOCOL_MESSAGE_LENGTH) {
739 0 : LogPrint(BCLog::NET, "Header error: Size too large (%s, %u bytes), peer=%d\n", SanitizeString(hdr.GetCommand()), hdr.nMessageSize, m_node_id);
740 0 : return -1;
741 : }
742 :
743 : // switch state to reading message data
744 3 : in_data = true;
745 :
746 3 : return nCopy;
747 7 : }
748 :
749 1 : int V1Transport::readData(Span<const uint8_t> msg_bytes)
750 : {
751 1 : AssertLockHeld(m_recv_mutex);
752 1 : unsigned int nRemaining = hdr.nMessageSize - nDataPos;
753 1 : unsigned int nCopy = std::min<unsigned int>(nRemaining, msg_bytes.size());
754 :
755 1 : if (vRecv.size() < nDataPos + nCopy) {
756 : // Allocate up to 256 KiB ahead, but never more than the total message size.
757 1 : vRecv.resize(std::min(hdr.nMessageSize, nDataPos + nCopy + 256 * 1024));
758 1 : }
759 :
760 1 : hasher.Write(msg_bytes.first(nCopy));
761 1 : memcpy(&vRecv[nDataPos], msg_bytes.data(), nCopy);
762 1 : nDataPos += nCopy;
763 :
764 1 : return nCopy;
765 : }
766 :
767 2 : const uint256& V1Transport::GetMessageHash() const
768 : {
769 2 : AssertLockHeld(m_recv_mutex);
770 2 : assert(CompleteInternal());
771 2 : if (data_hash.IsNull())
772 2 : hasher.Finalize(data_hash);
773 2 : return data_hash;
774 : }
775 :
776 2 : CNetMessage V1Transport::GetReceivedMessage(const std::chrono::microseconds time, bool& reject_message)
777 : {
778 2 : AssertLockNotHeld(m_recv_mutex);
779 : // Initialize out parameter
780 2 : reject_message = false;
781 : // decompose a single CNetMessage from the TransportDeserializer
782 2 : LOCK(m_recv_mutex);
783 2 : CNetMessage msg(std::move(vRecv));
784 :
785 : // store message type string, time, and sizes
786 2 : msg.m_type = hdr.GetCommand();
787 2 : msg.m_time = time;
788 2 : msg.m_message_size = hdr.nMessageSize;
789 2 : msg.m_raw_message_size = hdr.nMessageSize + CMessageHeader::HEADER_SIZE;
790 :
791 2 : uint256 hash = GetMessageHash();
792 :
793 : // We just received a message off the wire, harvest entropy from the time (and the message checksum)
794 2 : RandAddEvent(ReadLE32(hash.begin()));
795 :
796 : // Check checksum and header message type string
797 2 : if (memcmp(hash.begin(), hdr.pchChecksum, CMessageHeader::CHECKSUM_SIZE) != 0) {
798 0 : LogPrint(BCLog::NET, "Header error: Wrong checksum (%s, %u bytes), expected %s was %s, peer=%d\n",
799 : SanitizeString(msg.m_type), msg.m_message_size,
800 : HexStr(Span{hash}.first(CMessageHeader::CHECKSUM_SIZE)),
801 : HexStr(hdr.pchChecksum),
802 : m_node_id);
803 0 : reject_message = true;
804 2 : } else if (!hdr.IsCommandValid()) {
805 0 : LogPrint(BCLog::NET, "Header error: Invalid message type (%s, %u bytes), peer=%d\n",
806 : SanitizeString(hdr.GetCommand()), msg.m_message_size, m_node_id);
807 0 : reject_message = true;
808 0 : }
809 :
810 : // Always reset the network deserializer (prepare for the next message)
811 2 : Reset();
812 2 : return msg;
813 2 : }
814 :
815 2 : bool V1Transport::SetMessageToSend(CSerializedNetMsg& msg) noexcept
816 : {
817 2 : AssertLockNotHeld(m_send_mutex);
818 : // Determine whether a new message can be set.
819 2 : LOCK(m_send_mutex);
820 2 : if (m_sending_header || m_bytes_sent < m_message_to_send.data.size()) return false;
821 :
822 : // create dbl-sha256 checksum
823 2 : uint256 hash = Hash(msg.data);
824 :
825 : // create header
826 2 : CMessageHeader hdr(m_magic_bytes, msg.m_type.c_str(), msg.data.size());
827 2 : memcpy(hdr.pchChecksum, hash.begin(), CMessageHeader::CHECKSUM_SIZE);
828 :
829 : // serialize header
830 2 : m_header_to_send.clear();
831 2 : CVectorWriter{SER_NETWORK, INIT_PROTO_VERSION, m_header_to_send, 0, hdr};
832 :
833 : // update state
834 2 : m_message_to_send = std::move(msg);
835 2 : m_sending_header = true;
836 2 : m_bytes_sent = 0;
837 2 : return true;
838 2 : }
839 :
840 54 : Transport::BytesToSend V1Transport::GetBytesToSend(bool have_next_message) const noexcept
841 : {
842 54 : AssertLockNotHeld(m_send_mutex);
843 54 : LOCK(m_send_mutex);
844 54 : if (m_sending_header) {
845 4 : return {Span{m_header_to_send}.subspan(m_bytes_sent),
846 : // We have more to send after the header if the message has payload, or if there
847 : // is a next message after that.
848 2 : have_next_message || !m_message_to_send.data.empty(),
849 2 : m_message_to_send.m_type
850 : };
851 : } else {
852 52 : return {Span{m_message_to_send.data}.subspan(m_bytes_sent),
853 : // We only have more to send after this message's payload if there is another
854 : // message.
855 : have_next_message,
856 52 : m_message_to_send.m_type
857 : };
858 : }
859 54 : }
860 :
861 3 : void V1Transport::MarkBytesSent(size_t bytes_sent) noexcept
862 : {
863 3 : AssertLockNotHeld(m_send_mutex);
864 3 : LOCK(m_send_mutex);
865 3 : m_bytes_sent += bytes_sent;
866 3 : if (m_sending_header && m_bytes_sent == m_header_to_send.size()) {
867 : // We're done sending a message's header. Switch to sending its data bytes.
868 2 : m_sending_header = false;
869 2 : m_bytes_sent = 0;
870 3 : } else if (!m_sending_header && m_bytes_sent == m_message_to_send.data.size()) {
871 : // We're done sending a message's data. Wipe the data vector to reduce memory consumption.
872 1 : ClearShrink(m_message_to_send.data);
873 1 : m_bytes_sent = 0;
874 1 : }
875 3 : }
876 :
877 40 : size_t V1Transport::GetSendMemoryUsage() const noexcept
878 : {
879 40 : AssertLockNotHeld(m_send_mutex);
880 40 : LOCK(m_send_mutex);
881 : // Don't count sending-side fields besides m_message_to_send, as they're all small and bounded.
882 40 : return m_message_to_send.GetMemoryUsage();
883 40 : }
884 :
885 : namespace {
886 :
887 : /** List of short messages as defined in BIP324, in order.
888 : *
889 : * Only message types that are actually implemented in this codebase need to be listed, as other
890 : * messages get ignored anyway - whether we know how to decode them or not.
891 : */
892 0 : const std::array<std::string, 33> V2_BITCOIN_IDS = {
893 146 : "", // 12 bytes follow encoding the message type like in V1
894 146 : NetMsgType::ADDR,
895 146 : NetMsgType::BLOCK,
896 146 : NetMsgType::BLOCKTXN,
897 146 : NetMsgType::CMPCTBLOCK,
898 146 : "", /* FEEFILTER is not implemented in Dash */
899 146 : NetMsgType::FILTERADD,
900 146 : NetMsgType::FILTERCLEAR,
901 146 : NetMsgType::FILTERLOAD,
902 146 : NetMsgType::GETBLOCKS,
903 146 : NetMsgType::GETBLOCKTXN,
904 146 : NetMsgType::GETDATA,
905 146 : NetMsgType::GETHEADERS,
906 146 : NetMsgType::HEADERS,
907 146 : NetMsgType::INV,
908 146 : NetMsgType::MEMPOOL,
909 146 : NetMsgType::MERKLEBLOCK,
910 146 : NetMsgType::NOTFOUND,
911 146 : NetMsgType::PING,
912 146 : NetMsgType::PONG,
913 146 : NetMsgType::SENDCMPCT,
914 146 : NetMsgType::TX,
915 146 : NetMsgType::GETCFILTERS,
916 146 : NetMsgType::CFILTER,
917 146 : NetMsgType::GETCFHEADERS,
918 146 : NetMsgType::CFHEADERS,
919 146 : NetMsgType::GETCFCHECKPT,
920 146 : NetMsgType::CFCHECKPT,
921 146 : NetMsgType::ADDRV2,
922 : // Unimplemented message types that are assigned in BIP324:
923 146 : "",
924 146 : "",
925 146 : "",
926 146 : ""
927 : };
928 :
929 : /** List of short messages allocated in Dash's reserved namespace, in order.
930 : *
931 : * Slots should not be reused unless the switchover has already been done
932 : * by a protocol upgrade, the old message is no longer supported by the client
933 : * and a new slot wasn't already allotted for the message.
934 : */
935 0 : const std::array<std::string, 41> V2_DASH_IDS = {
936 146 : NetMsgType::SPORK,
937 146 : NetMsgType::GETSPORKS,
938 146 : NetMsgType::SENDDSQUEUE,
939 146 : NetMsgType::DSACCEPT,
940 146 : NetMsgType::DSVIN,
941 146 : NetMsgType::DSFINALTX,
942 146 : NetMsgType::DSSIGNFINALTX,
943 146 : NetMsgType::DSCOMPLETE,
944 146 : NetMsgType::DSSTATUSUPDATE,
945 146 : NetMsgType::DSTX,
946 146 : NetMsgType::DSQUEUE,
947 146 : NetMsgType::SYNCSTATUSCOUNT,
948 146 : NetMsgType::MNGOVERNANCESYNC,
949 146 : NetMsgType::MNGOVERNANCEOBJECT,
950 146 : NetMsgType::MNGOVERNANCEOBJECTVOTE,
951 146 : NetMsgType::GETMNLISTDIFF,
952 146 : NetMsgType::MNLISTDIFF,
953 146 : NetMsgType::QSENDRECSIGS,
954 146 : NetMsgType::QFCOMMITMENT,
955 146 : NetMsgType::QCONTRIB,
956 146 : NetMsgType::QCOMPLAINT,
957 146 : NetMsgType::QJUSTIFICATION,
958 146 : NetMsgType::QPCOMMITMENT,
959 146 : NetMsgType::QWATCH,
960 146 : NetMsgType::QSIGSESANN,
961 146 : NetMsgType::QSIGSHARESINV,
962 146 : NetMsgType::QGETSIGSHARES,
963 146 : NetMsgType::QBSIGSHARES,
964 146 : NetMsgType::QSIGREC,
965 146 : NetMsgType::QSIGSHARE,
966 146 : NetMsgType::QGETDATA,
967 146 : NetMsgType::QDATA,
968 146 : NetMsgType::CLSIG,
969 146 : NetMsgType::ISDLOCK,
970 146 : NetMsgType::MNAUTH,
971 146 : NetMsgType::GETHEADERS2,
972 146 : NetMsgType::SENDHEADERS2,
973 146 : NetMsgType::HEADERS2,
974 146 : NetMsgType::GETQUORUMROTATIONINFO,
975 146 : NetMsgType::QUORUMROTATIONINFO,
976 146 : NetMsgType::PLATFORMBAN
977 : };
978 :
979 : /** Explicit version requirements for Dash v2 short IDs added after baseline.
980 : *
981 : * Maps message type to the minimum protocol version required to use its short ID.
982 : * Only contains messages that were added AFTER the initial v2 implementation.
983 : * Messages not in this map are baseline messages available at BIP324_DASH_BASELINE_VERSION.
984 : *
985 : * IMPORTANT:
986 : * - Never remove entries (historical compatibility)
987 : * - Add new messages as they are introduced
988 : */
989 146 : static const std::map<std::string, int> V2_NEW_SHORT_ID_VERSIONS = {
990 146 : {NetMsgType::PLATFORMBAN, PLATFORMBAN_V2_SHORT_ID_VERSION},
991 : };
992 :
993 : /** Get the minimum protocol version required for a message's short ID.
994 : *
995 : * @param message_type The message type to check
996 : * @return Minimum protocol version required
997 : */
998 4 : int GetMessageMinVersion(const std::string& message_type)
999 : {
1000 4 : auto it = V2_NEW_SHORT_ID_VERSIONS.find(message_type);
1001 4 : if (it != V2_NEW_SHORT_ID_VERSIONS.end()) {
1002 2 : return it->second;
1003 : }
1004 : // Not in map means it's a baseline message
1005 2 : return BIP324_DASH_BASELINE_VERSION;
1006 4 : }
1007 :
1008 : /** A complete set of short IDs
1009 : *
1010 : * Bitcoin takes up short IDs up to 128 (lower half) while Dash can take
1011 : * up short IDs between 128 and 256 (upper half) most of the array will
1012 : * have entries that correspond to nothing.
1013 : *
1014 : * To distinguish between entries that are *meant* to correspond to
1015 : * nothing versus empty space, use IsValidV2ShortID()
1016 : */
1017 48165 : constexpr std::array<std::string_view, 256> V2ShortIDs() {
1018 : static_assert(std::size(V2_BITCOIN_IDS) <= 128);
1019 : static_assert(std::size(V2_DASH_IDS) <= 128);
1020 :
1021 48165 : std::array<std::string_view, 256> ret{};
1022 48165 : std::fill(ret.begin(), ret.end(), "");
1023 48165 : std::copy(V2_BITCOIN_IDS.begin(), V2_BITCOIN_IDS.end(), ret.begin());
1024 48165 : std::copy(V2_DASH_IDS.begin(), V2_DASH_IDS.end(), ret.begin() + 128);
1025 48165 : return ret;
1026 : }
1027 :
1028 37362 : bool IsValidV2ShortID(uint8_t first_byte) {
1029 : // Since we have filled the namespace of short IDs, we have to preserve
1030 : // the expected behaviour of coming up short when going beyond Bitcoin's
1031 : // and Dash's *used* slots. We do this by checking if the byte is within
1032 : // the range where a valid message is expected to reside.
1033 69921 : return first_byte < std::size(V2_BITCOIN_IDS) ||
1034 32559 : (first_byte >= 128 && static_cast<uint8_t>(first_byte - 128) < std::size(V2_DASH_IDS));
1035 : }
1036 :
1037 : class V2MessageMap
1038 : {
1039 : std::unordered_map<std::string, uint8_t> m_map;
1040 :
1041 : public:
1042 292 : V2MessageMap() noexcept
1043 146 : {
1044 37376 : for (size_t i = 1; i < std::size(V2ShortIDs()); ++i) {
1045 37230 : if (IsValidV2ShortID(i)) {
1046 10658 : m_map.emplace(V2ShortIDs()[i], i);
1047 10658 : }
1048 37230 : }
1049 292 : }
1050 :
1051 54 : std::optional<uint8_t> operator()(const std::string& message_name) const noexcept
1052 : {
1053 54 : auto it = m_map.find(message_name);
1054 54 : if (it == m_map.end()) return std::nullopt;
1055 4 : return it->second;
1056 54 : }
1057 : };
1058 :
1059 146 : const V2MessageMap V2_MESSAGE_MAP;
1060 :
1061 78 : std::vector<uint8_t> GenerateRandomGarbage() noexcept
1062 : {
1063 78 : std::vector<uint8_t> ret;
1064 78 : FastRandomContext rng;
1065 78 : ret.resize(rng.randrange(V2Transport::MAX_GARBAGE_LEN + 1));
1066 78 : rng.fillrand(MakeWritableByteSpan(ret));
1067 78 : return ret;
1068 78 : }
1069 :
1070 : } // namespace
1071 :
1072 77 : void V2Transport::StartSendingHandshake() noexcept
1073 : {
1074 77 : AssertLockHeld(m_send_mutex);
1075 77 : Assume(m_send_state == SendState::AWAITING_KEY);
1076 77 : Assume(m_send_buffer.empty());
1077 : // Initialize the send buffer with ellswift pubkey + provided garbage.
1078 77 : m_send_buffer.resize(EllSwiftPubKey::size() + m_send_garbage.size());
1079 77 : std::copy(std::begin(m_cipher.GetOurPubKey()), std::end(m_cipher.GetOurPubKey()), MakeWritableByteSpan(m_send_buffer).begin());
1080 77 : std::copy(m_send_garbage.begin(), m_send_garbage.end(), m_send_buffer.begin() + EllSwiftPubKey::size());
1081 : // We cannot wipe m_send_garbage as it will still be used as AAD later in the handshake.
1082 77 : }
1083 :
1084 234 : V2Transport::V2Transport(NodeId nodeid, bool initiating, int type_in, int version_in, const CKey& key, Span<const std::byte> ent32, std::vector<uint8_t> garbage) noexcept :
1085 78 : m_cipher{key, ent32}, m_initiating{initiating}, m_nodeid{nodeid},
1086 78 : m_v1_fallback{nodeid, type_in, version_in}, m_recv_type{type_in}, m_recv_version{version_in},
1087 : m_recv_state{initiating ? RecvState::KEY : RecvState::KEY_MAYBE_V1},
1088 : m_send_garbage{std::move(garbage)},
1089 : m_send_state{initiating ? SendState::AWAITING_KEY : SendState::MAYBE_V1}
1090 156 : {
1091 : Assume(m_send_garbage.size() <= MAX_GARBAGE_LEN);
1092 : // Start sending immediately if we're the initiator of the connection.
1093 : if (initiating) {
1094 : LOCK(m_send_mutex);
1095 : StartSendingHandshake();
1096 : }
1097 78 : }
1098 :
1099 78 : V2Transport::V2Transport(NodeId nodeid, bool initiating, int type_in, int version_in) noexcept :
1100 156 : V2Transport{nodeid, initiating, type_in, version_in, GenerateRandomKey(),
1101 156 : MakeByteSpan(GetRandHash()), GenerateRandomGarbage()} { }
1102 :
1103 764 : void V2Transport::SetReceiveState(RecvState recv_state) noexcept
1104 : {
1105 764 : AssertLockHeld(m_recv_mutex);
1106 : // Enforce allowed state transitions.
1107 764 : switch (m_recv_state) {
1108 : case RecvState::KEY_MAYBE_V1:
1109 36 : Assume(recv_state == RecvState::KEY || recv_state == RecvState::V1);
1110 36 : break;
1111 : case RecvState::KEY:
1112 76 : Assume(recv_state == RecvState::GARB_GARBTERM);
1113 76 : break;
1114 : case RecvState::GARB_GARBTERM:
1115 74 : Assume(recv_state == RecvState::VERSION);
1116 74 : break;
1117 : case RecvState::VERSION:
1118 74 : Assume(recv_state == RecvState::APP);
1119 74 : break;
1120 : case RecvState::APP:
1121 252 : Assume(recv_state == RecvState::APP_READY);
1122 252 : break;
1123 : case RecvState::APP_READY:
1124 252 : Assume(recv_state == RecvState::APP);
1125 252 : break;
1126 : case RecvState::V1:
1127 0 : Assume(false); // V1 state cannot be left
1128 0 : break;
1129 : }
1130 : // Change state.
1131 764 : m_recv_state = recv_state;
1132 764 : }
1133 :
1134 112 : void V2Transport::SetSendState(SendState send_state) noexcept
1135 : {
1136 112 : AssertLockHeld(m_send_mutex);
1137 : // Enforce allowed state transitions.
1138 112 : switch (m_send_state) {
1139 : case SendState::MAYBE_V1:
1140 36 : Assume(send_state == SendState::V1 || send_state == SendState::AWAITING_KEY);
1141 36 : break;
1142 : case SendState::AWAITING_KEY:
1143 76 : Assume(send_state == SendState::READY);
1144 76 : break;
1145 : case SendState::READY:
1146 : case SendState::V1:
1147 0 : Assume(false); // Final states
1148 0 : break;
1149 : }
1150 : // Change state.
1151 112 : m_send_state = send_state;
1152 112 : }
1153 :
1154 2941 : bool V2Transport::ReceivedMessageComplete() const noexcept
1155 : {
1156 2941 : AssertLockNotHeld(m_recv_mutex);
1157 2941 : LOCK(m_recv_mutex);
1158 2941 : if (m_recv_state == RecvState::V1) return m_v1_fallback.ReceivedMessageComplete();
1159 :
1160 2935 : return m_recv_state == RecvState::APP_READY;
1161 2941 : }
1162 :
1163 36 : void V2Transport::ProcessReceivedMaybeV1Bytes() noexcept
1164 : {
1165 36 : AssertLockHeld(m_recv_mutex);
1166 36 : AssertLockNotHeld(m_send_mutex);
1167 36 : Assume(m_recv_state == RecvState::KEY_MAYBE_V1);
1168 : // We still have to determine if this is a v1 or v2 connection. The bytes being received could
1169 : // be the beginning of either a v1 packet (network magic + "version\x00\x00\x00\x00\x00"), or
1170 : // of a v2 public key. BIP324 specifies that a mismatch with this 16-byte string should trigger
1171 : // sending of the key.
1172 36 : std::array<uint8_t, V1_PREFIX_LEN> v1_prefix = {0, 0, 0, 0, 'v', 'e', 'r', 's', 'i', 'o', 'n', 0, 0, 0, 0, 0};
1173 36 : std::copy(std::begin(Params().MessageStart()), std::end(Params().MessageStart()), v1_prefix.begin());
1174 36 : Assume(m_recv_buffer.size() <= v1_prefix.size());
1175 36 : if (!std::equal(m_recv_buffer.begin(), m_recv_buffer.end(), v1_prefix.begin())) {
1176 : // Mismatch with v1 prefix, so we can assume a v2 connection.
1177 35 : SetReceiveState(RecvState::KEY); // Convert to KEY state, leaving received bytes around.
1178 : // Transition the sender to AWAITING_KEY state and start sending.
1179 35 : LOCK(m_send_mutex);
1180 35 : SetSendState(SendState::AWAITING_KEY);
1181 35 : StartSendingHandshake();
1182 36 : } else if (m_recv_buffer.size() == v1_prefix.size()) {
1183 : // Full match with the v1 prefix, so fall back to v1 behavior.
1184 1 : LOCK(m_send_mutex);
1185 1 : Span<const uint8_t> feedback{m_recv_buffer};
1186 : // Feed already received bytes to v1 transport. It should always accept these, because it's
1187 : // less than the size of a v1 header, and these are the first bytes fed to m_v1_fallback.
1188 1 : bool ret = m_v1_fallback.ReceivedBytes(feedback);
1189 1 : Assume(feedback.empty());
1190 1 : Assume(ret);
1191 1 : SetReceiveState(RecvState::V1);
1192 1 : SetSendState(SendState::V1);
1193 : // Reset v2 transport buffers to save memory.
1194 1 : ClearShrink(m_recv_buffer);
1195 1 : ClearShrink(m_send_buffer);
1196 1 : } else {
1197 : // We have not received enough to distinguish v1 from v2 yet. Wait until more bytes come.
1198 : }
1199 36 : }
1200 :
1201 160 : bool V2Transport::ProcessReceivedKeyBytes() noexcept
1202 : {
1203 160 : AssertLockHeld(m_recv_mutex);
1204 160 : AssertLockNotHeld(m_send_mutex);
1205 160 : Assume(m_recv_state == RecvState::KEY);
1206 160 : Assume(m_recv_buffer.size() <= EllSwiftPubKey::size());
1207 :
1208 : // As a special exception, if bytes 4-16 of the key on a responder connection match the
1209 : // corresponding bytes of a V1 version message, but bytes 0-4 don't match the network magic
1210 : // (if they did, we'd have switched to V1 state already), assume this is a peer from
1211 : // another network, and disconnect them. They will almost certainly disconnect us too when
1212 : // they receive our uniformly random key and garbage, but detecting this case specially
1213 : // means we can log it.
1214 : static constexpr std::array<uint8_t, 12> MATCH = {'v', 'e', 'r', 's', 'i', 'o', 'n', 0, 0, 0, 0, 0};
1215 : static constexpr size_t OFFSET = sizeof(CMessageHeader::MessageStartChars);
1216 160 : if (!m_initiating && m_recv_buffer.size() >= OFFSET + MATCH.size()) {
1217 89 : if (std::equal(MATCH.begin(), MATCH.end(), m_recv_buffer.begin() + OFFSET)) {
1218 1 : LogPrint(BCLog::NET, "V2 transport error: V1 peer with wrong MessageStart %s\n",
1219 : HexStr(Span(m_recv_buffer).first(OFFSET)));
1220 1 : return false;
1221 : }
1222 88 : }
1223 :
1224 159 : if (m_recv_buffer.size() == EllSwiftPubKey::size()) {
1225 : // Other side's key has been fully received, and can now be Diffie-Hellman combined with
1226 : // our key to initialize the encryption ciphers.
1227 :
1228 : // Initialize the ciphers.
1229 76 : EllSwiftPubKey ellswift(MakeByteSpan(m_recv_buffer));
1230 76 : LOCK(m_send_mutex);
1231 76 : m_cipher.Initialize(ellswift, m_initiating);
1232 :
1233 : // Switch receiver state to GARB_GARBTERM.
1234 76 : SetReceiveState(RecvState::GARB_GARBTERM);
1235 76 : m_recv_buffer.clear();
1236 :
1237 : // Switch sender state to READY.
1238 76 : SetSendState(SendState::READY);
1239 :
1240 : // Append the garbage terminator to the send buffer.
1241 76 : m_send_buffer.resize(m_send_buffer.size() + BIP324Cipher::GARBAGE_TERMINATOR_LEN);
1242 152 : std::copy(m_cipher.GetSendGarbageTerminator().begin(),
1243 76 : m_cipher.GetSendGarbageTerminator().end(),
1244 76 : MakeWritableByteSpan(m_send_buffer).last(BIP324Cipher::GARBAGE_TERMINATOR_LEN).begin());
1245 :
1246 : // Construct version packet in the send buffer, with the sent garbage data as AAD.
1247 76 : m_send_buffer.resize(m_send_buffer.size() + BIP324Cipher::EXPANSION + VERSION_CONTENTS.size());
1248 76 : m_cipher.Encrypt(
1249 76 : /*contents=*/VERSION_CONTENTS,
1250 76 : /*aad=*/MakeByteSpan(m_send_garbage),
1251 : /*ignore=*/false,
1252 76 : /*output=*/MakeWritableByteSpan(m_send_buffer).last(BIP324Cipher::EXPANSION + VERSION_CONTENTS.size()));
1253 : // We no longer need the garbage.
1254 76 : ClearShrink(m_send_garbage);
1255 76 : } else {
1256 : // We still have to receive more key bytes.
1257 : }
1258 159 : return true;
1259 160 : }
1260 :
1261 146848 : bool V2Transport::ProcessReceivedGarbageBytes() noexcept
1262 : {
1263 146848 : AssertLockHeld(m_recv_mutex);
1264 146848 : Assume(m_recv_state == RecvState::GARB_GARBTERM);
1265 146848 : Assume(m_recv_buffer.size() <= MAX_GARBAGE_LEN + BIP324Cipher::GARBAGE_TERMINATOR_LEN);
1266 146848 : if (m_recv_buffer.size() >= BIP324Cipher::GARBAGE_TERMINATOR_LEN) {
1267 145708 : if (MakeByteSpan(m_recv_buffer).last(BIP324Cipher::GARBAGE_TERMINATOR_LEN) == m_cipher.GetReceiveGarbageTerminator()) {
1268 : // Garbage terminator received. Store garbage to authenticate it as AAD later.
1269 74 : m_recv_aad = std::move(m_recv_buffer);
1270 74 : m_recv_aad.resize(m_recv_aad.size() - BIP324Cipher::GARBAGE_TERMINATOR_LEN);
1271 74 : m_recv_buffer.clear();
1272 74 : SetReceiveState(RecvState::VERSION);
1273 145708 : } else if (m_recv_buffer.size() == MAX_GARBAGE_LEN + BIP324Cipher::GARBAGE_TERMINATOR_LEN) {
1274 : // We've reached the maximum length for garbage + garbage terminator, and the
1275 : // terminator still does not match. Abort.
1276 2 : LogPrint(BCLog::NET, "V2 transport error: missing garbage terminator, peer=%d\n", m_nodeid);
1277 2 : return false;
1278 : } else {
1279 : // We still need to receive more garbage and/or garbage terminator bytes.
1280 : }
1281 145706 : } else {
1282 : // We have less than GARBAGE_TERMINATOR_LEN (16) bytes, so we certainly need to receive
1283 : // more first.
1284 : }
1285 146846 : return true;
1286 146848 : }
1287 :
1288 99563 : bool V2Transport::ProcessReceivedPacketBytes() noexcept
1289 : {
1290 99563 : AssertLockHeld(m_recv_mutex);
1291 99563 : Assume(m_recv_state == RecvState::VERSION || m_recv_state == RecvState::APP);
1292 :
1293 : // The maximum permitted contents length for a packet, consisting of:
1294 : // - 0x00 byte: indicating long message type encoding
1295 : // - 12 bytes of message type
1296 : // - payload
1297 : static constexpr size_t MAX_CONTENTS_LEN =
1298 : 1 + CMessageHeader::COMMAND_SIZE +
1299 : std::min<size_t>(MAX_SIZE, MAX_PROTOCOL_MESSAGE_LENGTH);
1300 :
1301 99563 : if (m_recv_buffer.size() == BIP324Cipher::LENGTH_LEN) {
1302 : // Length descriptor received.
1303 49134 : m_recv_len = m_cipher.DecryptLength(MakeByteSpan(m_recv_buffer));
1304 49134 : if (m_recv_len > MAX_CONTENTS_LEN) {
1305 10 : LogPrint(BCLog::NET, "V2 transport error: packet too large (%u bytes), peer=%d\n", m_recv_len, m_nodeid);
1306 10 : return false;
1307 : }
1308 99553 : } else if (m_recv_buffer.size() > BIP324Cipher::LENGTH_LEN && m_recv_buffer.size() == m_recv_len + BIP324Cipher::EXPANSION) {
1309 : // Ciphertext received, decrypt it into m_recv_decode_buffer.
1310 : // Note that it is impossible to reach this branch without hitting the branch above first,
1311 : // as GetMaxBytesToProcess only allows up to LENGTH_LEN into the buffer before that point.
1312 49124 : m_recv_decode_buffer.resize(m_recv_len);
1313 49124 : bool ignore{false};
1314 98248 : bool ret = m_cipher.Decrypt(
1315 49124 : /*input=*/MakeByteSpan(m_recv_buffer).subspan(BIP324Cipher::LENGTH_LEN),
1316 49124 : /*aad=*/MakeByteSpan(m_recv_aad),
1317 : /*ignore=*/ignore,
1318 49124 : /*contents=*/MakeWritableByteSpan(m_recv_decode_buffer));
1319 49124 : if (!ret) {
1320 10 : LogPrint(BCLog::NET, "V2 transport error: packet decryption failure (%u bytes), peer=%d\n", m_recv_len, m_nodeid);
1321 10 : return false;
1322 : }
1323 : // We have decrypted a valid packet with the AAD we expected, so clear the expected AAD.
1324 49114 : ClearShrink(m_recv_aad);
1325 : // Feed the last 4 bytes of the Poly1305 authentication tag (and its timing) into our RNG.
1326 49114 : RandAddEvent(ReadLE32(m_recv_buffer.data() + m_recv_buffer.size() - 4));
1327 :
1328 : // At this point we have a valid packet decrypted into m_recv_decode_buffer. If it's not a
1329 : // decoy, which we simply ignore, use the current state to decide what to do with it.
1330 49114 : if (!ignore) {
1331 326 : switch (m_recv_state) {
1332 : case RecvState::VERSION:
1333 : // Version message received; transition to application phase. The contents is
1334 : // ignored, but can be used for future extensions.
1335 74 : SetReceiveState(RecvState::APP);
1336 74 : break;
1337 : case RecvState::APP:
1338 : // Application message decrypted correctly. It can be extracted using GetMessage().
1339 252 : SetReceiveState(RecvState::APP_READY);
1340 252 : break;
1341 : default:
1342 : // Any other state is invalid (this function should not have been called).
1343 0 : Assume(false);
1344 0 : }
1345 326 : }
1346 : // Wipe the receive buffer where the next packet will be received into.
1347 49114 : ClearShrink(m_recv_buffer);
1348 : // In all but APP_READY state, we can wipe the decoded contents.
1349 49114 : if (m_recv_state != RecvState::APP_READY) ClearShrink(m_recv_decode_buffer);
1350 49114 : } else {
1351 : // We either have less than 3 bytes, so we don't know the packet's length yet, or more
1352 : // than 3 bytes but less than the packet's full ciphertext. Wait until those arrive.
1353 : }
1354 99543 : return true;
1355 99563 : }
1356 :
1357 246878 : size_t V2Transport::GetMaxBytesToProcess() noexcept
1358 : {
1359 246878 : AssertLockHeld(m_recv_mutex);
1360 246878 : switch (m_recv_state) {
1361 : case RecvState::KEY_MAYBE_V1:
1362 : // During the KEY_MAYBE_V1 state we do not allow more than the length of v1 prefix into the
1363 : // receive buffer.
1364 36 : Assume(m_recv_buffer.size() <= V1_PREFIX_LEN);
1365 : // As long as we're not sure if this is a v1 or v2 connection, don't receive more than what
1366 : // is strictly necessary to distinguish the two (16 bytes). If we permitted more than
1367 : // the v1 header size (24 bytes), we may not be able to feed the already-received bytes
1368 : // back into the m_v1_fallback V1 transport.
1369 36 : return V1_PREFIX_LEN - m_recv_buffer.size();
1370 : case RecvState::KEY:
1371 : // During the KEY state, we only allow the 64-byte key into the receive buffer.
1372 160 : Assume(m_recv_buffer.size() <= EllSwiftPubKey::size());
1373 : // As long as we have not received the other side's public key, don't receive more than
1374 : // that (64 bytes), as garbage follows, and locating the garbage terminator requires the
1375 : // key exchange first.
1376 160 : return EllSwiftPubKey::size() - m_recv_buffer.size();
1377 : case RecvState::GARB_GARBTERM:
1378 : // Process garbage bytes one by one (because terminator may appear anywhere).
1379 146848 : return 1;
1380 : case RecvState::VERSION:
1381 : case RecvState::APP:
1382 : // These three states all involve decoding a packet. Process the length descriptor first,
1383 : // so that we know where the current packet ends (and we don't process bytes from the next
1384 : // packet or decoy yet). Then, process the ciphertext bytes of the current packet.
1385 99563 : if (m_recv_buffer.size() < BIP324Cipher::LENGTH_LEN) {
1386 49154 : return BIP324Cipher::LENGTH_LEN - m_recv_buffer.size();
1387 : } else {
1388 : // Note that BIP324Cipher::EXPANSION is the total difference between contents size
1389 : // and encoded packet size, which includes the 3 bytes due to the packet length.
1390 : // When transitioning from receiving the packet length to receiving its ciphertext,
1391 : // the encrypted packet length is left in the receive buffer.
1392 50409 : return BIP324Cipher::EXPANSION + m_recv_len - m_recv_buffer.size();
1393 : }
1394 : case RecvState::APP_READY:
1395 : // No bytes can be processed until GetMessage() is called.
1396 271 : return 0;
1397 : case RecvState::V1:
1398 : // Not allowed (must be dealt with by the caller).
1399 0 : Assume(false);
1400 0 : return 0;
1401 : }
1402 0 : Assume(false); // unreachable
1403 0 : return 0;
1404 246878 : }
1405 :
1406 198758 : bool V2Transport::ReceivedBytes(Span<const uint8_t>& msg_bytes) noexcept
1407 : {
1408 395348 : AssertLockNotHeld(m_recv_mutex);
1409 : /** How many bytes to allocate in the receive buffer at most above what is received so far. */
1410 : static constexpr size_t MAX_RESERVE_AHEAD = 256 * 1024;
1411 :
1412 2168 : LOCK(m_recv_mutex);
1413 2168 : if (m_recv_state == RecvState::V1) return m_v1_fallback.ReceivedBytes(msg_bytes);
1414 :
1415 : // Process the provided bytes in msg_bytes in a loop. In each iteration a nonzero number of
1416 : // bytes (decided by GetMaxBytesToProcess) are taken from the beginning om msg_bytes, and
1417 : // appended to m_recv_buffer. Then, depending on the receiver state, one of the
1418 : // ProcessReceived*Bytes functions is called to process the bytes in that buffer.
1419 248747 : while (!msg_bytes.empty()) {
1420 : // Decide how many bytes to copy from msg_bytes to m_recv_buffer.
1421 246878 : size_t max_read = GetMaxBytesToProcess();
1422 :
1423 : // Reserve space in the buffer if there is not enough.
1424 246878 : if (m_recv_buffer.size() + std::min(msg_bytes.size(), max_read) > m_recv_buffer.capacity()) {
1425 98373 : switch (m_recv_state) {
1426 : case RecvState::KEY_MAYBE_V1:
1427 : case RecvState::KEY:
1428 : case RecvState::GARB_GARBTERM:
1429 : // During the initial states (key/garbage), allocate once to fit the maximum (4111
1430 : // bytes).
1431 98373 : m_recv_buffer.reserve(MAX_GARBAGE_LEN + BIP324Cipher::GARBAGE_TERMINATOR_LEN);
1432 78 : break;
1433 : case RecvState::VERSION:
1434 : case RecvState::APP: {
1435 : // During states where a packet is being received, as much as is expected but never
1436 : // more than MAX_RESERVE_AHEAD bytes in addition to what is received so far.
1437 : // This means attackers that want to cause us to waste allocated memory are limited
1438 : // to MAX_RESERVE_AHEAD above the largest allowed message contents size, and to
1439 : // MAX_RESERVE_AHEAD more than they've actually sent us.
1440 0 : size_t alloc_add = std::min(max_read, msg_bytes.size() + MAX_RESERVE_AHEAD);
1441 98295 : m_recv_buffer.reserve(m_recv_buffer.size() + alloc_add);
1442 98295 : break;
1443 : }
1444 : case RecvState::APP_READY:
1445 : // The buffer is empty in this state.
1446 0 : Assume(m_recv_buffer.empty());
1447 0 : break;
1448 : case RecvState::V1:
1449 : // Should have bailed out above.
1450 0 : Assume(false);
1451 0 : break;
1452 : }
1453 98373 : }
1454 :
1455 : // Can't read more than provided input.
1456 246878 : max_read = std::min(msg_bytes.size(), max_read);
1457 : // Copy data to buffer.
1458 246878 : m_recv_buffer.insert(m_recv_buffer.end(), UCharCast(msg_bytes.data()), UCharCast(msg_bytes.data() + max_read));
1459 246878 : msg_bytes = msg_bytes.subspan(max_read);
1460 :
1461 : // Process data in the buffer.
1462 246878 : switch (m_recv_state) {
1463 : case RecvState::KEY_MAYBE_V1:
1464 36 : ProcessReceivedMaybeV1Bytes();
1465 36 : if (m_recv_state == RecvState::V1) return true;
1466 35 : break;
1467 :
1468 : case RecvState::KEY:
1469 160 : if (!ProcessReceivedKeyBytes()) return false;
1470 159 : break;
1471 :
1472 : case RecvState::GARB_GARBTERM:
1473 146848 : if (!ProcessReceivedGarbageBytes()) return false;
1474 146846 : break;
1475 :
1476 : case RecvState::VERSION:
1477 : case RecvState::APP:
1478 99563 : if (!ProcessReceivedPacketBytes()) return false;
1479 99543 : break;
1480 :
1481 : case RecvState::APP_READY:
1482 271 : return true;
1483 :
1484 : case RecvState::V1:
1485 : // We should have bailed out before.
1486 0 : Assume(false);
1487 0 : break;
1488 : }
1489 : // Make sure we have made progress before continuing.
1490 246583 : Assume(max_read > 0);
1491 : }
1492 :
1493 1869 : return true;
1494 2168 : }
1495 :
1496 252 : std::optional<std::string> V2Transport::GetMessageType(Span<const uint8_t>& contents) noexcept
1497 : {
1498 252 : if (contents.size() == 0) return std::nullopt; // Empty contents
1499 252 : uint8_t first_byte = contents[0];
1500 252 : contents = contents.subspan(1); // Strip first byte.
1501 :
1502 252 : if (first_byte != 0) {
1503 : // Short (1 byte) encoding.
1504 132 : if (IsValidV2ShortID(first_byte)) {
1505 : // Valid short message id.
1506 131 : return std::string{V2ShortIDs()[first_byte]};
1507 : } else {
1508 : // Unknown short message id.
1509 1 : return std::nullopt;
1510 : }
1511 : }
1512 :
1513 120 : if (contents.size() < CMessageHeader::COMMAND_SIZE) {
1514 10 : return std::nullopt; // Long encoding needs 12 message type bytes.
1515 : }
1516 :
1517 110 : size_t msg_type_len{0};
1518 830 : while (msg_type_len < CMessageHeader::COMMAND_SIZE && contents[msg_type_len] != 0) {
1519 : // Verify that message type bytes before the first 0x00 are in range.
1520 720 : if (contents[msg_type_len] < ' ' || contents[msg_type_len] > 0x7F) {
1521 0 : return {};
1522 : }
1523 720 : ++msg_type_len;
1524 : }
1525 110 : std::string ret{reinterpret_cast<const char*>(contents.data()), msg_type_len};
1526 610 : while (msg_type_len < CMessageHeader::COMMAND_SIZE) {
1527 : // Verify that message type bytes after the first 0x00 are also 0x00.
1528 550 : if (contents[msg_type_len] != 0) return {};
1529 500 : ++msg_type_len;
1530 : }
1531 : // Strip message type bytes of contents.
1532 60 : contents = contents.subspan(CMessageHeader::COMMAND_SIZE);
1533 60 : return ret;
1534 252 : }
1535 :
1536 252 : CNetMessage V2Transport::GetReceivedMessage(std::chrono::microseconds time, bool& reject_message) noexcept
1537 : {
1538 252 : AssertLockNotHeld(m_recv_mutex);
1539 252 : LOCK(m_recv_mutex);
1540 252 : if (m_recv_state == RecvState::V1) return m_v1_fallback.GetReceivedMessage(time, reject_message);
1541 :
1542 252 : Assume(m_recv_state == RecvState::APP_READY);
1543 252 : Span<const uint8_t> contents{m_recv_decode_buffer};
1544 252 : auto msg_type = GetMessageType(contents);
1545 252 : CDataStream ret(m_recv_type, m_recv_version);
1546 252 : CNetMessage msg{std::move(ret)};
1547 : // Note that BIP324Cipher::EXPANSION also includes the length descriptor size.
1548 252 : msg.m_raw_message_size = m_recv_decode_buffer.size() + BIP324Cipher::EXPANSION;
1549 252 : if (msg_type) {
1550 191 : reject_message = false;
1551 191 : msg.m_type = std::move(*msg_type);
1552 191 : msg.m_time = time;
1553 191 : msg.m_message_size = contents.size();
1554 191 : msg.m_recv.resize(contents.size());
1555 191 : std::copy(contents.begin(), contents.end(), UCharCast(msg.m_recv.data()));
1556 191 : } else {
1557 61 : LogPrint(BCLog::NET, "V2 transport error: invalid message type (%u bytes contents), peer=%d\n", m_recv_decode_buffer.size(), m_nodeid);
1558 61 : reject_message = true;
1559 : }
1560 252 : ClearShrink(m_recv_decode_buffer);
1561 252 : SetReceiveState(RecvState::APP);
1562 :
1563 252 : return msg;
1564 252 : }
1565 :
1566 54 : bool V2Transport::SetMessageToSend(CSerializedNetMsg& msg) noexcept
1567 : {
1568 54 : AssertLockNotHeld(m_send_mutex);
1569 54 : LOCK(m_send_mutex);
1570 54 : if (m_send_state == SendState::V1) return m_v1_fallback.SetMessageToSend(msg);
1571 : // We only allow adding a new message to be sent when in the READY state (so the packet cipher
1572 : // is available) and the send buffer is empty. This limits the number of messages in the send
1573 : // buffer to just one, and leaves the responsibility for queueing them up to the caller.
1574 54 : if (!(m_send_state == SendState::READY && m_send_buffer.empty())) return false;
1575 : // Construct contents (encoding message type + payload).
1576 54 : std::vector<uint8_t> contents;
1577 54 : auto short_message_id = V2_MESSAGE_MAP(msg.m_type);
1578 :
1579 54 : if (short_message_id.has_value() && m_peer_version >= GetMessageMinVersion(msg.m_type)) {
1580 : // Use short encoding (1 byte)
1581 3 : contents.resize(1 + msg.data.size());
1582 3 : contents[0] = *short_message_id;
1583 3 : std::copy(msg.data.begin(), msg.data.end(), contents.begin() + 1);
1584 3 : } else {
1585 : // Use long encoding (13 bytes for message type)
1586 : // Initialize with zeroes, and then write the message type string starting at offset 1.
1587 : // This means contents[0] and the unused positions in contents[1..13] remain 0x00.
1588 51 : contents.resize(1 + CMessageHeader::COMMAND_SIZE + msg.data.size(), 0);
1589 51 : std::copy(msg.m_type.begin(), msg.m_type.end(), contents.data() + 1);
1590 51 : std::copy(msg.data.begin(), msg.data.end(), contents.begin() + 1 + CMessageHeader::COMMAND_SIZE);
1591 : }
1592 : // Construct ciphertext in send buffer.
1593 54 : m_send_buffer.resize(contents.size() + BIP324Cipher::EXPANSION);
1594 54 : m_cipher.Encrypt(MakeByteSpan(contents), {}, false, MakeWritableByteSpan(m_send_buffer));
1595 54 : m_send_type = msg.m_type;
1596 : // Release memory
1597 54 : ClearShrink(msg.data);
1598 54 : return true;
1599 54 : }
1600 :
1601 80 : void V2Transport::SetPeerVersion(int version) noexcept
1602 : {
1603 80 : AssertLockNotHeld(m_send_mutex);
1604 80 : LOCK(m_send_mutex);
1605 80 : m_peer_version = version;
1606 80 : }
1607 :
1608 2941 : Transport::BytesToSend V2Transport::GetBytesToSend(bool have_next_message) const noexcept
1609 : {
1610 2941 : AssertLockNotHeld(m_send_mutex);
1611 2941 : LOCK(m_send_mutex);
1612 2941 : if (m_send_state == SendState::V1) return m_v1_fallback.GetBytesToSend(have_next_message);
1613 :
1614 2935 : if (m_send_state == SendState::MAYBE_V1) Assume(m_send_buffer.empty());
1615 2935 : Assume(m_send_pos <= m_send_buffer.size());
1616 2935 : return {
1617 2935 : Span{m_send_buffer}.subspan(m_send_pos),
1618 : // We only have more to send after the current m_send_buffer if there is a (next)
1619 : // message to be sent, and we're capable of sending packets. */
1620 2935 : have_next_message && m_send_state == SendState::READY,
1621 2935 : m_send_type
1622 : };
1623 2941 : }
1624 :
1625 1018 : void V2Transport::MarkBytesSent(size_t bytes_sent) noexcept
1626 : {
1627 1018 : AssertLockNotHeld(m_send_mutex);
1628 1018 : LOCK(m_send_mutex);
1629 1018 : if (m_send_state == SendState::V1) return m_v1_fallback.MarkBytesSent(bytes_sent);
1630 :
1631 1018 : if (m_send_state == SendState::AWAITING_KEY && m_send_pos == 0 && bytes_sent > 0) {
1632 50 : LogPrint(BCLog::NET, "start sending v2 handshake to peer=%d\n", m_nodeid);
1633 50 : }
1634 :
1635 1018 : m_send_pos += bytes_sent;
1636 1018 : Assume(m_send_pos <= m_send_buffer.size());
1637 1018 : if (m_send_pos >= CMessageHeader::HEADER_SIZE) {
1638 913 : m_sent_v1_header_worth = true;
1639 913 : }
1640 : // Wipe the buffer when everything is sent.
1641 1018 : if (m_send_pos == m_send_buffer.size()) {
1642 161 : m_send_pos = 0;
1643 161 : ClearShrink(m_send_buffer);
1644 161 : }
1645 1018 : }
1646 :
1647 0 : bool V2Transport::ShouldReconnectV1() const noexcept
1648 : {
1649 0 : AssertLockNotHeld(m_send_mutex);
1650 0 : AssertLockNotHeld(m_recv_mutex);
1651 : // Only outgoing connections need reconnection.
1652 0 : if (!m_initiating) return false;
1653 :
1654 0 : LOCK(m_recv_mutex);
1655 : // We only reconnect in the very first state and when the receive buffer is empty. Together
1656 : // these conditions imply nothing has been received so far.
1657 0 : if (m_recv_state != RecvState::KEY) return false;
1658 0 : if (!m_recv_buffer.empty()) return false;
1659 : // Check if we've sent enough for the other side to disconnect us (if it was V1).
1660 0 : LOCK(m_send_mutex);
1661 0 : return m_sent_v1_header_worth;
1662 0 : }
1663 :
1664 0 : size_t V2Transport::GetSendMemoryUsage() const noexcept
1665 : {
1666 0 : AssertLockNotHeld(m_send_mutex);
1667 0 : LOCK(m_send_mutex);
1668 0 : if (m_send_state == SendState::V1) return m_v1_fallback.GetSendMemoryUsage();
1669 :
1670 0 : return sizeof(m_send_buffer) + memusage::DynamicUsage(m_send_buffer);
1671 0 : }
1672 :
1673 71 : Transport::Info V2Transport::GetInfo() const noexcept
1674 : {
1675 71 : AssertLockNotHeld(m_recv_mutex);
1676 71 : LOCK(m_recv_mutex);
1677 71 : if (m_recv_state == RecvState::V1) return m_v1_fallback.GetInfo();
1678 :
1679 71 : Transport::Info info;
1680 :
1681 : // Do not report v2 and session ID until the version packet has been received
1682 : // and verified (confirming that the other side very likely has the same keys as us).
1683 142 : if (m_recv_state != RecvState::KEY_MAYBE_V1 && m_recv_state != RecvState::KEY &&
1684 71 : m_recv_state != RecvState::GARB_GARBTERM && m_recv_state != RecvState::VERSION) {
1685 71 : info.transport_type = TransportProtocolType::V2;
1686 71 : info.session_id = uint256(MakeUCharSpan(m_cipher.GetSessionID()));
1687 71 : } else {
1688 0 : info.transport_type = TransportProtocolType::DETECTING;
1689 : }
1690 :
1691 71 : return info;
1692 71 : }
1693 :
1694 0 : std::pair<size_t, bool> CConnman::SocketSendData(CNode& node) const
1695 : {
1696 0 : auto it = node.vSendMsg.begin();
1697 0 : size_t nSentSize = 0;
1698 0 : bool data_left{false}; //!< second return value (whether unsent data remains)
1699 0 : std::optional<bool> expected_more;
1700 :
1701 0 : while (true) {
1702 0 : if (it != node.vSendMsg.end()) {
1703 : // If possible, move one message from the send queue to the transport. This fails when
1704 : // there is an existing message still being sent, or (for v2 transports) when the
1705 : // handshake has not yet completed.
1706 0 : size_t memusage = it->GetMemoryUsage();
1707 0 : if (node.m_transport->SetMessageToSend(*it)) {
1708 : // Update memory usage of send buffer (as *it will be deleted).
1709 0 : node.m_send_memusage -= memusage;
1710 0 : ++it;
1711 0 : }
1712 0 : }
1713 0 : const auto& [data, more, msg_type] = node.m_transport->GetBytesToSend(it != node.vSendMsg.end());
1714 : // We rely on the 'more' value returned by GetBytesToSend to correctly predict whether more
1715 : // bytes are still to be sent, to correctly set the MSG_MORE flag. As a sanity check,
1716 : // verify that the previously returned 'more' was correct.
1717 0 : if (expected_more.has_value()) Assume(!data.empty() == *expected_more);
1718 0 : expected_more = more;
1719 0 : data_left = !data.empty(); // will be overwritten on next loop if all of data gets sent
1720 0 : int nBytes = 0;
1721 0 : if (!data.empty()) {
1722 0 : LOCK(node.m_sock_mutex);
1723 : // There is no socket in case we've already disconnected, or in test cases without
1724 : // real connections. In these cases, we bail out immediately and just leave things
1725 : // in the send queue and transport.
1726 0 : if (!node.m_sock) {
1727 0 : break;
1728 : }
1729 0 : int flags = MSG_NOSIGNAL | MSG_DONTWAIT;
1730 : #ifdef MSG_MORE
1731 : if (more) {
1732 : flags |= MSG_MORE;
1733 : }
1734 : #endif
1735 0 : nBytes = node.m_sock->Send(reinterpret_cast<const char*>(data.data()), data.size(), flags);
1736 0 : }
1737 0 : if (nBytes > 0) {
1738 0 : node.m_last_send = GetTime<std::chrono::seconds>();
1739 0 : node.nSendBytes += nBytes;
1740 : // Notify transport that bytes have been processed.
1741 0 : node.m_transport->MarkBytesSent(nBytes);
1742 : // Update statistics per message type.
1743 0 : if (!msg_type.empty()) { // don't report v2 handshake bytes for now
1744 0 : node.AccountForSentBytes(msg_type, nBytes);
1745 0 : }
1746 0 : nSentSize += nBytes;
1747 0 : if ((size_t)nBytes != data.size()) {
1748 : // could not send full message; stop sending more
1749 0 : node.fCanSendData = false;
1750 0 : break;
1751 : }
1752 0 : } else {
1753 0 : if (nBytes < 0) {
1754 : // error
1755 0 : int nErr = WSAGetLastError();
1756 0 : if (nErr != WSAEWOULDBLOCK && nErr != WSAEMSGSIZE && nErr != WSAEINTR && nErr != WSAEINPROGRESS) {
1757 0 : LogPrint(BCLog::NET, "socket send error for peer=%d: %s\n", node.GetId(), NetworkErrorString(nErr));
1758 0 : node.fDisconnect = true;
1759 0 : }
1760 0 : }
1761 0 : break;
1762 : }
1763 : }
1764 :
1765 0 : node.fPauseSend = node.m_send_memusage + node.m_transport->GetSendMemoryUsage() > nSendBufferMaxSize;
1766 :
1767 0 : if (it == node.vSendMsg.end()) {
1768 0 : assert(node.m_send_memusage == 0);
1769 0 : }
1770 0 : node.vSendMsg.erase(node.vSendMsg.begin(), it);
1771 0 : node.nSendMsgSize = node.vSendMsg.size();
1772 0 : return {nSentSize, data_left};
1773 0 : }
1774 :
1775 0 : std::vector<NodeEvictionCandidate> CConnman::GetEvictionCandidates() const
1776 : {
1777 0 : std::vector<NodeEvictionCandidate> vEvictionCandidates;
1778 0 : READ_LOCK(m_nodes_mutex);
1779 :
1780 0 : for (const CNode* node : m_nodes) {
1781 0 : if (node->fDisconnect)
1782 0 : continue;
1783 :
1784 0 : if (m_active_masternode) {
1785 : // This handles eviction protected nodes. Nodes are always protected for a short time after the connection
1786 : // was accepted. This short time is meant for the VERSION/VERACK exchange and the possible MNAUTH that might
1787 : // follow when the incoming connection is from another masternode. When a message other than MNAUTH
1788 : // is received after VERSION/VERACK, the protection is lifted immediately.
1789 0 : bool isProtected = GetTime<std::chrono::seconds>() - node->m_connected < INBOUND_EVICTION_PROTECTION_TIME;
1790 0 : if (node->nTimeFirstMessageReceived.load() != 0s && !node->fFirstMessageIsMNAUTH) {
1791 0 : isProtected = false;
1792 0 : }
1793 : // if MNAUTH was valid, the node is always protected (and at the same time not accounted when
1794 : // checking incoming connection limits)
1795 0 : if (!node->GetVerifiedProRegTxHash().IsNull()) {
1796 0 : isProtected = true;
1797 0 : }
1798 0 : if (isProtected) {
1799 0 : continue;
1800 : }
1801 0 : }
1802 :
1803 0 : NodeEvictionCandidate candidate{
1804 0 : .id = node->GetId(),
1805 0 : .m_connected = node->m_connected,
1806 0 : .m_min_ping_time = node->m_min_ping_time,
1807 0 : .m_last_block_time = node->m_last_block_time,
1808 0 : .m_last_tx_time = node->m_last_tx_time,
1809 0 : .fRelevantServices = node->m_has_all_wanted_services,
1810 0 : .m_relay_txs = node->m_relays_txs.load(),
1811 0 : .fBloomFilter = node->m_bloom_filter_loaded.load(),
1812 0 : .nKeyedNetGroup = node->nKeyedNetGroup,
1813 0 : .prefer_evict = node->m_prefer_evict,
1814 0 : .m_is_local = node->addr.IsLocal(),
1815 0 : .m_network = node->ConnectedThroughNetwork(),
1816 0 : .m_noban = node->HasPermission(NetPermissionFlags::NoBan),
1817 0 : .m_conn_type = node->m_conn_type,
1818 : };
1819 0 : vEvictionCandidates.push_back(candidate);
1820 : }
1821 :
1822 0 : return vEvictionCandidates;
1823 0 : }
1824 :
1825 : /** Try to find a connection to evict when the node is full.
1826 : * Extreme care must be taken to avoid opening the node to attacker
1827 : * triggered network partitioning.
1828 : * The strategy used here is to protect a small number of peers
1829 : * for each of several distinct characteristics which are difficult
1830 : * to forge. In order to partition a node the attacker must be
1831 : * simultaneously better at all of them than honest peers.
1832 : */
1833 0 : bool CConnman::AttemptToEvictConnection()
1834 : {
1835 0 : std::vector<NodeEvictionCandidate> vEvictionCandidates = GetEvictionCandidates();
1836 0 : const std::optional<NodeId> node_id_to_evict = SelectNodeToEvict(std::move(vEvictionCandidates));
1837 0 : if (!node_id_to_evict) {
1838 0 : return false;
1839 : }
1840 0 : return WithNodeMutable(*node_id_to_evict, [](CNode* pnode){
1841 0 : LogPrint(BCLog::NET_NETCONN, "selected %s connection for eviction peer=%d; disconnecting\n", pnode->ConnectionTypeAsString(), pnode->GetId());
1842 0 : pnode->fDisconnect = true;
1843 0 : return true;
1844 0 : }).value_or(false);
1845 0 : }
1846 :
1847 0 : void CConnman::AcceptConnection(const ListenSocket& hListenSocket, CMasternodeSync& mn_sync) {
1848 : struct sockaddr_storage sockaddr;
1849 0 : socklen_t len = sizeof(sockaddr);
1850 0 : auto sock = hListenSocket.sock->Accept((struct sockaddr*)&sockaddr, &len);
1851 0 : CAddress addr;
1852 :
1853 0 : if (!sock) {
1854 0 : const int nErr = WSAGetLastError();
1855 0 : if (nErr != WSAEWOULDBLOCK) {
1856 0 : LogPrintf("socket error accept failed: %s\n", NetworkErrorString(nErr));
1857 0 : }
1858 0 : return;
1859 : }
1860 :
1861 0 : if (!addr.SetSockAddr((const struct sockaddr*)&sockaddr)) {
1862 0 : LogPrintLevel(BCLog::NET, BCLog::Level::Warning, "Unknown socket family\n");
1863 0 : } else {
1864 0 : addr = CAddress{MaybeFlipIPv6toCJDNS(addr), NODE_NONE};
1865 : }
1866 :
1867 0 : const CAddress addr_bind{MaybeFlipIPv6toCJDNS(GetBindAddress(*sock)), NODE_NONE};
1868 :
1869 0 : NetPermissionFlags permission_flags = NetPermissionFlags::None;
1870 0 : hListenSocket.AddSocketPermissionFlags(permission_flags);
1871 :
1872 0 : CreateNodeFromAcceptedSocket(std::move(sock), permission_flags, addr_bind, addr, mn_sync);
1873 0 : }
1874 :
1875 0 : void CConnman::CreateNodeFromAcceptedSocket(std::unique_ptr<Sock>&& sock,
1876 : NetPermissionFlags permission_flags,
1877 : const CAddress& addr_bind,
1878 : const CAddress& addr,
1879 : CMasternodeSync& mn_sync)
1880 : {
1881 0 : int nInbound = 0;
1882 0 : int nVerifiedInboundMasternodes = 0;
1883 0 : int nMaxInbound = nMaxConnections - m_max_outbound;
1884 :
1885 0 : AddWhitelistPermissionFlags(permission_flags, addr);
1886 0 : if (NetPermissions::HasFlag(permission_flags, NetPermissionFlags::Implicit)) {
1887 0 : NetPermissions::ClearFlag(permission_flags, NetPermissionFlags::Implicit);
1888 0 : if (gArgs.GetBoolArg("-whitelistforcerelay", DEFAULT_WHITELISTFORCERELAY)) NetPermissions::AddFlag(permission_flags, NetPermissionFlags::ForceRelay);
1889 0 : if (gArgs.GetBoolArg("-whitelistrelay", DEFAULT_WHITELISTRELAY)) NetPermissions::AddFlag(permission_flags, NetPermissionFlags::Relay);
1890 0 : NetPermissions::AddFlag(permission_flags, NetPermissionFlags::Mempool);
1891 0 : NetPermissions::AddFlag(permission_flags, NetPermissionFlags::NoBan);
1892 0 : }
1893 :
1894 :
1895 : {
1896 0 : READ_LOCK(m_nodes_mutex);
1897 0 : for (const CNode* pnode : m_nodes) {
1898 0 : if (pnode->IsInboundConn()) {
1899 0 : nInbound++;
1900 0 : if (!pnode->GetVerifiedProRegTxHash().IsNull()) {
1901 0 : nVerifiedInboundMasternodes++;
1902 0 : }
1903 0 : }
1904 : }
1905 0 : }
1906 :
1907 0 : std::string strDropped;
1908 0 : if (fLogIPs) {
1909 0 : strDropped = strprintf("connection from %s dropped", addr.ToStringAddrPort());
1910 0 : } else {
1911 0 : strDropped = "connection dropped";
1912 : }
1913 :
1914 0 : if (!fNetworkActive) {
1915 0 : LogPrint(BCLog::NET_NETCONN, "%s: not accepting new connections\n", strDropped);
1916 0 : return;
1917 : }
1918 :
1919 0 : if (!sock->IsSelectable(/*is_select=*/::g_socket_events_mode == SocketEventsMode::Select)) {
1920 0 : LogPrintf("%s: non-selectable socket\n", strDropped);
1921 0 : return;
1922 : }
1923 :
1924 : // According to the internet TCP_NODELAY is not carried into accepted sockets
1925 : // on all platforms. Set it again here just to be sure.
1926 0 : const int on{1};
1927 0 : if (sock->SetSockOpt(IPPROTO_TCP, TCP_NODELAY, &on, sizeof(on)) == SOCKET_ERROR) {
1928 0 : LogPrint(BCLog::NET, "connection from %s: unable to set TCP_NODELAY, continuing anyway\n",
1929 : addr.ToStringAddrPort());
1930 0 : }
1931 :
1932 : // Don't accept connections from banned peers.
1933 0 : bool banned = m_banman && m_banman->IsBanned(addr);
1934 0 : if (!NetPermissions::HasFlag(permission_flags, NetPermissionFlags::NoBan) && banned)
1935 : {
1936 0 : LogPrint(BCLog::NET_NETCONN, "%s (banned)\n", strDropped);
1937 0 : return;
1938 : }
1939 :
1940 : // Only accept connections from discouraged peers if our inbound slots aren't (almost) full.
1941 0 : bool discouraged = m_banman && m_banman->IsDiscouraged(addr);
1942 0 : if (!NetPermissions::HasFlag(permission_flags, NetPermissionFlags::NoBan) && nInbound + 1 >= nMaxInbound && discouraged)
1943 : {
1944 0 : LogPrint(BCLog::NET_NETCONN, "connection from %s dropped (discouraged)\n", addr.ToStringAddrPort());
1945 0 : return;
1946 : }
1947 :
1948 : // Evict connections until we are below nMaxInbound. In case eviction protection resulted in nodes to not be evicted
1949 : // before, they might get evicted in batches now (after the protection timeout).
1950 : // We don't evict verified MN connections and also don't take them into account when checking limits. We can do this
1951 : // because we know that such connections are naturally limited by the total number of MNs, so this is not usable
1952 : // for attacks.
1953 0 : while (nInbound - nVerifiedInboundMasternodes >= nMaxInbound)
1954 : {
1955 0 : if (!AttemptToEvictConnection()) {
1956 : // No connection to evict, disconnect the new connection
1957 0 : LogPrint(BCLog::NET_NETCONN, "failed to find an eviction candidate - connection dropped (full)\n");
1958 0 : return;
1959 : }
1960 0 : nInbound--;
1961 : }
1962 :
1963 : // don't accept incoming connections until blockchain is synced
1964 0 : if (m_active_masternode && !mn_sync.IsBlockchainSynced()) {
1965 0 : LogPrint(BCLog::NET_NETCONN, "AcceptConnection -- blockchain is not synced yet, skipping inbound connection attempt\n");
1966 0 : return;
1967 : }
1968 :
1969 0 : NodeId id = GetNewNodeId();
1970 0 : uint64_t nonce = GetDeterministicRandomizer(RANDOMIZER_ID_LOCALHOSTNONCE).Write(id).Finalize();
1971 :
1972 0 : ServiceFlags nodeServices = nLocalServices;
1973 0 : if (NetPermissions::HasFlag(permission_flags, NetPermissionFlags::BloomFilter)) {
1974 0 : nodeServices = static_cast<ServiceFlags>(nodeServices | NODE_BLOOM);
1975 0 : }
1976 :
1977 0 : const bool inbound_onion = std::find(m_onion_binds.begin(), m_onion_binds.end(), addr_bind) != m_onion_binds.end();
1978 : // The V2Transport transparently falls back to V1 behavior when an incoming V1 connection is
1979 : // detected, so use it whenever we signal NODE_P2P_V2.
1980 0 : const bool use_v2transport(nodeServices & NODE_P2P_V2);
1981 :
1982 0 : CNode* pnode = new CNode(id,
1983 0 : std::move(sock),
1984 0 : addr,
1985 0 : CalculateKeyedNetGroup(addr),
1986 0 : nonce,
1987 0 : addr_bind,
1988 0 : /*addrNameIn=*/"",
1989 : ConnectionType::INBOUND,
1990 0 : inbound_onion,
1991 0 : CNodeOptions{
1992 0 : .permission_flags = permission_flags,
1993 : .prefer_evict = discouraged,
1994 : .recv_flood_size = nReceiveFloodSize,
1995 : .use_v2transport = use_v2transport,
1996 : });
1997 : pnode->AddRef();
1998 : m_msgproc->InitializeNode(*pnode, nodeServices);
1999 :
2000 : {
2001 : LOCK(pnode->m_sock_mutex);
2002 : if (fLogIPs) {
2003 : LogPrint(BCLog::NET_NETCONN, "connection from %s accepted, sock=%d, peer=%d\n", addr.ToStringAddrPort(), pnode->m_sock->Get(), pnode->GetId());
2004 : } else {
2005 : LogPrint(BCLog::NET_NETCONN, "connection accepted, sock=%d, peer=%d\n", pnode->m_sock->Get(), pnode->GetId());
2006 : }
2007 : }
2008 :
2009 : {
2010 : LOCK(m_nodes_mutex);
2011 : m_nodes.push_back(pnode);
2012 : }
2013 : {
2014 : LOCK2(cs_mapSocketToNode, pnode->m_sock_mutex);
2015 : mapSocketToNode.emplace(pnode->m_sock->Get(), pnode);
2016 : if (m_edge_trig_events) {
2017 : if (!m_edge_trig_events->RegisterEvents(pnode->m_sock->Get())) {
2018 : LogPrint(BCLog::NET, "EdgeTriggeredEvents::RegisterEvents() failed\n");
2019 : }
2020 : }
2021 : if (m_wakeup_pipe) {
2022 : m_wakeup_pipe->Write();
2023 : }
2024 : }
2025 :
2026 : // We received a new connection, harvest entropy from the time (and our peer count)
2027 : RandAddEvent((uint32_t)id);
2028 0 : }
2029 :
2030 0 : bool CConnman::AddConnection(const std::string& address, ConnectionType conn_type, bool use_v2transport = false)
2031 : {
2032 0 : AssertLockNotHeld(m_unused_i2p_sessions_mutex);
2033 0 : std::optional<int> max_connections;
2034 0 : switch (conn_type) {
2035 : case ConnectionType::INBOUND:
2036 : case ConnectionType::MANUAL:
2037 0 : return false;
2038 : case ConnectionType::OUTBOUND_FULL_RELAY:
2039 0 : max_connections = m_max_outbound_full_relay;
2040 0 : break;
2041 : case ConnectionType::BLOCK_RELAY:
2042 0 : max_connections = m_max_outbound_block_relay;
2043 0 : break;
2044 : // no limit for ADDR_FETCH because -seednode has no limit either
2045 : case ConnectionType::ADDR_FETCH:
2046 0 : break;
2047 : // no limit for FEELER connections since they're short-lived
2048 : case ConnectionType::FEELER:
2049 0 : break;
2050 : } // no default case, so the compiler can warn about missing cases
2051 :
2052 : // Count existing connections
2053 0 : int existing_connections = WITH_READ_LOCK(m_nodes_mutex,
2054 : return std::count_if(m_nodes.begin(), m_nodes.end(), [conn_type](CNode* node) { return node->m_conn_type == conn_type; }););
2055 :
2056 : // Max connections of specified type already exist
2057 0 : if (max_connections != std::nullopt && existing_connections >= max_connections) return false;
2058 :
2059 : // Max total outbound connections already exist
2060 0 : CSemaphoreGrant grant(*semOutbound, true);
2061 0 : if (!grant) return false;
2062 :
2063 0 : OpenNetworkConnection(CAddress(), false, std::move(grant), address.c_str(), conn_type, /*use_v2transport=*/use_v2transport);
2064 0 : return true;
2065 0 : }
2066 :
2067 0 : void CConnman::DisconnectNodes()
2068 : {
2069 0 : AssertLockNotHeld(m_nodes_mutex);
2070 0 : AssertLockNotHeld(m_reconnections_mutex);
2071 :
2072 : // Use a temporary variable to accumulate desired reconnections, so we don't need
2073 : // m_reconnections_mutex while holding m_nodes_mutex.
2074 0 : decltype(m_reconnections) reconnections_to_add;
2075 :
2076 : // Quick check without exclusive lock to avoid unnecessary locking
2077 : // Note: This is a best-effort optimization. If network is active and no nodes
2078 : // are marked for disconnect, we can skip the expensive exclusive lock.
2079 : // If network is inactive or nodes need cleanup, we must take the lock.
2080 0 : bool has_to_disconnect = false;
2081 0 : if (fNetworkActive) {
2082 : {
2083 0 : READ_LOCK(m_nodes_mutex);
2084 0 : for (CNode* pnode : m_nodes) {
2085 0 : if (pnode->fDisconnect) {
2086 0 : has_to_disconnect = true;
2087 0 : break;
2088 : }
2089 : }
2090 0 : }
2091 : // Only return early if network is active AND no nodes need disconnection
2092 : // AND no disconnected nodes need cleanup (checked below)
2093 0 : if (!has_to_disconnect && m_nodes_disconnected.empty()) {
2094 0 : return;
2095 : }
2096 0 : }
2097 :
2098 0 : if (has_to_disconnect || !fNetworkActive) {
2099 0 : LOCK(m_nodes_mutex);
2100 :
2101 0 : if (!fNetworkActive) {
2102 : // Disconnect any connected nodes
2103 0 : for (CNode* pnode : m_nodes) {
2104 0 : if (!pnode->fDisconnect) {
2105 0 : LogPrint(BCLog::NET_NETCONN, "Network not active, dropping peer=%d\n", pnode->GetId());
2106 0 : pnode->fDisconnect = true;
2107 0 : }
2108 : }
2109 0 : }
2110 :
2111 : // Disconnect unused nodes, if we have any
2112 0 : for (auto it = m_nodes.begin(); it != m_nodes.end(); )
2113 : {
2114 0 : CNode* pnode = *it;
2115 0 : if (pnode->fDisconnect)
2116 : {
2117 : // If we were the ones who initiated the disconnect, we must assume that the other side wants to see
2118 : // pending messages. If the other side initiated the disconnect (or disconnected after we've shutdown
2119 : // the socket), we can be pretty sure that they are not interested in any pending messages anymore and
2120 : // thus can immediately close the socket.
2121 0 : if (!pnode->fOtherSideDisconnected) {
2122 0 : if (pnode->nDisconnectLingerTime.load() == SteadyClock::time_point{}) {
2123 : // let's not immediately close the socket but instead wait for at least 100ms so that there is a
2124 : // chance to flush all/some pending data. Otherwise the other side might not receive REJECT messages
2125 : // that were pushed right before setting fDisconnect=true
2126 : // Flushing must happen in two places to ensure data can be received by the other side:
2127 : // 1. vSendMsg must be empty and all messages sent via send(). This is ensured by SocketHandler()
2128 : // being called before DisconnectNodes and also by the linger time
2129 : // 2. Internal socket send buffers must be flushed. This is ensured solely by the linger time
2130 0 : pnode->nDisconnectLingerTime = SteadyClock::now() + 100ms;
2131 0 : }
2132 0 : if (SteadyClock::now() < pnode->nDisconnectLingerTime.load()) {
2133 : // everything flushed to the kernel?
2134 0 : const auto& [to_send, more, _msg_type] = pnode->m_transport->GetBytesToSend(pnode->nSendMsgSize != 0);
2135 0 : const bool queue_is_empty{to_send.empty() && !more};
2136 0 : if (!pnode->fSocketShutdown && queue_is_empty) {
2137 0 : LOCK(pnode->m_sock_mutex);
2138 0 : if (pnode->m_sock) {
2139 : // Give the other side a chance to detect the disconnect as early as possible (recv() will return 0)
2140 0 : ::shutdown(pnode->m_sock->Get(), SD_SEND);
2141 0 : }
2142 0 : pnode->fSocketShutdown = true;
2143 0 : }
2144 0 : ++it;
2145 0 : continue;
2146 : }
2147 0 : }
2148 :
2149 0 : if (fLogIPs) {
2150 0 : LogPrintf("ThreadSocketHandler -- removing node: peer=%d addr=%s nRefCount=%d fInbound=%d m_masternode_connection=%d m_masternode_iqr_connection=%d\n",
2151 : pnode->GetId(), pnode->addr.ToStringAddrPort(), pnode->GetRefCount(), pnode->IsInboundConn(), pnode->m_masternode_connection, pnode->m_masternode_iqr_connection);
2152 0 : } else {
2153 0 : LogPrintf("ThreadSocketHandler -- removing node: peer=%d nRefCount=%d fInbound=%d m_masternode_connection=%d m_masternode_iqr_connection=%d\n",
2154 : pnode->GetId(), pnode->GetRefCount(), pnode->IsInboundConn(), pnode->m_masternode_connection, pnode->m_masternode_iqr_connection);
2155 : }
2156 :
2157 : // remove from m_nodes
2158 0 : it = m_nodes.erase(it);
2159 :
2160 : // Add to reconnection list if appropriate. We don't reconnect right here, because
2161 : // the creation of a connection is a blocking operation (up to several seconds),
2162 : // and we don't want to hold up the socket handler thread for that long.
2163 0 : if (pnode->m_transport->ShouldReconnectV1()) {
2164 0 : reconnections_to_add.push_back({
2165 0 : .addr_connect = pnode->addr,
2166 0 : .grant = std::move(pnode->grantOutbound),
2167 0 : .destination = pnode->m_dest,
2168 0 : .conn_type = pnode->m_conn_type,
2169 : .use_v2transport = false,
2170 0 : .masternode_connection = pnode->m_masternode_connection,
2171 0 : .masternode_probe_connection = pnode->m_masternode_probe_connection});
2172 0 : LogPrint(BCLog::NET, "retrying with v1 transport protocol for peer=%d\n", pnode->GetId());
2173 0 : }
2174 :
2175 : // release outbound grant (if any)
2176 0 : pnode->grantOutbound.Release();
2177 :
2178 : // close socket and cleanup
2179 0 : pnode->CloseSocketDisconnect(this);
2180 :
2181 : // update connection count by network
2182 0 : if (pnode->IsManualOrFullOutboundConn()) --m_network_conn_counts[pnode->addr.GetNetwork()];
2183 :
2184 : // hold in disconnected pool until all refs are released
2185 0 : pnode->Release();
2186 0 : m_nodes_disconnected.push_back(pnode);
2187 0 : } else {
2188 0 : ++it;
2189 : }
2190 : }
2191 0 : }
2192 : {
2193 : // Delete disconnected nodes
2194 0 : std::list<CNode*> nodes_disconnected_copy = m_nodes_disconnected;
2195 0 : for (auto it = m_nodes_disconnected.begin(); it != m_nodes_disconnected.end(); )
2196 : {
2197 0 : CNode* pnode = *it;
2198 : // Destroy the object only after other threads have stopped using it.
2199 0 : if (pnode->GetRefCount() <= 0) {
2200 0 : it = m_nodes_disconnected.erase(it);
2201 0 : DeleteNode(pnode);
2202 0 : } else {
2203 0 : ++it;
2204 : }
2205 : }
2206 0 : }
2207 : {
2208 : // Move entries from reconnections_to_add to m_reconnections.
2209 0 : LOCK(m_reconnections_mutex);
2210 0 : m_reconnections.splice(m_reconnections.end(), std::move(reconnections_to_add));
2211 0 : }
2212 0 : }
2213 :
2214 0 : void CConnman::NotifyNumConnectionsChanged(CMasternodeSync& mn_sync)
2215 : {
2216 0 : size_t nodes_size = WITH_READ_LOCK(m_nodes_mutex, return m_nodes.size(););
2217 :
2218 : // If we had zero connections before and new connections now or if we just dropped
2219 : // to zero connections reset the sync process if its outdated.
2220 0 : if ((nodes_size > 0 && nPrevNodeCount == 0) || (nodes_size == 0 && nPrevNodeCount > 0)) {
2221 0 : mn_sync.Reset();
2222 0 : }
2223 :
2224 0 : if (nodes_size != nPrevNodeCount) {
2225 0 : nPrevNodeCount = nodes_size;
2226 0 : if (m_client_interface) {
2227 0 : m_client_interface->NotifyNumConnectionsChanged(nodes_size);
2228 0 : }
2229 :
2230 0 : CalculateNumConnectionsChangedStats();
2231 0 : }
2232 0 : }
2233 :
2234 0 : void CConnman::CalculateNumConnectionsChangedStats()
2235 : {
2236 0 : if (!::g_stats_client->active()) {
2237 0 : return;
2238 : }
2239 :
2240 : // count various node attributes for statsD
2241 0 : int fullNodes = 0;
2242 0 : int spvNodes = 0;
2243 0 : int inboundNodes = 0;
2244 0 : int outboundNodes = 0;
2245 0 : int ipv4Nodes = 0;
2246 0 : int ipv6Nodes = 0;
2247 0 : int torNodes = 0;
2248 0 : mapMsgTypeSize mapRecvBytesMsgStats;
2249 0 : mapMsgTypeSize mapSentBytesMsgStats;
2250 0 : for (const std::string &msg : getAllNetMessageTypes()) {
2251 0 : mapRecvBytesMsgStats[msg] = 0;
2252 0 : mapSentBytesMsgStats[msg] = 0;
2253 : }
2254 0 : mapRecvBytesMsgStats[NET_MESSAGE_TYPE_OTHER] = 0;
2255 0 : mapSentBytesMsgStats[NET_MESSAGE_TYPE_OTHER] = 0;
2256 0 : const NodesSnapshot snap{*this, /* cond = */ CConnman::FullyConnectedOnly};
2257 0 : for (auto pnode : snap.Nodes()) {
2258 0 : WITH_LOCK(pnode->cs_vRecv, pnode->UpdateRecvMapWithStats(mapRecvBytesMsgStats));
2259 0 : WITH_LOCK(pnode->cs_vSend, pnode->UpdateSentMapWithStats(mapSentBytesMsgStats));
2260 0 : if (pnode->m_bloom_filter_loaded.load()) {
2261 0 : spvNodes++;
2262 0 : } else {
2263 0 : fullNodes++;
2264 : }
2265 0 : if(pnode->IsInboundConn())
2266 0 : inboundNodes++;
2267 : else
2268 0 : outboundNodes++;
2269 0 : if(pnode->addr.IsIPv4())
2270 0 : ipv4Nodes++;
2271 0 : if(pnode->addr.IsIPv6())
2272 0 : ipv6Nodes++;
2273 0 : if(pnode->addr.IsTor())
2274 0 : torNodes++;
2275 0 : const auto last_ping_time = count_microseconds(pnode->m_last_ping_time);
2276 0 : if (last_ping_time > 0)
2277 0 : ::g_stats_client->timing("peers.ping_us", last_ping_time, 1.0f);
2278 : }
2279 0 : for (const std::string &msg : getAllNetMessageTypes()) {
2280 0 : ::g_stats_client->gauge("bandwidth.message." + msg + ".totalBytesReceived", mapRecvBytesMsgStats[msg], 1.0f);
2281 0 : ::g_stats_client->gauge("bandwidth.message." + msg + ".totalBytesSent", mapSentBytesMsgStats[msg], 1.0f);
2282 : }
2283 0 : ::g_stats_client->gauge("peers.totalConnections", nPrevNodeCount, 1.0f);
2284 0 : ::g_stats_client->gauge("peers.spvNodeConnections", spvNodes, 1.0f);
2285 0 : ::g_stats_client->gauge("peers.fullNodeConnections", fullNodes, 1.0f);
2286 0 : ::g_stats_client->gauge("peers.inboundConnections", inboundNodes, 1.0f);
2287 0 : ::g_stats_client->gauge("peers.outboundConnections", outboundNodes, 1.0f);
2288 0 : ::g_stats_client->gauge("peers.ipv4Connections", ipv4Nodes, 1.0f);
2289 0 : ::g_stats_client->gauge("peers.ipv6Connections", ipv6Nodes, 1.0f);
2290 0 : ::g_stats_client->gauge("peers.torConnections", torNodes, 1.0f);
2291 0 : }
2292 :
2293 6 : bool CConnman::ShouldRunInactivityChecks(const CNode& node, std::chrono::seconds now) const
2294 : {
2295 6 : return node.m_connected + m_peer_connect_timeout < now;
2296 : }
2297 :
2298 0 : bool CConnman::InactivityCheck(const CNode& node) const
2299 : {
2300 : // Tests that see disconnects after using mocktime can start nodes with a
2301 : // large timeout. For example, -peertimeout=999999999.
2302 0 : const auto now{GetTime<std::chrono::seconds>()};
2303 0 : const auto last_send{node.m_last_send.load()};
2304 0 : const auto last_recv{node.m_last_recv.load()};
2305 :
2306 0 : if (!ShouldRunInactivityChecks(node, now)) return false;
2307 :
2308 0 : if (last_recv.count() == 0 || last_send.count() == 0) {
2309 0 : LogPrint(BCLog::NET, "socket no message in first %i seconds, %d %d peer=%d\n", count_seconds(m_peer_connect_timeout), last_recv.count() != 0, last_send.count() != 0, node.GetId());
2310 0 : return true;
2311 : }
2312 :
2313 0 : if (now > last_send + TIMEOUT_INTERVAL) {
2314 0 : LogPrint(BCLog::NET, "socket sending timeout: %is peer=%d\n", count_seconds(now - last_send), node.GetId());
2315 0 : return true;
2316 : }
2317 :
2318 0 : if (now > last_recv + TIMEOUT_INTERVAL) {
2319 0 : LogPrint(BCLog::NET, "socket receive timeout: %is peer=%d\n", count_seconds(now - last_recv), node.GetId());
2320 0 : return true;
2321 : }
2322 :
2323 0 : if (!node.fSuccessfullyConnected) {
2324 0 : if (node.m_transport->GetInfo().transport_type == TransportProtocolType::DETECTING) {
2325 0 : LogPrint(BCLog::NET, "V2 handshake timeout peer=%d\n", node.GetId());
2326 0 : } else {
2327 0 : LogPrint(BCLog::NET, "version handshake timeout peer=%d\n", node.GetId());
2328 : }
2329 0 : return true;
2330 : }
2331 :
2332 0 : return false;
2333 0 : }
2334 :
2335 0 : Sock::EventsPerSock CConnman::GenerateWaitSockets(Span<CNode* const> nodes)
2336 : {
2337 0 : Sock::EventsPerSock events_per_sock;
2338 :
2339 0 : for (const ListenSocket& hListenSocket : vhListenSocket) {
2340 0 : events_per_sock.emplace(hListenSocket.sock->Get(), Sock::Events{Sock::RECV});
2341 : }
2342 :
2343 0 : for (CNode* pnode : nodes) {
2344 0 : bool select_recv = !pnode->fHasRecvData;
2345 0 : bool select_send = !pnode->fCanSendData;
2346 0 : if (!select_recv && !select_send) continue;
2347 :
2348 0 : LOCK(pnode->m_sock_mutex);
2349 0 : if (!pnode->m_sock) {
2350 0 : continue;
2351 : }
2352 :
2353 0 : Sock::Event requested{0};
2354 0 : if (select_send) {
2355 0 : requested |= Sock::SEND;
2356 0 : }
2357 0 : if (select_recv) {
2358 0 : requested |= Sock::RECV;
2359 0 : }
2360 :
2361 0 : events_per_sock.emplace(pnode->m_sock->Get(), Sock::Events{requested});
2362 0 : }
2363 :
2364 0 : if (m_wakeup_pipe) {
2365 : // We add a pipe to the read set so that the select() call can be woken up from the outside
2366 : // This is done when data is added to send buffers (vSendMsg) or when new peers are added
2367 : // This is currently only implemented for POSIX compliant systems. This means that Windows will fall back to
2368 : // timing out after 50ms and then trying to send. This is ok as we assume that heavy-load daemons are usually
2369 : // run on Linux and friends.
2370 0 : events_per_sock.emplace(m_wakeup_pipe->m_pipe[0], Sock::Events{Sock::RECV});
2371 0 : }
2372 :
2373 0 : return events_per_sock;
2374 0 : }
2375 :
2376 0 : void CConnman::SocketHandler(CMasternodeSync& mn_sync)
2377 : {
2378 0 : AssertLockNotHeld(m_total_bytes_sent_mutex);
2379 :
2380 0 : Sock::EventsPerSock events_per_sock;
2381 :
2382 0 : bool only_poll = [this]() {
2383 : // Check if we have work to do and thus should avoid waiting for events
2384 0 : READ_LOCK(m_nodes_mutex); // We acquire this to avoid the pointers stored in mapSendableNodes and mapReceivableNodes being invalidated by ThreadSocketHandler
2385 0 : LOCK(cs_sendable_receivable_nodes);
2386 0 : if (!mapReceivableNodes.empty()) {
2387 0 : return true;
2388 : }
2389 0 : for (const auto& p : mapSendableNodes) {
2390 0 : const auto& [to_send, more, _msg_type] = p.second->m_transport->GetBytesToSend(p.second->nSendMsgSize != 0);
2391 0 : if (!to_send.empty()) {
2392 0 : return true;
2393 : }
2394 : }
2395 0 : return false;
2396 0 : }();
2397 :
2398 : {
2399 0 : const NodesSnapshot snap{*this, /* cond = */ CConnman::AllNodes, /* shuffle = */ false};
2400 :
2401 0 : const auto timeout = std::chrono::milliseconds(only_poll ? 0 : SELECT_TIMEOUT_MILLISECONDS);
2402 0 : const bool is_lt = socketEventsMode == SocketEventsMode::Poll || socketEventsMode == SocketEventsMode::Select;
2403 :
2404 : // Check for the readiness of the already connected sockets and the
2405 : // listening sockets in one call ("readiness" as in poll(2) or
2406 : // select(2)). If none are ready, wait for a short while and return
2407 : // empty sets.
2408 0 : events_per_sock = GenerateWaitSockets(snap.Nodes());
2409 0 : if ((is_lt && events_per_sock.empty()) || !Sock::WaitManyInternal(timeout, events_per_sock, SocketEventsParams{socketEventsMode, GetModeFileDescriptor(), ToggleWakeupPipe})) {
2410 0 : if (is_lt) {
2411 0 : interruptNet.sleep_for(std::chrono::milliseconds(SELECT_TIMEOUT_MILLISECONDS));
2412 0 : }
2413 0 : }
2414 :
2415 : // Drain the wakeup pipe
2416 0 : if (m_wakeup_pipe && events_per_sock.find(m_wakeup_pipe->m_pipe[0]) != events_per_sock.end()) {
2417 0 : m_wakeup_pipe->Drain();
2418 0 : }
2419 :
2420 : // Service (send/receive) each of the already connected nodes.
2421 0 : SocketHandlerConnected(events_per_sock);
2422 0 : }
2423 :
2424 : // Accept new connections from listening sockets.
2425 0 : SocketHandlerListening(events_per_sock, mn_sync);
2426 0 : }
2427 :
2428 0 : void CConnman::SocketHandlerConnected(const Sock::EventsPerSock& events_per_sock)
2429 : {
2430 0 : AssertLockNotHeld(m_total_bytes_sent_mutex);
2431 :
2432 0 : if (interruptNet) return;
2433 :
2434 0 : std::set<CNode*> node_err_set;
2435 0 : std::set<CNode*> node_recv_set;
2436 0 : std::set<CNode*> node_send_set;
2437 : {
2438 0 : LOCK(cs_mapSocketToNode);
2439 0 : for (const auto& [sock, events] : events_per_sock) {
2440 0 : auto it = mapSocketToNode.find(sock);
2441 0 : if (it == mapSocketToNode.end()) continue;
2442 0 : if (events.occurred & Sock::ERR) {
2443 0 : it->second->AddRef();
2444 0 : node_err_set.emplace(it->second);
2445 0 : }
2446 0 : if (events.occurred & Sock::RECV) {
2447 0 : if (events.occurred & Sock::ERR) continue;
2448 0 : LOCK(cs_sendable_receivable_nodes);
2449 0 : auto jt = mapReceivableNodes.emplace(it->second->GetId(), it->second);
2450 0 : assert(jt.first->second == it->second);
2451 0 : it->second->fHasRecvData = true;
2452 0 : }
2453 0 : if (events.occurred & Sock::SEND) {
2454 0 : LOCK(cs_sendable_receivable_nodes);
2455 0 : auto jt = mapSendableNodes.emplace(it->second->GetId(), it->second);
2456 0 : assert(jt.first->second == it->second);
2457 0 : it->second->fCanSendData = true;
2458 0 : }
2459 : }
2460 0 : }
2461 :
2462 0 : ForEachNode(AllNodes, [&](CNode* pnode) {
2463 0 : const auto& [to_send, more, _msg_type] = pnode->m_transport->GetBytesToSend(pnode->nSendMsgSize != 0);
2464 : // Collect nodes that have a receivable socket, implement the following logic:
2465 : // * If there is data to send, try sending data. As this only
2466 : // happens when optimistic write failed, we choose to first drain the
2467 : // write buffer in this case before receiving more. This avoids
2468 : // needlessly queueing received data, if the remote peer is not themselves
2469 : // receiving data. This means properly utilizing TCP flow control signalling.
2470 : // * Otherwise, if there is space left in the receive buffer (!fPauseRecv), try
2471 : // receiving data (which should succeed as the socket signalled as receivable).
2472 0 : if (pnode->fHasRecvData && !pnode->fPauseRecv && !pnode->fDisconnect &&
2473 0 : (!pnode->m_transport->ReceivedMessageComplete() || to_send.empty())) {
2474 0 : pnode->AddRef();
2475 0 : node_recv_set.emplace(pnode);
2476 0 : }
2477 :
2478 : // Collect nodes that have data to send and have a socket with non-empty write buffers
2479 0 : if (pnode->fCanSendData && (!pnode->m_transport->ReceivedMessageComplete() || !to_send.empty())) {
2480 0 : pnode->AddRef();
2481 0 : node_send_set.emplace(pnode);
2482 0 : }
2483 0 : });
2484 :
2485 0 : for (CNode* pnode : node_send_set) {
2486 0 : if (interruptNet) {
2487 0 : break;
2488 : }
2489 :
2490 : // Send data
2491 0 : auto [bytes_sent, data_left] = WITH_LOCK(pnode->cs_vSend, return SocketSendData(*pnode));
2492 0 : if (bytes_sent) {
2493 0 : RecordBytesSent(bytes_sent);
2494 :
2495 : // If both receiving and (non-optimistic) sending were possible, we first attempt
2496 : // sending. If that succeeds, but does not fully drain the send queue, do not
2497 : // attempt to receive. This avoids needlessly queueing data if the remote peer
2498 : // is slow at receiving data, by means of TCP flow control. We only do this when
2499 : // sending actually succeeded to make sure progress is always made; otherwise a
2500 : // deadlock would be possible when both sides have data to send, but neither is
2501 : // receiving.
2502 0 : if (data_left && node_recv_set.erase(pnode)) {
2503 0 : pnode->Release();
2504 0 : }
2505 0 : }
2506 : }
2507 :
2508 0 : for (CNode* pnode : node_err_set)
2509 : {
2510 0 : if (interruptNet) {
2511 0 : break;
2512 : }
2513 : // let recv() return errors and then handle it
2514 0 : SocketRecvData(pnode);
2515 : }
2516 :
2517 0 : for (CNode* pnode : node_recv_set)
2518 : {
2519 0 : if (interruptNet) {
2520 0 : break;
2521 : }
2522 0 : if (pnode->fPauseRecv) {
2523 0 : continue;
2524 : }
2525 :
2526 0 : SocketRecvData(pnode);
2527 : }
2528 :
2529 0 : for (auto& node : node_err_set) {
2530 0 : node->Release();
2531 : }
2532 0 : for (auto& node : node_recv_set) {
2533 0 : node->Release();
2534 : }
2535 0 : for (auto& node : node_send_set) {
2536 0 : node->Release();
2537 : }
2538 :
2539 0 : if (interruptNet) {
2540 0 : return;
2541 : }
2542 :
2543 : {
2544 0 : LOCK(cs_sendable_receivable_nodes);
2545 : // remove nodes from mapSendableNodes, so that the next iteration knows that there is no work to do
2546 : // (even if there are pending messages to be sent)
2547 0 : for (auto it = mapSendableNodes.begin(); it != mapSendableNodes.end(); ) {
2548 0 : if (!it->second->fCanSendData) {
2549 0 : it = mapSendableNodes.erase(it);
2550 0 : } else {
2551 0 : ++it;
2552 : }
2553 : }
2554 : // clean up mapReceivableNodes from nodes that were receivable in the last iteration but aren't anymore
2555 0 : for (auto it = mapReceivableNodes.begin(); it != mapReceivableNodes.end(); ) {
2556 0 : if (!it->second->fHasRecvData) {
2557 0 : it = mapReceivableNodes.erase(it);
2558 0 : } else {
2559 0 : ++it;
2560 : }
2561 : }
2562 0 : }
2563 0 : }
2564 :
2565 0 : void CConnman::SocketHandlerListening(const Sock::EventsPerSock& events_per_sock, CMasternodeSync& mn_sync)
2566 : {
2567 0 : for (const ListenSocket& listen_socket : vhListenSocket) {
2568 0 : if (interruptNet) {
2569 0 : return;
2570 : }
2571 0 : const auto it = events_per_sock.find(listen_socket.sock->Get());
2572 0 : if (it != events_per_sock.end() && (it->second.occurred & Sock::RECV)) {
2573 0 : AcceptConnection(listen_socket, mn_sync);
2574 0 : }
2575 : }
2576 0 : }
2577 :
2578 0 : size_t CConnman::SocketRecvData(CNode *pnode)
2579 : {
2580 : // typical socket buffer is 8K-64K
2581 : uint8_t pchBuf[0x10000];
2582 0 : int nBytes = 0;
2583 : {
2584 0 : LOCK(pnode->m_sock_mutex);
2585 0 : if (!pnode->m_sock)
2586 0 : return 0;
2587 0 : nBytes = recv(pnode->m_sock->Get(), (char*)pchBuf, sizeof(pchBuf), MSG_DONTWAIT);
2588 0 : if (nBytes < (int)sizeof(pchBuf)) {
2589 0 : pnode->fHasRecvData = false;
2590 0 : }
2591 0 : }
2592 0 : if (nBytes > 0)
2593 : {
2594 0 : bool notify = false;
2595 0 : if (!pnode->ReceiveMsgBytes(Span<const uint8_t>(pchBuf, nBytes), notify)) {
2596 0 : READ_LOCK(m_nodes_mutex); // is this here for lock ordering?
2597 0 : pnode->CloseSocketDisconnect(this);
2598 0 : }
2599 0 : RecordBytesRecv(nBytes);
2600 0 : if (notify) {
2601 0 : pnode->MarkReceivedMsgsForProcessing();
2602 0 : WakeMessageHandler();
2603 0 : }
2604 0 : }
2605 0 : else if (nBytes == 0)
2606 : {
2607 : // socket closed gracefully
2608 0 : if (!pnode->fDisconnect) {
2609 0 : LogPrint(BCLog::NET, "socket closed for peer=%d\n", pnode->GetId());
2610 0 : }
2611 0 : READ_LOCK(m_nodes_mutex); // is this here for lock ordering?
2612 0 : pnode->fOtherSideDisconnected = true; // avoid lingering
2613 0 : pnode->CloseSocketDisconnect(this);
2614 0 : }
2615 0 : else if (nBytes < 0)
2616 : {
2617 : // error
2618 0 : int nErr = WSAGetLastError();
2619 0 : if (nErr != WSAEWOULDBLOCK && nErr != WSAEMSGSIZE && nErr != WSAEINTR && nErr != WSAEINPROGRESS)
2620 : {
2621 0 : if (!pnode->fDisconnect){
2622 0 : LogPrint(BCLog::NET, "socket recv error for peer=%d: %s\n", pnode->GetId(), NetworkErrorString(nErr));
2623 0 : }
2624 0 : READ_LOCK(m_nodes_mutex); // is this here for lock ordering?
2625 0 : pnode->fOtherSideDisconnected = true; // avoid lingering
2626 0 : pnode->CloseSocketDisconnect(this);
2627 0 : }
2628 0 : }
2629 0 : if (nBytes < 0) {
2630 0 : return 0;
2631 : }
2632 0 : return (size_t)nBytes;
2633 0 : }
2634 :
2635 0 : void CConnman::ThreadSocketHandler(CMasternodeSync& mn_sync)
2636 : {
2637 0 : AssertLockNotHeld(m_total_bytes_sent_mutex);
2638 :
2639 0 : auto nLastCleanupNodes = SteadyClock::time_point{};
2640 :
2641 0 : while (!interruptNet)
2642 : {
2643 : // Handle sockets before we do the next round of disconnects. This allows us to flush send buffers one last time
2644 : // before actually closing sockets. Receiving is however skipped in case a peer is pending to be disconnected
2645 0 : SocketHandler(mn_sync);
2646 0 : if (SteadyClock::now() - nLastCleanupNodes > 1s) {
2647 0 : ForEachNode(AllNodes, [&](CNode* pnode) {
2648 0 : if (InactivityCheck(*pnode)) pnode->fDisconnect = true;
2649 0 : });
2650 0 : nLastCleanupNodes = SteadyClock::now();
2651 0 : }
2652 0 : DisconnectNodes();
2653 0 : NotifyNumConnectionsChanged(mn_sync);
2654 : }
2655 0 : }
2656 :
2657 0 : void CConnman::WakeMessageHandler()
2658 : {
2659 : {
2660 0 : LOCK(mutexMsgProc);
2661 0 : fMsgProcWake = true;
2662 0 : }
2663 0 : condMsgProc.notify_one();
2664 0 : }
2665 :
2666 0 : void CConnman::ThreadDNSAddressSeed()
2667 : {
2668 0 : FastRandomContext rng;
2669 0 : std::vector<std::string> seeds = Params().DNSSeeds();
2670 0 : Shuffle(seeds.begin(), seeds.end(), rng);
2671 0 : int seeds_right_now = 0; // Number of seeds left before testing if we have enough connections
2672 0 : int found = 0;
2673 :
2674 0 : if (gArgs.GetBoolArg("-forcednsseed", DEFAULT_FORCEDNSSEED)) {
2675 : // When -forcednsseed is provided, query all.
2676 0 : seeds_right_now = seeds.size();
2677 0 : } else if (addrman.Size() == 0) {
2678 : // If we have no known peers, query all.
2679 : // This will occur on the first run, or if peers.dat has been
2680 : // deleted.
2681 0 : seeds_right_now = seeds.size();
2682 0 : }
2683 :
2684 : // goal: only query DNS seed if address need is acute
2685 : // * If we have a reasonable number of peers in addrman, spend
2686 : // some time trying them first. This improves user privacy by
2687 : // creating fewer identifying DNS requests, reduces trust by
2688 : // giving seeds less influence on the network topology, and
2689 : // reduces traffic to the seeds.
2690 : // * When querying DNS seeds query a few at once, this ensures
2691 : // that we don't give DNS seeds the ability to eclipse nodes
2692 : // that query them.
2693 : // * If we continue having problems, eventually query all the
2694 : // DNS seeds, and if that fails too, also try the fixed seeds.
2695 : // (done in ThreadOpenConnections)
2696 0 : const std::chrono::seconds seeds_wait_time = (addrman.Size() >= DNSSEEDS_DELAY_PEER_THRESHOLD ? DNSSEEDS_DELAY_MANY_PEERS : DNSSEEDS_DELAY_FEW_PEERS);
2697 :
2698 0 : for (const std::string& seed : seeds) {
2699 0 : if (seeds_right_now == 0) {
2700 0 : seeds_right_now += DNSSEEDS_TO_QUERY_AT_ONCE;
2701 :
2702 0 : if (addrman.Size() > 0) {
2703 0 : LogPrintf("Waiting %d seconds before querying DNS seeds.\n", seeds_wait_time.count());
2704 0 : std::chrono::seconds to_wait = seeds_wait_time;
2705 0 : while (to_wait.count() > 0) {
2706 : // if sleeping for the MANY_PEERS interval, wake up
2707 : // early to see if we have enough peers and can stop
2708 : // this thread entirely freeing up its resources
2709 0 : std::chrono::seconds w = std::min(DNSSEEDS_DELAY_FEW_PEERS, to_wait);
2710 0 : if (!interruptNet.sleep_for(w)) return;
2711 0 : to_wait -= w;
2712 :
2713 0 : int nRelevant = 0;
2714 : {
2715 0 : READ_LOCK(m_nodes_mutex);
2716 0 : for (const CNode* pnode : m_nodes) {
2717 0 : if (pnode->fSuccessfullyConnected && pnode->IsFullOutboundConn() && !pnode->m_masternode_probe_connection) ++nRelevant;
2718 : }
2719 0 : }
2720 0 : if (nRelevant >= 2) {
2721 0 : if (found > 0) {
2722 0 : LogPrintf("%d addresses found from DNS seeds\n", found);
2723 0 : LogPrintf("P2P peers available. Finished DNS seeding.\n");
2724 0 : } else {
2725 0 : LogPrintf("P2P peers available. Skipped DNS seeding.\n");
2726 : }
2727 0 : return;
2728 : }
2729 : }
2730 0 : }
2731 0 : }
2732 :
2733 0 : if (interruptNet) return;
2734 :
2735 : // hold off on querying seeds if P2P network deactivated
2736 0 : if (!fNetworkActive) {
2737 0 : LogPrintf("Waiting for network to be reactivated before querying DNS seeds.\n");
2738 0 : do {
2739 0 : if (!interruptNet.sleep_for(std::chrono::seconds{1})) return;
2740 0 : } while (!fNetworkActive);
2741 0 : }
2742 :
2743 0 : LogPrintf("Loading addresses from DNS seed %s\n", seed);
2744 : // If -proxy is in use, we make an ADDR_FETCH connection to the DNS resolved peer address
2745 : // for the base dns seed domain in chainparams
2746 0 : if (HaveNameProxy()) {
2747 0 : AddAddrFetch(seed);
2748 0 : } else {
2749 0 : std::vector<CAddress> vAdd;
2750 0 : ServiceFlags requiredServiceBits = GetDesirableServiceFlags(NODE_NONE);
2751 0 : std::string host = strprintf("x%x.%s", requiredServiceBits, seed);
2752 0 : CNetAddr resolveSource;
2753 0 : if (!resolveSource.SetInternal(host)) {
2754 0 : continue;
2755 : }
2756 : // Limit number of IPs learned from a single DNS seed. This limit exists to prevent the results from
2757 : // one DNS seed from dominating AddrMan. Note that the number of results from a UDP DNS query is
2758 : // bounded to 33 already, but it is possible for it to use TCP where a larger number of results can be
2759 : // returned.
2760 0 : unsigned int nMaxIPs = 32;
2761 0 : const auto addresses{LookupHost(host, nMaxIPs, true)};
2762 0 : if (!addresses.empty()) {
2763 0 : for (const CNetAddr& ip : addresses) {
2764 0 : CAddress addr = CAddress(CService(ip, Params().GetDefaultPort()), requiredServiceBits);
2765 0 : addr.nTime = rng.rand_uniform_delay(Now<NodeSeconds>() - 3 * 24h, -4 * 24h); // use a random age between 3 and 7 days old
2766 0 : vAdd.push_back(addr);
2767 0 : found++;
2768 0 : }
2769 0 : addrman.Add(vAdd, resolveSource);
2770 0 : } else {
2771 : // If the seed does not support a subdomain with our desired service bits,
2772 : // we make an ADDR_FETCH connection to the DNS resolved peer address for the
2773 : // base dns seed domain in chainparams
2774 0 : AddAddrFetch(seed);
2775 : }
2776 0 : }
2777 0 : --seeds_right_now;
2778 : }
2779 0 : LogPrintf("%d addresses found from DNS seeds\n", found);
2780 0 : }
2781 :
2782 0 : void CConnman::DumpAddresses()
2783 : {
2784 0 : const auto start{SteadyClock::now()};
2785 :
2786 0 : DumpPeerAddresses(::gArgs, addrman);
2787 :
2788 0 : LogPrint(BCLog::NET, "Flushed %d addresses to peers.dat %dms\n",
2789 : addrman.Size(), Ticks<std::chrono::milliseconds>(SteadyClock::now() - start));
2790 0 : }
2791 :
2792 0 : void CConnman::ProcessAddrFetch()
2793 : {
2794 0 : AssertLockNotHeld(m_unused_i2p_sessions_mutex);
2795 0 : std::string strDest;
2796 : {
2797 0 : LOCK(m_addr_fetches_mutex);
2798 0 : if (m_addr_fetches.empty())
2799 0 : return;
2800 0 : strDest = m_addr_fetches.front();
2801 0 : m_addr_fetches.pop_front();
2802 0 : }
2803 : // Attempt v2 connection if we support v2 - we'll reconnect with v1 if our
2804 : // peer doesn't support it or immediately disconnects us for another reason.
2805 0 : const bool use_v2transport(GetLocalServices() & NODE_P2P_V2);
2806 0 : CAddress addr;
2807 0 : CSemaphoreGrant grant(*semOutbound, /*fTry=*/true);
2808 0 : if (grant) {
2809 0 : OpenNetworkConnection(addr, false, std::move(grant), strDest.c_str(), ConnectionType::ADDR_FETCH, use_v2transport);
2810 0 : }
2811 0 : }
2812 :
2813 3 : bool CConnman::GetTryNewOutboundPeer() const
2814 : {
2815 3 : return m_try_another_outbound_peer;
2816 : }
2817 :
2818 637 : void CConnman::SetTryNewOutboundPeer(bool flag)
2819 : {
2820 637 : m_try_another_outbound_peer = flag;
2821 637 : LogPrint(BCLog::NET, "setting try another outbound peer=%s\n", flag ? "true" : "false");
2822 637 : }
2823 :
2824 0 : void CConnman::StartExtraBlockRelayPeers()
2825 : {
2826 0 : LogPrint(BCLog::NET, "enabling extra block-relay-only peers\n");
2827 0 : m_start_extra_block_relay_peers = true;
2828 0 : }
2829 :
2830 : // Return the number of peers we have over our outbound connection limit
2831 : // Exclude peers that are marked for disconnect, or are going to be
2832 : // disconnected soon (eg ADDR_FETCH and FEELER)
2833 : // Also exclude peers that haven't finished initial connection handshake yet
2834 : // (so that we don't decide we're over our desired connection limit, and then
2835 : // evict some peer that has finished the handshake)
2836 10 : int CConnman::GetExtraFullOutboundCount() const
2837 : {
2838 10 : int full_outbound_peers = 0;
2839 : {
2840 10 : READ_LOCK(m_nodes_mutex);
2841 76 : for (const CNode* pnode : m_nodes) {
2842 : // don't count outbound masternodes
2843 66 : if (pnode->m_masternode_connection) {
2844 0 : continue;
2845 : }
2846 66 : if (pnode->fSuccessfullyConnected && !pnode->fDisconnect && pnode->IsFullOutboundConn() && !pnode->m_masternode_probe_connection) {
2847 54 : ++full_outbound_peers;
2848 54 : }
2849 : }
2850 10 : }
2851 10 : return std::max(full_outbound_peers - m_max_outbound_full_relay, 0);
2852 0 : }
2853 :
2854 10 : int CConnman::GetExtraBlockRelayCount() const
2855 : {
2856 10 : int block_relay_peers = 0;
2857 : {
2858 10 : READ_LOCK(m_nodes_mutex);
2859 76 : for (const CNode* pnode : m_nodes) {
2860 66 : if (pnode->fSuccessfullyConnected && !pnode->fDisconnect && pnode->IsBlockOnlyConn()) {
2861 11 : ++block_relay_peers;
2862 11 : }
2863 : }
2864 10 : }
2865 10 : return std::max(block_relay_peers - m_max_outbound_block_relay, 0);
2866 0 : }
2867 :
2868 0 : std::unordered_set<Network> CConnman::GetReachableEmptyNetworks() const
2869 : {
2870 0 : std::unordered_set<Network> networks{};
2871 0 : for (int n = 0; n < NET_MAX; n++) {
2872 0 : enum Network net = (enum Network)n;
2873 0 : if (net == NET_UNROUTABLE || net == NET_INTERNAL) continue;
2874 0 : if (g_reachable_nets.Contains(net) && addrman.Size(net, std::nullopt) == 0) {
2875 0 : networks.insert(net);
2876 0 : }
2877 0 : }
2878 0 : return networks;
2879 0 : }
2880 :
2881 38 : bool CConnman::MultipleManualOrFullOutboundConns(Network net) const
2882 : {
2883 38 : AssertSharedLockHeld(m_nodes_mutex);
2884 38 : return m_network_conn_counts[net] > 1;
2885 : }
2886 :
2887 0 : bool CConnman::MaybePickPreferredNetwork(std::optional<Network>& network)
2888 : {
2889 0 : std::array<Network, 5> nets{NET_IPV4, NET_IPV6, NET_ONION, NET_I2P, NET_CJDNS};
2890 0 : Shuffle(nets.begin(), nets.end(), FastRandomContext());
2891 :
2892 0 : READ_LOCK(m_nodes_mutex);
2893 0 : for (const auto net : nets) {
2894 0 : if (g_reachable_nets.Contains(net) && m_network_conn_counts[net] == 0 && addrman.Size(net) != 0) {
2895 0 : network = net;
2896 0 : return true;
2897 : }
2898 : }
2899 :
2900 0 : return false;
2901 0 : }
2902 :
2903 0 : void CConnman::ThreadOpenConnections(const std::vector<std::string> connect, CDeterministicMNManager& dmnman)
2904 : {
2905 0 : AssertLockNotHeld(m_unused_i2p_sessions_mutex);
2906 0 : AssertLockNotHeld(m_reconnections_mutex);
2907 0 : FastRandomContext rng;
2908 : // Connect to specific addresses
2909 0 : if (!connect.empty())
2910 : {
2911 : // Attempt v2 connection if we support v2 - we'll reconnect with v1 if our
2912 : // peer doesn't support it or immediately disconnects us for another reason.
2913 0 : const bool use_v2transport(GetLocalServices() & NODE_P2P_V2);
2914 0 : for (int64_t nLoop = 0;; nLoop++)
2915 : {
2916 0 : for (const std::string& strAddr : connect)
2917 : {
2918 0 : CAddress addr(CService(), NODE_NONE);
2919 0 : OpenNetworkConnection(addr, false, {}, strAddr.c_str(), ConnectionType::MANUAL, /*use_v2transport=*/use_v2transport);
2920 0 : for (int i = 0; i < 10 && i < nLoop; i++)
2921 : {
2922 0 : if (!interruptNet.sleep_for(std::chrono::milliseconds(500)))
2923 0 : return;
2924 0 : }
2925 0 : }
2926 0 : if (!interruptNet.sleep_for(std::chrono::milliseconds(500)))
2927 0 : return;
2928 0 : PerformReconnections();
2929 0 : }
2930 : }
2931 :
2932 : // Initiate network connections
2933 0 : auto start = GetTime<std::chrono::microseconds>();
2934 :
2935 : // Minimum time before next feeler connection (in microseconds).
2936 0 : auto next_feeler = GetExponentialRand(start, FEELER_INTERVAL);
2937 0 : auto next_extra_block_relay = GetExponentialRand(start, EXTRA_BLOCK_RELAY_ONLY_PEER_INTERVAL);
2938 0 : auto next_extra_network_peer{GetExponentialRand(start, EXTRA_NETWORK_PEER_INTERVAL)};
2939 0 : const bool dnsseed = gArgs.GetBoolArg("-dnsseed", DEFAULT_DNSSEED);
2940 0 : bool add_fixed_seeds = gArgs.GetBoolArg("-fixedseeds", DEFAULT_FIXEDSEEDS);
2941 0 : const bool use_seednodes{gArgs.IsArgSet("-seednode")};
2942 :
2943 0 : if (!add_fixed_seeds) {
2944 0 : LogPrintf("Fixed seeds are disabled\n");
2945 0 : }
2946 :
2947 0 : while (!interruptNet)
2948 : {
2949 0 : ProcessAddrFetch();
2950 :
2951 0 : if (!interruptNet.sleep_for(std::chrono::milliseconds(500)))
2952 0 : return;
2953 :
2954 0 : PerformReconnections();
2955 :
2956 0 : CSemaphoreGrant grant(*semOutbound);
2957 0 : if (interruptNet)
2958 0 : return;
2959 :
2960 0 : const std::unordered_set<Network> fixed_seed_networks{GetReachableEmptyNetworks()};
2961 0 : if (add_fixed_seeds && !fixed_seed_networks.empty()) {
2962 : // When the node starts with an empty peers.dat, there are a few other sources of peers before
2963 : // we fallback on to fixed seeds: -dnsseed, -seednode, -addnode
2964 : // If none of those are available, we fallback on to fixed seeds immediately, else we allow
2965 : // 60 seconds for any of those sources to populate addrman.
2966 0 : bool add_fixed_seeds_now = false;
2967 : // It is cheapest to check if enough time has passed first.
2968 0 : if (GetTime<std::chrono::seconds>() > start + std::chrono::minutes{1}) {
2969 0 : add_fixed_seeds_now = true;
2970 0 : LogPrintf("Adding fixed seeds as 60 seconds have passed and addrman is empty for at least one reachable network\n");
2971 0 : }
2972 :
2973 : // Perform cheap checks before locking a mutex.
2974 0 : else if (!dnsseed && !use_seednodes) {
2975 0 : LOCK(m_added_nodes_mutex);
2976 0 : if (m_added_node_params.empty()) {
2977 0 : add_fixed_seeds_now = true;
2978 0 : LogPrintf("Adding fixed seeds as -dnsseed=0 (or IPv4/IPv6 connections are disabled via -onlynet) and neither -addnode nor -seednode are provided\n");
2979 0 : }
2980 0 : }
2981 :
2982 0 : if (add_fixed_seeds_now) {
2983 0 : std::vector<CAddress> seed_addrs{ConvertSeeds(Params().FixedSeeds())};
2984 : // We will not make outgoing connections to peers that are unreachable
2985 : // (e.g. because of -onlynet configuration).
2986 : // Therefore, we do not add them to addrman in the first place.
2987 : // In case previously unreachable networks become reachable
2988 : // (e.g. in case of -onlynet changes by the user), fixed seeds will
2989 : // be loaded only for networks for which we have no addresses.
2990 0 : seed_addrs.erase(std::remove_if(seed_addrs.begin(), seed_addrs.end(),
2991 0 : [&fixed_seed_networks](const CAddress& addr) { return fixed_seed_networks.count(addr.GetNetwork()) == 0; }),
2992 0 : seed_addrs.end());
2993 0 : CNetAddr local;
2994 0 : local.SetInternal("fixedseeds");
2995 0 : addrman.Add(seed_addrs, local);
2996 0 : add_fixed_seeds = false;
2997 0 : LogPrintf("Added %d fixed seeds from reachable networks.\n", seed_addrs.size());
2998 0 : }
2999 0 : }
3000 :
3001 : //
3002 : // Choose an address to connect to based on most recently seen
3003 : //
3004 0 : CAddress addrConnect;
3005 :
3006 : // Only connect out to one peer per ipv4/ipv6 network group (/16 for IPv4).
3007 : // This is only done for mainnet and testnet
3008 0 : int nOutboundFullRelay = 0;
3009 0 : int nOutboundBlockRelay = 0;
3010 0 : int nOutboundOnionRelay = 0;
3011 0 : int outbound_privacy_network_peers = 0;
3012 0 : std::set<std::vector<unsigned char>> outbound_ipv46_peer_netgroups;
3013 :
3014 0 : if (!Params().AllowMultipleAddressesFromGroup()) {
3015 0 : READ_LOCK(m_nodes_mutex);
3016 0 : for (const CNode* pnode : m_nodes) {
3017 0 : if (pnode->IsFullOutboundConn() && !pnode->m_masternode_connection) nOutboundFullRelay++;
3018 0 : if (pnode->IsBlockOnlyConn()) nOutboundBlockRelay++;
3019 0 : if (pnode->IsFullOutboundConn() && pnode->ConnectedThroughNetwork() == Network::NET_ONION) nOutboundOnionRelay++;
3020 :
3021 : // Make sure our persistent outbound slots to ipv4/ipv6 peers belong to different netgroups.
3022 0 : switch (pnode->m_conn_type) {
3023 : // We currently don't take inbound connections into account. Since they are
3024 : // free to make, an attacker could make them to prevent us from connecting to
3025 : // certain peers.
3026 : case ConnectionType::INBOUND:
3027 : // Short-lived outbound connections should not affect how we select outbound
3028 : // peers from addrman.
3029 : case ConnectionType::ADDR_FETCH:
3030 : case ConnectionType::FEELER:
3031 0 : break;
3032 : case ConnectionType::MANUAL:
3033 : case ConnectionType::OUTBOUND_FULL_RELAY:
3034 : case ConnectionType::BLOCK_RELAY:
3035 0 : const CAddress address{pnode->addr};
3036 0 : if (address.IsTor() || address.IsI2P() || address.IsCJDNS()) {
3037 : // Since our addrman-groups for these networks are
3038 : // random, without relation to the route we
3039 : // take to connect to these peers or to the
3040 : // difficulty in obtaining addresses with diverse
3041 : // groups, we don't worry about diversity with
3042 : // respect to our addrman groups when connecting to
3043 : // these networks.
3044 0 : ++outbound_privacy_network_peers;
3045 0 : } else {
3046 0 : outbound_ipv46_peer_netgroups.insert(m_netgroupman.GetGroup(address));
3047 : }
3048 0 : } // no default case, so the compiler can warn about missing cases
3049 : }
3050 0 : }
3051 :
3052 0 : std::set<uint256> setConnectedMasternodes;
3053 : {
3054 0 : READ_LOCK(m_nodes_mutex);
3055 0 : for (const CNode* pnode : m_nodes) {
3056 0 : auto verifiedProRegTxHash = pnode->GetVerifiedProRegTxHash();
3057 0 : if (!verifiedProRegTxHash.IsNull()) {
3058 0 : setConnectedMasternodes.emplace(verifiedProRegTxHash);
3059 0 : }
3060 : }
3061 0 : }
3062 :
3063 0 : ConnectionType conn_type = ConnectionType::OUTBOUND_FULL_RELAY;
3064 0 : auto now = GetTime<std::chrono::microseconds>();
3065 0 : bool anchor = false;
3066 0 : bool fFeeler = false;
3067 0 : std::optional<Network> preferred_net;
3068 0 : bool onion_only = false;
3069 :
3070 : // Determine what type of connection to open. Opening
3071 : // BLOCK_RELAY connections to addresses from anchors.dat gets the highest
3072 : // priority. Then we open OUTBOUND_FULL_RELAY priority until we
3073 : // meet our full-relay capacity. Then we open BLOCK_RELAY connection
3074 : // until we hit our block-relay-only peer limit.
3075 : // GetTryNewOutboundPeer() gets set when a stale tip is detected, so we
3076 : // try opening an additional OUTBOUND_FULL_RELAY connection. If none of
3077 : // these conditions are met, check to see if it's time to try an extra
3078 : // block-relay-only peer (to confirm our tip is current, see below) or the next_feeler
3079 : // timer to decide if we should open a FEELER.
3080 :
3081 0 : if (!m_anchors.empty() && (nOutboundBlockRelay < m_max_outbound_block_relay)) {
3082 0 : conn_type = ConnectionType::BLOCK_RELAY;
3083 0 : anchor = true;
3084 0 : } else if (nOutboundFullRelay < m_max_outbound_full_relay) {
3085 : // OUTBOUND_FULL_RELAY
3086 0 : } else if (nOutboundBlockRelay < m_max_outbound_block_relay) {
3087 0 : conn_type = ConnectionType::BLOCK_RELAY;
3088 0 : } else if (GetTryNewOutboundPeer()) {
3089 : // OUTBOUND_FULL_RELAY
3090 0 : } else if (now > next_extra_block_relay && m_start_extra_block_relay_peers) {
3091 : // Periodically connect to a peer (using regular outbound selection
3092 : // methodology from addrman) and stay connected long enough to sync
3093 : // headers, but not much else.
3094 : //
3095 : // Then disconnect the peer, if we haven't learned anything new.
3096 : //
3097 : // The idea is to make eclipse attacks very difficult to pull off,
3098 : // because every few minutes we're finding a new peer to learn headers
3099 : // from.
3100 : //
3101 : // This is similar to the logic for trying extra outbound (full-relay)
3102 : // peers, except:
3103 : // - we do this all the time on an exponential timer, rather than just when
3104 : // our tip is stale
3105 : // - we potentially disconnect our next-youngest block-relay-only peer, if our
3106 : // newest block-relay-only peer delivers a block more recently.
3107 : // See the eviction logic in net_processing.cpp.
3108 : //
3109 : // Because we can promote these connections to block-relay-only
3110 : // connections, they do not get their own ConnectionType enum
3111 : // (similar to how we deal with extra outbound peers).
3112 0 : next_extra_block_relay = GetExponentialRand(now, EXTRA_BLOCK_RELAY_ONLY_PEER_INTERVAL);
3113 0 : conn_type = ConnectionType::BLOCK_RELAY;
3114 0 : } else if (now > next_feeler) {
3115 0 : next_feeler = GetExponentialRand(now, FEELER_INTERVAL);
3116 0 : conn_type = ConnectionType::FEELER;
3117 0 : fFeeler = true;
3118 0 : } else if (nOutboundFullRelay == m_max_outbound_full_relay &&
3119 0 : m_max_outbound_full_relay == MAX_OUTBOUND_FULL_RELAY_CONNECTIONS &&
3120 0 : now > next_extra_network_peer &&
3121 0 : MaybePickPreferredNetwork(preferred_net)) {
3122 : // Full outbound connection management: Attempt to get at least one
3123 : // outbound peer from each reachable network by making extra connections
3124 : // and then protecting "only" peers from a network during outbound eviction.
3125 : // This is not attempted if the user changed -maxconnections to a value
3126 : // so low that less than MAX_OUTBOUND_FULL_RELAY_CONNECTIONS are made,
3127 : // to prevent interactions with otherwise protected outbound peers.
3128 0 : next_extra_network_peer = GetExponentialRand(now, EXTRA_NETWORK_PEER_INTERVAL);
3129 0 : } else if (nOutboundOnionRelay < m_max_outbound_onion && g_reachable_nets.Contains(Network::NET_ONION)) {
3130 0 : onion_only = true;
3131 0 : } else {
3132 : // skip to next iteration of while loop
3133 0 : continue;
3134 : }
3135 :
3136 0 : addrman.ResolveCollisions();
3137 :
3138 0 : auto mnList = dmnman.GetListAtChainTip();
3139 :
3140 0 : const auto current_time{NodeClock::now()};
3141 0 : int nTries = 0;
3142 0 : while (!interruptNet)
3143 : {
3144 0 : if (anchor && !m_anchors.empty()) {
3145 0 : const CAddress addr = m_anchors.back();
3146 0 : m_anchors.pop_back();
3147 0 : if (!addr.IsValid() || IsLocal(addr) || !g_reachable_nets.Contains(addr) ||
3148 0 : !HasAllDesirableServiceFlags(addr.nServices) ||
3149 0 : outbound_ipv46_peer_netgroups.count(m_netgroupman.GetGroup(addr))) continue;
3150 0 : addrConnect = addr;
3151 0 : LogPrint(BCLog::NET, "Trying to make an anchor connection to %s\n", addrConnect.ToStringAddrPort());
3152 0 : break;
3153 0 : }
3154 :
3155 : // If we didn't find an appropriate destination after trying 100 addresses fetched from addrman,
3156 : // stop this loop, and let the outer loop run again (which sleeps, adds seed nodes, recalculates
3157 : // already-connected network ranges, ...) before trying new addrman addresses.
3158 0 : nTries++;
3159 0 : if (nTries > 100)
3160 0 : break;
3161 :
3162 0 : CAddress addr;
3163 0 : NodeSeconds addr_last_try{0s};
3164 :
3165 0 : if (fFeeler) {
3166 : // First, try to get a tried table collision address. This returns
3167 : // an empty (invalid) address if there are no collisions to try.
3168 0 : std::tie(addr, addr_last_try) = addrman.SelectTriedCollision();
3169 :
3170 0 : if (!addr.IsValid()) {
3171 : // No tried table collisions. Select a new table address
3172 : // for our feeler.
3173 0 : std::tie(addr, addr_last_try) = addrman.Select(true);
3174 0 : } else if (AlreadyConnectedToAddress(addr)) {
3175 : // If test-before-evict logic would have us connect to a
3176 : // peer that we're already connected to, just mark that
3177 : // address as Good(). We won't be able to initiate the
3178 : // connection anyway, so this avoids inadvertently evicting
3179 : // a currently-connected peer.
3180 0 : addrman.Good(addr);
3181 : // Select a new table address for our feeler instead.
3182 0 : std::tie(addr, addr_last_try) = addrman.Select(true);
3183 0 : }
3184 0 : } else {
3185 : // Not a feeler
3186 : // If preferred_net has a value set, pick an extra outbound
3187 : // peer from that network. The eviction logic in net_processing
3188 : // ensures that a peer from another network will be evicted.
3189 0 : std::tie(addr, addr_last_try) = addrman.Select(false, preferred_net);
3190 : }
3191 :
3192 : // Require outbound IPv4/IPv6 connections, other than feelers, to be to distinct network groups
3193 0 : if (!fFeeler && outbound_ipv46_peer_netgroups.count(m_netgroupman.GetGroup(addr))) {
3194 0 : continue;
3195 : }
3196 :
3197 : // if we selected an invalid address, restart
3198 0 : if (!addr.IsValid()) {
3199 0 : break;
3200 : }
3201 :
3202 : // don't connect to ourselves
3203 0 : if (addr.GetPort() == GetListenPort() && IsLocal(addr)) {
3204 0 : break;
3205 : }
3206 :
3207 0 : if (!g_reachable_nets.Contains(addr)) {
3208 0 : continue;
3209 : }
3210 :
3211 0 : if (onion_only && !addr.IsTor()) {
3212 0 : continue;
3213 : }
3214 :
3215 : // only consider very recently tried nodes after 30 failed attempts
3216 0 : if (current_time - addr_last_try < 10min && nTries < 30) {
3217 0 : continue;
3218 : }
3219 :
3220 : // for non-feelers, require all the services we'll want,
3221 : // for feelers, only require they be a full node (only because most
3222 : // SPV clients don't have a good address DB available)
3223 0 : if (!fFeeler && !HasAllDesirableServiceFlags(addr.nServices)) {
3224 0 : continue;
3225 0 : } else if (fFeeler && !MayHaveUsefulAddressDB(addr.nServices)) {
3226 0 : continue;
3227 : }
3228 :
3229 : // Do not connect to prohibited ports, unless 50 invalid addresses have been selected already.
3230 0 : if (nTries < 50 && IsBadPort(addr.GetPort())) {
3231 0 : continue;
3232 : }
3233 :
3234 : // Do not make automatic outbound connections to addnode peers, to
3235 : // not use our limited outbound slots for them and to ensure
3236 : // addnode connections benefit from their intended protections.
3237 0 : if (AddedNodesContain(addr)) {
3238 0 : LogPrint(BCLog::NET, "Not making automatic %s%s connection to %s peer selected for manual (addnode) connection%s\n",
3239 : preferred_net.has_value() ? "network-specific " : "",
3240 : ConnectionTypeAsString(conn_type), GetNetworkName(addr.GetNetwork()),
3241 : fLogIPs ? strprintf(": %s", addr.ToStringAddrPort()) : "");
3242 0 : continue;
3243 : }
3244 :
3245 : // don't try to connect to masternodes that we already have a connection to (most likely inbound)
3246 0 : if (auto dmn = mnList.GetMNByService(addr); dmn && setConnectedMasternodes.count(dmn->proTxHash)) {
3247 0 : continue;
3248 : }
3249 :
3250 0 : addrConnect = addr;
3251 0 : break;
3252 0 : }
3253 :
3254 0 : if (addrConnect.IsValid()) {
3255 0 : if (fFeeler) {
3256 : // Add small amount of random noise before connection to avoid synchronization.
3257 0 : if (!interruptNet.sleep_for(rng.rand_uniform_duration<CThreadInterrupt::Clock>(FEELER_SLEEP_WINDOW))) {
3258 0 : return;
3259 : }
3260 0 : if (fLogIPs) {
3261 0 : LogPrint(BCLog::NET, "Making feeler connection to %s\n", addrConnect.ToStringAddrPort());
3262 0 : } else {
3263 0 : LogPrint(BCLog::NET, "Making feeler connection\n");
3264 : }
3265 0 : }
3266 :
3267 0 : if (preferred_net != std::nullopt) LogPrint(BCLog::NET, "Making network specific connection to %s on %s.\n", addrConnect.ToStringAddrPort(), GetNetworkName(preferred_net.value()));
3268 :
3269 : // Record addrman failure attempts when node has at least 2 persistent outbound connections to peers with
3270 : // different netgroups in ipv4/ipv6 networks + all peers in Tor/I2P/CJDNS networks.
3271 : // Don't record addrman failure attempts when node is offline. This can be identified since all local
3272 : // network connections (if any) belong in the same netgroup, and the size of `outbound_ipv46_peer_netgroups` would only be 1.
3273 0 : const bool count_failures{((int)outbound_ipv46_peer_netgroups.size() + outbound_privacy_network_peers) >= std::min(nMaxConnections - 1, 2)};
3274 : // Use BIP324 transport when both us and them have NODE_V2_P2P set.
3275 0 : const bool use_v2transport(addrConnect.nServices & GetLocalServices() & NODE_P2P_V2);
3276 0 : OpenNetworkConnection(addrConnect, count_failures, std::move(grant), /*strDest=*/nullptr, conn_type, use_v2transport);
3277 0 : }
3278 0 : }
3279 0 : }
3280 :
3281 0 : std::vector<CAddress> CConnman::GetCurrentBlockRelayOnlyConns() const
3282 : {
3283 0 : std::vector<CAddress> ret;
3284 0 : READ_LOCK(m_nodes_mutex);
3285 0 : for (const CNode* pnode : m_nodes) {
3286 0 : if (pnode->IsBlockRelayOnly()) {
3287 0 : ret.push_back(pnode->addr);
3288 0 : }
3289 : }
3290 :
3291 0 : return ret;
3292 0 : }
3293 :
3294 3 : std::vector<AddedNodeInfo> CConnman::GetAddedNodeInfo(bool include_connected) const
3295 : {
3296 3 : std::vector<AddedNodeInfo> ret;
3297 :
3298 3 : std::list<AddedNodeParams> lAddresses(0);
3299 : {
3300 3 : LOCK(m_added_nodes_mutex);
3301 3 : ret.reserve(m_added_node_params.size());
3302 3 : std::copy(m_added_node_params.cbegin(), m_added_node_params.cend(), std::back_inserter(lAddresses));
3303 3 : }
3304 :
3305 :
3306 : // Build a map of all already connected addresses (by IP:port and by name) to inbound/outbound and resolved CService
3307 3 : std::map<CService, bool> mapConnected;
3308 3 : std::map<std::string, std::pair<bool, CService>> mapConnectedByName;
3309 : {
3310 3 : READ_LOCK(m_nodes_mutex);
3311 19 : for (const CNode* pnode : m_nodes) {
3312 16 : if (pnode->addr.IsValid()) {
3313 16 : mapConnected[pnode->addr] = pnode->IsInboundConn();
3314 16 : }
3315 16 : std::string addrName{pnode->m_addr_name};
3316 16 : if (!addrName.empty()) {
3317 16 : mapConnectedByName[std::move(addrName)] = std::make_pair(pnode->IsInboundConn(), static_cast<const CService&>(pnode->addr));
3318 16 : }
3319 16 : }
3320 3 : }
3321 :
3322 18 : for (const auto& addr : lAddresses) {
3323 15 : CService service{MaybeFlipIPv6toCJDNS(LookupNumeric(addr.m_added_node, Params().GetDefaultPort(addr.m_added_node)))};
3324 15 : AddedNodeInfo addedNode{addr, CService(), false, false};
3325 15 : if (service.IsValid()) {
3326 : // strAddNode is an IP:port
3327 15 : auto it = mapConnected.find(service);
3328 15 : if (it != mapConnected.end()) {
3329 15 : if (!include_connected) {
3330 5 : continue;
3331 : }
3332 10 : addedNode.resolvedAddress = service;
3333 10 : addedNode.fConnected = true;
3334 10 : addedNode.fInbound = it->second;
3335 10 : }
3336 10 : } else {
3337 : // strAddNode is a name
3338 0 : auto it = mapConnectedByName.find(addr.m_added_node);
3339 0 : if (it != mapConnectedByName.end()) {
3340 0 : if (!include_connected) {
3341 0 : continue;
3342 : }
3343 0 : addedNode.resolvedAddress = it->second.second;
3344 0 : addedNode.fConnected = true;
3345 0 : addedNode.fInbound = it->second.first;
3346 0 : }
3347 : }
3348 10 : ret.emplace_back(std::move(addedNode));
3349 15 : }
3350 :
3351 3 : return ret;
3352 3 : }
3353 :
3354 0 : void CConnman::ThreadOpenAddedConnections()
3355 : {
3356 0 : AssertLockNotHeld(m_unused_i2p_sessions_mutex);
3357 0 : AssertLockNotHeld(m_reconnections_mutex);
3358 0 : while (true)
3359 : {
3360 0 : CSemaphoreGrant grant(*semAddnode);
3361 0 : std::vector<AddedNodeInfo> vInfo = GetAddedNodeInfo(/*include_connected=*/false);
3362 0 : bool tried = false;
3363 0 : for (const AddedNodeInfo& info : vInfo) {
3364 0 : if (!grant) {
3365 : // If we've used up our semaphore and need a new one, let's not wait here since while we are waiting
3366 : // the addednodeinfo state might change.
3367 0 : break;
3368 : }
3369 0 : tried = true;
3370 0 : CAddress addr(CService(), NODE_NONE);
3371 0 : OpenNetworkConnection(addr, false, std::move(grant), info.m_params.m_added_node.c_str(), ConnectionType::MANUAL, info.m_params.m_use_v2transport);
3372 0 : if (!interruptNet.sleep_for(std::chrono::milliseconds(500))) return;
3373 0 : grant = CSemaphoreGrant(*semAddnode, /*fTry=*/true);
3374 0 : }
3375 : // See if any reconnections are desired.
3376 0 : PerformReconnections();
3377 : // Retry every 60 seconds if a connection was attempted, otherwise two seconds
3378 0 : if (!interruptNet.sleep_for(std::chrono::seconds(tried ? 60 : 2)))
3379 0 : return;
3380 0 : }
3381 0 : }
3382 :
3383 0 : void CConnman::ThreadOpenMasternodeConnections(CDeterministicMNManager& dmnman, CMasternodeMetaMan& mn_metaman,
3384 : CMasternodeSync& mn_sync)
3385 : {
3386 : // Connecting to specific addresses, no masternode connections available
3387 0 : if (gArgs.IsArgSet("-connect") && gArgs.GetArgs("-connect").size() > 0)
3388 0 : return;
3389 :
3390 0 : auto& chainParams = Params();
3391 :
3392 0 : bool didConnect = false;
3393 0 : while (!interruptNet)
3394 : {
3395 0 : auto sleepTime = std::chrono::milliseconds(1000);
3396 0 : if (didConnect) {
3397 0 : sleepTime = std::chrono::milliseconds(100);
3398 0 : }
3399 0 : if (!interruptNet.sleep_for(sleepTime))
3400 0 : return;
3401 :
3402 0 : didConnect = false;
3403 :
3404 0 : if (!fNetworkActive || !m_masternode_thread_active || !mn_sync.IsBlockchainSynced()) continue;
3405 :
3406 0 : std::unordered_set<CService, CServiceHash> connectedNodes;
3407 0 : Uint256HashMap</*fInbound=*/bool> connectedProRegTxHashes;
3408 0 : ForEachNode([&](const CNode* pnode) {
3409 0 : connectedNodes.emplace(pnode->addr);
3410 0 : if (auto verifiedProRegTxHash = pnode->GetVerifiedProRegTxHash(); !verifiedProRegTxHash.IsNull()) {
3411 0 : connectedProRegTxHashes.emplace(verifiedProRegTxHash, pnode->IsInboundConn());
3412 0 : }
3413 0 : });
3414 :
3415 0 : auto mnList = dmnman.GetListAtChainTip();
3416 :
3417 0 : if (interruptNet)
3418 0 : return;
3419 :
3420 0 : int64_t nANow = GetTime<std::chrono::seconds>().count();
3421 0 : constexpr const auto &_func_ = __func__;
3422 :
3423 : // NOTE: Process only one pending masternode at a time
3424 :
3425 0 : MasternodeProbeConn isProbe = MasternodeProbeConn::IsNotConnection;
3426 :
3427 0 : const auto getPendingQuorumNodes = [&]() SHARED_LOCKS_REQUIRED(m_nodes_mutex) EXCLUSIVE_LOCKS_REQUIRED(cs_vPendingMasternodes) {
3428 0 : AssertSharedLockHeld(m_nodes_mutex);
3429 0 : AssertLockHeld(cs_vPendingMasternodes);
3430 0 : std::vector<CDeterministicMNCPtr> ret;
3431 0 : for (const auto& group : masternodeQuorumNodes) {
3432 0 : for (const auto& proRegTxHash : group.second) {
3433 0 : if (connectedProRegTxHashes.count(proRegTxHash)) {
3434 0 : continue;
3435 : }
3436 0 : auto dmn = mnList.GetMN(proRegTxHash);
3437 0 : if (!dmn) {
3438 0 : continue;
3439 : }
3440 0 : const auto addr2 = dmn->pdmnState->netInfo->GetPrimary();
3441 0 : CNode* pnode = FindNodeMutable(addr2, /*fExcludeDisconnecting=*/false);
3442 0 : if (pnode && (pnode->m_masternode_connection || pnode->fDisconnect)) {
3443 : // node is either a masternode or disconnecting, skip it
3444 0 : continue;
3445 : }
3446 0 : if (connectedNodes.count(addr2)) {
3447 : // we probably connected to it before it became a masternode
3448 : // or maybe we are still waiting for mnauth
3449 0 : bool slow_handshake = pnode && pnode->nTimeFirstMessageReceived.load() != 0s &&
3450 0 : GetTime<std::chrono::seconds>() - pnode->nTimeFirstMessageReceived.load() > 5s;
3451 0 : if (slow_handshake) {
3452 : // clearly not expecting mnauth to take that long even if it wasn't the first message
3453 : // we received (as it should normally), disconnect
3454 0 : LogPrint(BCLog::NET_NETCONN, "CConnman::%s -- dropping non-mnauth connection to %s, service=%s\n",
3455 : _func_, proRegTxHash.ToString(), addr2.ToStringAddrPort());
3456 0 : pnode->fDisconnect = true;
3457 0 : }
3458 : // either way - it's not ready, skip it for now
3459 0 : continue;
3460 : }
3461 : // back off connecting to an address if we already tried recently
3462 0 : int64_t last_attempt = mn_metaman.GetLastOutboundAttempt(dmn->proTxHash);
3463 0 : if (nANow - last_attempt < chainParams.LLMQConnectionRetryTimeout()) {
3464 0 : continue;
3465 : }
3466 : // all checks passed
3467 0 : ret.emplace_back(dmn);
3468 0 : }
3469 : }
3470 0 : return ret;
3471 0 : };
3472 :
3473 0 : const auto getPendingProbes = [&]() EXCLUSIVE_LOCKS_REQUIRED(cs_vPendingMasternodes) {
3474 0 : AssertLockHeld(cs_vPendingMasternodes);
3475 0 : std::vector<CDeterministicMNCPtr> ret;
3476 0 : for (auto it = masternodePendingProbes.begin(); it != masternodePendingProbes.end(); ) {
3477 0 : auto dmn = mnList.GetMN(*it);
3478 0 : if (!dmn) {
3479 0 : it = masternodePendingProbes.erase(it);
3480 0 : continue;
3481 : }
3482 0 : bool connectedAndOutbound = connectedProRegTxHashes.count(dmn->proTxHash) && !connectedProRegTxHashes[dmn->proTxHash];
3483 0 : if (connectedAndOutbound) {
3484 : // we already have an outbound connection to this MN so there is no theed to probe it again
3485 0 : mn_metaman.SetLastOutboundSuccess(dmn->proTxHash, nANow);
3486 0 : it = masternodePendingProbes.erase(it);
3487 0 : continue;
3488 : }
3489 :
3490 0 : ++it;
3491 :
3492 0 : int64_t lastAttempt = mn_metaman.GetLastOutboundAttempt(dmn->proTxHash);
3493 : // back off trying connecting to an address if we already tried recently
3494 0 : if (nANow - lastAttempt < chainParams.LLMQConnectionRetryTimeout()) {
3495 0 : continue;
3496 : }
3497 0 : ret.emplace_back(dmn);
3498 0 : }
3499 0 : return ret;
3500 0 : };
3501 :
3502 0 : auto getConnectToDmn = [&]() -> CDeterministicMNCPtr {
3503 : // don't hold lock while calling OpenMasternodeConnection as cs_main is locked deep inside
3504 0 : READ_LOCK(m_nodes_mutex);
3505 0 : LOCK(cs_vPendingMasternodes);
3506 :
3507 0 : if (!vPendingMasternodes.empty()) {
3508 0 : auto dmn = mnList.GetValidMN(vPendingMasternodes.front());
3509 0 : vPendingMasternodes.erase(vPendingMasternodes.begin());
3510 : // Check if we should connect to this masternode
3511 : // We already hold m_nodes_mutex here, so check m_masternode_connection directly
3512 0 : if (dmn && !connectedNodes.count(dmn->pdmnState->netInfo->GetPrimary())) {
3513 0 : const CNode* pnode = FindNode(dmn->pdmnState->netInfo->GetPrimary(), /*fExcludeDisconnecting=*/false);
3514 0 : if (pnode == nullptr || (!pnode->m_masternode_connection && !pnode->fDisconnect)) {
3515 0 : LogPrint(BCLog::NET_NETCONN, "CConnman::%s -- opening pending masternode connection to %s, service=%s\n",
3516 : _func_, dmn->proTxHash.ToString(),
3517 : dmn->pdmnState->netInfo->GetPrimary().ToStringAddrPort());
3518 0 : return dmn;
3519 : }
3520 0 : }
3521 0 : }
3522 :
3523 0 : if (const auto pending = getPendingQuorumNodes(); !pending.empty()) {
3524 : // not-null
3525 0 : auto dmn = pending[GetRand(pending.size())];
3526 0 : LogPrint(BCLog::NET_NETCONN, "CConnman::%s -- opening quorum connection to %s, service=%s\n",
3527 : _func_, dmn->proTxHash.ToString(), dmn->pdmnState->netInfo->GetPrimary().ToStringAddrPort());
3528 0 : return dmn;
3529 0 : }
3530 :
3531 0 : if (const auto pending = getPendingProbes(); !pending.empty()) {
3532 : // not-null
3533 0 : auto dmn = pending[GetRand(pending.size())];
3534 0 : masternodePendingProbes.erase(dmn->proTxHash);
3535 0 : isProbe = MasternodeProbeConn::IsConnection;
3536 :
3537 0 : LogPrint(BCLog::NET_NETCONN, "CConnman::%s -- probing masternode %s, service=%s\n", _func_, dmn->proTxHash.ToString(), dmn->pdmnState->netInfo->GetPrimary().ToStringAddrPort());
3538 0 : return dmn;
3539 0 : }
3540 0 : return nullptr;
3541 0 : };
3542 :
3543 0 : CDeterministicMNCPtr connectToDmn = getConnectToDmn();
3544 :
3545 0 : if (connectToDmn == nullptr) {
3546 0 : continue;
3547 : }
3548 :
3549 0 : didConnect = true;
3550 :
3551 0 : mn_metaman.SetLastOutboundAttempt(connectToDmn->proTxHash, nANow);
3552 :
3553 0 : OpenMasternodeConnection(CAddress(connectToDmn->pdmnState->netInfo->GetPrimary(), NODE_NETWORK), /*use_v2transport=*/GetLocalServices() & NODE_P2P_V2, isProbe);
3554 : // should be in the list now if connection was opened
3555 0 : bool connected = ForNode(connectToDmn->pdmnState->netInfo->GetPrimary(), CConnman::AllNodes, [&](const CNode* pnode) {
3556 0 : if (pnode->fDisconnect) {
3557 0 : return false;
3558 : }
3559 0 : return true;
3560 0 : });
3561 0 : if (!connected) {
3562 0 : LogPrint(BCLog::NET_NETCONN, "CConnman::%s -- connection failed for masternode %s, service=%s\n", __func__, connectToDmn->proTxHash.ToString(), connectToDmn->pdmnState->netInfo->GetPrimary().ToStringAddrPort());
3563 : // Will take a few consequent failed attempts to PoSe-punish a MN.
3564 0 : if (mn_metaman.OutboundFailedTooManyTimes(connectToDmn->proTxHash)) {
3565 0 : LogPrint(BCLog::NET_NETCONN, "CConnman::%s -- failed to connect to masternode %s too many times\n", __func__, connectToDmn->proTxHash.ToString());
3566 0 : }
3567 0 : }
3568 0 : }
3569 0 : }
3570 :
3571 : // if successful, this moves the passed grant to the constructed node
3572 0 : void CConnman::OpenNetworkConnection(const CAddress& addrConnect, bool fCountFailure, CSemaphoreGrant&& grant_outbound,
3573 : const char *pszDest, ConnectionType conn_type, bool use_v2transport,
3574 : MasternodeConn masternode_connection, MasternodeProbeConn masternode_probe_connection)
3575 : {
3576 0 : AssertLockNotHeld(m_unused_i2p_sessions_mutex);
3577 0 : assert(conn_type != ConnectionType::INBOUND);
3578 :
3579 : //
3580 : // Initiate outbound network connection
3581 : //
3582 0 : if (interruptNet) {
3583 0 : return;
3584 : }
3585 0 : if (!fNetworkActive) {
3586 0 : return;
3587 : }
3588 :
3589 0 : auto getIpStr = [&]() {
3590 0 : if (fLogIPs) {
3591 0 : return addrConnect.ToStringAddrPort();
3592 : } else {
3593 0 : return std::string("new peer");
3594 : }
3595 0 : };
3596 :
3597 0 : if (!pszDest) {
3598 : // banned, discouraged or exact match?
3599 0 : if ((m_banman && (m_banman->IsDiscouraged(addrConnect) || m_banman->IsBanned(addrConnect))) || AlreadyConnectedToAddress(addrConnect))
3600 0 : return;
3601 : // connecting to ourselves?
3602 0 : if (addrConnect.GetPort() == GetListenPort() && IsLocal(addrConnect)) {
3603 0 : return;
3604 : }
3605 0 : } else if (ExistsNode(std::string(pszDest))) {
3606 0 : return;
3607 : }
3608 :
3609 0 : LogPrint(BCLog::NET_NETCONN, "CConnman::%s -- connecting to %s\n", __func__, getIpStr());
3610 0 : CNode* pnode = ConnectNode(addrConnect, pszDest, fCountFailure, conn_type, use_v2transport);
3611 :
3612 0 : if (!pnode) {
3613 0 : LogPrint(BCLog::NET_NETCONN, "CConnman::%s -- ConnectNode failed for %s\n", __func__, getIpStr());
3614 0 : return;
3615 : }
3616 :
3617 : {
3618 0 : LOCK(pnode->m_sock_mutex);
3619 0 : LogPrint(BCLog::NET_NETCONN, "CConnman::%s -- successfully connected to %s, sock=%d, peer=%d\n", __func__, getIpStr(), pnode->m_sock->Get(), pnode->GetId());
3620 0 : }
3621 :
3622 0 : pnode->grantOutbound = std::move(grant_outbound);
3623 :
3624 0 : if (masternode_connection == MasternodeConn::IsConnection)
3625 0 : pnode->m_masternode_connection = true;
3626 0 : if (masternode_probe_connection == MasternodeProbeConn::IsConnection)
3627 0 : pnode->m_masternode_probe_connection = true;
3628 :
3629 : {
3630 0 : LOCK2(cs_mapSocketToNode, pnode->m_sock_mutex);
3631 0 : mapSocketToNode.emplace(pnode->m_sock->Get(), pnode);
3632 0 : }
3633 :
3634 0 : m_msgproc->InitializeNode(*pnode, nLocalServices);
3635 : {
3636 0 : LOCK(m_nodes_mutex);
3637 0 : m_nodes.push_back(pnode);
3638 :
3639 : // update connection count by network
3640 0 : if (pnode->IsManualOrFullOutboundConn()) ++m_network_conn_counts[pnode->addr.GetNetwork()];
3641 0 : }
3642 : {
3643 0 : if (m_edge_trig_events) {
3644 0 : LOCK(pnode->m_sock_mutex);
3645 0 : if (!m_edge_trig_events->RegisterEvents(pnode->m_sock->Get())) {
3646 0 : LogPrint(BCLog::NET, "EdgeTriggeredEvents::RegisterEvents() failed\n");
3647 0 : }
3648 0 : }
3649 0 : if (m_wakeup_pipe) {
3650 0 : m_wakeup_pipe->Write();
3651 0 : }
3652 : }
3653 0 : }
3654 :
3655 0 : void CConnman::OpenMasternodeConnection(const CAddress &addrConnect, bool use_v2transport, MasternodeProbeConn probe) {
3656 0 : OpenNetworkConnection(addrConnect, false, {}, /*strDest=*/nullptr, ConnectionType::OUTBOUND_FULL_RELAY,
3657 0 : use_v2transport, MasternodeConn::IsConnection, probe);
3658 0 : }
3659 :
3660 : Mutex NetEventsInterface::g_msgproc_mutex;
3661 :
3662 0 : void CConnman::ThreadMessageHandler()
3663 : {
3664 0 : LOCK(NetEventsInterface::g_msgproc_mutex);
3665 :
3666 0 : auto nLastSendMessagesTimeMasternodes = SteadyClock::time_point{};
3667 :
3668 0 : FastRandomContext rng;
3669 0 : while (!flagInterruptMsgProc)
3670 : {
3671 0 : bool fMoreWork = false;
3672 :
3673 0 : bool fSkipSendMessagesForMasternodes = true;
3674 0 : if (SteadyClock::now() - nLastSendMessagesTimeMasternodes >= 100ms) {
3675 0 : fSkipSendMessagesForMasternodes = false;
3676 0 : nLastSendMessagesTimeMasternodes = SteadyClock::now();
3677 0 : }
3678 :
3679 : // Randomize the order in which we process messages from/to our peers.
3680 : // This prevents attacks in which an attacker exploits having multiple
3681 : // consecutive connections in the m_nodes list.
3682 0 : const NodesSnapshot snap{*this, /* cond = */ CConnman::AllNodes, /* shuffle = */ true};
3683 :
3684 0 : for (CNode* pnode : snap.Nodes()) {
3685 0 : if (pnode->fDisconnect)
3686 0 : continue;
3687 :
3688 : // Receive messages
3689 0 : bool fMoreNodeWork = m_msgproc->ProcessMessages(pnode, flagInterruptMsgProc);
3690 0 : fMoreWork |= (fMoreNodeWork && !pnode->fPauseSend);
3691 0 : if (flagInterruptMsgProc)
3692 0 : return;
3693 : // Send messages
3694 0 : if (!fSkipSendMessagesForMasternodes || !pnode->m_masternode_connection) {
3695 0 : m_msgproc->SendMessages(pnode);
3696 0 : }
3697 :
3698 0 : if (flagInterruptMsgProc)
3699 0 : return;
3700 : }
3701 :
3702 0 : WAIT_LOCK(mutexMsgProc, lock);
3703 0 : if (!fMoreWork) {
3704 0 : condMsgProc.wait_until(lock, std::chrono::steady_clock::now() + std::chrono::milliseconds(100), [this]() EXCLUSIVE_LOCKS_REQUIRED(mutexMsgProc) { return fMsgProcWake; });
3705 0 : }
3706 0 : fMsgProcWake = false;
3707 0 : }
3708 0 : }
3709 :
3710 0 : void CConnman::ThreadI2PAcceptIncoming(CMasternodeSync& mn_sync)
3711 : {
3712 : static constexpr auto err_wait_begin = 1s;
3713 : static constexpr auto err_wait_cap = 5min;
3714 0 : auto err_wait = err_wait_begin;
3715 :
3716 0 : bool advertising_listen_addr = false;
3717 0 : i2p::Connection conn;
3718 :
3719 0 : auto SleepOnFailure = [&]() {
3720 0 : interruptNet.sleep_for(err_wait);
3721 0 : if (err_wait < err_wait_cap) {
3722 0 : err_wait += 1s;
3723 0 : }
3724 0 : };
3725 :
3726 0 : while (!interruptNet) {
3727 :
3728 0 : if (!m_i2p_sam_session->Listen(conn)) {
3729 0 : if (advertising_listen_addr && conn.me.IsValid()) {
3730 0 : RemoveLocal(conn.me);
3731 0 : advertising_listen_addr = false;
3732 0 : }
3733 0 : SleepOnFailure();
3734 0 : continue;
3735 : }
3736 :
3737 0 : if (!advertising_listen_addr) {
3738 0 : AddLocal(conn.me, LOCAL_MANUAL);
3739 0 : advertising_listen_addr = true;
3740 0 : }
3741 :
3742 0 : if (!m_i2p_sam_session->Accept(conn)) {
3743 0 : SleepOnFailure();
3744 0 : continue;
3745 : }
3746 :
3747 0 : CreateNodeFromAcceptedSocket(std::move(conn.sock), NetPermissionFlags::None,
3748 0 : CAddress{conn.me, NODE_NONE}, CAddress{conn.peer, NODE_NONE}, mn_sync);
3749 :
3750 0 : err_wait = err_wait_begin;
3751 : }
3752 0 : }
3753 :
3754 0 : bool CConnman::BindListenPort(const CService& addrBind, bilingual_str& strError, NetPermissionFlags permissions)
3755 : {
3756 0 : int nOne = 1;
3757 :
3758 : // Create socket for listening for incoming connections
3759 : struct sockaddr_storage sockaddr;
3760 0 : socklen_t len = sizeof(sockaddr);
3761 0 : if (!addrBind.GetSockAddr((struct sockaddr*)&sockaddr, &len))
3762 : {
3763 0 : strError = strprintf(Untranslated("Error: Bind address family for %s not supported"), addrBind.ToStringAddrPort());
3764 0 : LogPrintLevel(BCLog::NET, BCLog::Level::Error, "%s\n", strError.original);
3765 0 : return false;
3766 : }
3767 :
3768 0 : std::unique_ptr<Sock> sock = CreateSock(addrBind.GetSAFamily());
3769 0 : if (!sock) {
3770 0 : strError = strprintf(Untranslated("Couldn't open socket for incoming connections (socket returned error %s)"), NetworkErrorString(WSAGetLastError()));
3771 0 : LogPrintLevel(BCLog::NET, BCLog::Level::Error, "%s\n", strError.original);
3772 0 : return false;
3773 : }
3774 :
3775 : // Allow binding if the port is still in TIME_WAIT state after
3776 : // the program was closed and restarted.
3777 0 : if (sock->SetSockOpt(SOL_SOCKET, SO_REUSEADDR, (sockopt_arg_type)&nOne, sizeof(int)) == SOCKET_ERROR) {
3778 0 : strError = strprintf(Untranslated("Error setting SO_REUSEADDR on socket: %s, continuing anyway"), NetworkErrorString(WSAGetLastError()));
3779 0 : LogPrintf("%s\n", strError.original);
3780 0 : }
3781 :
3782 : // some systems don't have IPV6_V6ONLY but are always v6only; others do have the option
3783 : // and enable it by default or not. Try to enable it, if possible.
3784 0 : if (addrBind.IsIPv6()) {
3785 : #ifdef IPV6_V6ONLY
3786 0 : if (sock->SetSockOpt(IPPROTO_IPV6, IPV6_V6ONLY, (sockopt_arg_type)&nOne, sizeof(int)) == SOCKET_ERROR) {
3787 0 : strError = strprintf(Untranslated("Error setting IPV6_V6ONLY on socket: %s, continuing anyway"), NetworkErrorString(WSAGetLastError()));
3788 0 : LogPrintf("%s\n", strError.original);
3789 0 : }
3790 : #endif
3791 : #ifdef WIN32
3792 : int nProtLevel = PROTECTION_LEVEL_UNRESTRICTED;
3793 : if (sock->SetSockOpt(IPPROTO_IPV6, IPV6_PROTECTION_LEVEL, (const char*)&nProtLevel, sizeof(int)) == SOCKET_ERROR) {
3794 : strError = strprintf(Untranslated("Error setting IPV6_PROTECTION_LEVEL on socket: %s, continuing anyway"), NetworkErrorString(WSAGetLastError()));
3795 : LogPrintf("%s\n", strError.original);
3796 : }
3797 : #endif
3798 0 : }
3799 :
3800 0 : if (sock->Bind(reinterpret_cast<struct sockaddr*>(&sockaddr), len) == SOCKET_ERROR) {
3801 0 : int nErr = WSAGetLastError();
3802 0 : if (nErr == WSAEADDRINUSE)
3803 0 : strError = strprintf(_("Unable to bind to %s on this computer. %s is probably already running."), addrBind.ToStringAddrPort(), PACKAGE_NAME);
3804 : else
3805 0 : strError = strprintf(_("Unable to bind to %s on this computer (bind returned error %s)"), addrBind.ToStringAddrPort(), NetworkErrorString(nErr));
3806 0 : LogPrintLevel(BCLog::NET, BCLog::Level::Error, "%s\n", strError.original);
3807 0 : return false;
3808 : }
3809 0 : LogPrintf("Bound to %s\n", addrBind.ToStringAddrPort());
3810 :
3811 : // Listen for incoming connections
3812 0 : if (sock->Listen(SOMAXCONN) == SOCKET_ERROR)
3813 : {
3814 0 : strError = strprintf(_("Listening for incoming connections failed (listen returned error %s)"), NetworkErrorString(WSAGetLastError()));
3815 0 : LogPrintLevel(BCLog::NET, BCLog::Level::Error, "%s\n", strError.original);
3816 0 : return false;
3817 : }
3818 :
3819 0 : if (m_edge_trig_events && !m_edge_trig_events->AddSocket(sock->Get())) {
3820 0 : LogPrintf("Error: EdgeTriggeredEvents::AddSocket() failed\n");
3821 0 : return false;
3822 : }
3823 :
3824 0 : vhListenSocket.emplace_back(std::move(sock), permissions);
3825 :
3826 0 : return true;
3827 0 : }
3828 :
3829 0 : void Discover()
3830 : {
3831 0 : if (!fDiscover)
3832 0 : return;
3833 :
3834 : #ifdef WIN32
3835 : // Get local host IP
3836 : char pszHostName[256] = "";
3837 : if (gethostname(pszHostName, sizeof(pszHostName)) != SOCKET_ERROR)
3838 : {
3839 : const std::vector<CNetAddr> addresses{LookupHost(pszHostName, 0, true)};
3840 : for (const CNetAddr& addr : addresses)
3841 : {
3842 : if (AddLocal(addr, LOCAL_IF))
3843 : LogPrintf("%s: %s - %s\n", __func__, pszHostName, addr.ToStringAddr());
3844 : }
3845 : }
3846 : #elif (HAVE_DECL_GETIFADDRS && HAVE_DECL_FREEIFADDRS)
3847 : // Get local host ip
3848 : struct ifaddrs* myaddrs;
3849 0 : if (getifaddrs(&myaddrs) == 0)
3850 : {
3851 0 : for (struct ifaddrs* ifa = myaddrs; ifa != nullptr; ifa = ifa->ifa_next)
3852 : {
3853 0 : if (ifa->ifa_addr == nullptr) continue;
3854 0 : if ((ifa->ifa_flags & IFF_UP) == 0) continue;
3855 0 : if ((ifa->ifa_flags & IFF_LOOPBACK) != 0) continue;
3856 0 : if (ifa->ifa_addr->sa_family == AF_INET)
3857 : {
3858 0 : struct sockaddr_in* s4 = (struct sockaddr_in*)(ifa->ifa_addr);
3859 0 : CNetAddr addr(s4->sin_addr);
3860 0 : if (AddLocal(addr, LOCAL_IF))
3861 0 : LogPrintf("%s: IPv4 %s: %s\n", __func__, ifa->ifa_name, addr.ToStringAddr());
3862 0 : }
3863 0 : else if (ifa->ifa_addr->sa_family == AF_INET6)
3864 : {
3865 0 : struct sockaddr_in6* s6 = (struct sockaddr_in6*)(ifa->ifa_addr);
3866 0 : CNetAddr addr(s6->sin6_addr);
3867 0 : if (AddLocal(addr, LOCAL_IF))
3868 0 : LogPrintf("%s: IPv6 %s: %s\n", __func__, ifa->ifa_name, addr.ToStringAddr());
3869 0 : }
3870 0 : }
3871 0 : freeifaddrs(myaddrs);
3872 0 : }
3873 : #endif
3874 0 : }
3875 :
3876 634 : void CConnman::SetNetworkActive(bool active, CMasternodeSync* const mn_sync)
3877 : {
3878 634 : LogPrintf("%s: %s\n", __func__, active);
3879 :
3880 634 : if (fNetworkActive == active) {
3881 632 : return;
3882 : }
3883 :
3884 2 : fNetworkActive = active;
3885 :
3886 : // mn_sync is nullptr during app initialization with `-networkactive=`
3887 2 : if (mn_sync) {
3888 : // Always call the Reset() if the network gets enabled/disabled to make sure the sync process
3889 : // gets a reset if its outdated..
3890 2 : mn_sync->Reset();
3891 2 : }
3892 :
3893 2 : if (m_client_interface) {
3894 0 : m_client_interface->NotifyNetworkActiveChanged(fNetworkActive);
3895 0 : }
3896 634 : }
3897 :
3898 633 : CConnman::CConnman(uint64_t nSeed0In, uint64_t nSeed1In, AddrMan& addrman_in,
3899 : const NetGroupManager& netgroupman, bool network_active)
3900 : : addrman(addrman_in)
3901 : , m_netgroupman{netgroupman}
3902 : , nSeed0(nSeed0In)
3903 : , nSeed1(nSeed1In)
3904 1 : {
3905 : // Make sure we never set the default port to a bad port
3906 : for (int n = 0; n < NET_MAX; ++n) {
3907 : const bool is_bad_port = IsBadPort(Params().GetDefaultPort(static_cast<Network>(n)));
3908 : assert(!is_bad_port);
3909 : }
3910 :
3911 : SetTryNewOutboundPeer(false);
3912 :
3913 : Options connOptions;
3914 : Init(connOptions);
3915 : SetNetworkActive(network_active, /* mn_sync = */ nullptr);
3916 1 : }
3917 :
3918 0 : NodeId CConnman::GetNewNodeId()
3919 : {
3920 0 : return nLastNodeId.fetch_add(1, std::memory_order_relaxed);
3921 : }
3922 :
3923 :
3924 0 : bool CConnman::Bind(const CService& addr_, unsigned int flags, NetPermissionFlags permissions)
3925 : {
3926 0 : const CService addr{MaybeFlipIPv6toCJDNS(addr_)};
3927 :
3928 0 : bilingual_str strError;
3929 0 : if (!BindListenPort(addr, strError, permissions)) {
3930 0 : if ((flags & BF_REPORT_ERROR) && m_client_interface) {
3931 0 : m_client_interface->ThreadSafeMessageBox(strError, "", CClientUIInterface::MSG_ERROR);
3932 0 : }
3933 0 : return false;
3934 : }
3935 :
3936 0 : if (addr.IsRoutable() && fDiscover && !(flags & BF_DONT_ADVERTISE) && !NetPermissions::HasFlag(permissions, NetPermissionFlags::NoBan)) {
3937 0 : AddLocal(addr, LOCAL_BIND);
3938 0 : }
3939 :
3940 0 : return true;
3941 0 : }
3942 :
3943 0 : bool CConnman::InitBinds(const Options& options)
3944 : {
3945 0 : bool fBound = false;
3946 0 : for (const auto& addrBind : options.vBinds) {
3947 0 : fBound |= Bind(addrBind, BF_REPORT_ERROR, NetPermissionFlags::None);
3948 : }
3949 0 : for (const auto& addrBind : options.vWhiteBinds) {
3950 0 : fBound |= Bind(addrBind.m_service, BF_REPORT_ERROR, addrBind.m_flags);
3951 : }
3952 0 : for (const auto& addr_bind : options.onion_binds) {
3953 0 : fBound |= Bind(addr_bind, BF_DONT_ADVERTISE, NetPermissionFlags::None);
3954 : }
3955 0 : if (options.bind_on_any) {
3956 : struct in_addr inaddr_any;
3957 0 : inaddr_any.s_addr = htonl(INADDR_ANY);
3958 0 : struct in6_addr inaddr6_any = IN6ADDR_ANY_INIT;
3959 0 : fBound |= Bind(CService(inaddr6_any, GetListenPort()), BF_NONE, NetPermissionFlags::None);
3960 0 : fBound |= Bind(CService(inaddr_any, GetListenPort()), !fBound ? BF_REPORT_ERROR : BF_NONE, NetPermissionFlags::None);
3961 0 : }
3962 0 : return fBound;
3963 0 : }
3964 :
3965 0 : bool CConnman::Start(CDeterministicMNManager& dmnman, CMasternodeMetaMan& mn_metaman, CMasternodeSync& mn_sync,
3966 : CScheduler& scheduler, const Options& connOptions)
3967 : {
3968 0 : AssertLockNotHeld(m_total_bytes_sent_mutex);
3969 0 : Init(connOptions);
3970 :
3971 0 : if (socketEventsMode == SocketEventsMode::EPoll || socketEventsMode == SocketEventsMode::KQueue) {
3972 0 : m_edge_trig_events = std::make_unique<EdgeTriggeredEvents>(socketEventsMode);
3973 0 : if (!m_edge_trig_events->IsValid()) {
3974 0 : LogPrintf("Unable to initialize EdgeTriggeredEvents instance\n");
3975 0 : m_edge_trig_events.reset();
3976 0 : return false;
3977 : }
3978 0 : }
3979 :
3980 0 : if (fListen && !InitBinds(connOptions)) {
3981 0 : if (m_client_interface) {
3982 0 : m_client_interface->ThreadSafeMessageBox(
3983 0 : _("Failed to listen on any port. Use -listen=0 if you want this."),
3984 0 : "", CClientUIInterface::MSG_ERROR);
3985 0 : }
3986 0 : return false;
3987 : }
3988 :
3989 0 : Proxy i2p_sam;
3990 0 : if (GetProxy(NET_I2P, i2p_sam) && connOptions.m_i2p_accept_incoming) {
3991 0 : m_i2p_sam_session = std::make_unique<i2p::sam::Session>(gArgs.GetDataDirNet() / "i2p_private_key",
3992 0 : i2p_sam, &interruptNet);
3993 0 : }
3994 :
3995 0 : for (const auto& strDest : connOptions.vSeedNodes) {
3996 0 : AddAddrFetch(strDest);
3997 : }
3998 :
3999 0 : if (m_use_addrman_outgoing) {
4000 : // Load addresses from anchors.dat
4001 0 : m_anchors = ReadAnchors(gArgs.GetDataDirNet() / ANCHORS_DATABASE_FILENAME);
4002 0 : if (m_anchors.size() > MAX_BLOCK_RELAY_ONLY_ANCHORS) {
4003 0 : m_anchors.resize(MAX_BLOCK_RELAY_ONLY_ANCHORS);
4004 0 : }
4005 0 : LogPrintf("%i block-relay-only anchors will be tried for connections.\n", m_anchors.size());
4006 0 : }
4007 :
4008 0 : if (m_client_interface) {
4009 0 : m_client_interface->InitMessage(_("Starting network threads…").translated);
4010 0 : }
4011 :
4012 0 : fAddressesInitialized = true;
4013 :
4014 0 : if (semOutbound == nullptr) {
4015 : // initialize semaphore
4016 0 : semOutbound = std::make_unique<CSemaphore>(std::min(m_max_outbound, nMaxConnections));
4017 0 : }
4018 0 : if (semAddnode == nullptr) {
4019 : // initialize semaphore
4020 0 : semAddnode = std::make_unique<CSemaphore>(nMaxAddnode);
4021 0 : }
4022 :
4023 : //
4024 : // Start threads
4025 : //
4026 0 : assert(m_msgproc);
4027 0 : InterruptSocks5(false);
4028 0 : interruptNet.reset();
4029 0 : flagInterruptMsgProc = false;
4030 :
4031 : {
4032 0 : LOCK(mutexMsgProc);
4033 0 : fMsgProcWake = false;
4034 0 : }
4035 :
4036 : #ifdef USE_WAKEUP_PIPE
4037 0 : m_wakeup_pipe = std::make_unique<WakeupPipe>(m_edge_trig_events.get());
4038 0 : if (!m_wakeup_pipe->IsValid()) {
4039 : /* We log the error but do not halt initialization */
4040 0 : LogPrintf("Unable to initialize WakeupPipe instance\n");
4041 0 : m_wakeup_pipe.reset();
4042 0 : }
4043 : #endif /* USE_WAKEUP_PIPE */
4044 :
4045 : // Send and receive from sockets, accept connections
4046 0 : threadSocketHandler = std::thread(&util::TraceThread, "net", [this, &mn_sync] { ThreadSocketHandler(mn_sync); });
4047 :
4048 0 : if (!gArgs.GetBoolArg("-dnsseed", DEFAULT_DNSSEED))
4049 0 : LogPrintf("DNS seeding disabled\n");
4050 : else
4051 0 : threadDNSAddressSeed = std::thread(&util::TraceThread, "dnsseed", [this] { ThreadDNSAddressSeed(); });
4052 :
4053 : // Initiate manual connections
4054 0 : threadOpenAddedConnections = std::thread(&util::TraceThread, "addcon", [this] { ThreadOpenAddedConnections(); });
4055 :
4056 0 : if (connOptions.m_use_addrman_outgoing && !connOptions.m_specified_outgoing.empty()) {
4057 0 : if (m_client_interface) {
4058 0 : m_client_interface->ThreadSafeMessageBox(
4059 0 : _("Cannot provide specific connections and have addrman find outgoing connections at the same time."),
4060 0 : "", CClientUIInterface::MSG_ERROR);
4061 0 : }
4062 0 : return false;
4063 : }
4064 0 : if (connOptions.m_use_addrman_outgoing || !connOptions.m_specified_outgoing.empty()) {
4065 0 : threadOpenConnections = std::thread(
4066 0 : &util::TraceThread, "opencon",
4067 0 : [this, connect = connOptions.m_specified_outgoing, &dmnman] { ThreadOpenConnections(connect, dmnman); });
4068 0 : }
4069 :
4070 : // Initiate masternode connections
4071 0 : threadOpenMasternodeConnections = std::thread(&util::TraceThread, "mncon", [this, &dmnman, &mn_metaman, &mn_sync] {
4072 0 : ThreadOpenMasternodeConnections(dmnman, mn_metaman, mn_sync);
4073 0 : });
4074 :
4075 : // Process messages
4076 0 : threadMessageHandler = std::thread(&util::TraceThread, "msghand", [this] { ThreadMessageHandler(); });
4077 :
4078 0 : if (m_i2p_sam_session) {
4079 0 : threadI2PAcceptIncoming =
4080 0 : std::thread(&util::TraceThread, "i2paccept", [this, &mn_sync] { ThreadI2PAcceptIncoming(mn_sync); });
4081 0 : }
4082 :
4083 : // Dump network addresses
4084 0 : scheduler.scheduleEvery([this] { DumpAddresses(); }, DUMP_PEERS_INTERVAL);
4085 :
4086 : // Run the ASMap Health check once and then schedule it to run every 24h.
4087 0 : if (m_netgroupman.UsingASMap()) {
4088 0 : ASMapHealthCheck();
4089 0 : scheduler.scheduleEvery([this] { ASMapHealthCheck(); }, ASMAP_HEALTH_CHECK_INTERVAL);
4090 0 : }
4091 :
4092 0 : return true;
4093 0 : }
4094 :
4095 : class CNetCleanup
4096 : {
4097 : public:
4098 : CNetCleanup() = default;
4099 :
4100 0 : ~CNetCleanup()
4101 0 : {
4102 : #ifdef WIN32
4103 : // Shutdown Windows Sockets
4104 : WSACleanup();
4105 : #endif
4106 0 : }
4107 : };
4108 : static CNetCleanup instance_of_cnetcleanup;
4109 :
4110 0 : void CExplicitNetCleanup::callCleanup()
4111 : {
4112 : // Explicit call to destructor of CNetCleanup because it's not implicitly called
4113 : // when the wallet is restarted from within the wallet itself.
4114 : CNetCleanup tmp;
4115 0 : }
4116 :
4117 631 : void CConnman::Interrupt()
4118 : {
4119 : {
4120 631 : LOCK(mutexMsgProc);
4121 631 : flagInterruptMsgProc = true;
4122 631 : }
4123 631 : condMsgProc.notify_all();
4124 :
4125 631 : interruptNet();
4126 631 : InterruptSocks5(true);
4127 :
4128 631 : if (semOutbound) {
4129 0 : for (int i=0; i<m_max_outbound; i++) {
4130 0 : semOutbound->post();
4131 0 : }
4132 0 : }
4133 :
4134 631 : if (semAddnode) {
4135 0 : for (int i=0; i<nMaxAddnode; i++) {
4136 0 : semAddnode->post();
4137 0 : }
4138 0 : }
4139 631 : }
4140 :
4141 809 : void CConnman::StopThreads()
4142 : {
4143 809 : if (threadI2PAcceptIncoming.joinable()) {
4144 0 : threadI2PAcceptIncoming.join();
4145 0 : }
4146 809 : if (threadMessageHandler.joinable())
4147 0 : threadMessageHandler.join();
4148 809 : if (threadOpenMasternodeConnections.joinable())
4149 0 : threadOpenMasternodeConnections.join();
4150 809 : if (threadOpenConnections.joinable())
4151 0 : threadOpenConnections.join();
4152 809 : if (threadOpenAddedConnections.joinable())
4153 0 : threadOpenAddedConnections.join();
4154 809 : if (threadDNSAddressSeed.joinable())
4155 0 : threadDNSAddressSeed.join();
4156 809 : if (threadSocketHandler.joinable())
4157 0 : threadSocketHandler.join();
4158 809 : }
4159 :
4160 809 : void CConnman::StopNodes()
4161 : {
4162 809 : if (fAddressesInitialized) {
4163 0 : DumpAddresses();
4164 0 : fAddressesInitialized = false;
4165 :
4166 0 : if (m_use_addrman_outgoing) {
4167 : // Anchor connections are only dumped during clean shutdown.
4168 0 : std::vector<CAddress> anchors_to_dump = GetCurrentBlockRelayOnlyConns();
4169 0 : if (anchors_to_dump.size() > MAX_BLOCK_RELAY_ONLY_ANCHORS) {
4170 0 : anchors_to_dump.resize(MAX_BLOCK_RELAY_ONLY_ANCHORS);
4171 0 : }
4172 0 : DumpAnchors(gArgs.GetDataDirNet() / ANCHORS_DATABASE_FILENAME, anchors_to_dump);
4173 0 : }
4174 0 : }
4175 :
4176 : // Delete peer connections.
4177 809 : std::vector<CNode*> nodes;
4178 1618 : WITH_LOCK(m_nodes_mutex, nodes.swap(m_nodes));
4179 809 : for (CNode *pnode : nodes) {
4180 0 : pnode->CloseSocketDisconnect(this);
4181 0 : DeleteNode(pnode);
4182 : }
4183 :
4184 : // Close listening sockets.
4185 809 : if (m_edge_trig_events) {
4186 0 : for (ListenSocket& hListenSocket : vhListenSocket) {
4187 0 : if (hListenSocket.sock) {
4188 0 : m_edge_trig_events->RemoveSocket(hListenSocket.sock->Get());
4189 0 : }
4190 : }
4191 0 : }
4192 :
4193 809 : for (CNode* pnode : m_nodes_disconnected) {
4194 0 : DeleteNode(pnode);
4195 : }
4196 1618 : WITH_LOCK(cs_mapSocketToNode, mapSocketToNode.clear());
4197 : {
4198 809 : LOCK(cs_sendable_receivable_nodes);
4199 809 : mapReceivableNodes.clear();
4200 809 : }
4201 809 : m_nodes_disconnected.clear();
4202 809 : vhListenSocket.clear();
4203 1618 : WITH_LOCK(m_reconnections_mutex, m_reconnections.clear());
4204 809 : semOutbound.reset();
4205 809 : semAddnode.reset();
4206 : /**
4207 : * m_wakeup_pipe must be reset *before* m_edge_trig_events as it may
4208 : * attempt to call EdgeTriggeredEvents::UnregisterPipe() in its destructor
4209 : */
4210 809 : m_wakeup_pipe.reset();
4211 809 : m_edge_trig_events.reset();
4212 809 : }
4213 :
4214 0 : void CConnman::DeleteNode(CNode* pnode)
4215 : {
4216 0 : assert(pnode);
4217 0 : m_msgproc->FinalizeNode(*pnode);
4218 0 : delete pnode;
4219 0 : }
4220 :
4221 1258 : CConnman::~CConnman()
4222 627 : {
4223 631 : Interrupt();
4224 631 : Stop();
4225 1258 : }
4226 :
4227 0 : std::vector<CAddress> CConnman::GetAddresses(size_t max_addresses, size_t max_pct, std::optional<Network> network, const bool filtered) const
4228 : {
4229 0 : std::vector<CAddress> addresses = addrman.GetAddr(max_addresses, max_pct, network, filtered);
4230 0 : if (m_banman) {
4231 0 : addresses.erase(std::remove_if(addresses.begin(), addresses.end(),
4232 0 : [this](const CAddress& addr){return m_banman->IsDiscouraged(addr) || m_banman->IsBanned(addr);}),
4233 0 : addresses.end());
4234 0 : }
4235 0 : return addresses;
4236 0 : }
4237 :
4238 0 : std::vector<CAddress> CConnman::GetAddresses(CNode& requestor, size_t max_addresses, size_t max_pct)
4239 : {
4240 0 : auto local_socket_bytes = requestor.addrBind.GetAddrBytes();
4241 0 : uint64_t cache_id = GetDeterministicRandomizer(RANDOMIZER_ID_ADDRCACHE)
4242 0 : .Write(requestor.ConnectedThroughNetwork())
4243 0 : .Write(local_socket_bytes.data(), local_socket_bytes.size())
4244 : // For outbound connections, the port of the bound address is randomly
4245 : // assigned by the OS and would therefore not be useful for seeding.
4246 0 : .Write(requestor.IsInboundConn() ? requestor.addrBind.GetPort() : 0)
4247 0 : .Finalize();
4248 0 : const auto current_time = GetTime<std::chrono::microseconds>();
4249 0 : auto r = m_addr_response_caches.emplace(cache_id, CachedAddrResponse{});
4250 : CachedAddrResponse& cache_entry = r.first->second;
4251 : if (cache_entry.m_cache_entry_expiration < current_time) { // If emplace() added new one it has expiration 0.
4252 : cache_entry.m_addrs_response_cache = GetAddresses(max_addresses, max_pct, /*network=*/std::nullopt);
4253 : // Choosing a proper cache lifetime is a trade-off between the privacy leak minimization
4254 : // and the usefulness of ADDR responses to honest users.
4255 : //
4256 : // Longer cache lifetime makes it more difficult for an attacker to scrape
4257 : // enough AddrMan data to maliciously infer something useful.
4258 : // By the time an attacker scraped enough AddrMan records, most of
4259 : // the records should be old enough to not leak topology info by
4260 : // e.g. analyzing real-time changes in timestamps.
4261 : //
4262 : // It takes only several hundred requests to scrape everything from an AddrMan containing 100,000 nodes,
4263 : // so ~24 hours of cache lifetime indeed makes the data less inferable by the time
4264 : // most of it could be scraped (considering that timestamps are updated via
4265 : // ADDR self-announcements and when nodes communicate).
4266 : // We also should be robust to those attacks which may not require scraping *full* victim's AddrMan
4267 : // (because even several timestamps of the same handful of nodes may leak privacy).
4268 : //
4269 : // On the other hand, longer cache lifetime makes ADDR responses
4270 : // outdated and less useful for an honest requestor, e.g. if most nodes
4271 : // in the ADDR response are no longer active.
4272 : //
4273 : // However, the churn in the network is known to be rather low. Since we consider
4274 : // nodes to be "terrible" (see IsTerrible()) if the timestamps are older than 30 days,
4275 : // max. 24 hours of "penalty" due to cache shouldn't make any meaningful difference
4276 : // in terms of the freshness of the response.
4277 : cache_entry.m_cache_entry_expiration = current_time + std::chrono::hours(21) + GetRandMillis(std::chrono::hours(6));
4278 : }
4279 : return cache_entry.m_addrs_response_cache;
4280 0 : }
4281 :
4282 7 : bool CConnman::AddNode(const AddedNodeParams& add)
4283 : {
4284 7 : const CService resolved(LookupNumeric(add.m_added_node, Params().GetDefaultPort(add.m_added_node)));
4285 7 : const bool resolved_is_valid{resolved.IsValid()};
4286 :
4287 7 : LOCK(m_added_nodes_mutex);
4288 17 : for (const auto& it : m_added_node_params) {
4289 24 : if (add.m_added_node == it.m_added_node || (resolved_is_valid && resolved == LookupNumeric(it.m_added_node, Params().GetDefaultPort(it.m_added_node)))) return false;
4290 : }
4291 :
4292 5 : m_added_node_params.push_back(add);
4293 5 : return true;
4294 7 : }
4295 :
4296 0 : bool CConnman::RemoveAddedNode(const std::string& strNode)
4297 : {
4298 0 : LOCK(m_added_nodes_mutex);
4299 0 : for (auto it = m_added_node_params.begin(); it != m_added_node_params.end(); ++it) {
4300 0 : if (strNode == it->m_added_node) {
4301 0 : m_added_node_params.erase(it);
4302 0 : return true;
4303 : }
4304 0 : }
4305 0 : return false;
4306 0 : }
4307 :
4308 6 : bool CConnman::AddedNodesContain(const CAddress& addr) const
4309 : {
4310 6 : AssertLockNotHeld(m_added_nodes_mutex);
4311 6 : const std::string addr_str{addr.ToStringAddr()};
4312 6 : const std::string addr_port_str{addr.ToStringAddrPort()};
4313 6 : LOCK(m_added_nodes_mutex);
4314 6 : return (m_added_node_params.size() < 24 // bound the query to a reasonable limit
4315 6 : && std::any_of(m_added_node_params.cbegin(), m_added_node_params.cend(),
4316 26 : [&](const auto& p) { return p.m_added_node == addr_str || p.m_added_node == addr_port_str; }));
4317 6 : }
4318 :
4319 0 : bool CConnman::AddPendingMasternode(const uint256& proTxHash)
4320 : {
4321 0 : LOCK(cs_vPendingMasternodes);
4322 0 : if (std23::ranges::contains(vPendingMasternodes, proTxHash)) {
4323 0 : return false;
4324 : }
4325 :
4326 0 : vPendingMasternodes.push_back(proTxHash);
4327 0 : return true;
4328 0 : }
4329 :
4330 0 : void CConnman::SetMasternodeQuorumNodes(Consensus::LLMQType llmqType, const uint256& quorumHash, const Uint256HashSet& proTxHashes)
4331 : {
4332 0 : LOCK(cs_vPendingMasternodes);
4333 0 : auto it = masternodeQuorumNodes.emplace(std::make_pair(llmqType, quorumHash), proTxHashes);
4334 0 : if (!it.second) {
4335 0 : it.first->second = proTxHashes;
4336 0 : }
4337 0 : }
4338 :
4339 0 : void CConnman::SetMasternodeQuorumRelayMembers(Consensus::LLMQType llmqType, const uint256& quorumHash, const Uint256HashSet& proTxHashes)
4340 : {
4341 : {
4342 0 : LOCK(cs_vPendingMasternodes);
4343 0 : auto it = masternodeQuorumRelayMembers.emplace(std::make_pair(llmqType, quorumHash), proTxHashes);
4344 0 : if (!it.second) {
4345 0 : it.first->second = proTxHashes;
4346 0 : }
4347 0 : }
4348 :
4349 : // Update existing connections
4350 0 : ForEachNode([&](CNode* pnode) {
4351 0 : auto verifiedProRegTxHash = pnode->GetVerifiedProRegTxHash();
4352 0 : if (!verifiedProRegTxHash.IsNull() && !pnode->m_masternode_iqr_connection && IsMasternodeQuorumRelayMember(verifiedProRegTxHash)) {
4353 : // Tell our peer that we're interested in plain LLMQ recovered signatures.
4354 : // Otherwise the peer would only announce/send messages resulting from QRECSIG,
4355 : // e.g. InstantSend locks or ChainLocks. SPV and regular full nodes should not send
4356 : // this message as they are usually only interested in the higher level messages.
4357 0 : const CNetMsgMaker msgMaker(pnode->GetCommonVersion());
4358 0 : PushMessage(pnode, msgMaker.Make(NetMsgType::QSENDRECSIGS, true));
4359 0 : pnode->m_masternode_iqr_connection = true;
4360 0 : }
4361 0 : });
4362 0 : }
4363 :
4364 0 : bool CConnman::HasMasternodeQuorumNodes(Consensus::LLMQType llmqType, const uint256& quorumHash) const
4365 : {
4366 0 : LOCK(cs_vPendingMasternodes);
4367 0 : return masternodeQuorumNodes.count(std::make_pair(llmqType, quorumHash));
4368 0 : }
4369 :
4370 0 : Uint256HashSet CConnman::GetMasternodeQuorums(Consensus::LLMQType llmqType) const
4371 : {
4372 0 : LOCK(cs_vPendingMasternodes);
4373 0 : Uint256HashSet result;
4374 0 : for (const auto& p : masternodeQuorumNodes) {
4375 0 : if (p.first.first != llmqType) {
4376 0 : continue;
4377 : }
4378 0 : result.emplace(p.first.second);
4379 : }
4380 0 : return result;
4381 0 : }
4382 :
4383 0 : std::vector<NodeId> CConnman::GetMasternodeQuorumNodes(Consensus::LLMQType llmqType, const uint256& quorumHash) const
4384 : {
4385 0 : READ_LOCK(m_nodes_mutex);
4386 0 : LOCK(cs_vPendingMasternodes);
4387 0 : auto it = masternodeQuorumNodes.find(std::make_pair(llmqType, quorumHash));
4388 0 : if (it == masternodeQuorumNodes.end()) {
4389 0 : return {};
4390 : }
4391 0 : const auto& proRegTxHashes = it->second;
4392 :
4393 0 : std::vector<NodeId> nodes;
4394 :
4395 0 : auto IsMasternodeQuorumNode = [&](const CNode* n) {
4396 0 : if (n->fDisconnect) return false;
4397 0 : const auto h = n->GetVerifiedProRegTxHash();
4398 0 : return n->qwatch || (!h.IsNull() && proRegTxHashes.contains(h));
4399 0 : };
4400 :
4401 0 : for (NodeId id : m_nodes
4402 0 : | std::views::filter(IsMasternodeQuorumNode)
4403 0 : | std::views::transform([](const CNode* n){ return n->GetId(); })) {
4404 0 : nodes.push_back(id);
4405 : }
4406 0 : return nodes;
4407 0 : }
4408 :
4409 0 : void CConnman::RemoveMasternodeQuorumNodes(Consensus::LLMQType llmqType, const uint256& quorumHash)
4410 : {
4411 0 : LOCK(cs_vPendingMasternodes);
4412 0 : masternodeQuorumNodes.erase(std::make_pair(llmqType, quorumHash));
4413 0 : masternodeQuorumRelayMembers.erase(std::make_pair(llmqType, quorumHash));
4414 0 : }
4415 :
4416 0 : bool CConnman::IsMasternodeQuorumNode(const CNode* pnode, const CDeterministicMNList& tip_mn_list) const
4417 : {
4418 : // Let's see if this is an outgoing connection to an address that is known to be a masternode
4419 : // We however only need to know this if the node did not authenticate itself as a MN yet
4420 0 : uint256 assumedProTxHash;
4421 0 : if (pnode->GetVerifiedProRegTxHash().IsNull() && !pnode->IsInboundConn()) {
4422 0 : auto dmn = tip_mn_list.GetMNByService(pnode->addr);
4423 0 : if (dmn == nullptr) {
4424 : // This is definitely not a masternode
4425 0 : return false;
4426 : }
4427 0 : assumedProTxHash = dmn->proTxHash;
4428 0 : }
4429 :
4430 0 : LOCK(cs_vPendingMasternodes);
4431 0 : for (const auto& p : masternodeQuorumNodes) {
4432 0 : if (!pnode->GetVerifiedProRegTxHash().IsNull()) {
4433 0 : if (p.second.count(pnode->GetVerifiedProRegTxHash())) {
4434 0 : return true;
4435 : }
4436 0 : } else if (!assumedProTxHash.IsNull()) {
4437 0 : if (p.second.count(assumedProTxHash)) {
4438 0 : return true;
4439 : }
4440 0 : }
4441 : }
4442 0 : return false;
4443 0 : }
4444 :
4445 0 : bool CConnman::IsMasternodeQuorumRelayMember(const uint256& protxHash)
4446 : {
4447 0 : if (protxHash.IsNull()) {
4448 0 : return false;
4449 : }
4450 0 : LOCK(cs_vPendingMasternodes);
4451 0 : for (const auto& p : masternodeQuorumRelayMembers) {
4452 0 : if (p.second.count(protxHash)) {
4453 0 : return true;
4454 : }
4455 : }
4456 0 : return false;
4457 0 : }
4458 :
4459 0 : void CConnman::AddPendingProbeConnections(const Uint256HashSet& proTxHashes)
4460 : {
4461 0 : LOCK(cs_vPendingMasternodes);
4462 0 : masternodePendingProbes.insert(proTxHashes.begin(), proTxHashes.end());
4463 0 : }
4464 :
4465 18 : size_t CConnman::GetNodeCount(ConnectionDirection flags) const
4466 : {
4467 18 : READ_LOCK(m_nodes_mutex);
4468 :
4469 18 : int nNum = 0;
4470 18 : for (const auto& pnode : m_nodes) {
4471 0 : if (pnode->fDisconnect) {
4472 0 : continue;
4473 : }
4474 0 : if ((flags & ConnectionDirection::Verified) && pnode->GetVerifiedProRegTxHash().IsNull()) {
4475 0 : continue;
4476 : }
4477 0 : if (flags & (pnode->IsInboundConn() ? ConnectionDirection::In : ConnectionDirection::Out)) {
4478 0 : nNum++;
4479 0 : } else if (flags == ConnectionDirection::Verified) {
4480 0 : nNum++;
4481 0 : }
4482 : }
4483 :
4484 18 : return nNum;
4485 18 : }
4486 :
4487 0 : std::map<CNetAddr, LocalServiceInfo> CConnman::getNetLocalAddresses() const
4488 : {
4489 0 : LOCK(g_maplocalhost_mutex);
4490 0 : return mapLocalHost;
4491 0 : }
4492 :
4493 0 : size_t CConnman::GetMaxOutboundNodeCount()
4494 : {
4495 0 : return m_max_outbound;
4496 : }
4497 :
4498 3 : size_t CConnman::GetMaxOutboundOnionNodeCount()
4499 : {
4500 3 : return m_max_outbound_onion;
4501 : }
4502 :
4503 4 : uint32_t CConnman::GetMappedAS(const CNetAddr& addr) const
4504 : {
4505 4 : return m_netgroupman.GetMappedAS(addr);
4506 : }
4507 :
4508 0 : void CConnman::GetNodeStats(std::vector<CNodeStats>& vstats) const
4509 : {
4510 0 : vstats.clear();
4511 0 : READ_LOCK(m_nodes_mutex);
4512 0 : vstats.reserve(m_nodes.size());
4513 0 : for (const CNode* pnode : m_nodes) {
4514 0 : if (pnode->fDisconnect) {
4515 0 : continue;
4516 : }
4517 0 : vstats.emplace_back();
4518 0 : pnode->CopyStats(vstats.back());
4519 0 : vstats.back().m_mapped_as = GetMappedAS(pnode->addr);
4520 : }
4521 0 : }
4522 :
4523 0 : bool CConnman::DisconnectNode(const std::string& strNode)
4524 : {
4525 0 : return WithNodeMutable(strNode, [&](CNode* pnode){
4526 0 : LogPrint(BCLog::NET_NETCONN, "disconnect by address%s matched peer=%d; disconnecting\n", (fLogIPs ? strprintf("=%s", strNode) : ""), pnode->GetId());
4527 0 : pnode->fDisconnect = true;
4528 0 : return true;
4529 0 : }).value_or(false);
4530 : }
4531 :
4532 11 : bool CConnman::DisconnectNode(const CSubNet& subnet)
4533 : {
4534 11 : bool disconnected = false;
4535 11 : READ_LOCK(m_nodes_mutex);
4536 17 : for (CNode* pnode : m_nodes) {
4537 6 : if (subnet.Match(pnode->addr)) {
4538 3 : LogPrint(BCLog::NET_NETCONN, "disconnect by subnet%s matched peer=%d; disconnecting\n", (fLogIPs ? strprintf("=%s", subnet.ToString()) : ""), pnode->GetId());
4539 3 : pnode->fDisconnect = true;
4540 3 : disconnected = true;
4541 3 : }
4542 : }
4543 11 : return disconnected;
4544 11 : }
4545 :
4546 6 : bool CConnman::DisconnectNode(const CNetAddr& addr)
4547 : {
4548 6 : return DisconnectNode(CSubNet(addr));
4549 0 : }
4550 :
4551 0 : bool CConnman::DisconnectNode(NodeId id)
4552 : {
4553 0 : return WithNodeMutable(id, [&](CNode* pnode){
4554 0 : LogPrint(BCLog::NET_NETCONN, "disconnect by id peer=%d; disconnecting\n", pnode->GetId());
4555 0 : pnode->fDisconnect = true;
4556 0 : return true;
4557 0 : }).value_or(false);
4558 : }
4559 :
4560 0 : void CConnman::RecordBytesRecv(uint64_t bytes)
4561 : {
4562 0 : nTotalBytesRecv += bytes;
4563 0 : ::g_stats_client->count("bandwidth.bytesReceived", bytes, 0.1f);
4564 0 : ::g_stats_client->gauge("bandwidth.totalBytesReceived", nTotalBytesRecv.load(), 0.01f);
4565 0 : }
4566 :
4567 0 : void CConnman::RecordBytesSent(uint64_t bytes)
4568 : {
4569 0 : AssertLockNotHeld(m_total_bytes_sent_mutex);
4570 0 : LOCK(m_total_bytes_sent_mutex);
4571 :
4572 0 : nTotalBytesSent += bytes;
4573 0 : ::g_stats_client->count("bandwidth.bytesSent", bytes, 0.01f);
4574 0 : ::g_stats_client->gauge("bandwidth.totalBytesSent", nTotalBytesSent, 0.01f);
4575 :
4576 0 : const auto now = GetTime<std::chrono::seconds>();
4577 0 : if (nMaxOutboundCycleStartTime + MAX_UPLOAD_TIMEFRAME < now)
4578 : {
4579 : // timeframe expired, reset cycle
4580 0 : nMaxOutboundCycleStartTime = now;
4581 0 : nMaxOutboundTotalBytesSentInCycle = 0;
4582 0 : }
4583 :
4584 0 : nMaxOutboundTotalBytesSentInCycle += bytes;
4585 0 : }
4586 :
4587 0 : uint64_t CConnman::GetMaxOutboundTarget() const
4588 : {
4589 0 : AssertLockNotHeld(m_total_bytes_sent_mutex);
4590 0 : LOCK(m_total_bytes_sent_mutex);
4591 0 : return nMaxOutboundLimit;
4592 0 : }
4593 :
4594 0 : std::chrono::seconds CConnman::GetMaxOutboundTimeframe() const
4595 : {
4596 0 : return MAX_UPLOAD_TIMEFRAME;
4597 : }
4598 :
4599 0 : std::chrono::seconds CConnman::GetMaxOutboundTimeLeftInCycle() const
4600 : {
4601 0 : AssertLockNotHeld(m_total_bytes_sent_mutex);
4602 0 : LOCK(m_total_bytes_sent_mutex);
4603 0 : return GetMaxOutboundTimeLeftInCycle_();
4604 0 : }
4605 :
4606 0 : std::chrono::seconds CConnman::GetMaxOutboundTimeLeftInCycle_() const
4607 : {
4608 0 : AssertLockHeld(m_total_bytes_sent_mutex);
4609 :
4610 0 : if (nMaxOutboundLimit == 0)
4611 0 : return 0s;
4612 :
4613 0 : if (nMaxOutboundCycleStartTime.count() == 0)
4614 0 : return MAX_UPLOAD_TIMEFRAME;
4615 :
4616 0 : const std::chrono::seconds cycleEndTime = nMaxOutboundCycleStartTime + MAX_UPLOAD_TIMEFRAME;
4617 0 : const auto now = GetTime<std::chrono::seconds>();
4618 0 : return (cycleEndTime < now) ? 0s : cycleEndTime - now;
4619 0 : }
4620 :
4621 0 : bool CConnman::OutboundTargetReached(bool historicalBlockServingLimit) const
4622 : {
4623 0 : AssertLockNotHeld(m_total_bytes_sent_mutex);
4624 0 : LOCK(m_total_bytes_sent_mutex);
4625 0 : if (nMaxOutboundLimit == 0)
4626 0 : return false;
4627 :
4628 0 : if (historicalBlockServingLimit)
4629 : {
4630 : // keep a large enough buffer to at least relay each block once
4631 0 : const std::chrono::seconds timeLeftInCycle = GetMaxOutboundTimeLeftInCycle_();
4632 0 : const uint64_t buffer = timeLeftInCycle / std::chrono::minutes{10} * MaxBlockSize();
4633 0 : if (buffer >= nMaxOutboundLimit || nMaxOutboundTotalBytesSentInCycle >= nMaxOutboundLimit - buffer)
4634 0 : return true;
4635 0 : }
4636 0 : else if (nMaxOutboundTotalBytesSentInCycle >= nMaxOutboundLimit)
4637 0 : return true;
4638 :
4639 0 : return false;
4640 0 : }
4641 :
4642 0 : uint64_t CConnman::GetOutboundTargetBytesLeft() const
4643 : {
4644 0 : AssertLockNotHeld(m_total_bytes_sent_mutex);
4645 0 : LOCK(m_total_bytes_sent_mutex);
4646 0 : if (nMaxOutboundLimit == 0)
4647 0 : return 0;
4648 :
4649 0 : return (nMaxOutboundTotalBytesSentInCycle >= nMaxOutboundLimit) ? 0 : nMaxOutboundLimit - nMaxOutboundTotalBytesSentInCycle;
4650 0 : }
4651 :
4652 0 : uint64_t CConnman::GetTotalBytesRecv() const
4653 : {
4654 0 : return nTotalBytesRecv;
4655 : }
4656 :
4657 0 : uint64_t CConnman::GetTotalBytesSent() const
4658 : {
4659 0 : AssertLockNotHeld(m_total_bytes_sent_mutex);
4660 0 : LOCK(m_total_bytes_sent_mutex);
4661 0 : return nTotalBytesSent;
4662 0 : }
4663 :
4664 815 : ServiceFlags CConnman::GetLocalServices() const
4665 : {
4666 815 : return nLocalServices;
4667 : }
4668 :
4669 40 : static std::unique_ptr<Transport> MakeTransport(NodeId id, bool use_v2transport, bool inbound) noexcept
4670 : {
4671 40 : if (use_v2transport) {
4672 0 : return std::make_unique<V2Transport>(id, /*initiating=*/!inbound, SER_NETWORK, INIT_PROTO_VERSION);
4673 : } else {
4674 40 : return std::make_unique<V1Transport>(id, SER_NETWORK, INIT_PROTO_VERSION);
4675 : }
4676 40 : }
4677 :
4678 80 : CNode::CNode(NodeId idIn,
4679 : std::shared_ptr<Sock> sock,
4680 : const CAddress& addrIn,
4681 : uint64_t nKeyedNetGroupIn,
4682 : uint64_t nLocalHostNonceIn,
4683 : const CAddress& addrBindIn,
4684 : const std::string& addrNameIn,
4685 : ConnectionType conn_type_in,
4686 : bool inbound_onion,
4687 : CNodeOptions&& node_opts)
4688 40 : : m_transport{MakeTransport(idIn, node_opts.use_v2transport, conn_type_in == ConnectionType::INBOUND)},
4689 40 : m_permission_flags{node_opts.permission_flags},
4690 40 : m_sock{sock},
4691 : m_connected{GetTime<std::chrono::seconds>()},
4692 : addr{addrIn},
4693 : addrBind{addrBindIn},
4694 : m_addr_name{addrNameIn.empty() ? addr.ToStringAddrPort() : addrNameIn},
4695 : m_dest(addrNameIn),
4696 : m_inbound_onion{inbound_onion},
4697 : m_prefer_evict{node_opts.prefer_evict},
4698 : nKeyedNetGroup{nKeyedNetGroupIn},
4699 : m_conn_type{conn_type_in},
4700 : id{idIn},
4701 : nLocalHostNonce{nLocalHostNonceIn},
4702 : m_recv_flood_size{node_opts.recv_flood_size},
4703 : m_i2p_sam_session{std::move(node_opts.i2p_sam_session)}
4704 40 : {
4705 : if (inbound_onion) assert(conn_type_in == ConnectionType::INBOUND);
4706 :
4707 : for (const std::string &msg : getAllNetMessageTypes())
4708 : mapRecvBytesPerMsgType[msg] = 0;
4709 : mapRecvBytesPerMsgType[NET_MESSAGE_TYPE_OTHER] = 0;
4710 :
4711 : if (fLogIPs) {
4712 : LogPrint(BCLog::NET, "Added connection to %s peer=%d\n", m_addr_name, id);
4713 : } else {
4714 : LogPrint(BCLog::NET, "Added connection peer=%d\n", id);
4715 : }
4716 40 : }
4717 :
4718 2 : void CNode::MarkReceivedMsgsForProcessing()
4719 : {
4720 2 : AssertLockNotHeld(m_msg_process_queue_mutex);
4721 :
4722 2 : size_t nQuorumSizeAdded = 0;
4723 2 : size_t nNormalSizeAdded = 0;
4724 2 : std::list<CNetMessage> quorumMsgs;
4725 2 : std::list<CNetMessage> normalMsgs;
4726 :
4727 : // Classify messages into quorum-priority and normal queues
4728 4 : for (auto it = vRecvMsg.begin(); it != vRecvMsg.end();) {
4729 2 : auto& msg = *it;
4730 : // vRecvMsg contains only completed CNetMessage
4731 : // the single possible partially deserialized message are held by TransportDeserializer
4732 2 : if (IsQuorumPriorityMessage(msg.m_type)) {
4733 0 : quorumMsgs.splice(quorumMsgs.end(), vRecvMsg, it++);
4734 0 : nQuorumSizeAdded += msg.m_raw_message_size;
4735 0 : } else {
4736 2 : normalMsgs.splice(normalMsgs.end(), vRecvMsg, it++);
4737 2 : nNormalSizeAdded += msg.m_raw_message_size;
4738 : }
4739 : }
4740 :
4741 2 : LOCK(m_msg_process_queue_mutex);
4742 : // Splice classified messages into appropriate queues
4743 2 : m_msg_quorum_queue.splice(m_msg_quorum_queue.end(), quorumMsgs);
4744 2 : m_msg_quorum_queue_size += nQuorumSizeAdded;
4745 2 : m_msg_process_queue.splice(m_msg_process_queue.end(), normalMsgs);
4746 2 : m_msg_process_queue_size += nNormalSizeAdded;
4747 : // Compute backpressure over combined size of both queues
4748 2 : fPauseRecv = (m_msg_quorum_queue_size + m_msg_process_queue_size) > m_recv_flood_size;
4749 2 : }
4750 :
4751 2 : std::optional<std::pair<CNetMessage, bool>> CNode::PollMessage()
4752 : {
4753 2 : LOCK(m_msg_process_queue_mutex);
4754 :
4755 : // Ratio-based processing: process N quorum messages for every 1 normal message
4756 : // This ensures forward progress for both queues while strongly prioritizing quorum messages
4757 : // However, if normal queue is empty, process quorum messages in bursts (like old algorithm)
4758 2 : constexpr size_t QUORUM_TO_NORMAL_RATIO = 100;
4759 :
4760 : // Check if we should process normal queue for forward progress
4761 : // Only apply ratio when both queues have messages to allow burst processing when normal queue is empty
4762 4 : bool skip_quorum_processing = !m_msg_process_queue.empty() &&
4763 2 : m_quorum_msg_count_since_normal >= QUORUM_TO_NORMAL_RATIO;
4764 :
4765 : // Prioritize quorum queue: pop from it first if non-empty and ratio not reached
4766 : // If normal queue is empty, process quorum messages without ratio limit (burst mode)
4767 2 : if (!m_msg_quorum_queue.empty() && !skip_quorum_processing) {
4768 0 : std::list<CNetMessage> msgs;
4769 : // Just take one message from quorum queue
4770 0 : msgs.splice(msgs.begin(), m_msg_quorum_queue, m_msg_quorum_queue.begin());
4771 0 : m_msg_quorum_queue_size -= msgs.front().m_raw_message_size;
4772 : // Only increment counter if normal queue has messages (to track ratio)
4773 : // If normal queue is empty, don't increment so we can process bursts quickly
4774 0 : if (!m_msg_process_queue.empty()) {
4775 0 : ++m_quorum_msg_count_since_normal;
4776 0 : }
4777 : // Compute backpressure over combined size of both queues
4778 0 : fPauseRecv = (m_msg_quorum_queue_size + m_msg_process_queue_size) > m_recv_flood_size;
4779 : // Return true for 'more' if either queue has remaining messages
4780 0 : return std::make_pair(std::move(msgs.front()), !m_msg_quorum_queue.empty() || !m_msg_process_queue.empty());
4781 0 : }
4782 :
4783 : // Process normal queue (either because quorum queue is empty or ratio reached)
4784 2 : if (m_msg_process_queue.empty()) return std::nullopt;
4785 :
4786 2 : std::list<CNetMessage> msgs;
4787 : // Just take one message
4788 2 : msgs.splice(msgs.begin(), m_msg_process_queue, m_msg_process_queue.begin());
4789 2 : m_msg_process_queue_size -= msgs.front().m_raw_message_size;
4790 2 : m_quorum_msg_count_since_normal = 0; // Reset counter after processing normal message
4791 : // Compute backpressure over combined size of both queues
4792 2 : fPauseRecv = (m_msg_quorum_queue_size + m_msg_process_queue_size) > m_recv_flood_size;
4793 :
4794 2 : return std::make_pair(std::move(msgs.front()), !m_msg_quorum_queue.empty() || !m_msg_process_queue.empty());
4795 2 : }
4796 :
4797 55 : bool CConnman::NodeFullyConnected(const CNode* pnode)
4798 : {
4799 55 : return pnode && pnode->fSuccessfullyConnected && !pnode->fDisconnect;
4800 : }
4801 :
4802 40 : void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
4803 : {
4804 40 : AssertLockNotHeld(m_total_bytes_sent_mutex);
4805 40 : size_t nMessageSize = msg.data.size();
4806 40 : LogPrint(BCLog::NET, "sending %s (%d bytes) peer=%d\n", msg.m_type, nMessageSize, pnode->GetId());
4807 40 : if (gArgs.GetBoolArg("-capturemessages", false)) {
4808 10 : CaptureMessage(pnode->addr, msg.m_type, msg.data, /*is_incoming=*/false);
4809 10 : }
4810 :
4811 : TRACE6(net, outbound_message,
4812 : pnode->GetId(),
4813 : pnode->m_addr_name.c_str(),
4814 : pnode->ConnectionTypeAsString().c_str(),
4815 : msg.m_type.c_str(),
4816 : msg.data.size(),
4817 : msg.data.data()
4818 : );
4819 :
4820 40 : ::g_stats_client->count(strprintf("bandwidth.message.%s.bytesSent", msg.m_type), nMessageSize, 1.0f);
4821 40 : ::g_stats_client->inc(strprintf("message.sent.%s", msg.m_type), 1.0f);
4822 :
4823 : {
4824 40 : LOCK(pnode->cs_vSend);
4825 : // Check if the transport still has unsent bytes, and indicate to it that we're about to
4826 : // give it a message to send.
4827 65 : const auto& [to_send, more, _msg_type] =
4828 40 : pnode->m_transport->GetBytesToSend(/*have_next_message=*/true);
4829 40 : const bool queue_was_empty{to_send.empty() && pnode->vSendMsg.empty()};
4830 :
4831 : // Update memory usage of send buffer.
4832 40 : pnode->m_send_memusage += msg.GetMemoryUsage();
4833 40 : if (pnode->m_send_memusage + pnode->m_transport->GetSendMemoryUsage() > nSendBufferMaxSize) pnode->fPauseSend = true;
4834 : // Move message to vSendMsg queue.
4835 40 : pnode->vSendMsg.push_back(std::move(msg));
4836 40 : pnode->nSendMsgSize = pnode->vSendMsg.size();
4837 :
4838 : // If there was nothing to send before, and there is now (predicted by the "more" value
4839 : // returned by the GetBytesToSend call above), attempt "optimistic write":
4840 : // because the poll/select loop may pause for SELECT_TIMEOUT_MILLISECONDS before actually
4841 : // doing a send, try sending from the calling thread if the queue was empty before.
4842 : // With a V1Transport, more will always be true here, because adding a message always
4843 : // results in sendable bytes there, but with V2Transport this is not the case (it may
4844 : // still be in the handshake).
4845 40 : if (queue_was_empty && more) {
4846 25 : if (m_wakeup_pipe && m_wakeup_pipe->m_need_wakeup.load()) {
4847 0 : m_wakeup_pipe->Write();
4848 0 : }
4849 25 : }
4850 40 : }
4851 40 : }
4852 :
4853 0 : bool CConnman::ForNode(const CService& addr, std::function<bool(const CNode* pnode)> cond, std::function<bool(CNode* pnode)> func)
4854 : {
4855 0 : READ_LOCK(m_nodes_mutex);
4856 0 : CNode* found = FindNodeMutable(addr, false);
4857 0 : return found != nullptr && cond(found) && func(found);
4858 0 : }
4859 :
4860 0 : bool CConnman::ForNode(const CService& addr, std::function<bool(const CNode* pnode)> cond, std::function<bool(const CNode* pnode)> func) const
4861 : {
4862 0 : READ_LOCK(m_nodes_mutex);
4863 0 : const CNode* found = FindNode(addr, false);
4864 0 : return found != nullptr && cond(found) && func(found);
4865 0 : }
4866 :
4867 7 : bool CConnman::ForNode(NodeId id, std::function<bool(const CNode* pnode)> cond, std::function<bool(CNode* pnode)> func)
4868 : {
4869 7 : READ_LOCK(m_nodes_mutex);
4870 7 : CNode* found = FindNodeMutable(id, false);
4871 7 : return found != nullptr && cond(found) && func(found);
4872 7 : }
4873 :
4874 0 : bool CConnman::ForNode(NodeId id, std::function<bool(const CNode* pnode)> cond, std::function<bool(const CNode* pnode)> func) const
4875 : {
4876 0 : READ_LOCK(m_nodes_mutex);
4877 0 : const CNode* found = FindNode(id, false);
4878 0 : return found != nullptr && cond(found) && func(found);
4879 0 : }
4880 :
4881 0 : bool CConnman::IsMasternodeOrDisconnectRequested(const CService& addr) const
4882 : {
4883 0 : return ForNode(addr, AllNodes, [](const CNode* pnode){
4884 0 : return pnode->m_masternode_connection || pnode->fDisconnect;
4885 : });
4886 0 : }
4887 :
4888 0 : CConnman::NodesSnapshot::NodesSnapshot(const CConnman& connman, std::function<bool(const CNode* pnode)> filter,
4889 : bool shuffle)
4890 0 : {
4891 : {
4892 0 : READ_LOCK(connman.m_nodes_mutex);
4893 0 : m_nodes_copy.reserve(connman.m_nodes.size());
4894 :
4895 0 : for (auto& node : connman.m_nodes) {
4896 0 : if (!filter(node)) continue;
4897 0 : node->AddRef();
4898 0 : m_nodes_copy.push_back(node);
4899 : }
4900 0 : }
4901 0 : if (shuffle) {
4902 0 : Shuffle(m_nodes_copy.begin(), m_nodes_copy.end(), FastRandomContext{});
4903 0 : }
4904 0 : }
4905 :
4906 0 : CConnman::NodesSnapshot::~NodesSnapshot()
4907 0 : {
4908 0 : for (auto& node : m_nodes_copy) {
4909 0 : node->Release();
4910 : }
4911 0 : }
4912 :
4913 0 : CSipHasher CConnman::GetDeterministicRandomizer(uint64_t id) const
4914 : {
4915 0 : return CSipHasher(nSeed0, nSeed1).Write(id);
4916 : }
4917 :
4918 0 : uint64_t CConnman::CalculateKeyedNetGroup(const CAddress& address) const
4919 : {
4920 0 : std::vector<unsigned char> vchNetGroup(m_netgroupman.GetGroup(address));
4921 :
4922 0 : return GetDeterministicRandomizer(RANDOMIZER_ID_NETGROUP).Write(vchNetGroup.data(), vchNetGroup.size()).Finalize();
4923 0 : }
4924 :
4925 0 : void CConnman::PerformReconnections()
4926 : {
4927 0 : AssertLockNotHeld(m_reconnections_mutex);
4928 0 : AssertLockNotHeld(m_unused_i2p_sessions_mutex);
4929 0 : while (true) {
4930 : // Move first element of m_reconnections to todo (avoiding an allocation inside the lock).
4931 0 : decltype(m_reconnections) todo;
4932 : {
4933 0 : LOCK(m_reconnections_mutex);
4934 0 : if (m_reconnections.empty()) break;
4935 0 : todo.splice(todo.end(), m_reconnections, m_reconnections.begin());
4936 0 : }
4937 :
4938 0 : auto& item = *todo.begin();
4939 0 : OpenNetworkConnection(item.addr_connect,
4940 : // We only reconnect if the first attempt to connect succeeded at
4941 : // connection time, but then failed after the CNode object was
4942 : // created. Since we already know connecting is possible, do not
4943 : // count failure to reconnect.
4944 : /*fCountFailure=*/false,
4945 0 : std::move(item.grant),
4946 0 : item.destination.empty() ? nullptr : item.destination.c_str(),
4947 0 : item.conn_type,
4948 0 : item.use_v2transport,
4949 0 : item.masternode_connection ? MasternodeConn::IsConnection : MasternodeConn::IsNotConnection,
4950 0 : item.masternode_probe_connection ? MasternodeProbeConn::IsConnection : MasternodeProbeConn::IsNotConnection);
4951 0 : }
4952 0 : }
4953 :
4954 0 : void CConnman::ASMapHealthCheck()
4955 : {
4956 0 : const std::vector<CAddress> v4_addrs{GetAddresses(/*max_addresses=*/ 0, /*max_pct=*/ 0, Network::NET_IPV4, /*filtered=*/ false)};
4957 0 : const std::vector<CAddress> v6_addrs{GetAddresses(/*max_addresses=*/ 0, /*max_pct=*/ 0, Network::NET_IPV6, /*filtered=*/ false)};
4958 0 : std::vector<CNetAddr> clearnet_addrs;
4959 0 : clearnet_addrs.reserve(v4_addrs.size() + v6_addrs.size());
4960 0 : std::transform(v4_addrs.begin(), v4_addrs.end(), std::back_inserter(clearnet_addrs),
4961 0 : [](const CAddress& addr) { return static_cast<CNetAddr>(addr); });
4962 0 : std::transform(v6_addrs.begin(), v6_addrs.end(), std::back_inserter(clearnet_addrs),
4963 0 : [](const CAddress& addr) { return static_cast<CNetAddr>(addr); });
4964 0 : m_netgroupman.ASMapHealthCheck(clearnet_addrs);
4965 0 : }
4966 :
4967 : // Dump binary message to file, with timestamp.
4968 7 : static void CaptureMessageToFile(const CAddress& addr,
4969 : const std::string& msg_type,
4970 : Span<const unsigned char> data,
4971 : bool is_incoming)
4972 : {
4973 : // Note: This function captures the message at the time of processing,
4974 : // not at socket receive/send time.
4975 : // This ensures that the messages are always in order from an application
4976 : // layer (processing) perspective.
4977 7 : auto now = GetTime<std::chrono::microseconds>();
4978 :
4979 : // Windows folder names cannot include a colon
4980 7 : std::string clean_addr = addr.ToStringAddrPort();
4981 7 : std::replace(clean_addr.begin(), clean_addr.end(), ':', '_');
4982 :
4983 7 : fs::path base_path = gArgs.GetDataDirNet() / "message_capture" / fs::u8path(clean_addr);
4984 7 : fs::create_directories(base_path);
4985 :
4986 7 : fs::path path = base_path / (is_incoming ? "msgs_recv.dat" : "msgs_sent.dat");
4987 7 : AutoFile f{fsbridge::fopen(path, "ab")};
4988 :
4989 7 : ser_writedata64(f, now.count());
4990 7 : f.write(MakeByteSpan(msg_type));
4991 34 : for (auto i = msg_type.length(); i < CMessageHeader::COMMAND_SIZE; ++i) {
4992 27 : f << uint8_t{'\0'};
4993 27 : }
4994 7 : uint32_t size = data.size();
4995 7 : ser_writedata32(f, size);
4996 7 : f.write(AsBytes(data));
4997 7 : }
4998 :
4999 : std::function<void(const CAddress& addr,
5000 : const std::string& msg_type,
5001 : Span<const unsigned char> data,
5002 : bool is_incoming)>
5003 146 : CaptureMessage = CaptureMessageToFile;
|
