ADC/net/adc_netservice_asio.h
Timur A. Fatkhullin ebe93f1d74 ...
2024-06-17 23:39:58 +03:00

737 lines
27 KiB
C++

#pragma once
/*
ABSTRACT DEVICE COMPONENTS LIBRARY
ASIO-library implementation of network service
*/
#ifdef USE_ASIO_LIBRARY
#include <future>
#include "adc_netservice.h"
#include <asio/basic_datagram_socket.hpp>
#include <asio/basic_seq_packet_socket.hpp>
#include <asio/basic_stream_socket.hpp>
#include <asio/compose.hpp>
#include <asio/ip/tcp.hpp>
#include <asio/ip/udp.hpp>
#include <asio/local/seq_packet_protocol.hpp>
#include <asio/local/stream_protocol.hpp>
#include <asio/read_until.hpp>
#include <asio/steady_timer.hpp>
#include <asio/streambuf.hpp>
#include <asio/use_future.hpp>
#include <asio/write.hpp>
#ifdef USE_OPENSSL_WITH_ASIO
#include <asio/ssl.hpp>
#include <asio/ssl/stream.hpp>
#endif
#include <concepts>
#include "adc_netmsg.h"
namespace adc::traits
{
// still only TCP, UDP and UNIX
template <typename T>
concept adc_asio_inet_proto_c =
std::derived_from<T, asio::ip::tcp> || std::derived_from<T, asio::ip::udp> ||
std::derived_from<T, asio::local::seq_packet_protocol> || std::derived_from<T, asio::local::stream_protocol>;
template <typename T>
concept adc_asio_inet_stream_proto_c =
std::derived_from<T, asio::ip::tcp> || std::derived_from<T, asio::local::stream_protocol>;
} // namespace adc::traits
namespace adc::impl
{
template <traits::adc_asio_inet_proto_c InetProtoT>
class AdcNetServiceASIO : public InetProtoT
{
public:
using socket_t = typename InetProtoT::socket;
using endpoint_t = typename InetProtoT::endpoint;
using acceptor_t =
std::conditional_t<std::derived_from<socket_t, asio::basic_datagram_socket<typename socket_t::protocol_type>>,
std::nullptr_t,
typename InetProtoT::acceptor>;
using inet_proto_t = InetProtoT;
typedef std::chrono::steady_clock::duration timeout_t; // nanoseconds resolution
using streambuff_iter_t = asio::buffers_iterator<asio::streambuf::const_buffers_type>;
AdcNetServiceASIO(asio::io_context& io_context) : _ioContext(io_context), _socket(io_context), _acceptor(io_context)
{
}
virtual ~AdcNetServiceASIO() = default;
template <traits::adc_time_duration_c TimeoutT, asio::completion_token_for<void(std::error_code)> CompletionTokenT>
auto asyncAccept(const endpoint_t& endpoint, const TimeoutT& timeout, CompletionTokenT&& token)
{
// no acceptor for UDP-sockets
if constexpr (!std::is_null_pointer_v<acceptor_t>) {
return;
}
auto timer = getDeadlineTimer(timeout);
enum { starting, finishing };
// wrapper
return asio::async_compose<CompletionTokenT, void(std::error_code)>(
[timer = std::move(timer), state = starting, &endpoint, this](auto& self,
const std::error_code& ec = {}) mutable {
if (!ec) {
switch (state) {
case starting:
state = finishing;
try {
_acceptor = acceptor_t(_ioContext, endpoint);
} catch (std::system_error err) {
timer->cancel();
self.complete(err.code());
return;
}
return _acceptor.async_accept(_socket, std::move(self));
break;
case finishing:
break;
default:
break;
}
}
timer->cancel();
self.complete(ec);
},
token, _socket);
}
template <traits::adc_time_duration_c TimeoutT, asio::completion_token_for<void(std::error_code)> CompletionTokenT>
auto asyncConnect(const endpoint_t& endpoint, const TimeoutT& timeout, CompletionTokenT&& token)
{
auto timer = getDeadlineTimer(timeout);
enum { starting, finishing };
// wrapper
return asio::async_compose<CompletionTokenT, void(std::error_code)>(
[timer = std::move(timer), state = starting, &endpoint, this](auto& self,
const std::error_code& ec = {}) mutable {
if (!ec) {
switch (state) {
case starting:
state = finishing;
return _socket.async_connect(endpoint, std::move(self));
break;
case finishing:
break;
default:
break;
}
}
timer->cancel();
self.complete(ec);
},
token, _socket);
}
template <traits::adc_netmessage_c NetMessageT,
traits::adc_time_duration_c TimeoutT,
asio::completion_token_for<void(std::error_code)> CompletionTokenT>
auto asyncSend(const NetMessageT& msg, const TimeoutT& timeout, CompletionTokenT&& token)
{
enum { starting, finishing };
// create buffer sequence
std::vector<asio::const_buffer> buff;
std::ranges::for_each(msg.template bytesView<std::vector<std::string_view>>(),
[&buff](const auto& el) { buff.emplace_back(el); });
auto timer = getDeadlineTimer(timeout);
// wrapper
return asio::async_compose<CompletionTokenT, void(std::error_code)>(
[buff = std::move(buff), timer = std::move(timer), state = starting, this](
auto& self, const std::error_code& ec = {}, size_t = 0) mutable {
if (!ec) {
switch (state) {
case starting:
state = finishing;
if constexpr (std::derived_from<
socket_t, asio::basic_stream_socket<typename socket_t::protocol_type>>) {
return asio::async_write(_socket, buff, std::move(self));
} else if constexpr (std::derived_from<socket_t, asio::basic_datagram_socket<
typename socket_t::protocol_type>>) {
return _socket.async_send(buff, std::move(self));
} else if constexpr (std::derived_from<socket_t, asio::basic_seq_packet_socket<
typename socket_t::protocol_type>>) {
return _socket.async_send(buff, std::move(self));
} else {
static_assert(false, "UNKNOWN ASIO-LIBRARY SOCKET TYPE!!!");
}
break;
case finishing:
timer->cancel();
break;
default:
break;
}
}
self.complete(ec);
},
token, _socket);
}
template <traits::adc_netmessage_c NetMessageT, traits::adc_time_duration_c TimeoutT, typename CompletionTokenT>
auto asyncReceive(const TimeoutT& timeout, CompletionTokenT&& token)
{
enum { starting, finishing };
std::unique_ptr<asio::socket_base::message_flags> out_flags;
auto timer = getDeadlineTimer(timeout); // armed timer
return asio::async_compose<CompletionTokenT, void(const std::error_code&, const NetMessageT&)>(
[timer = std::move(timer), out_flags = std::move(out_flags), state = starting, this](
auto& self, const std::error_code& ec = {}, size_t = 0) mutable {
if (!ec) {
switch (state) {
case starting:
state = finishing;
if constexpr (std::derived_from<
socket_t, asio::basic_stream_socket<typename socket_t::protocol_type>>) {
return asio::async_read_until(
_socket, _streamBuffer,
[this](auto begin, auto end) { return this->matchCondition(begin, end); },
std::move(self));
} else if constexpr (std::derived_from<socket_t, asio::basic_datagram_socket<
typename socket_t::protocol_type>>) {
return _socket.receive(_streamBuffer,
std::move(self)); // datagram, so it should be received at once
} else if constexpr (std::derived_from<socket_t, asio::basic_seq_packet_socket<
typename socket_t::protocol_type>>) {
return _socket.receive(_streamBuffer, *out_flags,
std::move(self)); // datagram, so it should be received at once
} else {
static_assert(false, "UNKNOWN ASIO-LIBRARY SOCKET TYPE!!!");
}
break;
case finishing:
timer->cancel();
break;
default:
break;
}
auto begin_it = streambuff_iter_t::begin(_streamBuffer.data());
auto end_it = begin_it + _streamBuffer.data().size();
// check for byte sequence is valid byte sequence and find the limits
// (stream buffer may contain number of bytes more than requred by protocol)
auto res = this->matchCondition(begin_it, end_it);
if (!res.second) {
self.complete(std::make_error_code(std::errc::protocol_error),
NetMessageT()); // return an empty message
} else {
auto nbytes = std::distance(begin_it, res.first);
NetMessageT msg;
auto msg_it = this->fromLowLevel(begin_it, res.first);
msg.setFromBytes(msg_it.first, msg_it.second);
_streamBuffer.consume(nbytes);
self.complete(ec, msg);
}
} else {
self.complete(ec, NetMessageT()); // return an empty message
return;
}
},
token, _socket);
}
template <traits::adc_time_duration_c TimeoutT>
auto connect(const endpoint_t& endpoint, const TimeoutT& timeout)
{
std::future<void> ftr = asyncConnect(endpoint, timeout, asio::use_future);
ftr.get();
}
template <traits::adc_netmessage_c NetMessageT, traits::adc_time_duration_c TimeoutT>
auto send(const NetMessageT& msg, const TimeoutT& timeout)
{
std::future<void> ftr = asyncSend(msg, timeout, asio::use_future);
ftr.get();
}
template <traits::adc_netmessage_c NetMessageT, traits::adc_time_duration_c TimeoutT>
auto receive(const TimeoutT& timeout)
{
std::future<NetMessageT> ftr = asyncReceive(timeout, asio::use_future);
return ftr.get();
}
std::error_code close(asio::socket_base::shutdown_type stype = asio::socket_base::shutdown_both)
{
std::error_code ec;
_socket.shutdown(stype, ec);
if (!ec) {
_socket.close(ec);
}
return ec;
}
protected:
asio::io_context& _ioContext;
socket_t _socket;
acceptor_t _acceptor;
asio::streambuf _streamBuffer;
template <traits::adc_time_duration_c TimeoutT>
std::unique_ptr<asio::steady_timer> getDeadlineTimer(const TimeoutT& timeout, bool arm = true)
{
std::unique_ptr<asio::steady_timer> timer(_socket.get_executor());
if (arm) {
timer->expires_after(std::chrono::duration_cast<timeout_t>(timeout));
timer->async_wait([this](const std::error_code& ec) {
if (!ec) {
_socket.cancel(std::make_error_code(std::errc::timed_out));
}
});
}
return timer;
}
};
#ifdef USE_OPENSSL_WITH_ASIO
template <traits::adc_asio_inet_stream_proto_c InetProtoT = asio::ip::tcp>
class AdcNetServiceAsioTls : public InetProtoT
{
public:
using socket_t = typename InetProtoT::socket;
using stream_t = asio::ssl::stream<socket_t>;
using endpoint_t = typename InetProtoT::endpoint;
using acceptor_t = typename InetProtoT::acceptor;
using inet_proto_t = InetProtoT;
typedef std::chrono::steady_clock::duration timeout_t; // nanoseconds resolution
// TLS certificate attributes comparison function:
// 'serial' - as returned by OpenSSL BN_bn2hex
// 'fingerprint' - as returned by OpenSSL X509_digest
// 'depth' - depth in chain
// the function must return 0 - if comparison failed; otherwise - something != 0
typedef std::function<int(const std::string& serial, const std::vector<unsigned char>& fingerprint, int depth)>
cert_comp_func_t;
using streambuff_iter_t = asio::buffers_iterator<asio::streambuf::const_buffers_type>;
AdcNetServiceAsioTls(asio::io_context& io_context,
asio::ssl::context&& tls_context = asio::ssl::context(asio::ssl::context::tlsv13),
asio::ssl::verify_mode verify_mode = asio::ssl::verify_peer)
: _ioContext(io_context),
_tlsStream(socket_t(io_context), io_context),
_tlsContext(std::move(tls_context)),
_acceptor(io_context),
_tlsPeerVerifyMode(verify_mode),
_tlsCertFingerprintDigest("sha256"),
_tlsCertCompFunc(nullptr)
{
}
template <traits::adc_time_duration_c TimeoutT, asio::completion_token_for<void(std::error_code)> CompletionTokenT>
auto asyncAccept(const endpoint_t& endpoint, const TimeoutT& timeout, CompletionTokenT&& token)
{
auto sock = std::make_unique<socket_t>(_ioContext);
auto timer = getDeadlineTimer(timeout);
enum { starting, handshaking, finishing };
// wrapper
return asio::async_compose<CompletionTokenT, void(std::error_code)>(
[timer = std::move(timer), sock = std::move(sock), state = starting, &endpoint, this](
auto& self, std::error_code ec = {}) mutable {
if (!ec) {
switch (state) {
case starting:
state = handshaking;
try {
_acceptor = acceptor_t(_ioContext, endpoint);
return _acceptor.async_accept(*sock, std::move(self));
} catch (std::system_error err) {
ec = err.code();
}
break;
case handshaking:
state = finishing;
_tlsStream = stream_t(std::move(sock), _tlsContext);
_tlsStream->set_verify_mode(_tlsPeerVerifyMode, ec);
if (!ec) {
_tlsStream->set_verify_callback(
[this](bool preverified, asio::ssl::verify_context& ctx) {
return verifyCertificate(preverified ? 1 : 0, ctx.native_handle());
});
return _tlsStream->async_handshake(asio::ssl::stream_base::server, std::move(self));
}
break;
case finishing:
break;
default:
break;
}
}
timer->cancel();
self.complete(ec);
},
token, _ioContext);
}
template <traits::adc_time_duration_c TimeoutT, asio::completion_token_for<void(std::error_code)> CompletionTokenT>
auto asyncConnect(const endpoint_t& endpoint, const TimeoutT& timeout, CompletionTokenT&& token)
{
auto sock = std::make_unique<socket_t>(_ioContext);
auto timer = getDeadlineTimer(timeout);
enum { starting, handshaking, finishing };
// wrapper
return asio::async_compose<CompletionTokenT, void(std::error_code)>(
[timer = std::move(timer), sock = std::move(sock), state = starting, &endpoint, this](
auto& self, std::error_code ec = {}) mutable {
if (!ec) {
switch (state) {
case starting:
state = handshaking;
return sock->async_connect(endpoint, std::move(self));
break;
case handshaking:
state = finishing;
_tlsStream = stream_t(std::move(*sock), _tlsContext);
return _tlsStream.async_handshake(asio::ssl::stream_base::client, std::move(self));
break;
case finishing:
break;
default:
break;
}
}
timer->cancel();
self.complete(ec);
},
token, _ioContext);
}
template <traits::adc_netmessage_c NetMessageT,
traits::adc_time_duration_c TimeoutT,
asio::completion_token_for<void(std::error_code)> CompletionTokenT>
auto asyncSend(const NetMessageT& msg, const TimeoutT& timeout, CompletionTokenT&& token)
{
enum { starting, finishing };
// create buffer sequence
std::vector<asio::const_buffer> buff;
std::ranges::for_each(msg.template bytesView<std::vector<std::string_view>>(),
[&buff](const auto& el) { buff.emplace_back(el); });
auto timer = getDeadlineTimer(timeout);
// wrapper
return asio::async_compose<CompletionTokenT, void(std::error_code)>(
[buff = std::move(buff), timer = std::move(timer), state = starting, this](
auto& self, const std::error_code& ec = {}, size_t = 0) mutable {
if (!ec) {
switch (state) {
case starting:
state = finishing;
return asio::async_write(_tlsStream, buff, std::move(self));
break;
case finishing:
timer->cancel();
break;
default:
break;
}
}
self.complete(ec);
},
token, _ioContext);
}
template <traits::adc_netmessage_c NetMessageT, traits::adc_time_duration_c TimeoutT, typename CompletionTokenT>
auto asyncReceive(const TimeoutT& timeout, CompletionTokenT&& token)
{
enum { starting, finishing };
std::unique_ptr<asio::socket_base::message_flags> out_flags;
auto timer = getDeadlineTimer(timeout); // armed timer
return asio::async_compose<CompletionTokenT, void(const std::error_code&, const NetMessageT&)>(
[timer = std::move(timer), out_flags = std::move(out_flags), state = starting, this](
auto& self, const std::error_code& ec = {}, size_t = 0) mutable {
if (!ec) {
switch (state) {
case starting:
state = finishing;
return asio::async_read_until(
_tlsStream, _streamBuffer,
[this](auto begin, auto end) { return this->matchCondition(begin, end); },
std::move(self));
break;
case finishing:
timer->cancel();
break;
default:
break;
}
auto begin_it = streambuff_iter_t::begin(_streamBuffer.data());
auto end_it = begin_it + _streamBuffer.data().size();
// check for byte sequence is valid byte sequence and find the limits
// (stream buffer may contain number of bytes more than requred by protocol)
auto res = this->matchCondition(begin_it, end_it);
if (!res.second) {
self.complete(std::make_error_code(std::errc::protocol_error),
NetMessageT()); // return an empty message
} else {
auto nbytes = std::distance(begin_it, res.first);
NetMessageT msg;
auto msg_it = this->fromLowLevel(begin_it, res.first);
msg.setFromBytes(msg_it.first, msg_it.second);
_streamBuffer.consume(nbytes);
self.complete(ec, msg);
}
} else {
self.complete(ec, NetMessageT()); // return an empty message
return;
}
},
token, _ioContext);
}
template <traits::adc_time_duration_c TimeoutT>
auto connect(const endpoint_t& endpoint, const TimeoutT& timeout)
{
std::future<void> ftr = asyncConnect(endpoint, timeout, asio::use_future);
ftr.get();
}
template <traits::adc_netmessage_c NetMessageT, traits::adc_time_duration_c TimeoutT>
auto send(const NetMessageT& msg, const TimeoutT& timeout)
{
std::future<void> ftr = asyncSend(msg, timeout, asio::use_future);
ftr.get();
}
template <traits::adc_netmessage_c NetMessageT, traits::adc_time_duration_c TimeoutT>
auto receive(const TimeoutT& timeout)
{
std::future<NetMessageT> ftr = asyncReceive(timeout, asio::use_future);
return ftr.get();
}
std::error_code close(asio::socket_base::shutdown_type stype = asio::socket_base::shutdown_both)
{
std::error_code ec;
_tlsStream.shutdown(ec); // shutdown OpenSSL stream
if (!ec) {
_tlsStream.lowest_layer().shutdown(stype, ec);
if (!ec) {
_tlsStream.lowest_layer().close(ec);
}
}
return ec;
}
// special TLS-related methods
void setPeerVerifyMode(asio::ssl::verify_mode mode = asio::ssl::verify_peer)
{
// restart TLS server?!!
_tlsPeerVerifyMode = mode;
}
asio::ssl::verify_mode getPeerVerifyMode() const
{
return _tlsPeerVerifyMode;
}
AdcNetServiceAsioTls& setTLSCertFingerprintDigest(const std::string& digest = "sha256")
{
// check for validness?!!
_tlsCertFingerprintDigest = digest;
}
std::string getTLSCertFingerprintDigest() const
{
return _tlsCertFingerprintDigest;
}
template <std::invocable<const std::string&, const std::vector<unsigned char>&, int> FuncT>
AdcNetServiceAsioTls& setTLSCertCompFunc(FuncT&& func)
{
_tlsCertCompFunc = static_cast<cert_comp_func_t>(std::forward<FuncT>(func));
}
protected:
asio::io_context& _ioContext;
acceptor_t _acceptor;
stream_t _tlsStream;
asio::ssl::context _tlsContext;
asio::ssl::verify_mode _tlsPeerVerifyMode;
std::string _tlsCertFingerprintDigest;
cert_comp_func_t _tlsCertCompFunc;
asio::streambuf _streamBuffer;
// reference implementation
virtual bool verifyCertificate(int preverified_ok, X509_STORE_CTX* store)
{
if (preverified_ok == 0) {
int err = X509_STORE_CTX_get_error(store);
auto err_str = X509_verify_cert_error_string(err);
// log_error("TLS certificate verification error: {}", err_str);
return preverified_ok;
}
char subject_name[256];
int depth;
ASN1_INTEGER* serial;
BIGNUM* bnser;
X509* cert = X509_STORE_CTX_get_current_cert(store);
if (cert != NULL) {
depth = X509_STORE_CTX_get_error_depth(store);
X509_NAME_oneline(X509_get_subject_name(cert), subject_name, 256);
serial = X509_get_serialNumber(cert); // IT IS INTERNAL POINTER SO IT MUST NOT BE FREED UP!!!
bnser = ASN1_INTEGER_to_BN(serial, NULL);
auto serial_hex = BN_bn2hex(bnser);
// log_debug("Received TLS certificate: SUBJECT = {}, SERIAL = {}, DEPTH = {}", subject_name, serial_hex,
// depth);
// if no compare function then do not compute fingerprint
if (_tlsCertCompFunc) {
// compute certificate fingerprint
unsigned char digest_buff[EVP_MAX_MD_SIZE];
const EVP_MD* digest = EVP_get_digestbyname(_tlsCertFingerprintDigest.c_str());
unsigned int N;
if (X509_digest(cert, digest, digest_buff, &N)) {
preverified_ok = _tlsCertCompFunc(std::string(serial_hex),
std::vector<unsigned char>(digest_buff, digest_buff + N), depth);
} else {
// log_error("Cannot compute client certificate fingerprint! Cannot verify the certificate!");
preverified_ok = 0;
}
}
BN_free(bnser);
OPENSSL_free(serial_hex);
} else {
// log_error("OpenSSL error: cannot get current certificate");
preverified_ok = 0;
}
return preverified_ok;
}
};
#endif
typedef AdcNetService<impl::AdcNetServiceASIO<asio::ip::tcp>> AdcNetServiceAsioTcp;
typedef AdcNetService<impl::AdcNetServiceASIO<asio::ip::udp>> AdcNetServiceAsioUdp;
typedef AdcNetService<impl::AdcNetServiceASIO<asio::local::seq_packet_protocol>> AdcNetServiceAsioLocalSeqPack;
typedef AdcNetService<impl::AdcNetServiceASIO<asio::local::stream_protocol>> AdcNetServiceAsioLocalStream;
} // namespace adc::impl
namespace adc::traits
{
template <typename T>
concept adc_netservice_asio_c = requires {
typename T::inet_proto_t;
#ifdef USE_OPENSSL_WITH_ASIO
requires std::derived_from<T, adc::impl::AdcNetServiceASIO<typename T::inet_proto_t>> ||
std::derived_from<T, adc::impl::AdcNetServiceAsioTls<typename T::inet_proto_t>>;
#else
requires std::derived_from<T, adc::impl::AdcNetServiceASIO<typename T::inet_proto_t>>;
#endif
};
} // namespace adc::traits
#endif