snapshot: Use privileged decorator for actions

Tests:

- DONE: Functional tests work
- DONE: Initial setup work on btrfs filesystem
- Not tested: Upgrading from older versions
- DONE: After backup is restored for snapshot app, snapper daemon is reloaded
- DONE: All configuration values are updated as expected
  - DONE: Values show up correctly in app page
  - DONE: Configuration files contain the proper values
- DONE: New snapshot can be created, gets listed in the snapshots list
- DONE: Enabling/disabling apt snapshotting works
  - DONE: Configuration file is updated
  - DONE: App page shows the correct value
- DONE: Deleting snapshots works, snapshot is removed from the list
- FAIL: Rolling back snapshots works (#2144)

Signed-off-by: Sunil Mohan Adapa <sunil@medhas.org>
Reviewed-by: James Valleroy <jvalleroy@mailbox.org>

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'],
+        subprocess.run([
-                       check=True)
+            '/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',
+                '/usr/share/plinth/actions/actions',
-                'yes'
+                'snapshot',
-            ], check=True)
+                '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'
+            '/usr/share/plinth/actions/actions', 'snapshot',
-        ], check=True)
+            'disable_apt_snapshot'
+        ], input='{"args": ["no"], "kwargs": {}}'.encode(), check=True)
 
 
 def _disable_searx() -> bool:

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',
+            privileged.setup(old_version)
-                                  ['setup', '--old-version',
-                                   str(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 = privileged.get_config()
-    output = json.loads(output)
 
     def get_boolean_choice(status):
         return ('yes', 'Enabled') if status else ('no', 'Disabled')

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():
+@privileged
-    """Return parsed command line arguments as dictionary."""
+def setup(old_version: int):
-    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):
     """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):
+@privileged
-    """Set flag to Enable/Disable apt software snapshots in config files"""
+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):
+@privileged
-    command = ['snapper', 'set-config'] + arguments.config.split()
+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(_):
+@privileged
-    config = _get_config()
+def get_config() -> dict[str, str]:
-    print(json.dumps(config))
+    """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',
+        'snapper', 'rollback', '--description', 'created by rollback', number
-        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()

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 = privileged.list_()
-    snapshots = json.loads(output)
     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 = privileged.list_()
-    snapshots = json.loads(output)
     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',
+                privileged.delete(snapshot['number'])
-                                      ['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 = privileged.list_()
-    snapshots = json.loads(output)
-
     snapshot = None
     for current_snapshot in snapshots:
         if current_snapshot['number'] == number: