package server import ( "context" "encoding/json" "errors" "fmt" "log" "net/http" "net/url" "strconv" "strings" "sync" "time" "pve-confsync/internal/adapter" "pve-confsync/internal/config" "pve-confsync/internal/engine" "pve-confsync/internal/model" "pve-confsync/internal/service" "pve-confsync/internal/store" ) // readTimeout bounds a comparison (reading every cluster). const readTimeout = 2 * time.Minute // ---- state (phase 1: clusters and visual diff) ---- // handleState returns the last comparison; refresh=1 reads the clusters again. func (s *Server) handleState(w http.ResponseWriter, r *http.Request) { s.stateMu.Lock() defer s.stateMu.Unlock() if s.state == nil || r.URL.Query().Get("refresh") == "1" { cfg, err := s.store.Config(r.Context()) if errors.Is(err, store.ErrNoReference) { // nothing to compare yet: no cluster at all, or no primary chosen cls, _ := s.store.Clusters(r.Context()) setup := "clusters" if len(cls) > 0 { setup = "primary" } writeJSON(w, http.StatusOK, map[string]any{"setup": setup}) return } if err != nil { writeError(w, http.StatusInternalServerError, err.Error(), "config") return } ctx, cancel := context.WithTimeout(r.Context(), readTimeout) defer cancel() s.state, s.stateAt = service.Analyze(ctx, cfg, nil), time.Now() } writeJSON(w, http.StatusOK, map[string]any{"state": s.state}) } // invalidate forgets the cached comparison (configuration changed, apply done). func (s *Server) invalidate() { s.stateMu.Lock() s.state = nil s.stateMu.Unlock() } // remember merges the clients of a partial comparison into the cached one, so // that the screen shows what a dry-run just read. func (s *Server) remember(st *service.State, all bool) { s.stateMu.Lock() defer s.stateMu.Unlock() if all || s.state == nil { if all { s.state, s.stateAt = st, time.Now() } return } s.state.Reference = st.Reference for _, c := range st.Clients { for i := range s.state.Clients { if s.state.Clients[i].Name == c.Name { s.state.Clients[i] = c } } } } // ---- dry-run and apply (phase 2) ---- type selection struct { Scope string `json:"scope"` // global (several targets), client (one target) or object Targets []string `json:"targets"` // global scope: the targets (empty: all of them) Client string `json:"client"` // client and object scopes Objects []string `json:"objects"` // object scope: "kind/id" Skip []model.Kind `json:"skip"` // kinds left out of this run ("not synchronized") Digests map[string]string `json:"digests"` // apply: digest of each reviewed plan Confirm string `json:"confirm"` // apply with SDN changes: target name (primary name for global) } func (sel *selection) check() error { if len(sel.Skip) > 0 { sel.Skip = adapter.Bundle(sel.Skip) } switch sel.Scope { case "global": sel.Client, sel.Objects = "", nil case "client": sel.Objects, sel.Targets = nil, nil if sel.Client == "" { return errors.New("client required") } case "object": sel.Targets = nil if sel.Client == "" || len(sel.Objects) == 0 { return errors.New("client and objects required") } default: return fmt.Errorf("unknown scope %q", sel.Scope) } return nil } func (sel selection) clients() []string { if sel.Client == "" { return sel.Targets } return []string{sel.Client} } // all tells whether the selection reads every target (the cached comparison is replaced). func (sel selection) all() bool { return sel.Scope == "global" && len(sel.Targets) == 0 } // plan reads the clusters and returns the plans of the selection. func (s *Server) plan(ctx context.Context, sel selection) (*config.Config, *service.State, []service.ClientPlan, error) { cfg, err := s.store.Config(ctx) if err != nil { return nil, nil, nil, err } for _, c := range sel.clients() { if _, ok := cfg.Client(c); !ok { return nil, nil, nil, fmt.Errorf("unknown target %q", c) } } rctx, cancel := context.WithTimeout(ctx, readTimeout) defer cancel() st := service.Analyze(rctx, cfg, sel.clients()) s.remember(st, sel.all()) return cfg, st, st.Plans(sel.Objects, sel.Skip), nil } type runDetail struct { Plans []service.ClientPlan `json:"plans"` Results map[string]*clientResult `json:"results,omitempty"` } type clientResult struct { Status string `json:"status"` // running, ok, error, noop, skipped Error string `json:"error,omitempty"` Steps []engine.Step `json:"steps"` } func (s *Server) handleDryRun(w http.ResponseWriter, r *http.Request) { var sel selection if err := readJSON(r, &sel); err != nil { writeError(w, http.StatusBadRequest, err.Error(), "request") return } if err := sel.check(); err != nil { writeError(w, http.StatusBadRequest, err.Error(), "request") return } id := identity(r) _, _, plans, err := s.plan(r.Context(), sel) if err != nil { writeError(w, http.StatusBadRequest, err.Error(), "config") return } status, summary := dryRunSummary(plans) d, _ := json.Marshal(runDetail{Plans: plans}) run := store.Run{User: id.User, Mode: "dry-run", Scope: sel.Scope, Clients: planClients(plans), Objects: sel.Objects, Status: status, Summary: summary, Detail: d} runID, err := s.store.StartRun(r.Context(), run) if err != nil { writeError(w, http.StatusInternalServerError, err.Error(), "store") return } _ = s.store.UpdateRun(r.Context(), runID, status, summary, d) s.audit(r, id, "dry-run", target(sel), true, summary) writeJSON(w, http.StatusOK, map[string]any{"run": runID, "plans": plans, "summary": summary}) } func planClients(plans []service.ClientPlan) []string { out := make([]string, 0, len(plans)) for _, p := range plans { out = append(out, p.Client) } return out } func target(sel selection) string { var t string switch { case sel.Scope == "client": t = sel.Client case sel.Scope == "object": t = sel.Client + ": " + strings.Join(sel.Objects, ", ") case len(sel.Targets) > 0: t = "targets " + strings.Join(sel.Targets, ", ") default: t = "all targets" } if len(sel.Skip) > 0 { var ks []string for _, k := range sel.Skip { ks = append(ks, string(k)) } t += " (not replicated: " + strings.Join(ks, ", ") + ")" } return t } // dryRunSummary counts the changes: "3 changes on 2 clients", or the errors. func dryRunSummary(plans []service.ClientPlan) (string, string) { changes, clients, blocked := 0, 0, 0 for _, p := range plans { if p.Error != "" || len(p.Plan.Errors) > 0 { blocked++ continue } if n := len(p.Plan.Changes); n > 0 { changes += n clients++ } } switch { case blocked > 0: return store.RunBlocked, fmt.Sprintf("%d change(s) on %d target(s), %d target(s) in error", changes, clients, blocked) case changes == 0: return store.RunNoop, "no difference" } return store.RunOK, fmt.Sprintf("%d change(s) on %d target(s)", changes, clients) } // handleApply recomputes the plans, checks that they are the ones reviewed in the // dry-run (digests) and the confirmation of the SDN changes, then applies them in // the background, client after client. Progress goes to the journal. func (s *Server) handleApply(w http.ResponseWriter, r *http.Request) { var sel selection if err := readJSON(r, &sel); err != nil { writeError(w, http.StatusBadRequest, err.Error(), "request") return } if err := sel.check(); err != nil { writeError(w, http.StatusBadRequest, err.Error(), "request") return } id := identity(r) if !s.applyMu.TryLock() { writeError(w, http.StatusConflict, "a replication is already running", "busy") return } started := false defer func() { if !started { s.applyMu.Unlock() } }() cfg, st, plans, err := s.plan(r.Context(), sel) if err != nil { writeError(w, http.StatusBadRequest, err.Error(), "config") return } var todo []service.ClientPlan sdn := false for _, p := range plans { switch { case p.Error != "": s.audit(r, id, "apply", target(sel), false, p.Client+": "+p.Error) writeError(w, http.StatusConflict, p.Client+": "+p.Error, "read") return case len(p.Plan.Errors) > 0: s.audit(r, id, "apply", target(sel), false, p.Client+": plan en erreur") writeError(w, http.StatusConflict, p.Client+": "+strings.Join(p.Plan.Errors, "; "), "plan") return case sel.Digests[p.Client] != p.Digest: s.audit(r, id, "apply", target(sel), false, p.Client+": the state changed since the dry-run") writeJSON(w, http.StatusConflict, map[string]any{"error": p.Client + ": the state changed since the dry-run, run it again", "code": "stale", "plans": plans}) return case p.Plan.Empty(): continue } sdn = sdn || p.SDN todo = append(todo, p) } if len(todo) == 0 { writeJSON(w, http.StatusOK, map[string]any{"run": 0, "summary": "no difference"}) return } if sdn { want := sel.Client if sel.Scope == "global" { want = cfg.Reference.Name } if sel.Confirm != want { writeError(w, http.StatusBadRequest, "SDN changes: confirmation required ("+want+")", "confirm") return } } det := &runDetail{Plans: todo, Results: map[string]*clientResult{}} for _, p := range todo { det.Results[p.Client] = &clientResult{Status: "pending", Steps: []engine.Step{}} } d, _ := json.Marshal(det) runID, err := s.store.StartRun(r.Context(), store.Run{User: id.User, Mode: "apply", Scope: sel.Scope, Clients: planClients(todo), Objects: sel.Objects, Status: store.RunRunning, Summary: "en cours", Detail: d}) if err != nil { writeError(w, http.StatusInternalServerError, err.Error(), "store") return } s.audit(r, id, "apply-start", target(sel), true, fmt.Sprintf("journal #%d", runID)) started = true s.jobs.Add(1) // the request ends now: the job keeps a copy that outlives it (audit IP) rr := r.Clone(context.WithoutCancel(r.Context())) go func() { defer s.jobs.Done() defer s.applyMu.Unlock() s.runApply(rr, id, runID, sel, st, det) }() writeJSON(w, http.StatusAccepted, map[string]any{"run": runID}) } // runApply applies the plans one client after the other; a failed client does // not stop the next ones (each has its own SDN lock and rollback). func (s *Server) runApply(r *http.Request, id Identity, runID int64, sel selection, st *service.State, det *runDetail) { ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) defer cancel() var mu sync.Mutex // det is read by save while a step is appended save := func(status, summary string) { mu.Lock() d, _ := json.Marshal(det) mu.Unlock() if err := s.store.UpdateRun(ctx, runID, status, summary, d); err != nil { log.Printf("journal #%d: %v", runID, err) } } failed, done := 0, 0 for _, p := range det.Plans { res := det.Results[p.Client] res.Status = store.RunRunning save(store.RunRunning, fmt.Sprintf("%s en cours", p.Client)) err := st.Apply(ctx, p, sel.Objects, sel.Skip, func(step engine.Step) { mu.Lock() res.Steps = append(res.Steps, step) mu.Unlock() save(store.RunRunning, fmt.Sprintf("%s : %s", p.Client, step.Action)) }) if err != nil { failed++ res.Status, res.Error = store.RunError, err.Error() log.Printf("apply %s (journal #%d): %v", p.Client, runID, err) s.audit(r, id, "apply", p.Client, false, err.Error()) } else { done++ res.Status = store.RunOK s.audit(r, id, "apply", p.Client, true, fmt.Sprintf("%d change(s), verified", len(p.Plan.Changes))) } } status := store.RunOK summary := fmt.Sprintf("%d target(s) replicated and verified", done) if failed > 0 { status = store.RunError summary = fmt.Sprintf("%d target(s) failed, %d replicated", failed, done) } save(status, summary) s.invalidate() } func (s *Server) handleRuns(w http.ResponseWriter, r *http.Request) { limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) before, _ := strconv.ParseInt(r.URL.Query().Get("before"), 10, 64) runs, err := s.store.Runs(r.Context(), limit, before) if err != nil { writeError(w, http.StatusInternalServerError, err.Error(), "store") return } if runs == nil { runs = []store.Run{} } writeJSON(w, http.StatusOK, map[string]any{"runs": runs}) } func (s *Server) handleRun(w http.ResponseWriter, r *http.Request) { id, _ := strconv.ParseInt(r.PathValue("id"), 10, 64) run, err := s.store.Run(r.Context(), id) if errors.Is(err, store.ErrNotFound) { writeError(w, http.StatusNotFound, "unknown run", "notfound") return } if err != nil { writeError(w, http.StatusInternalServerError, err.Error(), "store") return } writeJSON(w, http.StatusOK, run) } // ---- clusters and settings (admin) ---- func (s *Server) handleClusters(w http.ResponseWriter, r *http.Request) { cls, err := s.store.Clusters(r.Context()) if err != nil { writeError(w, http.StatusInternalServerError, err.Error(), "store") return } if cls == nil { cls = []store.Cluster{} } // connection details are for the administrators if identity(r).Role != store.RoleAdmin { for i, c := range cls { cls[i] = store.Cluster{Name: c.Name, Role: c.Role, URL: c.URL, Mapping: c.Mapping, Policies: c.Policies, LastCheck: c.LastCheck} } } writeJSON(w, http.StatusOK, map[string]any{"clusters": cls}) } func (s *Server) handleSaveCluster(w http.ResponseWriter, r *http.Request) { var c store.Cluster if err := readJSON(r, &c); err != nil { writeError(w, http.StatusBadRequest, err.Error(), "request") return } old := r.PathValue("name") if old == "-" { // creation old = "" } if err := validateCluster(c); err != nil { writeError(w, http.StatusBadRequest, err.Error(), "invalid") return } err := s.store.SaveCluster(r.Context(), old, c) detail := fmt.Sprintf("url=%s role=%s tls=%s ssh=%s:%d", c.URL, c.Role, c.TLS, c.SSHHost, c.SSHPort) if c.Password != "" { detail += " password=changed" } if old != "" && old != c.Name { detail += " renamed from " + old } s.audit(r, Identity{}, "cluster-save", c.Name, err == nil, errText(err, detail)) if err != nil { writeError(w, http.StatusBadRequest, err.Error(), "invalid") return } s.invalidate() writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) } // validateCluster checks the address and the mapping and policies of a cluster. func validateCluster(c store.Cluster) error { u, err := url.Parse(strings.TrimSpace(c.URL)) if err != nil || u.Scheme != "https" || u.Host == "" { return errors.New("url: https://host:8006 expected") } if c.User != "" && c.User != "root@pam" { return errors.New("user: root@pam expected (the same account is used over SSH)") } if c.Timeout != "" { if _, err := time.ParseDuration(c.Timeout); err != nil { return fmt.Errorf("timeout: %w", err) } } if c.SSHPort < 0 || c.SSHPort > 65535 { return errors.New("ssh port: 1-65535 expected") } cfg := &config.Config{Kinds: model.Kinds, Reference: config.Cluster{Name: "\x00ref"}, Clients: []config.Client{{Cluster: config.Cluster{Name: c.Name}, Mapping: c.Mapping, Policies: c.Policies}}} return cfg.Validate() } func errText(err error, ok string) string { if err != nil { return err.Error() } return ok } func (s *Server) handleDeleteCluster(w http.ResponseWriter, r *http.Request) { name := r.PathValue("name") err := s.store.DeleteCluster(r.Context(), name) s.audit(r, Identity{}, "cluster-delete", name, err == nil, errText(err, "")) if err != nil { writeError(w, http.StatusNotFound, err.Error(), "notfound") return } s.invalidate() writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) } // handleRole makes a cluster the primary, a target or disabled. func (s *Server) handleRole(w http.ResponseWriter, r *http.Request) { var req struct { Role string `json:"role"` } if err := readJSON(r, &req); err != nil { writeError(w, http.StatusBadRequest, err.Error(), "request") return } name := r.PathValue("name") err := s.store.SetRole(r.Context(), name, req.Role) s.audit(r, Identity{}, "cluster-role", name, err == nil, errText(err, req.Role)) switch { case errors.Is(err, store.ErrNotFound): writeError(w, http.StatusNotFound, err.Error(), "notfound") return case err != nil: writeError(w, http.StatusBadRequest, err.Error(), "invalid") return } s.invalidate() writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) } // handleCheckCluster tests the API (root@pam, PVE 9.x) and SSH. On the first // connection the TLS certificate (mode "fp") and the SSH host key are recorded // (trust on first use); trust=true accepts a new SSH host key. func (s *Server) handleCheckCluster(w http.ResponseWriter, r *http.Request) { var req struct { Trust bool `json:"trust"` } if r.ContentLength > 0 { if err := readJSON(r, &req); err != nil { writeError(w, http.StatusBadRequest, err.Error(), "request") return } } name := r.PathValue("name") c, cc, err := s.store.Cluster(r.Context(), name) if errors.Is(err, store.ErrNotFound) { writeError(w, http.StatusNotFound, "unknown cluster", "notfound") return } if err != nil { writeError(w, http.StatusInternalServerError, err.Error(), "store") return } ctx, cancel := context.WithTimeout(r.Context(), 45*time.Second) defer cancel() res := service.CheckCluster(ctx, cc, c.TLS == store.TLSPin, req.Trust) chk := &store.Check{At: time.Now().UTC(), OK: res.OK, Error: res.Error, Warnings: res.Warnings, Version: res.Version, API: res.API, SSH: res.SSH} for _, n := range res.Nodes { chk.Nodes = append(chk.Nodes, n.Name) } // a TLS certificate is pinned only once the API answered with it fp := "" if res.API { fp = res.TLSFingerprint } if err := s.store.Pin(r.Context(), name, fp, res.SSHHostKey, chk); err != nil { writeError(w, http.StatusInternalServerError, err.Error(), "store") return } detail := res.Error if fp != "" { detail += " TLS certificate recorded " + fp } if res.SSHHostKey != "" { detail += " SSH key recorded " + res.SSHFingerprint } s.audit(r, Identity{}, "cluster-check", name, res.OK, strings.TrimSpace(detail)) if fp != "" || res.SSHHostKey != "" { s.invalidate() } writeJSON(w, http.StatusOK, res) } // handleForget clears the pinned certificate and SSH host key of a cluster // (node reinstalled, new certificate): they are recorded again on the next test. func (s *Server) handleForget(w http.ResponseWriter, r *http.Request) { name := r.PathValue("name") err := s.store.Forget(r.Context(), name) s.audit(r, Identity{}, "cluster-forget", name, err == nil, errText(err, "TLS fingerprint and SSH key forgotten")) if err != nil { writeError(w, http.StatusNotFound, err.Error(), "notfound") return } s.invalidate() writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) } func (s *Server) handleSettings(w http.ResponseWriter, r *http.Request) { st, err := s.store.Settings(r.Context()) if err != nil { writeError(w, http.StatusInternalServerError, err.Error(), "store") return } writeJSON(w, http.StatusOK, map[string]any{"settings": st, "allKinds": model.Kinds, "deps": adapter.KindDeps, "bundles": adapter.KindBundles}) } func (s *Server) handleSaveSettings(w http.ResponseWriter, r *http.Request) { var st store.Settings if err := readJSON(r, &st); err != nil { writeError(w, http.StatusBadRequest, err.Error(), "request") return } st.Kinds = adapter.Bundle(st.Kinds) cfg := &config.Config{Kinds: st.Kinds, Objects: st.Objects, Policies: st.Policies, Prune: st.Prune, Reference: config.Cluster{Name: "\x00ref"}} err := cfg.Validate() if err == nil && len(st.Kinds) == 0 { err = errors.New("select at least one kind") } if err == nil { err = s.store.SaveSettings(r.Context(), st) } b, _ := json.Marshal(st) s.audit(r, Identity{}, "settings-save", "replication", err == nil, errText(err, string(b))) if err != nil { writeError(w, http.StatusBadRequest, err.Error(), "invalid") return } s.invalidate() writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) } // ---- users and audit ---- func (s *Server) handleUsers(w http.ResponseWriter, r *http.Request) { us, err := s.store.Users(r.Context()) if err != nil { writeError(w, http.StatusInternalServerError, err.Error(), "store") return } if us == nil { us = []store.User{} } writeJSON(w, http.StatusOK, map[string]any{"users": us}) } func (s *Server) handleSaveUser(w http.ResponseWriter, r *http.Request) { var req struct { Role string `json:"role"` Password string `json:"password"` } if err := readJSON(r, &req); err != nil { writeError(w, http.StatusBadRequest, err.Error(), "request") return } name := r.PathValue("name") err := s.store.SaveUser(r.Context(), name, req.Role, req.Password) detail := "role " + req.Role if req.Password != "" { detail += ", password set" } s.audit(r, Identity{}, "user-save", name, err == nil, errText(err, detail)) if err != nil { writeError(w, http.StatusBadRequest, err.Error(), "invalid") return } writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) } func (s *Server) handleDeleteUser(w http.ResponseWriter, r *http.Request) { name := r.PathValue("name") err := s.store.DeleteUser(r.Context(), name) s.audit(r, Identity{}, "user-delete", name, err == nil, errText(err, "")) if err != nil { writeError(w, http.StatusBadRequest, err.Error(), "invalid") return } writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) } // handlePassword lets a local user change their own password. func (s *Server) handlePassword(w http.ResponseWriter, r *http.Request) { id := identity(r) var req struct { Old string `json:"old"` New string `json:"new"` } if err := readJSON(r, &req); err != nil { writeError(w, http.StatusBadRequest, err.Error(), "request") return } if id.Mode != "local" { writeError(w, http.StatusBadRequest, "password managed by the identity provider", "oidc") return } u, ok := s.store.Authenticate(r.Context(), id.User, req.Old) if !ok { time.Sleep(loginFailDelay) s.audit(r, id, "password-change", id.User, false, "ancien mot de passe incorrect") writeError(w, http.StatusForbidden, "wrong current password", "password") return } err := s.store.SaveUser(r.Context(), id.User, u.Role, req.New) s.audit(r, id, "password-change", id.User, err == nil, errText(err, "")) if err != nil { writeError(w, http.StatusBadRequest, err.Error(), "invalid") return } // the new epoch ends the other sessions; this one gets a new cookie if nu, err := s.store.User(r.Context(), id.User); err == nil { s.auth.setSession(w, r, session{User: nu.Name, Role: nu.Role, Mode: "local", Epoch: nu.Epoch}) } writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) } func (s *Server) handleAudit(w http.ResponseWriter, r *http.Request) { limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) before, _ := strconv.ParseInt(r.URL.Query().Get("before"), 10, 64) ev, err := s.store.Events(r.Context(), limit, before) if err != nil { writeError(w, http.StatusInternalServerError, err.Error(), "store") return } if ev == nil { ev = []store.Event{} } writeJSON(w, http.StatusOK, map[string]any{"events": ev}) }