src/allmydata/mutable/retrieve.py

   1
   2 import time
   3 from itertools import count
   4 from zope.interface import implements
   5 from twisted.internet import defer
   6 from twisted.python import failure
   7 from twisted.internet.interfaces import IPushProducer, IConsumer
   8 from foolscap.api import eventually, fireEventually
   9 from allmydata.interfaces import IRetrieveStatus, NotEnoughSharesError, \
  10      DownloadStopped, MDMF_VERSION, SDMF_VERSION
  11 from allmydata.util import hashutil, log, mathutil
  12 from allmydata.util.dictutil import DictOfSets
  13 from allmydata import hashtree, codec
  14 from allmydata.storage.server import si_b2a
  15 from pycryptopp.cipher.aes import AES
  16 from pycryptopp.publickey import rsa
  17
  18 from allmydata.mutable.common import CorruptShareError, UncoordinatedWriteError
  19 from allmydata.mutable.layout import MDMFSlotReadProxy
  20
  21 class RetrieveStatus:
  22     implements(IRetrieveStatus)
  23     statusid_counter = count(0)
  24     def __init__(self):
  25         self.timings = {}
  26         self.timings["fetch_per_server"] = {}
  27         self.timings["decode"] = 0.0
  28         self.timings["decrypt"] = 0.0
  29         self.timings["cumulative_verify"] = 0.0
  30         self.problems = {}
  31         self.active = True
  32         self.storage_index = None
  33         self.helper = False
  34         self.encoding = ("?","?")
  35         self.size = None
  36         self.status = "Not started"
  37         self.progress = 0.0
  38         self.counter = self.statusid_counter.next()
  39         self.started = time.time()
  40
  41     def get_started(self):
  42         return self.started
  43     def get_storage_index(self):
  44         return self.storage_index
  45     def get_encoding(self):
  46         return self.encoding
  47     def using_helper(self):
  48         return self.helper
  49     def get_size(self):
  50         return self.size
  51     def get_status(self):
  52         return self.status
  53     def get_progress(self):
  54         return self.progress
  55     def get_active(self):
  56         return self.active
  57     def get_counter(self):
  58         return self.counter
  59
  60     def add_fetch_timing(self, peerid, elapsed):
  61         if peerid not in self.timings["fetch_per_server"]:
  62             self.timings["fetch_per_server"][peerid] = []
  63         self.timings["fetch_per_server"][peerid].append(elapsed)
  64     def accumulate_decode_time(self, elapsed):
  65         self.timings["decode"] += elapsed
  66     def accumulate_decrypt_time(self, elapsed):
  67         self.timings["decrypt"] += elapsed
  68     def set_storage_index(self, si):
  69         self.storage_index = si
  70     def set_helper(self, helper):
  71         self.helper = helper
  72     def set_encoding(self, k, n):
  73         self.encoding = (k, n)
  74     def set_size(self, size):
  75         self.size = size
  76     def set_status(self, status):
  77         self.status = status
  78     def set_progress(self, value):
  79         self.progress = value
  80     def set_active(self, value):
  81         self.active = value
  82
  83 class Marker:
  84     pass
  85
  86 class Retrieve:
  87     # this class is currently single-use. Eventually (in MDMF) we will make
  88     # it multi-use, in which case you can call download(range) multiple
  89     # times, and each will have a separate response chain. However the
  90     # Retrieve object will remain tied to a specific version of the file, and
  91     # will use a single ServerMap instance.
  92     implements(IPushProducer)
  93
  94     def __init__(self, filenode, servermap, verinfo, fetch_privkey=False,
  95                  verify=False):
  96         self._node = filenode
  97         assert self._node.get_pubkey()
  98         self._storage_index = filenode.get_storage_index()
  99         assert self._node.get_readkey()
 100         self._last_failure = None
 101         prefix = si_b2a(self._storage_index)[:5]
 102         self._log_number = log.msg("Retrieve(%s): starting" % prefix)
 103         self._outstanding_queries = {} # maps (peerid,shnum) to start_time
 104         self._running = True
 105         self._decoding = False
 106         self._bad_shares = set()
 107
 108         self.servermap = servermap
 109         assert self._node.get_pubkey()
 110         self.verinfo = verinfo
 111         # during repair, we may be called upon to grab the private key, since
 112         # it wasn't picked up during a verify=False checker run, and we'll
 113         # need it for repair to generate a new version.
 114         self._need_privkey = verify or (fetch_privkey
 115                                         and not self._node.get_privkey())
 116
 117         if self._need_privkey:
 118             # TODO: Evaluate the need for this. We'll use it if we want
 119             # to limit how many queries are on the wire for the privkey
 120             # at once.
 121             self._privkey_query_markers = [] # one Marker for each time we've
 122                                              # tried to get the privkey.
 123
 124         # verify means that we are using the downloader logic to verify all
 125         # of our shares. This tells the downloader a few things.
 126         #
 127         # 1. We need to download all of the shares.
 128         # 2. We don't need to decode or decrypt the shares, since our
 129         #    caller doesn't care about the plaintext, only the
 130         #    information about which shares are or are not valid.
 131         # 3. When we are validating readers, we need to validate the
 132         #    signature on the prefix. Do we? We already do this in the
 133         #    servermap update?
 134         self._verify = verify
 135
 136         self._status = RetrieveStatus()
 137         self._status.set_storage_index(self._storage_index)
 138         self._status.set_helper(False)
 139         self._status.set_progress(0.0)
 140         self._status.set_active(True)
 141         (seqnum, root_hash, IV, segsize, datalength, k, N, prefix,
 142          offsets_tuple) = self.verinfo
 143         self._status.set_size(datalength)
 144         self._status.set_encoding(k, N)
 145         self.readers = {}
 146         self._stopped = False
 147         self._pause_deferred = None
 148         self._offset = None
 149         self._read_length = None
 150         self.log("got seqnum %d" % self.verinfo[0])
 151
 152
 153     def get_status(self):
 154         return self._status
 155
 156     def log(self, *args, **kwargs):
 157         if "parent" not in kwargs:
 158             kwargs["parent"] = self._log_number
 159         if "facility" not in kwargs:
 160             kwargs["facility"] = "tahoe.mutable.retrieve"
 161         return log.msg(*args, **kwargs)
 162
 163     def _set_current_status(self, state):
 164         seg = "%d/%d" % (self._current_segment, self._last_segment)
 165         self._status.set_status("segment %s (%s)" % (seg, state))
 166
 167     ###################
 168     # IPushProducer
 169
 170     def pauseProducing(self):
 171         """
 172         I am called by my download target if we have produced too much
 173         data for it to handle. I make the downloader stop producing new
 174         data until my resumeProducing method is called.
 175         """
 176         if self._pause_deferred is not None:
 177             return
 178
 179         # fired when the download is unpaused.
 180         self._old_status = self._status.get_status()
 181         self._set_current_status("paused")
 182
 183         self._pause_deferred = defer.Deferred()
 184
 185
 186     def resumeProducing(self):
 187         """
 188         I am called by my download target once it is ready to begin
 189         receiving data again.
 190         """
 191         if self._pause_deferred is None:
 192             return
 193
 194         p = self._pause_deferred
 195         self._pause_deferred = None
 196         self._status.set_status(self._old_status)
 197
 198         eventually(p.callback, None)
 199
 200     def stopProducing(self):
 201         self._stopped = True
 202         self.resumeProducing()
 203
 204
 205     def _check_for_paused(self, res):
 206         """
 207         I am called just before a write to the consumer. I return a
 208         Deferred that eventually fires with the data that is to be
 209         written to the consumer. If the download has not been paused,
 210         the Deferred fires immediately. Otherwise, the Deferred fires
 211         when the downloader is unpaused.
 212         """
 213         if self._stopped:
 214             raise DownloadStopped("our Consumer called stopProducing()")
 215         if self._pause_deferred is not None:
 216             d = defer.Deferred()
 217             self._pause_deferred.addCallback(lambda ignored: d.callback(res))
 218             return d
 219         return defer.succeed(res)
 220
 221
 222     def download(self, consumer=None, offset=0, size=None):
 223         assert IConsumer.providedBy(consumer) or self._verify
 224
 225         if consumer:
 226             self._consumer = consumer
 227             # we provide IPushProducer, so streaming=True, per
 228             # IConsumer.
 229             self._consumer.registerProducer(self, streaming=True)
 230
 231         self._done_deferred = defer.Deferred()
 232         self._offset = offset
 233         self._read_length = size
 234         self._setup_download()
 235         self._setup_encoding_parameters()
 236         self.log("starting download")
 237         self._started_fetching = time.time()
 238         # The download process beyond this is a state machine.
 239         # _add_active_peers will select the peers that we want to use
 240         # for the download, and then attempt to start downloading. After
 241         # each segment, it will check for doneness, reacting to broken
 242         # peers and corrupt shares as necessary. If it runs out of good
 243         # peers before downloading all of the segments, _done_deferred
 244         # will errback.  Otherwise, it will eventually callback with the
 245         # contents of the mutable file.
 246         self.loop()
 247         return self._done_deferred
 248
 249     def loop(self):
 250         d = fireEventually(None) # avoid #237 recursion limit problem
 251         d.addCallback(lambda ign: self._activate_enough_peers())
 252         d.addCallback(lambda ign: self._download_current_segment())
 253         # when we're done, _download_current_segment will call _done. If we
 254         # aren't, it will call loop() again.
 255         d.addErrback(self._error)
 256
 257     def _setup_download(self):
 258         self._started = time.time()
 259         self._status.set_status("Retrieving Shares")
 260
 261         # how many shares do we need?
 262         (seqnum,
 263          root_hash,
 264          IV,
 265          segsize,
 266          datalength,
 267          k,
 268          N,
 269          prefix,
 270          offsets_tuple) = self.verinfo
 271
 272         # first, which servers can we use?
 273         versionmap = self.servermap.make_versionmap()
 274         shares = versionmap[self.verinfo]
 275         # this sharemap is consumed as we decide to send requests
 276         self.remaining_sharemap = DictOfSets()
 277         for (shnum, peerid, timestamp) in shares:
 278             self.remaining_sharemap.add(shnum, peerid)
 279             # If the servermap update fetched anything, it fetched at least 1
 280             # KiB, so we ask for that much.
 281             # TODO: Change the cache methods to allow us to fetch all of the
 282             # data that they have, then change this method to do that.
 283             any_cache = self._node._read_from_cache(self.verinfo, shnum,
 284                                                     0, 1000)
 285             ss = self.servermap.connections[peerid]
 286             reader = MDMFSlotReadProxy(ss,
 287                                        self._storage_index,
 288                                        shnum,
 289                                        any_cache)
 290             reader.peerid = peerid
 291             self.readers[shnum] = reader
 292         assert len(self.remaining_sharemap) >= k
 293
 294         self.shares = {} # maps shnum to validated blocks
 295         self._active_readers = [] # list of active readers for this dl.
 296         self._block_hash_trees = {} # shnum => hashtree
 297
 298         # We need one share hash tree for the entire file; its leaves
 299         # are the roots of the block hash trees for the shares that
 300         # comprise it, and its root is in the verinfo.
 301         self.share_hash_tree = hashtree.IncompleteHashTree(N)
 302         self.share_hash_tree.set_hashes({0: root_hash})
 303
 304     def decode(self, blocks_and_salts, segnum):
 305         """
 306         I am a helper method that the mutable file update process uses
 307         as a shortcut to decode and decrypt the segments that it needs
 308         to fetch in order to perform a file update. I take in a
 309         collection of blocks and salts, and pick some of those to make a
 310         segment with. I return the plaintext associated with that
 311         segment.
 312         """
 313         # shnum => block hash tree. Unused, but setup_encoding_parameters will
 314         # want to set this.
 315         self._block_hash_trees = None
 316         self._setup_encoding_parameters()
 317
 318         # This is the form expected by decode.
 319         blocks_and_salts = blocks_and_salts.items()
 320         blocks_and_salts = [(True, [d]) for d in blocks_and_salts]
 321
 322         d = self._decode_blocks(blocks_and_salts, segnum)
 323         d.addCallback(self._decrypt_segment)
 324         return d
 325
 326
 327     def _setup_encoding_parameters(self):
 328         """
 329         I set up the encoding parameters, including k, n, the number
 330         of segments associated with this file, and the segment decoders.
 331         """
 332         (seqnum,
 333          root_hash,
 334          IV,
 335          segsize,
 336          datalength,
 337          k,
 338          n,
 339          known_prefix,
 340          offsets_tuple) = self.verinfo
 341         self._required_shares = k
 342         self._total_shares = n
 343         self._segment_size = segsize
 344         self._data_length = datalength
 345
 346         if not IV:
 347             self._version = MDMF_VERSION
 348         else:
 349             self._version = SDMF_VERSION
 350
 351         if datalength and segsize:
 352             self._num_segments = mathutil.div_ceil(datalength, segsize)
 353             self._tail_data_size = datalength % segsize
 354         else:
 355             self._num_segments = 0
 356             self._tail_data_size = 0
 357
 358         self._segment_decoder = codec.CRSDecoder()
 359         self._segment_decoder.set_params(segsize, k, n)
 360
 361         if  not self._tail_data_size:
 362             self._tail_data_size = segsize
 363
 364         self._tail_segment_size = mathutil.next_multiple(self._tail_data_size,
 365                                                          self._required_shares)
 366         if self._tail_segment_size == self._segment_size:
 367             self._tail_decoder = self._segment_decoder
 368         else:
 369             self._tail_decoder = codec.CRSDecoder()
 370             self._tail_decoder.set_params(self._tail_segment_size,
 371                                           self._required_shares,
 372                                           self._total_shares)
 373
 374         self.log("got encoding parameters: "
 375                  "k: %d "
 376                  "n: %d "
 377                  "%d segments of %d bytes each (%d byte tail segment)" % \
 378                  (k, n, self._num_segments, self._segment_size,
 379                   self._tail_segment_size))
 380
 381         if self._block_hash_trees is not None:
 382             for i in xrange(self._total_shares):
 383                 # So we don't have to do this later.
 384                 self._block_hash_trees[i] = hashtree.IncompleteHashTree(self._num_segments)
 385
 386         # Our last task is to tell the downloader where to start and
 387         # where to stop. We use three parameters for that:
 388         #   - self._start_segment: the segment that we need to start
 389         #     downloading from.
 390         #   - self._current_segment: the next segment that we need to
 391         #     download.
 392         #   - self._last_segment: The last segment that we were asked to
 393         #     download.
 394         #
 395         #  We say that the download is complete when
 396         #  self._current_segment > self._last_segment. We use
 397         #  self._start_segment and self._last_segment to know when to
 398         #  strip things off of segments, and how much to strip.
 399         if self._offset:
 400             self.log("got offset: %d" % self._offset)
 401             # our start segment is the first segment containing the
 402             # offset we were given.
 403             start = self._offset // self._segment_size
 404
 405             assert start < self._num_segments
 406             self._start_segment = start
 407             self.log("got start segment: %d" % self._start_segment)
 408         else:
 409             self._start_segment = 0
 410
 411
 412         # If self._read_length is None, then we want to read the whole
 413         # file. Otherwise, we want to read only part of the file, and
 414         # need to figure out where to stop reading.
 415         if self._read_length is not None:
 416             # our end segment is the last segment containing part of the
 417             # segment that we were asked to read.
 418             self.log("got read length %d" % self._read_length)
 419             if self._read_length != 0:
 420                 end_data = self._offset + self._read_length
 421
 422                 # We don't actually need to read the byte at end_data,
 423                 # but the one before it.
 424                 end = (end_data - 1) // self._segment_size
 425
 426                 assert end < self._num_segments
 427                 self._last_segment = end
 428             else:
 429                 self._last_segment = self._start_segment
 430             self.log("got end segment: %d" % self._last_segment)
 431         else:
 432             self._last_segment = self._num_segments - 1
 433
 434         self._current_segment = self._start_segment
 435
 436     def _activate_enough_peers(self):
 437         """
 438         I populate self._active_readers with enough active readers to
 439         retrieve the contents of this mutable file. I am called before
 440         downloading starts, and (eventually) after each validation
 441         error, connection error, or other problem in the download.
 442         """
 443         # TODO: It would be cool to investigate other heuristics for
 444         # reader selection. For instance, the cost (in time the user
 445         # spends waiting for their file) of selecting a really slow peer
 446         # that happens to have a primary share is probably more than
 447         # selecting a really fast peer that doesn't have a primary
 448         # share. Maybe the servermap could be extended to provide this
 449         # information; it could keep track of latency information while
 450         # it gathers more important data, and then this routine could
 451         # use that to select active readers.
 452         #
 453         # (these and other questions would be easier to answer with a
 454         #  robust, configurable tahoe-lafs simulator, which modeled node
 455         #  failures, differences in node speed, and other characteristics
 456         #  that we expect storage servers to have.  You could have
 457         #  presets for really stable grids (like allmydata.com),
 458         #  friendnets, make it easy to configure your own settings, and
 459         #  then simulate the effect of big changes on these use cases
 460         #  instead of just reasoning about what the effect might be. Out
 461         #  of scope for MDMF, though.)
 462
 463         # XXX: Why don't format= log messages work here?
 464
 465         known_shnums = set(self.remaining_sharemap.keys())
 466         used_shnums = set([r.shnum for r in self._active_readers])
 467         unused_shnums = known_shnums - used_shnums
 468
 469         if self._verify:
 470             new_shnums = unused_shnums # use them all
 471         elif len(self._active_readers) < self._required_shares:
 472             # need more shares
 473             more = self._required_shares - len(self._active_readers)
 474             # We favor lower numbered shares, since FEC is faster with
 475             # primary shares than with other shares, and lower-numbered
 476             # shares are more likely to be primary than higher numbered
 477             # shares.
 478             new_shnums = sorted(unused_shnums)[:more]
 479             if len(new_shnums) < more:
 480                 # We don't have enough readers to retrieve the file; fail.
 481                 self._raise_notenoughshareserror()
 482         else:
 483             new_shnums = []
 484
 485         self.log("adding %d new peers to the active list" % len(new_shnums))
 486         for shnum in new_shnums:
 487             reader = self.readers[shnum]
 488             self._active_readers.append(reader)
 489             self.log("added reader for share %d" % shnum)
 490             # Each time we add a reader, we check to see if we need the
 491             # private key. If we do, we politely ask for it and then continue
 492             # computing. If we find that we haven't gotten it at the end of
 493             # segment decoding, then we'll take more drastic measures.
 494             if self._need_privkey and not self._node.is_readonly():
 495                 d = reader.get_encprivkey()
 496                 d.addCallback(self._try_to_validate_privkey, reader)
 497                 # XXX: don't just drop the Deferred. We need error-reporting
 498                 # but not flow-control here.
 499         assert len(self._active_readers) >= self._required_shares
 500
 501     def _try_to_validate_prefix(self, prefix, reader):
 502         """
 503         I check that the prefix returned by a candidate server for
 504         retrieval matches the prefix that the servermap knows about
 505         (and, hence, the prefix that was validated earlier). If it does,
 506         I return True, which means that I approve of the use of the
 507         candidate server for segment retrieval. If it doesn't, I return
 508         False, which means that another server must be chosen.
 509         """
 510         (seqnum,
 511          root_hash,
 512          IV,
 513          segsize,
 514          datalength,
 515          k,
 516          N,
 517          known_prefix,
 518          offsets_tuple) = self.verinfo
 519         if known_prefix != prefix:
 520             self.log("prefix from share %d doesn't match" % reader.shnum)
 521             raise UncoordinatedWriteError("Mismatched prefix -- this could "
 522                                           "indicate an uncoordinated write")
 523         # Otherwise, we're okay -- no issues.
 524
 525
 526     def _remove_reader(self, reader):
 527         """
 528         At various points, we will wish to remove a peer from
 529         consideration and/or use. These include, but are not necessarily
 530         limited to:
 531
 532             - A connection error.
 533             - A mismatched prefix (that is, a prefix that does not match
 534               our conception of the version information string).
 535             - A failing block hash, salt hash, or share hash, which can
 536               indicate disk failure/bit flips, or network trouble.
 537
 538         This method will do that. I will make sure that the
 539         (shnum,reader) combination represented by my reader argument is
 540         not used for anything else during this download. I will not
 541         advise the reader of any corruption, something that my callers
 542         may wish to do on their own.
 543         """
 544         # TODO: When you're done writing this, see if this is ever
 545         # actually used for something that _mark_bad_share isn't. I have
 546         # a feeling that they will be used for very similar things, and
 547         # that having them both here is just going to be an epic amount
 548         # of code duplication.
 549         #
 550         # (well, okay, not epic, but meaningful)
 551         self.log("removing reader %s" % reader)
 552         # Remove the reader from _active_readers
 553         self._active_readers.remove(reader)
 554         # TODO: self.readers.remove(reader)?
 555         for shnum in list(self.remaining_sharemap.keys()):
 556             self.remaining_sharemap.discard(shnum, reader.peerid)
 557
 558
 559     def _mark_bad_share(self, reader, f):
 560         """
 561         I mark the (peerid, shnum) encapsulated by my reader argument as
 562         a bad share, which means that it will not be used anywhere else.
 563
 564         There are several reasons to want to mark something as a bad
 565         share. These include:
 566
 567             - A connection error to the peer.
 568             - A mismatched prefix (that is, a prefix that does not match
 569               our local conception of the version information string).
 570             - A failing block hash, salt hash, share hash, or other
 571               integrity check.
 572
 573         This method will ensure that readers that we wish to mark bad
 574         (for these reasons or other reasons) are not used for the rest
 575         of the download. Additionally, it will attempt to tell the
 576         remote peer (with no guarantee of success) that its share is
 577         corrupt.
 578         """
 579         self.log("marking share %d on server %s as bad" % \
 580                  (reader.shnum, reader))
 581         prefix = self.verinfo[-2]
 582         self.servermap.mark_bad_share(reader.peerid,
 583                                       reader.shnum,
 584                                       prefix)
 585         self._remove_reader(reader)
 586         self._bad_shares.add((reader.peerid, reader.shnum, f))
 587         self._status.problems[reader.peerid] = f
 588         self._last_failure = f
 589         self.notify_server_corruption(reader.peerid, reader.shnum,
 590                                       str(f.value))
 591
 592
 593     def _download_current_segment(self):
 594         """
 595         I download, validate, decode, decrypt, and assemble the segment
 596         that this Retrieve is currently responsible for downloading.
 597         """
 598         assert len(self._active_readers) >= self._required_shares
 599         if self._current_segment > self._last_segment:
 600             # No more segments to download, we're done.
 601             self.log("got plaintext, done")
 602             return self._done()
 603         self.log("on segment %d of %d" %
 604                  (self._current_segment + 1, self._num_segments))
 605         d = self._process_segment(self._current_segment)
 606         d.addCallback(lambda ign: self.loop())
 607         return d
 608
 609     def _process_segment(self, segnum):
 610         """
 611         I download, validate, decode, and decrypt one segment of the
 612         file that this Retrieve is retrieving. This means coordinating
 613         the process of getting k blocks of that file, validating them,
 614         assembling them into one segment with the decoder, and then
 615         decrypting them.
 616         """
 617         self.log("processing segment %d" % segnum)
 618
 619         # TODO: The old code uses a marker. Should this code do that
 620         # too? What did the Marker do?
 621         assert len(self._active_readers) >= self._required_shares
 622
 623         # We need to ask each of our active readers for its block and
 624         # salt. We will then validate those. If validation is
 625         # successful, we will assemble the results into plaintext.
 626         ds = []
 627         for reader in self._active_readers:
 628             started = time.time()
 629             d = reader.get_block_and_salt(segnum)
 630             d2 = self._get_needed_hashes(reader, segnum)
 631             dl = defer.DeferredList([d, d2], consumeErrors=True)
 632             dl.addCallback(self._validate_block, segnum, reader, started)
 633             dl.addErrback(self._validation_or_decoding_failed, [reader])
 634             ds.append(dl)
 635         dl = defer.DeferredList(ds)
 636         if self._verify:
 637             dl.addCallback(lambda ignored: "")
 638             dl.addCallback(self._set_segment)
 639         else:
 640             dl.addCallback(self._maybe_decode_and_decrypt_segment, segnum)
 641         return dl
 642
 643
 644     def _maybe_decode_and_decrypt_segment(self, blocks_and_salts, segnum):
 645         """
 646         I take the results of fetching and validating the blocks from a
 647         callback chain in another method. If the results are such that
 648         they tell me that validation and fetching succeeded without
 649         incident, I will proceed with decoding and decryption.
 650         Otherwise, I will do nothing.
 651         """
 652         self.log("trying to decode and decrypt segment %d" % segnum)
 653         failures = False
 654         for block_and_salt in blocks_and_salts:
 655             if not block_and_salt[0] or block_and_salt[1] == None:
 656                 self.log("some validation operations failed; not proceeding")
 657                 failures = True
 658                 break
 659         if not failures:
 660             self.log("everything looks ok, building segment %d" % segnum)
 661             d = self._decode_blocks(blocks_and_salts, segnum)
 662             d.addCallback(self._decrypt_segment)
 663             d.addErrback(self._validation_or_decoding_failed,
 664                          self._active_readers)
 665             # check to see whether we've been paused before writing
 666             # anything.
 667             d.addCallback(self._check_for_paused)
 668             d.addCallback(self._set_segment)
 669             return d
 670         else:
 671             return defer.succeed(None)
 672
 673
 674     def _set_segment(self, segment):
 675         """
 676         Given a plaintext segment, I register that segment with the
 677         target that is handling the file download.
 678         """
 679         self.log("got plaintext for segment %d" % self._current_segment)
 680         if self._current_segment == self._start_segment:
 681             # We're on the first segment. It's possible that we want
 682             # only some part of the end of this segment, and that we
 683             # just downloaded the whole thing to get that part. If so,
 684             # we need to account for that and give the reader just the
 685             # data that they want.
 686             n = self._offset % self._segment_size
 687             self.log("stripping %d bytes off of the first segment" % n)
 688             self.log("original segment length: %d" % len(segment))
 689             segment = segment[n:]
 690             self.log("new segment length: %d" % len(segment))
 691
 692         if self._current_segment == self._last_segment and self._read_length is not None:
 693             # We're on the last segment. It's possible that we only want
 694             # part of the beginning of this segment, and that we
 695             # downloaded the whole thing anyway. Make sure to give the
 696             # caller only the portion of the segment that they want to
 697             # receive.
 698             extra = self._read_length
 699             if self._start_segment != self._last_segment:
 700                 extra -= self._segment_size - \
 701                             (self._offset % self._segment_size)
 702             extra %= self._segment_size
 703             self.log("original segment length: %d" % len(segment))
 704             segment = segment[:extra]
 705             self.log("new segment length: %d" % len(segment))
 706             self.log("only taking %d bytes of the last segment" % extra)
 707
 708         if not self._verify:
 709             self._consumer.write(segment)
 710         else:
 711             # we don't care about the plaintext if we are doing a verify.
 712             segment = None
 713         self._current_segment += 1
 714
 715
 716     def _validation_or_decoding_failed(self, f, readers):
 717         """
 718         I am called when a block or a salt fails to correctly validate, or when
 719         the decryption or decoding operation fails for some reason.  I react to
 720         this failure by notifying the remote server of corruption, and then
 721         removing the remote peer from further activity.
 722         """
 723         assert isinstance(readers, list)
 724         bad_shnums = [reader.shnum for reader in readers]
 725
 726         self.log("validation or decoding failed on share(s) %s, peer(s) %s "
 727                  ", segment %d: %s" % \
 728                  (bad_shnums, readers, self._current_segment, str(f)))
 729         for reader in readers:
 730             self._mark_bad_share(reader, f)
 731         return
 732
 733
 734     def _validate_block(self, results, segnum, reader, started):
 735         """
 736         I validate a block from one share on a remote server.
 737         """
 738         # Grab the part of the block hash tree that is necessary to
 739         # validate this block, then generate the block hash root.
 740         self.log("validating share %d for segment %d" % (reader.shnum,
 741                                                              segnum))
 742         elapsed = time.time() - started
 743         self._status.add_fetch_timing(reader.peerid, elapsed)
 744         self._set_current_status("validating blocks")
 745         # Did we fail to fetch either of the things that we were
 746         # supposed to? Fail if so.
 747         if not results[0][0] and results[1][0]:
 748             # handled by the errback handler.
 749
 750             # These all get batched into one query, so the resulting
 751             # failure should be the same for all of them, so we can just
 752             # use the first one.
 753             assert isinstance(results[0][1], failure.Failure)
 754
 755             f = results[0][1]
 756             raise CorruptShareError(reader.peerid,
 757                                     reader.shnum,
 758                                     "Connection error: %s" % str(f))
 759
 760         block_and_salt, block_and_sharehashes = results
 761         block, salt = block_and_salt[1]
 762         blockhashes, sharehashes = block_and_sharehashes[1]
 763
 764         blockhashes = dict(enumerate(blockhashes[1]))
 765         self.log("the reader gave me the following blockhashes: %s" % \
 766                  blockhashes.keys())
 767         self.log("the reader gave me the following sharehashes: %s" % \
 768                  sharehashes[1].keys())
 769         bht = self._block_hash_trees[reader.shnum]
 770
 771         if bht.needed_hashes(segnum, include_leaf=True):
 772             try:
 773                 bht.set_hashes(blockhashes)
 774             except (hashtree.BadHashError, hashtree.NotEnoughHashesError, \
 775                     IndexError), e:
 776                 raise CorruptShareError(reader.peerid,
 777                                         reader.shnum,
 778                                         "block hash tree failure: %s" % e)
 779
 780         if self._version == MDMF_VERSION:
 781             blockhash = hashutil.block_hash(salt + block)
 782         else:
 783             blockhash = hashutil.block_hash(block)
 784         # If this works without an error, then validation is
 785         # successful.
 786         try:
 787            bht.set_hashes(leaves={segnum: blockhash})
 788         except (hashtree.BadHashError, hashtree.NotEnoughHashesError, \
 789                 IndexError), e:
 790             raise CorruptShareError(reader.peerid,
 791                                     reader.shnum,
 792                                     "block hash tree failure: %s" % e)
 793
 794         # Reaching this point means that we know that this segment
 795         # is correct. Now we need to check to see whether the share
 796         # hash chain is also correct.
 797         # SDMF wrote share hash chains that didn't contain the
 798         # leaves, which would be produced from the block hash tree.
 799         # So we need to validate the block hash tree first. If
 800         # successful, then bht[0] will contain the root for the
 801         # shnum, which will be a leaf in the share hash tree, which
 802         # will allow us to validate the rest of the tree.
 803         if self.share_hash_tree.needed_hashes(reader.shnum,
 804                                               include_leaf=True) or \
 805                                               self._verify:
 806             try:
 807                 self.share_hash_tree.set_hashes(hashes=sharehashes[1],
 808                                             leaves={reader.shnum: bht[0]})
 809             except (hashtree.BadHashError, hashtree.NotEnoughHashesError, \
 810                     IndexError), e:
 811                 raise CorruptShareError(reader.peerid,
 812                                         reader.shnum,
 813                                         "corrupt hashes: %s" % e)
 814
 815         self.log('share %d is valid for segment %d' % (reader.shnum,
 816                                                        segnum))
 817         return {reader.shnum: (block, salt)}
 818
 819
 820     def _get_needed_hashes(self, reader, segnum):
 821         """
 822         I get the hashes needed to validate segnum from the reader, then return
 823         to my caller when this is done.
 824         """
 825         bht = self._block_hash_trees[reader.shnum]
 826         needed = bht.needed_hashes(segnum, include_leaf=True)
 827         # The root of the block hash tree is also a leaf in the share
 828         # hash tree. So we don't need to fetch it from the remote
 829         # server. In the case of files with one segment, this means that
 830         # we won't fetch any block hash tree from the remote server,
 831         # since the hash of each share of the file is the entire block
 832         # hash tree, and is a leaf in the share hash tree. This is fine,
 833         # since any share corruption will be detected in the share hash
 834         # tree.
 835         #needed.discard(0)
 836         self.log("getting blockhashes for segment %d, share %d: %s" % \
 837                  (segnum, reader.shnum, str(needed)))
 838         d1 = reader.get_blockhashes(needed, force_remote=True)
 839         if self.share_hash_tree.needed_hashes(reader.shnum):
 840             need = self.share_hash_tree.needed_hashes(reader.shnum)
 841             self.log("also need sharehashes for share %d: %s" % (reader.shnum,
 842                                                                  str(need)))
 843             d2 = reader.get_sharehashes(need, force_remote=True)
 844         else:
 845             d2 = defer.succeed({}) # the logic in the next method
 846                                    # expects a dict
 847         dl = defer.DeferredList([d1, d2], consumeErrors=True)
 848         return dl
 849
 850
 851     def _decode_blocks(self, blocks_and_salts, segnum):
 852         """
 853         I take a list of k blocks and salts, and decode that into a
 854         single encrypted segment.
 855         """
 856         d = {}
 857         # We want to merge our dictionaries to the form
 858         # {shnum: blocks_and_salts}
 859         #
 860         # The dictionaries come from validate block that way, so we just
 861         # need to merge them.
 862         for block_and_salt in blocks_and_salts:
 863             d.update(block_and_salt[1])
 864
 865         # All of these blocks should have the same salt; in SDMF, it is
 866         # the file-wide IV, while in MDMF it is the per-segment salt. In
 867         # either case, we just need to get one of them and use it.
 868         #
 869         # d.items()[0] is like (shnum, (block, salt))
 870         # d.items()[0][1] is like (block, salt)
 871         # d.items()[0][1][1] is the salt.
 872         salt = d.items()[0][1][1]
 873         # Next, extract just the blocks from the dict. We'll use the
 874         # salt in the next step.
 875         share_and_shareids = [(k, v[0]) for k, v in d.items()]
 876         d2 = dict(share_and_shareids)
 877         shareids = []
 878         shares = []
 879         for shareid, share in d2.items():
 880             shareids.append(shareid)
 881             shares.append(share)
 882
 883         self._set_current_status("decoding")
 884         started = time.time()
 885         assert len(shareids) >= self._required_shares, len(shareids)
 886         # zfec really doesn't want extra shares
 887         shareids = shareids[:self._required_shares]
 888         shares = shares[:self._required_shares]
 889         self.log("decoding segment %d" % segnum)
 890         if segnum == self._num_segments - 1:
 891             d = defer.maybeDeferred(self._tail_decoder.decode, shares, shareids)
 892         else:
 893             d = defer.maybeDeferred(self._segment_decoder.decode, shares, shareids)
 894         def _process(buffers):
 895             segment = "".join(buffers)
 896             self.log(format="now decoding segment %(segnum)s of %(numsegs)s",
 897                      segnum=segnum,
 898                      numsegs=self._num_segments,
 899                      level=log.NOISY)
 900             self.log(" joined length %d, datalength %d" %
 901                      (len(segment), self._data_length))
 902             if segnum == self._num_segments - 1:
 903                 size_to_use = self._tail_data_size
 904             else:
 905                 size_to_use = self._segment_size
 906             segment = segment[:size_to_use]
 907             self.log(" segment len=%d" % len(segment))
 908             self._status.accumulate_decode_time(time.time() - started)
 909             return segment, salt
 910         d.addCallback(_process)
 911         return d
 912
 913
 914     def _decrypt_segment(self, segment_and_salt):
 915         """
 916         I take a single segment and its salt, and decrypt it. I return
 917         the plaintext of the segment that is in my argument.
 918         """
 919         segment, salt = segment_and_salt
 920         self._set_current_status("decrypting")
 921         self.log("decrypting segment %d" % self._current_segment)
 922         started = time.time()
 923         key = hashutil.ssk_readkey_data_hash(salt, self._node.get_readkey())
 924         decryptor = AES(key)
 925         plaintext = decryptor.process(segment)
 926         self._status.accumulate_decrypt_time(time.time() - started)
 927         return plaintext
 928
 929
 930     def notify_server_corruption(self, peerid, shnum, reason):
 931         ss = self.servermap.connections[peerid]
 932         ss.callRemoteOnly("advise_corrupt_share",
 933                           "mutable", self._storage_index, shnum, reason)
 934
 935
 936     def _try_to_validate_privkey(self, enc_privkey, reader):
 937         alleged_privkey_s = self._node._decrypt_privkey(enc_privkey)
 938         alleged_writekey = hashutil.ssk_writekey_hash(alleged_privkey_s)
 939         if alleged_writekey != self._node.get_writekey():
 940             self.log("invalid privkey from %s shnum %d" %
 941                      (reader, reader.shnum),
 942                      level=log.WEIRD, umid="YIw4tA")
 943             if self._verify:
 944                 self.servermap.mark_bad_share(reader.peerid, reader.shnum,
 945                                               self.verinfo[-2])
 946                 e = CorruptShareError(reader.peerid,
 947                                       reader.shnum,
 948                                       "invalid privkey")
 949                 f = failure.Failure(e)
 950                 self._bad_shares.add((reader.peerid, reader.shnum, f))
 951             return
 952
 953         # it's good
 954         self.log("got valid privkey from shnum %d on reader %s" %
 955                  (reader.shnum, reader))
 956         privkey = rsa.create_signing_key_from_string(alleged_privkey_s)
 957         self._node._populate_encprivkey(enc_privkey)
 958         self._node._populate_privkey(privkey)
 959         self._need_privkey = False
 960
 961
 962
 963     def _done(self):
 964         """
 965         I am called by _download_current_segment when the download process
 966         has finished successfully. After making some useful logging
 967         statements, I return the decrypted contents to the owner of this
 968         Retrieve object through self._done_deferred.
 969         """
 970         self._running = False
 971         self._status.set_active(False)
 972         now = time.time()
 973         self._status.timings['total'] = now - self._started
 974         self._status.timings['fetch'] = now - self._started_fetching
 975         self._status.set_status("Finished")
 976         self._status.set_progress(1.0)
 977
 978         # remember the encoding parameters, use them again next time
 979         (seqnum, root_hash, IV, segsize, datalength, k, N, prefix,
 980          offsets_tuple) = self.verinfo
 981         self._node._populate_required_shares(k)
 982         self._node._populate_total_shares(N)
 983
 984         if self._verify:
 985             ret = list(self._bad_shares)
 986             self.log("done verifying, found %d bad shares" % len(ret))
 987         else:
 988             # TODO: upload status here?
 989             ret = self._consumer
 990             self._consumer.unregisterProducer()
 991         eventually(self._done_deferred.callback, ret)
 992
 993
 994     def _raise_notenoughshareserror(self):
 995         """
 996         I am called by _activate_enough_peers when there are not enough
 997         active peers left to complete the download. After making some
 998         useful logging statements, I throw an exception to that effect
 999         to the caller of this Retrieve object through
1000         self._done_deferred.
1001         """
1002
1003         format = ("ran out of peers: "
1004                   "have %(have)d of %(total)d segments "
1005                   "found %(bad)d bad shares "
1006                   "encoding %(k)d-of-%(n)d")
1007         args = {"have": self._current_segment,
1008                 "total": self._num_segments,
1009                 "need": self._last_segment,
1010                 "k": self._required_shares,
1011                 "n": self._total_shares,
1012                 "bad": len(self._bad_shares)}
1013         raise NotEnoughSharesError("%s, last failure: %s" %
1014                                    (format % args, str(self._last_failure)))
1015
1016     def _error(self, f):
1017         # all errors, including NotEnoughSharesError, land here
1018         self._running = False
1019         self._status.set_active(False)
1020         now = time.time()
1021         self._status.timings['total'] = now - self._started
1022         self._status.timings['fetch'] = now - self._started_fetching
1023         self._status.set_status("Failed")
1024         eventually(self._done_deferred.errback, f)