Merge remote branch 'origin/development' into secured-meshms

commandline.c

@@ -1562,6 +1562,9 @@ int app_rhizome_clean(const struct cli_parsed *parsed, void *context)
 {
   if (config.debug.verbose)
     DEBUG_cli_parsed(parsed);
+  int verify = cli_arg(parsed, "verify", NULL, NULL, NULL) == 0;
+  if (verify)
+    verify_bundles();
   struct rhizome_cleanup_report report;
   if (rhizome_cleanup(&report) == -1)
     return -1;
@@ -2353,7 +2356,7 @@ struct cli_schema command_line_options[]={
 	"Deliver all new content to the specified Rhizome Direct server. Return when done."},
   {app_rhizome_direct_sync,{"rhizome","direct","pull","[<url>]",NULL}, 0,
 	"Fetch all new content from the specified Rhizome Direct server. Return when done."},
-  {app_rhizome_clean,{"rhizome","clean",NULL}, 0,
+  {app_rhizome_clean,{"rhizome","clean","[verify]",NULL}, 0,
 	"Remove stale and orphaned content from the Rhizome store"},
   {app_keyring_create,{"keyring","create",NULL}, 0,
    "Create a new keyring file."},

constants.h

@@ -97,7 +97,6 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 #define PAYLOAD_FLAG_ONE_HOP (1<<2)
 #define PAYLOAD_FLAG_CIPHERED (1<<4)
 #define PAYLOAD_FLAG_SIGNED (1<<5)
-#define PAYLOAD_FLAG_DUPLICATE (1<<6)
 
 /* Time-to-live is a 'uint5_t'.
  */

keyring.c

@@ -523,7 +523,8 @@ const struct keytype keytypes[] = {
       .dumper = dump_raw_hex
     },
   [KEYTYPE_RHIZOME] = {
-      /* Only the private key (Rhizome Secret) is stored, because the public key is never used.
+      /* The Rhizome Secret (a large, unguessable number) is stored in the private key field, and
+       * the public key field is not used.
        */
       .private_key_size = 32,
       .public_key_size = 0,

log.c

@@ -389,29 +389,39 @@ static int _log_iterator_next(_log_iterator *it, int level)
   return 0;
 }
 
-static void _log_iterator_printf_nl(_log_iterator *it, int level, struct __sourceloc whence, const char *fmt, ...)
+static void _log_iterator_vprintf_nl(_log_iterator *it, int level, struct __sourceloc whence, const char *fmt, va_list ap)
 {
-  va_list ap;
   _log_iterator_rewind(it);
   while (_log_iterator_next(it, level)) {
     _log_prefix_whence(it, whence);
-    va_start(ap, fmt);
+    va_list ap1;
-    vxprintf(it->xpf, fmt, ap);
+    va_copy(ap1, ap);
-    va_end(ap);
+    vxprintf(it->xpf, fmt, ap1);
+    va_end(ap1);
   }
 }
 
+static void _log_iterator_printf_nl(_log_iterator *it, int level, struct __sourceloc whence, const char *fmt, ...)
+{
+  va_list ap;
+  va_start(ap, fmt);
+  _log_iterator_vprintf_nl(it, level, whence, fmt, ap);
+  va_end(ap);
+}
+
+static void _logs_vprintf_nl(int level, struct __sourceloc whence, const char *fmt, va_list ap)
+{
+  _log_iterator it;
+  _log_iterator_start(&it);
+  _log_iterator_vprintf_nl(&it, level, whence, fmt, ap);
+}
+
 static void _logs_printf_nl(int level, struct __sourceloc whence, const char *fmt, ...)
 {
   va_list ap;
-  _log_iterator it;
-  _log_iterator_start(&it);
-  while (_log_iterator_next(&it, level)) {
-    _log_prefix_whence(&it, whence);
   va_start(ap, fmt);
-    vxprintf(it.xpf, fmt, ap);
+  _logs_vprintf_nl(level, whence, fmt, ap);
   va_end(ap);
-  }
 }
 
 const char *log_file_directory_path()
@@ -674,16 +684,26 @@ void logString(int level, struct __sourceloc whence, const char *str)
 
 void logMessage(int level, struct __sourceloc whence, const char *fmt, ...)
 {
+  if (level != LOG_LEVEL_SILENT) {
     va_list ap;
+    va_start(ap, fmt);
+    vlogMessage(level, whence, fmt, ap);
+    va_end(ap);
+  }
+}
+
+void vlogMessage(int level, struct __sourceloc whence, const char *fmt, va_list ap)
+{
   if (level != LOG_LEVEL_SILENT) {
     _log_iterator it;
     _log_iterator_start(&it);
     _rotate_log_file(&it);
     while (_log_iterator_next(&it, level)) {
       _log_prefix_whence(&it, whence);
-      va_start(ap, fmt);
+      va_list ap1;
-      vxprintf(it.xpf, fmt, ap);
+      va_copy(ap1, ap);
-      va_end(ap);
+      vxprintf(it.xpf, fmt, ap1);
+      va_end(ap1);
     }
   }
 }

overlay_interface.c

@@ -251,7 +251,7 @@ overlay_interface_read_any(struct sched_ent *alarm){
 	     inet_ntoa(src),
 	     interface->name);
     
-    if (packetOkOverlay(interface, packet, plen, recvttl, &src_addr, addrlen)) {
+    if (packetOkOverlay(interface, packet, plen, recvttl, &src_addr, addrlen)<0) {
       if (config.debug.rejecteddata) {
 	WHYF("Malformed packet (length = %d)",plen);
 	dump("the malformed packet",packet,plen);
@@ -562,7 +562,7 @@ static void interface_read_dgram(struct overlay_interface *interface){
 	   inet_ntoa(src),
 	   interface->name);
   }
-  if (packetOkOverlay(interface, packet, plen, recvttl, &src_addr, addrlen)) {
+  if (packetOkOverlay(interface, packet, plen, recvttl, &src_addr, addrlen)<0) {
     if (config.debug.rejecteddata) {
       WHYF("Malformed packet (length = %d)",plen);
       dump("the malformed packet",packet,plen);
@@ -703,7 +703,7 @@ static void interface_read_stream(struct overlay_interface *interface){
   while (state->src_offset < state->src_size) {
     int ret = slip_decode(state);
     if (ret==1){
-      if (packetOkOverlay(interface, state->dst, state->packet_length, -1, NULL, -1)) {
+      if (packetOkOverlay(interface, state->dst, state->packet_length, -1, NULL, -1)<0) {
 	if (config.debug.rejecteddata) {
 	  WHYF("Malformed packet (length = %d)",state->packet_length);
 	  dump("the malformed packet",state->dst,state->packet_length);

overlay_link.c

@@ -255,7 +255,11 @@ overlay_mdp_service_probe(overlay_mdp_frame *mdp)
   peer->address.sin_family = AF_INET;
   peer->address.sin_addr = probe.addr.sin_addr;
   peer->address.sin_port = probe.addr.sin_port;
-  set_reachable(peer, REACHABLE_UNICAST | (peer->reachable & REACHABLE_DIRECT));
+  int r=REACHABLE_UNICAST;
+  // Don't turn assumed|broadcast into unicast|broadcast
+  if (!(peer->reachable & REACHABLE_ASSUMED))
+    r |= (peer->reachable & REACHABLE_DIRECT);
+  set_reachable(peer, r);
   RETURN(0);
   OUT();
 }

overlay_mdp.c

@@ -835,7 +835,7 @@ static int routing_table(struct subscriber *subscriber, void *context){
   reply.packetTypeAndFlags=MDP_TX;
   reply.out.payload_length=sizeof(struct overlay_route_record);
   memcpy(r->sid, subscriber->sid, SID_SIZE);
-  r->reachable = subscriber_is_reachable(subscriber);
+  r->reachable = subscriber->reachable;
   
   if (subscriber->reachable==REACHABLE_INDIRECT && subscriber->next_hop)
     memcpy(r->neighbour, subscriber->next_hop->sid, SID_SIZE);

overlay_packetformats.c

@@ -227,7 +227,7 @@ int parseMdpPacketHeader(struct decode_context *context, struct overlay_frame *f
   }else
     frame->type=OF_TYPE_DATA;
 
-  if (context->packet_version >0){
+  if (context->packet_version >= 1){
     int seq = ob_get(buffer);
     if (seq == -1)
       RETURN(WHY("Unable to read packet seq"));

overlay_payload.c

@@ -73,7 +73,7 @@ static int overlay_frame_build_header(int packet_version, struct decode_context
     if (ob_append_byte(buff, type)) return -1;
   }
 
-  if (packet_version>0)
+  if (packet_version >= 1)
     if (ob_append_byte(buff, sequence))
       return -1;
   

overlay_queue.c

@@ -494,7 +494,7 @@ overlay_stuff_packet(struct outgoing_packet *packet, overlay_txqueue *queue, tim
     }else{
       // is this packet going our way?
       if (frame->interface!=packet->interface ||
-	  frame->packet_version==packet->packet_version ||
+	  frame->packet_version!=packet->packet_version ||
 	  memcmp(&packet->dest, &frame->recvaddr, sizeof(packet->dest))!=0){
 	goto skip;
       }

rhizome.h

@@ -211,6 +211,7 @@ extern sqlite3 *rhizome_db;
 
 int rhizome_opendb();
 int rhizome_close_db();
+void verify_bundles();
 
 struct rhizome_cleanup_report {
     int deleted_stale_incoming_files;

rhizome_bundle.c

@@ -108,7 +108,7 @@ int rhizome_read_manifest_file(rhizome_manifest *m, const char *filename, int bu
   }
 
   m->manifest_all_bytes=m->manifest_bytes;
-
+  m->var_count=0;
   /* Parse out variables, signature etc */
   int have_service = 0;
   int have_id = 0;

rhizome_database.c

@@ -165,7 +165,7 @@ void sqlite_log(void *ignored, int result, const char *msg){
   WARNF("Sqlite: %d %s", result, msg);
 }
 
-static void verify_bundles(){
+void verify_bundles(){
   // assume that only the manifest itself can be trusted
   // fetch all manifests and reinsert them.
   sqlite_retry_state retry = SQLITE_RETRY_STATE_DEFAULT;
@@ -192,7 +192,7 @@ static void verify_bundles(){
     }
     if (ret!=0){
       DEBUGF("Removing invalid manifest entry @%lld", rowid);
-      //sqlite_exec_void_retry(&retry, "DELETE FROM MANIFESTS WHERE ROWID=%lld;", rowid);
+      sqlite_exec_void_retry(&retry, "DELETE FROM MANIFESTS WHERE ROWID=%lld;", rowid);
     }
     rhizome_manifest_free(m);
   }
@@ -444,17 +444,18 @@ sqlite3_stmt *_sqlite_prepare(struct __sourceloc __whence, sqlite_retry_state *r
 
 sqlite3_stmt *_sqlite_prepare_loglevel(struct __sourceloc __whence, int log_level, sqlite_retry_state *retry, strbuf stmt)
 {
+  IN();
   sqlite3_stmt *statement = NULL;
   if (strbuf_overrun(stmt)) {
     WHYF("SQL overrun: %s", strbuf_str(stmt));
-    return NULL;
+    RETURN(NULL);
   }
   if (!rhizome_db && rhizome_opendb() == -1)
-    return NULL;
+    RETURN(NULL);
   while (1) {
     switch (sqlite3_prepare_v2(rhizome_db, strbuf_str(stmt), -1, &statement, NULL)) {
       case SQLITE_OK:
-	return statement;
+	RETURN(statement);
       case SQLITE_BUSY:
       case SQLITE_LOCKED:
 	if (retry && _sqlite_retry(__whence, retry, strbuf_str(stmt))) {
@@ -464,13 +465,14 @@ sqlite3_stmt *_sqlite_prepare_loglevel(struct __sourceloc __whence, int log_leve
       default:
 	LOGF(log_level, "query invalid, %s: %s", sqlite3_errmsg(rhizome_db), strbuf_str(stmt));
 	sqlite3_finalize(statement);
-	return NULL;
+	RETURN(NULL);
     }
   }
 }
 
 int _sqlite_step_retry(struct __sourceloc __whence, int log_level, sqlite_retry_state *retry, sqlite3_stmt *statement)
 {
+  IN();
   int ret = -1;
   sqlite_trace_whence = &__whence;
   while (statement) {
@@ -499,6 +501,7 @@ int _sqlite_step_retry(struct __sourceloc __whence, int log_level, sqlite_retry_
     }
   }
   sqlite_trace_whence = NULL;
+  OUT();
   return ret;
 }
 
@@ -1773,48 +1776,14 @@ int rhizome_delete_file(const char *fileid)
   return rhizome_delete_file_retry(&retry, fileid);
 }
 
-time_ms_t lookup_time=0;
-time_ms_t last_bar_lookup=0;
 int rhizome_is_bar_interesting(unsigned char *bar){
   IN();
-
-  if (last_bar_lookup>(gettime_ms()-1000)) {
-    /* Looking up rhizome bundles by BAR can be slow.
-       If so, then ignore some percentage of BARs. */
-    if (lookup_time>10) {
-      // if >10ms then only check 1 in 8 BARs
-      if (random()&0x7) RETURN(0);    
-    }
-    if (lookup_time>100) {
-      // if >10ms then only check 1 in 32 (1 in 8 of 1 in 4) BARs
-      if (random()&0x3) RETURN(0);   
-    }
-  }
-
-  time_ms_t start_time=gettime_ms();
-
-  int64_t version = rhizome_bar_version(bar);
-  unsigned char log2_size = bar[RHIZOME_BAR_FILESIZE_OFFSET];
   int ret=1;
+  int64_t version = rhizome_bar_version(bar);
   char id_hex[RHIZOME_MANIFEST_ID_STRLEN];
   tohex(id_hex, &bar[RHIZOME_BAR_PREFIX_OFFSET], RHIZOME_BAR_PREFIX_BYTES);
   strcat(id_hex, "%");
 
-  // are we ignoring this manifest?
-  if (rhizome_ignore_manifest_check(&bar[RHIZOME_BAR_PREFIX_OFFSET], RHIZOME_BAR_PREFIX_BYTES)){
-    DEBUGF("Ignoring %s", id_hex);
-    RETURN(0);
-  }
-
-  // do we have free space in a fetch queue?
-  if (log2_size!=0xFF && rhizome_fetch_has_queue_space(log2_size)!=1)
-    RETURN(0);
-
-  // are we already fetching this bundle [or later]?
-  rhizome_manifest *m=rhizome_fetch_search(&bar[RHIZOME_BAR_PREFIX_OFFSET], RHIZOME_BAR_PREFIX_BYTES);
-  if (m && m->version >= version)
-    RETURN(0);
-  
   // do we have this bundle [or later]?
   sqlite_retry_state retry = SQLITE_RETRY_STATE_DEFAULT;
   sqlite3_stmt *statement = sqlite_prepare(&retry, 
@@ -1833,11 +1802,6 @@ int rhizome_is_bar_interesting(unsigned char *bar){
   }  
   sqlite3_finalize(statement);
 
-  time_ms_t end_time=gettime_ms();
-  lookup_time=end_time-start_time;
-  last_bar_lookup=end_time;
-  if (lookup_time>50) WARNF("Looking up a BAR took %lldms",lookup_time);
-
   RETURN(ret);
   OUT();
 }

rhizome_packetformats.c

@@ -254,6 +254,8 @@ error:
   return -1;
 }
 
+time_ms_t lookup_time=0;
+
 int overlay_rhizome_saw_advertisements(int i, struct overlay_frame *f, long long now)
 {
   IN();
@@ -388,13 +390,50 @@ int overlay_rhizome_saw_advertisements(int i, struct overlay_frame *f, long long
   mdp.out.payload_length=0;
   
   // parse BAR's
-  while(ob_remaining(f->payload)>0){
+  unsigned char *bars[50];
-    unsigned char *bar=ob_get_bytes_ptr(f->payload, RHIZOME_BAR_BYTES);
+  int bar_count=0;
+  while(ob_remaining(f->payload)>0 && bar_count<50){
+    unsigned char *bar;
+    bars[bar_count]=bar=ob_get_bytes_ptr(f->payload, RHIZOME_BAR_BYTES);
     if (!bar){
       WARNF("Expected whole BAR @%x (only %d bytes remain)", ob_position(f->payload), ob_remaining(f->payload));
       break;
     }
-    if (rhizome_is_bar_interesting(bar)==1){
+
+    // are we ignoring this manifest?
+    if (rhizome_ignore_manifest_check(&bar[RHIZOME_BAR_PREFIX_OFFSET], RHIZOME_BAR_PREFIX_BYTES))
+      continue;
+
+    // do we have free space in a fetch queue?
+    unsigned char log2_size = bar[RHIZOME_BAR_FILESIZE_OFFSET];
+    if (log2_size!=0xFF && rhizome_fetch_has_queue_space(log2_size)!=1)
+      continue;
+
+    int64_t version = rhizome_bar_version(bar);
+    // are we already fetching this bundle [or later]?
+    rhizome_manifest *m=rhizome_fetch_search(&bar[RHIZOME_BAR_PREFIX_OFFSET], RHIZOME_BAR_PREFIX_BYTES);
+    if (m && m->version >= version)
+      continue;
+
+    bar_count++;
+  }
+
+  // perform costly database lookups
+  int index;
+  int test_count=0;
+  int max_tests = (lookup_time?(int)(40 / lookup_time):bar_count);
+  if (max_tests<=0)
+    max_tests=2;
+
+  time_ms_t start_time = gettime_ms();
+
+  for (index=0;index<bar_count;index++){
+    if (test_count > max_tests || gettime_ms() - start_time >40)
+      break;
+    if (bar_count > max_tests && random()%bar_count >= max_tests)
+      continue;
+    test_count++;
+    if (rhizome_is_bar_interesting(bars[index])==1){
       // add a request for the manifest
       if (mdp.out.payload_length==0){
 	bcopy(my_subscriber->sid,mdp.out.src.sid,SID_SIZE);
@@ -409,12 +448,19 @@ int overlay_rhizome_saw_advertisements(int i, struct overlay_frame *f, long long
 	
 	mdp.out.queue=OQ_ORDINARY;
       }
-      DEBUGF("Requesting manifest for BAR %s", alloca_tohex(bar, RHIZOME_BAR_BYTES));
+      DEBUGF("Requesting manifest for BAR %s", alloca_tohex(bars[index], RHIZOME_BAR_BYTES));
-      bcopy(bar, &mdp.out.payload[mdp.out.payload_length], RHIZOME_BAR_BYTES);
+      bcopy(bars[index], &mdp.out.payload[mdp.out.payload_length], RHIZOME_BAR_BYTES);
       mdp.out.payload_length+=RHIZOME_BAR_BYTES;
     }
   }
   
+  time_ms_t end_time=gettime_ms();
+
+  if (test_count)
+    lookup_time=(end_time-start_time)/test_count;
+  else
+    lookup_time = (end_time - start_time);
+
   if (mdp.out.payload_length>0)
     overlay_mdp_dispatch(&mdp,0 /* system generated */,NULL,0);
 
@@ -422,3 +468,4 @@ int overlay_rhizome_saw_advertisements(int i, struct overlay_frame *f, long long
   RETURN(0);
   OUT();
 }
+

route_link.c

@@ -274,15 +274,16 @@ static void update_path_score(struct neighbour *neighbour, struct link *link){
 
 static int find_best_link(struct subscriber *subscriber)
 {
+  IN();
   if (subscriber->reachable==REACHABLE_SELF)
-    return 0;
+    RETURN(0);
 
   struct link_state *state = get_link_state(subscriber);
   if (state->route_version == route_version)
-    return 0;
+    RETURN(0);
 
   if (state->calculating)
-    return -1;
+    RETURN(-1);
   state->calculating = 1;
 
   struct neighbour *neighbour = neighbours;
@@ -341,7 +342,7 @@ next:
 
   int reachable = subscriber->reachable;
   if (next_hop == NULL){
-    if (subscriber->reachable&REACHABLE_BROADCAST && !(subscriber->reachable & REACHABLE_ASSUMED))
+    if ((subscriber->reachable&REACHABLE_DIRECT) != REACHABLE_UNICAST)
       reachable = REACHABLE_NONE;
   } else if (next_hop == subscriber){
     // reset the state of any unicast probe's if the interface has changed
@@ -350,7 +351,7 @@ next:
       subscriber->last_probe=0;
       bzero(&subscriber->address, sizeof subscriber->address);
     }
-    reachable = REACHABLE_BROADCAST | (subscriber->reachable & REACHABLE_UNICAST);
+    reachable = REACHABLE_BROADCAST | (reachable & REACHABLE_UNICAST);
     next_hop = NULL;
     subscriber->interface = interface;
   } else {
@@ -377,7 +378,7 @@ next:
     state->next_update = now;
   }
 
-  return 0;
+  RETURN(0);
 }
 
 static int monitor_announce(struct subscriber *subscriber, void *context){
@@ -505,7 +506,6 @@ static void free_neighbour(struct neighbour **neighbour_ptr){
   n->root=NULL;
   *neighbour_ptr = n->_next;
   free(n);
-  route_version++;
 }
 
 static void clean_neighbours(time_ms_t now)
@@ -527,6 +527,10 @@ static void clean_neighbours(time_ms_t now)
         list = &link->_next;
       }
     }
+    // when all links to a neighbour that we are routing through expire, force a routing calculation update
+    struct link_state *state = get_link_state(n->subscriber);
+    if (state->next_hop == n->subscriber && (n->neighbour_link_timeout < now || !n->links) && state->route_version == route_version)
+      route_version++;
     if (!n->links){
       free_neighbour(n_ptr);
     }else{
@@ -698,7 +702,8 @@ static void link_send(struct sched_ent *alarm)
   if (neighbours){
     alarm->deadline = alarm->alarm;
     schedule(alarm);
-  }
+  }else
+    alarm->alarm=0;
 }
 
 static void update_alarm(time_ms_t limit){

tests/routing

@@ -358,6 +358,28 @@ test_offline() {
       wait_until --timeout=30 instance_offline +C
 }
 
+doc_lose_neighbours="Lose and regain neighbours"
+setup_lose_neighbours() {
+   setup_servald
+   assert_no_servald_processes
+   foreach_instance +A +B +C create_single_identity
+   foreach_instance +A +B add_interface 1
+   foreach_instance +B +C add_interface 2
+   foreach_instance +A +B +C start_routing_instance
+}
+test_lose_neighbours() {
+   wait_until path_exists +A +B +C
+   wait_until path_exists +C +B +A
+   stop_servald_server +B
+   foreach_instance +A +C \
+      wait_until --timeout=30 instance_offline +B
+   set_instance +A
+   wait_until --timeout=30 instance_offline +C
+   start_servald_server +B
+   wait_until path_exists +A +B +C
+   wait_until path_exists +C +B +A
+}
+
 setup_multi_interface() {
    setup_servald
    assert_no_servald_processes