hub v0.72.0 — R-70 + R-71c: offsite delivery-state detector, card, stuck event, R-39(a)-guarded self-heal restage

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NKSN3gSg4TKVBBqkwW2djR
2026-07-23 12:59:04 +02:00
parent c801cee647
commit 1133aade73
13 changed files with 1179 additions and 3 deletions
@@ -0,0 +1,188 @@
+package monitor
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"time"
+
+	"gitea.dooplex.hu/admin/felhom-hub/internal/offsite"
+	"gitea.dooplex.hu/admin/felhom-hub/internal/store"
+)
+
+// R-70 + R-71(c): the offsite delivery-state checker. Reads the shared detector
+// (offsite.DeliveryStateFor) for every offsite-enabled customer and drives two consumers:
+//
+//   - the LOUD EVENT: `offsite_delivery_stuck` (warning → operator email per existing dispatcher
+//     rules) when the burned-credential shape persists past stuckAfter;
+//   - the SELF-HEAL (R-71c): invoke the EXISTING Re-issue path — never a second delivery
+//     mechanism — when the shape is unambiguous, then surface `offsite_credential_restaged`
+//     (warning) so the operator ALWAYS knows it fired.
+//
+// Cooldowns are durable: both events rate-limit off store.LastEventAt (the events table), so a hub
+// restart cannot flood or silently re-heal. A repeating pattern surfaces as repeating events on a
+// 24 h cadence, never as a silent retry loop.
+//
+// THE R-39(a) GUARD (mandatory, enforced HERE because the store deliberately clobbers):
+// SaveOneTimeSecret is last-write-wins by design — Re-issue depends on supersede. Restaging on top
+// of an UNCONSUMED secret would clobber a password a box may be about to consume (the operator may
+// have clicked Re-issue between this checker's derive and its act). So the heal re-reads the
+// secret row IMMEDIATELY before acting and refuses unless it is still a CONSUMED row.
+
+const (
+	eventDeliveryStuck      = "offsite_delivery_stuck"      // hub-internal (not in allowedEventTypes, like pbsdr_*)
+	eventCredentialRestaged = "offsite_credential_restaged" // hub-internal
+	// stuckAfter: consumed_awaiting_apply is normal for seconds (F10 repair: consume→applied in
+	// 4 s). An hour of it means the apply will never come without intervention.
+	stuckAfter = time.Hour
+	// stuckCooldown / healCooldown: per-customer, durable via the events table.
+	stuckCooldown = 24 * time.Hour
+	healCooldown  = 24 * time.Hour
+	// healMinReports: at least this many consecutive offbox-less reports after consumed_at before
+	// the self-heal may fire — the box must be alive and reporting, just not applied.
+	healMinReports = 4
+)
+
+// OffsiteReissuer is the narrow reissue surface the self-heal needs — satisfied by
+// *web.Server.ReissueOffsiteForCustomer (the pbsdrheal.Reissuer precedent; avoids an import
+// cycle and guarantees the heal IS the designed Re-issue path, not a sibling mechanism).
+type OffsiteReissuer interface {
+	ReissueOffsiteForCustomer(ctx context.Context, customerID string) error
+}
+
+// OffsiteDeliveryChecker runs on the shared monitor ticker.
+type OffsiteDeliveryChecker struct {
+	store    *store.Store
+	reissuer OffsiteReissuer // nil → self-heal disabled (no provisioner configured); detector+event still run
+	onEvent  EventNotifyFunc
+	logger   *log.Logger
+	now      func() time.Time
+}
+
+// NewOffsiteDeliveryChecker constructs the checker. reissuer may be nil (heal disabled).
+func NewOffsiteDeliveryChecker(s *store.Store, reissuer OffsiteReissuer, onEvent EventNotifyFunc, logger *log.Logger) *OffsiteDeliveryChecker {
+	return &OffsiteDeliveryChecker{store: s, reissuer: reissuer, onEvent: onEvent, logger: logger, now: time.Now}
+}
+
+// Check derives the delivery state for every offsite-enabled active customer and applies the
+// event + self-heal rules. Never returns an error — a checker failure must not take the ticker
+// down (log-and-continue, like every sibling checker).
+func (c *OffsiteDeliveryChecker) Check() {
+	configs, err := c.store.ListCustomerConfigs()
+	if err != nil {
+		c.logger.Printf("[WARN] offsite-delivery: list configs: %v", err)
+		return
+	}
+	for _, cfg := range configs {
+		if cfg.Status != "active" {
+			continue // blocked/inactive customers are not delivery-monitored (and never healed)
+		}
+		d, err := offsite.ReadDescriptor(cfg.ConfigJSON)
+		if err != nil || d == nil || !d.Enabled {
+			continue // unparseable config never drives a heal; the config UI owns that failure
+		}
+		status, err := offsite.DeliveryStateFor(c.store, cfg.CustomerID)
+		if err != nil {
+			c.logger.Printf("[WARN] offsite-delivery: %s: derive: %v", cfg.CustomerID, err)
+			continue
+		}
+		if status.State != offsite.DeliveryConsumedAwaitingApply {
+			continue // applied / staged / no_secret: card-rendered states, no event or heal (yet)
+		}
+		age := c.now().Sub(status.Since)
+		if age < stuckAfter {
+			continue // normal convergence window
+		}
+		c.maybeEmitStuck(cfg.CustomerID, status, age)
+		c.maybeHeal(cfg.CustomerID, status)
+	}
+}
+
+// maybeEmitStuck emits offsite_delivery_stuck (warning) once per stuckCooldown per customer.
+func (c *OffsiteDeliveryChecker) maybeEmitStuck(customerID string, status offsite.DeliveryStatus, age time.Duration) {
+	last, err := c.store.LastEventAt(customerID, eventDeliveryStuck)
+	if err != nil {
+		c.logger.Printf("[WARN] offsite-delivery: %s: cooldown read: %v", customerID, err)
+		return
+	}
+	if !last.IsZero() && c.now().Sub(last) < stuckCooldown {
+		return
+	}
+	msg := fmt.Sprintf("Offsite delivery stuck: one-time password consumed %s ago and %d report(s) since carry no offbox target — the credential is likely burned (apply died between consume and persist). Re-issue delivers a fresh one.",
+		age.Round(time.Minute), status.ReportsSinceConsume)
+	details, _ := json.Marshal(map[string]any{
+		"state":                 string(status.State),
+		"consumed_at":           status.Since.UTC().Format(time.RFC3339),
+		"reports_since_consume": status.ReportsSinceConsume,
+	})
+	c.emit(customerID, eventDeliveryStuck, "warning", msg, string(details))
+}
+
+// maybeHeal fires the R-71c self-heal when the burned-credential shape is unambiguous:
+// consumed ≥ stuckAfter ago, ≥ healMinReports consecutive reports since with ZERO offbox evidence,
+// one heal per healCooldown — and the R-39(a) guard holds at act time.
+func (c *OffsiteDeliveryChecker) maybeHeal(customerID string, status offsite.DeliveryStatus) {
+	if c.reissuer == nil {
+		return
+	}
+	if status.ReportsSinceConsume < healMinReports || status.OffsiteReportsSinceConsume != 0 {
+		return // box not reporting enough, or offbox evidence exists (regressed-apply shape) → operator's call
+	}
+	last, err := c.store.LastEventAt(customerID, eventCredentialRestaged)
+	if err != nil {
+		c.logger.Printf("[WARN] offsite-delivery: %s: heal rate-limit read: %v", customerID, err)
+		return
+	}
+	if !last.IsZero() && c.now().Sub(last) < healCooldown {
+		return // one restage per customer per 24 h — repeats surface as repeated events only
+	}
+	// THE R-39(a) GUARD — re-read the secret row immediately before acting. The derive above is a
+	// snapshot; an operator Re-issue may have staged a FRESH UNCONSUMED secret since (TOCTOU).
+	// SaveOneTimeSecret clobbers by design, so acting now would burn that fresh password.
+	info, err := c.store.GetOneTimeSecretInfo(customerID)
+	if err != nil {
+		c.logger.Printf("[WARN] offsite-delivery: %s: guard read: %v", customerID, err)
+		return
+	}
+	if info == nil || info.ConsumedAt.IsZero() {
+		c.logger.Printf("[INFO] offsite-delivery: %s: heal refused — secret row is now %s (R-39(a) guard: never restage over an unconsumed secret)",
+			customerID, secretShape(info))
+		return
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
+	defer cancel()
+	if err := c.reissuer.ReissueOffsiteForCustomer(ctx, customerID); err != nil {
+		c.logger.Printf("[WARN] offsite-delivery: %s: self-heal reissue failed: %v", customerID, err)
+		return
+	}
+	c.logger.Printf("[INFO] offsite-delivery: %s: self-heal restage fired (consumed_at %s, %d offbox-less reports since)",
+		customerID, status.Since.UTC().Format(time.RFC3339), status.ReportsSinceConsume)
+	details, _ := json.Marshal(map[string]any{
+		"burned_consumed_at":    status.Since.UTC().Format(time.RFC3339),
+		"reports_since_consume": status.ReportsSinceConsume,
+	})
+	c.emit(customerID, eventCredentialRestaged, "warning",
+		"Offsite credential re-staged automatically: the previous one-time password was consumed but never applied (burned mid-delivery). The box picks the fresh password up on its next config refresh.",
+		string(details))
+}
+
+func secretShape(info *store.OneTimeSecretInfo) string {
+	if info == nil {
+		return "absent"
+	}
+	return "unconsumed (staged " + info.CreatedAt.UTC().Format(time.RFC3339) + ")"
+}
+
+// emit saves the event (audit trail first) and then notifies — the OffsiteChecker convention:
+// SaveEvent failure logs and SKIPS the notification (an email without its audit row lies).
+func (c *OffsiteDeliveryChecker) emit(customerID, eventType, severity, message, details string) {
+	if _, err := c.store.SaveEvent(customerID, eventType, severity, message, details, "hub"); err != nil {
+		c.logger.Printf("[WARN] offsite-delivery: %s: save %s: %v", customerID, eventType, err)
+		return
+	}
+	c.logger.Printf("[INFO] offsite-delivery: %s: %s (%s)", customerID, eventType, severity)
+	if c.onEvent != nil {
+		c.onEvent(customerID, eventType, severity, message, details, "hub")
+	}
+}
@@ -0,0 +1,302 @@
+package monitor
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"log"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"gitea.dooplex.hu/admin/felhom-hub/internal/store"
+)
+
+// R-70/R-71c checker tests. Real store, fake reissuer (records calls AND mimics the production
+// side effect — a restage IS a SaveOneTimeSecret — so a clobber is observable on the row itself),
+// captured onEvent, injected clock.
+
+const (
+	dtReportNoOffsite   = `{"health":{"status":"ok"}}`
+	dtReportWithOffsite = `{"health":{"status":"ok"},"offsite":{"enabled":true}}`
+	dtOffsiteConfig     = `{"offsite":{"enabled":true,"type":"shared","host":"h","user":"u","repo_path":"/home/felhom-repo","quota_gb":50}}`
+)
+
+type fakeReissuer struct {
+	mu    sync.Mutex
+	st    *store.Store
+	calls []string
+}
+
+func (f *fakeReissuer) ReissueOffsiteForCustomer(_ context.Context, customerID string) error {
+	f.mu.Lock()
+	f.calls = append(f.calls, customerID)
+	f.mu.Unlock()
+	// The production path's essential side effect: ReissueCredentials → SaveOneTimeSecret
+	// (last-write-wins clobber). Mimicked so the R-39(a) tests can observe what a wrongly-fired
+	// heal would DO to the row, not merely that it was called.
+	return f.st.SaveOneTimeSecret(customerID, "fresh-from-heal")
+}
+
+func (f *fakeReissuer) count() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return len(f.calls)
+}
+
+type dtHarness struct {
+	st       *store.Store
+	reissuer *fakeReissuer
+	checker  *OffsiteDeliveryChecker
+	events   *[]string // "type:severity"
+}
+
+func newDTHarness(t *testing.T, withReissuer bool) dtHarness {
+	t.Helper()
+	st, err := store.New(filepath.Join(t.TempDir(), "d.db"), log.New(io.Discard, "", 0))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { st.Close() })
+	if err := st.SaveCustomerConfig(&store.CustomerConfig{
+		CustomerID: "c1", CustomerName: "C", Domain: "c1.hu", APIKey: "k", RetrievalPassword: "p",
+		ConfigJSON: dtOffsiteConfig,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	events := &[]string{}
+	var mu sync.Mutex
+	onEvent := func(_, eventType, severity, _, _, _ string) {
+		mu.Lock()
+		*events = append(*events, eventType+":"+severity)
+		mu.Unlock()
+	}
+	var ri *fakeReissuer
+	var riIface OffsiteReissuer
+	if withReissuer {
+		ri = &fakeReissuer{st: st}
+		riIface = ri
+	}
+	c := NewOffsiteDeliveryChecker(st, riIface, onEvent, log.New(io.Discard, "", 0))
+	return dtHarness{st: st, reissuer: ri, checker: c, events: events}
+}
+
+// burnedFixture puts c1 into the F10 shape: consumed >1h ago, N offbox-less reports since.
+func (h dtHarness) burnedFixture(t *testing.T, reports int) {
+	t.Helper()
+	if err := h.st.SaveOneTimeSecret("c1", "x"); err != nil {
+		t.Fatal(err)
+	}
+	consumed := time.Now().UTC().Add(-2 * time.Hour).Format("2006-01-02 15:04:05")
+	staged := time.Now().UTC().Add(-3 * time.Hour).Format("2006-01-02 15:04:05")
+	if err := h.st.SetOneTimeSecretTimesForTest("c1", staged, consumed); err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < reports; i++ {
+		if err := h.st.SaveReport("c1", []byte(dtReportNoOffsite)); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func (h dtHarness) savedEvents(t *testing.T, eventType string) []store.Event {
+	t.Helper()
+	all, err := h.st.GetRecentEvents("c1", 50)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var out []store.Event
+	for _, e := range all {
+		if e.EventType == eventType {
+			out = append(out, e)
+		}
+	}
+	return out
+}
+
+// Scenario: the stuck event fires once with severity WARNING, and the durable 24h cooldown holds
+// across a second pass (delete the LastEventAt guard in maybeEmitStuck → this fails with 2 events —
+// the executable red-proof of the cooldown).
+func TestDeliveryChecker_StuckEvent_WarningOncePer24h(t *testing.T) {
+	h := newDTHarness(t, false)
+	h.burnedFixture(t, 5)
+
+	h.checker.Check()
+	h.checker.Check() // same tick shape again — cooldown must swallow it
+
+	saved := h.savedEvents(t, "offsite_delivery_stuck")
+	if len(saved) != 1 {
+		t.Fatalf("stuck events = %d, want exactly 1 (24h per-customer cooldown)", len(saved))
+	}
+	if saved[0].Severity != "warning" {
+		t.Fatalf("severity = %q, want warning (operator email tier; never info-silent, never critical)", saved[0].Severity)
+	}
+	var details map[string]any
+	if err := json.Unmarshal([]byte(saved[0].DetailsJSON), &details); err != nil || details["consumed_at"] == "" {
+		t.Fatalf("details must carry consumed_at (mióta), got %s err %v", saved[0].DetailsJSON, err)
+	}
+	if got := *h.events; len(got) != 1 || got[0] != "offsite_delivery_stuck:warning" {
+		t.Fatalf("dispatched = %v, want exactly [offsite_delivery_stuck:warning]", got)
+	}
+}
+
+// Scenario: the R-71c self-heal fires EXACTLY once — one reissue call, the restaged event
+// (warning), and the 24h rate limit blocks a re-trigger even when the shape recurs (delete the
+// LastEventAt guard in maybeHeal → the second pass fires again and this fails — the rate-limit
+// red-proof).
+func TestDeliveryChecker_SelfHeal_FiresOnceAndRateLimits(t *testing.T) {
+	h := newDTHarness(t, true)
+	h.burnedFixture(t, 4)
+
+	h.checker.Check()
+	if h.reissuer.count() != 1 {
+		t.Fatalf("reissue calls = %d, want 1", h.reissuer.count())
+	}
+	restaged := h.savedEvents(t, "offsite_credential_restaged")
+	if len(restaged) != 1 || restaged[0].Severity != "warning" {
+		t.Fatalf("restaged events = %v, want exactly 1 with severity warning (the operator ALWAYS learns a heal fired)", restaged)
+	}
+
+	// The shape recurs (box burned the fresh one too): back into consumed_awaiting_apply.
+	h.burnedFixture(t, 4)
+	h.checker.Check()
+	if h.reissuer.count() != 1 {
+		t.Fatalf("reissue calls after recurrence = %d, want STILL 1 (one restage per customer per 24h; repeats surface as events only)", h.reissuer.count())
+	}
+}
+
+// Not-enough-evidence gates: under healMinReports offbox-less reports → no heal; any offbox
+// evidence since consume (regressed-apply shape) → no heal. The stuck EVENT still fires (age gate
+// alone) — visibility never waits for the heal's stricter bar.
+func TestDeliveryChecker_SelfHeal_EvidenceGates(t *testing.T) {
+	h := newDTHarness(t, true)
+	h.burnedFixture(t, 3) // 3 < healMinReports
+	h.checker.Check()
+	if h.reissuer.count() != 0 {
+		t.Fatalf("reissue with 3 reports = %d calls, want 0 (needs >= 4)", h.reissuer.count())
+	}
+	if len(h.savedEvents(t, "offsite_delivery_stuck")) != 1 {
+		t.Fatal("stuck event must fire regardless of the heal's stricter evidence bar")
+	}
+
+	// add offbox evidence AFTER consume, then more offbox-less reports — mixed history: no heal
+	if err := h.st.SaveReport("c1", []byte(dtReportWithOffsite)); err != nil {
+		t.Fatal(err)
+	}
+	if err := h.st.SaveReport("c1", []byte(dtReportNoOffsite)); err != nil {
+		t.Fatal(err)
+	}
+	h.checker.Check()
+	if h.reissuer.count() != 0 {
+		t.Fatalf("reissue on mixed offbox history = %d calls, want 0 (regressed-apply is the operator's call)", h.reissuer.count())
+	}
+}
+
+// THE CLOBBER RED-PROOF (R-39(a), mandatory per spec): an operator Re-issue lands between the
+// checker's derive and its act (simulated via the onEvent hook, which runs after the stuck event
+// and before maybeHeal). The act-time guard re-reads the row, finds it UNCONSUMED, and refuses —
+// zero reissue calls, row untouched. Remove the `info.ConsumedAt.IsZero()` refusal in maybeHeal →
+// the fake fires SaveOneTimeSecret and this test FAILS on both assertions (calls=1, row clobbered)
+// — proving the fixture would have been clobbered.
+func TestDeliveryChecker_R39aGuard_NeverRestagesOverUnconsumed(t *testing.T) {
+	h := newDTHarness(t, true)
+	h.burnedFixture(t, 4)
+
+	// The TOCTOU: the moment the stuck event dispatches, the "operator" stages a fresh secret.
+	operatorStaged := "2026-07-23 12:00:00"
+	*h.events = nil
+	base := h.checker.onEvent
+	h.checker.onEvent = func(cid, et, sev, msg, det, src string) {
+		if et == "offsite_delivery_stuck" {
+			if err := h.st.SaveOneTimeSecret("c1", "operator-fresh"); err != nil {
+				t.Errorf("mid-tick stage: %v", err)
+			}
+			if err := h.st.SetOneTimeSecretTimesForTest("c1", operatorStaged, ""); err != nil {
+				t.Errorf("mid-tick stamp: %v", err)
+			}
+		}
+		base(cid, et, sev, msg, det, src)
+	}
+
+	h.checker.Check()
+
+	if h.reissuer.count() != 0 {
+		t.Fatalf("reissue calls = %d, want 0 — the R-39(a) guard must refuse over an unconsumed secret", h.reissuer.count())
+	}
+	info, err := h.st.GetOneTimeSecretInfo("c1")
+	if err != nil || info == nil {
+		t.Fatalf("secret row: %v / %v", info, err)
+	}
+	if !info.ConsumedAt.IsZero() || !info.CreatedAt.Equal(time.Date(2026, 7, 23, 12, 0, 0, 0, time.UTC)) {
+		t.Fatalf("the operator's fresh secret was CLOBBERED (created_at=%v consumed_at=%v) — R-39(a) violated", info.CreatedAt, info.ConsumedAt)
+	}
+	if len(h.savedEvents(t, "offsite_credential_restaged")) != 0 {
+		t.Fatal("no restaged event may exist for a refused heal")
+	}
+}
+
+// The demo-felhom live shape, full Check(): applied + stale unconsumed staged secret → ZERO events,
+// ZERO reissue calls, row untouched. Precedence (applied wins) is the first line of defense; the
+// R-39(a) guard is the second.
+func TestDeliveryChecker_AppliedWithStaleStaged_Untouched(t *testing.T) {
+	h := newDTHarness(t, true)
+	if err := h.st.SaveReport("c1", []byte(dtReportWithOffsite)); err != nil {
+		t.Fatal(err)
+	}
+	if err := h.st.SaveOneTimeSecret("c1", "stale"); err != nil {
+		t.Fatal(err)
+	}
+	if err := h.st.SetOneTimeSecretTimesForTest("c1", "2026-07-21 08:29:29", ""); err != nil {
+		t.Fatal(err)
+	}
+
+	h.checker.Check()
+
+	if h.reissuer.count() != 0 {
+		t.Fatalf("reissue calls = %d, want 0 (demo-felhom shape is healthy)", h.reissuer.count())
+	}
+	if len(*h.events) != 0 {
+		t.Fatalf("events = %v, want none", *h.events)
+	}
+	info, _ := h.st.GetOneTimeSecretInfo("c1")
+	if info == nil || !info.ConsumedAt.IsZero() || !info.CreatedAt.Equal(time.Date(2026, 7, 21, 8, 29, 29, 0, time.UTC)) {
+		t.Fatalf("fixture row mutated: %+v — the live specimen must survive the checker untouched", info)
+	}
+}
+
+// Guard rails: young consumed state (inside stuckAfter) is silent; offsite-disabled and
+// non-active customers are skipped entirely.
+func TestDeliveryChecker_QuietShapes(t *testing.T) {
+	h := newDTHarness(t, true)
+	// consumed 5 minutes ago — normal convergence window
+	if err := h.st.SaveOneTimeSecret("c1", "x"); err != nil {
+		t.Fatal(err)
+	}
+	recent := time.Now().UTC().Add(-5 * time.Minute).Format("2006-01-02 15:04:05")
+	if err := h.st.SetOneTimeSecretTimesForTest("c1", recent, recent); err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < 5; i++ {
+		if err := h.st.SaveReport("c1", []byte(dtReportNoOffsite)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	h.checker.Check()
+	if len(*h.events) != 0 || h.reissuer.count() != 0 {
+		t.Fatalf("young consumed state must be silent, got events=%v calls=%d", *h.events, h.reissuer.count())
+	}
+
+	// offsite disabled → skipped even in a stuck-looking shape
+	if err := h.st.SaveCustomerConfig(&store.CustomerConfig{
+		CustomerID: "c1", CustomerName: "C", Domain: "c1.hu", APIKey: "k", RetrievalPassword: "p",
+		ConfigJSON: `{"offsite":{"enabled":false}}`,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	h.burnedFixture(t, 5)
+	h.checker.Check()
+	if len(*h.events) != 0 || h.reissuer.count() != 0 {
+		t.Fatalf("disabled offsite must be skipped, got events=%v calls=%d", *h.events, h.reissuer.count())
+	}
+}