package web // PBS DR tier (SLICE 1): the hub-side provisioning flow behind the customer config form's // "PBS DR tier (ep0)" section. On enable+save: verify the host's WG peer exists (the agent // self-registers its pubkey — absence is fail-closed, nothing to allocate here) → provision the // per-customer ep0 tenancy over the tenantsync channel → store the token secret CONSUME-ONCE // (host-scoped; the agent fetches it via POST /api/v1/hosts/{id}/pbs/consume-token) → merge the // NON-SECRET descriptor into the host's desired_json under "pbs_dr" (the admin-set path; // SetHostDesired bumps the generation, which is the agent's change signal). // // Fail-closed like applyOffsite: any error means NO descriptor write, NO bump, NO half-enabled // state. Idempotent: an already-provisioned descriptor short-circuits — no tenantsync call, no // second secret, no spurious bump. Disable rewrites the descriptor with enabled=false (bump) — // the ep0 namespace/token are NOT deprovisioned (the offsite-disable precedent: data deletion is // a deliberate, separate decision). import ( "context" "database/sql" "encoding/json" "errors" "fmt" "net/http" "regexp" "strings" "time" "gitea.dooplex.hu/admin/felhom-hub/internal/store" "gitea.dooplex.hu/admin/felhom-hub/internal/tenantsync" ) // tenancyProvisioner is the tenantsync seam — satisfied by *tenantsync.Client; tests inject a fake. type tenancyProvisioner interface { Provision(ctx context.Context, customerID string) (*tenantsync.Result, error) Reissue(ctx context.Context, customerID string) (*tenantsync.Result, error) } // SetTenantSync enables PBS DR tier provisioning (optional). Without it, saving a config with the // tier enabled returns an error (not configured on this hub); the form section still renders. func (s *Server) SetTenantSync(p tenancyProvisioner) { s.tenantsync = p } // pbsDRDescriptor is the NON-SECRET pbs_dr block in a host's desired_json. It NEVER carries the // token secret (that is host_pbs_secrets custody, consume-once). type pbsDRDescriptor struct { Enabled bool `json:"enabled"` StorageID string `json:"storage_id,omitempty"` PBSTunnelIP string `json:"pbs_tunnel_ip,omitempty"` Datastore string `json:"datastore,omitempty"` Namespace string `json:"namespace,omitempty"` TokenID string `json:"token_id,omitempty"` Fingerprint string `json:"fingerprint,omitempty"` } // defaultPBSStorageID is the storage-entry id the agent bridge creates on a customer box. The DEMO // host's adopted manual entry is `felhom-offsite` — the descriptor carries the id so the bridge is // name-agnostic and the felhom-pbs collision on the demo dissolves (spike naming note). const defaultPBSStorageID = "felhom-pbs" // pveStorageIDRe validates the operator-typed storage id (PVE storage-id grammar, conservative). var pveStorageIDRe = regexp.MustCompile(`^[A-Za-z][A-Za-z0-9_.-]{0,27}$`) // readPBSDR extracts the pbs_dr descriptor from a host's desired_json ("" / absent → nil). func readPBSDR(desiredJSON string) *pbsDRDescriptor { var doc struct { PBSDR *pbsDRDescriptor `json:"pbs_dr"` } if err := json.Unmarshal([]byte(desiredJSON), &doc); err != nil { return nil } return doc.PBSDR } // mergePBSDR merges the descriptor under the "pbs_dr" key of a desired_json object, preserving // every other key (the operator blob, dr blocks…). Same shape as offsite.MergeDescriptor. func mergePBSDR(desiredJSON string, d *pbsDRDescriptor) (string, error) { obj := map[string]json.RawMessage{} if strings.TrimSpace(desiredJSON) != "" && desiredJSON != "{}" { if err := json.Unmarshal([]byte(desiredJSON), &obj); err != nil { return "", fmt.Errorf("pbsdr: parse desired_json: %w", err) } } db, err := json.Marshal(d) if err != nil { return "", err } obj["pbs_dr"] = db out, err := json.Marshal(obj) if err != nil { return "", err } return string(out), nil } // applyPBSDR handles the config form's DR-tier section on create/update. Called BEFORE // SaveCustomerConfig (fail-closed on REAL provisioning failures: those abort the whole save). // The descriptor lives in the HOST's desired_json — ConfigJSON never carries pbs_dr. // // v0.51.0 (DR-tier-by-default): the switch is cfg.DRTier (the stored per-customer flag, set from // the form by the handler) — and an UNMET PRECONDITION is no longer an error. New customers // default the flag ON before any host exists; the cascade converges later: host enrolls → WG // peer registers (the api hook auto-provisions, scenario A) → descriptor applies on the agent's // next tick. The form save stores the intent and reports the honest waiting stage; only an // actual provisioning FAILURE (tenantsync error, token mismatch…) still fails the save. func (s *Server) applyPBSDR(ctx context.Context, r *http.Request, cfg *store.CustomerConfig) error { enabled := cfg.DRTier storageID := strings.TrimSpace(r.FormValue("pbsdr_storage_id")) if storageID == "" { storageID = defaultPBSStorageID } host, err := s.store.GetHostByCustomer(cfg.CustomerID) if err != nil { return fmt.Errorf("pbsdr: host lookup: %w", err) } if host == nil { if enabled { s.logger.Printf("[INFO] pbsdr: DR tier ON for %s, no host enrolled yet — the descriptor applies once the cascade is ready (host → WG peer → apply)", cfg.CustomerID) } return nil // the flag is stored; nothing host-side to converge yet } cur := readPBSDR(host.DesiredJSON) if !enabled { // Disable = descriptor enabled:false (coords kept; NO ep0 deprovision). Only when there is // something to disable — otherwise a pure no-op (no spurious bump). if cur == nil || !cur.Enabled { return nil } cur.Enabled = false merged, err := mergePBSDR(host.DesiredJSON, cur) if err != nil { return err } gen, err := s.store.SetHostDesired(host.HostID, []byte(merged)) if err != nil { return fmt.Errorf("pbsdr: desired-state write: %w", err) } s.logger.Printf("[INFO] pbsdr disabled for %s (host %s, gen %d; ep0 tenancy kept)", cfg.CustomerID, host.HostID, gen) return nil } if !pveStorageIDRe.MatchString(storageID) { return fmt.Errorf("invalid PBS storage id %q", storageID) } // Already provisioned → success-no-op (descriptor re-served unchanged; NO new secret, no // tenantsync call). Only an actual change (re-enable, storage id edit) rewrites + bumps. if cur != nil && cur.Namespace != "" { if cur.Enabled && cur.StorageID == storageID { return nil } cur.Enabled = true cur.StorageID = storageID merged, err := mergePBSDR(host.DesiredJSON, cur) if err != nil { return err } gen, err := s.store.SetHostDesired(host.HostID, []byte(merged)) if err != nil { return fmt.Errorf("pbsdr: desired-state write: %w", err) } s.logger.Printf("[INFO] pbsdr descriptor updated for %s (host %s, gen %d; tenancy unchanged)", cfg.CustomerID, host.HostID, gen) return nil } // Fresh provision — the shared atom (also fired by the WG-registration hook, scenario A). // An unmet precondition is an honest waiting stage, never a save-blocking error. blocked, err := s.pbsdrProvisionAtom(ctx, cfg.CustomerID, host, storageID) if err != nil { return err } if blocked != "" { s.logger.Printf("[INFO] pbsdr: DR tier ON for %s — waiting: %s", cfg.CustomerID, blocked) } return nil } // pbsdrProvisionAtom runs the fresh-provision cascade step for a host with NO provisioned // descriptor yet: preconditions → tenantsync provision → consume-once secret → descriptor merge // + generation bump. Returns (blockedReason, err): a non-empty blockedReason is an HONEST // waiting stage (tunnel peer absent, endpoint unset, tenantsync unconfigured — the cascade // retries at its next opportunity); err is a REAL provisioning failure and must stay loud // (fail-closed on the form path, error-logged on the hook path). func (s *Server) pbsdrProvisionAtom(ctx context.Context, customerID string, host *store.Host, storageID string) (string, error) { if s.tenantsync == nil { return "PBS DR provisioning is not configured on this hub (no tenantsync key)", nil } if _, err := s.store.GetWGPeerForHost(host.HostID); err == sql.ErrNoRows { return fmt.Sprintf("host %s has not reported a WG key yet — the tunnel peer must exist before the PBS DR tier", host.HostID), nil } else if err != nil { return "", fmt.Errorf("pbsdr: wg peer lookup: %w", err) } ep, err := s.store.GetWGEndpoint() if err == sql.ErrNoRows { return "wg endpoint not configured — register ep0 before the PBS DR tier", nil } else if err != nil { return "", fmt.Errorf("pbsdr: wg endpoint read: %w", err) } // Detach from the request context (the applyOffsite F1 precedent): once provisioning starts, // provision→store-secret→descriptor must complete; an impatient re-click must not cancel // between the ep0 mutation and the consume-once store write (a stranded token would need a // manual re-issue). The absolute timeout still bounds a hung SSH exec. ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Minute) defer cancel() res, err := s.tenantsync.Provision(ctx, customerID) if errors.Is(err, tenantsync.ErrTokenExists) { // ep0 has a token but the hub has no descriptor — state mismatch (lost hub state or a // half-torn earlier attempt). Never silently re-key: the operator decides via Re-issue. return "", fmt.Errorf("the endpoint already holds a PBS token for %s but the hub has no descriptor — use the explicit \"Re-issue PBS credentials\" action", customerID) } if err != nil { return "", err } // The atom: secret first (consume-once custody), then descriptor+bump (the agent's signal). if err := s.store.SaveHostPBSSecret(host.HostID, res.TokenSecret); err != nil { return "", fmt.Errorf("pbsdr: store one-time token secret: %w", err) } desc := &pbsDRDescriptor{ Enabled: true, StorageID: storageID, PBSTunnelIP: ep.PBSTunnelIP, Datastore: res.Datastore, Namespace: res.Namespace, TokenID: res.TokenID, Fingerprint: res.Fingerprint, } merged, err := mergePBSDR(host.DesiredJSON, desc) if err != nil { return "", err } gen, err := s.store.SetHostDesired(host.HostID, []byte(merged)) if err != nil { return "", fmt.Errorf("pbsdr: desired-state write: %w", err) } s.logger.Printf("[INFO] pbsdr provisioned for %s (host %s, ns %s, token_id %s, gen %d; secret stored consume-once, withheld from logs)", customerID, host.HostID, res.Namespace, res.TokenID, gen) return "", nil } // PBSDRAutoProvision is the WG-registration hook (v0.51.0, scenario A): the api handler calls it // (via the main.go closure) right after a host's first WG peer registration. If the customer's // DR-tier flag is ON and no descriptor is provisioned yet, it runs the provisioning atom — the // descriptor then applies on the agent's next desired-state tick with ZERO operator steps. // Never fails the caller: outcomes are logged (an error here is retried by the next config save). func (s *Server) PBSDRAutoProvision(ctx context.Context, customerID string) { cfg, err := s.store.GetCustomerConfig(customerID) if err != nil || cfg == nil || !cfg.DRTier { return } host, err := s.store.GetHostByCustomer(customerID) if err != nil || host == nil { return } if cur := readPBSDR(host.DesiredJSON); cur != nil && cur.Namespace != "" { return // already provisioned — nothing to converge here } storageID := defaultPBSStorageID if cur := readPBSDR(host.DesiredJSON); cur != nil && cur.StorageID != "" { storageID = cur.StorageID } blocked, err := s.pbsdrProvisionAtom(ctx, customerID, host, storageID) if err != nil { s.logger.Printf("[ERROR] pbsdr auto-provision for %s (WG-registration hook): %v — save the customer config to retry", customerID, err) return } if blocked != "" { s.logger.Printf("[INFO] pbsdr auto-provision for %s still waiting: %s", customerID, blocked) return } s.logger.Printf("[INFO] pbsdr auto-provisioned for %s on WG registration (hands-free cascade)", customerID) } // handlePBSDRReissue explicitly re-keys the customer's ep0 PBS token (the offsite F4 precedent): // tenantsync reissue → fresh consume-once secret → descriptor refresh + generation bump so the // agent re-runs its bridge and consumes the fresh secret. The secret value is never logged. func (s *Server) handlePBSDRReissue(w http.ResponseWriter, r *http.Request, customerID string) { if s.tenantsync == nil { http.Error(w, "PBS DR provisioning is not configured on this hub", http.StatusBadGateway) return } cfg, err := s.store.GetCustomerConfig(customerID) if err != nil || cfg == nil { http.NotFound(w, r) return } host, err := s.store.GetHostByCustomer(customerID) if err != nil || host == nil { http.Error(w, "No host enrolled for this customer", http.StatusBadRequest) return } cur := readPBSDR(host.DesiredJSON) if cur == nil || cur.Namespace == "" { http.Error(w, "No provisioned PBS DR tier for this customer", http.StatusBadRequest) return } // Same detached-ctx discipline as applyPBSDR: reissue→store→bump is the atom. ctx, cancel := context.WithTimeout(context.WithoutCancel(r.Context()), 2*time.Minute) defer cancel() res, err := s.tenantsync.Reissue(ctx, customerID) if err != nil { s.logger.Printf("[ERROR] pbsdr reissue for %s: %v", customerID, err) http.Error(w, "PBS credential re-issue failed: "+err.Error(), http.StatusBadGateway) return } if err := s.store.SaveHostPBSSecret(host.HostID, res.TokenSecret); err != nil { s.logger.Printf("[ERROR] pbsdr reissue for %s: secret store: %v", customerID, err) http.Error(w, "Re-issued on the endpoint but storing the secret failed — re-issue again", http.StatusInternalServerError) return } cur.TokenID = res.TokenID cur.Fingerprint = res.Fingerprint cur.Datastore = res.Datastore cur.Namespace = res.Namespace merged, err := mergePBSDR(host.DesiredJSON, cur) if err == nil { _, err = s.store.SetHostDesired(host.HostID, []byte(merged)) } if err != nil { s.logger.Printf("[ERROR] pbsdr reissue for %s: descriptor bump: %v", customerID, err) http.Error(w, "Credential re-issued but the descriptor bump failed — save the config once to trigger the pickup", http.StatusInternalServerError) return } s.logger.Printf("[INFO] pbsdr credentials re-issued for %s (host %s; fresh consume-once secret stored)", customerID, host.HostID) http.Redirect(w, r, "/customers/"+customerID+"?flash=pbsdr_reissued#tab=edit", http.StatusSeeOther) } // pbsDRView is the config form's render model for the DR-tier section, including the v0.51.0 // cascade stages (host → WG peer → descriptor → escrow) — one flag, ordered rollout, honest // intermediate states (scenario D). type pbsDRView struct { Supported bool // tenantsync configured on this hub NoHost bool // no enrolled host for the customer (cascade stage 1 waiting) HostID string DRTier bool // the stored per-customer flag (operator INTENT) Enabled bool // descriptor enabled (host-side reality) StorageID string // current or the default Provisioned bool Namespace string TokenID string WGPeer bool // cascade stage 2: the host has registered its tunnel peer EscrowPresent bool // cascade stage 4: the ceremony ran (blob in custody) } // pbsDRViewFor loads the section state for the form. Read-only; every error degrades to a // zero-ish view (the section still renders). drTier is the customer's stored flag (the caller // has the config; "" customerID = the create form). func (s *Server) pbsDRViewFor(customerID string, drTier bool) pbsDRView { v := pbsDRView{Supported: s.tenantsync != nil, StorageID: defaultPBSStorageID, DRTier: drTier} if customerID == "" { v.NoHost = true return v } host, err := s.store.GetHostByCustomer(customerID) if err != nil || host == nil { v.NoHost = true return v } v.HostID = host.HostID if _, err := s.store.GetWGPeerForHost(host.HostID); err == nil { v.WGPeer = true } if escrow, err := s.store.GetHostEscrow(host.HostID); err == nil && escrow != nil { v.EscrowPresent = true } if d := readPBSDR(host.DesiredJSON); d != nil { v.Enabled = d.Enabled if d.StorageID != "" { v.StorageID = d.StorageID } v.Provisioned = d.Namespace != "" v.Namespace = d.Namespace v.TokenID = d.TokenID } return v }