# sb-builder Phase 2 (Packages and slackrepo builds) Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Package browser with a structured hint view, slackrepo build/rebuild/update/remove and update-all as previewed jobs, a publish gate with a separate Publish action, and signing key unlock for jobs without a tty. **Architecture:** slackrepo_setup becomes the home of the slackrepo config (site values moved to a root-only `.local`), gains `sb-publish` and the hook gates. `sbb-remote` 0.2.0 adds `inventory`, `slackrepo` and `publish` kinds (kinds now return argv plus env), `gpg-status` and `gpg-unlock` (passphrase on stdin only). The GUI adds `hints.py` (pure slackrepo hint rules), the Packages tab, Run it with a publish box and the passphrase dialog in the Jobs tab, the key indicator and tile links. Spec: `docs/superpowers/specs/2026-10-19-sb-builder-phase2-design.md`. **Tech Stack:** as phase 1: Python 3.11+ stdlib, PyQt6, bash; gpg 2.5 on the VM. **Verified while planning:** all code below ran in a scratch copy: 68 tests pass, offscreen. Against the real VM (helper copied to `/tmp`, temp state dir): `inventory` returns 9957 packages in 2.1s warm / 6.1s cold, 1.5 MB, 599 with a hint, hint key counts match the census; `gpg-status` reports `{"unlocked": false}` for the configured key; the Packages tab rendered that inventory correctly. The config hook gates were exercised with stubs for every `OPT_DRY_RUN`/`OPT_PREVIEW`/`SBB_PUBLISH` combination. `make-local.py` was run against a copy of the live config (5 keys found, mode 600, refuses to overwrite); that copy was shredded. **Rules for whoever executes this:** - Commits are GPG-signed; never disable signing. Global hooks scan for personal data; a rejection is right until proven otherwise. - Never stage `HANDOFF.md`. Never commit or print site values (rsync target, Gotify URL or key, hostnames). - Anything touching the VM or deleting files asks the user first: Task 1 Step 5 and Task 5. - GUI tests must run offscreen (the test file forces it). If a dialog ever appears on the desktop, stop: a fake remote is missing an answer. --- ## File map | file | change | |---|---| | slackrepo_setup: `slackrepo_SBo-danix.conf`, `sb-publish`, `SBo-danix.local.example`, `gpg-agent.conf`, `make-local.py`, `README.md`, `AGENTS.md` | Task 1 | | `remote/sbb-remote` | 0.2.0: new config keys, kinds return (argv, env), `slackrepo`/`publish` kinds, `inventory`, `gpg-status`, `gpg-unlock`, `stdin=DEVNULL` everywhere | | `remote/config.example.toml` | new keys | | `tests/test_remote.py` | 37 tests | | `sb_builder/__init__.py` | `VERSION = "0.2.0"` | | `sb_builder/config.py`, `config.example.toml` | `[builds] publish` | | `sb_builder/remote.py` | `call(..., stdin=)` | | `sb_builder/hints.py` | new: slackrepo hint rules | | `sb_builder/packages.py` | new: Packages tab | | `sb_builder/jobs.py` | Run it, publish box, signing gate | | `sb_builder/dashboard.py` | clickable cards | | `sb_builder/mainwindow.py` | Packages tab, key indicator, inventory reload after jobs | | `tests/test_gui.py` | 31 tests | | `README.md`, `AGENTS.md` | phase 2 docs | --- ### Task 1: slackrepo config, sb-publish and gates (repo slackrepo_setup) **Files:** create `slackrepo_SBo-danix.conf`, `sb-publish`, `SBo-danix.local.example`, `gpg-agent.conf`, `make-local.py`; modify `README.md`, `AGENTS.md`. `slackrepo_SBo-danix.conf` is the live VM config of 2026-10-19 with: a header that sources `/etc/slackrepo/SBo-danix.local`; `manage_packages_hook` and `gotify_hook` also checking `OPT_PREVIEW`; `rsync_push_hook` calling `sb-publish` and honouring `SBB_PUBLISH=n` and `OPT_PREVIEW`; Gotify URL and key read from `SB_GOTIFY_URL`/`SB_GOTIFY_KEY`. No site value is in the file. - [ ] **Step 1: Apply the patch** Save as `/tmp/slackrepo_setup-p2.patch`, then: ````diff diff '--color=auto' -ruN a/AGENTS.md b/AGENTS.md --- a/AGENTS.md +++ b/AGENTS.md @@ -43,6 +43,22 @@ Do **not** add a hintfile generator to this script — it would create a second owner of HINT_DIR and fight mkhint's `.bak` handling. +## slackrepo config and publishing + +`slackrepo_SBo-danix.conf` is the source of truth for `/etc/slackrepo/slackrepo_SBo-danix.conf`. +Never commit site values (rsync target, passwords, Gotify URL or key, hostnames): they live in the +root-only `/etc/slackrepo/SBo-danix.local`, which the config sources; `SBo-danix.local.example` +documents the keys with placeholders. slackrepo unsets only its own known config names after +sourcing (`varnames`, `genrepnames`, `initnames`, `OPTIONLIST` in `slackrepo`), so the `SB_*` +variables survive into the hooks, which run as functions in slackrepo's shell. + +Gates in the hooks: `SBB_PUBLISH=n` holds back `rsync_push_hook` (sb-builder sets it for +unpublished runs; `sb-publish` pushes later). `--preview` does not set `OPT_DRY_RUN` +(`cmdfunctions.sh`), so hooks check `OPT_PREVIEW` too. To check the gates without a VM, source +the config with `/usr/local/bin/` rewritten to a stub dir (stub `sb-publish`, +`manage-packages.sh`, `curl`) and call the hooks under each `OPT_DRY_RUN`/`OPT_PREVIEW`/ +`SBB_PUBLISH` combination. + ## Run ```bash diff '--color=auto' -ruN a/README.md b/README.md --- a/README.md +++ b/README.md @@ -25,6 +25,19 @@ `--yes` applies without asking, `--no` only shows the plan (for non-interactive callers such as sb-builder). `SBO_REPO` overrides the repo path. +- `slackrepo_SBo-danix.conf`: the slackrepo config for this repo + (`/etc/slackrepo/`). Site values (rsync target, Gotify URL and key) are not + in it: it sources the root-only `/etc/slackrepo/SBo-danix.local`, see + `SBo-danix.local.example`. Hooks honour `SBB_PUBLISH=n` (no rsync push, set by + sb-builder for unpublished runs), and a `--preview` neither notifies Gotify + nor runs `manage-packages.sh` for real. +- `sb-publish`: pushes the package repository to the mirror. Called by the + config's `rsync_push_hook`; run it by hand to publish builds made with + `SBB_PUBLISH=n`. +- `gpg-agent.conf`: root's gpg-agent config, with `allow-preset-passphrase` so + sb-builder can unlock the package signing key for jobs that have no tty. +- `make-local.py`: one-off migration, builds `SBo-danix.local` from the values + in an older config that still had them inline. - `sbpull.bash`: `sbpull {personal|pentesting} [--yes|--no]` shell function. Pulls one overlay subtree and runs `sbshadow` (passing the flag on), to carry overlay changes without a full rebuild. Source it from root's @@ -33,8 +46,11 @@ ## Install ```bash -install -m 755 slackrepo_setup_SBo-danix.sh sbshadow /usr/local/bin/ +install -m 755 slackrepo_setup_SBo-danix.sh sbshadow sb-publish /usr/local/bin/ echo ". /path/to/sbpull.bash" >> ~/.bashrc +install -m 644 slackrepo_SBo-danix.conf /etc/slackrepo/ +install -m 600 SBo-danix.local.example /etc/slackrepo/SBo-danix.local # then edit +install -m 600 gpg-agent.conf /root/.gnupg/ && gpgconf --reload gpg-agent ``` Optional: `mkhint` (mkhintfile repo) for phantom-dep hintfile diff '--color=auto' -ruN a/SBo-danix.local.example b/SBo-danix.local.example --- a/SBo-danix.local.example +++ b/SBo-danix.local.example @@ -0,0 +1,8 @@ +# /etc/slackrepo/SBo-danix.local: site settings for slackrepo_SBo-danix.conf +# and sb-publish. Not in git: it holds the Gotify key. Keep it root-only: +# install -m 600 SBo-danix.local.example /etc/slackrepo/SBo-danix.local +SB_RSYNC_TARGET='rsync://user@mirror.example.org/packages' +SB_RSYNC_PASSWORD_FILE='/root/.rsync-password' +SB_RSYNC_EXCLUDES='/root/.repo-excludes.txt' +SB_GOTIFY_URL='http://gotify.example.org/message' +SB_GOTIFY_KEY='replace-me' diff '--color=auto' -ruN a/gpg-agent.conf b/gpg-agent.conf --- a/gpg-agent.conf +++ b/gpg-agent.conf @@ -0,0 +1,10 @@ +pinentry-program /usr/bin/pinentry-curses +enable-ssh-support +# cache passwords for both gpg and ssh for 4h +max-cache-ttl-ssh 14400 +default-cache-ttl 14400 +# expire passwords after 6h +max-cache-ttl 21600 +max-cache-ttl-ssh 21600 +# let sb-builder unlock the signing key: jobs have no tty for pinentry +allow-preset-passphrase diff '--color=auto' -ruN a/make-local.py b/make-local.py --- a/make-local.py +++ b/make-local.py @@ -0,0 +1,31 @@ +#!/usr/bin/env python3 +# Copyright (C) 2026 Danilo M. +# +# 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. See the LICENSE file for details. + +# Run on the VM, as root, BEFORE installing the new slackrepo_SBo-danix.conf: +# copies the site values out of the old config into SBo-danix.local (mode 600). +# Prints only which keys it found, never their values. +import os, re, sys +old, dest = sys.argv[1], sys.argv[2] +if os.path.exists(dest): + sys.exit(f"{dest} exists, left alone") +s = open(old).read() +found = { + "SB_RSYNC_TARGET": re.search(r"local rsync_target='([^']*)'", s), + "SB_RSYNC_EXCLUDES": re.search(r"local rsync_excludes='([^']*)'", s), + "SB_RSYNC_PASSWORD_FILE": re.search(r"local rsync_password_file='([^']*)'", s), + "SB_GOTIFY_KEY": re.search(r'X-Gotify-Key: ([^"]*)"', s), + "SB_GOTIFY_URL": re.search(r'\n\s*"(https?://[^"]*)" >/dev/null', s), +} +missing = [k for k, m in found.items() if not m] +if missing: + sys.exit(f"not found in {old}: {', '.join(missing)}") +fd = os.open(dest, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) +with os.fdopen(fd, "w") as f: + f.write("# site settings for slackrepo_SBo-danix.conf and sb-publish (root-only)\n") + for k, m in found.items(): + f.write(f"{k}='{m.group(1)}'\n") +print(f"wrote {dest}: {', '.join(found)}") diff '--color=auto' -ruN a/sb-publish b/sb-publish --- a/sb-publish +++ b/sb-publish @@ -0,0 +1,35 @@ +#!/bin/bash +# Copyright (C) 2026 Danilo M. +# +# 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. See the LICENSE file for details. + +# sb-publish: push the local package repository to the public mirror. +# +# Usage: sb-publish [PKGREPO] (default /repo) +# +# Called by rsync_push_hook in slackrepo_SBo-danix.conf after a build, unless +# the build ran with SBB_PUBLISH=n; run it by hand (or from sb-builder's +# Publish action) to push such builds later. Settings come from the root-only +# SB_LOCAL file (default /etc/slackrepo/SBo-danix.local): +# SB_RSYNC_TARGET rsync destination (required) +# SB_RSYNC_PASSWORD_FILE --password-file, if the file exists +# SB_RSYNC_EXCLUDES --exclude-from, if the file exists + +set -euo pipefail + +SB_LOCAL=${SB_LOCAL:-/etc/slackrepo/SBo-danix.local} +[ -r "$SB_LOCAL" ] && . "$SB_LOCAL" +: "${SB_RSYNC_TARGET:?SB_RSYNC_TARGET is not set in $SB_LOCAL}" + +pkgrepo=${1:-/repo} +args=( -havz --delete-after --stats ) +[ -n "${SB_RSYNC_EXCLUDES:-}" ] && [ -f "$SB_RSYNC_EXCLUDES" ] && \ + args+=( --exclude-from="$SB_RSYNC_EXCLUDES" ) +[ -n "${SB_RSYNC_PASSWORD_FILE:-}" ] && [ -f "$SB_RSYNC_PASSWORD_FILE" ] && \ + args+=( --password-file="$SB_RSYNC_PASSWORD_FILE" ) + +echo "sb-publish: pushing $pkgrepo to $SB_RSYNC_TARGET ..." +rsync "${args[@]}" "$pkgrepo"/ "$SB_RSYNC_TARGET" +echo "sb-publish: rsync push complete." diff '--color=auto' -ruN a/slackrepo_SBo-danix.conf b/slackrepo_SBo-danix.conf --- a/slackrepo_SBo-danix.conf +++ b/slackrepo_SBo-danix.conf @@ -0,0 +1,310 @@ +# slackrepo configuration file for the SBo-danix personal repository. +# +# Priority order: +# (1) Environment variables +# (2) ~/.slackreporc +# (3) /etc/slackrepo/slackrepo_SBo-danix.conf +# +# %REPO%, %SLACKVER% and %ARCH% are replaced at runtime. + +#------------------------------------------------------------------------------- +# Filestore locations + +# Personal SlackBuilds repo (managed as a git fork of SBo): +# Site settings (rsync target, Gotify URL and key) live outside git in a +# root-only file, see SBo-danix.local.example in slackrepo_setup. +SB_LOCAL='/etc/slackrepo/SBo-danix.local' +[ -r "$SB_LOCAL" ] && . "$SB_LOCAL" + +SBREPO="/var/lib/sbopkg/SBo-danix" + +# Cached sources: +SRCREPO="/var/lib/slackrepo/%REPO%/source" + +# Built packages: +#PKGREPO="/var/lib/slackrepo/%REPO%/packages/%SLACKVER%/%ARCH%" +PKGREPO="/repo" + +# Backups (comment out to disable): +#PKGBACKUP="/var/lib/slackrepo/%REPO%/backups/%SLACKVER%/%ARCH%" + +# Local hintfiles (for overriding SBo packages without touching the SlackBuild): +HINTDIR="/etc/slackrepo/%REPO%/hintfiles" + +# Default hintfiles from slackrepo-hints package (optional): +DEFAULT_HINTDIR="/etc/slackrepo/%REPO%/default_hintfiles/%SLACKVER%" + +# Logs: +LOGDIR="/var/log/slackrepo/%REPO%" + +# Database: +DATABASE="/var/lib/slackrepo/%REPO%/database_%REPO%_%ARCH%.sqlite3" + +# Temporary build space (needs plenty of room): +TMP="/tmp/%REPO%" + +#------------------------------------------------------------------------------- +# Package tag +# Using _danix instead of _SBo to identify your personal builds. +TAG='_danix' + +# Package compression: +PKGTYPE='txz' + +#------------------------------------------------------------------------------- +# Build control + +SUBSTITUTE="" +NUMJOBS='' +ARCH='' + +#------------------------------------------------------------------------------- +# Control argument defaults + +VERBOSE='n' +VERY_VERBOSE='n' +DRY_RUN='n' +INSTALL='n' +CHROOT='/' +LINT='n' +KEEP_TMP='n' +COLOR='auto' +NICE='5' +NOWARNING='' + +#------------------------------------------------------------------------------- +# Git branch setup +# +# INIT_GITCLONE is only used if SBREPO does not exist yet. +# INIT_GITBRANCH must match the branch you have checked out in SBREPO. +# Since we use a personal branch, gitfetch_hook is replaced below with a +# custom hook that fetches upstream and rebases your branch on top of it. + +INIT_GITCLONE='https://github.com/Ponce/slackbuilds.git' +INIT_GITBRANCH='danix-current' + +#------------------------------------------------------------------------------- +# Hooks + +function danix_gitfetch_hook +{ + local upstream_branch='current' + local personal_branch='danix-current' + + cd "$SR_SBREPO" || return 1 + + local currbranch + currbranch=$(git rev-parse --abbrev-ref HEAD) + if [ "$currbranch" != "$personal_branch" ]; then + echo "slackrepo: WARNING: SBREPO is on branch '$currbranch', expected '$personal_branch'. Skipping git update." + return 0 + fi + + local muck + muck=$(git status -s .) + if [ -n "$muck" ]; then + echo "slackrepo: WARNING: SBREPO has uncommitted changes. Skipping git update." + return 0 + fi + + echo "slackrepo: Fetching upstream SBo..." + git fetch origin || { echo "slackrepo: WARNING: git fetch failed."; return 1; } + + echo "slackrepo: Rebasing $personal_branch onto origin/$upstream_branch..." + git rebase --rebase-merges -X theirs origin/"$upstream_branch" || { + echo "slackrepo: ERROR: rebase failed. Resolve conflicts in $SR_SBREPO and run 'git rebase --continue'." + return 1 + } + + return 0 +} + +HOOK_START=( danix_gitfetch_hook ) + +function manage_packages_hook +{ + local args=() + # --preview does not set OPT_DRY_RUN (slackrepo cmdfunctions.sh), check both + if [ "$OPT_DRY_RUN" = 'y' ] || [ "$OPT_PREVIEW" = 'y' ]; then + args+=( --dry-run ) + fi + REPO="$SR_PKGREPO" /usr/local/bin/manage-packages.sh "${args[@]}" +} + +#------------------------------------------------------------------------------- +# generate web files + +function gen_web_hook +{ + if [ "$OPT_DRY_RUN" = 'y' ] || [ ! -s "$CHANGELOG" ]; then + return 0 + fi + SR_PKGREPO="$SR_PKGREPO" /usr/local/bin/gen_web_hook.sh + local stat=$? + [ $stat -ne 0 ] && log_warning "gen_web_hook failed, status $stat" + return $stat +} + +#------------------------------------------------------------------------------- +# dep template generator +# Writes /.template for each successfully built package +# that has one or more recursive dependencies. + +TEMPLATE_DIR='/repo/templates' + +function dep_template_hook +{ + local template_dir="${TEMPLATE_DIR:-/repo/templates}" + + [ "$OPT_DRY_RUN" = 'y' ] && return 0 + [ ${#OKLIST[@]} -eq 0 ] && return 0 + + mkdir -p "$template_dir" || return 1 + + local itemid prgnam dep depname outfile + for itemid in "${OKLIST[@]}"; do + [ -z "${FULLDEPS[$itemid]}" ] && continue + + prgnam="${ITEMPRGNAM[$itemid]}" + outfile="$template_dir/${prgnam}.template" + + { + for dep in ${FULLDEPS[$itemid]}; do + depname="${ITEMPRGNAM[$dep]}" + printf '%s\n' "${depname:-$dep}" + done + printf '%s\n' "$prgnam" + } > "$outfile" + + log_normal "dep_template_hook: wrote $outfile" + done + + return 0 +} + +#------------------------------------------------------------------------------- +# rsync push config + +function rsync_push_hook +{ + # SBB_PUBLISH=n (set by sb-builder for an unpublished run) holds the push + # back; sb-publish pushes later. Unset means publish, as from a terminal. + if [ "$OPT_DRY_RUN" = 'y' ] || [ "$OPT_PREVIEW" = 'y' ] || [ ! -s "$CHANGELOG" ] \ + || [ "${SBB_PUBLISH:-y}" = 'n' ]; then + return 0 + fi + /usr/local/bin/sb-publish "$SR_PKGREPO" + local rsyncstat=$? + if [ "$rsyncstat" != 0 ]; then + log_warning "rsync_push_hook failed, status $rsyncstat" + return 1 + fi + return 0 +} + +function recap_hook +{ + # TODOLIST is per-item scratch (overwritten on each build_command call), so it + # only ever holds the last item. Build the recap from the cumulative status + # lists that slackrepo dedupes before running the finish hooks. + local recaplist=( + "${OKLIST[@]}" "${FAILEDLIST[@]}" "${ABORTEDLIST[@]}" + "${SKIPPEDLIST[@]}" "${UNSUPPORTEDLIST[@]}" + ) + recaplist=( $(printf '%s\n' "${recaplist[@]}" | sort -u) ) + [ ${#recaplist[@]} -eq 0 ] && return 0 + + local col_name=4 col_ver=7 col_status=6 + local itemid prgnam ver stat + for itemid in "${recaplist[@]}"; do + prgnam="${ITEMPRGNAM[$itemid]}" + ver="${HINT_VERSION[$itemid]:-${INFOVERSION[$itemid]:-?}}" + stat="${STATUS[$itemid]:-?}" + [ ${#prgnam} -gt $col_name ] && col_name=${#prgnam} + [ ${#ver} -gt $col_ver ] && col_ver=${#ver} + [ ${#stat} -gt $col_status ] && col_status=${#stat} + done + + local sep + sep="+-$(printf '%*s' "$col_name" '' | tr ' ' '-')-+-$(printf '%*s' "$col_ver" '' | tr ' ' '-')-+-$(printf '%*s' "$col_status" '' | tr ' ' '-')-+" + + echo "" + echo "Build recap:" + echo "$sep" + printf "| %-*s | %-*s | %-*s |\n" "$col_name" "Name" "$col_ver" "Version" "$col_status" "Status" + echo "$sep" + for itemid in "${recaplist[@]}"; do + prgnam="${ITEMPRGNAM[$itemid]}" + ver="${HINT_VERSION[$itemid]:-${INFOVERSION[$itemid]:-?}}" + stat="${STATUS[$itemid]:-?}" + printf "| %-*s | %-*s | %-*s |\n" "$col_name" "$prgnam" "$col_ver" "$ver" "$col_status" "$stat" + done + echo "$sep" + echo "" + + return 0 +} + +function gotify_hook +{ + # --preview does not set OPT_DRY_RUN (slackrepo cmdfunctions.sh), check both + if [ "$OPT_DRY_RUN" = 'y' ] || [ "$OPT_PREVIEW" = 'y' ]; then + return 0 + fi + if [ -z "${SB_GOTIFY_URL:-}" ] || [ -z "${SB_GOTIFY_KEY:-}" ]; then + log_warning "SB_GOTIFY_URL/SB_GOTIFY_KEY not set in $SB_LOCAL, skipping gotify_hook" + return 0 + fi + + local curl_bin + curl_bin=$(which curl 2>/dev/null) + if [ ! -x "$curl_bin" ]; then + log_warning "curl not found, skipping gotify_hook" + return 1 + fi + + local ok="${#OKLIST[@]}" + local failed="${#FAILEDLIST[@]}" + local aborted="${#ABORTEDLIST[@]}" + local skipped="${#SKIPPEDLIST[@]}" + + local title msg priority + if [ $(( failed + aborted )) -gt 0 ]; then + title="slackrepo [$CMD]: FAILED" + priority=8 + else + title="slackrepo [$CMD]: OK" + priority=5 + fi + + msg="OK: $ok Failed: $failed Aborted: $aborted Skipped: $skipped" + [ ${#OKLIST[@]} -gt 0 ] && msg+=$'\nOK: '"$(printf '%s ' "${OKLIST[@]}")" + [ ${#FAILEDLIST[@]} -gt 0 ] && msg+=$'\nFailed: '"$(printf '%s ' "${FAILEDLIST[@]}")" + [ ${#ABORTEDLIST[@]} -gt 0 ] && msg+=$'\nAborted: '"$(printf '%s ' "${ABORTEDLIST[@]}")" + + "$curl_bin" -s \ + -H "X-Gotify-Key: $SB_GOTIFY_KEY" \ + -F "title=$title" \ + -F "message=$msg" \ + -F "priority=$priority" \ + "$SB_GOTIFY_URL" >/dev/null + local stat=$? + [ $stat -ne 0 ] && log_warning "gotify_hook failed, status $stat" + return $stat +} + +function bell_hook +{ + printf '\a' + return 0 +} + +HOOK_FINISH=( manage_packages_hook bell_hook genrepos_hook gen_web_hook dep_template_hook rsync_push_hook recap_hook gotify_hook ) + +#------------------------------------------------------------------------------- +# gen_repos_files.sh (disabled by default) + +USE_GENREPOS='1' + +#------------------------------------------------------------------------------- +export XZ_OPT="--threads=0" ```` ```bash cd ~/Programming/GIT/slackrepo_setup && git apply /tmp/slackrepo_setup-p2.patch && chmod 755 sb-publish bash -n slackrepo_SBo-danix.conf && bash -n sb-publish && python3 -m py_compile make-local.py && echo ok grep -cE 'X-Gotify-Key: [A-Za-z0-9]{8,}|rsync://[^$]' slackrepo_SBo-danix.conf # expect 0 ``` - [ ] **Step 2: Check the hook gates with stubs** ```bash T=$(mktemp -d); mkdir $T/stub for b in sb-publish manage-packages.sh; do printf '#!/bin/sh\necho "%s $*"\n' $b > $T/stub/$b; done printf '#!/bin/sh\necho "curl $*" >> %s/curl.log\n' $T > $T/stub/curl; chmod 755 $T/stub/* sed "s|/usr/local/bin/|$T/stub/|g; s|^SB_LOCAL=.*|SB_LOCAL=/nonexistent|" \ ~/Programming/GIT/slackrepo_setup/slackrepo_SBo-danix.conf > $T/test.conf for e in "OPT_DRY_RUN=n OPT_PREVIEW=n" "OPT_DRY_RUN=n OPT_PREVIEW=n SBB_PUBLISH=n" \ "OPT_DRY_RUN=n OPT_PREVIEW=y" "OPT_DRY_RUN=y OPT_PREVIEW=n"; do echo "== $e"; : > $T/curl.log env $e PATH=$T/stub:$PATH bash -c 'log_warning() { echo "WARN $*"; }; log_normal() { :; } . '"$T"'/test.conf; SB_GOTIFY_URL=http://example.org/m SB_GOTIFY_KEY=k CHANGELOG='"$T"'/cl; echo x > $CHANGELOG; SR_PKGREPO=/repo; CMD=build; OKLIST=(a) rsync_push_hook; manage_packages_hook; gotify_hook' grep -o 'X-Gotify-Key: k\|http://example.org/m' $T/curl.log | tr '\n' ' '; echo done ``` Expected: ``` == OPT_DRY_RUN=n OPT_PREVIEW=n sb-publish /repo manage-packages.sh X-Gotify-Key: k http://example.org/m == OPT_DRY_RUN=n OPT_PREVIEW=n SBB_PUBLISH=n manage-packages.sh X-Gotify-Key: k http://example.org/m == OPT_DRY_RUN=n OPT_PREVIEW=y manage-packages.sh --dry-run == OPT_DRY_RUN=y OPT_PREVIEW=n manage-packages.sh --dry-run ``` - [ ] **Step 3: Commit and push slackrepo_setup** ```bash cd ~/Programming/GIT/slackrepo_setup git add slackrepo_SBo-danix.conf sb-publish SBo-danix.local.example gpg-agent.conf make-local.py README.md AGENTS.md git commit -F - <<'EOF' Track the slackrepo config; add sb-publish and publish gates slackrepo_SBo-danix.conf is now kept here. Site values (rsync target, Gotify URL and key) moved to the root-only SBo-danix.local it sources; make-local.py migrates them out of an old inline config. The rsync push moved into sb-publish; rsync_push_hook calls it and holds back when SBB_PUBLISH=n, so sb-builder can build without publishing and publish later. --preview does not set OPT_DRY_RUN, so the hooks now also check OPT_PREVIEW: previews no longer notify Gotify or run manage-packages.sh for real. gpg-agent.conf adds allow-preset-passphrase so sb-builder can unlock the signing key for jobs without a tty. Co-Authored-By: Claude Opus 5.5 EOF git log --format='%h %G? %s' -1 && git push ``` - [ ] **Step 4: Remove the stale config copy (ask the user first)** `~/Programming/GIT/repo-html-structure/slackrepo_SBo-danix.conf` is untracked there and may still hold the Gotify key. With the user's OK: `shred -u ~/Programming/GIT/repo-html-structure/slackrepo_SBo-danix.conf`. - [ ] **Step 5: VM deploy is Task 5.** --- ### Task 2: helper 0.2.0 **Files:** replace `remote/sbb-remote`, `remote/config.example.toml`, `tests/test_remote.py`. - [ ] **Step 1: Write the tests** (replace `tests/test_remote.py`): ````python # sb-builder - self-checks for the remote helper # Copyright (C) 2026 Danilo M. # # 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. """Run: python3 -m unittest discover -s tests""" import contextlib import fcntl import importlib.machinery import importlib.util import io import json import os import shutil import sqlite3 import tempfile import threading import time import unittest from unittest import mock SCRIPT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "remote", "sbb-remote") _loader = importlib.machinery.SourceFileLoader("sbb_remote", SCRIPT) sbb = importlib.util.module_from_spec(importlib.util.spec_from_loader("sbb_remote", _loader)) _loader.exec_module(sbb) class Base(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, self.tmp) conf = os.path.join(self.tmp, "config.toml") with open(conf, "w") as f: f.write(f'state_dir = "{self.tmp}"\n' f'repo_dir = "{self.tmp}/repo"\n' f'slackrepo_db = "{self.tmp}/slackrepo.sqlite3"\n' f'hintdir = "{self.tmp}/hints"\n' f'genreprc = "{self.tmp}/genreprc"\n') patcher = mock.patch.dict(os.environ, {"SBB_CONFIG": conf}) patcher.start() self.addCleanup(patcher.stop) self.cfg = sbb.load_config() os.makedirs(os.path.join(self.tmp, "jobs")) def make_job(self, argv, **meta): jid = "20260101-000000-test" d = os.path.join(self.tmp, "jobs", jid) os.mkdir(d) sbb.write_meta(d, {"id": jid, "kind": "test", "args": [], "argv": argv, "status": "starting", "created": time.time(), **meta}) open(os.path.join(d, "log"), "wb").close() return jid, d def make_pkg(self, path, info="", hint=None, desc=None): d = os.path.join(self.cfg["repo_dir"], path) pkg = os.path.basename(path) os.makedirs(d, exist_ok=True) open(os.path.join(d, pkg + ".SlackBuild"), "w").close() with open(os.path.join(d, pkg + ".info"), "w") as f: f.write(info) if desc: with open(os.path.join(d, "slack-desc"), "w") as f: f.write(f"{pkg}: {pkg} ({desc})\n{pkg}:\n") if hint is not None: os.makedirs(self.cfg["hintdir"], exist_ok=True) with open(os.path.join(self.cfg["hintdir"], pkg + ".hint"), "w") as f: f.write(hint) def fake_bin(self, name, script): bindir = os.path.join(self.tmp, "bin") os.makedirs(bindir, exist_ok=True) path = os.path.join(bindir, name) with open(path, "w") as f: f.write("#!/bin/sh\n" + script) os.chmod(path, 0o755) if bindir not in os.environ["PATH"]: patcher = mock.patch.dict(os.environ, {"PATH": bindir + os.pathsep + os.environ["PATH"]}) patcher.start() self.addCleanup(patcher.stop) return path def hold_lock(self): f = open(os.path.join(self.tmp, "lock"), "a") fcntl.flock(f, fcntl.LOCK_EX) self.addCleanup(f.close) def run_main(self, *argv): out = io.StringIO() with contextlib.redirect_stdout(out): rc = sbb.main(list(argv)) return rc, json.loads(out.getvalue()) def wait_status(self, d, statuses): for _ in range(100): if sbb.read_meta(d)["status"] in statuses: break time.sleep(0.05) return sbb.read_meta(d) class ConfigTest(Base): def write(self, text): p = os.path.join(self.tmp, "c.toml") with open(p, "w") as f: f.write(text) return p def test_missing_file_gives_defaults(self): self.assertEqual(sbb.load_config(os.path.join(self.tmp, "nope.toml")), sbb.DEFAULTS) def test_override(self): self.assertEqual(sbb.load_config(self.write('overlays = ["a"]\n'))["overlays"], ["a"]) def test_example_matches_defaults(self): example = os.path.join(os.path.dirname(SCRIPT), "config.example.toml") self.assertEqual(sbb.load_config(example), sbb.DEFAULTS) def test_unknown_key(self): with self.assertRaisesRegex(sbb.Error, "bogus: unknown key"): sbb.load_config(self.write("bogus = 1\n")) def test_wrong_type(self): with self.assertRaisesRegex(sbb.Error, "state_dir: expected str"): sbb.load_config(self.write("state_dir = 1\n")) class KindsTest(Base): def test_sbpull_argv(self): self.assertEqual(sbb.KINDS["sbpull"](self.cfg, ["personal", "--no"]), (["bash", "-c", '. "$1" && sbpull "$2" "$3"', "_", self.cfg["sbpull_script"], "personal", "--no"], {})) def test_sbshadow_argv(self): self.assertEqual(sbb.KINDS["sbshadow"](self.cfg, ["--yes"]), (["sbshadow", "--yes"], {})) def test_slackrepo_argv(self): self.make_pkg("misc/nuclei") self.make_pkg("personal/foo") kind = sbb.KINDS["slackrepo"] self.assertEqual(kind(self.cfg, ["update", "--preview", "nuclei", "foo", "nuclei"]), (["slackrepo", "--repo=SBo-danix", "update", "--preview", "nuclei", "foo"], {})) self.assertEqual(kind(self.cfg, ["build", "--no-publish", "foo"]), (["slackrepo", "--repo=SBo-danix", "build", "foo"], {"SBB_PUBLISH": "n"})) self.assertEqual(kind(self.cfg, ["update"]), (["slackrepo", "--repo=SBo-danix", "update"], {})) def test_slackrepo_rejects(self): self.make_pkg("misc/nuclei") for args in (["install", "nuclei"], ["build"], ["remove"], ["build", "nope"], ["build", "../etc"], ["build", "--force", "nuclei"]): with self.subTest(args=args), self.assertRaises(sbb.Error): sbb.KINDS["slackrepo"](self.cfg, args) def test_publish(self): self.assertEqual(sbb.KINDS["publish"](self.cfg, []), (["sb-publish"], {})) with self.assertRaises(sbb.Error): sbb.KINDS["publish"](self.cfg, ["x"]) def test_rejects_bad_args(self): for kind, args in (("sbpull", ["evil", "--no"]), ("sbpull", ["personal"]), ("sbpull", ["personal", "--force"]), ("sbshadow", []), ("sbshadow", ["--yes", "--no"])): with self.subTest(kind=kind, args=args), self.assertRaises(sbb.Error): sbb.KINDS[kind](self.cfg, args) def test_unknown_kind(self): with self.assertRaisesRegex(sbb.Error, "unknown kind"): sbb.cmd_job_start(self.cfg, "rm", "-rf", "/") class MainTest(Base): def test_unknown_command(self): rc, data = self.run_main("nope") self.assertEqual(rc, 1) self.assertIn("usage", data["error"]) def test_version(self): self.assertEqual(self.run_main("version"), (0, {"version": sbb.VERSION, "overlays": ["personal", "pentesting"]})) class InventoryTest(Base): def test_rows(self): self.make_pkg("misc/nuclei", 'VERSION="3.11.0"\nMAINTAINER="Some One"\n' 'REQUIRES="go \\\n git"\n', desc="Fast scanner") self.make_pkg("personal/foo", 'VERSION="1.0"\n', hint='VERSION="1.1" # bumped\nDELREQUIRES="a b"\nOPTIONS="X=$(uname -m)"\n') os.makedirs(os.path.join(self.cfg["repo_dir"], "misc", "notapkg")) db = sqlite3.connect(self.cfg["slackrepo_db"]) db.executescript(""" create table buildresults (itemid text primary key, time text, result text); create table revisions (itemid text, dep text, deplist text, version text, built text, rev text, os text, hintcksum text, primary key (itemid, dep)); insert into buildresults values ('misc/nuclei', '5', 'ok'), ('nuclei', '1', 'bad'); insert into revisions values ('misc/nuclei', '/', '/', '3.11.0', '1788978222', 'r', 'o', '/'); """) db.close() inv = sbb.cmd_inventory(self.cfg) rows = [dict(zip(inv["columns"], r)) for r in inv["rows"]] self.assertEqual([r["path"] for r in rows], ["misc/nuclei", "personal/foo"]) nuclei, foo = rows self.assertEqual((nuclei["version"], nuclei["built_version"], nuclei["built_time"], nuclei["result"], nuclei["maintainer"], nuclei["requires"], nuclei["summary"], nuclei["hint_path"]), ("3.11.0", "3.11.0", 1788978222, "ok", "Some One", ["go", "git"], "Fast scanner", None)) self.assertEqual((foo["version"], foo["info_version"], foo["hint_version"], foo["result"]), ("1.1", "1.0", "1.1", None)) self.assertEqual(foo["hints"]["DELREQUIRES"], "a b") self.assertEqual(foo["hints"]["OPTIONS"], {"value": '"X=$(uname -m)"', "shell": True}) json.dumps(inv) def test_hint_search_order(self): self.make_pkg("misc/foo") sub = os.path.join(self.cfg["hintdir"], "misc", "foo") os.makedirs(sub) with open(os.path.join(sub, "foo.hint"), "w") as f: f.write('VERSION="2"\n') self.assertEqual(sbb.find_hint(self.cfg, "misc/foo"), os.path.join(sub, "foo.hint")) with open(os.path.join(self.cfg["hintdir"], "foo.hint"), "w") as f: f.write('VERSION="3"\n') self.assertEqual(sbb.find_hint(self.cfg, "misc/foo"), os.path.join(self.cfg["hintdir"], "foo.hint")) in_repo = os.path.join(self.cfg["repo_dir"], "misc", "foo", "foo.hint") open(in_repo, "w").close() self.assertEqual(sbb.find_hint(self.cfg, "misc/foo"), in_repo) def test_missing_db(self): self.make_pkg("misc/foo", 'VERSION="1"\n') row = sbb.cmd_inventory(self.cfg)["rows"][0] self.assertEqual(row[4:7], [None, None, None]) class GpgTest(Base): def setup_key(self, cached): with open(self.cfg["genreprc"], "w") as f: f.write('REPOSOWNERGPG="ABCD1234"\nUSE_GPGAGENT=1\n') self.fake_bin("gpg", 'case "$*" in\n' ' *--list-secret-keys*) printf "sec:u:255:22:K:1::::::scESC:\\n' 'grp:::::::::GRIPSIGN:\\nssb:u:255:18:S:1::::::e:\\ngrp:::::::::GRIPENC:\\n" ;;\n' f' *-bas*) [ -f "{self.tmp}/good" ] ;;\n' 'esac\n') flag = "1" if cached else "-" self.fake_bin("gpg-connect-agent", f'echo "S KEYINFO GRIPSIGN D - - {flag} P - - -"; echo OK\n') def test_status(self): self.assertEqual(sbb.cmd_gpg_status(self.cfg), {"unlocked": None, "key": None}) self.setup_key(cached=False) self.assertEqual(sbb.cmd_gpg_status(self.cfg), {"unlocked": False, "key": "ABCD1234"}) self.setup_key(cached=True) self.assertEqual(sbb.cmd_gpg_status(self.cfg)["unlocked"], True) def test_locked_refuses_real_run_only(self): self.setup_key(cached=False) self.make_pkg("misc/foo") with self.assertRaisesRegex(sbb.Error, "signing key locked"): sbb.cmd_job_start(self.cfg, "slackrepo", "build", "foo") self.fake_bin("slackrepo", "exit 0\n") self.assertIn("id", sbb.cmd_job_start(self.cfg, "slackrepo", "build", "--preview", "foo")) def test_unlock(self): self.setup_key(cached=False) log = os.path.join(self.tmp, "preset.log") preset = self.fake_bin("preset", f'echo "$* $(cat)" >> {log}\n') with mock.patch.object(sbb, "PRESET", preset): with self.assertRaisesRegex(sbb.Error, "bad passphrase"): sbb.cmd_gpg_unlock(self.cfg, stdin=io.BytesIO(b"wrong\n")) open(os.path.join(self.tmp, "good"), "w").close() self.assertEqual(sbb.cmd_gpg_unlock(self.cfg, stdin=io.BytesIO(b"right\n")), {"unlocked": True}) with open(log) as f: self.assertEqual(f.read().splitlines(), ["--preset GRIPSIGN wrong", "--forget GRIPSIGN ", "--preset GRIPSIGN right"]) class JobsTest(Base): def log(self, d): with open(os.path.join(d, "log"), "rb") as f: return f.read() def test_run_ok(self): jid, d = self.make_job(["sh", "-c", "echo hello"]) sbb.cmd_run(self.cfg, jid) m = sbb.read_meta(d) self.assertEqual((m["status"], m["exit"]), ("ok", 0)) self.assertEqual(self.log(d), b"hello\n") def test_run_failed(self): jid, d = self.make_job(["sh", "-c", "exit 3"]) sbb.cmd_run(self.cfg, jid) m = sbb.read_meta(d) self.assertEqual((m["status"], m["exit"]), ("failed", 3)) def test_run_missing_program(self): jid, d = self.make_job(["/nonexistent/prog"]) sbb.cmd_run(self.cfg, jid) self.assertEqual(sbb.read_meta(d)["status"], "failed") self.assertIn(b"sbb-remote:", self.log(d)) def test_kill(self): jid, d = self.make_job(["sleep", "30"]) t = threading.Thread(target=sbb.cmd_run, args=(self.cfg, jid)) t.start() self.wait_status(d, ("running",)) self.assertEqual(sbb.cmd_job_kill(self.cfg, jid), {"ok": True}) t.join(5) self.assertEqual(sbb.read_meta(d)["status"], "killed") def test_kill_not_running(self): jid, _ = self.make_job(["true"], status="ok") with self.assertRaisesRegex(sbb.Error, "not running"): sbb.cmd_job_kill(self.cfg, jid) def test_busy(self): self.hold_lock() jid, d = self.make_job(["true"]) sbb.cmd_run(self.cfg, jid) self.assertEqual(sbb.read_meta(d)["status"], "busy") def test_lost_when_lock_free(self): self.make_job(["true"], status="running", pid=1) self.assertEqual(sbb.cmd_jobs(self.cfg)["jobs"][0]["status"], "lost") def test_running_kept_while_lock_held(self): self.hold_lock() self.make_job(["true"], status="running", pid=1) self.assertEqual(sbb.cmd_jobs(self.cfg)["jobs"][0]["status"], "running") def test_starting_grace(self): _, d = self.make_job(["true"]) self.assertEqual(sbb.cmd_jobs(self.cfg)["jobs"][0]["status"], "starting") sbb.write_meta(d, dict(sbb.read_meta(d), created=time.time() - 60)) self.assertEqual(sbb.cmd_jobs(self.cfg)["jobs"][0]["status"], "lost") def test_follow_from_offset(self): jid, d = self.make_job(["true"], status="ok") with open(os.path.join(d, "log"), "wb") as f: f.write(b"hello world") out = io.BytesIO() sbb.cmd_job_follow(self.cfg, jid, "6", out=out) self.assertEqual(out.getvalue(), b"world") def test_bad_job_id(self): with self.assertRaisesRegex(sbb.Error, "bad job id"): sbb.cmd_job_follow(self.cfg, "../../etc", "0") def fake_sbshadow(self): bindir = os.path.join(self.tmp, "bin") os.mkdir(bindir) fake = os.path.join(bindir, "sbshadow") with open(fake, "w") as f: f.write('#!/bin/sh\necho "shadow: cat/$1"\n') os.chmod(fake, 0o755) patcher = mock.patch.dict(os.environ, {"PATH": bindir + os.pathsep + os.environ["PATH"]}) patcher.start() self.addCleanup(patcher.stop) def test_env_reaches_child(self): jid, d = self.make_job(["sh", "-c", 'echo "publish=$SBB_PUBLISH"'], env={"SBB_PUBLISH": "n"}) sbb.cmd_run(self.cfg, jid) self.assertEqual(self.log(d), b"publish=n\n") def test_start_end_to_end(self): self.fake_sbshadow() rc, data = self.run_main("job-start", "sbshadow", "--no") self.assertEqual(rc, 0) d = os.path.join(self.tmp, "jobs", data["id"]) self.assertEqual(self.wait_status(d, ("ok", "failed", "lost"))["status"], "ok") self.assertEqual(self.log(d), b"shadow: cat/--no\n") def test_start_busy(self): self.fake_sbshadow() self.hold_lock() jid, _ = self.make_job(["true"], status="running", pid=1) self.assertEqual(self.run_main("job-start", "sbshadow", "--no"), (0, {"busy": jid})) self.assertEqual(os.listdir(os.path.join(self.tmp, "jobs")), [jid]) class StatsTest(Base): def test_missing_sources_are_null(self): s = sbb.cmd_stats(self.cfg) self.assertIsNotNone(s["load"]) self.assertIsNone(s["packages"]) self.assertIsNone(s["slackrepo"]) self.assertEqual(s["sync"], {"personal": None, "pentesting": None}) json.dumps(s) def test_counts(self): r = self.cfg["repo_dir"] for p in ("cat/foo", "cat/bar", "personal/foo", "personal/baz", "pentesting/nmap"): os.makedirs(os.path.join(r, p)) open(os.path.join(r, p, os.path.basename(p) + ".SlackBuild"), "w").close() os.makedirs(os.path.join(r, "cat", "notapkg")) def fake_git(cfg, *args): if args[0] == "ls-tree": return "cat/foo/foo.SlackBuild\ncat/foo/README\ncat/old/old.SlackBuild\n" return "1792351223\n" with mock.patch.object(sbb, "_git", fake_git): s = sbb.cmd_stats(self.cfg) self.assertEqual(s["packages"], {"total": 5, "personal": 2, "pentesting": 1, "shadowed": 1}) self.assertEqual(s["sync"]["personal"], 1792351223) def test_slackrepo(self): db = sqlite3.connect(self.cfg["slackrepo_db"]) db.executescript(""" create table buildresults (itemid text primary key, time text, result text); create table revisions (itemid text, dep text, deplist text, version text, built text, rev text, os text, hintcksum text, primary key (itemid, dep)); insert into buildresults values ('a', '1', 'ok'), ('b', '1', 'ok'), ('c', '1', 'failed'); insert into revisions values ('cat/a', '/', '/', '1', '1', 'r', 'o', '/'), ('cat/a', 'dep', '/', '1', '1', 'r', 'o', '/'); """) db.close() self.assertEqual(sbb.cmd_stats(self.cfg)["slackrepo"], {"results": {"ok": 2, "failed": 1}, "built": 1}) if __name__ == "__main__": unittest.main() ```` - [ ] **Step 2: Run them, expect failures** Run: `python3 -m unittest tests.test_remote # # 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.2.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", } 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" 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 /.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: results = dict(db.execute("select result, count(*) from buildresults 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 cmd_stats(cfg): dirs = {"repo": cfg["repo_dir"], "state": cfg["state_dir"]} return { "load": _try(os.getloadavg), "uptime": _try(_uptime), "disk_free": {k: _try(lambda p=p: shutil.disk_usage(p).free) for k, p in dirs.items()}, "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 .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 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, "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:])) ```` Replace `remote/config.example.toml`: ````toml # sbb-remote config on the VM: /etc/sb-builder/config.toml # Every key is optional; these are the 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 = "" # searched after hintdir when set genreprc = "/root/.genreprc" # REPOSOWNERGPG names the signing key ```` - [ ] **Step 4: Run the tests** Run: `python3 -m unittest tests.test_remote EOF git log --format='%h %G? %s' -1 ``` --- ### Task 3: GUI **Files:** replace `sb_builder/__init__.py`, `sb_builder/config.py`, `config.example.toml`, `sb_builder/remote.py`, `sb_builder/jobs.py`, `sb_builder/dashboard.py`, `sb_builder/mainwindow.py`, `tests/test_gui.py`; create `sb_builder/hints.py`, `sb_builder/packages.py`. - [ ] **Step 1: Write the tests** (replace `tests/test_gui.py`): ````python # sb-builder - self-checks for the GUI # Copyright (C) 2026 Danilo M. # # 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. """Run: python3 -m unittest discover -s tests (headless, no VM needed)""" import os os.environ["QT_QPA_PLATFORM"] = "offscreen" # never the real display, even if the session sets one import importlib.machinery # noqa: E402 import importlib.util # noqa: E402 import sys # noqa: E402 import tempfile # noqa: E402 import time # noqa: E402 import unittest # noqa: E402 ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..") sys.path.insert(0, ROOT) from PyQt6.QtCore import QCoreApplication, QEvent, QObject, QProcess, pyqtSignal # noqa: E402 from PyQt6.QtWidgets import QApplication # noqa: E402 from sb_builder import VERSION, config # noqa: E402 from sb_builder import hints as hintlib # noqa: E402 from sb_builder import packages as pk # noqa: E402 from sb_builder.mainwindow import MainWindow # noqa: E402 from sb_builder.remote import DOWN, MISSING, parse # noqa: E402 app = QApplication.instance() or QApplication([]) class ConfigTest(unittest.TestCase): def load(self, text): with tempfile.NamedTemporaryFile("w", suffix=".toml", delete=False) as f: f.write(text) self.addCleanup(os.unlink, f.name) return config.load(f.name) def test_missing_file(self): self.assertEqual(config.load("/nonexistent/config.toml"), (config.DEFAULTS, None)) def test_example_matches_defaults(self): self.assertEqual(config.load(os.path.join(ROOT, "config.example.toml")), (config.DEFAULTS, None)) def test_override(self): cfg, err = self.load("[refresh]\njobs = 2\n") self.assertIsNone(err) self.assertEqual(cfg["refresh"], {"dashboard": 30, "jobs": 2}) def test_unknown_key(self): cfg, err = self.load("[remote]\nbogus = 1\n") self.assertIn("remote.bogus", err) self.assertEqual(cfg, config.DEFAULTS) def test_wrong_type(self): cfg, err = self.load('[vm]\nautostart = "yes"\n') self.assertIn("vm.autostart: expected bool", err) self.assertEqual(cfg, config.DEFAULTS) class ParseTest(unittest.TestCase): def test_cases(self): self.assertEqual(parse(255, b"", b""), (None, DOWN)) self.assertEqual(parse(127, b"", b""), (None, MISSING)) self.assertEqual(parse(0, b'{"a": 1}', b""), ({"a": 1}, None)) self.assertEqual(parse(1, b'{"error": "boom"}', b""), (None, "boom")) self.assertEqual(parse(1, b"", b"bash: x\n"), (None, "bash: x")) class VersionTest(unittest.TestCase): def test_gui_matches_helper(self): script = os.path.join(ROOT, "remote", "sbb-remote") loader = importlib.machinery.SourceFileLoader("sbb_remote", script) sbb = importlib.util.module_from_spec(importlib.util.spec_from_loader("sbb_remote", loader)) loader.exec_module(sbb) self.assertEqual(sbb.VERSION, VERSION) class FakeRemote: def __init__(self, answers): self.answers = answers self.calls = [] self.virsh_answer = (1, "") def call(self, cb, *args, stdin=None): self.calls.append(args if stdin is None else (*args, stdin)) answer = self.answers.get(args[0], (None, DOWN)) cb(*(answer(args, stdin) if callable(answer) else answer)) def virsh(self, cb, *args): self.calls.append(("virsh", *args)) cb(*self.virsh_answer) def follow(self, jid, offset): p = QProcess(app) # parented, like Remote's, so it outlives this call p.start("true", []) return p class FakeNotifier(QObject): clicked = pyqtSignal(str) def __init__(self): super().__init__() self.sent = [] def send(self, title, body, jid=""): self.sent.append((title, body, jid)) STATS = {"load": [0.5, 0.4, 0.3], "uptime": 3600.0, "disk_free": {"repo": 10 * 2**30, "state": None}, "packages": {"total": 10, "personal": 2, "pentesting": 1, "shadowed": 1}, "sync": {"personal": None, "pentesting": 1792351223}, "slackrepo": None} JOB = {"id": "20260101-000000-sbshadow", "kind": "sbshadow", "args": ["--no"], "argv": [], "status": "running", "created": 0, "start": 0, "pid": 5} def inv_row(path, version="1.0", built=None, result=None, hints=None, requires=(), summary=None, maintainer="Some One", info_version=None): return [path, version, info_version or version, None, built, 1788978222 if built else None, result, maintainer, list(requires), summary, "/h/x.hint" if hints else None, hints] COLS = ["path", "version", "info_version", "hint_version", "built_version", "built_time", "result", "maintainer", "requires", "summary", "hint_path", "hints"] INVENTORY = {"columns": COLS, "rows": [ inv_row("misc/nuclei", "3.11.1", built="3.11.0", result="ok", summary="Fast scanner", requires=["go"]), inv_row("personal/foo", built="1.0", result="ok", hints={"DELREQUIRES": "x"}, requires=["x", "y"]), inv_row("pentesting/nmap", built="7.9", result="bad"), inv_row("network/curl"), ]} PREVIEW = {"id": "20260101-000000-slackrepo", "kind": "slackrepo", "status": "ok", "args": ["update", "--preview", "nuclei"], "argv": [], "created": 0, "start": 0, "end": 1} class HintsTest(unittest.TestCase): def view(self, hints, requires=(), info_version="1.0"): return hintlib.interpret(hints, "pkg", info_version, requires) def section(self, view, title): return dict(view["sections"]).get(title) def test_no_hint(self): v = self.view(None, requires=["a"]) self.assertEqual((v["skip"], v["sections"], v["deps"], v["ignored"]), (None, [], [("a", "")], [])) def test_version_arch_and_forced_checksum_skip(self): v = self.view({"VERSION": "2.0", "ARCH": "x86_64", "DOWNLOAD": "http://a/src.tgz", "DOWNLOAD_x86_64": "http://a/bin-x64.tgz http://a/extra.tgz"}) self.assertEqual(self.section(v, "Version override"), [ ("version", "1.0 \u2192 2.0"), ("arch", "x86_64"), ("download", "http://a/bin-x64.tgz"), ("download", "http://a/extra.tgz"), ("md5sum", "not checked")]) def test_arch_variant_empty_falls_back(self): v = self.view({"ARCH": "x86_64", "DOWNLOAD": "http://a/s", "MD5SUM": "abc", "MD5SUM_x86_64": ""}) self.assertEqual(self.section(v, "Version override")[1:], [("download", "http://a/s"), ("md5sum", "abc")]) def test_no_download_forms(self): for hints in ({"DOWNLOAD": "no"}, {"NODOWNLOAD": "yes"}): with self.subTest(hints=hints): self.assertEqual(self.section(self.view(hints), "Version override"), [("download", "no download, the source must already be in the cache")]) self.assertEqual(self.view(hints)["ignored"], []) def test_dependencies(self): v = self.view({"DELREQUIRES": "b %README%", "ADDREQUIRES": "z", "BUILDTIME": "cmake", "PRAGMA": "python3 need_net", "CONFLICTS": "old"}, requires=["a", "b", "%README%"]) self.assertEqual(v["deps"], [("a", ""), ("b", "del"), ("%README%", "del"), ("python3", "pragma"), ("z", "add"), ("cmake", "build")]) self.assertEqual(v["conflicts"], ["old"]) self.assertEqual(self.section(v, "Pragmas")[1], ("need_net", "network not blocked during the build")) def test_skip_forms(self): self.assertIsNone(self.view({"SKIP": "no"})["skip"]) self.assertEqual(self.view({"SKIP": "yes"})["skip"], "always skipped") self.assertEqual(self.view({"SKIP": "remove nc first"})["skip"], "remove nc first") def test_build_settings(self): v = self.view({"OPTIONS": "A=yes B=no", "NUMJOBS": "-j1", "INSTALL": "N", "ANSWER": "ok\\nyes", "NOWARNINGS": "Nonstandard"}) self.assertEqual(self.section(v, "Build settings"), [ ("option", "A=yes"), ("option", "B=no"), ("make jobs", "-j1"), ("install after build", "no"), ("answers to the build", "ok \u23ce yes"), ("hide warnings matching", "Nonstandard")]) self.assertEqual(dict(self.section(self.view({"INSTALL": "1"}), "Build settings")), {"install after build": "yes"}) def test_users_and_groups(self): v = self.view({"GROUPADD": "209:postgres 300", "USERADD": "209:postgres:-d/var/lib/pgsql:-r"}) self.assertEqual(self.section(v, "Host changes"), [ ("group", "postgres (gid 209)"), ("group", "pkg (gid 300)"), ("user", "postgres (uid 209), group postgres, home /var/lib/pgsql, shell /bin/false, -r")]) def test_pragma_forms(self): self.assertEqual(hintlib.pragma_meaning("x86arch=i686"), "on 32 bit x86, build with ARCH=i686") self.assertEqual(hintlib.pragma_meaning("unset_DISTCC_HOSTS"), "unset DISTCC_HOSTS for the build") self.assertEqual(hintlib.pragma_meaning("bogus"), "unknown pragma") def test_ignored_and_shell(self): v = self.view({"ANSWERS": "x", "FOO": "1", "DOWNLOAD_arm": "u", "OPTIONS": {"value": '"X=$(uname -m)"', "shell": True}}) self.assertEqual(v["ignored"], ["ANSWERS", "FOO"]) self.assertEqual(v["shell"], ["OPTIONS"]) def test_cleanup(self): self.assertEqual(self.section(self.view({"CLEANUP": "depmod -a"}), "On uninstall"), [("run", "depmod -a")]) class FilterTest(unittest.TestCase): def setUp(self): self.rows = {r["name"]: r for r in pk.load_rows(INVENTORY)} def test_status(self): self.assertEqual({n: r["status"] for n, r in self.rows.items()}, {"nuclei": "outdated", "foo": "ok", "nmap": "failed", "curl": "never"}) def test_query(self): self.assertEqual(pk.parse_query("Nuc cat:misc bogus:x status:"), [(None, "nuc"), ("category", "misc"), (None, "bogus:x"), (None, "status:")]) def names(self, text="", overlays=(), statuses=()): terms = pk.parse_query(text) return sorted(n for n, r in self.rows.items() if pk.matches(r, terms, overlays, statuses)) def test_matching(self): self.assertEqual(self.names("scanner"), ["nuclei"]) self.assertEqual(self.names("req:go"), ["nuclei"]) self.assertEqual(self.names("maint:some status:o"), ["foo", "nuclei"]) self.assertEqual(self.names(overlays={"personal", "pentesting"}), ["foo", "nmap"]) self.assertEqual(self.names(overlays={"upstream"}, statuses={"never", "failed"}), ["curl"]) self.assertEqual(self.names("cat:misc", statuses={"ok"}), []) class SmokeTest(unittest.TestCase): def window(self, **answers): remote = FakeRemote({"version": ({"version": VERSION, "overlays": ["personal"]}, None), "stats": (STATS, None), "jobs": ({"jobs": []}, None), "gpg-status": ({"unlocked": True, "key": "K"}, None), "inventory": (INVENTORY, None), **answers}) w = MainWindow(config.load("/nonexistent")[0], remote, FakeNotifier()) self.addCleanup(w.deleteLater) self.addCleanup(lambda: w.jobs_tab and w.jobs_tab.stop()) return w, remote def test_tabs_and_dashboard(self): w, _ = self.window() w.poll_jobs() self.assertEqual(w.conn.text(), "● connected") self.assertEqual(w.tabs.tabText(0), "Dashboard") tiles = w.dashboard.tiles self.assertEqual(tiles["Packages"].value.text(), "10") self.assertEqual(tiles["Packages"].detail.text(), "2\u00a0personal · 1\u00a0pentesting · 1\u00a0shadowed") self.assertEqual(tiles["Built"].value.text(), "n/a") self.assertIn("10.0 GiB free", tiles["VM load"].detail.text()) self.assertEqual(w.dashboard.legend.text(), "n/a") jobs = w.open_jobs() self.assertIs(w.open_jobs(), jobs) self.assertEqual(w.tabs.count(), 2) self.assertEqual(jobs.start_menu.actions()[0].text(), "sbpull personal (preview)") def test_end_notification_once(self): w, remote = self.window(jobs=({"jobs": [JOB]}, None)) w.poll_jobs() remote.answers["jobs"] = ({"jobs": [dict(JOB, status="failed", end=5, exit=1)]}, None) w.poll_jobs() w.poll_jobs() self.assertEqual(w.notifier.sent, [("sb-builder: failed", "sbshadow --no", JOB["id"])]) def test_down_offers_vm_start(self): w, remote = self.window(jobs=(None, DOWN)) remote.virsh_answer = (0, "shut off") w.poll_jobs() self.assertIn(("virsh", "domstate", "buildsystem"), remote.calls) self.assertEqual(w.conn.text(), "● VM off") self.assertFalse(w.vm_btn.isHidden()) def test_results_bar(self): w, _ = self.window(stats=(dict(STATS, slackrepo={"results": {"failed": 1, "ok": 3, "removed": 1}, "built": 4}), None)) w.poll_jobs() self.assertEqual([(label, n) for label, n, _ in w.dashboard.bar.parts], [("ok", 3), ("failed", 1), ("removed", 1)]) self.assertEqual(w.dashboard.tiles["Built"].detail.text(), "of 10 · 6 never built") def test_close_jobs_tab_while_following(self): # regression: the follow process ended after the tab was deleted, its # finished slot touched dead widgets and PyQt aborted the app w, remote = self.window(jobs=({"jobs": [JOB]}, None)) procs = QObject() self.addCleanup(procs.deleteLater) def follow(jid, offset): p = QProcess(procs) p.start("sleep", ["5"]) return p remote.follow = follow w.poll_jobs() jobs = w.open_jobs() jobs.select(JOB["id"]) self.assertIsNotNone(jobs.proc) w._close_tab(w.tabs.indexOf(jobs)) QCoreApplication.sendPostedEvents(None, QEvent.Type.DeferredDelete.value) end = time.monotonic() + 1 while time.monotonic() < end: # let the killed process report in app.processEvents() self.assertIsNone(w.jobs_tab) def test_packages_tab(self): w, remote = self.window(**{"job-start": ({"id": "20260101-000001-slackrepo"}, None)}) w.poll_jobs() tab = w.open_packages() self.assertEqual(tab.proxy.rowCount(), 4) tab.chips["failed"].setChecked(True) self.assertEqual(tab.proxy.rowCount(), 1) tab.chips["failed"].setChecked(False) started = [] tab.start_requested.connect(started.append) tab.act("update") self.assertEqual(started, []) # nothing checked, no current row tab.table.setCurrentIndex(tab.proxy.index(1, 1)) self.assertIn("x", tab.detail.text()) tab.act("update") tab.model.checked.update({"nuclei", "curl"}) tab.act("build") tab.update_all() self.assertEqual(started, [["slackrepo", "update", "--preview", "foo"], ["slackrepo", "build", "--preview", "curl", "nuclei"], ["slackrepo", "update", "--preview"]]) def test_run_preview_with_signing(self): w, remote = self.window(jobs=({"jobs": [PREVIEW]}, None), **{"gpg-status": ({"unlocked": False, "key": "K"}, None), "gpg-unlock": lambda args, stdin: ( (None, "bad passphrase") if stdin == b"wrong\n" else ({}, None)), "job-start": ({"id": "20260101-000001-slackrepo"}, None)}) w.poll_jobs() jobs = w.open_jobs() jobs.select(PREVIEW["id"]) self.assertTrue(jobs.run_btn.isVisibleTo(jobs)) answers = iter(["wrong", "right"]) jobs.ask_passphrase = lambda retry: next(answers) jobs.publish_box.setChecked(False) jobs.run_preview() self.assertIn(("gpg-unlock", b"wrong\n"), remote.calls) self.assertIn(("gpg-unlock", b"right\n"), remote.calls) self.assertEqual(remote.calls[-2], ("job-start", "slackrepo", "update", "--no-publish", "nuclei")) def test_signing_cancel_starts_nothing(self): w, remote = self.window(jobs=({"jobs": [PREVIEW]}, None), **{"gpg-status": ({"unlocked": False, "key": "K"}, None)}) w.poll_jobs() jobs = w.open_jobs() jobs.select(PREVIEW["id"]) jobs.ask_passphrase = lambda retry: None jobs.run_preview() self.assertFalse(any(c[0] == "job-start" for c in remote.calls)) def test_key_indicator_and_tile_links(self): w, _ = self.window(**{"gpg-status": ({"unlocked": False, "key": "K"}, None)}) w.poll_jobs() self.assertEqual(w.key.text(), "🔒 key locked") w.dashboard.tiles["Packages"].clicked.emit() self.assertIs(w.tabs.currentWidget(), w.packages_tab) w.dashboard.jobs.clicked.emit() self.assertIs(w.tabs.currentWidget(), w.jobs_tab) def test_log_carriage_return(self): w, _ = self.window() jobs = w.open_jobs() jobs.append_log("10%\r50%\r100%\ndone\r\n") self.assertEqual(jobs.log.toPlainText(), "100%\ndone\n") if __name__ == "__main__": unittest.main() ```` - [ ] **Step 2: Run them, expect failures** Run: `python3 -m unittest tests.test_gui # # 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. """What a slackrepo hint file does, the way slackrepo reads it. Rules from slackrepo's libexec/functions.d/parsefunctions.sh (hint and fixup departments) and slackrepo.hint(5). Pure functions, no Qt. """ import os import re KNOWN = {"SKIP", "VERSION", "OPTIONS", "GROUPADD", "USERADD", "NOWARNING", "NOWARNINGS", "ADDREQUIRES", "DELREQUIRES", "BUILDTIME", "CONFLICTS", "INSTALL", "NUMJOBS", "ANSWER", "CLEANUP", "PRAGMA", "SPECIAL", "ARCH", "DOWNLOAD", "MD5SUM", "SHA256SUM", "NODOWNLOAD"} # NODOWNLOAD: written by mkhint, shown as DOWNLOAD="no" by user decision PER_ARCH = re.compile(r"^(DOWNLOAD|MD5SUM|SHA256SUM)_\w+$") SHELLS = {"sh", "bash", "zsh", "ksh", "csh", "tcsh", "false", "nologin", "true"} PRAGMAS = { "python3": "force python3 support: adds python3 to the deps, sets PYTHON3=yes", "kernelmodule": "rebuild when the host kernel changes; depmod on uninstall", "kernel": "rebuild when the host kernel changes", "kernelversion": "uname reports $KERNEL as the kernel version", "multilib_ldflags": "patch the SlackBuild to set LDFLAGS on a multilib host", "wget": "download with a fake wget user agent", "curl": "download with a fake curl user agent", "download_basename": "symlink downloads to the URL's basename", "download_urldecode": "URL-decode DOWNLOAD values before use", "need_net": "network not blocked during the build", "need_X": "X display not blocked during the build", "noexport_ARCH": "set ARCH by patching the SlackBuild, not via the environment", "noexport_BUILD": "set BUILD by patching the SlackBuild, not via the environment", "noexport_TAG": "set TAG by patching the SlackBuild, not via the environment", "noremove": "do not uninstall packages before building", "nofakeroot": "build as root with sudo instead of fakeroot", "no_make_test": "comment out 'make test' in the SlackBuild", "stubs-32": "provide /usr/include/gnu/stubs-32.h during the build", "abstar": "patch the SlackBuild to use tar --absolute-names", } def _raw(v): return v["value"] if isinstance(v, dict) else v def pragma_meaning(p): if p in PRAGMAS: return PRAGMAS[p] if p.startswith("x86arch="): return f"on 32 bit x86, build with ARCH={p[8:]}" if p.startswith("unset_"): return f"unset {p[6:]} for the build" return "unknown pragma" def parse_groupadd(value, pkg): """[(gid, name)] like slackrepo: numeric field is the gid, other the name.""" out = [] for spec in value.split(): gid, name = None, pkg for field in spec.split(":"): if field[:1].isdigit(): gid = field elif field: name = field out.append((gid, name)) return out def parse_useradd(value, pkg): """[dict(uid, name, group, home, shell, extra)] with slackrepo's defaults.""" out = [] for spec in value.split(): u = {"uid": None, "name": pkg, "group": "", "home": "/dev/null", "shell": "/bin/false", "extra": []} for field in spec.split(":"): if field.startswith("-g"): u["group"] = field[2:] elif field.startswith("-d"): u["home"] = field[2:] elif field.startswith("-s"): u["shell"] = field[2:] elif field.startswith("-"): u["extra"].append(f"{field[:2]} {field[2:]}".strip()) elif field.startswith("/"): # ponytail: slackrepo tests -x on the build host; guess by name instead key = "shell" if os.path.basename(field) in SHELLS else "home" u[key] = field elif field[:1].isdigit(): u["uid"] = field elif field: u["name"] = field u["group"] = u["group"] or u["name"] out.append(u) return out def dependencies(hints, requires): """[(dep, mark)] as slackrepo computes them; mark in '', del, pragma, add, build.""" deleted = set(hints.get("DELREQUIRES", "").split()) | {"%README%"} out = [(r, "del" if r in deleted else "") for r in requires] if "python3" in hints.get("PRAGMA", hints.get("SPECIAL", "")).split(): out.append(("python3", "pragma")) out += [(r, "add") for r in hints.get("ADDREQUIRES", "").split()] out += [(r, "build") for r in hints.get("BUILDTIME", "").split()] return out def interpret(hints, pkg, info_version=None, requires=()): """Structured view of one package's hints. hints: {key: value or {"value": raw, "shell": True}} from the helper, or None. Returns {"skip", "sections": [(title, [(label, value)])], "deps", "conflicts", "ignored", "shell"}. """ raw = hints or {} shell = sorted(k for k, v in raw.items() if isinstance(v, dict)) h = {k: _raw(v) for k, v in raw.items()} view = {"skip": None, "sections": [], "deps": dependencies(h, list(requires)), "conflicts": h.get("CONFLICTS", "").split(), "shell": shell, "ignored": sorted(k for k in h if k not in KNOWN and not PER_ARCH.match(k))} skip = h.get("SKIP") if skip is not None and skip.lower() != "no": view["skip"] = "always skipped" if skip.lower() == "yes" else skip # version, downloads and checksums, resolved together as slackrepo does arch = h.get("ARCH") resolved = {} for var in ("DOWNLOAD", "MD5SUM", "SHA256SUM"): resolved[var] = h.get(f"{var}_{arch}") if arch and h.get(f"{var}_{arch}") else h.get(var) if h.get("VERSION"): resolved["MD5SUM"] = resolved["MD5SUM"] or "no" resolved["SHA256SUM"] = resolved["SHA256SUM"] or "no" rows = [] if h.get("VERSION"): rows.append(("version", f"{info_version or '?'} → {h['VERSION']}")) if arch is not None: rows.append(("arch", arch or "(empty)")) if resolved["DOWNLOAD"] == "no" or h.get("NODOWNLOAD", "").lower() == "yes": rows.append(("download", "no download, the source must already be in the cache")) elif resolved["DOWNLOAD"]: rows += [("download", url) for url in resolved["DOWNLOAD"].split()] if resolved["MD5SUM"] == "no": rows.append(("md5sum", "not checked")) elif resolved["MD5SUM"]: rows.append(("md5sum", resolved["MD5SUM"])) if any(k.startswith("SHA256SUM") for k in h): rows.append(("sha256sum", "not checked" if resolved["SHA256SUM"] == "no" else resolved["SHA256SUM"])) if rows: view["sections"].append(("Version override", rows)) rows = [("option", o) for o in h.get("OPTIONS", "").split()] if "NUMJOBS" in h: rows.append(("make jobs", h["NUMJOBS"])) if h.get("INSTALL"): rows.append(("install after build", "no" if h["INSTALL"][:1] in "Nn0" else "yes")) if "ANSWER" in h: rows.append(("answers to the build", h["ANSWER"].replace("\\n", " ⏎ "))) nowarning = h.get("NOWARNING") or h.get("NOWARNINGS") if nowarning: rows.append(("hide warnings matching", nowarning)) if rows: view["sections"].append(("Build settings", rows)) pragmas = h.get("PRAGMA") or h.get("SPECIAL") or "" if pragmas: view["sections"].append(("Pragmas", [(p, pragma_meaning(p)) for p in pragmas.split()])) rows = [("group", f"{name} (gid {gid or '?'})") for gid, name in parse_groupadd(h.get("GROUPADD", ""), pkg)] for u in parse_useradd(h.get("USERADD", ""), pkg): extra = f", {' '.join(u['extra'])}" if u["extra"] else "" rows.append(("user", f"{u['name']} (uid {u['uid'] or '?'}), group {u['group']}, " f"home {u['home']}, shell {u['shell']}{extra}")) if rows: view["sections"].append(("Host changes", rows)) if h.get("CLEANUP"): view["sections"].append(("On uninstall", [("run", h["CLEANUP"])])) return view ```` - [ ] **Step 4: Packages tab** (create `sb_builder/packages.py`): ````python # sb-builder - Packages tab # Copyright (C) 2026 Danilo M. # # 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. """Packages tab: search the inventory, read a package's hints, start slackrepo previews.""" import html from PyQt6.QtCore import QAbstractTableModel, QModelIndex, QSortFilterProxyModel, Qt, pyqtSignal from PyQt6.QtGui import QColor from PyQt6.QtWidgets import (QHBoxLayout, QLabel, QLineEdit, QMessageBox, QPushButton, QScrollArea, QSplitter, QTableView, QVBoxLayout, QWidget) from . import hints as hintlib from .dashboard import STATUS from .fmt import NA, fmt_time OVERLAYS = ("personal", "pentesting") STATUSES = ("ok", "failed", "never", "outdated") COLUMNS = ("", "name", "category", "version", "built", "status", "maintainer") STATUS_COLOR = {"ok": STATUS["ok"], "failed": STATUS["failed"], "outdated": STATUS["aborted"]} FIELDS = {"cat": "category", "maint": "maintainer", "req": "requires", "status": "status"} def status(row): # ponytail: outdated compares versions only; slackrepo also rebuilds on # SlackBuild or dependency changes, which its preview shows if not row["built_version"] and not row["result"]: return "never" if row["result"] in ("failed", "bad", "aborted"): return "failed" if row["built_version"] != row["version"]: return "outdated" return "ok" def overlay(row): cat = row["path"].split("/", 1)[0] return cat if cat in OVERLAYS else "upstream" def parse_query(text): """[(field or None, term)], lower-cased; unknown prefixes count as plain text.""" terms = [] for word in text.lower().split(): field, sep, term = word.partition(":") if sep and field in FIELDS and term: terms.append((FIELDS[field], term)) else: terms.append((None, word)) return terms def matches(row, terms, overlays=(), statuses=()): """All terms match, and the row is in a toggled chip of each non-empty group.""" if overlays and overlay(row) not in overlays: return False if statuses and row["status"] not in statuses: return False for field, term in terms: if field is None: hay = f"{row['name']} {row['summary'] or ''}" elif field == "requires": hay = " ".join(row["requires"]) else: hay = row[field] or "" if term not in hay.lower(): return False return True def load_rows(inventory): rows = [] for values in inventory["rows"]: row = dict(zip(inventory["columns"], values)) row["category"], row["name"] = row["path"].split("/", 1) row["status"] = status(row) rows.append(row) return rows class Model(QAbstractTableModel): def __init__(self): super().__init__() self.rows = [] self.checked = set() # package names def set_rows(self, rows): self.beginResetModel() self.rows = rows names = {r["name"] for r in rows} self.checked &= names self.endResetModel() def rowCount(self, parent=QModelIndex()): return 0 if parent.isValid() else len(self.rows) def columnCount(self, parent=QModelIndex()): return len(COLUMNS) def headerData(self, section, orientation, role=Qt.ItemDataRole.DisplayRole): if orientation == Qt.Orientation.Horizontal and role == Qt.ItemDataRole.DisplayRole: return COLUMNS[section] return None def flags(self, index): f = Qt.ItemFlag.ItemIsEnabled | Qt.ItemFlag.ItemIsSelectable return f | Qt.ItemFlag.ItemIsUserCheckable if index.column() == 0 else f def data(self, index, role=Qt.ItemDataRole.DisplayRole): row, col = self.rows[index.row()], index.column() if role == Qt.ItemDataRole.CheckStateRole and col == 0: return Qt.CheckState.Checked if row["name"] in self.checked else Qt.CheckState.Unchecked if role == Qt.ItemDataRole.DisplayRole: return (None, row["name"], row["category"], row["version"], row["built_version"] or "", row["status"], row["maintainer"] or "")[col] if role == Qt.ItemDataRole.ForegroundRole and col == 5 and row["status"] in STATUS_COLOR: return QColor(STATUS_COLOR[row["status"]]) if role == Qt.ItemDataRole.UserRole: return row return None def setData(self, index, value, role=Qt.ItemDataRole.EditRole): if role != Qt.ItemDataRole.CheckStateRole or index.column() != 0: return False name = self.rows[index.row()]["name"] if Qt.CheckState(value) == Qt.CheckState.Checked: self.checked.add(name) else: self.checked.discard(name) self.dataChanged.emit(index, index) return True class Filter(QSortFilterProxyModel): def __init__(self): super().__init__() self.terms, self.overlays, self.statuses = [], set(), set() def update(self, text, overlays, statuses): self.terms, self.overlays, self.statuses = parse_query(text), overlays, statuses self.invalidateFilter() def filterAcceptsRow(self, source_row, _parent): row = self.sourceModel().rows[source_row] return matches(row, self.terms, self.overlays, self.statuses) def chip(text): b = QPushButton(text, checkable=True) b.setStyleSheet("QPushButton { border: 1px solid palette(mid); border-radius: 10px;" " padding: 2px 10px; }" "QPushButton:checked { background: palette(highlight);" " color: palette(highlighted-text); }") return b def detail_html(row): """The detail pane for one package: facts, dependencies, then its hints.""" view = hintlib.interpret(row["hints"], row["name"], row["info_version"], row["requires"]) e = html.escape out = [f"

{e(row['name'])}

{e(row['path'])}" f"{' · ' + e(row['summary']) if row['summary'] else ''}

"] if view["skip"]: out.append(f'

⚠ Skipped by slackrepo: ' f"{e(view['skip'])}

") facts = [(".info version", row["info_version"]), ("hint version", row["hint_version"]), ("built version", row["built_version"]), ("built", fmt_time(row["built_time"]) if row["built_time"] else None), ("last result", row["result"]), ("maintainer", row["maintainer"])] out.append(table([(k, v or NA) for k, v in facts])) marks = {"": "", "del": "{} (DELREQUIRES)", "add": "+{} (ADDREQUIRES)", "build": "+{} (build time only)", "pragma": "+{} (pragma)"} deps = [(marks[m] or "{}").format(e(d)) for d, m in view["deps"]] out.append(f"

Dependencies

{'   '.join(deps) or 'none'}

") if view["conflicts"]: out.append(f"

Conflicts: {e(' '.join(view['conflicts']))}

") for title, rows in view["sections"]: out.append(f"

{e(title)}

" + table(rows)) if view["shell"]: out.append(f"

Expanded by the shell at build time, shown as written: " f"{e(', '.join(view['shell']))}

") if view["ignored"]: out.append(f'

Not read by slackrepo: ' f"{e(', '.join(view['ignored']))}

") if row["hint_path"]: out.append(f"

hint: {e(row['hint_path'])}

") return "".join(out) def table(rows): cells = "".join(f"{html.escape(str(k))}" f"{html.escape(str(v))}" for k, v in rows) return f"{cells}
" class PackagesTab(QWidget): start_requested = pyqtSignal(list) # job-start arguments def __init__(self, remote, parent=None): super().__init__(parent) self.remote = remote self.model = Model() self.proxy = Filter() self.proxy.setSourceModel(self.model) self.query = QLineEdit(placeholderText="search (cat: maint: req: status:)") self.query.textChanged.connect(self.refilter) self.chips = {} chips = QHBoxLayout() for group in (("personal", "pentesting", "upstream"), STATUSES): for name in group: self.chips[name] = chip(name) self.chips[name].toggled.connect(self.refilter) chips.addWidget(self.chips[name]) chips.addSpacing(16) chips.addStretch() self.refresh_btn = QPushButton("Refresh") self.refresh_btn.clicked.connect(self.reload) chips.addWidget(self.refresh_btn) self.table = QTableView(sortingEnabled=True) self.table.setModel(self.proxy) self.table.setSelectionBehavior(QTableView.SelectionBehavior.SelectRows) self.table.setSelectionMode(QTableView.SelectionMode.SingleSelection) self.table.verticalHeader().hide() self.table.horizontalHeader().setStretchLastSection(True) for col, width in enumerate((28, 220, 110, 120, 120, 80)): self.table.setColumnWidth(col, width) self.table.selectionModel().currentRowChanged.connect(self.show_detail) self.detail = QLabel(textFormat=Qt.TextFormat.RichText, wordWrap=True, alignment=Qt.AlignmentFlag.AlignTop) self.detail.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse) self.detail.setContentsMargins(12, 8, 12, 8) scroll = QScrollArea(widgetResizable=True) scroll.setWidget(self.detail) split = QSplitter() split.addWidget(self.table) split.addWidget(scroll) split.setSizes([700, 400]) self.count = QLabel() bar = QHBoxLayout() bar.addWidget(self.count) for label, slot in (("Select visible", self.select_visible), ("Clear", self.clear)): b = QPushButton(label) b.clicked.connect(slot) bar.addWidget(b) bar.addStretch() for action in ("build", "update", "rebuild", "remove"): b = QPushButton(action.capitalize()) b.clicked.connect(lambda _c, a=action: self.act(a)) bar.addWidget(b) for label, slot in (("Update all", self.update_all), ("Publish", self.publish)): b = QPushButton(label) b.clicked.connect(slot) bar.addWidget(b) layout = QVBoxLayout(self) layout.addWidget(self.query) layout.addLayout(chips) layout.addWidget(split) layout.addLayout(bar) self.model.dataChanged.connect(lambda *_: self.update_count()) self.update_count() def reload(self): self.count.setText("loading…") self.remote.call(self._loaded, "inventory") def _loaded(self, data, err): if err: self.count.setText(f"inventory failed: {err}") return self.model.set_rows(load_rows(data)) self.update_count() def refilter(self): on = {n for n, b in self.chips.items() if b.isChecked()} self.proxy.update(self.query.text(), on - set(STATUSES), on & set(STATUSES)) self.update_count() def update_count(self): self.count.setText(f"{self.proxy.rowCount():,} shown · {len(self.model.checked)} selected") def select_visible(self): for i in range(self.proxy.rowCount()): self.model.checked.add(self.proxy.index(i, 0).data(Qt.ItemDataRole.UserRole)["name"]) self.model.layoutChanged.emit() self.update_count() def clear(self): self.model.checked.clear() self.model.layoutChanged.emit() self.update_count() def targets(self): """Checked packages, else the current row; sorted names.""" if self.model.checked: return sorted(self.model.checked) current = self.table.currentIndex() return [current.data(Qt.ItemDataRole.UserRole)["name"]] if current.isValid() else [] def act(self, action): names = self.targets() if names: self.start_requested.emit(["slackrepo", action, "--preview", *names]) def update_all(self): self.start_requested.emit(["slackrepo", "update", "--preview"]) def publish(self): if QMessageBox.question(self, "Publish", "Push the package repository to the mirror now?") \ == QMessageBox.StandardButton.Yes: self.start_requested.emit(["publish"]) def show_detail(self, current, _previous): row = current.data(Qt.ItemDataRole.UserRole) if current.isValid() else None self.detail.setText(detail_html(row) if row else "") ```` - [ ] **Step 5: Version, config, transport** Replace `sb_builder/__init__.py`: ````python # sb-builder - GUI package # Copyright (C) 2026 Danilo M. # # 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. """sb-builder: Qt6 GUI driving the SlackBuild tooling on a build VM.""" VERSION = "0.2.0" # must match remote/sbb-remote VERSION ```` Replace `sb_builder/config.py`: ````python # sb-builder - GUI configuration # Copyright (C) 2026 Danilo M. # # 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. """GUI configuration: built-in defaults overlaid by the user's TOML file.""" import copy import os import tomllib DEFAULTS = { "remote": {"target": "buildsystem", "ssh_options": []}, "refresh": {"dashboard": 30, "jobs": 5}, "notify": {"enabled": True, "events": ["finished", "failed", "killed", "lost", "vm-up"]}, "vm": {"libvirt_uri": "qemu:///system", "domain": "buildsystem", "autostart": False}, "builds": {"publish": True}, } def path(): base = os.environ.get("XDG_CONFIG_HOME") or os.path.expanduser("~/.config") return os.path.join(base, "sb-builder", "config.toml") def load(p=None): """Return (config, error). On any problem: the defaults plus the error text.""" p = p or path() try: with open(p, "rb") as f: data = tomllib.load(f) except FileNotFoundError: return copy.deepcopy(DEFAULTS), None except (OSError, tomllib.TOMLDecodeError) as e: return copy.deepcopy(DEFAULTS), f"{p}: {e}" cfg = copy.deepcopy(DEFAULTS) for sec, keys in data.items(): if sec not in DEFAULTS or not isinstance(keys, dict): return copy.deepcopy(DEFAULTS), f"{p}: unknown section [{sec}]" for k, v in keys.items(): if k not in DEFAULTS[sec]: return copy.deepcopy(DEFAULTS), f"{p}: unknown key {sec}.{k}" want = type(DEFAULTS[sec][k]) if type(v) is not want: return copy.deepcopy(DEFAULTS), f"{p}: {sec}.{k}: expected {want.__name__}" cfg[sec][k] = v return cfg, None ```` Replace `config.example.toml`: ````toml # sb-builder GUI config: ~/.config/sb-builder/config.toml # Every key is optional; these are the defaults. [remote] target = "buildsystem" # any ssh destination: host alias, user@host ssh_options = [] # extra ssh args, e.g. ["-p", "2222"] [refresh] dashboard = 30 # seconds, while the Dashboard tab is visible jobs = 5 # seconds; doubles up to 30 while the VM is down [notify] enabled = true events = ["finished", "failed", "killed", "lost", "vm-up"] [vm] libvirt_uri = "qemu:///system" domain = "buildsystem" # libvirt domain of the VM; "" disables VM control autostart = false # true: start the VM without asking when it is off [builds] publish = true # initial state of the "publish" box next to Run it ```` Replace `sb_builder/remote.py`: ````python # sb-builder - ssh and virsh transport # Copyright (C) 2026 Danilo M. # # 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. """The only code that talks to the VM: ssh to sbb-remote, and local virsh.""" import json import os import shlex from PyQt6.QtCore import QObject, QProcess, QStandardPaths DOWN = "ssh: connection failed" MISSING = "sbb-remote is not installed on the VM" def parse(rc, out, err): """Map one sbb-remote call to (data, error).""" if rc == 255: return None, DOWN if rc == 127: return None, MISSING try: data = json.loads(out) except ValueError: return None, err.decode(errors="replace").strip() or f"exit {rc}" if rc != 0 or "error" in data: return None, data.get("error", f"exit {rc}") return data, None class Remote(QObject): def __init__(self, cfg, parent=None): super().__init__(parent) self.cfg = cfg # shared dict, updated in place on config reload run = QStandardPaths.writableLocation(QStandardPaths.StandardLocation.RuntimeLocation) self.control = os.path.join(run or "/tmp", "sbb-%C") def ssh_args(self, *args): r = self.cfg["remote"] # ssh joins its args into a remote shell command: quote every one return ["-o", "ControlMaster=auto", "-o", f"ControlPath={self.control}", "-o", "ControlPersist=10m", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5", *r["ssh_options"], r["target"], "sbb-remote", *map(shlex.quote, args)] def call(self, cb, *args, stdin=None): """Run sbb-remote , then cb(data, error). stdin: bytes for the helper.""" self._run("ssh", self.ssh_args(*args), lambda rc, out, err: cb(*parse(rc, out, err)), stdin) def virsh(self, cb, *args): """Run local virsh against the configured libvirt URI, then cb(rc, text).""" argv = ["-c", self.cfg["vm"]["libvirt_uri"], *args] self._run("virsh", argv, lambda rc, out, err: cb(rc, (out or err).decode(errors="replace").strip())) def follow(self, jid, offset): """Start job-follow; the caller wires readyReadStandardOutput and finished.""" p = QProcess(self) p.start("ssh", self.ssh_args("job-follow", jid, str(offset))) return p def _run(self, prog, argv, cb, stdin=None): p = QProcess(self) def done(rc, _status=None): cb(rc, bytes(p.readAllStandardOutput()), bytes(p.readAllStandardError())) p.deleteLater() p.finished.connect(done) p.errorOccurred.connect( lambda e: e == QProcess.ProcessError.FailedToStart and done(-1)) p.start(prog, argv) if stdin is not None: p.write(stdin) p.closeWriteChannel() ```` - [ ] **Step 6: Jobs tab, Dashboard, main window** Replace `sb_builder/jobs.py`: ````python # sb-builder - Jobs tab # Copyright (C) 2026 Danilo M. # # 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. """Jobs tab: job list, live log through job-follow, start and kill.""" import codecs import re from PyQt6.QtCore import Qt, pyqtSignal from PyQt6.QtGui import QFontDatabase, QTextCursor from PyQt6.QtWidgets import (QCheckBox, QHBoxLayout, QInputDialog, QLineEdit, QListWidget, QListWidgetItem, QMenu, QMessageBox, QPlainTextEdit, QPushButton, QSplitter, QToolButton, QVBoxLayout, QWidget) from .fmt import describe ANSI = re.compile(rb"\x1b\[[0-9;?]*[A-Za-z]") SHADOW_PLAN = re.compile(r"^(un)?shadow: ", re.M) ACTIVE = ("starting", "running") BAD = ("failed", "killed", "lost") class JobsTab(QWidget): refresh_requested = pyqtSignal() def __init__(self, remote, publish=True, parent=None): super().__init__(parent) self.remote = remote self.pending = () self.jobs = {} self.current = None self.offset = 0 self.proc = None self.decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") self.start_menu = QMenu(self) start = QToolButton(text="Start", popupMode=QToolButton.ToolButtonPopupMode.InstantPopup) start.setMenu(self.start_menu) self.apply_btn = QPushButton("Apply shadowing") self.apply_btn.clicked.connect(lambda: self.confirm_start( "Apply the shadowing plan to the repo (sbshadow --yes)?", "sbshadow", "--yes")) self.apply_btn.hide() self.publish_box = QCheckBox("publish", checked=publish) self.publish_box.setToolTip("Push to the mirror when the build ends; " "unticked, use Publish later") self.run_btn = QPushButton("Run it") self.run_btn.clicked.connect(self.run_preview) self.kill_btn = QPushButton("Kill") self.kill_btn.clicked.connect(self.kill) bar = QHBoxLayout() for w in (start, self.apply_btn, self.run_btn, self.publish_box): bar.addWidget(w) bar.addStretch() bar.addWidget(self.kill_btn) self.list = QListWidget() self.list.currentItemChanged.connect(self._on_select) self.log = QPlainTextEdit(readOnly=True) self.log.setFont(QFontDatabase.systemFont(QFontDatabase.SystemFont.FixedFont)) self.log.setMaximumBlockCount(50000) # ponytail: older lines drop off; the full log stays on the VM split = QSplitter() split.addWidget(self.list) split.addWidget(self.log) split.setSizes([300, 700]) layout = QVBoxLayout(self) layout.addLayout(bar) layout.addWidget(split) self.set_overlays([]) def set_overlays(self, overlays): m = self.start_menu m.clear() for o in overlays: m.addAction(f"sbpull {o} (preview)", lambda o=o: self.start("sbpull", o, "--no")) m.addSeparator() m.addAction("sbshadow (preview)", lambda: self.start("sbshadow", "--no")) m.addAction("sbshadow (apply)…", lambda: self.confirm_start( "Run sbshadow --yes on the repo?", "sbshadow", "--yes")) def set_jobs(self, jobs): self.jobs = {j["id"]: j for j in jobs} self.list.blockSignals(True) self.list.clear() for j in jobs: item = QListWidgetItem(describe(j)) item.setData(Qt.ItemDataRole.UserRole, j["id"]) if j["status"] in BAD: item.setForeground(Qt.GlobalColor.red) self.list.addItem(item) if j["id"] == self.current: self.list.setCurrentItem(item) self.list.blockSignals(False) j = self.jobs.get(self.current) if j and j["status"] in ACTIVE and self.proc is None: self._follow() # stream dropped while the job kept running: resume at offset self._update_buttons() def select(self, jid): for i in range(self.list.count()): if self.list.item(i).data(Qt.ItemDataRole.UserRole) == jid: self.list.setCurrentItem(self.list.item(i)) return self._show(jid) # not listed yet: just started def append_log(self, text): cur = self.log.textCursor() cur.movePosition(QTextCursor.MoveOperation.End) # ponytail: a \r split across two chunks is printed as-is for i, part in enumerate(text.replace("\r\n", "\n").split("\r")): if i: # carriage return: the next text replaces the current line cur.movePosition(QTextCursor.MoveOperation.StartOfBlock, QTextCursor.MoveMode.KeepAnchor) cur.removeSelectedText() cur.insertText(part) self.log.setTextCursor(cur) def stop(self): if self.proc: p, self.proc = self.proc, None # disconnect first: finished arrives later, maybe after this tab is deleted p.readyReadStandardOutput.disconnect() p.finished.disconnect() p.kill() p.deleteLater() def start(self, *args): self.pending = args self.remote.call(self._started, "job-start", *args) def confirm_start(self, question, *args): if QMessageBox.question(self, "Confirm", question) == QMessageBox.StandardButton.Yes: self.start(*args) def kill(self): if QMessageBox.question(self, "Kill job", f"Send SIGTERM to {self.current}?") \ == QMessageBox.StandardButton.Yes: self.remote.call(lambda _d, err: err and QMessageBox.warning(self, "Kill failed", err), "job-kill", self.current) def run_preview(self): """Start the previewed slackrepo run for real.""" args = [a for a in self.jobs[self.current]["args"] if a != "--preview"] if not self.publish_box.isChecked(): args.insert(1, "--no-publish") self.start_signed("slackrepo", *args) def ask_passphrase(self, retry): text, ok = QInputDialog.getText( self, "Signing key", ("Wrong passphrase, try again." if retry else "Passphrase of the package signing key:"), QLineEdit.EchoMode.Password) return text if ok else None def start_signed(self, *args, retry=False): """Unlock the signing key first when it is locked, then start the job.""" def status(data, err): if err: QMessageBox.warning(self, "Signing key", err) elif data["unlocked"] is False: self._unlock(args, retry) else: self.start(*args) self.remote.call(status, "gpg-status") def _unlock(self, args, retry): passphrase = self.ask_passphrase(retry) if passphrase is None: return def unlocked(_data, err): if err == "bad passphrase": self._unlock(args, True) elif err: QMessageBox.warning(self, "Signing key", err) else: self.start(*args) self.remote.call(unlocked, "gpg-unlock", stdin=passphrase.encode() + b"\n") def _started(self, data, err): if err == "signing key locked": self.start_signed(*self.pending) # locked since the last check: ask again return if err: QMessageBox.warning(self, "Start failed", err) return if "busy" in data: box = QMessageBox(QMessageBox.Icon.Information, "Busy", f"busy: {data['busy']}", parent=self) show = box.addButton("Show job", QMessageBox.ButtonRole.AcceptRole) box.addButton(QMessageBox.StandardButton.Close) box.exec() if box.clickedButton() is show and data["busy"]: self.select(data["busy"]) return self.select(data["id"]) self.refresh_requested.emit() def _on_select(self, item, _prev): jid = item.data(Qt.ItemDataRole.UserRole) if item else None if jid != self.current: self._show(jid) def _show(self, jid): self.stop() self.current, self.offset = jid, 0 self.decoder.reset() self.log.clear() if jid: self._follow() self._update_buttons() def _follow(self): jid = self.current p = self.proc = self.remote.follow(jid, self.offset) p.readyReadStandardOutput.connect(lambda: self._chunk(p, jid)) p.finished.connect(lambda *_: self._ended(p)) def _chunk(self, p, jid): data = bytes(p.readAllStandardOutput()) if jid != self.current: return self.offset += len(data) self.append_log(self.decoder.decode(ANSI.sub(b"", data))) def _ended(self, p): if p is self.proc: self.proc = None p.deleteLater() self._update_buttons() def _update_buttons(self): j = self.jobs.get(self.current) self.kill_btn.setEnabled(bool(j) and j["status"] == "running") self.apply_btn.setVisible( bool(j) and j["status"] == "ok" and "--no" in j["args"] and bool(SHADOW_PLAN.search(self.log.toPlainText()))) preview = bool(j) and j["kind"] == "slackrepo" and j["status"] == "ok" \ and "--preview" in j["args"] self.run_btn.setVisible(preview) self.publish_box.setVisible(preview) ```` Replace `sb_builder/dashboard.py`: ````python # sb-builder - Dashboard tab # Copyright (C) 2026 Danilo M. # # 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. """Dashboard tab: KPI tiles, slackrepo results bar, recent jobs, from stats and jobs.""" import html import time from PyQt6.QtCore import QEvent, QRectF, Qt, pyqtSignal from PyQt6.QtGui import QColor, QFont, QPainter, QPalette from PyQt6.QtWidgets import (QFrame, QGridLayout, QHBoxLayout, QLabel, QPushButton, QToolTip, QVBoxLayout, QWidget) from .fmt import ICON, NA, fmt_age, fmt_bytes, fmt_duration, fmt_time, job_title # reserved status colors (dataviz reference palette); always shown with a text label STATUS = {"ok": "#0ca30c", "aborted": "#fab219", "bad": "#ec835a", "failed": "#d03b3b", "killed": "#fab219", "lost": "#ec835a"} GAP = 2 # px of surface between bar segments def num(n): return NA if n is None else f"{n:,}" def muted(label): label.setForegroundRole(QPalette.ColorRole.PlaceholderText) return label class Card(QFrame): """Rounded panel with a muted title; content goes into self.body.""" clicked = pyqtSignal() def __init__(self, title): super().__init__(objectName="card") self.body = QVBoxLayout(self) self.body.setContentsMargins(16, 12, 16, 14) self.body.addWidget(muted(QLabel(title))) def link(self, slot): """Make the whole card open something.""" self.clicked.connect(slot) self.setCursor(Qt.CursorShape.PointingHandCursor) def mouseReleaseEvent(self, event): if event.button() == Qt.MouseButton.LeftButton: self.clicked.emit() super().mouseReleaseEvent(event) class Tile(Card): """Stat tile: title, one large value, a muted detail line.""" def __init__(self, title): super().__init__(title) self.value = QLabel(NA) f = self.value.font() f.setPointSizeF(f.pointSizeF() * 2.2) f.setWeight(QFont.Weight.DemiBold) self.value.setFont(f) self.detail = muted(QLabel()) self.detail.setWordWrap(True) self.body.addWidget(self.value) self.body.addWidget(self.detail) self.body.addStretch() def set(self, value, detail=""): self.value.setText(value) self.detail.setText(detail) class StackedBar(QWidget): """Horizontal part-to-whole bar; hover a segment for its label and count.""" def __init__(self): super().__init__() self.parts = [] # (label, count, QColor) self.setFixedHeight(14) def set_parts(self, parts): self.parts = [p for p in parts if p[1]] self.update() def rects(self): total = sum(c for _, c, _ in self.parts) if not total: return [] room = self.width() - GAP * (len(self.parts) - 1) out, x = [], 0.0 for label, count, color in self.parts: w = max(2.0, room * count / total) out.append((QRectF(x, 0, w, self.height()), label, count, color)) x += w + GAP return out def paintEvent(self, _event): p = QPainter(self) p.setRenderHint(QPainter.RenderHint.Antialiasing) p.setPen(Qt.PenStyle.NoPen) for rect, _label, _count, color in self.rects(): p.setBrush(color) p.drawRoundedRect(rect, 4, 4) def event(self, e): if e.type() == QEvent.Type.ToolTip: hit = next((f"{label}: {count:,}" for rect, label, count, _ in self.rects() if rect.contains(e.position())), None) if hit: QToolTip.showText(e.globalPosition().toPoint(), hit, self) else: QToolTip.hideText() return True return super().event(e) class Dashboard(QWidget): def __init__(self, parent=None): super().__init__(parent) self.setStyleSheet("QFrame#card { background: palette(base);" " border: 1px solid palette(mid); border-radius: 10px; }") self.tiles = {name: Tile(name) for name in ("Packages", "Built", "Last pull", "VM load")} self.results = Card("slackrepo results") self.bar = StackedBar() self.legend = QLabel(NA) self.legend.setTextFormat(Qt.TextFormat.RichText) self.legend.setWordWrap(True) self.results.body.addWidget(self.bar) self.results.body.addWidget(self.legend) self.results.body.addStretch() self.orphans = Card("SBo orphans") self.orphans.body.addWidget(muted(QLabel("coming in phase 5"))) self.orphans.body.addStretch() self.jobs = Card("Recent jobs") self.jobs_list = QLabel("no jobs yet") self.jobs_list.setTextFormat(Qt.TextFormat.RichText) self.jobs.body.addWidget(self.jobs_list) self.jobs.body.addStretch() self.stale = muted(QLabel()) self.refresh_btn = QPushButton("Refresh") top = QHBoxLayout() top.addWidget(self.stale) top.addStretch() top.addWidget(self.refresh_btn) grid = QGridLayout() grid.setSpacing(12) for i, tile in enumerate(self.tiles.values()): grid.addWidget(tile, 0, i) grid.addWidget(self.results, 1, 0, 1, 3) grid.addWidget(self.orphans, 1, 3) grid.addWidget(self.jobs, 2, 0, 1, 4) grid.setRowStretch(2, 1) layout = QVBoxLayout(self) layout.setContentsMargins(16, 12, 16, 16) layout.addLayout(top) layout.addLayout(grid) def cards(self): return [*self.tiles.values(), self.results, self.orphans, self.jobs] def show_stats(self, s): p, sr, sync = s.get("packages"), s.get("slackrepo"), s.get("sync") or {} t = self.tiles if p is None: t["Packages"].set(NA) else: extra = " · ".join(f"{num(v)}\u00a0{k}" for k, v in p.items() if k != "total") t["Packages"].set(num(p["total"]), extra) if sr is None: t["Built"].set(NA, "no slackrepo database") else: never = "" if p is None else f" · {num(p['total'] - sr['built'])} never built" t["Built"].set(num(sr["built"]), f"of {num(p and p['total'])}{never}") known = [ts for ts in sync.values() if ts is not None] t["Last pull"].set(fmt_age(min(known)) if known else NA, " · ".join(f"{o} {fmt_duration(None if ts is None else time.time() - ts)}" for o, ts in sync.items())) load, disk = s.get("load"), s.get("disk_free") or {} t["VM load"].set(NA if load is None else f"{load[0]:.2f}", f"up {fmt_duration(s.get('uptime'))} · {fmt_bytes(disk.get('repo'))} free") self._show_results(sr) def _show_results(self, sr): if sr is None: self.bar.set_parts([]) self.legend.setText(NA) return gray = self.palette().color(QPalette.ColorRole.PlaceholderText) # Mid can match the background order = sorted(sr["results"].items(), key=lambda kv: (list(STATUS).index(kv[0]) if kv[0] in STATUS else 99, kv[0])) self.bar.set_parts([(k, v, QColor(STATUS.get(k)) if k in STATUS else gray) for k, v in order]) self.legend.setText("   ".join( f'■ {html.escape(k)} {num(v)}' for k, v in order)) def show_jobs(self, jobs): rows = [] for j in jobs[:10]: color = STATUS.get(j["status"]) icon = ICON.get(j["status"], "?") if color: icon = f'{icon}' dur = None if "start" not in j else j.get("end", time.time()) - j["start"] rows.append(f"{icon} {html.escape(job_title(j))}  " f"{j['status']}  " f"{fmt_time(j.get('start', j['created']))}  " f"{fmt_duration(dur)}") self.jobs_list.setText(f"{''.join(rows)}
" if rows else "no jobs yet") def set_stale(self, since): """Grey the cards when the VM stopped answering; None clears it.""" for card in self.cards(): card.setEnabled(since is None) self.stale.setText("" if since is None else f"stale since {time.strftime('%H:%M', time.localtime(since))}") ```` Replace `sb_builder/mainwindow.py`: ````python # sb-builder - main window # Copyright (C) 2026 Danilo M. # # 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. """Main window: tabs, status bar, connection polling, VM control, notifications.""" import time from PyQt6.QtCore import QMetaType, QObject, QTimer, pyqtSignal, pyqtSlot from PyQt6.QtDBus import QDBusArgument, QDBusConnection, QDBusInterface, QDBusMessage from PyQt6.QtWidgets import QLabel, QMainWindow, QMessageBox, QPushButton, QTabBar, QTabWidget from . import VERSION, config from .dashboard import Dashboard from .fmt import job_title from .jobs import JobsTab from .packages import PackagesTab from .remote import DOWN, MISSING NOTIFY = "org.freedesktop.Notifications" NOTIFY_PATH = "/org/freedesktop/Notifications" END_EVENT = {"ok": "finished", "failed": "failed", "killed": "killed", "lost": "lost"} MAX_RETRY_MS = 30000 class Notifier(QObject): """freedesktop notifications over D-Bus; clicked(job id) on the default action.""" clicked = pyqtSignal(str) def __init__(self, parent=None): super().__init__(parent) bus = QDBusConnection.sessionBus() self.iface = QDBusInterface(NOTIFY, NOTIFY_PATH, NOTIFY, bus) bus.connect(NOTIFY, NOTIFY_PATH, NOTIFY, "ActionInvoked", self._action) self.ids = {} # notification id -> job id def send(self, title, body, jid=""): reply = self.iface.call( "Notify", "sb-builder", QDBusArgument(0, QMetaType.Type.UInt.value), "", title, body, QDBusArgument(["default", "Open"], QMetaType.Type.QStringList.value), {}, -1) if reply.type() == QDBusMessage.MessageType.ReplyMessage: self.ids[reply.arguments()[0]] = jid @pyqtSlot("uint", "QString") def _action(self, nid, _key): if nid in self.ids: self.clicked.emit(self.ids.pop(nid)) class MainWindow(QMainWindow): def __init__(self, cfg, remote, notifier=None): super().__init__() self.cfg, self.remote = cfg, remote self.notifier = notifier or Notifier(self) self.notifier.clicked.connect(self.open_job) self.jobs_tab = None self.packages_tab = None self.seen = None # job id -> status at the last poll; None before the first self.online = None # None until the first answer self.overlays = [] self.vm_action = None # "start" or "resume" offered on the VM button self.vm_woken = False # we started the VM: notify once ssh answers self.running_id = None self.setWindowTitle("sb-builder") self.tabs = QTabWidget(tabsClosable=True) self.tabs.tabCloseRequested.connect(self._close_tab) self.tabs.currentChanged.connect(lambda _i: self.poll_stats()) self.dashboard = Dashboard() self.dashboard.refresh_btn.clicked.connect(self.poll_stats) self.dashboard.tiles["Packages"].link(self.open_packages) self.dashboard.jobs.link(self.open_jobs) self.tabs.addTab(self.dashboard, "Dashboard") self.tabs.tabBar().setTabButton(0, QTabBar.ButtonPosition.RightSide, None) self.setCentralWidget(self.tabs) file_menu = self.menuBar().addMenu("&File") file_menu.addAction("&Reload config", self.reload_config) file_menu.addAction("&Quit", self.close).setShortcut("Ctrl+Q") open_menu = self.menuBar().addMenu("&Open") open_menu.addAction("&Packages", self.open_packages) open_menu.addAction("&Jobs", self.open_jobs) self.conn = QLabel("● connecting") self.key = QLabel() self.key.hide() self.vm_btn = QPushButton() self.vm_btn.clicked.connect(lambda: self._vm_do(self.vm_action)) self.vm_btn.hide() self.job_btn = QPushButton(flat=True) self.job_btn.clicked.connect(lambda: self.open_job(self.running_id)) self.job_btn.hide() for w in (self.job_btn, self.key, self.vm_btn, self.conn): self.statusBar().addPermanentWidget(w) self.jobs_timer = QTimer(self, timeout=self.poll_jobs) self.stats_timer = QTimer(self, timeout=self.poll_slow) self._start_timers() QTimer.singleShot(0, self.poll_jobs) # ── polling ───────────────────────────────────────────────────────────── def _start_timers(self): self.jobs_timer.start(self.cfg["refresh"]["jobs"] * 1000) self.stats_timer.start(self.cfg["refresh"]["dashboard"] * 1000) def poll_jobs(self): self.remote.call(self._got_jobs, "jobs") def poll_slow(self): self.poll_stats() if self.online: self.remote.call(self._got_key, "gpg-status") def _got_key(self, data, err): if err: return self._failed(err) self.key.setVisible(data["unlocked"] is not None) self.key.setText("🔓 key unlocked" if data["unlocked"] else "🔒 key locked") self.key.setToolTip(f"package signing key {data['key']}") def poll_stats(self): if self.online and self.tabs.currentWidget() is self.dashboard: self.remote.call(self._got_stats, "stats") def _got_jobs(self, data, err): if err: return self._failed(err) self._up() jobs = data["jobs"] self._notify_ends(jobs) self.dashboard.show_jobs(jobs) if self.jobs_tab: self.jobs_tab.set_jobs(jobs) run = next((j for j in jobs if j["status"] == "running"), None) self.running_id = run and run["id"] self.job_btn.setVisible(run is not None) if run: self.job_btn.setText(f"▶ {job_title(run)}") def _got_stats(self, data, err): if err: return self._failed(err) self.dashboard.show_stats(data) def _got_version(self, data, err): if err: return self._failed(err) self.overlays = data.get("overlays", []) if self.jobs_tab: self.jobs_tab.set_overlays(self.overlays) if data["version"] != VERSION: self.conn.setText(f"● helper {data['version']}, GUI {VERSION}: run ./install.sh --remote") def _up(self): if self.online: return self.online = True self.conn.setText("● connected") self.vm_btn.hide() self.dashboard.set_stale(None) self.jobs_timer.setInterval(self.cfg["refresh"]["jobs"] * 1000) self.remote.call(self._got_version, "version") self.poll_slow() if self.vm_woken: self.vm_woken = False self.notify("vm-up", f"{self.cfg['vm']['domain']} is up") def _failed(self, err): if err == MISSING: self.conn.setText("● sbb-remote missing: run ./install.sh --remote") return if err != DOWN: self.statusBar().showMessage(err, 10000) return if self.online is not False: self.online = False self.dashboard.set_stale(time.time()) base = self.cfg["refresh"]["jobs"] * 1000 self.jobs_timer.setInterval(min(self.jobs_timer.interval() * 2, max(MAX_RETRY_MS, base))) self._check_vm() # ── VM control ────────────────────────────────────────────────────────── def _check_vm(self): if not self.cfg["vm"]["domain"]: self.conn.setText("● down") return self.remote.virsh(self._got_domstate, "domstate", self.cfg["vm"]["domain"]) def _got_domstate(self, rc, state): if self.online: return # ssh came back while virsh ran self.vm_btn.hide() if rc != 0: self.conn.setText("● down") elif state == "shut off": self.conn.setText("● VM off") if self.cfg["vm"]["autostart"]: self._vm_do("start") else: self._offer("Start VM", "start") elif state == "paused": self.conn.setText("● VM paused") self._offer("Resume VM", "resume") elif state == "running": self.conn.setText("● VM running, ssh unreachable") else: self.conn.setText(f"● VM {state}") def _offer(self, label, action): self.vm_action = action self.vm_btn.setText(label) self.vm_btn.show() def _vm_do(self, action): self.vm_btn.hide() self.vm_woken = True self.conn.setText(f"● VM {action} requested") self.remote.virsh( lambda rc, out: rc and self.statusBar().showMessage(f"virsh {action}: {out}", 10000), action, self.cfg["vm"]["domain"]) # ── notifications ─────────────────────────────────────────────────────── def notify(self, event, body, jid=""): n = self.cfg["notify"] if n["enabled"] and event in n["events"]: self.notifier.send(f"sb-builder: {event}", body, jid) def _notify_ends(self, jobs): prev, self.seen = self.seen, {j["id"]: j["status"] for j in jobs} if prev is None: return # first poll: old jobs are history, not news for j in jobs: event = END_EVENT.get(j["status"]) if event and prev.get(j["id"], "running") != j["status"]: self.notify(event, job_title(j), j["id"]) if j["kind"] in ("slackrepo", "publish") and self.packages_tab: self.packages_tab.reload() # build state changed # ── tabs ──────────────────────────────────────────────────────────────── def open_jobs(self): if self.jobs_tab is None: self.jobs_tab = JobsTab(self.remote, self.cfg["builds"]["publish"]) self.jobs_tab.refresh_requested.connect(self.poll_jobs) self.jobs_tab.set_overlays(self.overlays) self.tabs.addTab(self.jobs_tab, "Jobs") self.poll_jobs() self.tabs.setCurrentWidget(self.jobs_tab) return self.jobs_tab def open_packages(self): if self.packages_tab is None: self.packages_tab = PackagesTab(self.remote) self.packages_tab.start_requested.connect(lambda args: self.open_jobs().start(*args)) self.tabs.addTab(self.packages_tab, "Packages") self.packages_tab.reload() self.tabs.setCurrentWidget(self.packages_tab) return self.packages_tab def open_job(self, jid): self.show() self.raise_() self.activateWindow() if jid: self.open_jobs().select(jid) def _close_tab(self, i): w = self.tabs.widget(i) if w is self.dashboard: return self.tabs.removeTab(i) if w is self.jobs_tab: self.jobs_tab = None w.stop() if w is self.packages_tab: self.packages_tab = None w.deleteLater() def reload_config(self): cfg, err = config.load() if err: QMessageBox.warning(self, "Config error", f"{err}\nUsing defaults.") self.cfg.clear() self.cfg.update(cfg) # Remote holds the same dict self._start_timers() ```` - [ ] **Step 7: Run the whole suite** Run: `python3 -m unittest discover -s tests EOF git log --format='%h %G? %s' -1 ``` --- ### Task 4: docs - [ ] **Step 1:** Replace `README.md`: ````markdown # sb-builder Desktop GUI for managing a Slackware SlackBuilds build VM from one place. It runs slackrepo builds, overlay syncs (sbpull, sbshadow), hintfile work (mkhint) and SBo orphan discovery on the VM over ssh, as background jobs that keep running when the GUI disconnects and can be reattached later. A dashboard shows VM and repository state, and the GUI can start the VM through libvirt when it is off. Work in progress. See `docs/superpowers/specs/` for the design. ## Requirements - Desktop: Python 3.11 or newer, PyQt6 (with QtDBus), ssh access to the VM, optionally `virsh` for starting the VM - VM: Python 3.11 or newer, root ssh login, the [slackrepo_setup](https://forge.slackware.nl/danix/slackrepo_setup) tools (`sbpull`, `sbshadow`) and mkhint ## Install ./install.sh # ~/bin/sb-builder -> this repo, GUI config if missing ./install.sh --remote # helper to the VM, its config if missing (asks first) `BINDIR` changes the launcher directory, `SBB_TARGET` the ssh destination for `--remote` (default `buildsystem`). Configs: - GUI: `~/.config/sb-builder/config.toml`, see `config.example.toml` - VM: `/etc/sb-builder/config.toml`, see `remote/config.example.toml` Every key is optional. `File > Reload config` rereads the GUI config. ## Usage sb-builder - **Dashboard**: VM load, uptime and free disk, package counts, slackrepo build results, last overlay pull, recent jobs. Refreshes while visible. - **Open > Jobs**: start a job from the Start menu (`sbpull ` preview, `sbshadow` preview or apply), follow its log live, kill it. After a preview that lists `shadow:` or `unshadow:` lines, "Apply shadowing" runs `sbshadow --yes`. - **Open > Packages** (or click the Packages tile): every package of the repository, with built version and status (`ok`, `failed`, `never`, `outdated`). Search by name or summary, or with `cat:`, `maint:`, `req:`, `status:`; toggle the overlay and status chips. The detail pane shows the package's versions, its dependencies after hints (dropped ones struck through), and what its hint file does, section by section. - Tick packages (or select one row) and press Build, Update, Rebuild or Remove; Update all updates everything. Each first runs a slackrepo `--preview` job. When the preview is fine, **Run it** in the Jobs tab starts the real run; untick **publish** to keep the packages local and press **Publish** later to push them. - Real runs sign the repository. When the package signing key is locked (status bar 🔒), the GUI asks for its passphrase first and hands it to the VM's gpg-agent; it is never stored. Jobs run on the VM detached from the ssh session, one at a time. Closing the GUI or losing the connection does not stop them; reopening the job resumes the log. Job state and logs live in `/var/lib/sb-builder/jobs/` on the VM. When ssh fails and `vm.domain` is set, the status bar shows the libvirt state of the VM and offers to start or resume it. Desktop notifications report finished, failed, killed and lost jobs, and the VM coming up. ## License Copyright (C) 2026 Danilo M. GPLv2 only. See [LICENSE](LICENSE) for the full text. ## Development Approach This project is developed using AI-assisted tools. Code is generated with the help of AI based on human-provided specifications, design decisions, and iterative feedback. All contributions are reviewed, tested, and curated by the maintainer before being included in the codebase. AI is used as a productivity and exploration tool, while human oversight remains central to all decisions. The goal is to combine the flexibility of AI-assisted development with standard open-source practices such as transparency, review, and accountability. ```` - [ ] **Step 2:** Replace `AGENTS.md`: ````markdown # AGENTS.md Guidance for AI coding agents working in this repository. ## What this is `sb-builder`: a PyQt6 desktop GUI that drives the SlackBuild tooling on a Slackware -current build VM (slackrepo, mkhint, sbpull, sbshadow) over ssh. Two programs: the GUI (`sb_builder/`, PyQt6 incl. QtDBus, stdlib otherwise) and `remote/sbb-remote`, a single-file stdlib-only Python helper deployed on the VM. Do not add third-party dependencies to either. Design and phasing: `docs/superpowers/specs/`. Phase 1 (core) is `2026-10-19-sb-builder-phase1-design.md`. ## How it works - Every GUI action is `ssh sbb-remote [args]`; the helper answers with JSON. `sb_builder/remote.py` is the only code that runs ssh (and local `virsh`). - Trust boundary: the helper runs only a fixed table of job kinds and builds argv itself from validated args. Never pass shell strings or paths from the GUI to the helper. - Jobs run detached on the VM (`setsid`, `stdin=/dev/null`), one at a time under a flock, logging to `/jobs//`. The GUI can disconnect and reattach with `job-follow `. - Exception, by design: `gpg-unlock` reads the signing key passphrase from stdin (`remote.call(..., stdin=...)`) and feeds it to `gpg-preset-passphrase`. Every other subprocess in the helper runs with `stdin=DEVNULL`, so nothing can read the passphrase off the ssh channel. Never put the passphrase in argv, env, logs or files. - Hints are parsed (`shlex`), never sourced. `sb_builder/hints.py` mirrors slackrepo's own rules (`../GITHUB/slackrepo/libexec/functions.d/parsefunctions.sh`); check that file when changing it. - Jobs never get input. Any tool run as a job must be non-interactive (`--yes`/`--no` style flags). If a tool needs a flag it lacks, add it in that tool's own repo (e.g. `slackrepo_setup` for sbshadow, `mkhintfile` for mkhint) rather than working around it here. ## Conventions - GUI config: `$XDG_CONFIG_HOME/sb-builder/config.toml`. Helper config: `/etc/sb-builder/config.toml` on the VM. Both read with `tomllib`, both have shipped `config.example.toml` files, defaults live in code. - One file per GUI tab. - Deploying to the VM is outward-facing: confirm before `install.sh --remote`. - Placeholders only in code, docs, tests and commits. ## Testing python3 -m unittest discover -s tests `test_gui.py` forces `QT_QPA_PLATFORM=offscreen` (the desktop session sets its own, and a modal dialog on the real display blocks the run). A fake remote must answer every call a test triggers: an unanswered call returns "ssh: connection failed", which opens a warning dialog that blocks forever offscreen. Headless (`test_gui.py` sets `QT_QPA_PLATFORM=offscreen`), no VM needed. `test_remote.py` loads `remote/sbb-remote` with `SourceFileLoader` (no `.py` extension) and points it at a temp state dir through `SBB_CONFIG`. Two `ResourceWarning: subprocess ... is still running` lines are expected: the detached `_run` is never waited for. To try the helper on the VM without installing it, copy it to `/tmp` there and run it with `SBB_CONFIG` pointing at a config whose `state_dir` is under `/tmp`. ## Repo facts - License: GPLv2 only (`LICENSE`, header in each source file). - Work directly on master. - `.superpowers/` is brainstorming scratch, ignored. ```` - [ ] **Step 3: Commit and push** ```bash git add README.md AGENTS.md git commit -F - <<'EOF' Document the Packages tab, publishing and signing key unlock Co-Authored-By: Claude Opus 5.5 EOF git log --format='%h %G? %s' -1 && git push ``` --- ### Task 5: deploy and first real run (with the user) Ask before each VM step. Order matters: the `.local` file must be built from the old config before the new config replaces it. - [ ] **Step 1: Inspect targets** ```bash ssh buildsystem 'ls -l /etc/slackrepo/slackrepo_SBo-danix.conf /etc/slackrepo/SBo-danix.local /usr/local/bin/sb-publish /root/.gnupg/gpg-agent.conf 2>&1' ``` Expected: config and agent conf exist; `.local` and `sb-publish` do not. - [ ] **Step 2: Build SBo-danix.local on the VM from the old config** (values never leave the VM) ```bash cd ~/Programming/GIT/slackrepo_setup scp make-local.py buildsystem:/tmp/make-local.py ssh buildsystem 'cp -p /etc/slackrepo/slackrepo_SBo-danix.conf /etc/slackrepo/slackrepo_SBo-danix.conf.pre-sb-builder && python3 /tmp/make-local.py /etc/slackrepo/slackrepo_SBo-danix.conf /etc/slackrepo/SBo-danix.local && rm /tmp/make-local.py && stat -c "%a %n" /etc/slackrepo/SBo-danix.local' ``` Expected: `wrote /etc/slackrepo/SBo-danix.local: SB_RSYNC_TARGET, SB_RSYNC_EXCLUDES, SB_RSYNC_PASSWORD_FILE, SB_GOTIFY_KEY, SB_GOTIFY_URL` and `600 /etc/slackrepo/SBo-danix.local`. - [ ] **Step 3: Install config, sb-publish, agent conf** ```bash cd ~/Programming/GIT/slackrepo_setup scp slackrepo_SBo-danix.conf buildsystem:/etc/slackrepo/slackrepo_SBo-danix.conf scp sb-publish buildsystem:/usr/local/bin/sb-publish scp gpg-agent.conf buildsystem:/root/.gnupg/gpg-agent.conf ssh buildsystem 'chmod 755 /usr/local/bin/sb-publish; chmod 600 /root/.gnupg/gpg-agent.conf; gpgconf --reload gpg-agent; bash -n /etc/slackrepo/slackrepo_SBo-danix.conf && echo config ok; bash -c ". /etc/slackrepo/SBo-danix.local; [ -n \"\$SB_RSYNC_TARGET\" ] && echo local ok"' ``` - [ ] **Step 4: Deploy the helper**: `cd ~/Programming/GIT/sb-builder && ./install.sh --remote`. Expected last line: `{"version": "0.2.0", "overlays": ["personal", "pentesting"]}`. Existing `/etc/sb-builder/config.toml` is kept; its defaults cover the new keys. - [ ] **Step 5: User runs `sb-builder`**, checks: status bar shows 🔒 key locked; Packages tile opens the tab with ~9957 rows; chips and `cat:personal` filter; a hinted package shows its hint sections; DELREQUIRES deps struck through. - [ ] **Step 6: Preview, run unpublished, publish**: pick one personal package that shows `outdated`, Update. Preview job ends ✔ with [Run it]; untick publish; Run it asks for the passphrase (a wrong one asks again); the job runs and ends ✔; status bar shows 🔓. On the VM: `ls -l /repo/personal//` shows the new package and the mirror is not updated yet. Then Publish: the `publish` job runs `sb-publish` and ends ✔. - [ ] **Step 7: Note the preset passphrase lifetime**: check the 🔓 indicator again after more than 4h (default-cache-ttl) and after 6h (max-cache-ttl); record in the spec whether presets expire with the TTLs. Either way the GUI asks again when locked. - [ ] **Step 8: Terminal check**: from a VM shell, `slackrepo --preview update `: no Gotify message arrives (was sent before this change). Record anything that misbehaves as a follow-up; do not patch it silently.