akshara/usr/lib/akshara/utils/update.py

import filecmp
import os
import subprocess
import sys
from pathlib import Path

from classes.rootfs import RootFS
from utils import output, users

from . import helpers
from .gen_rootfs import gen_rootfs


def update_cleanup() -> None:
    """Clean-up from previous rebase/update."""

    # FIXME: should not ideally handle /var/cache/blendOS explicitly
    subprocess.run(
        ["umount", "-l", "/var/cache/akshara/rootfs/var/cache/blendOS"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )

    subprocess.run(
        [
            "rm",
            "-rf",
            "/var/cache/akshara/rootfs",
            "/.update_rootfs",
        ]
    )


def merge_etc(new_rootfs: RootFS, overrides_keep_new: dict) -> None:
    """Merges /etc trees.

    Args:
        new_rootfs: New RootFS instance.
        overrides_keep_new: Dictionary comprising overrides and whether to keep new.
    """
    subprocess.run(["cp", "-ax", f"{new_rootfs}/etc", f"{new_rootfs}/usr/etc"])

    if not os.path.isdir("/usr/etc"):
        subprocess.run(["rm", "-rf", "/usr/etc"])
        subprocess.run(["cp", "-ax", "/etc", "/usr/etc"])

    etc_diff = filecmp.dircmp("/etc/", "/usr/etc/")

    def handle_diff_etc_files(dcmp):
        dir_name = dcmp.left.replace("/etc/", f"{new_rootfs}/etc/", 1)
        subprocess.run(["mkdir", "-p", dir_name])
        for name in dcmp.left_only:
            subprocess.run(["cp", "-ax", "--", os.path.join(dcmp.left, name), dir_name])
        for name in dcmp.diff_files:
            subprocess.run(["cp", "-ax", "--", os.path.join(dcmp.left, name), dir_name])
        for sub_dcmp in dcmp.subdirs.values():
            handle_diff_etc_files(sub_dcmp)

    handle_diff_etc_files(etc_diff)

    for override, keep_new in overrides_keep_new.items():
        if (
            override.startswith("/etc/")
            and os.path.exists(override)
            and new_rootfs.exists(override)
        ):
            subprocess.run(["rm", "-rf", f"{new_rootfs}/{override}"])
            if keep_new:
                subprocess.run(
                    [
                        "cp",
                        "-ax",
                        f"{new_rootfs}/usr/{override}",
                        f"{new_rootfs}/{override}",
                    ]
                )
            else:
                subprocess.run(
                    [
                        "cp",
                        "-ax",
                        override,
                        f"{new_rootfs}/{override}",
                    ]
                )


def merge_var(new_rootfs: RootFS, overrides_keep_new: dict) -> None:
    """Merges /var trees.

    Args:
        new_rootfs: Path to rootfs.
        overrides_keep_new: Dictionary comprising overrides and whether to keep new.
    """
    subprocess.run(["cp", "-ax", f"{new_rootfs}/var", f"{new_rootfs}/usr/var"])
    subprocess.run(["rm", "-rf", f"{new_rootfs}/var/lib"])
    subprocess.run(["cp", "-ax", "/var/lib", f"{new_rootfs}/var/lib"])

    var_lib_diff = filecmp.dircmp(
        f"{new_rootfs}/usr/var/lib/", f"{new_rootfs}/var/lib/"
    )

    dir_name = f"{new_rootfs}/var/lib/"
    for name in var_lib_diff.left_only:
        if os.path.isdir(os.path.join(var_lib_diff.left, name)):
            subprocess.run(
                ["cp", "-ax", os.path.join(var_lib_diff.left, name), dir_name]
            )

    for override, keep_new in overrides_keep_new.items():
        if (
            override.startswith("/var/")
            and os.path.exists(override)
            and new_rootfs.exists(override)
        ):
            subprocess.run(["rm", "-rf", f"{new_rootfs}/{override}"])
            if keep_new:
                subprocess.run(
                    [
                        "cp",
                        "-ax",
                        f"{new_rootfs}/usr/{override}",
                        f"{new_rootfs}/{override}",
                    ]
                )
            else:
                subprocess.run(
                    [
                        "cp",
                        "-ax",
                        override,
                        f"{new_rootfs}/{override}",
                    ]
                )


def replace_boot_files() -> None:
    """Replace files in /boot with those from new rootfs."""
    new_boot_files = []

    for f in os.listdir("/.update_rootfs/boot"):
        if not os.path.isdir(f"/.update_rootfs/boot/{f}"):
            subprocess.run(["mv", f"/.update_rootfs/boot/{f}", "/boot"])
            new_boot_files.append(f)

    for f in os.listdir("/boot"):
        if not os.path.isdir(f"/boot/{f}"):
            if f not in new_boot_files:
                subprocess.run(["rm", "-f", f"/boot/{f}"])

    subprocess.run(["grub-mkconfig", "-o", "/boot/grub/grub.cfg"])


def update() -> None:
    """Update system.

    Args:
        image_name: Name of image to rebase to.
    """

    update_cleanup()

    system_config = helpers.get_system_config()

    output.info("generating new rootfs")

    subprocess.run(["rm", "-rf", "/var/cache/akshara"])
    Path("/var/cache/akshara/rootfs").mkdir(parents=True, exist_ok=True)

    new_rootfs = gen_rootfs(system_config, "/var/cache/akshara/rootfs")

    overrides_keep_new = (
        {
            override["path"]: override["keep"] == "new"
            for override in system_config["override"]
        }
        if isinstance(system_config.get("override"), list)
        else {}
    )

    merge_etc(new_rootfs, overrides_keep_new)

    try:
        # Store new_passwd_entries for users.merge_group() call
        new_passwd_entries = users.merge_passwd(new_rootfs)
    except Exception:
        output.error("malformed /etc/passwd")
        sys.exit(1)

    try:
        users.merge_shadow(new_rootfs)
    except Exception:
        output.error("malformed /etc/shadow")
        sys.exit(1)

    try:
        users.merge_group(new_rootfs, new_passwd_entries)
    except Exception:
        output.error("malformed /etc/group")
        sys.exit(1)

    try:
        users.merge_gshadow(new_rootfs, new_passwd_entries)
    except Exception:
        output.error("malformed /etc/shadow")
        sys.exit(1)

    merge_var(new_rootfs, overrides_keep_new)

    if (
        len(
            [
                kernel
                for kernel in os.listdir(f"{new_rootfs}/boot")
                if kernel.startswith("vmlinuz")
            ]
        )
        == 0
    ):
        output.error("new rootfs contains no kernel")
        output.error("refusing to proceed with applying update")
        exit(1)

    subprocess.run(["cp", "-ax", str(new_rootfs), "/.update_rootfs"])

    replace_boot_files()

    print()