mirror of
https://github.com/freedombox/FreedomBox.git
synced 2026-01-21 07:55:00 +00:00
01492895c4
- Downloaded archives can't be fully extracted as tar.gz is incomplete at the end and corrupt. This is due to complete gzip streaming implementation that is does not flush the final bytes of gzip stream. Remove custom implementation and get gzipped stream directly from borg. - Fix mimetype for .tar.gz to application/gzip. Signed-off-by: Sunil Mohan Adapa <sunil@medhas.org> Reviewed-by: James Valleroy <jvalleroy@mailbox.org>
328 lines
11 KiB
Python
Executable File
328 lines
11 KiB
Python
Executable File
#!/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 <http://www.gnu.org/licenses/>.
|
|
#
|
|
"""
|
|
Wrapper to handle backups using borg-backups.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
|
|
from plinth.modules.backups import MANIFESTS_FOLDER
|
|
|
|
TIMEOUT = 5
|
|
|
|
|
|
def parse_arguments():
|
|
"""Return parsed command line arguments as dictionary."""
|
|
parser = argparse.ArgumentParser()
|
|
subparsers = parser.add_subparsers(dest='subcommand', help='Sub command')
|
|
|
|
setup = subparsers.add_parser(
|
|
'setup', help='Create repository if it does not already exist')
|
|
|
|
init = subparsers.add_parser('init', help='Initialize a repository')
|
|
init.add_argument('--encryption', help='Encryption of the repository',
|
|
required=True)
|
|
|
|
info = subparsers.add_parser('info', help='Show repository information')
|
|
|
|
list_repo = subparsers.add_parser('list-repo',
|
|
help='List repository contents')
|
|
|
|
create_archive = subparsers.add_parser('create-archive',
|
|
help='Create archive')
|
|
create_archive.add_argument('--paths', help='Paths to include in archive',
|
|
nargs='+')
|
|
|
|
delete_archive = subparsers.add_parser('delete-archive',
|
|
help='Delete archive')
|
|
|
|
export_help = 'Export archive contents as tar on stdout'
|
|
export_tar = subparsers.add_parser('export-tar', help=export_help)
|
|
|
|
get_archive_apps = subparsers.add_parser(
|
|
'get-archive-apps', help='Get list of apps included in archive')
|
|
|
|
restore_archive = subparsers.add_parser(
|
|
'restore-archive', help='Restore files from an archive')
|
|
restore_archive.add_argument('--destination', help='Destination',
|
|
required=True)
|
|
|
|
for cmd in [
|
|
info, init, list_repo, create_archive, delete_archive, export_tar,
|
|
get_archive_apps, restore_archive, setup
|
|
]:
|
|
cmd.add_argument('--path', help='Repository or Archive path',
|
|
required=False)
|
|
cmd.add_argument('--ssh-keyfile', help='Path of private ssh key',
|
|
default=None)
|
|
cmd.add_argument('--encryption-passphrase',
|
|
help='Encryption passphrase', default=None)
|
|
|
|
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)
|
|
|
|
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)
|
|
|
|
subparsers.required = True
|
|
return parser.parse_args()
|
|
|
|
|
|
def subcommand_setup(arguments):
|
|
"""Create repository if it does not already exist."""
|
|
try:
|
|
run(['borg', 'info', arguments.path], arguments=arguments, check=True)
|
|
except subprocess.CalledProcessError:
|
|
path = os.path.dirname(arguments.path)
|
|
if not os.path.exists(path):
|
|
os.makedirs(path)
|
|
|
|
init_repository(arguments, encryption='none')
|
|
|
|
|
|
def init_repository(arguments, encryption):
|
|
"""Initialize a local or remote borg repository"""
|
|
if encryption != 'none':
|
|
if not hasattr(arguments, 'encryption_passphrase') or not \
|
|
arguments.encryption_passphrase:
|
|
raise ValueError('No encryption passphrase provided')
|
|
|
|
cmd = ['borg', 'init', '--encryption', encryption, arguments.path]
|
|
run(cmd, arguments=arguments)
|
|
|
|
|
|
def subcommand_init(arguments):
|
|
"""Initialize the borg repository."""
|
|
init_repository(arguments, encryption=arguments.encryption)
|
|
|
|
|
|
def subcommand_info(arguments):
|
|
"""Show repository information."""
|
|
run(['borg', 'info', '--json', arguments.path], arguments=arguments)
|
|
|
|
|
|
def subcommand_list_repo(arguments):
|
|
"""List repository contents."""
|
|
run(['borg', 'list', '--json', arguments.path], arguments=arguments)
|
|
|
|
|
|
def subcommand_create_archive(arguments):
|
|
"""Create archive."""
|
|
paths = filter(os.path.exists, arguments.paths)
|
|
run(['borg', 'create', '--json', arguments.path] + list(paths),
|
|
arguments=arguments)
|
|
|
|
|
|
def subcommand_delete_archive(arguments):
|
|
"""Delete archive."""
|
|
run(['borg', 'delete', arguments.path], arguments)
|
|
|
|
|
|
def _extract(archive_path, destination, locations=None, env=None):
|
|
"""Extract archive contents."""
|
|
if not env:
|
|
env = dict(os.environ)
|
|
env['LANG'] = 'C.UTF-8'
|
|
|
|
prev_dir = os.getcwd()
|
|
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."""
|
|
run(['borg', 'export-tar', arguments.path, '-', '--tar-filter=gzip'],
|
|
arguments=arguments)
|
|
|
|
|
|
def _read_archive_file(archive, filepath, env=None):
|
|
"""Read the content of a file inside an archive"""
|
|
arguments = ['borg', 'extract', archive, filepath, '--stdout']
|
|
return subprocess.check_output(arguments, env=env).decode()
|
|
|
|
|
|
def subcommand_get_archive_apps(arguments):
|
|
"""Get list of apps included in archive."""
|
|
env = get_env(arguments)
|
|
manifest_folder = os.path.relpath(MANIFESTS_FOLDER, '/')
|
|
borg_call = [
|
|
'borg', 'list', arguments.path, manifest_folder, '--format',
|
|
'{path}{NEWLINE}'
|
|
]
|
|
timeout = None
|
|
if 'BORG_RSH' in env and 'SSHPASS' not in env:
|
|
timeout = TIMEOUT
|
|
try:
|
|
manifest_path = subprocess.check_output(borg_call, env=env,
|
|
timeout=timeout).decode()\
|
|
.strip()
|
|
except subprocess.CalledProcessError:
|
|
sys.exit(1)
|
|
|
|
manifest = None
|
|
if manifest_path:
|
|
manifest_data = _read_archive_file(arguments.path, manifest_path,
|
|
env=env)
|
|
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 isinstance(manifest, list):
|
|
apps = manifest
|
|
elif isinstance(manifest, dict) and 'apps' in manifest:
|
|
apps = manifest['apps']
|
|
else:
|
|
raise RuntimeError('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 tar_handle:
|
|
filenames = tar_handle.getnames()
|
|
for name in filenames:
|
|
if 'var/lib/plinth/backups-manifests/' in name \
|
|
and name.endswith('.json'):
|
|
manifest_data = tar_handle.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."""
|
|
env = get_env(arguments)
|
|
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,
|
|
env=env)
|
|
|
|
|
|
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 read_password():
|
|
"""Read the password from stdin."""
|
|
if sys.stdin.isatty():
|
|
return ''
|
|
|
|
return ''.join(sys.stdin)
|
|
|
|
|
|
def get_env(arguments, use_credentials=False):
|
|
"""Create encryption and ssh kwargs out of given arguments"""
|
|
env = dict(os.environ, BORG_RELOCATED_REPO_ACCESS_IS_OK='yes')
|
|
# always provide BORG_PASSPHRASE (also if empty) so borg does not get stuck
|
|
# while asking for a passphrase.
|
|
passphrase = arguments.encryption_passphrase or ''
|
|
env['BORG_PASSPHRASE'] = passphrase
|
|
if use_credentials:
|
|
if arguments.ssh_keyfile:
|
|
env['BORG_RSH'] = 'ssh -i %s' % arguments.ssh_keyfile
|
|
else:
|
|
password = read_password()
|
|
if password:
|
|
env['SSHPASS'] = password
|
|
env['BORG_RSH'] = 'sshpass -e ssh -o StrictHostKeyChecking=no'
|
|
else:
|
|
raise ValueError('could not find credentials')
|
|
|
|
return env
|
|
|
|
|
|
def run(cmd, arguments, check=True):
|
|
"""Wrap the command with ssh password or keyfile authentication"""
|
|
# Set a timeout to not get stuck if the remote server asks for a password.
|
|
timeout = None
|
|
use_credentials = False
|
|
if '@' in arguments.path:
|
|
timeout = TIMEOUT
|
|
use_credentials = True
|
|
|
|
env = get_env(arguments, use_credentials=use_credentials)
|
|
subprocess.run(cmd, check=check, env=env, timeout=timeout)
|
|
|
|
|
|
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()
|