From 316e507df853a065dd15dc80d3130ea0139a0a98 Mon Sep 17 00:00:00 2001 From: orignal Date: Tue, 30 Dec 2025 15:00:00 -0500 Subject: [PATCH] implement i2p.streaming.maxConnsPerMinute tunnel param --- libi2pd/Destination.cpp | 98 ++++++------ libi2pd/Destination.h | 18 ++- libi2pd/Streaming.cpp | 263 ++++++++++++++++++------------- libi2pd/Streaming.h | 24 +-- libi2pd_client/ClientContext.cpp | 43 ++--- 5 files changed, 254 insertions(+), 192 deletions(-) diff --git a/libi2pd/Destination.cpp b/libi2pd/Destination.cpp index 93995d83..6100224b 100644 --- a/libi2pd/Destination.cpp +++ b/libi2pd/Destination.cpp @@ -59,7 +59,7 @@ namespace client if (ratchetsInboundTags && ratchetsInboundTags < i2p::garlic::ECIESX25519_MIN_NUM_GENERATED_TAGS) ratchetsInboundTags = i2p::garlic::ECIESX25519_MIN_NUM_GENERATED_TAGS; SetNumRatchetInboundTags (ratchetsInboundTags); - } + } auto explicitPeersStr = (*params)[I2CP_PARAM_EXPLICIT_PEERS]; if (!explicitPeersStr.empty ()) { @@ -124,7 +124,7 @@ namespace client int minLatency = 0; if (params->Get (I2CP_PARAM_MIN_TUNNEL_LATENCY, minLatency)) { - if (minLatency > 0 && maxLatency > 0) + if (minLatency > 0 && maxLatency > 0) { // set tunnel pool latency LogPrint(eLogInfo, "Destination: Requiring tunnel latency [", minLatency, "ms, ", maxLatency, "ms]"); @@ -168,7 +168,7 @@ namespace client SaveTags (); CleanUp (); // GarlicDestination } - + bool LeaseSetDestination::Reconfigure(const i2p::util::Mapping& params) { params.Get (I2CP_PARAM_DONT_PUBLISH_LEASESET, m_IsPublic); @@ -190,7 +190,7 @@ namespace client params.Get (I2CP_PARAM_MIN_TUNNEL_LATENCY, minLatency); int maxLatency = 0; params.Get (I2CP_PARAM_MAX_TUNNEL_LATENCY, maxLatency); - + SetNumTags (numTags); SetNumRatchetInboundTags (numRatchetInboundTags); pool->RequireLatency(minLatency, maxLatency); @@ -326,15 +326,15 @@ namespace client m_IncomingMsgsQueue.push_back (msg); } if (empty) - boost::asio::post (m_Service, [s = shared_from_this ()]() - { + boost::asio::post (m_Service, [s = shared_from_this ()]() + { std::list > receivedMsgs; { std::lock_guard l(s->m_IncomingMsgsQueueMutex); s->m_IncomingMsgsQueue.swap (receivedMsgs); } for (auto& it: receivedMsgs) - s->HandleGarlicMessage (it); + s->HandleGarlicMessage (it); }); } @@ -348,7 +348,7 @@ namespace client { I2NPMessageType typeID = (I2NPMessageType)(buf[I2NP_HEADER_TYPEID_OFFSET]); uint32_t msgID = bufbe32toh (buf + I2NP_HEADER_MSGID_OFFSET); - LeaseSetDestination::HandleCloveI2NPMessage (typeID, buf + I2NP_HEADER_SIZE, + LeaseSetDestination::HandleCloveI2NPMessage (typeID, buf + I2NP_HEADER_SIZE, GetI2NPMessageLength(buf, len) - I2NP_HEADER_SIZE, msgID, nullptr); } @@ -439,9 +439,9 @@ namespace client if (buf[DATABASE_STORE_TYPE_OFFSET] == i2p::data::NETDB_STORE_TYPE_LEASESET) leaseSet = std::make_shared (buf + offset, len - offset); // LeaseSet else - { - leaseSet = std::make_shared (buf[DATABASE_STORE_TYPE_OFFSET], - buf + offset, len - offset, true, shared_from_this (), + { + leaseSet = std::make_shared (buf[DATABASE_STORE_TYPE_OFFSET], + buf + offset, len - offset, true, shared_from_this (), from ? from->GetRemoteStaticKeyType () : GetPreferredCryptoType () ); // LeaseSet2 if (from) { @@ -451,9 +451,9 @@ namespace client { LogPrint (eLogError, "Destination: Remote LeaseSet static key mismatch"); leaseSet = nullptr; - } - } - } + } + } + } if (leaseSet && leaseSet->IsValid () && leaseSet->GetIdentHash () == key && !leaseSet->IsExpired ()) { if (leaseSet->GetIdentHash () != GetIdentHash ()) @@ -484,7 +484,7 @@ namespace client if (request->requestedBlindedKey) { auto ls2 = std::make_shared (buf + offset, len - offset, - request->requestedBlindedKey, shared_from_this (), + request->requestedBlindedKey, shared_from_this (), m_LeaseSetPrivKey ? ((const uint8_t *)*m_LeaseSetPrivKey) : nullptr, GetPreferredCryptoType ()); if (ls2->IsValid () && !ls2->IsExpired ()) { @@ -979,9 +979,10 @@ namespace client m_StreamingOutboundSpeed (DEFAULT_MAX_OUTBOUND_SPEED), m_StreamingInboundSpeed (DEFAULT_MAX_INBOUND_SPEED), m_StreamingMaxConcurrentStreams (DEFAULT_MAX_CONCURRENT_STREAMS), + m_StreamingMaxConnsPerMinute (DEFAULT_MAX_CONNS_PER_MINUTE), m_StreamingMaxWindowSize (i2p::stream::MAX_WINDOW_SIZE), - m_IsStreamingAnswerPings (DEFAULT_ANSWER_PINGS), m_IsStreamingDontSign (DEFAULT_DONT_SIGN), - m_LastPort (0), m_DatagramDestination (nullptr), m_RefCounter (0), + m_IsStreamingAnswerPings (DEFAULT_ANSWER_PINGS), m_IsStreamingDontSign (DEFAULT_DONT_SIGN), + m_LastPort (0), m_DatagramDestination (nullptr), m_RefCounter (0), m_LastPublishedTimestamp (0), m_ReadyChecker(service) { if (keys.IsOfflineSignature () && GetLeaseSetType () == i2p::data::NETDB_STORE_TYPE_LEASESET) @@ -1004,12 +1005,12 @@ namespace client i2p::data::CryptoKeyType cryptoType = std::stoi(it1); #if !OPENSSL_PQ if (cryptoType <= i2p::data::CRYPTO_KEY_TYPE_ECIES_X25519_AEAD) // skip PQ keys if not supported -#endif +#endif { if (!m_PreferredCryptoType && cryptoType) m_PreferredCryptoType = cryptoType; // first non-zero in the list encryptionKeyTypes.insert (cryptoType); - } + } } catch (std::exception& ex) { @@ -1024,7 +1025,7 @@ namespace client encryptionKeyTypes.insert ( { GetIdentity ()->GetCryptoKeyType (), #if OPENSSL_PQ i2p::data::CRYPTO_KEY_TYPE_ECIES_MLKEM768_X25519_AEAD, -#endif +#endif i2p::data::CRYPTO_KEY_TYPE_ECIES_X25519_AEAD }); // usually 0,4 or 0,6,4 if post quantum for (auto& it: encryptionKeyTypes) @@ -1052,12 +1053,13 @@ namespace client params->Get (I2CP_PARAM_STREAMING_MAX_OUTBOUND_SPEED, m_StreamingOutboundSpeed); params->Get (I2CP_PARAM_STREAMING_MAX_INBOUND_SPEED, m_StreamingInboundSpeed); params->Get (I2CP_PARAM_STREAMING_MAX_CONCURRENT_STREAMS, m_StreamingMaxConcurrentStreams); + params->Get (I2CP_PARAM_STREAMING_MAX_CONNS_PER_MINUTE, m_StreamingMaxConnsPerMinute); if (params->Get (I2CP_PARAM_STREAMING_MAX_WINDOW_SIZE, m_StreamingMaxWindowSize) && - (m_StreamingMaxWindowSize < i2p::stream::MIN_WINDOW_SIZE)) + (m_StreamingMaxWindowSize < i2p::stream::MIN_WINDOW_SIZE)) m_StreamingMaxWindowSize = i2p::stream::MIN_WINDOW_SIZE; params->Get (I2CP_PARAM_STREAMING_ANSWER_PINGS, m_IsStreamingAnswerPings); params->Get (I2CP_PARAM_STREAMING_DONT_SIGN, m_IsStreamingDontSign); - + if (GetLeaseSetType () == i2p::data::NETDB_STORE_TYPE_ENCRYPTED_LEASESET2) { // authentication for encrypted LeaseSet @@ -1161,9 +1163,9 @@ namespace client } break; case PROTOCOL_TYPE_DATAGRAM: - case PROTOCOL_TYPE_RAW: + case PROTOCOL_TYPE_RAW: case PROTOCOL_TYPE_DATAGRAM2: - case PROTOCOL_TYPE_DATAGRAM3: + case PROTOCOL_TYPE_DATAGRAM3: // datagram protocol if (m_DatagramDestination) m_DatagramDestination->HandleDataMessagePayload (fromPort, toPort, buf, length, buf[9], from); @@ -1358,15 +1360,15 @@ namespace client return nullptr; } - i2p::datagram::DatagramDestination * ClientDestination::CreateDatagramDestination (bool gzip, + i2p::datagram::DatagramDestination * ClientDestination::CreateDatagramDestination (bool gzip, i2p::datagram::DatagramVersion version) { if (!m_DatagramDestination) - { + { if (!GetNumRatchetInboundTags ()) SetNumRatchetInboundTags (i2p::garlic::ECIESX25519_MAX_NUM_GENERATED_TAGS); // set max tags if not specified m_DatagramDestination = new i2p::datagram::DatagramDestination (GetSharedFromThis (), gzip, version); - } + } return m_DatagramDestination; } @@ -1390,20 +1392,20 @@ namespace client std::string ident = GetIdentHash().ToBase32(); std::string path = i2p::fs::DataDirPath("destinations", ident + "." + std::to_string (keys->keyType) + ".dat"); std::ifstream f(path, std::ifstream::binary); - if (f) + if (f) { size_t len = 0; if (keys->keyType == i2p::data::CRYPTO_KEY_TYPE_ELGAMAL) len = 512; else if (keys->keyType == i2p::data::CRYPTO_KEY_TYPE_ECIES_X25519_AEAD) - { + { f.seekg (0, std::ios::end); len = f.tellg(); f.seekg (0, std::ios::beg); } - + if (len == 512) - { + { char pub[256], priv[256]; f.read (pub, 256); memcpy (keys->pub.data(), pub, keys->pub.size()); @@ -1414,8 +1416,8 @@ namespace client { f.read ((char *)keys->pub.data(), keys->pub.size()); f.read ((char *)keys->priv.data(), keys->priv.size()); - } - if (f) + } + if (f) return; else LogPrint(eLogWarning, "Destination: Can't read keys from ", path); @@ -1425,9 +1427,9 @@ namespace client memset (keys->priv.data (), 0, keys->priv.size ()); memset (keys->pub.data (), 0, keys->pub.size ()); keys->GenerateKeys (); - + std::ofstream f1 (path, std::ofstream::binary | std::ofstream::out); - if (f1) + if (f1) { f1.write ((char *)keys->pub.data (), keys->pub.size ()); f1.write ((char *)keys->priv.data (), keys->priv.size ()); @@ -1458,23 +1460,23 @@ namespace client { LogPrint (eLogError, "Destinations: No encryption keys"); return; - } - + } + i2p::data::LocalLeaseSet2::EncryptionKeys keySections; std::shared_ptr preferredSection; if (m_EncryptionKeys.size () == 1) preferredSection = m_EncryptionKeys.begin ()->second; // only key else - { + { for (const auto& it: m_EncryptionKeys) if (it.first == m_PreferredCryptoType) preferredSection = it.second; else keySections.push_back (it.second); - } - if (preferredSection) + } + if (preferredSection) keySections.push_front (preferredSection); // make preferred first - + auto publishedTimestamp = i2p::util::GetSecondsSinceEpoch (); if (publishedTimestamp <= m_LastPublishedTimestamp) { @@ -1509,10 +1511,10 @@ namespace client auto it = m_EncryptionKeys.find (preferredCrypto); if (it != m_EncryptionKeys.end ()) encryptionKey = it->second; - } + } if (!encryptionKey) encryptionKey = m_EncryptionKeys.rbegin ()->second; - } + } if (encryptionKey) return encryptionKey->decryptor->Decrypt (encrypted, data); else @@ -1524,18 +1526,18 @@ namespace client { #if __cplusplus >= 202002L // C++20 return m_EncryptionKeys.contains (keyType); -#else +#else return m_EncryptionKeys.count (keyType) > 0; -#endif +#endif } i2p::data::CryptoKeyType ClientDestination::GetRatchetsHighestCryptoType () const { if (m_EncryptionKeys.empty ()) return 0; - auto cryptoType = m_EncryptionKeys.rbegin ()->first; + auto cryptoType = m_EncryptionKeys.rbegin ()->first; return cryptoType >= i2p::data::CRYPTO_KEY_TYPE_ECIES_X25519_AEAD ? cryptoType : 0; } - + const uint8_t * ClientDestination::GetEncryptionPublicKey (i2p::data::CryptoKeyType keyType) const { auto it = m_EncryptionKeys.find (keyType); @@ -1571,7 +1573,7 @@ namespace client return false; } - RunnableClientDestination::RunnableClientDestination (const i2p::data::PrivateKeys& keys, + RunnableClientDestination::RunnableClientDestination (const i2p::data::PrivateKeys& keys, bool isPublic, const i2p::util::Mapping * params): RunnableService ("Destination"), ClientDestination (GetIOService (), keys, isPublic, params) { diff --git a/libi2pd/Destination.h b/libi2pd/Destination.h index 8565baaf..d5ed00a6 100644 --- a/libi2pd/Destination.h +++ b/libi2pd/Destination.h @@ -101,10 +101,12 @@ namespace client const int DEFAULT_STREAMING_PROFILE = STREAMING_PROFILE_BULK; const char I2CP_PARAM_STREAMING_MAX_CONCURRENT_STREAMS[] = "i2p.streaming.maxConcurrentStreams"; const int DEFAULT_MAX_CONCURRENT_STREAMS = 2048; + const char I2CP_PARAM_STREAMING_MAX_CONNS_PER_MINUTE[] ="i2p.streaming.maxConnsPerMinute"; // per dest + const int DEFAULT_MAX_CONNS_PER_MINUTE = 0; // unlimited const char I2CP_PARAM_STREAMING_MAX_WINDOW_SIZE[] = "i2p.streaming.maxWindowSize"; const char I2CP_PARAM_STREAMING_DONT_SIGN[] = "i2p.streaming.dontSign"; const int DEFAULT_DONT_SIGN = false; - + typedef std::function stream)> StreamRequestComplete; class LeaseSetDestination: public i2p::garlic::GarlicDestination, @@ -169,7 +171,7 @@ namespace client // implements GarlicDestination void HandleI2NPMessage (const uint8_t * buf, size_t len) override; - bool HandleCloveI2NPMessage (I2NPMessageType typeID, const uint8_t * payload, + bool HandleCloveI2NPMessage (I2NPMessageType typeID, const uint8_t * payload, size_t len, uint32_t msgID, i2p::garlic::ECIESX25519AEADRatchetSession * from) override; void SetLeaseSet (std::shared_ptr newLeaseSet); @@ -181,7 +183,7 @@ namespace client // I2CP virtual void HandleDataMessage (const uint8_t * buf, size_t len, i2p::garlic::ECIESX25519AEADRatchetSession * from) = 0; virtual void CreateNewLeaseSet (const std::vector >& tunnels) = 0; - + private: void UpdateLeaseSet (); @@ -210,7 +212,7 @@ namespace client std::list > m_IncomingMsgsQueue; mutable std::mutex m_IncomingMsgsQueueMutex; - + std::shared_ptr m_Pool; std::mutex m_LeaseSetMutex; std::shared_ptr m_LeaseSet; @@ -273,6 +275,7 @@ namespace client int GetStreamingOutboundSpeed () const { return m_StreamingOutboundSpeed; } int GetStreamingInboundSpeed () const { return m_StreamingInboundSpeed; } int GetStreamingMaxConcurrentStreams () const { return m_StreamingMaxConcurrentStreams; } + int GetStreamingMaxConnsPerMinute () const { return m_StreamingMaxConnsPerMinute; } bool IsStreamingAnswerPings () const { return m_IsStreamingAnswerPings; } bool IsStreamingDontSign () const { return m_IsStreamingDontSign; } int GetStreamingMaxWindowSize () const { return m_StreamingMaxWindowSize; } @@ -298,7 +301,7 @@ namespace client // I2CP void HandleDataMessage (const uint8_t * buf, size_t len, i2p::garlic::ECIESX25519AEADRatchetSession * from) override; void CreateNewLeaseSet (const std::vector >& tunnels) override; - + private: std::shared_ptr GetSharedFromThis () { @@ -315,8 +318,9 @@ namespace client i2p::data::PrivateKeys m_Keys; std::map > m_EncryptionKeys; // last is most preferable i2p::data::CryptoKeyType m_PreferredCryptoType; - - int m_StreamingAckDelay,m_StreamingOutboundSpeed, m_StreamingInboundSpeed, m_StreamingMaxConcurrentStreams, m_StreamingMaxWindowSize; + + int m_StreamingAckDelay,m_StreamingOutboundSpeed, m_StreamingInboundSpeed, + m_StreamingMaxConcurrentStreams, m_StreamingMaxConnsPerMinute, m_StreamingMaxWindowSize; bool m_IsStreamingAnswerPings, m_IsStreamingDontSign; std::shared_ptr m_StreamingDestination; // default std::map > m_StreamingDestinationsByPorts; diff --git a/libi2pd/Streaming.cpp b/libi2pd/Streaming.cpp index 0a00f142..a00ebd75 100644 --- a/libi2pd/Streaming.cpp +++ b/libi2pd/Streaming.cpp @@ -43,9 +43,9 @@ namespace stream m_Buffers.clear (); m_Size = 0; return offset; - } + } else - { + { while (!m_Buffers.empty () && offset < len) { auto nextBuffer = m_Buffers.front (); @@ -67,7 +67,7 @@ namespace stream } } m_Size -= offset; - } + } return offset; } @@ -87,21 +87,21 @@ namespace stream m_SendStreamID (0), m_SequenceNumber (0), m_DropWindowDelaySequenceNumber (INITIAL_WINDOW_SIZE), m_TunnelsChangeSequenceNumber (0), m_LastReceivedSequenceNumber (-1), m_PreviousReceivedSequenceNumber (-1), m_LastConfirmedReceivedSequenceNumber (0), // for limit inbound speed - m_Status (eStreamStatusNew), m_IsIncoming (false), m_IsAckSendScheduled (false), m_IsNAcked (false), m_IsFirstACK (false), - m_IsResendNeeded (false), m_IsFirstRttSample (false), m_IsSendTime (true), + m_Status (eStreamStatusNew), m_IsIncoming (false), m_IsAckSendScheduled (false), m_IsNAcked (false), m_IsFirstACK (false), + m_IsResendNeeded (false), m_IsFirstRttSample (false), m_IsSendTime (true), m_IsWinDropped (true), m_IsChoking2 (false), m_IsClientChoked (false), m_IsClientChoked2 (false), - m_IsTimeOutResend (false), m_IsImmediateAckRequested (false), m_IsRemoteLeaseChangeInProgress (false), - m_IsBufferEmpty (false), m_IsJavaClient (false), m_DontSign (local.GetOwner ()->IsStreamingDontSign ()), - m_LocalDestination (local), m_RemoteLeaseSet (remote), m_ReceiveTimer (m_Service), + m_IsTimeOutResend (false), m_IsImmediateAckRequested (false), m_IsRemoteLeaseChangeInProgress (false), + m_IsBufferEmpty (false), m_IsJavaClient (false), m_DontSign (local.GetOwner ()->IsStreamingDontSign ()), + m_LocalDestination (local), 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_RTT (INITIAL_RTT), m_MinRTT (INITIAL_RTT), + m_NumReceivedBytes (0), m_Port (port), m_RTT (INITIAL_RTT), m_MinRTT (INITIAL_RTT), m_SlowRTT (INITIAL_RTT), m_FastRTT (INITIAL_RTT), m_WindowSize (INITIAL_WINDOW_SIZE), - m_MaxWindowSize (local.GetOwner ()->GetStreamingMaxWindowSize ()), m_LastWindowDropSize (0), + m_MaxWindowSize (local.GetOwner ()->GetStreamingMaxWindowSize ()), m_LastWindowDropSize (0), m_WindowDropTargetSize (0), m_WindowIncCounter (0), m_RTO (INITIAL_RTO), - m_AckDelay (local.GetOwner ()->GetStreamingAckDelay ()), m_PrevRTTSample (INITIAL_RTT), - m_Jitter (0), m_MinPacingTime (0), m_PacingTime (INITIAL_PACING_TIME), m_PacingTimeRem (0), - m_LastSendTime (0), m_LastACKRecieveTime (0), m_ACKRecieveInterval (local.GetOwner ()->GetStreamingAckDelay ()), - m_RemoteLeaseChangeTime (0), m_LastWindowIncTime (0), m_LastACKRequestTime (0), m_LastACKSendTime (0), + m_AckDelay (local.GetOwner ()->GetStreamingAckDelay ()), m_PrevRTTSample (INITIAL_RTT), + m_Jitter (0), m_MinPacingTime (0), m_PacingTime (INITIAL_PACING_TIME), m_PacingTimeRem (0), + m_LastSendTime (0), m_LastACKRecieveTime (0), m_ACKRecieveInterval (local.GetOwner ()->GetStreamingAckDelay ()), + m_RemoteLeaseChangeTime (0), m_LastWindowIncTime (0), m_LastACKRequestTime (0), m_LastACKSendTime (0), m_PacketACKInterval (1), m_PacketACKIntervalRem (0), // for limit inbound speed m_NumResendAttempts (0), m_NumPacketsToSend (0), m_JitterAccum (0), m_JitterDiv (1), m_MTU (STREAMING_MTU) { @@ -110,7 +110,7 @@ namespace stream auto outboundSpeed = local.GetOwner ()->GetStreamingOutboundSpeed (); if (outboundSpeed) m_MinPacingTime = (1000000LL*STREAMING_MTU)/outboundSpeed; - + auto inboundSpeed = local.GetOwner ()->GetStreamingInboundSpeed (); // for limit inbound speed if (inboundSpeed) m_PacketACKInterval = (1000000LL*STREAMING_MTU)/inboundSpeed; @@ -120,19 +120,19 @@ namespace stream m_Service (service), m_SendStreamID (0), m_SequenceNumber (0), m_DropWindowDelaySequenceNumber (INITIAL_WINDOW_SIZE), m_TunnelsChangeSequenceNumber (0), m_LastReceivedSequenceNumber (-1), m_PreviousReceivedSequenceNumber (-1), m_LastConfirmedReceivedSequenceNumber (0), // for limit inbound speed - m_Status (eStreamStatusNew), m_IsIncoming (true), m_IsAckSendScheduled (false), m_IsNAcked (false), m_IsFirstACK (false), - m_IsResendNeeded (false), m_IsFirstRttSample (false), m_IsSendTime (true), + m_Status (eStreamStatusNew), m_IsIncoming (true), m_IsAckSendScheduled (false), m_IsNAcked (false), m_IsFirstACK (false), + m_IsResendNeeded (false), m_IsFirstRttSample (false), m_IsSendTime (true), m_IsWinDropped (true), m_IsChoking2 (false), m_IsClientChoked (false), m_IsClientChoked2 (false), m_IsTimeOutResend (false), m_IsImmediateAckRequested (false), m_IsRemoteLeaseChangeInProgress (false), m_IsBufferEmpty (false), m_IsJavaClient (false), m_DontSign (local.GetOwner ()->IsStreamingDontSign ()), m_LocalDestination (local),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_MinRTT (INITIAL_RTT), m_SlowRTT (INITIAL_RTT), m_FastRTT (INITIAL_RTT), - m_WindowSize (INITIAL_WINDOW_SIZE), m_MaxWindowSize (local.GetOwner ()->GetStreamingMaxWindowSize ()), + m_WindowSize (INITIAL_WINDOW_SIZE), m_MaxWindowSize (local.GetOwner ()->GetStreamingMaxWindowSize ()), m_LastWindowDropSize (0), m_WindowDropTargetSize (0), m_WindowIncCounter (0), m_RTO (INITIAL_RTO), - m_AckDelay (local.GetOwner ()->GetStreamingAckDelay ()),m_PrevRTTSample (INITIAL_RTT), m_Jitter (0), + m_AckDelay (local.GetOwner ()->GetStreamingAckDelay ()),m_PrevRTTSample (INITIAL_RTT), m_Jitter (0), m_MinPacingTime (0), m_PacingTime (INITIAL_PACING_TIME), m_PacingTimeRem (0), m_LastSendTime (0), - m_LastACKRecieveTime (0), m_ACKRecieveInterval (local.GetOwner ()->GetStreamingAckDelay ()), + m_LastACKRecieveTime (0), m_ACKRecieveInterval (local.GetOwner ()->GetStreamingAckDelay ()), m_RemoteLeaseChangeTime (0), m_LastWindowIncTime (0), m_LastACKRequestTime (0), m_LastACKSendTime (0), m_PacketACKInterval (1), m_PacketACKIntervalRem (0), // for limit inbound speed m_NumResendAttempts (0), m_NumPacketsToSend (0), m_JitterAccum (0), m_JitterDiv (1), m_MTU (STREAMING_MTU) @@ -141,7 +141,7 @@ namespace stream auto outboundSpeed = local.GetOwner ()->GetStreamingOutboundSpeed (); if (outboundSpeed) m_MinPacingTime = (1000000LL*STREAMING_MTU)/outboundSpeed; - + auto inboundSpeed = local.GetOwner ()->GetStreamingInboundSpeed (); // for limit inbound speed if (inboundSpeed) m_PacketACKInterval = (1000000LL*STREAMING_MTU)/inboundSpeed; @@ -166,7 +166,7 @@ namespace stream } void Stream::CleanUp () - { + { if (m_RoutingSession && !m_SentPackets.empty ()) // free up space in shared window { int numPackets = m_SentPackets.size (); @@ -175,7 +175,7 @@ namespace stream if (numSentPackets < 0) numSentPackets = 0; m_RoutingSession->SetNumSentPackets (numSentPackets); } - + m_SendBuffer.CleanUp (); while (!m_ReceiveQueue.empty ()) { @@ -183,7 +183,7 @@ namespace stream m_ReceiveQueue.pop (); m_LocalDestination.DeletePacket (packet); } - + m_NACKedPackets.clear (); for (auto it: m_SentPackets) @@ -201,10 +201,10 @@ namespace stream { m_LocalDestination.DeletePacket (packet); return; - } + } m_NumReceivedBytes += packet->GetLength (); if (!m_SendStreamID) - { + { m_SendStreamID = packet->GetReceiveStreamID (); if (!m_RemoteIdentity && !packet->from && packet->GetNACKCount () == 8 && // first incoming packet memcmp (packet->GetNACKs (), m_LocalDestination.GetOwner ()->GetIdentHash (), 32)) @@ -212,8 +212,8 @@ namespace stream LogPrint (eLogWarning, "Streaming: Destination mismatch for ", m_LocalDestination.GetOwner ()->GetIdentHash ().ToBase32 ()); m_LocalDestination.DeletePacket (packet); return; - } - } + } + } if (!packet->IsNoAck ()) // ack received ProcessAck (packet); @@ -225,7 +225,7 @@ namespace stream if (flags) // plain ack with options ProcessOptions (flags, packet); - else + else // plain ack { LogPrint (eLogDebug, "Streaming: Plain ACK received"); @@ -252,7 +252,7 @@ namespace stream // we have received next in sequence message ProcessPacket (packet); if (m_Status == eStreamStatusTerminated) return; - + // we should also try stored messages if any for (auto it = m_SavedPackets.begin (); it != m_SavedPackets.end ();) { @@ -329,7 +329,7 @@ namespace stream for (uint32_t i = nextSeqn; i < seqn; i++) numNacks++; nextSeqn = seqn + 1; } - + if (numNacks + (receivedSeqn - lastSavedSeq) >= 256) { m_LocalDestination.DeletePacket (packet); @@ -345,14 +345,14 @@ namespace stream if (m_LastReceivedSequenceNumber >= 0) { if (!m_IsAckSendScheduled) - { - // send NACKs for missing messages - SendQuickAck (); + { + // send NACKs for missing messages + SendQuickAck (); auto ackTimeout = m_RTT/10; if (ackTimeout > m_AckDelay) ackTimeout = m_AckDelay; ScheduleAck (ackTimeout); - } - } + } + } else // wait for SYN ScheduleAck (SYN_TIMEOUT); @@ -413,7 +413,7 @@ namespace stream { LogPrint (eLogInfo, "Streaming: Invalid option size ", optionSize, " Discarded"); return false; - } + } if (!flags) return true; bool immediateAckRequested = false; if (flags & PACKET_FLAG_DELAY_REQUESTED) @@ -475,11 +475,11 @@ namespace stream } optionData += m_RemoteIdentity->GetFullLen (); if (!m_RemoteLeaseSet) - { + { LogPrint (eLogDebug, "Streaming: Incoming stream from ", m_RemoteIdentity->GetIdentHash ().ToBase32 (), ", sSID=", m_SendStreamID, ", rSID=", m_RecvStreamID); - if (packet->from) // try to obtain LeaseSet if came from ratchets session + if (packet->from) // try to obtain LeaseSet if came from ratchets session m_RemoteLeaseSet = m_LocalDestination.GetOwner ()->FindLeaseSet (m_RemoteIdentity->GetIdentHash ()); - } + } if (packet->from && m_RemoteLeaseSet) { // stream came from ratchets session and static key must match one from LeaseSet @@ -487,10 +487,10 @@ namespace stream m_RemoteLeaseSet->Encrypt (nullptr, staticKey); if (memcmp (packet->from->GetRemoteStaticKey (), staticKey, 32)) { - LogPrint (eLogError, "Streaming: Remote LeaseSet static key mismatch for stream from ", + LogPrint (eLogError, "Streaming: Remote LeaseSet static key mismatch for stream from ", m_RemoteIdentity->GetIdentHash ().ToBase32 ()); return false; - } + } verified = true; if (!(flags & PACKET_FLAG_SIGNATURE_INCLUDED)) m_DontSign = true; // don't sign if the remote didn't sign @@ -508,7 +508,7 @@ namespace stream { verified = false; if (packet->from) - { + { if (!m_RemoteLeaseSet && m_RemoteIdentity) m_RemoteLeaseSet = m_LocalDestination.GetOwner ()->FindLeaseSet (m_RemoteIdentity->GetIdentHash ()); if (m_RemoteLeaseSet) @@ -517,17 +517,17 @@ namespace stream m_RemoteLeaseSet->Encrypt (nullptr, staticKey); if (memcmp (packet->from->GetRemoteStaticKey (), staticKey, 32)) { - LogPrint (eLogError, "Streaming: Remote LeaseSet static key mismatch for stream from ", + LogPrint (eLogError, "Streaming: Remote LeaseSet static key mismatch for stream from ", m_RemoteIdentity->GetIdentHash ().ToBase32 ()); return false; - } + } verified = true; - } + } else // invalid stream, safe to close verified = true; } - } - + } + if (flags & PACKET_FLAG_OFFLINE_SIGNATURE) { if (!m_RemoteIdentity) @@ -538,7 +538,7 @@ namespace stream if (verified) { // skip offline signature - optionData += 4; // timestamp + optionData += 4; // timestamp uint16_t keyType = bufbe16toh (optionData); optionData += 2; // key type std::unique_ptr transientVerifier (i2p::data::IdentityEx::CreateVerifier (keyType)); if (!transientVerifier) @@ -550,7 +550,7 @@ namespace stream optionData += m_RemoteIdentity->GetSignatureLen (); // signature } else - { + { // if we have it in LeaseSet already we don't need to parse it again if (m_RemoteLeaseSet) m_TransientVerifier = m_RemoteLeaseSet->GetTransientVerifier (); if (m_TransientVerifier) @@ -572,7 +572,7 @@ namespace stream return false; } } - } + } } if (flags & PACKET_FLAG_SIGNATURE_INCLUDED) @@ -582,9 +582,9 @@ namespace stream { LogPrint (eLogError, "Streaming: Signature too big, ", signatureLen, " bytes"); return false; - } + } if (!verified) // packet was not verified through session - { + { // verify actual signature if (signatureLen <= 256) { @@ -608,7 +608,7 @@ namespace stream m_TransientVerifier->Verify (packet->GetBuffer (), packet->GetLength (), signature.data ()) : m_RemoteIdentity->Verify (packet->GetBuffer (), packet->GetLength (), signature.data ()); } - } + } if (verified) optionData += signatureLen; else @@ -621,9 +621,9 @@ namespace stream { LogPrint (eLogError, "Streaming: Missing signature, sSID=", m_SendStreamID, ", rSID=", m_RecvStreamID); return false; - } - if (immediateAckRequested) - { + } + if (immediateAckRequested) + { auto ts = i2p::util::GetMillisecondsSinceEpoch (); if (m_LastACKSendTime != ts) // preventing multiple acks when reading m_SavedPackets { @@ -753,7 +753,7 @@ namespace stream { m_RTT = (m_PrevRTTSample + rttSample) / 2; } - if (!m_IsWinDropped) + if (!m_IsWinDropped) { m_SlowRTT = SLOWRTT_EWMA_ALPHA * m_RTT + (1.0 - SLOWRTT_EWMA_ALPHA) * m_SlowRTT; m_FastRTT = RTT_EWMA_ALPHA * m_RTT + (1.0 - RTT_EWMA_ALPHA) * m_FastRTT; @@ -792,10 +792,10 @@ namespace stream } UpdatePacingTime (); m_PrevRTTSample = rttSample; - + bool wasInitial = m_RTO == INITIAL_RTO; m_RTO = std::max (MIN_RTO, (int)(m_RTT * 1.3 + m_Jitter + m_ACKRecieveInterval)); // TODO: implement it better - + if (wasInitial) ScheduleResend (); } @@ -917,7 +917,7 @@ namespace stream if (buffer) s->m_SendBuffer.Add (std::move(buffer)); s->SendBuffer (); - }); + }); } void Stream::SendBuffer () @@ -990,18 +990,18 @@ namespace stream packet[size] = 0; size++; // NACK count } else - { + { packet[size] = 8; size++; // NACK count memcpy (packet + size, m_RemoteIdentity->GetIdentHash (), 32); size += 32; - } + } } else - { + { packet[size] = 0; size++; // NACK count - } + } packet[size] = m_RTO/1000; size++; // resend delay if (m_Status == eStreamStatusNew) @@ -1035,7 +1035,7 @@ namespace stream size += m_SendBuffer.Get (packet + size, m_MTU); // payload } else - { + { if (isOfflineSignature) { const auto& offlineSignature = m_LocalDestination.GetOwner ()->GetPrivateKeys ().GetOfflineSignature (); @@ -1048,7 +1048,7 @@ namespace stream htobe16buf (optionsSize, packet + size - 2 - optionsSize); // actual options size size += m_SendBuffer.Get (packet + size, m_MTU); // payload m_LocalDestination.GetOwner ()->Sign (packet, size, signature); - } + } } else { @@ -1094,7 +1094,7 @@ namespace stream } SendPackets (packets); m_LastSendTime = ts; - m_IsSendTime = false; + m_IsSendTime = false; if (m_RoutingSession) { int numSentPackets = m_RoutingSession->NumSentPackets (); @@ -1137,7 +1137,7 @@ namespace stream { auto seqn = it->GetSeqn (); // for limit inbound speed - if (m_LastConfirmedReceivedSequenceNumber + numPackets < int(seqn)) + if (m_LastConfirmedReceivedSequenceNumber + numPackets < int(seqn)) { if (!m_IsAckSendScheduled) { @@ -1226,7 +1226,7 @@ namespace stream size++; // NACK count } packet[size] = 0; - size++; // resend delay + size++; // resend delay bool requestImmediateAck = false; if (!choking) requestImmediateAck = m_LastSendTime && ts > m_LastSendTime + REQUEST_IMMEDIATE_ACK_INTERVAL && @@ -1246,8 +1246,8 @@ namespace stream m_LastSendTime = ts; m_IsImmediateAckRequested = true; } - } - else + } + else htobuf16 (packet + size, 0); // no options size += 2; // options size p.len = size; @@ -1282,7 +1282,7 @@ namespace stream if (m_DontSign) htobe16buf (optionsSize, packet + size - 2 - optionsSize); // actual options size else - { + { if (isOfflineSignature) { const auto& offlineSignature = m_LocalDestination.GetOwner ()->GetPrivateKeys ().GetOfflineSignature (); @@ -1295,7 +1295,7 @@ namespace stream size += signatureLen; // signature htobe16buf (optionsSize, packet + size - 2 - optionsSize); // actual options size m_LocalDestination.GetOwner ()->Sign (packet, size, signature); - } + } p.len = size; SendPackets (std::vector { &p }); LogPrint (eLogDebug, "Streaming: Ping of ", p.len, " bytes sent"); @@ -1359,9 +1359,9 @@ namespace stream { memset (packet + size, 0, 2); // no options size += 2; // options size - } + } else - { + { if (isOfflineSignature) { const auto& offlineSignature = m_LocalDestination.GetOwner ()->GetPrivateKeys ().GetOfflineSignature (); @@ -1375,7 +1375,7 @@ namespace stream memset (packet + size, 0, signatureLen); size += signatureLen; // signature m_LocalDestination.GetOwner ()->Sign (packet, size, signature); - } + } p->len = size; boost::asio::post (m_Service, std::bind (&Stream::SendPacket, shared_from_this (), p)); @@ -1440,15 +1440,15 @@ namespace stream } } if (!m_RoutingSession || m_RoutingSession->IsTerminated () || !m_RoutingSession->IsReadyToSend ()) // expired and detached or new session sent - { + { m_RoutingSession = m_LocalDestination.GetOwner ()->GetRoutingSession (m_RemoteLeaseSet, true, !m_IsIncoming || m_SequenceNumber > 1); if (!m_RoutingSession) { LogPrint (eLogError, "Streaming: Can't obtain routing session, sSID=", m_SendStreamID); Terminate (); return; - } - } + } + } if (!m_CurrentOutboundTunnel && m_RoutingSession) // first message to send { // try to get shared path first @@ -1579,7 +1579,7 @@ namespace stream shared_from_this (), std::placeholders::_1)); } } - + void Stream::HandleSendTimer (const boost::system::error_code& ecode) { if (ecode != boost::asio::error::operation_aborted) @@ -1588,16 +1588,16 @@ namespace stream if (m_LastSendTime && ts*1000 > m_LastSendTime*1000 + m_PacingTime) { if (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; if (m_WindowIncCounter && (m_WindowSize < m_MaxWindowSize || m_WindowDropTargetSize) && !m_SendBuffer.IsEmpty () && m_PacingTime > m_MinPacingTime) { @@ -1687,7 +1687,7 @@ namespace stream ResendPacket (); // send one packet per RTO, waiting for ack } } - + void Stream::ResendPacket () { // check for resend attempts @@ -1743,7 +1743,7 @@ namespace stream } } } - + // select tunnels if necessary and send if (packets.size () > 0 && m_IsSendTime) { @@ -1819,8 +1819,8 @@ namespace stream m_AckSendTimer.expires_from_now (boost::posix_time::milliseconds(timeout)); m_AckSendTimer.async_wait (std::bind (&Stream::HandleAckSendTimer, shared_from_this (), std::placeholders::_1)); - } - + } + void Stream::HandleAckSendTimer (const boost::system::error_code& ecode) { if (m_IsAckSendScheduled) @@ -1838,11 +1838,11 @@ namespace stream { auto ts = i2p::util::GetMillisecondsSinceEpoch (); if (ts > m_RoutingSession->GetLeaseSetSubmissionTime () + i2p::garlic::LEASESET_CONFIRMATION_TIMEOUT) - { + { // seems something went wrong and we should re-select tunnels m_CurrentOutboundTunnel = nullptr; m_CurrentRemoteLease = nullptr; - } + } } if (m_LastReceivedSequenceNumber == 0 && m_SequenceNumber == 1) { @@ -1871,7 +1871,7 @@ namespace stream { LogPrint (eLogWarning, "Streaming: LeaseSet ", m_RemoteIdentity->GetIdentHash ().ToBase32 (), m_RemoteLeaseSet ? " expired" : " not found"); if (!m_IsIncoming) // outgoing - { + { auto requestCallback = [s = shared_from_this ()](std::shared_ptr ls) { if (!ls && s->m_Status == eStreamStatusOpen) // LeaseSet not found @@ -1879,9 +1879,9 @@ namespace stream // close the socket without sending FIN or RST s->m_Status = eStreamStatusClosed; s->AsyncClose (); - } + } }; - + if (m_RemoteLeaseSet && m_RemoteLeaseSet->IsPublishedEncrypted ()) { m_LocalDestination.GetOwner ()->RequestDestinationWithEncryptedLeaseSet ( @@ -1891,7 +1891,7 @@ namespace stream else { m_RemoteLeaseSet = nullptr; - m_LocalDestination.GetOwner ()->RequestDestination (m_RemoteIdentity->GetIdentHash (), requestCallback); // try to request for a next attempt + m_LocalDestination.GetOwner ()->RequestDestination (m_RemoteIdentity->GetIdentHash (), requestCallback); // try to request for a next attempt } } else // incoming @@ -1899,7 +1899,7 @@ namespace stream // just close the socket without sending FIN or RST m_Status = eStreamStatusClosed; AsyncClose (); - } + } } else { @@ -1980,7 +1980,7 @@ namespace stream m_RTO = INITIAL_RTO; if (m_RoutingSession) m_RoutingSession->SetSharedRoutingPath (nullptr); // TODO: count failures - } + } void Stream::UpdatePacingTime () { @@ -1991,7 +1991,7 @@ namespace stream m_PacingTime = std::round (rtt*1000/m_WindowSize); if (m_MinPacingTime && m_PacingTime < m_MinPacingTime) m_PacingTime = m_MinPacingTime; - } + } void Stream::ProcessWindowDrop () { @@ -2038,8 +2038,8 @@ namespace stream } StreamingDestination::StreamingDestination (std::shared_ptr owner, uint16_t localPort, bool gzip): - m_Owner (owner), m_LocalPort (localPort), m_Gzip (gzip), - m_PendingIncomingTimer (m_Owner->GetService ()), + m_Owner (owner), m_LocalPort (localPort), m_Gzip (gzip), + m_PendingIncomingTimer (m_Owner->GetService ()), m_LastCleanupTime (i2p::util::GetSecondsSinceEpoch ()) { } @@ -2128,9 +2128,9 @@ namespace stream if (m_Owner->GetStreamingMaxConcurrentStreams () > 0 && (int)m_Streams.size () > m_Owner->GetStreamingMaxConcurrentStreams ()) { LogPrint(eLogWarning, "Streaming: Number of streams exceeds ", m_Owner->GetStreamingMaxConcurrentStreams ()); - DeletePacket (packet); + DeletePacket (packet); return; - } + } auto incomingStream = CreateNewIncomingStream (receiveStreamID); incomingStream->HandleNextPacket (packet); // SYN if (!incomingStream->GetRemoteLeaseSet ()) @@ -2138,7 +2138,26 @@ namespace stream LogPrint (eLogWarning, "Streaming: No remote LeaseSet for incoming stream. Terminated"); incomingStream->Terminate (); // can't send FIN anyway return; - } + } + if (m_Owner->GetStreamingMaxConnsPerMinute () > 0) + { + auto ts = i2p::util::GetSecondsSinceEpoch (); + auto ident = incomingStream->GetRemoteIdentity (); + if (ident) + { + auto& numConnectionsList = m_NumIncomingConnectionsPerSecond[ident->GetIdentHash ()]; // find or create new + CleanupExpiredNumConnectionsPerSecond (numConnectionsList, ts); + if ((int)numConnectionsList.size () >= m_Owner->GetStreamingMaxConnsPerMinute ()) + { + LogPrint (eLogWarning, "Streaming: Number of incoming streams from ",ident->GetIdentHash ().ToBase32 (), + " exceeds ", m_Owner->GetStreamingMaxConnsPerMinute (), " streams per minute"); + incomingStream->Terminate (); + return; + } + else + numConnectionsList.emplace_back (ts); + } + } // handle saved packets if any { @@ -2249,6 +2268,26 @@ namespace stream { m_PacketsPool.CleanUp (); m_I2NPMsgsPool.CleanUp (); + if (!m_NumIncomingConnectionsPerSecond.empty ()) + { + if (stream) + { + auto ident = stream->GetRemoteIdentity (); + if (ident) + { + auto it = m_NumIncomingConnectionsPerSecond.find (ident->GetIdentHash ()); + if (it != m_NumIncomingConnectionsPerSecond.end ()) + CleanupExpiredNumConnectionsPerSecond (it->second, ts); + } + } + for (auto it = m_NumIncomingConnectionsPerSecond.begin (); it != m_NumIncomingConnectionsPerSecond.end ();) + { + if (it->second.empty ()) + it = m_NumIncomingConnectionsPerSecond.erase (it); + else + it++; + } + } m_LastCleanupTime = ts; } } @@ -2293,12 +2332,12 @@ namespace stream { m_PongHandler = handler; } - + void StreamingDestination::ResetPongHandler () { m_PongHandler = nullptr; } - + void StreamingDestination::AcceptOnce (const Acceptor& acceptor) { boost::asio::post (m_Owner->GetService (), [acceptor, this](void) @@ -2366,7 +2405,7 @@ namespace stream { uncompressed->from = from; HandleNextPacket (uncompressed); - } + } else DeletePacket (uncompressed); } @@ -2402,6 +2441,18 @@ namespace stream uint32_t StreamingDestination::GetRandom () { return m_Owner ? m_Owner->GetRng ()() : rand (); - } + } + + void StreamingDestination::CleanupExpiredNumConnectionsPerSecond (std::list& numConnectionsList, uint64_t ts) + { + if (numConnectionsList.empty ()) return; + auto it = numConnectionsList.begin (); + while (it != numConnectionsList.end ()) + { + if (*it + 60 >= ts) break; // 1 minute + it++; + } + numConnectionsList.erase (numConnectionsList.begin (), it); + } } } diff --git a/libi2pd/Streaming.h b/libi2pd/Streaming.h index a251e890..b53b73d8 100644 --- a/libi2pd/Streaming.h +++ b/libi2pd/Streaming.h @@ -52,11 +52,11 @@ namespace stream const size_t STREAMING_MTU = 1730; const size_t STREAMING_MTU_RATCHETS = 1812; -#if OPENSSL_PQ +#if OPENSSL_PQ const size_t MAX_PACKET_SIZE = 8192; #else const size_t MAX_PACKET_SIZE = 4096; -#endif +#endif const size_t COMPRESSION_THRESHOLD_SIZE = 66; const int MAX_NUM_RESEND_ATTEMPTS = 10; const int INITIAL_WINDOW_SIZE = 10; @@ -80,11 +80,11 @@ namespace stream const uint16_t DELAY_CHOKING_2 = 65535; // in milliseconds 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_VARIANCE = 3200; // in milliseconds + const uint64_t REQUEST_IMMEDIATE_ACK_INTERVAL = 7500; // in milliseconds + const uint64_t REQUEST_IMMEDIATE_ACK_INTERVAL_VARIANCE = 3200; // in milliseconds const bool LOSS_BASED_CONTROL_ENABLED = 0; // 0/1 const uint64_t STREAMING_DESTINATION_POOLS_CLEANUP_INTERVAL = 646; // in seconds - + struct Packet { size_t len, offset; @@ -262,7 +262,7 @@ namespace stream void ProcessWindowDrop (); void ResetWindowSize (); void CancelRemoteLeaseChange (); - + private: boost::asio::io_context& m_Service; @@ -333,7 +333,7 @@ namespace stream std::shared_ptr AcceptStream (int timeout = 0); // sync void SetPongHandler (const PongHandler& handler); void ResetPongHandler (); - + std::shared_ptr GetOwner () const { return m_Owner; }; void SetOwner (std::shared_ptr owner) { m_Owner = owner; }; uint16_t GetLocalPort () const { return m_LocalPort; }; @@ -351,8 +351,13 @@ namespace stream std::shared_ptr CreateNewIncomingStream (uint32_t receiveStreamID); void HandlePendingIncomingTimer (const boost::system::error_code& ecode); + void CleanupExpiredNumConnectionsPerSecond (std::list& numConnectionsList, uint64_t ts); + private: + i2p::util::MemoryPool m_PacketsPool; + i2p::util::MemoryPool > m_I2NPMsgsPool; + std::shared_ptr m_Owner; uint16_t m_LocalPort; bool m_Gzip; // gzip compression of data messages @@ -366,10 +371,9 @@ namespace stream boost::asio::deadline_timer m_PendingIncomingTimer; std::unordered_map > m_SavedPackets; // receiveStreamID->packets, arrived before SYN - i2p::util::MemoryPool m_PacketsPool; - i2p::util::MemoryPool > m_I2NPMsgsPool; uint64_t m_LastCleanupTime; // in seconds - + std::unordered_map > m_NumIncomingConnectionsPerSecond; // dest-> list of timestamps in seconds + public: i2p::data::GzipInflator m_Inflator; diff --git a/libi2pd_client/ClientContext.cpp b/libi2pd_client/ClientContext.cpp index 0fb81dd0..e55fe217 100644 --- a/libi2pd_client/ClientContext.cpp +++ b/libi2pd_client/ClientContext.cpp @@ -270,10 +270,10 @@ namespace client { #if __cplusplus >= 202002L // C++20 if (filename.starts_with ("transient")) -#else +#else std::string_view transient("transient"); if (!filename.compare (0, transient.length (), transient)) // starts with transient -#endif +#endif { keys = i2p::data::PrivateKeys::CreateRandomKeys (sigType, cryptoType, true); LogPrint (eLogInfo, "Clients: New transient keys address ", m_AddressBook.ToAddress(keys.GetPublic ()->GetIdentHash ()), " created"); @@ -359,7 +359,7 @@ namespace client return localDestination; } - std::shared_ptr ClientContext::CreateNewMatchedTunnelDestination(const i2p::data::PrivateKeys &keys, + std::shared_ptr ClientContext::CreateNewMatchedTunnelDestination(const i2p::data::PrivateKeys &keys, const std::string & name, const i2p::util::Mapping * params) { auto localDestination = std::make_shared(keys, name, params); @@ -424,7 +424,7 @@ namespace client i2p::util::Mapping params; ReadI2CPOptionsFromConfig ("shareddest.", params); params.Insert (I2CP_PARAM_OUTBOUND_NICKNAME, "SharedDest"); - + m_SharedLocalDestination = CreateNewLocalDestination (false, i2p::data::SIGNING_KEY_TYPE_EDDSA_SHA512_ED25519, i2p::data::CRYPTO_KEY_TYPE_ELGAMAL, ¶ms); // non-public, EDDSA m_SharedLocalDestination->Acquire (); @@ -451,7 +451,7 @@ namespace client } template - void ClientContext::ReadI2CPOptionsGroup (const Section& section, const std::string& group, + void ClientContext::ReadI2CPOptionsGroup (const Section& section, const std::string& group, i2p::util::Mapping& options) const { for (auto it: section.second) @@ -477,6 +477,7 @@ namespace client options.Insert (I2CP_PARAM_STREAMING_MAX_OUTBOUND_SPEED, GetI2CPOption(section, I2CP_PARAM_STREAMING_MAX_OUTBOUND_SPEED, DEFAULT_MAX_OUTBOUND_SPEED)); options.Insert (I2CP_PARAM_STREAMING_MAX_INBOUND_SPEED, GetI2CPOption(section, I2CP_PARAM_STREAMING_MAX_INBOUND_SPEED, DEFAULT_MAX_INBOUND_SPEED)); options.Insert (I2CP_PARAM_STREAMING_MAX_CONCURRENT_STREAMS, GetI2CPOption(section, I2CP_PARAM_STREAMING_MAX_CONCURRENT_STREAMS, DEFAULT_MAX_CONCURRENT_STREAMS)); + options.Insert (I2CP_PARAM_STREAMING_MAX_CONNS_PER_MINUTE, GetI2CPOption(section, I2CP_PARAM_STREAMING_MAX_CONNS_PER_MINUTE, DEFAULT_MAX_CONNS_PER_MINUTE)); options.Insert (I2CP_PARAM_STREAMING_ANSWER_PINGS, GetI2CPOption(section, I2CP_PARAM_STREAMING_ANSWER_PINGS, isServer ? DEFAULT_ANSWER_PINGS : false)); options.Insert (I2CP_PARAM_STREAMING_DONT_SIGN, GetI2CPOption(section, I2CP_PARAM_STREAMING_DONT_SIGN, DEFAULT_DONT_SIGN)); options.Insert (I2CP_PARAM_STREAMING_PROFILE, GetI2CPOption(section, I2CP_PARAM_STREAMING_PROFILE, DEFAULT_STREAMING_PROFILE)); @@ -484,9 +485,9 @@ namespace client options.Insert (I2CP_PARAM_LEASESET_TYPE, GetI2CPOption(section, I2CP_PARAM_LEASESET_TYPE, DEFAULT_LEASESET_TYPE)); #if OPENSSL_PQ std::string encType = GetI2CPStringOption(section, I2CP_PARAM_LEASESET_ENCRYPTION_TYPE, isServer ? "6,4" : "6,4,0"); -#else +#else std::string encType = GetI2CPStringOption(section, I2CP_PARAM_LEASESET_ENCRYPTION_TYPE, isServer ? "4" : "4,0"); -#endif +#endif if (encType.length () > 0) options.Insert (I2CP_PARAM_LEASESET_ENCRYPTION_TYPE, encType); std::string privKey = GetI2CPStringOption(section, I2CP_PARAM_LEASESET_PRIV_KEY, ""); if (privKey.length () > 0) options.Insert (I2CP_PARAM_LEASESET_PRIV_KEY, privKey); @@ -559,9 +560,9 @@ namespace client { #if __cplusplus >= 202002L // C++20 if (!it.ends_with (".conf")) continue; -#else +#else if (it.substr(it.size() - 5) != ".conf") continue; // skip files which not ends with ".conf" -#endif +#endif LogPrint(eLogDebug, "Clients: Tunnels extra config file: ", it); ReadTunnels (it, numClientTunnels, numServerTunnels); } @@ -608,7 +609,7 @@ namespace client i2p::data::SigningKeyType sigType = section.second.get (I2P_CLIENT_TUNNEL_SIGNATURE_TYPE, i2p::data::SIGNING_KEY_TYPE_EDDSA_SHA512_ED25519); #if !OPENSSL_PQ if (sigType >= i2p::data::SIGNING_KEY_TYPE_MLDSA44) sigType = i2p::data::SIGNING_KEY_TYPE_EDDSA_SHA512_ED25519; -#endif +#endif i2p::data::CryptoKeyType cryptoType = section.second.get (I2P_CLIENT_TUNNEL_CRYPTO_TYPE, i2p::data::CRYPTO_KEY_TYPE_ELGAMAL); // I2CP i2p::util::Mapping options; @@ -645,7 +646,7 @@ namespace client } } - if (type == I2P_TUNNELS_SECTION_TYPE_UDPCLIENT) + if (type == I2P_TUNNELS_SECTION_TYPE_UDPCLIENT) { // udp client // TODO: hostnames @@ -655,7 +656,7 @@ namespace client bool gzip = section.second.get (I2P_CLIENT_TUNNEL_GZIP, true); int datagramVersion = (i2p::datagram::DatagramVersion)section.second.get (UDP_CLIENT_TUNNEL_DATAGRAM_VERSION, (int)i2p::datagram::eDatagramV3); - auto clientTunnel = std::make_shared (name, dest, end, + auto clientTunnel = std::make_shared (name, dest, end, localDestination, destinationPort, gzip, (i2p::datagram::DatagramVersion)datagramVersion); auto ins = m_ClientForwards.insert (std::make_pair (end, clientTunnel)); @@ -678,8 +679,8 @@ namespace client LogPrint(eLogError, "Clients: I2P Client forward for endpoint ", end, " already exists"); } - } - else + } + else { boost::asio::ip::tcp::endpoint clientEndpoint; std::shared_ptr clientTunnel; @@ -788,7 +789,7 @@ namespace client if (keys == "shareddest") localDestination = m_SharedLocalDestination; else - { + { auto it = destinations.find (keys); if (it != destinations.end ()) { @@ -809,7 +810,7 @@ namespace client else localDestination->SetPublic (true); } - } + } if (type == I2P_TUNNELS_SECTION_TYPE_UDPSERVER) { // udp server tunnel @@ -849,16 +850,16 @@ namespace client std::shared_ptr serverTunnel; if (type == I2P_TUNNELS_SECTION_TYPE_HTTP) - { + { std::string hostOverride = section.second.get (I2P_SERVER_TUNNEL_HOST_OVERRIDE, ""); bool i2pheaders = section.second.get (I2P_SERVER_TUNNEL_I2P_HEADERS, true); serverTunnel = std::make_shared (name, host, port, localDestination, hostOverride, inPort, gzip, i2pheaders); - } + } else if (type == I2P_TUNNELS_SECTION_TYPE_IRC) { std::string webircpass = section.second.get (I2P_SERVER_TUNNEL_WEBIRC_PASSWORD, ""); serverTunnel = std::make_shared (name, host, port, localDestination, webircpass, inPort, gzip); - } + } else // regular server tunnel by default serverTunnel = std::make_shared (name, host, port, localDestination, inPort, gzip); @@ -937,7 +938,7 @@ namespace client i2p::data::SigningKeyType sigType; i2p::config::GetOption("httpproxy.signaturetype", sigType); #if !OPENSSL_PQ if (sigType >= i2p::data::SIGNING_KEY_TYPE_MLDSA44) sigType = i2p::data::SIGNING_KEY_TYPE_EDDSA_SHA512_ED25519; -#endif +#endif LogPrint(eLogInfo, "Clients: Starting HTTP Proxy at ", httpProxyAddr, ":", httpProxyPort); if (httpProxyKeys == "shareddest") { @@ -989,7 +990,7 @@ namespace client i2p::data::SigningKeyType sigType; i2p::config::GetOption("socksproxy.signaturetype", sigType); #if !OPENSSL_PQ if (sigType >= i2p::data::SIGNING_KEY_TYPE_MLDSA44) sigType = i2p::data::SIGNING_KEY_TYPE_EDDSA_SHA512_ED25519; -#endif +#endif LogPrint(eLogInfo, "Clients: Starting SOCKS Proxy at ", socksProxyAddr, ":", socksProxyPort); if (socksProxyKeys == "shareddest") {