Files
felhom-agent/internal/capability/probe_test.go
T
admin 25024d9dda capability: agent privileged-capability self-probe (manifest + build-test + runtime snapshot) v0.44.0
New internal/capability: Manifest of required sudo -n grants + Prober that LISTS each
via 'sudo -n -l' (never executes) + binary-exists check → ok/degraded snapshot on the hub
report. Build-time test asserts manifest⊆sudoers (red-proof: dropping lxc-info FAILs the
gate). Startup logs N/N ok + ERROR per degraded. Serve-degraded; no allowlist change.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EPZ4GJ8L5Jqf8UiPwbn1kt
2026-06-29 18:43:49 +02:00

118 lines
4.2 KiB
Go

package capability
import (
"context"
"errors"
"strings"
"testing"
)
// fakeRunner returns a canned error per (command line) and records calls. deny holds binaries (or
// the bare "sudo -n -l" preflight) that should fail; everything else exits 0.
type fakeRunner struct {
preflightErr error
denyBinary map[string]bool // binary path → policy-denied
calls int
executedReal bool // set if a probed command was ever run WITHOUT -l (must never happen)
}
func (f *fakeRunner) Run(_ context.Context, name string, args ...string) ([]byte, []byte, error) {
f.calls++
// Preflight is `sudo -n -l` (exactly 2 args, no `--`).
if name == "sudo" && len(args) == 2 && args[0] == "-n" && args[1] == "-l" {
return nil, nil, f.preflightErr
}
// Every real probe must be a LIST: `sudo -n -l -- <binary> …`.
if name != "sudo" || len(args) < 4 || args[0] != "-n" || args[1] != "-l" || args[2] != "--" {
f.executedReal = true
return nil, nil, nil
}
binary := args[3]
if f.denyBinary[binary] {
return nil, nil, errors.New("sudo: a password is required")
}
return nil, nil, nil
}
func find(statuses []Status, name string) Status {
for _, s := range statuses {
if s.Name == name {
return s
}
}
return Status{}
}
// §7-A: all grants present + binaries exist → every capability ok.
func TestProbe_AllOK(t *testing.T) {
r := &fakeRunner{denyBinary: map[string]bool{}}
p := Prober{Runner: r, Exists: func(string) bool { return true }}
statuses := p.Probe(context.Background())
ok, total, degraded := Summarize(statuses)
if total != len(Manifest()) {
t.Fatalf("total=%d want %d", total, len(Manifest()))
}
if ok != total || len(degraded) != 0 {
t.Fatalf("expected all ok, got %d/%d (degraded: %+v)", ok, total, degraded)
}
if r.executedReal {
t.Fatal("probe executed a command without -l (must be list-only)")
}
}
// §7-B: one grant denied → that capability degraded "sudo policy denied", others ok. Serve-degraded.
func TestProbe_OneDenied(t *testing.T) {
r := &fakeRunner{denyBinary: map[string]bool{"/usr/bin/lxc-info": true}}
p := Prober{Runner: r, Exists: func(string) bool { return true }}
statuses := p.Probe(context.Background())
gi := find(statuses, "guest-init-pid")
if gi.Status != StatusDegraded || gi.Reason != "sudo policy denied" {
t.Fatalf("guest-init-pid = %+v, want degraded/sudo policy denied", gi)
}
if !gi.Critical {
t.Fatal("guest-init-pid should be Critical")
}
// A sibling stays ok.
if s := find(statuses, "drive-bind"); s.Status != StatusOK {
t.Fatalf("drive-bind = %+v, want ok", s)
}
ok, total, _ := Summarize(statuses)
if ok != total-1 {
t.Fatalf("expected exactly one degraded, got ok=%d total=%d", ok, total)
}
}
// §7-D: binary missing but policy granted → degraded "binary not found".
func TestProbe_BinaryMissing(t *testing.T) {
r := &fakeRunner{denyBinary: map[string]bool{}}
p := Prober{Runner: r, Exists: func(path string) bool { return path != "/usr/bin/lxc-info" }}
statuses := p.Probe(context.Background())
gi := find(statuses, "guest-init-pid")
if gi.Status != StatusDegraded || gi.Reason != "binary not found" {
t.Fatalf("guest-init-pid = %+v, want degraded/binary not found", gi)
}
}
// §8 aggregate: sudo itself unavailable for the user → ONE aggregate degraded, not N.
func TestProbe_SudoUnavailableAggregates(t *testing.T) {
r := &fakeRunner{preflightErr: errors.New("Sorry, user felhom-agent may not run sudo"), denyBinary: map[string]bool{}}
p := Prober{Runner: r, Exists: func(string) bool { return true }}
statuses := p.Probe(context.Background())
if len(statuses) != 1 {
t.Fatalf("expected 1 aggregate status, got %d", len(statuses))
}
s := statuses[0]
if s.Name != "sudo" || s.Status != StatusDegraded || !s.Critical || !strings.Contains(s.Reason, "drop-in not installed") {
t.Fatalf("aggregate = %+v, want critical degraded sudo-unavailable", s)
}
}
// Probe must never raise — even with a nil runner (e.g. a dev path) it returns statuses.
func TestProbe_NilRunnerNoPanic(t *testing.T) {
p := Prober{Runner: nil, Exists: func(string) bool { return true }}
if got := len(p.Probe(context.Background())); got != len(Manifest()) {
t.Fatalf("nil-runner probe returned %d statuses, want %d", got, len(Manifest()))
}
}