archive.py 27 KB

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