move Bob's peer tests from SSU2 session to server

This commit is contained in:
orignal 2024-09-28 22:05:25 -04:00
parent abbe1fea64
commit 15cd4feade
4 changed files with 56 additions and 40 deletions

View File

@ -152,6 +152,7 @@ namespace transport
m_SessionsByRouterHash.clear (); m_SessionsByRouterHash.clear ();
m_PendingOutgoingSessions.clear (); m_PendingOutgoingSessions.clear ();
m_Relays.clear (); m_Relays.clear ();
m_PeerTests.clear ();
m_Introducers.clear (); m_Introducers.clear ();
m_IntroducersV6.clear (); m_IntroducersV6.clear ();
m_ConnectedRecently.clear (); m_ConnectedRecently.clear ();
@ -621,6 +622,23 @@ namespace transport
return nullptr; return nullptr;
} }
bool SSU2Server::AddPeerTest (uint32_t nonce, std::shared_ptr<SSU2Session> aliceSession, uint64_t ts)
{
return m_PeerTests.emplace (nonce, std::pair{ aliceSession, ts }).second;
}
std::shared_ptr<SSU2Session> SSU2Server::GetPeerTest (uint32_t nonce)
{
auto it = m_PeerTests.find (nonce);
if (it != m_PeerTests.end ())
{
auto s = it->second.first.lock ();
m_PeerTests.erase (it);
return s;
}
return nullptr;
}
bool SSU2Server::AddRequestedPeerTest (uint32_t nonce, std::shared_ptr<SSU2PeerTestSession> session, uint64_t ts) bool SSU2Server::AddRequestedPeerTest (uint32_t nonce, std::shared_ptr<SSU2PeerTestSession> session, uint64_t ts)
{ {
return m_RequestedPeerTests.emplace (nonce, std::pair{ session, ts }).second; return m_RequestedPeerTests.emplace (nonce, std::pair{ session, ts }).second;
@ -1065,6 +1083,17 @@ namespace transport
it++; it++;
} }
for (auto it = m_PeerTests.begin (); it != m_PeerTests.end ();)
{
if (ts > it->second.second + SSU2_PEER_TEST_EXPIRATION_TIMEOUT || it->second.first.expired ())
{
LogPrint (eLogInfo, "SSU2: Peer test nonce ", it->first, " was not responded in ", SSU2_PEER_TEST_EXPIRATION_TIMEOUT, " seconds or session invalid. Deleted");
it = m_PeerTests.erase (it);
}
else
it++;
}
for (auto it = m_IncomingTokens.begin (); it != m_IncomingTokens.end (); ) for (auto it = m_IncomingTokens.begin (); it != m_IncomingTokens.end (); )
{ {
if (ts > it->second.second) if (ts > it->second.second)

View File

@ -111,6 +111,9 @@ namespace transport
void RemoveRelay (uint32_t tag); void RemoveRelay (uint32_t tag);
std::shared_ptr<SSU2Session> FindRelaySession (uint32_t tag); std::shared_ptr<SSU2Session> FindRelaySession (uint32_t tag);
bool AddPeerTest (uint32_t nonce, std::shared_ptr<SSU2Session> aliceSession, uint64_t ts);
std::shared_ptr<SSU2Session> GetPeerTest (uint32_t nonce);
bool AddRequestedPeerTest (uint32_t nonce, std::shared_ptr<SSU2PeerTestSession> session, uint64_t ts); bool AddRequestedPeerTest (uint32_t nonce, std::shared_ptr<SSU2PeerTestSession> session, uint64_t ts);
std::shared_ptr<SSU2PeerTestSession> GetRequestedPeerTest (uint32_t nonce); std::shared_ptr<SSU2PeerTestSession> GetRequestedPeerTest (uint32_t nonce);
@ -185,6 +188,7 @@ namespace transport
mutable std::mutex m_PendingOutgoingSessionsMutex; mutable std::mutex m_PendingOutgoingSessionsMutex;
std::map<boost::asio::ip::udp::endpoint, std::pair<uint64_t, uint32_t> > m_IncomingTokens, m_OutgoingTokens; // remote endpoint -> (token, expires in seconds) std::map<boost::asio::ip::udp::endpoint, std::pair<uint64_t, uint32_t> > m_IncomingTokens, m_OutgoingTokens; // remote endpoint -> (token, expires in seconds)
std::unordered_map<uint32_t, std::weak_ptr<SSU2Session> > m_Relays; // we are introducer, relay tag -> session std::unordered_map<uint32_t, std::weak_ptr<SSU2Session> > m_Relays; // we are introducer, relay tag -> session
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::MemoryPoolMt<Packets> m_PacketsArrayPool;

View File

@ -283,7 +283,6 @@ namespace transport
m_SentPackets.clear (); m_SentPackets.clear ();
m_IncompleteMessages.clear (); m_IncompleteMessages.clear ();
m_RelaySessions.clear (); m_RelaySessions.clear ();
m_PeerTests.clear ();
m_ReceivedI2NPMsgIDs.clear (); m_ReceivedI2NPMsgIDs.clear ();
m_Server.RemoveSession (m_SourceConnID); m_Server.RemoveSession (m_SourceConnID);
transports.PeerDisconnected (shared_from_this ()); transports.PeerDisconnected (shared_from_this ());
@ -2145,7 +2144,7 @@ namespace transport
GetRemoteIdentity ()->GetIdentHash ()); GetRemoteIdentity ()->GetIdentHash ());
if (session) // session with Charlie if (session) // session with Charlie
{ {
session->m_PeerTests.emplace (nonce, std::make_pair (shared_from_this (), i2p::util::GetSecondsSinceEpoch ())); m_Server.AddPeerTest (nonce, shared_from_this (), ts/1000);
auto packet = m_Server.GetSentPacketsPool ().AcquireShared (); auto packet = m_Server.GetSentPacketsPool ().AcquireShared ();
// Alice's RouterInfo // Alice's RouterInfo
auto r = i2p::data::netdb.FindRouter (GetRemoteIdentity ()->GetIdentHash ()); auto r = i2p::data::netdb.FindRouter (GetRemoteIdentity ()->GetIdentHash ());
@ -2251,34 +2250,29 @@ namespace transport
} }
case 3: // Bob from Charlie case 3: // Bob from Charlie
{ {
auto it = m_PeerTests.find (nonce); auto aliceSession = m_Server.GetPeerTest (nonce);
if (it != m_PeerTests.end ()) if (aliceSession && aliceSession->IsEstablished ())
{ {
auto aliceSession = it->second.first.lock (); uint8_t payload[SSU2_MAX_PACKET_SIZE];
if (aliceSession && aliceSession->IsEstablished ()) // Charlie's RouterInfo
{ auto r = i2p::data::netdb.FindRouter (GetRemoteIdentity ()->GetIdentHash ());
uint8_t payload[SSU2_MAX_PACKET_SIZE]; if (r && (r->IsUnreachable () || !i2p::data::netdb.PopulateRouterInfoBuffer (r))) r = nullptr;
// Charlie's RouterInfo size_t payloadSize = r ? CreateRouterInfoBlock (payload, m_MaxPayloadSize - len - 32, r) : 0;
auto r = i2p::data::netdb.FindRouter (GetRemoteIdentity ()->GetIdentHash ()); if (!payloadSize && r)
if (r && (r->IsUnreachable () || !i2p::data::netdb.PopulateRouterInfoBuffer (r))) r = nullptr; aliceSession->SendFragmentedMessage (CreateDatabaseStoreMsg (r));
size_t payloadSize = r ? CreateRouterInfoBlock (payload, m_MaxPayloadSize - len - 32, r) : 0; if (payloadSize + len + 16 > m_MaxPayloadSize)
if (!payloadSize && r) {
aliceSession->SendFragmentedMessage (CreateDatabaseStoreMsg (r)); // doesn't fit one message, send RouterInfo in separate message
if (payloadSize + len + 16 > m_MaxPayloadSize)
{
// doesn't fit one message, send RouterInfo in separate message
aliceSession->SendData (payload, payloadSize);
payloadSize = 0;
}
// PeerTest to Alice
payloadSize += CreatePeerTestBlock (payload + payloadSize, m_MaxPayloadSize, 4,
(SSU2PeerTestCode)buf[1], GetRemoteIdentity ()->GetIdentHash (), buf + offset, len - offset);
if (payloadSize < m_MaxPayloadSize)
payloadSize += CreatePaddingBlock (payload + payloadSize, m_MaxPayloadSize - payloadSize);
aliceSession->SendData (payload, payloadSize); aliceSession->SendData (payload, payloadSize);
} payloadSize = 0;
m_PeerTests.erase (it); }
} // PeerTest to Alice
payloadSize += CreatePeerTestBlock (payload + payloadSize, m_MaxPayloadSize, 4,
(SSU2PeerTestCode)buf[1], GetRemoteIdentity ()->GetIdentHash (), buf + offset, len - offset);
if (payloadSize < m_MaxPayloadSize)
payloadSize += CreatePaddingBlock (payload + payloadSize, m_MaxPayloadSize - payloadSize);
aliceSession->SendData (payload, payloadSize);
}
else else
LogPrint (eLogWarning, "SSU2: Unknown peer test 3 nonce ", nonce); LogPrint (eLogWarning, "SSU2: Unknown peer test 3 nonce ", nonce);
break; break;
@ -3030,16 +3024,6 @@ namespace transport
else else
++it; ++it;
} }
for (auto it = m_PeerTests.begin (); it != m_PeerTests.end ();)
{
if (ts > it->second.second + SSU2_PEER_TEST_EXPIRATION_TIMEOUT || it->second.first.expired ())
{
LogPrint (eLogWarning, "SSU2: Peer test nonce ", it->first, " was not responded in ", SSU2_PEER_TEST_EXPIRATION_TIMEOUT, " seconds or session invalid. Deleted");
it = m_PeerTests.erase (it);
}
else
++it;
}
if (m_PathChallenge) if (m_PathChallenge)
RequestTermination (eSSU2TerminationReasonNormalClose); RequestTermination (eSSU2TerminationReasonNormalClose);
} }

View File

@ -373,7 +373,6 @@ namespace transport
std::map<uint32_t, std::shared_ptr<SSU2SentPacket> > m_SentPackets; // packetNum -> packet std::map<uint32_t, std::shared_ptr<SSU2SentPacket> > m_SentPackets; // packetNum -> packet
std::unordered_map<uint32_t, std::shared_ptr<SSU2IncompleteMessage> > m_IncompleteMessages; // msgID -> I2NP std::unordered_map<uint32_t, std::shared_ptr<SSU2IncompleteMessage> > m_IncompleteMessages; // msgID -> I2NP
std::unordered_map<uint32_t, std::pair <std::shared_ptr<SSU2Session>, uint64_t > > m_RelaySessions; // nonce->(Alice, timestamp) for Bob or nonce->(Charlie, timestamp) for Alice std::unordered_map<uint32_t, std::pair <std::shared_ptr<SSU2Session>, uint64_t > > m_RelaySessions; // nonce->(Alice, timestamp) for Bob or nonce->(Charlie, timestamp) for Alice
std::unordered_map<uint32_t, std::pair <std::weak_ptr<SSU2Session>, uint64_t > > m_PeerTests; // nonce->(Alice, timestamp). We are Bob
std::list<std::shared_ptr<I2NPMessage> > m_SendQueue; std::list<std::shared_ptr<I2NPMessage> > m_SendQueue;
i2p::I2NPMessagesHandler m_Handler; i2p::I2NPMessagesHandler m_Handler;
bool m_IsDataReceived; bool m_IsDataReceived;