felhom-agent/internal/pbs/client.go

package pbs

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

	"gitea.dooplex.hu/admin/felhom-agent/internal/httpx"
)

// 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
	namespace  string // "" = root ns (whole-datastore); set = per-customer tenant scope (S4)
}

// 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)
	Namespace   string // PBS namespace (from storage.cfg `namespace`); "" = root. S4 per-customer tenancy.
	Timeout     time.Duration

	// IdleConnTimeout bounds how long this client's idle keep-alive connections are retained.
	// Zero means httpx.DefaultIdleConnTimeout (90s) — it does NOT mean "no timeout", which is the
	// R-344 defect. Production leaves it unset; only tests set it, to avoid a 90-second wait.
	IdleConnTimeout 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,
		namespace:  cfg.Namespace,
		http: &http.Client{
			Timeout: timeout,
			// R-344: this transport MUST come from httpx. pbsTargetsFromPVE (cmd/felhom-agent/
			// main.go) builds a fresh Client every cycle and drops the previous one, so a
			// transport with no idle timeout strands one connection per cycle, forever, on both
			// sides. That leaked 388 sockets onto ep0 in 46 hours. Pinned by
			// TestAbandonedClientsReleaseTheirConnections — do not inline an http.Transport here.
			Transport: httpx.NewTransport(tlsCfg, cfg.IdleConnTimeout),
		},
	}, nil
}

// Verify triggers a verify (POST /admin/datastore/<ds>/verify) and returns the task UPID; the
// cheap, key-free, ciphertext-level integrity check (doc 03 §8). When the client is namespace-
// scoped (S4 per-customer tenancy) the verify is confined to that namespace (`ns=`), which a
// DatastoreBackup token can trigger on its OWN namespace — no Datastore.Verify / admin widening
// (Phase-1 confirmed live 2026-07-04). Root-ns (unscoped) clients verify the whole datastore as
// before (needs Datastore.Verify, e.g. the DooPlex felhom-pbs n100 token).
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 c.namespace != "" {
		form.Set("ns", c.namespace)
	}
	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). A namespace-scoped
// client (S4) lists ONLY its namespace (`?ns=`) — the unscoped call targets the datastore root,
// which a per-customer DatastoreBackup token cannot read (Phase-1: 403 without ns). Root-ns
// clients list the root namespace as before.
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 c.namespace != "" {
		path += "?ns=" + url.QueryEscape(c.namespace)
	}
	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]
}

// ErrUnauthorized is returned by ProbeAuth when PBS REJECTS the credential (HTTP 401).
//
// It is a distinct sentinel because a rejected credential and an unreachable server demand opposite
// responses: 401 is terminal until the credential is replaced (the hub must re-key), while a dial
// error is transient and must NOT trigger a re-issue — mistaking one for the other would either
// leave a dead tier green (the R-39 failure) or burn a fresh secret on every network blip.
var ErrUnauthorized = errors.New("pbs: unauthorized (401) — the token secret is not accepted")

// ProbeAuth asks PBS the cheapest question that requires authentication: GET /version.
//
// WHY THIS EXISTS (R-39 leg c). The DR tier could be `applied` and dead at the same time: PVE holds
// a storage entry, the agent's marker says converged, and every PBS request 401s because the entry
// is pinned to a superseded credential. Nothing noticed, because the one loop that could — the
// 15-minute PBS verify loop — could not even READ the credential to test it (the non-root agent
// writes /etc/pve/priv/storage/<id>.pw through a root wrapper and had no read verb). With the
// wrapper's `read` verb this probe finally closes that gap, and its result becomes a LOUD
// `auth_failed` state the hub self-heals instead of a Warn-and-skip.
//
// /version is deliberate: it needs no datastore, no namespace and no privileges beyond a valid
// token, so a 401 here means the CREDENTIAL is bad — not that a datastore is missing or an ACL is
// too narrow. That distinction is what makes the state safe to auto-remediate.
func (c *Client) ProbeAuth(ctx context.Context) error {
	err := c.do(ctx, http.MethodGet, "/version", nil)
	if err == nil {
		return nil
	}
	if isUnauthorized(err) {
		return ErrUnauthorized
	}
	return err
}

// isUnauthorized classifies a doBody error as an authentication rejection. doBody formats non-2xx as
// "... -> HTTP <code>: <body>", so the code is matched on that shape. 403 is deliberately NOT
// included: a valid token with too narrow an ACL is a permissions problem, and re-keying it would
// mint credentials forever without fixing anything.
func isUnauthorized(err error) bool {
	return err != nil && strings.Contains(err.Error(), "-> HTTP 401")
}

// 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
}