Finish test and implementation of directory service

.gitignore

@@ -17,8 +17,8 @@ nacl/nacl-20110221/build
 nacl/naclinc.txt
 nacl/nacllib.txt
 serval.c
-servald
-directory_service
+/servald
+/directory_service
 *.so
 test.*.log
 testlog

Makefile.in

@@ -72,6 +72,7 @@ MONITORCLIENTSRCS=conf.c \
 	strbuf_helpers.c
 
 MDPCLIENTSRCS=conf.c \
+	dataformats.c \
 	mkdir.c \
 	log.c \
 	mdp_client.c \

directory_client.c

@@ -17,6 +17,18 @@
 
 struct subscriber *directory_service;
 
+static void directory_update(struct sched_ent *alarm);
+
+static struct profile_total directory_timing={
+  .name="directory_update",
+};
+
+struct sched_ent directory_alarm={
+  .function=directory_update,
+  .stats=&directory_timing,
+};
+#define DIRECTORY_UPDATE_INTERVAL 300000
+
 // send a registration packet
 static void directory_send(struct subscriber *directory_service, const unsigned char *sid, const char *did, const char *name){
   overlay_mdp_frame request;
@@ -32,7 +44,9 @@ static void directory_send(struct subscriber *directory_service, const unsigned
   request.out.dst.port=MDP_PORT_DIRECTORY;
   request.out.payload_length = snprintf((char *)request.out.payload, sizeof(request.out.payload), 
 					"%s|%s", did, name);
-  
+  // Used by tests
+  INFOF("Sending directory registration for %s, %s, %s to %s", 
+	alloca_tohex(sid,7), did, name, alloca_tohex(directory_service->sid, 7));
   overlay_mdp_dispatch(&request, 0, NULL, 0);
 }
 
@@ -48,8 +62,8 @@ static void directory_send_keyring(struct subscriber *directory_service){
       
       for(k2=0; k2 < i->keypair_count; k2++){
 	if (i->keypairs[k2]->type==KEYTYPE_DID){
-	  const char *unpackedDid = (const char *) i->keypairs[kp]->private_key;
-	  const char *name = (const char *) i->keypairs[kp]->public_key;
+	  const char *unpackedDid = (const char *) i->keypairs[k2]->private_key;
+	  const char *name = (const char *) i->keypairs[k2]->public_key;
 	  
 	  directory_send(directory_service, packedSid, unpackedDid, name);
 	  // send the first DID only
@@ -81,15 +95,33 @@ static int load_directory_config(){
   return load_subscriber_address(directory_service);
 }
 
-int directory_interface_up(overlay_interface *interface){
-  // reload config, now that an interface is up
+static void directory_update(struct sched_ent *alarm){
   load_directory_config();
   
   if (directory_service){
-    if (subscriber_is_reachable(directory_service) != REACHABLE_NONE)
+    if (subscriber_is_reachable(directory_service) != REACHABLE_NONE){
       directory_send_keyring(directory_service);
-    else
+      
+      alarm->alarm = gettime_ms() + DIRECTORY_UPDATE_INTERVAL;
+      alarm->deadline = alarm->alarm + 10000;
+      schedule(alarm);
+    }else
       DEBUGF("Directory service is not reachable");
   }
+}
+
+int directory_service_init(){
+  directory_update(&directory_alarm);
   return 0;
 }
+
+// called when we discover a route to the directory service SID
+int directory_registration(){
+  // give the route & SAS keys a moment to propagate
+  unschedule(&directory_alarm);
+  directory_alarm.alarm = gettime_ms() + 200;
+  directory_alarm.deadline = directory_alarm.alarm + 10000;
+  schedule(&directory_alarm);
+  return 0;
+}
+

directory_service.c

@@ -56,20 +56,20 @@ static struct item *find_item(const char *key, int create){
 }
 
 static void store(char *key, char *value){
-  // used by tests
-  INFOF("PUBLISHED \"%s\" = \"%s\"", key, value);
-  
   struct item *item = find_item(key, 1);
   strncpy(item->value,value,sizeof(item->value));
   item->value[sizeof(item->value) -1]=0;
+  // expire after 20 minutes
+  item->expires = gettime_ms()+1200000;
+  // used by tests
+  fprintf(stderr, "PUBLISHED \"%s\" = \"%s\"\n", key, value);
 }
 
 static const char *retrieve(char *key){
-  INFOF("RESOLVING \"%s\"", key);
-  
   struct item *item = find_item(key, 0);
-  if (item)
+  if (item && item->expires > gettime_ms()){
     return item->value;
+  }
   return NULL;
 }
 
@@ -77,11 +77,11 @@ static void add_record(){
   int ttl;
   overlay_mdp_frame mdp;
   
-  if (!overlay_mdp_recv(&mdp, &ttl))
+  if (overlay_mdp_recv(&mdp, &ttl))
     return;
   
-  if (mdp.packetTypeAndFlags|=MDP_NOCRYPT){
-    WHY("Only encrypted packets will be considered for publishing");
+  if (mdp.packetTypeAndFlags&MDP_NOCRYPT){
+    fprintf(stderr, "Only encrypted packets will be considered for publishing\n");
     return;
   }
   
@@ -115,6 +115,8 @@ static void process_line(char *line){
   const char *response = retrieve(did);
   if (response)
     printf("%s|%s|\n",token,response);
+  printf("DONE\n");
+  fflush(stdout);
 }
 
 static void resolve_request(){
@@ -149,8 +151,10 @@ int main(int argc, char **argv){
 
   // bind for incoming directory updates
   unsigned char srcsid[SID_SIZE];
-  if (overlay_mdp_getmyaddr(0,srcsid)) return WHY("Could not get local address");
-  if (overlay_mdp_bind(srcsid,MDP_PORT_DIRECTORY)) return WHY("Could not bind to MDP socket");
+  if (overlay_mdp_getmyaddr(0,srcsid))
+    return WHY("Could not get local address");
+  if (overlay_mdp_bind(srcsid,MDP_PORT_DIRECTORY))
+    return WHY("Could not bind to MDP socket");
   
   set_nonblock(STDIN_FILENO);
   
@@ -159,8 +163,11 @@ int main(int argc, char **argv){
   fds[1].fd = mdp_client_socket;
   fds[1].events = POLLIN;
   
+  printf("STARTED\n");
+  fflush(stdout);
+  
   while(1){
-    int r = poll(fds, 2, 60000);
+    int r = poll(fds, 2, 100);
     if (r>0){
       if (fds[0].revents & POLLIN)
 	resolve_request();

keyring.c

@@ -696,7 +696,7 @@ int keyring_decrypt_pkr(keyring_file *k,keyring_context *c,
     if (id->keypairs[i]->type == KEYTYPE_CRYPTOBOX){
       struct subscriber *subscriber = find_subscriber(id->keypairs[i]->public_key, SID_SIZE, 1);
       if (subscriber){
-	subscriber->reachable=REACHABLE_SELF;
+	set_reachable(subscriber, REACHABLE_SELF);
 	if (!my_subscriber)
 	  my_subscriber=subscriber;
       }
@@ -909,7 +909,7 @@ keyring_identity *keyring_create_identity(keyring_file *k,keyring_context *c, co
   // add new identity to in memory table
   struct subscriber *subscriber = find_subscriber(id->keypairs[1]->public_key, SID_SIZE, 1);
   if (subscriber){
-    subscriber->reachable=REACHABLE_SELF;
+    set_reachable(subscriber, REACHABLE_SELF);
     if (!my_subscriber)
       my_subscriber=subscriber;
   }

net.c

@@ -22,7 +22,9 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 #include <fcntl.h>
 #include <sys/socket.h>
 #include <netinet/in.h>
+#include <time.h>
 #include "net.h"
+#include "serval.h"
 
 int _set_nonblock(int fd, struct __sourceloc where)
 {
@@ -221,3 +223,28 @@ int urandombytes(unsigned char *x, unsigned long long xlen)
   return 0;
 }
 
+time_ms_t gettime_ms()
+{
+  struct timeval nowtv;
+  // If gettimeofday() fails or returns an invalid value, all else is lost!
+  if (gettimeofday(&nowtv, NULL) == -1)
+    FATAL_perror("gettimeofday");
+  if (nowtv.tv_sec < 0 || nowtv.tv_usec < 0 || nowtv.tv_usec >= 1000000)
+    FATALF("gettimeofday returned tv_sec=%ld tv_usec=%ld", nowtv.tv_sec, nowtv.tv_usec);
+  return nowtv.tv_sec * 1000LL + nowtv.tv_usec / 1000;
+}
+
+// Returns sleep time remaining.
+time_ms_t sleep_ms(time_ms_t milliseconds)
+{
+  if (milliseconds <= 0)
+    return 0;
+  struct timespec delay;
+  struct timespec remain;
+  delay.tv_sec = milliseconds / 1000;
+  delay.tv_nsec = (milliseconds % 1000) * 1000000;
+  if (nanosleep(&delay, &remain) == -1 && errno != EINTR)
+    FATALF_perror("nanosleep(tv_sec=%ld, tv_nsec=%ld)", delay.tv_sec, delay.tv_nsec);
+  return remain.tv_sec * 1000 + remain.tv_nsec / 1000000;
+}
+

overlay.c

@@ -153,6 +153,9 @@ schedule(&_sched_##X); }
   // start the dna helper if configured
   dna_helper_start();
   
+  // preload directory service information
+  directory_service_init();
+  
   /* Pick next rhizome files to grab every few seconds
      from the priority list continuously being built from observed
      bundle announcements */

overlay_address.c

@@ -192,6 +192,45 @@ int subscriber_is_reachable(struct subscriber *subscriber){
   return subscriber->reachable;
 }
 
+int set_reachable(struct subscriber *subscriber, int reachable){
+  if (subscriber->reachable==reachable)
+    return 0;
+  int old_value=subscriber->reachable;
+  
+  subscriber->reachable=reachable;
+  
+  // these log messages may be used in tests
+  switch(reachable){
+    case REACHABLE_NONE:
+      DEBUGF("%s is not reachable", alloca_tohex_sid(subscriber->sid));
+      break;
+    case REACHABLE_SELF:
+      break;
+      
+    case REACHABLE_DIRECT:
+      DEBUGF("%s is now reachable directly", alloca_tohex_sid(subscriber->sid));
+      break;
+    case REACHABLE_INDIRECT:
+      DEBUGF("%s is now reachable indirectly", alloca_tohex_sid(subscriber->sid));
+      break;
+    case REACHABLE_UNICAST:
+      DEBUGF("%s is now reachable via unicast", alloca_tohex_sid(subscriber->sid));
+      break;
+    case REACHABLE_BROADCAST:
+      DEBUGF("%s is now reachable via broadcast", alloca_tohex_sid(subscriber->sid));
+      break;
+  }
+  
+  // Hacky layering violation...
+  if (subscriber==directory_service &&
+      (old_value==REACHABLE_NONE||old_value==REACHABLE_BROADCAST) &&
+      (reachable!=REACHABLE_NONE&&reachable!=REACHABLE_BROADCAST)
+      )
+    directory_registration();
+  
+  return 0;
+}
+
 // mark the subscriber as reachable via reply unicast packet
 int reachable_unicast(struct subscriber *subscriber, overlay_interface *interface, struct in_addr addr, int port){
   if (subscriber->reachable!=REACHABLE_NONE && subscriber->reachable!=REACHABLE_UNICAST)
@@ -200,12 +239,8 @@ int reachable_unicast(struct subscriber *subscriber, overlay_interface *interfac
   if (subscriber->node)
     return WHYF("Subscriber %s is already known for overlay routing", alloca_tohex_sid(subscriber->sid));
   
-  // may be used in tests
-  if (subscriber->reachable==REACHABLE_NONE)
-    DEBUGF("ADD DIRECT ROUTE TO %s via %s:%d", alloca_tohex_sid(subscriber->sid), inet_ntoa(addr), port);
-  
   subscriber->interface = interface;
-  subscriber->reachable = REACHABLE_UNICAST;
+  set_reachable(subscriber, REACHABLE_UNICAST);
   subscriber->address.sin_family = AF_INET;
   subscriber->address.sin_addr = addr;
   subscriber->address.sin_port = htons(port);
@@ -222,6 +257,7 @@ int load_subscriber_address(struct subscriber *subscriber){
   
   snprintf(buff, sizeof(buff), "%s.interface", sid_hex);
   const char *interface_name = confValueGet(buff, NULL);
+  // no unicast configuration? just return.
   if (!interface_name)
     return 1;
   

overlay_address.h

@@ -98,10 +98,12 @@ struct broadcast{
 };
 
 extern struct subscriber *my_subscriber;
+extern struct subscriber *directory_service;
 
 struct subscriber *find_subscriber(const unsigned char *sid, int len, int create);
 void enum_subscribers(struct subscriber *start, int(*callback)(struct subscriber *, void *), void *context);
 int subscriber_is_reachable(struct subscriber *subscriber);
+int set_reachable(struct subscriber *subscriber, int reachable);
 int reachable_unicast(struct subscriber *subscriber, overlay_interface *interface, struct in_addr addr, int port);
 int load_subscriber_address(struct subscriber *subscriber);
 

overlay_interface.c

@@ -419,9 +419,6 @@ overlay_interface_init_socket(int interface_index)
   if (my_subscriber)
     my_subscriber->send_full = 1;
   
-  // try to register ourselves with a directory service
-  directory_interface_up(interface);
-  
   return 0;
 }
 

overlay_route.c

@@ -184,7 +184,7 @@ overlay_node *get_node(struct subscriber *subscriber, int create){
     memset(subscriber->node,0,sizeof(overlay_node));
     subscriber->node->subscriber = subscriber;
     // if we're taking over routing calculations, make sure we invalidate any other calculations first
-    subscriber->reachable=REACHABLE_NONE;
+    set_reachable(subscriber, REACHABLE_NONE);
     // This info message is used by tests; don't alter or remove it.
     INFOF("ADD OVERLAY NODE sid=%s", alloca_tohex_sid(subscriber->sid));
   }
@@ -248,7 +248,7 @@ int overlay_route_ack_selfannounce(struct overlay_frame *f,
 
   /* Try to use broadcast if we don't have a route yet */
   if (out->destination->reachable == REACHABLE_NONE)
-    out->destination->reachable = REACHABLE_BROADCAST;
+    set_reachable(out->destination, REACHABLE_BROADCAST);
   
   /* Set the time in the ack. Use the last sequence number we have seen
      from this neighbour, as that may be helpful information for that neighbour
@@ -504,31 +504,14 @@ int overlay_route_recalc_node_metrics(overlay_node *n, time_ms_t now)
      if it goes down, we probably don't need to say anything at all.
   */
   
-  int diff=best_score-n->best_link_score;
+  int diff=best_score - n->best_link_score;
   if (diff>0) {
     overlay_route_please_advertise(n);
     if (debug&DEBUG_OVERLAYROUTEMONITOR) overlay_route_dump();
   }
+  int old_best = n->best_link_score;
   
   /* Remember new reachability information */
-  if (n->subscriber->reachable!=reachable){
-    switch (reachable){
-      case REACHABLE_DIRECT:
-	DEBUGF("%s is now reachable directly", alloca_tohex_sid(n->subscriber->sid));
-	break;
-      case REACHABLE_INDIRECT:
-	DEBUGF("%s is now reachable indirectly", alloca_tohex_sid(n->subscriber->sid));
-	break;
-      case REACHABLE_NONE:
-	DEBUGF("%s is not reachable", alloca_tohex_sid(n->subscriber->sid));
-	break;
-      case REACHABLE_BROADCAST:
-	DEBUGF("%s is now reachable via broadcast", alloca_tohex_sid(n->subscriber->sid));
-	break;
-    }
-  }
-    
-  n->subscriber->reachable=reachable;
   switch (reachable){
     case REACHABLE_INDIRECT:
       n->subscriber->next_hop = next_hop;
@@ -538,11 +521,14 @@ int overlay_route_recalc_node_metrics(overlay_node *n, time_ms_t now)
       n->subscriber->address = interface->broadcast_address;
       break;
   }
+  n->best_link_score=best_score;
+  n->best_observation=best_observation;
+  set_reachable(n->subscriber, reachable);
   
-  if (n->best_link_score && !best_score){
+  if (old_best && !best_score){
     INFOF("PEER UNREACHABLE, sid=%s", alloca_tohex_sid(n->subscriber->sid));
     monitor_announce_unreachable_peer(n->subscriber->sid);
-  }else if(best_score && !n->best_link_score){
+  }else if(best_score && !old_best){
     INFOF("PEER REACHABLE, sid=%s", alloca_tohex_sid(n->subscriber->sid));
     /* Make sure node is advertised soon */
     overlay_route_please_advertise(n);
@@ -552,9 +538,6 @@ int overlay_route_recalc_node_metrics(overlay_node *n, time_ms_t now)
     keyring_find_sas_public(keyring, n->subscriber->sid);
   }
   
-  n->best_link_score=best_score;
-  n->best_observation=best_observation;
-
   return 0;
 }
 

serval.h

@@ -865,7 +865,8 @@ int overlay_interface_register(char *name,
 overlay_interface * overlay_interface_find(struct in_addr addr);
 overlay_interface * overlay_interface_find_name(const char *name);
 
-int directory_interface_up(overlay_interface *interface);
+int directory_registration();
+int directory_service_init();
 
 #ifdef HAVE_VOIPTEST
 int app_pa_phone(int argc, const char *const *argv, struct command_line_option *o);

server.c

@@ -18,7 +18,6 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 */
 
 #include <signal.h>
-#include <time.h>
 #include <unistd.h>
 #include <sys/socket.h>
 #include <netinet/in.h>
@@ -52,31 +51,6 @@ void crash_handler(int signal);
 int getKeyring(char *s);
 int createServerSocket();
 
-time_ms_t gettime_ms()
-{
-  struct timeval nowtv;
-  // If gettimeofday() fails or returns an invalid value, all else is lost!
-  if (gettimeofday(&nowtv, NULL) == -1)
-    FATAL_perror("gettimeofday");
-  if (nowtv.tv_sec < 0 || nowtv.tv_usec < 0 || nowtv.tv_usec >= 1000000)
-    FATALF("gettimeofday returned tv_sec=%ld tv_usec=%ld", nowtv.tv_sec, nowtv.tv_usec);
-  return nowtv.tv_sec * 1000LL + nowtv.tv_usec / 1000;
-}
-
-// Returns sleep time remaining.
-time_ms_t sleep_ms(time_ms_t milliseconds)
-{
-  if (milliseconds <= 0)
-    return 0;
-  struct timespec delay;
-  struct timespec remain;
-  delay.tv_sec = milliseconds / 1000;
-  delay.tv_nsec = (milliseconds % 1000) * 1000000;
-  if (nanosleep(&delay, &remain) == -1 && errno != EINTR)
-    FATALF_perror("nanosleep(tv_sec=%ld, tv_nsec=%ld)", delay.tv_sec, delay.tv_nsec);
-  return remain.tv_sec * 1000 + remain.tv_nsec / 1000000;
-}
-
 /** Return the PID of the currently running server process, return 0 if there is none.
  */
 int server_pid()

tests/directory_service

@@ -0,0 +1,72 @@
+#!/bin/bash
+
+# Tests for Directory Services.
+# Copyright 2012 Serval Project
+#
+# This program 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 2
+# of the License, or (at your option) any later version.
+#
+# This program 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 this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+
+source "${0%/*}/../testframework.sh"
+source "${0%/*}/../testdefs.sh"
+
+configure_servald_server() {
+   executeOk_servald config set log.show_pid on
+   executeOk_servald config set log.show_time on
+   executeOk_servald config set debug.mdprequests Yes
+}
+
+setup() {
+   setup_servald
+   assert_no_servald_processes
+   foreach_instance +A +B +C create_single_identity
+   set_instance +A
+   executeOk_servald config set dna.helper.executable "$servald_build_root/directory_service"
+   executeOk_servald config set debug.dnahelper on
+   foreach_instance +B +C executeOk_servald config set directory.service $SIDA
+   start_servald_instances +A +B +C
+}
+
+teardown() {
+   stop_all_servald_servers
+   kill_all_servald_processes
+   assert_no_servald_processes
+}
+
+is_published() {
+   grep "PUBLISHED" $LOGA || return 1
+   return 0   
+}
+
+sent_directory_request() {
+   grep "Sending directory registration" $LOGB || return 1
+   return 0
+}
+
+doc_publish="Publish and retrieve a directory entry"
+test_publish() {
+   wait_until sent_directory_request
+   wait_until is_published
+   stop_servald_server +B
+   stop_servald_server +C
+   set_instance +A
+   executeOk_servald dna lookup "$DIDB"
+   assertStdoutLineCount '==' 1
+   assertStdoutGrep --matches=1 "^sid://$SIDB/$DIDB:$DIDB:$NAMEB\$"
+   executeOk_servald dna lookup "$DIDC"
+   assertStdoutLineCount '==' 1
+   assertStdoutGrep --matches=1 "^sid://$SIDC/$DIDC:$DIDC:$NAMEC\$"
+   return
+}
+
+runTests "$@"