diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 4ef2ff262c..4394455bd1 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -112,6 +112,25 @@ replay through another cluster endpoint. Coordination retains at most 4,096 active/cooling endpoint-guest entries and expires cooldowns. Explicit API permission refusals and unsupported commands remain errors, not successes. +Guest commands also prevent transport-level replay: each uses a fresh single-use +HTTP/1 connection, with no pooled-connection or HTTP/2-stream retries and no +redirect following. The client preserves its configured TLS verification or +fingerprint, proxy, DNS dialer, authentication and request bounds. An opaque +transport or custom TLS dialer defers without dispatch rather than substituting +a different trust policy. A fully received redirect discards its body and +starts the same per-VM uncertainty cooldown; its destination is never contacted. +Ordinary config/resource reads retain pooling, HTTP/2, redirects and session +recovery. This deliberately trades one new connection per guest command for +single-attempt safety; it does not reserve the QGA channel or establish thaw. +Wire controls drop a reply only after receiving a command (including when a +backup lock then appears), and exercise every read and redirect code. TLS +controls retain CA/fingerprint trust, reject untrusted peers and verify command +HTTP/1 isolation alongside ordinary HTTP/2. The connected monitoring controls +retain old cache timestamps and History while live CPU continues through lost +reply/redirect deferral. `TestGuestAgentTransportDeferralPreservesLiveCounterReceipts` +also pins independent current CPU/I/O, explicit zero/null/missing counter presence +and receipt timestamps without clearing guest-command uncertainty. + Deferrals retain the original metadata/memory cache timestamps and supported-OS failure evidence. Locked guests retain labelled last-known disks/identity and previous-snapshot memory, or unavailable memory without trustworthy evidence. diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index 30ab89b6c5..3971067d22 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -2974,3 +2974,19 @@ func TestBroadcastProjectionMatchesPreviousPipeline(t *testing.T) { func TestGuestAgentBackupMonitoringContract(t *testing.T) { testGuestAgentBackupMonitoringLifecycle(t) } + +func TestGuestAgentTransportMonitoringContract(t *testing.T) { + for _, reason := range []string{"agent-redirect", "agent-transport-unverified"} { + for _, prefix := range []string{"", "prev-"} { + if !guestAgentDiskDeferred(prefix + reason) { + t.Errorf("transport uncertainty %q is not labelled deferred", prefix+reason) + } + } + } + for _, reason := range []string{"", "permission-denied", "agent-not-running", "agent-error"} { + if guestAgentDiskDeferred(reason) { + t.Errorf("completed/ordinary error %q became transport uncertainty", reason) + } + } + testGuestAgentTransportDeferralKeepsLastKnownHistory(t) +} diff --git a/internal/monitoring/guest_agent_transport_integration_test.go b/internal/monitoring/guest_agent_transport_integration_test.go new file mode 100644 index 0000000000..a649a07c6f --- /dev/null +++ b/internal/monitoring/guest_agent_transport_integration_test.go @@ -0,0 +1,122 @@ +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/config" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" +) + +func testGuestAgentTransportDeferralKeepsLastKnownHistory(t *testing.T) { + for _, kind := range []string{"lost reply", "redirect"} { + t.Run(kind, func(t *testing.T) { + const mib = uint64(1024 * 1024) + var phase, calls, redirected atomic.Int32 + var lost atomic.Bool + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "/config"): + fmt.Fprint(w, `{"data":{}}`) + case strings.HasSuffix(r.URL.Path, "/status/current"): + fmt.Fprintf(w, `{"data":{"status":"running","agent":1,"maxmem":%d,"mem":%d}}`, 8*mib, 8*mib) + case strings.Contains(r.URL.Path, "/agent/"): + calls.Add(1) + if r.URL.Query().Get("redirect") == "1" { + redirected.Add(1) + } + if phase.Load() == 1 && lost.CompareAndSwap(false, true) { + if kind == "redirect" { + http.Redirect(w, r, r.URL.Path+"?redirect=1", http.StatusTemporaryRedirect) + return + } + conn, _, err := w.(http.Hijacker).Hijack() + if err != nil { + t.Error(err) + return + } + conn.Close() + return + } + switch { + case strings.HasSuffix(r.URL.Path, "file-read"): + available := 5120 + if phase.Load() == 1 { + available = 4096 + } + json.NewEncoder(w).Encode(map[string]any{"data": map[string]any{"content": fmt.Sprintf("MemTotal: 8192 kB\nMemAvailable: %d kB\n", available)}}) + case strings.HasSuffix(r.URL.Path, "get-fsinfo"): + fmt.Fprintf(w, `{"data":{"result":[{"mountpoint":"/","type":"ext4","total-bytes":%d,"used-bytes":%d}]}}`, 1000*mib, 300*mib) + case strings.HasSuffix(r.URL.Path, "network-get-interfaces"): + fmt.Fprint(w, `{"data":{"result":[{"name":"eth0","ip-addresses":[{"ip-address":"192.0.2.10","ip-address-type":"ipv4","prefix":24}]}]}}`) + case strings.HasSuffix(r.URL.Path, "get-osinfo"): + fmt.Fprint(w, `{"data":{"name":"Linux","version":"fixture"}}`) + default: + fmt.Fprint(w, `{"data":{"result":{"version":"1.0"}}}`) + } + default: + http.NotFound(w, r) + } + })) + defer server.Close() + client, err := proxmox.NewClient(proxmox.ClientConfig{Host: server.URL, TokenName: "fixture@pve!pulse", TokenValue: "fixture", Timeout: time.Second}) + if err != nil { + t.Fatal(err) + } + m := &Monitor{config: &config.Config{}, rateTracker: NewRateTracker(), metricsHistory: NewMetricsHistory(32, time.Hour), guestMetadataLimiter: make(map[string]time.Time)} + res := proxmox.ClusterResource{Type: "qemu", Node: "node", Name: "guest", VMID: 105, Status: "running", MaxMem: 8 * mib, Mem: 8 * mib, MaxDisk: 1000 * mib, CPU: 0.1} + identity := "transport:node:105" + var previous *models.VM + build := func() (models.VM, string) { + vm, raw, source, notes, at, ok := m.buildVMFromClusterResource(context.Background(), "transport", res, client, identity, nil, previous) + if !ok { + t.Fatal("VM disappeared") + } + m.recordGuestSnapshot("transport", "qemu", "node", 105, GuestMemorySnapshot{Name: vm.Name, Status: vm.Status, RetrievedAt: at, MemorySource: source, Memory: vm.Memory, Raw: raw, Notes: notes}) + m.recordGuestMetrics([]models.VM{vm}, nil, time.Now().Add(-time.Second)) + previous = &vm + return vm, source + } + initial, source := build() + if source != "guest-agent-meminfo" || initial.Memory.Used != int64(3*mib) || initial.Disk.Used != int64(300*mib) || calls.Load() != 5 { + t.Fatal("healthy baseline absent") + } + memKey, metadataKey := guestMemoryCacheKey("transport", "node", 105), guestMetadataCacheKey("transport", "node", 105) + memory := m.vmAgentMemCache[memKey] + memory.fetchedAt = time.Now().Add(-2 * vmAgentMemCacheTTL) + m.vmAgentMemCache[memKey] = memory + metadata := m.guestMetadataCache[metadataKey] + metadata.fetchedAt = time.Now().Add(-2 * guestMetadataCacheTTL) + m.guestMetadataCache[metadataKey] = metadata + m.guestMetadataLimiter = make(map[string]time.Time) + phase.Store(1) + res.CPU = 0.2 + for i := 0; i < 2; i++ { + deferred, source := build() + if deferred.ID != initial.ID || deferred.GuestAgentStatus != "deferred" || deferred.DiskStatusReason != "prev-agent-cooldown" || deferred.Memory.Used != initial.Memory.Used || source != "previous-snapshot" || deferred.Disk.Used != initial.Disk.Used || !reflect.DeepEqual(deferred.NetworkInterfaces, initial.NetworkInterfaces) { + t.Fatalf("uncertain command lost truthful continuity: source=%s vm=%+v", source, deferred) + } + if !reflect.DeepEqual(m.vmAgentMemCache[memKey], memory) || !reflect.DeepEqual(m.guestMetadataCache[metadataKey], metadata) { + t.Fatal("uncertain command renewed or replaced cached evidence") + } + } + if calls.Load() != 6 || redirected.Load() != 0 { + t.Fatalf("commands/redirected after uncertainty = %d/%d, want 6/0", calls.Load(), redirected.Load()) + } + for metric, want := range map[string]int{"cpu": 3, "memory": 1, "memoryused": 1, "disk": 1} { + if got := len(m.metricsHistory.GetGuestMetrics(identity, metric, time.Hour)); got != want { + t.Errorf("%s History points = %d, want %d", metric, got, want) + } + } + }) + } +} diff --git a/internal/monitoring/monitor_pve_guest_builders.go b/internal/monitoring/monitor_pve_guest_builders.go index 3f1423fa2c..30081fbb2d 100644 --- a/internal/monitoring/monitor_pve_guest_builders.go +++ b/internal/monitoring/monitor_pve_guest_builders.go @@ -891,7 +891,7 @@ func (m *Monitor) updateVMDisksFromGuestAgentFSInfo( func guestAgentDiskDeferred(reason string) bool { switch strings.TrimPrefix(reason, "prev-") { - case "vm-locked", "lock-unverified", "agent-busy", "agent-cooldown", "agent-capacity", "agent-timeout", "agent-response-incomplete": + case "vm-locked", "lock-unverified", "agent-busy", "agent-cooldown", "agent-capacity", "agent-timeout", "agent-response-incomplete", "agent-redirect", "agent-transport-unverified": return true } return false diff --git a/pkg/proxmox/client.go b/pkg/proxmox/client.go index 12750b34e0..483226bc05 100644 --- a/pkg/proxmox/client.go +++ b/pkg/proxmox/client.go @@ -523,6 +523,10 @@ func (c *Client) request(ctx context.Context, method, path string, data url.Valu } func (c *Client) requestWithRetry(ctx context.Context, method, path string, data url.Values, retriedAfter401 bool) (*http.Response, error) { + return c.requestWithRetryUsingClient(ctx, method, path, data, retriedAfter401, c.httpClient) +} + +func (c *Client) requestWithRetryUsingClient(ctx context.Context, method, path string, data url.Values, retriedAfter401 bool, httpClient *http.Client) (*http.Response, error) { // Re-authenticate if needed if c.config.Password != "" && c.auth.tokenName == "" && time.Now().After(c.auth.expiresAt) { if err := c.authenticate(ctx); err != nil { @@ -570,7 +574,7 @@ func (c *Client) requestWithRetry(ctx context.Context, method, path string, data } } - resp, err := c.httpClient.Do(req) + resp, err := httpClient.Do(req) if err != nil { return nil, err } @@ -589,7 +593,7 @@ func (c *Client) requestWithRetry(ctx context.Context, method, path string, data if err := c.authenticate(ctx); err != nil { return nil, fmt.Errorf("re-authentication failed after 401: %w", err) } - return c.requestWithRetry(ctx, method, path, data, true) + return c.requestWithRetryUsingClient(ctx, method, path, data, true, httpClient) } // Create base error with helpful guidance for common issues diff --git a/pkg/proxmox/guest_agent_guard.go b/pkg/proxmox/guest_agent_guard.go index fce6be56b2..80e3268b72 100644 --- a/pkg/proxmox/guest_agent_guard.go +++ b/pkg/proxmox/guest_agent_guard.go @@ -168,13 +168,18 @@ func (c *Client) getGuestAgent(ctx context.Context, path, node string, vmid int) } uncertain := false defer func() { release(uncertain) }() + commandClient, err := guestAgentCommandClient(c.httpClient) + if err != nil { + return nil, err + } + defer commandClient.CloseIdleConnections() if err := c.verifyGuestAgentUnlocked(ctx, node, vmid); err != nil { return nil, err } // A password-session 401 must not replay this command after re-authentication: // the operation lock could change between attempts. Ordinary API reads still // own their usual session recovery; this guest admission is single-attempt. - resp, err := c.requestWithRetry(ctx, http.MethodGet, path, nil, true) + resp, err := c.requestWithRetryUsingClient(ctx, http.MethodGet, path, nil, true, commandClient) if err != nil { var incomplete *responseBodyReadError if errors.As(err, &incomplete) { @@ -198,6 +203,12 @@ func (c *Client) getGuestAgent(ctx context.Context, path, node string, vmid int) uncertain = true return nil, &guestAgentDeferredError{reason: "agent-response-incomplete", cause: errors.Join(readErr, closeErr)} } + if resp.StatusCode >= 300 && resp.StatusCode < 400 { + // The original endpoint may have consumed the command. Neither its + // completion nor another endpoint's operation lock is established. + uncertain = true + return nil, &guestAgentDeferredError{reason: "agent-redirect"} + } // A backup may have started while the command was in flight. Do not publish // its payload as fresh telemetry if lock clearance cannot still be verified. if err := c.verifyGuestAgentUnlocked(ctx, node, vmid); err != nil { diff --git a/pkg/proxmox/guest_agent_transport.go b/pkg/proxmox/guest_agent_transport.go new file mode 100644 index 0000000000..8f61a78fd6 --- /dev/null +++ b/pkg/proxmox/guest_agent_transport.go @@ -0,0 +1,40 @@ +package proxmox + +import "net/http" + +// Guest-agent GETs enqueue serial QGA work and are not replayable resource +// reads. A pooled net/http Transport can transparently resubmit a GET after +// losing a reused connection's reply, independently of application retries. +// HTTP/2 can also retry streams. Use a fresh, single-use HTTP/1 connection and +// refuse redirects, while preserving this client's TLS, proxy, dial and timeout +// policy. Ordinary API reads keep their existing pooled client and recovery. +func guestAgentCommandClient(client *http.Client) (*http.Client, error) { + if client == nil { + return nil, &guestAgentDeferredError{reason: "agent-transport-unverified"} + } + roundTripper := client.Transport + if roundTripper == nil { + roundTripper = http.DefaultTransport + } + base, ok := roundTripper.(*http.Transport) + if !ok || base == nil || base.DialTLS != nil || base.DialTLSContext != nil { + // An opaque transport or TLS dialer cannot establish these no-replay + // properties. Do not silently fall back to another trust/dial policy. + return nil, &guestAgentDeferredError{reason: "agent-transport-unverified"} + } + transport := base.Clone() + transport.DisableKeepAlives = true + transport.ForceAttemptHTTP2 = false + transport.Protocols = new(http.Protocols) + transport.Protocols.SetHTTP1(true) + transport.TLSNextProto = nil + if transport.TLSClientConfig != nil { + transport.TLSClientConfig.NextProtos = []string{"http/1.1"} + } + commandClient := *client + commandClient.Transport = transport + commandClient.CheckRedirect = func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse + } + return &commandClient, nil +} diff --git a/pkg/proxmox/guest_agent_transport_policy_test.go b/pkg/proxmox/guest_agent_transport_policy_test.go new file mode 100644 index 0000000000..cb6bdfa8f0 --- /dev/null +++ b/pkg/proxmox/guest_agent_transport_policy_test.go @@ -0,0 +1,183 @@ +package proxmox + +import ( + "context" + "crypto/sha256" + "crypto/tls" + "crypto/x509" + "errors" + "fmt" + "net" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestGuestAgentCommandClientPreservesPolicyWithoutMutation(t *testing.T) { + var proxyCalls, dialCalls, verificationCalls int + originalError := errors.New("fixture policy") + base := &http.Transport{ + Proxy: func(*http.Request) (*url.URL, error) { proxyCalls++; return nil, originalError }, + DialContext: func(context.Context, string, string) (net.Conn, error) { dialCalls++; return nil, originalError }, + TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS13, ServerName: "fixture.invalid", NextProtos: []string{"h2", "http/1.1"}, VerifyPeerCertificate: func([][]byte, [][]*x509.Certificate) error { verificationCalls++; return originalError }}, + MaxConnsPerHost: 7, ResponseHeaderTimeout: 2 * time.Second, + ForceAttemptHTTP2: true, + } + original := &http.Client{Transport: base, Timeout: 5 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return originalError }} + command, err := guestAgentCommandClient(original) + if err != nil { + t.Fatal(err) + } + transport := command.Transport.(*http.Transport) + if transport == base || transport.TLSClientConfig == base.TLSClientConfig || command == original { + t.Fatal("command policy aliases mutable original configuration") + } + if command.Timeout != original.Timeout || transport.MaxConnsPerHost != 7 || transport.ResponseHeaderTimeout != 2*time.Second { + t.Fatal("command lost request bounds") + } + if transport.TLSClientConfig.MinVersion != tls.VersionTLS13 || transport.TLSClientConfig.ServerName != "fixture.invalid" { + t.Fatal("command lost TLS policy") + } + transport.Proxy(nil) + transport.DialContext(context.Background(), "tcp", "fixture.invalid") + if !errors.Is(transport.TLSClientConfig.VerifyPeerCertificate(nil, nil), originalError) || proxyCalls != 1 || dialCalls != 1 || verificationCalls != 1 { + t.Fatal("command replaced configured proxy, dial or verification policy") + } + if !transport.DisableKeepAlives || transport.ForceAttemptHTTP2 || !transport.Protocols.HTTP1() || transport.Protocols.HTTP2() || transport.Protocols.UnencryptedHTTP2() || transport.TLSNextProto != nil { + t.Fatal("command transport admits connection/stream replay") + } + if strings.Join(transport.TLSClientConfig.NextProtos, ",") != "http/1.1" { + t.Fatal("command still advertises an alternate stream protocol") + } + if !errors.Is(command.CheckRedirect(nil, nil), http.ErrUseLastResponse) { + t.Fatal("command follows redirects") + } + if base.DisableKeepAlives || !base.ForceAttemptHTTP2 || base.Protocols != nil || strings.Join(base.TLSClientConfig.NextProtos, ",") != "h2,http/1.1" || !errors.Is(original.CheckRedirect(nil, nil), originalError) { + t.Fatal("command changed ordinary transport/redirect policy") + } +} + +func TestGuestAgentCommandClientRejectsUnverifiableTransports(t *testing.T) { + var calls atomic.Int32 + for name, client := range map[string]*http.Client{ + "missing client": nil, + "opaque transport": {Transport: roundTripperFunc(func(*http.Request) (*http.Response, error) { calls.Add(1); return nil, errors.New("must not execute") })}, + "typed nil transport": {Transport: (*http.Transport)(nil)}, + "TLS dial": {Transport: &http.Transport{DialTLS: func(string, string) (net.Conn, error) { calls.Add(1); return nil, errors.New("must not execute") }}}, + "TLS context dial": {Transport: &http.Transport{DialTLSContext: func(context.Context, string, string) (net.Conn, error) { + calls.Add(1) + return nil, errors.New("must not execute") + }}}, + } { + t.Run(name, func(t *testing.T) { + c := &Client{httpClient: client} + if _, err := c.GetVMAgentInfo(context.Background(), "node", 105); GuestAgentDeferredReason(err) != "agent-transport-unverified" { + t.Fatalf("unverified transport admitted: %v", err) + } + }) + } + if calls.Load() != 0 { + t.Fatal("transport policy failure dispatched work") + } + if command, err := guestAgentCommandClient(&http.Client{}); err != nil || command == nil { + t.Fatalf("standard default transport not admitted: %v", err) + } +} + +func TestGuestAgentCommandUsesTLSHTTP1AndSeparateConnections(t *testing.T) { + for _, mode := range []string{"trusted CA", "pinned fingerprint"} { + t.Run(mode, func(t *testing.T) { + var configs, commands atomic.Int32 + var mu sync.Mutex + peers := make(map[string]struct{}) + server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/config") { + configs.Add(1) + if r.ProtoMajor != 2 { + t.Error("ordinary API stopped using configured HTTP/2") + } + fmt.Fprint(w, `{"data":{}}`) + return + } + commands.Add(1) + if r.TLS == nil || r.ProtoMajor != 1 || !r.Close { + t.Error("guest command did not use single-use TLS HTTP/1") + } + if r.Header.Get("Authorization") != "PVEAPIToken=fixture@pve!pulse=fixture" { + t.Error("guest command lost configured authentication") + } + mu.Lock() + peers[r.RemoteAddr] = struct{}{} + mu.Unlock() + backupAgentPayload(w, r) + })) + server.EnableHTTP2 = true + server.StartTLS() + defer server.Close() + cfg := ClientConfig{Host: server.URL, TokenName: "fixture@pve!pulse", TokenValue: "fixture", VerifySSL: true, Timeout: time.Second} + if mode == "pinned fingerprint" { + cfg.Fingerprint = fmt.Sprintf("%x", sha256.Sum256(server.Certificate().Raw)) + } + client, err := NewClient(cfg) + if err != nil { + t.Fatal(err) + } + transport := client.httpClient.Transport.(*http.Transport) + if mode == "trusted CA" { + roots := x509.NewCertPool() + roots.AddCert(server.Certificate()) + transport.TLSClientConfig.RootCAs = roots + } + transport.ForceAttemptHTTP2 = true + transport.Protocols = new(http.Protocols) + transport.Protocols.SetHTTP1(true) + transport.Protocols.SetHTTP2(true) + for name, read := range backupAgentReads() { + t.Run(name, func(t *testing.T) { + if err := read(context.Background(), client, 105); err != nil { + t.Fatal(err) + } + }) + } + if configs.Load() != 12 || commands.Load() != 6 || len(peers) != 6 { + t.Fatalf("config/command/connection counts = %d/%d/%d, want 12/6/6", configs.Load(), commands.Load(), len(peers)) + } + if transport.DisableKeepAlives || !transport.Protocols.HTTP2() { + t.Fatal("command mutated shared ordinary transport") + } + }) + } +} + +func TestGuestAgentCommandTransportStillRejectsUntrustedTLS(t *testing.T) { + var requests atomic.Int32 + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { requests.Add(1); backupAgentPayload(w, r) })) + defer server.Close() + for _, fingerprint := range []string{"", strings.Repeat("00", 32)} { + client, err := NewClient(ClientConfig{Host: server.URL, TokenName: "fixture@pve!pulse", TokenValue: "fixture", VerifySSL: true, Fingerprint: fingerprint, Timeout: time.Second}) + if err != nil { + t.Fatal(err) + } + command, err := guestAgentCommandClient(client.httpClient) + if err != nil { + t.Fatal(err) + } + req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, server.URL+"/agent/info", nil) + if err != nil { + t.Fatal(err) + } + if response, err := command.Do(req); err == nil { + response.Body.Close() + t.Fatal("command transport accepted untrusted peer") + } + command.CloseIdleConnections() + } + if requests.Load() != 0 { + t.Fatal("untrusted TLS reached a guest endpoint") + } +} diff --git a/pkg/proxmox/guest_agent_transport_test.go b/pkg/proxmox/guest_agent_transport_test.go new file mode 100644 index 0000000000..f21b3be05f --- /dev/null +++ b/pkg/proxmox/guest_agent_transport_test.go @@ -0,0 +1,159 @@ +package proxmox + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" +) + +// A server that consumes a command and loses its reply models uncertain +// completion, not an idle socket that closed before receiving any work. +func TestGuestAgentWireDoesNotReplayLostReply(t *testing.T) { + for _, startsBackup := range []bool{false, true} { + for name, read := range backupAgentReads() { + t.Run(fmt.Sprintf("%s/backup=%t", name, startsBackup), func(t *testing.T) { + var commands, configs atomic.Int32 + var locked atomic.Bool + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/config") { + configs.Add(1) + if locked.Load() { + fmt.Fprint(w, `{"data":{"lock":"backup"}}`) + } else { + fmt.Fprint(w, `{"data":{}}`) + } + return + } + if strings.Contains(r.URL.Path, "/agent/") { + if commands.Add(1) == 1 { + locked.Store(startsBackup) + conn, _, err := w.(http.Hijacker).Hijack() + if err != nil { + t.Error(err) + return + } + // The request has reached the endpoint before the connection is lost. + conn.Close() + return + } + backupAgentPayload(w, r) + return + } + http.NotFound(w, r) + })) + defer server.Close() + client := backupTestClient(t, server.URL) + if err := read(context.Background(), client, 105); GuestAgentDeferredReason(err) != "agent-timeout" { + t.Errorf("lost command reply was not deferred: %v", err) + } + if got := commands.Load(); got != 1 { + t.Errorf("commands after lost reply = %d, want exactly one", got) + } + if got := configs.Load(); got != 1 { + t.Errorf("config reads after uncertain command = %d, want only pre-dispatch verification", got) + } + // A different method must not follow the uncertain command either. + if _, err := client.GetVMAgentInfo(context.Background(), "node", 105); GuestAgentDeferredReason(err) != "agent-cooldown" { + t.Errorf("subsequent command escaped VM cooldown: %v", err) + } + if got := commands.Load(); got != 1 { + t.Errorf("commands after cooldown read = %d, want one", got) + } + }) + } + } +} + +func TestGuestAgentWireDoesNotFollowRedirect(t *testing.T) { + for _, status := range []int{301, 302, 303, 307, 308} { + for name, read := range backupAgentReads() { + t.Run(fmt.Sprintf("%s/status=%d", name, status), func(t *testing.T) { + var first, redirected atomic.Int32 + destination := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + redirected.Add(1) + backupAgentPayload(w, r) + })) + defer destination.Close() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + return + } + first.Add(1) + http.Redirect(w, r, destination.URL+r.URL.Path, status) + })) + defer server.Close() + client := backupTestClient(t, server.URL) + if err := read(context.Background(), client, 105); GuestAgentDeferredReason(err) != "agent-redirect" { + t.Errorf("redirect response was not deferred: %v", err) + } + if got := redirected.Load(); got != 0 { + t.Errorf("unverified redirected commands = %d, want zero", got) + } + if got := first.Load(); got != 1 { + t.Errorf("initial commands = %d, want one", got) + } + if _, err := client.GetVMAgentInfo(context.Background(), "node", 105); GuestAgentDeferredReason(err) != "agent-cooldown" { + t.Errorf("redirect completion uncertainty escaped VM cooldown: %v", err) + } + if got := first.Load(); got != 1 { + t.Errorf("command replay after redirect = %d, want one", got) + } + if got := redirected.Load(); got != 0 { + t.Errorf("redirect destination reached during cooldown = %d", got) + } + }) + } + } +} + +func TestGuestAgentTransportLeavesOrdinaryRecoveryIntact(t *testing.T) { + var calls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if calls.Add(1) == 2 { + conn, _, err := w.(http.Hijacker).Hijack() + if err != nil { + t.Error(err) + return + } + conn.Close() + return + } + fmt.Fprint(w, `{"data":[]}`) + })) + defer server.Close() + client := backupTestClient(t, server.URL) + // Warm the ordinary transport. Its idempotent resource reads still recover + // a lost reply on that reused connection; only QGA commands are single-use. + for i := 0; i < 2; i++ { + if _, err := client.GetNodes(context.Background()); err != nil { + t.Fatal(err) + } + } + if got := calls.Load(); got != 3 { + t.Errorf("ordinary wire reads = %d, want warm read plus recovered pair", got) + } +} + +func TestGuestAgentTransportLeavesOrdinaryRedirectIntact(t *testing.T) { + var calls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + if r.URL.Path == "/api2/json/nodes" { + http.Redirect(w, r, "/redirected/nodes", http.StatusTemporaryRedirect) + return + } + fmt.Fprint(w, `{"data":[]}`) + })) + defer server.Close() + if _, err := backupTestClient(t, server.URL).GetNodes(context.Background()); err != nil { + t.Fatal(err) + } + if got := calls.Load(); got != 2 { + t.Errorf("ordinary redirect reads = %d, want two", got) + } +} diff --git a/pkg/proxmox/io_counters_test.go b/pkg/proxmox/io_counters_test.go index 1354c757e6..3bb74bbb5e 100644 --- a/pkg/proxmox/io_counters_test.go +++ b/pkg/proxmox/io_counters_test.go @@ -176,3 +176,61 @@ func TestUnverifiedVMConfigPreservesStatusCounterObservations(t *testing.T) { t.Errorf("guest commands with incomplete config = %d, want zero", got) } } + +func TestGuestAgentTransportDeferralPreservesLiveCounterReceipts(t *testing.T) { + for _, kind := range []string{"lost reply", "redirect"} { + t.Run(kind, func(t *testing.T) { + var commands, statusCalls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "/config"): + fmt.Fprint(w, `{"data":{}}`) + case strings.HasSuffix(r.URL.Path, "/status/current"): + statusCalls.Add(1) + fmt.Fprint(w, `{"data":{"status":"running","cpu":0.25,"diskread":0,"diskwrite":null,"netin":42}}`) + case strings.Contains(r.URL.Path, "/agent/"): + commands.Add(1) + if kind == "redirect" { + http.Redirect(w, r, "/unverified/agent", http.StatusTemporaryRedirect) + return + } + conn, _, err := w.(http.Hijacker).Hijack() + if err != nil { + t.Error(err) + return + } + conn.Close() + default: + http.NotFound(w, r) + } + })) + defer server.Close() + client := backupTestClient(t, server.URL) + wantReason := "agent-timeout" + if kind == "redirect" { + wantReason = "agent-redirect" + } + if _, err := client.GetVMFSInfo(context.Background(), "node", 105); GuestAgentDeferredReason(err) != wantReason { + t.Fatalf("uncertain command not deferred: %v", err) + } + // The QGA uncertainty is not missing or zero CPU/I/O evidence. Ordinary + // status still has its own completed response and observation receipt. + before := time.Now() + status, err := client.GetVMStatus(context.Background(), "node", 105) + after := time.Now() + if err != nil { + t.Fatal(err) + } + presence := status.IOCounters.Effective() + if status.CPU != 0.25 || status.DiskRead != 0 || status.NetIn != 42 || !presence.DiskRead || !presence.NetworkIn || presence.DiskWrite || presence.NetworkOut || status.ObservedAt.Before(before) || status.ObservedAt.After(after) { + t.Fatalf("guest transport deferral changed current counter presence/receipt: %+v", status) + } + if _, err := client.GetVMAgentInfo(context.Background(), "node", 105); GuestAgentDeferredReason(err) != "agent-cooldown" { + t.Fatalf("status response erased command uncertainty: %v", err) + } + if commands.Load() != 1 || statusCalls.Load() != 1 { + t.Fatalf("guest/status calls = %d/%d, want 1/1", commands.Load(), statusCalls.Load()) + } + }) + } +}