v0.6.0: slice 6 Phase B — PBS offsite tier (verify + PBS-API client + reporting)
Spike-proven that backup/restore-to-PBS reuse Phase A unchanged; the only new code is the verify capability, a small PBS-API client, and PBSSnapshot reporting. - internal/pbs: fingerprint-pinned, token-authed PBS-API client (Verify/Snapshots/ TaskStatus, node-from-UPID; secret read from /etc/pve/priv/storage/<id>.pw at runtime, never logged) + the verify maintenance loop (own cadence, default 6h, NOT gated/journaled, like the watchdog) + SnapshotStore. - hub: PBSSnapshot filled (namespace/type/id/time/size/owner/protected/encrypted/ verify_state/verify_upid); PBSReporter collector seam; cross-repo golden + bidirectional key-set tests; hub handler parses pbs_snapshots + logs a failed-verify WARN. - backup: report the ACTUAL vzdump mode (parsed from the task log; PVE may downgrade snapshot->stop). proxmox.Storage.Username. config PBSVerifyCadence/secret-dir. --selftest=pbs-verify. Backup/restore-to-PBS unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,209 @@
|
||||
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))
|
||||
if err := c.do(ctx, http.MethodPost, path, &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]
|
||||
}
|
||||
|
||||
// do performs a request, sets the token auth header, and decodes the JSON body into out. The
|
||||
// auth header carries the secret and is NEVER logged.
|
||||
func (c *Client) do(ctx context.Context, method, path string, out any) error {
|
||||
req, err := http.NewRequestWithContext(ctx, method, c.base+path, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
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()
|
||||
body, _ := 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(body))
|
||||
}
|
||||
if out != nil {
|
||||
if err := json.Unmarshal(body, 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
|
||||
}
|
||||
@@ -0,0 +1,161 @@
|
||||
package pbs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// fingerprintOf returns the SHA-256 hex of a TLS test server's leaf cert (what the client pins).
|
||||
func fingerprintOf(ts *httptest.Server) string {
|
||||
sum := sha256.Sum256(ts.Certificate().Raw)
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
// newPBSTestServer spins up a TLS server routing requests to fn, plus its fingerprint.
|
||||
func newPBSTestServer(t *testing.T, fn http.HandlerFunc) (*httptest.Server, string) {
|
||||
t.Helper()
|
||||
ts := httptest.NewTLSServer(fn)
|
||||
t.Cleanup(ts.Close)
|
||||
return ts, fingerprintOf(ts)
|
||||
}
|
||||
|
||||
// hostPort splits an httptest URL "https://127.0.0.1:PORT" into host + port for NewClient.
|
||||
func hostPort(t *testing.T, url string) (string, int) {
|
||||
t.Helper()
|
||||
hp := strings.TrimPrefix(url, "https://")
|
||||
host, port, ok := strings.Cut(hp, ":")
|
||||
if !ok {
|
||||
t.Fatalf("bad url %q", url)
|
||||
}
|
||||
p := 0
|
||||
for _, c := range port {
|
||||
p = p*10 + int(c-'0')
|
||||
}
|
||||
return host, p
|
||||
}
|
||||
|
||||
// TestClient_FingerprintPinEnforced is the headline: a WRONG fingerprint is rejected; the
|
||||
// RIGHT one succeeds (mirrors the slice-1 PVE pin test).
|
||||
func TestClient_FingerprintPinEnforced(t *testing.T) {
|
||||
ts, fp := newPBSTestServer(t, func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Write([]byte(`{"data":[]}`))
|
||||
})
|
||||
host, port := hostPort(t, ts.URL)
|
||||
|
||||
// Wrong fingerprint → connection rejected at the TLS pin.
|
||||
wrong := strings.Repeat("ab", 32)
|
||||
bad, _ := NewClient(Config{Server: host, Port: port, Fingerprint: wrong, TokenID: "u@pbs!t", Secret: "s"})
|
||||
if _, err := bad.Snapshots(context.Background(), "ds"); err == nil || !strings.Contains(err.Error(), "pin mismatch") {
|
||||
t.Fatalf("wrong fingerprint must be rejected with a pin mismatch, got %v", err)
|
||||
}
|
||||
|
||||
// Correct fingerprint → succeeds.
|
||||
good, err := NewClient(Config{Server: host, Port: port, Fingerprint: fp, TokenID: "u@pbs!t", Secret: "s"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := good.Snapshots(context.Background(), "ds"); err != nil {
|
||||
t.Fatalf("correct fingerprint should connect: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestClient_TokenHeaderAndNeverLogged asserts the auth header is the PBS token form and that
|
||||
// the secret is not exposed by the client's printed form.
|
||||
func TestClient_TokenHeaderAndNeverLogged(t *testing.T) {
|
||||
const secret = "super-secret-token-value-xyz"
|
||||
var gotAuth string
|
||||
ts, fp := newPBSTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
gotAuth = r.Header.Get("Authorization")
|
||||
w.Write([]byte(`{"data":[]}`))
|
||||
})
|
||||
host, port := hostPort(t, ts.URL)
|
||||
c, _ := NewClient(Config{Server: host, Port: port, Fingerprint: fp, TokenID: "felhom@pbs!n100", Secret: secret})
|
||||
if _, err := c.Snapshots(context.Background(), "ds"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if gotAuth != "PBSAPIToken=felhom@pbs!n100:"+secret {
|
||||
t.Errorf("auth header = %q, want PBSAPIToken=<id>:<secret>", gotAuth)
|
||||
}
|
||||
// The secret lives only in the unexported authHeader and is never handed to a logger (the
|
||||
// client takes no logger). The "never logged" property of the verify LOOP — which does log
|
||||
// — is asserted in TestVerifyLoop_NeverLogsSecret (verify_test.go).
|
||||
}
|
||||
|
||||
func TestClient_VerifyParsesUPID(t *testing.T) {
|
||||
ts, fp := newPBSTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost || !strings.Contains(r.URL.Path, "/admin/datastore/ds/verify") {
|
||||
t.Errorf("unexpected request %s %s", r.Method, r.URL.Path)
|
||||
}
|
||||
w.Write([]byte(`{"data":"UPID:dooplex:00034582:5269BDD7:00000005:6A282176:verify:ds:felhom@pbs!n100:"}`))
|
||||
})
|
||||
host, port := hostPort(t, ts.URL)
|
||||
c, _ := NewClient(Config{Server: host, Port: port, Fingerprint: fp, TokenID: "u@pbs!t", Secret: "s"})
|
||||
upid, err := c.Verify(context.Background(), "ds")
|
||||
if err != nil || !strings.HasPrefix(upid, "UPID:dooplex:") {
|
||||
t.Fatalf("Verify upid=%q err=%v", upid, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClient_SnapshotsMapToHub(t *testing.T) {
|
||||
body := `{"data":[
|
||||
{"backup-type":"ct","backup-id":"9001","backup-time":1781014713,"size":2518889256,
|
||||
"owner":"felhom@pbs!n100","protected":false,
|
||||
"verification":{"state":"ok","upid":"UPID:dooplex:..:verify:ds:u:"},
|
||||
"files":[{"filename":"pct.conf.blob","crypt-mode":"encrypt","size":240},
|
||||
{"filename":"index.json.blob","crypt-mode":"sign-only","size":646}]},
|
||||
{"backup-type":"ct","backup-id":"9002","backup-time":1781000000,"size":10,"owner":"x",
|
||||
"files":[{"filename":"root.pxar.didx","crypt-mode":"encrypt","size":10}]}
|
||||
]}`
|
||||
ts, fp := newPBSTestServer(t, func(w http.ResponseWriter, _ *http.Request) { w.Write([]byte(body)) })
|
||||
host, port := hostPort(t, ts.URL)
|
||||
c, _ := NewClient(Config{Server: host, Port: port, Fingerprint: fp, TokenID: "u@pbs!t", Secret: "s"})
|
||||
snaps, err := c.Snapshots(context.Background(), "ds")
|
||||
if err != nil || len(snaps) != 2 {
|
||||
t.Fatalf("snaps=%d err=%v", len(snaps), err)
|
||||
}
|
||||
|
||||
// First: verified ok, encrypted, RFC3339 time, default namespace.
|
||||
h0 := snaps[0].ToHub()
|
||||
if h0.VerifyState != VerifyOK || !h0.Encrypted || h0.Namespace != "root" {
|
||||
t.Errorf("h0 = %+v, want verify=ok encrypted=true ns=root", h0)
|
||||
}
|
||||
if h0.BackupTime != "2026-06-09T14:18:33Z" {
|
||||
t.Errorf("backup_time = %q, want RFC3339 UTC", h0.BackupTime)
|
||||
}
|
||||
// Second: NO verification field → verify_state "none".
|
||||
if h1 := snaps[1].ToHub(); h1.VerifyState != VerifyNone {
|
||||
t.Errorf("absent verification must map to %q, got %q", VerifyNone, h1.VerifyState)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNodeFromUPID(t *testing.T) {
|
||||
cases := map[string]string{
|
||||
"UPID:dooplex:00034582:5269BDD7:00000005:6A282176:verify:ds:u:": "dooplex",
|
||||
"UPID:demo-felhom:00:00:00:00:vzdump:9001:root@pam:": "demo-felhom",
|
||||
"not-a-upid": "",
|
||||
"": "",
|
||||
}
|
||||
for in, want := range cases {
|
||||
if got := NodeFromUPID(in); got != want {
|
||||
t.Errorf("NodeFromUPID(%q) = %q, want %q", in, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestNormalizeFingerprint(t *testing.T) {
|
||||
if _, err := normalizeFingerprint("3b:95:5a"); err == nil {
|
||||
t.Error("short fingerprint must error")
|
||||
}
|
||||
if _, err := normalizeFingerprint(strings.Repeat("g", 64)); err == nil {
|
||||
t.Error("non-hex must error")
|
||||
}
|
||||
got, err := normalizeFingerprint("3B:95:" + strings.Repeat("a", 60))
|
||||
if err != nil || strings.Contains(got, ":") || got != strings.ToLower(got) {
|
||||
t.Errorf("normalize = %q err=%v (want lowercased, colons stripped)", got, err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
// Package pbs is the agent's PBS (Proxmox Backup Server) API client + the verify
|
||||
// maintenance loop (doc 03 §8, slice 6 Phase B).
|
||||
//
|
||||
// PBS is a SEPARATE server (its own host:8007, its own API + token auth), distinct from the
|
||||
// PVE proxmox.Client — so this is the agent's SECOND privileged external surface and gets the
|
||||
// same slice-1 discipline: TLS fingerprint-pinned to the PBS leaf cert, token auth, typed,
|
||||
// context-aware, NO shell.
|
||||
//
|
||||
// The spike (felhom.eu/documentation/tests/phase5-pbs-spike-findings.md) proved that
|
||||
// backup-to-PBS and restore-from-PBS reuse Phase A UNCHANGED (PBS is just a storage target +
|
||||
// a volid). The only genuinely new code here is:
|
||||
// - client.go — the PBS-API client: Verify, Snapshots, TaskStatus (node from the UPID).
|
||||
// - report.go — Snapshot → hub.PBSSnapshot mapping + a SnapshotStore implementing the hub
|
||||
// PBSReporter seam (the collector reads it; hub does not import pbs).
|
||||
// - verify.go — the verify maintenance loop on its OWN cadence (the cheap, frequent,
|
||||
// ciphertext-level integrity check — needs NO encryption key, unlike the
|
||||
// full self-restore-test). It is a reporting/maintenance task like the
|
||||
// slice-5 watchdog: it does NOT go through the reconcile gate/journal.
|
||||
//
|
||||
// The encryption key is never needed here (verify is ciphertext-level), and the token secret
|
||||
// is read at runtime from /etc/pve/priv/storage/<id>.pw — referenced by location, never
|
||||
// logged or committed (zero-knowledge holds; the PBS server has no client key — spike B6).
|
||||
package pbs
|
||||
@@ -0,0 +1,48 @@
|
||||
package pbs
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// pinnedTLS builds a tls.Config that pins the PBS server's leaf cert by SHA-256 — the same
|
||||
// model as the PVE client (proxmox/tls.go). PBS serves a self-signed cert, so we disable the
|
||||
// default chain check but enforce an exact-cert match: a spoofed PBS presents a different
|
||||
// fingerprint and is rejected. fingerprint is hex with optional colons (the form in
|
||||
// /etc/pve/storage.cfg and the slice-5 durable_id).
|
||||
func pinnedTLS(fingerprint string) (*tls.Config, error) {
|
||||
want, err := normalizeFingerprint(fingerprint)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &tls.Config{
|
||||
InsecureSkipVerify: true, //nolint:gosec // replaced by the exact-cert pin below
|
||||
VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
|
||||
if len(rawCerts) == 0 {
|
||||
return fmt.Errorf("pbs: TLS pin: peer presented no certificate")
|
||||
}
|
||||
got := sha256.Sum256(rawCerts[0])
|
||||
if hex.EncodeToString(got[:]) != want {
|
||||
return fmt.Errorf("pbs: TLS pin mismatch: server cert sha256 does not match configured fingerprint")
|
||||
}
|
||||
return nil
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
// normalizeFingerprint lowercases and strips colons/whitespace, validating a 64-char
|
||||
// (32-byte) hex SHA-256.
|
||||
func normalizeFingerprint(fp string) (string, error) {
|
||||
s := strings.ToLower(strings.NewReplacer(":", "", " ", "", "\t", "").Replace(fp))
|
||||
if len(s) != 64 {
|
||||
return "", fmt.Errorf("pbs: fingerprint must be a SHA-256 (64 hex chars), got %d", len(s))
|
||||
}
|
||||
if _, err := hex.DecodeString(s); err != nil {
|
||||
return "", fmt.Errorf("pbs: fingerprint is not valid hex: %w", err)
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
package pbs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
|
||||
)
|
||||
|
||||
// verify-state constants for the reported PBSSnapshot.
|
||||
const (
|
||||
VerifyOK = "ok"
|
||||
VerifyFailed = "failed"
|
||||
VerifyNone = "none" // no verify has run yet (PBS omits the verification field)
|
||||
)
|
||||
|
||||
// ToHub maps a PBS API Snapshot to the hub.PBSSnapshot wire record (slice 6 Phase B). The
|
||||
// backup-time epoch becomes RFC3339; `encrypted` is derived from the data files'
|
||||
// crypt-mode (any "encrypt" → encrypted); verify_state is "none" until a verify runs.
|
||||
func (s Snapshot) ToHub() hub.PBSSnapshot {
|
||||
ns := s.Namespace
|
||||
if ns == "" {
|
||||
ns = "root"
|
||||
}
|
||||
out := hub.PBSSnapshot{
|
||||
Namespace: ns,
|
||||
BackupType: s.BackupType,
|
||||
BackupID: s.BackupID,
|
||||
BackupTime: time.Unix(s.BackupTime, 0).UTC().Format(time.RFC3339),
|
||||
SizeBytes: s.Size,
|
||||
Owner: s.Owner,
|
||||
Protected: s.Protected,
|
||||
Encrypted: s.encrypted(),
|
||||
VerifyState: VerifyNone,
|
||||
}
|
||||
if s.Verification != nil {
|
||||
out.VerifyState = s.Verification.State
|
||||
out.VerifyUPID = s.Verification.UPID
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// encrypted reports whether the snapshot's DATA is client-side encrypted (any data file with
|
||||
// crypt-mode "encrypt"; index.json is "sign-only" and is ignored).
|
||||
func (s Snapshot) encrypted() bool {
|
||||
for _, f := range s.Files {
|
||||
if f.CryptMode == "encrypt" {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// SnapshotStore holds the latest reported PBS snapshots per datastore — the point-in-time
|
||||
// state the host-report surfaces. The verify loop writes it; the collector reads it via the
|
||||
// hub PBSReporter seam. Mutex-guarded (concurrent collector vs loop).
|
||||
type SnapshotStore struct {
|
||||
mu sync.Mutex
|
||||
byDatastore map[string][]hub.PBSSnapshot
|
||||
}
|
||||
|
||||
// NewSnapshotStore builds an empty store.
|
||||
func NewSnapshotStore() *SnapshotStore {
|
||||
return &SnapshotStore{byDatastore: map[string][]hub.PBSSnapshot{}}
|
||||
}
|
||||
|
||||
// Record replaces the snapshot set for a datastore.
|
||||
func (s *SnapshotStore) Record(datastore string, snaps []hub.PBSSnapshot) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.byDatastore[datastore] = snaps
|
||||
}
|
||||
|
||||
// PBSSnapshots implements hub.PBSReporter — all known snapshots across datastores.
|
||||
func (s *SnapshotStore) PBSSnapshots(context.Context) []hub.PBSSnapshot {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
out := []hub.PBSSnapshot{}
|
||||
for _, snaps := range s.byDatastore {
|
||||
out = append(out, snaps...)
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,127 @@
|
||||
package pbs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
|
||||
)
|
||||
|
||||
// DefaultVerifyCadence is the verify maintenance interval — more frequent than the full
|
||||
// self-restore-test, because it's the cheap, key-free, ciphertext-level integrity check (§8).
|
||||
const DefaultVerifyCadence = 6 * time.Hour
|
||||
|
||||
// Target is one PBS datastore to verify, with its client.
|
||||
type Target struct {
|
||||
Datastore string
|
||||
Client *Client
|
||||
}
|
||||
|
||||
// Targets resolves the current set of PBS datastores to verify (re-derived each cycle from
|
||||
// the PVE storage config — wired in main.go so pbs stays decoupled from how clients are built).
|
||||
type Targets func(ctx context.Context) ([]Target, error)
|
||||
|
||||
// VerifyLoop is the verify maintenance loop (slice 6 Phase B). It runs on its OWN cadence and
|
||||
// is a reporting/maintenance task like the slice-5 watchdog — it does NOT go through the
|
||||
// reconcile gate/journal (it mutates no guest). Each cycle, per datastore: trigger a verify →
|
||||
// poll the task → re-list snapshots → record the per-snapshot verify-state for the report.
|
||||
type VerifyLoop struct {
|
||||
targets Targets
|
||||
store *SnapshotStore
|
||||
cadence time.Duration
|
||||
logger *slog.Logger
|
||||
}
|
||||
|
||||
// VerifyLoopOptions configures a VerifyLoop.
|
||||
type VerifyLoopOptions struct {
|
||||
Targets Targets
|
||||
Store *SnapshotStore
|
||||
Cadence time.Duration // 0 → default 6h; negative → disabled
|
||||
Logger *slog.Logger
|
||||
}
|
||||
|
||||
// NewVerifyLoop builds a VerifyLoop.
|
||||
func NewVerifyLoop(opts VerifyLoopOptions) *VerifyLoop {
|
||||
logger := opts.Logger
|
||||
if logger == nil {
|
||||
logger = slog.Default()
|
||||
}
|
||||
cadence := opts.Cadence
|
||||
if cadence == 0 {
|
||||
cadence = DefaultVerifyCadence
|
||||
}
|
||||
return &VerifyLoop{targets: opts.Targets, store: opts.Store, cadence: cadence, logger: logger}
|
||||
}
|
||||
|
||||
// Run verifies on the cadence until ctx is cancelled. It does an immediate first pass (so a
|
||||
// freshly-started agent reports snapshot inventory + verify-state promptly), then on each
|
||||
// tick. A negative cadence (or nil targets/store) disables it. Returns nil on cancellation.
|
||||
func (l *VerifyLoop) Run(ctx context.Context) error {
|
||||
if l.cadence < 0 || l.targets == nil || l.store == nil {
|
||||
l.logger.Info("pbs: verify loop disabled")
|
||||
<-ctx.Done()
|
||||
return nil
|
||||
}
|
||||
l.logger.Info("pbs: verify loop starting", "cadence", l.cadence)
|
||||
l.tick(ctx) // immediate inventory + verify
|
||||
t := time.NewTicker(l.cadence)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
l.logger.Info("pbs: verify loop shutting down", "reason", ctx.Err())
|
||||
return nil
|
||||
case <-t.C:
|
||||
l.tick(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// RunOnce performs a single synchronous verify+list pass over all targets (used by the
|
||||
// selftest harness and the live runbook). Same work as one cadence tick.
|
||||
func (l *VerifyLoop) RunOnce(ctx context.Context) { l.tick(ctx) }
|
||||
|
||||
// tick verifies + re-lists each target datastore once. Deterministic enough to drive directly
|
||||
// in tests. A per-target error is logged and skipped (other datastores still report).
|
||||
func (l *VerifyLoop) tick(ctx context.Context) {
|
||||
targets, err := l.targets(ctx)
|
||||
if err != nil {
|
||||
l.logger.Warn("pbs: verify loop could not resolve targets; skipping", "err", err)
|
||||
return
|
||||
}
|
||||
for _, t := range targets {
|
||||
l.verifyOne(ctx, t)
|
||||
}
|
||||
}
|
||||
|
||||
// verifyOne triggers a verify, waits for it, then re-lists + records the snapshots' state.
|
||||
func (l *VerifyLoop) verifyOne(ctx context.Context, t Target) {
|
||||
if upid, err := t.Client.Verify(ctx, t.Datastore); err != nil {
|
||||
// Verify-trigger failure is non-fatal: still re-list so we report current state.
|
||||
l.logger.Warn("pbs: verify trigger failed; reporting current snapshot state", "datastore", t.Datastore, "err", err)
|
||||
} else if err := t.Client.WaitVerify(ctx, upid, 2*time.Second, 30*time.Minute); err != nil {
|
||||
l.logger.Warn("pbs: verify task wait failed; reporting current snapshot state", "datastore", t.Datastore, "err", err)
|
||||
}
|
||||
|
||||
snaps, err := t.Client.Snapshots(ctx, t.Datastore)
|
||||
if err != nil {
|
||||
l.logger.Warn("pbs: snapshot list failed", "datastore", t.Datastore, "err", err)
|
||||
return
|
||||
}
|
||||
out := make([]hub.PBSSnapshot, 0, len(snaps))
|
||||
failed := 0
|
||||
for _, s := range snaps {
|
||||
h := s.ToHub()
|
||||
if h.VerifyState == VerifyFailed {
|
||||
failed++
|
||||
}
|
||||
out = append(out, h)
|
||||
}
|
||||
l.store.Record(t.Datastore, out)
|
||||
if failed > 0 {
|
||||
l.logger.Error("pbs: datastore has FAILED-verify snapshots", "datastore", t.Datastore, "failed", failed, "total", len(out))
|
||||
} else {
|
||||
l.logger.Info("pbs: verify cycle complete", "datastore", t.Datastore, "snapshots", len(out))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,100 @@
|
||||
package pbs
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// verifyServer is a fake PBS that answers verify POSTs, task status, and a snapshot list whose
|
||||
// verification.state is controlled by `state` ("" → no verification field → none).
|
||||
func verifyServer(t *testing.T, state string) (*httptest.Server, string) {
|
||||
t.Helper()
|
||||
verification := ""
|
||||
if state != "" {
|
||||
verification = `"verification":{"state":"` + state + `","upid":"UPID:dooplex:0:0:0:0:verify:ds:u:"},`
|
||||
}
|
||||
body := `{"data":[{"backup-type":"ct","backup-id":"9001","backup-time":1781014713,"size":10,"owner":"u",` +
|
||||
verification + `"files":[{"filename":"root.pxar.didx","crypt-mode":"encrypt","size":10}]}]}`
|
||||
return newPBSTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
switch {
|
||||
case strings.Contains(r.URL.Path, "/verify"):
|
||||
w.Write([]byte(`{"data":"UPID:dooplex:0:0:0:0:verify:ds:u:"}`))
|
||||
case strings.Contains(r.URL.Path, "/status"):
|
||||
w.Write([]byte(`{"data":{"status":"stopped","exitstatus":"OK","node":"dooplex"}}`))
|
||||
default: // snapshots
|
||||
w.Write([]byte(body))
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestVerifyLoop_RecordsState(t *testing.T) {
|
||||
ts, fp := verifyServer(t, "ok")
|
||||
host, port := hostPort(t, ts.URL)
|
||||
c, _ := NewClient(Config{Server: host, Port: port, Fingerprint: fp, TokenID: "u@pbs!t", Secret: "s"})
|
||||
store := NewSnapshotStore()
|
||||
loop := NewVerifyLoop(VerifyLoopOptions{
|
||||
Targets: func(context.Context) ([]Target, error) { return []Target{{Datastore: "ds", Client: c}}, nil },
|
||||
Store: store,
|
||||
})
|
||||
loop.RunOnce(context.Background())
|
||||
|
||||
snaps := store.PBSSnapshots(context.Background())
|
||||
if len(snaps) != 1 || snaps[0].VerifyState != VerifyOK {
|
||||
t.Fatalf("loop should record 1 ok snapshot, got %+v", snaps)
|
||||
}
|
||||
}
|
||||
|
||||
func TestVerifyLoop_FailedVerifyRecorded(t *testing.T) {
|
||||
ts, fp := verifyServer(t, "failed")
|
||||
host, port := hostPort(t, ts.URL)
|
||||
c, _ := NewClient(Config{Server: host, Port: port, Fingerprint: fp, TokenID: "u@pbs!t", Secret: "s"})
|
||||
store := NewSnapshotStore()
|
||||
NewVerifyLoop(VerifyLoopOptions{
|
||||
Targets: func(context.Context) ([]Target, error) { return []Target{{Datastore: "ds", Client: c}}, nil },
|
||||
Store: store,
|
||||
}).RunOnce(context.Background())
|
||||
if s := store.PBSSnapshots(context.Background()); len(s) != 1 || s[0].VerifyState != VerifyFailed {
|
||||
t.Fatalf("failed verify must be recorded, got %+v", s)
|
||||
}
|
||||
}
|
||||
|
||||
// TestVerifyLoop_NeverLogsSecret runs a full cycle with the loop's logger capturing output and
|
||||
// asserts the token secret never appears in any log line.
|
||||
func TestVerifyLoop_NeverLogsSecret(t *testing.T) {
|
||||
const secret = "tok-secret-DO-NOT-LOG-7f3a"
|
||||
ts, fp := verifyServer(t, "ok")
|
||||
host, port := hostPort(t, ts.URL)
|
||||
c, _ := NewClient(Config{Server: host, Port: port, Fingerprint: fp, TokenID: "felhom@pbs!n100", Secret: secret})
|
||||
|
||||
var buf bytes.Buffer
|
||||
logger := slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}))
|
||||
loop := NewVerifyLoop(VerifyLoopOptions{
|
||||
Targets: func(context.Context) ([]Target, error) { return []Target{{Datastore: "ds", Client: c}}, nil },
|
||||
Store: NewSnapshotStore(),
|
||||
Logger: logger,
|
||||
})
|
||||
loop.RunOnce(context.Background())
|
||||
if strings.Contains(buf.String(), secret) {
|
||||
t.Fatalf("the verify loop logged the token secret")
|
||||
}
|
||||
}
|
||||
|
||||
func TestVerifyLoop_DisabledByNegativeCadence(t *testing.T) {
|
||||
loop := NewVerifyLoop(VerifyLoopOptions{
|
||||
Targets: func(context.Context) ([]Target, error) { return nil, nil },
|
||||
Store: NewSnapshotStore(),
|
||||
Cadence: -1,
|
||||
})
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- loop.Run(ctx) }()
|
||||
cancel()
|
||||
if err := <-done; err != nil {
|
||||
t.Fatalf("disabled loop Run should return nil on cancel, got %v", err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user