mirror of
https://github.com/freedombox/FreedomBox.git
synced 2026-01-21 07:55:00 +00:00
12ea72eb0a
- Only do the configuration based on version run the remaining setup code without considering version. - Don't force upgrade/setup configuration when doing from version 4 and above. - Force set cleanup=yes values during migration too. Signed-off-by: Sunil Mohan Adapa <sunil@medhas.org>
341 lines
11 KiB
Python
Executable File
341 lines
11 KiB
Python
Executable File
#!/usr/bin/python3
|
|
#
|
|
# 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/>.
|
|
#
|
|
"""
|
|
Configuration helper for filesystem snapshots.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import signal
|
|
import subprocess
|
|
|
|
import augeas
|
|
import dbus
|
|
|
|
FSTAB = '/etc/fstab'
|
|
AUG_FSTAB = '/files/etc/fstab'
|
|
DEFAULT_FILE = '/etc/default/snapper'
|
|
|
|
|
|
def parse_arguments():
|
|
"""Return parsed command line arguments as dictionary."""
|
|
parser = argparse.ArgumentParser()
|
|
subparsers = parser.add_subparsers(dest='subcommand', help='Sub command')
|
|
|
|
subparser = subparsers.add_parser('setup', help='Configure snapper')
|
|
subparser.add_argument(
|
|
'--old-version', type=int, required=True,
|
|
help='Earlier version of the app that is already setup.')
|
|
subparsers.add_parser('list', help='List snapshots')
|
|
subparsers.add_parser('create', help='Create snapshot')
|
|
subparsers.add_parser('get-config', help='Configurations of snapshot')
|
|
|
|
subparser = subparsers.add_parser('delete',
|
|
help='Delete a snapshot by number')
|
|
subparser.add_argument('number', help='Number of snapshot to delete')
|
|
|
|
subparser = subparsers.add_parser('delete-all',
|
|
help='Delete all the snapshots')
|
|
|
|
subparser = subparsers.add_parser('set-config',
|
|
help='Configure automatic snapshots')
|
|
subparser.add_argument('config')
|
|
subparsers.add_parser('kill-daemon',
|
|
help='Kill snapperd to reload configuration')
|
|
|
|
subparser = subparsers.add_parser('rollback', help='Rollback to snapshot')
|
|
subparser.add_argument('number', help='Number of snapshot to rollback to')
|
|
|
|
subparser = subparsers.add_parser('disable-apt-snapshot',
|
|
help='enable/disable apt snapshots')
|
|
subparser.add_argument('state')
|
|
|
|
subparsers.required = True
|
|
return parser.parse_args()
|
|
|
|
|
|
def subcommand_setup(arguments):
|
|
"""Configure snapper."""
|
|
# Check if root config exists.
|
|
command = ['snapper', 'list-configs']
|
|
process = subprocess.run(command, stdout=subprocess.PIPE, check=True)
|
|
output = process.stdout.decode()
|
|
|
|
# Create root config if needed.
|
|
if 'root' not in output:
|
|
command = ['snapper', 'create-config', '/']
|
|
subprocess.run(command, check=True)
|
|
|
|
_add_fstab_entry('/')
|
|
if arguments.old_version == 0:
|
|
_set_default_config()
|
|
elif arguments.old_version <= 3:
|
|
_migrate_config_from_version_3()
|
|
else:
|
|
pass # After version 4 and above don't reset configuration
|
|
|
|
|
|
def _migrate_config_from_version_3():
|
|
"""Upgrade configuration from version <=3.
|
|
|
|
- This configuration was not using ranges for limits which would make free
|
|
space setting unused.
|
|
- Force set yes to cleanups.
|
|
- Reset all number cleanup settings.
|
|
- Make free space setting 30% by default instead of 20%.
|
|
|
|
"""
|
|
config = _get_config()
|
|
|
|
def convert_to_range(key):
|
|
value = config[key]
|
|
value = value if '-' in value else '0-{}'.format(value)
|
|
return '{}={}'.format(key, value)
|
|
|
|
command = [
|
|
'snapper',
|
|
'set-config',
|
|
'TIMELINE_CLEANUP=yes',
|
|
'TIMELINE_MIN_AGE=0',
|
|
convert_to_range('TIMELINE_LIMIT_HOURLY'),
|
|
convert_to_range('TIMELINE_LIMIT_DAILY'),
|
|
convert_to_range('TIMELINE_LIMIT_WEEKLY'),
|
|
convert_to_range('TIMELINE_LIMIT_MONTHLY'),
|
|
convert_to_range('TIMELINE_LIMIT_YEARLY'),
|
|
'NUMBER_CLEANUP=yes',
|
|
'NUMBER_MIN_AGE=0',
|
|
'NUMBER_LIMIT=0-100',
|
|
'NUMBER_LIMIT_IMPORTANT=0-20',
|
|
'EMPTY_PRE_POST_MIN_AGE=0',
|
|
'FREE_LIMIT=0.3',
|
|
]
|
|
subprocess.run(command, check=True)
|
|
|
|
|
|
def _set_default_config():
|
|
command = [
|
|
'snapper',
|
|
'set-config',
|
|
'TIMELINE_CLEANUP=yes',
|
|
'TIMELINE_CREATE=yes',
|
|
'TIMELINE_MIN_AGE=0',
|
|
'TIMELINE_LIMIT_HOURLY=0-10',
|
|
'TIMELINE_LIMIT_DAILY=0-3',
|
|
'TIMELINE_LIMIT_WEEKLY=0-2',
|
|
'TIMELINE_LIMIT_MONTHLY=0-2',
|
|
'TIMELINE_LIMIT_YEARLY=0-0',
|
|
'NUMBER_CLEANUP=yes',
|
|
'NUMBER_MIN_AGE=0',
|
|
'NUMBER_LIMIT=0-100',
|
|
'NUMBER_LIMIT_IMPORTANT=0-20',
|
|
'EMPTY_PRE_POST_MIN_AGE=0',
|
|
'FREE_LIMIT=0.3',
|
|
]
|
|
subprocess.run(command, check=True)
|
|
|
|
|
|
def _add_fstab_entry(mount_point):
|
|
"""Add mountpoint for subvolumes."""
|
|
snapshots_mount_point = os.path.join(mount_point, '.snapshots')
|
|
|
|
aug = augeas.Augeas(flags=augeas.Augeas.NO_LOAD +
|
|
augeas.Augeas.NO_MODL_AUTOLOAD)
|
|
aug.set('/augeas/load/Fstab/lens', 'Fstab.lns')
|
|
aug.set('/augeas/load/Fstab/incl[last() + 1]', FSTAB)
|
|
aug.load()
|
|
|
|
spec = None
|
|
for entry in aug.match(AUG_FSTAB + '/*'):
|
|
entry_mount_point = aug.get(entry + '/file')
|
|
if entry_mount_point == snapshots_mount_point:
|
|
return
|
|
|
|
if entry_mount_point == mount_point and \
|
|
aug.get(entry + '/vfstype') == 'btrfs':
|
|
spec = aug.get(entry + '/spec')
|
|
|
|
if spec:
|
|
aug.set(AUG_FSTAB + '/01/spec', spec)
|
|
aug.set(AUG_FSTAB + '/01/file', snapshots_mount_point)
|
|
aug.set(AUG_FSTAB + '/01/vfstype', 'btrfs')
|
|
aug.set(AUG_FSTAB + '/01/opt', 'subvol')
|
|
aug.set(AUG_FSTAB + '/01/opt/value', '.snapshots')
|
|
aug.set(AUG_FSTAB + '/01/dump', '0')
|
|
aug.set(AUG_FSTAB + '/01/passno', '1')
|
|
aug.save()
|
|
|
|
|
|
def _get_snapper_list():
|
|
command = ['snapper', 'list']
|
|
process = subprocess.run(command, stdout=subprocess.PIPE, check=True)
|
|
return process.stdout.decode().splitlines()
|
|
|
|
|
|
def subcommand_list(_):
|
|
"""List snapshots."""
|
|
lines = _get_snapper_list()
|
|
keys = ('number', 'type', 'pre_number', 'date', 'user', 'cleanup',
|
|
'description')
|
|
snapshots = []
|
|
for line in lines[2:]:
|
|
parts = [part.strip('* ') for part in line.split('|')]
|
|
snapshots.append(dict(zip(keys, parts)))
|
|
|
|
default = _get_default_snapshot()
|
|
for snapshot in snapshots:
|
|
snapshot['is_default'] = (snapshot['number'] == default)
|
|
|
|
print(json.dumps(snapshots))
|
|
|
|
|
|
def _get_default_snapshot():
|
|
"""Return the default snapshot by looking at default subvolume."""
|
|
command = ['btrfs', 'subvolume', 'get-default', '/']
|
|
process = subprocess.run(command, stdout=subprocess.PIPE, check=True)
|
|
output = process.stdout.decode()
|
|
|
|
output_parts = output.split()
|
|
if len(output_parts) >= 9:
|
|
path = output.split()[8]
|
|
path_parts = path.split('/')
|
|
if len(path_parts) == 3 and path_parts[0] == '.snapshots':
|
|
return path_parts[1]
|
|
|
|
return None
|
|
|
|
|
|
def subcommand_disable_apt_snapshot(arguments):
|
|
"""Set flag to Enable/Disable apt software snapshots in config files"""
|
|
# Initialize Augeas
|
|
aug = augeas.Augeas(flags=augeas.Augeas.NO_LOAD +
|
|
augeas.Augeas.NO_MODL_AUTOLOAD)
|
|
aug.set('/augeas/load/Shellvars/lens', 'Shellvars.lns')
|
|
aug.set('/augeas/load/Shellvars/incl[last() + 1]', DEFAULT_FILE)
|
|
aug.load()
|
|
|
|
aug.set('/files' + DEFAULT_FILE + '/DISABLE_APT_SNAPSHOT', arguments.state)
|
|
aug.save()
|
|
|
|
|
|
def subcommand_create(_):
|
|
"""Create snapshot."""
|
|
command = ['snapper', 'create', '--description', 'manually created']
|
|
subprocess.run(command, check=True)
|
|
|
|
|
|
def subcommand_delete(arguments):
|
|
"""Delete a snapshot by number."""
|
|
command = ['snapper', 'delete', arguments.number]
|
|
subprocess.run(command, check=True)
|
|
|
|
|
|
def subcommand_delete_all(_):
|
|
"""Delete all the snapshots (except the active one)."""
|
|
lines = _get_snapper_list()
|
|
snapshot_range = [line.split('|')[0].strip() for line in lines[3:]]
|
|
default_snapshot = _get_default_snapshot()
|
|
if snapshot_range:
|
|
if default_snapshot:
|
|
index = snapshot_range.index(default_snapshot)
|
|
range_before = snapshot_range[:index]
|
|
range_after = snapshot_range[index + 1:]
|
|
to_delete = [range_before, range_after]
|
|
else:
|
|
to_delete = [snapshot_range]
|
|
|
|
delete_args = filter(None, map(_get_delete_arg, to_delete))
|
|
for arg in delete_args:
|
|
subprocess.run(['snapper', 'delete', arg], check=True)
|
|
|
|
|
|
def _get_delete_arg(range_list):
|
|
"""Return 'a-b' when given ['a', ..., 'b']."""
|
|
if not range_list:
|
|
return None
|
|
elif len(range_list) == 1:
|
|
return range_list[0]
|
|
else:
|
|
return range_list[0] + '-' + range_list[-1]
|
|
|
|
|
|
def subcommand_set_config(arguments):
|
|
command = ['snapper', 'set-config'] + arguments.config.split()
|
|
subprocess.run(command, check=True)
|
|
|
|
|
|
def _get_config():
|
|
command = ['snapper', 'get-config']
|
|
process = subprocess.run(command, stdout=subprocess.PIPE, check=True)
|
|
lines = process.stdout.decode().splitlines()
|
|
config = {}
|
|
for line in lines[2:]:
|
|
parts = [part.strip() for part in line.split('|')]
|
|
config[parts[0]] = parts[1]
|
|
return config
|
|
|
|
|
|
def subcommand_get_config(_):
|
|
config = _get_config()
|
|
print(json.dumps(config))
|
|
|
|
|
|
def subcommand_kill_daemon(_):
|
|
"""Kill the snapper daemon.
|
|
|
|
This is generally not necessary because we do configuration changes via
|
|
snapperd. However, when the configuration is restored from a backup. We
|
|
need to kill the daemon to reload configuration.
|
|
|
|
Ideally, we should be able to reload/terminate the service using systemd.
|
|
|
|
"""
|
|
bus = dbus.SystemBus()
|
|
|
|
dbus_object = bus.get_object('org.freedesktop.DBus', '/')
|
|
dbus_interface = dbus.Interface(dbus_object,
|
|
dbus_interface='org.freedesktop.DBus')
|
|
try:
|
|
pid = dbus_interface.GetConnectionUnixProcessID('org.opensuse.Snapper')
|
|
except dbus.exceptions.DBusException:
|
|
pass
|
|
else:
|
|
os.kill(pid, signal.SIGTERM)
|
|
|
|
|
|
def subcommand_rollback(arguments):
|
|
"""Rollback to snapshot."""
|
|
command = [
|
|
'snapper', 'rollback', '--description', 'created by rollback',
|
|
arguments.number
|
|
]
|
|
subprocess.run(command, check=True)
|
|
|
|
|
|
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()
|