src/multicast/gnunet-service-multicast.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2009 GNUnet e.V.
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      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      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
  18      Boston, MA 02110-1301, USA.
  19 */
  20
  21 /**
  22  * @file multicast/gnunet-service-multicast.c
  23  * @brief program that does multicast
  24  * @author Christian Grothoff
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "gnunet_signatures.h"
  29 #include "gnunet_applications.h"
  30 #include "gnunet_statistics_service.h"
  31 #include "gnunet_core_service.h"
  32 #include "gnunet_cadet_service.h"
  33 #include "gnunet_multicast_service.h"
  34 #include "multicast.h"
  35
  36 /**
  37  * Handle to our current configuration.
  38  */
  39 static const struct GNUNET_CONFIGURATION_Handle *cfg;
  40
  41 /**
  42  * Server handle.
  43  */
  44 static struct GNUNET_SERVER_Handle *server;
  45
  46 /**
  47  * Core handle.
  48  * Only used during initialization.
  49  */
  50 static struct GNUNET_CORE_Handle *core;
  51
  52 /**
  53  * CADET handle.
  54  */
  55 static struct GNUNET_CADET_Handle *cadet;
  56
  57 /**
  58  * Identity of this peer.
  59  */
  60 static struct GNUNET_PeerIdentity this_peer;
  61
  62 /**
  63  * Handle to the statistics service.
  64  */
  65 static struct GNUNET_STATISTICS_Handle *stats;
  66
  67 /**
  68  * Notification context, simplifies client broadcasts.
  69  */
  70 static struct GNUNET_SERVER_NotificationContext *nc;
  71
  72 /**
  73  * All connected origin clients.
  74  * Group's pub_key_hash -> struct Origin * (uniq)
  75  */
  76 static struct GNUNET_CONTAINER_MultiHashMap *origins;
  77
  78 /**
  79  * All connected member clients.
  80  * Group's pub_key_hash -> struct Member * (multi)
  81  */
  82 static struct GNUNET_CONTAINER_MultiHashMap *members;
  83
  84 /**
  85  * Connected member clients per group.
  86  * Group's pub_key_hash -> Member's pub_key_hash (uniq) -> struct Member * (uniq)
  87  */
  88 static struct GNUNET_CONTAINER_MultiHashMap *group_members;
  89
  90 /**
  91  * Incoming CADET channels with connected children in the tree.
  92  * Group's pub_key_hash -> struct Channel * (multi)
  93  */
  94 static struct GNUNET_CONTAINER_MultiHashMap *channels_in;
  95
  96 /**
  97  * Outgoing CADET channels connecting to parents in the tree.
  98  * Group's pub_key_hash -> struct Channel * (multi)
  99  */
 100 static struct GNUNET_CONTAINER_MultiHashMap *channels_out;
 101
 102 /**
 103  * Incoming replay requests from CADET.
 104  * Group's pub_key_hash ->
 105  *   H(fragment_id, message_id, fragment_offset, flags) -> struct Channel *
 106  */
 107 static struct GNUNET_CONTAINER_MultiHashMap *replay_req_cadet;
 108
 109 /**
 110  * Incoming replay requests from clients.
 111  * Group's pub_key_hash ->
 112  *   H(fragment_id, message_id, fragment_offset, flags) -> struct GNUNET_SERVER_Client *
 113  */
 114 static struct GNUNET_CONTAINER_MultiHashMap *replay_req_client;
 115
 116
 117 /**
 118  * Join status of a remote peer.
 119  */
 120 enum JoinStatus
 121 {
 122   JOIN_REFUSED  = -1,
 123   JOIN_NOT_ASKED = 0,
 124   JOIN_WAITING   = 1,
 125   JOIN_ADMITTED  = 2,
 126 };
 127
 128 enum ChannelDirection
 129 {
 130   DIR_INCOMING = 0,
 131   DIR_OUTGOING = 1,
 132 };
 133
 134
 135 /**
 136  * Context for a CADET channel.
 137  */
 138 struct Channel
 139 {
 140   /**
 141    * Group the channel belongs to.
 142    *
 143    * Only set for outgoing channels.
 144    */
 145   struct Group *grp;
 146
 147   /**
 148    * CADET channel.
 149    */
 150   struct GNUNET_CADET_Channel *channel;
 151
 152   /**
 153    * CADET transmission handle.
 154    */
 155   struct GNUNET_CADET_TransmitHandle *tmit_handle;
 156
 157   /**
 158    * Public key of the target group.
 159    */
 160   struct GNUNET_CRYPTO_EddsaPublicKey group_pub_key;
 161
 162   /**
 163    * Hash of @a group_pub_key.
 164    */
 165   struct GNUNET_HashCode group_pub_hash;
 166
 167   /**
 168    * Public key of the joining member.
 169    */
 170   struct GNUNET_CRYPTO_EcdsaPublicKey member_pub_key;
 171
 172   /**
 173    * Remote peer identity.
 174    */
 175   struct GNUNET_PeerIdentity peer;
 176
 177   /**
 178    * Is the remote peer admitted to the group?
 179    * @see enum JoinStatus
 180    */
 181   int8_t join_status;
 182
 183   /**
 184    * Number of messages waiting to be sent to CADET.
 185    */
 186   uint8_t msgs_pending;
 187
 188   /**
 189    * Channel direction.
 190    * @see enum ChannelDirection
 191    */
 192   uint8_t direction;
 193 };
 194
 195
 196 /**
 197  * List of connected clients.
 198  */
 199 struct ClientList
 200 {
 201   struct ClientList *prev;
 202   struct ClientList *next;
 203   struct GNUNET_SERVER_Client *client;
 204 };
 205
 206 /**
 207  * Common part of the client context for both an origin and member.
 208  */
 209 struct Group
 210 {
 211   struct ClientList *clients_head;
 212   struct ClientList *clients_tail;
 213
 214   /**
 215    * Public key of the group.
 216    */
 217   struct GNUNET_CRYPTO_EddsaPublicKey pub_key;
 218
 219   /**
 220    * Hash of @a pub_key.
 221    */
 222   struct GNUNET_HashCode pub_key_hash;
 223
 224   /**
 225    * Is this an origin (#GNUNET_YES), or member (#GNUNET_NO)?
 226    */
 227   uint8_t is_origin;
 228
 229   /**
 230    * Is the client disconnected? #GNUNET_YES or #GNUNET_NO
 231    */
 232   uint8_t disconnected;
 233 };
 234
 235
 236 /**
 237  * Client context for a group's origin.
 238  */
 239 struct Origin
 240 {
 241   struct Group grp;
 242
 243   /**
 244    * Private key of the group.
 245    */
 246   struct GNUNET_CRYPTO_EddsaPrivateKey priv_key;
 247
 248   /**
 249    * Last message fragment ID sent to the group.
 250    */
 251   uint64_t max_fragment_id;
 252 };
 253
 254
 255 /**
 256  * Client context for a group member.
 257  */
 258 struct Member
 259 {
 260   struct Group grp;
 261
 262   /**
 263    * Private key of the member.
 264    */
 265   struct GNUNET_CRYPTO_EcdsaPrivateKey priv_key;
 266
 267   /**
 268    * Public key of the member.
 269    */
 270   struct GNUNET_CRYPTO_EcdsaPublicKey pub_key;
 271
 272   /**
 273    * Hash of @a pub_key.
 274    */
 275   struct GNUNET_HashCode pub_key_hash;
 276
 277   /**
 278    * Join request sent to the origin / members.
 279    */
 280   struct MulticastJoinRequestMessage *join_req;
 281
 282   /**
 283    * Join decision sent in reply to our request.
 284    *
 285    * Only a positive decision is stored here, in case of a negative decision the
 286    * client is disconnected.
 287    */
 288   struct MulticastJoinDecisionMessageHeader *join_dcsn;
 289
 290   /**
 291    * CADET channel to the origin.
 292    */
 293   struct Channel *origin_channel;
 294
 295   /**
 296    * Peer identity of origin.
 297    */
 298   struct GNUNET_PeerIdentity origin;
 299
 300   /**
 301    * Peer identity of relays (other members to connect).
 302    */
 303   struct GNUNET_PeerIdentity *relays;
 304
 305   /**
 306    * Last request fragment ID sent to the origin.
 307    */
 308   uint64_t max_fragment_id;
 309
 310   /**
 311    * Number of @a relays.
 312    */
 313   uint32_t relay_count;
 314 };
 315
 316
 317 struct ReplayRequestKey
 318 {
 319   uint64_t fragment_id;
 320   uint64_t message_id;
 321   uint64_t fragment_offset;
 322   uint64_t flags;
 323 };
 324
 325
 326 /**
 327  * Task run during shutdown.
 328  *
 329  * @param cls unused
 330  */
 331 static void
 332 shutdown_task (void *cls)
 333 {
 334   if (NULL != core)
 335   {
 336     GNUNET_CORE_disconnect (core);
 337     core = NULL;
 338   }
 339   if (NULL != cadet)
 340   {
 341     GNUNET_CADET_disconnect (cadet);
 342     cadet = NULL;
 343   }
 344   if (NULL != stats)
 345   {
 346     GNUNET_STATISTICS_destroy (stats, GNUNET_YES);
 347     stats = NULL;
 348   }
 349   /* FIXME: do more clean up here */
 350 }
 351
 352
 353 /**
 354  * Clean up origin data structures after a client disconnected.
 355  */
 356 static void
 357 cleanup_origin (struct Origin *orig)
 358 {
 359   struct Group *grp = &orig->grp;
 360   GNUNET_CONTAINER_multihashmap_remove (origins, &grp->pub_key_hash, orig);
 361 }
 362
 363
 364 /**
 365  * Clean up member data structures after a client disconnected.
 366  */
 367 static void
 368 cleanup_member (struct Member *mem)
 369 {
 370   struct Group *grp = &mem->grp;
 371   struct GNUNET_CONTAINER_MultiHashMap *
 372     grp_mem = GNUNET_CONTAINER_multihashmap_get (group_members,
 373                                                  &grp->pub_key_hash);
 374   GNUNET_assert (NULL != grp_mem);
 375   GNUNET_CONTAINER_multihashmap_remove (grp_mem, &mem->pub_key_hash, mem);
 376
 377   if (0 == GNUNET_CONTAINER_multihashmap_size (grp_mem))
 378   {
 379     GNUNET_CONTAINER_multihashmap_remove (group_members, &grp->pub_key_hash,
 380                                           grp_mem);
 381     GNUNET_CONTAINER_multihashmap_destroy (grp_mem);
 382   }
 383   if (NULL != mem->join_dcsn)
 384   {
 385     GNUNET_free (mem->join_dcsn);
 386     mem->join_dcsn = NULL;
 387   }
 388   GNUNET_CONTAINER_multihashmap_remove (members, &grp->pub_key_hash, mem);
 389 }
 390
 391
 392 /**
 393  * Clean up group data structures after a client disconnected.
 394  */
 395 static void
 396 cleanup_group (struct Group *grp)
 397 {
 398   (GNUNET_YES == grp->is_origin)
 399     ? cleanup_origin ((struct Origin *) grp)
 400     : cleanup_member ((struct Member *) grp);
 401
 402   GNUNET_free (grp);
 403 }
 404
 405
 406 void
 407 replay_key_hash (uint64_t fragment_id, uint64_t message_id,
 408                  uint64_t fragment_offset, uint64_t flags,
 409                  struct GNUNET_HashCode *key_hash)
 410 {
 411   struct ReplayRequestKey key = {
 412     .fragment_id = fragment_id,
 413     .message_id = message_id,
 414     .fragment_offset = fragment_offset,
 415     .flags = flags,
 416   };
 417   GNUNET_CRYPTO_hash (&key, sizeof (key), key_hash);
 418 }
 419
 420
 421 /**
 422  * Remove channel from replay request hashmap.
 423  *
 424  * @param chn
 425  *        Channel to remove.
 426  *
 427  * @return #GNUNET_YES if there are more entries to process,
 428  *         #GNUNET_NO when reached end of hashmap.
 429  */
 430 static int
 431 replay_req_remove_cadet (struct Channel *chn)
 432 {
 433   struct GNUNET_CONTAINER_MultiHashMap *
 434     grp_replay_req = GNUNET_CONTAINER_multihashmap_get (replay_req_cadet,
 435                                                         &chn->grp->pub_key_hash);
 436   if (NULL == grp_replay_req)
 437     return GNUNET_NO;
 438
 439   struct GNUNET_CONTAINER_MultiHashMapIterator *
 440     it = GNUNET_CONTAINER_multihashmap_iterator_create (grp_replay_req);
 441   struct GNUNET_HashCode key;
 442   const struct Channel *c;
 443   while (GNUNET_YES
 444          == GNUNET_CONTAINER_multihashmap_iterator_next (it, &key,
 445                                                          (const void **) &c))
 446   {
 447     if (c == chn)
 448     {
 449       GNUNET_CONTAINER_multihashmap_remove (grp_replay_req, &key, chn);
 450       return GNUNET_YES;
 451     }
 452   }
 453   GNUNET_CONTAINER_multihashmap_iterator_destroy (it);
 454   return GNUNET_NO;
 455 }
 456
 457
 458 /**
 459  * Remove client from replay request hashmap.
 460  *
 461  * @param client
 462  *        Client to remove.
 463  *
 464  * @return #GNUNET_YES if there are more entries to process,
 465  *         #GNUNET_NO when reached end of hashmap.
 466  */
 467 static int
 468 replay_req_remove_client (struct Group *grp, struct GNUNET_SERVER_Client *client)
 469 {
 470   struct GNUNET_CONTAINER_MultiHashMap *
 471     grp_replay_req = GNUNET_CONTAINER_multihashmap_get (replay_req_client,
 472                                                         &grp->pub_key_hash);
 473   if (NULL == grp_replay_req)
 474     return GNUNET_NO;
 475
 476   struct GNUNET_CONTAINER_MultiHashMapIterator *
 477     it = GNUNET_CONTAINER_multihashmap_iterator_create (grp_replay_req);
 478   struct GNUNET_HashCode key;
 479   const struct GNUNET_SERVER_Client *c;
 480   while (GNUNET_YES
 481          == GNUNET_CONTAINER_multihashmap_iterator_next (it, &key,
 482                                                          (const void **) &c))
 483   {
 484     if (c == client)
 485     {
 486       GNUNET_CONTAINER_multihashmap_remove (replay_req_client, &key, client);
 487       return GNUNET_YES;
 488     }
 489   }
 490   GNUNET_CONTAINER_multihashmap_iterator_destroy (it);
 491   return GNUNET_NO;
 492 }
 493
 494
 495 /**
 496  * Called whenever a client is disconnected.
 497  *
 498  * Frees our resources associated with that client.
 499  *
 500  * @param cls  Closure.
 501  * @param client  Client handle.
 502  */
 503 static void
 504 client_notify_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
 505 {
 506   if (NULL == client)
 507     return;
 508
 509   struct Group *grp
 510     = GNUNET_SERVER_client_get_user_context (client, struct Group);
 511
 512   if (NULL == grp)
 513   {
 514     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 515                 "%p User context is NULL in client_disconnect()\n", grp);
 516     GNUNET_break (0);
 517     return;
 518   }
 519
 520   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 521               "%p Client (%s) disconnected from group %s\n",
 522               grp, (GNUNET_YES == grp->is_origin) ? "origin" : "member",
 523               GNUNET_h2s (&grp->pub_key_hash));
 524
 525   struct ClientList *cl = grp->clients_head;
 526   while (NULL != cl)
 527   {
 528     if (cl->client == client)
 529     {
 530       GNUNET_CONTAINER_DLL_remove (grp->clients_head, grp->clients_tail, cl);
 531       GNUNET_free (cl);
 532       break;
 533     }
 534     cl = cl->next;
 535   }
 536
 537   while (GNUNET_YES == replay_req_remove_client (grp, client));
 538
 539   if (NULL == grp->clients_head)
 540   { /* Last client disconnected. */
 541 #if FIXME
 542     if (NULL != grp->tmit_head)
 543     { /* Send pending messages via CADET before cleanup. */
 544       transmit_message (grp);
 545     }
 546     else
 547 #endif
 548     {
 549       cleanup_group (grp);
 550     }
 551   }
 552 }
 553
 554
 555 /**
 556  * Send message to a client.
 557  */
 558 static void
 559 client_send (struct GNUNET_SERVER_Client *client,
 560              const struct GNUNET_MessageHeader *msg)
 561 {
 562   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 563               "%p Sending message to client.\n", client);
 564
 565   GNUNET_SERVER_notification_context_add (nc, client);
 566   GNUNET_SERVER_notification_context_unicast (nc, client, msg, GNUNET_NO);
 567 }
 568
 569
 570 /**
 571  * Send message to all clients connected to the group.
 572  */
 573 static void
 574 client_send_group (const struct Group *grp,
 575                    const struct GNUNET_MessageHeader *msg)
 576 {
 577   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 578               "%p Sending message to all clients of the group.\n", grp);
 579
 580   struct ClientList *cl = grp->clients_head;
 581   while (NULL != cl)
 582   {
 583     GNUNET_SERVER_notification_context_add (nc, cl->client);
 584     GNUNET_SERVER_notification_context_unicast (nc, cl->client, msg, GNUNET_NO);
 585     cl = cl->next;
 586   }
 587 }
 588
 589
 590 /**
 591  * Iterator callback for sending a message to origin clients.
 592  */
 593 static int
 594 client_send_origin_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash,
 595                        void *origin)
 596 {
 597   const struct GNUNET_MessageHeader *msg = cls;
 598   struct Member *orig = origin;
 599
 600   client_send_group (&orig->grp, msg);
 601   return GNUNET_YES;
 602 }
 603
 604
 605 /**
 606  * Iterator callback for sending a message to member clients.
 607  */
 608 static int
 609 client_send_member_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash,
 610                        void *member)
 611 {
 612   const struct GNUNET_MessageHeader *msg = cls;
 613   struct Member *mem = member;
 614
 615   if (NULL != mem->join_dcsn)
 616   { /* Only send message to admitted members */
 617     client_send_group (&mem->grp, msg);
 618   }
 619   return GNUNET_YES;
 620 }
 621
 622
 623 /**
 624  * Send message to all origin and member clients connected to the group.
 625  *
 626  * @param pub_key_hash
 627  *        H(key_pub) of the group.
 628  * @param msg
 629  *        Message to send.
 630  */
 631 static int
 632 client_send_all (struct GNUNET_HashCode *pub_key_hash,
 633                  const struct GNUNET_MessageHeader *msg)
 634 {
 635   int n = 0;
 636   n += GNUNET_CONTAINER_multihashmap_get_multiple (origins, pub_key_hash,
 637                                                    client_send_origin_cb,
 638                                                    (void *) msg);
 639   n += GNUNET_CONTAINER_multihashmap_get_multiple (members, pub_key_hash,
 640                                                    client_send_member_cb,
 641                                                    (void *) msg);
 642   return n;
 643 }
 644
 645
 646 /**
 647  * Send message to a random origin client or a random member client.
 648  *
 649  * @param grp  The group to send @a msg to.
 650  * @param msg  Message to send.
 651  */
 652 static int
 653 client_send_random (struct GNUNET_HashCode *pub_key_hash,
 654                     const struct GNUNET_MessageHeader *msg)
 655 {
 656   int n = 0;
 657   n = GNUNET_CONTAINER_multihashmap_get_random (origins, client_send_origin_cb,
 658                                                  (void *) msg);
 659   if (n <= 0)
 660     n = GNUNET_CONTAINER_multihashmap_get_random (members, client_send_member_cb,
 661                                                    (void *) msg);
 662   return n;
 663 }
 664
 665
 666 /**
 667  * Send message to all origin clients connected to the group.
 668  *
 669  * @param pub_key_hash
 670  *        H(key_pub) of the group.
 671  * @param msg
 672  *        Message to send.
 673  */
 674 static int
 675 client_send_origin (struct GNUNET_HashCode *pub_key_hash,
 676                     const struct GNUNET_MessageHeader *msg)
 677 {
 678   int n = 0;
 679   n += GNUNET_CONTAINER_multihashmap_get_multiple (origins, pub_key_hash,
 680                                                    client_send_origin_cb,
 681                                                    (void *) msg);
 682   return n;
 683 }
 684
 685
 686 /**
 687  * Send fragment acknowledgement to all clients of the channel.
 688  *
 689  * @param pub_key_hash
 690  *        H(key_pub) of the group.
 691  */
 692 static void
 693 client_send_ack (struct GNUNET_HashCode *pub_key_hash)
 694 {
 695   static struct GNUNET_MessageHeader *msg = NULL;
 696   if (NULL == msg)
 697   {
 698     msg = GNUNET_malloc (sizeof (*msg));
 699     msg->type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_FRAGMENT_ACK);
 700     msg->size = htons (sizeof (*msg));
 701   }
 702   client_send_all (pub_key_hash, msg);
 703 }
 704
 705
 706 struct CadetTransmitClosure
 707 {
 708   struct Channel *chn;
 709   const struct GNUNET_MessageHeader *msg;
 710 };
 711
 712
 713 /**
 714  * CADET is ready to transmit a message.
 715  */
 716 size_t
 717 cadet_notify_transmit_ready (void *cls, size_t buf_size, void *buf)
 718 {
 719   if (0 == buf_size)
 720   {
 721     /* FIXME: connection closed */
 722     return 0;
 723   }
 724   struct CadetTransmitClosure *tcls = cls;
 725   struct Channel *chn = tcls->chn;
 726   uint16_t msg_size = ntohs (tcls->msg->size);
 727   GNUNET_assert (msg_size <= buf_size);
 728   memcpy (buf, tcls->msg, msg_size);
 729   GNUNET_free (tcls);
 730
 731   if (0 == chn->msgs_pending)
 732   {
 733     GNUNET_break (0);
 734   }
 735   else if (0 == --chn->msgs_pending)
 736   {
 737     client_send_ack (&chn->group_pub_hash);
 738   }
 739   return msg_size;
 740 }
 741
 742
 743 /**
 744  * Send a message to a CADET channel.
 745  *
 746  * @param chn  Channel.
 747  * @param msg  Message.
 748  */
 749 static void
 750 cadet_send_channel (struct Channel *chn, const struct GNUNET_MessageHeader *msg)
 751 {
 752   struct CadetTransmitClosure *tcls = GNUNET_malloc (sizeof (*tcls));
 753   tcls->chn = chn;
 754   tcls->msg = msg;
 755
 756   chn->msgs_pending++;
 757   chn->tmit_handle
 758     = GNUNET_CADET_notify_transmit_ready (chn->channel, GNUNET_NO,
 759                                           GNUNET_TIME_UNIT_FOREVER_REL,
 760                                           ntohs (msg->size),
 761                                           &cadet_notify_transmit_ready,
 762                                           (void *) msg);
 763   GNUNET_assert (NULL != chn->tmit_handle);
 764 }
 765
 766
 767 /**
 768  * Create new outgoing CADET channel.
 769  *
 770  * @param peer
 771  *        Peer to connect to.
 772  * @param group_pub_key
 773  *        Public key of group the channel belongs to.
 774  * @param group_pub_hash
 775  *        Hash of @a group_pub_key.
 776  *
 777  * @return Channel.
 778  */
 779 static struct Channel *
 780 cadet_channel_create (struct Group *grp, struct GNUNET_PeerIdentity *peer)
 781 {
 782   struct Channel *chn = GNUNET_malloc (sizeof (*chn));
 783   chn->grp = grp;
 784   chn->group_pub_key = grp->pub_key;
 785   chn->group_pub_hash = grp->pub_key_hash;
 786   chn->peer = *peer;
 787   chn->direction = DIR_OUTGOING;
 788   chn->join_status = JOIN_WAITING;
 789   chn->channel = GNUNET_CADET_channel_create (cadet, chn, &chn->peer,
 790                                               GNUNET_APPLICATION_TYPE_MULTICAST,
 791                                               GNUNET_CADET_OPTION_RELIABLE);
 792   GNUNET_CONTAINER_multihashmap_put (channels_out, &chn->group_pub_hash, chn,
 793                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
 794   return chn;
 795 }
 796
 797
 798 /**
 799  * Create CADET channel and send a join request.
 800  */
 801 static void
 802 cadet_send_join_request (struct Member *mem)
 803 {
 804   mem->origin_channel = cadet_channel_create (&mem->grp, &mem->origin);
 805   cadet_send_channel (mem->origin_channel, &mem->join_req->header);
 806
 807   uint32_t i;
 808   for (i = 0; i < mem->relay_count; i++)
 809   {
 810     struct Channel *
 811       chn = cadet_channel_create (&mem->grp, &mem->relays[i]);
 812     cadet_send_channel (chn, &mem->join_req->header);
 813   }
 814 }
 815
 816
 817 static int
 818 cadet_send_join_decision_cb (void *cls,
 819                              const struct GNUNET_HashCode *group_pub_hash,
 820                              void *channel)
 821 {
 822   const struct MulticastJoinDecisionMessageHeader *hdcsn = cls;
 823   struct Channel *chn = channel;
 824
 825   if (0 == memcmp (&hdcsn->member_pub_key, &chn->member_pub_key, sizeof (chn->member_pub_key))
 826       && 0 == memcmp (&hdcsn->peer, &chn->peer, sizeof (chn->peer)))
 827   {
 828     cadet_send_channel (chn, &hdcsn->header);
 829     return GNUNET_NO;
 830   }
 831   return GNUNET_YES;
 832 }
 833
 834
 835 /**
 836  * Send join decision to a remote peer.
 837  */
 838 static void
 839 cadet_send_join_decision (struct Group *grp,
 840                           const struct MulticastJoinDecisionMessageHeader *hdcsn)
 841 {
 842   GNUNET_CONTAINER_multihashmap_get_multiple (channels_in, &grp->pub_key_hash,
 843                                               &cadet_send_join_decision_cb,
 844                                               (void *) hdcsn);
 845 }
 846
 847
 848 /**
 849  * Iterator callback for sending a message to origin clients.
 850  */
 851 static int
 852 cadet_send_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash,
 853                void *channel)
 854 {
 855   const struct GNUNET_MessageHeader *msg = cls;
 856   struct Channel *chn = channel;
 857   if (JOIN_ADMITTED == chn->join_status)
 858     cadet_send_channel (chn, msg);
 859   return GNUNET_YES;
 860 }
 861
 862
 863 /**
 864  * Send message to all connected children.
 865  */
 866 static int
 867 cadet_send_children (struct GNUNET_HashCode *pub_key_hash,
 868                      const struct GNUNET_MessageHeader *msg)
 869 {
 870   int n = 0;
 871   if (channels_in != NULL)
 872     n += GNUNET_CONTAINER_multihashmap_get_multiple (channels_in, pub_key_hash,
 873                                                      cadet_send_cb, (void *) msg);
 874   return n;
 875 }
 876
 877
 878 /**
 879  * Send message to all connected parents.
 880  */
 881 static int
 882 cadet_send_parents (struct GNUNET_HashCode *pub_key_hash,
 883                     const struct GNUNET_MessageHeader *msg)
 884 {
 885   int n = 0;
 886   if (channels_in != NULL)
 887     n += GNUNET_CONTAINER_multihashmap_get_multiple (channels_out, pub_key_hash,
 888                                                      cadet_send_cb, (void *) msg);
 889   return n;
 890 }
 891
 892
 893 /**
 894  * Handle a connecting client starting an origin.
 895  */
 896 static void
 897 client_recv_origin_start (void *cls, struct GNUNET_SERVER_Client *client,
 898                           const struct GNUNET_MessageHeader *m)
 899 {
 900   const struct MulticastOriginStartMessage *
 901     msg = (const struct MulticastOriginStartMessage *) m;
 902
 903   struct GNUNET_CRYPTO_EddsaPublicKey pub_key;
 904   struct GNUNET_HashCode pub_key_hash;
 905
 906   GNUNET_CRYPTO_eddsa_key_get_public (&msg->group_key, &pub_key);
 907   GNUNET_CRYPTO_hash (&pub_key, sizeof (pub_key), &pub_key_hash);
 908
 909   struct Origin *
 910     orig = GNUNET_CONTAINER_multihashmap_get (origins, &pub_key_hash);
 911   struct Group *grp;
 912
 913   if (NULL == orig)
 914   {
 915     orig = GNUNET_new (struct Origin);
 916     orig->priv_key = msg->group_key;
 917     orig->max_fragment_id = GNUNET_ntohll (msg->max_fragment_id);
 918     grp = &orig->grp;
 919     grp->is_origin = GNUNET_YES;
 920     grp->pub_key = pub_key;
 921     grp->pub_key_hash = pub_key_hash;
 922
 923     GNUNET_CONTAINER_multihashmap_put (origins, &grp->pub_key_hash, orig,
 924                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
 925   }
 926   else
 927   {
 928     grp = &orig->grp;
 929   }
 930
 931   struct ClientList *cl = GNUNET_new (struct ClientList);
 932   cl->client = client;
 933   GNUNET_CONTAINER_DLL_insert (grp->clients_head, grp->clients_tail, cl);
 934
 935   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 936               "%p Client connected as origin to group %s.\n",
 937               orig, GNUNET_h2s (&grp->pub_key_hash));
 938
 939   GNUNET_SERVER_client_set_user_context (client, grp);
 940   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 941 }
 942
 943
 944 /**
 945  * Handle a connecting client joining a group.
 946  */
 947 static void
 948 client_recv_member_join (void *cls, struct GNUNET_SERVER_Client *client,
 949                          const struct GNUNET_MessageHeader *m)
 950 {
 951   const struct MulticastMemberJoinMessage *
 952     msg = (const struct MulticastMemberJoinMessage *) m;
 953   uint16_t msg_size = ntohs (msg->header.size);
 954
 955   struct GNUNET_CRYPTO_EcdsaPublicKey mem_pub_key;
 956   struct GNUNET_HashCode pub_key_hash, mem_pub_key_hash;
 957
 958   GNUNET_CRYPTO_ecdsa_key_get_public (&msg->member_key, &mem_pub_key);
 959   GNUNET_CRYPTO_hash (&mem_pub_key, sizeof (mem_pub_key), &mem_pub_key_hash);
 960   GNUNET_CRYPTO_hash (&msg->group_pub_key, sizeof (msg->group_pub_key), &pub_key_hash);
 961
 962   struct GNUNET_CONTAINER_MultiHashMap *
 963     grp_mem = GNUNET_CONTAINER_multihashmap_get (group_members, &pub_key_hash);
 964   struct Member *mem = NULL;
 965   struct Group *grp;
 966
 967   if (NULL != grp_mem)
 968   {
 969     mem = GNUNET_CONTAINER_multihashmap_get (grp_mem, &mem_pub_key_hash);
 970   }
 971   if (NULL == mem)
 972   {
 973     mem = GNUNET_new (struct Member);
 974     mem->priv_key = msg->member_key;
 975     mem->pub_key = mem_pub_key;
 976     mem->pub_key_hash = mem_pub_key_hash;
 977     mem->max_fragment_id = 0; // FIXME
 978
 979     grp = &mem->grp;
 980     grp->is_origin = GNUNET_NO;
 981     grp->pub_key = msg->group_pub_key;
 982     grp->pub_key_hash = pub_key_hash;
 983
 984     if (NULL == grp_mem)
 985     {
 986       grp_mem = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
 987       GNUNET_CONTAINER_multihashmap_put (group_members, &grp->pub_key_hash, grp_mem,
 988                                          GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
 989     }
 990     GNUNET_CONTAINER_multihashmap_put (grp_mem, &mem->pub_key_hash, mem,
 991                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
 992     GNUNET_CONTAINER_multihashmap_put (members, &grp->pub_key_hash, mem,
 993                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
 994   }
 995   else
 996   {
 997     grp = &mem->grp;
 998   }
 999
1000   struct ClientList *cl = GNUNET_new (struct ClientList);
1001   cl->client = client;
1002   GNUNET_CONTAINER_DLL_insert (grp->clients_head, grp->clients_tail, cl);
1003
1004   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1005               "%p Client connected to group %s..\n",
1006               mem, GNUNET_h2s (&grp->pub_key_hash));
1007   char *str = GNUNET_CRYPTO_ecdsa_public_key_to_string (&mem->pub_key);
1008   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1009               "%p ..as member %s (%s).\n",
1010               mem, GNUNET_h2s (&mem->pub_key_hash), str);
1011   GNUNET_free (str);
1012
1013   GNUNET_SERVER_client_set_user_context (client, grp);
1014
1015   if (NULL != mem->join_dcsn)
1016   { /* Already got a join decision, send it to client. */
1017     GNUNET_SERVER_notification_context_add (nc, client);
1018     GNUNET_SERVER_notification_context_unicast (nc, client,
1019                                                 (struct GNUNET_MessageHeader *)
1020                                                 mem->join_dcsn,
1021                                                 GNUNET_NO);
1022   }
1023   else
1024   { /* First client of the group, send join request. */
1025     struct GNUNET_PeerIdentity *relays = (struct GNUNET_PeerIdentity *) &msg[1];
1026     uint32_t relay_count = ntohl (msg->relay_count);
1027     uint16_t relay_size = relay_count * sizeof (*relays);
1028     struct GNUNET_MessageHeader *join_msg = NULL;
1029     uint16_t join_msg_size = 0;
1030     if (sizeof (*msg) + relay_size + sizeof (struct GNUNET_MessageHeader)
1031         <= msg_size)
1032     {
1033       join_msg = (struct GNUNET_MessageHeader *)
1034         (((char *) &msg[1]) + relay_size);
1035       join_msg_size = ntohs (join_msg->size);
1036     }
1037     if (sizeof (*msg) + relay_size + join_msg_size != msg_size)
1038     {
1039       GNUNET_break (0);
1040       GNUNET_SERVER_client_disconnect (client);
1041       return;
1042     }
1043
1044     struct MulticastJoinRequestMessage *
1045       req = GNUNET_malloc (sizeof (*req) + join_msg_size);
1046     req->header.size = htons (sizeof (*req) + join_msg_size);
1047     req->header.type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_REQUEST);
1048     req->group_pub_key = grp->pub_key;
1049     req->peer = this_peer;
1050     GNUNET_CRYPTO_ecdsa_key_get_public (&mem->priv_key, &req->member_pub_key);
1051     if (0 < join_msg_size)
1052       memcpy (&req[1], join_msg, join_msg_size);
1053
1054     req->member_pub_key = mem->pub_key;
1055     req->purpose.size = htonl (msg_size
1056                                - sizeof (req->header)
1057                                - sizeof (req->reserved)
1058                                - sizeof (req->signature));
1059     req->purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST);
1060
1061     if (GNUNET_OK != GNUNET_CRYPTO_ecdsa_sign (&mem->priv_key, &req->purpose,
1062                                                &req->signature))
1063     {
1064       /* FIXME: handle error */
1065       GNUNET_assert (0);
1066     }
1067
1068     if (NULL != mem->join_req)
1069       GNUNET_free (mem->join_req);
1070     mem->join_req = req;
1071
1072     if (0 == client_send_origin (&grp->pub_key_hash, &mem->join_req->header))
1073     { /* No local origins, send to remote origin */
1074       cadet_send_join_request (mem);
1075     }
1076   }
1077   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1078 }
1079
1080
1081 static void
1082 client_send_join_decision (struct Member *mem,
1083                            const struct MulticastJoinDecisionMessageHeader *hdcsn)
1084 {
1085   client_send_group (&mem->grp, &hdcsn->header);
1086
1087   const struct MulticastJoinDecisionMessage *
1088     dcsn = (const struct MulticastJoinDecisionMessage *) &hdcsn[1];
1089   if (GNUNET_YES == ntohl (dcsn->is_admitted))
1090   { /* Member admitted, store join_decision. */
1091     uint16_t dcsn_size = ntohs (dcsn->header.size);
1092     mem->join_dcsn = GNUNET_malloc (dcsn_size);
1093     memcpy (mem->join_dcsn, dcsn, dcsn_size);
1094   }
1095   else
1096   { /* Refused entry, but replay would be still possible for past members. */
1097   }
1098 }
1099
1100
1101 /**
1102  * Join decision from client.
1103  */
1104 static void
1105 client_recv_join_decision (void *cls, struct GNUNET_SERVER_Client *client,
1106                            const struct GNUNET_MessageHeader *m)
1107 {
1108   struct Group *
1109     grp = GNUNET_SERVER_client_get_user_context (client, struct Group);
1110   const struct MulticastJoinDecisionMessageHeader *
1111     hdcsn = (const struct MulticastJoinDecisionMessageHeader *) m;
1112
1113   if (NULL == grp)
1114   {
1115     GNUNET_break (0);
1116     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1117     return;
1118   }
1119   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1120               "%p Got join decision from client for group %s..\n",
1121               grp, GNUNET_h2s (&grp->pub_key_hash));
1122
1123   struct GNUNET_CONTAINER_MultiHashMap *
1124     grp_mem = GNUNET_CONTAINER_multihashmap_get (group_members,
1125                                                  &grp->pub_key_hash);
1126   struct Member *mem = NULL;
1127   if (NULL != grp_mem)
1128   {
1129     struct GNUNET_HashCode member_key_hash;
1130     GNUNET_CRYPTO_hash (&hdcsn->member_pub_key, sizeof (hdcsn->member_pub_key),
1131                         &member_key_hash);
1132     mem = GNUNET_CONTAINER_multihashmap_get (grp_mem, &member_key_hash);
1133     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1134                 "%p ..and member %s: %p\n",
1135                 grp, GNUNET_h2s (&member_key_hash), mem);
1136   }
1137   if (NULL != mem)
1138   { /* Found local member */
1139     client_send_join_decision (mem, hdcsn);
1140   }
1141   else
1142   { /* Look for remote member */
1143     cadet_send_join_decision (grp, hdcsn);
1144   }
1145   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1146 }
1147
1148
1149 /**
1150  * Incoming message from a client.
1151  */
1152 static void
1153 client_recv_multicast_message (void *cls, struct GNUNET_SERVER_Client *client,
1154                                const struct GNUNET_MessageHeader *m)
1155 {
1156   struct Group *
1157     grp = GNUNET_SERVER_client_get_user_context (client, struct Group);
1158   struct GNUNET_MULTICAST_MessageHeader *out;
1159   struct Origin *orig;
1160
1161   if (NULL == grp)
1162   {
1163     GNUNET_break (0);
1164     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1165     return;
1166   }
1167   GNUNET_assert (GNUNET_YES == grp->is_origin);
1168   orig = (struct Origin *) grp;
1169
1170   /* FIXME: yucky, should use separate message structs for P2P and CS! */
1171   out = (struct GNUNET_MULTICAST_MessageHeader *) GNUNET_copy_message (m);
1172   out->fragment_id = GNUNET_htonll (++orig->max_fragment_id);
1173   out->purpose.size = htonl (ntohs (out->header.size)
1174                              - sizeof (out->header)
1175                              - sizeof (out->hop_counter)
1176                              - sizeof (out->signature));
1177   out->purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_MULTICAST_MESSAGE);
1178
1179   if (GNUNET_OK != GNUNET_CRYPTO_eddsa_sign (&orig->priv_key, &out->purpose,
1180                                              &out->signature))
1181   {
1182     GNUNET_assert (0);
1183   }
1184
1185   client_send_all (&grp->pub_key_hash, &out->header);
1186   if (0 == cadet_send_children (&grp->pub_key_hash, &out->header))
1187   {
1188     client_send_ack (&grp->pub_key_hash);
1189   }
1190   GNUNET_free (out);
1191
1192   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1193 }
1194
1195
1196 /**
1197  * Incoming request from a client.
1198  */
1199 static void
1200 client_recv_multicast_request (void *cls, struct GNUNET_SERVER_Client *client,
1201                                const struct GNUNET_MessageHeader *m)
1202 {
1203   struct Group *grp = GNUNET_SERVER_client_get_user_context (client, struct Group);
1204   struct Member *mem;
1205   struct GNUNET_MULTICAST_RequestHeader *out;
1206   if (NULL == grp)
1207   {
1208     GNUNET_break (0);
1209     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1210     return;
1211   }
1212   GNUNET_assert (GNUNET_NO == grp->is_origin);
1213   mem = (struct Member *) grp;
1214
1215   /* FIXME: yucky, should use separate message structs for P2P and CS! */
1216   out = (struct GNUNET_MULTICAST_RequestHeader *) GNUNET_copy_message (m);
1217   out->member_pub_key = mem->pub_key;
1218   out->fragment_id = GNUNET_ntohll (++mem->max_fragment_id);
1219   out->purpose.size = htonl (ntohs (out->header.size)
1220                              - sizeof (out->header)
1221                              - sizeof (out->member_pub_key)
1222                              - sizeof (out->signature));
1223   out->purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST);
1224
1225   if (GNUNET_OK != GNUNET_CRYPTO_ecdsa_sign (&mem->priv_key, &out->purpose,
1226                                              &out->signature))
1227   {
1228     GNUNET_assert (0);
1229   }
1230
1231   uint8_t send_ack = GNUNET_YES;
1232   if (0 == client_send_origin (&grp->pub_key_hash, &out->header))
1233   { /* No local origins, send to remote origin */
1234     if (NULL != mem->origin_channel)
1235     {
1236       cadet_send_channel (mem->origin_channel, &out->header);
1237       send_ack = GNUNET_NO;
1238     }
1239     else
1240     {
1241       /* FIXME: not yet connected to origin */
1242       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1243       GNUNET_free (out);
1244       return;
1245     }
1246   }
1247   if (GNUNET_YES == send_ack)
1248   {
1249     client_send_ack (&grp->pub_key_hash);
1250   }
1251   GNUNET_free (out);
1252   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1253 }
1254
1255
1256 /**
1257  * Incoming replay request from a client.
1258  */
1259 static void
1260 client_recv_replay_request (void *cls, struct GNUNET_SERVER_Client *client,
1261                             const struct GNUNET_MessageHeader *m)
1262 {
1263   struct Group *grp = GNUNET_SERVER_client_get_user_context (client, struct Group);
1264   struct Member *mem;
1265   if (NULL == grp)
1266   {
1267     GNUNET_break (0);
1268     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1269     return;
1270   }
1271   GNUNET_assert (GNUNET_NO == grp->is_origin);
1272   mem = (struct Member *) grp;
1273
1274   struct GNUNET_CONTAINER_MultiHashMap *
1275     grp_replay_req = GNUNET_CONTAINER_multihashmap_get (replay_req_client,
1276                                                         &grp->pub_key_hash);
1277   if (NULL == grp_replay_req)
1278   {
1279     grp_replay_req = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
1280     GNUNET_CONTAINER_multihashmap_put (replay_req_client,
1281                                        &grp->pub_key_hash, grp_replay_req,
1282                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
1283   }
1284   struct MulticastReplayRequestMessage *
1285     rep = (struct MulticastReplayRequestMessage *) m;
1286   struct GNUNET_HashCode key_hash;
1287   replay_key_hash (rep->fragment_id, rep->message_id, rep->fragment_offset,
1288                    rep->flags, &key_hash);
1289   GNUNET_CONTAINER_multihashmap_put (grp_replay_req, &key_hash, client,
1290                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1291
1292   if (0 == client_send_origin (&grp->pub_key_hash, m))
1293   { /* No local origin, replay from remote members / origin. */
1294     if (NULL != mem->origin_channel)
1295     {
1296       cadet_send_channel (mem->origin_channel, m);
1297     }
1298     else
1299     {
1300       /* FIXME: not yet connected to origin */
1301       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1302       return;
1303     }
1304   }
1305   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1306 }
1307
1308
1309 static int
1310 cadet_send_replay_response_cb (void *cls,
1311                                const struct GNUNET_HashCode *key_hash,
1312                                void *value)
1313 {
1314   struct Channel *chn = value;
1315   struct GNUNET_MessageHeader *msg = cls;
1316
1317   cadet_send_channel (chn, msg);
1318   return GNUNET_OK;
1319 }
1320
1321
1322 static int
1323 client_send_replay_response_cb (void *cls,
1324                                 const struct GNUNET_HashCode *key_hash,
1325                                 void *value)
1326 {
1327   struct GNUNET_SERVER_Client *client = value;
1328   struct GNUNET_MessageHeader *msg = cls;
1329
1330   client_send (client, msg);
1331   return GNUNET_OK;
1332 }
1333
1334
1335 /**
1336  * End of replay response from a client.
1337  */
1338 static void
1339 client_recv_replay_response_end (void *cls, struct GNUNET_SERVER_Client *client,
1340                                  const struct GNUNET_MessageHeader *m)
1341 {
1342   struct Group *grp = GNUNET_SERVER_client_get_user_context (client, struct Group);
1343   if (NULL == grp)
1344   {
1345     GNUNET_break (0);
1346     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1347     return;
1348   }
1349
1350   struct MulticastReplayResponseMessage *
1351     res = (struct MulticastReplayResponseMessage *) m;
1352
1353   struct GNUNET_HashCode key_hash;
1354   replay_key_hash (res->fragment_id, res->message_id, res->fragment_offset,
1355                    res->flags, &key_hash);
1356
1357   struct GNUNET_CONTAINER_MultiHashMap *
1358     grp_replay_req_cadet = GNUNET_CONTAINER_multihashmap_get (replay_req_cadet,
1359                                                                 &grp->pub_key_hash);
1360   if (NULL != grp_replay_req_cadet)
1361   {
1362     GNUNET_CONTAINER_multihashmap_remove_all (grp_replay_req_cadet, &key_hash);
1363   }
1364   struct GNUNET_CONTAINER_MultiHashMap *
1365     grp_replay_req_client = GNUNET_CONTAINER_multihashmap_get (replay_req_client,
1366                                                                &grp->pub_key_hash);
1367   if (NULL != grp_replay_req_client)
1368   {
1369     GNUNET_CONTAINER_multihashmap_remove_all (grp_replay_req_client, &key_hash);
1370   }
1371   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1372 }
1373
1374
1375 /**
1376  * Incoming replay response from a client.
1377  *
1378  * Respond with a multicast message on success, or otherwise with an error code.
1379  */
1380 static void
1381 client_recv_replay_response (void *cls, struct GNUNET_SERVER_Client *client,
1382                              const struct GNUNET_MessageHeader *m)
1383 {
1384   struct Group *grp = GNUNET_SERVER_client_get_user_context (client, struct Group);
1385   if (NULL == grp)
1386   {
1387     GNUNET_break (0);
1388     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1389     return;
1390   }
1391
1392   struct MulticastReplayResponseMessage *
1393     res = (struct MulticastReplayResponseMessage *) m;
1394
1395   const struct GNUNET_MessageHeader *msg = m;
1396   if (GNUNET_MULTICAST_REC_OK == res->error_code)
1397   {
1398     msg = (struct GNUNET_MessageHeader *) &res[1];
1399   }
1400
1401   struct GNUNET_HashCode key_hash;
1402   replay_key_hash (res->fragment_id, res->message_id, res->fragment_offset,
1403                    res->flags, &key_hash);
1404
1405   struct GNUNET_CONTAINER_MultiHashMap *
1406     grp_replay_req_cadet = GNUNET_CONTAINER_multihashmap_get (replay_req_cadet,
1407                                                               &grp->pub_key_hash);
1408   if (NULL != grp_replay_req_cadet)
1409   {
1410     GNUNET_CONTAINER_multihashmap_get_multiple (grp_replay_req_cadet, &key_hash,
1411                                                 cadet_send_replay_response_cb,
1412                                                 (void *) msg);
1413   }
1414   if (GNUNET_MULTICAST_REC_OK == res->error_code)
1415   {
1416     struct GNUNET_CONTAINER_MultiHashMap *
1417       grp_replay_req_client = GNUNET_CONTAINER_multihashmap_get (replay_req_client,
1418                                                                  &grp->pub_key_hash);
1419     if (NULL != grp_replay_req_client)
1420     {
1421       GNUNET_CONTAINER_multihashmap_get_multiple (grp_replay_req_client, &key_hash,
1422                                                   client_send_replay_response_cb,
1423                                                   (void *) msg);
1424     }
1425   }
1426   else
1427   {
1428     client_recv_replay_response_end (cls, client, m);
1429     return;
1430   }
1431   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1432 }
1433
1434
1435 /**
1436  * A new client connected.
1437  */
1438 static void
1439 client_notify_connect (void *cls, struct GNUNET_SERVER_Client *client)
1440 {
1441   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Client connected: %p\n", client);
1442   /* FIXME: send connect ACK */
1443 }
1444
1445
1446 /**
1447  * Message handlers for the server.
1448  */
1449 static const struct GNUNET_SERVER_MessageHandler server_handlers[] = {
1450   { client_recv_origin_start, NULL,
1451     GNUNET_MESSAGE_TYPE_MULTICAST_ORIGIN_START, 0 },
1452
1453   { client_recv_member_join, NULL,
1454     GNUNET_MESSAGE_TYPE_MULTICAST_MEMBER_JOIN, 0 },
1455
1456   { client_recv_join_decision, NULL,
1457     GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_DECISION, 0 },
1458
1459   { client_recv_multicast_message, NULL,
1460     GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE, 0 },
1461
1462   { client_recv_multicast_request, NULL,
1463     GNUNET_MESSAGE_TYPE_MULTICAST_REQUEST, 0 },
1464
1465   { client_recv_replay_request, NULL,
1466     GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST, 0 },
1467
1468   { client_recv_replay_response, NULL,
1469     GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE, 0 },
1470
1471   { client_recv_replay_response_end, NULL,
1472     GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE_END, 0 },
1473
1474   { NULL, NULL, 0, 0 }
1475 };
1476
1477
1478 /**
1479  * New incoming CADET channel.
1480  */
1481 static void *
1482 cadet_notify_channel_new (void *cls,
1483                           struct GNUNET_CADET_Channel *channel,
1484                           const struct GNUNET_PeerIdentity *initiator,
1485                           uint32_t port,
1486                           enum GNUNET_CADET_ChannelOption options)
1487 {
1488   return NULL;
1489 }
1490
1491
1492 /**
1493  * CADET channel is being destroyed.
1494  */
1495 static void
1496 cadet_notify_channel_end (void *cls,
1497                           const struct GNUNET_CADET_Channel *channel,
1498                           void *ctx)
1499 {
1500   if (NULL == ctx)
1501     return;
1502
1503   struct Channel *chn = ctx;
1504   if (NULL != chn->grp)
1505   {
1506     if (GNUNET_NO == chn->grp->is_origin)
1507     {
1508       struct Member *mem = (struct Member *) chn->grp;
1509       if (chn == mem->origin_channel)
1510         mem->origin_channel = NULL;
1511     }
1512   }
1513
1514   while (GNUNET_YES == replay_req_remove_cadet (chn));
1515
1516   GNUNET_free (chn);
1517 }
1518
1519
1520 /**
1521  * Incoming join request message from CADET.
1522  */
1523 int
1524 cadet_recv_join_request (void *cls,
1525                          struct GNUNET_CADET_Channel *channel,
1526                          void **ctx,
1527                          const struct GNUNET_MessageHeader *m)
1528 {
1529   const struct MulticastJoinRequestMessage *
1530     req = (const struct MulticastJoinRequestMessage *) m;
1531   uint16_t size = ntohs (m->size);
1532   if (size < sizeof (*req))
1533   {
1534     GNUNET_break_op (0);
1535     return GNUNET_SYSERR;
1536   }
1537   if (NULL != *ctx)
1538   {
1539     GNUNET_break_op (0);
1540     return GNUNET_SYSERR;
1541   }
1542   if (ntohl (req->purpose.size) != (size
1543                                     - sizeof (req->header)
1544                                     - sizeof (req->reserved)
1545                                     - sizeof (req->signature)))
1546   {
1547     GNUNET_break_op (0);
1548     return GNUNET_SYSERR;
1549   }
1550   if (GNUNET_OK !=
1551       GNUNET_CRYPTO_ecdsa_verify (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST,
1552                                   &req->purpose, &req->signature,
1553                                   &req->member_pub_key))
1554   {
1555     GNUNET_break_op (0);
1556     return GNUNET_SYSERR;
1557   }
1558
1559   struct GNUNET_HashCode group_pub_hash;
1560   GNUNET_CRYPTO_hash (&req->group_pub_key, sizeof (req->group_pub_key), &group_pub_hash);
1561
1562   struct Channel *chn = GNUNET_malloc (sizeof *chn);
1563   chn->channel = channel;
1564   chn->group_pub_key = req->group_pub_key;
1565   chn->group_pub_hash = group_pub_hash;
1566   chn->member_pub_key = req->member_pub_key;
1567   chn->peer = req->peer;
1568   chn->join_status = JOIN_WAITING;
1569   GNUNET_CONTAINER_multihashmap_put (channels_in, &chn->group_pub_hash, chn,
1570                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1571
1572   client_send_all (&group_pub_hash, m);
1573   return GNUNET_OK;
1574 }
1575
1576
1577 /**
1578  * Incoming join decision message from CADET.
1579  */
1580 int
1581 cadet_recv_join_decision (void *cls,
1582                           struct GNUNET_CADET_Channel *channel,
1583                           void **ctx,
1584                           const struct GNUNET_MessageHeader *m)
1585 {
1586   const struct MulticastJoinDecisionMessage *
1587     dcsn = (const struct MulticastJoinDecisionMessage *) m;
1588   uint16_t size = ntohs (m->size);
1589   if (size < sizeof (*dcsn))
1590   {
1591     GNUNET_break_op (0);
1592     return GNUNET_SYSERR;
1593   }
1594   struct Channel *chn = *ctx;
1595   if (NULL == chn)
1596   {
1597     GNUNET_break_op (0);
1598     return GNUNET_SYSERR;
1599   }
1600   if (NULL == chn->grp || GNUNET_NO != chn->grp->is_origin)
1601   {
1602     GNUNET_break_op (0);
1603     return GNUNET_SYSERR;
1604   }
1605   switch (chn->join_status)
1606   {
1607   case JOIN_REFUSED:
1608     return GNUNET_SYSERR;
1609
1610   case JOIN_ADMITTED:
1611     return GNUNET_OK;
1612
1613   case JOIN_NOT_ASKED:
1614   case JOIN_WAITING:
1615     break;
1616   }
1617
1618   struct MulticastJoinDecisionMessageHeader *
1619     hdcsn = GNUNET_malloc (sizeof (*hdcsn) + size);
1620   hdcsn->peer = chn->peer;
1621   memcpy (&hdcsn[1], dcsn, sizeof (*hdcsn) + size);
1622
1623   struct Member *mem = (struct Member *) chn->grp;
1624   client_send_join_decision (mem, hdcsn);
1625   GNUNET_free (hdcsn);
1626   if (GNUNET_YES == ntohs (dcsn->is_admitted))
1627   {
1628     chn->join_status = JOIN_ADMITTED;
1629     return GNUNET_OK;
1630   }
1631   else
1632   {
1633     chn->join_status = JOIN_REFUSED;
1634     return GNUNET_SYSERR;
1635   }
1636 }
1637
1638 /**
1639  * Incoming multicast message from CADET.
1640  */
1641 int
1642 cadet_recv_message (void *cls,
1643                     struct GNUNET_CADET_Channel *channel,
1644                     void **ctx,
1645                     const struct GNUNET_MessageHeader *m)
1646 {
1647   const struct GNUNET_MULTICAST_MessageHeader *
1648     msg = (const struct GNUNET_MULTICAST_MessageHeader *) m;
1649   uint16_t size = ntohs (m->size);
1650   if (size < sizeof (*msg))
1651   {
1652     GNUNET_break_op (0);
1653     return GNUNET_SYSERR;
1654   }
1655   struct Channel *chn = *ctx;
1656   if (NULL == chn)
1657   {
1658     GNUNET_break_op (0);
1659     return GNUNET_SYSERR;
1660   }
1661   if (ntohl (msg->purpose.size) != (size
1662                                     - sizeof (msg->header)
1663                                     - sizeof (msg->hop_counter)
1664                                     - sizeof (msg->signature)))
1665   {
1666     GNUNET_break_op (0);
1667     return GNUNET_SYSERR;
1668   }
1669   if (GNUNET_OK !=
1670       GNUNET_CRYPTO_eddsa_verify (GNUNET_SIGNATURE_PURPOSE_MULTICAST_MESSAGE,
1671                                   &msg->purpose, &msg->signature,
1672                                   &chn->group_pub_key))
1673   {
1674     GNUNET_break_op (0);
1675     return GNUNET_SYSERR;
1676   }
1677
1678   client_send_all (&chn->group_pub_hash, m);
1679   return GNUNET_OK;
1680 }
1681
1682
1683 /**
1684  * Incoming multicast request message from CADET.
1685  */
1686 int
1687 cadet_recv_request (void *cls,
1688                     struct GNUNET_CADET_Channel *channel,
1689                     void **ctx,
1690                     const struct GNUNET_MessageHeader *m)
1691 {
1692   const struct GNUNET_MULTICAST_RequestHeader *
1693     req = (const struct GNUNET_MULTICAST_RequestHeader *) m;
1694   uint16_t size = ntohs (m->size);
1695   if (size < sizeof (*req))
1696   {
1697     GNUNET_break_op (0);
1698     return GNUNET_SYSERR;
1699   }
1700   struct Channel *chn = *ctx;
1701   if (NULL == chn)
1702   {
1703     GNUNET_break_op (0);
1704     return GNUNET_SYSERR;
1705   }
1706   if (ntohl (req->purpose.size) != (size
1707                                     - sizeof (req->header)
1708                                     - sizeof (req->member_pub_key)
1709                                     - sizeof (req->signature)))
1710   {
1711     GNUNET_break_op (0);
1712     return GNUNET_SYSERR;
1713   }
1714   if (GNUNET_OK !=
1715       GNUNET_CRYPTO_ecdsa_verify (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST,
1716                                   &req->purpose, &req->signature,
1717                                   &req->member_pub_key))
1718   {
1719     GNUNET_break_op (0);
1720     return GNUNET_SYSERR;
1721   }
1722
1723   client_send_origin (&chn->group_pub_hash, m);
1724   return GNUNET_OK;
1725 }
1726
1727
1728 /**
1729  * Incoming multicast replay request from CADET.
1730  */
1731 int
1732 cadet_recv_replay_request (void *cls,
1733                            struct GNUNET_CADET_Channel *channel,
1734                            void **ctx,
1735                            const struct GNUNET_MessageHeader *m)
1736 {
1737   struct MulticastReplayRequestMessage rep;
1738   uint16_t size = ntohs (m->size);
1739   if (size < sizeof (rep))
1740   {
1741     GNUNET_break_op (0);
1742     return GNUNET_SYSERR;
1743   }
1744   struct Channel *chn = *ctx;
1745
1746   memcpy (&rep, m, sizeof (rep));
1747   memcpy (&rep.member_pub_key, &chn->member_pub_key, sizeof (chn->member_pub_key));
1748
1749   struct GNUNET_CONTAINER_MultiHashMap *
1750     grp_replay_req = GNUNET_CONTAINER_multihashmap_get (replay_req_cadet,
1751                                                         &chn->grp->pub_key_hash);
1752   if (NULL == grp_replay_req)
1753   {
1754     grp_replay_req = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
1755     GNUNET_CONTAINER_multihashmap_put (replay_req_cadet,
1756                                        &chn->grp->pub_key_hash, grp_replay_req,
1757                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
1758   }
1759   struct GNUNET_HashCode key_hash;
1760   replay_key_hash (rep.fragment_id, rep.message_id, rep.fragment_offset,
1761                    rep.flags, &key_hash);
1762   GNUNET_CONTAINER_multihashmap_put (grp_replay_req, &key_hash, chn,
1763                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1764
1765   client_send_random (&chn->group_pub_hash, &rep.header);
1766   return GNUNET_OK;
1767 }
1768
1769
1770 /**
1771  * Incoming multicast replay response from CADET.
1772  */
1773 int
1774 cadet_recv_replay_response (void *cls,
1775                             struct GNUNET_CADET_Channel *channel,
1776                             void **ctx,
1777                             const struct GNUNET_MessageHeader *m)
1778 {
1779   struct Channel *chn = *ctx;
1780
1781   /* @todo FIXME: got replay error response, send request to other members */
1782
1783   return GNUNET_OK;
1784 }
1785
1786
1787 /**
1788  * Message handlers for CADET.
1789  */
1790 static const struct GNUNET_CADET_MessageHandler cadet_handlers[] = {
1791   { cadet_recv_join_request,
1792     GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_REQUEST, 0 },
1793
1794   { cadet_recv_message,
1795     GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE, 0 },
1796
1797   { cadet_recv_request,
1798     GNUNET_MESSAGE_TYPE_MULTICAST_REQUEST, 0 },
1799
1800   { cadet_recv_replay_request,
1801     GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST, 0 },
1802
1803   { cadet_recv_replay_response,
1804     GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE, 0 },
1805
1806   { NULL, 0, 0 }
1807 };
1808
1809
1810 /**
1811  * Listening ports for CADET.
1812  */
1813 static const uint32_t cadet_ports[] = { GNUNET_APPLICATION_TYPE_MULTICAST, 0 };
1814
1815
1816 /**
1817  * Connected to core service.
1818  */
1819 static void
1820 core_connected_cb  (void *cls, const struct GNUNET_PeerIdentity *my_identity)
1821 {
1822   this_peer = *my_identity;
1823
1824   stats = GNUNET_STATISTICS_create ("multicast", cfg);
1825   origins = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
1826   members = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
1827   group_members = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
1828   channels_in = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
1829   channels_out = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
1830   replay_req_cadet = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
1831   replay_req_client = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
1832
1833   cadet = GNUNET_CADET_connect (cfg, NULL,
1834                                 &cadet_notify_channel_new,
1835                                 &cadet_notify_channel_end,
1836                                 cadet_handlers, cadet_ports);
1837
1838   nc = GNUNET_SERVER_notification_context_create (server, 1);
1839   GNUNET_SERVER_add_handlers (server, server_handlers);
1840   GNUNET_SERVER_disconnect_notify (server,
1841                                    &client_notify_disconnect, NULL);
1842   GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
1843                                  NULL);
1844 }
1845
1846
1847 /**
1848  * Service started.
1849  *
1850  * @param cls closure
1851  * @param server the initialized server
1852  * @param cfg configuration to use
1853  */
1854 static void
1855 run (void *cls,
1856      struct GNUNET_SERVER_Handle *srv,
1857      const struct GNUNET_CONFIGURATION_Handle *c)
1858 {
1859   cfg = c;
1860   server = srv;
1861   GNUNET_SERVER_connect_notify (server, &client_notify_connect, NULL);
1862   core = GNUNET_CORE_connect (cfg, NULL, &core_connected_cb, NULL, NULL,
1863                               NULL, GNUNET_NO, NULL, GNUNET_NO, NULL);
1864 }
1865
1866
1867 /**
1868  * The main function for the multicast service.
1869  *
1870  * @param argc number of arguments from the command line
1871  * @param argv command line arguments
1872  * @return 0 ok, 1 on error
1873  */
1874 int
1875 main (int argc, char *const *argv)
1876 {
1877   return (GNUNET_OK ==
1878           GNUNET_SERVICE_run (argc, argv, "multicast",
1879                               GNUNET_SERVICE_OPTION_NONE, &run, NULL)) ? 0 : 1;
1880 }
1881
1882 /* end of gnunet-service-multicast.c */