src/allmydata/test/test_mutable.py

   1
   2 import itertools, struct
   3 from twisted.trial import unittest
   4 from twisted.internet import defer
   5 from twisted.python import failure, log
   6 from allmydata import mutable, uri, dirnode2
   7 from allmydata.dirnode2 import split_netstring
   8 from allmydata.util.hashutil import netstring, tagged_hash
   9 from allmydata.encode import NotEnoughPeersError
  10 from allmydata.interfaces import IURI, INewDirectoryURI, IDirnodeURI, \
  11      IMutableFileURI
  12
  13 import sha
  14 from allmydata.Crypto.Util.number import bytes_to_long
  15
  16 class Netstring(unittest.TestCase):
  17     def test_split(self):
  18         a = netstring("hello") + netstring("world")
  19         self.failUnlessEqual(split_netstring(a, 2), ("hello", "world"))
  20         self.failUnlessEqual(split_netstring(a, 2, False), ("hello", "world"))
  21         self.failUnlessEqual(split_netstring(a, 2, True),
  22                              ("hello", "world", ""))
  23         self.failUnlessRaises(ValueError, split_netstring, a+" extra", 2)
  24         self.failUnlessRaises(ValueError, split_netstring, a+" extra", 2, False)
  25
  26     def test_extra(self):
  27         a = netstring("hello")
  28         self.failUnlessEqual(split_netstring(a, 1, True), ("hello", ""))
  29         b = netstring("hello") + "extra stuff"
  30         self.failUnlessEqual(split_netstring(b, 1, True),
  31                              ("hello", "extra stuff"))
  32
  33     def test_nested(self):
  34         a = netstring("hello") + netstring("world") + "extra stuff"
  35         b = netstring("a") + netstring("is") + netstring(a) + netstring(".")
  36         top = split_netstring(b, 4)
  37         self.failUnlessEqual(len(top), 4)
  38         self.failUnlessEqual(top[0], "a")
  39         self.failUnlessEqual(top[1], "is")
  40         self.failUnlessEqual(top[2], a)
  41         self.failUnlessEqual(top[3], ".")
  42         self.failUnlessRaises(ValueError, split_netstring, a, 2)
  43         self.failUnlessRaises(ValueError, split_netstring, a, 2, False)
  44         bottom = split_netstring(a, 2, True)
  45         self.failUnlessEqual(bottom, ("hello", "world", "extra stuff"))
  46
  47 class FakeFilenode(mutable.MutableFileNode):
  48     counter = itertools.count(1)
  49     all_contents = {}
  50     all_rw_friends = {}
  51
  52     def create(self, initial_contents, wait_for_numpeers=None):
  53         d = mutable.MutableFileNode.create(self, initial_contents, wait_for_numpeers=None)
  54         def _then(res):
  55             self.all_contents[self.get_uri()] = initial_contents
  56             return res
  57         d.addCallback(_then)
  58         return d
  59     def init_from_uri(self, myuri):
  60         mutable.MutableFileNode.init_from_uri(self, myuri)
  61         return self
  62     def replace(self, newdata, wait_for_numpeers=None):
  63         self.all_contents[self.get_uri()] = initial_contents
  64         return defer.succeed(self)
  65     def _generate_pubprivkeys(self):
  66         count = self.counter.next()
  67         return FakePubKey(count), FakePrivKey(count)
  68     def _publish(self, initial_contents, wait_for_numpeers):
  69         self.all_contents[self.get_uri()] = initial_contents
  70         return defer.succeed(self)
  71
  72     def download_to_data(self):
  73         if self.is_readonly():
  74             assert self.all_rw_friends.has_key(self.get_uri()), (self.get_uri(), id(self.all_rw_friends))
  75             return defer.succeed(self.all_contents[self.all_rw_friends[self.get_uri()]])
  76         else:
  77             return defer.succeed(self.all_contents[self.get_uri()])
  78     def replace(self, newdata, wait_for_numpeers=None):
  79         self.all_contents[self.get_uri()] = newdata
  80         return defer.succeed(None)
  81
  82 class FakePublish(mutable.Publish):
  83     def _do_query(self, conn, peerid, peer_storage_servers, storage_index):
  84         assert conn[0] == peerid
  85         shares = self._peers[peerid]
  86         return defer.succeed(shares)
  87
  88     def _do_testreadwrite(self, peerid, peer_storage_servers, secrets,
  89                           tw_vectors, read_vector):
  90         # always-pass: parrot the test vectors back to them.
  91         readv = {}
  92         for shnum, (testv, datav, new_length) in tw_vectors.items():
  93             for (offset, length, op, specimen) in testv:
  94                 assert op in ("le", "eq", "ge")
  95             readv[shnum] = [ specimen
  96                              for (offset, length, op, specimen)
  97                              in testv ]
  98         answer = (True, readv)
  99         return defer.succeed(answer)
 100
 101
 102 class FakeNewDirectoryNode(dirnode2.NewDirectoryNode):
 103     filenode_class = FakeFilenode
 104
 105 class FakeIntroducerClient:
 106     def when_enough_peers(self, numpeers):
 107         return defer.succeed(None)
 108
 109 class FakeClient:
 110     def __init__(self, num_peers=10):
 111         self._num_peers = num_peers
 112         self._peerids = [tagged_hash("peerid", "%d" % i)[:20]
 113                          for i in range(self._num_peers)]
 114         self.introducer_client = FakeIntroducerClient()
 115     def log(self, msg):
 116         log.msg(msg)
 117
 118     def get_renewal_secret(self):
 119         return "I hereby permit you to renew my files"
 120     def get_cancel_secret(self):
 121         return "I hereby permit you to cancel my leases"
 122
 123     def create_empty_dirnode(self, wait_for_numpeers):
 124         n = FakeNewDirectoryNode(self)
 125         d = n.create(wait_for_numpeers=wait_for_numpeers)
 126         d.addCallback(lambda res: n)
 127         return d
 128
 129     def create_dirnode_from_uri(self, u):
 130         return FakeNewDirectoryNode(self).init_from_uri(u)
 131
 132     def create_mutable_file(self, contents="", wait_for_numpeers=None):
 133         n = FakeFilenode(self)
 134         d = n.create(contents, wait_for_numpeers=wait_for_numpeers)
 135         d.addCallback(lambda res: n)
 136         return d
 137
 138     def create_node_from_uri(self, u):
 139         u = IURI(u)
 140         if INewDirectoryURI.providedBy(u):
 141             return self.create_dirnode_from_uri(u)
 142         assert IMutableFileURI.providedBy(u)
 143         res = FakeFilenode(self).init_from_uri(u)
 144         return res
 145
 146     def get_permuted_peers(self, key, include_myself=True):
 147         """
 148         @return: list of (permuted-peerid, peerid, connection,)
 149         """
 150         peers_and_connections = [(pid, (pid,)) for pid in self._peerids]
 151         results = []
 152         for peerid, connection in peers_and_connections:
 153             assert isinstance(peerid, str)
 154             permuted = bytes_to_long(sha.new(key + peerid).digest())
 155             results.append((permuted, peerid, connection))
 156         results.sort()
 157         return results
 158
 159 class Filenode(unittest.TestCase):
 160     def setUp(self):
 161         self.client = FakeClient()
 162
 163     def test_create(self):
 164         d = self.client.create_mutable_file(wait_for_numpeers=1)
 165         def _created(n):
 166             d = n.replace("contents 1")
 167             d.addCallback(lambda res: self.failUnlessIdentical(res, None))
 168             d.addCallback(lambda res: n.download_to_data())
 169             d.addCallback(lambda res: self.failUnlessEqual(res, "contents 1"))
 170             d.addCallback(lambda res: n.replace("contents 2"))
 171             d.addCallback(lambda res: n.download_to_data())
 172             d.addCallback(lambda res: self.failUnlessEqual(res, "contents 2"))
 173             return d
 174         d.addCallback(_created)
 175         return d
 176
 177     def test_create_with_initial_contents(self):
 178         d = self.client.create_mutable_file("contents 1")
 179         def _created(n):
 180             d = n.download_to_data()
 181             d.addCallback(lambda res: self.failUnlessEqual(res, "contents 1"))
 182             d.addCallback(lambda res: n.replace("contents 2"))
 183             d.addCallback(lambda res: n.download_to_data())
 184             d.addCallback(lambda res: self.failUnlessEqual(res, "contents 2"))
 185             return d
 186         d.addCallback(_created)
 187         return d
 188
 189
 190 class Publish(unittest.TestCase):
 191     def test_encrypt(self):
 192         c = FakeClient()
 193         fn = FakeFilenode(c)
 194         # .create usually returns a Deferred, but we happen to know it's
 195         # synchronous
 196         CONTENTS = "some initial contents"
 197         fn.create(CONTENTS, wait_for_numpeers=1)
 198         p = mutable.Publish(fn)
 199         target_info = None
 200         d = defer.maybeDeferred(p._encrypt_and_encode, target_info,
 201                                 CONTENTS, "READKEY", "IV"*8, 3, 10)
 202         def _done( ((shares, share_ids),
 203                     required_shares, total_shares,
 204                     segsize, data_length, target_info2) ):
 205             self.failUnlessEqual(len(shares), 10)
 206             for sh in shares:
 207                 self.failUnless(isinstance(sh, str))
 208                 self.failUnlessEqual(len(sh), 7)
 209             self.failUnlessEqual(len(share_ids), 10)
 210             self.failUnlessEqual(required_shares, 3)
 211             self.failUnlessEqual(total_shares, 10)
 212             self.failUnlessEqual(segsize, 21)
 213             self.failUnlessEqual(data_length, len(CONTENTS))
 214             self.failUnlessIdentical(target_info, target_info2)
 215         d.addCallback(_done)
 216         return d
 217
 218     def test_generate(self):
 219         c = FakeClient()
 220         fn = FakeFilenode(c)
 221         # .create usually returns a Deferred, but we happen to know it's
 222         # synchronous
 223         CONTENTS = "some initial contents"
 224         fn.create(CONTENTS, wait_for_numpeers=1)
 225         p = mutable.Publish(fn)
 226         r = mutable.Retrieve(fn)
 227         # make some fake shares
 228         shares_and_ids = ( ["%07d" % i for i in range(10)], range(10) )
 229         target_info = None
 230         p._privkey = FakePrivKey(0)
 231         p._encprivkey = "encprivkey"
 232         p._pubkey = FakePubKey(0)
 233         d = defer.maybeDeferred(p._generate_shares,
 234                                 (shares_and_ids,
 235                                  3, 10,
 236                                  21, # segsize
 237                                  len(CONTENTS),
 238                                  target_info),
 239                                 3, # seqnum
 240                                 "IV"*8)
 241         def _done( (seqnum, root_hash, final_shares, target_info2) ):
 242             self.failUnlessEqual(seqnum, 3)
 243             self.failUnlessEqual(len(root_hash), 32)
 244             self.failUnless(isinstance(final_shares, dict))
 245             self.failUnlessEqual(len(final_shares), 10)
 246             self.failUnlessEqual(sorted(final_shares.keys()), range(10))
 247             for i,sh in final_shares.items():
 248                 self.failUnless(isinstance(sh, str))
 249                 self.failUnlessEqual(len(sh), 381)
 250                 # feed the share through the unpacker as a sanity-check
 251                 pieces = mutable.unpack_share(sh)
 252                 (u_seqnum, u_root_hash, IV, k, N, segsize, datalen,
 253                  pubkey, signature, share_hash_chain, block_hash_tree,
 254                  share_data, enc_privkey) = pieces
 255                 self.failUnlessEqual(u_seqnum, 3)
 256                 self.failUnlessEqual(u_root_hash, root_hash)
 257                 self.failUnlessEqual(k, 3)
 258                 self.failUnlessEqual(N, 10)
 259                 self.failUnlessEqual(segsize, 21)
 260                 self.failUnlessEqual(datalen, len(CONTENTS))
 261                 self.failUnlessEqual(pubkey, FakePubKey(0).serialize())
 262                 sig_material = struct.pack(">BQ32s16s BBQQ",
 263                                            0, seqnum, root_hash, IV,
 264                                            k, N, segsize, datalen)
 265                 self.failUnlessEqual(signature,
 266                                      FakePrivKey(0).sign(sig_material))
 267                 self.failUnless(isinstance(share_hash_chain, dict))
 268                 self.failUnlessEqual(len(share_hash_chain), 4) # ln2(10)++
 269                 for shnum,share_hash in share_hash_chain.items():
 270                     self.failUnless(isinstance(shnum, int))
 271                     self.failUnless(isinstance(share_hash, str))
 272                     self.failUnlessEqual(len(share_hash), 32)
 273                 self.failUnless(isinstance(block_hash_tree, list))
 274                 self.failUnlessEqual(len(block_hash_tree), 1) # very small tree
 275                 self.failUnlessEqual(IV, "IV"*8)
 276                 self.failUnlessEqual(len(share_data), len("%07d" % 1))
 277                 self.failUnlessEqual(enc_privkey, "encprivkey")
 278             self.failUnlessIdentical(target_info, target_info2)
 279         d.addCallback(_done)
 280         return d
 281
 282     def setup_for_sharemap(self, num_peers):
 283         c = FakeClient(num_peers)
 284         fn = FakeFilenode(c)
 285         # .create usually returns a Deferred, but we happen to know it's
 286         # synchronous
 287         CONTENTS = "some initial contents"
 288         fn.create(CONTENTS)
 289         p = FakePublish(fn)
 290         p._storage_index = "\x00"*32
 291         #r = mutable.Retrieve(fn)
 292         p._peers = {}
 293         for peerid in c._peerids:
 294             p._peers[peerid] = {}
 295         return c, p
 296
 297     def shouldFail(self, expected_failure, which, call, *args, **kwargs):
 298         substring = kwargs.pop("substring", None)
 299         d = defer.maybeDeferred(call, *args, **kwargs)
 300         def _done(res):
 301             if isinstance(res, failure.Failure):
 302                 res.trap(expected_failure)
 303                 if substring:
 304                     self.failUnless(substring in str(res),
 305                                     "substring '%s' not in '%s'"
 306                                     % (substring, str(res)))
 307             else:
 308                 self.fail("%s was supposed to raise %s, not get '%s'" %
 309                           (which, expected_failure, res))
 310         d.addBoth(_done)
 311         return d
 312
 313     def test_sharemap_20newpeers(self):
 314         c, p = self.setup_for_sharemap(20)
 315
 316         total_shares = 10
 317         d = p._query_peers(total_shares)
 318         def _done(target_info):
 319             (target_map, shares_per_peer, peer_storage_servers) = target_info
 320             shares_per_peer = {}
 321             for shnum in target_map:
 322                 for (peerid, old_seqnum, old_R) in target_map[shnum]:
 323                     #print "shnum[%d]: send to %s [oldseqnum=%s]" % \
 324                     #      (shnum, idlib.b2a(peerid), old_seqnum)
 325                     if peerid not in shares_per_peer:
 326                         shares_per_peer[peerid] = 1
 327                     else:
 328                         shares_per_peer[peerid] += 1
 329             # verify that we're sending only one share per peer
 330             for peerid, count in shares_per_peer.items():
 331                 self.failUnlessEqual(count, 1)
 332         d.addCallback(_done)
 333         return d
 334
 335     def test_sharemap_3newpeers(self):
 336         c, p = self.setup_for_sharemap(3)
 337
 338         total_shares = 10
 339         d = p._query_peers(total_shares)
 340         def _done(target_info):
 341             (target_map, shares_per_peer, peer_storage_servers) = target_info
 342             shares_per_peer = {}
 343             for shnum in target_map:
 344                 for (peerid, old_seqnum, old_R) in target_map[shnum]:
 345                     if peerid not in shares_per_peer:
 346                         shares_per_peer[peerid] = 1
 347                     else:
 348                         shares_per_peer[peerid] += 1
 349             # verify that we're sending 3 or 4 shares per peer
 350             for peerid, count in shares_per_peer.items():
 351                 self.failUnless(count in (3,4), count)
 352         d.addCallback(_done)
 353         return d
 354
 355     def test_sharemap_nopeers(self):
 356         c, p = self.setup_for_sharemap(0)
 357
 358         total_shares = 10
 359         d = self.shouldFail(NotEnoughPeersError, "test_sharemap_nopeers",
 360                             p._query_peers, total_shares)
 361         return d
 362
 363     def test_write(self):
 364         total_shares = 10
 365         c, p = self.setup_for_sharemap(20)
 366         p._privkey = FakePrivKey(0)
 367         p._encprivkey = "encprivkey"
 368         p._pubkey = FakePubKey(0)
 369         # make some fake shares
 370         CONTENTS = "some initial contents"
 371         shares_and_ids = ( ["%07d" % i for i in range(10)], range(10) )
 372         d = defer.maybeDeferred(p._query_peers, total_shares)
 373         IV = "IV"*8
 374         d.addCallback(lambda target_info:
 375                       p._generate_shares( (shares_and_ids,
 376                                            3, total_shares,
 377                                            21, # segsize
 378                                            len(CONTENTS),
 379                                            target_info),
 380                                           3, # seqnum
 381                                           IV))
 382         d.addCallback(p._send_shares, IV)
 383         def _done((surprised, dispatch_map)):
 384             self.failIf(surprised, "surprised!")
 385         d.addCallback(_done)
 386         return d
 387
 388     def setup_for_publish(self, num_peers):
 389         c = FakeClient(num_peers)
 390         fn = FakeFilenode(c)
 391         # .create usually returns a Deferred, but we happen to know it's
 392         # synchronous
 393         fn.create("")
 394         p = FakePublish(fn)
 395         p._peers = {}
 396         for peerid in c._peerids:
 397             p._peers[peerid] = {}
 398         return c, fn, p
 399
 400     def test_publish(self):
 401         c, fn, p = self.setup_for_publish(20)
 402         # make sure the length of our contents string is not a multiple of k,
 403         # to exercise the padding code.
 404         d = p.publish("New contents of the mutable filenode.")
 405         def _done(res):
 406             # TODO: examine peers and check on their shares
 407             pass
 408         d.addCallback(_done)
 409         return d
 410
 411
 412 class FakePubKey:
 413     def __init__(self, count):
 414         self.count = count
 415     def serialize(self):
 416         return "PUBKEY-%d" % self.count
 417     def verify(self, msg, signature):
 418         return True
 419
 420 class FakePrivKey:
 421     def __init__(self, count):
 422         self.count = count
 423     def serialize(self):
 424         return "PRIVKEY-%d" % self.count
 425     def sign(self, data):
 426         return "SIGN(%s)" % data
 427
 428 class Dirnode(unittest.TestCase):
 429     def setUp(self):
 430         self.client = FakeClient()
 431
 432     def test_create(self):
 433         self.expected_manifest = []
 434
 435         d = self.client.create_empty_dirnode(wait_for_numpeers=1)
 436         def _then(n):
 437             self.failUnless(n.is_mutable())
 438             u = n.get_uri()
 439             self.failUnless(u)
 440             self.failUnless(u.startswith("URI:DIR2:"), u)
 441             u_ro = n.get_readonly_uri()
 442             self.failUnless(u_ro.startswith("URI:DIR2-RO:"), u_ro)
 443             u_v = n.get_verifier()
 444             self.failUnless(u_v.startswith("URI:DIR2-Verifier:"), u_v)
 445             self.expected_manifest.append(u_v)
 446
 447             d = n.list()
 448             d.addCallback(lambda res: self.failUnlessEqual(res, {}))
 449             d.addCallback(lambda res: n.has_child("missing"))
 450             d.addCallback(lambda res: self.failIf(res))
 451             fake_file_uri = uri.WriteableSSKFileURI("a"*16,"b"*32)
 452             ffu_v = fake_file_uri.get_verifier().to_string()
 453             self.expected_manifest.append(ffu_v)
 454             d.addCallback(lambda res: n.set_uri("child", fake_file_uri))
 455
 456             d.addCallback(lambda res: n.create_empty_directory("subdir", wait_for_numpeers=1))
 457             def _created(subdir):
 458                 self.failUnless(isinstance(subdir, FakeNewDirectoryNode))
 459                 self.subdir = subdir
 460                 new_v = subdir.get_verifier()
 461                 self.expected_manifest.append(new_v)
 462             d.addCallback(_created)
 463
 464             d.addCallback(lambda res: n.list())
 465             d.addCallback(lambda children:
 466                           self.failUnlessEqual(sorted(children.keys()),
 467                                                sorted(["child", "subdir"])))
 468
 469             d.addCallback(lambda res: n.build_manifest())
 470             def _check_manifest(manifest):
 471                 self.failUnlessEqual(sorted(manifest),
 472                                      sorted(self.expected_manifest))
 473             d.addCallback(_check_manifest)
 474
 475             def _add_subsubdir(res):
 476                 return self.subdir.create_empty_directory("subsubdir", wait_for_numpeers=1)
 477             d.addCallback(_add_subsubdir)
 478             d.addCallback(lambda res: n.get_child_at_path("subdir/subsubdir"))
 479             d.addCallback(lambda subsubdir:
 480                           self.failUnless(isinstance(subsubdir,
 481                                                      FakeNewDirectoryNode)))
 482             d.addCallback(lambda res: n.get_child_at_path(""))
 483             d.addCallback(lambda res: self.failUnlessEqual(res.get_uri(),
 484                                                            n.get_uri()))
 485
 486             d.addCallback(lambda res: n.get_metadata_for("child"))
 487             d.addCallback(lambda metadata: self.failUnlessEqual(metadata, {}))
 488
 489             d.addCallback(lambda res: n.delete("subdir"))
 490             d.addCallback(lambda old_child:
 491                           self.failUnlessEqual(old_child.get_uri(),
 492                                                self.subdir.get_uri()))
 493
 494             d.addCallback(lambda res: n.list())
 495             d.addCallback(lambda children:
 496                           self.failUnlessEqual(sorted(children.keys()),
 497                                                sorted(["child"])))
 498
 499             return d
 500
 501         d.addCallback(_then)
 502
 503         return d
 504