guix-gnunet/c-code/rehash-service.c

/* rehash --- a decentralised hash<->hash store
   Copyright © 2020 Maxime Devos <maxime.devos@student.kuleuven.be>

   This file is part of rehash.

   rehash 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.

   rehash 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 General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with rehash.  If not, see <http://www.gnu.org/licenses/>. */

#include "platform.h"
#include <stdio.h>
#include <gnunet/gnunet_service_lib.h>
#include <gnunet/gnunet_container_lib.h>
#include <gnunet/gnunet_dht_service.h>
#include <gnunet/gnunet_datastore_service.h>
#include "rehash_service.h"
#include "extra_gnunet_protocols.h"
#include "rehash.h"
#include "rehash_crypto.h"
#include "rehash_dht.h"

/* FIXME no magic constants

   An upper bound on hash sizes found in the wild. */
#define MAGIC 64

static struct GNUNET_DHT_Handle *dht_handle;
static struct GNUNET_DATASTORE_Handle *ds_handle;
/* TODO bring old data back into the DHT */
/* TODO monitor the network for known bad hashes,
   and kill them when passing the local peer. */

enum ServerContextType
{
  SERVER_CONTEXT_GET,
  SERVER_CONTEXT_PUT,
};
struct ClientContext;

/** Information about in-process hash->hash lookup
    requests. */
struct GetContext
{
  enum ServerContextType ctx;
  uint32_t request_id;
  struct ClientContext *client;

  /* TODO the API allows controlling
     where to find hashes ( )*/

  /* Never NULL actually. */
  struct GNUNET_DHT_GetHandle *dht_get;
  /* If NULL, the input hash was not present
     in the datastore. */
  struct GNUNET_DATASTORE_QueueEntry *datastore_get;
};

/** Information about in-process hash->hash insertions. */
struct PutContext
{
  enum ServerContextType ctx;
  uint32_t request_id;
  struct ClientContext *client;

  /* If NULL, the mapping has been send onto
     the DHT. Once datastore_put is completed,
     the task FIXME??? will periodically
     reenter the hashes into the DHT. */
  struct GNUNET_DHT_PutHandle *dht_put;
  /* If NULL, the mapping has been inserted into
     the datastore. */
  struct GNUNET_DATASTORE_QueueEntry *datastore_put;
};

struct ClientContext
{
  struct GNUNET_SERVICE_Client *client;
  /* Requests from C->S */
  struct GNUNET_CONTAINER_MultiHashMap32 *requests;
};

/**
 * Callback run when shutting down the rehash service
 */
static void
shutdown_task (void *cls)
{
  /* FIXME perhaps let store requests complete first? */
  GNUNET_DHT_disconnect (dht_handle);
  GNUNET_DATASTORE_disconnect (ds_handle, GNUNET_NO);
  /* prevent accidental use-after-free */
  dht_handle = NULL;
  ds_handle = NULL;
}

/**
 * Callback to initialise the rehash service
 */
static void
init_cb (void *cls, const struct GNUNET_CONFIGURATION_Handle *cfg,
         struct GNUNET_SERVICE_Handle *sh)
{
  /* TODO: reasonable length */
  dht_handle = GNUNET_DHT_connect (cfg, 16);
  ds_handle  = GNUNET_DATASTORE_connect (cfg);
  /* TODO: what to do in these cases,
     and is this possible? */
  GNUNET_assert(dht_handle != NULL);
  GNUNET_assert(ds_handle != NULL);
  GNUNET_SCHEDULER_add_shutdown (&shutdown_task, NULL);
}

/**
 * Callback for when a client connects to the service.
 */
static void *
connect_cb(void *cls, struct GNUNET_SERVICE_Client *c,
           struct GNUNET_MQ_Handle *mq)
{
  struct ClientContext *ctx;
  ctx = GNUNET_new(struct ClientContext);
  ctx->client = c;
  /* length is a wild guess */
  ctx->requests = GNUNET_CONTAINER_multihashmap32_create (8);
  /* TODO when is this possible */
  GNUNET_assert(ctx->requests != NULL);
  return ctx;
}

/**
 * Callback for when a client is disconnected from the service.
 */
static void
disconnect_cb(void *cls, struct GNUNET_SERVICE_Client *c, void *internal_cls)
{
  /* TODO free elements of cls->requests, abort requests ... */
}

static int
check_get (void *cls, const struct REHASH_GetMessage *msg)
{
  struct ClientContext *ctx;
  uint16_t header_size;
  uint32_t input_length; /* TODO perhaps uint16_t */
  ctx = cls;
  header_size = ntohs(msg->header.size);
  input_length = ntohl(msg->input_length);
  if (header_size - sizeof(struct REHASH_GetMessage) != input_length)
  {
    GNUNET_break(0);
    return GNUNET_SYSERR;
  }

  /* Detect duplicate message ids */
  if (GNUNET_CONTAINER_multihashmap32_contains (ctx->requests, msg->request_id))
    /* TODO check if GNUNET_break below is redundant */
    return GNUNET_SYSERR;

  /* Detect unsupported options */
  if (ntohl(msg->options) > 1)
  {
    GNUNET_break(0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}

/* Inform the client a hash has been found */
static void
inform_hash_found (struct GetContext *ctx,
		   struct GNUNET_TIME_Absolute exp,
		   size_t size,
		   const void *data)
{
  struct GNUNET_SERVICE_Client *client;
  /* TODO: send a REHASH_ResultMessage */
  struct GNUNET_MQ_Envelope *ev;
  struct REHASH_ResultMessage *msg;
  /* Prevent buffer overflow. */
  if (size > MAGIC) {
    GNUNET_break (0);
    return;
  }
  client = ctx->client->client;

  /* Inform the client of the found message */
  ev = GNUNET_MQ_msg_extra (msg, size, GNUNET_MESSAGE_TYPE_REHASH_CLIENT_RESULT);
  msg->request_id = ctx->request_id;
  msg->output_length = htonl (size);
  msg->exp = GNUNET_TIME_absolute_hton (exp);
  memcpy(&msg[1], data, size);

  GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (client), ev);
}

static void
dht_get_iterator(void *cls,
                 struct GNUNET_TIME_Absolute exp,
                 const struct GNUNET_HashCode *key,
                 const struct GNUNET_PeerIdentity *get_path,
                 unsigned int get_path_length,
                 const struct GNUNET_PeerIdentity *put_path,
                 unsigned int put_path_length,
                 enum GNUNET_BLOCK_Type type,
                 size_t size,
                 const void *data)
{
  /* TODO: who should free get_path, put_path?*/
  struct GetContext *ctx;
  ctx = cls;
  GNUNET_assert (ctx->ctx == SERVER_CONTEXT_GET);
  inform_hash_found (ctx, exp, size, data);
}

/**
 * Callback for when a hash->hash mapping has been found
 * in the datastore.
 */
static void
datastore_get_cb (void *cls,
		  const struct GNUNET_HashCode *key,
		  size_t size,
		  const void *data,
		  enum GNUNET_BLOCK_Type type,
		  uint32_t priority,
		  uint32_t anonymity,
		  uint32_t replication,
		  struct GNUNET_TIME_Absolute expiration,
		  uint64_t uid)
{
  struct GetContext *ctx;
  GNUNET_assert (ctx->ctx == SERVER_CONTEXT_GET);

  /* TODO should ctx->datastore_get be freed? */
  ctx->datastore_get = NULL;
  if ((data == NULL) || (key == NULL))
    /* No entry found!
       Or our request was dropped, not sure. */
    /* TODO figure out exact condition */
    return;

  /* prevent overflow*/
  if (size > MAGIC)
    return;

  inform_hash_found (ctx, expiration, size, data);
}

static void
handle_get (void *cls, const struct REHASH_GetMessage *msg)
{
  int ret;
  struct GetContext *ctx;
  struct ClientContext *c;
  struct GNUNET_DHT_GetHandle *h;
  struct GNUNET_HashCode key;
  c = cls;
  /* TODO: define protocols for anonymous GET */
  GNUNET_assert(ntohl(msg->anonymity_level) == 0);
  GNUNET_assert(0 && "understood get!");
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
	      "Client started a search for a hash\n");
  if (GNUNET_OK !=
      REHASH_obfuscated_query_from_hash
      (ntohl(msg->out_type),
       ntohl(msg->in_type),
       (const char *) &msg[1],
       ntohl(msg->input_length),
       &key))
    /* TODO bweh? disconnect? */
    return;

  /* TODO: desired replication level */
  ctx = GNUNET_new(struct GetContext);
  *ctx = (struct GetContext) {};
  ctx->ctx = SERVER_CONTEXT_GET;
  ctx->client = c;
  ctx->request_id = msg->request_id;

  /* Remember the request */
  ret = GNUNET_CONTAINER_multihashmap32_put
    (c->requests, ctx->request_id, ctx,
     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
  GNUNET_assert (ret == GNUNET_OK);

  /* Start looking in the DHT
     TODO: maybe look in the datastore first
     before bothering the network? */
  ctx->dht_get = REHASH_dht_get_start
    (dht_handle, &key, 1, GNUNET_DHT_RO_NONE,
     &dht_get_iterator, ctx);
  /* Start looking in the datastore.

     Technically redundant, as the mappings
     in the datastore are periodically
     entered into the DHT. However, there
     are realistic situations where waiting
     on the DHT to catch up may not be ideal:

     E.g. the DHT service (or dhtcache maybe?)
     might have recently restarted (e.g. due to a
     system upgrade) and the peer isn't connected
     to useful other peers (e.g. network is down). */
  ctx->datastore_get = GNUNET_DATASTORE_get_key
    (ds_handle, 0, 0, &key,
     GNUNET_BLOCK_TYPE_REHASH,
     /* FIXME priority and what's this about dropping requests? */
     100,
     100,
     &datastore_get_cb,
     cls);
  /* Note: datastore_get can be NULL */

  /* Allow more queries, and possibly aborts */
  GNUNET_SERVICE_client_continue (ctx->client->client);
  /* TODO: free h eventually */
}

static int
check_get_stop (void *cls, const struct REHASH_GetStopMessage *msg)
{
  GNUNET_assert(0);
}

static void
handle_get_stop (void *cls, const struct REHASH_GetStopMessage *msg)
{
}

static int
check_put (void *cls, const struct REHASH_PutMessage *msg)
{
  struct ClientContext *ctx;
  uint32_t input_length;
  uint32_t output_length;
  ctx = cls;
  input_length = ntohl(msg->input_length);
  output_length = ntohl(msg->output_length);
  /* Prevent overflow TODO no magic values
     TODO not really invalid necessarily */
  if (input_length > MAGIC)
    return GNUNET_SYSERR;
  if (output_length > MAGIC)
    return GNUNET_SYSERR;
  if (input_length + output_length + sizeof(*msg)
      != ntohs(msg->header.size))
    return GNUNET_SYSERR;
  /* Detect duplicate message ids */
  if (GNUNET_CONTAINER_multihashmap32_contains (ctx->requests, msg->request_id))
    /* TODO check if GNUNET_break below is redundant */
    return GNUNET_SYSERR;
  /* TODO prevent saving of hashes of incorrect lengths */
  return GNUNET_OK;
}

/* A part of a hash->hash insertion seems to have
   succeeded, perhaps tell that to the client? */
static void
perhaps_complete_put (struct PutContext *ctx)
{
  struct REHASH_PutStatusMessage *msg;
  struct GNUNET_SERVICE_Client *client;
  struct GNUNET_MQ_Envelope *ev;
  uint32_t request_id;

  if (ctx->datastore_put || ctx->dht_put)
    /* Still something to do! */
    return;

  /* Done! (This actually is somewhat optimistic,
     as the DHT and datastore service may perform
     the actual insertion in the background.

     The status update is still useful though:
     consider a misconfigured system (or a memory loaded
     system where the OOM killer killed some
     critical GNUnet services).

     In that case, it is useful for the client to know if
     an insertion was ‘probably successful’,.
     even if there is a tiny race window. Don't expect
     hard guarantees of rehash, use a database or
     file system instead if these are required. )*/
  client = ctx->client->client;
  request_id = ctx->request_id;
  GNUNET_free (ctx);

  /* Inform the client of the found message */
  /* TODO ordering abort / done messages.
     What if the client just sent an abort?*/
  ev = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_REHASH_PUT_DONE);
  msg->request_id = request_id;
  msg->flags = htonl(REHASH_PUT_COMPLETED);
  /* ev is consumed by GNUNET_MQ_send */
  GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (client), ev);
}

static void
datastore_put_cb (void *cls,
		  int32_t success,
		  struct GNUNET_TIME_Absolute min_expiration,
		  const char *msg)
{
  /* TODO: do something I guess?
     SYSERR: failure
     NO: already exists
     YES: exists
     msg: error message */
  struct PutContext *ctx;
  ctx = cls;
  /* TODO should this be aborted? */
  ctx->datastore_put = NULL;
  perhaps_complete_put (ctx);
}

static void
dht_put_cb (void *cls)
{
  struct PutContext *ctx;
  ctx = cls;
  /* This has been freed */
  ctx->dht_put = NULL;
  perhaps_complete_put (ctx);
}

static void
handle_put (void *cls, const struct REHASH_PutMessage *msg)
{
  int ret;
  struct ClientContext *client;
  struct PutContext *ctx;
  struct GNUNET_HashCode query;
  char *dest;
  const char *out_data;
  const char *in_data;
  ssize_t expected_length;
  ssize_t serialised_length;
  GNUNET_assert(ntohl(msg->anonymity_level) == 0);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
	      "Client entered a hash->hash mapping to save\n");
  /* TODO also put into datastore */
  if (GNUNET_OK !=
      REHASH_obfuscated_query_from_hash
      (ntohl(msg->out_type),
       ntohl(msg->in_type),
       (const char *) &msg[1],
       ntohl(msg->input_length),
       &query))
  {
    /* TODO: ? TODO */
    GNUNET_break(0);
    return;
  }

  /* Prepare data to put into the DHT and datastore */
  in_data = (const char *) &msg[1];
  out_data = in_data + ntohl(msg->input_length);
  expected_length = REHASH_data_size_for_mapping
    (ntohl(msg->output_length));
  dest = GNUNET_malloc (expected_length);
  serialised_length = REHASH_data_for_mapping
    (ntohl(msg->out_type),
     ntohl(msg->in_type),
     out_data,
     in_data,
     ntohl(msg->output_length),
     ntohl(msg->input_length),
     dest,
     expected_length);
  if (serialised_length != expected_length)
  {
    GNUNET_free (dest);
    GNUNET_break (0);
    /* TODO error message */
    /* TODO interaction with aborting */
    /* FIXME either disconnect the client
       or resume the client! */
    return;
  }

  client = cls;

  /* Allow aborting the request */
  ctx = GNUNET_new(struct PutContext);
  *ctx = (struct PutContext) {};
  ctx->ctx = SERVER_CONTEXT_PUT;
  ctx->request_id = msg->request_id;
  ctx->client = client;

  /* Remember the request */
  ret = GNUNET_CONTAINER_multihashmap32_put
    (client->requests, ctx->request_id, ctx,
     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
  GNUNET_assert (ret == GNUNET_OK);

  /* Put data into the DHT */
  ctx->dht_put = REHASH_dht_put
    (dht_handle,
     &query,
     ntohl(msg->replication_level),
     GNUNET_DHT_RO_NONE,
     serialised_length,
     dest,
     GNUNET_TIME_absolute_ntoh(msg->expiration_time),
     /* TODO callback */
     &dht_put_cb,
     ctx);
  ctx->datastore_put = GNUNET_DATASTORE_put
    (ds_handle,
     0 /* reservation id*/,
     &query,
     serialised_length,
     dest,
     GNUNET_BLOCK_TYPE_REHASH,
     /* FIXME priority */
     1000 /* ntohl (msg->priority) */,
     ntohl (msg->anonymity_level),
     ntohl (msg->replication_level),
     GNUNET_TIME_absolute_ntoh (msg->expiration_time),
     /* XXX ??? queue properties */
     100,
     100,
     &datastore_put_cb,
     ctx);

  /* TODO: has dest to be kept not-freed
     while datastore_put and dht_put are running? */
  GNUNET_free (dest);
  /* TODO free h eventually (at abort perhaps?) */
  GNUNET_SERVICE_client_continue (client->client);
}

GNUNET_SERVICE_MAIN
("rehash",
 GNUNET_SERVICE_OPTION_NONE,
 &init_cb,
 &connect_cb,
 &disconnect_cb,
 NULL,
 /* TODO MQ handlers! */
 GNUNET_MQ_hd_var_size (get,
                        GNUNET_MESSAGE_TYPE_REHASH_CLIENT_GET,
                        struct REHASH_GetMessage,
                        NULL),
 GNUNET_MQ_hd_var_size (get_stop,
                        GNUNET_MESSAGE_TYPE_REHASH_CLIENT_GET_STOP,
                        struct REHASH_GetStopMessage,
                        NULL),
 GNUNET_MQ_hd_var_size (put,
                        GNUNET_MESSAGE_TYPE_REHASH_CLIENT_PUT,
                        struct REHASH_PutMessage,
                        NULL),
 GNUNET_MQ_handler_end ());