backup.py 32 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877
  1. #
  2. # The Qubes OS Project, http://www.qubes-os.org
  3. #
  4. # Copyright (C) 2013-2017 Marek Marczykowski-Górecki
  5. # <marmarek@invisiblethingslab.com>
  6. # Copyright (C) 2013 Olivier Médoc <o_medoc@yahoo.fr>
  7. #
  8. # This library is free software; you can redistribute it and/or
  9. # modify it under the terms of the GNU Lesser General Public
  10. # License as published by the Free Software Foundation; either
  11. # version 2.1 of the License, or (at your option) any later version.
  12. #
  13. # This library is distributed in the hope that it will be useful,
  14. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  15. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  16. # Lesser General Public License for more details.
  17. #
  18. # You should have received a copy of the GNU Lesser General Public
  19. # License along with this library; if not, see <https://www.gnu.org/licenses/>.
  20. #
  21. #
  22. from __future__ import unicode_literals
  23. import asyncio
  24. import datetime
  25. import fcntl
  26. import functools
  27. import grp
  28. import itertools
  29. import logging
  30. import os
  31. import pwd
  32. import re
  33. import shutil
  34. import stat
  35. import string
  36. import subprocess
  37. import tempfile
  38. import termios
  39. import time
  40. from .utils import size_to_human
  41. import qubes
  42. import qubes.storage
  43. import qubes.storage.file
  44. import qubes.vm.templatevm
  45. QUEUE_ERROR = "ERROR"
  46. QUEUE_FINISHED = "FINISHED"
  47. HEADER_FILENAME = 'backup-header'
  48. DEFAULT_CRYPTO_ALGORITHM = 'aes-256-cbc'
  49. # 'scrypt' is not exactly HMAC algorithm, but a tool we use to
  50. # integrity-protect the data
  51. DEFAULT_HMAC_ALGORITHM = 'scrypt'
  52. DEFAULT_COMPRESSION_FILTER = 'gzip'
  53. CURRENT_BACKUP_FORMAT_VERSION = '4'
  54. # Maximum size of error message get from process stderr (including VM process)
  55. MAX_STDERR_BYTES = 1024
  56. # header + qubes.xml max size
  57. HEADER_QUBES_XML_MAX_SIZE = 1024 * 1024
  58. # hmac file max size - regardless of backup format version!
  59. HMAC_MAX_SIZE = 4096
  60. BLKSIZE = 512
  61. _re_alphanum = re.compile(r'^[A-Za-z0-9-]*$')
  62. class BackupCanceledError(qubes.exc.QubesException):
  63. def __init__(self, msg, tmpdir=None):
  64. super(BackupCanceledError, self).__init__(msg)
  65. self.tmpdir = tmpdir
  66. class BackupHeader:
  67. '''Structure describing backup-header file included as the first file in
  68. backup archive
  69. '''
  70. # pylint: disable=too-few-public-methods
  71. header_keys = {
  72. 'version': 'version',
  73. 'encrypted': 'encrypted',
  74. 'compressed': 'compressed',
  75. 'compression-filter': 'compression_filter',
  76. 'crypto-algorithm': 'crypto_algorithm',
  77. 'hmac-algorithm': 'hmac_algorithm',
  78. 'backup-id': 'backup_id'
  79. }
  80. bool_options = ['encrypted', 'compressed']
  81. int_options = ['version']
  82. def __init__(self,
  83. version=None,
  84. encrypted=None,
  85. compressed=None,
  86. compression_filter=None,
  87. hmac_algorithm=None,
  88. crypto_algorithm=None,
  89. backup_id=None):
  90. # repeat the list to help code completion...
  91. self.version = version
  92. self.encrypted = encrypted
  93. self.compressed = compressed
  94. # Options introduced in backup format 3+, which always have a header,
  95. # so no need for fallback in function parameter
  96. self.compression_filter = compression_filter
  97. self.hmac_algorithm = hmac_algorithm
  98. self.crypto_algorithm = crypto_algorithm
  99. self.backup_id = backup_id
  100. def save(self, filename):
  101. with open(filename, "w") as f_header:
  102. # make sure 'version' is the first key
  103. f_header.write('version={}\n'.format(self.version))
  104. for key, attr in self.header_keys.items():
  105. if key == 'version':
  106. continue
  107. if getattr(self, attr) is None:
  108. continue
  109. f_header.write("{!s}={!s}\n".format(key, getattr(self, attr)))
  110. class SendWorker:
  111. # pylint: disable=too-few-public-methods
  112. def __init__(self, queue, base_dir, backup_stdout):
  113. super(SendWorker, self).__init__()
  114. self.queue = queue
  115. self.base_dir = base_dir
  116. self.backup_stdout = backup_stdout
  117. self.log = logging.getLogger('qubes.backup')
  118. @asyncio.coroutine
  119. def run(self):
  120. self.log.debug("Started sending thread")
  121. while True:
  122. filename = yield from self.queue.get()
  123. if filename in (QUEUE_FINISHED, QUEUE_ERROR):
  124. break
  125. self.log.debug("Sending file {}".format(filename))
  126. # This tar used for sending data out need to be as simple, as
  127. # simple, as featureless as possible. It will not be
  128. # verified before untaring.
  129. tar_final_cmd = ["tar", "-cO", "--posix",
  130. "-C", self.base_dir, filename]
  131. # pylint: disable=not-an-iterable
  132. final_proc = yield from asyncio.create_subprocess_exec(
  133. *tar_final_cmd,
  134. stdout=self.backup_stdout)
  135. retcode = yield from final_proc.wait()
  136. if retcode >= 2:
  137. # handle only exit code 2 (tar fatal error) or
  138. # greater (call failed?)
  139. raise qubes.exc.QubesException(
  140. "ERROR: Failed to write the backup, out of disk space? "
  141. "Check console output or ~/.xsession-errors for details.")
  142. # Delete the file as we don't need it anymore
  143. self.log.debug("Removing file {}".format(filename))
  144. os.remove(os.path.join(self.base_dir, filename))
  145. self.log.debug("Finished sending thread")
  146. @asyncio.coroutine
  147. def launch_proc_with_pty(args, stdin=None, stdout=None, stderr=None, echo=True):
  148. """Similar to pty.fork, but handle stdin/stdout according to parameters
  149. instead of connecting to the pty
  150. :return tuple (subprocess.Popen, pty_master)
  151. """
  152. def set_ctty(ctty_fd, master_fd):
  153. os.setsid()
  154. os.close(master_fd)
  155. fcntl.ioctl(ctty_fd, termios.TIOCSCTTY, 0)
  156. if not echo:
  157. termios_p = termios.tcgetattr(ctty_fd)
  158. # termios_p.c_lflags
  159. termios_p[3] &= ~termios.ECHO
  160. termios.tcsetattr(ctty_fd, termios.TCSANOW, termios_p)
  161. (pty_master, pty_slave) = os.openpty()
  162. # pylint: disable=not-an-iterable
  163. p = yield from asyncio.create_subprocess_exec(*args,
  164. stdin=stdin,
  165. stdout=stdout,
  166. stderr=stderr,
  167. preexec_fn=lambda: set_ctty(pty_slave, pty_master))
  168. os.close(pty_slave)
  169. return p, open(pty_master, 'wb+', buffering=0)
  170. @asyncio.coroutine
  171. def launch_scrypt(action, input_name, output_name, passphrase):
  172. '''
  173. Launch 'scrypt' process, pass passphrase to it and return
  174. subprocess.Popen object.
  175. :param action: 'enc' or 'dec'
  176. :param input_name: input path or '-' for stdin
  177. :param output_name: output path or '-' for stdout
  178. :param passphrase: passphrase
  179. :type passphrase: bytes
  180. :return: subprocess.Popen object
  181. '''
  182. command_line = ['scrypt', action, input_name, output_name]
  183. (p, pty) = yield from launch_proc_with_pty(command_line,
  184. stdin=subprocess.PIPE if input_name == '-' else None,
  185. stdout=subprocess.PIPE if output_name == '-' else None,
  186. stderr=subprocess.PIPE,
  187. echo=False)
  188. if action == 'enc':
  189. prompts = (b'Please enter passphrase: ', b'Please confirm passphrase: ')
  190. else:
  191. prompts = (b'Please enter passphrase: ',)
  192. for prompt in prompts:
  193. actual_prompt = yield from p.stderr.read(len(prompt))
  194. if actual_prompt != prompt:
  195. raise qubes.exc.QubesException(
  196. 'Unexpected prompt from scrypt: {}'.format(actual_prompt))
  197. pty.write(passphrase + b'\n')
  198. pty.flush()
  199. # save it here, so garbage collector would not close it (which would kill
  200. # the child)
  201. p.pty = pty
  202. return p
  203. class Backup:
  204. '''Backup operation manager. Usage:
  205. >>> app = qubes.Qubes()
  206. >>> # optional - you can use 'None' to use default list (based on
  207. >>> # vm.include_in_backups property)
  208. >>> vms = [app.domains[name] for name in ['my-vm1', 'my-vm2', 'my-vm3']]
  209. >>> exclude_vms = []
  210. >>> options = {
  211. >>> 'encrypted': True,
  212. >>> 'compressed': True,
  213. >>> 'passphrase': 'This is very weak backup passphrase',
  214. >>> 'target_vm': app.domains['sys-usb'],
  215. >>> 'target_dir': '/media/disk',
  216. >>> }
  217. >>> backup_op = Backup(app, vms, exclude_vms, **options)
  218. >>> print(backup_op.get_backup_summary())
  219. >>> asyncio.get_event_loop().run_until_complete(backup_op.backup_do())
  220. See attributes of this object for all available options.
  221. '''
  222. # pylint: disable=too-many-instance-attributes
  223. class FileToBackup:
  224. # pylint: disable=too-few-public-methods
  225. def __init__(self, file_path, subdir=None, name=None, size=None):
  226. if size is None:
  227. size = qubes.storage.file.get_disk_usage(file_path)
  228. if subdir is None:
  229. abs_file_path = os.path.abspath(file_path)
  230. abs_base_dir = os.path.abspath(
  231. qubes.config.system_path["qubes_base_dir"]) + '/'
  232. abs_file_dir = os.path.dirname(abs_file_path) + '/'
  233. (nothing, directory, subdir) = \
  234. abs_file_dir.partition(abs_base_dir)
  235. assert nothing == ""
  236. assert directory == abs_base_dir
  237. else:
  238. if subdir and not subdir.endswith('/'):
  239. subdir += '/'
  240. #: real path to the file
  241. self.path = file_path
  242. #: size of the file
  243. self.size = size
  244. #: directory in backup archive where file should be placed
  245. self.subdir = subdir
  246. #: use this name in the archive (aka rename)
  247. self.name = os.path.basename(file_path)
  248. if name is not None:
  249. self.name = name
  250. class VMToBackup:
  251. # pylint: disable=too-few-public-methods
  252. def __init__(self, vm, files, subdir):
  253. self.vm = vm
  254. self.files = files
  255. self.subdir = subdir
  256. @property
  257. def size(self):
  258. return functools.reduce(lambda x, y: x + y.size, self.files, 0)
  259. def __init__(self, app, vms_list=None, exclude_list=None, **kwargs):
  260. """
  261. If vms = None, use default list based on vm.include_in_backups property;
  262. exclude_list is always applied
  263. """
  264. super(Backup, self).__init__()
  265. #: progress of the backup - bytes handled of the current VM
  266. self.chunk_size = 100 * 1024 * 1024
  267. self._current_vm_bytes = 0
  268. #: progress of the backup - bytes handled of finished VMs
  269. self._done_vms_bytes = 0
  270. #: total backup size (set by :py:meth:`get_files_to_backup`)
  271. self.total_backup_bytes = 0
  272. #: application object
  273. self.app = app
  274. #: directory for temporary files - set after creating the directory
  275. self.tmpdir = None
  276. # Backup settings - defaults
  277. #: should the backup be compressed?
  278. self.compressed = True
  279. #: what passphrase should be used to intergrity protect (and encrypt)
  280. #: the backup; required
  281. self.passphrase = None
  282. #: custom compression filter; a program which process stdin to stdout
  283. self.compression_filter = DEFAULT_COMPRESSION_FILTER
  284. #: VM to which backup should be sent (if any)
  285. self.target_vm = None
  286. #: directory to save backup in (either in dom0 or target VM,
  287. #: depending on :py:attr:`target_vm`
  288. self.target_dir = None
  289. #: callback for progress reporting. Will be called with one argument
  290. #: - progress in percents
  291. self.progress_callback = None
  292. self.last_progress_time = time.time()
  293. #: backup ID, needs to be unique (for a given user),
  294. #: not necessary unpredictable; automatically generated
  295. self.backup_id = datetime.datetime.now().strftime(
  296. '%Y%m%dT%H%M%S-' + str(os.getpid()))
  297. for key, value in kwargs.items():
  298. if hasattr(self, key):
  299. setattr(self, key, value)
  300. else:
  301. raise AttributeError(key)
  302. self.log = logging.getLogger('qubes.backup')
  303. if exclude_list is None:
  304. exclude_list = []
  305. if vms_list is None:
  306. vms_list = [vm for vm in app.domains if vm.include_in_backups]
  307. # Apply exclude list
  308. self.vms_for_backup = [vm for vm in vms_list
  309. if vm.name not in exclude_list]
  310. self._files_to_backup = self.get_files_to_backup()
  311. def __del__(self):
  312. if self.tmpdir and os.path.exists(self.tmpdir):
  313. shutil.rmtree(self.tmpdir)
  314. def get_files_to_backup(self):
  315. files_to_backup = {}
  316. for vm in self.vms_for_backup:
  317. if vm.qid == 0:
  318. # handle dom0 later
  319. continue
  320. subdir = 'vm%d/' % vm.qid
  321. vm_files = []
  322. for name, volume in vm.volumes.items():
  323. if not volume.save_on_stop:
  324. continue
  325. vm_files.append(self.FileToBackup(
  326. volume.export(),
  327. subdir,
  328. name + '.img',
  329. volume.usage))
  330. vm_files.extend(self.FileToBackup(i, subdir)
  331. for i in vm.fire_event('backup-get-files'))
  332. firewall_conf = os.path.join(vm.dir_path, vm.firewall_conf)
  333. if os.path.exists(firewall_conf):
  334. vm_files.append(self.FileToBackup(firewall_conf, subdir))
  335. if not vm_files:
  336. # subdir/ is needed in the tar file, otherwise restore
  337. # of a (Disp)VM without any backed up files is going
  338. # to fail. Adding a zero-sized file here happens to be
  339. # more straightforward than adding an empty directory.
  340. empty = self.FileToBackup("/var/run/qubes/empty", subdir)
  341. assert empty.size == 0
  342. vm_files.append(empty)
  343. files_to_backup[vm.qid] = self.VMToBackup(vm, vm_files, subdir)
  344. # Dom0 user home
  345. if 0 in [vm.qid for vm in self.vms_for_backup]:
  346. local_user = grp.getgrnam('qubes').gr_mem[0]
  347. home_dir = pwd.getpwnam(local_user).pw_dir
  348. # Home dir should have only user-owned files, so fix it now
  349. # to prevent permissions problems - some root-owned files can
  350. # left after 'sudo bash' and similar commands
  351. subprocess.check_call(['sudo', 'chown', '-R', local_user, home_dir])
  352. home_to_backup = [
  353. self.FileToBackup(home_dir, 'dom0-home/')]
  354. vm_files = home_to_backup
  355. files_to_backup[0] = self.VMToBackup(self.app.domains[0],
  356. vm_files,
  357. os.path.join('dom0-home', os.path.basename(home_dir)))
  358. self.total_backup_bytes = functools.reduce(
  359. lambda x, y: x + y.size, files_to_backup.values(), 0)
  360. return files_to_backup
  361. def get_backup_summary(self):
  362. summary = ""
  363. fields_to_display = [
  364. {"name": "VM", "width": 16},
  365. {"name": "type", "width": 12},
  366. {"name": "size", "width": 12}
  367. ]
  368. # Display the header
  369. for field in fields_to_display:
  370. fmt = "{{0:-^{0}}}-+".format(field["width"] + 1)
  371. summary += fmt.format('-')
  372. summary += "\n"
  373. for field in fields_to_display:
  374. fmt = "{{0:>{0}}} |".format(field["width"] + 1)
  375. summary += fmt.format(field["name"])
  376. summary += "\n"
  377. for field in fields_to_display:
  378. fmt = "{{0:-^{0}}}-+".format(field["width"] + 1)
  379. summary += fmt.format('-')
  380. summary += "\n"
  381. files_to_backup = self._files_to_backup
  382. for qid, vm_info in files_to_backup.items():
  383. summary_line = ""
  384. fmt = "{{0:>{0}}} |".format(fields_to_display[0]["width"] + 1)
  385. summary_line += fmt.format(vm_info.vm.name)
  386. fmt = "{{0:>{0}}} |".format(fields_to_display[1]["width"] + 1)
  387. if qid == 0:
  388. summary_line += fmt.format("User home")
  389. elif isinstance(vm_info.vm, qubes.vm.templatevm.TemplateVM):
  390. summary_line += fmt.format("Template VM")
  391. else:
  392. summary_line += fmt.format("VM" + (" + Sys" if
  393. vm_info.vm.updateable else ""))
  394. vm_size = vm_info.size
  395. fmt = "{{0:>{0}}} |".format(fields_to_display[2]["width"] + 1)
  396. summary_line += fmt.format(size_to_human(vm_size))
  397. if qid != 0 and vm_info.vm.is_running():
  398. summary_line += " <-- The VM is running, backup will contain " \
  399. "its state from before its start!"
  400. summary += summary_line + "\n"
  401. for field in fields_to_display:
  402. fmt = "{{0:-^{0}}}-+".format(field["width"] + 1)
  403. summary += fmt.format('-')
  404. summary += "\n"
  405. fmt = "{{0:>{0}}} |".format(fields_to_display[0]["width"] + 1)
  406. summary += fmt.format("Total size:")
  407. fmt = "{{0:>{0}}} |".format(
  408. fields_to_display[1]["width"] + 1 + 2 + fields_to_display[2][
  409. "width"] + 1)
  410. summary += fmt.format(size_to_human(self.total_backup_bytes))
  411. summary += "\n"
  412. for field in fields_to_display:
  413. fmt = "{{0:-^{0}}}-+".format(field["width"] + 1)
  414. summary += fmt.format('-')
  415. summary += "\n"
  416. vms_not_for_backup = [vm.name for vm in self.app.domains
  417. if vm not in self.vms_for_backup]
  418. summary += "VMs not selected for backup:\n - " + "\n - ".join(
  419. sorted(vms_not_for_backup)) + "\n"
  420. return summary
  421. @asyncio.coroutine
  422. def _prepare_backup_header(self):
  423. header_file_path = os.path.join(self.tmpdir, HEADER_FILENAME)
  424. backup_header = BackupHeader(
  425. version=CURRENT_BACKUP_FORMAT_VERSION,
  426. hmac_algorithm=DEFAULT_HMAC_ALGORITHM,
  427. encrypted=True,
  428. compressed=self.compressed,
  429. compression_filter=self.compression_filter,
  430. backup_id=self.backup_id,
  431. )
  432. backup_header.save(header_file_path)
  433. # Start encrypt, scrypt will also handle integrity
  434. # protection
  435. scrypt_passphrase = '{filename}!'.format(
  436. filename=HEADER_FILENAME).encode() + self.passphrase
  437. scrypt = yield from launch_scrypt(
  438. 'enc', header_file_path, header_file_path + '.hmac',
  439. scrypt_passphrase)
  440. retcode = yield from scrypt.wait()
  441. if retcode:
  442. raise qubes.exc.QubesException(
  443. "Failed to compute hmac of header file: "
  444. + scrypt.stderr.read())
  445. return HEADER_FILENAME, HEADER_FILENAME + ".hmac"
  446. def _send_progress_update(self):
  447. if not self.total_backup_bytes:
  448. return
  449. if callable(self.progress_callback):
  450. if time.time() - self.last_progress_time >= 1: # avoid flooding
  451. progress = (
  452. 100 * (self._done_vms_bytes + self._current_vm_bytes) /
  453. self.total_backup_bytes)
  454. self.last_progress_time = time.time()
  455. # pylint: disable=not-callable
  456. self.progress_callback(progress)
  457. def _add_vm_progress(self, bytes_done):
  458. self._current_vm_bytes += bytes_done
  459. self._send_progress_update()
  460. @asyncio.coroutine
  461. def _split_and_send(self, input_stream, file_basename,
  462. output_queue):
  463. '''Split *input_stream* into parts of max *chunk_size* bytes and send
  464. to *output_queue*.
  465. :param input_stream: stream (asyncio reader stream) of data to split
  466. :param file_basename: basename (i.e. without part number and '.enc')
  467. of output files
  468. :param output_queue: asyncio.Queue instance to put produced files to
  469. - queue will get only filenames of written chunks
  470. '''
  471. # Wait for compressor (tar) process to finish or for any
  472. # error of other subprocesses
  473. i = 0
  474. run_error = "size_limit"
  475. scrypt = None
  476. while run_error == "size_limit":
  477. # Prepare a first chunk
  478. chunkfile = file_basename + ".%03d.enc" % i
  479. i += 1
  480. # Start encrypt, scrypt will also handle integrity
  481. # protection
  482. scrypt_passphrase = \
  483. '{backup_id}!{filename}!'.format(
  484. backup_id=self.backup_id,
  485. filename=os.path.relpath(chunkfile[:-4],
  486. self.tmpdir)).encode() + self.passphrase
  487. try:
  488. scrypt = yield from launch_scrypt(
  489. "enc", "-", chunkfile, scrypt_passphrase)
  490. run_error = yield from handle_streams(
  491. input_stream,
  492. scrypt.stdin,
  493. self.chunk_size,
  494. self._add_vm_progress
  495. )
  496. self.log.debug(
  497. "handle_streams returned: {}".format(run_error))
  498. except:
  499. scrypt.terminate()
  500. raise
  501. scrypt.stdin.close()
  502. yield from scrypt.wait()
  503. self.log.debug("scrypt return code: {}".format(
  504. scrypt.returncode))
  505. # Send the chunk to the backup target
  506. yield from output_queue.put(
  507. os.path.relpath(chunkfile, self.tmpdir))
  508. @asyncio.coroutine
  509. def _wrap_and_send_files(self, files_to_backup, output_queue):
  510. for vm_info in files_to_backup:
  511. for file_info in vm_info.files:
  512. self.log.debug("Backing up {}".format(file_info))
  513. backup_tempfile = os.path.join(
  514. self.tmpdir, file_info.subdir,
  515. file_info.name)
  516. self.log.debug("Using temporary location: {}".format(
  517. backup_tempfile))
  518. # Ensure the temporary directory exists
  519. if not os.path.isdir(os.path.dirname(backup_tempfile)):
  520. os.makedirs(os.path.dirname(backup_tempfile))
  521. # The first tar cmd can use any complex feature as we want.
  522. # Files will be verified before untaring this.
  523. # Prefix the path in archive with filename["subdir"] to have it
  524. # verified during untar
  525. tar_cmdline = (["tar", "-Pc", '--sparse',
  526. '-C', os.path.dirname(file_info.path)] +
  527. (['--dereference'] if
  528. file_info.subdir != "dom0-home/" else []) +
  529. ['--xform=s:^%s:%s\\0:' % (
  530. os.path.basename(file_info.path),
  531. file_info.subdir),
  532. os.path.basename(file_info.path)
  533. ])
  534. file_stat = os.stat(file_info.path)
  535. if stat.S_ISBLK(file_stat.st_mode) or \
  536. file_info.name != os.path.basename(file_info.path):
  537. # tar doesn't handle content of block device, use our
  538. # writer
  539. # also use our tar writer when renaming file
  540. assert not stat.S_ISDIR(file_stat.st_mode), \
  541. "Renaming directories not supported"
  542. tar_cmdline = ['python3', '-m', 'qubes.tarwriter',
  543. '--override-name=%s' % (
  544. os.path.join(file_info.subdir, os.path.basename(
  545. file_info.name))),
  546. file_info.path]
  547. if self.compressed:
  548. tar_cmdline.insert(-2,
  549. "--use-compress-program=%s" % self.compression_filter)
  550. self.log.debug(" ".join(tar_cmdline))
  551. # Pipe: tar-sparse | scrypt | tar | backup_target
  552. # TODO: log handle stderr
  553. # pylint: disable=not-an-iterable
  554. tar_sparse = yield from asyncio.create_subprocess_exec(
  555. *tar_cmdline, stdout=subprocess.PIPE)
  556. try:
  557. yield from self._split_and_send(
  558. tar_sparse.stdout,
  559. backup_tempfile,
  560. output_queue)
  561. except:
  562. try:
  563. tar_sparse.terminate()
  564. except ProcessLookupError:
  565. pass
  566. raise
  567. yield from tar_sparse.wait()
  568. if tar_sparse.returncode:
  569. raise qubes.exc.QubesException(
  570. 'Failed to archive {} file'.format(file_info.path))
  571. # This VM done, update progress
  572. self._done_vms_bytes += vm_info.size
  573. self._current_vm_bytes = 0
  574. self._send_progress_update()
  575. yield from output_queue.put(QUEUE_FINISHED)
  576. @staticmethod
  577. @asyncio.coroutine
  578. def _monitor_process(proc, error_message):
  579. try:
  580. yield from proc.wait()
  581. except:
  582. proc.terminate()
  583. raise
  584. if proc.returncode:
  585. if proc.stderr is not None:
  586. proc_stderr = (yield from proc.stderr.read())
  587. proc_stderr = proc_stderr.decode('ascii', errors='ignore')
  588. proc_stderr = ''.join(
  589. c for c in proc_stderr if c in string.printable and
  590. c not in '\r\n%{}')
  591. error_message += ': ' + proc_stderr
  592. raise qubes.exc.QubesException(error_message)
  593. @staticmethod
  594. @asyncio.coroutine
  595. def _cancel_on_error(future, previous_task):
  596. '''If further element of chain fail, cancel previous one to
  597. avoid deadlock.
  598. When earlier element of chain fail, it will be handled by
  599. :py:meth:`backup_do`.
  600. The chain is:
  601. :py:meth:`_wrap_and_send_files` -> :py:class:`SendWorker` -> vmproc
  602. '''
  603. try:
  604. yield from future
  605. except: # pylint: disable=bare-except
  606. previous_task.cancel()
  607. @asyncio.coroutine
  608. def backup_do(self):
  609. # pylint: disable=too-many-statements
  610. if self.passphrase is None:
  611. raise qubes.exc.QubesException("No passphrase set")
  612. if not isinstance(self.passphrase, bytes):
  613. self.passphrase = self.passphrase.encode('utf-8')
  614. qubes_xml = self.app.store
  615. self.tmpdir = tempfile.mkdtemp()
  616. shutil.copy(qubes_xml, os.path.join(self.tmpdir, 'qubes.xml'))
  617. qubes_xml = os.path.join(self.tmpdir, 'qubes.xml')
  618. backup_app = qubes.Qubes(qubes_xml, offline_mode=True)
  619. backup_app.events_enabled = False
  620. files_to_backup = self._files_to_backup
  621. # make sure backup_content isn't set initially
  622. for vm in backup_app.domains:
  623. vm.events_enabled = False
  624. vm.features['backup-content'] = False
  625. for qid, vm_info in files_to_backup.items():
  626. # VM is included in the backup
  627. backup_app.domains[qid].features['backup-content'] = True
  628. backup_app.domains[qid].features['backup-path'] = vm_info.subdir
  629. backup_app.domains[qid].features['backup-size'] = vm_info.size
  630. backup_app.save()
  631. del backup_app
  632. vmproc = None
  633. if self.target_vm is not None:
  634. # Prepare the backup target (Qubes service call)
  635. # If APPVM, STDOUT is a PIPE
  636. read_fd, write_fd = os.pipe()
  637. vmproc = yield from self.target_vm.run_service('qubes.Backup',
  638. stdin=read_fd,
  639. stderr=subprocess.PIPE,
  640. stdout=subprocess.DEVNULL)
  641. os.close(read_fd)
  642. os.write(write_fd, (self.target_dir.
  643. replace("\r", "").replace("\n", "") + "\n").encode())
  644. backup_stdout = write_fd
  645. else:
  646. # Prepare the backup target (local file)
  647. if os.path.isdir(self.target_dir):
  648. backup_target = self.target_dir + "/qubes-{0}". \
  649. format(time.strftime("%Y-%m-%dT%H%M%S"))
  650. else:
  651. backup_target = self.target_dir
  652. # Create the target directory
  653. if not os.path.exists(os.path.dirname(self.target_dir)):
  654. raise qubes.exc.QubesException(
  655. "ERROR: the backup directory for {0} does not exists".
  656. format(self.target_dir))
  657. # If not APPVM, STDOUT is a local file
  658. backup_stdout = open(backup_target, 'wb')
  659. # Tar with tape length does not deals well with stdout
  660. # (close stdout between two tapes)
  661. # For this reason, we will use named pipes instead
  662. self.log.debug("Working in {}".format(self.tmpdir))
  663. self.log.debug("Will backup: {}".format(files_to_backup))
  664. header_files = yield from self._prepare_backup_header()
  665. # Setup worker to send encrypted data chunks to the backup_target
  666. to_send = asyncio.Queue(10)
  667. send_proc = SendWorker(to_send, self.tmpdir, backup_stdout)
  668. send_task = asyncio.ensure_future(send_proc.run())
  669. vmproc_task = None
  670. if vmproc is not None:
  671. vmproc_task = asyncio.ensure_future(
  672. self._monitor_process(vmproc,
  673. 'Writing backup to VM {} failed'.format(
  674. self.target_vm.name)))
  675. asyncio.ensure_future(self._cancel_on_error(
  676. vmproc_task, send_task))
  677. for file_name in header_files:
  678. yield from to_send.put(file_name)
  679. qubes_xml_info = self.VMToBackup(
  680. None,
  681. [self.FileToBackup(qubes_xml, '')],
  682. ''
  683. )
  684. inner_archive_task = asyncio.ensure_future(
  685. self._wrap_and_send_files(
  686. itertools.chain([qubes_xml_info], files_to_backup.values()),
  687. to_send
  688. ))
  689. asyncio.ensure_future(
  690. self._cancel_on_error(send_task, inner_archive_task))
  691. try:
  692. try:
  693. yield from inner_archive_task
  694. except:
  695. yield from to_send.put(QUEUE_ERROR)
  696. # in fact we may be handling CancelledError, induced by
  697. # exception in send_task or vmproc_task (and propagated by
  698. # self._cancel_on_error call above); in such a case this
  699. # yield from will raise exception, covering CancelledError -
  700. # this is intended behaviour
  701. if vmproc_task:
  702. yield from vmproc_task
  703. yield from send_task
  704. raise
  705. yield from send_task
  706. finally:
  707. if isinstance(backup_stdout, int):
  708. os.close(backup_stdout)
  709. else:
  710. backup_stdout.close()
  711. try:
  712. if vmproc_task:
  713. yield from vmproc_task
  714. finally:
  715. shutil.rmtree(self.tmpdir)
  716. # Save date of last backup, only when backup succeeded
  717. for qid, vm_info in files_to_backup.items():
  718. if vm_info.vm:
  719. vm_info.vm.backup_timestamp = \
  720. int(datetime.datetime.now().strftime('%s'))
  721. self.app.save()
  722. @asyncio.coroutine
  723. def handle_streams(stream_in, stream_out, size_limit=None,
  724. progress_callback=None):
  725. '''
  726. Copy stream_in to all streams_out and monitor all mentioned processes.
  727. If any of them terminate with non-zero code, interrupt the process. Copy
  728. at most `size_limit` data (if given).
  729. :param stream_in: StreamReader object to read data from
  730. :param stream_out: StreamWriter object to write data to
  731. :param size_limit: int maximum data amount to process
  732. :param progress_callback: callable function to report progress, will be
  733. given copied data size (it should accumulate internally)
  734. :return: "size_limit" or None (no error)
  735. '''
  736. buffer_size = 409600
  737. bytes_copied = 0
  738. while True:
  739. if size_limit:
  740. to_copy = min(buffer_size, size_limit - bytes_copied)
  741. if to_copy <= 0:
  742. return "size_limit"
  743. else:
  744. to_copy = buffer_size
  745. buf = yield from stream_in.read(to_copy)
  746. if not buf:
  747. # done
  748. break
  749. if callable(progress_callback):
  750. progress_callback(len(buf))
  751. stream_out.write(buf)
  752. bytes_copied += len(buf)
  753. return None
  754. # vim:sw=4:et: