lokinet/llarp/iwp/session.cpp

#include <llarp/crypto.hpp>
#include <llarp/iwp/server.hpp>
#include <llarp/iwp/session.hpp>
#include "address_info.hpp"
#include "buffer.hpp"
#include "link/encoder.hpp"

#include "llarp/ev.h"  // for handle_frame_encrypt

static void
handle_crypto_outbound(void *u)
{
  llarp_link_session *self = static_cast< llarp_link_session * >(u);
  self->EncryptOutboundFrames();
  self->working = false;
}

// TODO: move this orphan function?
static void
handle_frame_encrypt(iwp_async_frame *frame)
{
  llarp_link_session *self = static_cast< llarp_link_session * >(frame->user);
  llarp::Debug("tx ", frame->sz);
  if(llarp_ev_udp_sendto(self->udp, self->addr, frame->buf, frame->sz) == -1)
    llarp::Warn("sendto failed");
}

llarp_link_session::llarp_link_session(llarp_udp_io *u, llarp_async_iwp *i, llarp_crypto *c,
                     llarp_logic *l, const byte_t *seckey, const llarp::Addr &a)
      : udp(u)
      , crypto(c)
      , iwp(i)
      , logic(l)
      , outboundFrames("iwp_outbound")
      //, decryptedFrames("iwp_inbound")
      , addr(a)
      , state(eInitial)
{
  if(seckey)
    eph_seckey = seckey;
  else
    crypto->encryption_keygen(eph_seckey);
  llarp_rc_clear(&remote_router);
  crypto->randbytes(token, 32);
  llarp::Info("session created");
}

llarp_link_session::~llarp_link_session()
{
  llarp_rc_free(&remote_router);
  frame.clear();
}

bool
llarp_link_session::sendto(llarp_buffer_t msg)
{
  auto id = frame.txids++;
  // llarp::Debug("session sending to, number", id);
  llarp::ShortHash digest;
  crypto->shorthash(digest, msg);
  transit_message *m = new transit_message(msg, digest, id);
  add_outbound_message(id, m);
  return true;
}

bool
llarp_link_session::timedout(llarp_time_t now, llarp_time_t timeout)
{
  auto diff = now - frame.lastEvent;
  return diff >= timeout;
}

bool
llarp_link_session::has_timed_out()
{
  auto now = llarp_time_now_ms();
  return timedout(now);
}

static void
send_keepalive(void *user)
{
  llarp_link_session *self = static_cast< llarp_link_session * >(user);
  // if both sides agree on invalidation
  if(self->is_invalidated())
  {
    // don't send keepalive
    llarp::Info("session cant send keepalive because were invalid");
    return;
  }
  // all zeros means keepalive
  byte_t tmp[8] = {0};
  // set flags for tx
  frame_header hdr(tmp);
  hdr.flags() = self->frame.txflags;

  // send frame after encrypting
  auto buf            = llarp::StackBuffer< decltype(tmp) >(tmp);
  self->now           = llarp_time_now_ms();
  self->lastKeepalive = self->now;
  self->encrypt_frame_async_send(buf.base, buf.sz);
  self->pump();
  self->PumpCryptoOutbound();
}

void
llarp_link_session::close()
{
  // set our side invalidated and close async when the other side also marks
  // as session invalidated
  frame.txflags |= eSessionInvalidated;
  // TODO: add timer for session invalidation
  llarp_logic_queue_job(logic, {this, &send_keepalive});
}

void
llarp_link_session::session_established()
{
  llarp::RouterID remote = remote_router.pubkey;
  llarp::Info("Session to ", remote, " established");
  EnterState(eEstablished);
  serv->MapAddr(addr, remote_router.pubkey);
  llarp_logic_cancel_call(logic, establish_job_id);
}

llarp_rc *
llarp_link_session::get_remote_router()
{
  return &remote_router;
}

void
llarp_link_session::add_outbound_message(uint64_t id, transit_message *msg)
{
  llarp::Debug("add outbound message ", id, " of size ",
               msg->msginfo.totalsize(),
               " numfrags=", (int)msg->msginfo.numfrags(),
               " lastfrag=", (int)msg->msginfo.lastfrag());

  frame.queue_tx(id, msg);
  pump();
  PumpCryptoOutbound();
}

bool
llarp_link_session::CheckRCValid()
{
  // verify signatuire
  if(!llarp_rc_verify_sig(crypto, &remote_router))
    return false;

  auto &list = remote_router.addrs->list;
  if(list.size() == 0)  // the remote node is a client node so accept it
    return true;
  // check if the RC owns a pubkey that we are using
  for(auto &ai : list)
  {
    if(memcmp(ai.enc_key, remote, PUBKEYSIZE) == 0)
      return true;
  }
  return false;
}

bool
llarp_link_session::IsEstablished()
{
  return state == eEstablished;
}

void
llarp_link_session::send_LIM()
{
  llarp::Debug("send LIM");
  llarp::ShortHash digest;
  // 64 bytes overhead for link message
  byte_t tmp[MAX_RC_SIZE + 64];
  auto buf = llarp::StackBuffer< decltype(tmp) >(tmp);
  // return a llarp_buffer_t of encoded link message
  if(llarp::EncodeLIM(&buf, our_router))
  {
    // rewind message buffer
    buf.sz  = buf.cur - buf.base;
    buf.cur = buf.base;
    // hash message buffer
    crypto->shorthash(digest, buf);
    auto id  = frame.txids++;
    auto msg = new transit_message(buf, digest, id);
    // put into outbound send queue
    add_outbound_message(id, msg);
    // enter state
    EnterState(eLIMSent);
  }
  else
    llarp::Error("LIM Encode failed");
}

static void
handle_generated_session_start(iwp_async_session_start *start)
{
  llarp_link_session *link = static_cast< llarp_link_session * >(start->user);
  link->working            = false;
  if(llarp_ev_udp_sendto(link->udp, link->addr, start->buf, start->sz) == -1)
    llarp::Error("sendto failed");
  link->EnterState(llarp_link_session::State::eSessionStartSent);
  link->serv->remove_intro_from(link->addr);
}

static void
handle_verify_intro(iwp_async_intro *intro)
{
  llarp_link_session *self = static_cast< llarp_link_session * >(intro->user);
  self->working            = false;
  if(!intro->buf)
  {
    self->serv->remove_intro_from(self->addr);
    llarp::Error("intro verify failed from ", self->addr, " via ",
                 self->serv->addr);
    delete self;
    return;
  }
  self->intro_ack();
}

static void
handle_verify_introack(iwp_async_introack *introack)
{
  llarp_link_session *link =
      static_cast< llarp_link_session * >(introack->user);

  link->working = false;
  if(introack->buf == nullptr)
  {
    // invalid signature
    llarp::Error("introack verify failed from ", link->addr);
    link->serv->remove_intro_from(link->addr);
    link->serv->RemoveSessionByAddr(link->addr);
    return;
  }
  link->EnterState(llarp_link_session::eIntroAckRecv);
  link->session_start();
}

static void
handle_establish_timeout(void *user, uint64_t orig, uint64_t left)
{
  if(orig == 0)
    return;
  llarp_link_session *self = static_cast< llarp_link_session * >(user);
  self->establish_job_id   = 0;
  if(self->establish_job)
  {
    auto job            = self->establish_job;
    self->establish_job = nullptr;
    job->link           = self->serv->parent;
    if(self->IsEstablished())
    {
      job->session = self->parent;
    }
    else
    {
      // timer timeout
      job->session = nullptr;
    }
    job->result(job);
  }
}

void
llarp_link_session::done()
{
  auto logic = serv->logic;
  serv->remove_intro_from(addr);
  if(establish_job_id)
  {
    llarp_logic_remove_call(logic, establish_job_id);
    handle_establish_timeout(this, 0, 0);
  }
  if(pump_recv_timer_id)
  {
    llarp_logic_remove_call(logic, pump_recv_timer_id);
  }
  if(pump_send_timer_id)
  {
    llarp_logic_remove_call(logic, pump_send_timer_id);
  }
}

void
llarp_link_session::PumpCryptoOutbound()
{
  working = true;
  llarp_threadpool_queue_job(serv->worker, {this, &handle_crypto_outbound});
}

// void llarp_link_session::PumpCodelInbound()
// {
//   pump_recv_timer_id =
//       llarp_logic_call_later(logic,
//                              {decryptedFrames.nextTickInterval, this,
//                               &handle_inbound_codel_delayed});
// }

void
llarp_link_session::EnterState(State st)
{
  llarp::Debug("EnterState - entering state: ", st,
               state == eLIMSent ? "eLIMSent" : "",
               state == eSessionStartSent ? "eSessionStartSent" : "");
  frame.alive();
  state = st;
  if(state == eSessionStartSent || state == eIntroAckSent)
  {
    // llarp::Info("EnterState - ",  state==eLIMSent?"eLIMSent":"",
    // state==eSessionStartSent?"eSessionStartSent":"");
    // PumpCodelInbound();
    // PumpCodelOutbound();
    PumpCryptoOutbound();
    // StartInboundCodel();
  }
}

void
llarp_link_session::on_intro(const void *buf, size_t sz)
{
  llarp::Debug("session onintro");
  if(sz >= sizeof(workbuf))
  {
    // too big?
    llarp::Error("intro too big");
    delete this;
    return;
  }
  if(serv->has_intro_from(addr))
  {
    llarp::Error("duplicate intro from ", addr);
    delete this;
    return;
  }
  serv->put_intro_from(this);
  // copy so we own it
  memcpy(workbuf, buf, sz);
  intro.buf = workbuf;
  intro.sz  = sz;
  // give secret key
  intro.secretkey = eph_seckey;
  // and nonce
  intro.nonce = intro.buf + 32;
  intro.user  = this;
  // set call back hook
  intro.hook = &handle_verify_intro;
  // put remote pubkey into this buffer
  intro.remote_pubkey = remote;

  // call
  EnterState(eIntroRecv);
  working = true;
  iwp_call_async_verify_intro(iwp, &intro);
}

void
llarp_link_session::on_intro_ack(const void *buf, size_t sz)
{
  if(sz >= sizeof(workbuf))
  {
    // too big?
    llarp::Error("introack too big");
    serv->RemoveSessionByAddr(addr);
    return;
  }
  serv->put_intro_from(this);
  // copy buffer so we own it
  memcpy(workbuf, buf, sz);
  // set intro ack parameters
  introack.buf           = workbuf;
  introack.sz            = sz;
  introack.nonce         = workbuf + 32;
  introack.remote_pubkey = remote;
  introack.token         = token;
  introack.secretkey     = eph_seckey;
  introack.user          = this;
  introack.hook          = &handle_verify_introack;
  // async verify
  working = true;
  iwp_call_async_verify_introack(iwp, &introack);
}

bool
llarp_link_session::is_invalidated() const
{
  return frame.flags_agree(eSessionInvalidated);
}

llarp_link *
llarp_link_session::get_parent()
{
  return serv;
}

bool
llarp_link_session::Tick(llarp_time_t now)
{
  if(timedout(now, SESSION_TIMEOUT))
  {
    // we are timed out
    // when we are done doing stuff with all of our frames from the crypto
    // workers we are done
    llarp::Warn("Tick - ", addr, " timed out with ", frames, " frames left");
    return !working;
  }
  if(is_invalidated())
  {
    // both sides agreeed to session invalidation
    // terminate our session when all of our frames from the crypto workers
    // are done
    llarp::Warn("Tick - ", addr, " invaldiated session with ", frames,
                " frames left");
    return !working;
  }
  if(state == eLIMSent || state == eEstablished)
  {
    if(now - lastKeepalive > KEEP_ALIVE_INTERVAL)
      send_keepalive(this);
  }

  // pump frame state
  if(state == eEstablished)
  {
    // llarp::Debug("Tick - pumping and retransmitting because we're
    // eEstablished");

    frame.retransmit(now);
    pump();
    PumpCryptoOutbound();
  }
  return !frame.process_inbound_queue();
}

void
llarp_link_session::EncryptOutboundFrames()
{
  std::queue< iwp_async_frame * > outq;
  outboundFrames.Process(outq);
  while(outq.size())
  {
    auto &front = outq.front();

    // if(iwp_encrypt_frame(&front))
    // q.push(front);
    if(iwp_encrypt_frame(front))
      handle_frame_encrypt(front);
    delete front;
    outq.pop();
  }
}

static void
handle_verify_session_start(iwp_async_session_start *s)
{
  llarp_link_session *self = static_cast< llarp_link_session * >(s->user);
  self->serv->remove_intro_from(self->addr);
  self->working = false;
  if(!s->buf)
  {
    // verify fail
    // TODO: remove session?
    llarp::Warn("session start verify failed from ", self->addr);
    self->serv->RemoveSessionByAddr(self->addr);
    return;
  }
  self->send_LIM();
}

static void
handle_introack_generated(iwp_async_introack *i)
{
  llarp_link_session *link = static_cast< llarp_link_session * >(i->user);
  if(i->buf && link->serv->has_intro_from(link->addr))
  {
    // track it with the server here
    if(link->serv->has_session_to(link->addr))
    {
      // duplicate session
      llarp::Warn("duplicate session to ", link->addr);
      return;
    }
    link->frame.alive();
    link->EnterState(llarp_link_session::State::eIntroAckSent);
    link->serv->put_session(link->addr, link);
    llarp::Debug("send introack to ", link->addr, " via ", link->serv->addr);
    llarp_ev_udp_sendto(link->udp, link->addr, i->buf, i->sz);
  }
  else
  {
    // failed to generate?
    llarp::Warn("failed to generate introack");
  }
}

static void
handle_generated_intro(iwp_async_intro *i)
{
  llarp_link_session *link = static_cast< llarp_link_session * >(i->user);
  link->working            = false;
  if(i->buf)
  {
    llarp::Info("send intro to ", link->addr);
    if(llarp_ev_udp_sendto(link->udp, link->addr, i->buf, i->sz) == -1)
    {
      llarp::Warn("send intro failed");
      return;
    }
    link->EnterState(llarp_link_session::eIntroSent);
  }
  else
  {
    llarp::Warn("failed to generate intro");
  }
}

void
llarp_link_session::introduce(uint8_t *pub)
{
  llarp::Debug("session introduce");
  memcpy(remote, pub, PUBKEYSIZE);
  intro.buf   = workbuf;
  size_t w0sz = (rand() % MAX_PAD);
  intro.sz    = (32 * 3) + w0sz;
  // randomize w0
  if(w0sz)
  {
    crypto->randbytes(intro.buf + (32 * 3), w0sz);
  }

  intro.nonce     = intro.buf + 32;
  intro.secretkey = eph_seckey;
  // copy in pubkey
  intro.remote_pubkey = remote;
  // randomize nonce
  crypto->randbytes(intro.nonce, 32);
  // async generate intro packet
  intro.user = this;
  intro.hook = &handle_generated_intro;
  working    = true;
  llarp::Info("try introduce to transport adddress ", llarp::RouterID(remote));
  iwp_call_async_gen_intro(iwp, &intro);
  // start introduce timer
  establish_job_id =
      llarp_logic_call_later(logic, {5000, this, &handle_establish_timeout});
}

static void
handle_frame_decrypt(iwp_async_frame *frame)
{
  llarp_link_session *self = static_cast< llarp_link_session * >(frame->user);
  llarp::Debug("rx ", frame->sz);
  if(frame->success)
  {
    if(self->frame.process(frame->buf + 64, frame->sz - 64))
    {
      self->frame.alive();
      self->pump();
    }
    else
      llarp::Error("invalid frame from ", self->addr);
  }
  else
    llarp::Error("decrypt frame fail from ", self->addr);
}

void
llarp_link_session::decrypt_frame(const void *buf, size_t sz)
{
  if(sz > 64)
  {
    // auto frame = alloc_frame(inboundFrames, buf, sz);
    // inboundFrames.Put(frame);
    auto f = alloc_frame(buf, sz);
    /*
    if(iwp_decrypt_frame(f))
    {
      decryptedFrames.Put(f);
      if(state == eEstablished)
      {
        if(pump_recv_timer_id == 0)
          PumpCodelInbound();
      }
      else
        ManualPumpInboundCodel();
    }
    else
      llarp::Warn("decrypt frame fail");
   */
    f->hook = &handle_frame_decrypt;
    iwp_call_async_frame_decrypt(iwp, f);
  }
  else
    llarp::Warn("short packet of ", sz, " bytes");
}

void
llarp_link_session::session_start()
{
  llarp::Info("session gen start");
  size_t w2sz = rand() % MAX_PAD;
  start.buf   = workbuf;
  start.sz    = w2sz + (32 * 3);
  start.nonce = workbuf + 32;
  crypto->randbytes(start.nonce, 32);
  start.token = token;
  memcpy(start.buf + 64, token, 32);
  if(w2sz)
    crypto->randbytes(start.buf + (32 * 3), w2sz);
  start.remote_pubkey = remote;
  start.secretkey     = eph_seckey;
  start.sessionkey    = sessionkey;
  start.user          = this;
  start.hook          = &handle_generated_session_start;
  working             = true;
  iwp_call_async_gen_session_start(iwp, &start);
}

void
llarp_link_session::on_session_start(const void *buf, size_t sz)
{
  llarp::Info("session start");
  if(sz > sizeof(workbuf))
  {
    llarp::Debug("session start too big");
    return;
  }
  // own the buffer
  memcpy(workbuf, buf, sz);
  // verify session start
  start.buf           = workbuf;
  start.sz            = sz;
  start.nonce         = workbuf + 32;
  start.token         = token;
  start.remote_pubkey = remote;
  start.secretkey     = eph_seckey;
  start.sessionkey    = sessionkey;
  start.user          = this;
  start.hook          = &handle_verify_session_start;
  working             = true;
  iwp_call_async_verify_session_start(iwp, &start);
}

void
llarp_link_session::intro_ack()
{
  if(serv->has_session_to(addr))
  {
    llarp::Warn("won't ack intro for duplicate session from ", addr);
    return;
  }
  llarp::Debug("session introack");
  uint16_t w1sz = rand() % MAX_PAD;
  introack.buf  = workbuf;
  introack.sz   = (32 * 3) + w1sz;
  // randomize padding
  if(w1sz)
    crypto->randbytes(introack.buf + (32 * 3), w1sz);

  // randomize nonce
  introack.nonce = introack.buf + 32;
  crypto->randbytes(introack.nonce, 32);
  // token
  introack.token = token;

  // keys
  introack.remote_pubkey = remote;
  introack.secretkey     = eph_seckey;

  // call
  introack.user = this;
  introack.hook = &handle_introack_generated;
  working       = true;
  iwp_call_async_gen_introack(iwp, &introack);
}

// this is called from net thread
void
llarp_link_session::recv(const void *buf, size_t sz)
{
  // llarp::Debug("session recv", state);

  // frame_header hdr((byte_t *)buf);
  // llarp::Debug("recv - message header type ", (int)hdr.msgtype());

  now = llarp_time_now_ms();
  switch(state)
  {
    case eInitial:
    case eIntroRecv:
      // got intro
      llarp::Debug("session recv - intro");
      on_intro(buf, sz);
      break;
    case eIntroSent:
      // got intro ack
      llarp::Debug("session recv - introack");
      on_intro_ack(buf, sz);
      break;
    case eIntroAckSent:
      // probably a session start
      llarp::Debug("session recv - sessionstart");
      on_session_start(buf, sz);
      break;

    case eSessionStartSent:
    case eLIMSent:
    case eEstablished:
      // session is started
      llarp::Debug("session recv - ",
                   state == eSessionStartSent ? "startsent" : "",
                   state == eLIMSent ? "limset" : "",
                   state == eEstablished ? "established" : "");

      decrypt_frame(buf, sz);
      break;
    default:
      llarp::Error("session recv - invalid state");
      // invalid state?
      break;
  }
}

// TODO: fix orphan
iwp_async_frame *
llarp_link_session::alloc_frame(const void *buf, size_t sz)
{
  // TODO don't hard code 1500
  if(sz > 1500)
  {
    llarp::Warn("alloc frame - frame too big, >1500");
    return nullptr;
  }

  iwp_async_frame *frame = new iwp_async_frame;
  if(buf)
    memcpy(frame->buf, buf, sz);
  frame->iwp        = iwp;
  frame->sz         = sz;
  frame->user       = this;
  frame->sessionkey = sessionkey;
  /// TODO: this could be rather slow
  // frame->created = now;
  // llarp::Info("alloc_frame putting into q");
  // q.Put(frame);
  return frame;
}

void
llarp_link_session::encrypt_frame_async_send(const void *buf, size_t sz)
{
  // 64 bytes frame overhead for nonce and hmac
  iwp_async_frame *frame = alloc_frame(nullptr, sz + 64);
  memcpy(frame->buf + 64, buf, sz);
  // maybe add upto 128 random bytes to the packet
  auto padding = rand() % MAX_PAD;
  if(padding)
    crypto->randbytes(frame->buf + 64 + sz, padding);
  frame->sz += padding;
  // frame is modified, so now we can push it to queue
  outboundFrames.Put(frame);
}

void
llarp_link_session::pump()
{
  // llarp::Info("session pump");
  // TODO: in codel the timestamp may cause excssive drop when all the
  // packets have a similar timestamp
  now = llarp_time_now_ms();
  llarp_buffer_t buf;
  while(frame.next_frame(&buf))
  {
    encrypt_frame_async_send(buf.base, buf.sz);
    frame.pop_next_frame();
  }
}