/* This file is part of TON Blockchain Library. TON Blockchain Library is free software: you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License as published by the Free Software Foundation, either version 2 of the License, or (at your option) any later version. TON Blockchain Library 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 Lesser General Public License for more details. You should have received a copy of the GNU Lesser General Public License along with TON Blockchain Library. If not, see . Copyright 2017-2019 Telegram Systems LLP */ #include "td/utils/Random.h" #include "adnl/utils.hpp" #include "dht/dht.h" #include "overlay.hpp" #include "auto/tl/ton_api.hpp" #include "keys/encryptor.h" namespace ton { namespace overlay { td::actor::ActorOwn Overlay::create(td::actor::ActorId keyring, td::actor::ActorId adnl, td::actor::ActorId manager, td::actor::ActorId dht_node, adnl::AdnlNodeIdShort local_id, OverlayIdFull overlay_id, std::unique_ptr callback, OverlayPrivacyRules rules) { auto R = td::actor::create_actor("overlay", keyring, adnl, manager, dht_node, local_id, std::move(overlay_id), true, std::vector(), std::move(callback), std::move(rules)); return td::actor::ActorOwn(std::move(R)); } td::actor::ActorOwn Overlay::create(td::actor::ActorId keyring, td::actor::ActorId adnl, td::actor::ActorId manager, td::actor::ActorId dht_node, adnl::AdnlNodeIdShort local_id, OverlayIdFull overlay_id, std::vector nodes, std::unique_ptr callback, OverlayPrivacyRules rules) { auto R = td::actor::create_actor("overlay", keyring, adnl, manager, dht_node, local_id, std::move(overlay_id), false, std::move(nodes), std::move(callback), std::move(rules)); return td::actor::ActorOwn(std::move(R)); } OverlayImpl::OverlayImpl(td::actor::ActorId keyring, td::actor::ActorId adnl, td::actor::ActorId manager, td::actor::ActorId dht_node, adnl::AdnlNodeIdShort local_id, OverlayIdFull overlay_id, bool pub, std::vector nodes, std::unique_ptr callback, OverlayPrivacyRules rules) : keyring_(keyring) , adnl_(adnl) , manager_(manager) , dht_node_(dht_node) , local_id_(local_id) , id_full_(std::move(overlay_id)) , callback_(std::move(callback)) , public_(pub) , rules_(std::move(rules)) { overlay_id_ = id_full_.compute_short_id(); VLOG(OVERLAY_INFO) << this << ": creating " << (public_ ? "public" : "private"); for (auto &node : nodes) { CHECK(!public_); auto X = OverlayNode{node, overlay_id_}; do_add_peer(std::move(X)); } update_neighbours(static_cast(nodes.size())); } void OverlayImpl::process_query(adnl::AdnlNodeIdShort src, ton_api::overlay_getRandomPeers &query, td::Promise promise) { if (public_) { VLOG(OVERLAY_DEBUG) << this << ": received " << query.peers_->nodes_.size() << " nodes from " << src << " in getRandomPeers query"; std::vector nodes; for (auto &n : query.peers_->nodes_) { auto N = OverlayNode::create(n); if (N.is_ok()) { nodes.emplace_back(N.move_as_ok()); } } add_peers(std::move(nodes)); send_random_peers(src, std::move(promise)); } else { VLOG(OVERLAY_WARNING) << this << ": DROPPING getRandomPeers query from " << src << " in private overlay"; promise.set_error(td::Status::Error(ErrorCode::protoviolation, "overlay is private")); } } void OverlayImpl::process_query(adnl::AdnlNodeIdShort src, ton_api::overlay_getBroadcast &query, td::Promise promise) { auto it = broadcasts_.find(query.hash_); if (it == broadcasts_.end()) { VLOG(OVERLAY_NOTICE) << this << ": received getBroadcastQuery(" << query.hash_ << ") from " << src << " but broadcast is unknown"; promise.set_value(create_serialize_tl_object()); return; } if (delivered_broadcasts_.find(query.hash_) != delivered_broadcasts_.end()) { VLOG(OVERLAY_DEBUG) << this << ": received getBroadcastQuery(" << query.hash_ << ") from " << src << " but broadcast already deleted"; promise.set_value(create_serialize_tl_object()); return; } VLOG(OVERLAY_DEBUG) << this << ": received getBroadcastQuery(" << query.hash_ << ") from " << src << " sending broadcast"; promise.set_value(it->second->serialize()); } void OverlayImpl::process_query(adnl::AdnlNodeIdShort src, ton_api::overlay_getBroadcastList &query, td::Promise promise) { VLOG(OVERLAY_WARNING) << this << ": DROPPING getBroadcastList query"; promise.set_error(td::Status::Error(ErrorCode::protoviolation, "dropping get broadcast list query")); } /*void OverlayImpl::process_query(adnl::AdnlNodeIdShort src, adnl::AdnlQueryId query_id, ton_api::overlay_customQuery &query) { callback_->receive_query(src, query_id, id_, std::move(query.data_)); } */ void OverlayImpl::receive_query(adnl::AdnlNodeIdShort src, td::BufferSlice data, td::Promise promise) { if (!public_) { auto P = peers_.get(src); if (P == nullptr) { VLOG(OVERLAY_WARNING) << this << ": received query in private overlay from unknown source " << src; promise.set_error(td::Status::Error(ErrorCode::protoviolation, "overlay is private")); return; } } auto R = fetch_tl_object(data.clone(), true); if (R.is_error()) { // allow custom query to be here callback_->receive_query(src, overlay_id_, std::move(data), std::move(promise)); return; } auto Q = R.move_as_ok(); VLOG(OVERLAY_EXTRA_DEBUG) << this << "query from " << src << ": " << ton_api::to_string(Q); ton_api::downcast_call(*Q.get(), [&](auto &object) { this->process_query(src, object, std::move(promise)); }); } td::Status OverlayImpl::process_broadcast(adnl::AdnlNodeIdShort message_from, tl_object_ptr bcast) { return BroadcastSimple::create(this, std::move(bcast)); } td::Status OverlayImpl::process_broadcast(adnl::AdnlNodeIdShort message_from, tl_object_ptr b) { return OverlayFecBroadcastPart::create(this, std::move(b)); } td::Status OverlayImpl::process_broadcast(adnl::AdnlNodeIdShort message_from, tl_object_ptr b) { return OverlayFecBroadcastPart::create(this, std::move(b)); } td::Status OverlayImpl::process_broadcast(adnl::AdnlNodeIdShort message_from, tl_object_ptr bcast) { return td::Status::Error(ErrorCode::protoviolation, PSTRING() << "received strange message broadcastNotFound from " << message_from); } td::Status OverlayImpl::process_broadcast(adnl::AdnlNodeIdShort message_from, tl_object_ptr msg) { auto it = fec_broadcasts_.find(msg->hash_); if (it != fec_broadcasts_.end()) { VLOG(OVERLAY_DEBUG) << this << ": received fec opt-out message from " << message_from << " for broadcast " << msg->hash_; it->second->add_received(message_from); } else { VLOG(OVERLAY_DEBUG) << this << ": received fec opt-out message from " << message_from << " for unknown broadcast " << msg->hash_; } return td::Status::OK(); } td::Status OverlayImpl::process_broadcast(adnl::AdnlNodeIdShort message_from, tl_object_ptr msg) { auto it = fec_broadcasts_.find(msg->hash_); if (it != fec_broadcasts_.end()) { VLOG(OVERLAY_DEBUG) << this << ": received fec completed message from " << message_from << " for broadcast " << msg->hash_; it->second->add_completed(message_from); } else { VLOG(OVERLAY_DEBUG) << this << ": received fec completed message from " << message_from << " for unknown broadcast " << msg->hash_; } return td::Status::OK(); } td::Status OverlayImpl::process_broadcast(adnl::AdnlNodeIdShort message_from, tl_object_ptr msg) { VLOG(OVERLAY_DEBUG) << this << ": received unicast from " << message_from; callback_->receive_message(message_from, overlay_id_, std::move(msg->data_)); return td::Status::OK(); } void OverlayImpl::receive_message(adnl::AdnlNodeIdShort src, td::BufferSlice data) { if (!public_) { if (peers_.get(src) == nullptr) { VLOG(OVERLAY_WARNING) << this << ": received query in private overlay from unknown source " << src; return; } } auto X = fetch_tl_object(data.clone(), true); if (X.is_error()) { VLOG(OVERLAY_DEBUG) << this << ": received custom message"; callback_->receive_message(src, overlay_id_, std::move(data)); return; } auto Q = X.move_as_ok(); ton_api::downcast_call(*Q.get(), [Self = this, &Q, &src](auto &object) { Self->process_broadcast(src, move_tl_object_as>(Q)); }); } void OverlayImpl::alarm() { bcast_gc(); if (public_) { if (peers_.size() > 0) { auto P = get_random_peer(); if (P) { send_random_peers(P->get_id(), {}); } } if (next_dht_query_.is_in_past()) { auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result res) { td::actor::send_closure(SelfId, &OverlayImpl::receive_dht_nodes, std::move(res), true); }); td::actor::send_closure(dht_node_, &dht::Dht::get_value, dht::DhtKey{overlay_id_.pubkey_hash(), "nodes", 0}, std::move(P)); next_dht_query_ = td::Timestamp::in(td::Random::fast(60.0, 100.0)); } if (update_db_at_.is_in_past()) { if (peers_.size() > 0) { std::vector vec; for (td::uint32 i = 0; i < 20; i++) { vec.push_back(get_random_peer()->get()); } td::actor::send_closure(manager_, &OverlayManager::save_to_db, local_id_, overlay_id_, std::move(vec)); } update_db_at_ = td::Timestamp::in(60.0); } alarm_timestamp() = td::Timestamp::in(1.0); } else { update_neighbours(0); alarm_timestamp() = td::Timestamp::in(60.0 + td::Random::fast(0, 100) * 0.6); } } void OverlayImpl::receive_dht_nodes(td::Result res, bool dummy) { CHECK(public_); if (res.is_ok()) { auto v = res.move_as_ok(); auto R = fetch_tl_object(v.value().clone(), true); if (R.is_ok()) { auto r = R.move_as_ok(); VLOG(OVERLAY_INFO) << this << ": received " << r->nodes_.size() << " nodes from overlay"; VLOG(OVERLAY_EXTRA_DEBUG) << this << ": nodes: " << ton_api::to_string(r); std::vector nodes; for (auto &n : r->nodes_) { auto N = OverlayNode::create(n); if (N.is_ok()) { nodes.emplace_back(N.move_as_ok()); } } add_peers(std::move(nodes)); } else { VLOG(OVERLAY_WARNING) << this << ": incorrect value in DHT for overlay nodes: " << R.move_as_error(); } } else { VLOG(OVERLAY_NOTICE) << this << ": can not get value from DHT: " << res.move_as_error(); } VLOG(OVERLAY_INFO) << this << ": adding self node to DHT overlay's nodes"; auto P = td::PromiseCreator::lambda([SelfId = actor_id(this), oid = print_id()](td::Result R) { if (R.is_error()) { LOG(ERROR) << oid << "cannot get self node"; return; } td::actor::send_closure(SelfId, &OverlayImpl::update_dht_nodes, R.move_as_ok()); }); get_self_node(std::move(P)); } void OverlayImpl::update_dht_nodes(OverlayNode node) { if (!public_) { return; } auto nodes = create_tl_object(std::vector>()); nodes->nodes_.emplace_back(node.tl()); dht::DhtKey dht_key{overlay_id_.pubkey_hash(), "nodes", 0}; auto update_rule = dht::DhtUpdateRuleOverlayNodes::create(); dht::DhtKeyDescription dht_key_descr(std::move(dht_key), id_full_.pubkey(), update_rule.move_as_ok(), td::BufferSlice()); dht_key_descr.check().ensure(); dht::DhtValue value{std::move(dht_key_descr), serialize_tl_object(nodes, true), static_cast(td::Clocks::system() + 3600), td::BufferSlice()}; value.check().ensure(); auto P = td::PromiseCreator::lambda([oid = print_id()](td::Result res) { if (res.is_error()) { VLOG(OVERLAY_NOTICE) << oid << ": error storing to DHT: " << res.move_as_error(); } }); td::actor::send_closure(dht_node_, &dht::Dht::set_value, std::move(value), std::move(P)); } void OverlayImpl::bcast_gc() { while (broadcasts_.size() > max_data_bcasts()) { auto bcast = BroadcastSimple::from_list_node(bcast_data_lru_.get()); CHECK(bcast); auto hash = bcast->get_hash(); broadcasts_.erase(hash); if (delivered_broadcasts_.insert(hash).second) { bcast_lru_.push(hash); } } while (fec_broadcasts_.size() > 0) { auto bcast = BroadcastFec::from_list_node(bcast_fec_lru_.prev); CHECK(bcast); if (bcast->get_date() > td::Clocks::system() - 60) { break; } auto hash = bcast->get_hash(); CHECK(fec_broadcasts_.count(hash) == 1); fec_broadcasts_.erase(hash); if (delivered_broadcasts_.insert(hash).second) { bcast_lru_.push(hash); } } while (bcast_lru_.size() > max_bcasts()) { auto Id = bcast_lru_.front(); bcast_lru_.pop(); CHECK(delivered_broadcasts_.erase(Id)); } CHECK(delivered_broadcasts_.size() == bcast_lru_.size()); } void OverlayImpl::send_message_to_neighbours(td::BufferSlice data) { for (auto &n : neighbours_) { td::actor::send_closure(manager_, &OverlayManager::send_message, n, local_id_, overlay_id_, data.clone()); } } void OverlayImpl::send_broadcast(PublicKeyHash send_as, td::uint32 flags, td::BufferSlice data) { auto S = BroadcastSimple::create_new(actor_id(this), keyring_, send_as, std::move(data), flags); if (S.is_error()) { LOG(WARNING) << "failed to send broadcast: " << S; } } void OverlayImpl::send_broadcast_fec(PublicKeyHash send_as, td::uint32 flags, td::BufferSlice data) { OverlayOutboundFecBroadcast::create(std::move(data), flags, actor_id(this), send_as); } void OverlayImpl::print(td::StringBuilder &sb) { sb << this; } td::Status OverlayImpl::check_date(td::uint32 date) { auto n = td::Clocks::system(); if (date < n - 20) { return td::Status::Error(ErrorCode::notready, "too old broadcast"); } if (date > n + 20) { return td::Status::Error(ErrorCode::notready, "too new broadcast"); } return td::Status::OK(); } td::Status OverlayImpl::check_source_eligible(PublicKey source, const Certificate *cert, td::uint32 size) { if (size == 0) { return td::Status::Error(ErrorCode::protoviolation, "empty broadcast"); } auto short_id = source.compute_short_id(); auto r = rules_.max_size(source.compute_short_id()); if (r >= size) { return td::Status::OK(); } if (!cert) { return td::Status::Error(ErrorCode::protoviolation, "source is not eligible"); } TRY_STATUS(cert->check(short_id, overlay_id_, static_cast(td::Clocks::system()), size)); auto issuer_short = cert->issuer_hash(); if (rules_.max_size(issuer_short) < size) { return td::Status::Error(ErrorCode::protoviolation, "bad certificate"); } return td::Status::OK(); } td::Status OverlayImpl::check_delivered(BroadcastHash hash) { if (delivered_broadcasts_.count(hash) == 1 || broadcasts_.count(hash) == 1) { return td::Status::Error(ErrorCode::notready, "duplicate broadcast"); } else { return td::Status::OK(); } } BroadcastFec *OverlayImpl::get_fec_broadcast(BroadcastHash hash) { auto it = fec_broadcasts_.find(hash); if (it == fec_broadcasts_.end()) { return nullptr; } else { return it->second.get(); } } void OverlayImpl::register_fec_broadcast(std::unique_ptr bcast) { auto hash = bcast->get_hash(); bcast_fec_lru_.put(bcast.get()); fec_broadcasts_.emplace(hash, std::move(bcast)); bcast_gc(); } void OverlayImpl::get_self_node(td::Promise promise) { OverlayNode s{local_id_, overlay_id_}; auto to_sign = s.to_sign(); auto P = td::PromiseCreator::lambda([oid = print_id(), s = std::move(s), promise = std::move(promise)]( td::Result> R) mutable { if (R.is_error()) { auto S = R.move_as_error(); LOG(ERROR) << oid << ": failed to get self node: " << S; promise.set_error(std::move(S)); return; } auto V = R.move_as_ok(); s.update_signature(std::move(V.first)); s.update_adnl_id(adnl::AdnlNodeIdFull{V.second}); promise.set_value(std::move(s)); }); td::actor::send_closure(keyring_, &keyring::Keyring::sign_add_get_public_key, local_id_.pubkey_hash(), std::move(to_sign), std::move(P)); } void OverlayImpl::send_new_fec_broadcast_part(PublicKeyHash local_id, Overlay::BroadcastDataHash data_hash, td::uint32 size, td::uint32 flags, td::BufferSlice part, td::uint32 seqno, fec::FecType fec_type, td::uint32 date) { auto S = OverlayFecBroadcastPart::create_new(this, actor_id(this), local_id, data_hash, size, flags, std::move(part), seqno, std::move(fec_type), date); if (S.is_error() && S.code() != ErrorCode::notready) { LOG(WARNING) << "failed to send broadcast part: " << S; } } void OverlayImpl::deliver_broadcast(PublicKeyHash source, td::BufferSlice data) { callback_->receive_broadcast(source, overlay_id_, std::move(data)); } void OverlayImpl::failed_to_create_fec_broadcast(td::Status reason) { if (reason.code() == ErrorCode::notready) { LOG(DEBUG) << "failed to receive fec broadcast: " << reason; } else { LOG(WARNING) << "failed to receive fec broadcast: " << reason; } } void OverlayImpl::created_fec_broadcast(PublicKeyHash local_id, std::unique_ptr bcast) { bcast->update_overlay(this); auto S = bcast->run(); if (S.is_error() && S.code() != ErrorCode::notready) { LOG(WARNING) << "failed to send fec broadcast: " << S; } } void OverlayImpl::failed_to_create_simple_broadcast(td::Status reason) { if (reason.code() == ErrorCode::notready) { LOG(DEBUG) << "failed to send simple broadcast: " << reason; } else { LOG(WARNING) << "failed to send simple broadcast: " << reason; } } void OverlayImpl::created_simple_broadcast(std::unique_ptr bcast) { bcast->update_overlay(this); auto S = bcast->run(); register_simple_broadcast(std::move(bcast)); if (S.is_error() && S.code() != ErrorCode::notready) { LOG(WARNING) << "failed to receive fec broadcast: " << S; } } void OverlayImpl::register_simple_broadcast(std::unique_ptr bcast) { auto hash = bcast->get_hash(); bcast_data_lru_.put(bcast.get()); broadcasts_.emplace(hash, std::move(bcast)); bcast_gc(); } td::Result OverlayImpl::get_encryptor(PublicKey source) { auto short_id = source.compute_short_id(); auto it = encryptor_map_.find(short_id); if (it != encryptor_map_.end()) { return it->second->get(); } TRY_RESULT(e, source.create_encryptor()); auto res = e.get(); auto cache = std::make_unique(short_id, std::move(e)); encryptor_lru_.put(cache.get()); encryptor_map_.emplace(short_id, std::move(cache)); while (encryptor_map_.size() > max_encryptors()) { auto x = CachedEncryptor::from_list_node(encryptor_lru_.get()); auto id = x->id(); encryptor_map_.erase(id); } return res; } std::shared_ptr OverlayImpl::get_certificate(PublicKeyHash source) { auto it = certs_.find(source); return (it == certs_.end()) ? nullptr : it->second; } void OverlayImpl::set_privacy_rules(OverlayPrivacyRules rules) { rules_ = std::move(rules); } } // namespace overlay } // namespace ton