feat(felhomsshd): dedicated OOB sshd instance + port-adaptive belt (H1 Parts 2-4 agent)

internal/felhomsshd: agent-managed felhom-sshd (claim port [8822,2222,8022,62222]
loud-fail-on-exhaustion; render config→sshd -t→reload never-restart-on-change
[SF-2]; operator authorized_keys from the hub block outside ~/.ssh [SF-3]); the
static-table nft belt mutating ONLY @operator_ips + @ssh_port [trap 4]; health/heal
(reset-failed-then-restart with 10min cooldown, NEVER restart onto an invalid
config) + the oob heartbeat stanza. configs/felhom-sshd.service (SAFE, no
RuntimeDirectory [SF-1]). FELHOM_SSHD + FELHOM_OOB sudoers (set-elements only).
oob.enabled config DEFAULT FALSE. Wired into main like wgtunnel.

Non-hollow tests: claim clean/contention/idempotent/exhaustion; config
safe+byte-stable+refuses-:22; belt mutate-then-idempotent + never-touches-rules;
heal no-restart-on-invalid-config + cooldown; status reflects block.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PSK5g6qYLknKj8u3QAFEr6

internal/felhomsshd/belt_health_test.go

@@ -0,0 +1,178 @@
+package felhomsshd
+
+import (
+	"context"
+	"io"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
+)
+
+// scriptRunner records every exec and returns scripted stdout per "name+first-arg" key; scripted
+// errors per the same key. Everything else returns empty success.
+type scriptRunner struct {
+	mu     sync.Mutex
+	calls  [][]string
+	stdout map[string]string
+	errs   map[string]error
+}
+
+func newScriptRunner() *scriptRunner {
+	return &scriptRunner{stdout: map[string]string{}, errs: map[string]error{}}
+}
+
+func key(name string, args ...string) string {
+	// `nft list set inet felhom_oob <setName>` → key by the set name (last arg), so the two set reads
+	// are distinguishable. Everything else keys by first arg.
+	if name == "nft" && len(args) > 0 && args[0] == "list" {
+		return "nft list " + args[len(args)-1]
+	}
+	if len(args) > 0 {
+		return name + " " + args[0]
+	}
+	return name
+}
+
+func (r *scriptRunner) Run(_ context.Context, name string, args ...string) ([]byte, []byte, error) {
+	r.mu.Lock()
+	r.calls = append(r.calls, append([]string{name}, args...))
+	k := key(name, args...)
+	out := r.stdout[k]
+	err := r.errs[k]
+	r.mu.Unlock()
+	if err != nil {
+		return nil, []byte("scripted failure"), err
+	}
+	return []byte(out), nil, nil
+}
+
+func (r *scriptRunner) RunStdin(ctx context.Context, _ io.Reader, name string, args ...string) ([]byte, []byte, error) {
+	return r.Run(ctx, name, args...)
+}
+
+func (r *scriptRunner) count(name, firstArg string) int {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	n := 0
+	for _, c := range r.calls {
+		if c[0] == name && len(c) > 1 && c[1] == firstArg {
+			n++
+		}
+	}
+	return n
+}
+
+func (r *scriptRunner) sawRuleMutation() bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	for _, c := range r.calls {
+		// trap 4: the agent must NEVER run `nft add rule` / `nft -f` / `nft flush ruleset|table`.
+		if c[0] == "nft" && len(c) > 1 {
+			if c[1] == "-f" || (len(c) > 2 && c[1] == "add" && c[2] == "rule") {
+				return true
+			}
+			if c[1] == "flush" && len(c) > 2 && (c[2] == "ruleset" || c[2] == "table") {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+func TestBelt_SyncMutatesThenIdempotent(t *testing.T) {
+	r := newScriptRunner()
+	// first read: both sets empty (no "elements" block)
+	r.stdout["nft list operator_ips"] = "set operator_ips {\n\ttype ipv4_addr\n}"
+	r.stdout["nft list ssh_port"] = "set ssh_port {\n\ttype inet_service\n}"
+	b := NewBelt(r, nil)
+	b.Sync(context.Background(), 8822, "10.77.0.250")
+
+	// mutations happened: flush + add for BOTH sets
+	if r.count("nft", "flush") < 2 || r.count("nft", "add") < 2 {
+		t.Fatalf("first sync must flush+add both sets; flushes=%d adds=%d", r.count("nft", "flush"), r.count("nft", "add"))
+	}
+	if r.sawRuleMutation() {
+		t.Fatal("belt must NEVER mutate rules — only set elements (trap 4)")
+	}
+
+	// second sync: the reads now return the desired elements → ZERO mutations (idempotent)
+	r2 := newScriptRunner()
+	r2.stdout["nft list operator_ips"] = "elements = { 10.77.0.250 }"
+	r2.stdout["nft list ssh_port"] = "elements = { 8822 }"
+	b2 := NewBelt(r2, nil)
+	b2.Sync(context.Background(), 8822, "10.77.0.250")
+	if r2.count("nft", "flush") != 0 || r2.count("nft", "add") != 0 {
+		t.Fatalf("idempotent sync must do ZERO mutations; flushes=%d adds=%d", r2.count("nft", "flush"), r2.count("nft", "add"))
+	}
+}
+
+func TestBelt_OperatorEmptyEmptiesSet(t *testing.T) {
+	r := newScriptRunner()
+	r.stdout["nft list operator_ips"] = "elements = { 10.77.0.250 }" // currently has an operator IP
+	r.stdout["nft list ssh_port"] = "elements = { 8822 }"
+	b := NewBelt(r, nil)
+	b.Sync(context.Background(), 8822, "") // OOB off → operator set must be emptied
+	// operator_ips: current {250} != desired {} → flush, no add. ssh_port: {} vs {8822}...
+	if r.count("nft", "flush") < 1 {
+		t.Fatal("emptying the operator set must flush it")
+	}
+}
+
+// health: inactive + INVALID config → NO restart (the red-proof); inactive + valid → restart.
+func TestHeal_NoRestartOnInvalidConfig(t *testing.T) {
+	r := newScriptRunner()
+	r.errs["sshd -t"] = context.DeadlineExceeded // sshd -t FAILS (config invalid)
+	m := newTestManager(r, false)                // inactive
+	m.HealAndCheck(context.Background(), 8822)
+	if r.count("systemctl", "restart") != 0 {
+		t.Fatal("a broken config must NOT trigger a restart (never degraded→dead)")
+	}
+}
+
+func TestHeal_RestartsWhenDownWithValidConfigThenCooldown(t *testing.T) {
+	r := newScriptRunner() // sshd -t passes (no error)
+	m := newTestManager(r, false)
+	m.HealAndCheck(context.Background(), 8822)
+	if r.count("systemctl", "restart") != 1 {
+		t.Fatalf("down + valid config → exactly one restart, got %d", r.count("systemctl", "restart"))
+	}
+	// cooldown: a second call inside the window → still one restart
+	m.HealAndCheck(context.Background(), 8822)
+	if r.count("systemctl", "restart") != 1 {
+		t.Fatalf("restart cooldown violated: %d restarts", r.count("systemctl", "restart"))
+	}
+}
+
+func TestStatus_ReflectsBlockAndPort(t *testing.T) {
+	r := newScriptRunner()
+	r.stdout["sshd -T"] = "port 8822\nsomethingelse yes\n"
+	m := newTestManager(r, true) // active
+	m.port = 8822
+	st := m.Status(context.Background(), &hub.WireWireguard{OOBPeerIP: "10.77.0.250", OOBOperatorSSHKey: "ssh-ed25519 AAAA op"})
+	if !st.FelhomSshdActive || st.FelhomSshdPort != 8822 {
+		t.Fatalf("status: %+v", st)
+	}
+	if !st.OperatorPeerConfigured || !st.OperatorKeyConfigured {
+		t.Fatalf("status must reflect operator peer + key configured: %+v", st)
+	}
+}
+
+// newTestManager builds a Manager with injected active-state + a now clock; sshd -T/-t go through the
+// runner. isFree unused here.
+func newTestManager(r *scriptRunner, active bool) *Manager {
+	m := NewManager(r, "/tmp/felhomsshd-test", nil)
+	m.isActive = func(context.Context) bool { return active }
+	m.isFailed = func(context.Context) bool { return false }
+	m.now = func() time.Time { return time.Unix(1783270000, 0) }
+	return m
+}
+
+func TestConfig_NoRuntimeDirectoryString(t *testing.T) {
+	c, _ := renderConfig(2222)
+	if strings.Contains(c, "RuntimeDirectory") {
+		t.Fatal("SF-1: config must never contain RuntimeDirectory")
+	}
+}