From 9dcd5697ccb6b4e83bdbc5f7b3e37dcdbfbd6572 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Wed, 14 Oct 2026 11:35:55 +0100 Subject: [PATCH] Keep incomplete Proxmox guest inventories out of publication Require whole bounded, unambiguous enumerations on the efficient and node fallback routes. Preserve scoped empty inventories, LXC integer compatibility and ordinary counter/receipt semantics. Validate retained readings, History, breaches and template policy through actual fallback and recovery. Change-source: pulse-maintainer --- .../v6/internal/subsystems/monitoring.md | 32 +++ .../guest_inventory_admission_test.go | 208 ++++++++++++++++++ pkg/proxmox/client.go | 44 ++-- pkg/proxmox/client_api_more_test.go | 2 +- .../cluster_client_additional_api_test.go | 4 +- pkg/proxmox/guest_inventory_response.go | 85 +++++++ pkg/proxmox/guest_inventory_response_test.go | 102 +++++++++ pkg/proxmox/io_counters_test.go | 64 ++++++ pkg/proxmox/node_inventory_response.go | 6 +- 9 files changed, 514 insertions(+), 33 deletions(-) create mode 100644 internal/monitoring/guest_inventory_admission_test.go create mode 100644 pkg/proxmox/guest_inventory_response.go create mode 100644 pkg/proxmox/guest_inventory_response_test.go diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 905b41c152..466c366b66 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -7661,3 +7661,35 @@ The canonical Docker-ingest source guard follows both guest enumeration routes through `publishPVEGuests` to the canonical state update. Publication reconciliation updates its former direct-call assertions without changing Docker ingestion or the runtime ownership gate; the behavioral withdrawal controls remain required. + +### Complete guest enumeration before publication (14 October 2026) + +PVE `cluster/resources`, node `qemu` and node `lxc` enumeration consume and +close the whole existing 8 MiB-bounded HTTP response before admission. Only +HTTP 200 containing a complete UTF-8 JSON object with an explicit canonical +`data` array is authoritative inventory. Missing/null data, duplicate or +case-variant supported fields, conflicting errors, trailing values, incomplete +bodies, invalid rows and duplicate guest identities fail the whole enumeration; +no partial inventory or provider body is returned as a successful reading. +Cluster guest identities require a node and positive cluster-wide unique VMID. +Node lists require positive unique VMIDs; existing LXC string integers remain. +Scoped empty arrays, optional names/status/metrics, null counter presence, +templates, unique provider metadata and unfiltered non-guest resources remain +supported. Observation timestamps are assigned only to an admitted inventory. + +A refused efficient enumeration can use the existing ordinary node fallback. +Failed node enumerations retain their earlier guest observations, History and +backup-template eligibility; a complete fallback can still remove deleted +guests and restore ordinary sampling. Complete cluster recovery returns to +its efficient route. This changes no TLS/pin, credentials, permissions, +polling cadence, guest-command/backup-lock checks, completion uncertainty, +publication ownership or grace-period rules. + +`guest_inventory_response_test.go` covers complete-body and all-or-nothing +admission on all three real HTTP readers; `io_counters_test.go` covers +scoped-empty, extended fields, counter presence and receipt-time controls. +`guest_inventory_admission_test.go` covers actual efficient/fallback collection, +owned publication, retained observation times, breach/History/template +continuity, independent-connection preservation and normal recovery/removal. +These controlled source proofs do not establish native thaw/restart recovery, +installed trust, report attribution or release availability. diff --git a/internal/monitoring/guest_inventory_admission_test.go b/internal/monitoring/guest_inventory_admission_test.go new file mode 100644 index 0000000000..c88f2993e0 --- /dev/null +++ b/internal/monitoring/guest_inventory_admission_test.go @@ -0,0 +1,208 @@ +package monitoring + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" + "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" +) + +// Walk the actual efficient enumeration -> fallback -> accepted publication -> +// canonical projection. Malformed successes cannot remove guests, refresh their +// last observation, clear a breach or invalidate template backup eligibility. +// All endpoints are synthetic; no backup, guest command or live send occurs. +func TestGuestInventoryFailurePreservesOwnedReadingsAndRecovers(t *testing.T) { + for _, failure := range []string{"missing-data", "trailing-value", "truncated-prefix", "duplicate-id"} { + t.Run(failure, func(t *testing.T) { + t.Setenv("PULSE_DATA_DIR", t.TempDir()) + var phase, commands, clusterReads, listReads atomic.Int32 + corrupt := func(w http.ResponseWriter, prefix string) { + switch failure { + case "missing-data": + fmt.Fprint(w, `{}`) + case "trailing-value": + fmt.Fprint(w, prefix+` {"errors":{"message":"fixture-private-detail"}}`) + case "truncated-prefix": + w.Header().Set("Content-Length", fmt.Sprint(len(prefix)+100)) + fmt.Fprint(w, prefix) + case "duplicate-id": + var payload struct { + Data []json.RawMessage `json:"data"` + } + if err := json.Unmarshal([]byte(prefix), &payload); err != nil { + t.Error(err) + return + } + payload.Data = append(payload.Data, payload.Data[0]) + _ = json.NewEncoder(w).Encode(payload) + } + } + const vm = `{"vmid":105,"type":"qemu","node":"member","name":"kept-vm","status":"running","lock":"backup","cpu":0.95,"maxmem":1073741824,"maxdisk":2147483648}` + const extra = `{"vmid":106,"type":"qemu","node":"member","name":"later-deleted-vm","status":"running","lock":"backup","cpu":0.1,"maxmem":1073741824}` + const ct = `{"vmid":205,"type":"lxc","node":"member","name":"kept-container","status":"running","lock":"backup","cpu":0.15,"mem":104857600,"maxmem":1073741824,"disk":268435456,"maxdisk":2147483648}` + const template = `{"vmid":999,"type":"qemu","node":"member","template":1}` + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + p := phase.Load() + switch { + case strings.HasSuffix(r.URL.Path, "/cluster/resources"): + clusterReads.Add(1) + if p == 1 || p == 2 { + corrupt(w, `{"data":[`+vm+`]}`) + return + } + if p == 4 { + http.NotFound(w, r) + return + } + rows := vm + `,` + ct + if p == 0 { + rows += `,` + extra + `,` + template + } + if p == 3 { + rows += `,{"vmid":108,"type":"qemu","node":"member","name":"new-vm","status":"running","lock":"backup","cpu":0,"maxmem":1073741824}` + } + fmt.Fprint(w, `{"data":[`+rows+`]}`) + case strings.HasSuffix(r.URL.Path, "/member/qemu"): + listReads.Add(1) + if p == 1 { + corrupt(w, `{"data":[`+vm+`]}`) + return + } + if p == 4 { + fmt.Fprint(w, `{"data":[]}`) + return + } + fmt.Fprint(w, `{"data":[`+vm+`]}`) + case strings.HasSuffix(r.URL.Path, "/member/lxc"): + listReads.Add(1) + if p == 1 { + corrupt(w, `{"data":[`+ct+`]}`) + return + } + if p == 4 { + fmt.Fprint(w, `{"data":[]}`) + return + } + // Keep the existing older-PVE string identifier compatibility on fallback. + fmt.Fprint(w, `{"data":[`+strings.Replace(ct, `"vmid":205`, `"vmid":"205"`, 1)+`]}`) + case strings.Contains(r.URL.Path, "/agent/"): + commands.Add(1) + t.Error("inventory proof dispatched an in-guest command") + http.Error(w, "unexpected command", 500) + case strings.HasSuffix(r.URL.Path, "/qemu/105/status/current"), strings.HasSuffix(r.URL.Path, "/qemu/106/status/current"), strings.HasSuffix(r.URL.Path, "/qemu/108/status/current"): + fmt.Fprint(w, `{"data":{"status":"running","lock":"backup","maxmem":1073741824,"mem":104857600}}`) + case strings.HasSuffix(r.URL.Path, "/lxc/205/status/current"): + fmt.Fprint(w, `{"data":{"vmid":205,"status":"running","lock":"backup","maxmem":1073741824,"mem":104857600,"maxdisk":2147483648,"disk":268435456}}`) + case strings.HasSuffix(r.URL.Path, "/config"): + fmt.Fprint(w, `{"data":{"lock":"backup"}}`) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + client, err := proxmox.NewClient(proxmox.ClientConfig{Host: server.URL, TokenName: "fixture@pve!pulse", TokenValue: "fixture", Timeout: 10 * time.Second}) + if err != nil { + t.Fatal(err) + } + m := newTestPVEMonitor("inventory") + t.Cleanup(m.alertManager.Stop) + t.Cleanup(m.notificationMgr.Stop) + m.replacePVEClient("inventory", client) + alertCfg := m.alertManager.GetConfig() + alertCfg.Enabled = true + alertCfg.GuestDefaults.CPU = &alerts.HysteresisThreshold{Trigger: 80, Clear: 75} + alertCfg.MetricTimeThresholds = map[string]map[string]int{"guest": {"cpu": 0}} + m.alertManager.UpdateConfig(alertCfg) + run := func() models.StateSnapshot { + t.Helper() + if err := m.pollGuestsWithFallback(context.Background(), "inventory", &config.PVEInstance{MonitorVMs: true, MonitorContainers: true}, client, []proxmox.Node{{Node: "member", Status: "online"}}, map[string]string{"member": "online"}); err != nil { + t.Fatal(err) + } + return m.state.GetSnapshot() + } + // A second registered connection has its own observation and must survive + // both failed enumerations and actual removals on this one. + peer := models.VM{ID: "peer:other:305", Instance: "peer", Node: "other", VMID: 305, Name: "independent-peer", Type: "qemu", Status: "running", LastSeen: time.Now()} + m.state.UpdateVMsForInstance("peer", []models.VM{peer}) + original := run() + if len(original.VMs) != 3 || len(original.Containers) != 1 { + t.Fatalf("initial complete inventory not established: %+v", original) + } + templateKey := pveBackupTemplateSubjectKey("inventory", "qemu", "member", 999) + if _, ok := m.pveBackupTemplateSubjects["inventory"][templateKey]; !ok { + t.Fatal("initial template not established") + } + originalHistory := m.metricsHistory.GetGuestMetrics("inventory:member:105", "cpu", time.Hour) + originalAlerts := m.alertManager.GetActiveAlerts() + if len(originalHistory) != 1 || len(originalAlerts) != 1 { + t.Fatalf("initial history/breach not established: %d/%d", len(originalHistory), len(originalAlerts)) + } + emit := func(label string, snapshot models.StateSnapshot) { + registry := unifiedresources.NewRegistry(nil) + registry.IngestSnapshot(snapshot) + projection, err := json.Marshal(map[string]any{"case": failure, "phase": label, "snapshot": snapshot, "resources": registry.List()}) + if err != nil { + t.Fatal(err) + } + t.Logf("GUEST_INVENTORY_PROJECTION %s", projection) + } + emit("initial", original) + phase.Store(1) + failed := run() + if !reflect.DeepEqual(failed.VMs, original.VMs) || !reflect.DeepEqual(failed.Containers, original.Containers) { + t.Error("incomplete inventory changed retained guests or their observation time") + } + if !reflect.DeepEqual(m.metricsHistory.GetGuestMetrics("inventory:member:105", "cpu", time.Hour), originalHistory) { + t.Error("failed inventory fabricated a new History sample") + } + if !reflect.DeepEqual(m.alertManager.GetActiveAlerts(), originalAlerts) { + t.Error("failed inventory changed an existing breach") + } + if _, ok := m.pveBackupTemplateSubjects["inventory"][templateKey]; !ok { + t.Error("failed inventory erased backup-template eligibility") + } + emit("both-inventories-refused", failed) + phase.Store(2) + fallback := run() + if len(fallback.VMs) != 2 || len(fallback.Containers) != 1 || fallback.VMs[0].VMID == 106 { + t.Error("complete fallback failed to publish a real deletion") + } + if _, ok := m.pveBackupTemplateSubjects["inventory"][templateKey]; ok { + t.Error("complete fallback retained a deleted template") + } + if len(m.metricsHistory.GetGuestMetrics("inventory:member:105", "cpu", time.Hour)) != 2 { + t.Error("complete fallback failed to resume History") + } + emit("complete-node-fallback", fallback) + before := listReads.Load() + phase.Store(3) + recovered := run() + if len(recovered.VMs) != 3 || len(recovered.Containers) != 1 || listReads.Load() != before { + t.Error("complete cluster recovery did not return to the efficient route") + } + emit("complete-cluster-recovery", recovered) + phase.Store(4) + empty := run() + if len(empty.VMs) != 1 || empty.VMs[0].ID != peer.ID || len(empty.Containers) != 0 { + t.Error("explicit empty node inventories failed to remove guests within their scope") + } + if commands.Load() != 0 || clusterReads.Load() != 5 || listReads.Load() != 6 || len(m.pveGuestPolls) != 0 { + t.Errorf("unexpected command/retry/owner leak: %d/%d/%d/%d", commands.Load(), clusterReads.Load(), listReads.Load(), len(m.pveGuestPolls)) + } + emit("complete-scoped-empty", empty) + }) + } +} diff --git a/pkg/proxmox/client.go b/pkg/proxmox/client.go index 9138ac63da..118594c7df 100644 --- a/pkg/proxmox/client.go +++ b/pkg/proxmox/client.go @@ -1272,18 +1272,13 @@ func (c *Client) GetVMs(ctx context.Context, node string) ([]VM, error) { if err != nil { return nil, err } - defer resp.Body.Close() - - var result struct { - Data []VM `json:"data"` - } - - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + values, err := readGuestInventory(resp, func(vm VM) (string, bool) { return guestInventoryID(vm.VMID) }) + if err != nil { return nil, err } - stampVMObservation(result.Data, time.Now().UTC()) - return result.Data, nil + stampVMObservation(values, time.Now().UTC()) + return values, nil } // GetContainers returns all LXC containers on a specific node @@ -1292,18 +1287,13 @@ func (c *Client) GetContainers(ctx context.Context, node string) ([]Container, e if err != nil { return nil, err } - defer resp.Body.Close() - - var result struct { - Data []Container `json:"data"` - } - - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + values, err := readGuestInventory(resp, func(container Container) (string, bool) { return guestInventoryID(int(container.VMID)) }) + if err != nil { return nil, err } - stampContainerObservation(result.Data, time.Now().UTC()) - return result.Data, nil + stampContainerObservation(values, time.Now().UTC()) + return values, nil } // GetStorage returns storage information for a specific node @@ -2580,18 +2570,18 @@ func (c *Client) GetClusterResources(ctx context.Context, resourceType string) ( if err != nil { return nil, err } - defer resp.Body.Close() - - var result struct { - Data []ClusterResource `json:"data"` - } - - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + values, err := readGuestInventory(resp, func(resource ClusterResource) (string, bool) { + if resourceType == "vm" && resource.Type != "qemu" && resource.Type != "lxc" { + return "", false + } + return clusterInventoryID(resource) + }) + if err != nil { return nil, err } - stampClusterResourceObservation(result.Data, time.Now().UTC()) - return result.Data, nil + stampClusterResourceObservation(values, time.Now().UTC()) + return values, nil } // ClusterOptions holds selected Proxmox datacenter configuration options. diff --git a/pkg/proxmox/client_api_more_test.go b/pkg/proxmox/client_api_more_test.go index bf66d6c96f..4613900818 100644 --- a/pkg/proxmox/client_api_more_test.go +++ b/pkg/proxmox/client_api_more_test.go @@ -297,7 +297,7 @@ func TestClientStatusAndResources(t *testing.T) { }) case "/api2/json/cluster/resources": writeJSON(t, w, map[string]interface{}{ - "data": []ClusterResource{{ID: "qemu/100", Type: "vm", VMID: 100, Pool: "prod-vms"}}, + "data": []ClusterResource{{ID: "qemu/100", Type: "qemu", Node: "node1", VMID: 100, Pool: "prod-vms"}}, }) case "/api2/json/nodes/node1/lxc/101/interfaces": writeJSON(t, w, map[string]interface{}{ diff --git a/pkg/proxmox/cluster_client_additional_api_test.go b/pkg/proxmox/cluster_client_additional_api_test.go index 0c6d65ccff..3a30489b77 100644 --- a/pkg/proxmox/cluster_client_additional_api_test.go +++ b/pkg/proxmox/cluster_client_additional_api_test.go @@ -275,7 +275,7 @@ func TestClusterClient_GetClusterResources(t *testing.T) { return } if strings.HasPrefix(r.URL.Path, "/api2/json/cluster/resources") { - fmt.Fprint(w, `{"data":[{"id":"qemu/100","type":"qemu","node":"node1"}]}`) + fmt.Fprint(w, `{"data":[{"id":"qemu/100","type":"qemu","node":"node1","vmid":100}]}`) return } w.WriteHeader(http.StatusNotFound) @@ -302,7 +302,7 @@ func TestClusterClient_GetClusterResources_AllTypes(t *testing.T) { return } if r.URL.Path == "/api2/json/cluster/resources" { - fmt.Fprint(w, `{"data":[{"id":"lxc/200","type":"lxc","node":"node1"}]}`) + fmt.Fprint(w, `{"data":[{"id":"lxc/200","type":"lxc","node":"node1","vmid":200}]}`) return } w.WriteHeader(http.StatusNotFound) diff --git a/pkg/proxmox/guest_inventory_response.go b/pkg/proxmox/guest_inventory_response.go new file mode 100644 index 0000000000..87c92dd9d4 --- /dev/null +++ b/pkg/proxmox/guest_inventory_response.go @@ -0,0 +1,85 @@ +package proxmox + +import ( + "bytes" + "encoding/json" + "errors" + "net/http" + "strconv" + "strings" + "unicode/utf8" +) + +// Inventories own removal. A malformed success is neither a complete empty +// enumeration nor an authentication refusal; keep its body out of diagnostics. +var errInvalidGuestInventoryResponse = errors.New("invalid Proxmox guest inventory response") + +// The three enumerations share identity, telemetry and metadata fields. Reject +// case-insensitive struct overwrites without excluding unique provider metadata +// or changing the existing endpoint-specific decoders (notably LXC FlexInt). +var guestInventoryFields = []string{ + "id", "type", "node", "pool", "status", "name", "vmid", "cpu", "cpus", "maxcpu", "mem", "maxmem", "swap", "maxswap", "disk", "maxdisk", "netin", "netout", "diskread", "diskwrite", "uptime", "template", "tags", "lock", "agent", "hostname", "ip", "ip6", "ipv4", "ipv6", "network", "diskinfo", "rootfs", +} + +func readGuestInventory[T any](resp *http.Response, identity func(T) (string, bool)) ([]T, error) { + body, readErr := readResponseBodyLimited(resp.Body) + closeErr := resp.Body.Close() + if readErr != nil || closeErr != nil { + return nil, errors.Join(readErr, closeErr) + } + if resp.StatusCode != http.StatusOK || !utf8.Valid(body) { + return nil, errInvalidGuestInventoryResponse + } + envelope, valid := inventoryResponseObject(body, "data", "errors") + if !valid { + return nil, errInvalidGuestInventoryResponse + } + if failure, exists := envelope["errors"]; exists && !bytes.Equal(bytes.TrimSpace(failure), []byte("null")) { + return nil, errInvalidGuestInventoryResponse + } + data := bytes.TrimSpace(envelope["data"]) + // Preserve a scoped empty array, never a missing/null inventory. Read and + // close the whole bounded HTTP body before publishing even a valid prefix. + if len(data) == 0 || data[0] != '[' { + return nil, errInvalidGuestInventoryResponse + } + var rows []json.RawMessage + if json.Unmarshal(data, &rows) != nil { + return nil, errInvalidGuestInventoryResponse + } + values := make([]T, 0, len(rows)) + seen := make(map[string]bool, len(rows)) + for _, row := range rows { + if _, valid := inventoryResponseObject(row, guestInventoryFields...); !valid { + return nil, errInvalidGuestInventoryResponse + } + var value T + if json.Unmarshal(row, &value) != nil { + return nil, errInvalidGuestInventoryResponse + } + key, valid := identity(value) + if !valid || seen[key] { + return nil, errInvalidGuestInventoryResponse + } + seen[key] = true + values = append(values, value) + } + return values, nil +} + +func guestInventoryID(vmid int) (string, bool) { + return strconv.Itoa(vmid), vmid > 0 +} + +func clusterInventoryID(resource ClusterResource) (string, bool) { + switch resource.Type { + case "qemu", "lxc": + key, valid := guestInventoryID(resource.VMID) + // VMIDs are cluster-wide, including across a migration or guest kind. + return "guest/" + key, valid && strings.TrimSpace(resource.Node) != "" + default: + // Unfiltered calls may include nodes, storage, pools and future provider + // resource kinds. They retain their provider identity and optional fields. + return resource.ID, strings.TrimSpace(resource.Type) != "" && strings.TrimSpace(resource.ID) != "" + } +} diff --git a/pkg/proxmox/guest_inventory_response_test.go b/pkg/proxmox/guest_inventory_response_test.go new file mode 100644 index 0000000000..d8f9cb47d4 --- /dev/null +++ b/pkg/proxmox/guest_inventory_response_test.go @@ -0,0 +1,102 @@ +package proxmox + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "sync/atomic" + "testing" + "time" +) + +// Enumerations drive removal, backup-template policy and History. Exercise the +// real HTTP readers on both production routes, with no QGA commands or retries. +func TestGuestInventoryResponseAdmission(t *testing.T) { + for _, kind := range []string{"cluster", "qemu", "lxc"} { + t.Run(kind, func(t *testing.T) { + row := `{"vmid":105,"type":"qemu","node":"member","name":"fixture","status":"running"}` + inventory := `{"data":[` + row + `]}` + cases := []struct { + name, body string + incomplete bool + }{ + {"missing-data", `{}`, false}, + {"null-data", `{"data":null}`, false}, + {"null-envelope", `null`, false}, + {"conflicting-error", `{"data":[],"errors":{"message":"fixture-private-detail"}}`, false}, + {"case-data", `{"Data":[]}`, false}, + {"duplicate-data", `{"data":[` + row + `],"data":[]}`, false}, + {"escaped-duplicate-data", `{"data":[],"\u0064ata":[` + row + `]}`, false}, + {"case-overwrite", `{"data":[` + row + `],"Data":[]}`, false}, + {"trailing-value", inventory + ` {"data":[]}`, false}, + {"trailing-garbage", inventory + `fixture-private-detail`, false}, + {"truncated-after-prefix", inventory, true}, + {"oversized-after-prefix", inventory + strings.Repeat(" ", int(maxResponseBodyBytes)), false}, + {"invalid-utf8", `{"data":[` + row + `],"metadata":"` + string([]byte{0xff}) + `"}`, false}, + {"null-row", `{"data":[` + row + `,null]}`, false}, + {"missing-guest-id", `{"data":[{"type":"qemu","node":"member"}]}`, false}, + {"zero-guest-id", `{"data":[{"vmid":0,"type":"qemu","node":"member"}]}`, false}, + {"negative-guest-id", `{"data":[{"vmid":-105,"type":"qemu","node":"member"}]}`, false}, + {"duplicate-field", `{"data":[{"vmid":106,"vmid":105,"type":"qemu","node":"member"}]}`, false}, + {"case-id", `{"data":[{"vmid":105,"VMID":106,"type":"qemu","node":"member"}]}`, false}, + {"duplicate-row", `{"data":[` + row + `,` + row + `]}`, false}, + {"bad-row-after-good", `{"data":[` + row + `,{"vmid":106,"type":"qemu","node":"member","mem":"not-a-number"}]}`, false}, + } + if kind == "cluster" { + cases = append(cases, struct { + name, body string + incomplete bool + }{"missing-node", `{"data":[{"vmid":105,"type":"qemu"}]}`, false}, + struct { + name, body string + incomplete bool + }{"missing-type", `{"data":[{"vmid":105,"node":"member"}]}`, false}, + struct { + name, body string + incomplete bool + }{"duplicate-migrated-id", `{"data":[` + row + `,{"vmid":105,"type":"qemu","node":"other"}]}`, false}) + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + var requests atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + if tc.incomplete { + w.Header().Set("Content-Length", fmt.Sprint(len(tc.body)+100)) + } else { + w.(http.Flusher).Flush() + } + fmt.Fprint(w, tc.body) + })) + defer server.Close() + c, err := NewClient(ClientConfig{Host: server.URL, TokenName: "fixture@pve!pulse", TokenValue: "fixture", Timeout: 15 * time.Second}) + if err != nil { + t.Fatal(err) + } + defer c.httpClient.CloseIdleConnections() + var values any + switch kind { + case "cluster": + values, err = c.GetClusterResources(context.Background(), "vm") + case "qemu": + values, err = c.GetVMs(context.Background(), "member") + case "lxc": + values, err = c.GetContainers(context.Background(), "member") + } + if err == nil || !reflect.ValueOf(values).IsNil() { + t.Errorf("unverified %s inventory admitted: rows=%+v err=%v", kind, values, err) + } + if err != nil && strings.Contains(err.Error(), "fixture-private-detail") { + t.Error("private response body became an inventory diagnostic") + } + if requests.Load() != 1 { + t.Errorf("enumeration was replayed: %d requests", requests.Load()) + } + }) + } + }) + } +} diff --git a/pkg/proxmox/io_counters_test.go b/pkg/proxmox/io_counters_test.go index c914db7a61..e43487dbe2 100644 --- a/pkg/proxmox/io_counters_test.go +++ b/pkg/proxmox/io_counters_test.go @@ -253,3 +253,67 @@ func TestVMStatusMemoryAvailabilityPresence(t *testing.T) { t.Run("invalid-numbers", testVMMemInfoAvailabilityRejectsInvalidNumbers) t.Run("reused-decode", testVMMemInfoAvailabilityPresenceDoesNotLeakAcrossDecode) } + +// Complete enumeration still owns the original counter-presence and receipt- +// time semantics; scoped-empty and older LXC encodings remain compatible. +func TestGuestInventoryAcceptedScopedAndExtendedResponses(t *testing.T) { + for _, kind := range []string{"cluster", "qemu", "lxc"} { + for _, body := range []string{ + `{"data":[]}`, `{"data":[],"errors":null,"total":0}`, + `{"data":[{"vmid":105,"type":"qemu","node":"member","name":"fixture","status":"stopped","pool":"department","cpu":0,"maxcpu":4,"cpus":4,"mem":0,"maxmem":1024,"disk":0,"maxdisk":4096,"diskread":0,"netin":null,"uptime":0,"template":1,"tags":"ops","lock":"backup","agent":1,"vendor-metadata":{"version":1}}],"digest":"fixture-metadata"}`, + } { + t.Run(kind+"/"+body, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { fmt.Fprint(w, body) })) + defer server.Close() + c := backupTestClient(t, server.URL) + defer c.httpClient.CloseIdleConnections() + switch kind { + case "cluster": + got, err := c.GetClusterResources(context.Background(), "vm") + if err != nil { + t.Fatal(err) + } + if strings.Contains(body, `"vmid"`) && (len(got) != 1 || got[0].VMID != 105 || got[0].Node != "member" || got[0].MaxCPU != 4 || got[0].Pool != "department" || got[0].Lock != "backup" || got[0].Template != 1 || got[0].IOCounters.DiskRead != true || got[0].IOCounters.NetworkIn || got[0].ObservedAt.IsZero()) { + t.Fatalf("supported cluster fields changed: %+v", got) + } + case "qemu": + got, err := c.GetVMs(context.Background(), "member") + if err != nil { + t.Fatal(err) + } + if strings.Contains(body, `"vmid"`) && (len(got) != 1 || got[0].VMID != 105 || got[0].CPUs != 4 || got[0].Agent != 1 || got[0].Lock != "backup" || got[0].Pool != "department" || got[0].IOCounters.NetworkIn || got[0].ObservedAt.IsZero()) { + t.Fatalf("supported VM fields changed: %+v", got) + } + case "lxc": + got, err := c.GetContainers(context.Background(), "member") + if err != nil { + t.Fatal(err) + } + if strings.Contains(body, `"vmid"`) && (len(got) != 1 || int(got[0].VMID) != 105 || int(got[0].CPUs) != 4 || got[0].Pool != "department" || got[0].IOCounters.NetworkIn || got[0].ObservedAt.IsZero()) { + t.Fatalf("supported container fields changed: %+v", got) + } + } + }) + } + } + t.Run("lxc-string-identifiers", func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + fmt.Fprint(w, `{"data":[{"vmid":"205","cpus":"2","name":"container"}]}`) + })) + defer server.Close() + got, err := backupTestClient(t, server.URL).GetContainers(context.Background(), "member") + if err != nil || len(got) != 1 || int(got[0].VMID) != 205 || int(got[0].CPUs) != 2 { + t.Fatalf("string LXC identifiers lost: %+v %v", got, err) + } + }) + t.Run("all-resource-types", func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + fmt.Fprint(w, `{"data":[{"id":"node/member","type":"node","node":"member"},{"id":"storage/member/local","type":"storage","node":"member"},{"id":"pool/department","type":"pool"},{"id":"future/provider","type":"future-type"}]}`) + })) + defer server.Close() + got, err := backupTestClient(t, server.URL).GetClusterResources(context.Background(), "") + if err != nil || len(got) != 4 { + t.Fatalf("non-guest resource compatibility lost: %+v %v", got, err) + } + }) +} diff --git a/pkg/proxmox/node_inventory_response.go b/pkg/proxmox/node_inventory_response.go index a8b05b3b43..f9a90c164f 100644 --- a/pkg/proxmox/node_inventory_response.go +++ b/pkg/proxmox/node_inventory_response.go @@ -16,7 +16,7 @@ func decodeNodeInventory(body []byte) ([]Node, error) { if !utf8.Valid(body) { return nil, errInvalidNodeInventoryResponse } - envelope, valid := nodeInventoryObject(body, "data", "errors") + envelope, valid := inventoryResponseObject(body, "data", "errors") if !valid { return nil, errInvalidNodeInventoryResponse } @@ -36,7 +36,7 @@ func decodeNodeInventory(body []byte) ([]Node, error) { nodes := make([]Node, 0, len(rows)) seen := make(map[string]bool, len(rows)) for _, row := range rows { - fields, valid := nodeInventoryObject(row, "node", "status", "cpu", "maxcpu", "mem", "maxmem", "disk", "maxdisk", "uptime", "level") + fields, valid := inventoryResponseObject(row, "node", "status", "cpu", "maxcpu", "mem", "maxmem", "disk", "maxdisk", "uptime", "level") if !valid { return nil, errInvalidNodeInventoryResponse } @@ -57,7 +57,7 @@ func decodeNodeInventory(body []byte) ([]Node, error) { // The existing object reader rejects duplicates and extra JSON values. Keep // unique provider metadata, but never let encoding/json's case-insensitive // struct matching replace canonical inventory fields with another spelling. -func nodeInventoryObject(body []byte, canonical ...string) (map[string]json.RawMessage, bool) { +func inventoryResponseObject(body []byte, canonical ...string) (map[string]json.RawMessage, bool) { fields, valid := guestAgentResponseFields(body) if !valid { return nil, false