egress-proxies-operator/internal/provider/proxmox/client.go

package proxmox

import (
	"bytes"
	"context"
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"mime/multipart"
	"net"
	"net/http"
	"net/url"
	"os"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"

	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing"
)

const (
	dialTimeout         = 10 * time.Second
	tlsHandshakeTimeout = 10 * time.Second
	idleConnTimeout     = 90 * time.Second
	maxIdleConnsPerHost = 4
	// maxErrorBody caps how much of a failed response is read for its
	// message; Proxmox error bodies are a few hundred bytes.
	maxErrorBody = 64 << 10
)

// errCredentials marks a token file that is missing or unusable, so the
// Checker can report CredentialsUnavailable instead of an API failure.
var errCredentials = errors.New("proxmox token file unusable")

// tokenPattern is <user>@<realm>!<tokenid>=<secret> (spec entry 14).
var tokenPattern = regexp.MustCompile(`^[^@\s]+@[^!\s]+![^=\s]+=\S+$`)

// httpAPI is the production adapter behind the api seam: a thin net/http
// client for /api2/json (spec entry 17). It is safe for concurrent use.
type httpAPI struct {
	base      *url.URL
	tokenFile string
	caFile    string
	insecure  bool
	// timeout bounds one request, response included. Read when the client
	// is built; tests shrink it before the first call.
	timeout time.Duration
	wrap    func(http.RoundTripper) http.RoundTripper

	// mu guards client, built on first use because the CA is loaded lazily
	// and only cached once it loads (spec entry 14).
	mu     sync.Mutex
	client *http.Client
}

var _ api = (*httpAPI)(nil)

// newHTTPAPI builds the adapter without touching the network or any file.
func newHTTPAPI(cfg provider.ProxmoxConfig, opts ...tracing.Option) (*httpAPI, error) {
	u, err := url.Parse(cfg.URL)
	if err != nil {
		return nil, fmt.Errorf("proxmox.url: %w", err)
	}
	return &httpAPI{
		base:      u.JoinPath("api2", "json"),
		tokenFile: cfg.TokenFile,
		caFile:    cfg.CAFile,
		insecure:  cfg.InsecureSkipTLSVerify,
		timeout:   time.Duration(cfg.RequestTimeoutSeconds) * time.Second,
		wrap:      tracing.RestConfigWrapper(opts...),
	}, nil
}

// httpClient returns the shared client, building it once the CA loads. A
// missing CA file is Transient and retried on the next call; one without a
// certificate is Permanent.
func (a *httpAPI) httpClient() (*http.Client, error) {
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.client != nil {
		return a.client, nil
	}
	tlsCfg := &tls.Config{MinVersion: tls.VersionTLS12, InsecureSkipVerify: a.insecure} //nolint:gosec // opt-in, warned about at startup
	if a.caFile != "" && !a.insecure {
		pem, err := os.ReadFile(a.caFile)
		if err != nil {
			return nil, fmt.Errorf("%w: reading CA file: %w", provider.ErrTransient, err)
		}
		pool := x509.NewCertPool()
		if !pool.AppendCertsFromPEM(pem) {
			return nil, permanentf("CA file %s holds no PEM certificate", a.caFile)
		}
		tlsCfg.RootCAs = pool
	}
	transport := &http.Transport{
		Proxy:                 http.ProxyFromEnvironment,
		DialContext:           (&net.Dialer{Timeout: dialTimeout, KeepAlive: 30 * time.Second}).DialContext,
		TLSClientConfig:       tlsCfg,
		TLSHandshakeTimeout:   tlsHandshakeTimeout,
		ResponseHeaderTimeout: a.timeout,
		IdleConnTimeout:       idleConnTimeout,
		MaxIdleConnsPerHost:   maxIdleConnsPerHost,
		// pveproxy speaks HTTP/1.1 only (F18); a non-nil empty map keeps
		// the transport from offering h2.
		ForceAttemptHTTP2: false,
		TLSNextProto:      map[string]func(string, *tls.Conn) http.RoundTripper{},
	}
	a.client = &http.Client{Transport: a.wrap(transport), Timeout: a.timeout}
	return a.client, nil
}

// readToken reads the token file on every request, so a rotated Secret
// takes effect at once. Neither error names the file's content.
func readToken(path string) (string, error) {
	raw, err := os.ReadFile(path)
	if errors.Is(err, fs.ErrNotExist) {
		return "", fmt.Errorf("%w: %w: %s does not exist yet", provider.ErrTransient, errCredentials, path)
	}
	if err != nil {
		return "", fmt.Errorf("%w: reading %s: %w", errCredentials, path, err)
	}
	token := strings.TrimSpace(string(raw))
	if !tokenPattern.MatchString(token) {
		return "", fmt.Errorf("%w: %w: %s is not <user>@<realm>!<tokenid>=<secret>", provider.ErrPermanent, errCredentials, path)
	}
	return token, nil
}

// apiPath joins path segments, escaping each, so a volid's "/" travels as
// %2F as Phase 0 sent it.
func apiPath(segments ...string) string {
	escaped := make([]string, len(segments))
	for i, s := range segments {
		escaped[i] = url.PathEscape(s)
	}
	return strings.Join(escaped, "/")
}

func vmPath(node string, vmid int, rest ...string) string {
	return apiPath(append([]string{"nodes", node, "qemu", strconv.Itoa(vmid)}, rest...)...)
}

// do sends one request and decodes the response's data member into out,
// which may be nil. POST and PUT send form as the body; every method sends
// query in the URL. Any non-2xx answer is an *apiError.
func (a *httpAPI) do(ctx context.Context, method, path string, query, form url.Values, out any) error {
	var body io.Reader
	contentType := ""
	if form != nil {
		body = strings.NewReader(form.Encode())
		contentType = "application/x-www-form-urlencoded"
	}
	return a.send(ctx, method, path, query, body, -1, contentType, out)
}

// send is do with a caller-built body. length is the body's size, or -1
// when do built it from a string the request can measure itself.
func (a *httpAPI) send(ctx context.Context, method, path string, query url.Values, body io.Reader, length int64, contentType string, out any) error {
	token, err := readToken(a.tokenFile)
	if err != nil {
		return err
	}
	client, err := a.httpClient()
	if err != nil {
		return err
	}
	u := a.base.JoinPath(path)
	u.RawQuery = query.Encode()
	req, err := http.NewRequestWithContext(ctx, method, u.String(), body)
	if err != nil {
		return fmt.Errorf("building %s %s: %w", method, path, err)
	}
	if length >= 0 {
		req.ContentLength = length
	}
	if contentType != "" {
		req.Header.Set("Content-Type", contentType)
	}
	req.Header.Set("Accept", "application/json")
	req.Header.Set("Authorization", "PVEAPIToken="+token)
	resp, err := client.Do(req)
	if err != nil {
		return err
	}
	defer func() {
		_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxErrorBody))
		_ = resp.Body.Close()
	}()
	if resp.StatusCode < 200 || resp.StatusCode > 299 {
		raw, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBody))
		reason := strings.TrimPrefix(resp.Status, strconv.Itoa(resp.StatusCode))
		return redact(newAPIError(resp.StatusCode, reason, raw), token)
	}
	var env struct {
		Data json.RawMessage `json:"data"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&env); err != nil {
		return fmt.Errorf("decoding %s %s: %w", method, path, err)
	}
	if out == nil || len(env.Data) == 0 {
		return nil
	}
	if err := json.Unmarshal(env.Data, out); err != nil {
		return fmt.Errorf("decoding %s %s data: %w", method, path, err)
	}
	return nil
}

// redact removes the token's secret from an error Proxmox echoed back, so
// it cannot reach a log line, a span or a Proxy's status (spec entry 14).
func redact(e *apiError, token string) *apiError {
	_, secret, _ := strings.Cut(token, "=")
	if secret == "" {
		return e
	}
	e.Message = strings.ReplaceAll(e.Message, secret, "<redacted>")
	for k, v := range e.Errors {
		e.Errors[k] = strings.ReplaceAll(v, secret, "<redacted>")
	}
	return e
}

// decodeConfig renders every config value as a string: Proxmox sends some
// scalars as JSON numbers (cores) and others as strings (memory). A null
// value is dropped.
func decodeConfig(raw map[string]json.RawMessage) (vmConfig, error) {
	cfg := make(vmConfig, len(raw))
	for k, v := range raw {
		switch {
		case len(v) == 0 || string(v) == "null":
			continue
		case v[0] == '"':
			var s string
			if err := json.Unmarshal(v, &s); err != nil {
				return nil, fmt.Errorf("config key %s: %w", k, err)
			}
			cfg[k] = s
		default:
			cfg[k] = string(v)
		}
	}
	return cfg, nil
}

func (a *httpAPI) Version(ctx context.Context) (v versionInfo, err error) {
	err = a.do(ctx, http.MethodGet, "version", nil, nil, &v)
	return v, err
}

func (a *httpAPI) NodeStatus(ctx context.Context, node string) (s nodeStatus, err error) {
	err = a.do(ctx, http.MethodGet, apiPath("nodes", node, "status"), nil, nil, &s)
	return s, err
}

func (a *httpAPI) StorageStatus(ctx context.Context, node, storage string) (s storageStatus, err error) {
	err = a.do(ctx, http.MethodGet, apiPath("nodes", node, "storage", storage, "status"), nil, nil, &s)
	return s, err
}

func (a *httpAPI) ClusterResources(ctx context.Context) (rs []vmResource, err error) {
	err = a.do(ctx, http.MethodGet, "cluster/resources", url.Values{"type": {"vm"}}, nil, &rs)
	return rs, err
}

func (a *httpAPI) VMConfig(ctx context.Context, node string, vmid int) (vmConfig, error) {
	var raw map[string]json.RawMessage
	if err := a.do(ctx, http.MethodGet, vmPath(node, vmid, "config"), nil, nil, &raw); err != nil {
		return nil, err
	}
	return decodeConfig(raw)
}

func (a *httpAPI) SetVMConfig(ctx context.Context, node string, vmid int, params map[string]string) error {
	form := url.Values{}
	for k, v := range params {
		form.Set(k, v)
	}
	return a.do(ctx, http.MethodPut, vmPath(node, vmid, "config"), nil, form, nil)
}

func (a *httpAPI) Clone(ctx context.Context, node string, vmid int, p cloneParams) (upid string, err error) {
	form := url.Values{
		"newid":       {strconv.Itoa(p.NewID)},
		"name":        {p.Name},
		"description": {p.Description},
		"full":        {"1"},
	}
	if p.Storage != "" {
		form.Set("storage", p.Storage)
	}
	if p.Target != "" {
		form.Set("target", p.Target)
	}
	err = a.do(ctx, http.MethodPost, vmPath(node, vmid, "clone"), nil, form, &upid)
	return upid, err
}

func (a *httpAPI) Resize(ctx context.Context, node string, vmid int, disk, size string) (upid string, err error) {
	form := url.Values{"disk": {disk}, "size": {size}}
	err = a.do(ctx, http.MethodPut, vmPath(node, vmid, "resize"), nil, form, &upid)
	return upid, err
}

func (a *httpAPI) Start(ctx context.Context, node string, vmid int) (upid string, err error) {
	err = a.do(ctx, http.MethodPost, vmPath(node, vmid, "status", "start"), nil, url.Values{}, &upid)
	return upid, err
}

func (a *httpAPI) Stop(ctx context.Context, node string, vmid int) (upid string, err error) {
	err = a.do(ctx, http.MethodPost, vmPath(node, vmid, "status", "stop"), nil, url.Values{}, &upid)
	return upid, err
}

func (a *httpAPI) VMStatus(ctx context.Context, node string, vmid int) (s vmStatus, err error) {
	err = a.do(ctx, http.MethodGet, vmPath(node, vmid, "status", "current"), nil, nil, &s)
	return s, err
}

// Destroy purges the VM from backup jobs and HA and removes disks no config
// references, so nothing of it is left behind (spec entry 11).
func (a *httpAPI) Destroy(ctx context.Context, node string, vmid int) (upid string, err error) {
	query := url.Values{"purge": {"1"}, "destroy-unreferenced-disks": {"1"}}
	err = a.do(ctx, http.MethodDelete, vmPath(node, vmid), query, nil, &upid)
	return upid, err
}

func (a *httpAPI) TaskStatus(ctx context.Context, node, upid string) (s taskStatus, err error) {
	err = a.do(ctx, http.MethodGet, apiPath("nodes", node, "tasks", upid, "status"), nil, nil, &s)
	return s, err
}

func (a *httpAPI) AgentNetworkInterfaces(ctx context.Context, node string, vmid int) ([]agentInterface, error) {
	var out struct {
		Result []agentInterface `json:"result"`
	}
	err := a.do(ctx, http.MethodGet, vmPath(node, vmid, "agent", "network-get-interfaces"), nil, nil, &out)
	return out.Result, err
}

// UploadISO streams r as the file part of a multipart upload (F12). The
// multipart framing is rendered up front, so the request carries an exact
// Content-Length, as Phase 0's curl -F upload did, while r itself is read
// in the transport's chunks and never held in memory.
func (a *httpAPI) UploadISO(ctx context.Context, node, storage, filename string, r io.Reader, size int64) (upid string, err error) {
	var frame bytes.Buffer
	mw := multipart.NewWriter(&frame)
	if err := mw.WriteField("content", "iso"); err != nil {
		return "", fmt.Errorf("framing upload: %w", err)
	}
	if _, err := mw.CreateFormFile("filename", filename); err != nil {
		return "", fmt.Errorf("framing upload: %w", err)
	}
	head := bytes.Clone(frame.Bytes())
	frame.Reset()
	if err := mw.Close(); err != nil {
		return "", fmt.Errorf("framing upload: %w", err)
	}
	tail := frame.Bytes()
	if r == nil {
		r = bytes.NewReader(nil)
	}
	body := io.MultiReader(bytes.NewReader(head), io.LimitReader(r, size), bytes.NewReader(tail))
	length := int64(len(head)) + size + int64(len(tail))
	err = a.send(ctx, http.MethodPost, apiPath("nodes", node, "storage", storage, "upload"), nil, body, length, mw.FormDataContentType(), &upid)
	return upid, err
}

func (a *httpAPI) DeleteVolume(ctx context.Context, node, storage, volid string) (upid string, err error) {
	err = a.do(ctx, http.MethodDelete, apiPath("nodes", node, "storage", storage, "content", volid), nil, nil, &upid)
	return upid, err
}