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 PBS DR section on create/update. Called BEFORE // SaveCustomerConfig (fail-closed: an error must abort the whole save). The descriptor lives in // the HOST's desired_json — ConfigJSON never carries pbs_dr (single source of truth). func (s *Server) applyPBSDR(ctx context.Context, r *http.Request, cfg *store.CustomerConfig) error { enabled := r.FormValue("pbsdr_enabled") == "on" || r.FormValue("pbsdr_enabled") == "true" 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 { return nil // nothing enrolled, nothing enabled — nothing to do } return fmt.Errorf("no host enrolled for customer %s yet — the PBS DR tier needs the enrolled host", cfg.CustomerID) } 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. Fail-closed preconditions first. if s.tenantsync == nil { return fmt.Errorf("PBS DR provisioning is not configured on this hub (no tenantsync key)") } if _, err := s.store.GetWGPeerForHost(host.HostID); err == sql.ErrNoRows { return fmt.Errorf("host %s has not reported a WG key yet — the tunnel peer must exist before the PBS DR tier", host.HostID) } else if err != nil { return fmt.Errorf("pbsdr: wg peer lookup: %w", err) } ep, err := s.store.GetWGEndpoint() if err == sql.ErrNoRows { return fmt.Errorf("wg endpoint not configured — register ep0 before the PBS DR tier") } 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, cfg.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", cfg.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)", cfg.CustomerID, host.HostID, res.Namespace, res.TokenID, gen) return nil } // 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", http.StatusSeeOther) } // pbsDRView is the config form's render model for the PBS DR section. type pbsDRView struct { Supported bool // tenantsync configured on this hub NoHost bool // no enrolled host for the customer (enable would fail-closed) HostID string Enabled bool StorageID string // current or the default Provisioned bool Namespace string TokenID string } // pbsDRViewFor loads the section state for the form. Read-only; every error degrades to a // zero-ish view (the section still renders). func (s *Server) pbsDRViewFor(customerID string) pbsDRView { v := pbsDRView{Supported: s.tenantsync != nil, StorageID: defaultPBSStorageID} 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 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 }