fix(felhomsshd): don't empty the belt/authkeys on an unfetched block (operator lockout)

Mirrors wgtunnel fetched=false-is-never-a-teardown: until the desired-state is
delivered, @operator_ips + felhom-op authorized_keys are left untouched (a nil
block on agent restart would otherwise empty @operator_ips and lock the operator
out until the next fetch). Belt.Sync split into SyncPort (always) + SyncOperator
(fetched only).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PSK5g6qYLknKj8u3QAFEr6
This commit is contained in:
2026-07-05 22:51:30 +02:00
parent b99d02b7a9
commit 2bf9354c0e
3 changed files with 43 additions and 25 deletions
+15 -12
View File
@@ -32,12 +32,21 @@ func NewBelt(runner proxmox.Runner, logger *slog.Logger) *Belt {
return &Belt{runner: runner, logger: logger, table: "felhom_oob"}
}
// Sync converges @operator_ips + @ssh_port on the desired values. operatorIP "" empties @operator_ips
// (belt drops all tunnel SSH — OOB off). port 0 empties @ssh_port. Every value is validated before it
// reaches nft (netip for the IP, int range for the port), so the coarse sudoers wildcard can never be
// abused. A missing table (host-install not run / pre-H1 box) degrades to a single logged warning.
func (b *Belt) Sync(ctx context.Context, port int, operatorIP string) {
// desired sets
// SyncPort converges @ssh_port on the claimed port (0 empties it). Always safe to call — the port
// comes from the agent's own claim, not the hub desired-state.
func (b *Belt) SyncPort(ctx context.Context, port int) {
wantPorts := []string{}
if port > 0 && port <= 65535 {
wantPorts = []string{strconv.Itoa(port)}
}
b.syncSet(ctx, "ssh_port", wantPorts)
}
// SyncOperator converges @operator_ips on the operator /32 (operatorIP "" = OOB explicitly off →
// empty the set). Call ONLY when the desired-state has actually been FETCHED — a nil/unfetched block
// must NOT empty the set (that would lock the operator out until the next fetch, the wgtunnel
// fetched=false-is-never-a-teardown rule). Every value is netip-validated before it reaches nft.
func (b *Belt) SyncOperator(ctx context.Context, operatorIP string) {
wantOps := []string{}
if operatorIP != "" {
ip, err := netip.ParseAddr(operatorIP)
@@ -47,13 +56,7 @@ func (b *Belt) Sync(ctx context.Context, port int, operatorIP string) {
}
wantOps = []string{ip.String()}
}
wantPorts := []string{}
if port > 0 && port <= 65535 {
wantPorts = []string{strconv.Itoa(port)}
}
b.syncSet(ctx, "operator_ips", wantOps)
b.syncSet(ctx, "ssh_port", wantPorts)
}
var nftElemRe = regexp.MustCompile(`elements\s*=\s*\{([^}]*)\}`)
+5 -4
View File
@@ -88,7 +88,8 @@ func TestBelt_SyncMutatesThenIdempotent(t *testing.T) {
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")
b.SyncPort(context.Background(), 8822)
b.SyncOperator(context.Background(), "10.77.0.250")
// mutations happened: flush + add for BOTH sets
if r.count("nft", "flush") < 2 || r.count("nft", "add") < 2 {
@@ -103,7 +104,8 @@ func TestBelt_SyncMutatesThenIdempotent(t *testing.T) {
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")
b2.SyncPort(context.Background(), 8822)
b2.SyncOperator(context.Background(), "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"))
}
@@ -114,8 +116,7 @@ func TestBelt_OperatorEmptyEmptiesSet(t *testing.T) {
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}...
b.SyncOperator(context.Background(), "") // OOB explicitly off → operator set must be emptied
if r.count("nft", "flush") < 1 {
t.Fatal("emptying the operator set must flush it")
}
+21 -7
View File
@@ -20,6 +20,7 @@ type Loop struct {
logger *slog.Logger
mu sync.Mutex
fetched bool // a desired-state document has been delivered (false = never a teardown signal)
block *hub.WireWireguard
nudge chan struct{}
@@ -42,6 +43,7 @@ func (l *Loop) OnDesiredState(_ context.Context, resp *hub.DesiredStateResponse)
return
}
l.mu.Lock()
l.fetched = true
l.block = resp.DesiredState.Wireguard
l.mu.Unlock()
select {
@@ -50,10 +52,10 @@ func (l *Loop) OnDesiredState(_ context.Context, resp *hub.DesiredStateResponse)
}
}
func (l *Loop) snapshot() *hub.WireWireguard {
func (l *Loop) snapshot() (bool, *hub.WireWireguard) {
l.mu.Lock()
defer l.mu.Unlock()
return l.block
return l.fetched, l.block
}
// Run reconciles immediately, then on every tick or desired-state nudge, until ctx is cancelled.
@@ -72,15 +74,26 @@ func (l *Loop) Run(ctx context.Context) error {
}
}
// reconcile runs one full pass: instance apply → belt sync → health/heal.
// reconcile runs one full pass: instance apply → belt sync → health/heal. The `fetched` gate mirrors
// wgtunnel: until a desired-state document arrives, the operator inputs (authorized_keys, @operator_ips)
// are LEFT UNTOUCHED — an unfetched block must never empty the belt (operator lockout) or wipe the key.
func (l *Loop) reconcile(ctx context.Context) {
block := l.snapshot()
port, err := l.mgr.Apply(ctx, block)
fetched, block := l.snapshot()
// Pass the block to Apply only when fetched, so authorized_keys is applied only from real desired
// state (a nil block on Apply skips the authorized_keys write, leaving the existing file).
applyBlock := block
if !fetched {
applyBlock = nil
}
port, err := l.mgr.Apply(ctx, applyBlock)
if err != nil {
return // Apply logged; a claim/exhaustion or install error — retry next tick
}
if l.belt != nil {
l.belt.Sync(ctx, port, oobPeerIP(block))
l.belt.SyncPort(ctx, port) // always — the port is the agent's own claim
if fetched {
l.belt.SyncOperator(ctx, oobPeerIP(block)) // only from real desired state
}
}
l.mgr.HealAndCheck(ctx, port)
}
@@ -95,5 +108,6 @@ func oobPeerIP(block *hub.WireWireguard) string {
// OOBStatus implements the hub collector's reporter seam (Part 4).
func (l *Loop) OOBStatus(ctx context.Context) *hub.OOBStatus {
return l.mgr.Status(ctx, l.snapshot())
_, block := l.snapshot()
return l.mgr.Status(ctx, block)
}