From fd9229c4676b158137532bb61aee0ffb84408263 Mon Sep 17 00:00:00 2001 From: orignal Date: Wed, 30 Sep 2020 17:12:28 -0400 Subject: [PATCH] ping/pong for streaming --- libi2pd/Destination.cpp | 10 +++++++--- libi2pd/Destination.h | 6 +++++- libi2pd/Streaming.cpp | 31 +++++++++++++++++++++++++++++++ libi2pd/Streaming.h | 2 ++ libi2pd_client/ClientContext.cpp | 1 + 5 files changed, 46 insertions(+), 4 deletions(-) diff --git a/libi2pd/Destination.cpp b/libi2pd/Destination.cpp index cd0ad9ab..af3b4cbc 100644 --- a/libi2pd/Destination.cpp +++ b/libi2pd/Destination.cpp @@ -849,8 +849,9 @@ namespace client ClientDestination::ClientDestination (boost::asio::io_service& service, const i2p::data::PrivateKeys& keys, bool isPublic, const std::map * params): - LeaseSetDestination (service, isPublic, params), - m_Keys (keys), m_StreamingAckDelay (DEFAULT_INITIAL_ACK_DELAY), + LeaseSetDestination (service, isPublic, params), + m_Keys (keys), m_StreamingAckDelay (DEFAULT_INITIAL_ACK_DELAY), + m_IsStreamingAnswerPings (DEFAULT_ANSWER_PINGS), m_DatagramDestination (nullptr), m_RefCounter (0), m_ReadyChecker(service) { @@ -918,7 +919,10 @@ namespace client auto it = params->find (I2CP_PARAM_STREAMING_INITIAL_ACK_DELAY); if (it != params->end ()) m_StreamingAckDelay = std::stoi(it->second); - + it = params->find (I2CP_PARAM_STREAMING_ANSWER_PINGS); + if (it != params->end ()) + m_IsStreamingAnswerPings = (it->second == "true"); + if (GetLeaseSetType () == i2p::data::NETDB_STORE_TYPE_ENCRYPTED_LEASESET2) { // authentication for encrypted LeaseSet diff --git a/libi2pd/Destination.h b/libi2pd/Destination.h index 1d5c0a55..a7567534 100644 --- a/libi2pd/Destination.h +++ b/libi2pd/Destination.h @@ -78,6 +78,8 @@ namespace client // streaming const char I2CP_PARAM_STREAMING_INITIAL_ACK_DELAY[] = "i2p.streaming.initialAckDelay"; const int DEFAULT_INITIAL_ACK_DELAY = 200; // milliseconds + const char I2CP_PARAM_STREAMING_ANSWER_PINGS[] = "i2p.streaming.answerPings"; + const int DEFAULT_ANSWER_PINGS = true; typedef std::function stream)> StreamRequestComplete; @@ -242,7 +244,8 @@ namespace client bool IsAcceptingStreams () const; void AcceptOnce (const i2p::stream::StreamingDestination::Acceptor& acceptor); int GetStreamingAckDelay () const { return m_StreamingAckDelay; } - + bool IsStreamingAnswerPings () const { return m_IsStreamingAnswerPings; } + // datagram i2p::datagram::DatagramDestination * GetDatagramDestination () const { return m_DatagramDestination; }; i2p::datagram::DatagramDestination * CreateDatagramDestination (bool gzip = true); @@ -275,6 +278,7 @@ namespace client std::unique_ptr m_ECIESx25519EncryptionKey; int m_StreamingAckDelay; + bool m_IsStreamingAnswerPings; std::shared_ptr m_StreamingDestination; // default std::map > m_StreamingDestinationsByPorts; i2p::datagram::DatagramDestination * m_DatagramDestination; diff --git a/libi2pd/Streaming.cpp b/libi2pd/Streaming.cpp index ff8915c0..ab08f41f 100644 --- a/libi2pd/Streaming.cpp +++ b/libi2pd/Streaming.cpp @@ -351,6 +351,28 @@ namespace stream return true; } + void Stream::HandlePing (Packet * packet) + { + uint16_t flags = packet->GetFlags (); + if (ProcessOptions (flags, packet) && m_RemoteIdentity) + { + // send pong + Packet p; + memset (p.buf, 0, 22); // minimal header all zeroes + memcpy (p.buf + 4, packet->buf, 4); // but receiveStreamID is the sendStreamID from the ping + htobe16buf (p.buf + 18, PACKET_FLAG_ECHO); // and echo flag + ssize_t payloadLen = packet->len - (packet->GetPayload () - packet->buf); + if (payloadLen > 0) + memcpy (p.buf + 22, packet->GetPayload (), payloadLen); + else + payloadLen = 0; + p.len = payloadLen + 22; + SendPackets (std::vector { &p }); + LogPrint (eLogDebug, "Streaming: Pong of ", p.len, " bytes sent"); + } + m_LocalDestination.DeletePacket (packet); + } + void Stream::ProcessAck (Packet * packet) { bool acknowledged = false; @@ -609,6 +631,7 @@ namespace stream packet[size] = 0; size++; // NACK count } + packet[size] = 0; size++; // resend delay htobuf16 (packet + size, 0); // no flags set size += 2; // flags @@ -666,6 +689,7 @@ namespace stream size += 4; // ack Through packet[size] = 0; size++; // NACK count + packet[size] = 0; size++; // resend delay htobe16buf (packet + size, PACKET_FLAG_CLOSE | PACKET_FLAG_SIGNATURE_INCLUDED); size += 2; // flags @@ -1016,6 +1040,13 @@ namespace stream auto it = m_Streams.find (sendStreamID); if (it != m_Streams.end ()) it->second->HandleNextPacket (packet); + else if (packet->IsEcho () && m_Owner->IsStreamingAnswerPings ()) + { + // ping + LogPrint (eLogInfo, "Streaming: Ping received sSID=", sendStreamID); + auto s = std::make_shared (m_Owner->GetService (), *this); + s->HandlePing (packet); + } else { LogPrint (eLogInfo, "Streaming: Unknown stream sSID=", sendStreamID); diff --git a/libi2pd/Streaming.h b/libi2pd/Streaming.h index a56b0565..e8b8db91 100644 --- a/libi2pd/Streaming.h +++ b/libi2pd/Streaming.h @@ -87,6 +87,7 @@ namespace stream bool IsSYN () const { return GetFlags () & PACKET_FLAG_SYNCHRONIZE; }; bool IsNoAck () const { return GetFlags () & PACKET_FLAG_NO_ACK; }; + bool IsEcho () const { return GetFlags () & PACKET_FLAG_ECHO; }; }; struct PacketCmp @@ -168,6 +169,7 @@ namespace stream StreamingDestination& GetLocalDestination () { return m_LocalDestination; }; void HandleNextPacket (Packet * packet); + void HandlePing (Packet * packet); size_t Send (const uint8_t * buf, size_t len); void AsyncSend (const uint8_t * buf, size_t len, SendHandler handler); diff --git a/libi2pd_client/ClientContext.cpp b/libi2pd_client/ClientContext.cpp index e1316a7f..a70add7d 100644 --- a/libi2pd_client/ClientContext.cpp +++ b/libi2pd_client/ClientContext.cpp @@ -453,6 +453,7 @@ namespace client options[I2CP_PARAM_MIN_TUNNEL_LATENCY] = GetI2CPOption(section, I2CP_PARAM_MIN_TUNNEL_LATENCY, DEFAULT_MIN_TUNNEL_LATENCY); options[I2CP_PARAM_MAX_TUNNEL_LATENCY] = GetI2CPOption(section, I2CP_PARAM_MAX_TUNNEL_LATENCY, DEFAULT_MAX_TUNNEL_LATENCY); options[I2CP_PARAM_STREAMING_INITIAL_ACK_DELAY] = GetI2CPOption(section, I2CP_PARAM_STREAMING_INITIAL_ACK_DELAY, DEFAULT_INITIAL_ACK_DELAY); + options[I2CP_PARAM_STREAMING_ANSWER_PINGS] = GetI2CPOption(section, I2CP_PARAM_STREAMING_ANSWER_PINGS, DEFAULT_ANSWER_PINGS); options[I2CP_PARAM_LEASESET_TYPE] = GetI2CPOption(section, I2CP_PARAM_LEASESET_TYPE, DEFAULT_LEASESET_TYPE); std::string encType = GetI2CPStringOption(section, I2CP_PARAM_LEASESET_ENCRYPTION_TYPE, ""); if (encType.length () > 0) options[I2CP_PARAM_LEASESET_ENCRYPTION_TYPE] = encType;