Skip to content

Commit

Permalink
don't connect peer test session. Use weak_ptr for seesions by hash
Browse files Browse the repository at this point in the history
  • Loading branch information
@orignal
orignal committed Sep 25, 2024
1 parent edb2ba7 commit 0912de5
Show file tree
Hide file tree
Showing 4 changed files with 67 additions and 42 deletions.
84 changes: 48 additions & 36 deletions libi2pd/SSU2.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -450,7 +450,8 @@ namespace transport
if (session)
{
m_Sessions.emplace (session->GetConnID (), session);
AddSessionByRouterHash (session);
if (session->GetState () != eSSU2SessionStatePeerTest)
AddSessionByRouterHash (session);
}
}

Expand All @@ -459,13 +460,16 @@ namespace transport
auto it = m_Sessions.find (connID);
if (it != m_Sessions.end ())
{
auto ident = it->second->GetRemoteIdentity ();
if (ident)
{
auto it1 = m_SessionsByRouterHash.find (ident->GetIdentHash ());
if (it1 != m_SessionsByRouterHash.end () && it->second == it1->second)
m_SessionsByRouterHash.erase (it1);
}
if (it->second->GetState () != eSSU2SessionStatePeerTest)
{
auto ident = it->second->GetRemoteIdentity ();
if (ident)
{
auto it1 = m_SessionsByRouterHash.find (ident->GetIdentHash ());
if (it1 != m_SessionsByRouterHash.end () && it->second == it1->second.lock ())
m_SessionsByRouterHash.erase (it1);
}
}
if (m_LastSession == it->second)
m_LastSession = nullptr;
m_Sessions.erase (it);
Expand All @@ -480,16 +484,20 @@ namespace transport
if (ident)
{
auto ret = m_SessionsByRouterHash.emplace (ident->GetIdentHash (), session);
if (!ret.second && ret.first->second != session)
if (!ret.second)
{
// session already exists
LogPrint (eLogWarning, "SSU2: Session to ", ident->GetIdentHash ().ToBase64 (), " already exists");
// move unsent msgs to new session
ret.first->second->MoveSendQueue (session);
// terminate existing
GetService ().post (std::bind (&SSU2Session::RequestTermination, ret.first->second, eSSU2TerminationReasonReplacedByNewSession));
// update session
ret.first->second = session;
auto oldSession = ret.first->second.lock ();
if (oldSession != session)
{
// session already exists
LogPrint (eLogWarning, "SSU2: Session to ", ident->GetIdentHash ().ToBase64 (), " already exists");
// move unsent msgs to new session
oldSession->MoveSendQueue (session);
// terminate existing
GetService ().post (std::bind (&SSU2Session::RequestTermination, oldSession, eSSU2TerminationReasonReplacedByNewSession));
// update session
ret.first->second = session;
}
}
}
}
Expand All @@ -506,7 +514,7 @@ namespace transport
{
auto it = m_SessionsByRouterHash.find (ident);
if (it != m_SessionsByRouterHash.end ())
return it->second;
return it->second.lock ();
return nullptr;
}

Expand Down Expand Up @@ -761,11 +769,9 @@ namespace transport
if (it != m_SessionsByRouterHash.end ())
{
// session with router found, trying to send peer test if requested
if (peerTest && it->second->IsEstablished ())
{
auto session = it->second;
auto session = it->second.lock ();
if (peerTest && session && session->IsEstablished ())
GetService ().post ([session]() { session->SendPeerTest (); });
}
return false;
}
// check is no pending session
Expand Down Expand Up @@ -825,11 +831,15 @@ namespace transport
auto it1 = m_SessionsByRouterHash.find (it.iH);
if (it1 != m_SessionsByRouterHash.end ())
{
auto addr = it1->second->GetAddress ();
if (addr && addr->IsIntroducer ())
auto s = it1->second.lock ();
if (s)
{
it1->second->Introduce (session, it.iTag);
return;
auto addr = s->GetAddress ();
if (addr && addr->IsIntroducer ())
{
s->Introduce (session, it.iTag);
return;
}
}
}
else
Expand Down Expand Up @@ -936,17 +946,19 @@ namespace transport
if (!router) return false;
auto addr = v4 ? router->GetSSU2V4Address () : router->GetSSU2V6Address ();
if (!addr) return false;
std::shared_ptr<SSU2Session> session;
auto it = m_SessionsByRouterHash.find (router->GetIdentHash ());
if (it != m_SessionsByRouterHash.end ())
session = it->second.lock ();
if (session)
{
auto remoteAddr = it->second->GetAddress ();
auto remoteAddr = session->GetAddress ();
if (!remoteAddr || !remoteAddr->IsPeerTesting () ||
(v4 && !remoteAddr->IsV4 ()) || (!v4 && !remoteAddr->IsV6 ())) return false;
auto s = it->second;
if (s->IsEstablished ())
GetService ().post ([s]() { s->SendPeerTest (); });
(v4 && !remoteAddr->IsV4 ()) || (!v4 && !remoteAddr->IsV6 ())) return false;
if (session->IsEstablished ())
GetService ().post ([session]() { session->SendPeerTest (); });
else
s->SetOnEstablished ([s]() { s->SendPeerTest (); });
session->SetOnEstablished ([session]() { session->SendPeerTest (); });
return true;
}
else
Expand Down Expand Up @@ -997,7 +1009,7 @@ namespace transport

for (auto it = m_SessionsByRouterHash.begin (); it != m_SessionsByRouterHash.begin ();)
{
if (it->second && it->second->GetState () == eSSU2SessionStateTerminated)
if (it->second.expired ())
it = m_SessionsByRouterHash.erase (it);
else
it++;
Expand Down Expand Up @@ -1183,7 +1195,7 @@ namespace transport
auto it1 = m_SessionsByRouterHash.find (it);
if (it1 != m_SessionsByRouterHash.end ())
{
session = it1->second;
session = it1->second.lock ();
excluded.insert (it);
}
if (session && session->IsEstablished () && session->GetRelayTag () && session->IsOutgoing () && // still session with introducer?
Expand Down Expand Up @@ -1217,8 +1229,8 @@ namespace transport
auto it1 = m_SessionsByRouterHash.find (it);
if (it1 != m_SessionsByRouterHash.end ())
{
auto session = it1->second;
if (session->IsEstablished () && session->GetRelayTag () && session->IsOutgoing ())
auto session = it1->second.lock ();
if (session && session->IsEstablished () && session->GetRelayTag () && session->IsOutgoing ())
{
session->SetCreationTime (session->GetCreationTime () + SSU2_TO_INTRODUCER_SESSION_DURATION);
if (std::find (newList.begin (), newList.end (), it) == newList.end ())
Expand Down
2 changes: 1 addition & 1 deletion libi2pd/SSU2.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -162,7 +162,7 @@ namespace transport
boost::asio::ip::udp::socket m_SocketV4, m_SocketV6;
boost::asio::ip::address m_AddressV4, m_AddressV6;
std::unordered_map<uint64_t, std::shared_ptr<SSU2Session> > m_Sessions;
std::unordered_map<i2p::data::IdentHash, std::shared_ptr<SSU2Session> > m_SessionsByRouterHash;
std::unordered_map<i2p::data::IdentHash, std::weak_ptr<SSU2Session> > m_SessionsByRouterHash;
std::map<boost::asio::ip::udp::endpoint, std::shared_ptr<SSU2Session> > m_PendingOutgoingSessions;
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)
Expand Down
15 changes: 13 additions & 2 deletions libi2pd/SSU2Session.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3140,15 +3140,15 @@ namespace transport
SendPeerTest (7, buf + offset, len - offset);
else
LogPrint (eLogWarning, "SSU2: Unknown address for peer test 6");
GetServer ().RemoveSession (~htobe64 (((uint64_t)nonce << 32) | nonce));
GetServer ().RemoveSession (GetConnID ());
break;
}
case 7: // Alice from Charlie 2
{
auto addr = GetAddress ();
if (addr && addr->IsV6 ())
i2p::context.SetStatusV6 (eRouterStatusOK); // set status OK for ipv6 even if from SSU2
GetServer ().RemoveSession (htobe64 (((uint64_t)nonce << 32) | nonce));
GetServer ().RemoveSession (GetConnID ());
break;
}
default:
Expand Down Expand Up @@ -3203,5 +3203,16 @@ namespace transport
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;
}
}
}
8 changes: 5 additions & 3 deletions libi2pd/SSU2Session.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -248,7 +248,7 @@ namespace transport
void SetOnEstablished (OnEstablished e) { m_OnEstablished = e; };
OnEstablished GetOnEstablished () const { return m_OnEstablished; };

void Connect ();
virtual void Connect ();
bool Introduce (std::shared_ptr<SSU2Session> session, uint32_t relayTag);
void WaitForIntroduction ();
void SendPeerTest (); // Alice, Data message
Expand All @@ -268,7 +268,7 @@ namespace transport
SSU2SessionState GetState () const { return m_State; };
void SetState (SSU2SessionState state) { m_State = state; };

bool ProcessFirstIncomingMessage (uint64_t connID, uint8_t * buf, size_t len);
virtual bool ProcessFirstIncomingMessage (uint64_t connID, uint8_t * buf, size_t len);
bool ProcessSessionCreated (uint8_t * buf, size_t len);
bool ProcessSessionConfirmed (uint8_t * buf, size_t len);
bool ProcessRetry (uint8_t * buf, size_t len);
Expand Down Expand Up @@ -404,7 +404,9 @@ namespace transport
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
Expand Down

0 comments on commit 0912de5

Please sign in to comment.