matthewtran.com/scripts/provision.py

#!/usr/bin/env python3

import base64
import json
import secrets
import subprocess
import yaml

if __name__ == "__main__":
    cfg = json.load(open("config/server.json"))
    but = {
        "variant": "fcos",
        "version": "1.6.0",
    }

    # recommend keys if needed
    if "var_key" not in cfg["core"]:
        print(f'cfg["core"]["var_key"] doesn\'t exist, try "{base64.b64encode(secrets.token_bytes(64)).decode("utf-8")}"')
        exit(1)
    for i, d in enumerate(cfg["drives"]):
        if "key" not in d:
            print(f'cfg["drives"][{i}]["key"] doesn\'t exist, try "{base64.b64encode(secrets.token_bytes(64)).decode("utf-8")}"')
            exit(1)

    # configure root drive
    but["storage"] = {
        "disks": [
            {
                "device": "/dev/disk/by-id/coreos-boot-disk",
                "wipe_table": False,
                "partitions": [
                    {
                        "number": 4,
                        "label": "root",
                        "size_mib": 16384,
                        "resize": True,
                    },
                    {
                        "label": "var",
                        "size_mib": 0,
                    },
                ],
            },
        ],
        "raid": [],
        "luks": [
            {
                "name": "root",
                "label": "luks-root",
                "device": "/dev/disk/by-partlabel/root",
                "wipe_volume": True,
                "clevis": { "tpm2": True },
            },
            {
                "name": "var",
                "device": "/dev/disk/by-partlabel/var",
                "wipe_volume": cfg["core"]["var_wipe"],
                "key_file": { "inline": base64.b64decode(cfg["core"]["var_key"]) },
            },
        ],
        "filesystems": [
            {
                "device": "/dev/mapper/root",
                "format": "xfs",
                "wipe_filesystem": True,
                "label": "root",
            },
            {
                "path": "/var",
                "device": "/dev/mapper/var",
                "format": "xfs",
                "wipe_filesystem": cfg["core"]["var_wipe"],
                "with_mount_unit": True,
            },
        ],
        "files": [],
        "directories": [],
    }

    # add additional drives
    for d in cfg["drives"]:
        raid = len(d["devices"]) > 1
        if raid:
            but["storage"]["raid"].append({
                "name": d["name"],
                "level": "raid1",
                "devices": d["devices"],
            })
        but["storage"]["luks"].append({
            "name": d["name"],
            "device": f"/dev/md/{d["name"]}" if raid else d["devices"][0],
            "wipe_volume": d["wipe"],
            "key_file": { "inline": base64.b64decode(d["key"]) },
        })
        but["storage"]["filesystems"].append({
            "path": f"/var/mnt/{d["name"]}",
            "device": f"/dev/mapper/{d["name"]}",
            "format": "ext4",
            "wipe_filesystem": d["wipe"],
            "with_mount_unit": True,
        })
        but["storage"]["directories"].append({
            "path": f"/var/mnt/{d["name"]}",
            "user": { "name": "core" },
            "group": { "name": "core" },
        })

    # add SSH keys
    assert(len(cfg["core"]["ssh_keys"]) > 0)
    but["passwd"] = {
        "users": [
            {
                "name": "core",
                "ssh_authorized_keys": cfg["core"]["ssh_keys"],
            },
        ],
    }

    # add packages
    # TODO update once done https://github.com/coreos/fedora-coreos-tracker/issues/681
    but["systemd"] = {
        "units": [
            {
                "name": "rpm-ostree-install.service",
                "enabled": True,
                "contents": "\n".join([
                    "[Unit]",
                    "Description=Install packages",
                    "Wants=network-online.target",
                    "After=network-online.target",
                    "Before=zincati.service",
                    "ConditionPathExists=!/etc/rpm/%N.stamp",
                    "[Service]",
                    "Type=oneshot",
                    "RemainAfterExit=yes",
                    "ExecStart=/usr/bin/rpm-ostree install -y --allow-inactive " + " ".join([
                        "avahi",
                        "htop",
                        "vim",
                    ]),
                    "ExecStart=/bin/touch /etc/rpm/%N.stamp",
                    "ExecStart=/bin/systemctl --no-block reboot",
                    "[Install]",
                    "WantedBy=multi-user.target",
                ]),
            },
        ],
    }

    # set hostname
    but["storage"]["files"].append({
        "path": "/etc/hostname",
        "mode": 0o644,
        "contents": { "inline": cfg["core"]["hostname"] },
    })

    # allow unprivileged port access
    but["storage"]["files"].append({
        "path": "/etc/sysctl.d/99-unprivileged-ports.conf",
        "mode": 0o644,
        "contents": { "inline": "net.ipv4.ip_unprivileged_port_start=80" },
    })


    # TODO make server build images on first boot?
    # TODO serve backup.zip to restore on first boot? only if wipe specified

    # TODO convert all to quadlets? whatever compose likes
    # TODO enable bedrock => check idle cpu
    # TODO reduce disk logging?


    with open("config/server.bu", "w") as f:
        f.write(yaml.dump(but, sort_keys=False))
    subprocess.check_output(["butane", "-p", "-s", "-o", "config/server.ign", "config/server.bu"])
    print("WARNING - Using unencrypted connections without authentication, ensure LAN is secure!")