felhom-agent/internal/pbsdr/manager.go

// Package pbsdr is the PBS-DR-tier apply-bridge (slice 2) — the host-side twin of the
// controller's offsite apply-bridge. It consumes the hub's `pbs_dr` desired-state descriptor
// (slice 1) and converges the box: the pbs storage entry exists, K exists, the agent token can
// write to it, and the ceremony one-liner finds its storage id.
//
// The laws this package encodes (spike SPIKE-pbs-tier-provisioning-2026-07-10 + the offsite
// bridge precedent — do not "simplify" any of them):
//
//   - SET-ONLY: re-apply never removes the storage entry — entry deletion destroys K
//     (un-decryptable backups). The wrapper has no deletion verb; this package never asks for one.
//   - ADOPTION FIRST: an existing healthy entry is adopted non-destructively — verified, granted,
//     marked — with NO consume. Tenancy identity (namespace/username/datastore) is ENTRY-OWNED on
//     adoption; a descriptor that names a different namespace does not repoint a live entry (the
//     demo's manually-built felhom-offsite tenancy is the canonical case).
//   - VERIFY-PIN-BEFORE-CONSUME: the PBS fingerprint is probed over the tunnel against the
//     descriptor BEFORE the one-time secret is consumed (the offsite ordering law).
//   - CONSUMED-BUT-FAILED IS LOUD: after a consume, any failure lands in a persistent alarming
//     report state. The bridge never silently retries a burned secret; it recovers ONLY when the
//     operator stages a fresh one (hub Re-issue) — the next consume then succeeds.
//   - The secret rides the wrapper's STDIN (sudo logs argv) and is never logged.
package pbsdr

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net/netip"
	"os"
	"path/filepath"
	"regexp"
	"strings"
	"sync"
	"time"

	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
	"gitea.dooplex.hu/admin/felhom-agent/internal/pbs"
	"gitea.dooplex.hu/admin/felhom-agent/internal/proxmox"
)

// WrapperPath is the pinned sudoers vector (configs/felhom-pbs-apply).
const WrapperPath = "/usr/local/sbin/felhom-pbs-apply"

// StorageReader is the PVE read seam (satisfied by *proxmox.Client; tests fake it).
type StorageReader interface {
	StorageEntry(ctx context.Context, id string) (*proxmox.StorageEntryConfig, bool, error)
	StorageActive(ctx context.Context, id string) (bool, error)
}

// SecretConsumer is the hub consume-once seam (satisfied by *hub.Client; tests fake it).
type SecretConsumer interface {
	ConsumePBSToken(ctx context.Context) (string, error)
}

var storageIDRe = regexp.MustCompile(`^[A-Za-z][A-Za-z0-9_.-]{0,27}$`)
var nameRe = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9_.-]{0,31}$`)
var tokenIDRe = regexp.MustCompile(`^[A-Za-z0-9_.-]+@[A-Za-z0-9]+![A-Za-z0-9_.-]+$`)
var fingerprintRe = regexp.MustCompile(`^([A-Fa-f0-9]{2}:){31}[A-Fa-f0-9]{2}$`)

const markerName = "marker.json"
const consumedFailedName = "consumed-failed.json"

// marker is the idempotency record: the descriptor hash last converged + how.
type marker struct {
	Hash      string `json:"hash"`
	State     string `json:"state"` // "adopted" | "applied"
	AppliedAt string `json:"applied_at"`
}

// consumedFailed is the LOUD persistent state: a one-time secret was burned and the apply failed.
type consumedFailed struct {
	Hash    string `json:"hash"`
	Message string `json:"message"`
	At      string `json:"at"`
}

// Manager converges the box toward the pbs_dr descriptor. All deps are seams for tests.
type Manager struct {
	runner     proxmox.Runner
	px         StorageReader
	hub        SecretConsumer
	stateDir   string // <agent-state>/pbsdr
	secretDir  string // cfg.Backup pbs secret dir (the wrapper's copy target)
	configPath string // agent.json — for the escrow.pbs_storage_id seed ("" = no seeding)
	logger     *slog.Logger

	// probeFP is the verify-pin-before-consume seam (default pbs.ProbeFingerprint).
	probeFP func(ctx context.Context, server, fingerprint string) error
	now     func() time.Time

	mu     sync.Mutex
	status *hub.PBSDRStatus // latest snapshot for the report stanza
}

// NewManager builds the bridge manager.
func NewManager(runner proxmox.Runner, px StorageReader, hubc SecretConsumer, stateDir, secretDir, configPath string, logger *slog.Logger) *Manager {
	if logger == nil {
		logger = slog.Default()
	}
	return &Manager{
		runner: runner, px: px, hub: hubc,
		stateDir: filepath.Join(stateDir, "pbsdr"), secretDir: secretDir, configPath: configPath,
		logger:  logger,
		probeFP: pbs.ProbeFingerprint,
		now:     func() time.Time { return time.Now().UTC() },
	}
}

func (m *Manager) markerPath() string         { return filepath.Join(m.stateDir, markerName) }
func (m *Manager) consumedFailedPath() string { return filepath.Join(m.stateDir, consumedFailedName) }

func (m *Manager) setStatus(s *hub.PBSDRStatus) {
	m.mu.Lock()
	m.status = s
	m.mu.Unlock()
}

// Status returns the latest bridge state (the report stanza; nil before the first Apply).
func (m *Manager) Status() *hub.PBSDRStatus {
	m.mu.Lock()
	defer m.mu.Unlock()
	return m.status
}

// DRConfigured reports whether the DR tier is configured ON for this box — the capability
// prober's GatePBSDR answer (v0.86.0). True when the last-seen descriptor was enabled (any live
// status except "disabled"), or, before the first desired-state fetch of this process, when a
// previously-converged marker exists (so an applied box never flaps to inactive across an agent
// restart). False = no descriptor ever / descriptor disabled → healthy pbsdr capabilities report
// "inactive (disabled by configuration)" instead of ok.
func (m *Manager) DRConfigured() bool {
	m.mu.Lock()
	st := m.status
	m.mu.Unlock()
	if st != nil {
		return st.State != "disabled"
	}
	return m.loadMarker() != nil
}

// NoteAuthResult implements pbs.AuthSink (R-39 leg c): the credential probe's verdict for one
// storage, turned into the DR bridge's reported state.
//
// This is the leg that makes `applied` mean something. Until v0.91.0 the agent could not read the
// credential it had written (root-only path, no wrapper read verb), so a tier pinned to a superseded
// secret reported `applied` forever while every PBS request 401'd — and the hub, seeing `applied`,
// had no reason to re-key. Now a rejection becomes a LOUD `auth_failed` that pbsdrheal escalates to
// a fresh mint; the fresh mint advances the secret generation; the descriptor hash moves; and Apply
// finally re-consumes.
//
// Rules that keep it safe:
//   - Only a REJECTION (401) sets the state. An unreachable PBS is UNKNOWN and must never re-key.
//   - Only the storage this box's descriptor actually names is considered; a host may carry other
//     PBS entries that are none of the DR tier's business.
//   - Recovery is self-clearing: a subsequent successful probe restores the converged state from the
//     marker, so the operator does not have to acknowledge a fault that fixed itself.
func (m *Manager) NoteAuthResult(storageID string, unauthorized bool, detail string) {
	m.mu.Lock()
	st := m.status
	m.mu.Unlock()
	// No descriptor seen yet, or this is not our storage → not our business.
	if st == nil || st.StorageID == "" || storageID == "" || st.StorageID != storageID {
		return
	}
	if unauthorized {
		if st.State == "auth_failed" {
			return // already loud; do not churn the report
		}
		m.logger.Error("pbsdr: the DR endpoint REJECTED this box's credential — the tier is applied and DEAD",
			"storage_id", storageID, "previous_state", st.State)
		m.setStatus(&hub.PBSDRStatus{
			State: "auth_failed", StorageID: st.StorageID, Namespace: st.Namespace, AppliedAt: st.AppliedAt,
			AuthFailed: true,
			Message:    detail + " — awaiting fresh credentials from the hub (automatic)",
		})
		return
	}
	// A clean probe clears a previously-loud auth failure by restoring the converged marker state.
	if st.State == "auth_failed" && detail == "" {
		mk := m.loadMarker()
		restored := "applied"
		appliedAt := st.AppliedAt
		if mk != nil {
			restored, appliedAt = mk.State, mk.AppliedAt
		}
		m.logger.Info("pbsdr: credential accepted again — clearing auth_failed", "storage_id", storageID, "state", restored)
		m.setStatus(&hub.PBSDRStatus{State: restored, StorageID: st.StorageID, Namespace: st.Namespace, AppliedAt: appliedAt})
	}
}

// descriptorHash is the idempotency key: sha256 of the canonical (struct-ordered) JSON.
func descriptorHash(b *hub.WirePBSDR) string {
	j, _ := json.Marshal(b)
	sum := sha256.Sum256(j)
	return hex.EncodeToString(sum[:])
}

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.Hash == "" {
		return nil
	}
	return &mk
}

func (m *Manager) loadConsumedFailed() *consumedFailed {
	raw, err := os.ReadFile(m.consumedFailedPath())
	if err != nil {
		return nil
	}
	var cf consumedFailed
	if json.Unmarshal(raw, &cf) != nil {
		return nil
	}
	return &cf
}

func (m *Manager) writeState(path string, v any) error {
	if err := os.MkdirAll(m.stateDir, 0o700); err != nil {
		return err
	}
	raw, err := json.Marshal(v)
	if err != nil {
		return err
	}
	tmp := path + ".tmp"
	if err := os.WriteFile(tmp, raw, 0o600); err != nil {
		return err
	}
	return os.Rename(tmp, path)
}

// validate checks every descriptor field BEFORE any exec (the fine gate under the coarse sudoers).
func validate(b *hub.WirePBSDR) error {
	if !storageIDRe.MatchString(b.StorageID) {
		return fmt.Errorf("bad storage_id %q", b.StorageID)
	}
	if !nameRe.MatchString(b.Datastore) {
		return fmt.Errorf("bad datastore %q", b.Datastore)
	}
	if !nameRe.MatchString(b.Namespace) {
		return fmt.Errorf("bad namespace %q", b.Namespace)
	}
	if !tokenIDRe.MatchString(b.TokenID) {
		return fmt.Errorf("bad token_id %q", b.TokenID)
	}
	if !fingerprintRe.MatchString(b.Fingerprint) {
		return fmt.Errorf("bad fingerprint (want 32-pair colon sha256)")
	}
	if _, err := netip.ParseAddr(b.PBSTunnelIP); err != nil {
		return fmt.Errorf("bad pbs_tunnel_ip %q", b.PBSTunnelIP)
	}
	return nil
}

// Apply converges toward the descriptor. fetched=false (no desired data yet) is never a signal.
func (m *Manager) Apply(ctx context.Context, fetched bool, block *hub.WirePBSDR) {
	if !fetched {
		return
	}
	if block == nil {
		// Absent block: pre-slice-1 hub or tier never enabled — a silent no-op (old-hub compat).
		return
	}
	if !block.Enabled {
		m.setStatus(&hub.PBSDRStatus{State: "disabled", StorageID: block.StorageID, Namespace: block.Namespace})
		return // NO teardown in this slice — deprovision is a deliberate future op
	}
	if err := validate(block); err != nil {
		m.logger.Error("pbsdr: descriptor invalid; refusing", "err", err)
		m.setStatus(&hub.PBSDRStatus{State: "verify_failed", StorageID: block.StorageID, Namespace: block.Namespace,
			Message: "descriptor invalid: " + err.Error()})
		return
	}

	h := descriptorHash(block)
	cf := m.loadConsumedFailed()
	if mk := m.loadMarker(); mk != nil && mk.Hash == h && (cf == nil || cf.Hash != h) {
		m.setStatus(&hub.PBSDRStatus{State: mk.State, StorageID: block.StorageID, Namespace: block.Namespace, AppliedAt: mk.AppliedAt})
		return // idempotent: this exact descriptor already converged
	}

	entry, found, err := m.px.StorageEntry(ctx, block.StorageID)
	if err != nil {
		// R-22 self-grant (F4, tests/VALIDATION-n100-baremetal): on a NON-DEFAULT storage id the
		// agent token holds no ACL on /storage/<id> yet, so this token-auth pre-check
		// (GET /storage/<id>) 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)
		m.logger.Error("pbsdr: " + msg)
		m.setStatus(&hub.PBSDRStatus{State: "verify_failed", StorageID: block.StorageID, Message: msg})
		return
	}

	if found {
		active, err := m.px.StorageActive(ctx, block.StorageID)
		if err != nil {
			m.logger.Warn("pbsdr: storage status probe failed (transient)", "err", err)
			return
		}
		if active {
			m.adopt(ctx, block, entry, h)
			return
		}
		// Exists but unhealthy → the recovery path: verify → consume → reconcile-with-password.
	}

	// VERIFY-PIN-BEFORE-CONSUME (the ordering law): a mismatched or unreachable PBS aborts here —
	// the one-time secret is untouched and the bridge simply retries next tick.
	if err := m.probeFP(ctx, block.PBSTunnelIP, block.Fingerprint); err != nil {
		m.logger.Warn("pbsdr: PBS fingerprint verify failed BEFORE consume (nothing consumed; retrying)", "err", err)
		m.setStatus(&hub.PBSDRStatus{State: "verify_failed", StorageID: block.StorageID, Namespace: block.Namespace,
			Message: "pre-consume fingerprint verify: " + err.Error()})
		return
	}

	secret, err := m.hub.ConsumePBSToken(ctx)
	if errors.Is(err, hub.ErrNoPBSSecret) {
		if cf != nil && cf.Hash == h {
			// The burned-secret dead-end: stay LOUD until the operator re-issues (fresh secret).
			m.setStatus(&hub.PBSDRStatus{State: "consumed_failed", StorageID: block.StorageID, Namespace: block.Namespace,
				ConsumedFailed: true, Message: cf.Message + " — awaiting operator re-issue (hub: Re-issue PBS credentials)"})
			return
		}
		m.setStatus(&hub.PBSDRStatus{State: "waiting_secret", StorageID: block.StorageID, Namespace: block.Namespace,
			Message: "verified; no unconsumed token secret staged on the hub"})
		return
	}
	if err != nil {
		m.logger.Warn("pbsdr: consume-token failed (transient; nothing consumed hub-side on error)", "err", err)
		return
	}
	m.logger.Info("pbsdr: one-time token secret consumed (single-use; value withheld from logs)",
		"storage_id", block.StorageID, "secret_len", len(secret))

	// From here the secret is BURNED — every failure below is the loud persistent state.
	if !found {
		_, errOut, err := m.runner.RunStdin(ctx, strings.NewReader(secret+"\n"), WrapperPath,
			"create", block.StorageID, block.PBSTunnelIP, block.Datastore, block.Namespace,
			block.TokenID, block.Fingerprint, m.secretDir)
		if err != nil {
			m.consumedFail(h, fmt.Sprintf("wrapper create failed: %v (stderr: %s)", err, tail(errOut)), block)
			return
		}
	} else {
		_, errOut, err := m.runner.RunStdin(ctx, strings.NewReader(secret+"\n"), WrapperPath,
			"reconcile", block.StorageID, block.PBSTunnelIP, block.Namespace,
			block.TokenID, block.Fingerprint, m.secretDir)
		if err != nil {
			m.consumedFail(h, fmt.Sprintf("wrapper reconcile failed: %v (stderr: %s)", err, tail(errOut)), block)
			return
		}
	}
	if _, errOut, err := m.runner.Run(ctx, WrapperPath, "grant", block.StorageID); err != nil {
		m.consumedFail(h, fmt.Sprintf("wrapper grant failed: %v (stderr: %s)", err, tail(errOut)), block)
		return
	}
	active, err := m.px.StorageActive(ctx, block.StorageID)
	if err != nil || !active {
		m.consumedFail(h, fmt.Sprintf("post-apply status probe failed (active=%v err=%v)", active, err), block)
		return
	}

	m.finishConverged(block, h, "applied")
}

// adopt is the non-destructive existing-entry path: NO consume, tenancy identity entry-owned.
func (m *Manager) adopt(ctx context.Context, block *hub.WirePBSDR, entry *proxmox.StorageEntryConfig, h string) {
	if _, errOut, err := m.runner.Run(ctx, WrapperPath, "grant", block.StorageID); err != nil {
		// No secret involved — a grant failure is transient, retried next tick.
		m.logger.Warn("pbsdr: adoption grant failed (retrying next tick)", "err", err, "stderr", tail(errOut))
		return
	}
	note := ""
	if entry.Namespace != block.Namespace {
		note = fmt.Sprintf("adopted entry keeps its own tenancy (namespace %q; descriptor says %q — entry wins, never repointed)",
			entry.Namespace, block.Namespace)
		m.logger.Info("pbsdr: " + note)
	}
	m.finishConverged(block, h, "adopted")
	if note != "" {
		st := m.Status()
		st.Message = note
		st.Namespace = entry.Namespace // report the REAL tenancy
		m.setStatus(st)
	}
}

// finishConverged writes the marker, clears any consumed-failed state, seeds the escrow storage
// id, and publishes the converged status.
func (m *Manager) finishConverged(block *hub.WirePBSDR, h, state string) {
	at := m.now().Format(time.RFC3339)
	if err := m.writeState(m.markerPath(), marker{Hash: h, State: state, AppliedAt: at}); err != nil {
		m.logger.Error("pbsdr: marker write failed (converged, but will re-run next tick)", "err", err)
	}
	os.Remove(m.consumedFailedPath())
	msg := ""
	if err := m.seedEscrowStorageID(block.StorageID); err != nil {
		msg = "escrow.pbs_storage_id seed failed: " + err.Error() + " (set it manually before the ceremony)"
		m.logger.Warn("pbsdr: " + msg)
	}
	m.logger.Info("pbsdr: converged", "state", state, "storage_id", block.StorageID)
	m.setStatus(&hub.PBSDRStatus{State: state, StorageID: block.StorageID, Namespace: block.Namespace,
		AppliedAt: at, Message: msg})
}

// consumedFail records the LOUD persistent burned-secret state.
func (m *Manager) consumedFail(h, msg string, block *hub.WirePBSDR) {
	m.logger.Error("pbsdr: CONSUMED-BUT-FAILED — the one-time secret is burned; NOT retrying silently. "+
		"Operator action: Re-issue PBS credentials on the hub.", "detail", msg, "storage_id", block.StorageID)
	if err := m.writeState(m.consumedFailedPath(), consumedFailed{Hash: h, Message: msg, At: m.now().Format(time.RFC3339)}); err != nil {
		m.logger.Error("pbsdr: consumed-failed state write failed", "err", err)
	}
	m.setStatus(&hub.PBSDRStatus{State: "consumed_failed", StorageID: block.StorageID, Namespace: block.Namespace,
		ConsumedFailed: true, Message: msg})
}

// seedEscrowStorageID sets escrow.pbs_storage_id in agent.json when empty/absent — the bare
// `--selftest=escrow-create` one-liner must find its storage with no flags. A different existing
// value is NEVER clobbered (warn-and-keep). Unknown config keys are preserved verbatim
// (map[string]RawMessage read-modify-write, atomic rename).
func (m *Manager) seedEscrowStorageID(storageID string) error {
	if m.configPath == "" {
		return nil
	}
	raw, err := os.ReadFile(m.configPath)
	if err != nil {
		return err
	}
	var doc map[string]json.RawMessage
	if err := json.Unmarshal(raw, &doc); err != nil {
		return fmt.Errorf("parse %s: %w", m.configPath, err)
	}
	var esc map[string]json.RawMessage
	if cur, ok := doc["escrow"]; ok {
		if err := json.Unmarshal(cur, &esc); err != nil {
			return fmt.Errorf("parse escrow section: %w", err)
		}
	} else {
		esc = map[string]json.RawMessage{}
	}
	if cur, ok := esc["pbs_storage_id"]; ok {
		var existing string
		_ = json.Unmarshal(cur, &existing)
		if existing == storageID {
			return nil // already seeded
		}
		if existing != "" {
			m.logger.Warn("pbsdr: escrow.pbs_storage_id already set differently — keeping it",
				"existing", existing, "descriptor", storageID)
			return nil
		}
	}
	idJSON, _ := json.Marshal(storageID)
	esc["pbs_storage_id"] = idJSON
	escJSON, err := json.Marshal(esc)
	if err != nil {
		return err
	}
	doc["escrow"] = escJSON
	out, err := json.MarshalIndent(doc, "", "  ")
	if err != nil {
		return err
	}
	st, err := os.Stat(m.configPath)
	if err != nil {
		return err
	}
	// The config DIRECTORY (/etc/felhom-agent) is root-owned while the FILE is agent-owned —
	// tmp+rename is impossible for the non-root agent (rename mutates the directory; live
	// finding on the demo). So: park a recovery copy in the agent state dir, then rewrite the
	// file IN PLACE (O_TRUNC). The agent is the file's only writer and the content is small;
	// a torn write is recoverable from the parked copy.
	if err := os.MkdirAll(m.stateDir, 0o700); err != nil {
		return err
	}
	if err := os.WriteFile(filepath.Join(m.stateDir, "agent.json.pre-seed"), raw, 0o600); err != nil {
		return fmt.Errorf("parking the pre-seed copy: %w", err)
	}
	f, err := os.OpenFile(m.configPath, os.O_WRONLY|os.O_TRUNC, st.Mode().Perm())
	if err != nil {
		return err
	}
	if _, err := f.Write(out); err != nil {
		f.Close()
		return err
	}
	if err := f.Close(); err != nil {
		return err
	}
	m.logger.Info("pbsdr: seeded escrow.pbs_storage_id (the ceremony one-liner needs no flags)", "storage_id", storageID)
	return nil
}

func tail(b []byte) string {
	s := strings.TrimSpace(string(b))
	if len(s) > 300 {
		s = s[len(s)-300:]
	}
	return s
}