#!/usr/bin/python3 # -*- mode: python -*- # # This file is part of FreedomBox. # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as # published by the Free Software Foundation, either version 3 of the # License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . # """ Configuration helper for backups. """ import argparse import json import os import subprocess import sys import tarfile from plinth.errors import ActionError from plinth.modules.backups import MANIFESTS_FOLDER, REPOSITORY def parse_arguments(): """Return parsed command line arguments as dictionary.""" parser = argparse.ArgumentParser() subparsers = parser.add_subparsers(dest='subcommand', help='Sub command') subparsers.add_parser( 'setup', help='Create repository if it does not already exist') subparsers.add_parser('info', help='Show repository information') subparsers.add_parser('list', help='List repository contents') create = subparsers.add_parser('create', help='Create archive') create.add_argument('--name', help='Archive name', required=True) create.add_argument('--paths', help='Paths to include in archive', nargs='+') delete = subparsers.add_parser('delete', help='Delete archive') delete.add_argument('--name', help='Archive name', required=True) export_help='Export archive contents as tar on stdout' export_tar = subparsers.add_parser('export-tar', help=export_help) export_tar.add_argument('--name', help='Archive name)', required=True) get_exported_archive_apps = subparsers.add_parser( 'get-exported-archive-apps', help='Get list of apps included in exported archive file') get_exported_archive_apps.add_argument( '--path', help='Tarball file path', required=True) get_archive_apps = subparsers.add_parser( 'get-archive-apps', help='Get list of apps included in archive') get_archive_apps.add_argument( '--path', help='Path of the archive', required=True) restore_exported_archive = subparsers.add_parser( 'restore-exported-archive', help='Restore files from an exported archive') restore_exported_archive.add_argument('--path', help='Tarball file path', required=True) restore_archive = subparsers.add_parser( 'restore-archive', help='Restore files from an archive') restore_archive.add_argument('--path', help='Archive path', required=True) restore_archive.add_argument('--destination', help='Destination', required=True) subparsers.required = True return parser.parse_args() def subcommand_setup(_): """Create repository if it does not already exist.""" try: subprocess.run(['borg', 'info', REPOSITORY], check=True) except: path = os.path.dirname(REPOSITORY) if not os.path.exists(path): os.makedirs(path) subprocess.run(['borg', 'init', '--encryption', 'none', REPOSITORY]) def subcommand_info(_): """Show repository information.""" subprocess.run(['borg', 'info', '--json', REPOSITORY], check=True) def subcommand_list(_): """List repository contents.""" subprocess.run(['borg', 'list', '--json', REPOSITORY], check=True) def subcommand_create(arguments): """Create archive.""" paths = filter(os.path.exists, arguments.paths) subprocess.run([ 'borg', 'create', '--json', REPOSITORY + '::' + arguments.name, ] + list(paths), check=True) def subcommand_delete(arguments): """Delete archive.""" subprocess.run(['borg', 'delete', REPOSITORY + '::' + arguments.name], check=True) def _extract(archive_path, destination, locations=None): """Extract archive contents.""" prev_dir = os.getcwd() env = dict(os.environ, LANG='C.UTF-8') borg_call = ['borg', 'extract', archive_path] # do not extract any files when we get an empty locations list if locations is not None: borg_call.extend(locations) try: os.chdir(os.path.expanduser(destination)) # TODO: with python 3.7 use subprocess.run with the 'capture_output' # argument process = subprocess.run(borg_call, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if process.returncode != 0: error = process.stderr.decode() # Don't fail on the borg error when no files were matched if "never matched" not in error: raise subprocess.CalledProcessError(process.returncode, process.args) finally: os.chdir(prev_dir) def subcommand_export_tar(arguments): """Export archive contents as tar stream on stdout.""" subprocess.run([ 'borg', 'export-tar', REPOSITORY + '::' + arguments.name, '-'], check=True) def _read_archive_file(archive, filepath): """Read the content of a file inside an archive""" arguments = ['borg', 'extract', archive, filepath, '--stdout'] return subprocess.check_output(arguments).decode() def subcommand_get_archive_apps(arguments): """Get list of apps included in archive.""" manifest_folder = os.path.relpath(MANIFESTS_FOLDER, '/') borg_call = ['borg', 'list', arguments.path, manifest_folder, '--format', '{path}{NEWLINE}'] try: manifest_path = subprocess.check_output(borg_call).decode().strip() except subprocess.CalledProcessError: sys.exit(1) manifest = None if manifest_path: manifest_data = _read_archive_file(arguments.path, manifest_path) manifest = json.loads(manifest_data) if manifest: for app in _get_apps_of_manifest(manifest): print(app['name']) def _get_apps_of_manifest(manifest): """ Get apps of a manifest. Supports both dict format as well as list format of plinth <=0.42 """ if type(manifest) is list: apps = manifest elif type(manifest) is dict and 'apps' in manifest: apps = manifest['apps'] else: raise ActionError('Unknown manifest format') return apps def subcommand_get_exported_archive_apps(arguments): """Get list of apps included in an exported archive file.""" manifest = None with tarfile.open(arguments.path) as t: filenames = t.getnames() for name in filenames: if 'var/lib/plinth/backups-manifests/' in name \ and name.endswith('.json'): manifest_data = t.extractfile(name).read() manifest = json.loads(manifest_data) break if manifest: for app in _get_apps_of_manifest(manifest): print(app['name']) def subcommand_restore_archive(arguments): """Restore files from an archive.""" locations_data = ''.join(sys.stdin) _locations = json.loads(locations_data) locations = _locations['directories'] + _locations['files'] locations = [os.path.relpath(location, '/') for location in locations] _extract(arguments.path, arguments.destination, locations=locations) def subcommand_restore_exported_archive(arguments): """Restore files from an exported archive.""" locations_data = ''.join(sys.stdin) locations = json.loads(locations_data) with tarfile.open(arguments.path) as tar_handle: for member in tar_handle.getmembers(): path = '/' + member.name if path in locations['files']: tar_handle.extract(member, '/') else: for directory in locations['directories']: if path.startswith(directory): tar_handle.extract(member, '/') break def main(): """Parse arguments and perform all duties.""" arguments = parse_arguments() subcommand = arguments.subcommand.replace('-', '_') subcommand_method = globals()['subcommand_' + subcommand] subcommand_method(arguments) if __name__ == '__main__': main()