SSU.cpp

#include <string.h>
#include <boost/bind.hpp>
#include "Log.h"
#include "Timestamp.h"
#include "RouterContext.h"
#include "NetDb.h"
#include "SSU.h"

namespace i2p
{
namespace transport
{

	SSUServer::SSUServer (const boost::asio::ip::address & addr, int port):
		m_OnlyV6(true), m_IsRunning(false),
		m_Thread (nullptr), m_ThreadV6 (nullptr), m_ReceiversThread (nullptr),
		m_Work (m_Service), m_WorkV6 (m_ServiceV6), m_ReceiversWork (m_ReceiversService), 
		m_EndpointV6 (addr, port), 
		m_Socket (m_ReceiversService, m_Endpoint), m_SocketV6 (m_ReceiversService), 
		m_IntroducersUpdateTimer (m_Service), m_PeerTestsCleanupTimer (m_Service)	
	{
		m_SocketV6.open (boost::asio::ip::udp::v6());
		m_SocketV6.set_option (boost::asio::ip::v6_only (true));
		m_SocketV6.set_option (boost::asio::socket_base::receive_buffer_size (65535));
		m_SocketV6.set_option (boost::asio::socket_base::send_buffer_size (65535));
		m_SocketV6.bind (m_EndpointV6);
	}
	
	SSUServer::SSUServer (int port):
		m_OnlyV6(false), m_IsRunning(false),
		m_Thread (nullptr), m_ThreadV6 (nullptr), m_ReceiversThread (nullptr),
		m_Work (m_Service), m_WorkV6 (m_ServiceV6), m_ReceiversWork (m_ReceiversService), 
		m_Endpoint (boost::asio::ip::udp::v4 (), port), m_EndpointV6 (boost::asio::ip::udp::v6 (), port), 
		m_Socket (m_ReceiversService, m_Endpoint), m_SocketV6 (m_ReceiversService), 
		m_IntroducersUpdateTimer (m_Service), m_PeerTestsCleanupTimer (m_Service)	
	{
		
		m_Socket.set_option (boost::asio::socket_base::receive_buffer_size (65535));
		m_Socket.set_option (boost::asio::socket_base::send_buffer_size (65535));
		if (context.SupportsV6 ())
		{
			m_SocketV6.open (boost::asio::ip::udp::v6());
			m_SocketV6.set_option (boost::asio::ip::v6_only (true));
			m_SocketV6.set_option (boost::asio::socket_base::receive_buffer_size (65535));
			m_SocketV6.set_option (boost::asio::socket_base::send_buffer_size (65535));
			m_SocketV6.bind (m_EndpointV6);
		}
	}
	
	SSUServer::~SSUServer ()
	{
	}

	void SSUServer::Start ()
	{
		m_IsRunning = true;
		m_ReceiversThread = new std::thread (std::bind (&SSUServer::RunReceivers, this));
		if (!m_OnlyV6)
		{
			m_Thread = new std::thread (std::bind (&SSUServer::Run, this));
			m_ReceiversService.post (std::bind (&SSUServer::Receive, this));
		}
		if (context.SupportsV6 ())
		{	
			m_ThreadV6 = new std::thread (std::bind (&SSUServer::RunV6, this));
			m_ReceiversService.post (std::bind (&SSUServer::ReceiveV6, this));	
		}
		SchedulePeerTestsCleanupTimer ();	
		ScheduleIntroducersUpdateTimer (); // wait for 30 seconds and decide if we need introducers
	}

	void SSUServer::Stop ()
	{
		DeleteAllSessions ();
		m_IsRunning = false;
		m_Service.stop ();
		m_Socket.close ();
		m_ServiceV6.stop ();
		m_SocketV6.close ();
		m_ReceiversService.stop ();
		if (m_ReceiversThread)
		{	
			m_ReceiversThread->join (); 
			delete m_ReceiversThread;
			m_ReceiversThread = nullptr;
		}
		if (m_Thread)
		{	
			m_Thread->join (); 
			delete m_Thread;
			m_Thread = nullptr;
		}
		if (m_ThreadV6)
		{	
			m_ThreadV6->join (); 
			delete m_ThreadV6;
			m_ThreadV6 = nullptr;
		}
	}

	void SSUServer::Run () 
	{ 
		while (m_IsRunning)
		{
			try
			{	
				m_Service.run ();
			}
			catch (std::exception& ex)
			{
				LogPrint (eLogError, "SSU: server runtime exception: ", ex.what ());
			}	
		}	
	}

	void SSUServer::RunV6 () 
	{ 
		while (m_IsRunning)
		{
			try
			{	
				m_ServiceV6.run ();
			}
			catch (std::exception& ex)
			{
				LogPrint (eLogError, "SSU: v6 server runtime exception: ", ex.what ());
			}	
		}	
	}	

	void SSUServer::RunReceivers () 
	{ 
		while (m_IsRunning)
		{
			try
			{	
				m_ReceiversService.run ();
			}
			catch (std::exception& ex)
			{
				LogPrint (eLogError, "SSU: receivers runtime exception: ", ex.what ());
			}	
		}	
	}	
	
	void SSUServer::AddRelay (uint32_t tag, const boost::asio::ip::udp::endpoint& relay)
	{
		m_Relays[tag] = relay;
	}	

	std::shared_ptr<SSUSession> SSUServer::FindRelaySession (uint32_t tag)
	{
		auto it = m_Relays.find (tag);
		if (it != m_Relays.end ())
			return FindSession (it->second);
		return nullptr;
	}

	void SSUServer::Send (const uint8_t * buf, size_t len, const boost::asio::ip::udp::endpoint& to)
	{
		if (to.protocol () == boost::asio::ip::udp::v4()) 
			m_Socket.send_to (boost::asio::buffer (buf, len), to);
		else
			m_SocketV6.send_to (boost::asio::buffer (buf, len), to);
	}	

	void SSUServer::Receive ()
	{
		SSUPacket * packet = new SSUPacket ();
		m_Socket.async_receive_from (boost::asio::buffer (packet->buf, SSU_MTU_V4), packet->from,
			std::bind (&SSUServer::HandleReceivedFrom, this, std::placeholders::_1, std::placeholders::_2, packet)); 
	}

	void SSUServer::ReceiveV6 ()
	{
		SSUPacket * packet = new SSUPacket ();
		m_SocketV6.async_receive_from (boost::asio::buffer (packet->buf, SSU_MTU_V6), packet->from,
			std::bind (&SSUServer::HandleReceivedFromV6, this, std::placeholders::_1, std::placeholders::_2, packet)); 
	}	

	void SSUServer::HandleReceivedFrom (const boost::system::error_code& ecode, std::size_t bytes_transferred, SSUPacket * packet)
	{
		if (!ecode)
		{
			packet->len = bytes_transferred;
			std::vector<SSUPacket *> packets;
			packets.push_back (packet);

			boost::system::error_code ec;
			size_t moreBytes = m_Socket.available(ec);
			while (moreBytes && packets.size () < 25)
			{
				packet = new SSUPacket ();
				packet->len = m_Socket.receive_from (boost::asio::buffer (packet->buf, SSU_MTU_V4), packet->from);
				packets.push_back (packet);
				moreBytes = m_Socket.available();
			}

			m_Service.post (std::bind (&SSUServer::HandleReceivedPackets, this, packets, &m_Sessions));
			Receive ();
		}
		else
		{	
			LogPrint (eLogError, "SSU: receive error: ", ecode.message ());
			delete packet;
		}	
	}

	void SSUServer::HandleReceivedFromV6 (const boost::system::error_code& ecode, std::size_t bytes_transferred, SSUPacket * packet)
	{
		if (!ecode)
		{
			packet->len = bytes_transferred;
			std::vector<SSUPacket *> packets;
			packets.push_back (packet);

			size_t moreBytes = m_SocketV6.available ();
			while (moreBytes && packets.size () < 25)
			{
				packet = new SSUPacket ();
				packet->len = m_SocketV6.receive_from (boost::asio::buffer (packet->buf, SSU_MTU_V6), packet->from);
				packets.push_back (packet);
				moreBytes = m_SocketV6.available();
			}

			m_ServiceV6.post (std::bind (&SSUServer::HandleReceivedPackets, this, packets, &m_SessionsV6));
			ReceiveV6 ();
		}
		else
		{	
			LogPrint (eLogError, "SSU: v6 receive error: ", ecode.message ());
			delete packet;
		}	
	}

	void SSUServer::HandleReceivedPackets (std::vector<SSUPacket *> packets, 
		std::map<boost::asio::ip::udp::endpoint, std::shared_ptr<SSUSession> > * sessions)
	{
		std::shared_ptr<SSUSession> session;	
		for (auto it1: packets)
		{
			auto packet = it1;
			try
			{	
				if (!session || session->GetRemoteEndpoint () != packet->from) // we received packet for other session than previous
				{
					if (session) session->FlushData ();
					auto it = sessions->find (packet->from);
					if (it != sessions->end ())
						session = it->second;
					if (!session)
					{
						session = std::make_shared<SSUSession> (*this, packet->from);
						session->WaitForConnect ();
						(*sessions)[packet->from] = session;
						LogPrint (eLogDebug, "SSU: new session from ", packet->from.address ().to_string (), ":", packet->from.port (), " created");
					}
				}
				session->ProcessNextMessage (packet->buf, packet->len, packet->from);
			}	
			catch (std::exception& ex)
			{
				LogPrint (eLogError, "SSU: HandleReceivedPackets ", ex.what ());
				if (session) session->FlushData ();
				session = nullptr;
			}	
			delete packet;
		}
		if (session) session->FlushData ();
	}

	std::shared_ptr<SSUSession> SSUServer::FindSession (std::shared_ptr<const i2p::data::RouterInfo> router) const
	{
		if (!router) return nullptr;
		auto address = router->GetSSUAddress (true); // v4 only
 		if (!address) return nullptr;
		auto session = FindSession (boost::asio::ip::udp::endpoint (address->host, address->port));
		if (session || !context.SupportsV6 ())
			return session;
		// try v6
		address = router->GetSSUV6Address (); 
		if (!address) return nullptr;
		return FindSession (boost::asio::ip::udp::endpoint (address->host, address->port));
	}	

	std::shared_ptr<SSUSession> SSUServer::FindSession (const boost::asio::ip::udp::endpoint& e) const
	{
		auto& sessions = e.address ().is_v6 () ?  m_SessionsV6 : m_Sessions; 
		auto it = sessions.find (e);
		if (it != sessions.end ())
			return it->second;
		else
			return nullptr;
	}
	
	void SSUServer::CreateSession (std::shared_ptr<const i2p::data::RouterInfo> router, bool peerTest)
	{
		auto address = router->GetSSUAddress (!context.SupportsV6 ());
		if (address)
			CreateSession (router, address->host, address->port, peerTest);
		else
			LogPrint (eLogWarning, "SSU: Router ", i2p::data::GetIdentHashAbbreviation (router->GetIdentHash ()), " doesn't have SSU address");
	}
	
	void SSUServer::CreateSession (std::shared_ptr<const i2p::data::RouterInfo> router,
		const boost::asio::ip::address& addr, int port, bool peerTest)
	{
		if (router)
		{
			if (router->UsesIntroducer ())
				m_Service.post (std::bind (&SSUServer::CreateSessionThroughIntroducer, this, router, peerTest)); // always V4 thread
			else
			{
				boost::asio::ip::udp::endpoint remoteEndpoint (addr, port);
				auto& s = addr.is_v6 () ? m_ServiceV6 : m_Service;
				s.post (std::bind (&SSUServer::CreateDirectSession, this, router, remoteEndpoint, peerTest));
			}
		}
	}

	void SSUServer::CreateDirectSession (std::shared_ptr<const i2p::data::RouterInfo> router, boost::asio::ip::udp::endpoint remoteEndpoint, bool peerTest)
	{	
		auto& sessions = remoteEndpoint.address ().is_v6 () ? m_SessionsV6 : m_Sessions;	
		auto it = sessions.find (remoteEndpoint);
		if (it != sessions.end ())
		{	
			auto session = it->second;
			if (peerTest && session->GetState () == eSessionStateEstablished)
				session->SendPeerTest ();
		}	
		else
		{
			// otherwise create new session					
			auto session = std::make_shared<SSUSession> (*this, remoteEndpoint, router, peerTest);
			sessions[remoteEndpoint] = session;
			// connect 					
			LogPrint (eLogDebug, "SSU: Creating new session to [", i2p::data::GetIdentHashAbbreviation (router->GetIdentHash ()), "] ",
				remoteEndpoint.address ().to_string (), ":", remoteEndpoint.port ());
			session->Connect ();
		}
	}
	
	void SSUServer::CreateSessionThroughIntroducer (std::shared_ptr<const i2p::data::RouterInfo> router, bool peerTest)
	{
		if (router && router->UsesIntroducer ())
		{
			auto address = router->GetSSUAddress (true); // v4 only for now
			if (address)
			{
				boost::asio::ip::udp::endpoint remoteEndpoint (address->host, address->port);
				auto it = m_Sessions.find (remoteEndpoint);
				// check if session if presented alredy
				if (it != m_Sessions.end ())
				{	
					auto session = it->second;
					if (peerTest && session->GetState () == eSessionStateEstablished)
						session->SendPeerTest ();
					return; 
				}		
				// create new session					
				int numIntroducers = address->introducers.size ();
				if (numIntroducers > 0)
				{
					std::shared_ptr<SSUSession> introducerSession;
					const i2p::data::RouterInfo::Introducer * introducer = nullptr;
					// we might have a session to introducer already
					for (int i = 0; i < numIntroducers; i++)
					{
						auto intr = &(address->introducers[i]);
						boost::asio::ip::udp::endpoint ep (intr->iHost, intr->iPort);
						if (ep.address ().is_v4 ()) // ipv4 only
						{	
							if (!introducer) introducer = intr; // we pick first one for now
							it = m_Sessions.find (ep); 
							if (it != m_Sessions.end ())
							{
								introducerSession = it->second;
								break; 
							}	
						}
					}
					if (!introducer)
					{
						LogPrint (eLogWarning, "SSU: Can't connect to unreachable router and no ipv4 introducers present");
						return;
					}				

					if (introducerSession) // session found 
						LogPrint (eLogWarning, "SSU: Session to introducer already exists");
					else // create new
					{
						LogPrint (eLogDebug, "SSU: Creating new session to introducer ", introducer->iHost);
						boost::asio::ip::udp::endpoint introducerEndpoint (introducer->iHost, introducer->iPort);
						introducerSession = std::make_shared<SSUSession> (*this, introducerEndpoint, router);
						m_Sessions[introducerEndpoint] = introducerSession;													
					}
					// create session	
					auto session = std::make_shared<SSUSession> (*this, remoteEndpoint, router, peerTest);
					m_Sessions[remoteEndpoint] = session;
					// introduce
					LogPrint (eLogInfo, "SSU: Introduce new session to [", i2p::data::GetIdentHashAbbreviation (router->GetIdentHash ()),
							"] through introducer ", introducer->iHost, ":", introducer->iPort);
					session->WaitForIntroduction ();	
					if (i2p::context.GetRouterInfo ().UsesIntroducer ()) // if we are unreachable
					{
						uint8_t buf[1];
						Send (buf, 0, remoteEndpoint); // send HolePunch
					}	
					introducerSession->Introduce (*introducer, router);
				}
				else	
					LogPrint (eLogWarning, "SSU: Can't connect to unreachable router and no introducers present");
			}
			else
				LogPrint (eLogWarning, "SSU: Router ", i2p::data::GetIdentHashAbbreviation (router->GetIdentHash ()), " doesn't have SSU address");
		}
	}

	void SSUServer::DeleteSession (std::shared_ptr<SSUSession> session)
	{
		if (session)
		{
			session->Close ();
			auto& ep = session->GetRemoteEndpoint ();
			if (ep.address ().is_v6 ())
				m_SessionsV6.erase (ep);
			else
				m_Sessions.erase (ep);
		}	
	}	

	void SSUServer::DeleteAllSessions ()
	{
		for (auto it: m_Sessions)
			it.second->Close ();
		m_Sessions.clear ();

		for (auto it: m_SessionsV6)
			it.second->Close ();
		m_SessionsV6.clear ();
	}

	template<typename Filter>
	std::shared_ptr<SSUSession> SSUServer::GetRandomV4Session (Filter filter) // v4 only
	{
		std::vector<std::shared_ptr<SSUSession> > filteredSessions;
		for (auto s :m_Sessions)
			if (filter (s.second)) filteredSessions.push_back (s.second);
		if (filteredSessions.size () > 0)
		{
			auto ind = rand () % filteredSessions.size ();
			return filteredSessions[ind];
		}
		return nullptr;	
	}

	std::shared_ptr<SSUSession> SSUServer::GetRandomEstablishedV4Session (std::shared_ptr<const SSUSession> excluded) // v4 only
	{
		return GetRandomV4Session (
			[excluded](std::shared_ptr<SSUSession> session)->bool 
			{ 
				return session->GetState () == eSessionStateEstablished && !session->IsV6 () && 
					session != excluded; 
			}
								);
	}

	std::set<SSUSession *> SSUServer::FindIntroducers (int maxNumIntroducers)
	{
		uint32_t ts = i2p::util::GetSecondsSinceEpoch ();
		std::set<SSUSession *> ret;
		for (int i = 0; i < maxNumIntroducers; i++)
		{
			auto session = GetRandomV4Session (
				[&ret, ts](std::shared_ptr<SSUSession> session)->bool 
				{ 
					return session->GetRelayTag () && !ret.count (session.get ()) &&
						session->GetState () == eSessionStateEstablished &&
						ts < session->GetCreationTime () + SSU_TO_INTRODUCER_SESSION_DURATION; 
				}
											);	
			if (session)
			{
				ret.insert (session.get ());
				break;
			}	
		}
		return ret;
	}

	void SSUServer::ScheduleIntroducersUpdateTimer ()
	{
		m_IntroducersUpdateTimer.expires_from_now (boost::posix_time::seconds(SSU_KEEP_ALIVE_INTERVAL));
		m_IntroducersUpdateTimer.async_wait (std::bind (&SSUServer::HandleIntroducersUpdateTimer,
			this, std::placeholders::_1));	
	}

	void SSUServer::HandleIntroducersUpdateTimer (const boost::system::error_code& ecode)
	{
		if (ecode != boost::asio::error::operation_aborted)
		{
			// timeout expired
			if (i2p::context.GetStatus () == eRouterStatusTesting)
			{
				// we still don't know if we need introducers
				ScheduleIntroducersUpdateTimer ();
				return;
			}	
			if (i2p::context.GetStatus () == eRouterStatusOK) return; // we don't need introducers anymore
			// we are firewalled
			if (!i2p::context.IsUnreachable ()) i2p::context.SetUnreachable ();
			std::list<boost::asio::ip::udp::endpoint> newList;
			size_t numIntroducers = 0;
			uint32_t ts = i2p::util::GetSecondsSinceEpoch ();
			for (auto it :m_Introducers)
			{	
				auto session = FindSession (it);
				if (session && ts < session->GetCreationTime () + SSU_TO_INTRODUCER_SESSION_DURATION)
				{
					session->SendKeepAlive ();
					newList.push_back (it);
					numIntroducers++;
				}
				else	
					i2p::context.RemoveIntroducer (it);
			}

			if (numIntroducers < SSU_MAX_NUM_INTRODUCERS)
			{
				// create new
				auto introducers = FindIntroducers (SSU_MAX_NUM_INTRODUCERS);
				if (introducers.size () > 0)
				{
					for (auto it1: introducers)
					{
						auto& ep = it1->GetRemoteEndpoint ();
						i2p::data::RouterInfo::Introducer introducer;
						introducer.iHost = ep.address ();
						introducer.iPort = ep.port ();
						introducer.iTag = it1->GetRelayTag ();
						introducer.iKey = it1->GetIntroKey ();
						if (i2p::context.AddIntroducer (introducer))
						{	
							newList.push_back (ep);
							if (newList.size () >= SSU_MAX_NUM_INTRODUCERS) break;
						}	
					}	
				}	
			}	
			m_Introducers = newList;
			if (m_Introducers.size () < SSU_MAX_NUM_INTRODUCERS)
			{
				auto introducer = i2p::data::netdb.GetRandomIntroducer ();
				if (introducer)
					CreateSession (introducer);
			}	
			ScheduleIntroducersUpdateTimer ();
		}	
	}

	void SSUServer::NewPeerTest (uint32_t nonce, PeerTestParticipant role, std::shared_ptr<SSUSession> session)
	{
		m_PeerTests[nonce] = { i2p::util::GetMillisecondsSinceEpoch (), role, session };
	}

	PeerTestParticipant SSUServer::GetPeerTestParticipant (uint32_t nonce)
	{
		auto it = m_PeerTests.find (nonce);
		if (it != m_PeerTests.end ())
			return it->second.role;
		else
			return ePeerTestParticipantUnknown;
	}	

	std::shared_ptr<SSUSession> SSUServer::GetPeerTestSession (uint32_t nonce)
	{
		auto it = m_PeerTests.find (nonce);
		if (it != m_PeerTests.end ())
			return it->second.session;
		else
			return nullptr;
	}

	void SSUServer::UpdatePeerTest (uint32_t nonce, PeerTestParticipant role)
	{
		auto it = m_PeerTests.find (nonce);
		if (it != m_PeerTests.end ())
			it->second.role = role;
	}	
	
	void SSUServer::RemovePeerTest (uint32_t nonce)
	{
		m_PeerTests.erase (nonce);
	}	

	void SSUServer::SchedulePeerTestsCleanupTimer ()
	{
		m_PeerTestsCleanupTimer.expires_from_now (boost::posix_time::seconds(SSU_PEER_TEST_TIMEOUT));
		m_PeerTestsCleanupTimer.async_wait (std::bind (&SSUServer::HandlePeerTestsCleanupTimer,
			this, std::placeholders::_1));	
	}

	void SSUServer::HandlePeerTestsCleanupTimer (const boost::system::error_code& ecode)
	{
		if (ecode != boost::asio::error::operation_aborted)
		{
			int numDeleted = 0;	
			uint64_t ts = i2p::util::GetMillisecondsSinceEpoch ();	
			for (auto it = m_PeerTests.begin (); it != m_PeerTests.end ();)
			{
				if (ts > it->second.creationTime + SSU_PEER_TEST_TIMEOUT*1000LL)
				{
					numDeleted++;
					it = m_PeerTests.erase (it);
				}
				else
					it++;	 
			}
			if (numDeleted > 0)
				LogPrint (eLogDebug, "SSU: ", numDeleted, " peer tests have been expired");
			SchedulePeerTestsCleanupTimer ();
		}
	}
}
}