From a659e5dc09a117fe9a1cd381b0e7e58ac36147be Mon Sep 17 00:00:00 2001 From: kisfenyo Date: Thu, 16 Jul 2026 22:47:22 +0200 Subject: [PATCH] v0.89.0: pbsdr self-grant (R-22) + escrow config live-reload + agent-plane poke listener (Direction-2a) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - pbsdr: on a 403 pre-check (non-default storage id, no ACL yet) self-grant via the root wrapper then re-read, instead of aborting before the grant — closes F4/R-22. Red-proof TestSelfGrant_PreCheck403DoesNotAbortBeforeGrant. - escrow preflight: late-bound CurrentPBSStorageID re-reads agent.json so a pbsdr-seeded pbs_storage_id flips the row green in-process (no restart). Red-proof TestEscrowPreflight_PBSStorageIDLiveReload. - internal/poke: contentless UDP poke listener bound exclusively to the box WG /32 (port 51822), leading-edge debounced, fires the hub-loop out-of-band trigger for an immediate desired-state cycle. First slice of R-13. Red-proofs TestBindConfinement + TestDebounceCoalescesBurst. --- CHANGELOG.md | 38 +++++ REUSE.md | 2 + cmd/felhom-agent/main.go | 44 ++++- internal/hub/loop.go | 13 +- internal/localapi/escrow_ceremony.go | 21 ++- internal/localapi/escrow_ceremony_test.go | 78 +++++++++ internal/pbsdr/manager.go | 33 +++- internal/pbsdr/manager_test.go | 62 +++++++ internal/poke/poke.go | 197 ++++++++++++++++++++++ internal/poke/poke_test.go | 147 ++++++++++++++++ internal/wgtunnel/manager.go | 23 +++ 11 files changed, 643 insertions(+), 15 deletions(-) create mode 100644 internal/poke/poke.go create mode 100644 internal/poke/poke_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index b7de577..13083ef 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,41 @@ +## v0.89.0 — agent train: PBS-DR self-grant (R-22) + escrow config live-reload + agent-plane poke listener (2026-07-16) + +Three bundled agent-plane items. Green: `go build ./... && go vet ./... && go test ./...` all pass. + +- **Item 1 — pbsdr self-grant (R-22, closes the F4 root cause from `tests/VALIDATION-n100`)** + (`internal/pbsdr/manager.go`): on a NON-DEFAULT PBS storage id the agent token holds no ACL on + `/storage/` yet, so the reconcile tick's token-auth pre-check `StorageEntry` (GET + `/storage/`) 403s and — pre-fix — aborted BEFORE the root-run wrapper `grant` that creates + that very ACL: a permanent self-deadlock that bit every legacy/descriptor-provisioned id (the + demo's `felhom-offsite`). Fix: on an `IsForbidden` pre-check error ONLY, run + `felhom-pbs-apply grant ` now (root, no secret, no pre-existing entry — `pveum acl modify` on a + path is unconditional), re-read once, then flow the normal adoption/create path. The pre-check is + KEPT (once the ACL exists it short-circuits the happy path cheaply); every other error stays + transient. Grant-then-still-failing surfaces `verify_failed` loudly, never a silent retry storm. + Red-proof: `TestSelfGrant_PreCheck403DoesNotAbortBeforeGrant` (fake 403 on the pre-check → pre-fix + never reaches the grant, calls=[] → FAIL; fixed → self-grant + converge, no secret consumed). +- **Item 2 — escrow config live-reload** (`internal/localapi/escrow_ceremony.go`, + `cmd/felhom-agent/main.go`): the pbsdr bridge seeds `escrow.pbs_storage_id` into agent.json on DR + convergence, but `/escrow/preflight` read a daemon-start snapshot and stayed red until a service + restart. New late-bound `EscrowCeremonyConfig.CurrentPBSStorageID` resolver (mirrors the existing + `DRConfigured func() bool` pattern) re-reads config from disk at preflight time — exactly what the + ceremony subprocess itself loads, so the row flips green in-process, no restart. Falls back to the + boot snapshot on read error / all-env config. Red-proof: + `TestEscrowPreflight_PBSStorageIDLiveReload` (seed after boot → pre-fix stays red → FAIL). +- **Item 3 — agent-plane poke listener (Direction-2a, `SPIKE-immediate-sync-transport-2026-07-16`)** + (new `internal/poke`, `internal/wgtunnel.LoadAssignedAddr`, `cmd/felhom-agent/main.go`, + `internal/hub/loop.go`): a CONTENTLESS UDP poke relayed hub → ep0 forced-command → wg0-origin → + this listener fires ONE immediate desired-state cycle (the hub control loop's out-of-band report + trigger — the SAME channel the storage watchdog uses; fan-in, cap-1 coalescing). The socket binds + EXCLUSIVELY to the box's WG /32 (`registered.json`; never 0.0.0.0, never the LAN iface) on the + fixed port **51822**; payload is ignored entirely (a forged/replayed poke costs at most one extra + debounced tick); leading-edge debounce (`DebounceWindow`) coalesces a burst into ≤1 tick. Enabled + whenever `wg_tunnel.enabled`; a lost poke is harmless (the 15-min cycle still reconciles). This is + the FIRST concrete slice of the OOB/mutual-repair arc (R-13) — the listener+trigger only, nothing + more. Red-proofs: `TestBindConfinement` (wildcard bind → bound to `::` → FAIL) and + `TestDebounceCoalescesBurst` (guard removed → 10 fires for 10 pokes → FAIL). Port registered in + REUSE.md as a shared cross-repo constant (hub sender + ep0 `felhom-poke`). + ## v0.88.0 — controller-driven escrow ceremony: --output=json + localapi job + one-shot R claim (2026-07-13) The agent half of the customer-facing recovery-code wizard (controller v0.127.0; every mechanism diff --git a/REUSE.md b/REUSE.md index 120a1d7..dec95b7 100644 --- a/REUSE.md +++ b/REUSE.md @@ -93,6 +93,8 @@ | `escrow.Create` | internal/escrow/escrow.go | `Create(ctx, CreateOptions) (CreateResult, R, error)` | PBS-key escrow (zero-knowledge) | Recovery code returned SEPARATELY from the result (anti-log); self-verifies recoverability | | `escrow.CeremonyBinary` / `CeremonyArgs()` / `CeremonyOutput` | internal/escrow/ceremony.go | the ONE fixed sudo self-invocation argv + the `--output=json` wire object (v1) | controller-driven ceremony (v0.88.0) | SINGLE SOURCE shared by the localapi exec, the capability manifest entry, and (byte-identically) the FELHOM_ESCROW sudoers line — `TestEscrowCeremonyArgvPinned` + `TestManifestCoveredBySudoers` lock all three. Never flag-helpers, never `--`→`-` (spike §2.2) | | localapi escrow ceremony job | internal/localapi/escrow_ceremony.go | `POST /escrow/ceremony` + status + ONE-SHOT claim + preflight | the wizard's agent half | R lives ONLY in `Server.escrowR` (NEVER the job struct — snapshots must be structurally R-free); zeroed on claim/supersede/10-min TTL (`unclaimed_void`); in-memory BY DESIGN (restart loses R safely; re-run supersedes); subprocess stdout is SECRET-BEARING → parsed then zeroed, never logged | +| `poke.Listener` + `poke.Port` | internal/poke/poke.go | `NewListener(resolve, trigger, port, logger)`; `poke.Port = 51822` | agent-plane immediate-sync (Direction-2a, v0.89.0) | Binds a contentless UDP socket EXCLUSIVELY to the box's WG /32 (`wgtunnel.LoadAssignedAddr`), fires the hub-loop out-of-band trigger. **Port 51822 is a SHARED cross-repo contract** — the hub poke sender + the ep0 `felhom-poke` forced-command target the SAME number; change one → change all three. Contentless (payload ignored), leading-edge debounced (`DebounceWindow`), WG-confined (kernel EKEYREJECTED refuses non-peer /32s). Wired only when `wg_tunnel.enabled` | +| `wgtunnel.LoadAssignedAddr` | internal/wgtunnel/manager.go | `LoadAssignedAddr(stateDir) (netip.Addr, bool)` | the box's own WG /32 without a Manager | Reads `registered.json`; ok=false until registered; strips the /32 → bare addr (the poke bind target) | ## 2. Canonical patterns (copy structure from THE named file) diff --git a/cmd/felhom-agent/main.go b/cmd/felhom-agent/main.go index 97a83db..8b89589 100644 --- a/cmd/felhom-agent/main.go +++ b/cmd/felhom-agent/main.go @@ -17,6 +17,7 @@ import ( "log/slog" "net" "net/http" + "net/netip" "os" "os/signal" "path/filepath" @@ -41,6 +42,7 @@ import ( "gitea.dooplex.hu/admin/felhom-agent/internal/mgmtplane" "gitea.dooplex.hu/admin/felhom-agent/internal/pbs" "gitea.dooplex.hu/admin/felhom-agent/internal/pbsdr" + "gitea.dooplex.hu/admin/felhom-agent/internal/poke" "gitea.dooplex.hu/admin/felhom-agent/internal/provision" "gitea.dooplex.hu/admin/felhom-agent/internal/proxmox" "gitea.dooplex.hu/admin/felhom-agent/internal/reconcile" @@ -53,7 +55,7 @@ import ( // version is the agent version. Overridable at build time with // -ldflags "-X main.version="; defaults to the in-repo CHANGELOG version. -var version = "0.86.0" +var version = "0.89.0" // runGuestHook is the PVE hook body (`felhom-agent guest-hook `). On pre-start it // creates placeholder dirs for any absent bind-mount source so the guest always boots (the C1 net); @@ -728,8 +730,22 @@ func runDaemon(cfg config.Config, logger *slog.Logger, logRing *applog.Ring) int // the pbsdr manager is constructed further down; the closure reads drConfigured at call time // (nil until then → preflight reports the tier not applied, which is the honest pre-wire answer). escrowCeremonyCfg := &localapi.EscrowCeremonyConfig{ - SudoPath: cfg.Privileged.SudoPath, - PBSStorageID: cfg.Escrow.PBSStorageID, + SudoPath: cfg.Privileged.SudoPath, + PBSStorageID: cfg.Escrow.PBSStorageID, + // Live-reload (v0.89.0): the pbsdr bridge seeds escrow.pbs_storage_id into agent.json on DR + // convergence; re-read it from disk at preflight time so the row flips green without a + // restart. Reading the same file the ceremony subprocess loads keeps the preflight honest. + // On a read error, fall back to the daemon-start snapshot. cfg.SourcePath == "" (all-env + // config) → nothing to re-read, the snapshot stands. + CurrentPBSStorageID: func() string { + if cfg.SourcePath == "" { + return cfg.Escrow.PBSStorageID + } + if c, err := config.Load(cfg.SourcePath); err == nil { + return c.Escrow.PBSStorageID + } + return cfg.Escrow.PBSStorageID + }, HubConfigured: cfg.Hub.URL != "" && cfg.Hub.HostID != "" && cfg.Hub.APIKey != "", DRConfigured: func() bool { if drConfigured != nil { @@ -771,6 +787,7 @@ func runDaemon(cfg config.Config, logger *slog.Logger, logRing *applog.Ring) int // registration, no report stanza. wgServers := 0 var wgLoop *wgtunnel.Loop + var pokeListener *poke.Listener { wt := cfg.WGTunnel.WithDefaults() if wt.Enabled { @@ -785,6 +802,22 @@ func runDaemon(cfg config.Config, logger *slog.Logger, logRing *applog.Ring) int desiredSyncer.AddConsumer(wgLoop) // raw desired-state → the wireguard block collector.SetWireguardReporter(wgLoop) // heartbeat status stanza logger.Info("wgtunnel: enabled", "interval_s", wt.IntervalSeconds, "state_dir", wt.StateDir) + + // Agent-plane immediate-sync LISTENER (Direction-2a, v0.89.0). Binds a contentless UDP + // poke socket EXCLUSIVELY to the box's WG /32 (from registered.json) and nudges the hub + // control loop's out-of-band report trigger — the SAME channel the storage watchdog + // uses (fan-in; the loop coalesces). Enabled whenever the tunnel is (WG is the only path + // a poke can arrive on); a lost poke is harmless — the 15-min cycle still reconciles. + pokeListener = poke.NewListener( + func() (netip.Addr, bool) { return wgtunnel.LoadAssignedAddr(wt.StateDir) }, + func() { + select { + case storageTrigger <- struct{}{}: + default: + } + }, + poke.Port, logger) + logger.Info("poke: agent-plane sync listener enabled", "port", poke.Port) } } @@ -840,13 +873,16 @@ func runDaemon(cfg config.Config, logger *slog.Logger, logRing *applog.Ring) int // Run reconcile, the hub loop, the storage watchdog, the restore-test scheduler, the PBS // verify loop, (optionally) the local-API server, and (optionally) the LAN resolver loop // concurrently; any one returning ends the daemon (ctx cancel tears down the rest). - errc := make(chan error, 7) + errc := make(chan error, 8) go func() { errc <- engine.Run(ctx, interval) }() go func() { errc <- loop.Run(ctx) }() go func() { errc <- watchdog.Run(ctx) }() go func() { errc <- scheduler.Run(ctx) }() go func() { errc <- pbsLoop.Run(ctx) }() go func() { errc <- pbsdrLoop.Run(ctx) }() + if pokeListener != nil { + go func() { errc <- pokeListener.Run(ctx) }() // agent-plane immediate-sync listener (v0.89.0) + } if localSrv != nil { localServers = 1 // Host-reboot remount fix: BEFORE binding into the guest, re-assert enrolled drive MOUNTS on the diff --git a/internal/hub/loop.go b/internal/hub/loop.go index 7d553b7..1c9435c 100644 --- a/internal/hub/loop.go +++ b/internal/hub/loop.go @@ -81,9 +81,10 @@ func NewLoop(collector collectorIface, client reporter, interval time.Duration, } // SetTrigger wires an out-of-band report channel. A receive on it runs one extra -// collect→report cycle immediately WITHOUT disturbing the regular ticker cadence — used by -// the storage watchdog to push a disconnect to the hub in seconds. The watchdog debounces, -// so this fires at most once per debounce window. +// collect→report cycle immediately WITHOUT disturbing the regular ticker cadence. Two producers +// fan into this one channel: the storage watchdog (push a disconnect in seconds) and the +// agent-plane poke listener (v0.89.0 immediate-sync). Both debounce, and the channel is cap-1 +// non-blocking, so a burst coalesces to at most one pending extra cycle. func (l *Loop) SetTrigger(ch <-chan struct{}) { l.trigger = ch } // SetEnvelopeObserver wires the slice-10A desired-state sync hook. It is called once per cycle @@ -115,10 +116,10 @@ func (l *Loop) Run(ctx context.Context) error { ticker.Reset(interval) } case <-l.trigger: - // Out-of-band report (storage watchdog). Run a cycle now; keep the regular - // cadence (do not reset the ticker). The envelope's interval is still adopted + // Out-of-band report (storage watchdog OR agent-plane poke). Run a cycle now; keep the + // regular cadence (do not reset the ticker). The envelope's interval is still adopted // if it changed, mirroring the normal path. - l.logger.Info("hub: out-of-band report triggered (storage watchdog)") + l.logger.Info("hub: out-of-band report triggered (watchdog/poke)") next := l.cycle(ctx, interval) if next != interval { l.logger.Info("hub: poll interval changed", "from", interval, "to", next) diff --git a/internal/localapi/escrow_ceremony.go b/internal/localapi/escrow_ceremony.go index c67bab2..47926ad 100644 --- a/internal/localapi/escrow_ceremony.go +++ b/internal/localapi/escrow_ceremony.go @@ -34,8 +34,17 @@ import ( type EscrowCeremonyConfig struct { // SudoPath is the sudo binary ("" → "sudo"). SudoPath string - // PBSStorageID is cfg.Escrow.PBSStorageID ("" = not configured — preflight red). + // PBSStorageID is cfg.Escrow.PBSStorageID at daemon-start ("" = not configured — preflight + // red). It is the FALLBACK snapshot; the live value is CurrentPBSStorageID when wired. PBSStorageID string + // CurrentPBSStorageID, when set, is called at preflight time to read the LIVE + // escrow.pbs_storage_id. The pbsdr bridge SEEDS this key into agent.json on DR convergence + // (finishConverged → seedEscrowStorageID); a static snapshot taken at daemon start would then + // stay red until a service restart (v0.89.0 live-reload). The wired closure re-reads config + // from disk — exactly what the ceremony subprocess itself loads — so the preflight reflects the + // real state the bare `--selftest=escrow-create` one-liner will see. nil → the PBSStorageID + // snapshot is used (old wiring / tests). + CurrentPBSStorageID func() string // HubConfigured: hub url + host id + api key all present (the --upload target). HubConfigured bool // DRConfigured answers "is the DR tier applied on this box?" (pbsdr.Manager.DRConfigured, @@ -362,8 +371,14 @@ func (s *Server) handleEscrowPreflight(w http.ResponseWriter, r *http.Request, v } items := make([]item, 0, 6) - items = append(items, item{ID: "pbs_storage_id", OK: cfg.PBSStorageID != "", - Detail: map[bool]string{true: cfg.PBSStorageID, false: "escrow.pbs_storage_id not configured"}[cfg.PBSStorageID != ""]}) + // Live-reload (v0.89.0): prefer the current on-disk value over the daemon-start snapshot, so a + // pbsdr convergence that just seeded escrow.pbs_storage_id flips this row green with no restart. + storageID := cfg.PBSStorageID + if cfg.CurrentPBSStorageID != nil { + storageID = cfg.CurrentPBSStorageID() + } + items = append(items, item{ID: "pbs_storage_id", OK: storageID != "", + Detail: map[bool]string{true: storageID, false: "escrow.pbs_storage_id not configured"}[storageID != ""]}) drOK := cfg.DRConfigured != nil && cfg.DRConfigured() items = append(items, item{ID: "dr_tier", OK: drOK, diff --git a/internal/localapi/escrow_ceremony_test.go b/internal/localapi/escrow_ceremony_test.go index f9b12ef..c0efd01 100644 --- a/internal/localapi/escrow_ceremony_test.go +++ b/internal/localapi/escrow_ceremony_test.go @@ -5,6 +5,8 @@ import ( "encoding/json" "fmt" "net/http" + "os" + "path/filepath" "strings" "sync/atomic" "testing" @@ -61,6 +63,82 @@ func startAndWait(t *testing.T, srv *Server, h http.Handler) { } } +// TestEscrowPreflight_PBSStorageIDLiveReload pins the v0.89.0 live-reload: the pbsdr bridge seeds +// escrow.pbs_storage_id into agent.json on DR convergence; the preflight must see that seed IN THE +// SAME PROCESS (no restart). The daemon-start snapshot (cfg.PBSStorageID) is empty; only the +// late-bound CurrentPBSStorageID resolver (re-reads disk) reflects the seed. RED-PROOF: the pre-fix +// preflight read cfg.PBSStorageID directly and ignored the resolver, so step 3 stays red → FAIL. +func TestEscrowPreflight_PBSStorageIDLiveReload(t *testing.T) { + dir := t.TempDir() + cfgPath := filepath.Join(dir, "agent.json") + // Pre-convergence: agent.json has no escrow.pbs_storage_id. + if err := os.WriteFile(cfgPath, []byte(`{"escrow":{"posture":"zero_knowledge"}}`), 0o600); err != nil { + t.Fatalf("seed config: %v", err) + } + readCurrent := func() string { + raw, err := os.ReadFile(cfgPath) + if err != nil { + return "" + } + var doc struct { + Escrow struct { + PBSStorageID string `json:"pbs_storage_id"` + } `json:"escrow"` + } + _ = json.Unmarshal(raw, &doc) + return doc.Escrow.PBSStorageID + } + + srv := newEscrowTestServer(t, okRunner(nil)) + srv.escrowCeremony.PBSStorageID = "" // daemon-start snapshot: empty (the pre-fix source) + srv.escrowCeremony.CurrentPBSStorageID = readCurrent // live resolver → current disk state + h := srv.Handler() + + rowOK := func(t *testing.T) (bool, string) { + t.Helper() + w := do(t, h, "GET", "/escrow/preflight", "A", "") + if w.Code != http.StatusOK { + t.Fatalf("preflight status %d: %s", w.Code, w.Body.String()) + } + var resp struct { + Data struct { + Items []struct { + ID string `json:"id"` + OK bool `json:"ok"` + Detail string `json:"detail"` + } `json:"items"` + } `json:"data"` + } + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatalf("decode preflight: %v (%s)", err, w.Body.String()) + } + for _, it := range resp.Data.Items { + if it.ID == "pbs_storage_id" { + return it.OK, it.Detail + } + } + t.Fatal("pbs_storage_id row missing from preflight") + return false, "" + } + + // 1. Before convergence: the row is RED. + if ok, _ := rowOK(t); ok { + t.Fatal("pbs_storage_id row is green before any seed") + } + // 2. Convergence seeds the id IN-PROCESS (simulating pbsdr finishConverged → seedEscrowStorageID). + if err := os.WriteFile(cfgPath, []byte(`{"escrow":{"posture":"zero_knowledge","pbs_storage_id":"felhom-offsite"}}`), 0o600); err != nil { + t.Fatalf("seed after convergence: %v", err) + } + // 3. Same process, NO restart: the row flips GREEN and reports the seeded id. + ok, detail := rowOK(t) + if !ok { + t.Fatal("pbs_storage_id row still red after the in-process seed — the preflight is not live-reloading (restart-only)") + } + if detail != "felhom-offsite" { + t.Fatalf("pbs_storage_id detail = %q, want the seeded id felhom-offsite", detail) + } +} + // Scenario A/D happy path: run → done → claim ONCE (R delivered, no-store) → 410 on re-claim, // holder zeroed. R never appears in the start or status payloads. func TestEscrowCeremony_OneShotClaim(t *testing.T) { diff --git a/internal/pbsdr/manager.go b/internal/pbsdr/manager.go index 5990fb9..d462b2e 100644 --- a/internal/pbsdr/manager.go +++ b/internal/pbsdr/manager.go @@ -239,8 +239,37 @@ func (m *Manager) Apply(ctx context.Context, fetched bool, block *hub.WirePBSDR) entry, found, err := m.px.StorageEntry(ctx, block.StorageID) if err != nil { - m.logger.Warn("pbsdr: storage-entry read failed (transient; retrying next tick)", "err", err) - return + // R-22 self-grant (F4, tests/VALIDATION-n100-baremetal): on a NON-DEFAULT storage id the + // agent token holds no ACL on /storage/ yet, so this token-auth pre-check + // (GET /storage/) 403s. Aborting here would deadlock permanently — the root-run wrapper + // `grant` that CREATES that very ACL is only reached further down (adoption / create paths). + // So on a 403 ONLY, run the grant now (root, no secret, no pre-existing entry required — + // `pveum acl modify` on a path is unconditional) and re-read once; the retry then flows the + // normal adoption/create path. Every OTHER error stays transient (retry next tick). The + // pre-check itself is KEPT: once the ACL exists the read succeeds and short-circuits the + // happy path cheaply — we only stop the 403 from being a first-contact dead-end. + var ae *proxmox.APIError + if !errors.As(err, &ae) || !ae.IsForbidden() { + m.logger.Warn("pbsdr: storage-entry read failed (transient; retrying next tick)", "err", err) + return + } + m.logger.Info("pbsdr: pre-check 403 (token has no ACL on this storage id yet) — self-granting via the root wrapper, then re-reading (R-22)", + "storage_id", block.StorageID) + if _, errOut, gerr := m.runner.Run(ctx, WrapperPath, "grant", block.StorageID); gerr != nil { + m.logger.Warn("pbsdr: self-grant failed (retrying next tick)", "err", gerr, "stderr", tail(errOut)) + m.setStatus(&hub.PBSDRStatus{State: "verify_failed", StorageID: block.StorageID, Namespace: block.Namespace, + Message: "pre-check 403 and self-grant failed: " + gerr.Error()}) + return + } + entry, found, err = m.px.StorageEntry(ctx, block.StorageID) + if err != nil { + // Grant succeeded but the read STILL fails → not the ACL bootstrap after all; surface it + // loudly rather than looping silently. + m.logger.Warn("pbsdr: storage-entry read still failing after self-grant (retrying next tick)", "err", err) + m.setStatus(&hub.PBSDRStatus{State: "verify_failed", StorageID: block.StorageID, Namespace: block.Namespace, + Message: "storage read failed even after self-grant: " + err.Error()}) + return + } } if found && entry.Type != "pbs" { msg := fmt.Sprintf("storage id %s exists with type %q (not pbs) — refusing to touch it", block.StorageID, entry.Type) diff --git a/internal/pbsdr/manager_test.go b/internal/pbsdr/manager_test.go index 7554dda..0f4c1be 100644 --- a/internal/pbsdr/manager_test.go +++ b/internal/pbsdr/manager_test.go @@ -67,9 +67,19 @@ type fakeStorage struct { found bool active []bool // consumed per StorageActive call; last value repeats calls int + // entryErrs is consumed per StorageEntry call (nil = the normal (entry,found,nil) answer); + // after the slice is exhausted every call answers normally. Lets a test model the R-22 + // pre-check 403 that self-grant must recover from without aborting. + entryErrs []error + entryCalls int } func (f *fakeStorage) StorageEntry(context.Context, string) (*proxmox.StorageEntryConfig, bool, error) { + i := f.entryCalls + f.entryCalls++ + if i < len(f.entryErrs) && f.entryErrs[i] != nil { + return nil, false, f.entryErrs[i] + } return f.entry, f.found, nil } @@ -182,6 +192,58 @@ func TestFreshPath_SecretOnStdinNeverArgv(t *testing.T) { } } +// TestSelfGrant_PreCheck403DoesNotAbortBeforeGrant pins R-22 (F4 from tests/VALIDATION-n100): +// on a non-default storage id whose ACL the token lacks, the token-auth pre-check GET /storage/ +// 403s. The fix must NOT abort — it must run the root-run `grant` (which creates that very ACL), +// re-read, and converge. RED-PROOF: the pre-fix code returns on the StorageEntry error before any +// runner call, so NO grant runs and the box never converges — this test then fails on +// "self-grant never ran". No secret may be consumed on this (adoption) path. +func TestSelfGrant_PreCheck403DoesNotAbortBeforeGrant(t *testing.T) { + r := &fakeRunner{} + forbidden := &proxmox.APIError{StatusCode: 403, Method: "GET", Path: "/storage/felhom-offsite", + Body: "Permission check failed (/storage/felhom-offsite, Datastore.Audit)"} + st := &fakeStorage{ + // First read 403s (no ACL); after the self-grant the entry reads healthy → adoption. + entryErrs: []error{forbidden}, + entry: &proxmox.StorageEntryConfig{Storage: "felhom-offsite", Type: "pbs", + Namespace: "peti", Fingerprint: testFP}, + found: true, + active: []bool{true}, + } + c := &fakeConsumer{secret: "MUST-NOT-BURN"} + m, _ := newTestManager(t, r, st, c) + block := testBlock() + block.StorageID = "felhom-offsite" + block.Namespace = "peti" + + m.Apply(context.Background(), true, block) + + // A 403 on IsForbidden must be recognised as such (regression guard on the type assertion). + if !forbidden.IsForbidden() { + t.Fatal("precondition: crafted APIError is not IsForbidden") + } + calls := r.recorded() + grants := 0 + for _, call := range calls { + if len(call.Args) > 0 && call.Args[0] == "grant" { + grants++ + } + } + if grants == 0 { + t.Fatalf("self-grant never ran — the pre-check 403 aborted before the root grant (R-22 regression); calls=%+v", calls) + } + if c.calls != 0 { + t.Fatalf("a secret was consumed on the self-grant/adoption path (%d calls) — the no-consume law", c.calls) + } + if s := m.Status(); s == nil || (s.State != "adopted" && s.State != "applied") { + t.Fatalf("status = %+v, want converged (adopted/applied) after self-grant", s) + } + // The pre-check was re-read (not dropped): 2 StorageEntry calls — the 403, then the post-grant read. + if st.entryCalls < 2 { + t.Fatalf("StorageEntry called %d times — the post-grant re-read is missing", st.entryCalls) + } +} + func TestVerifyPinBeforeConsume(t *testing.T) { r := &fakeRunner{} st := &fakeStorage{found: false} diff --git a/internal/poke/poke.go b/internal/poke/poke.go new file mode 100644 index 0000000..af5ec13 --- /dev/null +++ b/internal/poke/poke.go @@ -0,0 +1,197 @@ +// Package poke is the agent-plane immediate-sync LISTENER (Direction-2a, +// SPIKE-immediate-sync-transport-2026-07-16). It receives a CONTENTLESS UDP "sync now" poke +// relayed hub → ep0 forced-command → wg0-origin → here, and fires ONE immediate desired-state +// cycle (the hub control loop's out-of-band report trigger). This collapses a user-triggered +// agent-plane config change (e.g. a pbsdr descriptor) from the 15-min report cycle to the +// spike-measured ~0.42 s path — while the 15-min cycle remains the guarantee (a lost poke is +// harmless by construction). +// +// The security posture is the spike's, unchanged: +// - BIND CONFINEMENT: the socket binds EXCLUSIVELY to the box's WireGuard /32 (10.77.0.x from +// registered.json). Never 0.0.0.0, never the LAN interface — a datagram to the LAN address +// reaches no socket. The only way in is over the tunnel, and WireGuard refuses to encrypt to a +// /32 no registered peer owns (spike P1 EKEYREJECTED), so a poke can only originate from ep0. +// - CONTENTLESS: the payload is ignored ENTIRELY — any datagram means only "tick now". A forged +// or replayed poke costs at most one extra debounced tick; it carries no config, no auth +// handshake, no version. There is nothing in a poke to trust. +// - DEBOUNCE: a burst of pokes within DebounceWindow yields ≤1 extra tick (leading-edge). The +// downstream trigger channel is itself cap-1 coalescing, so a tick already running collapses +// further pokes into one follow-up — never N queued. +package poke + +import ( + "context" + "fmt" + "log/slog" + "net" + "net/netip" + "sync" + "time" +) + +// Port is the FIXED UDP port the listener binds on the box's WG address and the hub/ep0 +// forced-command targets. It is a shared cross-repo contract (documented in REUSE.md and the ep0 +// felhom-poke forced-command): change it in one place → change it in all three. +const Port = 51822 + +// DebounceWindow coalesces a burst of pokes into a single immediate tick (leading-edge). Two +// distinct operator changes inside one window still converge in one tick because desired-state is +// level-triggered (the tick fetches the LATEST state, not a delta). +const DebounceWindow = 2 * time.Second + +// rebindInterval is how long the listener waits before re-attempting a bind after WG is +// unregistered/down or a serve loop ends. +const rebindInterval = 5 * time.Second + +// readBufSize bounds a single datagram read. A poke is contentless; the buffer only needs to be +// large enough to drain whatever a (harmless) sender emits. +const readBufSize = 512 + +// Listener binds the poke socket and fires the report trigger on each (debounced) poke. +type Listener struct { + // resolve returns the box's current WG address (ok=false until registered). Called before each + // bind attempt so a late registration / rare re-address is picked up on the next cycle. + resolve func() (netip.Addr, bool) + // trigger is the non-blocking nudge into the hub control loop's out-of-band report channel. + trigger func() + port int + window time.Duration + logger *slog.Logger + + mu sync.Mutex + lastFire time.Time + local netip.AddrPort // set once bound (for tests / logging) + boundCh chan struct{} // closed once the first bind succeeds + now func() time.Time +} + +// NewListener builds a listener. port<=0 uses the fixed Port. A nil trigger is a no-op (safe). +func NewListener(resolve func() (netip.Addr, bool), trigger func(), port int, logger *slog.Logger) *Listener { + if logger == nil { + logger = slog.Default() + } + if trigger == nil { + trigger = func() {} + } + if port <= 0 { + port = Port + } + return &Listener{ + resolve: resolve, trigger: trigger, port: port, + window: DebounceWindow, + logger: logger, + boundCh: make(chan struct{}), + now: time.Now, + } +} + +// onPoke applies the leading-edge debounce and fires the trigger at most once per window. Returns +// true iff it fired (the test hook for the debounce red-proof). +func (l *Listener) onPoke() bool { + l.mu.Lock() + now := l.now() + if !l.lastFire.IsZero() && now.Sub(l.lastFire) < l.window { + l.mu.Unlock() + return false + } + l.lastFire = now + l.mu.Unlock() + l.trigger() + return true +} + +// Run resolves the WG address and serves pokes until ctx is cancelled, rebinding across a WG +// down/re-address. It NEVER returns nil early (the daemon treats a returned goroutine as a reason +// to exit) — only ctx cancellation ends it. +func (l *Listener) Run(ctx context.Context) error { + for { + if err := ctx.Err(); err != nil { + return err + } + addr, ok := l.resolve() + if !ok { + l.logger.Debug("poke: no WG address yet (box not registered) — will retry") + if !l.sleep(ctx, rebindInterval) { + return ctx.Err() + } + continue + } + if err := l.serve(ctx, addr); err != nil && ctx.Err() == nil { + l.logger.Warn("poke: listener stopped; will rebind", "err", err, "wg_addr", addr.String()) + if !l.sleep(ctx, rebindInterval) { + return ctx.Err() + } + } + } +} + +// serve binds EXCLUSIVELY to addr:port and delivers each datagram to onPoke until ctx or a read +// error ends it. +func (l *Listener) serve(ctx context.Context, addr netip.Addr) error { + bind := netip.AddrPortFrom(addr, uint16(l.port)).String() + var lc net.ListenConfig + pc, err := lc.ListenPacket(ctx, "udp", bind) + if err != nil { + return fmt.Errorf("bind %s: %w", bind, err) + } + defer pc.Close() + + l.mu.Lock() + if ap, ok := netip.ParseAddrPort(pc.LocalAddr().String()); ok == nil { + l.local = ap + } + select { + case <-l.boundCh: // already closed + default: + close(l.boundCh) + } + l.mu.Unlock() + l.logger.Info("poke: listening for hub sync-pokes (WG-confined, contentless)", "addr", pc.LocalAddr().String()) + + // Unblock ReadFrom on shutdown. + go func() { + <-ctx.Done() + pc.Close() + }() + + buf := make([]byte, readBufSize) + for { + _, from, err := pc.ReadFrom(buf) + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + return fmt.Errorf("read: %w", err) + } + // Contentless: the payload is discarded. ANY datagram = "tick now". + if l.onPoke() { + l.logger.Info("poke received → triggering an immediate desired-state cycle", "from", from.String()) + } else { + l.logger.Debug("poke received but coalesced within the debounce window", "from", from.String()) + } + } +} + +// BoundAddrPort waits (bounded by ctx) until the socket is bound and returns its address. Test +// helper; production never needs it. +func (l *Listener) BoundAddrPort(ctx context.Context) (netip.AddrPort, bool) { + select { + case <-l.boundCh: + l.mu.Lock() + defer l.mu.Unlock() + return l.local, true + case <-ctx.Done(): + return netip.AddrPort{}, false + } +} + +func (l *Listener) sleep(ctx context.Context, d time.Duration) bool { + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return false + case <-t.C: + return true + } +} diff --git a/internal/poke/poke_test.go b/internal/poke/poke_test.go new file mode 100644 index 0000000..c4a3b61 --- /dev/null +++ b/internal/poke/poke_test.go @@ -0,0 +1,147 @@ +package poke + +import ( + "context" + "io" + "log/slog" + "net" + "net/netip" + "sync/atomic" + "testing" + "time" +) + +func discard() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) } + +// waitFor polls cond up to d, returning true if it became true. +func waitFor(d time.Duration, cond func() bool) bool { + deadline := time.Now().Add(d) + for time.Now().Before(deadline) { + if cond() { + return true + } + time.Sleep(2 * time.Millisecond) + } + return cond() +} + +// TestBindConfinement is red-proof (a): the listener binds EXCLUSIVELY to the address the resolver +// returns (the box's WG /32), never the wildcard 0.0.0.0/::. A wildcard bind would ALSO accept the +// LAN address — the exact leak the confinement forbids. We assert the bound socket's local address +// IS the specific resolved address and is NOT unspecified, and that a datagram to it fires the +// trigger. RED-PROOF: a 0.0.0.0 bind makes BoundAddrPort().Addr().IsUnspecified() true → FAIL. +func TestBindConfinement(t *testing.T) { + want := netip.MustParseAddr("127.0.0.1") // stand-in for the WG /32 on a portable loopback + var fires atomic.Int32 + l := NewListener( + func() (netip.Addr, bool) { return want, true }, + func() { fires.Add(1) }, + 0, // OS-chosen port so the test never collides + discard(), + ) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go l.Run(ctx) + + bctx, bcancel := context.WithTimeout(context.Background(), 3*time.Second) + defer bcancel() + bound, ok := l.BoundAddrPort(bctx) + if !ok { + t.Fatal("listener never bound") + } + if bound.Addr() != want { + t.Fatalf("bound to %s, want the specific WG addr %s (a wildcard bind would leak to the LAN)", bound.Addr(), want) + } + if bound.Addr().IsUnspecified() { + t.Fatal("listener bound to the wildcard address (0.0.0.0/::) — NOT confined to the WG /32") + } + + // Functional: a datagram to the WG address fires exactly one trigger. + conn, err := net.Dial("udp", bound.String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + if _, err := conn.Write([]byte("")); err != nil { // contentless + t.Fatalf("write poke: %v", err) + } + if !waitFor(2*time.Second, func() bool { return fires.Load() >= 1 }) { + t.Fatal("a poke to the bound WG address did not fire the trigger") + } +} + +// TestDebounceCoalescesBurst is red-proof (b): a burst of M pokes within DebounceWindow yields ≤1 +// trigger; after the window elapses a further poke fires again. onPoke uses an injected clock so +// the test is deterministic. RED-PROOF: remove the leading-edge guard (fire every time) → M fires +// for M pokes → the "want 1" assertion FAILS. +func TestDebounceCoalescesBurst(t *testing.T) { + var fires atomic.Int32 + l := NewListener(nil, func() { fires.Add(1) }, 0, discard()) + + base := time.Unix(1_700_000_000, 0) + var clock atomic.Int64 + clock.Store(base.UnixNano()) + l.now = func() time.Time { return time.Unix(0, clock.Load()) } + + // A burst of 10 pokes at the SAME instant → exactly one fire. + const burst = 10 + for i := 0; i < burst; i++ { + l.onPoke() + } + if got := fires.Load(); got != 1 { + t.Fatalf("burst of %d pokes fired %d triggers, want exactly 1 (debounce coalescing)", burst, got) + } + + // Still inside the window → no additional fire. + clock.Store(base.Add(l.window - time.Millisecond).UnixNano()) + l.onPoke() + if got := fires.Load(); got != 1 { + t.Fatalf("a poke inside the debounce window fired again (total %d), want still 1", got) + } + + // Window elapsed → the next poke fires (immediacy is not lost after the window). + clock.Store(base.Add(l.window + time.Millisecond).UnixNano()) + l.onPoke() + if got := fires.Load(); got != 2 { + t.Fatalf("a poke after the window did not fire (total %d), want 2", got) + } +} + +// TestRunWaitsForRegistration: with no WG address yet, Run keeps retrying (never returns early, +// never binds) — a lost/absent poke path is harmless; the 15-min cycle still reconciles. +func TestRunWaitsForRegistration(t *testing.T) { + var resolved atomic.Bool + l := NewListener( + func() (netip.Addr, bool) { + if resolved.Load() { + return netip.MustParseAddr("127.0.0.1"), true + } + return netip.Addr{}, false + }, + func() {}, 0, discard(), + ) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan struct{}) + go func() { l.Run(ctx); close(done) }() + + // Not bound while unregistered. + nb, nbcancel := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer nbcancel() + if _, ok := l.BoundAddrPort(nb); ok { + t.Fatal("listener bound before the box registered") + } + // Registration appears → it binds within a rebind cycle. + resolved.Store(true) + bctx, bcancel := context.WithTimeout(context.Background(), 3*rebindInterval) + defer bcancel() + if _, ok := l.BoundAddrPort(bctx); !ok { + t.Fatal("listener never bound after registration appeared") + } + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Run did not exit on ctx cancel") + } +} diff --git a/internal/wgtunnel/manager.go b/internal/wgtunnel/manager.go index 02a557b..db88087 100644 --- a/internal/wgtunnel/manager.go +++ b/internal/wgtunnel/manager.go @@ -162,6 +162,29 @@ func (m *Manager) loadMarker() *marker { return &mk } +// LoadAssignedAddr reads the box's assigned WG address from the registration marker +// (/wg/registered.json) WITHOUT constructing a Manager — the poke listener (v0.89.0) +// binds EXCLUSIVELY to this /32, so it needs the bare address, not the /32 prefix. Returns +// ok=false until the box has registered (marker absent / unparsable / empty). The address is +// stable across the box's life (preserved on re-key / reinstall — the WG IP is host-scoped). +func LoadAssignedAddr(stateDir string) (netip.Addr, bool) { + raw, err := os.ReadFile(filepath.Join(stateDir, "wg", markerName)) + if err != nil { + return netip.Addr{}, false + } + var mk marker + if json.Unmarshal(raw, &mk) != nil || mk.AssignedIP == "" { + return netip.Addr{}, false + } + if pfx, err := netip.ParsePrefix(mk.AssignedIP); err == nil { + return pfx.Addr(), true + } + if a, err := netip.ParseAddr(mk.AssignedIP); err == nil { // tolerate a bare addr + return a, true + } + return netip.Addr{}, false +} + func (m *Manager) writeMarker(mk marker) error { if err := os.MkdirAll(m.wgDir(), 0o700); err != nil { return err