path: root/remote/sbb-remote

#!/usr/bin/env python3
# sbb-remote - sb-builder helper on the build VM
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""Helper run over ssh by the sb-builder GUI.

Every command prints one JSON object on stdout, except job-follow, which
streams raw log bytes. gpg-unlock reads the passphrase from stdin. Jobs run detached (setsid, stdin=/dev/null), one at a
time under a flock, from a fixed table of kinds: the GUI sends a kind and
its arguments, this file validates them and builds the argv itself.
"""

import fcntl
import glob
import hashlib
import json
import os
import re
import shlex
import shutil
import signal
import sqlite3
import subprocess
import sys
import tempfile
import time
import tomllib
import urllib.request

VERSION = "0.6.0"
DEFAULTS = {
    "state_dir": "/var/lib/sb-builder",
    "repo_dir": "/var/lib/sbopkg/SBo-danix",
    "upstream_branch": "current",
    "slackrepo_db": "/var/lib/slackrepo/SBo-danix/database_SBo-danix_x86_64.sqlite3",
    "sbpull_script": "/usr/local/share/slackrepo_setup/sbpull.bash",
    "overlays": ["personal", "pentesting"],
    "slackrepo_repo": "SBo-danix",
    "hintdir": "/etc/slackrepo/SBo-danix/hintfiles",
    "default_hintdir": "",
    "genreprc": "/root/.genreprc",
    "slackrepo_logdir": "/var/log/slackrepo/SBo-danix",
    "manifest_packages": ["neovim", "openvino"],  # bundled deps: left to mkhint until phase 4b
}
FLAGS = ("--yes", "--no")
JOB_ID = re.compile(r"^\d{8}-\d{6}-[a-z]+(-[a-z]+)*(-\d+)?$")
PKG = re.compile(r"^[^/]+/([^/]+)/\1\.SlackBuild$")
PKGNAME = re.compile(r"^[A-Za-z0-9._+-]+$")
VERSION_ARG = re.compile(r"^[A-Za-z0-9._+-]{1,64}$")
ACTIONS = ("build", "rebuild", "update", "remove")
PRESET = "/usr/libexec/gpg-preset-passphrase"
ITEM = re.compile(r"^[A-Za-z0-9._+-]+/[A-Za-z0-9._+-]+$")  # category/pkg
LOG_TAIL = 512 * 1024  # bytes of the log returned; the failure is at the end
COMMAND_LOGS = ("build.log", "rebuild.log", "update.log", "remove.log")  # one per slackrepo command
ACTIVE = ("starting", "running")
START_GRACE = 10  # seconds a job may sit in "starting" before it counts as lost


class Error(Exception):
    pass


def load_config(path=None):
    path = path or os.environ.get("SBB_CONFIG", "/etc/sb-builder/config.toml")
    cfg = dict(DEFAULTS)
    try:
        with open(path, "rb") as f:
            data = tomllib.load(f)
    except FileNotFoundError:
        return cfg
    except tomllib.TOMLDecodeError as e:
        raise Error(f"config: {e}")
    for k, v in data.items():
        if k not in DEFAULTS:
            raise Error(f"config: {k}: unknown key")
        if type(v) is not type(DEFAULTS[k]):
            raise Error(f"config: {k}: expected {type(DEFAULTS[k]).__name__}")
        cfg[k] = v
    for k in ("overlays", "manifest_packages"):
        if not all(isinstance(o, str) for o in cfg[k]):
            raise Error(f"config: {k}: expected list of strings")
    return cfg


# ── job kinds: (cfg, args) -> (argv, env), raising Error on bad args ────────

def _flag(args):
    if len(args) != 1 or args[0] not in FLAGS:
        raise Error("expected exactly one of --yes, --no")
    return args[0]


def kind_sbpull(cfg, args):
    if not args or args[0] not in cfg["overlays"]:
        raise Error(f"sbpull: overlay must be one of {cfg['overlays']}")
    return ["bash", "-c", '. "$1" && sbpull "$2" "$3"', "_",
            cfg["sbpull_script"], args[0], _flag(args[1:])], {}


def kind_sbshadow(cfg, args):
    return ["sbshadow", _flag(args)], {}


def kind_slackrepo(cfg, args):
    if not args or args[0] not in ACTIONS:
        raise Error(f"slackrepo: action must be one of {', '.join(ACTIONS)}")
    action, rest = args[0], args[1:]
    opts = [a for a in rest if a.startswith("--")]
    for o in opts:
        if o not in ("--preview", "--no-publish"):
            raise Error(f"slackrepo: unknown option {o}")
    known = {os.path.basename(d) for d in _pkgdirs(cfg["repo_dir"], "*/*")}
    pkgs = []
    for p in (a for a in rest if not a.startswith("--")):
        if not PKGNAME.match(p) or p not in known:
            raise Error(f"slackrepo: no such package: {p}")
        if p not in pkgs:
            pkgs.append(p)
    if not pkgs and action != "update":
        raise Error(f"slackrepo: {action} needs at least one package")
    argv = ["slackrepo", f"--repo={cfg['slackrepo_repo']}", action]
    if "--preview" in opts:
        argv.append("--preview")
    return argv + pkgs, {"SBB_PUBLISH": "n"} if "--no-publish" in opts else {}


def kind_publish(cfg, args):
    if args:
        raise Error("publish takes no arguments")
    return ["sb-publish"], {}


def _pkgmap(cfg):
    """{package: category/package} of the repo."""
    return {os.path.basename(d): os.path.relpath(d, cfg["repo_dir"])
            for d in _pkgdirs(cfg["repo_dir"], "*/*")}


def _hint_target(cfg, pkg, pkgs=None):
    """(category/pkg, <hintdir>/<pkg>.hint); refuses a hint slackrepo finds elsewhere first."""
    pkgs = pkgs or _pkgmap(cfg)
    if not PKGNAME.match(pkg or "") or pkg not in pkgs:
        raise Error(f"no such package: {pkg}")
    target = os.path.join(cfg["hintdir"], pkg + ".hint")
    found = find_hint(cfg, pkgs[pkg])
    if found and found != target:
        raise Error(f"hint lives at {found}, not managed")
    return pkgs[pkg], target


def hint_args(cfg, kind, args):
    """Validated ([(pkg, version or None)], no_dl) of a hint-update or hint-new job."""
    no_dl = "--no-dl" in args
    for o in (a for a in args if a.startswith("--") and a != "--no-dl"):
        raise Error(f"{kind}: unknown option {o}")
    rest = [a for a in args if a != "--no-dl"]
    if kind == "hint-new":
        if not rest:
            raise Error("usage: hint-new <pkg>... [--no-dl]")
        pairs = [(pkg, None) for pkg in rest]  # the .info pin; Update sets a version later
    else:
        if not rest:
            raise Error("hint-update needs at least one pkg=version")
        if not all("=" in a for a in rest):
            raise Error("hint-update: expected pkg=version")
        pairs = [tuple(a.split("=", 1)) for a in rest]
    pkgs = _pkgmap(cfg)
    seen = set()
    for pkg, ver in pairs:
        if ver is not None and not VERSION_ARG.match(ver):
            raise Error(f"bad version: {ver}")
        _, target = _hint_target(cfg, pkg, pkgs)
        if pkg in seen:
            raise Error(f"{pkg} named twice")
        seen.add(pkg)
        if ver is not None and pkg in cfg["manifest_packages"]:
            raise Error(f"{pkg}: bundled deps: use mkhint until phase 4b")
        if kind == "hint-new" and os.path.exists(target):
            raise Error(f"{pkg}: hint exists")
    return pairs, no_dl


def _self_job(kind):
    """A job kind run by this file itself (_hint), argv-free like the others."""
    def build(cfg, args):
        hint_args(cfg, kind, args)
        return [sys.executable, os.path.abspath(__file__), "_hint", kind, *args], {}
    return build


KINDS = {"sbpull": kind_sbpull, "sbshadow": kind_sbshadow,
         "slackrepo": kind_slackrepo, "publish": kind_publish,
         **{k: _self_job(k) for k in ("hint-update", "hint-new")}}


def signs(kind, args):
    """A real slackrepo run ends in gen_repos_files.sh, which signs."""
    return kind == "slackrepo" and "--preview" not in args


# ── job state ───────────────────────────────────────────────────────────────

def job_dir(cfg, jid):
    if not JOB_ID.match(jid or ""):
        raise Error(f"bad job id: {jid}")
    d = os.path.join(cfg["state_dir"], "jobs", jid)
    if not os.path.isdir(d):
        raise Error(f"no such job: {jid}")
    return d


def read_meta(d):
    with open(os.path.join(d, "meta.json")) as f:
        return json.load(f)


def write_meta(d, meta):
    tmp = os.path.join(d, "meta.json.tmp")
    with open(tmp, "w") as f:
        json.dump(meta, f)
    os.replace(tmp, os.path.join(d, "meta.json"))


def lock_file(cfg):
    return open(os.path.join(cfg["state_dir"], "lock"), "a")


def refresh(cfg, d, meta):
    """Mark an active job lost when no runner holds the lock.

    _run writes the final status before it releases the lock, so a free lock
    plus a status that is still active (re-read under the lock) means the
    runner died: killed, crashed, or the VM rebooted.
    """
    if meta["status"] not in ACTIVE:
        return meta
    if meta["status"] == "starting" and time.time() - meta["created"] < START_GRACE:
        return meta
    with lock_file(cfg) as lk:
        try:
            fcntl.flock(lk, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            return meta
        meta = read_meta(d)
        if meta["status"] in ACTIVE:
            meta["status"] = "lost"
            write_meta(d, meta)
    return meta


def all_jobs(cfg):
    out = []
    for d in sorted(glob.glob(os.path.join(cfg["state_dir"], "jobs", "*")), reverse=True):
        try:
            out.append(refresh(cfg, d, read_meta(d)))
        except (OSError, ValueError):
            pass  # a dir being created or removed right now
    return out


# ── commands ────────────────────────────────────────────────────────────────

def cmd_version(cfg):
    return {"version": VERSION, "overlays": cfg["overlays"]}


def cmd_jobs(cfg):
    # ponytail: logs and job dirs are kept forever; prune once disk usage shows on the dashboard
    return {"jobs": all_jobs(cfg)}


def cmd_job_start(cfg, kind=None, *args):
    if kind not in KINDS:
        raise Error(f"unknown kind: {kind}")
    argv, env = KINDS[kind](cfg, list(args))
    if signs(kind, args) and gpg_unlocked(cfg) is False:
        raise Error("signing key locked")
    base = time.strftime("%Y%m%d-%H%M%S-") + kind
    jid, n = base, 1
    while True:
        d = os.path.join(cfg["state_dir"], "jobs", jid)
        try:
            os.mkdir(d)
            break
        except FileExistsError:
            n += 1
            jid = f"{base}-{n}"
    write_meta(d, {"id": jid, "kind": kind, "args": list(args), "argv": argv,
                   "env": env, "status": "starting", "created": time.time()})
    open(os.path.join(d, "log"), "wb").close()
    subprocess.Popen([sys.executable, os.path.abspath(__file__), "_run", jid],
                     stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
                     stderr=subprocess.DEVNULL, start_new_session=True)
    for _ in range(60):
        status = read_meta(d)["status"]
        if status == "busy":
            shutil.rmtree(d)
            return {"busy": next((m["id"] for m in all_jobs(cfg) if m["status"] == "running"), None)}
        if status != "starting":
            break
        time.sleep(0.05)
    return {"id": jid}


def cmd_run(cfg, jid=None):
    d = job_dir(cfg, jid)
    meta = read_meta(d)
    with lock_file(cfg) as lk:
        try:
            fcntl.flock(lk, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            meta["status"] = "busy"
            write_meta(d, meta)
            return None
        with open(os.path.join(d, "log"), "ab") as log:
            meta["start"] = time.time()
            try:
                p = subprocess.Popen(meta["argv"], stdin=subprocess.DEVNULL, stdout=log,
                                     stderr=subprocess.STDOUT, start_new_session=True,
                                     env={**os.environ, **meta.get("env", {})})
            except OSError as e:
                log.write(f"sbb-remote: {e}\n".encode())
                rc = 127
            else:
                meta.update(pid=p.pid, status="running")
                write_meta(d, meta)
                rc = p.wait()
        meta.update(end=time.time(), exit=rc,
                    status="ok" if rc == 0 else "killed" if rc == -signal.SIGTERM else "failed")
        write_meta(d, meta)
    return None


def cmd_job_follow(cfg, jid=None, offset="0", out=None):
    d = job_dir(cfg, jid)
    out = out or sys.stdout.buffer
    with open(os.path.join(d, "log"), "rb") as f:
        f.seek(int(offset))
        while True:
            # status first, then read: the last read after the job ends drains the log
            done = refresh(cfg, d, read_meta(d))["status"] not in ACTIVE
            chunk = f.read()
            if chunk:
                out.write(chunk)
                out.flush()
            elif done:
                return None
            else:
                time.sleep(0.5)


def cmd_job_kill(cfg, jid=None):
    meta = read_meta(job_dir(cfg, jid))
    if meta["status"] != "running":
        raise Error(f"job not running: {jid}")
    try:
        os.killpg(meta["pid"], signal.SIGTERM)
    except ProcessLookupError:
        pass  # ended on its own meanwhile
    return {"ok": True}


# ── stats ───────────────────────────────────────────────────────────────────

def _try(fn):
    try:
        return fn()
    except Exception:
        return None


def _git(cfg, *args):
    return subprocess.run(["git", "-C", cfg["repo_dir"], *args], stdin=subprocess.DEVNULL,
                          capture_output=True, text=True, check=True).stdout


def _pkgdirs(root, pattern):
    """Paths of the <pkg>/<pkg>.SlackBuild dirs matching pattern under root."""
    out = []
    for d in glob.glob(os.path.join(root, pattern, "")):
        d = os.path.dirname(d)
        if os.path.isfile(os.path.join(d, os.path.basename(d) + ".SlackBuild")):
            out.append(d)
    return out


def _pkgs(root, pattern):
    return [os.path.basename(d) for d in _pkgdirs(root, pattern)]


def _packages(cfg):
    r = cfg["repo_dir"]
    mine = {o: _pkgs(r, o + "/*") for o in cfg["overlays"]}
    tree = _git(cfg, "ls-tree", "-r", "--name-only", cfg["upstream_branch"]).splitlines()
    upstream = {m.group(1) for m in map(PKG.match, tree) if m}
    shadowed = upstream & {p for names in mine.values() for p in names}
    return {"total": len(_pkgs(r, "*/*")), **{o: len(n) for o, n in mine.items()},
            "shadowed": len(shadowed)}


def _slackrepo(cfg):
    db = sqlite3.connect(f"file:{cfg['slackrepo_db']}?mode=ro", uri=True)
    try:
        # legacy rows keyed by package file name, not category/pkg, match nothing: skip
        results = dict(db.execute("select result, count(*) from buildresults"
                                  " where itemid like '%/%' group by result"))
        built = db.execute("select count(distinct itemid) from revisions where dep = '/'").fetchone()[0]
    finally:
        db.close()
    return {"results": results, "built": built}


def _uptime():
    with open("/proc/uptime") as f:
        return float(f.read().split()[0])


def _cpu_times(path="/proc/stat"):
    """(busy, total) jiffies over all CPUs; idle and iowait are not busy."""
    with open(path) as f:
        fields = [int(x) for x in f.readline().split()[1:]]
    total = sum(fields[:8])  # guest time is already in user/nice
    return total - fields[3] - fields[4], total


def _cpu():
    """Busy percent over a short sample."""
    b1, t1 = _cpu_times()
    time.sleep(0.25)
    b2, t2 = _cpu_times()
    return round(100 * (b2 - b1) / max(t2 - t1, 1), 1)


def _meminfo(path="/proc/meminfo"):
    with open(path) as f:
        kb = {k: int(v.split()[0]) for k, v in (line.split(":", 1) for line in f)}
    return {"total": kb["MemTotal"] * 1024, "available": kb["MemAvailable"] * 1024}


def _slackpkg(lastupdate="/var/lib/slackpkg/LASTUPDATE", pkgdir="/var/log/packages"):
    """Last 'slackpkg update' (its LASTUPDATE stamp) and last package change."""
    def update():
        with open(lastupdate) as f:
            return int(f.read().split()[0])

    def upgrade():
        # ponytail: newest package entry; slackrepo INSTALL=n, so this is slackpkg's
        with os.scandir(pkgdir) as it:
            return int(max(e.stat().st_mtime for e in it))
    return {"update": _try(update), "upgrade": _try(upgrade)}


def cmd_stats(cfg):
    root = _try(lambda: shutil.disk_usage("/"))
    return {
        "load": _try(os.getloadavg),
        "uptime": _try(_uptime),
        "cpu": _try(_cpu),
        "mem": _try(_meminfo),
        "disk": {"total": root and root.total, "free": root and root.free},
        "slackpkg": _slackpkg(),
        "packages": _try(lambda: _packages(cfg)),
        "sync": {o: _try(lambda o=o: int(_git(cfg, "log", "-1", "--format=%ct", "--", o + "/")))
                 for o in cfg["overlays"]},
        "slackrepo": _try(lambda: _slackrepo(cfg)),
    }


# ── inventory ───────────────────────────────────────────────────────────────

def parse_assignments(path):
    """KEY=value assignments of a shell-style file, without running it.

    Values with a $ are returned as {"value": raw, "shell": True}: the shell
    would expand them, this parser does not.
    """
    with open(path, errors="replace") as f:
        text = f.read().replace("\\\n", " ")
    out = {}
    for line in text.splitlines():
        try:
            tokens = shlex.split(line, comments=True)
        except ValueError:
            continue  # unbalanced quotes: slackrepo would choke too, skip the line
        raw = line.split("=", 1)[1] if "=" in line else ""
        for t in tokens:
            name, eq, value = t.partition("=")
            if eq and re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", name):
                out[name] = {"value": raw.strip(), "shell": True} if "$" in raw else value
    return out


def _value(v):
    return v["value"] if isinstance(v, dict) else v


def find_hint(cfg, path):
    """slackrepo's search order (parsefunctions.sh): first <pkg>.hint found."""
    pkg = os.path.basename(path)
    dirs = [os.path.join(cfg["repo_dir"], path), cfg["hintdir"],
            os.path.join(cfg["hintdir"], path)]
    if cfg["default_hintdir"]:
        dirs += [cfg["default_hintdir"], os.path.join(cfg["default_hintdir"], path)]
    for d in dirs:
        f = os.path.join(d, pkg + ".hint")
        if os.path.isfile(f):
            return f
    return None


def _summary(pkgdir, pkg):
    try:
        with open(os.path.join(pkgdir, "slack-desc"), errors="replace") as f:
            for line in f:
                m = re.match(rf"^{re.escape(pkg)}: {re.escape(pkg)} \((.*)\)\s*$", line)
                if m:
                    return m.group(1)
    except OSError:
        pass
    return None


def _builds(cfg):
    """{path: (built_version, built_time)}, {path: result} from the slackrepo DB."""
    try:
        db = sqlite3.connect(f"file:{cfg['slackrepo_db']}?mode=ro", uri=True)
        try:
            built = {i: (v, int(t)) for i, v, t in
                     db.execute("select itemid, version, built from revisions where dep = '/'")}
            results = dict(db.execute("select itemid, result from buildresults"))
        finally:
            db.close()
    except sqlite3.Error:
        return {}, {}
    return built, results


def cmd_results(cfg):
    """Last build attempt per package, newest first."""
    db = sqlite3.connect(f"file:{cfg['slackrepo_db']}?mode=ro", uri=True)
    try:
        rows = db.execute("""
            select b.itemid, b.result, cast(b.time as integer), r.version
            from buildresults b left join revisions r on r.itemid = b.itemid and r.dep = '/'
            where b.itemid like '%/%' order by cast(b.time as integer) desc""").fetchall()
    finally:
        db.close()
    logdir = cfg["slackrepo_logdir"]
    return {"rows": [[i, res, t, v, os.path.isdir(os.path.join(logdir, i))]
                     for i, res, t, v in rows]}


def _tail(path, limit):
    with open(path, "rb") as f:
        f.seek(0, os.SEEK_END)
        size = f.tell()
        f.seek(max(0, size - limit))
        return f.read().decode(errors="replace"), size > limit


def cmd_build_log(cfg, item=None):
    if not ITEM.match(item or "") or ".." in item:
        raise Error(f"bad package path: {item}")
    d = os.path.join(cfg["slackrepo_logdir"], item)
    # the last attempt's log is the newest command log, not necessarily build.log
    logs = [os.path.join(d, n) for n in COMMAND_LOGS if os.path.isfile(os.path.join(d, n))]
    if not logs:
        raise Error(f"no build log for {item}")
    build = max(logs, key=os.path.getmtime)
    text, truncated = _tail(build, LOG_TAIL)
    resource = os.path.join(d, "resource.log")
    return {"name": os.path.basename(build), "build": text, "truncated": truncated,
            "mtime": int(os.path.getmtime(build)),
            "resource": _tail(resource, LOG_TAIL)[0] if os.path.isfile(resource) else None}


COLUMNS = ["path", "version", "info_version", "hint_version", "built_version", "built_time",
           "result", "maintainer", "requires", "summary", "hint_path", "hints"]


def cmd_inventory(cfg):
    built, results = _builds(cfg)
    rows = []
    for d in sorted(_pkgdirs(cfg["repo_dir"], "*/*")):
        path = os.path.relpath(d, cfg["repo_dir"])
        pkg = os.path.basename(d)
        try:
            info = parse_assignments(os.path.join(d, pkg + ".info"))
        except OSError:
            info = {}
        hint_path = find_hint(cfg, path)
        hints = parse_assignments(hint_path) if hint_path else None
        info_version = _value(info.get("VERSION"))
        hint_version = _value(hints.get("VERSION")) if hints else None
        bv, bt = built.get(path, (None, None))
        rows.append([path, hint_version or info_version, info_version, hint_version, bv, bt,
                     results.get(path), _value(info.get("MAINTAINER")),
                     (_value(info.get("REQUIRES")) or "").split(), _summary(d, pkg),
                     hint_path, hints])
    return {"columns": COLUMNS, "rows": rows}


# ── hint text ───────────────────────────────────────────────────────────────
# Hints are edited as text so comments, order and other keys stay as written.

PIN_KEYS = ("VERSION", "DOWNLOAD", "MD5SUM", "DOWNLOAD_x86_64", "MD5SUM_x86_64")
DL_PAIRS = (("DOWNLOAD", "MD5SUM"), ("DOWNLOAD_x86_64", "MD5SUM_x86_64"))
SKIP_URLS = ("UNSUPPORTED", "UNTESTED")
DL_TIMEOUT = 60


def _assignment(key):
    """KEY=value at line start; a quoted value may span lines. key is a regex."""
    return re.compile(rf'^(?:{key})=(?:"([^"]*)"|(\S*))[^\n]*\n?', re.M)


def hint_tokens(text, key):
    """Whitespace-separated words of KEY's value, [] when unset or empty."""
    m = _assignment(key).search(text)
    if not m:
        return []
    value = m.group(1) if m.group(1) is not None else m.group(2)
    return value.replace("\\\n", " ").split()


def hint_set(text, key, value):
    """KEY="value" in place of the first KEY line, else appended."""
    line = f'{key}="{value}"\n'
    m = _assignment(key).search(text)
    if m:
        return text[:m.start()] + line + text[m.end():]
    return text + ("\n" if text and not text.endswith("\n") else "") + line


def hint_drop(text, key):
    return _assignment(key).sub("", text)


def swap_version(text, old, new):
    """mkhint's bump of old to new, everywhere but in MD5SUM* values.

    Also the '-' form when either has '_' (dated URLs), and the dotless form of
    a dotted old version inside URLs, as a whole number (build_4200 for 4.2.00).
    """
    pairs = [(old, new)]
    if "_" in old or "_" in new:
        pairs.append((old.replace("_", "-"), new.replace("_", "-")))
    o, n = old.replace(".", ""), new.replace(".", "")
    dotless = "." in old and o != n
    out, pos = [], 0
    for m in [*_assignment(r"MD5SUM\w*").finditer(text), None]:
        part = text[pos:m.start() if m else len(text)]
        for a, b in pairs:
            part = part.replace(a, b)
        if dotless:
            part = re.sub(rf'(://[^\s"]*?)(?<!\d){re.escape(o)}(?!\d)',
                          lambda x: x.group(1) + n, part)
        out.append(part)
        if m:
            out.append(m.group(0))
            pos = m.end()
    return "".join(out)


def _pin(info):
    """The version pin of an .info: its VERSION, DOWNLOAD* and MD5SUM* lines."""
    lines = [m.group(0) for k in PIN_KEYS if (m := _assignment(k).search(info))]
    if not lines or not lines[0].startswith("VERSION="):
        raise Error("no VERSION in the .info")
    return "".join(ln.rstrip("\n") + "\n" for ln in lines)


def new_hint(info):
    """mkhint's hint from an .info: the pin, REQUIRES commented, ARCH.

    No phantom deps: a package that has them already has a hint, the
    DELREQUIRES one sbshadow's mkhint -F writes after every sync.
    """
    text = _pin(info)
    m = _assignment("REQUIRES").search(info)
    if m:
        text += "#" + m.group(0).rstrip("\n") + "\n"
    return text + 'ARCH="x86_64"\n'


def _md5_url(url):
    req = urllib.request.Request(url, headers={"User-Agent": "sbb-remote"})
    h = hashlib.md5()
    with urllib.request.urlopen(req, timeout=DL_TIMEOUT) as r:
        for chunk in iter(lambda: r.read(1 << 16), b""):
            h.update(chunk)
    return h.hexdigest()


def apply_version(text, info, new, no_dl, log):
    """The hint text at version new (None: keep the version), log(str) per download.

    An unpinned hint (DELREQUIRES only) gets the .info pin first. Downloads the
    first URL of each DOWNLOAD pair for its md5; continuation URLs keep theirs,
    as in mkhint. no_dl: no download, DOWNLOAD="no" in place of every
    DOWNLOAD*/MD5SUM* line, since DOWNLOAD_$ARCH would override it.
    """
    if not hint_tokens(text, "VERSION"):
        text = _pin(info) + ("" if hint_tokens(text, "ARCH") else 'ARCH="x86_64"\n') + text
    if new is not None:
        text = swap_version(text, hint_tokens(text, "VERSION")[0], new)
    if no_dl:
        return hint_drop(text, r"(?:DOWNLOAD|MD5SUM)\w*") + 'DOWNLOAD="no"\n'
    if new is None:
        return text
    for dl, sums in DL_PAIRS:
        urls = hint_tokens(text, dl)
        if not urls or urls[0] in SKIP_URLS:
            continue
        log(f"downloading {urls[0]}")
        text = hint_set(text, sums, " \\\n    ".join([_md5_url(urls[0]), *hint_tokens(text, sums)[1:]]))
    return text


# ── hint jobs and commands ──────────────────────────────────────────────────

def _read(path):
    try:
        with open(path, errors="replace") as f:
            return f.read()
    except FileNotFoundError:
        return None


def _write(path, text):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    tmp = path + ".tmp"
    with open(tmp, "w") as f:
        f.write(text)
    os.replace(tmp, path)


def _save(target, text):
    """Write a hint, its previous text to .bak; a new hint drops a stale .bak (revert = delete)."""
    if os.path.exists(target):
        shutil.copy2(target, target + ".bak")
    elif os.path.exists(target + ".bak"):
        os.remove(target + ".bak")
    _write(target, text)


def cmd_hint(cfg, kind=None, *args):
    """Body of the hint jobs: log lines, then SBB-RESULT {"ok": [...], "failed": [[pkg, why]]}."""
    if kind not in ("hint-update", "hint-new"):
        raise Error(f"unknown hint job: {kind}")
    ok, failed = [], []
    pkgs = _pkgmap(cfg)
    pairs, no_dl = hint_args(cfg, kind, list(args))
    for pkg, ver in pairs:
        try:
            path, target = _hint_target(cfg, pkg, pkgs)
            info = _read(os.path.join(cfg["repo_dir"], path, pkg + ".info")) or ""
            text = _read(target)
            if text is None:
                text = new_hint(info)
            _save(target, apply_version(text, info, ver, no_dl,
                                        lambda s, p=pkg: print(f"{p}: {s}", flush=True)))
            print(f"{pkg}: ok", flush=True)
            ok.append(pkg)
        except (Error, OSError) as e:
            print(f"{pkg}: failed: {e}", flush=True)
            failed.append([pkg, str(e)])
    print("SBB-RESULT " + json.dumps({"ok": ok, "failed": failed}), flush=True)
    if failed:
        raise Error(f"{len(failed)} failed")


def cmd_hint_diff(cfg, pkg=None):
    _, target = _hint_target(cfg, pkg)
    bak, current = _read(target + ".bak"), _read(target)
    if bak is None and current is None:
        raise Error("no backup")
    return {"bak": bak, "current": current}


def cmd_hint_revert(cfg, pkg=None):
    _, target = _hint_target(cfg, pkg)
    if os.path.exists(target + ".bak"):
        os.replace(target + ".bak", target)
    elif os.path.exists(target):
        os.remove(target)  # a new hint: reverting it is deleting it
    else:
        raise Error("nothing to revert")
    return {"ok": True}


def cmd_hint_delete(cfg, pkg=None):
    _, target = _hint_target(cfg, pkg)
    if not os.path.exists(target):
        raise Error(f"no hint for {pkg}")
    os.replace(target, target + ".bak")
    return {"ok": True}


def cmd_job_result(cfg, jid=None):
    """The SBB-RESULT of a hint job's log."""
    text, _ = _tail(os.path.join(job_dir(cfg, jid), "log"), LOG_TAIL)
    for line in reversed(text.splitlines()):
        if line.startswith("SBB-RESULT "):
            return json.loads(line.split(" ", 1)[1])
    raise Error(f"no result in {jid}")


# ── signing key ─────────────────────────────────────────────────────────────

def _signing_key(cfg):
    try:
        with open(cfg["genreprc"]) as f:
            m = re.search(r'^\s*REPOSOWNERGPG=["\']?([^"\'\s]+)', f.read(), re.M)
    except OSError:
        return None
    return m and m.group(1)


def _keygrips(key):
    """Keygrips of the signing-capable (sub)keys of key."""
    out = subprocess.run(["gpg", "--batch", "--with-colons", "--with-keygrip",
                          "--list-secret-keys", key], stdin=subprocess.DEVNULL,
                         capture_output=True, text=True).stdout
    grips, can_sign = [], False
    for line in out.splitlines():
        f = line.split(":")
        if f[0] in ("sec", "ssb"):
            can_sign = "s" in f[11]
        elif f[0] == "grp" and can_sign:
            grips.append(f[9])
    return grips


def gpg_unlocked(cfg):
    """True/False, or None when no signing key is configured or found."""
    key = _signing_key(cfg)
    grips = _keygrips(key) if key else []
    if not grips:
        return None
    for g in grips:
        out = subprocess.run(["gpg-connect-agent", f"keyinfo {g}", "/bye"],
                             stdin=subprocess.DEVNULL, capture_output=True, text=True).stdout
        info = next((ln.split() for ln in out.splitlines() if ln.startswith("S KEYINFO")), [])
        if len(info) < 7 or info[6] != "1":
            return False
    return True


def cmd_gpg_status(cfg):
    return {"unlocked": gpg_unlocked(cfg), "key": _signing_key(cfg)}


def cmd_gpg_unlock(cfg, stdin=None):
    key = _signing_key(cfg)
    grips = _keygrips(key) if key else []
    if not grips:
        raise Error("no signing key configured")
    passphrase = (stdin or sys.stdin.buffer).read().rstrip(b"\n")
    for g in grips:
        subprocess.run([PRESET, "--preset", g], input=passphrase, check=True,
                       capture_output=True)
    with tempfile.NamedTemporaryFile() as t:
        ok = subprocess.run(["gpg", "--batch", "--pinentry-mode", "error", "-u", key,
                             "-bas", "-o", "-", t.name], stdin=subprocess.DEVNULL,
                            capture_output=True).returncode == 0
    if not ok:
        for g in grips:
            subprocess.run([PRESET, "--forget", g], stdin=subprocess.DEVNULL, capture_output=True)
        raise Error("bad passphrase")
    return {"unlocked": True}


COMMANDS = {
    "version": cmd_version,
    "stats": cmd_stats,
    "jobs": cmd_jobs,
    "job-start": cmd_job_start,
    "job-follow": cmd_job_follow,
    "job-kill": cmd_job_kill,
    "inventory": cmd_inventory,
    "results": cmd_results,
    "build-log": cmd_build_log,
    "gpg-status": cmd_gpg_status,
    "gpg-unlock": cmd_gpg_unlock,
    "hint-diff": cmd_hint_diff,
    "hint-revert": cmd_hint_revert,
    "hint-delete": cmd_hint_delete,
    "job-result": cmd_job_result,
    "_run": cmd_run,
    "_hint": cmd_hint,
}


def main(argv):
    try:
        if not argv or argv[0] not in COMMANDS:
            names = ",".join(c for c in COMMANDS if not c.startswith("_"))
            raise Error(f"usage: sbb-remote {{{names}}} [args]")
        cfg = load_config()
        os.makedirs(os.path.join(cfg["state_dir"], "jobs"), exist_ok=True)
        result = COMMANDS[argv[0]](cfg, *argv[1:])
    except BrokenPipeError:
        # the GUI hung up on job-follow; silence the flush at exit too
        os.dup2(os.open(os.devnull, os.O_WRONLY), sys.stdout.fileno())
        return 0
    except Exception as e:
        print(json.dumps({"error": str(e) if isinstance(e, Error) else f"{type(e).__name__}: {e}"}))
        return 1
    if result is not None:
        print(json.dumps(result))
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))