Skip to content

Instantly share code, notes, and snippets.

@KosmX

KosmX/mount.py

Created July 23, 2024 12:54
Show Gist options
  • Save KosmX/275441023f304f73d3f0be24ea8799db to your computer and use it in GitHub Desktop.
Save KosmX/275441023f304f73d3f0be24ea8799db to your computer and use it in GitHub Desktop.
Download ZIP
Waydroid mount.py using fuse instead of mount (for containers i think)
Raw
mount.py
# Copyright 2021 Oliver Smith
# SPDX-License-Identifier: GPL-3.0-or-later
import os
import tools.helpers.run
import tempfile
from tools.helpers.version import versiontuple, kernel_version
def ismount(folder):
"""
Ismount() implementation, that works for mount --bind.
Workaround for: https://bugs.python.org/issue29707
"""
folder = os.path.realpath(os.path.realpath(folder))
with open("/proc/mounts", "r") as handle:
for line in handle:
words = line.split()
if len(words) >= 2 and words[1] == folder:
return True
if words[0] == folder:
return True
return False
def bind(args, source, destination, create_folders=True, umount=False):
"""
Mount --bind a folder and create necessary directory structure.
:param umount: when destination is already a mount point, umount it first.
"""
# Check/umount destination
if ismount(destination):
if umount:
umount_all(args, destination)
else:
return
# Check/create folders
for path in [source, destination]:
if os.path.exists(path):
continue
if create_folders:
tools.helpers.run.user(args, ["mkdir", "-p", path])
else:
raise RuntimeError("Mount failed, folder does not exist: " +
path)
# Actually mount the folder
tools.helpers.run.user(args, ["mount", "-o", "bind", source, destination])
# Verify, that it has worked
if not ismount(destination):
raise RuntimeError("Mount failed: " + source + " -> " + destination)
def bind_file(args, source, destination, create_folders=False):
"""
Mount a file with the --bind option, and create the destination file,
if necessary.
"""
# Skip existing mountpoint
if ismount(destination):
return
# Create empty file
if not os.path.exists(destination):
if create_folders:
dir = os.path.dirname(destination)
if not os.path.isdir(dir):
tools.helpers.run.user(args, ["mkdir", "-p", dir])
tools.helpers.run.user(args, ["touch", destination])
# Mount
tools.helpers.run.user(args, ["mount", "-o", "bind", source,
destination])
def umount_all_list(prefix, source="/proc/mounts"):
"""
Parses `/proc/mounts` for all folders beginning with a prefix.
:source: can be changed for testcases
:returns: a list of folders, that need to be umounted
"""
ret = []
prefix = os.path.realpath(prefix)
with open(source, "r") as handle:
for line in handle:
words = line.split()
if len(words) < 2:
raise RuntimeError("Failed to parse line in " + source + ": " +
line)
mountpoint = words[1]
if mountpoint.startswith(prefix):
# Remove "\040(deleted)" suffix (#545)
deleted_str = r"\040(deleted)"
if mountpoint.endswith(deleted_str):
mountpoint = mountpoint[:-len(deleted_str)]
ret.append(mountpoint)
ret.sort(reverse=True)
return ret
def umount_all(args, folder):
"""
Umount all folders, that are mounted inside a given folder.
"""
all_list = umount_all_list(folder)
for mountpoint in all_list:
tools.helpers.run.user(args, ["umount", mountpoint])
for mountpoint in all_list:
if ismount(mountpoint):
raise RuntimeError("Failed to umount: " + mountpoint)
def mount(args, source, destination, create_folders=True, umount=False,
readonly=True, mount_type=None, options=None, force=True):
"""
Mount and create necessary directory structure.
:param umount: when destination is already a mount point, umount it first.
:param force: attempt mounting even if the mount point already exists.
"""
# Check/umount destination
if ismount(destination):
if umount:
umount_all(args, destination)
else:
if not force:
return
# Check/create folders
if not os.path.exists(destination):
if create_folders:
tools.helpers.run.user(args, ["mkdir", "-p", destination])
else:
raise RuntimeError("Mount failed, folder does not exist: " +
destination)
use_fuse = True
extra_args = []
opt_args = []
if use_fuse:
opt_args.append("default_permissions")
opt_args.append("allow_other")
if mount_type:
extra_args.extend(["-t", mount_type])
if readonly:
opt_args.append("ro")
if options:
opt_args.extend(options)
if opt_args:
extra_args.extend(["-o", ",".join(opt_args)])
if (use_fuse):
if mount_type == "overlay":
tools.helpers.run.user(args, ["fuse-overlayfs", *extra_args, source, destination])
else: # not overlayfs
if (not readonly):
raise RuntimeError("NOT IMPLEMENTED")
else:
temp_dir = tempfile.TemporaryDirectory(delete=False).name
tools.helpers.run.user(args, ["fuse2fs", *extra_args, source, temp_dir])
mount_overlay(args, [temp_dir], destination, readonly=True)
else:
# Actually mount the folder
tools.helpers.run.user(args, ["mount", *extra_args, source, destination])
# Verify, that it has worked
if not ismount(destination):
raise RuntimeError("Mount failed: " + source + " -> " + destination)
def mount_overlay(args, lower_dirs, destination, upper_dir=None, work_dir=None,
create_folders=True, readonly=True):
"""
Mount an overlay.
"""
dirs = [*lower_dirs]
options = ["lowerdir=" + (":".join(lower_dirs))]
if upper_dir:
dirs.append(upper_dir)
dirs.append(work_dir)
options.append("upperdir=" + upper_dir)
options.append("workdir=" + work_dir)
if kernel_version() >= versiontuple("4.17"):
options.append("xino=off")
for dir_path in dirs:
if not os.path.exists(dir_path):
if create_folders:
tools.helpers.run.user(args, ["mkdir", "-p", dir_path])
else:
raise RuntimeError("Mount failed, folder does not exist: " +
dir_path)
mount(args, "overlay", destination, mount_type="overlay", options=options,
readonly=readonly, create_folders=create_folders, force=True)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment