i2pd/Transports.cpp

402 lines
11 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 <boost/bind.hpp>
#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 ():
2014-10-27 19:08:50 +00:00
m_Thread (nullptr), m_Work (m_Service), m_NTCPAcceptor (nullptr), m_NTCPV6Acceptor (nullptr),
2014-04-04 20:29:40 +00:00
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)
{
2014-10-28 00:36:03 +00:00
if (address.transportStyle == RouterInfo::eTransportNTCP && address.host.is_v4 ())
2013-10-27 15:26:39 +00:00
{
m_NTCPAcceptor = new boost::asio::ip::tcp::acceptor (m_Service,
boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), address.port));
2014-01-23 21:10:33 +00:00
LogPrint ("Start listening TCP port ", address.port);
2014-11-25 21:30:15 +00:00
auto conn = std::make_shared<NTCPSession>(m_Service);
2013-10-27 15:26:39 +00:00
m_NTCPAcceptor->async_accept(conn->GetSocket (), boost::bind (&Transports::HandleAccept, this,
2014-10-28 00:36:03 +00:00
conn, boost::asio::placeholders::error));
2014-10-27 19:08:50 +00:00
if (context.SupportsV6 ())
{
2014-10-28 00:36:03 +00:00
m_NTCPV6Acceptor = new boost::asio::ip::tcp::acceptor (m_Service);
m_NTCPV6Acceptor->open (boost::asio::ip::tcp::v6());
m_NTCPV6Acceptor->set_option (boost::asio::ip::v6_only (true));
m_NTCPV6Acceptor->bind (boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v6(), address.port));
m_NTCPV6Acceptor->listen ();
2014-10-27 19:08:50 +00:00
LogPrint ("Start listening V6 TCP port ", address.port);
2014-11-25 21:30:15 +00:00
auto conn = std::make_shared<NTCPSession> (m_Service);
2014-10-27 19:08:50 +00:00
m_NTCPV6Acceptor->async_accept(conn->GetSocket (), boost::bind (&Transports::HandleAcceptV6,
2014-10-28 00:36:03 +00:00
this, conn, boost::asio::placeholders::error));
}
2014-02-25 00:25:26 +00:00
}
2014-10-29 19:02:48 +00:00
else 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
}
}
void Transports::Stop ()
{
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
}
m_NTCPSessions.clear ();
2014-11-25 21:30:15 +00:00
2014-04-09 01:56:34 +00:00
delete m_NTCPAcceptor;
2014-09-22 18:44:10 +00:00
m_NTCPAcceptor = nullptr;
2014-10-27 19:08:50 +00:00
delete m_NTCPV6Acceptor;
m_NTCPV6Acceptor = 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
}
}
2014-11-25 21:30:15 +00:00
void Transports::AddNTCPSession (std::shared_ptr<NTCPSession> session)
2013-10-27 15:26:39 +00:00
{
if (session)
2014-10-24 19:39:53 +00:00
m_NTCPSessions[session->GetRemoteIdentity ().GetIdentHash ()] = session;
2013-10-27 15:26:39 +00:00
}
2014-11-25 21:30:15 +00:00
void Transports::RemoveNTCPSession (std::shared_ptr<NTCPSession> session)
2013-10-27 15:26:39 +00:00
{
if (session)
2014-10-24 19:39:53 +00:00
m_NTCPSessions.erase (session->GetRemoteIdentity ().GetIdentHash ());
2013-10-27 15:26:39 +00:00
}
2014-11-25 21:30:15 +00:00
void Transports::HandleAccept (std::shared_ptr<NTCPSession> conn, const boost::system::error_code& error)
2013-10-27 15:26:39 +00:00
{
if (!error)
{
LogPrint ("Connected from ", conn->GetSocket ().remote_endpoint().address ().to_string ());
conn->ServerLogin ();
}
2014-11-25 21:30:15 +00:00
2013-10-27 15:26:39 +00:00
2014-09-22 17:28:46 +00:00
if (error != boost::asio::error::operation_aborted)
{
2014-11-25 21:30:15 +00:00
conn = std::make_shared<NTCPSession> (m_Service);
2014-09-22 17:28:46 +00:00
m_NTCPAcceptor->async_accept(conn->GetSocket (), boost::bind (&Transports::HandleAccept, this,
conn, boost::asio::placeholders::error));
}
2013-10-27 15:26:39 +00:00
}
2014-11-25 21:30:15 +00:00
void Transports::HandleAcceptV6 (std::shared_ptr<NTCPSession> conn, const boost::system::error_code& error)
2014-10-27 19:08:50 +00:00
{
if (!error)
{
LogPrint ("Connected from ", conn->GetSocket ().remote_endpoint().address ().to_string ());
conn->ServerLogin ();
}
if (error != boost::asio::error::operation_aborted)
{
2014-11-25 21:30:15 +00:00
conn = std::make_shared<NTCPSession> (m_Service);
2014-10-27 19:08:50 +00:00
m_NTCPV6Acceptor->async_accept(conn->GetSocket (), boost::bind (&Transports::HandleAcceptV6, this,
conn, boost::asio::placeholders::error));
}
}
2014-11-25 21:30:15 +00:00
void Transports::Connect (const boost::asio::ip::address& address, int port, std::shared_ptr<NTCPSession> conn)
2014-11-25 20:16:03 +00:00
{
LogPrint ("Connecting to ", address ,":", port);
conn->GetSocket ().async_connect (boost::asio::ip::tcp::endpoint (address, port),
boost::bind (&Transports::HandleConnect, this, boost::asio::placeholders::error, conn));
}
2014-11-25 21:30:15 +00:00
void Transports::HandleConnect (const boost::system::error_code& ecode, std::shared_ptr<NTCPSession> conn)
2014-11-25 20:16:03 +00:00
{
if (ecode)
{
LogPrint ("Connect error: ", ecode.message ());
if (ecode != boost::asio::error::operation_aborted)
{
i2p::data::netdb.SetUnreachable (conn->GetRemoteIdentity ().GetIdentHash (), true);
conn->Terminate ();
}
}
else
{
LogPrint ("Connected");
if (conn->GetSocket ().local_endpoint ().protocol () == boost::asio::ip::tcp::v6()) // ipv6
context.UpdateNTCPV6Address (conn->GetSocket ().local_endpoint ().address ());
conn->ClientLogin ();
}
}
2014-11-25 21:30:15 +00:00
std::shared_ptr<NTCPSession> Transports::GetNextNTCPSession ()
2013-10-27 15:26:39 +00:00
{
for (auto session: m_NTCPSessions)
if (session.second->IsEstablished ())
return session.second;
return 0;
}
2014-11-25 21:30:15 +00:00
std::shared_ptr<NTCPSession> Transports::FindNTCPSession (const i2p::data::IdentHash& ident)
2013-10-27 15:26:39 +00:00
{
2013-11-29 12:52:09 +00:00
auto it = m_NTCPSessions.find (ident);
2013-10-27 15:26:39 +00:00
if (it != m_NTCPSessions.end ())
return it->second;
return 0;
}
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
{
2013-11-29 12:52:09 +00:00
if (ident == i2p::context.GetRouterInfo ().GetIdentHash ())
2013-10-27 15:26:39 +00:00
// we send it to ourself
i2p::HandleI2NPMessage (msg);
2013-10-27 15:26:39 +00:00
else
2013-12-29 15:48:57 +00:00
m_Service.post (boost::bind (&Transports::PostMessage, this, ident, msg));
}
void Transports::PostMessage (const i2p::data::IdentHash& ident, i2p::I2NPMessage * msg)
{
auto session = FindNTCPSession (ident);
2014-03-26 01:17:03 +00:00
if (session)
session->SendI2NPMessage (msg);
else
2013-12-29 15:48:57 +00:00
{
auto r = netdb.FindRouter (ident);
2013-12-29 15:48:57 +00:00
if (r)
{
2014-11-24 17:26:11 +00:00
auto ssuSession = m_SSUServer ? m_SSUServer->FindSession (r) : nullptr;
2014-03-26 01:17:03 +00:00
if (ssuSession)
ssuSession->SendI2NPMessage (msg);
2014-01-19 15:05:54 +00:00
else
2014-02-11 00:27:55 +00:00
{
2014-03-26 01:17:03 +00:00
// existing session not found. create new
// try NTCP first if message size < 16K
auto address = r->GetNTCPAddress (!context.SupportsV6 ());
if (address && !r->UsesIntroducer () && !r->IsUnreachable () && msg->GetLength () < NTCP_MAX_MESSAGE_SIZE)
2014-03-26 01:17:03 +00:00
{
2014-11-25 21:30:15 +00:00
auto s = std::make_shared<NTCPSession> (m_Service, r);
2014-03-26 01:17:03 +00:00
AddNTCPSession (s);
s->SendI2NPMessage (msg);
2014-11-25 20:16:03 +00:00
Connect (address->host, address->port, s);
2014-03-26 01:17:03 +00:00
}
else
{
// then SSU
auto s = m_SSUServer ? m_SSUServer->GetSession (r) : nullptr;
2014-02-11 00:27:55 +00:00
if (s)
s->SendI2NPMessage (msg);
2014-03-26 01:17:03 +00:00
else
2014-07-03 14:33:23 +00:00
{
2014-03-26 01:17:03 +00:00
LogPrint ("No NTCP and SSU addresses available");
2014-07-03 14:33:23 +00:00
DeleteI2NPMessage (msg);
}
2014-02-11 00:27:55 +00:00
}
}
2013-12-29 15:48:57 +00:00
}
2014-01-19 15:05:54 +00:00
else
{
LogPrint ("Router not found. Requested");
i2p::data::netdb.RequestDestination (ident);
auto resendTimer = new boost::asio::deadline_timer (m_Service);
resendTimer->expires_from_now (boost::posix_time::seconds(5)); // 5 seconds
resendTimer->async_wait (boost::bind (&Transports::HandleResendTimer,
this, boost::asio::placeholders::error, resendTimer, ident, msg));
2014-01-19 15:05:54 +00:00
}
2013-10-27 15:26:39 +00:00
}
}
2014-02-09 02:06:40 +00:00
void Transports::HandleResendTimer (const boost::system::error_code& ecode,
boost::asio::deadline_timer * timer, const i2p::data::IdentHash& ident, i2p::I2NPMessage * msg)
{
auto r = netdb.FindRouter (ident);
if (r)
{
LogPrint ("Router found. Sending message");
PostMessage (ident, msg);
}
else
{
LogPrint ("Router not found. Failed to send message");
DeleteI2NPMessage (msg);
}
delete timer;
}
2014-11-24 17:26:11 +00:00
void Transports::CloseSession (std::shared_ptr<const i2p::data::RouterInfo> router)
{
if (!router) return;
m_Service.post (boost::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);
}
2013-10-27 15:26:39 +00:00
}
}