Prevent transparent replay of Proxmox guest-agent commands

Guest-agent GETs enqueue serial work: ordinary HTTP pooling, stream recovery and redirects can resend a command despite the application single-attempt guard. Use an isolated single-use HTTP/1 transport with the configured TLS, proxy, dial and timeout policy, and defer redirects without contacting their destination. Keep ordinary API recovery and current CPU/I/O counter presence/receipts unchanged. Pin wire-level lost replies, all read/redirect paths, TLS trust and connected deferred cache/History behaviour.

Change-source: pulse-maintainer

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.

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)
+}

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)
+				}
+			}
+		})
+	}
+}

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

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

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 {

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

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")
+	}
+}

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)
+	}
+}

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())
+			}
+		})
+	}
+}