From 8b9f427aa4882690662feea3e66f4a808bb36ac3 Mon Sep 17 00:00:00 2001 From: orignal Date: Wed, 27 Nov 2024 18:31:58 -0500 Subject: [PATCH] handle session handshakes in separate thread --- libi2pd/NTCP2.cpp | 135 ++++++++++++++++++++++++++++------------------ libi2pd/NTCP2.h | 20 ++++++- 2 files changed, 102 insertions(+), 53 deletions(-) diff --git a/libi2pd/NTCP2.cpp b/libi2pd/NTCP2.cpp index 38efc825..ca5e9260 100644 --- a/libi2pd/NTCP2.cpp +++ b/libi2pd/NTCP2.cpp @@ -491,7 +491,6 @@ namespace transport void NTCP2Session::HandleSessionRequestReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred) { - (void) bytes_transferred; if (ecode) { LogPrint (eLogWarning, "NTCP2: SessionRequest read error: ", ecode.message ()); @@ -499,38 +498,47 @@ namespace transport } else { - LogPrint (eLogDebug, "NTCP2: SessionRequest received ", bytes_transferred); - uint16_t paddingLen = 0; - bool clockSkew = false; - if (m_Establisher->ProcessSessionRequestMessage (paddingLen, clockSkew)) + boost::asio::post (m_Server.GetEstablisherService (), + [s = shared_from_this (), bytes_transferred] () + { + s->ProcessSessionRequest (bytes_transferred);; + }); + } + } + + void NTCP2Session::ProcessSessionRequest (size_t len) + { + LogPrint (eLogDebug, "NTCP2: SessionRequest received ", len); + uint16_t paddingLen = 0; + bool clockSkew = false; + if (m_Establisher->ProcessSessionRequestMessage (paddingLen, clockSkew)) + { + if (clockSkew) { - if (clockSkew) + // we don't care about padding, send SessionCreated and close session + SendSessionCreated (); + boost::asio::post (m_Server.GetService (), std::bind (&NTCP2Session::Terminate, shared_from_this ())); + } + else if (paddingLen > 0) + { + if (paddingLen <= NTCP2_SESSION_REQUEST_MAX_SIZE - 64) // session request is 287 bytes max { - // we don't care about padding, send SessionCreated and close session - SendSessionCreated (); - boost::asio::post (m_Server.GetService (), std::bind (&NTCP2Session::Terminate, shared_from_this ())); + boost::asio::async_read (m_Socket, boost::asio::buffer(m_Establisher->m_SessionRequestBuffer + 64, paddingLen), boost::asio::transfer_all (), + std::bind(&NTCP2Session::HandleSessionRequestPaddingReceived, shared_from_this (), std::placeholders::_1, std::placeholders::_2)); } - else if (paddingLen > 0) + else { - if (paddingLen <= NTCP2_SESSION_REQUEST_MAX_SIZE - 64) // session request is 287 bytes max - { - boost::asio::async_read (m_Socket, boost::asio::buffer(m_Establisher->m_SessionRequestBuffer + 64, paddingLen), boost::asio::transfer_all (), - std::bind(&NTCP2Session::HandleSessionRequestPaddingReceived, shared_from_this (), std::placeholders::_1, std::placeholders::_2)); - } - else - { - LogPrint (eLogWarning, "NTCP2: SessionRequest padding length ", (int)paddingLen, " is too long"); - Terminate (); - } + LogPrint (eLogWarning, "NTCP2: SessionRequest padding length ", (int)paddingLen, " is too long"); + boost::asio::post (m_Server.GetService (), std::bind (&NTCP2Session::Terminate, shared_from_this ())); } - else - SendSessionCreated (); } else - Terminate (); + SendSessionCreated (); } - } - + else + boost::asio::post (m_Server.GetService (), std::bind (&NTCP2Session::Terminate, shared_from_this ())); + } + void NTCP2Session::HandleSessionRequestPaddingReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred) { if (ecode) @@ -539,7 +547,13 @@ namespace transport Terminate (); } else - SendSessionCreated (); + { + boost::asio::post (m_Server.GetEstablisherService (), + [s = shared_from_this ()] () + { + s->SendSessionCreated (); + }); + } } void NTCP2Session::SendSessionCreated () @@ -561,35 +575,44 @@ namespace transport else { m_HandshakeInterval = i2p::util::GetMillisecondsSinceEpoch () - m_HandshakeInterval; - LogPrint (eLogDebug, "NTCP2: SessionCreated received ", bytes_transferred); - uint16_t paddingLen = 0; - if (m_Establisher->ProcessSessionCreatedMessage (paddingLen)) + boost::asio::post (m_Server.GetEstablisherService (), + [s = shared_from_this (), bytes_transferred] () + { + s->ProcessSessionCreated (bytes_transferred); + }); + } + } + + void NTCP2Session::ProcessSessionCreated (size_t len) + { + LogPrint (eLogDebug, "NTCP2: SessionCreated received ", len); + uint16_t paddingLen = 0; + if (m_Establisher->ProcessSessionCreatedMessage (paddingLen)) + { + if (paddingLen > 0) { - if (paddingLen > 0) + if (paddingLen <= NTCP2_SESSION_CREATED_MAX_SIZE - 64) // session created is 287 bytes max { - if (paddingLen <= NTCP2_SESSION_CREATED_MAX_SIZE - 64) // session created is 287 bytes max - { - boost::asio::async_read (m_Socket, boost::asio::buffer(m_Establisher->m_SessionCreatedBuffer + 64, paddingLen), boost::asio::transfer_all (), - std::bind(&NTCP2Session::HandleSessionCreatedPaddingReceived, shared_from_this (), std::placeholders::_1, std::placeholders::_2)); - } - else - { - LogPrint (eLogWarning, "NTCP2: SessionCreated padding length ", (int)paddingLen, " is too long"); - Terminate (); - } + boost::asio::async_read (m_Socket, boost::asio::buffer(m_Establisher->m_SessionCreatedBuffer + 64, paddingLen), boost::asio::transfer_all (), + std::bind(&NTCP2Session::HandleSessionCreatedPaddingReceived, shared_from_this (), std::placeholders::_1, std::placeholders::_2)); } else - SendSessionConfirmed (); + { + LogPrint (eLogWarning, "NTCP2: SessionCreated padding length ", (int)paddingLen, " is too long"); + boost::asio::post (m_Server.GetService (), std::bind (&NTCP2Session::Terminate, shared_from_this ())); + } } else - { - if (GetRemoteIdentity ()) - i2p::data::netdb.SetUnreachable (GetRemoteIdentity ()->GetIdentHash (), true); // assume wrong s key - Terminate (); - } + SendSessionConfirmed (); } - } - + else + { + if (GetRemoteIdentity ()) + i2p::data::netdb.SetUnreachable (GetRemoteIdentity ()->GetIdentHash (), true); // assume wrong s key + boost::asio::post (m_Server.GetService (), std::bind (&NTCP2Session::Terminate, shared_from_this ())); + } + } + void NTCP2Session::HandleSessionCreatedPaddingReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred) { if (ecode) @@ -600,7 +623,11 @@ namespace transport else { m_Establisher->m_SessionCreatedBufferLen += bytes_transferred; - SendSessionConfirmed (); + boost::asio::post (m_Server.GetEstablisherService (), + [s = shared_from_this ()] () + { + s->SendSessionConfirmed (); + }); } } @@ -679,7 +706,7 @@ namespace transport // part 2 std::vector buf(m_Establisher->m3p2Len - 16); // -MAC memset (nonce, 0, 12); // set nonce to 0 again - if (m_Establisher->ProcessSessionConfirmedMessagePart2 (nonce, buf.data ())) + if (m_Establisher->ProcessSessionConfirmedMessagePart2 (nonce, buf.data ())) // TODO:handle in establisher thread { KeyDerivationFunctionDataPhase (); // Bob data phase keys @@ -811,7 +838,11 @@ namespace transport void NTCP2Session::ClientLogin () { m_Establisher->CreateEphemeralKey (); - SendSessionRequest (); + boost::asio::post (m_Server.GetEstablisherService (), + [s = shared_from_this ()] () + { + s->SendSessionRequest (); + }); } void NTCP2Session::ServerLogin () @@ -1367,6 +1398,7 @@ namespace transport void NTCP2Server::Start () { + m_EstablisherService.Start (); if (!IsRunning ()) { StartIOService (); @@ -1476,6 +1508,7 @@ namespace transport m_TerminationTimer.cancel (); m_ProxyEndpoint = nullptr; } + m_EstablisherService.Stop (); StopIOService (); } diff --git a/libi2pd/NTCP2.h b/libi2pd/NTCP2.h index e1718bb0..7f5c97aa 100644 --- a/libi2pd/NTCP2.h +++ b/libi2pd/NTCP2.h @@ -172,9 +172,11 @@ namespace transport void HandleSessionRequestSent (const boost::system::error_code& ecode, std::size_t bytes_transferred); void HandleSessionRequestReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred); + void ProcessSessionRequest (size_t len); void HandleSessionRequestPaddingReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred); void HandleSessionCreatedSent (const boost::system::error_code& ecode, std::size_t bytes_transferred); void HandleSessionCreatedReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred); + void ProcessSessionCreated (size_t len); void HandleSessionCreatedPaddingReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred); void HandleSessionConfirmedSent (const boost::system::error_code& ecode, std::size_t bytes_transferred); void HandleSessionConfirmedReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred); @@ -239,6 +241,18 @@ namespace transport class NTCP2Server: private i2p::util::RunnableServiceWithWork { + private: + + class EstablisherService: public i2p::util::RunnableServiceWithWork + { + public: + + EstablisherService (): RunnableServiceWithWork ("NTCP2e") {}; + auto& GetService () { return GetIOService (); }; + void Start () { StartIOService (); }; + void Stop () { StopIOService (); }; + }; + public: enum ProxyType @@ -247,13 +261,14 @@ namespace transport eSocksProxy, eHTTPProxy }; - + NTCP2Server (); ~NTCP2Server (); void Start (); void Stop (); auto& GetService () { return GetIOService (); }; + auto& GetEstablisherService () { return m_EstablisherService.GetService (); }; std::mt19937& GetRng () { return m_Rng; }; bool AddNTCP2Session (std::shared_ptr session, bool incoming = false); @@ -294,7 +309,8 @@ namespace transport std::unique_ptr m_ProxyEndpoint; std::shared_ptr m_Address4, m_Address6, m_YggdrasilAddress; std::mt19937 m_Rng; - + EstablisherService m_EstablisherService; + public: // for HTTP/I2PControl