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