initial implementation of STREAM FORWARD

libi2pd_client/SAM.cpp

@@ -54,6 +54,7 @@ namespace client
 				break;
 			}
 			case eSAMSocketTypeAcceptor:
+			case eSAMSocketTypeForward:	
 			{
 				if (Session)
 				{
@@ -263,6 +264,8 @@ namespace client
 						ProcessStreamConnect (separator + 1, bytes_transferred - (separator - m_Buffer) - 1, bytes_transferred - (eol - m_Buffer) - 1);
 					else if (!strcmp (m_Buffer, SAM_STREAM_ACCEPT))
 						ProcessStreamAccept (separator + 1, bytes_transferred - (separator - m_Buffer) - 1);
+					else if (!strcmp (m_Buffer, SAM_STREAM_FORWARD))
+						ProcessStreamForward (separator + 1, bytes_transferred - (separator - m_Buffer) - 1);
 					else if (!strcmp (m_Buffer, SAM_DEST_GENERATE))
 						ProcessDestGenerate (separator + 1, bytes_transferred - (separator - m_Buffer) - 1);
 					else if (!strcmp (m_Buffer, SAM_NAMING_LOOKUP))
@@ -358,12 +361,12 @@ namespace client
 
 		std::shared_ptr<boost::asio::ip::udp::endpoint> forward = nullptr;
 		if ((type == eSAMSessionTypeDatagram || type == eSAMSessionTypeRaw) &&
-			params.find(SAM_VALUE_HOST) != params.end() && params.find(SAM_VALUE_PORT) != params.end())
+			params.find(SAM_PARAM_HOST) != params.end() && params.find(SAM_PARAM_PORT) != params.end())
 		{
 			// udp forward selected
 			boost::system::error_code e;
 			// TODO: support hostnames in udp forward
-			auto addr = boost::asio::ip::address::from_string(params[SAM_VALUE_HOST], e);
+			auto addr = boost::asio::ip::address::from_string(params[SAM_PARAM_HOST], e);
 			if (e)
 			{
 				// not an ip address
@@ -371,7 +374,7 @@ namespace client
 				return;
 			}
 
-			auto port = std::stoi(params[SAM_VALUE_PORT]);
+			auto port = std::stoi(params[SAM_PARAM_PORT]);
 			if (port == -1)
 			{
 				SendI2PError("Invalid port");
@@ -565,6 +568,51 @@ namespace client
 			SendMessageReply (SAM_STREAM_STATUS_INVALID_ID, strlen(SAM_STREAM_STATUS_INVALID_ID), true);
 	}
 
+	void SAMSocket::ProcessStreamForward (char * buf, size_t len)
+	{
+		LogPrint (eLogDebug, "SAM: stream forward: ", buf);
+		std::map<std::string, std::string> params;
+		ExtractParams (buf, params);
+		std::string& id = params[SAM_PARAM_ID];
+		auto session = m_Owner.FindSession (id);
+		if (!session)
+		{
+			SendMessageReply (SAM_STREAM_STATUS_INVALID_ID, strlen(SAM_STREAM_STATUS_INVALID_ID), true);
+			return;
+		}	
+		if (session->localDestination->IsAcceptingStreams ())
+		{	
+			SendI2PError ("Already accepting");
+			return;
+		}	
+		auto it = params.find (SAM_PARAM_PORT);
+		if (it == params.end ())
+		{
+			SendI2PError ("PORT is missing");
+			return;
+		}	
+		auto port = std::stoi (it->second);
+		if (port <= 0 || port >= 0xFFFF)
+		{
+			SendI2PError ("Invalid PORT");
+			return;
+		}	
+		boost::system::error_code ec;
+		auto ep = m_Socket.remote_endpoint (ec);
+		if (ec)
+		{
+			SendI2PError ("Socket error");
+			return;
+		}	
+		ep.port (port);
+		m_SocketType = eSAMSocketTypeForward;
+		m_ID = id;
+		m_IsAccepting = true;
+		session->localDestination->AcceptStreams (std::bind (&SAMSocket::HandleI2PForward, 
+			shared_from_this (), std::placeholders::_1, ep));
+		SendMessageReply (SAM_STREAM_STATUS_OK, strlen(SAM_STREAM_STATUS_OK), false);			
+	}	
+		
 	size_t SAMSocket::ProcessDatagramSend (char * buf, size_t len, const char * data)
 	{
 		LogPrint (eLogDebug, "SAM: datagram send: ", buf, " ", len);
@@ -917,6 +965,33 @@ namespace client
 			LogPrint (eLogWarning, "SAM: I2P acceptor has been reset");
 	}
 
+	void SAMSocket::HandleI2PForward (std::shared_ptr<i2p::stream::Stream> stream, 
+		boost::asio::ip::tcp::endpoint ep)
+	{
+		if (stream)
+		{
+			LogPrint (eLogDebug, "SAM: incoming forward I2P connection for session ", m_ID);
+			auto newSocket = std::make_shared<SAMSocket>(m_Owner);
+			newSocket->SetSocketType (eSAMSocketTypeStream);
+			auto s = shared_from_this ();
+			newSocket->GetSocket ().async_connect (ep, 
+				[s, newSocket, stream](const boost::system::error_code& ecode)
+			    {
+					if (!ecode)
+					{
+						s->m_Owner.AddSocket (newSocket);
+						newSocket->Receive ();
+						newSocket->m_Stream = stream;
+						newSocket->I2PReceive ();
+					}
+					else
+						stream->AsyncClose ();
+				});			
+		}
+		else
+			LogPrint (eLogWarning, "SAM: I2P forward acceptor has been reset");
+	}	
+		
 	void SAMSocket::HandleI2PDatagramReceive (const i2p::data::IdentityEx& from, uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len)
 	{
 		LogPrint (eLogDebug, "SAM: datagram received ", len);
@@ -1072,6 +1147,12 @@ namespace client
 			std::placeholders::_1, newSocket));
 	}
 
+	void SAMBridge::AddSocket(std::shared_ptr<SAMSocket> socket)
+	{
+		std::unique_lock<std::mutex> lock(m_OpenSocketsMutex);
+		m_OpenSockets.push_back(socket);
+	}		
+		
 	void SAMBridge::RemoveSocket(const std::shared_ptr<SAMSocket> & socket)
 	{
 		std::unique_lock<std::mutex> lock(m_OpenSocketsMutex);
@@ -1087,10 +1168,7 @@ namespace client
 			if (!ec)
 			{
 				LogPrint (eLogDebug, "SAM: new connection from ", ep);
-				{
+				AddSocket (socket);
-					std::unique_lock<std::mutex> l(m_OpenSocketsMutex);
-					m_OpenSockets.push_back(socket);
-				}
 				socket->ReceiveHandshake ();
 			}
 			else

libi2pd_client/SAM.h

@@ -48,6 +48,7 @@ namespace client
 	const char SAM_STREAM_STATUS_CANT_REACH_PEER[] = "STREAM STATUS RESULT=CANT_REACH_PEER\n";
 	const char SAM_STREAM_STATUS_I2P_ERROR[] = "STREAM STATUS RESULT=I2P_ERROR\n";
 	const char SAM_STREAM_ACCEPT[] = "STREAM ACCEPT";
+	const char SAM_STREAM_FORWARD[] = "STREAM FORWARD";
 	const char SAM_DATAGRAM_SEND[] = "DATAGRAM SEND";
 	const char SAM_RAW_SEND[] = "RAW SEND";
 	const char SAM_DEST_GENERATE[] = "DEST GENERATE";
@@ -69,14 +70,14 @@ namespace client
 	const char SAM_PARAM_SIGNATURE_TYPE[] = "SIGNATURE_TYPE";
 	const char SAM_PARAM_CRYPTO_TYPE[] = "CRYPTO_TYPE";
 	const char SAM_PARAM_SIZE[] = "SIZE";
+	const char SAM_PARAM_HOST[] = "HOST";
+	const char SAM_PARAM_PORT[] = "PORT";
 	const char SAM_VALUE_TRANSIENT[] = "TRANSIENT";
 	const char SAM_VALUE_STREAM[] = "STREAM";
 	const char SAM_VALUE_DATAGRAM[] = "DATAGRAM";
 	const char SAM_VALUE_RAW[] = "RAW";
 	const char SAM_VALUE_TRUE[] = "true";
 	const char SAM_VALUE_FALSE[] = "false";
-	const char SAM_VALUE_HOST[] = "HOST";
-	const char SAM_VALUE_PORT[] = "PORT";
 
 	enum SAMSocketType
 	{
@@ -84,6 +85,7 @@ namespace client
 		eSAMSocketTypeSession,
 		eSAMSocketTypeStream,
 		eSAMSocketTypeAcceptor,
+		eSAMSocketTypeForward,
 		eSAMSocketTypeTerminated
 	};
 
@@ -121,6 +123,7 @@ namespace client
 			void I2PReceive ();
 			void HandleI2PReceive (const boost::system::error_code& ecode, std::size_t bytes_transferred);
 			void HandleI2PAccept (std::shared_ptr<i2p::stream::Stream> stream);
+			void HandleI2PForward (std::shared_ptr<i2p::stream::Stream> stream, boost::asio::ip::tcp::endpoint ep);
 			void HandleWriteI2PData (const boost::system::error_code& ecode, size_t sz);
 			void HandleI2PDatagramReceive (const i2p::data::IdentityEx& from, uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len);
 			void HandleI2PRawDatagramReceive (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len);
@@ -128,6 +131,7 @@ namespace client
 			void ProcessSessionCreate (char * buf, size_t len);
 			void ProcessStreamConnect (char * buf, size_t len, size_t rem);
 			void ProcessStreamAccept (char * buf, size_t len);
+			void ProcessStreamForward (char * buf, size_t len);
 			void ProcessDestGenerate (char * buf, size_t len);
 			void ProcessNamingLookup (char * buf, size_t len);
 			void SendI2PError(const std::string & msg);
@@ -205,6 +209,7 @@ namespace client
 			/** send raw data to remote endpoint from our UDP Socket */
 			void SendTo(const uint8_t * buf, size_t len, std::shared_ptr<boost::asio::ip::udp::endpoint> remote);
 
+			void AddSocket(std::shared_ptr<SAMSocket> socket);
 			void RemoveSocket(const std::shared_ptr<SAMSocket> & socket);
 
 			bool ResolveSignatureType (const std::string& name, i2p::data::SigningKeyType& type) const;