python: setup: arch: installimage adjustments

Signed-off-by: Nathan Chancellor <[email protected]>

python/lib/setup.py

@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: MIT
 # Copyright (C) 2022-2023 Nathan Chancellor
 
+import copy
 import grp
 import ipaddress
 import os
@@ -19,6 +20,107 @@
 # pylint: enable=wrong-import-position
 
 
+class FstabItem:
+
+    def __init__(self, fs, directory, fstype, opts, dump, check):
+
+        self.fs = fs
+        self.dir = directory
+        self.type = fstype
+        self.opts = opts
+        self.dump = dump
+        self.check = check
+
+    def __str__(self):
+        order = ('fs', 'dir', 'type', 'opts', 'dump', 'check')
+        return ' '.join(getattr(self, attr) for attr in order)
+
+    def get_dev(self):
+        if (uuid := self.get_uuid()) and (uuid_path := Path('/dev/disk/by-uuid', uuid)).exists():
+            return uuid_path.resolve()
+        if self.fs.startswith('/dev'):
+            return Path(self.fs)
+        return None
+
+    def get_uuid(self):
+        if self.fs.startswith('UUID='):
+            return self.fs.split('=', maxsplit=1)[1]
+        return None
+
+
+class Fstab:
+
+    DEFAULT_ARCH_VFAT_OPTS = 'rw,relatime,fmask=0022,dmask=0022,codepage=437,iocharset=ascii,shortname=mixed,utf8,errors=remount-ro'
+
+    def __init__(self, init_str=''):
+
+        self.entries = {}
+
+        if not init_str:
+            init_str = Path('/etc/fstab').read_text(encoding='utf-8')
+
+        for line in init_str.splitlines():
+            if line := line.strip():
+                if line.startswith('#'):
+                    continue
+
+                item = FstabItem(*line.split())
+                self.entries[item.dir] = item
+
+    def __contains__(self, item):
+        return item in self.entries
+
+    def __delitem__(self, key):
+        del self.entries[key]
+
+    def __getitem__(self, item):
+        return self.entries[item]
+
+    def __setitem__(self, key, value):
+        if not isinstance(value, FstabItem):
+            raise TypeError
+        if value.dir == key:
+            self.entries[key] = value
+        else:
+            self.entries[key] = copy.copy(value)
+            self.entries[key].dir = key
+
+    def _gen_str(self):
+        header = ('# Static information about the filesystems.\n'
+                  '# See fstab(5) for details.\n'
+                  '# <file system> <dir> <type> <options> <dump> <pass>\n')
+        lines = []
+
+        for item in self.entries.values():
+            # If we have a UUID, try to find the corresponding block device for
+            # a comment identifying it.
+            if dev := item.get_dev():
+                lines.append(f"# {dev}")
+            lines.append(str(item))
+
+        return header + '\n'.join(lines) + '\n'
+
+    def __str__(self):
+        return self._gen_str()
+
+    # Adjust the options for the ESP and ext partitions on Hetzner systems to
+    # match what genfstab uses
+    def adjust_for_hetzner(self):
+        for item in self.entries.values():
+            if (item.type, item.opts) == ('vfat', 'umask=0077'):
+                item.opts = Fstab.DEFAULT_ARCH_VFAT_OPTS
+            if item.type.startswith('ext'):
+                if item.opts == 'defaults':
+                    item.opts = 'rw,relatime'
+
+                if item.check == '0':
+                    item.check = '1' if item.dir == '/' else '2'
+
+    def write(self, path=None):
+        (path if path else Path('/etc/fstab')).write_text(self._gen_str(), encoding='utf-8')
+        subprocess.run(['systemctl', 'daemon-reload'], check=True)
+
+
 def add_user_to_group(groupname, username):
     subprocess.run(['usermod', '-aG', groupname, username], check=True)
 
@@ -475,5 +577,10 @@ def using_systemd_boot():
     return subprocess.run(['bootctl', '--quiet', 'is-installed'], check=False).returncode == 0
 
 
+def umount_gracefully(folder):
+    if subprocess.run(['mountpoint', '-q', folder], check=False).returncode == 0:
+        subprocess.run(['umount', folder], check=True)
+
+
 def zypper(zypper_args):
     lib.utils.run_as_root(['zypper', *zypper_args])

python/lib/utils.py

@@ -66,12 +66,14 @@ def path_and_text(*args):
     return path, None
 
 
-def print_cmd(command, show_cmd_location=False):
+def print_cmd(command, show_cmd_location=False, end='\n'):
     if show_cmd_location:
         cmd_loc = '(container) ' if in_container() else '(host) '
     else:
         cmd_loc = ''
-    print(f"{cmd_loc}$ {' '.join([shlex.quote(str(elem)) for elem in command])}", flush=True)
+    print(f"{cmd_loc}$ {' '.join([shlex.quote(str(elem)) for elem in command])}",
+          end=end,
+          flush=True)
 
 
 def print_header(string):