more work on TCP communicator

1 files changed, 521 insertions, 62 deletions

diff --git a/src/transport/gnunet-communicator-tcp.c b/src/transport/gnunet-communicator-tcp.c
index 6e3faa44a..5a397c296 100644
--- a/src/transport/gnunet-communicator-tcp.c
+++ b/src/transport/gnunet-communicator-tcp.c
@@ -1,6 +1,6 @@
 /*
      This file is part of GNUnet
-     Copyright (C) 2010-2014, 2018 GNUnet e.V.
+     Copyright (C) 2010-2014, 2018, 2019 GNUnet e.V.
 
      GNUnet is free software: you can redistribute it and/or modify it
      under the terms of the GNU Affero General Public License as published
@@ -33,6 +33,7 @@
 #include "platform.h"
 #include "gnunet_util_lib.h"
 #include "gnunet_protocols.h"
+#include "gnunet_signatures.h"
 #include "gnunet_constants.h"
 #include "gnunet_nt_lib.h"
 #include "gnunet_statistics_service.h"
@@ -49,6 +50,23 @@
 #define DEFAULT_MAX_QUEUE_LENGTH 8
 
 /**
+ * Size of our IO buffers for ciphertext data. Must be at
+ * least UINT_MAX + sizeof (struct TCPBox).
+ */
+#define BUF_SIZE (2 * 64 * 1024 + sizeof (struct TCPBox))
+
+/**
+ * How often do we rekey based on time (at least)
+ */ 
+#define REKEY_TIME_INTERVAL GNUNET_TIME_UNIT_DAYS
+
+/**
+ * How often do we rekey based on number of bytes transmitted?
+ * (additionally randomized).
+ */ 
+#define REKEY_MAX_BYTES (1024LLU * 1024 * 1024 * 4LLU)
+
+/**
  * Address prefix used by the communicator.
  */
 #define COMMUNICATOR_ADDRESS_PREFIX "tcp"
@@ -60,19 +78,6 @@
 
 GNUNET_NETWORK_STRUCT_BEGIN
 
-/**
- * TCP initial bytes on the wire (in either direction), used to 
- * establish a shared secret.
- */
-struct TCPHandshake
-{
-  /**
-   * First bytes: ephemeral key for KX.
-   */
-  struct GNUNET_CRYPTO_EcdhePublicKey ephemeral;
-
-};
-
 
 /**
  * Signature we use to verify that the ephemeral key was really chosen by
@@ -139,7 +144,10 @@ struct TCPBox
 {
   
   /**
-   * Type is #GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_BOX.
+   * Type is #GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_BOX.  Warning: the
+   * header size EXCLUDES the size of the `struct TCPBox`. We usually
+   * never do this, but here the payload may truly be 64k *after* the
+   * TCPBox (as we have no MTU)!!
    */ 
   struct GNUNET_MessageHeader header;
 
@@ -153,7 +161,8 @@ struct TCPBox
    */ 
   struct GNUNET_ShortHashCode hmac;
 
-  /* followed by as may bytes of payload as indicated in @e header */
+  /* followed by as may bytes of payload as indicated in @e header,
+     excluding the TCPBox itself! */
   
 };
 
@@ -186,7 +195,7 @@ struct TCPRekey
   struct GNUNET_CRYPTO_EcdhePublicKey ephemeral;
   
   /**
-   * Sender's signature of type #GNUNET_SIGNATURE_COMMUNICATOR_TCP_HANDSHAKE
+   * Sender's signature of type #GNUNET_SIGNATURE_COMMUNICATOR_TCP_REKEY
    */
   struct GNUNET_CRYPTO_EddsaSignature sender_sig;
 
@@ -260,11 +269,17 @@ struct Queue
   struct GNUNET_HashCode in_hmac;
 
   /**
-   * Shared secret for HMAC generation on outgoing data.
+   * Shared secret for HMAC generation on outgoing data, ratcheted after
+   * each operation.
    */ 
   struct GNUNET_HashCode out_hmac;
 
   /**
+   * Our ephemeral key. Stored here temporarily during rekeying / key generation.
+   */
+  struct GNUNET_CRYPTO_EcdhePrivateKey ephemeral;
+  
+  /**
    * ID of read task for this connection.
    */
   struct GNUNET_SCHEDULER_Task *read_task;
@@ -297,12 +312,6 @@ struct Queue
   socklen_t address_len;
 
   /**
-   * Message currently scheduled for transmission, non-NULL if and only
-   * if this queue is in the #queue_head DLL.
-   */
-  const struct GNUNET_MessageHeader *msg;
-
-  /**
    * Message queue we are providing for the #ch.
    */
   struct GNUNET_MQ_Handle *mq;
@@ -318,6 +327,50 @@ struct Queue
   unsigned long long bytes_in_queue;
 
   /**
+   * Buffer for reading ciphertext from network into.
+   */
+  char cread_buf[BUF_SIZE];
+
+  /**
+   * buffer for writing ciphertext to network.
+   */
+  char cwrite_buf[BUF_SIZE];
+
+  /**
+   * Plaintext buffer for decrypted plaintext.
+   */
+  char pread_buf[UINT16_MAX + 1 + sizeof (struct TCPBox)];
+
+  /**
+   * Plaintext buffer for messages to be encrypted.
+   */
+  char pwrite_buf[UINT16_MAX + 1 + sizeof (struct TCPBox)];
+  
+  /**
+   * At which offset in the ciphertext read buffer should we
+   * append more ciphertext for transmission next?
+   */
+  size_t cread_off;
+
+  /**
+   * At which offset in the ciphertext write buffer should we
+   * append more ciphertext from reading next?
+   */
+  size_t cwrite_off;
+  
+  /**
+   * At which offset in the plaintext input buffer should we
+   * append more plaintext from decryption next?
+   */
+  size_t pread_off;
+  
+  /**
+   * At which offset in the plaintext output buffer should we
+   * append more plaintext for encryption next?
+   */
+  size_t pwrite_off;
+
+  /**
    * Timeout for this queue.
    */
   struct GNUNET_TIME_Absolute timeout;
@@ -326,7 +379,23 @@ struct Queue
    * Which network type does this queue use?
    */
   enum GNUNET_NetworkType nt;
+
+  /**
+   * Is MQ awaiting a #GNUNET_MQ_impl_send_continue() call?
+   */
+  int mq_awaits_continue;
   
+  /**
+   * Did we enqueue a finish message and are closing down the queue?
+   */
+  int finishing;
+
+  /**
+   * #GNUNET_YES after #inject_key() placed the rekey message into the
+   * plaintext buffer. Once the plaintext buffer is drained, this
+   * means we must switch to the new key material.
+   */
+  int rekey_state;
 };
 
 
@@ -370,6 +439,21 @@ static struct GNUNET_NETWORK_Handle *listen_sock;
  */
 static struct GNUNET_TRANSPORT_AddressIdentifier *ai;
 
+/**
+ * Our public key.
+ */
+static struct GNUNET_PeerIdentity my_identity;
+
+/**
+ * Our private key.
+ */
+static struct GNUNET_CRYPTO_EddsaPrivateKey *my_private_key;
+
+/**
+ * Our configuration.
+ */
+static const struct GNUNET_CONFIGURATION_Handle *cfg;
+
 
 /**
  * We have been notified that our listen socket has something to
@@ -407,7 +491,7 @@ queue_destroy (struct Queue *queue)
                                                        &queue->target,
                                                        queue));
   GNUNET_STATISTICS_set (stats,
-                         "# UNIX queues active",
+                         "# queues active",
                          GNUNET_CONTAINER_multipeermap_size (queue_map),
                          GNUNET_NO);
   if (NULL != queue->read_task)
@@ -435,6 +519,52 @@ queue_destroy (struct Queue *queue)
 
 
 /**
+ * Compute @a mac over @a buf, and ratched the @a hmac_secret.
+ *
+ * @param[in,out] hmac_secret secret for HMAC calculation
+ * @param buf buffer to MAC
+ * @param buf_size number of bytes in @a buf
+ * @param smac[out] where to write the HMAC
+ */
+static void
+hmac (struct GNUNET_HashCode *hmac_secret,
+      const void *buf,
+      size_t buf_size,
+      struct GNUNET_ShortHashCode *smac)
+{
+  struct GNUNET_HashCode mac;
+
+  GNUNET_CRYPTO_hmac_raw (hmac_secret,
+                          sizeof (struct GNUNET_HashCode),
+                          buf,
+                          buf_size,
+                          &mac);
+  /* truncate to `struct GNUNET_ShortHashCode` */
+  memcpy (smac,
+          &mac,
+          sizeof (struct GNUNET_ShortHashCode));
+  /* ratchet hmac key */
+  GNUNET_CRYPTO_hash (hmac_secret,
+                      sizeof (struct GNUNET_HashCode),
+                      hmac_secret);
+}
+
+
+/**
+ * Append a 'finish' message to the outgoing transmission. Once the
+ * finish has been transmitted, destroy the queue.
+ *
+ * @param queue queue to shut down nicely
+ */
+static void
+queue_finish (struct Queue *queue)
+{
+  // FIXME: try to send 'finish' message first!?
+  queue_destroy (queue);
+}
+
+
+/**
  * Queue read task. If we hit the timeout, disconnect it
  *
  * @param cls the `struct Queue *` to disconnect
@@ -444,12 +574,32 @@ queue_read (void *cls)
 {
   struct Queue *queue = cls;
   struct GNUNET_TIME_Relative left;
+  ssize_t rcvd;
 
   queue->read_task = NULL;
-  /* CHECK IF READ-ready, then perform read! */
+  /* FIXME: perform read! */
+  rcvd = GNUNET_NETWORK_socket_recv (queue->sock,
+                                     &queue->cread_buf[queue->cread_off],
+                                     BUF_SIZE - queue->cread_off);
+  if (-1 == rcvd)
+  {
+    // FIXME: error handling...
+  }
+  if (0 != rcvd)
+    /* update queue timeout */
+  queue->cread_off += rcvd;
+  if (queue->pread_off < sizeof (queue->pread_buf))
+  {
+    /* FIXME: decrypt */
   
+    /* FIXME: check plaintext for complete messages, if complete, hand to CORE */
+    /* FIXME: CORE flow control: suspend doing more until CORE has ACKed */
+  }
+  
+  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 (0 != left.rel_value_us) 
   {
     /* not actually our turn yet, but let's at least update
        the monitor, it may think we're about to die ... */
@@ -458,6 +608,7 @@ queue_read (void *cls)
                                        queue->sock,
                                        &queue_read,
                                        queue);
+
     return;
   }
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -465,8 +616,7 @@ queue_read (void *cls)
               queue,
               GNUNET_STRINGS_relative_time_to_string (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
                                                       GNUNET_YES));
-  // FIXME: try to send 'finish' message first!?
-  queue_destroy (queue);
+  queue_finish (queue);
 }
 
 
@@ -506,6 +656,175 @@ tcp_address_to_sockaddr (const char *bindto,
 
 
 /**
+ * Setup @a cipher based on shared secret @a dh and decrypting
+ * peer @a pid.
+ *
+ * @param dh shared secret
+ * @param pid decrypting peer's identity
+ * @param cipher[out] cipher to initialize
+ * @param hmac_key[out] HMAC key to initialize
+ */
+static void
+setup_cipher (const struct GNUNET_HashCode *dh,
+              const struct GNUNET_PeerIdentity *pid,
+              gcry_cipher_hd_t *cipher,
+              struct GNUNET_HashCode *hmac_key)
+{
+  char key[256/8];
+  char ctr[128/8];
+
+  gcry_cipher_open (cipher,
+                    GCRY_CIPHER_AES256 /* low level: go for speed */,
+                    GCRY_CIPHER_MODE_CTR,
+                    0 /* flags */);
+  GNUNET_assert (GNUNET_YES ==
+                 GNUNET_CRYPTO_kdf (key,
+                                    sizeof (key),
+                                    "TCP-key",
+                                    strlen ("TCP-key"),
+                                    dh,
+                                    sizeof (*dh),
+                                    pid,
+                                    sizeof (*pid),
+                                    NULL, 0));
+  gcry_cipher_setkey (*cipher,
+                      key,
+                      sizeof (key));
+  GNUNET_assert (GNUNET_YES ==
+                 GNUNET_CRYPTO_kdf (ctr,
+                                    sizeof (ctr),
+                                    "TCP-ctr",
+                                    strlen ("TCP-ctr"),
+                                    dh,
+                                    sizeof (*dh),
+                                    pid,
+                                    sizeof (*pid),
+                                    NULL, 0));
+  gcry_cipher_setctr (*cipher,
+                      ctr,
+                      sizeof (ctr));
+  GNUNET_assert (GNUNET_YES ==
+                 GNUNET_CRYPTO_kdf (hmac_key,
+                                    sizeof (struct GNUNET_HashCode),
+                                    "TCP-hmac",
+                                    strlen ("TCP-hmac"),
+                                    dh,
+                                    sizeof (*dh),
+                                    pid,
+                                    sizeof (*pid),
+                                    NULL, 0));
+}
+
+
+/**
+ * Setup cipher of @a queue for decryption.
+ *
+ * @param ephemeral ephemeral key we received from the other peer
+ * @param queue[in,out] queue to initialize decryption cipher for
+ */
+static void
+setup_in_cipher (const struct GNUNET_CRYPTO_EcdhePublicKey *ephemeral,
+                 struct Queue *queue)
+{
+  struct GNUNET_HashCode dh;
+  
+  GNUNET_CRYPTO_eddsa_ecdh (my_private_key,
+                            ephemeral,
+                            &dh);
+  setup_cipher (&dh,
+                &my_identity,
+                &queue->in_cipher,
+                &queue->in_hmac);
+}
+                
+
+/**
+ * Setup cipher for outgoing data stream based on target and
+ * our ephemeral private key.
+ *
+ * @param queue queue to setup outgoing (encryption) cipher for
+ */
+static void
+setup_out_cipher (struct Queue *queue)
+{
+  struct GNUNET_HashCode dh;
+  
+  GNUNET_CRYPTO_ecdh_eddsa (&queue->ephemeral,
+                            &queue->target.public_key,
+                            &dh);
+  /* we don't need the private key anymore, drop it! */
+  memset (&queue->ephemeral,
+          0,
+          sizeof (queue->ephemeral));
+  setup_cipher (&dh,
+                &queue->target,
+                &queue->out_cipher,
+                &queue->out_hmac);
+  
+  queue->rekey_time = GNUNET_TIME_relative_to_absolute (REKEY_TIME_INTERVAL);
+  queue->rekey_left_bytes = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
+                                                      REKEY_MAX_BYTES);
+}
+
+
+/**
+ * Inject a `struct TCPRekey` message into the queue's plaintext
+ * buffer.
+ *
+ * @param queue queue to perform rekeying on
+ */ 
+static void
+inject_rekey (struct Queue *queue)
+{
+  struct TCPRekey rekey;
+  struct TcpHandshakeSignature thp;
+  
+  GNUNET_assert (0 == queue->pwrite_off);
+  memset (&rekey,
+          0,
+          sizeof (rekey));
+  GNUNET_assert (GNUNET_OK ==
+                 GNUNET_CRYPTO_ecdhe_key_create2 (&queue->ephemeral));
+  rekey.header.type = ntohs (GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_REKEY);
+  rekey.header.size = ntohs (sizeof (rekey));
+  GNUNET_CRYPTO_ecdhe_key_get_public (&queue->ephemeral,
+                                      &rekey.ephemeral);
+  rekey.monotonic_time = GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get_monotonic (cfg));
+  thp.purpose.purpose = htonl (GNUNET_SIGNATURE_COMMUNICATOR_TCP_REKEY);
+  thp.purpose.size = htonl (sizeof (thp));
+  thp.sender = my_identity;
+  thp.receiver = queue->target;
+  thp.ephemeral = rekey.ephemeral;
+  thp.monotonic_time = rekey.monotonic_time;
+  GNUNET_assert (GNUNET_OK ==
+                 GNUNET_CRYPTO_eddsa_sign (my_private_key,
+                                           &thp.purpose,
+                                           &rekey.sender_sig));
+  hmac (&queue->out_hmac,
+        &rekey,
+        sizeof (rekey),
+        &rekey.hmac);
+  memcpy (queue->pwrite_buf,
+          &rekey,
+          sizeof (rekey));
+  queue->rekey_state = GNUNET_YES;
+}
+
+
+/**
+ * We encrypted the rekey message, now update actually swap the key
+ * material and update the key freshness parameters of @a queue.
+ */ 
+static void
+switch_key (struct Queue *queue)
+{
+  queue->rekey_state = GNUNET_NO; 
+  gcry_cipher_close (queue->out_cipher);
+  setup_out_cipher (queue);
+}
+
+
+/**
  * We have been notified that our socket is ready to write.
  * Then reschedule this function to be called again once more is available.
  *
@@ -515,17 +834,68 @@ static void
 queue_write (void *cls)
 {
   struct Queue *queue = cls;
-  const struct GNUNET_MessageHeader *msg = queue->msg;
-  size_t msg_size = ntohs (msg->size);
+  ssize_t sent;
 
   queue->write_task = NULL;
-  /* FIXME: send 'msg' */
-  /* FIXME: check if we have more messages pending */
-  queue->write_task 
-    = GNUNET_SCHEDULER_add_write_net (GNUNET_TIME_UNIT_FOREVER_REL,
-                                      queue->sock,
-                                      &queue_write,
-                                      queue);
+  sent = GNUNET_NETWORK_socket_send (queue->sock,
+                                     queue->cwrite_buf,
+                                     queue->cwrite_off);
+  if ( (-1 == sent) &&
+       (EAGAIN != errno) &&
+       (EINTR != errno) )
+  {
+    GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING,
+                         "send");
+    queue_destroy (queue);
+    return;                      
+  }
+  if (sent > 0)
+  {
+    size_t usent = (size_t) sent;
+
+    memmove (queue->cwrite_buf,
+             &queue->cwrite_buf[sent],
+             queue->cwrite_off - sent);
+    /* FIXME: update queue timeout */ 
+ }
+  /* can we encrypt more? (always encrypt full messages, needed
+     such that #mq_cancel() can work!) */
+  if (queue->cwrite_off + queue->pwrite_off <= BUF_SIZE)
+  {
+    GNUNET_assert (0 ==
+                   gcry_cipher_encrypt (queue->out_cipher,
+                                        &queue->cwrite_buf[queue->cwrite_off],
+                                        queue->pwrite_off,
+                                        queue->pwrite_buf,
+                                        queue->pwrite_off));
+    if (queue->rekey_left_bytes > queue->pwrite_off)
+      queue->rekey_left_bytes -= queue->pwrite_off;
+    else
+      queue->rekey_left_bytes = 0;
+    queue->cwrite_off += queue->pwrite_off;
+    queue->pwrite_off = 0;
+  }
+  if ( (GNUNET_YES == queue->rekey_state) &&
+       (0 == queue->pwrite_off) )
+    switch_key (queue);
+  if ( (0 == queue->pwrite_off) &&
+       ( (0 == queue->rekey_left_bytes) ||
+         (0 == GNUNET_TIME_absolute_get_remaining (queue->rekey_time).rel_value_us) ) )
+    inject_rekey (queue);
+  if ( (0 == queue->pwrite_off) &&
+       (! queue->finishing) &&
+       (queue->mq_awaits_continue) )
+  {
+    queue->mq_awaits_continue = GNUNET_NO;
+    GNUNET_MQ_impl_send_continue (queue->mq);
+  }
+  /* do we care to write more? */
+  if (0 < queue->cwrite_off)
+    queue->write_task 
+      = GNUNET_SCHEDULER_add_write_net (GNUNET_TIME_UNIT_FOREVER_REL,
+                                        queue->sock,
+                                        &queue_write,
+                                        queue);
 }
 
 
@@ -543,10 +913,25 @@ mq_send (struct GNUNET_MQ_Handle *mq,
          void *impl_state)
 {
   struct Queue *queue = impl_state;
+  uint16_t msize = ntohs (msg->size);
+  struct TCPBox box;
 
   GNUNET_assert (mq == queue->mq);
-  GNUNET_assert (NULL == queue->msg);
-  queue->msg = msg;
+  GNUNET_assert (0 == queue->pread_off);
+  box.header.type = htons (GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_BOX);
+  box.header.size = htons (msize);
+  hmac (&queue->out_hmac,
+        msg,
+        msize,
+        &box.hmac);
+  memcpy (&queue->pread_buf[queue->pread_off],
+          &box,
+          sizeof (box));
+  queue->pread_off += sizeof (box);
+  memcpy (&queue->pread_buf[queue->pread_off],
+          msg,
+          msize);
+  queue->pread_off += msize;
   GNUNET_assert (NULL != queue->sock);
   if (NULL == queue->write_task)
     queue->write_task =
@@ -574,7 +959,7 @@ mq_destroy (struct GNUNET_MQ_Handle *mq,
   if (mq == queue->mq)
   {
     queue->mq = NULL;
-    queue_destroy (queue);
+    queue_finish (queue);
   }
 }
 
@@ -591,14 +976,8 @@ mq_cancel (struct GNUNET_MQ_Handle *mq,
 {
   struct Queue *queue = impl_state;
 
-  GNUNET_assert (NULL != queue->msg);
-  queue->msg = NULL;
-  GNUNET_assert (NULL != queue->write_task);
-  if (1) // FIXME?
-  {
-    GNUNET_SCHEDULER_cancel (queue->write_task);
-    queue->write_task = NULL;
-  }
+  GNUNET_assert (0 != queue->pwrite_off);
+  queue->pwrite_off = 0;
 }
 
 
@@ -618,10 +997,10 @@ mq_error (void *cls,
   struct Queue *queue = cls;
 
   GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-              "TCP MQ error in queue to %s: %d\n",
+              "MQ error in queue to %s: %d\n",
               GNUNET_i2s (&queue->target),
               (int) error);
-  queue_destroy (queue);
+  queue_finish (queue);
 }
 
 
@@ -629,7 +1008,8 @@ mq_error (void *cls,
  * Creates a new outbound queue the transport service will use to send
  * data to another peer.
  *
- * @param peer the target peer
+ * @param sock the queue's socket
+ * @param target the target peer
  * @param cs inbound or outbound queue
  * @param in the address
  * @param in_len number of bytes in @a in
@@ -637,6 +1017,7 @@ mq_error (void *cls,
  */
 static struct Queue *
 setup_queue (struct GNUNET_NETWORK_Handle *sock,
+             const struct GNUNET_PeerIdentity *target,
              enum GNUNET_TRANSPORT_ConnectionStatus cs,
              const struct sockaddr *in,
              socklen_t in_len)
@@ -644,7 +1025,7 @@ setup_queue (struct GNUNET_NETWORK_Handle *sock,
   struct Queue *queue;
 
   queue = GNUNET_new (struct Queue);
-  // queue->target = *target; // FIXME: handle case that we don't know the target yet!
+  queue->target = *target; 
   queue->address = GNUNET_memdup (in,
                                   in_len);
   queue->address_len = in_len;
@@ -730,12 +1111,9 @@ listen_cb (void *cls);
 static void
 listen_cb (void *cls)
 {
-  char buf[65536] GNUNET_ALIGN;
   struct Queue *queue;
   struct sockaddr_storage in;
   socklen_t addrlen;
-  ssize_t ret;
-  uint16_t msize;
   struct GNUNET_NETWORK_Handle *sock;
 
   listen_task = NULL;
@@ -765,6 +1143,9 @@ listen_cb (void *cls)
                          "accept");
     return;
   }
+#if 0
+  // FIXME: setup proto-queue first here, until we have received the starting
+  // messages!
   queue = setup_queue (sock,
                        GNUNET_TRANSPORT_CS_INBOUND,
                        (struct sockaddr *) &in,
@@ -775,6 +1156,7 @@ listen_cb (void *cls)
                 _("Maximum number of TCP connections exceeded, dropping incoming connection\n"));
     return;
   }
+#endif
 }
 
 
@@ -804,6 +1186,10 @@ mq_init (void *cls,
   const char *path;
   struct sockaddr *in;
   socklen_t in_len;
+  struct GNUNET_NETWORK_Handle *sock;
+  struct GNUNET_CRYPTO_EcdhePublicKey epub;
+  struct TcpHandshakeSignature ths;
+  struct TCPConfirmation tc;
 
   if (0 != strncmp (address,
                     COMMUNICATOR_ADDRESS_PREFIX "-",
@@ -815,12 +1201,37 @@ mq_init (void *cls,
   path = &address[strlen (COMMUNICATOR_ADDRESS_PREFIX "-")];
   in = tcp_address_to_sockaddr (path,
                                 &in_len);
-#if FIXME
-  queue = setup_queue (peer,
+  
+  sock = GNUNET_NETWORK_socket_create (in->sa_family,
+                                       SOCK_STREAM,
+                                       IPPROTO_TCP);
+  if (NULL == sock)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                "socket(%d) failed: %s",
+                in->sa_family,
+                STRERROR (errno));
+    GNUNET_free (in);
+    return GNUNET_SYSERR;
+  }
+  if (GNUNET_OK !=
+      GNUNET_NETWORK_socket_connect (sock,
+                                     in,
+                                     in_len))
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                "connect to `%s' failed: %s",
+                address,
+                STRERROR (errno));
+    GNUNET_NETWORK_socket_close (sock);
+    GNUNET_free (in);
+    return GNUNET_SYSERR;
+  }
+  queue = setup_queue (sock,
+                       peer,
                        GNUNET_TRANSPORT_CS_OUTBOUND,
                        in,
                        in_len);
-#endif
   GNUNET_free (in);
   if (NULL == queue)
   {
@@ -828,8 +1239,39 @@ mq_init (void *cls,
                 "Failed to setup queue to %s at `%s'\n",
                 GNUNET_i2s (peer),
                 path);
+    GNUNET_NETWORK_socket_close (sock);
     return GNUNET_NO;
   }
+  GNUNET_assert (GNUNET_OK ==
+                 GNUNET_CRYPTO_ecdhe_key_create2 (&queue->ephemeral)); 
+  GNUNET_CRYPTO_ecdhe_key_get_public (&queue->ephemeral,
+                                      &epub);
+  setup_out_cipher (queue);
+  memcpy (queue->cwrite_buf,
+          &epub,
+          sizeof (epub));
+  queue->cwrite_off = sizeof (epub);
+  /* compute 'tc' and append in encrypted format to cwrite_buf */
+  tc.sender = my_identity;
+  tc.monotonic_time = GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get_monotonic (cfg));
+  ths.purpose.purpose = htonl (GNUNET_SIGNATURE_COMMUNICATOR_TCP_HANDSHAKE);
+  ths.purpose.size = htonl (sizeof (ths));
+  ths.sender = my_identity;
+  ths.receiver = queue->target;
+  ths.ephemeral = epub;
+  ths.monotonic_time = tc.monotonic_time;
+  GNUNET_assert (GNUNET_OK ==
+                 GNUNET_CRYPTO_eddsa_sign (my_private_key,
+                                           &ths.purpose,
+                                           &tc.sender_sig));
+  GNUNET_assert (0 ==
+                 gcry_cipher_encrypt (queue->out_cipher,
+                                      &queue->cwrite_buf[queue->cwrite_off],
+                                      sizeof (tc),
+                                      &tc,
+                                      sizeof (tc)));
+  queue->cwrite_off += sizeof (tc);
+  
   return GNUNET_OK;
 }
 
@@ -895,6 +1337,11 @@ do_shutdown (void *cls)
                                GNUNET_NO);
     stats = NULL;
   }
+  if (NULL != my_private_key)
+  {
+    GNUNET_free (my_private_key);
+    my_private_key = NULL;
+  }
 }
 
 
@@ -903,7 +1350,7 @@ do_shutdown (void *cls)
  * acknowledgement for this communicator (!) via a different return
  * path.
  *
- * Not applicable for UNIX.
+ * Not applicable for TCP.
  *
  * @param cls closure
  * @param sender which peer sent the notification
@@ -927,13 +1374,13 @@ enc_notify_cb (void *cls,
  * @param cls NULL (always)
  * @param args remaining command-line arguments
  * @param cfgfile name of the configuration file used (for saving, can be NULL!)
- * @param cfg configuration
+ * @param c configuration
  */
 static void
 run (void *cls,
      char *const *args,
      const char *cfgfile,
-     const struct GNUNET_CONFIGURATION_Handle *cfg)
+     const struct GNUNET_CONFIGURATION_Handle *c)
 {
   char *bindto;
   struct sockaddr *in;
@@ -941,6 +1388,7 @@ run (void *cls,
   char *my_addr;
   (void) cls;
 
+  cfg = c;
   if (GNUNET_OK !=
       GNUNET_CONFIGURATION_get_value_filename (cfg,
                                                COMMUNICATOR_CONFIG_SECTION,
@@ -1002,6 +1450,17 @@ run (void *cls,
                                     cfg);
   GNUNET_SCHEDULER_add_shutdown (&do_shutdown,
                                  NULL);
+  my_private_key = GNUNET_CRYPTO_eddsa_key_create_from_configuration (cfg);
+  if (NULL == my_private_key)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                _("Transport service is lacking key configuration settings. Exiting.\n"));
+    GNUNET_SCHEDULER_shutdown ();
+    return;
+  }
+  GNUNET_CRYPTO_eddsa_key_get_public (my_private_key,
+                                      &my_identity.public_key);
+
   listen_task = GNUNET_SCHEDULER_add_read_net (GNUNET_TIME_UNIT_FOREVER_REL,
                                                listen_sock,
                                                &listen_cb,