From b35f43d79e48116e18931c994a416497e3107924 Mon Sep 17 00:00:00 2001 From: orignal Date: Sun, 25 Oct 2020 17:20:15 -0400 Subject: [PATCH] initial implementation of STREAM FORWARD --- libi2pd_client/SAM.cpp | 92 ++++++++++++++++++++++++++++++++++++++---- libi2pd_client/SAM.h | 9 ++++- 2 files changed, 92 insertions(+), 9 deletions(-) diff --git a/libi2pd_client/SAM.cpp b/libi2pd_client/SAM.cpp index 63b9c7ed..2bdbf1ee 100644 --- a/libi2pd_client/SAM.cpp +++ b/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 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 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 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(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 socket) + { + std::unique_lock lock(m_OpenSocketsMutex); + m_OpenSockets.push_back(socket); + } + void SAMBridge::RemoveSocket(const std::shared_ptr & socket) { std::unique_lock lock(m_OpenSocketsMutex); @@ -1087,10 +1168,7 @@ namespace client if (!ec) { LogPrint (eLogDebug, "SAM: new connection from ", ep); - { - std::unique_lock l(m_OpenSocketsMutex); - m_OpenSockets.push_back(socket); - } + AddSocket (socket); socket->ReceiveHandshake (); } else diff --git a/libi2pd_client/SAM.h b/libi2pd_client/SAM.h index 7b1702f5..fafd7d1c 100644 --- a/libi2pd_client/SAM.h +++ b/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 stream); + void HandleI2PForward (std::shared_ptr 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 remote); + void AddSocket(std::shared_ptr socket); void RemoveSocket(const std::shared_ptr & socket); bool ResolveSignatureType (const std::string& name, i2p::data::SigningKeyType& type) const;