diff --git a/internal/desired/consumer_test.go b/internal/desired/consumer_test.go new file mode 100644 index 0000000..74fb976 --- /dev/null +++ b/internal/desired/consumer_test.go @@ -0,0 +1,90 @@ +package desired + +// S3 Group C — the raw-consumer fan-out seam: called on generation advance, NOT on no-advance, +// and a panicking consumer is contained (the guest reconcile path must never break). + +import ( + "context" + "io" + "log/slog" + "sync" + "testing" + + "gitea.dooplex.hu/admin/felhom-agent/internal/hub" + "gitea.dooplex.hu/admin/felhom-agent/internal/reconcile" +) + +type recordingConsumer struct { + mu sync.Mutex + calls []*hub.DesiredStateResponse + panic bool +} + +func (r *recordingConsumer) OnDesiredState(_ context.Context, resp *hub.DesiredStateResponse) { + r.mu.Lock() + r.calls = append(r.calls, resp) + r.mu.Unlock() + if r.panic { + panic("consumer exploded") + } +} + +func (r *recordingConsumer) count() int { r.mu.Lock(); defer r.mu.Unlock(); return len(r.calls) } + +type stubFetcher struct{ resp *hub.DesiredStateResponse } + +func (s *stubFetcher) FetchDesiredState(context.Context) (*hub.DesiredStateResponse, error) { + return s.resp, nil +} + +func testResp(gen int64) *hub.DesiredStateResponse { + return &hub.DesiredStateResponse{ + Generation: gen, + DesiredState: hub.WireDesiredState{ + Guests: []hub.WireDesiredGuest{}, + Wireguard: &hub.WireWireguard{Pubkey: "PK", AssignedIP: "10.77.0.2/32"}, + }, + } +} + +func TestSyncer_ConsumerCalledOnAdvanceOnly(t *testing.T) { + provider := reconcile.NewCachingProvider() + f := &stubFetcher{resp: testResp(2)} + s := NewSyncer(f, provider, slog.New(slog.NewTextHandler(io.Discard, nil))) + c := &recordingConsumer{} + s.AddConsumer(c) + + // Advance → fetch → consumer called with the raw doc (wireguard block intact). + s.OnEnvelope(context.Background(), &hub.ControlEnvelope{DesiredGeneration: 2}) + if c.count() != 1 { + t.Fatalf("consumer calls = %d, want 1", c.count()) + } + if c.calls[0].DesiredState.Wireguard == nil || c.calls[0].DesiredState.Wireguard.Pubkey != "PK" { + t.Fatalf("consumer got %+v — the raw wireguard block must ride through", c.calls[0].DesiredState.Wireguard) + } + + // No advance → no fetch → no consumer call (the negative). + s.OnEnvelope(context.Background(), &hub.ControlEnvelope{DesiredGeneration: 2}) + if c.count() != 1 { + t.Errorf("consumer called without a generation advance: %d", c.count()) + } +} + +func TestSyncer_PanickingConsumerContained(t *testing.T) { + provider := reconcile.NewCachingProvider() + f := &stubFetcher{resp: testResp(1)} + s := NewSyncer(f, provider, slog.New(slog.NewTextHandler(io.Discard, nil))) + bomb := &recordingConsumer{panic: true} + after := &recordingConsumer{} + s.AddConsumer(bomb) + s.AddConsumer(after) + + // Must not panic out; the second consumer still runs; the provider still updated. + s.OnEnvelope(context.Background(), &hub.ControlEnvelope{DesiredGeneration: 1}) + if after.count() != 1 { + t.Errorf("consumer after the panicking one not called: %d", after.count()) + } + if provider.Generation() != 1 { + t.Errorf("provider generation = %d, want 1 (guest path unaffected)", provider.Generation()) + } +} diff --git a/internal/desired/syncer.go b/internal/desired/syncer.go index 3ce12f3..15eccd2 100644 --- a/internal/desired/syncer.go +++ b/internal/desired/syncer.go @@ -22,11 +22,28 @@ type Fetcher interface { FetchDesiredState(ctx context.Context) (*hub.DesiredStateResponse, error) } +// RawConsumer receives the FULL fetched desired-state document after each successful +// generation-advance fetch (S3 seam — internal/wgtunnel consumes its wireguard block this way +// without the reconcile engine learning about tunnels). Implementations must not block: do the +// cheap store-and-nudge, never network/exec inline. +type RawConsumer interface { + OnDesiredState(ctx context.Context, resp *hub.DesiredStateResponse) +} + // Syncer keeps the engine's CachingProvider in step with the hub's authoritative desired-state. type Syncer struct { - fetcher Fetcher - provider *reconcile.CachingProvider - logger *slog.Logger + fetcher Fetcher + provider *reconcile.CachingProvider + consumers []RawConsumer + logger *slog.Logger +} + +// AddConsumer registers a raw desired-state consumer (nil-safe no-op). Not concurrency-safe — +// call during wiring, before the hub loop starts. +func (s *Syncer) AddConsumer(c RawConsumer) { + if c != nil { + s.consumers = append(s.consumers, c) + } } // NewSyncer builds a Syncer over the hub fetcher and the engine's provider. @@ -61,12 +78,27 @@ func (s *Syncer) OnEnvelope(ctx context.Context, env *hub.ControlEnvelope) { s.provider.Update(resp.Generation, state) s.logger.Info("desired: updated from hub", "generation", resp.Generation, "guests", len(state.Guests)) + // S3: fan the raw document out to registered consumers (wgtunnel etc). A panicking consumer + // is contained — the guest reconcile path must never break over a tunnel add-on. + for _, c := range s.consumers { + s.notifyConsumer(ctx, c, resp) + } if env.HasSignedOps { // 10A only notes the flag; fetching + verifying + executing signed ops is slice 10B. s.logger.Info("desired: hub reports pending signed ops (fetch/execute is slice 10B)") } } +// notifyConsumer delivers one raw document with panic containment. +func (s *Syncer) notifyConsumer(ctx context.Context, c RawConsumer, resp *hub.DesiredStateResponse) { + defer func() { + if r := recover(); r != nil { + s.logger.Error("desired: raw consumer panicked (contained)", "panic", r) + } + }() + c.OnDesiredState(ctx, resp) +} + // mapWire maps the hub wire desired-state to the reconcile domain. 10A acts only on guests; the // forward-compat fields (restore_directive — 10D — etc.) are carried on the wire and logged, but // not translated into actions here. diff --git a/internal/hub/collect.go b/internal/hub/collect.go index d307102..2e54c2e 100644 --- a/internal/hub/collect.go +++ b/internal/hub/collect.go @@ -49,6 +49,12 @@ type PBSReporter interface { PBSSnapshots(ctx context.Context) []PBSSnapshot } +// WireguardReporter is the S3 seam the wgtunnel loop plugs into (same consumer-side pattern — +// hub does not import wgtunnel). nil (feature disabled) → no wireguard stanza on the report. +type WireguardReporter interface { + WireguardStatus(ctx context.Context) *WireguardStatus +} + // Collector builds a HostReport from read-only sources. All deps are behind narrow // interfaces for unit testing. type Collector struct { @@ -61,6 +67,7 @@ type Collector struct { temp TempReader // slice 9: host CPU/chassis temp (nil-safe → nil temp) capProbe func(ctx context.Context) []capability.Status // v0.44.0: privileged-capability self-check (nil → empty) leafFP string // v0.48.0: served local-API leaf fp (static per process; "" when local API disabled) + wg WireguardReporter // S3: offsite-tunnel status (nil → stanza omitted) hostID string agentVersion string logger *slog.Logger @@ -110,6 +117,13 @@ func (c *Collector) SetLeafFingerprint(fp string) *Collector { return c } +// SetWireguardReporter wires the offsite-tunnel status source (S3; nil-safe → stanza omitted). +// Returns the collector for chaining. +func (c *Collector) SetWireguardReporter(w WireguardReporter) *Collector { + c.wg = w + return c +} + // Collect builds the report. Best-effort liveness: a failed NodeStatus is a hard // error (no useful report — the cycle skips the POST); a failed per-guest // GuestConfig degrades that guest to status="unknown" without spec but still sends; @@ -143,6 +157,11 @@ func (c *Collector) Collect(ctx context.Context) (*HostReport, error) { // DR recipe host-half — derived from the just-collected guest/storage/PBS facts (no new reads). // Secret-free by construction (identifiers/intents/sizes/coordinates only). report.DRRecipe = BuildDRRecipeHostHalf(report.Guests, report.StorageTargets, report.PBSSnapshots) + // S3: offsite-tunnel status stanza (nil reporter = feature disabled → omitted; the pubkey in + // it is the operator's revocation-recovery handle). + if c.wg != nil { + report.Wireguard = c.wg.WireguardStatus(ctx) + } return report, nil } diff --git a/internal/wgtunnel/loop.go b/internal/wgtunnel/loop.go new file mode 100644 index 0000000..3546398 --- /dev/null +++ b/internal/wgtunnel/loop.go @@ -0,0 +1,82 @@ +package wgtunnel + +import ( + "context" + "log/slog" + "sync" + "time" + + "gitea.dooplex.hu/admin/felhom-agent/internal/hub" +) + +// Loop drives the Manager on its own cadence (the lanresolver shape) AND consumes fetched +// desired-state via the desired.Syncer raw-consumer seam. It distinguishes "no desired data +// seen yet" (fetched=false — never a teardown signal) from "desired-state present without the +// wireguard block" (revocation). +type Loop struct { + mgr *Manager + interval time.Duration + logger *slog.Logger + + mu sync.Mutex + fetched bool + block *hub.WireWireguard + + nudge chan struct{} +} + +// NewLoop builds the loop. interval defaults to 60s. +func NewLoop(mgr *Manager, interval time.Duration, logger *slog.Logger) *Loop { + if interval <= 0 { + interval = 60 * time.Second + } + if logger == nil { + logger = slog.Default() + } + return &Loop{mgr: mgr, interval: interval, logger: logger, nudge: make(chan struct{}, 1)} +} + +// OnDesiredState implements desired.RawConsumer: store the latest wireguard block (or its +// absence) and nudge the loop. Non-blocking and panic-free by construction. +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 applies immediately, then on every tick or desired-state nudge, until ctx is cancelled. +func (l *Loop) Run(ctx context.Context) error { + fetched, block := l.snapshot() + l.mgr.Apply(ctx, fetched, block) + t := time.NewTicker(l.interval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-t.C: + case <-l.nudge: + } + fetched, block = l.snapshot() + l.mgr.Apply(ctx, fetched, block) + } +} + +// WireguardStatus implements the hub collector's WireguardReporter seam. +func (l *Loop) WireguardStatus(ctx context.Context) *hub.WireguardStatus { + return l.mgr.Status(ctx) +} diff --git a/internal/wgtunnel/manager.go b/internal/wgtunnel/manager.go new file mode 100644 index 0000000..c3ef138 --- /dev/null +++ b/internal/wgtunnel/manager.go @@ -0,0 +1,439 @@ +package wgtunnel + +import ( + "context" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "encoding/json" + "fmt" + "log/slog" + "net/netip" + "os" + "os/exec" + "path/filepath" + "regexp" + "strconv" + "strings" + "sync" + "time" + + "gitea.dooplex.hu/admin/felhom-agent/internal/hub" + "gitea.dooplex.hu/admin/felhom-agent/internal/proxmox" +) + +const ( + // Iface is the managed interface/unit name: wg-quick@wg-felhom. + Iface = "wg-felhom" + // confDest is the installed conf path — must match the FELHOM_WG sudoers entry EXACTLY. + confDest = "/etc/wireguard/wg-felhom.conf" + // unit is the systemd unit the sudoers allowlists. + unit = "wg-quick@wg-felhom" + + markerName = "registered.json" + lastAppliedName = "last-applied.sha" + stagedConfName = "wg-felhom.conf" + + // registration backoff bounds (Scenario D: no hot loop). + regBackoffMin = time.Minute + regBackoffMax = 15 * time.Minute +) + +// Registrar is the hub seam (satisfied by *hub.Client; tests inject a fake). +type Registrar interface { + RegisterWG(ctx context.Context, pubkey string) (*hub.WGRegisterResponse, error) +} + +// marker is the local registration record (/wg/registered.json). Its EXISTENCE is the +// registration gate: present → never register again except the pubkey-mismatch re-key path. +// KEPT on revocation (revoked stays revoked — doc 06 §3.5 completion). +type marker struct { + Pubkey string `json:"pubkey"` + AssignedIP string `json:"assigned_ip"` + Generation int64 `json:"generation"` +} + +// Manager renders + applies the wg-felhom conf and drives wg-quick@wg-felhom through the +// narrow-sudoers runner (the lanresolver shape). All mutations go through `runner`; the only +// unprivileged execs are `systemctl is-active` (world-readable state) via isActive. +type Manager struct { + runner proxmox.Runner + hub Registrar + stateDir string + logger *slog.Logger + + // injectable for tests + isActive func(ctx context.Context) bool + wgPresent func() bool + now func() time.Time + + mu sync.Mutex + nextRegAt time.Time + regBackoff time.Duration + keyBroken bool // corrupt key / partial state — loop idles until operator resolves + brokenAnnounced bool +} + +// NewManager builds a Manager. stateDir is the agent state dir (default /var/lib/felhom-agent — +// note the FELHOM_WG sudoers install entry hard-codes the staged path under it). +func NewManager(runner proxmox.Runner, registrar Registrar, stateDir string, logger *slog.Logger) *Manager { + if logger == nil { + logger = slog.Default() + } + return &Manager{ + runner: runner, + hub: registrar, + stateDir: stateDir, + logger: logger, + isActive: func(ctx context.Context) bool { + out, _ := exec.CommandContext(ctx, "systemctl", "is-active", unit).Output() + return strings.TrimSpace(string(out)) == "active" + }, + wgPresent: func() bool { + _, err := os.Stat("/usr/bin/wg") + return err == nil + }, + now: time.Now, + } +} + +func (m *Manager) wgDir() string { return filepath.Join(m.stateDir, "wg") } +func (m *Manager) markerPath() string { return filepath.Join(m.wgDir(), markerName) } +func (m *Manager) lastAppliedPath() string { return filepath.Join(m.wgDir(), lastAppliedName) } +func (m *Manager) stagedConfPath() string { return filepath.Join(m.wgDir(), stagedConfName) } + +func (m *Manager) loadMarker() *marker { + raw, err := os.ReadFile(m.markerPath()) + if err != nil { + return nil + } + var mk marker + if json.Unmarshal(raw, &mk) != nil || mk.Pubkey == "" { + return nil + } + return &mk +} + +func (m *Manager) writeMarker(mk marker) error { + if err := os.MkdirAll(m.wgDir(), 0o700); err != nil { + return err + } + raw, _ := json.Marshal(mk) + return os.WriteFile(m.markerPath(), raw, 0o600) +} + +// --- conf rendering (pure; every value strictly validated before it reaches the file) --- + +var dnsNameRe = regexp.MustCompile(`^[A-Za-z0-9.-]+$`) + +func validKeyB64(s string) error { + if len(s) != 44 { + return fmt.Errorf("key must be 44 base64 chars, got %d", len(s)) + } + raw, err := base64.StdEncoding.DecodeString(s) + if err != nil { + return fmt.Errorf("key is not valid base64") + } + if len(raw) != 32 { + return fmt.Errorf("key decodes to %d bytes, want 32", len(raw)) + } + return nil +} + +// renderConf builds the wg-felhom.conf content from the hub block + the local private key. +// Client-side constants per doc 06 §4: MTU 1420, AllowedIPs = pbs_tunnel_ip/32 (the tunnel +// carries ONLY box→PBS traffic), PersistentKeepalive 25. All inputs validated — nothing +// user-controlled is interpolatable (strict charsets, netip parses). +func renderConf(block *hub.WireWireguard, privB64 string) (string, error) { + if err := validKeyB64(privB64); err != nil { + return "", fmt.Errorf("wgtunnel: private key: %w", err) + } + if err := validKeyB64(block.Endpoint.ServerPubkey); err != nil { + return "", fmt.Errorf("wgtunnel: server_pubkey: %w", err) + } + addr, err := netip.ParsePrefix(block.AssignedIP) + if err != nil || addr.Bits() != 32 { + return "", fmt.Errorf("wgtunnel: assigned_ip %q is not an ip/32", block.AssignedIP) + } + pbsIP, err := netip.ParseAddr(block.Endpoint.PBSTunnelIP) + if err != nil { + return "", fmt.Errorf("wgtunnel: pbs_tunnel_ip %q is not an address", block.Endpoint.PBSTunnelIP) + } + if !dnsNameRe.MatchString(block.Endpoint.DNSName) { + return "", fmt.Errorf("wgtunnel: endpoint dns_name has invalid characters") + } + if block.Endpoint.WGPort < 1 || block.Endpoint.WGPort > 65535 { + return "", fmt.Errorf("wgtunnel: wg_port %d out of range", block.Endpoint.WGPort) + } + + var b strings.Builder + b.WriteString("# felhom offsite tunnel — agent-managed (S3); DO NOT EDIT\n") + b.WriteString("[Interface]\n") + fmt.Fprintf(&b, "PrivateKey = %s\n", privB64) + fmt.Fprintf(&b, "Address = %s\n", block.AssignedIP) + b.WriteString("MTU = 1420\n\n") + b.WriteString("[Peer]\n") + fmt.Fprintf(&b, "PublicKey = %s\n", block.Endpoint.ServerPubkey) + fmt.Fprintf(&b, "Endpoint = %s:%d\n", block.Endpoint.DNSName, block.Endpoint.WGPort) + fmt.Fprintf(&b, "AllowedIPs = %s/32\n", pbsIP) + b.WriteString("PersistentKeepalive = 25\n") + return b.String(), nil +} + +// --- the state machine (doc 06 §3.3/§3.5 + spec §7/§8) --- + +// Apply reconciles local reality toward (fetched, block). fetched=false means "no desired-state +// data seen yet this process" — NEVER a teardown signal (teardown only on a PRESENT desired-state +// without the block). +func (m *Manager) Apply(ctx context.Context, fetched bool, block *hub.WireWireguard) { + m.mu.Lock() + defer m.mu.Unlock() + + keyExists := false + if _, err := os.Stat(KeyFilePath(m.stateDir)); err == nil { + keyExists = true + } + mk := m.loadMarker() + + // Partial state: marker without key. Never guess — a fresh keygen here would silently orphan + // the hub-registered identity. Operator repairs (delete marker → fresh keygen+register). + if mk != nil && !keyExists { + m.announceBroken("wgtunnel: registration marker exists but the key file is missing — idling until the operator resolves (delete the marker for a fresh keygen+register)") + return + } + + pub, created, err := EnsureKey(m.stateDir) + if err != nil { + m.announceBroken("wgtunnel: " + err.Error()) + return + } + m.keyBroken, m.brokenAnnounced = false, false + if created { + m.logger.Info("wgtunnel: generated new WG keypair", "pubkey", pub) + } + + if mk == nil { + // Not registered (by local knowledge). Adopt if the hub already knows this exact key — + // the lost-marker-kept-key case; no re-register. + if fetched && block != nil && block.Pubkey == pub { + if err := m.writeMarker(marker{Pubkey: pub, AssignedIP: block.AssignedIP}); err != nil { + m.logger.Error("wgtunnel: writing adoption marker", "err", err) + return + } + m.logger.Info("wgtunnel: adopted existing hub registration (marker restored)", "pubkey", pub) + mk = m.loadMarker() + } else { + // One-shot registration (backoff-gated). Registration is gated ONLY on marker + // absence: a revoked box (marker present, block gone) never lands here. + m.registerLocked(ctx, pub) + return + } + } + + if !fetched { + return // no desired data yet — keep last-applied state untouched + } + + if block == nil { + // PRESENT desired-state without the block = the hub revoked us. Stop + disable, KEEP the + // marker (revoked stays revoked; the operator re-adds the peer using the reported pubkey). + m.teardownLocked(ctx) + return + } + + if block.Pubkey != pub { + // The hub's block is for another key (DR from escrowed key on a re-provisioned box, or + // stale hub state) — the S2 re-key-in-place path. Bounded by the same backoff. + m.logger.Warn("wgtunnel: desired block pubkey differs from local key — re-registering (re-key-in-place)", + "local", pub, "desired", block.Pubkey) + m.registerLocked(ctx, pub) + return // the bumped generation brings a corrected block on the next fetch + } + + m.ensureTunnelLocked(ctx, block) +} + +// announceBroken logs a broken-state error ONCE (then idles quietly until state changes). +func (m *Manager) announceBroken(msg string) { + if !m.brokenAnnounced { + m.logger.Error(msg) + m.brokenAnnounced = true + } + m.keyBroken = true +} + +// registerLocked runs one backoff-gated RegisterWG and writes the marker on success. +func (m *Manager) registerLocked(ctx context.Context, pub string) { + if m.now().Before(m.nextRegAt) { + return + } + resp, err := m.hub.RegisterWG(ctx, pub) + if err != nil { + if m.regBackoff == 0 { + m.regBackoff = regBackoffMin + } else if m.regBackoff *= 2; m.regBackoff > regBackoffMax { + m.regBackoff = regBackoffMax + } + m.nextRegAt = m.now().Add(m.regBackoff) + m.logger.Warn("wgtunnel: registration failed — will retry", "err", err, "backoff", m.regBackoff) + return + } + m.regBackoff, m.nextRegAt = 0, time.Time{} + if err := m.writeMarker(marker{Pubkey: pub, AssignedIP: resp.AssignedIP, Generation: resp.Generation}); err != nil { + m.logger.Error("wgtunnel: registration succeeded but marker write failed", "err", err) + return + } + m.logger.Info("wgtunnel: registered with hub", "pubkey", pub, + "assigned_ip", resp.AssignedIP, "existed", resp.Existed, "generation", resp.Generation, "sync", resp.Sync) +} + +// teardownLocked disables the tunnel once (idempotent via the last-applied hash file + a live +// service check). NO re-registration follows — the marker stays. +func (m *Manager) teardownLocked(ctx context.Context) { + applied := false + if _, err := os.Stat(m.lastAppliedPath()); err == nil { + applied = true + } + if !applied && !m.isActive(ctx) { + return // already torn down — nothing to exec (Scenario C's quiet ticks) + } + m.logger.Warn("wgtunnel: desired-state no longer carries the wireguard block — hub revoked this peer; disabling the tunnel (marker kept; NO re-registration)") + if _, errOut, err := m.runner.Run(ctx, "systemctl", "disable", "--now", unit); err != nil { + m.logger.Error("wgtunnel: disable failed", "err", err, "stderr", strings.TrimSpace(string(errOut))) + return + } + os.Remove(m.lastAppliedPath()) +} + +// ensureTunnelLocked renders + applies the conf and keeps the service alive (Scenario B). +func (m *Manager) ensureTunnelLocked(ctx context.Context, block *hub.WireWireguard) { + priv, err := readPrivateKeyB64(m.stateDir) + if err != nil { + m.announceBroken("wgtunnel: " + err.Error()) + return + } + conf, err := renderConf(block, priv) + if err != nil { + m.logger.Error("wgtunnel: refusing to apply invalid desired block", "err", err) + return + } + sum := sha256.Sum256([]byte(conf)) + hash := hex.EncodeToString(sum[:]) + last, _ := os.ReadFile(m.lastAppliedPath()) + active := m.isActive(ctx) + + if string(last) == hash { + if active { + return // steady state: zero execs + } + // self-heal: conf is current but the service is down (crash/manual stop) + m.logger.Info("wgtunnel: service inactive with current conf — re-enabling (self-heal)") + if err := m.ensureToolsLocked(ctx); err != nil { + return + } + if _, errOut, err := m.runner.Run(ctx, "systemctl", "enable", "--now", unit); err != nil { + m.logger.Error("wgtunnel: enable failed", "err", err, "stderr", strings.TrimSpace(string(errOut))) + } + return + } + + // Conf changed (first apply, endpoint re-key, ip change): stage → install → enable/restart. + if err := m.ensureToolsLocked(ctx); err != nil { + return + } + if err := os.MkdirAll(m.wgDir(), 0o700); err != nil { + m.logger.Error("wgtunnel: state dir", "err", err) + return + } + if err := os.WriteFile(m.stagedConfPath(), []byte(conf), 0o600); err != nil { + m.logger.Error("wgtunnel: staging conf", "err", err) + return + } + // argv matches the FELHOM_WG sudoers entry exactly (fixed source + dest). + if _, errOut, err := m.runner.Run(ctx, "install", "-o", "root", "-g", "root", "-m", "0600", "--", m.stagedConfPath(), confDest); err != nil { + m.logger.Error("wgtunnel: conf install failed", "err", err, "stderr", strings.TrimSpace(string(errOut))) + return + } + verb := []string{"enable", "--now", unit} + if active { + verb = []string{"restart", unit} // conf change on a running tunnel → restart (never reload) + } + if _, errOut, err := m.runner.Run(ctx, "systemctl", verb...); err != nil { + m.logger.Error("wgtunnel: systemctl "+verb[0]+" failed", "err", err, "stderr", strings.TrimSpace(string(errOut))) + return + } + if err := os.WriteFile(m.lastAppliedPath(), []byte(hash), 0o600); err != nil { + m.logger.Error("wgtunnel: recording applied hash", "err", err) + return + } + m.logger.Info("wgtunnel: tunnel conf applied", "endpoint", + fmt.Sprintf("%s:%d", block.Endpoint.DNSName, block.Endpoint.WGPort), "assigned_ip", block.AssignedIP, "action", verb[0]) +} + +// ensureToolsLocked installs wireguard-tools once when absent (dnsmasq-install precedent). +func (m *Manager) ensureToolsLocked(ctx context.Context) error { + if m.wgPresent() { + return nil + } + m.logger.Info("wgtunnel: wireguard-tools absent — installing") + if out, errOut, err := m.runner.Run(ctx, "apt-get", "install", "-y", "-q", "wireguard-tools"); err != nil { + m.logger.Error("wgtunnel: apt-get install wireguard-tools failed", + "err", err, "out", strings.TrimSpace(string(errOut))+strings.TrimSpace(string(out))) + return err + } + return nil +} + +// Status builds the heartbeat report stanza (doc 06 §4.6). Read-only: it NEVER creates keys or +// registers. The handshake read is the package's single wg invocation — `latest-handshakes` +// only (NEVER `dump`, whose interface line carries the private key). +func (m *Manager) Status(ctx context.Context) *hub.WireguardStatus { + m.mu.Lock() + defer m.mu.Unlock() + + st := &hub.WireguardStatus{} + if raw, err := os.ReadFile(KeyFilePath(m.stateDir)); err == nil { + if priv, derr := decodeKey(raw); derr == nil { + if pub, perr := derivePublic(priv); perr == nil { + st.Pubkey = pub + } + } + } + if mk := m.loadMarker(); mk != nil { + st.Registered = true + st.AssignedIP = mk.AssignedIP + } + st.Active = m.isActive(ctx) + if st.Active { + if out, _, err := m.runner.Run(ctx, "wg", "show", Iface, "latest-handshakes"); err == nil { + if age, ok := parseHandshakeAge(string(out), m.now()); ok { + st.LastHandshakeAgeS = &age + } + } + } + if st.Pubkey == "" && !st.Registered && !st.Active { + return st // still a valid (empty-ish) stanza; caller decides whether to attach + } + return st +} + +// parseHandshakeAge parses `wg show latest-handshakes` output ("\t"). +// epoch 0 = no handshake yet → not ok (nil in the report; nil ≠ 0). +func parseHandshakeAge(out string, now time.Time) (int64, bool) { + for _, line := range strings.Split(strings.TrimSpace(out), "\n") { + f := strings.Fields(line) + if len(f) != 2 { + continue + } + epoch, err := strconv.ParseInt(f[1], 10, 64) + if err != nil || epoch <= 0 { + continue + } + age := now.Unix() - epoch + if age < 0 { + age = 0 + } + return age, true + } + return 0, false +} diff --git a/internal/wgtunnel/manager_test.go b/internal/wgtunnel/manager_test.go new file mode 100644 index 0000000..8ccde7d --- /dev/null +++ b/internal/wgtunnel/manager_test.go @@ -0,0 +1,449 @@ +package wgtunnel + +// Group B — the manager state machine (Scenarios A-D). Non-hollow: every test asserts EXEC +// COUNTS on the recording runner (the no-exec negatives are the load-bearing ones), single +// registration, marker semantics, and the absence of key material in the captured log buffer. + +import ( + "bytes" + "context" + "errors" + "io" + "log/slog" + "os" + "strings" + "sync" + "testing" + "time" + + "gitea.dooplex.hu/admin/felhom-agent/internal/hub" +) + +// recordingRunner records every privileged exec; scripted errors per command name. +type recordingRunner struct { + mu sync.Mutex + calls [][]string + fail map[string]error // command name → error +} + +func (r *recordingRunner) Run(_ context.Context, name string, args ...string) ([]byte, []byte, error) { + r.mu.Lock() + r.calls = append(r.calls, append([]string{name}, args...)) + err := r.fail[name] + r.mu.Unlock() + if err != nil { + return nil, []byte("scripted failure"), err + } + if name == "wg" { + return []byte("SERVERPUB\t1783107115\n"), nil, nil + } + return nil, nil, nil +} + +func (r *recordingRunner) RunStdin(ctx context.Context, _ io.Reader, name string, args ...string) ([]byte, []byte, error) { + return r.Run(ctx, name, args...) +} + +func (r *recordingRunner) count(name string) int { + r.mu.Lock() + defer r.mu.Unlock() + n := 0 + for _, c := range r.calls { + if c[0] == name { + n++ + } + } + return n +} + +func (r *recordingRunner) total() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.calls) +} + +func (r *recordingRunner) lastSystemctl() []string { + r.mu.Lock() + defer r.mu.Unlock() + for i := len(r.calls) - 1; i >= 0; i-- { + if r.calls[i][0] == "systemctl" { + return r.calls[i] + } + } + return nil +} + +// fakeHub scripts RegisterWG. +type fakeHub struct { + mu sync.Mutex + calls int + resp *hub.WGRegisterResponse + err error +} + +func (f *fakeHub) RegisterWG(_ context.Context, pubkey string) (*hub.WGRegisterResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls++ + if f.err != nil { + return nil, f.err + } + if f.resp != nil { + return f.resp, nil + } + return &hub.WGRegisterResponse{Pubkey: pubkey, AssignedIP: "10.77.0.2/32", Generation: 3, Sync: "ok"}, nil +} + +func (f *fakeHub) count() int { f.mu.Lock(); defer f.mu.Unlock(); return f.calls } + +// testManager builds a Manager over temp state with all external probes faked. +// active is a pointer so tests flip service state mid-scenario. +func testManager(t *testing.T, fh *fakeHub) (*Manager, *recordingRunner, *bool, *bytes.Buffer) { + t.Helper() + rr := &recordingRunner{fail: map[string]error{}} + active := false + logBuf := &bytes.Buffer{} + logger := slog.New(slog.NewTextHandler(logBuf, &slog.HandlerOptions{Level: slog.LevelDebug})) + m := NewManager(rr, fh, t.TempDir(), logger) + m.isActive = func(ctx context.Context) bool { return active } + m.wgPresent = func() bool { return true } + return m, rr, &active, logBuf +} + +func testBlock(pub string) *hub.WireWireguard { + return &hub.WireWireguard{ + Endpoint: hub.WireWireguardEndpoint{ + DNSName: "ep0.felhom.eu", WGPort: 443, + ServerPubkey: "CQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQk=", PBSTunnelIP: "10.77.0.1", + }, + Pubkey: pub, + AssignedIP: "10.77.0.2/32", + } +} + +func localPub(t *testing.T, m *Manager) string { + t.Helper() + pub, _, err := EnsureKey(m.stateDir) + if err != nil { + t.Fatal(err) + } + return pub +} + +func TestRenderConf_Golden(t *testing.T) { + block := testBlock("ignored") + conf, err := renderConf(block, vectorPrivB64) + if err != nil { + t.Fatal(err) + } + want := `# felhom offsite tunnel — agent-managed (S3); DO NOT EDIT +[Interface] +PrivateKey = ` + vectorPrivB64 + ` +Address = 10.77.0.2/32 +MTU = 1420 + +[Peer] +PublicKey = CQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQk= +Endpoint = ep0.felhom.eu:443 +AllowedIPs = 10.77.0.1/32 +PersistentKeepalive = 25 +` + if conf != want { + t.Errorf("conf mismatch:\n--- got ---\n%s\n--- want ---\n%s", conf, want) + } + // Refusals: bad server key, bad ip, hostile dns_name. + bad := *block + bad.Endpoint.ServerPubkey = "short" + if _, err := renderConf(&bad, vectorPrivB64); err == nil { + t.Error("bad server pubkey accepted") + } + bad = *block + bad.AssignedIP = "10.77.0.2" // no /32 + if _, err := renderConf(&bad, vectorPrivB64); err == nil { + t.Error("assigned_ip without /32 accepted") + } + bad = *block + bad.Endpoint.DNSName = "evil host\ninjected" + if _, err := renderConf(&bad, vectorPrivB64); err == nil { + t.Error("hostile dns_name accepted") + } +} + +func TestScenarioA_RegisterThenApplyOrdering(t *testing.T) { + fh := &fakeHub{} + m, rr, active, _ := testManager(t, fh) + ctx := context.Background() + + // Tick 1: no desired data yet → keygen + ONE registration, NO conf/systemctl. + m.Apply(ctx, false, nil) + if fh.count() != 1 { + t.Fatalf("register calls = %d, want 1", fh.count()) + } + if rr.total() != 0 { + t.Fatalf("privileged execs before the block arrived: %v", rr.calls) + } + if m.loadMarker() == nil { + t.Fatal("marker not written after successful registration") + } + + // Tick 2 (still nothing fetched): NO second registration (marker gate). + m.Apply(ctx, false, nil) + if fh.count() != 1 { + t.Fatalf("register calls after marker = %d, want still 1 (marker gate)", fh.count()) + } + + // Block arrives → conf staged + installed + enabled. + pub := localPub(t, m) + m.Apply(ctx, true, testBlock(pub)) + if rr.count("install") != 1 || rr.count("systemctl") != 1 { + t.Fatalf("apply execs = %v", rr.calls) + } + if got := rr.lastSystemctl(); strings.Join(got, " ") != "systemctl enable --now wg-quick@wg-felhom" { + t.Errorf("systemctl argv = %v", got) + } + if _, err := os.Stat(m.stagedConfPath()); err != nil { + t.Error("staged conf missing") + } + // Steady state with service now active: NO further execs. + *active = true + before := rr.total() + m.Apply(ctx, true, testBlock(pub)) + if rr.total() != before { + t.Errorf("steady-state tick ran execs: %v", rr.calls[before:]) + } +} + +func TestScenarioB_ConfChangeRestartsAndSelfHeal(t *testing.T) { + fh := &fakeHub{} + m, rr, active, _ := testManager(t, fh) + ctx := context.Background() + m.Apply(ctx, false, nil) // register + pub := localPub(t, m) + m.Apply(ctx, true, testBlock(pub)) // first apply (enable) + *active = true + + // Endpoint re-key: server_pubkey changes → re-render + install + RESTART (not reload/enable). + changed := testBlock(pub) + changed.Endpoint.ServerPubkey = "AgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgI=" + m.Apply(ctx, true, changed) + if got := rr.lastSystemctl(); strings.Join(got, " ") != "systemctl restart wg-quick@wg-felhom" { + t.Errorf("conf change systemctl = %v, want restart", got) + } + + // Crash/manual stop with unchanged conf → self-heal enable --now. + *active = false + m.Apply(ctx, true, changed) + if got := rr.lastSystemctl(); strings.Join(got, " ") != "systemctl enable --now wg-quick@wg-felhom" { + t.Errorf("self-heal systemctl = %v, want enable --now", got) + } +} + +func TestScenarioC_RevocationDisablesAndNeverReregisters(t *testing.T) { + fh := &fakeHub{} + m, rr, active, _ := testManager(t, fh) + ctx := context.Background() + m.Apply(ctx, false, nil) + pub := localPub(t, m) + m.Apply(ctx, true, testBlock(pub)) + *active = true + + // The hub removes the peer: PRESENT desired-state, absent block. + m.Apply(ctx, true, nil) + if got := rr.lastSystemctl(); strings.Join(got, " ") != "systemctl disable --now wg-quick@wg-felhom" { + t.Fatalf("revocation systemctl = %v, want disable --now", got) + } + if m.loadMarker() == nil { + t.Fatal("marker removed on revocation — it must be KEPT (revoked stays revoked)") + } + *active = false + + // Multiple later ticks: NO re-register, NO install, NO enable (the load-bearing negatives). + regBefore, execBefore := fh.count(), rr.total() + for i := 0; i < 5; i++ { + m.Apply(ctx, true, nil) + } + if fh.count() != regBefore { + t.Errorf("re-registration after revocation: %d calls", fh.count()-regBefore) + } + if rr.total() != execBefore { + t.Errorf("execs after teardown: %v", rr.calls[execBefore:]) + } + + // Operator re-adds the peer → block returns → tunnel back WITHOUT registration. + m.Apply(ctx, true, testBlock(pub)) + if fh.count() != regBefore { + t.Errorf("re-add triggered a registration: %d", fh.count()-regBefore) + } + if got := rr.lastSystemctl(); strings.Join(got, " ") != "systemctl enable --now wg-quick@wg-felhom" { + t.Errorf("re-add systemctl = %v, want enable --now", got) + } +} + +func TestScenarioD_MismatchReregisterWithBackoff(t *testing.T) { + fh := &fakeHub{} + m, _, _, _ := testManager(t, fh) + ctx := context.Background() + now := time.Unix(1_800_000_000, 0) + m.now = func() time.Time { return now } + + m.Apply(ctx, false, nil) // initial registration + pub := localPub(t, m) + + // The hub serves a block for ANOTHER pubkey → exactly one re-register per backoff window. + other := testBlock("hHwNLDdSNPNl5mCVUYejc1oPdhPRYJ06ak2MU66qWiI=") + fh.err = errors.New("hub 500") + m.Apply(ctx, true, other) + if fh.count() != 2 { + t.Fatalf("mismatch register calls = %d, want 2 (initial + one retry)", fh.count()) + } + // Same instant: backoff gates the hot loop. + m.Apply(ctx, true, other) + m.Apply(ctx, true, other) + if fh.count() != 2 { + t.Fatalf("backoff not applied: %d calls", fh.count()) + } + // After the backoff window: one more attempt, this time succeeding → marker updated. + fh.err = nil + now = now.Add(2 * time.Minute) + m.Apply(ctx, true, other) + if fh.count() != 3 { + t.Fatalf("post-backoff register calls = %d, want 3", fh.count()) + } + if mk := m.loadMarker(); mk == nil || mk.Pubkey != pub { + t.Errorf("marker after re-key = %+v, want local pubkey %s", m.loadMarker(), pub) + } +} + +func TestAdoptLostMarkerWithKnownKey(t *testing.T) { + fh := &fakeHub{} + m, rr, _, _ := testManager(t, fh) + ctx := context.Background() + pub := localPub(t, m) // key exists, NO marker (lost state), hub already has the block + + m.Apply(ctx, true, testBlock(pub)) + if fh.count() != 0 { + t.Fatalf("adopt path re-registered: %d calls", fh.count()) + } + mk := m.loadMarker() + if mk == nil || mk.Pubkey != pub || mk.AssignedIP != "10.77.0.2/32" { + t.Fatalf("adoption marker = %+v", mk) + } + if rr.count("install") != 1 { + t.Errorf("adopt did not proceed to conf apply: %v", rr.calls) + } +} + +func TestNoTeardownOnAbsentData(t *testing.T) { + fh := &fakeHub{} + m, rr, active, _ := testManager(t, fh) + ctx := context.Background() + m.Apply(ctx, false, nil) + pub := localPub(t, m) + m.Apply(ctx, true, testBlock(pub)) + *active = true + before := rr.total() + + // Desired fetch stale/unavailable (fetched=false): keep last-applied state — NO teardown, + // NO execs (red-proof (d) anchor). + for i := 0; i < 3; i++ { + m.Apply(ctx, false, nil) + } + if rr.total() != before { + t.Errorf("absent-data tick ran execs (teardown-on-absent-data bug): %v", rr.calls[before:]) + } +} + +func TestPartialStateAndCorruptKeyIdle(t *testing.T) { + fh := &fakeHub{} + m, rr, _, _ := testManager(t, fh) + ctx := context.Background() + + // Marker without key file → idle, no registration, no execs, key NOT created. + if err := m.writeMarker(marker{Pubkey: "X", AssignedIP: "10.77.0.9/32"}); err != nil { + t.Fatal(err) + } + m.Apply(ctx, false, nil) + if fh.count() != 0 || rr.total() != 0 { + t.Fatalf("partial state acted: reg=%d execs=%d", fh.count(), rr.total()) + } + if _, err := os.Stat(KeyFilePath(m.stateDir)); !os.IsNotExist(err) { + t.Fatal("partial state minted a fresh key — never-guess violated") + } + + // Corrupt key file → idle likewise. + os.Remove(m.markerPath()) + os.MkdirAll(m.wgDir(), 0o700) + os.WriteFile(KeyFilePath(m.stateDir), []byte("garbage"), 0o600) + m.Apply(ctx, false, nil) + if fh.count() != 0 || rr.total() != 0 { + t.Fatalf("corrupt key acted: reg=%d execs=%d", fh.count(), rr.total()) + } +} + +func TestStatusStanzaAndHandshakeParse(t *testing.T) { + fh := &fakeHub{} + m, rr, active, _ := testManager(t, fh) + ctx := context.Background() + m.now = func() time.Time { return time.Unix(1783107215, 0) } // 100s after the scripted handshake + + m.Apply(ctx, false, nil) // keygen + register + *active = true + st := m.Status(ctx) + if st.Pubkey == "" || !st.Registered || !st.Active || st.AssignedIP != "10.77.0.2/32" { + t.Fatalf("status = %+v", st) + } + if st.LastHandshakeAgeS == nil || *st.LastHandshakeAgeS != 100 { + t.Fatalf("handshake age = %v, want 100", st.LastHandshakeAgeS) + } + // The ONLY wg invocation is latest-handshakes — never `dump`. + rr.mu.Lock() + for _, c := range rr.calls { + if c[0] == "wg" && strings.Join(c, " ") != "wg show wg-felhom latest-handshakes" { + t.Errorf("unexpected wg invocation: %v", c) + } + } + rr.mu.Unlock() + + // Inactive service → no wg exec, nil age. + *active = false + wgBefore := rr.count("wg") + st = m.Status(ctx) + if st.Active || st.LastHandshakeAgeS != nil { + t.Errorf("inactive status = %+v", st) + } + if rr.count("wg") != wgBefore { + t.Error("handshake read ran against an inactive service") + } +} + +func TestNoKeyMaterialInLogs(t *testing.T) { + fh := &fakeHub{} + m, _, active, logBuf := testManager(t, fh) + ctx := context.Background() + m.Apply(ctx, false, nil) + pub := localPub(t, m) + m.Apply(ctx, true, testBlock(pub)) + *active = true + m.Apply(ctx, true, nil) // revocation path too + m.Status(ctx) + + raw, err := os.ReadFile(KeyFilePath(m.stateDir)) + if err != nil { + t.Fatal(err) + } + privB64 := strings.TrimSpace(string(raw)) + if privB64 == "" || len(privB64) != 44 { + t.Fatalf("unexpected key file %q", privB64) + } + logs := logBuf.String() + if strings.Contains(logs, privB64) { + t.Fatal("PRIVATE KEY found in the log buffer") + } + // base64 of the raw key without padding variants too (defense against partial logging) + if trimmed := strings.TrimRight(privB64, "="); len(trimmed) > 20 && strings.Contains(logs, trimmed) { + t.Fatal("private key material (unpadded) found in the log buffer") + } + if !strings.Contains(logs, pub) { + t.Error("pubkey absent from logs — expected (pubkeys are fine to log)") + } +}