/* This file is part of cpp-ethereum. cpp-ethereum 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. cpp-ethereum 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 cpp-ethereum. If not, see . */ /** @file Host.cpp * @author Alex Leverington * @author Gav Wood * @date 2014 */ #include #include #include #include #include #include #include #include #include "Session.h" #include "Common.h" #include "Capability.h" #include "UPnP.h" #include "Host.h" using namespace std; using namespace dev; using namespace dev::p2p; Host::Host(std::string const& _clientVersion, NetworkPreferences const& _n, bool _start): Worker("p2p", 0), m_clientVersion(_clientVersion), m_netPrefs(_n), m_ifAddresses(Network::getInterfaceAddresses()), m_ioService(2), m_acceptorV4(m_ioService), m_key(KeyPair::create()) { for (auto address: m_ifAddresses) if (address.is_v4()) clog(NetNote) << "IP Address: " << address << " = " << (isPrivateAddress(address) ? "[LOCAL]" : "[PEER]"); clog(NetNote) << "Id:" << id().abridged(); if (_start) start(); } Host::~Host() { stop(); } void Host::start() { startWorking(); } void Host::stop() { // called to force io_service to kill any remaining tasks it might have - // such tasks may involve socket reads from Capabilities that maintain references // to resources we're about to free. { // Although m_run is set by stop() or start(), it effects m_runTimer so x_runTimer is used instead of a mutex for m_run. // when m_run == false, run() will cause this::run() to stop() ioservice Guard l(x_runTimer); // ignore if already stopped/stopping if (!m_run) return; m_run = false; } // wait for m_timer to reset (indicating network scheduler has stopped) while (!!m_timer) this_thread::sleep_for(chrono::milliseconds(50)); // stop worker thread stopWorking(); } void Host::doneWorking() { // reset ioservice (allows manually polling network, below) m_ioService.reset(); // shutdown acceptor m_acceptorV4.cancel(); if (m_acceptorV4.is_open()) m_acceptorV4.close(); // There maybe an incoming connection which started but hasn't finished. // Wait for acceptor to end itself instead of assuming it's complete. // This helps ensure a peer isn't stopped at the same time it's starting // and that socket for pending connection is closed. while (m_accepting) m_ioService.poll(); // stop capabilities (eth: stops syncing or block/tx broadcast) for (auto const& h: m_capabilities) h.second->onStopping(); // disconnect peers for (unsigned n = 0;; n = 0) { { RecursiveGuard l(x_peers); for (auto i: m_peers) if (auto p = i.second.lock()) if (p->isOpen()) { p->disconnect(ClientQuit); n++; } } if (!n) break; // poll so that peers send out disconnect packets m_ioService.poll(); } // stop network (again; helpful to call before subsequent reset()) m_ioService.stop(); // reset network (allows reusing ioservice in future) m_ioService.reset(); // finally, clear out peers (in case they're lingering) RecursiveGuard l(x_peers); m_peers.clear(); } unsigned Host::protocolVersion() const { return 2; } void Host::registerPeer(std::shared_ptr _s, CapDescs const& _caps) { if (!_s->m_node || !_s->m_node->id) { cwarn << "Attempting to register a peer without node information!"; return; } { RecursiveGuard l(x_peers); m_peers[_s->m_node->id] = _s; } unsigned o = (unsigned)UserPacket; for (auto const& i: _caps) if (haveCapability(i)) { _s->m_capabilities[i] = shared_ptr(m_capabilities[i]->newPeerCapability(_s.get(), o)); o += m_capabilities[i]->messageCount(); } } void Host::seal(bytes& _b) { _b[0] = 0x22; _b[1] = 0x40; _b[2] = 0x08; _b[3] = 0x91; uint32_t len = (uint32_t)_b.size() - 8; _b[4] = (len >> 24) & 0xff; _b[5] = (len >> 16) & 0xff; _b[6] = (len >> 8) & 0xff; _b[7] = len & 0xff; } shared_ptr Host::noteNode(NodeId _id, bi::tcp::endpoint _a, Origin _o, bool _ready, NodeId _oldId) { RecursiveGuard l(x_peers); if (_a.port() < 30300 || _a.port() > 30305) cwarn << "Weird port being recorded: " << _a.port(); if (_a.port() >= /*49152*/32768) { cwarn << "Private port being recorded - setting to 0"; _a = bi::tcp::endpoint(_a.address(), 0); } // cnote << "Node:" << _id.abridged() << _a << (_ready ? "ready" : "used") << _oldId.abridged() << (m_nodes.count(_id) ? "[have]" : "[NEW]"); // First check for another node with the same connection credentials, and put it in oldId if found. if (!_oldId) for (pair> const& n: m_nodes) if (n.second->address == _a && n.second->id != _id) { _oldId = n.second->id; break; } unsigned i; if (!m_nodes.count(_id)) { if (m_nodes.count(_oldId)) { i = m_nodes[_oldId]->index; m_nodes.erase(_oldId); m_nodesList[i] = _id; } else { i = m_nodesList.size(); m_nodesList.push_back(_id); } m_nodes[_id] = make_shared(); m_nodes[_id]->id = _id; m_nodes[_id]->index = i; m_nodes[_id]->idOrigin = _o; } else { i = m_nodes[_id]->index; m_nodes[_id]->idOrigin = max(m_nodes[_id]->idOrigin, _o); } m_nodes[_id]->address = _a; m_ready.extendAll(i); m_private.extendAll(i); if (_ready) m_ready += i; else m_ready -= i; if (!_a.port() || (isPrivateAddress(_a.address()) && !m_netPrefs.localNetworking)) m_private += i; else m_private -= i; // cnote << m_nodes[_id]->index << ":" << m_ready; m_hadNewNodes = true; return m_nodes[_id]; } Nodes Host::potentialPeers(RangeMask const& _known) { RecursiveGuard l(x_peers); Nodes ret; auto ns = (m_netPrefs.localNetworking ? _known : (m_private + _known)).inverted(); for (auto i: ns) ret.push_back(*m_nodes[m_nodesList[i]]); return ret; } void Host::determinePublic(string const& _publicAddress, bool _upnp) { m_peerAddresses.clear(); // no point continuing if there are no interface addresses or valid listen port if (!m_ifAddresses.size() || m_listenPort < 1) return; // populate interfaces we'll listen on (eth listens on all interfaces); ignores local for (auto addr: m_ifAddresses) if ((m_netPrefs.localNetworking || !isPrivateAddress(addr)) && !isLocalHostAddress(addr)) m_peerAddresses.insert(addr); // if user supplied address is a public address then we use it // if user supplied address is private, and localnetworking is enabled, we use it bi::address reqpublicaddr(bi::address(_publicAddress.empty() ? bi::address() : bi::address::from_string(_publicAddress))); bi::tcp::endpoint reqpublic(reqpublicaddr, m_listenPort); bool isprivate = isPrivateAddress(reqpublicaddr); bool ispublic = !isprivate && !isLocalHostAddress(reqpublicaddr); if (!reqpublicaddr.is_unspecified() && (ispublic || (isprivate && m_netPrefs.localNetworking))) { if (!m_peerAddresses.count(reqpublicaddr)) m_peerAddresses.insert(reqpublicaddr); m_public = reqpublic; return; } // if address wasn't provided, then use first public ipv4 address found for (auto addr: m_peerAddresses) if (addr.is_v4() && !isPrivateAddress(addr)) { m_public = bi::tcp::endpoint(*m_peerAddresses.begin(), m_listenPort); return; } // or find address via upnp if (_upnp) { bi::address upnpifaddr; bi::tcp::endpoint upnpep = Network::traverseNAT(m_ifAddresses, m_listenPort, upnpifaddr); if (!upnpep.address().is_unspecified() && !upnpifaddr.is_unspecified()) { if (!m_peerAddresses.count(upnpep.address())) m_peerAddresses.insert(upnpep.address()); m_public = upnpep; return; } } // or if no address provided, use private ipv4 address if local networking is enabled if (reqpublicaddr.is_unspecified()) if (m_netPrefs.localNetworking) for (auto addr: m_peerAddresses) if (addr.is_v4() && isPrivateAddress(addr)) { m_public = bi::tcp::endpoint(addr, m_listenPort); return; } // otherwise address is unspecified m_public = bi::tcp::endpoint(bi::address(), m_listenPort); } void Host::runAcceptor() { assert(m_listenPort > 0); if (m_run && !m_accepting) { clog(NetConnect) << "Listening on local port " << m_listenPort << " (public: " << m_public << ")"; m_accepting = true; m_socket.reset(new bi::tcp::socket(m_ioService)); m_acceptorV4.async_accept(*m_socket, [=](boost::system::error_code ec) { bool success = false; if (!ec) { try { try { clog(NetConnect) << "Accepted connection from " << m_socket->remote_endpoint(); } catch (...){} bi::address remoteAddress = m_socket->remote_endpoint().address(); // Port defaults to 0 - we let the hello tell us which port the peer listens to auto p = std::make_shared(this, std::move(*m_socket.release()), bi::tcp::endpoint(remoteAddress, 0)); p->start(); success = true; } catch (Exception const& _e) { clog(NetWarn) << "ERROR: " << diagnostic_information(_e); } catch (std::exception const& _e) { clog(NetWarn) << "ERROR: " << _e.what(); } } if (!success && m_socket->is_open()) { boost::system::error_code ec; m_socket->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec); m_socket->close(); } m_accepting = false; if (ec.value() < 1) runAcceptor(); }); } } string Host::pocHost() { vector strs; boost::split(strs, dev::Version, boost::is_any_of(".")); return "poc-" + strs[1] + ".ethdev.com"; } void Host::connect(std::string const& _addr, unsigned short _port) noexcept { if (!m_run) return; for (auto first: {true, false}) { try { if (first) { bi::tcp::resolver r(m_ioService); connect(r.resolve({_addr, toString(_port)})->endpoint()); } else connect(bi::tcp::endpoint(bi::address::from_string(_addr), _port)); break; } catch (Exception const& _e) { // Couldn't connect clog(NetConnect) << "Bad host " << _addr << "\n" << diagnostic_information(_e); } catch (exception const& e) { // Couldn't connect clog(NetConnect) << "Bad host " << _addr << " (" << e.what() << ")"; } } } void Host::connect(bi::tcp::endpoint const& _ep) { if (!m_run) return; clog(NetConnect) << "Attempting single-shot connection to " << _ep; bi::tcp::socket* s = new bi::tcp::socket(m_ioService); s->async_connect(_ep, [=](boost::system::error_code const& ec) { if (ec) clog(NetConnect) << "Connection refused to " << _ep << " (" << ec.message() << ")"; else { auto p = make_shared(this, std::move(*s), _ep); clog(NetConnect) << "Connected to " << _ep; p->start(); } delete s; }); } void Host::connect(std::shared_ptr const& _n) { if (!m_run) return; // prevent concurrently connecting to a node; todo: better abstraction Node *nptr = _n.get(); { Guard l(x_pendingNodeConns); if (m_pendingNodeConns.count(nptr)) return; m_pendingNodeConns.insert(nptr); } clog(NetConnect) << "Attempting connection to node" << _n->id.abridged() << "@" << _n->address << "from" << id().abridged(); _n->lastAttempted = std::chrono::system_clock::now(); _n->failedAttempts++; m_ready -= _n->index; bi::tcp::socket* s = new bi::tcp::socket(m_ioService); s->async_connect(_n->address, [=](boost::system::error_code const& ec) { if (ec) { clog(NetConnect) << "Connection refused to node" << _n->id.abridged() << "@" << _n->address << "(" << ec.message() << ")"; _n->lastDisconnect = TCPError; _n->lastAttempted = std::chrono::system_clock::now(); m_ready += _n->index; } else { clog(NetConnect) << "Connected to" << _n->id.abridged() << "@" << _n->address; _n->lastConnected = std::chrono::system_clock::now(); auto p = make_shared(this, std::move(*s), node(_n->id), true); // true because we don't care about ids matched for now. Once we have permenant IDs this will matter a lot more and we can institute a safer mechanism. p->start(); } delete s; Guard l(x_pendingNodeConns); m_pendingNodeConns.erase(nptr); }); } bool Host::havePeer(NodeId _id) const { RecursiveGuard l(x_peers); // Remove dead peers from list. for (auto i = m_peers.begin(); i != m_peers.end();) if (i->second.lock().get()) ++i; else i = m_peers.erase(i); return !!m_peers.count(_id); } unsigned Node::fallbackSeconds() const { switch (lastDisconnect) { case BadProtocol: return 30 * (failedAttempts + 1); case UselessPeer: case TooManyPeers: case ClientQuit: return 15 * (failedAttempts + 1); case NoDisconnect: return 0; default: if (failedAttempts < 5) return failedAttempts * 5; else if (failedAttempts < 15) return 25 + (failedAttempts - 5) * 10; else return 25 + 100 + (failedAttempts - 15) * 20; } } bool Node::shouldReconnect() const { return chrono::system_clock::now() > lastAttempted + chrono::seconds(fallbackSeconds()); } void Host::growPeers() { RecursiveGuard l(x_peers); int morePeers = (int)m_idealPeerCount - m_peers.size(); if (morePeers > 0) { auto toTry = m_ready; if (!m_netPrefs.localNetworking) toTry -= m_private; set ns; for (auto i: toTry) if (m_nodes[m_nodesList[i]]->shouldReconnect()) ns.insert(*m_nodes[m_nodesList[i]]); if (ns.size()) for (Node const& i: ns) { connect(m_nodes[i.id]); if (!--morePeers) return; } else for (auto const& i: m_peers) if (auto p = i.second.lock()) p->ensureNodesRequested(); } } void Host::prunePeers() { RecursiveGuard l(x_peers); // We'll keep at most twice as many as is ideal, halfing what counts as "too young to kill" until we get there. set dc; for (unsigned old = 15000; m_peers.size() - dc.size() > m_idealPeerCount * 2 && old > 100; old /= 2) if (m_peers.size() - dc.size() > m_idealPeerCount) { // look for worst peer to kick off // first work out how many are old enough to kick off. shared_ptr worst; unsigned agedPeers = 0; for (auto i: m_peers) if (!dc.count(i.first)) if (auto p = i.second.lock()) if (/*(m_mode != NodeMode::Host || p->m_caps != 0x01) &&*/ chrono::steady_clock::now() > p->m_connect + chrono::milliseconds(old)) // don't throw off new peers; peer-servers should never kick off other peer-servers. { ++agedPeers; if ((!worst || p->rating() < worst->rating() || (p->rating() == worst->rating() && p->m_connect > worst->m_connect))) // kill older ones worst = p; } if (!worst || agedPeers <= m_idealPeerCount) break; dc.insert(worst->id()); worst->disconnect(TooManyPeers); } // Remove dead peers from list. for (auto i = m_peers.begin(); i != m_peers.end();) if (i->second.lock().get()) ++i; else i = m_peers.erase(i); } PeerInfos Host::peers(bool _updatePing) const { if (!m_run) return PeerInfos(); RecursiveGuard l(x_peers); if (_updatePing) { const_cast(this)->pingAll(); this_thread::sleep_for(chrono::milliseconds(200)); } std::vector ret; for (auto& i: m_peers) if (auto j = i.second.lock()) if (j->m_socket.is_open()) ret.push_back(j->m_info); return ret; } void Host::run(boost::system::error_code const&) { if (!m_run) { // stopping io service allows running manual network operations for shutdown // and also stops blocking worker thread, allowing worker thread to exit m_ioService.stop(); // resetting timer signals network that nothing else can be scheduled to run m_timer.reset(); return; } m_lastTick += c_timerInterval; if (m_lastTick >= c_timerInterval * 10) { growPeers(); prunePeers(); m_lastTick = 0; } if (m_hadNewNodes) { for (auto p: m_peers) if (auto pp = p.second.lock()) pp->serviceNodesRequest(); m_hadNewNodes = false; } if (chrono::steady_clock::now() - m_lastPing > chrono::seconds(30)) // ping every 30s. { for (auto p: m_peers) if (auto pp = p.second.lock()) if (chrono::steady_clock::now() - pp->m_lastReceived > chrono::seconds(60)) pp->disconnect(PingTimeout); pingAll(); } auto runcb = [this](boost::system::error_code const& error) -> void { run(error); }; m_timer->expires_from_now(boost::posix_time::milliseconds(c_timerInterval)); m_timer->async_wait(runcb); } void Host::startedWorking() { asserts(!m_timer); { // prevent m_run from being set to true at same time as set to false by stop() // don't release mutex until m_timer is set so in case stop() is called at same // time, stop will wait on m_timer and graceful network shutdown. Guard l(x_runTimer); // create deadline timer m_timer.reset(new boost::asio::deadline_timer(m_ioService)); m_run = true; } // try to open acceptor (todo: ipv6) m_listenPort = Network::listen4(m_acceptorV4, m_netPrefs.listenPort); // start capability threads for (auto const& h: m_capabilities) h.second->onStarting(); // determine public IP, but only if we're able to listen for connections // todo: GUI when listen is unavailable in UI if (m_listenPort) { determinePublic(m_netPrefs.publicIP, m_netPrefs.upnp); if (m_listenPort > 0) runAcceptor(); } // if m_public address is valid then add us to node list // todo: abstract empty() and emplace logic if (!m_public.address().is_unspecified() && (m_nodes.empty() || m_nodes[m_nodesList[0]]->id != id())) noteNode(id(), m_public, Origin::Perfect, false); clog(NetNote) << "Id:" << id().abridged(); run(boost::system::error_code()); } void Host::doWork() { if (m_run) m_ioService.run(); } void Host::pingAll() { RecursiveGuard l(x_peers); for (auto& i: m_peers) if (auto j = i.second.lock()) j->ping(); m_lastPing = chrono::steady_clock::now(); } bytes Host::saveNodes() const { RLPStream nodes; int count = 0; { RecursiveGuard l(x_peers); for (auto const& i: m_nodes) { Node const& n = *(i.second); // TODO: PoC-7: Figure out why it ever shares these ports.//n.address.port() >= 30300 && n.address.port() <= 30305 && if (!n.dead && chrono::system_clock::now() - n.lastConnected < chrono::seconds(3600 * 48) && n.address.port() > 0 && n.address.port() < /*49152*/32768 && n.id != id() && !isPrivateAddress(n.address.address())) { nodes.appendList(10); if (n.address.address().is_v4()) nodes << n.address.address().to_v4().to_bytes(); else nodes << n.address.address().to_v6().to_bytes(); nodes << n.address.port() << n.id << (int)n.idOrigin << chrono::duration_cast(n.lastConnected.time_since_epoch()).count() << chrono::duration_cast(n.lastAttempted.time_since_epoch()).count() << n.failedAttempts << (unsigned)n.lastDisconnect << n.score << n.rating; count++; } } } RLPStream ret(3); ret << 0 << m_key.secret(); ret.appendList(count).appendRaw(nodes.out(), count); return ret.out(); } void Host::restoreNodes(bytesConstRef _b) { RecursiveGuard l(x_peers); RLP r(_b); if (r.itemCount() > 0 && r[0].isInt()) switch (r[0].toInt()) { case 0: { auto oldId = id(); m_key = KeyPair(r[1].toHash()); noteNode(id(), m_public, Origin::Perfect, false, oldId); for (auto i: r[2]) { bi::tcp::endpoint ep; if (i[0].itemCount() == 4) ep = bi::tcp::endpoint(bi::address_v4(i[0].toArray()), i[1].toInt()); else ep = bi::tcp::endpoint(bi::address_v6(i[0].toArray()), i[1].toInt()); auto id = (NodeId)i[2]; if (!m_nodes.count(id)) { auto o = (Origin)i[3].toInt(); auto n = noteNode(id, ep, o, true); n->lastConnected = chrono::system_clock::time_point(chrono::seconds(i[4].toInt())); n->lastAttempted = chrono::system_clock::time_point(chrono::seconds(i[5].toInt())); n->failedAttempts = i[6].toInt(); n->lastDisconnect = (DisconnectReason)i[7].toInt(); n->score = (int)i[8].toInt(); n->rating = (int)i[9].toInt(); } } } default:; } else for (auto i: r) { auto id = (NodeId)i[2]; if (!m_nodes.count(id)) { bi::tcp::endpoint ep; if (i[0].itemCount() == 4) ep = bi::tcp::endpoint(bi::address_v4(i[0].toArray()), i[1].toInt()); else ep = bi::tcp::endpoint(bi::address_v6(i[0].toArray()), i[1].toInt()); auto n = noteNode(id, ep, Origin::Self, true); } } }