135 lines
4.2 KiB
Go
135 lines
4.2 KiB
Go
package felhomsshd
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"sync"
|
|
"time"
|
|
|
|
"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
|
|
)
|
|
|
|
// Loop drives the felhom-sshd Manager on its own cadence (the wgtunnel/lanresolver shape) and
|
|
// consumes the hub desired-state's wireguard block via the desired.Syncer raw-consumer seam (for
|
|
// oob_peer_ip → the belt, and oob_operator_ssh_key → authorized_keys). Each tick: claim/render/reload
|
|
// the instance, sync the belt sets, and run the health/heal check.
|
|
type Loop struct {
|
|
mgr *Manager
|
|
belt *Belt // Part 3 (nil-safe: no belt sync when unset)
|
|
interval time.Duration
|
|
logger *slog.Logger
|
|
|
|
mu sync.Mutex
|
|
fetched bool // a desired-state document has been delivered (false = never a teardown signal)
|
|
block *hub.WireWireguard
|
|
// passed (R-1194) is set once the first reconcile pass has run. Until then the instance has never been rendered,
|
|
// so its health says nothing — and the first host report, sent before that pass, used to read „config invalid,
|
|
// port 0" and raise a false oob_degraded on EVERY fresh box (rehearsal 2026-10-20; reproduced 2026-10-21: the
|
|
// report at 20:34:00 read an absent config, the apply landed 0.4 s later).
|
|
passed bool
|
|
|
|
nudge chan struct{}
|
|
}
|
|
|
|
// NewLoop builds the loop. interval defaults to 60s. belt may be nil (belt sync skipped).
|
|
func NewLoop(mgr *Manager, belt *Belt, interval time.Duration, logger *slog.Logger) *Loop {
|
|
if interval <= 0 {
|
|
interval = 60 * time.Second
|
|
}
|
|
if logger == nil {
|
|
logger = slog.Default()
|
|
}
|
|
return &Loop{mgr: mgr, belt: belt, interval: interval, logger: logger, nudge: make(chan struct{}, 1)}
|
|
}
|
|
|
|
// OnDesiredState implements desired.RawConsumer: store the latest wireguard block and nudge.
|
|
func (l *Loop) OnDesiredState(_ context.Context, resp *hub.DesiredStateResponse) {
|
|
if resp == nil {
|
|
return
|
|
}
|
|
l.mu.Lock()
|
|
l.fetched = true
|
|
l.block = resp.DesiredState.Wireguard
|
|
l.mu.Unlock()
|
|
select {
|
|
case l.nudge <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
|
|
func (l *Loop) snapshot() (bool, *hub.WireWireguard) {
|
|
l.mu.Lock()
|
|
defer l.mu.Unlock()
|
|
return l.fetched, l.block
|
|
}
|
|
|
|
// Run reconciles immediately, then on every tick or desired-state nudge, until ctx is cancelled.
|
|
func (l *Loop) Run(ctx context.Context) error {
|
|
l.reconcile(ctx)
|
|
t := time.NewTicker(l.interval)
|
|
defer t.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-t.C:
|
|
case <-l.nudge:
|
|
}
|
|
l.reconcile(ctx)
|
|
}
|
|
}
|
|
|
|
// 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) {
|
|
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
|
|
}
|
|
defer l.markPassed()
|
|
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.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)
|
|
}
|
|
|
|
// oobPeerIP extracts the operator /32 source (bare IP) from the block, or "" when OOB is off.
|
|
func oobPeerIP(block *hub.WireWireguard) string {
|
|
if block == nil {
|
|
return ""
|
|
}
|
|
return block.OOBPeerIP
|
|
}
|
|
|
|
// markPassed records that a reconcile pass has run (R-1194) — after a failed Apply too, so a real, lasting failure
|
|
// is still reported from the next report on.
|
|
func (l *Loop) markPassed() {
|
|
l.mu.Lock()
|
|
l.passed = true
|
|
l.mu.Unlock()
|
|
}
|
|
|
|
// OOBStatus implements the hub collector's reporter seam (Part 4). nil — no stanza, which the hub does not evaluate —
|
|
// until the first reconcile pass has run (R-1194).
|
|
func (l *Loop) OOBStatus(ctx context.Context) *hub.OOBStatus {
|
|
l.mu.Lock()
|
|
passed := l.passed
|
|
l.mu.Unlock()
|
|
if !passed {
|
|
return nil
|
|
}
|
|
_, block := l.snapshot()
|
|
return l.mgr.Status(ctx, block)
|
|
}
|