Commit 195eba3a authored by Dominik Charousset's avatar Dominik Charousset

Fix SSL session management

parent 0d655dd9
...@@ -20,29 +20,40 @@ ...@@ -20,29 +20,40 @@
#ifndef CAF_OPENSSL_SESSION_HPP #ifndef CAF_OPENSSL_SESSION_HPP
#define CAF_OPENSSL_SESSION_HPP #define CAF_OPENSSL_SESSION_HPP
#include "caf/actor_system.hpp" #include "caf/config.hpp"
#include "caf/io/network/native_socket.hpp"
CAF_PUSH_WARNINGS
#include <openssl/ssl.h> #include <openssl/ssl.h>
CAF_POP_WARNINGS
#include "caf/actor_system.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/io/network/default_multiplexer.hpp"
namespace caf { namespace caf {
namespace openssl { namespace openssl {
using native_socket = io::network::native_socket; using native_socket = io::network::native_socket;
using rw_state = io::network::rw_state;
class session : public ref_counted { class session : public ref_counted {
public: public:
session(actor_system& sys); session(actor_system& sys);
~session(); ~session();
void init(); bool init();
bool read_some(size_t& result, native_socket fd, void* buf, size_t len); rw_state read_some(size_t& result, native_socket fd, void* buf, size_t len);
bool write_some(size_t& result, native_socket fd, const void* buf, rw_state write_some(size_t& result, native_socket fd, const void* buf,
size_t len); size_t len);
bool connect(native_socket fd); bool try_connect(native_socket fd);
bool try_accept(native_socket fd); bool try_accept(native_socket fd);
const char* openssl_passphrase(); const char* openssl_passphrase();
private: private:
rw_state do_some(int (*f)(SSL*, void*, int), size_t& result, void* buf,
size_t len, const char* debug_name);
SSL_CTX* create_ssl_context(); SSL_CTX* create_ssl_context();
std::string get_ssl_error(); std::string get_ssl_error();
void raise_ssl_error(std::string msg); void raise_ssl_error(std::string msg);
...@@ -52,8 +63,17 @@ private: ...@@ -52,8 +63,17 @@ private:
SSL_CTX* ctx_; SSL_CTX* ctx_;
SSL* ssl_; SSL* ssl_;
std::string openssl_passphrase_; std::string openssl_passphrase_;
bool connecting_;
bool accepting_;
}; };
/// @relates session
using session_ptr = intrusive_ptr<session>;
/// @relates session
session_ptr make_session(actor_system& sys, native_socket fd,
bool from_accepted_socket);
} // namespace openssl } // namespace openssl
} // namespace caf } // namespace caf
......
...@@ -19,8 +19,10 @@ ...@@ -19,8 +19,10 @@
#include "caf/openssl/manager.hpp" #include "caf/openssl/manager.hpp"
CAF_PUSH_WARNINGS
#include <openssl/err.h> #include <openssl/err.h>
#include <openssl/ssl.h> #include <openssl/ssl.h>
CAF_POP_WARNINGS
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
......
...@@ -51,15 +51,15 @@ using default_mpx = io::network::default_multiplexer; ...@@ -51,15 +51,15 @@ using default_mpx = io::network::default_multiplexer;
struct ssl_policy { struct ssl_policy {
ssl_policy(intrusive_ptr<session> session) : session_(std::move(session)) { ssl_policy(intrusive_ptr<session> session) : session_(std::move(session)) {
// nop
} }
bool read_some(size_t& result, native_socket fd, void* buf, rw_state read_some(size_t& result, native_socket fd, void* buf, size_t len) {
size_t len) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len)); CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len));
return session_->read_some(result, fd, buf, len); return session_->read_some(result, fd, buf, len);
} }
bool write_some(size_t& result, native_socket fd, const void* buf, rw_state write_some(size_t& result, native_socket fd, const void* buf,
size_t len) { size_t len) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len)); CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len));
return session_->write_some(result, fd, buf, len); return session_->write_some(result, fd, buf, len);
...@@ -87,15 +87,19 @@ private: ...@@ -87,15 +87,19 @@ private:
class scribe_impl : public io::scribe { class scribe_impl : public io::scribe {
public: public:
scribe_impl(default_mpx& mpx, native_socket sockfd, scribe_impl(default_mpx& mpx, native_socket sockfd,
intrusive_ptr<session> session) session_ptr sptr)
: scribe(io::network::conn_hdl_from_socket(sockfd)), : scribe(io::network::conn_hdl_from_socket(sockfd)),
launched_(false), launched_(false),
stream_(mpx, sockfd, session) { stream_(mpx, sockfd, std::move(sptr)) {
// nop // nop
} }
void configure_read(io::receive_policy::config config) override { ~scribe_impl() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
}
void configure_read(io::receive_policy::config config) override {
CAF_LOG_TRACE(CAF_ARG(config));
stream_.configure_read(config); stream_.configure_read(config);
if (!launched_) if (!launched_)
launch(); launch();
...@@ -144,13 +148,19 @@ class scribe_impl : public io::scribe { ...@@ -144,13 +148,19 @@ class scribe_impl : public io::scribe {
CAF_ASSERT(!launched_); CAF_ASSERT(!launched_);
launched_ = true; launched_ = true;
stream_.start(this); stream_.start(this);
// This schedules the scribe in case SSL still needs to call SSL_connect
// or SSL_accept. Otherwise, the backend simply removes the socket for
// write operations after the first "nop write".
stream_.force_empty_write(this);
} }
void add_to_loop() override { void add_to_loop() override {
CAF_LOG_TRACE("");
stream_.activate(this); stream_.activate(this);
} }
void remove_from_loop() override { void remove_from_loop() override {
CAF_LOG_TRACE("");
stream_.passivate(); stream_.passivate();
} }
...@@ -159,13 +169,10 @@ class scribe_impl : public io::scribe { ...@@ -159,13 +169,10 @@ class scribe_impl : public io::scribe {
io::network::stream_impl<ssl_policy> stream_; io::network::stream_impl<ssl_policy> stream_;
}; };
class doorman_impl : public io::doorman { class doorman_impl : public io::network::doorman_impl {
public: public:
doorman_impl(default_mpx& mx, native_socket sockfd, doorman_impl(default_mpx& mx, native_socket sockfd)
intrusive_ptr<session> session) : io::network::doorman_impl(mx, sockfd) {
: doorman(io::network::accept_hdl_from_socket(sockfd)),
acceptor_(mx, sockfd, session),
session_(std::move(session)) {
// nop // nop
} }
...@@ -178,49 +185,17 @@ public: ...@@ -178,49 +185,17 @@ public:
// further activities for the broker // further activities for the broker
return false; return false;
auto& dm = acceptor_.backend(); auto& dm = acceptor_.backend();
auto sptr = auto fd = acceptor_.accepted_socket();
make_counted<scribe_impl>(dm, acceptor_.accepted_socket(), session_); auto sssn = make_session(parent()->system(), fd, true);
auto hdl = sptr->hdl(); if (sssn == nullptr) {
parent()->add_scribe(std::move(sptr)); CAF_LOG_ERROR("Unable to create SSL session for accepted socket");
return doorman::new_connection(&dm, hdl); return false;
}
void stop_reading() override {
CAF_LOG_TRACE("");
acceptor_.stop_reading();
detach(&acceptor_.backend(), false);
}
void launch() override {
CAF_LOG_TRACE("");
acceptor_.start(this);
}
std::string addr() const override {
auto x = io::network::local_addr_of_fd(acceptor_.fd());
if (!x)
return "";
return std::move(*x);
}
uint16_t port() const override {
auto x = io::network::local_port_of_fd(acceptor_.fd());
if (!x)
return 0;
return *x;
}
void add_to_loop() override {
acceptor_.activate(this);
} }
auto scrb = make_counted<scribe_impl>(dm, fd, std::move(sssn));
void remove_from_loop() override { auto hdl = scrb->hdl();
acceptor_.passivate(); parent()->add_scribe(std::move(scrb));
return doorman::new_connection(&dm, hdl);
} }
private:
io::network::acceptor_impl<ssl_policy> acceptor_;
intrusive_ptr<session> session_;
}; };
class middleman_actor_impl : public io::middleman_actor_impl { class middleman_actor_impl : public io::middleman_actor_impl {
...@@ -230,26 +205,35 @@ public: ...@@ -230,26 +205,35 @@ public:
// nop // nop
} }
const char* name() const override {
return "openssl::middleman_actor";
}
protected: protected:
expected<io::scribe_ptr> connect(const std::string& host, expected<io::scribe_ptr> connect(const std::string& host,
uint16_t port) override { uint16_t port) override {
CAF_LOG_TRACE(CAF_ARG(host) << CAF_ARG(port));
auto fd = io::network::new_tcp_connection(host, port); auto fd = io::network::new_tcp_connection(host, port);
if (fd == io::network::invalid_native_socket) if (fd == io::network::invalid_native_socket)
return std::move(fd.error()); return std::move(fd.error());
auto session = new class session(system()); io::network::nonblocking(*fd, true);
if (session->connect(*fd)) auto sssn = make_session(system(), *fd, false);
return make_counted<scribe_impl>(mpx(), *fd, session); if (!sssn) {
else CAF_LOG_ERROR("Unable to create SSL session for connection");
return sec::cannot_connect_to_node; return sec::cannot_connect_to_node;
} }
CAF_LOG_DEBUG("successfully created an SSL session for:"
<< CAF_ARG(host) << CAF_ARG(port));
return make_counted<scribe_impl>(mpx(), *fd, std::move(sssn));
}
expected<io::doorman_ptr> open(uint16_t port, const char* addr, expected<io::doorman_ptr> open(uint16_t port, const char* addr,
bool reuse) override { bool reuse) override {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(reuse));
auto fd = io::network::new_tcp_acceptor_impl(port, addr, reuse); auto fd = io::network::new_tcp_acceptor_impl(port, addr, reuse);
if (fd == io::network::invalid_native_socket) if (fd == io::network::invalid_native_socket)
return std::move(fd.error()); return std::move(fd.error());
auto session = new class session(system()); return make_counted<doorman_impl>(mpx(), *fd);
return make_counted<doorman_impl>(mpx(), *fd, session);
} }
private: private:
......
...@@ -40,6 +40,8 @@ expected<strong_actor_ptr> remote_actor(actor_system& sys, ...@@ -40,6 +40,8 @@ expected<strong_actor_ptr> remote_actor(actor_system& sys,
CAF_LOG_TRACE(CAF_ARG(mpi) << CAF_ARG(host) << CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(mpi) << CAF_ARG(host) << CAF_ARG(port));
expected<strong_actor_ptr> res{strong_actor_ptr{nullptr}}; expected<strong_actor_ptr> res{strong_actor_ptr{nullptr}};
scoped_actor self{sys}; scoped_actor self{sys};
auto mm = sys.openssl_manager().actor_handle();
CAF_ASSERT(mm != nullptr);
self->request(sys.openssl_manager().actor_handle(), infinite, self->request(sys.openssl_manager().actor_handle(), infinite,
connect_atom::value, std::move(host), port) connect_atom::value, std::move(host), port)
.receive( .receive(
......
...@@ -19,11 +19,15 @@ ...@@ -19,11 +19,15 @@
#include "caf/openssl/session.hpp" #include "caf/openssl/session.hpp"
CAF_PUSH_WARNINGS
#include <sys/socket.h> #include <sys/socket.h>
#include <openssl/err.h> #include <openssl/err.h>
CAF_POP_WARNINGS
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/io/network/default_multiplexer.hpp"
#include "caf/openssl/manager.hpp" #include "caf/openssl/manager.hpp"
namespace caf { namespace caf {
...@@ -33,22 +37,31 @@ namespace { ...@@ -33,22 +37,31 @@ namespace {
int pem_passwd_cb(char* buf, int size, int, void* ptr) { int pem_passwd_cb(char* buf, int size, int, void* ptr) {
auto passphrase = reinterpret_cast<session*>(ptr)->openssl_passphrase(); auto passphrase = reinterpret_cast<session*>(ptr)->openssl_passphrase();
strncpy(buf, passphrase, size); strncpy(buf, passphrase, static_cast<size_t>(size));
buf[size - 1] = '\0'; buf[size - 1] = '\0';
return strlen(buf); return strlen(buf);
} }
} // namespace <anonymous> } // namespace <anonymous>
session::session(actor_system& sys) : sys_(sys) { session::session(actor_system& sys)
: sys_(sys),
ctx_(nullptr),
ssl_(nullptr),
connecting_(false),
accepting_(false) {
// nop // nop
} }
void session::init() { bool session::init() {
CAF_LOG_TRACE("");
ctx_ = create_ssl_context(); ctx_ = create_ssl_context();
ssl_ = SSL_new(ctx_); ssl_ = SSL_new(ctx_);
if (!ssl_) if (ssl_ == nullptr) {
CAF_RAISE_ERROR("cannot create SSL session"); CAF_LOG_ERROR("cannot create SSL session");
return false;
}
return true;
} }
session::~session() { session::~session() {
...@@ -56,47 +69,99 @@ session::~session() { ...@@ -56,47 +69,99 @@ session::~session() {
SSL_CTX_free(ctx_); SSL_CTX_free(ctx_);
} }
bool session::read_some(size_t& result, native_socket, void* buf, size_t len) { rw_state session::do_some(int (*f)(SSL*, void*, int), size_t& result, void* buf,
CAF_LOG_TRACE(CAF_ARG(len)); size_t len, const char* debug_name) {
if (len == 0) auto check_ssl_res = [&](int res) -> rw_state {
return 0; result = 0;
auto ret = SSL_read(ssl_, buf, len); switch (SSL_get_error(ssl_, res)) {
default:
CAF_LOG_INFO("SSL error:" << get_ssl_error());
return rw_state::failure;
case SSL_ERROR_WANT_READ:
CAF_LOG_DEBUG("SSL_ERROR_WANT_READ reported");
// Report success to poll on this socket.
if (len == 0 && strcmp(debug_name, "write_some") == 0)
return rw_state::indeterminate;
return rw_state::success;
case SSL_ERROR_WANT_WRITE:
CAF_LOG_DEBUG("SSL_ERROR_WANT_WRITE reported");
// Report success to poll on this socket.
return rw_state::success;
}
};
CAF_LOG_TRACE(CAF_ARG(len) << CAF_ARG(debug_name));
CAF_IGNORE_UNUSED(debug_name);
if (connecting_) {
CAF_LOG_DEBUG(debug_name << ": connecting");
auto res = SSL_connect(ssl_);
if (res == 1) {
CAF_LOG_DEBUG("SSL connection established");
connecting_ = false;
} else {
result = 0;
return check_ssl_res(res);
}
}
if (accepting_) {
CAF_LOG_DEBUG(debug_name << ": accepting");
auto res = SSL_accept(ssl_);
if (res == 1) {
CAF_LOG_DEBUG("SSL connection accepted");
accepting_ = false;
} else {
result = 0;
return check_ssl_res(res);
}
}
CAF_LOG_DEBUG(debug_name << ": calling SSL_write or SSL_read");
if (len == 0) {
result = 0;
return rw_state::indeterminate;
}
auto ret = f(ssl_, buf, static_cast<int>(len));
if (ret > 0) { if (ret > 0) {
result = ret; result = static_cast<size_t>(ret);
return true; return rw_state::success;
} }
result = 0; result = 0;
return handle_ssl_result(ret); return handle_ssl_result(ret) ? rw_state::success : rw_state::failure;
} }
bool session::write_some(size_t& result, native_socket, const void* buf, rw_state session::read_some(size_t& result, native_socket, void* buf,
size_t len) { size_t len) {
CAF_LOG_TRACE(CAF_ARG(len)); CAF_LOG_TRACE(CAF_ARG(len));
if (len == 0) return do_some(SSL_read, result, buf, len, "read_some");
return true;
auto ret = SSL_write(ssl_, buf, len);
if (ret > 0) {
result = static_cast<size_t>(ret);
return true;
}
result = 0;
return handle_ssl_result(ret);
} }
bool session::connect(native_socket fd) { rw_state session::write_some(size_t& result, native_socket, const void* buf,
init(); size_t len) {
CAF_LOG_TRACE(CAF_ARG(len));
auto wr_fun = [](SSL* sptr, void* vptr, int ptr_size) {
return SSL_write(sptr, vptr, ptr_size);
};
return do_some(wr_fun, result, const_cast<void*>(buf), len, "write_some");
}
bool session::try_connect(native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
SSL_set_fd(ssl_, fd); SSL_set_fd(ssl_, fd);
SSL_set_connect_state(ssl_); SSL_set_connect_state(ssl_);
auto ret = SSL_connect(ssl_); auto ret = SSL_connect(ssl_);
return ret > 0 || handle_ssl_result(ret); if (ret == 1)
return true;
connecting_ = true;
return handle_ssl_result(ret);
} }
bool session::try_accept(native_socket fd) { bool session::try_accept(native_socket fd) {
init(); CAF_LOG_TRACE(CAF_ARG(fd));
SSL_set_fd(ssl_, fd); SSL_set_fd(ssl_, fd);
SSL_set_accept_state(ssl_); SSL_set_accept_state(ssl_);
auto ret = SSL_accept(ssl_); auto ret = SSL_accept(ssl_);
return ret > 0 || handle_ssl_result(ret); if (ret == 1)
return true;
accepting_ = true;
return handle_ssl_result(ret);
} }
const char* session::openssl_passphrase() { const char* session::openssl_passphrase() {
...@@ -118,7 +183,7 @@ SSL_CTX* session::create_ssl_context() { ...@@ -118,7 +183,7 @@ SSL_CTX* session::create_ssl_context() {
if (cfg.openssl_passphrase.size() > 0) { if (cfg.openssl_passphrase.size() > 0) {
openssl_passphrase_ = cfg.openssl_passphrase; openssl_passphrase_ = cfg.openssl_passphrase;
SSL_CTX_set_default_passwd_cb(ctx, pem_passwd_cb); SSL_CTX_set_default_passwd_cb(ctx, pem_passwd_cb);
SSL_CTX_set_default_passwd_cb_userdata(ctx, (void*)this); SSL_CTX_set_default_passwd_cb_userdata(ctx, this);
} }
if (cfg.openssl_key.size() > 0 if (cfg.openssl_key.size() > 0
&& SSL_CTX_use_PrivateKey_file(ctx, cfg.openssl_key.c_str(), && SSL_CTX_use_PrivateKey_file(ctx, cfg.openssl_key.c_str(),
...@@ -143,7 +208,9 @@ SSL_CTX* session::create_ssl_context() { ...@@ -143,7 +208,9 @@ SSL_CTX* session::create_ssl_context() {
auto ecdh = EC_KEY_new_by_curve_name(NID_secp384r1); auto ecdh = EC_KEY_new_by_curve_name(NID_secp384r1);
if (!ecdh) if (!ecdh)
raise_ssl_error("cannot get ECDH curve"); raise_ssl_error("cannot get ECDH curve");
CAF_PUSH_WARNINGS
SSL_CTX_set_tmp_ecdh(ctx, ecdh); SSL_CTX_set_tmp_ecdh(ctx, ecdh);
CAF_POP_WARNINGS
if (SSL_CTX_set_cipher_list(ctx, "AECDH-AES256-SHA") != 1) if (SSL_CTX_set_cipher_list(ctx, "AECDH-AES256-SHA") != 1)
raise_ssl_error("cannot set anonymous cipher"); raise_ssl_error("cannot set anonymous cipher");
} }
...@@ -169,18 +236,35 @@ void session::raise_ssl_error(std::string msg) { ...@@ -169,18 +236,35 @@ void session::raise_ssl_error(std::string msg) {
bool session::handle_ssl_result(int ret) { bool session::handle_ssl_result(int ret) {
auto err = SSL_get_error(ssl_, ret); auto err = SSL_get_error(ssl_, ret);
switch (err) { switch (err) {
case SSL_ERROR_WANT_READ: // Try again next time. case SSL_ERROR_WANT_READ:
CAF_LOG_DEBUG("Nonblocking call to SSL returned want_read");
return true;
case SSL_ERROR_WANT_WRITE: case SSL_ERROR_WANT_WRITE:
CAF_LOG_DEBUG("Nonblocking call to SSL returned want_write");
return true; return true;
case SSL_ERROR_ZERO_RETURN: // Regular remote connection shutdown. case SSL_ERROR_ZERO_RETURN: // Regular remote connection shutdown.
case SSL_ERROR_SYSCALL: // Socket connection closed. case SSL_ERROR_SYSCALL: // Socket connection closed.
return false; return false;
default: // Other error default: // Other error
// TODO: Log somehow? CAF_LOG_INFO("SSL call failed:" << get_ssl_error());
// auto msg = get_ssl_error();
return false; return false;
} }
} }
session_ptr make_session(actor_system& sys, native_socket fd,
bool from_accepted_socket) {
auto ptr = make_counted<session>(sys);
if (!ptr->init())
return nullptr;
if (from_accepted_socket) {
if (!ptr->try_accept(fd))
return nullptr;
} else {
if (!ptr->try_connect(fd))
return nullptr;
}
return ptr;
}
} // namespace openssl } // namespace openssl
} // namespace caf } // namespace caf
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