Commit 3d232106 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/neverlord/caf-net-additions'

parents a0c97282 617955aa
......@@ -19,8 +19,12 @@
// -- convenience type aliases -------------------------------------------------
// The trait for translating between bytes on the wire and flow items. The
// binary default trait operates on binary::frame items.
using trait = caf::net::binary::default_trait;
// Takes care converting a byte stream into a sequence of messages on the wire.
using lpf = caf::net::length_prefix_framing;
using lpf = caf::net::length_prefix_framing::bind<trait>;
// An implicitly shared type for storing a binary frame.
using bin_frame = caf::net::binary::frame;
......
......@@ -17,8 +17,12 @@
// -- convenience type aliases -------------------------------------------------
// The trait for translating between bytes on the wire and flow items. The
// binary default trait operates on binary::frame items.
using trait = caf::net::binary::default_trait;
// Takes care converting a byte stream into a sequence of messages on the wire.
using lpf = caf::net::length_prefix_framing;
using lpf = caf::net::length_prefix_framing::bind<trait>;
// An implicitly shared type for storing a binary frame.
using bin_frame = caf::net::binary::frame;
......
......@@ -75,7 +75,8 @@ int caf_main(actor_system& sys, const config& cfg) {
auto [lpf_pull, app_push] = make_spsc_buffer_resource<bin_frame>();
auto [app_pull, lpf_push] = make_spsc_buffer_resource<bin_frame>();
// Spin up the network backend.
using lpf = caf::net::length_prefix_framing;
using trait = caf::net::binary::default_trait;
using lpf = caf::net::length_prefix_framing::bind<trait>;
auto conn = lpf::run(sys, *fd, std::move(lpf_pull), std::move(lpf_push));
// Spin up Qt.
auto [argc, argv] = cfg.c_args_remainder();
......
......@@ -140,6 +140,11 @@ public:
impl_->cancel();
}
consumer_adapter& operator=(std::nullptr_t) {
impl_ = nullptr;
return *this;
}
template <class Policy>
read_result pull(Policy policy, T& result) {
if (impl_)
......
......@@ -107,6 +107,11 @@ public:
// nop
}
producer_adapter& operator=(std::nullptr_t) {
impl_ = nullptr;
return *this;
}
~producer_adapter() {
if (impl_)
impl_->close();
......
......@@ -607,6 +607,9 @@ auto observable<T>::merge(Inputs&&... xs) {
static_assert(
sizeof...(Inputs) > 0,
"merge without arguments expects this observable to emit observables");
static_assert(
(std::is_same_v<Out, output_type_t<std::decay_t<Inputs>>> && ...),
"can only merge observables with the same observed type");
using impl_t = op::merge<Out>;
return make_observable<impl_t>(ctx(), *this, std::forward<Inputs>(xs)...);
}
......@@ -622,7 +625,10 @@ auto observable<T>::concat(Inputs&&... xs) {
} else {
static_assert(
sizeof...(Inputs) > 0,
"merge without arguments expects this observable to emit observables");
"concat without arguments expects this observable to emit observables");
static_assert(
(std::is_same_v<Out, output_type_t<std::decay_t<Inputs>>> && ...),
"can only concatenate observables with the same observed type");
using impl_t = op::concat<Out>;
return make_observable<impl_t>(ctx(), *this, std::forward<Inputs>(xs)...);
}
......
......@@ -67,9 +67,11 @@ caf_add_component(
src/net/ssl/context.cpp
src/net/ssl/dtls.cpp
src/net/ssl/format.cpp
src/net/ssl/password.cpp
src/net/ssl/startup.cpp
src/net/ssl/tls.cpp
src/net/ssl/transport.cpp
src/net/ssl/verify.cpp
src/net/stream_oriented.cpp
src/net/stream_socket.cpp
src/net/stream_transport.cpp
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/async/consumer_adapter.hpp"
#include "caf/async/producer_adapter.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/fwd.hpp"
#include "caf/net/flow_connector.hpp"
#include "caf/sec.hpp"
#include <utility>
namespace caf::detail {
/// Translates between a message-oriented transport and data flows.
template <class UpperLayer, class LowerLayer, class Trait>
class flow_bridge_base : public UpperLayer {
public:
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
/// Type for the consumer adapter. We consume the output of the application.
using consumer_type = async::consumer_adapter<output_type>;
/// Type for the producer adapter. We produce the input of the application.
using producer_type = async::producer_adapter<input_type>;
using connector_pointer = net::flow_connector_ptr<Trait>;
flow_bridge_base(async::execution_context_ptr loop, connector_pointer conn)
: loop_(std::move(loop)), conn_(std::move(conn)) {
// nop
}
virtual bool write(const output_type& item) = 0;
bool running() const noexcept {
return in_ || out_;
}
void self_ref(disposable ref) {
self_ref_ = std::move(ref);
}
// -- implementation of the lower_layer --------------------------------------
error start(LowerLayer* down, const settings& cfg) override {
CAF_ASSERT(down != nullptr);
down_ = down;
auto [err, pull, push] = conn_->on_request(cfg);
if (!err) {
auto do_wakeup = make_action([this] {
if (running())
prepare_send();
});
auto do_resume = make_action([this] { down_->request_messages(); });
auto do_cancel = make_action([this] {
if (!running()) {
down_->shutdown();
}
});
in_ = consumer_type::make(pull.try_open(), loop_, std::move(do_wakeup));
out_ = producer_type::make(push.try_open(), loop_, std::move(do_resume),
std::move(do_cancel));
conn_ = nullptr;
if (running())
return none;
else
return make_error(sec::runtime_error,
"cannot init flow bridge: no buffers");
} else {
conn_ = nullptr;
return err;
}
}
void prepare_send() override {
input_type tmp;
while (down_->can_send_more()) {
switch (in_.pull(async::delay_errors, tmp)) {
case async::read_result::ok:
if (!write(tmp)) {
abort(trait_.last_error());
down_->shutdown(trait_.last_error());
return;
}
break;
case async::read_result::stop:
in_ = nullptr;
abort(error{});
down_->shutdown();
return;
case async::read_result::abort:
in_ = nullptr;
abort(in_.abort_reason());
down_->shutdown(in_.abort_reason());
return;
default: // try later
return;
}
}
}
bool done_sending() override {
return !in_.has_consumer_event();
}
void abort(const error& reason) override {
CAF_LOG_TRACE(CAF_ARG(reason));
if (out_) {
if (!reason || reason == sec::connection_closed
|| reason == sec::socket_disconnected || reason == sec::disposed)
out_.close();
else
out_.abort(reason);
out_ = nullptr;
}
if (in_) {
in_.cancel();
in_ = nullptr;
}
self_ref_ = nullptr;
}
protected:
LowerLayer* down_;
/// The output of the application. Serialized to the socket.
consumer_type in_;
/// The input to the application. Deserialized from the socket.
producer_type out_;
/// Converts between raw bytes and native C++ objects.
Trait trait_;
/// Our event loop.
async::execution_context_ptr loop_;
/// Initializes the bridge. Disposed (set to null) after initializing.
connector_pointer conn_;
/// Type-erased handle to the @ref socket_manager. This reference is important
/// to keep the bridge alive while the manager is not registered for writing
/// or reading.
disposable self_ref_;
};
} // namespace caf::detail
......@@ -7,6 +7,7 @@
#include "caf/async/consumer_adapter.hpp"
#include "caf/async/producer_adapter.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/detail/flow_bridge_base.hpp"
#include "caf/fwd.hpp"
#include "caf/net/binary/lower_layer.hpp"
#include "caf/net/binary/upper_layer.hpp"
......@@ -17,153 +18,48 @@
namespace caf::net::binary {
/// Convenience alias for referring to the base type of @ref flow_bridge.
template <class Trait>
using flow_bridge_base_t
= detail::flow_bridge_base<upper_layer, lower_layer, Trait>;
/// Translates between a message-oriented transport and data flows.
template <class Trait>
class flow_bridge : public upper_layer {
class flow_bridge : public flow_bridge_base_t<Trait> {
public:
using super = flow_bridge_base_t<Trait>;
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
/// Type for the consumer adapter. We consume the output of the application.
using consumer_type = async::consumer_adapter<output_type>;
/// Type for the producer adapter. We produce the input of the application.
using producer_type = async::producer_adapter<input_type>;
using connector_pointer = flow_connector_ptr<Trait>;
explicit flow_bridge(async::execution_context_ptr loop,
connector_pointer conn)
: loop_(std::move(loop)), conn_(std::move(conn)) {
// nop
}
using super::super;
static std::unique_ptr<flow_bridge> make(async::execution_context_ptr loop,
connector_pointer conn) {
return std::make_unique<flow_bridge>(std::move(loop), std::move(conn));
}
bool write(const output_type& item) {
down_->begin_message();
auto& bytes = down_->message_buffer();
return trait_.convert(item, bytes) && down_->end_message();
}
bool running() const noexcept {
return in_ || out_;
}
void self_ref(disposable ref) {
self_ref_ = std::move(ref);
bool write(const output_type& item) override {
super::down_->begin_message();
auto& bytes = super::down_->message_buffer();
return super::trait_.convert(item, bytes) && super::down_->end_message();
}
// -- implementation of binary::lower_layer ----------------------------------
error start(binary::lower_layer* down, const settings& cfg) override {
CAF_ASSERT(down != nullptr);
down_ = down;
auto [err, pull, push] = conn_->on_request(cfg);
if (!err) {
auto do_wakeup = make_action([this] {
prepare_send();
if (!running())
down_->shutdown();
});
auto do_resume = make_action([this] { down_->request_messages(); });
auto do_cancel = make_action([this] {
if (!running()) {
down_->shutdown();
}
});
in_ = consumer_type::make(pull.try_open(), loop_, std::move(do_wakeup));
out_ = producer_type::make(push.try_open(), loop_, std::move(do_resume),
std::move(do_cancel));
conn_ = nullptr;
if (in_ && out_) {
return none;
} else {
return make_error(sec::runtime_error,
"cannot init flow bridge: no buffers");
}
} else {
conn_ = nullptr;
return err;
}
}
void prepare_send() override {
input_type tmp;
while (down_->can_send_more()) {
switch (in_.pull(async::delay_errors, tmp)) {
case async::read_result::ok:
if (!write(tmp)) {
down_->shutdown(trait_.last_error());
return;
}
break;
case async::read_result::stop:
down_->shutdown();
break;
case async::read_result::abort:
down_->shutdown(in_.abort_reason());
break;
default: // try later
return;
}
}
}
bool done_sending() override {
return !in_.has_consumer_event();
}
void abort(const error& reason) override {
CAF_LOG_TRACE(CAF_ARG(reason));
if (out_) {
if (reason == sec::connection_closed || reason == sec::socket_disconnected
|| reason == sec::disposed)
out_.close();
else
out_.abort(reason);
}
in_.cancel();
self_ref_ = nullptr;
}
ptrdiff_t consume(byte_span buf) override {
if (!out_)
if (!super::out_)
return -1;
input_type val;
if (!trait_.convert(buf, val))
if (!super::trait_.convert(buf, val))
return -1;
if (out_.push(std::move(val)) == 0)
down_->suspend_reading();
if (super::out_.push(std::move(val)) == 0)
super::down_->suspend_reading();
return static_cast<ptrdiff_t>(buf.size());
}
private:
lower_layer* down_;
/// The output of the application. Serialized to the socket.
consumer_type in_;
/// The input to the application. Deserialized from the socket.
producer_type out_;
/// Converts between raw bytes and native C++ objects.
Trait trait_;
/// Our event loop.
async::execution_context_ptr loop_;
/// Initializes the bridge. Disposed (set to null) after initializing.
connector_pointer conn_;
/// Type-erased handle to the @ref socket_manager. This reference is important
/// to keep the bridge alive while the manager is not registered for writing
/// or reading.
disposable self_ref_;
};
} // namespace caf::net::binary
......@@ -56,103 +56,111 @@ public:
// -- high-level factory functions -------------------------------------------
/// Describes the one-time connection event.
using connect_event_t
= cow_tuple<async::consumer_resource<binary::frame>, // Socket to App.
async::producer_resource<binary::frame>>; // App to Socket.
/// Runs length-prefix framing on given connection.
/// @param sys The host system.
/// @param conn A connected stream socket or SSL connection, depending on the
/// `Transport`.
/// @param pull Source for pulling data to send.
/// @param push Source for pushing received data.
template <class Connection>
static disposable run(actor_system& sys, Connection conn,
async::consumer_resource<binary::frame> pull,
async::producer_resource<binary::frame> push) {
using trait_t = binary::default_trait;
using transport_t = typename Connection::transport_type;
auto mpx = sys.network_manager().mpx_ptr();
auto fc = flow_connector<trait_t>::make_trivial(std::move(pull),
/// Binds a trait class to the framing protocol to enable a high-level API for
/// operating on flows.
template <class Trait>
struct bind {
/// Describes the one-time connection event.
using connect_event_t
= cow_tuple<async::consumer_resource<typename Trait::input_type>,
async::producer_resource<typename Trait::output_type>>;
/// Runs length-prefix framing on given connection.
/// @param sys The host system.
/// @param conn A connected stream socket or SSL connection, depending on
/// the
/// `Transport`.
/// @param pull Source for pulling data to send.
/// @param push Source for pushing received data.
template <class Connection>
static disposable
run(actor_system& sys, Connection conn,
async::consumer_resource<typename Trait::output_type> pull,
async::producer_resource<typename Trait::input_type> push) {
using transport_t = typename Connection::transport_type;
auto mpx = sys.network_manager().mpx_ptr();
auto fc = flow_connector<Trait>::make_trivial(std::move(pull),
std::move(push));
auto bridge = binary::flow_bridge<trait_t>::make(mpx, std::move(fc));
auto bridge_ptr = bridge.get();
auto impl = length_prefix_framing::make(std::move(bridge));
auto transport = transport_t::make(std::move(conn), std::move(impl));
auto ptr = socket_manager::make(mpx, std::move(transport));
bridge_ptr->self_ref(ptr->as_disposable());
mpx->start(ptr);
return disposable{std::move(ptr)};
}
/// Runs length-prefix framing on given connection.
/// @param sys The host system.
/// @param conn A connected stream socket or SSL connection, depending on the
/// `Transport`.
/// @param init Function object for setting up the created flows.
template <class Connection, class Init>
static disposable run(actor_system& sys, Connection conn, Init init) {
static_assert(std::is_invocable_v<Init, connect_event_t&&>,
"invalid signature found for init");
using frame_t = binary::frame;
auto [fd_pull, app_push] = async::make_spsc_buffer_resource<frame_t>();
auto [app_pull, fd_push] = async::make_spsc_buffer_resource<frame_t>();
auto result = run(sys, std::move(conn), std::move(fd_pull),
std::move(fd_push));
init(connect_event_t{std::move(app_pull), std::move(app_push)});
return result;
}
/// The default number of concurrently open connections when using `accept`.
static constexpr size_t default_max_connections = 128;
/// A producer resource for the acceptor. Any accepted connection is
/// represented by two buffers: one for input and one for output.
using acceptor_resource_t = async::producer_resource<connect_event_t>;
/// Describes the per-connection event.
using accept_event_t
= cow_tuple<async::consumer_resource<binary::frame>, // Socket to App.
async::producer_resource<binary::frame>>; // App to Socket.
/// Convenience function for creating an event listener resource and an
/// @ref acceptor_resource_t via @ref async::make_spsc_buffer_resource.
template <class... Ts>
static auto make_accept_event_resources() {
return async::make_spsc_buffer_resource<accept_event_t>();
}
/// Listens for incoming connection on @p fd.
/// @param sys The host system.
/// @param acc A connection acceptor such as @ref tcp_accept_socket or
/// @ref ssl::acceptor.
/// @param cfg Configures the acceptor. Currently, the only supported
/// configuration parameter is `max-connections`.
template <class Acceptor>
static disposable accept(actor_system& sys, Acceptor acc,
acceptor_resource_t out, const settings& cfg = {}) {
using transport_t = typename Acceptor::transport_type;
using trait_t = binary::default_trait;
using factory_t = cf_impl<transport_t>;
using conn_t = typename transport_t::connection_handle;
using impl_t = detail::accept_handler<Acceptor, conn_t>;
auto max_connections = get_or(cfg, defaults::net::max_connections);
if (auto buf = out.try_open()) {
auto& mpx = sys.network_manager().mpx();
auto conn = flow_connector<trait_t>::make_basic_server(std::move(buf));
auto factory = std::make_unique<factory_t>(std::move(conn));
auto impl = impl_t::make(std::move(acc), std::move(factory),
max_connections);
auto impl_ptr = impl.get();
auto ptr = net::socket_manager::make(&mpx, std::move(impl));
impl_ptr->self_ref(ptr->as_disposable());
mpx.start(ptr);
auto bridge = binary::flow_bridge<Trait>::make(mpx, std::move(fc));
auto bridge_ptr = bridge.get();
auto impl = length_prefix_framing::make(std::move(bridge));
auto transport = transport_t::make(std::move(conn), std::move(impl));
auto ptr = socket_manager::make(mpx, std::move(transport));
bridge_ptr->self_ref(ptr->as_disposable());
mpx->start(ptr);
return disposable{std::move(ptr)};
} else {
return {};
}
}
/// Runs length-prefix framing on given connection.
/// @param sys The host system.
/// @param conn A connected stream socket or SSL connection, depending on
/// the
/// `Transport`.
/// @param init Function object for setting up the created flows.
template <class Connection, class Init>
static disposable run(actor_system& sys, Connection conn, Init init) {
using app_in_t = typename Trait::input_type;
using app_out_t = typename Trait::output_type;
static_assert(std::is_invocable_v<Init, connect_event_t&&>,
"invalid signature found for init");
auto [app_pull, fd_push] = async::make_spsc_buffer_resource<app_in_t>();
auto [fd_pull, app_push] = async::make_spsc_buffer_resource<app_out_t>();
auto result = run(sys, std::move(conn), std::move(fd_pull),
std::move(fd_push));
init(connect_event_t{std::move(app_pull), std::move(app_push)});
return result;
}
/// The default number of concurrently open connections when using `accept`.
static constexpr size_t default_max_connections = 128;
/// A producer resource for the acceptor. Any accepted connection is
/// represented by two buffers: one for input and one for output.
using acceptor_resource_t = async::producer_resource<connect_event_t>;
/// Describes the per-connection event.
using accept_event_t
= cow_tuple<async::consumer_resource<typename Trait::input_type>,
async::producer_resource<typename Trait::output_type>>;
/// Convenience function for creating an event listener resource and an
/// @ref acceptor_resource_t via @ref async::make_spsc_buffer_resource.
static auto make_accept_event_resources() {
return async::make_spsc_buffer_resource<accept_event_t>();
}
/// Listens for incoming connection on @p fd.
/// @param sys The host system.
/// @param acc A connection acceptor such as @ref tcp_accept_socket or
/// @ref ssl::acceptor.
/// @param cfg Configures the acceptor. Currently, the only supported
/// configuration parameter is `max-connections`.
template <class Acceptor>
static disposable accept(actor_system& sys, Acceptor acc,
acceptor_resource_t out,
const settings& cfg = {}) {
using transport_t = typename Acceptor::transport_type;
using trait_t = binary::default_trait;
using factory_t = cf_impl<transport_t>;
using conn_t = typename transport_t::connection_handle;
using impl_t = detail::accept_handler<Acceptor, conn_t>;
auto max_connections = get_or(cfg, defaults::net::max_connections);
if (auto buf = out.try_open()) {
auto& mpx = sys.network_manager().mpx();
auto conn = flow_connector<trait_t>::make_basic_server(std::move(buf));
auto factory = std::make_unique<factory_t>(std::move(conn));
auto impl = impl_t::make(std::move(acc), std::move(factory),
max_connections);
auto impl_ptr = impl.get();
auto ptr = net::socket_manager::make(&mpx, std::move(impl));
impl_ptr->self_ref(ptr->as_disposable());
mpx.start(ptr);
return disposable{std::move(ptr)};
} else {
return {};
}
}
};
// -- implementation of stream_oriented::upper_layer -------------------------
......
......@@ -4,10 +4,6 @@
#pragma once
#include <memory>
#include <mutex>
#include <thread>
#include "caf/action.hpp"
#include "caf/async/execution_context.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
......@@ -19,6 +15,11 @@
#include "caf/ref_counted.hpp"
#include "caf/unordered_flat_map.hpp"
#include <deque>
#include <memory>
#include <mutex>
#include <thread>
extern "C" {
struct pollfd;
......@@ -150,6 +151,9 @@ public:
/// Applies all pending updates.
void apply_updates();
/// Runs all pending actions.
void run_actions();
/// Sets the thread ID to `std::this_thread::id()`.
void set_thread_id();
......@@ -213,6 +217,9 @@ protected:
/// Signals whether shutdown has been requested.
bool shutting_down_ = false;
/// Pending actions via `schedule`.
std::deque<action> pending_actions_;
/// Keeps track of watched disposables.
std::vector<disposable> watched_;
......
......@@ -69,7 +69,7 @@ public:
/// from a client.
[[nodiscard]] ptrdiff_t accept();
/// Gracefully closes the SSL connection.
/// Gracefully closes the SSL connection without closing the socket.
ptrdiff_t close();
// -- reading and writing ----------------------------------------------------
......
......@@ -8,9 +8,12 @@
#include "caf/net/ssl/dtls.hpp"
#include "caf/net/ssl/format.hpp"
#include "caf/net/ssl/fwd.hpp"
#include "caf/net/ssl/password.hpp"
#include "caf/net/ssl/tls.hpp"
#include "caf/net/ssl/verify.hpp"
#include "caf/net/stream_socket.hpp"
#include <cstring>
#include <string>
namespace caf::net::ssl {
......@@ -27,6 +30,9 @@ public:
/// The opaque implementation type.
struct impl;
/// Stores additional data provided by the user.
struct user_data;
// -- constructors, destructors, and assignment operators --------------------
context() = delete;
......@@ -65,6 +71,30 @@ public:
return pimpl_ == nullptr;
}
/// Overrides the verification mode for this context.
/// @note calls @c SSL_CTX_set_verify
void set_verify_mode(verify_t flags);
/// Overrides the callback to obtain the password for encrypted PEM files.
/// @note calls @c SSL_CTX_set_default_passwd_cb
template <typename PasswordCallback>
void set_password_callback(PasswordCallback callback) {
set_password_callback_impl(password::make_callback(std::move(callback)));
}
/// Overrides the callback to obtain the password for encrypted PEM files with
/// a function that always returns @p password.
/// @note calls @c SSL_CTX_set_default_passwd_cb
void set_password(std::string password) {
auto cb = [pw = std::move(password)](char* buf, int len,
password::purpose) {
strncpy(buf, pw.c_str(), static_cast<size_t>(len));
buf[len - 1] = '\0';
return static_cast<int>(pw.size());
};
set_password_callback(std::move(cb));
}
// -- native handles ---------------------------------------------------------
/// Reinterprets `native_handle` as the native implementation type and takes
......@@ -111,29 +141,57 @@ public:
/// e.g., `9d66eef0.0` and `9d66eef0.1`.
/// @returns `true` on success, `false` otherwise and `last_error` can be used
/// to retrieve a human-readable error representation.
[[nodiscard]] bool load_verify_dir(const char* path);
/// @note Calls @c SSL_CTX_load_verify_locations
[[nodiscard]] bool add_verify_path(const char* path);
/// @copydoc add_verify_path
[[nodiscard]] bool add_verify_path(const std::string& path) {
return add_verify_path(path.c_str());
}
/// Loads a CA certificate file.
/// @param path Null-terminated string with a path to a single PEM file.
/// @returns `true` on success, `false` otherwise and `last_error` can be used
/// to retrieve a human-readable error representation.
/// @note Calls @c SSL_CTX_load_verify_locations
[[nodiscard]] bool load_verify_file(const char* path);
/// @copydoc load_verify_file
[[nodiscard]] bool load_verify_file(const std::string& path) {
return load_verify_file(path.c_str());
}
/// Loads the first certificate found in given file.
/// @param path Null-terminated string with a path to a single PEM file.
[[nodiscard]] bool use_certificate_from_file(const char* path,
format file_format);
[[nodiscard]] bool use_certificate_file(const char* path, format file_format);
/// Loads a certificate chain from a PEM-formatted file.
/// @note calls @c SSL_CTX_use_certificate_chain_file
[[nodiscard]] bool use_certificate_chain_file(const char* path);
/// @copydoc use_certificate_chain_file
[[nodiscard]] bool use_certificate_chain_file(const std::string& path) {
return use_certificate_chain_file(path.c_str());
}
/// Loads the first private key found in given file.
[[nodiscard]] bool use_private_key_from_file(const char* path,
format file_format);
[[nodiscard]] bool use_private_key_file(const char* path, format file_format);
/// @copydoc use_private_key_file
[[nodiscard]] bool use_private_key_file(const std::string& path,
format file_format) {
return use_private_key_file(path.c_str(), file_format);
}
private:
constexpr explicit context(impl* ptr) : pimpl_(ptr) {
// nop
}
void set_password_callback_impl(password::callback_ptr callback);
impl* pimpl_;
user_data* data_ = nullptr;
};
} // namespace caf::net::ssl
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include <memory>
namespace caf::net::ssl::password {
/// Purpose of PEM password.
enum class purpose {
/// SSL asks the password is reading/decryption.
reading,
/// SSL asks the password is for writing/encryption.
writing
};
class CAF_NET_EXPORT callback {
public:
virtual ~callback();
/// Reads the password into the buffer @p buf of size @p len.
/// @returns Written characters on success, -1 otherwise.
virtual int operator()(char* buf, int len, purpose flag) = 0;
};
using callback_ptr = std::unique_ptr<callback>;
template <class PasswordCallback>
callback_ptr make_callback(PasswordCallback callback) {
struct impl : callback {
explicit impl(PasswordCallback&& cb) : cb_(std::move(cb)) {
// nop
}
int operator()(char* buf, int len, purpose flag) override {
return cb_(buf, len, flag);
}
PasswordCallback cb_;
};
return std::make_unique<impl>(std::move(callback));
}
} // namespace caf::net::ssl::password
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/default_enum_inspect.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include <string>
namespace caf::net::ssl {
/// Bitmask type for the SSL verification mode.
enum class verify_t {};
namespace verify {
CAF_NET_EXPORT extern const verify_t none;
CAF_NET_EXPORT extern const verify_t peer;
CAF_NET_EXPORT extern const verify_t fail_if_no_peer_cert;
CAF_NET_EXPORT extern const verify_t client_once;
} // namespace verify
constexpr int to_integer(verify_t x) {
return static_cast<int>(x);
}
inline verify_t& operator|=(verify_t& x, verify_t y) noexcept {
return x = static_cast<verify_t>(to_integer(x) | to_integer(y));
}
constexpr verify_t operator|(verify_t x, verify_t y) noexcept {
return static_cast<verify_t>(to_integer(x) | to_integer(y));
}
} // namespace caf::net::ssl
......@@ -9,6 +9,7 @@
#include "caf/detail/accept_handler.hpp"
#include "caf/detail/connection_factory.hpp"
#include "caf/net/flow_connector.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/ssl/acceptor.hpp"
#include "caf/net/ssl/transport.hpp"
#include "caf/net/web_socket/default_trait.hpp"
......@@ -36,6 +37,10 @@ using accept_event_t
template <class... Ts>
using acceptor_resource_t = async::producer_resource<accept_event_t<Ts...>>;
/// A consumer resource for processing accepted connections.
template <class... Ts>
using listener_resource_t = async::consumer_resource<accept_event_t<Ts...>>;
/// Convenience function for creating an event listener resource and an
/// @ref acceptor_resource_t via @ref async::make_spsc_buffer_resource.
template <class... Ts>
......
......@@ -7,6 +7,7 @@
#include "caf/async/consumer_adapter.hpp"
#include "caf/async/producer_adapter.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/detail/flow_bridge_base.hpp"
#include "caf/fwd.hpp"
#include "caf/net/flow_connector.hpp"
#include "caf/net/web_socket/lower_layer.hpp"
......@@ -18,170 +19,67 @@
namespace caf::net::web_socket {
/// Convenience alias for referring to the base type of @ref flow_bridge.
template <class Trait>
using flow_bridge_base_t
= detail::flow_bridge_base<upper_layer, lower_layer, Trait>;
/// Translates between a message-oriented transport and data flows.
template <class Trait>
class flow_bridge : public web_socket::upper_layer {
class flow_bridge : public flow_bridge_base_t<Trait> {
public:
using super = flow_bridge_base_t<Trait>;
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
/// Type for the consumer adapter. We consume the output of the application.
using consumer_type = async::consumer_adapter<output_type>;
/// Type for the producer adapter. We produce the input of the application.
using producer_type = async::producer_adapter<input_type>;
using request_type = request<Trait>;
using connector_pointer = flow_connector_ptr<Trait>;
explicit flow_bridge(async::execution_context_ptr loop,
connector_pointer conn)
: loop_(std::move(loop)), conn_(std::move(conn)) {
// nop
}
using super::super;
static std::unique_ptr<flow_bridge> make(async::execution_context_ptr loop,
connector_pointer conn) {
return std::make_unique<flow_bridge>(std::move(loop), std::move(conn));
}
bool write(const output_type& item) {
if (trait_.converts_to_binary(item)) {
down_->begin_binary_message();
auto& bytes = down_->binary_message_buffer();
return trait_.convert(item, bytes) && down_->end_binary_message();
bool write(const output_type& item) override {
if (super::trait_.converts_to_binary(item)) {
super::down_->begin_binary_message();
auto& bytes = super::down_->binary_message_buffer();
return super::trait_.convert(item, bytes)
&& super::down_->end_binary_message();
} else {
down_->begin_text_message();
auto& text = down_->text_message_buffer();
return trait_.convert(item, text) && down_->end_text_message();
super::down_->begin_text_message();
auto& text = super::down_->text_message_buffer();
return super::trait_.convert(item, text)
&& super::down_->end_text_message();
}
}
bool running() const noexcept {
return in_ || out_;
}
void self_ref(disposable ref) {
self_ref_ = std::move(ref);
}
// -- implementation of web_socket::lower_layer ------------------------------
error start(web_socket::lower_layer* down, const settings& cfg) override {
down_ = down;
auto [err, pull, push] = conn_->on_request(cfg);
if (!err) {
auto do_wakeup = make_action([this] {
prepare_send();
if (!running())
down_->shutdown();
});
auto do_resume = make_action([this] { down_->request_messages(); });
auto do_cancel = make_action([this] {
if (!running())
down_->shutdown();
});
in_ = consumer_type::make(pull.try_open(), loop_, std::move(do_wakeup));
out_ = producer_type::make(push.try_open(), loop_, std::move(do_resume),
std::move(do_cancel));
conn_ = nullptr;
if (running())
return none;
else
return make_error(sec::runtime_error,
"cannot init flow bridge: no buffers");
} else {
conn_ = nullptr;
return err;
}
}
void prepare_send() override {
input_type tmp;
while (down_->can_send_more()) {
switch (in_.pull(async::delay_errors, tmp)) {
case async::read_result::ok:
if (!write(tmp)) {
down_->shutdown(trait_.last_error());
return;
}
break;
case async::read_result::stop:
down_->shutdown();
break;
case async::read_result::abort:
down_->shutdown(in_.abort_reason());
break;
default: // try later
return;
}
}
}
bool done_sending() override {
return !in_.has_consumer_event();
}
void abort(const error& reason) override {
CAF_LOG_TRACE(CAF_ARG(reason));
if (out_) {
if (reason == sec::connection_closed || reason == sec::socket_disconnected
|| reason == sec::disposed) {
out_.close();
} else {
out_.abort(reason);
}
}
in_.cancel();
self_ref_ = nullptr;
}
ptrdiff_t consume_binary(byte_span buf) override {
if (!out_)
if (!super::out_)
return -1;
input_type val;
if (!trait_.convert(buf, val))
if (!super::trait_.convert(buf, val))
return -1;
if (out_.push(std::move(val)) == 0)
down_->suspend_reading();
if (super::out_.push(std::move(val)) == 0)
super::down_->suspend_reading();
return static_cast<ptrdiff_t>(buf.size());
}
ptrdiff_t consume_text(std::string_view buf) override {
if (!out_)
if (!super::out_)
return -1;
input_type val;
if (!trait_.convert(buf, val))
if (!super::trait_.convert(buf, val))
return -1;
if (out_.push(std::move(val)) == 0)
down_->suspend_reading();
if (super::out_.push(std::move(val)) == 0)
super::down_->suspend_reading();
return static_cast<ptrdiff_t>(buf.size());
}
private:
web_socket::lower_layer* down_;
/// The output of the application. Serialized to the socket.
consumer_type in_;
/// The input to the application. Deserialized from the socket.
producer_type out_;
/// Converts between raw bytes and native C++ objects.
Trait trait_;
/// Runs callbacks in the I/O event loop.
async::execution_context_ptr loop_;
/// Initializes the bridge. Disposed (set to null) after initializing.
connector_pointer conn_;
/// Type-erased handle to the @ref socket_manager. This reference is important
/// to keep the bridge alive while the manager is not registered for writing
/// or reading.
disposable self_ref_;
};
} // namespace caf::net::web_socket
......@@ -171,8 +171,12 @@ void multiplexer::deref_execution_context() const noexcept {
void multiplexer::schedule(action what) {
CAF_LOG_TRACE("");
auto ptr = std::move(what).as_intrusive_ptr().release();
write_to_pipe(pollset_updater::code::run_action, ptr);
if (std::this_thread::get_id() == tid_) {
pending_actions_.push_back(what);
} else {
auto ptr = std::move(what).as_intrusive_ptr().release();
write_to_pipe(pollset_updater::code::run_action, ptr);
}
}
void multiplexer::watch(disposable what) {
......@@ -309,23 +313,32 @@ void multiplexer::poll() {
void multiplexer::apply_updates() {
CAF_LOG_DEBUG("apply" << updates_.size() << "updates");
if (!updates_.empty()) {
for (auto& [fd, update] : updates_) {
if (auto index = index_of(fd); index == -1) {
if (update.events != 0) {
pollfd new_entry{socket_cast<socket_id>(fd), update.events, 0};
pollset_.emplace_back(new_entry);
managers_.emplace_back(std::move(update.mgr));
for (;;) {
if (!updates_.empty()) {
for (auto& [fd, update] : updates_) {
if (auto index = index_of(fd); index == -1) {
if (update.events != 0) {
pollfd new_entry{socket_cast<socket_id>(fd), update.events, 0};
pollset_.emplace_back(new_entry);
managers_.emplace_back(std::move(update.mgr));
}
} else if (update.events != 0) {
pollset_[index].events = update.events;
managers_[index].swap(update.mgr);
} else {
pollset_.erase(pollset_.begin() + index);
managers_.erase(managers_.begin() + index);
}
} else if (update.events != 0) {
pollset_[index].events = update.events;
managers_[index].swap(update.mgr);
} else {
pollset_.erase(pollset_.begin() + index);
managers_.erase(managers_.begin() + index);
}
updates_.clear();
}
while (!pending_actions_.empty()) {
auto next = std::move(pending_actions_.front());
pending_actions_.pop_front();
next.run();
}
updates_.clear();
if (updates_.empty())
return;
}
}
......
......@@ -45,9 +45,9 @@ void pollset_updater::handle_read_event() {
auto as_mgr = [](intptr_t ptr) {
return intrusive_ptr{reinterpret_cast<socket_manager*>(ptr), false};
};
auto run_action = [](intptr_t ptr) {
auto add_action = [this](intptr_t ptr) {
auto f = action{intrusive_ptr{reinterpret_cast<action::impl*>(ptr), false}};
f.run();
mpx_->pending_actions_.push_back(std::move(f));
};
for (;;) {
CAF_ASSERT((buf_.size() - buf_size_) > 0);
......@@ -65,7 +65,7 @@ void pollset_updater::handle_read_event() {
mpx_->do_start(as_mgr(ptr));
break;
case code::run_action:
run_action(ptr);
add_action(ptr);
break;
case code::shutdown:
CAF_ASSERT(ptr == 0);
......
......@@ -33,11 +33,11 @@ acceptor::make_with_cert_file(tcp_accept_socket fd, const char* cert_file_path,
if (!ctx) {
return {make_error(sec::runtime_error, "unable to create SSL context")};
}
if (!ctx->use_certificate_from_file(cert_file_path, file_format)) {
if (!ctx->use_certificate_file(cert_file_path, file_format)) {
return {make_error(sec::runtime_error, "unable to load certificate file",
ctx->last_error_string())};
}
if (!ctx->use_private_key_from_file(key_file_path, file_format)) {
if (!ctx->use_private_key_file(key_file_path, file_format)) {
return {make_error(sec::runtime_error, "unable to load private key file",
ctx->last_error_string())};
}
......
......@@ -23,22 +23,34 @@ auto native(context::impl* ptr) {
} // namespace
// -- member types -------------------------------------------------------------
struct context::user_data {
password::callback_ptr pw_callback;
};
// -- constructors, destructors, and assignment operators ----------------------
context::context(context&& other) {
pimpl_ = other.pimpl_;
other.pimpl_ = nullptr;
data_ = other.data_;
other.data_ = nullptr;
}
context& context::operator=(context&& other) {
SSL_CTX_free(native(pimpl_));
pimpl_ = other.pimpl_;
other.pimpl_ = nullptr;
delete data_;
data_ = other.data_;
other.data_ = nullptr;
return *this;
}
context::~context() {
SSL_CTX_free(native(pimpl_)); // Already checks for NULL.
delete data_;
}
// -- factories ----------------------------------------------------------------
......@@ -147,6 +159,34 @@ expected<context> context::make_client(dtls vmin, dtls vmax) {
return {make_error(sec::logic_error, errstr)};
}
// -- properties ---------------------------------------------------------------
void context::set_verify_mode(verify_t flags) {
auto ptr = native(pimpl_);
SSL_CTX_set_verify(ptr, to_integer(flags), SSL_CTX_get_verify_callback(ptr));
}
namespace {
int c_password_callback(char* buf, int size, int rwflag, void* ptr) {
if (ptr == nullptr)
return -1;
auto flag = static_cast<password::purpose>(rwflag);
auto fn = static_cast<password::callback*>(ptr);
return (*fn)(buf, size, flag);
}
} // namespace
void context::set_password_callback_impl(password::callback_ptr callback) {
if (data_ == nullptr)
data_ = new user_data;
auto ptr = native(pimpl_);
data_->pw_callback = std::move(callback);
SSL_CTX_set_default_passwd_cb(ptr, c_password_callback);
SSL_CTX_set_default_passwd_cb_userdata(ptr, data_->pw_callback.get());
}
// -- native handles -----------------------------------------------------------
context context::from_native(void* native_handle) {
......@@ -217,7 +257,7 @@ bool context::set_default_verify_paths() {
return SSL_CTX_set_default_verify_paths(native(pimpl_)) == 1;
}
bool context::load_verify_dir(const char* path) {
bool context::add_verify_path(const char* path) {
ERR_clear_error();
return SSL_CTX_load_verify_locations(native(pimpl_), nullptr, path) == 1;
}
......@@ -227,16 +267,21 @@ bool context::load_verify_file(const char* path) {
return SSL_CTX_load_verify_locations(native(pimpl_), path, nullptr) == 1;
}
bool context::use_certificate_from_file(const char* path, format file_format) {
bool context::use_certificate_file(const char* path, format file_format) {
ERR_clear_error();
auto nff = native(file_format);
return SSL_CTX_use_certificate_file(native(pimpl_), path, nff) == 1;
}
bool context::use_certificate_chain_file(const char* path) {
ERR_clear_error();
return SSL_CTX_use_certificate_file(native(pimpl_), path, native(file_format))
== 1;
return SSL_CTX_use_certificate_chain_file(native(pimpl_), path) == 1;
}
bool context::use_private_key_from_file(const char* path, format file_format) {
bool context::use_private_key_file(const char* path, format file_format) {
ERR_clear_error();
return SSL_CTX_use_PrivateKey_file(native(pimpl_), path, native(file_format))
== 1;
auto nff = native(file_format);
return SSL_CTX_use_PrivateKey_file(native(pimpl_), path, nff) == 1;
}
} // namespace caf::net::ssl
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/net/ssl/password.hpp"
namespace caf::net::ssl::password {
callback::~callback() {
// nop
}
} // namespace caf::net::ssl::password
......@@ -62,6 +62,8 @@ void startup() {
CRYPTO_set_dynlock_create_callback(dynlock_create);
CRYPTO_set_dynlock_lock_callback(dynlock_lock);
CRYPTO_set_dynlock_destroy_callback(dynlock_destroy);
#else
OPENSSL_init_ssl(0, nullptr);
#endif
}
......@@ -72,6 +74,10 @@ void cleanup() {
CRYPTO_set_dynlock_lock_callback(nullptr);
CRYPTO_set_dynlock_destroy_callback(nullptr);
mutexes.clear();
#else
ERR_free_strings();
EVP_cleanup();
CRYPTO_cleanup_all_ex_data();
#endif
}
......
#include "caf/net/ssl/verify.hpp"
#include <openssl/ssl.h>
namespace caf::net::ssl::verify {
namespace {
constexpr verify_t i2v(int x) {
return static_cast<verify_t>(x);
}
} // namespace
const verify_t none = i2v(SSL_VERIFY_NONE);
const verify_t peer = i2v(SSL_VERIFY_PEER);
const verify_t fail_if_no_peer_cert = i2v(SSL_VERIFY_FAIL_IF_NO_PEER_CERT);
const verify_t client_once = i2v(SSL_VERIFY_CLIENT_ONCE);
} // namespace caf::net::ssl::verify
......@@ -215,7 +215,6 @@ SCENARIO("calling suspend_reading temporarily halts receiving of messages") {
app_ptr->continue_reading();
mpx->apply_updates();
mpx->poll_once(true);
CHECK_EQ(mpx->mask_of(mgr), net::operation::read);
while (mpx->num_socket_managers() > 1u)
mpx->poll_once(true);
if (CHECK_EQ(buf->size(), 5u)) {
......@@ -248,7 +247,9 @@ SCENARIO("length_prefix_framing::run translates between flows and socket I/O") {
caf::actor_system sys{cfg};
auto buf = std::make_shared<std::vector<std::string>>();
caf::actor hdl;
net::length_prefix_framing::run(sys, fd2, [&](auto event) {
using trait = net::binary::default_trait;
using lpf = net::length_prefix_framing::bind<trait>;
lpf::run(sys, fd2, [&](auto event) {
hdl = sys.spawn([event, buf](event_based_actor* self) {
auto [pull, push] = event.data();
pull.observe_on(self)
......
......@@ -112,12 +112,12 @@ void dummy_tls_server(stream_socket fd, const char* cert_file,
auto guard = detail::make_scope_guard([fd] { close(fd); });
// Get and configure our SSL context.
auto ctx = unbox(ssl::context::make_server(ssl::tls::any));
if (!ctx.use_certificate_from_file(cert_file, ssl::format::pem)) {
if (!ctx.use_certificate_file(cert_file, ssl::format::pem)) {
std::cerr << "*** failed to load certificate_file: "
<< ctx.last_error_string() << '\n';
return;
}
if (!ctx.use_private_key_from_file(key_file, ssl::format::pem)) {
if (!ctx.use_private_key_file(key_file, ssl::format::pem)) {
std::cerr << "*** failed to load private key file: "
<< ctx.last_error_string() << '\n';
return;
......@@ -231,10 +231,10 @@ SCENARIO("ssl::transport::make_server performs the server handshake") {
mpx->set_thread_id();
std::ignore = mpx->init();
auto ctx = unbox(ssl::context::make_server(ssl::tls::any));
REQUIRE(ctx.use_certificate_from_file(cert_1_pem_path, //
ssl::format::pem));
REQUIRE(ctx.use_private_key_from_file(key_1_pem_path, //
ssl::format::pem));
REQUIRE(ctx.use_certificate_file(cert_1_pem_path, //
ssl::format::pem));
REQUIRE(ctx.use_private_key_file(key_1_pem_path, //
ssl::format::pem));
auto conn = unbox(ctx.new_connection(server_fd));
auto done = std::make_shared<bool>(false);
auto buf = std::make_shared<byte_buffer>();
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment