1 from foolscap.api import DeadReferenceError, RemoteException
2 from allmydata.hashtree import IncompleteHashTree
3 from allmydata.check_results import CheckResults
4 from allmydata.immutable import download
5 from allmydata.uri import CHKFileVerifierURI
6 from allmydata.util.assertutil import precondition
7 from allmydata.util import base32, idlib, deferredutil, dictutil, log
8 from allmydata.util.hashutil import file_renewal_secret_hash, \
9 file_cancel_secret_hash, bucket_renewal_secret_hash, \
10 bucket_cancel_secret_hash
12 from allmydata.immutable import layout
14 class Checker(log.PrefixingLogMixin):
15 """I query all servers to see if M uniquely-numbered shares are
18 If the verify flag was passed to my constructor, then for each share I
19 download every data block and all metadata from each server and perform a
20 cryptographic integrity check on all of it. If not, I just ask each
21 server 'Which shares do you have?' and believe its answer.
23 In either case, I wait until I have gotten responses from all servers.
24 This fact -- that I wait -- means that an ill-behaved server which fails
25 to answer my questions will make me wait indefinitely. If it is
26 ill-behaved in a way that triggers the underlying foolscap timeouts, then
27 I will wait only as long as those foolscap timeouts, but if it is
28 ill-behaved in a way which placates the foolscap timeouts but still
29 doesn't answer my question then I will wait indefinitely.
31 Before I send any new request to a server, I always ask the 'monitor'
32 object that was passed into my constructor whether this task has been
33 cancelled (by invoking its raise_if_cancelled() method).
35 def __init__(self, verifycap, servers, verify, add_lease, secret_holder,
37 assert precondition(isinstance(verifycap, CHKFileVerifierURI), verifycap, type(verifycap))
38 assert precondition(isinstance(servers, (set, frozenset)), servers)
39 for (serverid, serverrref) in servers:
40 assert precondition(isinstance(serverid, str))
42 prefix = "%s" % base32.b2a_l(verifycap.storage_index[:8], 60)
43 log.PrefixingLogMixin.__init__(self, facility="tahoe.immutable.checker", prefix=prefix)
45 self._verifycap = verifycap
47 self._monitor = monitor
48 self._servers = servers
49 self._verify = verify # bool: verify what the servers claim, or not?
50 self._add_lease = add_lease
52 frs = file_renewal_secret_hash(secret_holder.get_renewal_secret(),
53 self._verifycap.storage_index)
54 self.file_renewal_secret = frs
55 fcs = file_cancel_secret_hash(secret_holder.get_cancel_secret(),
56 self._verifycap.storage_index)
57 self.file_cancel_secret = fcs
59 def _get_renewal_secret(self, peerid):
60 return bucket_renewal_secret_hash(self.file_renewal_secret, peerid)
61 def _get_cancel_secret(self, peerid):
62 return bucket_cancel_secret_hash(self.file_cancel_secret, peerid)
64 def _get_buckets(self, server, storageindex, serverid):
65 """Return a deferred that eventually fires with ({sharenum: bucket},
66 serverid, success). In case the server is disconnected or returns a
67 Failure then it fires with ({}, serverid, False) (A server
68 disconnecting or returning a Failure when we ask it for buckets is
69 the same, for our purposes, as a server that says it has none, except
70 that we want to track and report whether or not each server
74 renew_secret = self._get_renewal_secret(serverid)
75 cancel_secret = self._get_cancel_secret(serverid)
76 d2 = server.callRemote("add_lease", storageindex,
77 renew_secret, cancel_secret)
78 d2.addErrback(self._add_lease_failed, serverid, storageindex)
80 d = server.callRemote("get_buckets", storageindex)
81 def _wrap_results(res):
82 return (res, serverid, True)
86 if f.check(DeadReferenceError):
88 self.log("failure from server on 'get_buckets' the REMOTE failure was:", facility="tahoe.immutable.checker", failure=f, level=level, umid="3uuBUQ")
89 return ({}, serverid, False)
91 d.addCallbacks(_wrap_results, _trap_errs)
94 def _add_lease_failed(self, f, peerid, storage_index):
95 # Older versions of Tahoe didn't handle the add-lease message very
96 # well: <=1.1.0 throws a NameError because it doesn't implement
97 # remote_add_lease(), 1.2.0/1.3.0 throw IndexError on unknown buckets
98 # (which is most of them, since we send add-lease to everybody,
99 # before we know whether or not they have any shares for us), and
100 # 1.2.0 throws KeyError even on known buckets due to an internal bug
101 # in the latency-measuring code.
103 # we want to ignore the known-harmless errors and log the others. In
104 # particular we want to log any local errors caused by coding
107 if f.check(DeadReferenceError):
109 if f.check(RemoteException):
110 if f.value.failure.check(KeyError, IndexError, NameError):
111 # this may ignore a bit too much, but that only hurts us
114 self.log(format="error in add_lease from [%(peerid)s]: %(f_value)s",
115 peerid=idlib.shortnodeid_b2a(peerid),
116 f_value=str(f.value),
118 level=log.WEIRD, umid="atbAxw")
120 # local errors are cause for alarm
122 format="local error in add_lease to [%(peerid)s]: %(f_value)s",
123 peerid=idlib.shortnodeid_b2a(peerid),
124 f_value=str(f.value),
125 level=log.WEIRD, umid="hEGuQg")
128 def _download_and_verify(self, serverid, sharenum, bucket):
129 """Start an attempt to download and verify every block in this bucket
130 and return a deferred that will eventually fire once the attempt
133 If you download and verify every block then fire with (True,
134 sharenum, None), else if the share data couldn't be parsed because it
135 was of an unknown version number fire with (False, sharenum,
136 'incompatible'), else if any of the blocks were invalid, fire with
137 (False, sharenum, 'corrupt'), else if the server disconnected (False,
138 sharenum, 'disconnect'), else if the server returned a Failure during
139 the process fire with (False, sharenum, 'failure').
141 If there is an internal error such as an uncaught exception in this
142 code, then the deferred will errback, but if there is a remote error
143 such as the server failing or the returned data being incorrect then
144 it will not errback -- it will fire normally with the indicated
147 vcap = self._verifycap
148 b = layout.ReadBucketProxy(bucket, serverid, vcap.storage_index)
149 veup = download.ValidatedExtendedURIProxy(b, vcap)
153 share_hash_tree = IncompleteHashTree(vcap.total_shares)
154 share_hash_tree.set_hashes({0: vup.share_root_hash})
156 vrbp = download.ValidatedReadBucketProxy(sharenum, b,
162 # note: normal download doesn't use get_all_sharehashes(),
163 # because it gets more data than necessary. We've discussed the
164 # security properties of having verification and download look
165 # identical (so the server couldn't, say, provide good responses
166 # for one and not the other), but I think that full verification
167 # is more important than defending against inconsistent server
168 # behavior. Besides, they can't pass the verifier without storing
169 # all the data, so there's not so much to be gained by behaving
171 d = vrbp.get_all_sharehashes()
172 # we fill share_hash_tree before fetching any blocks, so the
173 # block fetches won't send redundant share-hash-tree requests, to
174 # speed things up. Then we fetch+validate all the blockhashes.
175 d.addCallback(lambda ign: vrbp.get_all_blockhashes())
177 cht = IncompleteHashTree(vup.num_segments)
178 cht.set_hashes({0: vup.crypttext_root_hash})
179 d.addCallback(lambda ign: vrbp.get_all_crypttext_hashes(cht))
181 d.addCallback(lambda ign: vrbp)
183 d.addCallback(_got_ueb)
185 def _discard_result(r):
186 assert isinstance(r, str), r
189 def _get_blocks(vrbp):
191 for blocknum in range(veup.num_segments):
192 db = vrbp.get_block(blocknum)
193 db.addCallback(_discard_result)
195 # this gatherResults will fire once every block of this share has
196 # been downloaded and verified, or else it will errback.
197 return deferredutil.gatherResults(ds)
198 d.addCallback(_get_blocks)
200 # if none of those errbacked, the blocks (and the hashes above them)
203 return (True, sharenum, None)
204 d.addCallback(_all_good)
206 # but if anything fails, we'll land here
208 # We didn't succeed at fetching and verifying all the blocks of
209 # this share. Handle each reason for failure differently.
211 if f.check(DeadReferenceError):
212 return (False, sharenum, 'disconnect')
213 elif f.check(RemoteException):
214 return (False, sharenum, 'failure')
215 elif f.check(layout.ShareVersionIncompatible):
216 return (False, sharenum, 'incompatible')
217 elif f.check(layout.LayoutInvalid,
218 layout.RidiculouslyLargeURIExtensionBlock,
219 download.BadOrMissingHash,
220 download.BadURIExtensionHashValue):
221 return (False, sharenum, 'corrupt')
223 # if it wasn't one of those reasons, re-raise the error
229 def _verify_server_shares(self, serverid, ss):
230 """ Return a deferred which eventually fires with a tuple of
231 (set(sharenum), serverid, set(corruptsharenum),
232 set(incompatiblesharenum), success) showing all the shares verified
233 to be served by this server, and all the corrupt shares served by the
234 server, and all the incompatible shares served by the server. In case
235 the server is disconnected or returns a Failure then it fires with
236 the last element False.
238 A server disconnecting or returning a failure when we ask it for
239 shares is the same, for our purposes, as a server that says it has
240 none or offers invalid ones, except that we want to track and report
241 the server's behavior. Similarly, the presence of corrupt shares is
242 mainly of use for diagnostics -- you can typically treat it as just
243 like being no share at all by just observing its absence from the
244 verified shares dict and ignoring its presence in the corrupt shares
247 The 'success' argument means whether the server responded to *any*
248 queries during this process, so if it responded to some queries and
249 then disconnected and ceased responding, or returned a failure, it is
250 still marked with the True flag for 'success'.
252 d = self._get_buckets(ss, self._verifycap.storage_index, serverid)
254 def _got_buckets(result):
255 bucketdict, serverid, success = result
258 for (sharenum, bucket) in bucketdict.items():
259 d = self._download_and_verify(serverid, sharenum, bucket)
262 dl = deferredutil.gatherResults(shareverds)
264 def collect(results):
268 for succ, sharenum, whynot in results:
270 verified.add(sharenum)
272 if whynot == 'corrupt':
273 corrupt.add(sharenum)
274 elif whynot == 'incompatible':
275 incompatible.add(sharenum)
276 return (verified, serverid, corrupt, incompatible, success)
278 dl.addCallback(collect)
282 f.trap(RemoteException, DeadReferenceError)
283 return (set(), serverid, set(), set(), False)
285 d.addCallbacks(_got_buckets, _err)
288 def _check_server_shares(self, serverid, ss):
289 """Return a deferred which eventually fires with a tuple of
290 (set(sharenum), serverid, set(), set(), responded) showing all the
291 shares claimed to be served by this server. In case the server is
292 disconnected then it fires with (set() serverid, set(), set(), False)
293 (a server disconnecting when we ask it for buckets is the same, for
294 our purposes, as a server that says it has none, except that we want
295 to track and report whether or not each server responded.)"""
296 def _curry_empty_corrupted(res):
297 buckets, serverid, responded = res
298 return (set(buckets), serverid, set(), set(), responded)
299 d = self._get_buckets(ss, self._verifycap.storage_index, serverid)
300 d.addCallback(_curry_empty_corrupted)
303 def _format_results(self, results):
304 cr = CheckResults(self._verifycap, self._verifycap.storage_index)
306 d['count-shares-needed'] = self._verifycap.needed_shares
307 d['count-shares-expected'] = self._verifycap.total_shares
309 verifiedshares = dictutil.DictOfSets() # {sharenum: set(serverid)}
310 servers = {} # {serverid: set(sharenums)}
311 corruptsharelocators = [] # (serverid, storageindex, sharenum)
312 incompatiblesharelocators = [] # (serverid, storageindex, sharenum)
314 for theseverifiedshares, thisserverid, thesecorruptshares, theseincompatibleshares, thisresponded in results:
315 servers.setdefault(thisserverid, set()).update(theseverifiedshares)
316 for sharenum in theseverifiedshares:
317 verifiedshares.setdefault(sharenum, set()).add(thisserverid)
318 for sharenum in thesecorruptshares:
319 corruptsharelocators.append((thisserverid, self._verifycap.storage_index, sharenum))
320 for sharenum in theseincompatibleshares:
321 incompatiblesharelocators.append((thisserverid, self._verifycap.storage_index, sharenum))
323 d['count-shares-good'] = len(verifiedshares)
324 d['count-good-share-hosts'] = len([s for s in servers.keys() if servers[s]])
326 assert len(verifiedshares) <= self._verifycap.total_shares, (verifiedshares.keys(), self._verifycap.total_shares)
327 if len(verifiedshares) == self._verifycap.total_shares:
329 cr.set_summary("Healthy")
331 cr.set_healthy(False)
332 cr.set_summary("Not Healthy: %d shares (enc %d-of-%d)" %
333 (len(verifiedshares),
334 self._verifycap.needed_shares,
335 self._verifycap.total_shares))
336 if len(verifiedshares) >= self._verifycap.needed_shares:
337 cr.set_recoverable(True)
338 d['count-recoverable-versions'] = 1
339 d['count-unrecoverable-versions'] = 0
341 cr.set_recoverable(False)
342 d['count-recoverable-versions'] = 0
343 d['count-unrecoverable-versions'] = 1
345 d['servers-responding'] = list(servers)
346 d['sharemap'] = verifiedshares
347 # no such thing as wrong shares of an immutable file
348 d['count-wrong-shares'] = 0
349 d['list-corrupt-shares'] = corruptsharelocators
350 d['count-corrupt-shares'] = len(corruptsharelocators)
351 d['list-incompatible-shares'] = incompatiblesharelocators
352 d['count-incompatible-shares'] = len(incompatiblesharelocators)
355 # The file needs rebalancing if the set of servers that have at least
356 # one share is less than the number of uniquely-numbered shares
358 cr.set_needs_rebalancing(d['count-good-share-hosts'] < d['count-shares-good'])
367 for (serverid, ss) in self._servers:
368 ds.append(self._verify_server_shares(serverid, ss))
370 for (serverid, ss) in self._servers:
371 ds.append(self._check_server_shares(serverid, ss))
373 return deferredutil.gatherResults(ds).addCallback(self._format_results)