plasma-next/tools/lib/environment.py

# SPDX-License-Identifier: GPL-3.0-or-later
"""Debian build controller and clean sbuild environment."""

import hashlib
import json
import os
import shutil
import subprocess
import sys
import threading
from contextlib import contextmanager
from pathlib import Path

import project as p


@contextmanager
def keep_sudo_alive():
    """Refresh sudo authorization until the build batch exits."""
    subprocess.run(["sudo", "-v"], check=True)
    stopped = threading.Event()

    def refresh():
        while not stopped.wait(45):
            try:
                result = subprocess.run(
                    ["sudo", "-n", "-v"],
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL,
                    timeout=10,
                )
                if result.returncode == 0:
                    continue
            except (OSError, subprocess.TimeoutExpired):
                pass
            print(
                "Sudo authorization could not be refreshed. "
                "A later build may request authentication.",
                file=sys.stderr,
            )
            return

    worker = threading.Thread(target=refresh, name="sudo-refresh", daemon=True)
    worker.start()
    try:
        yield
    finally:
        stopped.set()
        worker.join()


def setup(apt_cache=True):
    if os.geteuid() == 0:
        raise ValueError(
            "Run tools/plasma-next setup as your regular user. It invokes sudo."
        )
    for name in ["debootstrap", "unshare", "findmnt", "python3"]:
        if not shutil.which(name):
            raise ValueError("Missing host tool: " + name)
    if not Path("/usr/share/keyrings/debian-archive-keyring.gpg").is_file():
        raise ValueError("Install the Debian archive keyring on the host first.")
    # Create the lock file as the invoking user before the sudo helper opens it,
    # and refuse to touch shared inputs while another operation holds it.
    p.lock().close()
    for name in [
        "input",
        "sources",
        "results",
        "records",
        "chroots",
        "repository",
        "tmp",
    ]:
        (p.WORK / name).mkdir(parents=True, exist_ok=True)
    for name in ["debian.sources", "release.json", "snapshot-indexes.json"]:
        shutil.copy2(p.ROOT / "config" / name, p.WORK / "input" / name)
    for name in ["configure-controller", "sbuild.conf"]:
        shutil.copy2(p.ROOT / "tools/lib" / name, p.WORK / "input" / name)
    subprocess.run(
        [
            "sudo",
            "bash",
            str(p.ROOT / "tools/lib/bootstrap-controller"),
            str(os.getuid()),
            str(os.getgid()),
            "enabled" if apt_cache else "disabled",
        ],
        check=True,
    )


def build(module, target, jobs=None, apt_cache=None):
    if not (p.WORK / "controller/etc/debian_version").exists():
        raise ValueError("Run tools/plasma-next setup first.")
    if apt_cache is None:
        setting = p.WORK / "controller/etc/plasma-next-apt-cache"
        apt_cache = not setting.exists() or setting.read_text().strip() == "enabled"
    if jobs is None:
        jobs = p.read(p.ROOT / "config/release.json")["jobs"]
    if not isinstance(jobs, int) or isinstance(jobs, bool) or jobs < 1:
        raise ValueError("Build jobs must be a positive integer")
    entry = p.packages()[module]
    recipe, version = p.recipe(module)
    control = p.fields((recipe / "control").read_text())
    plan = {
        "module": module,
        "source": entry["source"],
        "version": version,
        "packages": [x["Package"] for x in control[1:]],
        "architecture_all": all(x["Architecture"] == "all" for x in control[1:]),
        "jobs": jobs,
        "target": str(target.relative_to(p.WORK)),
        "prepared": p.read(p.WORK / "prepared" / (module + ".json")),
    }
    plan_path = p.WORK / "input/modules" / (module + ".json")
    state_path = p.WORK / "state" / (module + ".json")
    if plan_path.exists() and state_path.exists():
        previous = p.read(plan_path)
        state = p.read(state_path)
        previous_inputs = {
            key: value for key, value in previous.items() if key != "jobs"
        }
        current_inputs = {key: value for key, value in plan.items() if key != "jobs"}
        previous_identity = hashlib.sha256(
            json.dumps(previous, sort_keys=True).encode()
        ).hexdigest()
        if (
            previous_inputs == current_inputs
            and state["version"] == version
            and state["inputs"] == previous_identity
        ):
            # Parallelism alone does not invalidate a completed build.
            plan = previous
    p.write(plan_path, plan)
    shutil.copy2(p.ROOT / "config/release.json", p.WORK / "input/release.json")
    for name in ["apt-cache.py", "build-package.py"]:
        shutil.copy2(p.ROOT / "tools/lib" / name, p.WORK / "input" / name)
    command = ["python3", "/work/input/build-package.py", module]
    if apt_cache:
        command = ["python3", "/work/input/apt-cache.py", *command]
    subprocess.run(
        [
            "sudo",
            "bash",
            str(p.ROOT / "tools/lib/controller-enter"),
            "builder",
            *command,
        ],
        check=True,
    )


def reset():
    subprocess.run(
        ["sudo", "bash", str(p.ROOT / "tools/lib/reset-environment")],
        check=True,
    )