cephfs_mount

#!/usr/bin/env python3

"""
CephFS kernel mount tool.

(c) 2019-2022 Jonas Jelten <jj@sft.lol>
Released under GPLv3 or later.

Can mount CephFS without Ceph's `mount.ceph` helper tool,
instead, this is a pure-python implementation without any special dependencies.

If you want a simpler variant without the kernel keyring, use:
    mount -t ceph -o name=client.username,secret=BASE64SECRET mon1,mon2,mon3:/path destination
This even works without any `mount.ceph`-tool.

This tool features:
* uploading key in kernel secret store
* monitor connection test
* monitor host name resolution if the kernel doesn't support it

Call with --help to see things this tool can be configured for.
"""

import argparse
import base64
import configparser
import ctypes
import ctypes.util
import errno as errnos
import logging
import os
import pathlib
import shutil
import socket
import subprocess
import sys


LIBC = ctypes.CDLL(ctypes.util.find_library('c'), use_errno=True)
LIBC.mount.argtypes = (ctypes.c_char_p, ctypes.c_char_p,
                       ctypes.c_char_p, ctypes.c_ulong, ctypes.c_char_p)


def mount(source, target, filesystem, options='', mountflags=0):
    """
    Call the libc mount function.
    """
    result = LIBC.mount(source, target, filesystem, mountflags, options)
    if result < 0:
        errno = ctypes.get_errno()
        raise OSError(
            errno,
            "Failed to mount %s with %s to %s with options '%s': %s; "
            "look at dmesg for details" % (source.decode(),
                                           filesystem.decode(),
                                           target.decode(),
                                           options.decode(),
                                           os.strerror(errno)))


def ceph_available():
    """
    Check if the kernel supports mounting ceph.
    """
    with open("/proc/filesystems") as hdl:
        for line in hdl:
            if line.strip() == 'nodev\tceph':
                return True
    return False


def resolve_mons(mon_str):
    """
    Convert a string of monhost,monhost:port,...
    to monip,monip:port,...
    """

    # msgr v2 port
    mon_default_port = 3300

    mons = mon_str.split(",")
    resolved_mons = []
    reached_mon = False
    for mon in mons:
        mon_split = mon.split(":", 1)
        mon_host = mon_split[0]
        mon_port = int(mon_split[1]) if len(mon_split) == 2 else mon_default_port

        logging.info("testing connection to monitor %s:%d...", mon_host, mon_port)
        try:
            conn_test = socket.socket()
            conn_test.settimeout(5)
            conn_test.connect((mon_host, mon_port))
            mon_ip = conn_test.getpeername()[0]
            conn_test.close()
            reached_mon = True
            logging.info("  ok: %s", mon_ip)

        except OSError as exc:
            logging.info("  failed: %s", exc)
            try:
                # if the connection attempt failed,
                # we still try to add the mon ip to the list so the kernel has it.
                mon_ip = socket.getaddrinfo(mon_host, mon_port,
                                            type=socket.SOCK_STREAM)[0][4][0]
            except OSError as exc:
                continue

        mon_str = ":".join([mon_ip] + mon_split[1:])
        resolved_mons.append(mon_str)

    if not reached_mon:
        print("could not connect to at least one monitor. giving up.")
        sys.exit(1)

    return ",".join(resolved_mons)


def upload_key(key, keyid=None, username=None, no_kernel_keyring=False):
    """
    Upload the given shared secret to the kernel keyring.

    key: raw bytes, the secret key.
    keyid: name of the key created in the kernel keyring
    username: if no keyid is given, keyid=client.$username
    no_kernel_keyring: don't use the kernel keyring

    return the mount option that tells ceph about the secret id.
    """

    if not no_kernel_keyring and not (keyid or username):
        raise Exception("need either a keyid or a username "
                        "for adding keys to kernel keyring")

    if not no_kernel_keyring:
        # add key to kernel keyring for the process keyring

        keyutils_libname = ctypes.util.find_library("keyutils")
        if not keyutils_libname:
            print("Could not find 'keyutils' library. You can try to use --nokeyring.")
            sys.exit(1)

        # from keyutils.h
        key_serial_t = ctypes.c_int32
        key_spec_process_keyring = -2

        keyutils = ctypes.CDLL(keyutils_libname, use_errno=True)
        keyutils.add_key.restype = key_serial_t
        keyutils.add_key.argtypes = (
            ctypes.c_char_p,  # type
            ctypes.c_char_p,  # description
            ctypes.c_void_p,  # payload
            ctypes.c_size_t,  # payload_len
            key_serial_t,     # keyring
        )
        keyutils.request_key.restype = key_serial_t
        keyutils.request_key.argtypes = (
            ctypes.c_char_p,  # type
            ctypes.c_char_p,  # description
            ctypes.c_char_p,  # callout_info
            key_serial_t,     # keyring
        )

        key_name = keyid if keyid else "client.%s" % username

        if key:
            # push the key to the kernel
            serial = keyutils.add_key(b"ceph", key_name.encode(),
                                      key, len(key),
                                      key_spec_process_keyring)

            if serial == -1:
                errno = ctypes.get_errno()
                raise OSError(errno, "Failed to add key: %s" % (os.strerror(errno)))
        else:
            # verify the requested key exists
            # copy it to the process keyring
            serial = keyutils.request_key(b"ceph", key_name.encode(), None,
                                          key_spec_process_keyring)
            key_found = (serial != -1)
            if not key_found:
                logging.warning("Kernel probably does not know a ceph key with id='%s'", key_name)

        return "key=%s" % key_name

    # legacy-way to pass the secret to the kernel
    return "secret=%s" % base64.b64encode(key).decode()


def extract_flags(mount_options):
    """
    convert a list of mount options to a bitmask and leftover options.
    the bitmask is created by converting known options to its corresponding bits.
    """

    # flags from <sys/mount.h>
    flags = {
        "ro"                : 1 << 0,
        "nosuid"            : 1 << 1,
        "nodev"             : 1 << 2,
        "noexec"            : 1 << 3,
        "sync"              : 1 << 4,
        "remount"           : 1 << 5,
        "mandlock"          : 1 << 6,
        "noatime"           : 1 << 10,
        "nodiratime"        : 1 << 11,
        "relatime"          : 1 << 21,
    }

    inverse_flags = {
        "rw"                : ~flags["ro"],
        "suid"              : ~flags["nosuid"],
        "dev"               : ~flags["nodev"],
        "exec"              : ~flags["noexec"],
        "nobrl"             : ~flags["mandlock"],
        "nolock"            : ~flags["mandlock"],
    }

    ignored_flags = {
        "_netdev",   # systemd-feature to mount this fs after network is there
    }

    mountflags = 0
    leftover_options = []
    for option in mount_options:
        flag = flags.get(option)
        iflag = inverse_flags.get(option)

        if flag is not None:
            mountflags |= flag
        elif iflag is not None:
            mountflags &= iflag
        elif option not in ignored_flags:
            leftover_options.append(option)

    return mountflags, leftover_options


def ceph_mount(monitors, srcpath, mount_destination,
               username, keyfile, keyid=None, mount_options="",
               no_kernel_keyring=False, kernel_resolve=True):
    """
    Load the secret into the kernel and mount CephFS to the desired location.

    monitors: comma separated list of monitor hostnames/IPs
              if monitors == None, look in /etc/ceph/ceph.conf
    srcpath: path within cephfs, starting with /
    mount_destination: target directory to mount cephfs to
    username: client name without the prefix `client.`
    keyfile: keyring file or secretkey file for given username
    keyid: kernel keyring key id, determined automatically
    mount_options: option string as given to mount -o ...
    no_kernel_keyring: if true, don't load the secret into the kernel,
                       instead pass it to the mount syscall directly
    kernel_resolve: let the monitor hostnames be resolved by the kernel
    """

    if not ceph_available():
        try:
            # load the ceph kernel module
            mp_path = shutil.which("modprobe") or "/sbin/modprobe"
            subprocess.check_call([mp_path, "ceph"])

        except subprocess.CalledProcessError:
            print("Your kernel does not support mounting ceph filesystems. "
                  "Is the ceph kernel module available?")
            sys.exit(1)

    if not pathlib.Path(mount_destination).is_dir():
        print("mount: %s: mount point does not exist" % mount_destination)
        sys.exit(32)

    if not monitors:
        cephcfg = configparser.ConfigParser()
        with open("/etc/ceph/ceph.conf") as hdl:
            cephcfg.read_file(hdl)

        monitors = cephcfg['global']['mon_host'].replace(" ", "")

    if not kernel_resolve:
        monitors = resolve_mons(monitors)

    if keyfile:
        with open(keyfile) as hdl:
            keyfiledata = hdl.read().strip()

            # extract the base64-key from a client keyring file
            if "[client" in keyfiledata:
                cfg = configparser.ConfigParser()
                cfg.read_string(keyfiledata)
                if len(cfg.sections()) > 1:
                    print("%s has multiple [sections], "
                          "don't know which one is the right one" % keyfile)
                    sys.exit(1)

                key_b64 = cfg[cfg.sections()[0]]["key"]
            else:
                # the file only contains a key
                key_b64 = keyfiledata

            key = base64.b64decode(key_b64)

    key_option = upload_key(key, keyid, username, no_kernel_keyring)
    if not key_option:
        print("no secret was provided to the kernel")
        sys.exit(1)

    mount_flags, leftover_options = extract_flags(mount_options.split(","))

    mnt_options = []
    mnt_options.append("name=%s" % username)
    mnt_options.append(key_option)
    if leftover_options:
        mnt_options.extend(leftover_options)

    mnt_src = '%s:%s' % (monitors, srcpath)
    mnt_dst = mount_destination
    mnt_type = 'ceph'
    mnt_option_str = ",".join(mnt_options)

    logging.info("mounting ceph as %s from %s to %s...", username, mnt_src, mnt_dst)
    ret = mount(mnt_src.encode(),
                mnt_dst.encode(),
                mnt_type.encode(),
                mnt_option_str.encode(),
                mount_flags)

    if ret != 0:
        errno = ctypes.get_errno()
        if errno == errnos.ENODEV:
            raise OSError(errno, ("failed to mount: ceph is not supported by system "
                                  "(is the kernel module loaded?)"))

        if errno != 0:
            raise OSError(errno, "failed to mount: %s" % (os.strerror(errno)))


def clamp(number, smallest, largest):
    """
    return number but limit it to the inclusive given value range
    """
    return max(smallest, min(number, largest))


def log_setup(setting, default=1):
    """
    Perform setup for the logger.
    Run before any logging.log thingy is called.
    if setting is 0: the default is used, which is WARNING.
    else: setting + default is used.
    """

    levels = (logging.ERROR, logging.WARNING, logging.INFO,
              logging.DEBUG, logging.NOTSET)

    factor = clamp(default + setting, 0, len(levels) - 1)
    level = levels[factor]

    logging.basicConfig(level=level, format="%(message)s")
    logging.captureWarnings(True)


def main():
    """
    Parse arguments for configuring the CephFS mount.
    """

    cli = argparse.ArgumentParser()
    cli.add_argument("--nokeyring", action="store_true",
                     help="don't use the kernel keyring, pass the secret by commandline instead")
    cli.add_argument("--options", "-o",
                     help=("further comma-separated mount options like 'ro,relatime', "
                           "see 'man mount.cephfs'"))
    cli.add_argument("--username", "-u", default="admin",
                     help="ceph authentication username. default: %(default)s")
    cli.add_argument("--keyfile", "-k", default="/etc/ceph/ceph.client.admin.keyring",
                     help="keyring file or the base64 encoded secret for authenticating the user")
    cli.add_argument("--keyid", "-i",
                     help="kernel keyring key id to use for authentication. is generated by default.")
    cli.add_argument("--srcpath", "-s", default="/",
                     help="ceph source path to mount. default: %(default)s")
    cli.add_argument("--kernel-resolve", action="store_true",
                     help="let the kernel resolve the monitor IPs")
    cli.add_argument("-v", "--verbose", action="count", default=0,
                     help="increase program verbosity")
    cli.add_argument("-q", "--quiet", action="count", default=0,
                     help="decrease program verbosity")
    cli.add_argument("-m", "--monitors",
                     help=("commaseparated string of ceph monitor servers, e.g. mon1,mon2,mon3. "
                           "optionally, add :port for a monitor. omitting this option means to "
                           "look in /etc/ceph/ceph.conf for the monitors"))
    cli.add_argument("mount_destination",
                     help="path to destination director where to mount the ceph filesystem")

    args = cli.parse_args()

    log_setup(args.verbose - args.quiet)

    keyfile = args.keyfile
    if args.keyfile == cli.get_default("keyfile"):
        if args.username != cli.get_default("username"):
            keyfile = '/etc/ceph/ceph.client.%s.keyring' % args.username

    if args.nokeyring and not args.keyfile:
        cli.error("When not using the kernel keyring, use --keyfile to read a key from")

    if os.geteuid() != 0:
        print("need root privileges")
        sys.exit(1)

    ceph_mount(args.monitors, args.srcpath, args.mount_destination,
               args.username, keyfile, args.keyid,
               args.options, args.nokeyring, args.kernel_resolve)


if __name__ == "__main__":
    main()