diff --git a/internal/felhomsshd/belt.go b/internal/felhomsshd/belt.go index c61f945..49a17c1 100644 --- a/internal/felhomsshd/belt.go +++ b/internal/felhomsshd/belt.go @@ -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*\{([^}]*)\}`) diff --git a/internal/felhomsshd/belt_health_test.go b/internal/felhomsshd/belt_health_test.go index 031b17d..0d8c22e 100644 --- a/internal/felhomsshd/belt_health_test.go +++ b/internal/felhomsshd/belt_health_test.go @@ -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") } diff --git a/internal/felhomsshd/loop.go b/internal/felhomsshd/loop.go index 6ec3bb1..526314c 100644 --- a/internal/felhomsshd/loop.go +++ b/internal/felhomsshd/loop.go @@ -19,8 +19,9 @@ type Loop struct { interval time.Duration logger *slog.Logger - mu sync.Mutex - block *hub.WireWireguard + 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) }