felhom-agent/internal/pbs/client.go

package pbs

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
	"time"
)

// Client is the PBS-API client for ONE PBS server. Construct with NewClient. It is pure (no
// logger — it must never log the token secret); callers log around it.
type Client struct {
	base       string // https://<server>:<port>/api2/json
	authHeader string // "PBSAPIToken=<tokenid>:<secret>" — SECRET; never logged
	http       *http.Client
}

// Config builds a Client. Secret is read by the caller from /etc/pve/priv/storage/<id>.pw at
// runtime (referenced by location, never committed).
type Config struct {
	Server      string // PBS host (no scheme), e.g. "192.168.0.180"
	Port        int    // default 8007
	Fingerprint string // SHA-256 of the PBS leaf cert (colons optional)
	TokenID     string // e.g. "felhom@pbs!n100" (from storage.cfg `username`)
	Secret      string // token secret (from <id>.pw)
	Timeout     time.Duration
}

// NewClient builds a fingerprint-pinned, token-authed PBS client.
func NewClient(cfg Config) (*Client, error) {
	if cfg.Server == "" || cfg.Fingerprint == "" || cfg.TokenID == "" || cfg.Secret == "" {
		return nil, fmt.Errorf("pbs: NewClient needs server, fingerprint, tokenid and secret")
	}
	tlsCfg, err := pinnedTLS(cfg.Fingerprint)
	if err != nil {
		return nil, err
	}
	port := cfg.Port
	if port == 0 {
		port = 8007
	}
	timeout := cfg.Timeout
	if timeout == 0 {
		timeout = 30 * time.Second
	}
	return &Client{
		base:       fmt.Sprintf("https://%s:%d/api2/json", cfg.Server, port),
		authHeader: "PBSAPIToken=" + cfg.TokenID + ":" + cfg.Secret,
		http: &http.Client{
			Timeout:   timeout,
			Transport: &http.Transport{TLSClientConfig: tlsCfg},
		},
	}, nil
}

// Verify triggers a datastore verify (POST /admin/datastore/<ds>/verify) and returns the
// task UPID. With no snapshots it verifies the whole datastore; the cheap, key-free,
// ciphertext-level integrity check (doc 03 §8). Needs the token's Datastore.Verify (in
// DatastoreAdmin). Per-snapshot scoping is a future refinement; whole-datastore is the spike-
// proven path.
func (c *Client) Verify(ctx context.Context, datastore string, _ ...string) (string, error) {
	var out struct {
		Data string `json:"data"`
	}
	path := fmt.Sprintf("/admin/datastore/%s/verify", url.PathEscape(datastore))
	// ignore-verified=false → RE-verify even already-verified snapshots. This is what makes
	// the check actually detect corruption (PBS's default ignore-verified=true skips them, so
	// a chunk that rots after its first verify would never be re-checked). The cost is real
	// re-read I/O; for a large datastore a future refinement is outdated-after-based scoping.
	form := url.Values{"ignore-verified": {"false"}}
	if err := c.post(ctx, path, form, &out); err != nil {
		return "", err
	}
	return out.Data, nil
}

// Snapshot is one PBS snapshot as the API returns it (GET /admin/datastore/<ds>/snapshots).
type Snapshot struct {
	BackupType   string `json:"backup-type"` // ct | vm
	BackupID     string `json:"backup-id"`
	BackupTime   int64  `json:"backup-time"` // epoch seconds
	Size         int64  `json:"size"`
	Owner        string `json:"owner"`
	Protected    bool   `json:"protected"`
	Namespace    string `json:"ns"` // "" = root namespace
	Verification *struct {
		State string `json:"state"` // ok | failed
		UPID  string `json:"upid"`
	} `json:"verification"`
	Files []struct {
		Filename  string `json:"filename"`
		CryptMode string `json:"crypt-mode"` // encrypt | sign-only | (none)
		Size      int64  `json:"size"`
	} `json:"files"`
}

// Snapshots lists the datastore's snapshots (incl. the verification field).
func (c *Client) Snapshots(ctx context.Context, datastore string) ([]Snapshot, error) {
	var out struct {
		Data []Snapshot `json:"data"`
	}
	path := fmt.Sprintf("/admin/datastore/%s/snapshots", url.PathEscape(datastore))
	if err := c.do(ctx, http.MethodGet, path, &out); err != nil {
		return nil, err
	}
	return out.Data, nil
}

// TaskStatus is the subset of a PBS task status we need.
type TaskStatus struct {
	Status     string `json:"status"`     // running | stopped
	ExitStatus string `json:"exitstatus"` // present once stopped ("OK" or an error)
	Node       string `json:"node"`
}

// Running reports whether the task is still executing.
func (t TaskStatus) Running() bool { return t.Status == "running" }

// OK reports whether the task stopped successfully.
func (t TaskStatus) OK() bool { return t.Status == "stopped" && t.ExitStatus == "OK" }

// TaskStatus polls a verify task. The PBS node name is extracted from the UPID — querying the
// wrong node (e.g. "localhost") returns exitstatus "unknown" (the spike B4 gotcha).
func (c *Client) TaskStatus(ctx context.Context, upid string) (TaskStatus, error) {
	node := NodeFromUPID(upid)
	if node == "" {
		return TaskStatus{}, fmt.Errorf("pbs: cannot extract node from UPID %q", upid)
	}
	var out struct {
		Data TaskStatus `json:"data"`
	}
	path := fmt.Sprintf("/nodes/%s/tasks/%s/status", url.PathEscape(node), url.PathEscape(upid))
	if err := c.do(ctx, http.MethodGet, path, &out); err != nil {
		return TaskStatus{}, err
	}
	return out.Data, nil
}

// WaitVerify polls a verify task until it stops or ctx/timeout elapses (best-effort — a poll
// failure returns the error; the caller re-lists snapshots for the authoritative state).
func (c *Client) WaitVerify(ctx context.Context, upid string, poll, timeout time.Duration) error {
	if poll <= 0 {
		poll = 2 * time.Second
	}
	if timeout <= 0 {
		timeout = 10 * time.Minute
	}
	deadline := time.Now().Add(timeout)
	t := time.NewTicker(poll)
	defer t.Stop()
	for {
		st, err := c.TaskStatus(ctx, upid)
		if err != nil {
			return err
		}
		if !st.Running() {
			return nil
		}
		if time.Now().After(deadline) {
			return fmt.Errorf("pbs: verify task %s did not finish within %s", upid, timeout)
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-t.C:
		}
	}
}

// NodeFromUPID extracts the node from a PBS/PVE UPID ("UPID:<node>:<pid>:...").
func NodeFromUPID(upid string) string {
	parts := strings.Split(upid, ":")
	if len(parts) < 2 || parts[0] != "UPID" {
		return ""
	}
	return parts[1]
}

// post performs a form-encoded POST (PBS mutating ops take form params).
func (c *Client) post(ctx context.Context, path string, form url.Values, out any) error {
	return c.doBody(ctx, http.MethodPost, path, strings.NewReader(form.Encode()), "application/x-www-form-urlencoded", out)
}

// do performs a request (no body), sets the token auth header, and decodes JSON into out.
func (c *Client) do(ctx context.Context, method, path string, out any) error {
	return c.doBody(ctx, method, path, nil, "", out)
}

// doBody is the shared request path. The auth header carries the secret and is NEVER logged.
func (c *Client) doBody(ctx context.Context, method, path string, body io.Reader, contentType string, out any) error {
	req, err := http.NewRequestWithContext(ctx, method, c.base+path, body)
	if err != nil {
		return err
	}
	if contentType != "" {
		req.Header.Set("Content-Type", contentType)
	}
	req.Header.Set("Authorization", c.authHeader)
	resp, err := c.http.Do(req)
	if err != nil {
		return fmt.Errorf("pbs: %s %s: %w", method, path, err)
	}
	defer resp.Body.Close()
	respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return fmt.Errorf("pbs: %s %s -> HTTP %d: %s", method, path, resp.StatusCode, trimBody(respBody))
	}
	if out != nil {
		if err := json.Unmarshal(respBody, out); err != nil {
			return fmt.Errorf("pbs: decoding %s %s: %w", method, path, err)
		}
	}
	return nil
}

func trimBody(b []byte) string {
	s := strings.TrimSpace(string(b))
	if len(s) > 300 {
		return s[:300] + "…"
	}
	return s
}