|
|
|
|
@ -1,5 +1,6 @@
@@ -1,5 +1,6 @@
|
|
|
|
|
#include "I2PEndian.h" |
|
|
|
|
#include <string> |
|
|
|
|
#include <algorithm> |
|
|
|
|
#include <cryptopp/gzip.h> |
|
|
|
|
#include "Log.h" |
|
|
|
|
#include "RouterInfo.h" |
|
|
|
|
@ -15,16 +16,26 @@ namespace i2p
@@ -15,16 +16,26 @@ namespace i2p
|
|
|
|
|
namespace stream |
|
|
|
|
{ |
|
|
|
|
Stream::Stream (StreamingDestination * local, const i2p::data::LeaseSet * remote): |
|
|
|
|
m_SendStreamID (0), m_SequenceNumber (0), m_LocalDestination (local), m_RemoteLeaseSet (remote) |
|
|
|
|
m_SendStreamID (0), m_SequenceNumber (0), m_LastReceivedSequenceNumber (0), m_IsOpen (false), |
|
|
|
|
m_LocalDestination (local), m_RemoteLeaseSet (remote), m_OutboundTunnel (nullptr) |
|
|
|
|
{ |
|
|
|
|
m_RecvStreamID = i2p::context.GetRandomNumberGenerator ().GenerateWord32 (); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void Stream::HandleNextPacket (const uint8_t * buf, size_t len) |
|
|
|
|
Stream::~Stream () |
|
|
|
|
{ |
|
|
|
|
const uint8_t * end = buf + len; |
|
|
|
|
while (auto packet = m_ReceiveQueue.Get ()) |
|
|
|
|
delete packet; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void Stream::HandleNextPacket (Packet * packet) |
|
|
|
|
{ |
|
|
|
|
const uint8_t * buf = packet->buf; |
|
|
|
|
buf += 4; // sendStreamID
|
|
|
|
|
if (!m_SendStreamID) |
|
|
|
|
m_SendStreamID = be32toh (*(uint32_t *)buf); |
|
|
|
|
buf += 4; // receiveStreamID
|
|
|
|
|
uint32_t receivedSeqn = be32toh (*(uint32_t *)buf); |
|
|
|
|
buf += 4; // sequenceNum
|
|
|
|
|
buf += 4; // ackThrough
|
|
|
|
|
int nackCount = buf[0]; |
|
|
|
|
@ -57,12 +68,59 @@ namespace stream
@@ -57,12 +68,59 @@ namespace stream
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// we have reached payload section
|
|
|
|
|
std::string str((const char *)buf, end-buf); |
|
|
|
|
LogPrint ("Payload: ", str); |
|
|
|
|
LogPrint ("seqn=", receivedSeqn, ", flags=", flags); |
|
|
|
|
if (!receivedSeqn || receivedSeqn == m_LastReceivedSequenceNumber + 1) |
|
|
|
|
{ |
|
|
|
|
// we have received next message
|
|
|
|
|
packet->offset = buf - packet->buf; |
|
|
|
|
if (packet->GetLength () > 0) |
|
|
|
|
m_ReceiveQueue.Put (packet); |
|
|
|
|
else |
|
|
|
|
delete packet; |
|
|
|
|
|
|
|
|
|
m_LastReceivedSequenceNumber = receivedSeqn; |
|
|
|
|
SendQuickAck (); |
|
|
|
|
} |
|
|
|
|
else |
|
|
|
|
{ |
|
|
|
|
if (receivedSeqn <= m_LastReceivedSequenceNumber) |
|
|
|
|
{ |
|
|
|
|
// we have received duplicate. Most likely our outbound tunnel is dead
|
|
|
|
|
LogPrint ("Duplicate message ", receivedSeqn, " received"); |
|
|
|
|
m_OutboundTunnel = i2p::tunnel::tunnels.GetNextOutboundTunnel (); // pick another tunnel
|
|
|
|
|
if (m_OutboundTunnel) |
|
|
|
|
SendQuickAck (); // resend ack for previous message again
|
|
|
|
|
} |
|
|
|
|
else |
|
|
|
|
{ |
|
|
|
|
LogPrint ("Missing messages from ", m_LastReceivedSequenceNumber + 1, " to ", receivedSeqn - 1); |
|
|
|
|
// actually do nothing. just wait for missing message again
|
|
|
|
|
} |
|
|
|
|
delete packet; // packet dropped
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
if (flags & PACKET_FLAG_CLOSE) |
|
|
|
|
{ |
|
|
|
|
LogPrint ("Closed"); |
|
|
|
|
m_IsOpen = false; |
|
|
|
|
m_ReceiveQueue.WakeUp (); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
size_t Stream::Send (uint8_t * buf, size_t len, int timeout) |
|
|
|
|
{ |
|
|
|
|
if (!m_IsOpen) |
|
|
|
|
ConnectAndSend (buf, len); |
|
|
|
|
else |
|
|
|
|
{ |
|
|
|
|
// TODO: implement
|
|
|
|
|
} |
|
|
|
|
return len; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void Stream::ConnectAndSend (uint8_t * buf, size_t len) |
|
|
|
|
{ |
|
|
|
|
m_IsOpen = true; |
|
|
|
|
uint8_t packet[STREAMING_MTU]; |
|
|
|
|
size_t size = 0; |
|
|
|
|
*(uint32_t *)(packet + size) = htobe32 (m_SendStreamID); |
|
|
|
|
@ -78,30 +136,149 @@ namespace stream
@@ -78,30 +136,149 @@ namespace stream
|
|
|
|
|
size++; // resend delay
|
|
|
|
|
// TODO: for initial packet only, following packets have different falgs
|
|
|
|
|
*(uint16_t *)(packet + size) = htobe16 (PACKET_FLAG_SYNCHRONIZE | |
|
|
|
|
PACKET_FLAG_FROM_INCLUDED | PACKET_FLAG_SIGNATURE_INCLUDED | PACKET_FLAG_NO_ACK); |
|
|
|
|
PACKET_FLAG_FROM_INCLUDED | PACKET_FLAG_SIGNATURE_INCLUDED | |
|
|
|
|
PACKET_FLAG_MAX_PACKET_SIZE_INCLUDED | PACKET_FLAG_NO_ACK); |
|
|
|
|
size += 2; // flags
|
|
|
|
|
*(uint16_t *)(packet + size) = htobe16 (sizeof (i2p::data::Identity) + 40); // identity + signature
|
|
|
|
|
*(uint16_t *)(packet + size) = htobe16 (sizeof (i2p::data::Identity) + 40 + 2); // identity + signature + packet size
|
|
|
|
|
size += 2; // options size
|
|
|
|
|
memcpy (packet + size, &m_LocalDestination->GetIdentity (), sizeof (i2p::data::Identity)); |
|
|
|
|
size += sizeof (i2p::data::Identity); // from
|
|
|
|
|
*(uint16_t *)(packet + size) = htobe16 (STREAMING_MTU); |
|
|
|
|
size += 2; // max packet size
|
|
|
|
|
uint8_t * signature = packet + size; // set it later
|
|
|
|
|
memset (signature, 0, 40); // zeroes for now
|
|
|
|
|
size += 40; // signature
|
|
|
|
|
|
|
|
|
|
memcpy (packet + size, buf, len); |
|
|
|
|
size += len; // payload
|
|
|
|
|
m_LocalDestination->Sign (packet, size, signature); |
|
|
|
|
I2NPMessage * msg = i2p::garlic::routing.WrapSingleMessage (m_RemoteLeaseSet, |
|
|
|
|
I2NPMessage * msg = i2p::garlic::routing.WrapMessage (m_RemoteLeaseSet, |
|
|
|
|
CreateDataMessage (this, packet, size), m_LocalDestination->GetLeaseSet ()); |
|
|
|
|
|
|
|
|
|
auto outbound = i2p::tunnel::tunnels.GetNextOutboundTunnel (); |
|
|
|
|
if (outbound) |
|
|
|
|
if (!m_OutboundTunnel) |
|
|
|
|
m_OutboundTunnel = i2p::tunnel::tunnels.GetNextOutboundTunnel (); |
|
|
|
|
if (m_OutboundTunnel) |
|
|
|
|
{ |
|
|
|
|
auto& lease = m_RemoteLeaseSet->GetLeases ()[0]; // TODO:
|
|
|
|
|
outbound->SendTunnelDataMsg (lease.tunnelGateway, lease.tunnelID, msg); |
|
|
|
|
m_OutboundTunnel->SendTunnelDataMsg (lease.tunnelGateway, lease.tunnelID, msg); |
|
|
|
|
} |
|
|
|
|
else |
|
|
|
|
DeleteI2NPMessage (msg); |
|
|
|
|
return len; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void Stream::SendQuickAck () |
|
|
|
|
{ |
|
|
|
|
uint8_t packet[STREAMING_MTU]; |
|
|
|
|
size_t size = 0; |
|
|
|
|
*(uint32_t *)(packet + size) = htobe32 (m_SendStreamID); |
|
|
|
|
size += 4; // sendStreamID
|
|
|
|
|
*(uint32_t *)(packet + size) = htobe32 (m_RecvStreamID); |
|
|
|
|
size += 4; // receiveStreamID
|
|
|
|
|
*(uint32_t *)(packet + size) = 0; // this is plain Ack message
|
|
|
|
|
size += 4; // sequenceNum
|
|
|
|
|
*(uint32_t *)(packet + size) = htobe32 (m_LastReceivedSequenceNumber); |
|
|
|
|
size += 4; // ack Through
|
|
|
|
|
packet[size] = 0; |
|
|
|
|
size++; // NACK count
|
|
|
|
|
size++; // resend delay
|
|
|
|
|
*(uint16_t *)(packet + size) = 0; // nof flags set
|
|
|
|
|
size += 2; // flags
|
|
|
|
|
*(uint16_t *)(packet + size) = 0; // no options
|
|
|
|
|
size += 2; // options size
|
|
|
|
|
|
|
|
|
|
I2NPMessage * msg = i2p::garlic::routing.WrapMessage (m_RemoteLeaseSet, |
|
|
|
|
CreateDataMessage (this, packet, size)); |
|
|
|
|
if (m_OutboundTunnel) |
|
|
|
|
{ |
|
|
|
|
auto leases = m_RemoteLeaseSet->GetNonExpiredLeases (); |
|
|
|
|
if (!leases.empty ()) |
|
|
|
|
{ |
|
|
|
|
auto& lease = leases[0]; // TODO:
|
|
|
|
|
m_OutboundTunnel->SendTunnelDataMsg (lease.tunnelGateway, lease.tunnelID, msg); |
|
|
|
|
LogPrint ("Quick Ack sent"); |
|
|
|
|
} |
|
|
|
|
else |
|
|
|
|
{ |
|
|
|
|
LogPrint ("All leases are expired"); |
|
|
|
|
DeleteI2NPMessage (msg); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
else |
|
|
|
|
DeleteI2NPMessage (msg); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void Stream::Close () |
|
|
|
|
{ |
|
|
|
|
if (m_IsOpen) |
|
|
|
|
{ |
|
|
|
|
m_IsOpen = false; |
|
|
|
|
uint8_t packet[STREAMING_MTU]; |
|
|
|
|
size_t size = 0; |
|
|
|
|
*(uint32_t *)(packet + size) = htobe32 (m_SendStreamID); |
|
|
|
|
size += 4; // sendStreamID
|
|
|
|
|
*(uint32_t *)(packet + size) = htobe32 (m_RecvStreamID); |
|
|
|
|
size += 4; // receiveStreamID
|
|
|
|
|
*(uint32_t *)(packet + size) = htobe32 (m_SequenceNumber); |
|
|
|
|
size += 4; // sequenceNum
|
|
|
|
|
*(uint32_t *)(packet + size) = htobe32 (m_LastReceivedSequenceNumber); |
|
|
|
|
size += 4; // ack Through
|
|
|
|
|
packet[size] = 0; |
|
|
|
|
size++; // NACK count
|
|
|
|
|
size++; // resend delay
|
|
|
|
|
*(uint16_t *)(packet + size) = PACKET_FLAG_CLOSE | PACKET_FLAG_SIGNATURE_INCLUDED; |
|
|
|
|
size += 2; // flags
|
|
|
|
|
*(uint16_t *)(packet + size) = htobe16 (40); // 40 bytes signature
|
|
|
|
|
size += 2; // options size
|
|
|
|
|
uint8_t * signature = packet + size; |
|
|
|
|
memset (packet + size, 0, 40); |
|
|
|
|
size += 40; // signature
|
|
|
|
|
m_LocalDestination->Sign (packet, size, signature); |
|
|
|
|
|
|
|
|
|
I2NPMessage * msg = i2p::garlic::routing.WrapSingleMessage (m_RemoteLeaseSet, |
|
|
|
|
CreateDataMessage (this, packet, size)); |
|
|
|
|
if (m_OutboundTunnel) |
|
|
|
|
{ |
|
|
|
|
auto& lease = m_RemoteLeaseSet->GetLeases ()[0]; // TODO:
|
|
|
|
|
m_OutboundTunnel->SendTunnelDataMsg (lease.tunnelGateway, lease.tunnelID, msg); |
|
|
|
|
LogPrint ("FIN sent"); |
|
|
|
|
} |
|
|
|
|
else |
|
|
|
|
DeleteI2NPMessage (msg); |
|
|
|
|
m_ReceiveQueue.WakeUp (); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
size_t Stream::Receive (uint8_t * buf, size_t len, int timeout) |
|
|
|
|
{ |
|
|
|
|
if (!m_IsOpen) return 0; |
|
|
|
|
if (m_ReceiveQueue.IsEmpty ()) |
|
|
|
|
{ |
|
|
|
|
if (!timeout) return 0; |
|
|
|
|
if (!m_ReceiveQueue.Wait (timeout, 0)) |
|
|
|
|
return 0; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// either non-empty or we have received empty
|
|
|
|
|
size_t pos = 0; |
|
|
|
|
while (pos < len) |
|
|
|
|
{ |
|
|
|
|
Packet * packet = m_ReceiveQueue.Peek (); |
|
|
|
|
if (packet) |
|
|
|
|
{ |
|
|
|
|
size_t l = std::min (packet->GetLength (), len - pos); |
|
|
|
|
memcpy (buf + pos, packet->GetBuffer (), l); |
|
|
|
|
pos += l; |
|
|
|
|
packet->offset += l; |
|
|
|
|
if (!packet->GetLength ()) |
|
|
|
|
{ |
|
|
|
|
m_ReceiveQueue.Get (); |
|
|
|
|
delete packet; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
else // no more data available
|
|
|
|
|
break; |
|
|
|
|
} |
|
|
|
|
return pos; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
StreamingDestination * sharedLocalDestination = nullptr; |
|
|
|
|
@ -122,14 +299,17 @@ namespace stream
@@ -122,14 +299,17 @@ namespace stream
|
|
|
|
|
DeleteI2NPMessage (m_LeaseSet); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void StreamingDestination::HandleNextPacket (const uint8_t * buf, size_t len) |
|
|
|
|
void StreamingDestination::HandleNextPacket (Packet * packet) |
|
|
|
|
{ |
|
|
|
|
uint32_t sendStreamID = *(uint32_t *)(buf); |
|
|
|
|
uint32_t sendStreamID = be32toh (*(uint32_t *)(packet->buf)); |
|
|
|
|
auto it = m_Streams.find (sendStreamID); |
|
|
|
|
if (it != m_Streams.end ()) |
|
|
|
|
it->second->HandleNextPacket (buf, len); |
|
|
|
|
it->second->HandleNextPacket (packet); |
|
|
|
|
else |
|
|
|
|
{ |
|
|
|
|
LogPrint ("Unknown stream ", sendStreamID); |
|
|
|
|
delete packet; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
Stream * StreamingDestination::CreateNewStream (const i2p::data::LeaseSet * remote) |
|
|
|
|
@ -150,10 +330,10 @@ namespace stream
@@ -150,10 +330,10 @@ namespace stream
|
|
|
|
|
|
|
|
|
|
I2NPMessage * StreamingDestination::GetLeaseSet () |
|
|
|
|
{ |
|
|
|
|
if (!m_LeaseSet) |
|
|
|
|
if (m_LeaseSet) // temporary always create new LeaseSet
|
|
|
|
|
DeleteI2NPMessage (m_LeaseSet); |
|
|
|
|
m_LeaseSet = CreateLeaseSet (); |
|
|
|
|
else |
|
|
|
|
FillI2NPMessageHeader (m_LeaseSet, eI2NPDatabaseStore); // refresh msgID
|
|
|
|
|
|
|
|
|
|
return m_LeaseSet; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -173,11 +353,12 @@ namespace stream
@@ -173,11 +353,12 @@ namespace stream
|
|
|
|
|
size += 256; // encryption key
|
|
|
|
|
memset (buf + size, 0, 128); |
|
|
|
|
size += 128; // signing key
|
|
|
|
|
auto tunnel = i2p::tunnel::tunnels.GetNextInboundTunnel (); |
|
|
|
|
if (tunnel) |
|
|
|
|
{ |
|
|
|
|
buf[size] = 1; // 1 lease
|
|
|
|
|
auto tunnels = i2p::tunnel::tunnels.GetInboundTunnels (5); // 5 tunnels maximum
|
|
|
|
|
buf[size] = tunnels.size (); // num leases
|
|
|
|
|
size++; // num
|
|
|
|
|
for (auto it: tunnels) |
|
|
|
|
{ |
|
|
|
|
auto tunnel = it; |
|
|
|
|
memcpy (buf + size, (const uint8_t *)tunnel->GetNextIdentHash (), 32); |
|
|
|
|
size += 32; // tunnel_gw
|
|
|
|
|
*(uint32_t *)(buf + size) = htobe32 (tunnel->GetNextTunnelID ()); |
|
|
|
|
@ -187,11 +368,6 @@ namespace stream
@@ -187,11 +368,6 @@ namespace stream
|
|
|
|
|
*(uint64_t *)(buf + size) = htobe64 (ts); |
|
|
|
|
size += 8; // end_date
|
|
|
|
|
} |
|
|
|
|
else |
|
|
|
|
{ |
|
|
|
|
buf[size] = 0; // zero leases
|
|
|
|
|
size++; // num
|
|
|
|
|
} |
|
|
|
|
Sign (buf, size, buf+ size); |
|
|
|
|
size += 40; // signature
|
|
|
|
|
|
|
|
|
|
@ -213,7 +389,7 @@ namespace stream
@@ -213,7 +389,7 @@ namespace stream
|
|
|
|
|
return sharedLocalDestination->CreateNewStream (remote); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void CloseStream (Stream * stream) |
|
|
|
|
void DeleteStream (Stream * stream) |
|
|
|
|
{ |
|
|
|
|
if (sharedLocalDestination) |
|
|
|
|
sharedLocalDestination->DeleteStream (stream); |
|
|
|
|
@ -230,13 +406,19 @@ namespace stream
@@ -230,13 +406,19 @@ namespace stream
|
|
|
|
|
CryptoPP::Gunzip decompressor; |
|
|
|
|
decompressor.Put (buf, length); |
|
|
|
|
decompressor.MessageEnd(); |
|
|
|
|
uint8_t uncompressed[2048]; |
|
|
|
|
int uncompressedSize = decompressor.MaxRetrievable (); |
|
|
|
|
decompressor.Get (uncompressed, uncompressedSize); |
|
|
|
|
Packet * uncompressed = new Packet; |
|
|
|
|
uncompressed->offset = 0; |
|
|
|
|
uncompressed->len = decompressor.MaxRetrievable (); |
|
|
|
|
if (uncompressed->len > MAX_PACKET_SIZE) |
|
|
|
|
{ |
|
|
|
|
LogPrint ("Recieved packet size exceeds mac packer size"); |
|
|
|
|
uncompressed->len = MAX_PACKET_SIZE; |
|
|
|
|
} |
|
|
|
|
decompressor.Get (uncompressed->buf, uncompressed->len); |
|
|
|
|
// then forward to streaming engine
|
|
|
|
|
// TODO: we have onle one destination, might be more
|
|
|
|
|
if (sharedLocalDestination) |
|
|
|
|
sharedLocalDestination->HandleNextPacket (uncompressed, uncompressedSize); |
|
|
|
|
sharedLocalDestination->HandleNextPacket (uncompressed); |
|
|
|
|
} |
|
|
|
|
else |
|
|
|
|
LogPrint ("Data: protocol ", buf[9], " is not supported"); |
|
|
|
|
@ -246,6 +428,7 @@ namespace stream
@@ -246,6 +428,7 @@ namespace stream
|
|
|
|
|
{ |
|
|
|
|
I2NPMessage * msg = NewI2NPMessage (); |
|
|
|
|
CryptoPP::Gzip compressor; |
|
|
|
|
compressor.SetDeflateLevel (CryptoPP::Gzip::MIN_DEFLATE_LEVEL); |
|
|
|
|
compressor.Put (payload, len); |
|
|
|
|
compressor.MessageEnd(); |
|
|
|
|
int size = compressor.MaxRetrievable (); |
|
|
|
|
@ -253,6 +436,7 @@ namespace stream
@@ -253,6 +436,7 @@ namespace stream
|
|
|
|
|
*(uint32_t *)buf = htobe32 (size); // length
|
|
|
|
|
buf += 4; |
|
|
|
|
compressor.Get (buf, size); |
|
|
|
|
memset (buf + 4, 0, 4); // source and destination ports. TODO: fill with proper values later
|
|
|
|
|
buf[9] = 6; // streaming protocol
|
|
|
|
|
msg->len += size + 4; |
|
|
|
|
FillI2NPMessageHeader (msg, eI2NPData); |
|
|
|
|
|
mikhail4021
11 years ago