Merge pull request #371 from PurpleI2P/openssl

recent changes

Identity.cpp

@@ -234,10 +234,7 @@ namespace data
 	size_t IdentityEx::ToBuffer (uint8_t * buf, size_t len) const
 	{
 		const size_t fullLen = GetFullLen();
-		if (fullLen > len) {
-			// buffer is too small and may overflow somewhere else
-			return 0;
-		}
+		if (fullLen > len) return 0; // buffer is too small and may overflow somewhere else
 		memcpy (buf, &m_StandardIdentity, DEFAULT_IDENTITY_SIZE);
 		if (m_ExtendedLen > 0 && m_ExtendedBuffer)
 			memcpy (buf + DEFAULT_IDENTITY_SIZE, m_ExtendedBuffer, m_ExtendedLen);
@@ -247,9 +244,8 @@ namespace data
 	size_t IdentityEx::FromBase64(const std::string& s)
 	{
 		const size_t slen = s.length();
-		const size_t bufLen = Base64EncodingBufferSize(slen);
-		uint8_t buf[bufLen];
-		const size_t len = Base64ToByteStream (s.c_str(), slen, buf, bufLen);
+		uint8_t buf[slen]; // binary data can't exceed base64
+		const size_t len = Base64ToByteStream (s.c_str(), slen, buf, slen);
 		return FromBuffer (buf, len);
 	}	
 	

Streaming.cpp

@@ -787,6 +787,12 @@ namespace stream
 		
 	StreamingDestination::~StreamingDestination ()
 	{
+		for (auto it: m_SavedPackets)
+		{
+			for (auto it1: it.second) delete it1;
+			it.second.clear ();	
+		}
+		m_SavedPackets.clear ();
 	}
 
 	void StreamingDestination::Start ()
@@ -822,7 +828,20 @@ namespace stream
 			if (packet->IsSYN () && !packet->GetSeqn ()) // new incoming stream
 			{	
 				auto incomingStream = CreateNewIncomingStream ();
-				incomingStream->HandleNextPacket (packet);
+				uint32_t receiveStreamID = packet->GetReceiveStreamID ();
+				incomingStream->HandleNextPacket (packet); // SYN
+				// handle saved packets if any
+				{
+					auto it = m_SavedPackets.find (receiveStreamID);
+					if (it != m_SavedPackets.end ())
+					{
+						LogPrint (eLogDebug, "Streaming: Processing ", it->second.size (), " saved packets for receiveStreamID=", 	receiveStreamID);
+						for (auto it1: it->second)
+							incomingStream->HandleNextPacket (it1);
+						m_SavedPackets.erase (it);
+					}			
+				}	
+				// accept
 				if (m_Acceptor != nullptr)
 					m_Acceptor (incomingStream);
 				else
@@ -834,7 +853,7 @@ namespace stream
 						m_PendingIncomingTimer.cancel ();
 						m_PendingIncomingTimer.expires_from_now (boost::posix_time::seconds(PENDING_INCOMING_TIMEOUT));
 						m_PendingIncomingTimer.async_wait (std::bind (&StreamingDestination::HandlePendingIncomingTimer, 
-							this, std::placeholders::_1));
+							shared_from_this (), std::placeholders::_1));
 						LogPrint (eLogDebug, "Streaming: Pending incoming stream added");
 					}
 					else
@@ -854,9 +873,30 @@ namespace stream
 						it.second->HandleNextPacket (packet);
 						return;
 					}
-				// TODO: should queue it up
-				LogPrint (eLogError, "Streaming: Unknown stream receiveStreamID=", receiveStreamID);
-				delete packet;
+				// save follow on packet
+				auto it = m_SavedPackets.find (receiveStreamID);
+				if (it != m_SavedPackets.end ())
+					it->second.push_back (packet);
+				else
+				{
+					m_SavedPackets.emplace (receiveStreamID, std::list<Packet *>{ packet });
+					auto timer = std::make_shared<boost::asio::deadline_timer> (m_Owner->GetService ());
+					timer->expires_from_now (boost::posix_time::seconds(PENDING_INCOMING_TIMEOUT));
+					auto s = shared_from_this ();
+					timer->async_wait ([s,timer,receiveStreamID](const boost::system::error_code& ecode)
+					{
+						if (ecode != boost::asio::error::operation_aborted)
+						{
+							auto it = s->m_SavedPackets.find (receiveStreamID);
+							if (it != s->m_SavedPackets.end ())
+							{
+								for (auto it1: it->second) delete it1;
+								it->second.clear ();
+								s->m_SavedPackets.erase (it);
+							}
+						}
+					});
+				}
 			}	
 		}	
 	}	

Streaming.h

@@ -189,7 +189,7 @@ namespace stream
 			SendHandler m_SendHandler;
 	};
 
-	class StreamingDestination
+	class StreamingDestination: public std::enable_shared_from_this<StreamingDestination>
 	{
 		public:
 
@@ -222,10 +222,11 @@ namespace stream
 			std::shared_ptr<i2p::client::ClientDestination> m_Owner;
 			uint16_t m_LocalPort;
 			std::mutex m_StreamsMutex;
-			std::map<uint32_t, std::shared_ptr<Stream> > m_Streams;
+			std::map<uint32_t, std::shared_ptr<Stream> > m_Streams; // sendStreamID->stream
 			Acceptor m_Acceptor;
 			std::list<std::shared_ptr<Stream> > m_PendingIncomingStreams;
 			boost::asio::deadline_timer m_PendingIncomingTimer;
+			std::map<uint32_t, std::list<Packet *> > m_SavedPackets; // receiveStreamID->packets, arrived before SYN
 			
 		public:
 

filelist.mk

@@ -4,11 +4,11 @@ LIB_SRC = \
   Reseed.cpp RouterContext.cpp RouterInfo.cpp Signature.cpp SSU.cpp \
   SSUSession.cpp SSUData.cpp Streaming.cpp Identity.cpp TransitTunnel.cpp \
   Transports.cpp Tunnel.cpp TunnelEndpoint.cpp TunnelPool.cpp TunnelGateway.cpp \
-  Destination.cpp Base.cpp I2PEndian.cpp util.cpp api.cpp
+  Destination.cpp Base.cpp I2PEndian.cpp Config.cpp util.cpp api.cpp
 
 LIB_CLIENT_SRC = \
 	AddressBook.cpp BOB.cpp ClientContext.cpp I2PTunnel.cpp I2PService.cpp \
-	SAM.cpp SOCKS.cpp HTTPProxy.cpp Config.cpp
+	SAM.cpp SOCKS.cpp HTTPProxy.cpp
 
 # also: Daemon{Linux,Win32}.cpp will be added later
 DAEMON_SRC = \