Require complete Proxmox password sessions before publication

Reject incomplete or ambiguous 200 login bodies before replacing ticket, CSRF, expiry and PVE generation. Preserve retained sessions, joined renewal failures, normal encoding fallback and later ordinary recovery; cover real pinned startup/poll/broadcast consumption.

Contract-Neutral: Restore existing complete-session and failed-renewal preservation guarantees; no API/wire schema, TLS trust, credential authority, guest lifecycle, polling cadence or History contract delta.
Change-source: pulse-maintainer

docs/release-control/v6/internal/subsystems/monitoring.md

@@ -17,6 +17,36 @@
 
 ## Purpose
 
+### Proxmox password success admits a complete session, not just HTTP 200
+
+PVE and PBS consume the entire existing byte-bounded login body before
+publishing ticket, CSRF token and the normal two-hour expiry together. The
+shared `internal/securityutil/proxmox_session.go` parser requires canonical,
+unambiguous envelope/session keys, complete UTF-8 JSON, a nonempty cookie-safe
+ticket and a nonempty header-safe CSRF token. Unknown unique metadata such as
+username and capabilities remains supported. Partial/null, duplicate or
+case-conflicting credentials, extra JSON values, incomplete reads and oversized
+tails fail without replacing the retained tuple or advancing PVE generation.
+Validation errors contain no response credentials. A malformed success is not
+a provider credential refusal and does not trigger JSON-to-form fallback.
+
+`session_response_admission_test.go` in both client packages exercises these
+controls and eight joined expired readers, zero subsequent API/form requests
+on invalid success, retained state and distinct later ordinary recovery with
+coherent cookie/CSRF pairs. Cluster recovery additionally checks empty and
+partial success for the retained polling client, with and without a previous
+client. `TestPBSInvalidSessionKeepsFirstPollBoundary` exercises the real pinned
+constructor, normal startup retry, poll executor and production broadcast:
+invalid success installs no client, valid construction is not healthy telemetry,
+and only the accepted first poll publishes recovered node/datastore readings.
+
+TLS trust, credentials, auth encodings and 400/415 fallback, API-token paths,
+renewal ownership, retries/cadence, guest/backup fences and History are unchanged.
+The parser establishes response integrity, not cryptographic ticket validation
+or native/installed acceptance. It neither attributes nor resolves a reported
+PBS, Windows guest, restart or workload-safety incident.
+
+
 ### Proxmox recovery sessions honour the owning caller
 
 PVE/PBS context-aware constructors bind initial password login, including the

docs/release-control/v6/internal/subsystems/security-privacy.md

@@ -18,6 +18,36 @@
 
 ## Purpose
 
+### Proxmox password success admits a complete session, not just HTTP 200
+
+PVE and PBS consume the entire existing byte-bounded login body before
+publishing ticket, CSRF token and the normal two-hour expiry together. The
+shared `internal/securityutil/proxmox_session.go` parser requires canonical,
+unambiguous envelope/session keys, complete UTF-8 JSON, a nonempty cookie-safe
+ticket and a nonempty header-safe CSRF token. Unknown unique metadata such as
+username and capabilities remains supported. Partial/null, duplicate or
+case-conflicting credentials, extra JSON values, incomplete reads and oversized
+tails fail without replacing the retained tuple or advancing PVE generation.
+Validation errors contain no response credentials. A malformed success is not
+a provider credential refusal and does not trigger JSON-to-form fallback.
+
+`session_response_admission_test.go` in both client packages exercises these
+controls and eight joined expired readers, zero subsequent API/form requests
+on invalid success, retained state and distinct later ordinary recovery with
+coherent cookie/CSRF pairs. Cluster recovery additionally checks empty and
+partial success for the retained polling client, with and without a previous
+client. `TestPBSInvalidSessionKeepsFirstPollBoundary` exercises the real pinned
+constructor, normal startup retry, poll executor and production broadcast:
+invalid success installs no client, valid construction is not healthy telemetry,
+and only the accepted first poll publishes recovered node/datastore readings.
+
+TLS trust, credentials, auth encodings and 400/415 fallback, API-token paths,
+renewal ownership, retries/cadence, guest/backup fences and History are unchanged.
+The parser establishes response integrity, not cryptographic ticket validation
+or native/installed acceptance. It neither attributes nor resolves a reported
+PBS, Windows guest, restart or workload-safety incident.
+
+
 ### Rejected infrastructure URLs stay out of diagnostics
 
 PVE/PBS constructors validate a host before logging it. URL syntax errors omit

internal/monitoring/monitor_session_admission_test.go

@@ -0,0 +1,102 @@
+package monitoring
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"io"
+	"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"
+)
+
+// The real password constructor, normal retry, poll executor and broadcast
+// projection feed browser proof of the unchanged PBS consuming surface. No
+// guest command, backup, installed service, database or provider is exercised.
+func TestPBSInvalidSessionKeepsFirstPollBoundary(t *testing.T) {
+	var logins, reads atomic.Int32
+	server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.Copy(io.Discard, r.Body)
+		w.Header().Set("Content-Type", "application/json")
+		if r.URL.Path == "/api2/json/access/ticket" {
+			if logins.Add(1) == 1 {
+				// A syntactically valid 200 with no session, not a credential
+				// refusal and not an invitation to replay another encoding.
+				fmt.Fprint(w, `{"data":{}}`)
+				return
+			}
+			fmt.Fprint(w, `{"data":{"ticket":"fixture-ticket","CSRFPreventionToken":"fixture-csrf"}}`)
+			return
+		}
+		reads.Add(1)
+		if r.Method != http.MethodGet || !strings.Contains(r.Header.Get("Cookie"), "fixture-ticket") {
+			t.Errorf("non-ordinary or unauthenticated PBS fixture read: %s %s", r.Method, r.URL.Path)
+		}
+		switch r.URL.Path {
+		case "/api2/json/version":
+			fmt.Fprint(w, `{"data":{"version":"3.4.2"}}`)
+		case "/api2/json/nodes/localhost/status":
+			fmt.Fprint(w, `{"data":{"cpu":0.15,"memory":{"used":512,"total":1024},"uptime":120}}`)
+		case "/api2/json/admin/datastore":
+			fmt.Fprint(w, `{"data":[{"store":"backups","total":1000,"used":250,"avail":750}]}`)
+		case "/api2/json/admin/datastore/backups/status":
+			fmt.Fprint(w, `{"data":{"status":"available","total":1000,"used":250,"avail":750}}`)
+		default:
+			fmt.Fprint(w, `{"data":[]}`)
+		}
+	}))
+	defer server.Close()
+	pin := sha256.Sum256(server.Certificate().Raw)
+	instance := config.PBSInstance{Name: "session-pbs", Host: server.URL, User: "fixture@pbs", Password: "fixture-password", VerifySSL: true, Fingerprint: hex.EncodeToString(pin[:]), MonitorDatastores: true}
+	monitor := newPBSHealthAuthorityMonitor([]config.PBSInstance{instance})
+	project := func(phase string) {
+		snapshot, scope := monitor.currentStateWithScope()
+		frontend := monitor.buildBroadcastFrontendStateFromSnapshot(snapshot, scope)
+		wire, err := json.Marshal(struct {
+			Phase string               `json:"phase"`
+			State models.StateFrontend `json:"state"`
+		}{phase, frontend})
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Logf("PBS_SESSION_ADMISSION_PROJECTION_JSON=%s", wire)
+	}
+	monitor.initPBSClients(monitor.config)
+	assertPBSConnectionProjection(t, monitor, instance.Name, false, "offline")
+	if logins.Load() != 1 || reads.Load() != 0 || len(monitor.pbsClients) != 0 || len(newPBSPollProvider().ListInstances(monitor)) != 0 {
+		t.Fatal("unusable login success published a polling client or dispatched API/form work")
+	}
+	project("invalid-session")
+
+	originalDelays := connRetryDelays
+	connRetryDelays = []time.Duration{time.Millisecond}
+	defer func() { connRetryDelays = originalDelays }()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	monitor.retryFailedConnections(ctx)
+	if logins.Load() != 2 || reads.Load() != 0 || len(newPBSPollProvider().ListInstances(monitor)) != 1 {
+		t.Fatal("later normal retry did not install its accepted session exactly once")
+	}
+	assertPBSConnectionProjection(t, monitor, instance.Name, false, "offline")
+	project("awaiting-poll")
+	task, err := newPBSPollProvider().BuildPollTask(monitor, instance.Name)
+	if err != nil {
+		t.Fatal(err)
+	}
+	newRealExecutor(monitor).Execute(ctx, task)
+	assertPBSConnectionProjection(t, monitor, instance.Name, true, "online")
+	actual := pbsInstanceByName(t, monitor.state.GetSnapshot(), instance.Name)
+	if actual.CPU != 0.15 || actual.Memory != 50 || len(actual.Datastores) != 1 || actual.Datastores[0].Usage != 25 || logins.Load() != 2 || !reflect.DeepEqual(instance, monitor.config.PBSInstances[0]) {
+		t.Fatal("ordinary first poll lost readings, identity or saved trust")
+	}
+	project("recovered")
+}

internal/securityutil/proxmox_session.go

@@ -0,0 +1,94 @@
+package securityutil
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"io"
+	"strings"
+	"unicode/utf8"
+)
+
+// ErrInvalidProxmoxSession deliberately contains no response or credential
+// material. A malformed 200 is not a provider's credential refusal and must
+// not authorise a second login encoding or replace the last complete session.
+var ErrInvalidProxmoxSession = errors.New("invalid authentication response: expected a complete ticket and CSRF token")
+
+// ParseProxmoxSession validates a complete, already byte-bounded PVE/PBS login
+// body before either client publishes it. Metadata such as username and cap is
+// allowed, but duplicate keys and case-insensitive credential overrides are
+// not. This establishes response integrity, not server acceptance of a ticket.
+func ParseProxmoxSession(body []byte) (ticket, csrf string, err error) {
+	if !utf8.Valid(body) {
+		return "", "", ErrInvalidProxmoxSession
+	}
+	envelope, ok := proxmoxSessionObject(body, "data", "errors")
+	if !ok {
+		return "", "", ErrInvalidProxmoxSession
+	}
+	if failure, exists := envelope["errors"]; exists && !bytes.Equal(bytes.TrimSpace(failure), []byte("null")) {
+		return "", "", ErrInvalidProxmoxSession
+	}
+	session, ok := proxmoxSessionObject(envelope["data"], "ticket", "CSRFPreventionToken")
+	if !ok || json.Unmarshal(session["ticket"], &ticket) != nil || json.Unmarshal(session["CSRFPreventionToken"], &csrf) != nil ||
+		!proxmoxSessionValue(ticket, true) || !proxmoxSessionValue(csrf, false) {
+		return "", "", ErrInvalidProxmoxSession
+	}
+	return ticket, csrf, nil
+}
+
+func proxmoxSessionObject(body []byte, canonical ...string) (map[string]json.RawMessage, bool) {
+	decoder := json.NewDecoder(bytes.NewReader(body))
+	start, err := decoder.Token()
+	if err != nil || start != json.Delim('{') {
+		return nil, false
+	}
+	fields := make(map[string]json.RawMessage)
+	for decoder.More() {
+		key, err := decoder.Token()
+		if err != nil {
+			return nil, false
+		}
+		name, ok := key.(string)
+		if !ok {
+			return nil, false
+		}
+		if _, duplicate := fields[name]; duplicate {
+			return nil, false
+		}
+		for _, expected := range canonical {
+			if name != expected && strings.EqualFold(name, expected) {
+				return nil, false
+			}
+		}
+		var value json.RawMessage
+		if decoder.Decode(&value) != nil {
+			return nil, false
+		}
+		fields[name] = value
+	}
+	end, err := decoder.Token()
+	if err != nil || end != json.Delim('}') {
+		return nil, false
+	}
+	if _, err := decoder.Token(); err != io.EOF {
+		return nil, false
+	}
+	return fields, true
+}
+
+func proxmoxSessionValue(value string, cookie bool) bool {
+	if value == "" {
+		return false
+	}
+	for i := 0; i < len(value); i++ {
+		b := value[i]
+		// Credentials are opaque: do not trim, repair or log them. Require a
+		// visible ASCII header value and, for the unquoted ticket cookie, the
+		// cookie-octet subset. Real ticket delimiters and base64 stay supported.
+		if b <= ' ' || b >= 0x7f || (cookie && (b == '"' || b == ',' || b == ';' || b == '\\')) {
+			return false
+		}
+	}
+	return true
+}

internal/securityutil/proxmox_session_test.go

@@ -0,0 +1,30 @@
+package securityutil
+
+import (
+	"errors"
+	"strings"
+	"testing"
+)
+
+func TestParseProxmoxSession(t *testing.T) {
+	for _, ticket := range []string{"PVE:fixture@pam:1234::base64+/=", "PBS:fixture@pbs:1234::base64+/="} {
+		body := []byte(`{"data":{"ticket":"` + ticket + `","CSRFPreventionToken":"1234:base64+/=","username":"fixture@pam","cap":{"VM.Audit":1}},"errors":null}`)
+		gotTicket, csrf, err := ParseProxmoxSession(body)
+		if err != nil || gotTicket != ticket || csrf != "1234:base64+/=" {
+			t.Fatal("complete native-format fixture with metadata was not preserved")
+		}
+	}
+	for _, body := range []string{
+		`{"data":{"ticket":"fixture-sensitive","CSRFPreventionToken":"csrf"},"errors":{"password":"rejected"}}`,
+		`{"data":{"ticket":"fixture-sensitive","CSRFPreventionToken":"csrf"},"Errors":null}`,
+		`{"data":{"ticket":"fixture-sensitive","CSRFPreventionToken":"csrf","username":null,"username":"fixture"}}`,
+		`{"data":{"ticket":"fixture-sensitive","CSRFPreventionToken":"csrf"},"data":null}`,
+		`{"data":{"ticket":"fixture-sensitive","CSRFPreventionToken":"csrf"}} []`,
+		`[]`, `null`, `{"data":[]}`,
+	} {
+		ticket, csrf, err := ParseProxmoxSession([]byte(body))
+		if !errors.Is(err, ErrInvalidProxmoxSession) || ticket != "" || csrf != "" || strings.Contains(err.Error(), "fixture-sensitive") {
+			t.Fatal("invalid response admitted credentials or exposed their bytes")
+		}
+	}
+}

pkg/pbs/client.go

@@ -339,20 +339,18 @@ func (c *Client) handleAuthResponse(resp *http.Response) error {
 		return err
 	}
 
-	var result struct {
-		Data struct {
-			Ticket              string `json:"ticket"`
-			CSRFPreventionToken string `json:"CSRFPreventionToken"`
-		} `json:"data"`
+	body, err := readResponseBodyLimited(resp.Body)
+	if err != nil {
+		return err
 	}
-
-	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+	ticket, csrf, err := securityutil.ParseProxmoxSession(body)
+	if err != nil {
 		return err
 	}
 
 	c.authMu.Lock()
-	c.auth.ticket = result.Data.Ticket
-	c.auth.csrfToken = result.Data.CSRFPreventionToken
+	c.auth.ticket = ticket
+	c.auth.csrfToken = csrf
 	c.auth.expiresAt = time.Now().Add(2 * time.Hour) // PBS tickets expire after 2 hours
 	c.authMu.Unlock()
 

pkg/pbs/session_response_admission_test.go

@@ -0,0 +1,189 @@
+package pbs
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"reflect"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+type sessionAdmissionReadError struct{ io.Reader }
+
+func (r sessionAdmissionReadError) Read(p []byte) (int, error) {
+	n, err := r.Reader.Read(p)
+	if err == io.EOF {
+		err = io.ErrUnexpectedEOF
+	}
+	return n, err
+}
+
+// Synthetic response credentials never come from an installed service. A 200
+// alone must not replace a session or refresh its expiry/generation.
+func TestClientSessionResponseAdmission(t *testing.T) {
+	const good = `{"data":{"ticket":"fixture-ticket:+/=_-","CSRFPreventionToken":"fixture-csrf:+/=_-","username":"fixture@pam","cap":{"VM.Audit":1}}}`
+	tests := []struct {
+		name, body       string
+		valid, readError bool
+	}{
+		{"complete-session", good, true, false},
+		{"complete-whitespace", " \n" + good + "\t", true, false},
+		{"missing-data", `{}`, false, false},
+		{"null-data", `{"data":null}`, false, false},
+		{"empty-data", `{"data":{}}`, false, false},
+		{"missing-ticket", `{"data":{"CSRFPreventionToken":"fixture-sensitive-csrf"}}`, false, false},
+		{"null-ticket", `{"data":{"ticket":null,"CSRFPreventionToken":"fixture-sensitive-csrf"}}`, false, false},
+		{"blank-ticket", `{"data":{"ticket":" ","CSRFPreventionToken":"fixture-sensitive-csrf"}}`, false, false},
+		{"missing-csrf", `{"data":{"ticket":"fixture-sensitive-ticket"}}`, false, false},
+		{"null-csrf", `{"data":{"ticket":"fixture-sensitive-ticket","CSRFPreventionToken":null}}`, false, false},
+		{"empty-csrf", `{"data":{"ticket":"fixture-sensitive-ticket","CSRFPreventionToken":""}}`, false, false},
+		{"wrong-type-csrf", `{"data":{"ticket":"fixture-sensitive-ticket","CSRFPreventionToken":42}}`, false, false},
+		{"duplicate-data", `{"data":null,"data":{"ticket":"fixture-sensitive-ticket","CSRFPreventionToken":"fixture-sensitive-csrf"}}`, false, false},
+		{"duplicate-ticket", `{"data":{"ticket":"","ticket":"fixture-sensitive-ticket","CSRFPreventionToken":"fixture-sensitive-csrf"}}`, false, false},
+		{"duplicate-csrf", `{"data":{"ticket":"fixture-sensitive-ticket","CSRFPreventionToken":"","CSRFPreventionToken":"fixture-sensitive-csrf"}}`, false, false},
+		{"escaped-duplicate", `{"data":{"ticket":"","ti\u0063ket":"fixture-sensitive-ticket","CSRFPreventionToken":"fixture-sensitive-csrf"}}`, false, false},
+		{"noncanonical-data", strings.Replace(good, `"data"`, `"Data"`, 1), false, false},
+		{"noncanonical-ticket", strings.Replace(good, `"ticket"`, `"Ticket"`, 1), false, false},
+		{"noncanonical-csrf", strings.Replace(good, `"CSRFPreventionToken"`, `"csrfpreventiontoken"`, 1), false, false},
+		{"conflicting-case", strings.Replace(good, `"ticket":`, `"Ticket":null,"ticket":`, 1), false, false},
+		{"trailing-object", good + ` {"data":null}`, false, false},
+		{"trailing-invalid", good + ` not-json`, false, false},
+		{"unfinished-body", good + ` `, false, true},
+		{"oversized-tail", good + strings.Repeat(" ", int(maxResponseBodyBytes)), false, false},
+		{"broken-utf8", strings.Replace(good, "fixture-ticket", "fixture-"+string([]byte{0xff}), 1), false, false},
+		{"cookie-injection", strings.Replace(good, "fixture-ticket:+/=_-", `fixture-sensitive-ticket;other=authority`, 1), false, false},
+		{"header-injection", strings.Replace(good, "fixture-csrf:+/=_-", `fixture-sensitive-csrf\r\nAuthorization: other`, 1), false, false},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			c := &Client{auth: auth{user: "fixture", realm: "pam", ticket: "retained-ticket", csrfToken: "retained-csrf", expiresAt: time.Now().Add(-time.Hour)}}
+
+			before := c.auth
+			var body io.Reader = strings.NewReader(tc.body)
+			if tc.readError {
+				body = sessionAdmissionReadError{body}
+			}
+			resp := &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(body), ContentLength: -1}
+			err := c.handleAuthResponse(resp)
+			resp.Body.Close()
+			if !tc.valid {
+				if err == nil {
+					t.Error("unusable or incomplete 200 admitted as a session")
+				}
+				if !reflect.DeepEqual(c.auth, before) {
+					t.Error("rejected success replaced retained ticket/CSRF/expiry")
+				}
+
+				if shouldFallbackToForm(err) {
+					t.Error("invalid success authorised another login encoding")
+				}
+				if err != nil && (strings.Contains(err.Error(), "fixture-sensitive") || strings.Contains(err.Error(), "fixture-ticket") || strings.Contains(err.Error(), "fixture-csrf")) {
+					t.Error("response credentials leaked into validation error")
+				}
+				return
+			}
+			if err != nil {
+				t.Fatal(err)
+			}
+			if c.auth.ticket != "fixture-ticket:+/=_-" || c.auth.csrfToken != "fixture-csrf:+/=_-" || time.Until(c.auth.expiresAt) < 119*time.Minute || time.Until(c.auth.expiresAt) > 2*time.Hour {
+				t.Error("complete session or normal lifetime changed")
+			}
+
+		})
+	}
+}
+
+// Expired ordinary reads all share the same invalid success, retain their old
+// tuple and dispatch no API work. A distinct later read can accept a complete
+// session without resetting credentials, clocks, identities or safety fences.
+func TestClientSessionInvalidSuccessRecovery(t *testing.T) {
+	const readers = 8
+	var logins, reads, forms atomic.Int32
+	entered, release := make(chan struct{}), make(chan struct{})
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+	defer cancel()
+	server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.Copy(io.Discard, r.Body)
+		if r.URL.Path == "/api2/json/access/ticket" {
+			if r.Header.Get("Content-Type") == "application/x-www-form-urlencoded" {
+				forms.Add(1)
+			}
+			n := logins.Add(1)
+			if n == 2 {
+				close(entered)
+				select {
+				case <-release:
+				case <-ctx.Done():
+					return
+				}
+				fmt.Fprint(w, `{"data":{"ticket":"fixture-sensitive-ticket"}}`)
+				return
+			}
+			fmt.Fprintf(w, `{"data":{"ticket":"session-%d","CSRFPreventionToken":"csrf-%d"}}`, n, n)
+			return
+		}
+		reads.Add(1)
+		cookie, err := r.Cookie("PBSAuthCookie")
+		if err != nil || cookie.Value != "session-3" || (r.Method != http.MethodGet && r.Header.Get("CSRFPreventionToken") != "csrf-3") {
+			t.Error("ordinary recovered read lost ticket/CSRF coherence")
+			w.WriteHeader(403)
+			return
+		}
+		fmt.Fprint(w, `{"data":{"cpu":0.25,"memory":{"total":1024,"used":256},"uptime":60}}`)
+	}))
+	defer server.Close()
+	c, err := NewClient(ClientConfig{Host: server.URL, User: "fixture@pam", Password: "fixture-password", VerifySSL: true, Fingerprint: func() string { pin := sha256.Sum256(server.Certificate().Raw); return hex.EncodeToString(pin[:]) }(), Timeout: 5 * time.Second})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.httpClient.CloseIdleConnections()
+	c.authMu.Lock()
+	c.auth.expiresAt = time.Now().Add(-time.Hour)
+	before := c.auth
+	c.authMu.Unlock()
+	done := make(chan error, readers)
+	go func() { _, err := c.GetNodeStatus(ctx); done <- err }()
+	pbsSessionAwait(t, entered)
+	for i := 1; i < readers; i++ {
+		joined := &pbsSessionJoinContext{Context: ctx, joined: make(chan struct{})}
+		go func() { _, err := c.GetNodeStatus(joined); done <- err }()
+		pbsSessionAwait(t, joined.joined)
+	}
+	close(release)
+	for i := 0; i < readers; i++ {
+		if err := <-done; err == nil {
+			t.Error("joined expired read advanced past unusable login success")
+		} else if strings.Contains(err.Error(), "fixture-sensitive") {
+			t.Error("failed renewal error contains session bytes")
+		}
+	}
+	c.authMu.Lock()
+	unchanged := reflect.DeepEqual(c.auth, before)
+	c.authMu.Unlock()
+	if !unchanged || logins.Load() != 2 || reads.Load() != 0 || forms.Load() != 0 {
+		t.Errorf("invalid success changed session or dispatched work: same=%t auth/API/form=%d/%d/%d", unchanged, logins.Load(), reads.Load(), forms.Load())
+	}
+	if _, err := c.GetNodeStatus(ctx); err != nil {
+		t.Fatal(err)
+	}
+	resp, err := c.request(ctx, http.MethodPost, "/synthetic-read", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = io.Copy(io.Discard, resp.Body)
+	resp.Body.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if logins.Load() != 3 || reads.Load() != 2 || forms.Load() != 0 {
+		t.Errorf("later ordinary recovery auth/API/form=%d/%d/%d", logins.Load(), reads.Load(), forms.Load())
+	}
+	t.Log("eight joined failures, zero API/form dispatches, retained tuple; one later login plus coherent ordinary GET/POST recovery")
+}

pkg/proxmox/client.go

@@ -544,20 +544,18 @@ func (c *Client) handleAuthResponse(resp *http.Response) error {
 		return err
 	}
 
-	var result struct {
-		Data struct {
-			Ticket              string `json:"ticket"`
-			CSRFPreventionToken string `json:"CSRFPreventionToken"`
-		} `json:"data"`
+	body, err := readResponseBodyLimited(resp.Body)
+	if err != nil {
+		return err
 	}
-
-	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+	ticket, csrf, err := securityutil.ParseProxmoxSession(body)
+	if err != nil {
 		return err
 	}
 
 	c.authMu.Lock()
-	c.auth.ticket = result.Data.Ticket
-	c.auth.csrfToken = result.Data.CSRFPreventionToken
+	c.auth.ticket = ticket
+	c.auth.csrfToken = csrf
 	c.auth.expiresAt = time.Now().Add(2 * time.Hour) // PVE tickets expire after 2 hours
 	c.authGeneration++
 	c.authMu.Unlock()

pkg/proxmox/cluster_client_recovery_session_test.go

@@ -18,7 +18,7 @@ import (
 // establish that the second was accepted. All credentials and TLS peers here
 // are guest-local fixtures, not an installed authentication/recovery attempt.
 func testClusterRecoveryRequiresUsableSession(t *testing.T) {
-	for _, refusal := range []string{"unauthorised", "forbidden", "malformed", "form-refused"} {
+	for _, refusal := range []string{"unauthorised", "forbidden", "malformed", "empty-session", "partial-session", "form-refused"} {
 		for _, retained := range []bool{false, true} {
 			t.Run(fmt.Sprintf("%s/retained-%t", refusal, retained), func(t *testing.T) {
 				var failing atomic.Bool
@@ -38,6 +38,10 @@ func testClusterRecoveryRequiresUsableSession(t *testing.T) {
 								fmt.Fprint(w, `{"message":"fixture session refused"}`)
 							case "malformed":
 								fmt.Fprint(w, `{"data":`)
+							case "empty-session":
+								fmt.Fprint(w, `{"data":{}}`)
+							case "partial-session":
+								fmt.Fprint(w, `{"data":{"ticket":"fixture-ticket"}}`)
 							case "form-refused":
 								if r.Header.Get("Content-Type") == "application/json" {
 									w.WriteHeader(http.StatusUnsupportedMediaType)

pkg/proxmox/session_response_admission_test.go

@@ -0,0 +1,195 @@
+package proxmox
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"reflect"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+type sessionAdmissionReadError struct{ io.Reader }
+
+func (r sessionAdmissionReadError) Read(p []byte) (int, error) {
+	n, err := r.Reader.Read(p)
+	if err == io.EOF {
+		err = io.ErrUnexpectedEOF
+	}
+	return n, err
+}
+
+// Synthetic response credentials never come from an installed service. A 200
+// alone must not replace a session or refresh its expiry/generation.
+func TestClientSessionResponseAdmission(t *testing.T) {
+	const good = `{"data":{"ticket":"fixture-ticket:+/=_-","CSRFPreventionToken":"fixture-csrf:+/=_-","username":"fixture@pam","cap":{"VM.Audit":1}}}`
+	tests := []struct {
+		name, body       string
+		valid, readError bool
+	}{
+		{"complete-session", good, true, false},
+		{"complete-whitespace", " \n" + good + "\t", true, false},
+		{"missing-data", `{}`, false, false},
+		{"null-data", `{"data":null}`, false, false},
+		{"empty-data", `{"data":{}}`, false, false},
+		{"missing-ticket", `{"data":{"CSRFPreventionToken":"fixture-sensitive-csrf"}}`, false, false},
+		{"null-ticket", `{"data":{"ticket":null,"CSRFPreventionToken":"fixture-sensitive-csrf"}}`, false, false},
+		{"blank-ticket", `{"data":{"ticket":" ","CSRFPreventionToken":"fixture-sensitive-csrf"}}`, false, false},
+		{"missing-csrf", `{"data":{"ticket":"fixture-sensitive-ticket"}}`, false, false},
+		{"null-csrf", `{"data":{"ticket":"fixture-sensitive-ticket","CSRFPreventionToken":null}}`, false, false},
+		{"empty-csrf", `{"data":{"ticket":"fixture-sensitive-ticket","CSRFPreventionToken":""}}`, false, false},
+		{"wrong-type-csrf", `{"data":{"ticket":"fixture-sensitive-ticket","CSRFPreventionToken":42}}`, false, false},
+		{"duplicate-data", `{"data":null,"data":{"ticket":"fixture-sensitive-ticket","CSRFPreventionToken":"fixture-sensitive-csrf"}}`, false, false},
+		{"duplicate-ticket", `{"data":{"ticket":"","ticket":"fixture-sensitive-ticket","CSRFPreventionToken":"fixture-sensitive-csrf"}}`, false, false},
+		{"duplicate-csrf", `{"data":{"ticket":"fixture-sensitive-ticket","CSRFPreventionToken":"","CSRFPreventionToken":"fixture-sensitive-csrf"}}`, false, false},
+		{"escaped-duplicate", `{"data":{"ticket":"","ti\u0063ket":"fixture-sensitive-ticket","CSRFPreventionToken":"fixture-sensitive-csrf"}}`, false, false},
+		{"noncanonical-data", strings.Replace(good, `"data"`, `"Data"`, 1), false, false},
+		{"noncanonical-ticket", strings.Replace(good, `"ticket"`, `"Ticket"`, 1), false, false},
+		{"noncanonical-csrf", strings.Replace(good, `"CSRFPreventionToken"`, `"csrfpreventiontoken"`, 1), false, false},
+		{"conflicting-case", strings.Replace(good, `"ticket":`, `"Ticket":null,"ticket":`, 1), false, false},
+		{"trailing-object", good + ` {"data":null}`, false, false},
+		{"trailing-invalid", good + ` not-json`, false, false},
+		{"unfinished-body", good + ` `, false, true},
+		{"oversized-tail", good + strings.Repeat(" ", int(maxResponseBodyBytes)), false, false},
+		{"broken-utf8", strings.Replace(good, "fixture-ticket", "fixture-"+string([]byte{0xff}), 1), false, false},
+		{"cookie-injection", strings.Replace(good, "fixture-ticket:+/=_-", `fixture-sensitive-ticket;other=authority`, 1), false, false},
+		{"header-injection", strings.Replace(good, "fixture-csrf:+/=_-", `fixture-sensitive-csrf\r\nAuthorization: other`, 1), false, false},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			c := &Client{auth: auth{user: "fixture", realm: "pam", ticket: "retained-ticket", csrfToken: "retained-csrf", expiresAt: time.Now().Add(-time.Hour)}}
+			c.authGeneration = 3
+			before := c.auth
+			var body io.Reader = strings.NewReader(tc.body)
+			if tc.readError {
+				body = sessionAdmissionReadError{body}
+			}
+			resp := &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(body), ContentLength: -1}
+			err := c.handleAuthResponse(resp)
+			resp.Body.Close()
+			if !tc.valid {
+				if err == nil {
+					t.Error("unusable or incomplete 200 admitted as a session")
+				}
+				if !reflect.DeepEqual(c.auth, before) {
+					t.Error("rejected success replaced retained ticket/CSRF/expiry")
+				}
+				if c.authGeneration != 3 {
+					t.Error("invalid success advanced session generation")
+				}
+				if shouldFallbackToForm(err) {
+					t.Error("invalid success authorised another login encoding")
+				}
+				if err != nil && (strings.Contains(err.Error(), "fixture-sensitive") || strings.Contains(err.Error(), "fixture-ticket") || strings.Contains(err.Error(), "fixture-csrf")) {
+					t.Error("response credentials leaked into validation error")
+				}
+				return
+			}
+			if err != nil {
+				t.Fatal(err)
+			}
+			if c.auth.ticket != "fixture-ticket:+/=_-" || c.auth.csrfToken != "fixture-csrf:+/=_-" || time.Until(c.auth.expiresAt) < 119*time.Minute || time.Until(c.auth.expiresAt) > 2*time.Hour {
+				t.Error("complete session or normal lifetime changed")
+			}
+			if c.authGeneration != 4 {
+				t.Error("complete session did not advance once")
+			}
+		})
+	}
+}
+
+// Expired ordinary reads all share the same invalid success, retain their old
+// tuple and dispatch no API work. A distinct later read can accept a complete
+// session without resetting credentials, clocks, identities or safety fences.
+func TestClientSessionInvalidSuccessRecovery(t *testing.T) {
+	const readers = 8
+	var logins, reads, forms atomic.Int32
+	entered, release := make(chan struct{}), make(chan struct{})
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+	defer cancel()
+	server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.Copy(io.Discard, r.Body)
+		if r.URL.Path == "/api2/json/access/ticket" {
+			if r.Header.Get("Content-Type") == "application/x-www-form-urlencoded" {
+				forms.Add(1)
+			}
+			n := logins.Add(1)
+			if n == 2 {
+				close(entered)
+				select {
+				case <-release:
+				case <-ctx.Done():
+					return
+				}
+				fmt.Fprint(w, `{"data":{"ticket":"fixture-sensitive-ticket"}}`)
+				return
+			}
+			fmt.Fprintf(w, `{"data":{"ticket":"session-%d","CSRFPreventionToken":"csrf-%d"}}`, n, n)
+			return
+		}
+		reads.Add(1)
+		cookie, err := r.Cookie("PVEAuthCookie")
+		if err != nil || cookie.Value != "session-3" || (r.Method != http.MethodGet && r.Header.Get("CSRFPreventionToken") != "csrf-3") {
+			t.Error("ordinary recovered read lost ticket/CSRF coherence")
+			w.WriteHeader(403)
+			return
+		}
+		fmt.Fprint(w, `{"data":[{"node":"fixture-node","status":"online"}]}`)
+	}))
+	defer server.Close()
+	c, err := NewClient(ClientConfig{Host: server.URL, User: "fixture@pam", Password: "fixture-password", VerifySSL: true, Fingerprint: func() string { pin := sha256.Sum256(server.Certificate().Raw); return hex.EncodeToString(pin[:]) }(), Timeout: 5 * time.Second})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.httpClient.CloseIdleConnections()
+	c.authMu.Lock()
+	c.auth.expiresAt = time.Now().Add(-time.Hour)
+	before := c.auth
+	generation := c.authGeneration
+	c.authMu.Unlock()
+	done := make(chan error, readers)
+	go func() { _, err := c.GetNodes(ctx); done <- err }()
+	awaitSessionSignal(t, entered)
+	for i := 1; i < readers; i++ {
+		joined := &authJoinContext{Context: ctx, joined: make(chan struct{})}
+		go func() { _, err := c.GetNodes(joined); done <- err }()
+		awaitSessionSignal(t, joined.joined)
+	}
+	close(release)
+	for i := 0; i < readers; i++ {
+		if err := <-done; err == nil {
+			t.Error("joined expired read advanced past unusable login success")
+		} else if strings.Contains(err.Error(), "fixture-sensitive") {
+			t.Error("failed renewal error contains session bytes")
+		}
+	}
+	c.authMu.Lock()
+	unchanged := reflect.DeepEqual(c.auth, before)
+	unchanged = unchanged && c.authGeneration == generation
+	c.authMu.Unlock()
+	if !unchanged || logins.Load() != 2 || reads.Load() != 0 || forms.Load() != 0 {
+		t.Errorf("invalid success changed session or dispatched work: same=%t auth/API/form=%d/%d/%d", unchanged, logins.Load(), reads.Load(), forms.Load())
+	}
+	if _, err := c.GetNodes(ctx); err != nil {
+		t.Fatal(err)
+	}
+	resp, err := c.request(ctx, http.MethodPost, "/synthetic-read", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = io.Copy(io.Discard, resp.Body)
+	resp.Body.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if logins.Load() != 3 || reads.Load() != 2 || forms.Load() != 0 {
+		t.Errorf("later ordinary recovery auth/API/form=%d/%d/%d", logins.Load(), reads.Load(), forms.Load())
+	}
+	t.Log("eight joined failures, zero API/form dispatches, retained tuple; one later login plus coherent ordinary GET/POST recovery")
+}