From e270f90f8d808deb994876c257448bc32a3673c0 Mon Sep 17 00:00:00 2001 From: Jeff Becker Date: Thu, 17 Nov 2016 10:36:27 -0500 Subject: [PATCH] try fixing udp tunnel (probably broken) --- I2PTunnel.cpp | 124 +++++++++++++++++++++++++++++--------------------- I2PTunnel.h | 20 +++++--- 2 files changed, 86 insertions(+), 58 deletions(-) diff --git a/I2PTunnel.cpp b/I2PTunnel.cpp index 93bcff0c..01a3e7ca 100644 --- a/I2PTunnel.cpp +++ b/I2PTunnel.cpp @@ -531,36 +531,50 @@ namespace client void I2PUDPServerTunnel::HandleRecvFromI2P(const i2p::data::IdentityEx& from, uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len) { std::lock_guard lock(m_SessionsMutex); - auto session = ObtainUDPSession(from, toPort, fromPort); - session->IPSocket.send_to(boost::asio::buffer(buf, len), m_RemoteEndpoint); - session->LastActivity = i2p::util::GetMillisecondsSinceEpoch(); + auto & session = ObtainUDPSession(from, toPort, fromPort); + session.IPSocket.send_to(boost::asio::buffer(buf, len), m_RemoteEndpoint); + session.LastActivity = i2p::util::GetMillisecondsSinceEpoch(); } void I2PUDPServerTunnel::ExpireStale(const uint64_t delta) { std::lock_guard lock(m_SessionsMutex); uint64_t now = i2p::util::GetMillisecondsSinceEpoch(); - std::remove_if(m_Sessions.begin(), m_Sessions.end(), [now, delta](const UDPSession * u) -> bool { - return now - u->LastActivity >= delta; + std::remove_if(m_Sessions.begin(), m_Sessions.end(), [now, delta](const UDPSession & u) -> bool { + return now - u.LastActivity >= delta; }); } - - UDPSession * I2PUDPServerTunnel::ObtainUDPSession(const i2p::data::IdentityEx& from, uint16_t localPort, uint16_t remotePort) + + void I2PUDPClientTunnel::ExpireStale(const uint64_t delta) { + std::lock_guard lock(m_SessionsMutex); + uint64_t now = i2p::util::GetMillisecondsSinceEpoch(); + std::vector removePorts; + for (const auto & s : m_Sessions) { + if (now - std::get<1>(s.second) >= delta) + removePorts.push_back(s.first); + } + for(auto port : removePorts) { + m_Sessions.erase(port); + } + } + + UDPSession & I2PUDPServerTunnel::ObtainUDPSession(const i2p::data::IdentityEx& from, uint16_t localPort, uint16_t remotePort) { auto ih = from.GetIdentHash(); - for ( UDPSession * s : m_Sessions ) + for ( UDPSession & s : m_Sessions ) { - if ( s->Identity == ih) + if ( s.Identity == ih) { /** found existing session */ - LogPrint(eLogDebug, "UDPServer: found session ", s->IPSocket.local_endpoint(), " ", ih.ToBase32()); + LogPrint(eLogDebug, "UDPServer: found session ", s.IPSocket.local_endpoint(), " ", ih.ToBase32()); return s; } } /** create new udp session */ boost::asio::ip::udp::endpoint ep(m_LocalAddress, 0); - m_Sessions.push_back(new UDPSession(ep, m_LocalDest, m_RemoteEndpoint, &ih, localPort, remotePort)); - return m_Sessions.back(); + m_Sessions.push_back(UDPSession(ep, m_LocalDest, m_RemoteEndpoint, &ih, localPort, remotePort)); + auto & back = m_Sessions.back(); + return back; } UDPSession::UDPSession(boost::asio::ip::udp::endpoint localEndpoint, @@ -568,7 +582,6 @@ namespace client boost::asio::ip::udp::endpoint endpoint, const i2p::data::IdentHash * to, uint16_t ourPort, uint16_t theirPort) : m_Destination(localDestination->GetDatagramDestination()), - m_Service(localDestination->GetService()), IPSocket(localDestination->GetService(), localEndpoint), SendEndpoint(endpoint), LastActivity(i2p::util::GetMillisecondsSinceEpoch()), @@ -602,9 +615,8 @@ namespace client I2PUDPServerTunnel::I2PUDPServerTunnel(const std::string & name, std::shared_ptr localDestination, - const boost::asio::ip::address& localAddress, boost::asio::ip::udp::endpoint forwardTo, uint16_t port) : + boost::asio::ip::address localAddress, boost::asio::ip::udp::endpoint forwardTo, uint16_t port) : m_Name(name), - LocalPort(port), m_LocalAddress(localAddress), m_RemoteEndpoint(forwardTo) { @@ -630,16 +642,16 @@ namespace client { std::vector > sessions; std::lock_guard lock(m_SessionsMutex); - for ( UDPSession * s : m_Sessions ) + for ( UDPSession & s : m_Sessions ) { - if (!s->m_Destination) continue; - auto info = s->m_Destination->GetInfoForRemote(s->Identity); + if (!s.m_Destination) continue; + auto info = s.m_Destination->GetInfoForRemote(s.Identity); if(!info) continue; auto sinfo = std::make_shared(); sinfo->Name = m_Name; sinfo->LocalIdent = std::make_shared(m_LocalDest->GetIdentHash().data()); - sinfo->RemoteIdent = std::make_shared(s->Identity.data()); + sinfo->RemoteIdent = std::make_shared(s.Identity.data()); sinfo->CurrentIBGW = info->IBGW; sinfo->CurrentOBEP = info->OBEP; sessions.push_back(sinfo); @@ -652,13 +664,12 @@ namespace client std::shared_ptr localDestination, uint16_t remotePort) : m_Name(name), - m_Session(nullptr), m_RemoteDest(remoteDest), m_LocalDest(localDestination), m_LocalEndpoint(localEndpoint), m_RemoteIdent(nullptr), m_ResolveThread(nullptr), - LocalPort(localEndpoint.port()), + m_LocalSocket(localDestination->GetService(), localEndpoint), RemotePort(remotePort), m_cancel_resolve(false) { @@ -675,29 +686,34 @@ namespace client m_LocalDest->Start(); if (m_ResolveThread == nullptr) m_ResolveThread = new std::thread(std::bind(&I2PUDPClientTunnel::TryResolving, this)); + RecvFromLocal(); } + void I2PUDPClientTunnel::RecvFromLocal() + { + m_LocalSocket.async_receive_from(boost::asio::buffer(m_RecvBuff, I2P_UDP_MAX_MTU), + m_RecvEndpoint, std::bind(&I2PUDPClientTunnel::HandleRecvFromLocal, this, std::placeholders::_1, std::placeholders::_2)); + } + + void I2PUDPClientTunnel::HandleRecvFromLocal(const boost::system::error_code & ec, std::size_t transferred) + { + if(!m_RemoteIdent) return; // drop, remote not resolved + auto remotePort = m_RecvEndpoint.port(); + auto itr = m_Sessions.find(remotePort); + if (itr == m_Sessions.end()) { + // track new udp convo + m_Sessions[remotePort] = {boost::asio::ip::udp::endpoint(m_RecvEndpoint), 0}; + } + // send off to remote i2p destination + m_LocalDest->GetDatagramDestination()->SendDatagramTo(m_RecvBuff, transferred, *m_RemoteIdent, remotePort, RemotePort); + // mark convo as active + std::get<1>(m_Sessions[remotePort]) = i2p::util::GetMillisecondsSinceEpoch(); + } + std::vector > I2PUDPClientTunnel::GetSessions() { + // TODO: implement std::vector > infos; - if(m_Session && m_LocalDest) - { - auto s = m_Session; - if (s->m_Destination) - { - auto info = m_Session->m_Destination->GetInfoForRemote(s->Identity); - if(info) - { - auto sinfo = std::make_shared(); - sinfo->Name = m_Name; - sinfo->LocalIdent = std::make_shared(m_LocalDest->GetIdentHash().data()); - sinfo->RemoteIdent = std::make_shared(s->Identity.data()); - sinfo->CurrentIBGW = info->IBGW; - sinfo->CurrentOBEP = info->OBEP; - infos.push_back(sinfo); - } - } - } return infos; } @@ -717,26 +733,28 @@ namespace client return; } LogPrint(eLogInfo, "UDP Tunnel: resolved ", m_RemoteDest, " to ", m_RemoteIdent->ToBase32()); - // delete existing session - if(m_Session) delete m_Session; - - boost::asio::ip::udp::endpoint ep(boost::asio::ip::address::from_string("127.0.0.1"), 0); - m_Session = new UDPSession(m_LocalEndpoint, m_LocalDest, ep, m_RemoteIdent, LocalPort, RemotePort); } void I2PUDPClientTunnel::HandleRecvFromI2P(const i2p::data::IdentityEx& from, uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len) { if(m_RemoteIdent && from.GetIdentHash() == *m_RemoteIdent) { - // address match - if(m_Session) + auto itr = m_Sessions.find(toPort); + // found convo ? + if(itr != m_Sessions.end()) { - // tell session - LogPrint(eLogDebug, "UDP Client: got ", len, "B from ", from.GetIdentHash().ToBase32()); - m_Session->IPSocket.send_to(boost::asio::buffer(buf, len), m_Session->FromEndpoint); + // found convo + if (len > 0) { + LogPrint(eLogDebug, "UDP Client: got ", len, "B from ", from.GetIdentHash().ToBase32()); + uint8_t sendbuf[len]; + memcpy(sendbuf, buf, len); + m_LocalSocket.send_to(boost::asio::buffer(buf, len), std::get<0>(itr->second)); + // mark convo as active + std::get<1>(itr->second) = i2p::util::GetMillisecondsSinceEpoch(); + } } else - LogPrint(eLogWarning, "UDP Client: no session"); + LogPrint(eLogWarning, "UDP Client: not tracking udp session using port ", (int) toPort); } else LogPrint(eLogWarning, "UDP Client: unwarrented traffic from ", from.GetIdentHash().ToBase32()); @@ -747,7 +765,11 @@ namespace client auto dgram = m_LocalDest->GetDatagramDestination(); if (dgram) dgram->ResetReceiver(); - if (m_Session) delete m_Session; + m_Sessions.clear(); + + if(m_LocalSocket.is_open()) + m_LocalSocket.close(); + m_cancel_resolve = true; if(m_ResolveThread) diff --git a/I2PTunnel.h b/I2PTunnel.h index e6f0e84f..8e2b8cb4 100644 --- a/I2PTunnel.h +++ b/I2PTunnel.h @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -141,7 +142,6 @@ namespace client struct UDPSession { i2p::datagram::DatagramDestination * m_Destination; - boost::asio::io_service & m_Service; boost::asio::ip::udp::socket IPSocket; i2p::data::IdentHash Identity; boost::asio::ip::udp::endpoint FromEndpoint; @@ -189,7 +189,7 @@ namespace client public: I2PUDPServerTunnel(const std::string & name, std::shared_ptr localDestination, - const boost::asio::ip::address & localAddress, + boost::asio::ip::address localAddress, boost::asio::ip::udp::endpoint forwardTo, uint16_t port); ~I2PUDPServerTunnel(); /** expire stale udp conversations */ @@ -202,15 +202,14 @@ namespace client private: void HandleRecvFromI2P(const i2p::data::IdentityEx& from, uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len); - UDPSession * ObtainUDPSession(const i2p::data::IdentityEx& from, uint16_t localPort, uint16_t remotePort); + UDPSession & ObtainUDPSession(const i2p::data::IdentityEx& from, uint16_t localPort, uint16_t remotePort); private: const std::string m_Name; - const uint16_t LocalPort; boost::asio::ip::address m_LocalAddress; boost::asio::ip::udp::endpoint m_RemoteEndpoint; std::mutex m_SessionsMutex; - std::vector m_Sessions; + std::vector m_Sessions; std::shared_ptr m_LocalDest; }; @@ -228,18 +227,25 @@ namespace client bool IsLocalDestination(const i2p::data::IdentHash & destination) const { return destination == m_LocalDest->GetIdentHash(); } std::shared_ptr GetLocalDestination () const { return m_LocalDest; } + void ExpireStale(const uint64_t delta=I2P_UDP_SESSION_TIMEOUT); private: + typedef std::tuple UDPConvo; + void RecvFromLocal(); + void HandleRecvFromLocal(const boost::system::error_code & e, std::size_t transferred); void HandleRecvFromI2P(const i2p::data::IdentityEx& from, uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len); void TryResolving(); const std::string m_Name; - UDPSession * m_Session; + std::mutex m_SessionsMutex; + std::map m_Sessions; // maps i2p port -> local udp convo const std::string m_RemoteDest; std::shared_ptr m_LocalDest; const boost::asio::ip::udp::endpoint m_LocalEndpoint; i2p::data::IdentHash * m_RemoteIdent; std::thread * m_ResolveThread; - uint16_t LocalPort; + boost::asio::ip::udp::socket m_LocalSocket; + boost::asio::ip::udp::endpoint m_RecvEndpoint; + uint8_t m_RecvBuff[I2P_UDP_MAX_MTU]; uint16_t RemotePort; bool m_cancel_resolve; };