diff --git a/actions/upgrades b/actions/upgrades index 3365f991b..d5e7960f8 100755 --- a/actions/upgrades +++ b/actions/upgrades @@ -432,15 +432,18 @@ def _take_snapshot_and_disable() -> bool: Return whether snapshots shall be re-enabled at the end.""" if snapshot_is_supported(): print('Taking a snapshot before dist upgrade...', flush=True) - subprocess.run(['/usr/share/plinth/actions/snapshot', 'create'], - check=True) + subprocess.run([ + '/usr/share/plinth/actions/actions', 'snapshot', 'create', + '--no-args' + ], check=True) aug = snapshot_load_augeas() if is_apt_snapshots_enabled(aug): print('Disable apt snapshots during dist upgrade...', flush=True) subprocess.run([ - '/usr/share/plinth/actions/snapshot', 'disable-apt-snapshot', - 'yes' - ], check=True) + '/usr/share/plinth/actions/actions', + 'snapshot', + 'disable_apt_snapshot', + ], input='{"args": ["yes"], "kwargs": {}}'.encode(), check=True) return True else: print('Apt snapshots already disabled.', flush=True) @@ -456,8 +459,9 @@ def _restore_snapshots_config(reenable=False): if reenable: print('Re-enable apt snapshots...', flush=True) subprocess.run([ - '/usr/share/plinth/actions/snapshot', 'disable-apt-snapshot', 'no' - ], check=True) + '/usr/share/plinth/actions/actions', 'snapshot', + 'disable_apt_snapshot' + ], input='{"args": ["no"], "kwargs": {}}'.encode(), check=True) def _disable_searx() -> bool: diff --git a/plinth/modules/snapshot/__init__.py b/plinth/modules/snapshot/__init__.py index c16ccb4d6..d1e6ddf4d 100644 --- a/plinth/modules/snapshot/__init__.py +++ b/plinth/modules/snapshot/__init__.py @@ -1,22 +1,18 @@ # SPDX-License-Identifier: AGPL-3.0-or-later -""" -FreedomBox app to manage filesystem snapshots. -""" +"""FreedomBox app to manage filesystem snapshots.""" -import json import pathlib import augeas from django.utils.translation import gettext_lazy as _ -from plinth import actions from plinth import app as app_module from plinth import menu from plinth.modules import storage from plinth.modules.backups.components import BackupRestore from plinth.package import Packages -from . import manifest +from . import manifest, privileged _description = [ _('Snapshots allows creating and managing btrfs file system snapshots. ' @@ -71,9 +67,7 @@ class SnapshotApp(app_module.App): """Install and configure the app.""" super().setup(old_version) if is_supported(): - actions.superuser_run('snapshot', - ['setup', '--old-version', - str(old_version)]) + privileged.setup(old_version) self.enable() @@ -82,7 +76,7 @@ class SnapshotBackupRestore(BackupRestore): def restore_post(self, packet): """Run after restore.""" - actions.superuser_run('snapshot', ['kill-daemon']) + privileged.kill_daemon() def is_supported(): @@ -114,9 +108,9 @@ def is_apt_snapshots_enabled(aug): def get_configuration(): + """Return snapper configuration.""" aug = load_augeas() - output = actions.superuser_run('snapshot', ['get-config']) - output = json.loads(output) + output = privileged.get_config() def get_boolean_choice(status): return ('yes', 'Enabled') if status else ('no', 'Disabled') diff --git a/actions/snapshot b/plinth/modules/snapshot/privileged.py old mode 100755 new mode 100644 similarity index 72% rename from actions/snapshot rename to plinth/modules/snapshot/privileged.py index f74513089..f0d18e24a --- a/actions/snapshot +++ b/plinth/modules/snapshot/privileged.py @@ -1,11 +1,6 @@ -#!/usr/bin/python3 # SPDX-License-Identifier: AGPL-3.0-or-later -""" -Configuration helper for filesystem snapshots. -""" +"""Configuration helper for filesystem snapshots.""" -import argparse -import json import os import signal import subprocess @@ -13,46 +8,15 @@ import subprocess import augeas import dbus +from plinth.actions import privileged + 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('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): +@privileged +def setup(old_version: int): """Configure snapper.""" # Check if root config exists. command = ['snapper', 'list-configs'] @@ -65,9 +29,9 @@ def subcommand_setup(arguments): subprocess.run(command, check=True) _add_fstab_entry('/') - if arguments.old_version == 0: + if old_version == 0: _set_default_config() - elif arguments.old_version <= 3: + elif old_version <= 3: _migrate_config_from_version_3() else: pass # After version 4 and above don't reset configuration @@ -170,7 +134,8 @@ def _parse_number(number): return number.strip('-+*'), is_default, is_active -def subcommand_list(_): +@privileged +def list_() -> list[dict[str, str]]: """List snapshots.""" process = subprocess.run(['snapper', 'list'], stdout=subprocess.PIPE, check=True) @@ -189,7 +154,7 @@ def subcommand_list(_): snapshots.append(snapshot) snapshots.reverse() - print(json.dumps(snapshots)) + return snapshots def _get_default_snapshot(): @@ -208,8 +173,9 @@ def _get_default_snapshot(): return None -def subcommand_disable_apt_snapshot(arguments): - """Set flag to Enable/Disable apt software snapshots in config files""" +@privileged +def disable_apt_snapshot(state: str): + """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) @@ -217,24 +183,28 @@ def subcommand_disable_apt_snapshot(arguments): aug.set('/augeas/load/Shellvars/incl[last() + 1]', DEFAULT_FILE) aug.load() - aug.set('/files' + DEFAULT_FILE + '/DISABLE_APT_SNAPSHOT', arguments.state) + aug.set('/files' + DEFAULT_FILE + '/DISABLE_APT_SNAPSHOT', state) aug.save() -def subcommand_create(_): +@privileged +def create(): """Create snapshot.""" command = ['snapper', 'create', '--description', 'manually created'] subprocess.run(command, check=True) -def subcommand_delete(arguments): +@privileged +def delete(number: str): """Delete a snapshot by number.""" - command = ['snapper', 'delete', arguments.number] + command = ['snapper', 'delete', number] subprocess.run(command, check=True) -def subcommand_set_config(arguments): - command = ['snapper', 'set-config'] + arguments.config.split() +@privileged +def set_config(config: list[str]): + """Set snapper configuration.""" + command = ['snapper', 'set-config'] + config subprocess.run(command, check=True) @@ -249,12 +219,14 @@ def _get_config(): return config -def subcommand_get_config(_): - config = _get_config() - print(json.dumps(config)) +@privileged +def get_config() -> dict[str, str]: + """Return snapper configuration.""" + return _get_config() -def subcommand_kill_daemon(_): +@privileged +def kill_daemon(): """Kill the snapper daemon. This is generally not necessary because we do configuration changes via @@ -262,7 +234,6 @@ def subcommand_kill_daemon(_): need to kill the daemon to reload configuration. Ideally, we should be able to reload/terminate the service using systemd. - """ bus = dbus.SystemBus() @@ -277,23 +248,10 @@ def subcommand_kill_daemon(_): os.kill(pid, signal.SIGTERM) -def subcommand_rollback(arguments): +@privileged +def rollback(number: str): """Rollback to snapshot.""" command = [ - 'snapper', 'rollback', '--description', 'created by rollback', - arguments.number + 'snapper', 'rollback', '--description', 'created by rollback', 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() diff --git a/plinth/modules/snapshot/views.py b/plinth/modules/snapshot/views.py index e40df217a..86f892f76 100644 --- a/plinth/modules/snapshot/views.py +++ b/plinth/modules/snapshot/views.py @@ -1,9 +1,6 @@ # SPDX-License-Identifier: AGPL-3.0-or-later -""" -Views for snapshot module. -""" +"""Views for snapshot module.""" -import json import urllib.parse from django.contrib import messages @@ -14,14 +11,12 @@ from django.urls import reverse, reverse_lazy from django.utils.translation import gettext as _ from django.utils.translation import gettext_lazy -from plinth import actions from plinth import app as app_module -from plinth.errors import ActionError from plinth.modules import snapshot as snapshot_module from plinth.modules import storage from plinth.views import AppView -from . import get_configuration +from . import get_configuration, privileged from .forms import SnapshotForm # i18n for snapshot descriptions @@ -90,7 +85,7 @@ def manage(request): if request.method == 'POST': if 'create' in request.POST: - actions.superuser_run('snapshot', ['create']) + privileged.create() messages.success(request, _('Created snapshot.')) if 'delete_selected' in request.POST: to_delete = request.POST.getlist('snapshot_list') @@ -102,8 +97,7 @@ def manage(request): url = reverse('snapshot:delete-selected') return HttpResponseRedirect(f'{url}?{params}') - output = actions.superuser_run('snapshot', ['list']) - snapshots = json.loads(output) + snapshots = privileged.list_() has_deletable_snapshots = any([ snapshot for snapshot in snapshots if not snapshot['is_default'] and not snapshot['is_active'] @@ -148,15 +142,14 @@ def update_configuration(request, old_status, new_status): if old_status['enable_software_snapshots'] != new_status[ 'enable_software_snapshots']: if new_status['enable_software_snapshots'] == 'yes': - actions.superuser_run('snapshot', ['disable-apt-snapshot', 'no']) + privileged.disable_apt_snapshot('no') else: - actions.superuser_run('snapshot', ['disable-apt-snapshot', 'yes']) + privileged.disable_apt_snapshot('yes') try: - actions.superuser_run('snapshot', ['set-config', " ".join(config)]) - + privileged.set_config(list(config)) messages.success(request, _('Storage snapshots configuration updated')) - except ActionError as exception: + except Exception as exception: messages.error( request, _('Action error: {0} [{1}] [{2}]').format(exception.args[0], @@ -174,8 +167,7 @@ def delete_selected(request): if not to_delete: return redirect(reverse('snapshot:manage')) - output = actions.superuser_run('snapshot', ['list']) - snapshots = json.loads(output) + snapshots = privileged.list_() snapshots_to_delete = [ snapshot for snapshot in snapshots if snapshot['number'] in to_delete and not snapshot['is_active'] and not snapshot['is_default'] @@ -184,11 +176,10 @@ def delete_selected(request): if request.method == 'POST': try: for snapshot in snapshots_to_delete: - actions.superuser_run('snapshot', - ['delete', snapshot['number']]) + privileged.delete(snapshot['number']) messages.success(request, _('Deleted selected snapshots')) - except ActionError as exception: + except Exception as exception: if 'Config is in use.' in exception.args[2]: messages.error( request, @@ -208,7 +199,7 @@ def delete_selected(request): def rollback(request, number): """Show confirmation to rollback to a snapshot.""" if request.method == 'POST': - actions.superuser_run('snapshot', ['rollback', number]) + privileged.rollback(number) messages.success( request, _('Rolled back to snapshot #{number}.').format(number=number)) @@ -217,9 +208,7 @@ def rollback(request, number): _('The system must be restarted to complete the rollback.')) return redirect(reverse('power:restart')) - output = actions.superuser_run('snapshot', ['list']) - snapshots = json.loads(output) - + snapshots = privileged.list_() snapshot = None for current_snapshot in snapshots: if current_snapshot['number'] == number: