slice 10B: operator-signed destructive completion (offline key + signing CLI) (v0.16.0)
A destructive op runs ONLY on a pinned-key-verified, nonce-fresh, in-window, host-bound, durable-id-bound operator signature. New cmd/felhom-opsign signs canonical OpBlobs offline via ssh-keygen -Y sign (hardware-ready); the signing key is never in the hub or agent. New internal/signedjobs runner verifies each queued blob through the gate and only on all-pass runs the WipeExecutor, which re-resolves the DURABLE device id + re-inspects (8C) before mkfs — closing the 8C data-bearing-wipe pending_signature gap. New storage durable-device resolution; authz.CanonicalBlob promoted to production. Real-crypto tests assert valid executes and forged/replay/expired/retarget/non-pinned are rejected (executor never called). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,115 @@
|
||||
package signedjobs
|
||||
|
||||
import (
|
||||
"crypto/ed25519"
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"encoding/pem"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gitea.dooplex.hu/admin/felhom-agent/internal/authz"
|
||||
"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
|
||||
"golang.org/x/crypto/ssh"
|
||||
)
|
||||
|
||||
// In-test SSHSIG minter (mirrors internal/reconcile/mint_test.go's framing) so the runner tests
|
||||
// exercise the REAL authz.Verifier + reconcile.Gate over genuinely-signed blobs — a positive case
|
||||
// verifying proves the framing, and the adversarial cases (non-pinned/replay/expired/retarget/
|
||||
// forged) exercise the real rejection path end-to-end. Production authz stays verify-only.
|
||||
|
||||
const sshsigMagic = "SSHSIG"
|
||||
|
||||
type sshsigBlob struct {
|
||||
Version uint32
|
||||
PublicKey string
|
||||
Namespace string
|
||||
Reserved string
|
||||
HashAlgo string
|
||||
Signature string
|
||||
}
|
||||
|
||||
func signedDataForTest(ns string, msg []byte) []byte {
|
||||
h := sha256Sum(msg)
|
||||
body := ssh.Marshal(struct {
|
||||
Namespace string
|
||||
Reserved string
|
||||
HashAlgo string
|
||||
Hash []byte
|
||||
}{ns, "", "sha256", h})
|
||||
return append([]byte(sshsigMagic), body...)
|
||||
}
|
||||
|
||||
func mintArmor(pubMarshaled []byte, namespace string, message []byte, sign func([]byte) ssh.Signature) []byte {
|
||||
sb := &sshsigBlob{Version: 1, PublicKey: string(pubMarshaled), Namespace: namespace, Reserved: "", HashAlgo: "sha256"}
|
||||
sig := sign(signedDataForTest(namespace, message))
|
||||
sb.Signature = string(ssh.Marshal(&sig))
|
||||
raw := append([]byte(sshsigMagic), ssh.Marshal(sb)...)
|
||||
return pem.EncodeToMemory(&pem.Block{Type: "SSH SIGNATURE", Bytes: raw})
|
||||
}
|
||||
|
||||
type testSigner struct {
|
||||
pub ssh.PublicKey
|
||||
line string
|
||||
sign func([]byte) ssh.Signature
|
||||
}
|
||||
|
||||
func newTestSigner(t *testing.T) testSigner {
|
||||
t.Helper()
|
||||
pub, priv, err := ed25519.GenerateKey(rand.Reader)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sshPub, err := ssh.NewPublicKey(pub)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return testSigner{
|
||||
pub: sshPub,
|
||||
line: string(ssh.MarshalAuthorizedKey(sshPub)),
|
||||
sign: func(d []byte) ssh.Signature {
|
||||
return ssh.Signature{Format: ssh.KeyAlgoED25519, Blob: ed25519.Sign(priv, d)}
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (s testSigner) allowed(t *testing.T, keyID string, role authz.KeyRole) authz.AllowedSigner {
|
||||
t.Helper()
|
||||
as, err := authz.NewAllowedSigner(keyID, role, s.line)
|
||||
if err != nil {
|
||||
t.Fatalf("NewAllowedSigner: %v", err)
|
||||
}
|
||||
return as
|
||||
}
|
||||
|
||||
// mintJob builds a hub.JobWire carrying a signed storage_wipe envelope from the given signer.
|
||||
func mintJob(t *testing.T, s testSigner, jobID, host, guest, keyID, paramsJSON string, issued, expires time.Time) hub.JobWire {
|
||||
t.Helper()
|
||||
blob, err := authz.CanonicalBlob("storage_wipe", host, guest, keyID, randNonce(), paramsJSON, issued, expires)
|
||||
if err != nil {
|
||||
t.Fatalf("CanonicalBlob: %v", err)
|
||||
}
|
||||
sig := mintArmor(s.pub.Marshal(), authz.Namespace, blob, s.sign)
|
||||
env := Envelope{OpBlobB64: base64.StdEncoding.EncodeToString(blob), SigArmored: string(sig)}
|
||||
envJSON, _ := json.Marshal(env)
|
||||
return hub.JobWire{JobID: jobID, BlobB64: base64.StdEncoding.EncodeToString(envJSON)}
|
||||
}
|
||||
|
||||
func randNonce() string {
|
||||
var b [16]byte
|
||||
rand.Read(b[:])
|
||||
const hexd = "0123456789abcdef"
|
||||
out := make([]byte, 32)
|
||||
for i, x := range b {
|
||||
out[i*2] = hexd[x>>4]
|
||||
out[i*2+1] = hexd[x&0x0f]
|
||||
}
|
||||
return string(out)
|
||||
}
|
||||
|
||||
func sha256Sum(b []byte) []byte {
|
||||
h := sha256.Sum256(b)
|
||||
return h[:]
|
||||
}
|
||||
@@ -0,0 +1,216 @@
|
||||
// Package signedjobs is the slice-10B consumer of the hub's signed-jobs queue: it fetches each
|
||||
// opaque queued blob, runs it through the reversibility gate (the LOCKED authz pipeline: pinned-key
|
||||
// SSHSIG → namespace → allow-list-by-key-material → crypto → host target → time window → durable
|
||||
// nonce-burn), and ONLY on all-pass hands the verified+bound op to an Executor. The order is
|
||||
// verify → burn nonce (durable, inside Verify) → execute → report-completion, so no destructive op
|
||||
// runs on an unsigned, non-pinned-signer, replayed, expired, retargeted, or path-only blob, and an
|
||||
// interrupted op can't be re-authorized by replaying the same blob after a crash.
|
||||
package signedjobs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sync"
|
||||
|
||||
"gitea.dooplex.hu/admin/felhom-agent/internal/authz"
|
||||
"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
|
||||
"gitea.dooplex.hu/admin/felhom-agent/internal/reconcile"
|
||||
)
|
||||
|
||||
// Envelope is the queued signed-op job payload (the opaque bytes the hub stores per job). The
|
||||
// operator's signing CLI (cmd/felhom-opsign) produces it; this runner consumes it. The hub never
|
||||
// forges one (it holds no signing key) and never opens it.
|
||||
type Envelope struct {
|
||||
OpBlobB64 string `json:"op_blob_b64"` // base64 of the canonical OpBlob JSON
|
||||
SigArmored string `json:"sig_armored"` // armored SSHSIG over the op-blob (namespace felhom-op-v1)
|
||||
}
|
||||
|
||||
// JobSource fetches + clears jobs. Satisfied by *hub.Client.
|
||||
type JobSource interface {
|
||||
Jobs(ctx context.Context) ([]hub.JobWire, error)
|
||||
CompleteJob(ctx context.Context, jobID string) error
|
||||
}
|
||||
|
||||
// Authorizer is the reversibility gate. Satisfied by *reconcile.Gate.
|
||||
type Authorizer interface {
|
||||
Authorize(intent reconcile.Intent, signed *reconcile.SignedOp) reconcile.Decision
|
||||
}
|
||||
|
||||
// Executor runs a verified+bound destructive op. 10B wires the storage-wipe executor; other op
|
||||
// classes (guest_destroy, decommission, restore_overwrite — 10D) plug in per-slice. The executor
|
||||
// owns the RESOURCE-level anti-retarget (durable-id resolution + re-inspection at execution).
|
||||
type Executor interface {
|
||||
// Execute runs the op named by `op` with the VERIFIED canonical params. ErrNoExecutor means
|
||||
// "this op has no executor in this build" (left in the queue for a later slice).
|
||||
Execute(ctx context.Context, op string, params json.RawMessage) error
|
||||
}
|
||||
|
||||
// ErrNoExecutor signals an op class with no executor wired in this build (don't clear the job).
|
||||
var ErrNoExecutor = fmt.Errorf("signedjobs: no executor for this op class in this build")
|
||||
|
||||
// Runner polls the jobs queue and drives verify→execute for each job. Single-flight: a RunOnce
|
||||
// already in progress short-circuits a concurrent trigger.
|
||||
type Runner struct {
|
||||
source JobSource
|
||||
gate Authorizer
|
||||
exec Executor
|
||||
hostID string
|
||||
logger *slog.Logger
|
||||
|
||||
mu sync.Mutex
|
||||
running bool
|
||||
}
|
||||
|
||||
// NewRunner builds a runner. All deps are required.
|
||||
func NewRunner(source JobSource, gate Authorizer, exec Executor, hostID string, logger *slog.Logger) *Runner {
|
||||
if logger == nil {
|
||||
logger = slog.Default()
|
||||
}
|
||||
return &Runner{source: source, gate: gate, exec: exec, hostID: hostID, logger: logger}
|
||||
}
|
||||
|
||||
// OnEnvelope implements hub.EnvelopeObserver: when the heartbeat flags pending signed ops, run a
|
||||
// pass (off the heartbeat path so a long mkfs never blocks the loop). Single-flight.
|
||||
func (r *Runner) OnEnvelope(ctx context.Context, env *hub.ControlEnvelope) {
|
||||
if env == nil || !env.HasSignedOps {
|
||||
return
|
||||
}
|
||||
go func() {
|
||||
if _, err := r.RunOnce(context.Background()); err != nil {
|
||||
r.logger.Warn("signedjobs: pass failed", "err", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// RunOnce fetches the pending jobs and processes each. Returns the number of jobs PROCESSED
|
||||
// (executed or terminally rejected/cleared). A fetch error returns early; per-job errors are
|
||||
// logged and do not abort the pass.
|
||||
func (r *Runner) RunOnce(ctx context.Context) (int, error) {
|
||||
r.mu.Lock()
|
||||
if r.running {
|
||||
r.mu.Unlock()
|
||||
return 0, nil // a pass is already in progress
|
||||
}
|
||||
r.running = true
|
||||
r.mu.Unlock()
|
||||
defer func() { r.mu.Lock(); r.running = false; r.mu.Unlock() }()
|
||||
|
||||
jobs, err := r.source.Jobs(ctx)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("signedjobs: fetch jobs: %w", err)
|
||||
}
|
||||
processed := 0
|
||||
for _, j := range jobs {
|
||||
if r.processJob(ctx, j) {
|
||||
processed++
|
||||
}
|
||||
}
|
||||
return processed, nil
|
||||
}
|
||||
|
||||
// processJob handles one job; returns true if it was terminally handled (cleared from the queue).
|
||||
func (r *Runner) processJob(ctx context.Context, j hub.JobWire) bool {
|
||||
signed, ob, ok := r.decode(j)
|
||||
if !ok {
|
||||
// A malformed/forged envelope is permanently bad — clear it loudly (a queued garbage job
|
||||
// must not wedge the queue). No nonce was consumed (we never reached Verify).
|
||||
r.logger.Error("signedjobs: malformed job envelope — clearing", "job", j.JobID)
|
||||
r.complete(ctx, j.JobID)
|
||||
return true
|
||||
}
|
||||
|
||||
intent := reconcile.Intent{
|
||||
Class: reconcile.OpClass(ob.Op),
|
||||
HostID: ob.Target.HostID,
|
||||
GuestID: ob.Target.GuestID,
|
||||
ParamsJSON: ob.Params,
|
||||
Source: reconcile.SourceOneShotJob,
|
||||
// Provenance is the zero value (no agent-internal evidence) — a hub-sourced op is
|
||||
// classified by its op class alone; a destructive class MUST carry a valid signature.
|
||||
}
|
||||
dec := r.gate.Authorize(intent, signed)
|
||||
if !dec.Allowed {
|
||||
// The signature/binding failed (forged, replayed, expired, retargeted, role-denied, …).
|
||||
// The executor is NEVER called. This blob can never become valid, so clear it — but LOG
|
||||
// it as a security event (a compromised hub queuing a forged op lands here).
|
||||
r.logger.Warn("signedjobs: REJECTED signed op — executor not called",
|
||||
"job", j.JobID, "op", ob.Op, "reason", dec.Reason, "err", errStr(dec.Err))
|
||||
r.complete(ctx, j.JobID)
|
||||
return true
|
||||
}
|
||||
|
||||
// Allowed: the nonce is already durably burned (Verify, before this point). Execute.
|
||||
r.logger.Warn("signedjobs: AUTHORIZED signed op — executing",
|
||||
"job", j.JobID, "op", ob.Op, "key_id", dec.Verified.KeyID, "nonce", dec.Verified.Nonce)
|
||||
err := r.exec.Execute(ctx, ob.Op, ob.Params)
|
||||
switch {
|
||||
case err == nil:
|
||||
r.logger.Warn("signedjobs: signed op COMPLETED", "job", j.JobID, "op", ob.Op)
|
||||
r.complete(ctx, j.JobID)
|
||||
return true
|
||||
case errorsIs(err, ErrNoExecutor):
|
||||
// No executor for this op class in this build — leave it queued for the owning slice.
|
||||
r.logger.Info("signedjobs: no executor for op (left queued)", "job", j.JobID, "op", ob.Op)
|
||||
return false
|
||||
default:
|
||||
// Execution failed AFTER a passing verify — the nonce is spent, so re-running the same
|
||||
// blob would be rejected as a replay anyway. Clear it (the operator re-signs with a fresh
|
||||
// nonce to retry) and log the failure loudly.
|
||||
r.logger.Error("signedjobs: signed op execution FAILED (nonce spent — clearing)",
|
||||
"job", j.JobID, "op", ob.Op, "err", err)
|
||||
r.complete(ctx, j.JobID)
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
// decode unpacks a JobWire → SignedOp{op-blob, armored sig} + the parsed OpBlob (for routing).
|
||||
func (r *Runner) decode(j hub.JobWire) (*reconcile.SignedOp, authz.OpBlob, bool) {
|
||||
raw, err := base64.StdEncoding.DecodeString(j.BlobB64)
|
||||
if err != nil {
|
||||
return nil, authz.OpBlob{}, false
|
||||
}
|
||||
var env Envelope
|
||||
if err := json.Unmarshal(raw, &env); err != nil || env.OpBlobB64 == "" || env.SigArmored == "" {
|
||||
return nil, authz.OpBlob{}, false
|
||||
}
|
||||
opBlob, err := base64.StdEncoding.DecodeString(env.OpBlobB64)
|
||||
if err != nil {
|
||||
return nil, authz.OpBlob{}, false
|
||||
}
|
||||
var ob authz.OpBlob
|
||||
if err := json.Unmarshal(opBlob, &ob); err != nil || ob.Op == "" {
|
||||
return nil, authz.OpBlob{}, false
|
||||
}
|
||||
return &reconcile.SignedOp{Blob: opBlob, Sig: []byte(env.SigArmored)}, ob, true
|
||||
}
|
||||
|
||||
func (r *Runner) complete(ctx context.Context, jobID string) {
|
||||
if err := r.source.CompleteJob(ctx, jobID); err != nil {
|
||||
r.logger.Warn("signedjobs: clearing job failed (will retry next pass)", "job", jobID, "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
func errStr(e error) string {
|
||||
if e == nil {
|
||||
return ""
|
||||
}
|
||||
return e.Error()
|
||||
}
|
||||
|
||||
func errorsIs(err, target error) bool {
|
||||
for err != nil {
|
||||
if err == target {
|
||||
return true
|
||||
}
|
||||
type w interface{ Unwrap() error }
|
||||
if u, ok := err.(w); ok {
|
||||
err = u.Unwrap()
|
||||
} else {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,240 @@
|
||||
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")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
package signedjobs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
|
||||
"gitea.dooplex.hu/admin/felhom-agent/internal/storage"
|
||||
)
|
||||
|
||||
// opStorageWipe is the op class this executor serves (mirrors reconcile.ClassStorageWipe; the
|
||||
// literal avoids importing reconcile here just for the string).
|
||||
const opStorageWipe = "storage_wipe"
|
||||
|
||||
// WipeOps is the privileged storage surface the wipe executor needs (a subset of storage.HostOps).
|
||||
type WipeOps interface {
|
||||
InspectDevice(ctx context.Context, device string) (storage.DeviceProbe, error)
|
||||
Format(ctx context.Context, device, fstype string) error
|
||||
}
|
||||
|
||||
// wipeParams is the verified params of a storage_wipe op. The device is named by a DURABLE id
|
||||
// (never a mutable /dev path), so execution re-resolves it to the exact physical device.
|
||||
type wipeParams struct {
|
||||
DurableID string `json:"durable_id"`
|
||||
FSType string `json:"fstype"`
|
||||
}
|
||||
|
||||
// WipeExecutor is the slice-10B storage-wipe consumer — it CLOSES the 8C data-bearing-format
|
||||
// `pending_signature` gap. Given a gate-VERIFIED+bound storage_wipe op, it:
|
||||
// 1. resolves the op's DURABLE device id → the current /dev path (vanished/replaced → refuse);
|
||||
// 2. re-derives that device's durable id and requires it to MATCH the signed id (resource-level
|
||||
// anti-retarget — "wipe device X" wipes exactly X, not whatever is at /dev/sdb now);
|
||||
// 3. re-inspects (8C classifier) to confirm it is STILL the data-bearing target;
|
||||
// 4. mkfs.
|
||||
//
|
||||
// A path-only binding, a device that vanished/changed, or a target that is no longer data-bearing
|
||||
// is refused EVEN WITH a valid signature. The gate has already burned the nonce durably before we
|
||||
// run, so an interrupted wipe can't be re-authorized by replaying the blob.
|
||||
type WipeExecutor struct {
|
||||
ops WipeOps
|
||||
resolve func(durableID string) (string, error)
|
||||
derive func(device string) (string, error)
|
||||
logger *slog.Logger
|
||||
}
|
||||
|
||||
// NewWipeExecutor wires the production durable resolver/deriver (storage package).
|
||||
func NewWipeExecutor(ops WipeOps, logger *slog.Logger) *WipeExecutor {
|
||||
if logger == nil {
|
||||
logger = slog.Default()
|
||||
}
|
||||
return &WipeExecutor{
|
||||
ops: ops,
|
||||
resolve: storage.ResolveDurableDevice,
|
||||
derive: storage.DeviceDurableID,
|
||||
logger: logger,
|
||||
}
|
||||
}
|
||||
|
||||
// Execute implements signedjobs.Executor for the storage_wipe op class.
|
||||
func (w *WipeExecutor) Execute(ctx context.Context, op string, params json.RawMessage) error {
|
||||
if op != opStorageWipe {
|
||||
return ErrNoExecutor // not ours — the runner leaves it queued for the owning slice
|
||||
}
|
||||
var p wipeParams
|
||||
if err := json.Unmarshal(params, &p); err != nil {
|
||||
return fmt.Errorf("wipe: bad params: %w", err)
|
||||
}
|
||||
if p.DurableID == "" {
|
||||
// The whole point of the durable binding: a path-only op is refused (anti-retarget).
|
||||
return fmt.Errorf("wipe: op has no durable_id — refusing a path-only wipe binding")
|
||||
}
|
||||
|
||||
// 1. Resolve the durable id → current device (gone/replaced → refuse).
|
||||
device, err := w.resolve(p.DurableID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("wipe: durable id %q no longer resolves (device removed/replaced?) — refusing: %w", p.DurableID, err)
|
||||
}
|
||||
// 2. Anti-retarget: re-derive the resolved device's durable id; it MUST equal the signed id.
|
||||
got, err := w.derive(device)
|
||||
if err != nil {
|
||||
return fmt.Errorf("wipe: cannot re-derive durable id for %s — refusing: %w", device, err)
|
||||
}
|
||||
if got != p.DurableID {
|
||||
return fmt.Errorf("wipe: durable-id mismatch — %s now has id %q, signed id was %q — refusing", device, got, p.DurableID)
|
||||
}
|
||||
// 3. Re-inspect (8C classifier) — confirm the resolved device is STILL the data-bearing target.
|
||||
probe, err := w.ops.InspectDevice(ctx, device)
|
||||
if err != nil {
|
||||
return fmt.Errorf("wipe: re-inspect %s failed — refusing: %w", device, err)
|
||||
}
|
||||
if !probe.Probed {
|
||||
return fmt.Errorf("wipe: %s did not probe cleanly at execution — refusing", device)
|
||||
}
|
||||
if !probe.DataBearing() {
|
||||
// The signed op authorized wiping a DATA-BEARING device; if it is now blank, the target
|
||||
// changed since signing — refuse rather than wipe the wrong (or an unexpected) device.
|
||||
return fmt.Errorf("wipe: %s is no longer data-bearing (target changed since signing) — refusing", device)
|
||||
}
|
||||
|
||||
// 4. Execute. The nonce was durably burned by the gate's Verify BEFORE this point (crash-safety:
|
||||
// a replay after an interrupted wipe is rejected). This log line is the audit trail of a
|
||||
// destructive, operator-authorized wipe actually running.
|
||||
w.logger.Warn("wipe: executing operator-signed data-bearing wipe",
|
||||
"device", device, "durable_id", p.DurableID, "fstype", p.FSType, "data_reason", probe.Reason())
|
||||
if err := w.ops.Format(ctx, device, p.FSType); err != nil {
|
||||
return fmt.Errorf("wipe: mkfs %s %s: %w", p.FSType, device, err)
|
||||
}
|
||||
w.logger.Warn("wipe: operator-signed wipe complete", "device", device, "durable_id", p.DurableID)
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,137 @@
|
||||
package signedjobs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"gitea.dooplex.hu/admin/felhom-agent/internal/storage"
|
||||
)
|
||||
|
||||
// fakeWipeOps records Format calls and returns a configurable InspectDevice probe.
|
||||
type fakeWipeOps struct {
|
||||
probe storage.DeviceProbe
|
||||
inspectErr error
|
||||
formatCalls []string
|
||||
}
|
||||
|
||||
func (f *fakeWipeOps) InspectDevice(_ context.Context, device string) (storage.DeviceProbe, error) {
|
||||
p := f.probe
|
||||
p.Device = device
|
||||
return p, f.inspectErr
|
||||
}
|
||||
func (f *fakeWipeOps) Format(_ context.Context, device, _ string) error {
|
||||
f.formatCalls = append(f.formatCalls, device)
|
||||
return nil
|
||||
}
|
||||
|
||||
// newWipeExec builds a WipeExecutor with injected resolve/derive (so durable resolution is
|
||||
// deterministic in a unit test, no real /dev needed).
|
||||
func newWipeExec(ops WipeOps, resolve func(string) (string, error), derive func(string) (string, error)) *WipeExecutor {
|
||||
w := NewWipeExecutor(ops, quiet())
|
||||
w.resolve = resolve
|
||||
w.derive = derive
|
||||
return w
|
||||
}
|
||||
|
||||
const durable = "byid:wwn-0xtest"
|
||||
|
||||
func params(t *testing.T, durableID, fstype string) json.RawMessage {
|
||||
t.Helper()
|
||||
b, _ := json.Marshal(map[string]string{"durable_id": durableID, "fstype": fstype})
|
||||
return b
|
||||
}
|
||||
|
||||
// VALID: durable resolves, re-derives to the same id, device is data-bearing → Format runs.
|
||||
func TestWipe_ValidExecutes(t *testing.T) {
|
||||
ops := &fakeWipeOps{probe: storage.DeviceProbe{Probed: true, HasFilesystem: true, FSType: "ext4"}}
|
||||
w := newWipeExec(ops,
|
||||
func(id string) (string, error) { return "/dev/sdb", nil },
|
||||
func(dev string) (string, error) { return durable, nil })
|
||||
if err := w.Execute(context.Background(), "storage_wipe", params(t, durable, "ext4")); err != nil {
|
||||
t.Fatalf("valid wipe: %v", err)
|
||||
}
|
||||
if len(ops.formatCalls) != 1 || ops.formatCalls[0] != "/dev/sdb" {
|
||||
t.Fatalf("Format calls = %v, want one on /dev/sdb", ops.formatCalls)
|
||||
}
|
||||
}
|
||||
|
||||
// PATH-ONLY (no durable_id) → refused, Format NOT called.
|
||||
func TestWipe_PathOnlyRefused(t *testing.T) {
|
||||
ops := &fakeWipeOps{probe: storage.DeviceProbe{Probed: true, HasFilesystem: true}}
|
||||
w := newWipeExec(ops, func(string) (string, error) { return "/dev/sdb", nil }, func(string) (string, error) { return durable, nil })
|
||||
// params carry a raw device path, NOT a durable id.
|
||||
p, _ := json.Marshal(map[string]string{"device": "/dev/sdb", "fstype": "ext4"})
|
||||
if err := w.Execute(context.Background(), "storage_wipe", p); err == nil {
|
||||
t.Fatal("path-only wipe must be refused")
|
||||
}
|
||||
if len(ops.formatCalls) != 0 {
|
||||
t.Errorf("Format was called on a path-only binding: %v", ops.formatCalls)
|
||||
}
|
||||
}
|
||||
|
||||
// DURABLE MISMATCH: the resolved device re-derives to a DIFFERENT id → refused (anti-retarget).
|
||||
func TestWipe_DurableMismatchRefused(t *testing.T) {
|
||||
ops := &fakeWipeOps{probe: storage.DeviceProbe{Probed: true, HasFilesystem: true}}
|
||||
w := newWipeExec(ops,
|
||||
func(string) (string, error) { return "/dev/sdb", nil },
|
||||
func(string) (string, error) { return "byid:wwn-0xDIFFERENT", nil }) // re-derive disagrees
|
||||
if err := w.Execute(context.Background(), "storage_wipe", params(t, durable, "ext4")); err == nil {
|
||||
t.Fatal("a durable-id mismatch must be refused")
|
||||
}
|
||||
if len(ops.formatCalls) != 0 {
|
||||
t.Errorf("Format was called despite a durable-id mismatch: %v", ops.formatCalls)
|
||||
}
|
||||
}
|
||||
|
||||
// DEVICE GONE: the durable id no longer resolves → refused (device removed/replaced).
|
||||
func TestWipe_ResolveFailureRefused(t *testing.T) {
|
||||
ops := &fakeWipeOps{probe: storage.DeviceProbe{Probed: true, HasFilesystem: true}}
|
||||
w := newWipeExec(ops,
|
||||
func(string) (string, error) { return "", errors.New("no such by-id link") },
|
||||
func(string) (string, error) { return durable, nil })
|
||||
if err := w.Execute(context.Background(), "storage_wipe", params(t, durable, "ext4")); err == nil {
|
||||
t.Fatal("an unresolvable durable id must be refused")
|
||||
}
|
||||
if len(ops.formatCalls) != 0 {
|
||||
t.Errorf("Format was called on an unresolvable device: %v", ops.formatCalls)
|
||||
}
|
||||
}
|
||||
|
||||
// RE-INSPECT: the resolved device is no longer data-bearing (target changed) → refused.
|
||||
func TestWipe_ReinspectNonDataBearingRefused(t *testing.T) {
|
||||
ops := &fakeWipeOps{probe: storage.DeviceProbe{Probed: true /* blank: no fs/parts */}}
|
||||
w := newWipeExec(ops,
|
||||
func(string) (string, error) { return "/dev/sdb", nil },
|
||||
func(string) (string, error) { return durable, nil })
|
||||
if err := w.Execute(context.Background(), "storage_wipe", params(t, durable, "ext4")); err == nil {
|
||||
t.Fatal("a device that is no longer data-bearing must be refused (target changed)")
|
||||
}
|
||||
if len(ops.formatCalls) != 0 {
|
||||
t.Errorf("Format was called on a non-target device: %v", ops.formatCalls)
|
||||
}
|
||||
}
|
||||
|
||||
// RE-INSPECT: the device did not probe cleanly → refused (fail-safe).
|
||||
func TestWipe_NotProbedRefused(t *testing.T) {
|
||||
ops := &fakeWipeOps{probe: storage.DeviceProbe{Probed: false}}
|
||||
w := newWipeExec(ops,
|
||||
func(string) (string, error) { return "/dev/sdb", nil },
|
||||
func(string) (string, error) { return durable, nil })
|
||||
if err := w.Execute(context.Background(), "storage_wipe", params(t, durable, "ext4")); err == nil {
|
||||
t.Fatal("a device that did not probe cleanly must be refused")
|
||||
}
|
||||
if len(ops.formatCalls) != 0 {
|
||||
t.Errorf("Format was called on an unprobed device: %v", ops.formatCalls)
|
||||
}
|
||||
}
|
||||
|
||||
// A non-wipe op → ErrNoExecutor (left for the owning slice; Format not called).
|
||||
func TestWipe_OtherOpIsNoExecutor(t *testing.T) {
|
||||
ops := &fakeWipeOps{}
|
||||
w := newWipeExec(ops, func(string) (string, error) { return "/dev/sdb", nil }, func(string) (string, error) { return durable, nil })
|
||||
if err := w.Execute(context.Background(), "guest_destroy", params(t, durable, "ext4")); !errors.Is(err, ErrNoExecutor) {
|
||||
t.Fatalf("guest_destroy err = %v, want ErrNoExecutor", err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user