From cc72ea76e34c976313bca4c540bffd4ab4e37f35 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sun, 11 Oct 2026 20:43:26 +0100 Subject: [PATCH] Expose caller-owned cluster endpoint edits bypassing admission Discriminate live routing, first-use TLS trust and shared guest-command coordination after the constructor input is reused. Use only synthetic loopback services; no native guest workload or reporter recovery is claimed. Change-source: pulse-maintainer --- pkg/proxmox/cluster_client_api_test.go | 6 + .../cluster_client_endpoint_ownership_test.go | 167 ++++++++++++++++++ 2 files changed, 173 insertions(+) create mode 100644 pkg/proxmox/cluster_client_endpoint_ownership_test.go diff --git a/pkg/proxmox/cluster_client_api_test.go b/pkg/proxmox/cluster_client_api_test.go index 7663c5790a..7b6b13a1db 100644 --- a/pkg/proxmox/cluster_client_api_test.go +++ b/pkg/proxmox/cluster_client_api_test.go @@ -635,6 +635,12 @@ func TestClusterClientFingerprintTrustContract(t *testing.T) { testClusterClientFingerprintTrust(t) } +func TestClusterClientEndpointOwnershipContract(t *testing.T) { + t.Run("routing-and-first-use-trust", testClusterClientEndpointOwnershipRoutingAndTrust) + t.Run("shared-guest-admission", testClusterClientEndpointOwnershipGuestAdmission) + t.Run("input-shape", testClusterClientEndpointOwnershipShape) +} + // Guest admission equivalence must not broaden the existing certificate trust scope. func TestGuestAgentIPv4MappedTrustScope(t *testing.T) { const ipv4 = "https://192.0.2.10:8006/tenant" diff --git a/pkg/proxmox/cluster_client_endpoint_ownership_test.go b/pkg/proxmox/cluster_client_endpoint_ownership_test.go new file mode 100644 index 0000000000..b35925cdee --- /dev/null +++ b/pkg/proxmox/cluster_client_endpoint_ownership_test.go @@ -0,0 +1,167 @@ +package proxmox + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "slices" + "strings" + "sync/atomic" + "testing" + "time" +) + +func testClusterClientEndpointOwnershipRoutingAndTrust(t *testing.T) { + declared := newClusterTrustFixture(t) + other := newClusterTrustFixture(t) + declaredPin := clusterTrustPin(declared.active.Load()) + input := []string{declared.server.URL} + cc := NewClusterClient("owned-endpoints", clusterTrustConfig("https://primary.trust.invalid", declaredPin), input, nil) + + // Reusing the caller's configuration storage must not declare a new member, + // change the live failover target, or permit first-use trust for that target. + input[0] = other.server.URL + cc.mu.RLock() + mayTrustOther := cc.canTrustMemberOnFirstUseLocked(other.server.URL) + cc.mu.RUnlock() + if mayTrustOther { + t.Error("caller-owned endpoint edit declared a new first-use trust target") + } + before := declared.requests.Load() + if nodes, err := cc.GetNodes(context.Background()); err != nil || len(nodes) != 1 || declared.requests.Load() != before+1 { + t.Errorf("caller-owned endpoint edit interrupted ordinary pinned monitoring: %v", err) + } + + // Use a real local verifier mismatch, not provider text. Rejected admission + // must do no additional capture handshake, authenticated request or pin write. + lastErr := clusterTrustMismatch(t, other, declaredPin) + handshakes := other.handshakes.Load() + client, err, retried := cc.refreshTOFUFingerprintAndRetry(context.Background(), other.server.URL, time.Second, lastErr) + if client != nil { + defer client.httpClient.CloseIdleConnections() + } + if client != nil || retried || !errors.Is(err, lastErr) || other.handshakes.Load() != handshakes || other.requests.Load() != 0 { + t.Error("caller-owned endpoint edit admitted capture or authenticated first-use retry") + } + cc.mu.RLock() + _, captured := cc.endpointFingerprints[other.server.URL] + cc.mu.RUnlock() + if captured || cc.getEndpointFingerprint(declared.server.URL) != declaredPin { + t.Error("caller-owned endpoint edit changed the client's trust state") + } + t.Logf("original pinned monitoring preserved; injected endpoint authenticated requests=%d", other.requests.Load()) +} + +func testClusterClientEndpointOwnershipGuestAdmission(t *testing.T) { + isolateGuestAgentCapacity(t) + entered, finish := make(chan struct{}), make(chan struct{}) + var firstCommand atomic.Bool + var originalCommands, otherRequests atomic.Int32 + handler := func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "PVEAPIToken=fixture@pve!pulse=fixture" { + t.Error("fixture authentication changed") + } + switch { + case r.URL.Path == "/api2/json/nodes": + fmt.Fprint(w, `{"data":[{"node":"node","status":"online"}]}`) + case strings.HasSuffix(r.URL.Path, "/config"): + fmt.Fprint(w, `{"data":{}}`) + default: + originalCommands.Add(1) + if strings.HasSuffix(r.URL.Path, "/qemu/105/agent/get-fsinfo") && firstCommand.CompareAndSwap(false, true) { + fmt.Fprint(w, `{"data":{"result":`) + w.(http.Flusher).Flush() + close(entered) + <-finish + fmt.Fprint(w, `[]}}`) + return + } + backupAgentPayload(w, r) + } + } + primary := httptest.NewServer(http.HandlerFunc(handler)) + defer primary.Close() + member := httptest.NewServer(http.HandlerFunc(handler)) + defer member.Close() + other := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + otherRequests.Add(1) + if r.URL.Path == "/api2/json/nodes" { + fmt.Fprint(w, `{"data":[{"node":"other","status":"online"}]}`) + return + } + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } + backupAgentPayload(w, r) + })) + defer other.Close() + input := []string{member.URL} + cc := NewClusterClient("owned-guest-targets", ClientConfig{Host: primary.URL, TokenName: "fixture@pve!pulse", TokenValue: "fixture", Timeout: 5 * time.Second}, input, nil) + standalone := backupTestClient(t, primary.URL) + standalone.httpClient.Timeout = 5 * time.Second + done := make(chan error, 1) + go func() { _, err := standalone.GetVMFSInfo(context.Background(), "node", 105); done <- err }() + joined := false + complete := func() { + if !joined { + close(finish) + joined = true + if err := <-done; err != nil { + t.Errorf("original single guest command failed: %v", err) + } + } + } + defer complete() + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("original command did not reach the partial-body barrier") + } + input[0] = other.URL + reads := map[string]func() error{ + "filesystem": func() error { _, err := cc.GetVMFSInfo(context.Background(), "node", 105); return err }, + "os": func() error { _, err := cc.GetVMAgentInfo(context.Background(), "node", 105); return err }, + "version": func() error { _, err := cc.GetVMAgentVersion(context.Background(), "node", 105); return err }, + "network": func() error { _, err := cc.GetVMNetworkInterfaces(context.Background(), "node", 105); return err }, + "meminfo": func() error { + _, err := cc.GetVMMemoryAvailabilityFromAgent(context.Background(), "node", 105) + return err + }, + "legacy-memory": func() error { _, err := cc.GetVMMemAvailableFromAgent(context.Background(), "node", 105); return err }, + } + for name, read := range reads { + if err := read(); GuestAgentDeferredReason(err) != "agent-busy" { + t.Errorf("%s bypassed the originally registered guest channel: %v", name, err) + } + } + if originalCommands.Load() != 1 || otherRequests.Load() != 0 { + t.Errorf("caller-owned endpoint edit dispatched outside shared admission: original commands=%d, other requests=%d", originalCommands.Load(), otherRequests.Load()) + } + if _, err := cc.GetVMAgentInfo(context.Background(), "node", 106); err != nil { + t.Errorf("independent guest lost monitoring: %v", err) + } + complete() + if _, err := cc.GetVMFSInfo(context.Background(), "node", 105); err != nil { + t.Errorf("completed guest did not resume through the declared member: %v", err) + } + if originalCommands.Load() != 3 || otherRequests.Load() != 0 { + t.Errorf("recovery left declared targets: original commands=%d, other requests=%d", originalCommands.Load(), otherRequests.Load()) + } + t.Logf("six busy reads fenced; independent guest and completed guest resume; other requests=%d", otherRequests.Load()) +} + +func testClusterClientEndpointOwnershipShape(t *testing.T) { + for _, input := range [][]string{nil, {}, {"invalid://first", "invalid://second", "invalid://first"}} { + want := slices.Clone(input) + cc := NewClusterClient("owned-shape", ClientConfig{Host: "https://primary.shape.invalid", TokenName: "fixture@pve!pulse", TokenValue: "fixture"}, input, nil) + for i := range input { + input[i] = "invalid://changed" + } + if !slices.Equal(cc.endpoints, want) || (cc.endpoints == nil) != (want == nil) { + t.Error("endpoint snapshot changed order, duplicates, nil/empty shape or ownership") + } + } +}