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