src/transport/gnunet-communicator-tcp.c

   1 /*
   2      This file is part of GNUnet
   3      Copyright (C) 2010-2014, 2018, 2019 GNUnet e.V.
   4
   5      GNUnet is free software: you can redistribute it and/or modify it
   6      under the terms of the GNU Affero General Public License as published
   7      by the Free Software Foundation, either version 3 of the License,
   8      or (at your option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      Affero General Public License for more details.
  14
  15      You should have received a copy of the GNU Affero General Public License
  16      along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17
  18      SPDX-License-Identifier: AGPL3.0-or-later
  19 */
  20
  21 /**
  22  * @file transport/gnunet-communicator-tcp.c
  23  * @brief Transport plugin using TCP.
  24  * @author Christian Grothoff
  25  *
  26  * TODO:
  27  * - NAT service API change to handle address stops!
  28  * - address construction for HELLOs (FIXMEs, easy)
  29  */
  30 #include "platform.h"
  31 #include "gnunet_util_lib.h"
  32 #include "gnunet_protocols.h"
  33 #include "gnunet_signatures.h"
  34 #include "gnunet_constants.h"
  35 #include "gnunet_nt_lib.h"
  36 #include "gnunet_nat_service.h"
  37 #include "gnunet_statistics_service.h"
  38 #include "gnunet_transport_communication_service.h"
  39
  40 /**
  41  * How many messages do we keep at most in the queue to the
  42  * transport service before we start to drop (default,
  43  * can be changed via the configuration file).
  44  * Should be _below_ the level of the communicator API, as
  45  * otherwise we may read messages just to have them dropped
  46  * by the communicator API.
  47  */
  48 #define DEFAULT_MAX_QUEUE_LENGTH 8
  49
  50 /**
  51  * Size of our IO buffers for ciphertext data. Must be at
  52  * least UINT_MAX + sizeof (struct TCPBox).
  53  */
  54 #define BUF_SIZE (2 * 64 * 1024 + sizeof (struct TCPBox))
  55
  56 /**
  57  * How often do we rekey based on time (at least)
  58  */
  59 #define REKEY_TIME_INTERVAL GNUNET_TIME_UNIT_DAYS
  60
  61 /**
  62  * How long do we wait until we must have received the initial KX?
  63  */
  64 #define PROTO_QUEUE_TIMEOUT GNUNET_TIME_UNIT_MINUTES
  65
  66 /**
  67  * How often do we rekey based on number of bytes transmitted?
  68  * (additionally randomized).
  69  */
  70 #define REKEY_MAX_BYTES (1024LLU * 1024 * 1024 * 4LLU)
  71
  72 /**
  73  * Size of the initial key exchange message sent first in both
  74  * directions.
  75  */
  76 #define INITIAL_KX_SIZE (sizeof (struct GNUNET_CRYPTO_EcdhePublicKey)+sizeof (struct TCPConfirmation))
  77
  78
  79 /**
  80  * Address prefix used by the communicator.
  81  */
  82 #define COMMUNICATOR_ADDRESS_PREFIX "tcp"
  83
  84 /**
  85  * Configuration section used by the communicator.
  86  */
  87 #define COMMUNICATOR_CONFIG_SECTION "communicator-tcp"
  88
  89 GNUNET_NETWORK_STRUCT_BEGIN
  90
  91
  92 /**
  93  * Signature we use to verify that the ephemeral key was really chosen by
  94  * the specified sender.
  95  */
  96 struct TcpHandshakeSignature
  97 {
  98   /**
  99    * Purpose must be #GNUNET_SIGNATURE_COMMUNICATOR_TCP_HANDSHAKE
 100    */
 101   struct GNUNET_CRYPTO_EccSignaturePurpose purpose;
 102
 103   /**
 104    * Identity of the inititor of the TCP connection (TCP client).
 105    */
 106   struct GNUNET_PeerIdentity sender;
 107
 108   /**
 109    * Presumed identity of the target of the TCP connection (TCP server)
 110    */
 111   struct GNUNET_PeerIdentity receiver;
 112
 113   /**
 114    * Ephemeral key used by the @e sender.
 115    */
 116   struct GNUNET_CRYPTO_EcdhePublicKey ephemeral;
 117
 118   /**
 119    * Monotonic time of @e sender, to possibly help detect replay attacks
 120    * (if receiver persists times by sender).
 121    */
 122   struct GNUNET_TIME_AbsoluteNBO monotonic_time;
 123 };
 124
 125
 126 /**
 127  * Encrypted continuation of TCP initial handshake.
 128  */
 129 struct TCPConfirmation
 130 {
 131   /**
 132    * Sender's identity
 133    */
 134   struct GNUNET_PeerIdentity sender;
 135
 136   /**
 137    * Sender's signature of type #GNUNET_SIGNATURE_COMMUNICATOR_TCP_HANDSHAKE
 138    */
 139   struct GNUNET_CRYPTO_EddsaSignature sender_sig;
 140
 141   /**
 142    * Monotonic time of @e sender, to possibly help detect replay attacks
 143    * (if receiver persists times by sender).
 144    */
 145   struct GNUNET_TIME_AbsoluteNBO monotonic_time;
 146
 147 };
 148
 149
 150 /**
 151  * TCP message box.  Always sent encrypted!
 152  */
 153 struct TCPBox
 154 {
 155
 156   /**
 157    * Type is #GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_BOX.  Warning: the
 158    * header size EXCLUDES the size of the `struct TCPBox`. We usually
 159    * never do this, but here the payload may truly be 64k *after* the
 160    * TCPBox (as we have no MTU)!!
 161    */
 162   struct GNUNET_MessageHeader header;
 163
 164   /**
 165    * HMAC for the following encrypted message.  Yes, we MUST use
 166    * mac-then-encrypt here, as we want to hide the message sizes on
 167    * the wire (zero plaintext design!).  Using CTR mode padding oracle
 168    * attacks do not apply.  Besides, due to the use of ephemeral keys
 169    * (hopefully with effective replay protection from monotonic time!)
 170    * the attacker is limited in using the oracle.
 171    */
 172   struct GNUNET_ShortHashCode hmac;
 173
 174   /* followed by as may bytes of payload as indicated in @e header,
 175      excluding the TCPBox itself! */
 176
 177 };
 178
 179
 180 /**
 181  * TCP rekey message box.  Always sent encrypted!  Data after
 182  * this message will use the new key.
 183  */
 184 struct TCPRekey
 185 {
 186
 187   /**
 188    * Type is #GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_REKEY.
 189    */
 190   struct GNUNET_MessageHeader header;
 191
 192   /**
 193    * HMAC for the following encrypted message.  Yes, we MUST use
 194    * mac-then-encrypt here, as we want to hide the message sizes on
 195    * the wire (zero plaintext design!).  Using CTR mode padding oracle
 196    * attacks do not apply.  Besides, due to the use of ephemeral keys
 197    * (hopefully with effective replay protection from monotonic time!)
 198    * the attacker is limited in using the oracle.
 199    */
 200   struct GNUNET_ShortHashCode hmac;
 201
 202   /**
 203    * New ephemeral key.
 204    */
 205   struct GNUNET_CRYPTO_EcdhePublicKey ephemeral;
 206
 207   /**
 208    * Sender's signature of type #GNUNET_SIGNATURE_COMMUNICATOR_TCP_REKEY
 209    */
 210   struct GNUNET_CRYPTO_EddsaSignature sender_sig;
 211
 212   /**
 213    * Monotonic time of @e sender, to possibly help detect replay attacks
 214    * (if receiver persists times by sender).
 215    */
 216   struct GNUNET_TIME_AbsoluteNBO monotonic_time;
 217
 218 };
 219
 220
 221 /**
 222  * TCP finish. Sender asks for the connection to be closed.
 223  * Needed/useful in case we drop RST/FIN packets on the GNUnet
 224  * port due to the possibility of malicious RST/FIN injection.
 225  */
 226 struct TCPFinish
 227 {
 228
 229   /**
 230    * Type is #GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_FINISH.
 231    */
 232   struct GNUNET_MessageHeader header;
 233
 234   /**
 235    * HMAC for the following encrypted message.  Yes, we MUST use
 236    * mac-then-encrypt here, as we want to hide the message sizes on
 237    * the wire (zero plaintext design!).  Using CTR mode padding oracle
 238    * attacks do not apply.  Besides, due to the use of ephemeral keys
 239    * (hopefully with effective replay protection from monotonic time!)
 240    * the attacker is limited in using the oracle.
 241    */
 242   struct GNUNET_ShortHashCode hmac;
 243
 244 };
 245
 246
 247 GNUNET_NETWORK_STRUCT_END
 248
 249
 250 /**
 251  * Handle for a queue.
 252  */
 253 struct Queue
 254 {
 255
 256   /**
 257    * To whom are we talking to.
 258    */
 259   struct GNUNET_PeerIdentity target;
 260
 261   /**
 262    * socket that we transmit all data with on this queue
 263    */
 264   struct GNUNET_NETWORK_Handle *sock;
 265
 266   /**
 267    * cipher for decryption of incoming data.
 268    */
 269   gcry_cipher_hd_t in_cipher;
 270
 271   /**
 272    * cipher for encryption of outgoing data.
 273    */
 274   gcry_cipher_hd_t out_cipher;
 275
 276   /**
 277    * Shared secret for HMAC verification on incoming data.
 278    */
 279   struct GNUNET_HashCode in_hmac;
 280
 281   /**
 282    * Shared secret for HMAC generation on outgoing data, ratcheted after
 283    * each operation.
 284    */
 285   struct GNUNET_HashCode out_hmac;
 286
 287   /**
 288    * Our ephemeral key. Stored here temporarily during rekeying / key generation.
 289    */
 290   struct GNUNET_CRYPTO_EcdhePrivateKey ephemeral;
 291
 292   /**
 293    * ID of read task for this connection.
 294    */
 295   struct GNUNET_SCHEDULER_Task *read_task;
 296
 297   /**
 298    * ID of write task for this connection.
 299    */
 300   struct GNUNET_SCHEDULER_Task *write_task;
 301
 302   /**
 303    * Address of the other peer.
 304    */
 305   struct sockaddr *address;
 306
 307   /**
 308    * How many more bytes may we sent with the current @e out_cipher
 309    * before we should rekey?
 310    */
 311   uint64_t rekey_left_bytes;
 312
 313   /**
 314    * Until what time may we sent with the current @e out_cipher
 315    * before we should rekey?
 316    */
 317   struct GNUNET_TIME_Absolute rekey_time;
 318
 319   /**
 320    * Length of the address.
 321    */
 322   socklen_t address_len;
 323
 324   /**
 325    * Message queue we are providing for the #ch.
 326    */
 327   struct GNUNET_MQ_Handle *mq;
 328
 329   /**
 330    * handle for this queue with the #ch.
 331    */
 332   struct GNUNET_TRANSPORT_QueueHandle *qh;
 333
 334   /**
 335    * Number of bytes we currently have in our write queue.
 336    */
 337   unsigned long long bytes_in_queue;
 338
 339   /**
 340    * Buffer for reading ciphertext from network into.
 341    */
 342   char cread_buf[BUF_SIZE];
 343
 344   /**
 345    * buffer for writing ciphertext to network.
 346    */
 347   char cwrite_buf[BUF_SIZE];
 348
 349   /**
 350    * Plaintext buffer for decrypted plaintext.
 351    */
 352   char pread_buf[UINT16_MAX + 1 + sizeof (struct TCPBox)];
 353
 354   /**
 355    * Plaintext buffer for messages to be encrypted.
 356    */
 357   char pwrite_buf[UINT16_MAX + 1 + sizeof (struct TCPBox)];
 358
 359   /**
 360    * At which offset in the ciphertext read buffer should we
 361    * append more ciphertext for transmission next?
 362    */
 363   size_t cread_off;
 364
 365   /**
 366    * At which offset in the ciphertext write buffer should we
 367    * append more ciphertext from reading next?
 368    */
 369   size_t cwrite_off;
 370
 371   /**
 372    * At which offset in the plaintext input buffer should we
 373    * append more plaintext from decryption next?
 374    */
 375   size_t pread_off;
 376
 377   /**
 378    * At which offset in the plaintext output buffer should we
 379    * append more plaintext for encryption next?
 380    */
 381   size_t pwrite_off;
 382
 383   /**
 384    * Timeout for this queue.
 385    */
 386   struct GNUNET_TIME_Absolute timeout;
 387
 388   /**
 389    * How may messages did we pass from this queue to CORE for which we
 390    * have yet to receive an acknoweldgement that CORE is done with
 391    * them? If "large" (or even just non-zero), we should throttle
 392    * reading to provide flow control.  See also #DEFAULT_MAX_QUEUE_LENGTH
 393    * and #max_queue_length.
 394    */
 395   unsigned int backpressure;
 396
 397   /**
 398    * Which network type does this queue use?
 399    */
 400   enum GNUNET_NetworkType nt;
 401
 402   /**
 403    * Is MQ awaiting a #GNUNET_MQ_impl_send_continue() call?
 404    */
 405   int mq_awaits_continue;
 406
 407   /**
 408    * Did we enqueue a finish message and are closing down the queue?
 409    */
 410   int finishing;
 411
 412   /**
 413    * Did we technically destroy this queue, but kept the allocation
 414    * around because of @e backpressure not being zero yet? Used
 415    * simply to delay the final #GNUNET_free() operation until
 416    * #core_read_finished_cb() has been called.
 417    */
 418   int destroyed;
 419
 420   /**
 421    * #GNUNET_YES after #inject_key() placed the rekey message into the
 422    * plaintext buffer. Once the plaintext buffer is drained, this
 423    * means we must switch to the new key material.
 424    */
 425   int rekey_state;
 426
 427   /**
 428    * #GNUNET_YES if we just rekeyed and must thus possibly
 429    * re-decrypt ciphertext.
 430    */
 431   int rekeyed;
 432 };
 433
 434
 435 /**
 436  * Handle for an incoming connection where we do not yet have enough
 437  * information to setup a full queue.
 438  */
 439 struct ProtoQueue
 440 {
 441
 442   /**
 443    * Kept in a DLL.
 444    */
 445   struct ProtoQueue *next;
 446
 447   /**
 448    * Kept in a DLL.
 449    */
 450   struct ProtoQueue *prev;
 451
 452   /**
 453    * socket that we transmit all data with on this queue
 454    */
 455   struct GNUNET_NETWORK_Handle *sock;
 456
 457   /**
 458    * ID of read task for this connection.
 459    */
 460   struct GNUNET_SCHEDULER_Task *read_task;
 461
 462   /**
 463    * Address of the other peer.
 464    */
 465   struct sockaddr *address;
 466
 467   /**
 468    * Length of the address.
 469    */
 470   socklen_t address_len;
 471
 472   /**
 473    * Timeout for this protoqueue.
 474    */
 475   struct GNUNET_TIME_Absolute timeout;
 476
 477   /**
 478    * Buffer for reading all the information we need to upgrade from
 479    * protoqueue to queue.
 480    */
 481   char ibuf[INITIAL_KX_SIZE];
 482
 483   /**
 484    * Current offset for reading into @e ibuf.
 485    */
 486   size_t ibuf_off;
 487 };
 488
 489
 490 /**
 491  * ID of listen task
 492  */
 493 static struct GNUNET_SCHEDULER_Task *listen_task;
 494
 495 /**
 496  * Maximum queue length before we stop reading towards the transport service.
 497  */
 498 static unsigned long long max_queue_length;
 499
 500 /**
 501  * For logging statistics.
 502  */
 503 static struct GNUNET_STATISTICS_Handle *stats;
 504
 505 /**
 506  * Our environment.
 507  */
 508 static struct GNUNET_TRANSPORT_CommunicatorHandle *ch;
 509
 510 /**
 511  * Queues (map from peer identity to `struct Queue`)
 512  */
 513 static struct GNUNET_CONTAINER_MultiPeerMap *queue_map;
 514
 515 /**
 516  * Listen socket.
 517  */
 518 static struct GNUNET_NETWORK_Handle *listen_sock;
 519
 520 /**
 521  * Our public key.
 522  */
 523 static struct GNUNET_PeerIdentity my_identity;
 524
 525 /**
 526  * Our private key.
 527  */
 528 static struct GNUNET_CRYPTO_EddsaPrivateKey *my_private_key;
 529
 530 /**
 531  * Our configuration.
 532  */
 533 static const struct GNUNET_CONFIGURATION_Handle *cfg;
 534
 535 /**
 536  * Network scanner to determine network types.
 537  */
 538 static struct GNUNET_NT_InterfaceScanner *is;
 539
 540 /**
 541  * Connection to NAT service.
 542  */
 543 static struct GNUNET_NAT_Handle *nat;
 544
 545 /**
 546  * Protoqueues DLL head.
 547  */
 548 static struct ProtoQueue *proto_head;
 549
 550 /**
 551  * Protoqueues DLL tail.
 552  */
 553 static struct ProtoQueue *proto_tail;
 554
 555
 556 /**
 557  * We have been notified that our listen socket has something to
 558  * read. Do the read and reschedule this function to be called again
 559  * once more is available.
 560  *
 561  * @param cls NULL
 562  */
 563 static void
 564 listen_cb (void *cls);
 565
 566
 567 /**
 568  * Functions with this signature are called whenever we need
 569  * to close a queue due to a disconnect or failure to
 570  * establish a connection.
 571  *
 572  * @param queue queue to close down
 573  */
 574 static void
 575 queue_destroy (struct Queue *queue)
 576 {
 577   struct GNUNET_MQ_Handle *mq;
 578
 579   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 580               "Disconnecting queue for peer `%s'\n",
 581               GNUNET_i2s (&queue->target));
 582   if (NULL != (mq = queue->mq))
 583   {
 584     queue->mq = NULL;
 585     GNUNET_MQ_destroy (mq);
 586   }
 587   GNUNET_assert (GNUNET_YES ==
 588                  GNUNET_CONTAINER_multipeermap_remove (queue_map,
 589                                                        &queue->target,
 590                                                        queue));
 591   GNUNET_STATISTICS_set (stats,
 592                          "# queues active",
 593                          GNUNET_CONTAINER_multipeermap_size (queue_map),
 594                          GNUNET_NO);
 595   if (NULL != queue->read_task)
 596   {
 597     GNUNET_SCHEDULER_cancel (queue->read_task);
 598     queue->read_task = NULL;
 599   }
 600   if (NULL != queue->write_task)
 601   {
 602     GNUNET_SCHEDULER_cancel (queue->write_task);
 603     queue->write_task = NULL;
 604   }
 605   GNUNET_NETWORK_socket_close (queue->sock);
 606   gcry_cipher_close (queue->in_cipher);
 607   gcry_cipher_close (queue->out_cipher);
 608   GNUNET_free (queue->address);
 609   if (0 != queue->backpressure)
 610     queue->destroyed = GNUNET_YES;
 611   else
 612     GNUNET_free (queue);
 613   if (NULL == listen_task)
 614     listen_task = GNUNET_SCHEDULER_add_read_net (GNUNET_TIME_UNIT_FOREVER_REL,
 615                                                  listen_sock,
 616                                                  &listen_cb,
 617                                                  NULL);
 618 }
 619
 620
 621 /**
 622  * Compute @a mac over @a buf, and ratched the @a hmac_secret.
 623  *
 624  * @param[in,out] hmac_secret secret for HMAC calculation
 625  * @param buf buffer to MAC
 626  * @param buf_size number of bytes in @a buf
 627  * @param smac[out] where to write the HMAC
 628  */
 629 static void
 630 hmac (struct GNUNET_HashCode *hmac_secret,
 631       const void *buf,
 632       size_t buf_size,
 633       struct GNUNET_ShortHashCode *smac)
 634 {
 635   struct GNUNET_HashCode mac;
 636
 637   GNUNET_CRYPTO_hmac_raw (hmac_secret,
 638                           sizeof (struct GNUNET_HashCode),
 639                           buf,
 640                           buf_size,
 641                           &mac);
 642   /* truncate to `struct GNUNET_ShortHashCode` */
 643   memcpy (smac,
 644           &mac,
 645           sizeof (struct GNUNET_ShortHashCode));
 646   /* ratchet hmac key */
 647   GNUNET_CRYPTO_hash (hmac_secret,
 648                       sizeof (struct GNUNET_HashCode),
 649                       hmac_secret);
 650 }
 651
 652
 653 /**
 654  * Append a 'finish' message to the outgoing transmission. Once the
 655  * finish has been transmitted, destroy the queue.
 656  *
 657  * @param queue queue to shut down nicely
 658  */
 659 static void
 660 queue_finish (struct Queue *queue)
 661 {
 662   struct TCPFinish fin;
 663
 664   memset (&fin,
 665           0,
 666           sizeof (fin));
 667   fin.header.size = htons (sizeof (fin));
 668   fin.header.type = htons (GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_FINISH);
 669   hmac (&queue->out_hmac,
 670         &fin,
 671         sizeof (fin),
 672         &fin.hmac);
 673   /* if there is any message left in pwrite_buf, we
 674      overwrite it (possibly dropping the last message
 675      from CORE hard here) */
 676   memcpy (queue->pwrite_buf,
 677           &fin,
 678           sizeof (fin));
 679   queue->pwrite_off = sizeof (fin);
 680   /* This flag will ensure that #queue_write() no longer
 681      notifies CORE about the possibility of sending
 682      more data, and that #queue_write() will call
 683      #queue_destroy() once the @c fin was fully written. */
 684   queue->finishing = GNUNET_YES;
 685 }
 686
 687
 688 /**
 689  * Increment queue timeout due to activity.  We do not immediately
 690  * notify the monitor here as that might generate excessive
 691  * signalling.
 692  *
 693  * @param queue queue for which the timeout should be rescheduled
 694  */
 695 static void
 696 reschedule_queue_timeout (struct Queue *queue)
 697 {
 698   queue->timeout
 699     = GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
 700 }
 701
 702
 703 /**
 704  * Queue read task. If we hit the timeout, disconnect it
 705  *
 706  * @param cls the `struct Queue *` to disconnect
 707  */
 708 static void
 709 queue_read (void *cls);
 710
 711
 712 /**
 713  * Core tells us it is done processing a message that transport
 714  * received on a queue with status @a success.
 715  *
 716  * @param cls a `struct Queue *` where the message originally came from
 717  * @param success #GNUNET_OK on success
 718  */
 719 static void
 720 core_read_finished_cb (void *cls,
 721                        int success)
 722 {
 723   struct Queue *queue = cls;
 724
 725   if (GNUNET_OK != success)
 726     GNUNET_STATISTICS_update (stats,
 727                               "# messages lost in communicator API towards CORE",
 728                               1,
 729                               GNUNET_NO);
 730   queue->backpressure--;
 731   /* handle deferred queue destruction */
 732   if ( (queue->destroyed) &&
 733        (0 == queue->backpressure) )
 734   {
 735     GNUNET_free (queue);
 736     return;
 737   }
 738   reschedule_queue_timeout (queue);
 739   /* possibly unchoke reading, now that CORE made progress */
 740   if (NULL == queue->read_task)
 741     queue->read_task
 742       = GNUNET_SCHEDULER_add_read_net (GNUNET_TIME_absolute_get_remaining (queue->timeout),
 743                                        queue->sock,
 744                                        &queue_read,
 745                                        queue);
 746 }
 747
 748
 749 /**
 750  * We received @a plaintext_len bytes of @a plaintext on @a queue.
 751  * Pass it on to CORE.  If transmission is actually happening,
 752  * increase backpressure counter.
 753  *
 754  * @param queue the queue that received the plaintext
 755  * @param plaintext the plaintext that was received
 756  * @param plaintext_len number of bytes of plaintext received
 757  */
 758 static void
 759 pass_plaintext_to_core (struct Queue *queue,
 760                         const void *plaintext,
 761                         size_t plaintext_len)
 762 {
 763   const struct GNUNET_MessageHeader *hdr = plaintext;
 764   int ret;
 765
 766   if (ntohs (hdr->size) != plaintext_len)
 767   {
 768     /* NOTE: If we ever allow multiple CORE messages in one
 769        BOX, this will have to change! */
 770     GNUNET_break (0);
 771     return;
 772   }
 773   ret = GNUNET_TRANSPORT_communicator_receive (ch,
 774                                                &queue->target,
 775                                                hdr,
 776                                                &core_read_finished_cb,
 777                                                queue);
 778   if (GNUNET_OK == ret)
 779     queue->backpressure++;
 780   GNUNET_break (GNUNET_NO != ret); /* backpressure not working!? */
 781   if (GNUNET_SYSERR == ret)
 782     GNUNET_STATISTICS_update (stats,
 783                               "# bytes lost due to CORE not running",
 784                               plaintext_len,
 785                               GNUNET_NO);
 786 }
 787
 788
 789 /**
 790  * Setup @a cipher based on shared secret @a dh and decrypting
 791  * peer @a pid.
 792  *
 793  * @param dh shared secret
 794  * @param pid decrypting peer's identity
 795  * @param cipher[out] cipher to initialize
 796  * @param hmac_key[out] HMAC key to initialize
 797  */
 798 static void
 799 setup_cipher (const struct GNUNET_HashCode *dh,
 800               const struct GNUNET_PeerIdentity *pid,
 801               gcry_cipher_hd_t *cipher,
 802               struct GNUNET_HashCode *hmac_key)
 803 {
 804   char key[256/8];
 805   char ctr[128/8];
 806
 807   gcry_cipher_open (cipher,
 808                     GCRY_CIPHER_AES256 /* low level: go for speed */,
 809                     GCRY_CIPHER_MODE_CTR,
 810                     0 /* flags */);
 811   GNUNET_assert (GNUNET_YES ==
 812                  GNUNET_CRYPTO_kdf (key,
 813                                     sizeof (key),
 814                                     "TCP-key",
 815                                     strlen ("TCP-key"),
 816                                     dh,
 817                                     sizeof (*dh),
 818                                     pid,
 819                                     sizeof (*pid),
 820                                     NULL, 0));
 821   gcry_cipher_setkey (*cipher,
 822                       key,
 823                       sizeof (key));
 824   GNUNET_assert (GNUNET_YES ==
 825                  GNUNET_CRYPTO_kdf (ctr,
 826                                     sizeof (ctr),
 827                                     "TCP-ctr",
 828                                     strlen ("TCP-ctr"),
 829                                     dh,
 830                                     sizeof (*dh),
 831                                     pid,
 832                                     sizeof (*pid),
 833                                     NULL, 0));
 834   gcry_cipher_setctr (*cipher,
 835                       ctr,
 836                       sizeof (ctr));
 837   GNUNET_assert (GNUNET_YES ==
 838                  GNUNET_CRYPTO_kdf (hmac_key,
 839                                     sizeof (struct GNUNET_HashCode),
 840                                     "TCP-hmac",
 841                                     strlen ("TCP-hmac"),
 842                                     dh,
 843                                     sizeof (*dh),
 844                                     pid,
 845                                     sizeof (*pid),
 846                                     NULL, 0));
 847 }
 848
 849
 850 /**
 851  * Setup cipher of @a queue for decryption.
 852  *
 853  * @param ephemeral ephemeral key we received from the other peer
 854  * @param queue[in,out] queue to initialize decryption cipher for
 855  */
 856 static void
 857 setup_in_cipher (const struct GNUNET_CRYPTO_EcdhePublicKey *ephemeral,
 858                  struct Queue *queue)
 859 {
 860   struct GNUNET_HashCode dh;
 861
 862   GNUNET_CRYPTO_eddsa_ecdh (my_private_key,
 863                             ephemeral,
 864                             &dh);
 865   setup_cipher (&dh,
 866                 &my_identity,
 867                 &queue->in_cipher,
 868                 &queue->in_hmac);
 869 }
 870
 871
 872 /**
 873  * Handle @a rekey message on @a queue. The message was already
 874  * HMAC'ed, but we should additionally still check the signature.
 875  * Then we need to stop the old cipher and start afresh.
 876  *
 877  * @param queue the queue @a rekey was received on
 878  * @param rekey the rekey message
 879  */
 880 static void
 881 do_rekey (struct Queue *queue,
 882           const struct TCPRekey *rekey)
 883 {
 884   struct TcpHandshakeSignature thp;
 885
 886   thp.purpose.purpose = htonl (GNUNET_SIGNATURE_COMMUNICATOR_TCP_REKEY);
 887   thp.purpose.size = htonl (sizeof (thp));
 888   thp.sender = queue->target;
 889   thp.receiver = my_identity;
 890   thp.ephemeral = rekey->ephemeral;
 891   thp.monotonic_time = rekey->monotonic_time;
 892   if (GNUNET_OK !=
 893       GNUNET_CRYPTO_eddsa_verify (GNUNET_SIGNATURE_COMMUNICATOR_TCP_REKEY,
 894                                   &thp.purpose,
 895                                   &rekey->sender_sig,
 896                                   &queue->target.public_key))
 897   {
 898     GNUNET_break (0);
 899     queue_finish (queue);
 900     return;
 901   }
 902   gcry_cipher_close (queue->in_cipher);
 903   queue->rekeyed = GNUNET_YES;
 904   setup_in_cipher (&rekey->ephemeral,
 905                    queue);
 906 }
 907
 908
 909 /**
 910  * Test if we have received a full message in plaintext.
 911  * If so, handle it.
 912  *
 913  * @param queue queue to process inbound plaintext for
 914  * @return number of bytes of plaintext handled, 0 for none
 915  */
 916 static size_t
 917 try_handle_plaintext (struct Queue *queue)
 918 {
 919   const struct GNUNET_MessageHeader *hdr
 920     = (const struct GNUNET_MessageHeader *) queue->pread_buf;
 921   const struct TCPBox *box
 922     = (const struct TCPBox *) queue->pread_buf;
 923   const struct TCPRekey *rekey
 924     = (const struct TCPRekey *) queue->pread_buf;
 925   const struct TCPFinish *fin
 926     = (const struct TCPFinish *) queue->pread_buf;
 927   struct TCPRekey rekeyz;
 928   struct TCPFinish finz;
 929   struct GNUNET_ShortHashCode tmac;
 930   uint16_t type;
 931   size_t size = 0; /* make compiler happy */
 932
 933   if (sizeof (*hdr) > queue->pread_off)
 934     return 0; /* not even a header */
 935   type = ntohs (hdr->type);
 936   switch (type)
 937   {
 938   case GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_BOX:
 939     /* Special case: header size excludes box itself! */
 940     if (ntohs (hdr->size) + sizeof (struct TCPBox) > queue->pread_off)
 941       return 0;
 942     hmac (&queue->in_hmac,
 943           &box[1],
 944           ntohs (hdr->size),
 945           &tmac);
 946     if (0 != memcmp (&tmac,
 947                      &box->hmac,
 948                      sizeof (tmac)))
 949     {
 950       GNUNET_break_op (0);
 951       queue_finish (queue);
 952       return 0;
 953     }
 954     pass_plaintext_to_core (queue,
 955                             (const void *) &box[1],
 956                             ntohs (hdr->size));
 957     size = ntohs (hdr->size) + sizeof (*box);
 958     break;
 959   case GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_REKEY:
 960     if (sizeof (*rekey) > queue->pread_off)
 961       return 0;
 962     if (ntohs (hdr->size) != sizeof (*rekey))
 963     {
 964       GNUNET_break_op (0);
 965       queue_finish (queue);
 966       return 0;
 967     }
 968     rekeyz = *rekey;
 969     memset (&rekeyz.hmac,
 970             0,
 971             sizeof (rekeyz.hmac));
 972     hmac (&queue->in_hmac,
 973           &rekeyz,
 974           sizeof (rekeyz),
 975           &tmac);
 976     if (0 != memcmp (&tmac,
 977                      &box->hmac,
 978                      sizeof (tmac)))
 979     {
 980       GNUNET_break_op (0);
 981       queue_finish (queue);
 982       return 0;
 983     }
 984     do_rekey (queue,
 985               rekey);
 986     size = ntohs (hdr->size);
 987     break;
 988   case GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_FINISH:
 989     if (sizeof (*fin) > queue->pread_off)
 990       return 0;
 991     if (ntohs (hdr->size) != sizeof (*fin))
 992     {
 993       GNUNET_break_op (0);
 994       queue_finish (queue);
 995       return 0;
 996     }
 997     finz = *fin;
 998     memset (&finz.hmac,
 999             0,
1000             sizeof (finz.hmac));
1001     hmac (&queue->in_hmac,
1002           &rekeyz,
1003           sizeof (rekeyz),
1004           &tmac);
1005     if (0 != memcmp (&tmac,
1006                      &fin->hmac,
1007                      sizeof (tmac)))
1008     {
1009       GNUNET_break_op (0);
1010       queue_finish (queue);
1011       return 0;
1012     }
1013     /* handle FINISH by destroying queue */
1014     queue_destroy (queue);
1015     break;
1016   default:
1017     GNUNET_break_op (0);
1018     queue_finish (queue);
1019     return 0;
1020   }
1021   GNUNET_assert (0 != size);
1022   return size;
1023 }
1024
1025
1026 /**
1027  * Queue read task. If we hit the timeout, disconnect it
1028  *
1029  * @param cls the `struct Queue *` to disconnect
1030  */
1031 static void
1032 queue_read (void *cls)
1033 {
1034   struct Queue *queue = cls;
1035   struct GNUNET_TIME_Relative left;
1036   ssize_t rcvd;
1037
1038   queue->read_task = NULL;
1039   rcvd = GNUNET_NETWORK_socket_recv (queue->sock,
1040                                      &queue->cread_buf[queue->cread_off],
1041                                      BUF_SIZE - queue->cread_off);
1042   if (-1 == rcvd)
1043   {
1044     if ( (EAGAIN != errno) &&
1045          (EINTR != errno) )
1046     {
1047       GNUNET_log_strerror (GNUNET_ERROR_TYPE_DEBUG,
1048                            "recv");
1049       queue_finish (queue);
1050       return;
1051     }
1052     /* try again */
1053     queue->read_task
1054       = GNUNET_SCHEDULER_add_read_net (left,
1055                                        queue->sock,
1056                                        &queue_read,
1057                                        queue);
1058     return;
1059   }
1060   if (0 != rcvd)
1061     reschedule_queue_timeout (queue);
1062   queue->cread_off += rcvd;
1063   while ( (queue->pread_off < sizeof (queue->pread_buf)) &&
1064           (queue->cread_off > 0) )
1065   {
1066     size_t max = GNUNET_MIN (sizeof (queue->pread_buf) - queue->pread_off,
1067                              queue->cread_off);
1068     size_t done;
1069     size_t total;
1070
1071     GNUNET_assert (0 ==
1072                    gcry_cipher_decrypt (queue->in_cipher,
1073                                         &queue->pread_buf[queue->pread_off],
1074                                         max,
1075                                         queue->cread_buf,
1076                                         max));
1077     queue->pread_off += max;
1078     total = 0;
1079     while ( (GNUNET_NO == queue->rekeyed) &&
1080             (0 != (done = try_handle_plaintext (queue))) )
1081     {
1082       /* 'done' bytes of plaintext were used, shift buffer */
1083       GNUNET_assert (done <= queue->pread_off);
1084       /* NOTE: this memmove() could possibly sometimes be
1085          avoided if we pass 'total' into try_handle_plaintext()
1086          and use it at an offset into the buffer there! */
1087       memmove (queue->pread_buf,
1088                &queue->pread_buf[done],
1089                queue->pread_off - done);
1090       queue->pread_off -= done;
1091       total += done;
1092     }
1093     /* when we encounter a rekey message, the decryption above uses the
1094        wrong key for everything after the rekey; in that case, we have
1095        to re-do the decryption at 'total' instead of at 'max'. If there
1096        is no rekey and the last message is incomplete (max > total),
1097        it is safe to keep the decryption so we shift by 'max' */
1098     if (GNUNET_YES == queue->rekeyed)
1099     {
1100       max = total;
1101       queue->rekeyed = GNUNET_NO;
1102     }
1103     memmove (queue->cread_buf,
1104              &queue->cread_buf[max],
1105              queue->cread_off - max);
1106     queue->cread_off -= max;
1107   }
1108
1109   if (BUF_SIZE == queue->cread_off)
1110     return; /* buffer full, suspend reading */
1111   left = GNUNET_TIME_absolute_get_remaining (queue->timeout);
1112   if (0 != left.rel_value_us)
1113   {
1114     if (max_queue_length < queue->backpressure)
1115     {
1116       /* continue reading */
1117       queue->read_task
1118         = GNUNET_SCHEDULER_add_read_net (left,
1119                                          queue->sock,
1120                                          &queue_read,
1121                                          queue);
1122     }
1123     return;
1124   }
1125   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1126               "Queue %p was idle for %s, disconnecting\n",
1127               queue,
1128               GNUNET_STRINGS_relative_time_to_string (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
1129                                                       GNUNET_YES));
1130   queue_finish (queue);
1131 }
1132
1133
1134 /**
1135  * Convert TCP bind specification to a `struct sockaddr *`
1136  *
1137  * @param bindto bind specification to convert
1138  * @param[out] sock_len set to the length of the address
1139  * @return converted bindto specification
1140  */
1141 static struct sockaddr *
1142 tcp_address_to_sockaddr (const char *bindto,
1143                          socklen_t *sock_len)
1144 {
1145   struct sockaddr *in;
1146   unsigned int port;
1147   char dummy[2];
1148   char *colon;
1149   char *cp;
1150
1151   if (1 == SSCANF (bindto,
1152                    "%u%1s",
1153                    &port,
1154                    dummy))
1155   {
1156     /* interpreting value as just a PORT number */
1157     if (port > UINT16_MAX)
1158     {
1159       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1160                   "BINDTO specification `%s' invalid: value too large for port\n",
1161                   bindto);
1162       return NULL;
1163     }
1164     if (GNUNET_YES ==
1165         GNUNET_CONFIGURATION_get_value_yesno (cfg,
1166                                               COMMUNICATOR_CONFIG_SECTION,
1167                                               "DISABLE_V6"))
1168     {
1169       struct sockaddr_in *i4;
1170
1171       i4 = GNUNET_malloc (sizeof (struct sockaddr_in));
1172       i4->sin_family = AF_INET;
1173       i4->sin_port = htons ((uint16_t) port);
1174       *sock_len = sizeof (struct sockaddr_in);
1175       in = (struct sockaddr *) i4;
1176     }
1177     else
1178     {
1179       struct sockaddr_in6 *i6;
1180
1181       i6 = GNUNET_malloc (sizeof (struct sockaddr_in6));
1182       i6->sin6_family = AF_INET6;
1183       i6->sin6_port = htons ((uint16_t) port);
1184       *sock_len = sizeof (struct sockaddr_in6);
1185       in = (struct sockaddr *) i6;
1186     }
1187     return in;
1188   }
1189   cp = GNUNET_strdup (bindto);
1190   colon = strrchr (cp, ':');
1191   if (NULL != colon)
1192   {
1193     /* interpet value after colon as port */
1194     *colon = '\0';
1195     colon++;
1196     if (1 == SSCANF (colon,
1197                      "%u%1s",
1198                      &port,
1199                      dummy))
1200     {
1201       /* interpreting value as just a PORT number */
1202       if (port > UINT16_MAX)
1203       {
1204         GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1205                     "BINDTO specification `%s' invalid: value too large for port\n",
1206                     bindto);
1207         GNUNET_free (cp);
1208         return NULL;
1209       }
1210     }
1211     else
1212     {
1213       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1214                   "BINDTO specification `%s' invalid: last ':' not followed by number\n",
1215                   bindto);
1216       GNUNET_free (cp);
1217       return NULL;
1218     }
1219   }
1220   else
1221   {
1222     /* interpret missing port as 0, aka pick any free one */
1223     port = 0;
1224   }
1225   {
1226     /* try IPv4 */
1227     struct sockaddr_in v4;
1228
1229     if (1 == inet_pton (AF_INET,
1230                         cp,
1231                         &v4))
1232     {
1233       v4.sin_port = htons ((uint16_t) port);
1234       in = GNUNET_memdup (&v4,
1235                           sizeof (v4));
1236       *sock_len = sizeof (v4);
1237       GNUNET_free (cp);
1238       return in;
1239     }
1240   }
1241   {
1242     /* try IPv6 */
1243     struct sockaddr_in6 v6;
1244
1245     if (1 == inet_pton (AF_INET6,
1246                         cp,
1247                         &v6))
1248     {
1249       v6.sin6_port = htons ((uint16_t) port);
1250       in = GNUNET_memdup (&v6,
1251                           sizeof (v6));
1252       *sock_len = sizeof (v6);
1253       GNUNET_free (cp);
1254       return in;
1255     }
1256   }
1257   /* FIXME (feature!): maybe also try getnameinfo()? */
1258   GNUNET_free (cp);
1259   return NULL;
1260 }
1261
1262
1263 /**
1264  * Setup cipher for outgoing data stream based on target and
1265  * our ephemeral private key.
1266  *
1267  * @param queue queue to setup outgoing (encryption) cipher for
1268  */
1269 static void
1270 setup_out_cipher (struct Queue *queue)
1271 {
1272   struct GNUNET_HashCode dh;
1273
1274   GNUNET_CRYPTO_ecdh_eddsa (&queue->ephemeral,
1275                             &queue->target.public_key,
1276                             &dh);
1277   /* we don't need the private key anymore, drop it! */
1278   memset (&queue->ephemeral,
1279           0,
1280           sizeof (queue->ephemeral));
1281   setup_cipher (&dh,
1282                 &queue->target,
1283                 &queue->out_cipher,
1284                 &queue->out_hmac);
1285
1286   queue->rekey_time = GNUNET_TIME_relative_to_absolute (REKEY_TIME_INTERVAL);
1287   queue->rekey_left_bytes = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
1288                                                       REKEY_MAX_BYTES);
1289 }
1290
1291
1292 /**
1293  * Inject a `struct TCPRekey` message into the queue's plaintext
1294  * buffer.
1295  *
1296  * @param queue queue to perform rekeying on
1297  */
1298 static void
1299 inject_rekey (struct Queue *queue)
1300 {
1301   struct TCPRekey rekey;
1302   struct TcpHandshakeSignature thp;
1303
1304   GNUNET_assert (0 == queue->pwrite_off);
1305   memset (&rekey,
1306           0,
1307           sizeof (rekey));
1308   GNUNET_assert (GNUNET_OK ==
1309                  GNUNET_CRYPTO_ecdhe_key_create2 (&queue->ephemeral));
1310   rekey.header.type = ntohs (GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_REKEY);
1311   rekey.header.size = ntohs (sizeof (rekey));
1312   GNUNET_CRYPTO_ecdhe_key_get_public (&queue->ephemeral,
1313                                       &rekey.ephemeral);
1314   rekey.monotonic_time = GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get_monotonic (cfg));
1315   thp.purpose.purpose = htonl (GNUNET_SIGNATURE_COMMUNICATOR_TCP_REKEY);
1316   thp.purpose.size = htonl (sizeof (thp));
1317   thp.sender = my_identity;
1318   thp.receiver = queue->target;
1319   thp.ephemeral = rekey.ephemeral;
1320   thp.monotonic_time = rekey.monotonic_time;
1321   GNUNET_assert (GNUNET_OK ==
1322                  GNUNET_CRYPTO_eddsa_sign (my_private_key,
1323                                            &thp.purpose,
1324                                            &rekey.sender_sig));
1325   hmac (&queue->out_hmac,
1326         &rekey,
1327         sizeof (rekey),
1328         &rekey.hmac);
1329   memcpy (queue->pwrite_buf,
1330           &rekey,
1331           sizeof (rekey));
1332   queue->rekey_state = GNUNET_YES;
1333 }
1334
1335
1336 /**
1337  * We encrypted the rekey message, now update actually swap the key
1338  * material and update the key freshness parameters of @a queue.
1339  */
1340 static void
1341 switch_key (struct Queue *queue)
1342 {
1343   queue->rekey_state = GNUNET_NO;
1344   gcry_cipher_close (queue->out_cipher);
1345   setup_out_cipher (queue);
1346 }
1347
1348
1349 /**
1350  * We have been notified that our socket is ready to write.
1351  * Then reschedule this function to be called again once more is available.
1352  *
1353  * @param cls a `struct Queue`
1354  */
1355 static void
1356 queue_write (void *cls)
1357 {
1358   struct Queue *queue = cls;
1359   ssize_t sent;
1360
1361   queue->write_task = NULL;
1362   sent = GNUNET_NETWORK_socket_send (queue->sock,
1363                                      queue->cwrite_buf,
1364                                      queue->cwrite_off);
1365   if ( (-1 == sent) &&
1366        (EAGAIN != errno) &&
1367        (EINTR != errno) )
1368   {
1369     GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING,
1370                          "send");
1371     queue_destroy (queue);
1372     return;
1373   }
1374   if (sent > 0)
1375   {
1376     size_t usent = (size_t) sent;
1377
1378     memmove (queue->cwrite_buf,
1379              &queue->cwrite_buf[usent],
1380              queue->cwrite_off - usent);
1381     reschedule_queue_timeout (queue);
1382  }
1383   /* can we encrypt more? (always encrypt full messages, needed
1384      such that #mq_cancel() can work!) */
1385   if (queue->cwrite_off + queue->pwrite_off <= BUF_SIZE)
1386   {
1387     GNUNET_assert (0 ==
1388                    gcry_cipher_encrypt (queue->out_cipher,
1389                                         &queue->cwrite_buf[queue->cwrite_off],
1390                                         queue->pwrite_off,
1391                                         queue->pwrite_buf,
1392                                         queue->pwrite_off));
1393     if (queue->rekey_left_bytes > queue->pwrite_off)
1394       queue->rekey_left_bytes -= queue->pwrite_off;
1395     else
1396       queue->rekey_left_bytes = 0;
1397     queue->cwrite_off += queue->pwrite_off;
1398     queue->pwrite_off = 0;
1399   }
1400   if ( (GNUNET_YES == queue->rekey_state) &&
1401        (0 == queue->pwrite_off) )
1402     switch_key (queue);
1403   if ( (0 == queue->pwrite_off) &&
1404        ( (0 == queue->rekey_left_bytes) ||
1405          (0 == GNUNET_TIME_absolute_get_remaining (queue->rekey_time).rel_value_us) ) )
1406     inject_rekey (queue);
1407   if ( (0 == queue->pwrite_off) &&
1408        (! queue->finishing) &&
1409        (queue->mq_awaits_continue) )
1410   {
1411     queue->mq_awaits_continue = GNUNET_NO;
1412     GNUNET_MQ_impl_send_continue (queue->mq);
1413   }
1414   /* did we just finish writing 'finish'? */
1415   if ( (0 == queue->cwrite_off) &&
1416        (GNUNET_YES == queue->finishing) )
1417   {
1418     queue_destroy (queue);
1419     return;
1420   }
1421   /* do we care to write more? */
1422   if (0 < queue->cwrite_off)
1423     queue->write_task
1424       = GNUNET_SCHEDULER_add_write_net (GNUNET_TIME_UNIT_FOREVER_REL,
1425                                         queue->sock,
1426                                         &queue_write,
1427                                         queue);
1428 }
1429
1430
1431 /**
1432  * Signature of functions implementing the sending functionality of a
1433  * message queue.
1434  *
1435  * @param mq the message queue
1436  * @param msg the message to send
1437  * @param impl_state our `struct Queue`
1438  */
1439 static void
1440 mq_send (struct GNUNET_MQ_Handle *mq,
1441          const struct GNUNET_MessageHeader *msg,
1442          void *impl_state)
1443 {
1444   struct Queue *queue = impl_state;
1445   uint16_t msize = ntohs (msg->size);
1446   struct TCPBox box;
1447
1448   GNUNET_assert (mq == queue->mq);
1449   if (GNUNET_YES == queue->finishing)
1450     return; /* this queue is dying, drop msg */
1451   GNUNET_assert (0 == queue->pread_off);
1452   box.header.type = htons (GNUNET_MESSAGE_TYPE_COMMUNICATOR_TCP_BOX);
1453   box.header.size = htons (msize);
1454   hmac (&queue->out_hmac,
1455         msg,
1456         msize,
1457         &box.hmac);
1458   memcpy (&queue->pread_buf[queue->pread_off],
1459           &box,
1460           sizeof (box));
1461   queue->pread_off += sizeof (box);
1462   memcpy (&queue->pread_buf[queue->pread_off],
1463           msg,
1464           msize);
1465   queue->pread_off += msize;
1466   GNUNET_assert (NULL != queue->sock);
1467   if (NULL == queue->write_task)
1468     queue->write_task =
1469       GNUNET_SCHEDULER_add_write_net (GNUNET_TIME_UNIT_FOREVER_REL,
1470                                       queue->sock,
1471                                       &queue_write,
1472                                       queue);
1473 }
1474
1475
1476 /**
1477  * Signature of functions implementing the destruction of a message
1478  * queue.  Implementations must not free @a mq, but should take care
1479  * of @a impl_state.
1480  *
1481  * @param mq the message queue to destroy
1482  * @param impl_state our `struct Queue`
1483  */
1484 static void
1485 mq_destroy (struct GNUNET_MQ_Handle *mq,
1486             void *impl_state)
1487 {
1488   struct Queue *queue = impl_state;
1489
1490   if (mq == queue->mq)
1491   {
1492     queue->mq = NULL;
1493     queue_finish (queue);
1494   }
1495 }
1496
1497
1498 /**
1499  * Implementation function that cancels the currently sent message.
1500  *
1501  * @param mq message queue
1502  * @param impl_state our `struct Queue`
1503  */
1504 static void
1505 mq_cancel (struct GNUNET_MQ_Handle *mq,
1506            void *impl_state)
1507 {
1508   struct Queue *queue = impl_state;
1509
1510   GNUNET_assert (0 != queue->pwrite_off);
1511   queue->pwrite_off = 0;
1512 }
1513
1514
1515 /**
1516  * Generic error handler, called with the appropriate
1517  * error code and the same closure specified at the creation of
1518  * the message queue.
1519  * Not every message queue implementation supports an error handler.
1520  *
1521  * @param cls our `struct Queue`
1522  * @param error error code
1523  */
1524 static void
1525 mq_error (void *cls,
1526           enum GNUNET_MQ_Error error)
1527 {
1528   struct Queue *queue = cls;
1529
1530   GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1531               "MQ error in queue to %s: %d\n",
1532               GNUNET_i2s (&queue->target),
1533               (int) error);
1534   queue_finish (queue);
1535 }
1536
1537
1538 /**
1539  * Add the given @a queue to our internal data structure.  Setup the
1540  * MQ processing and inform transport that the queue is ready.  Must
1541  * be called after the KX for outgoing messages has been bootstrapped.
1542  *
1543  * @param queue queue to boot
1544  */
1545 static void
1546 boot_queue (struct Queue *queue,
1547             enum GNUNET_TRANSPORT_ConnectionStatus cs)
1548 {
1549   queue->nt = GNUNET_NT_scanner_get_type (is,
1550                                           queue->address,
1551                                           queue->address_len);
1552   (void) GNUNET_CONTAINER_multipeermap_put (queue_map,
1553                                             &queue->target,
1554                                             queue,
1555                                             GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1556   GNUNET_STATISTICS_set (stats,
1557                          "# queues active",
1558                          GNUNET_CONTAINER_multipeermap_size (queue_map),
1559                          GNUNET_NO);
1560   queue->timeout
1561     = GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
1562   queue->mq
1563     = GNUNET_MQ_queue_for_callbacks (&mq_send,
1564                                      &mq_destroy,
1565                                      &mq_cancel,
1566                                      queue,
1567                                      NULL,
1568                                      &mq_error,
1569                                      queue);
1570   {
1571     char *foreign_addr;
1572
1573     switch (queue->address->sa_family)
1574     {
1575     case AF_INET:
1576       GNUNET_asprintf (&foreign_addr,
1577                        "%s-%s:%d",
1578                        COMMUNICATOR_ADDRESS_PREFIX,
1579                        "inet-ntop-FIXME",
1580                        4242);
1581       break;
1582     case AF_INET6:
1583       GNUNET_asprintf (&foreign_addr,
1584                        "%s-%s:%d",
1585                        COMMUNICATOR_ADDRESS_PREFIX,
1586                        "inet-ntop-FIXME",
1587                        4242);
1588       break;
1589     default:
1590       GNUNET_assert (0);
1591     }
1592     queue->qh
1593       = GNUNET_TRANSPORT_communicator_mq_add (ch,
1594                                               &queue->target,
1595                                               foreign_addr,
1596                                               0 /* no MTU */,
1597                                               queue->nt,
1598                                               cs,
1599                                               queue->mq);
1600     GNUNET_free (foreign_addr);
1601   }
1602 }
1603
1604
1605 /**
1606  * Generate and transmit our ephemeral key and the signature for
1607  * the initial KX with the other peer.  Must be called first, before
1608  * any other bytes are ever written to the output buffer.  Note that
1609  * our cipher must already be initialized when calling this function.
1610  * Helper function for #start_initial_kx_out().
1611  *
1612  * @param queue queue to do KX for
1613  * @param epub our public key for the KX
1614  */
1615 static void
1616 transmit_kx (struct Queue *queue,
1617              const struct GNUNET_CRYPTO_EcdhePublicKey *epub)
1618 {
1619   struct TcpHandshakeSignature ths;
1620   struct TCPConfirmation tc;
1621
1622   memcpy (queue->cwrite_buf,
1623           epub,
1624           sizeof (*epub));
1625   queue->cwrite_off = sizeof (epub);
1626   /* compute 'tc' and append in encrypted format to cwrite_buf */
1627   tc.sender = my_identity;
1628   tc.monotonic_time = GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get_monotonic (cfg));
1629   ths.purpose.purpose = htonl (GNUNET_SIGNATURE_COMMUNICATOR_TCP_HANDSHAKE);
1630   ths.purpose.size = htonl (sizeof (ths));
1631   ths.sender = my_identity;
1632   ths.receiver = queue->target;
1633   ths.ephemeral = *epub;
1634   ths.monotonic_time = tc.monotonic_time;
1635   GNUNET_assert (GNUNET_OK ==
1636                  GNUNET_CRYPTO_eddsa_sign (my_private_key,
1637                                            &ths.purpose,
1638                                            &tc.sender_sig));
1639   GNUNET_assert (0 ==
1640                  gcry_cipher_encrypt (queue->out_cipher,
1641                                       &queue->cwrite_buf[queue->cwrite_off],
1642                                       sizeof (tc),
1643                                       &tc,
1644                                       sizeof (tc)));
1645   queue->cwrite_off += sizeof (tc);
1646 }
1647
1648
1649 /**
1650  * Initialize our key material for outgoing transmissions and
1651  * inform the other peer about it. Must be called first before
1652  * any data is sent.
1653  *
1654  * @param queue the queue to setup
1655  */
1656 static void
1657 start_initial_kx_out (struct Queue *queue)
1658 {
1659   struct GNUNET_CRYPTO_EcdhePublicKey epub;
1660
1661   GNUNET_assert (GNUNET_OK ==
1662                  GNUNET_CRYPTO_ecdhe_key_create2 (&queue->ephemeral));
1663   GNUNET_CRYPTO_ecdhe_key_get_public (&queue->ephemeral,
1664                                       &epub);
1665   setup_out_cipher (queue);
1666   transmit_kx (queue,
1667                &epub);
1668 }
1669
1670
1671 /**
1672  * We have received the first bytes from the other side on a @a queue.
1673  * Decrypt the @a tc contained in @a ibuf and check the signature.
1674  * Note that #setup_in_cipher() must have already been called.
1675  *
1676  * @param queue queue to decrypt initial bytes from other peer for
1677  * @param tc[out] where to store the result
1678  * @param ibuf incoming data, of size
1679  *        `INITIAL_KX_SIZE`
1680  * @return #GNUNET_OK if the signature was OK, #GNUNET_SYSERR if not
1681  */
1682 static int
1683 decrypt_and_check_tc (struct Queue *queue,
1684                       struct TCPConfirmation *tc,
1685                       char *ibuf)
1686 {
1687   struct TcpHandshakeSignature ths;
1688
1689   GNUNET_assert (0 ==
1690                  gcry_cipher_decrypt (queue->in_cipher,
1691                                       tc,
1692                                       sizeof (*tc),
1693                                       &ibuf[sizeof (struct GNUNET_CRYPTO_EcdhePublicKey)],
1694                                       sizeof (tc)));
1695   ths.purpose.purpose = htonl (GNUNET_SIGNATURE_COMMUNICATOR_TCP_HANDSHAKE);
1696   ths.purpose.size = htonl (sizeof (ths));
1697   ths.sender = tc->sender;
1698   ths.receiver = my_identity;
1699   memcpy (&ths.ephemeral,
1700           ibuf,
1701           sizeof (struct GNUNET_CRYPTO_EcdhePublicKey));
1702   ths.monotonic_time = tc->monotonic_time;
1703   return GNUNET_CRYPTO_eddsa_verify (GNUNET_SIGNATURE_COMMUNICATOR_TCP_HANDSHAKE,
1704                                      &ths.purpose,
1705                                      &tc->sender_sig,
1706                                      &tc->sender.public_key);
1707 }
1708
1709
1710 /**
1711  * Closes socket and frees memory associated with @a pq.
1712  *
1713  * @param pq proto queue to free
1714  */
1715 static void
1716 free_proto_queue (struct ProtoQueue *pq)
1717 {
1718   GNUNET_NETWORK_socket_close (pq->sock);
1719   GNUNET_free (pq->address);
1720   GNUNET_CONTAINER_DLL_remove (proto_head,
1721                                proto_tail,
1722                                pq);
1723   GNUNET_free (pq);
1724 }
1725
1726
1727 /**
1728  * Read from the socket of the proto queue until we have enough data
1729  * to upgrade to full queue.
1730  *
1731  * @param cls a `struct ProtoQueue`
1732  */
1733 static void
1734 proto_read_kx (void *cls)
1735 {
1736   struct ProtoQueue *pq = cls;
1737   ssize_t rcvd;
1738   struct GNUNET_TIME_Relative left;
1739   struct Queue *queue;
1740   struct TCPConfirmation tc;
1741
1742   pq->read_task = NULL;
1743   left = GNUNET_TIME_absolute_get_remaining (pq->timeout);
1744   if (0 == left.rel_value_us)
1745   {
1746     free_proto_queue (pq);
1747     return;
1748   }
1749   rcvd = GNUNET_NETWORK_socket_recv (pq->sock,
1750                                      &pq->ibuf[pq->ibuf_off],
1751                                      sizeof (pq->ibuf) - pq->ibuf_off);
1752   if (-1 == rcvd)
1753   {
1754     if ( (EAGAIN != errno) &&
1755          (EINTR != errno) )
1756     {
1757       GNUNET_log_strerror (GNUNET_ERROR_TYPE_DEBUG,
1758                            "recv");
1759       free_proto_queue (pq);
1760       return;
1761     }
1762     /* try again */
1763     pq->read_task = GNUNET_SCHEDULER_add_read_net (left,
1764                                                    pq->sock,
1765                                                    &proto_read_kx,
1766                                                    pq);
1767     return;
1768   }
1769   pq->ibuf_off += rcvd;
1770   if (pq->ibuf_off > sizeof (pq->ibuf))
1771   {
1772     /* read more */
1773     pq->read_task = GNUNET_SCHEDULER_add_read_net (left,
1774                                                    pq->sock,
1775                                                    &proto_read_kx,
1776                                                    pq);
1777     return;
1778   }
1779   /* we got all the data, let's find out who we are talking to! */
1780   queue = GNUNET_new (struct Queue);
1781   setup_in_cipher ((const struct GNUNET_CRYPTO_EcdhePublicKey *) pq->ibuf,
1782                    queue);
1783   if (GNUNET_OK !=
1784       decrypt_and_check_tc (queue,
1785                             &tc,
1786                             pq->ibuf))
1787   {
1788     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1789                 "Invalid TCP KX received from %s\n",
1790                 GNUNET_a2s (queue->address,
1791                             queue->address_len));
1792     gcry_cipher_close (queue->in_cipher);
1793     GNUNET_free (queue);
1794     free_proto_queue (pq);
1795     return;
1796   }
1797   queue->address = pq->address; /* steals reference */
1798   queue->address_len = pq->address_len;
1799   queue->target = tc.sender;
1800   start_initial_kx_out (queue);
1801   boot_queue (queue,
1802               GNUNET_TRANSPORT_CS_INBOUND);
1803   queue->read_task
1804     = GNUNET_SCHEDULER_add_read_net (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
1805                                      queue->sock,
1806                                      &queue_read,
1807                                      queue);
1808   GNUNET_CONTAINER_DLL_remove (proto_head,
1809                                proto_tail,
1810                                pq);
1811   GNUNET_free (pq);
1812 }
1813
1814
1815 /**
1816  * We have been notified that our listen socket has something to
1817  * read. Do the read and reschedule this function to be called again
1818  * once more is available.
1819  *
1820  * @param cls NULL
1821  */
1822 static void
1823 listen_cb (void *cls)
1824 {
1825   struct sockaddr_storage in;
1826   socklen_t addrlen;
1827   struct GNUNET_NETWORK_Handle *sock;
1828   struct ProtoQueue *pq;
1829
1830   listen_task = NULL;
1831   GNUNET_assert (NULL != listen_sock);
1832   addrlen = sizeof (in);
1833   memset (&in,
1834           0,
1835           sizeof (in));
1836   sock = GNUNET_NETWORK_socket_accept (listen_sock,
1837                                        (struct sockaddr *) &in,
1838                                        &addrlen);
1839   if ( (NULL == sock) &&
1840        ( (EMFILE == errno) ||
1841          (ENFILE == errno) ) )
1842     return; /* system limit reached, wait until connection goes down */
1843   listen_task = GNUNET_SCHEDULER_add_read_net (GNUNET_TIME_UNIT_FOREVER_REL,
1844                                                listen_sock,
1845                                                &listen_cb,
1846                                                NULL);
1847   if ( (NULL == sock) &&
1848        ( (EAGAIN == errno) ||
1849          (ENOBUFS == errno) ) )
1850     return;
1851   if (NULL == sock)
1852   {
1853     GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING,
1854                          "accept");
1855     return;
1856   }
1857   pq = GNUNET_new (struct ProtoQueue);
1858   pq->address_len = addrlen;
1859   pq->address = GNUNET_memdup (&in,
1860                                addrlen);
1861   pq->timeout = GNUNET_TIME_relative_to_absolute (PROTO_QUEUE_TIMEOUT);
1862   pq->sock = sock;
1863   pq->read_task = GNUNET_SCHEDULER_add_read_net (PROTO_QUEUE_TIMEOUT,
1864                                                  pq->sock,
1865                                                  &proto_read_kx,
1866                                                  pq);
1867   GNUNET_CONTAINER_DLL_insert (proto_head,
1868                                proto_tail,
1869                                pq);
1870 }
1871
1872
1873 /**
1874  * Read from the socket of the queue until we have enough data
1875  * to initialize the decryption logic and can switch to regular
1876  * reading.
1877  *
1878  * @param cls a `struct Queue`
1879  */
1880 static void
1881 queue_read_kx (void *cls)
1882 {
1883   struct Queue *queue = cls;
1884   ssize_t rcvd;
1885   struct GNUNET_TIME_Relative left;
1886   struct TCPConfirmation tc;
1887
1888   queue->read_task = NULL;
1889   left = GNUNET_TIME_absolute_get_remaining (queue->timeout);
1890   if (0 == left.rel_value_us)
1891   {
1892     queue_destroy (queue);
1893     return;
1894   }
1895   rcvd = GNUNET_NETWORK_socket_recv (queue->sock,
1896                                      &queue->cread_buf[queue->cread_off],
1897                                      BUF_SIZE - queue->cread_off);
1898   if (-1 == rcvd)
1899   {
1900     if ( (EAGAIN != errno) &&
1901          (EINTR != errno) )
1902     {
1903       GNUNET_log_strerror (GNUNET_ERROR_TYPE_DEBUG,
1904                            "recv");
1905       queue_destroy (queue);
1906       return;
1907     }
1908     queue->read_task = GNUNET_SCHEDULER_add_read_net (left,
1909                                                       queue->sock,
1910                                                       &queue_read_kx,
1911                                                       queue);
1912     return;
1913   }
1914   queue->cread_off += rcvd;
1915   if (queue->cread_off <
1916       INITIAL_KX_SIZE)
1917   {
1918     /* read more */
1919     queue->read_task = GNUNET_SCHEDULER_add_read_net (left,
1920                                                       queue->sock,
1921                                                       &queue_read_kx,
1922                                                       queue);
1923     return;
1924   }
1925   /* we got all the data, let's find out who we are talking to! */
1926   setup_in_cipher ((const struct GNUNET_CRYPTO_EcdhePublicKey *) queue->cread_buf,
1927                    queue);
1928   if (GNUNET_OK !=
1929       decrypt_and_check_tc (queue,
1930                             &tc,
1931                             queue->cread_buf))
1932   {
1933     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1934                 "Invalid TCP KX received from %s\n",
1935                 GNUNET_a2s (queue->address,
1936                             queue->address_len));
1937     queue_destroy (queue);
1938     return;
1939   }
1940   if (0 != memcmp (&tc.sender,
1941                    &queue->target,
1942                    sizeof (struct GNUNET_PeerIdentity)))
1943   {
1944     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1945                 "Invalid sender in TCP KX received from %s\n",
1946                 GNUNET_a2s (queue->address,
1947                             queue->address_len));
1948     queue_destroy (queue);
1949     return;
1950   }
1951
1952   /* update queue timeout */
1953   reschedule_queue_timeout (queue);
1954   /* prepare to continue with regular read task immediately */
1955   memmove (queue->cread_buf,
1956            &queue->cread_buf[INITIAL_KX_SIZE],
1957            queue->cread_off - (INITIAL_KX_SIZE));
1958   queue->cread_off -= INITIAL_KX_SIZE;
1959   queue->read_task = GNUNET_SCHEDULER_add_now (&queue_read,
1960                                                queue);
1961 }
1962
1963
1964 /**
1965  * Function called by the transport service to initialize a
1966  * message queue given address information about another peer.
1967  * If and when the communication channel is established, the
1968  * communicator must call #GNUNET_TRANSPORT_communicator_mq_add()
1969  * to notify the service that the channel is now up.  It is
1970  * the responsibility of the communicator to manage sane
1971  * retries and timeouts for any @a peer/@a address combination
1972  * provided by the transport service.  Timeouts and retries
1973  * do not need to be signalled to the transport service.
1974  *
1975  * @param cls closure
1976  * @param peer identity of the other peer
1977  * @param address where to send the message, human-readable
1978  *        communicator-specific format, 0-terminated, UTF-8
1979  * @return #GNUNET_OK on success, #GNUNET_SYSERR if the provided address is invalid
1980  */
1981 static int
1982 mq_init (void *cls,
1983          const struct GNUNET_PeerIdentity *peer,
1984          const char *address)
1985 {
1986   struct Queue *queue;
1987   const char *path;
1988   struct sockaddr *in;
1989   socklen_t in_len;
1990   struct GNUNET_NETWORK_Handle *sock;
1991
1992   if (0 != strncmp (address,
1993                     COMMUNICATOR_ADDRESS_PREFIX "-",
1994                     strlen (COMMUNICATOR_ADDRESS_PREFIX "-")))
1995   {
1996     GNUNET_break_op (0);
1997     return GNUNET_SYSERR;
1998   }
1999   path = &address[strlen (COMMUNICATOR_ADDRESS_PREFIX "-")];
2000   in = tcp_address_to_sockaddr (path,
2001                                 &in_len);
2002
2003   sock = GNUNET_NETWORK_socket_create (in->sa_family,
2004                                        SOCK_STREAM,
2005                                        IPPROTO_TCP);
2006   if (NULL == sock)
2007   {
2008     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
2009                 "socket(%d) failed: %s",
2010                 in->sa_family,
2011                 STRERROR (errno));
2012     GNUNET_free (in);
2013     return GNUNET_SYSERR;
2014   }
2015   if (GNUNET_OK !=
2016       GNUNET_NETWORK_socket_connect (sock,
2017                                      in,
2018                                      in_len))
2019   {
2020     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
2021                 "connect to `%s' failed: %s",
2022                 address,
2023                 STRERROR (errno));
2024     GNUNET_NETWORK_socket_close (sock);
2025     GNUNET_free (in);
2026     return GNUNET_SYSERR;
2027   }
2028
2029   queue = GNUNET_new (struct Queue);
2030   queue->target = *peer;
2031   queue->address = in;
2032   queue->address_len = in_len;
2033   queue->sock = sock;
2034   boot_queue (queue,
2035               GNUNET_TRANSPORT_CS_OUTBOUND);
2036   queue->read_task
2037     = GNUNET_SCHEDULER_add_read_net (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
2038                                      queue->sock,
2039                                      &queue_read_kx,
2040                                      queue);
2041   if (NULL == queue)
2042   {
2043     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
2044                 "Failed to setup queue to %s at `%s'\n",
2045                 GNUNET_i2s (peer),
2046                 path);
2047     GNUNET_NETWORK_socket_close (sock);
2048     return GNUNET_NO;
2049   }
2050   start_initial_kx_out (queue);
2051   return GNUNET_OK;
2052 }
2053
2054
2055 /**
2056  * Iterator over all message queues to clean up.
2057  *
2058  * @param cls NULL
2059  * @param target unused
2060  * @param value the queue to destroy
2061  * @return #GNUNET_OK to continue to iterate
2062  */
2063 static int
2064 get_queue_delete_it (void *cls,
2065                      const struct GNUNET_PeerIdentity *target,
2066                      void *value)
2067 {
2068   struct Queue *queue = value;
2069
2070   (void) cls;
2071   (void) target;
2072   queue_destroy (queue);
2073   return GNUNET_OK;
2074 }
2075
2076
2077 /**
2078  * Shutdown the UNIX communicator.
2079  *
2080  * @param cls NULL (always)
2081  */
2082 static void
2083 do_shutdown (void *cls)
2084 {
2085   if (NULL != nat)
2086   {
2087      GNUNET_NAT_unregister (nat);
2088      nat = NULL;
2089   }
2090   if (NULL != listen_task)
2091   {
2092     GNUNET_SCHEDULER_cancel (listen_task);
2093     listen_task = NULL;
2094   }
2095   if (NULL != listen_sock)
2096   {
2097     GNUNET_break (GNUNET_OK ==
2098                   GNUNET_NETWORK_socket_close (listen_sock));
2099     listen_sock = NULL;
2100   }
2101   GNUNET_CONTAINER_multipeermap_iterate (queue_map,
2102                                          &get_queue_delete_it,
2103                                          NULL);
2104   GNUNET_CONTAINER_multipeermap_destroy (queue_map);
2105   if (NULL != ch)
2106   {
2107     GNUNET_TRANSPORT_communicator_disconnect (ch);
2108     ch = NULL;
2109   }
2110   if (NULL != stats)
2111   {
2112     GNUNET_STATISTICS_destroy (stats,
2113                                GNUNET_NO);
2114     stats = NULL;
2115   }
2116   if (NULL != my_private_key)
2117   {
2118     GNUNET_free (my_private_key);
2119     my_private_key = NULL;
2120   }
2121   if (NULL != is)
2122   {
2123      GNUNET_NT_scanner_done (is);
2124      is = NULL;
2125   }
2126 }
2127
2128
2129 /**
2130  * Function called when the transport service has received an
2131  * acknowledgement for this communicator (!) via a different return
2132  * path.
2133  *
2134  * Not applicable for TCP.
2135  *
2136  * @param cls closure
2137  * @param sender which peer sent the notification
2138  * @param msg payload
2139  */
2140 static void
2141 enc_notify_cb (void *cls,
2142                const struct GNUNET_PeerIdentity *sender,
2143                const struct GNUNET_MessageHeader *msg)
2144 {
2145   (void) cls;
2146   (void) sender;
2147   (void) msg;
2148   GNUNET_break_op (0);
2149 }
2150
2151
2152 /**
2153  * Signature of the callback passed to #GNUNET_NAT_register() for
2154  * a function to call whenever our set of 'valid' addresses changes.
2155  *
2156  * @param cls closure
2157  * @param add_remove #GNUNET_YES to add a new public IP address,
2158  *                   #GNUNET_NO to remove a previous (now invalid) one
2159  * @param ac address class the address belongs to
2160  * @param addr either the previous or the new public IP address
2161  * @param addrlen actual length of the @a addr
2162  */
2163 static void
2164 nat_address_cb (void *cls,
2165                 int add_remove,
2166                 enum GNUNET_NAT_AddressClass ac,
2167                 const struct sockaddr *addr,
2168                 socklen_t addrlen)
2169 {
2170   char *my_addr;
2171   static struct GNUNET_TRANSPORT_AddressIdentifier *ai; // FIXME: store in *ctx of NAT!
2172
2173   if (GNUNET_YES == add_remove)
2174   {
2175     // FIXME: do better job at stringification of @a addr?
2176     GNUNET_asprintf (&my_addr,
2177                      "%s-%s",
2178                      COMMUNICATOR_ADDRESS_PREFIX,
2179                      GNUNET_a2s (addr,
2180                                  addrlen));
2181     // FIXME: translate 'ac' to 'nt'?
2182     ai = GNUNET_TRANSPORT_communicator_address_add (ch,
2183                                                     my_addr,
2184                                                     GNUNET_NT_LOOPBACK, // FIXME: wrong NT!
2185                                                     GNUNET_TIME_UNIT_FOREVER_REL);
2186     GNUNET_free (my_addr);
2187   }
2188   else
2189   {
2190     // FIXME: support removal! => improve NAT API!
2191     GNUNET_TRANSPORT_communicator_address_remove (ai);
2192     ai = NULL;
2193   }
2194 }
2195
2196
2197 /**
2198  * Setup communicator and launch network interactions.
2199  *
2200  * @param cls NULL (always)
2201  * @param args remaining command-line arguments
2202  * @param cfgfile name of the configuration file used (for saving, can be NULL!)
2203  * @param c configuration
2204  */
2205 static void
2206 run (void *cls,
2207      char *const *args,
2208      const char *cfgfile,
2209      const struct GNUNET_CONFIGURATION_Handle *c)
2210 {
2211   char *bindto;
2212   struct sockaddr *in;
2213   socklen_t in_len;
2214
2215   (void) cls;
2216   cfg = c;
2217   if (GNUNET_OK !=
2218       GNUNET_CONFIGURATION_get_value_filename (cfg,
2219                                                COMMUNICATOR_CONFIG_SECTION,
2220                                                "BINDTO",
2221                                                &bindto))
2222   {
2223     GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
2224                                COMMUNICATOR_CONFIG_SECTION,
2225                                "BINDTO");
2226     return;
2227   }
2228   if (GNUNET_OK !=
2229       GNUNET_CONFIGURATION_get_value_number (cfg,
2230                                              COMMUNICATOR_CONFIG_SECTION,
2231                                              "MAX_QUEUE_LENGTH",
2232                                              &max_queue_length))
2233     max_queue_length = DEFAULT_MAX_QUEUE_LENGTH;
2234
2235   in = tcp_address_to_sockaddr (bindto,
2236                                 &in_len);
2237   if (NULL == in)
2238   {
2239     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
2240                 "Failed to setup TCP socket address with path `%s'\n",
2241                 bindto);
2242     GNUNET_free (bindto);
2243     return;
2244   }
2245   listen_sock = GNUNET_NETWORK_socket_create (in->sa_family,
2246                                               SOCK_STREAM,
2247                                               IPPROTO_TCP);
2248   if (NULL == listen_sock)
2249   {
2250     GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR,
2251                          "socket");
2252     GNUNET_free (in);
2253     GNUNET_free (bindto);
2254     return;
2255   }
2256   if (GNUNET_OK !=
2257       GNUNET_NETWORK_socket_bind (listen_sock,
2258                                   in,
2259                                   in_len))
2260   {
2261     GNUNET_log_strerror_file (GNUNET_ERROR_TYPE_ERROR,
2262                               "bind",
2263                               bindto);
2264     GNUNET_NETWORK_socket_close (listen_sock);
2265     listen_sock = NULL;
2266     GNUNET_free (in);
2267     GNUNET_free (bindto);
2268     return;
2269   }
2270   GNUNET_free (in);
2271   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2272               "Bound to `%s'\n",
2273               bindto);
2274   GNUNET_free (bindto);
2275   stats = GNUNET_STATISTICS_create ("C-TCP",
2276                                     cfg);
2277   GNUNET_SCHEDULER_add_shutdown (&do_shutdown,
2278                                  NULL);
2279   is = GNUNET_NT_scanner_init ();
2280   my_private_key = GNUNET_CRYPTO_eddsa_key_create_from_configuration (cfg);
2281   if (NULL == my_private_key)
2282   {
2283     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
2284                 _("Transport service is lacking key configuration settings. Exiting.\n"));
2285     GNUNET_SCHEDULER_shutdown ();
2286     return;
2287   }
2288   GNUNET_CRYPTO_eddsa_key_get_public (my_private_key,
2289                                       &my_identity.public_key);
2290   /* start listening */
2291   listen_task = GNUNET_SCHEDULER_add_read_net (GNUNET_TIME_UNIT_FOREVER_REL,
2292                                                listen_sock,
2293                                                &listen_cb,
2294                                                NULL);
2295   queue_map = GNUNET_CONTAINER_multipeermap_create (10,
2296                                                     GNUNET_NO);
2297   ch = GNUNET_TRANSPORT_communicator_connect (cfg,
2298                                               COMMUNICATOR_CONFIG_SECTION,
2299                                               COMMUNICATOR_ADDRESS_PREFIX,
2300                                               GNUNET_TRANSPORT_CC_RELIABLE,
2301                                               &mq_init,
2302                                               NULL,
2303                                               &enc_notify_cb,
2304                                               NULL);
2305   if (NULL == ch)
2306   {
2307     GNUNET_break (0);
2308     GNUNET_SCHEDULER_shutdown ();
2309     return;
2310   }
2311   nat = GNUNET_NAT_register (cfg,
2312                              COMMUNICATOR_CONFIG_SECTION,
2313                              IPPROTO_TCP,
2314                              1 /* one address */,
2315                              (const struct sockaddr **) &in,
2316                              &in_len,
2317                              &nat_address_cb,
2318                              NULL /* FIXME: support reversal! */,
2319                              NULL /* closure */);
2320 }
2321
2322
2323 /**
2324  * The main function for the UNIX communicator.
2325  *
2326  * @param argc number of arguments from the command line
2327  * @param argv command line arguments
2328  * @return 0 ok, 1 on error
2329  */
2330 int
2331 main (int argc,
2332       char *const *argv)
2333 {
2334   static const struct GNUNET_GETOPT_CommandLineOption options[] = {
2335     GNUNET_GETOPT_OPTION_END
2336   };
2337   int ret;
2338
2339   if (GNUNET_OK !=
2340       GNUNET_STRINGS_get_utf8_args (argc, argv,
2341                                     &argc, &argv))
2342     return 2;
2343
2344   ret =
2345       (GNUNET_OK ==
2346        GNUNET_PROGRAM_run (argc, argv,
2347                            "gnunet-communicator-tcp",
2348                            _("GNUnet TCP communicator"),
2349                            options,
2350                            &run,
2351                            NULL)) ? 0 : 1;
2352   GNUNET_free ((void*) argv);
2353   return ret;
2354 }
2355
2356
2357 #if defined(LINUX) && defined(__GLIBC__)
2358 #include <malloc.h>
2359
2360 /**
2361  * MINIMIZE heap size (way below 128k) since this process doesn't need much.
2362  */
2363 void __attribute__ ((constructor))
2364 GNUNET_ARM_memory_init ()
2365 {
2366   mallopt (M_TRIM_THRESHOLD, 4 * 1024);
2367   mallopt (M_TOP_PAD, 1 * 1024);
2368   malloc_trim (0);
2369 }
2370 #endif
2371
2372 /* end of gnunet-communicator-tcp.c */