/* This file is part of GNUnet. Copyright (C) 2013, 2014, 2016 GNUnet e.V. GNUnet is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. GNUnet is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. */ /** * @file scalarproduct/gnunet-service-scalarproduct_bob.c * @brief scalarproduct service implementation * @author Christian M. Fuchs * @author Christian Grothoff */ #include "platform.h" #include #include #include "gnunet_util_lib.h" #include "gnunet_core_service.h" #include "gnunet_cadet_service.h" #include "gnunet_applications.h" #include "gnunet_protocols.h" #include "gnunet_scalarproduct_service.h" #include "gnunet_set_service.h" #include "scalarproduct.h" #include "gnunet-service-scalarproduct.h" #define LOG(kind,...) GNUNET_log_from (kind, "scalarproduct-bob", __VA_ARGS__) /** * An encrypted element key-value pair. */ struct MpiElement { /** * Key used to identify matching pairs of values to multiply. * Points into an existing data structure, to avoid copying * and doubling memory use. */ const struct GNUNET_HashCode *key; /** * Value represented (a). */ gcry_mpi_t value; }; /** * A scalarproduct session which tracks an offer for a * multiplication service by a local client. */ struct BobServiceSession { /** * (hopefully) unique transaction ID */ struct GNUNET_HashCode session_id; /** * The client this request is related to. */ struct GNUNET_SERVICE_Client *client; /** * Client message queue. */ struct GNUNET_MQ_Handle *client_mq; /** * All non-0-value'd elements transmitted to us. */ struct GNUNET_CONTAINER_MultiHashMap *intersected_elements; /** * Set of elements for which we will be conducting an intersection. * The resulting elements are then used for computing the scalar product. */ struct GNUNET_SET_Handle *intersection_set; /** * Set of elements for which will conduction an intersection. * the resulting elements are then used for computing the scalar product. */ struct GNUNET_SET_OperationHandle *intersection_op; /** * CADET port we are listening on. */ struct GNUNET_CADET_Port *port; /** * a(Alice) */ struct MpiElement *sorted_elements; /** * E(ai)(Bob) after applying the mask */ struct GNUNET_CRYPTO_PaillierCiphertext *e_a; /** * Bob's permutation p of R */ struct GNUNET_CRYPTO_PaillierCiphertext *r; /** * Bob's permutation q of R */ struct GNUNET_CRYPTO_PaillierCiphertext *r_prime; /** * Bob's "s" */ struct GNUNET_CRYPTO_PaillierCiphertext s; /** * Bob's "s'" */ struct GNUNET_CRYPTO_PaillierCiphertext s_prime; /** * Handle for our associated incoming CADET session, or NULL * if we have not gotten one yet. */ struct CadetIncomingSession *cadet; /** * How many elements will be supplied in total from the client. */ uint32_t total; /** * Already transferred elements (received) for multipart * messages from client. Always less than @e total. */ uint32_t client_received_element_count; /** * How many elements actually are used for the scalar product. * Size of the arrays in @e r and @e r_prime. Also sometimes * used as an index into the arrays during construction. */ uint32_t used_element_count; /** * Counts the number of values received from Alice by us. * Always less than @e used_element_count. */ uint32_t cadet_received_element_count; /** * Counts the number of values transmitted from us to Alice. * Always less than @e used_element_count. */ uint32_t cadet_transmitted_element_count; /** * State of this session. In * #GNUNET_SCALARPRODUCT_STATUS_ACTIVE while operation is * ongoing, afterwards in #GNUNET_SCALARPRODUCT_STATUS_SUCCESS or * #GNUNET_SCALARPRODUCT_STATUS_FAILURE. */ enum GNUNET_SCALARPRODUCT_ResponseStatus status; /** * Are we already in #destroy_service_session()? */ int in_destroy; /** * The CADET channel. */ struct GNUNET_CADET_Channel *channel; /** * Originator's peer identity. (Only for diagnostics.) */ struct GNUNET_PeerIdentity peer; /** * Public key of the remote service. */ struct GNUNET_CRYPTO_PaillierPublicKey remote_pubkey; /** * The message queue for this channel. */ struct GNUNET_MQ_Handle *cadet_mq; }; /** * GNUnet configuration handle */ static const struct GNUNET_CONFIGURATION_Handle *cfg; /** * Service's own public key */ static struct GNUNET_CRYPTO_PaillierPublicKey my_pubkey; /** * Service's own private key */ static struct GNUNET_CRYPTO_PaillierPrivateKey my_privkey; /** * Service's offset for values that could possibly be negative but are plaintext for encryption. */ static gcry_mpi_t my_offset; /** * Handle to the CADET service. */ static struct GNUNET_CADET_Handle *my_cadet; /** * Callback used to free the elements in the map. * * @param cls NULL * @param key key of the element * @param value the value to free */ static int free_element_cb (void *cls, const struct GNUNET_HashCode *key, void *value) { struct GNUNET_SCALARPRODUCT_Element *element = value; GNUNET_free (element); return GNUNET_OK; } /** * Destroy session state, we are done with it. * * @param session the session to free elements from */ static void destroy_service_session (struct BobServiceSession *s) { unsigned int i; if (GNUNET_YES == s->in_destroy) return; s->in_destroy = GNUNET_YES; if (NULL != s->client) { struct GNUNET_SERVICE_Client *c = s->client; s->client = NULL; GNUNET_SERVICE_client_drop (c); } if (NULL != s->intersected_elements) { GNUNET_CONTAINER_multihashmap_iterate (s->intersected_elements, &free_element_cb, NULL); GNUNET_CONTAINER_multihashmap_destroy (s->intersected_elements); s->intersected_elements = NULL; } if (NULL != s->intersection_op) { GNUNET_SET_operation_cancel (s->intersection_op); s->intersection_op = NULL; } if (NULL != s->intersection_set) { GNUNET_SET_destroy (s->intersection_set); s->intersection_set = NULL; } if (NULL != s->e_a) { GNUNET_free (s->e_a); s->e_a = NULL; } if (NULL != s->sorted_elements) { for (i=0;iused_element_count;i++) gcry_mpi_release (s->sorted_elements[i].value); GNUNET_free (s->sorted_elements); s->sorted_elements = NULL; } if (NULL != s->r) { GNUNET_free (s->r); s->r = NULL; } if (NULL != s->r_prime) { GNUNET_free (s->r_prime); s->r_prime = NULL; } if (NULL != s->port) { GNUNET_CADET_close_port (s->port); s->port = NULL; } if (NULL != s->channel) { GNUNET_CADET_channel_destroy (s->channel); s->channel = NULL; } GNUNET_free (s); } /** * Notify the client that the session has succeeded or failed. This * message gets sent to Bob's client if the operation completed or * Alice disconnected. * * @param session the associated client session to fail or succeed */ static void prepare_client_end_notification (struct BobServiceSession *session) { struct ClientResponseMessage *msg; struct GNUNET_MQ_Envelope *e; if (NULL == session->client_mq) return; /* no client left to be notified */ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending session-end notification with status %d to client for session %s\n", session->status, GNUNET_h2s (&session->session_id)); e = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SCALARPRODUCT_RESULT); msg->range = 0; msg->product_length = htonl (0); msg->status = htonl (session->status); GNUNET_MQ_send (session->client_mq, e); } /** * Function called whenever a channel is destroyed. Should clean up * any associated state. * * It must NOT call #GNUNET_CADET_channel_destroy() on the channel. * * @param cls the `struct BobServiceSession` * @param channel connection to the other end (henceforth invalid) */ static void cb_channel_destruction (void *cls, const struct GNUNET_CADET_Channel *channel) { struct BobServiceSession *s = cls; GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer disconnected, terminating session %s with peer %s\n", GNUNET_h2s (&s->session_id), GNUNET_i2s (&s->peer)); if (GNUNET_SCALARPRODUCT_STATUS_ACTIVE == s->status) { s->status = GNUNET_SCALARPRODUCT_STATUS_FAILURE; prepare_client_end_notification (s); } s->channel = NULL; destroy_service_session (s); } /** * MQ finished giving our last message to CADET, now notify * the client that we are finished. */ static void bob_cadet_done_cb (void *cls) { struct BobServiceSession *session = cls; session->status = GNUNET_SCALARPRODUCT_STATUS_SUCCESS; prepare_client_end_notification (session); } /** * Maximum count of elements we can put into a multipart message */ #define ELEMENT_CAPACITY ((GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE - 1 - sizeof (struct BobCryptodataMultipartMessage)) / sizeof (struct GNUNET_CRYPTO_PaillierCiphertext)) /** * Send a multipart chunk of a service response from Bob to Alice. * This element only contains the two permutations of R, R'. * * @param s the associated service session */ static void transmit_bobs_cryptodata_message_multipart (struct BobServiceSession *s) { struct GNUNET_CRYPTO_PaillierCiphertext *payload; struct BobCryptodataMultipartMessage *msg; struct GNUNET_MQ_Envelope *e; unsigned int i; unsigned int j; uint32_t todo_count; while (s->cadet_transmitted_element_count != s->used_element_count) { todo_count = s->used_element_count - s->cadet_transmitted_element_count; if (todo_count > ELEMENT_CAPACITY / 2) todo_count = ELEMENT_CAPACITY / 2; GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending %u additional crypto values to Alice\n", (unsigned int) todo_count); e = GNUNET_MQ_msg_extra (msg, todo_count * sizeof (struct GNUNET_CRYPTO_PaillierCiphertext) * 2, GNUNET_MESSAGE_TYPE_SCALARPRODUCT_BOB_CRYPTODATA_MULTIPART); msg->contained_element_count = htonl (todo_count); payload = (struct GNUNET_CRYPTO_PaillierCiphertext *) &msg[1]; for (i = s->cadet_transmitted_element_count, j = 0; i < s->cadet_transmitted_element_count + todo_count; i++) { //r[i][p] and r[i][q] GNUNET_memcpy (&payload[j++], &s->r[i], sizeof (struct GNUNET_CRYPTO_PaillierCiphertext)); GNUNET_memcpy (&payload[j++], &s->r_prime[i], sizeof (struct GNUNET_CRYPTO_PaillierCiphertext)); } s->cadet_transmitted_element_count += todo_count; if (s->cadet_transmitted_element_count == s->used_element_count) GNUNET_MQ_notify_sent (e, &bob_cadet_done_cb, s); GNUNET_MQ_send (s->cadet_mq, e); } GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "All values queued for Alice, Bob is done\n"); } /** * Bob generates the response message to be sent to Alice after * computing the values (1), (2), S and S'. * * (1)[]: $E_A(a_{pi(i)}) times E_A(- r_{pi(i)} - b_{pi(i)}) &= E_A(a_{pi(i)} - r_{pi(i)} - b_{pi(i)})$ * (2)[]: $E_A(a_{pi'(i)}) times E_A(- r_{pi'(i)}) &= E_A(a_{pi'(i)} - r_{pi'(i)})$ * S: $S := E_A(sum (r_i + b_i)^2)$ * S': $S' := E_A(sum r_i^2)$ * * @param s the associated requesting session with Alice */ static void transmit_bobs_cryptodata_message (struct BobServiceSession *s) { struct BobCryptodataMessage *msg; struct GNUNET_MQ_Envelope *e; struct GNUNET_CRYPTO_PaillierCiphertext *payload; unsigned int i; s->cadet_transmitted_element_count = ((GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE - 1 - sizeof (struct BobCryptodataMessage)) / sizeof (struct GNUNET_CRYPTO_PaillierCiphertext) / 2) - 1; if (s->cadet_transmitted_element_count > s->used_element_count) s->cadet_transmitted_element_count = s->used_element_count; e = GNUNET_MQ_msg_extra (msg, (2 + s->cadet_transmitted_element_count * 2) * sizeof (struct GNUNET_CRYPTO_PaillierCiphertext), GNUNET_MESSAGE_TYPE_SCALARPRODUCT_BOB_CRYPTODATA); msg->contained_element_count = htonl (s->cadet_transmitted_element_count); GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending %u/%u crypto values to Alice\n", (unsigned int) s->cadet_transmitted_element_count, (unsigned int) s->used_element_count); payload = (struct GNUNET_CRYPTO_PaillierCiphertext *) &msg[1]; GNUNET_memcpy (&payload[0], &s->s, sizeof (struct GNUNET_CRYPTO_PaillierCiphertext)); GNUNET_memcpy (&payload[1], &s->s_prime, sizeof (struct GNUNET_CRYPTO_PaillierCiphertext)); payload = &payload[2]; // convert k[][] for (i = 0; i < s->cadet_transmitted_element_count; i++) { //k[i][p] and k[i][q] GNUNET_memcpy (&payload[i * 2], &s->r[i], sizeof (struct GNUNET_CRYPTO_PaillierCiphertext)); GNUNET_memcpy (&payload[i * 2 + 1], &s->r_prime[i], sizeof (struct GNUNET_CRYPTO_PaillierCiphertext)); } if (s->cadet_transmitted_element_count == s->used_element_count) GNUNET_MQ_notify_sent (e, &bob_cadet_done_cb, s); GNUNET_MQ_send (s->cadet_mq, e); transmit_bobs_cryptodata_message_multipart (s); } #undef ELEMENT_CAPACITY /** * Computes the square sum over a vector of a given length. * * @param vector the vector to compute over * @param length the length of the vector * @return an MPI value containing the calculated sum, never NULL * TODO: code duplication with Alice! */ static gcry_mpi_t compute_square_sum (const gcry_mpi_t *vector, uint32_t length) { gcry_mpi_t elem; gcry_mpi_t sum; uint32_t i; GNUNET_assert (NULL != (sum = gcry_mpi_new (0))); GNUNET_assert (NULL != (elem = gcry_mpi_new (0))); for (i = 0; i < length; i++) { gcry_mpi_mul (elem, vector[i], vector[i]); gcry_mpi_add (sum, sum, elem); } gcry_mpi_release (elem); return sum; } /** * Compute the values * (1)[]: $E_A(a_{pi(i)}) otimes E_A(- r_{pi(i)} - b_{pi(i)}) &= E_A(a_{pi(i)} - r_{pi(i)} - b_{pi(i)})$ * (2)[]: $E_A(a_{pi'(i)}) otimes E_A(- r_{pi'(i)}) &= E_A(a_{pi'(i)} - r_{pi'(i)})$ * S: $S := E_A(sum (r_i + b_i)^2)$ * S': $S' := E_A(sum r_i^2)$ * * @param request the requesting session + bob's requesting peer * @return #GNUNET_OK on success */ static int compute_service_response (struct BobServiceSession *session) { uint32_t i; unsigned int *p; unsigned int *q; uint32_t count; gcry_mpi_t *rand; gcry_mpi_t tmp; const struct MpiElement *b; struct GNUNET_CRYPTO_PaillierCiphertext *a; struct GNUNET_CRYPTO_PaillierCiphertext *r; struct GNUNET_CRYPTO_PaillierCiphertext *r_prime; count = session->used_element_count; a = session->e_a; b = session->sorted_elements; q = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_WEAK, count); p = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_WEAK, count); rand = GNUNET_malloc (sizeof (gcry_mpi_t) * count); for (i = 0; i < count; i++) GNUNET_assert (NULL != (rand[i] = gcry_mpi_new (0))); r = GNUNET_malloc (sizeof (struct GNUNET_CRYPTO_PaillierCiphertext) * count); r_prime = GNUNET_malloc (sizeof (struct GNUNET_CRYPTO_PaillierCiphertext) * count); for (i = 0; i < count; i++) { int32_t svalue; svalue = (int32_t) GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, UINT32_MAX); // long to gcry_mpi_t if (svalue < 0) gcry_mpi_sub_ui (rand[i], rand[i], - svalue); else rand[i] = gcry_mpi_set_ui (rand[i], svalue); } tmp = gcry_mpi_new (0); // encrypt the element // for the sake of readability I decided to have dedicated permutation // vectors, which get rid of all the lookups in p/q. // however, ap/aq are not absolutely necessary but are just abstraction // Calculate Kp = E(S + a_pi) (+) E(S - r_pi - b_pi) for (i = 0; i < count; i++) { // E(S - r_pi - b_pi) gcry_mpi_sub (tmp, my_offset, rand[p[i]]); gcry_mpi_sub (tmp, tmp, b[p[i]].value); GNUNET_assert (2 == GNUNET_CRYPTO_paillier_encrypt (&session->remote_pubkey, tmp, 2, &r[i])); // E(S - r_pi - b_pi) * E(S + a_pi) == E(2*S + a - r - b) if (GNUNET_OK != GNUNET_CRYPTO_paillier_hom_add (&session->remote_pubkey, &r[i], &a[p[i]], &r[i])) { GNUNET_break_op (0); goto error_cleanup; } } // Calculate Kq = E(S + a_qi) (+) E(S - r_qi) for (i = 0; i < count; i++) { // E(S - r_qi) gcry_mpi_sub (tmp, my_offset, rand[q[i]]); GNUNET_assert (2 == GNUNET_CRYPTO_paillier_encrypt (&session->remote_pubkey, tmp, 2, &r_prime[i])); // E(S - r_qi) * E(S + a_qi) == E(2*S + a_qi - r_qi) if (GNUNET_OK != GNUNET_CRYPTO_paillier_hom_add (&session->remote_pubkey, &r_prime[i], &a[q[i]], &r_prime[i])) { GNUNET_break_op (0); goto error_cleanup; } } gcry_mpi_release (tmp); // Calculate S' = E(SUM( r_i^2 )) tmp = compute_square_sum (rand, count); GNUNET_assert (1 == GNUNET_CRYPTO_paillier_encrypt (&session->remote_pubkey, tmp, 1, &session->s_prime)); gcry_mpi_release (tmp); // Calculate S = E(SUM( (r_i + b_i)^2 )) for (i = 0; i < count; i++) gcry_mpi_add (rand[i], rand[i], b[i].value); tmp = compute_square_sum (rand, count); GNUNET_assert (1 == GNUNET_CRYPTO_paillier_encrypt (&session->remote_pubkey, tmp, 1, &session->s)); gcry_mpi_release (tmp); session->r = r; session->r_prime = r_prime; for (i = 0; i < count; i++) gcry_mpi_release (rand[i]); GNUNET_free (session->e_a); session->e_a = NULL; GNUNET_free (p); GNUNET_free (q); GNUNET_free (rand); return GNUNET_OK; error_cleanup: GNUNET_free (r); GNUNET_free (r_prime); gcry_mpi_release (tmp); GNUNET_free (p); GNUNET_free (q); for (i = 0; i < count; i++) gcry_mpi_release (rand[i]); GNUNET_free (rand); return GNUNET_SYSERR; } /** * Iterator to copy over messages from the hash map * into an array for sorting. * * @param cls the `struct BobServiceSession *` * @param key the key (unused) * @param value the `struct GNUNET_SCALARPRODUCT_Element *` * TODO: code duplication with Alice! */ static int copy_element_cb (void *cls, const struct GNUNET_HashCode *key, void *value) { struct BobServiceSession *s = cls; struct GNUNET_SCALARPRODUCT_Element *e = value; gcry_mpi_t mval; int64_t val; mval = gcry_mpi_new (0); val = (int64_t) GNUNET_ntohll (e->value); if (0 > val) gcry_mpi_sub_ui (mval, mval, -val); else gcry_mpi_add_ui (mval, mval, val); s->sorted_elements [s->used_element_count].value = mval; s->sorted_elements [s->used_element_count].key = &e->key; s->used_element_count++; return GNUNET_OK; } /** * Compare two `struct MpiValue`s by key for sorting. * * @param a pointer to first `struct MpiValue *` * @param b pointer to first `struct MpiValue *` * @return -1 for a < b, 0 for a=b, 1 for a > b. * TODO: code duplication with Alice! */ static int element_cmp (const void *a, const void *b) { const struct MpiElement *ma = a; const struct MpiElement *mb = b; return GNUNET_CRYPTO_hash_cmp (ma->key, mb->key); } /** * Intersection operation and receiving data via CADET from * Alice are both done, compute and transmit our reply via * CADET. * * @param s session to transmit reply for. */ static void transmit_cryptographic_reply (struct BobServiceSession *s) { struct GNUNET_CADET_Channel *channel; /* TODO: code duplication with Alice! */ LOG (GNUNET_ERROR_TYPE_DEBUG, "Received everything, building reply for Alice\n"); s->sorted_elements = GNUNET_malloc (GNUNET_CONTAINER_multihashmap_size (s->intersected_elements) * sizeof (struct MpiElement)); s->used_element_count = 0; GNUNET_CONTAINER_multihashmap_iterate (s->intersected_elements, ©_element_cb, s); qsort (s->sorted_elements, s->used_element_count, sizeof (struct MpiElement), &element_cmp); if (GNUNET_OK != compute_service_response (s)) { channel = s->channel; s->channel = NULL; GNUNET_CADET_channel_destroy (channel); return; } transmit_bobs_cryptodata_message (s); } /** * Check a multipart-chunk of a request from another service to * calculate a scalarproduct with us. * * @param cls the `struct BobServiceSession *` * @param msg the actual message * @return #GNUNET_OK to keep the connection open, * #GNUNET_SYSERR to close it (signal serious error) */ static int check_alices_cryptodata_message (void *cls, const struct AliceCryptodataMessage *msg) { struct BobServiceSession *s = cls; uint32_t contained_elements; size_t msg_length; uint16_t msize; unsigned int max; msize = ntohs (msg->header.size); contained_elements = ntohl (msg->contained_element_count); /* Our intersection may still be ongoing, but this is nevertheless an upper bound on the required array size */ max = GNUNET_CONTAINER_multihashmap_size (s->intersected_elements); msg_length = sizeof (struct AliceCryptodataMessage) + contained_elements * sizeof (struct GNUNET_CRYPTO_PaillierCiphertext); if ( (msize != msg_length) || (0 == contained_elements) || (contained_elements > UINT16_MAX) || (max < contained_elements + s->cadet_received_element_count) ) { GNUNET_break_op (0); return GNUNET_SYSERR; } return GNUNET_OK; } /** * Handle a multipart-chunk of a request from another service to * calculate a scalarproduct with us. * * @param cls the `struct BobServiceSession *` * @param msg the actual message */ static void handle_alices_cryptodata_message (void *cls, const struct AliceCryptodataMessage *msg) { struct BobServiceSession *s = cls; const struct GNUNET_CRYPTO_PaillierCiphertext *payload; uint32_t contained_elements; unsigned int max; contained_elements = ntohl (msg->contained_element_count); /* Our intersection may still be ongoing, but this is nevertheless an upper bound on the required array size */ max = GNUNET_CONTAINER_multihashmap_size (s->intersected_elements); GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Received %u crypto values from Alice\n", (unsigned int) contained_elements); payload = (const struct GNUNET_CRYPTO_PaillierCiphertext *) &msg[1]; if (NULL == s->e_a) s->e_a = GNUNET_new_array (max, struct GNUNET_CRYPTO_PaillierCiphertext); GNUNET_memcpy (&s->e_a[s->cadet_received_element_count], payload, sizeof (struct GNUNET_CRYPTO_PaillierCiphertext) * contained_elements); s->cadet_received_element_count += contained_elements; if ( (s->cadet_received_element_count == max) && (NULL == s->intersection_op) ) { /* intersection has finished also on our side, and we got the full set, so we can proceed with the CADET response(s) */ transmit_cryptographic_reply (s); } GNUNET_CADET_receive_done (s->channel); } /** * Callback for set operation results. Called for each element * that needs to be removed from the result set. * * @param cls closure with the `struct BobServiceSession` * @param element a result element, only valid if status is #GNUNET_SET_STATUS_OK * @param current_size current set size * @param status what has happened with the set intersection? */ static void cb_intersection_element_removed (void *cls, const struct GNUNET_SET_Element *element, uint64_t current_size, enum GNUNET_SET_Status status) { struct BobServiceSession *s = cls; struct GNUNET_SCALARPRODUCT_Element *se; switch (status) { case GNUNET_SET_STATUS_OK: /* this element has been removed from the set */ se = GNUNET_CONTAINER_multihashmap_get (s->intersected_elements, element->data); GNUNET_assert (NULL != se); LOG (GNUNET_ERROR_TYPE_DEBUG, "Removed element with key %s and value %lld\n", GNUNET_h2s (&se->key), (long long) GNUNET_ntohll (se->value)); GNUNET_assert (GNUNET_YES == GNUNET_CONTAINER_multihashmap_remove (s->intersected_elements, element->data, se)); GNUNET_free (se); return; case GNUNET_SET_STATUS_DONE: s->intersection_op = NULL; GNUNET_break (NULL == s->intersection_set); GNUNET_CADET_receive_done (s->channel); LOG (GNUNET_ERROR_TYPE_DEBUG, "Finished intersection, %d items remain\n", GNUNET_CONTAINER_multihashmap_size (s->intersected_elements)); if (s->client_received_element_count == GNUNET_CONTAINER_multihashmap_size (s->intersected_elements)) { /* CADET transmission from Alice is also already done, start with our own reply */ transmit_cryptographic_reply (s); } return; case GNUNET_SET_STATUS_HALF_DONE: /* unexpected for intersection */ GNUNET_break (0); return; case GNUNET_SET_STATUS_FAILURE: /* unhandled status code */ LOG (GNUNET_ERROR_TYPE_DEBUG, "Set intersection failed!\n"); s->intersection_op = NULL; if (NULL != s->intersection_set) { GNUNET_SET_destroy (s->intersection_set); s->intersection_set = NULL; } s->status = GNUNET_SCALARPRODUCT_STATUS_FAILURE; prepare_client_end_notification (s); return; default: GNUNET_break (0); return; } } /** * We've paired up a client session with an incoming CADET request. * Initiate set intersection work. * * @param s client session to start intersection for */ static void start_intersection (struct BobServiceSession *s) { GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Got session with key %s and %u elements, starting intersection.\n", GNUNET_h2s (&s->session_id), (unsigned int) s->total); s->intersection_op = GNUNET_SET_prepare (&s->peer, &s->session_id, NULL, GNUNET_SET_RESULT_REMOVED, (struct GNUNET_SET_Option[]) {{ 0 }}, &cb_intersection_element_removed, s); if (GNUNET_OK != GNUNET_SET_commit (s->intersection_op, s->intersection_set)) { GNUNET_break (0); s->status = GNUNET_SCALARPRODUCT_STATUS_FAILURE; prepare_client_end_notification (s); return; } GNUNET_SET_destroy (s->intersection_set); s->intersection_set = NULL; } /** * Handle a request from Alice to calculate a scalarproduct with us (Bob). * * @param cls the `struct BobServiceSession *` * @param msg the actual message */ static void handle_alices_computation_request (void *cls, const struct ServiceRequestMessage *msg) { struct BobServiceSession *s = cls; s->session_id = msg->session_id; // ?? s->remote_pubkey = msg->public_key; if (s->client_received_element_count == s->total) start_intersection (s); } /** * Function called for inbound channels on Bob's end. Does some * preliminary initialization, more happens after we get Alice's first * message. * * @param cls closure with the `struct BobServiceSession` * @param channel new handle to the channel * @param initiator peer that started the channel * @return session associated with the channel */ static void * cb_channel_incoming (void *cls, struct GNUNET_CADET_Channel *channel, const struct GNUNET_PeerIdentity *initiator) { struct BobServiceSession *s = cls; GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "New incoming channel from peer %s.\n", GNUNET_i2s (initiator)); GNUNET_CADET_close_port (s->port); s->port = NULL; s->channel = channel; s->peer = *initiator; s->cadet_mq = GNUNET_CADET_get_mq (s->channel); return s; } /** * We're receiving additional set data. Check it is well-formed. * * @param cls identification of the client * @param msg the actual message * @return #GNUNET_OK if @a msg is well-formed */ static int check_bob_client_message_multipart (void *cls, const struct ComputationBobCryptodataMultipartMessage *msg) { struct BobServiceSession *s = cls; uint32_t contained_count; uint16_t msize; msize = ntohs (msg->header.size); contained_count = ntohl (msg->element_count_contained); if ( (msize != (sizeof (struct ComputationBobCryptodataMultipartMessage) + contained_count * sizeof (struct GNUNET_SCALARPRODUCT_Element))) || (0 == contained_count) || (UINT16_MAX < contained_count) || (s->total == s->client_received_element_count) || (s->total < s->client_received_element_count + contained_count) ) { GNUNET_break (0); return GNUNET_SYSERR; } return GNUNET_OK; } /** * We're receiving additional set data. Add it to our * set and if we are done, initiate the transaction. * * @param cls identification of the client * @param msg the actual message */ static void handle_bob_client_message_multipart (void *cls, const struct ComputationBobCryptodataMultipartMessage *msg) { struct BobServiceSession *s = cls; uint32_t contained_count; const struct GNUNET_SCALARPRODUCT_Element *elements; struct GNUNET_SET_Element set_elem; struct GNUNET_SCALARPRODUCT_Element *elem; contained_count = ntohl (msg->element_count_contained); elements = (const struct GNUNET_SCALARPRODUCT_Element *) &msg[1]; for (uint32_t i = 0; i < contained_count; i++) { elem = GNUNET_new (struct GNUNET_SCALARPRODUCT_Element); GNUNET_memcpy (elem, &elements[i], sizeof (struct GNUNET_SCALARPRODUCT_Element)); if (GNUNET_SYSERR == GNUNET_CONTAINER_multihashmap_put (s->intersected_elements, &elem->key, elem, GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY)) { GNUNET_break (0); GNUNET_free (elem); continue; } set_elem.data = &elem->key; set_elem.size = sizeof (elem->key); set_elem.element_type = 0; GNUNET_SET_add_element (s->intersection_set, &set_elem, NULL, NULL); } s->client_received_element_count += contained_count; GNUNET_SERVICE_client_continue (s->client); if (s->total != s->client_received_element_count) { /* more to come */ return; } if (NULL == s->channel) { /* no Alice waiting for this request, wait for Alice */ return; } start_intersection (s); } /** * Handler for Bob's a client request message. Check @a msg is * well-formed. * * @param cls identification of the client * @param msg the actual message * @return #GNUNET_OK if @a msg is well-formed */ static int check_bob_client_message (void *cls, const struct BobComputationMessage *msg) { struct BobServiceSession *s = cls; uint32_t contained_count; uint32_t total_count; uint16_t msize; if (GNUNET_SCALARPRODUCT_STATUS_INIT != s->status) { GNUNET_break (0); return GNUNET_SYSERR; } msize = ntohs (msg->header.size); total_count = ntohl (msg->element_count_total); contained_count = ntohl (msg->element_count_contained); if ( (0 == total_count) || (0 == contained_count) || (UINT16_MAX < contained_count) || (msize != (sizeof (struct BobComputationMessage) + contained_count * sizeof (struct GNUNET_SCALARPRODUCT_Element))) ) { GNUNET_break_op (0); return GNUNET_SYSERR; } return GNUNET_OK; } /** * Handler for Bob's a client request message. Bob is in the response * role, keep the values + session and waiting for a matching session * or process a waiting request from Alice. * * @param cls identification of the client * @param msg the actual message */ static void handle_bob_client_message (void *cls, const struct BobComputationMessage *msg) { struct BobServiceSession *s = cls; struct GNUNET_MQ_MessageHandler cadet_handlers[] = { GNUNET_MQ_hd_fixed_size (alices_computation_request, GNUNET_MESSAGE_TYPE_SCALARPRODUCT_SESSION_INITIALIZATION, struct ServiceRequestMessage, NULL), GNUNET_MQ_hd_var_size (alices_cryptodata_message, GNUNET_MESSAGE_TYPE_SCALARPRODUCT_ALICE_CRYPTODATA, struct AliceCryptodataMessage, NULL), GNUNET_MQ_handler_end () }; uint32_t contained_count; uint32_t total_count; const struct GNUNET_SCALARPRODUCT_Element *elements; struct GNUNET_SET_Element set_elem; struct GNUNET_SCALARPRODUCT_Element *elem; total_count = ntohl (msg->element_count_total); contained_count = ntohl (msg->element_count_contained); s->status = GNUNET_SCALARPRODUCT_STATUS_ACTIVE; s->total = total_count; s->client_received_element_count = contained_count; s->session_id = msg->session_key; elements = (const struct GNUNET_SCALARPRODUCT_Element *) &msg[1]; s->intersected_elements = GNUNET_CONTAINER_multihashmap_create (s->total, GNUNET_YES); s->intersection_set = GNUNET_SET_create (cfg, GNUNET_SET_OPERATION_INTERSECTION); for (uint32_t i = 0; i < contained_count; i++) { if (0 == GNUNET_ntohll (elements[i].value)) continue; elem = GNUNET_new (struct GNUNET_SCALARPRODUCT_Element); GNUNET_memcpy (elem, &elements[i], sizeof (struct GNUNET_SCALARPRODUCT_Element)); if (GNUNET_SYSERR == GNUNET_CONTAINER_multihashmap_put (s->intersected_elements, &elem->key, elem, GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY)) { GNUNET_break (0); GNUNET_free (elem); continue; } set_elem.data = &elem->key; set_elem.size = sizeof (elem->key); set_elem.element_type = 0; GNUNET_SET_add_element (s->intersection_set, &set_elem, NULL, NULL); s->used_element_count++; } GNUNET_SERVICE_client_continue (s->client); /* We're ready, open the port */ s->port = GNUNET_CADET_open_port (my_cadet, &msg->session_key, &cb_channel_incoming, s, NULL, &cb_channel_destruction, cadet_handlers); if (NULL == s->port) { GNUNET_break (0); GNUNET_SERVICE_client_drop (s->client); return; } } /** * Task run during shutdown. * * @param cls unused */ static void shutdown_task (void *cls) { GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Shutting down, initiating cleanup.\n"); // FIXME: we have to cut our connections to CADET first! if (NULL != my_cadet) { GNUNET_CADET_disconnect (my_cadet); my_cadet = NULL; } } /** * A client connected. * * Setup the associated data structure. * * @param cls closure, NULL * @param client identification of the client * @param mq message queue to communicate with @a client * @return our `struct BobServiceSession` */ static void * client_connect_cb (void *cls, struct GNUNET_SERVICE_Client *client, struct GNUNET_MQ_Handle *mq) { struct BobServiceSession *s; s = GNUNET_new (struct BobServiceSession); s->client = client; s->client_mq = mq; return s; } /** * A client disconnected. * * Remove the associated session(s), release data structures * and cancel pending outgoing transmissions to the client. * * @param cls closure, NULL * @param client identification of the client * @param app_cls our `struct BobServiceSession` */ static void client_disconnect_cb (void *cls, struct GNUNET_SERVICE_Client *client, void *app_cls) { struct BobServiceSession *s = app_cls; GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Client disconnected from us.\n"); s->client = NULL; destroy_service_session (s); } /** * Initialization of the program and message handlers * * @param cls closure * @param c configuration to use * @param service the initialized service */ static void run (void *cls, const struct GNUNET_CONFIGURATION_Handle *c, struct GNUNET_SERVICE_Handle *service) { cfg = c; /* offset has to be sufficiently small to allow computation of: m1+m2 mod n == (S + a) + (S + b) mod n, if we have more complex operations, this factor needs to be lowered */ my_offset = gcry_mpi_new (GNUNET_CRYPTO_PAILLIER_BITS / 3); gcry_mpi_set_bit (my_offset, GNUNET_CRYPTO_PAILLIER_BITS / 3); GNUNET_CRYPTO_paillier_create (&my_pubkey, &my_privkey); my_cadet = GNUNET_CADET_connect (cfg); GNUNET_SCHEDULER_add_shutdown (&shutdown_task, NULL); if (NULL == my_cadet) { GNUNET_log (GNUNET_ERROR_TYPE_ERROR, _("Connect to CADET failed\n")); GNUNET_SCHEDULER_shutdown (); return; } } /** * Define "main" method using service macro. */ GNUNET_SERVICE_MAIN ("scalarproduct-bob", GNUNET_SERVICE_OPTION_NONE, &run, &client_connect_cb, &client_disconnect_cb, NULL, GNUNET_MQ_hd_var_size (bob_client_message, GNUNET_MESSAGE_TYPE_SCALARPRODUCT_CLIENT_TO_BOB, struct BobComputationMessage, NULL), GNUNET_MQ_hd_var_size (bob_client_message_multipart, GNUNET_MESSAGE_TYPE_SCALARPRODUCT_CLIENT_MULTIPART_BOB, struct ComputationBobCryptodataMultipartMessage, NULL), GNUNET_MQ_handler_end ()); /* end of gnunet-service-scalarproduct_bob.c */