#include <string.h> #include <stdio.h> #ifdef _MSC_VER #include <stdlib.h> #endif #include "Base.h" #include "Identity.h" #include "Log.h" #include "Destination.h" #include "ClientContext.h" #include "util.h" #include "SAM.h" namespace i2p { namespace client { SAMSocket::SAMSocket (SAMBridge& owner): m_Owner (owner), m_Socket (m_Owner.GetService ()), m_Timer (m_Owner.GetService ()), m_BufferOffset (0), m_SocketType (eSAMSocketTypeUnknown), m_IsSilent (false), m_Stream (nullptr), m_Session (nullptr) { } SAMSocket::~SAMSocket () { Terminate (); } void SAMSocket::CloseStream () { if (m_Stream) { m_Stream->Close (); m_Stream.reset (); } } void SAMSocket::Terminate () { CloseStream (); switch (m_SocketType) { case eSAMSocketTypeSession: m_Owner.CloseSession (m_ID); break; case eSAMSocketTypeStream: { if (m_Session) m_Session->DelSocket (shared_from_this ()); break; } case eSAMSocketTypeAcceptor: { if (m_Session) m_Session->DelSocket (shared_from_this ()); break; } default: ; } m_SocketType = eSAMSocketTypeTerminated; if (m_Socket.is_open()) m_Socket.close (); m_Session = nullptr; } void SAMSocket::ReceiveHandshake () { m_Socket.async_read_some (boost::asio::buffer(m_Buffer, SAM_SOCKET_BUFFER_SIZE), std::bind(&SAMSocket::HandleHandshakeReceived, shared_from_this (), std::placeholders::_1, std::placeholders::_2)); } void SAMSocket::HandleHandshakeReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred) { if (ecode) { LogPrint (eLogError, "SAM: handshake read error: ", ecode.message ()); if (ecode != boost::asio::error::operation_aborted) Terminate (); } else { m_Buffer[bytes_transferred] = 0; char * eol = (char *)memchr (m_Buffer, '\n', bytes_transferred); if (eol) *eol = 0; LogPrint (eLogDebug, "SAM: handshake ", m_Buffer); char * separator = strchr (m_Buffer, ' '); if (separator) { separator = strchr (separator + 1, ' '); if (separator) *separator = 0; } if (!strcmp (m_Buffer, SAM_HANDSHAKE)) { std::string version("3.0"); // try to find MIN and MAX, 3.0 if not found if (separator) { separator++; std::map<std::string, std::string> params; ExtractParams (separator, params); auto it = params.find (SAM_PARAM_MAX); // TODO: check MIN as well if (it != params.end ()) version = it->second; } if (version[0] == '3') // we support v3 (3.0 and 3.1) only { #ifdef _MSC_VER size_t l = sprintf_s (m_Buffer, SAM_SOCKET_BUFFER_SIZE, SAM_HANDSHAKE_REPLY, version.c_str ()); #else size_t l = snprintf (m_Buffer, SAM_SOCKET_BUFFER_SIZE, SAM_HANDSHAKE_REPLY, version.c_str ()); #endif boost::asio::async_write (m_Socket, boost::asio::buffer (m_Buffer, l), boost::asio::transfer_all (), std::bind(&SAMSocket::HandleHandshakeReplySent, shared_from_this (), std::placeholders::_1, std::placeholders::_2)); } else SendMessageReply (SAM_HANDSHAKE_I2P_ERROR, strlen (SAM_HANDSHAKE_I2P_ERROR), true); } else { LogPrint (eLogError, "SAM: handshake mismatch"); Terminate (); } } } void SAMSocket::HandleHandshakeReplySent (const boost::system::error_code& ecode, std::size_t bytes_transferred) { if (ecode) { LogPrint (eLogError, "SAM: handshake reply send error: ", ecode.message ()); if (ecode != boost::asio::error::operation_aborted) Terminate (); } else { m_Socket.async_read_some (boost::asio::buffer(m_Buffer, SAM_SOCKET_BUFFER_SIZE), std::bind(&SAMSocket::HandleMessage, shared_from_this (), std::placeholders::_1, std::placeholders::_2)); } } void SAMSocket::SendMessageReply (const char * msg, size_t len, bool close) { if (!m_IsSilent) boost::asio::async_write (m_Socket, boost::asio::buffer (msg, len), boost::asio::transfer_all (), std::bind(&SAMSocket::HandleMessageReplySent, shared_from_this (), std::placeholders::_1, std::placeholders::_2, close)); else { if (close) Terminate (); else Receive (); } } void SAMSocket::HandleMessageReplySent (const boost::system::error_code& ecode, std::size_t bytes_transferred, bool close) { if (ecode) { LogPrint (eLogError, "SAM: reply send error: ", ecode.message ()); if (ecode != boost::asio::error::operation_aborted) Terminate (); } else { if (close) Terminate (); else Receive (); } } void SAMSocket::HandleMessage (const boost::system::error_code& ecode, std::size_t bytes_transferred) { if (ecode) { LogPrint (eLogError, "SAM: read error: ", ecode.message ()); if (ecode != boost::asio::error::operation_aborted) Terminate (); } else if (m_SocketType == eSAMSocketTypeStream) HandleReceived (ecode, bytes_transferred); else { bytes_transferred += m_BufferOffset; m_BufferOffset = 0; m_Buffer[bytes_transferred] = 0; char * eol = (char *)memchr (m_Buffer, '\n', bytes_transferred); if (eol) { *eol = 0; char * separator = strchr (m_Buffer, ' '); if (separator) { separator = strchr (separator + 1, ' '); if (separator) *separator = 0; else separator = eol; if (!strcmp (m_Buffer, SAM_SESSION_CREATE)) ProcessSessionCreate (separator + 1, bytes_transferred - (separator - m_Buffer) - 1); else if (!strcmp (m_Buffer, SAM_STREAM_CONNECT)) ProcessStreamConnect (separator + 1, bytes_transferred - (separator - 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_DEST_GENERATE)) ProcessDestGenerate (); else if (!strcmp (m_Buffer, SAM_NAMING_LOOKUP)) ProcessNamingLookup (separator + 1, bytes_transferred - (separator - m_Buffer) - 1); else if (!strcmp (m_Buffer, SAM_DATAGRAM_SEND)) { size_t len = bytes_transferred - (separator - m_Buffer) - 1; size_t processed = ProcessDatagramSend (separator + 1, len, eol + 1); if (processed < len) { m_BufferOffset = len - processed; if (processed > 0) memmove (m_Buffer, separator + 1 + processed, m_BufferOffset); else { // restore string back *separator = ' '; *eol = '\n'; } } // since it's SAM v1 reply is not expected Receive (); } else { LogPrint (eLogError, "SAM: unexpected message ", m_Buffer); Terminate (); } } else { LogPrint (eLogError, "SAM: malformed message ", m_Buffer); Terminate (); } } else { LogPrint (eLogWarning, "SAM: incomplete message ", bytes_transferred); m_BufferOffset = bytes_transferred; // try to receive remaining message Receive (); } } } void SAMSocket::ProcessSessionCreate (char * buf, size_t len) { LogPrint (eLogDebug, "SAM: session create: ", buf); std::map<std::string, std::string> params; ExtractParams (buf, params); std::string& style = params[SAM_PARAM_STYLE]; std::string& id = params[SAM_PARAM_ID]; std::string& destination = params[SAM_PARAM_DESTINATION]; m_ID = id; if (m_Owner.FindSession (id)) { // session exists SendMessageReply (SAM_SESSION_CREATE_DUPLICATED_ID, strlen(SAM_SESSION_CREATE_DUPLICATED_ID), true); return; } // create destination m_Session = m_Owner.CreateSession (id, destination == SAM_VALUE_TRANSIENT ? "" : destination, ¶ms); if (m_Session) { m_SocketType = eSAMSocketTypeSession; if (style == SAM_VALUE_DATAGRAM) { auto dest = m_Session->localDestination->CreateDatagramDestination (); dest->SetReceiver (std::bind (&SAMSocket::HandleI2PDatagramReceive, shared_from_this (), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4, std::placeholders::_5)); } else { // start accepting streams because we're not a datagram session m_Session->localDestination->AcceptStreams (std::bind (&SAMSession::AcceptI2P, m_Session, std::placeholders::_1)); } if (m_Session->localDestination->IsReady ()) SendSessionCreateReplyOk (); else { m_Timer.expires_from_now (boost::posix_time::seconds(SAM_SESSION_READINESS_CHECK_INTERVAL)); m_Timer.async_wait (std::bind (&SAMSocket::HandleSessionReadinessCheckTimer, shared_from_this (), std::placeholders::_1)); } } else SendMessageReply (SAM_SESSION_CREATE_DUPLICATED_DEST, strlen(SAM_SESSION_CREATE_DUPLICATED_DEST), true); } void SAMSocket::HandleSessionReadinessCheckTimer (const boost::system::error_code& ecode) { if (ecode != boost::asio::error::operation_aborted) { if (m_Session->localDestination->IsReady ()) SendSessionCreateReplyOk (); else { m_Timer.expires_from_now (boost::posix_time::seconds(SAM_SESSION_READINESS_CHECK_INTERVAL)); m_Timer.async_wait (std::bind (&SAMSocket::HandleSessionReadinessCheckTimer, shared_from_this (), std::placeholders::_1)); } } } void SAMSocket::SendSessionCreateReplyOk () { uint8_t buf[1024]; char priv[1024]; size_t l = m_Session->localDestination->GetPrivateKeys ().ToBuffer (buf, 1024); size_t l1 = i2p::data::ByteStreamToBase64 (buf, l, priv, 1024); priv[l1] = 0; #ifdef _MSC_VER size_t l2 = sprintf_s (m_Buffer, SAM_SOCKET_BUFFER_SIZE, SAM_SESSION_CREATE_REPLY_OK, priv); #else size_t l2 = snprintf (m_Buffer, SAM_SOCKET_BUFFER_SIZE, SAM_SESSION_CREATE_REPLY_OK, priv); #endif SendMessageReply (m_Buffer, l2, false); } void SAMSocket::ProcessStreamConnect (char * buf, size_t len) { LogPrint (eLogDebug, "SAM: stream connect: ", buf); std::map<std::string, std::string> params; ExtractParams (buf, params); std::string& id = params[SAM_PARAM_ID]; std::string& destination = params[SAM_PARAM_DESTINATION]; std::string& silent = params[SAM_PARAM_SILENT]; if (silent == SAM_VALUE_TRUE) m_IsSilent = true; m_ID = id; m_Session = m_Owner.FindSession (id); if (m_Session) { auto dest = std::make_shared<i2p::data::IdentityEx> (); size_t len = dest->FromBase64(destination); if (len > 0) { context.GetAddressBook().InsertAddress(dest); auto leaseSet = m_Session->localDestination->FindLeaseSet(dest->GetIdentHash()); if (leaseSet) Connect(leaseSet); else { m_Session->localDestination->RequestDestination(dest->GetIdentHash(), std::bind(&SAMSocket::HandleConnectLeaseSetRequestComplete, shared_from_this(), std::placeholders::_1)); } } else SendMessageReply(SAM_SESSION_STATUS_INVALID_KEY, strlen(SAM_SESSION_STATUS_INVALID_KEY), true); } else SendMessageReply (SAM_STREAM_STATUS_INVALID_ID, strlen(SAM_STREAM_STATUS_INVALID_ID), true); } void SAMSocket::Connect (std::shared_ptr<const i2p::data::LeaseSet> remote) { m_SocketType = eSAMSocketTypeStream; m_Session->AddSocket (shared_from_this ()); m_Stream = m_Session->localDestination->CreateStream (remote); m_Stream->Send ((uint8_t *)m_Buffer, 0); // connect I2PReceive (); SendMessageReply (SAM_STREAM_STATUS_OK, strlen(SAM_STREAM_STATUS_OK), false); } void SAMSocket::HandleConnectLeaseSetRequestComplete (std::shared_ptr<i2p::data::LeaseSet> leaseSet) { if (leaseSet) Connect (leaseSet); else { LogPrint (eLogError, "SAM: destination to connect not found"); SendMessageReply (SAM_STREAM_STATUS_CANT_REACH_PEER, strlen(SAM_STREAM_STATUS_CANT_REACH_PEER), true); } } void SAMSocket::ProcessStreamAccept (char * buf, size_t len) { LogPrint (eLogDebug, "SAM: stream accept: ", buf); std::map<std::string, std::string> params; ExtractParams (buf, params); std::string& id = params[SAM_PARAM_ID]; std::string& silent = params[SAM_PARAM_SILENT]; if (silent == SAM_VALUE_TRUE) m_IsSilent = true; m_ID = id; m_Session = m_Owner.FindSession (id); if (m_Session) { m_SocketType = eSAMSocketTypeAcceptor; m_Session->AddSocket (shared_from_this ()); SendMessageReply (SAM_STREAM_STATUS_OK, strlen(SAM_STREAM_STATUS_OK), false); } else SendMessageReply (SAM_STREAM_STATUS_INVALID_ID, strlen(SAM_STREAM_STATUS_INVALID_ID), true); } void SAMSocket::Accept(std::shared_ptr<i2p::stream::Stream> stream) { if(stream) { m_SocketType = eSAMSocketTypeStream; HandleI2PAccept(stream); } else { SendMessageReply (SAM_STREAM_STATUS_I2P_ERROR, strlen(SAM_STREAM_STATUS_I2P_ERROR), true); auto s = shared_from_this (); m_Owner.GetService ().post ([s] { s->Terminate (); }); } } size_t SAMSocket::ProcessDatagramSend (char * buf, size_t len, const char * data) { LogPrint (eLogDebug, "SAM: datagram send: ", buf, " ", len); std::map<std::string, std::string> params; ExtractParams (buf, params); size_t size = std::stoi(params[SAM_PARAM_SIZE]), offset = data - buf; if (offset + size <= len) { if (m_Session) { auto d = m_Session->localDestination->GetDatagramDestination (); if (d) { i2p::data::IdentityEx dest; dest.FromBase64 (params[SAM_PARAM_DESTINATION]); d->SendDatagramTo ((const uint8_t *)data, size, dest.GetIdentHash ()); } else LogPrint (eLogError, "SAM: missing datagram destination"); } else LogPrint (eLogError, "SAM: session is not created from DATAGRAM SEND"); } else { LogPrint (eLogWarning, "SAM: sent datagram size ", size, " exceeds buffer ", len - offset); return 0; // try to receive more } return offset + size; } void SAMSocket::ProcessDestGenerate () { LogPrint (eLogDebug, "SAM: dest generate"); auto keys = i2p::data::PrivateKeys::CreateRandomKeys (); #ifdef _MSC_VER size_t len = sprintf_s (m_Buffer, SAM_SOCKET_BUFFER_SIZE, SAM_DEST_REPLY, keys.GetPublic ()->ToBase64 ().c_str (), keys.ToBase64 ().c_str ()); #else size_t len = snprintf (m_Buffer, SAM_SOCKET_BUFFER_SIZE, SAM_DEST_REPLY, keys.GetPublic ()->ToBase64 ().c_str (), keys.ToBase64 ().c_str ()); #endif SendMessageReply (m_Buffer, len, false); } void SAMSocket::ProcessNamingLookup (char * buf, size_t len) { LogPrint (eLogDebug, "SAM: naming lookup: ", buf); std::map<std::string, std::string> params; ExtractParams (buf, params); std::string& name = params[SAM_PARAM_NAME]; std::shared_ptr<const i2p::data::IdentityEx> identity; i2p::data::IdentHash ident; if (name == "ME") SendNamingLookupReply (m_Session->localDestination->GetIdentity ()); else if ((identity = context.GetAddressBook ().GetAddress (name)) != nullptr) SendNamingLookupReply (identity); else if (m_Session && m_Session->localDestination && context.GetAddressBook ().GetIdentHash (name, ident)) { auto leaseSet = m_Session->localDestination->FindLeaseSet (ident); if (leaseSet) SendNamingLookupReply (leaseSet->GetIdentity ()); else m_Session->localDestination->RequestDestination (ident, std::bind (&SAMSocket::HandleNamingLookupLeaseSetRequestComplete, shared_from_this (), std::placeholders::_1, ident)); } else { LogPrint (eLogError, "SAM: naming failed, unknown address ", name); #ifdef _MSC_VER size_t len = sprintf_s (m_Buffer, SAM_SOCKET_BUFFER_SIZE, SAM_NAMING_REPLY_INVALID_KEY, name.c_str()); #else size_t len = snprintf (m_Buffer, SAM_SOCKET_BUFFER_SIZE, SAM_NAMING_REPLY_INVALID_KEY, name.c_str()); #endif SendMessageReply (m_Buffer, len, false); } } void SAMSocket::HandleNamingLookupLeaseSetRequestComplete (std::shared_ptr<i2p::data::LeaseSet> leaseSet, i2p::data::IdentHash ident) { if (leaseSet) { context.GetAddressBook ().InsertAddress (leaseSet->GetIdentity ()); SendNamingLookupReply (leaseSet->GetIdentity ()); } else { LogPrint (eLogError, "SAM: naming lookup failed. LeaseSet for ", ident.ToBase32 (), " not found"); #ifdef _MSC_VER size_t len = sprintf_s (m_Buffer, SAM_SOCKET_BUFFER_SIZE, SAM_NAMING_REPLY_INVALID_KEY, context.GetAddressBook ().ToAddress (ident).c_str()); #else size_t len = snprintf (m_Buffer, SAM_SOCKET_BUFFER_SIZE, SAM_NAMING_REPLY_INVALID_KEY, context.GetAddressBook ().ToAddress (ident).c_str()); #endif SendMessageReply (m_Buffer, len, false); } } void SAMSocket::SendNamingLookupReply (std::shared_ptr<const i2p::data::IdentityEx> identity) { auto base64 = identity->ToBase64 (); #ifdef _MSC_VER size_t l = sprintf_s (m_Buffer, SAM_SOCKET_BUFFER_SIZE, SAM_NAMING_REPLY, base64.c_str ()); #else size_t l = snprintf (m_Buffer, SAM_SOCKET_BUFFER_SIZE, SAM_NAMING_REPLY, base64.c_str ()); #endif SendMessageReply (m_Buffer, l, false); } void SAMSocket::ExtractParams (char * buf, std::map<std::string, std::string>& params) { char * separator; do { separator = strchr (buf, ' '); if (separator) *separator = 0; char * value = strchr (buf, '='); if (value) { *value = 0; value++; params[buf] = value; } buf = separator + 1; } while (separator); } void SAMSocket::Receive () { if (m_BufferOffset >= SAM_SOCKET_BUFFER_SIZE) { LogPrint (eLogError, "SAM: Buffer is full, terminate"); Terminate (); return; } m_Socket.async_read_some (boost::asio::buffer(m_Buffer + m_BufferOffset, SAM_SOCKET_BUFFER_SIZE - m_BufferOffset), std::bind((m_SocketType == eSAMSocketTypeStream) ? &SAMSocket::HandleReceived : &SAMSocket::HandleMessage, shared_from_this (), std::placeholders::_1, std::placeholders::_2)); } void SAMSocket::HandleReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred) { if (ecode) { LogPrint (eLogError, "SAM: read error: ", ecode.message ()); if (ecode != boost::asio::error::operation_aborted) Terminate (); } else { if (m_Stream) { auto s = shared_from_this (); m_Stream->AsyncSend ((uint8_t *)m_Buffer, bytes_transferred, [s](const boost::system::error_code& ecode) { if (!ecode) s->Receive (); else s->m_Owner.GetService ().post ([s] { s->Terminate (); }); }); } } } void SAMSocket::I2PReceive () { if (m_Stream) { if (m_Stream->GetStatus () == i2p::stream::eStreamStatusNew || m_Stream->GetStatus () == i2p::stream::eStreamStatusOpen) // regular { m_Stream->AsyncReceive (boost::asio::buffer (m_StreamBuffer, SAM_SOCKET_BUFFER_SIZE), std::bind (&SAMSocket::HandleI2PReceive, shared_from_this (), std::placeholders::_1, std::placeholders::_2), SAM_SOCKET_CONNECTION_MAX_IDLE); } else // closed by peer { // get remaning data auto len = m_Stream->ReadSome (m_StreamBuffer, SAM_SOCKET_BUFFER_SIZE); if (len > 0) // still some data { boost::asio::async_write (m_Socket, boost::asio::buffer (m_StreamBuffer, len), std::bind (&SAMSocket::HandleWriteI2PData, shared_from_this (), std::placeholders::_1)); } else // no more data Terminate (); } } } void SAMSocket::HandleI2PReceive (const boost::system::error_code& ecode, std::size_t bytes_transferred) { if (ecode) { LogPrint (eLogError, "SAM: stream read error: ", ecode.message ()); if (ecode != boost::asio::error::operation_aborted) { if (bytes_transferred > 0) boost::asio::async_write (m_Socket, boost::asio::buffer (m_StreamBuffer, bytes_transferred), std::bind (&SAMSocket::HandleWriteI2PData, shared_from_this (), std::placeholders::_1)); // postpone termination else { auto s = shared_from_this (); m_Owner.GetService ().post ([s] { s->Terminate (); }); } } else { auto s = shared_from_this (); m_Owner.GetService ().post ([s] { s->Terminate (); }); } } else { boost::asio::async_write (m_Socket, boost::asio::buffer (m_StreamBuffer, bytes_transferred), std::bind (&SAMSocket::HandleWriteI2PData, shared_from_this (), std::placeholders::_1)); } } void SAMSocket::HandleWriteI2PData (const boost::system::error_code& ecode) { if (ecode) { LogPrint (eLogError, "SAM: socket write error: ", ecode.message ()); if (ecode != boost::asio::error::operation_aborted) Terminate (); } else I2PReceive (); } void SAMSocket::HandleI2PAccept (std::shared_ptr<i2p::stream::Stream> stream) { if (stream) { LogPrint (eLogDebug, "SAM: incoming I2P connection for session ", m_ID); m_Stream = stream; context.GetAddressBook ().InsertAddress (stream->GetRemoteIdentity ()); if (!m_IsSilent) { // get remote peer address auto ident_ptr = stream->GetRemoteIdentity(); const size_t ident_len = ident_ptr->GetFullLen(); uint8_t* ident = new uint8_t[ident_len]; // send remote peer address as base64 const size_t l = ident_ptr->ToBuffer (ident, ident_len); const size_t l1 = i2p::data::ByteStreamToBase64 (ident, l, (char *)m_StreamBuffer, SAM_SOCKET_BUFFER_SIZE); delete[] ident; m_StreamBuffer[l1] = '\n'; HandleI2PReceive (boost::system::error_code (), l1 +1); // we send identity like it has been received from stream } else I2PReceive (); } else LogPrint (eLogWarning, "SAM: I2P 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); auto base64 = from.ToBase64 (); #ifdef _MSC_VER size_t l = sprintf_s ((char *)m_StreamBuffer, SAM_SOCKET_BUFFER_SIZE, SAM_DATAGRAM_RECEIVED, base64.c_str (), len); #else size_t l = snprintf ((char *)m_StreamBuffer, SAM_SOCKET_BUFFER_SIZE, SAM_DATAGRAM_RECEIVED, base64.c_str (), len); #endif if (len < SAM_SOCKET_BUFFER_SIZE - l) { memcpy (m_StreamBuffer + l, buf, len); boost::asio::async_write (m_Socket, boost::asio::buffer (m_StreamBuffer, len + l), std::bind (&SAMSocket::HandleWriteI2PData, shared_from_this (), std::placeholders::_1)); } else LogPrint (eLogWarning, "SAM: received datagram size ", len," exceeds buffer"); } SAMSession::SAMSession (std::shared_ptr<ClientDestination> dest): localDestination (dest), m_BacklogPumper(dest->GetService()) { PumpBacklog(); } void SAMSession::AcceptI2P(std::shared_ptr<i2p::stream::Stream> stream) { if(!stream) return; // fail std::unique_lock<std::mutex> lock(m_SocketsMutex); if(m_Backlog.size() > SAM_MAX_ACCEPT_BACKLOG) { stream->Close(); return; } m_Backlog.push_back(stream); } void SAMSession::PumpBacklog() { // pump backlog every 100ms boost::posix_time::milliseconds dlt(100); m_BacklogPumper.expires_from_now(dlt); m_BacklogPumper.async_wait(std::bind(&SAMSession::HandlePumpBacklog, this, std::placeholders::_1)); } std::shared_ptr<SAMSocket> SAMSession::FindAcceptor() { for (auto & sock : m_Sockets) { auto t = sock->GetSocketType(); if(t == eSAMSocketTypeAcceptor) { return sock; } } return nullptr; } void SAMSession::HandlePumpBacklog(const boost::system::error_code & ec) { if(ec) return; { std::unique_lock<std::mutex> lock(m_SocketsMutex); auto itr = m_Backlog.begin(); while(itr != m_Backlog.end()) { auto sock = FindAcceptor(); if (sock) { sock->Accept(*itr); itr = m_Backlog.erase(itr); } else { ++itr; } } } PumpBacklog(); } void SAMSession::CloseStreams () { m_BacklogPumper.cancel(); localDestination->GetService().post([&] () { std::lock_guard<std::mutex> lock(m_SocketsMutex); for (auto& sock : m_Sockets) { sock->CloseStream(); } for(auto & stream : m_Backlog) { stream->Close(); } m_Sockets.clear(); m_Backlog.clear(); i2p::client::context.DeleteLocalDestination (localDestination); }); } SAMBridge::SAMBridge (const std::string& address, int port): m_IsRunning (false), m_Thread (nullptr), m_Acceptor (m_Service, boost::asio::ip::tcp::endpoint(boost::asio::ip::address::from_string(address), port)), m_DatagramEndpoint (boost::asio::ip::address::from_string(address), port-1), m_DatagramSocket (m_Service, m_DatagramEndpoint) { } SAMBridge::~SAMBridge () { if (m_IsRunning) Stop (); } void SAMBridge::Start () { Accept (); ReceiveDatagram (); m_IsRunning = true; m_Thread = new std::thread (std::bind (&SAMBridge::Run, this)); } void SAMBridge::Stop () { m_IsRunning = false; m_Acceptor.cancel (); for (auto& it: m_Sessions) it.second->CloseStreams (); m_Sessions.clear (); m_Service.stop (); if (m_Thread) { m_Thread->join (); delete m_Thread; m_Thread = nullptr; } } void SAMBridge::Run () { while (m_IsRunning) { try { m_Service.run (); } catch (std::exception& ex) { LogPrint (eLogError, "SAM: runtime exception: ", ex.what ()); } } } void SAMBridge::Accept () { auto newSocket = std::make_shared<SAMSocket> (*this); m_Acceptor.async_accept (newSocket->GetSocket (), std::bind (&SAMBridge::HandleAccept, this, std::placeholders::_1, newSocket)); } void SAMBridge::HandleAccept(const boost::system::error_code& ecode, std::shared_ptr<SAMSocket> socket) { if (!ecode) { boost::system::error_code ec; auto ep = socket->GetSocket ().remote_endpoint (ec); if (!ec) { LogPrint (eLogDebug, "SAM: new connection from ", ep); socket->ReceiveHandshake (); } else LogPrint (eLogError, "SAM: incoming connection error ", ec.message ()); } else LogPrint (eLogError, "SAM: accept error: ", ecode.message ()); if (ecode != boost::asio::error::operation_aborted) Accept (); } std::shared_ptr<SAMSession> SAMBridge::CreateSession (const std::string& id, const std::string& destination, const std::map<std::string, std::string> * params) { std::shared_ptr<ClientDestination> localDestination = nullptr; if (destination != "") { i2p::data::PrivateKeys keys; keys.FromBase64 (destination); localDestination = i2p::client::context.CreateNewLocalDestination (keys, true, params); } else // transient { // extract signature type i2p::data::SigningKeyType signatureType = i2p::data::SIGNING_KEY_TYPE_DSA_SHA1; if (params) { auto it = params->find (SAM_PARAM_SIGNATURE_TYPE); if (it != params->end ()) // TODO: extract string values signatureType = std::stoi(it->second); } localDestination = i2p::client::context.CreateNewLocalDestination (true, signatureType, params); } if (localDestination) { auto session = std::make_shared<SAMSession>(localDestination); std::unique_lock<std::mutex> l(m_SessionsMutex); auto ret = m_Sessions.insert (std::make_pair(id, session)); if (!ret.second) { LogPrint (eLogWarning, "SAM: Session ", id, " already exists"); } return ret.first->second; } return nullptr; } void SAMBridge::CloseSession (const std::string& id) { std::shared_ptr<SAMSession> session; { std::unique_lock<std::mutex> l(m_SessionsMutex); auto it = m_Sessions.find (id); if (it != m_Sessions.end ()) { session = it->second; m_Sessions.erase (it); } } if (session) { session->localDestination->StopAcceptingStreams (); session->CloseStreams (); } } std::shared_ptr<SAMSession> SAMBridge::FindSession (const std::string& id) const { std::unique_lock<std::mutex> l(m_SessionsMutex); auto it = m_Sessions.find (id); if (it != m_Sessions.end ()) return it->second; return nullptr; } void SAMBridge::ReceiveDatagram () { m_DatagramSocket.async_receive_from ( boost::asio::buffer (m_DatagramReceiveBuffer, i2p::datagram::MAX_DATAGRAM_SIZE), m_SenderEndpoint, std::bind (&SAMBridge::HandleReceivedDatagram, this, std::placeholders::_1, std::placeholders::_2)); } void SAMBridge::HandleReceivedDatagram (const boost::system::error_code& ecode, std::size_t bytes_transferred) { if (!ecode) { m_DatagramReceiveBuffer[bytes_transferred] = 0; char * eol = strchr ((char *)m_DatagramReceiveBuffer, '\n'); *eol = 0; eol++; size_t payloadLen = bytes_transferred - ((uint8_t *)eol - m_DatagramReceiveBuffer); LogPrint (eLogDebug, "SAM: datagram received ", m_DatagramReceiveBuffer," size=", payloadLen); char * sessionID = strchr ((char *)m_DatagramReceiveBuffer, ' '); if (sessionID) { sessionID++; char * destination = strchr (sessionID, ' '); if (destination) { *destination = 0; destination++; auto session = FindSession (sessionID); if (session) { i2p::data::IdentityEx dest; dest.FromBase64 (destination); session->localDestination->GetDatagramDestination ()-> SendDatagramTo ((uint8_t *)eol, payloadLen, dest.GetIdentHash ()); } else LogPrint (eLogError, "SAM: Session ", sessionID, " not found"); } else LogPrint (eLogError, "SAM: Missing destination key"); } else LogPrint (eLogError, "SAM: Missing sessionID"); ReceiveDatagram (); } else LogPrint (eLogError, "SAM: datagram receive error: ", ecode.message ()); } } }