archive.py 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717
  1. from datetime import datetime
  2. from getpass import getuser
  3. from itertools import groupby
  4. import errno
  5. import shutil
  6. import tempfile
  7. from attic.key import key_factory
  8. from attic.remote import cache_if_remote
  9. import msgpack
  10. import os
  11. import socket
  12. import stat
  13. import sys
  14. import time
  15. from io import BytesIO
  16. from attic import xattr
  17. from attic.platform import acl_get, acl_set
  18. from attic.chunker import Chunker
  19. from attic.hashindex import ChunkIndex
  20. from attic.helpers import parse_timestamp, Error, uid2user, user2uid, gid2group, group2gid, \
  21. Manifest, Statistics, decode_dict, st_mtime_ns, make_path_safe, StableDict, int_to_bigint, bigint_to_int
  22. ITEMS_BUFFER = 1024 * 1024
  23. CHUNK_MIN = 1024
  24. WINDOW_SIZE = 0xfff
  25. CHUNK_MASK = 0xffff
  26. utime_supports_fd = os.utime in getattr(os, 'supports_fd', {})
  27. utime_supports_follow_symlinks = os.utime in getattr(os, 'supports_follow_symlinks', {})
  28. has_mtime_ns = sys.version >= '3.3'
  29. has_lchmod = hasattr(os, 'lchmod')
  30. has_lchflags = hasattr(os, 'lchflags')
  31. # Python <= 3.2 raises OSError instead of PermissionError (See #164)
  32. try:
  33. PermissionError = PermissionError
  34. except NameError:
  35. PermissionError = OSError
  36. class DownloadPipeline:
  37. def __init__(self, repository, key):
  38. self.repository = repository
  39. self.key = key
  40. def unpack_many(self, ids, filter=None, preload=False):
  41. unpacker = msgpack.Unpacker(use_list=False)
  42. for data in self.fetch_many(ids):
  43. unpacker.feed(data)
  44. items = [decode_dict(item, (b'path', b'source', b'user', b'group')) for item in unpacker]
  45. if filter:
  46. items = [item for item in items if filter(item)]
  47. if preload:
  48. for item in items:
  49. if b'chunks' in item:
  50. self.repository.preload([c[0] for c in item[b'chunks']])
  51. for item in items:
  52. yield item
  53. def fetch_many(self, ids, is_preloaded=False):
  54. for id_, data in zip(ids, self.repository.get_many(ids, is_preloaded=is_preloaded)):
  55. yield self.key.decrypt(id_, data)
  56. class ChunkBuffer:
  57. BUFFER_SIZE = 1 * 1024 * 1024
  58. def __init__(self, key):
  59. self.buffer = BytesIO()
  60. self.packer = msgpack.Packer(unicode_errors='surrogateescape')
  61. self.chunks = []
  62. self.key = key
  63. self.chunker = Chunker(WINDOW_SIZE, CHUNK_MASK, CHUNK_MIN, self.key.chunk_seed)
  64. def add(self, item):
  65. self.buffer.write(self.packer.pack(StableDict(item)))
  66. if self.is_full():
  67. self.flush()
  68. def write_chunk(self, chunk):
  69. raise NotImplementedError
  70. def flush(self, flush=False):
  71. if self.buffer.tell() == 0:
  72. return
  73. self.buffer.seek(0)
  74. chunks = list(bytes(s) for s in self.chunker.chunkify(self.buffer))
  75. self.buffer.seek(0)
  76. self.buffer.truncate(0)
  77. # Leave the last partial chunk in the buffer unless flush is True
  78. end = None if flush or len(chunks) == 1 else -1
  79. for chunk in chunks[:end]:
  80. self.chunks.append(self.write_chunk(chunk))
  81. if end == -1:
  82. self.buffer.write(chunks[-1])
  83. def is_full(self):
  84. return self.buffer.tell() > self.BUFFER_SIZE
  85. class CacheChunkBuffer(ChunkBuffer):
  86. def __init__(self, cache, key, stats):
  87. super(CacheChunkBuffer, self).__init__(key)
  88. self.cache = cache
  89. self.stats = stats
  90. def write_chunk(self, chunk):
  91. id_, _, _ = self.cache.add_chunk(self.key.id_hash(chunk), chunk, self.stats)
  92. return id_
  93. class Archive:
  94. class DoesNotExist(Error):
  95. """Archive {} does not exist"""
  96. class AlreadyExists(Error):
  97. """Archive {} already exists"""
  98. def __init__(self, repository, key, manifest, name, cache=None, create=False,
  99. checkpoint_interval=300, numeric_owner=False):
  100. self.cwd = os.getcwd()
  101. self.key = key
  102. self.repository = repository
  103. self.cache = cache
  104. self.manifest = manifest
  105. self.hard_links = {}
  106. self.stats = Statistics()
  107. self.name = name
  108. self.checkpoint_interval = checkpoint_interval
  109. self.numeric_owner = numeric_owner
  110. self.pipeline = DownloadPipeline(self.repository, self.key)
  111. if create:
  112. self.items_buffer = CacheChunkBuffer(self.cache, self.key, self.stats)
  113. self.chunker = Chunker(WINDOW_SIZE, CHUNK_MASK, CHUNK_MIN, self.key.chunk_seed)
  114. if name in manifest.archives:
  115. raise self.AlreadyExists(name)
  116. self.last_checkpoint = time.time()
  117. i = 0
  118. while True:
  119. self.checkpoint_name = '%s.checkpoint%s' % (name, i and ('.%d' % i) or '')
  120. if not self.checkpoint_name in manifest.archives:
  121. break
  122. i += 1
  123. else:
  124. if name not in self.manifest.archives:
  125. raise self.DoesNotExist(name)
  126. info = self.manifest.archives[name]
  127. self.load(info[b'id'])
  128. def load(self, id):
  129. self.id = id
  130. data = self.key.decrypt(self.id, self.repository.get(self.id))
  131. self.metadata = msgpack.unpackb(data)
  132. if self.metadata[b'version'] != 1:
  133. raise Exception('Unknown archive metadata version')
  134. decode_dict(self.metadata, (b'name', b'hostname', b'username', b'time'))
  135. self.metadata[b'cmdline'] = [arg.decode('utf-8', 'surrogateescape') for arg in self.metadata[b'cmdline']]
  136. self.name = self.metadata[b'name']
  137. @property
  138. def ts(self):
  139. """Timestamp of archive creation in UTC"""
  140. return parse_timestamp(self.metadata[b'time'])
  141. def __repr__(self):
  142. return 'Archive(%r)' % self.name
  143. def iter_items(self, filter=None, preload=False):
  144. for item in self.pipeline.unpack_many(self.metadata[b'items'], filter=filter, preload=preload):
  145. yield item
  146. def add_item(self, item):
  147. self.items_buffer.add(item)
  148. if time.time() - self.last_checkpoint > self.checkpoint_interval:
  149. self.write_checkpoint()
  150. self.last_checkpoint = time.time()
  151. def write_checkpoint(self):
  152. self.save(self.checkpoint_name)
  153. del self.manifest.archives[self.checkpoint_name]
  154. self.cache.chunk_decref(self.id, self.stats)
  155. def save(self, name=None):
  156. name = name or self.name
  157. if name in self.manifest.archives:
  158. raise self.AlreadyExists(name)
  159. self.items_buffer.flush(flush=True)
  160. metadata = StableDict({
  161. 'version': 1,
  162. 'name': name,
  163. 'items': self.items_buffer.chunks,
  164. 'cmdline': sys.argv,
  165. 'hostname': socket.gethostname(),
  166. 'username': getuser(),
  167. 'time': datetime.utcnow().isoformat(),
  168. })
  169. data = msgpack.packb(metadata, unicode_errors='surrogateescape')
  170. self.id = self.key.id_hash(data)
  171. self.cache.add_chunk(self.id, data, self.stats)
  172. self.manifest.archives[name] = {'id': self.id, 'time': metadata['time']}
  173. self.manifest.write()
  174. self.repository.commit()
  175. self.cache.commit()
  176. def calc_stats(self, cache):
  177. def add(id):
  178. count, size, csize = cache.chunks[id]
  179. stats.update(size, csize, count == 1)
  180. cache.chunks[id] = count - 1, size, csize
  181. def add_file_chunks(chunks):
  182. for id, _, _ in chunks:
  183. add(id)
  184. # This function is a bit evil since it abuses the cache to calculate
  185. # the stats. The cache transaction must be rolled back afterwards
  186. unpacker = msgpack.Unpacker(use_list=False)
  187. cache.begin_txn()
  188. stats = Statistics()
  189. add(self.id)
  190. for id, chunk in zip(self.metadata[b'items'], self.repository.get_many(self.metadata[b'items'])):
  191. add(id)
  192. unpacker.feed(self.key.decrypt(id, chunk))
  193. for item in unpacker:
  194. if b'chunks' in item:
  195. stats.nfiles += 1
  196. add_file_chunks(item[b'chunks'])
  197. cache.rollback()
  198. return stats
  199. def extract_item(self, item, restore_attrs=True, dry_run=False):
  200. if dry_run:
  201. if b'chunks' in item:
  202. for _ in self.pipeline.fetch_many([c[0] for c in item[b'chunks']], is_preloaded=True):
  203. pass
  204. return
  205. dest = self.cwd
  206. if item[b'path'].startswith('/') or item[b'path'].startswith('..'):
  207. raise Exception('Path should be relative and local')
  208. path = os.path.join(dest, item[b'path'])
  209. # Attempt to remove existing files, ignore errors on failure
  210. try:
  211. st = os.lstat(path)
  212. if stat.S_ISDIR(st.st_mode):
  213. os.rmdir(path)
  214. else:
  215. os.unlink(path)
  216. except OSError:
  217. pass
  218. mode = item[b'mode']
  219. if stat.S_ISDIR(mode):
  220. if not os.path.exists(path):
  221. os.makedirs(path)
  222. if restore_attrs:
  223. self.restore_attrs(path, item)
  224. elif stat.S_ISREG(mode):
  225. if not os.path.exists(os.path.dirname(path)):
  226. os.makedirs(os.path.dirname(path))
  227. # Hard link?
  228. if b'source' in item:
  229. source = os.path.join(dest, item[b'source'])
  230. if os.path.exists(path):
  231. os.unlink(path)
  232. os.link(source, path)
  233. else:
  234. with open(path, 'wb') as fd:
  235. ids = [c[0] for c in item[b'chunks']]
  236. for data in self.pipeline.fetch_many(ids, is_preloaded=True):
  237. fd.write(data)
  238. fd.flush()
  239. self.restore_attrs(path, item, fd=fd.fileno())
  240. elif stat.S_ISFIFO(mode):
  241. if not os.path.exists(os.path.dirname(path)):
  242. os.makedirs(os.path.dirname(path))
  243. os.mkfifo(path)
  244. self.restore_attrs(path, item)
  245. elif stat.S_ISLNK(mode):
  246. if not os.path.exists(os.path.dirname(path)):
  247. os.makedirs(os.path.dirname(path))
  248. source = item[b'source']
  249. if os.path.exists(path):
  250. os.unlink(path)
  251. os.symlink(source, path)
  252. self.restore_attrs(path, item, symlink=True)
  253. elif stat.S_ISCHR(mode) or stat.S_ISBLK(mode):
  254. os.mknod(path, item[b'mode'], item[b'rdev'])
  255. self.restore_attrs(path, item)
  256. else:
  257. raise Exception('Unknown archive item type %r' % item[b'mode'])
  258. def restore_attrs(self, path, item, symlink=False, fd=None):
  259. xattrs = item.get(b'xattrs')
  260. if xattrs:
  261. for k, v in xattrs.items():
  262. try:
  263. xattr.setxattr(fd or path, k, v, follow_symlinks=False)
  264. except OSError as e:
  265. if e.errno != errno.ENOTSUP:
  266. raise
  267. uid = gid = None
  268. if not self.numeric_owner:
  269. uid = user2uid(item[b'user'])
  270. gid = group2gid(item[b'group'])
  271. uid = item[b'uid'] if uid is None else uid
  272. gid = item[b'gid'] if gid is None else gid
  273. # This code is a bit of a mess due to os specific differences
  274. try:
  275. if fd:
  276. os.fchown(fd, uid, gid)
  277. else:
  278. os.lchown(path, uid, gid)
  279. except OSError:
  280. pass
  281. if fd:
  282. os.fchmod(fd, item[b'mode'])
  283. elif not symlink:
  284. os.chmod(path, item[b'mode'])
  285. elif has_lchmod: # Not available on Linux
  286. os.lchmod(path, item[b'mode'])
  287. mtime = bigint_to_int(item[b'mtime'])
  288. if fd and utime_supports_fd: # Python >= 3.3
  289. os.utime(fd, None, ns=(mtime, mtime))
  290. elif utime_supports_follow_symlinks: # Python >= 3.3
  291. os.utime(path, None, ns=(mtime, mtime), follow_symlinks=False)
  292. elif not symlink:
  293. os.utime(path, (mtime / 1e9, mtime / 1e9))
  294. acl_set(path, item, self.numeric_owner)
  295. # Only available on OS X and FreeBSD
  296. if has_lchflags and b'bsdflags' in item:
  297. try:
  298. os.lchflags(path, item[b'bsdflags'])
  299. except OSError:
  300. pass
  301. def delete(self, stats):
  302. unpacker = msgpack.Unpacker(use_list=False)
  303. for items_id, data in zip(self.metadata[b'items'], self.repository.get_many(self.metadata[b'items'])):
  304. unpacker.feed(self.key.decrypt(items_id, data))
  305. self.cache.chunk_decref(items_id, stats)
  306. for item in unpacker:
  307. if b'chunks' in item:
  308. for chunk_id, size, csize in item[b'chunks']:
  309. self.cache.chunk_decref(chunk_id, stats)
  310. self.cache.chunk_decref(self.id, stats)
  311. del self.manifest.archives[self.name]
  312. def stat_attrs(self, st, path):
  313. item = {
  314. b'mode': st.st_mode,
  315. b'uid': st.st_uid, b'user': uid2user(st.st_uid),
  316. b'gid': st.st_gid, b'group': gid2group(st.st_gid),
  317. b'mtime': int_to_bigint(st_mtime_ns(st))
  318. }
  319. if self.numeric_owner:
  320. item[b'user'] = item[b'group'] = None
  321. xattrs = xattr.get_all(path, follow_symlinks=False)
  322. if xattrs:
  323. item[b'xattrs'] = StableDict(xattrs)
  324. if has_lchflags and st.st_flags:
  325. item[b'bsdflags'] = st.st_flags
  326. acl_get(path, item, st, self.numeric_owner)
  327. return item
  328. def process_item(self, path, st):
  329. item = {b'path': make_path_safe(path)}
  330. item.update(self.stat_attrs(st, path))
  331. self.add_item(item)
  332. def process_dev(self, path, st):
  333. item = {b'path': make_path_safe(path), b'rdev': st.st_rdev}
  334. item.update(self.stat_attrs(st, path))
  335. self.add_item(item)
  336. def process_symlink(self, path, st):
  337. source = os.readlink(path)
  338. item = {b'path': make_path_safe(path), b'source': source}
  339. item.update(self.stat_attrs(st, path))
  340. self.add_item(item)
  341. def process_file(self, path, st, cache):
  342. safe_path = make_path_safe(path)
  343. # Is it a hard link?
  344. if st.st_nlink > 1:
  345. source = self.hard_links.get((st.st_ino, st.st_dev))
  346. if (st.st_ino, st.st_dev) in self.hard_links:
  347. item = self.stat_attrs(st, path)
  348. item.update({b'path': safe_path, b'source': source})
  349. self.add_item(item)
  350. return
  351. else:
  352. self.hard_links[st.st_ino, st.st_dev] = safe_path
  353. path_hash = self.key.id_hash(os.path.join(self.cwd, path).encode('utf-8', 'surrogateescape'))
  354. ids = cache.file_known_and_unchanged(path_hash, st)
  355. chunks = None
  356. if ids is not None:
  357. # Make sure all ids are available
  358. for id_ in ids:
  359. if not cache.seen_chunk(id_):
  360. break
  361. else:
  362. chunks = [cache.chunk_incref(id_, self.stats) for id_ in ids]
  363. # Only chunkify the file if needed
  364. if chunks is None:
  365. with Archive._open_rb(path, st) as fd:
  366. chunks = []
  367. for chunk in self.chunker.chunkify(fd):
  368. chunks.append(cache.add_chunk(self.key.id_hash(chunk), chunk, self.stats))
  369. cache.memorize_file(path_hash, st, [c[0] for c in chunks])
  370. item = {b'path': safe_path, b'chunks': chunks}
  371. item.update(self.stat_attrs(st, path))
  372. self.stats.nfiles += 1
  373. self.add_item(item)
  374. @staticmethod
  375. def list_archives(repository, key, manifest, cache=None):
  376. for name, info in manifest.archives.items():
  377. yield Archive(repository, key, manifest, name, cache=cache)
  378. @staticmethod
  379. def _open_rb(path, st):
  380. flags_noatime = None
  381. euid = None
  382. def open_simple(p, s):
  383. return open(p, 'rb')
  384. def open_noatime_if_owner(p, s):
  385. if s.st_uid == euid:
  386. return os.fdopen(os.open(p, flags_noatime), 'rb')
  387. else:
  388. return open(p, 'rb')
  389. def open_noatime(p, s):
  390. try:
  391. fd = os.open(p, flags_noatime)
  392. except PermissionError:
  393. # Was this EPERM due to the O_NOATIME flag?
  394. fo = open(p, 'rb')
  395. # Yes, it was -- otherwise the above line would have thrown
  396. # another exception.
  397. euid = os.geteuid()
  398. # So in future, let's check whether the file is owned by us
  399. # before attempting to use O_NOATIME.
  400. Archive._open_rb = open_noatime_if_owner
  401. return fo
  402. return os.fdopen(fd, 'rb')
  403. o_noatime = getattr(os, 'O_NOATIME', None)
  404. if o_noatime is not None:
  405. flags_noatime = os.O_RDONLY | getattr(os, 'O_BINARY', 0) | o_noatime
  406. # Always use O_NOATIME version.
  407. Archive._open_rb = open_noatime
  408. else:
  409. # Always use non-O_NOATIME version.
  410. Archive._open_rb = open_simple
  411. return Archive._open_rb(path, st)
  412. class RobustUnpacker():
  413. """A restartable/robust version of the streaming msgpack unpacker
  414. """
  415. item_keys = [msgpack.packb(name) for name in ('path', 'mode', 'source', 'chunks', 'rdev', 'xattrs', 'user', 'group', 'uid', 'gid', 'mtime')]
  416. def __init__(self, validator):
  417. super(RobustUnpacker, self).__init__()
  418. self.validator = validator
  419. self._buffered_data = []
  420. self._resync = False
  421. self._unpacker = msgpack.Unpacker(object_hook=StableDict)
  422. def resync(self):
  423. self._buffered_data = []
  424. self._resync = True
  425. def feed(self, data):
  426. if self._resync:
  427. self._buffered_data.append(data)
  428. else:
  429. self._unpacker.feed(data)
  430. def __iter__(self):
  431. return self
  432. def __next__(self):
  433. if self._resync:
  434. data = b''.join(self._buffered_data)
  435. while self._resync:
  436. if not data:
  437. raise StopIteration
  438. # Abort early if the data does not look like a serialized dict
  439. if len(data) < 2 or ((data[0] & 0xf0) != 0x80) or ((data[1] & 0xe0) != 0xa0):
  440. data = data[1:]
  441. continue
  442. # Make sure it looks like an item dict
  443. for pattern in self.item_keys:
  444. if data[1:].startswith(pattern):
  445. break
  446. else:
  447. data = data[1:]
  448. continue
  449. self._unpacker = msgpack.Unpacker(object_hook=StableDict)
  450. self._unpacker.feed(data)
  451. try:
  452. item = next(self._unpacker)
  453. if self.validator(item):
  454. self._resync = False
  455. return item
  456. # Ignore exceptions that might be raised when feeding
  457. # msgpack with invalid data
  458. except (TypeError, ValueError, StopIteration):
  459. pass
  460. data = data[1:]
  461. else:
  462. return next(self._unpacker)
  463. class ArchiveChecker:
  464. def __init__(self):
  465. self.error_found = False
  466. self.possibly_superseded = set()
  467. self.tmpdir = tempfile.mkdtemp()
  468. def __del__(self):
  469. shutil.rmtree(self.tmpdir)
  470. def check(self, repository, repair=False):
  471. self.report_progress('Starting archive consistency check...')
  472. self.repair = repair
  473. self.repository = repository
  474. self.init_chunks()
  475. self.key = self.identify_key(repository)
  476. if not Manifest.MANIFEST_ID in self.chunks:
  477. self.manifest = self.rebuild_manifest()
  478. else:
  479. self.manifest, _ = Manifest.load(repository, key=self.key)
  480. self.rebuild_refcounts()
  481. self.verify_chunks()
  482. if not self.error_found:
  483. self.report_progress('Archive consistency check complete, no problems found.')
  484. return self.repair or not self.error_found
  485. def init_chunks(self):
  486. """Fetch a list of all object keys from repository
  487. """
  488. # Explicity set the initial hash table capacity to avoid performance issues
  489. # due to hash table "resonance"
  490. capacity = int(len(self.repository) * 1.2)
  491. self.chunks = ChunkIndex(capacity)
  492. marker = None
  493. while True:
  494. result = self.repository.list(limit=10000, marker=marker)
  495. if not result:
  496. break
  497. marker = result[-1]
  498. for id_ in result:
  499. self.chunks[id_] = (0, 0, 0)
  500. def report_progress(self, msg, error=False):
  501. if error:
  502. self.error_found = True
  503. print(msg, file=sys.stderr if error else sys.stdout)
  504. def identify_key(self, repository):
  505. cdata = repository.get(next(self.chunks.iteritems())[0])
  506. return key_factory(repository, cdata)
  507. def rebuild_manifest(self):
  508. """Rebuild the manifest object if it is missing
  509. Iterates through all objects in the repository looking for archive metadata blocks.
  510. """
  511. self.report_progress('Rebuilding missing manifest, this might take some time...', error=True)
  512. manifest = Manifest(self.key, self.repository)
  513. for chunk_id, _ in self.chunks.iteritems():
  514. cdata = self.repository.get(chunk_id)
  515. data = self.key.decrypt(chunk_id, cdata)
  516. # Some basic sanity checks of the payload before feeding it into msgpack
  517. if len(data) < 2 or ((data[0] & 0xf0) != 0x80) or ((data[1] & 0xe0) != 0xa0):
  518. continue
  519. if not b'cmdline' in data or not b'\xa7version\x01' in data:
  520. continue
  521. try:
  522. archive = msgpack.unpackb(data)
  523. except:
  524. continue
  525. if isinstance(archive, dict) and b'items' in archive and b'cmdline' in archive:
  526. self.report_progress('Found archive ' + archive[b'name'].decode('utf-8'), error=True)
  527. manifest.archives[archive[b'name'].decode('utf-8')] = {b'id': chunk_id, b'time': archive[b'time']}
  528. self.report_progress('Manifest rebuild complete', error=True)
  529. return manifest
  530. def rebuild_refcounts(self):
  531. """Rebuild object reference counts by walking the metadata
  532. Missing and/or incorrect data is repaired when detected
  533. """
  534. # Exclude the manifest from chunks
  535. del self.chunks[Manifest.MANIFEST_ID]
  536. def mark_as_possibly_superseded(id_):
  537. if self.chunks.get(id_, (0,))[0] == 0:
  538. self.possibly_superseded.add(id_)
  539. def add_callback(chunk):
  540. id_ = self.key.id_hash(chunk)
  541. cdata = self.key.encrypt(chunk)
  542. add_reference(id_, len(chunk), len(cdata), cdata)
  543. return id_
  544. def add_reference(id_, size, csize, cdata=None):
  545. try:
  546. count, _, _ = self.chunks[id_]
  547. self.chunks[id_] = count + 1, size, csize
  548. except KeyError:
  549. assert cdata is not None
  550. self.chunks[id_] = 1, size, csize
  551. if self.repair:
  552. self.repository.put(id_, cdata)
  553. def verify_file_chunks(item):
  554. """Verifies that all file chunks are present
  555. Missing file chunks will be replaced with new chunks of the same
  556. length containing all zeros.
  557. """
  558. offset = 0
  559. chunk_list = []
  560. for chunk_id, size, csize in item[b'chunks']:
  561. if not chunk_id in self.chunks:
  562. # If a file chunk is missing, create an all empty replacement chunk
  563. self.report_progress('{}: Missing file chunk detected (Byte {}-{})'.format(item[b'path'].decode('utf-8', 'surrogateescape'), offset, offset + size), error=True)
  564. data = bytes(size)
  565. chunk_id = self.key.id_hash(data)
  566. cdata = self.key.encrypt(data)
  567. csize = len(cdata)
  568. add_reference(chunk_id, size, csize, cdata)
  569. else:
  570. add_reference(chunk_id, size, csize)
  571. chunk_list.append((chunk_id, size, csize))
  572. offset += size
  573. item[b'chunks'] = chunk_list
  574. def robust_iterator(archive):
  575. """Iterates through all archive items
  576. Missing item chunks will be skipped and the msgpack stream will be restarted
  577. """
  578. unpacker = RobustUnpacker(lambda item: isinstance(item, dict) and b'path' in item)
  579. _state = 0
  580. def missing_chunk_detector(chunk_id):
  581. nonlocal _state
  582. if _state % 2 != int(not chunk_id in self.chunks):
  583. _state += 1
  584. return _state
  585. for state, items in groupby(archive[b'items'], missing_chunk_detector):
  586. items = list(items)
  587. if state % 2:
  588. self.report_progress('Archive metadata damage detected', error=True)
  589. continue
  590. if state > 0:
  591. unpacker.resync()
  592. for chunk_id, cdata in zip(items, repository.get_many(items)):
  593. unpacker.feed(self.key.decrypt(chunk_id, cdata))
  594. for item in unpacker:
  595. yield item
  596. repository = cache_if_remote(self.repository)
  597. num_archives = len(self.manifest.archives)
  598. for i, (name, info) in enumerate(list(self.manifest.archives.items()), 1):
  599. self.report_progress('Analyzing archive {} ({}/{})'.format(name, i, num_archives))
  600. archive_id = info[b'id']
  601. if not archive_id in self.chunks:
  602. self.report_progress('Archive metadata block is missing', error=True)
  603. del self.manifest.archives[name]
  604. continue
  605. mark_as_possibly_superseded(archive_id)
  606. cdata = self.repository.get(archive_id)
  607. data = self.key.decrypt(archive_id, cdata)
  608. archive = StableDict(msgpack.unpackb(data))
  609. if archive[b'version'] != 1:
  610. raise Exception('Unknown archive metadata version')
  611. decode_dict(archive, (b'name', b'hostname', b'username', b'time')) # fixme: argv
  612. items_buffer = ChunkBuffer(self.key)
  613. items_buffer.write_chunk = add_callback
  614. for item in robust_iterator(archive):
  615. if b'chunks' in item:
  616. verify_file_chunks(item)
  617. items_buffer.add(item)
  618. items_buffer.flush(flush=True)
  619. for previous_item_id in archive[b'items']:
  620. mark_as_possibly_superseded(previous_item_id)
  621. archive[b'items'] = items_buffer.chunks
  622. data = msgpack.packb(archive, unicode_errors='surrogateescape')
  623. new_archive_id = self.key.id_hash(data)
  624. cdata = self.key.encrypt(data)
  625. add_reference(new_archive_id, len(data), len(cdata), cdata)
  626. info[b'id'] = new_archive_id
  627. def verify_chunks(self):
  628. unused = set()
  629. for id_, (count, size, csize) in self.chunks.iteritems():
  630. if count == 0:
  631. unused.add(id_)
  632. orphaned = unused - self.possibly_superseded
  633. if orphaned:
  634. self.report_progress('{} orphaned objects found'.format(len(orphaned)), error=True)
  635. if self.repair:
  636. for id_ in unused:
  637. self.repository.delete(id_)
  638. self.manifest.write()
  639. self.repository.commit()