From 4ce01689166f201c552e45b4aa3da351f9ef5ae4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Wed, 11 Oct 2017 23:33:32 +0200 Subject: [PATCH] network engine: move ParsedMessage to separate file --- CMakeLists.txt | 1 + src/Makefile.am | 2 +- src/network_engine.cpp | 275 +------------------------------------ src/parsed_message.h | 303 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 306 insertions(+), 275 deletions(-) create mode 100644 src/parsed_message.h diff --git a/CMakeLists.txt b/CMakeLists.txt index 1a7241de..da8731da 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -107,6 +107,7 @@ list (APPEND opendht_SOURCES src/storage.h src/listener.h src/search.h + src/parsed_message.h src/request.h src/callbacks.cpp src/routing_table.cpp diff --git a/src/Makefile.am b/src/Makefile.am index 60536d99..634f2d88 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -9,6 +9,7 @@ libopendht_la_SOURCES = \ listener.h \ request.h \ search.h \ + parsed_message.h \ node_cache.cpp \ callbacks.cpp \ routing_table.cpp \ @@ -82,4 +83,3 @@ noinst_HEADERS = \ @top_builddir@/argon2/src/encoding.h \ @top_builddir@/argon2/src/thread.h endif - diff --git a/src/network_engine.cpp b/src/network_engine.cpp index 40c6b5fd..2b02c427 100644 --- a/src/network_engine.cpp +++ b/src/network_engine.cpp @@ -22,6 +22,7 @@ #include "request.h" #include "default_types.h" #include "log_enable.h" +#include "parsed_message.h" #include @@ -95,63 +96,6 @@ set_nonblocking(int fd, int nonblocking) #endif -enum class MessageType { - Error = 0, - Reply, - Ping, - FindNode, - GetValues, - AnnounceValue, - Refresh, - Listen, - ValueData -}; - -struct ParsedMessage { - MessageType type; - /* Node ID of the sender */ - InfoHash id; - /* Network id */ - NetId network {0}; - /* hash for which values are requested */ - InfoHash info_hash; - /* target id around which to find nodes */ - InfoHash target; - /* transaction id */ - TransId tid; - /* tid for packets going through request socket */ - TransId socket_id; - /* security token */ - Blob token; - /* the value id (announce confirmation) */ - Value::Id value_id; - /* time when value was first created */ - time_point created { time_point::max() }; - /* IPv4 nodes in response to a 'find' request */ - Blob nodes4_raw, nodes6_raw; - std::vector> nodes4, nodes6; - /* values to store or retreive request */ - std::vector> values; - /* index for fields values */ - std::vector> fields; - /** When part of the message header: {index -> (total size, {})} - * When part of partial value data: {index -> (offset, part_data)} */ - std::map> value_parts; - /* query describing a filter to apply on values. */ - Query query; - /* states if ipv4 or ipv6 request */ - want_t want; - /* error code in case of error */ - uint16_t error_code; - /* reported address by the distant node */ - std::string ua; - SockAddr addr; - void msgpack_unpack(msgpack::object o); - - bool append(const ParsedMessage& block); - bool complete(); -}; - struct NetworkEngine::PartialMessage { SockAddr from; time_point start; @@ -1250,185 +1194,6 @@ NetworkEngine::sendError(const SockAddr& addr, send(buffer.data(), buffer.size(), 0, addr); } -void -ParsedMessage::msgpack_unpack(msgpack::object msg) -{ - auto y = findMapValue(msg, "y"); - auto r = findMapValue(msg, "r"); - auto e = findMapValue(msg, "e"); - auto v = findMapValue(msg, "p"); - - if (auto t = findMapValue(msg, "t")) - tid = {t->as>()}; - - if (auto rv = findMapValue(msg, "v")) - ua = rv->as(); - - if (auto netid = findMapValue(msg, "n")) - network = netid->as(); - - std::string q; - if (auto rq = findMapValue(msg, "q")) { - if (rq->type != msgpack::type::STR) - throw msgpack::type_error(); - q = rq->as(); - } - - if (e) - type = MessageType::Error; - else if (r) - type = MessageType::Reply; - else if (v) - type = MessageType::ValueData; - else if (y and y->as() != "q") - throw msgpack::type_error(); - else if (q == "ping") - type = MessageType::Ping; - else if (q == "find") - type = MessageType::FindNode; - else if (q == "get") - type = MessageType::GetValues; - else if (q == "listen") - type = MessageType::Listen; - else if (q == "put") - type = MessageType::AnnounceValue; - else if (q == "refresh") - type = MessageType::Refresh; - else - throw msgpack::type_error(); - - if (type == MessageType::ValueData) { - if (v->type != msgpack::type::MAP) - throw msgpack::type_error(); - for (size_t i = 0; i < v->via.map.size; ++i) { - auto& vdat = v->via.map.ptr[i]; - auto o = findMapValue(vdat.val, "o"); - auto d = findMapValue(vdat.val, "d"); - if (not o or not d) - continue; - value_parts.emplace(vdat.key.as(), std::pair(o->as(), unpackBlob(*d))); - } - return; - } - - auto a = findMapValue(msg, "a"); - if (!a && !r && !e) - throw msgpack::type_error(); - auto& req = a ? *a : (r ? *r : *e); - - if (e) { - if (e->type != msgpack::type::ARRAY) - throw msgpack::type_error(); - error_code = e->via.array.ptr[0].as(); - } - - if (auto t = findMapValue(req, "sid")) - socket_id = {t->as>()}; - - if (auto rid = findMapValue(req, "id")) - id = {*rid}; - - if (auto rh = findMapValue(req, "h")) - info_hash = {*rh}; - - if (auto rtarget = findMapValue(req, "target")) - target = {*rtarget}; - - if (auto rquery = findMapValue(req, "q")) - query.msgpack_unpack(*rquery); - - if (auto otoken = findMapValue(req, "token")) - token = unpackBlob(*otoken); - - if (auto vid = findMapValue(req, "vid")) - value_id = vid->as(); - - if (auto rnodes4 = findMapValue(req, "n4")) - nodes4_raw = unpackBlob(*rnodes4); - - if (auto rnodes6 = findMapValue(req, "n6")) - nodes6_raw = unpackBlob(*rnodes6); - - if (auto sa = findMapValue(req, "sa")) { - if (sa->type != msgpack::type::BIN) - throw msgpack::type_error(); - auto l = sa->via.bin.size; - if (l == sizeof(in_addr)) { - addr.setFamily(AF_INET); - auto& a = addr.getIPv4(); - std::fill_n((uint8_t*)&a, sizeof(a), 0); - a.sin_port = 0; - std::copy_n(sa->via.bin.ptr, l, (char*)&a.sin_addr); - } else if (l == sizeof(in6_addr)) { - addr.setFamily(AF_INET6); - auto& a = addr.getIPv6(); - std::fill_n((uint8_t*)&a, sizeof(a), 0); - a.sin6_port = 0; - std::copy_n(sa->via.bin.ptr, l, (char*)&a.sin6_addr); - } - } else - addr = {}; - - if (auto rcreated = findMapValue(req, "c")) - created = from_time_t(rcreated->as()); - - if (auto rvalues = findMapValue(req, "values")) { - if (rvalues->type != msgpack::type::ARRAY) - throw msgpack::type_error(); - for (size_t i = 0; i < rvalues->via.array.size; i++) { - auto& packed_v = rvalues->via.array.ptr[i]; - if (packed_v.type == msgpack::type::POSITIVE_INTEGER) { - // Skip oversize values with a small margin for header overhead - if (packed_v.via.u64 > MAX_VALUE_SIZE + 32) - continue; - value_parts.emplace(i, std::make_pair(packed_v.via.u64, Blob{})); - } else { - try { - values.emplace_back(std::make_shared(rvalues->via.array.ptr[i])); - } catch (const std::exception& e) { - //DHT_LOG_WARN("Error reading value: %s", e.what()); - } - } - } - } else if (auto raw_fields = findMapValue(req, "fields")) { - if (auto rfields = findMapValue(*raw_fields, "f")) { - auto vfields = rfields->as>(); - if (auto rvalues = findMapValue(*raw_fields, "v")) { - if (rvalues->type != msgpack::type::ARRAY) - throw msgpack::type_error(); - size_t val_num = rvalues->via.array.size / vfields.size(); - for (size_t i = 0; i < val_num; ++i) { - try { - auto v = std::make_shared(); - v->msgpack_unpack_fields(vfields, *rvalues, i*vfields.size()); - fields.emplace_back(std::move(v)); - } catch (const std::exception& e) { } - } - } - } else { - throw msgpack::type_error(); - } - } - - if (auto w = findMapValue(req, "w")) { - if (w->type != msgpack::type::ARRAY) - throw msgpack::type_error(); - want = 0; - for (unsigned i=0; ivia.array.size; i++) { - auto& val = w->via.array.ptr[i]; - try { - auto w = val.as(); - if (w == AF_INET) - want |= WANT4; - else if(w == AF_INET6) - want |= WANT6; - } catch (const std::exception& e) {}; - } - } else { - want = -1; - } -} - void NetworkEngine::maintainRxBuffer(const TransId& tid) { @@ -1443,44 +1208,6 @@ NetworkEngine::maintainRxBuffer(const TransId& tid) } } -bool -ParsedMessage::append(const ParsedMessage& block) -{ - bool ret(false); - for (const auto& ve : block.value_parts) { - auto part_val = value_parts.find(ve.first); - if (part_val == value_parts.end() - || part_val->second.second.size() >= part_val->second.first) - continue; - // TODO: handle out-of-order packets - if (ve.second.first != part_val->second.second.size()) { - //std::cout << "skipping out-of-order packet" << std::endl; - continue; - } - ret = true; - part_val->second.second.insert(part_val->second.second.end(), - ve.second.second.begin(), - ve.second.second.end()); - } - return ret; -} - -bool -ParsedMessage::complete() -{ - for (auto& e : value_parts) { - //std::cout << "part " << e.first << ": " << e.second.second.size() << "/" << e.second.first << std::endl; - if (e.second.first > e.second.second.size()) - return false; - } - for (auto& e : value_parts) { - msgpack::unpacked msg; - msgpack::unpack(msg, (const char*)e.second.second.data(), e.second.second.size()); - values.emplace_back(std::make_shared(msg.get())); - } - return true; -} - } /* namespace net */ } /* namespace dht */ diff --git a/src/parsed_message.h b/src/parsed_message.h new file mode 100644 index 00000000..83c74ab2 --- /dev/null +++ b/src/parsed_message.h @@ -0,0 +1,303 @@ +/* + * Copyright (C) 2017 Savoir-faire Linux Inc. + * Author(s) : Adrien Béraud + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + #pragma once + +#include "infohash.h" +#include "sockaddr.h" + +#include + +namespace dht { +namespace net { + +enum class MessageType { + Error = 0, + Reply, + Ping, + FindNode, + GetValues, + AnnounceValue, + Refresh, + Listen, + ValueData +}; + +struct ParsedMessage { + MessageType type; + /* Node ID of the sender */ + InfoHash id; + /* Network id */ + NetId network {0}; + /* hash for which values are requested */ + InfoHash info_hash; + /* target id around which to find nodes */ + InfoHash target; + /* transaction id */ + TransId tid; + /* tid for packets going through request socket */ + TransId socket_id; + /* security token */ + Blob token; + /* the value id (announce confirmation) */ + Value::Id value_id; + /* time when value was first created */ + time_point created { time_point::max() }; + /* IPv4 nodes in response to a 'find' request */ + Blob nodes4_raw, nodes6_raw; + std::vector> nodes4, nodes6; + /* values to store or retreive request */ + std::vector> values; + /* index for fields values */ + std::vector> fields; + /** When part of the message header: {index -> (total size, {})} + * When part of partial value data: {index -> (offset, part_data)} */ + std::map> value_parts; + /* query describing a filter to apply on values. */ + Query query; + /* states if ipv4 or ipv6 request */ + want_t want; + /* error code in case of error */ + uint16_t error_code; + /* reported address by the distant node */ + std::string ua; + SockAddr addr; + void msgpack_unpack(msgpack::object o); + + bool append(const ParsedMessage& block); + bool complete(); +}; + +bool +ParsedMessage::append(const ParsedMessage& block) +{ + bool ret(false); + for (const auto& ve : block.value_parts) { + auto part_val = value_parts.find(ve.first); + if (part_val == value_parts.end() + || part_val->second.second.size() >= part_val->second.first) + continue; + // TODO: handle out-of-order packets + if (ve.second.first != part_val->second.second.size()) { + //std::cout << "skipping out-of-order packet" << std::endl; + continue; + } + ret = true; + part_val->second.second.insert(part_val->second.second.end(), + ve.second.second.begin(), + ve.second.second.end()); + } + return ret; +} + +bool +ParsedMessage::complete() +{ + for (auto& e : value_parts) { + //std::cout << "part " << e.first << ": " << e.second.second.size() << "/" << e.second.first << std::endl; + if (e.second.first > e.second.second.size()) + return false; + } + for (auto& e : value_parts) { + msgpack::unpacked msg; + msgpack::unpack(msg, (const char*)e.second.second.data(), e.second.second.size()); + values.emplace_back(std::make_shared(msg.get())); + } + return true; +} + +void +ParsedMessage::msgpack_unpack(msgpack::object msg) +{ + auto y = findMapValue(msg, "y"); + auto r = findMapValue(msg, "r"); + auto e = findMapValue(msg, "e"); + auto v = findMapValue(msg, "p"); + + if (auto t = findMapValue(msg, "t")) + tid = {t->as>()}; + + if (auto rv = findMapValue(msg, "v")) + ua = rv->as(); + + if (auto netid = findMapValue(msg, "n")) + network = netid->as(); + + std::string q; + if (auto rq = findMapValue(msg, "q")) { + if (rq->type != msgpack::type::STR) + throw msgpack::type_error(); + q = rq->as(); + } + + if (e) + type = MessageType::Error; + else if (r) + type = MessageType::Reply; + else if (v) + type = MessageType::ValueData; + else if (y and y->as() != "q") + throw msgpack::type_error(); + else if (q == "ping") + type = MessageType::Ping; + else if (q == "find") + type = MessageType::FindNode; + else if (q == "get") + type = MessageType::GetValues; + else if (q == "listen") + type = MessageType::Listen; + else if (q == "put") + type = MessageType::AnnounceValue; + else if (q == "refresh") + type = MessageType::Refresh; + else + throw msgpack::type_error(); + + if (type == MessageType::ValueData) { + if (v->type != msgpack::type::MAP) + throw msgpack::type_error(); + for (size_t i = 0; i < v->via.map.size; ++i) { + auto& vdat = v->via.map.ptr[i]; + auto o = findMapValue(vdat.val, "o"); + auto d = findMapValue(vdat.val, "d"); + if (not o or not d) + continue; + value_parts.emplace(vdat.key.as(), std::pair(o->as(), unpackBlob(*d))); + } + return; + } + + auto a = findMapValue(msg, "a"); + if (!a && !r && !e) + throw msgpack::type_error(); + auto& req = a ? *a : (r ? *r : *e); + + if (e) { + if (e->type != msgpack::type::ARRAY) + throw msgpack::type_error(); + error_code = e->via.array.ptr[0].as(); + } + + if (auto t = findMapValue(req, "sid")) + socket_id = {t->as>()}; + + if (auto rid = findMapValue(req, "id")) + id = {*rid}; + + if (auto rh = findMapValue(req, "h")) + info_hash = {*rh}; + + if (auto rtarget = findMapValue(req, "target")) + target = {*rtarget}; + + if (auto rquery = findMapValue(req, "q")) + query.msgpack_unpack(*rquery); + + if (auto otoken = findMapValue(req, "token")) + token = unpackBlob(*otoken); + + if (auto vid = findMapValue(req, "vid")) + value_id = vid->as(); + + if (auto rnodes4 = findMapValue(req, "n4")) + nodes4_raw = unpackBlob(*rnodes4); + + if (auto rnodes6 = findMapValue(req, "n6")) + nodes6_raw = unpackBlob(*rnodes6); + + if (auto sa = findMapValue(req, "sa")) { + if (sa->type != msgpack::type::BIN) + throw msgpack::type_error(); + auto l = sa->via.bin.size; + if (l == sizeof(in_addr)) { + addr.setFamily(AF_INET); + auto& a = addr.getIPv4(); + std::fill_n((uint8_t*)&a, sizeof(a), 0); + a.sin_port = 0; + std::copy_n(sa->via.bin.ptr, l, (char*)&a.sin_addr); + } else if (l == sizeof(in6_addr)) { + addr.setFamily(AF_INET6); + auto& a = addr.getIPv6(); + std::fill_n((uint8_t*)&a, sizeof(a), 0); + a.sin6_port = 0; + std::copy_n(sa->via.bin.ptr, l, (char*)&a.sin6_addr); + } + } else + addr = {}; + + if (auto rcreated = findMapValue(req, "c")) + created = from_time_t(rcreated->as()); + + if (auto rvalues = findMapValue(req, "values")) { + if (rvalues->type != msgpack::type::ARRAY) + throw msgpack::type_error(); + for (size_t i = 0; i < rvalues->via.array.size; i++) { + auto& packed_v = rvalues->via.array.ptr[i]; + if (packed_v.type == msgpack::type::POSITIVE_INTEGER) { + // Skip oversize values with a small margin for header overhead + if (packed_v.via.u64 > MAX_VALUE_SIZE + 32) + continue; + value_parts.emplace(i, std::make_pair(packed_v.via.u64, Blob{})); + } else { + try { + values.emplace_back(std::make_shared(rvalues->via.array.ptr[i])); + } catch (const std::exception& e) { + //DHT_LOG_WARN("Error reading value: %s", e.what()); + } + } + } + } else if (auto raw_fields = findMapValue(req, "fields")) { + if (auto rfields = findMapValue(*raw_fields, "f")) { + auto vfields = rfields->as>(); + if (auto rvalues = findMapValue(*raw_fields, "v")) { + if (rvalues->type != msgpack::type::ARRAY) + throw msgpack::type_error(); + size_t val_num = rvalues->via.array.size / vfields.size(); + for (size_t i = 0; i < val_num; ++i) { + try { + auto v = std::make_shared(); + v->msgpack_unpack_fields(vfields, *rvalues, i*vfields.size()); + fields.emplace_back(std::move(v)); + } catch (const std::exception& e) { } + } + } + } else { + throw msgpack::type_error(); + } + } + + if (auto w = findMapValue(req, "w")) { + if (w->type != msgpack::type::ARRAY) + throw msgpack::type_error(); + want = 0; + for (unsigned i=0; ivia.array.size; i++) { + auto& val = w->via.array.ptr[i]; + try { + auto w = val.as(); + if (w == AF_INET) + want |= WANT4; + else if(w == AF_INET6) + want |= WANT6; + } catch (const std::exception& e) {}; + } + } else { + want = -1; + } +} + +} /* namespace net */ +} /* namespace dht */