[oweals/gnunet.git] / src / transport / gnunet-communicator-tcp.c

/*
     This file is part of GNUnet
     Copyright (C) 2010-2014, 2018, 2019 GNUnet e.V.

     GNUnet is free software: you can redistribute it and/or modify it
     under the terms of the GNU Affero General Public License as published
     by the Free Software Foundation, either version 3 of the License,
     or (at your option) any later version.

     GNUnet is distributed in the hope that it will be useful, but
     WITHOUT ANY WARRANTY; without even the implied warranty of
     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
     Affero General Public License for more details.

     You should have received a copy of the GNU Affero General Public License
     along with this program.  If not, see <http://www.gnu.org/licenses/>.

     SPDX-License-Identifier: AGPL3.0-or-later
*/

/**
 * @file transport/gnunet-communicator-tcp.c
 * @brief Transport plugin using TCP.
 * @author Christian Grothoff
 *
 * TODO:
 * - support DNS names in BINDTO option (#5528)
 * - support NAT connection reversal method (#5529)
 * - support other TCP-specific NAT traversal methods (#5531)
 * - add replay protection support to the protocol by
 *   adding a nonce in the KX and requiring (!) a
 *   nounce ACK to be send within the first X bytes of
 *   data (#5530)
 */
#include "platform.h"
#include "gnunet_util_lib.h"
#include "gnunet_protocols.h"
#include "gnunet_signatures.h"
#include "gnunet_constants.h"
#include "gnunet_nt_lib.h"
#include "gnunet_nat_service.h"
#include "gnunet_statistics_service.h"
#include "gnunet_transport_communication_service.h"

/**
 * How long do we believe our addresses to remain up (before
 * the other peer should revalidate).
 */
#define ADDRESS_VALIDITY_PERIOD \
  GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_HOURS, 4)

/**
 * How many messages do we keep at most in the queue to the
 * transport service before we start to drop (default,
 * can be changed via the configuration file).
 * Should be _below_ the level of the communicator API, as
 * otherwise we may read messages just to have them dropped
 * by the communicator API.
 */
#define DEFAULT_MAX_QUEUE_LENGTH 8

/**
 * Size of our IO buffers for ciphertext data. Must be at
 * least UINT_MAX + sizeof (struct TCPBox).
 */
#define BUF_SIZE (2 * 64 * 1024 + sizeof (struct TCPBox))

/**
 * How often do we rekey based on time (at least)
 */
#define REKEY_TIME_INTERVAL GNUNET_TIME_UNIT_DAYS

/**
 * How long do we wait until we must have received the initial KX?
 */
#define PROTO_QUEUE_TIMEOUT GNUNET_TIME_UNIT_MINUTES

/**
 * How often do we rekey based on number of bytes transmitted?
 * (additionally randomized).
 */
#define REKEY_MAX_BYTES (1024LLU * 1024 * 1024 * 4LLU)

/**
 * Size of the initial key exchange message sent first in both
 * directions.
 */
#define INITIAL_KX_SIZE                           \
  (sizeof (struct GNUNET_CRYPTO_EcdhePublicKey) + \
   sizeof (struct TCPConfirmation))


/**
 * Address prefix used by the communicator.
 */
#define COMMUNICATOR_ADDRESS_PREFIX "tcp"

/**
 * Configuration section used by the communicator.
 */
#define COMMUNICATOR_CONFIG_SECTION "communicator-tcp"

GNUNET_NETWORK_STRUCT_BEGIN


/**
 * Signature we use to verify that the ephemeral key was really chosen by
 * the specified sender.
 */
struct TcpHandshakeSignature
{
  /**
   * Purpose must be #GNUNET_SIGNATURE_COMMUNICATOR_TCP_HANDSHAKE
   */
  struct GNUNET_CRYPTO_EccSignaturePurpose purpose;

  /**
   * Identity of the inititor of the TCP connection (TCP client).
   */
  struct GNUNET_PeerIdentity sender;

  /**
   * Presumed identity of the target of the TCP connection (TCP server)
   */
  struct GNUNET_PeerIdentity receiver;

  /**
   * Ephemeral key used by the @e sender.
   */
  struct GNUNET_CRYPTO_EcdhePublicKey ephemeral;

  /**
   * Monotonic time of @e sender, to possibly help detect replay attacks
   * (if receiver persists times by sender).
   */
  struct GNUNET_TIME_AbsoluteNBO monotonic_time;
};


/**
 * Encrypted continuation of TCP initial handshake.
 */
struct TCPConfirmation
{
  /**
   * Sender's identity
   */
  struct GNUNET_PeerIdentity sender;

  /**
   * Sender's signature of type #GNUNET_SIGNATURE_COMMUNICATOR_TCP_HANDSHAKE
   */
  struct GNUNET_CRYPTO_EddsaSignature sender_sig;

  /**
   * Monotonic time of @e sender, to possibly help detect replay attacks
   * (if receiver persists times by sender).
   */
  struct GNUNET_TIME_AbsoluteNBO monotonic_time;
};


/**
 * TCP message box.  Always sent encrypted!
 */
struct TCPBox
{

  /**
   * Type is #GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_BOX.  Warning: the
   * header size EXCLUDES the size of the `struct TCPBox`. We usually
   * never do this, but here the payload may truly be 64k *after* the
   * TCPBox (as we have no MTU)!!
   */
  struct GNUNET_MessageHeader header;

  /**
   * HMAC for the following encrypted message.  Yes, we MUST use
   * mac-then-encrypt here, as we want to hide the message sizes on
   * the wire (zero plaintext design!).  Using CTR mode, padding oracle
   * attacks do not apply.  Besides, due to the use of ephemeral keys
   * (hopefully with effective replay protection from monotonic time!)
   * the attacker is limited in using the oracle.
   */
  struct GNUNET_ShortHashCode hmac;

  /* followed by as may bytes of payload as indicated in @e header,
     excluding the TCPBox itself! */
};


/**
 * TCP rekey message box.  Always sent encrypted!  Data after
 * this message will use the new key.
 */
struct TCPRekey
{

  /**
   * Type is #GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_REKEY.
   */
  struct GNUNET_MessageHeader header;

  /**
   * HMAC for the following encrypted message.  Yes, we MUST use
   * mac-then-encrypt here, as we want to hide the message sizes on
   * the wire (zero plaintext design!).  Using CTR mode padding oracle
   * attacks do not apply.  Besides, due to the use of ephemeral keys
   * (hopefully with effective replay protection from monotonic time!)
   * the attacker is limited in using the oracle.
   */
  struct GNUNET_ShortHashCode hmac;

  /**
   * New ephemeral key.
   */
  struct GNUNET_CRYPTO_EcdhePublicKey ephemeral;

  /**
   * Sender's signature of type #GNUNET_SIGNATURE_COMMUNICATOR_TCP_REKEY
   */
  struct GNUNET_CRYPTO_EddsaSignature sender_sig;

  /**
   * Monotonic time of @e sender, to possibly help detect replay attacks
   * (if receiver persists times by sender).
   */
  struct GNUNET_TIME_AbsoluteNBO monotonic_time;
};


/**
 * TCP finish. Sender asks for the connection to be closed.
 * Needed/useful in case we drop RST/FIN packets on the GNUnet
 * port due to the possibility of malicious RST/FIN injection.
 */
struct TCPFinish
{

  /**
   * Type is #GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_FINISH.
   */
  struct GNUNET_MessageHeader header;

  /**
   * HMAC for the following encrypted message.  Yes, we MUST use
   * mac-then-encrypt here, as we want to hide the message sizes on
   * the wire (zero plaintext design!).  Using CTR mode padding oracle
   * attacks do not apply.  Besides, due to the use of ephemeral keys
   * (hopefully with effective replay protection from monotonic time!)
   * the attacker is limited in using the oracle.
   */
  struct GNUNET_ShortHashCode hmac;
};


GNUNET_NETWORK_STRUCT_END


/**
 * Handle for a queue.
 */
struct Queue
{

  /**
   * To whom are we talking to.
   */
  struct GNUNET_PeerIdentity target;

  /**
   * socket that we transmit all data with on this queue
   */
  struct GNUNET_NETWORK_Handle *sock;

  /**
   * cipher for decryption of incoming data.
   */
  gcry_cipher_hd_t in_cipher;

  /**
   * cipher for encryption of outgoing data.
   */
  gcry_cipher_hd_t out_cipher;

  /**
   * Shared secret for HMAC verification on incoming data.
   */
  struct GNUNET_HashCode in_hmac;

  /**
   * Shared secret for HMAC generation on outgoing data, ratcheted after
   * each operation.
   */
  struct GNUNET_HashCode out_hmac;

  /**
   * Our ephemeral key. Stored here temporarily during rekeying / key
   * generation.
   */
  struct GNUNET_CRYPTO_EcdhePrivateKey ephemeral;

  /**
   * ID of read task for this connection.
   */
  struct GNUNET_SCHEDULER_Task *read_task;

  /**
   * ID of write task for this connection.
   */
  struct GNUNET_SCHEDULER_Task *write_task;

  /**
   * Address of the other peer.
   */
  struct sockaddr *address;

  /**
   * How many more bytes may we sent with the current @e out_cipher
   * before we should rekey?
   */
  uint64_t rekey_left_bytes;

  /**
   * Until what time may we sent with the current @e out_cipher
   * before we should rekey?
   */
  struct GNUNET_TIME_Absolute rekey_time;

  /**
   * Length of the address.
   */
  socklen_t address_len;

  /**
   * Message queue we are providing for the #ch.
   */
  struct GNUNET_MQ_Handle *mq;

  /**
   * handle for this queue with the #ch.
   */
  struct GNUNET_TRANSPORT_QueueHandle *qh;

  /**
   * Number of bytes we currently have in our write queue.
   */
  unsigned long long bytes_in_queue;

  /**
   * Buffer for reading ciphertext from network into.
   */
  char cread_buf[BUF_SIZE];

  /**
   * buffer for writing ciphertext to network.
   */
  char cwrite_buf[BUF_SIZE];

  /**
   * Plaintext buffer for decrypted plaintext.
   */
  char pread_buf[UINT16_MAX + 1 + sizeof (struct TCPBox)];

  /**
   * Plaintext buffer for messages to be encrypted.
   */
  char pwrite_buf[UINT16_MAX + 1 + sizeof (struct TCPBox)];

  /**
   * At which offset in the ciphertext read buffer should we
   * append more ciphertext for transmission next?
   */
  size_t cread_off;

  /**
   * At which offset in the ciphertext write buffer should we
   * append more ciphertext from reading next?
   */
  size_t cwrite_off;

  /**
   * At which offset in the plaintext input buffer should we
   * append more plaintext from decryption next?
   */
  size_t pread_off;

  /**
   * At which offset in the plaintext output buffer should we
   * append more plaintext for encryption next?
   */
  size_t pwrite_off;

  /**
   * Timeout for this queue.
   */
  struct GNUNET_TIME_Absolute timeout;

  /**
   * How may messages did we pass from this queue to CORE for which we
   * have yet to receive an acknoweldgement that CORE is done with
   * them? If "large" (or even just non-zero), we should throttle
   * reading to provide flow control.  See also #DEFAULT_MAX_QUEUE_LENGTH
   * and #max_queue_length.
   */
  unsigned int backpressure;

  /**
   * Which network type does this queue use?
   */
  enum GNUNET_NetworkType nt;

  /**
   * Is MQ awaiting a #GNUNET_MQ_impl_send_continue() call?
   */
  int mq_awaits_continue;

  /**
   * Did we enqueue a finish message and are closing down the queue?
   */
  int finishing;

  /**
   * Did we technically destroy this queue, but kept the allocation
   * around because of @e backpressure not being zero yet? Used
   * simply to delay the final #GNUNET_free() operation until
   * #core_read_finished_cb() has been called.
   */
  int destroyed;

  /**
   * #GNUNET_YES if we just rekeyed and must thus possibly
   * re-decrypt ciphertext.
   */
  int rekeyed;
};


/**
 * Handle for an incoming connection where we do not yet have enough
 * information to setup a full queue.
 */
struct ProtoQueue
{

  /**
   * Kept in a DLL.
   */
  struct ProtoQueue *next;

  /**
   * Kept in a DLL.
   */
  struct ProtoQueue *prev;

  /**
   * socket that we transmit all data with on this queue
   */
  struct GNUNET_NETWORK_Handle *sock;

  /**
   * ID of read task for this connection.
   */
  struct GNUNET_SCHEDULER_Task *read_task;

  /**
   * Address of the other peer.
   */
  struct sockaddr *address;

  /**
   * Length of the address.
   */
  socklen_t address_len;

  /**
   * Timeout for this protoqueue.
   */
  struct GNUNET_TIME_Absolute timeout;

  /**
   * Buffer for reading all the information we need to upgrade from
   * protoqueue to queue.
   */
  char ibuf[INITIAL_KX_SIZE];

  /**
   * Current offset for reading into @e ibuf.
   */
  size_t ibuf_off;
};


/**
 * ID of listen task
 */
static struct GNUNET_SCHEDULER_Task *listen_task;

/**
 * Maximum queue length before we stop reading towards the transport service.
 */
static unsigned long long max_queue_length;

/**
 * For logging statistics.
 */
static struct GNUNET_STATISTICS_Handle *stats;

/**
 * Our environment.
 */
static struct GNUNET_TRANSPORT_CommunicatorHandle *ch;

/**
 * Queues (map from peer identity to `struct Queue`)
 */
static struct GNUNET_CONTAINER_MultiPeerMap *queue_map;

/**
 * Listen socket.
 */
static struct GNUNET_NETWORK_Handle *listen_sock;

/**
 * Our public key.
 */
static struct GNUNET_PeerIdentity my_identity;

/**
 * Our private key.
 */
static struct GNUNET_CRYPTO_EddsaPrivateKey *my_private_key;

/**
 * Our configuration.
 */
static const struct GNUNET_CONFIGURATION_Handle *cfg;

/**
 * Network scanner to determine network types.
 */
static struct GNUNET_NT_InterfaceScanner *is;

/**
 * Connection to NAT service.
 */
static struct GNUNET_NAT_Handle *nat;

/**
 * Protoqueues DLL head.
 */
static struct ProtoQueue *proto_head;

/**
 * Protoqueues DLL tail.
 */
static struct ProtoQueue *proto_tail;


/**
 * We have been notified that our listen socket has something to
 * read. Do the read and reschedule this function to be called again
 * once more is available.
 *
 * @param cls NULL
 */
static void
listen_cb (void *cls);


/**
 * Functions with this signature are called whenever we need
 * to close a queue due to a disconnect or failure to
 * establish a connection.
 *
 * @param queue queue to close down
 */
static void
queue_destroy (struct Queue *queue)
{
  struct GNUNET_MQ_Handle *mq;

  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Disconnecting queue for peer `%s'\n",
              GNUNET_i2s (&queue->target));
  if (NULL != (mq = queue->mq))
  {
    queue->mq = NULL;
    GNUNET_MQ_destroy (mq);
  }
  if (NULL != queue->qh)
  {
    GNUNET_TRANSPORT_communicator_mq_del (queue->qh);
    queue->qh = NULL;
  }
  GNUNET_assert (
    GNUNET_YES ==
    GNUNET_CONTAINER_multipeermap_remove (queue_map, &queue->target, queue));
  GNUNET_STATISTICS_set (stats,
                         "# queues active",
                         GNUNET_CONTAINER_multipeermap_size (queue_map),
                         GNUNET_NO);
  if (NULL != queue->read_task)
  {
    GNUNET_SCHEDULER_cancel (queue->read_task);
    queue->read_task = NULL;
  }
  if (NULL != queue->write_task)
  {
    GNUNET_SCHEDULER_cancel (queue->write_task);
    queue->write_task = NULL;
  }
  GNUNET_NETWORK_socket_close (queue->sock);
  gcry_cipher_close (queue->in_cipher);
  gcry_cipher_close (queue->out_cipher);
  GNUNET_free (queue->address);
  if (0 != queue->backpressure)
    queue->destroyed = GNUNET_YES;
  else
    GNUNET_free (queue);
  if (NULL == listen_task)
    listen_task = GNUNET_SCHEDULER_add_read_net (GNUNET_TIME_UNIT_FOREVER_REL,
                                                 listen_sock,
                                                 &listen_cb,
                                                 NULL);
}


/**
 * Compute @a mac over @a buf, and ratched the @a hmac_secret.
 *
 * @param[in,out] hmac_secret secret for HMAC calculation
 * @param buf buffer to MAC
 * @param buf_size number of bytes in @a buf
 * @param smac[out] where to write the HMAC
 */
static void
calculate_hmac (struct GNUNET_HashCode *hmac_secret,
                const void *buf,
                size_t buf_size,
                struct GNUNET_ShortHashCode *smac)
{
  struct GNUNET_HashCode mac;

  GNUNET_CRYPTO_hmac_raw (hmac_secret,
                          sizeof (struct GNUNET_HashCode),
                          buf,
                          buf_size,
                          &mac);
  /* truncate to `struct GNUNET_ShortHashCode` */
  memcpy (smac, &mac, sizeof (struct GNUNET_ShortHashCode));
  /* ratchet hmac key */
  GNUNET_CRYPTO_hash (hmac_secret,
                      sizeof (struct GNUNET_HashCode),
                      hmac_secret);
}


/**
 * Append a 'finish' message to the outgoing transmission. Once the
 * finish has been transmitted, destroy the queue.
 *
 * @param queue queue to shut down nicely
 */
static void
queue_finish (struct Queue *queue)
{
  struct TCPFinish fin;

  memset (&fin, 0, sizeof (fin));
  fin.header.size = htons (sizeof (fin));
  fin.header.type = htons (GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_FINISH);
  calculate_hmac (&queue->out_hmac, &fin, sizeof (fin), &fin.hmac);
  /* if there is any message left in pwrite_buf, we
     overwrite it (possibly dropping the last message
     from CORE hard here) */
  memcpy (queue->pwrite_buf, &fin, sizeof (fin));
  queue->pwrite_off = sizeof (fin);
  /* This flag will ensure that #queue_write() no longer
     notifies CORE about the possibility of sending
     more data, and that #queue_write() will call
     #queue_destroy() once the @c fin was fully written. */
  queue->finishing = GNUNET_YES;
}


/**
 * Increment queue timeout due to activity.  We do not immediately
 * notify the monitor here as that might generate excessive
 * signalling.
 *
 * @param queue queue for which the timeout should be rescheduled
 */
static void
reschedule_queue_timeout (struct Queue *queue)
{
  queue->timeout =
    GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
}


/**
 * Queue read task. If we hit the timeout, disconnect it
 *
 * @param cls the `struct Queue *` to disconnect
 */
static void
queue_read (void *cls);


/**
 * Core tells us it is done processing a message that transport
 * received on a queue with status @a success.
 *
 * @param cls a `struct Queue *` where the message originally came from
 * @param success #GNUNET_OK on success
 */
static void
core_read_finished_cb (void *cls, int success)
{
  struct Queue *queue = cls;

  if (GNUNET_OK != success)
    GNUNET_STATISTICS_update (stats,
                              "# messages lost in communicator API towards CORE",
                              1,
                              GNUNET_NO);
  queue->backpressure--;
  /* handle deferred queue destruction */
  if ((queue->destroyed) && (0 == queue->backpressure))
  {
    GNUNET_free (queue);
    return;
  }
  reschedule_queue_timeout (queue);
  /* possibly unchoke reading, now that CORE made progress */
  if (NULL == queue->read_task)
    queue->read_task =
      GNUNET_SCHEDULER_add_read_net (GNUNET_TIME_absolute_get_remaining (
                                       queue->timeout),
                                     queue->sock,
                                     &queue_read,
                                     queue);
}


/**
 * We received @a plaintext_len bytes of @a plaintext on @a queue.
 * Pass it on to CORE.  If transmission is actually happening,
 * increase backpressure counter.
 *
 * @param queue the queue that received the plaintext
 * @param plaintext the plaintext that was received
 * @param plaintext_len number of bytes of plaintext received
 */
static void
pass_plaintext_to_core (struct Queue *queue,
                        const void *plaintext,
                        size_t plaintext_len)
{
  const struct GNUNET_MessageHeader *hdr = plaintext;
  int ret;

  if (ntohs (hdr->size) != plaintext_len)
  {
    /* NOTE: If we ever allow multiple CORE messages in one
       BOX, this will have to change! */
    GNUNET_break (0);
    return;
  }
  ret = GNUNET_TRANSPORT_communicator_receive (ch,
                                               &queue->target,
                                               hdr,
                                               ADDRESS_VALIDITY_PERIOD,
                                               &core_read_finished_cb,
                                               queue);
  if (GNUNET_OK == ret)
    queue->backpressure++;
  GNUNET_break (GNUNET_NO != ret); /* backpressure not working!? */
  if (GNUNET_SYSERR == ret)
    GNUNET_STATISTICS_update (stats,
                              "# bytes lost due to CORE not running",
                              plaintext_len,
                              GNUNET_NO);
}


/**
 * Setup @a cipher based on shared secret @a dh and decrypting
 * peer @a pid.
 *
 * @param dh shared secret
 * @param pid decrypting peer's identity
 * @param cipher[out] cipher to initialize
 * @param hmac_key[out] HMAC key to initialize
 */
static void
setup_cipher (const struct GNUNET_HashCode *dh,
              const struct GNUNET_PeerIdentity *pid,
              gcry_cipher_hd_t *cipher,
              struct GNUNET_HashCode *hmac_key)
{
  char key[256 / 8];
  char ctr[128 / 8];

  gcry_cipher_open (cipher,
                    GCRY_CIPHER_AES256 /* low level: go for speed */,
                    GCRY_CIPHER_MODE_CTR,
                    0 /* flags */);
  GNUNET_assert (GNUNET_YES == GNUNET_CRYPTO_kdf (key,
                                                  sizeof (key),
                                                  "TCP-key",
                                                  strlen ("TCP-key"),
                                                  dh,
                                                  sizeof (*dh),
                                                  pid,
                                                  sizeof (*pid),
                                                  NULL,
                                                  0));
  gcry_cipher_setkey (*cipher, key, sizeof (key));
  GNUNET_assert (GNUNET_YES == GNUNET_CRYPTO_kdf (ctr,
                                                  sizeof (ctr),
                                                  "TCP-ctr",
                                                  strlen ("TCP-ctr"),
                                                  dh,
                                                  sizeof (*dh),
                                                  pid,
                                                  sizeof (*pid),
                                                  NULL,
                                                  0));
  gcry_cipher_setctr (*cipher, ctr, sizeof (ctr));
  GNUNET_assert (GNUNET_YES ==
                 GNUNET_CRYPTO_kdf (hmac_key,
                                    sizeof (struct GNUNET_HashCode),
                                    "TCP-hmac",
                                    strlen ("TCP-hmac"),
                                    dh,
                                    sizeof (*dh),
                                    pid,
                                    sizeof (*pid),
                                    NULL,
                                    0));
}


/**
 * Setup cipher of @a queue for decryption.
 *
 * @param ephemeral ephemeral key we received from the other peer
 * @param queue[in,out] queue to initialize decryption cipher for
 */
static void
setup_in_cipher (const struct GNUNET_CRYPTO_EcdhePublicKey *ephemeral,
                 struct Queue *queue)
{
  struct GNUNET_HashCode dh;

  GNUNET_CRYPTO_eddsa_ecdh (my_private_key, ephemeral, &dh);
  setup_cipher (&dh, &my_identity, &queue->in_cipher, &queue->in_hmac);
}


/**
 * Handle @a rekey message on @a queue. The message was already
 * HMAC'ed, but we should additionally still check the signature.
 * Then we need to stop the old cipher and start afresh.
 *
 * @param queue the queue @a rekey was received on
 * @param rekey the rekey message
 */
static void
do_rekey (struct Queue *queue, const struct TCPRekey *rekey)
{
  struct TcpHandshakeSignature thp;

  thp.purpose.purpose = htonl (GNUNET_SIGNATURE_COMMUNICATOR_TCP_REKEY);
  thp.purpose.size = htonl (sizeof (thp));
  thp.sender = queue->target;
  thp.receiver = my_identity;
  thp.ephemeral = rekey->ephemeral;
  thp.monotonic_time = rekey->monotonic_time;
  /* FIXME: check monotonic time is monotonic... */
  if (GNUNET_OK !=
      GNUNET_CRYPTO_eddsa_verify (GNUNET_SIGNATURE_COMMUNICATOR_TCP_REKEY,
                                  &thp.purpose,
                                  &rekey->sender_sig,
                                  &queue->target.public_key))
  {
    GNUNET_break (0);
    queue_finish (queue);
    return;
  }
  gcry_cipher_close (queue->in_cipher);
  queue->rekeyed = GNUNET_YES;
  setup_in_cipher (&rekey->ephemeral, queue);
}


/**
 * Test if we have received a full message in plaintext.
 * If so, handle it.
 *
 * @param queue queue to process inbound plaintext for
 * @return number of bytes of plaintext handled, 0 for none
 */
static size_t
try_handle_plaintext (struct Queue *queue)
{
  const struct GNUNET_MessageHeader *hdr =
    (const struct GNUNET_MessageHeader *) queue->pread_buf;
  const struct TCPBox *box = (const struct TCPBox *) queue->pread_buf;
  const struct TCPRekey *rekey = (const struct TCPRekey *) queue->pread_buf;
  const struct TCPFinish *fin = (const struct TCPFinish *) queue->pread_buf;
  struct TCPRekey rekeyz;
  struct TCPFinish finz;
  struct GNUNET_ShortHashCode tmac;
  uint16_t type;
  size_t size = 0; /* make compiler happy */

  if (sizeof (*hdr) > queue->pread_off)
    return 0; /* not even a header */
  type = ntohs (hdr->type);
  switch (type)
  {
  case GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_BOX:
    /* Special case: header size excludes box itself! */
    if (ntohs (hdr->size) + sizeof (struct TCPBox) > queue->pread_off)
      return 0;
    calculate_hmac (&queue->in_hmac, &box[1], ntohs (hdr->size), &tmac);
    if (0 != memcmp (&tmac, &box->hmac, sizeof (tmac)))
    {
      GNUNET_break_op (0);
      queue_finish (queue);
      return 0;
    }
    pass_plaintext_to_core (queue, (const void *) &box[1], ntohs (hdr->size));
    size = ntohs (hdr->size) + sizeof (*box);
    break;
  case GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_REKEY:
    if (sizeof (*rekey) > queue->pread_off)
      return 0;
    if (ntohs (hdr->size) != sizeof (*rekey))
    {
      GNUNET_break_op (0);
      queue_finish (queue);
      return 0;
    }
    rekeyz = *rekey;
    memset (&rekeyz.hmac, 0, sizeof (rekeyz.hmac));
    calculate_hmac (&queue->in_hmac, &rekeyz, sizeof (rekeyz), &tmac);
    if (0 != memcmp (&tmac, &box->hmac, sizeof (tmac)))
    {
      GNUNET_break_op (0);
      queue_finish (queue);
      return 0;
    }
    do_rekey (queue, rekey);
    size = ntohs (hdr->size);
    break;
  case GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_FINISH:
    if (sizeof (*fin) > queue->pread_off)
      return 0;
    if (ntohs (hdr->size) != sizeof (*fin))
    {
      GNUNET_break_op (0);
      queue_finish (queue);
      return 0;
    }
    finz = *fin;
    memset (&finz.hmac, 0, sizeof (finz.hmac));
    calculate_hmac (&queue->in_hmac, &rekeyz, sizeof (rekeyz), &tmac);
    if (0 != memcmp (&tmac, &fin->hmac, sizeof (tmac)))
    {
      GNUNET_break_op (0);
      queue_finish (queue);
      return 0;
    }
    /* handle FINISH by destroying queue */
    queue_destroy (queue);
    break;
  default:
    GNUNET_break_op (0);
    queue_finish (queue);
    return 0;
  }
  GNUNET_assert (0 != size);
  return size;
}


/**
 * Queue read task. If we hit the timeout, disconnect it
 *
 * @param cls the `struct Queue *` to disconnect
 */
static void
queue_read (void *cls)
{
  struct Queue *queue = cls;
  struct GNUNET_TIME_Relative left;
  ssize_t rcvd;

  queue->read_task = NULL;
  rcvd = GNUNET_NETWORK_socket_recv (queue->sock,
                                     &queue->cread_buf[queue->cread_off],
                                     BUF_SIZE - queue->cread_off);
  if (-1 == rcvd)
  {
    if ((EAGAIN != errno) && (EINTR != errno))
    {
      GNUNET_log_strerror (GNUNET_ERROR_TYPE_DEBUG, "recv");
      queue_finish (queue);
      return;
    }
    /* try again */
    queue->read_task =
      GNUNET_SCHEDULER_add_read_net (left, queue->sock, &queue_read, queue);
    return;
  }
  if (0 != rcvd)
    reschedule_queue_timeout (queue);
  queue->cread_off += rcvd;
  while ((queue->pread_off < sizeof (queue->pread_buf)) &&
         (queue->cread_off > 0))
  {
    size_t max = GNUNET_MIN (sizeof (queue->pread_buf) - queue->pread_off,
                             queue->cread_off);
    size_t done;
    size_t total;

    GNUNET_assert (0 ==
                   gcry_cipher_decrypt (queue->in_cipher,
                                        &queue->pread_buf[queue->pread_off],
                                        max,
                                        queue->cread_buf,
                                        max));
    queue->pread_off += max;
    total = 0;
    while ((GNUNET_NO == queue->rekeyed) &&
           (0 != (done = try_handle_plaintext (queue))))
    {
      /* 'done' bytes of plaintext were used, shift buffer */
      GNUNET_assert (done <= queue->pread_off);
      /* NOTE: this memmove() could possibly sometimes be
         avoided if we pass 'total' into try_handle_plaintext()
         and use it at an offset into the buffer there! */
      memmove (queue->pread_buf,
               &queue->pread_buf[done],
               queue->pread_off - done);
      queue->pread_off -= done;
      total += done;
    }
    /* when we encounter a rekey message, the decryption above uses the
       wrong key for everything after the rekey; in that case, we have
       to re-do the decryption at 'total' instead of at 'max'. If there
       is no rekey and the last message is incomplete (max > total),
       it is safe to keep the decryption so we shift by 'max' */
    if (GNUNET_YES == queue->rekeyed)
    {
      max = total;
      queue->rekeyed = GNUNET_NO;
    }
    memmove (queue->cread_buf, &queue->cread_buf[max], queue->cread_off - max);
    queue->cread_off -= max;
  }

  if (BUF_SIZE == queue->cread_off)
    return; /* buffer full, suspend reading */
  left = GNUNET_TIME_absolute_get_remaining (queue->timeout);
  if (0 != left.rel_value_us)
  {
    if (max_queue_length < queue->backpressure)
    {
      /* continue reading */
      queue->read_task =
        GNUNET_SCHEDULER_add_read_net (left, queue->sock, &queue_read, queue);
    }
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Queue %p was idle for %s, disconnecting\n",
              queue,
              GNUNET_STRINGS_relative_time_to_string (
                GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
                GNUNET_YES));
  queue_finish (queue);
}


/**
 * Convert TCP bind specification to a `struct sockaddr *`
 *
 * @param bindto bind specification to convert
 * @param[out] sock_len set to the length of the address
 * @return converted bindto specification
 */
static struct sockaddr *
tcp_address_to_sockaddr (const char *bindto, socklen_t *sock_len)
{
  struct sockaddr *in;
  unsigned int port;
  char dummy[2];
  char *colon;
  char *cp;

  if (1 == SSCANF (bindto, "%u%1s", &port, dummy))
  {
    /* interpreting value as just a PORT number */
    if (port > UINT16_MAX)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                  "BINDTO specification `%s' invalid: value too large for port\n",
                  bindto);
      return NULL;
    }
    if ((GNUNET_NO == GNUNET_NETWORK_test_pf (PF_INET6)) ||
        (GNUNET_YES ==
         GNUNET_CONFIGURATION_get_value_yesno (cfg,
                                               COMMUNICATOR_CONFIG_SECTION,
                                               "DISABLE_V6")))
    {
      struct sockaddr_in *i4;

      i4 = GNUNET_malloc (sizeof (struct sockaddr_in));
      i4->sin_family = AF_INET;
      i4->sin_port = htons ((uint16_t) port);
      *sock_len = sizeof (struct sockaddr_in);
      in = (struct sockaddr *) i4;
    }
    else
    {
      struct sockaddr_in6 *i6;

      i6 = GNUNET_malloc (sizeof (struct sockaddr_in6));
      i6->sin6_family = AF_INET6;
      i6->sin6_port = htons ((uint16_t) port);
      *sock_len = sizeof (struct sockaddr_in6);
      in = (struct sockaddr *) i6;
    }
    return in;
  }
  cp = GNUNET_strdup (bindto);
  colon = strrchr (cp, ':');
  if (NULL != colon)
  {
    /* interpet value after colon as port */
    *colon = '\0';
    colon++;
    if (1 == SSCANF (colon, "%u%1s", &port, dummy))
    {
      /* interpreting value as just a PORT number */
      if (port > UINT16_MAX)
      {
        GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                    "BINDTO specification `%s' invalid: value too large for port\n",
                    bindto);
        GNUNET_free (cp);
        return NULL;
      }
    }
    else
    {
      GNUNET_log (
        GNUNET_ERROR_TYPE_ERROR,
        "BINDTO specification `%s' invalid: last ':' not followed by number\n",
        bindto);
      GNUNET_free (cp);
      return NULL;
    }
  }
  else
  {
    /* interpret missing port as 0, aka pick any free one */
    port = 0;
  }
  {
    /* try IPv4 */
    struct sockaddr_in v4;

    if (1 == inet_pton (AF_INET, cp, &v4))
    {
      v4.sin_port = htons ((uint16_t) port);
      in = GNUNET_memdup (&v4, sizeof (v4));
      *sock_len = sizeof (v4);
      GNUNET_free (cp);
      return in;
    }
  }
  {
    /* try IPv6 */
    struct sockaddr_in6 v6;
    const char *start;

    start = cp;
    if (('[' == *cp) && (']' == cp[strlen (cp) - 1]))
    {
      start++; /* skip over '[' */
      cp[strlen (cp) - 1] = '\0'; /* eat ']' */
    }
    if (1 == inet_pton (AF_INET6, start, &v6))
    {
      v6.sin6_port = htons ((uint16_t) port);
      in = GNUNET_memdup (&v6, sizeof (v6));
      *sock_len = sizeof (v6);
      GNUNET_free (cp);
      return in;
    }
  }
  /* #5528 FIXME (feature!): maybe also try getnameinfo()? */
  GNUNET_free (cp);
  return NULL;
}


/**
 * Setup cipher for outgoing data stream based on target and
 * our ephemeral private key.
 *
 * @param queue queue to setup outgoing (encryption) cipher for
 */
static void
setup_out_cipher (struct Queue *queue)
{
  struct GNUNET_HashCode dh;

  GNUNET_CRYPTO_ecdh_eddsa (&queue->ephemeral, &queue->target.public_key, &dh);
  /* we don't need the private key anymore, drop it! */
  memset (&queue->ephemeral, 0, sizeof (queue->ephemeral));
  setup_cipher (&dh, &queue->target, &queue->out_cipher, &queue->out_hmac);
  queue->rekey_time = GNUNET_TIME_relative_to_absolute (REKEY_TIME_INTERVAL);
  queue->rekey_left_bytes =
    GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK, REKEY_MAX_BYTES);
}


/**
 * Inject a `struct TCPRekey` message into the queue's plaintext
 * buffer.
 *
 * @param queue queue to perform rekeying on
 */
static void
inject_rekey (struct Queue *queue)
{
  struct TCPRekey rekey;
  struct TcpHandshakeSignature thp;

  GNUNET_assert (0 == queue->pwrite_off);
  memset (&rekey, 0, sizeof (rekey));
  GNUNET_assert (GNUNET_OK ==
                 GNUNET_CRYPTO_ecdhe_key_create2 (&queue->ephemeral));
  rekey.header.type = ntohs (GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_REKEY);
  rekey.header.size = ntohs (sizeof (rekey));
  GNUNET_CRYPTO_ecdhe_key_get_public (&queue->ephemeral, &rekey.ephemeral);
  rekey.monotonic_time =
    GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get_monotonic (cfg));
  thp.purpose.purpose = htonl (GNUNET_SIGNATURE_COMMUNICATOR_TCP_REKEY);
  thp.purpose.size = htonl (sizeof (thp));
  thp.sender = my_identity;
  thp.receiver = queue->target;
  thp.ephemeral = rekey.ephemeral;
  thp.monotonic_time = rekey.monotonic_time;
  GNUNET_assert (GNUNET_OK == GNUNET_CRYPTO_eddsa_sign (my_private_key,
                                                        &thp.purpose,
                                                        &rekey.sender_sig));
  calculate_hmac (&queue->out_hmac, &rekey, sizeof (rekey), &rekey.hmac);
  memcpy (queue->pwrite_buf, &rekey, sizeof (rekey));
  queue->pwrite_off = sizeof (rekey);
}


/**
 * We have been notified that our socket is ready to write.
 * Then reschedule this function to be called again once more is available.
 *
 * @param cls a `struct Queue`
 */
static void
queue_write (void *cls)
{
  struct Queue *queue = cls;
  ssize_t sent;

  queue->write_task = NULL;
  if (0 != queue->cwrite_off)
  {
    sent = GNUNET_NETWORK_socket_send (queue->sock,
                                       queue->cwrite_buf,
                                       queue->cwrite_off);
    if ((-1 == sent) && (EAGAIN != errno) && (EINTR != errno))
    {
      GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING, "send");
      queue_destroy (queue);
      return;
    }
    if (sent > 0)
    {
      size_t usent = (size_t) sent;

      memmove (queue->cwrite_buf,
               &queue->cwrite_buf[usent],
               queue->cwrite_off - usent);
      reschedule_queue_timeout (queue);
    }
  }
  /* can we encrypt more? (always encrypt full messages, needed
     such that #mq_cancel() can work!) */
  if ((0 < queue->rekey_left_bytes) &&
      (queue->cwrite_off + queue->pwrite_off <= BUF_SIZE))
  {
    GNUNET_assert (0 ==
                   gcry_cipher_encrypt (queue->out_cipher,
                                        &queue->cwrite_buf[queue->cwrite_off],
                                        queue->pwrite_off,
                                        queue->pwrite_buf,
                                        queue->pwrite_off));
    if (queue->rekey_left_bytes > queue->pwrite_off)
      queue->rekey_left_bytes -= queue->pwrite_off;
    else
      queue->rekey_left_bytes = 0;
    queue->cwrite_off += queue->pwrite_off;
    queue->pwrite_off = 0;
  }
  if ((0 == queue->pwrite_off) &&
      ((0 == queue->rekey_left_bytes) ||
       (0 ==
        GNUNET_TIME_absolute_get_remaining (queue->rekey_time).rel_value_us)))
  {
    gcry_cipher_close (queue->out_cipher);
    setup_out_cipher (queue);
    inject_rekey (queue);
  }
  if ((0 == queue->pwrite_off) && (! queue->finishing) &&
      (queue->mq_awaits_continue))
  {
    queue->mq_awaits_continue = GNUNET_NO;
    GNUNET_MQ_impl_send_continue (queue->mq);
  }
  /* did we just finish writing 'finish'? */
  if ((0 == queue->cwrite_off) && (GNUNET_YES == queue->finishing))
  {
    queue_destroy (queue);
    return;
  }
  /* do we care to write more? */
  if ((0 < queue->cwrite_off) || (0 < queue->pwrite_off))
    queue->write_task =
      GNUNET_SCHEDULER_add_write_net (GNUNET_TIME_UNIT_FOREVER_REL,
                                      queue->sock,
                                      &queue_write,
                                      queue);
}


/**
 * Signature of functions implementing the sending functionality of a
 * message queue.
 *
 * @param mq the message queue
 * @param msg the message to send
 * @param impl_state our `struct Queue`
 */
static void
mq_send (struct GNUNET_MQ_Handle *mq,
         const struct GNUNET_MessageHeader *msg,
         void *impl_state)
{
  struct Queue *queue = impl_state;
  uint16_t msize = ntohs (msg->size);
  struct TCPBox box;

  GNUNET_assert (mq == queue->mq);
  if (GNUNET_YES == queue->finishing)
    return; /* this queue is dying, drop msg */
  GNUNET_assert (0 == queue->pread_off);
  box.header.type = htons (GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_BOX);
  box.header.size = htons (msize);
  calculate_hmac (&queue->out_hmac, msg, msize, &box.hmac);
  memcpy (&queue->pread_buf[queue->pread_off], &box, sizeof (box));
  queue->pread_off += sizeof (box);
  memcpy (&queue->pread_buf[queue->pread_off], msg, msize);
  queue->pread_off += msize;
  GNUNET_assert (NULL != queue->sock);
  if (NULL == queue->write_task)
    queue->write_task =
      GNUNET_SCHEDULER_add_write_net (GNUNET_TIME_UNIT_FOREVER_REL,
                                      queue->sock,
                                      &queue_write,
                                      queue);
}


/**
 * Signature of functions implementing the destruction of a message
 * queue.  Implementations must not free @a mq, but should take care
 * of @a impl_state.
 *
 * @param mq the message queue to destroy
 * @param impl_state our `struct Queue`
 */
static void
mq_destroy (struct GNUNET_MQ_Handle *mq, void *impl_state)
{
  struct Queue *queue = impl_state;

  if (mq == queue->mq)
  {
    queue->mq = NULL;
    queue_finish (queue);
  }
}


/**
 * Implementation function that cancels the currently sent message.
 *
 * @param mq message queue
 * @param impl_state our `struct Queue`
 */
static void
mq_cancel (struct GNUNET_MQ_Handle *mq, void *impl_state)
{
  struct Queue *queue = impl_state;

  GNUNET_assert (0 != queue->pwrite_off);
  queue->pwrite_off = 0;
}


/**
 * Generic error handler, called with the appropriate
 * error code and the same closure specified at the creation of
 * the message queue.
 * Not every message queue implementation supports an error handler.
 *
 * @param cls our `struct Queue`
 * @param error error code
 */
static void
mq_error (void *cls, enum GNUNET_MQ_Error error)
{
  struct Queue *queue = cls;

  GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
              "MQ error in queue to %s: %d\n",
              GNUNET_i2s (&queue->target),
              (int) error);
  queue_finish (queue);
}


/**
 * Add the given @a queue to our internal data structure.  Setup the
 * MQ processing and inform transport that the queue is ready.  Must
 * be called after the KX for outgoing messages has been bootstrapped.
 *
 * @param queue queue to boot
 */
static void
boot_queue (struct Queue *queue, enum GNUNET_TRANSPORT_ConnectionStatus cs)
{
  queue->nt =
    GNUNET_NT_scanner_get_type (is, queue->address, queue->address_len);
  (void) GNUNET_CONTAINER_multipeermap_put (
    queue_map,
    &queue->target,
    queue,
    GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
  GNUNET_STATISTICS_set (stats,
                         "# queues active",
                         GNUNET_CONTAINER_multipeermap_size (queue_map),
                         GNUNET_NO);
  queue->timeout =
    GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
  queue->mq = GNUNET_MQ_queue_for_callbacks (&mq_send,
                                             &mq_destroy,
                                             &mq_cancel,
                                             queue,
                                             NULL,
                                             &mq_error,
                                             queue);
  {
    char *foreign_addr;

    switch (queue->address->sa_family)
    {
    case AF_INET:
      GNUNET_asprintf (&foreign_addr,
                       "%s-%s",
                       COMMUNICATOR_ADDRESS_PREFIX,
                       GNUNET_a2s (queue->address, queue->address_len));
      break;
    case AF_INET6:
      GNUNET_asprintf (&foreign_addr,
                       "%s-%s",
                       COMMUNICATOR_ADDRESS_PREFIX,
                       GNUNET_a2s (queue->address, queue->address_len));
      break;
    default:
      GNUNET_assert (0);
    }
    queue->qh = GNUNET_TRANSPORT_communicator_mq_add (ch,
                                                      &queue->target,
                                                      foreign_addr,
                                                      0 /* no MTU */,
                                                      queue->nt,
                                                      cs,
                                                      queue->mq);
    GNUNET_free (foreign_addr);
  }
}


/**
 * Generate and transmit our ephemeral key and the signature for
 * the initial KX with the other peer.  Must be called first, before
 * any other bytes are ever written to the output buffer.  Note that
 * our cipher must already be initialized when calling this function.
 * Helper function for #start_initial_kx_out().
 *
 * @param queue queue to do KX for
 * @param epub our public key for the KX
 */
static void
transmit_kx (struct Queue *queue,
             const struct GNUNET_CRYPTO_EcdhePublicKey *epub)
{
  struct TcpHandshakeSignature ths;
  struct TCPConfirmation tc;

  memcpy (queue->cwrite_buf, epub, sizeof (*epub));
  queue->cwrite_off = sizeof (epub);
  /* compute 'tc' and append in encrypted format to cwrite_buf */
  tc.sender = my_identity;
  tc.monotonic_time =
    GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get_monotonic (cfg));
  ths.purpose.purpose = htonl (GNUNET_SIGNATURE_COMMUNICATOR_TCP_HANDSHAKE);
  ths.purpose.size = htonl (sizeof (ths));
  ths.sender = my_identity;
  ths.receiver = queue->target;
  ths.ephemeral = *epub;
  ths.monotonic_time = tc.monotonic_time;
  GNUNET_assert (GNUNET_OK == GNUNET_CRYPTO_eddsa_sign (my_private_key,
                                                        &ths.purpose,
                                                        &tc.sender_sig));
  GNUNET_assert (0 ==
                 gcry_cipher_encrypt (queue->out_cipher,
                                      &queue->cwrite_buf[queue->cwrite_off],
                                      sizeof (tc),
                                      &tc,
                                      sizeof (tc)));
  queue->cwrite_off += sizeof (tc);
}


/**
 * Initialize our key material for outgoing transmissions and
 * inform the other peer about it. Must be called first before
 * any data is sent.
 *
 * @param queue the queue to setup
 */
static void
start_initial_kx_out (struct Queue *queue)
{
  struct GNUNET_CRYPTO_EcdhePublicKey epub;

  GNUNET_assert (GNUNET_OK ==
                 GNUNET_CRYPTO_ecdhe_key_create2 (&queue->ephemeral));
  GNUNET_CRYPTO_ecdhe_key_get_public (&queue->ephemeral, &epub);
  setup_out_cipher (queue);
  transmit_kx (queue, &epub);
}


/**
 * We have received the first bytes from the other side on a @a queue.
 * Decrypt the @a tc contained in @a ibuf and check the signature.
 * Note that #setup_in_cipher() must have already been called.
 *
 * @param queue queue to decrypt initial bytes from other peer for
 * @param tc[out] where to store the result
 * @param ibuf incoming data, of size
 *        `INITIAL_KX_SIZE`
 * @return #GNUNET_OK if the signature was OK, #GNUNET_SYSERR if not
 */
static int
decrypt_and_check_tc (struct Queue *queue,
                      struct TCPConfirmation *tc,
                      char *ibuf)
{
  struct TcpHandshakeSignature ths;

  GNUNET_assert (
    0 ==
    gcry_cipher_decrypt (queue->in_cipher,
                         tc,
                         sizeof (*tc),
                         &ibuf[sizeof (struct GNUNET_CRYPTO_EcdhePublicKey)],
                         sizeof (tc)));
  ths.purpose.purpose = htonl (GNUNET_SIGNATURE_COMMUNICATOR_TCP_HANDSHAKE);
  ths.purpose.size = htonl (sizeof (ths));
  ths.sender = tc->sender;
  ths.receiver = my_identity;
  memcpy (&ths.ephemeral, ibuf, sizeof (struct GNUNET_CRYPTO_EcdhePublicKey));
  ths.monotonic_time = tc->monotonic_time;
  /* FIXME: check monotonic time against previous mono times 
     from this sender! */
  return GNUNET_CRYPTO_eddsa_verify (GNUNET_SIGNATURE_COMMUNICATOR_TCP_HANDSHAKE,
                                     &ths.purpose,
                                     &tc->sender_sig,
                                     &tc->sender.public_key);
}


/**
 * Closes socket and frees memory associated with @a pq.
 *
 * @param pq proto queue to free
 */
static void
free_proto_queue (struct ProtoQueue *pq)
{
  GNUNET_NETWORK_socket_close (pq->sock);
  GNUNET_free (pq->address);
  GNUNET_CONTAINER_DLL_remove (proto_head, proto_tail, pq);
  GNUNET_free (pq);
}


/**
 * Read from the socket of the proto queue until we have enough data
 * to upgrade to full queue.
 *
 * @param cls a `struct ProtoQueue`
 */
static void
proto_read_kx (void *cls)
{
  struct ProtoQueue *pq = cls;
  ssize_t rcvd;
  struct GNUNET_TIME_Relative left;
  struct Queue *queue;
  struct TCPConfirmation tc;

  pq->read_task = NULL;
  left = GNUNET_TIME_absolute_get_remaining (pq->timeout);
  if (0 == left.rel_value_us)
  {
    free_proto_queue (pq);
    return;
  }
  rcvd = GNUNET_NETWORK_socket_recv (pq->sock,
                                     &pq->ibuf[pq->ibuf_off],
                                     sizeof (pq->ibuf) - pq->ibuf_off);
  if (-1 == rcvd)
  {
    if ((EAGAIN != errno) && (EINTR != errno))
    {
      GNUNET_log_strerror (GNUNET_ERROR_TYPE_DEBUG, "recv");
      free_proto_queue (pq);
      return;
    }
    /* try again */
    pq->read_task =
      GNUNET_SCHEDULER_add_read_net (left, pq->sock, &proto_read_kx, pq);
    return;
  }
  pq->ibuf_off += rcvd;
  if (pq->ibuf_off > sizeof (pq->ibuf))
  {
    /* read more */
    pq->read_task =
      GNUNET_SCHEDULER_add_read_net (left, pq->sock, &proto_read_kx, pq);
    return;
  }
  /* we got all the data, let's find out who we are talking to! */
  queue = GNUNET_new (struct Queue);
  setup_in_cipher ((const struct GNUNET_CRYPTO_EcdhePublicKey *) pq->ibuf,
                   queue);
  if (GNUNET_OK != decrypt_and_check_tc (queue, &tc, pq->ibuf))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                "Invalid TCP KX received from %s\n",
                GNUNET_a2s (queue->address, queue->address_len));
    gcry_cipher_close (queue->in_cipher);
    GNUNET_free (queue);
    free_proto_queue (pq);
    return;
  }
  queue->address = pq->address; /* steals reference */
  queue->address_len = pq->address_len;
  queue->target = tc.sender;
  start_initial_kx_out (queue);
  boot_queue (queue, GNUNET_TRANSPORT_CS_INBOUND);
  queue->read_task =
    GNUNET_SCHEDULER_add_read_net (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
                                   queue->sock,
                                   &queue_read,
                                   queue);
  queue->write_task =
    GNUNET_SCHEDULER_add_write_net (GNUNET_TIME_UNIT_FOREVER_REL,
                                    queue->sock,
                                    &queue_write,
                                    queue);
  GNUNET_CONTAINER_DLL_remove (proto_head, proto_tail, pq);
  GNUNET_free (pq);
}


/**
 * We have been notified that our listen socket has something to
 * read. Do the read and reschedule this function to be called again
 * once more is available.
 *
 * @param cls NULL
 */
static void
listen_cb (void *cls)
{
  struct sockaddr_storage in;
  socklen_t addrlen;
  struct GNUNET_NETWORK_Handle *sock;
  struct ProtoQueue *pq;

  listen_task = NULL;
  GNUNET_assert (NULL != listen_sock);
  addrlen = sizeof (in);
  memset (&in, 0, sizeof (in));
  sock = GNUNET_NETWORK_socket_accept (listen_sock,
                                       (struct sockaddr *) &in,
                                       &addrlen);
  if ((NULL == sock) && ((EMFILE == errno) || (ENFILE == errno)))
    return; /* system limit reached, wait until connection goes down */
  listen_task = GNUNET_SCHEDULER_add_read_net (GNUNET_TIME_UNIT_FOREVER_REL,
                                               listen_sock,
                                               &listen_cb,
                                               NULL);
  if ((NULL == sock) && ((EAGAIN == errno) || (ENOBUFS == errno)))
    return;
  if (NULL == sock)
  {
    GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING, "accept");
    return;
  }
  pq = GNUNET_new (struct ProtoQueue);
  pq->address_len = addrlen;
  pq->address = GNUNET_memdup (&in, addrlen);
  pq->timeout = GNUNET_TIME_relative_to_absolute (PROTO_QUEUE_TIMEOUT);
  pq->sock = sock;
  pq->read_task = GNUNET_SCHEDULER_add_read_net (PROTO_QUEUE_TIMEOUT,
                                                 pq->sock,
                                                 &proto_read_kx,
                                                 pq);
  GNUNET_CONTAINER_DLL_insert (proto_head, proto_tail, pq);
}


/**
 * Read from the socket of the queue until we have enough data
 * to initialize the decryption logic and can switch to regular
 * reading.
 *
 * @param cls a `struct Queue`
 */
static void
queue_read_kx (void *cls)
{
  struct Queue *queue = cls;
  ssize_t rcvd;
  struct GNUNET_TIME_Relative left;
  struct TCPConfirmation tc;

  queue->read_task = NULL;
  left = GNUNET_TIME_absolute_get_remaining (queue->timeout);
  if (0 == left.rel_value_us)
  {
    queue_destroy (queue);
    return;
  }
  rcvd = GNUNET_NETWORK_socket_recv (queue->sock,
                                     &queue->cread_buf[queue->cread_off],
                                     BUF_SIZE - queue->cread_off);
  if (-1 == rcvd)
  {
    if ((EAGAIN != errno) && (EINTR != errno))
    {
      GNUNET_log_strerror (GNUNET_ERROR_TYPE_DEBUG, "recv");
      queue_destroy (queue);
      return;
    }
    queue->read_task =
      GNUNET_SCHEDULER_add_read_net (left, queue->sock, &queue_read_kx, queue);
    return;
  }
  queue->cread_off += rcvd;
  if (queue->cread_off < INITIAL_KX_SIZE)
  {
    /* read more */
    queue->read_task =
      GNUNET_SCHEDULER_add_read_net (left, queue->sock, &queue_read_kx, queue);
    return;
  }
  /* we got all the data, let's find out who we are talking to! */
  setup_in_cipher ((const struct GNUNET_CRYPTO_EcdhePublicKey *)
                     queue->cread_buf,
                   queue);
  if (GNUNET_OK != decrypt_and_check_tc (queue, &tc, queue->cread_buf))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                "Invalid TCP KX received from %s\n",
                GNUNET_a2s (queue->address, queue->address_len));
    queue_destroy (queue);
    return;
  }
  if (0 !=
      memcmp (&tc.sender, &queue->target, sizeof (struct GNUNET_PeerIdentity)))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Invalid sender in TCP KX received from %s\n",
                GNUNET_a2s (queue->address, queue->address_len));
    queue_destroy (queue);
    return;
  }

  /* update queue timeout */
  reschedule_queue_timeout (queue);
  /* prepare to continue with regular read task immediately */
  memmove (queue->cread_buf,
           &queue->cread_buf[INITIAL_KX_SIZE],
           queue->cread_off - (INITIAL_KX_SIZE));
  queue->cread_off -= INITIAL_KX_SIZE;
  queue->read_task = GNUNET_SCHEDULER_add_now (&queue_read, queue);
}


/**
 * Function called by the transport service to initialize a
 * message queue given address information about another peer.
 * If and when the communication channel is established, the
 * communicator must call #GNUNET_TRANSPORT_communicator_mq_add()
 * to notify the service that the channel is now up.  It is
 * the responsibility of the communicator to manage sane
 * retries and timeouts for any @a peer/@a address combination
 * provided by the transport service.  Timeouts and retries
 * do not need to be signalled to the transport service.
 *
 * @param cls closure
 * @param peer identity of the other peer
 * @param address where to send the message, human-readable
 *        communicator-specific format, 0-terminated, UTF-8
 * @return #GNUNET_OK on success, #GNUNET_SYSERR if the provided address is
 * invalid
 */
static int
mq_init (void *cls, const struct GNUNET_PeerIdentity *peer, const char *address)
{
  struct Queue *queue;
  const char *path;
  struct sockaddr *in;
  socklen_t in_len;
  struct GNUNET_NETWORK_Handle *sock;

  if (0 != strncmp (address,
                    COMMUNICATOR_ADDRESS_PREFIX "-",
                    strlen (COMMUNICATOR_ADDRESS_PREFIX "-")))
  {
    GNUNET_break_op (0);
    return GNUNET_SYSERR;
  }
  path = &address[strlen (COMMUNICATOR_ADDRESS_PREFIX "-")];
  in = tcp_address_to_sockaddr (path, &in_len);

  sock = GNUNET_NETWORK_socket_create (in->sa_family, SOCK_STREAM, IPPROTO_TCP);
  if (NULL == sock)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "socket(%d) failed: %s",
                in->sa_family,
                STRERROR (errno));
    GNUNET_free (in);
    return GNUNET_SYSERR;
  }
  if (GNUNET_OK != GNUNET_NETWORK_socket_connect (sock, in, in_len))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "connect to `%s' failed: %s",
                address,
                STRERROR (errno));
    GNUNET_NETWORK_socket_close (sock);
    GNUNET_free (in);
    return GNUNET_SYSERR;
  }

  queue = GNUNET_new (struct Queue);
  queue->target = *peer;
  queue->address = in;
  queue->address_len = in_len;
  queue->sock = sock;
  boot_queue (queue, GNUNET_TRANSPORT_CS_OUTBOUND);
  queue->read_task =
    GNUNET_SCHEDULER_add_read_net (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
                                   queue->sock,
                                   &queue_read_kx,
                                   queue);
  start_initial_kx_out (queue);
  queue->write_task =
    GNUNET_SCHEDULER_add_write_net (GNUNET_TIME_UNIT_FOREVER_REL,
                                    queue->sock,
                                    &queue_write,
                                    queue);
  return GNUNET_OK;
}


/**
 * Iterator over all message queues to clean up.
 *
 * @param cls NULL
 * @param target unused
 * @param value the queue to destroy
 * @return #GNUNET_OK to continue to iterate
 */
static int
get_queue_delete_it (void *cls,
                     const struct GNUNET_PeerIdentity *target,
                     void *value)
{
  struct Queue *queue = value;

  (void) cls;
  (void) target;
  queue_destroy (queue);
  return GNUNET_OK;
}


/**
 * Shutdown the UNIX communicator.
 *
 * @param cls NULL (always)
 */
static void
do_shutdown (void *cls)
{
  while (NULL != proto_head)
    free_proto_queue (proto_head);
  if (NULL != nat)
  {
    GNUNET_NAT_unregister (nat);
    nat = NULL;
  }
  if (NULL != listen_task)
  {
    GNUNET_SCHEDULER_cancel (listen_task);
    listen_task = NULL;
  }
  if (NULL != listen_sock)
  {
    GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (listen_sock));
    listen_sock = NULL;
  }
  GNUNET_CONTAINER_multipeermap_iterate (queue_map, &get_queue_delete_it, NULL);
  GNUNET_CONTAINER_multipeermap_destroy (queue_map);
  if (NULL != ch)
  {
    GNUNET_TRANSPORT_communicator_disconnect (ch);
    ch = NULL;
  }
  if (NULL != stats)
  {
    GNUNET_STATISTICS_destroy (stats, GNUNET_NO);
    stats = NULL;
  }
  if (NULL != my_private_key)
  {
    GNUNET_free (my_private_key);
    my_private_key = NULL;
  }
  if (NULL != is)
  {
    GNUNET_NT_scanner_done (is);
    is = NULL;
  }
}


/**
 * Function called when the transport service has received an
 * acknowledgement for this communicator (!) via a different return
 * path.
 *
 * Not applicable for TCP.
 *
 * @param cls closure
 * @param sender which peer sent the notification
 * @param msg payload
 */
static void
enc_notify_cb (void *cls,
               const struct GNUNET_PeerIdentity *sender,
               const struct GNUNET_MessageHeader *msg)
{
  (void) cls;
  (void) sender;
  (void) msg;
  GNUNET_break_op (0);
}


/**
 * Signature of the callback passed to #GNUNET_NAT_register() for
 * a function to call whenever our set of 'valid' addresses changes.
 *
 * @param cls closure
 * @param app_ctx[in,out] location where the app can store stuff
 *                  on add and retrieve it on remove
 * @param add_remove #GNUNET_YES to add a new public IP address,
 *                   #GNUNET_NO to remove a previous (now invalid) one
 * @param ac address class the address belongs to
 * @param addr either the previous or the new public IP address
 * @param addrlen actual length of the @a addr
 */
static void
nat_address_cb (void *cls,
                void **app_ctx,
                int add_remove,
                enum GNUNET_NAT_AddressClass ac,
                const struct sockaddr *addr,
                socklen_t addrlen)
{
  char *my_addr;
  struct GNUNET_TRANSPORT_AddressIdentifier *ai;

  if (GNUNET_YES == add_remove)
  {
    enum GNUNET_NetworkType nt;

    GNUNET_asprintf (&my_addr,
                     "%s-%s",
                     COMMUNICATOR_ADDRESS_PREFIX,
                     GNUNET_a2s (addr, addrlen));
    nt = GNUNET_NT_scanner_get_type (is, addr, addrlen);
    ai =
      GNUNET_TRANSPORT_communicator_address_add (ch,
                                                 my_addr,
                                                 nt,
                                                 GNUNET_TIME_UNIT_FOREVER_REL);
    GNUNET_free (my_addr);
    *app_ctx = ai;
  }
  else
  {
    ai = *app_ctx;
    GNUNET_TRANSPORT_communicator_address_remove (ai);
    *app_ctx = NULL;
  }
}


/**
 * Setup communicator and launch network interactions.
 *
 * @param cls NULL (always)
 * @param args remaining command-line arguments
 * @param cfgfile name of the configuration file used (for saving, can be NULL!)
 * @param c configuration
 */
static void
run (void *cls,
     char *const *args,
     const char *cfgfile,
     const struct GNUNET_CONFIGURATION_Handle *c)
{
  char *bindto;
  struct sockaddr *in;
  socklen_t in_len;
  struct sockaddr_storage in_sto;
  socklen_t sto_len;

  (void) cls;
  cfg = c;
  if (GNUNET_OK !=
      GNUNET_CONFIGURATION_get_value_filename (cfg,
                                               COMMUNICATOR_CONFIG_SECTION,
                                               "BINDTO",
                                               &bindto))
  {
    GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
                               COMMUNICATOR_CONFIG_SECTION,
                               "BINDTO");
    return;
  }
  if (GNUNET_OK !=
      GNUNET_CONFIGURATION_get_value_number (cfg,
                                             COMMUNICATOR_CONFIG_SECTION,
                                             "MAX_QUEUE_LENGTH",
                                             &max_queue_length))
    max_queue_length = DEFAULT_MAX_QUEUE_LENGTH;

  in = tcp_address_to_sockaddr (bindto, &in_len);
  if (NULL == in)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Failed to setup TCP socket address with path `%s'\n",
                bindto);
    GNUNET_free (bindto);
    return;
  }
  listen_sock =
    GNUNET_NETWORK_socket_create (in->sa_family, SOCK_STREAM, IPPROTO_TCP);
  if (NULL == listen_sock)
  {
    GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR, "socket");
    GNUNET_free (in);
    GNUNET_free (bindto);
    return;
  }
  if (GNUNET_OK != GNUNET_NETWORK_socket_bind (listen_sock, in, in_len))
  {
    GNUNET_log_strerror_file (GNUNET_ERROR_TYPE_ERROR, "bind", bindto);
    GNUNET_NETWORK_socket_close (listen_sock);
    listen_sock = NULL;
    GNUNET_free (in);
    GNUNET_free (bindto);
    return;
  }
  /* We might have bound to port 0, allowing the OS to figure it out;
     thus, get the real IN-address from the socket */
  sto_len = sizeof (in_sto);
  if (0 != getsockname (GNUNET_NETWORK_get_fd (listen_sock),
                        (struct sockaddr *) &in_sto,
                        &sto_len))
  {
    memcpy (&in_sto, in, in_len);
    sto_len = in_len;
  }
  GNUNET_free (in);
  GNUNET_free (bindto);
  in = (struct sockaddr *) &in_sto;
  in_len = sto_len;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Bound to `%s'\n",
              GNUNET_a2s ((const struct sockaddr *) &in_sto, sto_len));
  stats = GNUNET_STATISTICS_create ("C-TCP", cfg);
  GNUNET_SCHEDULER_add_shutdown (&do_shutdown, NULL);
  is = GNUNET_NT_scanner_init ();
  my_private_key = GNUNET_CRYPTO_eddsa_key_create_from_configuration (cfg);
  if (NULL == my_private_key)
  {
    GNUNET_log (
      GNUNET_ERROR_TYPE_ERROR,
      _ (
        "Transport service is lacking key configuration settings. Exiting.\n"));
    GNUNET_SCHEDULER_shutdown ();
    return;
  }
  GNUNET_CRYPTO_eddsa_key_get_public (my_private_key, &my_identity.public_key);
  /* start listening */
  listen_task = GNUNET_SCHEDULER_add_read_net (GNUNET_TIME_UNIT_FOREVER_REL,
                                               listen_sock,
                                               &listen_cb,
                                               NULL);
  queue_map = GNUNET_CONTAINER_multipeermap_create (10, GNUNET_NO);
  ch = GNUNET_TRANSPORT_communicator_connect (cfg,
                                              COMMUNICATOR_CONFIG_SECTION,
                                              COMMUNICATOR_ADDRESS_PREFIX,
                                              GNUNET_TRANSPORT_CC_RELIABLE,
                                              &mq_init,
                                              NULL,
                                              &enc_notify_cb,
                                              NULL);
  if (NULL == ch)
  {
    GNUNET_break (0);
    GNUNET_SCHEDULER_shutdown ();
    return;
  }
  nat = GNUNET_NAT_register (cfg,
                             COMMUNICATOR_CONFIG_SECTION,
                             IPPROTO_TCP,
                             1 /* one address */,
                             (const struct sockaddr **) &in,
                             &in_len,
                             &nat_address_cb,
                             NULL /* FIXME: support reversal: #5529 */,
                             NULL /* closure */);
}


/**
 * The main function for the UNIX communicator.
 *
 * @param argc number of arguments from the command line
 * @param argv command line arguments
 * @return 0 ok, 1 on error
 */
int
main (int argc, char *const *argv)
{
  static const struct GNUNET_GETOPT_CommandLineOption options[] = {
    GNUNET_GETOPT_OPTION_END};
  int ret;

  if (GNUNET_OK != GNUNET_STRINGS_get_utf8_args (argc, argv, &argc, &argv))
    return 2;

  ret = (GNUNET_OK == GNUNET_PROGRAM_run (argc,
                                          argv,
                                          "gnunet-communicator-tcp",
                                          _ ("GNUnet TCP communicator"),
                                          options,
                                          &run,
                                          NULL))
          ? 0
          : 1;
  GNUNET_free ((void *) argv);
  return ret;
}


/* end of gnunet-communicator-tcp.c */