i2pd/Transports.cpp

468 lines
12 KiB
C++
Raw Normal View History

2014-10-20 19:19:56 +00:00
#include <cryptopp/dh.h>
2013-10-27 15:26:39 +00:00
#include "Log.h"
2014-10-20 19:19:56 +00:00
#include "CryptoConst.h"
2013-10-27 15:26:39 +00:00
#include "RouterContext.h"
#include "I2NPProtocol.h"
#include "NetDb.h"
#include "Transports.h"
using namespace i2p::data;
namespace i2p
{
namespace transport
2013-10-27 15:26:39 +00:00
{
2014-10-20 19:19:56 +00:00
DHKeysPairSupplier::DHKeysPairSupplier (int size):
m_QueueSize (size), m_IsRunning (false), m_Thread (nullptr)
{
}
2014-04-04 20:29:40 +00:00
DHKeysPairSupplier::~DHKeysPairSupplier ()
{
Stop ();
}
void DHKeysPairSupplier::Start ()
{
m_IsRunning = true;
m_Thread = new std::thread (std::bind (&DHKeysPairSupplier::Run, this));
}
void DHKeysPairSupplier::Stop ()
{
m_IsRunning = false;
m_Acquired.notify_one ();
if (m_Thread)
{
m_Thread->join ();
delete m_Thread;
m_Thread = 0;
}
}
void DHKeysPairSupplier::Run ()
{
while (m_IsRunning)
{
int num;
while ((num = m_QueueSize - m_Queue.size ()) > 0)
CreateDHKeysPairs (num);
std::unique_lock<std::mutex> l(m_AcquiredMutex);
m_Acquired.wait (l); // wait for element gets aquired
}
}
void DHKeysPairSupplier::CreateDHKeysPairs (int num)
{
if (num > 0)
{
2014-10-20 19:19:56 +00:00
CryptoPP::DH dh (i2p::crypto::elgp, i2p::crypto::elgg);
2014-04-04 20:29:40 +00:00
for (int i = 0; i < num; i++)
{
2014-10-20 20:09:59 +00:00
i2p::transport::DHKeysPair * pair = new i2p::transport::DHKeysPair ();
2014-10-20 19:19:56 +00:00
dh.GenerateKeyPair(m_Rnd, pair->privateKey, pair->publicKey);
2014-07-07 16:34:19 +00:00
std::unique_lock<std::mutex> l(m_AcquiredMutex);
2014-04-04 20:29:40 +00:00
m_Queue.push (pair);
}
}
}
DHKeysPair * DHKeysPairSupplier::Acquire ()
2014-04-04 20:29:40 +00:00
{
if (!m_Queue.empty ())
{
2014-07-07 16:34:19 +00:00
std::unique_lock<std::mutex> l(m_AcquiredMutex);
2014-04-04 20:29:40 +00:00
auto pair = m_Queue.front ();
m_Queue.pop ();
m_Acquired.notify_one ();
return pair;
}
else // queue is empty, create new
{
DHKeysPair * pair = new DHKeysPair ();
2014-10-20 19:19:56 +00:00
CryptoPP::DH dh (i2p::crypto::elgp, i2p::crypto::elgg);
dh.GenerateKeyPair(m_Rnd, pair->privateKey, pair->publicKey);
2014-04-04 20:29:40 +00:00
return pair;
}
}
void DHKeysPairSupplier::Return (DHKeysPair * pair)
{
std::unique_lock<std::mutex> l(m_AcquiredMutex);
m_Queue.push (pair);
}
2013-10-27 15:26:39 +00:00
Transports transports;
Transports::Transports ():
2015-02-11 19:45:25 +00:00
m_IsRunning (false), m_Thread (nullptr), m_Work (m_Service), m_PeerCleanupTimer (m_Service),
2015-01-11 22:41:56 +00:00
m_NTCPServer (nullptr), m_SSUServer (nullptr),
m_DHKeysPairSupplier (5) // 5 pre-generated keys
2013-10-27 15:26:39 +00:00
{
}
Transports::~Transports ()
{
Stop ();
}
void Transports::Start ()
{
2014-04-04 20:29:40 +00:00
m_DHKeysPairSupplier.Start ();
2013-12-29 15:48:57 +00:00
m_IsRunning = true;
2013-10-27 15:26:39 +00:00
m_Thread = new std::thread (std::bind (&Transports::Run, this));
// create acceptors
auto addresses = context.GetRouterInfo ().GetAddresses ();
for (auto& address : addresses)
{
2015-01-11 22:41:56 +00:00
if (!m_NTCPServer)
2013-10-27 15:26:39 +00:00
{
2015-01-11 22:41:56 +00:00
m_NTCPServer = new NTCPServer (address.port);
m_NTCPServer->Start ();
2014-02-25 00:25:26 +00:00
}
2015-01-11 22:41:56 +00:00
if (address.transportStyle == RouterInfo::eTransportSSU && address.host.is_v4 ())
2014-02-25 00:25:26 +00:00
{
2014-01-23 21:10:33 +00:00
if (!m_SSUServer)
{
m_SSUServer = new SSUServer (address.port);
2014-01-23 21:10:33 +00:00
LogPrint ("Start listening UDP port ", address.port);
m_SSUServer->Start ();
2014-02-25 00:25:26 +00:00
DetectExternalIP ();
2014-01-23 21:10:33 +00:00
}
else
LogPrint ("SSU server already exists");
}
2013-10-27 15:26:39 +00:00
}
2015-02-11 19:45:25 +00:00
m_PeerCleanupTimer.expires_from_now (boost::posix_time::seconds(5*SESSION_CREATION_TIMEOUT));
m_PeerCleanupTimer.async_wait (std::bind (&Transports::HandlePeerCleanupTimer, this, std::placeholders::_1));
2013-10-27 15:26:39 +00:00
}
void Transports::Stop ()
{
2015-02-11 19:45:25 +00:00
m_PeerCleanupTimer.cancel ();
2015-01-13 03:53:35 +00:00
m_Peers.clear ();
2014-01-23 21:10:33 +00:00
if (m_SSUServer)
{
m_SSUServer->Stop ();
delete m_SSUServer;
2014-09-22 17:28:46 +00:00
m_SSUServer = nullptr;
2014-04-09 01:56:34 +00:00
}
2015-01-11 22:41:56 +00:00
if (m_NTCPServer)
{
m_NTCPServer->Stop ();
delete m_NTCPServer;
m_NTCPServer = nullptr;
}
2014-01-23 21:10:33 +00:00
2014-04-04 20:29:40 +00:00
m_DHKeysPairSupplier.Stop ();
2013-12-29 15:48:57 +00:00
m_IsRunning = false;
2013-10-27 15:26:39 +00:00
m_Service.stop ();
if (m_Thread)
{
m_Thread->join ();
delete m_Thread;
2014-09-22 17:28:46 +00:00
m_Thread = nullptr;
2013-10-27 15:26:39 +00:00
}
}
2013-11-29 12:52:09 +00:00
void Transports::Run ()
{
2013-12-29 15:48:57 +00:00
while (m_IsRunning)
2013-11-29 12:52:09 +00:00
{
2013-12-29 15:48:57 +00:00
try
{
m_Service.run ();
}
catch (std::exception& ex)
{
LogPrint ("Transports: ", ex.what ());
}
2013-11-29 12:52:09 +00:00
}
}
2013-10-27 15:26:39 +00:00
2013-11-29 12:52:09 +00:00
void Transports::SendMessage (const i2p::data::IdentHash& ident, i2p::I2NPMessage * msg)
2013-10-27 15:26:39 +00:00
{
2015-01-17 13:22:43 +00:00
m_Service.post (std::bind (&Transports::PostMessage, this, ident, msg));
2013-12-29 15:48:57 +00:00
}
2015-01-21 02:05:57 +00:00
void Transports::SendMessages (const i2p::data::IdentHash& ident, const std::vector<i2p::I2NPMessage *>& msgs)
{
m_Service.post (std::bind (&Transports::PostMessages, this, ident, msgs));
}
void Transports::PostMessage (i2p::data::IdentHash ident, i2p::I2NPMessage * msg)
2013-12-29 15:48:57 +00:00
{
if (ident == i2p::context.GetRouterInfo ().GetIdentHash ())
{
// we send it to ourself
i2p::HandleI2NPMessage (msg);
return;
}
2015-01-14 02:31:39 +00:00
auto it = m_Peers.find (ident);
if (it == m_Peers.end ())
2013-12-29 15:48:57 +00:00
{
auto r = netdb.FindRouter (ident);
2015-02-11 19:45:25 +00:00
it = m_Peers.insert (std::pair<i2p::data::IdentHash, Peer>(ident, { 0, r, nullptr,
i2p::util::GetSecondsSinceEpoch () })).first;
2015-01-14 02:31:39 +00:00
if (!ConnectToPeer (ident, it->second))
{
DeleteI2NPMessage (msg);
return;
}
}
if (it->second.session)
it->second.session->SendI2NPMessage (msg);
else
it->second.delayedMessages.push_back (msg);
}
void Transports::PostMessages (i2p::data::IdentHash ident, std::vector<i2p::I2NPMessage *> msgs)
2015-01-21 02:05:57 +00:00
{
if (ident == i2p::context.GetRouterInfo ().GetIdentHash ())
{
// we send it to ourself
for (auto it: msgs)
i2p::HandleI2NPMessage (it);
return;
}
auto it = m_Peers.find (ident);
if (it == m_Peers.end ())
{
auto r = netdb.FindRouter (ident);
2015-02-11 19:45:25 +00:00
it = m_Peers.insert (std::pair<i2p::data::IdentHash, Peer>(ident, { 0, r, nullptr,
i2p::util::GetSecondsSinceEpoch () })).first;
2015-01-21 02:05:57 +00:00
if (!ConnectToPeer (ident, it->second))
{
for (auto it1: msgs)
DeleteI2NPMessage (it1);
return;
}
}
if (it->second.session)
it->second.session->SendI2NPMessages (msgs);
else
{
for (auto it1: msgs)
it->second.delayedMessages.push_back (it1);
}
}
2015-01-14 02:31:39 +00:00
bool Transports::ConnectToPeer (const i2p::data::IdentHash& ident, Peer& peer)
{
if (peer.router) // we have RI already
{
if (!peer.numAttempts) // NTCP
{
peer.numAttempts++;
2015-01-16 20:25:44 +00:00
auto address = peer.router->GetNTCPAddress (!context.SupportsV6 ());
if (address)
{
2015-02-04 14:40:00 +00:00
#if BOOST_VERSION >= 104900
2015-01-16 20:25:44 +00:00
if (!address->host.is_unspecified ()) // we have address now
2015-02-04 14:40:00 +00:00
#else
boost::system::error_code ecode;
address->host.to_string (ecode);
if (!ecode)
#endif
2015-01-16 20:25:44 +00:00
{
if (!peer.router->UsesIntroducer () && !peer.router->IsUnreachable ())
{
auto s = std::make_shared<NTCPSession> (*m_NTCPServer, peer.router);
m_NTCPServer->Connect (address->host, address->port, s);
return true;
}
}
else // we don't have address
{
2015-01-17 04:01:40 +00:00
if (address->addressString.length () > 0) // trying to resolve
2015-01-16 20:25:44 +00:00
{
LogPrint (eLogInfo, "Resolving ", address->addressString);
NTCPResolve (address->addressString, ident);
return true;
}
}
2014-02-11 00:27:55 +00:00
}
2013-12-29 15:48:57 +00:00
}
2015-01-14 02:31:39 +00:00
else if (peer.numAttempts == 1)// SSU
2014-01-19 15:05:54 +00:00
{
2015-01-14 02:31:39 +00:00
peer.numAttempts++;
if (m_SSUServer)
{
if (m_SSUServer->GetSession (peer.router))
return true;
}
2014-01-19 15:05:54 +00:00
}
2015-01-14 02:31:39 +00:00
LogPrint (eLogError, "No NTCP and SSU addresses available");
2015-02-10 22:36:39 +00:00
if (peer.session) peer.session->Done ();
2015-01-14 02:31:39 +00:00
m_Peers.erase (ident);
return false;
}
else // otherwise request RI
{
LogPrint ("Router not found. Requested");
i2p::data::netdb.RequestDestination (ident, std::bind (
&Transports::RequestComplete, this, std::placeholders::_1, ident));
2013-10-27 15:26:39 +00:00
}
2015-01-14 02:31:39 +00:00
return true;
2013-10-27 15:26:39 +00:00
}
void Transports::RequestComplete (std::shared_ptr<const i2p::data::RouterInfo> r, const i2p::data::IdentHash& ident)
{
m_Service.post (std::bind (&Transports::HandleRequestComplete, this, r, ident));
}
void Transports::HandleRequestComplete (std::shared_ptr<const i2p::data::RouterInfo> r, const i2p::data::IdentHash& ident)
{
2015-01-14 02:31:39 +00:00
auto it = m_Peers.find (ident);
if (it != m_Peers.end ())
{
if (r)
{
LogPrint ("Router found. Trying to connect");
it->second.router = r;
ConnectToPeer (ident, it->second);
}
else
{
LogPrint ("Router not found. Failed to send messages");
m_Peers.erase (it);
}
}
}
2015-01-16 20:25:44 +00:00
2015-01-17 04:01:40 +00:00
void Transports::NTCPResolve (const std::string& addr, const i2p::data::IdentHash& ident)
2015-01-16 20:25:44 +00:00
{
auto resolver = std::make_shared<boost::asio::ip::tcp::resolver>(m_Service);
2015-01-17 13:22:43 +00:00
resolver->async_resolve (boost::asio::ip::tcp::resolver::query (addr, ""),
2015-01-16 20:25:44 +00:00
std::bind (&Transports::HandleNTCPResolve, this,
std::placeholders::_1, std::placeholders::_2, ident, resolver));
}
void Transports::HandleNTCPResolve (const boost::system::error_code& ecode, boost::asio::ip::tcp::resolver::iterator it,
2015-02-15 15:23:06 +00:00
i2p::data::IdentHash ident, std::shared_ptr<boost::asio::ip::tcp::resolver> resolver)
2015-01-16 20:25:44 +00:00
{
auto it1 = m_Peers.find (ident);
2015-02-15 15:23:06 +00:00
if (it1 != m_Peers.end ())
2015-01-16 20:25:44 +00:00
{
auto& peer = it1->second;
2015-02-15 15:23:06 +00:00
if (!ecode && peer.router)
2015-01-16 20:25:44 +00:00
{
auto address = (*it).endpoint ().address ();
LogPrint (eLogInfo, (*it).host_name (), " has been resolved to ", address);
auto addr = peer.router->GetNTCPAddress ();
if (addr)
{
auto s = std::make_shared<NTCPSession> (*m_NTCPServer, peer.router);
m_NTCPServer->Connect (address, addr->port, s);
return;
}
}
2015-02-15 15:23:06 +00:00
LogPrint (eLogError, "Unable to resolve NTCP address: ", ecode.message ());
m_Peers.erase (it1);
2015-01-16 20:25:44 +00:00
}
}
2014-11-24 17:26:11 +00:00
void Transports::CloseSession (std::shared_ptr<const i2p::data::RouterInfo> router)
{
if (!router) return;
2015-01-17 13:22:43 +00:00
m_Service.post (std::bind (&Transports::PostCloseSession, this, router));
}
2014-11-24 17:26:11 +00:00
void Transports::PostCloseSession (std::shared_ptr<const i2p::data::RouterInfo> router)
{
auto ssuSession = m_SSUServer ? m_SSUServer->FindSession (router) : nullptr;
if (ssuSession) // try SSU first
{
m_SSUServer->DeleteSession (ssuSession);
LogPrint ("SSU session closed");
}
// TODO: delete NTCP
}
2014-02-09 02:06:40 +00:00
void Transports::DetectExternalIP ()
{
2014-10-29 19:02:48 +00:00
for (int i = 0; i < 5; i++)
2014-02-09 02:06:40 +00:00
{
auto router = i2p::data::netdb.GetRandomRouter ();
if (router && router->IsSSU () && m_SSUServer)
m_SSUServer->GetSession (router, true); // peer test
2014-02-09 02:06:40 +00:00
}
}
2014-10-20 19:19:56 +00:00
DHKeysPair * Transports::GetNextDHKeysPair ()
2014-04-04 17:30:13 +00:00
{
2014-04-04 20:29:40 +00:00
return m_DHKeysPairSupplier.Acquire ();
2014-04-04 17:30:13 +00:00
}
void Transports::ReuseDHKeysPair (DHKeysPair * pair)
{
m_DHKeysPairSupplier.Return (pair);
}
2015-01-13 03:53:35 +00:00
void Transports::PeerConnected (std::shared_ptr<TransportSession> session)
{
m_Service.post([session, this]()
{
auto ident = session->GetRemoteIdentity ().GetIdentHash ();
auto it = m_Peers.find (ident);
if (it != m_Peers.end ())
{
if (!it->second.session)
{
it->second.session = session;
session->SendI2NPMessages (it->second.delayedMessages);
it->second.delayedMessages.clear ();
}
else
{
LogPrint (eLogError, "Session for ", ident.ToBase64 ().substr (0, 4), " already exists");
2015-02-07 01:53:48 +00:00
session->Done ();
}
2015-01-13 03:53:35 +00:00
}
2015-01-14 02:31:39 +00:00
else // incoming connection
2015-02-11 19:45:25 +00:00
m_Peers.insert (std::make_pair (ident, Peer{ 0, nullptr, session, i2p::util::GetSecondsSinceEpoch () }));
2015-01-13 03:53:35 +00:00
});
}
void Transports::PeerDisconnected (std::shared_ptr<TransportSession> session)
{
m_Service.post([session, this]()
{
auto ident = session->GetRemoteIdentity ().GetIdentHash ();
auto it = m_Peers.find (ident);
if (it != m_Peers.end () && (!it->second.session || it->second.session == session))
2015-01-14 02:31:39 +00:00
{
if (it->second.delayedMessages.size () > 0)
ConnectToPeer (ident, it->second);
else
m_Peers.erase (it);
}
2015-01-13 03:53:35 +00:00
});
}
2015-02-11 19:45:25 +00:00
void Transports::HandlePeerCleanupTimer (const boost::system::error_code& ecode)
{
2015-02-12 03:48:26 +00:00
if (ecode != boost::asio::error::operation_aborted)
2015-02-11 19:45:25 +00:00
{
2015-02-12 03:48:26 +00:00
auto ts = i2p::util::GetSecondsSinceEpoch ();
for (auto it = m_Peers.begin (); it != m_Peers.end (); )
2015-02-11 19:45:25 +00:00
{
2015-02-12 03:48:26 +00:00
if (!it->second.session && ts > it->second.creationTime + SESSION_CREATION_TIMEOUT)
{
LogPrint (eLogError, "Session to peer ", it->first.ToBase64 (), " has not been created in ", SESSION_CREATION_TIMEOUT, " seconds");
it = m_Peers.erase (it);
}
else
it++;
2015-02-11 19:45:25 +00:00
}
2015-02-12 03:48:26 +00:00
m_PeerCleanupTimer.expires_from_now (boost::posix_time::seconds(5*SESSION_CREATION_TIMEOUT));
m_PeerCleanupTimer.async_wait (std::bind (&Transports::HandlePeerCleanupTimer, this, std::placeholders::_1));
}
2015-02-11 19:45:25 +00:00
}
2013-10-27 15:26:39 +00:00
}
}