debian-forge/mounts/org.osbuild.ostree.deployment

#!/usr/bin/python3
"""
OSTree deployment mount service

This mount service will setup all needed bind mounts so
that a given `tree` will look like an active OSTree
deployment, very much as OSTree does during early boot.

More specifically it will:
  - setup the sysroot bindmount to the deployment
  - setup the shared var directory
  - bind the boot directory into the deployment

Host commands used: mount
"""

import os
import subprocess
import sys
from typing import Dict

from osbuild import mounts
from osbuild.util import ostree

SCHEMA_2 = """
"additionalProperties": false,
"required": ["name", "type"],
"properties": {
  "name": { "type": "string" },
  "type": { "type": "string" },
  "options": {
    "type": "object",
    "required": ["deployment"],
    "properties": {
      "source": {
        "type": "string",
        "pattern": "^(mount|tree)$",
        "default": "tree",
        "description": "The source of the OSTree filesystem tree. If 'mount', there should be a preceding mount defined that's mounted at /."
      },
      "deployment": {
        "type": "object",
        "additionalProperties": false,
        "required": ["osname", "ref"],
        "properties": {
          "osname": {
            "description": "Name of the stateroot to be used in the deployment",
            "type": "string"
          },
          "ref": {
            "description": "OStree ref to create and use for deployment",
            "type": "string"
          },
          "serial": {
            "description": "The deployment serial (usually '0')",
            "type": "number",
            "default": 0
          }
        }
      }
    }
  }
}
"""


class OSTreeDeploymentMount(mounts.MountService):

    def __init__(self, args):
        super().__init__(args)

        self.mountpoint = None
        self.check = False

    @staticmethod
    def bind_mount(source, target):
        subprocess.run([
            "mount", "--bind", "--make-private", source, target,
        ], check=True)

    def is_mounted(self):
        # Use `mountpoint` command here to determine if the mountpoint is mounted.
        # We would use os.path.ismount() here but that only works if a device is
        # mounted (i.e. it doesn't use the mountinfo file in the heuristic and
        # thus things like `mount --move` wouldn't show up). The exit codes from
        # `mountpoint` are:
        #
        #  0 success; the directory is a mountpoint, or device is block device on --devno
        #  1 failure; incorrect invocation, permissions or system error
        #  32 failure; the directory is not a mountpoint, or device is not a block device on --devno
        #
        cp = subprocess.run(["mountpoint", "-q", self.mountpoint], check=False)
        if cp.returncode not in [0, 32]:
            cp.check_returncode()  # will raise error
        return cp.returncode == 0

    def mount(self, args: Dict):

        tree = args["tree"]
        mountroot = args["root"]
        options = args["options"]

        source = options.get("source", "tree")
        deployment = options["deployment"]
        osname = deployment["osname"]
        ref = deployment["ref"]
        serial = deployment.get("serial", 0)

        # The user could specify either the tree or mountroot as the
        # place where we want the deployment to be mounted.
        if source == "mount":
            target = mountroot
        else:
            target = tree

        # create a private mountpoint for the target path, which is
        # needed in order to be able to move the deployment `root`
        # mountpoint here, which is contained inside tree, since
        # "moving a mount residing under a shared mount is invalid
        # and unsupported."
        #                                              - `mount(8)`
        self.bind_mount(target, target)

        deploy_root = ostree.deployment_path(target, osname, ref, serial)

        print(f"Deployment root at '{os.path.relpath(deploy_root, target)}'")
        print(f"mounting {deploy_root} -> {target}")

        var = os.path.join(target, "ostree", "deploy", osname, "var")
        boot = os.path.join(target, "boot")

        self.mountpoint = deploy_root
        self.bind_mount(deploy_root, deploy_root)  # prepare to move it later

        self.bind_mount(target, os.path.join(deploy_root, "sysroot"))
        self.bind_mount(var, os.path.join(deploy_root, "var"))
        self.bind_mount(boot, os.path.join(deploy_root, "boot"))

        subprocess.run([
            "mount", "--move", deploy_root, target,
        ], check=True)

        self.mountpoint = target
        self.check = True

    def umount(self):
        if self.mountpoint:
            subprocess.run(["sync", "-f", self.mountpoint],
                           check=self.check)
            subprocess.run(["umount", "-v", "-R", self.mountpoint],
                           check=self.check)

            # Handle bug in older util-linux mount where the
            # mountinfo/utab wouldn't have updated information
            # when mount --move is performed, which means that
            # umount -R wouldn't unmount all overmounted mounts
            # on the target because it was operating on outdated
            # information. The umount -R behavior is fixed in v2.39
            # of util-linux most likely by [1] or [2] or both. This
            # loop can be removed when all hosts we care about have
            # moved to v2.39+.
            # [1] https://github.com/karelzak/util-linux/commit/a04149fbb7c1952da1194d1514e298ff07dbc7ca
            # [2] https://github.com/karelzak/util-linux/commit/8cf6c5075780598fe3b30e7a7753d8323d093e22
            while self.is_mounted():
                print(f"extra unmount {self.mountpoint}")
                subprocess.run(["umount", "-v", self.mountpoint],
                               check=self.check)
            self.mountpoint = None


def main():
    service = OSTreeDeploymentMount.from_args(sys.argv[1:])
    service.main()


if __name__ == '__main__':
    main()