gnunet-communicator-tcp.c

Go to the documentation of this file.

/*
     This file is part of GNUnet
     Copyright (C) 2010-2014, 2018, 2019, 2026 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
 */
 
#include "platform.h"
#include "gnunet_common.h"
#include "gnunet_util_lib.h"
#include "gnunet_pils_service.h"
#include "gnunet_core_service.h"
#include "gnunet_peerstore_service.h"
#include "gnunet_protocols.h"
#include "gnunet_signatures.h"
#include "gnunet_constants.h"
#include "gnunet_nat_service.h"
#include "gnunet_statistics_service.h"
#include "gnunet_transport_communication_service.h"
#include "gnunet_resolver_service.h"
 
/* Shorthand for Logging */
#define LOG(kind, ...) GNUNET_log_from (kind, "communicator-tcp", __VA_ARGS__)
 
 
#define NAT_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10)
 
#define ADDRESS_VALIDITY_PERIOD \
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_HOURS, 4)
 
#define DEFAULT_MAX_QUEUE_LENGTH 8
 
#define BUF_SIZE (2 * 64 * 1024 + sizeof(struct TCPBox))
 
#define TCP_QUEUE_GONE SIZE_MAX
 
#define DEFAULT_REKEY_INTERVAL GNUNET_TIME_UNIT_DAYS
 
#define PROTO_QUEUE_TIMEOUT GNUNET_TIME_UNIT_MINUTES
 
#define FINISH_TIMEOUT GNUNET_TIME_relative_multiply ( \
          GNUNET_TIME_UNIT_SECONDS, 5)
 
#define REKEY_MAX_BYTES (1024LLU * 1024 * 400)
 
#define INITIAL_KX_SIZE                           \
        (sizeof(struct GNUNET_CRYPTO_EcdhePublicKey)   \
         + sizeof(struct TCPConfirmation))
 
#define INITIAL_CORE_KX_SIZE          \
        (sizeof(struct EphemeralKeyMessage)   \
         + sizeof(struct PingMessage) \
         + sizeof(struct PongMessage))
 
#define COMMUNICATOR_ADDRESS_PREFIX "tcp"
 
#define COMMUNICATOR_CONFIG_SECTION "communicator-tcp"
 
GNUNET_NETWORK_STRUCT_BEGIN
 
 
struct TcpHandshakeSignature
{
  struct GNUNET_CRYPTO_SignaturePurpose purpose;
 
  struct GNUNET_PeerIdentity sender;
 
  struct GNUNET_PeerIdentity receiver;
 
  struct GNUNET_CRYPTO_HpkeEncapsulation ephemeral;
 
  struct GNUNET_TIME_AbsoluteNBO monotonic_time;
 
  struct GNUNET_CRYPTO_ChallengeNonceP challenge;
};
 
struct TcpHandshakeAckSignature
{
  struct GNUNET_CRYPTO_SignaturePurpose purpose;
 
  struct GNUNET_PeerIdentity sender;
 
  struct GNUNET_PeerIdentity receiver;
 
  struct GNUNET_TIME_AbsoluteNBO monotonic_time;
 
  struct GNUNET_CRYPTO_ChallengeNonceP challenge;
};
 
struct TCPConfirmation
{
  struct GNUNET_PeerIdentity sender;
 
  struct GNUNET_CRYPTO_EddsaSignature sender_sig;
 
  struct GNUNET_TIME_AbsoluteNBO monotonic_time;
 
  struct GNUNET_CRYPTO_ChallengeNonceP challenge;
 
};
 
struct TCPConfirmationAck
{
 
 
  struct GNUNET_MessageHeader header;
 
  struct GNUNET_PeerIdentity sender;
 
  struct GNUNET_CRYPTO_EddsaSignature sender_sig;
 
  struct GNUNET_TIME_AbsoluteNBO monotonic_time;
 
  struct GNUNET_CRYPTO_ChallengeNonceP challenge;
 
};
 
struct TCPBox
{
  struct GNUNET_MessageHeader header;
 
  struct GNUNET_ShortHashCode hmac;
 
  /* followed by as may bytes of payload as indicated in @e header,
     excluding the TCPBox itself! */
};
 
 
struct TCPRekey
{
  struct GNUNET_MessageHeader header;
 
  struct GNUNET_ShortHashCode hmac;
 
  struct GNUNET_CRYPTO_HpkeEncapsulation ephemeral;
 
  struct GNUNET_CRYPTO_EddsaSignature sender_sig;
 
  struct GNUNET_TIME_AbsoluteNBO monotonic_time;
};
 
struct TcpRekeySignature
{
  struct GNUNET_CRYPTO_SignaturePurpose purpose;
 
  struct GNUNET_PeerIdentity sender;
 
  struct GNUNET_PeerIdentity receiver;
 
  struct GNUNET_CRYPTO_HpkeEncapsulation ephemeral;
 
  struct GNUNET_TIME_AbsoluteNBO monotonic_time;
};
 
struct TCPFinish
{
  struct GNUNET_MessageHeader header;
 
  struct GNUNET_ShortHashCode hmac;
};
 
struct TCPNATProbeMessage
{
  struct GNUNET_MessageHeader header;
 
  struct GNUNET_PeerIdentity clientIdentity;
};
 
GNUNET_NETWORK_STRUCT_END
 
struct PendingReversal
{
  /*
   * Timeout task.
   */
  struct GNUNET_SCHEDULER_Task *timeout_task;
 
  struct GNUNET_PeerIdentity target;
 
  struct sockaddr *in;
};
 
struct ListenTask
{
  struct GNUNET_SCHEDULER_Task *listen_task;
 
  struct GNUNET_NETWORK_Handle *listen_sock;
};
 
struct Queue
{
  struct GNUNET_PeerIdentity target;
 
  struct GNUNET_CRYPTO_HpkePublicKey target_hpke_key;
 
  struct GNUNET_NETWORK_Handle *listen_sock;
 
  struct GNUNET_NETWORK_Handle *sock;
 
  gcry_cipher_hd_t in_cipher;
 
  gcry_cipher_hd_t out_cipher;
 
  struct GNUNET_HashCode key;
 
  struct GNUNET_CRYPTO_AuthKey in_hmac;
 
  struct GNUNET_CRYPTO_AuthKey out_hmac;
 
  struct GNUNET_SCHEDULER_Task *read_task;
 
  struct GNUNET_SCHEDULER_Task *write_task;
 
  struct GNUNET_SCHEDULER_Task *finish_task;
 
  struct sockaddr *address;
 
  uint64_t rekey_left_bytes;
 
  struct GNUNET_TIME_Absolute rekey_time;
 
  socklen_t address_len;
 
  struct GNUNET_MQ_Handle *mq;
 
  struct GNUNET_TRANSPORT_QueueHandle *qh;
 
  unsigned long long bytes_in_queue;
 
  char cread_buf[BUF_SIZE];
 
  char cwrite_buf[BUF_SIZE];
 
  char pread_buf[UINT16_MAX + 1 + sizeof(struct TCPBox)];
 
  char pwrite_buf[UINT16_MAX + 1 + sizeof(struct TCPBox)];
 
  size_t cread_off;
 
  size_t cwrite_off;
 
  size_t pread_off;
 
  size_t pwrite_off;
 
  struct GNUNET_TIME_Absolute timeout;
 
  unsigned int backpressure;
 
  enum GNUNET_NetworkType nt;
 
  enum GNUNET_TRANSPORT_ConnectionStatus cs;
 
  int mq_awaits_continue;
 
  int finishing;
 
  int destroyed;
 
  int rekeyed;
 
  struct GNUNET_TIME_AbsoluteNBO rekey_monotonic_time;
 
  struct GNUNET_TIME_AbsoluteNBO handshake_monotonic_time;
 
  struct GNUNET_TIME_AbsoluteNBO handshake_ack_monotonic_time;
 
  struct GNUNET_CRYPTO_ChallengeNonceP challenge;
 
  struct GNUNET_CRYPTO_ChallengeNonceP challenge_received;
 
  struct GNUNET_PEERSTORE_IterateContext *rekey_monotime_get;
 
  struct GNUNET_PEERSTORE_IterateContext *handshake_monotime_get;
 
  struct GNUNET_PEERSTORE_IterateContext *handshake_ack_monotime_get;
 
  struct GNUNET_PEERSTORE_StoreContext *rekey_monotime_sc;
 
  struct GNUNET_PEERSTORE_StoreContext *handshake_monotime_sc;
 
  struct GNUNET_PEERSTORE_StoreContext *handshake_ack_monotime_sc;
 
  // TODO remove?
  size_t unverified_size;
 
  int initial_core_kx_done;
};
 
 
struct ProtoQueue
{
  struct ProtoQueue *next;
 
  struct ProtoQueue *prev;
 
  struct GNUNET_NETWORK_Handle *listen_sock;
 
  struct GNUNET_NETWORK_Handle *sock;
 
  struct GNUNET_SCHEDULER_Task *write_task;
 
  char write_buf[sizeof (struct TCPNATProbeMessage)];
 
  size_t write_off;
 
  struct GNUNET_SCHEDULER_Task *read_task;
 
  struct sockaddr *address;
 
  socklen_t address_len;
 
  struct GNUNET_TIME_Absolute timeout;
 
  char ibuf[INITIAL_KX_SIZE];
 
  size_t ibuf_off;
};
 
struct PortOnlyIpv4Ipv6
{
  struct sockaddr *addr_ipv4;
 
  socklen_t addr_len_ipv4;
 
  struct sockaddr *addr_ipv6;
 
  socklen_t addr_len_ipv6;
 
};
 
struct Addresses
{
  struct Addresses *next;
 
  struct Addresses *prev;
 
  struct sockaddr *addr;
 
  socklen_t addr_len;
 
};
 
 
static unsigned long long max_queue_length;
 
static struct GNUNET_PILS_KeyRing *key_ring;
 
static struct GNUNET_PILS_Handle *pils;
 
static struct GNUNET_STATISTICS_Handle *stats;
 
static struct GNUNET_TRANSPORT_CommunicatorHandle *ch;
 
static struct GNUNET_CONTAINER_MultiHashMap *queue_map;
 
static struct GNUNET_CONTAINER_MultiHashMap *lt_map;
 
static unsigned long long rekey_max_bytes;
 
static struct GNUNET_TIME_Relative rekey_interval;
 
static const struct GNUNET_CONFIGURATION_Handle *cfg;
 
static struct GNUNET_NT_InterfaceScanner *is;
 
static struct GNUNET_NAT_Handle *nat;
 
static struct ProtoQueue *proto_head;
 
static struct ProtoQueue *proto_tail;
 
struct GNUNET_RESOLVER_RequestHandle *resolve_request_handle;
 
static struct Addresses *addrs_head;
 
static struct Addresses *addrs_tail;
 
static int addrs_lens;
 
static struct GNUNET_PEERSTORE_Handle *peerstore;
 
static int shutdown_running = GNUNET_NO;
 
static int disable_v6;
 
static unsigned int bind_port;
 
static struct GNUNET_CONTAINER_MultiHashMap *pending_reversals;
 
static void
listen_cb (void *cls);
 
static void
eddsa_priv_to_hpke_key (const struct GNUNET_CRYPTO_EddsaPrivateKey *edpk,
                        struct GNUNET_CRYPTO_HpkePrivateKey *pk)
{
  struct GNUNET_CRYPTO_BlindablePrivateKey key;
  key.type = htonl (GNUNET_PUBLIC_KEY_TYPE_EDDSA);
  key.eddsa_key = *edpk;
  GNUNET_CRYPTO_hpke_sk_to_x25519 (&key,
                                   pk);
}
 
 
static void
eddsa_pub_to_hpke_key (const struct GNUNET_CRYPTO_EddsaPublicKey *edpk,
                       struct GNUNET_CRYPTO_HpkePublicKey *pk)
{
  struct GNUNET_CRYPTO_BlindablePublicKey key;
  key.type = htonl (GNUNET_PUBLIC_KEY_TYPE_EDDSA);
  key.eddsa_key = *edpk;
  GNUNET_CRYPTO_hpke_pk_to_x25519 (&key, pk);
}
 
 
static void
queue_destroy (struct Queue *queue)
{
  struct ListenTask *lt = NULL;
  struct GNUNET_HashCode h_sock;
  int sockfd;
 
  if (NULL != queue->listen_sock)
  {
    sockfd = GNUNET_NETWORK_get_fd (queue->listen_sock);
    GNUNET_CRYPTO_hash (&sockfd,
                        sizeof(int),
                        &h_sock);
 
    lt = GNUNET_CONTAINER_multihashmap_get (lt_map, &h_sock);
  }
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Disconnecting queue for peer `%s'\n",
              GNUNET_i2s (&queue->target));
  if (NULL != queue->rekey_monotime_sc)
  {
    GNUNET_PEERSTORE_store_cancel (queue->rekey_monotime_sc);
    queue->rekey_monotime_sc = NULL;
  }
  if (NULL != queue->handshake_monotime_sc)
  {
    GNUNET_PEERSTORE_store_cancel (queue->handshake_monotime_sc);
    queue->handshake_monotime_sc = NULL;
  }
  if (NULL != queue->handshake_ack_monotime_sc)
  {
    GNUNET_PEERSTORE_store_cancel (queue->handshake_ack_monotime_sc);
    queue->handshake_ack_monotime_sc = NULL;
  }
  if (NULL != queue->rekey_monotime_get)
  {
    GNUNET_PEERSTORE_iteration_stop (queue->rekey_monotime_get);
    queue->rekey_monotime_get = NULL;
  }
  if (NULL != queue->handshake_monotime_get)
  {
    GNUNET_PEERSTORE_iteration_stop (queue->handshake_monotime_get);
    queue->handshake_monotime_get = NULL;
  }
  if (NULL != queue->handshake_ack_monotime_get)
  {
    GNUNET_PEERSTORE_iteration_stop (queue->handshake_ack_monotime_get);
    queue->handshake_ack_monotime_get = NULL;
  }
  if (NULL != queue->qh)
  {
    GNUNET_TRANSPORT_communicator_mq_del (queue->qh);
    queue->qh = NULL;
  }
  GNUNET_assert (
    GNUNET_YES ==
    GNUNET_CONTAINER_multihashmap_remove (queue_map, &queue->key, queue));
  GNUNET_STATISTICS_set (stats,
                         "# queues active",
                         GNUNET_CONTAINER_multihashmap_size (queue_map),
                         GNUNET_NO);
  if (NULL != queue->finish_task)
  {
    GNUNET_SCHEDULER_cancel (queue->finish_task);
    queue->finish_task = NULL;
  }
  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;
  }
  if (GNUNET_SYSERR == GNUNET_NETWORK_socket_close (queue->sock))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "closing socket failed\n");
  }
  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 == lt)
    return;
 
  if ((! shutdown_running) && (NULL == lt->listen_task))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "add read net listen\n");
    lt->listen_task = GNUNET_SCHEDULER_add_read_net (
      GNUNET_TIME_UNIT_FOREVER_REL,
      lt->listen_sock,
      &listen_cb,
      lt);
  }
  else
    GNUNET_free (lt);
}
 
 
static void
calculate_hmac (struct GNUNET_CRYPTO_AuthKey *hmac_secret,
                const void *buf,
                size_t buf_size,
                struct GNUNET_ShortHashCode *smac)
{
  struct GNUNET_HashCode mac;
 
  GNUNET_CRYPTO_hmac (hmac_secret,
                      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),
                      (struct GNUNET_HashCode*) hmac_secret);
}
 
 
static void
queue_write (void *cls);
 
 
static void
finish_timeout_cb (void *cls)
{
  struct Queue *queue = cls;
 
  queue->finish_task = NULL;
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              "Queue to %s did not flush its FINISH in time, tearing it down\n",
              GNUNET_i2s (&queue->target));
  queue_destroy (queue);
}
 
 
static void
queue_finish (struct Queue *queue)
{
  struct TCPFinish fin;
 
  if (GNUNET_YES == queue->destroyed)
    return;   /* socket is already gone, we only await backpressure draining */
  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;
  /* ... but #queue_write() only ever re-arms itself while it still has data
     buffered, and #mq_send() returns early for a finishing queue.  So unless
     we arm the write task here, nothing writes the @c fin, #queue_destroy()
     is never reached, and transport is never told that this queue died: it
     keeps scheduling transmissions that #mq_send() silently drops, and CORE
     never learns about the disconnect. */
  if (NULL == queue->write_task)
    queue->write_task =
      GNUNET_SCHEDULER_add_write_net (GNUNET_TIME_UNIT_FOREVER_REL,
                                      queue->sock,
                                      &queue_write,
                                      queue);
  /* And if the peer is gone, the socket may never become writable at all. */
  if (NULL == queue->finish_task)
    queue->finish_task = GNUNET_SCHEDULER_add_delayed (FINISH_TIMEOUT,
                                                       &finish_timeout_cb,
                                                       queue);
}
 
 
static void
queue_read (void *cls);
 
 
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);
  if (NULL == queue)
    return;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "backpressure %u\n",
              queue->backpressure);
 
  queue->backpressure--;
  /* handle deferred queue destruction */
  if ((queue->destroyed) && (0 == queue->backpressure))
  {
    GNUNET_free (queue);
    return;
  }
  else if (GNUNET_YES != queue->destroyed)
  {
    queue->timeout =
      GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT
                                        );
    /* 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);
  }
}
 
 
static void
pass_plaintext_to_core (struct Queue *queue,
                        const void *plaintext,
                        size_t plaintext_len)
{
  const struct GNUNET_MessageHeader *hdr = plaintext;
  int ret;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "pass message from %s to core\n",
              GNUNET_i2s (&queue->target));
 
  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);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "passed to core\n");
  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);
}
 
 
static void
setup_cipher (const struct GNUNET_ShortHashCode *prk,
              const struct GNUNET_PeerIdentity *pid,
              gcry_cipher_hd_t *cipher,
              struct GNUNET_CRYPTO_AuthKey *hmac_key)
{
  char key[256 / 8];
  char ctr[128 / 8];
 
  GNUNET_assert (0 == gcry_cipher_open (cipher,
                                        GCRY_CIPHER_AES256 /* low level: go for speed */
                                        ,
                                        GCRY_CIPHER_MODE_CTR,
                                        0 /* flags */));
  GNUNET_assert (GNUNET_YES ==
                 GNUNET_CRYPTO_hkdf_expand (
                   key,
                   sizeof(key),
                   prk,
                   GNUNET_CRYPTO_kdf_arg_string ("gnunet-communicator-tcp-key"))
                 );
  GNUNET_assert (0 == gcry_cipher_setkey (*cipher, key, sizeof(key)));
  GNUNET_assert (GNUNET_YES ==
                 GNUNET_CRYPTO_hkdf_expand (
                   ctr,
                   sizeof(ctr),
                   prk,
                   GNUNET_CRYPTO_kdf_arg_string ("gnunet-communicator-tcp-ctr"))
                 );
  gcry_cipher_setctr (*cipher, ctr, sizeof(ctr));
  GNUNET_assert (GNUNET_YES ==
                 GNUNET_CRYPTO_hkdf_expand (
                   hmac_key,
                   sizeof(struct GNUNET_HashCode),
                   prk,
                   GNUNET_CRYPTO_kdf_arg_string ("gnunet-communicator-hmac")));
}
 
 
static void
rekey_monotime_store_cb (void *cls, int success)
{
  struct Queue *queue = cls;
  if (GNUNET_OK != success)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Failed to store rekey monotonic time in PEERSTORE!\n");
  }
  queue->rekey_monotime_sc = NULL;
  GNUNET_PEERSTORE_iteration_next (queue->rekey_monotime_get, 1);
}
 
 
static void
rekey_monotime_cb (void *cls,
                   const struct GNUNET_PEERSTORE_Record *record,
                   const char *emsg)
{
  struct Queue *queue = cls;
  struct GNUNET_TIME_AbsoluteNBO *mtbe;
  struct GNUNET_TIME_Absolute mt;
  const struct GNUNET_PeerIdentity *pid;
  struct GNUNET_TIME_AbsoluteNBO *rekey_monotonic_time;
 
  (void) emsg;
 
  rekey_monotonic_time = &queue->rekey_monotonic_time;
  pid = &queue->target;
  if (NULL == record)
  {
    queue->rekey_monotime_get = NULL;
    return;
  }
  if (sizeof(*mtbe) != record->value_size)
  {
    GNUNET_PEERSTORE_iteration_next (queue->rekey_monotime_get, 1);
    GNUNET_break (0);
    return;
  }
  mtbe = record->value;
  mt = GNUNET_TIME_absolute_ntoh (*mtbe);
  if (mt.abs_value_us > GNUNET_TIME_absolute_ntoh (
        queue->rekey_monotonic_time).abs_value_us)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Queue from %s dropped, rekey monotime in the past\n",
                GNUNET_i2s (&queue->target));
    GNUNET_break (0);
    GNUNET_PEERSTORE_iteration_stop (queue->rekey_monotime_get);
    queue->rekey_monotime_get = NULL;
    // FIXME: Why should we try to gracefully finish here??
    queue_finish (queue);
    return;
  }
  queue->rekey_monotime_sc = GNUNET_PEERSTORE_store (peerstore,
                                                     "transport_tcp_communicator",
                                                     pid,
                                                     GNUNET_PEERSTORE_TRANSPORT_TCP_COMMUNICATOR_REKEY,
                                                     rekey_monotonic_time,
                                                     sizeof(*
                                                            rekey_monotonic_time),
                                                     GNUNET_TIME_UNIT_FOREVER_ABS,
                                                     GNUNET_PEERSTORE_STOREOPTION_REPLACE,
                                                     &rekey_monotime_store_cb,
                                                     queue);
}
 
 
static void
setup_in_cipher_elligator (
  const struct GNUNET_CRYPTO_HpkeEncapsulation *c,
  struct Queue *queue)
{
  const struct GNUNET_PeerIdentity *my_identity;
  const struct GNUNET_CRYPTO_EddsaPrivateKey *my_private_key;
  struct GNUNET_CRYPTO_HpkePrivateKey my_hpke_key;
  struct GNUNET_ShortHashCode k;
 
  my_identity = GNUNET_PILS_get_identity (pils);
  my_private_key = GNUNET_PILS_key_ring_get_private_key (key_ring);
  GNUNET_assert ((my_identity) && (my_private_key));
 
  eddsa_priv_to_hpke_key (my_private_key, &my_hpke_key);
 
  GNUNET_CRYPTO_hpke_elligator_kem_decaps (&my_hpke_key,
                                           c,
                                           &k);
  setup_cipher (&k, my_identity, &queue->in_cipher, &queue->in_hmac);
}
 
 
static void
setup_in_cipher (const struct GNUNET_CRYPTO_HpkeEncapsulation *ephemeral,
                 struct Queue *queue)
{
  const struct GNUNET_PeerIdentity *my_identity;
  const struct GNUNET_CRYPTO_EddsaPrivateKey *my_private_key;
  struct GNUNET_ShortHashCode k;
 
  my_identity = GNUNET_PILS_get_identity (pils);
  my_private_key = GNUNET_PILS_key_ring_get_private_key (key_ring);
  GNUNET_assert ((my_identity) && (my_private_key));
 
  GNUNET_CRYPTO_eddsa_kem_decaps (my_private_key, ephemeral, &k);
  setup_cipher (&k, my_identity, &queue->in_cipher, &queue->in_hmac);
}
 
 
static void
do_rekey (struct Queue *queue, const struct TCPRekey *rekey)
{
  const struct GNUNET_PeerIdentity *my_identity;
  struct TcpRekeySignature thp;
 
  my_identity = GNUNET_PILS_get_identity (pils);
  GNUNET_assert (my_identity);
 
  thp.purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_COMMUNICATOR_TCP_REKEY);
  thp.purpose.size = htonl (sizeof(thp));
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "do_rekey size %u\n",
              thp.purpose.size);
  thp.sender = queue->target;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "sender %s\n",
              GNUNET_p2s (&thp.sender.public_key));
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "sender %s\n",
              GNUNET_p2s (&queue->target.public_key));
  thp.receiver = *my_identity;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "receiver %s\n",
              GNUNET_p2s (&thp.receiver.public_key));
  thp.ephemeral = rekey->ephemeral;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "ephemeral %s\n",
              GNUNET_e2s ((struct GNUNET_CRYPTO_EcdhePublicKey*) &thp.ephemeral)
              );
  thp.monotonic_time = rekey->monotonic_time;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "time %s\n",
              GNUNET_STRINGS_absolute_time_to_string (
                GNUNET_TIME_absolute_ntoh (thp.monotonic_time)));
  GNUNET_assert (ntohl ((&thp)->purpose.size) == sizeof (*(&thp)));
  if (GNUNET_OK !=
      GNUNET_CRYPTO_eddsa_verify (
        GNUNET_SIGNATURE_PURPOSE_COMMUNICATOR_TCP_REKEY,
        &thp,
        &rekey->sender_sig,
        &queue->target.public_key))
  {
    GNUNET_break (0);
    // FIXME Why should we try to gracefully finish here?
    queue_finish (queue);
    return;
  }
  queue->rekey_monotonic_time = rekey->monotonic_time;
  queue->rekey_monotime_get = GNUNET_PEERSTORE_iteration_start (peerstore,
                                                                "transport_tcp_communicator",
                                                                &queue->target,
                                                                GNUNET_PEERSTORE_TRANSPORT_TCP_COMMUNICATOR_REKEY,
                                                                &
                                                                rekey_monotime_cb,
                                                                queue);
  gcry_cipher_close (queue->in_cipher);
  queue->rekeyed = GNUNET_YES;
  setup_in_cipher (&rekey->ephemeral, queue);
}
 
 
static void
handshake_ack_monotime_store_cb (void *cls, int success)
{
  struct Queue *queue = cls;
 
  if (GNUNET_OK != success)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Failed to store handshake ack monotonic time in PEERSTORE!\n");
  }
  queue->handshake_ack_monotime_sc = NULL;
  GNUNET_PEERSTORE_iteration_next (queue->handshake_ack_monotime_get, 1);
}
 
 
static void
handshake_ack_monotime_cb (void *cls,
                           const struct GNUNET_PEERSTORE_Record *record,
                           const char *emsg)
{
  struct Queue *queue = cls;
  struct GNUNET_TIME_AbsoluteNBO *mtbe;
  struct GNUNET_TIME_Absolute mt;
  const struct GNUNET_PeerIdentity *pid;
  struct GNUNET_TIME_AbsoluteNBO *handshake_ack_monotonic_time;
 
  (void) emsg;
 
  handshake_ack_monotonic_time = &queue->handshake_ack_monotonic_time;
  pid = &queue->target;
  if (NULL == record)
  {
    queue->handshake_ack_monotime_get = NULL;
    return;
  }
  if (sizeof(*mtbe) != record->value_size)
  {
    GNUNET_PEERSTORE_iteration_next (queue->handshake_ack_monotime_get, 1);
    GNUNET_break (0);
    return;
  }
  mtbe = record->value;
  mt = GNUNET_TIME_absolute_ntoh (*mtbe);
  if (mt.abs_value_us > GNUNET_TIME_absolute_ntoh (
        queue->handshake_ack_monotonic_time).abs_value_us)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Queue from %s dropped, handshake ack monotime in the past\n",
                GNUNET_i2s (&queue->target));
    GNUNET_break (0);
    GNUNET_PEERSTORE_iteration_stop (queue->handshake_ack_monotime_get);
    queue->handshake_ack_monotime_get = NULL;
    // FIXME: Why should we try to gracefully finish here?
    queue_finish (queue);
    return;
  }
  queue->handshake_ack_monotime_sc =
    GNUNET_PEERSTORE_store (peerstore,
                            "transport_tcp_communicator",
                            pid,
                            GNUNET_PEERSTORE_TRANSPORT_TCP_COMMUNICATOR_HANDSHAKE_ACK,
                            handshake_ack_monotonic_time,
                            sizeof(*handshake_ack_monotonic_time),
                            GNUNET_TIME_UNIT_FOREVER_ABS,
                            GNUNET_PEERSTORE_STOREOPTION_REPLACE,
                            &handshake_ack_monotime_store_cb,
                            queue);
}
 
 
static void
send_challenge (struct GNUNET_CRYPTO_ChallengeNonceP challenge,
                struct Queue *queue)
{
  const struct GNUNET_PeerIdentity *my_identity;
  const struct GNUNET_CRYPTO_EddsaPrivateKey *my_private_key;
  struct TCPConfirmationAck tca;
  struct TcpHandshakeAckSignature thas;
 
  my_identity = GNUNET_PILS_get_identity (pils);
  my_private_key = GNUNET_PILS_key_ring_get_private_key (key_ring);
  GNUNET_assert ((my_identity) && (my_private_key));
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "sending challenge\n");
 
  tca.header.type = ntohs (
    GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_CONFIRMATION_ACK);
  tca.header.size = ntohs (sizeof(tca));
  tca.challenge = challenge;
  tca.sender = *my_identity;
  tca.monotonic_time =
    GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get_monotonic (cfg));
  thas.purpose.purpose = htonl (
    GNUNET_SIGNATURE_PURPOSE_COMMUNICATOR_TCP_HANDSHAKE_ACK);
  thas.purpose.size = htonl (sizeof(thas));
  thas.sender = *my_identity;
  thas.receiver = queue->target;
  thas.monotonic_time = tca.monotonic_time;
  thas.challenge = tca.challenge;
  GNUNET_CRYPTO_eddsa_sign (my_private_key,
                            &thas,
                            &tca.sender_sig);
  GNUNET_assert (0 ==
                 gcry_cipher_encrypt (queue->out_cipher,
                                      &queue->cwrite_buf[queue->cwrite_off],
                                      sizeof(tca),
                                      &tca,
                                      sizeof(tca)));
  queue->cwrite_off += sizeof(tca);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "sending challenge done\n");
}
 
 
static void
setup_out_cipher (struct Queue *queue, struct GNUNET_ShortHashCode *dh)
{
  setup_cipher (dh, &queue->target, &queue->out_cipher, &queue->out_hmac);
  queue->rekey_time = GNUNET_TIME_relative_to_absolute (rekey_interval);
  queue->rekey_left_bytes =
    GNUNET_CRYPTO_random_u64 (rekey_max_bytes);
}
 
 
static void
inject_rekey (struct Queue *queue)
{
  const struct GNUNET_PeerIdentity *my_identity;
  const struct GNUNET_CRYPTO_EddsaPrivateKey *my_private_key;
  struct TCPRekey rekey;
  struct TcpRekeySignature thp;
  struct GNUNET_ShortHashCode k;
 
  my_identity = GNUNET_PILS_get_identity (pils);
  my_private_key = GNUNET_PILS_key_ring_get_private_key (key_ring);
  GNUNET_assert ((my_identity) && (my_private_key));
 
  GNUNET_assert (0 == queue->pwrite_off);
  memset (&rekey, 0, sizeof(rekey));
  GNUNET_CRYPTO_eddsa_kem_encaps (&queue->target.public_key, &rekey.ephemeral,
                                  &k);
  rekey.header.type = ntohs (GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_REKEY);
  rekey.header.size = ntohs (sizeof(rekey));
  rekey.monotonic_time =
    GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get_monotonic (cfg));
  thp.purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_COMMUNICATOR_TCP_REKEY);
  thp.purpose.size = htonl (sizeof(thp));
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "inject_rekey size %u\n",
              thp.purpose.size);
  thp.sender = *my_identity;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "sender %s\n",
              GNUNET_p2s (&thp.sender.public_key));
  thp.receiver = queue->target;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "receiver %s\n",
              GNUNET_p2s (&thp.receiver.public_key));
  thp.ephemeral = rekey.ephemeral;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "ephemeral %s\n",
              GNUNET_e2s ((struct GNUNET_CRYPTO_EcdhePublicKey*) &thp.ephemeral)
              );
  thp.monotonic_time = rekey.monotonic_time;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "time %s\n",
              GNUNET_STRINGS_absolute_time_to_string (
                GNUNET_TIME_absolute_ntoh (thp.monotonic_time)));
  GNUNET_CRYPTO_eddsa_sign (my_private_key,
                            &thp,
                            &rekey.sender_sig);
  calculate_hmac (&queue->out_hmac, &rekey, sizeof(rekey), &rekey.hmac);
  /* Encrypt rekey message with 'old' cipher */
  GNUNET_assert (0 ==
                 gcry_cipher_encrypt (queue->out_cipher,
                                      &queue->cwrite_buf[queue->cwrite_off],
                                      sizeof(rekey),
                                      &rekey,
                                      sizeof(rekey)));
  queue->cwrite_off += sizeof(rekey);
  /* Setup new cipher for successive messages */
  gcry_cipher_close (queue->out_cipher);
  setup_out_cipher (queue, &k);
}
 
 
static void
pending_reversal_key (const struct sockaddr *in,
                      const struct GNUNET_PeerIdentity *target,
                      struct GNUNET_HashCode *key)
{
  struct sockaddr_in v4;
  struct GNUNET_HashContext *hsh;
 
  GNUNET_assert (AF_INET == in->sa_family);
  GNUNET_memcpy (&v4, in, sizeof(v4));
  v4.sin_port = 0;
  hsh = GNUNET_CRYPTO_hash_context_start ();
  GNUNET_CRYPTO_hash_context_read (hsh, &v4, sizeof(v4));
  GNUNET_CRYPTO_hash_context_read (hsh, target, sizeof(*target));
  GNUNET_CRYPTO_hash_context_finish (hsh, key);
}
 
 
static int
pending_reversals_delete_it (void *cls,
                             const struct GNUNET_HashCode *key,
                             void *value)
{
  struct PendingReversal *pending_reversal = value;
  (void) cls;
 
  if (NULL != pending_reversal->timeout_task)
  {
    GNUNET_SCHEDULER_cancel (pending_reversal->timeout_task);
    pending_reversal->timeout_task = NULL;
  }
  GNUNET_assert (GNUNET_YES == GNUNET_CONTAINER_multihashmap_remove (
                   pending_reversals,
                   key,
                   pending_reversal));
  GNUNET_free (pending_reversal->in);
  GNUNET_free (pending_reversal);
  return GNUNET_OK;
}
 
 
static void
check_and_remove_pending_reversal (struct sockaddr *in, sa_family_t sa_family,
                                   const struct GNUNET_PeerIdentity *sender)
{
  if (AF_INET == sa_family)
  {
    struct PendingReversal *pending_reversal;
    struct GNUNET_HashCode key;
 
    pending_reversal_key (in, sender, &key);
    pending_reversal = GNUNET_CONTAINER_multihashmap_get (pending_reversals,
                                                          &key);
    if (NULL != pending_reversal)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Removing completed pending reversal for `%s' at `%s'\n",
                  GNUNET_i2s (&pending_reversal->target),
                  GNUNET_a2s (in, sizeof (struct sockaddr)));
      pending_reversals_delete_it (NULL, &key, pending_reversal);
    }
  }
}
 
 
static void
free_proto_queue (struct ProtoQueue *pq)
{
  if (NULL != pq->listen_sock)
  {
    GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (pq->listen_sock));
    pq->listen_sock = NULL;
  }
  if (NULL != pq->read_task)
  {
    GNUNET_SCHEDULER_cancel (pq->read_task);
    pq->read_task = NULL;
  }
  if (NULL != pq->write_task)
  {
    GNUNET_SCHEDULER_cancel (pq->write_task);
    pq->write_task = NULL;
  }
  /* NOTE: deliberately no #check_and_remove_pending_reversal() here.  A proto
     queue is anonymous -- we have not completed the KX, so we do not know
     which peer this was.  Dropping "the" pending reversal for the IP used to
     cancel the reversal of an unrelated peer that happens to sit behind the
     same NAT.  A reversal that truly went nowhere expires on #NAT_TIMEOUT. */
  GNUNET_NETWORK_socket_close (pq->sock);
  GNUNET_free (pq->address);
  GNUNET_CONTAINER_DLL_remove (proto_head, proto_tail, pq);
  GNUNET_free (pq);
}
 
 
static void
proto_queue_write (void *cls)
{
  struct ProtoQueue *pq = cls;
  ssize_t sent;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "In proto queue write\n");
  pq->write_task = NULL;
  if (0 != pq->write_off)
  {
    sent = GNUNET_NETWORK_socket_send (pq->sock,
                                       pq->write_buf,
                                       pq->write_off);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Sent %lu bytes to TCP queue\n", sent);
    if ((-1 == sent) && (EAGAIN != errno) && (EINTR != errno))
    {
      GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING, "send");
      free_proto_queue (pq);
      return;
    }
    if (sent > 0)
    {
      size_t usent = (size_t) sent;
      pq->write_off -= usent;
      memmove (pq->write_buf,
               &pq->write_buf[usent],
               pq->write_off);
    }
  }
  /* do we care to write more? */
  if ((0 < pq->write_off))
    pq->write_task =
      GNUNET_SCHEDULER_add_write_net (GNUNET_TIME_UNIT_FOREVER_REL,
                                      pq->sock,
                                      &proto_queue_write,
                                      pq);
}
 
 
static void
queue_write (void *cls)
{
  struct Queue *queue = cls;
  ssize_t sent;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "In queue write\n");
  queue->write_task = NULL;
  if (0 != queue->cwrite_off)
  {
    sent = GNUNET_NETWORK_socket_send (queue->sock,
                                       queue->cwrite_buf,
                                       queue->cwrite_off);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Sent %lu bytes to TCP queue\n", sent);
    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;
      queue->cwrite_off -= usent;
      memmove (queue->cwrite_buf,
               &queue->cwrite_buf[usent],
               queue->cwrite_off);
      queue->timeout =
        GNUNET_TIME_relative_to_absolute (
          GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
    }
  }
  {
    /* can we encrypt more? (always encrypt full messages, needed
       such that #mq_cancel() can work!)
 
       NOTE: this used to be gated on a `we_do_not_need_to_rekey' flag
       computed as `0 < rekey_left_bytes - (cwrite_off + pwrite_off +
       sizeof (struct TCPRekey))'.  That subtraction is unsigned, so it
       wrapped to a huge value precisely when the rekey budget was about to
       run out -- inverting the test it was meant to express, and reporting
       "no rekey needed" even for a @e rekey_left_bytes of zero.  The single
       value it did get right, exact equality, was worse still: it let the
       branch below call #inject_rekey() with a non-empty @e pwrite_buf,
       which that function asserts against, and if neither rekey trigger
       held it wedged this queue into a busy write loop that never drained
       @e pwrite_buf again -- a stall the peer sees as silence while the
       socket stays open and the communicator keeps reporting the queue up.
 
       Encrypting a message first and rekeying immediately afterwards
       overshoots the byte budget by at most one message, which against the
       400 MB default is noise. */
    if ((queue->pwrite_off > 0) &&
        (queue->cwrite_off + queue->pwrite_off <= BUF_SIZE))
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Encrypting %lu bytes\n", queue->pwrite_off);
      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;
    }
    /* Rekey once the byte budget is spent or the interval elapsed.
       #inject_rekey() requires an empty plaintext buffer (it asserts on it)
       and enough room in the ciphertext buffer; if either is missing we
       simply rekey on a later #queue_write(), which the write task above
       keeps arming for as long as anything is still buffered. */
    if (((0 == queue->rekey_left_bytes) ||
         (0 == GNUNET_TIME_absolute_get_remaining (
            queue->rekey_time).rel_value_us)) &&
        ((0 == queue->pwrite_off) &&
         (queue->cwrite_off + sizeof (struct TCPRekey) <= BUF_SIZE)))
    {
      inject_rekey (queue);
    }
  }
  if ((0 == queue->pwrite_off) && (! queue->finishing) &&
      (GNUNET_YES == 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))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Finishing queue\n");
    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);
}
 
 
static size_t
try_handle_plaintext (struct Queue *queue)
{
  const struct GNUNET_MessageHeader *hdr;
  const struct TCPConfirmationAck *tca;
  const struct TCPBox *box;
  const struct TCPRekey *rekey;
  const struct TCPFinish *fin;
  struct TCPRekey rekeyz;
  struct TCPFinish finz;
  struct GNUNET_ShortHashCode tmac;
  uint16_t type;
  size_t size = 0;
  struct TcpHandshakeAckSignature thas;
  const struct GNUNET_PeerIdentity *my_identity;
  const struct GNUNET_CRYPTO_ChallengeNonceP challenge = queue->challenge;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "try handle plaintext!\n");
 
  hdr = (const struct GNUNET_MessageHeader *) queue->pread_buf;
  if ((sizeof(*hdr) > queue->pread_off))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Handling plaintext, not even a header!\n");
    return 0; /* not even a header */
  }
 
  if ((GNUNET_YES != queue->initial_core_kx_done) && (queue->unverified_size >
                                                      INITIAL_CORE_KX_SIZE))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Already received data of size %lu bigger than KX size %lu!\n",
                queue->unverified_size,
                INITIAL_CORE_KX_SIZE);
    GNUNET_break_op (0);
    queue_finish (queue);
    return 0;
  }
 
  type = ntohs (hdr->type);
  switch (type)
  {
  case GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_CONFIRMATION_ACK:
    tca = (const struct TCPConfirmationAck *) queue->pread_buf;
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "start processing ack\n");
    if (sizeof(*tca) > queue->pread_off)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Handling plaintext size of tca greater than pread offset.\n")
      ;
      return 0;
    }
    if (ntohs (hdr->size) != sizeof(*tca))
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Handling plaintext size does not match message type.\n");
      GNUNET_break_op (0);
      queue_finish (queue);
      return 0;
    }
 
    my_identity = GNUNET_PILS_get_identity (pils);
    GNUNET_assert (my_identity);
 
    thas.purpose.purpose = htonl (
      GNUNET_SIGNATURE_PURPOSE_COMMUNICATOR_TCP_HANDSHAKE_ACK);
    thas.purpose.size = htonl (sizeof(thas));
    thas.sender = tca->sender;
    thas.receiver = *my_identity;
    thas.monotonic_time = tca->monotonic_time;
    thas.challenge = tca->challenge;
 
    if (GNUNET_SYSERR == GNUNET_CRYPTO_eddsa_verify (
          GNUNET_SIGNATURE_PURPOSE_COMMUNICATOR_TCP_HANDSHAKE_ACK,
          &thas,
          &tca->sender_sig,
          &tca->sender.public_key))
    {
      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                  "Verification of signature failed!\n");
      GNUNET_break (0);
      queue_finish (queue);
      return 0;
    }
    if (0 != GNUNET_memcmp (&tca->challenge, &challenge))
    {
      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                  "Challenge in TCPConfirmationAck not correct!\n");
      GNUNET_break (0);
      queue_finish (queue);
      return 0;
    }
 
    queue->handshake_ack_monotime_get = GNUNET_PEERSTORE_iteration_start (
      peerstore,
      "transport_tcp_communicator",
      &queue->target,
      GNUNET_PEERSTORE_TRANSPORT_TCP_COMMUNICATOR_HANDSHAKE_ACK,
      &handshake_ack_monotime_cb,
      queue);
 
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Handling plaintext, ack processed!\n");
 
    if (GNUNET_TRANSPORT_CS_INBOUND ==     queue->cs)
    {
      send_challenge (queue->challenge_received, queue);
      queue->write_task =
        GNUNET_SCHEDULER_add_write_net (GNUNET_TIME_UNIT_FOREVER_REL,
                                        queue->sock,
                                        &queue_write,
                                        queue);
    }
    else if (GNUNET_TRANSPORT_CS_OUTBOUND ==     queue->cs)
    {
      check_and_remove_pending_reversal (queue->address,
                                         queue->address->sa_family,
                                         &queue->target);
    }
 
    queue->initial_core_kx_done = GNUNET_YES;
 
    {
      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,
                                                        UINT16_MAX, /* no MTU */
                                                        GNUNET_TRANSPORT_QUEUE_LENGTH_UNLIMITED,
                                                        0, /* Priority */
                                                        queue->nt,
                                                        queue->cs,
                                                        queue->mq);
 
      GNUNET_free (foreign_addr);
    }
 
    size = ntohs (hdr->size);
    break;
  case GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_BOX:
    /* Special case: header size excludes box itself! */
    box = (const struct TCPBox *) queue->pread_buf;
    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);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Handling plaintext, box processed!\n");
    GNUNET_STATISTICS_update (stats,
                              "# bytes decrypted with BOX",
                              size,
                              GNUNET_NO);
    GNUNET_STATISTICS_update (stats,
                              "# messages decrypted with BOX",
                              1,
                              GNUNET_NO);
    break;
 
  case GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_REKEY:
    rekey = (const struct TCPRekey *) queue->pread_buf;
    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, &rekey->hmac, sizeof(tmac)))
    {
      GNUNET_break_op (0);
      queue_finish (queue);
      return 0;
    }
    do_rekey (queue, rekey);
    size = ntohs (hdr->size);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Handling plaintext, rekey processed!\n");
    GNUNET_STATISTICS_update (stats,
                              "# rekeying successful",
                              1,
                              GNUNET_NO);
    break;
 
  case GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_FINISH:
    fin = (const struct TCPFinish *) queue->pread_buf;
    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, &finz, sizeof(finz), &tmac);
    if (0 != memcmp (&tmac, &fin->hmac, sizeof(tmac)))
    {
      GNUNET_break_op (0);
      queue_finish (queue);
      return 0;
    }
    /* handle FINISH by destroying queue */
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Handling plaintext, finish processed!\n");
    queue_destroy (queue);
    /* NOTE: must return here rather than `break'.  #queue_destroy() releases
       @a queue outright once its backpressure has drained, so neither the
       @e unverified_size accounting below nor #queue_read()'s loop may look
       at it again.  Falling through also hit `GNUNET_assert (0 != size)'
       with @a size still zero -- receiving a perfectly ordinary FINISH thus
       aborted the communicator, taking every other queue down with it. */
    return TCP_QUEUE_GONE;
 
  default:
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Handling plaintext, nothing processed!\n");
    GNUNET_break_op (0);
    queue_finish (queue);
    return 0;
  }
  GNUNET_assert (0 != size);
  if (-1 != queue->unverified_size)
    queue->unverified_size += size;
  return size;
}
 
 
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);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received %zd bytes from TCP queue\n", rcvd);
  if (-1 == rcvd)
  {
    if ((EAGAIN != errno) && (EINTR != errno))
    {
      GNUNET_log_strerror (GNUNET_ERROR_TYPE_DEBUG, "recv");
      queue_destroy (queue);
      return;
    }
    /* try again */
    left = GNUNET_TIME_absolute_get_remaining (queue->timeout);
    if (0 != left.rel_value_us)
    {
      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_destroy (queue);
    return;
  }
  if (0 == rcvd)
  {
    /* Orderly shutdown of connection */
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Socket for queue %p seems to have been closed\n", queue);
    queue_destroy (queue);
    return;
  }
  queue->timeout =
    GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
  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;
    size_t old_pread_off = queue->pread_off;
 
    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 (0 != (done = try_handle_plaintext (queue)))
    {
      if (TCP_QUEUE_GONE == done)
        return;   /* @a queue was destroyed under us, do not touch it */
      /* '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;
      /* The last plaintext was a rekey, abort for now */
      if (GNUNET_YES == queue->rekeyed)
        break;
    }
    /* 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'.
       However, we have to take into account that the plaintext buffer may have
       already contained data and not jumped too far ahead in the ciphertext.
       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 - old_pread_off;
      queue->rekeyed = GNUNET_NO;
      queue->pread_off = 0;
    }
    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_destroy (queue);
}
 
 
static struct sockaddr *
tcp_address_to_sockaddr_numeric_v6 (socklen_t *sock_len,
                                    struct sockaddr_in6 v6,
                                    unsigned int port)
{
  struct sockaddr *in;
 
  v6.sin6_family = AF_INET6;
  v6.sin6_port = htons ((uint16_t) port);
#if HAVE_SOCKADDR_IN_SIN_LEN
  v6.sin6_len = sizeof(struct sockaddr_in6);
#endif
  v6.sin6_flowinfo = 0;
  v6.sin6_scope_id = 0;
  in = GNUNET_memdup (&v6, sizeof(v6));
  *sock_len = sizeof(struct sockaddr_in6);
 
  return in;
}
 
 
static struct sockaddr *
tcp_address_to_sockaddr_numeric_v4 (socklen_t *sock_len,
                                    struct sockaddr_in v4,
                                    unsigned int port)
{
  struct sockaddr *in;
 
  v4.sin_family = AF_INET;
  v4.sin_port = htons ((uint16_t) port);
#if HAVE_SOCKADDR_IN_SIN_LEN
  v4.sin_len = sizeof(struct sockaddr_in);
#endif
  in = GNUNET_memdup (&v4, sizeof(v4));
  *sock_len = sizeof(struct sockaddr_in);
  return in;
}
 
 
static struct PortOnlyIpv4Ipv6 *
tcp_address_to_sockaddr_port_only (const char *bindto, unsigned int *port)
{
  struct PortOnlyIpv4Ipv6 *po;
  struct sockaddr_in *i4;
  struct sockaddr_in6 *i6;
  socklen_t sock_len_ipv4;
  socklen_t sock_len_ipv6;
 
  /* 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;
  }
 
  po = GNUNET_new (struct PortOnlyIpv4Ipv6);
 
  if (GNUNET_YES == disable_v6)
  {
    i4 = GNUNET_malloc (sizeof(struct sockaddr_in));
    po->addr_ipv4 = tcp_address_to_sockaddr_numeric_v4 (&sock_len_ipv4, *i4,
                                                        *port);
    po->addr_len_ipv4 = sock_len_ipv4;
  }
  else
  {
 
    i4 = GNUNET_malloc (sizeof(struct sockaddr_in));
    po->addr_ipv4 = tcp_address_to_sockaddr_numeric_v4 (&sock_len_ipv4, *i4,
                                                        *port);
    po->addr_len_ipv4 = sock_len_ipv4;
 
    i6 = GNUNET_malloc (sizeof(struct sockaddr_in6));
    po->addr_ipv6 = tcp_address_to_sockaddr_numeric_v6 (&sock_len_ipv6, *i6,
                                                        *port);
 
    po->addr_len_ipv6 = sock_len_ipv6;
 
    GNUNET_free (i6);
  }
 
  GNUNET_free (i4);
 
  return po;
}
 
 
static char *
extract_address (const char *bindto)
{
  char *addr;
  char *start;
  char *token;
  char *cp;
  char *rest = NULL;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "extract address with bindto %s\n",
              bindto);
 
  if (NULL == bindto)
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "bindto is NULL\n");
 
  cp = GNUNET_strdup (bindto);
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "extract address 2\n");
 
  start = cp;
  if (('[' == *cp) && (']' == cp[strlen (cp) - 1]))
  {
    start++;   /* skip over '['*/
    cp[strlen (cp) - 1] = '\0';  /* eat ']'*/
    addr = GNUNET_strdup (start);
  }
  else
  {
    token = strtok_r (cp, "]", &rest);
    if (strlen (bindto) == strlen (token))
    {
      token = strtok_r (cp, ":", &rest);
      addr = GNUNET_strdup (token);
    }
    else
    {
      token++;
      addr = GNUNET_strdup (token);
    }
  }
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "tcp address: %s\n",
              addr);
  GNUNET_free (cp);
  return addr;
}
 
 
static unsigned int
extract_port (const char *addr_and_port)
{
  unsigned int port;
  char dummy[2];
  char *token;
  char *addr;
  char *colon;
  char *cp;
  char *rest = NULL;
 
  if (NULL != addr_and_port)
  {
    cp = GNUNET_strdup (addr_and_port);
    token = strtok_r (cp, "]", &rest);
    if (strlen (addr_and_port) == strlen (token))
    {
      colon = strrchr (cp, ':');
      if (NULL == colon)
      {
        GNUNET_free (cp);
        return 0;
      }
      addr = colon;
      addr++;
    }
    else
    {
      token = strtok_r (NULL, "]", &rest);
      if (NULL == token)
      {
        GNUNET_free (cp);
        return 0;
      }
      else
      {
        addr = token;
        addr++;
      }
    }
 
 
    if (1 == sscanf (addr, "%u%1s", &port, dummy))
    {
      /* interpreting value as just a PORT number */
      if (port > UINT16_MAX)
      {
        GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                    "Port `%u' invalid: value too large for port\n",
                    port);
        GNUNET_free (cp);
        return 0;
      }
    }
    else
    {
      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                  "BINDTO specification invalid: last ':' not followed by number\n");
      GNUNET_free (cp);
      return 0;
    }
    GNUNET_free (cp);
  }
  else
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "return 0\n");
    /* interpret missing port as 0, aka pick any free one */
    port = 0;
  }
 
  return port;
}
 
 
static struct sockaddr *
tcp_address_to_sockaddr (const char *bindto, socklen_t *sock_len)
{
  struct sockaddr *in;
  unsigned int port;
  struct sockaddr_in v4;
  struct sockaddr_in6 v6;
  char *start;
 
  memset (&v4, 0, sizeof(v4));
  start = extract_address (bindto);
  GNUNET_assert (NULL != start);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "start %s\n",
              start);
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "!bindto %s\n",
              bindto);
 
 
  if (1 == inet_pton (AF_INET, start, &v4.sin_addr))
  {
    port = extract_port (bindto);
 
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "port %u\n",
                port);
 
    in = tcp_address_to_sockaddr_numeric_v4 (sock_len, v4, port);
  }
  else if (1 == inet_pton (AF_INET6, start, &v6.sin6_addr))
  {
    port = extract_port (bindto);
    in = tcp_address_to_sockaddr_numeric_v6 (sock_len, v6, port);
  }
  else
  {
    GNUNET_assert (0);
  }
 
  GNUNET_free (start);
  return in;
}
 
 
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_log (GNUNET_ERROR_TYPE_DEBUG,
              "In MQ send. Queue finishing: %s; write task running: %s\n",
              (GNUNET_YES == queue->finishing) ? "yes" : "no",
              (NULL == queue->write_task) ? "yes" : "no");
  GNUNET_assert (mq == queue->mq);
  queue->mq_awaits_continue = GNUNET_YES;
  if (GNUNET_YES == queue->finishing)
    return; /* this queue is dying, drop msg */
  GNUNET_assert (0 == queue->pwrite_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->pwrite_buf[queue->pwrite_off], &box, sizeof(box));
  queue->pwrite_off += sizeof(box);
  memcpy (&queue->pwrite_buf[queue->pwrite_off], msg, msize);
  queue->pwrite_off += msize;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "%lu bytes of plaintext to send\n", queue->pwrite_off);
  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);
}
 
 
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);
  }
}
 
 
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;
}
 
 
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);
}
 
 
static void
boot_queue (struct Queue *queue)
{
  queue->nt =
    GNUNET_NT_scanner_get_type (is, queue->address, queue->address_len);
  (void) GNUNET_CONTAINER_multihashmap_put (
    queue_map,
    &queue->key,
    queue,
    GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
  GNUNET_STATISTICS_set (stats,
                         "# queues active",
                         GNUNET_CONTAINER_multihashmap_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);
}
 
 
static void
transmit_kx (struct Queue *queue,
             const struct GNUNET_CRYPTO_HpkeEncapsulation *c)
{
  const struct GNUNET_PeerIdentity *my_identity;
  const struct GNUNET_CRYPTO_EddsaPrivateKey *my_private_key;
  struct TcpHandshakeSignature ths;
  struct TCPConfirmation tc;
 
  my_identity = GNUNET_PILS_get_identity (pils);
  my_private_key = GNUNET_PILS_key_ring_get_private_key (key_ring);
  GNUNET_assert ((my_identity) && (my_private_key));
 
  memcpy (queue->cwrite_buf, c, sizeof(*c));
  queue->cwrite_off = sizeof(*c);
  /* 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));
  GNUNET_CRYPTO_random_block (&tc.challenge,
                              sizeof(tc.challenge));
  ths.purpose.purpose = htonl (
    GNUNET_SIGNATURE_PURPOSE_COMMUNICATOR_TCP_HANDSHAKE);
  ths.purpose.size = htonl (sizeof(ths));
  ths.sender = *my_identity;
  ths.receiver = queue->target;
  ths.ephemeral = *c;
  ths.monotonic_time = tc.monotonic_time;
  ths.challenge = tc.challenge;
  GNUNET_CRYPTO_eddsa_sign (my_private_key,
                            &ths,
                            &tc.sender_sig);
  GNUNET_assert (0 ==
                 gcry_cipher_encrypt (queue->out_cipher,
                                      &queue->cwrite_buf[queue->cwrite_off],
                                      sizeof(tc),
                                      &tc,
                                      sizeof(tc)));
  queue->challenge = tc.challenge;
  queue->cwrite_off += sizeof(tc);
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "handshake written\n");
}
 
 
static void
start_initial_kx_out (struct Queue *queue)
{
  struct GNUNET_CRYPTO_HpkeEncapsulation c;
  struct GNUNET_ShortHashCode k;
 
  GNUNET_CRYPTO_hpke_elligator_kem_encaps (&queue->target_hpke_key,
                                           &c, &k);
  setup_out_cipher (queue, &k);
  transmit_kx (queue, &c);
}
 
 
static void
handshake_monotime_store_cb (void *cls, int success)
{
  struct Queue *queue = cls;
  if (GNUNET_OK != success)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Failed to store handshake monotonic time in PEERSTORE!\n");
  }
  queue->handshake_monotime_sc = NULL;
  GNUNET_PEERSTORE_iteration_next (queue->handshake_ack_monotime_get, 1);
}
 
 
static void
handshake_monotime_cb (void *cls,
                       const struct GNUNET_PEERSTORE_Record *record,
                       const char *emsg)
{
  struct Queue *queue = cls;
  struct GNUNET_TIME_AbsoluteNBO *mtbe;
  struct GNUNET_TIME_Absolute mt;
  const struct GNUNET_PeerIdentity *pid;
  struct GNUNET_TIME_AbsoluteNBO *handshake_monotonic_time;
 
  (void) emsg;
 
  handshake_monotonic_time = &queue->handshake_monotonic_time;
  pid = &queue->target;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "tcp handshake with us %s\n",
              GNUNET_i2s (GNUNET_PILS_get_identity (pils)));
  if (NULL == record)
  {
    queue->handshake_monotime_get = NULL;
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "tcp handshake from peer %s\n",
              GNUNET_i2s (pid));
  if (sizeof(*mtbe) != record->value_size)
  {
    GNUNET_PEERSTORE_iteration_next (queue->handshake_ack_monotime_get, 1);
    GNUNET_break (0);
    return;
  }
  mtbe = record->value;
  mt = GNUNET_TIME_absolute_ntoh (*mtbe);
  if (mt.abs_value_us > GNUNET_TIME_absolute_ntoh (
        queue->handshake_monotonic_time).abs_value_us)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Queue from %s dropped, handshake monotime in the past\n",
                GNUNET_i2s (&queue->target));
    GNUNET_break (0);
    GNUNET_PEERSTORE_iteration_stop (queue->handshake_ack_monotime_get);
    queue->handshake_ack_monotime_get = NULL;
    queue_finish (queue);
    return;
  }
  queue->handshake_monotime_sc = GNUNET_PEERSTORE_store (peerstore,
                                                         "transport_tcp_communicator",
                                                         pid,
                                                         GNUNET_PEERSTORE_TRANSPORT_TCP_COMMUNICATOR_HANDSHAKE,
                                                         handshake_monotonic_time,
                                                         sizeof(*
                                                                handshake_monotonic_time),
                                                         GNUNET_TIME_UNIT_FOREVER_ABS,
                                                         GNUNET_PEERSTORE_STOREOPTION_REPLACE,
                                                         &
                                                         handshake_monotime_store_cb,
                                                         queue);
}
 
 
static int
decrypt_and_check_tc (struct Queue *queue,
                      struct TCPConfirmation *tc,
                      char *ibuf)
{
  const struct GNUNET_PeerIdentity *my_identity;
  struct TcpHandshakeSignature ths;
  enum GNUNET_GenericReturnValue ret;
 
  my_identity = GNUNET_PILS_get_identity (pils);
  GNUNET_assert (my_identity);
 
  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_PURPOSE_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;
  ths.challenge = tc->challenge;
  ret = GNUNET_CRYPTO_eddsa_verify (
    GNUNET_SIGNATURE_PURPOSE_COMMUNICATOR_TCP_HANDSHAKE,
    &ths,
    &tc->sender_sig,
    &tc->sender.public_key);
  if (GNUNET_YES == ret)
    queue->handshake_monotime_get =
      GNUNET_PEERSTORE_iteration_start (peerstore,
                                        "transport_tcp_communicator",
                                        &queue->target,
                                        GNUNET_PEERSTORE_TRANSPORT_TCP_COMMUNICATOR_HANDSHAKE,
                                        &handshake_monotime_cb,
                                        queue);
  return ret;
}
 
 
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);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received %lu bytes to write in buffer of size %lu for KX from queue %p (expires in %"
              PRIu64 ")\n",
              rcvd, BUF_SIZE - queue->cread_off, queue, left.rel_value_us);
  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;
  }
  if (0 == rcvd)
  {
    /* Orderly shutdown of connection */
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Socket for queue %p seems to have been closed\n", queue);
    queue_destroy (queue);
    return;
  }
  queue->cread_off += rcvd;
  if (queue->cread_off < INITIAL_KX_SIZE)
  {
    /* read more */
    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                "%lu/%lu bytes of KX read. Rescheduling...\n",
                queue->cread_off, INITIAL_KX_SIZE);
    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_elligator (
    (const struct GNUNET_CRYPTO_HpkeEncapsulation*)
    queue->cread_buf,
    queue);
  if (GNUNET_OK != decrypt_and_check_tc (queue, &tc, queue->cread_buf))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "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;
  }
  send_challenge (tc.challenge, queue);
  queue->write_task =
    GNUNET_SCHEDULER_add_write_net (GNUNET_TIME_UNIT_FOREVER_REL,
                                    queue->sock,
                                    &queue_write,
                                    queue);
 
  /* update queue timeout */
  queue->timeout =
    GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
  /* prepare to continue with regular read task immediately */
  memmove (queue->cread_buf,
           &queue->cread_buf[INITIAL_KX_SIZE],
           queue->cread_off - (INITIAL_KX_SIZE));
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "cread_off is %lu bytes before adjusting\n",
              queue->cread_off);
  queue->cread_off -= INITIAL_KX_SIZE;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "cread_off set to %lu bytes\n",
              queue->cread_off);
  queue->read_task = GNUNET_SCHEDULER_add_now (&queue_read, queue);
}
 
 
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;
  GNUNET_SCHEDULER_TaskCallback read_task;
 
  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);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Proto received %lu bytes for KX\n", rcvd);
  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;
  }
  if (0 == rcvd)
  {
    /* Orderly shutdown of connection */
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Socket for proto queue %p seems to have been closed\n", pq);
    free_proto_queue (pq);
    return;
  }
  pq->ibuf_off += rcvd;
  if (sizeof (struct TCPNATProbeMessage) == pq->ibuf_off)
  {
    struct TCPNATProbeMessage *pm = (struct TCPNATProbeMessage *) pq->ibuf;
 
    check_and_remove_pending_reversal (pq->address, pq->address->sa_family,
                                       &pm->clientIdentity);
 
    queue = GNUNET_new (struct Queue);
    queue->target = pm->clientIdentity;
    eddsa_pub_to_hpke_key (&queue->target.public_key,
                           &queue->target_hpke_key);
    queue->cs = GNUNET_TRANSPORT_CS_OUTBOUND;
    read_task = &queue_read_kx;
  }
  else 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;
  }
  else
  {
    /* we got all the data, let's find out who we are talking to! */
    queue = GNUNET_new (struct Queue);
    setup_in_cipher_elligator (
      (const struct GNUNET_CRYPTO_HpkeEncapsulation *) 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 (pq->address, pq->address_len));
      gcry_cipher_close (queue->in_cipher);
      GNUNET_free (queue);
      free_proto_queue (pq);
      return;
    }
    queue->target = tc.sender;
    eddsa_pub_to_hpke_key (&queue->target.public_key,
                           &queue->target_hpke_key);
    queue->cs = GNUNET_TRANSPORT_CS_INBOUND;
    read_task = &queue_read;
  }
  queue->address = pq->address; /* steals reference */
  queue->address_len = pq->address_len;
  queue->listen_sock = pq->listen_sock;
  queue->sock = pq->sock;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "created queue with target %s\n",
              GNUNET_i2s (&queue->target));
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "start kx proto\n");
 
  start_initial_kx_out (queue);
  boot_queue (queue);
  queue->read_task =
    GNUNET_SCHEDULER_add_read_net (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
                                   queue->sock,
                                   read_task,
                                   queue);
  queue->write_task =
    GNUNET_SCHEDULER_add_write_net (GNUNET_TIME_UNIT_FOREVER_REL,
                                    queue->sock,
                                    &queue_write,
                                    queue);
  // TODO To early! Move it somewhere else.
  // send_challenge (tc.challenge, queue);
  queue->challenge_received = tc.challenge;
 
  GNUNET_CONTAINER_DLL_remove (proto_head, proto_tail, pq);
  GNUNET_free (pq);
}
 
 
static struct ProtoQueue *
create_proto_queue (struct GNUNET_NETWORK_Handle *sock,
                    struct sockaddr *in,
                    socklen_t addrlen)
{
  struct ProtoQueue *pq = GNUNET_new (struct ProtoQueue);
 
  if (NULL == sock)
  {
    // sock = GNUNET_CONNECTION_create_from_sockaddr (AF_INET, addr, addrlen);
    sock = GNUNET_NETWORK_socket_create (in->sa_family, SOCK_STREAM, 0);
    if (NULL == sock)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                  "socket(%d) failed: %s",
                  in->sa_family,
                  strerror (errno));
      GNUNET_free (in);
      GNUNET_free (pq);
      return NULL;
    }
    if ((GNUNET_OK != GNUNET_NETWORK_socket_connect (sock, in, addrlen)) &&
        (errno != EINPROGRESS))
    {
      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                  "connect to `%s' failed: %s",
                  GNUNET_a2s (in, addrlen),
                  strerror (errno));
      GNUNET_NETWORK_socket_close (sock);
      GNUNET_free (in);
      GNUNET_free (pq);
      return NULL;
    }
  }
  pq->address_len = addrlen;
  pq->address = in;
  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);
 
  return pq;
}
 
 
static void
listen_cb (void *cls)
{
  struct sockaddr_storage in;
  socklen_t addrlen;
  struct GNUNET_NETWORK_Handle *sock;
  struct ListenTask *lt;
  struct sockaddr *in_addr;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "listen_cb\n");
 
  lt = cls;
 
  lt->listen_task = NULL;
  GNUNET_assert (NULL != lt->listen_sock);
  addrlen = sizeof(in);
  memset (&in, 0, sizeof(in));
  sock = GNUNET_NETWORK_socket_accept (lt->listen_sock,
                                       (struct sockaddr*) &in,
                                       &addrlen);
  if ((NULL == sock) && ((EMFILE == errno) || (ENFILE == errno)))
    return; /* system limit reached, wait until connection goes down */
  lt->listen_task = GNUNET_SCHEDULER_add_read_net (GNUNET_TIME_UNIT_FOREVER_REL,
                                                   lt->listen_sock,
                                                   &listen_cb,
                                                   lt);
  if ((NULL == sock) && ((EAGAIN == errno) || (ENOBUFS == errno)))
    return;
  if (NULL == sock)
  {
    GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING, "accept");
    return;
  }
  in_addr = GNUNET_memdup (&in, addrlen);
  create_proto_queue (sock, in_addr, addrlen);
}
 
 
static void
try_connection_reversal (void *cls,
                         const struct sockaddr *addr,
                         socklen_t addrlen)
{
  const struct GNUNET_PeerIdentity *my_identity;
  struct TCPNATProbeMessage pm;
  struct ProtoQueue *pq;
  struct sockaddr *in_addr;
  (void) cls;
 
  my_identity = GNUNET_PILS_get_identity (pils);
  GNUNET_assert (my_identity);
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "addr->sa_family %d\n",
              addr->sa_family);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Try to connect back\n");
  in_addr = GNUNET_memdup (addr, addrlen);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "in_addr->sa_family %d\n",
              in_addr->sa_family);
  pq = create_proto_queue (NULL, in_addr, addrlen);
  if (NULL != pq)
  {
    pm.header.size = htons (sizeof(struct TCPNATProbeMessage));
    pm.header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_TCP_NAT_PROBE);
    pm.clientIdentity = *my_identity;
    memcpy (pq->write_buf, &pm, sizeof(struct TCPNATProbeMessage));
    pq->write_off = sizeof(struct TCPNATProbeMessage);
    pq->write_task = GNUNET_SCHEDULER_add_write_net (PROTO_QUEUE_TIMEOUT,
                                                     pq->sock,
                                                     &proto_queue_write,
                                                     pq);
  }
  else
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Couldn't create ProtoQueue for sending TCPNATProbeMessage\n");
  }
}
 
 
static void
pending_reversal_timeout (void *cls)
{
  struct PendingReversal *pending_reversal = cls;
  struct GNUNET_HashCode key;
 
  pending_reversal->timeout_task = NULL;
  pending_reversal_key (pending_reversal->in,
                        &pending_reversal->target,
                        &key);
  if (GNUNET_NO == GNUNET_CONTAINER_multihashmap_remove (pending_reversals,
                                                         &key,
                                                         pending_reversal))
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "No pending reversal found for address %s\n",
                GNUNET_a2s (pending_reversal->in,
                            sizeof (struct sockaddr)));
  GNUNET_free (pending_reversal->in);
  GNUNET_free (pending_reversal);
}
 
 
static int
mq_init (void *cls, const struct GNUNET_PeerIdentity *peer, const char *address)
{
  struct sockaddr *in;
  socklen_t in_len = 0;
  const char *path;
  struct sockaddr_in *v4;
  struct sockaddr_in6 *v6;
  unsigned int is_natd = GNUNET_NO;
  struct GNUNET_HashCode key;
  struct GNUNET_HashCode queue_map_key;
  struct GNUNET_HashContext *hsh;
  struct Queue *queue;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Connecting to %s at %s\n",
              GNUNET_i2s (peer),
              address);
  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);
 
  if (NULL == in)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Failed to setup TCP socket address\n");
    return GNUNET_SYSERR;
  }
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "in %s\n",
              GNUNET_a2s (in, in_len));
 
  if (AF_INET6 == in->sa_family)
  {
    const struct sockaddr_in6 *v6 = (const struct sockaddr_in6 *) in;
 
    /* A link-local address is only meaningful together with the interface
       it was learned on, and that scope does not survive being carried to
       another host in a HELLO.  connect() to one with a zero scope id fails
       with EINVAL every single time, so trying is pure noise: peers gossip
       these by the dozen and each attempt logged a WARNING. */
    if (IN6_IS_ADDR_LINKLOCAL (&v6->sin6_addr) && (0 == v6->sin6_scope_id))
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Skipping unusable link-local address `%s' for `%s'\n",
                  address,
                  GNUNET_i2s (peer));
      GNUNET_free (in);
      return GNUNET_SYSERR;
    }
  }
 
  hsh = GNUNET_CRYPTO_hash_context_start ();
  GNUNET_CRYPTO_hash_context_read (hsh, address, strlen (address));
  GNUNET_CRYPTO_hash_context_read (hsh, peer, sizeof (*peer));
  GNUNET_CRYPTO_hash_context_finish (hsh, &queue_map_key);
  queue = GNUNET_CONTAINER_multihashmap_get (queue_map, &queue_map_key);
 
  if (NULL != queue)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Queue for %s already exists or is in construction\n", address);
    GNUNET_free (in);
    return GNUNET_NO;
  }
  switch (in->sa_family)
  {
  case AF_INET:
    v4 = (struct sockaddr_in *) in;
    if (0 == v4->sin_port)
    {
      is_natd = GNUNET_YES;
      pending_reversal_key (in, peer, &key);
      if (GNUNET_YES == GNUNET_CONTAINER_multihashmap_contains (
            pending_reversals,
            &key))
      {
        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                    "There is already a request reversal for `%s' at `%s'\n",
                    GNUNET_i2s (peer),
                    address);
        GNUNET_free (in);
        return GNUNET_SYSERR;
      }
    }
    break;
 
  case AF_INET6:
    if (GNUNET_YES == disable_v6)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "IPv6 disabled, skipping %s\n", address);
      GNUNET_free (in);
      return GNUNET_SYSERR;
    }
    v6 = (struct sockaddr_in6 *) in;
    if (0 == v6->sin6_port)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                  "Request reversal for `%s' at `%s' not possible for an IPv6 address\n",
                  GNUNET_i2s (peer),
                  address);
      GNUNET_free (in);
      return GNUNET_SYSERR;
    }
    break;
 
  default:
    GNUNET_assert (0);
  }
 
  if (GNUNET_YES == is_natd)
  {
    struct sockaddr_in local_sa;
    struct PendingReversal *pending_reversal;
 
    memset (&local_sa, 0, sizeof(local_sa));
    local_sa.sin_family = AF_INET;
    local_sa.sin_port = htons (bind_port);
    /* We leave sin_address at 0, let the kernel figure it out,
       even if our bind() is more specific.  (May want to reconsider
       later.) */
    if (GNUNET_OK != GNUNET_NAT_request_reversal (nat, &local_sa, v4))
    {
      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                  "request reversal for `%s' at `%s' failed\n",
                  GNUNET_i2s (peer),
                  address);
      GNUNET_free (in);
      return GNUNET_SYSERR;
    }
    pending_reversal = GNUNET_new (struct PendingReversal);
    pending_reversal->in = in;
    GNUNET_assert (GNUNET_OK ==
                   GNUNET_CONTAINER_multihashmap_put (pending_reversals,
                                                      &key,
                                                      pending_reversal,
                                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
    pending_reversal->target = *peer;
    pending_reversal->timeout_task = GNUNET_SCHEDULER_add_delayed (
      NAT_TIMEOUT,
      &pending_reversal_timeout,
      pending_reversal);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Created NAT WAIT connection to `%s' at `%s'\n",
                GNUNET_i2s (peer),
                GNUNET_a2s (in, sizeof (struct sockaddr)));
  }
  else
  {
    struct GNUNET_NETWORK_Handle *sock;
 
    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)) &&
        (errno != EINPROGRESS))
    {
      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;
    eddsa_pub_to_hpke_key (&queue->target.public_key, &queue->target_hpke_key);
    queue->key = queue_map_key;
    queue->address = in;
    queue->address_len = in_len;
    queue->sock = sock;
    queue->cs = GNUNET_TRANSPORT_CS_OUTBOUND;
    boot_queue (queue);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "booted queue with target %s\n",
                GNUNET_i2s (&queue->target));
    // queue->mq_awaits_continue = GNUNET_YES;
    queue->read_task =
      GNUNET_SCHEDULER_add_read_net (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
                                     queue->sock,
                                     &queue_read_kx,
                                     queue);
 
 
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "start kx mq_init\n");
 
    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;
}
 
 
static int
get_lt_delete_it (void *cls,
                  const struct GNUNET_HashCode *key,
                  void *value)
{
  struct ListenTask *lt = value;
 
  (void) cls;
  (void) key;
  if (NULL != lt->listen_task)
  {
    GNUNET_SCHEDULER_cancel (lt->listen_task);
    lt->listen_task = NULL;
  }
  if (NULL != lt->listen_sock)
  {
    GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (lt->listen_sock));
    lt->listen_sock = NULL;
  }
  GNUNET_free (lt);
  return GNUNET_OK;
}
 
 
static int
get_queue_delete_it (void *cls,
                     const struct GNUNET_HashCode *target,
                     void *value)
{
  struct Queue *queue = value;
 
  (void) cls;
  (void) target;
  queue_destroy (queue);
  return GNUNET_OK;
}
 
 
static void
do_shutdown (void *cls)
{
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Shutdown %s!\n",
              shutdown_running ? "running" : "not running");
 
  if (GNUNET_YES == shutdown_running)
    return;
  else
    shutdown_running = GNUNET_YES;
 
  while (NULL != proto_head)
    free_proto_queue (proto_head);
  if (NULL != nat)
  {
    GNUNET_NAT_unregister (nat);
    nat = NULL;
  }
  GNUNET_CONTAINER_multihashmap_iterate (pending_reversals,
                                         &pending_reversals_delete_it, NULL);
  GNUNET_CONTAINER_multihashmap_destroy (pending_reversals);
  GNUNET_CONTAINER_multihashmap_iterate (lt_map, &get_lt_delete_it, NULL);
  GNUNET_CONTAINER_multihashmap_destroy (lt_map);
  GNUNET_CONTAINER_multihashmap_iterate (queue_map, &get_queue_delete_it, NULL);
  GNUNET_CONTAINER_multihashmap_destroy (queue_map);
  if (NULL != ch)
  {
    GNUNET_TRANSPORT_communicator_address_remove_all (ch);
    GNUNET_TRANSPORT_communicator_disconnect (ch);
    ch = NULL;
  }
  if (NULL != stats)
  {
    GNUNET_STATISTICS_destroy (stats, GNUNET_YES);
    stats = NULL;
  }
  if (NULL != is)
  {
    GNUNET_NT_scanner_done (is);
    is = NULL;
  }
  if (NULL != pils)
  {
    GNUNET_PILS_disconnect (pils);
    pils = NULL;
  }
  if (NULL != key_ring)
  {
    GNUNET_PILS_destroy_key_ring (key_ring);
    key_ring = NULL;
  }
  if (NULL != peerstore)
  {
    GNUNET_PEERSTORE_disconnect (peerstore);
    peerstore = NULL;
  }
  if (NULL != resolve_request_handle)
  {
    GNUNET_RESOLVER_request_cancel (resolve_request_handle);
    resolve_request_handle = NULL;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Shutdown done!\n");
}
 
 
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);
}
 
 
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;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "nat address cb %s %s\n",
              add_remove ? "add" : "remove",
              GNUNET_a2s (addr, addrlen));
 
  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;
  }
}
 
 
static void
add_addr (struct sockaddr *in, socklen_t in_len)
{
 
  struct Addresses *saddrs;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "add address %s\n",
              GNUNET_a2s (in, in_len));
 
  saddrs = GNUNET_new (struct Addresses);
  saddrs->addr = in;
  saddrs->addr_len = in_len;
  GNUNET_CONTAINER_DLL_insert (addrs_head, addrs_tail, saddrs);
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "after add address %s\n",
              GNUNET_a2s (in, in_len));
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "add address %s\n",
              GNUNET_a2s (saddrs->addr, saddrs->addr_len));
 
  addrs_lens++;
}
 
 
static int
init_socket (struct sockaddr *addr,
             socklen_t in_len)
{
  struct sockaddr_storage in_sto;
  socklen_t sto_len;
  struct GNUNET_NETWORK_Handle *listen_sock;
  struct ListenTask *lt;
  int sockfd;
  struct GNUNET_HashCode h_sock;
 
  if (NULL == addr)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Address is NULL.\n");
    return GNUNET_SYSERR;
  }
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "address %s\n",
              GNUNET_a2s (addr, in_len));
 
  listen_sock =
    GNUNET_NETWORK_socket_create (addr->sa_family, SOCK_STREAM, IPPROTO_TCP);
  if (NULL == listen_sock)
  {
    GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR, "socket");
    return GNUNET_SYSERR;
  }
 
  if (GNUNET_OK != GNUNET_NETWORK_socket_bind (listen_sock, addr, in_len))
  {
    GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR, "bind");
    GNUNET_NETWORK_socket_close (listen_sock);
    listen_sock = NULL;
    return GNUNET_SYSERR;
  }
 
  if (GNUNET_OK !=
      GNUNET_NETWORK_socket_listen (listen_sock,
                                    5))
  {
    GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR,
                         "listen");
    GNUNET_NETWORK_socket_close (listen_sock);
    listen_sock = NULL;
    return GNUNET_SYSERR;
  }
 
  /* 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, addr, in_len);
    sto_len = in_len;
  }
 
  // addr = (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));
  if (NULL == stats)
    stats = GNUNET_STATISTICS_create ("communicator-tcp", cfg);
 
  if (NULL == is)
    is = GNUNET_NT_scanner_init ();
 
  /* start listening */
 
  lt = GNUNET_new (struct ListenTask);
  lt->listen_sock = listen_sock;
 
  lt->listen_task = GNUNET_SCHEDULER_add_read_net (GNUNET_TIME_UNIT_FOREVER_REL,
                                                   listen_sock,
                                                   &listen_cb,
                                                   lt);
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "creating hash\n");
  sockfd = GNUNET_NETWORK_get_fd (lt->listen_sock);
  GNUNET_CRYPTO_hash (&sockfd,
                      sizeof(int),
                      &h_sock);
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "creating map\n");
  if (NULL == lt_map)
    lt_map = GNUNET_CONTAINER_multihashmap_create (2, GNUNET_NO);
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "creating map entry\n");
  GNUNET_assert (GNUNET_OK ==
                 GNUNET_CONTAINER_multihashmap_put (lt_map,
                                                    &h_sock,
                                                    lt,
                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "map entry created\n");
 
  if (NULL == queue_map)
    queue_map = GNUNET_CONTAINER_multihashmap_create (10, GNUNET_NO);
 
  if (NULL == ch)
    ch = GNUNET_TRANSPORT_communicator_connect (cfg,
                                                COMMUNICATOR_CONFIG_SECTION,
                                                COMMUNICATOR_ADDRESS_PREFIX,
                                                GNUNET_TRANSPORT_CC_RELIABLE,
                                                &mq_init,
                                                NULL,
                                                &enc_notify_cb,
                                                NULL,
                                                NULL);
 
  if (NULL == ch)
  {
    GNUNET_break (0);
    if (NULL != resolve_request_handle)
      GNUNET_RESOLVER_request_cancel (resolve_request_handle);
    GNUNET_SCHEDULER_shutdown ();
    return GNUNET_SYSERR;
  }
 
  add_addr (addr, in_len);
  return GNUNET_OK;
 
}
 
 
static void
nat_register ()
{
  struct sockaddr **saddrs;
  socklen_t *saddr_lens;
  int i;
  size_t len;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "starting nat register!\n");
  len = 0;
  i = 0;
  saddrs = GNUNET_malloc ((addrs_lens) * sizeof(struct sockaddr *));
  saddr_lens = GNUNET_malloc ((addrs_lens) * sizeof(socklen_t));
  for (struct Addresses *pos = addrs_head; NULL != pos; pos = pos->next)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "registering address %s\n",
                GNUNET_a2s (pos->addr, pos->addr_len));
 
    saddr_lens[i] = pos->addr_len;
    len += saddr_lens[i];
    saddrs[i] = GNUNET_memdup (pos->addr, saddr_lens[i]);
    i++;
  }
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "registering addresses %lu %lu %lu %lu\n",
              (addrs_lens) * sizeof(struct sockaddr *),
              (addrs_lens) * sizeof(socklen_t),
              len,
              sizeof(COMMUNICATOR_CONFIG_SECTION));
  nat = GNUNET_NAT_register (cfg,
                             COMMUNICATOR_CONFIG_SECTION,
                             IPPROTO_TCP,
                             addrs_lens,
                             (const struct sockaddr **) saddrs,
                             saddr_lens,
                             &nat_address_cb,
                             try_connection_reversal,
                             NULL /* closure */);
  for (i = addrs_lens - 1; i >= 0; i--)
    GNUNET_free (saddrs[i]);
  GNUNET_free (saddrs);
  GNUNET_free (saddr_lens);
 
  if (NULL == nat)
  {
    GNUNET_break (0);
    if (NULL != resolve_request_handle)
      GNUNET_RESOLVER_request_cancel (resolve_request_handle);
    GNUNET_SCHEDULER_shutdown ();
  }
}
 
 
static void
init_socket_resolv (void *cls,
                    const struct sockaddr *addr,
                    socklen_t in_len)
{
  struct sockaddr_in *v4;
  struct sockaddr_in6 *v6;
  struct sockaddr *in;
 
  (void) cls;
  if (NULL != addr)
  {
    if (AF_INET == addr->sa_family)
    {
      v4 = (struct sockaddr_in *) addr;
      in = tcp_address_to_sockaddr_numeric_v4 (&in_len, *v4, bind_port);// _global);
    }
    else if (AF_INET6 == addr->sa_family)
    {
      v6 = (struct sockaddr_in6 *) addr;
      in = tcp_address_to_sockaddr_numeric_v6 (&in_len, *v6, bind_port);// _global);
    }
    else
    {
      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                  "Address family %u not suitable (not AF_INET %u nor AF_INET6 %u \n",
                  addr->sa_family,
                  AF_INET,
                  AF_INET6);
      return;
    }
    init_socket (in, in_len);
  }
  else
  {
    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                "Address is NULL. This might be an error or the resolver finished resolving.\n");
    if (NULL == addrs_head)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                  "Resolver finished resolving, but we do not listen to an address!.\n");
      return;
    }
    nat_register ();
  }
}
 
 
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_in v4;
  struct sockaddr_in6 v6;
  char *start;
  unsigned int port;
  char dummy[2];
  char *rest = NULL;
  struct PortOnlyIpv4Ipv6 *po;
  socklen_t addr_len_ipv4;
  socklen_t addr_len_ipv6;
 
  (void) cls;
 
  pending_reversals = GNUNET_CONTAINER_multihashmap_create (16, GNUNET_NO);
  memset (&v4,0,sizeof(struct sockaddr_in));
  memset (&v6,0,sizeof(struct sockaddr_in6));
  cfg = c;
  if (GNUNET_OK !=
      GNUNET_CONFIGURATION_get_value_string (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;
  }
  if (GNUNET_OK !=
      GNUNET_CONFIGURATION_get_value_time (cfg,
                                           COMMUNICATOR_CONFIG_SECTION,
                                           "REKEY_INTERVAL",
                                           &rekey_interval))
  {
    rekey_interval = DEFAULT_REKEY_INTERVAL;
  }
  if (GNUNET_OK !=
      GNUNET_CONFIGURATION_get_value_number (cfg,
                                             COMMUNICATOR_CONFIG_SECTION,
                                             "REKEY_MAX_BYTES",
                                             &rekey_max_bytes))
  {
    rekey_max_bytes = REKEY_MAX_BYTES;
  }
  disable_v6 = GNUNET_NO;
  if ((GNUNET_NO == GNUNET_NETWORK_test_pf (PF_INET6)) ||
      (GNUNET_YES ==
       GNUNET_CONFIGURATION_get_value_yesno (cfg,
                                             COMMUNICATOR_CONFIG_SECTION,
                                             "DISABLE_V6")))
  {
    disable_v6 = GNUNET_YES;
  }
  key_ring = GNUNET_PILS_create_key_ring (cfg, NULL, NULL);
  GNUNET_assert (NULL != key_ring);
  pils = GNUNET_PILS_connect (cfg, NULL, NULL);
  GNUNET_assert (NULL != pils);
  peerstore = GNUNET_PEERSTORE_connect (cfg);
  if (NULL == peerstore)
  {
    GNUNET_free (bindto);
    GNUNET_break (0);
    GNUNET_SCHEDULER_shutdown ();
    return;
  }
 
  GNUNET_SCHEDULER_add_shutdown (&do_shutdown, NULL);
 
  if (1 == sscanf (bindto, "%u%1s", &bind_port, dummy))
  {
    po = tcp_address_to_sockaddr_port_only (bindto, &bind_port);
    addr_len_ipv4 = po->addr_len_ipv4;
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "address po %s\n",
                GNUNET_a2s (po->addr_ipv4, addr_len_ipv4));
    if (NULL != po->addr_ipv4)
    {
      init_socket (po->addr_ipv4, addr_len_ipv4);
    }
    if (NULL != po->addr_ipv6)
    {
      addr_len_ipv6 = po->addr_len_ipv6;
      init_socket (po->addr_ipv6, addr_len_ipv6);
    }
    GNUNET_free (po);
    nat_register ();
    GNUNET_free (bindto);
    return;
  }
 
  start = extract_address (bindto);
  // FIXME: check for NULL == start...
  if (1 == inet_pton (AF_INET, start, &v4.sin_addr))
  {
    bind_port = extract_port (bindto);
 
    in = tcp_address_to_sockaddr_numeric_v4 (&in_len, v4, bind_port);
    init_socket (in, in_len);
    nat_register ();
    GNUNET_free (start);
    GNUNET_free (bindto);
    return;
  }
 
  if (1 == inet_pton (AF_INET6, start, &v6.sin6_addr))
  {
    bind_port = extract_port (bindto);
    in = tcp_address_to_sockaddr_numeric_v6 (&in_len, v6, bind_port);
    init_socket (in, in_len);
    nat_register ();
    GNUNET_free (start);
    GNUNET_free (bindto);
    return;
  }
 
  bind_port = extract_port (bindto);
  resolve_request_handle = GNUNET_RESOLVER_ip_get (strtok_r (bindto,
                                                             ":",
                                                             &rest),
                                                   AF_UNSPEC,
                                                   GNUNET_TIME_UNIT_MINUTES,
                                                   &init_socket_resolv,
                                                   &port);
 
  GNUNET_free (bindto);
  GNUNET_free (start);
}
 
 
int
main (int argc, char *const *argv)
{
  static const struct GNUNET_GETOPT_CommandLineOption options[] = {
    GNUNET_GETOPT_OPTION_END
  };
  int ret;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Starting tcp communicator\n");
 
  ret = (GNUNET_OK ==
         GNUNET_PROGRAM_run (GNUNET_OS_project_data_gnunet (),
                             argc,
                             argv,
                             "gnunet-communicator-tcp",
                             _ ("GNUnet TCP communicator"),
                             options,
                             &run,
                             NULL))
        ? 0
        : 1;
  return ret;
}
 
 
/* end of gnunet-communicator-tcp.c */