42421988994

Committed 07 Nov 2026 10:44PM UTC coverage: 93.582% (-0.3%) from 93.863%

Build # 42421988994

Build Type

Pull #1011

github

Committed by

web-flow

Commit Message

Merge 62b70bb20 into ddc7eabb3

Pull Request Pull Request #1011: Read rpc channel post bodies as json-rpc regardless of content type.

Coverage Stats

70 of 70 new or added lines in 5 files covered. (100.0%)

58 existing lines in 7 files now uncovered.

14567 of 15566 relevant lines covered (93.58%)

2260.09 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

64.94

/src/tls/stream.cpp

/**
 * Copyright (c) 2011-2026 libbitcoin developers
 *
 * This file is part of libbitcoin.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
#include <bitcoin/network/tls/stream.hpp>

#include <bitcoin/network/define.hpp>
#include <bitcoin/network/tls/constants.hpp>

namespace libbitcoin {
namespace network {
namespace tls {

using namespace system;
using namespace std::placeholders;

BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT)

// Constructor.
// ----------------------------------------------------------------------------

stream::stream(io::socket&& socket, const context& context) NOEXCEPT
  : socket_(std::move(socket)),
    context_(context),
    server_(context),
    input_(record_header_size + maximum_ciphertext)
{
}

// Properties.
// ----------------------------------------------------------------------------

stream::executor_type stream::get_executor() NOEXCEPT
{
    return socket_.get_executor();
}

io::socket& stream::next_layer() NOEXCEPT
{
    return socket_;
}

const io::socket& stream::next_layer() const NOEXCEPT
{
    return socket_;
}

// Handshake.
// ----------------------------------------------------------------------------

void stream::async_handshake(result_handler&& handler) NOEXCEPT
{
    if (!context_.is_ready())
    {
        const code ec{ code{ error::tls_alert_internal_error } };
        io::post(get_executor(), std::bind(std::move(handler), ec));
        return;
    }

    receive(std::bind(&stream::handle_handshake,
        this, _1, std::move(handler)));
}

void stream::handle_handshake(const code& ec,
    const result_handler& handler) NOEXCEPT
{
    if (ec)
    {
        handler(ec);
        return;
    }

    if (server_.is_failed())
    {
        handler(failure());
        return;
    }

    if (server_.is_closed())
    {
        handler(error::peer_disconnect);
        return;
    }

    if (server_.is_established())
    {
        handler(error::success);
        return;
    }

    receive(std::bind(&stream::handle_handshake,
        this, _1, handler));
}

// Read.
// ----------------------------------------------------------------------------

void stream::do_read_some(const byte_span& out,
    const io_handler& handler) NOEXCEPT
{
    code ec{};
    size_t size{};
    if (try_read(ec, size, out))
    {
        io::post(get_executor(), std::bind(handler, ec, size));
        return;
    }

    receive(std::bind(&stream::handle_read_some,
        this, _1, out, handler));
}

// True if the read is complete without receiving.
bool stream::try_read(code& ec, size_t& size,
    const byte_span& out) NOEXCEPT
{
    ec = error::success;
    size = zero;
    if (out.empty())
        return true;

    if (is_nonzero(server_.readable()))
    {
        size = server_.read(out);
        return true;
    }

    if (server_.is_closed())
    {
        ec = error::peer_disconnect;
        return true;
    }

    if (server_.is_failed())
    {
        ec = failure();
        return true;
    }

    return false;
}

void stream::handle_read_some(const code& ec, const byte_span& out,
    const io_handler& handler) NOEXCEPT
{
    if (ec)
    {
        handler(ec, zero);
        return;
    }

    code result{};
    size_t size{};
    if (try_read(result, size, out))
    {
        handler(result, size);
        return;
    }

    receive(std::bind(&stream::handle_read_some,
        this, _1, out, handler));
}

// Write.
// ----------------------------------------------------------------------------

void stream::do_write_some(const const_byte_span& data,
    const io_handler& handler) NOEXCEPT
{
    if (data.empty())
    {
        io::post(get_executor(),
            std::bind(handler, code{ error::success }, zero));
        return;
    }

    const auto size = std::min(data.size(), maximum_plaintext);
    if (!server_.write(data.first(size)))
    {
        const auto ec = server_.is_failed() ? failure() :
            code{ error::connect_failed };

        io::post(get_executor(), std::bind(handler, ec, zero));
        return;
    }

    flush([handler, size](const code& ec) NOEXCEPT
    {
        handler(ec, ec ? zero : size);
    });
}

size_t stream::do_write(const const_byte_span& data, code& ec) NOEXCEPT
{
    ec = error::success;
    if (writing_)
    {
        ec = error::operation_failed;
        return zero;
    }

    if (data.empty())
        return zero;

    const auto size = std::min(data.size(), maximum_plaintext);
    if (!server_.write(data.first(size)))
    {
        ec = server_.is_failed() ? failure() : code{ error::connect_failed };
        return zero;
    }

    auto& output = server_.output();
    io::write(socket_, io::buffer(output), ec);
    output.clear();
    return ec ? zero : size;
}

// Shutdown.
// ----------------------------------------------------------------------------

void stream::async_shutdown(result_handler&& handler) NOEXCEPT
{
    if (server_.is_failed())
    {
        io::post(get_executor(),
            std::bind(std::move(handler), code{ error::success }));
        return;
    }

    server_.close();
    flush(std::bind(&stream::handle_close_sent,
        this, _1, std::move(handler)));
}

void stream::handle_close_sent(const code& ec,
    const result_handler& handler) NOEXCEPT
{
    if (ec)
    {
        handler(ec);
        return;
    }

    handle_close_received(ec, handler);
}

void stream::handle_close_received(const code& ec,
    const result_handler& handler) NOEXCEPT
{
    if (ec)
    {
        handler(ec);
        return;
    }

    if (server_.is_failed())
    {
        handler(failure());
        return;
    }

    if (server_.is_closed())
    {
        handler(error::success);
        return;
    }

    receive(std::bind(&stream::handle_close_received,
        this, _1, handler));
}

// Transport.
// ----------------------------------------------------------------------------

code stream::failure() const NOEXCEPT
{
    return code{ error::tls_to_error_code(server_.failure()) };
}

// Closure of the transport without close_notify truncates the stream.
code stream::transport(const code& ec) const NOEXCEPT
{
    if (ec == error::peer_disconnect && !server_.is_closed())
        return code{ error::tls_stream_truncated };

    return ec;
}

// A receive joins any receive in progress, as they share the input buffer.
void stream::receive(result_handler&& handler) NOEXCEPT
{
    receivers_.push_back(std::move(handler));
    if (receiving_)
        return;

    receiving_ = true;
    socket_.async_read_some(io::buffer(input_),
        std::bind(&stream::handle_receive,
            this, _1, _2));
}

// Records produced by receipt (alerts and key updates) are sent first.
void stream::handle_receive(const code& ec, size_t size) NOEXCEPT
{
    receiving_ = false;
    std::vector<result_handler> handlers{};
    handlers.swap(receivers_);

    if (ec)
    {
        const auto result = transport(ec);
        for (const auto& handler: handlers)
            handler(result);

        return;
    }

    server_.receive({ input_.data(), size });
    if (server_.output().empty())
    {
        for (const auto& handler: handlers)
            handler(error::success);

        return;
    }

    for (const auto& handler: handlers)
        flush(handler);
}

// The handler completes once all output preceding it is written.
void stream::flush(const result_handler& handler) NOEXCEPT
{
    waiting_.push_back(handler);
    if (!writing_)
        write_next();
}

void stream::write_next() NOEXCEPT
{
    std::vector<result_handler> handlers{};
    handlers.swap(waiting_);

    auto& output = server_.output();
    if (output.empty())
    {
        for (auto& handler: handlers)
            io::post(get_executor(),
                std::bind(std::move(handler), code{ error::success }));

        return;
    }

    writing_ = true;
    sending_.swap(output);
    output.clear();
    io::async_write(socket_, io::buffer(sending_),
        std::bind(&stream::handle_write,
            this, _1, std::move(handlers)));
}

void stream::handle_write(const code& ec,
    const std::vector<result_handler>& handlers) NOEXCEPT
{
    writing_ = false;
    sending_.clear();
    for (const auto& handler: handlers)
        handler(ec);

    if (!writing_ && !waiting_.empty())
        write_next();
}

BC_POP_WARNING()

} // namespace tls
} // namespace network
} // namespace libbitcoin

1	/**
2	* Copyright (c) 2011-2026 libbitcoin developers
3	*
4	* This file is part of libbitcoin.
5	*
6	* This program is free software: you can redistribute it and/or modify
7	* it under the terms of the GNU Affero General Public License as published by
8	* the Free Software Foundation, either version 3 of the License, or
9	* (at your option) any later version.
10	*
11	* This program is distributed in the hope that it will be useful,
12	* but WITHOUT ANY WARRANTY; without even the implied warranty of
13	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14	* GNU Affero General Public License for more details.
15	*
16	* You should have received a copy of the GNU Affero General Public License
17	* along with this program. If not, see <http://www.gnu.org/licenses/>.
18	*/
19	#include <bitcoin/network/tls/stream.hpp>
20
21	#include <bitcoin/network/define.hpp>
22	#include <bitcoin/network/tls/constants.hpp>
23
24	namespace libbitcoin {
25	namespace network {
26	namespace tls {
27
28	using namespace system;
29	using namespace std::placeholders;
30
31	BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT)
32
33	// Constructor.
34	// ----------------------------------------------------------------------------
35
36	stream::stream(io::socket&& socket, const context& context) NOEXCEPT	11 ✔
37	: socket_(std::move(socket)),	11 ✔
38	context_(context),	11 ✔
39	server_(context),	11 ✔
40	input_(record_header_size + maximum_ciphertext)	11 ✔
41	{
42	}	11 ✔
43
44	// Properties.
45	// ----------------------------------------------------------------------------
46
47	stream::executor_type stream::get_executor() NOEXCEPT	×
48	{
49	return socket_.get_executor();	×
50	}
51
52	io::socket& stream::next_layer() NOEXCEPT	12 ✔
53	{
54	return socket_;	12 ✔
55	}
56
57	const io::socket& stream::next_layer() const NOEXCEPT	×
58	{
59	return socket_;	×
60	}
61
62	// Handshake.
63	// ----------------------------------------------------------------------------
64
65	void stream::async_handshake(result_handler&& handler) NOEXCEPT	11 ✔
66	{
67	if (!context_.is_ready())	11 ✔
68	{
69	const code ec{ code{ error::tls_alert_internal_error } };	×
70	io::post(get_executor(), std::bind(std::move(handler), ec));	×
71	return;
72	}
73
74	receive(std::bind(&stream::handle_handshake,	11 ✔
75	this, _1, std::move(handler)));	22 ✔
76	}
77
78	void stream::handle_handshake(const code& ec,	22 ✔
79	const result_handler& handler) NOEXCEPT
80	{
81	if (ec)	22 ✔
82	{
83	handler(ec);	1 ✔
84	return;	1 ✔
85	}
86
87	if (server_.is_failed())	21 ✔
88	{
89	handler(failure());	1 ✔
90	return;	1 ✔
91	}
92
93	if (server_.is_closed())	20 ✔
94	{
95	handler(error::peer_disconnect);	×
96	return;	×
97	}
98
99	if (server_.is_established())	20 ✔
100	{
101	handler(error::success);	9 ✔
102	return;	9 ✔
103	}
104
105	receive(std::bind(&stream::handle_handshake,	11 ✔
106	this, _1, handler));	22 ✔
107	}
108
109	// Read.
110	// ----------------------------------------------------------------------------
111
112	void stream::do_read_some(const byte_span& out,	5 ✔
113	const io_handler& handler) NOEXCEPT
114	{
115	code ec{};	5 ✔
116	size_t size{};	5 ✔
117	if (try_read(ec, size, out))	5 ✔
118	{
119	io::post(get_executor(), std::bind(handler, ec, size));	×
120	return;	×
121	}
122
123	receive(std::bind(&stream::handle_read_some,	5 ✔
124	this, _1, out, handler));	10 ✔
125	}
126
127	// True if the read is complete without receiving.
128	bool stream::try_read(code& ec, size_t& size,	9 ✔
129	const byte_span& out) NOEXCEPT
130	{
131	ec = error::success;	9 ✔
132	size = zero;	9 ✔
133	if (out.empty())	9 ✔
134	return true;
135
136	if (is_nonzero(server_.readable()))	9 ✔
137	{
138	size = server_.read(out);	2 ✔
139	return true;	2 ✔
140	}
141
142	if (server_.is_closed())	7 ✔
143	{
144	ec = error::peer_disconnect;	1 ✔
145	return true;	1 ✔
146	}
147
148	if (server_.is_failed())	6 ✔
149	{
150	ec = failure();	1 ✔
151	return true;	1 ✔
152	}
153
154	return false;
155	}
156
157	void stream::handle_read_some(const code& ec, const byte_span& out,	5 ✔
158	const io_handler& handler) NOEXCEPT
159	{
160	if (ec)	5 ✔
161	{
162	handler(ec, zero);	1 ✔
163	return;	5 ✔
164	}
165
166	code result{};	4 ✔
167	size_t size{};	4 ✔
168	if (try_read(result, size, out))	4 ✔
169	{
170	handler(result, size);	4 ✔
171	return;	4 ✔
172	}
173
174	receive(std::bind(&stream::handle_read_some,	×
175	this, _1, out, handler));	×
176	}
177
178	// Write.
179	// ----------------------------------------------------------------------------
180
181	void stream::do_write_some(const const_byte_span& data,	10 ✔
182	const io_handler& handler) NOEXCEPT
183	{
184	if (data.empty())	10 ✔
185	{
186	io::post(get_executor(),	×
187	std::bind(handler, code{ error::success }, zero));	×
188	return;	×
189	}
190
191	const auto size = std::min(data.size(), maximum_plaintext);	10 ✔
192	if (!server_.write(data.first(size)))	10 ✔
193	{
194	const auto ec = server_.is_failed() ? failure() :	×
195	code{ error::connect_failed };	×
196
197	io::post(get_executor(), std::bind(handler, ec, zero));	×
198	return;
199	}
200
201	flush([handler, size](const code& ec) NOEXCEPT	10 ✔
202	{
203	handler(ec, ec ? zero : size);	10 ✔
204	});	10 ✔
205	}
206
207	size_t stream::do_write(const const_byte_span& data, code& ec) NOEXCEPT	×
208	{
209	ec = error::success;	×
210	if (writing_)	×
211	{
212	ec = error::operation_failed;	×
213	return zero;	×
214	}
215
216	if (data.empty())	×
217	return zero;
218
219	const auto size = std::min(data.size(), maximum_plaintext);	×
220	if (!server_.write(data.first(size)))	×
221	{
222	ec = server_.is_failed() ? failure() : code{ error::connect_failed };	×
223	return zero;	×
224	}
225
226	auto& output = server_.output();	×
227	io::write(socket_, io::buffer(output), ec);	×
228	output.clear();	×
229	return ec ? zero : size;	×
230	}
231
232	// Shutdown.
233	// ----------------------------------------------------------------------------
234
UNCOV 235	void stream::async_shutdown(result_handler&& handler) NOEXCEPT	×
236	{
UNCOV 237	if (server_.is_failed())	×
238	{
239	io::post(get_executor(),	×
240	std::bind(std::move(handler), code{ error::success }));	×
241	return;	×
242	}
243
UNCOV 244	server_.close();	×
UNCOV 245	flush(std::bind(&stream::handle_close_sent,	×
UNCOV 246	this, _1, std::move(handler)));	×
247	}
248
UNCOV 249	void stream::handle_close_sent(const code& ec,	×
250	const result_handler& handler) NOEXCEPT
251	{
UNCOV 252	if (ec)	×
253	{
254	handler(ec);	×
255	return;	×
256	}
257
UNCOV 258	handle_close_received(ec, handler);	×
259	}
260
UNCOV 261	void stream::handle_close_received(const code& ec,	×
262	const result_handler& handler) NOEXCEPT
263	{
UNCOV 264	if (ec)	×
265	{
UNCOV 266	handler(ec);	×
UNCOV 267	return;	×
268	}
269
UNCOV 270	if (server_.is_failed())	×
271	{
272	handler(failure());	×
273	return;	×
274	}
275
UNCOV 276	if (server_.is_closed())	×
277	{
278	handler(error::success);	×
279	return;	×
280	}
281
UNCOV 282	receive(std::bind(&stream::handle_close_received,	×
UNCOV 283	this, _1, handler));	×
284	}
285
286	// Transport.
287	// ----------------------------------------------------------------------------
288
289	code stream::failure() const NOEXCEPT	2 ✔
290	{
291	return code{ error::tls_to_error_code(server_.failure()) };	2 ✔
292	}
293
294	// Closure of the transport without close_notify truncates the stream.
295	code stream::transport(const code& ec) const NOEXCEPT	2 ✔
296	{
297	if (ec == error::peer_disconnect && !server_.is_closed())	2 ✔
298	return code{ error::tls_stream_truncated };	1 ✔
299
300	return ec;	1 ✔
301	}
302
303	// A receive joins any receive in progress, as they share the input buffer.
304	void stream::receive(result_handler&& handler) NOEXCEPT	27 ✔
305	{
306	receivers_.push_back(std::move(handler));	27 ✔
307	if (receiving_)	27 ✔
308	return;
309
310	receiving_ = true;	27 ✔
311	socket_.async_read_some(io::buffer(input_),	27 ✔
312	std::bind(&stream::handle_receive,	54 ✔
313	this, _1, _2));	54 ✔
314	}
315
316	// Records produced by receipt (alerts and key updates) are sent first.
317	void stream::handle_receive(const code& ec, size_t size) NOEXCEPT	27 ✔
318	{
319	receiving_ = false;	27 ✔
320	std::vector<result_handler> handlers{};	27 ✔
321	handlers.swap(receivers_);	27 ✔
322
323	if (ec)	27 ✔
324	{
325	const auto result = transport(ec);	2 ✔
326	for (const auto& handler: handlers)	4 ✔
327	handler(result);	2 ✔
328
329	return;
330	}
331
332	server_.receive({ input_.data(), size });	25 ✔
333	if (server_.output().empty())	25 ✔
334	{
335	for (const auto& handler: handlers)	22 ✔
336	handler(error::success);	11 ✔
337
338	return;	11 ✔
339	}
340
341	for (const auto& handler: handlers)	28 ✔
342	flush(handler);	14 ✔
343	}	27 ✔
344
345	// The handler completes once all output preceding it is written.
346	void stream::flush(const result_handler& handler) NOEXCEPT	24 ✔
347	{
348	waiting_.push_back(handler);	24 ✔
349	if (!writing_)	24 ✔
350	write_next();	24 ✔
351	}	24 ✔
352
353	void stream::write_next() NOEXCEPT	24 ✔
354	{
355	std::vector<result_handler> handlers{};	24 ✔
356	handlers.swap(waiting_);	24 ✔
357
358	auto& output = server_.output();	24 ✔
359	if (output.empty())	24 ✔
360	{
361	for (auto& handler: handlers)	×
362	io::post(get_executor(),	×
363	std::bind(std::move(handler), code{ error::success }));	×
364
365	return;
366	}
367
368	writing_ = true;	24 ✔
369	sending_.swap(output);	24 ✔
370	output.clear();	24 ✔
371	io::async_write(socket_, io::buffer(sending_),	24 ✔
372	std::bind(&stream::handle_write,	24 ✔
373	this, _1, std::move(handlers)));	24 ✔
374	}	24 ✔
375
376	void stream::handle_write(const code& ec,	24 ✔
377	const std::vector<result_handler>& handlers) NOEXCEPT
378	{
379	writing_ = false;	24 ✔
380	sending_.clear();	24 ✔
381	for (const auto& handler: handlers)	48 ✔
382	handler(ec);	24 ✔
383
384	if (!writing_ && !waiting_.empty())	24 ✔
385	write_next();	×
386	}	24 ✔
387
388	BC_POP_WARNING()
389
390	} // namespace tls
391	} // namespace network
392	} // namespace libbitcoin

libbitcoin / libbitcoin-network / 42421988994

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous