diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index e47eebf2a5..70935ac59b 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -42,9 +42,25 @@ through normalized authorities, invalid nonempty pins and a capture race. `TestClusterConflictingPinsEndpointProjection` uses the production endpoint builder, real clients and monitor projection in all three auth modes: native Online, identity and saved trust are not changed by failed Pulse reachability. +The producer also records contradictory exact duplicate saved rows before +its map/deduplication can overwrite them, and carries those endpoint conflicts +through initialisation, missing-client reconnect and cluster refresh. The +cluster owns a copy of that conflict set; aliases cannot escape it, and a +later repeated row cannot clear it. Normal configuration reload with agreeing +pins is required to resume that endpoint. A non-conflicting sibling is not +refused. The public constructor's optional conflict inputs preserve existing +callers and do not introduce a new trust mode. Producer tests cover both +routes, pin formatting, repeated contradictions and configuration ownership. +The projection cases include exact duplicates as well as URL aliases. +A credential-free agreement check also holds automatic topology refresh while +saved trust is contradictory: discovery cannot collapse the rows, move an +alias or substitute a pin to clear that refusal. Current trusted sibling +clients keep monitoring; topology changes wait for independently reconciled +configuration. `TestClusterPinConflictsSurviveMonitorLifecycle` exercises the +actual initialisation, missing-client retry and refresh/rebuild paths in all +three auth modes and both layouts, including normal refresh after reconciliation. This establishes source/consumer behaviour, not native recovery or attribution -of a guest, PBS or certificate report. Exact duplicate saved rows collapsed by -the builder's map are outside this URL-alias resolver contract. +of a guest, PBS or certificate report. ### Cluster recovery requires success of the operation actually probed diff --git a/docs/release-control/v6/internal/subsystems/security-privacy.md b/docs/release-control/v6/internal/subsystems/security-privacy.md index 339264d843..d9f14d1346 100644 --- a/docs/release-control/v6/internal/subsystems/security-privacy.md +++ b/docs/release-control/v6/internal/subsystems/security-privacy.md @@ -43,9 +43,25 @@ through normalized authorities, invalid nonempty pins and a capture race. `TestClusterConflictingPinsEndpointProjection` uses the production endpoint builder, real clients and monitor projection in all three auth modes: native Online, identity and saved trust are not changed by failed Pulse reachability. +The producer also records contradictory exact duplicate saved rows before +its map/deduplication can overwrite them, and carries those endpoint conflicts +through initialisation, missing-client reconnect and cluster refresh. The +cluster owns a copy of that conflict set; aliases cannot escape it, and a +later repeated row cannot clear it. Normal configuration reload with agreeing +pins is required to resume that endpoint. A non-conflicting sibling is not +refused. The public constructor's optional conflict inputs preserve existing +callers and do not introduce a new trust mode. Producer tests cover both +routes, pin formatting, repeated contradictions and configuration ownership. +The projection cases include exact duplicates as well as URL aliases. +A credential-free agreement check also holds automatic topology refresh while +saved trust is contradictory: discovery cannot collapse the rows, move an +alias or substitute a pin to clear that refusal. Current trusted sibling +clients keep monitoring; topology changes wait for independently reconciled +configuration. `TestClusterPinConflictsSurviveMonitorLifecycle` exercises the +actual initialisation, missing-client retry and refresh/rebuild paths in all +three auth modes and both layouts, including normal refresh after reconciliation. This establishes source/consumer behaviour, not native recovery or attribution -of a guest, PBS or certificate report. Exact duplicate saved rows collapsed by -the builder's map are outside this URL-alias resolver contract. +of a guest, PBS or certificate report. ### First-use cluster trust includes password construction diff --git a/internal/monitoring/monitor_additional_test.go b/internal/monitoring/monitor_additional_test.go index a2622b436a..25058ff5bc 100644 --- a/internal/monitoring/monitor_additional_test.go +++ b/internal/monitoring/monitor_additional_test.go @@ -335,7 +335,7 @@ func TestBuildClusterEndpointsForInit_RespectsDiscoveryPolicy(t *testing.T) { }, } - endpoints, _ := monitor.buildClusterEndpointsForInit(config.PVEInstance{ + endpoints, _, _ := monitor.buildClusterEndpointsForInit(config.PVEInstance{ Name: "cluster-a", Host: "https://main.local:8006", VerifySSL: true, diff --git a/internal/monitoring/monitor_client_init.go b/internal/monitoring/monitor_client_init.go index 05754727d3..24902da8a8 100644 --- a/internal/monitoring/monitor_client_init.go +++ b/internal/monitoring/monitor_client_init.go @@ -29,7 +29,7 @@ func (m *Monitor) initPVEClients(cfg *config.Config) { // Check if this is a cluster if pve.IsCluster && len(pve.ClusterEndpoints) > 0 { - endpoints, endpointFingerprints := m.buildClusterEndpointsForInit(pve) + endpoints, endpointFingerprints, conflictingEndpoints := m.buildClusterEndpointsForInit(pve) log.Info(). Str("cluster", pve.ClusterName). @@ -44,6 +44,7 @@ func (m *Monitor) initPVEClients(cfg *config.Config) { clientConfig, endpoints, endpointFingerprints, + conflictingEndpoints..., ) m.replacePVEClient(pve.Name, clusterClient) log.Info(). diff --git a/internal/monitoring/monitor_client_reconnect.go b/internal/monitoring/monitor_client_reconnect.go index 7734b21dfb..08cb8bf0f0 100644 --- a/internal/monitoring/monitor_client_reconnect.go +++ b/internal/monitoring/monitor_client_reconnect.go @@ -108,11 +108,11 @@ func (m *Monitor) retryFailedConnections(ctx context.Context) { } if pve.IsCluster && len(pve.ClusterEndpoints) > 0 { // Create cluster client - endpoints, endpointFingerprints := m.buildClusterEndpointsForReconnect(pve) + endpoints, endpointFingerprints, conflictingEndpoints := m.buildClusterEndpointsForReconnect(pve) clientConfig := config.CreateProxmoxConfig(&pve) clientConfig.Timeout = connectionTimeout - clusterClient, err := proxmox.NewClusterClientWithContext(ctx, pve.Name, clientConfig, endpoints, endpointFingerprints) + clusterClient, err := proxmox.NewClusterClientWithContext(ctx, pve.Name, clientConfig, endpoints, endpointFingerprints, conflictingEndpoints...) if ctx.Err() != nil { return } diff --git a/internal/monitoring/monitor_pve_client_endpoints.go b/internal/monitoring/monitor_pve_client_endpoints.go index 6e9394dab3..d8c93084f0 100644 --- a/internal/monitoring/monitor_pve_client_endpoints.go +++ b/internal/monitoring/monitor_pve_client_endpoints.go @@ -1,15 +1,17 @@ package monitoring import ( + "strings" + "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rs/zerolog/log" ) -func (m *Monitor) buildClusterEndpointsForInit(pve config.PVEInstance) ([]string, map[string]string) { +func (m *Monitor) buildClusterEndpointsForInit(pve config.PVEInstance) ([]string, map[string]string, []string) { return m.buildClusterClientEndpoints(pve) } -func (m *Monitor) buildClusterEndpointsForReconnect(pve config.PVEInstance) ([]string, map[string]string) { +func (m *Monitor) buildClusterEndpointsForReconnect(pve config.PVEInstance) ([]string, map[string]string, []string) { return m.buildClusterClientEndpoints(pve) } @@ -18,9 +20,11 @@ func (m *Monitor) buildClusterEndpointsForReconnect(pve config.PVEInstance) ([]s // are valid inside the cluster but unreachable from the Pulse container; those // addresses remain useful failover and reachability evidence, but they must // never displace the URL the operator proved reachable when adding the cluster. -func (m *Monitor) buildClusterClientEndpoints(pve config.PVEInstance) ([]string, map[string]string) { +func (m *Monitor) buildClusterClientEndpoints(pve config.PVEInstance) ([]string, map[string]string, []string) { endpoints := make([]string, 0, len(pve.ClusterEndpoints)+1) endpointFingerprints := make(map[string]string) + var conflictingEndpoints []string + conflicts := make(map[string]bool) discoveryCfg := monitorDiscoveryConfig(m) hasFingerprint := pve.Fingerprint != "" seen := make(map[string]struct{}, len(pve.ClusterEndpoints)+1) @@ -53,7 +57,17 @@ func (m *Monitor) buildClusterClientEndpoints(pve config.PVEInstance) ([]string, failoverCount++ } if ep.Fingerprint != "" { - endpointFingerprints[host] = ep.Fingerprint + if previous, exists := endpointFingerprints[host]; exists { + if strings.ToLower(strings.ReplaceAll(previous, ":", "")) != strings.ToLower(strings.ReplaceAll(ep.Fingerprint, ":", "")) && !conflicts[host] { + // The map cannot retain two values for the exact same URL. + // Carry the conflict separately; a later repeated row must + // not erase it or choose the certificate to trust. + conflicts[host] = true + conflictingEndpoints = append(conflictingEndpoints, host) + } + } else { + endpointFingerprints[host] = ep.Fingerprint + } } } @@ -71,5 +85,5 @@ func (m *Monitor) buildClusterClientEndpoints(pve config.PVEInstance) ([]string, Msg("Configured Proxmox cluster URL is empty; using first discovered member as API authority") } - return endpoints, endpointFingerprints + return endpoints, endpointFingerprints, conflictingEndpoints } diff --git a/internal/monitoring/monitor_pve_cluster.go b/internal/monitoring/monitor_pve_cluster.go index 965cf94905..9426470adc 100644 --- a/internal/monitoring/monitor_pve_cluster.go +++ b/internal/monitoring/monitor_pve_cluster.go @@ -514,6 +514,16 @@ func (m *Monitor) detectClusterMembership(ctx context.Context, instanceName stri // node addresses changed, the config is persisted and the failover client is // rebuilt so polling stops dialing the dead addresses. func (m *Monitor) refreshClusterEndpoints(instanceName string, instanceCfg *config.PVEInstance, clusterName string, discovered []config.ClusterEndpoint) { + // Membership discovery is not a trust decision. In particular it can + // collapse duplicate saved rows, or move an alias away from its conflicting + // authority. Keep the saved evidence and current sibling clients until an + // operator independently reconciles trust through normal configuration. + _, savedPins, conflicts := m.buildClusterClientEndpoints(*instanceCfg) + if err := proxmox.CheckEndpointFingerprintConflicts(savedPins, conflicts...); err != nil { + log.Warn().Str("instance", instanceName).Err(err). + Msg("Retaining cluster endpoints until conflicting saved TLS trust is reconciled") + return + } refreshed := mergeRefreshedClusterEndpoints(instanceCfg.ClusterEndpoints, discovered, instanceCfg.VerifySSL) if !clusterEndpointIdentityChanged(instanceCfg.ClusterEndpoints, refreshed) { return @@ -700,10 +710,10 @@ func (m *Monitor) rebuildPVEClusterClient(instanceName string) { return } - endpoints, endpointFingerprints := m.buildClusterEndpointsForReconnect(*pve) + endpoints, endpointFingerprints, conflictingEndpoints := m.buildClusterEndpointsForReconnect(*pve) clientConfig := config.CreateProxmoxConfig(pve) clientConfig.Timeout = m.config.ConnectionTimeout - clusterClient := proxmox.NewClusterClient(pve.Name, clientConfig, endpoints, endpointFingerprints) + clusterClient := proxmox.NewClusterClient(pve.Name, clientConfig, endpoints, endpointFingerprints, conflictingEndpoints...) m.replacePVEClient(instanceName, clusterClient) diff --git a/internal/monitoring/monitor_pve_cluster_refresh_test.go b/internal/monitoring/monitor_pve_cluster_refresh_test.go index b2edb65796..4693ae025f 100644 --- a/internal/monitoring/monitor_pve_cluster_refresh_test.go +++ b/internal/monitoring/monitor_pve_cluster_refresh_test.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "reflect" + "strings" "testing" "time" @@ -362,13 +363,16 @@ func TestBuildClusterEndpoints_PreservesConfiguredAuthority(t *testing.T) { for _, build := range []struct { name string - fn func(config.PVEInstance) ([]string, map[string]string) + fn func(config.PVEInstance) ([]string, map[string]string, []string) }{ {name: "initialization", fn: monitor.buildClusterEndpointsForInit}, {name: "reconnect", fn: monitor.buildClusterEndpointsForReconnect}, } { t.Run(build.name, func(t *testing.T) { - endpoints, fingerprints := build.fn(instance) + endpoints, fingerprints, conflicts := build.fn(instance) + if len(conflicts) != 0 { + t.Fatalf("non-conflicting member trust was refused: %v", conflicts) + } if len(endpoints) != len(want) { t.Fatalf("expected %d ordered endpoints, got %v", len(want), endpoints) } @@ -422,3 +426,49 @@ func TestStartupClientConstructionHonoursCancellation(t *testing.T) { func TestPBSStartupCancellationKeepsFirstPollBoundary(t *testing.T) { testPBSStartupCancellationKeepsFirstPollBoundary(t) } + +func TestBuildClusterEndpointsRetainsDuplicatePinConflicts(t *testing.T) { + monitor := &Monitor{config: &config.Config{}} + const member = "https://member.example.invalid:8006" + for _, build := range []struct { + name string + fn func(config.PVEInstance) ([]string, map[string]string, []string) + }{ + {"init", monitor.buildClusterEndpointsForInit}, {"reconnect", monitor.buildClusterEndpointsForReconnect}, + } { + for _, sequence := range [][]string{{"aa:bb", "AA:BB", "aabb"}, {"aa:bb", "cc:dd"}, {"cc:dd", "aa:bb", "cc:dd"}, {"aa:bb", "", "cc:dd", "aa:bb"}, {"", "aa:bb", "cc:dd"}, {"", "aa:bb", "AA:BB"}, {":", ":"}} { + t.Run(build.name+"/"+strings.Join(sequence, "_"), func(t *testing.T) { + inst := config.PVEInstance{Host: "https://primary.example.invalid:8006", VerifySSL: true} + for _, pin := range sequence { + inst.ClusterEndpoints = append(inst.ClusterEndpoints, config.ClusterEndpoint{NodeName: "member", Host: member, Fingerprint: pin}) + } + endpoints, pins, conflicts := build.fn(inst) + wantConflict := false + firstPin := "" + for _, pin := range sequence { + if firstPin == "" { + firstPin = pin + } + if pin != "" && strings.ToLower(strings.ReplaceAll(pin, ":", "")) != strings.ToLower(strings.ReplaceAll(firstPin, ":", "")) { + wantConflict = true + } + } + if len(endpoints) != 2 || pins[member] != firstPin { + t.Fatal("duplicate rows changed routing order or selected later trust") + } + if wantConflict { + if len(conflicts) != 1 || conflicts[0] != member { + t.Fatal("later row erased conflicting certificate evidence") + } + } else if len(conflicts) != 0 { + t.Fatal("agreeing/absent member pins created a conflict") + } + for i, ep := range inst.ClusterEndpoints { + if ep.Fingerprint != sequence[i] || ep.Host != member { + t.Fatal("producer changed saved configuration") + } + } + }) + } + } +} diff --git a/internal/monitoring/monitor_pve_pin_conflict_test.go b/internal/monitoring/monitor_pve_pin_conflict_test.go index 15aef0cbb5..ef634524b0 100644 --- a/internal/monitoring/monitor_pve_pin_conflict_test.go +++ b/internal/monitoring/monitor_pve_pin_conflict_test.go @@ -8,6 +8,7 @@ import ( "fmt" "net/http" "net/http/httptest" + "reflect" "strings" "sync/atomic" "testing" @@ -25,110 +26,235 @@ import ( // deployment/recovery, and says nothing about any reported incident's cause. func TestClusterConflictingPinsEndpointProjection(t *testing.T) { for _, mode := range []string{"token", "password-json", "password-form"} { - t.Run(mode, func(t *testing.T) { - var badRequests atomic.Int32 - endpoint := func(bad bool) *httptest.Server { - s := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if bad { - badRequests.Add(1) - } - switch r.URL.Path { - case "/api2/json/access/ticket": - if mode == "password-form" && r.Header.Get("Content-Type") == "application/json" { - w.WriteHeader(http.StatusUnsupportedMediaType) - fmt.Fprint(w, `{"message":"fixture requires form encoding"}`) - return + for _, layout := range []string{"url-aliases", "exact-duplicates"} { + t.Run(layout+"/"+mode, func(t *testing.T) { + var badRequests atomic.Int32 + endpoint := func(bad bool) *httptest.Server { + s := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if bad { + badRequests.Add(1) } - fmt.Fprint(w, `{"data":{"ticket":"fixture-ticket","CSRFPreventionToken":"fixture-csrf"}}`) - case "/api2/json/nodes": - if mode == "token" { - if r.Header.Get("Authorization") != "PVEAPIToken=fixture@pve!pulse=fixture" { - t.Error("lost token fixture auth") + switch r.URL.Path { + case "/api2/json/access/ticket": + if mode == "password-form" && r.Header.Get("Content-Type") == "application/json" { + w.WriteHeader(http.StatusUnsupportedMediaType) + fmt.Fprint(w, `{"message":"fixture requires form encoding"}`) + return } - } else if r.Header.Get("Cookie") != "PVEAuthCookie=fixture-ticket" { - t.Error("lost fixture session") + fmt.Fprint(w, `{"data":{"ticket":"fixture-ticket","CSRFPreventionToken":"fixture-csrf"}}`) + case "/api2/json/nodes": + if mode == "token" { + if r.Header.Get("Authorization") != "PVEAPIToken=fixture@pve!pulse=fixture" { + t.Error("lost token fixture auth") + } + } else if r.Header.Get("Cookie") != "PVEAuthCookie=fixture-ticket" { + t.Error("lost fixture session") + } + fmt.Fprint(w, `{"data":[{"node":"primary","status":"online","cpu":0.125},{"node":"member","status":"online","cpu":0.25},{"node":"member-alias","status":"online","cpu":0.25}]}`) + default: + t.Errorf("left ordinary login/nodes path: %s", r.URL.Path) + http.NotFound(w, r) + } + })) + t.Cleanup(s.Close) + return s + } + primary, member := endpoint(false), endpoint(true) + pin := func(s *httptest.Server) string { + sum := sha256.Sum256(s.Certificate().Raw) + return hex.EncodeToString(sum[:]) + } + inst := config.PVEInstance{Name: "pin-conflict-cluster", Host: primary.URL, IsCluster: true, ClusterName: "Fixture cluster", VerifySSL: true, Fingerprint: pin(primary), ClusterEndpoints: []config.ClusterEndpoint{ + {NodeName: "primary", NodeIdentity: "primary-identity", Host: primary.URL, Fingerprint: pin(primary)}, + {NodeName: "member", NodeIdentity: "member-identity", Host: member.URL, Fingerprint: pin(member)}, + {NodeName: "member-alias", NodeIdentity: "alias-identity", Host: member.URL + "/./", Fingerprint: strings.Repeat("0", 64)}, + }} + if layout == "exact-duplicates" { + inst.ClusterEndpoints[2].Host = member.URL + } + m := &Monitor{config: &config.Config{PVEInstances: []config.PVEInstance{inst}}} + construct := func(conflict bool) *proxmox.ClusterClient { + endpoints, pins, conflicts := m.buildClusterEndpointsForInit(inst) + wantEndpoints, wantConflicts := 3, 0 + if layout == "exact-duplicates" { + wantEndpoints = 2 + if conflict { + wantConflicts = 1 } - fmt.Fprint(w, `{"data":[{"node":"primary","status":"online","cpu":0.125},{"node":"member","status":"online","cpu":0.25},{"node":"member-alias","status":"online","cpu":0.25}]}`) - default: - t.Errorf("left ordinary login/nodes path: %s", r.URL.Path) - http.NotFound(w, r) } - })) - t.Cleanup(s.Close) - return s - } - primary, member := endpoint(false), endpoint(true) - pin := func(s *httptest.Server) string { - sum := sha256.Sum256(s.Certificate().Raw) - return hex.EncodeToString(sum[:]) - } - inst := config.PVEInstance{Name: "pin-conflict-cluster", Host: primary.URL, IsCluster: true, ClusterName: "Fixture cluster", VerifySSL: true, Fingerprint: pin(primary), ClusterEndpoints: []config.ClusterEndpoint{ - {NodeName: "primary", NodeIdentity: "primary-identity", Host: primary.URL, Fingerprint: pin(primary)}, - {NodeName: "member", NodeIdentity: "member-identity", Host: member.URL, Fingerprint: pin(member)}, - {NodeName: "member-alias", NodeIdentity: "alias-identity", Host: member.URL + "/./", Fingerprint: strings.Repeat("0", 64)}, - }} - m := &Monitor{config: &config.Config{PVEInstances: []config.PVEInstance{inst}}} - construct := func() *proxmox.ClusterClient { - endpoints, pins := m.buildClusterEndpointsForInit(inst) - if len(endpoints) != 3 || len(pins) != 3 { - t.Fatalf("saved alias evidence was collapsed: endpoints=%d pins=%d", len(endpoints), len(pins)) - } - cfg := proxmox.ClientConfig{Host: inst.Host, Fingerprint: inst.Fingerprint, VerifySSL: true, Timeout: 2 * time.Second} - if mode == "token" { - cfg.TokenName = "fixture@pve!pulse" - cfg.TokenValue = "fixture" - } else { - cfg.User = "fixture@pam" - cfg.Password = "fixture-password" - } - return proxmox.NewClusterClient(inst.Name, cfg, endpoints, pins) - } - project := func(cc *proxmox.ClusterClient, phase string, conflict bool) { - nodes, err := cc.GetNodes(context.Background()) - if err != nil || len(nodes) != 3 { - t.Fatalf("trusted primary lost ordinary monitoring: %v", err) - } - model := make([]models.Node, len(nodes)) - for i, node := range nodes { - model[i] = models.Node{Name: node.Node, Status: node.Status} - } - m.updateClusterEndpointStatus(inst.Name, &inst, cc, model) - for i, ep := range inst.ClusterEndpoints { - if !ep.Online || ep.NodeIdentity != []string{"primary-identity", "member-identity", "alias-identity"}[i] { - t.Fatal("trust refusal changed native online state or identity") + if len(endpoints) != wantEndpoints || len(pins) != wantEndpoints || len(conflicts) != wantConflicts { + t.Fatalf("saved alias evidence was collapsed: endpoints=%d pins=%d", len(endpoints), len(pins)) } - if ep.PulseReachable == nil || *ep.PulseReachable != (i == 0 || !conflict) || ep.LastPulseCheck == nil { - t.Fatalf("missing/false Pulse reachability observation: %+v", ep) - } - if i > 0 && conflict && !strings.Contains(ep.PulseError, "TLS trust conflict") { - t.Fatalf("member lost actionable trust diagnostic: %+v", ep) - } - if !conflict && ep.PulseError != "" { - t.Fatalf("agreed trust kept a stale failure: %+v", ep) + cfg := proxmox.ClientConfig{Host: inst.Host, Fingerprint: inst.Fingerprint, VerifySSL: true, Timeout: 2 * time.Second} + if mode == "token" { + cfg.TokenName = "fixture@pve!pulse" + cfg.TokenValue = "fixture" + } else { + cfg.User = "fixture@pam" + cfg.Password = "fixture-password" } + return proxmox.NewClusterClient(inst.Name, cfg, endpoints, pins, conflicts...) } - wire, err := json.Marshal(struct { - Case, Phase string - Endpoints []config.ClusterEndpoint - }{mode, phase, inst.ClusterEndpoints}) - if err != nil { - t.Fatal(err) + project := func(cc *proxmox.ClusterClient, phase string, conflict bool) { + nodes, err := cc.GetNodes(context.Background()) + if err != nil || len(nodes) != 3 { + t.Fatalf("trusted primary lost ordinary monitoring: %v", err) + } + model := make([]models.Node, len(nodes)) + for i, node := range nodes { + model[i] = models.Node{Name: node.Node, Status: node.Status} + } + m.updateClusterEndpointStatus(inst.Name, &inst, cc, model) + for i, ep := range inst.ClusterEndpoints { + if !ep.Online || ep.NodeIdentity != []string{"primary-identity", "member-identity", "alias-identity"}[i] { + t.Fatal("trust refusal changed native online state or identity") + } + if ep.PulseReachable == nil || *ep.PulseReachable != (i == 0 || !conflict) || ep.LastPulseCheck == nil { + t.Fatalf("missing/false Pulse reachability observation: %+v", ep) + } + if i > 0 && conflict && !strings.Contains(ep.PulseError, "TLS trust conflict") { + t.Fatalf("member lost actionable trust diagnostic: %+v", ep) + } + if !conflict && ep.PulseError != "" { + t.Fatalf("agreed trust kept a stale failure: %+v", ep) + } + } + wire, err := json.Marshal(struct { + Case, Phase string + Endpoints []config.ClusterEndpoint + }{layout + "/" + mode, phase, inst.ClusterEndpoints}) + if err != nil { + t.Fatal(err) + } + t.Logf("PIN_CONFLICT_ENDPOINT_PROJECTION_JSON=%s", wire) } - t.Logf("PIN_CONFLICT_ENDPOINT_PROJECTION_JSON=%s", wire) - } - refused := construct() - project(refused, "conflicting-saved-trust", true) - if badRequests.Load() != 0 || inst.ClusterEndpoints[2].Fingerprint != strings.Repeat("0", 64) { - t.Fatal("conflict reached credentials or rewrote saved trust") - } - // Normal source configuration reload, not mutation of the owned map or a - // forced/favourable repeat on the refused source. Native state stays online. - inst.ClusterEndpoints[2].Fingerprint = pin(member) - accepted := construct() - project(accepted, "reconciled-config-reload", false) - if badRequests.Load() == 0 { - t.Fatal("agreed saved trust never reached accepted ordinary use") - } - }) + refused := construct(true) + project(refused, "conflicting-saved-trust", true) + if badRequests.Load() != 0 || inst.ClusterEndpoints[2].Fingerprint != strings.Repeat("0", 64) { + t.Fatal("conflict reached credentials or rewrote saved trust") + } + // Normal source configuration reload, not mutation of the owned map or a + // forced/favourable repeat on the refused source. Native state stays online. + inst.ClusterEndpoints[2].Fingerprint = pin(member) + accepted := construct(false) + project(accepted, "reconciled-config-reload", false) + if badRequests.Load() == 0 { + t.Fatal("agreed saved trust never reached accepted ordinary use") + } + }) + } + } +} + +// Exercise the owning monitor routes, not just equivalent calls to their +// builder. Discovery is allowed to report a collapsed/changed member list, +// but must not make saved ambiguity disappear or interrupt trusted siblings. +func TestClusterPinConflictsSurviveMonitorLifecycle(t *testing.T) { + originalDelays := connRetryDelays + connRetryDelays = []time.Duration{time.Millisecond, time.Millisecond} + t.Cleanup(func() { connRetryDelays = originalDelays }) + for _, mode := range []string{"token", "password-json", "password-form"} { + for _, layout := range []string{"url-aliases", "exact-duplicates"} { + for _, route := range []string{"init", "reconnect", "refresh"} { + t.Run(layout+"/"+mode+"/"+route, func(t *testing.T) { + var badRequests, safeReads atomic.Int32 + endpoint := func(bad bool) *httptest.Server { + s := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if bad { + badRequests.Add(1) + } + switch r.URL.Path { + case "/api2/json/access/ticket": + if mode == "password-form" && r.Header.Get("Content-Type") == "application/json" { + w.WriteHeader(http.StatusUnsupportedMediaType) + return + } + fmt.Fprint(w, `{"data":{"ticket":"fixture-ticket","CSRFPreventionToken":"fixture-csrf"}}`) + case "/api2/json/nodes": + if !bad { + safeReads.Add(1) + } + fmt.Fprint(w, `{"data":[{"node":"primary","status":"online"},{"node":"member","status":"online"}]}`) + default: + t.Errorf("unexpected fixture request: %s", r.URL.Path) + http.NotFound(w, r) + } + })) + t.Cleanup(s.Close) + return s + } + primary, member := endpoint(false), endpoint(true) + pin := func(s *httptest.Server) string { + sum := sha256.Sum256(s.Certificate().Raw) + return hex.EncodeToString(sum[:]) + } + alias := member.URL + "/./" + if layout == "exact-duplicates" { + alias = member.URL + } + inst := config.PVEInstance{Name: "lifecycle-cluster", Host: primary.URL, IsCluster: true, ClusterName: "Fixture cluster", VerifySSL: true, Fingerprint: pin(primary), ClusterEndpoints: []config.ClusterEndpoint{ + {NodeName: "primary", NodeIdentity: "primary-id", Host: primary.URL, Fingerprint: pin(primary)}, + {NodeName: "member", NodeIdentity: "member-id", Host: member.URL, Fingerprint: pin(member)}, + {NodeName: "member-alias", NodeIdentity: "alias-id", Host: alias, Fingerprint: strings.Repeat("0", 64)}, + }} + if mode == "token" { + inst.TokenName, inst.TokenValue = "fixture@pve!pulse", "fixture" + } else { + inst.User, inst.Password = "fixture@pam", "fixture-password" + } + m := &Monitor{config: &config.Config{PVEInstances: []config.PVEInstance{inst}, ConnectionTimeout: 2 * time.Second}, pveClients: make(map[string]PVEClientInterface)} + switch route { + case "init": + m.initPVEClients(m.config) + case "reconnect": + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + m.retryFailedConnections(ctx) + if err := ctx.Err(); err != nil { + t.Fatalf("missing-client retry did not finish: %v", err) + } + case "refresh": + m.rebuildPVEClusterClient(inst.Name) + original := m.pveClients[inst.Name] + // Discovery both drops the contradictory row and changes a + // sibling's address. Neither is permission to reconcile trust. + discovered := append([]config.ClusterEndpoint(nil), inst.ClusterEndpoints[:2]...) + discovered[0].Host += "/" + m.refreshClusterEndpoints(inst.Name, &m.config.PVEInstances[0], inst.ClusterName, discovered) + if m.pveClients[inst.Name] != original || !reflect.DeepEqual(m.config.PVEInstances[0], inst) { + t.Fatal("discovery erased conflicting saved trust or replaced active sibling clients") + } + } + cc, ok := m.pveClients[inst.Name].(*proxmox.ClusterClient) + if !ok { + t.Fatal("owning monitor route did not retain its cluster client") + } + if _, err := cc.GetNodes(context.Background()); err != nil || safeReads.Load() == 0 || badRequests.Load() != 0 { + t.Fatalf("monitor route lost sibling use or dispatched conflicting trust: %v safe=%d bad=%d", err, safeReads.Load(), badRequests.Load()) + } + if health := cc.GetHealthStatusWithErrors()[member.URL]; health.Healthy || !strings.Contains(health.LastError, "TLS trust conflict") { + t.Fatalf("monitor route lost member's trust refusal: %+v", health) + } + if !reflect.DeepEqual(m.config.PVEInstances[0], inst) { + t.Fatal("monitor route changed saved configuration") + } + if route == "refresh" { + // A normal independently reconciled configuration removes + // the hold; automatic address refresh is not disabled forever. + m.config.PVEInstances[0].ClusterEndpoints[2].Fingerprint = pin(member) + discovered := append([]config.ClusterEndpoint(nil), m.config.PVEInstances[0].ClusterEndpoints...) + discovered[0].Host += "/" + m.refreshClusterEndpoints(inst.Name, &m.config.PVEInstances[0], inst.ClusterName, discovered) + reloaded := m.pveClients[inst.Name] + if reloaded == cc { + t.Fatal("reconciled configuration did not resume address refresh") + } + if _, err := reloaded.GetNodes(context.Background()); err != nil || badRequests.Load() == 0 { + t.Fatalf("reconciled refresh did not resume ordinary member use: %v", err) + } + } + }) + } + } } } diff --git a/pkg/proxmox/cluster_client.go b/pkg/proxmox/cluster_client.go index e6a3a4914b..07653ba026 100644 --- a/pkg/proxmox/cluster_client.go +++ b/pkg/proxmox/cluster_client.go @@ -20,18 +20,19 @@ import ( // ClusterClient wraps multiple Proxmox clients for cluster-aware operations type ClusterClient struct { - mu sync.RWMutex - clientInitMu sync.Mutex // Serialize lazy construction while a trust probe releases mu. - name string - clients map[string]*Client // Key is node name - endpoints []string // All available endpoints - endpointFingerprints map[string]string // Per-endpoint TLS fingerprints (TOFU) - nodeHealth map[string]bool // Track node health - lastHealthCheck map[string]time.Time // Track last health check time - lastError map[string]string // Track last error per endpoint - config ClientConfig // Base config (auth info) - rateLimitUntil map[string]time.Time // Cooldown window for rate-limited endpoints - recoveryInProgress bool // Prevent overlapping background recovery sweeps + mu sync.RWMutex + clientInitMu sync.Mutex // Serialize lazy construction while a trust probe releases mu. + name string + clients map[string]*Client // Key is node name + endpoints []string // All available endpoints + endpointFingerprints map[string]string // Per-endpoint TLS fingerprints (TOFU) + conflictingEndpointFingerprints map[string]bool // Exact duplicate saved rows refused by the producer. + nodeHealth map[string]bool // Track node health + lastHealthCheck map[string]time.Time // Track last health check time + lastError map[string]string // Track last error per endpoint + config ClientConfig // Base config (auth info) + rateLimitUntil map[string]time.Time // Cooldown window for rate-limited endpoints + recoveryInProgress bool // Prevent overlapping background recovery sweeps } const ( @@ -157,15 +158,17 @@ func sanitizeEndpointError(errMsg string) string { // NewClusterClient creates a new cluster-aware client. // endpointFingerprints is an optional map of endpoint URL -> TLS fingerprint for per-node certificate verification. // This enables TOFU (Trust On First Use) for clusters with unique self-signed certs per node. -func NewClusterClient(name string, config ClientConfig, endpoints []string, endpointFingerprints map[string]string) *ClusterClient { - cc, _ := NewClusterClientWithContext(context.Background(), name, config, endpoints, endpointFingerprints) +// Optional conflictingEndpoints retain contradictory exact duplicate saved rows +// which a map cannot represent. They refuse only those endpoint trust origins. +func NewClusterClient(name string, config ClientConfig, endpoints []string, endpointFingerprints map[string]string, conflictingEndpoints ...string) *ClusterClient { + cc, _ := NewClusterClientWithContext(context.Background(), name, config, endpoints, endpointFingerprints, conflictingEndpoints...) return cc } // NewClusterClientWithContext keeps synchronous startup checks and password // session construction within the caller's lifetime. Endpoint failures still // produce a recoverable cluster; cancellation does not publish a partial one. -func NewClusterClientWithContext(ctx context.Context, name string, config ClientConfig, endpoints []string, endpointFingerprints map[string]string) (*ClusterClient, error) { +func NewClusterClientWithContext(ctx context.Context, name string, config ClientConfig, endpoints []string, endpointFingerprints map[string]string, conflictingEndpoints ...string) (*ClusterClient, error) { if err := ctx.Err(); err != nil { return nil, err } @@ -180,16 +183,21 @@ func NewClusterClientWithContext(ctx context.Context, name string, config Client for endpoint, fingerprint := range endpointFingerprints { ownedFingerprints[endpoint] = fingerprint } + ownedConflicts := make(map[string]bool, len(conflictingEndpoints)) + for _, endpoint := range conflictingEndpoints { + ownedConflicts[endpoint] = true + } cc := &ClusterClient{ - name: name, - clients: make(map[string]*Client), - endpoints: endpoints, - endpointFingerprints: ownedFingerprints, - nodeHealth: make(map[string]bool), - lastHealthCheck: make(map[string]time.Time), - lastError: make(map[string]string), - config: config, - rateLimitUntil: make(map[string]time.Time), + name: name, + clients: make(map[string]*Client), + endpoints: endpoints, + endpointFingerprints: ownedFingerprints, + conflictingEndpointFingerprints: ownedConflicts, + nodeHealth: make(map[string]bool), + lastHealthCheck: make(map[string]time.Time), + lastError: make(map[string]string), + config: config, + rateLimitUntil: make(map[string]time.Time), } // Initialize all endpoints as unknown (will be tested on first use) @@ -240,6 +248,12 @@ func clusterEndpointKey(endpoint string) (string, bool) { func (cc *ClusterClient) knownEndpointFingerprintLocked(endpoint string) (string, bool, error) { key, valid := clusterEndpointKey(endpoint) + for conflict := range cc.conflictingEndpointFingerprints { + conflictKey, conflictValid := clusterEndpointKey(conflict) + if conflict == endpoint || (valid && conflictValid && conflictKey == key) { + return "", true, errConflictingEndpointFingerprints + } + } var fingerprint string known := false for storedEndpoint, fp := range cc.endpointFingerprints { @@ -268,6 +282,24 @@ func (cc *ClusterClient) knownEndpointFingerprintLocked(endpoint string) (string return fingerprint, known, nil } +// CheckEndpointFingerprintConflicts checks saved trust without constructing a +// client, probing TLS or changing any pin. Discovery must not discard ambiguity +// by collapsing saved rows or replacing their addresses during a refresh. +// This checks agreement only; malformed nonempty pins remain the TLS verifier's +// responsibility and are never treated as absent trust here. +func CheckEndpointFingerprintConflicts(fingerprints map[string]string, conflictingEndpoints ...string) error { + if len(conflictingEndpoints) > 0 { + return errConflictingEndpointFingerprints + } + cc := ClusterClient{endpointFingerprints: fingerprints} + for endpoint := range fingerprints { + if _, _, err := cc.knownEndpointFingerprintLocked(endpoint); err != nil { + return err + } + } + return nil +} + // Only a declared, previously untrusted member may use the primary pin as a // first-use bootstrap. A known member or the configured primary must keep its // established pin; recovery is never permission to replace it. diff --git a/pkg/proxmox/cluster_client_api_test.go b/pkg/proxmox/cluster_client_api_test.go index bcc26019ef..d9c1ee9fc1 100644 --- a/pkg/proxmox/cluster_client_api_test.go +++ b/pkg/proxmox/cluster_client_api_test.go @@ -930,7 +930,11 @@ func TestClusterSavedPinResolutionRejectsConflicts(t *testing.T) { {"member.example.invalid:8006/tenant/./", "https://member.example.invalid:8006/tenant"}, } for _, pair := range pairs { - cc := NewClusterClient("resolution", ClientConfig{Host: "https://primary.example.invalid", Fingerprint: "base-pin"}, nil, map[string]string{pair[0]: "saved-pin", pair[1]: "other-pin"}) + pins := map[string]string{pair[0]: "saved-pin", pair[1]: "other-pin"} + if err := CheckEndpointFingerprintConflicts(pins); !errors.Is(err, errConflictingEndpointFingerprints) { + t.Errorf("refresh check lost contradictory saved trust: %v", err) + } + cc := NewClusterClient("resolution", ClientConfig{Host: "https://primary.example.invalid", Fingerprint: "base-pin"}, nil, pins) for _, endpoint := range pair { fingerprint, err := cc.getEndpointFingerprint(endpoint) if fingerprint != "" || !errors.Is(err, errConflictingEndpointFingerprints) { @@ -948,7 +952,11 @@ func TestClusterSavedPinResolutionRejectsConflicts(t *testing.T) { // Invalid nonempty pin presentation cannot turn into unpinned CA/opt-out // admission, even when every saved spelling agrees after normalization. for _, pin := range []string{":", "::", " ", "not-a-pin"} { - cc := NewClusterClient("invalid-pin", ClientConfig{Host: "https://primary.example.invalid", Fingerprint: "base-pin", VerifySSL: false}, nil, map[string]string{"https://member.example.invalid:443/": pin, "https://MEMBER.example.invalid": pin}) + pins := map[string]string{"https://member.example.invalid:443/": pin, "https://MEMBER.example.invalid": pin} + if err := CheckEndpointFingerprintConflicts(pins); err != nil { + t.Errorf("refresh check refused agreeing nonempty pins: %v", err) + } + cc := NewClusterClient("invalid-pin", ClientConfig{Host: "https://primary.example.invalid", Fingerprint: "base-pin", VerifySSL: false}, nil, pins) got, err := cc.getEndpointFingerprint("https://member.example.invalid") if err != nil || got != pin || got == "" { t.Errorf("invalid saved pin became absent trust: %q %v", got, err) @@ -991,3 +999,25 @@ func TestClusterFirstUseCaptureDoesNotReconcileConflictingPins(t *testing.T) { t.Error("capture authenticated or reconciled contradictory saved trust") } } + +func TestClusterProducerPinConflictsAreOwnedAndScoped(t *testing.T) { + const original = "https://member.example.invalid:443/tenant/" + input := []string{original} + if err := CheckEndpointFingerprintConflicts(map[string]string{original: "saved-pin"}, input...); !errors.Is(err, errConflictingEndpointFingerprints) { + t.Errorf("refresh check discarded producer's duplicate-row conflict: %v", err) + } + cc := NewClusterClient("producer-conflict", ClientConfig{Host: "https://primary.example.invalid", Fingerprint: "primary-pin"}, nil, map[string]string{original: "saved-pin"}, input...) + input[0] = "https://other.example.invalid/tenant" + for _, endpoint := range []string{original, "https://MEMBER.example.invalid/tenant"} { + fingerprint, err := cc.getEndpointFingerprint(endpoint) + if fingerprint != "" || !errors.Is(err, errConflictingEndpointFingerprints) { + t.Errorf("producer's copied conflict was lost: %q %v", fingerprint, err) + } + } + for _, endpoint := range []string{"https://other.example.invalid/tenant", "https://member.example.invalid/other", "http://member.example.invalid/tenant", "https://member.example.invalid:8006/tenant"} { + fingerprint, err := cc.getEndpointFingerprint(endpoint) + if err != nil || fingerprint != "primary-pin" { + t.Errorf("producer conflict refused an independent authority/path: %q %v", fingerprint, err) + } + } +} diff --git a/pkg/proxmox/cluster_pin_conflict_test.go b/pkg/proxmox/cluster_pin_conflict_test.go index 52bbc79398..27ea0249e9 100644 --- a/pkg/proxmox/cluster_pin_conflict_test.go +++ b/pkg/proxmox/cluster_pin_conflict_test.go @@ -169,7 +169,11 @@ func TestClusterSavedPinScopesStayDistinct(t *testing.T) { const endpoint = "https://member.scope.invalid:8006/tenant" const pin = "saved-pin" for _, other := range []string{"http://member.scope.invalid:8006/tenant", "https://member.scope.invalid:8007/tenant", "https://other.scope.invalid:8006/tenant", "https://member.scope.invalid:8006/Tenant", "https://member.scope.invalid:8006/other"} { - cc := NewClusterClient("scope", ClientConfig{Host: endpoint, Fingerprint: "primary-pin"}, nil, map[string]string{endpoint: pin, other: "independent-pin"}) + pins := map[string]string{endpoint: pin, other: "independent-pin"} + if err := CheckEndpointFingerprintConflicts(pins); err != nil { + t.Errorf("independent origin/base path held refresh: %v", err) + } + cc := NewClusterClient("scope", ClientConfig{Host: endpoint, Fingerprint: "primary-pin"}, nil, pins) fingerprint, err := cc.getEndpointFingerprint(endpoint) if err != nil || fingerprint != pin { t.Errorf("distinct endpoint created a saved-trust conflict: %q %v", fingerprint, err)