diff --git a/internal/monitoring/pve_session_recovery_test.go b/internal/monitoring/pve_session_recovery_test.go new file mode 100644 index 0000000000..fc8d0b8ab8 --- /dev/null +++ b/internal/monitoring/pve_session_recovery_test.go @@ -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) +} diff --git a/pkg/proxmox/auth_session_concurrency_test.go b/pkg/proxmox/auth_session_concurrency_test.go new file mode 100644 index 0000000000..cd83e654ec --- /dev/null +++ b/pkg/proxmox/auth_session_concurrency_test.go @@ -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()) + } +} diff --git a/pkg/proxmox/client.go b/pkg/proxmox/client.go index 47f6a75a43..9613d2cca2 100644 --- a/pkg/proxmox/client.go +++ b/pkg/proxmox/client.go @@ -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