i2pd/libi2pd/Datagram.cpp

423 lines
13 KiB
C++
Raw Normal View History

/*
* Copyright (c) 2013-2020, 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
*/
2014-10-24 00:56:50 +04:00
#include <string.h>
2016-05-11 22:12:38 +03:00
#include "Crypto.h"
2014-10-22 23:30:25 +04:00
#include "Log.h"
2014-10-24 00:56:50 +04:00
#include "TunnelBase.h"
#include "RouterContext.h"
#include "Destination.h"
2014-10-22 23:30:25 +04:00
#include "Datagram.h"
namespace i2p
{
namespace datagram
{
2020-05-18 19:01:13 +03:00
DatagramDestination::DatagramDestination (std::shared_ptr<i2p::client::ClientDestination> owner, bool gzip):
m_Owner (owner), m_Receiver (nullptr), m_RawReceiver (nullptr), m_Gzip (gzip)
2014-10-24 00:56:50 +04:00
{
auto identityLen = m_Owner->GetIdentity ()->GetFullLen ();
m_From.resize (identityLen);
m_Owner->GetIdentity ()->ToBuffer (m_From.data (), identityLen);
m_Signature.resize (m_Owner->GetIdentity ()->GetSignatureLen ());
2014-10-24 00:56:50 +04:00
}
2018-01-06 06:48:51 +03:00
2015-11-03 17:15:49 +03:00
DatagramDestination::~DatagramDestination ()
{
m_Sessions.clear();
2015-11-03 17:15:49 +03:00
}
2016-12-12 21:40:24 +03:00
void DatagramDestination::SendDatagramTo(const uint8_t * payload, size_t len, const i2p::data::IdentHash & identity, uint16_t fromPort, uint16_t toPort)
2018-01-06 06:48:51 +03:00
{
2020-06-10 18:57:40 +03:00
auto session = ObtainSession(identity);
SendDatagram (session, payload, len, fromPort, toPort);
FlushSendQueue (session);
}
2019-07-10 04:33:55 +03:00
void DatagramDestination::SendRawDatagramTo(const uint8_t * payload, size_t len, const i2p::data::IdentHash & identity, uint16_t fromPort, uint16_t toPort)
{
2020-06-10 18:57:40 +03:00
auto session = ObtainSession(identity);
SendRawDatagram (session, payload, len, fromPort, toPort);
FlushSendQueue (session);
}
2020-06-10 02:20:24 +03:00
std::shared_ptr<DatagramSession> DatagramDestination::GetSession(const i2p::data::IdentHash & ident)
{
return ObtainSession(ident);
}
void DatagramDestination::SendDatagram (std::shared_ptr<DatagramSession> session, const uint8_t * payload, size_t len, uint16_t fromPort, uint16_t toPort)
{
if (session)
{
if (m_Owner->GetIdentity ()->GetSigningKeyType () == i2p::data::SIGNING_KEY_TYPE_DSA_SHA1)
{
uint8_t hash[32];
SHA256(payload, len, hash);
m_Owner->Sign (hash, 32, m_Signature.data ());
}
else
m_Owner->Sign (payload, len, m_Signature.data ());
auto msg = CreateDataMessage ({{m_From.data (), m_From.size ()}, {m_Signature.data (), m_Signature.size ()}, {payload, len}},
fromPort, toPort, false, !session->IsRatchets ()); // datagram
session->SendMsg(msg);
}
}
void DatagramDestination::SendRawDatagram (std::shared_ptr<DatagramSession> session, const uint8_t * payload, size_t len, uint16_t fromPort, uint16_t toPort)
{
if (session)
session->SendMsg(CreateDataMessage ({{payload, len}}, fromPort, toPort, true, !session->IsRatchets ())); // raw
}
void DatagramDestination::FlushSendQueue (std::shared_ptr<DatagramSession> session)
2020-06-09 23:26:45 +03:00
{
2020-06-10 02:20:24 +03:00
if (session)
session->FlushSendQueue ();
2020-06-09 23:26:45 +03:00
}
2016-10-09 17:55:55 +03:00
void DatagramDestination::HandleDatagram (uint16_t fromPort, uint16_t toPort,uint8_t * const &buf, size_t len)
2014-10-27 23:30:56 +03:00
{
i2p::data::IdentityEx identity;
size_t identityLen = identity.FromBuffer (buf, len);
const uint8_t * signature = buf + identityLen;
size_t headerLen = identityLen + identity.GetSignatureLen ();
bool verified = false;
if (identity.GetSigningKeyType () == i2p::data::SIGNING_KEY_TYPE_DSA_SHA1)
2015-03-27 05:17:26 +03:00
{
uint8_t hash[32];
2015-11-03 17:15:49 +03:00
SHA256(buf + headerLen, len - headerLen, hash);
2015-03-27 05:17:26 +03:00
verified = identity.Verify (hash, 32, signature);
2018-01-06 06:48:51 +03:00
}
else
2014-10-27 23:30:56 +03:00
verified = identity.Verify (buf + headerLen, len - headerLen, signature);
2018-01-06 06:48:51 +03:00
2014-10-27 23:30:56 +03:00
if (verified)
{
2016-12-12 21:40:24 +03:00
auto h = identity.GetIdentHash();
2017-01-13 19:45:52 +03:00
auto session = ObtainSession(h);
session->Ack();
auto r = FindReceiver(toPort);
if(r)
r(identity, fromPort, toPort, buf + headerLen, len -headerLen);
2014-10-31 23:44:44 +03:00
else
LogPrint (eLogWarning, "DatagramDestination: no receiver for port ", toPort);
2014-10-27 23:30:56 +03:00
}
else
2018-01-06 06:48:51 +03:00
LogPrint (eLogWarning, "Datagram signature verification failed");
2014-10-27 23:30:56 +03:00
}
2019-07-10 04:33:55 +03:00
void DatagramDestination::HandleRawDatagram (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len)
{
if (m_RawReceiver)
m_RawReceiver (fromPort, toPort, buf, len);
else
LogPrint (eLogWarning, "DatagramDestination: no receiver for raw datagram");
}
DatagramDestination::Receiver DatagramDestination::FindReceiver(uint16_t port)
{
std::lock_guard<std::mutex> lock(m_ReceiversMutex);
Receiver r = m_Receiver;
auto itr = m_ReceiversByPorts.find(port);
if (itr != m_ReceiversByPorts.end())
r = itr->second;
return r;
}
2019-07-10 04:33:55 +03:00
void DatagramDestination::HandleDataMessagePayload (uint16_t fromPort, uint16_t toPort, const uint8_t * buf, size_t len, bool isRaw)
2014-10-22 23:30:25 +04:00
{
// unzip it
uint8_t uncompressed[MAX_DATAGRAM_SIZE];
2015-11-03 17:15:49 +03:00
size_t uncompressedLen = m_Inflator.Inflate (buf, len, uncompressed, MAX_DATAGRAM_SIZE);
if (uncompressedLen)
2019-07-10 04:33:55 +03:00
{
if (isRaw)
HandleRawDatagram (fromPort, toPort, uncompressed, uncompressedLen);
else
2019-07-10 04:33:55 +03:00
HandleDatagram (fromPort, toPort, uncompressed, uncompressedLen);
}
2017-01-17 20:13:56 +03:00
else
LogPrint (eLogWarning, "Datagram: decompression failed");
2014-10-22 23:30:25 +04:00
}
2020-05-17 23:49:31 +03:00
std::shared_ptr<I2NPMessage> DatagramDestination::CreateDataMessage (
const std::vector<std::pair<const uint8_t *, size_t> >& payloads,
uint16_t fromPort, uint16_t toPort, bool isRaw, bool checksum)
2014-10-24 00:56:50 +04:00
{
auto msg = m_I2NPMsgsPool.AcquireShared ();
2014-10-24 00:56:50 +04:00
uint8_t * buf = msg->GetPayload ();
2015-11-03 17:15:49 +03:00
buf += 4; // reserve for length
2020-05-20 04:02:32 +03:00
size_t size = m_Gzip ? m_Deflator.Deflate (payloads, buf, msg->maxLen - msg->len) :
i2p::data::GzipNoCompression (payloads, buf, msg->maxLen - msg->len);
2015-11-03 17:15:49 +03:00
if (size)
{
htobe32buf (msg->GetPayload (), size); // length
htobe16buf (buf + 4, fromPort); // source port
2018-01-06 06:48:51 +03:00
htobe16buf (buf + 6, toPort); // destination port
buf[9] = isRaw ? i2p::client::PROTOCOL_TYPE_RAW : i2p::client::PROTOCOL_TYPE_DATAGRAM; // raw or datagram protocol
2018-01-06 06:48:51 +03:00
msg->len += size + 4;
2020-05-17 23:49:31 +03:00
msg->FillI2NPMessageHeader (eI2NPData, 0, checksum);
2018-01-06 06:48:51 +03:00
}
2015-11-03 17:15:49 +03:00
else
msg = nullptr;
2014-10-24 00:56:50 +04:00
return msg;
}
void DatagramDestination::CleanUp ()
2018-01-06 06:48:51 +03:00
{
2016-09-08 17:56:22 +03:00
if (m_Sessions.empty ()) return;
auto now = i2p::util::GetMillisecondsSinceEpoch();
LogPrint(eLogDebug, "DatagramDestination: clean up sessions");
2016-10-09 17:55:55 +03:00
std::unique_lock<std::mutex> lock(m_SessionsMutex);
// for each session ...
2018-01-06 06:48:51 +03:00
for (auto it = m_Sessions.begin (); it != m_Sessions.end (); )
{
// check if expired
if (now - it->second->LastActivity() >= DATAGRAM_SESSION_MAX_IDLE)
{
LogPrint(eLogInfo, "DatagramDestination: expiring idle session with ", it->first.ToBase32());
2017-01-17 20:13:56 +03:00
it->second->Stop ();
it = m_Sessions.erase (it); // we are expired
}
else
it++;
}
}
2018-01-06 06:48:51 +03:00
2016-12-12 21:40:24 +03:00
std::shared_ptr<DatagramSession> DatagramDestination::ObtainSession(const i2p::data::IdentHash & identity)
{
std::shared_ptr<DatagramSession> session = nullptr;
std::lock_guard<std::mutex> lock(m_SessionsMutex);
2016-12-12 21:40:24 +03:00
auto itr = m_Sessions.find(identity);
if (itr == m_Sessions.end()) {
// not found, create new session
2016-12-12 21:40:24 +03:00
session = std::make_shared<DatagramSession>(m_Owner, identity);
2017-01-17 20:13:56 +03:00
session->Start ();
2016-12-12 21:40:24 +03:00
m_Sessions[identity] = session;
} else {
session = itr->second;
}
return session;
}
2016-09-03 20:58:34 +03:00
std::shared_ptr<DatagramSession::Info> DatagramDestination::GetInfoForRemote(const i2p::data::IdentHash & remote)
{
std::lock_guard<std::mutex> lock(m_SessionsMutex);
for ( auto & item : m_Sessions)
{
if(item.first == remote) return std::make_shared<DatagramSession::Info>(item.second->GetSessionInfo());
}
return nullptr;
}
2018-01-06 06:48:51 +03:00
2019-07-10 04:33:55 +03:00
DatagramSession::DatagramSession(std::shared_ptr<i2p::client::ClientDestination> localDestination,
const i2p::data::IdentHash & remoteIdent) :
m_LocalDestination(localDestination),
2016-12-12 21:40:24 +03:00
m_RemoteIdent(remoteIdent),
2016-12-27 02:47:47 +03:00
m_RequestingLS(false)
2017-01-17 20:13:56 +03:00
{
}
void DatagramSession::Start ()
{
2016-12-12 21:40:24 +03:00
m_LastUse = i2p::util::GetMillisecondsSinceEpoch ();
}
2017-01-17 20:13:56 +03:00
void DatagramSession::Stop ()
{
}
void DatagramSession::SendMsg(std::shared_ptr<I2NPMessage> msg)
{
// we used this session
m_LastUse = i2p::util::GetMillisecondsSinceEpoch();
if (msg || m_SendQueue.empty ())
m_SendQueue.push_back(msg);
// flush queue right away if full
if (!msg || m_SendQueue.size() >= DATAGRAM_SEND_QUEUE_MAX_SIZE)
FlushSendQueue();
}
2016-09-03 20:58:34 +03:00
DatagramSession::Info DatagramSession::GetSessionInfo() const
{
if(!m_RoutingSession)
2016-12-12 21:40:24 +03:00
return DatagramSession::Info(nullptr, nullptr, m_LastUse);
2018-01-06 06:48:51 +03:00
2016-09-03 20:58:34 +03:00
auto routingPath = m_RoutingSession->GetSharedRoutingPath();
if (!routingPath)
2016-12-12 21:40:24 +03:00
return DatagramSession::Info(nullptr, nullptr, m_LastUse);
2016-09-03 20:58:34 +03:00
auto lease = routingPath->remoteLease;
auto tunnel = routingPath->outboundTunnel;
if(lease)
{
if(tunnel)
2016-12-12 21:40:24 +03:00
return DatagramSession::Info(lease->tunnelGateway, tunnel->GetEndpointIdentHash(), m_LastUse);
2016-09-03 20:58:34 +03:00
else
2016-12-12 21:40:24 +03:00
return DatagramSession::Info(lease->tunnelGateway, nullptr, m_LastUse);
2016-09-03 20:58:34 +03:00
}
else if(tunnel)
2016-12-12 21:40:24 +03:00
return DatagramSession::Info(nullptr, tunnel->GetEndpointIdentHash(), m_LastUse);
2016-09-03 20:58:34 +03:00
else
2016-12-12 21:40:24 +03:00
return DatagramSession::Info(nullptr, nullptr, m_LastUse);
}
2016-12-12 21:40:24 +03:00
void DatagramSession::Ack()
{
2016-12-12 21:40:24 +03:00
m_LastUse = i2p::util::GetMillisecondsSinceEpoch();
auto path = GetSharedRoutingPath();
if(path)
path->updateTime = i2p::util::GetSecondsSinceEpoch ();
2020-05-17 23:49:31 +03:00
if (IsRatchets ())
SendMsg (nullptr); // send empty message in case if we have some data to send
}
2016-12-12 21:40:24 +03:00
std::shared_ptr<i2p::garlic::GarlicRoutingPath> DatagramSession::GetSharedRoutingPath ()
{
2020-06-13 23:18:12 +03:00
if (!m_RemoteLeaseSet || m_RemoteLeaseSet->IsExpired ())
{
2020-06-13 23:18:12 +03:00
m_RemoteLeaseSet = m_LocalDestination->FindLeaseSet(m_RemoteIdent);
if (!m_RemoteLeaseSet)
{
if(!m_RequestingLS)
{
2016-12-27 02:47:47 +03:00
m_RequestingLS = true;
m_LocalDestination->RequestDestination(m_RemoteIdent, std::bind(&DatagramSession::HandleLeaseSetUpdated, this, std::placeholders::_1));
}
2016-12-12 21:40:24 +03:00
return nullptr;
2020-06-13 23:18:12 +03:00
}
}
if (!m_RoutingSession || m_RoutingSession->IsTerminated () || !m_RoutingSession->IsReadyToSend ())
{
bool found = false;
for (auto& it: m_PendingRoutingSessions)
if (it->GetOwner () && m_RoutingSession->IsReadyToSend ()) // found established session
{
m_RoutingSession = it;
m_PendingRoutingSessions.clear ();
found = true;
break;
}
if (!found)
{
m_RoutingSession = m_LocalDestination->GetRoutingSession(m_RemoteLeaseSet, true);
if (!m_RoutingSession->GetOwner () || !m_RoutingSession->IsReadyToSend ())
m_PendingRoutingSessions.push_back (m_RoutingSession);
}
}
2020-06-13 23:18:12 +03:00
2016-12-12 21:40:24 +03:00
auto path = m_RoutingSession->GetSharedRoutingPath();
if (path && m_RoutingSession->IsRatchets () &&
m_LastUse > m_RoutingSession->GetLastActivityTimestamp ()*1000 + DATAGRAM_SESSION_PATH_TIMEOUT)
{
m_RoutingSession->SetSharedRoutingPath (nullptr);
path = nullptr;
}
2020-06-13 23:18:12 +03:00
if (path)
{
if (path->outboundTunnel && !path->outboundTunnel->IsEstablished ())
2020-06-17 21:24:25 +03:00
{
2016-12-12 21:40:24 +03:00
// bad outbound tunnel, switch outbound tunnel
2020-06-13 23:18:12 +03:00
path->outboundTunnel = m_LocalDestination->GetTunnelPool()->GetNextOutboundTunnel(path->outboundTunnel);
2020-06-17 21:24:25 +03:00
if (!path->outboundTunnel)
m_RoutingSession->SetSharedRoutingPath (nullptr);
}
2020-06-13 23:18:12 +03:00
if (path->remoteLease && path->remoteLease->ExpiresWithin(DATAGRAM_SESSION_LEASE_HANDOVER_WINDOW))
{
2016-12-12 21:40:24 +03:00
// bad lease, switch to next one
2020-06-13 23:18:12 +03:00
if (m_RemoteLeaseSet)
{
auto ls = m_RemoteLeaseSet->GetNonExpiredLeasesExcluding(
[&](const i2p::data::Lease& l) -> bool
{
return l.tunnelID == path->remoteLease->tunnelID;
});
2016-12-12 21:40:24 +03:00
auto sz = ls.size();
2020-06-13 23:18:12 +03:00
if (sz)
{
2016-12-12 21:40:24 +03:00
auto idx = rand() % sz;
2020-06-13 23:18:12 +03:00
path->remoteLease = ls[idx];
2016-12-12 21:40:24 +03:00
}
2020-06-13 23:18:12 +03:00
else
2020-06-17 21:24:25 +03:00
m_RoutingSession->SetSharedRoutingPath (nullptr);
2020-06-13 23:18:12 +03:00
}
else
2020-06-17 21:24:25 +03:00
{
2016-12-12 21:40:24 +03:00
// no remote lease set?
LogPrint(eLogWarning, "DatagramSession: no cached remote lease set for ", m_RemoteIdent.ToBase32());
2020-06-17 21:24:25 +03:00
m_RoutingSession->SetSharedRoutingPath (nullptr);
}
}
2020-06-13 23:18:12 +03:00
}
else
{
2016-12-12 21:40:24 +03:00
// no current path, make one
path = std::make_shared<i2p::garlic::GarlicRoutingPath>();
2020-06-13 23:18:12 +03:00
path->outboundTunnel = m_LocalDestination->GetTunnelPool()->GetNextOutboundTunnel();
2020-06-17 21:24:25 +03:00
if (!path->outboundTunnel) return nullptr;
2020-06-13 23:18:12 +03:00
if (m_RemoteLeaseSet)
{
// pick random next good lease
auto ls = m_RemoteLeaseSet->GetNonExpiredLeases();
auto sz = ls.size();
if (sz)
2017-04-09 15:52:42 +03:00
{
2020-06-13 23:18:12 +03:00
auto idx = rand() % sz;
path->remoteLease = ls[idx];
2017-04-09 15:52:42 +03:00
}
2020-06-17 21:24:25 +03:00
else
return nullptr;
2020-06-13 23:18:12 +03:00
}
else
{
// no remote lease set currently, bail
LogPrint(eLogWarning, "DatagramSession: no remote lease set found for ", m_RemoteIdent.ToBase32());
return nullptr;
}
2016-12-12 21:40:24 +03:00
m_RoutingSession->SetSharedRoutingPath(path);
}
2016-12-12 21:40:24 +03:00
return path;
}
2016-12-12 21:40:24 +03:00
void DatagramSession::HandleLeaseSetUpdated(std::shared_ptr<i2p::data::LeaseSet> ls)
{
2016-12-27 02:47:47 +03:00
m_RequestingLS = false;
if(!ls) return;
2016-12-12 21:40:24 +03:00
// only update lease set if found and newer than previous lease set
uint64_t oldExpire = 0;
if(m_RemoteLeaseSet) oldExpire = m_RemoteLeaseSet->GetExpirationTime();
if(ls && ls->GetExpirationTime() > oldExpire) m_RemoteLeaseSet = ls;
}
2016-12-12 21:40:24 +03:00
void DatagramSession::FlushSendQueue ()
{
2020-06-09 23:26:45 +03:00
if (m_SendQueue.empty ()) return;
2016-12-12 21:40:24 +03:00
std::vector<i2p::tunnel::TunnelMessageBlock> send;
auto routingPath = GetSharedRoutingPath();
// if we don't have a routing path we will drop all queued messages
if(routingPath && routingPath->outboundTunnel && routingPath->remoteLease)
{
2016-12-12 21:40:24 +03:00
for (const auto & msg : m_SendQueue)
{
2016-12-12 21:40:24 +03:00
auto m = m_RoutingSession->WrapSingleMessage(msg);
if (m)
send.push_back(i2p::tunnel::TunnelMessageBlock{i2p::tunnel::eDeliveryTypeTunnel,routingPath->remoteLease->tunnelGateway, routingPath->remoteLease->tunnelID, m});
}
2016-12-12 21:40:24 +03:00
routingPath->outboundTunnel->SendTunnelDataMsg(send);
}
2016-12-12 21:40:24 +03:00
m_SendQueue.clear();
}
2014-10-22 23:30:25 +04:00
}
}