Refuse dulicated incoming pending session from same IP

libi2pd/NTCP2.cpp

@@ -378,6 +378,13 @@ namespace transport
 		}
 	}
 
+	void NTCP2Session::Close () 
+	{
+		m_Socket.close (); 
+		SetTerminationTimeout (NTCP2_ESTABLISH_TIMEOUT);
+		m_LastActivityTimestamp = i2p::util::GetSecondsSinceEpoch (); 
+	}
+		
 	void NTCP2Session::TerminateByTimeout ()
 	{
 		SendTerminationAndTerminate (eNTCP2IdleTimeout);
@@ -1289,7 +1296,7 @@ namespace transport
 			for (auto& it: ntcpSessions)
 				it.second->Terminate ();
 			for (auto& it: m_PendingIncomingSessions)
-				it->Terminate ();
+				it.second->Terminate ();
 		}
 		m_NTCP2Sessions.clear ();
 
@@ -1305,7 +1312,7 @@ namespace transport
 	{
 		if (!session) return false;
 		if (incoming)
-			m_PendingIncomingSessions.remove (session);
+			m_PendingIncomingSessions.erase (session->GetRemoteEndpoint ().address ());
 		if (!session->GetRemoteIdentity ()) return false;
 		auto& ident = session->GetRemoteIdentity ()->GetIdentHash ();
 		auto it = m_NTCP2Sessions.find (ident);
@@ -1413,13 +1420,22 @@ namespace transport
 			if (!ec)
 			{
 				LogPrint (eLogDebug, "NTCP2: Connected from ", ep);
-				if (conn)
+				if (!i2p::util::net::IsInReservedRange(ep.address ()))
 				{    
-					conn->SetRemoteEndpoint (ep);
-					conn->ServerLogin ();
-					m_PendingIncomingSessions.push_back (conn);
-					conn = nullptr;
+					if (conn)
+					{
+						if (m_PendingIncomingSessions.emplace (ep.address (), conn).second)
+						{	
+							conn->SetRemoteEndpoint (ep);
+							conn->ServerLogin ();
+							conn = nullptr;
+						}	
+						else
+							LogPrint (eLogInfo, "NTCP2: Incoming session from ", ep.address (), " is already pending");
+					}
 				}	
+				else
+					LogPrint (eLogError, "NTCP2: Incoming connection from invalid IP ", ep.address ());
 			}
 			else
 				LogPrint (eLogError, "NTCP2: Connected from error ", ec.message ());
@@ -1454,12 +1470,22 @@ namespace transport
 			if (!ec)
 			{
 				LogPrint (eLogDebug, "NTCP2: Connected from ", ep);
-				if (conn)
+				if (!i2p::util::net::IsInReservedRange(ep.address ()))
 				{    
-					conn->SetRemoteEndpoint (ep);
-					conn->ServerLogin ();
-					m_PendingIncomingSessions.push_back (conn);
+					if (conn)
+					{
+						if (m_PendingIncomingSessions.emplace (ep.address (), conn).second)
+						{	
+							conn->SetRemoteEndpoint (ep);
+							conn->ServerLogin ();
+							conn = nullptr;
+						}	
+						else
+							LogPrint (eLogInfo, "NTCP2: Incoming session from ", ep.address (), " is already pending");
+					}
 				}	
+				else
+					LogPrint (eLogError, "NTCP2: Incoming connection from invalid IP ", ep.address ());
 			}
 			else
 				LogPrint (eLogError, "NTCP2: Connected from error ", ec.message ());
@@ -1476,7 +1502,10 @@ namespace transport
 
 		if (error != boost::asio::error::operation_aborted)
 		{
-			conn = std::make_shared<NTCP2Session> (*this);
+			if (!conn) // connection is used, create new one
+				conn = std::make_shared<NTCP2Session> (*this);
+			else // reuse failed
+				conn->Close ();
 			m_NTCP2V6Acceptor->async_accept(conn->GetSocket (), std::bind (&NTCP2Server::HandleAcceptV6, this,
 				conn, std::placeholders::_1));
 		}
@@ -1507,12 +1536,12 @@ namespace transport
 			// pending
 			for (auto it = m_PendingIncomingSessions.begin (); it != m_PendingIncomingSessions.end ();)
 			{
-				if ((*it)->IsEstablished () || (*it)->IsTerminationTimeoutExpired (ts))
+				if (it->second->IsEstablished () || it->second->IsTerminationTimeoutExpired (ts))
 				{
-					(*it)->Terminate ();
+					it->second->Terminate ();
 					it = m_PendingIncomingSessions.erase (it); // established of expired
 				}
-				else if ((*it)->IsTerminated ())
+				else if (it->second->IsTerminated ())
 					it = m_PendingIncomingSessions.erase (it); // already terminated
 				else
 					it++;

libi2pd/NTCP2.h

@@ -135,7 +135,7 @@ namespace transport
 			void Terminate ();
 			void TerminateByTimeout ();
 			void Done () override;
-			void Close () { m_Socket.close (); }; // for accept
+			void Close (); // for accept
 			void DeleteNextReceiveBuffer (uint64_t ts);
 
 			boost::asio::ip::tcp::socket& GetSocket () { return m_Socket; };
@@ -277,7 +277,7 @@ namespace transport
 			boost::asio::deadline_timer m_TerminationTimer;
 			std::unique_ptr<boost::asio::ip::tcp::acceptor> m_NTCP2Acceptor, m_NTCP2V6Acceptor;
 			std::map<i2p::data::IdentHash, std::shared_ptr<NTCP2Session> > m_NTCP2Sessions;
-			std::list<std::shared_ptr<NTCP2Session> > m_PendingIncomingSessions;
+			std::map<boost::asio::ip::address, std::shared_ptr<NTCP2Session> > m_PendingIncomingSessions;
 
 			ProxyType m_ProxyType;
 			std::string m_ProxyAddress, m_ProxyAuthorization;