#!/usr/bin/env python3
"""boneio-containers — privileged container operations, by name only.

The ``boneio`` account is in the ``docker`` group, which is root-equivalent: a
container can bind-mount the host root and write to it. Removing the account
from that group is the fix, but the application still needs to start Node-RED
and reload Caddy — so those operations move here, behind a closed list of
verbs.

Two things make this worth doing rather than granting ``sudo docker compose``:

  * **The vector is the compose file, not the argument list.** ``docker compose
    up`` reads ``docker-compose.yaml`` from the project directory, and that file
    lives under /home/boneio. A sudoers rule permitting only ``docker compose
    up`` still lets anyone who can write that file run a container as root with
    the host filesystem mounted. So this helper refuses to act at all unless the
    compose file is root-owned and not writable by anyone else.
  * **No argument reaches docker from the caller.** Each verb expands to a fixed
    command. The only caller-supplied values are a log line count and a domain
    name, both validated.

Caddy runs either as the ``caddy`` service of that project or, once
/etc/boneio/proxy-native exists, as the apt package under systemd. The Caddy
verbs keep their names in both modes, so the application does not need to know
which one it is talking to; in the native mode none of them runs compose, since
the project no longer has a Caddy in it. Every Caddy verb, and ``up`` and
``down``, holds /run/boneio-proxy.lock, the lock the switch between the modes
holds for its whole run, so no verb acts on the mode the switch is leaving.
The switch itself is ``proxy-switch-start``: it runs ``proxy-switch-run`` in a
unit of its own, which puts the container back if the package does not answer;
``proxy-switch-recover``, run at boot, does the same for a switch cut short.

Usage:
    boneio-containers <verb> [argument]
    boneio-containers --list-verbs
    boneio-containers --selftest

Exit codes:
    0  success
    1  refused, or the underlying command failed
    2  the compose file is not root-owned, so nothing was run
"""

from __future__ import annotations

import argparse
import contextlib
import datetime
import fcntl
import hashlib
import http.client
import json
import logging
import logging.handlers
import os
import pwd
import re
import shutil
import socket
import ssl
import stat
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request
from pathlib import Path

LOG_FILE = "/var/log/boneio-containers.log"

#: The compose project. Hard-coded on purpose: a caller-supplied project
#: directory would be a caller-supplied compose file, which is the vector.
PROJECT_DIR = Path("/home/boneio/docker/nodered")
COMPOSE_FILE = PROJECT_DIR / "docker-compose.yaml"

#: Root-owned templates, restored from the pristine copy rather than written by
#: the application. The application used to write the compose file itself,
#: which is precisely what must stop.
TRUSTED_DIR = Path("/usr/lib/boneio/trusted")
COMPOSE_TEMPLATE = TRUSTED_DIR / "docker-compose.yaml"
COMPOSE_CLOUD_TEMPLATE = TRUSTED_DIR / "docker-compose-cloud.yaml"

CADDY_SERVICE = "caddy"
NODERED_SERVICE = "node-red"

#: Present when Caddy is the apt package rather than a container. Root-owned,
#: created only by the switch, so the caller cannot pick the mode.
NATIVE_MARKER = Path("/etc/boneio/proxy-native")
#: Present when the packaged Caddy should serve the cloud's certificate; read by
#: the root generator /usr/lib/boneio/proxy-config on every start and reload.
CLOUD_MARKER = Path("/etc/boneio/proxy-cloud")
#: The packaged Caddy's data directory, holding its local authority.
CADDY_DATA = Path("/var/lib/caddy/.local/share/caddy")
#: Shared with the switch, which holds it from the first step to the last.
#: In /run, which only root can write, not the world-writable /run/lock: there
#: another account could create it first and stall every Caddy verb.
PROXY_LOCK = Path("/run/boneio-proxy.lock")
#: Seconds a locked verb waits for the switch before refusing. Shorter than
#: the application's own 30 s, so the caller hears why rather than a timeout.
PROXY_LOCK_TIMEOUT = 20.0
#: Larger than any real root certificate, small enough that a planted file
#: cannot make root read the disk.
_MAX_ROOT_CA = 64 * 1024

#: The compose file once Caddy is the package: Node-RED alone.
COMPOSE_NATIVE_TEMPLATE = TRUSTED_DIR / "docker-compose-native-proxy.yaml"
#: The ports the panel chose; the container publishes HTTPS_PORT, and the
#: packaged Caddy listens on it.
ENV_FILE = PROJECT_DIR / ".env"
#: The container's authority, in a tree the ``boneio`` account can write.
OLD_CA = PROJECT_DIR / "caddy" / "data" / "caddy" / "pki" / "authorities" / "local"
#: The files an authority is; Caddy recreates everything else under pki.
CA_FILES = ("root.crt", "root.key", "intermediate.crt", "intermediate.key")
#: This helper's own path: the switch unit runs it again, as root, outside
#: boneio.service, whose restart must not kill the switch halfway.
HELPER_SELF = "/usr/sbin/boneio-containers"
#: A fixed name, so systemd itself refuses a second switch while one runs.
SWITCH_UNIT = "boneio-proxy-switch.service"
SWITCH_DIR = Path("/var/lib/boneio/proxy")
SWITCH_STATE = SWITCH_DIR / "switch.json"
SWITCH_LOG = SWITCH_DIR / "switch.log"
SWITCH_LOG_TAIL = 200
#: The container's compose file, kept so a failed switch can put it back.
COMPOSE_BEFORE = SWITCH_DIR / "compose.before"
#: What the generator compares the hostname with before keeping the authority.
HOSTNAME_FILE = SWITCH_DIR / "last_hostname"
#: The root generator, which also validates with the real Caddy.
PROXY_CONFIG = "/usr/lib/boneio/proxy-config"
#: What a migration installs before the switch may run. The drop-in above all:
#: without its condition the package's postinst starts a stock Caddy on :80.
SWITCH_PREREQUISITES = (
    Path("/etc/systemd/system/caddy.service.d/boneio.conf"),
    Path(PROXY_CONFIG),
)
#: The Caddy boneIO installs: the release's armhf .deb from GitHub, pinned by
#: the SHA-512 in that release's checksums file. This helper is signed with the
#: migrations, so the hash is what vouches for the package, and a new Caddy is
#: a new boneIO release. Not Caddy's apt repository on Cloudsmith: it answers
#: 402 whenever the project runs out of transfer quota. Nor Debian's caddy,
#: 2.6.2, which lacks what the generated Caddyfile uses.
CADDY_VERSION = "2.11.7"
CADDY_DEB_URL = (
    f"https://github.com/caddyserver/caddy/releases/download/v{CADDY_VERSION}/"
    f"caddy_{CADDY_VERSION}_linux_armv7.deb"
)
CADDY_DEB_SHA512 = (
    "9c41e47815b2b26b16318683884e08e67456a3407e295114ef58d823c037d57a"
    "8732ca63ebdbaab05745376aec29d1e58eda2f9ff78268755d6208dad2a1c507"
)
#: Root-only. A .deb here with the pinned hash is not downloaded again; the
#: image build puts it here, fetched on the PC.
CADDY_CACHE = Path("/var/cache/boneio")
#: How long a proxy has to answer after it was started, and how often it is
#: asked.
PROBE_TIMEOUT = 90.0
PROBE_INTERVAL = 2.0
#: A clean environment for every command of the switch, as for the system
#: update: nothing of the caller's is passed through.
APT_ENV = {
    "PATH": "/usr/sbin:/usr/bin:/sbin:/bin",
    "LC_ALL": "C",
    "DEBIAN_FRONTEND": "noninteractive",
    "NEEDRESTART_MODE": "l",
    "APT_LISTCHANGES_FRONTEND": "none",
}

_MAX_LOG_LINES = 2000

#: A container image tag, per Docker's own rules and no looser. Validated
#: before it can reach the compose file, which is what `docker compose up`
#: executes — the application used to rewrite that file itself to change the
#: Node-RED version, which is the same escalation path by another route.
_TAG_RE = re.compile(r"^[A-Za-z0-9_][A-Za-z0-9._-]{0,127}$")
#: The line this helper is willing to change, and nothing else in the file.
_NODERED_IMAGE_RE = re.compile(
    r"^(?P<prefix>\s*image:\s*nodered/node-red:)(?P<tag>\S+)(?P<suffix>\s*)$",
    re.MULTILINE,
)

#: The Caddy image line, in a template or in the live compose file.
_CADDY_IMAGE_RE = re.compile(
    r"^(?P<prefix>\s*image:\s*)(?P<image>caddy:\S+)(?P<suffix>\s*)$",
    re.MULTILINE,
)
#: What a pinned Caddy image looks like: an exact tag and, normally, the digest
#: of the multi-arch index. Checked even though it comes from a root-owned
#: template, because it is written into the file ``docker compose up`` executes.
_PINNED_CADDY_RE = re.compile(
    r"^caddy:[A-Za-z0-9_][A-Za-z0-9._-]{0,127}(@sha256:[0-9a-f]{64})?$"
)


def _configure_logging() -> logging.Logger:
    """Set up logging without making the log file a hard requirement.

    Returns:
        The helper's logger.
    """
    handlers: list[logging.Handler] = [logging.StreamHandler(sys.stderr)]
    try:
        handlers.append(
            logging.handlers.RotatingFileHandler(
                LOG_FILE, maxBytes=524_288, backupCount=1, encoding="utf-8"
            )
        )
    except OSError as exc:
        print(f"boneio-containers: cannot open {LOG_FILE}: {exc}", file=sys.stderr)
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s",
        handlers=handlers,
    )
    return logging.getLogger("boneio-containers")


_LOGGER = _configure_logging()


class Refused(Exception):
    """The request was rejected; nothing was run."""


#: Verb → the arguments that follow ``docker compose -f <file>``. Constants
#: only: nothing in this table can be influenced from outside, and building the
#: argv at call time keeps it honest about which compose file is in play rather
#: than freezing the path at import.
VERBS: dict[str, tuple[str, ...]] = {
    "status": ("ps", "--format", "json"),
    "up": ("up", "-d"),
    "down": ("down",),
    "pull": ("pull",),
    "start-nodered": ("up", "-d", NODERED_SERVICE),
    "stop-nodered": ("stop", NODERED_SERVICE),
    "restart-nodered": ("restart", NODERED_SERVICE),
    "pull-nodered": ("pull", NODERED_SERVICE),
    "start-caddy": ("up", "-d", CADDY_SERVICE),
    "restart-caddy": ("restart", CADDY_SERVICE),
    "reload-caddy": (
        # The container runs /tmp/Caddyfile, written by init-certs.sh on every
        # start. /etc/caddy/Caddyfile is the image's stock config: reloading it
        # takes HTTPS down until the container is restarted.
        "exec", CADDY_SERVICE, "caddy", "reload", "--config", "/tmp/Caddyfile"
    ),
}

#: What the lifecycle verbs mean for the packaged Caddy. Reload goes through
#: systemd so the root generator rewrites the configuration first.
NATIVE_VERBS: dict[str, tuple[str, ...]] = {
    "start-caddy": ("systemctl", "start", "caddy"),
    "restart-caddy": ("systemctl", "restart", "caddy"),
    "reload-caddy": ("systemctl", "reload", "caddy"),
}

#: Verbs that talk to the docker daemon rather than to a compose project.
DAEMON_VERBS: dict[str, tuple[str, ...]] = {
    "ps": ("ps", "-a", "--format", "json"),
    "names": ("ps", "-a", "--format", "{{.Names}}"),
}

#: How much of a container log the diagnostics bundle takes.
_CONTAINER_LOG_LINES = 200
#: A container name, per Docker's own rules. Checked, and then checked again
#: against the containers that actually exist — a name is the one piece of
#: caller data that reaches docker here, so a regex alone is not enough: it
#: would still admit anything shaped like a name, including a flag.
_CONTAINER_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$")


def _compose(*args: str) -> list[str]:
    """A docker compose command for the fixed project.

    Args:
        *args: Verb and arguments to pass to compose.

    Returns:
        The full argv.
    """
    return ["docker", "compose", "-f", str(COMPOSE_FILE), *args]


def _argv_for(verb: str) -> list[str]:
    """The command a fixed verb expands to.

    Args:
        verb: One of :data:`VERBS` or :data:`DAEMON_VERBS`.

    Returns:
        The full argv.

    Raises:
        KeyError: If the verb is not a fixed one.
    """
    if verb in DAEMON_VERBS:
        return ["docker", *DAEMON_VERBS[verb]]
    return _compose(*VERBS[verb])

#: Verbs that take one validated argument.
PARAMETERISED_VERBS = {
    "logs-caddy",
    "logs-nodered",
    "apply-cloud-template",
    "remove-cloud-template",
    "set-nodered-image",
    "logs-container",
}

#: Verbs that take no argument and are not one fixed compose command.
COMPOSITE_VERBS = {"caddy-image-state", "caddy-image-apply", "caddy-root-ca"}

#: Verbs whose meaning depends on the mode.
CADDY_VERBS = {
    "start-caddy", "restart-caddy", "reload-caddy", "logs-caddy",
    "apply-cloud-template", "remove-cloud-template",
    "caddy-image-state", "caddy-image-apply", "caddy-root-ca",
}
#: Verbs that wait for the switch: the Caddy verbs, the two that would
#: recreate or remove the Caddy container in the middle of it, and the one that
#: rewrites the compose file the switch replaces.
LOCKED_VERBS = CADDY_VERBS | {"up", "down", "set-nodered-image"}

#: The move from the Caddy container to the package. Not in LOCKED_VERBS: they
#: take the lock themselves where they need it, the run for its whole length.
SWITCH_VERBS = {
    "proxy-switch-start", "proxy-switch-run", "proxy-switch-state", "proxy-switch-recover",
    "caddy-install",
}

ALL_VERBS = sorted(
    set(VERBS) | set(DAEMON_VERBS) | PARAMETERISED_VERBS | COMPOSITE_VERBS | SWITCH_VERBS
)

#: Verbs that only read. They are allowed to run even when the compose file is
#: not yet root-owned, because refusing them would leave the UI unable to say
#: what is wrong.
READ_ONLY_VERBS = {
    "status", "ps", "names", "logs-caddy", "logs-nodered", "logs-container",
    "caddy-image-state", "caddy-root-ca", "proxy-switch-state",
}


def _assert_root() -> None:
    """Exit unless running as root."""
    if os.geteuid() != 0:
        _LOGGER.error("boneio-containers must be run as root via sudo.")
        sys.exit(1)


def _root_owned(path: Path) -> bool:
    """Whether *path* is a real file owned by root and not writable by others.

    Args:
        path: Path to inspect.

    Returns:
        True when root can trust the file's contents.
    """
    try:
        st = path.lstat()
    except OSError:
        return False
    if not os.path.isfile(path) or os.path.islink(path):
        return False
    return st.st_uid == 0 and not st.st_mode & 0o022


def _assert_compose_trustworthy() -> None:
    """Refuse to run compose against a file the caller could have written.

    Raises:
        Refused: If the compose file is missing or not root-owned.
    """
    if not _root_owned(COMPOSE_FILE):
        raise Refused(
            f"{COMPOSE_FILE} is missing, a symlink, or not owned by root. "
            "Running compose against it would let whoever can write that file "
            "start a container as root with the host filesystem mounted — the "
            "file is the vector, not the command line. Reinstall it from "
            f"{COMPOSE_TEMPLATE} with a migration."
        )


def _run(argv: list[str], timeout: int = 120) -> int:
    """Run a fixed command in the project directory.

    Args:
        argv: The command to run.
        timeout: Seconds to allow.

    Returns:
        Process exit status; output is passed through.
    """
    _LOGGER.info("RUN: %s", " ".join(argv))
    # Only compose needs the project directory; systemd's tools must not fail
    # because it is missing.
    cwd = str(PROJECT_DIR) if argv[0] == "docker" else "/"
    try:
        result = subprocess.run(
            argv, cwd=cwd, capture_output=True, text=True, timeout=timeout
        )
    except subprocess.TimeoutExpired:
        _LOGGER.error("timed out after %ds: %s", timeout, " ".join(argv))
        return 1
    except OSError as exc:
        _LOGGER.error("cannot run %s: %s", argv[0], exc)
        return 1

    if result.stdout:
        sys.stdout.write(result.stdout)
    if result.returncode != 0:
        _LOGGER.error("rc=%d: %s", result.returncode, result.stderr.strip())
        sys.stderr.write(result.stderr)
    return result.returncode


def _capture(argv: list[str], timeout: int = 30) -> tuple[int, str]:
    """Run a fixed command and return its output instead of passing it on.

    ``LC_ALL=C`` because sudo keeps the caller's locale, and apt translates the
    words that are parsed here.

    Args:
        argv: The command to run.
        timeout: Seconds to allow.

    Returns:
        ``(exit status, stdout)``; status 1 and no output when it cannot run.
    """
    cwd = str(PROJECT_DIR) if argv[0] == "docker" else "/"
    try:
        result = subprocess.run(
            argv, cwd=cwd, capture_output=True, text=True, timeout=timeout,
            env={**os.environ, "LC_ALL": "C"},
        )
    except (subprocess.TimeoutExpired, OSError) as exc:
        _LOGGER.error("cannot run %s: %s", " ".join(argv), exc)
        return 1, ""
    if result.stderr:
        sys.stderr.write(result.stderr)
    return result.returncode, result.stdout


def _line_count(argument: str | None) -> int:
    """A log tail length from the caller.

    Raises:
        Refused: If the line count is not a sane integer.
    """
    if argument is None:
        return 200
    # isascii: "²".isdigit() is true, and int() would then raise.
    if (
        not (argument.isascii() and argument.isdigit())
        or not 1 <= int(argument) <= _MAX_LOG_LINES
    ):
        raise Refused(
            f"log line count must be an integer between 1 and {_MAX_LOG_LINES}"
        )
    return int(argument)


def _logs(service: str, argument: str | None) -> int:
    """Show the tail of a service's log.

    Args:
        service: Compose service name.
        argument: Number of lines, as text.

    Returns:
        Process exit status.

    Raises:
        Refused: If the line count is not a sane integer.
    """
    lines = _line_count(argument)
    return _run(_compose("logs", "--tail", str(lines), "--no-color", service), timeout=60)


def _journal_caddy(lines: int) -> int:
    """The packaged Caddy's log, which systemd keeps."""
    return _run(
        ["journalctl", "-u", "caddy", "-n", str(lines), "--no-pager", "-o", "short-iso"],
        timeout=60,
    )


def _install_compose(source: Path) -> int:
    """Install a root-owned compose file, copied from a trusted template.

    Nothing is substituted into it. The cloud template serves a wildcard
    certificate and takes the hostname from the container itself, so no
    caller-supplied value belongs in this file — which is the point, since the
    file is what ``docker compose up`` executes.

    Args:
        source: Template in the pristine copy.

    Returns:
        0 on success.

    Raises:
        Refused: If the template is missing or not root-owned.
    """
    if not _root_owned(source):
        raise Refused(
            f"template {source} is missing or not root-owned. The compose file "
            "must come from the pristine copy, not from the application."
        )
    content = source.read_text(encoding="utf-8")

    PROJECT_DIR.mkdir(parents=True, exist_ok=True)
    _back_up_if_customised(content)
    with tempfile.NamedTemporaryFile(
        dir=PROJECT_DIR, delete=False, suffix=".tmp", mode="w", encoding="utf-8"
    ) as tmp:
        tmp.write(content)
        tmp_path = tmp.name
    try:
        os.chmod(tmp_path, 0o644)
        if os.geteuid() == 0:
            # The whole point of installing it from here is that it ends up
            # root-owned; running as root this cannot fail, and not running as
            # root there is no privilege to protect.
            os.chown(tmp_path, 0, 0)
        os.replace(tmp_path, COMPOSE_FILE)
    except OSError:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        raise
    _LOGGER.info("INSTALLED: %s from %s (root:root 0644)", COMPOSE_FILE, source.name)
    return 0


def _back_up_if_customised(replacement: str) -> None:
    """Keep a copy of a compose file that matches neither template.

    The code this replaces wrote a ``.yaml.bak`` before switching, which mattered
    for anyone who had hand-edited the file. Hand-editing is what is being taken
    away — the file is what ``docker compose up`` executes — but taking it away
    should not silently discard what somebody already wrote.

    Args:
        replacement: The content about to be installed.
    """
    if not COMPOSE_FILE.is_file() or COMPOSE_FILE.is_symlink():
        return
    try:
        current = COMPOSE_FILE.read_text(encoding="utf-8")
    except OSError as exc:
        _LOGGER.warning("Cannot read %s to back it up: %s", COMPOSE_FILE, exc)
        return
    if current == replacement:
        return

    known = []
    for template in (COMPOSE_TEMPLATE, COMPOSE_CLOUD_TEMPLATE, COMPOSE_NATIVE_TEMPLATE):
        try:
            known.append(template.read_text(encoding="utf-8"))
        except OSError:
            continue
    if current in known:
        # One of ours; the template it came from is the backup.
        return

    backup = COMPOSE_FILE.with_suffix(".yaml.bak")
    if backup.exists():
        _LOGGER.info("Keeping the existing backup at %s", backup)
        return
    try:
        backup.write_text(current, encoding="utf-8")
        os.chmod(backup, 0o644)
        _LOGGER.info(
            "Backed up a customised %s to %s before replacing it",
            COMPOSE_FILE, backup,
        )
    except OSError as exc:
        _LOGGER.warning("Could not back up %s: %s", COMPOSE_FILE, exc)


def _logs_container(argument: str | None) -> int:
    """Show the tail of one container's log, by name.

    The diagnostics bundle reads every container's log, and compose prefixes
    the names with the project (``nodered-caddy-1``, not ``caddy``), so a fixed
    list would produce "No such container" for exactly the logs somebody wanted.
    The name therefore has to come from the caller — and is accepted only if it
    matches a container that exists, which is a stronger check than any pattern.

    Args:
        argument: The container name.

    Returns:
        Process exit status.

    Raises:
        Refused: If the name is malformed or is not a container on this host.
    """
    if not argument or not _CONTAINER_NAME_RE.match(argument):
        raise Refused(f"not a valid container name: {argument!r}")
    if argument == CADDY_SERVICE and _proxy_native():
        # ``names`` lists the packaged Caddy too, so Diagnostics asks for it.
        return _journal_caddy(_CONTAINER_LOG_LINES)

    listing = subprocess.run(
        ["docker", *DAEMON_VERBS["names"]],
        cwd=str(PROJECT_DIR), capture_output=True, text=True, timeout=30,
    )
    if listing.returncode != 0:
        raise Refused(
            f"cannot list containers, so {argument!r} cannot be checked: "
            f"{listing.stderr.strip()}"
        )
    existing = {line.strip() for line in listing.stdout.splitlines() if line.strip()}
    if argument not in existing:
        raise Refused(f"no such container: {argument!r}")

    return _run(
        ["docker", "logs", "--tail", str(_CONTAINER_LOG_LINES), argument], timeout=60
    )


def _set_nodered_image(argument: str | None) -> int:
    """Change the Node-RED image tag in the compose file, and nothing else.

    The application used to rewrite the whole compose file to do this. That file
    is what ``docker compose up`` executes, so being able to write it is being
    able to run a container as root with the host filesystem mounted — the same
    hole as the docker group, reached through the update flow. Here the only
    thing that can change is one tag, matched by a fixed pattern.

    Args:
        argument: The image tag.

    Returns:
        0 on success.

    Raises:
        Refused: If the tag is implausible or the image line is not there.
    """
    if not argument or not _TAG_RE.match(argument):
        raise Refused(
            f"not a valid image tag: {argument!r}. Checked before it reaches the "
            "compose file, so a tag cannot smuggle YAML into it."
        )

    content = COMPOSE_FILE.read_text(encoding="utf-8")
    match = _NODERED_IMAGE_RE.search(content)
    if match is None:
        raise Refused(
            f"no 'image: nodered/node-red:<tag>' line in {COMPOSE_FILE}; refusing "
            "to guess what to edit"
        )
    if match.group("tag") == argument:
        _LOGGER.info("Node-RED image is already nodered/node-red:%s", argument)
        return 0

    updated = _NODERED_IMAGE_RE.sub(
        lambda m: f"{m.group('prefix')}{argument}{m.group('suffix')}", content, count=1
    )
    _back_up_if_customised(updated)

    with tempfile.NamedTemporaryFile(
        dir=PROJECT_DIR, delete=False, suffix=".tmp", mode="w", encoding="utf-8"
    ) as tmp:
        tmp.write(updated)
        tmp_path = tmp.name
    try:
        os.chmod(tmp_path, 0o644)
        if os.geteuid() == 0:
            os.chown(tmp_path, 0, 0)
        os.replace(tmp_path, COMPOSE_FILE)
    except OSError:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        raise
    _LOGGER.info(
        "Node-RED image set to nodered/node-red:%s (was %s)",
        argument, match.group("tag"),
    )
    return 0


def _caddy_images() -> tuple[str, str | None]:
    """The Caddy image this release pins, and the one the compose file uses.

    Returns:
        ``(pinned, configured)`` — configured is None when the compose file has
        no Caddy image line.

    Raises:
        Refused: If the template is not root-owned or pins nothing usable.
    """
    if not _root_owned(COMPOSE_TEMPLATE):
        raise Refused(f"template {COMPOSE_TEMPLATE} is missing or not root-owned")
    match = _CADDY_IMAGE_RE.search(COMPOSE_TEMPLATE.read_text(encoding="utf-8"))
    if match is None or not _PINNED_CADDY_RE.match(match.group("image")):
        raise Refused(f"{COMPOSE_TEMPLATE} pins no Caddy image this helper accepts")
    pinned = match.group("image")
    try:
        current = _CADDY_IMAGE_RE.search(COMPOSE_FILE.read_text(encoding="utf-8"))
    except OSError:
        current = None
    return pinned, current.group("image") if current else None


def _caddy_image_state() -> int:
    """Print the pinned and the configured Caddy image. Changes nothing.

    Returns:
        0, with JSON on stdout.
    """
    pinned, configured = _caddy_images()
    print(json.dumps({
        "pinned": pinned,
        "configured": configured,
        "update_available": configured != pinned,
    }))
    return 0


def _caddy_image_apply() -> int:
    """Move Caddy to the image this boneIO release pins.

    Caddy sat on ``caddy:2-alpine``: whatever that tag meant the day an image
    was built, never updated after, and different on every controller. The
    release now pins an exact version and digest in the root-owned template,
    and this takes that one value into the live compose file — the one line,
    nothing else, the way set-nodered-image changes only Node-RED's tag. There
    is no argument: which version Caddy runs is the release's decision, not the
    caller's.

    Then the image is pulled and only Caddy is recreated. The HTTPS panel drops
    for the seconds that takes.

    Returns:
        0 on success.

    Raises:
        Refused: If there is nothing to pin or no Caddy line to change.
    """
    pinned, configured = _caddy_images()
    if configured is None:
        raise Refused(f"no 'image: caddy:...' line in {COMPOSE_FILE}; refusing to guess")
    if configured != pinned:
        content = COMPOSE_FILE.read_text(encoding="utf-8")
        updated = _CADDY_IMAGE_RE.sub(
            lambda m: f"{m.group('prefix')}{pinned}{m.group('suffix')}", content, count=1
        )
        _back_up_if_customised(updated)
        with tempfile.NamedTemporaryFile(
            dir=PROJECT_DIR, delete=False, suffix=".tmp", mode="w", encoding="utf-8"
        ) as tmp:
            tmp.write(updated)
            tmp_path = tmp.name
        try:
            os.chmod(tmp_path, 0o644)
            if os.geteuid() == 0:
                os.chown(tmp_path, 0, 0)
            os.replace(tmp_path, COMPOSE_FILE)
        except OSError:
            try:
                os.unlink(tmp_path)
            except OSError:
                pass
            raise
        _LOGGER.info("Caddy image set to %s (was %s)", pinned, configured)
    rc = _run(_compose("pull", CADDY_SERVICE), timeout=900)
    if rc != 0:
        return rc
    return _run(_compose("up", "-d", CADDY_SERVICE), timeout=300)


def _apply_cloud_template(argument: str | None) -> int:
    """Switch the compose project to the cloud template.

    Args:
        argument: Must be absent. The template needs no parameter, so accepting
            one would only create a path for caller data to reach the file.

    Returns:
        Process exit status.

    Raises:
        Refused: If an argument is supplied.
    """
    if argument is not None:
        raise Refused("apply-cloud-template takes no argument")
    _LOGGER.info("applying the cloud compose template")
    return _install_compose(COMPOSE_CLOUD_TEMPLATE)


def _remove_cloud_template(argument: str | None) -> int:
    """Restore the plain compose template.

    Args:
        argument: Unused.

    Returns:
        Process exit status.
    """
    if argument is not None:
        raise Refused("remove-cloud-template takes no argument")
    _LOGGER.info("restoring the plain compose template")
    return _install_compose(COMPOSE_TEMPLATE)


def _proxy_native() -> bool:
    """Whether Caddy is the apt package rather than a container."""
    return NATIVE_MARKER.exists()


@contextlib.contextmanager
def _proxy_lock():
    """Hold the lock the switch between the modes holds, or refuse.

    The file is opened without following a link and used only if it is a
    regular file of root's own, so nothing planted in its place can have root
    create or lock a file elsewhere.

    Raises:
        Refused: If the lock is unusable, or still held after the wait.
    """
    try:
        fd = os.open(
            PROXY_LOCK, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600
        )
    except OSError as exc:
        raise Refused(f"cannot open {PROXY_LOCK}: {exc.strerror}") from None
    try:
        st = os.fstat(fd)
        if not stat.S_ISREG(st.st_mode) or st.st_uid != os.geteuid():
            raise Refused(f"{PROXY_LOCK} is not a regular file of root's own")
        deadline = time.monotonic() + PROXY_LOCK_TIMEOUT
        while True:
            try:
                fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
                break
            except BlockingIOError:
                if time.monotonic() >= deadline:
                    raise Refused(
                        f"{PROXY_LOCK} is still held after {PROXY_LOCK_TIMEOUT:g} s: "
                        "Caddy is being switched between the container and the "
                        "package. Try again when the switch has finished."
                    ) from None
                time.sleep(0.1)
        yield
    finally:
        os.close(fd)


def _touch(path: Path) -> None:
    """Create an empty marker, never through a link."""
    path.parent.mkdir(mode=0o755, parents=True, exist_ok=True)
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC, 0o644)
    # Whatever the caller's umask: the panel reads it as boneio.
    os.fchmod(fd, 0o644)
    os.close(fd)


def _set_cloud_marker(present: bool) -> int:
    """Have the packaged Caddy serve the cloud's certificate, or stop.

    The marker is the whole switch: the root generator reads it when
    ``systemctl reload`` runs it, and nothing from the caller reaches it.

    Returns:
        The reload's exit status.
    """
    if present:
        _touch(CLOUD_MARKER)
    else:
        with contextlib.suppress(FileNotFoundError):
            CLOUD_MARKER.unlink()
    _LOGGER.info("cloud marker %s", "set" if present else "cleared")
    return _run(list(NATIVE_VERBS["reload-caddy"]))


def _caddy_installed_version() -> str | None:
    """The installed Caddy package's version, or None.

    A package removed with its configuration kept still has a version in dpkg,
    so only an ``installed`` one counts.
    """
    rc, out = _capture(["dpkg-query", "-W", "--showformat=${db:Status-Status} ${Version}", "caddy"])
    status, _, version = out.strip().partition(" ")
    return version if rc == 0 and status == "installed" and version else None


def _sha512(path: Path) -> str:
    digest = hashlib.sha512()
    with open(path, "rb") as handle:
        for block in iter(lambda: handle.read(1 << 20), b""):
            digest.update(block)
    return digest.hexdigest()


def _fetch_caddy(log) -> Path:
    """The pinned Caddy .deb, downloaded unless the cache already holds it.

    Streamed to disk, not memory: the board has half a gigabyte of RAM.

    Raises:
        Refused: When it cannot be downloaded or is not the pinned file.
    """
    CADDY_CACHE.mkdir(mode=0o700, parents=True, exist_ok=True)
    # Root installs what it hashed here, so nobody else may be able to swap the
    # file in between: a directory that is a link, not root's, or writable by
    # group or others is refused, whoever made it.
    info = os.lstat(CADDY_CACHE)
    if not stat.S_ISDIR(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o022:
        raise Refused(f"{CADDY_CACHE} must be a directory only root can write")
    deb = CADDY_CACHE / CADDY_DEB_URL.rsplit("/", 1)[1]
    if deb.is_file() and _sha512(deb) == CADDY_DEB_SHA512:
        log.write(f"=== {deb} is already here\n")
        return deb
    log.write(f"=== downloading {CADDY_DEB_URL}\n")
    log.flush()
    fd, part = tempfile.mkstemp(dir=CADDY_CACHE, suffix=".part")
    try:
        with os.fdopen(fd, "wb") as out, urllib.request.urlopen(CADDY_DEB_URL, timeout=60) as resp:
            shutil.copyfileobj(resp, out, 1 << 20)
        got = _sha512(Path(part))
        if got != CADDY_DEB_SHA512:
            raise Refused(f"{CADDY_DEB_URL} has SHA-512 {got[:16]}…, not the pinned one")
        os.replace(part, deb)
    except OSError as exc:
        raise Refused(f"cannot download {CADDY_DEB_URL}: {exc}") from exc
    finally:
        with contextlib.suppress(FileNotFoundError):
            os.unlink(part)
    return deb


def _install_caddy(log) -> None:
    """Put the pinned Caddy package in place; nothing to do when it is.

    Raises:
        Refused: When the package cannot be had or installed.
    """
    if _caddy_installed_version() == CADDY_VERSION:
        log.write(f"=== caddy {CADDY_VERSION} is already installed\n")
        return
    deb = _fetch_caddy(log)
    # An earlier attempt cut off by a power loss mid-install leaves dpkg
    # half-configured, and apt-get refuses every install until this runs.
    # Not fatal: dpkg has no lock timeout, so an automatic update holding
    # the lock fails it at once, while the install below waits for it —
    # and fails clearly if dpkg really is broken.
    rc = _switch_cmd(["dpkg", "--configure", "-a"], log, 600)
    if rc != 0:
        log.write(f"=== dpkg --configure -a failed (rc={rc}); installing anyway\n")
    # apt-get rather than dpkg -i, so a missing dependency is fetched, not left
    # broken. The drop-in's condition keeps the postinst from starting Caddy.
    argv = [
        "apt-get", "install", "-y", "--no-install-recommends",
        "-o", "DPkg::Lock::Timeout=300", str(deb),
    ]
    rc = _switch_cmd(argv, log)
    if rc != 0:
        raise Refused(f"{' '.join(argv)} failed (rc={rc})")


def _native_image_state() -> int:
    """Print the installed and the candidate Caddy package. Changes nothing.

    A package removed with its configuration kept still has a version in dpkg,
    so only an ``installed`` one counts.

    Returns:
        0, with JSON on stdout.
    """
    installed = _caddy_installed_version()
    candidate = None
    _, out = _capture(["apt-cache", "policy", "caddy"])
    for line in out.splitlines():
        key, _, value = line.strip().partition(":")
        if key == "Candidate" and value.strip() not in ("", "(none)"):
            candidate = value.strip()
    print(json.dumps({"mode": "native", "installed": installed, "candidate": candidate}))
    return 0


def _native_caddy(verb: str, argument: str | None) -> int:
    """A Caddy verb, for the packaged Caddy. Never compose.

    Returns:
        Process exit status.

    Raises:
        Refused: On an argument the verb does not take, and on image apply.
    """
    if verb == "logs-caddy":
        return _journal_caddy(_line_count(argument))
    if argument is not None:
        raise Refused(f"verb {verb!r} takes no argument")
    if verb in NATIVE_VERBS:
        return _run(list(NATIVE_VERBS[verb]))
    if verb in ("apply-cloud-template", "remove-cloud-template"):
        return _set_cloud_marker(verb == "apply-cloud-template")
    if verb == "caddy-image-state":
        return _native_image_state()
    if verb == "caddy-image-apply":
        raise Refused("Caddy is pinned by boneIO; it updates with boneIO")
    if verb == "caddy-root-ca":
        return _caddy_root_ca()
    raise Refused(f"verb {verb!r} has no meaning for the packaged Caddy")


def _caddy_entry() -> dict:
    """The packaged Caddy, shaped like Docker's entry for a container.

    The application finds a service by ``Service`` or ``Name`` and checks
    ``State`` for ``running``, so these keys keep it working unchanged.
    """
    _, out = _capture(["systemctl", "is-active", "caddy"])
    word = out.strip() or "unknown"
    return {
        "Name": CADDY_SERVICE, "Names": CADDY_SERVICE, "Service": CADDY_SERVICE,
        "State": "running" if word in ("active", "reloading") else "exited",
        "Status": f"caddy.service {word}",
    }


def _with_native_caddy(verb: str) -> int:
    """Docker's answer to status, ps or names, with the packaged Caddy added.

    Compose prints either one JSON array or one object per line, depending on
    its version; the entry joins whichever form came back.

    Returns:
        Docker's exit status.
    """
    rc, out = _capture(_argv_for(verb), timeout=120)
    if verb == "names":
        extra = CADDY_SERVICE
    else:
        try:
            parsed = json.loads(out) if out.strip() else None
        except ValueError:
            parsed = None
        if isinstance(parsed, list):
            out, extra = "", json.dumps(parsed + [_caddy_entry()])
        else:
            extra = json.dumps(_caddy_entry())
    if out and not out.endswith("\n"):
        out += "\n"
    sys.stdout.write(out + extra + "\n")
    return rc


def _open_nofollow(path: Path, flags: int) -> int:
    """Open *path* without following a symlink at any component.

    ``O_NOFOLLOW`` alone covers only the last one: a directory swapped for a
    link would still lead root wherever it points.

    Returns:
        The open file descriptor.
    """
    parts = Path(path).parts
    fd = os.open("/", os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC)
    try:
        for name in parts[1:-1]:
            child = os.open(
                name,
                os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC,
                dir_fd=fd,
            )
            os.close(fd)
            fd = child
        return os.open(parts[-1], flags | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=fd)
    finally:
        os.close(fd)


def _caddy_root_ca() -> int:
    """Print the root certificate of Caddy's local authority.

    Read straight from Caddy's own data in both modes. In the container mode
    that is under /home/boneio, which the caller can write, so it is opened
    without following a link anywhere, non-blocking against a FIFO, and printed
    only if it is a small single-named regular file holding certificates and no
    key.

    Returns:
        0, with the PEM on stdout.

    Raises:
        Refused: If there is no certificate yet, or the file is not one.
    """
    base = CADDY_DATA if _proxy_native() else PROJECT_DIR / "caddy" / "data" / "caddy"
    path = base / "pki" / "authorities" / "local" / "root.crt"
    try:
        fd = _open_nofollow(path, os.O_RDONLY | os.O_NONBLOCK)
    except FileNotFoundError:
        raise Refused(f"{path} does not exist yet; Caddy creates it on its first start") from None
    except OSError as exc:
        raise Refused(f"{path} cannot be opened safely: {exc.strerror}") from None
    with os.fdopen(fd, "rb") as handle:
        st = os.fstat(handle.fileno())
        if not stat.S_ISREG(st.st_mode) or st.st_nlink != 1:
            raise Refused(f"{path} is not a plain file")
        data = handle.read(_MAX_ROOT_CA + 1)
    if (
        len(data) > _MAX_ROOT_CA
        or not data.lstrip().startswith(b"-----BEGIN CERTIFICATE-----")
        or b"PRIVATE KEY" in data
    ):
        raise Refused(f"{path} is not a certificate")
    sys.stdout.flush()
    sys.stdout.buffer.write(data)
    sys.stdout.flush()
    return 0


# ---------------------------------------------------------------- the switch
#
# Moving a controller from the Caddy container to the package. The order is
# what makes it safe: nothing the running proxy depends on changes until the
# package is installed and its configuration has passed the real Caddy's
# validation; from the moment the compose file is replaced, any failure puts
# the container back. The package then stays installed, so a later attempt
# does not download it again.


def _now() -> str:
    return datetime.datetime.now(datetime.UTC).isoformat(timespec="seconds")


def _switch_load() -> dict:
    """The last switch's record, or an empty dict when none has run."""
    try:
        data = json.loads(SWITCH_STATE.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {}
    return data if isinstance(data, dict) else {}


def _switch_save(state: dict) -> None:
    """Write the record atomically, readable by the panel."""
    SWITCH_DIR.mkdir(mode=0o755, parents=True, exist_ok=True)
    _write_root_file(SWITCH_STATE, json.dumps(state, indent=1))


def _write_root_file(path: Path, data: str) -> None:
    """Replace *path* atomically and durably with *data*, as root:root 0644.

    Mode and owner are set on the descriptor: the compose project's directory
    belongs to ``boneio``, who could put a link where the temporary file was.
    Durably, because a power loss is exactly what the switch's record and
    compose.before exist for: without the syncs, ext4 can bring back a new
    name with no data behind it.
    """
    with tempfile.NamedTemporaryFile(
        dir=path.parent, delete=False, suffix=".tmp", mode="w", encoding="utf-8"
    ) as tmp:
        os.fchmod(tmp.fileno(), 0o644)
        if os.geteuid() == 0:
            os.fchown(tmp.fileno(), 0, 0)
        tmp.write(data)
        tmp.flush()
        os.fsync(tmp.fileno())
        tmp_path = tmp.name
    try:
        os.replace(tmp_path, path)
    except OSError:
        with contextlib.suppress(OSError):
            os.unlink(tmp_path)
        raise
    directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC)
    try:
        os.fsync(directory)
    finally:
        os.close(directory)


def _open_switch_log(append: bool):
    """The switch's log, opened without following a link, readable by all."""
    SWITCH_DIR.mkdir(mode=0o755, parents=True, exist_ok=True)
    flags = os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC
    fd = os.open(SWITCH_LOG, flags | (os.O_APPEND if append else os.O_TRUNC), 0o644)
    os.fchmod(fd, 0o644)
    return os.fdopen(fd, "w", encoding="utf-8")


def _read_own(path: Path, limit: int | None = _MAX_ROOT_CA) -> str:
    """A file only this helper's owner can have written, read without links.

    Raises:
        OSError: If it cannot be opened that way.
        ValueError: If it is not a plain file of root's, or is too large.
    """
    fd = _open_nofollow(path, os.O_RDONLY | os.O_NONBLOCK)
    with os.fdopen(fd, "rb") as handle:
        st = os.fstat(handle.fileno())
        if (
            not stat.S_ISREG(st.st_mode) or st.st_uid != os.geteuid()
            or st.st_mode & 0o022
        ):
            raise ValueError(f"{path} is not a plain file of root's")
        data = handle.read() if limit is None else handle.read(limit + 1)
    if limit is not None and len(data) > limit:
        raise ValueError(f"{path} is too large")
    return data.decode("utf-8", "replace")


def _switch_unit_active() -> bool:
    """Whether a switch is running right now."""
    try:
        result = subprocess.run(
            ["systemctl", "is-active", SWITCH_UNIT],
            capture_output=True, text=True, timeout=15,
        )
    except (OSError, subprocess.SubprocessError):
        return False
    return result.stdout.strip() in ("active", "activating", "reloading")


def _in_switch_unit() -> bool:
    """Whether this process runs inside the switch's own unit."""
    try:
        cgroup = Path("/proc/self/cgroup").read_text(encoding="utf-8")
    except OSError:
        return False
    # In the system slice: a user's own unit of the same name is not it.
    return f"/system.slice/{SWITCH_UNIT}" in cgroup


def _switch_cmd(argv: list[str], log, timeout: int = 900) -> int:
    """Run one command of the switch, its output going straight to the log.

    Returns:
        Its exit status; 1 when it cannot run or runs out of time.
    """
    log.write(f"\n$ {' '.join(argv)}\n")
    log.flush()
    cwd = str(PROJECT_DIR) if argv[0] == "docker" else "/"
    try:
        return subprocess.run(
            argv, cwd=cwd, stdout=log, stderr=subprocess.STDOUT,
            stdin=subprocess.DEVNULL, env={**APT_ENV, "HOME": "/root"},
            timeout=timeout, check=False,
        ).returncode
    except subprocess.TimeoutExpired:
        log.write(f"\n=== killed after {timeout}s\n")
    except OSError as exc:
        log.write(f"\n=== cannot run {argv[0]}: {exc}\n")
    return 1


def _https_port() -> int:
    """The HTTPS port in the compose ``.env``, by the generator's rules.

    The file is ``boneio``'s, so it is opened as the root certificate is. Only
    the generator's per-value rules are repeated here; a port it refuses for
    colliding with another one makes the probe fail, and the switch roll back.
    """
    port = 8443
    try:
        fd = _open_nofollow(ENV_FILE, os.O_RDONLY | os.O_NONBLOCK)
    except OSError:
        return port
    with os.fdopen(fd, "rb") as handle:
        regular = stat.S_ISREG(os.fstat(handle.fileno()).st_mode)
        data = handle.read(_MAX_ROOT_CA + 1) if regular else b""
    if len(data) > _MAX_ROOT_CA:
        return port
    for line in data.decode("utf-8", "replace").splitlines():
        line = line.strip()
        if line.startswith("export "):
            line = line[len("export "):].lstrip()
        name, sep, value = line.partition("=")
        if not sep or name.strip() != "HTTPS_PORT":
            continue
        value = value.strip()
        if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'":
            value = value[1:-1]
        valid = len(value) <= 5 and value.isascii() and value.isdigit()
        port = int(value) if valid and 1024 <= int(value) <= 65535 else 8443
    return port


def _status(port: int, path: str) -> int | None:
    """The HTTP status of one request to the local proxy, None for no answer."""
    context = ssl.create_default_context()
    # The device's own authority, which root has no reason to trust.
    context.check_hostname = False
    context.verify_mode = ssl.CERT_NONE
    opener = urllib.request.build_opener(
        urllib.request.ProxyHandler({}), urllib.request.HTTPSHandler(context=context)
    )
    try:
        with opener.open(f"https://127.0.0.1:{port}{path}", timeout=5) as reply:
            return reply.status
    except urllib.error.HTTPError as exc:
        exc.close()
        return exc.code
    except (OSError, http.client.HTTPException, ValueError):
        return None


def _proxy_answers(port: int) -> bool:
    """Whether the proxy on *port* serves both of its routes.

    ``/nodered-status`` is answered by Caddy itself, whatever boneIO is doing.
    ``/api/init`` goes to the panel and only has to come back as HTTP: the
    starting page while boneIO restarts still proves the route.
    """
    return _status(port, "/nodered-status") == 200 and _status(port, "/api/init") is not None


def _wait_for_proxy(port: int) -> bool:
    """Ask the proxy until it answers or :data:`PROBE_TIMEOUT` has passed."""
    deadline = time.monotonic() + PROBE_TIMEOUT
    while True:
        if _proxy_answers(port):
            return True
        if time.monotonic() >= deadline:
            return False
        time.sleep(PROBE_INTERVAL)


def _shape(text: str) -> str:
    """*text* with the two lines this helper changes for the panel blanked."""
    text = _NODERED_IMAGE_RE.sub(lambda m: f"{m['prefix']}*{m['suffix']}", text, count=1)
    return _CADDY_IMAGE_RE.sub(lambda m: f"{m['prefix']}*{m['suffix']}", text, count=1)


def _template_shapes() -> list[str]:
    """The root-owned Caddy templates, as :func:`_shape` sees them."""
    return [
        _shape(template.read_text(encoding="utf-8"))
        for template in (COMPOSE_TEMPLATE, COMPOSE_CLOUD_TEMPLATE)
        if _root_owned(template)
    ]


def _switch_preconditions() -> tuple[str, str, bool, str]:
    """What the switch starts from, or why it cannot start.

    The live compose file must be root's and one of the two Caddy templates,
    up to the Node-RED version and the Caddy image, which this helper itself
    changes on the panel's behalf. Anything else is a file somebody edited,
    and replacing it would discard that. The Node-RED version is carried into
    the native compose file, so the switch does not also update Node-RED.

    Returns:
        The live compose file, the native one to install in its place,
        whether the live one is the cloud variant, and the Node-RED tag.

    Raises:
        Refused: If the switch must not run.
    """
    if _proxy_native():
        raise Refused("Caddy is already the package")
    _assert_compose_trustworthy()
    missing = [
        str(path) for path in (*SWITCH_PREREQUISITES, COMPOSE_NATIVE_TEMPLATE)
        if not _root_owned(path)
    ]
    if missing:
        raise Refused("not installed by a migration yet: " + ", ".join(missing))
    live = COMPOSE_FILE.read_text(encoding="utf-8")
    if _shape(live) not in _template_shapes():
        raise Refused(
            f"{COMPOSE_FILE} is not one of the Caddy templates; it is left as it is"
        )
    tag = _NODERED_IMAGE_RE.search(live)
    native = COMPOSE_NATIVE_TEMPLATE.read_text(encoding="utf-8")
    if tag is None or not _TAG_RE.match(tag["tag"]) or not _NODERED_IMAGE_RE.search(native):
        raise Refused("the Node-RED image cannot be carried over")
    native = _NODERED_IMAGE_RE.sub(
        lambda m: f"{m['prefix']}{tag['tag']}{m['suffix']}", native, count=1
    )
    # The init script is what differs between the two, as the panel tells them.
    return live, native, "init-certs-cloud.sh" in live, tag["tag"]


def _read_old_ca() -> dict[str, bytes]:
    """The container's authority, if every file of it is one root can vouch for.

    The tree is ``boneio``'s, so each file is opened without following a link
    anywhere, must be a single-named regular file under the size cap and look
    like what its name says. Nothing else under pki is read.

    Raises:
        ValueError: Naming the first file that is missing or unacceptable.
    """
    found: dict[str, bytes] = {}
    for name in CA_FILES:
        try:
            fd = _open_nofollow(OLD_CA / name, os.O_RDONLY | os.O_NONBLOCK)
        except OSError as exc:
            raise ValueError(f"{name}: {exc.strerror}") from None
        with os.fdopen(fd, "rb") as handle:
            st = os.fstat(handle.fileno())
            if not stat.S_ISREG(st.st_mode) or st.st_nlink != 1:
                raise ValueError(f"{name} is not a plain file")
            data = handle.read(_MAX_ROOT_CA + 1)
        if len(data) > _MAX_ROOT_CA:
            raise ValueError(f"{name} is too large")
        if name.endswith(".crt"):
            ok = data.lstrip().startswith(b"-----BEGIN CERTIFICATE-----") and b"PRIVATE KEY" not in data
        else:
            ok = data.lstrip().startswith(b"-----BEGIN") and b"PRIVATE KEY-----" in data
        if not ok:
            raise ValueError(f"{name} is not what its name says")
        found[name] = data
    return found


def _carry_over_ca(log) -> None:
    """Give the packaged Caddy the container's authority, unless it has one.

    People installed that authority's root; a new one would make every browser
    warn again. All four files or none: half an authority stops Caddy. The tree
    is built beside its destination and renamed into place, so an interrupted
    copy never leaves a ``pki`` that the next attempt would then keep.
    """
    pki = CADDY_DATA / "pki"
    if os.path.lexists(pki):
        log.write("=== the packaged Caddy has an authority already; it is kept\n")
        return
    try:
        files = _read_old_ca()
    except ValueError as exc:
        log.write(f"=== the container's authority is not carried over ({exc}); Caddy creates a new one\n")
        return
    owner = None
    if os.geteuid() == 0:
        entry = pwd.getpwnam("caddy")
        owner = (entry.pw_uid, entry.pw_gid)

    # The package creates Caddy's home; the data directories below it appear
    # on its first start, which has not happened.
    missing = []
    directory = CADDY_DATA
    while not os.path.lexists(directory):
        missing.append(directory)
        directory = directory.parent
    staging = CADDY_DATA / ".pki.boneio"
    shutil.rmtree(staging, ignore_errors=True)
    local = staging / "authorities" / "local"
    for directory in [*reversed(missing), staging, staging / "authorities", local]:
        os.mkdir(directory, 0o700)
        if owner:
            os.chown(directory, *owner, follow_symlinks=False)
    for name, data in files.items():
        mode = 0o600 if name.endswith(".key") else 0o644
        fd = os.open(
            local / name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC, mode
        )
        with os.fdopen(fd, "wb") as handle:
            os.fchmod(fd, mode)
            if owner:
                os.fchown(fd, *owner)
            handle.write(data)
    os.rename(staging, pki)
    log.write("=== the container's authority is carried over\n")


def _mirror_cloud_marker(cloud: bool) -> bool:
    """Make the cloud marker say what the compose file says.

    A marker left by an earlier attempt would otherwise have a plain device
    serve the cloud's certificate.

    Returns:
        True when this created it.
    """
    existed = os.path.lexists(CLOUD_MARKER)
    if cloud:
        _touch(CLOUD_MARKER)
        return not existed
    with contextlib.suppress(FileNotFoundError):
        CLOUD_MARKER.unlink()
    return False


def _container_compose(state: dict, log) -> str:
    """The container's compose file to put back.

    compose.before when it is root's, not empty and still one of the Caddy
    templates. Otherwise - a power loss can leave it empty - it is rebuilt
    from the root-owned template the run recorded, with its Node-RED version.

    Raises:
        Refused: If neither is available.
    """
    try:
        before = _read_own(COMPOSE_BEFORE)
        if before.strip() and _shape(before) in _template_shapes():
            return before
        log.write(f"=== {COMPOSE_BEFORE} is not a Caddy template\n")
    except (OSError, ValueError) as exc:
        log.write(f"=== {COMPOSE_BEFORE} cannot be used: {exc}\n")
    template = COMPOSE_CLOUD_TEMPLATE if state.get("cloud") else COMPOSE_TEMPLATE
    if not _root_owned(template):
        raise Refused(f"template {template} is missing or not root-owned")
    text = template.read_text(encoding="utf-8")
    tag = state.get("nodered_tag")
    if isinstance(tag, str) and _TAG_RE.match(tag):
        text = _NODERED_IMAGE_RE.sub(
            lambda m: f"{m['prefix']}{tag}{m['suffix']}", text, count=1
        )
    log.write(f"=== the compose file is rebuilt from {template.name}\n")
    return text


def _switch_rollback(log, state: dict, port: int) -> list[str]:
    """Put the container back. Every step is tried whatever the last one did.

    Returns:
        What went wrong; empty when the container answers again.
    """
    problems = []
    log.write("\n=== rollback\n")
    if _switch_cmd(["systemctl", "disable", "--now", "caddy"], log, 300) != 0:
        problems.append("the packaged Caddy could not be stopped")
    created_cloud = bool(state.get("created_cloud"))
    for marker in (NATIVE_MARKER, CLOUD_MARKER) if created_cloud else (NATIVE_MARKER,):
        try:
            marker.unlink()
        except FileNotFoundError:
            pass
        except OSError as exc:
            problems.append(f"{marker} could not be removed: {exc.strerror}")
    try:
        _write_root_file(COMPOSE_FILE, _container_compose(state, log))
    except Exception as exc:  # noqa: BLE001 - the rest must still be tried
        problems.append(f"the compose file could not be restored: {exc}")
    if _switch_cmd(_compose("up", "-d"), log) != 0:
        problems.append("docker compose up failed")
    elif not _wait_for_proxy(port):
        problems.append(f"the container did not answer on {port} within {PROBE_TIMEOUT:g} s")
    log.write(f"=== rollback {'incomplete: ' + '; '.join(problems) if problems else 'done'}\n")
    return problems


def _switch_steps(state: dict, log) -> None:
    """The switch itself, recording each step in *state*."""
    def step(name: str) -> None:
        state["step"] = name
        _switch_save(state)
        log.write(f"\n=== {name}\n")
        log.flush()

    def run(argv: list[str], timeout: int = 900) -> None:
        rc = _switch_cmd(argv, log, timeout)
        if rc != 0:
            raise Refused(f"{state['step']}: {' '.join(argv)} failed (rc={rc})")

    port = 8443
    try:
        step("preconditions")
        live, native, cloud, tag = _switch_preconditions()
        port = _https_port()
        step("install-caddy")
        _install_caddy(log)
        step("carry-over")
        _carry_over_ca(log)
        # The device's name now, not the one the container recorded (its
        # default is "boneio"): a different name makes the generator discard
        # the authority just carried over.
        _write_root_file(HOSTNAME_FILE, f"{socket.gethostname()}\n")
        # Recorded at once: recovery after a power loss removes the marker
        # only if this run made it.
        state["created_cloud"] = _mirror_cloud_marker(cloud)
        # What a rollback rebuilds the container's compose file from, should
        # compose.before itself not survive.
        state["cloud"], state["nodered_tag"] = cloud, tag
        _switch_save(state)
        step("check")
        run([PROXY_CONFIG, "--check"], 300)

        step("compose")
        _write_root_file(COMPOSE_BEFORE, live)
        # From here on a failure, or the boot after an interruption, rolls
        # back; compose.before is this run's, not an older attempt's.
        state["replaced"] = True
        _switch_save(state)
        _write_root_file(COMPOSE_FILE, native)
        # Removes the Caddy container, which frees 8091 and 8443.
        run(_compose("up", "-d", "--remove-orphans"))
        step("start")
        _touch(NATIVE_MARKER)
        run(["systemctl", "daemon-reload"], 120)
        run(["systemctl", "enable", "--now", "caddy"], 300)
        step("probe")
        if not _wait_for_proxy(port):
            raise Refused(
                f"probe: the packaged Caddy did not answer on {port} within {PROBE_TIMEOUT:g} s"
            )
        # Caddy creates its authority on the first start; the reload has the
        # generator export the root. Missing it costs only the download.
        if _switch_cmd(["systemctl", "reload", "caddy"], log, 120) != 0:
            log.write("=== the root certificate is exported on the next start instead\n")
        state["state"] = "done"
    # Any error at all: past the compose file, the rollback is what stands
    # between a failure and a controller with no proxy.
    except Exception as exc:  # noqa: BLE001
        state["error"] = str(exc) or type(exc).__name__
        log.write(f"\n=== FAILED: {state['error']}\n")
        _switch_undo(state, log, port)
        _LOGGER.error("proxy switch: %s", state["error"])


def _switch_undo(state: dict, log, port: int) -> None:
    """Undo what *state* records a failed or interrupted run as having done.

    Once the compose file was replaced, that is the whole rollback; before
    that, only a cloud marker the run created is left to remove.
    """
    if state.get("replaced"):
        try:
            problems = _switch_rollback(log, state, port)
        except Exception as exc:  # noqa: BLE001 - the record must say so
            problems = [f"stopped: {exc}"]
        state["state"] = "rolled_back"
        if problems:
            state["error"] += "; rollback: " + "; ".join(problems)
    else:
        state["state"] = "failed"
        if state.get("created_cloud"):
            with contextlib.suppress(OSError):
                CLOUD_MARKER.unlink()


def _undo_interrupted(state: dict, log) -> None:
    """Undo the run *state* records as cut short, and record that."""
    log.write("\n=== the last switch was interrupted; undoing it\n")
    state["error"] = "interrupted"
    _switch_undo(state, log, _https_port())
    state["at"] = _now()
    _switch_save(state)
    _LOGGER.error("proxy switch: %s", state["error"])


def _switch_recover() -> None:
    """Undo a switch that was cut short; the caller holds the proxy lock.

    A record saying ``running`` with no unit running it is a switch killed
    or powered off halfway, possibly in the moment when neither Caddy runs.
    Anything else is left alone, so this can run on every boot.
    """
    state = _switch_load()
    if state.get("state") != "running" or _switch_unit_active():
        return
    with _open_switch_log(append=True) as log:
        _undo_interrupted(state, log)


def _switch_run() -> int:
    """Move Caddy from the container to the package. Only inside its unit.

    Holds the proxy lock from the first step to the last, so no Caddy verb acts
    on the mode being left; everything in here therefore calls the unlocked
    internals.

    Raises:
        Refused: When called any other way.
    """
    if not _in_switch_unit():
        raise Refused(f"proxy-switch-run only runs inside {SWITCH_UNIT}; use proxy-switch-start")
    with _proxy_lock(), _open_switch_log(append=False) as log:
        previous = _switch_load()
        if previous.get("state") == "running":
            # This unit is the only one that runs a switch, and it is this
            # one: a record saying running is a run that was cut short.
            _undo_interrupted(previous, log)
        attempts = previous.get("attempts")
        state = {
            "state": "running", "step": None, "error": None,
            "attempts": (attempts if isinstance(attempts, int) else 0) + 1, "at": _now(),
            "created_cloud": False, "replaced": False,
        }
        _switch_save(state)
        try:
            _switch_steps(state, log)
        finally:
            state["at"] = _now()
            _switch_save(state)
    _LOGGER.info("proxy switch: %s", state["state"])
    return 0 if state["state"] == "done" else 1


def _switch_start() -> int:
    """Start the switch in its own unit, and return.

    Not here: this process is a child of boneio.service, and the switch must
    survive boneIO restarting. The fixed unit name means systemd itself
    refuses a second switch while one runs.

    An interrupted switch is undone by the run, inside the unit, before it
    starts again; it may have left the native marker behind.

    Raises:
        Refused: While a switch runs, or once Caddy is the package.
    """
    if _switch_unit_active():
        raise Refused("the switch to the packaged Caddy is already running")
    if _proxy_native() and _switch_load().get("state") != "running":
        raise Refused("Caddy is already the package")
    rc = _run([
        "systemd-run", "--unit", SWITCH_UNIT, "--collect", "--no-block", "--quiet",
        "--description", "boneIO switch to the packaged Caddy",
        # boneIO runs at CPUWeight=1000; the switch must not starve it.
        "--property", "CPUWeight=50", "--property", "IOWeight=50",
        HELPER_SELF, "proxy-switch-run",
    ])
    if rc == 0:
        print(json.dumps({"started": True}))
    return rc


def _switch_state() -> int:
    """Print the last switch's record and the end of its log. Changes nothing.

    A record still saying ``running`` with no unit left is a switch that was
    killed or lost power halfway; it is shown as failed, or the panel would
    wait for it forever.
    """
    state = {"state": None, "step": None, "error": None, "attempts": 0, "at": None}
    state.update(_switch_load())
    running = _switch_unit_active()
    if state["state"] == "running" and not running:
        state["state"], state["error"] = "failed", "interrupted"
    try:
        lines = _read_own(SWITCH_LOG, limit=None).splitlines()
    except (OSError, ValueError):
        lines = []
    print(json.dumps({
        **state, "running": running, "native": _proxy_native(),
        "log": "\n".join(lines[-SWITCH_LOG_TAIL:]),
    }))
    return 0


def _switch(verb: str, argument: str | None) -> int:
    if argument is not None:
        raise Refused(f"verb {verb!r} takes no argument")
    if verb == "proxy-switch-start":
        return _switch_start()
    if verb == "proxy-switch-run":
        return _switch_run()
    if verb == "caddy-install":
        # The image build's way in; it has no switch to run.
        with _proxy_lock():
            _install_caddy(sys.stdout)
        return 0
    if verb == "proxy-switch-recover":
        # Waiting for the lock is pointless while the switch itself runs.
        if not _switch_unit_active():
            with _proxy_lock():
                _switch_recover()
        return 0
    return _switch_state()


def selftest() -> int:
    """Check the helper can do its job, changing nothing.

    Returns:
        0 when healthy.
    """
    problems: list[str] = []
    if shutil.which("docker") is None:
        problems.append("docker is not installed")
    if not PROJECT_DIR.is_dir():
        problems.append(f"the compose project directory {PROJECT_DIR} does not exist")
    if not _root_owned(COMPOSE_FILE):
        problems.append(
            f"{COMPOSE_FILE} is not root-owned, so every write verb will refuse"
        )
    for template in (COMPOSE_TEMPLATE, COMPOSE_CLOUD_TEMPLATE):
        if not _root_owned(template):
            problems.append(f"template {template} is missing or not root-owned")
    if _proxy_native() and shutil.which("caddy") is None:
        problems.append(
            f"{NATIVE_MARKER} says Caddy is the package, but caddy is not installed"
        )

    for problem in problems:
        _LOGGER.error("selftest: %s", problem)
    if problems:
        return 1
    _LOGGER.info("selftest: %d verbs available, compose file trusted.", len(ALL_VERBS))
    return 0


def _dispatch(verb: str, argument: str | None) -> int:
    """Run one allowed verb, taking no lock.

    :func:`main` holds :func:`_proxy_lock` around this for :data:`LOCKED_VERBS`;
    code that already holds the lock calls this directly, since a second
    ``flock`` from the same process would wait on the first. The mode is read
    here, so under the lock it cannot change in between.

    Returns:
        Process exit status.

    Raises:
        Refused: If the request cannot be carried out safely.
    """
    native = _proxy_native()
    if native and verb in CADDY_VERBS:
        return _native_caddy(verb, argument)

    if verb not in READ_ONLY_VERBS:
        _assert_compose_trustworthy()

    if verb == "logs-caddy":
        return _logs(CADDY_SERVICE, argument)
    if verb == "logs-nodered":
        return _logs(NODERED_SERVICE, argument)
    if verb == "logs-container":
        return _logs_container(argument)
    if verb == "set-nodered-image":
        return _set_nodered_image(argument)
    if verb in COMPOSITE_VERBS and argument is not None:
        raise Refused(f"verb {verb!r} takes no argument")
    if verb == "caddy-image-state":
        return _caddy_image_state()
    if verb == "caddy-image-apply":
        return _caddy_image_apply()
    if verb == "caddy-root-ca":
        return _caddy_root_ca()
    if verb == "apply-cloud-template":
        return _apply_cloud_template(argument)
    if verb == "remove-cloud-template":
        return _remove_cloud_template(argument)

    if argument is not None:
        raise Refused(f"verb {verb!r} takes no argument")
    if native and verb in ("status", "ps", "names"):
        return _with_native_caddy(verb)
    # Pulling an image onto a BeagleBone takes minutes, as for Caddy above.
    timeout = 900 if VERBS.get(verb, ("",))[0] == "pull" else 120
    return _run(_argv_for(verb), timeout=timeout)


def main(argv: list[str] | None = None) -> int:
    """Entry point.

    Returns:
        Process exit status.
    """
    parser = argparse.ArgumentParser(
        description="boneIO privileged container operations"
    )
    parser.add_argument("verb", nargs="?", help=f"one of: {', '.join(ALL_VERBS)}")
    # A value starting with "-" would otherwise be parsed as an option and make
    # argparse exit before the domain validator ever sees it. The validator
    # should be the thing that rejects it, with a message that says why.
    parser.add_argument(
        "argument", nargs="?", default=None,
        help="log line count, or a domain",
    )
    parser.add_argument(
        "--list-verbs", action="store_true", help="print the allowed verbs as JSON"
    )
    parser.add_argument(
        "--selftest", action="store_true", help="check the helper, change nothing"
    )
    args, unparsed = parser.parse_known_args(argv)
    if unparsed:
        # Anything argparse could not place is still caller input, so it is
        # refused rather than ignored.
        if args.argument is None and len(unparsed) == 1:
            args.argument = unparsed[0]
        else:
            _LOGGER.error("REFUSED: unexpected arguments: %s", unparsed)
            return 1

    if args.list_verbs:
        print(json.dumps(ALL_VERBS))
        return 0

    _assert_root()

    if args.selftest:
        return selftest()

    if not args.verb:
        parser.error("a verb is required")

    try:
        if args.verb not in ALL_VERBS:
            raise Refused(
                f"unknown verb {args.verb!r}. Allowed: {', '.join(ALL_VERBS)}"
            )

        if args.verb in SWITCH_VERBS:
            # Before the compose check: a run refused for its preconditions
            # must still record why.
            return _switch(args.verb, args.argument)
        if args.verb in LOCKED_VERBS:
            with _proxy_lock():
                return _dispatch(args.verb, args.argument)
        return _dispatch(args.verb, args.argument)
    except Refused as exc:
        _LOGGER.error("REFUSED: %s", exc)
        return 2 if "not owned by root" in str(exc) else 1


if __name__ == "__main__":
    sys.exit(main())
