Commit 0744bcc7 authored by Dominik Charousset's avatar Dominik Charousset

Allow WebSocket clients to connect over TLS

parent 888058a9
......@@ -13,15 +13,11 @@
#include <iostream>
#include <utility>
static constexpr uint16_t default_port = 8080;
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<std::string>("host", "TCP endpoint to connect to")
.add<uint16_t>("port,p", "port for connecting to the host")
.add<std::string>("endpoint", "sets the Request-URI field")
.add<std::string>("protocols", "sets the Sec-WebSocket-Protocol field")
opt_group{custom_options_, "global"}
.add<caf::uri>("server,s", "URI for connecting to the server")
.add<std::string>("protocols,p", "sets the Sec-WebSocket-Protocol field")
.add<size_t>("max,m", "maximum number of message to receive");
}
};
......@@ -30,34 +26,27 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
namespace cn = caf::net;
namespace ws = cn::web_socket;
// Sanity checking.
auto host = caf::get_or(cfg, "host", "");
if (host.empty()) {
std::cerr << "*** missing mandatory argument: host\n";
auto server = caf::get_or(cfg, "server", caf::uri{});
if (!server.valid()) {
std::cerr << "*** mandatory argument server missing or invalid\n";
return EXIT_FAILURE;
}
// Ask user for the hello message.
std::string hello;
std::cout << "Please enter a hello message for the server: " << std::flush;
std::getline(std::cin, hello);
// Open up the TCP connection.
auto port = caf::get_or(cfg, "port", default_port);
cn::tcp_stream_socket fd;
if (auto maybe_fd = cn::make_connected_tcp_stream_socket(host, port)) {
std::cout << "*** connected to " << host << ':' << port << '\n';
fd = std::move(*maybe_fd);
} else {
std::cerr << "*** unable to connect to " << host << ':' << port << ": "
<< to_string(maybe_fd.error()) << '\n';
// Spin up the WebSocket.
auto hs_setup = [&cfg](ws::handshake& hs) {
if (auto str = caf::get_as<std::string>(cfg, "protocols");
str && !str->empty())
hs.protocols(std::move(*str));
};
auto conn = ws::connect(sys, server, hs_setup);
if (!conn) {
std::cerr << "*** failed to connect: " << to_string(conn.error()) << '\n';
return EXIT_FAILURE;
}
// Spin up the WebSocket.
ws::handshake hs;
hs.host(host);
hs.endpoint(caf::get_or(cfg, "endpoint", "/"));
if (auto str = caf::get_as<std::string>(cfg, "protocols");
str && !str->empty())
hs.protocols(std::move(*str));
ws::connect(sys, fd, std::move(hs), [&](const ws::connect_event_t& conn) {
conn->run([&](const ws::connect_event_t& conn) {
sys.spawn([conn, hello](caf::event_based_actor* self) {
auto [pull, push] = conn.data();
// Print everything from the server to stdout.
......@@ -73,6 +62,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
return std::move(in).as_observable();
}
})
.do_finally([] { std::cout << "Server has closed the connection\n"; })
.for_each([](const ws::frame& msg) {
if (msg.is_text()) {
std::cout << "Server: " << msg.as_text() << '\n';
......
......@@ -91,6 +91,9 @@ public:
/// The fragment component.
std::string fragment;
/// Offset to the path in `str`.
size_t path_offset = 0;
// -- properties -----------------------------------------------------------
bool valid() const noexcept {
......@@ -101,6 +104,10 @@ public:
return rc_.load() == 1;
}
std::string_view str_after_path_offset() const noexcept {
return {str.c_str() + path_offset, str.size() - path_offset};
}
// -- modifiers ------------------------------------------------------------
/// Assembles the human-readable string representation for this URI.
......@@ -182,6 +189,13 @@ public:
return impl_->fragment;
}
/// Returns the host sub-component of the authority as string.
std::string host_str() const;
/// Returns the path, query and fragment components (as they appear in the
/// encoded URI) with a leading '/'.
std::string path_query_fragment() const;
/// Returns a hash code over all components.
size_t hash_code() const noexcept;
......
......@@ -37,10 +37,12 @@ void uri::impl_type::assemble_str() {
str += ':';
if (authority.empty()) {
CAF_ASSERT(!path.empty());
path_offset = str.size();
uri::encode(str, path, true);
} else {
str += "//";
str += to_string(authority);
path_offset = str.size();
if (!path.empty()) {
str += '/';
uri::encode(str, path, true);
......@@ -74,6 +76,25 @@ uri::uri(impl_ptr ptr) : impl_(std::move(ptr)) {
CAF_ASSERT(impl_ != nullptr);
}
std::string uri::host_str() const {
const auto& host = impl_->authority.host;
if (std::holds_alternative<std::string>(host)) {
return std::get<std::string>(host);
} else {
return to_string(std::get<ip_address>(host));
}
}
std::string uri::path_query_fragment() const {
std::string result;
auto sub_str = impl_->str_after_path_offset();
if (sub_str.empty() || sub_str[0] != '/') {
result += '/';
}
result.insert(result.begin(), sub_str.begin(), sub_str.end());
return result;
}
size_t uri::hash_code() const noexcept {
return hash::fnv<size_t>::compute(str());
}
......
......@@ -78,6 +78,7 @@ caf_add_component(
src/net/this_host.cpp
src/net/udp_datagram_socket.cpp
src/net/web_socket/client.cpp
src/net/web_socket/connect.cpp
src/net/web_socket/default_trait.cpp
src/net/web_socket/frame.cpp
src/net/web_socket/framing.cpp
......
......@@ -4,12 +4,19 @@
#pragma once
#include "caf/async/spsc_buffer.hpp"
#include "caf/callback.hpp"
#include "caf/cow_tuple.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/disposable.hpp"
#include "caf/expected.hpp"
#include "caf/net/ssl/connection.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/web_socket/client.hpp"
#include "caf/net/web_socket/default_trait.hpp"
#include "caf/net/web_socket/flow_bridge.hpp"
#include "caf/net/web_socket/frame.hpp"
#include "caf/net/web_socket/handshake.hpp"
#include "caf/uri.hpp"
#include <utility>
namespace caf::net::web_socket {
......@@ -18,27 +25,88 @@ using connect_event_t
= cow_tuple<async::consumer_resource<frame>, // Socket to App.
async::producer_resource<frame>>; // App to Socket.
/// Tries to establish a WebSocket connection on @p fd.
/// @param sys The host system.
/// Wraps a successful connection setup.
class connect_state {
public:
connect_state() = delete;
connect_state(connect_state&&) = default;
connect_state(const connect_state&) = default;
connect_state& operator=(connect_state&&) = default;
connect_state& operator=(const connect_state&) = default;
connect_state(disposable worker, connect_event_t event)
: worker_(worker), event_(std::move(event)) {
// nop
}
template <class Init>
disposable run(Init init) {
init(std::move(event_));
return worker_;
}
private:
disposable worker_;
connect_event_t event_;
};
} // namespace caf::net::web_socket
namespace caf::detail {
using ws_handshake_setup = callback<void(net::web_socket::handshake&)>;
net::web_socket::connect_state CAF_NET_EXPORT ws_do_connect(
actor_system& sys, net::stream_socket fd, net::web_socket::handshake& hs);
net::web_socket::connect_state CAF_NET_EXPORT ws_do_connect(
actor_system& sys, net::ssl::connection conn, net::web_socket::handshake& hs);
expected<net::web_socket::connect_state>
CAF_NET_EXPORT ws_connect_impl(actor_system& sys, const caf::uri& dst,
ws_handshake_setup& setup);
} // namespace caf::detail
namespace caf::net::web_socket {
/// Starts a WebSocket connection on @p fd.
/// @param sys The parent system.
/// @param fd An connected socket.
/// @param init Function object for setting up the created flows.
template <class Transport = stream_transport, class Socket, class Init>
void connect(actor_system& sys, Socket fd, handshake hs, Init init) {
// TODO: connect() should return a disposable to stop the WebSocket.
using trait_t = default_trait;
template <class Init>
disposable
connect(actor_system& sys, stream_socket fd, handshake hs, Init init) {
static_assert(std::is_invocable_v<Init, connect_event_t&&>,
"invalid signature found for init");
auto [ws_pull, app_push] = async::make_spsc_buffer_resource<frame>();
auto [app_pull, ws_push] = async::make_spsc_buffer_resource<frame>();
using conn_t = flow_connector_trivial_impl<trait_t>;
auto mpx = sys.network_manager().mpx_ptr();
auto conn = std::make_shared<conn_t>(std::move(ws_pull), std::move(ws_push));
auto bridge = flow_bridge<trait_t>::make(mpx, std::move(conn));
auto impl = client::make(std::move(hs), std::move(bridge));
auto transport = Transport::make(fd, std::move(impl));
auto ptr = socket_manager::make(mpx, std::move(transport));
mpx->start(ptr);
init(connect_event_t{app_pull, app_push});
return detail::ws_do_connect(sys, fd, hs).run(std::move(init));
}
/// Starts a WebSocket connection on @p fd.
/// @param sys The parent system.
/// @param conn An established TCP with an TLS connection attached to it.
/// @param init Function object for setting up the created flows.
template <class Init>
disposable
connect(actor_system& sys, ssl::connection conn, handshake hs, Init init) {
static_assert(std::is_invocable_v<Init, connect_event_t&&>,
"invalid signature found for init");
return detail::ws_do_connect(sys, std::move(conn), hs).run(std::move(init));
}
/// Tries to connect to the host from the URI.
/// @param sys The parent system.
/// @param dst Encodes the destination host. URI scheme must be `ws` or `wss`.
/// @note Blocks the caller while trying to establish a TCP connection.
expected<connect_state> connect(actor_system& sys, const caf::uri& dst);
template <class HandshakeSetup>
expected<connect_state>
connect(actor_system& sys, const caf::uri& dst, HandshakeSetup&& setup) {
static_assert(std::is_invocable_v<HandshakeSetup, handshake&>,
"invalid signature found for the handshake setup function");
auto cb = make_callback(std::forward<HandshakeSetup>(setup));
return detail::ws_connect_impl(sys, dst, cb);
}
} // namespace caf::net::web_socket
// 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/web_socket/connect.hpp"
#include "caf/net/flow_connector.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/ssl/transport.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/net/web_socket/client.hpp"
#include "caf/net/web_socket/default_trait.hpp"
#include "caf/net/web_socket/flow_bridge.hpp"
namespace ws = caf::net::web_socket;
namespace caf::detail {
template <class Transport, class Connection>
ws::connect_state
ws_do_connect_impl(actor_system& sys, Connection conn, ws::handshake& hs) {
using trait_t = ws::default_trait;
using connector_t = net::flow_connector_trivial_impl<trait_t>;
auto [ws_pull, app_push] = async::make_spsc_buffer_resource<ws::frame>();
auto [app_pull, ws_push] = async::make_spsc_buffer_resource<ws::frame>();
auto mpx = sys.network_manager().mpx_ptr();
auto connector = std::make_shared<connector_t>(std::move(ws_pull),
std::move(ws_push));
auto bridge = ws::flow_bridge<trait_t>::make(mpx, std::move(connector));
auto impl = ws::client::make(std::move(hs), std::move(bridge));
auto fd = conn.fd();
auto transport = Transport::make(std::move(conn), std::move(impl));
transport->active_policy().connect(fd);
auto ptr = net::socket_manager::make(mpx, std::move(transport));
mpx->start(ptr);
return ws::connect_state{ptr->as_disposable(),
ws::connect_event_t{app_pull, app_push}};
}
ws::connect_state ws_do_connect(actor_system& sys, net::stream_socket fd,
ws::handshake& hs) {
return ws_do_connect_impl<net::stream_transport>(sys, fd, hs);
}
ws::connect_state ws_do_connect(actor_system& sys, net::ssl::connection conn,
ws::handshake& hs) {
return ws_do_connect_impl<net::ssl::transport>(sys, std::move(conn), hs);
}
expected<ws::connect_state> ws_connect_impl(actor_system& sys,
const caf::uri& dst,
ws_handshake_setup& setup) {
if (dst.scheme() != "ws" && dst.scheme() != "wss") {
return {make_error(sec::invalid_argument, "URI must use ws or wss scheme")};
}
auto fd = net::make_connected_tcp_stream_socket(dst.authority());
if (!fd) {
return {std::move(fd.error())};
}
net::web_socket::handshake hs;
hs.host(dst.host_str());
hs.endpoint(dst.path_query_fragment());
setup(hs);
if (dst.scheme() == "ws") {
return {ws_do_connect(sys, *fd, hs)};
} else {
auto ctx = net::ssl::context::make_client(net::ssl::tls::any);
if (!ctx) {
return {std::move(ctx.error())};
}
auto conn = ctx->new_connection(*fd);
if (!conn) {
return {std::move(conn.error())};
}
return {ws_do_connect(sys, std::move(*conn), hs)};
}
}
} // namespace caf::detail
namespace caf::net::web_socket {
expected<connect_state> connect(actor_system& sys, const caf::uri& dst) {
auto cb = make_callback([](net::web_socket::handshake&) {});
return detail::ws_connect_impl(sys, dst, cb);
}
} // namespace caf::net::web_socket
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