Merge pull request #378 from PurpleI2P/openssl

recent changes

Destination.cpp

@@ -17,7 +17,8 @@ namespace client
 			const std::map<std::string, std::string> * params):
 		m_IsRunning (false), m_Thread (nullptr), m_Work (m_Service),	
 		m_Keys (keys), m_IsPublic (isPublic), m_PublishReplyToken (0),
-		m_DatagramDestination (nullptr), m_PublishConfirmationTimer (m_Service), m_CleanupTimer (m_Service)
+		m_DatagramDestination (nullptr), m_PublishConfirmationTimer (m_Service), 
+		m_PublishVerificationTimer (m_Service), m_CleanupTimer (m_Service)
 	{
 		if (m_IsPublic)	
 			PersistTemporaryKeys ();
@@ -156,6 +157,8 @@ namespace client
 		if (m_IsRunning)
 		{	
 			m_CleanupTimer.cancel ();
+			m_PublishConfirmationTimer.cancel ();
+			m_PublishVerificationTimer.cancel ();
 			m_IsRunning = false;
 			m_StreamingDestination->Stop ();
 			m_StreamingDestination = nullptr;
@@ -298,8 +301,13 @@ namespace client
 				leaseSet = std::make_shared<i2p::data::LeaseSet> (buf + offset, len - offset);
 				if (leaseSet->IsValid ())
 				{
-					LogPrint (eLogDebug, "New remote LeaseSet added");
-					m_RemoteLeaseSets[buf + DATABASE_STORE_KEY_OFFSET] = leaseSet;
+					if (leaseSet->GetIdentHash () != GetIdentHash ())
+					{
+						LogPrint (eLogDebug, "New remote LeaseSet added");
+						m_RemoteLeaseSets[buf + DATABASE_STORE_KEY_OFFSET] = leaseSet;
+					}
+					else
+						LogPrint (eLogDebug, "Own remote LeaseSet dropped");
 				}
 				else
 				{
@@ -371,6 +379,10 @@ namespace client
 			LogPrint (eLogDebug, "Destination: Publishing LeaseSet confirmed");
 			m_ExcludedFloodfills.clear ();
 			m_PublishReplyToken = 0;
+			// schedule verification
+			m_PublishVerificationTimer.expires_from_now (boost::posix_time::seconds(PUBLISH_VERIFICATION_TIMEOUT));
+			m_PublishVerificationTimer.async_wait (std::bind (&ClientDestination::HandlePublishVerificationTimer,
+			shared_from_this (), std::placeholders::_1));	
 		}
 		else
 			i2p::garlic::GarlicDestination::HandleDeliveryStatusMessage (msg);
@@ -381,7 +393,10 @@ namespace client
 		i2p::garlic::GarlicDestination::SetLeaseSetUpdated ();	
 		UpdateLeaseSet ();
 		if (m_IsPublic)
+		{
+			m_PublishVerificationTimer.cancel ();
 			Publish ();
+		}
 	}
 		
 	void ClientDestination::Publish ()
@@ -402,7 +417,6 @@ namespace client
 			LogPrint (eLogError, "Destination: Can't publish LeaseSet. No outbound tunnels");
 			return;
 		}
-		std::set<i2p::data::IdentHash> excluded; 
 		auto floodfill = i2p::data::netdb.GetClosestFloodfill (m_LeaseSet->GetIdentHash (), m_ExcludedFloodfills);	
 		if (!floodfill)
 		{
@@ -416,7 +430,7 @@ namespace client
 		auto msg = WrapMessage (floodfill, i2p::CreateDatabaseStoreMsg (m_LeaseSet, m_PublishReplyToken));			
 		m_PublishConfirmationTimer.expires_from_now (boost::posix_time::seconds(PUBLISH_CONFIRMATION_TIMEOUT));
 		m_PublishConfirmationTimer.async_wait (std::bind (&ClientDestination::HandlePublishConfirmationTimer,
-			this, std::placeholders::_1));	
+			shared_from_this (), std::placeholders::_1));	
 		outbound->SendTunnelDataMsg (floodfill->GetIdentHash (), 0, msg);	
 	}
 
@@ -433,6 +447,33 @@ namespace client
 		}
 	}
 
+	void ClientDestination::HandlePublishVerificationTimer (const boost::system::error_code& ecode)
+	{
+		if (ecode != boost::asio::error::operation_aborted)
+		{	
+			auto s = shared_from_this ();
+			RequestLeaseSet (GetIdentHash (), 
+				[s](std::shared_ptr<i2p::data::LeaseSet> leaseSet)
+				{
+					if (leaseSet)
+					{
+						if (s->m_LeaseSet && *s->m_LeaseSet == *leaseSet)
+						{
+							// we got latest LeasetSet
+							LogPrint (eLogDebug, "Destination: published LeaseSet verified");
+							s->m_PublishVerificationTimer.expires_from_now (boost::posix_time::seconds(PUBLISH_REGULAR_VERIFICATION_INTERNAL));
+							s->m_PublishVerificationTimer.async_wait (std::bind (&ClientDestination::HandlePublishVerificationTimer, s, std::placeholders::_1));	
+							return;
+						}		
+					}	
+					else
+						LogPrint (eLogWarning, "Destination: couldn't find published LeaseSet");
+					// we have to publish again
+					s->Publish ();	
+				});
+		}
+	}
+
 	void ClientDestination::HandleDataMessage (const uint8_t * buf, size_t len)
 	{
 		uint32_t length = bufbe32toh (buf);
@@ -622,7 +663,7 @@ namespace client
 				});	
 			request->requestTimeoutTimer.expires_from_now (boost::posix_time::seconds(LEASESET_REQUEST_TIMEOUT));
 			request->requestTimeoutTimer.async_wait (std::bind (&ClientDestination::HandleRequestTimoutTimer,
-				this, std::placeholders::_1, dest));
+				shared_from_this (), std::placeholders::_1, dest));
 		}	
 		else
 			return false;

Destination.h

@@ -26,6 +26,8 @@ namespace client
 	const uint8_t PROTOCOL_TYPE_DATAGRAM = 17;
 	const uint8_t PROTOCOL_TYPE_RAW = 18;	
 	const int PUBLISH_CONFIRMATION_TIMEOUT = 5; // in seconds
+	const int PUBLISH_VERIFICATION_TIMEOUT = 10; // in seconds after successfull publish
+	const int PUBLISH_REGULAR_VERIFICATION_INTERNAL = 100; // in seconds periodically	
 	const int LEASESET_REQUEST_TIMEOUT = 5; // in seconds
 	const int MAX_LEASESET_REQUEST_TIMEOUT = 40; // in seconds
 	const int DESTINATION_CLEANUP_TIMEOUT = 3; // in minutes 
@@ -116,6 +118,7 @@ namespace client
 			void UpdateLeaseSet ();
 			void Publish ();
 			void HandlePublishConfirmationTimer (const boost::system::error_code& ecode);
+			void HandlePublishVerificationTimer (const boost::system::error_code& ecode);
 			void HandleDatabaseStoreMessage (const uint8_t * buf, size_t len);
 			void HandleDatabaseSearchReplyMessage (const uint8_t * buf, size_t len);
 			void HandleDeliveryStatusMessage (std::shared_ptr<I2NPMessage> msg);		
@@ -148,7 +151,7 @@ namespace client
 			std::map<uint16_t, std::shared_ptr<i2p::stream::StreamingDestination> > m_StreamingDestinationsByPorts;
 			i2p::datagram::DatagramDestination * m_DatagramDestination;
 	
-			boost::asio::deadline_timer m_PublishConfirmationTimer, m_CleanupTimer;
+			boost::asio::deadline_timer m_PublishConfirmationTimer, m_PublishVerificationTimer, m_CleanupTimer;
 
 		public:
 			

Garlic.h

@@ -62,6 +62,7 @@ namespace garlic
 	{
 		std::shared_ptr<i2p::tunnel::OutboundTunnel> outboundTunnel;
 		std::shared_ptr<const i2p::data::Lease> remoteLease;
+		int rtt; // RTT
 		uint32_t updateTime; // seconds since epoch
 	};	
 	

LeaseSet.cpp

@@ -130,7 +130,7 @@ namespace data
 			leases += 4; // tunnel ID
 			lease.endDate = bufbe64toh (leases); 
 			leases += 8; // end date
-			if (ts < lease.endDate)
+			if (ts < lease.endDate + LEASE_ENDDATE_THRESHOLD)
 			{	
 				if (lease.endDate > m_ExpirationTime)
 					m_ExpirationTime = lease.endDate;
@@ -151,7 +151,7 @@ namespace data
 			else
 				LogPrint (eLogWarning, "LeaseSet: Lease is expired already ");
 		}	
-		if (!m_ExpirationTime)
+		if (!m_ExpirationTime && m_Leases.empty ())
 		{
 			LogPrint (eLogWarning, "LeaseSet: all leases are expired. Dropped");
 			m_IsValid = false;
@@ -192,6 +192,12 @@ namespace data
 			if (ts < endDate)
 				leases.push_back (it);
 		}	
+		if (leases.empty () && withThreshold)
+		{
+			for (auto it: m_Leases)
+				if (ts < it->endDate + LEASE_ENDDATE_THRESHOLD)
+					leases.push_back (it);
+		}	
 		return leases;	
 	}	
 

LeaseSet.h

@@ -17,6 +17,7 @@ namespace tunnel
 
 namespace data
 {	
+	const int LEASE_ENDDATE_THRESHOLD = 9500; // in milliseconds
 	struct Lease
 	{
 		IdentHash tunnelGateway;
@@ -57,6 +58,8 @@ namespace data
 			bool IsExpired () const;
 			bool IsEmpty () const { return m_Leases.empty (); };
 			uint64_t GetExpirationTime () const { return m_ExpirationTime; };
+			bool operator== (const LeaseSet& other) const 
+			{ return m_BufferLen == other.m_BufferLen && !memcmp (m_Buffer, other.m_Buffer, m_BufferLen); }; 
 
 			// implements RoutingDestination
 			const IdentHash& GetIdentHash () const { return m_Identity->GetIdentHash (); };

NetDb.cpp

@@ -496,9 +496,11 @@ namespace data
 					std::set<IdentHash> excluded;
 					for (int i = 0; i < 3; i++)
 					{
-						auto floodfill = GetClosestFloodfill (ident, excluded);
+						auto floodfill = GetClosestFloodfill (ident, excluded, true); // we need a floodfill close than us only
 						if (floodfill)
 							transports.SendMessage (floodfill->GetIdentHash (), floodMsg);
+						else
+							break;
 					}	
 				}	
 				else
@@ -688,7 +690,7 @@ namespace data
 			    lookupType == DATABASE_LOOKUP_TYPE_NORMAL_LOOKUP))
 			{
 				auto leaseSet = FindLeaseSet (ident);
-				if (leaseSet) // we don't send back our LeaseSets
+				if (leaseSet && !leaseSet->IsExpired ()) // we don't send back our LeaseSets
 				{
 					LogPrint (eLogDebug, "NetDb: requested LeaseSet ", key, " found");
 					replyMsg = CreateDatabaseStoreMsg (leaseSet);
@@ -885,12 +887,15 @@ namespace data
 	}	
 
 	std::shared_ptr<const RouterInfo> NetDb::GetClosestFloodfill (const IdentHash& destination, 
-		const std::set<IdentHash>& excluded) const
+		const std::set<IdentHash>& excluded, bool closeThanUsOnly) const
 	{
 		std::shared_ptr<const RouterInfo> r;
 		XORMetric minMetric;
 		IdentHash destKey = CreateRoutingKey (destination);
-		minMetric.SetMax ();
+		if (closeThanUsOnly)
+			minMetric = destKey ^ i2p::context.GetIdentHash ();
+		else	
+			minMetric.SetMax ();
 		std::unique_lock<std::mutex> l(m_FloodfillsMutex);
 		for (auto it: m_Floodfills)
 		{	

NetDb.h

@@ -52,7 +52,7 @@ namespace data
 			std::shared_ptr<const RouterInfo> GetHighBandwidthRandomRouter (std::shared_ptr<const RouterInfo> compatibleWith) const;
 			std::shared_ptr<const RouterInfo> GetRandomPeerTestRouter () const;
 			std::shared_ptr<const RouterInfo> GetRandomIntroducer () const;
-			std::shared_ptr<const RouterInfo> GetClosestFloodfill (const IdentHash& destination, const std::set<IdentHash>& excluded) const;
+			std::shared_ptr<const RouterInfo> GetClosestFloodfill (const IdentHash& destination, const std::set<IdentHash>& excluded, bool closeThanUsOnly = false) const;
 			std::vector<IdentHash> GetClosestFloodfills (const IdentHash& destination, size_t num,
 				std::set<IdentHash>& excluded) const;
 			std::shared_ptr<const RouterInfo> GetClosestNonFloodfill (const IdentHash& destination, const std::set<IdentHash>& excluded) const;

Streaming.cpp

@@ -274,7 +274,7 @@ namespace stream
 				if (!seqn && m_RoutingSession) // first message confirmed
 					m_RoutingSession->SetSharedRoutingPath (
 						std::make_shared<i2p::garlic::GarlicRoutingPath> (
-							i2p::garlic::GarlicRoutingPath{m_CurrentOutboundTunnel, m_CurrentRemoteLease, 0}));
+							i2p::garlic::GarlicRoutingPath{m_CurrentOutboundTunnel, m_CurrentRemoteLease, m_RTT, 0}));
 			}
 			else
 				break;
@@ -605,6 +605,8 @@ namespace stream
 				{
 					m_CurrentOutboundTunnel = routingPath->outboundTunnel;
 					m_CurrentRemoteLease = routingPath->remoteLease;
+					m_RTT = routingPath->rtt;
+					m_RTO = m_RTT*1.5; // TODO: implement it better
 				}
 			}	
 		}	
@@ -727,7 +729,8 @@ namespace stream
 	{
 		if (!m_RemoteLeaseSet || m_RemoteLeaseSet->IsExpired ()) 
 		{
-			m_RemoteLeaseSet = m_LocalDestination.GetOwner ()->FindLeaseSet (m_RemoteIdentity->GetIdentHash ());
+			auto remoteLeaseSet = m_LocalDestination.GetOwner ()->FindLeaseSet (m_RemoteIdentity->GetIdentHash ());
+			if (remoteLeaseSet) m_RemoteLeaseSet = remoteLeaseSet; // renew if possible
 			if (!m_RemoteLeaseSet)	
 				LogPrint (eLogWarning, "Streaming: LeaseSet ", m_RemoteIdentity->GetIdentHash ().ToBase64 (), " not found");
 		}