lokinet/llarp/link/utp.cpp

917 lines
25 KiB
C++
Raw Normal View History

#include <llarp/link/utp.hpp>
#include "router.hpp"
#include <llarp/messages/link_intro.hpp>
2018-09-06 20:31:58 +00:00
#include <llarp/messages/discard.hpp>
#include <llarp/buffer.hpp>
#include <llarp/endian.h>
#include <utp.h>
2018-09-06 11:46:19 +00:00
#include <cassert>
2018-09-07 17:41:49 +00:00
#include <tuple>
2018-09-07 20:36:06 +00:00
#include <deque>
2018-09-06 20:31:58 +00:00
#ifdef __linux__
#include <linux/errqueue.h>
#include <netinet/ip_icmp.h>
#endif
namespace llarp
{
namespace utp
{
2018-09-06 20:31:58 +00:00
constexpr size_t FragmentHashSize = 32;
2018-09-07 17:41:49 +00:00
constexpr size_t FragmentNonceSize = 24;
constexpr size_t FragmentOverheadSize =
FragmentHashSize + FragmentNonceSize;
2018-09-07 20:36:06 +00:00
constexpr size_t FragmentBodyPayloadSize = 1024;
2018-09-07 17:41:49 +00:00
constexpr size_t FragmentBodyOverhead = sizeof(uint32_t) * 2;
constexpr size_t FragmentBodySize =
FragmentBodyOverhead + FragmentBodyPayloadSize;
2018-09-06 20:31:58 +00:00
constexpr size_t FragmentBufferSize =
2018-09-07 17:41:49 +00:00
FragmentOverheadSize + FragmentBodySize;
2018-09-06 11:46:19 +00:00
typedef llarp::AlignedBuffer< FragmentBufferSize > FragmentBuffer;
2018-09-06 20:31:58 +00:00
typedef llarp::AlignedBuffer< MAX_LINK_MSG_SIZE > MessageBuffer;
struct LinkLayer;
struct BaseSession : public ILinkSession
{
RouterContact remoteRC;
2018-09-06 11:46:19 +00:00
utp_socket* sock;
LinkLayer* parent;
bool gotLIM;
PubKey remoteTransportPubKey;
Addr remoteAddr;
SharedSecret sessionKey;
llarp_time_t lastActive;
2018-09-07 17:41:49 +00:00
const static llarp_time_t sessionTimeout = 30 * 1000;
2018-09-07 20:36:06 +00:00
std::deque< FragmentBuffer > sendq;
2018-09-07 17:41:49 +00:00
size_t sendBufOffset;
FragmentBuffer recvBuf;
2018-09-06 11:46:19 +00:00
size_t recvBufOffset;
2018-09-06 20:31:58 +00:00
MessageBuffer recvMsg;
size_t recvMsgOffset;
bool stalled = false;
2018-09-06 11:46:19 +00:00
2018-09-06 13:16:24 +00:00
void
2018-09-06 20:31:58 +00:00
Alive();
2018-09-06 13:16:24 +00:00
2018-09-06 11:46:19 +00:00
/// base
BaseSession(llarp_router* r);
2018-09-06 11:46:19 +00:00
/// outbound
BaseSession(llarp_router* r, utp_socket* s, const RouterContact& rc,
const AddressInfo& addr);
/// inbound
BaseSession(llarp_router* r, utp_socket* s, const Addr& remote);
2018-09-04 19:15:06 +00:00
enum State
{
eInitial,
eConnecting,
eLinkEstablished, // when utp connection is established
eCryptoHandshake, // crypto handshake initiated
eSessionReady, // session is ready
eClose // utp connection is closed
};
2018-09-06 11:46:19 +00:00
llarp_router*
Router();
State state;
void
2018-09-06 11:46:19 +00:00
OnLinkEstablished(LinkLayer* p)
{
2018-09-06 11:46:19 +00:00
parent = p;
EnterState(eLinkEstablished);
llarp::LogDebug("link established with ", remoteAddr);
}
2018-09-06 11:46:19 +00:00
void
EnterState(State st);
2018-09-04 19:15:06 +00:00
BaseSession();
virtual ~BaseSession();
void
2018-09-06 13:16:24 +00:00
PumpWrite()
{
2018-09-07 17:41:49 +00:00
while(sendq.size() > 0 && !stalled)
{
2018-09-07 17:41:49 +00:00
ssize_t expect = FragmentBufferSize - sendBufOffset;
ssize_t s = write_ll(sendq.front().data() + sendBufOffset, expect);
if(s != expect)
2018-09-06 20:31:58 +00:00
{
2018-09-07 17:41:49 +00:00
llarp::LogDebug("stalled at offset=", sendBufOffset, " sz=", s,
" to ", remoteAddr);
sendBufOffset += s;
2018-09-06 20:31:58 +00:00
stalled = true;
}
2018-09-07 17:41:49 +00:00
else
{
sendBufOffset = 0;
2018-09-07 20:36:06 +00:00
sendq.pop_front();
2018-09-07 17:41:49 +00:00
}
}
}
2018-09-06 20:31:58 +00:00
ssize_t
2018-09-07 17:41:49 +00:00
write_ll(byte_t* buf, size_t sz)
{
2018-09-06 13:16:24 +00:00
if(sock == nullptr)
{
llarp::LogWarn("write_ll failed: no socket");
2018-09-06 20:31:58 +00:00
return 0;
}
2018-09-07 17:41:49 +00:00
ssize_t s = utp_write(sock, buf, sz);
llarp::LogDebug("write_ll ", s, " of ", sz, " bytes to ", remoteAddr);
return s;
}
bool
2018-09-07 17:41:49 +00:00
VerifyThenDecrypt(byte_t* buf);
void
2018-09-07 20:36:06 +00:00
EncryptThenHash(const byte_t* ptr, uint32_t sz, bool isLastFragment);
bool
2018-09-06 11:46:19 +00:00
QueueWriteBuffers(llarp_buffer_t buf)
{
2018-09-06 13:16:24 +00:00
llarp::LogDebug("write ", buf.sz, " bytes to ", remoteAddr);
if(state != eSessionReady)
2018-09-04 19:15:06 +00:00
{
2018-09-06 13:16:24 +00:00
llarp::LogWarn("failed to send ", buf.sz,
" bytes on non ready session state=", state);
return false;
2018-09-04 19:15:06 +00:00
}
2018-09-07 17:41:49 +00:00
else
lastActive = llarp_time_now_ms();
2018-09-07 20:36:06 +00:00
size_t sz = buf.sz;
byte_t* ptr = buf.base;
while(sz)
{
2018-09-07 17:41:49 +00:00
uint32_t s = std::min(FragmentBodyPayloadSize, sz);
2018-09-07 20:36:06 +00:00
EncryptThenHash(ptr, s, ((sz - s) == 0));
ptr += s;
sz -= s;
}
return true;
}
2018-09-06 11:46:19 +00:00
void
Connect()
{
utp_connect(sock, remoteAddr, remoteAddr.SockLen());
EnterState(eConnecting);
}
void
OutboundLinkEstablished(LinkLayer* p)
{
OnLinkEstablished(p);
KeyExchangeNonce nonce;
nonce.Randomize();
gotLIM = true;
2018-09-06 20:31:58 +00:00
if(DoKeyExchange(Router()->crypto.transport_dh_client, nonce,
remoteTransportPubKey, Router()->encryption))
2018-09-06 11:46:19 +00:00
{
2018-09-07 17:41:49 +00:00
SendHandshake(nonce);
2018-09-06 11:46:19 +00:00
EnterState(eSessionReady);
2018-09-07 17:41:49 +00:00
SendKeepAlive();
2018-09-06 11:46:19 +00:00
}
}
// send our RC to the remote
void
2018-09-07 17:41:49 +00:00
SendHandshake(const KeyExchangeNonce& n)
2018-09-06 11:46:19 +00:00
{
2018-09-07 17:41:49 +00:00
FragmentBuffer tmp;
auto buf = InitBuffer(tmp.data(), tmp.size());
2018-09-06 11:46:19 +00:00
// fastforward buffer for handshake to fit before
buf.cur += sizeof(uint32_t) * 2;
byte_t* begin = buf.cur;
LinkIntroMessage msg;
msg.rc = Router()->rc;
2018-09-06 20:31:58 +00:00
msg.N = n;
2018-09-06 11:46:19 +00:00
if(!msg.BEncode(&buf))
{
llarp::LogError("failed to encode our RC for handshake");
2018-09-06 20:31:58 +00:00
Close();
2018-09-06 11:46:19 +00:00
return;
}
uint32_t sz = buf.cur - begin;
llarp::LogDebug("handshake is of size ", sz, " bytes");
// write handshake header
buf.cur = buf.base;
llarp_buffer_put_uint32(&buf, LLARP_PROTO_VERSION);
llarp_buffer_put_uint32(&buf, sz);
// send it
2018-09-07 17:41:49 +00:00
write_ll(tmp.data(), sz + (sizeof(uint32_t) * 2));
2018-09-06 11:46:19 +00:00
}
2018-09-04 19:15:06 +00:00
bool
2018-09-06 20:31:58 +00:00
DoKeyExchange(llarp_transport_dh_func dh, const KeyExchangeNonce& n,
const PubKey& other, const SecretKey& secret)
{
2018-09-06 20:31:58 +00:00
PubKey us = llarp::seckey_topublic(secret);
2018-09-07 17:41:49 +00:00
llarp::LogDebug("DH us=", us, " them=", other, " n=", n);
if(!dh(sessionKey, other, secret, n))
{
llarp::LogError("key exchange with ", other, " failed");
2018-09-06 20:31:58 +00:00
Close();
2018-09-04 19:15:06 +00:00
return false;
}
2018-09-04 19:15:06 +00:00
return true;
}
void
2018-09-06 11:46:19 +00:00
TickImpl(llarp_time_t now)
{
}
void
2018-09-07 17:41:49 +00:00
Close();
2018-09-07 20:36:06 +00:00
bool
RecvHandshake(const void* buf, size_t bufsz, LinkLayer* p);
bool
Recv(const void* buf, size_t sz)
{
2018-09-07 17:41:49 +00:00
if(state != eSessionReady)
{
llarp::LogWarn("session not ready via ", remoteAddr);
return false;
}
Alive();
byte_t* ptr = (const byte_t*)buf;
llarp::LogDebug("utp read ", sz, " from ", remoteAddr);
2018-09-06 20:31:58 +00:00
size_t s = sz;
2018-09-07 17:41:49 +00:00
// process leftovers
if(recvBufOffset)
{
2018-09-07 20:36:06 +00:00
auto left = FragmentBufferSize - recvBufOffset;
2018-09-07 17:41:49 +00:00
if(s >= left)
2018-09-06 20:31:58 +00:00
{
2018-09-07 20:36:06 +00:00
// yes it fills it
2018-09-07 17:41:49 +00:00
llarp::LogDebug("process leftovers, offset=", recvBufOffset,
" sz=", s, " left=", left);
memcpy(recvBuf.data() + recvBufOffset, ptr, left);
s -= left;
recvBufOffset = 0;
ptr += left;
2018-09-07 20:36:06 +00:00
if(!VerifyThenDecrypt(recvBuf.data()))
2018-09-06 20:31:58 +00:00
return false;
}
}
2018-09-07 20:36:06 +00:00
// process full fragments
2018-09-07 17:41:49 +00:00
while(s >= FragmentBufferSize)
{
2018-09-07 20:36:06 +00:00
recvBufOffset = 0;
2018-09-07 17:41:49 +00:00
llarp::LogDebug("process full sz=", s);
if(!VerifyThenDecrypt(ptr))
return false;
ptr += FragmentBufferSize;
s -= FragmentBufferSize;
}
if(s)
{
// hold onto leftovers
llarp::LogDebug("leftovers sz=", s);
2018-09-07 20:36:06 +00:00
memcpy(recvBuf.data() + recvBufOffset, ptr, s);
recvBufOffset += s;
2018-09-07 17:41:49 +00:00
}
return true;
}
bool
2018-09-06 11:46:19 +00:00
IsTimedOut(llarp_time_t now) const
{
if(now < lastActive)
return false;
2018-09-07 17:41:49 +00:00
auto dlt = now - lastActive;
if(dlt >= sessionTimeout)
{
llarp::LogDebug("session timeout reached for ", remoteAddr);
return true;
}
return false;
}
const PubKey&
2018-09-06 13:16:24 +00:00
RemotePubKey() const
{
return remoteRC.pubkey;
}
const Addr&
2018-09-07 17:41:49 +00:00
RemoteEndpoint() const
{
return remoteAddr;
}
void
MarkEstablished();
2018-09-04 19:15:06 +00:00
}; // namespace utp
struct LinkLayer : public ILinkLayer
{
utp_context* _utp_ctx = nullptr;
2018-09-06 11:46:19 +00:00
llarp_router* router = nullptr;
static uint64
2018-09-06 11:46:19 +00:00
OnRead(utp_callback_arguments* arg);
static uint64
SendTo(utp_callback_arguments* arg)
{
LinkLayer* l =
static_cast< LinkLayer* >(utp_context_get_userdata(arg->context));
2018-09-06 20:31:58 +00:00
llarp::LogDebug("utp_sendto ", Addr(*arg->address), " ", arg->len,
" bytes");
if(sendto(l->m_udp.fd, arg->buf, arg->len, arg->flags, arg->address,
arg->address_len)
== -1)
{
llarp::LogError("sendto failed: ", strerror(errno));
}
return 0;
}
static uint64
OnError(utp_callback_arguments* arg)
{
llarp::LogError(utp_error_code_names[arg->error_code]);
return 0;
}
static uint64
2018-09-06 11:46:19 +00:00
OnStateChange(utp_callback_arguments*);
static uint64
OnAccept(utp_callback_arguments*);
2018-09-06 20:31:58 +00:00
static uint64
OnLog(utp_callback_arguments* arg)
{
llarp::LogDebug(arg->buf);
return 0;
}
2018-09-06 11:46:19 +00:00
LinkLayer(llarp_router* r) : ILinkLayer()
{
2018-09-06 11:46:19 +00:00
router = r;
_utp_ctx = utp_init(2);
utp_context_set_userdata(_utp_ctx, this);
utp_set_callback(_utp_ctx, UTP_SENDTO, &LinkLayer::SendTo);
utp_set_callback(_utp_ctx, UTP_ON_ACCEPT, &LinkLayer::OnAccept);
2018-09-06 11:46:19 +00:00
utp_set_callback(_utp_ctx, UTP_ON_STATE_CHANGE,
&LinkLayer::OnStateChange);
utp_set_callback(_utp_ctx, UTP_ON_READ, &LinkLayer::OnRead);
2018-09-06 20:31:58 +00:00
utp_set_callback(_utp_ctx, UTP_ON_ERROR, &LinkLayer::OnError);
utp_set_callback(_utp_ctx, UTP_LOG, &LinkLayer::OnLog);
2018-09-04 19:15:06 +00:00
utp_context_set_option(_utp_ctx, UTP_LOG_NORMAL, 1);
utp_context_set_option(_utp_ctx, UTP_LOG_MTU, 1);
utp_context_set_option(_utp_ctx, UTP_LOG_DEBUG, 1);
2018-09-07 17:41:49 +00:00
utp_context_set_option(_utp_ctx, UTP_SNDBUF, MAX_LINK_MSG_SIZE * 16);
2018-09-06 20:31:58 +00:00
utp_context_set_option(_utp_ctx, UTP_RCVBUF, MAX_LINK_MSG_SIZE * 64);
}
~LinkLayer()
{
utp_destroy(_utp_ctx);
}
uint16_t
Rank() const
{
return 1;
}
void
RecvFrom(const Addr& from, const void* buf, size_t sz)
{
utp_process_udp(_utp_ctx, (const byte_t*)buf, sz, from, from.SockLen());
}
2018-09-06 20:31:58 +00:00
#ifdef __linux__
void
ProcessICMP()
{
do
{
byte_t vec_buf[4096], ancillary_buf[4096];
struct iovec iov = {vec_buf, sizeof(vec_buf)};
struct sockaddr_in remote;
struct msghdr msg;
ssize_t len;
struct cmsghdr* cmsg;
struct sock_extended_err* e;
struct sockaddr* icmp_addr;
struct sockaddr_in* icmp_sin;
memset(&msg, 0, sizeof(msg));
msg.msg_name = &remote;
msg.msg_namelen = sizeof(remote);
msg.msg_iov = &iov;
msg.msg_iovlen = 1;
msg.msg_flags = 0;
msg.msg_control = ancillary_buf;
msg.msg_controllen = sizeof(ancillary_buf);
len = recvmsg(m_udp.fd, &msg, MSG_ERRQUEUE | MSG_DONTWAIT);
if(len < 0)
{
if(errno == EAGAIN || errno == EWOULDBLOCK)
errno = 0;
else
llarp::LogError("failed to read icmp for utp ", strerror(errno));
return;
}
for(cmsg = CMSG_FIRSTHDR(&msg); cmsg; cmsg = CMSG_NXTHDR(&msg, cmsg))
{
if(cmsg->cmsg_type != IP_RECVERR)
{
continue;
}
if(cmsg->cmsg_level != SOL_IP)
{
continue;
}
e = (struct sock_extended_err*)CMSG_DATA(cmsg);
if(!e)
continue;
if(e->ee_origin != SO_EE_ORIGIN_ICMP)
{
continue;
}
icmp_addr = (struct sockaddr*)SO_EE_OFFENDER(e);
icmp_sin = (struct sockaddr_in*)icmp_addr;
if(icmp_sin->sin_port != 0)
{
continue;
}
if(e->ee_type == 3 && e->ee_code == 4)
{
utp_process_icmp_fragmentation(_utp_ctx, vec_buf, len,
(struct sockaddr*)&remote,
sizeof(remote), e->ee_info);
}
else
{
utp_process_icmp_error(_utp_ctx, vec_buf, len,
(struct sockaddr*)&remote, sizeof(remote));
}
}
} while(true);
}
#endif
void
Pump()
{
2018-09-07 17:41:49 +00:00
utp_issue_deferred_acks(_utp_ctx);
2018-09-06 20:31:58 +00:00
#ifdef __linux__
ProcessICMP();
#endif
2018-09-07 17:41:49 +00:00
ILinkLayer::Pump();
2018-09-06 20:31:58 +00:00
}
void
Stop()
{
}
2018-09-06 11:46:19 +00:00
llarp_router*
GetRouter();
bool
KeyGen(SecretKey& k)
{
2018-09-06 11:46:19 +00:00
router->crypto.encryption_keygen(k);
return true;
}
2018-09-06 20:31:58 +00:00
void
Tick(llarp_time_t now)
{
utp_check_timeouts(_utp_ctx);
ILinkLayer::Tick(now);
}
2018-09-06 13:16:24 +00:00
ILinkSession*
NewOutboundSession(const RouterContact& rc, const AddressInfo& addr);
utp_socket*
NewSocket()
{
return utp_create_socket(_utp_ctx);
}
const char*
Name() const
{
return "utp";
}
};
std::unique_ptr< ILinkLayer >
NewServer(llarp_router* r)
{
2018-09-04 19:15:06 +00:00
return std::unique_ptr< LinkLayer >(new LinkLayer(r));
}
2018-09-06 11:46:19 +00:00
BaseSession::BaseSession(llarp_router* r)
{
2018-09-07 17:41:49 +00:00
remoteTransportPubKey.Zero();
2018-09-06 20:31:58 +00:00
parent = nullptr;
recvMsgOffset = 0;
SendKeepAlive = [&]() -> bool {
2018-09-07 17:41:49 +00:00
if(false && sendq.size() == 0)
2018-09-06 20:31:58 +00:00
{
DiscardMessage msg;
byte_t tmp[128] = {0};
auto buf = llarp::StackBuffer< decltype(tmp) >(tmp);
if(!msg.BEncode(&buf))
return false;
buf.sz = buf.cur - buf.base;
buf.cur = buf.base;
if(!this->QueueWriteBuffers(buf))
return false;
}
return true;
};
2018-09-07 17:41:49 +00:00
sendBufOffset = 0;
2018-09-06 11:46:19 +00:00
recvBufOffset = 0;
TimedOut = [&](llarp_time_t now) -> bool {
2018-09-07 17:41:49 +00:00
return this->IsTimedOut(now) || this->state == eClose;
2018-09-06 11:46:19 +00:00
};
2018-09-06 13:16:24 +00:00
GetPubKey = std::bind(&BaseSession::RemotePubKey, this);
2018-09-06 11:46:19 +00:00
lastActive = llarp_time_now_ms();
2018-09-06 20:31:58 +00:00
Pump = std::bind(&BaseSession::PumpWrite, this);
2018-09-06 11:46:19 +00:00
Tick = std::bind(&BaseSession::TickImpl, this, std::placeholders::_1);
2018-09-07 17:41:49 +00:00
SendMessageBuffer = std::bind(&BaseSession::QueueWriteBuffers, this,
std::placeholders::_1);
IsEstablished = [=]() {
return this->state == eSessionReady || this->state == eLinkEstablished;
2018-09-06 11:46:19 +00:00
};
2018-09-07 17:41:49 +00:00
SendClose = std::bind(&BaseSession::Close, this);
GetRemoteEndpoint = std::bind(&BaseSession::RemoteEndpoint, this);
2018-09-06 11:46:19 +00:00
}
2018-09-06 11:46:19 +00:00
BaseSession::BaseSession(llarp_router* r, utp_socket* s,
const RouterContact& rc, const AddressInfo& addr)
: BaseSession(r)
{
2018-09-06 11:46:19 +00:00
remoteRC.Clear();
remoteTransportPubKey = addr.pubkey;
remoteRC = rc;
sock = s;
assert(utp_set_userdata(sock, this) == this);
2018-09-06 13:16:24 +00:00
assert(s == sock);
2018-09-06 11:46:19 +00:00
remoteAddr = addr;
Start = std::bind(&BaseSession::Connect, this);
}
2018-09-06 11:46:19 +00:00
BaseSession::BaseSession(llarp_router* r, utp_socket* s, const Addr& addr)
: BaseSession(r)
{
remoteRC.Clear();
sock = s;
2018-09-06 13:16:24 +00:00
assert(s == sock);
2018-09-06 11:46:19 +00:00
assert(utp_set_userdata(sock, this) == this);
remoteAddr = addr;
Start = []() {};
}
2018-09-06 11:46:19 +00:00
llarp_router*
BaseSession::Router()
{
2018-09-06 11:46:19 +00:00
return parent->router;
}
BaseSession::~BaseSession()
{
2018-09-06 13:16:24 +00:00
if(sock)
{
utp_set_userdata(sock, nullptr);
2018-09-07 17:41:49 +00:00
sock = nullptr;
2018-09-06 13:16:24 +00:00
}
}
2018-09-06 13:16:24 +00:00
ILinkSession*
LinkLayer::NewOutboundSession(const RouterContact& rc,
const AddressInfo& addr)
{
2018-09-06 13:16:24 +00:00
return new BaseSession(router, utp_create_socket(_utp_ctx), rc, addr);
2018-09-06 11:46:19 +00:00
}
uint64
LinkLayer::OnRead(utp_callback_arguments* arg)
{
2018-09-06 20:31:58 +00:00
LinkLayer* parent =
static_cast< LinkLayer* >(utp_context_get_userdata(arg->context));
2018-09-06 11:46:19 +00:00
BaseSession* self =
static_cast< BaseSession* >(utp_get_userdata(arg->socket));
2018-09-07 17:41:49 +00:00
2018-09-06 11:46:19 +00:00
if(self)
{
2018-09-06 13:16:24 +00:00
if(self->state == BaseSession::eClose)
{
return 0;
}
else if(self->state == BaseSession::eSessionReady)
2018-09-06 20:31:58 +00:00
{
2018-09-07 17:41:49 +00:00
if(!self->Recv(arg->buf, arg->len))
{
llarp::LogDebug("recv fail for ", self->remoteAddr);
self->Close();
return 0;
}
utp_read_drained(arg->socket);
2018-09-06 20:31:58 +00:00
}
2018-09-06 11:46:19 +00:00
else if(self->state == BaseSession::eLinkEstablished)
{
2018-09-07 20:36:06 +00:00
if(!self->RecvHandshake(arg->buf, arg->len, parent))
{
llarp::LogDebug("recv handshake failed for ", self->remoteAddr);
self->Close();
return 0;
}
2018-09-07 17:41:49 +00:00
utp_read_drained(arg->socket);
2018-09-06 11:46:19 +00:00
}
}
else
{
llarp::LogWarn("utp_socket got data with no underlying session");
}
return 0;
}
2018-09-06 11:46:19 +00:00
uint64
LinkLayer::OnStateChange(utp_callback_arguments* arg)
{
2018-09-06 11:46:19 +00:00
LinkLayer* l =
static_cast< LinkLayer* >(utp_context_get_userdata(arg->context));
BaseSession* session =
static_cast< BaseSession* >(utp_get_userdata(arg->socket));
2018-09-06 13:16:24 +00:00
if(session)
2018-09-06 11:46:19 +00:00
{
2018-09-06 13:16:24 +00:00
if(arg->state == UTP_STATE_CONNECT)
{
if(session->state == BaseSession::eClose)
{
return 0;
}
session->OutboundLinkEstablished(l);
}
2018-09-06 20:31:58 +00:00
else if(arg->state == UTP_STATE_WRITABLE)
{
2018-09-07 17:41:49 +00:00
if(session->IsEstablished())
{
llarp::LogDebug("write resumed for ", session->remoteAddr);
session->stalled = false;
session->PumpWrite();
}
2018-09-06 20:31:58 +00:00
}
2018-09-06 13:16:24 +00:00
else if(arg->state == UTP_STATE_EOF)
{
2018-09-07 17:41:49 +00:00
llarp::LogDebug("got eof from ", session->remoteAddr);
session->Close();
2018-09-06 13:16:24 +00:00
}
2018-09-06 11:46:19 +00:00
}
return 0;
}
uint64
LinkLayer::OnAccept(utp_callback_arguments* arg)
{
LinkLayer* self =
static_cast< LinkLayer* >(utp_context_get_userdata(arg->context));
Addr remote(*arg->address);
llarp::LogDebug("utp accepted from ", remote);
2018-09-06 11:46:19 +00:00
BaseSession* session = new BaseSession(self->router, arg->socket, remote);
self->PutSession(remote, session);
2018-09-06 11:46:19 +00:00
session->OnLinkEstablished(self);
return 0;
}
2018-09-06 11:46:19 +00:00
void
2018-09-07 20:36:06 +00:00
BaseSession::EncryptThenHash(const byte_t* ptr, uint32_t sz,
bool isLastFragment)
2018-09-06 11:46:19 +00:00
{
2018-09-07 17:41:49 +00:00
if(state != eSessionReady)
{
llarp::LogWarn("tried to send to non ready session on ", remoteAddr);
return;
}
2018-09-07 20:36:06 +00:00
FragmentBuffer buf;
2018-09-06 20:31:58 +00:00
llarp::LogDebug("encrypt then hash ", sz, " bytes last=", isLastFragment);
2018-09-06 11:46:19 +00:00
buf.Randomize();
2018-09-07 20:36:06 +00:00
byte_t* nonce = buf.data() + FragmentHashSize;
byte_t* body = nonce + FragmentNonceSize;
byte_t* base = body;
2018-09-06 11:46:19 +00:00
if(isLastFragment)
htobe32buf(body, 0);
2018-09-07 17:41:49 +00:00
else
htobe32buf(body, 1);
2018-09-06 11:46:19 +00:00
body += sizeof(uint32_t);
htobe32buf(body, sz);
body += sizeof(uint32_t);
memcpy(body, ptr, sz);
2018-09-07 20:36:06 +00:00
auto payload =
InitBuffer(base, FragmentBufferSize - FragmentOverheadSize);
// encrypt
2018-09-07 17:41:49 +00:00
Router()->crypto.xchacha20(payload, sessionKey, nonce);
2018-09-07 20:36:06 +00:00
payload.base = nonce;
2018-09-06 20:31:58 +00:00
payload.cur = payload.base;
2018-09-07 20:36:06 +00:00
payload.sz = FragmentBufferSize - FragmentHashSize;
// key'd hash
2018-09-07 17:41:49 +00:00
Router()->crypto.hmac(buf.data(), payload, sessionKey);
2018-09-07 20:36:06 +00:00
// push back
sendq.push_back(std::move(buf));
2018-09-06 11:46:19 +00:00
}
void
BaseSession::EnterState(State st)
{
2018-09-06 13:16:24 +00:00
state = st;
2018-09-06 11:46:19 +00:00
if(st == eSessionReady)
{
2018-09-07 17:41:49 +00:00
parent->MapAddr(remoteRC.pubkey, this);
2018-09-06 11:46:19 +00:00
Router()->HandleLinkSessionEstablished(remoteRC);
}
2018-09-06 20:31:58 +00:00
Alive();
2018-09-06 11:46:19 +00:00
}
bool
2018-09-07 17:41:49 +00:00
BaseSession::VerifyThenDecrypt(byte_t* buf)
2018-09-06 11:46:19 +00:00
{
2018-09-06 20:31:58 +00:00
llarp::LogDebug("verify then decrypt ", remoteAddr);
2018-09-06 11:46:19 +00:00
ShortHash digest;
2018-09-06 20:31:58 +00:00
2018-09-07 17:41:49 +00:00
auto hbuf = InitBuffer(buf + FragmentHashSize,
FragmentBufferSize - FragmentHashSize);
if(!Router()->crypto.hmac(digest.data(), hbuf, sessionKey))
2018-09-06 11:46:19 +00:00
{
llarp::LogError("keyed hash failed");
return false;
}
2018-09-07 17:41:49 +00:00
ShortHash expected(buf);
if(expected != digest)
2018-09-06 11:46:19 +00:00
{
2018-09-06 20:31:58 +00:00
llarp::LogError("Message Integrity Failed: got ", digest, " from ",
2018-09-07 17:41:49 +00:00
remoteAddr, " instead of ", expected);
llarp::DumpBuffer(InitBuffer(buf, FragmentBufferSize));
2018-09-06 11:46:19 +00:00
return false;
}
2018-09-07 17:41:49 +00:00
auto body = InitBuffer(buf + FragmentOverheadSize,
2018-09-06 11:46:19 +00:00
FragmentBufferSize - FragmentOverheadSize);
2018-09-07 17:41:49 +00:00
Router()->crypto.xchacha20(body, sessionKey, buf + FragmentHashSize);
2018-09-06 11:46:19 +00:00
uint32_t upper, lower;
if(!(llarp_buffer_read_uint32(&body, &upper)
&& llarp_buffer_read_uint32(&body, &lower)))
return false;
bool fragmentEnd = upper == 0;
2018-09-06 20:31:58 +00:00
llarp::LogDebug("fragment size ", lower, " from ", remoteAddr);
if(lower + recvMsgOffset > recvMsg.size())
2018-09-06 11:46:19 +00:00
{
llarp::LogError("Fragment too big: ", lower, " bytes");
return false;
}
2018-09-07 20:36:06 +00:00
memcpy(recvMsg.data() + recvMsgOffset, body.cur, lower);
2018-09-06 20:31:58 +00:00
recvMsgOffset += lower;
2018-09-06 11:46:19 +00:00
if(fragmentEnd)
{
// got a message
2018-09-06 20:31:58 +00:00
llarp::LogDebug("end of message from ", remoteAddr);
2018-09-07 20:36:06 +00:00
auto mbuf = InitBuffer(recvMsg.data(), recvMsgOffset);
if(!Router()->HandleRecvLinkMessageBuffer(this, mbuf))
2018-09-06 20:31:58 +00:00
{
llarp::LogWarn("failed to handle message from ", remoteAddr);
llarp::DumpBuffer(mbuf);
}
recvMsgOffset = 0;
2018-09-06 11:46:19 +00:00
}
return true;
}
2018-09-07 20:36:06 +00:00
bool
BaseSession::RecvHandshake(const void* buf, size_t bufsz, LinkLayer* p)
2018-09-06 11:46:19 +00:00
{
size_t sz = bufsz;
parent = p;
llarp::LogDebug("recv handshake ", sz, " from ", remoteAddr);
if(sz <= 8)
{
llarp::LogDebug("handshake too small from ", remoteAddr);
2018-09-06 20:31:58 +00:00
Close();
2018-09-07 20:36:06 +00:00
return false;
2018-09-06 11:46:19 +00:00
}
// process handshake header
2018-09-07 17:41:49 +00:00
byte_t* ptr = (byte_t*)buf;
2018-09-06 11:46:19 +00:00
uint32_t version = bufbe32toh(ptr);
if(version != LLARP_PROTO_VERSION)
{
llarp::LogWarn("protocol version missmatch ", version,
" != ", LLARP_PROTO_VERSION);
2018-09-06 20:31:58 +00:00
Close();
2018-09-07 20:36:06 +00:00
return false;
2018-09-06 11:46:19 +00:00
}
ptr += sizeof(uint32_t);
sz -= sizeof(uint32_t);
uint32_t limsz = bufbe32toh(ptr);
ptr += sizeof(uint32_t);
sz -= sizeof(uint32_t);
if(limsz > sz)
{
// not enough data
// TODO: don't bail here, continue reading
llarp::LogDebug("not enough data for handshake, want ", limsz,
" bytes but got ", sz);
2018-09-06 20:31:58 +00:00
Close();
2018-09-07 20:36:06 +00:00
return false;
2018-09-06 11:46:19 +00:00
}
2018-09-07 20:36:06 +00:00
llarp::LogDebug("from ", bufsz, " bytes reading ", limsz, " of ", sz,
" bytes");
2018-09-06 11:46:19 +00:00
// process LIM
auto mbuf = InitBuffer(ptr, limsz);
LinkIntroMessage msg(this);
if(!msg.BDecode(&mbuf))
{
llarp::LogError("Failed to parse LIM from ", remoteAddr);
llarp::DumpBuffer(mbuf);
2018-09-06 20:31:58 +00:00
Close();
2018-09-07 20:36:06 +00:00
return false;
2018-09-06 11:46:19 +00:00
}
if(!msg.HandleMessage(Router()))
{
llarp::LogError("failed to verify signature of rc");
2018-09-07 20:36:06 +00:00
return false;
2018-09-06 11:46:19 +00:00
}
2018-09-07 20:36:06 +00:00
gotLIM = true;
2018-09-07 17:41:49 +00:00
sz -= limsz;
2018-09-06 11:46:19 +00:00
remoteRC = msg.rc;
2018-09-06 20:31:58 +00:00
if(!DoKeyExchange(Router()->crypto.transport_dh_server, msg.N,
remoteRC.enckey, parent->TransportSecretKey()))
2018-09-07 20:36:06 +00:00
return false;
EnterState(eSessionReady);
2018-09-07 17:41:49 +00:00
if(sz)
{
llarp::LogDebug("got ", sz, " leftover from handshake from ",
remoteAddr);
2018-09-07 20:36:06 +00:00
return Recv(ptr + limsz, sz);
2018-09-07 17:41:49 +00:00
}
2018-09-07 20:36:06 +00:00
else
{
llarp::LogDebug("no leftovers in handshake from ", remoteAddr);
}
return true;
2018-09-06 20:31:58 +00:00
}
void
2018-09-07 17:41:49 +00:00
BaseSession::Close()
2018-09-06 20:31:58 +00:00
{
if(state != eClose)
{
2018-09-07 20:36:06 +00:00
if(sock)
{
utp_shutdown(sock, SHUT_RDWR);
utp_close(sock);
llarp::LogDebug("utp_close ", remoteAddr);
utp_set_userdata(sock, nullptr);
}
2018-09-06 20:31:58 +00:00
}
EnterState(eClose);
sock = nullptr;
}
void
BaseSession::Alive()
{
lastActive = llarp_time_now_ms();
2018-09-06 11:46:19 +00:00
}
} // namespace utp
} // namespace llarp