src/multicast/gnunet-service-multicast.c

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