Browse Source

Merge pull request #2118 from PurpleI2P/openssl

recent changes
master
orignal 3 weeks ago committed by GitHub
parent
commit
e695f1e060
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
  1. 26
      daemon/I2PControl.cpp
  2. 4
      daemon/UnixDaemon.cpp
  3. 2
      libi2pd/Crypto.cpp
  4. 7
      libi2pd/Destination.cpp
  5. 18
      libi2pd/Destination.h
  6. 69
      libi2pd/Family.cpp
  7. 6
      libi2pd/Family.h
  8. 2
      libi2pd/Garlic.cpp
  9. 2
      libi2pd/Garlic.h
  10. 26
      libi2pd/I2NPProtocol.cpp
  11. 3
      libi2pd/I2NPProtocol.h
  12. 4
      libi2pd/NTCP2.cpp
  13. 85
      libi2pd/NetDb.cpp
  14. 1
      libi2pd/NetDb.hpp
  15. 8
      libi2pd/Profiling.h
  16. 32
      libi2pd/Queue.h
  17. 2
      libi2pd/RouterContext.cpp
  18. 22
      libi2pd/RouterContext.h
  19. 8
      libi2pd/RouterInfo.cpp
  20. 6
      libi2pd/RouterInfo.h
  21. 177
      libi2pd/SSU2.cpp
  22. 36
      libi2pd/SSU2.h
  23. 355
      libi2pd/SSU2OutOfSession.cpp
  24. 86
      libi2pd/SSU2OutOfSession.h
  25. 428
      libi2pd/SSU2Session.cpp
  26. 54
      libi2pd/SSU2Session.h
  27. 105
      libi2pd/Streaming.cpp
  28. 9
      libi2pd/Streaming.h
  29. 25
      libi2pd/Transports.cpp
  30. 44
      libi2pd/Tunnel.cpp
  31. 6
      libi2pd/Tunnel.h
  32. 8
      libi2pd/TunnelPool.cpp
  33. 6
      libi2pd_client/I2PTunnel.cpp
  34. 2
      libi2pd_client/UDPTunnel.cpp

26
daemon/I2PControl.cpp

@ -45,15 +45,29 @@ namespace client
i2pcp_crt = i2p::fs::DataDirPath(i2pcp_crt); i2pcp_crt = i2p::fs::DataDirPath(i2pcp_crt);
if (i2pcp_key.at(0) != '/') if (i2pcp_key.at(0) != '/')
i2pcp_key = i2p::fs::DataDirPath(i2pcp_key); i2pcp_key = i2p::fs::DataDirPath(i2pcp_key);
if (!i2p::fs::Exists (i2pcp_crt) || !i2p::fs::Exists (i2pcp_key)) { if (!i2p::fs::Exists (i2pcp_crt) || !i2p::fs::Exists (i2pcp_key))
{
LogPrint (eLogInfo, "I2PControl: Creating new certificate for control connection"); LogPrint (eLogInfo, "I2PControl: Creating new certificate for control connection");
CreateCertificate (i2pcp_crt.c_str(), i2pcp_key.c_str()); CreateCertificate (i2pcp_crt.c_str(), i2pcp_key.c_str());
} else {
LogPrint(eLogDebug, "I2PControl: Using cert from ", i2pcp_crt);
} }
else
LogPrint(eLogDebug, "I2PControl: Using cert from ", i2pcp_crt);
m_SSLContext.set_options (boost::asio::ssl::context::default_workarounds | boost::asio::ssl::context::no_sslv2 | boost::asio::ssl::context::single_dh_use); m_SSLContext.set_options (boost::asio::ssl::context::default_workarounds | boost::asio::ssl::context::no_sslv2 | boost::asio::ssl::context::single_dh_use);
m_SSLContext.use_certificate_file (i2pcp_crt, boost::asio::ssl::context::pem); boost::system::error_code ec;
m_SSLContext.use_private_key_file (i2pcp_key, boost::asio::ssl::context::pem); m_SSLContext.use_certificate_file (i2pcp_crt, boost::asio::ssl::context::pem, ec);
if (!ec)
m_SSLContext.use_private_key_file (i2pcp_key, boost::asio::ssl::context::pem, ec);
if (ec)
{
LogPrint (eLogInfo, "I2PControl: Failed to load ceritifcate: ", ec.message (), ". Recreating");
CreateCertificate (i2pcp_crt.c_str(), i2pcp_key.c_str());
m_SSLContext.use_certificate_file (i2pcp_crt, boost::asio::ssl::context::pem, ec);
if (!ec)
m_SSLContext.use_private_key_file (i2pcp_key, boost::asio::ssl::context::pem, ec);
if (ec)
// give up
LogPrint (eLogError, "I2PControl: Can't load certificates");
}
// handlers // handlers
m_MethodHandlers["Authenticate"] = &I2PControlService::AuthenticateHandler; m_MethodHandlers["Authenticate"] = &I2PControlService::AuthenticateHandler;
@ -403,7 +417,7 @@ namespace client
X509_NAME_add_entry_by_txt (name, "O", MBSTRING_ASC, (unsigned char *)I2P_CONTROL_CERTIFICATE_ORGANIZATION, -1, -1, 0); // organization X509_NAME_add_entry_by_txt (name, "O", MBSTRING_ASC, (unsigned char *)I2P_CONTROL_CERTIFICATE_ORGANIZATION, -1, -1, 0); // organization
X509_NAME_add_entry_by_txt (name, "CN", MBSTRING_ASC, (unsigned char *)I2P_CONTROL_CERTIFICATE_COMMON_NAME, -1, -1, 0); // common name X509_NAME_add_entry_by_txt (name, "CN", MBSTRING_ASC, (unsigned char *)I2P_CONTROL_CERTIFICATE_COMMON_NAME, -1, -1, 0); // common name
X509_set_issuer_name (x509, name); // set issuer to ourselves X509_set_issuer_name (x509, name); // set issuer to ourselves
X509_sign (x509, pkey, EVP_sha1 ()); // sign X509_sign (x509, pkey, EVP_sha1 ()); // sign, last param must be NULL for EdDSA
// save cert // save cert
if ((f = fopen (crt_path, "wb")) != NULL) { if ((f = fopen (crt_path, "wb")) != NULL) {

4
daemon/UnixDaemon.cpp

@ -1,5 +1,5 @@
/* /*
* Copyright (c) 2013-2020, The PurpleI2P Project * Copyright (c) 2013-2024, The PurpleI2P Project
* *
* This file is part of Purple i2pd project and licensed under BSD3 * This file is part of Purple i2pd project and licensed under BSD3
* *
@ -25,6 +25,7 @@
#include "RouterContext.h" #include "RouterContext.h"
#include "ClientContext.h" #include "ClientContext.h"
#include "Transports.h" #include "Transports.h"
#include "util.h"
void handle_signal(int sig) void handle_signal(int sig)
{ {
@ -220,6 +221,7 @@ namespace i2p
void DaemonLinux::run () void DaemonLinux::run ()
{ {
i2p::util::SetThreadName ("i2pd-daemon");
while (running) while (running)
{ {
std::this_thread::sleep_for (std::chrono::seconds(1)); std::this_thread::sleep_for (std::chrono::seconds(1));

2
libi2pd/Crypto.cpp

@ -997,7 +997,7 @@ namespace crypto
} }
else else
{ {
#if defined(LIBRESSL_VERSION_NUMBER) #if defined(LIBRESSL_VERSION_NUMBER) && LIBRESSL_VERSION_NUMBER < 0x4000000fL
std::vector<uint8_t> m(msgLen + 16); std::vector<uint8_t> m(msgLen + 16);
if (msg == buf) if (msg == buf)
{ {

7
libi2pd/Destination.cpp

@ -588,9 +588,12 @@ namespace client
i2p::garlic::GarlicDestination::HandleDeliveryStatusMessage (msgID); i2p::garlic::GarlicDestination::HandleDeliveryStatusMessage (msgID);
} }
void LeaseSetDestination::SetLeaseSetUpdated () void LeaseSetDestination::SetLeaseSetUpdated (bool post)
{ {
UpdateLeaseSet (); if (post)
m_Service.post([s = shared_from_this ()]() { s->UpdateLeaseSet (); });
else
UpdateLeaseSet ();
} }
void LeaseSetDestination::Publish () void LeaseSetDestination::Publish ()

18
libi2pd/Destination.h

@ -142,15 +142,15 @@ namespace client
void CancelDestinationRequestWithEncryptedLeaseSet (std::shared_ptr<const i2p::data::BlindedPublicKey> dest, bool notify = true); void CancelDestinationRequestWithEncryptedLeaseSet (std::shared_ptr<const i2p::data::BlindedPublicKey> dest, bool notify = true);
// implements GarlicDestination // implements GarlicDestination
std::shared_ptr<const i2p::data::LocalLeaseSet> GetLeaseSet (); std::shared_ptr<const i2p::data::LocalLeaseSet> GetLeaseSet () override;
std::shared_ptr<i2p::tunnel::TunnelPool> GetTunnelPool () const { return m_Pool; } std::shared_ptr<i2p::tunnel::TunnelPool> GetTunnelPool () const override { return m_Pool; }
// override GarlicDestination // override GarlicDestination
bool SubmitSessionKey (const uint8_t * key, const uint8_t * tag); bool SubmitSessionKey (const uint8_t * key, const uint8_t * tag) override;
void SubmitECIESx25519Key (const uint8_t * key, uint64_t tag); void SubmitECIESx25519Key (const uint8_t * key, uint64_t tag) override;
void ProcessGarlicMessage (std::shared_ptr<I2NPMessage> msg); void ProcessGarlicMessage (std::shared_ptr<I2NPMessage> msg) override;
void ProcessDeliveryStatusMessage (std::shared_ptr<I2NPMessage> msg); void ProcessDeliveryStatusMessage (std::shared_ptr<I2NPMessage> msg) override;
void SetLeaseSetUpdated (); void SetLeaseSetUpdated (bool post) override;
bool IsPublic () const { return m_IsPublic; }; bool IsPublic () const { return m_IsPublic; };
void SetPublic (bool pub) { m_IsPublic = pub; }; void SetPublic (bool pub) { m_IsPublic = pub; };
@ -158,8 +158,8 @@ namespace client
protected: protected:
// implements GarlicDestination // implements GarlicDestination
void HandleI2NPMessage (const uint8_t * buf, size_t len); void HandleI2NPMessage (const uint8_t * buf, size_t len) override;
bool HandleCloveI2NPMessage (I2NPMessageType typeID, const uint8_t * payload, size_t len, uint32_t msgID); bool HandleCloveI2NPMessage (I2NPMessageType typeID, const uint8_t * payload, size_t len, uint32_t msgID) override;
void SetLeaseSet (std::shared_ptr<const i2p::data::LocalLeaseSet> newLeaseSet); void SetLeaseSet (std::shared_ptr<const i2p::data::LocalLeaseSet> newLeaseSet);
int GetLeaseSetType () const { return m_LeaseSetType; }; int GetLeaseSetType () const { return m_LeaseSetType; };

69
libi2pd/Family.cpp

@ -1,5 +1,5 @@
/* /*
* Copyright (c) 2013-2023, The PurpleI2P Project * Copyright (c) 2013-2024, The PurpleI2P Project
* *
* This file is part of Purple i2pd project and licensed under BSD3 * This file is part of Purple i2pd project and licensed under BSD3
* *
@ -7,7 +7,6 @@
*/ */
#include <string.h> #include <string.h>
#include <openssl/evp.h>
#include <openssl/ssl.h> #include <openssl/ssl.h>
#include "Crypto.h" #include "Crypto.h"
#include "FS.h" #include "FS.h"
@ -25,6 +24,8 @@ namespace data
Families::~Families () Families::~Families ()
{ {
for (auto it : m_SigningKeys)
if (it.second.first) EVP_PKEY_free (it.second.first);
} }
void Families::LoadCertificate (const std::string& filename) void Families::LoadCertificate (const std::string& filename)
@ -47,48 +48,16 @@ namespace data
cn += 3; cn += 3;
char * family = strstr (cn, ".family"); char * family = strstr (cn, ".family");
if (family) family[0] = 0; if (family) family[0] = 0;
} auto pkey = X509_get_pubkey (cert);
auto pkey = X509_get_pubkey (cert); if (pkey)
int keyType = EVP_PKEY_base_id (pkey);
switch (keyType)
{
case EVP_PKEY_DSA:
// TODO:
break;
case EVP_PKEY_EC:
{ {
EC_KEY * ecKey = EVP_PKEY_get1_EC_KEY (pkey); if (!m_SigningKeys.emplace (cn, std::make_pair(pkey, (int)m_SigningKeys.size () + 1)).second)
if (ecKey)
{ {
auto group = EC_KEY_get0_group (ecKey); EVP_PKEY_free (pkey);
if (group) LogPrint (eLogError, "Family: Duplicated family name ", cn);
{
int curve = EC_GROUP_get_curve_name (group);
if (curve == NID_X9_62_prime256v1)
{
uint8_t signingKey[64];
BIGNUM * x = BN_new(), * y = BN_new();
EC_POINT_get_affine_coordinates_GFp (group,
EC_KEY_get0_public_key (ecKey), x, y, NULL);
i2p::crypto::bn2buf (x, signingKey, 32);
i2p::crypto::bn2buf (y, signingKey + 32, 32);
BN_free (x); BN_free (y);
verifier = std::make_shared<i2p::crypto::ECDSAP256Verifier>();
verifier->SetPublicKey (signingKey);
}
else
LogPrint (eLogWarning, "Family: elliptic curve ", curve, " is not supported");
}
EC_KEY_free (ecKey);
} }
break;
} }
default:
LogPrint (eLogWarning, "Family: Certificate key type ", keyType, " is not supported");
} }
EVP_PKEY_free (pkey);
if (verifier && cn)
m_SigningKeys.emplace (cn, std::make_pair(verifier, (int)m_SigningKeys.size () + 1));
} }
SSL_free (ssl); SSL_free (ssl);
} }
@ -130,14 +99,22 @@ namespace data
LogPrint (eLogError, "Family: ", family, " is too long"); LogPrint (eLogError, "Family: ", family, " is too long");
return false; return false;
} }
memcpy (buf, family.c_str (), len);
memcpy (buf + len, (const uint8_t *)ident, 32);
len += 32;
Base64ToByteStream (signature, signatureLen, signatureBuf, 64);
auto it = m_SigningKeys.find (family); auto it = m_SigningKeys.find (family);
if (it != m_SigningKeys.end ()) if (it != m_SigningKeys.end () && it->second.first)
return it->second.first->Verify (buf, len, signatureBuf); {
memcpy (buf, family.c_str (), len);
memcpy (buf + len, (const uint8_t *)ident, 32);
len += 32;
auto signatureBufLen = Base64ToByteStream (signature, signatureLen, signatureBuf, 64);
if (signatureBufLen)
{
EVP_MD_CTX * ctx = EVP_MD_CTX_create ();
EVP_DigestVerifyInit (ctx, NULL, NULL, NULL, it->second.first);
auto ret = EVP_DigestVerify (ctx, signatureBuf, signatureBufLen, buf, len);
EVP_MD_CTX_destroy (ctx);
return ret;
}
}
// TODO: process key // TODO: process key
return true; return true;
} }

6
libi2pd/Family.h

@ -1,5 +1,5 @@
/* /*
* Copyright (c) 2013-2022, The PurpleI2P Project * Copyright (c) 2013-2024, The PurpleI2P Project
* *
* This file is part of Purple i2pd project and licensed under BSD3 * This file is part of Purple i2pd project and licensed under BSD3
* *
@ -12,7 +12,7 @@
#include <map> #include <map>
#include <string> #include <string>
#include <memory> #include <memory>
#include "Signature.h" #include <openssl/evp.h>
#include "Identity.h" #include "Identity.h"
namespace i2p namespace i2p
@ -37,7 +37,7 @@ namespace data
private: private:
std::map<std::string, std::pair<std::shared_ptr<i2p::crypto::Verifier>, FamilyID> > m_SigningKeys; // family -> (verifier, id) std::map<std::string, std::pair<EVP_PKEY *, FamilyID> > m_SigningKeys; // family -> (verification pkey, id)
}; };
std::string CreateFamilySignature (const std::string& family, const IdentHash& ident); std::string CreateFamilySignature (const std::string& family, const IdentHash& ident);

2
libi2pd/Garlic.cpp

@ -897,7 +897,7 @@ namespace garlic
} }
} }
void GarlicDestination::SetLeaseSetUpdated () void GarlicDestination::SetLeaseSetUpdated (bool post)
{ {
{ {
std::unique_lock<std::mutex> l(m_SessionsMutex); std::unique_lock<std::mutex> l(m_SessionsMutex);

2
libi2pd/Garlic.h

@ -253,7 +253,7 @@ namespace garlic
virtual void ProcessGarlicMessage (std::shared_ptr<I2NPMessage> msg); virtual void ProcessGarlicMessage (std::shared_ptr<I2NPMessage> msg);
virtual void ProcessDeliveryStatusMessage (std::shared_ptr<I2NPMessage> msg); virtual void ProcessDeliveryStatusMessage (std::shared_ptr<I2NPMessage> msg);
virtual void SetLeaseSetUpdated (); virtual void SetLeaseSetUpdated (bool post = false);
virtual std::shared_ptr<const i2p::data::LocalLeaseSet> GetLeaseSet () = 0; // TODO virtual std::shared_ptr<const i2p::data::LocalLeaseSet> GetLeaseSet () = 0; // TODO
virtual std::shared_ptr<i2p::tunnel::TunnelPool> GetTunnelPool () const = 0; virtual std::shared_ptr<i2p::tunnel::TunnelPool> GetTunnelPool () const = 0;

26
libi2pd/I2NPProtocol.cpp

@ -396,8 +396,26 @@ namespace i2p
return false; return false;
} }
uint8_t retCode = 0; uint8_t retCode = 0;
// decide if we should accept tunnel
bool accept = i2p::context.AcceptsTunnels ();
if (accept)
{
auto congestionLevel = i2p::context.GetCongestionLevel (false);
if (congestionLevel >= CONGESTION_LEVEL_MEDIUM)
{
if (congestionLevel < CONGESTION_LEVEL_FULL)
{
// random reject depending on congestion level
int level = i2p::tunnel::tunnels.GetRng ()() % (CONGESTION_LEVEL_FULL - CONGESTION_LEVEL_MEDIUM) + CONGESTION_LEVEL_MEDIUM;
if (congestionLevel > level)
accept = false;
}
else
accept = false;
}
}
// replace record to reply // replace record to reply
if (i2p::context.AcceptsTunnels () && i2p::context.GetCongestionLevel (false) < CONGESTION_LEVEL_FULL) if (accept)
{ {
auto transitTunnel = i2p::tunnel::CreateTransitTunnel ( auto transitTunnel = i2p::tunnel::CreateTransitTunnel (
bufbe32toh (clearText + ECIES_BUILD_REQUEST_RECORD_RECEIVE_TUNNEL_OFFSET), bufbe32toh (clearText + ECIES_BUILD_REQUEST_RECORD_RECEIVE_TUNNEL_OFFSET),
@ -932,14 +950,8 @@ namespace i2p
void I2NPMessagesHandler::Flush () void I2NPMessagesHandler::Flush ()
{ {
if (!m_TunnelMsgs.empty ()) if (!m_TunnelMsgs.empty ())
{
i2p::tunnel::tunnels.PostTunnelData (m_TunnelMsgs); i2p::tunnel::tunnels.PostTunnelData (m_TunnelMsgs);
m_TunnelMsgs.clear ();
}
if (!m_TunnelGatewayMsgs.empty ()) if (!m_TunnelGatewayMsgs.empty ())
{
i2p::tunnel::tunnels.PostTunnelData (m_TunnelGatewayMsgs); i2p::tunnel::tunnels.PostTunnelData (m_TunnelGatewayMsgs);
m_TunnelGatewayMsgs.clear ();
}
} }
} }

3
libi2pd/I2NPProtocol.h

@ -13,6 +13,7 @@
#include <string.h> #include <string.h>
#include <unordered_set> #include <unordered_set>
#include <memory> #include <memory>
#include <list>
#include <functional> #include <functional>
#include "Crypto.h" #include "Crypto.h"
#include "I2PEndian.h" #include "I2PEndian.h"
@ -328,7 +329,7 @@ namespace tunnel
private: private:
std::vector<std::shared_ptr<I2NPMessage> > m_TunnelMsgs, m_TunnelGatewayMsgs; std::list<std::shared_ptr<I2NPMessage> > m_TunnelMsgs, m_TunnelGatewayMsgs;
}; };
} }

4
libi2pd/NTCP2.cpp

@ -1822,7 +1822,7 @@ namespace transport
LogPrint(eLogError, "NTCP2: HTTP proxy write error ", ec.message()); LogPrint(eLogError, "NTCP2: HTTP proxy write error ", ec.message());
}); });
boost::asio::streambuf * readbuff = new boost::asio::streambuf; auto readbuff = std::make_shared<boost::asio::streambuf>();
boost::asio::async_read_until(conn->GetSocket(), *readbuff, "\r\n\r\n", boost::asio::async_read_until(conn->GetSocket(), *readbuff, "\r\n\r\n",
[readbuff, timer, conn] (const boost::system::error_code & ec, std::size_t transferred) [readbuff, timer, conn] (const boost::system::error_code & ec, std::size_t transferred)
{ {
@ -1842,7 +1842,6 @@ namespace transport
{ {
timer->cancel(); timer->cancel();
conn->ClientLogin(); conn->ClientLogin();
delete readbuff;
return; return;
} }
else else
@ -1852,7 +1851,6 @@ namespace transport
LogPrint(eLogError, "NTCP2: HTTP proxy gave malformed response"); LogPrint(eLogError, "NTCP2: HTTP proxy gave malformed response");
timer->cancel(); timer->cancel();
conn->Terminate(); conn->Terminate();
delete readbuff;
} }
}); });
break; break;

85
libi2pd/NetDb.cpp

@ -122,16 +122,18 @@ namespace data
uint64_t lastProfilesCleanup = i2p::util::GetMonotonicMilliseconds (), lastObsoleteProfilesCleanup = lastProfilesCleanup; uint64_t lastProfilesCleanup = i2p::util::GetMonotonicMilliseconds (), lastObsoleteProfilesCleanup = lastProfilesCleanup;
int16_t profilesCleanupVariance = 0, obsoleteProfilesCleanVariance = 0; int16_t profilesCleanupVariance = 0, obsoleteProfilesCleanVariance = 0;
std::list<std::shared_ptr<const I2NPMessage> > msgs;
while (m_IsRunning) while (m_IsRunning)
{ {
try try
{ {
auto msg = m_Queue.GetNextWithTimeout (1000); // 1 sec if (m_Queue.Wait (1,0)) // 1 sec
if (msg)
{ {
int numMsgs = 0; m_Queue.GetWholeQueue (msgs);
while (msg) while (!msgs.empty ())
{ {
auto msg = msgs.front (); msgs.pop_front ();
if (!msg) continue;
LogPrint(eLogDebug, "NetDb: Got request with type ", (int) msg->GetTypeID ()); LogPrint(eLogDebug, "NetDb: Got request with type ", (int) msg->GetTypeID ());
switch (msg->GetTypeID ()) switch (msg->GetTypeID ())
{ {
@ -145,9 +147,6 @@ namespace data
LogPrint (eLogError, "NetDb: Unexpected message type ", (int) msg->GetTypeID ()); LogPrint (eLogError, "NetDb: Unexpected message type ", (int) msg->GetTypeID ());
//i2p::HandleI2NPMessage (msg); //i2p::HandleI2NPMessage (msg);
} }
if (numMsgs > 100) break;
msg = m_Queue.Get ();
numMsgs++;
} }
} }
if (!m_IsRunning) break; if (!m_IsRunning) break;
@ -639,70 +638,74 @@ namespace data
if (checkForExpiration && uptime > i2p::transport::SSU2_TO_INTRODUCER_SESSION_DURATION) // 1 hour if (checkForExpiration && uptime > i2p::transport::SSU2_TO_INTRODUCER_SESSION_DURATION) // 1 hour
expirationTimeout = i2p::context.IsFloodfill () ? NETDB_FLOODFILL_EXPIRATION_TIMEOUT*1000LL : expirationTimeout = i2p::context.IsFloodfill () ? NETDB_FLOODFILL_EXPIRATION_TIMEOUT*1000LL :
NETDB_MIN_EXPIRATION_TIMEOUT*1000LL + (NETDB_MAX_EXPIRATION_TIMEOUT - NETDB_MIN_EXPIRATION_TIMEOUT)*1000LL*NETDB_MIN_ROUTERS/total; NETDB_MIN_EXPIRATION_TIMEOUT*1000LL + (NETDB_MAX_EXPIRATION_TIMEOUT - NETDB_MIN_EXPIRATION_TIMEOUT)*1000LL*NETDB_MIN_ROUTERS/total;
bool isOffline = checkForExpiration && i2p::transport::transports.GetNumPeers () < NETDB_MIN_TRANSPORTS; // enough routers and uptime, but no tranports
std::list<std::pair<std::string, std::shared_ptr<RouterInfo::Buffer> > > saveToDisk; std::list<std::pair<std::string, std::shared_ptr<RouterInfo::Buffer> > > saveToDisk;
std::list<std::string> removeFromDisk; std::list<std::string> removeFromDisk;
auto own = i2p::context.GetSharedRouterInfo (); auto own = i2p::context.GetSharedRouterInfo ();
for (auto& it: m_RouterInfos) for (auto [ident, r]: m_RouterInfos)
{ {
if (!it.second || it.second == own) continue; // skip own if (!r || r == own) continue; // skip own
std::string ident = it.second->GetIdentHashBase64(); if (r->IsBufferScheduledToDelete ()) // from previous SaveUpdated, we assume m_PersistingRouters complete
if (it.second->IsUpdated ())
{ {
if (it.second->GetBuffer ()) std::lock_guard<std::mutex> l(m_RouterInfosMutex); // possible collision between DeleteBuffer and Update
r->DeleteBuffer ();
}
if (r->IsUpdated ())
{
if (r->GetBuffer () && !r->IsUnreachable ())
{ {
// we have something to save // we have something to save
std::shared_ptr<RouterInfo::Buffer> buffer; std::shared_ptr<RouterInfo::Buffer> buffer;
{ {
std::lock_guard<std::mutex> l(m_RouterInfosMutex); // possible collision between DeleteBuffer and Update std::lock_guard<std::mutex> l(m_RouterInfosMutex); // possible collision between DeleteBuffer and Update
buffer = it.second->GetSharedBuffer (); buffer = r->CopyBuffer ();
it.second->DeleteBuffer (); r->ScheduleBufferToDelete ();
} }
if (buffer && !it.second->IsUnreachable ()) // don't save bad router if (buffer)
saveToDisk.push_back(std::make_pair(ident, buffer)); saveToDisk.push_back(std::make_pair(ident.ToBase64 (), buffer));
it.second->SetUnreachable (false);
} }
it.second->SetUpdated (false); r->SetUpdated (false);
updatedCount++; updatedCount++;
continue; continue;
} }
if (it.second->GetProfile ()->IsUnreachable ()) if (r->GetProfile ()->IsUnreachable ())
it.second->SetUnreachable (true); r->SetUnreachable (true);
// make router reachable back if too few routers or floodfills // make router reachable back if too few routers or floodfills
if (it.second->IsUnreachable () && (total - deletedCount < NETDB_MIN_ROUTERS || isLowRate || if (r->IsUnreachable () && (total - deletedCount < NETDB_MIN_ROUTERS || isLowRate || isOffline ||
(it.second->IsFloodfill () && totalFloodfills - deletedFloodfillsCount < NETDB_MIN_FLOODFILLS))) (r->IsFloodfill () && totalFloodfills - deletedFloodfillsCount < NETDB_MIN_FLOODFILLS)))
it.second->SetUnreachable (false); r->SetUnreachable (false);
if (!it.second->IsUnreachable ()) if (!r->IsUnreachable ())
{ {
// find & mark expired routers // find & mark expired routers
if (!it.second->GetCompatibleTransports (true)) // non reachable by any transport if (!r->GetCompatibleTransports (true)) // non reachable by any transport
it.second->SetUnreachable (true); r->SetUnreachable (true);
else if (ts + NETDB_EXPIRATION_TIMEOUT_THRESHOLD*1000LL < it.second->GetTimestamp ()) else if (ts + NETDB_EXPIRATION_TIMEOUT_THRESHOLD*1000LL < r->GetTimestamp ())
{ {
LogPrint (eLogWarning, "NetDb: RouterInfo is from future for ", (it.second->GetTimestamp () - ts)/1000LL, " seconds"); LogPrint (eLogWarning, "NetDb: RouterInfo is from future for ", (r->GetTimestamp () - ts)/1000LL, " seconds");
it.second->SetUnreachable (true); r->SetUnreachable (true);
} }
else if (checkForExpiration) else if (checkForExpiration)
{ {
if (ts > it.second->GetTimestamp () + expirationTimeout) if (ts > r->GetTimestamp () + expirationTimeout)
it.second->SetUnreachable (true); r->SetUnreachable (true);
else if ((ts > it.second->GetTimestamp () + expirationTimeout/2) && // more than half of expiration else if ((ts > r->GetTimestamp () + expirationTimeout/2) && // more than half of expiration
total > NETDB_NUM_ROUTERS_THRESHOLD && !it.second->IsHighBandwidth() && // low bandwidth total > NETDB_NUM_ROUTERS_THRESHOLD && !r->IsHighBandwidth() && // low bandwidth
!it.second->IsFloodfill() && (!i2p::context.IsFloodfill () || // non floodfill !r->IsFloodfill() && (!i2p::context.IsFloodfill () || // non floodfill
(CreateRoutingKey (it.second->GetIdentHash ()) ^ i2p::context.GetIdentHash ()).metric[0] >= 0x02)) // different first 7 bits (CreateRoutingKey (ident) ^ i2p::context.GetIdentHash ()).metric[0] >= 0x02)) // different first 7 bits
it.second->SetUnreachable (true); r->SetUnreachable (true);
} }
} }
// make router reachable back if connected now // make router reachable back if connected now
if (it.second->IsUnreachable () && i2p::transport::transports.IsConnected (it.second->GetIdentHash ())) if (r->IsUnreachable () && i2p::transport::transports.IsConnected (ident))
it.second->SetUnreachable (false); r->SetUnreachable (false);
if (it.second->IsUnreachable ()) if (r->IsUnreachable ())
{ {
if (it.second->IsFloodfill ()) deletedFloodfillsCount++; if (r->IsFloodfill ()) deletedFloodfillsCount++;
// delete RI file // delete RI file
removeFromDisk.push_back (ident); removeFromDisk.push_back (ident.ToBase64());
deletedCount++; deletedCount++;
if (total - deletedCount < NETDB_MIN_ROUTERS) checkForExpiration = false; if (total - deletedCount < NETDB_MIN_ROUTERS) checkForExpiration = false;
} }

1
libi2pd/NetDb.hpp

@ -39,6 +39,7 @@ namespace data
{ {
const int NETDB_MIN_ROUTERS = 90; const int NETDB_MIN_ROUTERS = 90;
const int NETDB_MIN_FLOODFILLS = 5; const int NETDB_MIN_FLOODFILLS = 5;
const int NETDB_MIN_TRANSPORTS = 10 ; // otherwise assume offline
const int NETDB_NUM_FLOODFILLS_THRESHOLD = 1200; const int NETDB_NUM_FLOODFILLS_THRESHOLD = 1200;
const int NETDB_NUM_ROUTERS_THRESHOLD = 4*NETDB_NUM_FLOODFILLS_THRESHOLD; const int NETDB_NUM_ROUTERS_THRESHOLD = 4*NETDB_NUM_FLOODFILLS_THRESHOLD;
const int NETDB_TUNNEL_CREATION_RATE_THRESHOLD = 10; // in % const int NETDB_TUNNEL_CREATION_RATE_THRESHOLD = 10; // in %

8
libi2pd/Profiling.h

@ -11,6 +11,7 @@
#include <memory> #include <memory>
#include <future> #include <future>
#include <boost/asio.hpp>
#include "Identity.h" #include "Identity.h"
namespace i2p namespace i2p
@ -68,6 +69,11 @@ namespace data
bool IsUseful() const; bool IsUseful() const;
bool IsDuplicated () const { return m_IsDuplicated; }; bool IsDuplicated () const { return m_IsDuplicated; };
const boost::asio::ip::udp::endpoint& GetLastEndpoint () const { return m_LastEndpoint; }
void SetLastEndpoint (const boost::asio::ip::udp::endpoint& ep) { m_LastEndpoint = ep; }
bool HasLastEndpoint (bool v4) const { return !m_LastEndpoint.address ().is_unspecified () && m_LastEndpoint.port () &&
((v4 && m_LastEndpoint.address ().is_v4 ()) || (!v4 && m_LastEndpoint.address ().is_v6 ())); }
private: private:
void UpdateTime (); void UpdateTime ();
@ -90,6 +96,8 @@ namespace data
uint32_t m_NumTimesRejected; uint32_t m_NumTimesRejected;
bool m_HasConnected; // successful trusted(incoming or NTCP2) connection bool m_HasConnected; // successful trusted(incoming or NTCP2) connection
bool m_IsDuplicated; bool m_IsDuplicated;
// connectivity
boost::asio::ip::udp::endpoint m_LastEndpoint; // SSU2 for non-published addresses
}; };
std::shared_ptr<RouterProfile> GetRouterProfile (const IdentHash& identHash); std::shared_ptr<RouterProfile> GetRouterProfile (const IdentHash& identHash);

32
libi2pd/Queue.h

@ -1,5 +1,5 @@
/* /*
* Copyright (c) 2013-2020, The PurpleI2P Project * Copyright (c) 2013-2024, The PurpleI2P Project
* *
* This file is part of Purple i2pd project and licensed under BSD3 * This file is part of Purple i2pd project and licensed under BSD3
* *
@ -9,8 +9,7 @@
#ifndef QUEUE_H__ #ifndef QUEUE_H__
#define QUEUE_H__ #define QUEUE_H__
#include <queue> #include <list>
#include <vector>
#include <mutex> #include <mutex>
#include <thread> #include <thread>
#include <condition_variable> #include <condition_variable>
@ -29,18 +28,16 @@ namespace util
void Put (Element e) void Put (Element e)
{ {
std::unique_lock<std::mutex> l(m_QueueMutex); std::unique_lock<std::mutex> l(m_QueueMutex);
m_Queue.push (std::move(e)); m_Queue.push_back (std::move(e));
m_NonEmpty.notify_one (); m_NonEmpty.notify_one ();
} }
template<template<typename, typename...>class Container, typename... R> void Put (std::list<Element>& list)
void Put (const Container<Element, R...>& vec)
{ {
if (!vec.empty ()) if (!list.empty ())
{ {
std::unique_lock<std::mutex> l(m_QueueMutex); std::unique_lock<std::mutex> l(m_QueueMutex);
for (const auto& it: vec) m_Queue.splice (m_Queue.end (), list);
m_Queue.push (std::move(it));
m_NonEmpty.notify_one (); m_NonEmpty.notify_one ();
} }
} }
@ -107,6 +104,19 @@ namespace util
return GetNonThreadSafe (true); return GetNonThreadSafe (true);
} }
void GetWholeQueue (std::list<Element>& queue)
{
if (!queue.empty ())
{
std::list<Element> newQueue;
queue.swap (newQueue);
}
{
std::unique_lock<std::mutex> l(m_QueueMutex);
m_Queue.swap (queue);
}
}
private: private:
Element GetNonThreadSafe (bool peek = false) Element GetNonThreadSafe (bool peek = false)
@ -115,7 +125,7 @@ namespace util
{ {
auto el = m_Queue.front (); auto el = m_Queue.front ();
if (!peek) if (!peek)
m_Queue.pop (); m_Queue.pop_front ();
return el; return el;
} }
return nullptr; return nullptr;
@ -123,7 +133,7 @@ namespace util
private: private:
std::queue<Element> m_Queue; std::list<Element> m_Queue;
std::mutex m_QueueMutex; std::mutex m_QueueMutex;
std::condition_variable m_NonEmpty; std::condition_variable m_NonEmpty;
}; };

2
libi2pd/RouterContext.cpp

@ -1489,7 +1489,7 @@ namespace i2p
void RouterContext::UpdateCongestion () void RouterContext::UpdateCongestion ()
{ {
auto c = i2p::data::RouterInfo::eLowCongestion; auto c = i2p::data::RouterInfo::eLowCongestion;
if (!AcceptsTunnels () || !m_ShareRatio) if (!AcceptsTunnels () || !m_ShareRatio || (m_Error == eRouterErrorSymmetricNAT && !SupportsV6 () && !SupportsMesh ()))
c = i2p::data::RouterInfo::eRejectAll; c = i2p::data::RouterInfo::eRejectAll;
else else
{ {

22
libi2pd/RouterContext.h

@ -146,7 +146,6 @@ namespace garlic
void SetNetID (int netID) { m_NetID = netID; }; void SetNetID (int netID) { m_NetID = netID; };
bool DecryptTunnelBuildRecord (const uint8_t * encrypted, uint8_t * data); bool DecryptTunnelBuildRecord (const uint8_t * encrypted, uint8_t * data);
bool DecryptTunnelShortRequestRecord (const uint8_t * encrypted, uint8_t * data); bool DecryptTunnelShortRequestRecord (const uint8_t * encrypted, uint8_t * data);
void SubmitECIESx25519Key (const uint8_t * key, uint64_t tag);
void UpdatePort (int port); // called from Daemon void UpdatePort (int port); // called from Daemon
void UpdateAddress (const boost::asio::ip::address& host); // called from SSU2 or Daemon void UpdateAddress (const boost::asio::ip::address& host); // called from SSU2 or Daemon
@ -186,24 +185,24 @@ namespace garlic
void UpdateTimestamp (uint64_t ts); // in seconds, called from NetDb before publishing void UpdateTimestamp (uint64_t ts); // in seconds, called from NetDb before publishing
// implements LocalDestination // implements LocalDestination
std::shared_ptr<const i2p::data::IdentityEx> GetIdentity () const { return m_Keys.GetPublic (); }; std::shared_ptr<const i2p::data::IdentityEx> GetIdentity () const override{ return m_Keys.GetPublic (); };
bool Decrypt (const uint8_t * encrypted, uint8_t * data, i2p::data::CryptoKeyType preferredCrypto) const; bool Decrypt (const uint8_t * encrypted, uint8_t * data, i2p::data::CryptoKeyType preferredCrypto) const override;
void Sign (const uint8_t * buf, int len, uint8_t * signature) const { m_Keys.Sign (buf, len, signature); }; void SetLeaseSetUpdated (bool post) override {};
void SetLeaseSetUpdated () {};
// implements GarlicDestination // implements GarlicDestination
std::shared_ptr<const i2p::data::LocalLeaseSet> GetLeaseSet () { return nullptr; }; std::shared_ptr<const i2p::data::LocalLeaseSet> GetLeaseSet () override { return nullptr; };
std::shared_ptr<i2p::tunnel::TunnelPool> GetTunnelPool () const; std::shared_ptr<i2p::tunnel::TunnelPool> GetTunnelPool () const override;
// override GarlicDestination // override GarlicDestination
void ProcessGarlicMessage (std::shared_ptr<I2NPMessage> msg); void ProcessGarlicMessage (std::shared_ptr<I2NPMessage> msg) override;
void ProcessDeliveryStatusMessage (std::shared_ptr<I2NPMessage> msg); void ProcessDeliveryStatusMessage (std::shared_ptr<I2NPMessage> msg) override;
void SubmitECIESx25519Key (const uint8_t * key, uint64_t tag) override;
protected: protected:
// implements GarlicDestination // implements GarlicDestination
void HandleI2NPMessage (const uint8_t * buf, size_t len); void HandleI2NPMessage (const uint8_t * buf, size_t len) override;
bool HandleCloveI2NPMessage (I2NPMessageType typeID, const uint8_t * payload, size_t len, uint32_t msgID); bool HandleCloveI2NPMessage (I2NPMessageType typeID, const uint8_t * payload, size_t len, uint32_t msgID) override;
private: private:
@ -216,6 +215,7 @@ namespace garlic
void UpdateSSU2Keys (); void UpdateSSU2Keys ();
bool Load (); bool Load ();
void SaveKeys (); void SaveKeys ();
void Sign (const uint8_t * buf, int len, uint8_t * signature) const { m_Keys.Sign (buf, len, signature); };
uint16_t SelectRandomPort () const; uint16_t SelectRandomPort () const;
void PublishNTCP2Address (std::shared_ptr<i2p::data::RouterInfo::Address> address, int port, bool publish) const; void PublishNTCP2Address (std::shared_ptr<i2p::data::RouterInfo::Address> address, int port, bool publish) const;

8
libi2pd/RouterInfo.cpp

@ -45,8 +45,9 @@ namespace data
RouterInfo::RouterInfo (const std::string& fullPath): RouterInfo::RouterInfo (const std::string& fullPath):
m_FamilyID (0), m_IsUpdated (false), m_IsUnreachable (false), m_IsFloodfill (false), m_FamilyID (0), m_IsUpdated (false), m_IsUnreachable (false), m_IsFloodfill (false),
m_SupportedTransports (0),m_ReachableTransports (0), m_PublishedTransports (0), m_IsBufferScheduledToDelete (false), m_SupportedTransports (0),
m_Caps (0), m_Version (0), m_Congestion (eLowCongestion) m_ReachableTransports (0), m_PublishedTransports (0), m_Caps (0), m_Version (0),
m_Congestion (eLowCongestion)
{ {
m_Addresses = AddressesPtr(new Addresses ()); // create empty list m_Addresses = AddressesPtr(new Addresses ()); // create empty list
m_Buffer = RouterInfo::NewBuffer (); // always RouterInfo's m_Buffer = RouterInfo::NewBuffer (); // always RouterInfo's
@ -55,7 +56,7 @@ namespace data
RouterInfo::RouterInfo (std::shared_ptr<Buffer>&& buf, size_t len): RouterInfo::RouterInfo (std::shared_ptr<Buffer>&& buf, size_t len):
m_FamilyID (0), m_IsUpdated (true), m_IsUnreachable (false), m_IsFloodfill (false), m_FamilyID (0), m_IsUpdated (true), m_IsUnreachable (false), m_IsFloodfill (false),
m_SupportedTransports (0), m_ReachableTransports (0), m_PublishedTransports (0), m_IsBufferScheduledToDelete (false), m_SupportedTransports (0), m_ReachableTransports (0), m_PublishedTransports (0),
m_Caps (0), m_Version (0), m_Congestion (eLowCongestion) m_Caps (0), m_Version (0), m_Congestion (eLowCongestion)
{ {
if (len <= MAX_RI_BUFFER_SIZE) if (len <= MAX_RI_BUFFER_SIZE)
@ -1140,6 +1141,7 @@ namespace data
if (len > m_Buffer->size ()) len = m_Buffer->size (); if (len > m_Buffer->size ()) len = m_Buffer->size ();
memcpy (m_Buffer->data (), buf, len); memcpy (m_Buffer->data (), buf, len);
m_Buffer->SetBufferLen (len); m_Buffer->SetBufferLen (len);
m_IsBufferScheduledToDelete = false;
} }
std::shared_ptr<RouterInfo::Buffer> RouterInfo::CopyBuffer () const std::shared_ptr<RouterInfo::Buffer> RouterInfo::CopyBuffer () const

6
libi2pd/RouterInfo.h

@ -290,9 +290,11 @@ namespace data
const uint8_t * GetBuffer () const { return m_Buffer ? m_Buffer->data () : nullptr; }; const uint8_t * GetBuffer () const { return m_Buffer ? m_Buffer->data () : nullptr; };
const uint8_t * LoadBuffer (const std::string& fullPath); // load if necessary const uint8_t * LoadBuffer (const std::string& fullPath); // load if necessary
size_t GetBufferLen () const { return m_Buffer ? m_Buffer->GetBufferLen () : 0; }; size_t GetBufferLen () const { return m_Buffer ? m_Buffer->GetBufferLen () : 0; };
void DeleteBuffer () { m_Buffer = nullptr; }; void DeleteBuffer () { m_Buffer = nullptr; m_IsBufferScheduledToDelete = false; };
std::shared_ptr<Buffer> GetSharedBuffer () const { return m_Buffer; }; std::shared_ptr<Buffer> GetSharedBuffer () const { return m_Buffer; };
std::shared_ptr<Buffer> CopyBuffer () const; std::shared_ptr<Buffer> CopyBuffer () const;
void ScheduleBufferToDelete () { m_IsBufferScheduledToDelete = true; };
bool IsBufferScheduledToDelete () const { return m_IsBufferScheduledToDelete; };
bool IsUpdated () const { return m_IsUpdated; }; bool IsUpdated () const { return m_IsUpdated; };
void SetUpdated (bool updated) { m_IsUpdated = updated; }; void SetUpdated (bool updated) { m_IsUpdated = updated; };
@ -354,7 +356,7 @@ namespace data
#else #else
AddressesPtr m_Addresses; AddressesPtr m_Addresses;
#endif #endif
bool m_IsUpdated, m_IsUnreachable, m_IsFloodfill; bool m_IsUpdated, m_IsUnreachable, m_IsFloodfill, m_IsBufferScheduledToDelete;
CompatibleTransports m_SupportedTransports, m_ReachableTransports, m_PublishedTransports; CompatibleTransports m_SupportedTransports, m_ReachableTransports, m_PublishedTransports;
uint8_t m_Caps; uint8_t m_Caps;
char m_BandwidthCap; char m_BandwidthCap;

177
libi2pd/SSU2.cpp

@ -157,6 +157,9 @@ namespace transport
m_IntroducersV6.clear (); m_IntroducersV6.clear ();
m_ConnectedRecently.clear (); m_ConnectedRecently.clear ();
m_RequestedPeerTests.clear (); m_RequestedPeerTests.clear ();
m_PacketsPool.ReleaseMt (m_ReceivedPacketsQueue);
m_ReceivedPacketsQueue.clear ();
} }
void SSU2Server::SetLocalAddress (const boost::asio::ip::address& localAddress) void SSU2Server::SetLocalAddress (const boost::asio::ip::address& localAddress)
@ -213,15 +216,16 @@ namespace transport
return ep.port (); return ep.port ();
} }
bool SSU2Server::IsConnectedRecently (const boost::asio::ip::udp::endpoint& ep) bool SSU2Server::IsConnectedRecently (const boost::asio::ip::udp::endpoint& ep, bool max)
{ {
if (!ep.port () || ep.address ().is_unspecified ()) return false; if (!ep.port () || ep.address ().is_unspecified ()) return false;
std::lock_guard<std::mutex> l(m_ConnectedRecentlyMutex);
auto it = m_ConnectedRecently.find (ep); auto it = m_ConnectedRecently.find (ep);
if (it != m_ConnectedRecently.end ()) if (it != m_ConnectedRecently.end ())
{ {
if (i2p::util::GetSecondsSinceEpoch () <= it->second + SSU2_HOLE_PUNCH_EXPIRATION) if (i2p::util::GetSecondsSinceEpoch () <= it->second + (max ? SSU2_MAX_HOLE_PUNCH_EXPIRATION : SSU2_MIN_HOLE_PUNCH_EXPIRATION))
return true; return true;
else else if (max)
m_ConnectedRecently.erase (it); m_ConnectedRecently.erase (it);
} }
return false; return false;
@ -230,7 +234,8 @@ namespace transport
void SSU2Server::AddConnectedRecently (const boost::asio::ip::udp::endpoint& ep, uint64_t ts) void SSU2Server::AddConnectedRecently (const boost::asio::ip::udp::endpoint& ep, uint64_t ts)
{ {
if (!ep.port () || ep.address ().is_unspecified () || if (!ep.port () || ep.address ().is_unspecified () ||
i2p::util::GetSecondsSinceEpoch () > ts + SSU2_HOLE_PUNCH_EXPIRATION) return; i2p::util::GetSecondsSinceEpoch () > ts + SSU2_MAX_HOLE_PUNCH_EXPIRATION) return;
std::lock_guard<std::mutex> l(m_ConnectedRecentlyMutex);
auto [it, added] = m_ConnectedRecently.try_emplace (ep, ts); auto [it, added] = m_ConnectedRecently.try_emplace (ep, ts);
if (!added && ts > it->second) if (!added && ts > it->second)
it->second = ts; // renew timestamp of existing endpoint it->second = ts; // renew timestamp of existing endpoint
@ -369,9 +374,9 @@ namespace transport
size_t moreBytes = socket.available (ec); size_t moreBytes = socket.available (ec);
if (!ec && moreBytes) if (!ec && moreBytes)
{ {
auto packets = m_PacketsArrayPool.AcquireMt (); std::list<Packet *> packets;
packets->AddPacket (packet); packets.push_back (packet);
while (moreBytes && packets->numPackets < SSU2_MAX_NUM_PACKETS_PER_BATCH) while (moreBytes && packets.size () < SSU2_MAX_NUM_PACKETS_PER_BATCH)
{ {
packet = m_PacketsPool.AcquireMt (); packet = m_PacketsPool.AcquireMt ();
packet->len = socket.receive_from (boost::asio::buffer (packet->buf, SSU2_MAX_PACKET_SIZE), packet->from, 0, ec); packet->len = socket.receive_from (boost::asio::buffer (packet->buf, SSU2_MAX_PACKET_SIZE), packet->from, 0, ec);
@ -379,13 +384,7 @@ namespace transport
{ {
i2p::transport::transports.UpdateReceivedBytes (packet->len); i2p::transport::transports.UpdateReceivedBytes (packet->len);
if (packet->len >= SSU2_MIN_RECEIVED_PACKET_SIZE) if (packet->len >= SSU2_MIN_RECEIVED_PACKET_SIZE)
{ packets.push_back (packet);
if (!packets->AddPacket (packet))
{
LogPrint (eLogError, "SSU2: Received packets array is full");
m_PacketsPool.ReleaseMt (packet);
}
}
else // drop too short packets else // drop too short packets
m_PacketsPool.ReleaseMt (packet); m_PacketsPool.ReleaseMt (packet);
moreBytes = socket.available(ec); moreBytes = socket.available(ec);
@ -398,10 +397,10 @@ namespace transport
break; break;
} }
} }
GetService ().post (std::bind (&SSU2Server::HandleReceivedPackets, this, packets)); InsertToReceivedPacketsQueue (packets);
} }
else else
GetService ().post (std::bind (&SSU2Server::HandleReceivedPacket, this, packet)); InsertToReceivedPacketsQueue (packet);
Receive (socket); Receive (socket);
} }
else else
@ -428,49 +427,68 @@ namespace transport
} }
} }
void SSU2Server::HandleReceivedPacket (Packet * packet) void SSU2Server::HandleReceivedPackets (std::list<Packet *>&& packets)
{
if (packets.empty ()) return;
if (m_IsThroughProxy)
for (auto it: packets)
ProcessNextPacketFromProxy (it->buf, it->len);
else
for (auto it: packets)
ProcessNextPacket (it->buf, it->len, it->from);
m_PacketsPool.ReleaseMt (packets);
if (m_LastSession && m_LastSession->GetState () != eSSU2SessionStateTerminated)
m_LastSession->FlushData ();
}
void SSU2Server::InsertToReceivedPacketsQueue (Packet * packet)
{ {
if (packet) if (!packet) return;
bool empty = false;
{ {
if (m_IsThroughProxy) std::lock_guard<std::mutex> l(m_ReceivedPacketsQueueMutex);
ProcessNextPacketFromProxy (packet->buf, packet->len); empty = m_ReceivedPacketsQueue.empty ();
else m_ReceivedPacketsQueue.push_back (packet);
ProcessNextPacket (packet->buf, packet->len, packet->from);
m_PacketsPool.ReleaseMt (packet);
if (m_LastSession && m_LastSession->GetState () != eSSU2SessionStateTerminated)
m_LastSession->FlushData ();
} }
if (empty)
GetService ().post([this]() { HandleReceivedPacketsQueue (); });
} }
void SSU2Server::HandleReceivedPackets (Packets * packets) void SSU2Server::InsertToReceivedPacketsQueue (std::list<Packet *>& packets)
{ {
if (!packets) return; if (packets.empty ()) return;
if (m_IsThroughProxy) bool empty = false;
for (size_t i = 0; i < packets->numPackets; i++) {
{ std::lock_guard<std::mutex> l(m_ReceivedPacketsQueueMutex);
auto& packet = (*packets)[i]; empty = m_ReceivedPacketsQueue.empty ();
ProcessNextPacketFromProxy (packet->buf, packet->len); m_ReceivedPacketsQueue.splice (m_ReceivedPacketsQueue.end (), packets);
} }
else if (empty)
for (size_t i = 0; i < packets->numPackets; i++) GetService ().post([this]() { HandleReceivedPacketsQueue (); });
{
auto& packet = (*packets)[i];
ProcessNextPacket (packet->buf, packet->len, packet->from);
}
m_PacketsPool.ReleaseMt (packets->data (), packets->numPackets);
m_PacketsArrayPool.ReleaseMt (packets);
if (m_LastSession && m_LastSession->GetState () != eSSU2SessionStateTerminated)
m_LastSession->FlushData ();
} }
void SSU2Server::AddSession (std::shared_ptr<SSU2Session> session) void SSU2Server::HandleReceivedPacketsQueue ()
{
std::list<Packet *> receivedPackets;
{
std::lock_guard<std::mutex> l(m_ReceivedPacketsQueueMutex);
m_ReceivedPacketsQueue.swap (receivedPackets);
}
HandleReceivedPackets (std::move (receivedPackets));
}
bool SSU2Server::AddSession (std::shared_ptr<SSU2Session> session)
{ {
if (session) if (session)
{ {
m_Sessions.emplace (session->GetConnID (), session); if (m_Sessions.emplace (session->GetConnID (), session).second)
if (session->GetState () != eSSU2SessionStatePeerTest) {
AddSessionByRouterHash (session); if (session->GetState () != eSSU2SessionStatePeerTest)
AddSessionByRouterHash (session);
return true;
}
} }
return false;
} }
void SSU2Server::RemoveSession (uint64_t connID) void SSU2Server::RemoveSession (uint64_t connID)
@ -703,6 +721,9 @@ namespace transport
m_LastSession->SetRemoteEndpoint (senderEndpoint); m_LastSession->SetRemoteEndpoint (senderEndpoint);
m_LastSession->ProcessPeerTest (buf, len); m_LastSession->ProcessPeerTest (buf, len);
break; break;
case eSSU2SessionStateHolePunch:
m_LastSession->ProcessFirstIncomingMessage (connID, buf, len); // SessionRequest
break;
case eSSU2SessionStateClosing: case eSSU2SessionStateClosing:
m_LastSession->ProcessData (buf, len, senderEndpoint); // we might receive termintaion block m_LastSession->ProcessData (buf, len, senderEndpoint); // we might receive termintaion block
if (m_LastSession && m_LastSession->GetState () == eSSU2SessionStateClosing) if (m_LastSession && m_LastSession->GetState () == eSSU2SessionStateClosing)
@ -816,6 +837,29 @@ namespace transport
} }
} }
bool SSU2Server::CheckPendingOutgoingSession (const boost::asio::ip::udp::endpoint& ep, bool peerTest)
{
auto s = FindPendingOutgoingSession (ep);
if (s)
{
if (peerTest)
{
// if peer test requested add it to the list for pending session
auto onEstablished = s->GetOnEstablished ();
if (onEstablished)
s->SetOnEstablished ([s, onEstablished]()
{
onEstablished ();
s->SendPeerTest ();
});
else
s->SetOnEstablished ([s]() { s->SendPeerTest (); });
}
return true;
}
return false;
}
bool SSU2Server::CreateSession (std::shared_ptr<const i2p::data::RouterInfo> router, bool SSU2Server::CreateSession (std::shared_ptr<const i2p::data::RouterInfo> router,
std::shared_ptr<const i2p::data::RouterInfo::Address> address, bool peerTest) std::shared_ptr<const i2p::data::RouterInfo::Address> address, bool peerTest)
{ {
@ -835,34 +879,28 @@ namespace transport
if (isValidEndpoint) if (isValidEndpoint)
{ {
if (i2p::transport::transports.IsInReservedRange(address->host)) return false; if (i2p::transport::transports.IsInReservedRange(address->host)) return false;
auto s = FindPendingOutgoingSession (boost::asio::ip::udp::endpoint (address->host, address->port)); if (CheckPendingOutgoingSession (boost::asio::ip::udp::endpoint (address->host, address->port), peerTest)) return false;
if (s)
{
if (peerTest)
{
// if peer test requested add it to the list for pending session
auto onEstablished = s->GetOnEstablished ();
if (onEstablished)
s->SetOnEstablished ([s, onEstablished]()
{
onEstablished ();
s->SendPeerTest ();
});
else
s->SetOnEstablished ([s]() { s->SendPeerTest (); });
}
return false;
}
} }
auto session = std::make_shared<SSU2Session> (*this, router, address); auto session = std::make_shared<SSU2Session> (*this, router, address);
if (!isValidEndpoint && router->GetProfile ()->HasLastEndpoint (address->IsV4 ()))
{
// router doesn't publish endpoint, but we connected before and hole punch might be alive
auto ep = router->GetProfile ()->GetLastEndpoint ();
if (IsConnectedRecently (ep, false))
{
if (CheckPendingOutgoingSession (ep, peerTest)) return false;
session->SetRemoteEndpoint (ep);
isValidEndpoint = true;
}
}
if (peerTest) if (peerTest)
session->SetOnEstablished ([session]() {session->SendPeerTest (); }); session->SetOnEstablished ([session]() {session->SendPeerTest (); });
if (address->UsesIntroducer ()) if (isValidEndpoint) // we know endpoint
GetService ().post (std::bind (&SSU2Server::ConnectThroughIntroducer, this, session));
else if (isValidEndpoint) // we can't connect without endpoint
GetService ().post ([session]() { session->Connect (); }); GetService ().post ([session]() { session->Connect (); });
else if (address->UsesIntroducer ()) // we don't know endpoint yet
GetService ().post (std::bind (&SSU2Server::ConnectThroughIntroducer, this, session));
else else
return false; return false;
} }
@ -1112,7 +1150,7 @@ namespace transport
for (auto it = m_ConnectedRecently.begin (); it != m_ConnectedRecently.end (); ) for (auto it = m_ConnectedRecently.begin (); it != m_ConnectedRecently.end (); )
{ {
if (ts > it->second + SSU2_HOLE_PUNCH_EXPIRATION) if (ts > it->second + SSU2_MAX_HOLE_PUNCH_EXPIRATION)
it = m_ConnectedRecently.erase (it); it = m_ConnectedRecently.erase (it);
else else
it++; it++;
@ -1138,7 +1176,6 @@ namespace transport
} }
m_PacketsPool.CleanUpMt (); m_PacketsPool.CleanUpMt ();
m_PacketsArrayPool.CleanUpMt ();
m_SentPacketsPool.CleanUp (); m_SentPacketsPool.CleanUp ();
m_IncompleteMessagesPool.CleanUp (); m_IncompleteMessagesPool.CleanUp ();
m_FragmentsPool.CleanUp (); m_FragmentsPool.CleanUp ();

36
libi2pd/SSU2.h

@ -12,11 +12,13 @@
#include <unordered_map> #include <unordered_map>
#include <unordered_set> #include <unordered_set>
#include <vector> #include <vector>
#include <list>
#include <array> #include <array>
#include <mutex> #include <mutex>
#include <random> #include <random>
#include "util.h" #include "util.h"
#include "SSU2Session.h" #include "SSU2Session.h"
#include "SSU2OutOfSession.h"
#include "Socks5.h" #include "Socks5.h"
namespace i2p namespace i2p
@ -40,8 +42,9 @@ namespace transport
const int SSU2_KEEP_ALIVE_INTERVAL = 15; // in seconds const int SSU2_KEEP_ALIVE_INTERVAL = 15; // in seconds
const int SSU2_KEEP_ALIVE_INTERVAL_VARIANCE = 4; // in seconds const int SSU2_KEEP_ALIVE_INTERVAL_VARIANCE = 4; // in seconds
const int SSU2_PROXY_CONNECT_RETRY_TIMEOUT = 30; // in seconds const int SSU2_PROXY_CONNECT_RETRY_TIMEOUT = 30; // in seconds
const int SSU2_HOLE_PUNCH_EXPIRATION = 150; // in seconds const int SSU2_MIN_HOLE_PUNCH_EXPIRATION = 30; // in seconds
const size_t SSU2_MAX_NUM_PACKETS_PER_BATCH = 32; const int SSU2_MAX_HOLE_PUNCH_EXPIRATION = 160; // in seconds
const size_t SSU2_MAX_NUM_PACKETS_PER_BATCH = 64;
class SSU2Server: private i2p::util::RunnableServiceWithWork class SSU2Server: private i2p::util::RunnableServiceWithWork
{ {
@ -52,20 +55,6 @@ namespace transport
boost::asio::ip::udp::endpoint from; boost::asio::ip::udp::endpoint from;
}; };
struct Packets: public std::array<Packet *, SSU2_MAX_NUM_PACKETS_PER_BATCH>
{
size_t numPackets = 0;
bool AddPacket (Packet *p)
{
if (p && numPackets < size ())
{
data()[numPackets] = p; numPackets++;
return true;
}
return false;
}
};
class ReceiveService: public i2p::util::RunnableService class ReceiveService: public i2p::util::RunnableService
{ {
public: public:
@ -89,14 +78,14 @@ namespace transport
bool UsesProxy () const { return m_IsThroughProxy; }; bool UsesProxy () const { return m_IsThroughProxy; };
bool IsSupported (const boost::asio::ip::address& addr) const; bool IsSupported (const boost::asio::ip::address& addr) const;
uint16_t GetPort (bool v4) const; uint16_t GetPort (bool v4) const;
bool IsConnectedRecently (const boost::asio::ip::udp::endpoint& ep); bool IsConnectedRecently (const boost::asio::ip::udp::endpoint& ep, bool max = true);
void AddConnectedRecently (const boost::asio::ip::udp::endpoint& ep, uint64_t ts); void AddConnectedRecently (const boost::asio::ip::udp::endpoint& ep, uint64_t ts);
std::mt19937& GetRng () { return m_Rng; } std::mt19937& GetRng () { return m_Rng; }
bool IsMaxNumIntroducers (bool v4) const { return (v4 ? m_Introducers.size () : m_IntroducersV6.size ()) >= SSU2_MAX_NUM_INTRODUCERS; } bool IsMaxNumIntroducers (bool v4) const { return (v4 ? m_Introducers.size () : m_IntroducersV6.size ()) >= SSU2_MAX_NUM_INTRODUCERS; }
bool IsSyncClockFromPeers () const { return m_IsSyncClockFromPeers; }; bool IsSyncClockFromPeers () const { return m_IsSyncClockFromPeers; };
void AdjustTimeOffset (int64_t offset, std::shared_ptr<const i2p::data::IdentityEx> from); void AdjustTimeOffset (int64_t offset, std::shared_ptr<const i2p::data::IdentityEx> from);
void AddSession (std::shared_ptr<SSU2Session> session); bool AddSession (std::shared_ptr<SSU2Session> session);
void RemoveSession (uint64_t connID); void RemoveSession (uint64_t connID);
void RequestRemoveSession (uint64_t connID); void RequestRemoveSession (uint64_t connID);
void AddSessionByRouterHash (std::shared_ptr<SSU2Session> session); void AddSessionByRouterHash (std::shared_ptr<SSU2Session> session);
@ -144,9 +133,11 @@ namespace transport
void Receive (boost::asio::ip::udp::socket& socket); void Receive (boost::asio::ip::udp::socket& socket);
void HandleReceivedFrom (const boost::system::error_code& ecode, size_t bytes_transferred, void HandleReceivedFrom (const boost::system::error_code& ecode, size_t bytes_transferred,
Packet * packet, boost::asio::ip::udp::socket& socket); Packet * packet, boost::asio::ip::udp::socket& socket);
void HandleReceivedPacket (Packet * packet); void HandleReceivedPackets (std::list<Packet *>&& packets);
void HandleReceivedPackets (Packets * packets);
void ProcessNextPacket (uint8_t * buf, size_t len, const boost::asio::ip::udp::endpoint& senderEndpoint); void ProcessNextPacket (uint8_t * buf, size_t len, const boost::asio::ip::udp::endpoint& senderEndpoint);
void InsertToReceivedPacketsQueue (Packet * packet);
void InsertToReceivedPacketsQueue (std::list<Packet *>& packets);
void HandleReceivedPacketsQueue ();
void ScheduleTermination (); void ScheduleTermination ();
void HandleTerminationTimer (const boost::system::error_code& ecode); void HandleTerminationTimer (const boost::system::error_code& ecode);
@ -157,6 +148,7 @@ namespace transport
void ScheduleResend (bool more); void ScheduleResend (bool more);
void HandleResendTimer (const boost::system::error_code& ecode); void HandleResendTimer (const boost::system::error_code& ecode);
bool CheckPendingOutgoingSession (const boost::asio::ip::udp::endpoint& ep, bool peerTest);
void ConnectThroughIntroducer (std::shared_ptr<SSU2Session> session); void ConnectThroughIntroducer (std::shared_ptr<SSU2Session> session);
std::vector<std::shared_ptr<SSU2Session> > FindIntroducers (int maxNumIntroducers, std::vector<std::shared_ptr<SSU2Session> > FindIntroducers (int maxNumIntroducers,
bool v4, const std::unordered_set<i2p::data::IdentHash>& excluded); bool v4, const std::unordered_set<i2p::data::IdentHash>& excluded);
@ -191,7 +183,6 @@ namespace transport
std::unordered_map<uint32_t, std::pair <std::weak_ptr<SSU2Session>, uint64_t > > m_PeerTests; // nonce->(Alice, timestamp). We are Bob std::unordered_map<uint32_t, std::pair <std::weak_ptr<SSU2Session>, uint64_t > > m_PeerTests; // nonce->(Alice, timestamp). We are Bob
std::list<std::pair<i2p::data::IdentHash, uint32_t> > m_Introducers, m_IntroducersV6; // introducers we are connected to std::list<std::pair<i2p::data::IdentHash, uint32_t> > m_Introducers, m_IntroducersV6; // introducers we are connected to
i2p::util::MemoryPoolMt<Packet> m_PacketsPool; i2p::util::MemoryPoolMt<Packet> m_PacketsPool;
i2p::util::MemoryPoolMt<Packets> m_PacketsArrayPool;
i2p::util::MemoryPool<SSU2SentPacket> m_SentPacketsPool; i2p::util::MemoryPool<SSU2SentPacket> m_SentPacketsPool;
i2p::util::MemoryPool<SSU2IncompleteMessage> m_IncompleteMessagesPool; i2p::util::MemoryPool<SSU2IncompleteMessage> m_IncompleteMessagesPool;
i2p::util::MemoryPool<SSU2IncompleteMessage::Fragment> m_FragmentsPool; i2p::util::MemoryPool<SSU2IncompleteMessage::Fragment> m_FragmentsPool;
@ -204,7 +195,10 @@ namespace transport
std::shared_ptr<const i2p::data::IdentityEx> m_PendingTimeOffsetFrom; std::shared_ptr<const i2p::data::IdentityEx> m_PendingTimeOffsetFrom;
std::mt19937 m_Rng; std::mt19937 m_Rng;
std::map<boost::asio::ip::udp::endpoint, uint64_t> m_ConnectedRecently; // endpoint -> last activity time in seconds std::map<boost::asio::ip::udp::endpoint, uint64_t> m_ConnectedRecently; // endpoint -> last activity time in seconds
mutable std::mutex m_ConnectedRecentlyMutex;
std::unordered_map<uint32_t, std::pair <std::weak_ptr<SSU2PeerTestSession>, uint64_t > > m_RequestedPeerTests; // nonce->(Alice, timestamp) std::unordered_map<uint32_t, std::pair <std::weak_ptr<SSU2PeerTestSession>, uint64_t > > m_RequestedPeerTests; // nonce->(Alice, timestamp)
std::list<Packet *> m_ReceivedPacketsQueue;
mutable std::mutex m_ReceivedPacketsQueueMutex;
// proxy // proxy
bool m_IsThroughProxy; bool m_IsThroughProxy;

355
libi2pd/SSU2OutOfSession.cpp

@ -0,0 +1,355 @@
/*
* Copyright (c) 2024, The PurpleI2P Project
*
* This file is part of Purple i2pd project and licensed under BSD3
*
* See full license text in LICENSE file at top of project tree
*/
#include "Log.h"
#include "SSU2.h"
#include "SSU2OutOfSession.h"
namespace i2p
{
namespace transport
{
SSU2PeerTestSession::SSU2PeerTestSession (SSU2Server& server, uint64_t sourceConnID, uint64_t destConnID):
SSU2Session (server, nullptr, nullptr, false),
m_MsgNumReceived (0), m_NumResends (0),m_IsConnectedRecently (false), m_IsStatusChanged (false),
m_PeerTestResendTimer (server.GetService ())
{
if (!sourceConnID) sourceConnID = ~destConnID;
if (!destConnID) destConnID = ~sourceConnID;
SetSourceConnID (sourceConnID);
SetDestConnID (destConnID);
SetState (eSSU2SessionStatePeerTest);
SetTerminationTimeout (SSU2_PEER_TEST_EXPIRATION_TIMEOUT);
}
bool SSU2PeerTestSession::ProcessPeerTest (uint8_t * buf, size_t len)
{
// we are Alice or Charlie, msgs 5,6,7
Header header;
memcpy (header.buf, buf, 16);
header.ll[0] ^= CreateHeaderMask (i2p::context.GetSSU2IntroKey (), buf + (len - 24));
header.ll[1] ^= CreateHeaderMask (i2p::context.GetSSU2IntroKey (), buf + (len - 12));
if (header.h.type != eSSU2PeerTest)
{
LogPrint (eLogWarning, "SSU2: Unexpected message type ", (int)header.h.type, " instead ", (int)eSSU2PeerTest);
return false;
}
if (len < 48)
{
LogPrint (eLogWarning, "SSU2: PeerTest message too short ", len);
return false;
}
uint8_t nonce[12] = {0};
uint64_t headerX[2]; // sourceConnID, token
i2p::crypto::ChaCha20 (buf + 16, 16, i2p::context.GetSSU2IntroKey (), nonce, (uint8_t *)headerX);
SetDestConnID (headerX[0]);
// decrypt and handle payload
uint8_t * payload = buf + 32;
CreateNonce (be32toh (header.h.packetNum), nonce);
uint8_t h[32];
memcpy (h, header.buf, 16);
memcpy (h + 16, &headerX, 16);
if (!i2p::crypto::AEADChaCha20Poly1305 (payload, len - 48, h, 32,
i2p::context.GetSSU2IntroKey (), nonce, payload, len - 48, false))
{
LogPrint (eLogWarning, "SSU2: PeerTest AEAD verification failed ");
return false;
}
HandlePayload (payload, len - 48);
SetIsDataReceived (false);
return true;
}
void SSU2PeerTestSession::HandleAddress (const uint8_t * buf, size_t len)
{
if (!ExtractEndpoint (buf, len, m_OurEndpoint))
LogPrint (eLogWarning, "SSU2: Can't hanlde address block from peer test message");
}
void SSU2PeerTestSession::HandlePeerTest (const uint8_t * buf, size_t len)
{
// msgs 5-7
if (len < 8) return;
uint8_t msg = buf[0];
if (msg <= m_MsgNumReceived)
{
LogPrint (eLogDebug, "SSU2: PeerTest msg num ", msg, " received after ", m_MsgNumReceived, ". Ignored");
return;
}
size_t offset = 3; // points to signed data after msg + code + flag
uint32_t nonce = bufbe32toh (buf + offset + 1); // 1 - ver
switch (msg) // msg
{
case 5: // Alice from Charlie 1
{
if (htobe64 (((uint64_t)nonce << 32) | nonce) == GetSourceConnID ())
{
m_PeerTestResendTimer.cancel (); // calcel delayed msg 6 if any
m_IsConnectedRecently = GetServer ().IsConnectedRecently (GetRemoteEndpoint ());
if (GetAddress ())
{
if (!m_IsConnectedRecently)
SetRouterStatus (eRouterStatusOK);
else if (m_IsStatusChanged && GetRouterStatus () == eRouterStatusFirewalled)
SetRouterStatus (eRouterStatusUnknown);
SendPeerTest (6, buf + offset, len - offset);
}
}
else
LogPrint (eLogWarning, "SSU2: Peer test 5 nonce mismatch ", nonce, " connID=", GetSourceConnID ());
break;
}
case 6: // Charlie from Alice
{
m_PeerTestResendTimer.cancel (); // no more msg 5 resends
if (GetAddress ())
SendPeerTest (7, buf + offset, len - offset);
else
LogPrint (eLogWarning, "SSU2: Unknown address for peer test 6");
GetServer ().AddConnectedRecently (GetRemoteEndpoint (), i2p::util::GetSecondsSinceEpoch ());
GetServer ().RequestRemoveSession (GetConnID ());
break;
}
case 7: // Alice from Charlie 2
{
m_PeerTestResendTimer.cancel (); // no more msg 6 resends
if (m_MsgNumReceived < 5 && m_OurEndpoint.port ()) // msg 5 was not received
{
if (m_OurEndpoint.address ().is_v4 ()) // ipv4
{
if (i2p::context.GetStatus () == eRouterStatusFirewalled)
{
if (m_OurEndpoint.port () != GetServer ().GetPort (true))
i2p::context.SetError (eRouterErrorSymmetricNAT);
else if (i2p::context.GetError () == eRouterErrorSymmetricNAT)
i2p::context.SetError (eRouterErrorNone);
}
}
else
{
if (i2p::context.GetStatusV6 () == eRouterStatusFirewalled)
{
if (m_OurEndpoint.port () != GetServer ().GetPort (false))
i2p::context.SetErrorV6 (eRouterErrorSymmetricNAT);
else if (i2p::context.GetErrorV6 () == eRouterErrorSymmetricNAT)
i2p::context.SetErrorV6 (eRouterErrorNone);
}
}
}
GetServer ().AddConnectedRecently (GetRemoteEndpoint (), i2p::util::GetSecondsSinceEpoch ());
GetServer ().RequestRemoveSession (GetConnID ());
break;
}
default:
LogPrint (eLogWarning, "SSU2: PeerTest unexpected msg num ", msg);
return;
}
m_MsgNumReceived = msg;
}
void SSU2PeerTestSession::SendPeerTest (uint8_t msg)
{
auto addr = GetAddress ();
if (!addr) return;
Header header;
uint8_t h[32], payload[SSU2_MAX_PACKET_SIZE];
// fill packet
header.h.connID = GetDestConnID (); // dest id
RAND_bytes (header.buf + 8, 4); // random packet num
header.h.type = eSSU2PeerTest;
header.h.flags[0] = 2; // ver
header.h.flags[1] = (uint8_t)i2p::context.GetNetID (); // netID
header.h.flags[2] = 0; // flag
memcpy (h, header.buf, 16);
htobuf64 (h + 16, GetSourceConnID ()); // source id
// payload
payload[0] = eSSU2BlkDateTime;
htobe16buf (payload + 1, 4);
htobe32buf (payload + 3, (i2p::util::GetMillisecondsSinceEpoch () + 500)/1000);
size_t payloadSize = 7;
if (msg == 6 || msg == 7)
payloadSize += CreateAddressBlock (payload + payloadSize, GetMaxPayloadSize () - payloadSize, GetRemoteEndpoint ());
payloadSize += CreatePeerTestBlock (payload + payloadSize, GetMaxPayloadSize () - payloadSize,
msg, eSSU2PeerTestCodeAccept, nullptr, m_SignedData.data (), m_SignedData.size ());
payloadSize += CreatePaddingBlock (payload + payloadSize, GetMaxPayloadSize () - payloadSize);
// encrypt
uint8_t n[12];
CreateNonce (be32toh (header.h.packetNum), n);
i2p::crypto::AEADChaCha20Poly1305 (payload, payloadSize, h, 32, addr->i, n, payload, payloadSize + 16, true);
payloadSize += 16;
header.ll[0] ^= CreateHeaderMask (addr->i, payload + (payloadSize - 24));
header.ll[1] ^= CreateHeaderMask (addr->i, payload + (payloadSize - 12));
memset (n, 0, 12);
i2p::crypto::ChaCha20 (h + 16, 16, addr->i, n, h + 16);
// send
GetServer ().Send (header.buf, 16, h + 16, 16, payload, payloadSize, GetRemoteEndpoint ());
}
void SSU2PeerTestSession::SendPeerTest (uint8_t msg, const uint8_t * signedData, size_t signedDataLen, bool delayed)
{
#if __cplusplus >= 202002L // C++20
m_SignedData.assign (signedData, signedData + signedDataLen);
#else
m_SignedData.resize (signedDataLen);
memcpy (m_SignedData.data (), signedData, signedDataLen);
#endif
if (!delayed)
SendPeerTest (msg);
// schedule resend for msgs 5 or 6
if (msg == 5 || msg == 6)
ScheduleResend (msg);
}
void SSU2PeerTestSession::SendPeerTest (uint8_t msg, const uint8_t * signedData, size_t signedDataLen,
std::shared_ptr<const i2p::data::RouterInfo::Address> addr, bool delayed)
{
if (!addr) return;
SetAddress (addr);
SendPeerTest (msg, signedData, signedDataLen, delayed);
}
void SSU2PeerTestSession::Connect ()
{
LogPrint (eLogError, "SSU2: Can't connect peer test session");
}
bool SSU2PeerTestSession::ProcessFirstIncomingMessage (uint64_t connID, uint8_t * buf, size_t len)
{
LogPrint (eLogError, "SSU2: Can't handle incoming message in peer test session");
return false;
}
void SSU2PeerTestSession::ScheduleResend (uint8_t msg)
{
if (m_NumResends < SSU2_PEER_TEST_MAX_NUM_RESENDS)
{
m_PeerTestResendTimer.expires_from_now (boost::posix_time::milliseconds(
SSU2_PEER_TEST_RESEND_INTERVAL + GetServer ().GetRng ()() % SSU2_PEER_TEST_RESEND_INTERVAL_VARIANCE));
std::weak_ptr<SSU2PeerTestSession> s(std::static_pointer_cast<SSU2PeerTestSession>(shared_from_this ()));
m_PeerTestResendTimer.async_wait ([s, msg](const boost::system::error_code& ecode)
{
if (ecode != boost::asio::error::operation_aborted)
{
auto s1 = s.lock ();
if (s1)
{
if (msg > s1->m_MsgNumReceived)
{
s1->SendPeerTest (msg);
s1->m_NumResends++;
s1->ScheduleResend (msg);
}
}
}
});
}
}
SSU2HolePunchSession::SSU2HolePunchSession (SSU2Server& server, uint32_t nonce,
const boost::asio::ip::udp::endpoint& remoteEndpoint,
std::shared_ptr<const i2p::data::RouterInfo::Address> addr):
SSU2Session (server), // we create full incoming session
m_NumResends (0), m_HolePunchResendTimer (server.GetService ())
{
// we are Charlie
uint64_t destConnID = htobe64 (((uint64_t)nonce << 32) | nonce); // dest id
uint32_t sourceConnID = ~destConnID;
SetSourceConnID (sourceConnID);
SetDestConnID (destConnID);
SetState (eSSU2SessionStateHolePunch);
SetRemoteEndpoint (remoteEndpoint);
SetAddress (addr);
SetTerminationTimeout (SSU2_RELAY_NONCE_EXPIRATION_TIMEOUT);
}
void SSU2HolePunchSession::SendHolePunch ()
{
auto addr = GetAddress ();
if (!addr) return;
auto& ep = GetRemoteEndpoint ();
LogPrint (eLogDebug, "SSU2: Sending HolePunch to ", ep);
Header header;
uint8_t h[32], payload[SSU2_MAX_PACKET_SIZE];
// fill packet
header.h.connID = GetDestConnID (); // dest id
RAND_bytes (header.buf + 8, 4); // random packet num
header.h.type = eSSU2HolePunch;
header.h.flags[0] = 2; // ver
header.h.flags[1] = (uint8_t)i2p::context.GetNetID (); // netID
header.h.flags[2] = 0; // flag
memcpy (h, header.buf, 16);
htobuf64 (h + 16, GetSourceConnID ()); // source id
RAND_bytes (h + 24, 8); // header token, to be ignored by Alice
// payload
payload[0] = eSSU2BlkDateTime;
htobe16buf (payload + 1, 4);
htobe32buf (payload + 3, (i2p::util::GetMillisecondsSinceEpoch () + 500)/1000);
size_t payloadSize = 7;
payloadSize += CreateAddressBlock (payload + payloadSize, GetMaxPayloadSize () - payloadSize, ep);
// relay response block
if (payloadSize + m_RelayResponseBlock.size () < GetMaxPayloadSize ())
{
memcpy (payload + payloadSize, m_RelayResponseBlock.data (), m_RelayResponseBlock.size ());
payloadSize += m_RelayResponseBlock.size ();
}
payloadSize += CreatePaddingBlock (payload + payloadSize, GetMaxPayloadSize () - payloadSize);
// encrypt
uint8_t n[12];
CreateNonce (be32toh (header.h.packetNum), n);
i2p::crypto::AEADChaCha20Poly1305 (payload, payloadSize, h, 32, addr->i, n, payload, payloadSize + 16, true);
payloadSize += 16;
header.ll[0] ^= CreateHeaderMask (addr->i, payload + (payloadSize - 24));
header.ll[1] ^= CreateHeaderMask (addr->i, payload + (payloadSize - 12));
memset (n, 0, 12);
i2p::crypto::ChaCha20 (h + 16, 16, addr->i, n, h + 16);
// send
GetServer ().Send (header.buf, 16, h + 16, 16, payload, payloadSize, ep);
}
void SSU2HolePunchSession::SendHolePunch (const uint8_t * relayResponseBlock, size_t relayResponseBlockLen)
{
#if __cplusplus >= 202002L // C++20
m_RelayResponseBlock.assign (relayResponseBlock, relayResponseBlock + relayResponseBlockLen);
#else
m_RelayResponseBlock.resize (relayResponseBlockLen);
memcpy (m_RelayResponseBlock.data (), relayResponseBlock, relayResponseBlockLen);
#endif
SendHolePunch ();
ScheduleResend ();
}
void SSU2HolePunchSession::ScheduleResend ()
{
if (m_NumResends < SSU2_HOLE_PUNCH_MAX_NUM_RESENDS)
{
m_HolePunchResendTimer.expires_from_now (boost::posix_time::milliseconds(
SSU2_HOLE_PUNCH_RESEND_INTERVAL + GetServer ().GetRng ()() % SSU2_HOLE_PUNCH_RESEND_INTERVAL_VARIANCE));
std::weak_ptr<SSU2HolePunchSession> s(std::static_pointer_cast<SSU2HolePunchSession>(shared_from_this ()));
m_HolePunchResendTimer.async_wait ([s](const boost::system::error_code& ecode)
{
if (ecode != boost::asio::error::operation_aborted)
{
auto s1 = s.lock ();
if (s1 && s1->GetState () == eSSU2SessionStateHolePunch)
{
s1->SendHolePunch ();
s1->m_NumResends++;
s1->ScheduleResend ();
}
}
});
}
}
bool SSU2HolePunchSession::ProcessFirstIncomingMessage (uint64_t connID, uint8_t * buf, size_t len)
{
m_HolePunchResendTimer.cancel ();
return SSU2Session::ProcessFirstIncomingMessage (connID, buf, len);
}
}
}

86
libi2pd/SSU2OutOfSession.h

@ -0,0 +1,86 @@
/*
* Copyright (c) 2024, The PurpleI2P Project
*
* This file is part of Purple i2pd project and licensed under BSD3
*
* See full license text in LICENSE file at top of project tree
*/
#ifndef SSU2_OUT_OF_SESSION_H__
#define SSU2_OUT_OF_SESSION_H__
#include <vector>
#include "SSU2Session.h"
namespace i2p
{
namespace transport
{
const int SSU2_PEER_TEST_RESEND_INTERVAL = 3000; // in milliseconds
const int SSU2_PEER_TEST_RESEND_INTERVAL_VARIANCE = 2000; // in milliseconds
const int SSU2_PEER_TEST_MAX_NUM_RESENDS = 3;
class SSU2PeerTestSession: public SSU2Session // for PeerTest msgs 5,6,7
{
public:
SSU2PeerTestSession (SSU2Server& server, uint64_t sourceConnID, uint64_t destConnID);
uint8_t GetMsgNumReceived () const { return m_MsgNumReceived; }
bool IsConnectedRecently () const { return m_IsConnectedRecently; }
void SetStatusChanged () { m_IsStatusChanged = true; }
void SendPeerTest (uint8_t msg, const uint8_t * signedData, size_t signedDataLen,
std::shared_ptr<const i2p::data::RouterInfo::Address> addr, bool delayed = false);
bool ProcessPeerTest (uint8_t * buf, size_t len) override;
void Connect () override; // outgoing
bool ProcessFirstIncomingMessage (uint64_t connID, uint8_t * buf, size_t len) override; // incoming
private:
void SendPeerTest (uint8_t msg, const uint8_t * signedData, size_t signedDataLen, bool delayed = false); // PeerTest message
void SendPeerTest (uint8_t msg); // send or resend m_SignedData
void HandlePeerTest (const uint8_t * buf, size_t len) override;
void HandleAddress (const uint8_t * buf, size_t len) override;
void ScheduleResend (uint8_t msg);
private:
uint8_t m_MsgNumReceived, m_NumResends;
bool m_IsConnectedRecently, m_IsStatusChanged;
std::vector<uint8_t> m_SignedData; // for resends
boost::asio::deadline_timer m_PeerTestResendTimer;
boost::asio::ip::udp::endpoint m_OurEndpoint; // as seen by peer
};
const int SSU2_HOLE_PUNCH_RESEND_INTERVAL = 1000; // in milliseconds
const int SSU2_HOLE_PUNCH_RESEND_INTERVAL_VARIANCE = 500; // in milliseconds
const int SSU2_HOLE_PUNCH_MAX_NUM_RESENDS = 3;
class SSU2HolePunchSession: public SSU2Session // Charlie
{
public:
SSU2HolePunchSession (SSU2Server& server, uint32_t nonce, const boost::asio::ip::udp::endpoint& remoteEndpoint,
std::shared_ptr<const i2p::data::RouterInfo::Address> addr);
void SendHolePunch (const uint8_t * relayResponseBlock, size_t relayResponseBlockLen);
bool ProcessFirstIncomingMessage (uint64_t connID, uint8_t * buf, size_t len) override; // SessionRequest
private:
void SendHolePunch ();
void ScheduleResend ();
private:
int m_NumResends;
std::vector<uint8_t> m_RelayResponseBlock;
boost::asio::deadline_timer m_HolePunchResendTimer;
};
}
}
#endif

428
libi2pd/SSU2Session.cpp

@ -13,17 +13,12 @@
#include "Gzip.h" #include "Gzip.h"
#include "NetDb.hpp" #include "NetDb.hpp"
#include "SSU2.h" #include "SSU2.h"
#include "SSU2Session.h"
namespace i2p namespace i2p
{ {
namespace transport namespace transport
{ {
static inline void CreateNonce (uint64_t seqn, uint8_t * nonce)
{
memset (nonce, 0, 4);
htole64buf (nonce + 4, seqn);
}
void SSU2IncompleteMessage::AttachNextFragment (const uint8_t * fragment, size_t fragmentSize) void SSU2IncompleteMessage::AttachNextFragment (const uint8_t * fragment, size_t fragmentSize)
{ {
if (msg->len + fragmentSize > msg->maxLen) if (msg->len + fragmentSize > msg->maxLen)
@ -88,7 +83,7 @@ namespace transport
std::shared_ptr<const i2p::data::RouterInfo::Address> addr, bool noise): std::shared_ptr<const i2p::data::RouterInfo::Address> addr, bool noise):
TransportSession (in_RemoteRouter, SSU2_CONNECT_TIMEOUT), TransportSession (in_RemoteRouter, SSU2_CONNECT_TIMEOUT),
m_Server (server), m_Address (addr), m_RemoteTransports (0), m_RemotePeerTestTransports (0), m_Server (server), m_Address (addr), m_RemoteTransports (0), m_RemotePeerTestTransports (0),
m_DestConnID (0), m_SourceConnID (0), m_State (eSSU2SessionStateUnknown), m_RemoteVersion (0), m_DestConnID (0), m_SourceConnID (0), m_State (eSSU2SessionStateUnknown),
m_SendPacketNum (0), m_ReceivePacketNum (0), m_LastDatetimeSentPacketNum (0), m_SendPacketNum (0), m_ReceivePacketNum (0), m_LastDatetimeSentPacketNum (0),
m_IsDataReceived (false), m_RTT (SSU2_UNKNOWN_RTT), m_IsDataReceived (false), m_RTT (SSU2_UNKNOWN_RTT),
m_MsgLocalExpirationTimeout (I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT_MAX), m_MsgLocalExpirationTimeout (I2NP_MESSAGE_LOCAL_EXPIRATION_TIMEOUT_MAX),
@ -108,6 +103,7 @@ namespace transport
InitNoiseXKState1 (*m_NoiseState, m_Address->s); InitNoiseXKState1 (*m_NoiseState, m_Address->s);
m_RemoteEndpoint = boost::asio::ip::udp::endpoint (m_Address->host, m_Address->port); m_RemoteEndpoint = boost::asio::ip::udp::endpoint (m_Address->host, m_Address->port);
m_RemoteTransports = in_RemoteRouter->GetCompatibleTransports (false); m_RemoteTransports = in_RemoteRouter->GetCompatibleTransports (false);
m_RemoteVersion = in_RemoteRouter->GetVersion ();
if (in_RemoteRouter->IsSSU2PeerTesting (true)) m_RemotePeerTestTransports |= i2p::data::RouterInfo::eSSU2V4; if (in_RemoteRouter->IsSSU2PeerTesting (true)) m_RemotePeerTestTransports |= i2p::data::RouterInfo::eSSU2V4;
if (in_RemoteRouter->IsSSU2PeerTesting (false)) m_RemotePeerTestTransports |= i2p::data::RouterInfo::eSSU2V6; if (in_RemoteRouter->IsSSU2PeerTesting (false)) m_RemotePeerTestTransports |= i2p::data::RouterInfo::eSSU2V6;
RAND_bytes ((uint8_t *)&m_DestConnID, 8); RAND_bytes ((uint8_t *)&m_DestConnID, 8);
@ -231,6 +227,13 @@ namespace transport
if (m_Server.AddPendingOutgoingSession (shared_from_this ())) if (m_Server.AddPendingOutgoingSession (shared_from_this ()))
{ {
m_Server.RemoveSession (GetConnID ()); m_Server.RemoveSession (GetConnID ());
// update endpoint in profile because we know it now
auto identity = GetRemoteIdentity ();
if (identity)
{
auto profile = i2p::data::GetRouterProfile (identity->GetIdentHash ());
if (profile) profile->SetLastEndpoint (m_RemoteEndpoint);
}
// connect // connect
LogPrint (eLogDebug, "SSU2: Connecting after introduction to ", GetIdentHashBase64()); LogPrint (eLogDebug, "SSU2: Connecting after introduction to ", GetIdentHashBase64());
Connect (); Connect ();
@ -1174,6 +1177,8 @@ namespace transport
" and actual endpoint ", m_RemoteEndpoint.address (), " from ", i2p::data::GetIdentHashAbbreviation (ri->GetIdentHash ())); " and actual endpoint ", m_RemoteEndpoint.address (), " from ", i2p::data::GetIdentHashAbbreviation (ri->GetIdentHash ()));
return false; return false;
} }
if (!m_Address->published)
ri->GetProfile ()->SetLastEndpoint (m_RemoteEndpoint);
SetRemoteIdentity (ri->GetRouterIdentity ()); SetRemoteIdentity (ri->GetRouterIdentity ());
AdjustMaxPayloadSize (); AdjustMaxPayloadSize ();
m_Server.AddSessionByRouterHash (shared_from_this ()); // we know remote router now m_Server.AddSessionByRouterHash (shared_from_this ()); // we know remote router now
@ -1181,6 +1186,7 @@ namespace transport
m_RemotePeerTestTransports = 0; m_RemotePeerTestTransports = 0;
if (ri->IsSSU2PeerTesting (true)) m_RemotePeerTestTransports |= i2p::data::RouterInfo::eSSU2V4; if (ri->IsSSU2PeerTesting (true)) m_RemotePeerTestTransports |= i2p::data::RouterInfo::eSSU2V4;
if (ri->IsSSU2PeerTesting (false)) m_RemotePeerTestTransports |= i2p::data::RouterInfo::eSSU2V6; if (ri->IsSSU2PeerTesting (false)) m_RemotePeerTestTransports |= i2p::data::RouterInfo::eSSU2V6;
m_RemoteVersion = ri->GetVersion ();
// handle other blocks // handle other blocks
HandlePayload (decryptedPayload.data () + riSize + 3, decryptedPayload.size () - riSize - 3); HandlePayload (decryptedPayload.data () + riSize + 3, decryptedPayload.size () - riSize - 3);
@ -1357,46 +1363,6 @@ namespace transport
return true; return true;
} }
void SSU2Session::SendHolePunch (uint32_t nonce, const boost::asio::ip::udp::endpoint& ep,
const uint8_t * introKey, uint64_t token)
{
// we are Charlie
LogPrint (eLogDebug, "SSU2: Sending HolePunch to ", ep);
Header header;
uint8_t h[32], payload[SSU2_MAX_PACKET_SIZE];
// fill packet
header.h.connID = htobe64 (((uint64_t)nonce << 32) | nonce); // dest id
RAND_bytes (header.buf + 8, 4); // random packet num
header.h.type = eSSU2HolePunch;
header.h.flags[0] = 2; // ver
header.h.flags[1] = (uint8_t)i2p::context.GetNetID (); // netID
header.h.flags[2] = 0; // flag
memcpy (h, header.buf, 16);
uint64_t c = ~header.h.connID;
memcpy (h + 16, &c, 8); // source id
RAND_bytes (h + 24, 8); // token
// payload
payload[0] = eSSU2BlkDateTime;
htobe16buf (payload + 1, 4);
htobe32buf (payload + 3, (i2p::util::GetMillisecondsSinceEpoch () + 500)/1000);
size_t payloadSize = 7;
payloadSize += CreateAddressBlock (payload + payloadSize, m_MaxPayloadSize - payloadSize, ep);
payloadSize += CreateRelayResponseBlock (payload + payloadSize, m_MaxPayloadSize - payloadSize,
eSSU2RelayResponseCodeAccept, nonce, token, ep.address ().is_v4 ());
payloadSize += CreatePaddingBlock (payload + payloadSize, m_MaxPayloadSize - payloadSize);
// encrypt
uint8_t n[12];
CreateNonce (be32toh (header.h.packetNum), n);
i2p::crypto::AEADChaCha20Poly1305 (payload, payloadSize, h, 32, introKey, n, payload, payloadSize + 16, true);
payloadSize += 16;
header.ll[0] ^= CreateHeaderMask (introKey, payload + (payloadSize - 24));
header.ll[1] ^= CreateHeaderMask (introKey, payload + (payloadSize - 12));
memset (n, 0, 12);
i2p::crypto::ChaCha20 (h + 16, 16, introKey, n, h + 16);
// send
m_Server.Send (header.buf, 16, h + 16, 16, payload, payloadSize, ep);
}
bool SSU2Session::ProcessHolePunch (uint8_t * buf, size_t len) bool SSU2Session::ProcessHolePunch (uint8_t * buf, size_t len)
{ {
// we are Alice // we are Alice
@ -1965,34 +1931,36 @@ namespace transport
return; return;
} }
auto mts = i2p::util::GetMillisecondsSinceEpoch (); auto mts = i2p::util::GetMillisecondsSinceEpoch ();
session->m_RelaySessions.emplace (bufbe32toh (buf + 1), // nonce uint32_t nonce = bufbe32toh (buf + 1);
std::make_pair (shared_from_this (), mts/1000) ); if (session->m_RelaySessions.emplace (nonce, std::make_pair (shared_from_this (), mts/1000)).second)
{
// send relay intro to Charlie // send relay intro to Charlie
auto r = i2p::data::netdb.FindRouter (GetRemoteIdentity ()->GetIdentHash ()); // Alice's RI auto r = i2p::data::netdb.FindRouter (GetRemoteIdentity ()->GetIdentHash ()); // Alice's RI
if (r && (r->IsUnreachable () || !i2p::data::netdb.PopulateRouterInfoBuffer (r))) r = nullptr; if (r && (r->IsUnreachable () || !i2p::data::netdb.PopulateRouterInfoBuffer (r))) r = nullptr;
if (!r) LogPrint (eLogWarning, "SSU2: RelayRequest Alice's router info not found"); if (!r) LogPrint (eLogWarning, "SSU2: RelayRequest Alice's router info not found");
auto packet = m_Server.GetSentPacketsPool ().AcquireShared (); auto packet = m_Server.GetSentPacketsPool ().AcquireShared ();
packet->payloadSize = r ? CreateRouterInfoBlock (packet->payload, m_MaxPayloadSize - len - 32, r) : 0; packet->payloadSize = r ? CreateRouterInfoBlock (packet->payload, m_MaxPayloadSize - len - 32, r) : 0;
if (!packet->payloadSize && r) if (!packet->payloadSize && r)
session->SendFragmentedMessage (CreateDatabaseStoreMsg (r)); session->SendFragmentedMessage (CreateDatabaseStoreMsg (r));
packet->payloadSize += CreateRelayIntroBlock (packet->payload + packet->payloadSize, m_MaxPayloadSize - packet->payloadSize, buf + 1, len -1); packet->payloadSize += CreateRelayIntroBlock (packet->payload + packet->payloadSize, m_MaxPayloadSize - packet->payloadSize, buf + 1, len -1);
if (packet->payloadSize < m_MaxPayloadSize) if (packet->payloadSize < m_MaxPayloadSize)
packet->payloadSize += CreatePaddingBlock (packet->payload + packet->payloadSize, m_MaxPayloadSize - packet->payloadSize); packet->payloadSize += CreatePaddingBlock (packet->payload + packet->payloadSize, m_MaxPayloadSize - packet->payloadSize);
uint32_t packetNum = session->SendData (packet->payload, packet->payloadSize); uint32_t packetNum = session->SendData (packet->payload, packet->payloadSize);
packet->sendTime = mts; packet->sendTime = mts;
// Charlie always responds with RelayResponse // Charlie always responds with RelayResponse
session->m_SentPackets.emplace (packetNum, packet); session->m_SentPackets.emplace (packetNum, packet);
}
else
LogPrint (eLogInfo, "SSU2: Relay request nonce ", nonce, " already exists. Ignore");
} }
void SSU2Session::HandleRelayIntro (const uint8_t * buf, size_t len, int attempts) void SSU2Session::HandleRelayIntro (const uint8_t * buf, size_t len, int attempts)
{ {
// we are Charlie // we are Charlie
auto mts = i2p::util::GetMillisecondsSinceEpoch ();
SSU2RelayResponseCode code = eSSU2RelayResponseCodeAccept; SSU2RelayResponseCode code = eSSU2RelayResponseCodeAccept;
uint64_t token = 0; boost::asio::ip::udp::endpoint ep;
bool isV4 = false; std::shared_ptr<const i2p::data::RouterInfo::Address> addr;
auto r = i2p::data::netdb.FindRouter (buf + 1); // Alice auto r = i2p::data::netdb.FindRouter (buf + 1); // Alice
if (r) if (r)
{ {
@ -2005,21 +1973,19 @@ namespace transport
s.Insert (buf + 47, asz); // Alice Port, Alice IP s.Insert (buf + 47, asz); // Alice Port, Alice IP
if (s.Verify (r->GetIdentity (), buf + 47 + asz)) if (s.Verify (r->GetIdentity (), buf + 47 + asz))
{ {
// send HolePunch // obtain and check endpoint and address for HolePunch
boost::asio::ip::udp::endpoint ep;
if (ExtractEndpoint (buf + 47, asz, ep)) if (ExtractEndpoint (buf + 47, asz, ep))
{ {
std::shared_ptr<const i2p::data::RouterInfo::Address> addr;
if (!ep.address ().is_unspecified () && ep.port ()) if (!ep.address ().is_unspecified () && ep.port ())
addr = ep.address ().is_v6 () ? r->GetSSU2V6Address () : r->GetSSU2V4Address ();
if (addr)
{ {
if (m_Server.IsSupported (ep.address ())) if (m_Server.IsSupported (ep.address ()))
{ {
token = m_Server.GetIncomingToken (ep); addr = ep.address ().is_v6 () ? r->GetSSU2V6Address () : r->GetSSU2V4Address ();
isV4 = ep.address ().is_v4 (); if (!addr)
SendHolePunch (bufbe32toh (buf + 33), ep, addr->i, token); {
m_Server.AddConnectedRecently (ep, mts/1000); LogPrint (eLogWarning, "SSU2: RelayIntro address for endpoint not found");
code = eSSU2RelayResponseCodeCharlieAliceIsUnknown;
}
} }
else else
{ {
@ -2029,7 +1995,7 @@ namespace transport
} }
else else
{ {
LogPrint (eLogWarning, "SSU2: RelayIntro unknown address"); LogPrint (eLogWarning, "SSU2: RelayIntro invalid endpoint");
code = eSSU2RelayResponseCodeCharlieAliceIsUnknown; code = eSSU2RelayResponseCodeCharlieAliceIsUnknown;
} }
} }
@ -2065,14 +2031,29 @@ namespace transport
} }
// send relay response to Bob // send relay response to Bob
auto packet = m_Server.GetSentPacketsPool ().AcquireShared (); auto packet = m_Server.GetSentPacketsPool ().AcquireShared ();
uint32_t nonce = bufbe32toh (buf + 33);
packet->payloadSize = CreateRelayResponseBlock (packet->payload, m_MaxPayloadSize, packet->payloadSize = CreateRelayResponseBlock (packet->payload, m_MaxPayloadSize,
code, bufbe32toh (buf + 33), token, isV4); code, nonce, m_Server.GetIncomingToken (ep), ep.address ().is_v4 ());
if (code == eSSU2RelayResponseCodeAccept && addr)
{
// send HolePunch
auto holePunchSession = std::make_shared<SSU2HolePunchSession>(m_Server, nonce, ep, addr);
if (m_Server.AddSession (holePunchSession))
holePunchSession->SendHolePunch (packet->payload, packet->payloadSize); // relay response block
else
{
LogPrint (eLogInfo, "SSU2: Relay intro nonce ", nonce, " already exists. Ignore");
return;
}
}
packet->payloadSize += CreatePaddingBlock (packet->payload + packet->payloadSize, m_MaxPayloadSize - packet->payloadSize); packet->payloadSize += CreatePaddingBlock (packet->payload + packet->payloadSize, m_MaxPayloadSize - packet->payloadSize);
/*uint32_t packetNum = */SendData (packet->payload, packet->payloadSize); uint32_t packetNum = SendData (packet->payload, packet->payloadSize);
// sometimes Bob doesn't ack this RelayResponse if (m_RemoteVersion >= SSU2_MIN_RELAY_RESPONSE_RESEND_VERSION)
// TODO: uncomment line below once the problem is resolved {
//packet->sendTime = mts; // sometimes Bob doesn't ack this RelayResponse in older versions
//m_SentPackets.emplace (packetNum, packet); packet->sendTime = i2p::util::GetMillisecondsSinceEpoch ();
m_SentPackets.emplace (packetNum, packet);
}
} }
void SSU2Session::HandleRelayResponse (const uint8_t * buf, size_t len) void SSU2Session::HandleRelayResponse (const uint8_t * buf, size_t len)
@ -2107,11 +2088,13 @@ namespace transport
memcpy (payload + 3, buf, len); // forward to Alice as is memcpy (payload + 3, buf, len); // forward to Alice as is
packet->payloadSize = len + 3; packet->payloadSize = len + 3;
packet->payloadSize += CreatePaddingBlock (payload + packet->payloadSize, m_MaxPayloadSize - packet->payloadSize); packet->payloadSize += CreatePaddingBlock (payload + packet->payloadSize, m_MaxPayloadSize - packet->payloadSize);
/*uint32_t packetNum = */it->second.first->SendData (packet->payload, packet->payloadSize); uint32_t packetNum = it->second.first->SendData (packet->payload, packet->payloadSize);
// sometimes Alice doesn't ack this RelayResponse if (m_RemoteVersion >= SSU2_MIN_RELAY_RESPONSE_RESEND_VERSION)
// TODO: uncomment line below once the problem is resolved {
//packet->sendTime = i2p::util::GetMillisecondsSinceEpoch (); // sometimes Alice doesn't ack this RelayResponse in older versions
//it->second.first->m_SentPackets.emplace (packetNum, packet); packet->sendTime = i2p::util::GetMillisecondsSinceEpoch ();
it->second.first->m_SentPackets.emplace (packetNum, packet);
}
} }
else else
{ {
@ -2179,29 +2162,33 @@ namespace transport
GetRemoteIdentity ()->GetIdentHash ()); GetRemoteIdentity ()->GetIdentHash ());
if (session) // session with Charlie if (session) // session with Charlie
{ {
m_Server.AddPeerTest (nonce, shared_from_this (), ts/1000); if (m_Server.AddPeerTest (nonce, shared_from_this (), ts/1000))
auto packet = m_Server.GetSentPacketsPool ().AcquireShared ();
// Alice's RouterInfo
auto r = i2p::data::netdb.FindRouter (GetRemoteIdentity ()->GetIdentHash ());
if (r && (r->IsUnreachable () || !i2p::data::netdb.PopulateRouterInfoBuffer (r))) r = nullptr;
packet->payloadSize = r ? CreateRouterInfoBlock (packet->payload, m_MaxPayloadSize - len - 32, r) : 0;
if (!packet->payloadSize && r)
session->SendFragmentedMessage (CreateDatabaseStoreMsg (r));
if (packet->payloadSize + len + 48 > m_MaxPayloadSize)
{ {
// doesn't fit one message, send RouterInfo in separate message auto packet = m_Server.GetSentPacketsPool ().AcquireShared ();
// Alice's RouterInfo
auto r = i2p::data::netdb.FindRouter (GetRemoteIdentity ()->GetIdentHash ());
if (r && (r->IsUnreachable () || !i2p::data::netdb.PopulateRouterInfoBuffer (r))) r = nullptr;
packet->payloadSize = r ? CreateRouterInfoBlock (packet->payload, m_MaxPayloadSize - len - 32, r) : 0;
if (!packet->payloadSize && r)
session->SendFragmentedMessage (CreateDatabaseStoreMsg (r));
if (packet->payloadSize + len + 48 > m_MaxPayloadSize)
{
// doesn't fit one message, send RouterInfo in separate message
uint32_t packetNum = session->SendData (packet->payload, packet->payloadSize, SSU2_FLAG_IMMEDIATE_ACK_REQUESTED);
packet->sendTime = ts;
session->m_SentPackets.emplace (packetNum, packet);
packet = m_Server.GetSentPacketsPool ().AcquireShared (); // new packet
}
// PeerTest to Charlie
packet->payloadSize += CreatePeerTestBlock (packet->payload + packet->payloadSize, m_MaxPayloadSize - packet->payloadSize, 2,
eSSU2PeerTestCodeAccept, GetRemoteIdentity ()->GetIdentHash (), buf + offset, len - offset);
packet->payloadSize += CreatePaddingBlock (packet->payload + packet->payloadSize, m_MaxPayloadSize - packet->payloadSize);
uint32_t packetNum = session->SendData (packet->payload, packet->payloadSize, SSU2_FLAG_IMMEDIATE_ACK_REQUESTED); uint32_t packetNum = session->SendData (packet->payload, packet->payloadSize, SSU2_FLAG_IMMEDIATE_ACK_REQUESTED);
packet->sendTime = ts; packet->sendTime = ts;
session->m_SentPackets.emplace (packetNum, packet); session->m_SentPackets.emplace (packetNum, packet);
packet = m_Server.GetSentPacketsPool ().AcquireShared (); // new packet
} }
// PeerTest to Charlie else
packet->payloadSize += CreatePeerTestBlock (packet->payload + packet->payloadSize, m_MaxPayloadSize - packet->payloadSize, 2, LogPrint (eLogInfo, "SSU2: Peer test 1 nonce ", nonce, " already exists. Ignored");
eSSU2PeerTestCodeAccept, GetRemoteIdentity ()->GetIdentHash (), buf + offset, len - offset);
packet->payloadSize += CreatePaddingBlock (packet->payload + packet->payloadSize, m_MaxPayloadSize - packet->payloadSize);
uint32_t packetNum = session->SendData (packet->payload, packet->payloadSize, SSU2_FLAG_IMMEDIATE_ACK_REQUESTED);
packet->sendTime = ts;
session->m_SentPackets.emplace (packetNum, packet);
} }
else else
{ {
@ -2343,16 +2330,16 @@ namespace transport
{ {
session->SetRemoteIdentity (r->GetIdentity ()); session->SetRemoteIdentity (r->GetIdentity ());
auto addr = r->GetSSU2Address (m_Address->IsV4 ()); auto addr = r->GetSSU2Address (m_Address->IsV4 ());
if (addr) if (addr && addr->IsPeerTesting ())
{ {
if (session->GetMsgNumReceived () >= 5) if (session->GetMsgNumReceived () >= 5)
{ {
// msg 5 already received // msg 5 already received and we know remote endpoint
if (session->GetMsgNumReceived () == 5) if (session->GetMsgNumReceived () == 5)
{ {
if (!session->IsConnectedRecently ()) if (!session->IsConnectedRecently ())
SetRouterStatus (eRouterStatusOK); SetRouterStatus (eRouterStatusOK);
// send msg 6 // send msg 6 immeditely
session->SendPeerTest (6, buf + offset, len - offset, addr); session->SendPeerTest (6, buf + offset, len - offset, addr);
} }
else else
@ -2363,6 +2350,12 @@ namespace transport
session->m_Address = addr; session->m_Address = addr;
if (GetTestingState ()) if (GetTestingState ())
{ {
// schedule msg 6 with delay
if (!addr->host.is_unspecified () && addr->port)
{
session->SetRemoteEndpoint (boost::asio::ip::udp::endpoint (addr->host, addr->port));
session->SendPeerTest (6, buf + offset, len - offset, addr, true);
}
SetTestingState (false); SetTestingState (false);
if (GetRouterStatus () != eRouterStatusFirewalled && addr->IsPeerTesting ()) if (GetRouterStatus () != eRouterStatusFirewalled && addr->IsPeerTesting ())
{ {
@ -2380,7 +2373,7 @@ namespace transport
} }
else else
{ {
LogPrint (eLogWarning, "SSU2: Peer test 4 address not found"); LogPrint (eLogWarning, "SSU2: Peer test 4 address not found or not supported");
session->Done (); session->Done ();
} }
} }
@ -3062,7 +3055,7 @@ namespace transport
{ {
if (ts > it->second.second + SSU2_RELAY_NONCE_EXPIRATION_TIMEOUT) if (ts > it->second.second + SSU2_RELAY_NONCE_EXPIRATION_TIMEOUT)
{ {
LogPrint (eLogWarning, "SSU2: Relay nonce ", it->first, " was not responded in ", SSU2_RELAY_NONCE_EXPIRATION_TIMEOUT, " seconds, deleted"); LogPrint (eLogInfo, "SSU2: Relay nonce ", it->first, " was not responded in ", SSU2_RELAY_NONCE_EXPIRATION_TIMEOUT, " seconds, deleted");
it = m_RelaySessions.erase (it); it = m_RelaySessions.erase (it);
} }
else else
@ -3086,216 +3079,5 @@ namespace transport
else if (!sent && !m_SentPackets.empty ()) // if only acks received, nothing sent and we still have something to resend else if (!sent && !m_SentPackets.empty ()) // if only acks received, nothing sent and we still have something to resend
Resend (i2p::util::GetMillisecondsSinceEpoch ()); // than right time to resend Resend (i2p::util::GetMillisecondsSinceEpoch ()); // than right time to resend
} }
SSU2PeerTestSession::SSU2PeerTestSession (SSU2Server& server, uint64_t sourceConnID, uint64_t destConnID):
SSU2Session (server, nullptr, nullptr, false),
m_MsgNumReceived (0), m_NumResends (0),m_IsConnectedRecently (false), m_IsStatusChanged (false),
m_PeerTestResendTimer (server.GetService ())
{
if (!sourceConnID) sourceConnID = ~destConnID;
if (!destConnID) destConnID = ~sourceConnID;
SetSourceConnID (sourceConnID);
SetDestConnID (destConnID);
SetState (eSSU2SessionStatePeerTest);
SetTerminationTimeout (SSU2_PEER_TEST_EXPIRATION_TIMEOUT);
}
bool SSU2PeerTestSession::ProcessPeerTest (uint8_t * buf, size_t len)
{
// we are Alice or Charlie, msgs 5,6,7
Header header;
memcpy (header.buf, buf, 16);
header.ll[0] ^= CreateHeaderMask (i2p::context.GetSSU2IntroKey (), buf + (len - 24));
header.ll[1] ^= CreateHeaderMask (i2p::context.GetSSU2IntroKey (), buf + (len - 12));
if (header.h.type != eSSU2PeerTest)
{
LogPrint (eLogWarning, "SSU2: Unexpected message type ", (int)header.h.type, " instead ", (int)eSSU2PeerTest);
return false;
}
if (len < 48)
{
LogPrint (eLogWarning, "SSU2: PeerTest message too short ", len);
return false;
}
uint8_t nonce[12] = {0};
uint64_t headerX[2]; // sourceConnID, token
i2p::crypto::ChaCha20 (buf + 16, 16, i2p::context.GetSSU2IntroKey (), nonce, (uint8_t *)headerX);
SetDestConnID (headerX[0]);
// decrypt and handle payload
uint8_t * payload = buf + 32;
CreateNonce (be32toh (header.h.packetNum), nonce);
uint8_t h[32];
memcpy (h, header.buf, 16);
memcpy (h + 16, &headerX, 16);
if (!i2p::crypto::AEADChaCha20Poly1305 (payload, len - 48, h, 32,
i2p::context.GetSSU2IntroKey (), nonce, payload, len - 48, false))
{
LogPrint (eLogWarning, "SSU2: PeerTest AEAD verification failed ");
return false;
}
HandlePayload (payload, len - 48);
SetIsDataReceived (false);
return true;
}
void SSU2PeerTestSession::HandlePeerTest (const uint8_t * buf, size_t len)
{
// msgs 5-7
if (len < 8) return;
uint8_t msg = buf[0];
if (msg <= m_MsgNumReceived)
{
LogPrint (eLogDebug, "SSU2: PeerTest msg num ", msg, " received after ", m_MsgNumReceived, ". Ignored");
return;
}
size_t offset = 3; // points to signed data after msg + code + flag
uint32_t nonce = bufbe32toh (buf + offset + 1); // 1 - ver
switch (msg) // msg
{
case 5: // Alice from Charlie 1
{
if (htobe64 (((uint64_t)nonce << 32) | nonce) == GetSourceConnID ())
{
m_IsConnectedRecently = GetServer ().IsConnectedRecently (GetRemoteEndpoint ());
if (GetAddress ())
{
if (!m_IsConnectedRecently)
SetRouterStatus (eRouterStatusOK);
else if (m_IsStatusChanged && GetRouterStatus () == eRouterStatusFirewalled)
SetRouterStatus (eRouterStatusUnknown);
SendPeerTest (6, buf + offset, len - offset);
}
}
else
LogPrint (eLogWarning, "SSU2: Peer test 5 nonce mismatch ", nonce, " connID=", GetSourceConnID ());
break;
}
case 6: // Charlie from Alice
{
m_PeerTestResendTimer.cancel (); // no more msg 5 resends
if (GetAddress ())
SendPeerTest (7, buf + offset, len - offset);
else
LogPrint (eLogWarning, "SSU2: Unknown address for peer test 6");
GetServer ().AddConnectedRecently (GetRemoteEndpoint (), i2p::util::GetSecondsSinceEpoch ());
GetServer ().RequestRemoveSession (GetConnID ());
break;
}
case 7: // Alice from Charlie 2
{
m_PeerTestResendTimer.cancel (); // no more msg 6 resends
auto addr = GetAddress ();
if (addr && addr->IsV6 ())
i2p::context.SetStatusV6 (eRouterStatusOK); // set status OK for ipv6 even if from SSU2
GetServer ().AddConnectedRecently (GetRemoteEndpoint (), i2p::util::GetSecondsSinceEpoch ());
GetServer ().RequestRemoveSession (GetConnID ());
break;
}
default:
LogPrint (eLogWarning, "SSU2: PeerTest unexpected msg num ", msg);
return;
}
m_MsgNumReceived = msg;
}
void SSU2PeerTestSession::SendPeerTest (uint8_t msg)
{
auto addr = GetAddress ();
if (!addr) return;
Header header;
uint8_t h[32], payload[SSU2_MAX_PACKET_SIZE];
// fill packet
header.h.connID = GetDestConnID (); // dest id
RAND_bytes (header.buf + 8, 4); // random packet num
header.h.type = eSSU2PeerTest;
header.h.flags[0] = 2; // ver
header.h.flags[1] = (uint8_t)i2p::context.GetNetID (); // netID
header.h.flags[2] = 0; // flag
memcpy (h, header.buf, 16);
htobuf64 (h + 16, GetSourceConnID ()); // source id
// payload
payload[0] = eSSU2BlkDateTime;
htobe16buf (payload + 1, 4);
htobe32buf (payload + 3, (i2p::util::GetMillisecondsSinceEpoch () + 500)/1000);
size_t payloadSize = 7;
if (msg == 6 || msg == 7)
payloadSize += CreateAddressBlock (payload + payloadSize, GetMaxPayloadSize () - payloadSize, GetRemoteEndpoint ());
payloadSize += CreatePeerTestBlock (payload + payloadSize, GetMaxPayloadSize () - payloadSize,
msg, eSSU2PeerTestCodeAccept, nullptr, m_SignedData.data (), m_SignedData.size ());
payloadSize += CreatePaddingBlock (payload + payloadSize, GetMaxPayloadSize () - payloadSize);
// encrypt
uint8_t n[12];
CreateNonce (be32toh (header.h.packetNum), n);
i2p::crypto::AEADChaCha20Poly1305 (payload, payloadSize, h, 32, addr->i, n, payload, payloadSize + 16, true);
payloadSize += 16;
header.ll[0] ^= CreateHeaderMask (addr->i, payload + (payloadSize - 24));
header.ll[1] ^= CreateHeaderMask (addr->i, payload + (payloadSize - 12));
memset (n, 0, 12);
i2p::crypto::ChaCha20 (h + 16, 16, addr->i, n, h + 16);
// send
GetServer ().Send (header.buf, 16, h + 16, 16, payload, payloadSize, GetRemoteEndpoint ());
}
void SSU2PeerTestSession::SendPeerTest (uint8_t msg, const uint8_t * signedData, size_t signedDataLen)
{
#if __cplusplus >= 202002L // C++20
m_SignedData.assign (signedData, signedData + signedDataLen);
#else
m_SignedData.resize (signedDataLen);
memcpy (m_SignedData.data (), signedData, signedDataLen);
#endif
SendPeerTest (msg);
// schedule resend for msgs 5 or 6
if (msg == 5 || msg == 6)
ScheduleResend ();
}
void SSU2PeerTestSession::SendPeerTest (uint8_t msg, const uint8_t * signedData, size_t signedDataLen,
std::shared_ptr<const i2p::data::RouterInfo::Address> addr)
{
if (!addr) return;
SetAddress (addr);
SendPeerTest (msg, signedData, signedDataLen);
}
void SSU2PeerTestSession::Connect ()
{
LogPrint (eLogError, "SSU2: Can't connect peer test session");
}
bool SSU2PeerTestSession::ProcessFirstIncomingMessage (uint64_t connID, uint8_t * buf, size_t len)
{
LogPrint (eLogError, "SSU2: Can't handle incoming message in peer test session");
return false;
}
void SSU2PeerTestSession::ScheduleResend ()
{
if (m_NumResends < SSU2_PEER_TEST_MAX_NUM_RESENDS)
{
m_PeerTestResendTimer.expires_from_now (boost::posix_time::milliseconds(
SSU2_PEER_TEST_RESEND_INTERVAL + GetServer ().GetRng ()() % SSU2_PEER_TEST_RESEND_INTERVAL_VARIANCE));
std::weak_ptr<SSU2PeerTestSession> s(std::static_pointer_cast<SSU2PeerTestSession>(shared_from_this ()));
m_PeerTestResendTimer.async_wait ([s](const boost::system::error_code& ecode)
{
if (ecode != boost::asio::error::operation_aborted)
{
auto s1 = s.lock ();
if (s1)
{
int msg = 0;
if (s1->m_MsgNumReceived < 6)
msg = (s1->m_MsgNumReceived == 5) ? 6 : 5;
if (msg) // 5 or 6
{
s1->SendPeerTest (msg);
s1->ScheduleResend ();
}
}
}
});
m_NumResends++;
}
}
} }
} }

54
libi2pd/SSU2Session.h

@ -15,6 +15,7 @@
#include <set> #include <set>
#include <list> #include <list>
#include <boost/asio.hpp> #include <boost/asio.hpp>
#include "version.h"
#include "Crypto.h" #include "Crypto.h"
#include "RouterInfo.h" #include "RouterInfo.h"
#include "RouterContext.h" #include "RouterContext.h"
@ -55,6 +56,7 @@ namespace transport
const int SSU2_MAX_NUM_ACK_RANGES = 32; // to send const int SSU2_MAX_NUM_ACK_RANGES = 32; // to send
const uint8_t SSU2_MAX_NUM_FRAGMENTS = 64; const uint8_t SSU2_MAX_NUM_FRAGMENTS = 64;
const int SSU2_SEND_DATETIME_NUM_PACKETS = 256; const int SSU2_SEND_DATETIME_NUM_PACKETS = 256;
const int SSU2_MIN_RELAY_RESPONSE_RESEND_VERSION = MAKE_VERSION_NUMBER(0, 9, 64); // 0.9.64
// flags // flags
const uint8_t SSU2_FLAG_IMMEDIATE_ACK_REQUESTED = 0x01; const uint8_t SSU2_FLAG_IMMEDIATE_ACK_REQUESTED = 0x01;
@ -112,6 +114,7 @@ namespace transport
eSSU2SessionStateTerminated, eSSU2SessionStateTerminated,
eSSU2SessionStateFailed, eSSU2SessionStateFailed,
eSSU2SessionStateIntroduced, eSSU2SessionStateIntroduced,
eSSU2SessionStateHolePunch,
eSSU2SessionStatePeerTest, eSSU2SessionStatePeerTest,
eSSU2SessionStateTokenRequestReceived eSSU2SessionStateTokenRequestReceived
}; };
@ -296,6 +299,8 @@ namespace transport
size_t CreatePaddingBlock (uint8_t * buf, size_t len, size_t minSize = 0); size_t CreatePaddingBlock (uint8_t * buf, size_t len, size_t minSize = 0);
size_t CreatePeerTestBlock (uint8_t * buf, size_t len, uint8_t msg, SSU2PeerTestCode code, const uint8_t * routerHash, const uint8_t * signedData, size_t signedDataLen); size_t CreatePeerTestBlock (uint8_t * buf, size_t len, uint8_t msg, SSU2PeerTestCode code, const uint8_t * routerHash, const uint8_t * signedData, size_t signedDataLen);
bool ExtractEndpoint (const uint8_t * buf, size_t size, boost::asio::ip::udp::endpoint& ep);
private: private:
void Terminate (); void Terminate ();
@ -320,7 +325,6 @@ namespace transport
uint32_t SendData (const uint8_t * buf, size_t len, uint8_t flags = 0); // returns packet num uint32_t SendData (const uint8_t * buf, size_t len, uint8_t flags = 0); // returns packet num
void SendQuickAck (); void SendQuickAck ();
void SendTermination (); void SendTermination ();
void SendHolePunch (uint32_t nonce, const boost::asio::ip::udp::endpoint& ep, const uint8_t * introKey, uint64_t token);
void SendPathResponse (const uint8_t * data, size_t len); void SendPathResponse (const uint8_t * data, size_t len);
void SendPathChallenge (); void SendPathChallenge ();
@ -328,8 +332,7 @@ namespace transport
void HandleRouterInfo (const uint8_t * buf, size_t len); void HandleRouterInfo (const uint8_t * buf, size_t len);
void HandleAck (const uint8_t * buf, size_t len); void HandleAck (const uint8_t * buf, size_t len);
void HandleAckRange (uint32_t firstPacketNum, uint32_t lastPacketNum, uint64_t ts); void HandleAckRange (uint32_t firstPacketNum, uint32_t lastPacketNum, uint64_t ts);
void HandleAddress (const uint8_t * buf, size_t len); virtual void HandleAddress (const uint8_t * buf, size_t len);
bool ExtractEndpoint (const uint8_t * buf, size_t size, boost::asio::ip::udp::endpoint& ep);
size_t CreateEndpoint (uint8_t * buf, size_t len, const boost::asio::ip::udp::endpoint& ep); size_t CreateEndpoint (uint8_t * buf, size_t len, const boost::asio::ip::udp::endpoint& ep);
std::shared_ptr<const i2p::data::RouterInfo::Address> FindLocalAddress () const; std::shared_ptr<const i2p::data::RouterInfo::Address> FindLocalAddress () const;
void AdjustMaxPayloadSize (); void AdjustMaxPayloadSize ();
@ -353,6 +356,7 @@ namespace transport
size_t CreateFollowOnFragmentBlock (uint8_t * buf, size_t len, std::shared_ptr<I2NPMessage> msg, uint8_t& fragmentNum, uint32_t msgID); size_t CreateFollowOnFragmentBlock (uint8_t * buf, size_t len, std::shared_ptr<I2NPMessage> msg, uint8_t& fragmentNum, uint32_t msgID);
size_t CreateRelayIntroBlock (uint8_t * buf, size_t len, const uint8_t * introData, size_t introDataLen); size_t CreateRelayIntroBlock (uint8_t * buf, size_t len, const uint8_t * introData, size_t introDataLen);
size_t CreateRelayResponseBlock (uint8_t * buf, size_t len, SSU2RelayResponseCode code, uint32_t nonce, uint64_t token, bool v4); size_t CreateRelayResponseBlock (uint8_t * buf, size_t len, SSU2RelayResponseCode code, uint32_t nonce, uint64_t token, bool v4);
size_t CreatePeerTestBlock (uint8_t * buf, size_t len, uint32_t nonce); // Alice size_t CreatePeerTestBlock (uint8_t * buf, size_t len, uint32_t nonce); // Alice
size_t CreateTerminationBlock (uint8_t * buf, size_t len); size_t CreateTerminationBlock (uint8_t * buf, size_t len);
@ -366,6 +370,7 @@ namespace transport
std::shared_ptr<const i2p::data::RouterInfo::Address> m_Address; std::shared_ptr<const i2p::data::RouterInfo::Address> m_Address;
boost::asio::ip::udp::endpoint m_RemoteEndpoint; boost::asio::ip::udp::endpoint m_RemoteEndpoint;
i2p::data::RouterInfo::CompatibleTransports m_RemoteTransports, m_RemotePeerTestTransports; i2p::data::RouterInfo::CompatibleTransports m_RemoteTransports, m_RemotePeerTestTransports;
int m_RemoteVersion;
uint64_t m_DestConnID, m_SourceConnID; uint64_t m_DestConnID, m_SourceConnID;
SSU2SessionState m_State; SSU2SessionState m_State;
uint8_t m_KeyDataSend[64], m_KeyDataReceive[64]; uint8_t m_KeyDataSend[64], m_KeyDataReceive[64];
@ -391,49 +396,18 @@ namespace transport
uint64_t m_LastResendTime, m_LastResendAttemptTime; // in milliseconds uint64_t m_LastResendTime, m_LastResendAttemptTime; // in milliseconds
}; };
const int SSU2_PEER_TEST_RESEND_INTERVAL = 3000; // in milliseconds
const int SSU2_PEER_TEST_RESEND_INTERVAL_VARIANCE = 2000; // in milliseconds
const int SSU2_PEER_TEST_MAX_NUM_RESENDS = 3;
class SSU2PeerTestSession: public SSU2Session // for PeerTest msgs 5,6,7
{
public:
SSU2PeerTestSession (SSU2Server& server, uint64_t sourceConnID, uint64_t destConnID);
uint8_t GetMsgNumReceived () const { return m_MsgNumReceived; }
bool IsConnectedRecently () const { return m_IsConnectedRecently; }
void SetStatusChanged () { m_IsStatusChanged = true; }
void SendPeerTest (uint8_t msg, const uint8_t * signedData, size_t signedDataLen,
std::shared_ptr<const i2p::data::RouterInfo::Address> addr);
bool ProcessPeerTest (uint8_t * buf, size_t len) override;
void Connect () override; // outgoing
bool ProcessFirstIncomingMessage (uint64_t connID, uint8_t * buf, size_t len) override; // incoming
private:
void SendPeerTest (uint8_t msg, const uint8_t * signedData, size_t signedDataLen); // PeerTest message
void SendPeerTest (uint8_t msg); // send or resend m_SignedData
void HandlePeerTest (const uint8_t * buf, size_t len) override;
void ScheduleResend ();
private:
uint8_t m_MsgNumReceived, m_NumResends;
bool m_IsConnectedRecently, m_IsStatusChanged;
std::vector<uint8_t> m_SignedData; // for resends
boost::asio::deadline_timer m_PeerTestResendTimer;
};
inline uint64_t CreateHeaderMask (const uint8_t * kh, const uint8_t * nonce) inline uint64_t CreateHeaderMask (const uint8_t * kh, const uint8_t * nonce)
{ {
uint64_t data = 0; uint64_t data = 0;
i2p::crypto::ChaCha20 ((uint8_t *)&data, 8, kh, nonce, (uint8_t *)&data); i2p::crypto::ChaCha20 ((uint8_t *)&data, 8, kh, nonce, (uint8_t *)&data);
return data; return data;
} }
inline void CreateNonce (uint64_t seqn, uint8_t * nonce)
{
memset (nonce, 0, 4);
htole64buf (nonce + 4, seqn);
}
} }
} }

105
libi2pd/Streaming.cpp

@ -73,14 +73,14 @@ namespace stream
m_LastConfirmedReceivedSequenceNumber (0), // for limit inbound speed m_LastConfirmedReceivedSequenceNumber (0), // for limit inbound speed
m_Status (eStreamStatusNew), m_IsAckSendScheduled (false), m_IsNAcked (false), m_IsFirstACK (false), m_Status (eStreamStatusNew), m_IsAckSendScheduled (false), m_IsNAcked (false), m_IsFirstACK (false),
m_IsResendNeeded (false), m_IsFirstRttSample (false), m_IsSendTime (true), m_IsWinDropped (false), m_IsResendNeeded (false), m_IsFirstRttSample (false), m_IsSendTime (true), m_IsWinDropped (false),
m_IsTimeOutResend (false), m_IsImmediateAckRequested (false), m_LocalDestination (local), m_IsTimeOutResend (false), m_IsImmediateAckRequested (false), m_IsRemoteLeaseChangeInProgress (false), m_LocalDestination (local),
m_RemoteLeaseSet (remote), m_ReceiveTimer (m_Service), m_SendTimer (m_Service), m_ResendTimer (m_Service), m_RemoteLeaseSet (remote), m_ReceiveTimer (m_Service), m_SendTimer (m_Service), m_ResendTimer (m_Service),
m_AckSendTimer (m_Service), m_NumSentBytes (0), m_NumReceivedBytes (0), m_Port (port), m_AckSendTimer (m_Service), m_NumSentBytes (0), m_NumReceivedBytes (0), m_Port (port),
m_RTT (INITIAL_RTT), m_SlowRTT (INITIAL_RTT), m_SlowRTT2 (INITIAL_RTT), m_WindowSize (INITIAL_WINDOW_SIZE), m_LastWindowDropSize (0), m_RTT (INITIAL_RTT), m_SlowRTT (INITIAL_RTT), m_SlowRTT2 (INITIAL_RTT), m_WindowSize (INITIAL_WINDOW_SIZE), m_LastWindowDropSize (0),
m_WindowDropTargetSize (0), m_WindowIncCounter (0), m_RTO (INITIAL_RTO), m_WindowDropTargetSize (0), m_WindowIncCounter (0), m_RTO (INITIAL_RTO),
m_AckDelay (local.GetOwner ()->GetStreamingAckDelay ()), m_PrevRTTSample (INITIAL_RTT), m_AckDelay (local.GetOwner ()->GetStreamingAckDelay ()), m_PrevRTTSample (INITIAL_RTT),
m_Jitter (0), m_MinPacingTime (0), m_Jitter (0), m_MinPacingTime (0),
m_PacingTime (INITIAL_PACING_TIME), m_PacingTimeRem (0), m_LastSendTime (0), m_PacingTime (INITIAL_PACING_TIME), m_PacingTimeRem (0), m_LastSendTime (0), m_RemoteLeaseChangeTime (0),
m_LastACKSendTime (0), m_PacketACKInterval (1), m_PacketACKIntervalRem (0), // for limit inbound speed m_LastACKSendTime (0), m_PacketACKInterval (1), m_PacketACKIntervalRem (0), // for limit inbound speed
m_NumResendAttempts (0), m_NumPacketsToSend (0), m_MTU (STREAMING_MTU) m_NumResendAttempts (0), m_NumPacketsToSend (0), m_MTU (STREAMING_MTU)
{ {
@ -101,13 +101,13 @@ namespace stream
m_LastConfirmedReceivedSequenceNumber (0), // for limit inbound speed m_LastConfirmedReceivedSequenceNumber (0), // for limit inbound speed
m_Status (eStreamStatusNew), m_IsAckSendScheduled (false), m_IsNAcked (false), m_IsFirstACK (false), m_Status (eStreamStatusNew), m_IsAckSendScheduled (false), m_IsNAcked (false), m_IsFirstACK (false),
m_IsResendNeeded (false), m_IsFirstRttSample (false), m_IsSendTime (true), m_IsWinDropped (false), m_IsResendNeeded (false), m_IsFirstRttSample (false), m_IsSendTime (true), m_IsWinDropped (false),
m_IsTimeOutResend (false), m_IsImmediateAckRequested (false), m_LocalDestination (local), m_IsTimeOutResend (false), m_IsImmediateAckRequested (false), m_IsRemoteLeaseChangeInProgress (false), m_LocalDestination (local),
m_ReceiveTimer (m_Service), m_SendTimer (m_Service), m_ResendTimer (m_Service), m_AckSendTimer (m_Service), m_ReceiveTimer (m_Service), m_SendTimer (m_Service), m_ResendTimer (m_Service), m_AckSendTimer (m_Service),
m_NumSentBytes (0), m_NumReceivedBytes (0), m_Port (0), m_RTT (INITIAL_RTT), m_SlowRTT (INITIAL_RTT), m_SlowRTT2 (INITIAL_RTT), m_NumSentBytes (0), m_NumReceivedBytes (0), m_Port (0), m_RTT (INITIAL_RTT), m_SlowRTT (INITIAL_RTT), m_SlowRTT2 (INITIAL_RTT),
m_WindowSize (INITIAL_WINDOW_SIZE), m_LastWindowDropSize (0), m_WindowDropTargetSize (0), m_WindowIncCounter (0), m_WindowSize (INITIAL_WINDOW_SIZE), m_LastWindowDropSize (0), m_WindowDropTargetSize (0), m_WindowIncCounter (0),
m_RTO (INITIAL_RTO), m_AckDelay (local.GetOwner ()->GetStreamingAckDelay ()), m_RTO (INITIAL_RTO), m_AckDelay (local.GetOwner ()->GetStreamingAckDelay ()),
m_PrevRTTSample (INITIAL_RTT), m_Jitter (0), m_MinPacingTime (0), m_PrevRTTSample (INITIAL_RTT), m_Jitter (0), m_MinPacingTime (0),
m_PacingTime (INITIAL_PACING_TIME), m_PacingTimeRem (0), m_LastSendTime (0), m_PacingTime (INITIAL_PACING_TIME), m_PacingTimeRem (0), m_LastSendTime (0), m_RemoteLeaseChangeTime (0),
m_LastACKSendTime (0), m_PacketACKInterval (1), m_PacketACKIntervalRem (0), // for limit inbound speed m_LastACKSendTime (0), m_PacketACKInterval (1), m_PacketACKIntervalRem (0), // for limit inbound speed
m_NumResendAttempts (0), m_NumPacketsToSend (0), m_MTU (STREAMING_MTU) m_NumResendAttempts (0), m_NumPacketsToSend (0), m_MTU (STREAMING_MTU)
{ {
@ -256,6 +256,7 @@ namespace stream
if (receivedSeqn <= m_PreviousReceivedSequenceNumber || receivedSeqn == m_LastReceivedSequenceNumber) if (receivedSeqn <= m_PreviousReceivedSequenceNumber || receivedSeqn == m_LastReceivedSequenceNumber)
{ {
m_CurrentOutboundTunnel = m_LocalDestination.GetOwner ()->GetTunnelPool ()->GetNextOutboundTunnel (m_CurrentOutboundTunnel); m_CurrentOutboundTunnel = m_LocalDestination.GetOwner ()->GetTunnelPool ()->GetNextOutboundTunnel (m_CurrentOutboundTunnel);
CancelRemoteLeaseChange ();
UpdateCurrentRemoteLease (); UpdateCurrentRemoteLease ();
} }
m_PreviousReceivedSequenceNumber = receivedSeqn; m_PreviousReceivedSequenceNumber = receivedSeqn;
@ -1104,6 +1105,7 @@ namespace stream
{ {
if (!m_RemoteLeaseSet) if (!m_RemoteLeaseSet)
{ {
CancelRemoteLeaseChange ();
UpdateCurrentRemoteLease (); UpdateCurrentRemoteLease ();
if (!m_RemoteLeaseSet) if (!m_RemoteLeaseSet)
{ {
@ -1127,9 +1129,30 @@ namespace stream
} }
auto ts = i2p::util::GetMillisecondsSinceEpoch (); auto ts = i2p::util::GetMillisecondsSinceEpoch ();
if (!m_CurrentRemoteLease || !m_CurrentRemoteLease->endDate || // excluded from LeaseSet if (!m_CurrentRemoteLease || !m_CurrentRemoteLease->endDate) // excluded from LeaseSet
ts >= m_CurrentRemoteLease->endDate - i2p::data::LEASE_ENDDATE_THRESHOLD) {
CancelRemoteLeaseChange ();
UpdateCurrentRemoteLease (true); UpdateCurrentRemoteLease (true);
}
if (m_RemoteLeaseChangeTime && m_IsRemoteLeaseChangeInProgress && ts > m_RemoteLeaseChangeTime + INITIAL_RTT)
{
CancelRemoteLeaseChange ();
m_CurrentRemoteLease = m_NextRemoteLease;
HalveWindowSize ();
}
auto currentRemoteLease = m_CurrentRemoteLease;
if (!m_IsRemoteLeaseChangeInProgress && m_RemoteLeaseSet && m_CurrentRemoteLease && ts >= m_CurrentRemoteLease->endDate - i2p::data::LEASE_ENDDATE_THRESHOLD)
{
auto leases = m_RemoteLeaseSet->GetNonExpiredLeases (false);
if (leases.size ())
{
m_IsRemoteLeaseChangeInProgress = true;
UpdateCurrentRemoteLease (true);
m_NextRemoteLease = m_CurrentRemoteLease;
}
else
UpdateCurrentRemoteLease (true);
}
if (m_CurrentRemoteLease && ts < m_CurrentRemoteLease->endDate + i2p::data::LEASE_ENDDATE_THRESHOLD) if (m_CurrentRemoteLease && ts < m_CurrentRemoteLease->endDate + i2p::data::LEASE_ENDDATE_THRESHOLD)
{ {
bool freshTunnel = false; bool freshTunnel = false;
@ -1166,6 +1189,11 @@ namespace stream
msg msg
}); });
m_NumSentBytes += it->GetLength (); m_NumSentBytes += it->GetLength ();
if (m_IsRemoteLeaseChangeInProgress && !m_RemoteLeaseChangeTime)
{
m_RemoteLeaseChangeTime = ts;
m_CurrentRemoteLease = currentRemoteLease; // change it back before new lease is confirmed
}
} }
m_CurrentOutboundTunnel->SendTunnelDataMsgs (msgs); m_CurrentOutboundTunnel->SendTunnelDataMsgs (msgs);
} }
@ -1209,7 +1237,8 @@ namespace stream
if (m_Status != eStreamStatusTerminated) if (m_Status != eStreamStatusTerminated)
{ {
m_SendTimer.cancel (); m_SendTimer.cancel ();
m_SendTimer.expires_from_now (boost::posix_time::microseconds(SEND_INTERVAL)); m_SendTimer.expires_from_now (boost::posix_time::microseconds(
SEND_INTERVAL + m_LocalDestination.GetRandom () % SEND_INTERVAL_VARIANCE));
m_SendTimer.async_wait (std::bind (&Stream::HandleSendTimer, m_SendTimer.async_wait (std::bind (&Stream::HandleSendTimer,
shared_from_this (), std::placeholders::_1)); shared_from_this (), std::placeholders::_1));
} }
@ -1222,10 +1251,19 @@ namespace stream
auto ts = i2p::util::GetMillisecondsSinceEpoch (); auto ts = i2p::util::GetMillisecondsSinceEpoch ();
if (m_LastSendTime && ts*1000 > m_LastSendTime*1000 + m_PacingTime) if (m_LastSendTime && ts*1000 > m_LastSendTime*1000 + m_PacingTime)
{ {
m_NumPacketsToSend = ((ts*1000 - m_LastSendTime*1000) + m_PacingTimeRem) / m_PacingTime; if (m_PacingTime)
m_PacingTimeRem = ((ts*1000 - m_LastSendTime*1000) + m_PacingTimeRem) - (m_NumPacketsToSend * m_PacingTime); {
auto numPackets = std::lldiv (m_PacingTimeRem + ts*1000 - m_LastSendTime*1000, m_PacingTime);
m_NumPacketsToSend = numPackets.quot;
m_PacingTimeRem = numPackets.rem;
}
else
{
LogPrint (eLogError, "Streaming: pacing time is zero");
m_NumPacketsToSend = 1; m_PacingTimeRem = 0;
}
m_IsSendTime = true; m_IsSendTime = true;
if (m_WindowIncCounter && m_WindowSize < MAX_WINDOW_SIZE && !m_SendBuffer.IsEmpty ()) if (m_WindowIncCounter && m_WindowSize < MAX_WINDOW_SIZE && !m_SendBuffer.IsEmpty () && m_PacingTime > m_MinPacingTime)
{ {
for (int i = 0; i < m_NumPacketsToSend; i++) for (int i = 0; i < m_NumPacketsToSend; i++)
{ {
@ -1238,10 +1276,12 @@ namespace stream
else else
m_WindowSize += (m_WindowSize - (1 - PREV_SPEED_KEEP_TIME_COEFF)) / m_WindowSize; m_WindowSize += (m_WindowSize - (1 - PREV_SPEED_KEEP_TIME_COEFF)) / m_WindowSize;
if (m_WindowSize > MAX_WINDOW_SIZE) m_WindowSize = MAX_WINDOW_SIZE; if (m_WindowSize > MAX_WINDOW_SIZE) m_WindowSize = MAX_WINDOW_SIZE;
m_WindowIncCounter --; m_WindowIncCounter--;
UpdatePacingTime ();
} }
else
break;
} }
UpdatePacingTime ();
} }
if (m_IsNAcked) if (m_IsNAcked)
ResendPacket (); ResendPacket ();
@ -1366,6 +1406,7 @@ namespace stream
} }
else else
{ {
CancelRemoteLeaseChange ();
UpdateCurrentRemoteLease (); // pick another lease UpdateCurrentRemoteLease (); // pick another lease
LogPrint (eLogWarning, "Streaming: Resend #", m_NumResendAttempts, LogPrint (eLogWarning, "Streaming: Resend #", m_NumResendAttempts,
", another remote lease has been selected for stream with rSID=", m_RecvStreamID, ", sSID=", m_SendStreamID); ", another remote lease has been selected for stream with rSID=", m_RecvStreamID, ", sSID=", m_SendStreamID);
@ -1506,22 +1547,9 @@ namespace stream
LogPrint (eLogWarning, "Streaming: Remote LeaseSet not found"); LogPrint (eLogWarning, "Streaming: Remote LeaseSet not found");
m_CurrentRemoteLease = nullptr; m_CurrentRemoteLease = nullptr;
} }
if (isLeaseChanged) if (isLeaseChanged && !m_IsRemoteLeaseChangeInProgress)
{ {
// drop window to initial upon RemoteLease change HalveWindowSize ();
m_RTO = INITIAL_RTO;
if (m_WindowSize > INITIAL_WINDOW_SIZE)
{
m_WindowDropTargetSize = std::max (m_WindowSize/2, (float)INITIAL_WINDOW_SIZE);
m_IsWinDropped = true;
}
else
m_WindowSize = INITIAL_WINDOW_SIZE;
m_LastWindowDropSize = 0;
m_WindowIncCounter = 0;
m_IsFirstRttSample = true;
m_IsFirstACK = true;
UpdatePacingTime ();
} }
} }
@ -1560,6 +1588,29 @@ namespace stream
UpdatePacingTime (); UpdatePacingTime ();
} }
void Stream::HalveWindowSize ()
{
m_RTO = INITIAL_RTO;
if (m_WindowSize > INITIAL_WINDOW_SIZE)
{
m_WindowDropTargetSize = std::max (m_WindowSize/2, (float)INITIAL_WINDOW_SIZE);
m_IsWinDropped = true;
}
else
m_WindowSize = INITIAL_WINDOW_SIZE;
m_LastWindowDropSize = 0;
m_WindowIncCounter = 0;
m_IsFirstRttSample = true;
m_IsFirstACK = true;
UpdatePacingTime ();
}
void Stream::CancelRemoteLeaseChange ()
{
m_RemoteLeaseChangeTime = 0;
m_IsRemoteLeaseChangeInProgress = false;
}
StreamingDestination::StreamingDestination (std::shared_ptr<i2p::client::ClientDestination> owner, uint16_t localPort, bool gzip): StreamingDestination::StreamingDestination (std::shared_ptr<i2p::client::ClientDestination> owner, uint16_t localPort, bool gzip):
m_Owner (owner), m_LocalPort (localPort), m_Gzip (gzip), m_Owner (owner), m_LocalPort (localPort), m_Gzip (gzip),
m_PendingIncomingTimer (m_Owner->GetService ()), m_PendingIncomingTimer (m_Owner->GetService ()),

9
libi2pd/Streaming.h

@ -69,7 +69,8 @@ namespace stream
const int PENDING_INCOMING_TIMEOUT = 10; // in seconds const int PENDING_INCOMING_TIMEOUT = 10; // in seconds
const int MAX_RECEIVE_TIMEOUT = 20; // in seconds const int MAX_RECEIVE_TIMEOUT = 20; // in seconds
const uint16_t DELAY_CHOKING = 60000; // in milliseconds const uint16_t DELAY_CHOKING = 60000; // in milliseconds
const uint64_t SEND_INTERVAL = 1000; // in microseconds const uint64_t SEND_INTERVAL = 10000; // in microseconds
const uint64_t SEND_INTERVAL_VARIANCE = 2000; // in microseconds
const uint64_t REQUEST_IMMEDIATE_ACK_INTERVAL = 7500; // in milliseconds const uint64_t REQUEST_IMMEDIATE_ACK_INTERVAL = 7500; // in milliseconds
const uint64_t REQUEST_IMMEDIATE_ACK_INTERVAL_VARIANCE = 3200; // in milliseconds const uint64_t REQUEST_IMMEDIATE_ACK_INTERVAL_VARIANCE = 3200; // in milliseconds
const bool LOSS_BASED_CONTROL_ENABLED = 1; // 0/1 const bool LOSS_BASED_CONTROL_ENABLED = 1; // 0/1
@ -248,6 +249,8 @@ namespace stream
void UpdatePacingTime (); void UpdatePacingTime ();
void ProcessWindowDrop (); void ProcessWindowDrop ();
void HalveWindowSize ();
void CancelRemoteLeaseChange ();
private: private:
@ -268,12 +271,14 @@ namespace stream
bool m_IsWinDropped; bool m_IsWinDropped;
bool m_IsTimeOutResend; bool m_IsTimeOutResend;
bool m_IsImmediateAckRequested; bool m_IsImmediateAckRequested;
bool m_IsRemoteLeaseChangeInProgress;
StreamingDestination& m_LocalDestination; StreamingDestination& m_LocalDestination;
std::shared_ptr<const i2p::data::IdentityEx> m_RemoteIdentity; std::shared_ptr<const i2p::data::IdentityEx> m_RemoteIdentity;
std::shared_ptr<const i2p::crypto::Verifier> m_TransientVerifier; // in case of offline key std::shared_ptr<const i2p::crypto::Verifier> m_TransientVerifier; // in case of offline key
std::shared_ptr<const i2p::data::LeaseSet> m_RemoteLeaseSet; std::shared_ptr<const i2p::data::LeaseSet> m_RemoteLeaseSet;
std::shared_ptr<i2p::garlic::GarlicRoutingSession> m_RoutingSession; std::shared_ptr<i2p::garlic::GarlicRoutingSession> m_RoutingSession;
std::shared_ptr<const i2p::data::Lease> m_CurrentRemoteLease; std::shared_ptr<const i2p::data::Lease> m_CurrentRemoteLease;
std::shared_ptr<const i2p::data::Lease> m_NextRemoteLease;
std::shared_ptr<i2p::tunnel::OutboundTunnel> m_CurrentOutboundTunnel; std::shared_ptr<i2p::tunnel::OutboundTunnel> m_CurrentOutboundTunnel;
std::queue<Packet *> m_ReceiveQueue; std::queue<Packet *> m_ReceiveQueue;
std::set<Packet *, PacketCmp> m_SavedPackets; std::set<Packet *, PacketCmp> m_SavedPackets;
@ -289,7 +294,7 @@ namespace stream
int m_WindowIncCounter, m_RTO, m_AckDelay, m_PrevRTTSample; int m_WindowIncCounter, m_RTO, m_AckDelay, m_PrevRTTSample;
double m_Jitter; double m_Jitter;
uint64_t m_MinPacingTime, m_PacingTime, m_PacingTimeRem, // microseconds uint64_t m_MinPacingTime, m_PacingTime, m_PacingTimeRem, // microseconds
m_LastSendTime; // miliseconds m_LastSendTime, m_RemoteLeaseChangeTime; // miliseconds
uint64_t m_LastACKSendTime, m_PacketACKInterval, m_PacketACKIntervalRem; // for limit inbound speed uint64_t m_LastACKSendTime, m_PacketACKInterval, m_PacketACKIntervalRem; // for limit inbound speed
int m_NumResendAttempts, m_NumPacketsToSend; int m_NumResendAttempts, m_NumPacketsToSend;
size_t m_MTU; size_t m_MTU;

25
libi2pd/Transports.cpp

@ -672,6 +672,31 @@ namespace transport
if (transport & compatibleTransports) if (transport & compatibleTransports)
peer->priority.push_back (transport); peer->priority.push_back (transport);
} }
if (peer->priority.empty ())
{
// try recently connected SSU2 if any
auto supportedTransports = context.GetRouterInfo ().GetCompatibleTransports (false) &
peer->router->GetCompatibleTransports (false);
if (supportedTransports & (i2p::data::RouterInfo::eSSU2V4 | i2p::data::RouterInfo::eSSU2V6))
{
auto ep = peer->router->GetProfile ()->GetLastEndpoint ();
if (!ep.address ().is_unspecified () && ep.port ())
{
if (ep.address ().is_v4 ())
{
if ((supportedTransports & i2p::data::RouterInfo::eSSU2V4) &&
m_SSU2Server->IsConnectedRecently (ep, false))
peer->priority.push_back (i2p::data::RouterInfo::eSSU2V4);
}
else if (ep.address ().is_v6 ())
{
if ((supportedTransports & i2p::data::RouterInfo::eSSU2V6) &&
m_SSU2Server->IsConnectedRecently (ep))
peer->priority.push_back (i2p::data::RouterInfo::eSSU2V6);
}
}
}
}
} }
void Transports::RequestComplete (std::shared_ptr<const i2p::data::RouterInfo> r, const i2p::data::IdentHash& ident) void Transports::RequestComplete (std::shared_ptr<const i2p::data::RouterInfo> r, const i2p::data::IdentHash& ident)

44
libi2pd/Tunnel.cpp

@ -250,7 +250,18 @@ namespace tunnel
void InboundTunnel::HandleTunnelDataMsg (std::shared_ptr<I2NPMessage>&& msg) void InboundTunnel::HandleTunnelDataMsg (std::shared_ptr<I2NPMessage>&& msg)
{ {
if (GetState () != eTunnelStateExpiring) SetState (eTunnelStateEstablished); // incoming messages means a tunnel is alive if (!IsEstablished () && GetState () != eTunnelStateExpiring)
{
// incoming messages means a tunnel is alive
SetState (eTunnelStateEstablished);
auto pool = GetTunnelPool ();
if (pool)
{
// update LeaseSet
auto dest = pool->GetLocalDestination ();
if (dest) dest->SetLeaseSetUpdated (true);
}
}
EncryptTunnelMsg (msg, msg); EncryptTunnelMsg (msg, msg);
msg->from = GetSharedFromThis (); msg->from = GetSharedFromThis ();
m_Endpoint.HandleDecryptedTunnelDataMsg (msg); m_Endpoint.HandleDecryptedTunnelDataMsg (msg);
@ -339,7 +350,8 @@ namespace tunnel
Tunnels::Tunnels (): m_IsRunning (false), m_Thread (nullptr), m_MaxNumTransitTunnels (DEFAULT_MAX_NUM_TRANSIT_TUNNELS), Tunnels::Tunnels (): m_IsRunning (false), m_Thread (nullptr), m_MaxNumTransitTunnels (DEFAULT_MAX_NUM_TRANSIT_TUNNELS),
m_TotalNumSuccesiveTunnelCreations (0), m_TotalNumFailedTunnelCreations (0), // for normal average m_TotalNumSuccesiveTunnelCreations (0), m_TotalNumFailedTunnelCreations (0), // for normal average
m_TunnelCreationSuccessRate (TCSR_START_VALUE), m_TunnelCreationAttemptsNum(0) m_TunnelCreationSuccessRate (TCSR_START_VALUE), m_TunnelCreationAttemptsNum(0),
m_Rng(i2p::util::GetMonotonicMicroseconds ()%1000000LL)
{ {
} }
@ -479,18 +491,21 @@ namespace tunnel
std::this_thread::sleep_for (std::chrono::seconds(1)); // wait for other parts are ready std::this_thread::sleep_for (std::chrono::seconds(1)); // wait for other parts are ready
uint64_t lastTs = 0, lastPoolsTs = 0, lastMemoryPoolTs = 0; uint64_t lastTs = 0, lastPoolsTs = 0, lastMemoryPoolTs = 0;
std::list<std::shared_ptr<I2NPMessage> > msgs;
while (m_IsRunning) while (m_IsRunning)
{ {
try try
{ {
auto msg = m_Queue.GetNextWithTimeout (1000); // 1 sec if (m_Queue.Wait (1,0)) // 1 sec
if (msg)
{ {
m_Queue.GetWholeQueue (msgs);
int numMsgs = 0; int numMsgs = 0;
uint32_t prevTunnelID = 0, tunnelID = 0; uint32_t prevTunnelID = 0, tunnelID = 0;
std::shared_ptr<TunnelBase> prevTunnel; std::shared_ptr<TunnelBase> prevTunnel;
do while (!msgs.empty ())
{ {
auto msg = msgs.front (); msgs.pop_front ();
if (!msg) continue;
std::shared_ptr<TunnelBase> tunnel; std::shared_ptr<TunnelBase> tunnel;
uint8_t typeID = msg->GetTypeID (); uint8_t typeID = msg->GetTypeID ();
switch (typeID) switch (typeID)
@ -530,17 +545,18 @@ namespace tunnel
LogPrint (eLogWarning, "Tunnel: Unexpected message type ", (int) typeID); LogPrint (eLogWarning, "Tunnel: Unexpected message type ", (int) typeID);
} }
msg = (numMsgs <= MAX_TUNNEL_MSGS_BATCH_SIZE) ? m_Queue.Get () : nullptr; prevTunnelID = tunnelID;
if (msg) prevTunnel = tunnel;
numMsgs++;
if (msgs.empty ())
{ {
prevTunnelID = tunnelID; if (numMsgs < MAX_TUNNEL_MSGS_BATCH_SIZE && !m_Queue.IsEmpty ())
prevTunnel = tunnel; m_Queue.GetWholeQueue (msgs); // try more
numMsgs++; else if (tunnel)
tunnel->FlushTunnelDataMsgs (); // otherwise flush last
} }
else if (tunnel)
tunnel->FlushTunnelDataMsgs ();
} }
while (msg);
} }
if (i2p::transport::transports.IsOnline()) if (i2p::transport::transports.IsOnline())
@ -826,7 +842,7 @@ namespace tunnel
if (msg) m_Queue.Put (msg); if (msg) m_Queue.Put (msg);
} }
void Tunnels::PostTunnelData (const std::vector<std::shared_ptr<I2NPMessage> >& msgs) void Tunnels::PostTunnelData (std::list<std::shared_ptr<I2NPMessage> >& msgs)
{ {
m_Queue.Put (msgs); m_Queue.Put (msgs);
} }

6
libi2pd/Tunnel.h

@ -18,6 +18,7 @@
#include <thread> #include <thread>
#include <mutex> #include <mutex>
#include <memory> #include <memory>
#include <random>
#include "util.h" #include "util.h"
#include "Queue.h" #include "Queue.h"
#include "Crypto.h" #include "Crypto.h"
@ -229,7 +230,7 @@ namespace tunnel
std::shared_ptr<InboundTunnel> CreateInboundTunnel (std::shared_ptr<TunnelConfig> config, std::shared_ptr<TunnelPool> pool, std::shared_ptr<OutboundTunnel> outboundTunnel); std::shared_ptr<InboundTunnel> CreateInboundTunnel (std::shared_ptr<TunnelConfig> config, std::shared_ptr<TunnelPool> pool, std::shared_ptr<OutboundTunnel> outboundTunnel);
std::shared_ptr<OutboundTunnel> CreateOutboundTunnel (std::shared_ptr<TunnelConfig> config, std::shared_ptr<TunnelPool> pool); std::shared_ptr<OutboundTunnel> CreateOutboundTunnel (std::shared_ptr<TunnelConfig> config, std::shared_ptr<TunnelPool> pool);
void PostTunnelData (std::shared_ptr<I2NPMessage> msg); void PostTunnelData (std::shared_ptr<I2NPMessage> msg);
void PostTunnelData (const std::vector<std::shared_ptr<I2NPMessage> >& msgs); void PostTunnelData (std::list<std::shared_ptr<I2NPMessage> >& msgs); // and cleanup msgs
void AddPendingTunnel (uint32_t replyMsgID, std::shared_ptr<InboundTunnel> tunnel); void AddPendingTunnel (uint32_t replyMsgID, std::shared_ptr<InboundTunnel> tunnel);
void AddPendingTunnel (uint32_t replyMsgID, std::shared_ptr<OutboundTunnel> tunnel); void AddPendingTunnel (uint32_t replyMsgID, std::shared_ptr<OutboundTunnel> tunnel);
std::shared_ptr<TunnelPool> CreateTunnelPool (int numInboundHops, std::shared_ptr<TunnelPool> CreateTunnelPool (int numInboundHops,
@ -244,6 +245,8 @@ namespace tunnel
uint32_t GetMaxNumTransitTunnels () const { return m_MaxNumTransitTunnels; }; uint32_t GetMaxNumTransitTunnels () const { return m_MaxNumTransitTunnels; };
int GetCongestionLevel() const { return m_MaxNumTransitTunnels ? CONGESTION_LEVEL_FULL * m_TransitTunnels.size() / m_MaxNumTransitTunnels : CONGESTION_LEVEL_FULL; } int GetCongestionLevel() const { return m_MaxNumTransitTunnels ? CONGESTION_LEVEL_FULL * m_TransitTunnels.size() / m_MaxNumTransitTunnels : CONGESTION_LEVEL_FULL; }
std::mt19937& GetRng () { return m_Rng; };
private: private:
template<class TTunnel> template<class TTunnel>
@ -307,6 +310,7 @@ namespace tunnel
int m_TotalNumSuccesiveTunnelCreations, m_TotalNumFailedTunnelCreations; int m_TotalNumSuccesiveTunnelCreations, m_TotalNumFailedTunnelCreations;
double m_TunnelCreationSuccessRate; double m_TunnelCreationSuccessRate;
int m_TunnelCreationAttemptsNum; int m_TunnelCreationAttemptsNum;
std::mt19937 m_Rng;
public: public:

8
libi2pd/TunnelPool.cpp

@ -141,7 +141,7 @@ namespace tunnel
m_InboundTunnels.insert (createdTunnel); m_InboundTunnels.insert (createdTunnel);
} }
if (m_LocalDestination) if (m_LocalDestination)
m_LocalDestination->SetLeaseSetUpdated (); m_LocalDestination->SetLeaseSetUpdated (true);
} }
void TunnelPool::TunnelExpired (std::shared_ptr<InboundTunnel> expiredTunnel) void TunnelPool::TunnelExpired (std::shared_ptr<InboundTunnel> expiredTunnel)
@ -330,7 +330,7 @@ namespace tunnel
} }
if (num < m_NumInboundTunnels && m_NumInboundHops <= 0 && m_LocalDestination) // zero hops IB if (num < m_NumInboundTunnels && m_NumInboundHops <= 0 && m_LocalDestination) // zero hops IB
m_LocalDestination->SetLeaseSetUpdated (); // update LeaseSet immediately m_LocalDestination->SetLeaseSetUpdated (true); // update LeaseSet immediately
} }
void TunnelPool::TestTunnels () void TunnelPool::TestTunnels ()
@ -377,10 +377,10 @@ namespace tunnel
it.second.second->SetState (eTunnelStateTestFailed); it.second.second->SetState (eTunnelStateTestFailed);
} }
if (failed && m_LocalDestination) if (failed && m_LocalDestination)
m_LocalDestination->SetLeaseSetUpdated (); m_LocalDestination->SetLeaseSetUpdated (true);
} }
if (m_LocalDestination) if (m_LocalDestination)
m_LocalDestination->SetLeaseSetUpdated (); m_LocalDestination->SetLeaseSetUpdated (true);
} }
else if (it.second.second->GetState () != eTunnelStateExpiring) else if (it.second.second->GetState () != eTunnelStateExpiring)
it.second.second->SetState (eTunnelStateTestFailed); it.second.second->SetState (eTunnelStateTestFailed);

6
libi2pd_client/I2PTunnel.cpp

@ -1,5 +1,5 @@
/* /*
* Copyright (c) 2013-2023, The PurpleI2P Project * Copyright (c) 2013-2024, The PurpleI2P Project
* *
* This file is part of Purple i2pd project and licensed under BSD3 * This file is part of Purple i2pd project and licensed under BSD3
* *
@ -784,7 +784,7 @@ namespace client
} }
if (!found) if (!found)
{ {
LogPrint (eLogError, "I2PTunnel: Unable to resolve to compatible address"); LogPrint (eLogError, "I2PTunnel: Unable to resolve ", m_Address, " to compatible address");
return; return;
} }
@ -794,7 +794,7 @@ namespace client
Accept (); Accept ();
} }
else else
LogPrint (eLogError, "I2PTunnel: Unable to resolve server tunnel address: ", ecode.message ()); LogPrint (eLogError, "I2PTunnel: Unable to resolve server tunnel address ", m_Address, ": ", ecode.message ());
} }
void I2PServerTunnel::SetAccessList (const std::set<i2p::data::IdentHash>& accessList) void I2PServerTunnel::SetAccessList (const std::set<i2p::data::IdentHash>& accessList)

2
libi2pd_client/UDPTunnel.cpp

@ -203,7 +203,7 @@ namespace client
std::vector<std::shared_ptr<DatagramSessionInfo> > sessions; std::vector<std::shared_ptr<DatagramSessionInfo> > sessions;
std::lock_guard<std::mutex> lock (m_SessionsMutex); std::lock_guard<std::mutex> lock (m_SessionsMutex);
for (auto it: m_Sessions) for (const auto &it: m_Sessions)
{ {
auto s = it.second; auto s = it.second;
if (!s->m_Destination) continue; if (!s->m_Destination) continue;

Loading…
Cancel
Save