Commit bf227239 authored by Joseph Noir's avatar Joseph Noir

Udp handshake received, not answered

parent f6fbd101
...@@ -102,7 +102,9 @@ inline bool operator!=(const header& lhs, const header& rhs) { ...@@ -102,7 +102,9 @@ inline bool operator!=(const header& lhs, const header& rhs) {
/// Checks whether given header contains a handshake. /// Checks whether given header contains a handshake.
inline bool is_handshake(const header& hdr) { inline bool is_handshake(const header& hdr) {
return hdr.operation == message_type::server_handshake return hdr.operation == message_type::server_handshake
|| hdr.operation == message_type::client_handshake; || hdr.operation == message_type::client_handshake
|| hdr.operation == message_type::udp_client_handshake
|| hdr.operation == message_type::udp_server_handshake;
} }
/// Checks wheter given header contains a heartbeat. /// Checks wheter given header contains a heartbeat.
......
...@@ -69,6 +69,10 @@ enum class message_type : uint8_t { ...@@ -69,6 +69,10 @@ enum class message_type : uint8_t {
/// ///
/// ![](heartbeat.png) /// ![](heartbeat.png)
heartbeat = 0x05, heartbeat = 0x05,
// TODO: udp handshakes
udp_client_handshake = 0x06,
udp_server_handshake = 0x07
}; };
/// @relates message_type /// @relates message_type
......
...@@ -43,7 +43,7 @@ constexpr invalid_dgram_doorman_handle_t invalid_dgram_doorman_handle ...@@ -43,7 +43,7 @@ constexpr invalid_dgram_doorman_handle_t invalid_dgram_doorman_handle
/// Generic type for identifying a datagram source. /// Generic type for identifying a datagram source.
class dgram_doorman_handle : public handle<dgram_doorman_handle, class dgram_doorman_handle : public handle<dgram_doorman_handle,
invalid_dgram_doorman_handle_t> { invalid_dgram_doorman_handle_t> {
public: public:
friend class handle<dgram_doorman_handle, invalid_dgram_doorman_handle_t>; friend class handle<dgram_doorman_handle, invalid_dgram_doorman_handle_t>;
......
...@@ -597,7 +597,11 @@ behavior basp_broker::make_behavior() { ...@@ -597,7 +597,11 @@ behavior basp_broker::make_behavior() {
<< CAF_ARG(msg.handle)); << CAF_ARG(msg.handle));
ctx.callback->deliver(make_error(sec::disconnect_during_handshake)); ctx.callback->deliver(make_error(sec::disconnect_during_handshake));
} }
// TODO: some error?
} else { } else {
if (ctx.callback) {
// TODO: if callback available, handle that
}
// TODO: Is there configuration necessary? // TODO: Is there configuration necessary?
// configure_datagram_size(...); // configure_datagram_size(...);
} }
...@@ -693,9 +697,11 @@ behavior basp_broker::make_behavior() { ...@@ -693,9 +697,11 @@ behavior basp_broker::make_behavior() {
// flush(msg.handle); // flush(msg.handle);
// TODO: is this right? // TODO: is this right?
}, },
/*
[=](const dgram_delegate_msg&) { [=](const dgram_delegate_msg&) {
std::cerr << "[BB] Received new delegate message" << std::endl; CAF_LOG_TRACE(CAF_ARG(msg.handle));
}, },
*/
// received from underlying broker implementation // received from underlying broker implementation
[=](const dgram_scribe_closed_msg& msg) { [=](const dgram_scribe_closed_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
...@@ -709,12 +715,6 @@ behavior basp_broker::make_behavior() { ...@@ -709,12 +715,6 @@ behavior basp_broker::make_behavior() {
CAF_ASSERT(nid == none CAF_ASSERT(nid == none
|| !state.instance.tbl().reachable(nid)); || !state.instance.tbl().reachable(nid));
}, },
// received from underlying broker implementation
[=](const dgram_acceptor_closed_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
auto port = local_port(msg.handle);
state.instance.remove_published_actor(port);
},
[=](const dgram_doorman_closed_msg& msg) { [=](const dgram_doorman_closed_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
std::cerr << "[BB] Received dgram_doorman_closed_msg" << std::endl; std::cerr << "[BB] Received dgram_doorman_closed_msg" << std::endl;
...@@ -777,7 +777,8 @@ behavior basp_broker::make_behavior() { ...@@ -777,7 +777,8 @@ behavior basp_broker::make_behavior() {
[=](connect_atom, dgram_scribe_handle hdl, [=](connect_atom, dgram_scribe_handle hdl,
const std::string& host, uint16_t port) { const std::string& host, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(hdl.id())); CAF_LOG_TRACE(CAF_ARG(hdl.id()));
std::cerr << "[BB] connect for dgram to " << host << ":" << port << std::endl; std::cerr << "[BB] \"connect\" for dgram to "
<< host << ":" << port << std::endl;
auto rp = make_response_promise(); auto rp = make_response_promise();
auto res = assign_dgram_scribe(hdl, host, port); auto res = assign_dgram_scribe(hdl, host, port);
if (res) { if (res) {
...@@ -786,8 +787,8 @@ behavior basp_broker::make_behavior() { ...@@ -786,8 +787,8 @@ behavior basp_broker::make_behavior() {
ctx.remote_port = port; ctx.remote_port = port;
ctx.callback = rp; ctx.callback = rp;
auto& bi = state.instance; auto& bi = state.instance;
//bi.write_server_handshake(context(), wr_buf(hdl), port); // bi.write_server_handshake(context(), wr_buf(hdl), port);
//bi.write_client_handshake(context(), wr_buf(hdl), local_port(hdl)); // bi.write_client_handshake(context(), wr_buf(hdl), local_port(hdl));
bi.write_udp_client_handshake(context(), wr_buf(hdl)); bi.write_udp_client_handshake(context(), wr_buf(hdl));
flush(hdl); flush(hdl);
configure_datagram_size(hdl, 1500); configure_datagram_size(hdl, 1500);
......
...@@ -58,7 +58,7 @@ bool dgram_doorman::delegate_msg(execution_unit* ctx, ...@@ -58,7 +58,7 @@ bool dgram_doorman::delegate_msg(execution_unit* ctx,
CAF_LOG_TRACE(CAF_ARG(endpoint) << CAF_ARG(num_bytes)); CAF_LOG_TRACE(CAF_ARG(endpoint) << CAF_ARG(num_bytes));
if (detached()) if (detached())
return false; return false;
using delegate_t = dgram_delegate_msg; using delegate_t = new_datagram_msg; // dgram_delegate_msg;
using tmp_t = mailbox_element_vals<delegate_t>; using tmp_t = mailbox_element_vals<delegate_t>;
auto guard = parent_; auto guard = parent_;
auto& buf = rd_buf(); auto& buf = rd_buf();
...@@ -68,7 +68,7 @@ bool dgram_doorman::delegate_msg(execution_unit* ctx, ...@@ -68,7 +68,7 @@ bool dgram_doorman::delegate_msg(execution_unit* ctx,
// msg_buf.resize(num_bytes); // msg_buf.resize(num_bytes);
tmp_t tmp{strong_actor_ptr{}, message_id::make(), tmp_t tmp{strong_actor_ptr{}, message_id::make(),
mailbox_element::forwarding_stack{}, mailbox_element::forwarding_stack{},
delegate_t{hdl(), endpoint, std::move(buf)}}; delegate_t{endpoint, std::move(buf)}};
invoke_mailbox_element_impl(ctx,tmp); invoke_mailbox_element_impl(ctx,tmp);
return true; return true;
} }
......
...@@ -87,6 +87,22 @@ bool client_handshake_valid(const header& hdr) { ...@@ -87,6 +87,22 @@ bool client_handshake_valid(const header& hdr) {
&& zero(hdr.operation_data); && zero(hdr.operation_data);
} }
bool udp_server_handshake_valid(const header& hdr) {
return valid(hdr.source_node)
&& valid(hdr.dest_node)
&& hdr.source_node != hdr.dest_node
&& zero(hdr.source_actor)
&& zero(hdr.dest_actor)
&& zero(hdr.operation_data);
}
bool udp_client_handshake_valid(const header& hdr) {
return valid(hdr.source_node)
&& !valid(hdr.dest_node)
&& zero(hdr.dest_actor)
&& !zero(hdr.operation_data);
}
bool dispatch_message_valid(const header& hdr) { bool dispatch_message_valid(const header& hdr) {
return valid(hdr.dest_node) return valid(hdr.dest_node)
&& (!zero(hdr.dest_actor) || hdr.has(header::named_receiver_flag)) && (!zero(hdr.dest_actor) || hdr.has(header::named_receiver_flag))
...@@ -141,6 +157,10 @@ bool valid(const header& hdr) { ...@@ -141,6 +157,10 @@ bool valid(const header& hdr) {
return kill_proxy_instance_valid(hdr); return kill_proxy_instance_valid(hdr);
case message_type::heartbeat: case message_type::heartbeat:
return heartbeat_valid(hdr); return heartbeat_valid(hdr);
case message_type::udp_server_handshake:
return udp_server_handshake_valid(hdr);
case message_type::udp_client_handshake:
return udp_client_handshake_valid(hdr);
} }
} }
......
...@@ -167,7 +167,7 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -167,7 +167,7 @@ connection_state instance::handle(execution_unit* ctx,
break; break;
} }
case message_type::client_handshake: { case message_type::client_handshake: {
if (tbl_.lookup_hdl(hdr.source_node)) { if (tbl_.lookup_hdl(hdr.source_node)) {
CAF_LOG_INFO("received second client handshake:" CAF_LOG_INFO("received second client handshake:"
<< CAF_ARG(hdr.source_node)); << CAF_ARG(hdr.source_node));
break; break;
...@@ -258,11 +258,291 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -258,11 +258,291 @@ connection_state instance::handle(execution_unit* ctx,
} }
bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) { bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
auto err = [&]() -> bool {
auto cb = make_callback([&](const node_id& nid) -> error {
callee_.purge_state(nid);
return none;
});
tbl_.erase(dm.handle, cb);
return false;
};
std::cerr << "New datagram received!" << std::endl;
static_cast<void>(ctx); static_cast<void>(ctx);
static_cast<void>(dm); static_cast<void>(dm);
static_cast<void>(hdr); static_cast<void>(hdr);
std::cerr << "New datagram received!" << std::endl; std::vector<char> hdr_buf{std::begin(dm.buf),
return false; std::begin(dm.buf) + basp::header_size};
std::vector<char> pl_buf{std::begin(dm.buf) + basp::header_size,
std::end(dm.buf)};
binary_deserializer bd{ctx, hdr_buf};
auto e = bd(hdr);
if (e || !valid(hdr)) {
CAF_LOG_WARNING("received invalid header:" << CAF_ARG(hdr));
std::cerr << "Received invalid header!" << std::endl;
return err();
}
if (hdr.payload_len > 0) {
// TODO: handle payload
}
CAF_LOG_DEBUG(CAF_ARG(hdr));
/*
// needs forwarding?
if (!is_handshake(hdr) && !is_heartbeat(hdr) && hdr.dest_node != this_node_) {
CAF_LOG_DEBUG("forward message");
auto path = lookup(hdr.dest_node);
if (path) {
binary_serializer bs{ctx, path->wr_buf};
auto e = bs(hdr);
if (e)
return err();
if (payload)
bs.apply_raw(payload->size(), payload->data());
tbl_.flush(*path);
notify<hook::message_forwarded>(hdr, payload);
} else {
CAF_LOG_INFO("cannot forward message, no route to destination");
if (hdr.source_node != this_node_) {
// TODO: signalize error back to sending node
auto reverse_path = lookup(hdr.source_node);
if (!reverse_path) {
CAF_LOG_WARNING("cannot send error message: no route to source");
} else {
CAF_LOG_WARNING("not implemented yet: signalize forward failure");
}
} else {
CAF_LOG_WARNING("lost packet with probably spoofed source");
}
notify<hook::message_forwarding_failed>(hdr, payload);
}
return await_header;
}
// function object for checking payload validity
auto payload_valid = [&]() -> bool {
return payload != nullptr && payload->size() == hdr.payload_len;
};
*/
// handle message to ourselves
switch (hdr.operation) {
case message_type::udp_server_handshake: {
std::cerr << "[I] message_type::udp_server_handshake" << std::endl;
break;
}
case message_type::udp_client_handshake: {
std::cerr << "[I] message_type::udp_client_handshake" << std::endl;
// udp scribes start with a client handshake when sending the first
// message, should be answered by a server handshake!
/*
actor_id aid = invalid_actor_id;
std::set<std::string> sigs;
if (payload_valid()) {
binary_deserializer bd{ctx, *payload};
std::string remote_appid;
auto e = bd(remote_appid);
if (e)
return err();
if (remote_appid != callee_.system().config().middleman_app_identifier) {
CAF_LOG_ERROR("app identifier mismatch");
return err();
}
e = bd(aid, sigs);
if (e)
return err();
} else {
CAF_LOG_ERROR("fail to receive the app identifier");
return err();
}
// close self connection after handshake is done
if (hdr.source_node == this_node_) {
CAF_LOG_INFO("close connection to self immediately");
callee_.finalize_handshake(hdr.source_node, aid, sigs);
return err();
}
// close this connection if we already have a direct connection
if (tbl_.lookup_hdl(hdr.source_node)) {
CAF_LOG_INFO("close connection since we already have a "
"direct connection: " << CAF_ARG(hdr.source_node));
callee_.finalize_handshake(hdr.source_node, aid, sigs);
return err();
}
// add direct route to this node and remove any indirect entry
CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node));
tbl_.add(dm.handle, hdr.source_node);
// auto was_indirect = tbl_.erase_indirect(hdr.source_node);
// write handshake as client in response
auto path = tbl_.lookup(hdr.source_node);
if (!path) {
CAF_LOG_ERROR("no route to host after server handshake");
return err();
}
write_client_handshake(ctx, path->wr_buf, hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node);
callee_.finalize_handshake(hdr.source_node, aid, sigs);
flush(*path);
*/
break;
}
case message_type::server_handshake: {
std::cerr << "[I] message_type::server_handshake" << std::endl;
/*
actor_id aid = invalid_actor_id;
std::set<std::string> sigs;
if (payload_valid()) {
binary_deserializer bd{ctx, *payload};
std::string remote_appid;
auto e = bd(remote_appid);
if (e)
return err();
if (remote_appid != callee_.system().config().middleman_app_identifier) {
CAF_LOG_ERROR("app identifier mismatch");
return err();
}
e = bd(aid, sigs);
if (e)
return err();
} else {
CAF_LOG_ERROR("fail to receive the app identifier");
return err();
}
// close self connection after handshake is done
if (hdr.source_node == this_node_) {
CAF_LOG_INFO("close connection to self immediately");
callee_.finalize_handshake(hdr.source_node, aid, sigs);
return err();
}
// close this connection if we already have a direct connection
if (tbl_.lookup_hdl(hdr.source_node)) {
CAF_LOG_INFO("close connection since we already have a "
"direct connection: " << CAF_ARG(hdr.source_node));
callee_.finalize_handshake(hdr.source_node, aid, sigs);
return err();
}
// add direct route to this node and remove any indirect entry
CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node));
tbl_.add(dm.handle, hdr.source_node);
// auto was_indirect = tbl_.erase_indirect(hdr.source_node);
// write handshake as client in response
auto path = tbl_.lookup(hdr.source_node);
if (!path) {
CAF_LOG_ERROR("no route to host after server handshake");
return err();
}
write_client_handshake(ctx, path->wr_buf, hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node);
callee_.finalize_handshake(hdr.source_node, aid, sigs);
flush(*path);
*/
break;
}
case message_type::client_handshake: {
std::cerr << "[I] message_type::client_handshake" << std::endl;
/*
if (tbl_.lookup_hdl(hdr.source_node)) {
CAF_LOG_INFO("received second client handshake:"
<< CAF_ARG(hdr.source_node));
break;
}
if (payload_valid()) {
binary_deserializer bd{ctx, *payload};
std::string remote_appid;
auto e = bd(remote_appid);
if (e)
return err();
if (remote_appid != callee_.system().config().middleman_app_identifier) {
CAF_LOG_ERROR("app identifier mismatch");
return err();
}
} else {
CAF_LOG_ERROR("fail to receive the app identifier");
return err();
}
// add direct route to this node and remove any indirect entry
CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node));
tbl_.add(dm.handle, hdr.source_node);
// auto was_indirect = tbl_.erase_indirect(hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node);
*/
break;
}
case message_type::dispatch_message: {
std::cerr << "[I] message_type::dispatch_message" << std::endl;
/*
if (!payload_valid())
return err();
// in case the sender of this message was received via a third node,
// we assume that that node to offers a route to the original source
auto last_hop = tbl_.lookup_node(dm.handle);
// TODO: if we don't need this anymore, what does the rest do here?
//if (hdr.source_node != none
// && hdr.source_node != this_node_
// && last_hop != hdr.source_node
// && tbl_.lookup_hdl(hdr.source_node))
// && tbl_.add_indirect(last_hop, hdr.source_node))
// callee_.learned_new_node_indirectly(hdr.source_node);
binary_deserializer bd{ctx, *payload};
auto receiver_name = static_cast<atom_value>(0);
std::vector<strong_actor_ptr> forwarding_stack;
message msg;
if (hdr.has(header::named_receiver_flag)) {
auto e = bd(receiver_name);
if (e)
return err();
}
auto e = bd(forwarding_stack, msg);
if (e)
return err();
CAF_LOG_DEBUG(CAF_ARG(forwarding_stack) << CAF_ARG(msg));
if (hdr.has(header::named_receiver_flag))
callee_.deliver(hdr.source_node, hdr.source_actor, receiver_name,
message_id::from_integer_value(hdr.operation_data),
forwarding_stack, msg);
else
callee_.deliver(hdr.source_node, hdr.source_actor, hdr.dest_actor,
message_id::from_integer_value(hdr.operation_data),
forwarding_stack, msg);
*/
return err();
break;
}
case message_type::announce_proxy:
std::cerr << "[I] message_type::announce_proxy" << std::endl;
/*
callee_.proxy_announced(hdr.source_node, hdr.dest_actor);
*/
break;
case message_type::kill_proxy: {
std::cerr << "[I] message_type::kill_proxy" << std::endl;
/*
if (!payload_valid())
return err();
binary_deserializer bd{ctx, *payload};
error fail_state;
auto e = bd(fail_state);
if (e)
return err();
callee_.kill_proxy(hdr.source_node, hdr.source_actor, fail_state);
*/
break;
}
case message_type::heartbeat: {
std::cerr << "[I] message_type::heartbeat" << std::endl;
/*
CAF_LOG_TRACE("received heartbeat: " << CAF_ARG(hdr.source_node));
std::cerr << "[I] Received hearbeat: " << to_string(hdr.source_node)
<< std::endl;
callee_.handle_heartbeat(hdr.source_node);
*/
return true;
}
default:
CAF_LOG_ERROR("invalid operation");
std::cerr << "[I] Invalid operation" << std::endl;
return err();
}
// TODO: Is this reachable?
return err();
}; };
void instance::handle_heartbeat(execution_unit* ctx) { void instance::handle_heartbeat(execution_unit* ctx) {
...@@ -460,7 +740,7 @@ void instance::write_client_handshake(execution_unit* ctx, ...@@ -460,7 +740,7 @@ void instance::write_client_handshake(execution_unit* ctx,
void instance::write_udp_client_handshake(execution_unit* ctx, void instance::write_udp_client_handshake(execution_unit* ctx,
buffer_type& buf) { buffer_type& buf) {
header hdr{message_type::client_handshake, 0, 0, 0, header hdr{message_type::udp_client_handshake, 0, 0, version,
this_node_, none, invalid_actor_id, invalid_actor_id}; this_node_, none, invalid_actor_id, invalid_actor_id};
write(ctx, buf, hdr); write(ctx, buf, hdr);
} }
......
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