introclient: only give ascii key_s to storagebroker

This was triggered when the initial Introducer connection failed, so the
node read the introducer_cache.yaml from disk. That always returns
unicode strings, and the StorageFarmBroker insisted that it's
server-IDs (aka "key_s") were bytestrings.

The tests were extended to exercise the code that loads from disk and
delivers to the StorageFarmBroker, and more preconditions were put in
place to catch this sort of thing earlier next time.

closes ticket:2817

src/allmydata/introducer/client.py

@@ -11,6 +11,7 @@
 from allmydata.util import log, yamlutil
 from allmydata.util.rrefutil import add_version_to_remote_reference
 from allmydata.util.keyutil import BadSignatureError
+from allmydata.util.assertutil import precondition
 
 class InvalidCacheError(Exception):
     pass
@@ -103,8 +104,9 @@ def _load_announcements(self):
                 log.err(InvalidCacheError("not a dict: %r" % (server_params,)),
                         level=log.WEIRD)
                 continue
-            self._deliver_announcements(server_params['key_s'],
-                                        server_params['ann'])
+            # everything coming from yamlutil.safe_load is unicode
+            key_s = server_params['key_s'].encode("ascii")
+            self._deliver_announcements(key_s, server_params['ann'])
 
     def _save_announcements(self):
         announcements = []
@@ -157,6 +159,7 @@ def subscribe_to(self, service_name, cb, *args, **kwargs):
         self._subscribed_service_names.add(service_name)
         self._maybe_subscribe()
         for index,(ann,key_s,when) in self._inbound_announcements.items():
+            precondition(isinstance(key_s, str), key_s)
             servicename = index[0]
             if servicename == service_name:
                 eventually(cb, key_s, ann, *args, **kwargs)
@@ -232,6 +235,7 @@ def got_announcements(self, announcements, lp=None):
                 # this might raise UnknownKeyError or bad-sig error
                 ann, key_s = unsign_from_foolscap(ann_t)
                 # key is "v0-base32abc123"
+                precondition(isinstance(key_s, str), key_s)
             except BadSignatureError:
                 self.log("bad signature on inbound announcement: %s" % (ann_t,),
                          parent=lp, level=log.WEIRD, umid="ZAU15Q")
@@ -241,6 +245,7 @@ def got_announcements(self, announcements, lp=None):
             self._process_announcement(ann, key_s)
 
     def _process_announcement(self, ann, key_s):
+        precondition(isinstance(key_s, str), key_s)
         self._debug_counts["inbound_announcement"] += 1
         service_name = str(ann["service-name"])
         if service_name not in self._subscribed_service_names:
@@ -312,6 +317,7 @@ def _process_announcement(self, ann, key_s):
         self._deliver_announcements(key_s, ann)
 
     def _deliver_announcements(self, key_s, ann):
+        precondition(isinstance(key_s, str), key_s)
         service_name = str(ann["service-name"])
         for (service_name2,cb,args,kwargs) in self._local_subscribers:
             if service_name2 == service_name:

src/allmydata/test/test_introducer.py

@@ -749,6 +749,7 @@ def test_client_cache(self):
         ann_t = make_ann_t(ic, furl1, sk, 1)
 
         ic.got_announcements([ann_t])
+        yield flushEventualQueue()
 
         # check the cache for the announcement
         announcements = self._load_cache(cache_filepath)
@@ -763,6 +764,7 @@ def test_client_cache(self):
         furl2 = furl1 + "er"
         ann_t2 = make_ann_t(ic, furl2, sk, 2)
         ic.got_announcements([ann_t2])
+        yield flushEventualQueue()
         announcements = self._load_cache(cache_filepath)
         self.failUnlessEqual(len(announcements), 1)
         self.failUnlessEqual(announcements[0]['key_s'], pub1)
@@ -778,6 +780,7 @@ def test_client_cache(self):
         furl3 = "pb://onug64tu@127.0.0.1:456/short"
         ann_t3 = make_ann_t(ic, furl3, sk2, 1)
         ic.got_announcements([ann_t3])
+        yield flushEventualQueue()
 
         announcements = self._load_cache(cache_filepath)
         self.failUnlessEqual(len(announcements), 2)
@@ -788,6 +791,7 @@ def test_client_cache(self):
                                   for a in announcements]))
 
         # test loading
+        yield flushEventualQueue()
         ic2 = IntroducerClient(None, "introducer.furl", u"my_nickname",
                                "my_version", "oldest_version", {}, fakeseq,
                                ic._cache_filepath)
@@ -804,6 +808,12 @@ def got(key_s, ann):
         self.failUnlessEqual(announcements[pub2]["anonymous-storage-FURL"],
                              furl3)
 
+        c2 = TahoeClient(basedir)
+        c2.introducer_client._load_announcements()
+        yield flushEventualQueue()
+        self.assertEqual(c2.storage_broker.get_all_serverids(),
+                         frozenset([pub1, pub2]))
+
 class ClientSeqnums(unittest.TestCase):
     def test_client(self):
         basedir = "introducer/ClientSeqnums/test_client"