transport.c

GNUnet是一个安全的点对点网络框架

      (ntohs (hello->header.type) != GNUNET_P2P_PROTO_HELLO))
    return GNUNET_SYSERR;       /* invalid */
  prot = ntohs (hello->protocol);
  if ((prot >= tapis_count) || (tapis[prot] == NULL))
    return GNUNET_SYSERR;       /* not supported */
  return tapis[prot]->hello_verify (hello);
}

/**
 * Get the MTU for a given transport type.
 */
static int
transportGetMTU (unsigned short ttype)
{
  if ((ttype >= tapis_count) || (tapis[ttype] == NULL))
    return GNUNET_SYSERR;
  return tapis[ttype]->mtu;
}

/**
 * Create a hello advertisement for the given
 * transport type for this node.
 */
static GNUNET_MessageHello *
transportCreatehello (unsigned short ttype)
{
  GNUNET_TransportAPI *tapi;
  GNUNET_MessageHello *hello;

  GNUNET_mutex_lock (tapis_lock);
  if (ttype == GNUNET_TRANSPORT_PROTOCOL_NUMBER_ANY)
    {
      unsigned int *perm;

      perm = GNUNET_permute (GNUNET_RANDOM_QUALITY_WEAK, tapis_count);
      ttype = tapis_count - 1;
      while ((ttype < tapis_count) &&
             ((tapis[perm[ttype]] == NULL) ||
              (tapis[perm[ttype]]->hello == NULL)))
        ttype--;                /* unsigned, will wrap around! */
      if (ttype >= tapis_count)
        {
          GNUNET_free (perm);
          GNUNET_mutex_unlock (tapis_lock);
          return NULL;
        }
      ttype = perm[ttype];
      GNUNET_free (perm);
    }
  else
    {
      if ((ttype >= tapis_count) || (tapis[ttype] == NULL))
        {
          GNUNET_GE_LOG (ectx,
                         GNUNET_GE_DEBUG | GNUNET_GE_BULK | GNUNET_GE_USER,
                         _("No transport of type %d known.\n"), ttype);
          GNUNET_mutex_unlock (tapis_lock);
          return NULL;
        }
    }
  tapi = tapis[ttype];
  if (tapi->hello == NULL)
    {
      GNUNET_mutex_unlock (tapis_lock);
      return NULL;              /* send-only transport */
    }
  hello = GNUNET_malloc (GNUNET_sizeof_hello (tapi->hello));
  memcpy (hello, tapi->hello, GNUNET_sizeof_hello (tapi->hello));
  GNUNET_mutex_unlock (tapis_lock);
  return hello;
}

/**
 * Get a message consisting of (if possible) all addresses that this
 * node is currently advertising.  This method is used to send out
 * possible ways to contact this node when sending a (plaintext) PING
 * during node discovery. Note that if we have many transport
 * implementations, it may not be possible to advertise all of our
 * addresses in one message, thus the caller can bound the size of the
 * advertisements.
 *
 * @param maxLen the maximum size of the hello message collection in bytes
 * @param buff where to write the hello messages
 * @return the number of bytes written to buff, -1 on error
 */
static int
getAdvertisedhellos (unsigned int maxLen, char *buff)
{
  int i;
  int tcount;
  GNUNET_MessageHello **hellos;
  int used;

  GNUNET_mutex_lock (tapis_lock);
  tcount = 0;
  for (i = 0; i < tapis_count; i++)
    if (tapis[i] != NULL)
      tcount++;

  hellos = GNUNET_malloc (tcount * sizeof (GNUNET_MessageHello *));
  tcount = 0;
  for (i = 0; i < tapis_count; i++)
    {
      if (tapis[i] != NULL)
        {
          hellos[tcount] = transportCreatehello (i);
          if (NULL != hellos[tcount])
            tcount++;
        }
    }
  GNUNET_mutex_unlock (tapis_lock);
  if (tcount == 0)
    {
      GNUNET_GE_LOG (ectx,
                     GNUNET_GE_INFO | GNUNET_GE_USER | GNUNET_GE_REQUEST,
                     _("No transport succeeded in creating a hello!\n"));
      GNUNET_free (hellos);
      return GNUNET_SYSERR;
    }
  used = 0;
  while (tcount > 0)
    {
      i = GNUNET_random_u32 (GNUNET_RANDOM_QUALITY_WEAK, tcount);       /* select a hello at random */
      if ((unsigned int) GNUNET_sizeof_hello (hellos[i]) <= maxLen - used)
        {
          memcpy (&buff[used], hellos[i], GNUNET_sizeof_hello (hellos[i]));
          used += GNUNET_sizeof_hello (hellos[i]);
        }
      GNUNET_free (hellos[i]);
      hellos[i] = hellos[--tcount];
    }
  for (i = 0; i < tcount; i++)
    GNUNET_free (hellos[i]);
  GNUNET_free (hellos);
  if (used == 0)
    GNUNET_GE_LOG (ectx,
                   GNUNET_GE_DEBUG | GNUNET_GE_DEVELOPER | GNUNET_GE_REQUEST,
                   "No HELLOs fit in %u bytes.\n", maxLen);
  return used;
}

static void
initHello (void *cls)
{
  GNUNET_TransportAPI *tapi = cls;
  GNUNET_MessageHello *hello;

  createSignedhello (tapi);
  hello = transportCreatehello (tapi->protocol_number);
  if (NULL != hello)
    {
      identity->addHost (hello);
      GNUNET_free (hello);
    }
}


static void
doneHelper (GNUNET_TransportAPI * tapi, void *unused)
{
  /* In the (rare) case that we shutdown transports
     before the cron-jobs had a chance to run, stop
     the cron-jobs */
  GNUNET_cron_del_job (coreAPI->cron, &initHello, 0, tapi);
}

static void
unloadTransport (int i)
{
  void (*ptr) ();

  doneHelper (tapis[i], NULL);
  GNUNET_cron_del_job (coreAPI->cron,
                       &createSignedhello, HELLO_RECREATE_FREQ, tapis[i]);
  ptr = GNUNET_plugin_resolve_function (tapis[i]->library_handle,
                                        "donetransport_", GNUNET_NO);
  if (ptr != NULL)
    ptr ();
  GNUNET_free (tapis[i]->transport_name);
  GNUNET_free_non_null (tapis[i]->hello);
  tapis[i]->hello = NULL;
  GNUNET_plugin_unload (tapis[i]->library_handle);
  tapis[i] = NULL;
}


/**
 * Actually start the transport services and begin
 * receiving messages.
 */
static void
startTransports (GNUNET_TransportPacketProcessor mpp)
{
  int i;

  ctapi.receive = mpp;
  for (i = 0; i < tapis_count; i++)
    if (tapis[i] != NULL)
      {
        if (GNUNET_OK != tapis[i]->server_start ())
          unloadTransport (i);
      }
}

/**
 * Stop the transport services, stop receiving messages.
 */
static void
stopTransports ()
{
  int i;

  for (i = 0; i < tapis_count; i++)
    if (tapis[i] != NULL)
      tapis[i]->server_stop ();
  ctapi.receive = NULL;
}

static void
initHelper (GNUNET_TransportAPI * tapi, void *unused)
{
  /* Creation of HELLOs takes longer if a locally
     unresolvable hostname ((Dyn)DNS) was specified
     as this host's address and we have no network
     connection at the moment.  Use of gethostbyname()
     blocks the startup process in this case.
     This is why we create the HELLOs in another
     thread. */
  GNUNET_cron_add_job (coreAPI->cron, &initHello, 0, 0, tapi);
}

/**
 * Test if the transport would even try to send
 * a message of the given size and importance
 * for the given session.<br>
 * This function is used to check if the core should
 * even bother to construct (and encrypt) this kind
 * of message.
 *
 * @return GNUNET_YES if the transport would try (i.e. queue
 *         the message or call the OS to send),
 *         GNUNET_NO if the transport would just drop the message,
 *         GNUNET_SYSERR if the size/session is invalid
 */
static int
testWouldTry (GNUNET_TSession * tsession, unsigned int size, int important)
{
  if (tsession == NULL)
    return GNUNET_SYSERR;
  if ((tsession->ttype >= tapis_count) || (tapis[tsession->ttype] == NULL))
    return GNUNET_SYSERR;
  return tapis[tsession->ttype]->send_now_test (tsession, size, important);
}

/**
 * Initialize the transport layer.
 */
GNUNET_Transport_ServiceAPI *
provide_module_transport (GNUNET_CoreAPIForPlugins * capi)
{
  static GNUNET_Transport_ServiceAPI ret;
  GNUNET_TransportAPI *tapi;
  GNUNET_TransportMainMethod tptr;
  char *dso;
  char *next;
  char *pos;
  struct GNUNET_PluginHandle *lib;
  GNUNET_EncName myself;

  ectx = capi->ectx;
  if (-1 == GNUNET_GC_get_configuration_value_number (capi->cfg,
                                                      "GNUNETD",
                                                      "HELLOEXPIRES",
                                                      1,
                                                      GNUNET_MAX_HELLO_EXPIRES
                                                      / 60, 60, &hello_live))
    return NULL;
  hello_live *= 60;

  GNUNET_GE_ASSERT (ectx, sizeof (GNUNET_MessageHello) == 600);
  identity = capi->service_request ("identity");
  if (identity == NULL)
    {
      GNUNET_GE_BREAK (ectx, 0);
      return NULL;
    }
  coreAPI = capi;
  ctapi.version = 1;
  ctapi.my_identity = coreAPI->my_identity;
  ctapi.ectx = coreAPI->ectx;
  ctapi.cfg = coreAPI->cfg;
  ctapi.load_monitor = coreAPI->load_monitor;
  ctapi.cron = coreAPI->cron;
  ctapi.receive = NULL;         /* initialized LATER! */
  ctapi.service_request = coreAPI->service_request;
  ctapi.service_release = coreAPI->service_release;
  ctapi.tsession_assert_unused = coreAPI->tsession_assert_unused;

  GNUNET_array_grow (tapis, tapis_count,
                     GNUNET_TRANSPORT_PROTOCOL_NUMBER_UDP + 1);

  tapis_lock = GNUNET_mutex_create (GNUNET_YES);
  lock = GNUNET_mutex_create (GNUNET_NO);

  /* now load transports */
  dso = NULL;
  GNUNET_GE_ASSERT (ectx,
                    -1 != GNUNET_GC_get_configuration_value_string (capi->cfg,
                                                                    "GNUNETD",
                                                                    "TRANSPORTS",
                                                                    "udp tcp nat",
                                                                    &dso));
  if (strlen (dso) != 0)
    {
      GNUNET_GE_LOG (ectx,
                     GNUNET_GE_INFO | GNUNET_GE_USER | GNUNET_GE_BULK,
                     _("Loading transports `%s'\n"), dso);
      next = dso;
      do
        {
          pos = next;
          while ((*next != '\0') && (*next != ' '))
            next++;
          if (*next == '\0')
            next = NULL;        /* terminate! */
          else
            {
              *next = '\0';     /* add 0-termination for pos */
              next++;
            }
          lib = GNUNET_plugin_load (ectx, "libgnunettransport_", pos);
          if (lib == NULL)
            {
              GNUNET_GE_LOG (ectx,
                             GNUNET_GE_ERROR | GNUNET_GE_USER |
                             GNUNET_GE_ADMIN | GNUNET_GE_IMMEDIATE,
                             _("Could not load transport plugin `%s'\n"),
                             pos);
              continue;
            }
          tptr =
            GNUNET_plugin_resolve_function (lib, "inittransport_",
                                            GNUNET_YES);
          if (tptr == NULL)
            {
              GNUNET_GE_LOG (ectx,
                             GNUNET_GE_ERROR | GNUNET_GE_ADMIN |
                             GNUNET_GE_USER | GNUNET_GE_DEVELOPER |
                             GNUNET_GE_IMMEDIATE,
                             _
                             ("Transport library `%s' did not provide required function '%s%s'.\n"),
                             pos, "inittransport_", pos);
              GNUNET_plugin_unload (lib);
              continue;
            }
          tapi = tptr (&ctapi);
          if (tapi == NULL)
            {
              GNUNET_plugin_unload (lib);
              continue;
            }
          tapi->library_handle = lib;
          tapi->transport_name = GNUNET_strdup (pos);
          if (GNUNET_OK != addTransport (tapi))
            {
              void (*ptr) ();

              GNUNET_free (tapi->transport_name);
              ptr =
                GNUNET_plugin_resolve_function (lib, "donetransport_",
                                                GNUNET_NO);
              if (ptr != NULL)
                ptr ();
              GNUNET_plugin_unload (lib);
            }
          else
            {
              GNUNET_GE_LOG (ectx,
                             GNUNET_GE_INFO | GNUNET_GE_USER | GNUNET_GE_BULK,
                             _("Loaded transport `%s'\n"), pos);
            }
        }
      while (next != NULL);
    }
  GNUNET_free (dso);

  IF_GELOG (ectx,
            GNUNET_GE_INFO | GNUNET_GE_REQUEST | GNUNET_GE_USER,
            GNUNET_hash_to_enc (&coreAPI->my_identity->hashPubKey, &myself));
  GNUNET_GE_LOG (ectx,
                 GNUNET_GE_INFO | GNUNET_GE_REQUEST | GNUNET_GE_USER,
                 _("I am peer `%s'.\n"), &myself);
  forEachTransport (&initHelper, NULL);

  ret.start = &startTransports;
  ret.stop = &stopTransports;
  ret.test_available = &isTransportAvailable;
  ret.add = &addTransport;
  ret.iterate_available = &forEachTransport;
  ret.connect = &transportConnect;
  ret.connect_freely = &transportConnectFreely;
  ret.associate = &transportAssociate;
  ret.cost_get = &transportGetCost;
  ret.send = &transportSend;
  ret.disconnect = &transportDisconnect;
  ret.hello_verify = &transportVerifyHello;
  ret.hello_to_address = &helloToAddress;
  ret.mtu_get = &transportGetMTU;
  ret.hello_create = &transportCreatehello;
  ret.hello_advertisements_get = &getAdvertisedhellos;
  ret.send_now_test = &testWouldTry;
  ret.assert_associated = &assertAssociated;

  return &ret;
}


/**
 * Shutdown the transport layer.
 */
int
release_module_transport ()
{
  int i;

  forEachTransport (&doneHelper, NULL);
  for (i = 0; i < tapis_count; i++)
    if (tapis[i] != NULL)
      unloadTransport (i);
  GNUNET_mutex_destroy (tapis_lock);
  GNUNET_mutex_destroy (lock);
  tapis_lock = NULL;
  GNUNET_array_grow (tapis, tapis_count, 0);

  coreAPI->service_release (identity);
  identity = NULL;
  coreAPI = NULL;
  return GNUNET_OK;
}


/* end of transport.c */