pve-confsync/internal/pve/client.go

// Package pve is a minimal Proxmox VE API client: ticket auth with the root@pam
// credentials, generic JSON, form-encoded writes, task polling and version check.
// No typed SDK on purpose: objects stay generic (map[string]any) and are
// interpreted by the adapters. What the API does not cover (the password hashes
// of the pve realm) goes through a Shell, an SSH session on a node.
package pve

import (
	"context"
	"crypto/sha256"
	"crypto/tls"
	"crypto/x509"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
)

// Options describes how to reach and authenticate against one cluster.
type Options struct {
	Name        string // cluster name in the config (for messages only)
	BaseURL     string // https://host:8006
	User        string // user@realm, root@pam by default
	Password    string // never logged
	CAPEM       []byte // PEM bundle to trust (optional)
	Fingerprint string // SHA-256 of the leaf certificate, "AA:BB:..." (optional, pinning)
	Insecure    bool   // skip TLS verification (lab only)
	Timeout     time.Duration
	Shell       Shell // SSH access to a node (optional)
}

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

// ticketTTL: PVE tickets live 2 hours (pve-access-control AccessControl.pm,
// $ticket_lifetime); a new one is requested well before.
const ticketTTL = 90 * time.Minute

// Client talks to the PVE API of one cluster.
type Client struct {
	name     string
	base     *url.URL
	user     string
	password string
	http     *http.Client
	shell    Shell

	mu       sync.Mutex
	ticket   string
	csrf     string
	ticketAt time.Time
}

// ErrNoShell: the operation needs an SSH access to a node.
var ErrNoShell = errors.New("SSH access required")

// APIError is a non-2xx answer of the API.
type APIError struct {
	Method, Path string
	Status       int
	Message      string
	Errors       map[string]string
}

func (e *APIError) Error() string {
	msg := fmt.Sprintf("%s %s: HTTP %d", e.Method, e.Path, e.Status)
	if e.Message != "" {
		msg += ": " + e.Message
	}
	if len(e.Errors) > 0 {
		keys := make([]string, 0, len(e.Errors))
		for k := range e.Errors {
			keys = append(keys, k)
		}
		sort.Strings(keys)
		for _, k := range keys {
			msg += fmt.Sprintf(" [%s: %s]", k, strings.TrimSpace(e.Errors[k]))
		}
	}
	return msg
}

// New builds a client. The token secret is kept in the Authorization header only.
func New(o Options) (*Client, error) {
	u, err := url.Parse(strings.TrimRight(o.BaseURL, "/"))
	if err != nil || u.Scheme == "" || u.Host == "" {
		return nil, fmt.Errorf("cluster %s: invalid url %q", o.Name, o.BaseURL)
	}
	if o.User == "" {
		o.User = "root@pam"
	}
	if o.Password == "" {
		return nil, fmt.Errorf("cluster %s: the password of %s is required", o.Name, o.User)
	}
	tlsCfg, err := TLSConfig(o.Name, o.CAPEM, o.Fingerprint, o.Insecure)
	if err != nil {
		return nil, err
	}
	timeout := o.Timeout
	if timeout == 0 {
		timeout = 30 * time.Second
	}
	return &Client{
		name:     o.Name,
		base:     u,
		user:     o.User,
		password: o.Password,
		shell:    o.Shell,
		http: &http.Client{
			Timeout:   timeout,
			Transport: &http.Transport{TLSClientConfig: tlsCfg, Proxy: http.ProxyFromEnvironment},
		},
	}, nil
}

// TLSConfig builds the TLS settings of a cluster: certificate pinning (self-signed
// PVE certificates), a CA bundle, the system authorities, or no verification.
func TLSConfig(name string, caPEM []byte, fingerprint string, insecure bool) (*tls.Config, error) {
	tlsCfg := &tls.Config{MinVersion: tls.VersionTLS12}
	switch {
	case fingerprint != "":
		want := normalizeFingerprint(fingerprint)
		// Self-signed PVE certificates: pin the leaf certificate instead of a CA.
		tlsCfg.InsecureSkipVerify = true
		tlsCfg.VerifyPeerCertificate = func(raw [][]byte, _ [][]*x509.Certificate) error {
			if len(raw) == 0 {
				return errors.New("no peer certificate")
			}
			sum := sha256.Sum256(raw[0])
			if got := hex.EncodeToString(sum[:]); got != want {
				return fmt.Errorf("certificate fingerprint mismatch (got %s)", FormatFingerprint(got))
			}
			return nil
		}
	case len(caPEM) > 0:
		pool := x509.NewCertPool()
		if !pool.AppendCertsFromPEM(caPEM) {
			return nil, fmt.Errorf("cluster %s: no certificate in the CA bundle", name)
		}
		tlsCfg.RootCAs = pool
	case insecure:
		tlsCfg.InsecureSkipVerify = true
	}
	return tlsCfg, nil
}

// PeerFingerprint connects to the API and returns the SHA-256 fingerprint of the
// certificate it presents, without verifying it: the administrator pins it on
// the first connection (trust on first use) and sees it in the cluster list.
func PeerFingerprint(ctx context.Context, baseURL string) (string, error) {
	u, err := url.Parse(baseURL)
	if err != nil || u.Host == "" {
		return "", fmt.Errorf("invalid url %q", baseURL)
	}
	host := u.Host
	if u.Port() == "" {
		host += ":8006"
	}
	d := tls.Dialer{Config: &tls.Config{MinVersion: tls.VersionTLS12, InsecureSkipVerify: true}}
	conn, err := d.DialContext(ctx, "tcp", host)
	if err != nil {
		return "", err
	}
	defer conn.Close()
	certs := conn.(*tls.Conn).ConnectionState().PeerCertificates
	if len(certs) == 0 {
		return "", errors.New("no peer certificate")
	}
	sum := sha256.Sum256(certs[0].Raw)
	return FormatFingerprint(hex.EncodeToString(sum[:])), nil
}

// FormatFingerprint prints a SHA-256 fingerprint like PVE does ("AA:BB:...").
func FormatFingerprint(fp string) string {
	fp = strings.ToUpper(normalizeFingerprint(fp))
	var b strings.Builder
	for i := 0; i < len(fp); i += 2 {
		if i > 0 {
			b.WriteByte(':')
		}
		b.WriteString(fp[i:min(i+2, len(fp))])
	}
	return b.String()
}

// Name returns the cluster name given in the options.
func (c *Client) Name() string { return c.name }

// Shell returns the SSH access to a node, or ErrNoShell.
func (c *Client) Shell() (Shell, error) {
	if c.shell == nil {
		return nil, fmt.Errorf("cluster %s: %w", c.name, ErrNoShell)
	}
	return c.shell, nil
}

func normalizeFingerprint(s string) string {
	return strings.ToLower(strings.NewReplacer(":", "", " ", "").Replace(s))
}

// login gets a ticket and a CSRF token (POST /access/ticket). Accounts with a
// second factor are refused: the tool cannot answer the TFA challenge.
func (c *Client) login(ctx context.Context) error {
	var res struct {
		Ticket  string `json:"ticket"`
		CSRF    string `json:"CSRFPreventionToken"`
		NeedTFA any    `json:"NeedTFA"`
	}
	f := url.Values{"username": {c.user}, "password": {c.password}}
	if err := c.send(ctx, http.MethodPost, "access/ticket", nil, f, &res, false); err != nil {
		var ae *APIError
		if errors.As(err, &ae) && ae.Status == http.StatusUnauthorized {
			return fmt.Errorf("cluster %s: authentication of %s failed", c.name, c.user)
		}
		return err
	}
	if Str(res.NeedTFA) == "1" {
		return fmt.Errorf("cluster %s: %s has a second factor (TFA), not supported", c.name, c.user)
	}
	if res.Ticket == "" {
		return fmt.Errorf("cluster %s: no ticket in the login answer", c.name)
	}
	c.ticket, c.csrf, c.ticketAt = res.Ticket, res.CSRF, time.Now()
	return nil
}

// Login checks the credentials (and renews the ticket when needed).
func (c *Client) Login(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.ticket != "" && time.Since(c.ticketAt) < ticketTTL {
		return nil
	}
	return c.login(ctx)
}

// Get reads path (relative to /api2/json) and decodes "data" into out.
func (c *Client) Get(ctx context.Context, path string, query url.Values, out any) error {
	return c.do(ctx, http.MethodGet, path, query, nil, out)
}

// Post, Put and Delete send form-encoded parameters (the encoding pvesh and the GUI use).
func (c *Client) Post(ctx context.Context, path string, params url.Values, out any) error {
	return c.do(ctx, http.MethodPost, path, nil, params, out)
}

func (c *Client) Put(ctx context.Context, path string, params url.Values, out any) error {
	return c.do(ctx, http.MethodPut, path, nil, params, out)
}

// Delete passes its parameters in the query string, as the PVE API expects for DELETE.
func (c *Client) Delete(ctx context.Context, path string, params url.Values, out any) error {
	return c.do(ctx, http.MethodDelete, path, params, nil, out)
}

func (c *Client) do(ctx context.Context, method, path string, query, form url.Values, out any) error {
	if err := c.Login(ctx); err != nil {
		return err
	}
	err := c.send(ctx, method, path, query, form, out, true)
	var ae *APIError
	if errors.As(err, &ae) && ae.Status == http.StatusUnauthorized {
		// ticket expired early (key rotation on the cluster): log in again once
		c.mu.Lock()
		c.ticket = ""
		c.mu.Unlock()
		if err := c.Login(ctx); err != nil {
			return err
		}
		err = c.send(ctx, method, path, query, form, out, true)
	}
	return err
}

func (c *Client) send(ctx context.Context, method, path string, query, form url.Values, out any, auth bool) error {
	u := *c.base
	u.Path = "/api2/json/" + strings.TrimLeft(path, "/")
	if len(query) > 0 {
		u.RawQuery = query.Encode()
	}
	var body io.Reader
	if form != nil {
		body = strings.NewReader(form.Encode())
	}
	req, err := http.NewRequestWithContext(ctx, method, u.String(), body)
	if err != nil {
		return err
	}
	if auth {
		c.mu.Lock()
		req.AddCookie(&http.Cookie{Name: "PVEAuthCookie", Value: c.ticket})
		if method != http.MethodGet {
			req.Header.Set("CSRFPreventionToken", c.csrf)
		}
		c.mu.Unlock()
	}
	req.Header.Set("Accept", "application/json")
	if form != nil {
		req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
	}
	resp, err := c.http.Do(req)
	if err != nil {
		// url.Error contains the URL only, never the ticket nor the password.
		return fmt.Errorf("cluster %s: %w", c.name, err)
	}
	defer resp.Body.Close()
	raw, err := io.ReadAll(io.LimitReader(resp.Body, 32<<20))
	if err != nil {
		return fmt.Errorf("cluster %s: %s %s: %w", c.name, method, path, err)
	}
	var env struct {
		Data    json.RawMessage   `json:"data"`
		Message string            `json:"message"`
		Errors  map[string]string `json:"errors"`
	}
	_ = json.Unmarshal(raw, &env)
	if resp.StatusCode < 200 || resp.StatusCode > 299 {
		msg := env.Message
		if msg == "" {
			// pveproxy puts the reason in the status line ("596 Broken pipe", "403 Permission check failed")
			msg = strings.TrimSpace(strings.TrimPrefix(resp.Status, strconv.Itoa(resp.StatusCode)))
		}
		return &APIError{Method: method, Path: path, Status: resp.StatusCode, Message: msg, Errors: env.Errors}
	}
	if out == nil || len(env.Data) == 0 || string(env.Data) == "null" {
		return nil
	}
	d := json.NewDecoder(strings.NewReader(string(env.Data)))
	d.UseNumber()
	if err := d.Decode(out); err != nil {
		return fmt.Errorf("cluster %s: %s %s: decode: %w", c.name, method, path, err)
	}
	return nil
}