i2pd/SSUData.cpp

374 lines
10 KiB
C++
Raw Normal View History

#include <stdlib.h>
2014-07-20 14:38:39 +00:00
#include <boost/bind.hpp>
2014-04-22 15:39:26 +00:00
#include "Log.h"
2014-07-20 14:38:39 +00:00
#include "Timestamp.h"
2014-04-22 15:39:26 +00:00
#include "SSU.h"
#include "SSUData.h"
namespace i2p
{
namespace ssu
{
SSUData::SSUData (SSUSession& session):
2014-07-20 14:38:39 +00:00
m_Session (session), m_ResendTimer (session.m_Server.GetService ())
2014-04-22 15:39:26 +00:00
{
}
SSUData::~SSUData ()
{
for (auto it: m_IncomleteMessages)
if (it.second)
{
DeleteI2NPMessage (it.second->msg);
delete it.second;
}
2014-06-12 15:14:22 +00:00
for (auto it: m_SentMessages)
2014-07-20 14:38:39 +00:00
delete it.second;
2014-04-22 15:39:26 +00:00
}
2014-06-12 15:14:22 +00:00
void SSUData::ProcessSentMessageAck (uint32_t msgID)
{
auto it = m_SentMessages.find (msgID);
if (it != m_SentMessages.end ())
{
2014-07-20 14:38:39 +00:00
delete it->second;
2014-06-12 15:14:22 +00:00
m_SentMessages.erase (it);
2014-07-20 14:38:39 +00:00
if (m_SentMessages.empty ())
m_ResendTimer.cancel ();
2014-06-12 15:14:22 +00:00
}
}
2014-07-17 19:22:32 +00:00
void SSUData::ProcessAcks (uint8_t *& buf, uint8_t flag)
2014-04-22 15:39:26 +00:00
{
if (flag & DATA_FLAG_EXPLICIT_ACKS_INCLUDED)
{
// explicit ACKs
uint8_t numAcks =*buf;
buf++;
2014-06-12 15:14:22 +00:00
for (int i = 0; i < numAcks; i++)
ProcessSentMessageAck (be32toh (((uint32_t *)buf)[i]));
2014-04-22 15:39:26 +00:00
buf += numAcks*4;
}
if (flag & DATA_FLAG_ACK_BITFIELDS_INCLUDED)
{
// explicit ACK bitfields
uint8_t numBitfields =*buf;
buf++;
for (int i = 0; i < numBitfields; i++)
{
2014-07-17 19:00:33 +00:00
uint32_t msgID = be32toh (*(uint32_t *)buf);
2014-04-22 15:39:26 +00:00
buf += 4; // msgID
2014-07-19 22:53:02 +00:00
auto it = m_SentMessages.find (msgID);
2014-07-17 19:00:33 +00:00
// process individual Ack bitfields
bool isNonLast = false;
int fragment = 0;
do
{
uint8_t bitfield = *buf;
isNonLast = bitfield & 0x80;
bitfield &= 0x7F; // clear MSB
if (bitfield && it != m_SentMessages.end ())
{
2014-07-20 14:38:39 +00:00
int numSentFragments = it->second->fragments.size ();
2014-07-17 19:00:33 +00:00
// process bits
uint8_t mask = 0x40;
for (int j = 0; j < 7; j++)
{
if (bitfield & mask)
{
if (fragment < numSentFragments)
{
2014-07-20 14:38:39 +00:00
delete it->second->fragments[fragment];
it->second->fragments[fragment] = nullptr;
2014-07-17 19:00:33 +00:00
}
}
fragment++;
mask >>= 1;
}
}
2014-04-22 15:39:26 +00:00
buf++;
2014-07-17 19:00:33 +00:00
}
while (isNonLast);
2014-04-22 15:39:26 +00:00
}
2014-07-17 19:22:32 +00:00
}
}
void SSUData::ProcessFragments (uint8_t * buf)
{
2014-04-22 15:39:26 +00:00
uint8_t numFragments = *buf; // number of fragments
buf++;
for (int i = 0; i < numFragments; i++)
{
uint32_t msgID = be32toh (*(uint32_t *)buf); // message ID
buf += 4;
uint8_t frag[4];
frag[0] = 0;
memcpy (frag + 1, buf, 3);
buf += 3;
uint32_t fragmentInfo = be32toh (*(uint32_t *)frag); // fragment info
uint16_t fragmentSize = fragmentInfo & 0x1FFF; // bits 0 - 13
bool isLast = fragmentInfo & 0x010000; // bit 16
uint8_t fragmentNum = fragmentInfo >> 17; // bits 23 - 17
LogPrint ("SSU data fragment ", (int)fragmentNum, " of message ", msgID, " size=", (int)fragmentSize, isLast ? " last" : " non-last");
2014-07-15 02:06:58 +00:00
// find message with msgID
2014-04-22 15:39:26 +00:00
I2NPMessage * msg = nullptr;
2014-07-15 02:06:58 +00:00
IncompleteMessage * incompleteMessage = nullptr;
auto it = m_IncomleteMessages.find (msgID);
if (it != m_IncomleteMessages.end ())
{
// message exists
incompleteMessage = it->second;
msg = incompleteMessage->msg;
}
else
2014-04-22 15:39:26 +00:00
{
2014-07-15 02:06:58 +00:00
// create new message
2014-04-22 15:39:26 +00:00
msg = NewI2NPMessage ();
2014-07-15 02:06:58 +00:00
msg->len -= sizeof (I2NPHeaderShort);
incompleteMessage = new IncompleteMessage (msg);
m_IncomleteMessages[msgID] = incompleteMessage;
}
2014-04-22 15:39:26 +00:00
2014-07-15 02:06:58 +00:00
// handle current fragment
if (fragmentNum == incompleteMessage->nextFragmentNum)
{
// expected fragment
memcpy (msg->buf + msg->len, buf, fragmentSize);
msg->len += fragmentSize;
incompleteMessage->nextFragmentNum++;
if (!isLast && !incompleteMessage->savedFragments.empty ())
2014-04-22 15:39:26 +00:00
{
2014-07-15 02:06:58 +00:00
// try saved fragments
for (auto it1 = incompleteMessage->savedFragments.begin (); it1 != incompleteMessage->savedFragments.end ();)
2014-04-22 15:39:26 +00:00
{
2014-07-15 02:06:58 +00:00
auto savedFragment = *it1;
if (savedFragment->fragmentNum == incompleteMessage->nextFragmentNum)
2014-04-22 15:39:26 +00:00
{
2014-07-15 02:06:58 +00:00
memcpy (msg->buf + msg->len, savedFragment->buf, savedFragment->len);
msg->len += savedFragment->len;
isLast = savedFragment->isLast;
incompleteMessage->nextFragmentNum++;
incompleteMessage->savedFragments.erase (it1++);
delete savedFragment;
}
2014-04-22 15:39:26 +00:00
else
2014-07-15 02:06:58 +00:00
break;
2014-07-15 12:03:45 +00:00
}
if (isLast)
LogPrint ("Message ", msgID, " complete");
2014-07-15 02:06:58 +00:00
}
}
else
{
if (fragmentNum < incompleteMessage->nextFragmentNum)
// duplicate fragment
LogPrint ("Duplicate fragment ", (int)fragmentNum, " of message ", msgID, ". Ignored");
else
{
// missing fragment
LogPrint ("Missing fragments from ", (int)incompleteMessage->nextFragmentNum, " to ", fragmentNum - 1, " of message ", msgID);
2014-07-15 12:03:45 +00:00
auto savedFragment = new Fragment (fragmentNum, buf, fragmentSize, isLast);
if (!incompleteMessage->savedFragments.insert (savedFragment).second)
{
LogPrint ("Fragment ", (int)fragmentNum, " of message ", msgID, " already saved");
delete savedFragment;
}
2014-04-22 15:39:26 +00:00
}
2014-07-15 02:06:58 +00:00
isLast = false;
}
if (isLast)
{
// delete incomplete message
delete incompleteMessage;
m_IncomleteMessages.erase (msgID);
// process message
SendMsgAck (msgID);
msg->FromSSU (msgID);
if (m_Session.GetState () == eSessionStateEstablished)
i2p::HandleI2NPMessage (msg);
else
{
// we expect DeliveryStatus
if (msg->GetHeader ()->typeID == eI2NPDeliveryStatus)
{
LogPrint ("SSU session established");
m_Session.Established ();
}
else
LogPrint ("SSU unexpected message ", (int)msg->GetHeader ()->typeID);
DeleteI2NPMessage (msg);
}
}
else
SendFragmentAck (msgID, fragmentNum);
2014-04-22 15:39:26 +00:00
buf += fragmentSize;
}
}
2014-07-17 19:22:32 +00:00
void SSUData::ProcessMessage (uint8_t * buf, size_t len)
{
//uint8_t * start = buf;
uint8_t flag = *buf;
buf++;
LogPrint ("Process SSU data flags=", (int)flag);
// process acks if presented
if (flag & (DATA_FLAG_ACK_BITFIELDS_INCLUDED | DATA_FLAG_EXPLICIT_ACKS_INCLUDED))
ProcessAcks (buf, flag);
// extended data if presented
if (flag & DATA_FLAG_EXTENDED_DATA_INCLUDED)
{
uint8_t extendedDataSize = *buf;
buf++; // size
LogPrint ("SSU extended data of ", extendedDataSize, " bytes presented");
buf += extendedDataSize;
}
// process data
ProcessFragments (buf);
}
2014-06-10 14:39:29 +00:00
void SSUData::Send (i2p::I2NPMessage * msg)
{
2014-06-12 15:14:22 +00:00
uint32_t msgID = msg->ToSSU ();
2014-07-06 21:48:16 +00:00
if (m_SentMessages.count (msgID) > 0)
{
LogPrint ("SSU message ", msgID, " already sent");
DeleteI2NPMessage (msg);
return;
2014-07-20 14:38:39 +00:00
}
if (m_SentMessages.empty ()) // schedule resend at first message only
ScheduleResend ();
SentMessage * sentMessage = new SentMessage;
m_SentMessages[msgID] = sentMessage;
sentMessage->nextResendTime = i2p::util::GetSecondsSinceEpoch () + RESEND_INTERVAL;
sentMessage->numResends = 0;
auto& fragments = sentMessage->fragments;
2014-06-12 15:14:22 +00:00
msgID = htobe32 (msgID);
2014-06-10 14:39:29 +00:00
size_t payloadSize = SSU_MTU - sizeof (SSUHeader) - 9; // 9 = flag + #frg(1) + messageID(4) + frag info (3)
size_t len = msg->GetLength ();
uint8_t * msgBuf = msg->GetSSUHeader ();
uint32_t fragmentNum = 0;
while (len > 0)
{
2014-07-20 14:38:39 +00:00
Fragment * fragment = new Fragment;
uint8_t * buf = fragment->buf;
fragments.push_back (fragment);
2014-06-12 15:14:22 +00:00
uint8_t * payload = buf + sizeof (SSUHeader);
2014-06-10 14:39:29 +00:00
*payload = DATA_FLAG_WANT_REPLY; // for compatibility
payload++;
*payload = 1; // always 1 message fragment per message
payload++;
*(uint32_t *)payload = msgID;
payload += 4;
bool isLast = (len <= payloadSize);
size_t size = isLast ? len : payloadSize;
uint32_t fragmentInfo = (fragmentNum << 17);
if (isLast)
fragmentInfo |= 0x010000;
fragmentInfo |= size;
fragmentInfo = htobe32 (fragmentInfo);
memcpy (payload, (uint8_t *)(&fragmentInfo) + 1, 3);
payload += 3;
memcpy (payload, msgBuf, size);
size += payload - buf;
if (size & 0x0F) // make sure 16 bytes boundary
size = ((size >> 4) + 1) << 4; // (/16 + 1)*16
2014-07-20 14:38:39 +00:00
fragment->len = size;
2014-06-10 14:39:29 +00:00
// encrypt message with session key
m_Session.FillHeaderAndEncrypt (PAYLOAD_TYPE_DATA, buf, size);
m_Session.Send (buf, size);
if (!isLast)
{
len -= payloadSize;
msgBuf += payloadSize;
}
else
len = 0;
fragmentNum++;
}
DeleteI2NPMessage (msg);
}
2014-06-10 15:19:31 +00:00
void SSUData::SendMsgAck (uint32_t msgID)
{
uint8_t buf[48 + 18]; // actual length is 44 = 37 + 7 but pad it to multiple of 16
uint8_t * payload = buf + sizeof (SSUHeader);
*payload = DATA_FLAG_EXPLICIT_ACKS_INCLUDED; // flag
payload++;
*payload = 1; // number of ACKs
payload++;
*(uint32_t *)(payload) = htobe32 (msgID); // msgID
payload += 4;
*payload = 0; // number of fragments
// encrypt message with session key
m_Session.FillHeaderAndEncrypt (PAYLOAD_TYPE_DATA, buf, 48);
m_Session.Send (buf, 48);
}
void SSUData::SendFragmentAck (uint32_t msgID, int fragmentNum)
{
if (fragmentNum > 64)
{
LogPrint ("Fragment number ", fragmentNum, " exceeds 64");
return;
}
uint8_t buf[64 + 18];
uint8_t * payload = buf + sizeof (SSUHeader);
*payload = DATA_FLAG_ACK_BITFIELDS_INCLUDED; // flag
payload++;
*payload = 1; // number of ACK bitfields
payload++;
// one ack
*(uint32_t *)(payload) = htobe32 (msgID); // msgID
payload += 4;
div_t d = div (fragmentNum, 7);
memset (payload, 0x80, d.quot); // 0x80 means non-last
payload += d.quot;
*payload = 0x40 >> d.rem; // set corresponding bit
payload++;
*payload = 0; // number of fragments
size_t len = d.quot < 4 ? 48 : 64; // 48 = 37 + 7 + 4 (3+1)
// encrypt message with session key
m_Session.FillHeaderAndEncrypt (PAYLOAD_TYPE_DATA, buf, len);
m_Session.Send (buf, len);
}
2014-07-20 14:38:39 +00:00
void SSUData::ScheduleResend()
{
m_ResendTimer.cancel ();
m_ResendTimer.expires_from_now (boost::posix_time::seconds(RESEND_INTERVAL));
m_ResendTimer.async_wait (boost::bind (&SSUData::HandleResendTimer,
this, boost::asio::placeholders::error));
}
void SSUData::HandleResendTimer (const boost::system::error_code& ecode)
{
if (ecode != boost::asio::error::operation_aborted)
{
uint32_t ts = i2p::util::GetSecondsSinceEpoch ();
2014-07-22 02:13:57 +00:00
for (auto it : m_SentMessages)
2014-07-20 14:38:39 +00:00
{
2014-07-22 02:13:57 +00:00
if (ts >= it.second->nextResendTime && it.second->numResends < MAX_NUM_RESENDS)
2014-07-20 14:38:39 +00:00
{
2014-07-22 02:13:57 +00:00
for (auto f: it.second->fragments)
if (f) m_Session.Send (f->buf, f->len); // resend
2014-07-20 14:38:39 +00:00
2014-07-22 02:13:57 +00:00
it.second->numResends++;
it.second->nextResendTime += it.second->numResends*RESEND_INTERVAL;
2014-07-20 14:38:39 +00:00
}
}
2014-07-22 02:13:57 +00:00
ScheduleResend ();
2014-07-20 14:38:39 +00:00
}
}
2014-04-22 15:39:26 +00:00
}
}