#!/usr/bin/env python3
"""boneio-system — the remaining privileged system operations, by name.

Three features needed root and got it the two worst ways available: a broad
sudo rule, or an API endpoint that asked the operator for their system password.
The password is the same on every controller that shipped, so an endpoint
collecting it is a way to intercept it; and the rules it stood in for were
wildcards — ``ip link set can0 *`` and a ``sed -i`` expression built in the
application and run as root on /boot/uEnv.txt.

This replaces both with named operations:

    can-up <iface> <bitrate>       down, set bitrate, up
    can-down <iface>
    can-restart <iface>            recover from bus-off without a full cycle
    overlay-get                    read the device-tree overlay from uEnv.txt
    overlay-set <basename>         change it, from a fixed list of overlays
    hostname-set <name>
    ntp-get                        read the configured NTP servers
    ntp-set <servers|default>      point systemd-timesyncd at given servers
    mqtt-password <account>        set a broker password, read from stdin
    mqtt-reload                    have the broker re-read its password file
    service-password-state         is the boneio login locked, shipped or set
    service-password-init          set it once, from stdin — see below
    os-update-state                apt: last run, reboot needed, will the kernel boot
    os-update-start <check|upgrade>  run apt in its own unit, return at once
    os-update-run <check|upgrade>  what that unit runs; refused anywhere else
    os-update-log                  the end of the last run's log
    os-autoupdate-set <on|off>     automatic security updates (never a reboot)

Every argument is checked against a closed set before anything runs, and the
uEnv.txt edit is done here with a compiled pattern rather than by handing a
``sed`` expression to root. Nothing takes a password: the sudoers rule for this
helper is NOPASSWD, which is safe precisely because the vocabulary is closed.

Exit codes:
    0  success
    1  refused, or the underlying command failed
"""

from __future__ import annotations

import argparse
import ipaddress
import json
import logging
import logging.handlers
import os
import re
import shutil
import subprocess
import sys
import tempfile
import threading
import time
from pathlib import Path

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

#: The CAN interfaces this board has, plus the virtual one the schema allows for
#: testing. A name from the caller would otherwise reach `ip link set`, where it
#: is not just a device but a namespace and a rename away from something else.
CAN_INTERFACES = ("can0", "can1", "vcan0")
#: Bitrates the schema accepts, and only those. Kept identical to the `allowed`
#: list under `can.bitrate` in boneio/schema/schema.yaml: a helper that takes
#: more than the schema does is a wider grant than anything can ask for, and one
#: that takes fewer silently breaks a configuration the schema called valid.
CAN_BITRATES = (10_000, 20_000, 50_000, 100_000, 125_000, 250_000, 500_000,
                1_000_000)

#: Device-tree overlays shipped for this board. The application already kept
#: this list to reject unknown filenames; it is repeated here because the
#: helper cannot trust the application's copy of it.
VALID_OVERLAYS = (
    "BONEIO-BLACK-PINS-v0.2-v0.3.dtbo",
    "BONEIO-BLACK-PINS-v0.4-v0.8.dtbo",
    "BONEIO-BLACK-PINS-v1.0.dtbo",
    "BONEIO-BLACK-PINS.dtbo",
)

UENV_PATHS = (Path("/boot/firmware/uEnv.txt"), Path("/boot/uEnv.txt"))

#: An uncommented uboot_overlay_addr line naming a boneIO overlay, with or
#: without a directory prefix. Compiled here rather than passed to `sed -i` as
#: an expression the application composes.
_OVERLAY_LINE_RE = re.compile(
    r"^(?P<indent>[^\S\n]*)"
    r"(?P<key>uboot_overlay_addr[0-9]*=)"
    r"(?P<dir>.*/)?"
    r"(?P<name>BONEIO-BLACK-PINS[^\s]*)"
    r"(?P<rest>.*)$"
)

#: A hostname: DNS label rules, which is stricter than hostnamectl's own.
_HOSTNAME_RE = re.compile(r"^(?!-)[a-z0-9-]{1,63}(?<!-)$")

#: Where the NTP servers chosen in the web UI are written. A drop-in rather than
#: timesyncd.conf itself, so the distribution's file stays untouched and
#: "use the defaults" is a file deletion rather than an edit that has to guess
#: what the defaults were.
NTP_DROPIN = Path("/etc/systemd/timesyncd.conf.d/boneio.conf")

#: At most this many servers. systemd accepts more; the UI offers this many and
#: a bounded list keeps the generated file a fixed, reviewable shape.
NTP_MAX_SERVERS = 5

#: The broker's password file, and the accounts the application manages in it.
#: Naming them here rather than taking any account keeps this from becoming a
#: way to add a broker login nobody asked for.
MOSQUITTO_PASSWD = Path("/etc/mosquitto/passwd")
MQTT_ACCOUNTS = ("boneio", "homeassistant", "mqtt")
#: mosquitto_passwd rewrites the file with a mode of its own, so the permissions
#: have to be re-applied after every write. Found the hard way: the file shipped
#: 0644 and was seen at 0704, which let any local account take the hashes for an
#: offline crack (F-11). The broker runs as its own user, so the group read bit
#: has to stay or the fix takes the broker down with the exposure.
MOSQUITTO_PASSWD_MODE = 0o640
MOSQUITTO_PASSWD_OWNER = ("root", "mosquitto")

#: The login this helper serves, and the only account whose password it may
#: touch. Not a parameter: a verb that set "an account's password" would be a
#: way to give any local account a password its owner never chose.
SERVICE_ACCOUNT = "boneio"
SHADOW = Path("/etc/shadow")
#: What every image before 1.6 shipped as that account's password, identical on
#: every unit and published. Recognised so the panel can say it is still there;
#: never written by this helper.
SHIPPED_PASSWORD = "Black"
#: Written once the owner's password has been set. Root-owned and outside the
#: application's reach, so the one-shot cannot be re-armed by relocking the
#: account — see _service_password_init.
SERVICE_PASSWORD_FLAG = Path("/var/lib/boneio/service-password.set")
#: Generous, and well above anything a person types. chpasswd has no limit of
#: its own worth relying on, and a caller able to hand root a megabyte on stdin
#: is a caller who should be told no.
SERVICE_PASSWORD_MAX = 1024

#: One DNS name. Addresses are validated with :mod:`ipaddress` instead, so this
#: only has to cover hostnames.
#:
#: The point of validating at all is not that systemd would choke on a bad name.
#: It is that this value is written into a systemd configuration file: a space,
#: a newline or a bracket in it would let the caller append directives of their
#: own to a file parsed as root. The character set below has no way to express
#: any of that.
_NTP_HOSTNAME_RE = re.compile(
    r"^(?=.{1,253}$)(?!-)[A-Za-z0-9-]{1,63}(?<!-)"
    r"(\.(?!-)[A-Za-z0-9-]{1,63}(?<!-))*$"
)

#: This helper's own path: the transient update unit runs it again, as root,
#: outside boneio.service.
HELPER_SELF = "/usr/sbin/boneio-system"
#: What the caller may ask of apt. Nothing else reaches it.
OS_UPDATE_MODES = ("check", "upgrade")
#: A fixed name, so systemd itself refuses a second update while one runs.
OS_UPDATE_UNIT = "boneio-os-update.service"
OS_UPDATE_STATE = Path("/var/lib/boneio/os-update/state.json")
OS_UPDATE_LOG = Path("/var/log/boneio-os-update.log")
OS_UPDATE_LOG_TAIL = 400
#: A kernel upgrade alone unpacks ~60 MB and builds an initramfs next to the old
#: one; the eMMC rootfs has little to spare. Below this, refuse up front rather
#: than let dpkg run out of space halfway through.
OS_UPDATE_MIN_FREE_MB = 300
#: Debian writes these only when a package asks; the kernel check below is what
#: actually tells a BeagleBone it needs a reboot.
REBOOT_REQUIRED = Path("/run/reboot-required")
REBOOT_REQUIRED_PKGS = Path("/run/reboot-required.pkgs")
#: Packages whose new code is not in use until the device restarts.
REBOOT_PACKAGE_PREFIXES = ("linux-image", "libc6", "systemd", "dbus", "libssl")
BOOT_DIR = Path("/boot")
#: A clean environment for apt: nothing from the caller's is passed through.
#: NEEDRESTART_MODE=l lists stale services instead of restarting them — the
#: panel asks for a reboot instead of services bouncing under the operator.
APT_ENV = {
    "PATH": "/usr/sbin:/usr/bin:/sbin:/bin",
    "LC_ALL": "C",
    "DEBIAN_FRONTEND": "noninteractive",
    "NEEDRESTART_MODE": "l",
    "APT_LISTCHANGES_FRONTEND": "none",
}
#: --force-confold keeps configuration files the device has changed. Migrations
#: install mosquitto, sshd and journald settings; a package upgrade that put the
#: distribution's defaults back would silently undo that hardening.
APT_OPTIONS = (
    "-o", "DPkg::Lock::Timeout=300",
    "-o", "Dpkg::Options::=--force-confdef",
    "-o", "Dpkg::Options::=--force-confold",
)
#: Automatic security updates: the switch the panel flips, and the timers that
#: run them. The origin restriction and the no-reboot rule live in
#: 52boneio-unattended, which this helper never writes.
AUTOUPDATE_PERIODIC = Path("/etc/apt/apt.conf.d/52boneio-periodic")
AUTOUPDATE_TIMERS = ("apt-daily.timer", "apt-daily-upgrade.timer")
AUTOUPDATE_LOG = Path("/var/log/unattended-upgrades/unattended-upgrades.log")
AUTOUPDATE_BINARY = Path("/usr/bin/unattended-upgrade")
_PERIODIC_TEMPLATE = (
    "// Installed by boneIO migration 1.6.22; switched from the panel through\n"
    "// boneio-system os-autoupdate-set. Overrides 20auto-upgrades.\n"
    'APT::Periodic::Update-Package-Lists "{value}";\n'
    'APT::Periodic::Unattended-Upgrade "{value}";\n'
)
_PERIODIC_ON_RE = re.compile(r'^APT::Periodic::Unattended-Upgrade\s+"1";', re.MULTILINE)
#: ``2026-09-25 06:31:12,114 INFO Packages that will be upgraded: libssl3t64 openssl``
_UNATTENDED_PACKAGES_RE = re.compile(
    r"^(?P<when>\d{4}-\d\d-\d\d \d\d:\d\d:\d\d),\d+ INFO Packages that will be upgraded: (?P<pkgs>.*)$"
)
_UNATTENDED_RUN_RE = re.compile(
    r"^(?P<when>\d{4}-\d\d-\d\d \d\d:\d\d:\d\d),\d+ INFO Starting unattended upgrades script"
)

#: ``Inst libssl3t64 [3.5.1-1] (3.5.4-1~deb13u1 Debian-Security:13/stable [armhf])``
#: — the bracketed installed version is absent for a package that is new.
_APT_INST_RE = re.compile(
    r"^Inst (?P<name>\S+) (?:\[(?P<old>[^\]]+)\] )?\((?P<new>\S+)"
)

VERBS = (
    "can-up", "can-down", "can-restart", "overlay-get", "overlay-set",
    "hostname-set",
    "ntp-get", "ntp-set",
    "mqtt-password", "mqtt-reload",
    "service-password-state", "service-password-init",
    "os-update-state", "os-update-start", "os-update-run", "os-update-log",
    "os-autoupdate-set",
)


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-system: 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-system")


_LOGGER = _configure_logging()


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


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


def _run(argv: list[str], timeout: int = 30, tolerate: bool = False) -> int:
    """Run a fixed command.

    Args:
        argv: The command.
        timeout: Seconds to allow.
        tolerate: Report a failure but do not treat it as one.

    Returns:
        Process exit status, or 0 when *tolerate* and it failed.
    """
    _LOGGER.info("RUN: %s", " ".join(argv))
    try:
        result = subprocess.run(argv, capture_output=True, text=True, timeout=timeout)
    except subprocess.TimeoutExpired:
        _LOGGER.error("timed out after %ds: %s", timeout, " ".join(argv))
        return 0 if tolerate else 1
    except OSError as exc:
        _LOGGER.error("cannot run %s: %s", argv[0], exc)
        return 0 if tolerate else 1

    if result.stdout:
        sys.stdout.write(result.stdout)
    if result.returncode != 0:
        _LOGGER.log(
            logging.WARNING if tolerate else logging.ERROR,
            "rc=%d: %s", result.returncode, result.stderr.strip(),
        )
        if not tolerate:
            sys.stderr.write(result.stderr)
    return 0 if tolerate else result.returncode


# ------------------------------------------------------------------------ CAN


def _check_interface(name: str | None) -> str:
    """Validate a CAN interface name.

    Args:
        name: Candidate name.

    Returns:
        The name.

    Raises:
        Refused: If it is not one of this board's interfaces.
    """
    if name not in CAN_INTERFACES:
        raise Refused(
            f"not a CAN interface on this board: {name!r}. Allowed: "
            f"{', '.join(CAN_INTERFACES)}"
        )
    return name


def _can_up(interface: str | None, bitrate: str | None) -> int:
    """Bring a CAN interface up at a given bitrate.

    Args:
        interface: ``can0`` or ``can1``.
        bitrate: One of :data:`CAN_BITRATES`, as text.

    Returns:
        Process exit status.

    Raises:
        Refused: If either argument is not in its allowed set.
    """
    interface = _check_interface(interface)
    if bitrate is None or not bitrate.isdigit() or int(bitrate) not in CAN_BITRATES:
        raise Refused(
            f"unsupported bitrate: {bitrate!r}. Allowed: "
            f"{', '.join(str(rate) for rate in CAN_BITRATES)}"
        )

    # Down first, tolerating an interface that was not up.
    _run(["ip", "link", "set", interface, "down"], tolerate=True)
    status = _run(
        ["ip", "link", "set", interface, "type", "can", "bitrate", bitrate]
    )
    if status != 0:
        return status
    return _run(["ip", "link", "set", interface, "up"])


def _can_restart(interface: str | None) -> int:
    """Bring a CAN controller out of bus-off.

    A controller that has counted too many errors stops transmitting until it is
    restarted. The bring-up cycle does that too — down clears the state and up
    starts a fresh controller — but this is the primitive the kernel offers for
    exactly that job, and it keeps the configured bitrate, so a recovery does not
    have to know what the bus is running at.

    Args:
        interface: ``can0``, ``can1`` or ``vcan0``.

    Returns:
        Process exit status.
    """
    return _run(
        ["ip", "link", "set", _check_interface(interface), "type", "can", "restart"]
    )


def _can_down(interface: str | None) -> int:
    """Take a CAN interface down.

    Args:
        interface: ``can0`` or ``can1``.

    Returns:
        Process exit status.
    """
    return _run(["ip", "link", "set", _check_interface(interface), "down"])


# -------------------------------------------------------------------- overlay


def _find_uenv() -> Path:
    """The active uEnv.txt.

    Returns:
        Its path.

    Raises:
        Refused: If there is none.
    """
    for path in UENV_PATHS:
        if path.is_file() and not path.is_symlink():
            return path
    raise Refused(f"no uEnv.txt at any of {[str(p) for p in UENV_PATHS]}")


def _overlay_get() -> int:
    """Print the overlay currently configured in uEnv.txt.

    Returns:
        0, with the overlay name on stdout, or ``null`` when none is set.
    """
    uenv = _find_uenv()
    for line in uenv.read_text(encoding="utf-8").splitlines():
        if line.lstrip().startswith("#"):
            continue
        match = _OVERLAY_LINE_RE.match(line)
        if match:
            print(json.dumps({"overlay": match.group("name"), "uenv": str(uenv)}))
            return 0
    print(json.dumps({"overlay": None, "uenv": str(uenv)}))
    return 0


def _overlay_set(overlay: str | None) -> int:
    """Point uEnv.txt at one of the shipped overlays.

    The application used to build a ``sed -i`` expression and have root run it.
    The edit happens here instead, line by line, against a compiled pattern —
    and commented lines are left alone, as before.

    Args:
        overlay: Overlay basename.

    Returns:
        0 on success.

    Raises:
        Refused: If the overlay is not one that ships with this board.
    """
    if overlay not in VALID_OVERLAYS:
        raise Refused(
            f"unknown overlay: {overlay!r}. Allowed: {', '.join(VALID_OVERLAYS)}"
        )

    uenv = _find_uenv()
    original = uenv.read_text(encoding="utf-8")
    lines = original.splitlines(keepends=True)

    changed = 0
    for index, line in enumerate(lines):
        if line.lstrip().startswith("#"):
            continue
        match = _OVERLAY_LINE_RE.match(line.rstrip("\n"))
        if not match:
            continue
        if match.group("name") == overlay:
            continue
        newline = "\n" if line.endswith("\n") else ""
        lines[index] = (
            f"{match.group('indent')}{match.group('key')}"
            f"{match.group('dir') or ''}{overlay}{match.group('rest')}{newline}"
        )
        changed += 1

    if not changed:
        _LOGGER.info("uEnv.txt already points at %s", overlay)
        return 0

    backup = uenv.with_suffix(uenv.suffix + ".boneio.bak")
    if not backup.exists():
        shutil.copy2(uenv, backup)
        _LOGGER.info("Backed up %s to %s", uenv, backup)

    with tempfile.NamedTemporaryFile(
        dir=uenv.parent, delete=False, suffix=".tmp", mode="w", encoding="utf-8"
    ) as tmp:
        tmp.write("".join(lines))
        tmp_path = tmp.name
    try:
        os.chmod(tmp_path, uenv.stat().st_mode & 0o7777)
        if os.geteuid() == 0:
            os.chown(tmp_path, 0, 0)
        os.replace(tmp_path, uenv)
    except OSError:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        raise
    _LOGGER.info("Overlay set to %s in %s (%d line(s))", overlay, uenv, changed)
    return 0


# ------------------------------------------------------------------- hostname


def _mqtt_password(account: str | None) -> int:
    """Set a broker password for one of the managed accounts.

    The password is read from stdin, never from the argument list. The rule this
    replaces was ``mosquitto_passwd -b /etc/mosquitto/passwd <account> *``, so
    the new password stood in the process table for as long as the command ran —
    visible to every local account, and to anything sampling ``ps``.

    Args:
        account: One of :data:`MQTT_ACCOUNTS`.

    Returns:
        Process exit status.

    Raises:
        Refused: If the account is not one this application manages, or the
            password is empty or contains a newline.
    """
    if account not in MQTT_ACCOUNTS:
        raise Refused(
            f"not an account this device manages: {account!r}. Allowed: "
            f"{', '.join(MQTT_ACCOUNTS)}"
        )
    if not MOSQUITTO_PASSWD.is_file() or MOSQUITTO_PASSWD.is_symlink():
        raise Refused(f"{MOSQUITTO_PASSWD} is missing or is a symlink")

    password = sys.stdin.read()
    # Only the trailing newline a shell adds; anything else would be part of
    # the password, and mosquitto_passwd takes it on one line.
    password = password[:-1] if password.endswith("\n") else password
    if not password:
        raise Refused("no password on stdin")
    if "\n" in password or "\r" in password:
        raise Refused("a password cannot contain a line break")

    # -b takes the password as an argument, which is what we are getting away
    # from; the interactive form reads it from stdin instead.
    try:
        result = subprocess.run(
            ["mosquitto_passwd", str(MOSQUITTO_PASSWD), account],
            input=f"{password}\n{password}\n",
            capture_output=True,
            text=True,
            timeout=30,
        )
    except (OSError, subprocess.SubprocessError) as exc:
        raise Refused(f"cannot run mosquitto_passwd: {exc}") from exc

    if result.returncode != 0:
        # Deliberately not echoing stdout/stderr wholesale: mosquitto_passwd is
        # not expected to print a password back, but this is not the place to
        # find out the hard way.
        _LOGGER.error("mosquitto_passwd failed for %s (rc=%d)", account, result.returncode)
        return result.returncode

    _restore_passwd_permissions()
    _LOGGER.info("Broker password set for %s", account)
    return 0


def _restore_passwd_permissions() -> None:
    """Put the password file back to root:mosquitto 0640 after a write."""
    try:
        import grp
        import pwd

        os.chmod(MOSQUITTO_PASSWD, MOSQUITTO_PASSWD_MODE)
        uid = pwd.getpwnam(MOSQUITTO_PASSWD_OWNER[0]).pw_uid
        gid = grp.getgrnam(MOSQUITTO_PASSWD_OWNER[1]).gr_gid
        os.chown(MOSQUITTO_PASSWD, uid, gid)
    except (KeyError, OSError) as exc:
        _LOGGER.error(
            "Could not restore %s to %s:%s %04o: %s — the hashes may be readable "
            "by other local accounts",
            MOSQUITTO_PASSWD, *MOSQUITTO_PASSWD_OWNER, MOSQUITTO_PASSWD_MODE, exc,
        )


def _mqtt_reload() -> int:
    """Have the broker re-read its password file."""
    return _run(["systemctl", "reload", "mosquitto"])


def _hostname_set(name: str | None) -> int:
    """Set the system hostname.

    Args:
        name: The new hostname.

    Returns:
        Process exit status.

    Raises:
        Refused: If it is not a plausible DNS label.
    """
    if not name or not _HOSTNAME_RE.match(name):
        raise Refused(
            f"not a valid hostname: {name!r}. Lower-case letters, digits and "
            "hyphens, up to 63 characters, not starting or ending with a hyphen."
        )
    return _run(["hostnamectl", "set-hostname", name])


# ------------------------------------------------------------------------ NTP


def _check_ntp_server(value: str) -> str:
    """Validate one NTP server: an IP address or a DNS name.

    Args:
        value: Candidate server.

    Returns:
        The value, unchanged.

    Raises:
        Refused: If it is neither a valid address nor a plausible DNS name.
    """
    try:
        return str(ipaddress.ip_address(value))
    except ValueError:
        pass
    if not _NTP_HOSTNAME_RE.match(value):
        raise Refused(
            f"not a usable NTP server: {value!r}. Give an IPv4 or IPv6 address, "
            "or a DNS name made of letters, digits, hyphens and dots."
        )
    return value


def _parse_ntp_servers(argument: str | None) -> list[str]:
    """Split and validate the comma-separated server list.

    Args:
        argument: ``"a,b,c"``, or ``""``/``"default"`` to mean "no drop-in".

    Returns:
        The validated servers, possibly empty.

    Raises:
        Refused: On an invalid entry, a duplicate, or too many of them.
    """
    if argument is None or argument.strip().lower() in ("", "default"):
        return []

    servers: list[str] = []
    for raw in argument.split(","):
        entry = raw.strip()
        if not entry:
            continue
        checked = _check_ntp_server(entry)
        if checked in servers:
            raise Refused(f"duplicate NTP server: {checked!r}")
        servers.append(checked)

    if len(servers) > NTP_MAX_SERVERS:
        raise Refused(
            f"too many NTP servers: {len(servers)}. At most {NTP_MAX_SERVERS}."
        )
    return servers


def _read_ntp_dropin() -> list[str]:
    """Return the servers currently in the drop-in, empty if there is none."""
    try:
        text = NTP_DROPIN.read_text(encoding="utf-8")
    except OSError:
        return []
    for line in text.splitlines():
        stripped = line.strip()
        if stripped.startswith("NTP="):
            return stripped[4:].split()
    return []


def _ntp_get() -> int:
    """Print the configured NTP servers as JSON.

    Returns:
        0 always; an absent drop-in is a valid state, not a failure.
    """
    print(
        json.dumps(
            {
                "servers": _read_ntp_dropin(),
                "dropin": NTP_DROPIN.exists(),
                "path": str(NTP_DROPIN),
            }
        )
    )
    return 0


def _ntp_set(argument: str | None) -> int:
    """Write (or remove) the timesyncd drop-in naming the NTP servers.

    An empty list removes the drop-in, which returns the device to whatever the
    distribution configured — the only honest meaning of "use the defaults".

    The daemon is restarted with ``try-restart``, not ``restart``: if NTP has
    been switched off, changing which servers *would* be used must not quietly
    switch synchronisation back on.

    Args:
        argument: Comma-separated servers, or ``""``/``"default"``.

    Returns:
        0 on success.

    Raises:
        Refused: If any server is invalid.
    """
    servers = _parse_ntp_servers(argument)

    if not servers:
        if NTP_DROPIN.exists():
            NTP_DROPIN.unlink()
            _LOGGER.info("Removed %s; using the distribution defaults.", NTP_DROPIN)
        else:
            _LOGGER.info("No NTP drop-in to remove; already on the defaults.")
        return _run(["systemctl", "try-restart", "systemd-timesyncd"], tolerate=True)

    content = (
        "# Written by boneio-system. Edits here are replaced on the next change\n"
        "# from the boneIO web interface.\n"
        "[Time]\n"
        f"NTP={' '.join(servers)}\n"
    )

    NTP_DROPIN.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile(
        dir=NTP_DROPIN.parent, 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:
            os.chown(tmp_path, 0, 0)
        os.replace(tmp_path, NTP_DROPIN)
    except OSError:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        raise

    _LOGGER.info("NTP servers set to %s in %s", " ".join(servers), NTP_DROPIN)
    return _run(["systemctl", "try-restart", "systemd-timesyncd"], tolerate=True)


# ---------------------------------------------------------- service password
#
# The boneio account carries a password-gated (ALL:ALL) ALL, so its password is
# the root password. Images before 1.6 shipped it as "Black" on every unit;
# images from 1.6 ship it locked, and the owner's first password — the one they
# give the first-run wizard — becomes it.
#
# Setting it is the one thing here that could undo F-04. This helper runs as
# root, without a password, for anyone holding the boneio account. An operation
# that sets that account's password whenever asked is boneio -> root in two
# steps: set a password, then sudo with it. So it is allowed in exactly the two
# states where it grants nothing new:
#
#   locked   nobody has chosen a password yet — the factory state.
#   shipped  the password is "Black". Whoever holds boneio already has root,
#            through sudo with a password everybody knows.
#   empty    no password at all, which is the same thing without the typing.
#
# and never where the owner has chosen one. The flag makes it once-only even if
# somebody later locks the account again by hand.


def _shadow_hash(account: str = SERVICE_ACCOUNT) -> str | None:
    """The password field of *account* in /etc/shadow, or None if absent.

    Args:
        account: The account to look up.

    Returns:
        The hash field, possibly empty or carrying a lock prefix.

    Raises:
        Refused: If the shadow file cannot be read.
    """
    try:
        lines = SHADOW.read_text(encoding="utf-8").splitlines()
    except OSError as exc:
        raise Refused(f"cannot read {SHADOW}: {exc}") from exc
    for line in lines:
        name, _, rest = line.partition(":")
        if name == account:
            return rest.split(":", 1)[0]
    return None


def _crypt_matches(password: str, stored: str) -> bool | None:
    """Whether *password* verifies against the crypt(3) hash *stored*.

    Through libcrypt directly: Python 3.13 removed the crypt module, and the
    alternatives — su, sudo -S, an expect script — all mean trying the password
    against a live login and leaving a failed attempt in the log for every
    check.

    Returns:
        True or False, or None when libcrypt cannot be used here.
    """
    import ctypes
    import ctypes.util

    name = ctypes.util.find_library("crypt") or "libcrypt.so.1"
    try:
        libcrypt = ctypes.CDLL(name)
    except OSError:
        return None
    libcrypt.crypt.restype = ctypes.c_char_p
    libcrypt.crypt.argtypes = (ctypes.c_char_p, ctypes.c_char_p)
    result = libcrypt.crypt(password.encode(), stored.encode())
    if not result:
        return None
    return result.decode(errors="replace") == stored


def _service_password_state() -> str:
    """Classify the boneio login.

    Returns:
        ``locked``, ``empty``, ``shipped``, ``set`` or ``unknown``.

    Raises:
        Refused: If the account does not exist.
    """
    stored = _shadow_hash()
    if stored is None:
        raise Refused(f"no {SERVICE_ACCOUNT} account in {SHADOW}")
    if stored.startswith(("!", "*")):
        return "locked"
    if stored == "":
        # An empty field is a login with no password at all.
        return "empty"
    matches = _crypt_matches(SHIPPED_PASSWORD, stored)
    if matches is None:
        return "unknown"
    return "shipped" if matches else "set"


def _service_password_state_verb() -> int:
    """Print the state as JSON. Changes nothing.

    Returns:
        Process exit status.
    """
    state = _service_password_state()
    print(json.dumps({"state": state, "initialised": SERVICE_PASSWORD_FLAG.exists()}))
    return 0


def _service_password_init() -> int:
    """Set the boneio password once, from stdin.

    Returns:
        Process exit status.

    Raises:
        Refused: If a password has already been set this way, if the owner has
            chosen one, or if the password on stdin is unusable.
    """
    if SERVICE_PASSWORD_FLAG.exists():
        raise Refused(
            "the service account password has already been set once; change it "
            "with passwd, which asks for the current one"
        )
    state = _service_password_state()
    if state not in ("locked", "shipped", "empty"):
        raise Refused(
            f"the service account password is {state!r}, not one the factory "
            "left; it is the owner's to change, with passwd"
        )

    password = sys.stdin.read()
    password = password[:-1] if password.endswith("\n") else password
    if not password:
        raise Refused("no password on stdin")
    if "\n" in password or "\r" in password:
        raise Refused("a password cannot contain a line break")
    if len(password) > SERVICE_PASSWORD_MAX:
        raise Refused(f"a password longer than {SERVICE_PASSWORD_MAX} characters")

    try:
        result = subprocess.run(
            ["chpasswd"],
            # chpasswd splits on the first colon, so one in the password is
            # fine. Over stdin, never as an argument: ps would show it.
            input=f"{SERVICE_ACCOUNT}:{password}\n",
            capture_output=True,
            text=True,
            timeout=30,
        )
    except (OSError, subprocess.SubprocessError) as exc:
        raise Refused(f"cannot run chpasswd: {exc}") from exc
    if result.returncode != 0:
        _LOGGER.error("chpasswd failed for %s (rc=%d)", SERVICE_ACCOUNT, result.returncode)
        return result.returncode

    # A password set from the expired 1.6-era state would still demand a change
    # at the first login; the owner has just chosen this one, so it is current.
    subprocess.run(
        ["chage", "-d", str(int(time.time()) // 86400), SERVICE_ACCOUNT],
        capture_output=True,
        timeout=30,
    )

    SERVICE_PASSWORD_FLAG.parent.mkdir(parents=True, exist_ok=True)
    SERVICE_PASSWORD_FLAG.write_text(
        f"set from state {state} at {int(time.time())}\n", encoding="utf-8"
    )
    os.chmod(SERVICE_PASSWORD_FLAG, 0o644)
    _LOGGER.info("Service account password set (was %s)", state)
    return 0


# ----------------------------------------------------------------- OS updates
#
# A controller shipped a year ago still runs the kernel, OpenSSL and everything
# else it left the factory with: updating boneIO does not touch the system under
# it. These verbs give the panel `apt-get dist-upgrade`, and the reason they look
# the way they do is F-04. apt run as root with anything the caller chooses is a
# root shell — `-o APT::Update::Pre-Invoke=...`, a `.deb` path, another
# sources.list. So the caller picks a mode from a closed set and nothing else;
# the command lines, options and repositories are all fixed here, and what gets
# installed is whatever the signed Debian and BeagleBoard archives already
# configured on the device publish.


def _os_update_list(text: str) -> list[dict[str, str | None]]:
    """Parse the ``Inst`` lines of ``apt-get -s dist-upgrade``.

    Args:
        text: The simulation's output.

    Returns:
        One entry per package: name, installed version (None when new),
        candidate version.
    """
    packages: list[dict[str, str | None]] = []
    for line in text.splitlines():
        match = _APT_INST_RE.match(line)
        if match:
            packages.append({
                "name": match.group("name"),
                "from": match.group("old"),
                "to": match.group("new"),
            })
    return packages


def _os_update_unit_active() -> bool:
    """Whether an update is running right now.

    Returns:
        True while the transient unit exists and is not finished.
    """
    try:
        result = subprocess.run(
            ["systemctl", "is-active", OS_UPDATE_UNIT],
            capture_output=True, text=True, timeout=15,
        )
    except (OSError, subprocess.SubprocessError):
        return False
    return result.stdout.strip() in ("active", "activating", "reloading")


def _os_update_load() -> dict:
    """The last run's record.

    Returns:
        The stored state, or an empty dict when nothing has run yet.
    """
    try:
        data = json.loads(OS_UPDATE_STATE.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {}
    return data if isinstance(data, dict) else {}


def _os_update_save(state: dict) -> None:
    """Write the run's record atomically.

    Args:
        state: What to store.
    """
    OS_UPDATE_STATE.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile(
        dir=OS_UPDATE_STATE.parent, delete=False, suffix=".tmp", mode="w",
        encoding="utf-8",
    ) as tmp:
        json.dump(state, tmp, indent=1)
        tmp_path = tmp.name
    os.chmod(tmp_path, 0o644)
    os.replace(tmp_path, OS_UPDATE_STATE)


def _configured_overlay() -> tuple[str, str | None] | None:
    """The boneIO overlay uEnv.txt asks U-Boot for, if any.

    Returns:
        ``(name, directory)`` — the directory as written, or None when the line
        uses the bare name U-Boot resolves against /boot/dtbs/$uname_r/.
    """
    try:
        uenv = _find_uenv()
    except Refused:
        return None
    for line in uenv.read_text(encoding="utf-8").splitlines():
        if line.lstrip().startswith("#"):
            continue
        match = _OVERLAY_LINE_RE.match(line)
        if match:
            return match.group("name"), match.group("dir")
    return None


def _on_boot(path: str) -> Path:
    """A path from uEnv.txt, with /boot/ taken relative to BOOT_DIR."""
    return BOOT_DIR / path[len("/boot/"):] if path.startswith("/boot/") else Path(path)


def _overlay_to_bare_name(uenv: Path) -> int:
    """Drop the directory from boneIO overlay lines in uEnv.txt.

    Older images wrote the overlay as a full path into one kernel's dtbs
    directory, e.g. ``/boot/dtbs/6.18.2-bone12/overlays/BONEIO-BLACK-PINS.dtbo``.
    That keeps loading the old kernel's copy after an upgrade, and stops loading
    anything once that kernel is removed. The bare name is what U-Boot resolves
    against /boot/dtbs/$uname_r/, so it follows every kernel.

    Args:
        uenv: The file to edit.

    Returns:
        How many lines were changed.
    """
    original = uenv.read_text(encoding="utf-8")
    lines = original.splitlines(keepends=True)
    changed = 0
    for index, line in enumerate(lines):
        if line.lstrip().startswith("#"):
            continue
        match = _OVERLAY_LINE_RE.match(line.rstrip("\n"))
        if not match or not match.group("dir"):
            continue
        newline = "\n" if line.endswith("\n") else ""
        lines[index] = (
            f"{match.group('indent')}{match.group('key')}"
            f"{match.group('name')}{match.group('rest')}{newline}"
        )
        changed += 1
    if not changed:
        return 0
    backup = uenv.with_suffix(uenv.suffix + ".boneio.bak")
    if not backup.exists():
        shutil.copy2(uenv, backup)
    with tempfile.NamedTemporaryFile(
        dir=uenv.parent, delete=False, suffix=".tmp", mode="w", encoding="utf-8"
    ) as tmp:
        tmp.write("".join(lines))
        tmp_path = tmp.name
    os.chmod(tmp_path, uenv.stat().st_mode & 0o7777)
    os.replace(tmp_path, uenv)
    _LOGGER.warning("uEnv.txt: overlay path replaced by its bare name (%d line(s))", changed)
    return changed


def _boot_kernel() -> str | None:
    """The kernel U-Boot will load: ``uname_r`` in uEnv.txt."""
    for path in UENV_PATHS:
        if not path.is_file():
            continue
        for line in path.read_text(encoding="utf-8").splitlines():
            if line.startswith("uname_r="):
                value = line.split("=", 1)[1].strip()
                if value:
                    return value
    return None


def _kernel_check(repair: bool = False) -> dict[str, str | None]:
    """Will the next boot come up with the boneIO pinmux?

    A kernel upgrade brings a new /boot/dtbs/<version>/ that has never heard of
    the boneIO overlay. U-Boot looks the overlay up there by bare name, and when
    it is missing it boots the stock BeagleBone pinmux without a word: 1-Wire,
    CAN and the buzzer are simply gone, and nothing in userspace says why. The
    kernel postinst hook (zz-boneio-overlay) is meant to copy it across; this
    checks that it did, and with *repair* does the copy itself — only for a file
    from VALID_OVERLAYS, only from another kernel's dtbs directory.

    Args:
        repair: Copy a missing overlay into the boot kernel's directory.

    Returns:
        ``status`` (ok, repaired, problem), ``kernel`` (what will boot),
        ``running`` and ``message``.
    """
    running = os.uname().release
    kernel = _boot_kernel()
    report: dict[str, str | None] = {
        "status": "ok", "kernel": kernel, "running": running, "message": None,
    }

    def _problem(message: str) -> dict[str, str | None]:
        report["status"] = "problem"
        report["message"] = message
        return report

    if not kernel:
        return _problem("uEnv.txt names no kernel (uname_r)")
    for name in (f"vmlinuz-{kernel}", f"initrd.img-{kernel}"):
        if not (BOOT_DIR / name).is_file():
            return _problem(f"/boot/{name} is missing")
    dtbs = BOOT_DIR / "dtbs" / kernel
    if not dtbs.is_dir():
        return _problem(f"/boot/dtbs/{kernel}/ is missing")

    configured = _configured_overlay()
    if configured is None:
        return report
    overlay, directory = configured
    if overlay not in VALID_OVERLAYS:
        return _problem(f"uEnv.txt names an overlay this board does not ship: {overlay}")

    notes: list[str] = []
    if not (dtbs / overlay).is_file():
        if not repair:
            return _problem(f"{overlay} is missing from /boot/dtbs/{kernel}/")
        running_dtbs = BOOT_DIR / "dtbs" / running
        sources = [running_dtbs / overlay, running_dtbs / "overlays" / overlay]
        sources += sorted((BOOT_DIR / "dtbs").glob(f"*/{overlay}"))
        sources += sorted((BOOT_DIR / "dtbs").glob(f"*/overlays/{overlay}"))
        source = next(
            (s for s in sources if s.is_file() and dtbs not in (s.parent, s.parent.parent)),
            None,
        )
        if source is None:
            return _problem(
                f"{overlay} is missing from /boot/dtbs/{kernel}/ and no other kernel has it"
            )
        (dtbs / "overlays").mkdir(exist_ok=True)
        shutil.copy2(source, dtbs / overlay)
        shutil.copy2(source, dtbs / "overlays" / overlay)
        _LOGGER.warning("Copied %s from %s into /boot/dtbs/%s/", overlay, source.parent, kernel)
        notes.append(f"{overlay} was missing for kernel {kernel}; copied from {source.parent}")

    if directory:
        # U-Boot reads exactly this path, not /boot/dtbs/$uname_r/.
        written = f"{directory}{overlay}"
        if repair:
            _overlay_to_bare_name(_find_uenv())
            notes.append(f"uEnv.txt pointed at {written}; now uses the bare name")
        elif not _on_boot(written).is_file():
            return _problem(f"uEnv.txt loads the overlay from {written}, which does not exist")
        else:
            report["message"] = (
                f"uEnv.txt loads the overlay from {written}, not from the boot kernel's directory"
            )

    if notes:
        report["status"] = "repaired"
        report["message"] = "; ".join(notes)
    return report


def _os_update_state_verb() -> int:
    """Print what the panel needs to show. Changes nothing.

    Returns:
        0, with a JSON object on stdout.
    """
    state = _os_update_load()
    kernel = _kernel_check(repair=False)
    reboot_reasons: list[str] = []
    if REBOOT_REQUIRED.exists():
        try:
            pkgs = REBOOT_REQUIRED_PKGS.read_text(encoding="utf-8").split()
        except OSError:
            pkgs = []
        reboot_reasons.append("packages: " + ", ".join(pkgs) if pkgs else "packages")
    if kernel["kernel"] and kernel["kernel"] != kernel["running"]:
        reboot_reasons.append(f"kernel {kernel['running']} -> {kernel['kernel']}")
    if state.get("reboot_recommended") and (state.get("finished") or 0) > _boot_time():
        reboot_reasons.append("core libraries updated")
    usage = shutil.disk_usage("/")
    print(json.dumps({
        "running": _os_update_unit_active(),
        "last": state or None,
        "reboot_required": bool(reboot_reasons),
        "reboot_reasons": reboot_reasons,
        "kernel": kernel,
        "free_mb": usage.free // (1024 * 1024),
        "min_free_mb": OS_UPDATE_MIN_FREE_MB,
        "autoupdate": _autoupdate_state(),
    }))
    return 0


def _boot_time() -> float:
    """When the running system booted, as a Unix timestamp."""
    try:
        for line in Path("/proc/stat").read_text(encoding="utf-8").splitlines():
            if line.startswith("btime "):
                return float(line.split()[1])
    except (OSError, ValueError):
        pass
    return 0.0


def _os_update_start(mode: str | None) -> int:
    """Start a check or an upgrade in its own systemd unit, and return.

    The work does not run here, and it must not: this process is a child of
    boneio.service, and an upgrade that restarts boneIO — through needrestart,
    or because python itself was upgraded — would kill dpkg halfway through.
    A transient unit lives outside that cgroup, and its fixed name means
    systemd itself refuses a second run while one is going.

    Args:
        mode: ``check`` or ``upgrade``.

    Returns:
        0 when the unit was started.

    Raises:
        Refused: For an unknown mode, or while a run is in progress.
    """
    if mode not in OS_UPDATE_MODES:
        raise Refused(f"unknown mode: {mode!r}. Allowed: {', '.join(OS_UPDATE_MODES)}")
    if _os_update_unit_active():
        raise Refused("an operating system update is already running")
    rc = _run([
        "systemd-run", "--unit", OS_UPDATE_UNIT, "--collect", "--no-block", "--quiet",
        "--description", "boneIO operating system update",
        # boneIO runs at CPUWeight=1000; the update must not starve it.
        "--property", "CPUWeight=50", "--property", "IOWeight=50",
        HELPER_SELF, "os-update-run", mode,
    ])
    if rc == 0:
        print(json.dumps({"started": mode}))
    return rc


def _in_update_unit() -> bool:
    """Whether this process runs inside the update's own unit."""
    try:
        cgroup = Path("/proc/self/cgroup").read_text(encoding="utf-8")
    except OSError:
        return False
    return f"/{OS_UPDATE_UNIT}" in cgroup


def _stream(argv: list[str], log, timeout: int) -> tuple[int, str]:
    """Run one apt step, writing its output to *log* as it comes.

    A dist-upgrade on a BeagleBone configures packages for tens of minutes;
    written only at the end, the log the panel shows sits on the step's header
    the whole time and looks hung.

    Args:
        argv: The command, with APT_ENV as its whole environment.
        log: Open log file.
        timeout: Seconds before the step is killed.

    Returns:
        Exit status and the full output.
    """
    process = subprocess.Popen(
        argv, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
        env=APT_ENV, stdin=subprocess.DEVNULL, errors="replace",
    )
    timer = threading.Timer(timeout, process.kill)
    timer.start()
    output: list[str] = []
    try:
        assert process.stdout is not None
        for line in process.stdout:
            output.append(line)
            log.write(line)
            log.flush()
        returncode = process.wait()
    finally:
        timer.cancel()
    if returncode < 0:
        log.write(f"\n=== killed after {timeout}s\n")
    return returncode, "".join(output)


def _os_update_run(mode: str | None) -> int:
    """Do the work. Only inside the unit that os-update-start created.

    Args:
        mode: ``check`` or ``upgrade``.

    Returns:
        0 on success.

    Raises:
        Refused: For an unknown mode, or when called any other way.
    """
    if mode not in OS_UPDATE_MODES:
        raise Refused(f"unknown mode: {mode!r}. Allowed: {', '.join(OS_UPDATE_MODES)}")
    if not _in_update_unit():
        raise Refused(f"os-update-run only runs inside {OS_UPDATE_UNIT}; use os-update-start")

    state: dict = {
        "mode": mode, "started": time.time(), "finished": None,
        "result": "running", "step": None, "message": None,
        "packages": [], "reboot_recommended": False, "kernel": None,
    }
    _os_update_save(state)
    OS_UPDATE_LOG.parent.mkdir(parents=True, exist_ok=True)

    with OS_UPDATE_LOG.open("w", encoding="utf-8") as log:
        os.chmod(OS_UPDATE_LOG, 0o644)

        def step(name: str, argv: list[str], timeout: int) -> str:
            state["step"] = name
            _os_update_save(state)
            log.write(f"\n=== {name}: {' '.join(argv)}\n")
            log.flush()
            returncode, output = _stream(argv, log, timeout)
            if returncode != 0:
                raise Refused(f"{name} failed (rc={returncode})")
            return output

        try:
            if mode == "upgrade":
                free_mb = shutil.disk_usage("/").free // (1024 * 1024)
                if free_mb < OS_UPDATE_MIN_FREE_MB:
                    raise Refused(
                        f"only {free_mb} MB free on /, {OS_UPDATE_MIN_FREE_MB} MB needed"
                    )
                # An upgrade cut short by a power loss leaves dpkg half-done, and
                # apt refuses to do anything until this has run.
                step("dpkg-configure", ["dpkg", "--configure", "-a"], 1800)
            step("update", ["apt-get", *APT_OPTIONS, "update"], 900)
            simulated = step(
                "simulate", ["apt-get", *APT_OPTIONS, "-s", "dist-upgrade"], 600
            )
            state["packages"] = _os_update_list(simulated)
            if mode == "upgrade" and state["packages"]:
                step("dist-upgrade", ["apt-get", *APT_OPTIONS, "-y", "dist-upgrade"], 3 * 3600)
                step("autoremove", ["apt-get", *APT_OPTIONS, "-y", "autoremove", "--purge"], 1800)
                step("clean", ["apt-get", "clean"], 300)
                state["reboot_recommended"] = any(
                    p["name"].startswith(REBOOT_PACKAGE_PREFIXES) for p in state["packages"]
                )
            if mode == "upgrade":
                state["kernel"] = _kernel_check(repair=True)
                log.write(f"\n=== kernel check: {json.dumps(state['kernel'])}\n")
            state["result"] = (
                "problem" if (state["kernel"] or {}).get("status") == "problem" else "success"
            )
        except (Refused, OSError, subprocess.SubprocessError) as exc:
            state["result"] = "failed"
            state["message"] = str(exc)
            log.write(f"\n=== FAILED: {exc}\n")
        finally:
            state["finished"] = time.time()
            state["step"] = None
            _os_update_save(state)

    _LOGGER.info("os-update %s: %s (%d package(s))", mode, state["result"], len(state["packages"]))
    return 0 if state["result"] == "success" else 1


def _autoupdate_state() -> dict:
    """Whether automatic security updates are on, and what they last did.

    Returns:
        ``installed``, ``enabled``, ``last_run`` and ``last_packages``.
    """
    try:
        periodic_on = bool(_PERIODIC_ON_RE.search(AUTOUPDATE_PERIODIC.read_text(encoding="utf-8")))
    except OSError:
        periodic_on = False
    timers_on = True
    for timer in AUTOUPDATE_TIMERS:
        try:
            result = subprocess.run(
                ["systemctl", "is-enabled", timer], capture_output=True, text=True, timeout=15
            )
        except (OSError, subprocess.SubprocessError):
            timers_on = False
            break
        if result.stdout.strip() != "enabled":
            timers_on = False
    last_run = last_packages_at = None
    last_packages: list[str] = []
    try:
        lines = AUTOUPDATE_LOG.read_text(encoding="utf-8", errors="replace").splitlines()
    except OSError:
        lines = []
    for line in lines:
        run = _UNATTENDED_RUN_RE.match(line)
        if run:
            last_run = run.group("when")
        pkgs = _UNATTENDED_PACKAGES_RE.match(line)
        if pkgs:
            last_packages_at = pkgs.group("when")
            last_packages = pkgs.group("pkgs").split()
    return {
        "installed": AUTOUPDATE_BINARY.exists(),
        "configured": AUTOUPDATE_PERIODIC.exists(),
        "enabled": periodic_on and timers_on,
        "last_run": last_run,
        "last_packages_at": last_packages_at,
        "last_packages": last_packages,
    }


def _os_autoupdate_set(argument: str | None) -> int:
    """Switch automatic security updates on or off.

    Only the switch: which archive they come from (Debian-Security alone) and
    that they never reboot are in 52boneio-unattended, installed by a signed
    migration and not writable from here. The file written is a fixed template
    with one of two values, so nothing the caller sends reaches apt's config.

    Args:
        argument: ``on`` or ``off``.

    Returns:
        0 on success.

    Raises:
        Refused: For anything else, or before migration 1.6.22 has run.
    """
    if argument not in ("on", "off"):
        raise Refused(f"expected on or off, not {argument!r}")
    if not AUTOUPDATE_PERIODIC.exists():
        raise Refused(
            f"{AUTOUPDATE_PERIODIC} is missing; apply boneIO migration 1.6.22 first"
        )
    value = "1" if argument == "on" else "0"
    with tempfile.NamedTemporaryFile(
        dir=AUTOUPDATE_PERIODIC.parent, delete=False, suffix=".tmp", mode="w",
        encoding="utf-8",
    ) as tmp:
        tmp.write(_PERIODIC_TEMPLATE.format(value=value))
        tmp_path = tmp.name
    os.chmod(tmp_path, 0o644)
    os.replace(tmp_path, AUTOUPDATE_PERIODIC)
    verb = "enable" if argument == "on" else "disable"
    rc = 0
    for timer in AUTOUPDATE_TIMERS:
        rc |= _run(["systemctl", verb, "--now", timer])
    _LOGGER.info("Automatic security updates switched %s", argument)
    if rc == 0:
        print(json.dumps(_autoupdate_state()))
    return rc


def _os_update_log() -> int:
    """Print the end of the last run's log.

    Returns:
        0, with the log text on stdout (empty when nothing has run).
    """
    try:
        lines = OS_UPDATE_LOG.read_text(encoding="utf-8", errors="replace").splitlines()
    except OSError:
        lines = []
    print("\n".join(lines[-OS_UPDATE_LOG_TAIL:]))
    return 0


# ----------------------------------------------------------------- selftest


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

    Returns:
        0 when healthy.
    """
    problems: list[str] = []
    if shutil.which("ip") is None:
        problems.append("ip is not installed, so CAN cannot be configured")
    if shutil.which("hostnamectl") is None:
        problems.append("hostnamectl is not installed")
    if shutil.which("systemctl") is None:
        problems.append("systemctl is not installed, so NTP servers cannot be applied")
    if shutil.which("chpasswd") is None:
        problems.append("chpasswd is not installed, so the service password cannot be set")
    try:
        _find_uenv()
    except Refused as exc:
        problems.append(str(exc))

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


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

    Returns:
        Process exit status.
    """
    parser = argparse.ArgumentParser(description="boneIO privileged system operations")
    parser.add_argument("verb", nargs="?", help=f"one of: {', '.join(VERBS)}")
    parser.add_argument("argument", nargs="?", default=None)
    parser.add_argument("value", nargs="?", default=None)
    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 args.list_verbs:
        print(json.dumps(list(VERBS)))
        return 0

    _assert_root()

    if args.selftest:
        return selftest()

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

    try:
        if unparsed:
            # Argparse could not place these, and they are still caller input.
            if args.argument is None and len(unparsed) == 1:
                args.argument = unparsed[0]
            elif args.value is None and len(unparsed) == 1:
                args.value = unparsed[0]
            else:
                raise Refused(f"unexpected arguments: {unparsed}")

        if args.verb == "can-up":
            return _can_up(args.argument, args.value)
        if args.verb == "can-restart":
            return _can_restart(args.argument)
        if args.verb == "can-down":
            return _can_down(args.argument)
        if args.verb == "overlay-get":
            return _overlay_get()
        if args.verb == "overlay-set":
            return _overlay_set(args.argument)
        if args.verb == "mqtt-password":
            return _mqtt_password(args.argument)
        if args.verb == "mqtt-reload":
            return _mqtt_reload()
        if args.verb == "service-password-state":
            return _service_password_state_verb()
        if args.verb == "service-password-init":
            return _service_password_init()
        if args.verb == "hostname-set":
            return _hostname_set(args.argument)
        if args.verb == "ntp-get":
            return _ntp_get()
        if args.verb == "ntp-set":
            return _ntp_set(args.argument)
        if args.verb == "os-update-state":
            return _os_update_state_verb()
        if args.verb == "os-update-start":
            return _os_update_start(args.argument)
        if args.verb == "os-update-run":
            return _os_update_run(args.argument)
        if args.verb == "os-update-log":
            return _os_update_log()
        if args.verb == "os-autoupdate-set":
            return _os_autoupdate_set(args.argument)
        raise Refused(f"unknown verb {args.verb!r}. Allowed: {', '.join(VERBS)}")
    except Refused as exc:
        _LOGGER.error("REFUSED: %s", exc)
        return 1


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