mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-15 19:00:26 +00:00
Share Proxmox session renewal across concurrent polling reads
Publish ticket, CSRF and expiry as one locked snapshot, and bind expiry/401 renewal to the rejected session generation. Concurrent readers share one authentication attempt while retaining their own cancellation and ordinary single-retry behaviour. Keep guest commands single-attempt and token/TLS policy unchanged. Contract-Neutral: Synchronise existing Proxmox password-session renewal without changing API fields, TLS policy or guest-command admission and replay rules. Change-source: pulse-maintainer
This commit is contained in:
parent
0d8ade2b3b
commit
97e4c8eac7
3 changed files with 506 additions and 19 deletions
107
internal/monitoring/pve_session_recovery_test.go
Normal file
107
internal/monitoring/pve_session_recovery_test.go
Normal file
|
|
@ -0,0 +1,107 @@
|
|||
package monitoring
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"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/internal/unifiedresources"
|
||||
"github.com/rcourtman/pulse-go-rewrite/pkg/proxmox"
|
||||
)
|
||||
|
||||
// Exercise the changed password client through the real parallel node poller,
|
||||
// registry and frontend projection. No native QGA/SSH, backup or database is
|
||||
// used. The emitted synthetic resources also feed the retained pixel proof.
|
||||
func TestPVESessionRenewalProducesTrustedNodeReadback(t *testing.T) {
|
||||
const count = 8
|
||||
var authCalls, rejectedStatus, acceptedStatus, agentCalls atomic.Int32
|
||||
allOldReads := make(chan struct{})
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
||||
defer cancel()
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
switch {
|
||||
case strings.HasSuffix(r.URL.Path, "/access/ticket"):
|
||||
n := authCalls.Add(1)
|
||||
fmt.Fprintf(w, `{"data":{"ticket":"synthetic-session-%d","CSRFPreventionToken":"synthetic-csrf-%d"}}`, n, n)
|
||||
case r.URL.Path == "/api2/json/nodes":
|
||||
nodes := make([]proxmox.Node, count)
|
||||
for i := range nodes {
|
||||
nodes[i] = proxmox.Node{Node: fmt.Sprintf("session-node-%02d", i+1), Status: "online", CPU: .05, MaxCPU: 4, MaxMem: 2 << 30, Mem: 1 << 30, MaxDisk: 100 << 30, Disk: 25 << 30, Uptime: 7200}
|
||||
}
|
||||
json.NewEncoder(w).Encode(map[string]any{"data": nodes})
|
||||
case strings.HasSuffix(r.URL.Path, "/status"):
|
||||
cookie, err := r.Cookie("PVEAuthCookie")
|
||||
if err != nil {
|
||||
t.Error("status missing synthetic session")
|
||||
w.WriteHeader(401)
|
||||
return
|
||||
}
|
||||
if cookie.Value == "synthetic-session-1" {
|
||||
if rejectedStatus.Add(1) == count {
|
||||
close(allOldReads)
|
||||
}
|
||||
select {
|
||||
case <-allOldReads:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
fmt.Fprint(w, `{"message":"synthetic expired session"}`)
|
||||
return
|
||||
}
|
||||
acceptedStatus.Add(1)
|
||||
fmt.Fprint(w, `{"data":{"cpu":0.05,"memory":{"total":2147483648,"used":536870912,"free":1073741824,"available":1610612736},"rootfs":{"total":107374182400,"used":26843545600,"free":80530636800},"uptime":7200,"loadavg":[0.1,0.1,0.1],"kversion":"fixture-kernel","pveversion":"pve-manager/9.2.21/fixture","cpuinfo":{"model":"Synthetic CPU","cores":4,"sockets":1,"mhz":"2000"}}}`)
|
||||
case strings.Contains(r.URL.Path, "/agent/"):
|
||||
agentCalls.Add(1)
|
||||
t.Error("node recovery queued guest work")
|
||||
w.WriteHeader(500)
|
||||
default:
|
||||
fmt.Fprint(w, `{"data":[]}`) // ordinary network/apt inventory only
|
||||
}
|
||||
}))
|
||||
defer server.Close()
|
||||
client, err := proxmox.NewClient(proxmox.ClientConfig{Host: server.URL, User: "synthetic@pam", Password: "synthetic-password", Timeout: 5 * time.Second})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rawNodes, err := client.GetNodes(ctx)
|
||||
if err != nil || len(rawNodes) != count {
|
||||
t.Fatalf("ordinary node discovery: %v", err)
|
||||
}
|
||||
instance := config.PVEInstance{Name: "session-fixture", Host: server.URL, IsCluster: true, ClusterName: "Session recovery"}
|
||||
m := &Monitor{config: &config.Config{PVEInstances: []config.PVEInstance{instance}}, nodeLastOnline: make(map[string]time.Time)}
|
||||
nodes, statuses, _ := m.pollPVENodesParallel(ctx, instance.Name, &instance, client, rawNodes, "healthy", nil, false)
|
||||
if len(nodes) != count || len(statuses) != count || authCalls.Load() != 2 || rejectedStatus.Load() != count || acceptedStatus.Load() != count || agentCalls.Load() != 0 {
|
||||
t.Fatalf("node recovery counts nodes/statuses/auth/rejected/accepted/guest=%d/%d/%d/%d/%d/%d", len(nodes), len(statuses), authCalls.Load(), rejectedStatus.Load(), acceptedStatus.Load(), agentCalls.Load())
|
||||
}
|
||||
for _, node := range nodes {
|
||||
if node.Status != "online" || node.ConnectionHealth != "healthy" || !node.Memory.HasKnownUsage() || node.Memory.Usage != 25 || node.Disk.Usage != 25 {
|
||||
t.Fatalf("session recovery lost trusted ordinary fields: %+v", node)
|
||||
}
|
||||
}
|
||||
registry := unifiedresources.NewRegistry(nil)
|
||||
registry.IngestSnapshot(models.StateSnapshot{Nodes: nodes})
|
||||
resources, _ := convertPresentationResourcesForBroadcast(registry.ListForPresentation())
|
||||
if len(resources) != count {
|
||||
t.Fatal("node projection lost recovered nodes")
|
||||
}
|
||||
for _, resource := range resources {
|
||||
if resource.Memory == nil || resource.Memory.Current != 25 || resource.CPU == nil || resource.CPU.Current != 5 {
|
||||
t.Fatalf("rendered projection changed cache-aware memory/CPU: %+v", resource)
|
||||
}
|
||||
}
|
||||
payload, err := json.Marshal(map[string]any{"resources": resources, "authentications": authCalls.Load(), "rejectedStatusReads": rejectedStatus.Load(), "acceptedStatusReads": acceptedStatus.Load(), "guestCommands": agentCalls.Load()})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Logf("PULSE_SESSION_RECOVERY_JSON=%s", payload)
|
||||
}
|
||||
325
pkg/proxmox/auth_session_concurrency_test.go
Normal file
325
pkg/proxmox/auth_session_concurrency_test.go
Normal file
|
|
@ -0,0 +1,325 @@
|
|||
package proxmox
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// All reads have already left with the same rejected session before renewal
|
||||
// can start. Late 401s must reuse the new session, not invalidate it again.
|
||||
// The fixture issues synthetic tickets only and never queues guest work.
|
||||
func TestClientSessionConcurrent401(t *testing.T) {
|
||||
const readers = 16
|
||||
var authentications, oldRequests, newRequests, mismatchedPairs atomic.Int32
|
||||
oldArrived := make(chan struct{})
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if strings.HasSuffix(r.URL.Path, "/access/ticket") {
|
||||
n := authentications.Add(1)
|
||||
fmt.Fprintf(w, `{"data":{"ticket":"session-%d","CSRFPreventionToken":"csrf-%d"}}`, n, n)
|
||||
return
|
||||
}
|
||||
cookie, err := r.Cookie("PVEAuthCookie")
|
||||
if err != nil {
|
||||
t.Error("request missing synthetic session")
|
||||
w.WriteHeader(401)
|
||||
return
|
||||
}
|
||||
if cookie.Value == "session-1" {
|
||||
if oldRequests.Add(1) == readers {
|
||||
close(oldArrived)
|
||||
}
|
||||
select {
|
||||
case <-oldArrived:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
fmt.Fprint(w, `{"message":"synthetic session expired"}`)
|
||||
return
|
||||
}
|
||||
if r.Method != http.MethodGet {
|
||||
n := strings.TrimPrefix(cookie.Value, "session-")
|
||||
if r.Header.Get("CSRFPreventionToken") != "csrf-"+n {
|
||||
mismatchedPairs.Add(1)
|
||||
w.WriteHeader(403)
|
||||
return
|
||||
}
|
||||
}
|
||||
newRequests.Add(1)
|
||||
fmt.Fprint(w, `{"data":[{"node":"session-safe-node","status":"online","cpu":0.05,"maxcpu":4,"mem":268435456,"maxmem":2147483648}]}`)
|
||||
}))
|
||||
defer server.Close()
|
||||
c, err := NewClient(ClientConfig{Host: server.URL, User: "synthetic@pam", Password: "synthetic-password", Timeout: 5 * time.Second})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer c.httpClient.CloseIdleConnections()
|
||||
errors := make(chan error, readers)
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < readers; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
nodes, err := c.GetNodes(ctx)
|
||||
if err == nil && (len(nodes) != 1 || nodes[0].Node != "session-safe-node" || nodes[0].Status != "online") {
|
||||
err = fmt.Errorf("renewed session lost ordinary node result")
|
||||
}
|
||||
errors <- err
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
close(errors)
|
||||
for err := range errors {
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}
|
||||
// Check ticket/CSRF pairing through the same production request path.
|
||||
for i := 0; i < readers; i++ {
|
||||
resp, err := c.request(ctx, http.MethodPost, "/synthetic-read", nil)
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
continue
|
||||
}
|
||||
_, err = readResponseBodyLimited(resp.Body)
|
||||
resp.Body.Close()
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}
|
||||
if got := authentications.Load(); got != 2 {
|
||||
t.Errorf("session-renewal stampede: authentications=%d, want initial plus one renewal", got)
|
||||
}
|
||||
if oldRequests.Load() != readers || newRequests.Load() != 2*readers || mismatchedPairs.Load() != 0 {
|
||||
t.Errorf("old/new/pair-mismatch=%d/%d/%d, want %d/%d/0", oldRequests.Load(), newRequests.Load(), mismatchedPairs.Load(), readers, 2*readers)
|
||||
}
|
||||
t.Log("ordinary reads=" + strconv.Itoa(readers) + "; one shared renewal required; coherent ticket/CSRF pairs checked")
|
||||
}
|
||||
|
||||
// Done signals the exact point at which a caller joins a pending renewal.
|
||||
// There is no sleep or production-only instrumentation in these controls.
|
||||
type authJoinContext struct {
|
||||
context.Context
|
||||
joined chan struct{}
|
||||
once sync.Once
|
||||
}
|
||||
|
||||
func (c *authJoinContext) Done() <-chan struct{} {
|
||||
c.once.Do(func() { close(c.joined) })
|
||||
return c.Context.Done()
|
||||
}
|
||||
|
||||
func awaitSessionSignal(t *testing.T, signal <-chan struct{}) {
|
||||
t.Helper()
|
||||
select {
|
||||
case <-signal:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("session fixture barrier did not complete")
|
||||
}
|
||||
}
|
||||
|
||||
func expireSessionFixture(c *Client) {
|
||||
c.authMu.Lock()
|
||||
c.auth.expiresAt = time.Now().Add(-time.Hour)
|
||||
c.authMu.Unlock()
|
||||
}
|
||||
|
||||
func TestClientSessionConcurrentExpiryAndFailureRecovery(t *testing.T) {
|
||||
for _, failRenewal := range []bool{false, true} {
|
||||
t.Run(fmt.Sprintf("failure=%t", failRenewal), func(t *testing.T) {
|
||||
const readers = 8
|
||||
entered, finish := make(chan struct{}), make(chan struct{})
|
||||
var authCalls, apiCalls atomic.Int32
|
||||
var fail atomic.Bool
|
||||
fail.Store(failRenewal)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if strings.HasSuffix(r.URL.Path, "/access/ticket") {
|
||||
n := authCalls.Add(1)
|
||||
if n == 2 {
|
||||
close(entered)
|
||||
select {
|
||||
case <-finish:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
if fail.Load() {
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
fmt.Fprint(w, "synthetic renewal unavailable")
|
||||
return
|
||||
}
|
||||
}
|
||||
fmt.Fprintf(w, `{"data":{"ticket":"session-%d","CSRFPreventionToken":"csrf-%d"}}`, n, n)
|
||||
return
|
||||
}
|
||||
apiCalls.Add(1)
|
||||
fmt.Fprint(w, `{"data":[{"node":"session-safe-node","status":"online"}]}`)
|
||||
}))
|
||||
defer server.Close()
|
||||
c, err := NewClient(ClientConfig{Host: server.URL, User: "synthetic@pam", Password: "synthetic-password", Timeout: 5 * time.Second})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer c.httpClient.CloseIdleConnections()
|
||||
expireSessionFixture(c)
|
||||
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)
|
||||
}
|
||||
if authCalls.Load() != 2 || apiCalls.Load() != 0 {
|
||||
t.Errorf("expired readers bypassed shared renewal: auth/api=%d/%d", authCalls.Load(), apiCalls.Load())
|
||||
}
|
||||
close(finish)
|
||||
for i := 0; i < readers; i++ {
|
||||
err := <-done
|
||||
if failRenewal {
|
||||
if err == nil || !strings.Contains(err.Error(), "synthetic renewal unavailable") {
|
||||
t.Errorf("renewal failure not shared: %v", err)
|
||||
}
|
||||
} else if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}
|
||||
if authCalls.Load() != 2 {
|
||||
t.Error("joined callers retried completed renewal")
|
||||
}
|
||||
if failRenewal {
|
||||
snapshot, _ := c.authSnapshot()
|
||||
if snapshot.ticket != "session-1" || snapshot.csrfToken != "csrf-1" || !snapshot.expiresAt.Before(time.Now()) {
|
||||
t.Error("failed renewal published or refreshed the old session")
|
||||
}
|
||||
if apiCalls.Load() != 0 {
|
||||
t.Error("failed authentication admitted API reads")
|
||||
}
|
||||
// A new ordinary call may recover; the failed attempt is not
|
||||
// automatically replayed by any of its joined callers.
|
||||
fail.Store(false)
|
||||
if nodes, err := c.GetNodes(ctx); err != nil || len(nodes) != 1 {
|
||||
t.Errorf("ordinary later recovery failed: %v", err)
|
||||
}
|
||||
if authCalls.Load() != 3 || apiCalls.Load() != 1 {
|
||||
t.Errorf("later recovery counts auth/api=%d/%d", authCalls.Load(), apiCalls.Load())
|
||||
}
|
||||
} else if apiCalls.Load() != readers {
|
||||
t.Errorf("renewed session lost %d reads", readers-apiCalls.Load())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientSessionRenewalWaiterCancellation(t *testing.T) {
|
||||
entered, finish := make(chan struct{}), make(chan struct{})
|
||||
var authCalls, apiCalls atomic.Int32
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if strings.HasSuffix(r.URL.Path, "/access/ticket") {
|
||||
n := authCalls.Add(1)
|
||||
if n == 2 {
|
||||
close(entered)
|
||||
select {
|
||||
case <-finish:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
fmt.Fprintf(w, `{"data":{"ticket":"session-%d","CSRFPreventionToken":"csrf-%d"}}`, n, n)
|
||||
return
|
||||
}
|
||||
apiCalls.Add(1)
|
||||
fmt.Fprint(w, `{"data":[]}`)
|
||||
}))
|
||||
defer server.Close()
|
||||
c, err := NewClient(ClientConfig{Host: server.URL, User: "synthetic@pam", Password: "synthetic-password", Timeout: 5 * time.Second})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer c.httpClient.CloseIdleConnections()
|
||||
expireSessionFixture(c)
|
||||
owner := make(chan error, 1)
|
||||
go func() { _, err := c.GetNodes(ctx); owner <- err }()
|
||||
awaitSessionSignal(t, entered)
|
||||
waiterCtx, cancelWaiter := context.WithCancel(ctx)
|
||||
defer cancelWaiter()
|
||||
joined := &authJoinContext{Context: waiterCtx, joined: make(chan struct{})}
|
||||
waiter := make(chan error, 1)
|
||||
go func() { _, err := c.GetNodes(joined); waiter <- err }()
|
||||
awaitSessionSignal(t, joined.joined)
|
||||
cancelWaiter()
|
||||
select {
|
||||
case err := <-waiter:
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Errorf("cancelled waiter result: %v", err)
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("cancelled waiter remained blocked behind another caller's renewal")
|
||||
}
|
||||
if authCalls.Load() != 2 || apiCalls.Load() != 0 {
|
||||
t.Error("cancelled waiter replayed renewal or admitted a read")
|
||||
}
|
||||
select {
|
||||
case err := <-owner:
|
||||
t.Fatalf("waiter cancellation ended the owner's renewal: %v", err)
|
||||
default:
|
||||
}
|
||||
close(finish)
|
||||
if err := <-owner; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := c.GetNodes(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if authCalls.Load() != 2 || apiCalls.Load() != 2 {
|
||||
t.Errorf("owner/later recovery counts auth/api=%d/%d", authCalls.Load(), apiCalls.Load())
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientSessionTokenAuthDoesNotRenew(t *testing.T) {
|
||||
var authCalls, apiCalls atomic.Int32
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if strings.HasSuffix(r.URL.Path, "/access/ticket") {
|
||||
authCalls.Add(1)
|
||||
w.WriteHeader(500)
|
||||
return
|
||||
}
|
||||
apiCalls.Add(1)
|
||||
if r.Header.Get("Authorization") != "PVEAPIToken=synthetic@pve!pulse=synthetic-token" || r.Header.Get("Cookie") != "" || r.Header.Get("CSRFPreventionToken") != "" {
|
||||
t.Error("token read acquired session headers")
|
||||
}
|
||||
fmt.Fprint(w, `{"data":[]}`)
|
||||
}))
|
||||
defer server.Close()
|
||||
c, err := NewClient(ClientConfig{Host: server.URL, TokenName: "synthetic@pve!pulse", TokenValue: "synthetic-token", Password: "unused-synthetic-password", Timeout: time.Second})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer c.httpClient.CloseIdleConnections()
|
||||
done := make(chan error, 16)
|
||||
for i := 0; i < 16; i++ {
|
||||
go func() { _, err := c.GetNodes(context.Background()); done <- err }()
|
||||
}
|
||||
for i := 0; i < 16; i++ {
|
||||
if err := <-done; err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}
|
||||
if authCalls.Load() != 0 || apiCalls.Load() != 16 {
|
||||
t.Errorf("token auth/API calls=%d/%d", authCalls.Load(), apiCalls.Load())
|
||||
}
|
||||
}
|
||||
|
|
@ -11,6 +11,7 @@ import (
|
|||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
"unicode"
|
||||
|
||||
|
|
@ -255,6 +256,24 @@ type Client struct {
|
|||
httpClient *http.Client
|
||||
auth auth
|
||||
config ClientConfig
|
||||
|
||||
// A polling client is shared by concurrent node and guest reads. Publish
|
||||
// each ticket/CSRF/expiry tuple atomically and share only session renewal,
|
||||
// not the subsequent API requests or their caller-owned cancellation.
|
||||
authMu sync.Mutex
|
||||
authGeneration uint64
|
||||
authInFlight *authRefresh
|
||||
}
|
||||
|
||||
type authRefresh struct {
|
||||
done chan struct{}
|
||||
err error
|
||||
}
|
||||
|
||||
func (c *Client) authSnapshot() (auth, uint64) {
|
||||
c.authMu.Lock()
|
||||
defer c.authMu.Unlock()
|
||||
return c.auth, c.authGeneration
|
||||
}
|
||||
|
||||
func (c *Client) apiBaseURL() (*url.URL, error) {
|
||||
|
|
@ -398,16 +417,47 @@ func NewClient(cfg ClientConfig) (*Client, error) {
|
|||
|
||||
// authenticate performs password-based authentication
|
||||
func (c *Client) authenticate(ctx context.Context) error {
|
||||
username := c.auth.user + "@" + c.auth.realm
|
||||
password := c.config.Password
|
||||
return c.renewAuthentication(ctx, nil)
|
||||
}
|
||||
|
||||
if err := c.authenticateJSON(ctx, username, password); err == nil {
|
||||
return nil
|
||||
} else if shouldFallbackToForm(err) {
|
||||
return c.authenticateForm(ctx, username, password)
|
||||
} else {
|
||||
// expected binds expiry or a 401 to the session actually used by that read.
|
||||
// A late rejection of an old ticket cannot trigger another renewal after a
|
||||
// concurrent request has already replaced it. Explicit authentication (nil)
|
||||
// still owns the existing JSON/form fallback and failed-attempt behaviour.
|
||||
func (c *Client) renewAuthentication(ctx context.Context, expected *uint64) error {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
c.authMu.Lock()
|
||||
if expected != nil && c.authGeneration != *expected {
|
||||
c.authMu.Unlock()
|
||||
return nil
|
||||
}
|
||||
if pending := c.authInFlight; pending != nil {
|
||||
c.authMu.Unlock()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-pending.done:
|
||||
return pending.err
|
||||
}
|
||||
}
|
||||
pending := &authRefresh{done: make(chan struct{})}
|
||||
c.authInFlight = pending
|
||||
username := c.auth.user + "@" + c.auth.realm
|
||||
c.authMu.Unlock()
|
||||
|
||||
password := c.config.Password
|
||||
err := c.authenticateJSON(ctx, username, password)
|
||||
if shouldFallbackToForm(err) {
|
||||
err = c.authenticateForm(ctx, username, password)
|
||||
}
|
||||
c.authMu.Lock()
|
||||
pending.err = err
|
||||
c.authInFlight = nil
|
||||
close(pending.done)
|
||||
c.authMu.Unlock()
|
||||
return err
|
||||
}
|
||||
|
||||
func (c *Client) authenticateJSON(ctx context.Context, username, password string) error {
|
||||
|
|
@ -488,9 +538,12 @@ func (c *Client) handleAuthResponse(resp *http.Response) error {
|
|||
return err
|
||||
}
|
||||
|
||||
c.authMu.Lock()
|
||||
c.auth.ticket = result.Data.Ticket
|
||||
c.auth.csrfToken = result.Data.CSRFPreventionToken
|
||||
c.auth.expiresAt = time.Now().Add(2 * time.Hour) // PVE tickets expire after 2 hours
|
||||
c.authGeneration++
|
||||
c.authMu.Unlock()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
@ -528,10 +581,12 @@ func (c *Client) requestWithRetry(ctx context.Context, method, path string, data
|
|||
|
||||
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 {
|
||||
auth, generation := c.authSnapshot()
|
||||
if c.config.Password != "" && auth.tokenName == "" && time.Now().After(auth.expiresAt) {
|
||||
if err := c.renewAuthentication(ctx, &generation); err != nil {
|
||||
return nil, fmt.Errorf("re-authentication failed: %w", err)
|
||||
}
|
||||
auth, generation = c.authSnapshot()
|
||||
}
|
||||
|
||||
var body io.Reader
|
||||
|
|
@ -554,23 +609,23 @@ func (c *Client) requestWithRetryUsingClient(ctx context.Context, method, path s
|
|||
}
|
||||
|
||||
// Set authentication
|
||||
if c.auth.tokenName != "" && c.auth.tokenValue != "" {
|
||||
if auth.tokenName != "" && auth.tokenValue != "" {
|
||||
// API token authentication
|
||||
authHeader := fmt.Sprintf("PVEAPIToken=%s@%s!%s=%s",
|
||||
c.auth.user, c.auth.realm, c.auth.tokenName, c.auth.tokenValue)
|
||||
auth.user, auth.realm, auth.tokenName, auth.tokenValue)
|
||||
req.Header.Set("Authorization", authHeader)
|
||||
// NEVER log the actual token value - only log that we're using token auth
|
||||
maskedHeader := fmt.Sprintf("PVEAPIToken=%s@%s!%s=***",
|
||||
c.auth.user, c.auth.realm, c.auth.tokenName)
|
||||
auth.user, auth.realm, auth.tokenName)
|
||||
log.Debug().
|
||||
Str("authHeader", maskedHeader).
|
||||
Str("url", req.URL.String()).
|
||||
Msg("Setting API token authentication")
|
||||
} else if c.auth.ticket != "" {
|
||||
} else if auth.ticket != "" {
|
||||
// Ticket authentication
|
||||
req.Header.Set("Cookie", "PVEAuthCookie="+c.auth.ticket)
|
||||
if method != "GET" && c.auth.csrfToken != "" {
|
||||
req.Header.Set("CSRFPreventionToken", c.auth.csrfToken)
|
||||
req.Header.Set("Cookie", "PVEAuthCookie="+auth.ticket)
|
||||
if method != "GET" && auth.csrfToken != "" {
|
||||
req.Header.Set("CSRFPreventionToken", auth.csrfToken)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -589,8 +644,8 @@ func (c *Client) requestWithRetryUsingClient(ctx context.Context, method, path s
|
|||
|
||||
// Password sessions can be invalidated server-side before local expiry.
|
||||
// Retry once after forcing re-authentication on 401.
|
||||
if resp.StatusCode == http.StatusUnauthorized && !retriedAfter401 && c.config.Password != "" && c.auth.tokenName == "" {
|
||||
if err := c.authenticate(ctx); err != nil {
|
||||
if resp.StatusCode == http.StatusUnauthorized && !retriedAfter401 && c.config.Password != "" && auth.tokenName == "" {
|
||||
if err := c.renewAuthentication(ctx, &generation); err != nil {
|
||||
return nil, fmt.Errorf("re-authentication failed after 401: %w", err)
|
||||
}
|
||||
return c.requestWithRetryUsingClient(ctx, method, path, data, true, httpClient)
|
||||
|
|
@ -600,7 +655,7 @@ func (c *Client) requestWithRetryUsingClient(ctx context.Context, method, path s
|
|||
var apiErr error
|
||||
if resp.StatusCode == 403 && c.config.TokenName != "" {
|
||||
// Special case for 403 with API token - this is usually a permission issue
|
||||
apiErr = fmt.Errorf("API error 403 (Forbidden): The API token does not have sufficient permissions. Note: In Proxmox GUI, permissions must be set on the USER (not just the token). Please verify the user '%s@%s' has the required permissions", c.auth.user, c.auth.realm)
|
||||
apiErr = fmt.Errorf("API error 403 (Forbidden): The API token does not have sufficient permissions. Note: In Proxmox GUI, permissions must be set on the USER (not just the token). Please verify the user '%s@%s' has the required permissions", auth.user, auth.realm)
|
||||
} else if resp.StatusCode == 595 {
|
||||
// 595 can mean authentication failed OR trying to access an offline node in a cluster
|
||||
// Check if this is a node-specific endpoint
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue