path: root/v-i

#!/usr/bin/python3

import argparse
import glob
import os
import shutil
import sys
import tempfile
import yaml
from subprocess import run


def log(msg):
    print("INSTALLER:", msg)


def physical_volumes():
    log("list physical volumes")
    p = run(["pvdisplay", "-C", "--noheadings"], capture_output=True, check=True)
    lines = p.stdout.decode().splitlines()
    pvs = []
    for line in lines:
        words = line.split()
        pv = words[0]
        vg = words[1]
        pvs.append({"pv": pv, "vg": vg})
    return pvs


def logical_volumes():
    log("list logical volumes")
    p = run(["lvdisplay", "-C", "--noheadings"], capture_output=True, check=True)
    lines = p.stdout.decode().splitlines()
    lvs = []
    for line in lines:
        words = line.split()
        lv = words[0]
        vg = words[1]
        path = os.path.realpath(f"/dev/{vg}/{lv}")
        lvs.append({"lv": lv, "vg": vg, "path": path})
    return lvs


def volume_groups():
    log("list volume groups")
    p = run(["vgdisplay", "-C", "--noheadings"], capture_output=True, check=True)
    lines = p.stdout.decode().splitlines()
    return [line.split()[0] for line in lines if line.strip()]


def mount_points():
    log("find mount all points")
    mounts = {}
    for line in open("/proc/mounts").readlines():
        words = line.split()
        dev = os.path.realpath(words[0]) if words[0].startswith("/") else words[0]
        mounts[dev] = {
            "mount": words[1],
            "type": words[2],
        }
    return mounts


def find_mount_points(mounts, dev):
    log(f"find mount points using {dev}")
    res = []
    if dev in mounts:
        root = mounts[dev]["mount"]
        for m in mounts.values():
            if m["mount"].startswith(root):
                res.append(m)
    return res


def is_luks(path):
    log(f"is {path} a LUKS device?")
    p = run(["cryptsetup", "isLuks", path], check=False)
    return p.returncode == 0


def clean_up_disks():
    log("clean up disks from old installs")

    mounts = mount_points()
    pvs = physical_volumes()
    lvs = logical_volumes()
    vgs = volume_groups()

    for lv in lvs:
        for m in find_mount_points(mounts, lv["path"]):
            log(f"unmount {m['mount']}")
            run(["umount", m["mount"]])

    for vg in vgs:
        log(f"remove volume group {vg}")
        run(["vgremove", "--yes", vg], check=True)

    for pv in pvs:
        log(f"remove physical volume {pv}")
        run(["pvremove", pv["pv"]], check=True)
        if is_luks(pv["pv"]):
            run(["cryptsetup", "close", pv["pv"]], check=True)

    for mapping in glob.glob("/dev/mapper/*"):
        if not mapping.endswith("/control"):
            log(f"open LUKS volume {mapping} (just in case it is one)")
            run(["cryptsetup", "close", mapping], check=False)


def vmdb_spec(cryptsetup_password, playbook, extra_vars):
    device = "{{ image }}"
    spec = {
        "steps": [
            {
                "mklabel": "gpt",
                "device": device,
            },
            {
                "mkpart": "primary",
                "device": device,
                "start": "0%",
                "end": "500M",
                "tag": "efi",
            },
            {
                "mkpart": "primary",
                "device": device,
                "start": "500M",
                "end": "1G",
                "tag": "boot",
            },
        ]
    }

    # Set up pv0 for lvm2, either encrypted or cleartext.
    if cryptsetup_password:
        spec["steps"].extend(
            [
                {
                    "mkpart": "primary",
                    "device": device,
                    "start": "1G",
                    "end": "100%",
                    "tag": "cryptsetup0",
                },
                {
                    "cryptsetup": "cryptsetup0",
                    "password": cryptsetup_password,
                    "name": "pv0",
                },
            ]
        )
    else:
        spec["steps"].extend(
            [
                {
                    "mkpart": "primary",
                    "device": device,
                    "start": "1G",
                    "end": "100%",
                    "tag": "pv0",
                },
            ]
        )

    # Create file systems and install Debian.
    spec["steps"].extend(
        [
            {
                "mkfs": "vfat",
                "partition": "efi",
            },
            {
                "mkfs": "ext2",
                "partition": "boot",
            },
            {
                "vgcreate": "vg0",
                "physical": ["pv0"],
            },
            {
                "lvcreate": "vg0",
                "name": "root",
                "size": "10G",
            },
            {
                "mkfs": "ext4",
                "partition": "root",
            },
            {
                "mount": "root",
            },
            {
                "mount": "boot",
                "dirname": "/boot",
                "mount-on": "root",
            },
            {
                "mount": "efi",
                "dirname": "/boot/efi",
                "mount-on": "boot",
            },
            {
                "unpack-rootfs": "root",
            },
            {
                "debootstrap": "bullseye",
                "mirror": "http://deb.debian.org/debian",
                "target": "root",
                "unless": "rootfs_unpacked",
            },
            {
                "apt": "install",
                "packages": [
                    "console-setup",
                    "dosfstools",
                    "ifupdown",
                    "linux-image-amd64",
                    "locales-all",
                    "lvm2",
                    "psmisc",
                    "python3",
                    "ssh",
                    "strace",
                ],
                "tag": "root",
                "unless": "rootfs_unpacked",
            },
            {
                "cache-rootfs": "root",
                "unless": "rootfs_unpacked",
            },
            {
                # This MUST be after the debootstrap step.
                "virtual-filesystems": "root",
            },
            {
                "fstab": "root",
            },
            {
                # These MUST come after the fstab step so that they add the
                # crypttab in the initramfs.
                "apt": "install",
                "packages": [
                    "cryptsetup",
                    "cryptsetup-initramfs",
                ],
                "tag": "root",
            },
            {
                # This also MUST come outside the rootfs caching, as it install
                # things outside the file systems.
                "grub": "uefi",
                "tag": "root",
                "efi": "efi",
                "quiet": True,
                "image-dev": device,
            },
        ]
    )

    # If a playbook has been specified, add an ansible step.
    if playbook:
        spec["steps"].append(
            {"ansible": "root", "playbook": playbook, "extra_vars": extra_vars}
        )

    return spec


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--verbose", action="store_true")
    p.add_argument("--log", default="install.log")
    p.add_argument("--cache", default="cache.tar.gz")
    p.add_argument("--playbook")
    p.add_argument("--vars")
    p.add_argument("--luks")
    p.add_argument("device")
    args = p.parse_args()

    extra_vars = {}
    if args.vars:
        with open(args.vars) as f:
            extra_vars = yaml.safe_load(f)

    clean_up_disks()

    spec = vmdb_spec(args.luks, args.playbook, extra_vars)
    tmp = tempfile.mkdtemp()
    specfile = os.path.join(tmp, "spec.yaml")
    if args.verbose:
        yaml.dump(spec, stream=sys.stdout, indent=4)
    with open(specfile, "w") as f:
        yaml.dump(spec, stream=f, indent=4)

    log(f"run vmdb2 to install on {args.device}")
    env = dict(os.environ)
    env["ANSIBLE_STDOUT_CALLBACK"] = "yaml"
    env["ANSIBLE_NOCOWS"] = "1"
    env["ANSIBLE_LOG_PATH"] = "ansible.log"
    run(
        [
            "vmdb2",
            "--verbose",
            f"--rootfs-tarball={args.cache}",
            f"--log={args.log}",
            f"--image={args.device}",
            specfile,
        ],
        check=True,
    )

    log("cleanup")
    shutil.rmtree(tmp)

    log("OK, done")


main()