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, readonly=True, mount_type=None, options=None):
 113     """
 114     Mount and create necessary directory structure.
 115     :param umount: when destination is already a mount point, umount it first.
 116     """
 117     # Check/umount destination
 118     if ismount(destination):
 119         if umount:
 120             umount_all(args, destination)
 121         else:
 122             return
 123
 124     # Check/create folders
 125     if not os.path.exists(destination):
 126         if create_folders:
 127             tools.helpers.run.user(args, ["mkdir", "-p", destination])
 128         else:
 129             raise RuntimeError("Mount failed, folder does not exist: " +
 130                             destination)
 131
 132     extra_args = []
 133     opt_args = []
 134     if mount_type:
 135         extra_args.extend(["-t", mount_type])
 136     if readonly:
 137         opt_args.append("ro")
 138     if options:
 139         opt_args.extend(options)
 140     if opt_args:
 141         extra_args.extend(["-o", ",".join(opt_args)])
 142
 143     # Actually mount the folder
 144     tools.helpers.run.user(args, ["mount", *extra_args, source, destination])
 145
 146     # Verify, that it has worked
 147     if not ismount(destination):
 148         raise RuntimeError("Mount failed: " + source + " -> " + destination)