try fixing udp tunnel (probably broken)

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<std::mutex> lock(m_SessionsMutex);
-    auto session = ObtainUDPSession(from, toPort, fromPort);
+    auto & session = ObtainUDPSession(from, toPort, fromPort);
-    session->IPSocket.send_to(boost::asio::buffer(buf, len), m_RemoteEndpoint);
+    session.IPSocket.send_to(boost::asio::buffer(buf, len), m_RemoteEndpoint);
-    session->LastActivity = i2p::util::GetMillisecondsSinceEpoch();
+    session.LastActivity = i2p::util::GetMillisecondsSinceEpoch();
     
   }
 
   void I2PUDPServerTunnel::ExpireStale(const uint64_t delta) {
     std::lock_guard<std::mutex> lock(m_SessionsMutex);
     uint64_t now = i2p::util::GetMillisecondsSinceEpoch();
-    std::remove_if(m_Sessions.begin(), m_Sessions.end(), [now, delta](const UDPSession * u) -> bool {
+    std::remove_if(m_Sessions.begin(), m_Sessions.end(), [now, delta](const UDPSession & u) -> bool {
-      return now - u->LastActivity >= delta;
+      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<std::mutex> lock(m_SessionsMutex);
+    uint64_t now = i2p::util::GetMillisecondsSinceEpoch();
+		std::vector<uint16_t> 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));
+    m_Sessions.push_back(UDPSession(ep, m_LocalDest, m_RemoteEndpoint, &ih, localPort, remotePort));
-    return m_Sessions.back();
+		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<i2p::client::ClientDestination> 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<std::shared_ptr<DatagramSessionInfo> > sessions;
 		std::lock_guard<std::mutex> lock(m_SessionsMutex);
-		for ( UDPSession * s : m_Sessions )
+		for ( UDPSession & s : m_Sessions )
 		{
-			if (!s->m_Destination) continue;
+			if (!s.m_Destination) continue;
-			auto info = s->m_Destination->GetInfoForRemote(s->Identity);
+			auto info = s.m_Destination->GetInfoForRemote(s.Identity);
 			if(!info) continue;
 
 			auto sinfo = std::make_shared<DatagramSessionInfo>();
 			sinfo->Name = m_Name;
 			sinfo->LocalIdent = std::make_shared<i2p::data::IdentHash>(m_LocalDest->GetIdentHash().data());
-			sinfo->RemoteIdent = std::make_shared<i2p::data::IdentHash>(s->Identity.data());
+			sinfo->RemoteIdent = std::make_shared<i2p::data::IdentHash>(s.Identity.data());
 			sinfo->CurrentIBGW = info->IBGW;
 			sinfo->CurrentOBEP = info->OBEP;
 			sessions.push_back(sinfo);
@@ -652,13 +664,12 @@ namespace client
 		std::shared_ptr<i2p::client::ClientDestination> 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();
 	}
 
-	std::vector<std::shared_ptr<DatagramSessionInfo> > I2PUDPClientTunnel::GetSessions()
+	void I2PUDPClientTunnel::RecvFromLocal()
 	{
-		std::vector<std::shared_ptr<DatagramSessionInfo> > infos;
+		m_LocalSocket.async_receive_from(boost::asio::buffer(m_RecvBuff, I2P_UDP_MAX_MTU),
-		if(m_Session && m_LocalDest)
+			m_RecvEndpoint, std::bind(&I2PUDPClientTunnel::HandleRecvFromLocal, this, std::placeholders::_1, std::placeholders::_2));
-		{
-			auto s = m_Session;
-			if (s->m_Destination)
-			{
-				auto info = m_Session->m_Destination->GetInfoForRemote(s->Identity);
-				if(info)
-				{
-					auto sinfo = std::make_shared<DatagramSessionInfo>();
-					sinfo->Name = m_Name;
-					sinfo->LocalIdent = std::make_shared<i2p::data::IdentHash>(m_LocalDest->GetIdentHash().data());
-					sinfo->RemoteIdent = std::make_shared<i2p::data::IdentHash>(s->Identity.data());
-					sinfo->CurrentIBGW = info->IBGW;
-					sinfo->CurrentOBEP = info->OBEP;
-					infos.push_back(sinfo);
 	}
+
+	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<std::shared_ptr<DatagramSessionInfo> > I2PUDPClientTunnel::GetSessions()
+	{
+		// TODO: implement
+		std::vector<std::shared_ptr<DatagramSessionInfo> > infos;
 		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
+			auto itr = m_Sessions.find(toPort);
-			if(m_Session)
+			// found convo ?
+			if(itr != m_Sessions.end())
 			{
-				// tell session
+				// found convo
+				if (len > 0) {
 					LogPrint(eLogDebug, "UDP Client: got ", len, "B from ", from.GetIdentHash().ToBase32());
-				m_Session->IPSocket.send_to(boost::asio::buffer(buf, len), m_Session->FromEndpoint);
+					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)

I2PTunnel.h

@@ -4,6 +4,7 @@
 #include <inttypes.h>
 #include <string>
 #include <set>
+#include <tuple>
 #include <memory>
 #include <sstream>
 #include <boost/asio.hpp>
@@ -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<i2p::client::ClientDestination> 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<UDPSession*> m_Sessions;
+			std::vector<UDPSession> m_Sessions;
 			std::shared_ptr<i2p::client::ClientDestination> m_LocalDest;
 	};
 
@@ -228,18 +227,25 @@ namespace client
 			bool IsLocalDestination(const i2p::data::IdentHash & destination) const { return destination == m_LocalDest->GetIdentHash(); }
 
 			std::shared_ptr<ClientDestination> GetLocalDestination () const { return m_LocalDest; }
+			void ExpireStale(const uint64_t delta=I2P_UDP_SESSION_TIMEOUT);
 
 		private:
+		typedef std::tuple<boost::asio::ip::udp::endpoint, uint64_t> 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<uint16_t, UDPConvo > m_Sessions; // maps i2p port -> local udp convo
 			const std::string m_RemoteDest;
 			std::shared_ptr<i2p::client::ClientDestination> 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;
 	};