dht_tracker.cpp

LINUX下

			catch (std::exception&)
			{
				TORRENT_LOG(dht_tracker) << "   client: generic";
			};
#endif

			std::string const& msg_type = e["y"].string();

			if (msg_type == "r")
			{
#ifdef TORRENT_DHT_VERBOSE_LOGGING
				TORRENT_LOG(dht_tracker) << "   reply: transaction: "
					<< m.transaction_id;
#endif

				m.reply = true;
				entry const& r = e["r"];
				std::string const& id = r["id"].string();
				if (id.size() != 20) throw std::runtime_error("invalid size of id");
				std::copy(id.begin(), id.end(), m.id.begin());

				if (entry const* n = r.find_key("values"))
				{
					m.peers.clear();
					if (n->list().size() == 1)
					{
						// assume it's mainline format
						std::string const& peers = n->list().front().string();
						std::string::const_iterator i = peers.begin();
						std::string::const_iterator end = peers.end();

						while (std::distance(i, end) >= 6)
							m.peers.push_back(read_v4_endpoint<tcp::endpoint>(i));
					}
					else
					{
						// assume it's uTorrent/libtorrent format
						read_endpoint_list<tcp::endpoint>(n, m.peers);
					}
#ifdef TORRENT_DHT_VERBOSE_LOGGING
					TORRENT_LOG(dht_tracker) << "   peers: " << m.peers.size();
#endif
				}

				m.nodes.clear();
				if (entry const* n = r.find_key("nodes"))				
				{
					std::string const& nodes = n->string();
					std::string::const_iterator i = nodes.begin();
					std::string::const_iterator end = nodes.end();

					while (std::distance(i, end) >= 26)
					{
						node_id id;
						std::copy(i, i + 20, id.begin());
						i += 20;
						m.nodes.push_back(libtorrent::dht::node_entry(
							id, read_v4_endpoint<udp::endpoint>(i)));
					}
#ifdef TORRENT_DHT_VERBOSE_LOGGING
					TORRENT_LOG(dht_tracker) << "   nodes: " << m.nodes.size();
#endif
				}

				if (entry const* n = r.find_key("nodes2"))				
				{
					entry::list_type const& contacts = n->list();
					for (entry::list_type::const_iterator i = contacts.begin()
						, end(contacts.end()); i != end; ++i)
					{
						std::string const& p = i->string();
						if (p.size() < 6 + 20) continue;
						std::string::const_iterator in = p.begin();

						node_id id;
						std::copy(in, in + 20, id.begin());
						in += 20;
						if (p.size() == 6 + 20)
							m.nodes.push_back(libtorrent::dht::node_entry(
								id, read_v4_endpoint<udp::endpoint>(in)));
						else if (p.size() == 18 + 20)
							m.nodes.push_back(libtorrent::dht::node_entry(
								id, read_v6_endpoint<udp::endpoint>(in)));
					}
#ifdef TORRENT_DHT_VERBOSE_LOGGING
					TORRENT_LOG(dht_tracker) << "   nodes2 + nodes: " << m.nodes.size();
#endif
				}

				entry const* token = r.find_key("token");
				if (token) m.write_token = *token;
			}
			else if (msg_type == "q")
			{
				m.reply = false;
				entry const& a = e["a"];
				std::string const& id = a["id"].string();
				if (id.size() != 20) throw std::runtime_error("invalid size of id");
				std::copy(id.begin(), id.end(), m.id.begin());

				std::string request_kind(e["q"].string());
#ifdef TORRENT_DHT_VERBOSE_LOGGING
				TORRENT_LOG(dht_tracker) << "   query: " << request_kind;
#endif

				if (request_kind == "ping")
				{
					m.message_id = libtorrent::dht::messages::ping;
				}
				else if (request_kind == "find_node")
				{
					std::string const& target = a["target"].string();
					if (target.size() != 20) throw std::runtime_error("invalid size of target id");
					std::copy(target.begin(), target.end(), m.info_hash.begin());
#ifdef TORRENT_DHT_VERBOSE_LOGGING
					TORRENT_LOG(dht_tracker) << "   target: "
						<< boost::lexical_cast<std::string>(m.info_hash);
#endif

					m.message_id = libtorrent::dht::messages::find_node;
				}
				else if (request_kind == "get_peers")
				{
					std::string const& info_hash = a["info_hash"].string();
					if (info_hash.size() != 20) throw std::runtime_error("invalid size of info-hash");
					std::copy(info_hash.begin(), info_hash.end(), m.info_hash.begin());
					m.message_id = libtorrent::dht::messages::get_peers;
#ifdef TORRENT_DHT_VERBOSE_LOGGING
					TORRENT_LOG(dht_tracker) << "   info_hash: "
						<< boost::lexical_cast<std::string>(m.info_hash);
#endif
				}
				else if (request_kind == "announce_peer")
				{
#ifdef TORRENT_DHT_VERBOSE_LOGGING
					++m_announces;
#endif
					std::string const& info_hash = a["info_hash"].string();
					if (info_hash.size() != 20)
						throw std::runtime_error("invalid size of info-hash");
					std::copy(info_hash.begin(), info_hash.end(), m.info_hash.begin());
					m.port = a["port"].integer();
					m.write_token = a["token"];
					m.message_id = libtorrent::dht::messages::announce_peer;
#ifdef TORRENT_DHT_VERBOSE_LOGGING
					TORRENT_LOG(dht_tracker) << "   info_hash: "
						<< boost::lexical_cast<std::string>(m.info_hash);
					TORRENT_LOG(dht_tracker) << "   port: " << m.port;

					if (!m_dht.verify_token(m))
						++m_failed_announces;
#endif
				}
				else
				{
#ifdef TORRENT_DHT_VERBOSE_LOGGING
					TORRENT_LOG(dht_tracker) << "  *** UNSUPPORTED REQUEST *** : "
						<< request_kind;
#endif
					throw std::runtime_error("unsupported request: " + request_kind);
				}
			}
			else if (msg_type == "e")
			{
				entry::list_type const& list = e["e"].list();
				m.message_id = messages::error;
				m.error_msg = list.back().string();
				m.error_code = list.front().integer();
#ifdef TORRENT_DHT_VERBOSE_LOGGING
				TORRENT_LOG(dht_tracker) << "   incoming error: " << m.error_code << " "
					<< m.error_msg;
#endif
				throw std::runtime_error("DHT error message");
			}
			else
			{
#ifdef TORRENT_DHT_VERBOSE_LOGGING
				TORRENT_LOG(dht_tracker) << "  *** UNSUPPORTED MESSAGE TYPE *** : "
					<< msg_type;
#endif
				throw std::runtime_error("unsupported message type: " + msg_type);
			}

#ifdef TORRENT_DHT_VERBOSE_LOGGING
			if (!m.reply)
			{
				++m_queries_received[m.message_id];
				m_queries_bytes_received[m.message_id] += int(bytes_transferred);
			}
			TORRENT_LOG(dht_tracker) << e;
#endif
			TORRENT_ASSERT(m.message_id != messages::error);
			m_dht.incoming(m);
		}
		catch (std::exception& e)
		{
#ifdef TORRENT_DHT_VERBOSE_LOGGING
			int current_buffer = (m_buffer + 1) & 1;
			std::string msg(m_in_buf[current_buffer].begin()
				, m_in_buf[current_buffer].begin() + bytes_transferred);
			TORRENT_LOG(dht_tracker) << "invalid incoming packet: "
				<< e.what() << "\n" << msg << "\n";
#endif
		}
	}
	catch (std::exception& e)
	{
		TORRENT_ASSERT(false);
	};

	entry dht_tracker::state() const
	{
		entry ret(entry::dictionary_t);
		{
			entry nodes(entry::list_t);
			for (node_impl::iterator i(m_dht.begin())
				, end(m_dht.end()); i != end; ++i)
			{
				std::string node;
				std::back_insert_iterator<std::string> out(node);
				write_endpoint(i->addr, out);
				nodes.list().push_back(entry(node));
			}
			bucket_t cache;
			m_dht.replacement_cache(cache);
			for (bucket_t::iterator i(cache.begin())
				, end(cache.end()); i != end; ++i)
			{
				std::string node;
				std::back_insert_iterator<std::string> out(node);
				write_endpoint(i->addr, out);
				nodes.list().push_back(entry(node));
			}
			if (!nodes.list().empty())
				ret["nodes"] = nodes;
		}
		
		ret["node-id"] = boost::lexical_cast<std::string>(m_dht.nid());
		return ret;
	}

	void dht_tracker::add_node(udp::endpoint node)
	{
		m_dht.add_node(node);
	}

	void dht_tracker::add_node(std::pair<std::string, int> const& node)
	{
		udp::resolver::query q(node.first, lexical_cast<std::string>(node.second));
		m_host_resolver.async_resolve(q, m_strand.wrap(
			bind(&dht_tracker::on_name_lookup, self(), _1, _2)));
	}

	void dht_tracker::on_name_lookup(asio::error_code const& e
		, udp::resolver::iterator host) try
	{
		if (e || host == udp::resolver::iterator()) return;
		if (!m_socket.is_open()) return;
		add_node(host->endpoint());
	}
	catch (std::exception&)
	{
		TORRENT_ASSERT(false);
	};

	void dht_tracker::add_router_node(std::pair<std::string, int> const& node)
	{
		udp::resolver::query q(node.first, lexical_cast<std::string>(node.second));
		m_host_resolver.async_resolve(q, m_strand.wrap(
			bind(&dht_tracker::on_router_name_lookup, self(), _1, _2)));
	}

	void dht_tracker::on_router_name_lookup(asio::error_code const& e
		, udp::resolver::iterator host) try
	{
		if (e || host == udp::resolver::iterator()) return;
		if (!m_socket.is_open()) return;
		m_dht.add_router_node(host->endpoint());
	}
	catch (std::exception&)
	{
		TORRENT_ASSERT(false);
	};

	void dht_tracker::on_bootstrap()
	{}

	namespace
	{
		void write_nodes_entry(entry& r, libtorrent::dht::msg const& m)
		{
			bool ipv6_nodes = false;
			r["nodes"] = entry(entry::string_t);
			entry& n = r["nodes"];
			std::back_insert_iterator<std::string> out(n.string());
			for (msg::nodes_t::const_iterator i = m.nodes.begin()
				, end(m.nodes.end()); i != end; ++i)
			{
				if (!i->addr.address().is_v4())
				{
					ipv6_nodes = true;
					continue;
				}
				std::copy(i->id.begin(), i->id.end(), out);
				write_endpoint(i->addr, out);
			}

			if (ipv6_nodes)
			{
				r["nodes2"] = entry(entry::list_t);
				entry& p = r["nodes2"];
				std::string endpoint;
				for (msg::nodes_t::const_iterator i = m.nodes.begin()
					, end(m.nodes.end()); i != end; ++i)
				{
					if (!i->addr.address().is_v6()) continue;
					endpoint.resize(18 + 20);
					std::string::iterator out = endpoint.begin();
					std::copy(i->id.begin(), i->id.end(), out);
					out += 20;
					write_endpoint(i->addr, out);
					endpoint.resize(out - endpoint.begin());
					p.list().push_back(entry(endpoint));
				}
			}
#ifdef TORRENT_DHT_VERBOSE_LOGGING
			TORRENT_LOG(dht_tracker) << "   nodes: " << m.nodes.size();
#endif
		}
	}

	void dht_tracker::send_packet(msg const& m)
		try
	{
		using libtorrent::bencode;
		using libtorrent::entry;
		entry e(entry::dictionary_t);
		TORRENT_ASSERT(!m.transaction_id.empty() || m.message_id == messages::error);
		e["t"] = m.transaction_id;
		static char const version_str[] = {'L', 'T'
			, LIBTORRENT_VERSION_MAJOR, LIBTORRENT_VERSION_MINOR};
		e["v"] = std::string(version_str, version_str + 4);

#ifdef TORRENT_DHT_VERBOSE_LOGGING
		TORRENT_LOG(dht_tracker) << time_now_string()
			<< " SENDING [" << m.addr << "]:";
		TORRENT_LOG(dht_tracker) << "   transaction: " << m.transaction_id;
#endif

		if (m.message_id == messages::error)
		{
			TORRENT_ASSERT(m.reply);
			e["y"] = "e";
			entry error_list(entry::list_t);
			TORRENT_ASSERT(m.error_code > 200 && m.error_code <= 204);
			error_list.list().push_back(entry(m.error_code));
			error_list.list().push_back(entry(m.error_msg));
			e["e"] = error_list;
#ifdef TORRENT_DHT_VERBOSE_LOGGING
			TORRENT_LOG(dht_tracker) << time_now_string()
				<< "   outgoing error: " << m.error_code << " " << m.error_msg;
#endif
		}
		else if (m.reply)
		{
			e["y"] = "r";
			e["r"] = entry(entry::dictionary_t);
			entry& r = e["r"];
			r["id"] = std::string(m.id.begin(), m.id.end());

#ifdef TORRENT_DHT_VERBOSE_LOGGING
			TORRENT_LOG(dht_tracker) << time_now_string()
				<< "   reply: " << messages::ids[m.message_id];
#endif

			if (m.write_token.type() != entry::undefined_t)
				r["token"] = m.write_token;

			switch (m.message_id)
			{
				case messages::ping:
					break;
				case messages::find_node:
				{
					write_nodes_entry(r, m);
					break;
				}
				case messages::get_peers:
				{
					if (m.peers.empty())
					{
						write_nodes_entry(r, m);
					}
					else
					{
						r["values"] = entry(entry::list_t);
						entry& p = r["values"];
						std::string endpoint;
						for (msg::peers_t::const_iterator i = m.peers.begin()
							, end(m.peers.end()); i != end; ++i)
						{
							endpoint.resize(18);
							std::string::iterator out = endpoint.begin();
							write_endpoint(*i, out);
							endpoint.resize(out - endpoint.begin());
							p.list().push_back(entry(endpoint));
						}
#ifdef TORRENT_DHT_VERBOSE_LOGGING
						TORRENT_LOG(dht_tracker) << "   peers: " << m.peers.size();
#endif
					}
					break;
				}

				case messages::announce_peer:
					break;
				break;
			}
		}
		else
		{
			e["y"] = "q";
			e["a"] = entry(entry::dictionary_t);
			entry& a = e["a"];
			a["id"] = std::string(m.id.begin(), m.id.end());

			if (m.write_token.type() != entry::undefined_t)
				a["token"] = m.write_token;
			TORRENT_ASSERT(m.message_id <= messages::error);
			e["q"] = messages::ids[m.message_id];

#ifdef TORRENT_DHT_VERBOSE_LOGGING
			TORRENT_LOG(dht_tracker) << "   query: "
				<< messages::ids[m.message_id];
#endif

			switch (m.message_id)
			{
				case messages::find_node:
				{
					a["target"] = std::string(m.info_hash.begin(), m.info_hash.end());
#ifdef TORRENT_DHT_VERBOSE_LOGGING
					TORRENT_LOG(dht_tracker) << "   target: "
						<< boost::lexical_cast<std::string>(m.info_hash);
#endif
					break;
				}
				case messages::get_peers:
				{
					a["info_hash"] = std::string(m.info_hash.begin(), m.info_hash.end());
#ifdef TORRENT_DHT_VERBOSE_LOGGING
					TORRENT_LOG(dht_tracker) << "   info_hash: "
						<< boost::lexical_cast<std::string>(m.info_hash);
#endif
					break;	
				}
				case messages::announce_peer:
					a["port"] = m_settings.service_port;
					a["info_hash"] = std::string(m.info_hash.begin(), m.info_hash.end());
					a["token"] = m.write_token;
#ifdef TORRENT_DHT_VERBOSE_LOGGING
					TORRENT_LOG(dht_tracker) << "   port: "
						<< m_settings.service_port
						<< "   info_hash: " << boost::lexical_cast<std::string>(m.info_hash);
#endif
					break;
				default: break;
			}

		}

		m_send_buf.clear();
		bencode(std::back_inserter(m_send_buf), e);
		asio::error_code ec;
		m_socket.send_to(asio::buffer(&m_send_buf[0]
			, (int)m_send_buf.size()), m.addr, 0, ec);
		if (ec) return;

#ifdef TORRENT_DHT_VERBOSE_LOGGING
		m_total_out_bytes += m_send_buf.size();
		
		if (m.reply)
		{
			++m_replies_sent[m.message_id];
			m_replies_bytes_sent[m.message_id] += int(m_send_buf.size());
		}
		else
		{
			m_queries_out_bytes += m_send_buf.size();
		}
		TORRENT_LOG(dht_tracker) << e;
#endif

		if (!m.piggy_backed_ping) return;
		
		msg pm;
		pm.reply = false;
		pm.piggy_backed_ping = false;
		pm.message_id = messages::ping;
		pm.transaction_id = m.ping_transaction_id;
		pm.id = m.id;
		pm.addr = m.addr;
	
		send_packet(pm);	
	}
	catch (std::exception&)
	{
		// m_send may fail with "no route to host"
		// but it shouldn't throw since an error code
		// is passed in instead
		TORRENT_ASSERT(false);
	}

}}