pve-confsync/internal/nodesync/read.go

package nodesync

import (
	"bufio"
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"fmt"
	"math/bits"
	"net/url"
	"path"
	"sort"
	"strconv"
	"strings"
	"time"

	"pve-confsync/internal/model"
	"pve-confsync/internal/pve"
)

// Shell runs a command as root on the node, stdin fed with in (sshx.Client).
type Shell interface {
	Run(ctx context.Context, cmd string, in []byte) ([]byte, error)
}

// Endpoint is one node: the API of its cluster (calls proxied to the node)
// and SSH as root on the node itself.
type Endpoint struct {
	Key     string // "cluster/node"
	Cluster string
	Node    string
	API     *pve.Client
	Shell   Shell // nil: SSH host key not recorded yet
	// Dial opens a new SSH access, to check that the node still answers after
	// a network reload (nil: Shell is used).
	Dial func() (Shell, error)
}

// File is a file read on a node.
type File struct {
	Path    string `json:"path"`
	Mode    string `json:"mode"`  // octal, e.g. 644
	Owner   string `json:"owner"` // user:group
	Hash    string `json:"hash"`  // SHA-256 of the content
	Content []byte `json:"-"`
}

// Snapshot is what was read on one node.
type Snapshot struct {
	Key     string                     `json:"key"`
	Version string                     `json:"version,omitempty"`
	Network []model.Object             `json:"-"` // interfaces (id = iface), canonical fields
	Options map[string][]string        `json:"-"` // iface -> lines the API does not write (bond-miimon…)
	DNS     map[string]any             `json:"-"`
	Files   map[string]map[string]File `json:"-"` // item -> path -> file
	// Pending: /etc/network/interfaces.new exists (network changes not applied yet).
	Pending  bool      `json:"pending,omitempty"`
	Warnings []string  `json:"warnings,omitempty"`
	ReadAt   time.Time `json:"readAt"`
}

// maxFile: larger files are not replicated (configuration files are small).
const maxFile = 1 << 20

// Read loads the given items from a node.
func Read(ctx context.Context, ep Endpoint, items []string) (Snapshot, error) {
	snap := Snapshot{Key: ep.Key, Files: map[string]map[string]File{}, ReadAt: time.Now().UTC()}
	v, err := NodeVersion(ctx, ep)
	if err != nil {
		return snap, err
	}
	snap.Version = v
	var files []Item
	for _, id := range items {
		it, err := ItemByID(id)
		if err != nil {
			return snap, err
		}
		switch it.Kind {
		case KindNetwork:
			if snap.Network, snap.Options, err = readNetwork(ctx, ep); err != nil {
				return snap, fmt.Errorf("%s: read network: %w", ep.Key, err)
			}
		case KindDNS:
			if snap.DNS, err = readDNS(ctx, ep); err != nil {
				return snap, fmt.Errorf("%s: read dns: %w", ep.Key, err)
			}
		case KindFile:
			files = append(files, it)
		}
	}
	// SSH is needed for the files, and for the pending network changes
	if ep.Shell == nil {
		return snap, fmt.Errorf("%s: SSH host key not recorded yet, test the node first", ep.Key)
	}
	out, err := ep.Shell.Run(ctx, readScript(files), nil)
	if err != nil {
		return snap, fmt.Errorf("%s: read files: %w", ep.Key, err)
	}
	got, pending, warns, err := parseRead(out)
	if err != nil {
		return snap, fmt.Errorf("%s: read files: %w", ep.Key, err)
	}
	snap.Pending = pending
	snap.Warnings = append(snap.Warnings, warns...)
	for _, it := range files {
		m := map[string]File{}
		for p, f := range got {
			if matchItem(it, p) {
				m[p] = f
			}
		}
		snap.Files[it.ID] = m
	}
	return snap, nil
}

// NodeVersion reads the pve-manager version of the node and refuses anything
// but PVE 9.x.
func NodeVersion(ctx context.Context, ep Endpoint) (string, error) {
	var v struct {
		Version string `json:"version"`
		Release string `json:"release"`
	}
	if err := ep.API.Get(ctx, "nodes/"+url.PathEscape(ep.Node)+"/version", nil, &v); err != nil {
		return "", fmt.Errorf("%s: version check: %w", ep.Key, err)
	}
	ref := v.Release
	if ref == "" {
		ref = v.Version
	}
	if major, err := strconv.Atoi(strings.SplitN(ref, ".", 2)[0]); err != nil || major != pve.SupportedMajor {
		return v.Version, fmt.Errorf("%s: node runs PVE %q, only %d.x is supported", ep.Key, ref, pve.SupportedMajor)
	}
	return v.Version, nil
}

func matchItem(it Item, p string) bool {
	for _, x := range it.Exclude {
		if x == p {
			return false
		}
	}
	for _, pat := range it.Paths {
		if ok, _ := path.Match(pat, p); ok {
			return true
		}
	}
	return false
}

// readScript lists the files of the items: one header line per file
// ("F <mode> <user:group> <path>", "L <size> <path>" when too large) then
// its content in base64 on one line; "PENDING" when interfaces.new exists.
// The first line is a marker the fake node of the tests understands.
func readScript(files []Item) string {
	var pats []string
	for _, it := range files {
		pats = append(pats, it.Paths...)
	}
	var b strings.Builder
	b.WriteString("# confsync:read\n")
	if len(pats) > 0 {
		fmt.Fprintf(&b, "for f in %s; do\n", strings.Join(pats, " "))
		b.WriteString(`  [ -f "$f" ] && [ ! -L "$f" ] || continue
  s=$(stat -c %s "$f")
  if [ "$s" -gt ` + strconv.Itoa(maxFile) + ` ]; then printf 'L %s %s\n' "$s" "$f"; continue; fi
  printf 'F %s %s\n' "$(stat -c '%a %U:%G' "$f")" "$f"
  base64 -w0 < "$f"; echo
done
`)
	}
	b.WriteString("[ -e /etc/network/interfaces.new ] && echo PENDING\ntrue\n")
	return b.String()
}

func parseRead(out []byte) (map[string]File, bool, []string, error) {
	files := map[string]File{}
	pending := false
	var warns []string
	sc := bufio.NewScanner(bytes.NewReader(out))
	sc.Buffer(make([]byte, 64<<10), 4*maxFile)
	for sc.Scan() {
		line := sc.Text()
		switch {
		case line == "PENDING":
			pending = true
		case strings.HasPrefix(line, "L "):
			p := strings.SplitN(line, " ", 3)
			if len(p) == 3 {
				warns = append(warns, fmt.Sprintf("%s: %s bytes, too large, not replicated", p[2], p[1]))
			}
		case strings.HasPrefix(line, "F "):
			p := strings.SplitN(line, " ", 4)
			if len(p) != 4 || !sc.Scan() {
				return nil, false, nil, fmt.Errorf("unexpected output %q", line)
			}
			data, err := base64.StdEncoding.DecodeString(strings.TrimSpace(sc.Text()))
			if err != nil {
				return nil, false, nil, fmt.Errorf("%s: %w", p[3], err)
			}
			if !safePath(p[3]) {
				warns = append(warns, fmt.Sprintf("%s: unexpected characters in the path, not replicated", p[3]))
				continue
			}
			files[p[3]] = File{Path: p[3], Mode: p[1], Owner: p[2], Hash: hash(data), Content: data}
		}
	}
	return files, pending, warns, sc.Err()
}

// safePath: the paths go into shell scripts between single quotes.
func safePath(p string) bool {
	if !strings.HasPrefix(p, "/") || strings.Contains(p, "..") {
		return false
	}
	for _, r := range p {
		if !(r == '/' || r == '.' || r == '-' || r == '_' || r == '+' || r == '@' ||
			(r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9')) {
			return false
		}
	}
	return true
}

func hash(b []byte) string {
	s := sha256.Sum256(b)
	return hex.EncodeToString(s[:])
}

// ---- network (GET /nodes/{node}/network) ----

// netWritable are the parameters of POST/PUT /nodes/{node}/network[/{iface}]
// (pve-manager PVE/API2/Network.pm, $confdesc), in their cidr form: the GET
// returns address/netmask and cidr, the canonical form keeps cidr only.
var netWritable = map[string]bool{
	"type": true, "comments": true, "comments6": true, "autostart": true,
	"bridge_vlan_aware": true, "bridge_vids": true, "bridge_ports": true,
	"ovs_ports": true, "ovs_tag": true, "ovs_options": true, "ovs_bridge": true, "ovs_bonds": true,
	"slaves": true, "bond_mode": true, "bond-primary": true, "bond_xmit_hash_policy": true,
	"vlan-raw-device": true, "vlan-id": true,
	"gateway": true, "cidr": true, "gateway6": true, "cidr6": true, "mtu": true,
}

// netTypes are the interface types handled (fabric, vnet and unknown are not).
var netTypes = map[string]bool{
	"eth": true, "bond": true, "bridge": true, "vlan": true, "alias": true,
	"OVSBridge": true, "OVSBond": true, "OVSPort": true, "OVSIntPort": true,
}

// physical interfaces cannot be created: they must exist on the target.
func physical(typ string) bool { return typ == "eth" }

func readNetwork(ctx context.Context, ep Endpoint) ([]model.Object, map[string][]string, error) {
	var raw []map[string]any
	if err := ep.API.Get(ctx, "nodes/"+url.PathEscape(ep.Node)+"/network", nil, &raw); err != nil {
		return nil, nil, err
	}
	var objs []model.Object
	opts := map[string][]string{}
	for _, r := range raw {
		id := pve.Str(r["iface"])
		typ := pve.Str(r["type"])
		if id == "" || id == "lo" || !netTypes[typ] {
			continue
		}
		f := map[string]any{}
		for k, v := range r {
			if !netWritable[k] || v == nil {
				continue
			}
			s := strings.TrimRight(pve.Str(v), "\n")
			if s == "" {
				continue
			}
			f[k] = s
		}
		// a static address without cidr (older parsers): rebuild it
		if _, ok := f["cidr"]; !ok {
			if a, m := pve.Str(r["address"]), pve.Str(r["netmask"]); a != "" && m != "" {
				f["cidr"] = a + "/" + maskBits(m)
			}
		}
		if l := listOf(r["options"]); len(l) > 0 {
			opts[id] = l
		}
		if l := listOf(r["options6"]); len(l) > 0 {
			opts[id] = append(opts[id], l...)
		}
		objs = append(objs, model.Object{Kind: "network", ID: id, Fields: f})
	}
	sort.Slice(objs, func(i, j int) bool { return objs[i].ID < objs[j].ID })
	return objs, opts, nil
}

func listOf(v any) []string {
	l, _ := v.([]any)
	out := make([]string, 0, len(l))
	for _, e := range l {
		out = append(out, pve.Str(e))
	}
	sort.Strings(out)
	return out
}

// maskBits converts 255.255.255.0 to 24 (a prefix length is kept).
func maskBits(m string) string {
	if !strings.Contains(m, ".") {
		return m
	}
	n := 0
	for _, p := range strings.Split(m, ".") {
		b, _ := strconv.Atoi(p)
		n += bits.OnesCount8(uint8(b))
	}
	return strconv.Itoa(n)
}

// ---- dns (GET /nodes/{node}/dns) ----

var dnsFields = []string{"search", "dns1", "dns2", "dns3"}

func readDNS(ctx context.Context, ep Endpoint) (map[string]any, error) {
	var raw map[string]any
	if err := ep.API.Get(ctx, "nodes/"+url.PathEscape(ep.Node)+"/dns", nil, &raw); err != nil {
		return nil, err
	}
	out := map[string]any{}
	for _, k := range dnsFields {
		if s := pve.Str(raw[k]); s != "" {
			out[k] = s
		}
	}
	return out, nil
}