felhom-agent/internal/signedjobs/runner_test.go

package signedjobs

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"io"
	"log/slog"
	"sync"
	"testing"
	"time"

	"gitea.dooplex.hu/admin/felhom-agent/internal/authz"
	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
	"gitea.dooplex.hu/admin/felhom-agent/internal/reconcile"
)

func quiet() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) }

const testHost = "demo-felhom-01"

// jobsQueue is an in-memory JobSource for the runner tests.
type jobsQueue struct {
	mu        sync.Mutex
	jobs      []hub.JobWire
	completed []string
}

func (q *jobsQueue) add(j hub.JobWire) { q.mu.Lock(); q.jobs = append(q.jobs, j); q.mu.Unlock() }
func (q *jobsQueue) Jobs(context.Context) ([]hub.JobWire, error) {
	q.mu.Lock()
	defer q.mu.Unlock()
	out := append([]hub.JobWire(nil), q.jobs...)
	return out, nil
}
func (q *jobsQueue) CompleteJob(_ context.Context, jobID string) error {
	q.mu.Lock()
	defer q.mu.Unlock()
	q.completed = append(q.completed, jobID)
	// also remove from the pending list so a re-run doesn't re-process it
	kept := q.jobs[:0]
	for _, j := range q.jobs {
		if j.JobID != jobID {
			kept = append(kept, j)
		}
	}
	q.jobs = kept
	return nil
}
func (q *jobsQueue) wasCompleted(jobID string) bool {
	q.mu.Lock()
	defer q.mu.Unlock()
	for _, id := range q.completed {
		if id == jobID {
			return true
		}
	}
	return false
}

// corruptSig flips bytes in the envelope's armored signature (a hub forging/altering a blob).
func corruptSig(t *testing.T, j hub.JobWire) hub.JobWire {
	t.Helper()
	raw, _ := base64.StdEncoding.DecodeString(j.BlobB64)
	var env Envelope
	json.Unmarshal(raw, &env)
	// Replace the armored signature with a structurally-valid-but-wrong one: re-arm random bytes.
	env.SigArmored = env.SigArmored[:len(env.SigArmored)/2] + "AAAA" + env.SigArmored[len(env.SigArmored)/2:]
	out, _ := json.Marshal(env)
	j.BlobB64 = base64.StdEncoding.EncodeToString(out)
	return j
}

// fakeExecutor records Execute calls and returns a configurable error.
type fakeExecutor struct {
	mu    sync.Mutex
	calls []string // ops executed
	err   error
}

func (f *fakeExecutor) Execute(_ context.Context, op string, _ json.RawMessage) error {
	f.mu.Lock()
	f.calls = append(f.calls, op)
	f.mu.Unlock()
	return f.err
}
func (f *fakeExecutor) count() int { f.mu.Lock(); defer f.mu.Unlock(); return len(f.calls) }

// newRealGateRunner builds a runner with the REAL gate+verifier pinned to `signer`, an in-memory
// nonce store, plus a fake jobs source + fake executor. wipeParams is the params to sign.
func newRealGateRunner(t *testing.T, signer testSigner) (*Runner, *jobsQueue, *fakeExecutor) {
	t.Helper()
	store := authz.NewMemoryNonceStore()
	verifier := authz.New([]authz.AllowedSigner{signer.allowed(t, "ops-1", authz.RoleOperational)}, store, testHost)
	gate := reconcile.NewGate(verifier, testHost, nil, quiet())
	src := &jobsQueue{}
	exec := &fakeExecutor{}
	return NewRunner(src, gate, exec, testHost, quiet()), src, exec
}

const wipeParamsJSON = `{"durable_id":"byid:wwn-0xtest","fstype":"ext4"}`

// VALID: a correctly-signed wipe → executor runs once + the job is cleared.
func TestRunner_ValidSignedWipeExecutes(t *testing.T) {
	s := newTestSigner(t)
	r, src, exec := newRealGateRunner(t, s)
	now := time.Now().UTC()
	src.add(mintJob(t, s, "j1", testHost, "", "ops-1", wipeParamsJSON, now, now.Add(time.Hour)))

	n, err := r.RunOnce(context.Background())
	if err != nil {
		t.Fatalf("RunOnce: %v", err)
	}
	if n != 1 || exec.count() != 1 || exec.calls[0] != "storage_wipe" {
		t.Fatalf("executor calls = %v (n=%d), want one storage_wipe", exec.calls, n)
	}
	if !src.wasCompleted("j1") {
		t.Error("valid job was not cleared after execution")
	}
}

// REPLAY: the same blob resubmitted (a second job) → rejected, executor NOT called again.
func TestRunner_ReplayRejected(t *testing.T) {
	s := newTestSigner(t)
	r, src, exec := newRealGateRunner(t, s)
	now := time.Now().UTC()
	job := mintJob(t, s, "j1", testHost, "", "ops-1", wipeParamsJSON, now, now.Add(time.Hour))
	src.add(job)
	if _, err := r.RunOnce(context.Background()); err != nil {
		t.Fatal(err)
	}
	if exec.count() != 1 {
		t.Fatalf("first run executed %d times, want 1", exec.count())
	}
	// Resubmit the IDENTICAL signed blob (same nonce) as a new job.
	replay := job
	replay.JobID = "j1-replay"
	src.add(replay)
	if _, err := r.RunOnce(context.Background()); err != nil {
		t.Fatal(err)
	}
	if exec.count() != 1 {
		t.Errorf("replay caused a second execution (count=%d) — nonce-burn failed", exec.count())
	}
	if !src.wasCompleted("j1-replay") {
		t.Error("replayed job should be cleared (rejected)")
	}
}

// NON-PINNED signer → rejected, executor NOT called.
func TestRunner_NonPinnedSignerRejected(t *testing.T) {
	pinned := newTestSigner(t)
	attacker := newTestSigner(t) // a DIFFERENT key, not pinned
	r, src, exec := newRealGateRunner(t, pinned)
	now := time.Now().UTC()
	src.add(mintJob(t, attacker, "j1", testHost, "", "ops-1", wipeParamsJSON, now, now.Add(time.Hour)))

	r.RunOnce(context.Background())
	if exec.count() != 0 {
		t.Errorf("a non-pinned signature was EXECUTED (count=%d) — must be rejected", exec.count())
	}
	if !src.wasCompleted("j1") {
		t.Error("rejected job should be cleared")
	}
}

// EXPIRED → rejected, executor NOT called.
func TestRunner_ExpiredRejected(t *testing.T) {
	s := newTestSigner(t)
	r, src, exec := newRealGateRunner(t, s)
	past := time.Now().UTC().Add(-2 * time.Hour)
	src.add(mintJob(t, s, "j1", testHost, "", "ops-1", wipeParamsJSON, past, past.Add(time.Hour))) // expired an hour ago

	r.RunOnce(context.Background())
	if exec.count() != 0 {
		t.Errorf("an EXPIRED op was executed (count=%d)", exec.count())
	}
}

// RETARGETED (Target.HostID = another host) → rejected, executor NOT called.
func TestRunner_RetargetRejected(t *testing.T) {
	s := newTestSigner(t)
	r, src, exec := newRealGateRunner(t, s)
	now := time.Now().UTC()
	src.add(mintJob(t, s, "j1", "some-other-host", "", "ops-1", wipeParamsJSON, now, now.Add(time.Hour)))

	r.RunOnce(context.Background())
	if exec.count() != 0 {
		t.Errorf("an op targeting another host was executed on this host (count=%d) — anti-retarget failed", exec.count())
	}
}

// FORGED (a compromised hub queues a blob with a garbage signature) → rejected, executor NOT called.
func TestRunner_ForgedSignatureRejected(t *testing.T) {
	s := newTestSigner(t)
	r, src, exec := newRealGateRunner(t, s)
	now := time.Now().UTC()
	job := mintJob(t, s, "j1", testHost, "", "ops-1", wipeParamsJSON, now, now.Add(time.Hour))
	// Corrupt the envelope's signature (simulate a hub forging/altering the blob).
	job = corruptSig(t, job)
	src.add(job)

	r.RunOnce(context.Background())
	if exec.count() != 0 {
		t.Errorf("a forged/altered signature was executed (count=%d) — the compromised-hub case must be rejected", exec.count())
	}
	if !src.wasCompleted("j1") {
		t.Error("forged job should be cleared")
	}
}

// No verifier pinned (no signers) → every signed op is refused pending_signature, executor not called.
func TestRunner_NoSignersAllRejected(t *testing.T) {
	s := newTestSigner(t)
	gate := reconcile.NewGate(nil, testHost, nil, quiet()) // nil verifier
	src := &jobsQueue{}
	exec := &fakeExecutor{}
	r := NewRunner(src, gate, exec, testHost, quiet())
	now := time.Now().UTC()
	src.add(mintJob(t, s, "j1", testHost, "", "ops-1", wipeParamsJSON, now, now.Add(time.Hour)))

	r.RunOnce(context.Background())
	if exec.count() != 0 {
		t.Errorf("with no pinned signer a signed op executed (count=%d) — must be pending_signature", exec.count())
	}
}

// A malformed envelope → cleared without ever calling the executor.
func TestRunner_MalformedEnvelopeCleared(t *testing.T) {
	s := newTestSigner(t)
	r, src, exec := newRealGateRunner(t, s)
	src.add(hub.JobWire{JobID: "bad", BlobB64: "!!!not base64!!!"})
	r.RunOnce(context.Background())
	if exec.count() != 0 {
		t.Errorf("a malformed envelope reached the executor (count=%d)", exec.count())
	}
	if !src.wasCompleted("bad") {
		t.Error("malformed job should be cleared")
	}
}