+ return GNUNET_YES;
+}
+
+
+/**
+ * Send join decision to a remote peer.
+ */
+static void
+cadet_send_join_decision (struct Group *grp,
+ const struct MulticastJoinDecisionMessageHeader *hdcsn)
+{
+ GNUNET_CONTAINER_multihashmap_get_multiple (channels_in, &grp->pub_key_hash,
+ &cadet_send_join_decision_cb,
+ (void *) hdcsn);
+}
+
+
+/**
+ * Iterator callback for sending a message to origin clients.
+ */
+static int
+cadet_send_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash,
+ void *channel)
+{
+ const struct GNUNET_MessageHeader *msg = cls;
+ struct Channel *chn = channel;
+ if (JOIN_ADMITTED == chn->join_status)
+ cadet_send_channel (chn, msg);
+ return GNUNET_YES;
+}
+
+
+/**
+ * Send message to all connected children.
+ */
+static int
+cadet_send_children (struct GNUNET_HashCode *pub_key_hash,
+ const struct GNUNET_MessageHeader *msg)
+{
+ int n = 0;
+ if (channels_in != NULL)
+ n += GNUNET_CONTAINER_multihashmap_get_multiple (channels_in, pub_key_hash,
+ cadet_send_cb, (void *) msg);
+ return n;
+}
+
+
+#if 0 // unused as yet
+/**
+ * Send message to all connected parents.
+ */
+static int
+cadet_send_parents (struct GNUNET_HashCode *pub_key_hash,
+ const struct GNUNET_MessageHeader *msg)
+{
+ int n = 0;
+ if (channels_in != NULL)
+ n += GNUNET_CONTAINER_multihashmap_get_multiple (channels_out, pub_key_hash,
+ cadet_send_cb, (void *) msg);
+ return n;
+}
+#endif
+
+
+/**
+ * CADET channel connect handler.
+ *
+ * @see GNUNET_CADET_ConnectEventHandler()
+ */
+static void *
+cadet_notify_connect (void *cls,
+ struct GNUNET_CADET_Channel *channel,
+ const struct GNUNET_PeerIdentity *source)
+{
+ struct Channel *chn = GNUNET_malloc (sizeof *chn);
+ chn->group = cls;
+ chn->channel = channel;
+ chn->direction = DIR_INCOMING;
+ chn->join_status = JOIN_NOT_ASKED;
+
+ GNUNET_CONTAINER_multihashmap_put (channels_in, &chn->group_pub_hash, chn,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+ return chn;
+}
+
+
+/**
+ * CADET window size change handler.
+ *
+ * @see GNUNET_CADET_WindowSizeEventHandler()
+ */
+static void
+cadet_notify_window_change (void *cls,
+ const struct GNUNET_CADET_Channel *channel,
+ int window_size)
+{
+ struct Channel *chn = cls;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "%p Window size changed to %d. Pending messages: %u\n",
+ chn, window_size, chn->msgs_pending);
+
+ chn->is_connected = GNUNET_YES;
+ chn->window_size = (int32_t) window_size;
+
+ for (int i = 0; i < window_size; i++)
+ {
+ if (0 < chn->msgs_pending)
+ {
+ client_send_ack (&chn->group_pub_hash);
+ chn->msgs_pending--;
+ }
+ else
+ {
+ break;
+ }
+ }
+}
+
+
+/**
+ * CADET channel disconnect handler.
+ *
+ * @see GNUNET_CADET_DisconnectEventHandler()
+ */
+static void
+cadet_notify_disconnect (void *cls,
+ const struct GNUNET_CADET_Channel *channel)
+{
+ if (NULL == cls)
+ return;
+
+ struct Channel *chn = cls;
+ if (NULL != chn->group)
+ {
+ if (GNUNET_NO == chn->group->is_origin)
+ {
+ struct Member *mem = (struct Member *) chn->group;
+ if (chn == mem->origin_channel)
+ mem->origin_channel = NULL;
+ }
+ }
+
+ int ret;
+ do
+ {
+ ret = replay_req_remove_cadet (chn);
+ }
+ while (GNUNET_YES == ret);
+
+ GNUNET_free (chn);
+}
+
+
+static int
+check_cadet_join_request (void *cls,
+ const struct MulticastJoinRequestMessage *req)
+{
+ struct Channel *chn = cls;
+
+ if (NULL == chn
+ || JOIN_NOT_ASKED != chn->join_status)
+ {
+ return GNUNET_SYSERR;
+ }
+
+ uint16_t size = ntohs (req->header.size);
+ if (size < sizeof (*req))
+ {
+ GNUNET_break_op (0);
+ return GNUNET_SYSERR;
+ }
+ if (ntohl (req->purpose.size) != (size
+ - sizeof (req->header)
+ - sizeof (req->reserved)
+ - sizeof (req->signature)))
+ {
+ GNUNET_break_op (0);
+ return GNUNET_SYSERR;
+ }
+ if (GNUNET_OK !=
+ GNUNET_CRYPTO_ecdsa_verify (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST,
+ &req->purpose, &req->signature,
+ &req->member_pub_key))
+ {
+ GNUNET_break_op (0);
+ return GNUNET_SYSERR;
+ }
+
+ return GNUNET_OK;
+}
+
+
+/**
+ * Incoming join request message from CADET.
+ */
+static void
+handle_cadet_join_request (void *cls,
+ const struct MulticastJoinRequestMessage *req)
+{
+ struct Channel *chn = cls;
+ GNUNET_CADET_receive_done (chn->channel);
+
+ struct GNUNET_HashCode group_pub_hash;
+ GNUNET_CRYPTO_hash (&req->group_pub_key, sizeof (req->group_pub_key), &group_pub_hash);
+ chn->group_pub_key = req->group_pub_key;
+ chn->group_pub_hash = group_pub_hash;
+ chn->member_pub_key = req->member_pub_key;
+ chn->peer = req->peer;
+ chn->join_status = JOIN_WAITING;
+
+ client_send_all (&group_pub_hash, &req->header);
+}
+
+
+static int
+check_cadet_join_decision (void *cls,
+ const struct MulticastJoinDecisionMessageHeader *hdcsn)
+{
+ uint16_t size = ntohs (hdcsn->header.size);
+ if (size < sizeof (struct MulticastJoinDecisionMessageHeader) +
+ sizeof (struct MulticastJoinDecisionMessage))
+ {
+ GNUNET_break_op (0);
+ return GNUNET_SYSERR;
+ }
+
+ struct Channel *chn = cls;
+ if (NULL == chn)
+ {
+ GNUNET_break (0);
+ return GNUNET_SYSERR;
+ }
+ if (NULL == chn->group || GNUNET_NO != chn->group->is_origin)
+ {
+ GNUNET_break (0);
+ return GNUNET_SYSERR;
+ }
+ switch (chn->join_status)
+ {
+ case JOIN_REFUSED:
+ return GNUNET_SYSERR;
+
+ case JOIN_ADMITTED:
+ return GNUNET_OK;
+
+ case JOIN_NOT_ASKED:
+ case JOIN_WAITING:
+ break;
+ }
+
+ return GNUNET_OK;
+}
+
+
+/**
+ * Incoming join decision message from CADET.
+ */
+static void
+handle_cadet_join_decision (void *cls,
+ const struct MulticastJoinDecisionMessageHeader *hdcsn)
+{
+ const struct MulticastJoinDecisionMessage *
+ dcsn = (const struct MulticastJoinDecisionMessage *) &hdcsn[1];
+
+ struct Channel *chn = cls;
+ GNUNET_CADET_receive_done (chn->channel);
+
+ // FIXME: do we need to copy chn->peer or compare it with hdcsn->peer?
+ struct Member *mem = (struct Member *) chn->group;
+ client_send_join_decision (mem, hdcsn);
+ if (GNUNET_YES == ntohl (dcsn->is_admitted))
+ {
+ chn->join_status = JOIN_ADMITTED;
+ }
+ else
+ {
+ chn->join_status = JOIN_REFUSED;
+ cadet_channel_destroy (chn);
+ }
+}
+
+
+static int
+check_cadet_message (void *cls,
+ const struct GNUNET_MULTICAST_MessageHeader *msg)
+{
+ uint16_t size = ntohs (msg->header.size);
+ if (size < sizeof (*msg))
+ {
+ GNUNET_break_op (0);
+ return GNUNET_SYSERR;
+ }
+
+ struct Channel *chn = cls;
+ if (NULL == chn)
+ {
+ GNUNET_break (0);
+ return GNUNET_SYSERR;
+ }
+ if (ntohl (msg->purpose.size) != (size
+ - sizeof (msg->header)
+ - sizeof (msg->hop_counter)
+ - sizeof (msg->signature)))
+ {
+ GNUNET_break_op (0);
+ return GNUNET_SYSERR;
+ }
+ if (GNUNET_OK !=
+ GNUNET_CRYPTO_eddsa_verify (GNUNET_SIGNATURE_PURPOSE_MULTICAST_MESSAGE,
+ &msg->purpose, &msg->signature,
+ &chn->group_pub_key))
+ {
+ GNUNET_break_op (0);
+ return GNUNET_SYSERR;
+ }
+
+ return GNUNET_OK;
+}
+
+
+/**
+ * Incoming multicast message from CADET.
+ */
+static void
+handle_cadet_message (void *cls,
+ const struct GNUNET_MULTICAST_MessageHeader *msg)
+{
+ struct Channel *chn = cls;
+ GNUNET_CADET_receive_done (chn->channel);
+ client_send_all (&chn->group_pub_hash, &msg->header);
+}
+
+
+static int
+check_cadet_request (void *cls,
+ const struct GNUNET_MULTICAST_RequestHeader *req)
+{
+ uint16_t size = ntohs (req->header.size);
+ if (size < sizeof (*req))
+ {
+ GNUNET_break_op (0);
+ return GNUNET_SYSERR;
+ }
+
+ struct Channel *chn = cls;
+ if (NULL == chn)
+ {
+ GNUNET_break (0);
+ return GNUNET_SYSERR;
+ }
+ if (ntohl (req->purpose.size) != (size
+ - sizeof (req->header)
+ - sizeof (req->member_pub_key)
+ - sizeof (req->signature)))
+ {
+ GNUNET_break_op (0);
+ return GNUNET_SYSERR;
+ }
+ if (GNUNET_OK !=
+ GNUNET_CRYPTO_ecdsa_verify (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST,
+ &req->purpose, &req->signature,
+ &req->member_pub_key))
+ {
+ GNUNET_break_op (0);
+ return GNUNET_SYSERR;
+ }
+
+ return GNUNET_OK;
+}
+
+
+/**
+ * Incoming multicast request message from CADET.
+ */
+static void
+handle_cadet_request (void *cls,
+ const struct GNUNET_MULTICAST_RequestHeader *req)
+{
+ struct Channel *chn = cls;
+ GNUNET_CADET_receive_done (chn->channel);
+ client_send_origin (&chn->group_pub_hash, &req->header);
+}
+
+
+static int
+check_cadet_replay_request (void *cls,
+ const struct MulticastReplayRequestMessage *req)
+{
+ uint16_t size = ntohs (req->header.size);
+ if (size < sizeof (*req))
+ {
+ GNUNET_break_op (0);
+ return GNUNET_SYSERR;
+ }
+
+ struct Channel *chn = cls;
+ if (NULL == chn)
+ {
+ GNUNET_break_op (0);
+ return GNUNET_SYSERR;
+ }
+
+ return GNUNET_OK;
+}
+
+
+/**
+ * Incoming multicast replay request from CADET.
+ */
+static void
+handle_cadet_replay_request (void *cls,
+ const struct MulticastReplayRequestMessage *req)
+{
+ struct Channel *chn = cls;
+ GNUNET_CADET_receive_done (chn->channel);
+
+ struct MulticastReplayRequestMessage rep = *req;
+ GNUNET_memcpy (&rep.member_pub_key, &chn->member_pub_key, sizeof (chn->member_pub_key));
+
+ struct GNUNET_CONTAINER_MultiHashMap *
+ grp_replay_req = GNUNET_CONTAINER_multihashmap_get (replay_req_cadet,
+ &chn->group->pub_key_hash);
+ if (NULL == grp_replay_req)
+ {
+ grp_replay_req = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
+ GNUNET_CONTAINER_multihashmap_put (replay_req_cadet,
+ &chn->group->pub_key_hash, grp_replay_req,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
+ }
+ struct GNUNET_HashCode key_hash;
+ replay_key_hash (rep.fragment_id, rep.message_id, rep.fragment_offset,
+ rep.flags, &key_hash);
+ GNUNET_CONTAINER_multihashmap_put (grp_replay_req, &key_hash, chn,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+
+ client_send_random (&chn->group_pub_hash, &rep.header);
+}
+
+
+static int
+check_cadet_replay_response (void *cls,
+ const struct MulticastReplayResponseMessage *res)
+{
+ struct Channel *chn = cls;
+ if (NULL == chn)
+ {
+ GNUNET_break (0);
+ return GNUNET_SYSERR;
+ }
+ return GNUNET_OK;
+}
+
+
+/**
+ * Incoming multicast replay response from CADET.
+ */
+static void
+handle_cadet_replay_response (void *cls,
+ const struct MulticastReplayResponseMessage *res)
+{
+ struct Channel *chn = cls;
+ GNUNET_CADET_receive_done (chn->channel);
+
+ /* @todo FIXME: got replay error response, send request to other members */
+}
+
+
+static void
+group_set_cadet_port_hash (struct Group *grp)
+{
+ struct CadetPort {
+ struct GNUNET_CRYPTO_EddsaPublicKey pub_key;
+ uint32_t app_type;
+ } port = {
+ grp->pub_key,
+ GNUNET_APPLICATION_TYPE_MULTICAST,
+ };
+ GNUNET_CRYPTO_hash (&port, sizeof (port), &grp->cadet_port_hash);
+}
+
+
+
+/**
+ * Create new outgoing CADET channel.
+ *
+ * @param peer
+ * Peer to connect to.
+ * @param group_pub_key
+ * Public key of group the channel belongs to.
+ * @param group_pub_hash
+ * Hash of @a group_pub_key.
+ *
+ * @return Channel.
+ */
+static struct Channel *
+cadet_channel_create (struct Group *grp, struct GNUNET_PeerIdentity *peer)
+{
+ struct Channel *chn = GNUNET_malloc (sizeof (*chn));
+ chn->group = grp;
+ chn->group_pub_key = grp->pub_key;
+ chn->group_pub_hash = grp->pub_key_hash;
+ chn->peer = *peer;
+ chn->direction = DIR_OUTGOING;
+ chn->is_connected = GNUNET_NO;
+ chn->join_status = JOIN_WAITING;
+
+ struct GNUNET_MQ_MessageHandler cadet_handlers[] = {
+ GNUNET_MQ_hd_var_size (cadet_message,
+ GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE,
+ struct GNUNET_MULTICAST_MessageHeader,
+ chn),
+
+ GNUNET_MQ_hd_var_size (cadet_join_decision,
+ GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_DECISION,
+ struct MulticastJoinDecisionMessageHeader,
+ chn),
+
+ GNUNET_MQ_hd_var_size (cadet_replay_request,
+ GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST,
+ struct MulticastReplayRequestMessage,
+ chn),
+
+ GNUNET_MQ_hd_var_size (cadet_replay_response,
+ GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE,
+ struct MulticastReplayResponseMessage,
+ chn),
+
+ GNUNET_MQ_handler_end ()
+ };
+
+ chn->channel = GNUNET_CADET_channel_creatE (cadet, chn, &chn->peer,
+ &grp->cadet_port_hash,
+ GNUNET_CADET_OPTION_RELIABLE,
+ cadet_notify_window_change,
+ cadet_notify_disconnect,
+ cadet_handlers);
+ GNUNET_CONTAINER_multihashmap_put (channels_out, &chn->group_pub_hash, chn,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+ return chn;
+}
+
+
+/**
+ * Destroy outgoing CADET channel.
+ */
+static void
+cadet_channel_destroy (struct Channel *chn)
+{
+ GNUNET_CADET_channel_destroy (chn->channel);
+ GNUNET_CONTAINER_multihashmap_remove_all (channels_out, &chn->group_pub_hash);
+ GNUNET_free (chn);
+}
+
+/**
+ * Handle a connecting client starting an origin.
+ */
+static void
+handle_client_origin_start (void *cls,
+ const struct MulticastOriginStartMessage *msg)
+{
+ struct Client *c = cls;
+ struct GNUNET_SERVICE_Client *client = c->client;
+
+ struct GNUNET_CRYPTO_EddsaPublicKey pub_key;
+ struct GNUNET_HashCode pub_key_hash;
+
+ GNUNET_CRYPTO_eddsa_key_get_public (&msg->group_key, &pub_key);
+ GNUNET_CRYPTO_hash (&pub_key, sizeof (pub_key), &pub_key_hash);
+
+ struct Origin *
+ orig = GNUNET_CONTAINER_multihashmap_get (origins, &pub_key_hash);
+ struct Group *grp;
+
+ if (NULL == orig)
+ {
+ orig = GNUNET_new (struct Origin);
+ orig->priv_key = msg->group_key;
+ orig->max_fragment_id = GNUNET_ntohll (msg->max_fragment_id);
+
+ grp = c->group = &orig->group;
+ grp->origin = orig;
+ grp->is_origin = GNUNET_YES;
+ grp->pub_key = pub_key;
+ grp->pub_key_hash = pub_key_hash;
+
+ GNUNET_CONTAINER_multihashmap_put (origins, &grp->pub_key_hash, orig,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
+
+ group_set_cadet_port_hash (grp);
+
+ struct GNUNET_MQ_MessageHandler cadet_handlers[] = {
+ GNUNET_MQ_hd_var_size (cadet_message,
+ GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE,
+ struct GNUNET_MULTICAST_MessageHeader,
+ grp),
+
+ GNUNET_MQ_hd_var_size (cadet_request,
+ GNUNET_MESSAGE_TYPE_MULTICAST_REQUEST,
+ struct GNUNET_MULTICAST_RequestHeader,
+ grp),
+
+ GNUNET_MQ_hd_var_size (cadet_join_request,
+ GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_REQUEST,
+ struct MulticastJoinRequestMessage,
+ grp),
+
+ GNUNET_MQ_hd_var_size (cadet_replay_request,
+ GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST,
+ struct MulticastReplayRequestMessage,
+ grp),
+
+ GNUNET_MQ_hd_var_size (cadet_replay_response,
+ GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE,
+ struct MulticastReplayResponseMessage,
+ grp),
+
+ GNUNET_MQ_handler_end ()
+ };
+
+
+ orig->cadet_port = GNUNET_CADET_open_porT (cadet,
+ &grp->cadet_port_hash,
+ cadet_notify_connect,
+ NULL,
+ cadet_notify_window_change,
+ cadet_notify_disconnect,
+ cadet_handlers);
+ }
+ else
+ {
+ grp = &orig->group;
+ }
+
+ struct ClientList *cl = GNUNET_new (struct ClientList);
+ cl->client = client;
+ GNUNET_CONTAINER_DLL_insert (grp->clients_head, grp->clients_tail, cl);
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p Client connected as origin to group %s.\n",
+ orig, GNUNET_h2s (&grp->pub_key_hash));
+ GNUNET_SERVICE_client_continue (client);
+}
+
+
+static int
+check_client_member_join (void *cls,
+ const struct MulticastMemberJoinMessage *msg)
+{
+ uint16_t msg_size = ntohs (msg->header.size);
+ struct GNUNET_PeerIdentity *relays = (struct GNUNET_PeerIdentity *) &msg[1];
+ uint32_t relay_count = ntohl (msg->relay_count);
+ uint16_t relay_size = relay_count * sizeof (*relays);
+ struct GNUNET_MessageHeader *join_msg = NULL;
+ uint16_t join_msg_size = 0;
+ if (sizeof (*msg) + relay_size + sizeof (struct GNUNET_MessageHeader)
+ <= msg_size)
+ {
+ join_msg = (struct GNUNET_MessageHeader *)
+ (((char *) &msg[1]) + relay_size);
+ join_msg_size = ntohs (join_msg->size);
+ }
+ return
+ msg_size == (sizeof (*msg) + relay_size + join_msg_size)
+ ? GNUNET_OK
+ : GNUNET_SYSERR;
+}
+
+
+/**
+ * Handle a connecting client joining a group.
+ */
+static void
+handle_client_member_join (void *cls,
+ const struct MulticastMemberJoinMessage *msg)
+{
+ struct Client *c = cls;
+ struct GNUNET_SERVICE_Client *client = c->client;
+
+ uint16_t msg_size = ntohs (msg->header.size);
+
+ struct GNUNET_CRYPTO_EcdsaPublicKey mem_pub_key;
+ struct GNUNET_HashCode pub_key_hash, mem_pub_key_hash;
+
+ GNUNET_CRYPTO_ecdsa_key_get_public (&msg->member_key, &mem_pub_key);
+ GNUNET_CRYPTO_hash (&mem_pub_key, sizeof (mem_pub_key), &mem_pub_key_hash);
+ GNUNET_CRYPTO_hash (&msg->group_pub_key, sizeof (msg->group_pub_key), &pub_key_hash);
+
+ struct GNUNET_CONTAINER_MultiHashMap *
+ grp_mem = GNUNET_CONTAINER_multihashmap_get (group_members, &pub_key_hash);
+ struct Member *mem = NULL;
+ struct Group *grp;
+
+ if (NULL != grp_mem)
+ {
+ mem = GNUNET_CONTAINER_multihashmap_get (grp_mem, &mem_pub_key_hash);
+ }
+ if (NULL == mem)
+ {
+ mem = GNUNET_new (struct Member);
+ mem->origin = msg->origin;
+ mem->priv_key = msg->member_key;
+ mem->pub_key = mem_pub_key;