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 json
import os
import re
import shlex
import shutil
import signal
import sqlite3
import subprocess
import sys
import tempfile
import time
import tomllib

VERSION = "0.4.1"
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",
}
FLAGS = ("--yes", "--no")
JOB_ID = re.compile(r"^\d{8}-\d{6}-[a-z]+(-\d+)?$")
PKG = re.compile(r"^[^/]+/([^/]+)/\1\.SlackBuild$")
PKGNAME = re.compile(r"^[A-Za-z0-9._+-]+$")
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
    if not all(isinstance(o, str) for o in cfg["overlays"]):
        raise Error("config: overlays: 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"], {}


KINDS = {"sbpull": kind_sbpull, "sbshadow": kind_sbshadow,
         "slackrepo": kind_slackrepo, "publish": kind_publish}


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}


# ── 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,
    "_run": cmd_run,
}


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:]))