tools/helpers/mount.py

   1 # Copyright 2021 Oliver Smith
   2 # SPDX-License-Identifier: GPL-3.0-or-later
   3 import os
   4 import tools.helpers.run
   5 from tools.helpers.version import versiontuple, kernel_version
   6
   7
   8 def ismount(folder):
   9     """
  10     Ismount() implementation, that works for mount --bind.
  11     Workaround for: https://bugs.python.org/issue29707
  12     """
  13     folder = os.path.realpath(os.path.realpath(folder))
  14     with open("/proc/mounts", "r") as handle:
  15         for line in handle:
  16             words = line.split()
  17             if len(words) >= 2 and words[1] == folder:
  18                 return True
  19             if words[0] == folder:
  20                 return True
  21     return False
  22
  23
  24 def bind(args, source, destination, create_folders=True, umount=False):
  25     """
  26     Mount --bind a folder and create necessary directory structure.
  27     :param umount: when destination is already a mount point, umount it first.
  28     """
  29     # Check/umount destination
  30     if ismount(destination):
  31         if umount:
  32             umount_all(args, destination)
  33         else:
  34             return
  35
  36     # Check/create folders
  37     for path in [source, destination]:
  38         if os.path.exists(path):
  39             continue
  40         if create_folders:
  41             tools.helpers.run.user(args, ["mkdir", "-p", path])
  42         else:
  43             raise RuntimeError("Mount failed, folder does not exist: " +
  44                                path)
  45
  46     # Actually mount the folder
  47     tools.helpers.run.user(args, ["mount", "-o", "bind", source, destination])
  48
  49     # Verify, that it has worked
  50     if not ismount(destination):
  51         raise RuntimeError("Mount failed: " + source + " -> " + destination)
  52
  53
  54 def bind_file(args, source, destination, create_folders=False):
  55     """
  56     Mount a file with the --bind option, and create the destination file,
  57     if necessary.
  58     """
  59     # Skip existing mountpoint
  60     if ismount(destination):
  61         return
  62
  63     # Create empty file
  64     if not os.path.exists(destination):
  65         if create_folders:
  66             dir = os.path.dirname(destination)
  67             if not os.path.isdir(dir):
  68                 tools.helpers.run.user(args, ["mkdir", "-p", dir])
  69
  70         tools.helpers.run.user(args, ["touch", destination])
  71
  72     # Mount
  73     tools.helpers.run.user(args, ["mount", "-o", "bind", source,
  74                                 destination])
  75
  76
  77 def umount_all_list(prefix, source="/proc/mounts"):
  78     """
  79     Parses `/proc/mounts` for all folders beginning with a prefix.
  80     :source: can be changed for testcases
  81     :returns: a list of folders, that need to be umounted
  82     """
  83     ret = []
  84     prefix = os.path.realpath(prefix)
  85     with open(source, "r") as handle:
  86         for line in handle:
  87             words = line.split()
  88             if len(words) < 2:
  89                 raise RuntimeError("Failed to parse line in " + source + ": " +
  90                                    line)
  91             mountpoint = words[1]
  92             if mountpoint.startswith(prefix):
  93                 # Remove "\040(deleted)" suffix (#545)
  94                 deleted_str = r"\040(deleted)"
  95                 if mountpoint.endswith(deleted_str):
  96                     mountpoint = mountpoint[:-len(deleted_str)]
  97                 ret.append(mountpoint)
  98     ret.sort(reverse=True)
  99     return ret
 100
 101
 102 def umount_all(args, folder):
 103     """
 104     Umount all folders, that are mounted inside a given folder.
 105     """
 106     all_list = umount_all_list(folder)
 107     for mountpoint in all_list:
 108         tools.helpers.run.user(args, ["umount", mountpoint])
 109     for mountpoint in all_list:
 110         if ismount(mountpoint):
 111             raise RuntimeError("Failed to umount: " + mountpoint)
 112
 113 def mount(args, source, destination, create_folders=True, umount=False,
 114           readonly=True, mount_type=None, options=None, force=True):
 115     """
 116     Mount and create necessary directory structure.
 117     :param umount: when destination is already a mount point, umount it first.
 118     :param force: attempt mounting even if the mount point already exists.
 119     """
 120     # Check/umount destination
 121     if ismount(destination):
 122         if umount:
 123             umount_all(args, destination)
 124         else:
 125             if not force:
 126                 return
 127
 128     # Check/create folders
 129     if not os.path.exists(destination):
 130         if create_folders:
 131             tools.helpers.run.user(args, ["mkdir", "-p", destination])
 132         else:
 133             raise RuntimeError("Mount failed, folder does not exist: " +
 134                             destination)
 135
 136     extra_args = []
 137     opt_args = []
 138     if mount_type:
 139         extra_args.extend(["-t", mount_type])
 140     if readonly:
 141         opt_args.append("ro")
 142     if options:
 143         opt_args.extend(options)
 144     if opt_args:
 145         extra_args.extend(["-o", ",".join(opt_args)])
 146
 147     # Actually mount the folder
 148     tools.helpers.run.user(args, ["mount", *extra_args, source, destination])
 149
 150     # Verify, that it has worked
 151     if not ismount(destination):
 152         raise RuntimeError("Mount failed: " + source + " -> " + destination)
 153
 154 def mount_overlay(args, lower_dirs, destination, upper_dir=None, work_dir=None,
 155                   create_folders=True, readonly=True):
 156     """
 157     Mount an overlay.
 158     """
 159     dirs = [*lower_dirs]
 160     options = ["lowerdir=" + (":".join(lower_dirs))]
 161
 162     if upper_dir:
 163         dirs.append(upper_dir)
 164         dirs.append(work_dir)
 165         options.append("upperdir=" + upper_dir)
 166         options.append("workdir=" + work_dir)
 167
 168     if kernel_version() >= versiontuple("4.17"):
 169         options.append("xino=off")
 170
 171     for dir_path in dirs:
 172         if not os.path.exists(dir_path):
 173             if create_folders:
 174                 tools.helpers.run.user(args, ["mkdir", "-p", dir_path])
 175             else:
 176                 raise RuntimeError("Mount failed, folder does not exist: " +
 177                                    dir_path)
 178
 179     mount(args, "overlay", destination, mount_type="overlay", options=options,
 180           readonly=readonly, create_folders=create_folders, force=True)