From 5b93558bd0d7b46bf004a198f15282d217e5c7d8 Mon Sep 17 00:00:00 2001 From: weko Date: Thu, 25 Jan 2024 01:05:58 +0000 Subject: [PATCH] Add support for multiple udp server tunnels on one destionation --- libi2pd/Datagram.cpp | 68 ++++++++++++++++++++++++++++++-- libi2pd/Datagram.h | 26 ++++++------ libi2pd_client/ClientContext.cpp | 2 +- libi2pd_client/SAM.cpp | 53 ++++++++++++++----------- libi2pd_client/UDPTunnel.cpp | 46 ++++++++++++++------- libi2pd_client/UDPTunnel.h | 3 +- 6 files changed, 142 insertions(+), 56 deletions(-) diff --git a/libi2pd/Datagram.cpp b/libi2pd/Datagram.cpp index 64738ebe..c99db584 100644 --- a/libi2pd/Datagram.cpp +++ b/libi2pd/Datagram.cpp @@ -19,7 +19,7 @@ namespace i2p namespace datagram { DatagramDestination::DatagramDestination (std::shared_ptr owner, bool gzip): - m_Owner (owner), m_Receiver (nullptr), m_RawReceiver (nullptr), m_Gzip (gzip) + m_Owner (owner), m_DefaultReceiver (nullptr), m_DefaultRawReceiver (nullptr), m_Gzip (gzip) { if (m_Gzip) m_Deflator.reset (new i2p::data::GzipDeflator); @@ -119,19 +119,79 @@ namespace datagram void DatagramDestination::HandleRawDatagram (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len) { - if (m_RawReceiver) - m_RawReceiver (fromPort, toPort, buf, len); + auto r = FindRawReceiver(toPort); + + if (r) + r (fromPort, toPort, buf, len); else LogPrint (eLogWarning, "DatagramDestination: no receiver for raw datagram"); } + void DatagramDestination::SetReceiver (const Receiver& receiver, uint16_t port) + { + std::lock_guard lock(m_ReceiversMutex); + m_ReceiversByPorts[port] = receiver; + if (!m_DefaultReceiver) { + m_DefaultReceiver = receiver; + m_DefaultReceiverPort = port; + } + } + + void DatagramDestination::ResetReceiver (uint16_t port) + { + std::lock_guard lock(m_ReceiversMutex); + m_ReceiversByPorts.erase (port); + if (m_DefaultReceiverPort == port) { + m_DefaultReceiver = nullptr; + m_DefaultReceiverPort = 0; + } + } + + + void DatagramDestination::SetRawReceiver (const RawReceiver& receiver, uint16_t port) + { + std::lock_guard lock(m_RawReceiversMutex); + m_RawReceiversByPorts[port] = receiver; + if (!m_DefaultRawReceiver) { + m_DefaultRawReceiver = receiver; + m_DefaultRawReceiverPort = port; + } + }; + + void DatagramDestination::ResetRawReceiver (uint16_t port) + { + std::lock_guard lock(m_RawReceiversMutex); + m_RawReceiversByPorts.erase (port); + if (m_DefaultRawReceiverPort == port) { + m_DefaultRawReceiver = nullptr; + m_DefaultRawReceiverPort = 0; + } + } + + DatagramDestination::Receiver DatagramDestination::FindReceiver(uint16_t port) { std::lock_guard lock(m_ReceiversMutex); - Receiver r = m_Receiver; + Receiver r = nullptr; auto itr = m_ReceiversByPorts.find(port); if (itr != m_ReceiversByPorts.end()) r = itr->second; + else { + r = m_DefaultReceiver; + } + return r; + } + + DatagramDestination::RawReceiver DatagramDestination::FindRawReceiver(uint16_t port) + { + std::lock_guard lock(m_RawReceiversMutex); + RawReceiver r = nullptr; + auto itr = m_RawReceiversByPorts.find(port); + if (itr != m_RawReceiversByPorts.end()) + r = itr->second; + else { + r = m_DefaultRawReceiver; + } return r; } diff --git a/libi2pd/Datagram.h b/libi2pd/Datagram.h index e6d1f7b6..fd6af909 100644 --- a/libi2pd/Datagram.h +++ b/libi2pd/Datagram.h @@ -126,14 +126,12 @@ namespace datagram void HandleDataMessagePayload (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len, bool isRaw = false); - void SetReceiver (const Receiver& receiver) { m_Receiver = receiver; }; - void ResetReceiver () { m_Receiver = nullptr; }; - void SetReceiver (const Receiver& receiver, uint16_t port) { std::lock_guard lock(m_ReceiversMutex); m_ReceiversByPorts[port] = receiver; }; - void ResetReceiver (uint16_t port) { std::lock_guard lock(m_ReceiversMutex); m_ReceiversByPorts.erase (port); }; + void SetReceiver (const Receiver& receiver, uint16_t port); + void ResetReceiver (uint16_t port); - void SetRawReceiver (const RawReceiver& receiver) { m_RawReceiver = receiver; }; - void ResetRawReceiver () { m_RawReceiver = nullptr; }; + void SetRawReceiver (const RawReceiver& receiver, uint16_t port); + void ResetRawReceiver (uint16_t port); std::shared_ptr GetInfoForRemote(const i2p::data::IdentHash & remote); @@ -150,20 +148,26 @@ namespace datagram void HandleDatagram (uint16_t fromPort, uint16_t toPort, uint8_t *const& buf, size_t len); void HandleRawDatagram (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len); - /** find a receiver by port, if none by port is found try default receiver, otherwise returns nullptr */ Receiver FindReceiver(uint16_t port); + RawReceiver FindRawReceiver(uint16_t port); private: std::shared_ptr m_Owner; - Receiver m_Receiver; // default - RawReceiver m_RawReceiver; // default - bool m_Gzip; // gzip compression of data messages + std::mutex m_SessionsMutex; std::map m_Sessions; + + Receiver m_DefaultReceiver; + RawReceiver m_DefaultRawReceiver; + uint16_t m_DefaultReceiverPort; + uint16_t m_DefaultRawReceiverPort; std::mutex m_ReceiversMutex; - std::map m_ReceiversByPorts; + std::mutex m_RawReceiversMutex; + std::unordered_map m_ReceiversByPorts; + std::unordered_map m_RawReceiversByPorts; + bool m_Gzip; // gzip compression of data messages i2p::data::GzipInflator m_Inflator; std::unique_ptr m_Deflator; std::vector m_From, m_Signature; diff --git a/libi2pd_client/ClientContext.cpp b/libi2pd_client/ClientContext.cpp index d8c0bd2d..4efe252d 100644 --- a/libi2pd_client/ClientContext.cpp +++ b/libi2pd_client/ClientContext.cpp @@ -775,7 +775,7 @@ namespace client address = "127.0.0.1"; } auto localAddress = boost::asio::ip::address::from_string(address); - auto serverTunnel = std::make_shared(name, localDestination, localAddress, endpoint, port, gzip); + auto serverTunnel = std::make_shared(name, localDestination, localAddress, endpoint, inPort, gzip); if(!isUniqueLocal) { LogPrint(eLogInfo, "Clients: Disabling loopback address mapping"); diff --git a/libi2pd_client/SAM.cpp b/libi2pd_client/SAM.cpp index 1a3afb58..e4cbae8a 100644 --- a/libi2pd_client/SAM.cpp +++ b/libi2pd_client/SAM.cpp @@ -415,12 +415,17 @@ namespace client { session->UDPEndpoint = forward; auto dest = session->GetLocalDestination ()->CreateDatagramDestination (); + auto port = std::stoi(params[SAM_PARAM_PORT]); if (type == eSAMSessionTypeDatagram) dest->SetReceiver (std::bind (&SAMSocket::HandleI2PDatagramReceive, shared_from_this (), - std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4, std::placeholders::_5)); + std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4, std::placeholders::_5), + port + ); else // raw dest->SetRawReceiver (std::bind (&SAMSocket::HandleI2PRawDatagramReceive, shared_from_this (), - std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4)); + std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4), + port + ); } if (session->GetLocalDestination ()->IsReady ()) @@ -524,7 +529,7 @@ namespace client if (addr->IsIdentHash ()) { if (session->GetLocalDestination ()->GetIdentHash () != addr->identHash) - { + { auto leaseSet = session->GetLocalDestination ()->FindLeaseSet(addr->identHash); if (leaseSet) Connect(leaseSet, session); @@ -556,7 +561,7 @@ namespace client if (session) { if (session->GetLocalDestination ()->SupportsEncryptionType (remote->GetEncryptionType ())) - { + { m_SocketType = eSAMSocketTypeStream; m_Stream = session->GetLocalDestination ()->CreateStream (remote); if (m_Stream) @@ -570,7 +575,7 @@ namespace client SendMessageReply (SAM_STREAM_STATUS_INVALID_ID, strlen(SAM_STREAM_STATUS_INVALID_ID), true); } else - SendStreamCantReachPeer ("Incompatible crypto"); + SendStreamCantReachPeer ("Incompatible crypto"); } else SendMessageReply (SAM_STREAM_STATUS_INVALID_ID, strlen(SAM_STREAM_STATUS_INVALID_ID), true); @@ -583,7 +588,7 @@ namespace client else { LogPrint (eLogError, "SAM: Destination to connect not found"); - SendStreamCantReachPeer ("LeaseSet not found"); + SendStreamCantReachPeer ("LeaseSet not found"); } } @@ -612,27 +617,27 @@ namespace client session->GetLocalDestination ()->AcceptOnce (std::bind (&SAMSocket::HandleI2PAccept, shared_from_this (), std::placeholders::_1)); } else - { + { auto ts = i2p::util::GetSecondsSinceEpoch (); while (!session->acceptQueue.empty () && session->acceptQueue.front ().second + SAM_SESSION_MAX_ACCEPT_INTERVAL > ts) - { + { auto socket = session->acceptQueue.front ().first; session->acceptQueue.pop_front (); if (socket) m_Owner.GetService ().post (std::bind(&SAMSocket::TerminateClose, socket)); - } + } if (session->acceptQueue.size () < SAM_SESSION_MAX_ACCEPT_QUEUE_SIZE) { // already accepting, queue up SendMessageReply (SAM_STREAM_STATUS_OK, strlen(SAM_STREAM_STATUS_OK), false); session->acceptQueue.push_back (std::make_pair(shared_from_this(), ts)); - } - else - { + } + else + { LogPrint (eLogInfo, "SAM: Session ", m_ID, " accept queue is full ", session->acceptQueue.size ()); SendStreamI2PError ("Already accepting"); - } - } + } + } } else SendMessageReply (SAM_STREAM_STATUS_INVALID_ID, strlen(SAM_STREAM_STATUS_INVALID_ID), true); @@ -875,8 +880,8 @@ namespace client size_t len = snprintf (m_Buffer, SAM_SOCKET_BUFFER_SIZE, reply, msg.c_str()); #endif SendMessageReply (m_Buffer, len, true); - } - + } + void SAMSocket::SendSessionI2PError(const std::string & msg) { LogPrint (eLogError, "SAM: Session I2P error: ", msg); @@ -886,14 +891,14 @@ namespace client void SAMSocket::SendStreamI2PError(const std::string & msg) { LogPrint (eLogError, "SAM: Stream I2P error: ", msg); - SendReplyWithMessage (SAM_STREAM_STATUS_I2P_ERROR, msg); + SendReplyWithMessage (SAM_STREAM_STATUS_I2P_ERROR, msg); } void SAMSocket::SendStreamCantReachPeer(const std::string & msg) { - SendReplyWithMessage (SAM_STREAM_STATUS_CANT_REACH_PEER, msg); - } - + SendReplyWithMessage (SAM_STREAM_STATUS_CANT_REACH_PEER, msg); + } + void SAMSocket::HandleNamingLookupLeaseSetRequestComplete (std::shared_ptr leaseSet, std::string name) { if (leaseSet) @@ -1093,14 +1098,14 @@ namespace client // pending acceptors auto ts = i2p::util::GetSecondsSinceEpoch (); while (!session->acceptQueue.empty () && session->acceptQueue.front ().second + SAM_SESSION_MAX_ACCEPT_INTERVAL > ts) - { + { auto socket = session->acceptQueue.front ().first; session->acceptQueue.pop_front (); if (socket) m_Owner.GetService ().post (std::bind(&SAMSocket::TerminateClose, socket)); - } + } if (!session->acceptQueue.empty ()) - { + { auto socket = session->acceptQueue.front ().first; session->acceptQueue.pop_front (); if (socket && socket->GetSocketType () == eSAMSocketTypeAcceptor) @@ -1108,7 +1113,7 @@ namespace client socket->m_IsAccepting = true; session->GetLocalDestination ()->AcceptOnce (std::bind (&SAMSocket::HandleI2PAccept, socket, std::placeholders::_1)); } - } + } } if (!m_IsSilent) { diff --git a/libi2pd_client/UDPTunnel.cpp b/libi2pd_client/UDPTunnel.cpp index 0bf3202f..cd17bbf0 100644 --- a/libi2pd_client/UDPTunnel.cpp +++ b/libi2pd_client/UDPTunnel.cpp @@ -32,7 +32,7 @@ namespace client auto it = m_Sessions.find (GetSessionIndex (fromPort, toPort)); if (it != m_Sessions.end ()) m_LastSession = it->second; - else + else m_LastSession = nullptr; } if (m_LastSession) @@ -47,7 +47,7 @@ namespace client std::lock_guard lock(m_SessionsMutex); uint64_t now = i2p::util::GetMillisecondsSinceEpoch(); auto itr = m_Sessions.begin(); - while(itr != m_Sessions.end()) + while(itr != m_Sessions.end()) { if(now - itr->second->LastActivity >= delta ) itr = m_Sessions.erase(itr); @@ -88,10 +88,10 @@ namespace client { LogPrint(eLogWarning, "UDPServer: Session with from ", remotePort, " and to ", localPort, " ports already exists. But from differend address. Removed"); m_Sessions.erase (it); - } - } - } - + } + } + } + boost::asio::ip::address addr; /** create new udp session */ if(m_IsUniqueLocal && m_LocalAddress.is_loopback()) @@ -102,7 +102,7 @@ namespace client else addr = m_LocalAddress; - auto s = std::make_shared(boost::asio::ip::udp::endpoint(addr, 0), + auto s = std::make_shared(boost::asio::ip::udp::endpoint(addr, 0), m_LocalDest, m_RemoteEndpoint, ih, localPort, remotePort); std::lock_guard lock(m_SessionsMutex); m_Sessions.emplace (idx, s); @@ -163,9 +163,9 @@ namespace client } I2PUDPServerTunnel::I2PUDPServerTunnel (const std::string & name, std::shared_ptr localDestination, - const boost::asio::ip::address& localAddress, const boost::asio::ip::udp::endpoint& forwardTo, uint16_t port, bool gzip) : + const boost::asio::ip::address& localAddress, const boost::asio::ip::udp::endpoint& forwardTo, uint16_t inPort, bool gzip) : m_IsUniqueLocal (true), m_Name (name), m_LocalAddress (localAddress), - m_RemoteEndpoint (forwardTo), m_LocalDest (localDestination), m_Gzip (gzip) + m_RemoteEndpoint (forwardTo), m_LocalDest (localDestination), m_inPort(inPort), m_Gzip (gzip) { } @@ -179,14 +179,23 @@ namespace client m_LocalDest->Start (); auto dgram = m_LocalDest->CreateDatagramDestination (m_Gzip); - dgram->SetReceiver (std::bind (&I2PUDPServerTunnel::HandleRecvFromI2P, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4, std::placeholders::_5)); - dgram->SetRawReceiver (std::bind (&I2PUDPServerTunnel::HandleRecvFromI2PRaw, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4)); + dgram->SetReceiver ( + std::bind (&I2PUDPServerTunnel::HandleRecvFromI2P, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4, std::placeholders::_5), + m_inPort + ); + dgram->SetRawReceiver ( + std::bind (&I2PUDPServerTunnel::HandleRecvFromI2PRaw, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4), + m_inPort + ); } void I2PUDPServerTunnel::Stop () { auto dgram = m_LocalDest->GetDatagramDestination (); - if (dgram) dgram->ResetReceiver (); + if (dgram) { + dgram->ResetReceiver (m_inPort); + dgram->ResetRawReceiver (m_inPort); + } } std::vector > I2PUDPServerTunnel::GetSessions () @@ -240,9 +249,13 @@ namespace client dgram->SetReceiver (std::bind (&I2PUDPClientTunnel::HandleRecvFromI2P, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4, - std::placeholders::_5)); + std::placeholders::_5), + RemotePort + ); dgram->SetRawReceiver (std::bind (&I2PUDPClientTunnel::HandleRecvFromI2PRaw, this, - std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4)); + std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4), + RemotePort + ); m_LocalDest->Start (); if (m_ResolveThread == nullptr) @@ -253,7 +266,10 @@ namespace client void I2PUDPClientTunnel::Stop () { auto dgram = m_LocalDest->GetDatagramDestination (); - if (dgram) dgram->ResetReceiver (); + if (dgram) { + dgram->ResetReceiver (RemotePort); + dgram->ResetRawReceiver (RemotePort); + } m_cancel_resolve = true; m_Sessions.clear(); diff --git a/libi2pd_client/UDPTunnel.h b/libi2pd_client/UDPTunnel.h index 6dff41ea..5650124c 100644 --- a/libi2pd_client/UDPTunnel.h +++ b/libi2pd_client/UDPTunnel.h @@ -105,7 +105,7 @@ namespace client void HandleRecvFromI2PRaw (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len); UDPSessionPtr ObtainUDPSession (const i2p::data::IdentityEx& from, uint16_t localPort, uint16_t remotePort); uint32_t GetSessionIndex (uint16_t fromPort, uint16_t toPort) const { return ((uint32_t)fromPort << 16) + toPort; } - + private: bool m_IsUniqueLocal; @@ -116,6 +116,7 @@ namespace client std::unordered_map m_Sessions; // (from port, to port)->session std::shared_ptr m_LocalDest; UDPSessionPtr m_LastSession; + uint16_t m_inPort; bool m_Gzip; public: