snapshot: Fix mounting /.snapshots subvolume and use automounting

Closes: #2085.

- Read the list of snapshots and properly determine the full subvolume name to
be used for mounting the .snapshots subvolume.

- Use systemd .mount units instead of editing fstab. Fstab editing is dangerous
and could result in system not booting properly. systemd units are better suited
for tool based editing while /etc/fstab is recommended for humans.

- Use automount feature provided by systemd using autofs to perform mounting. This
means that the backing filesystem is only accessed and mounted when the mount
point is accessed by a program. Parse errors in the mount/automount file and
incorrect mount parameters are also tolerated well with failure to boot.

Tests:

- On a fresh Debian Bullseye install with btrfs. Install FreedomBox with the
changes, create and delete manual snapshots. Rollback to a snapshot should also
work. /.snapshots should contain all the files inside each of the snapshots.

- After rebooting into a rolled back snapshot, create/delete and restore to a
snapshot should work. /.snapshots should contain all the files inside each of
the snapshots.

- Introduce an error in .mount file such the mount operation will fail. Reboot
the machine. Reboot is successful. /.snapshots is still mounted as autofs.
Trying to access /.snapshots will result in error during mount operation.

- On a vagrant box without changes. Install freedombox and ensure snapshot app
setup has been run. This creates the /etc/fstab entry. Apply the patches.
snapshot app will run and remove the mount line in /etc/fstab and create the
.mount entry. /.snapshots is still mounted but not because of .automount. After
reboot, /.snapshots is mounted with autofs and also with btrfs. Unmounting
/.snapshots and then trying to run 'ls /.snapshots' will perform the mount again.

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

plinth/modules/snapshot/__init__.py

@@ -38,7 +38,7 @@ class SnapshotApp(app_module.App):
 
     app_id = 'snapshot'
 
-    _version = 4
+    _version = 5
 
     can_be_disabled = False
 

plinth/modules/snapshot/privileged.py

@@ -2,16 +2,17 @@
 """Configuration helper for filesystem snapshots."""
 
 import os
+import pathlib
 import signal
 import subprocess
 
 import augeas
 import dbus
 
+from plinth import action_utils
 from plinth.actions import privileged
 
 FSTAB = '/etc/fstab'
-AUG_FSTAB = '/files/etc/fstab'
 DEFAULT_FILE = '/etc/default/snapper'
 
 
@@ -28,7 +29,11 @@ def setup(old_version: int):
         command = ['snapper', 'create-config', '/']
         subprocess.run(command, check=True)
 
-    _add_fstab_entry('/')
+    if old_version and old_version <= 4:
+        _remove_fstab_entry('/')
+
+    _add_automount_unit('/')
+
     if old_version == 0:
         _set_default_config()
     elif old_version <= 3:
@@ -96,37 +101,118 @@ def _set_default_config():
     subprocess.run(command, check=True)
 
 
-def _add_fstab_entry(mount_point):
-    """Add mountpoint for subvolumes."""
+def _remove_fstab_entry(mount_point):
+    """Remove mountpoint for subvolumes that was added by previous versions.
+
+    The .snapshots mount is not needed, at least for the recent versions of
+    snapper. This removal code can be dropped after release of Debian Bullseye
+    + 1.
+    """
     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.transform('Fstab', '/etc/fstab')
+    aug.set('/augeas/context', '/files/etc/fstab')
     aug.load()
 
     spec = None
-    for entry in aug.match(AUG_FSTAB + '/*'):
+    for entry in aug.match('*'):
         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')
+        for entry in aug.match('*'):
+            if (aug.get(entry + '/spec') == spec
+                    and aug.get(entry + '/file') == snapshots_mount_point
+                    and aug.get(entry + '/vfstype') == 'btrfs'
+                    and aug.get(entry + '/opt') == 'subvol'
+                    and aug.get(entry + '/opt/value') == '.snapshots'):
+                aug.remove(entry)
+
         aug.save()
 
 
+def _systemd_path_escape(path):
+    """Escape a string using systemd path rules."""
+    process = subprocess.run(['systemd-escape', '--path', path],
+                             stdout=subprocess.PIPE, check=True)
+    return process.stdout.decode().strip()
+
+
+def _get_subvolume_path(mount_point):
+    """Return the subvolume path for .snapshots in a filesystem."""
+    # -o causes the list of subvolumes directly under the given mount point
+    process = subprocess.run(['btrfs', 'subvolume', 'list', '-o', mount_point],
+                             stdout=subprocess.PIPE, check=True)
+    for line in process.stdout.decode().splitlines():
+        entry = line.split()
+
+        # -o also causes the full path of the subvolume to be listed. This can
+        # -be used directly for mounting.
+        subvolume_path = entry[-1]
+        if '/' in subvolume_path:
+            path_parts = subvolume_path.split('/')
+            if len(path_parts) != 2 or path_parts[1] != '.snapshots':
+                continue
+        elif subvolume_path != '.snapshots':
+            continue
+
+        return subvolume_path
+
+    raise KeyError(f'.snapshots subvolume not found in {mount_point}')
+
+
+def _add_automount_unit(mount_point):
+    """Add a systemd automount unit for mounting .snapshots subvolume."""
+    aug = augeas.Augeas(flags=augeas.Augeas.NO_LOAD +
+                        augeas.Augeas.NO_MODL_AUTOLOAD)
+    aug.transform('Fstab', '/etc/fstab')
+    aug.set('/augeas/context', '/files/etc/fstab')
+    aug.load()
+
+    what = None
+    for entry in aug.match('*'):
+        entry_mount_point = aug.get(entry + '/file')
+        if (entry_mount_point == mount_point
+                and aug.get(entry + '/vfstype') == 'btrfs'):
+            what = aug.get(entry + '/spec')
+
+    snapshots_mount_point = os.path.join(mount_point, '.snapshots')
+    unit_name = _systemd_path_escape(snapshots_mount_point)
+    subvolume = _get_subvolume_path(mount_point)
+    mount_file = pathlib.Path(f'/etc/systemd/system/{unit_name}.mount')
+    mount_file.write_text(f'''# SPDX-License-Identifier: AGPL-3.0-or-later
+
+[Unit]
+Description=Mount for Snapshots Subvolume (FreedomBox)
+Documentation=man:snapper(8)
+
+[Mount]
+What={what}
+Where={snapshots_mount_point}
+Type=btrfs
+Options=subvol={subvolume}
+''')
+    mount_file = pathlib.Path(f'/etc/systemd/system/{unit_name}.automount')
+    mount_file.write_text(f'''# SPDX-License-Identifier: AGPL-3.0-or-later
+
+[Unit]
+Description=Automount for Snapshots Subvolume (FreedomBox)
+Documentation=man:snapper(8)
+
+[Automount]
+Where={snapshots_mount_point}
+
+[Install]
+WantedBy=local-fs.target
+''')
+    action_utils.service_daemon_reload()
+    action_utils.service_enable(f'{unit_name}.automount')
+
+
 def _parse_number(number):
     """Parse the char following the number and return status of snapshot."""
     is_default = number[-1] in ('+', '*')

plinth/modules/snapshot/tests/test_privileged.py

@@ -0,0 +1,23 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""
+Test module for privileged snapshot operations.
+"""
+
+import pathlib
+
+import pytest
+
+from plinth.modules.snapshot import privileged
+
+systemctl_path = pathlib.Path('/usr/bin/systemctl')
+systemd_installed = pytest.mark.skipif(not systemctl_path.exists(),
+                                       reason='systemd not available')
+
+
+@pytest.mark.parametrize('input_path,escaped_path',
+                         [('.snapshots', '\\x2esnapshots'), ('/', '-'),
+                          ('/home/user', 'home-user'), (':_.', ':_.')])
+@systemd_installed
+def test_systemd_path_escape(input_path, escaped_path):
+    """Test escaping systemd paths."""
+    assert escaped_path == privileged._systemd_path_escape(input_path)