1 // Copyright (c) 2020-2022 The Limenka developers
2 // Distributed under the MIT software license, see the accompanying
3 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
4
5 #ifndef LIMENKA_TEST_UTIL_NET_H
6 #define LIMENKA_TEST_UTIL_NET_H
7
8 #include <compat/compat.h>
9 #include <netmessagemaker.h>
10 #include <net.h>
11 #include <net_permissions.h>
12 #include <net_processing.h>
13 #include <netaddress.h>
14 #include <node/connection_types.h>
15 #include <node/eviction.h>
16 #include <span.h>
17 #include <sync.h>
18 #include <util/sock.h>
19
20 #include <algorithm>
21 #include <array>
22 #include <cassert>
23 #include <chrono>
24 #include <condition_variable>
25 #include <cstdint>
26 #include <cstring>
27 #include <memory>
28 #include <optional>
29 #include <string>
30 #include <unordered_map>
31 #include <vector>
32
33 class FastRandomContext;
34
35 struct ConnmanTestMsg : public CConnman {
36 using CConnman::CConnman;
37
38 void SetMsgProc(NetEventsInterface* msgproc)
39 {
40 m_msgproc = msgproc;
41 }
42
43 void SetPeerConnectTimeout(std::chrono::seconds timeout)
44 {
45 m_peer_connect_timeout = timeout;
46 }
47
48 std::vector<CNode*> TestNodes()
49 {
50 LOCK(m_nodes_mutex);
51 return m_nodes;
52 }
53
54 void AddTestNode(CNode& node)
55 {
56 LOCK(m_nodes_mutex);
57 m_nodes.push_back(&node);
58
59 if (node.IsManualOrFullOutboundConn()) ++m_network_conn_counts[node.addr.GetNetwork()];
60 }
61
62 void ClearTestNodes()
63 {
64 LOCK(m_nodes_mutex);
65 for (CNode* node : m_nodes) {
66 delete node;
67 }
68 m_nodes.clear();
69 }
70
71 void Handshake(CNode& node,
72 bool successfully_connected,
73 ServiceFlags remote_services,
74 ServiceFlags local_services,
75 int32_t version,
76 bool relay_txs)
77 EXCLUSIVE_LOCKS_REQUIRED(NetEventsInterface::g_msgproc_mutex);
78
79 bool ProcessMessagesOnce(CNode& node) EXCLUSIVE_LOCKS_REQUIRED(NetEventsInterface::g_msgproc_mutex)
80 {
81 return m_msgproc->ProcessMessages(&node, flagInterruptMsgProc);
82 }
83
84 void NodeReceiveMsgBytes(CNode& node, Span<const uint8_t> msg_bytes, bool& complete) const;
85
86 bool ReceiveMsgFrom(CNode& node, CSerializedNetMsg&& ser_msg) const;
87 void FlushSendBuffer(CNode& node) const;
88
89 bool AlreadyConnectedPublic(const CAddress& addr) { return AlreadyConnectedToAddress(addr); };
90
91 CNode* ConnectNodePublic(PeerManager& peerman, const char* pszDest, ConnectionType conn_type)
92 EXCLUSIVE_LOCKS_REQUIRED(!m_unused_i2p_sessions_mutex);
93 };
94
95 constexpr ServiceFlags ALL_SERVICE_FLAGS[]{
96 NODE_NONE,
97 NODE_NETWORK,
98 NODE_BLOOM,
99 NODE_WITNESS,
100 NODE_COMPACT_FILTERS,
101 NODE_NETWORK_LIMITED,
102 NODE_P2P_V2,
103 };
104
105 constexpr NetPermissionFlags ALL_NET_PERMISSION_FLAGS[]{
106 NetPermissionFlags::None,
107 NetPermissionFlags::BloomFilter,
108 NetPermissionFlags::Relay,
109 NetPermissionFlags::ForceRelay,
110 NetPermissionFlags::NoBan,
111 NetPermissionFlags::Mempool,
112 NetPermissionFlags::Addr,
113 NetPermissionFlags::Download,
114 NetPermissionFlags::Implicit,
115 NetPermissionFlags::All,
116 };
117
118 constexpr ConnectionType ALL_CONNECTION_TYPES[]{
119 ConnectionType::INBOUND,
120 ConnectionType::OUTBOUND_FULL_RELAY,
121 ConnectionType::MANUAL,
122 ConnectionType::FEELER,
123 ConnectionType::BLOCK_RELAY,
124 ConnectionType::ADDR_FETCH,
125 };
126
127 constexpr auto ALL_NETWORKS = std::array{
128 Network::NET_UNROUTABLE,
129 Network::NET_IPV4,
130 Network::NET_IPV6,
131 Network::NET_ONION,
132 Network::NET_I2P,
133 Network::NET_CJDNS,
134 Network::NET_INTERNAL,
135 };
136
137 /**
138 * A mocked Sock alternative that succeeds on all operations.
139 * Returns infinite amount of 0x0 bytes on reads.
140 */
141 class ZeroSock : public Sock
142 {
143 public:
144 ZeroSock();
145
146 ~ZeroSock() override;
147
148 ssize_t Send(const void*, size_t len, int) const override;
149
150 ssize_t Recv(void* buf, size_t len, int flags) const override;
151
152 int Connect(const sockaddr*, socklen_t) const override;
153
154 int Bind(const sockaddr*, socklen_t) const override;
155
156 int Listen(int) const override;
157
158 std::unique_ptr<Sock> Accept(sockaddr* addr, socklen_t* addr_len) const override;
159
160 int GetSockOpt(int level, int opt_name, void* opt_val, socklen_t* opt_len) const override;
161
162 int SetSockOpt(int, int, const void*, socklen_t) const override;
163
164 int GetSockName(sockaddr* name, socklen_t* name_len) const override;
165
166 bool SetNonBlocking() const override;
167
168 bool IsSelectable() const override;
169
170 bool Wait(std::chrono::milliseconds timeout,
171 Event requested,
172 Event* occurred = nullptr) const override;
173
174 bool WaitMany(std::chrono::milliseconds timeout, EventsPerSock& events_per_sock) const override;
175
176 private:
177 ZeroSock& operator=(Sock&& other) override;
178 };
179
180 /**
181 * A mocked Sock alternative that returns a statically contained data upon read and succeeds
182 * and ignores all writes. The data to be returned is given to the constructor and when it is
183 * exhausted an EOF is returned by further reads.
184 */
185 class StaticContentsSock : public ZeroSock
186 {
187 public:
188 explicit StaticContentsSock(const std::string& contents);
189
190 /**
191 * Return parts of the contents that was provided at construction until it is exhausted
192 * and then return 0 (EOF).
193 */
194 ssize_t Recv(void* buf, size_t len, int flags) const override;
195
196 bool IsConnected(std::string&) const override
197 {
198 return true;
199 }
200
201 private:
202 StaticContentsSock& operator=(Sock&& other) override;
203
204 const std::string m_contents;
205 mutable size_t m_consumed{0};
206 };
207
208 /**
209 * A mocked Sock alternative that allows providing the data to be returned by Recv()
210 * and inspecting the data that has been supplied to Send().
211 */
212 class DynSock : public ZeroSock
213 {
214 public:
215 /**
216 * Unidirectional bytes or CNetMessage queue (FIFO).
217 */
218 class Pipe
219 {
220 public:
221 /**
222 * Get bytes and remove them from the pipe.
223 * @param[in] buf Destination to write bytes to.
224 * @param[in] len Write up to this number of bytes.
225 * @param[in] flags Same as the flags of `recv(2)`. Just `MSG_PEEK` is honored.
226 * @return The number of bytes written to `buf`. `0` if `Eof()` has been called.
227 * If no bytes are available then `-1` is returned and `errno` is set to `EAGAIN`.
228 */
229 ssize_t GetBytes(void* buf, size_t len, int flags = 0) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
230
231 /**
232 * Deserialize a `CNetMessage` and remove it from the pipe.
233 * If not enough bytes are available then the function will wait. If parsing fails
234 * or EOF is signaled to the pipe, then `std::nullopt` is returned.
235 */
236 std::optional<CNetMessage> GetNetMsg() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
237
238 /**
239 * Push bytes to the pipe.
240 */
241 void PushBytes(const void* buf, size_t len) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
242
243 /**
244 * Construct and push CNetMessage to the pipe.
245 */
246 template <typename... Args>
247 void PushNetMsg(const std::string& type, Args&&... payload) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
248
249 /**
250 * Signal end-of-file on the receiving end (`GetBytes()` or `GetNetMsg()`).
251 */
252 void Eof() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex);
253
254 private:
255 /**
256 * Return when there is some data to read or EOF has been signaled.
257 * @param[in,out] lock Unique lock that must have been derived from `m_mutex` by `WAIT_LOCK(m_mutex, lock)`.
258 */
259 void WaitForDataOrEof(UniqueLock<Mutex>& lock) EXCLUSIVE_LOCKS_REQUIRED(m_mutex);
260
261 Mutex m_mutex;
262 std::condition_variable m_cond;
263 std::vector<uint8_t> m_data GUARDED_BY(m_mutex);
264 bool m_eof GUARDED_BY(m_mutex){false};
265 };
266
267 struct Pipes {
268 Pipe recv;
269 Pipe send;
270 };
271
272 /**
273 * A basic thread-safe queue, used for queuing sockets to be returned by Accept().
274 */
275 class Queue
276 {
277 public:
278 using S = std::unique_ptr<DynSock>;
279
280 void Push(S s) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
281 {
282 LOCK(m_mutex);
283 m_queue.push(std::move(s));
284 }
285
286 std::optional<S> Pop() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
287 {
288 LOCK(m_mutex);
289 if (m_queue.empty()) {
290 return std::nullopt;
291 }
292 S front{std::move(m_queue.front())};
293 m_queue.pop();
294 return front;
295 }
296
297 bool Empty() const EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
298 {
299 LOCK(m_mutex);
300 return m_queue.empty();
301 }
302
303 private:
304 mutable Mutex m_mutex;
305 std::queue<S> m_queue GUARDED_BY(m_mutex);
306 };
307
308 /**
309 * Create a new mocked sock.
310 * @param[in] pipes Send/recv pipes used by the Send() and Recv() methods.
311 * @param[in] accept_sockets Sockets to return by the Accept() method.
312 */
313 explicit DynSock(std::shared_ptr<Pipes> pipes, std::shared_ptr<Queue> accept_sockets);
314
315 ~DynSock();
316
317 ssize_t Recv(void* buf, size_t len, int flags) const override;
318
319 ssize_t Send(const void* buf, size_t len, int) const override;
320
321 std::unique_ptr<Sock> Accept(sockaddr* addr, socklen_t* addr_len) const override;
322
323 bool Wait(std::chrono::milliseconds timeout,
324 Event requested,
325 Event* occurred = nullptr) const override;
326
327 bool WaitMany(std::chrono::milliseconds timeout, EventsPerSock& events_per_sock) const override;
328
329 private:
330 DynSock& operator=(Sock&&) override;
331
332 std::shared_ptr<Pipes> m_pipes;
333 std::shared_ptr<Queue> m_accept_sockets;
334 };
335
336 template <typename... Args>
337 void DynSock::Pipe::PushNetMsg(const std::string& type, Args&&... payload)
338 {
339 auto msg = NetMsg::Make(type, std::forward<Args>(payload)...);
340 V1Transport transport{NodeId{0}};
341
342 const bool queued{transport.SetMessageToSend(msg)};
343 assert(queued);
344
345 LOCK(m_mutex);
346
347 for (;;) {
348 const auto& [bytes, _more, _msg_type] = transport.GetBytesToSend(/*have_next_message=*/true);
349 if (bytes.empty()) {
350 break;
351 }
352 m_data.insert(m_data.end(), bytes.begin(), bytes.end());
353 transport.MarkBytesSent(bytes.size());
354 }
355
356 m_cond.notify_all();
357 }
358
359 std::vector<NodeEvictionCandidate> GetRandomNodeEvictionCandidates(int n_candidates, FastRandomContext& random_context);
360
361 #endif // LIMENKA_TEST_UTIL_NET_H
362