pve-confsync/internal/server/nodes.go

package server

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net/http"
	"strings"
	"sync"
	"time"

	"pve-confsync/internal/nodesync"
	"pve-confsync/internal/pve"
	"pve-confsync/internal/service"
	"pve-confsync/internal/sshx"
	"pve-confsync/internal/store"
)

// ---- node replication: nodes (SSH access), profiles, state, dry-run, apply ----

// handleNodes lists the nodes, the profiles and the catalog of items.
func (s *Server) handleNodes(w http.ResponseWriter, r *http.Request) {
	nodes, err := s.store.Nodes(r.Context())
	if err != nil {
		writeError(w, http.StatusInternalServerError, err.Error(), "store")
		return
	}
	profiles, err := s.store.Profiles(r.Context())
	if err != nil {
		writeError(w, http.StatusInternalServerError, err.Error(), "store")
		return
	}
	if identity(r).Role != store.RoleAdmin {
		for i, n := range nodes {
			nodes[i] = store.Node{Key: n.Key, Cluster: n.Cluster, Node: n.Node, ClusterRole: n.ClusterRole, LastCheck: n.LastCheck}
		}
	}
	writeJSON(w, http.StatusOK, map[string]any{"nodes": nodes, "profiles": profiles, "items": nodesync.Items,
		"defaults": nodesync.DefaultPolicies})
}

// handleDiscover reads the nodes of every cluster (GET /cluster/status) and
// records them with their address in the cluster.
func (s *Server) handleDiscover(w http.ResponseWriter, r *http.Request) {
	cls, err := s.store.Clusters(r.Context())
	if err != nil {
		writeError(w, http.StatusInternalServerError, err.Error(), "store")
		return
	}
	ctx, cancel := context.WithTimeout(r.Context(), readTimeout)
	defer cancel()
	errs := map[string]string{}
	found := 0
	for _, c := range cls {
		if c.Role == store.ClusterDisabled {
			continue
		}
		_, cc, err := s.store.Cluster(ctx, c.Name)
		var nodes []service.DiscoveredNode
		if err == nil {
			nodes, err = service.DiscoverNodes(ctx, cc)
		}
		if err == nil {
			var d []store.Discovered
			for _, n := range nodes {
				d = append(d, store.Discovered{Name: n.Name, IP: n.IP})
			}
			err = s.store.SetNodes(ctx, c.Name, d)
			found += len(d)
		}
		if err != nil {
			errs[c.Name] = err.Error()
		}
	}
	s.audit(r, Identity{}, "nodes-discover", "all clusters", len(errs) == 0, fmt.Sprintf("%d node(s)", found))
	s.invalidateNodes()
	writeJSON(w, http.StatusOK, map[string]any{"found": found, "errors": errs})
}

func (s *Server) handleSaveNode(w http.ResponseWriter, r *http.Request) {
	var req struct {
		SSHHost       string `json:"sshHost"`
		SSHPort       int    `json:"sshPort"`
		Password      string `json:"password"`
		ClearPassword bool   `json:"clearPassword"`
	}
	if err := readJSON(r, &req); err != nil {
		writeError(w, http.StatusBadRequest, err.Error(), "request")
		return
	}
	cl, node := r.PathValue("cluster"), r.PathValue("node")
	err := s.store.SaveNode(r.Context(), cl, node, req.SSHHost, req.SSHPort, req.Password, req.ClearPassword)
	detail := fmt.Sprintf("ssh=%s:%d", req.SSHHost, req.SSHPort)
	if req.Password != "" {
		detail += " password=changed"
	}
	if req.ClearPassword {
		detail += " password=cluster"
	}
	s.audit(r, Identity{}, "node-save", store.NodeKey(cl, node), err == nil, errText(err, detail))
	switch {
	case errors.Is(err, store.ErrNotFound):
		writeError(w, http.StatusNotFound, "unknown node", "notfound")
	case err != nil:
		writeError(w, http.StatusBadRequest, err.Error(), "invalid")
	default:
		s.invalidateNodes()
		writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
	}
}

// handleCheckNode tests SSH on a node; the first test records its host key.
func (s *Server) handleCheckNode(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
		}
	}
	cl, node := r.PathValue("cluster"), r.PathValue("node")
	a, err := s.store.NodeAccessOf(r.Context(), cl, node)
	if errors.Is(err, store.ErrNotFound) {
		writeError(w, http.StatusNotFound, "unknown node", "notfound")
		return
	}
	if err != nil {
		writeError(w, http.StatusInternalServerError, err.Error(), "store")
		return
	}
	ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
	defer cancel()
	res := service.CheckNode(ctx, node, a.SSH, req.Trust)
	chk := &store.Check{At: time.Now().UTC(), OK: res.OK, Error: res.Error, Version: res.Version, SSH: res.OK}
	if err := s.store.PinNode(r.Context(), cl, node, res.SSHHostKey, chk); err != nil {
		writeError(w, http.StatusInternalServerError, err.Error(), "store")
		return
	}
	detail := res.Error
	if res.SSHHostKey != "" {
		detail += " SSH key recorded " + res.SSHFingerprint
	}
	s.audit(r, Identity{}, "node-check", store.NodeKey(cl, node), res.OK, strings.TrimSpace(detail))
	s.invalidateNodes()
	writeJSON(w, http.StatusOK, res)
}

func (s *Server) handleForgetNode(w http.ResponseWriter, r *http.Request) {
	cl, node := r.PathValue("cluster"), r.PathValue("node")
	err := s.store.ForgetNode(r.Context(), cl, node)
	s.audit(r, Identity{}, "node-forget", store.NodeKey(cl, node), err == nil, errText(err, "SSH key forgotten"))
	if err != nil {
		writeError(w, http.StatusNotFound, err.Error(), "notfound")
		return
	}
	s.invalidateNodes()
	writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
}

func (s *Server) handleSaveProfile(w http.ResponseWriter, r *http.Request) {
	var p nodesync.Profile
	if err := readJSON(r, &p); err != nil {
		writeError(w, http.StatusBadRequest, err.Error(), "request")
		return
	}
	old := r.PathValue("name")
	if old == "-" {
		old = ""
	}
	err := s.store.SaveProfile(r.Context(), old, p)
	b, _ := json.Marshal(p)
	s.audit(r, Identity{}, "profile-save", p.Name, err == nil, errText(err, string(b)))
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error(), "invalid")
		return
	}
	s.invalidateNodes()
	writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
}

func (s *Server) handleDeleteProfile(w http.ResponseWriter, r *http.Request) {
	name := r.PathValue("name")
	err := s.store.DeleteProfile(r.Context(), name)
	s.audit(r, Identity{}, "profile-delete", name, err == nil, errText(err, ""))
	if err != nil {
		writeError(w, http.StatusNotFound, err.Error(), "notfound")
		return
	}
	s.invalidateNodes()
	writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
}

// endpoints builds the access to every node: one API client per cluster,
// SSH as root on each node whose host key is recorded.
func (s *Server) endpoints(ctx context.Context) (map[string]nodesync.Endpoint, error) {
	acc, err := s.store.NodeAccesses(ctx)
	if err != nil {
		return nil, err
	}
	apis := map[string]*pve.Client{}
	out := map[string]nodesync.Endpoint{}
	for key, a := range acc {
		c, ok := apis[a.Cluster.Name]
		if !ok {
			if c, err = a.Cluster.PVE(); err != nil {
				return nil, err
			}
			apis[a.Cluster.Name] = c
		}
		ep := nodesync.Endpoint{Key: key, Cluster: a.Node.Cluster, Node: a.Node.Node, API: c}
		if a.SSH.HostKey != "" {
			if sh, err := sshx.New(a.SSH); err == nil {
				ep.Shell = sh
			}
		}
		out[key] = ep
	}
	return out, nil
}

func (s *Server) invalidateNodes() {
	s.nodeMu.Lock()
	s.nodeStates = nil
	s.nodeMu.Unlock()
}

// handleNodeState returns the last comparison of a profile; refresh=1 reads
// the nodes again.
func (s *Server) handleNodeState(w http.ResponseWriter, r *http.Request) {
	name := r.URL.Query().Get("profile")
	s.nodeMu.Lock()
	st := s.nodeStates[name]
	s.nodeMu.Unlock()
	if st == nil || r.URL.Query().Get("refresh") == "1" {
		prof, err := s.store.Profile(r.Context(), name)
		if errors.Is(err, store.ErrNotFound) {
			writeError(w, http.StatusNotFound, "unknown profile", "notfound")
			return
		}
		if err != nil {
			writeError(w, http.StatusInternalServerError, err.Error(), "store")
			return
		}
		eps, err := s.endpoints(r.Context())
		if err != nil {
			writeError(w, http.StatusInternalServerError, err.Error(), "config")
			return
		}
		ctx, cancel := context.WithTimeout(r.Context(), readTimeout)
		defer cancel()
		st = nodesync.Analyze(ctx, &prof, eps, nil)
		s.rememberNodes(st, true)
	}
	writeJSON(w, http.StatusOK, map[string]any{"state": st})
}

// rememberNodes keeps a comparison; a partial one (some targets) only
// replaces the targets it read.
func (s *Server) rememberNodes(st *nodesync.State, all bool) {
	s.nodeMu.Lock()
	defer s.nodeMu.Unlock()
	if s.nodeStates == nil {
		s.nodeStates = map[string]*nodesync.State{}
	}
	old := s.nodeStates[st.Profile]
	if all || old == nil {
		if all {
			s.nodeStates[st.Profile] = st
		}
		return
	}
	cp := *old
	cp.Master = st.Master
	cp.Targets = append([]nodesync.Target(nil), old.Targets...)
	for _, t := range st.Targets {
		for i := range cp.Targets {
			if cp.Targets[i].Key == t.Key {
				cp.Targets[i] = t
			}
		}
	}
	s.nodeStates[st.Profile] = &cp
}

type nodeSelection struct {
	Profile string            `json:"profile"`
	Scope   string            `json:"scope"`   // global (several targets), client (one target) or object
	Targets []string          `json:"targets"` // global: the targets (empty: all of them)
	Client  string            `json:"client"`  // client and object scopes
	Objects []string          `json:"objects"` // object scope: "item/id"
	Digests map[string]string `json:"digests"`
	Confirm string            `json:"confirm"` // network changes: target (profile name for several)
}

func (sel *nodeSelection) check() error {
	if sel.Profile == "" {
		return errors.New("profile required")
	}
	s := selection{Scope: sel.Scope, Targets: sel.Targets, Client: sel.Client, Objects: sel.Objects}
	if err := s.check(); err != nil {
		return err
	}
	sel.Targets, sel.Client, sel.Objects = s.Targets, s.Client, s.Objects
	return nil
}

func (sel nodeSelection) clients() []string {
	if sel.Client == "" {
		return sel.Targets
	}
	return []string{sel.Client}
}

func (sel nodeSelection) target() string {
	t := sel.Profile + ": "
	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"
	}
	return t
}

func (s *Server) nodePlan(ctx context.Context, sel nodeSelection) (*nodesync.State, []nodesync.TargetPlan, error) {
	prof, err := s.store.Profile(ctx, sel.Profile)
	if err != nil {
		if errors.Is(err, store.ErrNotFound) {
			err = fmt.Errorf("unknown profile %q", sel.Profile)
		}
		return nil, nil, err
	}
	for _, c := range sel.clients() {
		found := false
		for _, t := range prof.Targets {
			found = found || t == c
		}
		if !found {
			return nil, nil, fmt.Errorf("unknown target %q", c)
		}
	}
	eps, err := s.endpoints(ctx)
	if err != nil {
		return nil, nil, err
	}
	rctx, cancel := context.WithTimeout(ctx, readTimeout)
	defer cancel()
	st := nodesync.Analyze(rctx, &prof, eps, sel.clients())
	s.rememberNodes(st, sel.Scope == "global" && len(sel.Targets) == 0)
	return st, st.Plans(sel.Objects), nil
}

// node runs are journaled with the scopes "nodes", "node" and "node-object".
func nodeScope(scope string) string {
	switch scope {
	case "global":
		return "nodes"
	case "client":
		return "node"
	}
	return "node-object"
}

type nodeRunDetail struct {
	Profile string                 `json:"profile"`
	Plans   []nodesync.TargetPlan  `json:"plans"`
	Results map[string]*nodeResult `json:"results,omitempty"`
}

type nodeResult struct {
	Status string          `json:"status"`
	Error  string          `json:"error,omitempty"`
	Steps  []nodesync.Step `json:"steps"`
}

func nodeSummary(plans []nodesync.TargetPlan) (string, string) {
	changes, targets, 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
			targets++
		}
	}
	switch {
	case blocked > 0:
		return store.RunBlocked, fmt.Sprintf("%d change(s) on %d target(s), %d target(s) in error", changes, targets, blocked)
	case changes == 0:
		return store.RunNoop, "no difference"
	}
	return store.RunOK, fmt.Sprintf("%d change(s) on %d target(s)", changes, targets)
}

func nodePlanTargets(plans []nodesync.TargetPlan) []string {
	out := make([]string, 0, len(plans))
	for _, p := range plans {
		out = append(out, p.Client)
	}
	return out
}

func (s *Server) handleNodeDryRun(w http.ResponseWriter, r *http.Request) {
	var sel nodeSelection
	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.nodePlan(r.Context(), sel)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error(), "config")
		return
	}
	status, summary := nodeSummary(plans)
	d, _ := json.Marshal(nodeRunDetail{Profile: sel.Profile, Plans: plans})
	runID, err := s.store.StartRun(r.Context(), store.Run{User: id.User, Mode: "dry-run", Scope: nodeScope(sel.Scope),
		Clients: nodePlanTargets(plans), Objects: sel.Objects, Status: status, Summary: summary, Detail: d})
	if err != nil {
		writeError(w, http.StatusInternalServerError, err.Error(), "store")
		return
	}
	_ = s.store.UpdateRun(r.Context(), runID, status, summary, d)
	s.audit(r, id, "node-dry-run", sel.target(), true, summary)
	writeJSON(w, http.StatusOK, map[string]any{"run": runID, "plans": plans, "summary": summary})
}

// handleNodeApply recomputes the plans, checks the digests of the dry-run and
// the typed confirmation of the network changes, then applies them in the
// background, one target after the other.
func (s *Server) handleNodeApply(w http.ResponseWriter, r *http.Request) {
	var sel nodeSelection
	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()
		}
	}()
	st, plans, err := s.nodePlan(r.Context(), sel)
	if err != nil {
		writeError(w, http.StatusBadRequest, err.Error(), "config")
		return
	}
	var todo []nodesync.TargetPlan
	network := false
	for _, p := range plans {
		switch {
		case p.Error != "":
			s.audit(r, id, "node-apply", sel.target(), 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, "node-apply", sel.target(), false, p.Client+": plan en erreur")
			writeError(w, http.StatusConflict, p.Client+": "+p.Plan.Errors[0].Message, "plan")
			return
		case sel.Digests[p.Client] != p.Digest:
			s.audit(r, id, "node-apply", sel.target(), 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
		}
		network = network || p.Network
		todo = append(todo, p)
	}
	if len(todo) == 0 {
		writeJSON(w, http.StatusOK, map[string]any{"run": 0, "summary": "no difference"})
		return
	}
	if network {
		want := sel.Client
		if sel.Scope == "global" {
			want = sel.Profile
		}
		if sel.Confirm != want {
			writeError(w, http.StatusBadRequest, "network changes: confirmation required ("+want+")", "confirm")
			return
		}
	}
	det := &nodeRunDetail{Profile: sel.Profile, Plans: todo, Results: map[string]*nodeResult{}}
	for _, p := range todo {
		det.Results[p.Client] = &nodeResult{Status: "pending", Steps: []nodesync.Step{}}
	}
	d, _ := json.Marshal(det)
	runID, err := s.store.StartRun(r.Context(), store.Run{User: id.User, Mode: "apply", Scope: nodeScope(sel.Scope),
		Clients: nodePlanTargets(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, "node-apply-start", sel.target(), true, fmt.Sprintf("journal #%d", runID))
	started = true
	s.jobs.Add(1)
	rr := r.Clone(context.WithoutCancel(r.Context()))
	go func() {
		defer s.jobs.Done()
		defer s.applyMu.Unlock()
		s.runNodeApply(rr, id, runID, sel, st, det)
	}()
	writeJSON(w, http.StatusAccepted, map[string]any{"run": runID})
}

// runNodeApply applies the plans one target after the other and stops at the
// first failure: the nodes are often of the same cluster, a configuration
// that breaks one would break the next.
func (s *Server) runNodeApply(r *http.Request, id Identity, runID int64, sel nodeSelection, st *nodesync.State, det *nodeRunDetail) {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
	defer cancel()
	var mu sync.Mutex
	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]
		if failed > 0 {
			res.Status = "skipped"
			continue
		}
		res.Status = store.RunRunning
		save(store.RunRunning, p.Client+" en cours")
		err := st.Apply(ctx, p, sel.Objects, func(step nodesync.Step) {
			mu.Lock()
			res.Steps = append(res.Steps, step)
			mu.Unlock()
			save(store.RunRunning, p.Client+" : "+step.Action)
		})
		if err != nil {
			failed++
			res.Status, res.Error = store.RunError, err.Error()
			log.Printf("node apply %s (journal #%d): %v", p.Client, runID, err)
			s.audit(r, id, "node-apply", p.Client, false, err.Error())
			continue
		}
		done++
		res.Status = store.RunOK
		s.audit(r, id, "node-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.invalidateNodes()
}